55 static void dlm_mle_node_down(struct dlm_ctxt *dlm,
59 static void dlm_mle_node_up(struct dlm_ctxt *dlm,
65 static int dlm_do_assert_master(struct dlm_ctxt *dlm,
70 static inline int dlm_mle_equal(struct dlm_ctxt *dlm,  in dlm_mle_equal()  argument
75 	if (dlm != mle->dlm)  in dlm_mle_equal()
92 			struct dlm_ctxt *dlm,
98 static int dlm_find_mle(struct dlm_ctxt *dlm,
106 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,
110 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,
114 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
121 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,
123 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,
125 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,
128 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,
178 static inline void __dlm_mle_attach_hb_events(struct dlm_ctxt *dlm,  in __dlm_mle_attach_hb_events()  argument
181 	assert_spin_locked(&dlm->spinlock);  in __dlm_mle_attach_hb_events()
183 	list_add_tail(&mle->hb_events, &dlm->mle_hb_events);  in __dlm_mle_attach_hb_events()
187 static inline void __dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,  in __dlm_mle_detach_hb_events()  argument
195 static inline void dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,  in dlm_mle_detach_hb_events()  argument
198 	spin_lock(&dlm->spinlock);  in dlm_mle_detach_hb_events()
199 	__dlm_mle_detach_hb_events(dlm, mle);  in dlm_mle_detach_hb_events()
200 	spin_unlock(&dlm->spinlock);  in dlm_mle_detach_hb_events()
205 	struct dlm_ctxt *dlm;  in dlm_get_mle_inuse()  local
206 	dlm = mle->dlm;  in dlm_get_mle_inuse()
208 	assert_spin_locked(&dlm->spinlock);  in dlm_get_mle_inuse()
209 	assert_spin_locked(&dlm->master_lock);  in dlm_get_mle_inuse()
216 	struct dlm_ctxt *dlm;  in dlm_put_mle_inuse()  local
217 	dlm = mle->dlm;  in dlm_put_mle_inuse()
219 	spin_lock(&dlm->spinlock);  in dlm_put_mle_inuse()
220 	spin_lock(&dlm->master_lock);  in dlm_put_mle_inuse()
223 	spin_unlock(&dlm->master_lock);  in dlm_put_mle_inuse()
224 	spin_unlock(&dlm->spinlock);  in dlm_put_mle_inuse()
231 	struct dlm_ctxt *dlm;  in __dlm_put_mle()  local
232 	dlm = mle->dlm;  in __dlm_put_mle()
234 	assert_spin_locked(&dlm->spinlock);  in __dlm_put_mle()
235 	assert_spin_locked(&dlm->master_lock);  in __dlm_put_mle()
250 	struct dlm_ctxt *dlm;  in dlm_put_mle()  local
251 	dlm = mle->dlm;  in dlm_put_mle()
253 	spin_lock(&dlm->spinlock);  in dlm_put_mle()
254 	spin_lock(&dlm->master_lock);  in dlm_put_mle()
256 	spin_unlock(&dlm->master_lock);  in dlm_put_mle()
257 	spin_unlock(&dlm->spinlock);  in dlm_put_mle()
267 			struct dlm_ctxt *dlm,  in dlm_init_mle()  argument
272 	assert_spin_locked(&dlm->spinlock);  in dlm_init_mle()
274 	mle->dlm = dlm;  in dlm_init_mle()
306 	atomic_inc(&dlm->mle_tot_count[mle->type]);  in dlm_init_mle()
307 	atomic_inc(&dlm->mle_cur_count[mle->type]);  in dlm_init_mle()
310 	memcpy(mle->node_map, dlm->domain_map, sizeof(mle->node_map));  in dlm_init_mle()
311 	memcpy(mle->vote_map, dlm->domain_map, sizeof(mle->vote_map));  in dlm_init_mle()
312 	clear_bit(dlm->node_num, mle->vote_map);  in dlm_init_mle()
313 	clear_bit(dlm->node_num, mle->node_map);  in dlm_init_mle()
316 	__dlm_mle_attach_hb_events(dlm, mle);  in dlm_init_mle()
319 void __dlm_unlink_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)  in __dlm_unlink_mle()  argument
321 	assert_spin_locked(&dlm->spinlock);  in __dlm_unlink_mle()
322 	assert_spin_locked(&dlm->master_lock);  in __dlm_unlink_mle()
328 void __dlm_insert_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)  in __dlm_insert_mle()  argument
332 	assert_spin_locked(&dlm->master_lock);  in __dlm_insert_mle()
334 	bucket = dlm_master_hash(dlm, mle->mnamehash);  in __dlm_insert_mle()
339 static int dlm_find_mle(struct dlm_ctxt *dlm,  in dlm_find_mle()  argument
347 	assert_spin_locked(&dlm->master_lock);  in dlm_find_mle()
350 	bucket = dlm_master_hash(dlm, hash);  in dlm_find_mle()
352 		if (!dlm_mle_equal(dlm, tmpmle, name, namelen))  in dlm_find_mle()
361 void dlm_hb_event_notify_attached(struct dlm_ctxt *dlm, int idx, int node_up)  in dlm_hb_event_notify_attached()  argument
365 	assert_spin_locked(&dlm->spinlock);  in dlm_hb_event_notify_attached()
367 	list_for_each_entry(mle, &dlm->mle_hb_events, hb_events) {  in dlm_hb_event_notify_attached()
369 			dlm_mle_node_up(dlm, mle, NULL, idx);  in dlm_hb_event_notify_attached()
371 			dlm_mle_node_down(dlm, mle, NULL, idx);  in dlm_hb_event_notify_attached()
375 static void dlm_mle_node_down(struct dlm_ctxt *dlm,  in dlm_mle_node_down()  argument
389 static void dlm_mle_node_up(struct dlm_ctxt *dlm,  in dlm_mle_node_up()  argument
424 	struct dlm_ctxt *dlm;  in dlm_mle_release()  local
427 	dlm = mle->dlm;  in dlm_mle_release()
429 	assert_spin_locked(&dlm->spinlock);  in dlm_mle_release()
430 	assert_spin_locked(&dlm->master_lock);  in dlm_mle_release()
436 	__dlm_unlink_mle(dlm, mle);  in dlm_mle_release()
439 	__dlm_mle_detach_hb_events(dlm, mle);  in dlm_mle_release()
441 	atomic_dec(&dlm->mle_cur_count[mle->type]);  in dlm_mle_release()
489 	struct dlm_ctxt *dlm;  in dlm_lockres_release()  local
492 	dlm = res->dlm;  in dlm_lockres_release()
501 	atomic_dec(&dlm->res_cur_count);  in dlm_lockres_release()
545 static void dlm_init_lockres(struct dlm_ctxt *dlm,  in dlm_init_lockres()  argument
576 	res->dlm = dlm;  in dlm_init_lockres()
580 	atomic_inc(&dlm->res_tot_count);  in dlm_init_lockres()
581 	atomic_inc(&dlm->res_cur_count);  in dlm_init_lockres()
585 	dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);  in dlm_init_lockres()
592 	spin_lock(&dlm->spinlock);  in dlm_init_lockres()
593 	list_add_tail(&res->tracking, &dlm->tracking_list);  in dlm_init_lockres()
594 	spin_unlock(&dlm->spinlock);  in dlm_init_lockres()
600 struct dlm_lock_resource *dlm_new_lockres(struct dlm_ctxt *dlm,  in dlm_new_lockres()  argument
614 	dlm_init_lockres(dlm, res, name, namelen);  in dlm_new_lockres()
623 void dlm_lockres_set_refmap_bit(struct dlm_ctxt *dlm,  in dlm_lockres_set_refmap_bit()  argument
634 void dlm_lockres_clear_refmap_bit(struct dlm_ctxt *dlm,  in dlm_lockres_clear_refmap_bit()  argument
645 static void __dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,  in __dlm_lockres_grab_inflight_ref()  argument
650 	mlog(0, "%s: res %.*s, inflight++: now %u, %ps()\n", dlm->name,  in __dlm_lockres_grab_inflight_ref()
655 void dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,  in dlm_lockres_grab_inflight_ref()  argument
659 	__dlm_lockres_grab_inflight_ref(dlm, res);  in dlm_lockres_grab_inflight_ref()
662 void dlm_lockres_drop_inflight_ref(struct dlm_ctxt *dlm,  in dlm_lockres_drop_inflight_ref()  argument
671 	mlog(0, "%s: res %.*s, inflight--: now %u, %ps()\n", dlm->name,  in dlm_lockres_drop_inflight_ref()
678 void __dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,  in __dlm_lockres_grab_inflight_worker()  argument
684 			dlm->name, res->lockname.len, res->lockname.name,  in __dlm_lockres_grab_inflight_worker()
688 static void __dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,  in __dlm_lockres_drop_inflight_worker()  argument
695 			dlm->name, res->lockname.len, res->lockname.name,  in __dlm_lockres_drop_inflight_worker()
699 static void dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,  in dlm_lockres_drop_inflight_worker()  argument
703 	__dlm_lockres_drop_inflight_worker(dlm, res);  in dlm_lockres_drop_inflight_worker()
723 struct dlm_lock_resource * dlm_get_lock_resource(struct dlm_ctxt *dlm,  in dlm_get_lock_resource()  argument
745 	spin_lock(&dlm->spinlock);  in dlm_get_lock_resource()
746 	tmpres = __dlm_lookup_lockres_full(dlm, lockid, namelen, hash);  in dlm_get_lock_resource()
748 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
775 			BUG_ON(tmpres->owner == dlm->node_num);  in dlm_get_lock_resource()
785 		dlm_lockres_grab_inflight_ref(dlm, tmpres);  in dlm_get_lock_resource()
789 			spin_lock(&dlm->track_lock);  in dlm_get_lock_resource()
797 			spin_unlock(&dlm->track_lock);  in dlm_get_lock_resource()
805 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
811 		res = dlm_new_lockres(dlm, lockid, namelen);  in dlm_get_lock_resource()
823 		dlm_change_lockres_owner(dlm, res, dlm->node_num);  in dlm_get_lock_resource()
824 		__dlm_insert_lockres(dlm, res);  in dlm_get_lock_resource()
825 		dlm_lockres_grab_inflight_ref(dlm, res);  in dlm_get_lock_resource()
827 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
833 	spin_lock(&dlm->master_lock);  in dlm_get_lock_resource()
836 	blocked = dlm_find_mle(dlm, &mle, (char *)lockid, namelen);  in dlm_get_lock_resource()
853 			BUG_ON(mig && mle->master == dlm->node_num);  in dlm_get_lock_resource()
857 			     dlm->name, namelen, lockid,  in dlm_get_lock_resource()
859 			spin_unlock(&dlm->master_lock);  in dlm_get_lock_resource()
860 			spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
864 				dlm_mle_detach_hb_events(dlm, mle);  in dlm_get_lock_resource()
878 		dlm_init_mle(mle, DLM_MLE_MASTER, dlm, res, NULL, 0);  in dlm_get_lock_resource()
879 		set_bit(dlm->node_num, mle->maybe_map);  in dlm_get_lock_resource()
880 		__dlm_insert_mle(dlm, mle);  in dlm_get_lock_resource()
886 		bit = find_next_bit(dlm->recovery_map, O2NM_MAX_NODES, 0);  in dlm_get_lock_resource()
890 			     dlm->name, namelen, (char *)lockid, bit);  in dlm_get_lock_resource()
901 	__dlm_insert_lockres(dlm, res);  in dlm_get_lock_resource()
904 	__dlm_lockres_grab_inflight_ref(dlm, res);  in dlm_get_lock_resource()
911 	spin_unlock(&dlm->master_lock);  in dlm_get_lock_resource()
912 	spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
921 			     "master $RECOVERY lock now\n", dlm->name);  in dlm_get_lock_resource()
922 			if (!dlm_pre_master_reco_lockres(dlm, res))  in dlm_get_lock_resource()
926 				    "change\n", dlm->name);  in dlm_get_lock_resource()
932 		dlm_kick_recovery_thread(dlm);  in dlm_get_lock_resource()
934 		dlm_wait_for_recovery(dlm);  in dlm_get_lock_resource()
936 		spin_lock(&dlm->spinlock);  in dlm_get_lock_resource()
937 		bit = find_next_bit(dlm->recovery_map, O2NM_MAX_NODES, 0);  in dlm_get_lock_resource()
941 			     dlm->name, namelen, (char *)lockid, bit);  in dlm_get_lock_resource()
945 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
948 			dlm_wait_for_node_recovery(dlm, bit, 10000);  in dlm_get_lock_resource()
970 			     "master is %u, keep going\n", dlm->name, namelen,  in dlm_get_lock_resource()
977 	ret = dlm_wait_for_lock_mastery(dlm, res, mle, &blocked);  in dlm_get_lock_resource()
981 		     "request now, blocked=%d\n", dlm->name, res->lockname.len,  in dlm_get_lock_resource()
986 			     dlm->name, res->lockname.len,  in dlm_get_lock_resource()
995 	mlog(0, "%s: res %.*s, Mastered by %u\n", dlm->name, res->lockname.len,  in dlm_get_lock_resource()
1001 	dlm_mle_detach_hb_events(dlm, mle);  in dlm_get_lock_resource()
1023 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,  in dlm_wait_for_lock_mastery()  argument
1040 		mlog(0, "%s:%.*s: owner is suddenly %u\n", dlm->name,  in dlm_wait_for_lock_mastery()
1045 		if (res->owner != dlm->node_num) {  in dlm_wait_for_lock_mastery()
1070 		     dlm->name, res->lockname.len, res->lockname.name);  in dlm_wait_for_lock_mastery()
1071 		ret = dlm_restart_lock_mastery(dlm, res, mle, *blocked);  in dlm_wait_for_lock_mastery()
1075 			     dlm->name, res->lockname.len, res->lockname.name,  in dlm_wait_for_lock_mastery()
1085 		     "rechecking now\n", dlm->name, res->lockname.len,  in dlm_wait_for_lock_mastery()
1091 			     "for %s:%.*s\n", dlm->name, res->lockname.len,  in dlm_wait_for_lock_mastery()
1105 			if (dlm->node_num <= bit) {  in dlm_wait_for_lock_mastery()
1109 				mle->master = dlm->node_num;  in dlm_wait_for_lock_mastery()
1137 			mlog(0, "%s:%.*s: waiting again\n", dlm->name,  in dlm_wait_for_lock_mastery()
1148 		m = dlm->node_num;  in dlm_wait_for_lock_mastery()
1151 		ret = dlm_do_assert_master(dlm, res, mle->vote_map, 0);  in dlm_wait_for_lock_mastery()
1169 	dlm_change_lockres_owner(dlm, res, m);  in dlm_wait_for_lock_mastery()
1235 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,  in dlm_restart_lock_mastery()  argument
1282 						     "now\n", dlm->name,  in dlm_restart_lock_mastery()
1298 						     dlm->name,  in dlm_restart_lock_mastery()
1316 				set_bit(dlm->node_num, mle->maybe_map);  in dlm_restart_lock_mastery()
1338 	struct dlm_ctxt *dlm = mle->dlm;  in dlm_do_master_request()  local
1343 	request.node_idx = dlm->node_num;  in dlm_do_master_request()
1351 	ret = o2net_send_message(DLM_MASTER_REQUEST_MSG, dlm->key, &request,  in dlm_do_master_request()
1387 			     "reference\n", dlm->name, res->lockname.len,  in dlm_do_master_request()
1433 	struct dlm_ctxt *dlm = data;  in dlm_master_request_handler()  local
1444 	if (!dlm_grab(dlm))  in dlm_master_request_handler()
1447 	if (!dlm_domain_fully_joined(dlm)) {  in dlm_master_request_handler()
1462 	spin_lock(&dlm->spinlock);  in dlm_master_request_handler()
1463 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);  in dlm_master_request_handler()
1465 		spin_unlock(&dlm->spinlock);  in dlm_master_request_handler()
1492 		if (res->owner == dlm->node_num) {  in dlm_master_request_handler()
1493 			dlm_lockres_set_refmap_bit(dlm, res, request->node_idx);  in dlm_master_request_handler()
1526 		spin_lock(&dlm->master_lock);  in dlm_master_request_handler()
1527 		found = dlm_find_mle(dlm, &tmpmle, name, namelen);  in dlm_master_request_handler()
1541 			if (tmpmle->master == dlm->node_num) {  in dlm_master_request_handler()
1552 			if (tmpmle->master == dlm->node_num) {  in dlm_master_request_handler()
1558 				dlm_lockres_set_refmap_bit(dlm, res,  in dlm_master_request_handler()
1571 		spin_unlock(&dlm->master_lock);  in dlm_master_request_handler()
1587 	spin_lock(&dlm->master_lock);  in dlm_master_request_handler()
1588 	found = dlm_find_mle(dlm, &tmpmle, name, namelen);  in dlm_master_request_handler()
1593 			spin_unlock(&dlm->master_lock);  in dlm_master_request_handler()
1594 			spin_unlock(&dlm->spinlock);  in dlm_master_request_handler()
1607 		dlm_init_mle(mle, DLM_MLE_BLOCK, dlm, NULL, name, namelen);  in dlm_master_request_handler()
1609 		__dlm_insert_mle(dlm, mle);  in dlm_master_request_handler()
1615 		if (tmpmle->master == dlm->node_num) {  in dlm_master_request_handler()
1632 	spin_unlock(&dlm->master_lock);  in dlm_master_request_handler()
1633 	spin_unlock(&dlm->spinlock);  in dlm_master_request_handler()
1654 			     dlm->node_num, res->lockname.len, res->lockname.name);  in dlm_master_request_handler()
1656 		ret = dlm_dispatch_assert_master(dlm, res, 0, request->node_idx,  in dlm_master_request_handler()
1665 			__dlm_lockres_grab_inflight_worker(dlm, res);  in dlm_master_request_handler()
1674 		dlm_put(dlm);  in dlm_master_request_handler()
1688 static int dlm_do_assert_master(struct dlm_ctxt *dlm,  in dlm_do_assert_master()  argument
1718 		assert.node_idx = dlm->node_num;  in dlm_do_assert_master()
1723 		tmpret = o2net_send_message(DLM_ASSERT_MASTER_MSG, dlm->key,  in dlm_do_assert_master()
1728 			     DLM_ASSERT_MASTER_MSG, dlm->key, to);  in dlm_do_assert_master()
1742 			spin_lock(&dlm->spinlock);  in dlm_do_assert_master()
1743 			spin_lock(&dlm->master_lock);  in dlm_do_assert_master()
1744 			if (dlm_find_mle(dlm, &mle, (char *)lockname,  in dlm_do_assert_master()
1749 			spin_unlock(&dlm->master_lock);  in dlm_do_assert_master()
1750 			spin_unlock(&dlm->spinlock);  in dlm_do_assert_master()
1772 			dlm_lockres_set_refmap_bit(dlm, res, to);  in dlm_do_assert_master()
1800 	struct dlm_ctxt *dlm = data;  in dlm_assert_master_handler()  local
1810 	if (!dlm_grab(dlm))  in dlm_assert_master_handler()
1823 	spin_lock(&dlm->spinlock);  in dlm_assert_master_handler()
1829 	spin_lock(&dlm->master_lock);  in dlm_assert_master_handler()
1830 	if (!dlm_find_mle(dlm, &mle, name, namelen)) {  in dlm_assert_master_handler()
1863 				     dlm->name, namelen, name,  in dlm_assert_master_handler()
1868 				     dlm->name, namelen, name,  in dlm_assert_master_handler()
1871 				spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
1872 				spin_unlock(&dlm->spinlock);  in dlm_assert_master_handler()
1877 	spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
1881 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);  in dlm_assert_master_handler()
1954 				if (nn != dlm->node_num && nn != assert->node_idx) {  in dlm_assert_master_handler()
1972 			       		dlm->node_num, mle->new_master);  in dlm_assert_master_handler()
1975 				dlm_change_lockres_owner(dlm, res, mle->new_master);  in dlm_assert_master_handler()
1978 				dlm_change_lockres_owner(dlm, res, mle->master);  in dlm_assert_master_handler()
1989 		spin_lock(&dlm->master_lock);  in dlm_assert_master_handler()
2006 			     "inuse=%d\n", dlm->name, namelen, name,  in dlm_assert_master_handler()
2010 		__dlm_unlink_mle(dlm, mle);  in dlm_assert_master_handler()
2011 		__dlm_mle_detach_hb_events(dlm, mle);  in dlm_assert_master_handler()
2020 		spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
2028 	spin_unlock(&dlm->spinlock);  in dlm_assert_master_handler()
2038 	dlm_put(dlm);  in dlm_assert_master_handler()
2046 			     assert->node_idx, dlm->name, namelen, name);  in dlm_assert_master_handler()
2053 		     dlm->name, namelen, name, assert->node_idx);  in dlm_assert_master_handler()
2063 	spin_lock(&dlm->master_lock);  in dlm_assert_master_handler()
2066 	spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
2067 	spin_unlock(&dlm->spinlock);  in dlm_assert_master_handler()
2069 	dlm_put(dlm);  in dlm_assert_master_handler()
2087 int dlm_dispatch_assert_master(struct dlm_ctxt *dlm,  in dlm_dispatch_assert_master()  argument
2098 	dlm_init_work_item(dlm, item, dlm_assert_master_worker, NULL);  in dlm_dispatch_assert_master()
2109 	spin_lock(&dlm->work_lock);  in dlm_dispatch_assert_master()
2110 	list_add_tail(&item->list, &dlm->work_list);  in dlm_dispatch_assert_master()
2111 	spin_unlock(&dlm->work_lock);  in dlm_dispatch_assert_master()
2113 	queue_work(dlm->dlm_worker, &dlm->dispatched_work);  in dlm_dispatch_assert_master()
2119 	struct dlm_ctxt *dlm = data;  in dlm_assert_master_worker()  local
2128 	dlm = item->dlm;  in dlm_assert_master_worker()
2134 	spin_lock(&dlm->spinlock);  in dlm_assert_master_worker()
2135 	memcpy(nodemap, dlm->domain_map, sizeof(nodemap));  in dlm_assert_master_worker()
2136 	spin_unlock(&dlm->spinlock);  in dlm_assert_master_worker()
2138 	clear_bit(dlm->node_num, nodemap);  in dlm_assert_master_worker()
2144 		bit = dlm->node_num;  in dlm_assert_master_worker()
2174 		     res->lockname.len, res->lockname.name, dlm->node_num);  in dlm_assert_master_worker()
2175 	ret = dlm_do_assert_master(dlm, res, nodemap, flags);  in dlm_assert_master_worker()
2183 	dlm_lockres_release_ast(dlm, res);  in dlm_assert_master_worker()
2186 	dlm_lockres_drop_inflight_worker(dlm, res);  in dlm_assert_master_worker()
2203 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,  in dlm_pre_master_reco_lockres()  argument
2211 	spin_lock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2212 	dlm_node_iter_init(dlm->domain_map, &iter);  in dlm_pre_master_reco_lockres()
2213 	spin_unlock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2217 		if (nodenum == dlm->node_num)  in dlm_pre_master_reco_lockres()
2219 		ret = dlm_do_master_requery(dlm, res, nodenum, &master);  in dlm_pre_master_reco_lockres()
2231 			spin_lock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2232 			if (test_bit(master, dlm->recovery_map)) {  in dlm_pre_master_reco_lockres()
2236 				     "lock.  must wait.\n", dlm->name,  in dlm_pre_master_reco_lockres()
2240 			spin_unlock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2241 			mlog(0, "%s: reco lock master is %u\n", dlm->name,  in dlm_pre_master_reco_lockres()
2253 int dlm_drop_lockres_ref(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)  in dlm_drop_lockres_ref()  argument
2265 	deref.node_idx = dlm->node_num;  in dlm_drop_lockres_ref()
2269 	ret = o2net_send_message(DLM_DEREF_LOCKRES_MSG, dlm->key,  in dlm_drop_lockres_ref()
2273 		     dlm->name, namelen, lockname, ret, res->owner);  in dlm_drop_lockres_ref()
2277 		     dlm->name, namelen, lockname, res->owner, r);  in dlm_drop_lockres_ref()
2287 	struct dlm_ctxt *dlm = data;  in dlm_deref_lockres_handler()  local
2299 	if (!dlm_grab(dlm))  in dlm_deref_lockres_handler()
2317 	spin_lock(&dlm->spinlock);  in dlm_deref_lockres_handler()
2318 	res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);  in dlm_deref_lockres_handler()
2320 		spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_handler()
2322 		     dlm->name, namelen, name);  in dlm_deref_lockres_handler()
2325 	spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_handler()
2333 			dlm_lockres_clear_refmap_bit(dlm, res, node);  in dlm_deref_lockres_handler()
2341 			dlm_lockres_calc_usage(dlm, res);  in dlm_deref_lockres_handler()
2344 		     	"but it is already dropped!\n", dlm->name,  in dlm_deref_lockres_handler()
2359 	dlm_init_work_item(dlm, item, dlm_deref_lockres_worker, NULL);  in dlm_deref_lockres_handler()
2363 	spin_lock(&dlm->work_lock);  in dlm_deref_lockres_handler()
2364 	list_add_tail(&item->list, &dlm->work_list);  in dlm_deref_lockres_handler()
2365 	spin_unlock(&dlm->work_lock);  in dlm_deref_lockres_handler()
2367 	queue_work(dlm->dlm_worker, &dlm->dispatched_work);  in dlm_deref_lockres_handler()
2373 	dlm_put(dlm);  in dlm_deref_lockres_handler()
2380 	struct dlm_ctxt *dlm;  in dlm_deref_lockres_worker()  local
2385 	dlm = item->dlm;  in dlm_deref_lockres_worker()
2393 		dlm_lockres_clear_refmap_bit(dlm, res, node);  in dlm_deref_lockres_worker()
2400 		     dlm->name, res->lockname.len, res->lockname.name, node);  in dlm_deref_lockres_worker()
2401 		dlm_lockres_calc_usage(dlm, res);  in dlm_deref_lockres_worker()
2404 		     "but it is already dropped!\n", dlm->name,  in dlm_deref_lockres_worker()
2419 static int dlm_is_lockres_migrateable(struct dlm_ctxt *dlm,  in dlm_is_lockres_migrateable()  argument
2438 	if (res->owner != dlm->node_num)  in dlm_is_lockres_migrateable()
2444 			if (lock->ml.node != dlm->node_num) {  in dlm_is_lockres_migrateable()
2450 			     "%s list\n", dlm->name, res->lockname.len,  in dlm_is_lockres_migrateable()
2465 	mlog(0, "%s: res %.*s, Migrateable\n", dlm->name, res->lockname.len,  in dlm_is_lockres_migrateable()
2476 static int dlm_migrate_lockres(struct dlm_ctxt *dlm,  in dlm_migrate_lockres()  argument
2488 	if (!dlm_grab(dlm))  in dlm_migrate_lockres()
2496 	mlog(0, "%s: Migrating %.*s to node %u\n", dlm->name, namelen, name,  in dlm_migrate_lockres()
2518 	spin_lock(&dlm->spinlock);  in dlm_migrate_lockres()
2519 	spin_lock(&dlm->master_lock);  in dlm_migrate_lockres()
2520 	ret = dlm_add_migration_mle(dlm, res, mle, &oldmle, name,  in dlm_migrate_lockres()
2521 				    namelen, target, dlm->node_num);  in dlm_migrate_lockres()
2527 	spin_unlock(&dlm->master_lock);  in dlm_migrate_lockres()
2528 	spin_unlock(&dlm->spinlock);  in dlm_migrate_lockres()
2540 	if (dlm_mark_lockres_migrating(dlm, res, target) < 0) {  in dlm_migrate_lockres()
2554 		dlm_mle_detach_hb_events(dlm, oldmle);  in dlm_migrate_lockres()
2560 			dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2578 	flush_workqueue(dlm->dlm_worker);  in dlm_migrate_lockres()
2584 	ret = dlm_send_one_lockres(dlm, res, mres, target,  in dlm_migrate_lockres()
2591 		dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2599 			dlm_wait_for_node_death(dlm, target,  in dlm_migrate_lockres()
2627 			     dlm->name, res->lockname.len, res->lockname.name);  in dlm_migrate_lockres()
2630 			if (dlm_is_node_dead(dlm, target)) {  in dlm_migrate_lockres()
2633 				     dlm->name, res->lockname.len,  in dlm_migrate_lockres()
2637 				dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2648 			     dlm->name, res->lockname.len, res->lockname.name);  in dlm_migrate_lockres()
2653 	dlm_set_lockres_owner(dlm, res, target);  in dlm_migrate_lockres()
2655 	dlm_remove_nonlocal_locks(dlm, res);  in dlm_migrate_lockres()
2660 	dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2664 	dlm_lockres_calc_usage(dlm, res);  in dlm_migrate_lockres()
2669 		dlm_kick_thread(dlm, res);  in dlm_migrate_lockres()
2679 	dlm_put(dlm);  in dlm_migrate_lockres()
2681 	mlog(0, "%s: Migrating %.*s to %u, returns %d\n", dlm->name, namelen,  in dlm_migrate_lockres()
2698 int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)  in dlm_empty_lockres()  argument
2704 	assert_spin_locked(&dlm->spinlock);  in dlm_empty_lockres()
2707 	if (dlm_is_lockres_migrateable(dlm, res))  in dlm_empty_lockres()
2708 		target = dlm_pick_migration_target(dlm, res);  in dlm_empty_lockres()
2715 	spin_unlock(&dlm->spinlock);  in dlm_empty_lockres()
2717 	ret = dlm_migrate_lockres(dlm, res, target);  in dlm_empty_lockres()
2720 		     dlm->name, res->lockname.len, res->lockname.name,  in dlm_empty_lockres()
2722 	spin_lock(&dlm->spinlock);  in dlm_empty_lockres()
2727 int dlm_lock_basts_flushed(struct dlm_ctxt *dlm, struct dlm_lock *lock)  in dlm_lock_basts_flushed()  argument
2730 	spin_lock(&dlm->ast_lock);  in dlm_lock_basts_flushed()
2734 	spin_unlock(&dlm->ast_lock);  in dlm_lock_basts_flushed()
2738 static int dlm_migration_can_proceed(struct dlm_ctxt *dlm,  in dlm_migration_can_proceed()  argument
2749 	spin_lock(&dlm->spinlock);  in dlm_migration_can_proceed()
2750 	if (!test_bit(mig_target, dlm->domain_map))  in dlm_migration_can_proceed()
2752 	spin_unlock(&dlm->spinlock);  in dlm_migration_can_proceed()
2756 static int dlm_lockres_is_dirty(struct dlm_ctxt *dlm,  in dlm_lockres_is_dirty()  argument
2767 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,  in dlm_mark_lockres_migrating()  argument
2774 	       res->lockname.len, res->lockname.name, dlm->node_num,  in dlm_mark_lockres_migrating()
2787 	dlm_kick_thread(dlm, res);  in dlm_mark_lockres_migrating()
2795 	wait_event(dlm->ast_wq, !dlm_lockres_is_dirty(dlm, res));  in dlm_mark_lockres_migrating()
2796 	dlm_lockres_release_ast(dlm, res);  in dlm_mark_lockres_migrating()
2804 	ret = wait_event_interruptible_timeout(dlm->migration_wq,  in dlm_mark_lockres_migrating()
2805 		   dlm_migration_can_proceed(dlm, res, target),  in dlm_mark_lockres_migrating()
2810 		       test_bit(target, dlm->domain_map) ? "no":"yes");  in dlm_mark_lockres_migrating()
2814 		       test_bit(target, dlm->domain_map) ? "no":"yes");  in dlm_mark_lockres_migrating()
2816 	if (!dlm_migration_can_proceed(dlm, res, target)) {  in dlm_mark_lockres_migrating()
2823 	spin_lock(&dlm->spinlock);  in dlm_mark_lockres_migrating()
2824 	if (!test_bit(target, dlm->domain_map)) {  in dlm_mark_lockres_migrating()
2829 	spin_unlock(&dlm->spinlock);  in dlm_mark_lockres_migrating()
2859 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,  in dlm_remove_nonlocal_locks()  argument
2868 	BUG_ON(res->owner == dlm->node_num);  in dlm_remove_nonlocal_locks()
2872 			if (lock->ml.node != dlm->node_num) {  in dlm_remove_nonlocal_locks()
2880 				dlm_lockres_clear_refmap_bit(dlm, res,  in dlm_remove_nonlocal_locks()
2898 		if (bit != dlm->node_num) {  in dlm_remove_nonlocal_locks()
2900 			     "migrating lockres, clearing\n", dlm->name,  in dlm_remove_nonlocal_locks()
2902 			dlm_lockres_clear_refmap_bit(dlm, res, bit);  in dlm_remove_nonlocal_locks()
2913 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,  in dlm_pick_migration_target()  argument
2922 	assert_spin_locked(&dlm->spinlock);  in dlm_pick_migration_target()
2929 			if (lock->ml.node == dlm->node_num)  in dlm_pick_migration_target()
2931 			if (test_bit(lock->ml.node, dlm->exit_domain_map))  in dlm_pick_migration_target()
2945 		if (noderef == dlm->node_num)  in dlm_pick_migration_target()
2947 		if (test_bit(noderef, dlm->exit_domain_map))  in dlm_pick_migration_target()
2959 static int dlm_do_migrate_request(struct dlm_ctxt *dlm,  in dlm_do_migrate_request()  argument
2983 		spin_lock(&dlm->spinlock);  in dlm_do_migrate_request()
2984 		skip = (!test_bit(nodenum, dlm->domain_map));  in dlm_do_migrate_request()
2985 		spin_unlock(&dlm->spinlock);  in dlm_do_migrate_request()
2991 		ret = o2net_send_message(DLM_MIGRATE_REQUEST_MSG, dlm->key,  in dlm_do_migrate_request()
2996 			     "MIGRATE_REQUEST to node %u\n", dlm->name,  in dlm_do_migrate_request()
3013 			     dlm->name, res->lockname.len, res->lockname.name,  in dlm_do_migrate_request()
3016 			dlm_lockres_set_refmap_bit(dlm, res, nodenum);  in dlm_do_migrate_request()
3039 	struct dlm_ctxt *dlm = data;  in dlm_migrate_request_handler()  local
3047 	if (!dlm_grab(dlm))  in dlm_migrate_request_handler()
3063 	spin_lock(&dlm->spinlock);  in dlm_migrate_request_handler()
3064 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);  in dlm_migrate_request_handler()
3082 	spin_lock(&dlm->master_lock);  in dlm_migrate_request_handler()
3084 	ret = dlm_add_migration_mle(dlm, res, mle, &oldmle,  in dlm_migrate_request_handler()
3089 	spin_unlock(&dlm->master_lock);  in dlm_migrate_request_handler()
3091 	spin_unlock(&dlm->spinlock);  in dlm_migrate_request_handler()
3095 		dlm_mle_detach_hb_events(dlm, oldmle);  in dlm_migrate_request_handler()
3102 	dlm_put(dlm);  in dlm_migrate_request_handler()
3113 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,  in dlm_add_migration_mle()  argument
3125 	assert_spin_locked(&dlm->spinlock);  in dlm_add_migration_mle()
3126 	assert_spin_locked(&dlm->master_lock);  in dlm_add_migration_mle()
3129 	found = dlm_find_mle(dlm, oldmle, (char *)name, namelen);  in dlm_add_migration_mle()
3134 			if (master == dlm->node_num) {  in dlm_add_migration_mle()
3157 			__dlm_unlink_mle(dlm, tmp);  in dlm_add_migration_mle()
3158 			__dlm_mle_detach_hb_events(dlm, tmp);  in dlm_add_migration_mle()
3164 						"migration\n", dlm->name,  in dlm_add_migration_mle()
3173 	dlm_init_mle(mle, DLM_MLE_MIGRATION, dlm, res, name, namelen);  in dlm_add_migration_mle()
3180 	__dlm_insert_mle(dlm, mle);  in dlm_add_migration_mle()
3188 static struct dlm_lock_resource *dlm_reset_mleres_owner(struct dlm_ctxt *dlm,  in dlm_reset_mleres_owner()  argument
3194 	res = __dlm_lookup_lockres(dlm, mle->mname, mle->mnamelen,  in dlm_reset_mleres_owner()
3197 		spin_unlock(&dlm->master_lock);  in dlm_reset_mleres_owner()
3201 		dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);  in dlm_reset_mleres_owner()
3202 		dlm_move_lockres_to_recovery_list(dlm, res);  in dlm_reset_mleres_owner()
3207 		__dlm_mle_detach_hb_events(dlm, mle);  in dlm_reset_mleres_owner()
3210 		spin_lock(&dlm->master_lock);  in dlm_reset_mleres_owner()
3212 		spin_unlock(&dlm->master_lock);  in dlm_reset_mleres_owner()
3218 static void dlm_clean_migration_mle(struct dlm_ctxt *dlm,  in dlm_clean_migration_mle()  argument
3221 	__dlm_mle_detach_hb_events(dlm, mle);  in dlm_clean_migration_mle()
3224 	__dlm_unlink_mle(dlm, mle);  in dlm_clean_migration_mle()
3231 static void dlm_clean_block_mle(struct dlm_ctxt *dlm,  in dlm_clean_block_mle()  argument
3255 		__dlm_mle_detach_hb_events(dlm, mle);  in dlm_clean_block_mle()
3260 void dlm_clean_master_list(struct dlm_ctxt *dlm, u8 dead_node)  in dlm_clean_master_list()  argument
3268 	mlog(0, "dlm=%s, dead node=%u\n", dlm->name, dead_node);  in dlm_clean_master_list()
3270 	assert_spin_locked(&dlm->spinlock);  in dlm_clean_master_list()
3273 	spin_lock(&dlm->master_lock);  in dlm_clean_master_list()
3275 		bucket = dlm_master_hash(dlm, i);  in dlm_clean_master_list()
3291 				dlm_clean_block_mle(dlm, mle, dead_node);  in dlm_clean_master_list()
3314 						dlm->name, dead_node,  in dlm_clean_master_list()
3321 			dlm_clean_migration_mle(dlm, mle);  in dlm_clean_master_list()
3324 			     "%u to %u!\n", dlm->name, dead_node, mle->master,  in dlm_clean_master_list()
3332 			res = dlm_reset_mleres_owner(dlm, mle);  in dlm_clean_master_list()
3341 	spin_unlock(&dlm->master_lock);  in dlm_clean_master_list()
3344 int dlm_finish_migration(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,  in dlm_finish_migration()  argument
3350 	spin_lock(&dlm->spinlock);  in dlm_finish_migration()
3351 	dlm_node_iter_init(dlm->domain_map, &iter);  in dlm_finish_migration()
3353 	clear_bit(dlm->node_num, iter.node_map);  in dlm_finish_migration()
3354 	spin_unlock(&dlm->spinlock);  in dlm_finish_migration()
3360 	dlm_lockres_set_refmap_bit(dlm, res, old_master);  in dlm_finish_migration()
3364 	ret = dlm_do_migrate_request(dlm, res, old_master,  in dlm_finish_migration()
3365 				     dlm->node_num, &iter);  in dlm_finish_migration()
3375 	ret = dlm_do_assert_master(dlm, res, iter.node_map,  in dlm_finish_migration()
3387 	ret = dlm_do_assert_master(dlm, res, iter.node_map,  in dlm_finish_migration()
3399 	dlm_set_lockres_owner(dlm, res, dlm->node_num);  in dlm_finish_migration()
3403 	dlm_kick_thread(dlm, res);  in dlm_finish_migration()
3442 void dlm_lockres_release_ast(struct dlm_ctxt *dlm,  in dlm_lockres_release_ast()  argument
3458 	wake_up(&dlm->migration_wq);  in dlm_lockres_release_ast()
3461 void dlm_force_free_mles(struct dlm_ctxt *dlm)  in dlm_force_free_mles()  argument
3474 	spin_lock(&dlm->spinlock);  in dlm_force_free_mles()
3475 	spin_lock(&dlm->master_lock);  in dlm_force_free_mles()
3477 	BUG_ON(dlm->dlm_state != DLM_CTXT_LEAVING);  in dlm_force_free_mles()
3478 	BUG_ON((find_next_bit(dlm->domain_map, O2NM_MAX_NODES, 0) < O2NM_MAX_NODES));  in dlm_force_free_mles()
3481 		bucket = dlm_master_hash(dlm, i);  in dlm_force_free_mles()
3490 			__dlm_unlink_mle(dlm, mle);  in dlm_force_free_mles()
3491 			__dlm_mle_detach_hb_events(dlm, mle);  in dlm_force_free_mles()
3495 	spin_unlock(&dlm->master_lock);  in dlm_force_free_mles()
3496 	spin_unlock(&dlm->spinlock);  in dlm_force_free_mles()