linux - fs/dlm/lock.c

6346 lines · 163.4 KiB

1	// SPDX-License-Identifier: GPL-2.0-only
2	/******************************************************************************
3	*******************************************************************************
4	**
5	** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved.
6	**
7	**
8	*******************************************************************************
9	******************************************************************************/
10
11	/* Central locking logic has four stages:
12
13	dlm_lock()
14	dlm_unlock()
15
16	request_lock(ls, lkb)
17	convert_lock(ls, lkb)
18	unlock_lock(ls, lkb)
19	cancel_lock(ls, lkb)
20
21	_request_lock(r, lkb)
22	_convert_lock(r, lkb)
23	_unlock_lock(r, lkb)
24	_cancel_lock(r, lkb)
25
26	do_request(r, lkb)
27	do_convert(r, lkb)
28	do_unlock(r, lkb)
29	do_cancel(r, lkb)
30
31	Stage 1 (lock, unlock) is mainly about checking input args and
32	splitting into one of the four main operations:
33
34	dlm_lock = request_lock
35	dlm_lock+CONVERT = convert_lock
36	dlm_unlock = unlock_lock
37	dlm_unlock+CANCEL = cancel_lock
38
39	Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
40	provided to the next stage.
41
42	Stage 3, _xxxx_lock(), determines if the operation is local or remote.
43	When remote, it calls send_xxxx(), when local it calls do_xxxx().
44
45	Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
46	given rsb and lkb and queues callbacks.
47
48	For remote operations, send_xxxx() results in the corresponding do_xxxx()
49	function being executed on the remote node. The connecting send/receive
50	calls on local (L) and remote (R) nodes:
51
52	L: send_xxxx() -> R: receive_xxxx()
53	R: do_xxxx()
54	L: receive_xxxx_reply() <- R: send_xxxx_reply()
55	*/
56	#include <trace/events/dlm.h>
57
58	#include <linux/types.h>
59	#include <linux/rbtree.h>
60	#include <linux/slab.h>
61	#include "dlm_internal.h"
62	#include <linux/dlm_device.h>
63	#include "memory.h"
64	#include "midcomms.h"
65	#include "requestqueue.h"
66	#include "util.h"
67	#include "dir.h"
68	#include "member.h"
69	#include "lockspace.h"
70	#include "ast.h"
71	#include "lock.h"
72	#include "rcom.h"
73	#include "recover.h"
74	#include "lvb_table.h"
75	#include "user.h"
76	#include "config.h"
77
78	static int send_request(struct dlm_rsb r, struct dlm_lkb lkb);
79	static int send_convert(struct dlm_rsb r, struct dlm_lkb lkb);
80	static int send_unlock(struct dlm_rsb r, struct dlm_lkb lkb);
81	static int send_cancel(struct dlm_rsb r, struct dlm_lkb lkb);
82	static int send_grant(struct dlm_rsb r, struct dlm_lkb lkb);
83	static int send_bast(struct dlm_rsb r, struct dlm_lkb lkb, int mode);
84	static int send_lookup(struct dlm_rsb r, struct dlm_lkb lkb);
85	static int send_remove(struct dlm_rsb *r);
86	static int _request_lock(struct dlm_rsb r, struct dlm_lkb lkb);
87	static int _cancel_lock(struct dlm_rsb r, struct dlm_lkb lkb);
88	static void __receive_convert_reply(struct dlm_rsb r, struct dlm_lkb lkb,
89	const struct dlm_message *ms, bool local);
90	static int receive_extralen(const struct dlm_message *ms);
91	static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
92	static void deactivate_rsb(struct kref *kref);
93
94	/*
95	* Lock compatibility matrix - thanks Steve
96	* UN = Unlocked state. Not really a state, used as a flag
97	* PD = Padding. Used to make the matrix a nice power of two in size
98	* Other states are the same as the VMS DLM.
99	* Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
100	*/
101
102	static const int __dlm_compat_matrix[8][8] = {
103	/* UN NL CR CW PR PW EX PD */
104	{1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
105	{1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
106	{1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
107	{1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
108	{1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
109	{1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
110	{1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
111	{0, 0, 0, 0, 0, 0, 0, 0} /* PD */
112	};
113
114	/*
115	* This defines the direction of transfer of LVB data.
116	* Granted mode is the row; requested mode is the column.
117	* Usage: matrix[grmode+1][rqmode+1]
118	* 1 = LVB is returned to the caller
119	* 0 = LVB is written to the resource
120	* -1 = nothing happens to the LVB
121	*/
122
123	const int dlm_lvb_operations[8][8] = {
124	/* UN NL CR CW PR PW EX PD*/
125	{ -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
126	{ -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
127	{ -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
128	{ -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
129	{ -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
130	{ -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
131	{ -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
132	{ -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
133	};
134
135	#define modes_compat(gr, rq) \
136	__dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
137
138	int dlm_modes_compat(int mode1, int mode2)
139	{
140	return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
141	}
142
143	/*
144	* Compatibility matrix for conversions with QUECVT set.
145	* Granted mode is the row; requested mode is the column.
146	* Usage: matrix[grmode+1][rqmode+1]
147	*/
148
149	static const int __quecvt_compat_matrix[8][8] = {
150	/* UN NL CR CW PR PW EX PD */
151	{0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
152	{0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
153	{0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
154	{0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
155	{0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
156	{0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
157	{0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
158	{0, 0, 0, 0, 0, 0, 0, 0} /* PD */
159	};
160
161	void dlm_print_lkb(struct dlm_lkb *lkb)
162	{
163	printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
164	"sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
165	lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
166	dlm_iflags_val(lkb), lkb->lkb_status, lkb->lkb_rqmode,
167	lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
168	(unsigned long long)lkb->lkb_recover_seq);
169	}
170
171	static void dlm_print_rsb(struct dlm_rsb *r)
172	{
173	printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
174	"rlc %d name %s\n",
175	r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
176	r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
177	r->res_name);
178	}
179
180	void dlm_dump_rsb(struct dlm_rsb *r)
181	{
182	struct dlm_lkb *lkb;
183
184	dlm_print_rsb(r);
185
186	printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
187	list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
188	printk(KERN_ERR "rsb lookup list\n");
189	list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
190	dlm_print_lkb(lkb);
191	printk(KERN_ERR "rsb grant queue:\n");
192	list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
193	dlm_print_lkb(lkb);
194	printk(KERN_ERR "rsb convert queue:\n");
195	list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
196	dlm_print_lkb(lkb);
197	printk(KERN_ERR "rsb wait queue:\n");
198	list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
199	dlm_print_lkb(lkb);
200	}
201
202	/* Threads cannot use the lockspace while it's being recovered */
203
204	void dlm_lock_recovery(struct dlm_ls *ls)
205	{
206	down_read(&ls->ls_in_recovery);
207	}
208
209	void dlm_unlock_recovery(struct dlm_ls *ls)
210	{
211	up_read(&ls->ls_in_recovery);
212	}
213
214	int dlm_lock_recovery_try(struct dlm_ls *ls)
215	{
216	return down_read_trylock(&ls->ls_in_recovery);
217	}
218
219	static inline int can_be_queued(struct dlm_lkb *lkb)
220	{
221	return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
222	}
223
224	static inline int force_blocking_asts(struct dlm_lkb *lkb)
225	{
226	return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
227	}
228
229	static inline int is_demoted(struct dlm_lkb *lkb)
230	{
231	return test_bit(DLM_SBF_DEMOTED_BIT, &lkb->lkb_sbflags);
232	}
233
234	static inline int is_altmode(struct dlm_lkb *lkb)
235	{
236	return test_bit(DLM_SBF_ALTMODE_BIT, &lkb->lkb_sbflags);
237	}
238
239	static inline int is_granted(struct dlm_lkb *lkb)
240	{
241	return (lkb->lkb_status == DLM_LKSTS_GRANTED);
242	}
243
244	static inline int is_remote(struct dlm_rsb *r)
245	{
246	DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
247	return !!r->res_nodeid;
248	}
249
250	static inline int is_process_copy(struct dlm_lkb *lkb)
251	{
252	return lkb->lkb_nodeid &&
253	!test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags);
254	}
255
256	static inline int is_master_copy(struct dlm_lkb *lkb)
257	{
258	return test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags);
259	}
260
261	static inline int middle_conversion(struct dlm_lkb *lkb)
262	{
263	if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) \|\|
264	(lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
265	return 1;
266	return 0;
267	}
268
269	static inline int down_conversion(struct dlm_lkb *lkb)
270	{
271	return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
272	}
273
274	static inline int is_overlap_unlock(struct dlm_lkb *lkb)
275	{
276	return test_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
277	}
278
279	static inline int is_overlap_cancel(struct dlm_lkb *lkb)
280	{
281	return test_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
282	}
283
284	static inline int is_overlap(struct dlm_lkb *lkb)
285	{
286	return test_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags) \|\|
287	test_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
288	}
289
290	static void queue_cast(struct dlm_rsb r, struct dlm_lkb lkb, int rv)
291	{
292	if (is_master_copy(lkb))
293	return;
294
295	DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
296
297	if (rv == -DLM_ECANCEL &&
298	test_and_clear_bit(DLM_IFL_DEADLOCK_CANCEL_BIT, &lkb->lkb_iflags))
299	rv = -EDEADLK;
300
301	dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, dlm_sbflags_val(lkb));
302	}
303
304	static inline void queue_cast_overlap(struct dlm_rsb r, struct dlm_lkb lkb)
305	{
306	queue_cast(r, lkb,
307	is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
308	}
309
310	static void queue_bast(struct dlm_rsb r, struct dlm_lkb lkb, int rqmode)
311	{
312	if (is_master_copy(lkb)) {
313	send_bast(r, lkb, rqmode);
314	} else {
315	dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
316	}
317	}
318
319	/*
320	* Basic operations on rsb's and lkb's
321	*/
322
323	static inline unsigned long rsb_toss_jiffies(void)
324	{
325	return jiffies + (READ_ONCE(dlm_config.ci_toss_secs) * HZ);
326	}
327
328	/* This is only called to add a reference when the code already holds
329	a valid reference to the rsb, so there's no need for locking. */
330
331	static inline void hold_rsb(struct dlm_rsb *r)
332	{
333	/* inactive rsbs are not ref counted */
334	WARN_ON(rsb_flag(r, RSB_INACTIVE));
335	kref_get(&r->res_ref);
336	}
337
338	void dlm_hold_rsb(struct dlm_rsb *r)
339	{
340	hold_rsb(r);
341	}
342
343	/* TODO move this to lib/refcount.c */
344	static __must_check bool
345	dlm_refcount_dec_and_write_lock_bh(refcount_t r, rwlock_t lock)
346	__cond_acquires(true, lock)
347	{
348	if (refcount_dec_not_one(r))
349	return false;
350
351	write_lock_bh(lock);
352	if (!refcount_dec_and_test(r)) {
353	write_unlock_bh(lock);
354	return false;
355	}
356
357	return true;
358	}
359
360	/* TODO move this to include/linux/kref.h */
361	static inline int dlm_kref_put_write_lock_bh(struct kref *kref,
362	void (release)(struct kref kref),
363	rwlock_t *lock)
364	{
365	if (dlm_refcount_dec_and_write_lock_bh(&kref->refcount, lock)) {
366	release(kref);
367	return 1;
368	}
369
370	return 0;
371	}
372
373	static void put_rsb(struct dlm_rsb *r)
374	{
375	struct dlm_ls *ls = r->res_ls;
376	int rv;
377
378	rv = dlm_kref_put_write_lock_bh(&r->res_ref, deactivate_rsb,
379	&ls->ls_rsbtbl_lock);
380	if (rv)
381	write_unlock_bh(&ls->ls_rsbtbl_lock);
382	}
383
384	void dlm_put_rsb(struct dlm_rsb *r)
385	{
386	put_rsb(r);
387	}
388
389	/* connected with timer_delete_sync() in dlm_ls_stop() to stop
390	* new timers when recovery is triggered and don't run them
391	* again until a resume_scan_timer() tries it again.
392	*/
393	static void enable_scan_timer(struct dlm_ls *ls, unsigned long jiffies)
394	{
395	if (!dlm_locking_stopped(ls))
396	mod_timer(&ls->ls_scan_timer, jiffies);
397	}
398
399	/* This function tries to resume the timer callback if a rsb
400	* is on the scan list and no timer is pending. It might that
401	* the first entry is on currently executed as timer callback
402	* but we don't care if a timer queued up again and does
403	* nothing. Should be a rare case.
404	*/
405	void resume_scan_timer(struct dlm_ls *ls)
406	{
407	struct dlm_rsb *r;
408
409	spin_lock_bh(&ls->ls_scan_lock);
410	r = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb,
411	res_scan_list);
412	if (r && !timer_pending(&ls->ls_scan_timer))
413	enable_scan_timer(ls, r->res_toss_time);
414	spin_unlock_bh(&ls->ls_scan_lock);
415	}
416
417	/* ls_rsbtbl_lock must be held */
418
419	static void del_scan(struct dlm_ls ls, struct dlm_rsb r)
420	{
421	struct dlm_rsb *first;
422
423	/* active rsbs should never be on the scan list */
424	WARN_ON(!rsb_flag(r, RSB_INACTIVE));
425
426	spin_lock_bh(&ls->ls_scan_lock);
427	r->res_toss_time = 0;
428
429	/* if the rsb is not queued do nothing */
430	if (list_empty(&r->res_scan_list))
431	goto out;
432
433	/* get the first element before delete */
434	first = list_first_entry(&ls->ls_scan_list, struct dlm_rsb,
435	res_scan_list);
436	list_del_init(&r->res_scan_list);
437	/* check if the first element was the rsb we deleted */
438	if (first == r) {
439	/* try to get the new first element, if the list
440	* is empty now try to delete the timer, if we are
441	* too late we don't care.
442	*
443	* if the list isn't empty and a new first element got
444	* in place, set the new timer expire time.
445	*/
446	first = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb,
447	res_scan_list);
448	if (!first)
449	timer_delete(&ls->ls_scan_timer);
450	else
451	enable_scan_timer(ls, first->res_toss_time);
452	}
453
454	out:
455	spin_unlock_bh(&ls->ls_scan_lock);
456	}
457
458	static void add_scan(struct dlm_ls ls, struct dlm_rsb r)
459	{
460	int our_nodeid = dlm_our_nodeid();
461	struct dlm_rsb *first;
462
463	/* A dir record for a remote master rsb should never be on the scan list. */
464	WARN_ON(!dlm_no_directory(ls) &&
465	(r->res_master_nodeid != our_nodeid) &&
466	(dlm_dir_nodeid(r) == our_nodeid));
467
468	/* An active rsb should never be on the scan list. */
469	WARN_ON(!rsb_flag(r, RSB_INACTIVE));
470
471	/* An rsb should not already be on the scan list. */
472	WARN_ON(!list_empty(&r->res_scan_list));
473
474	spin_lock_bh(&ls->ls_scan_lock);
475	/* set the new rsb absolute expire time in the rsb */
476	r->res_toss_time = rsb_toss_jiffies();
477	if (list_empty(&ls->ls_scan_list)) {
478	/* if the queue is empty add the element and it's
479	* our new expire time
480	*/
481	list_add_tail(&r->res_scan_list, &ls->ls_scan_list);
482	enable_scan_timer(ls, r->res_toss_time);
483	} else {
484	/* try to get the maybe new first element and then add
485	* to this rsb with the oldest expire time to the end
486	* of the queue. If the list was empty before this
487	* rsb expire time is our next expiration if it wasn't
488	* the now new first elemet is our new expiration time
489	*/
490	first = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb,
491	res_scan_list);
492	list_add_tail(&r->res_scan_list, &ls->ls_scan_list);
493	if (!first)
494	enable_scan_timer(ls, r->res_toss_time);
495	else
496	enable_scan_timer(ls, first->res_toss_time);
497	}
498	spin_unlock_bh(&ls->ls_scan_lock);
499	}
500
501	/* if we hit contention we do in 250 ms a retry to trylock.
502	* if there is any other mod_timer in between we don't care
503	* about that it expires earlier again this is only for the
504	* unlikely case nothing happened in this time.
505	*/
506	#define DLM_TOSS_TIMER_RETRY (jiffies + msecs_to_jiffies(250))
507
508	/* Called by lockspace scan_timer to free unused rsb's. */
509
510	void dlm_rsb_scan(struct timer_list *timer)
511	{
512	struct dlm_ls *ls = timer_container_of(ls, timer, ls_scan_timer);
513	int our_nodeid = dlm_our_nodeid();
514	struct dlm_rsb *r;
515	int rv;
516
517	while (1) {
518	/* interrupting point to leave iteration when
519	* recovery waits for timer_delete_sync(), recovery
520	* will take care to delete everything in scan list.
521	*/
522	if (dlm_locking_stopped(ls))
523	break;
524
525	rv = spin_trylock(&ls->ls_scan_lock);
526	if (!rv) {
527	/* rearm again try timer */
528	enable_scan_timer(ls, DLM_TOSS_TIMER_RETRY);
529	break;
530	}
531
532	r = list_first_entry_or_null(&ls->ls_scan_list, struct dlm_rsb,
533	res_scan_list);
534	if (!r) {
535	/* the next add_scan will enable the timer again */
536	spin_unlock(&ls->ls_scan_lock);
537	break;
538	}
539
540	/*
541	* If the first rsb is not yet expired, then stop because the
542	* list is sorted with nearest expiration first.
543	*/
544	if (time_before(jiffies, r->res_toss_time)) {
545	/* rearm with the next rsb to expire in the future */
546	enable_scan_timer(ls, r->res_toss_time);
547	spin_unlock(&ls->ls_scan_lock);
548	break;
549	}
550
551	/* in find_rsb_dir/nodir there is a reverse order of this
552	* lock, however this is only a trylock if we hit some
553	* possible contention we try it again.
554	*/
555	rv = write_trylock(&ls->ls_rsbtbl_lock);
556	if (!rv) {
557	spin_unlock(&ls->ls_scan_lock);
558	/* rearm again try timer */
559	enable_scan_timer(ls, DLM_TOSS_TIMER_RETRY);
560	break;
561	}
562
563	list_del(&r->res_slow_list);
564	rhashtable_remove_fast(&ls->ls_rsbtbl, &r->res_node,
565	dlm_rhash_rsb_params);
566	rsb_clear_flag(r, RSB_HASHED);
567
568	/* ls_rsbtbl_lock is not needed when calling send_remove() */
569	write_unlock(&ls->ls_rsbtbl_lock);
570
571	list_del_init(&r->res_scan_list);
572	spin_unlock(&ls->ls_scan_lock);
573
574	/* An rsb that is a dir record for a remote master rsb
575	* cannot be removed, and should not have a timer enabled.
576	*/
577	WARN_ON(!dlm_no_directory(ls) &&
578	(r->res_master_nodeid != our_nodeid) &&
579	(dlm_dir_nodeid(r) == our_nodeid));
580
581	/* We're the master of this rsb but we're not
582	* the directory record, so we need to tell the
583	* dir node to remove the dir record
584	*/
585	if (!dlm_no_directory(ls) &&
586	(r->res_master_nodeid == our_nodeid) &&
587	(dlm_dir_nodeid(r) != our_nodeid))
588	send_remove(r);
589
590	free_inactive_rsb(r);
591	}
592	}
593
594	/* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
595	unlock any spinlocks, go back and call pre_rsb_struct again.
596	Otherwise, take an rsb off the list and return it. */
597
598	static int get_rsb_struct(struct dlm_ls ls, const void name, int len,
599	struct dlm_rsb **r_ret)
600	{
601	struct dlm_rsb *r;
602
603	r = dlm_allocate_rsb();
604	if (!r)
605	return -ENOMEM;
606
607	r->res_ls = ls;
608	r->res_length = len;
609	memcpy(r->res_name, name, len);
610	spin_lock_init(&r->res_lock);
611
612	INIT_LIST_HEAD(&r->res_lookup);
613	INIT_LIST_HEAD(&r->res_grantqueue);
614	INIT_LIST_HEAD(&r->res_convertqueue);
615	INIT_LIST_HEAD(&r->res_waitqueue);
616	INIT_LIST_HEAD(&r->res_root_list);
617	INIT_LIST_HEAD(&r->res_scan_list);
618	INIT_LIST_HEAD(&r->res_recover_list);
619	INIT_LIST_HEAD(&r->res_masters_list);
620
621	*r_ret = r;
622	return 0;
623	}
624
625	int dlm_search_rsb_tree(struct rhashtable rhash, const void name,
626	unsigned int len, struct dlm_rsb **r_ret)
627	{
628	struct dlm_rsb_key key = {
629	.len = len,
630	};
631
632	if (len > DLM_RESNAME_MAXLEN)
633	return -EINVAL;
634
635	memcpy(key.name, name, len);
636	*r_ret = rhashtable_lookup_fast(rhash, &key, dlm_rhash_rsb_params);
637	if (*r_ret)
638	return 0;
639
640	return -EBADR;
641	}
642
643	static int rsb_insert(struct dlm_rsb rsb, struct rhashtable rhash)
644	{
645	int rv;
646
647	rv = rhashtable_insert_fast(rhash, &rsb->res_node,
648	dlm_rhash_rsb_params);
649	if (!rv)
650	rsb_set_flag(rsb, RSB_HASHED);
651
652	return rv;
653	}
654
655	/*
656	* Find rsb in rsbtbl and potentially create/add one
657	*
658	* Delaying the release of rsb's has a similar benefit to applications keeping
659	* NL locks on an rsb, but without the guarantee that the cached master value
660	* will still be valid when the rsb is reused. Apps aren't always smart enough
661	* to keep NL locks on an rsb that they may lock again shortly; this can lead
662	* to excessive master lookups and removals if we don't delay the release.
663	*
664	* Searching for an rsb means looking through both the normal list and toss
665	* list. When found on the toss list the rsb is moved to the normal list with
666	* ref count of 1; when found on normal list the ref count is incremented.
667	*
668	* rsb's on the keep list are being used locally and refcounted.
669	* rsb's on the toss list are not being used locally, and are not refcounted.
670	*
671	* The toss list rsb's were either
672	* - previously used locally but not any more (were on keep list, then
673	* moved to toss list when last refcount dropped)
674	* - created and put on toss list as a directory record for a lookup
675	* (we are the dir node for the res, but are not using the res right now,
676	* but some other node is)
677	*
678	* The purpose of find_rsb() is to return a refcounted rsb for local use.
679	* So, if the given rsb is on the toss list, it is moved to the keep list
680	* before being returned.
681	*
682	* deactivate_rsb() happens when all local usage of the rsb is done, i.e. no
683	* more refcounts exist, so the rsb is moved from the keep list to the
684	* toss list.
685	*
686	* rsb's on both keep and toss lists are used for doing a name to master
687	* lookups. rsb's that are in use locally (and being refcounted) are on
688	* the keep list, rsb's that are not in use locally (not refcounted) and
689	* only exist for name/master lookups are on the toss list.
690	*
691	* rsb's on the toss list who's dir_nodeid is not local can have stale
692	* name/master mappings. So, remote requests on such rsb's can potentially
693	* return with an error, which means the mapping is stale and needs to
694	* be updated with a new lookup. (The idea behind MASTER UNCERTAIN and
695	* first_lkid is to keep only a single outstanding request on an rsb
696	* while that rsb has a potentially stale master.)
697	*/
698
699	static int find_rsb_dir(struct dlm_ls ls, const void name, int len,
700	uint32_t hash, int dir_nodeid, int from_nodeid,
701	unsigned int flags, struct dlm_rsb **r_ret)
702	{
703	struct dlm_rsb *r = NULL;
704	int our_nodeid = dlm_our_nodeid();
705	int from_local = 0;
706	int from_other = 0;
707	int from_dir = 0;
708	int create = 0;
709	int error;
710
711	if (flags & R_RECEIVE_REQUEST) {
712	if (from_nodeid == dir_nodeid)
713	from_dir = 1;
714	else
715	from_other = 1;
716	} else if (flags & R_REQUEST) {
717	from_local = 1;
718	}
719
720	/*
721	* flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
722	* from_nodeid has sent us a lock in dlm_recover_locks, believing
723	* we're the new master. Our local recovery may not have set
724	* res_master_nodeid to our_nodeid yet, so allow either. Don't
725	* create the rsb; dlm_recover_process_copy() will handle EBADR
726	* by resending.
727	*
728	* If someone sends us a request, we are the dir node, and we do
729	* not find the rsb anywhere, then recreate it. This happens if
730	* someone sends us a request after we have removed/freed an rsb.
731	* (They sent a request instead of lookup because they are using
732	* an rsb taken from their scan list.)
733	*/
734
735	if (from_local \|\| from_dir \|\|
736	(from_other && (dir_nodeid == our_nodeid))) {
737	create = 1;
738	}
739
740	retry:
741	error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
742	if (error)
743	goto do_new;
744
745	/* check if the rsb is active under read lock - likely path */
746	read_lock_bh(&ls->ls_rsbtbl_lock);
747	if (!rsb_flag(r, RSB_HASHED)) {
748	read_unlock_bh(&ls->ls_rsbtbl_lock);
749	error = -EBADR;
750	goto do_new;
751	}
752
753	/*
754	* rsb is active, so we can't check master_nodeid without lock_rsb.
755	*/
756
757	if (rsb_flag(r, RSB_INACTIVE)) {
758	read_unlock_bh(&ls->ls_rsbtbl_lock);
759	goto do_inactive;
760	}
761
762	kref_get(&r->res_ref);
763	read_unlock_bh(&ls->ls_rsbtbl_lock);
764	goto out;
765
766
767	do_inactive:
768	write_lock_bh(&ls->ls_rsbtbl_lock);
769
770	/*
771	* The expectation here is that the rsb will have HASHED and
772	* INACTIVE flags set, and that the rsb can be moved from
773	* inactive back to active again. However, between releasing
774	* the read lock and acquiring the write lock, this rsb could
775	* have been removed from rsbtbl, and had HASHED cleared, to
776	* be freed. To deal with this case, we would normally need
777	* to repeat dlm_search_rsb_tree while holding the write lock,
778	* but rcu allows us to simply check the HASHED flag, because
779	* the rcu read lock means the rsb will not be freed yet.
780	* If the HASHED flag is not set, then the rsb is being freed,
781	* so we add a new rsb struct. If the HASHED flag is set,
782	* and INACTIVE is not set, it means another thread has
783	* made the rsb active, as we're expecting to do here, and
784	* we just repeat the lookup (this will be very unlikely.)
785	*/
786	if (rsb_flag(r, RSB_HASHED)) {
787	if (!rsb_flag(r, RSB_INACTIVE)) {
788	write_unlock_bh(&ls->ls_rsbtbl_lock);
789	goto retry;
790	}
791	} else {
792	write_unlock_bh(&ls->ls_rsbtbl_lock);
793	error = -EBADR;
794	goto do_new;
795	}
796
797	/*
798	* rsb found inactive (master_nodeid may be out of date unless
799	* we are the dir_nodeid or were the master) No other thread
800	* is using this rsb because it's inactive, so we can
801	* look at or update res_master_nodeid without lock_rsb.
802	*/
803
804	if ((r->res_master_nodeid != our_nodeid) && from_other) {
805	/* our rsb was not master, and another node (not the dir node)
806	has sent us a request */
807	log_debug(ls, "find_rsb inactive from_other %d master %d dir %d %s",
808	from_nodeid, r->res_master_nodeid, dir_nodeid,
809	r->res_name);
810	write_unlock_bh(&ls->ls_rsbtbl_lock);
811	error = -ENOTBLK;
812	goto out;
813	}
814
815	if ((r->res_master_nodeid != our_nodeid) && from_dir) {
816	/* don't think this should ever happen */
817	log_error(ls, "find_rsb inactive from_dir %d master %d",
818	from_nodeid, r->res_master_nodeid);
819	dlm_print_rsb(r);
820	/* fix it and go on */
821	r->res_master_nodeid = our_nodeid;
822	r->res_nodeid = 0;
823	rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
824	r->res_first_lkid = 0;
825	}
826
827	if (from_local && (r->res_master_nodeid != our_nodeid)) {
828	/* Because we have held no locks on this rsb,
829	res_master_nodeid could have become stale. */
830	rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
831	r->res_first_lkid = 0;
832	}
833
834	/* we always deactivate scan timer for the rsb, when
835	* we move it out of the inactive state as rsb state
836	* can be changed and scan timers are only for inactive
837	* rsbs.
838	*/
839	del_scan(ls, r);
840	list_move(&r->res_slow_list, &ls->ls_slow_active);
841	rsb_clear_flag(r, RSB_INACTIVE);
842	kref_init(&r->res_ref); /* ref is now used in active state */
843	write_unlock_bh(&ls->ls_rsbtbl_lock);
844
845	goto out;
846
847
848	do_new:
849	/*
850	* rsb not found
851	*/
852
853	if (error == -EBADR && !create)
854	goto out;
855
856	error = get_rsb_struct(ls, name, len, &r);
857	if (WARN_ON_ONCE(error))
858	goto out;
859
860	r->res_hash = hash;
861	r->res_dir_nodeid = dir_nodeid;
862	kref_init(&r->res_ref);
863
864	if (from_dir) {
865	/* want to see how often this happens */
866	log_debug(ls, "find_rsb new from_dir %d recreate %s",
867	from_nodeid, r->res_name);
868	r->res_master_nodeid = our_nodeid;
869	r->res_nodeid = 0;
870	goto out_add;
871	}
872
873	if (from_other && (dir_nodeid != our_nodeid)) {
874	/* should never happen */
875	log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
876	from_nodeid, dir_nodeid, our_nodeid, r->res_name);
877	dlm_free_rsb(r);
878	r = NULL;
879	error = -ENOTBLK;
880	goto out;
881	}
882
883	if (from_other) {
884	log_debug(ls, "find_rsb new from_other %d dir %d %s",
885	from_nodeid, dir_nodeid, r->res_name);
886	}
887
888	if (dir_nodeid == our_nodeid) {
889	/* When we are the dir nodeid, we can set the master
890	node immediately */
891	r->res_master_nodeid = our_nodeid;
892	r->res_nodeid = 0;
893	} else {
894	/* set_master will send_lookup to dir_nodeid */
895	r->res_master_nodeid = 0;
896	r->res_nodeid = -1;
897	}
898
899	out_add:
900
901	write_lock_bh(&ls->ls_rsbtbl_lock);
902	error = rsb_insert(r, &ls->ls_rsbtbl);
903	if (error == -EEXIST) {
904	/* somebody else was faster and it seems the
905	* rsb exists now, we do a whole relookup
906	*/
907	write_unlock_bh(&ls->ls_rsbtbl_lock);
908	dlm_free_rsb(r);
909	goto retry;
910	} else if (!error) {
911	list_add(&r->res_slow_list, &ls->ls_slow_active);
912	}
913	write_unlock_bh(&ls->ls_rsbtbl_lock);
914	out:
915	*r_ret = r;
916	return error;
917	}
918
919	/* During recovery, other nodes can send us new MSTCPY locks (from
920	dlm_recover_locks) before we've made ourself master (in
921	dlm_recover_masters). */
922
923	static int find_rsb_nodir(struct dlm_ls ls, const void name, int len,
924	uint32_t hash, int dir_nodeid, int from_nodeid,
925	unsigned int flags, struct dlm_rsb **r_ret)
926	{
927	struct dlm_rsb *r = NULL;
928	int our_nodeid = dlm_our_nodeid();
929	int recover = (flags & R_RECEIVE_RECOVER);
930	int error;
931
932	retry:
933	error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
934	if (error)
935	goto do_new;
936
937	/* check if the rsb is in active state under read lock - likely path */
938	read_lock_bh(&ls->ls_rsbtbl_lock);
939	if (!rsb_flag(r, RSB_HASHED)) {
940	read_unlock_bh(&ls->ls_rsbtbl_lock);
941	goto do_new;
942	}
943
944	if (rsb_flag(r, RSB_INACTIVE)) {
945	read_unlock_bh(&ls->ls_rsbtbl_lock);
946	goto do_inactive;
947	}
948
949	/*
950	* rsb is active, so we can't check master_nodeid without lock_rsb.
951	*/
952
953	kref_get(&r->res_ref);
954	read_unlock_bh(&ls->ls_rsbtbl_lock);
955
956	goto out;
957
958
959	do_inactive:
960	write_lock_bh(&ls->ls_rsbtbl_lock);
961
962	/* See comment in find_rsb_dir. */
963	if (rsb_flag(r, RSB_HASHED)) {
964	if (!rsb_flag(r, RSB_INACTIVE)) {
965	write_unlock_bh(&ls->ls_rsbtbl_lock);
966	goto retry;
967	}
968	} else {
969	write_unlock_bh(&ls->ls_rsbtbl_lock);
970	goto do_new;
971	}
972
973
974	/*
975	* rsb found inactive. No other thread is using this rsb because
976	* it's inactive, so we can look at or update res_master_nodeid
977	* without lock_rsb.
978	*/
979
980	if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
981	/* our rsb is not master, and another node has sent us a
982	request; this should never happen */
983	log_error(ls, "find_rsb inactive from_nodeid %d master %d dir %d",
984	from_nodeid, r->res_master_nodeid, dir_nodeid);
985	dlm_print_rsb(r);
986	write_unlock_bh(&ls->ls_rsbtbl_lock);
987	error = -ENOTBLK;
988	goto out;
989	}
990
991	if (!recover && (r->res_master_nodeid != our_nodeid) &&
992	(dir_nodeid == our_nodeid)) {
993	/* our rsb is not master, and we are dir; may as well fix it;
994	this should never happen */
995	log_error(ls, "find_rsb inactive our %d master %d dir %d",
996	our_nodeid, r->res_master_nodeid, dir_nodeid);
997	dlm_print_rsb(r);
998	r->res_master_nodeid = our_nodeid;
999	r->res_nodeid = 0;
1000	}
1001
1002	del_scan(ls, r);
1003	list_move(&r->res_slow_list, &ls->ls_slow_active);
1004	rsb_clear_flag(r, RSB_INACTIVE);
1005	kref_init(&r->res_ref);
1006	write_unlock_bh(&ls->ls_rsbtbl_lock);
1007
1008	goto out;
1009
1010
1011	do_new:
1012	/*
1013	* rsb not found
1014	*/
1015
1016	error = get_rsb_struct(ls, name, len, &r);
1017	if (WARN_ON_ONCE(error))
1018	goto out;
1019
1020	r->res_hash = hash;
1021	r->res_dir_nodeid = dir_nodeid;
1022	r->res_master_nodeid = dir_nodeid;
1023	r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
1024	kref_init(&r->res_ref);
1025
1026	write_lock_bh(&ls->ls_rsbtbl_lock);
1027	error = rsb_insert(r, &ls->ls_rsbtbl);
1028	if (error == -EEXIST) {
1029	/* somebody else was faster and it seems the
1030	* rsb exists now, we do a whole relookup
1031	*/
1032	write_unlock_bh(&ls->ls_rsbtbl_lock);
1033	dlm_free_rsb(r);
1034	goto retry;
1035	} else if (!error) {
1036	list_add(&r->res_slow_list, &ls->ls_slow_active);
1037	}
1038	write_unlock_bh(&ls->ls_rsbtbl_lock);
1039
1040	out:
1041	*r_ret = r;
1042	return error;
1043	}
1044
1045	/*
1046	* rsb rcu usage
1047	*
1048	* While rcu read lock is held, the rsb cannot be freed,
1049	* which allows a lookup optimization.
1050	*
1051	* Two threads are accessing the same rsb concurrently,
1052	* the first (A) is trying to use the rsb, the second (B)
1053	* is trying to free the rsb.
1054	*
1055	* thread A thread B
1056	* (trying to use rsb) (trying to free rsb)
1057	*
1058	* A1. rcu read lock
1059	* A2. rsbtbl read lock
1060	* A3. look up rsb in rsbtbl
1061	* A4. rsbtbl read unlock
1062	* B1. rsbtbl write lock
1063	* B2. look up rsb in rsbtbl
1064	* B3. remove rsb from rsbtbl
1065	* B4. clear rsb HASHED flag
1066	* B5. rsbtbl write unlock
1067	* B6. begin freeing rsb using rcu...
1068	*
1069	* (rsb is inactive, so try to make it active again)
1070	* A5. read rsb HASHED flag (safe because rsb is not freed yet)
1071	* A6. the rsb HASHED flag is not set, which it means the rsb
1072	* is being removed from rsbtbl and freed, so don't use it.
1073	* A7. rcu read unlock
1074	*
1075	* B7. ...finish freeing rsb using rcu
1076	* A8. create a new rsb
1077	*
1078	* Without the rcu optimization, steps A5-8 would need to do
1079	* an extra rsbtbl lookup:
1080	* A5. rsbtbl write lock
1081	* A6. look up rsb in rsbtbl, not found
1082	* A7. rsbtbl write unlock
1083	* A8. create a new rsb
1084	*/
1085
1086	static int find_rsb(struct dlm_ls ls, const void name, int len,
1087	int from_nodeid, unsigned int flags,
1088	struct dlm_rsb **r_ret)
1089	{
1090	int dir_nodeid;
1091	uint32_t hash;
1092	int rv;
1093
1094	if (len > DLM_RESNAME_MAXLEN)
1095	return -EINVAL;
1096
1097	hash = jhash(name, len, 0);
1098	dir_nodeid = dlm_hash2nodeid(ls, hash);
1099
1100	rcu_read_lock();
1101	if (dlm_no_directory(ls))
1102	rv = find_rsb_nodir(ls, name, len, hash, dir_nodeid,
1103	from_nodeid, flags, r_ret);
1104	else
1105	rv = find_rsb_dir(ls, name, len, hash, dir_nodeid,
1106	from_nodeid, flags, r_ret);
1107	rcu_read_unlock();
1108	return rv;
1109	}
1110
1111	/* we have received a request and found that res_master_nodeid != our_nodeid,
1112	so we need to return an error or make ourself the master */
1113
1114	static int validate_master_nodeid(struct dlm_ls ls, struct dlm_rsb r,
1115	int from_nodeid)
1116	{
1117	if (dlm_no_directory(ls)) {
1118	log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
1119	from_nodeid, r->res_master_nodeid,
1120	r->res_dir_nodeid);
1121	dlm_print_rsb(r);
1122	return -ENOTBLK;
1123	}
1124
1125	if (from_nodeid != r->res_dir_nodeid) {
1126	/* our rsb is not master, and another node (not the dir node)
1127	has sent us a request. this is much more common when our
1128	master_nodeid is zero, so limit debug to non-zero. */
1129
1130	if (r->res_master_nodeid) {
1131	log_debug(ls, "validate master from_other %d master %d "
1132	"dir %d first %x %s", from_nodeid,
1133	r->res_master_nodeid, r->res_dir_nodeid,
1134	r->res_first_lkid, r->res_name);
1135	}
1136	return -ENOTBLK;
1137	} else {
1138	/* our rsb is not master, but the dir nodeid has sent us a
1139	request; this could happen with master 0 / res_nodeid -1 */
1140
1141	if (r->res_master_nodeid) {
1142	log_error(ls, "validate master from_dir %d master %d "
1143	"first %x %s",
1144	from_nodeid, r->res_master_nodeid,
1145	r->res_first_lkid, r->res_name);
1146	}
1147
1148	r->res_master_nodeid = dlm_our_nodeid();
1149	r->res_nodeid = 0;
1150	return 0;
1151	}
1152	}
1153
1154	static void __dlm_master_lookup(struct dlm_ls ls, struct dlm_rsb r, int our_nodeid,
1155	int from_nodeid, bool is_inactive, unsigned int flags,
1156	int r_nodeid, int result)
1157	{
1158	int fix_master = (flags & DLM_LU_RECOVER_MASTER);
1159	int from_master = (flags & DLM_LU_RECOVER_DIR);
1160
1161	if (r->res_dir_nodeid != our_nodeid) {
1162	/* should not happen, but may as well fix it and carry on */
1163	log_error(ls, "%s res_dir %d our %d %s", __func__,
1164	r->res_dir_nodeid, our_nodeid, r->res_name);
1165	r->res_dir_nodeid = our_nodeid;
1166	}
1167
1168	if (fix_master && r->res_master_nodeid && dlm_is_removed(ls, r->res_master_nodeid)) {
1169	/* Recovery uses this function to set a new master when
1170	* the previous master failed. Setting NEW_MASTER will
1171	* force dlm_recover_masters to call recover_master on this
1172	* rsb even though the res_nodeid is no longer removed.
1173	*/
1174
1175	r->res_master_nodeid = from_nodeid;
1176	r->res_nodeid = from_nodeid;
1177	rsb_set_flag(r, RSB_NEW_MASTER);
1178
1179	if (is_inactive) {
1180	/* I don't think we should ever find it inactive. */
1181	log_error(ls, "%s fix_master inactive", __func__);
1182	dlm_dump_rsb(r);
1183	}
1184	}
1185
1186	if (from_master && (r->res_master_nodeid != from_nodeid)) {
1187	/* this will happen if from_nodeid became master during
1188	* a previous recovery cycle, and we aborted the previous
1189	* cycle before recovering this master value
1190	*/
1191
1192	log_limit(ls, "%s from_master %d master_nodeid %d res_nodeid %d first %x %s",
1193	__func__, from_nodeid, r->res_master_nodeid,
1194	r->res_nodeid, r->res_first_lkid, r->res_name);
1195
1196	if (r->res_master_nodeid == our_nodeid) {
1197	log_error(ls, "from_master %d our_master", from_nodeid);
1198	dlm_dump_rsb(r);
1199	goto ret_assign;
1200	}
1201
1202	r->res_master_nodeid = from_nodeid;
1203	r->res_nodeid = from_nodeid;
1204	rsb_set_flag(r, RSB_NEW_MASTER);
1205	}
1206
1207	if (!r->res_master_nodeid) {
1208	/* this will happen if recovery happens while we're looking
1209	* up the master for this rsb
1210	*/
1211
1212	log_debug(ls, "%s master 0 to %d first %x %s", __func__,
1213	from_nodeid, r->res_first_lkid, r->res_name);
1214	r->res_master_nodeid = from_nodeid;
1215	r->res_nodeid = from_nodeid;
1216	}
1217
1218	if (!from_master && !fix_master &&
1219	(r->res_master_nodeid == from_nodeid)) {
1220	/* this can happen when the master sends remove, the dir node
1221	* finds the rsb on the active list and ignores the remove,
1222	* and the former master sends a lookup
1223	*/
1224
1225	log_limit(ls, "%s from master %d flags %x first %x %s",
1226	__func__, from_nodeid, flags, r->res_first_lkid,
1227	r->res_name);
1228	}
1229
1230	ret_assign:
1231	*r_nodeid = r->res_master_nodeid;
1232	if (result)
1233	*result = DLM_LU_MATCH;
1234	}
1235
1236	/*
1237	* We're the dir node for this res and another node wants to know the
1238	* master nodeid. During normal operation (non recovery) this is only
1239	* called from receive_lookup(); master lookups when the local node is
1240	* the dir node are done by find_rsb().
1241	*
1242	* normal operation, we are the dir node for a resource
1243	* . _request_lock
1244	* . set_master
1245	* . send_lookup
1246	* . receive_lookup
1247	* . dlm_master_lookup flags 0
1248	*
1249	* recover directory, we are rebuilding dir for all resources
1250	* . dlm_recover_directory
1251	* . dlm_rcom_names
1252	* remote node sends back the rsb names it is master of and we are dir of
1253	* . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
1254	* we either create new rsb setting remote node as master, or find existing
1255	* rsb and set master to be the remote node.
1256	*
1257	* recover masters, we are finding the new master for resources
1258	* . dlm_recover_masters
1259	* . recover_master
1260	* . dlm_send_rcom_lookup
1261	* . receive_rcom_lookup
1262	* . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
1263	*/
1264
1265	static int _dlm_master_lookup(struct dlm_ls ls, int from_nodeid, const char name,
1266	int len, unsigned int flags, int r_nodeid, int result)
1267	{
1268	struct dlm_rsb *r = NULL;
1269	uint32_t hash;
1270	int our_nodeid = dlm_our_nodeid();
1271	int dir_nodeid, error;
1272
1273	if (len > DLM_RESNAME_MAXLEN)
1274	return -EINVAL;
1275
1276	if (from_nodeid == our_nodeid) {
1277	log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
1278	our_nodeid, flags);
1279	return -EINVAL;
1280	}
1281
1282	hash = jhash(name, len, 0);
1283	dir_nodeid = dlm_hash2nodeid(ls, hash);
1284	if (dir_nodeid != our_nodeid) {
1285	log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
1286	from_nodeid, dir_nodeid, our_nodeid, hash,
1287	ls->ls_num_nodes);
1288	*r_nodeid = -1;
1289	return -EINVAL;
1290	}
1291
1292	retry:
1293	error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
1294	if (error)
1295	goto not_found;
1296
1297	/* check if the rsb is active under read lock - likely path */
1298	read_lock_bh(&ls->ls_rsbtbl_lock);
1299	if (!rsb_flag(r, RSB_HASHED)) {
1300	read_unlock_bh(&ls->ls_rsbtbl_lock);
1301	goto not_found;
1302	}
1303
1304	if (rsb_flag(r, RSB_INACTIVE)) {
1305	read_unlock_bh(&ls->ls_rsbtbl_lock);
1306	goto do_inactive;
1307	}
1308
1309	/* because the rsb is active, we need to lock_rsb before
1310	* checking/changing re_master_nodeid
1311	*/
1312
1313	hold_rsb(r);
1314	read_unlock_bh(&ls->ls_rsbtbl_lock);
1315	lock_rsb(r);
1316
1317	__dlm_master_lookup(ls, r, our_nodeid, from_nodeid, false,
1318	flags, r_nodeid, result);
1319
1320	/* the rsb was active */
1321	unlock_rsb(r);
1322	put_rsb(r);
1323
1324	return 0;
1325
1326	do_inactive:
1327	/* unlikely path - check if still part of ls_rsbtbl */
1328	write_lock_bh(&ls->ls_rsbtbl_lock);
1329
1330	/* see comment in find_rsb_dir */
1331	if (rsb_flag(r, RSB_HASHED)) {
1332	if (!rsb_flag(r, RSB_INACTIVE)) {
1333	write_unlock_bh(&ls->ls_rsbtbl_lock);
1334	/* something as changed, very unlikely but
1335	* try again
1336	*/
1337	goto retry;
1338	}
1339	} else {
1340	write_unlock_bh(&ls->ls_rsbtbl_lock);
1341	goto not_found;
1342	}
1343
1344	/* because the rsb is inactive, it's not refcounted and lock_rsb
1345	is not used, but is protected by the rsbtbl lock */
1346
1347	__dlm_master_lookup(ls, r, our_nodeid, from_nodeid, true, flags,
1348	r_nodeid, result);
1349
1350	/* A dir record rsb should never be on scan list.
1351	* Except when we are the dir and master node.
1352	* This function should only be called by the dir
1353	* node.
1354	*/
1355	WARN_ON(!list_empty(&r->res_scan_list) &&
1356	r->res_master_nodeid != our_nodeid);
1357
1358	write_unlock_bh(&ls->ls_rsbtbl_lock);
1359
1360	return 0;
1361
1362	not_found:
1363	error = get_rsb_struct(ls, name, len, &r);
1364	if (WARN_ON_ONCE(error))
1365	goto out;
1366
1367	r->res_hash = hash;
1368	r->res_dir_nodeid = our_nodeid;
1369	r->res_master_nodeid = from_nodeid;
1370	r->res_nodeid = from_nodeid;
1371	rsb_set_flag(r, RSB_INACTIVE);
1372
1373	write_lock_bh(&ls->ls_rsbtbl_lock);
1374	error = rsb_insert(r, &ls->ls_rsbtbl);
1375	if (error == -EEXIST) {
1376	/* somebody else was faster and it seems the
1377	* rsb exists now, we do a whole relookup
1378	*/
1379	write_unlock_bh(&ls->ls_rsbtbl_lock);
1380	dlm_free_rsb(r);
1381	goto retry;
1382	} else if (error) {
1383	write_unlock_bh(&ls->ls_rsbtbl_lock);
1384	/* should never happen */
1385	dlm_free_rsb(r);
1386	goto retry;
1387	}
1388
1389	list_add(&r->res_slow_list, &ls->ls_slow_inactive);
1390	write_unlock_bh(&ls->ls_rsbtbl_lock);
1391
1392	if (result)
1393	*result = DLM_LU_ADD;
1394	*r_nodeid = from_nodeid;
1395	out:
1396	return error;
1397	}
1398
1399	int dlm_master_lookup(struct dlm_ls ls, int from_nodeid, const char name,
1400	int len, unsigned int flags, int r_nodeid, int result)
1401	{
1402	int rv;
1403	rcu_read_lock();
1404	rv = _dlm_master_lookup(ls, from_nodeid, name, len, flags, r_nodeid, result);
1405	rcu_read_unlock();
1406	return rv;
1407	}
1408
1409	static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1410	{
1411	struct dlm_rsb *r;
1412
1413	read_lock_bh(&ls->ls_rsbtbl_lock);
1414	list_for_each_entry(r, &ls->ls_slow_active, res_slow_list) {
1415	if (r->res_hash == hash)
1416	dlm_dump_rsb(r);
1417	}
1418	read_unlock_bh(&ls->ls_rsbtbl_lock);
1419	}
1420
1421	void dlm_dump_rsb_name(struct dlm_ls ls, const char name, int len)
1422	{
1423	struct dlm_rsb *r = NULL;
1424	int error;
1425
1426	rcu_read_lock();
1427	error = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
1428	if (error)
1429	goto out;
1430
1431	dlm_dump_rsb(r);
1432	out:
1433	rcu_read_unlock();
1434	}
1435
1436	static void deactivate_rsb(struct kref *kref)
1437	{
1438	struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1439	struct dlm_ls *ls = r->res_ls;
1440	int our_nodeid = dlm_our_nodeid();
1441
1442	DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1443	rsb_set_flag(r, RSB_INACTIVE);
1444	list_move(&r->res_slow_list, &ls->ls_slow_inactive);
1445
1446	/*
1447	* When the rsb becomes unused, there are two possibilities:
1448	* 1. Leave the inactive rsb in place (don't remove it).
1449	* 2. Add it to the scan list to be removed.
1450	*
1451	* 1 is done when the rsb is acting as the dir record
1452	* for a remotely mastered rsb. The rsb must be left
1453	* in place as an inactive rsb to act as the dir record.
1454	*
1455	* 2 is done when a) the rsb is not the master and not the
1456	* dir record, b) when the rsb is both the master and the
1457	* dir record, c) when the rsb is master but not dir record.
1458	*
1459	* (If no directory is used, the rsb can always be removed.)
1460	*/
1461	if (dlm_no_directory(ls) \|\|
1462	(r->res_master_nodeid == our_nodeid \|\|
1463	dlm_dir_nodeid(r) != our_nodeid))
1464	add_scan(ls, r);
1465
1466	if (r->res_lvbptr) {
1467	dlm_free_lvb(r->res_lvbptr);
1468	r->res_lvbptr = NULL;
1469	}
1470	}
1471
1472	void free_inactive_rsb(struct dlm_rsb *r)
1473	{
1474	WARN_ON_ONCE(!rsb_flag(r, RSB_INACTIVE));
1475
1476	DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1477	DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1478	DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1479	DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1480	DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1481	DLM_ASSERT(list_empty(&r->res_scan_list), dlm_dump_rsb(r););
1482	DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1483	DLM_ASSERT(list_empty(&r->res_masters_list), dlm_dump_rsb(r););
1484
1485	dlm_free_rsb(r);
1486	}
1487
1488	/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1489	The rsb must exist as long as any lkb's for it do. */
1490
1491	static void attach_lkb(struct dlm_rsb r, struct dlm_lkb lkb)
1492	{
1493	hold_rsb(r);
1494	lkb->lkb_resource = r;
1495	}
1496
1497	static void detach_lkb(struct dlm_lkb *lkb)
1498	{
1499	if (lkb->lkb_resource) {
1500	put_rsb(lkb->lkb_resource);
1501	lkb->lkb_resource = NULL;
1502	}
1503	}
1504
1505	static int _create_lkb(struct dlm_ls ls, struct dlm_lkb *lkb_ret,
1506	unsigned long start, unsigned long end)
1507	{
1508	struct xa_limit limit;
1509	struct dlm_lkb *lkb;
1510	int rv;
1511
1512	limit.max = end;
1513	limit.min = start;
1514
1515	lkb = dlm_allocate_lkb();
1516	if (!lkb)
1517	return -ENOMEM;
1518
1519	lkb->lkb_last_bast_cb_mode = DLM_LOCK_IV;
1520	lkb->lkb_last_cast_cb_mode = DLM_LOCK_IV;
1521	lkb->lkb_last_cb_mode = DLM_LOCK_IV;
1522	lkb->lkb_nodeid = -1;
1523	lkb->lkb_grmode = DLM_LOCK_IV;
1524	kref_init(&lkb->lkb_ref);
1525	INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1526	INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1527
1528	write_lock_bh(&ls->ls_lkbxa_lock);
1529	rv = xa_alloc(&ls->ls_lkbxa, &lkb->lkb_id, lkb, limit, GFP_ATOMIC);
1530	write_unlock_bh(&ls->ls_lkbxa_lock);
1531
1532	if (rv < 0) {
1533	log_error(ls, "create_lkb xa error %d", rv);
1534	dlm_free_lkb(lkb);
1535	return rv;
1536	}
1537
1538	*lkb_ret = lkb;
1539	return 0;
1540	}
1541
1542	static int create_lkb(struct dlm_ls ls, struct dlm_lkb *lkb_ret)
1543	{
1544	return _create_lkb(ls, lkb_ret, 1, ULONG_MAX);
1545	}
1546
1547	static int find_lkb(struct dlm_ls ls, uint32_t lkid, struct dlm_lkb *lkb_ret)
1548	{
1549	struct dlm_lkb *lkb;
1550
1551	rcu_read_lock();
1552	lkb = xa_load(&ls->ls_lkbxa, lkid);
1553	if (lkb) {
1554	/* check if lkb is still part of lkbxa under lkbxa_lock as
1555	* the lkb_ref is tight to the lkbxa data structure, see
1556	* __put_lkb().
1557	*/
1558	read_lock_bh(&ls->ls_lkbxa_lock);
1559	if (kref_read(&lkb->lkb_ref))
1560	kref_get(&lkb->lkb_ref);
1561	else
1562	lkb = NULL;
1563	read_unlock_bh(&ls->ls_lkbxa_lock);
1564	}
1565	rcu_read_unlock();
1566
1567	*lkb_ret = lkb;
1568	return lkb ? 0 : -ENOENT;
1569	}
1570
1571	static void kill_lkb(struct kref *kref)
1572	{
1573	struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1574
1575	/* All work is done after the return from kref_put() so we
1576	can release the write_lock before the detach_lkb */
1577
1578	DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1579	}
1580
1581	/* __put_lkb() is used when an lkb may not have an rsb attached to
1582	it so we need to provide the lockspace explicitly */
1583
1584	static int __put_lkb(struct dlm_ls ls, struct dlm_lkb lkb)
1585	{
1586	uint32_t lkid = lkb->lkb_id;
1587	int rv;
1588
1589	rv = dlm_kref_put_write_lock_bh(&lkb->lkb_ref, kill_lkb,
1590	&ls->ls_lkbxa_lock);
1591	if (rv) {
1592	xa_erase(&ls->ls_lkbxa, lkid);
1593	write_unlock_bh(&ls->ls_lkbxa_lock);
1594
1595	detach_lkb(lkb);
1596
1597	/* for local/process lkbs, lvbptr points to caller's lksb */
1598	if (lkb->lkb_lvbptr && is_master_copy(lkb))
1599	dlm_free_lvb(lkb->lkb_lvbptr);
1600	dlm_free_lkb(lkb);
1601	}
1602
1603	return rv;
1604	}
1605
1606	int dlm_put_lkb(struct dlm_lkb *lkb)
1607	{
1608	struct dlm_ls *ls;
1609
1610	DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1611	DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1612
1613	ls = lkb->lkb_resource->res_ls;
1614	return __put_lkb(ls, lkb);
1615	}
1616
1617	/* This is only called to add a reference when the code already holds
1618	a valid reference to the lkb, so there's no need for locking. */
1619
1620	static inline void hold_lkb(struct dlm_lkb *lkb)
1621	{
1622	kref_get(&lkb->lkb_ref);
1623	}
1624
1625	static void unhold_lkb_assert(struct kref *kref)
1626	{
1627	struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1628
1629	DLM_ASSERT(false, dlm_print_lkb(lkb););
1630	}
1631
1632	/* This is called when we need to remove a reference and are certain
1633	it's not the last ref. e.g. del_lkb is always called between a
1634	find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1635	put_lkb would work fine, but would involve unnecessary locking */
1636
1637	static inline void unhold_lkb(struct dlm_lkb *lkb)
1638	{
1639	kref_put(&lkb->lkb_ref, unhold_lkb_assert);
1640	}
1641
1642	static void lkb_add_ordered(struct list_head new, struct list_head head,
1643	int mode)
1644	{
1645	struct dlm_lkb lkb = NULL, iter;
1646
1647	list_for_each_entry(iter, head, lkb_statequeue)
1648	if (iter->lkb_rqmode < mode) {
1649	lkb = iter;
1650	list_add_tail(new, &iter->lkb_statequeue);
1651	break;
1652	}
1653
1654	if (!lkb)
1655	list_add_tail(new, head);
1656	}
1657
1658	/* add/remove lkb to rsb's grant/convert/wait queue */
1659
1660	static void add_lkb(struct dlm_rsb r, struct dlm_lkb lkb, int status)
1661	{
1662	kref_get(&lkb->lkb_ref);
1663
1664	DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1665
1666	lkb->lkb_timestamp = ktime_get();
1667
1668	lkb->lkb_status = status;
1669
1670	switch (status) {
1671	case DLM_LKSTS_WAITING:
1672	if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1673	list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1674	else
1675	list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1676	break;
1677	case DLM_LKSTS_GRANTED:
1678	/* convention says granted locks kept in order of grmode */
1679	lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1680	lkb->lkb_grmode);
1681	break;
1682	case DLM_LKSTS_CONVERT:
1683	if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1684	list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1685	else
1686	list_add_tail(&lkb->lkb_statequeue,
1687	&r->res_convertqueue);
1688	break;
1689	default:
1690	DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1691	}
1692	}
1693
1694	static void del_lkb(struct dlm_rsb r, struct dlm_lkb lkb)
1695	{
1696	lkb->lkb_status = 0;
1697	list_del(&lkb->lkb_statequeue);
1698	unhold_lkb(lkb);
1699	}
1700
1701	static void move_lkb(struct dlm_rsb r, struct dlm_lkb lkb, int sts)
1702	{
1703	del_lkb(r, lkb);
1704	add_lkb(r, lkb, sts);
1705	}
1706
1707	static int msg_reply_type(int mstype)
1708	{
1709	switch (mstype) {
1710	case DLM_MSG_REQUEST:
1711	return DLM_MSG_REQUEST_REPLY;
1712	case DLM_MSG_CONVERT:
1713	return DLM_MSG_CONVERT_REPLY;
1714	case DLM_MSG_UNLOCK:
1715	return DLM_MSG_UNLOCK_REPLY;
1716	case DLM_MSG_CANCEL:
1717	return DLM_MSG_CANCEL_REPLY;
1718	case DLM_MSG_LOOKUP:
1719	return DLM_MSG_LOOKUP_REPLY;
1720	}
1721	return -1;
1722	}
1723
1724	/* add/remove lkb from global waiters list of lkb's waiting for
1725	a reply from a remote node */
1726
1727	static void add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1728	{
1729	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1730
1731	spin_lock_bh(&ls->ls_waiters_lock);
1732	if (lkb->lkb_wait_type \|\| is_overlap_cancel(lkb)) {
1733	switch (mstype) {
1734	case DLM_MSG_UNLOCK:
1735	set_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
1736	break;
1737	case DLM_MSG_CANCEL:
1738	set_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
1739	break;
1740	default:
1741	/* should never happen as validate_lock_args() checks
1742	* on lkb_wait_type and validate_unlock_args() only
1743	* creates UNLOCK or CANCEL messages.
1744	*/
1745	WARN_ON_ONCE(1);
1746	goto out;
1747	}
1748	lkb->lkb_wait_count++;
1749	hold_lkb(lkb);
1750
1751	log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1752	lkb->lkb_id, lkb->lkb_wait_type, mstype,
1753	lkb->lkb_wait_count, dlm_iflags_val(lkb));
1754	goto out;
1755	}
1756
1757	DLM_ASSERT(!lkb->lkb_wait_count,
1758	dlm_print_lkb(lkb);
1759	printk("wait_count %d\n", lkb->lkb_wait_count););
1760
1761	lkb->lkb_wait_count++;
1762	lkb->lkb_wait_type = mstype;
1763	lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1764	hold_lkb(lkb);
1765	list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1766	out:
1767	spin_unlock_bh(&ls->ls_waiters_lock);
1768	}
1769
1770	/* We clear the RESEND flag because we might be taking an lkb off the waiters
1771	list as part of process_requestqueue (e.g. a lookup that has an optimized
1772	request reply on the requestqueue) between dlm_recover_waiters_pre() which
1773	set RESEND and dlm_recover_waiters_post() */
1774
1775	static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1776	const struct dlm_message *ms)
1777	{
1778	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1779	int overlap_done = 0;
1780
1781	if (mstype == DLM_MSG_UNLOCK_REPLY &&
1782	test_and_clear_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags)) {
1783	log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1784	overlap_done = 1;
1785	goto out_del;
1786	}
1787
1788	if (mstype == DLM_MSG_CANCEL_REPLY &&
1789	test_and_clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags)) {
1790	log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1791	overlap_done = 1;
1792	goto out_del;
1793	}
1794
1795	/* Cancel state was preemptively cleared by a successful convert,
1796	see next comment, nothing to do. */
1797
1798	if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1799	(lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1800	log_debug(ls, "remwait %x cancel_reply wait_type %d",
1801	lkb->lkb_id, lkb->lkb_wait_type);
1802	return -1;
1803	}
1804
1805	/* Remove for the convert reply, and premptively remove for the
1806	cancel reply. A convert has been granted while there's still
1807	an outstanding cancel on it (the cancel is moot and the result
1808	in the cancel reply should be 0). We preempt the cancel reply
1809	because the app gets the convert result and then can follow up
1810	with another op, like convert. This subsequent op would see the
1811	lingering state of the cancel and fail with -EBUSY. */
1812
1813	if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1814	(lkb->lkb_wait_type == DLM_MSG_CONVERT) && ms && !ms->m_result &&
1815	test_and_clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags)) {
1816	log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1817	lkb->lkb_id);
1818	lkb->lkb_wait_type = 0;
1819	lkb->lkb_wait_count--;
1820	unhold_lkb(lkb);
1821	goto out_del;
1822	}
1823
1824	/* N.B. type of reply may not always correspond to type of original
1825	msg due to lookup->request optimization, verify others? */
1826
1827	if (lkb->lkb_wait_type) {
1828	lkb->lkb_wait_type = 0;
1829	goto out_del;
1830	}
1831
1832	log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1833	lkb->lkb_id, ms ? le32_to_cpu(ms->m_header.h_nodeid) : 0,
1834	lkb->lkb_remid, mstype, dlm_iflags_val(lkb));
1835	return -1;
1836
1837	out_del:
1838	/* the force-unlock/cancel has completed and we haven't recvd a reply
1839	to the op that was in progress prior to the unlock/cancel; we
1840	give up on any reply to the earlier op. FIXME: not sure when/how
1841	this would happen */
1842
1843	if (overlap_done && lkb->lkb_wait_type) {
1844	log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1845	lkb->lkb_id, mstype, lkb->lkb_wait_type);
1846	lkb->lkb_wait_count--;
1847	unhold_lkb(lkb);
1848	lkb->lkb_wait_type = 0;
1849	}
1850
1851	DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1852
1853	clear_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags);
1854	lkb->lkb_wait_count--;
1855	if (!lkb->lkb_wait_count)
1856	list_del_init(&lkb->lkb_wait_reply);
1857	unhold_lkb(lkb);
1858	return 0;
1859	}
1860
1861	static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1862	{
1863	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1864	int error;
1865
1866	spin_lock_bh(&ls->ls_waiters_lock);
1867	error = _remove_from_waiters(lkb, mstype, NULL);
1868	spin_unlock_bh(&ls->ls_waiters_lock);
1869	return error;
1870	}
1871
1872	/* Handles situations where we might be processing a "fake" or "local" reply in
1873	* the recovery context which stops any locking activity. Only debugfs might
1874	* change the lockspace waiters but they will held the recovery lock to ensure
1875	* remove_from_waiters_ms() in local case will be the only user manipulating the
1876	* lockspace waiters in recovery context.
1877	*/
1878
1879	static int remove_from_waiters_ms(struct dlm_lkb *lkb,
1880	const struct dlm_message *ms, bool local)
1881	{
1882	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1883	int error;
1884
1885	if (!local)
1886	spin_lock_bh(&ls->ls_waiters_lock);
1887	else
1888	WARN_ON_ONCE(!rwsem_is_locked(&ls->ls_in_recovery) \|\|
1889	!dlm_locking_stopped(ls));
1890	error = _remove_from_waiters(lkb, le32_to_cpu(ms->m_type), ms);
1891	if (!local)
1892	spin_unlock_bh(&ls->ls_waiters_lock);
1893	return error;
1894	}
1895
1896	/* lkb is master or local copy */
1897
1898	static void set_lvb_lock(struct dlm_rsb r, struct dlm_lkb lkb)
1899	{
1900	int b, len = r->res_ls->ls_lvblen;
1901
1902	/* b=1 lvb returned to caller
1903	b=0 lvb written to rsb or invalidated
1904	b=-1 do nothing */
1905
1906	b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1907
1908	if (b == 1) {
1909	if (!lkb->lkb_lvbptr)
1910	return;
1911
1912	if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1913	return;
1914
1915	if (!r->res_lvbptr)
1916	return;
1917
1918	memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1919	lkb->lkb_lvbseq = r->res_lvbseq;
1920
1921	} else if (b == 0) {
1922	if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1923	rsb_set_flag(r, RSB_VALNOTVALID);
1924	return;
1925	}
1926
1927	if (!lkb->lkb_lvbptr)
1928	return;
1929
1930	if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1931	return;
1932
1933	if (!r->res_lvbptr)
1934	r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1935
1936	if (!r->res_lvbptr)
1937	return;
1938
1939	memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1940	r->res_lvbseq++;
1941	lkb->lkb_lvbseq = r->res_lvbseq;
1942	rsb_clear_flag(r, RSB_VALNOTVALID);
1943	}
1944
1945	if (rsb_flag(r, RSB_VALNOTVALID))
1946	set_bit(DLM_SBF_VALNOTVALID_BIT, &lkb->lkb_sbflags);
1947	}
1948
1949	static void set_lvb_unlock(struct dlm_rsb r, struct dlm_lkb lkb)
1950	{
1951	if (lkb->lkb_grmode < DLM_LOCK_PW)
1952	return;
1953
1954	if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1955	rsb_set_flag(r, RSB_VALNOTVALID);
1956	return;
1957	}
1958
1959	if (!lkb->lkb_lvbptr)
1960	return;
1961
1962	if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1963	return;
1964
1965	if (!r->res_lvbptr)
1966	r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1967
1968	if (!r->res_lvbptr)
1969	return;
1970
1971	memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1972	r->res_lvbseq++;
1973	rsb_clear_flag(r, RSB_VALNOTVALID);
1974	}
1975
1976	/* lkb is process copy (pc) */
1977
1978	static void set_lvb_lock_pc(struct dlm_rsb r, struct dlm_lkb lkb,
1979	const struct dlm_message *ms)
1980	{
1981	int b;
1982
1983	if (!lkb->lkb_lvbptr)
1984	return;
1985
1986	if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1987	return;
1988
1989	b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1990	if (b == 1) {
1991	int len = receive_extralen(ms);
1992	if (len > r->res_ls->ls_lvblen)
1993	len = r->res_ls->ls_lvblen;
1994	memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1995	lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
1996	}
1997	}
1998
1999	/* Manipulate lkb's on rsb's convert/granted/waiting queues
2000	remove_lock -- used for unlock, removes lkb from granted
2001	revert_lock -- used for cancel, moves lkb from convert to granted
2002	grant_lock -- used for request and convert, adds lkb to granted or
2003	moves lkb from convert or waiting to granted
2004
2005	Each of these is used for master or local copy lkb's. There is
2006	also a _pc() variation used to make the corresponding change on
2007	a process copy (pc) lkb. */
2008
2009	static void _remove_lock(struct dlm_rsb r, struct dlm_lkb lkb)
2010	{
2011	del_lkb(r, lkb);
2012	lkb->lkb_grmode = DLM_LOCK_IV;
2013	/* this unhold undoes the original ref from create_lkb()
2014	so this leads to the lkb being freed */
2015	unhold_lkb(lkb);
2016	}
2017
2018	static void remove_lock(struct dlm_rsb r, struct dlm_lkb lkb)
2019	{
2020	set_lvb_unlock(r, lkb);
2021	_remove_lock(r, lkb);
2022	}
2023
2024	static void remove_lock_pc(struct dlm_rsb r, struct dlm_lkb lkb)
2025	{
2026	_remove_lock(r, lkb);
2027	}
2028
2029	/* returns: 0 did nothing
2030	1 moved lock to granted
2031	-1 removed lock */
2032
2033	static int revert_lock(struct dlm_rsb r, struct dlm_lkb lkb)
2034	{
2035	int rv = 0;
2036
2037	lkb->lkb_rqmode = DLM_LOCK_IV;
2038
2039	switch (lkb->lkb_status) {
2040	case DLM_LKSTS_GRANTED:
2041	break;
2042	case DLM_LKSTS_CONVERT:
2043	move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2044	rv = 1;
2045	break;
2046	case DLM_LKSTS_WAITING:
2047	del_lkb(r, lkb);
2048	lkb->lkb_grmode = DLM_LOCK_IV;
2049	/* this unhold undoes the original ref from create_lkb()
2050	so this leads to the lkb being freed */
2051	unhold_lkb(lkb);
2052	rv = -1;
2053	break;
2054	default:
2055	log_print("invalid status for revert %d", lkb->lkb_status);
2056	}
2057	return rv;
2058	}
2059
2060	static int revert_lock_pc(struct dlm_rsb r, struct dlm_lkb lkb)
2061	{
2062	return revert_lock(r, lkb);
2063	}
2064
2065	static void _grant_lock(struct dlm_rsb r, struct dlm_lkb lkb)
2066	{
2067	if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2068	lkb->lkb_grmode = lkb->lkb_rqmode;
2069	if (lkb->lkb_status)
2070	move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2071	else
2072	add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2073	}
2074
2075	lkb->lkb_rqmode = DLM_LOCK_IV;
2076	lkb->lkb_highbast = 0;
2077	}
2078
2079	static void grant_lock(struct dlm_rsb r, struct dlm_lkb lkb)
2080	{
2081	set_lvb_lock(r, lkb);
2082	_grant_lock(r, lkb);
2083	}
2084
2085	static void grant_lock_pc(struct dlm_rsb r, struct dlm_lkb lkb,
2086	const struct dlm_message *ms)
2087	{
2088	set_lvb_lock_pc(r, lkb, ms);
2089	_grant_lock(r, lkb);
2090	}
2091
2092	/* called by grant_pending_locks() which means an async grant message must
2093	be sent to the requesting node in addition to granting the lock if the
2094	lkb belongs to a remote node. */
2095
2096	static void grant_lock_pending(struct dlm_rsb r, struct dlm_lkb lkb)
2097	{
2098	grant_lock(r, lkb);
2099	if (is_master_copy(lkb))
2100	send_grant(r, lkb);
2101	else
2102	queue_cast(r, lkb, 0);
2103	}
2104
2105	/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2106	change the granted/requested modes. We're munging things accordingly in
2107	the process copy.
2108	CONVDEADLK: our grmode may have been forced down to NL to resolve a
2109	conversion deadlock
2110	ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2111	compatible with other granted locks */
2112
2113	static void munge_demoted(struct dlm_lkb *lkb)
2114	{
2115	if (lkb->lkb_rqmode == DLM_LOCK_IV \|\| lkb->lkb_grmode == DLM_LOCK_IV) {
2116	log_print("munge_demoted %x invalid modes gr %d rq %d",
2117	lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2118	return;
2119	}
2120
2121	lkb->lkb_grmode = DLM_LOCK_NL;
2122	}
2123
2124	static void munge_altmode(struct dlm_lkb lkb, const struct dlm_message ms)
2125	{
2126	if (ms->m_type != cpu_to_le32(DLM_MSG_REQUEST_REPLY) &&
2127	ms->m_type != cpu_to_le32(DLM_MSG_GRANT)) {
2128	log_print("munge_altmode %x invalid reply type %d",
2129	lkb->lkb_id, le32_to_cpu(ms->m_type));
2130	return;
2131	}
2132
2133	if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2134	lkb->lkb_rqmode = DLM_LOCK_PR;
2135	else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2136	lkb->lkb_rqmode = DLM_LOCK_CW;
2137	else {
2138	log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2139	dlm_print_lkb(lkb);
2140	}
2141	}
2142
2143	static inline int first_in_list(struct dlm_lkb lkb, struct list_head head)
2144	{
2145	struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2146	lkb_statequeue);
2147	if (lkb->lkb_id == first->lkb_id)
2148	return 1;
2149
2150	return 0;
2151	}
2152
2153	/* Check if the given lkb conflicts with another lkb on the queue. */
2154
2155	static int queue_conflict(struct list_head head, struct dlm_lkb lkb)
2156	{
2157	struct dlm_lkb *this;
2158
2159	list_for_each_entry(this, head, lkb_statequeue) {
2160	if (this == lkb)
2161	continue;
2162	if (!modes_compat(this, lkb))
2163	return 1;
2164	}
2165	return 0;
2166	}
2167
2168	/*
2169	* "A conversion deadlock arises with a pair of lock requests in the converting
2170	* queue for one resource. The granted mode of each lock blocks the requested
2171	* mode of the other lock."
2172	*
2173	* Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2174	* convert queue from being granted, then deadlk/demote lkb.
2175	*
2176	* Example:
2177	* Granted Queue: empty
2178	* Convert Queue: NL->EX (first lock)
2179	* PR->EX (second lock)
2180	*
2181	* The first lock can't be granted because of the granted mode of the second
2182	* lock and the second lock can't be granted because it's not first in the
2183	* list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2184	* demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2185	* flag set and return DEMOTED in the lksb flags.
2186	*
2187	* Originally, this function detected conv-deadlk in a more limited scope:
2188	* - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2189	* - if lkb1 was the first entry in the queue (not just earlier), and was
2190	* blocked by the granted mode of lkb2, and there was nothing on the
2191	* granted queue preventing lkb1 from being granted immediately, i.e.
2192	* lkb2 was the only thing preventing lkb1 from being granted.
2193	*
2194	* That second condition meant we'd only say there was conv-deadlk if
2195	* resolving it (by demotion) would lead to the first lock on the convert
2196	* queue being granted right away. It allowed conversion deadlocks to exist
2197	* between locks on the convert queue while they couldn't be granted anyway.
2198	*
2199	* Now, we detect and take action on conversion deadlocks immediately when
2200	* they're created, even if they may not be immediately consequential. If
2201	* lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2202	* mode that would prevent lkb1's conversion from being granted, we do a
2203	* deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2204	* I think this means that the lkb_is_ahead condition below should always
2205	* be zero, i.e. there will never be conv-deadlk between two locks that are
2206	* both already on the convert queue.
2207	*/
2208
2209	static int conversion_deadlock_detect(struct dlm_rsb r, struct dlm_lkb lkb2)
2210	{
2211	struct dlm_lkb *lkb1;
2212	int lkb_is_ahead = 0;
2213
2214	list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2215	if (lkb1 == lkb2) {
2216	lkb_is_ahead = 1;
2217	continue;
2218	}
2219
2220	if (!lkb_is_ahead) {
2221	if (!modes_compat(lkb2, lkb1))
2222	return 1;
2223	} else {
2224	if (!modes_compat(lkb2, lkb1) &&
2225	!modes_compat(lkb1, lkb2))
2226	return 1;
2227	}
2228	}
2229	return 0;
2230	}
2231
2232	/*
2233	* Return 1 if the lock can be granted, 0 otherwise.
2234	* Also detect and resolve conversion deadlocks.
2235	*
2236	* lkb is the lock to be granted
2237	*
2238	* now is 1 if the function is being called in the context of the
2239	* immediate request, it is 0 if called later, after the lock has been
2240	* queued.
2241	*
2242	* recover is 1 if dlm_recover_grant() is trying to grant conversions
2243	* after recovery.
2244	*
2245	* References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2246	*/
2247
2248	static int _can_be_granted(struct dlm_rsb r, struct dlm_lkb lkb, int now,
2249	int recover)
2250	{
2251	int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2252
2253	/*
2254	* 6-10: Version 5.4 introduced an option to address the phenomenon of
2255	* a new request for a NL mode lock being blocked.
2256	*
2257	* 6-11: If the optional EXPEDITE flag is used with the new NL mode
2258	* request, then it would be granted. In essence, the use of this flag
2259	* tells the Lock Manager to expedite theis request by not considering
2260	* what may be in the CONVERTING or WAITING queues... As of this
2261	* writing, the EXPEDITE flag can be used only with new requests for NL
2262	* mode locks. This flag is not valid for conversion requests.
2263	*
2264	* A shortcut. Earlier checks return an error if EXPEDITE is used in a
2265	* conversion or used with a non-NL requested mode. We also know an
2266	* EXPEDITE request is always granted immediately, so now must always
2267	* be 1. The full condition to grant an expedite request: (now &&
2268	* !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2269	* therefore be shortened to just checking the flag.
2270	*/
2271
2272	if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2273	return 1;
2274
2275	/*
2276	* A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2277	* added to the remaining conditions.
2278	*/
2279
2280	if (queue_conflict(&r->res_grantqueue, lkb))
2281	return 0;
2282
2283	/*
2284	* 6-3: By default, a conversion request is immediately granted if the
2285	* requested mode is compatible with the modes of all other granted
2286	* locks
2287	*/
2288
2289	if (queue_conflict(&r->res_convertqueue, lkb))
2290	return 0;
2291
2292	/*
2293	* The RECOVER_GRANT flag means dlm_recover_grant() is granting
2294	* locks for a recovered rsb, on which lkb's have been rebuilt.
2295	* The lkb's may have been rebuilt on the queues in a different
2296	* order than they were in on the previous master. So, granting
2297	* queued conversions in order after recovery doesn't make sense
2298	* since the order hasn't been preserved anyway. The new order
2299	* could also have created a new "in place" conversion deadlock.
2300	* (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2301	* After recovery, there would be no granted locks, and possibly
2302	* NL->EX, PR->EX, an in-place conversion deadlock.) So, after
2303	* recovery, grant conversions without considering order.
2304	*/
2305
2306	if (conv && recover)
2307	return 1;
2308
2309	/*
2310	* 6-5: But the default algorithm for deciding whether to grant or
2311	* queue conversion requests does not by itself guarantee that such
2312	* requests are serviced on a "first come first serve" basis. This, in
2313	* turn, can lead to a phenomenon known as "indefinate postponement".
2314	*
2315	* 6-7: This issue is dealt with by using the optional QUECVT flag with
2316	* the system service employed to request a lock conversion. This flag
2317	* forces certain conversion requests to be queued, even if they are
2318	* compatible with the granted modes of other locks on the same
2319	* resource. Thus, the use of this flag results in conversion requests
2320	* being ordered on a "first come first servce" basis.
2321	*
2322	* DCT: This condition is all about new conversions being able to occur
2323	* "in place" while the lock remains on the granted queue (assuming
2324	* nothing else conflicts.) IOW if QUECVT isn't set, a conversion
2325	* doesn't _have_ to go onto the convert queue where it's processed in
2326	* order. The "now" variable is necessary to distinguish converts
2327	* being received and processed for the first time now, because once a
2328	* convert is moved to the conversion queue the condition below applies
2329	* requiring fifo granting.
2330	*/
2331
2332	if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2333	return 1;
2334
2335	/*
2336	* Even if the convert is compat with all granted locks,
2337	* QUECVT forces it behind other locks on the convert queue.
2338	*/
2339
2340	if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2341	if (list_empty(&r->res_convertqueue))
2342	return 1;
2343	else
2344	return 0;
2345	}
2346
2347	/*
2348	* The NOORDER flag is set to avoid the standard vms rules on grant
2349	* order.
2350	*/
2351
2352	if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2353	return 1;
2354
2355	/*
2356	* 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2357	* granted until all other conversion requests ahead of it are granted
2358	* and/or canceled.
2359	*/
2360
2361	if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2362	return 1;
2363
2364	/*
2365	* 6-4: By default, a new request is immediately granted only if all
2366	* three of the following conditions are satisfied when the request is
2367	* issued:
2368	* - The queue of ungranted conversion requests for the resource is
2369	* empty.
2370	* - The queue of ungranted new requests for the resource is empty.
2371	* - The mode of the new request is compatible with the most
2372	* restrictive mode of all granted locks on the resource.
2373	*/
2374
2375	if (now && !conv && list_empty(&r->res_convertqueue) &&
2376	list_empty(&r->res_waitqueue))
2377	return 1;
2378
2379	/*
2380	* 6-4: Once a lock request is in the queue of ungranted new requests,
2381	* it cannot be granted until the queue of ungranted conversion
2382	* requests is empty, all ungranted new requests ahead of it are
2383	* granted and/or canceled, and it is compatible with the granted mode
2384	* of the most restrictive lock granted on the resource.
2385	*/
2386
2387	if (!now && !conv && list_empty(&r->res_convertqueue) &&
2388	first_in_list(lkb, &r->res_waitqueue))
2389	return 1;
2390
2391	return 0;
2392	}
2393
2394	static int can_be_granted(struct dlm_rsb r, struct dlm_lkb lkb, int now,
2395	int recover, int *err)
2396	{
2397	int rv;
2398	int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2399	int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2400
2401	if (err)
2402	*err = 0;
2403
2404	rv = _can_be_granted(r, lkb, now, recover);
2405	if (rv)
2406	goto out;
2407
2408	/*
2409	* The CONVDEADLK flag is non-standard and tells the dlm to resolve
2410	* conversion deadlocks by demoting grmode to NL, otherwise the dlm
2411	* cancels one of the locks.
2412	*/
2413
2414	if (is_convert && can_be_queued(lkb) &&
2415	conversion_deadlock_detect(r, lkb)) {
2416	if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2417	lkb->lkb_grmode = DLM_LOCK_NL;
2418	set_bit(DLM_SBF_DEMOTED_BIT, &lkb->lkb_sbflags);
2419	} else if (err) {
2420	*err = -EDEADLK;
2421	} else {
2422	log_print("can_be_granted deadlock %x now %d",
2423	lkb->lkb_id, now);
2424	dlm_dump_rsb(r);
2425	}
2426	goto out;
2427	}
2428
2429	/*
2430	* The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2431	* to grant a request in a mode other than the normal rqmode. It's a
2432	* simple way to provide a big optimization to applications that can
2433	* use them.
2434	*/
2435
2436	if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2437	alt = DLM_LOCK_PR;
2438	else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2439	alt = DLM_LOCK_CW;
2440
2441	if (alt) {
2442	lkb->lkb_rqmode = alt;
2443	rv = _can_be_granted(r, lkb, now, 0);
2444	if (rv)
2445	set_bit(DLM_SBF_ALTMODE_BIT, &lkb->lkb_sbflags);
2446	else
2447	lkb->lkb_rqmode = rqmode;
2448	}
2449	out:
2450	return rv;
2451	}
2452
2453	/* Returns the highest requested mode of all blocked conversions; sets
2454	cw if there's a blocked conversion to DLM_LOCK_CW. */
2455
2456	static int grant_pending_convert(struct dlm_rsb r, int high, int cw,
2457	unsigned int *count)
2458	{
2459	struct dlm_lkb lkb, s;
2460	int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2461	int hi, demoted, quit, grant_restart, demote_restart;
2462	int deadlk;
2463
2464	quit = 0;
2465	restart:
2466	grant_restart = 0;
2467	demote_restart = 0;
2468	hi = DLM_LOCK_IV;
2469
2470	list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2471	demoted = is_demoted(lkb);
2472	deadlk = 0;
2473
2474	if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2475	grant_lock_pending(r, lkb);
2476	grant_restart = 1;
2477	if (count)
2478	(*count)++;
2479	continue;
2480	}
2481
2482	if (!demoted && is_demoted(lkb)) {
2483	log_print("WARN: pending demoted %x node %d %s",
2484	lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2485	demote_restart = 1;
2486	continue;
2487	}
2488
2489	if (deadlk) {
2490	/*
2491	* If DLM_LKB_NODLKWT flag is set and conversion
2492	* deadlock is detected, we request blocking AST and
2493	* down (or cancel) conversion.
2494	*/
2495	if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
2496	if (lkb->lkb_highbast < lkb->lkb_rqmode) {
2497	queue_bast(r, lkb, lkb->lkb_rqmode);
2498	lkb->lkb_highbast = lkb->lkb_rqmode;
2499	}
2500	} else {
2501	log_print("WARN: pending deadlock %x node %d %s",
2502	lkb->lkb_id, lkb->lkb_nodeid,
2503	r->res_name);
2504	dlm_dump_rsb(r);
2505	}
2506	continue;
2507	}
2508
2509	hi = max_t(int, lkb->lkb_rqmode, hi);
2510
2511	if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2512	*cw = 1;
2513	}
2514
2515	if (grant_restart)
2516	goto restart;
2517	if (demote_restart && !quit) {
2518	quit = 1;
2519	goto restart;
2520	}
2521
2522	return max_t(int, high, hi);
2523	}
2524
2525	static int grant_pending_wait(struct dlm_rsb r, int high, int cw,
2526	unsigned int *count)
2527	{
2528	struct dlm_lkb lkb, s;
2529
2530	list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2531	if (can_be_granted(r, lkb, 0, 0, NULL)) {
2532	grant_lock_pending(r, lkb);
2533	if (count)
2534	(*count)++;
2535	} else {
2536	high = max_t(int, lkb->lkb_rqmode, high);
2537	if (lkb->lkb_rqmode == DLM_LOCK_CW)
2538	*cw = 1;
2539	}
2540	}
2541
2542	return high;
2543	}
2544
2545	/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2546	on either the convert or waiting queue.
2547	high is the largest rqmode of all locks blocked on the convert or
2548	waiting queue. */
2549
2550	static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2551	{
2552	if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2553	if (gr->lkb_highbast < DLM_LOCK_EX)
2554	return 1;
2555	return 0;
2556	}
2557
2558	if (gr->lkb_highbast < high &&
2559	!__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2560	return 1;
2561	return 0;
2562	}
2563
2564	static void grant_pending_locks(struct dlm_rsb r, unsigned int count)
2565	{
2566	struct dlm_lkb lkb, s;
2567	int high = DLM_LOCK_IV;
2568	int cw = 0;
2569
2570	if (!is_master(r)) {
2571	log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2572	dlm_dump_rsb(r);
2573	return;
2574	}
2575
2576	high = grant_pending_convert(r, high, &cw, count);
2577	high = grant_pending_wait(r, high, &cw, count);
2578
2579	if (high == DLM_LOCK_IV)
2580	return;
2581
2582	/*
2583	* If there are locks left on the wait/convert queue then send blocking
2584	* ASTs to granted locks based on the largest requested mode (high)
2585	* found above.
2586	*/
2587
2588	list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2589	if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2590	if (cw && high == DLM_LOCK_PR &&
2591	lkb->lkb_grmode == DLM_LOCK_PR)
2592	queue_bast(r, lkb, DLM_LOCK_CW);
2593	else
2594	queue_bast(r, lkb, high);
2595	lkb->lkb_highbast = high;
2596	}
2597	}
2598	}
2599
2600	static int modes_require_bast(struct dlm_lkb gr, struct dlm_lkb rq)
2601	{
2602	if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) \|\|
2603	(gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2604	if (gr->lkb_highbast < DLM_LOCK_EX)
2605	return 1;
2606	return 0;
2607	}
2608
2609	if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2610	return 1;
2611	return 0;
2612	}
2613
2614	static void send_bast_queue(struct dlm_rsb r, struct list_head head,
2615	struct dlm_lkb *lkb)
2616	{
2617	struct dlm_lkb *gr;
2618
2619	list_for_each_entry(gr, head, lkb_statequeue) {
2620	/* skip self when sending basts to convertqueue */
2621	if (gr == lkb)
2622	continue;
2623	if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2624	queue_bast(r, gr, lkb->lkb_rqmode);
2625	gr->lkb_highbast = lkb->lkb_rqmode;
2626	}
2627	}
2628	}
2629
2630	static void send_blocking_asts(struct dlm_rsb r, struct dlm_lkb lkb)
2631	{
2632	send_bast_queue(r, &r->res_grantqueue, lkb);
2633	}
2634
2635	static void send_blocking_asts_all(struct dlm_rsb r, struct dlm_lkb lkb)
2636	{
2637	send_bast_queue(r, &r->res_grantqueue, lkb);
2638	send_bast_queue(r, &r->res_convertqueue, lkb);
2639	}
2640
2641	/* set_master(r, lkb) -- set the master nodeid of a resource
2642
2643	The purpose of this function is to set the nodeid field in the given
2644	lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2645	known, it can just be copied to the lkb and the function will return
2646	0. If the rsb's nodeid is _not_ known, it needs to be looked up
2647	before it can be copied to the lkb.
2648
2649	When the rsb nodeid is being looked up remotely, the initial lkb
2650	causing the lookup is kept on the ls_waiters list waiting for the
2651	lookup reply. Other lkb's waiting for the same rsb lookup are kept
2652	on the rsb's res_lookup list until the master is verified.
2653
2654	Return values:
2655	0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2656	1: the rsb master is not available and the lkb has been placed on
2657	a wait queue
2658	*/
2659
2660	static int set_master(struct dlm_rsb r, struct dlm_lkb lkb)
2661	{
2662	int our_nodeid = dlm_our_nodeid();
2663
2664	if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2665	rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2666	r->res_first_lkid = lkb->lkb_id;
2667	lkb->lkb_nodeid = r->res_nodeid;
2668	return 0;
2669	}
2670
2671	if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2672	list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2673	return 1;
2674	}
2675
2676	if (r->res_master_nodeid == our_nodeid) {
2677	lkb->lkb_nodeid = 0;
2678	return 0;
2679	}
2680
2681	if (r->res_master_nodeid) {
2682	lkb->lkb_nodeid = r->res_master_nodeid;
2683	return 0;
2684	}
2685
2686	if (dlm_dir_nodeid(r) == our_nodeid) {
2687	/* This is a somewhat unusual case; find_rsb will usually
2688	have set res_master_nodeid when dir nodeid is local, but
2689	there are cases where we become the dir node after we've
2690	past find_rsb and go through _request_lock again.
2691	confirm_master() or process_lookup_list() needs to be
2692	called after this. */
2693	log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2694	lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2695	r->res_name);
2696	r->res_master_nodeid = our_nodeid;
2697	r->res_nodeid = 0;
2698	lkb->lkb_nodeid = 0;
2699	return 0;
2700	}
2701
2702	r->res_first_lkid = lkb->lkb_id;
2703	send_lookup(r, lkb);
2704	return 1;
2705	}
2706
2707	static void process_lookup_list(struct dlm_rsb *r)
2708	{
2709	struct dlm_lkb lkb, safe;
2710
2711	list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2712	list_del_init(&lkb->lkb_rsb_lookup);
2713	_request_lock(r, lkb);
2714	}
2715	}
2716
2717	/* confirm_master -- confirm (or deny) an rsb's master nodeid */
2718
2719	static void confirm_master(struct dlm_rsb *r, int error)
2720	{
2721	struct dlm_lkb *lkb;
2722
2723	if (!r->res_first_lkid)
2724	return;
2725
2726	switch (error) {
2727	case 0:
2728	case -EINPROGRESS:
2729	r->res_first_lkid = 0;
2730	process_lookup_list(r);
2731	break;
2732
2733	case -EAGAIN:
2734	case -EBADR:
2735	case -ENOTBLK:
2736	/* the remote request failed and won't be retried (it was
2737	a NOQUEUE, or has been canceled/unlocked); make a waiting
2738	lkb the first_lkid */
2739
2740	r->res_first_lkid = 0;
2741
2742	if (!list_empty(&r->res_lookup)) {
2743	lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2744	lkb_rsb_lookup);
2745	list_del_init(&lkb->lkb_rsb_lookup);
2746	r->res_first_lkid = lkb->lkb_id;
2747	_request_lock(r, lkb);
2748	}
2749	break;
2750
2751	default:
2752	log_error(r->res_ls, "confirm_master unknown error %d", error);
2753	}
2754	}
2755
2756	static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2757	int namelen, void (ast)(void astparam),
2758	void *astparam,
2759	void (bast)(void astparam, int mode),
2760	struct dlm_args *args)
2761	{
2762	int rv = -EINVAL;
2763
2764	/* check for invalid arg usage */
2765
2766	if (mode < 0 \|\| mode > DLM_LOCK_EX)
2767	goto out;
2768
2769	if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2770	goto out;
2771
2772	if (flags & DLM_LKF_CANCEL)
2773	goto out;
2774
2775	if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2776	goto out;
2777
2778	if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2779	goto out;
2780
2781	if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2782	goto out;
2783
2784	if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2785	goto out;
2786
2787	if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2788	goto out;
2789
2790	if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2791	goto out;
2792
2793	if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2794	goto out;
2795
2796	if (!ast \|\| !lksb)
2797	goto out;
2798
2799	if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2800	goto out;
2801
2802	if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2803	goto out;
2804
2805	/* these args will be copied to the lkb in validate_lock_args,
2806	it cannot be done now because when converting locks, fields in
2807	an active lkb cannot be modified before locking the rsb */
2808
2809	args->flags = flags;
2810	args->astfn = ast;
2811	args->astparam = astparam;
2812	args->bastfn = bast;
2813	args->mode = mode;
2814	args->lksb = lksb;
2815	rv = 0;
2816	out:
2817	return rv;
2818	}
2819
2820	static int set_unlock_args(uint32_t flags, void astarg, struct dlm_args args)
2821	{
2822	if (flags & ~(DLM_LKF_CANCEL \| DLM_LKF_VALBLK \| DLM_LKF_IVVALBLK \|
2823	DLM_LKF_FORCEUNLOCK))
2824	return -EINVAL;
2825
2826	if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2827	return -EINVAL;
2828
2829	args->flags = flags;
2830	args->astparam = astarg;
2831	return 0;
2832	}
2833
2834	static int validate_lock_args(struct dlm_ls ls, struct dlm_lkb lkb,
2835	struct dlm_args *args)
2836	{
2837	int rv = -EBUSY;
2838
2839	if (args->flags & DLM_LKF_CONVERT) {
2840	if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2841	goto out;
2842
2843	/* lock not allowed if there's any op in progress */
2844	if (lkb->lkb_wait_type \|\| lkb->lkb_wait_count)
2845	goto out;
2846
2847	if (is_overlap(lkb))
2848	goto out;
2849
2850	rv = -EINVAL;
2851	if (test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags))
2852	goto out;
2853
2854	if (args->flags & DLM_LKF_QUECVT &&
2855	!__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2856	goto out;
2857	}
2858
2859	lkb->lkb_exflags = args->flags;
2860	dlm_set_sbflags_val(lkb, 0);
2861	lkb->lkb_astfn = args->astfn;
2862	lkb->lkb_astparam = args->astparam;
2863	lkb->lkb_bastfn = args->bastfn;
2864	lkb->lkb_rqmode = args->mode;
2865	lkb->lkb_lksb = args->lksb;
2866	lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2867	lkb->lkb_ownpid = (int) current->pid;
2868	rv = 0;
2869	out:
2870	switch (rv) {
2871	case 0:
2872	break;
2873	case -EINVAL:
2874	/* annoy the user because dlm usage is wrong */
2875	WARN_ON(1);
2876	log_error(ls, "%s %d %x %x %x %d %d", __func__,
2877	rv, lkb->lkb_id, dlm_iflags_val(lkb), args->flags,
2878	lkb->lkb_status, lkb->lkb_wait_type);
2879	break;
2880	default:
2881	log_debug(ls, "%s %d %x %x %x %d %d", __func__,
2882	rv, lkb->lkb_id, dlm_iflags_val(lkb), args->flags,
2883	lkb->lkb_status, lkb->lkb_wait_type);
2884	break;
2885	}
2886
2887	return rv;
2888	}
2889
2890	/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2891	for success */
2892
2893	/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2894	because there may be a lookup in progress and it's valid to do
2895	cancel/unlockf on it */
2896
2897	static int validate_unlock_args(struct dlm_lkb lkb, struct dlm_args args)
2898	{
2899	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2900	int rv = -EBUSY;
2901
2902	/* normal unlock not allowed if there's any op in progress */
2903	if (!(args->flags & (DLM_LKF_CANCEL \| DLM_LKF_FORCEUNLOCK)) &&
2904	(lkb->lkb_wait_type \|\| lkb->lkb_wait_count))
2905	goto out;
2906
2907	/* an lkb may be waiting for an rsb lookup to complete where the
2908	lookup was initiated by another lock */
2909
2910	if (!list_empty(&lkb->lkb_rsb_lookup)) {
2911	if (args->flags & (DLM_LKF_CANCEL \| DLM_LKF_FORCEUNLOCK)) {
2912	log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2913	list_del_init(&lkb->lkb_rsb_lookup);
2914	queue_cast(lkb->lkb_resource, lkb,
2915	args->flags & DLM_LKF_CANCEL ?
2916	-DLM_ECANCEL : -DLM_EUNLOCK);
2917	unhold_lkb(lkb); /* undoes create_lkb() */
2918	}
2919	/* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2920	goto out;
2921	}
2922
2923	rv = -EINVAL;
2924	if (test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags)) {
2925	log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2926	dlm_print_lkb(lkb);
2927	goto out;
2928	}
2929
2930	/* an lkb may still exist even though the lock is EOL'ed due to a
2931	* cancel, unlock or failed noqueue request; an app can't use these
2932	* locks; return same error as if the lkid had not been found at all
2933	*/
2934
2935	if (test_bit(DLM_IFL_ENDOFLIFE_BIT, &lkb->lkb_iflags)) {
2936	log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2937	rv = -ENOENT;
2938	goto out;
2939	}
2940
2941	if (is_overlap_unlock(lkb))
2942	goto out;
2943
2944	/* cancel not allowed with another cancel/unlock in progress */
2945
2946	if (args->flags & DLM_LKF_CANCEL) {
2947	if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2948	goto out;
2949
2950	if (is_overlap_cancel(lkb))
2951	goto out;
2952
2953	if (test_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags)) {
2954	set_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
2955	rv = -EBUSY;
2956	goto out;
2957	}
2958
2959	/* there's nothing to cancel */
2960	if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2961	!lkb->lkb_wait_type) {
2962	rv = -EBUSY;
2963	goto out;
2964	}
2965
2966	switch (lkb->lkb_wait_type) {
2967	case DLM_MSG_LOOKUP:
2968	case DLM_MSG_REQUEST:
2969	set_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
2970	rv = -EBUSY;
2971	goto out;
2972	case DLM_MSG_UNLOCK:
2973	case DLM_MSG_CANCEL:
2974	goto out;
2975	}
2976	/* add_to_waiters() will set OVERLAP_CANCEL */
2977	goto out_ok;
2978	}
2979
2980	/* do we need to allow a force-unlock if there's a normal unlock
2981	already in progress? in what conditions could the normal unlock
2982	fail such that we'd want to send a force-unlock to be sure? */
2983
2984	if (args->flags & DLM_LKF_FORCEUNLOCK) {
2985	if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2986	goto out;
2987
2988	if (test_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags)) {
2989	set_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
2990	rv = -EBUSY;
2991	goto out;
2992	}
2993
2994	switch (lkb->lkb_wait_type) {
2995	case DLM_MSG_LOOKUP:
2996	case DLM_MSG_REQUEST:
2997	set_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
2998	rv = -EBUSY;
2999	goto out;
3000	case DLM_MSG_UNLOCK:
3001	goto out;
3002	}
3003	/* add_to_waiters() will set OVERLAP_UNLOCK */
3004	}
3005
3006	out_ok:
3007	/* an overlapping op shouldn't blow away exflags from other op */
3008	lkb->lkb_exflags \|= args->flags;
3009	dlm_set_sbflags_val(lkb, 0);
3010	lkb->lkb_astparam = args->astparam;
3011	rv = 0;
3012	out:
3013	switch (rv) {
3014	case 0:
3015	break;
3016	case -EINVAL:
3017	/* annoy the user because dlm usage is wrong */
3018	WARN_ON(1);
3019	log_error(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
3020	lkb->lkb_id, dlm_iflags_val(lkb), lkb->lkb_exflags,
3021	args->flags, lkb->lkb_wait_type,
3022	lkb->lkb_resource->res_name);
3023	break;
3024	default:
3025	log_debug(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
3026	lkb->lkb_id, dlm_iflags_val(lkb), lkb->lkb_exflags,
3027	args->flags, lkb->lkb_wait_type,
3028	lkb->lkb_resource->res_name);
3029	break;
3030	}
3031
3032	return rv;
3033	}
3034
3035	/*
3036	* Four stage 4 varieties:
3037	* do_request(), do_convert(), do_unlock(), do_cancel()
3038	* These are called on the master node for the given lock and
3039	* from the central locking logic.
3040	*/
3041
3042	static int do_request(struct dlm_rsb r, struct dlm_lkb lkb)
3043	{
3044	int error = 0;
3045
3046	if (can_be_granted(r, lkb, 1, 0, NULL)) {
3047	grant_lock(r, lkb);
3048	queue_cast(r, lkb, 0);
3049	goto out;
3050	}
3051
3052	if (can_be_queued(lkb)) {
3053	error = -EINPROGRESS;
3054	add_lkb(r, lkb, DLM_LKSTS_WAITING);
3055	goto out;
3056	}
3057
3058	error = -EAGAIN;
3059	queue_cast(r, lkb, -EAGAIN);
3060	out:
3061	return error;
3062	}
3063
3064	static void do_request_effects(struct dlm_rsb r, struct dlm_lkb lkb,
3065	int error)
3066	{
3067	switch (error) {
3068	case -EAGAIN:
3069	if (force_blocking_asts(lkb))
3070	send_blocking_asts_all(r, lkb);
3071	break;
3072	case -EINPROGRESS:
3073	send_blocking_asts(r, lkb);
3074	break;
3075	}
3076	}
3077
3078	static int do_convert(struct dlm_rsb r, struct dlm_lkb lkb)
3079	{
3080	int error = 0;
3081	int deadlk = 0;
3082
3083	/* changing an existing lock may allow others to be granted */
3084
3085	if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3086	grant_lock(r, lkb);
3087	queue_cast(r, lkb, 0);
3088	goto out;
3089	}
3090
3091	/* can_be_granted() detected that this lock would block in a conversion
3092	deadlock, so we leave it on the granted queue and return EDEADLK in
3093	the ast for the convert. */
3094
3095	if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
3096	/* it's left on the granted queue */
3097	revert_lock(r, lkb);
3098	queue_cast(r, lkb, -EDEADLK);
3099	error = -EDEADLK;
3100	goto out;
3101	}
3102
3103	/* is_demoted() means the can_be_granted() above set the grmode
3104	to NL, and left us on the granted queue. This auto-demotion
3105	(due to CONVDEADLK) might mean other locks, and/or this lock, are
3106	now grantable. We have to try to grant other converting locks
3107	before we try again to grant this one. */
3108
3109	if (is_demoted(lkb)) {
3110	grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3111	if (_can_be_granted(r, lkb, 1, 0)) {
3112	grant_lock(r, lkb);
3113	queue_cast(r, lkb, 0);
3114	goto out;
3115	}
3116	/* else fall through and move to convert queue */
3117	}
3118
3119	if (can_be_queued(lkb)) {
3120	error = -EINPROGRESS;
3121	del_lkb(r, lkb);
3122	add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3123	goto out;
3124	}
3125
3126	error = -EAGAIN;
3127	queue_cast(r, lkb, -EAGAIN);
3128	out:
3129	return error;
3130	}
3131
3132	static void do_convert_effects(struct dlm_rsb r, struct dlm_lkb lkb,
3133	int error)
3134	{
3135	switch (error) {
3136	case 0:
3137	grant_pending_locks(r, NULL);
3138	/* grant_pending_locks also sends basts */
3139	break;
3140	case -EAGAIN:
3141	if (force_blocking_asts(lkb))
3142	send_blocking_asts_all(r, lkb);
3143	break;
3144	case -EINPROGRESS:
3145	send_blocking_asts(r, lkb);
3146	break;
3147	}
3148	}
3149
3150	static int do_unlock(struct dlm_rsb r, struct dlm_lkb lkb)
3151	{
3152	remove_lock(r, lkb);
3153	queue_cast(r, lkb, -DLM_EUNLOCK);
3154	return -DLM_EUNLOCK;
3155	}
3156
3157	static void do_unlock_effects(struct dlm_rsb r, struct dlm_lkb lkb,
3158	int error)
3159	{
3160	grant_pending_locks(r, NULL);
3161	}
3162
3163	/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3164
3165	static int do_cancel(struct dlm_rsb r, struct dlm_lkb lkb)
3166	{
3167	int error;
3168
3169	error = revert_lock(r, lkb);
3170	if (error) {
3171	queue_cast(r, lkb, -DLM_ECANCEL);
3172	return -DLM_ECANCEL;
3173	}
3174	return 0;
3175	}
3176
3177	static void do_cancel_effects(struct dlm_rsb r, struct dlm_lkb lkb,
3178	int error)
3179	{
3180	if (error)
3181	grant_pending_locks(r, NULL);
3182	}
3183
3184	/*
3185	* Four stage 3 varieties:
3186	* _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3187	*/
3188
3189	/* add a new lkb to a possibly new rsb, called by requesting process */
3190
3191	static int _request_lock(struct dlm_rsb r, struct dlm_lkb lkb)
3192	{
3193	int error;
3194
3195	/* set_master: sets lkb nodeid from r */
3196
3197	error = set_master(r, lkb);
3198	if (error < 0)
3199	goto out;
3200	if (error) {
3201	error = 0;
3202	goto out;
3203	}
3204
3205	if (is_remote(r)) {
3206	/* receive_request() calls do_request() on remote node */
3207	error = send_request(r, lkb);
3208	} else {
3209	error = do_request(r, lkb);
3210	/* for remote locks the request_reply is sent
3211	between do_request and do_request_effects */
3212	do_request_effects(r, lkb, error);
3213	}
3214	out:
3215	return error;
3216	}
3217
3218	/* change some property of an existing lkb, e.g. mode */
3219
3220	static int _convert_lock(struct dlm_rsb r, struct dlm_lkb lkb)
3221	{
3222	int error;
3223
3224	if (is_remote(r)) {
3225	/* receive_convert() calls do_convert() on remote node */
3226	error = send_convert(r, lkb);
3227	} else {
3228	error = do_convert(r, lkb);
3229	/* for remote locks the convert_reply is sent
3230	between do_convert and do_convert_effects */
3231	do_convert_effects(r, lkb, error);
3232	}
3233
3234	return error;
3235	}
3236
3237	/* remove an existing lkb from the granted queue */
3238
3239	static int _unlock_lock(struct dlm_rsb r, struct dlm_lkb lkb)
3240	{
3241	int error;
3242
3243	if (is_remote(r)) {
3244	/* receive_unlock() calls do_unlock() on remote node */
3245	error = send_unlock(r, lkb);
3246	} else {
3247	error = do_unlock(r, lkb);
3248	/* for remote locks the unlock_reply is sent
3249	between do_unlock and do_unlock_effects */
3250	do_unlock_effects(r, lkb, error);
3251	}
3252
3253	return error;
3254	}
3255
3256	/* remove an existing lkb from the convert or wait queue */
3257
3258	static int _cancel_lock(struct dlm_rsb r, struct dlm_lkb lkb)
3259	{
3260	int error;
3261
3262	if (is_remote(r)) {
3263	/* receive_cancel() calls do_cancel() on remote node */
3264	error = send_cancel(r, lkb);
3265	} else {
3266	error = do_cancel(r, lkb);
3267	/* for remote locks the cancel_reply is sent
3268	between do_cancel and do_cancel_effects */
3269	do_cancel_effects(r, lkb, error);
3270	}
3271
3272	return error;
3273	}
3274
3275	/*
3276	* Four stage 2 varieties:
3277	* request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3278	*/
3279
3280	static int request_lock(struct dlm_ls ls, struct dlm_lkb lkb,
3281	const void *name, int len,
3282	struct dlm_args *args)
3283	{
3284	struct dlm_rsb *r;
3285	int error;
3286
3287	error = validate_lock_args(ls, lkb, args);
3288	if (error)
3289	return error;
3290
3291	error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3292	if (error)
3293	return error;
3294
3295	lock_rsb(r);
3296
3297	attach_lkb(r, lkb);
3298	lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3299
3300	error = _request_lock(r, lkb);
3301
3302	unlock_rsb(r);
3303	put_rsb(r);
3304	return error;
3305	}
3306
3307	static int convert_lock(struct dlm_ls ls, struct dlm_lkb lkb,
3308	struct dlm_args *args)
3309	{
3310	struct dlm_rsb *r;
3311	int error;
3312
3313	r = lkb->lkb_resource;
3314
3315	hold_rsb(r);
3316	lock_rsb(r);
3317
3318	error = validate_lock_args(ls, lkb, args);
3319	if (error)
3320	goto out;
3321
3322	error = _convert_lock(r, lkb);
3323	out:
3324	unlock_rsb(r);
3325	put_rsb(r);
3326	return error;
3327	}
3328
3329	static int unlock_lock(struct dlm_ls ls, struct dlm_lkb lkb,
3330	struct dlm_args *args)
3331	{
3332	struct dlm_rsb *r;
3333	int error;
3334
3335	r = lkb->lkb_resource;
3336
3337	hold_rsb(r);
3338	lock_rsb(r);
3339
3340	error = validate_unlock_args(lkb, args);
3341	if (error)
3342	goto out;
3343
3344	error = _unlock_lock(r, lkb);
3345	out:
3346	unlock_rsb(r);
3347	put_rsb(r);
3348	return error;
3349	}
3350
3351	static int cancel_lock(struct dlm_ls ls, struct dlm_lkb lkb,
3352	struct dlm_args *args)
3353	{
3354	struct dlm_rsb *r;
3355	int error;
3356
3357	r = lkb->lkb_resource;
3358
3359	hold_rsb(r);
3360	lock_rsb(r);
3361
3362	error = validate_unlock_args(lkb, args);
3363	if (error)
3364	goto out;
3365
3366	error = _cancel_lock(r, lkb);
3367	out:
3368	unlock_rsb(r);
3369	put_rsb(r);
3370	return error;
3371	}
3372
3373	/*
3374	* Two stage 1 varieties: dlm_lock() and dlm_unlock()
3375	*/
3376
3377	int dlm_lock(dlm_lockspace_t *lockspace,
3378	int mode,
3379	struct dlm_lksb *lksb,
3380	uint32_t flags,
3381	const void *name,
3382	unsigned int namelen,
3383	uint32_t parent_lkid,
3384	void (ast) (void astarg),
3385	void *astarg,
3386	void (bast) (void astarg, int mode))
3387	{
3388	struct dlm_ls *ls;
3389	struct dlm_lkb *lkb;
3390	struct dlm_args args;
3391	int error, convert = flags & DLM_LKF_CONVERT;
3392
3393	ls = dlm_find_lockspace_local(lockspace);
3394	if (!ls)
3395	return -EINVAL;
3396
3397	dlm_lock_recovery(ls);
3398
3399	if (convert)
3400	error = find_lkb(ls, lksb->sb_lkid, &lkb);
3401	else
3402	error = create_lkb(ls, &lkb);
3403
3404	if (error)
3405	goto out;
3406
3407	trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
3408
3409	error = set_lock_args(mode, lksb, flags, namelen, ast, astarg, bast,
3410	&args);
3411	if (error)
3412	goto out_put;
3413
3414	if (convert)
3415	error = convert_lock(ls, lkb, &args);
3416	else
3417	error = request_lock(ls, lkb, name, namelen, &args);
3418
3419	if (error == -EINPROGRESS)
3420	error = 0;
3421	out_put:
3422	trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, true);
3423
3424	if (convert \|\| error)
3425	__put_lkb(ls, lkb);
3426	if (error == -EAGAIN \|\| error == -EDEADLK)
3427	error = 0;
3428	out:
3429	dlm_unlock_recovery(ls);
3430	dlm_put_lockspace(ls);
3431	return error;
3432	}
3433
3434	int dlm_unlock(dlm_lockspace_t *lockspace,
3435	uint32_t lkid,
3436	uint32_t flags,
3437	struct dlm_lksb *lksb,
3438	void *astarg)
3439	{
3440	struct dlm_ls *ls;
3441	struct dlm_lkb *lkb;
3442	struct dlm_args args;
3443	int error;
3444
3445	ls = dlm_find_lockspace_local(lockspace);
3446	if (!ls)
3447	return -EINVAL;
3448
3449	dlm_lock_recovery(ls);
3450
3451	error = find_lkb(ls, lkid, &lkb);
3452	if (error)
3453	goto out;
3454
3455	trace_dlm_unlock_start(ls, lkb, flags);
3456
3457	error = set_unlock_args(flags, astarg, &args);
3458	if (error)
3459	goto out_put;
3460
3461	if (flags & DLM_LKF_CANCEL)
3462	error = cancel_lock(ls, lkb, &args);
3463	else
3464	error = unlock_lock(ls, lkb, &args);
3465
3466	if (error == -DLM_EUNLOCK \|\| error == -DLM_ECANCEL)
3467	error = 0;
3468	if (error == -EBUSY && (flags & (DLM_LKF_CANCEL \| DLM_LKF_FORCEUNLOCK)))
3469	error = 0;
3470	out_put:
3471	trace_dlm_unlock_end(ls, lkb, flags, error);
3472
3473	dlm_put_lkb(lkb);
3474	out:
3475	dlm_unlock_recovery(ls);
3476	dlm_put_lockspace(ls);
3477	return error;
3478	}
3479
3480	/*
3481	* send/receive routines for remote operations and replies
3482	*
3483	* send_args
3484	* send_common
3485	* send_request receive_request
3486	* send_convert receive_convert
3487	* send_unlock receive_unlock
3488	* send_cancel receive_cancel
3489	* send_grant receive_grant
3490	* send_bast receive_bast
3491	* send_lookup receive_lookup
3492	* send_remove receive_remove
3493	*
3494	* send_common_reply
3495	* receive_request_reply send_request_reply
3496	* receive_convert_reply send_convert_reply
3497	* receive_unlock_reply send_unlock_reply
3498	* receive_cancel_reply send_cancel_reply
3499	* receive_lookup_reply send_lookup_reply
3500	*/
3501
3502	static int _create_message(struct dlm_ls *ls, int mb_len,
3503	int to_nodeid, int mstype,
3504	struct dlm_message **ms_ret,
3505	struct dlm_mhandle **mh_ret)
3506	{
3507	struct dlm_message *ms;
3508	struct dlm_mhandle *mh;
3509	char *mb;
3510
3511	/* get_buffer gives us a message handle (mh) that we need to
3512	pass into midcomms_commit and a message buffer (mb) that we
3513	write our data into */
3514
3515	mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, &mb);
3516	if (!mh)
3517	return -ENOBUFS;
3518
3519	ms = (struct dlm_message *) mb;
3520
3521	ms->m_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR \| DLM_HEADER_MINOR);
3522	ms->m_header.u.h_lockspace = cpu_to_le32(ls->ls_global_id);
3523	ms->m_header.h_nodeid = cpu_to_le32(dlm_our_nodeid());
3524	ms->m_header.h_length = cpu_to_le16(mb_len);
3525	ms->m_header.h_cmd = DLM_MSG;
3526
3527	ms->m_type = cpu_to_le32(mstype);
3528
3529	*mh_ret = mh;
3530	*ms_ret = ms;
3531	return 0;
3532	}
3533
3534	static int create_message(struct dlm_rsb r, struct dlm_lkb lkb,
3535	int to_nodeid, int mstype,
3536	struct dlm_message **ms_ret,
3537	struct dlm_mhandle **mh_ret)
3538	{
3539	int mb_len = sizeof(struct dlm_message);
3540
3541	switch (mstype) {
3542	case DLM_MSG_REQUEST:
3543	case DLM_MSG_LOOKUP:
3544	case DLM_MSG_REMOVE:
3545	mb_len += r->res_length;
3546	break;
3547	case DLM_MSG_CONVERT:
3548	case DLM_MSG_UNLOCK:
3549	case DLM_MSG_REQUEST_REPLY:
3550	case DLM_MSG_CONVERT_REPLY:
3551	case DLM_MSG_GRANT:
3552	if (lkb && lkb->lkb_lvbptr && (lkb->lkb_exflags & DLM_LKF_VALBLK))
3553	mb_len += r->res_ls->ls_lvblen;
3554	break;
3555	}
3556
3557	return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3558	ms_ret, mh_ret);
3559	}
3560
3561	/* further lowcomms enhancements or alternate implementations may make
3562	the return value from this function useful at some point */
3563
3564	static int send_message(struct dlm_mhandle mh, struct dlm_message ms,
3565	const void *name, int namelen)
3566	{
3567	dlm_midcomms_commit_mhandle(mh, name, namelen);
3568	return 0;
3569	}
3570
3571	static void send_args(struct dlm_rsb r, struct dlm_lkb lkb,
3572	struct dlm_message *ms)
3573	{
3574	ms->m_nodeid = cpu_to_le32(lkb->lkb_nodeid);
3575	ms->m_pid = cpu_to_le32(lkb->lkb_ownpid);
3576	ms->m_lkid = cpu_to_le32(lkb->lkb_id);
3577	ms->m_remid = cpu_to_le32(lkb->lkb_remid);
3578	ms->m_exflags = cpu_to_le32(lkb->lkb_exflags);
3579	ms->m_sbflags = cpu_to_le32(dlm_sbflags_val(lkb));
3580	ms->m_flags = cpu_to_le32(dlm_dflags_val(lkb));
3581	ms->m_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
3582	ms->m_status = cpu_to_le32(lkb->lkb_status);
3583	ms->m_grmode = cpu_to_le32(lkb->lkb_grmode);
3584	ms->m_rqmode = cpu_to_le32(lkb->lkb_rqmode);
3585	ms->m_hash = cpu_to_le32(r->res_hash);
3586
3587	/* m_result and m_bastmode are set from function args,
3588	not from lkb fields */
3589
3590	if (lkb->lkb_bastfn)
3591	ms->m_asts \|= cpu_to_le32(DLM_CB_BAST);
3592	if (lkb->lkb_astfn)
3593	ms->m_asts \|= cpu_to_le32(DLM_CB_CAST);
3594
3595	/* compare with switch in create_message; send_remove() doesn't
3596	use send_args() */
3597
3598	switch (ms->m_type) {
3599	case cpu_to_le32(DLM_MSG_REQUEST):
3600	case cpu_to_le32(DLM_MSG_LOOKUP):
3601	memcpy(ms->m_extra, r->res_name, r->res_length);
3602	break;
3603	case cpu_to_le32(DLM_MSG_CONVERT):
3604	case cpu_to_le32(DLM_MSG_UNLOCK):
3605	case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3606	case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3607	case cpu_to_le32(DLM_MSG_GRANT):
3608	if (!lkb->lkb_lvbptr \|\| !(lkb->lkb_exflags & DLM_LKF_VALBLK))
3609	break;
3610	memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3611	break;
3612	}
3613	}
3614
3615	static int send_common(struct dlm_rsb r, struct dlm_lkb lkb, int mstype)
3616	{
3617	struct dlm_message *ms;
3618	struct dlm_mhandle *mh;
3619	int to_nodeid, error;
3620
3621	to_nodeid = r->res_nodeid;
3622
3623	add_to_waiters(lkb, mstype, to_nodeid);
3624	error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3625	if (error)
3626	goto fail;
3627
3628	send_args(r, lkb, ms);
3629
3630	error = send_message(mh, ms, r->res_name, r->res_length);
3631	if (error)
3632	goto fail;
3633	return 0;
3634
3635	fail:
3636	remove_from_waiters(lkb, msg_reply_type(mstype));
3637	return error;
3638	}
3639
3640	static int send_request(struct dlm_rsb r, struct dlm_lkb lkb)
3641	{
3642	return send_common(r, lkb, DLM_MSG_REQUEST);
3643	}
3644
3645	static int send_convert(struct dlm_rsb r, struct dlm_lkb lkb)
3646	{
3647	int error;
3648
3649	error = send_common(r, lkb, DLM_MSG_CONVERT);
3650
3651	/* down conversions go without a reply from the master */
3652	if (!error && down_conversion(lkb)) {
3653	remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3654	r->res_ls->ls_local_ms.m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
3655	r->res_ls->ls_local_ms.m_result = 0;
3656	__receive_convert_reply(r, lkb, &r->res_ls->ls_local_ms, true);
3657	}
3658
3659	return error;
3660	}
3661
3662	/* FIXME: if this lkb is the only lock we hold on the rsb, then set
3663	MASTER_UNCERTAIN to force the next request on the rsb to confirm
3664	that the master is still correct. */
3665
3666	static int send_unlock(struct dlm_rsb r, struct dlm_lkb lkb)
3667	{
3668	return send_common(r, lkb, DLM_MSG_UNLOCK);
3669	}
3670
3671	static int send_cancel(struct dlm_rsb r, struct dlm_lkb lkb)
3672	{
3673	return send_common(r, lkb, DLM_MSG_CANCEL);
3674	}
3675
3676	static int send_grant(struct dlm_rsb r, struct dlm_lkb lkb)
3677	{
3678	struct dlm_message *ms;
3679	struct dlm_mhandle *mh;
3680	int to_nodeid, error;
3681
3682	to_nodeid = lkb->lkb_nodeid;
3683
3684	error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3685	if (error)
3686	goto out;
3687
3688	send_args(r, lkb, ms);
3689
3690	ms->m_result = 0;
3691
3692	error = send_message(mh, ms, r->res_name, r->res_length);
3693	out:
3694	return error;
3695	}
3696
3697	static int send_bast(struct dlm_rsb r, struct dlm_lkb lkb, int mode)
3698	{
3699	struct dlm_message *ms;
3700	struct dlm_mhandle *mh;
3701	int to_nodeid, error;
3702
3703	to_nodeid = lkb->lkb_nodeid;
3704
3705	error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3706	if (error)
3707	goto out;
3708
3709	send_args(r, lkb, ms);
3710
3711	ms->m_bastmode = cpu_to_le32(mode);
3712
3713	error = send_message(mh, ms, r->res_name, r->res_length);
3714	out:
3715	return error;
3716	}
3717
3718	static int send_lookup(struct dlm_rsb r, struct dlm_lkb lkb)
3719	{
3720	struct dlm_message *ms;
3721	struct dlm_mhandle *mh;
3722	int to_nodeid, error;
3723
3724	to_nodeid = dlm_dir_nodeid(r);
3725
3726	add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3727	error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3728	if (error)
3729	goto fail;
3730
3731	send_args(r, lkb, ms);
3732
3733	error = send_message(mh, ms, r->res_name, r->res_length);
3734	if (error)
3735	goto fail;
3736	return 0;
3737
3738	fail:
3739	remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3740	return error;
3741	}
3742
3743	static int send_remove(struct dlm_rsb *r)
3744	{
3745	struct dlm_message *ms;
3746	struct dlm_mhandle *mh;
3747	int to_nodeid, error;
3748
3749	to_nodeid = dlm_dir_nodeid(r);
3750
3751	error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3752	if (error)
3753	goto out;
3754
3755	memcpy(ms->m_extra, r->res_name, r->res_length);
3756	ms->m_hash = cpu_to_le32(r->res_hash);
3757
3758	error = send_message(mh, ms, r->res_name, r->res_length);
3759	out:
3760	return error;
3761	}
3762
3763	static int send_common_reply(struct dlm_rsb r, struct dlm_lkb lkb,
3764	int mstype, int rv)
3765	{
3766	struct dlm_message *ms;
3767	struct dlm_mhandle *mh;
3768	int to_nodeid, error;
3769
3770	to_nodeid = lkb->lkb_nodeid;
3771
3772	error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3773	if (error)
3774	goto out;
3775
3776	send_args(r, lkb, ms);
3777
3778	ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3779
3780	error = send_message(mh, ms, r->res_name, r->res_length);
3781	out:
3782	return error;
3783	}
3784
3785	static int send_request_reply(struct dlm_rsb r, struct dlm_lkb lkb, int rv)
3786	{
3787	return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3788	}
3789
3790	static int send_convert_reply(struct dlm_rsb r, struct dlm_lkb lkb, int rv)
3791	{
3792	return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3793	}
3794
3795	static int send_unlock_reply(struct dlm_rsb r, struct dlm_lkb lkb, int rv)
3796	{
3797	return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3798	}
3799
3800	static int send_cancel_reply(struct dlm_rsb r, struct dlm_lkb lkb, int rv)
3801	{
3802	return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3803	}
3804
3805	static int send_lookup_reply(struct dlm_ls *ls,
3806	const struct dlm_message *ms_in, int ret_nodeid,
3807	int rv)
3808	{
3809	struct dlm_rsb *r = &ls->ls_local_rsb;
3810	struct dlm_message *ms;
3811	struct dlm_mhandle *mh;
3812	int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid);
3813
3814	error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3815	if (error)
3816	goto out;
3817
3818	ms->m_lkid = ms_in->m_lkid;
3819	ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3820	ms->m_nodeid = cpu_to_le32(ret_nodeid);
3821
3822	error = send_message(mh, ms, ms_in->m_extra, receive_extralen(ms_in));
3823	out:
3824	return error;
3825	}
3826
3827	/* which args we save from a received message depends heavily on the type
3828	of message, unlike the send side where we can safely send everything about
3829	the lkb for any type of message */
3830
3831	static void receive_flags(struct dlm_lkb lkb, const struct dlm_message ms)
3832	{
3833	lkb->lkb_exflags = le32_to_cpu(ms->m_exflags);
3834	dlm_set_sbflags_val(lkb, le32_to_cpu(ms->m_sbflags));
3835	dlm_set_dflags_val(lkb, le32_to_cpu(ms->m_flags));
3836	}
3837
3838	static void receive_flags_reply(struct dlm_lkb *lkb,
3839	const struct dlm_message *ms,
3840	bool local)
3841	{
3842	if (local)
3843	return;
3844
3845	dlm_set_sbflags_val(lkb, le32_to_cpu(ms->m_sbflags));
3846	dlm_set_dflags_val(lkb, le32_to_cpu(ms->m_flags));
3847	}
3848
3849	static int receive_extralen(const struct dlm_message *ms)
3850	{
3851	return (le16_to_cpu(ms->m_header.h_length) -
3852	sizeof(struct dlm_message));
3853	}
3854
3855	static int receive_lvb(struct dlm_ls ls, struct dlm_lkb lkb,
3856	const struct dlm_message *ms)
3857	{
3858	int len;
3859
3860	if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3861	if (!lkb->lkb_lvbptr)
3862	lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3863	if (!lkb->lkb_lvbptr)
3864	return -ENOMEM;
3865	len = receive_extralen(ms);
3866	if (len > ls->ls_lvblen)
3867	len = ls->ls_lvblen;
3868	memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3869	}
3870	return 0;
3871	}
3872
3873	static void fake_bastfn(void *astparam, int mode)
3874	{
3875	log_print("fake_bastfn should not be called");
3876	}
3877
3878	static void fake_astfn(void *astparam)
3879	{
3880	log_print("fake_astfn should not be called");
3881	}
3882
3883	static int receive_request_args(struct dlm_ls ls, struct dlm_lkb lkb,
3884	const struct dlm_message *ms)
3885	{
3886	lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3887	lkb->lkb_ownpid = le32_to_cpu(ms->m_pid);
3888	lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3889	lkb->lkb_grmode = DLM_LOCK_IV;
3890	lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3891
3892	lkb->lkb_bastfn = (ms->m_asts & cpu_to_le32(DLM_CB_BAST)) ? &fake_bastfn : NULL;
3893	lkb->lkb_astfn = (ms->m_asts & cpu_to_le32(DLM_CB_CAST)) ? &fake_astfn : NULL;
3894
3895	if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3896	/* lkb was just created so there won't be an lvb yet */
3897	lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3898	if (!lkb->lkb_lvbptr)
3899	return -ENOMEM;
3900	}
3901
3902	return 0;
3903	}
3904
3905	static int receive_convert_args(struct dlm_ls ls, struct dlm_lkb lkb,
3906	const struct dlm_message *ms)
3907	{
3908	if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3909	return -EBUSY;
3910
3911	if (receive_lvb(ls, lkb, ms))
3912	return -ENOMEM;
3913
3914	lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3915	lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
3916
3917	return 0;
3918	}
3919
3920	static int receive_unlock_args(struct dlm_ls ls, struct dlm_lkb lkb,
3921	const struct dlm_message *ms)
3922	{
3923	if (receive_lvb(ls, lkb, ms))
3924	return -ENOMEM;
3925	return 0;
3926	}
3927
3928	/* We fill in the local-lkb fields with the info that send_xxxx_reply()
3929	uses to send a reply and that the remote end uses to process the reply. */
3930
3931	static void setup_local_lkb(struct dlm_ls ls, const struct dlm_message ms)
3932	{
3933	struct dlm_lkb *lkb = &ls->ls_local_lkb;
3934	lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3935	lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3936	}
3937
3938	/* This is called after the rsb is locked so that we can safely inspect
3939	fields in the lkb. */
3940
3941	static int validate_message(struct dlm_lkb lkb, const struct dlm_message ms)
3942	{
3943	int from = le32_to_cpu(ms->m_header.h_nodeid);
3944	int error = 0;
3945
3946	/* currently mixing of user/kernel locks are not supported */
3947	if (ms->m_flags & cpu_to_le32(BIT(DLM_DFL_USER_BIT)) &&
3948	!test_bit(DLM_DFL_USER_BIT, &lkb->lkb_dflags)) {
3949	log_error(lkb->lkb_resource->res_ls,
3950	"got user dlm message for a kernel lock");
3951	error = -EINVAL;
3952	goto out;
3953	}
3954
3955	switch (ms->m_type) {
3956	case cpu_to_le32(DLM_MSG_CONVERT):
3957	case cpu_to_le32(DLM_MSG_UNLOCK):
3958	case cpu_to_le32(DLM_MSG_CANCEL):
3959	if (!is_master_copy(lkb) \|\| lkb->lkb_nodeid != from)
3960	error = -EINVAL;
3961	break;
3962
3963	case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3964	case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
3965	case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
3966	case cpu_to_le32(DLM_MSG_GRANT):
3967	case cpu_to_le32(DLM_MSG_BAST):
3968	if (!is_process_copy(lkb) \|\| lkb->lkb_nodeid != from)
3969	error = -EINVAL;
3970	break;
3971
3972	case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3973	if (!is_process_copy(lkb))
3974	error = -EINVAL;
3975	else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3976	error = -EINVAL;
3977	break;
3978
3979	default:
3980	error = -EINVAL;
3981	}
3982
3983	out:
3984	if (error)
3985	log_error(lkb->lkb_resource->res_ls,
3986	"ignore invalid message %d from %d %x %x %x %d",
3987	le32_to_cpu(ms->m_type), from, lkb->lkb_id,
3988	lkb->lkb_remid, dlm_iflags_val(lkb),
3989	lkb->lkb_nodeid);
3990	return error;
3991	}
3992
3993	static int receive_request(struct dlm_ls ls, const struct dlm_message ms)
3994	{
3995	struct dlm_lkb *lkb;
3996	struct dlm_rsb *r;
3997	int from_nodeid;
3998	int error, namelen = 0;
3999
4000	from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4001
4002	error = create_lkb(ls, &lkb);
4003	if (error)
4004	goto fail;
4005
4006	receive_flags(lkb, ms);
4007	set_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags);
4008	error = receive_request_args(ls, lkb, ms);
4009	if (error) {
4010	__put_lkb(ls, lkb);
4011	goto fail;
4012	}
4013
4014	/* The dir node is the authority on whether we are the master
4015	for this rsb or not, so if the master sends us a request, we should
4016	recreate the rsb if we've destroyed it. This race happens when we
4017	send a remove message to the dir node at the same time that the dir
4018	node sends us a request for the rsb. */
4019
4020	namelen = receive_extralen(ms);
4021
4022	error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4023	R_RECEIVE_REQUEST, &r);
4024	if (error) {
4025	__put_lkb(ls, lkb);
4026	goto fail;
4027	}
4028
4029	lock_rsb(r);
4030
4031	if (r->res_master_nodeid != dlm_our_nodeid()) {
4032	error = validate_master_nodeid(ls, r, from_nodeid);
4033	if (error) {
4034	unlock_rsb(r);
4035	put_rsb(r);
4036	__put_lkb(ls, lkb);
4037	goto fail;
4038	}
4039	}
4040
4041	attach_lkb(r, lkb);
4042	error = do_request(r, lkb);
4043	send_request_reply(r, lkb, error);
4044	do_request_effects(r, lkb, error);
4045
4046	unlock_rsb(r);
4047	put_rsb(r);
4048
4049	if (error == -EINPROGRESS)
4050	error = 0;
4051	if (error)
4052	dlm_put_lkb(lkb);
4053	return 0;
4054
4055	fail:
4056	/* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4057	and do this receive_request again from process_lookup_list once
4058	we get the lookup reply. This would avoid a many repeated
4059	ENOTBLK request failures when the lookup reply designating us
4060	as master is delayed. */
4061
4062	if (error != -ENOTBLK) {
4063	log_limit(ls, "receive_request %x from %d %d",
4064	le32_to_cpu(ms->m_lkid), from_nodeid, error);
4065	}
4066
4067	setup_local_lkb(ls, ms);
4068	send_request_reply(&ls->ls_local_rsb, &ls->ls_local_lkb, error);
4069	return error;
4070	}
4071
4072	static int receive_convert(struct dlm_ls ls, const struct dlm_message ms)
4073	{
4074	struct dlm_lkb *lkb;
4075	struct dlm_rsb *r;
4076	int error, reply = 1;
4077
4078	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4079	if (error)
4080	goto fail;
4081
4082	if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4083	log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4084	"remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4085	(unsigned long long)lkb->lkb_recover_seq,
4086	le32_to_cpu(ms->m_header.h_nodeid),
4087	le32_to_cpu(ms->m_lkid));
4088	error = -ENOENT;
4089	dlm_put_lkb(lkb);
4090	goto fail;
4091	}
4092
4093	r = lkb->lkb_resource;
4094
4095	hold_rsb(r);
4096	lock_rsb(r);
4097
4098	error = validate_message(lkb, ms);
4099	if (error)
4100	goto out;
4101
4102	receive_flags(lkb, ms);
4103
4104	error = receive_convert_args(ls, lkb, ms);
4105	if (error) {
4106	send_convert_reply(r, lkb, error);
4107	goto out;
4108	}
4109
4110	reply = !down_conversion(lkb);
4111
4112	error = do_convert(r, lkb);
4113	if (reply)
4114	send_convert_reply(r, lkb, error);
4115	do_convert_effects(r, lkb, error);
4116	out:
4117	unlock_rsb(r);
4118	put_rsb(r);
4119	dlm_put_lkb(lkb);
4120	return 0;
4121
4122	fail:
4123	setup_local_lkb(ls, ms);
4124	send_convert_reply(&ls->ls_local_rsb, &ls->ls_local_lkb, error);
4125	return error;
4126	}
4127
4128	static int receive_unlock(struct dlm_ls ls, const struct dlm_message ms)
4129	{
4130	struct dlm_lkb *lkb;
4131	struct dlm_rsb *r;
4132	int error;
4133
4134	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4135	if (error)
4136	goto fail;
4137
4138	if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4139	log_error(ls, "receive_unlock %x remid %x remote %d %x",
4140	lkb->lkb_id, lkb->lkb_remid,
4141	le32_to_cpu(ms->m_header.h_nodeid),
4142	le32_to_cpu(ms->m_lkid));
4143	error = -ENOENT;
4144	dlm_put_lkb(lkb);
4145	goto fail;
4146	}
4147
4148	r = lkb->lkb_resource;
4149
4150	hold_rsb(r);
4151	lock_rsb(r);
4152
4153	error = validate_message(lkb, ms);
4154	if (error)
4155	goto out;
4156
4157	receive_flags(lkb, ms);
4158
4159	error = receive_unlock_args(ls, lkb, ms);
4160	if (error) {
4161	send_unlock_reply(r, lkb, error);
4162	goto out;
4163	}
4164
4165	error = do_unlock(r, lkb);
4166	send_unlock_reply(r, lkb, error);
4167	do_unlock_effects(r, lkb, error);
4168	out:
4169	unlock_rsb(r);
4170	put_rsb(r);
4171	dlm_put_lkb(lkb);
4172	return 0;
4173
4174	fail:
4175	setup_local_lkb(ls, ms);
4176	send_unlock_reply(&ls->ls_local_rsb, &ls->ls_local_lkb, error);
4177	return error;
4178	}
4179
4180	static int receive_cancel(struct dlm_ls ls, const struct dlm_message ms)
4181	{
4182	struct dlm_lkb *lkb;
4183	struct dlm_rsb *r;
4184	int error;
4185
4186	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4187	if (error)
4188	goto fail;
4189
4190	receive_flags(lkb, ms);
4191
4192	r = lkb->lkb_resource;
4193
4194	hold_rsb(r);
4195	lock_rsb(r);
4196
4197	error = validate_message(lkb, ms);
4198	if (error)
4199	goto out;
4200
4201	error = do_cancel(r, lkb);
4202	send_cancel_reply(r, lkb, error);
4203	do_cancel_effects(r, lkb, error);
4204	out:
4205	unlock_rsb(r);
4206	put_rsb(r);
4207	dlm_put_lkb(lkb);
4208	return 0;
4209
4210	fail:
4211	setup_local_lkb(ls, ms);
4212	send_cancel_reply(&ls->ls_local_rsb, &ls->ls_local_lkb, error);
4213	return error;
4214	}
4215
4216	static int receive_grant(struct dlm_ls ls, const struct dlm_message ms)
4217	{
4218	struct dlm_lkb *lkb;
4219	struct dlm_rsb *r;
4220	int error;
4221
4222	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4223	if (error)
4224	return error;
4225
4226	r = lkb->lkb_resource;
4227
4228	hold_rsb(r);
4229	lock_rsb(r);
4230
4231	error = validate_message(lkb, ms);
4232	if (error)
4233	goto out;
4234
4235	receive_flags_reply(lkb, ms, false);
4236	if (is_altmode(lkb))
4237	munge_altmode(lkb, ms);
4238	grant_lock_pc(r, lkb, ms);
4239	queue_cast(r, lkb, 0);
4240	out:
4241	unlock_rsb(r);
4242	put_rsb(r);
4243	dlm_put_lkb(lkb);
4244	return 0;
4245	}
4246
4247	static int receive_bast(struct dlm_ls ls, const struct dlm_message ms)
4248	{
4249	struct dlm_lkb *lkb;
4250	struct dlm_rsb *r;
4251	int error;
4252
4253	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4254	if (error)
4255	return error;
4256
4257	r = lkb->lkb_resource;
4258
4259	hold_rsb(r);
4260	lock_rsb(r);
4261
4262	error = validate_message(lkb, ms);
4263	if (error)
4264	goto out;
4265
4266	queue_bast(r, lkb, le32_to_cpu(ms->m_bastmode));
4267	lkb->lkb_highbast = le32_to_cpu(ms->m_bastmode);
4268	out:
4269	unlock_rsb(r);
4270	put_rsb(r);
4271	dlm_put_lkb(lkb);
4272	return 0;
4273	}
4274
4275	static void receive_lookup(struct dlm_ls ls, const struct dlm_message ms)
4276	{
4277	int len, error, ret_nodeid, from_nodeid, our_nodeid;
4278
4279	from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4280	our_nodeid = dlm_our_nodeid();
4281
4282	len = receive_extralen(ms);
4283
4284	error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4285	&ret_nodeid, NULL);
4286
4287	/* Optimization: we're master so treat lookup as a request */
4288	if (!error && ret_nodeid == our_nodeid) {
4289	receive_request(ls, ms);
4290	return;
4291	}
4292	send_lookup_reply(ls, ms, ret_nodeid, error);
4293	}
4294
4295	static void receive_remove(struct dlm_ls ls, const struct dlm_message ms)
4296	{
4297	char name[DLM_RESNAME_MAXLEN+1];
4298	struct dlm_rsb *r;
4299	int rv, len, dir_nodeid, from_nodeid;
4300
4301	from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4302
4303	len = receive_extralen(ms);
4304
4305	if (len > DLM_RESNAME_MAXLEN) {
4306	log_error(ls, "receive_remove from %d bad len %d",
4307	from_nodeid, len);
4308	return;
4309	}
4310
4311	dir_nodeid = dlm_hash2nodeid(ls, le32_to_cpu(ms->m_hash));
4312	if (dir_nodeid != dlm_our_nodeid()) {
4313	log_error(ls, "receive_remove from %d bad nodeid %d",
4314	from_nodeid, dir_nodeid);
4315	return;
4316	}
4317
4318	/*
4319	* Look for inactive rsb, if it's there, free it.
4320	* If the rsb is active, it's being used, and we should ignore this
4321	* message. This is an expected race between the dir node sending a
4322	* request to the master node at the same time as the master node sends
4323	* a remove to the dir node. The resolution to that race is for the
4324	* dir node to ignore the remove message, and the master node to
4325	* recreate the master rsb when it gets a request from the dir node for
4326	* an rsb it doesn't have.
4327	*/
4328
4329	memset(name, 0, sizeof(name));
4330	memcpy(name, ms->m_extra, len);
4331
4332	rcu_read_lock();
4333	rv = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);
4334	if (rv) {
4335	rcu_read_unlock();
4336	/* should not happen */
4337	log_error(ls, "%s from %d not found %s", __func__,
4338	from_nodeid, name);
4339	return;
4340	}
4341
4342	write_lock_bh(&ls->ls_rsbtbl_lock);
4343	if (!rsb_flag(r, RSB_HASHED)) {
4344	rcu_read_unlock();
4345	write_unlock_bh(&ls->ls_rsbtbl_lock);
4346	/* should not happen */
4347	log_error(ls, "%s from %d got removed during removal %s",
4348	__func__, from_nodeid, name);
4349	return;
4350	}
4351	/* at this stage the rsb can only being freed here */
4352	rcu_read_unlock();
4353
4354	if (!rsb_flag(r, RSB_INACTIVE)) {
4355	if (r->res_master_nodeid != from_nodeid) {
4356	/* should not happen */
4357	log_error(ls, "receive_remove on active rsb from %d master %d",
4358	from_nodeid, r->res_master_nodeid);
4359	dlm_print_rsb(r);
4360	write_unlock_bh(&ls->ls_rsbtbl_lock);
4361	return;
4362	}
4363
4364	/* Ignore the remove message, see race comment above. */
4365
4366	log_debug(ls, "receive_remove from %d master %d first %x %s",
4367	from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4368	name);
4369	write_unlock_bh(&ls->ls_rsbtbl_lock);
4370	return;
4371	}
4372
4373	if (r->res_master_nodeid != from_nodeid) {
4374	log_error(ls, "receive_remove inactive from %d master %d",
4375	from_nodeid, r->res_master_nodeid);
4376	dlm_print_rsb(r);
4377	write_unlock_bh(&ls->ls_rsbtbl_lock);
4378	return;
4379	}
4380
4381	list_del(&r->res_slow_list);
4382	rhashtable_remove_fast(&ls->ls_rsbtbl, &r->res_node,
4383	dlm_rhash_rsb_params);
4384	rsb_clear_flag(r, RSB_HASHED);
4385	write_unlock_bh(&ls->ls_rsbtbl_lock);
4386
4387	free_inactive_rsb(r);
4388	}
4389
4390	static void receive_purge(struct dlm_ls ls, const struct dlm_message ms)
4391	{
4392	do_purge(ls, le32_to_cpu(ms->m_nodeid), le32_to_cpu(ms->m_pid));
4393	}
4394
4395	static int receive_request_reply(struct dlm_ls *ls,
4396	const struct dlm_message *ms)
4397	{
4398	struct dlm_lkb *lkb;
4399	struct dlm_rsb *r;
4400	int error, mstype, result;
4401	int from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4402
4403	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4404	if (error)
4405	return error;
4406
4407	r = lkb->lkb_resource;
4408	hold_rsb(r);
4409	lock_rsb(r);
4410
4411	error = validate_message(lkb, ms);
4412	if (error)
4413	goto out;
4414
4415	mstype = lkb->lkb_wait_type;
4416	error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4417	if (error) {
4418	log_error(ls, "receive_request_reply %x remote %d %x result %d",
4419	lkb->lkb_id, from_nodeid, le32_to_cpu(ms->m_lkid),
4420	from_dlm_errno(le32_to_cpu(ms->m_result)));
4421	dlm_dump_rsb(r);
4422	goto out;
4423	}
4424
4425	/* Optimization: the dir node was also the master, so it took our
4426	lookup as a request and sent request reply instead of lookup reply */
4427	if (mstype == DLM_MSG_LOOKUP) {
4428	r->res_master_nodeid = from_nodeid;
4429	r->res_nodeid = from_nodeid;
4430	lkb->lkb_nodeid = from_nodeid;
4431	}
4432
4433	/* this is the value returned from do_request() on the master */
4434	result = from_dlm_errno(le32_to_cpu(ms->m_result));
4435
4436	switch (result) {
4437	case -EAGAIN:
4438	/* request would block (be queued) on remote master */
4439	queue_cast(r, lkb, -EAGAIN);
4440	confirm_master(r, -EAGAIN);
4441	unhold_lkb(lkb); /* undoes create_lkb() */
4442	break;
4443
4444	case -EINPROGRESS:
4445	case 0:
4446	/* request was queued or granted on remote master */
4447	receive_flags_reply(lkb, ms, false);
4448	lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
4449	if (is_altmode(lkb))
4450	munge_altmode(lkb, ms);
4451	if (result) {
4452	add_lkb(r, lkb, DLM_LKSTS_WAITING);
4453	} else {
4454	grant_lock_pc(r, lkb, ms);
4455	queue_cast(r, lkb, 0);
4456	}
4457	confirm_master(r, result);
4458	break;
4459
4460	case -EBADR:
4461	case -ENOTBLK:
4462	/* find_rsb failed to find rsb or rsb wasn't master */
4463	log_limit(ls, "receive_request_reply %x from %d %d "
4464	"master %d dir %d first %x %s", lkb->lkb_id,
4465	from_nodeid, result, r->res_master_nodeid,
4466	r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4467
4468	if (r->res_dir_nodeid != dlm_our_nodeid() &&
4469	r->res_master_nodeid != dlm_our_nodeid()) {
4470	/* cause _request_lock->set_master->send_lookup */
4471	r->res_master_nodeid = 0;
4472	r->res_nodeid = -1;
4473	lkb->lkb_nodeid = -1;
4474	}
4475
4476	if (is_overlap(lkb)) {
4477	/* we'll ignore error in cancel/unlock reply */
4478	queue_cast_overlap(r, lkb);
4479	confirm_master(r, result);
4480	unhold_lkb(lkb); /* undoes create_lkb() */
4481	} else {
4482	_request_lock(r, lkb);
4483
4484	if (r->res_master_nodeid == dlm_our_nodeid())
4485	confirm_master(r, 0);
4486	}
4487	break;
4488
4489	default:
4490	log_error(ls, "receive_request_reply %x error %d",
4491	lkb->lkb_id, result);
4492	}
4493
4494	if ((result == 0 \|\| result == -EINPROGRESS) &&
4495	test_and_clear_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags)) {
4496	log_debug(ls, "receive_request_reply %x result %d unlock",
4497	lkb->lkb_id, result);
4498	clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
4499	send_unlock(r, lkb);
4500	} else if ((result == -EINPROGRESS) &&
4501	test_and_clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT,
4502	&lkb->lkb_iflags)) {
4503	log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4504	clear_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
4505	send_cancel(r, lkb);
4506	} else {
4507	clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT, &lkb->lkb_iflags);
4508	clear_bit(DLM_IFL_OVERLAP_UNLOCK_BIT, &lkb->lkb_iflags);
4509	}
4510	out:
4511	unlock_rsb(r);
4512	put_rsb(r);
4513	dlm_put_lkb(lkb);
4514	return 0;
4515	}
4516
4517	static void __receive_convert_reply(struct dlm_rsb r, struct dlm_lkb lkb,
4518	const struct dlm_message *ms, bool local)
4519	{
4520	/* this is the value returned from do_convert() on the master */
4521	switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4522	case -EAGAIN:
4523	/* convert would block (be queued) on remote master */
4524	queue_cast(r, lkb, -EAGAIN);
4525	break;
4526
4527	case -EDEADLK:
4528	receive_flags_reply(lkb, ms, local);
4529	revert_lock_pc(r, lkb);
4530	queue_cast(r, lkb, -EDEADLK);
4531	break;
4532
4533	case -EINPROGRESS:
4534	/* convert was queued on remote master */
4535	receive_flags_reply(lkb, ms, local);
4536	if (is_demoted(lkb))
4537	munge_demoted(lkb);
4538	del_lkb(r, lkb);
4539	add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4540	break;
4541
4542	case 0:
4543	/* convert was granted on remote master */
4544	receive_flags_reply(lkb, ms, local);
4545	if (is_demoted(lkb))
4546	munge_demoted(lkb);
4547	grant_lock_pc(r, lkb, ms);
4548	queue_cast(r, lkb, 0);
4549	break;
4550
4551	default:
4552	log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4553	lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4554	le32_to_cpu(ms->m_lkid),
4555	from_dlm_errno(le32_to_cpu(ms->m_result)));
4556	dlm_print_rsb(r);
4557	dlm_print_lkb(lkb);
4558	}
4559	}
4560
4561	static void _receive_convert_reply(struct dlm_lkb *lkb,
4562	const struct dlm_message *ms, bool local)
4563	{
4564	struct dlm_rsb *r = lkb->lkb_resource;
4565	int error;
4566
4567	hold_rsb(r);
4568	lock_rsb(r);
4569
4570	error = validate_message(lkb, ms);
4571	if (error)
4572	goto out;
4573
4574	error = remove_from_waiters_ms(lkb, ms, local);
4575	if (error)
4576	goto out;
4577
4578	__receive_convert_reply(r, lkb, ms, local);
4579	out:
4580	unlock_rsb(r);
4581	put_rsb(r);
4582	}
4583
4584	static int receive_convert_reply(struct dlm_ls *ls,
4585	const struct dlm_message *ms)
4586	{
4587	struct dlm_lkb *lkb;
4588	int error;
4589
4590	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4591	if (error)
4592	return error;
4593
4594	_receive_convert_reply(lkb, ms, false);
4595	dlm_put_lkb(lkb);
4596	return 0;
4597	}
4598
4599	static void _receive_unlock_reply(struct dlm_lkb *lkb,
4600	const struct dlm_message *ms, bool local)
4601	{
4602	struct dlm_rsb *r = lkb->lkb_resource;
4603	int error;
4604
4605	hold_rsb(r);
4606	lock_rsb(r);
4607
4608	error = validate_message(lkb, ms);
4609	if (error)
4610	goto out;
4611
4612	error = remove_from_waiters_ms(lkb, ms, local);
4613	if (error)
4614	goto out;
4615
4616	/* this is the value returned from do_unlock() on the master */
4617
4618	switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4619	case -DLM_EUNLOCK:
4620	receive_flags_reply(lkb, ms, local);
4621	remove_lock_pc(r, lkb);
4622	queue_cast(r, lkb, -DLM_EUNLOCK);
4623	break;
4624	case -ENOENT:
4625	break;
4626	default:
4627	log_error(r->res_ls, "receive_unlock_reply %x error %d",
4628	lkb->lkb_id, from_dlm_errno(le32_to_cpu(ms->m_result)));
4629	}
4630	out:
4631	unlock_rsb(r);
4632	put_rsb(r);
4633	}
4634
4635	static int receive_unlock_reply(struct dlm_ls *ls,
4636	const struct dlm_message *ms)
4637	{
4638	struct dlm_lkb *lkb;
4639	int error;
4640
4641	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4642	if (error)
4643	return error;
4644
4645	_receive_unlock_reply(lkb, ms, false);
4646	dlm_put_lkb(lkb);
4647	return 0;
4648	}
4649
4650	static void _receive_cancel_reply(struct dlm_lkb *lkb,
4651	const struct dlm_message *ms, bool local)
4652	{
4653	struct dlm_rsb *r = lkb->lkb_resource;
4654	int error;
4655
4656	hold_rsb(r);
4657	lock_rsb(r);
4658
4659	error = validate_message(lkb, ms);
4660	if (error)
4661	goto out;
4662
4663	error = remove_from_waiters_ms(lkb, ms, local);
4664	if (error)
4665	goto out;
4666
4667	/* this is the value returned from do_cancel() on the master */
4668
4669	switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4670	case -DLM_ECANCEL:
4671	receive_flags_reply(lkb, ms, local);
4672	revert_lock_pc(r, lkb);
4673	queue_cast(r, lkb, -DLM_ECANCEL);
4674	break;
4675	case 0:
4676	break;
4677	default:
4678	log_error(r->res_ls, "receive_cancel_reply %x error %d",
4679	lkb->lkb_id,
4680	from_dlm_errno(le32_to_cpu(ms->m_result)));
4681	}
4682	out:
4683	unlock_rsb(r);
4684	put_rsb(r);
4685	}
4686
4687	static int receive_cancel_reply(struct dlm_ls *ls,
4688	const struct dlm_message *ms)
4689	{
4690	struct dlm_lkb *lkb;
4691	int error;
4692
4693	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4694	if (error)
4695	return error;
4696
4697	_receive_cancel_reply(lkb, ms, false);
4698	dlm_put_lkb(lkb);
4699	return 0;
4700	}
4701
4702	static void receive_lookup_reply(struct dlm_ls *ls,
4703	const struct dlm_message *ms)
4704	{
4705	struct dlm_lkb *lkb;
4706	struct dlm_rsb *r;
4707	int error, ret_nodeid;
4708	int do_lookup_list = 0;
4709
4710	error = find_lkb(ls, le32_to_cpu(ms->m_lkid), &lkb);
4711	if (error) {
4712	log_error(ls, "%s no lkid %x", __func__,
4713	le32_to_cpu(ms->m_lkid));
4714	return;
4715	}
4716
4717	/* ms->m_result is the value returned by dlm_master_lookup on dir node
4718	FIXME: will a non-zero error ever be returned? */
4719
4720	r = lkb->lkb_resource;
4721	hold_rsb(r);
4722	lock_rsb(r);
4723
4724	error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4725	if (error)
4726	goto out;
4727
4728	ret_nodeid = le32_to_cpu(ms->m_nodeid);
4729
4730	/* We sometimes receive a request from the dir node for this
4731	rsb before we've received the dir node's loookup_reply for it.
4732	The request from the dir node implies we're the master, so we set
4733	ourself as master in receive_request_reply, and verify here that
4734	we are indeed the master. */
4735
4736	if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4737	/* This should never happen */
4738	log_error(ls, "receive_lookup_reply %x from %d ret %d "
4739	"master %d dir %d our %d first %x %s",
4740	lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4741	ret_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
4742	dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4743	}
4744
4745	if (ret_nodeid == dlm_our_nodeid()) {
4746	r->res_master_nodeid = ret_nodeid;
4747	r->res_nodeid = 0;
4748	do_lookup_list = 1;
4749	r->res_first_lkid = 0;
4750	} else if (ret_nodeid == -1) {
4751	/* the remote node doesn't believe it's the dir node */
4752	log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4753	lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid));
4754	r->res_master_nodeid = 0;
4755	r->res_nodeid = -1;
4756	lkb->lkb_nodeid = -1;
4757	} else {
4758	/* set_master() will set lkb_nodeid from r */
4759	r->res_master_nodeid = ret_nodeid;
4760	r->res_nodeid = ret_nodeid;
4761	}
4762
4763	if (is_overlap(lkb)) {
4764	log_debug(ls, "receive_lookup_reply %x unlock %x",
4765	lkb->lkb_id, dlm_iflags_val(lkb));
4766	queue_cast_overlap(r, lkb);
4767	unhold_lkb(lkb); /* undoes create_lkb() */
4768	goto out_list;
4769	}
4770
4771	_request_lock(r, lkb);
4772
4773	out_list:
4774	if (do_lookup_list)
4775	process_lookup_list(r);
4776	out:
4777	unlock_rsb(r);
4778	put_rsb(r);
4779	dlm_put_lkb(lkb);
4780	}
4781
4782	static void _receive_message(struct dlm_ls ls, const struct dlm_message ms,
4783	uint32_t saved_seq)
4784	{
4785	int error = 0, noent = 0;
4786
4787	if (WARN_ON_ONCE(!dlm_is_member(ls, le32_to_cpu(ms->m_header.h_nodeid)))) {
4788	log_limit(ls, "receive %d from non-member %d %x %x %d",
4789	le32_to_cpu(ms->m_type),
4790	le32_to_cpu(ms->m_header.h_nodeid),
4791	le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4792	from_dlm_errno(le32_to_cpu(ms->m_result)));
4793	return;
4794	}
4795
4796	switch (ms->m_type) {
4797
4798	/* messages sent to a master node */
4799
4800	case cpu_to_le32(DLM_MSG_REQUEST):
4801	error = receive_request(ls, ms);
4802	break;
4803
4804	case cpu_to_le32(DLM_MSG_CONVERT):
4805	error = receive_convert(ls, ms);
4806	break;
4807
4808	case cpu_to_le32(DLM_MSG_UNLOCK):
4809	error = receive_unlock(ls, ms);
4810	break;
4811
4812	case cpu_to_le32(DLM_MSG_CANCEL):
4813	noent = 1;
4814	error = receive_cancel(ls, ms);
4815	break;
4816
4817	/* messages sent from a master node (replies to above) */
4818
4819	case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
4820	error = receive_request_reply(ls, ms);
4821	break;
4822
4823	case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4824	error = receive_convert_reply(ls, ms);
4825	break;
4826
4827	case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4828	error = receive_unlock_reply(ls, ms);
4829	break;
4830
4831	case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4832	error = receive_cancel_reply(ls, ms);
4833	break;
4834
4835	/* messages sent from a master node (only two types of async msg) */
4836
4837	case cpu_to_le32(DLM_MSG_GRANT):
4838	noent = 1;
4839	error = receive_grant(ls, ms);
4840	break;
4841
4842	case cpu_to_le32(DLM_MSG_BAST):
4843	noent = 1;
4844	error = receive_bast(ls, ms);
4845	break;
4846
4847	/* messages sent to a dir node */
4848
4849	case cpu_to_le32(DLM_MSG_LOOKUP):
4850	receive_lookup(ls, ms);
4851	break;
4852
4853	case cpu_to_le32(DLM_MSG_REMOVE):
4854	receive_remove(ls, ms);
4855	break;
4856
4857	/* messages sent from a dir node (remove has no reply) */
4858
4859	case cpu_to_le32(DLM_MSG_LOOKUP_REPLY):
4860	receive_lookup_reply(ls, ms);
4861	break;
4862
4863	/* other messages */
4864
4865	case cpu_to_le32(DLM_MSG_PURGE):
4866	receive_purge(ls, ms);
4867	break;
4868
4869	default:
4870	log_error(ls, "unknown message type %d",
4871	le32_to_cpu(ms->m_type));
4872	}
4873
4874	/*
4875	* When checking for ENOENT, we're checking the result of
4876	* find_lkb(m_remid):
4877	*
4878	* The lock id referenced in the message wasn't found. This may
4879	* happen in normal usage for the async messages and cancel, so
4880	* only use log_debug for them.
4881	*
4882	* Some errors are expected and normal.
4883	*/
4884
4885	if (error == -ENOENT && noent) {
4886	log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4887	le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4888	le32_to_cpu(ms->m_header.h_nodeid),
4889	le32_to_cpu(ms->m_lkid), saved_seq);
4890	} else if (error == -ENOENT) {
4891	log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4892	le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4893	le32_to_cpu(ms->m_header.h_nodeid),
4894	le32_to_cpu(ms->m_lkid), saved_seq);
4895
4896	if (ms->m_type == cpu_to_le32(DLM_MSG_CONVERT))
4897	dlm_dump_rsb_hash(ls, le32_to_cpu(ms->m_hash));
4898	}
4899
4900	if (error == -EINVAL) {
4901	log_error(ls, "receive %d inval from %d lkid %x remid %x "
4902	"saved_seq %u",
4903	le32_to_cpu(ms->m_type),
4904	le32_to_cpu(ms->m_header.h_nodeid),
4905	le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4906	saved_seq);
4907	}
4908	}
4909
4910	/* If the lockspace is in recovery mode (locking stopped), then normal
4911	messages are saved on the requestqueue for processing after recovery is
4912	done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4913	messages off the requestqueue before we process new ones. This occurs right
4914	after recovery completes when we transition from saving all messages on
4915	requestqueue, to processing all the saved messages, to processing new
4916	messages as they arrive. */
4917
4918	static void dlm_receive_message(struct dlm_ls ls, const struct dlm_message ms,
4919	int nodeid)
4920	{
4921	try_again:
4922	read_lock_bh(&ls->ls_requestqueue_lock);
4923	if (test_bit(LSFL_RECV_MSG_BLOCKED, &ls->ls_flags)) {
4924	/* If we were a member of this lockspace, left, and rejoined,
4925	other nodes may still be sending us messages from the
4926	lockspace generation before we left. */
4927	if (WARN_ON_ONCE(!ls->ls_generation)) {
4928	read_unlock_bh(&ls->ls_requestqueue_lock);
4929	log_limit(ls, "receive %d from %d ignore old gen",
4930	le32_to_cpu(ms->m_type), nodeid);
4931	return;
4932	}
4933
4934	read_unlock_bh(&ls->ls_requestqueue_lock);
4935	write_lock_bh(&ls->ls_requestqueue_lock);
4936	/* recheck because we hold writelock now */
4937	if (!test_bit(LSFL_RECV_MSG_BLOCKED, &ls->ls_flags)) {
4938	write_unlock_bh(&ls->ls_requestqueue_lock);
4939	goto try_again;
4940	}
4941
4942	dlm_add_requestqueue(ls, nodeid, ms);
4943	write_unlock_bh(&ls->ls_requestqueue_lock);
4944	} else {
4945	_receive_message(ls, ms, 0);
4946	read_unlock_bh(&ls->ls_requestqueue_lock);
4947	}
4948	}
4949
4950	/* This is called by dlm_recoverd to process messages that were saved on
4951	the requestqueue. */
4952
4953	void dlm_receive_message_saved(struct dlm_ls ls, const struct dlm_message ms,
4954	uint32_t saved_seq)
4955	{
4956	_receive_message(ls, ms, saved_seq);
4957	}
4958
4959	/* This is called by the midcomms layer when something is received for
4960	the lockspace. It could be either a MSG (normal message sent as part of
4961	standard locking activity) or an RCOM (recovery message sent as part of
4962	lockspace recovery). */
4963
4964	void dlm_receive_buffer(const union dlm_packet *p, int nodeid)
4965	{
4966	const struct dlm_header *hd = &p->header;
4967	struct dlm_ls *ls;
4968	int type = 0;
4969
4970	switch (hd->h_cmd) {
4971	case DLM_MSG:
4972	type = le32_to_cpu(p->message.m_type);
4973	break;
4974	case DLM_RCOM:
4975	type = le32_to_cpu(p->rcom.rc_type);
4976	break;
4977	default:
4978	log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
4979	return;
4980	}
4981
4982	if (le32_to_cpu(hd->h_nodeid) != nodeid) {
4983	log_print("invalid h_nodeid %d from %d lockspace %x",
4984	le32_to_cpu(hd->h_nodeid), nodeid,
4985	le32_to_cpu(hd->u.h_lockspace));
4986	return;
4987	}
4988
4989	ls = dlm_find_lockspace_global(le32_to_cpu(hd->u.h_lockspace));
4990	if (!ls) {
4991	if (dlm_config.ci_log_debug) {
4992	printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
4993	"%u from %d cmd %d type %d\n",
4994	le32_to_cpu(hd->u.h_lockspace), nodeid,
4995	hd->h_cmd, type);
4996	}
4997
4998	if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
4999	dlm_send_ls_not_ready(nodeid, &p->rcom);
5000	return;
5001	}
5002
5003	/* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5004	be inactive (in this ls) before transitioning to recovery mode */
5005
5006	read_lock_bh(&ls->ls_recv_active);
5007	if (hd->h_cmd == DLM_MSG)
5008	dlm_receive_message(ls, &p->message, nodeid);
5009	else if (hd->h_cmd == DLM_RCOM)
5010	dlm_receive_rcom(ls, &p->rcom, nodeid);
5011	else
5012	log_error(ls, "invalid h_cmd %d from %d lockspace %x",
5013	hd->h_cmd, nodeid, le32_to_cpu(hd->u.h_lockspace));
5014	read_unlock_bh(&ls->ls_recv_active);
5015
5016	dlm_put_lockspace(ls);
5017	}
5018
5019	static void recover_convert_waiter(struct dlm_ls ls, struct dlm_lkb lkb,
5020	struct dlm_message *ms_local)
5021	{
5022	if (middle_conversion(lkb) \|\| lkb->lkb_rqmode >= lkb->lkb_grmode)
5023	set_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags);
5024
5025	/* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5026	conversions are async; there's no reply from the remote master */
5027	}
5028
5029	/* A waiting lkb needs recovery if the master node has failed, or
5030	the master node is changing (only when no directory is used) */
5031
5032	static int waiter_needs_recovery(struct dlm_ls ls, struct dlm_lkb lkb,
5033	int dir_nodeid)
5034	{
5035	if (dlm_no_directory(ls))
5036	return 1;
5037
5038	if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5039	return 1;
5040
5041	return 0;
5042	}
5043
5044	/* Recovery for locks that are waiting for replies from nodes that are now
5045	gone. We can just complete unlocks and cancels by faking a reply from the
5046	dead node. Requests and up-conversions we flag to be resent after
5047	recovery. Down-conversions can just be completed with a fake reply like
5048	unlocks. Conversions between PR and CW need special attention. */
5049
5050	void dlm_recover_waiters_pre(struct dlm_ls *ls)
5051	{
5052	struct dlm_lkb lkb, safe;
5053	struct dlm_message *ms_local;
5054	int wait_type, local_unlock_result, local_cancel_result;
5055	int dir_nodeid;
5056
5057	ms_local = kmalloc_obj(*ms_local);
5058	if (!ms_local)
5059	return;
5060
5061	list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5062
5063	dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5064
5065	/* exclude debug messages about unlocks because there can be so
5066	many and they aren't very interesting */
5067
5068	if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5069	log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5070	"lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5071	lkb->lkb_id,
5072	lkb->lkb_remid,
5073	lkb->lkb_wait_type,
5074	lkb->lkb_resource->res_nodeid,
5075	lkb->lkb_nodeid,
5076	lkb->lkb_wait_nodeid,
5077	dir_nodeid);
5078	}
5079
5080	/* all outstanding lookups, regardless of destination will be
5081	resent after recovery is done */
5082
5083	if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5084	set_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags);
5085	continue;
5086	}
5087
5088	if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5089	continue;
5090
5091	wait_type = lkb->lkb_wait_type;
5092	local_unlock_result = -DLM_EUNLOCK;
5093	local_cancel_result = -DLM_ECANCEL;
5094
5095	/* Main reply may have been received leaving a zero wait_type,
5096	but a reply for the overlapping op may not have been
5097	received. In that case we need to fake the appropriate
5098	reply for the overlap op. */
5099
5100	if (!wait_type) {
5101	if (is_overlap_cancel(lkb)) {
5102	wait_type = DLM_MSG_CANCEL;
5103	if (lkb->lkb_grmode == DLM_LOCK_IV)
5104	local_cancel_result = 0;
5105	}
5106	if (is_overlap_unlock(lkb)) {
5107	wait_type = DLM_MSG_UNLOCK;
5108	if (lkb->lkb_grmode == DLM_LOCK_IV)
5109	local_unlock_result = -ENOENT;
5110	}
5111
5112	log_debug(ls, "rwpre overlap %x %x %d %d %d",
5113	lkb->lkb_id, dlm_iflags_val(lkb), wait_type,
5114	local_cancel_result, local_unlock_result);
5115	}
5116
5117	switch (wait_type) {
5118
5119	case DLM_MSG_REQUEST:
5120	set_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags);
5121	break;
5122
5123	case DLM_MSG_CONVERT:
5124	recover_convert_waiter(ls, lkb, ms_local);
5125	break;
5126
5127	case DLM_MSG_UNLOCK:
5128	hold_lkb(lkb);
5129	memset(ms_local, 0, sizeof(struct dlm_message));
5130	ms_local->m_type = cpu_to_le32(DLM_MSG_UNLOCK_REPLY);
5131	ms_local->m_result = cpu_to_le32(to_dlm_errno(local_unlock_result));
5132	ms_local->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5133	_receive_unlock_reply(lkb, ms_local, true);
5134	dlm_put_lkb(lkb);
5135	break;
5136
5137	case DLM_MSG_CANCEL:
5138	hold_lkb(lkb);
5139	memset(ms_local, 0, sizeof(struct dlm_message));
5140	ms_local->m_type = cpu_to_le32(DLM_MSG_CANCEL_REPLY);
5141	ms_local->m_result = cpu_to_le32(to_dlm_errno(local_cancel_result));
5142	ms_local->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5143	_receive_cancel_reply(lkb, ms_local, true);
5144	dlm_put_lkb(lkb);
5145	break;
5146
5147	default:
5148	log_error(ls, "invalid lkb wait_type %d %d",
5149	lkb->lkb_wait_type, wait_type);
5150	}
5151	schedule();
5152	}
5153	kfree(ms_local);
5154	}
5155
5156	static struct dlm_lkb find_resend_waiter(struct dlm_ls ls)
5157	{
5158	struct dlm_lkb lkb = NULL, iter;
5159
5160	spin_lock_bh(&ls->ls_waiters_lock);
5161	list_for_each_entry(iter, &ls->ls_waiters, lkb_wait_reply) {
5162	if (test_bit(DLM_IFL_RESEND_BIT, &iter->lkb_iflags)) {
5163	hold_lkb(iter);
5164	lkb = iter;
5165	break;
5166	}
5167	}
5168	spin_unlock_bh(&ls->ls_waiters_lock);
5169
5170	return lkb;
5171	}
5172
5173	/*
5174	* Forced state reset for locks that were in the middle of remote operations
5175	* when recovery happened (i.e. lkbs that were on the waiters list, waiting
5176	* for a reply from a remote operation.) The lkbs remaining on the waiters
5177	* list need to be reevaluated; some may need resending to a different node
5178	* than previously, and some may now need local handling rather than remote.
5179	*
5180	* First, the lkb state for the voided remote operation is forcibly reset,
5181	* equivalent to what remove_from_waiters() would normally do:
5182	* . lkb removed from ls_waiters list
5183	* . lkb wait_type cleared
5184	* . lkb waiters_count cleared
5185	* . lkb ref count decremented for each waiters_count (almost always 1,
5186	* but possibly 2 in case of cancel/unlock overlapping, which means
5187	* two remote replies were being expected for the lkb.)
5188	*
5189	* Second, the lkb is reprocessed like an original operation would be,
5190	* by passing it to _request_lock or _convert_lock, which will either
5191	* process the lkb operation locally, or send it to a remote node again
5192	* and put the lkb back onto the waiters list.
5193	*
5194	* When reprocessing the lkb, we may find that it's flagged for an overlapping
5195	* force-unlock or cancel, either from before recovery began, or after recovery
5196	* finished. If this is the case, the unlock/cancel is done directly, and the
5197	* original operation is not initiated again (no _request_lock/_convert_lock.)
5198	*/
5199
5200	int dlm_recover_waiters_post(struct dlm_ls *ls)
5201	{
5202	struct dlm_lkb *lkb;
5203	struct dlm_rsb *r;
5204	int error = 0, mstype, err, oc, ou;
5205
5206	while (1) {
5207	if (dlm_locking_stopped(ls)) {
5208	log_debug(ls, "recover_waiters_post aborted");
5209	error = -EINTR;
5210	break;
5211	}
5212
5213	/*
5214	* Find an lkb from the waiters list that's been affected by
5215	* recovery node changes, and needs to be reprocessed. Does
5216	* hold_lkb(), adding a refcount.
5217	*/
5218	lkb = find_resend_waiter(ls);
5219	if (!lkb)
5220	break;
5221
5222	r = lkb->lkb_resource;
5223	hold_rsb(r);
5224	lock_rsb(r);
5225
5226	/*
5227	* If the lkb has been flagged for a force unlock or cancel,
5228	* then the reprocessing below will be replaced by just doing
5229	* the unlock/cancel directly.
5230	*/
5231	mstype = lkb->lkb_wait_type;
5232	oc = test_and_clear_bit(DLM_IFL_OVERLAP_CANCEL_BIT,
5233	&lkb->lkb_iflags);
5234	ou = test_and_clear_bit(DLM_IFL_OVERLAP_UNLOCK_BIT,
5235	&lkb->lkb_iflags);
5236	err = 0;
5237
5238	log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5239	"lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5240	"overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5241	r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5242	dlm_dir_nodeid(r), oc, ou);
5243
5244	/*
5245	* No reply to the pre-recovery operation will now be received,
5246	* so a forced equivalent of remove_from_waiters() is needed to
5247	* reset the waiters state that was in place before recovery.
5248	*/
5249
5250	clear_bit(DLM_IFL_RESEND_BIT, &lkb->lkb_iflags);
5251
5252	/* Forcibly clear wait_type */
5253	lkb->lkb_wait_type = 0;
5254
5255	/*
5256	* Forcibly reset wait_count and associated refcount. The
5257	* wait_count will almost always be 1, but in case of an
5258	* overlapping unlock/cancel it could be 2: see where
5259	* add_to_waiters() finds the lkb is already on the waiters
5260	* list and does lkb_wait_count++; hold_lkb().
5261	*/
5262	while (lkb->lkb_wait_count) {
5263	lkb->lkb_wait_count--;
5264	unhold_lkb(lkb);
5265	}
5266
5267	/* Forcibly remove from waiters list */
5268	spin_lock_bh(&ls->ls_waiters_lock);
5269	list_del_init(&lkb->lkb_wait_reply);
5270	spin_unlock_bh(&ls->ls_waiters_lock);
5271
5272	/*
5273	* The lkb is now clear of all prior waiters state and can be
5274	* processed locally, or sent to remote node again, or directly
5275	* cancelled/unlocked.
5276	*/
5277
5278	if (oc \|\| ou) {
5279	/* do an unlock or cancel instead of resending */
5280	switch (mstype) {
5281	case DLM_MSG_LOOKUP:
5282	case DLM_MSG_REQUEST:
5283	queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5284	-DLM_ECANCEL);
5285	unhold_lkb(lkb); /* undoes create_lkb() */
5286	break;
5287	case DLM_MSG_CONVERT:
5288	if (oc) {
5289	queue_cast(r, lkb, -DLM_ECANCEL);
5290	} else {
5291	lkb->lkb_exflags \|= DLM_LKF_FORCEUNLOCK;
5292	_unlock_lock(r, lkb);
5293	}
5294	break;
5295	default:
5296	err = 1;
5297	}
5298	} else {
5299	switch (mstype) {
5300	case DLM_MSG_LOOKUP:
5301	case DLM_MSG_REQUEST:
5302	_request_lock(r, lkb);
5303	if (r->res_nodeid != -1 && is_master(r))
5304	confirm_master(r, 0);
5305	break;
5306	case DLM_MSG_CONVERT:
5307	_convert_lock(r, lkb);
5308	break;
5309	default:
5310	err = 1;
5311	}
5312	}
5313
5314	if (err) {
5315	log_error(ls, "waiter %x msg %d r_nodeid %d "
5316	"dir_nodeid %d overlap %d %d",
5317	lkb->lkb_id, mstype, r->res_nodeid,
5318	dlm_dir_nodeid(r), oc, ou);
5319	}
5320	unlock_rsb(r);
5321	put_rsb(r);
5322	dlm_put_lkb(lkb);
5323	}
5324
5325	return error;
5326	}
5327
5328	static void purge_mstcpy_list(struct dlm_ls ls, struct dlm_rsb r,
5329	struct list_head *list)
5330	{
5331	struct dlm_lkb lkb, safe;
5332
5333	list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5334	if (!is_master_copy(lkb))
5335	continue;
5336
5337	/* don't purge lkbs we've added in recover_master_copy for
5338	the current recovery seq */
5339
5340	if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5341	continue;
5342
5343	del_lkb(r, lkb);
5344
5345	/* this put should free the lkb */
5346	if (!dlm_put_lkb(lkb))
5347	log_error(ls, "purged mstcpy lkb not released");
5348	}
5349	}
5350
5351	void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5352	{
5353	struct dlm_ls *ls = r->res_ls;
5354
5355	purge_mstcpy_list(ls, r, &r->res_grantqueue);
5356	purge_mstcpy_list(ls, r, &r->res_convertqueue);
5357	purge_mstcpy_list(ls, r, &r->res_waitqueue);
5358	}
5359
5360	static void purge_dead_list(struct dlm_ls ls, struct dlm_rsb r,
5361	struct list_head *list,
5362	int nodeid_gone, unsigned int *count)
5363	{
5364	struct dlm_lkb lkb, safe;
5365
5366	list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5367	if (!is_master_copy(lkb))
5368	continue;
5369
5370	if ((lkb->lkb_nodeid == nodeid_gone) \|\|
5371	dlm_is_removed(ls, lkb->lkb_nodeid)) {
5372
5373	/* tell recover_lvb to invalidate the lvb
5374	because a node holding EX/PW failed */
5375	if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5376	(lkb->lkb_grmode >= DLM_LOCK_PW)) {
5377	rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5378	}
5379
5380	del_lkb(r, lkb);
5381
5382	/* this put should free the lkb */
5383	if (!dlm_put_lkb(lkb))
5384	log_error(ls, "purged dead lkb not released");
5385
5386	rsb_set_flag(r, RSB_RECOVER_GRANT);
5387
5388	(*count)++;
5389	}
5390	}
5391	}
5392
5393	/* Get rid of locks held by nodes that are gone. */
5394
5395	void dlm_recover_purge(struct dlm_ls ls, const struct list_head root_list)
5396	{
5397	struct dlm_rsb *r;
5398	struct dlm_member *memb;
5399	int nodes_count = 0;
5400	int nodeid_gone = 0;
5401	unsigned int lkb_count = 0;
5402
5403	/* cache one removed nodeid to optimize the common
5404	case of a single node removed */
5405
5406	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5407	nodes_count++;
5408	nodeid_gone = memb->nodeid;
5409	}
5410
5411	if (!nodes_count)
5412	return;
5413
5414	list_for_each_entry(r, root_list, res_root_list) {
5415	lock_rsb(r);
5416	if (r->res_nodeid != -1 && is_master(r)) {
5417	purge_dead_list(ls, r, &r->res_grantqueue,
5418	nodeid_gone, &lkb_count);
5419	purge_dead_list(ls, r, &r->res_convertqueue,
5420	nodeid_gone, &lkb_count);
5421	purge_dead_list(ls, r, &r->res_waitqueue,
5422	nodeid_gone, &lkb_count);
5423	}
5424	unlock_rsb(r);
5425
5426	cond_resched();
5427	}
5428
5429	if (lkb_count)
5430	log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5431	lkb_count, nodes_count);
5432	}
5433
5434	static struct dlm_rsb find_grant_rsb(struct dlm_ls ls)
5435	{
5436	struct dlm_rsb *r;
5437
5438	read_lock_bh(&ls->ls_rsbtbl_lock);
5439	list_for_each_entry(r, &ls->ls_slow_active, res_slow_list) {
5440	if (!rsb_flag(r, RSB_RECOVER_GRANT))
5441	continue;
5442	if (!is_master(r)) {
5443	rsb_clear_flag(r, RSB_RECOVER_GRANT);
5444	continue;
5445	}
5446	hold_rsb(r);
5447	read_unlock_bh(&ls->ls_rsbtbl_lock);
5448	return r;
5449	}
5450	read_unlock_bh(&ls->ls_rsbtbl_lock);
5451	return NULL;
5452	}
5453
5454	/*
5455	* Attempt to grant locks on resources that we are the master of.
5456	* Locks may have become grantable during recovery because locks
5457	* from departed nodes have been purged (or not rebuilt), allowing
5458	* previously blocked locks to now be granted. The subset of rsb's
5459	* we are interested in are those with lkb's on either the convert or
5460	* waiting queues.
5461	*
5462	* Simplest would be to go through each master rsb and check for non-empty
5463	* convert or waiting queues, and attempt to grant on those rsbs.
5464	* Checking the queues requires lock_rsb, though, for which we'd need
5465	* to release the rsbtbl lock. This would make iterating through all
5466	* rsb's very inefficient. So, we rely on earlier recovery routines
5467	* to set RECOVER_GRANT on any rsb's that we should attempt to grant
5468	* locks for.
5469	*/
5470
5471	void dlm_recover_grant(struct dlm_ls *ls)
5472	{
5473	struct dlm_rsb *r;
5474	unsigned int count = 0;
5475	unsigned int rsb_count = 0;
5476	unsigned int lkb_count = 0;
5477
5478	while (1) {
5479	r = find_grant_rsb(ls);
5480	if (!r)
5481	break;
5482
5483	rsb_count++;
5484	count = 0;
5485	lock_rsb(r);
5486	/* the RECOVER_GRANT flag is checked in the grant path */
5487	grant_pending_locks(r, &count);
5488	rsb_clear_flag(r, RSB_RECOVER_GRANT);
5489	lkb_count += count;
5490	confirm_master(r, 0);
5491	unlock_rsb(r);
5492	put_rsb(r);
5493	cond_resched();
5494	}
5495
5496	if (lkb_count)
5497	log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5498	lkb_count, rsb_count);
5499	}
5500
5501	static struct dlm_lkb search_remid_list(struct list_head head, int nodeid,
5502	uint32_t remid)
5503	{
5504	struct dlm_lkb *lkb;
5505
5506	list_for_each_entry(lkb, head, lkb_statequeue) {
5507	if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5508	return lkb;
5509	}
5510	return NULL;
5511	}
5512
5513	static struct dlm_lkb search_remid(struct dlm_rsb r, int nodeid,
5514	uint32_t remid)
5515	{
5516	struct dlm_lkb *lkb;
5517
5518	lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5519	if (lkb)
5520	return lkb;
5521	lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5522	if (lkb)
5523	return lkb;
5524	lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5525	if (lkb)
5526	return lkb;
5527	return NULL;
5528	}
5529
5530	/* needs at least dlm_rcom + rcom_lock */
5531	static int receive_rcom_lock_args(struct dlm_ls ls, struct dlm_lkb lkb,
5532	struct dlm_rsb r, const struct dlm_rcom rc)
5533	{
5534	struct rcom_lock rl = (struct rcom_lock ) rc->rc_buf;
5535
5536	if (rl->rl_rqmode < DLM_LOCK_IV \|\| rl->rl_rqmode > DLM_LOCK_EX \|\|
5537	rl->rl_grmode < DLM_LOCK_IV \|\| rl->rl_grmode > DLM_LOCK_EX)
5538	return -EINVAL;
5539
5540	lkb->lkb_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5541	lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5542	lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5543	lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5544	dlm_set_dflags_val(lkb, le32_to_cpu(rl->rl_flags));
5545	set_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags);
5546	lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5547	lkb->lkb_rqmode = rl->rl_rqmode;
5548	lkb->lkb_grmode = rl->rl_grmode;
5549	/* don't set lkb_status because add_lkb wants to itself */
5550
5551	lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5552	lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5553
5554	if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5555	int lvblen = le16_to_cpu(rc->rc_header.h_length) -
5556	sizeof(struct dlm_rcom) - sizeof(struct rcom_lock);
5557	if (lvblen > ls->ls_lvblen)
5558	return -EINVAL;
5559	lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5560	if (!lkb->lkb_lvbptr)
5561	return -ENOMEM;
5562	memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5563	}
5564
5565	/* Conversions between PR and CW (middle modes) need special handling.
5566	The real granted mode of these converting locks cannot be determined
5567	until all locks have been rebuilt on the rsb (recover_conversion) */
5568
5569	if (rl->rl_status == DLM_LKSTS_CONVERT && middle_conversion(lkb)) {
5570	/* We may need to adjust grmode depending on other granted locks. */
5571	log_rinfo(ls, "%s %x middle convert gr %d rq %d remote %d %x",
5572	__func__, lkb->lkb_id, lkb->lkb_grmode,
5573	lkb->lkb_rqmode, lkb->lkb_nodeid, lkb->lkb_remid);
5574	rsb_set_flag(r, RSB_RECOVER_CONVERT);
5575	}
5576
5577	return 0;
5578	}
5579
5580	/* This lkb may have been recovered in a previous aborted recovery so we need
5581	to check if the rsb already has an lkb with the given remote nodeid/lkid.
5582	If so we just send back a standard reply. If not, we create a new lkb with
5583	the given values and send back our lkid. We send back our lkid by sending
5584	back the rcom_lock struct we got but with the remid field filled in. */
5585
5586	/* needs at least dlm_rcom + rcom_lock */
5587	int dlm_recover_master_copy(struct dlm_ls ls, const struct dlm_rcom rc,
5588	__le32 rl_remid, __le32 rl_result)
5589	{
5590	struct rcom_lock rl = (struct rcom_lock ) rc->rc_buf;
5591	struct dlm_rsb *r;
5592	struct dlm_lkb *lkb;
5593	uint32_t remid = 0;
5594	int from_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5595	int error;
5596
5597	/* init rl_remid with rcom lock rl_remid */
5598	*rl_remid = rl->rl_remid;
5599
5600	if (rl->rl_parent_lkid) {
5601	error = -EOPNOTSUPP;
5602	goto out;
5603	}
5604
5605	remid = le32_to_cpu(rl->rl_lkid);
5606
5607	/* In general we expect the rsb returned to be R_MASTER, but we don't
5608	have to require it. Recovery of masters on one node can overlap
5609	recovery of locks on another node, so one node can send us MSTCPY
5610	locks before we've made ourselves master of this rsb. We can still
5611	add new MSTCPY locks that we receive here without any harm; when
5612	we make ourselves master, dlm_recover_masters() won't touch the
5613	MSTCPY locks we've received early. */
5614
5615	error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5616	from_nodeid, R_RECEIVE_RECOVER, &r);
5617	if (error)
5618	goto out;
5619
5620	lock_rsb(r);
5621
5622	if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5623	log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5624	from_nodeid, remid);
5625	error = -EBADR;
5626	goto out_unlock;
5627	}
5628
5629	lkb = search_remid(r, from_nodeid, remid);
5630	if (lkb) {
5631	error = -EEXIST;
5632	goto out_remid;
5633	}
5634
5635	error = create_lkb(ls, &lkb);
5636	if (error)
5637	goto out_unlock;
5638
5639	error = receive_rcom_lock_args(ls, lkb, r, rc);
5640	if (error) {
5641	__put_lkb(ls, lkb);
5642	goto out_unlock;
5643	}
5644
5645	attach_lkb(r, lkb);
5646	add_lkb(r, lkb, rl->rl_status);
5647	ls->ls_recover_locks_in++;
5648
5649	if (!list_empty(&r->res_waitqueue) \|\| !list_empty(&r->res_convertqueue))
5650	rsb_set_flag(r, RSB_RECOVER_GRANT);
5651
5652	out_remid:
5653	/* this is the new value returned to the lock holder for
5654	saving in its process-copy lkb */
5655	*rl_remid = cpu_to_le32(lkb->lkb_id);
5656
5657	lkb->lkb_recover_seq = ls->ls_recover_seq;
5658
5659	out_unlock:
5660	unlock_rsb(r);
5661	put_rsb(r);
5662	out:
5663	if (error && error != -EEXIST)
5664	log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5665	from_nodeid, remid, error);
5666	*rl_result = cpu_to_le32(error);
5667	return error;
5668	}
5669
5670	/* needs at least dlm_rcom + rcom_lock */
5671	int dlm_recover_process_copy(struct dlm_ls ls, const struct dlm_rcom rc,
5672	uint64_t seq)
5673	{
5674	struct rcom_lock rl = (struct rcom_lock ) rc->rc_buf;
5675	struct dlm_rsb *r;
5676	struct dlm_lkb *lkb;
5677	uint32_t lkid, remid;
5678	int error, result;
5679
5680	lkid = le32_to_cpu(rl->rl_lkid);
5681	remid = le32_to_cpu(rl->rl_remid);
5682	result = le32_to_cpu(rl->rl_result);
5683
5684	error = find_lkb(ls, lkid, &lkb);
5685	if (error) {
5686	log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5687	lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5688	result);
5689	return error;
5690	}
5691
5692	r = lkb->lkb_resource;
5693	hold_rsb(r);
5694	lock_rsb(r);
5695
5696	if (!is_process_copy(lkb)) {
5697	log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5698	lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5699	result);
5700	dlm_dump_rsb(r);
5701	unlock_rsb(r);
5702	put_rsb(r);
5703	dlm_put_lkb(lkb);
5704	return -EINVAL;
5705	}
5706
5707	switch (result) {
5708	case -EBADR:
5709	/* There's a chance the new master received our lock before
5710	dlm_recover_master_reply(), this wouldn't happen if we did
5711	a barrier between recover_masters and recover_locks. */
5712
5713	log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5714	lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5715	result);
5716
5717	dlm_send_rcom_lock(r, lkb, seq);
5718	goto out;
5719	case -EEXIST:
5720	case 0:
5721	lkb->lkb_remid = remid;
5722	break;
5723	default:
5724	log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5725	lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5726	result);
5727	}
5728
5729	/* an ack for dlm_recover_locks() which waits for replies from
5730	all the locks it sends to new masters */
5731	dlm_recovered_lock(r);
5732	out:
5733	unlock_rsb(r);
5734	put_rsb(r);
5735	dlm_put_lkb(lkb);
5736
5737	return 0;
5738	}
5739
5740	int dlm_user_request(struct dlm_ls ls, struct dlm_user_args ua,
5741	int mode, uint32_t flags, void *name, unsigned int namelen)
5742	{
5743	struct dlm_lkb *lkb;
5744	struct dlm_args args;
5745	bool do_put = true;
5746	int error;
5747
5748	dlm_lock_recovery(ls);
5749
5750	error = create_lkb(ls, &lkb);
5751	if (error) {
5752	kfree(ua);
5753	goto out;
5754	}
5755
5756	trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
5757
5758	if (flags & DLM_LKF_VALBLK) {
5759	ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5760	if (!ua->lksb.sb_lvbptr) {
5761	kfree(ua);
5762	error = -ENOMEM;
5763	goto out_put;
5764	}
5765	}
5766	error = set_lock_args(mode, &ua->lksb, flags, namelen, fake_astfn, ua,
5767	fake_bastfn, &args);
5768	if (error) {
5769	kfree(ua->lksb.sb_lvbptr);
5770	ua->lksb.sb_lvbptr = NULL;
5771	kfree(ua);
5772	goto out_put;
5773	}
5774
5775	/* After ua is attached to lkb it will be freed by dlm_free_lkb().
5776	When DLM_DFL_USER_BIT is set, the dlm knows that this is a userspace
5777	lock and that lkb_astparam is the dlm_user_args structure. */
5778	set_bit(DLM_DFL_USER_BIT, &lkb->lkb_dflags);
5779	error = request_lock(ls, lkb, name, namelen, &args);
5780
5781	switch (error) {
5782	case 0:
5783	break;
5784	case -EINPROGRESS:
5785	error = 0;
5786	break;
5787	case -EAGAIN:
5788	error = 0;
5789	fallthrough;
5790	default:
5791	goto out_put;
5792	}
5793
5794	/* add this new lkb to the per-process list of locks */
5795	spin_lock_bh(&ua->proc->locks_spin);
5796	hold_lkb(lkb);
5797	list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5798	spin_unlock_bh(&ua->proc->locks_spin);
5799	do_put = false;
5800	out_put:
5801	trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, false);
5802	if (do_put)
5803	__put_lkb(ls, lkb);
5804	out:
5805	dlm_unlock_recovery(ls);
5806	return error;
5807	}
5808
5809	int dlm_user_convert(struct dlm_ls ls, struct dlm_user_args ua_tmp,
5810	int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
5811	{
5812	struct dlm_lkb *lkb;
5813	struct dlm_args args;
5814	struct dlm_user_args *ua;
5815	int error;
5816
5817	dlm_lock_recovery(ls);
5818
5819	error = find_lkb(ls, lkid, &lkb);
5820	if (error)
5821	goto out;
5822
5823	trace_dlm_lock_start(ls, lkb, NULL, 0, mode, flags);
5824
5825	/* user can change the params on its lock when it converts it, or
5826	add an lvb that didn't exist before */
5827
5828	ua = lkb->lkb_ua;
5829
5830	if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5831	ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5832	if (!ua->lksb.sb_lvbptr) {
5833	error = -ENOMEM;
5834	goto out_put;
5835	}
5836	}
5837	if (lvb_in && ua->lksb.sb_lvbptr)
5838	memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5839
5840	ua->xid = ua_tmp->xid;
5841	ua->castparam = ua_tmp->castparam;
5842	ua->castaddr = ua_tmp->castaddr;
5843	ua->bastparam = ua_tmp->bastparam;
5844	ua->bastaddr = ua_tmp->bastaddr;
5845	ua->user_lksb = ua_tmp->user_lksb;
5846
5847	error = set_lock_args(mode, &ua->lksb, flags, 0, fake_astfn, ua,
5848	fake_bastfn, &args);
5849	if (error)
5850	goto out_put;
5851
5852	error = convert_lock(ls, lkb, &args);
5853
5854	if (error == -EINPROGRESS \|\| error == -EAGAIN \|\| error == -EDEADLK)
5855	error = 0;
5856	out_put:
5857	trace_dlm_lock_end(ls, lkb, NULL, 0, mode, flags, error, false);
5858	dlm_put_lkb(lkb);
5859	out:
5860	dlm_unlock_recovery(ls);
5861	kfree(ua_tmp);
5862	return error;
5863	}
5864
5865	/*
5866	* The caller asks for an orphan lock on a given resource with a given mode.
5867	* If a matching lock exists, it's moved to the owner's list of locks and
5868	* the lkid is returned.
5869	*/
5870
5871	int dlm_user_adopt_orphan(struct dlm_ls ls, struct dlm_user_args ua_tmp,
5872	int mode, uint32_t flags, void *name, unsigned int namelen,
5873	uint32_t *lkid)
5874	{
5875	struct dlm_lkb lkb = NULL, iter;
5876	struct dlm_user_args *ua;
5877	int found_other_mode = 0;
5878	int rv = 0;
5879
5880	spin_lock_bh(&ls->ls_orphans_lock);
5881	list_for_each_entry(iter, &ls->ls_orphans, lkb_ownqueue) {
5882	if (iter->lkb_resource->res_length != namelen)
5883	continue;
5884	if (memcmp(iter->lkb_resource->res_name, name, namelen))
5885	continue;
5886	if (iter->lkb_grmode != mode) {
5887	found_other_mode = 1;
5888	continue;
5889	}
5890
5891	lkb = iter;
5892	list_del_init(&iter->lkb_ownqueue);
5893	clear_bit(DLM_DFL_ORPHAN_BIT, &iter->lkb_dflags);
5894	*lkid = iter->lkb_id;
5895	break;
5896	}
5897	spin_unlock_bh(&ls->ls_orphans_lock);
5898
5899	if (!lkb && found_other_mode) {
5900	rv = -EAGAIN;
5901	goto out;
5902	}
5903
5904	if (!lkb) {
5905	rv = -ENOENT;
5906	goto out;
5907	}
5908
5909	lkb->lkb_exflags = flags;
5910	lkb->lkb_ownpid = (int) current->pid;
5911
5912	ua = lkb->lkb_ua;
5913
5914	ua->proc = ua_tmp->proc;
5915	ua->xid = ua_tmp->xid;
5916	ua->castparam = ua_tmp->castparam;
5917	ua->castaddr = ua_tmp->castaddr;
5918	ua->bastparam = ua_tmp->bastparam;
5919	ua->bastaddr = ua_tmp->bastaddr;
5920	ua->user_lksb = ua_tmp->user_lksb;
5921
5922	/*
5923	* The lkb reference from the ls_orphans list was not
5924	* removed above, and is now considered the reference
5925	* for the proc locks list.
5926	*/
5927
5928	spin_lock_bh(&ua->proc->locks_spin);
5929	list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5930	spin_unlock_bh(&ua->proc->locks_spin);
5931	out:
5932	kfree(ua_tmp);
5933	return rv;
5934	}
5935
5936	int dlm_user_unlock(struct dlm_ls ls, struct dlm_user_args ua_tmp,
5937	uint32_t flags, uint32_t lkid, char *lvb_in)
5938	{
5939	struct dlm_lkb *lkb;
5940	struct dlm_args args;
5941	struct dlm_user_args *ua;
5942	int error;
5943
5944	dlm_lock_recovery(ls);
5945
5946	error = find_lkb(ls, lkid, &lkb);
5947	if (error)
5948	goto out;
5949
5950	trace_dlm_unlock_start(ls, lkb, flags);
5951
5952	ua = lkb->lkb_ua;
5953
5954	if (lvb_in && ua->lksb.sb_lvbptr)
5955	memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5956	if (ua_tmp->castparam)
5957	ua->castparam = ua_tmp->castparam;
5958	ua->user_lksb = ua_tmp->user_lksb;
5959
5960	error = set_unlock_args(flags, ua, &args);
5961	if (error)
5962	goto out_put;
5963
5964	error = unlock_lock(ls, lkb, &args);
5965
5966	if (error == -DLM_EUNLOCK)
5967	error = 0;
5968	/* from validate_unlock_args() */
5969	if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
5970	error = 0;
5971	if (error)
5972	goto out_put;
5973
5974	spin_lock_bh(&ua->proc->locks_spin);
5975	/* dlm_user_add_cb() may have already taken lkb off the proc list */
5976	if (!list_empty(&lkb->lkb_ownqueue))
5977	list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
5978	spin_unlock_bh(&ua->proc->locks_spin);
5979	out_put:
5980	trace_dlm_unlock_end(ls, lkb, flags, error);
5981	dlm_put_lkb(lkb);
5982	out:
5983	dlm_unlock_recovery(ls);
5984	kfree(ua_tmp);
5985	return error;
5986	}
5987
5988	int dlm_user_cancel(struct dlm_ls ls, struct dlm_user_args ua_tmp,
5989	uint32_t flags, uint32_t lkid)
5990	{
5991	struct dlm_lkb *lkb;
5992	struct dlm_args args;
5993	struct dlm_user_args *ua;
5994	int error;
5995
5996	dlm_lock_recovery(ls);
5997
5998	error = find_lkb(ls, lkid, &lkb);
5999	if (error)
6000	goto out;
6001
6002	trace_dlm_unlock_start(ls, lkb, flags);
6003
6004	ua = lkb->lkb_ua;
6005	if (ua_tmp->castparam)
6006	ua->castparam = ua_tmp->castparam;
6007	ua->user_lksb = ua_tmp->user_lksb;
6008
6009	error = set_unlock_args(flags, ua, &args);
6010	if (error)
6011	goto out_put;
6012
6013	error = cancel_lock(ls, lkb, &args);
6014
6015	if (error == -DLM_ECANCEL)
6016	error = 0;
6017	/* from validate_unlock_args() */
6018	if (error == -EBUSY)
6019	error = 0;
6020	out_put:
6021	trace_dlm_unlock_end(ls, lkb, flags, error);
6022	dlm_put_lkb(lkb);
6023	out:
6024	dlm_unlock_recovery(ls);
6025	kfree(ua_tmp);
6026	return error;
6027	}
6028
6029	int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6030	{
6031	struct dlm_lkb *lkb;
6032	struct dlm_args args;
6033	struct dlm_user_args *ua;
6034	struct dlm_rsb *r;
6035	int error;
6036
6037	dlm_lock_recovery(ls);
6038
6039	error = find_lkb(ls, lkid, &lkb);
6040	if (error)
6041	goto out;
6042
6043	trace_dlm_unlock_start(ls, lkb, flags);
6044
6045	ua = lkb->lkb_ua;
6046
6047	error = set_unlock_args(flags, ua, &args);
6048	if (error)
6049	goto out_put;
6050
6051	/* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6052
6053	r = lkb->lkb_resource;
6054	hold_rsb(r);
6055	lock_rsb(r);
6056
6057	error = validate_unlock_args(lkb, &args);
6058	if (error)
6059	goto out_r;
6060	set_bit(DLM_IFL_DEADLOCK_CANCEL_BIT, &lkb->lkb_iflags);
6061
6062	error = _cancel_lock(r, lkb);
6063	out_r:
6064	unlock_rsb(r);
6065	put_rsb(r);
6066
6067	if (error == -DLM_ECANCEL)
6068	error = 0;
6069	/* from validate_unlock_args() */
6070	if (error == -EBUSY)
6071	error = 0;
6072	out_put:
6073	trace_dlm_unlock_end(ls, lkb, flags, error);
6074	dlm_put_lkb(lkb);
6075	out:
6076	dlm_unlock_recovery(ls);
6077	return error;
6078	}
6079
6080	/* lkb's that are removed from the waiters list by revert are just left on the
6081	orphans list with the granted orphan locks, to be freed by purge */
6082
6083	static int orphan_proc_lock(struct dlm_ls ls, struct dlm_lkb lkb)
6084	{
6085	struct dlm_args args;
6086	int error;
6087
6088	hold_lkb(lkb); /* reference for the ls_orphans list */
6089	spin_lock_bh(&ls->ls_orphans_lock);
6090	list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6091	spin_unlock_bh(&ls->ls_orphans_lock);
6092
6093	set_unlock_args(0, lkb->lkb_ua, &args);
6094
6095	error = cancel_lock(ls, lkb, &args);
6096	if (error == -DLM_ECANCEL)
6097	error = 0;
6098	return error;
6099	}
6100
6101	/* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6102	granted. Regardless of what rsb queue the lock is on, it's removed and
6103	freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6104	if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6105
6106	static int unlock_proc_lock(struct dlm_ls ls, struct dlm_lkb lkb)
6107	{
6108	struct dlm_args args;
6109	int error;
6110
6111	set_unlock_args(DLM_LKF_FORCEUNLOCK \| DLM_LKF_IVVALBLK,
6112	lkb->lkb_ua, &args);
6113
6114	error = unlock_lock(ls, lkb, &args);
6115	if (error == -DLM_EUNLOCK)
6116	error = 0;
6117	return error;
6118	}
6119
6120	/* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6121	(which does lock_rsb) due to deadlock with receiving a message that does
6122	lock_rsb followed by dlm_user_add_cb() */
6123
6124	static struct dlm_lkb del_proc_lock(struct dlm_ls ls,
6125	struct dlm_user_proc *proc)
6126	{
6127	struct dlm_lkb *lkb = NULL;
6128
6129	spin_lock_bh(&ls->ls_clear_proc_locks);
6130	if (list_empty(&proc->locks))
6131	goto out;
6132
6133	lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6134	list_del_init(&lkb->lkb_ownqueue);
6135
6136	if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6137	set_bit(DLM_DFL_ORPHAN_BIT, &lkb->lkb_dflags);
6138	else
6139	set_bit(DLM_IFL_DEAD_BIT, &lkb->lkb_iflags);
6140	out:
6141	spin_unlock_bh(&ls->ls_clear_proc_locks);
6142	return lkb;
6143	}
6144
6145	/* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6146	1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6147	which we clear here. */
6148
6149	/* proc CLOSING flag is set so no more device_reads should look at proc->asts
6150	list, and no more device_writes should add lkb's to proc->locks list; so we
6151	shouldn't need to take asts_spin or locks_spin here. this assumes that
6152	device reads/writes/closes are serialized -- FIXME: we may need to serialize
6153	them ourself. */
6154
6155	void dlm_clear_proc_locks(struct dlm_ls ls, struct dlm_user_proc proc)
6156	{
6157	struct dlm_callback cb, cb_safe;
6158	struct dlm_lkb lkb, safe;
6159
6160	dlm_lock_recovery(ls);
6161
6162	while (1) {
6163	lkb = del_proc_lock(ls, proc);
6164	if (!lkb)
6165	break;
6166	if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6167	orphan_proc_lock(ls, lkb);
6168	else
6169	unlock_proc_lock(ls, lkb);
6170
6171	/* this removes the reference for the proc->locks list
6172	added by dlm_user_request, it may result in the lkb
6173	being freed */
6174
6175	dlm_put_lkb(lkb);
6176	}
6177
6178	spin_lock_bh(&ls->ls_clear_proc_locks);
6179
6180	/* in-progress unlocks */
6181	list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6182	list_del_init(&lkb->lkb_ownqueue);
6183	set_bit(DLM_IFL_DEAD_BIT, &lkb->lkb_iflags);
6184	dlm_put_lkb(lkb);
6185	}
6186
6187	list_for_each_entry_safe(cb, cb_safe, &proc->asts, list) {
6188	list_del(&cb->list);
6189	dlm_free_cb(cb);
6190	}
6191
6192	spin_unlock_bh(&ls->ls_clear_proc_locks);
6193	dlm_unlock_recovery(ls);
6194	}
6195
6196	static void purge_proc_locks(struct dlm_ls ls, struct dlm_user_proc proc)
6197	{
6198	struct dlm_callback cb, cb_safe;
6199	struct dlm_lkb lkb, safe;
6200
6201	while (1) {
6202	lkb = NULL;
6203	spin_lock_bh(&proc->locks_spin);
6204	if (!list_empty(&proc->locks)) {
6205	lkb = list_entry(proc->locks.next, struct dlm_lkb,
6206	lkb_ownqueue);
6207	list_del_init(&lkb->lkb_ownqueue);
6208	}
6209	spin_unlock_bh(&proc->locks_spin);
6210
6211	if (!lkb)
6212	break;
6213
6214	set_bit(DLM_IFL_DEAD_BIT, &lkb->lkb_iflags);
6215	unlock_proc_lock(ls, lkb);
6216	dlm_put_lkb(lkb); /* ref from proc->locks list */
6217	}
6218
6219	spin_lock_bh(&proc->locks_spin);
6220	list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6221	list_del_init(&lkb->lkb_ownqueue);
6222	set_bit(DLM_IFL_DEAD_BIT, &lkb->lkb_iflags);
6223	dlm_put_lkb(lkb);
6224	}
6225	spin_unlock_bh(&proc->locks_spin);
6226
6227	spin_lock_bh(&proc->asts_spin);
6228	list_for_each_entry_safe(cb, cb_safe, &proc->asts, list) {
6229	list_del(&cb->list);
6230	dlm_free_cb(cb);
6231	}
6232	spin_unlock_bh(&proc->asts_spin);
6233	}
6234
6235	/* pid of 0 means purge all orphans */
6236
6237	static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6238	{
6239	struct dlm_lkb lkb, safe;
6240
6241	spin_lock_bh(&ls->ls_orphans_lock);
6242	list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6243	if (pid && lkb->lkb_ownpid != pid)
6244	continue;
6245	unlock_proc_lock(ls, lkb);
6246	list_del_init(&lkb->lkb_ownqueue);
6247	dlm_put_lkb(lkb);
6248	}
6249	spin_unlock_bh(&ls->ls_orphans_lock);
6250	}
6251
6252	static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6253	{
6254	struct dlm_message *ms;
6255	struct dlm_mhandle *mh;
6256	int error;
6257
6258	error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6259	DLM_MSG_PURGE, &ms, &mh);
6260	if (error)
6261	return error;
6262	ms->m_nodeid = cpu_to_le32(nodeid);
6263	ms->m_pid = cpu_to_le32(pid);
6264
6265	return send_message(mh, ms, NULL, 0);
6266	}
6267
6268	int dlm_user_purge(struct dlm_ls ls, struct dlm_user_proc proc,
6269	int nodeid, int pid)
6270	{
6271	int error = 0;
6272
6273	if (nodeid && (nodeid != dlm_our_nodeid())) {
6274	error = send_purge(ls, nodeid, pid);
6275	} else {
6276	dlm_lock_recovery(ls);
6277	if (pid == current->pid)
6278	purge_proc_locks(ls, proc);
6279	else
6280	do_purge(ls, nodeid, pid);
6281	dlm_unlock_recovery(ls);
6282	}
6283	return error;
6284	}
6285
6286	/* debug functionality */
6287	int dlm_debug_add_lkb(struct dlm_ls ls, uint32_t lkb_id, char name, int len,
6288	int lkb_nodeid, unsigned int lkb_dflags, int lkb_status)
6289	{
6290	struct dlm_lksb *lksb;
6291	struct dlm_lkb *lkb;
6292	struct dlm_rsb *r;
6293	int error;
6294
6295	/* we currently can't set a valid user lock */
6296	if (lkb_dflags & BIT(DLM_DFL_USER_BIT))
6297	return -EOPNOTSUPP;
6298
6299	lksb = kzalloc_obj(*lksb, GFP_NOFS);
6300	if (!lksb)
6301	return -ENOMEM;
6302
6303	error = _create_lkb(ls, &lkb, lkb_id, lkb_id + 1);
6304	if (error) {
6305	kfree(lksb);
6306	return error;
6307	}
6308
6309	dlm_set_dflags_val(lkb, lkb_dflags);
6310	lkb->lkb_nodeid = lkb_nodeid;
6311	lkb->lkb_lksb = lksb;
6312	/* user specific pointer, just don't have it NULL for kernel locks */
6313	if (~lkb_dflags & BIT(DLM_DFL_USER_BIT))
6314	lkb->lkb_astparam = (void *)0xDEADBEEF;
6315
6316	error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
6317	if (error) {
6318	kfree(lksb);
6319	__put_lkb(ls, lkb);
6320	return error;
6321	}
6322
6323	lock_rsb(r);
6324	attach_lkb(r, lkb);
6325	add_lkb(r, lkb, lkb_status);
6326	unlock_rsb(r);
6327	put_rsb(r);
6328
6329	return 0;
6330	}
6331
6332	int dlm_debug_add_lkb_to_waiters(struct dlm_ls *ls, uint32_t lkb_id,
6333	int mstype, int to_nodeid)
6334	{
6335	struct dlm_lkb *lkb;
6336	int error;
6337
6338	error = find_lkb(ls, lkb_id, &lkb);
6339	if (error)
6340	return error;
6341
6342	add_to_waiters(lkb, mstype, to_nodeid);
6343	dlm_put_lkb(lkb);
6344	return 0;
6345	}
6346