1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved.
6 ** This copyrighted material is made available to anyone wishing to use,
7 ** modify, copy, or redistribute it subject to the terms and conditions
8 ** of the GNU General Public License v.2.
10 *******************************************************************************
11 ******************************************************************************/
13 /* Central locking logic has four stages:
33 Stage 1 (lock, unlock) is mainly about checking input args and
34 splitting into one of the four main operations:
36 dlm_lock = request_lock
37 dlm_lock+CONVERT = convert_lock
38 dlm_unlock = unlock_lock
39 dlm_unlock+CANCEL = cancel_lock
41 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
42 provided to the next stage.
44 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
45 When remote, it calls send_xxxx(), when local it calls do_xxxx().
47 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
48 given rsb and lkb and queues callbacks.
50 For remote operations, send_xxxx() results in the corresponding do_xxxx()
51 function being executed on the remote node. The connecting send/receive
52 calls on local (L) and remote (R) nodes:
54 L: send_xxxx() -> R: receive_xxxx()
56 L: receive_xxxx_reply() <- R: send_xxxx_reply()
58 #include <linux/types.h>
59 #include <linux/rbtree.h>
60 #include <linux/slab.h>
61 #include "dlm_internal.h"
62 #include <linux/dlm_device.h>
65 #include "requestqueue.h"
69 #include "lockspace.h"
74 #include "lvb_table.h"
78 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
79 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
80 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
81 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
82 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
83 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
84 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
85 static int send_remove(struct dlm_rsb *r);
86 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
87 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
88 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
89 struct dlm_message *ms);
90 static int receive_extralen(struct dlm_message *ms);
91 static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
92 static void del_timeout(struct dlm_lkb *lkb);
93 static void toss_rsb(struct kref *kref);
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)
103 static 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 */
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
124 const 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 */
136 #define modes_compat(gr, rq) \
137 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
139 int dlm_modes_compat(int mode1, int mode2)
141 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
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]
150 static 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 */
162 void dlm_print_lkb(struct dlm_lkb *lkb)
164 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
165 "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
166 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
167 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
168 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
169 (unsigned long long)lkb->lkb_recover_seq);
172 static void dlm_print_rsb(struct dlm_rsb *r)
174 printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
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,
181 void dlm_dump_rsb(struct dlm_rsb *r)
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)
192 printk(KERN_ERR "rsb grant queue:\n");
193 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
195 printk(KERN_ERR "rsb convert queue:\n");
196 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
198 printk(KERN_ERR "rsb wait queue:\n");
199 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
203 /* Threads cannot use the lockspace while it's being recovered */
205 static inline void dlm_lock_recovery(struct dlm_ls *ls)
207 down_read(&ls->ls_in_recovery);
210 void dlm_unlock_recovery(struct dlm_ls *ls)
212 up_read(&ls->ls_in_recovery);
215 int dlm_lock_recovery_try(struct dlm_ls *ls)
217 return down_read_trylock(&ls->ls_in_recovery);
220 static inline int can_be_queued(struct dlm_lkb *lkb)
222 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
225 static inline int force_blocking_asts(struct dlm_lkb *lkb)
227 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
230 static inline int is_demoted(struct dlm_lkb *lkb)
232 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
235 static inline int is_altmode(struct dlm_lkb *lkb)
237 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
240 static inline int is_granted(struct dlm_lkb *lkb)
242 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
245 static inline int is_remote(struct dlm_rsb *r)
247 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
248 return !!r->res_nodeid;
251 static inline int is_process_copy(struct dlm_lkb *lkb)
253 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
256 static inline int is_master_copy(struct dlm_lkb *lkb)
258 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
261 static inline int middle_conversion(struct dlm_lkb *lkb)
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))
269 static inline int down_conversion(struct dlm_lkb *lkb)
271 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
274 static inline int is_overlap_unlock(struct dlm_lkb *lkb)
276 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
279 static inline int is_overlap_cancel(struct dlm_lkb *lkb)
281 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
284 static inline int is_overlap(struct dlm_lkb *lkb)
286 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
287 DLM_IFL_OVERLAP_CANCEL));
290 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
292 if (is_master_copy(lkb))
297 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
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;
306 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
307 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
311 dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
314 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
317 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
320 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
322 if (is_master_copy(lkb)) {
323 send_bast(r, lkb, rqmode);
325 dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
330 * Basic operations on rsb's and lkb's
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. */
336 static inline void hold_rsb(struct dlm_rsb *r)
338 kref_get(&r->res_ref);
341 void dlm_hold_rsb(struct dlm_rsb *r)
346 /* When all references to the rsb are gone it's transferred to
347 the tossed list for later disposal. */
349 static void put_rsb(struct dlm_rsb *r)
351 struct dlm_ls *ls = r->res_ls;
352 uint32_t bucket = r->res_bucket;
354 spin_lock(&ls->ls_rsbtbl[bucket].lock);
355 kref_put(&r->res_ref, toss_rsb);
356 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
359 void dlm_put_rsb(struct dlm_rsb *r)
364 static int pre_rsb_struct(struct dlm_ls *ls)
366 struct dlm_rsb *r1, *r2;
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);
374 spin_unlock(&ls->ls_new_rsb_spin);
376 r1 = dlm_allocate_rsb(ls);
377 r2 = dlm_allocate_rsb(ls);
379 spin_lock(&ls->ls_new_rsb_spin);
381 list_add(&r1->res_hashchain, &ls->ls_new_rsb);
382 ls->ls_new_rsb_count++;
385 list_add(&r2->res_hashchain, &ls->ls_new_rsb);
386 ls->ls_new_rsb_count++;
388 count = ls->ls_new_rsb_count;
389 spin_unlock(&ls->ls_new_rsb_spin);
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. */
400 static int get_rsb_struct(struct dlm_ls *ls, char *name, int len,
401 struct dlm_rsb **r_ret)
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);
415 r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
416 list_del(&r->res_hashchain);
417 /* Convert the empty list_head to a NULL rb_node for tree usage: */
418 memset(&r->res_hashnode, 0, sizeof(struct rb_node));
419 ls->ls_new_rsb_count--;
420 spin_unlock(&ls->ls_new_rsb_spin);
424 memcpy(r->res_name, name, len);
425 mutex_init(&r->res_mutex);
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);
438 static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
440 char maxname[DLM_RESNAME_MAXLEN];
442 memset(maxname, 0, DLM_RESNAME_MAXLEN);
443 memcpy(maxname, name, nlen);
444 return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
447 int dlm_search_rsb_tree(struct rb_root *tree, char *name, int len,
448 struct dlm_rsb **r_ret)
450 struct rb_node *node = tree->rb_node;
455 r = rb_entry(node, struct dlm_rsb, res_hashnode);
456 rc = rsb_cmp(r, name, len);
458 node = node->rb_left;
460 node = node->rb_right;
472 static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
474 struct rb_node **newn = &tree->rb_node;
475 struct rb_node *parent = NULL;
479 struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
483 rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
485 newn = &parent->rb_left;
487 newn = &parent->rb_right;
489 log_print("rsb_insert match");
496 rb_link_node(&rsb->res_hashnode, parent, newn);
497 rb_insert_color(&rsb->res_hashnode, tree);
502 * Find rsb in rsbtbl and potentially create/add one
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.
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.
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.
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)
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.
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
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.
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.)
545 static 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)
550 struct dlm_rsb *r = NULL;
551 int our_nodeid = dlm_our_nodeid();
558 if (flags & R_RECEIVE_REQUEST) {
559 if (from_nodeid == dir_nodeid)
563 } else if (flags & R_REQUEST) {
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
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.)
582 if (from_local || from_dir ||
583 (from_other && (dir_nodeid == our_nodeid))) {
589 error = pre_rsb_struct(ls);
594 spin_lock(&ls->ls_rsbtbl[b].lock);
596 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
601 * rsb is active, so we can't check master_nodeid without lock_rsb.
604 kref_get(&r->res_ref);
610 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
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.
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,
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);
636 /* fix it and go on */
637 r->res_master_nodeid = our_nodeid;
639 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
640 r->res_first_lkid = 0;
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. */
646 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
647 r->res_first_lkid = 0;
650 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
651 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
660 if (error == -EBADR && !create)
663 error = get_rsb_struct(ls, name, len, &r);
664 if (error == -EAGAIN) {
665 spin_unlock(&ls->ls_rsbtbl[b].lock);
673 r->res_dir_nodeid = dir_nodeid;
674 kref_init(&r->res_ref);
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;
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);
696 log_debug(ls, "find_rsb new from_other %d dir %d %s",
697 from_nodeid, dir_nodeid, r->res_name);
700 if (dir_nodeid == our_nodeid) {
701 /* When we are the dir nodeid, we can set the master
703 r->res_master_nodeid = our_nodeid;
706 /* set_master will send_lookup to dir_nodeid */
707 r->res_master_nodeid = 0;
712 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
714 spin_unlock(&ls->ls_rsbtbl[b].lock);
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). */
724 static 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)
729 struct dlm_rsb *r = NULL;
730 int our_nodeid = dlm_our_nodeid();
731 int recover = (flags & R_RECEIVE_RECOVER);
735 error = pre_rsb_struct(ls);
739 spin_lock(&ls->ls_rsbtbl[b].lock);
741 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
746 * rsb is active, so we can't check master_nodeid without lock_rsb.
749 kref_get(&r->res_ref);
754 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
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.
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);
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);
781 r->res_master_nodeid = our_nodeid;
785 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
786 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
795 error = get_rsb_struct(ls, name, len, &r);
796 if (error == -EAGAIN) {
797 spin_unlock(&ls->ls_rsbtbl[b].lock);
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);
810 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
812 spin_unlock(&ls->ls_rsbtbl[b].lock);
818 static int find_rsb(struct dlm_ls *ls, char *name, int len, int from_nodeid,
819 unsigned int flags, struct dlm_rsb **r_ret)
824 if (len > DLM_RESNAME_MAXLEN)
827 hash = jhash(name, len, 0);
828 b = hash & (ls->ls_rsbtbl_size - 1);
830 dir_nodeid = dlm_hash2nodeid(ls, hash);
832 if (dlm_no_directory(ls))
833 return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
834 from_nodeid, flags, r_ret);
836 return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
837 from_nodeid, flags, r_ret);
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 */
843 static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
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,
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. */
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);
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 */
870 if (r->res_master_nodeid) {
871 log_error(ls, "validate master from_dir %d master %d "
873 from_nodeid, r->res_master_nodeid,
874 r->res_first_lkid, r->res_name);
877 r->res_master_nodeid = dlm_our_nodeid();
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().
889 * normal operation, we are the dir node for a resource
894 * . dlm_master_lookup flags 0
896 * recover directory, we are rebuilding dir for all resources
897 * . dlm_recover_directory
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.
904 * recover masters, we are finding the new master for resources
905 * . dlm_recover_masters
907 * . dlm_send_rcom_lookup
908 * . receive_rcom_lookup
909 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
912 int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
913 unsigned int flags, int *r_nodeid, int *result)
915 struct dlm_rsb *r = NULL;
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;
922 if (len > DLM_RESNAME_MAXLEN)
925 if (from_nodeid == our_nodeid) {
926 log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
931 hash = jhash(name, len, 0);
932 b = hash & (ls->ls_rsbtbl_size - 1);
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,
944 error = pre_rsb_struct(ls);
948 spin_lock(&ls->ls_rsbtbl[b].lock);
949 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
951 /* because the rsb is active, we need to lock_rsb before
952 checking/changing re_master_nodeid */
955 spin_unlock(&ls->ls_rsbtbl[b].lock);
960 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
964 /* because the rsb is inactive (on toss list), it's not refcounted
965 and lock_rsb is not used, but is protected by the rsbtbl lock */
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;
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. */
982 r->res_master_nodeid = from_nodeid;
983 r->res_nodeid = from_nodeid;
984 rsb_set_flag(r, RSB_NEW_MASTER);
987 /* I don't think we should ever find it on toss list. */
988 log_error(ls, "dlm_master_lookup fix_master on toss");
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 */
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);
1003 if (r->res_master_nodeid == our_nodeid) {
1004 log_error(ls, "from_master %d our_master", from_nodeid);
1006 dlm_send_rcom_lookup_dump(r, from_nodeid);
1010 r->res_master_nodeid = from_nodeid;
1011 r->res_nodeid = from_nodeid;
1012 rsb_set_flag(r, RSB_NEW_MASTER);
1015 if (!r->res_master_nodeid) {
1016 /* this will happen if recovery happens while we're looking
1017 up the master for this rsb */
1019 log_debug(ls, "dlm_master_lookup master 0 to %d first %x %s",
1020 from_nodeid, r->res_first_lkid, r->res_name);
1021 r->res_master_nodeid = from_nodeid;
1022 r->res_nodeid = from_nodeid;
1025 if (!from_master && !fix_master &&
1026 (r->res_master_nodeid == from_nodeid)) {
1027 /* this can happen when the master sends remove, the dir node
1028 finds the rsb on the keep list and ignores the remove,
1029 and the former master sends a lookup */
1031 log_limit(ls, "dlm_master_lookup from master %d flags %x "
1032 "first %x %s", from_nodeid, flags,
1033 r->res_first_lkid, r->res_name);
1037 *r_nodeid = r->res_master_nodeid;
1039 *result = DLM_LU_MATCH;
1042 r->res_toss_time = jiffies;
1043 /* the rsb was inactive (on toss list) */
1044 spin_unlock(&ls->ls_rsbtbl[b].lock);
1046 /* the rsb was active */
1053 error = get_rsb_struct(ls, name, len, &r);
1054 if (error == -EAGAIN) {
1055 spin_unlock(&ls->ls_rsbtbl[b].lock);
1063 r->res_dir_nodeid = our_nodeid;
1064 r->res_master_nodeid = from_nodeid;
1065 r->res_nodeid = from_nodeid;
1066 kref_init(&r->res_ref);
1067 r->res_toss_time = jiffies;
1069 error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1071 /* should never happen */
1073 spin_unlock(&ls->ls_rsbtbl[b].lock);
1078 *result = DLM_LU_ADD;
1079 *r_nodeid = from_nodeid;
1082 spin_unlock(&ls->ls_rsbtbl[b].lock);
1086 static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1092 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1093 spin_lock(&ls->ls_rsbtbl[i].lock);
1094 for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1095 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1096 if (r->res_hash == hash)
1099 spin_unlock(&ls->ls_rsbtbl[i].lock);
1103 void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
1105 struct dlm_rsb *r = NULL;
1109 hash = jhash(name, len, 0);
1110 b = hash & (ls->ls_rsbtbl_size - 1);
1112 spin_lock(&ls->ls_rsbtbl[b].lock);
1113 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1117 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1123 spin_unlock(&ls->ls_rsbtbl[b].lock);
1126 static void toss_rsb(struct kref *kref)
1128 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1129 struct dlm_ls *ls = r->res_ls;
1131 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1132 kref_init(&r->res_ref);
1133 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1134 rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
1135 r->res_toss_time = jiffies;
1136 ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
1137 if (r->res_lvbptr) {
1138 dlm_free_lvb(r->res_lvbptr);
1139 r->res_lvbptr = NULL;
1143 /* See comment for unhold_lkb */
1145 static void unhold_rsb(struct dlm_rsb *r)
1148 rv = kref_put(&r->res_ref, toss_rsb);
1149 DLM_ASSERT(!rv, dlm_dump_rsb(r););
1152 static void kill_rsb(struct kref *kref)
1154 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1156 /* All work is done after the return from kref_put() so we
1157 can release the write_lock before the remove and free. */
1159 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1160 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1161 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1162 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1163 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1164 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1167 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1168 The rsb must exist as long as any lkb's for it do. */
1170 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1173 lkb->lkb_resource = r;
1176 static void detach_lkb(struct dlm_lkb *lkb)
1178 if (lkb->lkb_resource) {
1179 put_rsb(lkb->lkb_resource);
1180 lkb->lkb_resource = NULL;
1184 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1186 struct dlm_lkb *lkb;
1189 lkb = dlm_allocate_lkb(ls);
1193 lkb->lkb_nodeid = -1;
1194 lkb->lkb_grmode = DLM_LOCK_IV;
1195 kref_init(&lkb->lkb_ref);
1196 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1197 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1198 INIT_LIST_HEAD(&lkb->lkb_time_list);
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);
1203 idr_preload(GFP_NOFS);
1204 spin_lock(&ls->ls_lkbidr_spin);
1205 rv = idr_alloc(&ls->ls_lkbidr, lkb, 1, 0, GFP_NOWAIT);
1208 spin_unlock(&ls->ls_lkbidr_spin);
1212 log_error(ls, "create_lkb idr error %d", rv);
1221 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1223 struct dlm_lkb *lkb;
1225 spin_lock(&ls->ls_lkbidr_spin);
1226 lkb = idr_find(&ls->ls_lkbidr, lkid);
1228 kref_get(&lkb->lkb_ref);
1229 spin_unlock(&ls->ls_lkbidr_spin);
1232 return lkb ? 0 : -ENOENT;
1235 static void kill_lkb(struct kref *kref)
1237 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1239 /* All work is done after the return from kref_put() so we
1240 can release the write_lock before the detach_lkb */
1242 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1245 /* __put_lkb() is used when an lkb may not have an rsb attached to
1246 it so we need to provide the lockspace explicitly */
1248 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
1250 uint32_t lkid = lkb->lkb_id;
1252 spin_lock(&ls->ls_lkbidr_spin);
1253 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
1254 idr_remove(&ls->ls_lkbidr, lkid);
1255 spin_unlock(&ls->ls_lkbidr_spin);
1259 /* for local/process lkbs, lvbptr points to caller's lksb */
1260 if (lkb->lkb_lvbptr && is_master_copy(lkb))
1261 dlm_free_lvb(lkb->lkb_lvbptr);
1265 spin_unlock(&ls->ls_lkbidr_spin);
1270 int dlm_put_lkb(struct dlm_lkb *lkb)
1274 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1275 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1277 ls = lkb->lkb_resource->res_ls;
1278 return __put_lkb(ls, lkb);
1281 /* This is only called to add a reference when the code already holds
1282 a valid reference to the lkb, so there's no need for locking. */
1284 static inline void hold_lkb(struct dlm_lkb *lkb)
1286 kref_get(&lkb->lkb_ref);
1289 /* This is called when we need to remove a reference and are certain
1290 it's not the last ref. e.g. del_lkb is always called between a
1291 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1292 put_lkb would work fine, but would involve unnecessary locking */
1294 static inline void unhold_lkb(struct dlm_lkb *lkb)
1297 rv = kref_put(&lkb->lkb_ref, kill_lkb);
1298 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
1301 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1304 struct dlm_lkb *lkb = NULL;
1306 list_for_each_entry(lkb, head, lkb_statequeue)
1307 if (lkb->lkb_rqmode < mode)
1310 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
1313 /* add/remove lkb to rsb's grant/convert/wait queue */
1315 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1317 kref_get(&lkb->lkb_ref);
1319 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1321 lkb->lkb_timestamp = ktime_get();
1323 lkb->lkb_status = status;
1326 case DLM_LKSTS_WAITING:
1327 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1328 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1330 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1332 case DLM_LKSTS_GRANTED:
1333 /* convention says granted locks kept in order of grmode */
1334 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1337 case DLM_LKSTS_CONVERT:
1338 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1339 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1341 list_add_tail(&lkb->lkb_statequeue,
1342 &r->res_convertqueue);
1345 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1349 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1351 lkb->lkb_status = 0;
1352 list_del(&lkb->lkb_statequeue);
1356 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1360 add_lkb(r, lkb, sts);
1364 static int msg_reply_type(int mstype)
1367 case DLM_MSG_REQUEST:
1368 return DLM_MSG_REQUEST_REPLY;
1369 case DLM_MSG_CONVERT:
1370 return DLM_MSG_CONVERT_REPLY;
1371 case DLM_MSG_UNLOCK:
1372 return DLM_MSG_UNLOCK_REPLY;
1373 case DLM_MSG_CANCEL:
1374 return DLM_MSG_CANCEL_REPLY;
1375 case DLM_MSG_LOOKUP:
1376 return DLM_MSG_LOOKUP_REPLY;
1381 static int nodeid_warned(int nodeid, int num_nodes, int *warned)
1385 for (i = 0; i < num_nodes; i++) {
1390 if (warned[i] == nodeid)
1396 void dlm_scan_waiters(struct dlm_ls *ls)
1398 struct dlm_lkb *lkb;
1400 s64 debug_maxus = 0;
1401 u32 debug_scanned = 0;
1402 u32 debug_expired = 0;
1406 if (!dlm_config.ci_waitwarn_us)
1409 mutex_lock(&ls->ls_waiters_mutex);
1411 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1412 if (!lkb->lkb_wait_time)
1417 us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time));
1419 if (us < dlm_config.ci_waitwarn_us)
1422 lkb->lkb_wait_time = 0;
1425 if (us > debug_maxus)
1429 num_nodes = ls->ls_num_nodes;
1430 warned = kcalloc(num_nodes, sizeof(int), GFP_KERNEL);
1434 if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned))
1437 log_error(ls, "waitwarn %x %lld %d us check connection to "
1438 "node %d", lkb->lkb_id, (long long)us,
1439 dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid);
1441 mutex_unlock(&ls->ls_waiters_mutex);
1445 log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us",
1446 debug_scanned, debug_expired,
1447 dlm_config.ci_waitwarn_us, (long long)debug_maxus);
1450 /* add/remove lkb from global waiters list of lkb's waiting for
1451 a reply from a remote node */
1453 static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1455 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1458 mutex_lock(&ls->ls_waiters_mutex);
1460 if (is_overlap_unlock(lkb) ||
1461 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1466 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1468 case DLM_MSG_UNLOCK:
1469 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1471 case DLM_MSG_CANCEL:
1472 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1478 lkb->lkb_wait_count++;
1481 log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1482 lkb->lkb_id, lkb->lkb_wait_type, mstype,
1483 lkb->lkb_wait_count, lkb->lkb_flags);
1487 DLM_ASSERT(!lkb->lkb_wait_count,
1489 printk("wait_count %d\n", lkb->lkb_wait_count););
1491 lkb->lkb_wait_count++;
1492 lkb->lkb_wait_type = mstype;
1493 lkb->lkb_wait_time = ktime_get();
1494 lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1496 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1499 log_error(ls, "addwait error %x %d flags %x %d %d %s",
1500 lkb->lkb_id, error, lkb->lkb_flags, mstype,
1501 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1502 mutex_unlock(&ls->ls_waiters_mutex);
1506 /* We clear the RESEND flag because we might be taking an lkb off the waiters
1507 list as part of process_requestqueue (e.g. a lookup that has an optimized
1508 request reply on the requestqueue) between dlm_recover_waiters_pre() which
1509 set RESEND and dlm_recover_waiters_post() */
1511 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1512 struct dlm_message *ms)
1514 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1515 int overlap_done = 0;
1517 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1518 log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1519 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1524 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1525 log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1526 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1531 /* Cancel state was preemptively cleared by a successful convert,
1532 see next comment, nothing to do. */
1534 if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1535 (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1536 log_debug(ls, "remwait %x cancel_reply wait_type %d",
1537 lkb->lkb_id, lkb->lkb_wait_type);
1541 /* Remove for the convert reply, and premptively remove for the
1542 cancel reply. A convert has been granted while there's still
1543 an outstanding cancel on it (the cancel is moot and the result
1544 in the cancel reply should be 0). We preempt the cancel reply
1545 because the app gets the convert result and then can follow up
1546 with another op, like convert. This subsequent op would see the
1547 lingering state of the cancel and fail with -EBUSY. */
1549 if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1550 (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1551 is_overlap_cancel(lkb) && ms && !ms->m_result) {
1552 log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1554 lkb->lkb_wait_type = 0;
1555 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1556 lkb->lkb_wait_count--;
1561 /* N.B. type of reply may not always correspond to type of original
1562 msg due to lookup->request optimization, verify others? */
1564 if (lkb->lkb_wait_type) {
1565 lkb->lkb_wait_type = 0;
1569 log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1570 lkb->lkb_id, ms ? ms->m_header.h_nodeid : 0, lkb->lkb_remid,
1571 mstype, lkb->lkb_flags);
1575 /* the force-unlock/cancel has completed and we haven't recvd a reply
1576 to the op that was in progress prior to the unlock/cancel; we
1577 give up on any reply to the earlier op. FIXME: not sure when/how
1578 this would happen */
1580 if (overlap_done && lkb->lkb_wait_type) {
1581 log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1582 lkb->lkb_id, mstype, lkb->lkb_wait_type);
1583 lkb->lkb_wait_count--;
1585 lkb->lkb_wait_type = 0;
1588 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1590 lkb->lkb_flags &= ~DLM_IFL_RESEND;
1591 lkb->lkb_wait_count--;
1592 if (!lkb->lkb_wait_count)
1593 list_del_init(&lkb->lkb_wait_reply);
1598 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1600 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1603 mutex_lock(&ls->ls_waiters_mutex);
1604 error = _remove_from_waiters(lkb, mstype, NULL);
1605 mutex_unlock(&ls->ls_waiters_mutex);
1609 /* Handles situations where we might be processing a "fake" or "stub" reply in
1610 which we can't try to take waiters_mutex again. */
1612 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1614 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1617 if (ms->m_flags != DLM_IFL_STUB_MS)
1618 mutex_lock(&ls->ls_waiters_mutex);
1619 error = _remove_from_waiters(lkb, ms->m_type, ms);
1620 if (ms->m_flags != DLM_IFL_STUB_MS)
1621 mutex_unlock(&ls->ls_waiters_mutex);
1625 /* If there's an rsb for the same resource being removed, ensure
1626 that the remove message is sent before the new lookup message.
1627 It should be rare to need a delay here, but if not, then it may
1628 be worthwhile to add a proper wait mechanism rather than a delay. */
1630 static void wait_pending_remove(struct dlm_rsb *r)
1632 struct dlm_ls *ls = r->res_ls;
1634 spin_lock(&ls->ls_remove_spin);
1635 if (ls->ls_remove_len &&
1636 !rsb_cmp(r, ls->ls_remove_name, ls->ls_remove_len)) {
1637 log_debug(ls, "delay lookup for remove dir %d %s",
1638 r->res_dir_nodeid, r->res_name);
1639 spin_unlock(&ls->ls_remove_spin);
1643 spin_unlock(&ls->ls_remove_spin);
1647 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1648 * read by other threads in wait_pending_remove. ls_remove_names
1649 * and ls_remove_lens are only used by the scan thread, so they do
1650 * not need protection.
1653 static void shrink_bucket(struct dlm_ls *ls, int b)
1655 struct rb_node *n, *next;
1658 int our_nodeid = dlm_our_nodeid();
1659 int remote_count = 0;
1660 int need_shrink = 0;
1663 memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1665 spin_lock(&ls->ls_rsbtbl[b].lock);
1667 if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1668 spin_unlock(&ls->ls_rsbtbl[b].lock);
1672 for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1674 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1676 /* If we're the directory record for this rsb, and
1677 we're not the master of it, then we need to wait
1678 for the master node to send us a dir remove for
1679 before removing the dir record. */
1681 if (!dlm_no_directory(ls) &&
1682 (r->res_master_nodeid != our_nodeid) &&
1683 (dlm_dir_nodeid(r) == our_nodeid)) {
1689 if (!time_after_eq(jiffies, r->res_toss_time +
1690 dlm_config.ci_toss_secs * HZ)) {
1694 if (!dlm_no_directory(ls) &&
1695 (r->res_master_nodeid == our_nodeid) &&
1696 (dlm_dir_nodeid(r) != our_nodeid)) {
1698 /* We're the master of this rsb but we're not
1699 the directory record, so we need to tell the
1700 dir node to remove the dir record. */
1702 ls->ls_remove_lens[remote_count] = r->res_length;
1703 memcpy(ls->ls_remove_names[remote_count], r->res_name,
1704 DLM_RESNAME_MAXLEN);
1707 if (remote_count >= DLM_REMOVE_NAMES_MAX)
1712 if (!kref_put(&r->res_ref, kill_rsb)) {
1713 log_error(ls, "tossed rsb in use %s", r->res_name);
1717 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1722 ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1724 ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1725 spin_unlock(&ls->ls_rsbtbl[b].lock);
1728 * While searching for rsb's to free, we found some that require
1729 * remote removal. We leave them in place and find them again here
1730 * so there is a very small gap between removing them from the toss
1731 * list and sending the removal. Keeping this gap small is
1732 * important to keep us (the master node) from being out of sync
1733 * with the remote dir node for very long.
1735 * From the time the rsb is removed from toss until just after
1736 * send_remove, the rsb name is saved in ls_remove_name. A new
1737 * lookup checks this to ensure that a new lookup message for the
1738 * same resource name is not sent just before the remove message.
1741 for (i = 0; i < remote_count; i++) {
1742 name = ls->ls_remove_names[i];
1743 len = ls->ls_remove_lens[i];
1745 spin_lock(&ls->ls_rsbtbl[b].lock);
1746 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1748 spin_unlock(&ls->ls_rsbtbl[b].lock);
1749 log_debug(ls, "remove_name not toss %s", name);
1753 if (r->res_master_nodeid != our_nodeid) {
1754 spin_unlock(&ls->ls_rsbtbl[b].lock);
1755 log_debug(ls, "remove_name master %d dir %d our %d %s",
1756 r->res_master_nodeid, r->res_dir_nodeid,
1761 if (r->res_dir_nodeid == our_nodeid) {
1762 /* should never happen */
1763 spin_unlock(&ls->ls_rsbtbl[b].lock);
1764 log_error(ls, "remove_name dir %d master %d our %d %s",
1765 r->res_dir_nodeid, r->res_master_nodeid,
1770 if (!time_after_eq(jiffies, r->res_toss_time +
1771 dlm_config.ci_toss_secs * HZ)) {
1772 spin_unlock(&ls->ls_rsbtbl[b].lock);
1773 log_debug(ls, "remove_name toss_time %lu now %lu %s",
1774 r->res_toss_time, jiffies, name);
1778 if (!kref_put(&r->res_ref, kill_rsb)) {
1779 spin_unlock(&ls->ls_rsbtbl[b].lock);
1780 log_error(ls, "remove_name in use %s", name);
1784 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1786 /* block lookup of same name until we've sent remove */
1787 spin_lock(&ls->ls_remove_spin);
1788 ls->ls_remove_len = len;
1789 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1790 spin_unlock(&ls->ls_remove_spin);
1791 spin_unlock(&ls->ls_rsbtbl[b].lock);
1795 /* allow lookup of name again */
1796 spin_lock(&ls->ls_remove_spin);
1797 ls->ls_remove_len = 0;
1798 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1799 spin_unlock(&ls->ls_remove_spin);
1805 void dlm_scan_rsbs(struct dlm_ls *ls)
1809 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1810 shrink_bucket(ls, i);
1811 if (dlm_locking_stopped(ls))
1817 static void add_timeout(struct dlm_lkb *lkb)
1819 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1821 if (is_master_copy(lkb))
1824 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1825 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1826 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1829 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1834 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1835 mutex_lock(&ls->ls_timeout_mutex);
1837 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1838 mutex_unlock(&ls->ls_timeout_mutex);
1841 static void del_timeout(struct dlm_lkb *lkb)
1843 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1845 mutex_lock(&ls->ls_timeout_mutex);
1846 if (!list_empty(&lkb->lkb_time_list)) {
1847 list_del_init(&lkb->lkb_time_list);
1850 mutex_unlock(&ls->ls_timeout_mutex);
1853 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1854 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1855 and then lock rsb because of lock ordering in add_timeout. We may need
1856 to specify some special timeout-related bits in the lkb that are just to
1857 be accessed under the timeout_mutex. */
1859 void dlm_scan_timeout(struct dlm_ls *ls)
1862 struct dlm_lkb *lkb;
1863 int do_cancel, do_warn;
1867 if (dlm_locking_stopped(ls))
1872 mutex_lock(&ls->ls_timeout_mutex);
1873 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1875 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1876 lkb->lkb_timestamp));
1878 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1879 wait_us >= (lkb->lkb_timeout_cs * 10000))
1882 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1883 wait_us >= dlm_config.ci_timewarn_cs * 10000)
1886 if (!do_cancel && !do_warn)
1891 mutex_unlock(&ls->ls_timeout_mutex);
1893 if (!do_cancel && !do_warn)
1896 r = lkb->lkb_resource;
1901 /* clear flag so we only warn once */
1902 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1903 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1905 dlm_timeout_warn(lkb);
1909 log_debug(ls, "timeout cancel %x node %d %s",
1910 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1911 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1912 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1914 _cancel_lock(r, lkb);
1923 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1924 dlm_recoverd before checking/setting ls_recover_begin. */
1926 void dlm_adjust_timeouts(struct dlm_ls *ls)
1928 struct dlm_lkb *lkb;
1929 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1931 ls->ls_recover_begin = 0;
1932 mutex_lock(&ls->ls_timeout_mutex);
1933 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1934 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1935 mutex_unlock(&ls->ls_timeout_mutex);
1937 if (!dlm_config.ci_waitwarn_us)
1940 mutex_lock(&ls->ls_waiters_mutex);
1941 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1942 if (ktime_to_us(lkb->lkb_wait_time))
1943 lkb->lkb_wait_time = ktime_get();
1945 mutex_unlock(&ls->ls_waiters_mutex);
1948 /* lkb is master or local copy */
1950 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1952 int b, len = r->res_ls->ls_lvblen;
1954 /* b=1 lvb returned to caller
1955 b=0 lvb written to rsb or invalidated
1958 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1961 if (!lkb->lkb_lvbptr)
1964 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1970 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1971 lkb->lkb_lvbseq = r->res_lvbseq;
1973 } else if (b == 0) {
1974 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1975 rsb_set_flag(r, RSB_VALNOTVALID);
1979 if (!lkb->lkb_lvbptr)
1982 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1986 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1991 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1993 lkb->lkb_lvbseq = r->res_lvbseq;
1994 rsb_clear_flag(r, RSB_VALNOTVALID);
1997 if (rsb_flag(r, RSB_VALNOTVALID))
1998 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
2001 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2003 if (lkb->lkb_grmode < DLM_LOCK_PW)
2006 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
2007 rsb_set_flag(r, RSB_VALNOTVALID);
2011 if (!lkb->lkb_lvbptr)
2014 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2018 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
2023 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2025 rsb_clear_flag(r, RSB_VALNOTVALID);
2028 /* lkb is process copy (pc) */
2030 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2031 struct dlm_message *ms)
2035 if (!lkb->lkb_lvbptr)
2038 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2041 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
2043 int len = receive_extralen(ms);
2044 if (len > r->res_ls->ls_lvblen)
2045 len = r->res_ls->ls_lvblen;
2046 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2047 lkb->lkb_lvbseq = ms->m_lvbseq;
2051 /* Manipulate lkb's on rsb's convert/granted/waiting queues
2052 remove_lock -- used for unlock, removes lkb from granted
2053 revert_lock -- used for cancel, moves lkb from convert to granted
2054 grant_lock -- used for request and convert, adds lkb to granted or
2055 moves lkb from convert or waiting to granted
2057 Each of these is used for master or local copy lkb's. There is
2058 also a _pc() variation used to make the corresponding change on
2059 a process copy (pc) lkb. */
2061 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2064 lkb->lkb_grmode = DLM_LOCK_IV;
2065 /* this unhold undoes the original ref from create_lkb()
2066 so this leads to the lkb being freed */
2070 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2072 set_lvb_unlock(r, lkb);
2073 _remove_lock(r, lkb);
2076 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2078 _remove_lock(r, lkb);
2081 /* returns: 0 did nothing
2082 1 moved lock to granted
2085 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2089 lkb->lkb_rqmode = DLM_LOCK_IV;
2091 switch (lkb->lkb_status) {
2092 case DLM_LKSTS_GRANTED:
2094 case DLM_LKSTS_CONVERT:
2095 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2098 case DLM_LKSTS_WAITING:
2100 lkb->lkb_grmode = DLM_LOCK_IV;
2101 /* this unhold undoes the original ref from create_lkb()
2102 so this leads to the lkb being freed */
2107 log_print("invalid status for revert %d", lkb->lkb_status);
2112 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2114 return revert_lock(r, lkb);
2117 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2119 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2120 lkb->lkb_grmode = lkb->lkb_rqmode;
2121 if (lkb->lkb_status)
2122 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2124 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2127 lkb->lkb_rqmode = DLM_LOCK_IV;
2128 lkb->lkb_highbast = 0;
2131 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2133 set_lvb_lock(r, lkb);
2134 _grant_lock(r, lkb);
2137 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2138 struct dlm_message *ms)
2140 set_lvb_lock_pc(r, lkb, ms);
2141 _grant_lock(r, lkb);
2144 /* called by grant_pending_locks() which means an async grant message must
2145 be sent to the requesting node in addition to granting the lock if the
2146 lkb belongs to a remote node. */
2148 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2151 if (is_master_copy(lkb))
2154 queue_cast(r, lkb, 0);
2157 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2158 change the granted/requested modes. We're munging things accordingly in
2160 CONVDEADLK: our grmode may have been forced down to NL to resolve a
2162 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2163 compatible with other granted locks */
2165 static void munge_demoted(struct dlm_lkb *lkb)
2167 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2168 log_print("munge_demoted %x invalid modes gr %d rq %d",
2169 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2173 lkb->lkb_grmode = DLM_LOCK_NL;
2176 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2178 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
2179 ms->m_type != DLM_MSG_GRANT) {
2180 log_print("munge_altmode %x invalid reply type %d",
2181 lkb->lkb_id, ms->m_type);
2185 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2186 lkb->lkb_rqmode = DLM_LOCK_PR;
2187 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2188 lkb->lkb_rqmode = DLM_LOCK_CW;
2190 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2195 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2197 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2199 if (lkb->lkb_id == first->lkb_id)
2205 /* Check if the given lkb conflicts with another lkb on the queue. */
2207 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2209 struct dlm_lkb *this;
2211 list_for_each_entry(this, head, lkb_statequeue) {
2214 if (!modes_compat(this, lkb))
2221 * "A conversion deadlock arises with a pair of lock requests in the converting
2222 * queue for one resource. The granted mode of each lock blocks the requested
2223 * mode of the other lock."
2225 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2226 * convert queue from being granted, then deadlk/demote lkb.
2229 * Granted Queue: empty
2230 * Convert Queue: NL->EX (first lock)
2231 * PR->EX (second lock)
2233 * The first lock can't be granted because of the granted mode of the second
2234 * lock and the second lock can't be granted because it's not first in the
2235 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2236 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2237 * flag set and return DEMOTED in the lksb flags.
2239 * Originally, this function detected conv-deadlk in a more limited scope:
2240 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2241 * - if lkb1 was the first entry in the queue (not just earlier), and was
2242 * blocked by the granted mode of lkb2, and there was nothing on the
2243 * granted queue preventing lkb1 from being granted immediately, i.e.
2244 * lkb2 was the only thing preventing lkb1 from being granted.
2246 * That second condition meant we'd only say there was conv-deadlk if
2247 * resolving it (by demotion) would lead to the first lock on the convert
2248 * queue being granted right away. It allowed conversion deadlocks to exist
2249 * between locks on the convert queue while they couldn't be granted anyway.
2251 * Now, we detect and take action on conversion deadlocks immediately when
2252 * they're created, even if they may not be immediately consequential. If
2253 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2254 * mode that would prevent lkb1's conversion from being granted, we do a
2255 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2256 * I think this means that the lkb_is_ahead condition below should always
2257 * be zero, i.e. there will never be conv-deadlk between two locks that are
2258 * both already on the convert queue.
2261 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2263 struct dlm_lkb *lkb1;
2264 int lkb_is_ahead = 0;
2266 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2272 if (!lkb_is_ahead) {
2273 if (!modes_compat(lkb2, lkb1))
2276 if (!modes_compat(lkb2, lkb1) &&
2277 !modes_compat(lkb1, lkb2))
2285 * Return 1 if the lock can be granted, 0 otherwise.
2286 * Also detect and resolve conversion deadlocks.
2288 * lkb is the lock to be granted
2290 * now is 1 if the function is being called in the context of the
2291 * immediate request, it is 0 if called later, after the lock has been
2294 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2297 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2300 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2303 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2306 * 6-10: Version 5.4 introduced an option to address the phenomenon of
2307 * a new request for a NL mode lock being blocked.
2309 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2310 * request, then it would be granted. In essence, the use of this flag
2311 * tells the Lock Manager to expedite theis request by not considering
2312 * what may be in the CONVERTING or WAITING queues... As of this
2313 * writing, the EXPEDITE flag can be used only with new requests for NL
2314 * mode locks. This flag is not valid for conversion requests.
2316 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
2317 * conversion or used with a non-NL requested mode. We also know an
2318 * EXPEDITE request is always granted immediately, so now must always
2319 * be 1. The full condition to grant an expedite request: (now &&
2320 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2321 * therefore be shortened to just checking the flag.
2324 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2328 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2329 * added to the remaining conditions.
2332 if (queue_conflict(&r->res_grantqueue, lkb))
2336 * 6-3: By default, a conversion request is immediately granted if the
2337 * requested mode is compatible with the modes of all other granted
2341 if (queue_conflict(&r->res_convertqueue, lkb))
2345 * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2346 * locks for a recovered rsb, on which lkb's have been rebuilt.
2347 * The lkb's may have been rebuilt on the queues in a different
2348 * order than they were in on the previous master. So, granting
2349 * queued conversions in order after recovery doesn't make sense
2350 * since the order hasn't been preserved anyway. The new order
2351 * could also have created a new "in place" conversion deadlock.
2352 * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2353 * After recovery, there would be no granted locks, and possibly
2354 * NL->EX, PR->EX, an in-place conversion deadlock.) So, after
2355 * recovery, grant conversions without considering order.
2358 if (conv && recover)
2362 * 6-5: But the default algorithm for deciding whether to grant or
2363 * queue conversion requests does not by itself guarantee that such
2364 * requests are serviced on a "first come first serve" basis. This, in
2365 * turn, can lead to a phenomenon known as "indefinate postponement".
2367 * 6-7: This issue is dealt with by using the optional QUECVT flag with
2368 * the system service employed to request a lock conversion. This flag
2369 * forces certain conversion requests to be queued, even if they are
2370 * compatible with the granted modes of other locks on the same
2371 * resource. Thus, the use of this flag results in conversion requests
2372 * being ordered on a "first come first servce" basis.
2374 * DCT: This condition is all about new conversions being able to occur
2375 * "in place" while the lock remains on the granted queue (assuming
2376 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
2377 * doesn't _have_ to go onto the convert queue where it's processed in
2378 * order. The "now" variable is necessary to distinguish converts
2379 * being received and processed for the first time now, because once a
2380 * convert is moved to the conversion queue the condition below applies
2381 * requiring fifo granting.
2384 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2388 * Even if the convert is compat with all granted locks,
2389 * QUECVT forces it behind other locks on the convert queue.
2392 if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2393 if (list_empty(&r->res_convertqueue))
2400 * The NOORDER flag is set to avoid the standard vms rules on grant
2404 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2408 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2409 * granted until all other conversion requests ahead of it are granted
2413 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2417 * 6-4: By default, a new request is immediately granted only if all
2418 * three of the following conditions are satisfied when the request is
2420 * - The queue of ungranted conversion requests for the resource is
2422 * - The queue of ungranted new requests for the resource is empty.
2423 * - The mode of the new request is compatible with the most
2424 * restrictive mode of all granted locks on the resource.
2427 if (now && !conv && list_empty(&r->res_convertqueue) &&
2428 list_empty(&r->res_waitqueue))
2432 * 6-4: Once a lock request is in the queue of ungranted new requests,
2433 * it cannot be granted until the queue of ungranted conversion
2434 * requests is empty, all ungranted new requests ahead of it are
2435 * granted and/or canceled, and it is compatible with the granted mode
2436 * of the most restrictive lock granted on the resource.
2439 if (!now && !conv && list_empty(&r->res_convertqueue) &&
2440 first_in_list(lkb, &r->res_waitqueue))
2446 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2447 int recover, int *err)
2450 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2451 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2456 rv = _can_be_granted(r, lkb, now, recover);
2461 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2462 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2463 * cancels one of the locks.
2466 if (is_convert && can_be_queued(lkb) &&
2467 conversion_deadlock_detect(r, lkb)) {
2468 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2469 lkb->lkb_grmode = DLM_LOCK_NL;
2470 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2471 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
2475 log_print("can_be_granted deadlock %x now %d",
2484 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2485 * to grant a request in a mode other than the normal rqmode. It's a
2486 * simple way to provide a big optimization to applications that can
2490 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2492 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2496 lkb->lkb_rqmode = alt;
2497 rv = _can_be_granted(r, lkb, now, 0);
2499 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2501 lkb->lkb_rqmode = rqmode;
2507 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
2508 for locks pending on the convert list. Once verified (watch for these
2509 log_prints), we should be able to just call _can_be_granted() and not
2510 bother with the demote/deadlk cases here (and there's no easy way to deal
2511 with a deadlk here, we'd have to generate something like grant_lock with
2512 the deadlk error.) */
2514 /* Returns the highest requested mode of all blocked conversions; sets
2515 cw if there's a blocked conversion to DLM_LOCK_CW. */
2517 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2518 unsigned int *count)
2520 struct dlm_lkb *lkb, *s;
2521 int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2522 int hi, demoted, quit, grant_restart, demote_restart;
2531 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2532 demoted = is_demoted(lkb);
2535 if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2536 grant_lock_pending(r, lkb);
2543 if (!demoted && is_demoted(lkb)) {
2544 log_print("WARN: pending demoted %x node %d %s",
2545 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2551 log_print("WARN: pending deadlock %x node %d %s",
2552 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2557 hi = max_t(int, lkb->lkb_rqmode, hi);
2559 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2565 if (demote_restart && !quit) {
2570 return max_t(int, high, hi);
2573 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2574 unsigned int *count)
2576 struct dlm_lkb *lkb, *s;
2578 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2579 if (can_be_granted(r, lkb, 0, 0, NULL)) {
2580 grant_lock_pending(r, lkb);
2584 high = max_t(int, lkb->lkb_rqmode, high);
2585 if (lkb->lkb_rqmode == DLM_LOCK_CW)
2593 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2594 on either the convert or waiting queue.
2595 high is the largest rqmode of all locks blocked on the convert or
2598 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2600 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2601 if (gr->lkb_highbast < DLM_LOCK_EX)
2606 if (gr->lkb_highbast < high &&
2607 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2612 static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2614 struct dlm_lkb *lkb, *s;
2615 int high = DLM_LOCK_IV;
2618 if (!is_master(r)) {
2619 log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2624 high = grant_pending_convert(r, high, &cw, count);
2625 high = grant_pending_wait(r, high, &cw, count);
2627 if (high == DLM_LOCK_IV)
2631 * If there are locks left on the wait/convert queue then send blocking
2632 * ASTs to granted locks based on the largest requested mode (high)
2636 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2637 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2638 if (cw && high == DLM_LOCK_PR &&
2639 lkb->lkb_grmode == DLM_LOCK_PR)
2640 queue_bast(r, lkb, DLM_LOCK_CW);
2642 queue_bast(r, lkb, high);
2643 lkb->lkb_highbast = high;
2648 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2650 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2651 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2652 if (gr->lkb_highbast < DLM_LOCK_EX)
2657 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2662 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2663 struct dlm_lkb *lkb)
2667 list_for_each_entry(gr, head, lkb_statequeue) {
2668 /* skip self when sending basts to convertqueue */
2671 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2672 queue_bast(r, gr, lkb->lkb_rqmode);
2673 gr->lkb_highbast = lkb->lkb_rqmode;
2678 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2680 send_bast_queue(r, &r->res_grantqueue, lkb);
2683 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2685 send_bast_queue(r, &r->res_grantqueue, lkb);
2686 send_bast_queue(r, &r->res_convertqueue, lkb);
2689 /* set_master(r, lkb) -- set the master nodeid of a resource
2691 The purpose of this function is to set the nodeid field in the given
2692 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2693 known, it can just be copied to the lkb and the function will return
2694 0. If the rsb's nodeid is _not_ known, it needs to be looked up
2695 before it can be copied to the lkb.
2697 When the rsb nodeid is being looked up remotely, the initial lkb
2698 causing the lookup is kept on the ls_waiters list waiting for the
2699 lookup reply. Other lkb's waiting for the same rsb lookup are kept
2700 on the rsb's res_lookup list until the master is verified.
2703 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2704 1: the rsb master is not available and the lkb has been placed on
2708 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2710 int our_nodeid = dlm_our_nodeid();
2712 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2713 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2714 r->res_first_lkid = lkb->lkb_id;
2715 lkb->lkb_nodeid = r->res_nodeid;
2719 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2720 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2724 if (r->res_master_nodeid == our_nodeid) {
2725 lkb->lkb_nodeid = 0;
2729 if (r->res_master_nodeid) {
2730 lkb->lkb_nodeid = r->res_master_nodeid;
2734 if (dlm_dir_nodeid(r) == our_nodeid) {
2735 /* This is a somewhat unusual case; find_rsb will usually
2736 have set res_master_nodeid when dir nodeid is local, but
2737 there are cases where we become the dir node after we've
2738 past find_rsb and go through _request_lock again.
2739 confirm_master() or process_lookup_list() needs to be
2740 called after this. */
2741 log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2742 lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2744 r->res_master_nodeid = our_nodeid;
2746 lkb->lkb_nodeid = 0;
2750 wait_pending_remove(r);
2752 r->res_first_lkid = lkb->lkb_id;
2753 send_lookup(r, lkb);
2757 static void process_lookup_list(struct dlm_rsb *r)
2759 struct dlm_lkb *lkb, *safe;
2761 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2762 list_del_init(&lkb->lkb_rsb_lookup);
2763 _request_lock(r, lkb);
2768 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
2770 static void confirm_master(struct dlm_rsb *r, int error)
2772 struct dlm_lkb *lkb;
2774 if (!r->res_first_lkid)
2780 r->res_first_lkid = 0;
2781 process_lookup_list(r);
2787 /* the remote request failed and won't be retried (it was
2788 a NOQUEUE, or has been canceled/unlocked); make a waiting
2789 lkb the first_lkid */
2791 r->res_first_lkid = 0;
2793 if (!list_empty(&r->res_lookup)) {
2794 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2796 list_del_init(&lkb->lkb_rsb_lookup);
2797 r->res_first_lkid = lkb->lkb_id;
2798 _request_lock(r, lkb);
2803 log_error(r->res_ls, "confirm_master unknown error %d", error);
2807 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2808 int namelen, unsigned long timeout_cs,
2809 void (*ast) (void *astparam),
2811 void (*bast) (void *astparam, int mode),
2812 struct dlm_args *args)
2816 /* check for invalid arg usage */
2818 if (mode < 0 || mode > DLM_LOCK_EX)
2821 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2824 if (flags & DLM_LKF_CANCEL)
2827 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2830 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2833 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2836 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2839 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2842 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2845 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2851 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2854 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2857 /* these args will be copied to the lkb in validate_lock_args,
2858 it cannot be done now because when converting locks, fields in
2859 an active lkb cannot be modified before locking the rsb */
2861 args->flags = flags;
2863 args->astparam = astparam;
2864 args->bastfn = bast;
2865 args->timeout = timeout_cs;
2873 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2875 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2876 DLM_LKF_FORCEUNLOCK))
2879 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2882 args->flags = flags;
2883 args->astparam = astarg;
2887 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2888 struct dlm_args *args)
2892 if (args->flags & DLM_LKF_CONVERT) {
2893 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2896 if (lkb->lkb_wait_type)
2899 if (is_overlap(lkb))
2903 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2906 if (args->flags & DLM_LKF_QUECVT &&
2907 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2911 lkb->lkb_exflags = args->flags;
2912 lkb->lkb_sbflags = 0;
2913 lkb->lkb_astfn = args->astfn;
2914 lkb->lkb_astparam = args->astparam;
2915 lkb->lkb_bastfn = args->bastfn;
2916 lkb->lkb_rqmode = args->mode;
2917 lkb->lkb_lksb = args->lksb;
2918 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2919 lkb->lkb_ownpid = (int) current->pid;
2920 lkb->lkb_timeout_cs = args->timeout;
2924 log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2925 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2926 lkb->lkb_status, lkb->lkb_wait_type,
2927 lkb->lkb_resource->res_name);
2931 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2934 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2935 because there may be a lookup in progress and it's valid to do
2936 cancel/unlockf on it */
2938 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2940 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2943 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2944 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2949 /* an lkb may still exist even though the lock is EOL'ed due to a
2950 cancel, unlock or failed noqueue request; an app can't use these
2951 locks; return same error as if the lkid had not been found at all */
2953 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2954 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2959 /* an lkb may be waiting for an rsb lookup to complete where the
2960 lookup was initiated by another lock */
2962 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2963 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2964 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2965 list_del_init(&lkb->lkb_rsb_lookup);
2966 queue_cast(lkb->lkb_resource, lkb,
2967 args->flags & DLM_LKF_CANCEL ?
2968 -DLM_ECANCEL : -DLM_EUNLOCK);
2969 unhold_lkb(lkb); /* undoes create_lkb() */
2971 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2976 /* cancel not allowed with another cancel/unlock in progress */
2978 if (args->flags & DLM_LKF_CANCEL) {
2979 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2982 if (is_overlap(lkb))
2985 /* don't let scand try to do a cancel */
2988 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2989 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2994 /* there's nothing to cancel */
2995 if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2996 !lkb->lkb_wait_type) {
3001 switch (lkb->lkb_wait_type) {
3002 case DLM_MSG_LOOKUP:
3003 case DLM_MSG_REQUEST:
3004 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3007 case DLM_MSG_UNLOCK:
3008 case DLM_MSG_CANCEL:
3011 /* add_to_waiters() will set OVERLAP_CANCEL */
3015 /* do we need to allow a force-unlock if there's a normal unlock
3016 already in progress? in what conditions could the normal unlock
3017 fail such that we'd want to send a force-unlock to be sure? */
3019 if (args->flags & DLM_LKF_FORCEUNLOCK) {
3020 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3023 if (is_overlap_unlock(lkb))
3026 /* don't let scand try to do a cancel */
3029 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3030 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3035 switch (lkb->lkb_wait_type) {
3036 case DLM_MSG_LOOKUP:
3037 case DLM_MSG_REQUEST:
3038 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3041 case DLM_MSG_UNLOCK:
3044 /* add_to_waiters() will set OVERLAP_UNLOCK */
3048 /* normal unlock not allowed if there's any op in progress */
3050 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
3054 /* an overlapping op shouldn't blow away exflags from other op */
3055 lkb->lkb_exflags |= args->flags;
3056 lkb->lkb_sbflags = 0;
3057 lkb->lkb_astparam = args->astparam;
3061 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
3062 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3063 args->flags, lkb->lkb_wait_type,
3064 lkb->lkb_resource->res_name);
3069 * Four stage 4 varieties:
3070 * do_request(), do_convert(), do_unlock(), do_cancel()
3071 * These are called on the master node for the given lock and
3072 * from the central locking logic.
3075 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3079 if (can_be_granted(r, lkb, 1, 0, NULL)) {
3081 queue_cast(r, lkb, 0);
3085 if (can_be_queued(lkb)) {
3086 error = -EINPROGRESS;
3087 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3093 queue_cast(r, lkb, -EAGAIN);
3098 static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3103 if (force_blocking_asts(lkb))
3104 send_blocking_asts_all(r, lkb);
3107 send_blocking_asts(r, lkb);
3112 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3117 /* changing an existing lock may allow others to be granted */
3119 if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3121 queue_cast(r, lkb, 0);
3125 /* can_be_granted() detected that this lock would block in a conversion
3126 deadlock, so we leave it on the granted queue and return EDEADLK in
3127 the ast for the convert. */
3130 /* it's left on the granted queue */
3131 revert_lock(r, lkb);
3132 queue_cast(r, lkb, -EDEADLK);
3137 /* is_demoted() means the can_be_granted() above set the grmode
3138 to NL, and left us on the granted queue. This auto-demotion
3139 (due to CONVDEADLK) might mean other locks, and/or this lock, are
3140 now grantable. We have to try to grant other converting locks
3141 before we try again to grant this one. */
3143 if (is_demoted(lkb)) {
3144 grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3145 if (_can_be_granted(r, lkb, 1, 0)) {
3147 queue_cast(r, lkb, 0);
3150 /* else fall through and move to convert queue */
3153 if (can_be_queued(lkb)) {
3154 error = -EINPROGRESS;
3156 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3162 queue_cast(r, lkb, -EAGAIN);
3167 static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3172 grant_pending_locks(r, NULL);
3173 /* grant_pending_locks also sends basts */
3176 if (force_blocking_asts(lkb))
3177 send_blocking_asts_all(r, lkb);
3180 send_blocking_asts(r, lkb);
3185 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3187 remove_lock(r, lkb);
3188 queue_cast(r, lkb, -DLM_EUNLOCK);
3189 return -DLM_EUNLOCK;
3192 static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3195 grant_pending_locks(r, NULL);
3198 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3200 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3204 error = revert_lock(r, lkb);
3206 queue_cast(r, lkb, -DLM_ECANCEL);
3207 return -DLM_ECANCEL;
3212 static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3216 grant_pending_locks(r, NULL);
3220 * Four stage 3 varieties:
3221 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3224 /* add a new lkb to a possibly new rsb, called by requesting process */
3226 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3230 /* set_master: sets lkb nodeid from r */
3232 error = set_master(r, lkb);
3241 /* receive_request() calls do_request() on remote node */
3242 error = send_request(r, lkb);
3244 error = do_request(r, lkb);
3245 /* for remote locks the request_reply is sent
3246 between do_request and do_request_effects */
3247 do_request_effects(r, lkb, error);
3253 /* change some property of an existing lkb, e.g. mode */
3255 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3260 /* receive_convert() calls do_convert() on remote node */
3261 error = send_convert(r, lkb);
3263 error = do_convert(r, lkb);
3264 /* for remote locks the convert_reply is sent
3265 between do_convert and do_convert_effects */
3266 do_convert_effects(r, lkb, error);
3272 /* remove an existing lkb from the granted queue */
3274 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3279 /* receive_unlock() calls do_unlock() on remote node */
3280 error = send_unlock(r, lkb);
3282 error = do_unlock(r, lkb);
3283 /* for remote locks the unlock_reply is sent
3284 between do_unlock and do_unlock_effects */
3285 do_unlock_effects(r, lkb, error);
3291 /* remove an existing lkb from the convert or wait queue */
3293 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3298 /* receive_cancel() calls do_cancel() on remote node */
3299 error = send_cancel(r, lkb);
3301 error = do_cancel(r, lkb);
3302 /* for remote locks the cancel_reply is sent
3303 between do_cancel and do_cancel_effects */
3304 do_cancel_effects(r, lkb, error);
3311 * Four stage 2 varieties:
3312 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3315 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
3316 int len, struct dlm_args *args)
3321 error = validate_lock_args(ls, lkb, args);
3325 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3332 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3334 error = _request_lock(r, lkb);
3341 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3342 struct dlm_args *args)
3347 r = lkb->lkb_resource;
3352 error = validate_lock_args(ls, lkb, args);
3356 error = _convert_lock(r, lkb);
3363 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3364 struct dlm_args *args)
3369 r = lkb->lkb_resource;
3374 error = validate_unlock_args(lkb, args);
3378 error = _unlock_lock(r, lkb);
3385 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3386 struct dlm_args *args)
3391 r = lkb->lkb_resource;
3396 error = validate_unlock_args(lkb, args);
3400 error = _cancel_lock(r, lkb);
3408 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
3411 int dlm_lock(dlm_lockspace_t *lockspace,
3413 struct dlm_lksb *lksb,
3416 unsigned int namelen,
3417 uint32_t parent_lkid,
3418 void (*ast) (void *astarg),
3420 void (*bast) (void *astarg, int mode))
3423 struct dlm_lkb *lkb;
3424 struct dlm_args args;
3425 int error, convert = flags & DLM_LKF_CONVERT;
3427 ls = dlm_find_lockspace_local(lockspace);
3431 dlm_lock_recovery(ls);
3434 error = find_lkb(ls, lksb->sb_lkid, &lkb);
3436 error = create_lkb(ls, &lkb);
3441 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3442 astarg, bast, &args);
3447 error = convert_lock(ls, lkb, &args);
3449 error = request_lock(ls, lkb, name, namelen, &args);
3451 if (error == -EINPROGRESS)
3454 if (convert || error)
3456 if (error == -EAGAIN || error == -EDEADLK)
3459 dlm_unlock_recovery(ls);
3460 dlm_put_lockspace(ls);
3464 int dlm_unlock(dlm_lockspace_t *lockspace,
3467 struct dlm_lksb *lksb,
3471 struct dlm_lkb *lkb;
3472 struct dlm_args args;
3475 ls = dlm_find_lockspace_local(lockspace);
3479 dlm_lock_recovery(ls);
3481 error = find_lkb(ls, lkid, &lkb);
3485 error = set_unlock_args(flags, astarg, &args);
3489 if (flags & DLM_LKF_CANCEL)
3490 error = cancel_lock(ls, lkb, &args);
3492 error = unlock_lock(ls, lkb, &args);
3494 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3496 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3501 dlm_unlock_recovery(ls);
3502 dlm_put_lockspace(ls);
3507 * send/receive routines for remote operations and replies
3511 * send_request receive_request
3512 * send_convert receive_convert
3513 * send_unlock receive_unlock
3514 * send_cancel receive_cancel
3515 * send_grant receive_grant
3516 * send_bast receive_bast
3517 * send_lookup receive_lookup
3518 * send_remove receive_remove
3521 * receive_request_reply send_request_reply
3522 * receive_convert_reply send_convert_reply
3523 * receive_unlock_reply send_unlock_reply
3524 * receive_cancel_reply send_cancel_reply
3525 * receive_lookup_reply send_lookup_reply
3528 static int _create_message(struct dlm_ls *ls, int mb_len,
3529 int to_nodeid, int mstype,
3530 struct dlm_message **ms_ret,
3531 struct dlm_mhandle **mh_ret)
3533 struct dlm_message *ms;
3534 struct dlm_mhandle *mh;
3537 /* get_buffer gives us a message handle (mh) that we need to
3538 pass into lowcomms_commit and a message buffer (mb) that we
3539 write our data into */
3541 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
3545 memset(mb, 0, mb_len);
3547 ms = (struct dlm_message *) mb;
3549 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3550 ms->m_header.h_lockspace = ls->ls_global_id;
3551 ms->m_header.h_nodeid = dlm_our_nodeid();
3552 ms->m_header.h_length = mb_len;
3553 ms->m_header.h_cmd = DLM_MSG;
3555 ms->m_type = mstype;
3562 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3563 int to_nodeid, int mstype,
3564 struct dlm_message **ms_ret,
3565 struct dlm_mhandle **mh_ret)
3567 int mb_len = sizeof(struct dlm_message);
3570 case DLM_MSG_REQUEST:
3571 case DLM_MSG_LOOKUP:
3572 case DLM_MSG_REMOVE:
3573 mb_len += r->res_length;
3575 case DLM_MSG_CONVERT:
3576 case DLM_MSG_UNLOCK:
3577 case DLM_MSG_REQUEST_REPLY:
3578 case DLM_MSG_CONVERT_REPLY:
3580 if (lkb && lkb->lkb_lvbptr)
3581 mb_len += r->res_ls->ls_lvblen;
3585 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3589 /* further lowcomms enhancements or alternate implementations may make
3590 the return value from this function useful at some point */
3592 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
3594 dlm_message_out(ms);
3595 dlm_lowcomms_commit_buffer(mh);
3599 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3600 struct dlm_message *ms)
3602 ms->m_nodeid = lkb->lkb_nodeid;
3603 ms->m_pid = lkb->lkb_ownpid;
3604 ms->m_lkid = lkb->lkb_id;
3605 ms->m_remid = lkb->lkb_remid;
3606 ms->m_exflags = lkb->lkb_exflags;
3607 ms->m_sbflags = lkb->lkb_sbflags;
3608 ms->m_flags = lkb->lkb_flags;
3609 ms->m_lvbseq = lkb->lkb_lvbseq;
3610 ms->m_status = lkb->lkb_status;
3611 ms->m_grmode = lkb->lkb_grmode;
3612 ms->m_rqmode = lkb->lkb_rqmode;
3613 ms->m_hash = r->res_hash;
3615 /* m_result and m_bastmode are set from function args,
3616 not from lkb fields */
3618 if (lkb->lkb_bastfn)
3619 ms->m_asts |= DLM_CB_BAST;
3621 ms->m_asts |= DLM_CB_CAST;
3623 /* compare with switch in create_message; send_remove() doesn't
3626 switch (ms->m_type) {
3627 case DLM_MSG_REQUEST:
3628 case DLM_MSG_LOOKUP:
3629 memcpy(ms->m_extra, r->res_name, r->res_length);
3631 case DLM_MSG_CONVERT:
3632 case DLM_MSG_UNLOCK:
3633 case DLM_MSG_REQUEST_REPLY:
3634 case DLM_MSG_CONVERT_REPLY:
3636 if (!lkb->lkb_lvbptr)
3638 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3643 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3645 struct dlm_message *ms;
3646 struct dlm_mhandle *mh;
3647 int to_nodeid, error;
3649 to_nodeid = r->res_nodeid;
3651 error = add_to_waiters(lkb, mstype, to_nodeid);
3655 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3659 send_args(r, lkb, ms);
3661 error = send_message(mh, ms);
3667 remove_from_waiters(lkb, msg_reply_type(mstype));
3671 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3673 return send_common(r, lkb, DLM_MSG_REQUEST);
3676 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3680 error = send_common(r, lkb, DLM_MSG_CONVERT);
3682 /* down conversions go without a reply from the master */
3683 if (!error && down_conversion(lkb)) {
3684 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3685 r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS;
3686 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3687 r->res_ls->ls_stub_ms.m_result = 0;
3688 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3694 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
3695 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3696 that the master is still correct. */
3698 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3700 return send_common(r, lkb, DLM_MSG_UNLOCK);
3703 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3705 return send_common(r, lkb, DLM_MSG_CANCEL);
3708 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3710 struct dlm_message *ms;
3711 struct dlm_mhandle *mh;
3712 int to_nodeid, error;
3714 to_nodeid = lkb->lkb_nodeid;
3716 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3720 send_args(r, lkb, ms);
3724 error = send_message(mh, ms);
3729 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3731 struct dlm_message *ms;
3732 struct dlm_mhandle *mh;
3733 int to_nodeid, error;
3735 to_nodeid = lkb->lkb_nodeid;
3737 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3741 send_args(r, lkb, ms);
3743 ms->m_bastmode = mode;
3745 error = send_message(mh, ms);
3750 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3752 struct dlm_message *ms;
3753 struct dlm_mhandle *mh;
3754 int to_nodeid, error;
3756 to_nodeid = dlm_dir_nodeid(r);
3758 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3762 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3766 send_args(r, lkb, ms);
3768 error = send_message(mh, ms);
3774 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3778 static int send_remove(struct dlm_rsb *r)
3780 struct dlm_message *ms;
3781 struct dlm_mhandle *mh;
3782 int to_nodeid, error;
3784 to_nodeid = dlm_dir_nodeid(r);
3786 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3790 memcpy(ms->m_extra, r->res_name, r->res_length);
3791 ms->m_hash = r->res_hash;
3793 error = send_message(mh, ms);
3798 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3801 struct dlm_message *ms;
3802 struct dlm_mhandle *mh;
3803 int to_nodeid, error;
3805 to_nodeid = lkb->lkb_nodeid;
3807 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3811 send_args(r, lkb, ms);
3815 error = send_message(mh, ms);
3820 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3822 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3825 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3827 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3830 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3832 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3835 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3837 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3840 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3841 int ret_nodeid, int rv)
3843 struct dlm_rsb *r = &ls->ls_stub_rsb;
3844 struct dlm_message *ms;
3845 struct dlm_mhandle *mh;
3846 int error, nodeid = ms_in->m_header.h_nodeid;
3848 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3852 ms->m_lkid = ms_in->m_lkid;
3854 ms->m_nodeid = ret_nodeid;
3856 error = send_message(mh, ms);
3861 /* which args we save from a received message depends heavily on the type
3862 of message, unlike the send side where we can safely send everything about
3863 the lkb for any type of message */
3865 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3867 lkb->lkb_exflags = ms->m_exflags;
3868 lkb->lkb_sbflags = ms->m_sbflags;
3869 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3870 (ms->m_flags & 0x0000FFFF);
3873 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3875 if (ms->m_flags == DLM_IFL_STUB_MS)
3878 lkb->lkb_sbflags = ms->m_sbflags;
3879 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3880 (ms->m_flags & 0x0000FFFF);
3883 static int receive_extralen(struct dlm_message *ms)
3885 return (ms->m_header.h_length - sizeof(struct dlm_message));
3888 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3889 struct dlm_message *ms)
3893 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3894 if (!lkb->lkb_lvbptr)
3895 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3896 if (!lkb->lkb_lvbptr)
3898 len = receive_extralen(ms);
3899 if (len > ls->ls_lvblen)
3900 len = ls->ls_lvblen;
3901 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3906 static void fake_bastfn(void *astparam, int mode)
3908 log_print("fake_bastfn should not be called");
3911 static void fake_astfn(void *astparam)
3913 log_print("fake_astfn should not be called");
3916 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3917 struct dlm_message *ms)
3919 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3920 lkb->lkb_ownpid = ms->m_pid;
3921 lkb->lkb_remid = ms->m_lkid;
3922 lkb->lkb_grmode = DLM_LOCK_IV;
3923 lkb->lkb_rqmode = ms->m_rqmode;
3925 lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
3926 lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
3928 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3929 /* lkb was just created so there won't be an lvb yet */
3930 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3931 if (!lkb->lkb_lvbptr)
3938 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3939 struct dlm_message *ms)
3941 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3944 if (receive_lvb(ls, lkb, ms))
3947 lkb->lkb_rqmode = ms->m_rqmode;
3948 lkb->lkb_lvbseq = ms->m_lvbseq;
3953 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3954 struct dlm_message *ms)
3956 if (receive_lvb(ls, lkb, ms))
3961 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3962 uses to send a reply and that the remote end uses to process the reply. */
3964 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3966 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3967 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3968 lkb->lkb_remid = ms->m_lkid;
3971 /* This is called after the rsb is locked so that we can safely inspect
3972 fields in the lkb. */
3974 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3976 int from = ms->m_header.h_nodeid;
3979 /* currently mixing of user/kernel locks are not supported */
3980 if (ms->m_flags & DLM_IFL_USER && ~lkb->lkb_flags & DLM_IFL_USER) {
3981 log_error(lkb->lkb_resource->res_ls,
3982 "got user dlm message for a kernel lock");
3987 switch (ms->m_type) {
3988 case DLM_MSG_CONVERT:
3989 case DLM_MSG_UNLOCK:
3990 case DLM_MSG_CANCEL:
3991 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3995 case DLM_MSG_CONVERT_REPLY:
3996 case DLM_MSG_UNLOCK_REPLY:
3997 case DLM_MSG_CANCEL_REPLY:
4000 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
4004 case DLM_MSG_REQUEST_REPLY:
4005 if (!is_process_copy(lkb))
4007 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
4017 log_error(lkb->lkb_resource->res_ls,
4018 "ignore invalid message %d from %d %x %x %x %d",
4019 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
4020 lkb->lkb_flags, lkb->lkb_nodeid);
4024 static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
4026 char name[DLM_RESNAME_MAXLEN + 1];
4027 struct dlm_message *ms;
4028 struct dlm_mhandle *mh;
4033 memset(name, 0, sizeof(name));
4034 memcpy(name, ms_name, len);
4036 hash = jhash(name, len, 0);
4037 b = hash & (ls->ls_rsbtbl_size - 1);
4039 dir_nodeid = dlm_hash2nodeid(ls, hash);
4041 log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
4043 spin_lock(&ls->ls_rsbtbl[b].lock);
4044 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4046 spin_unlock(&ls->ls_rsbtbl[b].lock);
4047 log_error(ls, "repeat_remove on keep %s", name);
4051 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4053 spin_unlock(&ls->ls_rsbtbl[b].lock);
4054 log_error(ls, "repeat_remove on toss %s", name);
4058 /* use ls->remove_name2 to avoid conflict with shrink? */
4060 spin_lock(&ls->ls_remove_spin);
4061 ls->ls_remove_len = len;
4062 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
4063 spin_unlock(&ls->ls_remove_spin);
4064 spin_unlock(&ls->ls_rsbtbl[b].lock);
4066 rv = _create_message(ls, sizeof(struct dlm_message) + len,
4067 dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
4071 memcpy(ms->m_extra, name, len);
4074 send_message(mh, ms);
4076 spin_lock(&ls->ls_remove_spin);
4077 ls->ls_remove_len = 0;
4078 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
4079 spin_unlock(&ls->ls_remove_spin);
4082 static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4084 struct dlm_lkb *lkb;
4087 int error, namelen = 0;
4089 from_nodeid = ms->m_header.h_nodeid;
4091 error = create_lkb(ls, &lkb);
4095 receive_flags(lkb, ms);
4096 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4097 error = receive_request_args(ls, lkb, ms);
4103 /* The dir node is the authority on whether we are the master
4104 for this rsb or not, so if the master sends us a request, we should
4105 recreate the rsb if we've destroyed it. This race happens when we
4106 send a remove message to the dir node at the same time that the dir
4107 node sends us a request for the rsb. */
4109 namelen = receive_extralen(ms);
4111 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4112 R_RECEIVE_REQUEST, &r);
4120 if (r->res_master_nodeid != dlm_our_nodeid()) {
4121 error = validate_master_nodeid(ls, r, from_nodeid);
4131 error = do_request(r, lkb);
4132 send_request_reply(r, lkb, error);
4133 do_request_effects(r, lkb, error);
4138 if (error == -EINPROGRESS)
4145 /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4146 and do this receive_request again from process_lookup_list once
4147 we get the lookup reply. This would avoid a many repeated
4148 ENOTBLK request failures when the lookup reply designating us
4149 as master is delayed. */
4151 /* We could repeatedly return -EBADR here if our send_remove() is
4152 delayed in being sent/arriving/being processed on the dir node.
4153 Another node would repeatedly lookup up the master, and the dir
4154 node would continue returning our nodeid until our send_remove
4157 We send another remove message in case our previous send_remove
4158 was lost/ignored/missed somehow. */
4160 if (error != -ENOTBLK) {
4161 log_limit(ls, "receive_request %x from %d %d",
4162 ms->m_lkid, from_nodeid, error);
4165 if (namelen && error == -EBADR) {
4166 send_repeat_remove(ls, ms->m_extra, namelen);
4170 setup_stub_lkb(ls, ms);
4171 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4175 static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4177 struct dlm_lkb *lkb;
4179 int error, reply = 1;
4181 error = find_lkb(ls, ms->m_remid, &lkb);
4185 if (lkb->lkb_remid != ms->m_lkid) {
4186 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4187 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4188 (unsigned long long)lkb->lkb_recover_seq,
4189 ms->m_header.h_nodeid, ms->m_lkid);
4195 r = lkb->lkb_resource;
4200 error = validate_message(lkb, ms);
4204 receive_flags(lkb, ms);
4206 error = receive_convert_args(ls, lkb, ms);
4208 send_convert_reply(r, lkb, error);
4212 reply = !down_conversion(lkb);
4214 error = do_convert(r, lkb);
4216 send_convert_reply(r, lkb, error);
4217 do_convert_effects(r, lkb, error);
4225 setup_stub_lkb(ls, ms);
4226 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4230 static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4232 struct dlm_lkb *lkb;
4236 error = find_lkb(ls, ms->m_remid, &lkb);
4240 if (lkb->lkb_remid != ms->m_lkid) {
4241 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4242 lkb->lkb_id, lkb->lkb_remid,
4243 ms->m_header.h_nodeid, ms->m_lkid);
4249 r = lkb->lkb_resource;
4254 error = validate_message(lkb, ms);
4258 receive_flags(lkb, ms);
4260 error = receive_unlock_args(ls, lkb, ms);
4262 send_unlock_reply(r, lkb, error);
4266 error = do_unlock(r, lkb);
4267 send_unlock_reply(r, lkb, error);
4268 do_unlock_effects(r, lkb, error);
4276 setup_stub_lkb(ls, ms);
4277 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4281 static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4283 struct dlm_lkb *lkb;
4287 error = find_lkb(ls, ms->m_remid, &lkb);
4291 receive_flags(lkb, ms);
4293 r = lkb->lkb_resource;
4298 error = validate_message(lkb, ms);
4302 error = do_cancel(r, lkb);
4303 send_cancel_reply(r, lkb, error);
4304 do_cancel_effects(r, lkb, error);
4312 setup_stub_lkb(ls, ms);
4313 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4317 static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4319 struct dlm_lkb *lkb;
4323 error = find_lkb(ls, ms->m_remid, &lkb);
4327 r = lkb->lkb_resource;
4332 error = validate_message(lkb, ms);
4336 receive_flags_reply(lkb, ms);
4337 if (is_altmode(lkb))
4338 munge_altmode(lkb, ms);
4339 grant_lock_pc(r, lkb, ms);
4340 queue_cast(r, lkb, 0);
4348 static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4350 struct dlm_lkb *lkb;
4354 error = find_lkb(ls, ms->m_remid, &lkb);
4358 r = lkb->lkb_resource;
4363 error = validate_message(lkb, ms);
4367 queue_bast(r, lkb, ms->m_bastmode);
4368 lkb->lkb_highbast = ms->m_bastmode;
4376 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4378 int len, error, ret_nodeid, from_nodeid, our_nodeid;
4380 from_nodeid = ms->m_header.h_nodeid;
4381 our_nodeid = dlm_our_nodeid();
4383 len = receive_extralen(ms);
4385 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4388 /* Optimization: we're master so treat lookup as a request */
4389 if (!error && ret_nodeid == our_nodeid) {
4390 receive_request(ls, ms);
4393 send_lookup_reply(ls, ms, ret_nodeid, error);
4396 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4398 char name[DLM_RESNAME_MAXLEN+1];
4401 int rv, len, dir_nodeid, from_nodeid;
4403 from_nodeid = ms->m_header.h_nodeid;
4405 len = receive_extralen(ms);
4407 if (len > DLM_RESNAME_MAXLEN) {
4408 log_error(ls, "receive_remove from %d bad len %d",
4413 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
4414 if (dir_nodeid != dlm_our_nodeid()) {
4415 log_error(ls, "receive_remove from %d bad nodeid %d",
4416 from_nodeid, dir_nodeid);
4420 /* Look for name on rsbtbl.toss, if it's there, kill it.
4421 If it's on rsbtbl.keep, it's being used, and we should ignore this
4422 message. This is an expected race between the dir node sending a
4423 request to the master node at the same time as the master node sends
4424 a remove to the dir node. The resolution to that race is for the
4425 dir node to ignore the remove message, and the master node to
4426 recreate the master rsb when it gets a request from the dir node for
4427 an rsb it doesn't have. */
4429 memset(name, 0, sizeof(name));
4430 memcpy(name, ms->m_extra, len);
4432 hash = jhash(name, len, 0);
4433 b = hash & (ls->ls_rsbtbl_size - 1);
4435 spin_lock(&ls->ls_rsbtbl[b].lock);
4437 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4439 /* verify the rsb is on keep list per comment above */
4440 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4442 /* should not happen */
4443 log_error(ls, "receive_remove from %d not found %s",
4445 spin_unlock(&ls->ls_rsbtbl[b].lock);
4448 if (r->res_master_nodeid != from_nodeid) {
4449 /* should not happen */
4450 log_error(ls, "receive_remove keep from %d master %d",
4451 from_nodeid, r->res_master_nodeid);
4453 spin_unlock(&ls->ls_rsbtbl[b].lock);
4457 log_debug(ls, "receive_remove from %d master %d first %x %s",
4458 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4460 spin_unlock(&ls->ls_rsbtbl[b].lock);
4464 if (r->res_master_nodeid != from_nodeid) {
4465 log_error(ls, "receive_remove toss from %d master %d",
4466 from_nodeid, r->res_master_nodeid);
4468 spin_unlock(&ls->ls_rsbtbl[b].lock);
4472 if (kref_put(&r->res_ref, kill_rsb)) {
4473 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4474 spin_unlock(&ls->ls_rsbtbl[b].lock);
4477 log_error(ls, "receive_remove from %d rsb ref error",
4480 spin_unlock(&ls->ls_rsbtbl[b].lock);
4484 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4486 do_purge(ls, ms->m_nodeid, ms->m_pid);
4489 static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4491 struct dlm_lkb *lkb;
4493 int error, mstype, result;
4494 int from_nodeid = ms->m_header.h_nodeid;
4496 error = find_lkb(ls, ms->m_remid, &lkb);
4500 r = lkb->lkb_resource;
4504 error = validate_message(lkb, ms);
4508 mstype = lkb->lkb_wait_type;
4509 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4511 log_error(ls, "receive_request_reply %x remote %d %x result %d",
4512 lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result);
4517 /* Optimization: the dir node was also the master, so it took our
4518 lookup as a request and sent request reply instead of lookup reply */
4519 if (mstype == DLM_MSG_LOOKUP) {
4520 r->res_master_nodeid = from_nodeid;
4521 r->res_nodeid = from_nodeid;
4522 lkb->lkb_nodeid = from_nodeid;
4525 /* this is the value returned from do_request() on the master */
4526 result = ms->m_result;
4530 /* request would block (be queued) on remote master */
4531 queue_cast(r, lkb, -EAGAIN);
4532 confirm_master(r, -EAGAIN);
4533 unhold_lkb(lkb); /* undoes create_lkb() */
4538 /* request was queued or granted on remote master */
4539 receive_flags_reply(lkb, ms);
4540 lkb->lkb_remid = ms->m_lkid;
4541 if (is_altmode(lkb))
4542 munge_altmode(lkb, ms);
4544 add_lkb(r, lkb, DLM_LKSTS_WAITING);
4547 grant_lock_pc(r, lkb, ms);
4548 queue_cast(r, lkb, 0);
4550 confirm_master(r, result);
4555 /* find_rsb failed to find rsb or rsb wasn't master */
4556 log_limit(ls, "receive_request_reply %x from %d %d "
4557 "master %d dir %d first %x %s", lkb->lkb_id,
4558 from_nodeid, result, r->res_master_nodeid,
4559 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4561 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4562 r->res_master_nodeid != dlm_our_nodeid()) {
4563 /* cause _request_lock->set_master->send_lookup */
4564 r->res_master_nodeid = 0;
4566 lkb->lkb_nodeid = -1;
4569 if (is_overlap(lkb)) {
4570 /* we'll ignore error in cancel/unlock reply */
4571 queue_cast_overlap(r, lkb);
4572 confirm_master(r, result);
4573 unhold_lkb(lkb); /* undoes create_lkb() */
4575 _request_lock(r, lkb);
4577 if (r->res_master_nodeid == dlm_our_nodeid())
4578 confirm_master(r, 0);
4583 log_error(ls, "receive_request_reply %x error %d",
4584 lkb->lkb_id, result);
4587 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4588 log_debug(ls, "receive_request_reply %x result %d unlock",
4589 lkb->lkb_id, result);
4590 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4591 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4592 send_unlock(r, lkb);
4593 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4594 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4595 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4596 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4597 send_cancel(r, lkb);
4599 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4600 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4609 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4610 struct dlm_message *ms)
4612 /* this is the value returned from do_convert() on the master */
4613 switch (ms->m_result) {
4615 /* convert would block (be queued) on remote master */
4616 queue_cast(r, lkb, -EAGAIN);
4620 receive_flags_reply(lkb, ms);
4621 revert_lock_pc(r, lkb);
4622 queue_cast(r, lkb, -EDEADLK);
4626 /* convert was queued on remote master */
4627 receive_flags_reply(lkb, ms);
4628 if (is_demoted(lkb))
4631 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4636 /* convert was granted on remote master */
4637 receive_flags_reply(lkb, ms);
4638 if (is_demoted(lkb))
4640 grant_lock_pc(r, lkb, ms);
4641 queue_cast(r, lkb, 0);
4645 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4646 lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid,
4653 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4655 struct dlm_rsb *r = lkb->lkb_resource;
4661 error = validate_message(lkb, ms);
4665 /* stub reply can happen with waiters_mutex held */
4666 error = remove_from_waiters_ms(lkb, ms);
4670 __receive_convert_reply(r, lkb, ms);
4676 static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4678 struct dlm_lkb *lkb;
4681 error = find_lkb(ls, ms->m_remid, &lkb);
4685 _receive_convert_reply(lkb, ms);
4690 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4692 struct dlm_rsb *r = lkb->lkb_resource;
4698 error = validate_message(lkb, ms);
4702 /* stub reply can happen with waiters_mutex held */
4703 error = remove_from_waiters_ms(lkb, ms);
4707 /* this is the value returned from do_unlock() on the master */
4709 switch (ms->m_result) {
4711 receive_flags_reply(lkb, ms);
4712 remove_lock_pc(r, lkb);
4713 queue_cast(r, lkb, -DLM_EUNLOCK);
4718 log_error(r->res_ls, "receive_unlock_reply %x error %d",
4719 lkb->lkb_id, ms->m_result);
4726 static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4728 struct dlm_lkb *lkb;
4731 error = find_lkb(ls, ms->m_remid, &lkb);
4735 _receive_unlock_reply(lkb, ms);
4740 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4742 struct dlm_rsb *r = lkb->lkb_resource;
4748 error = validate_message(lkb, ms);
4752 /* stub reply can happen with waiters_mutex held */
4753 error = remove_from_waiters_ms(lkb, ms);
4757 /* this is the value returned from do_cancel() on the master */
4759 switch (ms->m_result) {
4761 receive_flags_reply(lkb, ms);
4762 revert_lock_pc(r, lkb);
4763 queue_cast(r, lkb, -DLM_ECANCEL);
4768 log_error(r->res_ls, "receive_cancel_reply %x error %d",
4769 lkb->lkb_id, ms->m_result);
4776 static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4778 struct dlm_lkb *lkb;
4781 error = find_lkb(ls, ms->m_remid, &lkb);
4785 _receive_cancel_reply(lkb, ms);
4790 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4792 struct dlm_lkb *lkb;
4794 int error, ret_nodeid;
4795 int do_lookup_list = 0;
4797 error = find_lkb(ls, ms->m_lkid, &lkb);
4799 log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid);
4803 /* ms->m_result is the value returned by dlm_master_lookup on dir node
4804 FIXME: will a non-zero error ever be returned? */
4806 r = lkb->lkb_resource;
4810 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4814 ret_nodeid = ms->m_nodeid;
4816 /* We sometimes receive a request from the dir node for this
4817 rsb before we've received the dir node's loookup_reply for it.
4818 The request from the dir node implies we're the master, so we set
4819 ourself as master in receive_request_reply, and verify here that
4820 we are indeed the master. */
4822 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4823 /* This should never happen */
4824 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4825 "master %d dir %d our %d first %x %s",
4826 lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid,
4827 r->res_master_nodeid, r->res_dir_nodeid,
4828 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4831 if (ret_nodeid == dlm_our_nodeid()) {
4832 r->res_master_nodeid = ret_nodeid;
4835 r->res_first_lkid = 0;
4836 } else if (ret_nodeid == -1) {
4837 /* the remote node doesn't believe it's the dir node */
4838 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4839 lkb->lkb_id, ms->m_header.h_nodeid);
4840 r->res_master_nodeid = 0;
4842 lkb->lkb_nodeid = -1;
4844 /* set_master() will set lkb_nodeid from r */
4845 r->res_master_nodeid = ret_nodeid;
4846 r->res_nodeid = ret_nodeid;
4849 if (is_overlap(lkb)) {
4850 log_debug(ls, "receive_lookup_reply %x unlock %x",
4851 lkb->lkb_id, lkb->lkb_flags);
4852 queue_cast_overlap(r, lkb);
4853 unhold_lkb(lkb); /* undoes create_lkb() */
4857 _request_lock(r, lkb);
4861 process_lookup_list(r);
4868 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4871 int error = 0, noent = 0;
4873 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
4874 log_limit(ls, "receive %d from non-member %d %x %x %d",
4875 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4876 ms->m_remid, ms->m_result);
4880 switch (ms->m_type) {
4882 /* messages sent to a master node */
4884 case DLM_MSG_REQUEST:
4885 error = receive_request(ls, ms);
4888 case DLM_MSG_CONVERT:
4889 error = receive_convert(ls, ms);
4892 case DLM_MSG_UNLOCK:
4893 error = receive_unlock(ls, ms);
4896 case DLM_MSG_CANCEL:
4898 error = receive_cancel(ls, ms);
4901 /* messages sent from a master node (replies to above) */
4903 case DLM_MSG_REQUEST_REPLY:
4904 error = receive_request_reply(ls, ms);
4907 case DLM_MSG_CONVERT_REPLY:
4908 error = receive_convert_reply(ls, ms);
4911 case DLM_MSG_UNLOCK_REPLY:
4912 error = receive_unlock_reply(ls, ms);
4915 case DLM_MSG_CANCEL_REPLY:
4916 error = receive_cancel_reply(ls, ms);
4919 /* messages sent from a master node (only two types of async msg) */
4923 error = receive_grant(ls, ms);
4928 error = receive_bast(ls, ms);
4931 /* messages sent to a dir node */
4933 case DLM_MSG_LOOKUP:
4934 receive_lookup(ls, ms);
4937 case DLM_MSG_REMOVE:
4938 receive_remove(ls, ms);
4941 /* messages sent from a dir node (remove has no reply) */
4943 case DLM_MSG_LOOKUP_REPLY:
4944 receive_lookup_reply(ls, ms);
4947 /* other messages */
4950 receive_purge(ls, ms);
4954 log_error(ls, "unknown message type %d", ms->m_type);
4958 * When checking for ENOENT, we're checking the result of
4959 * find_lkb(m_remid):
4961 * The lock id referenced in the message wasn't found. This may
4962 * happen in normal usage for the async messages and cancel, so
4963 * only use log_debug for them.
4965 * Some errors are expected and normal.
4968 if (error == -ENOENT && noent) {
4969 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4970 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4971 ms->m_lkid, saved_seq);
4972 } else if (error == -ENOENT) {
4973 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4974 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4975 ms->m_lkid, saved_seq);
4977 if (ms->m_type == DLM_MSG_CONVERT)
4978 dlm_dump_rsb_hash(ls, ms->m_hash);
4981 if (error == -EINVAL) {
4982 log_error(ls, "receive %d inval from %d lkid %x remid %x "
4984 ms->m_type, ms->m_header.h_nodeid,
4985 ms->m_lkid, ms->m_remid, saved_seq);
4989 /* If the lockspace is in recovery mode (locking stopped), then normal
4990 messages are saved on the requestqueue for processing after recovery is
4991 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4992 messages off the requestqueue before we process new ones. This occurs right
4993 after recovery completes when we transition from saving all messages on
4994 requestqueue, to processing all the saved messages, to processing new
4995 messages as they arrive. */
4997 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
5000 if (dlm_locking_stopped(ls)) {
5001 /* If we were a member of this lockspace, left, and rejoined,
5002 other nodes may still be sending us messages from the
5003 lockspace generation before we left. */
5004 if (!ls->ls_generation) {
5005 log_limit(ls, "receive %d from %d ignore old gen",
5006 ms->m_type, nodeid);
5010 dlm_add_requestqueue(ls, nodeid, ms);
5012 dlm_wait_requestqueue(ls);
5013 _receive_message(ls, ms, 0);
5017 /* This is called by dlm_recoverd to process messages that were saved on
5018 the requestqueue. */
5020 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
5023 _receive_message(ls, ms, saved_seq);
5026 /* This is called by the midcomms layer when something is received for
5027 the lockspace. It could be either a MSG (normal message sent as part of
5028 standard locking activity) or an RCOM (recovery message sent as part of
5029 lockspace recovery). */
5031 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
5033 struct dlm_header *hd = &p->header;
5037 switch (hd->h_cmd) {
5039 dlm_message_in(&p->message);
5040 type = p->message.m_type;
5043 dlm_rcom_in(&p->rcom);
5044 type = p->rcom.rc_type;
5047 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5051 if (hd->h_nodeid != nodeid) {
5052 log_print("invalid h_nodeid %d from %d lockspace %x",
5053 hd->h_nodeid, nodeid, hd->h_lockspace);
5057 ls = dlm_find_lockspace_global(hd->h_lockspace);
5059 if (dlm_config.ci_log_debug) {
5060 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5061 "%u from %d cmd %d type %d\n",
5062 hd->h_lockspace, nodeid, hd->h_cmd, type);
5065 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5066 dlm_send_ls_not_ready(nodeid, &p->rcom);
5070 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5071 be inactive (in this ls) before transitioning to recovery mode */
5073 down_read(&ls->ls_recv_active);
5074 if (hd->h_cmd == DLM_MSG)
5075 dlm_receive_message(ls, &p->message, nodeid);
5077 dlm_receive_rcom(ls, &p->rcom, nodeid);
5078 up_read(&ls->ls_recv_active);
5080 dlm_put_lockspace(ls);
5083 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5084 struct dlm_message *ms_stub)
5086 if (middle_conversion(lkb)) {
5088 memset(ms_stub, 0, sizeof(struct dlm_message));
5089 ms_stub->m_flags = DLM_IFL_STUB_MS;
5090 ms_stub->m_type = DLM_MSG_CONVERT_REPLY;
5091 ms_stub->m_result = -EINPROGRESS;
5092 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5093 _receive_convert_reply(lkb, ms_stub);
5095 /* Same special case as in receive_rcom_lock_args() */
5096 lkb->lkb_grmode = DLM_LOCK_IV;
5097 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5100 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5101 lkb->lkb_flags |= DLM_IFL_RESEND;
5104 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5105 conversions are async; there's no reply from the remote master */
5108 /* A waiting lkb needs recovery if the master node has failed, or
5109 the master node is changing (only when no directory is used) */
5111 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5114 if (dlm_no_directory(ls))
5117 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5123 /* Recovery for locks that are waiting for replies from nodes that are now
5124 gone. We can just complete unlocks and cancels by faking a reply from the
5125 dead node. Requests and up-conversions we flag to be resent after
5126 recovery. Down-conversions can just be completed with a fake reply like
5127 unlocks. Conversions between PR and CW need special attention. */
5129 void dlm_recover_waiters_pre(struct dlm_ls *ls)
5131 struct dlm_lkb *lkb, *safe;
5132 struct dlm_message *ms_stub;
5133 int wait_type, stub_unlock_result, stub_cancel_result;
5136 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5140 mutex_lock(&ls->ls_waiters_mutex);
5142 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5144 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5146 /* exclude debug messages about unlocks because there can be so
5147 many and they aren't very interesting */
5149 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5150 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5151 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5155 lkb->lkb_resource->res_nodeid,
5157 lkb->lkb_wait_nodeid,
5161 /* all outstanding lookups, regardless of destination will be
5162 resent after recovery is done */
5164 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5165 lkb->lkb_flags |= DLM_IFL_RESEND;
5169 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5172 wait_type = lkb->lkb_wait_type;
5173 stub_unlock_result = -DLM_EUNLOCK;
5174 stub_cancel_result = -DLM_ECANCEL;
5176 /* Main reply may have been received leaving a zero wait_type,
5177 but a reply for the overlapping op may not have been
5178 received. In that case we need to fake the appropriate
5179 reply for the overlap op. */
5182 if (is_overlap_cancel(lkb)) {
5183 wait_type = DLM_MSG_CANCEL;
5184 if (lkb->lkb_grmode == DLM_LOCK_IV)
5185 stub_cancel_result = 0;
5187 if (is_overlap_unlock(lkb)) {
5188 wait_type = DLM_MSG_UNLOCK;
5189 if (lkb->lkb_grmode == DLM_LOCK_IV)
5190 stub_unlock_result = -ENOENT;
5193 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5194 lkb->lkb_id, lkb->lkb_flags, wait_type,
5195 stub_cancel_result, stub_unlock_result);
5198 switch (wait_type) {
5200 case DLM_MSG_REQUEST:
5201 lkb->lkb_flags |= DLM_IFL_RESEND;
5204 case DLM_MSG_CONVERT:
5205 recover_convert_waiter(ls, lkb, ms_stub);
5208 case DLM_MSG_UNLOCK:
5210 memset(ms_stub, 0, sizeof(struct dlm_message));
5211 ms_stub->m_flags = DLM_IFL_STUB_MS;
5212 ms_stub->m_type = DLM_MSG_UNLOCK_REPLY;
5213 ms_stub->m_result = stub_unlock_result;
5214 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5215 _receive_unlock_reply(lkb, ms_stub);
5219 case DLM_MSG_CANCEL:
5221 memset(ms_stub, 0, sizeof(struct dlm_message));
5222 ms_stub->m_flags = DLM_IFL_STUB_MS;
5223 ms_stub->m_type = DLM_MSG_CANCEL_REPLY;
5224 ms_stub->m_result = stub_cancel_result;
5225 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5226 _receive_cancel_reply(lkb, ms_stub);
5231 log_error(ls, "invalid lkb wait_type %d %d",
5232 lkb->lkb_wait_type, wait_type);
5236 mutex_unlock(&ls->ls_waiters_mutex);
5240 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5242 struct dlm_lkb *lkb;
5245 mutex_lock(&ls->ls_waiters_mutex);
5246 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
5247 if (lkb->lkb_flags & DLM_IFL_RESEND) {
5253 mutex_unlock(&ls->ls_waiters_mutex);
5260 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
5261 master or dir-node for r. Processing the lkb may result in it being placed
5264 /* We do this after normal locking has been enabled and any saved messages
5265 (in requestqueue) have been processed. We should be confident that at
5266 this point we won't get or process a reply to any of these waiting
5267 operations. But, new ops may be coming in on the rsbs/locks here from
5268 userspace or remotely. */
5270 /* there may have been an overlap unlock/cancel prior to recovery or after
5271 recovery. if before, the lkb may still have a pos wait_count; if after, the
5272 overlap flag would just have been set and nothing new sent. we can be
5273 confident here than any replies to either the initial op or overlap ops
5274 prior to recovery have been received. */
5276 int dlm_recover_waiters_post(struct dlm_ls *ls)
5278 struct dlm_lkb *lkb;
5280 int error = 0, mstype, err, oc, ou;
5283 if (dlm_locking_stopped(ls)) {
5284 log_debug(ls, "recover_waiters_post aborted");
5289 lkb = find_resend_waiter(ls);
5293 r = lkb->lkb_resource;
5297 mstype = lkb->lkb_wait_type;
5298 oc = is_overlap_cancel(lkb);
5299 ou = is_overlap_unlock(lkb);
5302 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5303 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5304 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5305 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5306 dlm_dir_nodeid(r), oc, ou);
5308 /* At this point we assume that we won't get a reply to any
5309 previous op or overlap op on this lock. First, do a big
5310 remove_from_waiters() for all previous ops. */
5312 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5313 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5314 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5315 lkb->lkb_wait_type = 0;
5316 /* drop all wait_count references we still
5317 * hold a reference for this iteration.
5319 while (lkb->lkb_wait_count) {
5320 lkb->lkb_wait_count--;
5323 mutex_lock(&ls->ls_waiters_mutex);
5324 list_del_init(&lkb->lkb_wait_reply);
5325 mutex_unlock(&ls->ls_waiters_mutex);
5328 /* do an unlock or cancel instead of resending */
5330 case DLM_MSG_LOOKUP:
5331 case DLM_MSG_REQUEST:
5332 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5334 unhold_lkb(lkb); /* undoes create_lkb() */
5336 case DLM_MSG_CONVERT:
5338 queue_cast(r, lkb, -DLM_ECANCEL);
5340 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5341 _unlock_lock(r, lkb);
5349 case DLM_MSG_LOOKUP:
5350 case DLM_MSG_REQUEST:
5351 _request_lock(r, lkb);
5353 confirm_master(r, 0);
5355 case DLM_MSG_CONVERT:
5356 _convert_lock(r, lkb);
5364 log_error(ls, "waiter %x msg %d r_nodeid %d "
5365 "dir_nodeid %d overlap %d %d",
5366 lkb->lkb_id, mstype, r->res_nodeid,
5367 dlm_dir_nodeid(r), oc, ou);
5377 static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5378 struct list_head *list)
5380 struct dlm_lkb *lkb, *safe;
5382 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5383 if (!is_master_copy(lkb))
5386 /* don't purge lkbs we've added in recover_master_copy for
5387 the current recovery seq */
5389 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5394 /* this put should free the lkb */
5395 if (!dlm_put_lkb(lkb))
5396 log_error(ls, "purged mstcpy lkb not released");
5400 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5402 struct dlm_ls *ls = r->res_ls;
5404 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5405 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5406 purge_mstcpy_list(ls, r, &r->res_waitqueue);
5409 static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5410 struct list_head *list,
5411 int nodeid_gone, unsigned int *count)
5413 struct dlm_lkb *lkb, *safe;
5415 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5416 if (!is_master_copy(lkb))
5419 if ((lkb->lkb_nodeid == nodeid_gone) ||
5420 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5422 /* tell recover_lvb to invalidate the lvb
5423 because a node holding EX/PW failed */
5424 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5425 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5426 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5431 /* this put should free the lkb */
5432 if (!dlm_put_lkb(lkb))
5433 log_error(ls, "purged dead lkb not released");
5435 rsb_set_flag(r, RSB_RECOVER_GRANT);
5442 /* Get rid of locks held by nodes that are gone. */
5444 void dlm_recover_purge(struct dlm_ls *ls)
5447 struct dlm_member *memb;
5448 int nodes_count = 0;
5449 int nodeid_gone = 0;
5450 unsigned int lkb_count = 0;
5452 /* cache one removed nodeid to optimize the common
5453 case of a single node removed */
5455 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5457 nodeid_gone = memb->nodeid;
5463 down_write(&ls->ls_root_sem);
5464 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5468 purge_dead_list(ls, r, &r->res_grantqueue,
5469 nodeid_gone, &lkb_count);
5470 purge_dead_list(ls, r, &r->res_convertqueue,
5471 nodeid_gone, &lkb_count);
5472 purge_dead_list(ls, r, &r->res_waitqueue,
5473 nodeid_gone, &lkb_count);
5479 up_write(&ls->ls_root_sem);
5482 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5483 lkb_count, nodes_count);
5486 static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5491 spin_lock(&ls->ls_rsbtbl[bucket].lock);
5492 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5493 r = rb_entry(n, struct dlm_rsb, res_hashnode);
5495 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5497 if (!is_master(r)) {
5498 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5502 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5505 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5510 * Attempt to grant locks on resources that we are the master of.
5511 * Locks may have become grantable during recovery because locks
5512 * from departed nodes have been purged (or not rebuilt), allowing
5513 * previously blocked locks to now be granted. The subset of rsb's
5514 * we are interested in are those with lkb's on either the convert or
5517 * Simplest would be to go through each master rsb and check for non-empty
5518 * convert or waiting queues, and attempt to grant on those rsbs.
5519 * Checking the queues requires lock_rsb, though, for which we'd need
5520 * to release the rsbtbl lock. This would make iterating through all
5521 * rsb's very inefficient. So, we rely on earlier recovery routines
5522 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5526 void dlm_recover_grant(struct dlm_ls *ls)
5530 unsigned int count = 0;
5531 unsigned int rsb_count = 0;
5532 unsigned int lkb_count = 0;
5535 r = find_grant_rsb(ls, bucket);
5537 if (bucket == ls->ls_rsbtbl_size - 1)
5545 /* the RECOVER_GRANT flag is checked in the grant path */
5546 grant_pending_locks(r, &count);
5547 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5549 confirm_master(r, 0);
5556 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5557 lkb_count, rsb_count);
5560 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5563 struct dlm_lkb *lkb;
5565 list_for_each_entry(lkb, head, lkb_statequeue) {
5566 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5572 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5575 struct dlm_lkb *lkb;
5577 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5580 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5583 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5589 /* needs at least dlm_rcom + rcom_lock */
5590 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5591 struct dlm_rsb *r, struct dlm_rcom *rc)
5593 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5595 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
5596 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5597 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5598 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5599 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5600 lkb->lkb_flags |= DLM_IFL_MSTCPY;
5601 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5602 lkb->lkb_rqmode = rl->rl_rqmode;
5603 lkb->lkb_grmode = rl->rl_grmode;
5604 /* don't set lkb_status because add_lkb wants to itself */
5606 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5607 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5609 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5610 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
5611 sizeof(struct rcom_lock);
5612 if (lvblen > ls->ls_lvblen)
5614 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5615 if (!lkb->lkb_lvbptr)
5617 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5620 /* Conversions between PR and CW (middle modes) need special handling.
5621 The real granted mode of these converting locks cannot be determined
5622 until all locks have been rebuilt on the rsb (recover_conversion) */
5624 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5625 middle_conversion(lkb)) {
5626 rl->rl_status = DLM_LKSTS_CONVERT;
5627 lkb->lkb_grmode = DLM_LOCK_IV;
5628 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5634 /* This lkb may have been recovered in a previous aborted recovery so we need
5635 to check if the rsb already has an lkb with the given remote nodeid/lkid.
5636 If so we just send back a standard reply. If not, we create a new lkb with
5637 the given values and send back our lkid. We send back our lkid by sending
5638 back the rcom_lock struct we got but with the remid field filled in. */
5640 /* needs at least dlm_rcom + rcom_lock */
5641 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5643 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5645 struct dlm_lkb *lkb;
5647 int from_nodeid = rc->rc_header.h_nodeid;
5650 if (rl->rl_parent_lkid) {
5651 error = -EOPNOTSUPP;
5655 remid = le32_to_cpu(rl->rl_lkid);
5657 /* In general we expect the rsb returned to be R_MASTER, but we don't
5658 have to require it. Recovery of masters on one node can overlap
5659 recovery of locks on another node, so one node can send us MSTCPY
5660 locks before we've made ourselves master of this rsb. We can still
5661 add new MSTCPY locks that we receive here without any harm; when
5662 we make ourselves master, dlm_recover_masters() won't touch the
5663 MSTCPY locks we've received early. */
5665 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5666 from_nodeid, R_RECEIVE_RECOVER, &r);
5672 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5673 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5674 from_nodeid, remid);
5679 lkb = search_remid(r, from_nodeid, remid);
5685 error = create_lkb(ls, &lkb);
5689 error = receive_rcom_lock_args(ls, lkb, r, rc);
5696 add_lkb(r, lkb, rl->rl_status);
5698 ls->ls_recover_locks_in++;
5700 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5701 rsb_set_flag(r, RSB_RECOVER_GRANT);
5704 /* this is the new value returned to the lock holder for
5705 saving in its process-copy lkb */
5706 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5708 lkb->lkb_recover_seq = ls->ls_recover_seq;
5714 if (error && error != -EEXIST)
5715 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5716 from_nodeid, remid, error);
5717 rl->rl_result = cpu_to_le32(error);
5721 /* needs at least dlm_rcom + rcom_lock */
5722 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5724 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5726 struct dlm_lkb *lkb;
5727 uint32_t lkid, remid;
5730 lkid = le32_to_cpu(rl->rl_lkid);
5731 remid = le32_to_cpu(rl->rl_remid);
5732 result = le32_to_cpu(rl->rl_result);
5734 error = find_lkb(ls, lkid, &lkb);
5736 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5737 lkid, rc->rc_header.h_nodeid, remid, result);
5741 r = lkb->lkb_resource;
5745 if (!is_process_copy(lkb)) {
5746 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5747 lkid, rc->rc_header.h_nodeid, remid, result);
5757 /* There's a chance the new master received our lock before
5758 dlm_recover_master_reply(), this wouldn't happen if we did
5759 a barrier between recover_masters and recover_locks. */
5761 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5762 lkid, rc->rc_header.h_nodeid, remid, result);
5764 dlm_send_rcom_lock(r, lkb);
5768 lkb->lkb_remid = remid;
5771 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5772 lkid, rc->rc_header.h_nodeid, remid, result);
5775 /* an ack for dlm_recover_locks() which waits for replies from
5776 all the locks it sends to new masters */
5777 dlm_recovered_lock(r);
5786 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5787 int mode, uint32_t flags, void *name, unsigned int namelen,
5788 unsigned long timeout_cs)
5790 struct dlm_lkb *lkb;
5791 struct dlm_args args;
5794 dlm_lock_recovery(ls);
5796 error = create_lkb(ls, &lkb);
5802 if (flags & DLM_LKF_VALBLK) {
5803 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5804 if (!ua->lksb.sb_lvbptr) {
5811 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5812 fake_astfn, ua, fake_bastfn, &args);
5814 kfree(ua->lksb.sb_lvbptr);
5815 ua->lksb.sb_lvbptr = NULL;
5821 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5822 When DLM_IFL_USER is set, the dlm knows that this is a userspace
5823 lock and that lkb_astparam is the dlm_user_args structure. */
5824 lkb->lkb_flags |= DLM_IFL_USER;
5825 error = request_lock(ls, lkb, name, namelen, &args);
5841 /* add this new lkb to the per-process list of locks */
5842 spin_lock(&ua->proc->locks_spin);
5844 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5845 spin_unlock(&ua->proc->locks_spin);
5847 dlm_unlock_recovery(ls);
5851 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5852 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5853 unsigned long timeout_cs)
5855 struct dlm_lkb *lkb;
5856 struct dlm_args args;
5857 struct dlm_user_args *ua;
5860 dlm_lock_recovery(ls);
5862 error = find_lkb(ls, lkid, &lkb);
5866 /* user can change the params on its lock when it converts it, or
5867 add an lvb that didn't exist before */
5871 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5872 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5873 if (!ua->lksb.sb_lvbptr) {
5878 if (lvb_in && ua->lksb.sb_lvbptr)
5879 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5881 ua->xid = ua_tmp->xid;
5882 ua->castparam = ua_tmp->castparam;
5883 ua->castaddr = ua_tmp->castaddr;
5884 ua->bastparam = ua_tmp->bastparam;
5885 ua->bastaddr = ua_tmp->bastaddr;
5886 ua->user_lksb = ua_tmp->user_lksb;
5888 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5889 fake_astfn, ua, fake_bastfn, &args);
5893 error = convert_lock(ls, lkb, &args);
5895 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5900 dlm_unlock_recovery(ls);
5906 * The caller asks for an orphan lock on a given resource with a given mode.
5907 * If a matching lock exists, it's moved to the owner's list of locks and
5908 * the lkid is returned.
5911 int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5912 int mode, uint32_t flags, void *name, unsigned int namelen,
5913 unsigned long timeout_cs, uint32_t *lkid)
5915 struct dlm_lkb *lkb;
5916 struct dlm_user_args *ua;
5917 int found_other_mode = 0;
5921 mutex_lock(&ls->ls_orphans_mutex);
5922 list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) {
5923 if (lkb->lkb_resource->res_length != namelen)
5925 if (memcmp(lkb->lkb_resource->res_name, name, namelen))
5927 if (lkb->lkb_grmode != mode) {
5928 found_other_mode = 1;
5933 list_del_init(&lkb->lkb_ownqueue);
5934 lkb->lkb_flags &= ~DLM_IFL_ORPHAN;
5935 *lkid = lkb->lkb_id;
5938 mutex_unlock(&ls->ls_orphans_mutex);
5940 if (!found && found_other_mode) {
5950 lkb->lkb_exflags = flags;
5951 lkb->lkb_ownpid = (int) current->pid;
5955 ua->proc = ua_tmp->proc;
5956 ua->xid = ua_tmp->xid;
5957 ua->castparam = ua_tmp->castparam;
5958 ua->castaddr = ua_tmp->castaddr;
5959 ua->bastparam = ua_tmp->bastparam;
5960 ua->bastaddr = ua_tmp->bastaddr;
5961 ua->user_lksb = ua_tmp->user_lksb;
5964 * The lkb reference from the ls_orphans list was not
5965 * removed above, and is now considered the reference
5966 * for the proc locks list.
5969 spin_lock(&ua->proc->locks_spin);
5970 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5971 spin_unlock(&ua->proc->locks_spin);
5977 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5978 uint32_t flags, uint32_t lkid, char *lvb_in)
5980 struct dlm_lkb *lkb;
5981 struct dlm_args args;
5982 struct dlm_user_args *ua;
5985 dlm_lock_recovery(ls);
5987 error = find_lkb(ls, lkid, &lkb);
5993 if (lvb_in && ua->lksb.sb_lvbptr)
5994 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5995 if (ua_tmp->castparam)
5996 ua->castparam = ua_tmp->castparam;
5997 ua->user_lksb = ua_tmp->user_lksb;
5999 error = set_unlock_args(flags, ua, &args);
6003 error = unlock_lock(ls, lkb, &args);
6005 if (error == -DLM_EUNLOCK)
6007 /* from validate_unlock_args() */
6008 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
6013 spin_lock(&ua->proc->locks_spin);
6014 /* dlm_user_add_cb() may have already taken lkb off the proc list */
6015 if (!list_empty(&lkb->lkb_ownqueue))
6016 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
6017 spin_unlock(&ua->proc->locks_spin);
6021 dlm_unlock_recovery(ls);
6026 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6027 uint32_t flags, uint32_t lkid)
6029 struct dlm_lkb *lkb;
6030 struct dlm_args args;
6031 struct dlm_user_args *ua;
6034 dlm_lock_recovery(ls);
6036 error = find_lkb(ls, lkid, &lkb);
6041 if (ua_tmp->castparam)
6042 ua->castparam = ua_tmp->castparam;
6043 ua->user_lksb = ua_tmp->user_lksb;
6045 error = set_unlock_args(flags, ua, &args);
6049 error = cancel_lock(ls, lkb, &args);
6051 if (error == -DLM_ECANCEL)
6053 /* from validate_unlock_args() */
6054 if (error == -EBUSY)
6059 dlm_unlock_recovery(ls);
6064 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6066 struct dlm_lkb *lkb;
6067 struct dlm_args args;
6068 struct dlm_user_args *ua;
6072 dlm_lock_recovery(ls);
6074 error = find_lkb(ls, lkid, &lkb);
6080 error = set_unlock_args(flags, ua, &args);
6084 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6086 r = lkb->lkb_resource;
6090 error = validate_unlock_args(lkb, &args);
6093 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6095 error = _cancel_lock(r, lkb);
6100 if (error == -DLM_ECANCEL)
6102 /* from validate_unlock_args() */
6103 if (error == -EBUSY)
6108 dlm_unlock_recovery(ls);
6112 /* lkb's that are removed from the waiters list by revert are just left on the
6113 orphans list with the granted orphan locks, to be freed by purge */
6115 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6117 struct dlm_args args;
6120 hold_lkb(lkb); /* reference for the ls_orphans list */
6121 mutex_lock(&ls->ls_orphans_mutex);
6122 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6123 mutex_unlock(&ls->ls_orphans_mutex);
6125 set_unlock_args(0, lkb->lkb_ua, &args);
6127 error = cancel_lock(ls, lkb, &args);
6128 if (error == -DLM_ECANCEL)
6133 /* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6134 granted. Regardless of what rsb queue the lock is on, it's removed and
6135 freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6136 if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6138 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6140 struct dlm_args args;
6143 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6144 lkb->lkb_ua, &args);
6146 error = unlock_lock(ls, lkb, &args);
6147 if (error == -DLM_EUNLOCK)
6152 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6153 (which does lock_rsb) due to deadlock with receiving a message that does
6154 lock_rsb followed by dlm_user_add_cb() */
6156 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6157 struct dlm_user_proc *proc)
6159 struct dlm_lkb *lkb = NULL;
6161 mutex_lock(&ls->ls_clear_proc_locks);
6162 if (list_empty(&proc->locks))
6165 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6166 list_del_init(&lkb->lkb_ownqueue);
6168 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6169 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6171 lkb->lkb_flags |= DLM_IFL_DEAD;
6173 mutex_unlock(&ls->ls_clear_proc_locks);
6177 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6178 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6179 which we clear here. */
6181 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
6182 list, and no more device_writes should add lkb's to proc->locks list; so we
6183 shouldn't need to take asts_spin or locks_spin here. this assumes that
6184 device reads/writes/closes are serialized -- FIXME: we may need to serialize
6187 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6189 struct dlm_lkb *lkb, *safe;
6191 dlm_lock_recovery(ls);
6194 lkb = del_proc_lock(ls, proc);
6198 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6199 orphan_proc_lock(ls, lkb);
6201 unlock_proc_lock(ls, lkb);
6203 /* this removes the reference for the proc->locks list
6204 added by dlm_user_request, it may result in the lkb
6210 mutex_lock(&ls->ls_clear_proc_locks);
6212 /* in-progress unlocks */
6213 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6214 list_del_init(&lkb->lkb_ownqueue);
6215 lkb->lkb_flags |= DLM_IFL_DEAD;
6219 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6220 memset(&lkb->lkb_callbacks, 0,
6221 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6222 list_del_init(&lkb->lkb_cb_list);
6226 mutex_unlock(&ls->ls_clear_proc_locks);
6227 dlm_unlock_recovery(ls);
6230 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6232 struct dlm_lkb *lkb, *safe;
6236 spin_lock(&proc->locks_spin);
6237 if (!list_empty(&proc->locks)) {
6238 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6240 list_del_init(&lkb->lkb_ownqueue);
6242 spin_unlock(&proc->locks_spin);
6247 lkb->lkb_flags |= DLM_IFL_DEAD;
6248 unlock_proc_lock(ls, lkb);
6249 dlm_put_lkb(lkb); /* ref from proc->locks list */
6252 spin_lock(&proc->locks_spin);
6253 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6254 list_del_init(&lkb->lkb_ownqueue);
6255 lkb->lkb_flags |= DLM_IFL_DEAD;
6258 spin_unlock(&proc->locks_spin);
6260 spin_lock(&proc->asts_spin);
6261 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6262 memset(&lkb->lkb_callbacks, 0,
6263 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6264 list_del_init(&lkb->lkb_cb_list);
6267 spin_unlock(&proc->asts_spin);
6270 /* pid of 0 means purge all orphans */
6272 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6274 struct dlm_lkb *lkb, *safe;
6276 mutex_lock(&ls->ls_orphans_mutex);
6277 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6278 if (pid && lkb->lkb_ownpid != pid)
6280 unlock_proc_lock(ls, lkb);
6281 list_del_init(&lkb->lkb_ownqueue);
6284 mutex_unlock(&ls->ls_orphans_mutex);
6287 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6289 struct dlm_message *ms;
6290 struct dlm_mhandle *mh;
6293 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6294 DLM_MSG_PURGE, &ms, &mh);
6297 ms->m_nodeid = nodeid;
6300 return send_message(mh, ms);
6303 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6304 int nodeid, int pid)
6308 if (nodeid && (nodeid != dlm_our_nodeid())) {
6309 error = send_purge(ls, nodeid, pid);
6311 dlm_lock_recovery(ls);
6312 if (pid == current->pid)
6313 purge_proc_locks(ls, proc);
6315 do_purge(ls, nodeid, pid);
6316 dlm_unlock_recovery(ls);