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