38 static void dlm_mle_node_down(struct dlm_ctxt *dlm,
42 static void dlm_mle_node_up(struct dlm_ctxt *dlm,
48 static int dlm_do_assert_master(struct dlm_ctxt *dlm,
53 static inline int dlm_mle_equal(struct dlm_ctxt *dlm,  in dlm_mle_equal()  argument
58 	if (dlm != mle->dlm)  in dlm_mle_equal()
75 			struct dlm_ctxt *dlm,
81 static int dlm_find_mle(struct dlm_ctxt *dlm,
89 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,
93 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,
97 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
104 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,
106 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,
108 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,
111 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,
161 static inline void __dlm_mle_attach_hb_events(struct dlm_ctxt *dlm,  in __dlm_mle_attach_hb_events()  argument
164 	assert_spin_locked(&dlm->spinlock);  in __dlm_mle_attach_hb_events()
166 	list_add_tail(&mle->hb_events, &dlm->mle_hb_events);  in __dlm_mle_attach_hb_events()
170 static inline void __dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,  in __dlm_mle_detach_hb_events()  argument
178 static inline void dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,  in dlm_mle_detach_hb_events()  argument
181 	spin_lock(&dlm->spinlock);  in dlm_mle_detach_hb_events()
182 	__dlm_mle_detach_hb_events(dlm, mle);  in dlm_mle_detach_hb_events()
183 	spin_unlock(&dlm->spinlock);  in dlm_mle_detach_hb_events()
188 	struct dlm_ctxt *dlm;  in dlm_get_mle_inuse()  local
189 	dlm = mle->dlm;  in dlm_get_mle_inuse()
191 	assert_spin_locked(&dlm->spinlock);  in dlm_get_mle_inuse()
192 	assert_spin_locked(&dlm->master_lock);  in dlm_get_mle_inuse()
199 	struct dlm_ctxt *dlm;  in dlm_put_mle_inuse()  local
200 	dlm = mle->dlm;  in dlm_put_mle_inuse()
202 	spin_lock(&dlm->spinlock);  in dlm_put_mle_inuse()
203 	spin_lock(&dlm->master_lock);  in dlm_put_mle_inuse()
206 	spin_unlock(&dlm->master_lock);  in dlm_put_mle_inuse()
207 	spin_unlock(&dlm->spinlock);  in dlm_put_mle_inuse()
214 	struct dlm_ctxt *dlm;  in __dlm_put_mle()  local
215 	dlm = mle->dlm;  in __dlm_put_mle()
217 	assert_spin_locked(&dlm->spinlock);  in __dlm_put_mle()
218 	assert_spin_locked(&dlm->master_lock);  in __dlm_put_mle()
233 	struct dlm_ctxt *dlm;  in dlm_put_mle()  local
234 	dlm = mle->dlm;  in dlm_put_mle()
236 	spin_lock(&dlm->spinlock);  in dlm_put_mle()
237 	spin_lock(&dlm->master_lock);  in dlm_put_mle()
239 	spin_unlock(&dlm->master_lock);  in dlm_put_mle()
240 	spin_unlock(&dlm->spinlock);  in dlm_put_mle()
250 			struct dlm_ctxt *dlm,  in dlm_init_mle()  argument
255 	assert_spin_locked(&dlm->spinlock);  in dlm_init_mle()
257 	mle->dlm = dlm;  in dlm_init_mle()
289 	atomic_inc(&dlm->mle_tot_count[mle->type]);  in dlm_init_mle()
290 	atomic_inc(&dlm->mle_cur_count[mle->type]);  in dlm_init_mle()
293 	memcpy(mle->node_map, dlm->domain_map, sizeof(mle->node_map));  in dlm_init_mle()
294 	memcpy(mle->vote_map, dlm->domain_map, sizeof(mle->vote_map));  in dlm_init_mle()
295 	clear_bit(dlm->node_num, mle->vote_map);  in dlm_init_mle()
296 	clear_bit(dlm->node_num, mle->node_map);  in dlm_init_mle()
299 	__dlm_mle_attach_hb_events(dlm, mle);  in dlm_init_mle()
302 void __dlm_unlink_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)  in __dlm_unlink_mle()  argument
304 	assert_spin_locked(&dlm->spinlock);  in __dlm_unlink_mle()
305 	assert_spin_locked(&dlm->master_lock);  in __dlm_unlink_mle()
311 void __dlm_insert_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)  in __dlm_insert_mle()  argument
315 	assert_spin_locked(&dlm->master_lock);  in __dlm_insert_mle()
317 	bucket = dlm_master_hash(dlm, mle->mnamehash);  in __dlm_insert_mle()
322 static int dlm_find_mle(struct dlm_ctxt *dlm,  in dlm_find_mle()  argument
330 	assert_spin_locked(&dlm->master_lock);  in dlm_find_mle()
333 	bucket = dlm_master_hash(dlm, hash);  in dlm_find_mle()
335 		if (!dlm_mle_equal(dlm, tmpmle, name, namelen))  in dlm_find_mle()
344 void dlm_hb_event_notify_attached(struct dlm_ctxt *dlm, int idx, int node_up)  in dlm_hb_event_notify_attached()  argument
348 	assert_spin_locked(&dlm->spinlock);  in dlm_hb_event_notify_attached()
350 	list_for_each_entry(mle, &dlm->mle_hb_events, hb_events) {  in dlm_hb_event_notify_attached()
352 			dlm_mle_node_up(dlm, mle, NULL, idx);  in dlm_hb_event_notify_attached()
354 			dlm_mle_node_down(dlm, mle, NULL, idx);  in dlm_hb_event_notify_attached()
358 static void dlm_mle_node_down(struct dlm_ctxt *dlm,  in dlm_mle_node_down()  argument
372 static void dlm_mle_node_up(struct dlm_ctxt *dlm,  in dlm_mle_node_up()  argument
406 	struct dlm_ctxt *dlm;  in dlm_mle_release()  local
409 	dlm = mle->dlm;  in dlm_mle_release()
411 	assert_spin_locked(&dlm->spinlock);  in dlm_mle_release()
412 	assert_spin_locked(&dlm->master_lock);  in dlm_mle_release()
418 	__dlm_unlink_mle(dlm, mle);  in dlm_mle_release()
421 	__dlm_mle_detach_hb_events(dlm, mle);  in dlm_mle_release()
423 	atomic_dec(&dlm->mle_cur_count[mle->type]);  in dlm_mle_release()
467 	struct dlm_ctxt *dlm;  in dlm_lockres_release()  local
470 	dlm = res->dlm;  in dlm_lockres_release()
479 	atomic_dec(&dlm->res_cur_count);  in dlm_lockres_release()
523 static void dlm_init_lockres(struct dlm_ctxt *dlm,  in dlm_init_lockres()  argument
554 	res->dlm = dlm;  in dlm_init_lockres()
558 	atomic_inc(&dlm->res_tot_count);  in dlm_init_lockres()
559 	atomic_inc(&dlm->res_cur_count);  in dlm_init_lockres()
563 	dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);  in dlm_init_lockres()
570 	spin_lock(&dlm->track_lock);  in dlm_init_lockres()
571 	list_add_tail(&res->tracking, &dlm->tracking_list);  in dlm_init_lockres()
572 	spin_unlock(&dlm->track_lock);  in dlm_init_lockres()
578 struct dlm_lock_resource *dlm_new_lockres(struct dlm_ctxt *dlm,  in dlm_new_lockres()  argument
592 	dlm_init_lockres(dlm, res, name, namelen);  in dlm_new_lockres()
601 void dlm_lockres_set_refmap_bit(struct dlm_ctxt *dlm,  in dlm_lockres_set_refmap_bit()  argument
612 void dlm_lockres_clear_refmap_bit(struct dlm_ctxt *dlm,  in dlm_lockres_clear_refmap_bit()  argument
623 static void __dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,  in __dlm_lockres_grab_inflight_ref()  argument
628 	mlog(0, "%s: res %.*s, inflight++: now %u, %ps()\n", dlm->name,  in __dlm_lockres_grab_inflight_ref()
633 void dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,  in dlm_lockres_grab_inflight_ref()  argument
637 	__dlm_lockres_grab_inflight_ref(dlm, res);  in dlm_lockres_grab_inflight_ref()
640 void dlm_lockres_drop_inflight_ref(struct dlm_ctxt *dlm,  in dlm_lockres_drop_inflight_ref()  argument
649 	mlog(0, "%s: res %.*s, inflight--: now %u, %ps()\n", dlm->name,  in dlm_lockres_drop_inflight_ref()
656 void __dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,  in __dlm_lockres_grab_inflight_worker()  argument
662 			dlm->name, res->lockname.len, res->lockname.name,  in __dlm_lockres_grab_inflight_worker()
666 static void __dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,  in __dlm_lockres_drop_inflight_worker()  argument
673 			dlm->name, res->lockname.len, res->lockname.name,  in __dlm_lockres_drop_inflight_worker()
677 static void dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,  in dlm_lockres_drop_inflight_worker()  argument
681 	__dlm_lockres_drop_inflight_worker(dlm, res);  in dlm_lockres_drop_inflight_worker()
701 struct dlm_lock_resource * dlm_get_lock_resource(struct dlm_ctxt *dlm,  in dlm_get_lock_resource()  argument
723 	spin_lock(&dlm->spinlock);  in dlm_get_lock_resource()
724 	tmpres = __dlm_lookup_lockres_full(dlm, lockid, namelen, hash);  in dlm_get_lock_resource()
726 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
753 			BUG_ON(tmpres->owner == dlm->node_num);  in dlm_get_lock_resource()
763 		dlm_lockres_grab_inflight_ref(dlm, tmpres);  in dlm_get_lock_resource()
767 			spin_lock(&dlm->track_lock);  in dlm_get_lock_resource()
775 			spin_unlock(&dlm->track_lock);  in dlm_get_lock_resource()
783 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
789 		res = dlm_new_lockres(dlm, lockid, namelen);  in dlm_get_lock_resource()
801 		dlm_change_lockres_owner(dlm, res, dlm->node_num);  in dlm_get_lock_resource()
802 		__dlm_insert_lockres(dlm, res);  in dlm_get_lock_resource()
803 		dlm_lockres_grab_inflight_ref(dlm, res);  in dlm_get_lock_resource()
805 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
811 	spin_lock(&dlm->master_lock);  in dlm_get_lock_resource()
814 	blocked = dlm_find_mle(dlm, &mle, (char *)lockid, namelen);  in dlm_get_lock_resource()
831 			BUG_ON(mig && mle->master == dlm->node_num);  in dlm_get_lock_resource()
835 			     dlm->name, namelen, lockid,  in dlm_get_lock_resource()
837 			spin_unlock(&dlm->master_lock);  in dlm_get_lock_resource()
838 			spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
842 				dlm_mle_detach_hb_events(dlm, mle);  in dlm_get_lock_resource()
856 		dlm_init_mle(mle, DLM_MLE_MASTER, dlm, res, NULL, 0);  in dlm_get_lock_resource()
857 		set_bit(dlm->node_num, mle->maybe_map);  in dlm_get_lock_resource()
858 		__dlm_insert_mle(dlm, mle);  in dlm_get_lock_resource()
864 		bit = find_first_bit(dlm->recovery_map, O2NM_MAX_NODES);  in dlm_get_lock_resource()
868 			     dlm->name, namelen, (char *)lockid, bit);  in dlm_get_lock_resource()
879 	__dlm_insert_lockres(dlm, res);  in dlm_get_lock_resource()
882 	__dlm_lockres_grab_inflight_ref(dlm, res);  in dlm_get_lock_resource()
889 	spin_unlock(&dlm->master_lock);  in dlm_get_lock_resource()
890 	spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
899 			     "master $RECOVERY lock now\n", dlm->name);  in dlm_get_lock_resource()
900 			if (!dlm_pre_master_reco_lockres(dlm, res))  in dlm_get_lock_resource()
904 				    "change\n", dlm->name);  in dlm_get_lock_resource()
910 		dlm_kick_recovery_thread(dlm);  in dlm_get_lock_resource()
912 		dlm_wait_for_recovery(dlm);  in dlm_get_lock_resource()
914 		spin_lock(&dlm->spinlock);  in dlm_get_lock_resource()
915 		bit = find_first_bit(dlm->recovery_map, O2NM_MAX_NODES);  in dlm_get_lock_resource()
919 			     dlm->name, namelen, (char *)lockid, bit);  in dlm_get_lock_resource()
923 		spin_unlock(&dlm->spinlock);  in dlm_get_lock_resource()
926 			dlm_wait_for_node_recovery(dlm, bit, 10000);  in dlm_get_lock_resource()
948 			     "master is %u, keep going\n", dlm->name, namelen,  in dlm_get_lock_resource()
955 	ret = dlm_wait_for_lock_mastery(dlm, res, mle, &blocked);  in dlm_get_lock_resource()
959 		     "request now, blocked=%d\n", dlm->name, res->lockname.len,  in dlm_get_lock_resource()
964 			     dlm->name, res->lockname.len,  in dlm_get_lock_resource()
973 	mlog(0, "%s: res %.*s, Mastered by %u\n", dlm->name, res->lockname.len,  in dlm_get_lock_resource()
979 	dlm_mle_detach_hb_events(dlm, mle);  in dlm_get_lock_resource()
1001 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,  in dlm_wait_for_lock_mastery()  argument
1018 		mlog(0, "%s:%.*s: owner is suddenly %u\n", dlm->name,  in dlm_wait_for_lock_mastery()
1023 		if (res->owner != dlm->node_num) {  in dlm_wait_for_lock_mastery()
1048 		     dlm->name, res->lockname.len, res->lockname.name);  in dlm_wait_for_lock_mastery()
1049 		ret = dlm_restart_lock_mastery(dlm, res, mle, *blocked);  in dlm_wait_for_lock_mastery()
1053 			     dlm->name, res->lockname.len, res->lockname.name,  in dlm_wait_for_lock_mastery()
1063 		     "rechecking now\n", dlm->name, res->lockname.len,  in dlm_wait_for_lock_mastery()
1069 			     "for %s:%.*s\n", dlm->name, res->lockname.len,  in dlm_wait_for_lock_mastery()
1083 			if (dlm->node_num <= bit) {  in dlm_wait_for_lock_mastery()
1087 				mle->master = dlm->node_num;  in dlm_wait_for_lock_mastery()
1108 			mlog(0, "%s:%.*s: waiting again\n", dlm->name,  in dlm_wait_for_lock_mastery()
1119 		m = dlm->node_num;  in dlm_wait_for_lock_mastery()
1122 		ret = dlm_do_assert_master(dlm, res, mle->vote_map, 0);  in dlm_wait_for_lock_mastery()
1140 	dlm_change_lockres_owner(dlm, res, m);  in dlm_wait_for_lock_mastery()
1206 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,  in dlm_restart_lock_mastery()  argument
1253 						     "now\n", dlm->name,  in dlm_restart_lock_mastery()
1269 						     dlm->name,  in dlm_restart_lock_mastery()
1287 				set_bit(dlm->node_num, mle->maybe_map);  in dlm_restart_lock_mastery()
1309 	struct dlm_ctxt *dlm = mle->dlm;  in dlm_do_master_request()  local
1314 	request.node_idx = dlm->node_num;  in dlm_do_master_request()
1322 	ret = o2net_send_message(DLM_MASTER_REQUEST_MSG, dlm->key, &request,  in dlm_do_master_request()
1358 			     "reference\n", dlm->name, res->lockname.len,  in dlm_do_master_request()
1404 	struct dlm_ctxt *dlm = data;  in dlm_master_request_handler()  local
1415 	if (!dlm_grab(dlm))  in dlm_master_request_handler()
1418 	if (!dlm_domain_fully_joined(dlm)) {  in dlm_master_request_handler()
1433 	spin_lock(&dlm->spinlock);  in dlm_master_request_handler()
1434 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);  in dlm_master_request_handler()
1436 		spin_unlock(&dlm->spinlock);  in dlm_master_request_handler()
1463 		if (res->owner == dlm->node_num) {  in dlm_master_request_handler()
1464 			dlm_lockres_set_refmap_bit(dlm, res, request->node_idx);  in dlm_master_request_handler()
1497 		spin_lock(&dlm->master_lock);  in dlm_master_request_handler()
1498 		found = dlm_find_mle(dlm, &tmpmle, name, namelen);  in dlm_master_request_handler()
1512 			if (tmpmle->master == dlm->node_num) {  in dlm_master_request_handler()
1523 			if (tmpmle->master == dlm->node_num) {  in dlm_master_request_handler()
1529 				dlm_lockres_set_refmap_bit(dlm, res,  in dlm_master_request_handler()
1542 		spin_unlock(&dlm->master_lock);  in dlm_master_request_handler()
1558 	spin_lock(&dlm->master_lock);  in dlm_master_request_handler()
1559 	found = dlm_find_mle(dlm, &tmpmle, name, namelen);  in dlm_master_request_handler()
1564 			spin_unlock(&dlm->master_lock);  in dlm_master_request_handler()
1565 			spin_unlock(&dlm->spinlock);  in dlm_master_request_handler()
1578 		dlm_init_mle(mle, DLM_MLE_BLOCK, dlm, NULL, name, namelen);  in dlm_master_request_handler()
1580 		__dlm_insert_mle(dlm, mle);  in dlm_master_request_handler()
1584 		if (tmpmle->master == dlm->node_num) {  in dlm_master_request_handler()
1600 	spin_unlock(&dlm->master_lock);  in dlm_master_request_handler()
1601 	spin_unlock(&dlm->spinlock);  in dlm_master_request_handler()
1616 			     dlm->node_num, res->lockname.len, res->lockname.name);  in dlm_master_request_handler()
1618 		ret = dlm_dispatch_assert_master(dlm, res, 0, request->node_idx,  in dlm_master_request_handler()
1627 			__dlm_lockres_grab_inflight_worker(dlm, res);  in dlm_master_request_handler()
1636 		dlm_put(dlm);  in dlm_master_request_handler()
1650 static int dlm_do_assert_master(struct dlm_ctxt *dlm,  in dlm_do_assert_master()  argument
1680 		assert.node_idx = dlm->node_num;  in dlm_do_assert_master()
1685 		tmpret = o2net_send_message(DLM_ASSERT_MASTER_MSG, dlm->key,  in dlm_do_assert_master()
1690 			     DLM_ASSERT_MASTER_MSG, dlm->key, to);  in dlm_do_assert_master()
1704 			spin_lock(&dlm->spinlock);  in dlm_do_assert_master()
1705 			spin_lock(&dlm->master_lock);  in dlm_do_assert_master()
1706 			if (dlm_find_mle(dlm, &mle, (char *)lockname,  in dlm_do_assert_master()
1711 			spin_unlock(&dlm->master_lock);  in dlm_do_assert_master()
1712 			spin_unlock(&dlm->spinlock);  in dlm_do_assert_master()
1734 			dlm_lockres_set_refmap_bit(dlm, res, to);  in dlm_do_assert_master()
1762 	struct dlm_ctxt *dlm = data;  in dlm_assert_master_handler()  local
1772 	if (!dlm_grab(dlm))  in dlm_assert_master_handler()
1785 	spin_lock(&dlm->spinlock);  in dlm_assert_master_handler()
1791 	spin_lock(&dlm->master_lock);  in dlm_assert_master_handler()
1792 	if (!dlm_find_mle(dlm, &mle, name, namelen)) {  in dlm_assert_master_handler()
1825 				     dlm->name, namelen, name,  in dlm_assert_master_handler()
1830 				     dlm->name, namelen, name,  in dlm_assert_master_handler()
1833 				spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
1834 				spin_unlock(&dlm->spinlock);  in dlm_assert_master_handler()
1839 	spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
1843 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);  in dlm_assert_master_handler()
1916 				if (nn != dlm->node_num && nn != assert->node_idx) {  in dlm_assert_master_handler()
1934 			       		dlm->node_num, mle->new_master);  in dlm_assert_master_handler()
1937 				dlm_change_lockres_owner(dlm, res, mle->new_master);  in dlm_assert_master_handler()
1940 				dlm_change_lockres_owner(dlm, res, mle->master);  in dlm_assert_master_handler()
1951 		spin_lock(&dlm->master_lock);  in dlm_assert_master_handler()
1968 			     "inuse=%d\n", dlm->name, namelen, name,  in dlm_assert_master_handler()
1972 		__dlm_unlink_mle(dlm, mle);  in dlm_assert_master_handler()
1973 		__dlm_mle_detach_hb_events(dlm, mle);  in dlm_assert_master_handler()
1982 		spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
1990 	spin_unlock(&dlm->spinlock);  in dlm_assert_master_handler()
2000 	dlm_put(dlm);  in dlm_assert_master_handler()
2008 			     assert->node_idx, dlm->name, namelen, name);  in dlm_assert_master_handler()
2015 		     dlm->name, namelen, name, assert->node_idx);  in dlm_assert_master_handler()
2025 	spin_lock(&dlm->master_lock);  in dlm_assert_master_handler()
2028 	spin_unlock(&dlm->master_lock);  in dlm_assert_master_handler()
2029 	spin_unlock(&dlm->spinlock);  in dlm_assert_master_handler()
2031 	dlm_put(dlm);  in dlm_assert_master_handler()
2049 int dlm_dispatch_assert_master(struct dlm_ctxt *dlm,  in dlm_dispatch_assert_master()  argument
2060 	dlm_init_work_item(dlm, item, dlm_assert_master_worker, NULL);  in dlm_dispatch_assert_master()
2071 	spin_lock(&dlm->work_lock);  in dlm_dispatch_assert_master()
2072 	list_add_tail(&item->list, &dlm->work_list);  in dlm_dispatch_assert_master()
2073 	spin_unlock(&dlm->work_lock);  in dlm_dispatch_assert_master()
2075 	queue_work(dlm->dlm_worker, &dlm->dispatched_work);  in dlm_dispatch_assert_master()
2081 	struct dlm_ctxt *dlm = data;  in dlm_assert_master_worker()  local
2090 	dlm = item->dlm;  in dlm_assert_master_worker()
2096 	spin_lock(&dlm->spinlock);  in dlm_assert_master_worker()
2097 	memcpy(nodemap, dlm->domain_map, sizeof(nodemap));  in dlm_assert_master_worker()
2098 	spin_unlock(&dlm->spinlock);  in dlm_assert_master_worker()
2100 	clear_bit(dlm->node_num, nodemap);  in dlm_assert_master_worker()
2106 		bit = dlm->node_num;  in dlm_assert_master_worker()
2136 		     res->lockname.len, res->lockname.name, dlm->node_num);  in dlm_assert_master_worker()
2137 	ret = dlm_do_assert_master(dlm, res, nodemap, flags);  in dlm_assert_master_worker()
2145 	dlm_lockres_release_ast(dlm, res);  in dlm_assert_master_worker()
2148 	dlm_lockres_drop_inflight_worker(dlm, res);  in dlm_assert_master_worker()
2165 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,  in dlm_pre_master_reco_lockres()  argument
2173 	spin_lock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2174 	dlm_node_iter_init(dlm->domain_map, &iter);  in dlm_pre_master_reco_lockres()
2175 	spin_unlock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2179 		if (nodenum == dlm->node_num)  in dlm_pre_master_reco_lockres()
2181 		ret = dlm_do_master_requery(dlm, res, nodenum, &master);  in dlm_pre_master_reco_lockres()
2193 			spin_lock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2194 			if (test_bit(master, dlm->recovery_map)) {  in dlm_pre_master_reco_lockres()
2198 				     "lock.  must wait.\n", dlm->name,  in dlm_pre_master_reco_lockres()
2202 			spin_unlock(&dlm->spinlock);  in dlm_pre_master_reco_lockres()
2203 			mlog(0, "%s: reco lock master is %u\n", dlm->name,  in dlm_pre_master_reco_lockres()
2215 int dlm_drop_lockres_ref(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)  in dlm_drop_lockres_ref()  argument
2227 	deref.node_idx = dlm->node_num;  in dlm_drop_lockres_ref()
2231 	ret = o2net_send_message(DLM_DEREF_LOCKRES_MSG, dlm->key,  in dlm_drop_lockres_ref()
2235 		     dlm->name, namelen, lockname, ret, res->owner);  in dlm_drop_lockres_ref()
2239 		     dlm->name, namelen, lockname, res->owner, r);  in dlm_drop_lockres_ref()
2252 	struct dlm_ctxt *dlm = data;  in dlm_deref_lockres_handler()  local
2264 	if (!dlm_grab(dlm))  in dlm_deref_lockres_handler()
2282 	spin_lock(&dlm->spinlock);  in dlm_deref_lockres_handler()
2283 	res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);  in dlm_deref_lockres_handler()
2285 		spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_handler()
2287 		     dlm->name, namelen, name);  in dlm_deref_lockres_handler()
2290 	spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_handler()
2298 			dlm_lockres_clear_refmap_bit(dlm, res, node);  in dlm_deref_lockres_handler()
2306 			dlm_lockres_calc_usage(dlm, res);  in dlm_deref_lockres_handler()
2309 		     	"but it is already dropped!\n", dlm->name,  in dlm_deref_lockres_handler()
2324 	dlm_init_work_item(dlm, item, dlm_deref_lockres_worker, NULL);  in dlm_deref_lockres_handler()
2328 	spin_lock(&dlm->work_lock);  in dlm_deref_lockres_handler()
2329 	list_add_tail(&item->list, &dlm->work_list);  in dlm_deref_lockres_handler()
2330 	spin_unlock(&dlm->work_lock);  in dlm_deref_lockres_handler()
2332 	queue_work(dlm->dlm_worker, &dlm->dispatched_work);  in dlm_deref_lockres_handler()
2338 	dlm_put(dlm);  in dlm_deref_lockres_handler()
2346 	struct dlm_ctxt *dlm = data;  in dlm_deref_lockres_done_handler()  local
2356 	if (!dlm_grab(dlm))  in dlm_deref_lockres_done_handler()
2374 	spin_lock(&dlm->spinlock);  in dlm_deref_lockres_done_handler()
2375 	res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);  in dlm_deref_lockres_done_handler()
2377 		spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_done_handler()
2379 		     dlm->name, namelen, name);  in dlm_deref_lockres_done_handler()
2386 		spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_done_handler()
2388 			"but it is already derefed!\n", dlm->name,  in dlm_deref_lockres_done_handler()
2394 	__dlm_do_purge_lockres(dlm, res);  in dlm_deref_lockres_done_handler()
2398 	spin_unlock(&dlm->spinlock);  in dlm_deref_lockres_done_handler()
2404 	dlm_put(dlm);  in dlm_deref_lockres_done_handler()
2408 static void dlm_drop_lockres_ref_done(struct dlm_ctxt *dlm,  in dlm_drop_lockres_ref_done()  argument
2421 	deref.node_idx = dlm->node_num;  in dlm_drop_lockres_ref_done()
2425 	ret = o2net_send_message(DLM_DEREF_LOCKRES_DONE, dlm->key,  in dlm_drop_lockres_ref_done()
2429 				" to node %u\n", dlm->name, namelen,  in dlm_drop_lockres_ref_done()
2434 		     dlm->name, namelen, lockname, node, r);  in dlm_drop_lockres_ref_done()
2441 	struct dlm_ctxt *dlm;  in dlm_deref_lockres_worker()  local
2446 	dlm = item->dlm;  in dlm_deref_lockres_worker()
2454 		dlm_lockres_clear_refmap_bit(dlm, res, node);  in dlm_deref_lockres_worker()
2459 	dlm_drop_lockres_ref_done(dlm, res, node);  in dlm_deref_lockres_worker()
2463 		     dlm->name, res->lockname.len, res->lockname.name, node);  in dlm_deref_lockres_worker()
2464 		dlm_lockres_calc_usage(dlm, res);  in dlm_deref_lockres_worker()
2467 		     "but it is already dropped!\n", dlm->name,  in dlm_deref_lockres_worker()
2482 static int dlm_is_lockres_migratable(struct dlm_ctxt *dlm,  in dlm_is_lockres_migratable()  argument
2502 	if (res->owner != dlm->node_num)  in dlm_is_lockres_migratable()
2508 			if (lock->ml.node != dlm->node_num) {  in dlm_is_lockres_migratable()
2514 			     "%s list\n", dlm->name, res->lockname.len,  in dlm_is_lockres_migratable()
2529 	mlog(0, "%s: res %.*s, Migratable\n", dlm->name, res->lockname.len,  in dlm_is_lockres_migratable()
2540 static int dlm_migrate_lockres(struct dlm_ctxt *dlm,  in dlm_migrate_lockres()  argument
2552 	if (!dlm_grab(dlm))  in dlm_migrate_lockres()
2558 	mlog(0, "%s: Migrating %.*s to node %u\n", dlm->name, namelen, name,  in dlm_migrate_lockres()
2580 	spin_lock(&dlm->spinlock);  in dlm_migrate_lockres()
2581 	spin_lock(&dlm->master_lock);  in dlm_migrate_lockres()
2582 	ret = dlm_add_migration_mle(dlm, res, mle, &oldmle, name,  in dlm_migrate_lockres()
2583 				    namelen, target, dlm->node_num);  in dlm_migrate_lockres()
2591 	spin_unlock(&dlm->master_lock);  in dlm_migrate_lockres()
2592 	spin_unlock(&dlm->spinlock);  in dlm_migrate_lockres()
2604 	if (dlm_mark_lockres_migrating(dlm, res, target) < 0) {  in dlm_migrate_lockres()
2618 		dlm_mle_detach_hb_events(dlm, oldmle);  in dlm_migrate_lockres()
2624 			dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2642 	flush_workqueue(dlm->dlm_worker);  in dlm_migrate_lockres()
2648 	ret = dlm_send_one_lockres(dlm, res, mres, target,  in dlm_migrate_lockres()
2655 		dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2663 			dlm_wait_for_node_death(dlm, target,  in dlm_migrate_lockres()
2691 			     dlm->name, res->lockname.len, res->lockname.name);  in dlm_migrate_lockres()
2694 			if (dlm_is_node_dead(dlm, target)) {  in dlm_migrate_lockres()
2697 				     dlm->name, res->lockname.len,  in dlm_migrate_lockres()
2701 				dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2712 			     dlm->name, res->lockname.len, res->lockname.name);  in dlm_migrate_lockres()
2717 	dlm_set_lockres_owner(dlm, res, target);  in dlm_migrate_lockres()
2719 	dlm_remove_nonlocal_locks(dlm, res);  in dlm_migrate_lockres()
2724 	dlm_mle_detach_hb_events(dlm, mle);  in dlm_migrate_lockres()
2728 	dlm_lockres_calc_usage(dlm, res);  in dlm_migrate_lockres()
2733 		dlm_kick_thread(dlm, res);  in dlm_migrate_lockres()
2743 	dlm_put(dlm);  in dlm_migrate_lockres()
2745 	mlog(0, "%s: Migrating %.*s to %u, returns %d\n", dlm->name, namelen,  in dlm_migrate_lockres()
2760 int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)  in dlm_empty_lockres()  argument
2761 	__must_hold(&dlm->spinlock)  in dlm_empty_lockres()
2767 	assert_spin_locked(&dlm->spinlock);  in dlm_empty_lockres()
2770 	if (dlm_is_lockres_migratable(dlm, res))  in dlm_empty_lockres()
2771 		target = dlm_pick_migration_target(dlm, res);  in dlm_empty_lockres()
2778 	spin_unlock(&dlm->spinlock);  in dlm_empty_lockres()
2780 	ret = dlm_migrate_lockres(dlm, res, target);  in dlm_empty_lockres()
2783 		     dlm->name, res->lockname.len, res->lockname.name,  in dlm_empty_lockres()
2785 	spin_lock(&dlm->spinlock);  in dlm_empty_lockres()
2790 int dlm_lock_basts_flushed(struct dlm_ctxt *dlm, struct dlm_lock *lock)  in dlm_lock_basts_flushed()  argument
2793 	spin_lock(&dlm->ast_lock);  in dlm_lock_basts_flushed()
2797 	spin_unlock(&dlm->ast_lock);  in dlm_lock_basts_flushed()
2801 static int dlm_migration_can_proceed(struct dlm_ctxt *dlm,  in dlm_migration_can_proceed()  argument
2812 	spin_lock(&dlm->spinlock);  in dlm_migration_can_proceed()
2813 	if (!test_bit(mig_target, dlm->domain_map))  in dlm_migration_can_proceed()
2815 	spin_unlock(&dlm->spinlock);  in dlm_migration_can_proceed()
2819 static int dlm_lockres_is_dirty(struct dlm_ctxt *dlm,  in dlm_lockres_is_dirty()  argument
2830 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,  in dlm_mark_lockres_migrating()  argument
2837 	       res->lockname.len, res->lockname.name, dlm->node_num,  in dlm_mark_lockres_migrating()
2850 	dlm_kick_thread(dlm, res);  in dlm_mark_lockres_migrating()
2858 	wait_event(dlm->ast_wq, !dlm_lockres_is_dirty(dlm, res));  in dlm_mark_lockres_migrating()
2859 	dlm_lockres_release_ast(dlm, res);  in dlm_mark_lockres_migrating()
2867 	ret = wait_event_interruptible_timeout(dlm->migration_wq,  in dlm_mark_lockres_migrating()
2868 		   dlm_migration_can_proceed(dlm, res, target),  in dlm_mark_lockres_migrating()
2873 		       test_bit(target, dlm->domain_map) ? "no":"yes");  in dlm_mark_lockres_migrating()
2877 		       test_bit(target, dlm->domain_map) ? "no":"yes");  in dlm_mark_lockres_migrating()
2879 	if (!dlm_migration_can_proceed(dlm, res, target)) {  in dlm_mark_lockres_migrating()
2886 	spin_lock(&dlm->spinlock);  in dlm_mark_lockres_migrating()
2887 	if (!test_bit(target, dlm->domain_map)) {  in dlm_mark_lockres_migrating()
2892 	spin_unlock(&dlm->spinlock);  in dlm_mark_lockres_migrating()
2922 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,  in dlm_remove_nonlocal_locks()  argument
2931 	BUG_ON(res->owner == dlm->node_num);  in dlm_remove_nonlocal_locks()
2935 			if (lock->ml.node != dlm->node_num) {  in dlm_remove_nonlocal_locks()
2943 				dlm_lockres_clear_refmap_bit(dlm, res,  in dlm_remove_nonlocal_locks()
2961 		if (bit != dlm->node_num) {  in dlm_remove_nonlocal_locks()
2963 			     "migrating lockres, clearing\n", dlm->name,  in dlm_remove_nonlocal_locks()
2965 			dlm_lockres_clear_refmap_bit(dlm, res, bit);  in dlm_remove_nonlocal_locks()
2976 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,  in dlm_pick_migration_target()  argument
2985 	assert_spin_locked(&dlm->spinlock);  in dlm_pick_migration_target()
2992 			if (lock->ml.node == dlm->node_num)  in dlm_pick_migration_target()
2994 			if (test_bit(lock->ml.node, dlm->exit_domain_map))  in dlm_pick_migration_target()
3008 		if (noderef == dlm->node_num)  in dlm_pick_migration_target()
3010 		if (test_bit(noderef, dlm->exit_domain_map))  in dlm_pick_migration_target()
3022 static int dlm_do_migrate_request(struct dlm_ctxt *dlm,  in dlm_do_migrate_request()  argument
3046 		spin_lock(&dlm->spinlock);  in dlm_do_migrate_request()
3047 		skip = (!test_bit(nodenum, dlm->domain_map));  in dlm_do_migrate_request()
3048 		spin_unlock(&dlm->spinlock);  in dlm_do_migrate_request()
3054 		ret = o2net_send_message(DLM_MIGRATE_REQUEST_MSG, dlm->key,  in dlm_do_migrate_request()
3059 			     "MIGRATE_REQUEST to node %u\n", dlm->name,  in dlm_do_migrate_request()
3076 			     dlm->name, res->lockname.len, res->lockname.name,  in dlm_do_migrate_request()
3079 			dlm_lockres_set_refmap_bit(dlm, res, nodenum);  in dlm_do_migrate_request()
3102 	struct dlm_ctxt *dlm = data;  in dlm_migrate_request_handler()  local
3110 	if (!dlm_grab(dlm))  in dlm_migrate_request_handler()
3126 	spin_lock(&dlm->spinlock);  in dlm_migrate_request_handler()
3127 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);  in dlm_migrate_request_handler()
3145 	spin_lock(&dlm->master_lock);  in dlm_migrate_request_handler()
3147 	ret = dlm_add_migration_mle(dlm, res, mle, &oldmle,  in dlm_migrate_request_handler()
3155 	spin_unlock(&dlm->master_lock);  in dlm_migrate_request_handler()
3157 	spin_unlock(&dlm->spinlock);  in dlm_migrate_request_handler()
3161 		dlm_mle_detach_hb_events(dlm, oldmle);  in dlm_migrate_request_handler()
3168 	dlm_put(dlm);  in dlm_migrate_request_handler()
3179 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,  in dlm_add_migration_mle()  argument
3191 	assert_spin_locked(&dlm->spinlock);  in dlm_add_migration_mle()
3192 	assert_spin_locked(&dlm->master_lock);  in dlm_add_migration_mle()
3195 	found = dlm_find_mle(dlm, oldmle, (char *)name, namelen);  in dlm_add_migration_mle()
3200 			if (master == dlm->node_num) {  in dlm_add_migration_mle()
3224 			__dlm_unlink_mle(dlm, tmp);  in dlm_add_migration_mle()
3225 			__dlm_mle_detach_hb_events(dlm, tmp);  in dlm_add_migration_mle()
3231 						"migration\n", dlm->name,  in dlm_add_migration_mle()
3240 	dlm_init_mle(mle, DLM_MLE_MIGRATION, dlm, res, name, namelen);  in dlm_add_migration_mle()
3247 	__dlm_insert_mle(dlm, mle);  in dlm_add_migration_mle()
3255 static struct dlm_lock_resource *dlm_reset_mleres_owner(struct dlm_ctxt *dlm,  in dlm_reset_mleres_owner()  argument
3261 	res = __dlm_lookup_lockres(dlm, mle->mname, mle->mnamelen,  in dlm_reset_mleres_owner()
3264 		spin_unlock(&dlm->master_lock);  in dlm_reset_mleres_owner()
3268 		dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);  in dlm_reset_mleres_owner()
3269 		dlm_move_lockres_to_recovery_list(dlm, res);  in dlm_reset_mleres_owner()
3274 		__dlm_mle_detach_hb_events(dlm, mle);  in dlm_reset_mleres_owner()
3277 		spin_lock(&dlm->master_lock);  in dlm_reset_mleres_owner()
3279 		spin_unlock(&dlm->master_lock);  in dlm_reset_mleres_owner()
3285 static void dlm_clean_migration_mle(struct dlm_ctxt *dlm,  in dlm_clean_migration_mle()  argument
3288 	__dlm_mle_detach_hb_events(dlm, mle);  in dlm_clean_migration_mle()
3291 	__dlm_unlink_mle(dlm, mle);  in dlm_clean_migration_mle()
3298 static void dlm_clean_block_mle(struct dlm_ctxt *dlm,  in dlm_clean_block_mle()  argument
3322 		__dlm_mle_detach_hb_events(dlm, mle);  in dlm_clean_block_mle()
3327 void dlm_clean_master_list(struct dlm_ctxt *dlm, u8 dead_node)  in dlm_clean_master_list()  argument
3335 	mlog(0, "dlm=%s, dead node=%u\n", dlm->name, dead_node);  in dlm_clean_master_list()
3337 	assert_spin_locked(&dlm->spinlock);  in dlm_clean_master_list()
3340 	spin_lock(&dlm->master_lock);  in dlm_clean_master_list()
3342 		bucket = dlm_master_hash(dlm, i);  in dlm_clean_master_list()
3358 				dlm_clean_block_mle(dlm, mle, dead_node);  in dlm_clean_master_list()
3381 						dlm->name, dead_node,  in dlm_clean_master_list()
3388 			dlm_clean_migration_mle(dlm, mle);  in dlm_clean_master_list()
3391 			     "%u to %u!\n", dlm->name, dead_node, mle->master,  in dlm_clean_master_list()
3399 			res = dlm_reset_mleres_owner(dlm, mle);  in dlm_clean_master_list()
3408 	spin_unlock(&dlm->master_lock);  in dlm_clean_master_list()
3411 int dlm_finish_migration(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,  in dlm_finish_migration()  argument
3417 	spin_lock(&dlm->spinlock);  in dlm_finish_migration()
3418 	dlm_node_iter_init(dlm->domain_map, &iter);  in dlm_finish_migration()
3420 	clear_bit(dlm->node_num, iter.node_map);  in dlm_finish_migration()
3421 	spin_unlock(&dlm->spinlock);  in dlm_finish_migration()
3427 	dlm_lockres_set_refmap_bit(dlm, res, old_master);  in dlm_finish_migration()
3431 	ret = dlm_do_migrate_request(dlm, res, old_master,  in dlm_finish_migration()
3432 				     dlm->node_num, &iter);  in dlm_finish_migration()
3442 	ret = dlm_do_assert_master(dlm, res, iter.node_map,  in dlm_finish_migration()
3454 	ret = dlm_do_assert_master(dlm, res, iter.node_map,  in dlm_finish_migration()
3466 	dlm_set_lockres_owner(dlm, res, dlm->node_num);  in dlm_finish_migration()
3470 	dlm_kick_thread(dlm, res);  in dlm_finish_migration()
3509 void dlm_lockres_release_ast(struct dlm_ctxt *dlm,  in dlm_lockres_release_ast()  argument
3525 	wake_up(&dlm->migration_wq);  in dlm_lockres_release_ast()
3528 void dlm_force_free_mles(struct dlm_ctxt *dlm)  in dlm_force_free_mles()  argument
3541 	spin_lock(&dlm->spinlock);  in dlm_force_free_mles()
3542 	spin_lock(&dlm->master_lock);  in dlm_force_free_mles()
3544 	BUG_ON(dlm->dlm_state != DLM_CTXT_LEAVING);  in dlm_force_free_mles()
3545 	BUG_ON((find_first_bit(dlm->domain_map, O2NM_MAX_NODES) < O2NM_MAX_NODES));  in dlm_force_free_mles()
3548 		bucket = dlm_master_hash(dlm, i);  in dlm_force_free_mles()
3557 			__dlm_unlink_mle(dlm, mle);  in dlm_force_free_mles()
3558 			__dlm_mle_detach_hb_events(dlm, mle);  in dlm_force_free_mles()
3562 	spin_unlock(&dlm->master_lock);  in dlm_force_free_mles()
3563 	spin_unlock(&dlm->spinlock);  in dlm_force_free_mles()