projects / linux-2.6-block.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
Merge branch 'for-2.6.25' of git://git.kernel.org/pub/scm/linux/kernel/git/paulus...
[linux-2.6-block.git] / fs / dlm / lock.c
CommitLineData
e7fd4179
DT		 1/******************************************************************************
2*******************************************************************************
3**
46b43eed 4** Copyright (C) 2005-2008 Red Hat, Inc. All rights reserved.
e7fd4179
DT		 5**
6** This copyrighted material is made available to anyone wishing to use,
7** modify, copy, or redistribute it subject to the terms and conditions
8** of the GNU General Public License v.2.
9**
10*******************************************************************************
11******************************************************************************/
12
13/* Central locking logic has four stages:
14
15 dlm_lock()
16 dlm_unlock()
17
18 request_lock(ls, lkb)
19 convert_lock(ls, lkb)
20 unlock_lock(ls, lkb)
21 cancel_lock(ls, lkb)
22
23 _request_lock(r, lkb)
24 _convert_lock(r, lkb)
25 _unlock_lock(r, lkb)
26 _cancel_lock(r, lkb)
27
28 do_request(r, lkb)
29 do_convert(r, lkb)
30 do_unlock(r, lkb)
31 do_cancel(r, lkb)
32
33 Stage 1 (lock, unlock) is mainly about checking input args and
34 splitting into one of the four main operations:
35
36 dlm_lock = request_lock
37 dlm_lock+CONVERT = convert_lock
38 dlm_unlock = unlock_lock
39 dlm_unlock+CANCEL = cancel_lock
40
41 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
42 provided to the next stage.
43
44 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
45 When remote, it calls send_xxxx(), when local it calls do_xxxx().
46
47 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
48 given rsb and lkb and queues callbacks.
49
50 For remote operations, send_xxxx() results in the corresponding do_xxxx()
51 function being executed on the remote node. The connecting send/receive
52 calls on local (L) and remote (R) nodes:
53
54 L: send_xxxx() -> R: receive_xxxx()
55 R: do_xxxx()
56 L: receive_xxxx_reply() <- R: send_xxxx_reply()
57*/
597d0cae 58#include <linux/types.h>
e7fd4179 59#include "dlm_internal.h"
597d0cae 60#include <linux/dlm_device.h>
e7fd4179
DT		 61#include "memory.h"
62#include "lowcomms.h"
63#include "requestqueue.h"
64#include "util.h"
65#include "dir.h"
66#include "member.h"
67#include "lockspace.h"
68#include "ast.h"
69#include "lock.h"
70#include "rcom.h"
71#include "recover.h"
72#include "lvb_table.h"
597d0cae 73#include "user.h"
e7fd4179
DT		 74#include "config.h"
75
76static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
77static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
78static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
79static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
80static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
81static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
82static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
83static int send_remove(struct dlm_rsb *r);
84static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
3ae1acf9 85static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
e7fd4179
DT		 86static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
87 struct dlm_message *ms);
88static int receive_extralen(struct dlm_message *ms);
8499137d 89static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
3ae1acf9 90static void del_timeout(struct dlm_lkb *lkb);
e7fd4179
DT		 91
92/*
93 * Lock compatibilty matrix - thanks Steve
94 * UN = Unlocked state. Not really a state, used as a flag
95 * PD = Padding. Used to make the matrix a nice power of two in size
96 * Other states are the same as the VMS DLM.
97 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
98 */
99
100static const int __dlm_compat_matrix[8][8] = {
101 /* UN NL CR CW PR PW EX PD */
102 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
103 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
104 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
105 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
106 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
107 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
108 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
109 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
110};
111
112/*
113 * This defines the direction of transfer of LVB data.
114 * Granted mode is the row; requested mode is the column.
115 * Usage: matrix[grmode+1][rqmode+1]
116 * 1 = LVB is returned to the caller
117 * 0 = LVB is written to the resource
118 * -1 = nothing happens to the LVB
119 */
120
121const int dlm_lvb_operations[8][8] = {
122 /* UN NL CR CW PR PW EX PD*/
123 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
124 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
125 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
126 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
127 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
128 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
129 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
130 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
131};
e7fd4179
DT		 132
133#define modes_compat(gr, rq) \
134 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
135
136int dlm_modes_compat(int mode1, int mode2)
137{
138 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
139}
140
141/*
142 * Compatibility matrix for conversions with QUECVT set.
143 * Granted mode is the row; requested mode is the column.
144 * Usage: matrix[grmode+1][rqmode+1]
145 */
146
147static const int __quecvt_compat_matrix[8][8] = {
148 /* UN NL CR CW PR PW EX PD */
149 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
150 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
151 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
152 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
153 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
154 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
155 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
156 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
157};
158
597d0cae 159void dlm_print_lkb(struct dlm_lkb *lkb)
e7fd4179
DT		 160{
161 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
162 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
163 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
164 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
165 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
166}
167
168void dlm_print_rsb(struct dlm_rsb *r)
169{
170 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
171 r->res_nodeid, r->res_flags, r->res_first_lkid,
172 r->res_recover_locks_count, r->res_name);
173}
174
a345da3e
DT		 175void dlm_dump_rsb(struct dlm_rsb *r)
176{
177 struct dlm_lkb *lkb;
178
179 dlm_print_rsb(r);
180
181 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
182 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
183 printk(KERN_ERR "rsb lookup list\n");
184 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
185 dlm_print_lkb(lkb);
186 printk(KERN_ERR "rsb grant queue:\n");
187 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
188 dlm_print_lkb(lkb);
189 printk(KERN_ERR "rsb convert queue:\n");
190 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
191 dlm_print_lkb(lkb);
192 printk(KERN_ERR "rsb wait queue:\n");
193 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
194 dlm_print_lkb(lkb);
195}
196
e7fd4179
DT		 197/* Threads cannot use the lockspace while it's being recovered */
198
85e86edf 199static inline void dlm_lock_recovery(struct dlm_ls *ls)
e7fd4179
DT		 200{
201 down_read(&ls->ls_in_recovery);
202}
203
85e86edf 204void dlm_unlock_recovery(struct dlm_ls *ls)
e7fd4179
DT		 205{
206 up_read(&ls->ls_in_recovery);
207}
208
85e86edf 209int dlm_lock_recovery_try(struct dlm_ls *ls)
e7fd4179
DT		 210{
211 return down_read_trylock(&ls->ls_in_recovery);
212}
213
214static inline int can_be_queued(struct dlm_lkb *lkb)
215{
216 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
217}
218
219static inline int force_blocking_asts(struct dlm_lkb *lkb)
220{
221 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
222}
223
224static inline int is_demoted(struct dlm_lkb *lkb)
225{
226 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
227}
228
7d3c1feb
DT		 229static inline int is_altmode(struct dlm_lkb *lkb)
230{
231 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
232}
233
234static inline int is_granted(struct dlm_lkb *lkb)
235{
236 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
237}
238
e7fd4179
DT		 239static inline int is_remote(struct dlm_rsb *r)
240{
241 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
242 return !!r->res_nodeid;
243}
244
245static inline int is_process_copy(struct dlm_lkb *lkb)
246{
247 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
248}
249
250static inline int is_master_copy(struct dlm_lkb *lkb)
251{
252 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
253 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
90135925 254 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
e7fd4179
DT		 255}
256
257static inline int middle_conversion(struct dlm_lkb *lkb)
258{
259 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
260 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
90135925
DT		 261 return 1;
262 return 0;
e7fd4179
DT		 263}
264
265static inline int down_conversion(struct dlm_lkb *lkb)
266{
267 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
268}
269
ef0c2bb0
DT		 270static inline int is_overlap_unlock(struct dlm_lkb *lkb)
271{
272 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
273}
274
275static inline int is_overlap_cancel(struct dlm_lkb *lkb)
276{
277 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
278}
279
280static inline int is_overlap(struct dlm_lkb *lkb)
281{
282 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
283 DLM_IFL_OVERLAP_CANCEL));
284}
285
e7fd4179
DT		 286static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
287{
288 if (is_master_copy(lkb))
289 return;
290
3ae1acf9
DT		 291 del_timeout(lkb);
292
e7fd4179
DT		 293 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
294
3ae1acf9
DT		 295 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
296 timeout caused the cancel then return -ETIMEDOUT */
297 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
298 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
299 rv = -ETIMEDOUT;
300 }
301
8b4021fa
DT		 302 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
303 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
304 rv = -EDEADLK;
305 }
306
e7fd4179
DT		 307 lkb->lkb_lksb->sb_status = rv;
308 lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
309
310 dlm_add_ast(lkb, AST_COMP);
311}
312
ef0c2bb0
DT		 313static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
314{
315 queue_cast(r, lkb,
316 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
317}
318
e7fd4179
DT		 319static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
320{
321 if (is_master_copy(lkb))
322 send_bast(r, lkb, rqmode);
323 else {
324 lkb->lkb_bastmode = rqmode;
325 dlm_add_ast(lkb, AST_BAST);
326 }
327}
328
329/*
330 * Basic operations on rsb's and lkb's
331 */
332
333static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
334{
335 struct dlm_rsb *r;
336
52bda2b5 337 r = dlm_allocate_rsb(ls, len);
e7fd4179
DT		 338 if (!r)
339 return NULL;
340
341 r->res_ls = ls;
342 r->res_length = len;
343 memcpy(r->res_name, name, len);
90135925 344 mutex_init(&r->res_mutex);
e7fd4179
DT		 345
346 INIT_LIST_HEAD(&r->res_lookup);
347 INIT_LIST_HEAD(&r->res_grantqueue);
348 INIT_LIST_HEAD(&r->res_convertqueue);
349 INIT_LIST_HEAD(&r->res_waitqueue);
350 INIT_LIST_HEAD(&r->res_root_list);
351 INIT_LIST_HEAD(&r->res_recover_list);
352
353 return r;
354}
355
356static int search_rsb_list(struct list_head *head, char *name, int len,
357 unsigned int flags, struct dlm_rsb **r_ret)
358{
359 struct dlm_rsb *r;
360 int error = 0;
361
362 list_for_each_entry(r, head, res_hashchain) {
363 if (len == r->res_length && !memcmp(name, r->res_name, len))
364 goto found;
365 }
597d0cae 366 return -EBADR;
e7fd4179
DT		 367
368 found:
369 if (r->res_nodeid && (flags & R_MASTER))
370 error = -ENOTBLK;
371 *r_ret = r;
372 return error;
373}
374
375static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
376 unsigned int flags, struct dlm_rsb **r_ret)
377{
378 struct dlm_rsb *r;
379 int error;
380
381 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
382 if (!error) {
383 kref_get(&r->res_ref);
384 goto out;
385 }
386 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
387 if (error)
388 goto out;
389
390 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
391
392 if (dlm_no_directory(ls))
393 goto out;
394
395 if (r->res_nodeid == -1) {
396 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
397 r->res_first_lkid = 0;
398 } else if (r->res_nodeid > 0) {
399 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
400 r->res_first_lkid = 0;
401 } else {
402 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
403 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
404 }
405 out:
406 *r_ret = r;
407 return error;
408}
409
410static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
411 unsigned int flags, struct dlm_rsb **r_ret)
412{
413 int error;
414 write_lock(&ls->ls_rsbtbl[b].lock);
415 error = _search_rsb(ls, name, len, b, flags, r_ret);
416 write_unlock(&ls->ls_rsbtbl[b].lock);
417 return error;
418}
419
420/*
421 * Find rsb in rsbtbl and potentially create/add one
422 *
423 * Delaying the release of rsb's has a similar benefit to applications keeping
424 * NL locks on an rsb, but without the guarantee that the cached master value
425 * will still be valid when the rsb is reused. Apps aren't always smart enough
426 * to keep NL locks on an rsb that they may lock again shortly; this can lead
427 * to excessive master lookups and removals if we don't delay the release.
428 *
429 * Searching for an rsb means looking through both the normal list and toss
430 * list. When found on the toss list the rsb is moved to the normal list with
431 * ref count of 1; when found on normal list the ref count is incremented.
432 */
433
434static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
435 unsigned int flags, struct dlm_rsb **r_ret)
436{
437 struct dlm_rsb *r, *tmp;
438 uint32_t hash, bucket;
439 int error = 0;
440
441 if (dlm_no_directory(ls))
442 flags |= R_CREATE;
443
444 hash = jhash(name, namelen, 0);
445 bucket = hash & (ls->ls_rsbtbl_size - 1);
446
447 error = search_rsb(ls, name, namelen, bucket, flags, &r);
448 if (!error)
449 goto out;
450
597d0cae 451 if (error == -EBADR && !(flags & R_CREATE))
e7fd4179
DT		 452 goto out;
453
454 /* the rsb was found but wasn't a master copy */
455 if (error == -ENOTBLK)
456 goto out;
457
458 error = -ENOMEM;
459 r = create_rsb(ls, name, namelen);
460 if (!r)
461 goto out;
462
463 r->res_hash = hash;
464 r->res_bucket = bucket;
465 r->res_nodeid = -1;
466 kref_init(&r->res_ref);
467
468 /* With no directory, the master can be set immediately */
469 if (dlm_no_directory(ls)) {
470 int nodeid = dlm_dir_nodeid(r);
471 if (nodeid == dlm_our_nodeid())
472 nodeid = 0;
473 r->res_nodeid = nodeid;
474 }
475
476 write_lock(&ls->ls_rsbtbl[bucket].lock);
477 error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
478 if (!error) {
479 write_unlock(&ls->ls_rsbtbl[bucket].lock);
52bda2b5 480 dlm_free_rsb(r);
e7fd4179
DT		 481 r = tmp;
482 goto out;
483 }
484 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
485 write_unlock(&ls->ls_rsbtbl[bucket].lock);
486 error = 0;
487 out:
488 *r_ret = r;
489 return error;
490}
491
e7fd4179
DT		 492/* This is only called to add a reference when the code already holds
493 a valid reference to the rsb, so there's no need for locking. */
494
495static inline void hold_rsb(struct dlm_rsb *r)
496{
497 kref_get(&r->res_ref);
498}
499
500void dlm_hold_rsb(struct dlm_rsb *r)
501{
502 hold_rsb(r);
503}
504
505static void toss_rsb(struct kref *kref)
506{
507 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
508 struct dlm_ls *ls = r->res_ls;
509
510 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
511 kref_init(&r->res_ref);
512 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
513 r->res_toss_time = jiffies;
514 if (r->res_lvbptr) {
52bda2b5 515 dlm_free_lvb(r->res_lvbptr);
e7fd4179
DT		 516 r->res_lvbptr = NULL;
517 }
518}
519
520/* When all references to the rsb are gone it's transfered to
521 the tossed list for later disposal. */
522
523static void put_rsb(struct dlm_rsb *r)
524{
525 struct dlm_ls *ls = r->res_ls;
526 uint32_t bucket = r->res_bucket;
527
528 write_lock(&ls->ls_rsbtbl[bucket].lock);
529 kref_put(&r->res_ref, toss_rsb);
530 write_unlock(&ls->ls_rsbtbl[bucket].lock);
531}
532
533void dlm_put_rsb(struct dlm_rsb *r)
534{
535 put_rsb(r);
536}
537
538/* See comment for unhold_lkb */
539
540static void unhold_rsb(struct dlm_rsb *r)
541{
542 int rv;
543 rv = kref_put(&r->res_ref, toss_rsb);
a345da3e 544 DLM_ASSERT(!rv, dlm_dump_rsb(r););
e7fd4179
DT		 545}
546
547static void kill_rsb(struct kref *kref)
548{
549 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
550
551 /* All work is done after the return from kref_put() so we
552 can release the write_lock before the remove and free. */
553
a345da3e
DT		 554 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
555 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
556 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
557 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
558 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
559 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
e7fd4179
DT		 560}
561
562/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
563 The rsb must exist as long as any lkb's for it do. */
564
565static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
566{
567 hold_rsb(r);
568 lkb->lkb_resource = r;
569}
570
571static void detach_lkb(struct dlm_lkb *lkb)
572{
573 if (lkb->lkb_resource) {
574 put_rsb(lkb->lkb_resource);
575 lkb->lkb_resource = NULL;
576 }
577}
578
579static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
580{
581 struct dlm_lkb *lkb, *tmp;
582 uint32_t lkid = 0;
583 uint16_t bucket;
584
52bda2b5 585 lkb = dlm_allocate_lkb(ls);
e7fd4179
DT		 586 if (!lkb)
587 return -ENOMEM;
588
589 lkb->lkb_nodeid = -1;
590 lkb->lkb_grmode = DLM_LOCK_IV;
591 kref_init(&lkb->lkb_ref);
34e22bed 592 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
ef0c2bb0 593 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
3ae1acf9 594 INIT_LIST_HEAD(&lkb->lkb_time_list);
e7fd4179
DT		 595
596 get_random_bytes(&bucket, sizeof(bucket));
597 bucket &= (ls->ls_lkbtbl_size - 1);
598
599 write_lock(&ls->ls_lkbtbl[bucket].lock);
600
601 /* counter can roll over so we must verify lkid is not in use */
602
603 while (lkid == 0) {
ce03f12b 604 lkid = (bucket << 16) | ls->ls_lkbtbl[bucket].counter++;
e7fd4179
DT		 605
606 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
607 lkb_idtbl_list) {
608 if (tmp->lkb_id != lkid)
609 continue;
610 lkid = 0;
611 break;
612 }
613 }
614
615 lkb->lkb_id = lkid;
616 list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
617 write_unlock(&ls->ls_lkbtbl[bucket].lock);
618
619 *lkb_ret = lkb;
620 return 0;
621}
622
623static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
624{
e7fd4179 625 struct dlm_lkb *lkb;
ce03f12b 626 uint16_t bucket = (lkid >> 16);
e7fd4179
DT		 627
628 list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
629 if (lkb->lkb_id == lkid)
630 return lkb;
631 }
632 return NULL;
633}
634
635static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
636{
637 struct dlm_lkb *lkb;
ce03f12b 638 uint16_t bucket = (lkid >> 16);
e7fd4179
DT		 639
640 if (bucket >= ls->ls_lkbtbl_size)
641 return -EBADSLT;
642
643 read_lock(&ls->ls_lkbtbl[bucket].lock);
644 lkb = __find_lkb(ls, lkid);
645 if (lkb)
646 kref_get(&lkb->lkb_ref);
647 read_unlock(&ls->ls_lkbtbl[bucket].lock);
648
649 *lkb_ret = lkb;
650 return lkb ? 0 : -ENOENT;
651}
652
653static void kill_lkb(struct kref *kref)
654{
655 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
656
657 /* All work is done after the return from kref_put() so we
658 can release the write_lock before the detach_lkb */
659
660 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
661}
662
b3f58d8f
DT		 663/* __put_lkb() is used when an lkb may not have an rsb attached to
664 it so we need to provide the lockspace explicitly */
665
666static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
e7fd4179 667{
ce03f12b 668 uint16_t bucket = (lkb->lkb_id >> 16);
e7fd4179
DT		 669
670 write_lock(&ls->ls_lkbtbl[bucket].lock);
671 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
672 list_del(&lkb->lkb_idtbl_list);
673 write_unlock(&ls->ls_lkbtbl[bucket].lock);
674
675 detach_lkb(lkb);
676
677 /* for local/process lkbs, lvbptr points to caller's lksb */
678 if (lkb->lkb_lvbptr && is_master_copy(lkb))
52bda2b5
DT		 679 dlm_free_lvb(lkb->lkb_lvbptr);
680 dlm_free_lkb(lkb);
e7fd4179
DT		 681 return 1;
682 } else {
683 write_unlock(&ls->ls_lkbtbl[bucket].lock);
684 return 0;
685 }
686}
687
688int dlm_put_lkb(struct dlm_lkb *lkb)
689{
b3f58d8f
DT		 690 struct dlm_ls *ls;
691
692 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
693 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
694
695 ls = lkb->lkb_resource->res_ls;
696 return __put_lkb(ls, lkb);
e7fd4179
DT		 697}
698
699/* This is only called to add a reference when the code already holds
700 a valid reference to the lkb, so there's no need for locking. */
701
702static inline void hold_lkb(struct dlm_lkb *lkb)
703{
704 kref_get(&lkb->lkb_ref);
705}
706
707/* This is called when we need to remove a reference and are certain
708 it's not the last ref. e.g. del_lkb is always called between a
709 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
710 put_lkb would work fine, but would involve unnecessary locking */
711
712static inline void unhold_lkb(struct dlm_lkb *lkb)
713{
714 int rv;
715 rv = kref_put(&lkb->lkb_ref, kill_lkb);
716 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
717}
718
719static void lkb_add_ordered(struct list_head *new, struct list_head *head,
720 int mode)
721{
722 struct dlm_lkb *lkb = NULL;
723
724 list_for_each_entry(lkb, head, lkb_statequeue)
725 if (lkb->lkb_rqmode < mode)
726 break;
727
728 if (!lkb)
729 list_add_tail(new, head);
730 else
731 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
732}
733
734/* add/remove lkb to rsb's grant/convert/wait queue */
735
736static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
737{
738 kref_get(&lkb->lkb_ref);
739
740 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
741
742 lkb->lkb_status = status;
743
744 switch (status) {
745 case DLM_LKSTS_WAITING:
746 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
747 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
748 else
749 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
750 break;
751 case DLM_LKSTS_GRANTED:
752 /* convention says granted locks kept in order of grmode */
753 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
754 lkb->lkb_grmode);
755 break;
756 case DLM_LKSTS_CONVERT:
757 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
758 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
759 else
760 list_add_tail(&lkb->lkb_statequeue,
761 &r->res_convertqueue);
762 break;
763 default:
764 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
765 }
766}
767
768static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
769{
770 lkb->lkb_status = 0;
771 list_del(&lkb->lkb_statequeue);
772 unhold_lkb(lkb);
773}
774
775static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
776{
777 hold_lkb(lkb);
778 del_lkb(r, lkb);
779 add_lkb(r, lkb, sts);
780 unhold_lkb(lkb);
781}
782
ef0c2bb0
DT		 783static int msg_reply_type(int mstype)
784{
785 switch (mstype) {
786 case DLM_MSG_REQUEST:
787 return DLM_MSG_REQUEST_REPLY;
788 case DLM_MSG_CONVERT:
789 return DLM_MSG_CONVERT_REPLY;
790 case DLM_MSG_UNLOCK:
791 return DLM_MSG_UNLOCK_REPLY;
792 case DLM_MSG_CANCEL:
793 return DLM_MSG_CANCEL_REPLY;
794 case DLM_MSG_LOOKUP:
795 return DLM_MSG_LOOKUP_REPLY;
796 }
797 return -1;
798}
799
e7fd4179
DT		 800/* add/remove lkb from global waiters list of lkb's waiting for
801 a reply from a remote node */
802
ef0c2bb0 803static int add_to_waiters(struct dlm_lkb *lkb, int mstype)
e7fd4179
DT		 804{
805 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
ef0c2bb0 806 int error = 0;
e7fd4179 807
90135925 808 mutex_lock(&ls->ls_waiters_mutex);
ef0c2bb0
DT		 809
810 if (is_overlap_unlock(lkb) ||
811 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
812 error = -EINVAL;
813 goto out;
814 }
815
816 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
817 switch (mstype) {
818 case DLM_MSG_UNLOCK:
819 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
820 break;
821 case DLM_MSG_CANCEL:
822 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
823 break;
824 default:
825 error = -EBUSY;
826 goto out;
827 }
828 lkb->lkb_wait_count++;
829 hold_lkb(lkb);
830
831 log_debug(ls, "add overlap %x cur %d new %d count %d flags %x",
832 lkb->lkb_id, lkb->lkb_wait_type, mstype,
833 lkb->lkb_wait_count, lkb->lkb_flags);
e7fd4179
DT		 834 goto out;
835 }
ef0c2bb0
DT		 836
837 DLM_ASSERT(!lkb->lkb_wait_count,
838 dlm_print_lkb(lkb);
839 printk("wait_count %d\n", lkb->lkb_wait_count););
840
841 lkb->lkb_wait_count++;
e7fd4179 842 lkb->lkb_wait_type = mstype;
ef0c2bb0 843 hold_lkb(lkb);
e7fd4179
DT		 844 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
845 out:
ef0c2bb0
DT		 846 if (error)
847 log_error(ls, "add_to_waiters %x error %d flags %x %d %d %s",
848 lkb->lkb_id, error, lkb->lkb_flags, mstype,
849 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
90135925 850 mutex_unlock(&ls->ls_waiters_mutex);
ef0c2bb0 851 return error;
e7fd4179
DT		 852}
853
b790c3b7
DT		 854/* We clear the RESEND flag because we might be taking an lkb off the waiters
855 list as part of process_requestqueue (e.g. a lookup that has an optimized
856 request reply on the requestqueue) between dlm_recover_waiters_pre() which
857 set RESEND and dlm_recover_waiters_post() */
858
ef0c2bb0 859static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype)
e7fd4179 860{
ef0c2bb0
DT		 861 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
862 int overlap_done = 0;
e7fd4179 863
ef0c2bb0
DT		 864 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
865 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
866 overlap_done = 1;
867 goto out_del;
e7fd4179 868 }
ef0c2bb0
DT		 869
870 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
871 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
872 overlap_done = 1;
873 goto out_del;
874 }
875
876 /* N.B. type of reply may not always correspond to type of original
877 msg due to lookup->request optimization, verify others? */
878
879 if (lkb->lkb_wait_type) {
880 lkb->lkb_wait_type = 0;
881 goto out_del;
882 }
883
884 log_error(ls, "remove_from_waiters lkid %x flags %x types %d %d",
885 lkb->lkb_id, lkb->lkb_flags, mstype, lkb->lkb_wait_type);
886 return -1;
887
888 out_del:
889 /* the force-unlock/cancel has completed and we haven't recvd a reply
890 to the op that was in progress prior to the unlock/cancel; we
891 give up on any reply to the earlier op. FIXME: not sure when/how
892 this would happen */
893
894 if (overlap_done && lkb->lkb_wait_type) {
895 log_error(ls, "remove_from_waiters %x reply %d give up on %d",
896 lkb->lkb_id, mstype, lkb->lkb_wait_type);
897 lkb->lkb_wait_count--;
898 lkb->lkb_wait_type = 0;
899 }
900
901 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
902
b790c3b7 903 lkb->lkb_flags &= ~DLM_IFL_RESEND;
ef0c2bb0
DT		 904 lkb->lkb_wait_count--;
905 if (!lkb->lkb_wait_count)
906 list_del_init(&lkb->lkb_wait_reply);
e7fd4179 907 unhold_lkb(lkb);
ef0c2bb0 908 return 0;
e7fd4179
DT		 909}
910
ef0c2bb0 911static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
e7fd4179
DT		 912{
913 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
914 int error;
915
90135925 916 mutex_lock(&ls->ls_waiters_mutex);
ef0c2bb0 917 error = _remove_from_waiters(lkb, mstype);
90135925 918 mutex_unlock(&ls->ls_waiters_mutex);
e7fd4179
DT		 919 return error;
920}
921
ef0c2bb0
DT		 922/* Handles situations where we might be processing a "fake" or "stub" reply in
923 which we can't try to take waiters_mutex again. */
924
925static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
926{
927 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
928 int error;
929
930 if (ms != &ls->ls_stub_ms)
931 mutex_lock(&ls->ls_waiters_mutex);
932 error = _remove_from_waiters(lkb, ms->m_type);
933 if (ms != &ls->ls_stub_ms)
934 mutex_unlock(&ls->ls_waiters_mutex);
935 return error;
936}
937
e7fd4179
DT		 938static void dir_remove(struct dlm_rsb *r)
939{
940 int to_nodeid;
941
942 if (dlm_no_directory(r->res_ls))
943 return;
944
945 to_nodeid = dlm_dir_nodeid(r);
946 if (to_nodeid != dlm_our_nodeid())
947 send_remove(r);
948 else
949 dlm_dir_remove_entry(r->res_ls, to_nodeid,
950 r->res_name, r->res_length);
951}
952
953/* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
954 found since they are in order of newest to oldest? */
955
956static int shrink_bucket(struct dlm_ls *ls, int b)
957{
958 struct dlm_rsb *r;
959 int count = 0, found;
960
961 for (;;) {
90135925 962 found = 0;
e7fd4179
DT		 963 write_lock(&ls->ls_rsbtbl[b].lock);
964 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
965 res_hashchain) {
966 if (!time_after_eq(jiffies, r->res_toss_time +
68c817a1 967 dlm_config.ci_toss_secs * HZ))
e7fd4179 968 continue;
90135925 969 found = 1;
e7fd4179
DT		 970 break;
971 }
972
973 if (!found) {
974 write_unlock(&ls->ls_rsbtbl[b].lock);
975 break;
976 }
977
978 if (kref_put(&r->res_ref, kill_rsb)) {
979 list_del(&r->res_hashchain);
980 write_unlock(&ls->ls_rsbtbl[b].lock);
981
982 if (is_master(r))
983 dir_remove(r);
52bda2b5 984 dlm_free_rsb(r);
e7fd4179
DT		 985 count++;
986 } else {
987 write_unlock(&ls->ls_rsbtbl[b].lock);
988 log_error(ls, "tossed rsb in use %s", r->res_name);
989 }
990 }
991
992 return count;
993}
994
995void dlm_scan_rsbs(struct dlm_ls *ls)
996{
997 int i;
998
e7fd4179
DT		 999 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1000 shrink_bucket(ls, i);
85e86edf
DT		 1001 if (dlm_locking_stopped(ls))
1002 break;
e7fd4179
DT		 1003 cond_resched();
1004 }
1005}
1006
3ae1acf9
DT		 1007static void add_timeout(struct dlm_lkb *lkb)
1008{
1009 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1010
84d8cd69
DT		 1011 if (is_master_copy(lkb)) {
1012 lkb->lkb_timestamp = jiffies;
3ae1acf9 1013 return;
84d8cd69 1014 }
3ae1acf9
DT		 1015
1016 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1017 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1018 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1019 goto add_it;
1020 }
84d8cd69
DT		 1021 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1022 goto add_it;
3ae1acf9
DT		 1023 return;
1024
1025 add_it:
1026 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1027 mutex_lock(&ls->ls_timeout_mutex);
1028 hold_lkb(lkb);
1029 lkb->lkb_timestamp = jiffies;
1030 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1031 mutex_unlock(&ls->ls_timeout_mutex);
1032}
1033
1034static void del_timeout(struct dlm_lkb *lkb)
1035{
1036 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1037
1038 mutex_lock(&ls->ls_timeout_mutex);
1039 if (!list_empty(&lkb->lkb_time_list)) {
1040 list_del_init(&lkb->lkb_time_list);
1041 unhold_lkb(lkb);
1042 }
1043 mutex_unlock(&ls->ls_timeout_mutex);
1044}
1045
1046/* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1047 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1048 and then lock rsb because of lock ordering in add_timeout. We may need
1049 to specify some special timeout-related bits in the lkb that are just to
1050 be accessed under the timeout_mutex. */
1051
1052void dlm_scan_timeout(struct dlm_ls *ls)
1053{
1054 struct dlm_rsb *r;
1055 struct dlm_lkb *lkb;
1056 int do_cancel, do_warn;
1057
1058 for (;;) {
1059 if (dlm_locking_stopped(ls))
1060 break;
1061
1062 do_cancel = 0;
1063 do_warn = 0;
1064 mutex_lock(&ls->ls_timeout_mutex);
1065 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1066
1067 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1068 time_after_eq(jiffies, lkb->lkb_timestamp +
1069 lkb->lkb_timeout_cs * HZ/100))
1070 do_cancel = 1;
1071
1072 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1073 time_after_eq(jiffies, lkb->lkb_timestamp +
1074 dlm_config.ci_timewarn_cs * HZ/100))
1075 do_warn = 1;
1076
1077 if (!do_cancel && !do_warn)
1078 continue;
1079 hold_lkb(lkb);
1080 break;
1081 }
1082 mutex_unlock(&ls->ls_timeout_mutex);
1083
1084 if (!do_cancel && !do_warn)
1085 break;
1086
1087 r = lkb->lkb_resource;
1088 hold_rsb(r);
1089 lock_rsb(r);
1090
1091 if (do_warn) {
1092 /* clear flag so we only warn once */
1093 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1094 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1095 del_timeout(lkb);
1096 dlm_timeout_warn(lkb);
1097 }
1098
1099 if (do_cancel) {
b3cab7b9 1100 log_debug(ls, "timeout cancel %x node %d %s",
639aca41 1101 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
3ae1acf9
DT		 1102 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1103 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1104 del_timeout(lkb);
1105 _cancel_lock(r, lkb);
1106 }
1107
1108 unlock_rsb(r);
1109 unhold_rsb(r);
1110 dlm_put_lkb(lkb);
1111 }
1112}
1113
1114/* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1115 dlm_recoverd before checking/setting ls_recover_begin. */
1116
1117void dlm_adjust_timeouts(struct dlm_ls *ls)
1118{
1119 struct dlm_lkb *lkb;
1120 long adj = jiffies - ls->ls_recover_begin;
1121
1122 ls->ls_recover_begin = 0;
1123 mutex_lock(&ls->ls_timeout_mutex);
1124 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1125 lkb->lkb_timestamp += adj;
1126 mutex_unlock(&ls->ls_timeout_mutex);
1127}
1128
e7fd4179
DT		 1129/* lkb is master or local copy */
1130
1131static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1132{
1133 int b, len = r->res_ls->ls_lvblen;
1134
1135 /* b=1 lvb returned to caller
1136 b=0 lvb written to rsb or invalidated
1137 b=-1 do nothing */
1138
1139 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1140
1141 if (b == 1) {
1142 if (!lkb->lkb_lvbptr)
1143 return;
1144
1145 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1146 return;
1147
1148 if (!r->res_lvbptr)
1149 return;
1150
1151 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1152 lkb->lkb_lvbseq = r->res_lvbseq;
1153
1154 } else if (b == 0) {
1155 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1156 rsb_set_flag(r, RSB_VALNOTVALID);
1157 return;
1158 }
1159
1160 if (!lkb->lkb_lvbptr)
1161 return;
1162
1163 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1164 return;
1165
1166 if (!r->res_lvbptr)
52bda2b5 1167 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
e7fd4179
DT		 1168
1169 if (!r->res_lvbptr)
1170 return;
1171
1172 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1173 r->res_lvbseq++;
1174 lkb->lkb_lvbseq = r->res_lvbseq;
1175 rsb_clear_flag(r, RSB_VALNOTVALID);
1176 }
1177
1178 if (rsb_flag(r, RSB_VALNOTVALID))
1179 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1180}
1181
1182static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1183{
1184 if (lkb->lkb_grmode < DLM_LOCK_PW)
1185 return;
1186
1187 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1188 rsb_set_flag(r, RSB_VALNOTVALID);
1189 return;
1190 }
1191
1192 if (!lkb->lkb_lvbptr)
1193 return;
1194
1195 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1196 return;
1197
1198 if (!r->res_lvbptr)
52bda2b5 1199 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
e7fd4179
DT		 1200
1201 if (!r->res_lvbptr)
1202 return;
1203
1204 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1205 r->res_lvbseq++;
1206 rsb_clear_flag(r, RSB_VALNOTVALID);
1207}
1208
1209/* lkb is process copy (pc) */
1210
1211static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1212 struct dlm_message *ms)
1213{
1214 int b;
1215
1216 if (!lkb->lkb_lvbptr)
1217 return;
1218
1219 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1220 return;
1221
597d0cae 1222 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
e7fd4179
DT		 1223 if (b == 1) {
1224 int len = receive_extralen(ms);
1225 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1226 lkb->lkb_lvbseq = ms->m_lvbseq;
1227 }
1228}
1229
1230/* Manipulate lkb's on rsb's convert/granted/waiting queues
1231 remove_lock -- used for unlock, removes lkb from granted
1232 revert_lock -- used for cancel, moves lkb from convert to granted
1233 grant_lock -- used for request and convert, adds lkb to granted or
1234 moves lkb from convert or waiting to granted
1235
1236 Each of these is used for master or local copy lkb's. There is
1237 also a _pc() variation used to make the corresponding change on
1238 a process copy (pc) lkb. */
1239
1240static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1241{
1242 del_lkb(r, lkb);
1243 lkb->lkb_grmode = DLM_LOCK_IV;
1244 /* this unhold undoes the original ref from create_lkb()
1245 so this leads to the lkb being freed */
1246 unhold_lkb(lkb);
1247}
1248
1249static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1250{
1251 set_lvb_unlock(r, lkb);
1252 _remove_lock(r, lkb);
1253}
1254
1255static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1256{
1257 _remove_lock(r, lkb);
1258}
1259
ef0c2bb0
DT		 1260/* returns: 0 did nothing
1261 1 moved lock to granted
1262 -1 removed lock */
1263
1264static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
e7fd4179 1265{
ef0c2bb0
DT		 1266 int rv = 0;
1267
e7fd4179
DT		 1268 lkb->lkb_rqmode = DLM_LOCK_IV;
1269
1270 switch (lkb->lkb_status) {
597d0cae
DT		 1271 case DLM_LKSTS_GRANTED:
1272 break;
e7fd4179
DT		 1273 case DLM_LKSTS_CONVERT:
1274 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
ef0c2bb0 1275 rv = 1;
e7fd4179
DT		 1276 break;
1277 case DLM_LKSTS_WAITING:
1278 del_lkb(r, lkb);
1279 lkb->lkb_grmode = DLM_LOCK_IV;
1280 /* this unhold undoes the original ref from create_lkb()
1281 so this leads to the lkb being freed */
1282 unhold_lkb(lkb);
ef0c2bb0 1283 rv = -1;
e7fd4179
DT		 1284 break;
1285 default:
1286 log_print("invalid status for revert %d", lkb->lkb_status);
1287 }
ef0c2bb0 1288 return rv;
e7fd4179
DT		 1289}
1290
ef0c2bb0 1291static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
e7fd4179 1292{
ef0c2bb0 1293 return revert_lock(r, lkb);
e7fd4179
DT		 1294}
1295
1296static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1297{
1298 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1299 lkb->lkb_grmode = lkb->lkb_rqmode;
1300 if (lkb->lkb_status)
1301 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1302 else
1303 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1304 }
1305
1306 lkb->lkb_rqmode = DLM_LOCK_IV;
e7fd4179
DT		 1307}
1308
1309static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1310{
1311 set_lvb_lock(r, lkb);
1312 _grant_lock(r, lkb);
1313 lkb->lkb_highbast = 0;
1314}
1315
1316static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1317 struct dlm_message *ms)
1318{
1319 set_lvb_lock_pc(r, lkb, ms);
1320 _grant_lock(r, lkb);
1321}
1322
1323/* called by grant_pending_locks() which means an async grant message must
1324 be sent to the requesting node in addition to granting the lock if the
1325 lkb belongs to a remote node. */
1326
1327static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1328{
1329 grant_lock(r, lkb);
1330 if (is_master_copy(lkb))
1331 send_grant(r, lkb);
1332 else
1333 queue_cast(r, lkb, 0);
1334}
1335
7d3c1feb
DT		 1336/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1337 change the granted/requested modes. We're munging things accordingly in
1338 the process copy.
1339 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1340 conversion deadlock
1341 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1342 compatible with other granted locks */
1343
1344static void munge_demoted(struct dlm_lkb *lkb, struct dlm_message *ms)
1345{
1346 if (ms->m_type != DLM_MSG_CONVERT_REPLY) {
1347 log_print("munge_demoted %x invalid reply type %d",
1348 lkb->lkb_id, ms->m_type);
1349 return;
1350 }
1351
1352 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
1353 log_print("munge_demoted %x invalid modes gr %d rq %d",
1354 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
1355 return;
1356 }
1357
1358 lkb->lkb_grmode = DLM_LOCK_NL;
1359}
1360
1361static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
1362{
1363 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
1364 ms->m_type != DLM_MSG_GRANT) {
1365 log_print("munge_altmode %x invalid reply type %d",
1366 lkb->lkb_id, ms->m_type);
1367 return;
1368 }
1369
1370 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
1371 lkb->lkb_rqmode = DLM_LOCK_PR;
1372 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
1373 lkb->lkb_rqmode = DLM_LOCK_CW;
1374 else {
1375 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
1376 dlm_print_lkb(lkb);
1377 }
1378}
1379
e7fd4179
DT		 1380static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1381{
1382 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1383 lkb_statequeue);
1384 if (lkb->lkb_id == first->lkb_id)
90135925 1385 return 1;
e7fd4179 1386
90135925 1387 return 0;
e7fd4179
DT		 1388}
1389
e7fd4179
DT		 1390/* Check if the given lkb conflicts with another lkb on the queue. */
1391
1392static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1393{
1394 struct dlm_lkb *this;
1395
1396 list_for_each_entry(this, head, lkb_statequeue) {
1397 if (this == lkb)
1398 continue;
3bcd3687 1399 if (!modes_compat(this, lkb))
90135925 1400 return 1;
e7fd4179 1401 }
90135925 1402 return 0;
e7fd4179
DT		 1403}
1404
1405/*
1406 * "A conversion deadlock arises with a pair of lock requests in the converting
1407 * queue for one resource. The granted mode of each lock blocks the requested
1408 * mode of the other lock."
1409 *
c85d65e9
DT		 1410 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1411 * convert queue from being granted, then deadlk/demote lkb.
e7fd4179
DT		 1412 *
1413 * Example:
1414 * Granted Queue: empty
1415 * Convert Queue: NL->EX (first lock)
1416 * PR->EX (second lock)
1417 *
1418 * The first lock can't be granted because of the granted mode of the second
1419 * lock and the second lock can't be granted because it's not first in the
c85d65e9
DT		 1420 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1421 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1422 * flag set and return DEMOTED in the lksb flags.
e7fd4179 1423 *
c85d65e9
DT		 1424 * Originally, this function detected conv-deadlk in a more limited scope:
1425 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1426 * - if lkb1 was the first entry in the queue (not just earlier), and was
1427 * blocked by the granted mode of lkb2, and there was nothing on the
1428 * granted queue preventing lkb1 from being granted immediately, i.e.
1429 * lkb2 was the only thing preventing lkb1 from being granted.
1430 *
1431 * That second condition meant we'd only say there was conv-deadlk if
1432 * resolving it (by demotion) would lead to the first lock on the convert
1433 * queue being granted right away. It allowed conversion deadlocks to exist
1434 * between locks on the convert queue while they couldn't be granted anyway.
1435 *
1436 * Now, we detect and take action on conversion deadlocks immediately when
1437 * they're created, even if they may not be immediately consequential. If
1438 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1439 * mode that would prevent lkb1's conversion from being granted, we do a
1440 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1441 * I think this means that the lkb_is_ahead condition below should always
1442 * be zero, i.e. there will never be conv-deadlk between two locks that are
1443 * both already on the convert queue.
e7fd4179
DT		 1444 */
1445
c85d65e9 1446static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
e7fd4179 1447{
c85d65e9
DT		 1448 struct dlm_lkb *lkb1;
1449 int lkb_is_ahead = 0;
e7fd4179 1450
c85d65e9
DT		 1451 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
1452 if (lkb1 == lkb2) {
1453 lkb_is_ahead = 1;
e7fd4179
DT		 1454 continue;
1455 }
1456
c85d65e9
DT		 1457 if (!lkb_is_ahead) {
1458 if (!modes_compat(lkb2, lkb1))
1459 return 1;
1460 } else {
1461 if (!modes_compat(lkb2, lkb1) &&
1462 !modes_compat(lkb1, lkb2))
1463 return 1;
1464 }
e7fd4179 1465 }
90135925 1466 return 0;
e7fd4179
DT		 1467}
1468
1469/*
1470 * Return 1 if the lock can be granted, 0 otherwise.
1471 * Also detect and resolve conversion deadlocks.
1472 *
1473 * lkb is the lock to be granted
1474 *
1475 * now is 1 if the function is being called in the context of the
1476 * immediate request, it is 0 if called later, after the lock has been
1477 * queued.
1478 *
1479 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1480 */
1481
1482static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1483{
1484 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1485
1486 /*
1487 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1488 * a new request for a NL mode lock being blocked.
1489 *
1490 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1491 * request, then it would be granted. In essence, the use of this flag
1492 * tells the Lock Manager to expedite theis request by not considering
1493 * what may be in the CONVERTING or WAITING queues... As of this
1494 * writing, the EXPEDITE flag can be used only with new requests for NL
1495 * mode locks. This flag is not valid for conversion requests.
1496 *
1497 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1498 * conversion or used with a non-NL requested mode. We also know an
1499 * EXPEDITE request is always granted immediately, so now must always
1500 * be 1. The full condition to grant an expedite request: (now &&
1501 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1502 * therefore be shortened to just checking the flag.
1503 */
1504
1505 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
90135925 1506 return 1;
e7fd4179
DT		 1507
1508 /*
1509 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1510 * added to the remaining conditions.
1511 */
1512
1513 if (queue_conflict(&r->res_grantqueue, lkb))
1514 goto out;
1515
1516 /*
1517 * 6-3: By default, a conversion request is immediately granted if the
1518 * requested mode is compatible with the modes of all other granted
1519 * locks
1520 */
1521
1522 if (queue_conflict(&r->res_convertqueue, lkb))
1523 goto out;
1524
1525 /*
1526 * 6-5: But the default algorithm for deciding whether to grant or
1527 * queue conversion requests does not by itself guarantee that such
1528 * requests are serviced on a "first come first serve" basis. This, in
1529 * turn, can lead to a phenomenon known as "indefinate postponement".
1530 *
1531 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1532 * the system service employed to request a lock conversion. This flag
1533 * forces certain conversion requests to be queued, even if they are
1534 * compatible with the granted modes of other locks on the same
1535 * resource. Thus, the use of this flag results in conversion requests
1536 * being ordered on a "first come first servce" basis.
1537 *
1538 * DCT: This condition is all about new conversions being able to occur
1539 * "in place" while the lock remains on the granted queue (assuming
1540 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1541 * doesn't _have_ to go onto the convert queue where it's processed in
1542 * order. The "now" variable is necessary to distinguish converts
1543 * being received and processed for the first time now, because once a
1544 * convert is moved to the conversion queue the condition below applies
1545 * requiring fifo granting.
1546 */
1547
1548 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
90135925 1549 return 1;
e7fd4179
DT		 1550
1551 /*
3bcd3687
DT		 1552 * The NOORDER flag is set to avoid the standard vms rules on grant
1553 * order.
e7fd4179
DT		 1554 */
1555
1556 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
90135925 1557 return 1;
e7fd4179
DT		 1558
1559 /*
1560 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1561 * granted until all other conversion requests ahead of it are granted
1562 * and/or canceled.
1563 */
1564
1565 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
90135925 1566 return 1;
e7fd4179
DT		 1567
1568 /*
1569 * 6-4: By default, a new request is immediately granted only if all
1570 * three of the following conditions are satisfied when the request is
1571 * issued:
1572 * - The queue of ungranted conversion requests for the resource is
1573 * empty.
1574 * - The queue of ungranted new requests for the resource is empty.
1575 * - The mode of the new request is compatible with the most
1576 * restrictive mode of all granted locks on the resource.
1577 */
1578
1579 if (now && !conv && list_empty(&r->res_convertqueue) &&
1580 list_empty(&r->res_waitqueue))
90135925 1581 return 1;
e7fd4179
DT		 1582
1583 /*
1584 * 6-4: Once a lock request is in the queue of ungranted new requests,
1585 * it cannot be granted until the queue of ungranted conversion
1586 * requests is empty, all ungranted new requests ahead of it are
1587 * granted and/or canceled, and it is compatible with the granted mode
1588 * of the most restrictive lock granted on the resource.
1589 */
1590
1591 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1592 first_in_list(lkb, &r->res_waitqueue))
90135925 1593 return 1;
e7fd4179 1594 out:
90135925 1595 return 0;
e7fd4179
DT		 1596}
1597
c85d65e9
DT		 1598static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
1599 int *err)
e7fd4179 1600{
e7fd4179
DT		 1601 int rv;
1602 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
c85d65e9
DT		 1603 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
1604
1605 if (err)
1606 *err = 0;
e7fd4179
DT		 1607
1608 rv = _can_be_granted(r, lkb, now);
1609 if (rv)
1610 goto out;
1611
c85d65e9
DT		 1612 /*
1613 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1614 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1615 * cancels one of the locks.
1616 */
1617
1618 if (is_convert && can_be_queued(lkb) &&
1619 conversion_deadlock_detect(r, lkb)) {
1620 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
1621 lkb->lkb_grmode = DLM_LOCK_NL;
1622 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1623 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1624 if (err)
1625 *err = -EDEADLK;
1626 else {
1627 log_print("can_be_granted deadlock %x now %d",
1628 lkb->lkb_id, now);
1629 dlm_dump_rsb(r);
1630 }
1631 }
e7fd4179 1632 goto out;
c85d65e9 1633 }
e7fd4179 1634
c85d65e9
DT		 1635 /*
1636 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1637 * to grant a request in a mode other than the normal rqmode. It's a
1638 * simple way to provide a big optimization to applications that can
1639 * use them.
1640 */
1641
1642 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
e7fd4179 1643 alt = DLM_LOCK_PR;
c85d65e9 1644 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
e7fd4179
DT		 1645 alt = DLM_LOCK_CW;
1646
1647 if (alt) {
1648 lkb->lkb_rqmode = alt;
1649 rv = _can_be_granted(r, lkb, now);
1650 if (rv)
1651 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1652 else
1653 lkb->lkb_rqmode = rqmode;
1654 }
1655 out:
1656 return rv;
1657}
1658
c85d65e9
DT		 1659/* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1660 for locks pending on the convert list. Once verified (watch for these
1661 log_prints), we should be able to just call _can_be_granted() and not
1662 bother with the demote/deadlk cases here (and there's no easy way to deal
1663 with a deadlk here, we'd have to generate something like grant_lock with
1664 the deadlk error.) */
1665
36509258
DT		 1666/* Returns the highest requested mode of all blocked conversions; sets
1667 cw if there's a blocked conversion to DLM_LOCK_CW. */
c85d65e9 1668
36509258 1669static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
e7fd4179
DT		 1670{
1671 struct dlm_lkb *lkb, *s;
1672 int hi, demoted, quit, grant_restart, demote_restart;
c85d65e9 1673 int deadlk;
e7fd4179
DT		 1674
1675 quit = 0;
1676 restart:
1677 grant_restart = 0;
1678 demote_restart = 0;
1679 hi = DLM_LOCK_IV;
1680
1681 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1682 demoted = is_demoted(lkb);
c85d65e9
DT		 1683 deadlk = 0;
1684
1685 if (can_be_granted(r, lkb, 0, &deadlk)) {
e7fd4179
DT		 1686 grant_lock_pending(r, lkb);
1687 grant_restart = 1;
c85d65e9 1688 continue;
e7fd4179 1689 }
c85d65e9
DT		 1690
1691 if (!demoted && is_demoted(lkb)) {
1692 log_print("WARN: pending demoted %x node %d %s",
1693 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1694 demote_restart = 1;
1695 continue;
1696 }
1697
1698 if (deadlk) {
1699 log_print("WARN: pending deadlock %x node %d %s",
1700 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1701 dlm_dump_rsb(r);
1702 continue;
1703 }
1704
1705 hi = max_t(int, lkb->lkb_rqmode, hi);
36509258
DT		 1706
1707 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
1708 *cw = 1;
e7fd4179
DT		 1709 }
1710
1711 if (grant_restart)
1712 goto restart;
1713 if (demote_restart && !quit) {
1714 quit = 1;
1715 goto restart;
1716 }
1717
1718 return max_t(int, high, hi);
1719}
1720
36509258 1721static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
e7fd4179
DT		 1722{
1723 struct dlm_lkb *lkb, *s;
1724
1725 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
c85d65e9 1726 if (can_be_granted(r, lkb, 0, NULL))
e7fd4179 1727 grant_lock_pending(r, lkb);
36509258 1728 else {
e7fd4179 1729 high = max_t(int, lkb->lkb_rqmode, high);
36509258
DT		 1730 if (lkb->lkb_rqmode == DLM_LOCK_CW)
1731 *cw = 1;
1732 }
e7fd4179
DT		 1733 }
1734
1735 return high;
1736}
1737
36509258
DT		 1738/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
1739 on either the convert or waiting queue.
1740 high is the largest rqmode of all locks blocked on the convert or
1741 waiting queue. */
1742
1743static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
1744{
1745 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
1746 if (gr->lkb_highbast < DLM_LOCK_EX)
1747 return 1;
1748 return 0;
1749 }
1750
1751 if (gr->lkb_highbast < high &&
1752 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
1753 return 1;
1754 return 0;
1755}
1756
e7fd4179
DT		 1757static void grant_pending_locks(struct dlm_rsb *r)
1758{
1759 struct dlm_lkb *lkb, *s;
1760 int high = DLM_LOCK_IV;
36509258 1761 int cw = 0;
e7fd4179 1762
a345da3e 1763 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
e7fd4179 1764
36509258
DT		 1765 high = grant_pending_convert(r, high, &cw);
1766 high = grant_pending_wait(r, high, &cw);
e7fd4179
DT		 1767
1768 if (high == DLM_LOCK_IV)
1769 return;
1770
1771 /*
1772 * If there are locks left on the wait/convert queue then send blocking
1773 * ASTs to granted locks based on the largest requested mode (high)
36509258 1774 * found above.
e7fd4179
DT		 1775 */
1776
1777 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
36509258
DT		 1778 if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
1779 if (cw && high == DLM_LOCK_PR)
1780 queue_bast(r, lkb, DLM_LOCK_CW);
1781 else
1782 queue_bast(r, lkb, high);
e7fd4179
DT		 1783 lkb->lkb_highbast = high;
1784 }
1785 }
1786}
1787
36509258
DT		 1788static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
1789{
1790 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
1791 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
1792 if (gr->lkb_highbast < DLM_LOCK_EX)
1793 return 1;
1794 return 0;
1795 }
1796
1797 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
1798 return 1;
1799 return 0;
1800}
1801
e7fd4179
DT		 1802static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1803 struct dlm_lkb *lkb)
1804{
1805 struct dlm_lkb *gr;
1806
1807 list_for_each_entry(gr, head, lkb_statequeue) {
36509258 1808 if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
e7fd4179
DT		 1809 queue_bast(r, gr, lkb->lkb_rqmode);
1810 gr->lkb_highbast = lkb->lkb_rqmode;
1811 }
1812 }
1813}
1814
1815static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1816{
1817 send_bast_queue(r, &r->res_grantqueue, lkb);
1818}
1819
1820static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1821{
1822 send_bast_queue(r, &r->res_grantqueue, lkb);
1823 send_bast_queue(r, &r->res_convertqueue, lkb);
1824}
1825
1826/* set_master(r, lkb) -- set the master nodeid of a resource
1827
1828 The purpose of this function is to set the nodeid field in the given
1829 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1830 known, it can just be copied to the lkb and the function will return
1831 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1832 before it can be copied to the lkb.
1833
1834 When the rsb nodeid is being looked up remotely, the initial lkb
1835 causing the lookup is kept on the ls_waiters list waiting for the
1836 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1837 on the rsb's res_lookup list until the master is verified.
1838
1839 Return values:
1840 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1841 1: the rsb master is not available and the lkb has been placed on
1842 a wait queue
1843*/
1844
1845static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1846{
1847 struct dlm_ls *ls = r->res_ls;
755b5eb8 1848 int i, error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
e7fd4179
DT		 1849
1850 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1851 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1852 r->res_first_lkid = lkb->lkb_id;
1853 lkb->lkb_nodeid = r->res_nodeid;
1854 return 0;
1855 }
1856
1857 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1858 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1859 return 1;
1860 }
1861
1862 if (r->res_nodeid == 0) {
1863 lkb->lkb_nodeid = 0;
1864 return 0;
1865 }
1866
1867 if (r->res_nodeid > 0) {
1868 lkb->lkb_nodeid = r->res_nodeid;
1869 return 0;
1870 }
1871
a345da3e 1872 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
e7fd4179
DT		 1873
1874 dir_nodeid = dlm_dir_nodeid(r);
1875
1876 if (dir_nodeid != our_nodeid) {
1877 r->res_first_lkid = lkb->lkb_id;
1878 send_lookup(r, lkb);
1879 return 1;
1880 }
1881
755b5eb8 1882 for (i = 0; i < 2; i++) {
e7fd4179
DT		 1883 /* It's possible for dlm_scand to remove an old rsb for
1884 this same resource from the toss list, us to create
1885 a new one, look up the master locally, and find it
1886 already exists just before dlm_scand does the
1887 dir_remove() on the previous rsb. */
1888
1889 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1890 r->res_length, &ret_nodeid);
1891 if (!error)
1892 break;
1893 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1894 schedule();
1895 }
755b5eb8
DT		 1896 if (error && error != -EEXIST)
1897 return error;
e7fd4179
DT		 1898
1899 if (ret_nodeid == our_nodeid) {
1900 r->res_first_lkid = 0;
1901 r->res_nodeid = 0;
1902 lkb->lkb_nodeid = 0;
1903 } else {
1904 r->res_first_lkid = lkb->lkb_id;
1905 r->res_nodeid = ret_nodeid;
1906 lkb->lkb_nodeid = ret_nodeid;
1907 }
1908 return 0;
1909}
1910
1911static void process_lookup_list(struct dlm_rsb *r)
1912{
1913 struct dlm_lkb *lkb, *safe;
1914
1915 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
ef0c2bb0 1916 list_del_init(&lkb->lkb_rsb_lookup);
e7fd4179
DT		 1917 _request_lock(r, lkb);
1918 schedule();
1919 }
1920}
1921
1922/* confirm_master -- confirm (or deny) an rsb's master nodeid */
1923
1924static void confirm_master(struct dlm_rsb *r, int error)
1925{
1926 struct dlm_lkb *lkb;
1927
1928 if (!r->res_first_lkid)
1929 return;
1930
1931 switch (error) {
1932 case 0:
1933 case -EINPROGRESS:
1934 r->res_first_lkid = 0;
1935 process_lookup_list(r);
1936 break;
1937
1938 case -EAGAIN:
aec64e1b
DT		 1939 case -EBADR:
1940 case -ENOTBLK:
1941 /* the remote request failed and won't be retried (it was
1942 a NOQUEUE, or has been canceled/unlocked); make a waiting
1943 lkb the first_lkid */
e7fd4179
DT		 1944
1945 r->res_first_lkid = 0;
1946
1947 if (!list_empty(&r->res_lookup)) {
1948 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1949 lkb_rsb_lookup);
ef0c2bb0 1950 list_del_init(&lkb->lkb_rsb_lookup);
e7fd4179
DT		 1951 r->res_first_lkid = lkb->lkb_id;
1952 _request_lock(r, lkb);
1953 } else
1954 r->res_nodeid = -1;
1955 break;
1956
1957 default:
1958 log_error(r->res_ls, "confirm_master unknown error %d", error);
1959 }
1960}
1961
1962static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
d7db923e 1963 int namelen, unsigned long timeout_cs, void *ast,
3bcd3687 1964 void *astarg, void *bast, struct dlm_args *args)
e7fd4179
DT		 1965{
1966 int rv = -EINVAL;
1967
1968 /* check for invalid arg usage */
1969
1970 if (mode < 0 || mode > DLM_LOCK_EX)
1971 goto out;
1972
1973 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1974 goto out;
1975
1976 if (flags & DLM_LKF_CANCEL)
1977 goto out;
1978
1979 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1980 goto out;
1981
1982 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1983 goto out;
1984
1985 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1986 goto out;
1987
1988 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1989 goto out;
1990
1991 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
1992 goto out;
1993
1994 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
1995 goto out;
1996
1997 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
1998 goto out;
1999
2000 if (!ast || !lksb)
2001 goto out;
2002
2003 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2004 goto out;
2005
e7fd4179
DT		 2006 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2007 goto out;
2008
2009 /* these args will be copied to the lkb in validate_lock_args,
2010 it cannot be done now because when converting locks, fields in
2011 an active lkb cannot be modified before locking the rsb */
2012
2013 args->flags = flags;
2014 args->astaddr = ast;
2015 args->astparam = (long) astarg;
2016 args->bastaddr = bast;
d7db923e 2017 args->timeout = timeout_cs;
e7fd4179
DT		 2018 args->mode = mode;
2019 args->lksb = lksb;
e7fd4179
DT		 2020 rv = 0;
2021 out:
2022 return rv;
2023}
2024
2025static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2026{
2027 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2028 DLM_LKF_FORCEUNLOCK))
2029 return -EINVAL;
2030
ef0c2bb0
DT		 2031 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2032 return -EINVAL;
2033
e7fd4179
DT		 2034 args->flags = flags;
2035 args->astparam = (long) astarg;
2036 return 0;
2037}
2038
2039static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2040 struct dlm_args *args)
2041{
2042 int rv = -EINVAL;
2043
2044 if (args->flags & DLM_LKF_CONVERT) {
2045 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2046 goto out;
2047
2048 if (args->flags & DLM_LKF_QUECVT &&
2049 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2050 goto out;
2051
2052 rv = -EBUSY;
2053 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2054 goto out;
2055
2056 if (lkb->lkb_wait_type)
2057 goto out;
ef0c2bb0
DT		 2058
2059 if (is_overlap(lkb))
2060 goto out;
e7fd4179
DT		 2061 }
2062
2063 lkb->lkb_exflags = args->flags;
2064 lkb->lkb_sbflags = 0;
2065 lkb->lkb_astaddr = args->astaddr;
2066 lkb->lkb_astparam = args->astparam;
2067 lkb->lkb_bastaddr = args->bastaddr;
2068 lkb->lkb_rqmode = args->mode;
2069 lkb->lkb_lksb = args->lksb;
2070 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2071 lkb->lkb_ownpid = (int) current->pid;
d7db923e 2072 lkb->lkb_timeout_cs = args->timeout;
e7fd4179
DT		 2073 rv = 0;
2074 out:
2075 return rv;
2076}
2077
ef0c2bb0
DT		 2078/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2079 for success */
2080
2081/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2082 because there may be a lookup in progress and it's valid to do
2083 cancel/unlockf on it */
2084
e7fd4179
DT		 2085static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2086{
ef0c2bb0 2087 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
e7fd4179
DT		 2088 int rv = -EINVAL;
2089
ef0c2bb0
DT		 2090 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2091 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2092 dlm_print_lkb(lkb);
e7fd4179 2093 goto out;
ef0c2bb0 2094 }
e7fd4179 2095
ef0c2bb0
DT		 2096 /* an lkb may still exist even though the lock is EOL'ed due to a
2097 cancel, unlock or failed noqueue request; an app can't use these
2098 locks; return same error as if the lkid had not been found at all */
e7fd4179 2099
ef0c2bb0
DT		 2100 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2101 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2102 rv = -ENOENT;
e7fd4179 2103 goto out;
ef0c2bb0 2104 }
e7fd4179 2105
ef0c2bb0
DT		 2106 /* an lkb may be waiting for an rsb lookup to complete where the
2107 lookup was initiated by another lock */
2108
42dc1601
DT		 2109 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2110 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
ef0c2bb0
DT		 2111 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2112 list_del_init(&lkb->lkb_rsb_lookup);
2113 queue_cast(lkb->lkb_resource, lkb,
2114 args->flags & DLM_LKF_CANCEL ?
2115 -DLM_ECANCEL : -DLM_EUNLOCK);
2116 unhold_lkb(lkb); /* undoes create_lkb() */
ef0c2bb0 2117 }
42dc1601
DT		 2118 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2119 rv = -EBUSY;
2120 goto out;
ef0c2bb0
DT		 2121 }
2122
2123 /* cancel not allowed with another cancel/unlock in progress */
2124
2125 if (args->flags & DLM_LKF_CANCEL) {
2126 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2127 goto out;
2128
2129 if (is_overlap(lkb))
2130 goto out;
2131
3ae1acf9
DT		 2132 /* don't let scand try to do a cancel */
2133 del_timeout(lkb);
2134
ef0c2bb0
DT		 2135 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2136 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2137 rv = -EBUSY;
2138 goto out;
2139 }
2140
2141 switch (lkb->lkb_wait_type) {
2142 case DLM_MSG_LOOKUP:
2143 case DLM_MSG_REQUEST:
2144 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2145 rv = -EBUSY;
2146 goto out;
2147 case DLM_MSG_UNLOCK:
2148 case DLM_MSG_CANCEL:
2149 goto out;
2150 }
2151 /* add_to_waiters() will set OVERLAP_CANCEL */
2152 goto out_ok;
2153 }
2154
2155 /* do we need to allow a force-unlock if there's a normal unlock
2156 already in progress? in what conditions could the normal unlock
2157 fail such that we'd want to send a force-unlock to be sure? */
2158
2159 if (args->flags & DLM_LKF_FORCEUNLOCK) {
2160 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2161 goto out;
2162
2163 if (is_overlap_unlock(lkb))
2164 goto out;
e7fd4179 2165
3ae1acf9
DT		 2166 /* don't let scand try to do a cancel */
2167 del_timeout(lkb);
2168
ef0c2bb0
DT		 2169 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2170 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2171 rv = -EBUSY;
2172 goto out;
2173 }
2174
2175 switch (lkb->lkb_wait_type) {
2176 case DLM_MSG_LOOKUP:
2177 case DLM_MSG_REQUEST:
2178 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2179 rv = -EBUSY;
2180 goto out;
2181 case DLM_MSG_UNLOCK:
2182 goto out;
2183 }
2184 /* add_to_waiters() will set OVERLAP_UNLOCK */
2185 goto out_ok;
2186 }
2187
2188 /* normal unlock not allowed if there's any op in progress */
e7fd4179 2189 rv = -EBUSY;
ef0c2bb0 2190 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
e7fd4179
DT		 2191 goto out;
2192
2193 out_ok:
ef0c2bb0
DT		 2194 /* an overlapping op shouldn't blow away exflags from other op */
2195 lkb->lkb_exflags |= args->flags;
e7fd4179
DT		 2196 lkb->lkb_sbflags = 0;
2197 lkb->lkb_astparam = args->astparam;
e7fd4179
DT		 2198 rv = 0;
2199 out:
ef0c2bb0
DT		 2200 if (rv)
2201 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
2202 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
2203 args->flags, lkb->lkb_wait_type,
2204 lkb->lkb_resource->res_name);
e7fd4179
DT		 2205 return rv;
2206}
2207
2208/*
2209 * Four stage 4 varieties:
2210 * do_request(), do_convert(), do_unlock(), do_cancel()
2211 * These are called on the master node for the given lock and
2212 * from the central locking logic.
2213 */
2214
2215static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2216{
2217 int error = 0;
2218
c85d65e9 2219 if (can_be_granted(r, lkb, 1, NULL)) {
e7fd4179
DT		 2220 grant_lock(r, lkb);
2221 queue_cast(r, lkb, 0);
2222 goto out;
2223 }
2224
2225 if (can_be_queued(lkb)) {
2226 error = -EINPROGRESS;
2227 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2228 send_blocking_asts(r, lkb);
3ae1acf9 2229 add_timeout(lkb);
e7fd4179
DT		 2230 goto out;
2231 }
2232
2233 error = -EAGAIN;
2234 if (force_blocking_asts(lkb))
2235 send_blocking_asts_all(r, lkb);
2236 queue_cast(r, lkb, -EAGAIN);
2237
2238 out:
2239 return error;
2240}
2241
2242static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2243{
2244 int error = 0;
c85d65e9 2245 int deadlk = 0;
e7fd4179
DT		 2246
2247 /* changing an existing lock may allow others to be granted */
2248
c85d65e9 2249 if (can_be_granted(r, lkb, 1, &deadlk)) {
e7fd4179
DT		 2250 grant_lock(r, lkb);
2251 queue_cast(r, lkb, 0);
2252 grant_pending_locks(r);
2253 goto out;
2254 }
2255
c85d65e9
DT		 2256 /* can_be_granted() detected that this lock would block in a conversion
2257 deadlock, so we leave it on the granted queue and return EDEADLK in
2258 the ast for the convert. */
2259
2260 if (deadlk) {
2261 /* it's left on the granted queue */
2262 log_debug(r->res_ls, "deadlock %x node %d sts%d g%d r%d %s",
2263 lkb->lkb_id, lkb->lkb_nodeid, lkb->lkb_status,
2264 lkb->lkb_grmode, lkb->lkb_rqmode, r->res_name);
2265 revert_lock(r, lkb);
2266 queue_cast(r, lkb, -EDEADLK);
2267 error = -EDEADLK;
2268 goto out;
2269 }
2270
7d3c1feb
DT		 2271 /* is_demoted() means the can_be_granted() above set the grmode
2272 to NL, and left us on the granted queue. This auto-demotion
2273 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2274 now grantable. We have to try to grant other converting locks
2275 before we try again to grant this one. */
2276
2277 if (is_demoted(lkb)) {
36509258 2278 grant_pending_convert(r, DLM_LOCK_IV, NULL);
7d3c1feb
DT		 2279 if (_can_be_granted(r, lkb, 1)) {
2280 grant_lock(r, lkb);
2281 queue_cast(r, lkb, 0);
e7fd4179 2282 grant_pending_locks(r);
7d3c1feb
DT		 2283 goto out;
2284 }
2285 /* else fall through and move to convert queue */
2286 }
2287
2288 if (can_be_queued(lkb)) {
e7fd4179
DT		 2289 error = -EINPROGRESS;
2290 del_lkb(r, lkb);
2291 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2292 send_blocking_asts(r, lkb);
3ae1acf9 2293 add_timeout(lkb);
e7fd4179
DT		 2294 goto out;
2295 }
2296
2297 error = -EAGAIN;
2298 if (force_blocking_asts(lkb))
2299 send_blocking_asts_all(r, lkb);
2300 queue_cast(r, lkb, -EAGAIN);
2301
2302 out:
2303 return error;
2304}
2305
2306static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2307{
2308 remove_lock(r, lkb);
2309 queue_cast(r, lkb, -DLM_EUNLOCK);
2310 grant_pending_locks(r);
2311 return -DLM_EUNLOCK;
2312}
2313
ef0c2bb0 2314/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
907b9bce 2315
e7fd4179
DT		 2316static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2317{
ef0c2bb0
DT		 2318 int error;
2319
2320 error = revert_lock(r, lkb);
2321 if (error) {
2322 queue_cast(r, lkb, -DLM_ECANCEL);
2323 grant_pending_locks(r);
2324 return -DLM_ECANCEL;
2325 }
2326 return 0;
e7fd4179
DT		 2327}
2328
2329/*
2330 * Four stage 3 varieties:
2331 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2332 */
2333
2334/* add a new lkb to a possibly new rsb, called by requesting process */
2335
2336static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2337{
2338 int error;
2339
2340 /* set_master: sets lkb nodeid from r */
2341
2342 error = set_master(r, lkb);
2343 if (error < 0)
2344 goto out;
2345 if (error) {
2346 error = 0;
2347 goto out;
2348 }
2349
2350 if (is_remote(r))
2351 /* receive_request() calls do_request() on remote node */
2352 error = send_request(r, lkb);
2353 else
2354 error = do_request(r, lkb);
2355 out:
2356 return error;
2357}
2358
3bcd3687 2359/* change some property of an existing lkb, e.g. mode */
e7fd4179
DT		 2360
2361static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2362{
2363 int error;
2364
2365 if (is_remote(r))
2366 /* receive_convert() calls do_convert() on remote node */
2367 error = send_convert(r, lkb);
2368 else
2369 error = do_convert(r, lkb);
2370
2371 return error;
2372}
2373
2374/* remove an existing lkb from the granted queue */
2375
2376static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2377{
2378 int error;
2379
2380 if (is_remote(r))
2381 /* receive_unlock() calls do_unlock() on remote node */
2382 error = send_unlock(r, lkb);
2383 else
2384 error = do_unlock(r, lkb);
2385
2386 return error;
2387}
2388
2389/* remove an existing lkb from the convert or wait queue */
2390
2391static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2392{
2393 int error;
2394
2395 if (is_remote(r))
2396 /* receive_cancel() calls do_cancel() on remote node */
2397 error = send_cancel(r, lkb);
2398 else
2399 error = do_cancel(r, lkb);
2400
2401 return error;
2402}
2403
2404/*
2405 * Four stage 2 varieties:
2406 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2407 */
2408
2409static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
2410 int len, struct dlm_args *args)
2411{
2412 struct dlm_rsb *r;
2413 int error;
2414
2415 error = validate_lock_args(ls, lkb, args);
2416 if (error)
2417 goto out;
2418
2419 error = find_rsb(ls, name, len, R_CREATE, &r);
2420 if (error)
2421 goto out;
2422
2423 lock_rsb(r);
2424
2425 attach_lkb(r, lkb);
2426 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
2427
2428 error = _request_lock(r, lkb);
2429
2430 unlock_rsb(r);
2431 put_rsb(r);
2432
2433 out:
2434 return error;
2435}
2436
2437static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2438 struct dlm_args *args)
2439{
2440 struct dlm_rsb *r;
2441 int error;
2442
2443 r = lkb->lkb_resource;
2444
2445 hold_rsb(r);
2446 lock_rsb(r);
2447
2448 error = validate_lock_args(ls, lkb, args);
2449 if (error)
2450 goto out;
2451
2452 error = _convert_lock(r, lkb);
2453 out:
2454 unlock_rsb(r);
2455 put_rsb(r);
2456 return error;
2457}
2458
2459static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2460 struct dlm_args *args)
2461{
2462 struct dlm_rsb *r;
2463 int error;
2464
2465 r = lkb->lkb_resource;
2466
2467 hold_rsb(r);
2468 lock_rsb(r);
2469
2470 error = validate_unlock_args(lkb, args);
2471 if (error)
2472 goto out;
2473
2474 error = _unlock_lock(r, lkb);
2475 out:
2476 unlock_rsb(r);
2477 put_rsb(r);
2478 return error;
2479}
2480
2481static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2482 struct dlm_args *args)
2483{
2484 struct dlm_rsb *r;
2485 int error;
2486
2487 r = lkb->lkb_resource;
2488
2489 hold_rsb(r);
2490 lock_rsb(r);
2491
2492 error = validate_unlock_args(lkb, args);
2493 if (error)
2494 goto out;
2495
2496 error = _cancel_lock(r, lkb);
2497 out:
2498 unlock_rsb(r);
2499 put_rsb(r);
2500 return error;
2501}
2502
2503/*
2504 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2505 */
2506
2507int dlm_lock(dlm_lockspace_t *lockspace,
2508 int mode,
2509 struct dlm_lksb *lksb,
2510 uint32_t flags,
2511 void *name,
2512 unsigned int namelen,
2513 uint32_t parent_lkid,
2514 void (*ast) (void *astarg),
2515 void *astarg,
3bcd3687 2516 void (*bast) (void *astarg, int mode))
e7fd4179
DT		 2517{
2518 struct dlm_ls *ls;
2519 struct dlm_lkb *lkb;
2520 struct dlm_args args;
2521 int error, convert = flags & DLM_LKF_CONVERT;
2522
2523 ls = dlm_find_lockspace_local(lockspace);
2524 if (!ls)
2525 return -EINVAL;
2526
85e86edf 2527 dlm_lock_recovery(ls);
e7fd4179
DT		 2528
2529 if (convert)
2530 error = find_lkb(ls, lksb->sb_lkid, &lkb);
2531 else
2532 error = create_lkb(ls, &lkb);
2533
2534 if (error)
2535 goto out;
2536
d7db923e 2537 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3bcd3687 2538 astarg, bast, &args);
e7fd4179
DT		 2539 if (error)
2540 goto out_put;
2541
2542 if (convert)
2543 error = convert_lock(ls, lkb, &args);
2544 else
2545 error = request_lock(ls, lkb, name, namelen, &args);
2546
2547 if (error == -EINPROGRESS)
2548 error = 0;
2549 out_put:
2550 if (convert || error)
b3f58d8f 2551 __put_lkb(ls, lkb);
c85d65e9 2552 if (error == -EAGAIN || error == -EDEADLK)
e7fd4179
DT		 2553 error = 0;
2554 out:
85e86edf 2555 dlm_unlock_recovery(ls);
e7fd4179
DT		 2556 dlm_put_lockspace(ls);
2557 return error;
2558}
2559
2560int dlm_unlock(dlm_lockspace_t *lockspace,
2561 uint32_t lkid,
2562 uint32_t flags,
2563 struct dlm_lksb *lksb,
2564 void *astarg)
2565{
2566 struct dlm_ls *ls;
2567 struct dlm_lkb *lkb;
2568 struct dlm_args args;
2569 int error;
2570
2571 ls = dlm_find_lockspace_local(lockspace);
2572 if (!ls)
2573 return -EINVAL;
2574
85e86edf 2575 dlm_lock_recovery(ls);
e7fd4179
DT		 2576
2577 error = find_lkb(ls, lkid, &lkb);
2578 if (error)
2579 goto out;
2580
2581 error = set_unlock_args(flags, astarg, &args);
2582 if (error)
2583 goto out_put;
2584
2585 if (flags & DLM_LKF_CANCEL)
2586 error = cancel_lock(ls, lkb, &args);
2587 else
2588 error = unlock_lock(ls, lkb, &args);
2589
2590 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2591 error = 0;
ef0c2bb0
DT		 2592 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
2593 error = 0;
e7fd4179 2594 out_put:
b3f58d8f 2595 dlm_put_lkb(lkb);
e7fd4179 2596 out:
85e86edf 2597 dlm_unlock_recovery(ls);
e7fd4179
DT		 2598 dlm_put_lockspace(ls);
2599 return error;
2600}
2601
2602/*
2603 * send/receive routines for remote operations and replies
2604 *
2605 * send_args
2606 * send_common
2607 * send_request receive_request
2608 * send_convert receive_convert
2609 * send_unlock receive_unlock
2610 * send_cancel receive_cancel
2611 * send_grant receive_grant
2612 * send_bast receive_bast
2613 * send_lookup receive_lookup
2614 * send_remove receive_remove
2615 *
2616 * send_common_reply
2617 * receive_request_reply send_request_reply
2618 * receive_convert_reply send_convert_reply
2619 * receive_unlock_reply send_unlock_reply
2620 * receive_cancel_reply send_cancel_reply
2621 * receive_lookup_reply send_lookup_reply
2622 */
2623
7e4dac33
DT		 2624static int _create_message(struct dlm_ls *ls, int mb_len,
2625 int to_nodeid, int mstype,
2626 struct dlm_message **ms_ret,
2627 struct dlm_mhandle **mh_ret)
e7fd4179
DT		 2628{
2629 struct dlm_message *ms;
2630 struct dlm_mhandle *mh;
2631 char *mb;
e7fd4179
DT		 2632
2633 /* get_buffer gives us a message handle (mh) that we need to
2634 pass into lowcomms_commit and a message buffer (mb) that we
2635 write our data into */
2636
44f487a5 2637 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
e7fd4179
DT		 2638 if (!mh)
2639 return -ENOBUFS;
2640
2641 memset(mb, 0, mb_len);
2642
2643 ms = (struct dlm_message *) mb;
2644
2645 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
7e4dac33 2646 ms->m_header.h_lockspace = ls->ls_global_id;
e7fd4179
DT		 2647 ms->m_header.h_nodeid = dlm_our_nodeid();
2648 ms->m_header.h_length = mb_len;
2649 ms->m_header.h_cmd = DLM_MSG;
2650
2651 ms->m_type = mstype;
2652
2653 *mh_ret = mh;
2654 *ms_ret = ms;
2655 return 0;
2656}
2657
7e4dac33
DT		 2658static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2659 int to_nodeid, int mstype,
2660 struct dlm_message **ms_ret,
2661 struct dlm_mhandle **mh_ret)
2662{
2663 int mb_len = sizeof(struct dlm_message);
2664
2665 switch (mstype) {
2666 case DLM_MSG_REQUEST:
2667 case DLM_MSG_LOOKUP:
2668 case DLM_MSG_REMOVE:
2669 mb_len += r->res_length;
2670 break;
2671 case DLM_MSG_CONVERT:
2672 case DLM_MSG_UNLOCK:
2673 case DLM_MSG_REQUEST_REPLY:
2674 case DLM_MSG_CONVERT_REPLY:
2675 case DLM_MSG_GRANT:
2676 if (lkb && lkb->lkb_lvbptr)
2677 mb_len += r->res_ls->ls_lvblen;
2678 break;
2679 }
2680
2681 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
2682 ms_ret, mh_ret);
2683}
2684
e7fd4179
DT		 2685/* further lowcomms enhancements or alternate implementations may make
2686 the return value from this function useful at some point */
2687
2688static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2689{
2690 dlm_message_out(ms);
2691 dlm_lowcomms_commit_buffer(mh);
2692 return 0;
2693}
2694
2695static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2696 struct dlm_message *ms)
2697{
2698 ms->m_nodeid = lkb->lkb_nodeid;
2699 ms->m_pid = lkb->lkb_ownpid;
2700 ms->m_lkid = lkb->lkb_id;
2701 ms->m_remid = lkb->lkb_remid;
2702 ms->m_exflags = lkb->lkb_exflags;
2703 ms->m_sbflags = lkb->lkb_sbflags;
2704 ms->m_flags = lkb->lkb_flags;
2705 ms->m_lvbseq = lkb->lkb_lvbseq;
2706 ms->m_status = lkb->lkb_status;
2707 ms->m_grmode = lkb->lkb_grmode;
2708 ms->m_rqmode = lkb->lkb_rqmode;
2709 ms->m_hash = r->res_hash;
2710
2711 /* m_result and m_bastmode are set from function args,
2712 not from lkb fields */
2713
2714 if (lkb->lkb_bastaddr)
2715 ms->m_asts |= AST_BAST;
2716 if (lkb->lkb_astaddr)
2717 ms->m_asts |= AST_COMP;
2718
da49f36f
DT		 2719 /* compare with switch in create_message; send_remove() doesn't
2720 use send_args() */
e7fd4179 2721
da49f36f
DT		 2722 switch (ms->m_type) {
2723 case DLM_MSG_REQUEST:
2724 case DLM_MSG_LOOKUP:
2725 memcpy(ms->m_extra, r->res_name, r->res_length);
2726 break;
2727 case DLM_MSG_CONVERT:
2728 case DLM_MSG_UNLOCK:
2729 case DLM_MSG_REQUEST_REPLY:
2730 case DLM_MSG_CONVERT_REPLY:
2731 case DLM_MSG_GRANT:
2732 if (!lkb->lkb_lvbptr)
2733 break;
e7fd4179 2734 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
da49f36f
DT		 2735 break;
2736 }
e7fd4179
DT		 2737}
2738
2739static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2740{
2741 struct dlm_message *ms;
2742 struct dlm_mhandle *mh;
2743 int to_nodeid, error;
2744
ef0c2bb0
DT		 2745 error = add_to_waiters(lkb, mstype);
2746 if (error)
2747 return error;
e7fd4179
DT		 2748
2749 to_nodeid = r->res_nodeid;
2750
2751 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2752 if (error)
2753 goto fail;
2754
2755 send_args(r, lkb, ms);
2756
2757 error = send_message(mh, ms);
2758 if (error)
2759 goto fail;
2760 return 0;
2761
2762 fail:
ef0c2bb0 2763 remove_from_waiters(lkb, msg_reply_type(mstype));
e7fd4179
DT		 2764 return error;
2765}
2766
2767static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2768{
2769 return send_common(r, lkb, DLM_MSG_REQUEST);
2770}
2771
2772static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2773{
2774 int error;
2775
2776 error = send_common(r, lkb, DLM_MSG_CONVERT);
2777
2778 /* down conversions go without a reply from the master */
2779 if (!error && down_conversion(lkb)) {
ef0c2bb0
DT		 2780 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
2781 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
e7fd4179 2782 r->res_ls->ls_stub_ms.m_result = 0;
32f105a1 2783 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
e7fd4179
DT		 2784 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2785 }
2786
2787 return error;
2788}
2789
2790/* FIXME: if this lkb is the only lock we hold on the rsb, then set
2791 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2792 that the master is still correct. */
2793
2794static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2795{
2796 return send_common(r, lkb, DLM_MSG_UNLOCK);
2797}
2798
2799static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2800{
2801 return send_common(r, lkb, DLM_MSG_CANCEL);
2802}
2803
2804static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2805{
2806 struct dlm_message *ms;
2807 struct dlm_mhandle *mh;
2808 int to_nodeid, error;
2809
2810 to_nodeid = lkb->lkb_nodeid;
2811
2812 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2813 if (error)
2814 goto out;
2815
2816 send_args(r, lkb, ms);
2817
2818 ms->m_result = 0;
2819
2820 error = send_message(mh, ms);
2821 out:
2822 return error;
2823}
2824
2825static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2826{
2827 struct dlm_message *ms;
2828 struct dlm_mhandle *mh;
2829 int to_nodeid, error;
2830
2831 to_nodeid = lkb->lkb_nodeid;
2832
2833 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2834 if (error)
2835 goto out;
2836
2837 send_args(r, lkb, ms);
2838
2839 ms->m_bastmode = mode;
2840
2841 error = send_message(mh, ms);
2842 out:
2843 return error;
2844}
2845
2846static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2847{
2848 struct dlm_message *ms;
2849 struct dlm_mhandle *mh;
2850 int to_nodeid, error;
2851
ef0c2bb0
DT		 2852 error = add_to_waiters(lkb, DLM_MSG_LOOKUP);
2853 if (error)
2854 return error;
e7fd4179
DT		 2855
2856 to_nodeid = dlm_dir_nodeid(r);
2857
2858 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2859 if (error)
2860 goto fail;
2861
2862 send_args(r, lkb, ms);
2863
2864 error = send_message(mh, ms);
2865 if (error)
2866 goto fail;
2867 return 0;
2868
2869 fail:
ef0c2bb0 2870 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
e7fd4179
DT		 2871 return error;
2872}
2873
2874static int send_remove(struct dlm_rsb *r)
2875{
2876 struct dlm_message *ms;
2877 struct dlm_mhandle *mh;
2878 int to_nodeid, error;
2879
2880 to_nodeid = dlm_dir_nodeid(r);
2881
2882 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2883 if (error)
2884 goto out;
2885
2886 memcpy(ms->m_extra, r->res_name, r->res_length);
2887 ms->m_hash = r->res_hash;
2888
2889 error = send_message(mh, ms);
2890 out:
2891 return error;
2892}
2893
2894static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2895 int mstype, int rv)
2896{
2897 struct dlm_message *ms;
2898 struct dlm_mhandle *mh;
2899 int to_nodeid, error;
2900
2901 to_nodeid = lkb->lkb_nodeid;
2902
2903 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2904 if (error)
2905 goto out;
2906
2907 send_args(r, lkb, ms);
2908
2909 ms->m_result = rv;
2910
2911 error = send_message(mh, ms);
2912 out:
2913 return error;
2914}
2915
2916static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2917{
2918 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2919}
2920
2921static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2922{
2923 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2924}
2925
2926static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2927{
2928 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2929}
2930
2931static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2932{
2933 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2934}
2935
2936static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2937 int ret_nodeid, int rv)
2938{
2939 struct dlm_rsb *r = &ls->ls_stub_rsb;
2940 struct dlm_message *ms;
2941 struct dlm_mhandle *mh;
2942 int error, nodeid = ms_in->m_header.h_nodeid;
2943
2944 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2945 if (error)
2946 goto out;
2947
2948 ms->m_lkid = ms_in->m_lkid;
2949 ms->m_result = rv;
2950 ms->m_nodeid = ret_nodeid;
2951
2952 error = send_message(mh, ms);
2953 out:
2954 return error;
2955}
2956
2957/* which args we save from a received message depends heavily on the type
2958 of message, unlike the send side where we can safely send everything about
2959 the lkb for any type of message */
2960
2961static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2962{
2963 lkb->lkb_exflags = ms->m_exflags;
6f90a8b1 2964 lkb->lkb_sbflags = ms->m_sbflags;
e7fd4179
DT		 2965 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2966 (ms->m_flags & 0x0000FFFF);
2967}
2968
2969static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2970{
2971 lkb->lkb_sbflags = ms->m_sbflags;
2972 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2973 (ms->m_flags & 0x0000FFFF);
2974}
2975
2976static int receive_extralen(struct dlm_message *ms)
2977{
2978 return (ms->m_header.h_length - sizeof(struct dlm_message));
2979}
2980
e7fd4179
DT		 2981static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2982 struct dlm_message *ms)
2983{
2984 int len;
2985
2986 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2987 if (!lkb->lkb_lvbptr)
52bda2b5 2988 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
e7fd4179
DT		 2989 if (!lkb->lkb_lvbptr)
2990 return -ENOMEM;
2991 len = receive_extralen(ms);
2992 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2993 }
2994 return 0;
2995}
2996
2997static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2998 struct dlm_message *ms)
2999{
3000 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3001 lkb->lkb_ownpid = ms->m_pid;
3002 lkb->lkb_remid = ms->m_lkid;
3003 lkb->lkb_grmode = DLM_LOCK_IV;
3004 lkb->lkb_rqmode = ms->m_rqmode;
3005 lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
3006 lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
3007
8d07fd50
DT		 3008 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3009 /* lkb was just created so there won't be an lvb yet */
52bda2b5 3010 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
8d07fd50
DT		 3011 if (!lkb->lkb_lvbptr)
3012 return -ENOMEM;
3013 }
e7fd4179
DT		 3014
3015 return 0;
3016}
3017
3018static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3019 struct dlm_message *ms)
3020{
e7fd4179
DT		 3021 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3022 return -EBUSY;
3023
e7fd4179
DT		 3024 if (receive_lvb(ls, lkb, ms))
3025 return -ENOMEM;
3026
3027 lkb->lkb_rqmode = ms->m_rqmode;
3028 lkb->lkb_lvbseq = ms->m_lvbseq;
3029
3030 return 0;
3031}
3032
3033static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3034 struct dlm_message *ms)
3035{
e7fd4179
DT		 3036 if (receive_lvb(ls, lkb, ms))
3037 return -ENOMEM;
3038 return 0;
3039}
3040
3041/* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3042 uses to send a reply and that the remote end uses to process the reply. */
3043
3044static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3045{
3046 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3047 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3048 lkb->lkb_remid = ms->m_lkid;
3049}
3050
c54e04b0
DT		 3051/* This is called after the rsb is locked so that we can safely inspect
3052 fields in the lkb. */
3053
3054static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3055{
3056 int from = ms->m_header.h_nodeid;
3057 int error = 0;
3058
3059 switch (ms->m_type) {
3060 case DLM_MSG_CONVERT:
3061 case DLM_MSG_UNLOCK:
3062 case DLM_MSG_CANCEL:
3063 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3064 error = -EINVAL;
3065 break;
3066
3067 case DLM_MSG_CONVERT_REPLY:
3068 case DLM_MSG_UNLOCK_REPLY:
3069 case DLM_MSG_CANCEL_REPLY:
3070 case DLM_MSG_GRANT:
3071 case DLM_MSG_BAST:
3072 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3073 error = -EINVAL;
3074 break;
3075
3076 case DLM_MSG_REQUEST_REPLY:
3077 if (!is_process_copy(lkb))
3078 error = -EINVAL;
3079 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3080 error = -EINVAL;
3081 break;
3082
3083 default:
3084 error = -EINVAL;
3085 }
3086
3087 if (error)
3088 log_error(lkb->lkb_resource->res_ls,
3089 "ignore invalid message %d from %d %x %x %x %d",
3090 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
3091 lkb->lkb_flags, lkb->lkb_nodeid);
3092 return error;
3093}
3094
e7fd4179
DT		 3095static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3096{
3097 struct dlm_lkb *lkb;
3098 struct dlm_rsb *r;
3099 int error, namelen;
3100
3101 error = create_lkb(ls, &lkb);
3102 if (error)
3103 goto fail;
3104
3105 receive_flags(lkb, ms);
3106 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3107 error = receive_request_args(ls, lkb, ms);
3108 if (error) {
b3f58d8f 3109 __put_lkb(ls, lkb);
e7fd4179
DT		 3110 goto fail;
3111 }
3112
3113 namelen = receive_extralen(ms);
3114
3115 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
3116 if (error) {
b3f58d8f 3117 __put_lkb(ls, lkb);
e7fd4179
DT		 3118 goto fail;
3119 }
3120
3121 lock_rsb(r);
3122
3123 attach_lkb(r, lkb);
3124 error = do_request(r, lkb);
3125 send_request_reply(r, lkb, error);
3126
3127 unlock_rsb(r);
3128 put_rsb(r);
3129
3130 if (error == -EINPROGRESS)
3131 error = 0;
3132 if (error)
b3f58d8f 3133 dlm_put_lkb(lkb);
e7fd4179
DT		 3134 return;
3135
3136 fail:
3137 setup_stub_lkb(ls, ms);
3138 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3139}
3140
3141static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3142{
3143 struct dlm_lkb *lkb;
3144 struct dlm_rsb *r;
90135925 3145 int error, reply = 1;
e7fd4179
DT		 3146
3147 error = find_lkb(ls, ms->m_remid, &lkb);
3148 if (error)
3149 goto fail;
3150
3151 r = lkb->lkb_resource;
3152
3153 hold_rsb(r);
3154 lock_rsb(r);
3155
c54e04b0
DT		 3156 error = validate_message(lkb, ms);
3157 if (error)
3158 goto out;
3159
e7fd4179
DT		 3160 receive_flags(lkb, ms);
3161 error = receive_convert_args(ls, lkb, ms);
3162 if (error)
c54e04b0 3163 goto out_reply;
e7fd4179
DT		 3164 reply = !down_conversion(lkb);
3165
3166 error = do_convert(r, lkb);
c54e04b0 3167 out_reply:
e7fd4179
DT		 3168 if (reply)
3169 send_convert_reply(r, lkb, error);
c54e04b0 3170 out:
e7fd4179
DT		 3171 unlock_rsb(r);
3172 put_rsb(r);
b3f58d8f 3173 dlm_put_lkb(lkb);
e7fd4179
DT		 3174 return;
3175
3176 fail:
3177 setup_stub_lkb(ls, ms);
3178 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3179}
3180
3181static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3182{
3183 struct dlm_lkb *lkb;
3184 struct dlm_rsb *r;
3185 int error;
3186
3187 error = find_lkb(ls, ms->m_remid, &lkb);
3188 if (error)
3189 goto fail;
3190
3191 r = lkb->lkb_resource;
3192
3193 hold_rsb(r);
3194 lock_rsb(r);
3195
c54e04b0
DT		 3196 error = validate_message(lkb, ms);
3197 if (error)
3198 goto out;
3199
e7fd4179
DT		 3200 receive_flags(lkb, ms);
3201 error = receive_unlock_args(ls, lkb, ms);
3202 if (error)
c54e04b0 3203 goto out_reply;
e7fd4179
DT		 3204
3205 error = do_unlock(r, lkb);
c54e04b0 3206 out_reply:
e7fd4179 3207 send_unlock_reply(r, lkb, error);
c54e04b0 3208 out:
e7fd4179
DT		 3209 unlock_rsb(r);
3210 put_rsb(r);
b3f58d8f 3211 dlm_put_lkb(lkb);
e7fd4179
DT		 3212 return;
3213
3214 fail:
3215 setup_stub_lkb(ls, ms);
3216 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3217}
3218
3219static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3220{
3221 struct dlm_lkb *lkb;
3222 struct dlm_rsb *r;
3223 int error;
3224
3225 error = find_lkb(ls, ms->m_remid, &lkb);
3226 if (error)
3227 goto fail;
3228
3229 receive_flags(lkb, ms);
3230
3231 r = lkb->lkb_resource;
3232
3233 hold_rsb(r);
3234 lock_rsb(r);
3235
c54e04b0
DT		 3236 error = validate_message(lkb, ms);
3237 if (error)
3238 goto out;
3239
e7fd4179
DT		 3240 error = do_cancel(r, lkb);
3241 send_cancel_reply(r, lkb, error);
c54e04b0 3242 out:
e7fd4179
DT		 3243 unlock_rsb(r);
3244 put_rsb(r);
b3f58d8f 3245 dlm_put_lkb(lkb);
e7fd4179
DT		 3246 return;
3247
3248 fail:
3249 setup_stub_lkb(ls, ms);
3250 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3251}
3252
3253static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
3254{
3255 struct dlm_lkb *lkb;
3256 struct dlm_rsb *r;
3257 int error;
3258
3259 error = find_lkb(ls, ms->m_remid, &lkb);
3260 if (error) {
c54e04b0
DT		 3261 log_debug(ls, "receive_grant from %d no lkb %x",
3262 ms->m_header.h_nodeid, ms->m_remid);
e7fd4179
DT		 3263 return;
3264 }
e7fd4179
DT		 3265
3266 r = lkb->lkb_resource;
3267
3268 hold_rsb(r);
3269 lock_rsb(r);
3270
c54e04b0
DT		 3271 error = validate_message(lkb, ms);
3272 if (error)
3273 goto out;
3274
e7fd4179 3275 receive_flags_reply(lkb, ms);
7d3c1feb
DT		 3276 if (is_altmode(lkb))
3277 munge_altmode(lkb, ms);
e7fd4179
DT		 3278 grant_lock_pc(r, lkb, ms);
3279 queue_cast(r, lkb, 0);
c54e04b0 3280 out:
e7fd4179
DT		 3281 unlock_rsb(r);
3282 put_rsb(r);
b3f58d8f 3283 dlm_put_lkb(lkb);
e7fd4179
DT		 3284}
3285
3286static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
3287{
3288 struct dlm_lkb *lkb;
3289 struct dlm_rsb *r;
3290 int error;
3291
3292 error = find_lkb(ls, ms->m_remid, &lkb);
3293 if (error) {
c54e04b0
DT		 3294 log_debug(ls, "receive_bast from %d no lkb %x",
3295 ms->m_header.h_nodeid, ms->m_remid);
e7fd4179
DT		 3296 return;
3297 }
e7fd4179
DT		 3298
3299 r = lkb->lkb_resource;
3300
3301 hold_rsb(r);
3302 lock_rsb(r);
3303
c54e04b0
DT		 3304 error = validate_message(lkb, ms);
3305 if (error)
3306 goto out;
e7fd4179 3307
c54e04b0
DT		 3308 queue_bast(r, lkb, ms->m_bastmode);
3309 out:
e7fd4179
DT		 3310 unlock_rsb(r);
3311 put_rsb(r);
b3f58d8f 3312 dlm_put_lkb(lkb);
e7fd4179
DT		 3313}
3314
3315static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
3316{
3317 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
3318
3319 from_nodeid = ms->m_header.h_nodeid;
3320 our_nodeid = dlm_our_nodeid();
3321
3322 len = receive_extralen(ms);
3323
3324 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3325 if (dir_nodeid != our_nodeid) {
3326 log_error(ls, "lookup dir_nodeid %d from %d",
3327 dir_nodeid, from_nodeid);
3328 error = -EINVAL;
3329 ret_nodeid = -1;
3330 goto out;
3331 }
3332
3333 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
3334
3335 /* Optimization: we're master so treat lookup as a request */
3336 if (!error && ret_nodeid == our_nodeid) {
3337 receive_request(ls, ms);
3338 return;
3339 }
3340 out:
3341 send_lookup_reply(ls, ms, ret_nodeid, error);
3342}
3343
3344static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
3345{
3346 int len, dir_nodeid, from_nodeid;
3347
3348 from_nodeid = ms->m_header.h_nodeid;
3349
3350 len = receive_extralen(ms);
3351
3352 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3353 if (dir_nodeid != dlm_our_nodeid()) {
3354 log_error(ls, "remove dir entry dir_nodeid %d from %d",
3355 dir_nodeid, from_nodeid);
3356 return;
3357 }
3358
3359 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
3360}
3361
8499137d
DT		 3362static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
3363{
3364 do_purge(ls, ms->m_nodeid, ms->m_pid);
3365}
3366
e7fd4179
DT		 3367static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
3368{
3369 struct dlm_lkb *lkb;
3370 struct dlm_rsb *r;
ef0c2bb0 3371 int error, mstype, result;
e7fd4179
DT		 3372
3373 error = find_lkb(ls, ms->m_remid, &lkb);
3374 if (error) {
c54e04b0
DT		 3375 log_debug(ls, "receive_request_reply from %d no lkb %x",
3376 ms->m_header.h_nodeid, ms->m_remid);
e7fd4179
DT		 3377 return;
3378 }
e7fd4179 3379
e7fd4179
DT		 3380 r = lkb->lkb_resource;
3381 hold_rsb(r);
3382 lock_rsb(r);
3383
c54e04b0
DT		 3384 error = validate_message(lkb, ms);
3385 if (error)
3386 goto out;
3387
ef0c2bb0
DT		 3388 mstype = lkb->lkb_wait_type;
3389 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
3390 if (error)
3391 goto out;
3392
e7fd4179
DT		 3393 /* Optimization: the dir node was also the master, so it took our
3394 lookup as a request and sent request reply instead of lookup reply */
3395 if (mstype == DLM_MSG_LOOKUP) {
3396 r->res_nodeid = ms->m_header.h_nodeid;
3397 lkb->lkb_nodeid = r->res_nodeid;
3398 }
3399
ef0c2bb0
DT		 3400 /* this is the value returned from do_request() on the master */
3401 result = ms->m_result;
3402
3403 switch (result) {
e7fd4179 3404 case -EAGAIN:
ef0c2bb0 3405 /* request would block (be queued) on remote master */
e7fd4179
DT		 3406 queue_cast(r, lkb, -EAGAIN);
3407 confirm_master(r, -EAGAIN);
ef0c2bb0 3408 unhold_lkb(lkb); /* undoes create_lkb() */
e7fd4179
DT		 3409 break;
3410
3411 case -EINPROGRESS:
3412 case 0:
3413 /* request was queued or granted on remote master */
3414 receive_flags_reply(lkb, ms);
3415 lkb->lkb_remid = ms->m_lkid;
7d3c1feb
DT		 3416 if (is_altmode(lkb))
3417 munge_altmode(lkb, ms);
3ae1acf9 3418 if (result) {
e7fd4179 3419 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3ae1acf9
DT		 3420 add_timeout(lkb);
3421 } else {
e7fd4179
DT		 3422 grant_lock_pc(r, lkb, ms);
3423 queue_cast(r, lkb, 0);
3424 }
ef0c2bb0 3425 confirm_master(r, result);
e7fd4179
DT		 3426 break;
3427
597d0cae 3428 case -EBADR:
e7fd4179
DT		 3429 case -ENOTBLK:
3430 /* find_rsb failed to find rsb or rsb wasn't master */
ef0c2bb0
DT		 3431 log_debug(ls, "receive_request_reply %x %x master diff %d %d",
3432 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result);
e7fd4179
DT		 3433 r->res_nodeid = -1;
3434 lkb->lkb_nodeid = -1;
ef0c2bb0
DT		 3435
3436 if (is_overlap(lkb)) {
3437 /* we'll ignore error in cancel/unlock reply */
3438 queue_cast_overlap(r, lkb);
aec64e1b 3439 confirm_master(r, result);
ef0c2bb0
DT		 3440 unhold_lkb(lkb); /* undoes create_lkb() */
3441 } else
3442 _request_lock(r, lkb);
e7fd4179
DT		 3443 break;
3444
3445 default:
ef0c2bb0
DT		 3446 log_error(ls, "receive_request_reply %x error %d",
3447 lkb->lkb_id, result);
e7fd4179
DT		 3448 }
3449
ef0c2bb0
DT		 3450 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
3451 log_debug(ls, "receive_request_reply %x result %d unlock",
3452 lkb->lkb_id, result);
3453 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3454 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3455 send_unlock(r, lkb);
3456 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
3457 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
3458 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3459 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3460 send_cancel(r, lkb);
3461 } else {
3462 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3463 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3464 }
3465 out:
e7fd4179
DT		 3466 unlock_rsb(r);
3467 put_rsb(r);
b3f58d8f 3468 dlm_put_lkb(lkb);
e7fd4179
DT		 3469}
3470
3471static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3472 struct dlm_message *ms)
3473{
e7fd4179 3474 /* this is the value returned from do_convert() on the master */
ef0c2bb0 3475 switch (ms->m_result) {
e7fd4179
DT		 3476 case -EAGAIN:
3477 /* convert would block (be queued) on remote master */
3478 queue_cast(r, lkb, -EAGAIN);
3479 break;
3480
c85d65e9
DT		 3481 case -EDEADLK:
3482 receive_flags_reply(lkb, ms);
3483 revert_lock_pc(r, lkb);
3484 queue_cast(r, lkb, -EDEADLK);
3485 break;
3486
e7fd4179
DT		 3487 case -EINPROGRESS:
3488 /* convert was queued on remote master */
7d3c1feb
DT		 3489 receive_flags_reply(lkb, ms);
3490 if (is_demoted(lkb))
3491 munge_demoted(lkb, ms);
e7fd4179
DT		 3492 del_lkb(r, lkb);
3493 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3ae1acf9 3494 add_timeout(lkb);
e7fd4179
DT		 3495 break;
3496
3497 case 0:
3498 /* convert was granted on remote master */
3499 receive_flags_reply(lkb, ms);
7d3c1feb
DT		 3500 if (is_demoted(lkb))
3501 munge_demoted(lkb, ms);
e7fd4179
DT		 3502 grant_lock_pc(r, lkb, ms);
3503 queue_cast(r, lkb, 0);
3504 break;
3505
3506 default:
ef0c2bb0
DT		 3507 log_error(r->res_ls, "receive_convert_reply %x error %d",
3508 lkb->lkb_id, ms->m_result);
e7fd4179
DT		 3509 }
3510}
3511
3512static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3513{
3514 struct dlm_rsb *r = lkb->lkb_resource;
ef0c2bb0 3515 int error;
e7fd4179
DT		 3516
3517 hold_rsb(r);
3518 lock_rsb(r);
3519
c54e04b0
DT		 3520 error = validate_message(lkb, ms);
3521 if (error)
3522 goto out;
3523
ef0c2bb0
DT		 3524 /* stub reply can happen with waiters_mutex held */
3525 error = remove_from_waiters_ms(lkb, ms);
3526 if (error)
3527 goto out;
e7fd4179 3528
ef0c2bb0
DT		 3529 __receive_convert_reply(r, lkb, ms);
3530 out:
e7fd4179
DT		 3531 unlock_rsb(r);
3532 put_rsb(r);
3533}
3534
3535static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
3536{
3537 struct dlm_lkb *lkb;
3538 int error;
3539
3540 error = find_lkb(ls, ms->m_remid, &lkb);
3541 if (error) {
c54e04b0
DT		 3542 log_debug(ls, "receive_convert_reply from %d no lkb %x",
3543 ms->m_header.h_nodeid, ms->m_remid);
e7fd4179
DT		 3544 return;
3545 }
e7fd4179 3546
e7fd4179 3547 _receive_convert_reply(lkb, ms);
b3f58d8f 3548 dlm_put_lkb(lkb);
e7fd4179
DT		 3549}
3550
3551static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3552{
3553 struct dlm_rsb *r = lkb->lkb_resource;
ef0c2bb0 3554 int error;
e7fd4179
DT		 3555
3556 hold_rsb(r);
3557 lock_rsb(r);
3558
c54e04b0
DT		 3559 error = validate_message(lkb, ms);
3560 if (error)
3561 goto out;
3562
ef0c2bb0
DT		 3563 /* stub reply can happen with waiters_mutex held */
3564 error = remove_from_waiters_ms(lkb, ms);
3565 if (error)
3566 goto out;
3567
e7fd4179
DT		 3568 /* this is the value returned from do_unlock() on the master */
3569
ef0c2bb0 3570 switch (ms->m_result) {
e7fd4179
DT		 3571 case -DLM_EUNLOCK:
3572 receive_flags_reply(lkb, ms);
3573 remove_lock_pc(r, lkb);
3574 queue_cast(r, lkb, -DLM_EUNLOCK);
3575 break;
ef0c2bb0
DT		 3576 case -ENOENT:
3577 break;
e7fd4179 3578 default:
ef0c2bb0
DT		 3579 log_error(r->res_ls, "receive_unlock_reply %x error %d",
3580 lkb->lkb_id, ms->m_result);
e7fd4179 3581 }
ef0c2bb0 3582 out:
e7fd4179
DT		 3583 unlock_rsb(r);
3584 put_rsb(r);
3585}
3586
3587static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
3588{
3589 struct dlm_lkb *lkb;
3590 int error;
3591
3592 error = find_lkb(ls, ms->m_remid, &lkb);
3593 if (error) {
c54e04b0
DT		 3594 log_debug(ls, "receive_unlock_reply from %d no lkb %x",
3595 ms->m_header.h_nodeid, ms->m_remid);
e7fd4179
DT		 3596 return;
3597 }
e7fd4179 3598
e7fd4179 3599 _receive_unlock_reply(lkb, ms);
b3f58d8f 3600 dlm_put_lkb(lkb);
e7fd4179
DT		 3601}
3602
3603static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3604{
3605 struct dlm_rsb *r = lkb->lkb_resource;
ef0c2bb0 3606 int error;
e7fd4179
DT		 3607
3608 hold_rsb(r);
3609 lock_rsb(r);
3610
c54e04b0
DT		 3611 error = validate_message(lkb, ms);
3612 if (error)
3613 goto out;
3614
ef0c2bb0
DT		 3615 /* stub reply can happen with waiters_mutex held */
3616 error = remove_from_waiters_ms(lkb, ms);
3617 if (error)
3618 goto out;
3619
e7fd4179
DT		 3620 /* this is the value returned from do_cancel() on the master */
3621
ef0c2bb0 3622 switch (ms->m_result) {
e7fd4179
DT		 3623 case -DLM_ECANCEL:
3624 receive_flags_reply(lkb, ms);
3625 revert_lock_pc(r, lkb);
84d8cd69 3626 queue_cast(r, lkb, -DLM_ECANCEL);
ef0c2bb0
DT		 3627 break;
3628 case 0:
e7fd4179
DT		 3629 break;
3630 default:
ef0c2bb0
DT		 3631 log_error(r->res_ls, "receive_cancel_reply %x error %d",
3632 lkb->lkb_id, ms->m_result);
e7fd4179 3633 }
ef0c2bb0 3634 out:
e7fd4179
DT		 3635 unlock_rsb(r);
3636 put_rsb(r);
3637}
3638
3639static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
3640{
3641 struct dlm_lkb *lkb;
3642 int error;
3643
3644 error = find_lkb(ls, ms->m_remid, &lkb);
3645 if (error) {
c54e04b0
DT		 3646 log_debug(ls, "receive_cancel_reply from %d no lkb %x",
3647 ms->m_header.h_nodeid, ms->m_remid);
e7fd4179
DT		 3648 return;
3649 }
e7fd4179 3650
e7fd4179 3651 _receive_cancel_reply(lkb, ms);
b3f58d8f 3652 dlm_put_lkb(lkb);
e7fd4179
DT		 3653}
3654
3655static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
3656{
3657 struct dlm_lkb *lkb;
3658 struct dlm_rsb *r;
3659 int error, ret_nodeid;
3660
3661 error = find_lkb(ls, ms->m_lkid, &lkb);
3662 if (error) {
3663 log_error(ls, "receive_lookup_reply no lkb");
3664 return;
3665 }
3666
ef0c2bb0 3667 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
e7fd4179 3668 FIXME: will a non-zero error ever be returned? */
e7fd4179
DT		 3669
3670 r = lkb->lkb_resource;
3671 hold_rsb(r);
3672 lock_rsb(r);
3673
ef0c2bb0
DT		 3674 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3675 if (error)
3676 goto out;
3677
e7fd4179
DT		 3678 ret_nodeid = ms->m_nodeid;
3679 if (ret_nodeid == dlm_our_nodeid()) {
3680 r->res_nodeid = 0;
3681 ret_nodeid = 0;
3682 r->res_first_lkid = 0;
3683 } else {
3684 /* set_master() will copy res_nodeid to lkb_nodeid */
3685 r->res_nodeid = ret_nodeid;
3686 }
3687
ef0c2bb0
DT		 3688 if (is_overlap(lkb)) {
3689 log_debug(ls, "receive_lookup_reply %x unlock %x",
3690 lkb->lkb_id, lkb->lkb_flags);
3691 queue_cast_overlap(r, lkb);
3692 unhold_lkb(lkb); /* undoes create_lkb() */
3693 goto out_list;
3694 }
3695
e7fd4179
DT		 3696 _request_lock(r, lkb);
3697
ef0c2bb0 3698 out_list:
e7fd4179
DT		 3699 if (!ret_nodeid)
3700 process_lookup_list(r);
ef0c2bb0 3701 out:
e7fd4179
DT		 3702 unlock_rsb(r);
3703 put_rsb(r);
b3f58d8f 3704 dlm_put_lkb(lkb);
e7fd4179
DT		 3705}
3706
c36258b5 3707static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179 3708{
46b43eed
DT		 3709 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
3710 log_debug(ls, "ignore non-member message %d from %d %x %x %d",
3711 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
3712 ms->m_remid, ms->m_result);
3713 return;
3714 }
3715
e7fd4179
DT		 3716 switch (ms->m_type) {
3717
3718 /* messages sent to a master node */
3719
3720 case DLM_MSG_REQUEST:
3721 receive_request(ls, ms);
3722 break;
3723
3724 case DLM_MSG_CONVERT:
3725 receive_convert(ls, ms);
3726 break;
3727
3728 case DLM_MSG_UNLOCK:
3729 receive_unlock(ls, ms);
3730 break;
3731
3732 case DLM_MSG_CANCEL:
3733 receive_cancel(ls, ms);
3734 break;
3735
3736 /* messages sent from a master node (replies to above) */
3737
3738 case DLM_MSG_REQUEST_REPLY:
3739 receive_request_reply(ls, ms);
3740 break;
3741
3742 case DLM_MSG_CONVERT_REPLY:
3743 receive_convert_reply(ls, ms);
3744 break;
3745
3746 case DLM_MSG_UNLOCK_REPLY:
3747 receive_unlock_reply(ls, ms);
3748 break;
3749
3750 case DLM_MSG_CANCEL_REPLY:
3751 receive_cancel_reply(ls, ms);
3752 break;
3753
3754 /* messages sent from a master node (only two types of async msg) */
3755
3756 case DLM_MSG_GRANT:
3757 receive_grant(ls, ms);
3758 break;
3759
3760 case DLM_MSG_BAST:
3761 receive_bast(ls, ms);
3762 break;
3763
3764 /* messages sent to a dir node */
3765
3766 case DLM_MSG_LOOKUP:
3767 receive_lookup(ls, ms);
3768 break;
3769
3770 case DLM_MSG_REMOVE:
3771 receive_remove(ls, ms);
3772 break;
3773
3774 /* messages sent from a dir node (remove has no reply) */
3775
3776 case DLM_MSG_LOOKUP_REPLY:
3777 receive_lookup_reply(ls, ms);
3778 break;
3779
8499137d
DT		 3780 /* other messages */
3781
3782 case DLM_MSG_PURGE:
3783 receive_purge(ls, ms);
3784 break;
3785
e7fd4179
DT		 3786 default:
3787 log_error(ls, "unknown message type %d", ms->m_type);
3788 }
3789
e7fd4179 3790 dlm_astd_wake();
e7fd4179
DT		 3791}
3792
c36258b5
DT		 3793/* If the lockspace is in recovery mode (locking stopped), then normal
3794 messages are saved on the requestqueue for processing after recovery is
3795 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
3796 messages off the requestqueue before we process new ones. This occurs right
3797 after recovery completes when we transition from saving all messages on
3798 requestqueue, to processing all the saved messages, to processing new
3799 messages as they arrive. */
e7fd4179 3800
c36258b5
DT		 3801static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
3802 int nodeid)
3803{
3804 if (dlm_locking_stopped(ls)) {
3805 dlm_add_requestqueue(ls, nodeid, (struct dlm_header *) ms);
3806 } else {
3807 dlm_wait_requestqueue(ls);
3808 _receive_message(ls, ms);
3809 }
3810}
3811
3812/* This is called by dlm_recoverd to process messages that were saved on
3813 the requestqueue. */
3814
3815void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms)
3816{
3817 _receive_message(ls, ms);
3818}
3819
3820/* This is called by the midcomms layer when something is received for
3821 the lockspace. It could be either a MSG (normal message sent as part of
3822 standard locking activity) or an RCOM (recovery message sent as part of
3823 lockspace recovery). */
3824
3825void dlm_receive_buffer(struct dlm_header *hd, int nodeid)
3826{
3827 struct dlm_message *ms = (struct dlm_message *) hd;
3828 struct dlm_rcom *rc = (struct dlm_rcom *) hd;
3829 struct dlm_ls *ls;
3830 int type = 0;
3831
3832 switch (hd->h_cmd) {
3833 case DLM_MSG:
3834 dlm_message_in(ms);
3835 type = ms->m_type;
3836 break;
3837 case DLM_RCOM:
3838 dlm_rcom_in(rc);
3839 type = rc->rc_type;
3840 break;
3841 default:
3842 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
3843 return;
3844 }
3845
3846 if (hd->h_nodeid != nodeid) {
3847 log_print("invalid h_nodeid %d from %d lockspace %x",
3848 hd->h_nodeid, nodeid, hd->h_lockspace);
3849 return;
3850 }
3851
3852 ls = dlm_find_lockspace_global(hd->h_lockspace);
3853 if (!ls) {
594199eb
DT		 3854 if (dlm_config.ci_log_debug)
3855 log_print("invalid lockspace %x from %d cmd %d type %d",
3856 hd->h_lockspace, nodeid, hd->h_cmd, type);
c36258b5
DT		 3857
3858 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
3859 dlm_send_ls_not_ready(nodeid, rc);
3860 return;
3861 }
3862
3863 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
3864 be inactive (in this ls) before transitioning to recovery mode */
3865
3866 down_read(&ls->ls_recv_active);
3867 if (hd->h_cmd == DLM_MSG)
3868 dlm_receive_message(ls, ms, nodeid);
3869 else
3870 dlm_receive_rcom(ls, rc, nodeid);
3871 up_read(&ls->ls_recv_active);
3872
3873 dlm_put_lockspace(ls);
3874}
e7fd4179
DT		 3875
3876static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3877{
3878 if (middle_conversion(lkb)) {
3879 hold_lkb(lkb);
ef0c2bb0 3880 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
e7fd4179 3881 ls->ls_stub_ms.m_result = -EINPROGRESS;
075529b5 3882 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
c54e04b0 3883 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
e7fd4179
DT		 3884 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3885
3886 /* Same special case as in receive_rcom_lock_args() */
3887 lkb->lkb_grmode = DLM_LOCK_IV;
3888 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3889 unhold_lkb(lkb);
3890
3891 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3892 lkb->lkb_flags |= DLM_IFL_RESEND;
3893 }
3894
3895 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3896 conversions are async; there's no reply from the remote master */
3897}
3898
3899/* A waiting lkb needs recovery if the master node has failed, or
3900 the master node is changing (only when no directory is used) */
3901
3902static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3903{
3904 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3905 return 1;
3906
3907 if (!dlm_no_directory(ls))
3908 return 0;
3909
3910 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3911 return 1;
3912
3913 return 0;
3914}
3915
3916/* Recovery for locks that are waiting for replies from nodes that are now
3917 gone. We can just complete unlocks and cancels by faking a reply from the
3918 dead node. Requests and up-conversions we flag to be resent after
3919 recovery. Down-conversions can just be completed with a fake reply like
3920 unlocks. Conversions between PR and CW need special attention. */
3921
3922void dlm_recover_waiters_pre(struct dlm_ls *ls)
3923{
3924 struct dlm_lkb *lkb, *safe;
601342ce 3925 int wait_type, stub_unlock_result, stub_cancel_result;
e7fd4179 3926
90135925 3927 mutex_lock(&ls->ls_waiters_mutex);
e7fd4179
DT		 3928
3929 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3930 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3931 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3932
3933 /* all outstanding lookups, regardless of destination will be
3934 resent after recovery is done */
3935
3936 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3937 lkb->lkb_flags |= DLM_IFL_RESEND;
3938 continue;
3939 }
3940
3941 if (!waiter_needs_recovery(ls, lkb))
3942 continue;
3943
601342ce
DT		 3944 wait_type = lkb->lkb_wait_type;
3945 stub_unlock_result = -DLM_EUNLOCK;
3946 stub_cancel_result = -DLM_ECANCEL;
3947
3948 /* Main reply may have been received leaving a zero wait_type,
3949 but a reply for the overlapping op may not have been
3950 received. In that case we need to fake the appropriate
3951 reply for the overlap op. */
3952
3953 if (!wait_type) {
3954 if (is_overlap_cancel(lkb)) {
3955 wait_type = DLM_MSG_CANCEL;
3956 if (lkb->lkb_grmode == DLM_LOCK_IV)
3957 stub_cancel_result = 0;
3958 }
3959 if (is_overlap_unlock(lkb)) {
3960 wait_type = DLM_MSG_UNLOCK;
3961 if (lkb->lkb_grmode == DLM_LOCK_IV)
3962 stub_unlock_result = -ENOENT;
3963 }
3964
3965 log_debug(ls, "rwpre overlap %x %x %d %d %d",
3966 lkb->lkb_id, lkb->lkb_flags, wait_type,
3967 stub_cancel_result, stub_unlock_result);
3968 }
3969
3970 switch (wait_type) {
e7fd4179
DT		 3971
3972 case DLM_MSG_REQUEST:
3973 lkb->lkb_flags |= DLM_IFL_RESEND;
3974 break;
3975
3976 case DLM_MSG_CONVERT:
3977 recover_convert_waiter(ls, lkb);
3978 break;
3979
3980 case DLM_MSG_UNLOCK:
3981 hold_lkb(lkb);
ef0c2bb0 3982 ls->ls_stub_ms.m_type = DLM_MSG_UNLOCK_REPLY;
601342ce 3983 ls->ls_stub_ms.m_result = stub_unlock_result;
075529b5 3984 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
c54e04b0 3985 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
e7fd4179 3986 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
b3f58d8f 3987 dlm_put_lkb(lkb);
e7fd4179
DT		 3988 break;
3989
3990 case DLM_MSG_CANCEL:
3991 hold_lkb(lkb);
ef0c2bb0 3992 ls->ls_stub_ms.m_type = DLM_MSG_CANCEL_REPLY;
601342ce 3993 ls->ls_stub_ms.m_result = stub_cancel_result;
075529b5 3994 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
c54e04b0 3995 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
e7fd4179 3996 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
b3f58d8f 3997 dlm_put_lkb(lkb);
e7fd4179
DT		 3998 break;
3999
4000 default:
601342ce
DT		 4001 log_error(ls, "invalid lkb wait_type %d %d",
4002 lkb->lkb_wait_type, wait_type);
e7fd4179 4003 }
81456807 4004 schedule();
e7fd4179 4005 }
90135925 4006 mutex_unlock(&ls->ls_waiters_mutex);
e7fd4179
DT		 4007}
4008
ef0c2bb0 4009static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
e7fd4179
DT		 4010{
4011 struct dlm_lkb *lkb;
ef0c2bb0 4012 int found = 0;
e7fd4179 4013
90135925 4014 mutex_lock(&ls->ls_waiters_mutex);
e7fd4179
DT		 4015 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
4016 if (lkb->lkb_flags & DLM_IFL_RESEND) {
ef0c2bb0
DT		 4017 hold_lkb(lkb);
4018 found = 1;
e7fd4179
DT		 4019 break;
4020 }
4021 }
90135925 4022 mutex_unlock(&ls->ls_waiters_mutex);
e7fd4179 4023
ef0c2bb0 4024 if (!found)
e7fd4179 4025 lkb = NULL;
ef0c2bb0 4026 return lkb;
e7fd4179
DT		 4027}
4028
4029/* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
4030 master or dir-node for r. Processing the lkb may result in it being placed
4031 back on waiters. */
4032
ef0c2bb0
DT		 4033/* We do this after normal locking has been enabled and any saved messages
4034 (in requestqueue) have been processed. We should be confident that at
4035 this point we won't get or process a reply to any of these waiting
4036 operations. But, new ops may be coming in on the rsbs/locks here from
4037 userspace or remotely. */
4038
4039/* there may have been an overlap unlock/cancel prior to recovery or after
4040 recovery. if before, the lkb may still have a pos wait_count; if after, the
4041 overlap flag would just have been set and nothing new sent. we can be
4042 confident here than any replies to either the initial op or overlap ops
4043 prior to recovery have been received. */
4044
e7fd4179
DT		 4045int dlm_recover_waiters_post(struct dlm_ls *ls)
4046{
4047 struct dlm_lkb *lkb;
4048 struct dlm_rsb *r;
ef0c2bb0 4049 int error = 0, mstype, err, oc, ou;
e7fd4179
DT		 4050
4051 while (1) {
4052 if (dlm_locking_stopped(ls)) {
4053 log_debug(ls, "recover_waiters_post aborted");
4054 error = -EINTR;
4055 break;
4056 }
4057
ef0c2bb0
DT		 4058 lkb = find_resend_waiter(ls);
4059 if (!lkb)
e7fd4179
DT		 4060 break;
4061
4062 r = lkb->lkb_resource;
ef0c2bb0
DT		 4063 hold_rsb(r);
4064 lock_rsb(r);
4065
4066 mstype = lkb->lkb_wait_type;
4067 oc = is_overlap_cancel(lkb);
4068 ou = is_overlap_unlock(lkb);
4069 err = 0;
e7fd4179
DT		 4070
4071 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
4072 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
4073
ef0c2bb0
DT		 4074 /* At this point we assume that we won't get a reply to any
4075 previous op or overlap op on this lock. First, do a big
4076 remove_from_waiters() for all previous ops. */
4077
4078 lkb->lkb_flags &= ~DLM_IFL_RESEND;
4079 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4080 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4081 lkb->lkb_wait_type = 0;
4082 lkb->lkb_wait_count = 0;
4083 mutex_lock(&ls->ls_waiters_mutex);
4084 list_del_init(&lkb->lkb_wait_reply);
4085 mutex_unlock(&ls->ls_waiters_mutex);
4086 unhold_lkb(lkb); /* for waiters list */
4087
4088 if (oc || ou) {
4089 /* do an unlock or cancel instead of resending */
4090 switch (mstype) {
4091 case DLM_MSG_LOOKUP:
4092 case DLM_MSG_REQUEST:
4093 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
4094 -DLM_ECANCEL);
4095 unhold_lkb(lkb); /* undoes create_lkb() */
4096 break;
4097 case DLM_MSG_CONVERT:
4098 if (oc) {
4099 queue_cast(r, lkb, -DLM_ECANCEL);
4100 } else {
4101 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
4102 _unlock_lock(r, lkb);
4103 }
4104 break;
4105 default:
4106 err = 1;
4107 }
4108 } else {
4109 switch (mstype) {
4110 case DLM_MSG_LOOKUP:
4111 case DLM_MSG_REQUEST:
4112 _request_lock(r, lkb);
4113 if (is_master(r))
4114 confirm_master(r, 0);
4115 break;
4116 case DLM_MSG_CONVERT:
4117 _convert_lock(r, lkb);
4118 break;
4119 default:
4120 err = 1;
4121 }
e7fd4179 4122 }
ef0c2bb0
DT		 4123
4124 if (err)
4125 log_error(ls, "recover_waiters_post %x %d %x %d %d",
4126 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou);
4127 unlock_rsb(r);
4128 put_rsb(r);
4129 dlm_put_lkb(lkb);
e7fd4179
DT		 4130 }
4131
4132 return error;
4133}
4134
4135static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
4136 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
4137{
4138 struct dlm_ls *ls = r->res_ls;
4139 struct dlm_lkb *lkb, *safe;
4140
4141 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
4142 if (test(ls, lkb)) {
97a35d1e 4143 rsb_set_flag(r, RSB_LOCKS_PURGED);
e7fd4179
DT		 4144 del_lkb(r, lkb);
4145 /* this put should free the lkb */
b3f58d8f 4146 if (!dlm_put_lkb(lkb))
e7fd4179
DT		 4147 log_error(ls, "purged lkb not released");
4148 }
4149 }
4150}
4151
4152static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4153{
4154 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
4155}
4156
4157static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4158{
4159 return is_master_copy(lkb);
4160}
4161
4162static void purge_dead_locks(struct dlm_rsb *r)
4163{
4164 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
4165 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
4166 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
4167}
4168
4169void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
4170{
4171 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
4172 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
4173 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
4174}
4175
4176/* Get rid of locks held by nodes that are gone. */
4177
4178int dlm_purge_locks(struct dlm_ls *ls)
4179{
4180 struct dlm_rsb *r;
4181
4182 log_debug(ls, "dlm_purge_locks");
4183
4184 down_write(&ls->ls_root_sem);
4185 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
4186 hold_rsb(r);
4187 lock_rsb(r);
4188 if (is_master(r))
4189 purge_dead_locks(r);
4190 unlock_rsb(r);
4191 unhold_rsb(r);
4192
4193 schedule();
4194 }
4195 up_write(&ls->ls_root_sem);
4196
4197 return 0;
4198}
4199
97a35d1e
DT		 4200static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
4201{
4202 struct dlm_rsb *r, *r_ret = NULL;
4203
4204 read_lock(&ls->ls_rsbtbl[bucket].lock);
4205 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
4206 if (!rsb_flag(r, RSB_LOCKS_PURGED))
4207 continue;
4208 hold_rsb(r);
4209 rsb_clear_flag(r, RSB_LOCKS_PURGED);
4210 r_ret = r;
4211 break;
4212 }
4213 read_unlock(&ls->ls_rsbtbl[bucket].lock);
4214 return r_ret;
4215}
4216
4217void dlm_grant_after_purge(struct dlm_ls *ls)
e7fd4179
DT		 4218{
4219 struct dlm_rsb *r;
2b4e926a 4220 int bucket = 0;
e7fd4179 4221
2b4e926a
DT		 4222 while (1) {
4223 r = find_purged_rsb(ls, bucket);
4224 if (!r) {
4225 if (bucket == ls->ls_rsbtbl_size - 1)
4226 break;
4227 bucket++;
97a35d1e 4228 continue;
2b4e926a 4229 }
97a35d1e
DT		 4230 lock_rsb(r);
4231 if (is_master(r)) {
4232 grant_pending_locks(r);
4233 confirm_master(r, 0);
e7fd4179 4234 }
97a35d1e
DT		 4235 unlock_rsb(r);
4236 put_rsb(r);
2b4e926a 4237 schedule();
e7fd4179 4238 }
e7fd4179
DT		 4239}
4240
4241static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
4242 uint32_t remid)
4243{
4244 struct dlm_lkb *lkb;
4245
4246 list_for_each_entry(lkb, head, lkb_statequeue) {
4247 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
4248 return lkb;
4249 }
4250 return NULL;
4251}
4252
4253static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
4254 uint32_t remid)
4255{
4256 struct dlm_lkb *lkb;
4257
4258 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
4259 if (lkb)
4260 return lkb;
4261 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
4262 if (lkb)
4263 return lkb;
4264 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
4265 if (lkb)
4266 return lkb;
4267 return NULL;
4268}
4269
4270static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
4271 struct dlm_rsb *r, struct dlm_rcom *rc)
4272{
4273 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4274 int lvblen;
4275
4276 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
4277 lkb->lkb_ownpid = rl->rl_ownpid;
4278 lkb->lkb_remid = rl->rl_lkid;
4279 lkb->lkb_exflags = rl->rl_exflags;
4280 lkb->lkb_flags = rl->rl_flags & 0x0000FFFF;
4281 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4282 lkb->lkb_lvbseq = rl->rl_lvbseq;
4283 lkb->lkb_rqmode = rl->rl_rqmode;
4284 lkb->lkb_grmode = rl->rl_grmode;
4285 /* don't set lkb_status because add_lkb wants to itself */
4286
4287 lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
4288 lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
4289
e7fd4179 4290 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
52bda2b5 4291 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
e7fd4179
DT		 4292 if (!lkb->lkb_lvbptr)
4293 return -ENOMEM;
4294 lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
4295 sizeof(struct rcom_lock);
4296 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
4297 }
4298
4299 /* Conversions between PR and CW (middle modes) need special handling.
4300 The real granted mode of these converting locks cannot be determined
4301 until all locks have been rebuilt on the rsb (recover_conversion) */
4302
4303 if (rl->rl_wait_type == DLM_MSG_CONVERT && middle_conversion(lkb)) {
4304 rl->rl_status = DLM_LKSTS_CONVERT;
4305 lkb->lkb_grmode = DLM_LOCK_IV;
4306 rsb_set_flag(r, RSB_RECOVER_CONVERT);
4307 }
4308
4309 return 0;
4310}
4311
4312/* This lkb may have been recovered in a previous aborted recovery so we need
4313 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4314 If so we just send back a standard reply. If not, we create a new lkb with
4315 the given values and send back our lkid. We send back our lkid by sending
4316 back the rcom_lock struct we got but with the remid field filled in. */
4317
4318int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4319{
4320 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4321 struct dlm_rsb *r;
4322 struct dlm_lkb *lkb;
4323 int error;
4324
4325 if (rl->rl_parent_lkid) {
4326 error = -EOPNOTSUPP;
4327 goto out;
4328 }
4329
4330 error = find_rsb(ls, rl->rl_name, rl->rl_namelen, R_MASTER, &r);
4331 if (error)
4332 goto out;
4333
4334 lock_rsb(r);
4335
4336 lkb = search_remid(r, rc->rc_header.h_nodeid, rl->rl_lkid);
4337 if (lkb) {
4338 error = -EEXIST;
4339 goto out_remid;
4340 }
4341
4342 error = create_lkb(ls, &lkb);
4343 if (error)
4344 goto out_unlock;
4345
4346 error = receive_rcom_lock_args(ls, lkb, r, rc);
4347 if (error) {
b3f58d8f 4348 __put_lkb(ls, lkb);
e7fd4179
DT		 4349 goto out_unlock;
4350 }
4351
4352 attach_lkb(r, lkb);
4353 add_lkb(r, lkb, rl->rl_status);
4354 error = 0;
4355
4356 out_remid:
4357 /* this is the new value returned to the lock holder for
4358 saving in its process-copy lkb */
4359 rl->rl_remid = lkb->lkb_id;
4360
4361 out_unlock:
4362 unlock_rsb(r);
4363 put_rsb(r);
4364 out:
4365 if (error)
11b2498b 4366 log_debug(ls, "recover_master_copy %d %x", error, rl->rl_lkid);
e7fd4179
DT		 4367 rl->rl_result = error;
4368 return error;
4369}
4370
4371int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4372{
4373 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4374 struct dlm_rsb *r;
4375 struct dlm_lkb *lkb;
4376 int error;
4377
4378 error = find_lkb(ls, rl->rl_lkid, &lkb);
4379 if (error) {
4380 log_error(ls, "recover_process_copy no lkid %x", rl->rl_lkid);
4381 return error;
4382 }
4383
4384 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
4385
4386 error = rl->rl_result;
4387
4388 r = lkb->lkb_resource;
4389 hold_rsb(r);
4390 lock_rsb(r);
4391
4392 switch (error) {
dc200a88
DT		 4393 case -EBADR:
4394 /* There's a chance the new master received our lock before
4395 dlm_recover_master_reply(), this wouldn't happen if we did
4396 a barrier between recover_masters and recover_locks. */
4397 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
4398 (unsigned long)r, r->res_name);
4399 dlm_send_rcom_lock(r, lkb);
4400 goto out;
e7fd4179
DT		 4401 case -EEXIST:
4402 log_debug(ls, "master copy exists %x", lkb->lkb_id);
4403 /* fall through */
4404 case 0:
4405 lkb->lkb_remid = rl->rl_remid;
4406 break;
4407 default:
4408 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
4409 error, lkb->lkb_id);
4410 }
4411
4412 /* an ack for dlm_recover_locks() which waits for replies from
4413 all the locks it sends to new masters */
4414 dlm_recovered_lock(r);
dc200a88 4415 out:
e7fd4179
DT		 4416 unlock_rsb(r);
4417 put_rsb(r);
b3f58d8f 4418 dlm_put_lkb(lkb);
e7fd4179
DT		 4419
4420 return 0;
4421}
4422
597d0cae
DT		 4423int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
4424 int mode, uint32_t flags, void *name, unsigned int namelen,
d7db923e 4425 unsigned long timeout_cs)
597d0cae
DT		 4426{
4427 struct dlm_lkb *lkb;
4428 struct dlm_args args;
4429 int error;
4430
85e86edf 4431 dlm_lock_recovery(ls);
597d0cae
DT		 4432
4433 error = create_lkb(ls, &lkb);
4434 if (error) {
4435 kfree(ua);
4436 goto out;
4437 }
4438
4439 if (flags & DLM_LKF_VALBLK) {
62a0f623 4440 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
597d0cae
DT		 4441 if (!ua->lksb.sb_lvbptr) {
4442 kfree(ua);
4443 __put_lkb(ls, lkb);
4444 error = -ENOMEM;
4445 goto out;
4446 }
4447 }
4448
52bda2b5 4449 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
597d0cae
DT		 4450 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4451 lock and that lkb_astparam is the dlm_user_args structure. */
4452
d7db923e 4453 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
32f105a1 4454 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
597d0cae
DT		 4455 lkb->lkb_flags |= DLM_IFL_USER;
4456 ua->old_mode = DLM_LOCK_IV;
4457
4458 if (error) {
4459 __put_lkb(ls, lkb);
4460 goto out;
4461 }
4462
4463 error = request_lock(ls, lkb, name, namelen, &args);
4464
4465 switch (error) {
4466 case 0:
4467 break;
4468 case -EINPROGRESS:
4469 error = 0;
4470 break;
4471 case -EAGAIN:
4472 error = 0;
4473 /* fall through */
4474 default:
4475 __put_lkb(ls, lkb);
4476 goto out;
4477 }
4478
4479 /* add this new lkb to the per-process list of locks */
4480 spin_lock(&ua->proc->locks_spin);
ef0c2bb0 4481 hold_lkb(lkb);
597d0cae
DT		 4482 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
4483 spin_unlock(&ua->proc->locks_spin);
4484 out:
85e86edf 4485 dlm_unlock_recovery(ls);
597d0cae
DT		 4486 return error;
4487}
4488
4489int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
d7db923e
DT		 4490 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
4491 unsigned long timeout_cs)
597d0cae
DT		 4492{
4493 struct dlm_lkb *lkb;
4494 struct dlm_args args;
4495 struct dlm_user_args *ua;
4496 int error;
4497
85e86edf 4498 dlm_lock_recovery(ls);
597d0cae
DT		 4499
4500 error = find_lkb(ls, lkid, &lkb);
4501 if (error)
4502 goto out;
4503
4504 /* user can change the params on its lock when it converts it, or
4505 add an lvb that didn't exist before */
4506
4507 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4508
4509 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
62a0f623 4510 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
597d0cae
DT		 4511 if (!ua->lksb.sb_lvbptr) {
4512 error = -ENOMEM;
4513 goto out_put;
4514 }
4515 }
4516 if (lvb_in && ua->lksb.sb_lvbptr)
4517 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4518
d7db923e 4519 ua->xid = ua_tmp->xid;
597d0cae
DT		 4520 ua->castparam = ua_tmp->castparam;
4521 ua->castaddr = ua_tmp->castaddr;
4522 ua->bastparam = ua_tmp->bastparam;
4523 ua->bastaddr = ua_tmp->bastaddr;
10948eb4 4524 ua->user_lksb = ua_tmp->user_lksb;
597d0cae
DT		 4525 ua->old_mode = lkb->lkb_grmode;
4526
d7db923e
DT		 4527 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
4528 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
597d0cae
DT		 4529 if (error)
4530 goto out_put;
4531
4532 error = convert_lock(ls, lkb, &args);
4533
c85d65e9 4534 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
597d0cae
DT		 4535 error = 0;
4536 out_put:
4537 dlm_put_lkb(lkb);
4538 out:
85e86edf 4539 dlm_unlock_recovery(ls);
597d0cae
DT		 4540 kfree(ua_tmp);
4541 return error;
4542}
4543
4544int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4545 uint32_t flags, uint32_t lkid, char *lvb_in)
4546{
4547 struct dlm_lkb *lkb;
4548 struct dlm_args args;
4549 struct dlm_user_args *ua;
4550 int error;
4551
85e86edf 4552 dlm_lock_recovery(ls);
597d0cae
DT		 4553
4554 error = find_lkb(ls, lkid, &lkb);
4555 if (error)
4556 goto out;
4557
4558 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4559
4560 if (lvb_in && ua->lksb.sb_lvbptr)
4561 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
b434eda6
PC		 4562 if (ua_tmp->castparam)
4563 ua->castparam = ua_tmp->castparam;
cc346d55 4564 ua->user_lksb = ua_tmp->user_lksb;
597d0cae
DT		 4565
4566 error = set_unlock_args(flags, ua, &args);
4567 if (error)
4568 goto out_put;
4569
4570 error = unlock_lock(ls, lkb, &args);
4571
4572 if (error == -DLM_EUNLOCK)
4573 error = 0;
ef0c2bb0
DT		 4574 /* from validate_unlock_args() */
4575 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
4576 error = 0;
597d0cae
DT		 4577 if (error)
4578 goto out_put;
4579
4580 spin_lock(&ua->proc->locks_spin);
a1bc86e6
DT		 4581 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4582 if (!list_empty(&lkb->lkb_ownqueue))
4583 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
597d0cae 4584 spin_unlock(&ua->proc->locks_spin);
597d0cae
DT		 4585 out_put:
4586 dlm_put_lkb(lkb);
4587 out:
85e86edf 4588 dlm_unlock_recovery(ls);
ef0c2bb0 4589 kfree(ua_tmp);
597d0cae
DT		 4590 return error;
4591}
4592
4593int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4594 uint32_t flags, uint32_t lkid)
4595{
4596 struct dlm_lkb *lkb;
4597 struct dlm_args args;
4598 struct dlm_user_args *ua;
4599 int error;
4600
85e86edf 4601 dlm_lock_recovery(ls);
597d0cae
DT		 4602
4603 error = find_lkb(ls, lkid, &lkb);
4604 if (error)
4605 goto out;
4606
4607 ua = (struct dlm_user_args *)lkb->lkb_astparam;
b434eda6
PC		 4608 if (ua_tmp->castparam)
4609 ua->castparam = ua_tmp->castparam;
c059f70e 4610 ua->user_lksb = ua_tmp->user_lksb;
597d0cae
DT		 4611
4612 error = set_unlock_args(flags, ua, &args);
4613 if (error)
4614 goto out_put;
4615
4616 error = cancel_lock(ls, lkb, &args);
4617
4618 if (error == -DLM_ECANCEL)
4619 error = 0;
ef0c2bb0
DT		 4620 /* from validate_unlock_args() */
4621 if (error == -EBUSY)
4622 error = 0;
597d0cae
DT		 4623 out_put:
4624 dlm_put_lkb(lkb);
4625 out:
85e86edf 4626 dlm_unlock_recovery(ls);
ef0c2bb0 4627 kfree(ua_tmp);
597d0cae
DT		 4628 return error;
4629}
4630
8b4021fa
DT		 4631int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
4632{
4633 struct dlm_lkb *lkb;
4634 struct dlm_args args;
4635 struct dlm_user_args *ua;
4636 struct dlm_rsb *r;
4637 int error;
4638
4639 dlm_lock_recovery(ls);
4640
4641 error = find_lkb(ls, lkid, &lkb);
4642 if (error)
4643 goto out;
4644
4645 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4646
4647 error = set_unlock_args(flags, ua, &args);
4648 if (error)
4649 goto out_put;
4650
4651 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4652
4653 r = lkb->lkb_resource;
4654 hold_rsb(r);
4655 lock_rsb(r);
4656
4657 error = validate_unlock_args(lkb, &args);
4658 if (error)
4659 goto out_r;
4660 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
4661
4662 error = _cancel_lock(r, lkb);
4663 out_r:
4664 unlock_rsb(r);
4665 put_rsb(r);
4666
4667 if (error == -DLM_ECANCEL)
4668 error = 0;
4669 /* from validate_unlock_args() */
4670 if (error == -EBUSY)
4671 error = 0;
4672 out_put:
4673 dlm_put_lkb(lkb);
4674 out:
4675 dlm_unlock_recovery(ls);
4676 return error;
4677}
4678
ef0c2bb0
DT		 4679/* lkb's that are removed from the waiters list by revert are just left on the
4680 orphans list with the granted orphan locks, to be freed by purge */
4681
597d0cae
DT		 4682static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4683{
4684 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
ef0c2bb0
DT		 4685 struct dlm_args args;
4686 int error;
597d0cae 4687
ef0c2bb0
DT		 4688 hold_lkb(lkb);
4689 mutex_lock(&ls->ls_orphans_mutex);
4690 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
4691 mutex_unlock(&ls->ls_orphans_mutex);
597d0cae 4692
ef0c2bb0
DT		 4693 set_unlock_args(0, ua, &args);
4694
4695 error = cancel_lock(ls, lkb, &args);
4696 if (error == -DLM_ECANCEL)
4697 error = 0;
4698 return error;
597d0cae
DT		 4699}
4700
4701/* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4702 Regardless of what rsb queue the lock is on, it's removed and freed. */
4703
4704static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4705{
4706 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4707 struct dlm_args args;
4708 int error;
4709
597d0cae
DT		 4710 set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
4711
4712 error = unlock_lock(ls, lkb, &args);
4713 if (error == -DLM_EUNLOCK)
4714 error = 0;
4715 return error;
4716}
4717
ef0c2bb0
DT		 4718/* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4719 (which does lock_rsb) due to deadlock with receiving a message that does
4720 lock_rsb followed by dlm_user_add_ast() */
4721
4722static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
4723 struct dlm_user_proc *proc)
4724{
4725 struct dlm_lkb *lkb = NULL;
4726
4727 mutex_lock(&ls->ls_clear_proc_locks);
4728 if (list_empty(&proc->locks))
4729 goto out;
4730
4731 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
4732 list_del_init(&lkb->lkb_ownqueue);
4733
4734 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4735 lkb->lkb_flags |= DLM_IFL_ORPHAN;
4736 else
4737 lkb->lkb_flags |= DLM_IFL_DEAD;
4738 out:
4739 mutex_unlock(&ls->ls_clear_proc_locks);
4740 return lkb;
4741}
4742
597d0cae
DT		 4743/* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4744 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4745 which we clear here. */
4746
4747/* proc CLOSING flag is set so no more device_reads should look at proc->asts
4748 list, and no more device_writes should add lkb's to proc->locks list; so we
4749 shouldn't need to take asts_spin or locks_spin here. this assumes that
4750 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4751 them ourself. */
4752
4753void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4754{
4755 struct dlm_lkb *lkb, *safe;
4756
85e86edf 4757 dlm_lock_recovery(ls);
597d0cae 4758
ef0c2bb0
DT		 4759 while (1) {
4760 lkb = del_proc_lock(ls, proc);
4761 if (!lkb)
4762 break;
84d8cd69 4763 del_timeout(lkb);
ef0c2bb0 4764 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
597d0cae 4765 orphan_proc_lock(ls, lkb);
ef0c2bb0 4766 else
597d0cae 4767 unlock_proc_lock(ls, lkb);
597d0cae
DT		 4768
4769 /* this removes the reference for the proc->locks list
4770 added by dlm_user_request, it may result in the lkb
4771 being freed */
4772
4773 dlm_put_lkb(lkb);
4774 }
a1bc86e6 4775
ef0c2bb0
DT		 4776 mutex_lock(&ls->ls_clear_proc_locks);
4777
a1bc86e6
DT		 4778 /* in-progress unlocks */
4779 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4780 list_del_init(&lkb->lkb_ownqueue);
4781 lkb->lkb_flags |= DLM_IFL_DEAD;
4782 dlm_put_lkb(lkb);
4783 }
4784
4785 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
8a358ca8 4786 lkb->lkb_ast_type = 0;
a1bc86e6
DT		 4787 list_del(&lkb->lkb_astqueue);
4788 dlm_put_lkb(lkb);
4789 }
4790
597d0cae 4791 mutex_unlock(&ls->ls_clear_proc_locks);
85e86edf 4792 dlm_unlock_recovery(ls);
597d0cae 4793}
a1bc86e6 4794
8499137d
DT		 4795static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4796{
4797 struct dlm_lkb *lkb, *safe;
4798
4799 while (1) {
4800 lkb = NULL;
4801 spin_lock(&proc->locks_spin);
4802 if (!list_empty(&proc->locks)) {
4803 lkb = list_entry(proc->locks.next, struct dlm_lkb,
4804 lkb_ownqueue);
4805 list_del_init(&lkb->lkb_ownqueue);
4806 }
4807 spin_unlock(&proc->locks_spin);
4808
4809 if (!lkb)
4810 break;
4811
4812 lkb->lkb_flags |= DLM_IFL_DEAD;
4813 unlock_proc_lock(ls, lkb);
4814 dlm_put_lkb(lkb); /* ref from proc->locks list */
4815 }
4816
4817 spin_lock(&proc->locks_spin);
4818 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4819 list_del_init(&lkb->lkb_ownqueue);
4820 lkb->lkb_flags |= DLM_IFL_DEAD;
4821 dlm_put_lkb(lkb);
4822 }
4823 spin_unlock(&proc->locks_spin);
4824
4825 spin_lock(&proc->asts_spin);
4826 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4827 list_del(&lkb->lkb_astqueue);
4828 dlm_put_lkb(lkb);
4829 }
4830 spin_unlock(&proc->asts_spin);
4831}
4832
4833/* pid of 0 means purge all orphans */
4834
4835static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
4836{
4837 struct dlm_lkb *lkb, *safe;
4838
4839 mutex_lock(&ls->ls_orphans_mutex);
4840 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
4841 if (pid && lkb->lkb_ownpid != pid)
4842 continue;
4843 unlock_proc_lock(ls, lkb);
4844 list_del_init(&lkb->lkb_ownqueue);
4845 dlm_put_lkb(lkb);
4846 }
4847 mutex_unlock(&ls->ls_orphans_mutex);
4848}
4849
4850static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
4851{
4852 struct dlm_message *ms;
4853 struct dlm_mhandle *mh;
4854 int error;
4855
4856 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
4857 DLM_MSG_PURGE, &ms, &mh);
4858 if (error)
4859 return error;
4860 ms->m_nodeid = nodeid;
4861 ms->m_pid = pid;
4862
4863 return send_message(mh, ms);
4864}
4865
4866int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
4867 int nodeid, int pid)
4868{
4869 int error = 0;
4870
4871 if (nodeid != dlm_our_nodeid()) {
4872 error = send_purge(ls, nodeid, pid);
4873 } else {
85e86edf 4874 dlm_lock_recovery(ls);
8499137d
DT		 4875 if (pid == current->pid)
4876 purge_proc_locks(ls, proc);
4877 else
4878 do_purge(ls, nodeid, pid);
85e86edf 4879 dlm_unlock_recovery(ls);
8499137d
DT		 4880 }
4881 return error;
4882}
4883
Linux 6.x block layer and io_uring tree(s)