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