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