]> Git Repo - linux.git/blame - fs/dlm/lock.c
projects / linux.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
dlm: remove found label in dlm_master_lookup
[linux.git] / fs / dlm / lock.c
CommitLineData
2522fe45 1// SPDX-License-Identifier: GPL-2.0-only
e7fd4179
DT		 2/******************************************************************************
3*******************************************************************************
4**
7fe2b319 5** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved.
e7fd4179 6**
e7fd4179
DT		 7**
8*******************************************************************************
9******************************************************************************/
10
11/* Central locking logic has four stages:
12
13 dlm_lock()
14 dlm_unlock()
15
16 request_lock(ls, lkb)
17 convert_lock(ls, lkb)
18 unlock_lock(ls, lkb)
19 cancel_lock(ls, lkb)
20
21 _request_lock(r, lkb)
22 _convert_lock(r, lkb)
23 _unlock_lock(r, lkb)
24 _cancel_lock(r, lkb)
25
26 do_request(r, lkb)
27 do_convert(r, lkb)
28 do_unlock(r, lkb)
29 do_cancel(r, lkb)
30
31 Stage 1 (lock, unlock) is mainly about checking input args and
32 splitting into one of the four main operations:
33
34 dlm_lock = request_lock
35 dlm_lock+CONVERT = convert_lock
36 dlm_unlock = unlock_lock
37 dlm_unlock+CANCEL = cancel_lock
38
39 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
40 provided to the next stage.
41
42 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
43 When remote, it calls send_xxxx(), when local it calls do_xxxx().
44
45 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
46 given rsb and lkb and queues callbacks.
47
48 For remote operations, send_xxxx() results in the corresponding do_xxxx()
49 function being executed on the remote node. The connecting send/receive
50 calls on local (L) and remote (R) nodes:
51
52 L: send_xxxx() -> R: receive_xxxx()
53 R: do_xxxx()
54 L: receive_xxxx_reply() <- R: send_xxxx_reply()
55*/
f1d3b8f9
AA		 56#include <trace/events/dlm.h>
57
597d0cae 58#include <linux/types.h>
9beb3bf5 59#include <linux/rbtree.h>
5a0e3ad6 60#include <linux/slab.h>
e7fd4179 61#include "dlm_internal.h"
597d0cae 62#include <linux/dlm_device.h>
e7fd4179 63#include "memory.h"
a070a91c 64#include "midcomms.h"
e7fd4179
DT		 65#include "requestqueue.h"
66#include "util.h"
67#include "dir.h"
68#include "member.h"
69#include "lockspace.h"
70#include "ast.h"
71#include "lock.h"
72#include "rcom.h"
73#include "recover.h"
74#include "lvb_table.h"
597d0cae 75#include "user.h"
e7fd4179
DT		 76#include "config.h"
77
78static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
79static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
80static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
81static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
82static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
83static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
84static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
85static int send_remove(struct dlm_rsb *r);
86static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
3ae1acf9 87static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
e7fd4179
DT		 88static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
89 struct dlm_message *ms);
90static int receive_extralen(struct dlm_message *ms);
8499137d 91static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
3ae1acf9 92static void del_timeout(struct dlm_lkb *lkb);
c04fecb4 93static void toss_rsb(struct kref *kref);
e7fd4179
DT		 94
95/*
96 * Lock compatibilty matrix - thanks Steve
97 * UN = Unlocked state. Not really a state, used as a flag
98 * PD = Padding. Used to make the matrix a nice power of two in size
99 * Other states are the same as the VMS DLM.
100 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
101 */
102
103static const int __dlm_compat_matrix[8][8] = {
104 /* UN NL CR CW PR PW EX PD */
105 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
106 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
107 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
108 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
109 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
110 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
111 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
112 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
113};
114
115/*
116 * This defines the direction of transfer of LVB data.
117 * Granted mode is the row; requested mode is the column.
118 * Usage: matrix[grmode+1][rqmode+1]
119 * 1 = LVB is returned to the caller
120 * 0 = LVB is written to the resource
121 * -1 = nothing happens to the LVB
122 */
123
124const int dlm_lvb_operations[8][8] = {
125 /* UN NL CR CW PR PW EX PD*/
126 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
127 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
128 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
129 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
130 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
131 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
132 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
133 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
134};
e7fd4179
DT		 135
136#define modes_compat(gr, rq) \
137 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
138
139int dlm_modes_compat(int mode1, int mode2)
140{
141 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
142}
143
144/*
145 * Compatibility matrix for conversions with QUECVT set.
146 * Granted mode is the row; requested mode is the column.
147 * Usage: matrix[grmode+1][rqmode+1]
148 */
149
150static const int __quecvt_compat_matrix[8][8] = {
151 /* UN NL CR CW PR PW EX PD */
152 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
153 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
154 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
155 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
156 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
157 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
158 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
159 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
160};
161
597d0cae 162void dlm_print_lkb(struct dlm_lkb *lkb)
e7fd4179 163{
6d40c4a7 164 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
4875647a 165 "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
e7fd4179
DT		 166 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
167 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
4875647a
DT		 168 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
169 (unsigned long long)lkb->lkb_recover_seq);
e7fd4179
DT		 170}
171
170e19ab 172static void dlm_print_rsb(struct dlm_rsb *r)
e7fd4179 173{
c04fecb4
DT		 174 printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
175 "rlc %d name %s\n",
176 r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
177 r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
178 r->res_name);
e7fd4179
DT		 179}
180
a345da3e
DT		 181void dlm_dump_rsb(struct dlm_rsb *r)
182{
183 struct dlm_lkb *lkb;
184
185 dlm_print_rsb(r);
186
187 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
188 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
189 printk(KERN_ERR "rsb lookup list\n");
190 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
191 dlm_print_lkb(lkb);
192 printk(KERN_ERR "rsb grant queue:\n");
193 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
194 dlm_print_lkb(lkb);
195 printk(KERN_ERR "rsb convert queue:\n");
196 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
197 dlm_print_lkb(lkb);
198 printk(KERN_ERR "rsb wait queue:\n");
199 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
200 dlm_print_lkb(lkb);
201}
202
e7fd4179
DT		 203/* Threads cannot use the lockspace while it's being recovered */
204
85e86edf 205static inline void dlm_lock_recovery(struct dlm_ls *ls)
e7fd4179
DT		 206{
207 down_read(&ls->ls_in_recovery);
208}
209
85e86edf 210void dlm_unlock_recovery(struct dlm_ls *ls)
e7fd4179
DT		 211{
212 up_read(&ls->ls_in_recovery);
213}
214
85e86edf 215int dlm_lock_recovery_try(struct dlm_ls *ls)
e7fd4179
DT		 216{
217 return down_read_trylock(&ls->ls_in_recovery);
218}
219
220static inline int can_be_queued(struct dlm_lkb *lkb)
221{
222 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
223}
224
225static inline int force_blocking_asts(struct dlm_lkb *lkb)
226{
227 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
228}
229
230static inline int is_demoted(struct dlm_lkb *lkb)
231{
232 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
233}
234
7d3c1feb
DT		 235static inline int is_altmode(struct dlm_lkb *lkb)
236{
237 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
238}
239
240static inline int is_granted(struct dlm_lkb *lkb)
241{
242 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
243}
244
e7fd4179
DT		 245static inline int is_remote(struct dlm_rsb *r)
246{
247 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
248 return !!r->res_nodeid;
249}
250
251static inline int is_process_copy(struct dlm_lkb *lkb)
252{
253 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
254}
255
256static inline int is_master_copy(struct dlm_lkb *lkb)
257{
90135925 258 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
e7fd4179
DT		 259}
260
261static inline int middle_conversion(struct dlm_lkb *lkb)
262{
263 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
264 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
90135925
DT		 265 return 1;
266 return 0;
e7fd4179
DT		 267}
268
269static inline int down_conversion(struct dlm_lkb *lkb)
270{
271 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
272}
273
ef0c2bb0
DT		 274static inline int is_overlap_unlock(struct dlm_lkb *lkb)
275{
276 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
277}
278
279static inline int is_overlap_cancel(struct dlm_lkb *lkb)
280{
281 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
282}
283
284static inline int is_overlap(struct dlm_lkb *lkb)
285{
286 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
287 DLM_IFL_OVERLAP_CANCEL));
288}
289
e7fd4179
DT		 290static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
291{
292 if (is_master_copy(lkb))
293 return;
294
3ae1acf9
DT		 295 del_timeout(lkb);
296
e7fd4179
DT		 297 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
298
3ae1acf9
DT		 299 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
300 timeout caused the cancel then return -ETIMEDOUT */
301 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
302 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
303 rv = -ETIMEDOUT;
304 }
305
8b4021fa
DT		 306 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
307 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
308 rv = -EDEADLK;
309 }
310
23e8e1aa 311 dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
e7fd4179
DT		 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{
b6fa8796 322 if (is_master_copy(lkb)) {
e7fd4179 323 send_bast(r, lkb, rqmode);
b6fa8796 324 } else {
23e8e1aa 325 dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
b6fa8796 326 }
e7fd4179
DT		 327}
328
329/*
330 * Basic operations on rsb's and lkb's
331 */
332
c04fecb4
DT		 333/* This is only called to add a reference when the code already holds
334 a valid reference to the rsb, so there's no need for locking. */
335
336static inline void hold_rsb(struct dlm_rsb *r)
337{
338 kref_get(&r->res_ref);
339}
340
341void dlm_hold_rsb(struct dlm_rsb *r)
342{
343 hold_rsb(r);
344}
345
346/* When all references to the rsb are gone it's transferred to
347 the tossed list for later disposal. */
348
349static void put_rsb(struct dlm_rsb *r)
350{
351 struct dlm_ls *ls = r->res_ls;
352 uint32_t bucket = r->res_bucket;
353
354 spin_lock(&ls->ls_rsbtbl[bucket].lock);
355 kref_put(&r->res_ref, toss_rsb);
356 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
357}
358
359void dlm_put_rsb(struct dlm_rsb *r)
360{
361 put_rsb(r);
362}
363
3881ac04
DT		 364static int pre_rsb_struct(struct dlm_ls *ls)
365{
366 struct dlm_rsb *r1, *r2;
367 int count = 0;
368
369 spin_lock(&ls->ls_new_rsb_spin);
370 if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
371 spin_unlock(&ls->ls_new_rsb_spin);
372 return 0;
373 }
374 spin_unlock(&ls->ls_new_rsb_spin);
375
376 r1 = dlm_allocate_rsb(ls);
377 r2 = dlm_allocate_rsb(ls);
378
379 spin_lock(&ls->ls_new_rsb_spin);
380 if (r1) {
381 list_add(&r1->res_hashchain, &ls->ls_new_rsb);
382 ls->ls_new_rsb_count++;
383 }
384 if (r2) {
385 list_add(&r2->res_hashchain, &ls->ls_new_rsb);
386 ls->ls_new_rsb_count++;
387 }
388 count = ls->ls_new_rsb_count;
389 spin_unlock(&ls->ls_new_rsb_spin);
390
391 if (!count)
392 return -ENOMEM;
393 return 0;
394}
395
396/* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
397 unlock any spinlocks, go back and call pre_rsb_struct again.
398 Otherwise, take an rsb off the list and return it. */
399
400static int get_rsb_struct(struct dlm_ls *ls, char *name, int len,
401 struct dlm_rsb **r_ret)
e7fd4179
DT		 402{
403 struct dlm_rsb *r;
3881ac04
DT		 404 int count;
405
406 spin_lock(&ls->ls_new_rsb_spin);
407 if (list_empty(&ls->ls_new_rsb)) {
408 count = ls->ls_new_rsb_count;
409 spin_unlock(&ls->ls_new_rsb_spin);
410 log_debug(ls, "find_rsb retry %d %d %s",
411 count, dlm_config.ci_new_rsb_count, name);
412 return -EAGAIN;
413 }
e7fd4179 414
3881ac04
DT		 415 r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
416 list_del(&r->res_hashchain);
9beb3bf5
BP		 417 /* Convert the empty list_head to a NULL rb_node for tree usage: */
418 memset(&r->res_hashnode, 0, sizeof(struct rb_node));
3881ac04
DT		 419 ls->ls_new_rsb_count--;
420 spin_unlock(&ls->ls_new_rsb_spin);
e7fd4179
DT		 421
422 r->res_ls = ls;
423 r->res_length = len;
424 memcpy(r->res_name, name, len);
90135925 425 mutex_init(&r->res_mutex);
e7fd4179
DT		 426
427 INIT_LIST_HEAD(&r->res_lookup);
428 INIT_LIST_HEAD(&r->res_grantqueue);
429 INIT_LIST_HEAD(&r->res_convertqueue);
430 INIT_LIST_HEAD(&r->res_waitqueue);
431 INIT_LIST_HEAD(&r->res_root_list);
432 INIT_LIST_HEAD(&r->res_recover_list);
433
3881ac04
DT		 434 *r_ret = r;
435 return 0;
e7fd4179
DT		 436}
437
9beb3bf5
BP		 438static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
439{
440 char maxname[DLM_RESNAME_MAXLEN];
441
442 memset(maxname, 0, DLM_RESNAME_MAXLEN);
443 memcpy(maxname, name, nlen);
444 return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
445}
446
7210cb7a 447int dlm_search_rsb_tree(struct rb_root *tree, char *name, int len,
c04fecb4 448 struct dlm_rsb **r_ret)
e7fd4179 449{
9beb3bf5 450 struct rb_node *node = tree->rb_node;
e7fd4179 451 struct dlm_rsb *r;
9beb3bf5
BP		 452 int rc;
453
454 while (node) {
455 r = rb_entry(node, struct dlm_rsb, res_hashnode);
456 rc = rsb_cmp(r, name, len);
457 if (rc < 0)
458 node = node->rb_left;
459 else if (rc > 0)
460 node = node->rb_right;
461 else
e7fd4179
DT		 462 goto found;
463 }
18c60c0a 464 *r_ret = NULL;
597d0cae 465 return -EBADR;
e7fd4179
DT		 466
467 found:
e7fd4179 468 *r_ret = r;
c04fecb4 469 return 0;
e7fd4179
DT		 470}
471
9beb3bf5
BP		 472static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
473{
474 struct rb_node **newn = &tree->rb_node;
475 struct rb_node *parent = NULL;
476 int rc;
477
478 while (*newn) {
479 struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
480 res_hashnode);
481
482 parent = *newn;
483 rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
484 if (rc < 0)
485 newn = &parent->rb_left;
486 else if (rc > 0)
487 newn = &parent->rb_right;
488 else {
489 log_print("rsb_insert match");
490 dlm_dump_rsb(rsb);
491 dlm_dump_rsb(cur);
492 return -EEXIST;
493 }
494 }
495
496 rb_link_node(&rsb->res_hashnode, parent, newn);
497 rb_insert_color(&rsb->res_hashnode, tree);
498 return 0;
499}
500
c04fecb4
DT		 501/*
502 * Find rsb in rsbtbl and potentially create/add one
503 *
504 * Delaying the release of rsb's has a similar benefit to applications keeping
505 * NL locks on an rsb, but without the guarantee that the cached master value
506 * will still be valid when the rsb is reused. Apps aren't always smart enough
507 * to keep NL locks on an rsb that they may lock again shortly; this can lead
508 * to excessive master lookups and removals if we don't delay the release.
509 *
510 * Searching for an rsb means looking through both the normal list and toss
511 * list. When found on the toss list the rsb is moved to the normal list with
512 * ref count of 1; when found on normal list the ref count is incremented.
513 *
514 * rsb's on the keep list are being used locally and refcounted.
515 * rsb's on the toss list are not being used locally, and are not refcounted.
516 *
517 * The toss list rsb's were either
518 * - previously used locally but not any more (were on keep list, then
519 * moved to toss list when last refcount dropped)
520 * - created and put on toss list as a directory record for a lookup
521 * (we are the dir node for the res, but are not using the res right now,
522 * but some other node is)
523 *
524 * The purpose of find_rsb() is to return a refcounted rsb for local use.
525 * So, if the given rsb is on the toss list, it is moved to the keep list
526 * before being returned.
527 *
528 * toss_rsb() happens when all local usage of the rsb is done, i.e. no
529 * more refcounts exist, so the rsb is moved from the keep list to the
530 * toss list.
531 *
532 * rsb's on both keep and toss lists are used for doing a name to master
533 * lookups. rsb's that are in use locally (and being refcounted) are on
534 * the keep list, rsb's that are not in use locally (not refcounted) and
535 * only exist for name/master lookups are on the toss list.
536 *
537 * rsb's on the toss list who's dir_nodeid is not local can have stale
538 * name/master mappings. So, remote requests on such rsb's can potentially
539 * return with an error, which means the mapping is stale and needs to
540 * be updated with a new lookup. (The idea behind MASTER UNCERTAIN and
541 * first_lkid is to keep only a single outstanding request on an rsb
542 * while that rsb has a potentially stale master.)
543 */
544
545static int find_rsb_dir(struct dlm_ls *ls, char *name, int len,
546 uint32_t hash, uint32_t b,
547 int dir_nodeid, int from_nodeid,
548 unsigned int flags, struct dlm_rsb **r_ret)
e7fd4179 549{
c04fecb4
DT		 550 struct dlm_rsb *r = NULL;
551 int our_nodeid = dlm_our_nodeid();
552 int from_local = 0;
553 int from_other = 0;
554 int from_dir = 0;
555 int create = 0;
e7fd4179
DT		 556 int error;
557
c04fecb4
DT		 558 if (flags & R_RECEIVE_REQUEST) {
559 if (from_nodeid == dir_nodeid)
560 from_dir = 1;
561 else
562 from_other = 1;
563 } else if (flags & R_REQUEST) {
564 from_local = 1;
565 }
566
567 /*
568 * flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
569 * from_nodeid has sent us a lock in dlm_recover_locks, believing
570 * we're the new master. Our local recovery may not have set
571 * res_master_nodeid to our_nodeid yet, so allow either. Don't
572 * create the rsb; dlm_recover_process_copy() will handle EBADR
573 * by resending.
574 *
575 * If someone sends us a request, we are the dir node, and we do
576 * not find the rsb anywhere, then recreate it. This happens if
577 * someone sends us a request after we have removed/freed an rsb
578 * from our toss list. (They sent a request instead of lookup
579 * because they are using an rsb from their toss list.)
580 */
581
582 if (from_local || from_dir ||
583 (from_other && (dir_nodeid == our_nodeid))) {
584 create = 1;
e7fd4179 585 }
57638bf3 586
c04fecb4
DT		 587 retry:
588 if (create) {
589 error = pre_rsb_struct(ls);
590 if (error < 0)
591 goto out;
592 }
593
594 spin_lock(&ls->ls_rsbtbl[b].lock);
595
596 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
e7fd4179 597 if (error)
c04fecb4
DT		 598 goto do_toss;
599
600 /*
601 * rsb is active, so we can't check master_nodeid without lock_rsb.
602 */
e7fd4179 603
c04fecb4
DT		 604 kref_get(&r->res_ref);
605 error = 0;
606 goto out_unlock;
607
608
609 do_toss:
610 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
9beb3bf5 611 if (error)
c04fecb4 612 goto do_new;
e7fd4179 613
c04fecb4
DT		 614 /*
615 * rsb found inactive (master_nodeid may be out of date unless
616 * we are the dir_nodeid or were the master) No other thread
617 * is using this rsb because it's on the toss list, so we can
618 * look at or update res_master_nodeid without lock_rsb.
619 */
e7fd4179 620
c04fecb4
DT		 621 if ((r->res_master_nodeid != our_nodeid) && from_other) {
622 /* our rsb was not master, and another node (not the dir node)
623 has sent us a request */
624 log_debug(ls, "find_rsb toss from_other %d master %d dir %d %s",
625 from_nodeid, r->res_master_nodeid, dir_nodeid,
626 r->res_name);
627 error = -ENOTBLK;
628 goto out_unlock;
629 }
630
631 if ((r->res_master_nodeid != our_nodeid) && from_dir) {
632 /* don't think this should ever happen */
633 log_error(ls, "find_rsb toss from_dir %d master %d",
634 from_nodeid, r->res_master_nodeid);
635 dlm_print_rsb(r);
636 /* fix it and go on */
637 r->res_master_nodeid = our_nodeid;
638 r->res_nodeid = 0;
e7fd4179
DT		 639 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
640 r->res_first_lkid = 0;
c04fecb4
DT		 641 }
642
643 if (from_local && (r->res_master_nodeid != our_nodeid)) {
644 /* Because we have held no locks on this rsb,
645 res_master_nodeid could have become stale. */
e7fd4179
DT		 646 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
647 r->res_first_lkid = 0;
c04fecb4
DT		 648 }
649
650 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
651 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
652 goto out_unlock;
653
654
655 do_new:
656 /*
657 * rsb not found
658 */
659
660 if (error == -EBADR && !create)
661 goto out_unlock;
662
663 error = get_rsb_struct(ls, name, len, &r);
664 if (error == -EAGAIN) {
665 spin_unlock(&ls->ls_rsbtbl[b].lock);
666 goto retry;
667 }
668 if (error)
669 goto out_unlock;
670
671 r->res_hash = hash;
672 r->res_bucket = b;
673 r->res_dir_nodeid = dir_nodeid;
674 kref_init(&r->res_ref);
675
676 if (from_dir) {
677 /* want to see how often this happens */
678 log_debug(ls, "find_rsb new from_dir %d recreate %s",
679 from_nodeid, r->res_name);
680 r->res_master_nodeid = our_nodeid;
681 r->res_nodeid = 0;
682 goto out_add;
683 }
684
685 if (from_other && (dir_nodeid != our_nodeid)) {
686 /* should never happen */
687 log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
688 from_nodeid, dir_nodeid, our_nodeid, r->res_name);
689 dlm_free_rsb(r);
e8243f32 690 r = NULL;
c04fecb4
DT		 691 error = -ENOTBLK;
692 goto out_unlock;
693 }
694
695 if (from_other) {
696 log_debug(ls, "find_rsb new from_other %d dir %d %s",
697 from_nodeid, dir_nodeid, r->res_name);
698 }
699
700 if (dir_nodeid == our_nodeid) {
701 /* When we are the dir nodeid, we can set the master
702 node immediately */
703 r->res_master_nodeid = our_nodeid;
704 r->res_nodeid = 0;
e7fd4179 705 } else {
c04fecb4
DT		 706 /* set_master will send_lookup to dir_nodeid */
707 r->res_master_nodeid = 0;
708 r->res_nodeid = -1;
709 }
710
711 out_add:
712 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
713 out_unlock:
714 spin_unlock(&ls->ls_rsbtbl[b].lock);
715 out:
716 *r_ret = r;
717 return error;
718}
719
720/* During recovery, other nodes can send us new MSTCPY locks (from
721 dlm_recover_locks) before we've made ourself master (in
722 dlm_recover_masters). */
723
724static int find_rsb_nodir(struct dlm_ls *ls, char *name, int len,
725 uint32_t hash, uint32_t b,
726 int dir_nodeid, int from_nodeid,
727 unsigned int flags, struct dlm_rsb **r_ret)
728{
729 struct dlm_rsb *r = NULL;
730 int our_nodeid = dlm_our_nodeid();
731 int recover = (flags & R_RECEIVE_RECOVER);
732 int error;
733
734 retry:
735 error = pre_rsb_struct(ls);
736 if (error < 0)
737 goto out;
738
739 spin_lock(&ls->ls_rsbtbl[b].lock);
740
741 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
742 if (error)
743 goto do_toss;
744
745 /*
746 * rsb is active, so we can't check master_nodeid without lock_rsb.
747 */
748
749 kref_get(&r->res_ref);
750 goto out_unlock;
751
752
753 do_toss:
754 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
755 if (error)
756 goto do_new;
757
758 /*
759 * rsb found inactive. No other thread is using this rsb because
760 * it's on the toss list, so we can look at or update
761 * res_master_nodeid without lock_rsb.
762 */
763
764 if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
765 /* our rsb is not master, and another node has sent us a
766 request; this should never happen */
767 log_error(ls, "find_rsb toss from_nodeid %d master %d dir %d",
768 from_nodeid, r->res_master_nodeid, dir_nodeid);
769 dlm_print_rsb(r);
770 error = -ENOTBLK;
771 goto out_unlock;
e7fd4179 772 }
c04fecb4
DT		 773
774 if (!recover && (r->res_master_nodeid != our_nodeid) &&
775 (dir_nodeid == our_nodeid)) {
776 /* our rsb is not master, and we are dir; may as well fix it;
777 this should never happen */
778 log_error(ls, "find_rsb toss our %d master %d dir %d",
779 our_nodeid, r->res_master_nodeid, dir_nodeid);
780 dlm_print_rsb(r);
781 r->res_master_nodeid = our_nodeid;
782 r->res_nodeid = 0;
783 }
784
785 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
786 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
787 goto out_unlock;
788
789
790 do_new:
791 /*
792 * rsb not found
793 */
794
795 error = get_rsb_struct(ls, name, len, &r);
796 if (error == -EAGAIN) {
797 spin_unlock(&ls->ls_rsbtbl[b].lock);
798 goto retry;
799 }
800 if (error)
801 goto out_unlock;
802
803 r->res_hash = hash;
804 r->res_bucket = b;
805 r->res_dir_nodeid = dir_nodeid;
806 r->res_master_nodeid = dir_nodeid;
807 r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
808 kref_init(&r->res_ref);
809
810 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
811 out_unlock:
812 spin_unlock(&ls->ls_rsbtbl[b].lock);
e7fd4179
DT		 813 out:
814 *r_ret = r;
815 return error;
816}
817
c04fecb4
DT		 818static int find_rsb(struct dlm_ls *ls, char *name, int len, int from_nodeid,
819 unsigned int flags, struct dlm_rsb **r_ret)
820{
821 uint32_t hash, b;
822 int dir_nodeid;
823
824 if (len > DLM_RESNAME_MAXLEN)
825 return -EINVAL;
826
827 hash = jhash(name, len, 0);
828 b = hash & (ls->ls_rsbtbl_size - 1);
829
830 dir_nodeid = dlm_hash2nodeid(ls, hash);
831
832 if (dlm_no_directory(ls))
833 return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
834 from_nodeid, flags, r_ret);
835 else
836 return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
837 from_nodeid, flags, r_ret);
838}
839
840/* we have received a request and found that res_master_nodeid != our_nodeid,
841 so we need to return an error or make ourself the master */
842
843static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
844 int from_nodeid)
845{
846 if (dlm_no_directory(ls)) {
847 log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
848 from_nodeid, r->res_master_nodeid,
849 r->res_dir_nodeid);
850 dlm_print_rsb(r);
851 return -ENOTBLK;
852 }
853
854 if (from_nodeid != r->res_dir_nodeid) {
855 /* our rsb is not master, and another node (not the dir node)
856 has sent us a request. this is much more common when our
857 master_nodeid is zero, so limit debug to non-zero. */
858
859 if (r->res_master_nodeid) {
860 log_debug(ls, "validate master from_other %d master %d "
861 "dir %d first %x %s", from_nodeid,
862 r->res_master_nodeid, r->res_dir_nodeid,
863 r->res_first_lkid, r->res_name);
864 }
865 return -ENOTBLK;
866 } else {
867 /* our rsb is not master, but the dir nodeid has sent us a
868 request; this could happen with master 0 / res_nodeid -1 */
869
870 if (r->res_master_nodeid) {
871 log_error(ls, "validate master from_dir %d master %d "
872 "first %x %s",
873 from_nodeid, r->res_master_nodeid,
874 r->res_first_lkid, r->res_name);
875 }
876
877 r->res_master_nodeid = dlm_our_nodeid();
878 r->res_nodeid = 0;
879 return 0;
880 }
881}
882
e7fd4179 883/*
c04fecb4
DT		 884 * We're the dir node for this res and another node wants to know the
885 * master nodeid. During normal operation (non recovery) this is only
886 * called from receive_lookup(); master lookups when the local node is
887 * the dir node are done by find_rsb().
e7fd4179 888 *
c04fecb4
DT		 889 * normal operation, we are the dir node for a resource
890 * . _request_lock
891 * . set_master
892 * . send_lookup
893 * . receive_lookup
894 * . dlm_master_lookup flags 0
e7fd4179 895 *
c04fecb4
DT		 896 * recover directory, we are rebuilding dir for all resources
897 * . dlm_recover_directory
898 * . dlm_rcom_names
899 * remote node sends back the rsb names it is master of and we are dir of
900 * . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
901 * we either create new rsb setting remote node as master, or find existing
902 * rsb and set master to be the remote node.
903 *
904 * recover masters, we are finding the new master for resources
905 * . dlm_recover_masters
906 * . recover_master
907 * . dlm_send_rcom_lookup
908 * . receive_rcom_lookup
909 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
e7fd4179
DT		 910 */
911
c04fecb4
DT		 912int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
913 unsigned int flags, int *r_nodeid, int *result)
e7fd4179 914{
3881ac04 915 struct dlm_rsb *r = NULL;
c04fecb4
DT		 916 uint32_t hash, b;
917 int from_master = (flags & DLM_LU_RECOVER_DIR);
918 int fix_master = (flags & DLM_LU_RECOVER_MASTER);
919 int our_nodeid = dlm_our_nodeid();
920 int dir_nodeid, error, toss_list = 0;
ef58bcca 921
c04fecb4
DT		 922 if (len > DLM_RESNAME_MAXLEN)
923 return -EINVAL;
924
925 if (from_nodeid == our_nodeid) {
926 log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
927 our_nodeid, flags);
928 return -EINVAL;
3881ac04 929 }
e7fd4179 930
c04fecb4
DT		 931 hash = jhash(name, len, 0);
932 b = hash & (ls->ls_rsbtbl_size - 1);
e7fd4179 933
c04fecb4
DT		 934 dir_nodeid = dlm_hash2nodeid(ls, hash);
935 if (dir_nodeid != our_nodeid) {
936 log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
937 from_nodeid, dir_nodeid, our_nodeid, hash,
938 ls->ls_num_nodes);
939 *r_nodeid = -1;
940 return -EINVAL;
941 }
e7fd4179 942
3881ac04 943 retry:
c04fecb4
DT		 944 error = pre_rsb_struct(ls);
945 if (error < 0)
946 return error;
947
948 spin_lock(&ls->ls_rsbtbl[b].lock);
949 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
950 if (!error) {
951 /* because the rsb is active, we need to lock_rsb before
952 checking/changing re_master_nodeid */
953
954 hold_rsb(r);
955 spin_unlock(&ls->ls_rsbtbl[b].lock);
956 lock_rsb(r);
e91ce03b
AA		 957 } else {
958 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
959 if (error)
960 goto not_found;
3881ac04 961
e91ce03b
AA		 962 /* because the rsb is inactive (on toss list), it's not refcounted
963 * and lock_rsb is not used, but is protected by the rsbtbl lock
964 */
3881ac04 965
e91ce03b
AA		 966 toss_list = 1;
967 }
e7fd4179 968
c04fecb4
DT		 969 if (r->res_dir_nodeid != our_nodeid) {
970 /* should not happen, but may as well fix it and carry on */
971 log_error(ls, "dlm_master_lookup res_dir %d our %d %s",
972 r->res_dir_nodeid, our_nodeid, r->res_name);
973 r->res_dir_nodeid = our_nodeid;
974 }
975
976 if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
977 /* Recovery uses this function to set a new master when
978 the previous master failed. Setting NEW_MASTER will
979 force dlm_recover_masters to call recover_master on this
980 rsb even though the res_nodeid is no longer removed. */
981
982 r->res_master_nodeid = from_nodeid;
983 r->res_nodeid = from_nodeid;
984 rsb_set_flag(r, RSB_NEW_MASTER);
985
986 if (toss_list) {
987 /* I don't think we should ever find it on toss list. */
988 log_error(ls, "dlm_master_lookup fix_master on toss");
989 dlm_dump_rsb(r);
990 }
991 }
e7fd4179 992
c04fecb4
DT		 993 if (from_master && (r->res_master_nodeid != from_nodeid)) {
994 /* this will happen if from_nodeid became master during
995 a previous recovery cycle, and we aborted the previous
996 cycle before recovering this master value */
997
998 log_limit(ls, "dlm_master_lookup from_master %d "
999 "master_nodeid %d res_nodeid %d first %x %s",
1000 from_nodeid, r->res_master_nodeid, r->res_nodeid,
1001 r->res_first_lkid, r->res_name);
1002
1003 if (r->res_master_nodeid == our_nodeid) {
1004 log_error(ls, "from_master %d our_master", from_nodeid);
1005 dlm_dump_rsb(r);
c04fecb4
DT		 1006 goto out_found;
1007 }
1008
1009 r->res_master_nodeid = from_nodeid;
1010 r->res_nodeid = from_nodeid;
1011 rsb_set_flag(r, RSB_NEW_MASTER);
1012 }
1013
1014 if (!r->res_master_nodeid) {
1015 /* this will happen if recovery happens while we're looking
1016 up the master for this rsb */
1017
1018 log_debug(ls, "dlm_master_lookup master 0 to %d first %x %s",
1019 from_nodeid, r->res_first_lkid, r->res_name);
1020 r->res_master_nodeid = from_nodeid;
1021 r->res_nodeid = from_nodeid;
1022 }
1023
1024 if (!from_master && !fix_master &&
1025 (r->res_master_nodeid == from_nodeid)) {
1026 /* this can happen when the master sends remove, the dir node
1027 finds the rsb on the keep list and ignores the remove,
1028 and the former master sends a lookup */
1029
1030 log_limit(ls, "dlm_master_lookup from master %d flags %x "
1031 "first %x %s", from_nodeid, flags,
1032 r->res_first_lkid, r->res_name);
1033 }
1034
1035 out_found:
1036 *r_nodeid = r->res_master_nodeid;
1037 if (result)
1038 *result = DLM_LU_MATCH;
1039
1040 if (toss_list) {
1041 r->res_toss_time = jiffies;
1042 /* the rsb was inactive (on toss list) */
1043 spin_unlock(&ls->ls_rsbtbl[b].lock);
1044 } else {
1045 /* the rsb was active */
1046 unlock_rsb(r);
1047 put_rsb(r);
1048 }
1049 return 0;
e7fd4179 1050
c04fecb4
DT		 1051 not_found:
1052 error = get_rsb_struct(ls, name, len, &r);
3881ac04 1053 if (error == -EAGAIN) {
c04fecb4 1054 spin_unlock(&ls->ls_rsbtbl[b].lock);
3881ac04
DT		 1055 goto retry;
1056 }
1057 if (error)
1058 goto out_unlock;
e7fd4179
DT		 1059
1060 r->res_hash = hash;
c04fecb4
DT		 1061 r->res_bucket = b;
1062 r->res_dir_nodeid = our_nodeid;
1063 r->res_master_nodeid = from_nodeid;
1064 r->res_nodeid = from_nodeid;
e7fd4179 1065 kref_init(&r->res_ref);
c04fecb4 1066 r->res_toss_time = jiffies;
e7fd4179 1067
c04fecb4
DT		 1068 error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1069 if (error) {
1070 /* should never happen */
1071 dlm_free_rsb(r);
1072 spin_unlock(&ls->ls_rsbtbl[b].lock);
1073 goto retry;
e7fd4179 1074 }
c04fecb4
DT		 1075
1076 if (result)
1077 *result = DLM_LU_ADD;
1078 *r_nodeid = from_nodeid;
1079 error = 0;
3881ac04 1080 out_unlock:
c04fecb4 1081 spin_unlock(&ls->ls_rsbtbl[b].lock);
e7fd4179
DT		 1082 return error;
1083}
1084
6d40c4a7
DT		 1085static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1086{
1087 struct rb_node *n;
1088 struct dlm_rsb *r;
1089 int i;
1090
1091 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1092 spin_lock(&ls->ls_rsbtbl[i].lock);
1093 for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1094 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1095 if (r->res_hash == hash)
1096 dlm_dump_rsb(r);
1097 }
1098 spin_unlock(&ls->ls_rsbtbl[i].lock);
1099 }
1100}
1101
c04fecb4 1102void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
e7fd4179 1103{
c04fecb4
DT		 1104 struct dlm_rsb *r = NULL;
1105 uint32_t hash, b;
1106 int error;
e7fd4179 1107
c04fecb4
DT		 1108 hash = jhash(name, len, 0);
1109 b = hash & (ls->ls_rsbtbl_size - 1);
1110
1111 spin_lock(&ls->ls_rsbtbl[b].lock);
1112 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1113 if (!error)
1114 goto out_dump;
1115
1116 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1117 if (error)
1118 goto out;
1119 out_dump:
1120 dlm_dump_rsb(r);
1121 out:
1122 spin_unlock(&ls->ls_rsbtbl[b].lock);
e7fd4179
DT		 1123}
1124
1125static void toss_rsb(struct kref *kref)
1126{
1127 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1128 struct dlm_ls *ls = r->res_ls;
1129
1130 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1131 kref_init(&r->res_ref);
9beb3bf5
BP		 1132 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1133 rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
e7fd4179 1134 r->res_toss_time = jiffies;
f1172283 1135 ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
e7fd4179 1136 if (r->res_lvbptr) {
52bda2b5 1137 dlm_free_lvb(r->res_lvbptr);
e7fd4179
DT		 1138 r->res_lvbptr = NULL;
1139 }
1140}
1141
e7fd4179
DT		 1142/* See comment for unhold_lkb */
1143
1144static void unhold_rsb(struct dlm_rsb *r)
1145{
1146 int rv;
1147 rv = kref_put(&r->res_ref, toss_rsb);
a345da3e 1148 DLM_ASSERT(!rv, dlm_dump_rsb(r););
e7fd4179
DT		 1149}
1150
1151static void kill_rsb(struct kref *kref)
1152{
1153 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1154
1155 /* All work is done after the return from kref_put() so we
1156 can release the write_lock before the remove and free. */
1157
a345da3e
DT		 1158 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1159 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1160 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1161 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1162 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1163 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
e7fd4179
DT		 1164}
1165
1166/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1167 The rsb must exist as long as any lkb's for it do. */
1168
1169static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1170{
1171 hold_rsb(r);
1172 lkb->lkb_resource = r;
1173}
1174
1175static void detach_lkb(struct dlm_lkb *lkb)
1176{
1177 if (lkb->lkb_resource) {
1178 put_rsb(lkb->lkb_resource);
1179 lkb->lkb_resource = NULL;
1180 }
1181}
1182
75d25ffe
AA		 1183static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret,
1184 int start, int end)
e7fd4179 1185{
3d6aa675 1186 struct dlm_lkb *lkb;
2a86b3e7 1187 int rv;
e7fd4179 1188
52bda2b5 1189 lkb = dlm_allocate_lkb(ls);
e7fd4179
DT		 1190 if (!lkb)
1191 return -ENOMEM;
1192
1193 lkb->lkb_nodeid = -1;
1194 lkb->lkb_grmode = DLM_LOCK_IV;
1195 kref_init(&lkb->lkb_ref);
34e22bed 1196 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
ef0c2bb0 1197 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
3ae1acf9 1198 INIT_LIST_HEAD(&lkb->lkb_time_list);
23e8e1aa
DT		 1199 INIT_LIST_HEAD(&lkb->lkb_cb_list);
1200 mutex_init(&lkb->lkb_cb_mutex);
1201 INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
e7fd4179 1202
2a86b3e7 1203 idr_preload(GFP_NOFS);
3d6aa675 1204 spin_lock(&ls->ls_lkbidr_spin);
75d25ffe 1205 rv = idr_alloc(&ls->ls_lkbidr, lkb, start, end, GFP_NOWAIT);
2a86b3e7
TH		 1206 if (rv >= 0)
1207 lkb->lkb_id = rv;
3d6aa675 1208 spin_unlock(&ls->ls_lkbidr_spin);
2a86b3e7 1209 idr_preload_end();
e7fd4179 1210
3d6aa675
DT		 1211 if (rv < 0) {
1212 log_error(ls, "create_lkb idr error %d", rv);
23851e97 1213 dlm_free_lkb(lkb);
3d6aa675 1214 return rv;
e7fd4179
DT		 1215 }
1216
e7fd4179
DT		 1217 *lkb_ret = lkb;
1218 return 0;
1219}
1220
75d25ffe
AA		 1221static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1222{
1223 return _create_lkb(ls, lkb_ret, 1, 0);
1224}
1225
e7fd4179
DT		 1226static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1227{
1228 struct dlm_lkb *lkb;
e7fd4179 1229
3d6aa675
DT		 1230 spin_lock(&ls->ls_lkbidr_spin);
1231 lkb = idr_find(&ls->ls_lkbidr, lkid);
e7fd4179
DT		 1232 if (lkb)
1233 kref_get(&lkb->lkb_ref);
3d6aa675 1234 spin_unlock(&ls->ls_lkbidr_spin);
e7fd4179
DT		 1235
1236 *lkb_ret = lkb;
1237 return lkb ? 0 : -ENOENT;
1238}
1239
1240static void kill_lkb(struct kref *kref)
1241{
1242 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1243
1244 /* All work is done after the return from kref_put() so we
1245 can release the write_lock before the detach_lkb */
1246
1247 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1248}
1249
b3f58d8f
DT		 1250/* __put_lkb() is used when an lkb may not have an rsb attached to
1251 it so we need to provide the lockspace explicitly */
1252
1253static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
e7fd4179 1254{
3d6aa675 1255 uint32_t lkid = lkb->lkb_id;
e7fd4179 1256
3d6aa675 1257 spin_lock(&ls->ls_lkbidr_spin);
e7fd4179 1258 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
3d6aa675
DT		 1259 idr_remove(&ls->ls_lkbidr, lkid);
1260 spin_unlock(&ls->ls_lkbidr_spin);
e7fd4179
DT		 1261
1262 detach_lkb(lkb);
1263
1264 /* for local/process lkbs, lvbptr points to caller's lksb */
1265 if (lkb->lkb_lvbptr && is_master_copy(lkb))
52bda2b5
DT		 1266 dlm_free_lvb(lkb->lkb_lvbptr);
1267 dlm_free_lkb(lkb);
e7fd4179
DT		 1268 return 1;
1269 } else {
3d6aa675 1270 spin_unlock(&ls->ls_lkbidr_spin);
e7fd4179
DT		 1271 return 0;
1272 }
1273}
1274
1275int dlm_put_lkb(struct dlm_lkb *lkb)
1276{
b3f58d8f
DT		 1277 struct dlm_ls *ls;
1278
1279 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1280 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1281
1282 ls = lkb->lkb_resource->res_ls;
1283 return __put_lkb(ls, lkb);
e7fd4179
DT		 1284}
1285
1286/* This is only called to add a reference when the code already holds
1287 a valid reference to the lkb, so there's no need for locking. */
1288
1289static inline void hold_lkb(struct dlm_lkb *lkb)
1290{
1291 kref_get(&lkb->lkb_ref);
1292}
1293
1294/* This is called when we need to remove a reference and are certain
1295 it's not the last ref. e.g. del_lkb is always called between a
1296 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1297 put_lkb would work fine, but would involve unnecessary locking */
1298
1299static inline void unhold_lkb(struct dlm_lkb *lkb)
1300{
1301 int rv;
1302 rv = kref_put(&lkb->lkb_ref, kill_lkb);
1303 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
1304}
1305
1306static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1307 int mode)
1308{
1309 struct dlm_lkb *lkb = NULL;
1310
1311 list_for_each_entry(lkb, head, lkb_statequeue)
1312 if (lkb->lkb_rqmode < mode)
1313 break;
1314
99fb19d4 1315 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
e7fd4179
DT		 1316}
1317
1318/* add/remove lkb to rsb's grant/convert/wait queue */
1319
1320static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1321{
1322 kref_get(&lkb->lkb_ref);
1323
1324 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1325
eeda418d
DT		 1326 lkb->lkb_timestamp = ktime_get();
1327
e7fd4179
DT		 1328 lkb->lkb_status = status;
1329
1330 switch (status) {
1331 case DLM_LKSTS_WAITING:
1332 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1333 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1334 else
1335 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1336 break;
1337 case DLM_LKSTS_GRANTED:
1338 /* convention says granted locks kept in order of grmode */
1339 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1340 lkb->lkb_grmode);
1341 break;
1342 case DLM_LKSTS_CONVERT:
1343 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1344 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1345 else
1346 list_add_tail(&lkb->lkb_statequeue,
1347 &r->res_convertqueue);
1348 break;
1349 default:
1350 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1351 }
1352}
1353
1354static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1355{
1356 lkb->lkb_status = 0;
1357 list_del(&lkb->lkb_statequeue);
1358 unhold_lkb(lkb);
1359}
1360
1361static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1362{
1363 hold_lkb(lkb);
1364 del_lkb(r, lkb);
1365 add_lkb(r, lkb, sts);
1366 unhold_lkb(lkb);
1367}
1368
ef0c2bb0
DT		 1369static int msg_reply_type(int mstype)
1370{
1371 switch (mstype) {
1372 case DLM_MSG_REQUEST:
1373 return DLM_MSG_REQUEST_REPLY;
1374 case DLM_MSG_CONVERT:
1375 return DLM_MSG_CONVERT_REPLY;
1376 case DLM_MSG_UNLOCK:
1377 return DLM_MSG_UNLOCK_REPLY;
1378 case DLM_MSG_CANCEL:
1379 return DLM_MSG_CANCEL_REPLY;
1380 case DLM_MSG_LOOKUP:
1381 return DLM_MSG_LOOKUP_REPLY;
1382 }
1383 return -1;
1384}
1385
c6ff669b
DT		 1386static int nodeid_warned(int nodeid, int num_nodes, int *warned)
1387{
1388 int i;
1389
1390 for (i = 0; i < num_nodes; i++) {
1391 if (!warned[i]) {
1392 warned[i] = nodeid;
1393 return 0;
1394 }
1395 if (warned[i] == nodeid)
1396 return 1;
1397 }
1398 return 0;
1399}
1400
1401void dlm_scan_waiters(struct dlm_ls *ls)
1402{
1403 struct dlm_lkb *lkb;
c6ff669b
DT		 1404 s64 us;
1405 s64 debug_maxus = 0;
1406 u32 debug_scanned = 0;
1407 u32 debug_expired = 0;
1408 int num_nodes = 0;
1409 int *warned = NULL;
1410
1411 if (!dlm_config.ci_waitwarn_us)
1412 return;
1413
1414 mutex_lock(&ls->ls_waiters_mutex);
1415
1416 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1f3a8e49 1417 if (!lkb->lkb_wait_time)
c6ff669b
DT		 1418 continue;
1419
1420 debug_scanned++;
1421
1422 us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time));
1423
1424 if (us < dlm_config.ci_waitwarn_us)
1425 continue;
1426
1f3a8e49 1427 lkb->lkb_wait_time = 0;
c6ff669b
DT		 1428
1429 debug_expired++;
1430 if (us > debug_maxus)
1431 debug_maxus = us;
1432
1433 if (!num_nodes) {
1434 num_nodes = ls->ls_num_nodes;
fbb10081 1435 warned = kcalloc(num_nodes, sizeof(int), GFP_KERNEL);
c6ff669b
DT		 1436 }
1437 if (!warned)
1438 continue;
1439 if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned))
1440 continue;
1441
1442 log_error(ls, "waitwarn %x %lld %d us check connection to "
1443 "node %d", lkb->lkb_id, (long long)us,
1444 dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid);
1445 }
1446 mutex_unlock(&ls->ls_waiters_mutex);
5d70828a 1447 kfree(warned);
c6ff669b
DT		 1448
1449 if (debug_expired)
1450 log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us",
1451 debug_scanned, debug_expired,
1452 dlm_config.ci_waitwarn_us, (long long)debug_maxus);
1453}
1454
e7fd4179
DT		 1455/* add/remove lkb from global waiters list of lkb's waiting for
1456 a reply from a remote node */
1457
c6ff669b 1458static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
e7fd4179
DT		 1459{
1460 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
ef0c2bb0 1461 int error = 0;
e7fd4179 1462
90135925 1463 mutex_lock(&ls->ls_waiters_mutex);
ef0c2bb0
DT		 1464
1465 if (is_overlap_unlock(lkb) ||
1466 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1467 error = -EINVAL;
1468 goto out;
1469 }
1470
1471 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1472 switch (mstype) {
1473 case DLM_MSG_UNLOCK:
1474 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1475 break;
1476 case DLM_MSG_CANCEL:
1477 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1478 break;
1479 default:
1480 error = -EBUSY;
1481 goto out;
1482 }
1483 lkb->lkb_wait_count++;
1484 hold_lkb(lkb);
1485
43279e53 1486 log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
ef0c2bb0
DT		 1487 lkb->lkb_id, lkb->lkb_wait_type, mstype,
1488 lkb->lkb_wait_count, lkb->lkb_flags);
e7fd4179
DT		 1489 goto out;
1490 }
ef0c2bb0
DT		 1491
1492 DLM_ASSERT(!lkb->lkb_wait_count,
1493 dlm_print_lkb(lkb);
1494 printk("wait_count %d\n", lkb->lkb_wait_count););
1495
1496 lkb->lkb_wait_count++;
e7fd4179 1497 lkb->lkb_wait_type = mstype;
c6ff669b
DT		 1498 lkb->lkb_wait_time = ktime_get();
1499 lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
ef0c2bb0 1500 hold_lkb(lkb);
e7fd4179
DT		 1501 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1502 out:
ef0c2bb0 1503 if (error)
43279e53 1504 log_error(ls, "addwait error %x %d flags %x %d %d %s",
ef0c2bb0
DT		 1505 lkb->lkb_id, error, lkb->lkb_flags, mstype,
1506 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
90135925 1507 mutex_unlock(&ls->ls_waiters_mutex);
ef0c2bb0 1508 return error;
e7fd4179
DT		 1509}
1510
b790c3b7
DT		 1511/* We clear the RESEND flag because we might be taking an lkb off the waiters
1512 list as part of process_requestqueue (e.g. a lookup that has an optimized
1513 request reply on the requestqueue) between dlm_recover_waiters_pre() which
1514 set RESEND and dlm_recover_waiters_post() */
1515
43279e53
DT		 1516static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1517 struct dlm_message *ms)
e7fd4179 1518{
ef0c2bb0
DT		 1519 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1520 int overlap_done = 0;
e7fd4179 1521
ef0c2bb0 1522 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
43279e53 1523 log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
ef0c2bb0
DT		 1524 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1525 overlap_done = 1;
1526 goto out_del;
e7fd4179 1527 }
ef0c2bb0
DT		 1528
1529 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
43279e53 1530 log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
ef0c2bb0
DT		 1531 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1532 overlap_done = 1;
1533 goto out_del;
1534 }
1535
43279e53
DT		 1536 /* Cancel state was preemptively cleared by a successful convert,
1537 see next comment, nothing to do. */
1538
1539 if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1540 (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1541 log_debug(ls, "remwait %x cancel_reply wait_type %d",
1542 lkb->lkb_id, lkb->lkb_wait_type);
1543 return -1;
1544 }
1545
1546 /* Remove for the convert reply, and premptively remove for the
1547 cancel reply. A convert has been granted while there's still
1548 an outstanding cancel on it (the cancel is moot and the result
1549 in the cancel reply should be 0). We preempt the cancel reply
1550 because the app gets the convert result and then can follow up
1551 with another op, like convert. This subsequent op would see the
1552 lingering state of the cancel and fail with -EBUSY. */
1553
1554 if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1555 (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1556 is_overlap_cancel(lkb) && ms && !ms->m_result) {
1557 log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1558 lkb->lkb_id);
1559 lkb->lkb_wait_type = 0;
1560 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1561 lkb->lkb_wait_count--;
1562 goto out_del;
1563 }
1564
ef0c2bb0
DT		 1565 /* N.B. type of reply may not always correspond to type of original
1566 msg due to lookup->request optimization, verify others? */
1567
1568 if (lkb->lkb_wait_type) {
1569 lkb->lkb_wait_type = 0;
1570 goto out_del;
1571 }
1572
6d40c4a7 1573 log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
3428785a
AA		 1574 lkb->lkb_id, ms ? le32_to_cpu(ms->m_header.h_nodeid) : 0,
1575 lkb->lkb_remid, mstype, lkb->lkb_flags);
ef0c2bb0
DT		 1576 return -1;
1577
1578 out_del:
1579 /* the force-unlock/cancel has completed and we haven't recvd a reply
1580 to the op that was in progress prior to the unlock/cancel; we
1581 give up on any reply to the earlier op. FIXME: not sure when/how
1582 this would happen */
1583
1584 if (overlap_done && lkb->lkb_wait_type) {
43279e53 1585 log_error(ls, "remwait error %x reply %d wait_type %d overlap",
ef0c2bb0
DT		 1586 lkb->lkb_id, mstype, lkb->lkb_wait_type);
1587 lkb->lkb_wait_count--;
1588 lkb->lkb_wait_type = 0;
1589 }
1590
1591 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1592
b790c3b7 1593 lkb->lkb_flags &= ~DLM_IFL_RESEND;
ef0c2bb0
DT		 1594 lkb->lkb_wait_count--;
1595 if (!lkb->lkb_wait_count)
1596 list_del_init(&lkb->lkb_wait_reply);
e7fd4179 1597 unhold_lkb(lkb);
ef0c2bb0 1598 return 0;
e7fd4179
DT		 1599}
1600
ef0c2bb0 1601static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
e7fd4179
DT		 1602{
1603 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1604 int error;
1605
90135925 1606 mutex_lock(&ls->ls_waiters_mutex);
43279e53 1607 error = _remove_from_waiters(lkb, mstype, NULL);
90135925 1608 mutex_unlock(&ls->ls_waiters_mutex);
e7fd4179
DT		 1609 return error;
1610}
1611
ef0c2bb0
DT		 1612/* Handles situations where we might be processing a "fake" or "stub" reply in
1613 which we can't try to take waiters_mutex again. */
1614
1615static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1616{
1617 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1618 int error;
1619
00e99ccd 1620 if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS))
ef0c2bb0 1621 mutex_lock(&ls->ls_waiters_mutex);
00e99ccd
AA		 1622 error = _remove_from_waiters(lkb, le32_to_cpu(ms->m_type), ms);
1623 if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS))
ef0c2bb0
DT		 1624 mutex_unlock(&ls->ls_waiters_mutex);
1625 return error;
1626}
1627
05c32f47 1628/* If there's an rsb for the same resource being removed, ensure
21d9ac1a
AA		 1629 * that the remove message is sent before the new lookup message.
1630 */
1631
1632#define DLM_WAIT_PENDING_COND(ls, r) \
1633 (ls->ls_remove_len && \
1634 !rsb_cmp(r, ls->ls_remove_name, \
1635 ls->ls_remove_len))
e7fd4179 1636
05c32f47 1637static void wait_pending_remove(struct dlm_rsb *r)
e7fd4179 1638{
05c32f47
DT		 1639 struct dlm_ls *ls = r->res_ls;
1640 restart:
1641 spin_lock(&ls->ls_remove_spin);
21d9ac1a 1642 if (DLM_WAIT_PENDING_COND(ls, r)) {
05c32f47 1643 log_debug(ls, "delay lookup for remove dir %d %s",
21d9ac1a 1644 r->res_dir_nodeid, r->res_name);
05c32f47 1645 spin_unlock(&ls->ls_remove_spin);
21d9ac1a 1646 wait_event(ls->ls_remove_wait, !DLM_WAIT_PENDING_COND(ls, r));
05c32f47
DT		 1647 goto restart;
1648 }
1649 spin_unlock(&ls->ls_remove_spin);
1650}
e7fd4179 1651
05c32f47
DT		 1652/*
1653 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1654 * read by other threads in wait_pending_remove. ls_remove_names
1655 * and ls_remove_lens are only used by the scan thread, so they do
1656 * not need protection.
1657 */
c04fecb4 1658
05c32f47
DT		 1659static void shrink_bucket(struct dlm_ls *ls, int b)
1660{
1661 struct rb_node *n, *next;
1662 struct dlm_rsb *r;
1663 char *name;
1664 int our_nodeid = dlm_our_nodeid();
1665 int remote_count = 0;
f1172283 1666 int need_shrink = 0;
05c32f47 1667 int i, len, rv;
c04fecb4 1668
05c32f47 1669 memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
c04fecb4 1670
05c32f47 1671 spin_lock(&ls->ls_rsbtbl[b].lock);
f1172283
DT		 1672
1673 if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1674 spin_unlock(&ls->ls_rsbtbl[b].lock);
1675 return;
1676 }
1677
05c32f47
DT		 1678 for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1679 next = rb_next(n);
1680 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1681
1682 /* If we're the directory record for this rsb, and
1683 we're not the master of it, then we need to wait
1684 for the master node to send us a dir remove for
1685 before removing the dir record. */
1686
1687 if (!dlm_no_directory(ls) &&
1688 (r->res_master_nodeid != our_nodeid) &&
1689 (dlm_dir_nodeid(r) == our_nodeid)) {
1690 continue;
e7fd4179
DT		 1691 }
1692
f1172283
DT		 1693 need_shrink = 1;
1694
05c32f47
DT		 1695 if (!time_after_eq(jiffies, r->res_toss_time +
1696 dlm_config.ci_toss_secs * HZ)) {
1697 continue;
e7fd4179
DT		 1698 }
1699
05c32f47
DT		 1700 if (!dlm_no_directory(ls) &&
1701 (r->res_master_nodeid == our_nodeid) &&
1702 (dlm_dir_nodeid(r) != our_nodeid)) {
e7fd4179 1703
c04fecb4
DT		 1704 /* We're the master of this rsb but we're not
1705 the directory record, so we need to tell the
1706 dir node to remove the dir record. */
1707
05c32f47
DT		 1708 ls->ls_remove_lens[remote_count] = r->res_length;
1709 memcpy(ls->ls_remove_names[remote_count], r->res_name,
1710 DLM_RESNAME_MAXLEN);
1711 remote_count++;
c04fecb4 1712
05c32f47
DT		 1713 if (remote_count >= DLM_REMOVE_NAMES_MAX)
1714 break;
1715 continue;
1716 }
1717
1718 if (!kref_put(&r->res_ref, kill_rsb)) {
e7fd4179 1719 log_error(ls, "tossed rsb in use %s", r->res_name);
05c32f47 1720 continue;
e7fd4179 1721 }
05c32f47
DT		 1722
1723 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1724 dlm_free_rsb(r);
e7fd4179 1725 }
f1172283
DT		 1726
1727 if (need_shrink)
1728 ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1729 else
1730 ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
05c32f47 1731 spin_unlock(&ls->ls_rsbtbl[b].lock);
e7fd4179 1732
05c32f47
DT		 1733 /*
1734 * While searching for rsb's to free, we found some that require
1735 * remote removal. We leave them in place and find them again here
1736 * so there is a very small gap between removing them from the toss
1737 * list and sending the removal. Keeping this gap small is
1738 * important to keep us (the master node) from being out of sync
1739 * with the remote dir node for very long.
1740 *
1741 * From the time the rsb is removed from toss until just after
1742 * send_remove, the rsb name is saved in ls_remove_name. A new
1743 * lookup checks this to ensure that a new lookup message for the
1744 * same resource name is not sent just before the remove message.
1745 */
1746
1747 for (i = 0; i < remote_count; i++) {
1748 name = ls->ls_remove_names[i];
1749 len = ls->ls_remove_lens[i];
1750
1751 spin_lock(&ls->ls_rsbtbl[b].lock);
1752 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1753 if (rv) {
1754 spin_unlock(&ls->ls_rsbtbl[b].lock);
1755 log_debug(ls, "remove_name not toss %s", name);
1756 continue;
1757 }
1758
1759 if (r->res_master_nodeid != our_nodeid) {
1760 spin_unlock(&ls->ls_rsbtbl[b].lock);
1761 log_debug(ls, "remove_name master %d dir %d our %d %s",
1762 r->res_master_nodeid, r->res_dir_nodeid,
1763 our_nodeid, name);
1764 continue;
1765 }
1766
1767 if (r->res_dir_nodeid == our_nodeid) {
1768 /* should never happen */
1769 spin_unlock(&ls->ls_rsbtbl[b].lock);
1770 log_error(ls, "remove_name dir %d master %d our %d %s",
1771 r->res_dir_nodeid, r->res_master_nodeid,
1772 our_nodeid, name);
1773 continue;
1774 }
1775
1776 if (!time_after_eq(jiffies, r->res_toss_time +
1777 dlm_config.ci_toss_secs * HZ)) {
1778 spin_unlock(&ls->ls_rsbtbl[b].lock);
1779 log_debug(ls, "remove_name toss_time %lu now %lu %s",
1780 r->res_toss_time, jiffies, name);
1781 continue;
1782 }
1783
1784 if (!kref_put(&r->res_ref, kill_rsb)) {
1785 spin_unlock(&ls->ls_rsbtbl[b].lock);
1786 log_error(ls, "remove_name in use %s", name);
1787 continue;
1788 }
1789
1790 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1791
1792 /* block lookup of same name until we've sent remove */
1793 spin_lock(&ls->ls_remove_spin);
1794 ls->ls_remove_len = len;
1795 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1796 spin_unlock(&ls->ls_remove_spin);
1797 spin_unlock(&ls->ls_rsbtbl[b].lock);
21d9ac1a 1798 wake_up(&ls->ls_remove_wait);
05c32f47
DT		 1799
1800 send_remove(r);
1801
1802 /* allow lookup of name again */
1803 spin_lock(&ls->ls_remove_spin);
1804 ls->ls_remove_len = 0;
1805 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1806 spin_unlock(&ls->ls_remove_spin);
1807
1808 dlm_free_rsb(r);
1809 }
e7fd4179
DT		 1810}
1811
1812void dlm_scan_rsbs(struct dlm_ls *ls)
1813{
1814 int i;
1815
e7fd4179
DT		 1816 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1817 shrink_bucket(ls, i);
85e86edf
DT		 1818 if (dlm_locking_stopped(ls))
1819 break;
e7fd4179
DT		 1820 cond_resched();
1821 }
1822}
1823
3ae1acf9
DT		 1824static void add_timeout(struct dlm_lkb *lkb)
1825{
1826 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1827
eeda418d 1828 if (is_master_copy(lkb))
3ae1acf9 1829 return;
3ae1acf9
DT		 1830
1831 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1832 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1833 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1834 goto add_it;
1835 }
84d8cd69
DT		 1836 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1837 goto add_it;
3ae1acf9
DT		 1838 return;
1839
1840 add_it:
1841 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1842 mutex_lock(&ls->ls_timeout_mutex);
1843 hold_lkb(lkb);
3ae1acf9
DT		 1844 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1845 mutex_unlock(&ls->ls_timeout_mutex);
1846}
1847
1848static void del_timeout(struct dlm_lkb *lkb)
1849{
1850 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1851
1852 mutex_lock(&ls->ls_timeout_mutex);
1853 if (!list_empty(&lkb->lkb_time_list)) {
1854 list_del_init(&lkb->lkb_time_list);
1855 unhold_lkb(lkb);
1856 }
1857 mutex_unlock(&ls->ls_timeout_mutex);
1858}
1859
1860/* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1861 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1862 and then lock rsb because of lock ordering in add_timeout. We may need
1863 to specify some special timeout-related bits in the lkb that are just to
1864 be accessed under the timeout_mutex. */
1865
1866void dlm_scan_timeout(struct dlm_ls *ls)
1867{
1868 struct dlm_rsb *r;
1869 struct dlm_lkb *lkb;
1870 int do_cancel, do_warn;
eeda418d 1871 s64 wait_us;
3ae1acf9
DT		 1872
1873 for (;;) {
1874 if (dlm_locking_stopped(ls))
1875 break;
1876
1877 do_cancel = 0;
1878 do_warn = 0;
1879 mutex_lock(&ls->ls_timeout_mutex);
1880 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1881
eeda418d
DT		 1882 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1883 lkb->lkb_timestamp));
1884
3ae1acf9 1885 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
eeda418d 1886 wait_us >= (lkb->lkb_timeout_cs * 10000))
3ae1acf9
DT		 1887 do_cancel = 1;
1888
1889 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
eeda418d 1890 wait_us >= dlm_config.ci_timewarn_cs * 10000)
3ae1acf9
DT		 1891 do_warn = 1;
1892
1893 if (!do_cancel && !do_warn)
1894 continue;
1895 hold_lkb(lkb);
1896 break;
1897 }
1898 mutex_unlock(&ls->ls_timeout_mutex);
1899
1900 if (!do_cancel && !do_warn)
1901 break;
1902
1903 r = lkb->lkb_resource;
1904 hold_rsb(r);
1905 lock_rsb(r);
1906
1907 if (do_warn) {
1908 /* clear flag so we only warn once */
1909 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1910 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1911 del_timeout(lkb);
1912 dlm_timeout_warn(lkb);
1913 }
1914
1915 if (do_cancel) {
b3cab7b9 1916 log_debug(ls, "timeout cancel %x node %d %s",
639aca41 1917 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
3ae1acf9
DT		 1918 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1919 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1920 del_timeout(lkb);
1921 _cancel_lock(r, lkb);
1922 }
1923
1924 unlock_rsb(r);
1925 unhold_rsb(r);
1926 dlm_put_lkb(lkb);
1927 }
1928}
1929
1930/* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1931 dlm_recoverd before checking/setting ls_recover_begin. */
1932
1933void dlm_adjust_timeouts(struct dlm_ls *ls)
1934{
1935 struct dlm_lkb *lkb;
eeda418d 1936 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
3ae1acf9
DT		 1937
1938 ls->ls_recover_begin = 0;
1939 mutex_lock(&ls->ls_timeout_mutex);
1940 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
eeda418d 1941 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
3ae1acf9 1942 mutex_unlock(&ls->ls_timeout_mutex);
c6ff669b
DT		 1943
1944 if (!dlm_config.ci_waitwarn_us)
1945 return;
1946
1947 mutex_lock(&ls->ls_waiters_mutex);
1948 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1949 if (ktime_to_us(lkb->lkb_wait_time))
1950 lkb->lkb_wait_time = ktime_get();
1951 }
1952 mutex_unlock(&ls->ls_waiters_mutex);
3ae1acf9
DT		 1953}
1954
e7fd4179
DT		 1955/* lkb is master or local copy */
1956
1957static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1958{
1959 int b, len = r->res_ls->ls_lvblen;
1960
1961 /* b=1 lvb returned to caller
1962 b=0 lvb written to rsb or invalidated
1963 b=-1 do nothing */
1964
1965 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1966
1967 if (b == 1) {
1968 if (!lkb->lkb_lvbptr)
1969 return;
1970
1971 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1972 return;
1973
1974 if (!r->res_lvbptr)
1975 return;
1976
1977 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1978 lkb->lkb_lvbseq = r->res_lvbseq;
1979
1980 } else if (b == 0) {
1981 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1982 rsb_set_flag(r, RSB_VALNOTVALID);
1983 return;
1984 }
1985
1986 if (!lkb->lkb_lvbptr)
1987 return;
1988
1989 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1990 return;
1991
1992 if (!r->res_lvbptr)
52bda2b5 1993 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
e7fd4179
DT		 1994
1995 if (!r->res_lvbptr)
1996 return;
1997
1998 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1999 r->res_lvbseq++;
2000 lkb->lkb_lvbseq = r->res_lvbseq;
2001 rsb_clear_flag(r, RSB_VALNOTVALID);
2002 }
2003
2004 if (rsb_flag(r, RSB_VALNOTVALID))
2005 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
2006}
2007
2008static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2009{
2010 if (lkb->lkb_grmode < DLM_LOCK_PW)
2011 return;
2012
2013 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
2014 rsb_set_flag(r, RSB_VALNOTVALID);
2015 return;
2016 }
2017
2018 if (!lkb->lkb_lvbptr)
2019 return;
2020
2021 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2022 return;
2023
2024 if (!r->res_lvbptr)
52bda2b5 2025 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
e7fd4179
DT		 2026
2027 if (!r->res_lvbptr)
2028 return;
2029
2030 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2031 r->res_lvbseq++;
2032 rsb_clear_flag(r, RSB_VALNOTVALID);
2033}
2034
2035/* lkb is process copy (pc) */
2036
2037static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2038 struct dlm_message *ms)
2039{
2040 int b;
2041
2042 if (!lkb->lkb_lvbptr)
2043 return;
2044
2045 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2046 return;
2047
597d0cae 2048 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
e7fd4179
DT		 2049 if (b == 1) {
2050 int len = receive_extralen(ms);
cfa805f6
BVA		 2051 if (len > r->res_ls->ls_lvblen)
2052 len = r->res_ls->ls_lvblen;
e7fd4179 2053 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
00e99ccd 2054 lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
e7fd4179
DT		 2055 }
2056}
2057
2058/* Manipulate lkb's on rsb's convert/granted/waiting queues
2059 remove_lock -- used for unlock, removes lkb from granted
2060 revert_lock -- used for cancel, moves lkb from convert to granted
2061 grant_lock -- used for request and convert, adds lkb to granted or
2062 moves lkb from convert or waiting to granted
2063
2064 Each of these is used for master or local copy lkb's. There is
2065 also a _pc() variation used to make the corresponding change on
2066 a process copy (pc) lkb. */
2067
2068static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2069{
2070 del_lkb(r, lkb);
2071 lkb->lkb_grmode = DLM_LOCK_IV;
2072 /* this unhold undoes the original ref from create_lkb()
2073 so this leads to the lkb being freed */
2074 unhold_lkb(lkb);
2075}
2076
2077static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2078{
2079 set_lvb_unlock(r, lkb);
2080 _remove_lock(r, lkb);
2081}
2082
2083static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2084{
2085 _remove_lock(r, lkb);
2086}
2087
ef0c2bb0
DT		 2088/* returns: 0 did nothing
2089 1 moved lock to granted
2090 -1 removed lock */
2091
2092static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
e7fd4179 2093{
ef0c2bb0
DT		 2094 int rv = 0;
2095
e7fd4179
DT		 2096 lkb->lkb_rqmode = DLM_LOCK_IV;
2097
2098 switch (lkb->lkb_status) {
597d0cae
DT		 2099 case DLM_LKSTS_GRANTED:
2100 break;
e7fd4179
DT		 2101 case DLM_LKSTS_CONVERT:
2102 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
ef0c2bb0 2103 rv = 1;
e7fd4179
DT		 2104 break;
2105 case DLM_LKSTS_WAITING:
2106 del_lkb(r, lkb);
2107 lkb->lkb_grmode = DLM_LOCK_IV;
2108 /* this unhold undoes the original ref from create_lkb()
2109 so this leads to the lkb being freed */
2110 unhold_lkb(lkb);
ef0c2bb0 2111 rv = -1;
e7fd4179
DT		 2112 break;
2113 default:
2114 log_print("invalid status for revert %d", lkb->lkb_status);
2115 }
ef0c2bb0 2116 return rv;
e7fd4179
DT		 2117}
2118
ef0c2bb0 2119static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
e7fd4179 2120{
ef0c2bb0 2121 return revert_lock(r, lkb);
e7fd4179
DT		 2122}
2123
2124static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2125{
2126 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2127 lkb->lkb_grmode = lkb->lkb_rqmode;
2128 if (lkb->lkb_status)
2129 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2130 else
2131 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2132 }
2133
2134 lkb->lkb_rqmode = DLM_LOCK_IV;
4875647a 2135 lkb->lkb_highbast = 0;
e7fd4179
DT		 2136}
2137
2138static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2139{
2140 set_lvb_lock(r, lkb);
2141 _grant_lock(r, lkb);
e7fd4179
DT		 2142}
2143
2144static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2145 struct dlm_message *ms)
2146{
2147 set_lvb_lock_pc(r, lkb, ms);
2148 _grant_lock(r, lkb);
2149}
2150
2151/* called by grant_pending_locks() which means an async grant message must
2152 be sent to the requesting node in addition to granting the lock if the
2153 lkb belongs to a remote node. */
2154
2155static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2156{
2157 grant_lock(r, lkb);
2158 if (is_master_copy(lkb))
2159 send_grant(r, lkb);
2160 else
2161 queue_cast(r, lkb, 0);
2162}
2163
7d3c1feb
DT		 2164/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2165 change the granted/requested modes. We're munging things accordingly in
2166 the process copy.
2167 CONVDEADLK: our grmode may have been forced down to NL to resolve a
2168 conversion deadlock
2169 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2170 compatible with other granted locks */
2171
2a7ce0ed 2172static void munge_demoted(struct dlm_lkb *lkb)
7d3c1feb 2173{
7d3c1feb
DT		 2174 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2175 log_print("munge_demoted %x invalid modes gr %d rq %d",
2176 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2177 return;
2178 }
2179
2180 lkb->lkb_grmode = DLM_LOCK_NL;
2181}
2182
2183static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2184{
00e99ccd
AA		 2185 if (ms->m_type != cpu_to_le32(DLM_MSG_REQUEST_REPLY) &&
2186 ms->m_type != cpu_to_le32(DLM_MSG_GRANT)) {
7d3c1feb 2187 log_print("munge_altmode %x invalid reply type %d",
00e99ccd 2188 lkb->lkb_id, le32_to_cpu(ms->m_type));
7d3c1feb
DT		 2189 return;
2190 }
2191
2192 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2193 lkb->lkb_rqmode = DLM_LOCK_PR;
2194 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2195 lkb->lkb_rqmode = DLM_LOCK_CW;
2196 else {
2197 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2198 dlm_print_lkb(lkb);
2199 }
2200}
2201
e7fd4179
DT		 2202static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2203{
2204 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2205 lkb_statequeue);
2206 if (lkb->lkb_id == first->lkb_id)
90135925 2207 return 1;
e7fd4179 2208
90135925 2209 return 0;
e7fd4179
DT		 2210}
2211
e7fd4179
DT		 2212/* Check if the given lkb conflicts with another lkb on the queue. */
2213
2214static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2215{
2216 struct dlm_lkb *this;
2217
2218 list_for_each_entry(this, head, lkb_statequeue) {
2219 if (this == lkb)
2220 continue;
3bcd3687 2221 if (!modes_compat(this, lkb))
90135925 2222 return 1;
e7fd4179 2223 }
90135925 2224 return 0;
e7fd4179
DT		 2225}
2226
2227/*
2228 * "A conversion deadlock arises with a pair of lock requests in the converting
2229 * queue for one resource. The granted mode of each lock blocks the requested
2230 * mode of the other lock."
2231 *
c85d65e9
DT		 2232 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2233 * convert queue from being granted, then deadlk/demote lkb.
e7fd4179
DT		 2234 *
2235 * Example:
2236 * Granted Queue: empty
2237 * Convert Queue: NL->EX (first lock)
2238 * PR->EX (second lock)
2239 *
2240 * The first lock can't be granted because of the granted mode of the second
2241 * lock and the second lock can't be granted because it's not first in the
c85d65e9
DT		 2242 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2243 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2244 * flag set and return DEMOTED in the lksb flags.
e7fd4179 2245 *
c85d65e9
DT		 2246 * Originally, this function detected conv-deadlk in a more limited scope:
2247 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2248 * - if lkb1 was the first entry in the queue (not just earlier), and was
2249 * blocked by the granted mode of lkb2, and there was nothing on the
2250 * granted queue preventing lkb1 from being granted immediately, i.e.
2251 * lkb2 was the only thing preventing lkb1 from being granted.
2252 *
2253 * That second condition meant we'd only say there was conv-deadlk if
2254 * resolving it (by demotion) would lead to the first lock on the convert
2255 * queue being granted right away. It allowed conversion deadlocks to exist
2256 * between locks on the convert queue while they couldn't be granted anyway.
2257 *
2258 * Now, we detect and take action on conversion deadlocks immediately when
2259 * they're created, even if they may not be immediately consequential. If
2260 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2261 * mode that would prevent lkb1's conversion from being granted, we do a
2262 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2263 * I think this means that the lkb_is_ahead condition below should always
2264 * be zero, i.e. there will never be conv-deadlk between two locks that are
2265 * both already on the convert queue.
e7fd4179
DT		 2266 */
2267
c85d65e9 2268static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
e7fd4179 2269{
c85d65e9
DT		 2270 struct dlm_lkb *lkb1;
2271 int lkb_is_ahead = 0;
e7fd4179 2272
c85d65e9
DT		 2273 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2274 if (lkb1 == lkb2) {
2275 lkb_is_ahead = 1;
e7fd4179
DT		 2276 continue;
2277 }
2278
c85d65e9
DT		 2279 if (!lkb_is_ahead) {
2280 if (!modes_compat(lkb2, lkb1))
2281 return 1;
2282 } else {
2283 if (!modes_compat(lkb2, lkb1) &&
2284 !modes_compat(lkb1, lkb2))
2285 return 1;
2286 }
e7fd4179 2287 }
90135925 2288 return 0;
e7fd4179
DT		 2289}
2290
2291/*
2292 * Return 1 if the lock can be granted, 0 otherwise.
2293 * Also detect and resolve conversion deadlocks.
2294 *
2295 * lkb is the lock to be granted
2296 *
2297 * now is 1 if the function is being called in the context of the
2298 * immediate request, it is 0 if called later, after the lock has been
2299 * queued.
2300 *
c503a621
DT		 2301 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2302 * after recovery.
2303 *
e7fd4179
DT		 2304 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2305 */
2306
c503a621
DT		 2307static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2308 int recover)
e7fd4179
DT		 2309{
2310 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2311
2312 /*
2313 * 6-10: Version 5.4 introduced an option to address the phenomenon of
2314 * a new request for a NL mode lock being blocked.
2315 *
2316 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2317 * request, then it would be granted. In essence, the use of this flag
2318 * tells the Lock Manager to expedite theis request by not considering
2319 * what may be in the CONVERTING or WAITING queues... As of this
2320 * writing, the EXPEDITE flag can be used only with new requests for NL
2321 * mode locks. This flag is not valid for conversion requests.
2322 *
2323 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
2324 * conversion or used with a non-NL requested mode. We also know an
2325 * EXPEDITE request is always granted immediately, so now must always
2326 * be 1. The full condition to grant an expedite request: (now &&
2327 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2328 * therefore be shortened to just checking the flag.
2329 */
2330
2331 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
90135925 2332 return 1;
e7fd4179
DT		 2333
2334 /*
2335 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2336 * added to the remaining conditions.
2337 */
2338
2339 if (queue_conflict(&r->res_grantqueue, lkb))
c503a621 2340 return 0;
e7fd4179
DT		 2341
2342 /*
2343 * 6-3: By default, a conversion request is immediately granted if the
2344 * requested mode is compatible with the modes of all other granted
2345 * locks
2346 */
2347
2348 if (queue_conflict(&r->res_convertqueue, lkb))
c503a621
DT		 2349 return 0;
2350
2351 /*
2352 * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2353 * locks for a recovered rsb, on which lkb's have been rebuilt.
2354 * The lkb's may have been rebuilt on the queues in a different
2355 * order than they were in on the previous master. So, granting
2356 * queued conversions in order after recovery doesn't make sense
2357 * since the order hasn't been preserved anyway. The new order
2358 * could also have created a new "in place" conversion deadlock.
2359 * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2360 * After recovery, there would be no granted locks, and possibly
2361 * NL->EX, PR->EX, an in-place conversion deadlock.) So, after
2362 * recovery, grant conversions without considering order.
2363 */
2364
2365 if (conv && recover)
2366 return 1;
e7fd4179
DT		 2367
2368 /*
2369 * 6-5: But the default algorithm for deciding whether to grant or
2370 * queue conversion requests does not by itself guarantee that such
2371 * requests are serviced on a "first come first serve" basis. This, in
2372 * turn, can lead to a phenomenon known as "indefinate postponement".
2373 *
2374 * 6-7: This issue is dealt with by using the optional QUECVT flag with
2375 * the system service employed to request a lock conversion. This flag
2376 * forces certain conversion requests to be queued, even if they are
2377 * compatible with the granted modes of other locks on the same
2378 * resource. Thus, the use of this flag results in conversion requests
2379 * being ordered on a "first come first servce" basis.
2380 *
2381 * DCT: This condition is all about new conversions being able to occur
2382 * "in place" while the lock remains on the granted queue (assuming
2383 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
2384 * doesn't _have_ to go onto the convert queue where it's processed in
2385 * order. The "now" variable is necessary to distinguish converts
2386 * being received and processed for the first time now, because once a
2387 * convert is moved to the conversion queue the condition below applies
2388 * requiring fifo granting.
2389 */
2390
2391 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
90135925 2392 return 1;
e7fd4179 2393
53ad1c98
DT		 2394 /*
2395 * Even if the convert is compat with all granted locks,
2396 * QUECVT forces it behind other locks on the convert queue.
2397 */
2398
2399 if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2400 if (list_empty(&r->res_convertqueue))
2401 return 1;
2402 else
c503a621 2403 return 0;
53ad1c98
DT		 2404 }
2405
e7fd4179 2406 /*
3bcd3687
DT		 2407 * The NOORDER flag is set to avoid the standard vms rules on grant
2408 * order.
e7fd4179
DT		 2409 */
2410
2411 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
90135925 2412 return 1;
e7fd4179
DT		 2413
2414 /*
2415 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2416 * granted until all other conversion requests ahead of it are granted
2417 * and/or canceled.
2418 */
2419
2420 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
90135925 2421 return 1;
e7fd4179
DT		 2422
2423 /*
2424 * 6-4: By default, a new request is immediately granted only if all
2425 * three of the following conditions are satisfied when the request is
2426 * issued:
2427 * - The queue of ungranted conversion requests for the resource is
2428 * empty.
2429 * - The queue of ungranted new requests for the resource is empty.
2430 * - The mode of the new request is compatible with the most
2431 * restrictive mode of all granted locks on the resource.
2432 */
2433
2434 if (now && !conv && list_empty(&r->res_convertqueue) &&
2435 list_empty(&r->res_waitqueue))
90135925 2436 return 1;
e7fd4179
DT		 2437
2438 /*
2439 * 6-4: Once a lock request is in the queue of ungranted new requests,
2440 * it cannot be granted until the queue of ungranted conversion
2441 * requests is empty, all ungranted new requests ahead of it are
2442 * granted and/or canceled, and it is compatible with the granted mode
2443 * of the most restrictive lock granted on the resource.
2444 */
2445
2446 if (!now && !conv && list_empty(&r->res_convertqueue) &&
2447 first_in_list(lkb, &r->res_waitqueue))
90135925 2448 return 1;
c503a621 2449
90135925 2450 return 0;
e7fd4179
DT		 2451}
2452
c85d65e9 2453static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
c503a621 2454 int recover, int *err)
e7fd4179 2455{
e7fd4179
DT		 2456 int rv;
2457 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
c85d65e9
DT		 2458 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2459
2460 if (err)
2461 *err = 0;
e7fd4179 2462
c503a621 2463 rv = _can_be_granted(r, lkb, now, recover);
e7fd4179
DT		 2464 if (rv)
2465 goto out;
2466
c85d65e9
DT		 2467 /*
2468 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2469 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2470 * cancels one of the locks.
2471 */
2472
2473 if (is_convert && can_be_queued(lkb) &&
2474 conversion_deadlock_detect(r, lkb)) {
2475 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2476 lkb->lkb_grmode = DLM_LOCK_NL;
2477 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
294e7e45 2478 } else if (err) {
2479 *err = -EDEADLK;
2480 } else {
2481 log_print("can_be_granted deadlock %x now %d",
2482 lkb->lkb_id, now);
2483 dlm_dump_rsb(r);
c85d65e9 2484 }
e7fd4179 2485 goto out;
c85d65e9 2486 }
e7fd4179 2487
c85d65e9
DT		 2488 /*
2489 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2490 * to grant a request in a mode other than the normal rqmode. It's a
2491 * simple way to provide a big optimization to applications that can
2492 * use them.
2493 */
2494
2495 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
e7fd4179 2496 alt = DLM_LOCK_PR;
c85d65e9 2497 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
e7fd4179
DT		 2498 alt = DLM_LOCK_CW;
2499
2500 if (alt) {
2501 lkb->lkb_rqmode = alt;
c503a621 2502 rv = _can_be_granted(r, lkb, now, 0);
e7fd4179
DT		 2503 if (rv)
2504 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2505 else
2506 lkb->lkb_rqmode = rqmode;
2507 }
2508 out:
2509 return rv;
2510}
2511
36509258
DT		 2512/* Returns the highest requested mode of all blocked conversions; sets
2513 cw if there's a blocked conversion to DLM_LOCK_CW. */
c85d65e9 2514
4875647a
DT		 2515static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2516 unsigned int *count)
e7fd4179
DT		 2517{
2518 struct dlm_lkb *lkb, *s;
c503a621 2519 int recover = rsb_flag(r, RSB_RECOVER_GRANT);
e7fd4179 2520 int hi, demoted, quit, grant_restart, demote_restart;
c85d65e9 2521 int deadlk;
e7fd4179
DT		 2522
2523 quit = 0;
2524 restart:
2525 grant_restart = 0;
2526 demote_restart = 0;
2527 hi = DLM_LOCK_IV;
2528
2529 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2530 demoted = is_demoted(lkb);
c85d65e9
DT		 2531 deadlk = 0;
2532
c503a621 2533 if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
e7fd4179
DT		 2534 grant_lock_pending(r, lkb);
2535 grant_restart = 1;
4875647a
DT		 2536 if (count)
2537 (*count)++;
c85d65e9 2538 continue;
e7fd4179 2539 }
c85d65e9
DT		 2540
2541 if (!demoted && is_demoted(lkb)) {
2542 log_print("WARN: pending demoted %x node %d %s",
2543 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2544 demote_restart = 1;
2545 continue;
2546 }
2547
2548 if (deadlk) {
294e7e45 2549 /*
2550 * If DLM_LKB_NODLKWT flag is set and conversion
2551 * deadlock is detected, we request blocking AST and
2552 * down (or cancel) conversion.
2553 */
2554 if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
2555 if (lkb->lkb_highbast < lkb->lkb_rqmode) {
2556 queue_bast(r, lkb, lkb->lkb_rqmode);
2557 lkb->lkb_highbast = lkb->lkb_rqmode;
2558 }
2559 } else {
2560 log_print("WARN: pending deadlock %x node %d %s",
2561 lkb->lkb_id, lkb->lkb_nodeid,
2562 r->res_name);
2563 dlm_dump_rsb(r);
2564 }
c85d65e9
DT		 2565 continue;
2566 }
2567
2568 hi = max_t(int, lkb->lkb_rqmode, hi);
36509258
DT		 2569
2570 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2571 *cw = 1;
e7fd4179
DT		 2572 }
2573
2574 if (grant_restart)
2575 goto restart;
2576 if (demote_restart && !quit) {
2577 quit = 1;
2578 goto restart;
2579 }
2580
2581 return max_t(int, high, hi);
2582}
2583
4875647a
DT		 2584static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2585 unsigned int *count)
e7fd4179
DT		 2586{
2587 struct dlm_lkb *lkb, *s;
2588
2589 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
c503a621 2590 if (can_be_granted(r, lkb, 0, 0, NULL)) {
e7fd4179 2591 grant_lock_pending(r, lkb);
4875647a
DT		 2592 if (count)
2593 (*count)++;
2594 } else {
e7fd4179 2595 high = max_t(int, lkb->lkb_rqmode, high);
36509258
DT		 2596 if (lkb->lkb_rqmode == DLM_LOCK_CW)
2597 *cw = 1;
2598 }
e7fd4179
DT		 2599 }
2600
2601 return high;
2602}
2603
36509258
DT		 2604/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2605 on either the convert or waiting queue.
2606 high is the largest rqmode of all locks blocked on the convert or
2607 waiting queue. */
2608
2609static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2610{
2611 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2612 if (gr->lkb_highbast < DLM_LOCK_EX)
2613 return 1;
2614 return 0;
2615 }
2616
2617 if (gr->lkb_highbast < high &&
2618 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2619 return 1;
2620 return 0;
2621}
2622
4875647a 2623static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
e7fd4179
DT		 2624{
2625 struct dlm_lkb *lkb, *s;
2626 int high = DLM_LOCK_IV;
36509258 2627 int cw = 0;
e7fd4179 2628
4875647a
DT		 2629 if (!is_master(r)) {
2630 log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2631 dlm_dump_rsb(r);
2632 return;
2633 }
e7fd4179 2634
4875647a
DT		 2635 high = grant_pending_convert(r, high, &cw, count);
2636 high = grant_pending_wait(r, high, &cw, count);
e7fd4179
DT		 2637
2638 if (high == DLM_LOCK_IV)
2639 return;
2640
2641 /*
2642 * If there are locks left on the wait/convert queue then send blocking
2643 * ASTs to granted locks based on the largest requested mode (high)
36509258 2644 * found above.
e7fd4179
DT		 2645 */
2646
2647 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
e5dae548 2648 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
329fc4c3
DT		 2649 if (cw && high == DLM_LOCK_PR &&
2650 lkb->lkb_grmode == DLM_LOCK_PR)
36509258
DT		 2651 queue_bast(r, lkb, DLM_LOCK_CW);
2652 else
2653 queue_bast(r, lkb, high);
e7fd4179
DT		 2654 lkb->lkb_highbast = high;
2655 }
2656 }
2657}
2658
36509258
DT		 2659static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2660{
2661 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2662 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2663 if (gr->lkb_highbast < DLM_LOCK_EX)
2664 return 1;
2665 return 0;
2666 }
2667
2668 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2669 return 1;
2670 return 0;
2671}
2672
e7fd4179
DT		 2673static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2674 struct dlm_lkb *lkb)
2675{
2676 struct dlm_lkb *gr;
2677
2678 list_for_each_entry(gr, head, lkb_statequeue) {
314dd2a0
SW		 2679 /* skip self when sending basts to convertqueue */
2680 if (gr == lkb)
2681 continue;
e5dae548 2682 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
e7fd4179
DT		 2683 queue_bast(r, gr, lkb->lkb_rqmode);
2684 gr->lkb_highbast = lkb->lkb_rqmode;
2685 }
2686 }
2687}
2688
2689static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2690{
2691 send_bast_queue(r, &r->res_grantqueue, lkb);
2692}
2693
2694static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2695{
2696 send_bast_queue(r, &r->res_grantqueue, lkb);
2697 send_bast_queue(r, &r->res_convertqueue, lkb);
2698}
2699
2700/* set_master(r, lkb) -- set the master nodeid of a resource
2701
2702 The purpose of this function is to set the nodeid field in the given
2703 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2704 known, it can just be copied to the lkb and the function will return
2705 0. If the rsb's nodeid is _not_ known, it needs to be looked up
2706 before it can be copied to the lkb.
2707
2708 When the rsb nodeid is being looked up remotely, the initial lkb
2709 causing the lookup is kept on the ls_waiters list waiting for the
2710 lookup reply. Other lkb's waiting for the same rsb lookup are kept
2711 on the rsb's res_lookup list until the master is verified.
2712
2713 Return values:
2714 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2715 1: the rsb master is not available and the lkb has been placed on
2716 a wait queue
2717*/
2718
2719static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2720{
c04fecb4 2721 int our_nodeid = dlm_our_nodeid();
e7fd4179
DT		 2722
2723 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2724 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2725 r->res_first_lkid = lkb->lkb_id;
2726 lkb->lkb_nodeid = r->res_nodeid;
2727 return 0;
2728 }
2729
2730 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2731 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2732 return 1;
2733 }
2734
c04fecb4 2735 if (r->res_master_nodeid == our_nodeid) {
e7fd4179
DT		 2736 lkb->lkb_nodeid = 0;
2737 return 0;
2738 }
2739
c04fecb4
DT		 2740 if (r->res_master_nodeid) {
2741 lkb->lkb_nodeid = r->res_master_nodeid;
e7fd4179
DT		 2742 return 0;
2743 }
2744
c04fecb4
DT		 2745 if (dlm_dir_nodeid(r) == our_nodeid) {
2746 /* This is a somewhat unusual case; find_rsb will usually
2747 have set res_master_nodeid when dir nodeid is local, but
2748 there are cases where we become the dir node after we've
2749 past find_rsb and go through _request_lock again.
2750 confirm_master() or process_lookup_list() needs to be
2751 called after this. */
2752 log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2753 lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2754 r->res_name);
2755 r->res_master_nodeid = our_nodeid;
e7fd4179
DT		 2756 r->res_nodeid = 0;
2757 lkb->lkb_nodeid = 0;
c04fecb4 2758 return 0;
e7fd4179 2759 }
c04fecb4 2760
05c32f47
DT		 2761 wait_pending_remove(r);
2762
c04fecb4
DT		 2763 r->res_first_lkid = lkb->lkb_id;
2764 send_lookup(r, lkb);
2765 return 1;
e7fd4179
DT		 2766}
2767
2768static void process_lookup_list(struct dlm_rsb *r)
2769{
2770 struct dlm_lkb *lkb, *safe;
2771
2772 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
ef0c2bb0 2773 list_del_init(&lkb->lkb_rsb_lookup);
e7fd4179
DT		 2774 _request_lock(r, lkb);
2775 schedule();
2776 }
2777}
2778
2779/* confirm_master -- confirm (or deny) an rsb's master nodeid */
2780
2781static void confirm_master(struct dlm_rsb *r, int error)
2782{
2783 struct dlm_lkb *lkb;
2784
2785 if (!r->res_first_lkid)
2786 return;
2787
2788 switch (error) {
2789 case 0:
2790 case -EINPROGRESS:
2791 r->res_first_lkid = 0;
2792 process_lookup_list(r);
2793 break;
2794
2795 case -EAGAIN:
aec64e1b
DT		 2796 case -EBADR:
2797 case -ENOTBLK:
2798 /* the remote request failed and won't be retried (it was
2799 a NOQUEUE, or has been canceled/unlocked); make a waiting
2800 lkb the first_lkid */
e7fd4179
DT		 2801
2802 r->res_first_lkid = 0;
2803
2804 if (!list_empty(&r->res_lookup)) {
2805 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2806 lkb_rsb_lookup);
ef0c2bb0 2807 list_del_init(&lkb->lkb_rsb_lookup);
e7fd4179
DT		 2808 r->res_first_lkid = lkb->lkb_id;
2809 _request_lock(r, lkb);
761b9d3f 2810 }
e7fd4179
DT		 2811 break;
2812
2813 default:
2814 log_error(r->res_ls, "confirm_master unknown error %d", error);
2815 }
2816}
2817
2818static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
e5dae548
DT		 2819 int namelen, unsigned long timeout_cs,
2820 void (*ast) (void *astparam),
2821 void *astparam,
2822 void (*bast) (void *astparam, int mode),
2823 struct dlm_args *args)
e7fd4179
DT		 2824{
2825 int rv = -EINVAL;
2826
2827 /* check for invalid arg usage */
2828
2829 if (mode < 0 || mode > DLM_LOCK_EX)
2830 goto out;
2831
2832 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2833 goto out;
2834
2835 if (flags & DLM_LKF_CANCEL)
2836 goto out;
2837
2838 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2839 goto out;
2840
2841 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2842 goto out;
2843
2844 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2845 goto out;
2846
2847 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2848 goto out;
2849
2850 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2851 goto out;
2852
2853 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2854 goto out;
2855
2856 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2857 goto out;
2858
2859 if (!ast || !lksb)
2860 goto out;
2861
2862 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2863 goto out;
2864
e7fd4179
DT		 2865 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2866 goto out;
2867
2868 /* these args will be copied to the lkb in validate_lock_args,
2869 it cannot be done now because when converting locks, fields in
2870 an active lkb cannot be modified before locking the rsb */
2871
2872 args->flags = flags;
e5dae548
DT		 2873 args->astfn = ast;
2874 args->astparam = astparam;
2875 args->bastfn = bast;
d7db923e 2876 args->timeout = timeout_cs;
e7fd4179
DT		 2877 args->mode = mode;
2878 args->lksb = lksb;
e7fd4179
DT		 2879 rv = 0;
2880 out:
2881 return rv;
2882}
2883
2884static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2885{
2886 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2887 DLM_LKF_FORCEUNLOCK))
2888 return -EINVAL;
2889
ef0c2bb0
DT		 2890 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2891 return -EINVAL;
2892
e7fd4179 2893 args->flags = flags;
e5dae548 2894 args->astparam = astarg;
e7fd4179
DT		 2895 return 0;
2896}
2897
2898static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2899 struct dlm_args *args)
2900{
2901 int rv = -EINVAL;
2902
2903 if (args->flags & DLM_LKF_CONVERT) {
2904 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2905 goto out;
2906
2907 if (args->flags & DLM_LKF_QUECVT &&
2908 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2909 goto out;
2910
2911 rv = -EBUSY;
2912 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2913 goto out;
2914
67e4d8c5
AA		 2915 /* lock not allowed if there's any op in progress */
2916 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
e7fd4179 2917 goto out;
ef0c2bb0
DT		 2918
2919 if (is_overlap(lkb))
2920 goto out;
e7fd4179
DT		 2921 }
2922
2923 lkb->lkb_exflags = args->flags;
2924 lkb->lkb_sbflags = 0;
e5dae548 2925 lkb->lkb_astfn = args->astfn;
e7fd4179 2926 lkb->lkb_astparam = args->astparam;
e5dae548 2927 lkb->lkb_bastfn = args->bastfn;
e7fd4179
DT		 2928 lkb->lkb_rqmode = args->mode;
2929 lkb->lkb_lksb = args->lksb;
2930 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2931 lkb->lkb_ownpid = (int) current->pid;
d7db923e 2932 lkb->lkb_timeout_cs = args->timeout;
e7fd4179
DT		 2933 rv = 0;
2934 out:
43279e53
DT		 2935 if (rv)
2936 log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2937 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2938 lkb->lkb_status, lkb->lkb_wait_type,
2939 lkb->lkb_resource->res_name);
e7fd4179
DT		 2940 return rv;
2941}
2942
ef0c2bb0
DT		 2943/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2944 for success */
2945
2946/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2947 because there may be a lookup in progress and it's valid to do
2948 cancel/unlockf on it */
2949
e7fd4179
DT		 2950static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2951{
ef0c2bb0 2952 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
e7fd4179
DT		 2953 int rv = -EINVAL;
2954
ef0c2bb0
DT		 2955 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2956 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2957 dlm_print_lkb(lkb);
e7fd4179 2958 goto out;
ef0c2bb0 2959 }
e7fd4179 2960
ef0c2bb0
DT		 2961 /* an lkb may still exist even though the lock is EOL'ed due to a
2962 cancel, unlock or failed noqueue request; an app can't use these
2963 locks; return same error as if the lkid had not been found at all */
e7fd4179 2964
ef0c2bb0
DT		 2965 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2966 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2967 rv = -ENOENT;
e7fd4179 2968 goto out;
ef0c2bb0 2969 }
e7fd4179 2970
ef0c2bb0
DT		 2971 /* an lkb may be waiting for an rsb lookup to complete where the
2972 lookup was initiated by another lock */
2973
42dc1601
DT		 2974 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2975 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
ef0c2bb0
DT		 2976 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2977 list_del_init(&lkb->lkb_rsb_lookup);
2978 queue_cast(lkb->lkb_resource, lkb,
2979 args->flags & DLM_LKF_CANCEL ?
2980 -DLM_ECANCEL : -DLM_EUNLOCK);
2981 unhold_lkb(lkb); /* undoes create_lkb() */
ef0c2bb0 2982 }
42dc1601
DT		 2983 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2984 rv = -EBUSY;
2985 goto out;
ef0c2bb0
DT		 2986 }
2987
2988 /* cancel not allowed with another cancel/unlock in progress */
2989
2990 if (args->flags & DLM_LKF_CANCEL) {
2991 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2992 goto out;
2993
2994 if (is_overlap(lkb))
2995 goto out;
2996
3ae1acf9
DT		 2997 /* don't let scand try to do a cancel */
2998 del_timeout(lkb);
2999
ef0c2bb0
DT		 3000 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3001 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3002 rv = -EBUSY;
3003 goto out;
3004 }
3005
a536e381
DT		 3006 /* there's nothing to cancel */
3007 if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
3008 !lkb->lkb_wait_type) {
3009 rv = -EBUSY;
3010 goto out;
3011 }
3012
ef0c2bb0
DT		 3013 switch (lkb->lkb_wait_type) {
3014 case DLM_MSG_LOOKUP:
3015 case DLM_MSG_REQUEST:
3016 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3017 rv = -EBUSY;
3018 goto out;
3019 case DLM_MSG_UNLOCK:
3020 case DLM_MSG_CANCEL:
3021 goto out;
3022 }
3023 /* add_to_waiters() will set OVERLAP_CANCEL */
3024 goto out_ok;
3025 }
3026
3027 /* do we need to allow a force-unlock if there's a normal unlock
3028 already in progress? in what conditions could the normal unlock
3029 fail such that we'd want to send a force-unlock to be sure? */
3030
3031 if (args->flags & DLM_LKF_FORCEUNLOCK) {
3032 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3033 goto out;
3034
3035 if (is_overlap_unlock(lkb))
3036 goto out;
e7fd4179 3037
3ae1acf9
DT		 3038 /* don't let scand try to do a cancel */
3039 del_timeout(lkb);
3040
ef0c2bb0
DT		 3041 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3042 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3043 rv = -EBUSY;
3044 goto out;
3045 }
3046
3047 switch (lkb->lkb_wait_type) {
3048 case DLM_MSG_LOOKUP:
3049 case DLM_MSG_REQUEST:
3050 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3051 rv = -EBUSY;
3052 goto out;
3053 case DLM_MSG_UNLOCK:
3054 goto out;
3055 }
3056 /* add_to_waiters() will set OVERLAP_UNLOCK */
3057 goto out_ok;
3058 }
3059
3060 /* normal unlock not allowed if there's any op in progress */
e7fd4179 3061 rv = -EBUSY;
ef0c2bb0 3062 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
e7fd4179
DT		 3063 goto out;
3064
3065 out_ok:
ef0c2bb0
DT		 3066 /* an overlapping op shouldn't blow away exflags from other op */
3067 lkb->lkb_exflags |= args->flags;
e7fd4179
DT		 3068 lkb->lkb_sbflags = 0;
3069 lkb->lkb_astparam = args->astparam;
e7fd4179
DT		 3070 rv = 0;
3071 out:
ef0c2bb0
DT		 3072 if (rv)
3073 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
3074 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3075 args->flags, lkb->lkb_wait_type,
3076 lkb->lkb_resource->res_name);
e7fd4179
DT		 3077 return rv;
3078}
3079
3080/*
3081 * Four stage 4 varieties:
3082 * do_request(), do_convert(), do_unlock(), do_cancel()
3083 * These are called on the master node for the given lock and
3084 * from the central locking logic.
3085 */
3086
3087static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3088{
3089 int error = 0;
3090
c503a621 3091 if (can_be_granted(r, lkb, 1, 0, NULL)) {
e7fd4179
DT		 3092 grant_lock(r, lkb);
3093 queue_cast(r, lkb, 0);
3094 goto out;
3095 }
3096
3097 if (can_be_queued(lkb)) {
3098 error = -EINPROGRESS;
3099 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3ae1acf9 3100 add_timeout(lkb);
e7fd4179
DT		 3101 goto out;
3102 }
3103
3104 error = -EAGAIN;
e7fd4179 3105 queue_cast(r, lkb, -EAGAIN);
e7fd4179
DT		 3106 out:
3107 return error;
3108}
3109
cf6620ac
DT		 3110static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3111 int error)
3112{
3113 switch (error) {
3114 case -EAGAIN:
3115 if (force_blocking_asts(lkb))
3116 send_blocking_asts_all(r, lkb);
3117 break;
3118 case -EINPROGRESS:
3119 send_blocking_asts(r, lkb);
3120 break;
3121 }
3122}
3123
e7fd4179
DT		 3124static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3125{
3126 int error = 0;
c85d65e9 3127 int deadlk = 0;
e7fd4179
DT		 3128
3129 /* changing an existing lock may allow others to be granted */
3130
c503a621 3131 if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
e7fd4179
DT		 3132 grant_lock(r, lkb);
3133 queue_cast(r, lkb, 0);
e7fd4179
DT		 3134 goto out;
3135 }
3136
c85d65e9
DT		 3137 /* can_be_granted() detected that this lock would block in a conversion
3138 deadlock, so we leave it on the granted queue and return EDEADLK in
3139 the ast for the convert. */
3140
294e7e45 3141 if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
c85d65e9 3142 /* it's left on the granted queue */
c85d65e9
DT		 3143 revert_lock(r, lkb);
3144 queue_cast(r, lkb, -EDEADLK);
3145 error = -EDEADLK;
3146 goto out;
3147 }
3148
7d3c1feb
DT		 3149 /* is_demoted() means the can_be_granted() above set the grmode
3150 to NL, and left us on the granted queue. This auto-demotion
3151 (due to CONVDEADLK) might mean other locks, and/or this lock, are
3152 now grantable. We have to try to grant other converting locks
3153 before we try again to grant this one. */
3154
3155 if (is_demoted(lkb)) {
4875647a 3156 grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
c503a621 3157 if (_can_be_granted(r, lkb, 1, 0)) {
7d3c1feb
DT		 3158 grant_lock(r, lkb);
3159 queue_cast(r, lkb, 0);
7d3c1feb
DT		 3160 goto out;
3161 }
3162 /* else fall through and move to convert queue */
3163 }
3164
3165 if (can_be_queued(lkb)) {
e7fd4179
DT		 3166 error = -EINPROGRESS;
3167 del_lkb(r, lkb);
3168 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3ae1acf9 3169 add_timeout(lkb);
e7fd4179
DT		 3170 goto out;
3171 }
3172
3173 error = -EAGAIN;
e7fd4179 3174 queue_cast(r, lkb, -EAGAIN);
e7fd4179
DT		 3175 out:
3176 return error;
3177}
3178
cf6620ac
DT		 3179static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3180 int error)
3181{
3182 switch (error) {
3183 case 0:
4875647a 3184 grant_pending_locks(r, NULL);
cf6620ac
DT		 3185 /* grant_pending_locks also sends basts */
3186 break;
3187 case -EAGAIN:
3188 if (force_blocking_asts(lkb))
3189 send_blocking_asts_all(r, lkb);
3190 break;
3191 case -EINPROGRESS:
3192 send_blocking_asts(r, lkb);
3193 break;
3194 }
3195}
3196
e7fd4179
DT		 3197static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3198{
3199 remove_lock(r, lkb);
3200 queue_cast(r, lkb, -DLM_EUNLOCK);
e7fd4179
DT		 3201 return -DLM_EUNLOCK;
3202}
3203
cf6620ac
DT		 3204static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3205 int error)
3206{
4875647a 3207 grant_pending_locks(r, NULL);
cf6620ac
DT		 3208}
3209
ef0c2bb0 3210/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
c04fecb4 3211
e7fd4179
DT		 3212static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3213{
ef0c2bb0
DT		 3214 int error;
3215
3216 error = revert_lock(r, lkb);
3217 if (error) {
3218 queue_cast(r, lkb, -DLM_ECANCEL);
ef0c2bb0
DT		 3219 return -DLM_ECANCEL;
3220 }
3221 return 0;
e7fd4179
DT		 3222}
3223
cf6620ac
DT		 3224static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3225 int error)
3226{
3227 if (error)
4875647a 3228 grant_pending_locks(r, NULL);
cf6620ac
DT		 3229}
3230
e7fd4179
DT		 3231/*
3232 * Four stage 3 varieties:
3233 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3234 */
3235
3236/* add a new lkb to a possibly new rsb, called by requesting process */
3237
3238static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3239{
3240 int error;
3241
3242 /* set_master: sets lkb nodeid from r */
3243
3244 error = set_master(r, lkb);
3245 if (error < 0)
3246 goto out;
3247 if (error) {
3248 error = 0;
3249 goto out;
3250 }
3251
cf6620ac 3252 if (is_remote(r)) {
e7fd4179
DT		 3253 /* receive_request() calls do_request() on remote node */
3254 error = send_request(r, lkb);
cf6620ac 3255 } else {
e7fd4179 3256 error = do_request(r, lkb);
cf6620ac
DT		 3257 /* for remote locks the request_reply is sent
3258 between do_request and do_request_effects */
3259 do_request_effects(r, lkb, error);
3260 }
e7fd4179
DT		 3261 out:
3262 return error;
3263}
3264
3bcd3687 3265/* change some property of an existing lkb, e.g. mode */
e7fd4179
DT		 3266
3267static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3268{
3269 int error;
3270
cf6620ac 3271 if (is_remote(r)) {
e7fd4179
DT		 3272 /* receive_convert() calls do_convert() on remote node */
3273 error = send_convert(r, lkb);
cf6620ac 3274 } else {
e7fd4179 3275 error = do_convert(r, lkb);
cf6620ac
DT		 3276 /* for remote locks the convert_reply is sent
3277 between do_convert and do_convert_effects */
3278 do_convert_effects(r, lkb, error);
3279 }
e7fd4179
DT		 3280
3281 return error;
3282}
3283
3284/* remove an existing lkb from the granted queue */
3285
3286static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3287{
3288 int error;
3289
cf6620ac 3290 if (is_remote(r)) {
e7fd4179
DT		 3291 /* receive_unlock() calls do_unlock() on remote node */
3292 error = send_unlock(r, lkb);
cf6620ac 3293 } else {
e7fd4179 3294 error = do_unlock(r, lkb);
cf6620ac
DT		 3295 /* for remote locks the unlock_reply is sent
3296 between do_unlock and do_unlock_effects */
3297 do_unlock_effects(r, lkb, error);
3298 }
e7fd4179
DT		 3299
3300 return error;
3301}
3302
3303/* remove an existing lkb from the convert or wait queue */
3304
3305static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3306{
3307 int error;
3308
cf6620ac 3309 if (is_remote(r)) {
e7fd4179
DT		 3310 /* receive_cancel() calls do_cancel() on remote node */
3311 error = send_cancel(r, lkb);
cf6620ac 3312 } else {
e7fd4179 3313 error = do_cancel(r, lkb);
cf6620ac
DT		 3314 /* for remote locks the cancel_reply is sent
3315 between do_cancel and do_cancel_effects */
3316 do_cancel_effects(r, lkb, error);
3317 }
e7fd4179
DT		 3318
3319 return error;
3320}
3321
3322/*
3323 * Four stage 2 varieties:
3324 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3325 */
3326
3327static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
3328 int len, struct dlm_args *args)
3329{
3330 struct dlm_rsb *r;
3331 int error;
3332
3333 error = validate_lock_args(ls, lkb, args);
3334 if (error)
c04fecb4 3335 return error;
e7fd4179 3336
c04fecb4 3337 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
e7fd4179 3338 if (error)
c04fecb4 3339 return error;
e7fd4179
DT		 3340
3341 lock_rsb(r);
3342
3343 attach_lkb(r, lkb);
3344 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3345
3346 error = _request_lock(r, lkb);
3347
3348 unlock_rsb(r);
3349 put_rsb(r);
e7fd4179
DT		 3350 return error;
3351}
3352
3353static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3354 struct dlm_args *args)
3355{
3356 struct dlm_rsb *r;
3357 int error;
3358
3359 r = lkb->lkb_resource;
3360
3361 hold_rsb(r);
3362 lock_rsb(r);
3363
3364 error = validate_lock_args(ls, lkb, args);
3365 if (error)
3366 goto out;
3367
3368 error = _convert_lock(r, lkb);
3369 out:
3370 unlock_rsb(r);
3371 put_rsb(r);
3372 return error;
3373}
3374
3375static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3376 struct dlm_args *args)
3377{
3378 struct dlm_rsb *r;
3379 int error;
3380
3381 r = lkb->lkb_resource;
3382
3383 hold_rsb(r);
3384 lock_rsb(r);
3385
3386 error = validate_unlock_args(lkb, args);
3387 if (error)
3388 goto out;
3389
3390 error = _unlock_lock(r, lkb);
3391 out:
3392 unlock_rsb(r);
3393 put_rsb(r);
3394 return error;
3395}
3396
3397static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3398 struct dlm_args *args)
3399{
3400 struct dlm_rsb *r;
3401 int error;
3402
3403 r = lkb->lkb_resource;
3404
3405 hold_rsb(r);
3406 lock_rsb(r);
3407
3408 error = validate_unlock_args(lkb, args);
3409 if (error)
3410 goto out;
3411
3412 error = _cancel_lock(r, lkb);
3413 out:
3414 unlock_rsb(r);
3415 put_rsb(r);
3416 return error;
3417}
3418
3419/*
3420 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
3421 */
3422
3423int dlm_lock(dlm_lockspace_t *lockspace,
3424 int mode,
3425 struct dlm_lksb *lksb,
3426 uint32_t flags,
3427 void *name,
3428 unsigned int namelen,
3429 uint32_t parent_lkid,
3430 void (*ast) (void *astarg),
3431 void *astarg,
3bcd3687 3432 void (*bast) (void *astarg, int mode))
e7fd4179
DT		 3433{
3434 struct dlm_ls *ls;
3435 struct dlm_lkb *lkb;
3436 struct dlm_args args;
3437 int error, convert = flags & DLM_LKF_CONVERT;
3438
3439 ls = dlm_find_lockspace_local(lockspace);
3440 if (!ls)
3441 return -EINVAL;
3442
85e86edf 3443 dlm_lock_recovery(ls);
e7fd4179
DT		 3444
3445 if (convert)
3446 error = find_lkb(ls, lksb->sb_lkid, &lkb);
3447 else
3448 error = create_lkb(ls, &lkb);
3449
3450 if (error)
3451 goto out;
3452
f1d3b8f9
AA		 3453 trace_dlm_lock_start(ls, lkb, mode, flags);
3454
d7db923e 3455 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3bcd3687 3456 astarg, bast, &args);
e7fd4179
DT		 3457 if (error)
3458 goto out_put;
3459
3460 if (convert)
3461 error = convert_lock(ls, lkb, &args);
3462 else
3463 error = request_lock(ls, lkb, name, namelen, &args);
3464
3465 if (error == -EINPROGRESS)
3466 error = 0;
3467 out_put:
f1d3b8f9
AA		 3468 trace_dlm_lock_end(ls, lkb, mode, flags, error);
3469
e7fd4179 3470 if (convert || error)
b3f58d8f 3471 __put_lkb(ls, lkb);
c85d65e9 3472 if (error == -EAGAIN || error == -EDEADLK)
e7fd4179
DT		 3473 error = 0;
3474 out:
85e86edf 3475 dlm_unlock_recovery(ls);
e7fd4179
DT		 3476 dlm_put_lockspace(ls);
3477 return error;
3478}
3479
3480int dlm_unlock(dlm_lockspace_t *lockspace,
3481 uint32_t lkid,
3482 uint32_t flags,
3483 struct dlm_lksb *lksb,
3484 void *astarg)
3485{
3486 struct dlm_ls *ls;
3487 struct dlm_lkb *lkb;
3488 struct dlm_args args;
3489 int error;
3490
3491 ls = dlm_find_lockspace_local(lockspace);
3492 if (!ls)
3493 return -EINVAL;
3494
85e86edf 3495 dlm_lock_recovery(ls);
e7fd4179
DT		 3496
3497 error = find_lkb(ls, lkid, &lkb);
3498 if (error)
3499 goto out;
3500
f1d3b8f9
AA		 3501 trace_dlm_unlock_start(ls, lkb, flags);
3502
e7fd4179
DT		 3503 error = set_unlock_args(flags, astarg, &args);
3504 if (error)
3505 goto out_put;
3506
3507 if (flags & DLM_LKF_CANCEL)
3508 error = cancel_lock(ls, lkb, &args);
3509 else
3510 error = unlock_lock(ls, lkb, &args);
3511
3512 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3513 error = 0;
ef0c2bb0
DT		 3514 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3515 error = 0;
e7fd4179 3516 out_put:
f1d3b8f9
AA		 3517 trace_dlm_unlock_end(ls, lkb, flags, error);
3518
b3f58d8f 3519 dlm_put_lkb(lkb);
e7fd4179 3520 out:
85e86edf 3521 dlm_unlock_recovery(ls);
e7fd4179
DT		 3522 dlm_put_lockspace(ls);
3523 return error;
3524}
3525
3526/*
3527 * send/receive routines for remote operations and replies
3528 *
3529 * send_args
3530 * send_common
3531 * send_request receive_request
3532 * send_convert receive_convert
3533 * send_unlock receive_unlock
3534 * send_cancel receive_cancel
3535 * send_grant receive_grant
3536 * send_bast receive_bast
3537 * send_lookup receive_lookup
3538 * send_remove receive_remove
3539 *
3540 * send_common_reply
3541 * receive_request_reply send_request_reply
3542 * receive_convert_reply send_convert_reply
3543 * receive_unlock_reply send_unlock_reply
3544 * receive_cancel_reply send_cancel_reply
3545 * receive_lookup_reply send_lookup_reply
3546 */
3547
7e4dac33
DT		 3548static int _create_message(struct dlm_ls *ls, int mb_len,
3549 int to_nodeid, int mstype,
3550 struct dlm_message **ms_ret,
3551 struct dlm_mhandle **mh_ret)
e7fd4179
DT		 3552{
3553 struct dlm_message *ms;
3554 struct dlm_mhandle *mh;
3555 char *mb;
e7fd4179
DT		 3556
3557 /* get_buffer gives us a message handle (mh) that we need to
a070a91c 3558 pass into midcomms_commit and a message buffer (mb) that we
e7fd4179
DT		 3559 write our data into */
3560
a070a91c 3561 mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb);
e7fd4179
DT		 3562 if (!mh)
3563 return -ENOBUFS;
3564
e7fd4179
DT		 3565 ms = (struct dlm_message *) mb;
3566
3428785a
AA		 3567 ms->m_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3568 ms->m_header.u.h_lockspace = cpu_to_le32(ls->ls_global_id);
3569 ms->m_header.h_nodeid = cpu_to_le32(dlm_our_nodeid());
3570 ms->m_header.h_length = cpu_to_le16(mb_len);
e7fd4179
DT		 3571 ms->m_header.h_cmd = DLM_MSG;
3572
00e99ccd 3573 ms->m_type = cpu_to_le32(mstype);
e7fd4179
DT		 3574
3575 *mh_ret = mh;
3576 *ms_ret = ms;
3577 return 0;
3578}
3579
7e4dac33
DT		 3580static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3581 int to_nodeid, int mstype,
3582 struct dlm_message **ms_ret,
3583 struct dlm_mhandle **mh_ret)
3584{
3585 int mb_len = sizeof(struct dlm_message);
3586
3587 switch (mstype) {
3588 case DLM_MSG_REQUEST:
3589 case DLM_MSG_LOOKUP:
3590 case DLM_MSG_REMOVE:
3591 mb_len += r->res_length;
3592 break;
3593 case DLM_MSG_CONVERT:
3594 case DLM_MSG_UNLOCK:
3595 case DLM_MSG_REQUEST_REPLY:
3596 case DLM_MSG_CONVERT_REPLY:
3597 case DLM_MSG_GRANT:
3598 if (lkb && lkb->lkb_lvbptr)
3599 mb_len += r->res_ls->ls_lvblen;
3600 break;
3601 }
3602
3603 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3604 ms_ret, mh_ret);
3605}
3606
e7fd4179
DT		 3607/* further lowcomms enhancements or alternate implementations may make
3608 the return value from this function useful at some point */
3609
3610static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
3611{
a070a91c 3612 dlm_midcomms_commit_mhandle(mh);
e7fd4179
DT		 3613 return 0;
3614}
3615
3616static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3617 struct dlm_message *ms)
3618{
00e99ccd
AA		 3619 ms->m_nodeid = cpu_to_le32(lkb->lkb_nodeid);
3620 ms->m_pid = cpu_to_le32(lkb->lkb_ownpid);
3621 ms->m_lkid = cpu_to_le32(lkb->lkb_id);
3622 ms->m_remid = cpu_to_le32(lkb->lkb_remid);
3623 ms->m_exflags = cpu_to_le32(lkb->lkb_exflags);
3624 ms->m_sbflags = cpu_to_le32(lkb->lkb_sbflags);
3625 ms->m_flags = cpu_to_le32(lkb->lkb_flags);
3626 ms->m_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
3627 ms->m_status = cpu_to_le32(lkb->lkb_status);
3628 ms->m_grmode = cpu_to_le32(lkb->lkb_grmode);
3629 ms->m_rqmode = cpu_to_le32(lkb->lkb_rqmode);
3630 ms->m_hash = cpu_to_le32(r->res_hash);
e7fd4179
DT		 3631
3632 /* m_result and m_bastmode are set from function args,
3633 not from lkb fields */
3634
e5dae548 3635 if (lkb->lkb_bastfn)
00e99ccd 3636 ms->m_asts |= cpu_to_le32(DLM_CB_BAST);
e5dae548 3637 if (lkb->lkb_astfn)
00e99ccd 3638 ms->m_asts |= cpu_to_le32(DLM_CB_CAST);
e7fd4179 3639
da49f36f
DT		 3640 /* compare with switch in create_message; send_remove() doesn't
3641 use send_args() */
e7fd4179 3642
da49f36f 3643 switch (ms->m_type) {
00e99ccd
AA		 3644 case cpu_to_le32(DLM_MSG_REQUEST):
3645 case cpu_to_le32(DLM_MSG_LOOKUP):
da49f36f
DT		 3646 memcpy(ms->m_extra, r->res_name, r->res_length);
3647 break;
00e99ccd
AA		 3648 case cpu_to_le32(DLM_MSG_CONVERT):
3649 case cpu_to_le32(DLM_MSG_UNLOCK):
3650 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3651 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3652 case cpu_to_le32(DLM_MSG_GRANT):
da49f36f
DT		 3653 if (!lkb->lkb_lvbptr)
3654 break;
e7fd4179 3655 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
da49f36f
DT		 3656 break;
3657 }
e7fd4179
DT		 3658}
3659
3660static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3661{
3662 struct dlm_message *ms;
3663 struct dlm_mhandle *mh;
3664 int to_nodeid, error;
3665
c6ff669b
DT		 3666 to_nodeid = r->res_nodeid;
3667
3668 error = add_to_waiters(lkb, mstype, to_nodeid);
ef0c2bb0
DT		 3669 if (error)
3670 return error;
e7fd4179 3671
e7fd4179
DT		 3672 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3673 if (error)
3674 goto fail;
3675
3676 send_args(r, lkb, ms);
3677
3678 error = send_message(mh, ms);
3679 if (error)
3680 goto fail;
3681 return 0;
3682
3683 fail:
ef0c2bb0 3684 remove_from_waiters(lkb, msg_reply_type(mstype));
e7fd4179
DT		 3685 return error;
3686}
3687
3688static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3689{
3690 return send_common(r, lkb, DLM_MSG_REQUEST);
3691}
3692
3693static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3694{
3695 int error;
3696
3697 error = send_common(r, lkb, DLM_MSG_CONVERT);
3698
3699 /* down conversions go without a reply from the master */
3700 if (!error && down_conversion(lkb)) {
ef0c2bb0 3701 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
00e99ccd
AA		 3702 r->res_ls->ls_stub_ms.m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
3703 r->res_ls->ls_stub_ms.m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
e7fd4179
DT		 3704 r->res_ls->ls_stub_ms.m_result = 0;
3705 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3706 }
3707
3708 return error;
3709}
3710
3711/* FIXME: if this lkb is the only lock we hold on the rsb, then set
3712 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3713 that the master is still correct. */
3714
3715static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3716{
3717 return send_common(r, lkb, DLM_MSG_UNLOCK);
3718}
3719
3720static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3721{
3722 return send_common(r, lkb, DLM_MSG_CANCEL);
3723}
3724
3725static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3726{
3727 struct dlm_message *ms;
3728 struct dlm_mhandle *mh;
3729 int to_nodeid, error;
3730
3731 to_nodeid = lkb->lkb_nodeid;
3732
3733 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3734 if (error)
3735 goto out;
3736
3737 send_args(r, lkb, ms);
3738
3739 ms->m_result = 0;
3740
3741 error = send_message(mh, ms);
3742 out:
3743 return error;
3744}
3745
3746static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3747{
3748 struct dlm_message *ms;
3749 struct dlm_mhandle *mh;
3750 int to_nodeid, error;
3751
3752 to_nodeid = lkb->lkb_nodeid;
3753
3754 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3755 if (error)
3756 goto out;
3757
3758 send_args(r, lkb, ms);
3759
00e99ccd 3760 ms->m_bastmode = cpu_to_le32(mode);
e7fd4179
DT		 3761
3762 error = send_message(mh, ms);
3763 out:
3764 return error;
3765}
3766
3767static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3768{
3769 struct dlm_message *ms;
3770 struct dlm_mhandle *mh;
3771 int to_nodeid, error;
3772
c6ff669b
DT		 3773 to_nodeid = dlm_dir_nodeid(r);
3774
3775 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
ef0c2bb0
DT		 3776 if (error)
3777 return error;
e7fd4179 3778
e7fd4179
DT		 3779 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3780 if (error)
3781 goto fail;
3782
3783 send_args(r, lkb, ms);
3784
3785 error = send_message(mh, ms);
3786 if (error)
3787 goto fail;
3788 return 0;
3789
3790 fail:
ef0c2bb0 3791 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
e7fd4179
DT		 3792 return error;
3793}
3794
3795static int send_remove(struct dlm_rsb *r)
3796{
3797 struct dlm_message *ms;
3798 struct dlm_mhandle *mh;
3799 int to_nodeid, error;
3800
3801 to_nodeid = dlm_dir_nodeid(r);
3802
3803 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3804 if (error)
3805 goto out;
3806
3807 memcpy(ms->m_extra, r->res_name, r->res_length);
00e99ccd 3808 ms->m_hash = cpu_to_le32(r->res_hash);
e7fd4179
DT		 3809
3810 error = send_message(mh, ms);
3811 out:
3812 return error;
3813}
3814
3815static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3816 int mstype, int rv)
3817{
3818 struct dlm_message *ms;
3819 struct dlm_mhandle *mh;
3820 int to_nodeid, error;
3821
3822 to_nodeid = lkb->lkb_nodeid;
3823
3824 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3825 if (error)
3826 goto out;
3827
3828 send_args(r, lkb, ms);
3829
00e99ccd 3830 ms->m_result = cpu_to_le32(to_dlm_errno(rv));
e7fd4179
DT		 3831
3832 error = send_message(mh, ms);
3833 out:
3834 return error;
3835}
3836
3837static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3838{
3839 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3840}
3841
3842static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3843{
3844 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3845}
3846
3847static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3848{
3849 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3850}
3851
3852static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3853{
3854 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3855}
3856
3857static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3858 int ret_nodeid, int rv)
3859{
3860 struct dlm_rsb *r = &ls->ls_stub_rsb;
3861 struct dlm_message *ms;
3862 struct dlm_mhandle *mh;
3428785a 3863 int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid);
e7fd4179
DT		 3864
3865 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3866 if (error)
3867 goto out;
3868
3869 ms->m_lkid = ms_in->m_lkid;
00e99ccd
AA		 3870 ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3871 ms->m_nodeid = cpu_to_le32(ret_nodeid);
e7fd4179
DT		 3872
3873 error = send_message(mh, ms);
3874 out:
3875 return error;
3876}
3877
3878/* which args we save from a received message depends heavily on the type
3879 of message, unlike the send side where we can safely send everything about
3880 the lkb for any type of message */
3881
3882static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3883{
00e99ccd
AA		 3884 lkb->lkb_exflags = le32_to_cpu(ms->m_exflags);
3885 lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
e7fd4179 3886 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
00e99ccd 3887 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
e7fd4179
DT		 3888}
3889
3890static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3891{
00e99ccd 3892 if (ms->m_flags == cpu_to_le32(DLM_IFL_STUB_MS))
2a7ce0ed
DT		 3893 return;
3894
00e99ccd 3895 lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
e7fd4179 3896 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
00e99ccd 3897 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
e7fd4179
DT		 3898}
3899
3900static int receive_extralen(struct dlm_message *ms)
3901{
3428785a
AA		 3902 return (le16_to_cpu(ms->m_header.h_length) -
3903 sizeof(struct dlm_message));
e7fd4179
DT		 3904}
3905
e7fd4179
DT		 3906static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3907 struct dlm_message *ms)
3908{
3909 int len;
3910
3911 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3912 if (!lkb->lkb_lvbptr)
52bda2b5 3913 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
e7fd4179
DT		 3914 if (!lkb->lkb_lvbptr)
3915 return -ENOMEM;
3916 len = receive_extralen(ms);
cfa805f6
BVA		 3917 if (len > ls->ls_lvblen)
3918 len = ls->ls_lvblen;
e7fd4179
DT		 3919 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3920 }
3921 return 0;
3922}
3923
e5dae548
DT		 3924static void fake_bastfn(void *astparam, int mode)
3925{
3926 log_print("fake_bastfn should not be called");
3927}
3928
3929static void fake_astfn(void *astparam)
3930{
3931 log_print("fake_astfn should not be called");
3932}
3933
e7fd4179
DT		 3934static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3935 struct dlm_message *ms)
3936{
3428785a 3937 lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
00e99ccd
AA		 3938 lkb->lkb_ownpid = le32_to_cpu(ms->m_pid);
3939 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
e7fd4179 3940 lkb->lkb_grmode = DLM_LOCK_IV;
00e99ccd 3941 lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
e5dae548 3942
00e99ccd
AA		 3943 lkb->lkb_bastfn = (ms->m_asts & cpu_to_le32(DLM_CB_BAST)) ? &fake_bastfn : NULL;
3944 lkb->lkb_astfn = (ms->m_asts & cpu_to_le32(DLM_CB_CAST)) ? &fake_astfn : NULL;
e7fd4179 3945
8d07fd50
DT		 3946 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3947 /* lkb was just created so there won't be an lvb yet */
52bda2b5 3948 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
8d07fd50
DT		 3949 if (!lkb->lkb_lvbptr)
3950 return -ENOMEM;
3951 }
e7fd4179
DT		 3952
3953 return 0;
3954}
3955
3956static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3957 struct dlm_message *ms)
3958{
e7fd4179
DT		 3959 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3960 return -EBUSY;
3961
e7fd4179
DT		 3962 if (receive_lvb(ls, lkb, ms))
3963 return -ENOMEM;
3964
00e99ccd
AA		 3965 lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3966 lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
e7fd4179
DT		 3967
3968 return 0;
3969}
3970
3971static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3972 struct dlm_message *ms)
3973{
e7fd4179
DT		 3974 if (receive_lvb(ls, lkb, ms))
3975 return -ENOMEM;
3976 return 0;
3977}
3978
3979/* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3980 uses to send a reply and that the remote end uses to process the reply. */
3981
3982static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3983{
3984 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3428785a 3985 lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
00e99ccd 3986 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
e7fd4179
DT		 3987}
3988
c54e04b0
DT		 3989/* This is called after the rsb is locked so that we can safely inspect
3990 fields in the lkb. */
3991
3992static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3993{
3428785a 3994 int from = le32_to_cpu(ms->m_header.h_nodeid);
c54e04b0
DT		 3995 int error = 0;
3996
6c2e3bf6 3997 /* currently mixing of user/kernel locks are not supported */
00e99ccd
AA		 3998 if (ms->m_flags & cpu_to_le32(DLM_IFL_USER) &&
3999 ~lkb->lkb_flags & DLM_IFL_USER) {
6c2e3bf6
AA		 4000 log_error(lkb->lkb_resource->res_ls,
4001 "got user dlm message for a kernel lock");
4002 error = -EINVAL;
4003 goto out;
4004 }
4005
c54e04b0 4006 switch (ms->m_type) {
00e99ccd
AA		 4007 case cpu_to_le32(DLM_MSG_CONVERT):
4008 case cpu_to_le32(DLM_MSG_UNLOCK):
4009 case cpu_to_le32(DLM_MSG_CANCEL):
c54e04b0
DT		 4010 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
4011 error = -EINVAL;
4012 break;
4013
00e99ccd
AA		 4014 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4015 case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4016 case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4017 case cpu_to_le32(DLM_MSG_GRANT):
4018 case cpu_to_le32(DLM_MSG_BAST):
c54e04b0
DT		 4019 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
4020 error = -EINVAL;
4021 break;
4022
00e99ccd 4023 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
c54e04b0
DT		 4024 if (!is_process_copy(lkb))
4025 error = -EINVAL;
4026 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
4027 error = -EINVAL;
4028 break;
4029
4030 default:
4031 error = -EINVAL;
4032 }
4033
6c2e3bf6 4034out:
c54e04b0
DT		 4035 if (error)
4036 log_error(lkb->lkb_resource->res_ls,
4037 "ignore invalid message %d from %d %x %x %x %d",
00e99ccd
AA		 4038 le32_to_cpu(ms->m_type), from, lkb->lkb_id,
4039 lkb->lkb_remid, lkb->lkb_flags, lkb->lkb_nodeid);
c54e04b0
DT		 4040 return error;
4041}
4042
96006ea6
DT		 4043static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
4044{
4045 char name[DLM_RESNAME_MAXLEN + 1];
4046 struct dlm_message *ms;
4047 struct dlm_mhandle *mh;
4048 struct dlm_rsb *r;
4049 uint32_t hash, b;
4050 int rv, dir_nodeid;
4051
4052 memset(name, 0, sizeof(name));
4053 memcpy(name, ms_name, len);
4054
4055 hash = jhash(name, len, 0);
4056 b = hash & (ls->ls_rsbtbl_size - 1);
4057
4058 dir_nodeid = dlm_hash2nodeid(ls, hash);
4059
4060 log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
4061
4062 spin_lock(&ls->ls_rsbtbl[b].lock);
4063 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4064 if (!rv) {
4065 spin_unlock(&ls->ls_rsbtbl[b].lock);
4066 log_error(ls, "repeat_remove on keep %s", name);
4067 return;
4068 }
4069
4070 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4071 if (!rv) {
4072 spin_unlock(&ls->ls_rsbtbl[b].lock);
4073 log_error(ls, "repeat_remove on toss %s", name);
4074 return;
4075 }
4076
4077 /* use ls->remove_name2 to avoid conflict with shrink? */
4078
4079 spin_lock(&ls->ls_remove_spin);
4080 ls->ls_remove_len = len;
4081 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
4082 spin_unlock(&ls->ls_remove_spin);
4083 spin_unlock(&ls->ls_rsbtbl[b].lock);
21d9ac1a 4084 wake_up(&ls->ls_remove_wait);
96006ea6
DT		 4085
4086 rv = _create_message(ls, sizeof(struct dlm_message) + len,
4087 dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
4088 if (rv)
4089 return;
4090
4091 memcpy(ms->m_extra, name, len);
00e99ccd 4092 ms->m_hash = cpu_to_le32(hash);
96006ea6
DT		 4093
4094 send_message(mh, ms);
4095
4096 spin_lock(&ls->ls_remove_spin);
4097 ls->ls_remove_len = 0;
4098 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
4099 spin_unlock(&ls->ls_remove_spin);
4100}
4101
6d40c4a7 4102static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4103{
4104 struct dlm_lkb *lkb;
4105 struct dlm_rsb *r;
c04fecb4 4106 int from_nodeid;
96006ea6 4107 int error, namelen = 0;
e7fd4179 4108
3428785a 4109 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
c04fecb4 4110
e7fd4179
DT		 4111 error = create_lkb(ls, &lkb);
4112 if (error)
4113 goto fail;
4114
4115 receive_flags(lkb, ms);
4116 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4117 error = receive_request_args(ls, lkb, ms);
4118 if (error) {
b3f58d8f 4119 __put_lkb(ls, lkb);
e7fd4179
DT		 4120 goto fail;
4121 }
4122
c04fecb4
DT		 4123 /* The dir node is the authority on whether we are the master
4124 for this rsb or not, so if the master sends us a request, we should
4125 recreate the rsb if we've destroyed it. This race happens when we
4126 send a remove message to the dir node at the same time that the dir
4127 node sends us a request for the rsb. */
4128
e7fd4179
DT		 4129 namelen = receive_extralen(ms);
4130
c04fecb4
DT		 4131 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4132 R_RECEIVE_REQUEST, &r);
e7fd4179 4133 if (error) {
b3f58d8f 4134 __put_lkb(ls, lkb);
e7fd4179
DT		 4135 goto fail;
4136 }
4137
4138 lock_rsb(r);
4139
c04fecb4
DT		 4140 if (r->res_master_nodeid != dlm_our_nodeid()) {
4141 error = validate_master_nodeid(ls, r, from_nodeid);
4142 if (error) {
4143 unlock_rsb(r);
4144 put_rsb(r);
4145 __put_lkb(ls, lkb);
4146 goto fail;
4147 }
4148 }
4149
e7fd4179
DT		 4150 attach_lkb(r, lkb);
4151 error = do_request(r, lkb);
4152 send_request_reply(r, lkb, error);
cf6620ac 4153 do_request_effects(r, lkb, error);
e7fd4179
DT		 4154
4155 unlock_rsb(r);
4156 put_rsb(r);
4157
4158 if (error == -EINPROGRESS)
4159 error = 0;
4160 if (error)
b3f58d8f 4161 dlm_put_lkb(lkb);
6d40c4a7 4162 return 0;
e7fd4179
DT		 4163
4164 fail:
c04fecb4
DT		 4165 /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4166 and do this receive_request again from process_lookup_list once
4167 we get the lookup reply. This would avoid a many repeated
4168 ENOTBLK request failures when the lookup reply designating us
4169 as master is delayed. */
4170
4171 /* We could repeatedly return -EBADR here if our send_remove() is
4172 delayed in being sent/arriving/being processed on the dir node.
4173 Another node would repeatedly lookup up the master, and the dir
4174 node would continue returning our nodeid until our send_remove
96006ea6
DT		 4175 took effect.
4176
4177 We send another remove message in case our previous send_remove
4178 was lost/ignored/missed somehow. */
c04fecb4
DT		 4179
4180 if (error != -ENOTBLK) {
4181 log_limit(ls, "receive_request %x from %d %d",
00e99ccd 4182 le32_to_cpu(ms->m_lkid), from_nodeid, error);
c04fecb4
DT		 4183 }
4184
96006ea6
DT		 4185 if (namelen && error == -EBADR) {
4186 send_repeat_remove(ls, ms->m_extra, namelen);
4187 msleep(1000);
4188 }
4189
e7fd4179
DT		 4190 setup_stub_lkb(ls, ms);
4191 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
6d40c4a7 4192 return error;
e7fd4179
DT		 4193}
4194
6d40c4a7 4195static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4196{
4197 struct dlm_lkb *lkb;
4198 struct dlm_rsb *r;
90135925 4199 int error, reply = 1;
e7fd4179 4200
00e99ccd 4201 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
e7fd4179
DT		 4202 if (error)
4203 goto fail;
4204
00e99ccd 4205 if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4875647a
DT		 4206 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4207 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4208 (unsigned long long)lkb->lkb_recover_seq,
00e99ccd
AA		 4209 le32_to_cpu(ms->m_header.h_nodeid),
4210 le32_to_cpu(ms->m_lkid));
6d40c4a7 4211 error = -ENOENT;
c0174726 4212 dlm_put_lkb(lkb);
6d40c4a7
DT		 4213 goto fail;
4214 }
4215
e7fd4179
DT		 4216 r = lkb->lkb_resource;
4217
4218 hold_rsb(r);
4219 lock_rsb(r);
4220
c54e04b0
DT		 4221 error = validate_message(lkb, ms);
4222 if (error)
4223 goto out;
4224
e7fd4179 4225 receive_flags(lkb, ms);
cf6620ac 4226
e7fd4179 4227 error = receive_convert_args(ls, lkb, ms);
cf6620ac
DT		 4228 if (error) {
4229 send_convert_reply(r, lkb, error);
4230 goto out;
4231 }
4232
e7fd4179
DT		 4233 reply = !down_conversion(lkb);
4234
4235 error = do_convert(r, lkb);
e7fd4179
DT		 4236 if (reply)
4237 send_convert_reply(r, lkb, error);
cf6620ac 4238 do_convert_effects(r, lkb, error);
c54e04b0 4239 out:
e7fd4179
DT		 4240 unlock_rsb(r);
4241 put_rsb(r);
b3f58d8f 4242 dlm_put_lkb(lkb);
6d40c4a7 4243 return 0;
e7fd4179
DT		 4244
4245 fail:
4246 setup_stub_lkb(ls, ms);
4247 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
6d40c4a7 4248 return error;
e7fd4179
DT		 4249}
4250
6d40c4a7 4251static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4252{
4253 struct dlm_lkb *lkb;
4254 struct dlm_rsb *r;
4255 int error;
4256
00e99ccd 4257 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
e7fd4179
DT		 4258 if (error)
4259 goto fail;
4260
00e99ccd 4261 if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
6d40c4a7
DT		 4262 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4263 lkb->lkb_id, lkb->lkb_remid,
00e99ccd
AA		 4264 le32_to_cpu(ms->m_header.h_nodeid),
4265 le32_to_cpu(ms->m_lkid));
6d40c4a7 4266 error = -ENOENT;
c0174726 4267 dlm_put_lkb(lkb);
6d40c4a7
DT		 4268 goto fail;
4269 }
4270
e7fd4179
DT		 4271 r = lkb->lkb_resource;
4272
4273 hold_rsb(r);
4274 lock_rsb(r);
4275
c54e04b0
DT		 4276 error = validate_message(lkb, ms);
4277 if (error)
4278 goto out;
4279
e7fd4179 4280 receive_flags(lkb, ms);
cf6620ac 4281
e7fd4179 4282 error = receive_unlock_args(ls, lkb, ms);
cf6620ac
DT		 4283 if (error) {
4284 send_unlock_reply(r, lkb, error);
4285 goto out;
4286 }
e7fd4179
DT		 4287
4288 error = do_unlock(r, lkb);
e7fd4179 4289 send_unlock_reply(r, lkb, error);
cf6620ac 4290 do_unlock_effects(r, lkb, error);
c54e04b0 4291 out:
e7fd4179
DT		 4292 unlock_rsb(r);
4293 put_rsb(r);
b3f58d8f 4294 dlm_put_lkb(lkb);
6d40c4a7 4295 return 0;
e7fd4179
DT		 4296
4297 fail:
4298 setup_stub_lkb(ls, ms);
4299 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
6d40c4a7 4300 return error;
e7fd4179
DT		 4301}
4302
6d40c4a7 4303static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4304{
4305 struct dlm_lkb *lkb;
4306 struct dlm_rsb *r;
4307 int error;
4308
00e99ccd 4309 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
e7fd4179
DT		 4310 if (error)
4311 goto fail;
4312
4313 receive_flags(lkb, ms);
4314
4315 r = lkb->lkb_resource;
4316
4317 hold_rsb(r);
4318 lock_rsb(r);
4319
c54e04b0
DT		 4320 error = validate_message(lkb, ms);
4321 if (error)
4322 goto out;
4323
e7fd4179
DT		 4324 error = do_cancel(r, lkb);
4325 send_cancel_reply(r, lkb, error);
cf6620ac 4326 do_cancel_effects(r, lkb, error);
c54e04b0 4327 out:
e7fd4179
DT		 4328 unlock_rsb(r);
4329 put_rsb(r);
b3f58d8f 4330 dlm_put_lkb(lkb);
6d40c4a7 4331 return 0;
e7fd4179
DT		 4332
4333 fail:
4334 setup_stub_lkb(ls, ms);
4335 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
6d40c4a7 4336 return error;
e7fd4179
DT		 4337}
4338
6d40c4a7 4339static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4340{
4341 struct dlm_lkb *lkb;
4342 struct dlm_rsb *r;
4343 int error;
4344
00e99ccd 4345 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
6d40c4a7
DT		 4346 if (error)
4347 return error;
e7fd4179
DT		 4348
4349 r = lkb->lkb_resource;
4350
4351 hold_rsb(r);
4352 lock_rsb(r);
4353
c54e04b0
DT		 4354 error = validate_message(lkb, ms);
4355 if (error)
4356 goto out;
4357
e7fd4179 4358 receive_flags_reply(lkb, ms);
7d3c1feb
DT		 4359 if (is_altmode(lkb))
4360 munge_altmode(lkb, ms);
e7fd4179
DT		 4361 grant_lock_pc(r, lkb, ms);
4362 queue_cast(r, lkb, 0);
c54e04b0 4363 out:
e7fd4179
DT		 4364 unlock_rsb(r);
4365 put_rsb(r);
b3f58d8f 4366 dlm_put_lkb(lkb);
6d40c4a7 4367 return 0;
e7fd4179
DT		 4368}
4369
6d40c4a7 4370static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4371{
4372 struct dlm_lkb *lkb;
4373 struct dlm_rsb *r;
4374 int error;
4375
00e99ccd 4376 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
6d40c4a7
DT		 4377 if (error)
4378 return error;
e7fd4179
DT		 4379
4380 r = lkb->lkb_resource;
4381
4382 hold_rsb(r);
4383 lock_rsb(r);
4384
c54e04b0
DT		 4385 error = validate_message(lkb, ms);
4386 if (error)
4387 goto out;
e7fd4179 4388
00e99ccd
AA		 4389 queue_bast(r, lkb, le32_to_cpu(ms->m_bastmode));
4390 lkb->lkb_highbast = le32_to_cpu(ms->m_bastmode);
c54e04b0 4391 out:
e7fd4179
DT		 4392 unlock_rsb(r);
4393 put_rsb(r);
b3f58d8f 4394 dlm_put_lkb(lkb);
6d40c4a7 4395 return 0;
e7fd4179
DT		 4396}
4397
4398static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4399{
c04fecb4 4400 int len, error, ret_nodeid, from_nodeid, our_nodeid;
e7fd4179 4401
3428785a 4402 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
e7fd4179
DT		 4403 our_nodeid = dlm_our_nodeid();
4404
4405 len = receive_extralen(ms);
4406
c04fecb4
DT		 4407 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4408 &ret_nodeid, NULL);
e7fd4179
DT		 4409
4410 /* Optimization: we're master so treat lookup as a request */
4411 if (!error && ret_nodeid == our_nodeid) {
4412 receive_request(ls, ms);
4413 return;
4414 }
e7fd4179
DT		 4415 send_lookup_reply(ls, ms, ret_nodeid, error);
4416}
4417
4418static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4419{
c04fecb4
DT		 4420 char name[DLM_RESNAME_MAXLEN+1];
4421 struct dlm_rsb *r;
4422 uint32_t hash, b;
4423 int rv, len, dir_nodeid, from_nodeid;
e7fd4179 4424
3428785a 4425 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
e7fd4179
DT		 4426
4427 len = receive_extralen(ms);
4428
c04fecb4
DT		 4429 if (len > DLM_RESNAME_MAXLEN) {
4430 log_error(ls, "receive_remove from %d bad len %d",
4431 from_nodeid, len);
4432 return;
4433 }
4434
00e99ccd 4435 dir_nodeid = dlm_hash2nodeid(ls, le32_to_cpu(ms->m_hash));
e7fd4179 4436 if (dir_nodeid != dlm_our_nodeid()) {
c04fecb4
DT		 4437 log_error(ls, "receive_remove from %d bad nodeid %d",
4438 from_nodeid, dir_nodeid);
e7fd4179
DT		 4439 return;
4440 }
4441
c04fecb4
DT		 4442 /* Look for name on rsbtbl.toss, if it's there, kill it.
4443 If it's on rsbtbl.keep, it's being used, and we should ignore this
4444 message. This is an expected race between the dir node sending a
4445 request to the master node at the same time as the master node sends
4446 a remove to the dir node. The resolution to that race is for the
4447 dir node to ignore the remove message, and the master node to
4448 recreate the master rsb when it gets a request from the dir node for
4449 an rsb it doesn't have. */
4450
4451 memset(name, 0, sizeof(name));
4452 memcpy(name, ms->m_extra, len);
4453
4454 hash = jhash(name, len, 0);
4455 b = hash & (ls->ls_rsbtbl_size - 1);
4456
4457 spin_lock(&ls->ls_rsbtbl[b].lock);
4458
4459 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4460 if (rv) {
4461 /* verify the rsb is on keep list per comment above */
4462 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4463 if (rv) {
4464 /* should not happen */
4465 log_error(ls, "receive_remove from %d not found %s",
4466 from_nodeid, name);
4467 spin_unlock(&ls->ls_rsbtbl[b].lock);
4468 return;
4469 }
4470 if (r->res_master_nodeid != from_nodeid) {
4471 /* should not happen */
4472 log_error(ls, "receive_remove keep from %d master %d",
4473 from_nodeid, r->res_master_nodeid);
4474 dlm_print_rsb(r);
4475 spin_unlock(&ls->ls_rsbtbl[b].lock);
4476 return;
4477 }
4478
4479 log_debug(ls, "receive_remove from %d master %d first %x %s",
4480 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4481 name);
4482 spin_unlock(&ls->ls_rsbtbl[b].lock);
4483 return;
4484 }
4485
4486 if (r->res_master_nodeid != from_nodeid) {
4487 log_error(ls, "receive_remove toss from %d master %d",
4488 from_nodeid, r->res_master_nodeid);
4489 dlm_print_rsb(r);
4490 spin_unlock(&ls->ls_rsbtbl[b].lock);
4491 return;
4492 }
4493
4494 if (kref_put(&r->res_ref, kill_rsb)) {
4495 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4496 spin_unlock(&ls->ls_rsbtbl[b].lock);
4497 dlm_free_rsb(r);
4498 } else {
4499 log_error(ls, "receive_remove from %d rsb ref error",
4500 from_nodeid);
4501 dlm_print_rsb(r);
4502 spin_unlock(&ls->ls_rsbtbl[b].lock);
4503 }
e7fd4179
DT		 4504}
4505
8499137d
DT		 4506static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4507{
00e99ccd 4508 do_purge(ls, le32_to_cpu(ms->m_nodeid), le32_to_cpu(ms->m_pid));
8499137d
DT		 4509}
4510
6d40c4a7 4511static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4512{
4513 struct dlm_lkb *lkb;
4514 struct dlm_rsb *r;
ef0c2bb0 4515 int error, mstype, result;
3428785a 4516 int from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
e7fd4179 4517
00e99ccd 4518 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
6d40c4a7
DT		 4519 if (error)
4520 return error;
e7fd4179 4521
e7fd4179
DT		 4522 r = lkb->lkb_resource;
4523 hold_rsb(r);
4524 lock_rsb(r);
4525
c54e04b0
DT		 4526 error = validate_message(lkb, ms);
4527 if (error)
4528 goto out;
4529
ef0c2bb0
DT		 4530 mstype = lkb->lkb_wait_type;
4531 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4875647a
DT		 4532 if (error) {
4533 log_error(ls, "receive_request_reply %x remote %d %x result %d",
00e99ccd
AA		 4534 lkb->lkb_id, from_nodeid, le32_to_cpu(ms->m_lkid),
4535 from_dlm_errno(le32_to_cpu(ms->m_result)));
4875647a 4536 dlm_dump_rsb(r);
ef0c2bb0 4537 goto out;
4875647a 4538 }
ef0c2bb0 4539
e7fd4179
DT		 4540 /* Optimization: the dir node was also the master, so it took our
4541 lookup as a request and sent request reply instead of lookup reply */
4542 if (mstype == DLM_MSG_LOOKUP) {
c04fecb4
DT		 4543 r->res_master_nodeid = from_nodeid;
4544 r->res_nodeid = from_nodeid;
4545 lkb->lkb_nodeid = from_nodeid;
e7fd4179
DT		 4546 }
4547
ef0c2bb0 4548 /* this is the value returned from do_request() on the master */
00e99ccd 4549 result = from_dlm_errno(le32_to_cpu(ms->m_result));
ef0c2bb0
DT		 4550
4551 switch (result) {
e7fd4179 4552 case -EAGAIN:
ef0c2bb0 4553 /* request would block (be queued) on remote master */
e7fd4179
DT		 4554 queue_cast(r, lkb, -EAGAIN);
4555 confirm_master(r, -EAGAIN);
ef0c2bb0 4556 unhold_lkb(lkb); /* undoes create_lkb() */
e7fd4179
DT		 4557 break;
4558
4559 case -EINPROGRESS:
4560 case 0:
4561 /* request was queued or granted on remote master */
4562 receive_flags_reply(lkb, ms);
00e99ccd 4563 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
7d3c1feb
DT		 4564 if (is_altmode(lkb))
4565 munge_altmode(lkb, ms);
3ae1acf9 4566 if (result) {
e7fd4179 4567 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3ae1acf9
DT		 4568 add_timeout(lkb);
4569 } else {
e7fd4179
DT		 4570 grant_lock_pc(r, lkb, ms);
4571 queue_cast(r, lkb, 0);
4572 }
ef0c2bb0 4573 confirm_master(r, result);
e7fd4179
DT		 4574 break;
4575
597d0cae 4576 case -EBADR:
e7fd4179
DT		 4577 case -ENOTBLK:
4578 /* find_rsb failed to find rsb or rsb wasn't master */
c04fecb4
DT		 4579 log_limit(ls, "receive_request_reply %x from %d %d "
4580 "master %d dir %d first %x %s", lkb->lkb_id,
4581 from_nodeid, result, r->res_master_nodeid,
4582 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4583
4584 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4585 r->res_master_nodeid != dlm_our_nodeid()) {
4586 /* cause _request_lock->set_master->send_lookup */
4587 r->res_master_nodeid = 0;
4588 r->res_nodeid = -1;
4589 lkb->lkb_nodeid = -1;
4590 }
ef0c2bb0
DT		 4591
4592 if (is_overlap(lkb)) {
4593 /* we'll ignore error in cancel/unlock reply */
4594 queue_cast_overlap(r, lkb);
aec64e1b 4595 confirm_master(r, result);
ef0c2bb0 4596 unhold_lkb(lkb); /* undoes create_lkb() */
c04fecb4 4597 } else {
ef0c2bb0 4598 _request_lock(r, lkb);
c04fecb4
DT		 4599
4600 if (r->res_master_nodeid == dlm_our_nodeid())
4601 confirm_master(r, 0);
4602 }
e7fd4179
DT		 4603 break;
4604
4605 default:
ef0c2bb0
DT		 4606 log_error(ls, "receive_request_reply %x error %d",
4607 lkb->lkb_id, result);
e7fd4179
DT		 4608 }
4609
ef0c2bb0
DT		 4610 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4611 log_debug(ls, "receive_request_reply %x result %d unlock",
4612 lkb->lkb_id, result);
4613 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4614 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4615 send_unlock(r, lkb);
4616 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4617 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4618 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4619 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4620 send_cancel(r, lkb);
4621 } else {
4622 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4623 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4624 }
4625 out:
e7fd4179
DT		 4626 unlock_rsb(r);
4627 put_rsb(r);
b3f58d8f 4628 dlm_put_lkb(lkb);
6d40c4a7 4629 return 0;
e7fd4179
DT		 4630}
4631
4632static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4633 struct dlm_message *ms)
4634{
e7fd4179 4635 /* this is the value returned from do_convert() on the master */
00e99ccd 4636 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
e7fd4179
DT		 4637 case -EAGAIN:
4638 /* convert would block (be queued) on remote master */
4639 queue_cast(r, lkb, -EAGAIN);
4640 break;
4641
c85d65e9
DT		 4642 case -EDEADLK:
4643 receive_flags_reply(lkb, ms);
4644 revert_lock_pc(r, lkb);
4645 queue_cast(r, lkb, -EDEADLK);
4646 break;
4647
e7fd4179
DT		 4648 case -EINPROGRESS:
4649 /* convert was queued on remote master */
7d3c1feb
DT		 4650 receive_flags_reply(lkb, ms);
4651 if (is_demoted(lkb))
2a7ce0ed 4652 munge_demoted(lkb);
e7fd4179
DT		 4653 del_lkb(r, lkb);
4654 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3ae1acf9 4655 add_timeout(lkb);
e7fd4179
DT		 4656 break;
4657
4658 case 0:
4659 /* convert was granted on remote master */
4660 receive_flags_reply(lkb, ms);
7d3c1feb 4661 if (is_demoted(lkb))
2a7ce0ed 4662 munge_demoted(lkb);
e7fd4179
DT		 4663 grant_lock_pc(r, lkb, ms);
4664 queue_cast(r, lkb, 0);
4665 break;
4666
4667 default:
6d40c4a7 4668 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
3428785a 4669 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
00e99ccd
AA		 4670 le32_to_cpu(ms->m_lkid),
4671 from_dlm_errno(le32_to_cpu(ms->m_result)));
6d40c4a7
DT		 4672 dlm_print_rsb(r);
4673 dlm_print_lkb(lkb);
e7fd4179
DT		 4674 }
4675}
4676
4677static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4678{
4679 struct dlm_rsb *r = lkb->lkb_resource;
ef0c2bb0 4680 int error;
e7fd4179
DT		 4681
4682 hold_rsb(r);
4683 lock_rsb(r);
4684
c54e04b0
DT		 4685 error = validate_message(lkb, ms);
4686 if (error)
4687 goto out;
4688
ef0c2bb0
DT		 4689 /* stub reply can happen with waiters_mutex held */
4690 error = remove_from_waiters_ms(lkb, ms);
4691 if (error)
4692 goto out;
e7fd4179 4693
ef0c2bb0
DT		 4694 __receive_convert_reply(r, lkb, ms);
4695 out:
e7fd4179
DT		 4696 unlock_rsb(r);
4697 put_rsb(r);
4698}
4699
6d40c4a7 4700static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4701{
4702 struct dlm_lkb *lkb;
4703 int error;
4704
00e99ccd 4705 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
6d40c4a7
DT		 4706 if (error)
4707 return error;
e7fd4179 4708
e7fd4179 4709 _receive_convert_reply(lkb, ms);
b3f58d8f 4710 dlm_put_lkb(lkb);
6d40c4a7 4711 return 0;
e7fd4179
DT		 4712}
4713
4714static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4715{
4716 struct dlm_rsb *r = lkb->lkb_resource;
ef0c2bb0 4717 int error;
e7fd4179
DT		 4718
4719 hold_rsb(r);
4720 lock_rsb(r);
4721
c54e04b0
DT		 4722 error = validate_message(lkb, ms);
4723 if (error)
4724 goto out;
4725
ef0c2bb0
DT		 4726 /* stub reply can happen with waiters_mutex held */
4727 error = remove_from_waiters_ms(lkb, ms);
4728 if (error)
4729 goto out;
4730
e7fd4179
DT		 4731 /* this is the value returned from do_unlock() on the master */
4732
00e99ccd 4733 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
e7fd4179
DT		 4734 case -DLM_EUNLOCK:
4735 receive_flags_reply(lkb, ms);
4736 remove_lock_pc(r, lkb);
4737 queue_cast(r, lkb, -DLM_EUNLOCK);
4738 break;
ef0c2bb0
DT		 4739 case -ENOENT:
4740 break;
e7fd4179 4741 default:
ef0c2bb0 4742 log_error(r->res_ls, "receive_unlock_reply %x error %d",
00e99ccd 4743 lkb->lkb_id, from_dlm_errno(le32_to_cpu(ms->m_result)));
e7fd4179 4744 }
ef0c2bb0 4745 out:
e7fd4179
DT		 4746 unlock_rsb(r);
4747 put_rsb(r);
4748}
4749
6d40c4a7 4750static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4751{
4752 struct dlm_lkb *lkb;
4753 int error;
4754
00e99ccd 4755 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
6d40c4a7
DT		 4756 if (error)
4757 return error;
e7fd4179 4758
e7fd4179 4759 _receive_unlock_reply(lkb, ms);
b3f58d8f 4760 dlm_put_lkb(lkb);
6d40c4a7 4761 return 0;
e7fd4179
DT		 4762}
4763
4764static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4765{
4766 struct dlm_rsb *r = lkb->lkb_resource;
ef0c2bb0 4767 int error;
e7fd4179
DT		 4768
4769 hold_rsb(r);
4770 lock_rsb(r);
4771
c54e04b0
DT		 4772 error = validate_message(lkb, ms);
4773 if (error)
4774 goto out;
4775
ef0c2bb0
DT		 4776 /* stub reply can happen with waiters_mutex held */
4777 error = remove_from_waiters_ms(lkb, ms);
4778 if (error)
4779 goto out;
4780
e7fd4179
DT		 4781 /* this is the value returned from do_cancel() on the master */
4782
00e99ccd 4783 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
e7fd4179
DT		 4784 case -DLM_ECANCEL:
4785 receive_flags_reply(lkb, ms);
4786 revert_lock_pc(r, lkb);
84d8cd69 4787 queue_cast(r, lkb, -DLM_ECANCEL);
ef0c2bb0
DT		 4788 break;
4789 case 0:
e7fd4179
DT		 4790 break;
4791 default:
ef0c2bb0 4792 log_error(r->res_ls, "receive_cancel_reply %x error %d",
00e99ccd
AA		 4793 lkb->lkb_id,
4794 from_dlm_errno(le32_to_cpu(ms->m_result)));
e7fd4179 4795 }
ef0c2bb0 4796 out:
e7fd4179
DT		 4797 unlock_rsb(r);
4798 put_rsb(r);
4799}
4800
6d40c4a7 4801static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
e7fd4179
DT		 4802{
4803 struct dlm_lkb *lkb;
4804 int error;
4805
00e99ccd 4806 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
6d40c4a7
DT		 4807 if (error)
4808 return error;
e7fd4179 4809
e7fd4179 4810 _receive_cancel_reply(lkb, ms);
b3f58d8f 4811 dlm_put_lkb(lkb);
6d40c4a7 4812 return 0;
e7fd4179
DT		 4813}
4814
4815static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4816{
4817 struct dlm_lkb *lkb;
4818 struct dlm_rsb *r;
4819 int error, ret_nodeid;
c04fecb4 4820 int do_lookup_list = 0;
e7fd4179 4821
00e99ccd 4822 error = find_lkb(ls, le32_to_cpu(ms->m_lkid), &lkb);
e7fd4179 4823 if (error) {
00e99ccd
AA		 4824 log_error(ls, "%s no lkid %x", __func__,
4825 le32_to_cpu(ms->m_lkid));
e7fd4179
DT		 4826 return;
4827 }
4828
c04fecb4 4829 /* ms->m_result is the value returned by dlm_master_lookup on dir node
e7fd4179 4830 FIXME: will a non-zero error ever be returned? */
e7fd4179
DT		 4831
4832 r = lkb->lkb_resource;
4833 hold_rsb(r);
4834 lock_rsb(r);
4835
ef0c2bb0
DT		 4836 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4837 if (error)
4838 goto out;
4839
00e99ccd 4840 ret_nodeid = le32_to_cpu(ms->m_nodeid);
c04fecb4
DT		 4841
4842 /* We sometimes receive a request from the dir node for this
4843 rsb before we've received the dir node's loookup_reply for it.
4844 The request from the dir node implies we're the master, so we set
4845 ourself as master in receive_request_reply, and verify here that
4846 we are indeed the master. */
4847
4848 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4849 /* This should never happen */
4850 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4851 "master %d dir %d our %d first %x %s",
3428785a
AA		 4852 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4853 ret_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
c04fecb4
DT		 4854 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4855 }
4856
e7fd4179 4857 if (ret_nodeid == dlm_our_nodeid()) {
c04fecb4 4858 r->res_master_nodeid = ret_nodeid;
e7fd4179 4859 r->res_nodeid = 0;
c04fecb4 4860 do_lookup_list = 1;
e7fd4179 4861 r->res_first_lkid = 0;
c04fecb4
DT		 4862 } else if (ret_nodeid == -1) {
4863 /* the remote node doesn't believe it's the dir node */
4864 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
3428785a 4865 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid));
c04fecb4
DT		 4866 r->res_master_nodeid = 0;
4867 r->res_nodeid = -1;
4868 lkb->lkb_nodeid = -1;
e7fd4179 4869 } else {
c04fecb4
DT		 4870 /* set_master() will set lkb_nodeid from r */
4871 r->res_master_nodeid = ret_nodeid;
e7fd4179
DT		 4872 r->res_nodeid = ret_nodeid;
4873 }
4874
ef0c2bb0
DT		 4875 if (is_overlap(lkb)) {
4876 log_debug(ls, "receive_lookup_reply %x unlock %x",
4877 lkb->lkb_id, lkb->lkb_flags);
4878 queue_cast_overlap(r, lkb);
4879 unhold_lkb(lkb); /* undoes create_lkb() */
4880 goto out_list;
4881 }
4882
e7fd4179
DT		 4883 _request_lock(r, lkb);
4884
ef0c2bb0 4885 out_list:
c04fecb4 4886 if (do_lookup_list)
e7fd4179 4887 process_lookup_list(r);
ef0c2bb0 4888 out:
e7fd4179
DT		 4889 unlock_rsb(r);
4890 put_rsb(r);
b3f58d8f 4891 dlm_put_lkb(lkb);
e7fd4179
DT		 4892}
4893
6d40c4a7
DT		 4894static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4895 uint32_t saved_seq)
e7fd4179 4896{
6d40c4a7
DT		 4897 int error = 0, noent = 0;
4898
3428785a 4899 if (!dlm_is_member(ls, le32_to_cpu(ms->m_header.h_nodeid))) {
c04fecb4 4900 log_limit(ls, "receive %d from non-member %d %x %x %d",
00e99ccd
AA		 4901 le32_to_cpu(ms->m_type),
4902 le32_to_cpu(ms->m_header.h_nodeid),
4903 le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4904 from_dlm_errno(le32_to_cpu(ms->m_result)));
46b43eed
DT		 4905 return;
4906 }
4907
e7fd4179
DT		 4908 switch (ms->m_type) {
4909
4910 /* messages sent to a master node */
4911
00e99ccd 4912 case cpu_to_le32(DLM_MSG_REQUEST):
6d40c4a7 4913 error = receive_request(ls, ms);
e7fd4179
DT		 4914 break;
4915
00e99ccd 4916 case cpu_to_le32(DLM_MSG_CONVERT):
6d40c4a7 4917 error = receive_convert(ls, ms);
e7fd4179
DT		 4918 break;
4919
00e99ccd 4920 case cpu_to_le32(DLM_MSG_UNLOCK):
6d40c4a7 4921 error = receive_unlock(ls, ms);
e7fd4179
DT		 4922 break;
4923
00e99ccd 4924 case cpu_to_le32(DLM_MSG_CANCEL):
6d40c4a7
DT		 4925 noent = 1;
4926 error = receive_cancel(ls, ms);
e7fd4179
DT		 4927 break;
4928
4929 /* messages sent from a master node (replies to above) */
4930
00e99ccd 4931 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
6d40c4a7 4932 error = receive_request_reply(ls, ms);
e7fd4179
DT		 4933 break;
4934
00e99ccd 4935 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
6d40c4a7 4936 error = receive_convert_reply(ls, ms);
e7fd4179
DT		 4937 break;
4938
00e99ccd 4939 case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
6d40c4a7 4940 error = receive_unlock_reply(ls, ms);
e7fd4179
DT		 4941 break;
4942
00e99ccd 4943 case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
6d40c4a7 4944 error = receive_cancel_reply(ls, ms);
e7fd4179
DT		 4945 break;
4946
4947 /* messages sent from a master node (only two types of async msg) */
4948
00e99ccd 4949 case cpu_to_le32(DLM_MSG_GRANT):
6d40c4a7
DT		 4950 noent = 1;
4951 error = receive_grant(ls, ms);
e7fd4179
DT		 4952 break;
4953
00e99ccd 4954 case cpu_to_le32(DLM_MSG_BAST):
6d40c4a7
DT		 4955 noent = 1;
4956 error = receive_bast(ls, ms);
e7fd4179
DT		 4957 break;
4958
4959 /* messages sent to a dir node */
4960
00e99ccd 4961 case cpu_to_le32(DLM_MSG_LOOKUP):
e7fd4179
DT		 4962 receive_lookup(ls, ms);
4963 break;
4964
00e99ccd 4965 case cpu_to_le32(DLM_MSG_REMOVE):
e7fd4179
DT		 4966 receive_remove(ls, ms);
4967 break;
4968
4969 /* messages sent from a dir node (remove has no reply) */
4970
00e99ccd 4971 case cpu_to_le32(DLM_MSG_LOOKUP_REPLY):
e7fd4179
DT		 4972 receive_lookup_reply(ls, ms);
4973 break;
4974
8499137d
DT		 4975 /* other messages */
4976
00e99ccd 4977 case cpu_to_le32(DLM_MSG_PURGE):
8499137d
DT		 4978 receive_purge(ls, ms);
4979 break;
4980
e7fd4179 4981 default:
00e99ccd
AA		 4982 log_error(ls, "unknown message type %d",
4983 le32_to_cpu(ms->m_type));
e7fd4179 4984 }
6d40c4a7
DT		 4985
4986 /*
4987 * When checking for ENOENT, we're checking the result of
4988 * find_lkb(m_remid):
4989 *
4990 * The lock id referenced in the message wasn't found. This may
4991 * happen in normal usage for the async messages and cancel, so
4992 * only use log_debug for them.
4993 *
4875647a 4994 * Some errors are expected and normal.
6d40c4a7
DT		 4995 */
4996
4997 if (error == -ENOENT && noent) {
4875647a 4998 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
00e99ccd 4999 le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
3428785a 5000 le32_to_cpu(ms->m_header.h_nodeid),
00e99ccd 5001 le32_to_cpu(ms->m_lkid), saved_seq);
6d40c4a7 5002 } else if (error == -ENOENT) {
4875647a 5003 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
00e99ccd 5004 le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
3428785a 5005 le32_to_cpu(ms->m_header.h_nodeid),
00e99ccd 5006 le32_to_cpu(ms->m_lkid), saved_seq);
6d40c4a7 5007
00e99ccd
AA		 5008 if (ms->m_type == cpu_to_le32(DLM_MSG_CONVERT))
5009 dlm_dump_rsb_hash(ls, le32_to_cpu(ms->m_hash));
6d40c4a7 5010 }
4875647a
DT		 5011
5012 if (error == -EINVAL) {
5013 log_error(ls, "receive %d inval from %d lkid %x remid %x "
5014 "saved_seq %u",
00e99ccd
AA		 5015 le32_to_cpu(ms->m_type),
5016 le32_to_cpu(ms->m_header.h_nodeid),
5017 le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
5018 saved_seq);
4875647a 5019 }
e7fd4179
DT		 5020}
5021
c36258b5
DT		 5022/* If the lockspace is in recovery mode (locking stopped), then normal
5023 messages are saved on the requestqueue for processing after recovery is
5024 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
5025 messages off the requestqueue before we process new ones. This occurs right
5026 after recovery completes when we transition from saving all messages on
5027 requestqueue, to processing all the saved messages, to processing new
5028 messages as they arrive. */
e7fd4179 5029
c36258b5
DT		 5030static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
5031 int nodeid)
5032{
5033 if (dlm_locking_stopped(ls)) {
c04fecb4
DT		 5034 /* If we were a member of this lockspace, left, and rejoined,
5035 other nodes may still be sending us messages from the
5036 lockspace generation before we left. */
5037 if (!ls->ls_generation) {
5038 log_limit(ls, "receive %d from %d ignore old gen",
00e99ccd 5039 le32_to_cpu(ms->m_type), nodeid);
c04fecb4
DT		 5040 return;
5041 }
5042
8b0d8e03 5043 dlm_add_requestqueue(ls, nodeid, ms);
c36258b5
DT		 5044 } else {
5045 dlm_wait_requestqueue(ls);
6d40c4a7 5046 _receive_message(ls, ms, 0);
c36258b5
DT		 5047 }
5048}
5049
5050/* This is called by dlm_recoverd to process messages that were saved on
5051 the requestqueue. */
5052
6d40c4a7
DT		 5053void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
5054 uint32_t saved_seq)
c36258b5 5055{
6d40c4a7 5056 _receive_message(ls, ms, saved_seq);
c36258b5
DT		 5057}
5058
5059/* This is called by the midcomms layer when something is received for
5060 the lockspace. It could be either a MSG (normal message sent as part of
5061 standard locking activity) or an RCOM (recovery message sent as part of
5062 lockspace recovery). */
5063
eef7d739 5064void dlm_receive_buffer(union dlm_packet *p, int nodeid)
c36258b5 5065{
eef7d739 5066 struct dlm_header *hd = &p->header;
c36258b5
DT		 5067 struct dlm_ls *ls;
5068 int type = 0;
5069
5070 switch (hd->h_cmd) {
5071 case DLM_MSG:
00e99ccd 5072 type = le32_to_cpu(p->message.m_type);
c36258b5
DT		 5073 break;
5074 case DLM_RCOM:
2f9dbeda 5075 type = le32_to_cpu(p->rcom.rc_type);
c36258b5
DT		 5076 break;
5077 default:
5078 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5079 return;
5080 }
5081
3428785a 5082 if (le32_to_cpu(hd->h_nodeid) != nodeid) {
c36258b5 5083 log_print("invalid h_nodeid %d from %d lockspace %x",
3428785a
AA		 5084 le32_to_cpu(hd->h_nodeid), nodeid,
5085 le32_to_cpu(hd->u.h_lockspace));
c36258b5
DT		 5086 return;
5087 }
5088
3428785a 5089 ls = dlm_find_lockspace_global(le32_to_cpu(hd->u.h_lockspace));
c36258b5 5090 if (!ls) {
4875647a
DT		 5091 if (dlm_config.ci_log_debug) {
5092 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5093 "%u from %d cmd %d type %d\n",
3428785a
AA		 5094 le32_to_cpu(hd->u.h_lockspace), nodeid,
5095 hd->h_cmd, type);
4875647a 5096 }
c36258b5
DT		 5097
5098 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
eef7d739 5099 dlm_send_ls_not_ready(nodeid, &p->rcom);
c36258b5
DT		 5100 return;
5101 }
5102
5103 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5104 be inactive (in this ls) before transitioning to recovery mode */
5105
5106 down_read(&ls->ls_recv_active);
5107 if (hd->h_cmd == DLM_MSG)
eef7d739 5108 dlm_receive_message(ls, &p->message, nodeid);
c36258b5 5109 else
eef7d739 5110 dlm_receive_rcom(ls, &p->rcom, nodeid);
c36258b5
DT		 5111 up_read(&ls->ls_recv_active);
5112
5113 dlm_put_lockspace(ls);
5114}
e7fd4179 5115
2a7ce0ed
DT		 5116static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5117 struct dlm_message *ms_stub)
e7fd4179
DT		 5118{
5119 if (middle_conversion(lkb)) {
5120 hold_lkb(lkb);
2a7ce0ed 5121 memset(ms_stub, 0, sizeof(struct dlm_message));
00e99ccd
AA		 5122 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5123 ms_stub->m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
5124 ms_stub->m_result = cpu_to_le32(to_dlm_errno(-EINPROGRESS));
3428785a 5125 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
2a7ce0ed 5126 _receive_convert_reply(lkb, ms_stub);
e7fd4179
DT		 5127
5128 /* Same special case as in receive_rcom_lock_args() */
5129 lkb->lkb_grmode = DLM_LOCK_IV;
5130 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5131 unhold_lkb(lkb);
5132
5133 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5134 lkb->lkb_flags |= DLM_IFL_RESEND;
5135 }
5136
5137 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5138 conversions are async; there's no reply from the remote master */
5139}
5140
5141/* A waiting lkb needs recovery if the master node has failed, or
5142 the master node is changing (only when no directory is used) */
5143
13ef1111
DT		 5144static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5145 int dir_nodeid)
e7fd4179 5146{
4875647a 5147 if (dlm_no_directory(ls))
13ef1111
DT		 5148 return 1;
5149
4875647a 5150 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
e7fd4179
DT		 5151 return 1;
5152
5153 return 0;
5154}
5155
5156/* Recovery for locks that are waiting for replies from nodes that are now
5157 gone. We can just complete unlocks and cancels by faking a reply from the
5158 dead node. Requests and up-conversions we flag to be resent after
5159 recovery. Down-conversions can just be completed with a fake reply like
5160 unlocks. Conversions between PR and CW need special attention. */
5161
5162void dlm_recover_waiters_pre(struct dlm_ls *ls)
5163{
5164 struct dlm_lkb *lkb, *safe;
2a7ce0ed 5165 struct dlm_message *ms_stub;
601342ce 5166 int wait_type, stub_unlock_result, stub_cancel_result;
13ef1111 5167 int dir_nodeid;
e7fd4179 5168
102e67d4 5169 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
0d37eca7 5170 if (!ms_stub)
2a7ce0ed 5171 return;
2a7ce0ed 5172
90135925 5173 mutex_lock(&ls->ls_waiters_mutex);
e7fd4179
DT		 5174
5175 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
2a7ce0ed 5176
13ef1111
DT		 5177 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5178
2a7ce0ed
DT		 5179 /* exclude debug messages about unlocks because there can be so
5180 many and they aren't very interesting */
5181
5182 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
13ef1111
DT		 5183 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5184 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5185 lkb->lkb_id,
5186 lkb->lkb_remid,
5187 lkb->lkb_wait_type,
5188 lkb->lkb_resource->res_nodeid,
5189 lkb->lkb_nodeid,
5190 lkb->lkb_wait_nodeid,
5191 dir_nodeid);
2a7ce0ed 5192 }
e7fd4179
DT		 5193
5194 /* all outstanding lookups, regardless of destination will be
5195 resent after recovery is done */
5196
5197 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5198 lkb->lkb_flags |= DLM_IFL_RESEND;
5199 continue;
5200 }
5201
13ef1111 5202 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
e7fd4179
DT		 5203 continue;
5204
601342ce
DT		 5205 wait_type = lkb->lkb_wait_type;
5206 stub_unlock_result = -DLM_EUNLOCK;
5207 stub_cancel_result = -DLM_ECANCEL;
5208
5209 /* Main reply may have been received leaving a zero wait_type,
5210 but a reply for the overlapping op may not have been
5211 received. In that case we need to fake the appropriate
5212 reply for the overlap op. */
5213
5214 if (!wait_type) {
5215 if (is_overlap_cancel(lkb)) {
5216 wait_type = DLM_MSG_CANCEL;
5217 if (lkb->lkb_grmode == DLM_LOCK_IV)
5218 stub_cancel_result = 0;
5219 }
5220 if (is_overlap_unlock(lkb)) {
5221 wait_type = DLM_MSG_UNLOCK;
5222 if (lkb->lkb_grmode == DLM_LOCK_IV)
5223 stub_unlock_result = -ENOENT;
5224 }
5225
5226 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5227 lkb->lkb_id, lkb->lkb_flags, wait_type,
5228 stub_cancel_result, stub_unlock_result);
5229 }
5230
5231 switch (wait_type) {
e7fd4179
DT		 5232
5233 case DLM_MSG_REQUEST:
5234 lkb->lkb_flags |= DLM_IFL_RESEND;
5235 break;
5236
5237 case DLM_MSG_CONVERT:
2a7ce0ed 5238 recover_convert_waiter(ls, lkb, ms_stub);
e7fd4179
DT		 5239 break;
5240
5241 case DLM_MSG_UNLOCK:
5242 hold_lkb(lkb);
2a7ce0ed 5243 memset(ms_stub, 0, sizeof(struct dlm_message));
00e99ccd
AA		 5244 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5245 ms_stub->m_type = cpu_to_le32(DLM_MSG_UNLOCK_REPLY);
5246 ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_unlock_result));
3428785a 5247 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
2a7ce0ed 5248 _receive_unlock_reply(lkb, ms_stub);
b3f58d8f 5249 dlm_put_lkb(lkb);
e7fd4179
DT		 5250 break;
5251
5252 case DLM_MSG_CANCEL:
5253 hold_lkb(lkb);
2a7ce0ed 5254 memset(ms_stub, 0, sizeof(struct dlm_message));
00e99ccd
AA		 5255 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5256 ms_stub->m_type = cpu_to_le32(DLM_MSG_CANCEL_REPLY);
5257 ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_cancel_result));
3428785a 5258 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
2a7ce0ed 5259 _receive_cancel_reply(lkb, ms_stub);
b3f58d8f 5260 dlm_put_lkb(lkb);
e7fd4179
DT		 5261 break;
5262
5263 default:
601342ce
DT		 5264 log_error(ls, "invalid lkb wait_type %d %d",
5265 lkb->lkb_wait_type, wait_type);
e7fd4179 5266 }
81456807 5267 schedule();
e7fd4179 5268 }
90135925 5269 mutex_unlock(&ls->ls_waiters_mutex);
2a7ce0ed 5270 kfree(ms_stub);
e7fd4179
DT		 5271}
5272
ef0c2bb0 5273static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
e7fd4179
DT		 5274{
5275 struct dlm_lkb *lkb;
ef0c2bb0 5276 int found = 0;
e7fd4179 5277
90135925 5278 mutex_lock(&ls->ls_waiters_mutex);
e7fd4179
DT		 5279 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
5280 if (lkb->lkb_flags & DLM_IFL_RESEND) {
ef0c2bb0
DT		 5281 hold_lkb(lkb);
5282 found = 1;
e7fd4179
DT		 5283 break;
5284 }
5285 }
90135925 5286 mutex_unlock(&ls->ls_waiters_mutex);
e7fd4179 5287
ef0c2bb0 5288 if (!found)
e7fd4179 5289 lkb = NULL;
ef0c2bb0 5290 return lkb;
e7fd4179
DT		 5291}
5292
5293/* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
5294 master or dir-node for r. Processing the lkb may result in it being placed
5295 back on waiters. */
5296
ef0c2bb0
DT		 5297/* We do this after normal locking has been enabled and any saved messages
5298 (in requestqueue) have been processed. We should be confident that at
5299 this point we won't get or process a reply to any of these waiting
5300 operations. But, new ops may be coming in on the rsbs/locks here from
5301 userspace or remotely. */
5302
5303/* there may have been an overlap unlock/cancel prior to recovery or after
5304 recovery. if before, the lkb may still have a pos wait_count; if after, the
5305 overlap flag would just have been set and nothing new sent. we can be
5306 confident here than any replies to either the initial op or overlap ops
5307 prior to recovery have been received. */
5308
e7fd4179
DT		 5309int dlm_recover_waiters_post(struct dlm_ls *ls)
5310{
5311 struct dlm_lkb *lkb;
5312 struct dlm_rsb *r;
ef0c2bb0 5313 int error = 0, mstype, err, oc, ou;
e7fd4179
DT		 5314
5315 while (1) {
5316 if (dlm_locking_stopped(ls)) {
5317 log_debug(ls, "recover_waiters_post aborted");
5318 error = -EINTR;
5319 break;
5320 }
5321
ef0c2bb0
DT		 5322 lkb = find_resend_waiter(ls);
5323 if (!lkb)
e7fd4179
DT		 5324 break;
5325
5326 r = lkb->lkb_resource;
ef0c2bb0
DT		 5327 hold_rsb(r);
5328 lock_rsb(r);
5329
5330 mstype = lkb->lkb_wait_type;
5331 oc = is_overlap_cancel(lkb);
5332 ou = is_overlap_unlock(lkb);
5333 err = 0;
e7fd4179 5334
13ef1111
DT		 5335 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5336 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5337 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5338 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5339 dlm_dir_nodeid(r), oc, ou);
e7fd4179 5340
ef0c2bb0
DT		 5341 /* At this point we assume that we won't get a reply to any
5342 previous op or overlap op on this lock. First, do a big
5343 remove_from_waiters() for all previous ops. */
5344
5345 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5346 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5347 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5348 lkb->lkb_wait_type = 0;
5349 lkb->lkb_wait_count = 0;
5350 mutex_lock(&ls->ls_waiters_mutex);
5351 list_del_init(&lkb->lkb_wait_reply);
5352 mutex_unlock(&ls->ls_waiters_mutex);
5353 unhold_lkb(lkb); /* for waiters list */
5354
5355 if (oc || ou) {
5356 /* do an unlock or cancel instead of resending */
5357 switch (mstype) {
5358 case DLM_MSG_LOOKUP:
5359 case DLM_MSG_REQUEST:
5360 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5361 -DLM_ECANCEL);
5362 unhold_lkb(lkb); /* undoes create_lkb() */
5363 break;
5364 case DLM_MSG_CONVERT:
5365 if (oc) {
5366 queue_cast(r, lkb, -DLM_ECANCEL);
5367 } else {
5368 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5369 _unlock_lock(r, lkb);
5370 }
5371 break;
5372 default:
5373 err = 1;
5374 }
5375 } else {
5376 switch (mstype) {
5377 case DLM_MSG_LOOKUP:
5378 case DLM_MSG_REQUEST:
5379 _request_lock(r, lkb);
5380 if (is_master(r))
5381 confirm_master(r, 0);
5382 break;
5383 case DLM_MSG_CONVERT:
5384 _convert_lock(r, lkb);
5385 break;
5386 default:
5387 err = 1;
5388 }
e7fd4179 5389 }
ef0c2bb0 5390
13ef1111
DT		 5391 if (err) {
5392 log_error(ls, "waiter %x msg %d r_nodeid %d "
5393 "dir_nodeid %d overlap %d %d",
5394 lkb->lkb_id, mstype, r->res_nodeid,
5395 dlm_dir_nodeid(r), oc, ou);
5396 }
ef0c2bb0
DT		 5397 unlock_rsb(r);
5398 put_rsb(r);
5399 dlm_put_lkb(lkb);
e7fd4179
DT		 5400 }
5401
5402 return error;
5403}
5404
4875647a
DT		 5405static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5406 struct list_head *list)
e7fd4179 5407{
e7fd4179
DT		 5408 struct dlm_lkb *lkb, *safe;
5409
4875647a
DT		 5410 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5411 if (!is_master_copy(lkb))
5412 continue;
5413
5414 /* don't purge lkbs we've added in recover_master_copy for
5415 the current recovery seq */
5416
5417 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5418 continue;
5419
5420 del_lkb(r, lkb);
5421
5422 /* this put should free the lkb */
5423 if (!dlm_put_lkb(lkb))
5424 log_error(ls, "purged mstcpy lkb not released");
e7fd4179
DT		 5425 }
5426}
5427
4875647a 5428void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
e7fd4179 5429{
4875647a 5430 struct dlm_ls *ls = r->res_ls;
e7fd4179 5431
4875647a
DT		 5432 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5433 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5434 purge_mstcpy_list(ls, r, &r->res_waitqueue);
e7fd4179
DT		 5435}
5436
4875647a
DT		 5437static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5438 struct list_head *list,
5439 int nodeid_gone, unsigned int *count)
e7fd4179 5440{
4875647a 5441 struct dlm_lkb *lkb, *safe;
e7fd4179 5442
4875647a
DT		 5443 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5444 if (!is_master_copy(lkb))
5445 continue;
5446
5447 if ((lkb->lkb_nodeid == nodeid_gone) ||
5448 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5449
da8c6663
DT		 5450 /* tell recover_lvb to invalidate the lvb
5451 because a node holding EX/PW failed */
5452 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5453 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5454 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5455 }
5456
4875647a
DT		 5457 del_lkb(r, lkb);
5458
5459 /* this put should free the lkb */
5460 if (!dlm_put_lkb(lkb))
5461 log_error(ls, "purged dead lkb not released");
5462
5463 rsb_set_flag(r, RSB_RECOVER_GRANT);
5464
5465 (*count)++;
5466 }
5467 }
e7fd4179
DT		 5468}
5469
5470/* Get rid of locks held by nodes that are gone. */
5471
4875647a 5472void dlm_recover_purge(struct dlm_ls *ls)
e7fd4179
DT		 5473{
5474 struct dlm_rsb *r;
4875647a
DT		 5475 struct dlm_member *memb;
5476 int nodes_count = 0;
5477 int nodeid_gone = 0;
5478 unsigned int lkb_count = 0;
5479
5480 /* cache one removed nodeid to optimize the common
5481 case of a single node removed */
5482
5483 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5484 nodes_count++;
5485 nodeid_gone = memb->nodeid;
5486 }
e7fd4179 5487
4875647a
DT		 5488 if (!nodes_count)
5489 return;
e7fd4179
DT		 5490
5491 down_write(&ls->ls_root_sem);
5492 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5493 hold_rsb(r);
5494 lock_rsb(r);
4875647a
DT		 5495 if (is_master(r)) {
5496 purge_dead_list(ls, r, &r->res_grantqueue,
5497 nodeid_gone, &lkb_count);
5498 purge_dead_list(ls, r, &r->res_convertqueue,
5499 nodeid_gone, &lkb_count);
5500 purge_dead_list(ls, r, &r->res_waitqueue,
5501 nodeid_gone, &lkb_count);
5502 }
e7fd4179
DT		 5503 unlock_rsb(r);
5504 unhold_rsb(r);
4875647a 5505 cond_resched();
e7fd4179
DT		 5506 }
5507 up_write(&ls->ls_root_sem);
5508
4875647a 5509 if (lkb_count)
075f0177 5510 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
4875647a 5511 lkb_count, nodes_count);
e7fd4179
DT		 5512}
5513
4875647a 5514static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
97a35d1e 5515{
9beb3bf5 5516 struct rb_node *n;
4875647a 5517 struct dlm_rsb *r;
97a35d1e 5518
c7be761a 5519 spin_lock(&ls->ls_rsbtbl[bucket].lock);
9beb3bf5
BP		 5520 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5521 r = rb_entry(n, struct dlm_rsb, res_hashnode);
4875647a
DT		 5522
5523 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5524 continue;
c503a621
DT		 5525 if (!is_master(r)) {
5526 rsb_clear_flag(r, RSB_RECOVER_GRANT);
97a35d1e 5527 continue;
c503a621 5528 }
97a35d1e 5529 hold_rsb(r);
4875647a
DT		 5530 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5531 return r;
97a35d1e 5532 }
c7be761a 5533 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
4875647a 5534 return NULL;
97a35d1e
DT		 5535}
5536
4875647a
DT		 5537/*
5538 * Attempt to grant locks on resources that we are the master of.
5539 * Locks may have become grantable during recovery because locks
5540 * from departed nodes have been purged (or not rebuilt), allowing
5541 * previously blocked locks to now be granted. The subset of rsb's
5542 * we are interested in are those with lkb's on either the convert or
5543 * waiting queues.
5544 *
5545 * Simplest would be to go through each master rsb and check for non-empty
5546 * convert or waiting queues, and attempt to grant on those rsbs.
5547 * Checking the queues requires lock_rsb, though, for which we'd need
5548 * to release the rsbtbl lock. This would make iterating through all
5549 * rsb's very inefficient. So, we rely on earlier recovery routines
5550 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5551 * locks for.
5552 */
5553
5554void dlm_recover_grant(struct dlm_ls *ls)
e7fd4179
DT		 5555{
5556 struct dlm_rsb *r;
2b4e926a 5557 int bucket = 0;
4875647a
DT		 5558 unsigned int count = 0;
5559 unsigned int rsb_count = 0;
5560 unsigned int lkb_count = 0;
e7fd4179 5561
2b4e926a 5562 while (1) {
4875647a 5563 r = find_grant_rsb(ls, bucket);
2b4e926a
DT		 5564 if (!r) {
5565 if (bucket == ls->ls_rsbtbl_size - 1)
5566 break;
5567 bucket++;
97a35d1e 5568 continue;
2b4e926a 5569 }
4875647a
DT		 5570 rsb_count++;
5571 count = 0;
97a35d1e 5572 lock_rsb(r);
c503a621 5573 /* the RECOVER_GRANT flag is checked in the grant path */
4875647a 5574 grant_pending_locks(r, &count);
c503a621 5575 rsb_clear_flag(r, RSB_RECOVER_GRANT);
4875647a
DT		 5576 lkb_count += count;
5577 confirm_master(r, 0);
97a35d1e
DT		 5578 unlock_rsb(r);
5579 put_rsb(r);
4875647a 5580 cond_resched();
e7fd4179 5581 }
4875647a
DT		 5582
5583 if (lkb_count)
075f0177 5584 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
4875647a 5585 lkb_count, rsb_count);
e7fd4179
DT		 5586}
5587
5588static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5589 uint32_t remid)
5590{
5591 struct dlm_lkb *lkb;
5592
5593 list_for_each_entry(lkb, head, lkb_statequeue) {
5594 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5595 return lkb;
5596 }
5597 return NULL;
5598}
5599
5600static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5601 uint32_t remid)
5602{
5603 struct dlm_lkb *lkb;
5604
5605 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5606 if (lkb)
5607 return lkb;
5608 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5609 if (lkb)
5610 return lkb;
5611 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5612 if (lkb)
5613 return lkb;
5614 return NULL;
5615}
5616
ae773d0b 5617/* needs at least dlm_rcom + rcom_lock */
e7fd4179
DT		 5618static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5619 struct dlm_rsb *r, struct dlm_rcom *rc)
5620{
5621 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
e7fd4179 5622
3428785a 5623 lkb->lkb_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
163a1859
AV		 5624 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5625 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5626 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5627 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
e7fd4179 5628 lkb->lkb_flags |= DLM_IFL_MSTCPY;
163a1859 5629 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
e7fd4179
DT		 5630 lkb->lkb_rqmode = rl->rl_rqmode;
5631 lkb->lkb_grmode = rl->rl_grmode;
5632 /* don't set lkb_status because add_lkb wants to itself */
5633
8304d6f2
DT		 5634 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5635 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
e7fd4179 5636
e7fd4179 5637 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3428785a
AA		 5638 int lvblen = le16_to_cpu(rc->rc_header.h_length) -
5639 sizeof(struct dlm_rcom) - sizeof(struct rcom_lock);
a5dd0631
AV		 5640 if (lvblen > ls->ls_lvblen)
5641 return -EINVAL;
52bda2b5 5642 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
e7fd4179
DT		 5643 if (!lkb->lkb_lvbptr)
5644 return -ENOMEM;
e7fd4179
DT		 5645 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5646 }
5647
5648 /* Conversions between PR and CW (middle modes) need special handling.
5649 The real granted mode of these converting locks cannot be determined
5650 until all locks have been rebuilt on the rsb (recover_conversion) */
5651
163a1859
AV		 5652 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5653 middle_conversion(lkb)) {
e7fd4179
DT		 5654 rl->rl_status = DLM_LKSTS_CONVERT;
5655 lkb->lkb_grmode = DLM_LOCK_IV;
5656 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5657 }
5658
5659 return 0;
5660}
5661
5662/* This lkb may have been recovered in a previous aborted recovery so we need
5663 to check if the rsb already has an lkb with the given remote nodeid/lkid.
5664 If so we just send back a standard reply. If not, we create a new lkb with
5665 the given values and send back our lkid. We send back our lkid by sending
5666 back the rcom_lock struct we got but with the remid field filled in. */
5667
ae773d0b 5668/* needs at least dlm_rcom + rcom_lock */
e7fd4179
DT		 5669int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5670{
5671 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5672 struct dlm_rsb *r;
5673 struct dlm_lkb *lkb;
6d40c4a7 5674 uint32_t remid = 0;
3428785a 5675 int from_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
e7fd4179
DT		 5676 int error;
5677
5678 if (rl->rl_parent_lkid) {
5679 error = -EOPNOTSUPP;
5680 goto out;
5681 }
5682
6d40c4a7
DT		 5683 remid = le32_to_cpu(rl->rl_lkid);
5684
4875647a
DT		 5685 /* In general we expect the rsb returned to be R_MASTER, but we don't
5686 have to require it. Recovery of masters on one node can overlap
5687 recovery of locks on another node, so one node can send us MSTCPY
5688 locks before we've made ourselves master of this rsb. We can still
5689 add new MSTCPY locks that we receive here without any harm; when
5690 we make ourselves master, dlm_recover_masters() won't touch the
5691 MSTCPY locks we've received early. */
5692
c04fecb4
DT		 5693 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5694 from_nodeid, R_RECEIVE_RECOVER, &r);
e7fd4179
DT		 5695 if (error)
5696 goto out;
5697
c04fecb4
DT		 5698 lock_rsb(r);
5699
4875647a
DT		 5700 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5701 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
c04fecb4 5702 from_nodeid, remid);
4875647a 5703 error = -EBADR;
c04fecb4 5704 goto out_unlock;
4875647a
DT		 5705 }
5706
c04fecb4 5707 lkb = search_remid(r, from_nodeid, remid);
e7fd4179
DT		 5708 if (lkb) {
5709 error = -EEXIST;
5710 goto out_remid;
5711 }
5712
5713 error = create_lkb(ls, &lkb);
5714 if (error)
5715 goto out_unlock;
5716
5717 error = receive_rcom_lock_args(ls, lkb, r, rc);
5718 if (error) {
b3f58d8f 5719 __put_lkb(ls, lkb);
e7fd4179
DT		 5720 goto out_unlock;
5721 }
5722
5723 attach_lkb(r, lkb);
5724 add_lkb(r, lkb, rl->rl_status);
5725 error = 0;
4875647a
DT		 5726 ls->ls_recover_locks_in++;
5727
5728 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5729 rsb_set_flag(r, RSB_RECOVER_GRANT);
e7fd4179
DT		 5730
5731 out_remid:
5732 /* this is the new value returned to the lock holder for
5733 saving in its process-copy lkb */
163a1859 5734 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
e7fd4179 5735
4875647a
DT		 5736 lkb->lkb_recover_seq = ls->ls_recover_seq;
5737
e7fd4179
DT		 5738 out_unlock:
5739 unlock_rsb(r);
5740 put_rsb(r);
5741 out:
6d40c4a7 5742 if (error && error != -EEXIST)
075f0177 5743 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
c04fecb4 5744 from_nodeid, remid, error);
163a1859 5745 rl->rl_result = cpu_to_le32(error);
e7fd4179
DT		 5746 return error;
5747}
5748
ae773d0b 5749/* needs at least dlm_rcom + rcom_lock */
e7fd4179
DT		 5750int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5751{
5752 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5753 struct dlm_rsb *r;
5754 struct dlm_lkb *lkb;
6d40c4a7
DT		 5755 uint32_t lkid, remid;
5756 int error, result;
5757
5758 lkid = le32_to_cpu(rl->rl_lkid);
5759 remid = le32_to_cpu(rl->rl_remid);
5760 result = le32_to_cpu(rl->rl_result);
e7fd4179 5761
6d40c4a7 5762 error = find_lkb(ls, lkid, &lkb);
e7fd4179 5763 if (error) {
6d40c4a7 5764 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
3428785a
AA		 5765 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5766 result);
e7fd4179
DT		 5767 return error;
5768 }
5769
4875647a
DT		 5770 r = lkb->lkb_resource;
5771 hold_rsb(r);
5772 lock_rsb(r);
5773
6d40c4a7
DT		 5774 if (!is_process_copy(lkb)) {
5775 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
3428785a
AA		 5776 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5777 result);
4875647a
DT		 5778 dlm_dump_rsb(r);
5779 unlock_rsb(r);
5780 put_rsb(r);
5781 dlm_put_lkb(lkb);
6d40c4a7
DT		 5782 return -EINVAL;
5783 }
e7fd4179 5784
6d40c4a7 5785 switch (result) {
dc200a88
DT		 5786 case -EBADR:
5787 /* There's a chance the new master received our lock before
5788 dlm_recover_master_reply(), this wouldn't happen if we did
5789 a barrier between recover_masters and recover_locks. */
6d40c4a7
DT		 5790
5791 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
3428785a
AA		 5792 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5793 result);
6d40c4a7 5794
dc200a88
DT		 5795 dlm_send_rcom_lock(r, lkb);
5796 goto out;
e7fd4179 5797 case -EEXIST:
e7fd4179 5798 case 0:
6d40c4a7 5799 lkb->lkb_remid = remid;
e7fd4179
DT		 5800 break;
5801 default:
6d40c4a7 5802 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
3428785a
AA		 5803 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5804 result);
e7fd4179
DT		 5805 }
5806
5807 /* an ack for dlm_recover_locks() which waits for replies from
5808 all the locks it sends to new masters */
5809 dlm_recovered_lock(r);
dc200a88 5810 out:
e7fd4179
DT		 5811 unlock_rsb(r);
5812 put_rsb(r);
b3f58d8f 5813 dlm_put_lkb(lkb);
e7fd4179
DT		 5814
5815 return 0;
5816}
5817
597d0cae
DT		 5818int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5819 int mode, uint32_t flags, void *name, unsigned int namelen,
d7db923e 5820 unsigned long timeout_cs)
597d0cae
DT		 5821{
5822 struct dlm_lkb *lkb;
5823 struct dlm_args args;
5824 int error;
5825
85e86edf 5826 dlm_lock_recovery(ls);
597d0cae
DT		 5827
5828 error = create_lkb(ls, &lkb);
5829 if (error) {
5830 kfree(ua);
5831 goto out;
5832 }
5833
5834 if (flags & DLM_LKF_VALBLK) {
573c24c4 5835 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
597d0cae
DT		 5836 if (!ua->lksb.sb_lvbptr) {
5837 kfree(ua);
5838 __put_lkb(ls, lkb);
5839 error = -ENOMEM;
5840 goto out;
5841 }
5842 }
d7db923e 5843 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
e5dae548 5844 fake_astfn, ua, fake_bastfn, &args);
597d0cae 5845 if (error) {
d47b41ac
VA		 5846 kfree(ua->lksb.sb_lvbptr);
5847 ua->lksb.sb_lvbptr = NULL;
5848 kfree(ua);
597d0cae
DT		 5849 __put_lkb(ls, lkb);
5850 goto out;
5851 }
5852
d47b41ac
VA		 5853 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5854 When DLM_IFL_USER is set, the dlm knows that this is a userspace
5855 lock and that lkb_astparam is the dlm_user_args structure. */
5856 lkb->lkb_flags |= DLM_IFL_USER;
597d0cae
DT		 5857 error = request_lock(ls, lkb, name, namelen, &args);
5858
5859 switch (error) {
5860 case 0:
5861 break;
5862 case -EINPROGRESS:
5863 error = 0;
5864 break;
5865 case -EAGAIN:
5866 error = 0;
df561f66 5867 fallthrough;
597d0cae
DT		 5868 default:
5869 __put_lkb(ls, lkb);
5870 goto out;
5871 }
5872
5873 /* add this new lkb to the per-process list of locks */
5874 spin_lock(&ua->proc->locks_spin);
ef0c2bb0 5875 hold_lkb(lkb);
597d0cae
DT		 5876 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5877 spin_unlock(&ua->proc->locks_spin);
5878 out:
85e86edf 5879 dlm_unlock_recovery(ls);
597d0cae
DT		 5880 return error;
5881}
5882
5883int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
d7db923e
DT		 5884 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5885 unsigned long timeout_cs)
597d0cae
DT		 5886{
5887 struct dlm_lkb *lkb;
5888 struct dlm_args args;
5889 struct dlm_user_args *ua;
5890 int error;
5891
85e86edf 5892 dlm_lock_recovery(ls);
597d0cae
DT		 5893
5894 error = find_lkb(ls, lkid, &lkb);
5895 if (error)
5896 goto out;
5897
5898 /* user can change the params on its lock when it converts it, or
5899 add an lvb that didn't exist before */
5900
d292c0cc 5901 ua = lkb->lkb_ua;
597d0cae
DT		 5902
5903 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
573c24c4 5904 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
597d0cae
DT		 5905 if (!ua->lksb.sb_lvbptr) {
5906 error = -ENOMEM;
5907 goto out_put;
5908 }
5909 }
5910 if (lvb_in && ua->lksb.sb_lvbptr)
5911 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5912
d7db923e 5913 ua->xid = ua_tmp->xid;
597d0cae
DT		 5914 ua->castparam = ua_tmp->castparam;
5915 ua->castaddr = ua_tmp->castaddr;
5916 ua->bastparam = ua_tmp->bastparam;
5917 ua->bastaddr = ua_tmp->bastaddr;
10948eb4 5918 ua->user_lksb = ua_tmp->user_lksb;
597d0cae 5919
d7db923e 5920 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
e5dae548 5921 fake_astfn, ua, fake_bastfn, &args);
597d0cae
DT		 5922 if (error)
5923 goto out_put;
5924
5925 error = convert_lock(ls, lkb, &args);
5926
c85d65e9 5927 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
597d0cae
DT		 5928 error = 0;
5929 out_put:
5930 dlm_put_lkb(lkb);
5931 out:
85e86edf 5932 dlm_unlock_recovery(ls);
597d0cae
DT		 5933 kfree(ua_tmp);
5934 return error;
5935}
5936
2ab4bd8e
DT		 5937/*
5938 * The caller asks for an orphan lock on a given resource with a given mode.
5939 * If a matching lock exists, it's moved to the owner's list of locks and
5940 * the lkid is returned.
5941 */
5942
5943int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5944 int mode, uint32_t flags, void *name, unsigned int namelen,
5945 unsigned long timeout_cs, uint32_t *lkid)
5946{
5947 struct dlm_lkb *lkb;
5948 struct dlm_user_args *ua;
5949 int found_other_mode = 0;
5950 int found = 0;
5951 int rv = 0;
5952
5953 mutex_lock(&ls->ls_orphans_mutex);
5954 list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) {
5955 if (lkb->lkb_resource->res_length != namelen)
5956 continue;
5957 if (memcmp(lkb->lkb_resource->res_name, name, namelen))
5958 continue;
5959 if (lkb->lkb_grmode != mode) {
5960 found_other_mode = 1;
5961 continue;
5962 }
5963
5964 found = 1;
5965 list_del_init(&lkb->lkb_ownqueue);
5966 lkb->lkb_flags &= ~DLM_IFL_ORPHAN;
5967 *lkid = lkb->lkb_id;
5968 break;
5969 }
5970 mutex_unlock(&ls->ls_orphans_mutex);
5971
5972 if (!found && found_other_mode) {
5973 rv = -EAGAIN;
5974 goto out;
5975 }
5976
5977 if (!found) {
5978 rv = -ENOENT;
5979 goto out;
5980 }
5981
5982 lkb->lkb_exflags = flags;
5983 lkb->lkb_ownpid = (int) current->pid;
5984
5985 ua = lkb->lkb_ua;
5986
5987 ua->proc = ua_tmp->proc;
5988 ua->xid = ua_tmp->xid;
5989 ua->castparam = ua_tmp->castparam;
5990 ua->castaddr = ua_tmp->castaddr;
5991 ua->bastparam = ua_tmp->bastparam;
5992 ua->bastaddr = ua_tmp->bastaddr;
5993 ua->user_lksb = ua_tmp->user_lksb;
5994
5995 /*
5996 * The lkb reference from the ls_orphans list was not
5997 * removed above, and is now considered the reference
5998 * for the proc locks list.
5999 */
6000
6001 spin_lock(&ua->proc->locks_spin);
6002 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
6003 spin_unlock(&ua->proc->locks_spin);
6004 out:
6005 kfree(ua_tmp);
6006 return rv;
6007}
6008
597d0cae
DT		 6009int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6010 uint32_t flags, uint32_t lkid, char *lvb_in)
6011{
6012 struct dlm_lkb *lkb;
6013 struct dlm_args args;
6014 struct dlm_user_args *ua;
6015 int error;
6016
85e86edf 6017 dlm_lock_recovery(ls);
597d0cae
DT		 6018
6019 error = find_lkb(ls, lkid, &lkb);
6020 if (error)
6021 goto out;
6022
d292c0cc 6023 ua = lkb->lkb_ua;
597d0cae
DT		 6024
6025 if (lvb_in && ua->lksb.sb_lvbptr)
6026 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
b434eda6
PC		 6027 if (ua_tmp->castparam)
6028 ua->castparam = ua_tmp->castparam;
cc346d55 6029 ua->user_lksb = ua_tmp->user_lksb;
597d0cae
DT		 6030
6031 error = set_unlock_args(flags, ua, &args);
6032 if (error)
6033 goto out_put;
6034
6035 error = unlock_lock(ls, lkb, &args);
6036
6037 if (error == -DLM_EUNLOCK)
6038 error = 0;
ef0c2bb0
DT		 6039 /* from validate_unlock_args() */
6040 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
6041 error = 0;
597d0cae
DT		 6042 if (error)
6043 goto out_put;
6044
6045 spin_lock(&ua->proc->locks_spin);
23e8e1aa 6046 /* dlm_user_add_cb() may have already taken lkb off the proc list */
a1bc86e6
DT		 6047 if (!list_empty(&lkb->lkb_ownqueue))
6048 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
597d0cae 6049 spin_unlock(&ua->proc->locks_spin);
597d0cae
DT		 6050 out_put:
6051 dlm_put_lkb(lkb);
6052 out:
85e86edf 6053 dlm_unlock_recovery(ls);
ef0c2bb0 6054 kfree(ua_tmp);
597d0cae
DT		 6055 return error;
6056}
6057
6058int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6059 uint32_t flags, uint32_t lkid)
6060{
6061 struct dlm_lkb *lkb;
6062 struct dlm_args args;
6063 struct dlm_user_args *ua;
6064 int error;
6065
85e86edf 6066 dlm_lock_recovery(ls);
597d0cae
DT		 6067
6068 error = find_lkb(ls, lkid, &lkb);
6069 if (error)
6070 goto out;
6071
d292c0cc 6072 ua = lkb->lkb_ua;
b434eda6
PC		 6073 if (ua_tmp->castparam)
6074 ua->castparam = ua_tmp->castparam;
c059f70e 6075 ua->user_lksb = ua_tmp->user_lksb;
597d0cae
DT		 6076
6077 error = set_unlock_args(flags, ua, &args);
6078 if (error)
6079 goto out_put;
6080
6081 error = cancel_lock(ls, lkb, &args);
6082
6083 if (error == -DLM_ECANCEL)
6084 error = 0;
ef0c2bb0
DT		 6085 /* from validate_unlock_args() */
6086 if (error == -EBUSY)
6087 error = 0;
597d0cae
DT		 6088 out_put:
6089 dlm_put_lkb(lkb);
6090 out:
85e86edf 6091 dlm_unlock_recovery(ls);
ef0c2bb0 6092 kfree(ua_tmp);
597d0cae
DT		 6093 return error;
6094}
6095
8b4021fa
DT		 6096int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6097{
6098 struct dlm_lkb *lkb;
6099 struct dlm_args args;
6100 struct dlm_user_args *ua;
6101 struct dlm_rsb *r;
6102 int error;
6103
6104 dlm_lock_recovery(ls);
6105
6106 error = find_lkb(ls, lkid, &lkb);
6107 if (error)
6108 goto out;
6109
d292c0cc 6110 ua = lkb->lkb_ua;
8b4021fa
DT		 6111
6112 error = set_unlock_args(flags, ua, &args);
6113 if (error)
6114 goto out_put;
6115
6116 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6117
6118 r = lkb->lkb_resource;
6119 hold_rsb(r);
6120 lock_rsb(r);
6121
6122 error = validate_unlock_args(lkb, &args);
6123 if (error)
6124 goto out_r;
6125 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6126
6127 error = _cancel_lock(r, lkb);
6128 out_r:
6129 unlock_rsb(r);
6130 put_rsb(r);
6131
6132 if (error == -DLM_ECANCEL)
6133 error = 0;
6134 /* from validate_unlock_args() */
6135 if (error == -EBUSY)
6136 error = 0;
6137 out_put:
6138 dlm_put_lkb(lkb);
6139 out:
6140 dlm_unlock_recovery(ls);
6141 return error;
6142}
6143
ef0c2bb0
DT		 6144/* lkb's that are removed from the waiters list by revert are just left on the
6145 orphans list with the granted orphan locks, to be freed by purge */
6146
597d0cae
DT		 6147static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6148{
ef0c2bb0
DT		 6149 struct dlm_args args;
6150 int error;
597d0cae 6151
2ab4bd8e 6152 hold_lkb(lkb); /* reference for the ls_orphans list */
ef0c2bb0
DT		 6153 mutex_lock(&ls->ls_orphans_mutex);
6154 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6155 mutex_unlock(&ls->ls_orphans_mutex);
597d0cae 6156
d292c0cc 6157 set_unlock_args(0, lkb->lkb_ua, &args);
ef0c2bb0
DT		 6158
6159 error = cancel_lock(ls, lkb, &args);
6160 if (error == -DLM_ECANCEL)
6161 error = 0;
6162 return error;
597d0cae
DT		 6163}
6164
da8c6663
DT		 6165/* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6166 granted. Regardless of what rsb queue the lock is on, it's removed and
6167 freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6168 if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
597d0cae
DT		 6169
6170static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6171{
597d0cae
DT		 6172 struct dlm_args args;
6173 int error;
6174
da8c6663
DT		 6175 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6176 lkb->lkb_ua, &args);
597d0cae
DT		 6177
6178 error = unlock_lock(ls, lkb, &args);
6179 if (error == -DLM_EUNLOCK)
6180 error = 0;
6181 return error;
6182}
6183
ef0c2bb0
DT		 6184/* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6185 (which does lock_rsb) due to deadlock with receiving a message that does
23e8e1aa 6186 lock_rsb followed by dlm_user_add_cb() */
ef0c2bb0
DT		 6187
6188static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6189 struct dlm_user_proc *proc)
6190{
6191 struct dlm_lkb *lkb = NULL;
6192
6193 mutex_lock(&ls->ls_clear_proc_locks);
6194 if (list_empty(&proc->locks))
6195 goto out;
6196
6197 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6198 list_del_init(&lkb->lkb_ownqueue);
6199
6200 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6201 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6202 else
6203 lkb->lkb_flags |= DLM_IFL_DEAD;
6204 out:
6205 mutex_unlock(&ls->ls_clear_proc_locks);
6206 return lkb;
6207}
6208
23e8e1aa 6209/* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
597d0cae
DT		 6210 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6211 which we clear here. */
6212
6213/* proc CLOSING flag is set so no more device_reads should look at proc->asts
6214 list, and no more device_writes should add lkb's to proc->locks list; so we
6215 shouldn't need to take asts_spin or locks_spin here. this assumes that
6216 device reads/writes/closes are serialized -- FIXME: we may need to serialize
6217 them ourself. */
6218
6219void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6220{
6221 struct dlm_lkb *lkb, *safe;
6222
85e86edf 6223 dlm_lock_recovery(ls);
597d0cae 6224
ef0c2bb0
DT		 6225 while (1) {
6226 lkb = del_proc_lock(ls, proc);
6227 if (!lkb)
6228 break;
84d8cd69 6229 del_timeout(lkb);
ef0c2bb0 6230 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
597d0cae 6231 orphan_proc_lock(ls, lkb);
ef0c2bb0 6232 else
597d0cae 6233 unlock_proc_lock(ls, lkb);
597d0cae
DT		 6234
6235 /* this removes the reference for the proc->locks list
6236 added by dlm_user_request, it may result in the lkb
6237 being freed */
6238
6239 dlm_put_lkb(lkb);
6240 }
a1bc86e6 6241
ef0c2bb0
DT		 6242 mutex_lock(&ls->ls_clear_proc_locks);
6243
a1bc86e6
DT		 6244 /* in-progress unlocks */
6245 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6246 list_del_init(&lkb->lkb_ownqueue);
6247 lkb->lkb_flags |= DLM_IFL_DEAD;
6248 dlm_put_lkb(lkb);
6249 }
6250
23e8e1aa 6251 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
8304d6f2
DT		 6252 memset(&lkb->lkb_callbacks, 0,
6253 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
23e8e1aa 6254 list_del_init(&lkb->lkb_cb_list);
a1bc86e6
DT		 6255 dlm_put_lkb(lkb);
6256 }
6257
597d0cae 6258 mutex_unlock(&ls->ls_clear_proc_locks);
85e86edf 6259 dlm_unlock_recovery(ls);
597d0cae 6260}
a1bc86e6 6261
8499137d
DT		 6262static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6263{
6264 struct dlm_lkb *lkb, *safe;
6265
6266 while (1) {
6267 lkb = NULL;
6268 spin_lock(&proc->locks_spin);
6269 if (!list_empty(&proc->locks)) {
6270 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6271 lkb_ownqueue);
6272 list_del_init(&lkb->lkb_ownqueue);
6273 }
6274 spin_unlock(&proc->locks_spin);
6275
6276 if (!lkb)
6277 break;
6278
6279 lkb->lkb_flags |= DLM_IFL_DEAD;
6280 unlock_proc_lock(ls, lkb);
6281 dlm_put_lkb(lkb); /* ref from proc->locks list */
6282 }
6283
6284 spin_lock(&proc->locks_spin);
6285 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6286 list_del_init(&lkb->lkb_ownqueue);
6287 lkb->lkb_flags |= DLM_IFL_DEAD;
6288 dlm_put_lkb(lkb);
6289 }
6290 spin_unlock(&proc->locks_spin);
6291
6292 spin_lock(&proc->asts_spin);
23e8e1aa 6293 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
8304d6f2
DT		 6294 memset(&lkb->lkb_callbacks, 0,
6295 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
23e8e1aa 6296 list_del_init(&lkb->lkb_cb_list);
8499137d
DT		 6297 dlm_put_lkb(lkb);
6298 }
6299 spin_unlock(&proc->asts_spin);
6300}
6301
6302/* pid of 0 means purge all orphans */
6303
6304static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6305{
6306 struct dlm_lkb *lkb, *safe;
6307
6308 mutex_lock(&ls->ls_orphans_mutex);
6309 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6310 if (pid && lkb->lkb_ownpid != pid)
6311 continue;
6312 unlock_proc_lock(ls, lkb);
6313 list_del_init(&lkb->lkb_ownqueue);
6314 dlm_put_lkb(lkb);
6315 }
6316 mutex_unlock(&ls->ls_orphans_mutex);
6317}
6318
6319static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6320{
6321 struct dlm_message *ms;
6322 struct dlm_mhandle *mh;
6323 int error;
6324
6325 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6326 DLM_MSG_PURGE, &ms, &mh);
6327 if (error)
6328 return error;
00e99ccd
AA		 6329 ms->m_nodeid = cpu_to_le32(nodeid);
6330 ms->m_pid = cpu_to_le32(pid);
8499137d
DT		 6331
6332 return send_message(mh, ms);
6333}
6334
6335int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6336 int nodeid, int pid)
6337{
6338 int error = 0;
6339
2ab4bd8e 6340 if (nodeid && (nodeid != dlm_our_nodeid())) {
8499137d
DT		 6341 error = send_purge(ls, nodeid, pid);
6342 } else {
85e86edf 6343 dlm_lock_recovery(ls);
8499137d
DT		 6344 if (pid == current->pid)
6345 purge_proc_locks(ls, proc);
6346 else
6347 do_purge(ls, nodeid, pid);
85e86edf 6348 dlm_unlock_recovery(ls);
8499137d
DT		 6349 }
6350 return error;
6351}
6352
5054e79d
AA		 6353/* debug functionality */
6354int dlm_debug_add_lkb(struct dlm_ls *ls, uint32_t lkb_id, char *name, int len,
6355 int lkb_nodeid, unsigned int lkb_flags, int lkb_status)
6356{
6357 struct dlm_lksb *lksb;
6358 struct dlm_lkb *lkb;
6359 struct dlm_rsb *r;
6360 int error;
6361
6362 /* we currently can't set a valid user lock */
6363 if (lkb_flags & DLM_IFL_USER)
6364 return -EOPNOTSUPP;
6365
6366 lksb = kzalloc(sizeof(*lksb), GFP_NOFS);
6367 if (!lksb)
6368 return -ENOMEM;
6369
6370 error = _create_lkb(ls, &lkb, lkb_id, lkb_id + 1);
6371 if (error) {
6372 kfree(lksb);
6373 return error;
6374 }
6375
6376 lkb->lkb_flags = lkb_flags;
6377 lkb->lkb_nodeid = lkb_nodeid;
6378 lkb->lkb_lksb = lksb;
6379 /* user specific pointer, just don't have it NULL for kernel locks */
6380 if (~lkb_flags & DLM_IFL_USER)
6381 lkb->lkb_astparam = (void *)0xDEADBEEF;
6382
6383 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
6384 if (error) {
6385 kfree(lksb);
6386 __put_lkb(ls, lkb);
6387 return error;
6388 }
6389
6390 lock_rsb(r);
6391 attach_lkb(r, lkb);
6392 add_lkb(r, lkb, lkb_status);
6393 unlock_rsb(r);
6394 put_rsb(r);
6395
6396 return 0;
6397}
6398
63eab2b0
AA		 6399int dlm_debug_add_lkb_to_waiters(struct dlm_ls *ls, uint32_t lkb_id,
6400 int mstype, int to_nodeid)
6401{
6402 struct dlm_lkb *lkb;
6403 int error;
6404
6405 error = find_lkb(ls, lkb_id, &lkb);
6406 if (error)
6407 return error;
6408
6409 error = add_to_waiters(lkb, mstype, to_nodeid);
6410 dlm_put_lkb(lkb);
6411 return error;
6412}
6413
Empty description
This page took 1.788336 seconds and 4 git commands to generate.