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