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--;
1560 /* N.B. type of reply may not always correspond to type of original
1561 msg due to lookup->request optimization, verify others? */
1563 if (lkb->lkb_wait_type) {
1564 lkb->lkb_wait_type = 0;
1568 log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1569 lkb->lkb_id, ms ? ms->m_header.h_nodeid : 0, lkb->lkb_remid,
1570 mstype, lkb->lkb_flags);
1574 /* the force-unlock/cancel has completed and we haven't recvd a reply
1575 to the op that was in progress prior to the unlock/cancel; we
1576 give up on any reply to the earlier op. FIXME: not sure when/how
1577 this would happen */
1579 if (overlap_done && lkb->lkb_wait_type) {
1580 log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1581 lkb->lkb_id, mstype, lkb->lkb_wait_type);
1582 lkb->lkb_wait_count--;
1583 lkb->lkb_wait_type = 0;
1586 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1588 lkb->lkb_flags &= ~DLM_IFL_RESEND;
1589 lkb->lkb_wait_count--;
1590 if (!lkb->lkb_wait_count)
1591 list_del_init(&lkb->lkb_wait_reply);
1596 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1598 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1601 mutex_lock(&ls->ls_waiters_mutex);
1602 error = _remove_from_waiters(lkb, mstype, NULL);
1603 mutex_unlock(&ls->ls_waiters_mutex);
1607 /* Handles situations where we might be processing a "fake" or "stub" reply in
1608 which we can't try to take waiters_mutex again. */
1610 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1612 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1615 if (ms->m_flags != DLM_IFL_STUB_MS)
1616 mutex_lock(&ls->ls_waiters_mutex);
1617 error = _remove_from_waiters(lkb, ms->m_type, ms);
1618 if (ms->m_flags != DLM_IFL_STUB_MS)
1619 mutex_unlock(&ls->ls_waiters_mutex);
1623 /* If there's an rsb for the same resource being removed, ensure
1624 that the remove message is sent before the new lookup message.
1625 It should be rare to need a delay here, but if not, then it may
1626 be worthwhile to add a proper wait mechanism rather than a delay. */
1628 static void wait_pending_remove(struct dlm_rsb *r)
1630 struct dlm_ls *ls = r->res_ls;
1632 spin_lock(&ls->ls_remove_spin);
1633 if (ls->ls_remove_len &&
1634 !rsb_cmp(r, ls->ls_remove_name, ls->ls_remove_len)) {
1635 log_debug(ls, "delay lookup for remove dir %d %s",
1636 r->res_dir_nodeid, r->res_name);
1637 spin_unlock(&ls->ls_remove_spin);
1641 spin_unlock(&ls->ls_remove_spin);
1645 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1646 * read by other threads in wait_pending_remove. ls_remove_names
1647 * and ls_remove_lens are only used by the scan thread, so they do
1648 * not need protection.
1651 static void shrink_bucket(struct dlm_ls *ls, int b)
1653 struct rb_node *n, *next;
1656 int our_nodeid = dlm_our_nodeid();
1657 int remote_count = 0;
1658 int need_shrink = 0;
1661 memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1663 spin_lock(&ls->ls_rsbtbl[b].lock);
1665 if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1666 spin_unlock(&ls->ls_rsbtbl[b].lock);
1670 for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1672 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1674 /* If we're the directory record for this rsb, and
1675 we're not the master of it, then we need to wait
1676 for the master node to send us a dir remove for
1677 before removing the dir record. */
1679 if (!dlm_no_directory(ls) &&
1680 (r->res_master_nodeid != our_nodeid) &&
1681 (dlm_dir_nodeid(r) == our_nodeid)) {
1687 if (!time_after_eq(jiffies, r->res_toss_time +
1688 dlm_config.ci_toss_secs * HZ)) {
1692 if (!dlm_no_directory(ls) &&
1693 (r->res_master_nodeid == our_nodeid) &&
1694 (dlm_dir_nodeid(r) != our_nodeid)) {
1696 /* We're the master of this rsb but we're not
1697 the directory record, so we need to tell the
1698 dir node to remove the dir record. */
1700 ls->ls_remove_lens[remote_count] = r->res_length;
1701 memcpy(ls->ls_remove_names[remote_count], r->res_name,
1702 DLM_RESNAME_MAXLEN);
1705 if (remote_count >= DLM_REMOVE_NAMES_MAX)
1710 if (!kref_put(&r->res_ref, kill_rsb)) {
1711 log_error(ls, "tossed rsb in use %s", r->res_name);
1715 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1720 ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1722 ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1723 spin_unlock(&ls->ls_rsbtbl[b].lock);
1726 * While searching for rsb's to free, we found some that require
1727 * remote removal. We leave them in place and find them again here
1728 * so there is a very small gap between removing them from the toss
1729 * list and sending the removal. Keeping this gap small is
1730 * important to keep us (the master node) from being out of sync
1731 * with the remote dir node for very long.
1733 * From the time the rsb is removed from toss until just after
1734 * send_remove, the rsb name is saved in ls_remove_name. A new
1735 * lookup checks this to ensure that a new lookup message for the
1736 * same resource name is not sent just before the remove message.
1739 for (i = 0; i < remote_count; i++) {
1740 name = ls->ls_remove_names[i];
1741 len = ls->ls_remove_lens[i];
1743 spin_lock(&ls->ls_rsbtbl[b].lock);
1744 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1746 spin_unlock(&ls->ls_rsbtbl[b].lock);
1747 log_debug(ls, "remove_name not toss %s", name);
1751 if (r->res_master_nodeid != our_nodeid) {
1752 spin_unlock(&ls->ls_rsbtbl[b].lock);
1753 log_debug(ls, "remove_name master %d dir %d our %d %s",
1754 r->res_master_nodeid, r->res_dir_nodeid,
1759 if (r->res_dir_nodeid == our_nodeid) {
1760 /* should never happen */
1761 spin_unlock(&ls->ls_rsbtbl[b].lock);
1762 log_error(ls, "remove_name dir %d master %d our %d %s",
1763 r->res_dir_nodeid, r->res_master_nodeid,
1768 if (!time_after_eq(jiffies, r->res_toss_time +
1769 dlm_config.ci_toss_secs * HZ)) {
1770 spin_unlock(&ls->ls_rsbtbl[b].lock);
1771 log_debug(ls, "remove_name toss_time %lu now %lu %s",
1772 r->res_toss_time, jiffies, name);
1776 if (!kref_put(&r->res_ref, kill_rsb)) {
1777 spin_unlock(&ls->ls_rsbtbl[b].lock);
1778 log_error(ls, "remove_name in use %s", name);
1782 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1784 /* block lookup of same name until we've sent remove */
1785 spin_lock(&ls->ls_remove_spin);
1786 ls->ls_remove_len = len;
1787 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1788 spin_unlock(&ls->ls_remove_spin);
1789 spin_unlock(&ls->ls_rsbtbl[b].lock);
1793 /* allow lookup of name again */
1794 spin_lock(&ls->ls_remove_spin);
1795 ls->ls_remove_len = 0;
1796 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1797 spin_unlock(&ls->ls_remove_spin);
1803 void dlm_scan_rsbs(struct dlm_ls *ls)
1807 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1808 shrink_bucket(ls, i);
1809 if (dlm_locking_stopped(ls))
1815 static void add_timeout(struct dlm_lkb *lkb)
1817 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1819 if (is_master_copy(lkb))
1822 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1823 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1824 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1827 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1832 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1833 mutex_lock(&ls->ls_timeout_mutex);
1835 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1836 mutex_unlock(&ls->ls_timeout_mutex);
1839 static void del_timeout(struct dlm_lkb *lkb)
1841 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1843 mutex_lock(&ls->ls_timeout_mutex);
1844 if (!list_empty(&lkb->lkb_time_list)) {
1845 list_del_init(&lkb->lkb_time_list);
1848 mutex_unlock(&ls->ls_timeout_mutex);
1851 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1852 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1853 and then lock rsb because of lock ordering in add_timeout. We may need
1854 to specify some special timeout-related bits in the lkb that are just to
1855 be accessed under the timeout_mutex. */
1857 void dlm_scan_timeout(struct dlm_ls *ls)
1860 struct dlm_lkb *lkb;
1861 int do_cancel, do_warn;
1865 if (dlm_locking_stopped(ls))
1870 mutex_lock(&ls->ls_timeout_mutex);
1871 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1873 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1874 lkb->lkb_timestamp));
1876 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1877 wait_us >= (lkb->lkb_timeout_cs * 10000))
1880 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1881 wait_us >= dlm_config.ci_timewarn_cs * 10000)
1884 if (!do_cancel && !do_warn)
1889 mutex_unlock(&ls->ls_timeout_mutex);
1891 if (!do_cancel && !do_warn)
1894 r = lkb->lkb_resource;
1899 /* clear flag so we only warn once */
1900 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1901 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1903 dlm_timeout_warn(lkb);
1907 log_debug(ls, "timeout cancel %x node %d %s",
1908 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1909 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1910 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1912 _cancel_lock(r, lkb);
1921 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1922 dlm_recoverd before checking/setting ls_recover_begin. */
1924 void dlm_adjust_timeouts(struct dlm_ls *ls)
1926 struct dlm_lkb *lkb;
1927 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1929 ls->ls_recover_begin = 0;
1930 mutex_lock(&ls->ls_timeout_mutex);
1931 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1932 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1933 mutex_unlock(&ls->ls_timeout_mutex);
1935 if (!dlm_config.ci_waitwarn_us)
1938 mutex_lock(&ls->ls_waiters_mutex);
1939 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1940 if (ktime_to_us(lkb->lkb_wait_time))
1941 lkb->lkb_wait_time = ktime_get();
1943 mutex_unlock(&ls->ls_waiters_mutex);
1946 /* lkb is master or local copy */
1948 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1950 int b, len = r->res_ls->ls_lvblen;
1952 /* b=1 lvb returned to caller
1953 b=0 lvb written to rsb or invalidated
1956 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1959 if (!lkb->lkb_lvbptr)
1962 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1968 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1969 lkb->lkb_lvbseq = r->res_lvbseq;
1971 } else if (b == 0) {
1972 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1973 rsb_set_flag(r, RSB_VALNOTVALID);
1977 if (!lkb->lkb_lvbptr)
1980 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1984 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1989 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1991 lkb->lkb_lvbseq = r->res_lvbseq;
1992 rsb_clear_flag(r, RSB_VALNOTVALID);
1995 if (rsb_flag(r, RSB_VALNOTVALID))
1996 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1999 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2001 if (lkb->lkb_grmode < DLM_LOCK_PW)
2004 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
2005 rsb_set_flag(r, RSB_VALNOTVALID);
2009 if (!lkb->lkb_lvbptr)
2012 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2016 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
2021 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2023 rsb_clear_flag(r, RSB_VALNOTVALID);
2026 /* lkb is process copy (pc) */
2028 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2029 struct dlm_message *ms)
2033 if (!lkb->lkb_lvbptr)
2036 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2039 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
2041 int len = receive_extralen(ms);
2042 if (len > r->res_ls->ls_lvblen)
2043 len = r->res_ls->ls_lvblen;
2044 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2045 lkb->lkb_lvbseq = ms->m_lvbseq;
2049 /* Manipulate lkb's on rsb's convert/granted/waiting queues
2050 remove_lock -- used for unlock, removes lkb from granted
2051 revert_lock -- used for cancel, moves lkb from convert to granted
2052 grant_lock -- used for request and convert, adds lkb to granted or
2053 moves lkb from convert or waiting to granted
2055 Each of these is used for master or local copy lkb's. There is
2056 also a _pc() variation used to make the corresponding change on
2057 a process copy (pc) lkb. */
2059 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2062 lkb->lkb_grmode = DLM_LOCK_IV;
2063 /* this unhold undoes the original ref from create_lkb()
2064 so this leads to the lkb being freed */
2068 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2070 set_lvb_unlock(r, lkb);
2071 _remove_lock(r, lkb);
2074 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2076 _remove_lock(r, lkb);
2079 /* returns: 0 did nothing
2080 1 moved lock to granted
2083 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2087 lkb->lkb_rqmode = DLM_LOCK_IV;
2089 switch (lkb->lkb_status) {
2090 case DLM_LKSTS_GRANTED:
2092 case DLM_LKSTS_CONVERT:
2093 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2096 case DLM_LKSTS_WAITING:
2098 lkb->lkb_grmode = DLM_LOCK_IV;
2099 /* this unhold undoes the original ref from create_lkb()
2100 so this leads to the lkb being freed */
2105 log_print("invalid status for revert %d", lkb->lkb_status);
2110 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2112 return revert_lock(r, lkb);
2115 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2117 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2118 lkb->lkb_grmode = lkb->lkb_rqmode;
2119 if (lkb->lkb_status)
2120 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2122 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2125 lkb->lkb_rqmode = DLM_LOCK_IV;
2126 lkb->lkb_highbast = 0;
2129 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2131 set_lvb_lock(r, lkb);
2132 _grant_lock(r, lkb);
2135 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2136 struct dlm_message *ms)
2138 set_lvb_lock_pc(r, lkb, ms);
2139 _grant_lock(r, lkb);
2142 /* called by grant_pending_locks() which means an async grant message must
2143 be sent to the requesting node in addition to granting the lock if the
2144 lkb belongs to a remote node. */
2146 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2149 if (is_master_copy(lkb))
2152 queue_cast(r, lkb, 0);
2155 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2156 change the granted/requested modes. We're munging things accordingly in
2158 CONVDEADLK: our grmode may have been forced down to NL to resolve a
2160 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2161 compatible with other granted locks */
2163 static void munge_demoted(struct dlm_lkb *lkb)
2165 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2166 log_print("munge_demoted %x invalid modes gr %d rq %d",
2167 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2171 lkb->lkb_grmode = DLM_LOCK_NL;
2174 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2176 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
2177 ms->m_type != DLM_MSG_GRANT) {
2178 log_print("munge_altmode %x invalid reply type %d",
2179 lkb->lkb_id, ms->m_type);
2183 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2184 lkb->lkb_rqmode = DLM_LOCK_PR;
2185 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2186 lkb->lkb_rqmode = DLM_LOCK_CW;
2188 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2193 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2195 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2197 if (lkb->lkb_id == first->lkb_id)
2203 /* Check if the given lkb conflicts with another lkb on the queue. */
2205 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2207 struct dlm_lkb *this;
2209 list_for_each_entry(this, head, lkb_statequeue) {
2212 if (!modes_compat(this, lkb))
2219 * "A conversion deadlock arises with a pair of lock requests in the converting
2220 * queue for one resource. The granted mode of each lock blocks the requested
2221 * mode of the other lock."
2223 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2224 * convert queue from being granted, then deadlk/demote lkb.
2227 * Granted Queue: empty
2228 * Convert Queue: NL->EX (first lock)
2229 * PR->EX (second lock)
2231 * The first lock can't be granted because of the granted mode of the second
2232 * lock and the second lock can't be granted because it's not first in the
2233 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2234 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2235 * flag set and return DEMOTED in the lksb flags.
2237 * Originally, this function detected conv-deadlk in a more limited scope:
2238 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2239 * - if lkb1 was the first entry in the queue (not just earlier), and was
2240 * blocked by the granted mode of lkb2, and there was nothing on the
2241 * granted queue preventing lkb1 from being granted immediately, i.e.
2242 * lkb2 was the only thing preventing lkb1 from being granted.
2244 * That second condition meant we'd only say there was conv-deadlk if
2245 * resolving it (by demotion) would lead to the first lock on the convert
2246 * queue being granted right away. It allowed conversion deadlocks to exist
2247 * between locks on the convert queue while they couldn't be granted anyway.
2249 * Now, we detect and take action on conversion deadlocks immediately when
2250 * they're created, even if they may not be immediately consequential. If
2251 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2252 * mode that would prevent lkb1's conversion from being granted, we do a
2253 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2254 * I think this means that the lkb_is_ahead condition below should always
2255 * be zero, i.e. there will never be conv-deadlk between two locks that are
2256 * both already on the convert queue.
2259 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2261 struct dlm_lkb *lkb1;
2262 int lkb_is_ahead = 0;
2264 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2270 if (!lkb_is_ahead) {
2271 if (!modes_compat(lkb2, lkb1))
2274 if (!modes_compat(lkb2, lkb1) &&
2275 !modes_compat(lkb1, lkb2))
2283 * Return 1 if the lock can be granted, 0 otherwise.
2284 * Also detect and resolve conversion deadlocks.
2286 * lkb is the lock to be granted
2288 * now is 1 if the function is being called in the context of the
2289 * immediate request, it is 0 if called later, after the lock has been
2292 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2295 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2298 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2301 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2304 * 6-10: Version 5.4 introduced an option to address the phenomenon of
2305 * a new request for a NL mode lock being blocked.
2307 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2308 * request, then it would be granted. In essence, the use of this flag
2309 * tells the Lock Manager to expedite theis request by not considering
2310 * what may be in the CONVERTING or WAITING queues... As of this
2311 * writing, the EXPEDITE flag can be used only with new requests for NL
2312 * mode locks. This flag is not valid for conversion requests.
2314 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
2315 * conversion or used with a non-NL requested mode. We also know an
2316 * EXPEDITE request is always granted immediately, so now must always
2317 * be 1. The full condition to grant an expedite request: (now &&
2318 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2319 * therefore be shortened to just checking the flag.
2322 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2326 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2327 * added to the remaining conditions.
2330 if (queue_conflict(&r->res_grantqueue, lkb))
2334 * 6-3: By default, a conversion request is immediately granted if the
2335 * requested mode is compatible with the modes of all other granted
2339 if (queue_conflict(&r->res_convertqueue, lkb))
2343 * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2344 * locks for a recovered rsb, on which lkb's have been rebuilt.
2345 * The lkb's may have been rebuilt on the queues in a different
2346 * order than they were in on the previous master. So, granting
2347 * queued conversions in order after recovery doesn't make sense
2348 * since the order hasn't been preserved anyway. The new order
2349 * could also have created a new "in place" conversion deadlock.
2350 * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2351 * After recovery, there would be no granted locks, and possibly
2352 * NL->EX, PR->EX, an in-place conversion deadlock.) So, after
2353 * recovery, grant conversions without considering order.
2356 if (conv && recover)
2360 * 6-5: But the default algorithm for deciding whether to grant or
2361 * queue conversion requests does not by itself guarantee that such
2362 * requests are serviced on a "first come first serve" basis. This, in
2363 * turn, can lead to a phenomenon known as "indefinate postponement".
2365 * 6-7: This issue is dealt with by using the optional QUECVT flag with
2366 * the system service employed to request a lock conversion. This flag
2367 * forces certain conversion requests to be queued, even if they are
2368 * compatible with the granted modes of other locks on the same
2369 * resource. Thus, the use of this flag results in conversion requests
2370 * being ordered on a "first come first servce" basis.
2372 * DCT: This condition is all about new conversions being able to occur
2373 * "in place" while the lock remains on the granted queue (assuming
2374 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
2375 * doesn't _have_ to go onto the convert queue where it's processed in
2376 * order. The "now" variable is necessary to distinguish converts
2377 * being received and processed for the first time now, because once a
2378 * convert is moved to the conversion queue the condition below applies
2379 * requiring fifo granting.
2382 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2386 * Even if the convert is compat with all granted locks,
2387 * QUECVT forces it behind other locks on the convert queue.
2390 if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2391 if (list_empty(&r->res_convertqueue))
2398 * The NOORDER flag is set to avoid the standard vms rules on grant
2402 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2406 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2407 * granted until all other conversion requests ahead of it are granted
2411 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2415 * 6-4: By default, a new request is immediately granted only if all
2416 * three of the following conditions are satisfied when the request is
2418 * - The queue of ungranted conversion requests for the resource is
2420 * - The queue of ungranted new requests for the resource is empty.
2421 * - The mode of the new request is compatible with the most
2422 * restrictive mode of all granted locks on the resource.
2425 if (now && !conv && list_empty(&r->res_convertqueue) &&
2426 list_empty(&r->res_waitqueue))
2430 * 6-4: Once a lock request is in the queue of ungranted new requests,
2431 * it cannot be granted until the queue of ungranted conversion
2432 * requests is empty, all ungranted new requests ahead of it are
2433 * granted and/or canceled, and it is compatible with the granted mode
2434 * of the most restrictive lock granted on the resource.
2437 if (!now && !conv && list_empty(&r->res_convertqueue) &&
2438 first_in_list(lkb, &r->res_waitqueue))
2444 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2445 int recover, int *err)
2448 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2449 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2454 rv = _can_be_granted(r, lkb, now, recover);
2459 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2460 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2461 * cancels one of the locks.
2464 if (is_convert && can_be_queued(lkb) &&
2465 conversion_deadlock_detect(r, lkb)) {
2466 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2467 lkb->lkb_grmode = DLM_LOCK_NL;
2468 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2469 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
2473 log_print("can_be_granted deadlock %x now %d",
2482 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2483 * to grant a request in a mode other than the normal rqmode. It's a
2484 * simple way to provide a big optimization to applications that can
2488 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2490 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2494 lkb->lkb_rqmode = alt;
2495 rv = _can_be_granted(r, lkb, now, 0);
2497 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2499 lkb->lkb_rqmode = rqmode;
2505 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
2506 for locks pending on the convert list. Once verified (watch for these
2507 log_prints), we should be able to just call _can_be_granted() and not
2508 bother with the demote/deadlk cases here (and there's no easy way to deal
2509 with a deadlk here, we'd have to generate something like grant_lock with
2510 the deadlk error.) */
2512 /* Returns the highest requested mode of all blocked conversions; sets
2513 cw if there's a blocked conversion to DLM_LOCK_CW. */
2515 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2516 unsigned int *count)
2518 struct dlm_lkb *lkb, *s;
2519 int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2520 int hi, demoted, quit, grant_restart, demote_restart;
2529 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2530 demoted = is_demoted(lkb);
2533 if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2534 grant_lock_pending(r, lkb);
2541 if (!demoted && is_demoted(lkb)) {
2542 log_print("WARN: pending demoted %x node %d %s",
2543 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2549 log_print("WARN: pending deadlock %x node %d %s",
2550 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2555 hi = max_t(int, lkb->lkb_rqmode, hi);
2557 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2563 if (demote_restart && !quit) {
2568 return max_t(int, high, hi);
2571 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2572 unsigned int *count)
2574 struct dlm_lkb *lkb, *s;
2576 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2577 if (can_be_granted(r, lkb, 0, 0, NULL)) {
2578 grant_lock_pending(r, lkb);
2582 high = max_t(int, lkb->lkb_rqmode, high);
2583 if (lkb->lkb_rqmode == DLM_LOCK_CW)
2591 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2592 on either the convert or waiting queue.
2593 high is the largest rqmode of all locks blocked on the convert or
2596 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2598 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2599 if (gr->lkb_highbast < DLM_LOCK_EX)
2604 if (gr->lkb_highbast < high &&
2605 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2610 static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2612 struct dlm_lkb *lkb, *s;
2613 int high = DLM_LOCK_IV;
2616 if (!is_master(r)) {
2617 log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2622 high = grant_pending_convert(r, high, &cw, count);
2623 high = grant_pending_wait(r, high, &cw, count);
2625 if (high == DLM_LOCK_IV)
2629 * If there are locks left on the wait/convert queue then send blocking
2630 * ASTs to granted locks based on the largest requested mode (high)
2634 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2635 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2636 if (cw && high == DLM_LOCK_PR &&
2637 lkb->lkb_grmode == DLM_LOCK_PR)
2638 queue_bast(r, lkb, DLM_LOCK_CW);
2640 queue_bast(r, lkb, high);
2641 lkb->lkb_highbast = high;
2646 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2648 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2649 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2650 if (gr->lkb_highbast < DLM_LOCK_EX)
2655 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2660 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2661 struct dlm_lkb *lkb)
2665 list_for_each_entry(gr, head, lkb_statequeue) {
2666 /* skip self when sending basts to convertqueue */
2669 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2670 queue_bast(r, gr, lkb->lkb_rqmode);
2671 gr->lkb_highbast = lkb->lkb_rqmode;
2676 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2678 send_bast_queue(r, &r->res_grantqueue, lkb);
2681 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2683 send_bast_queue(r, &r->res_grantqueue, lkb);
2684 send_bast_queue(r, &r->res_convertqueue, lkb);
2687 /* set_master(r, lkb) -- set the master nodeid of a resource
2689 The purpose of this function is to set the nodeid field in the given
2690 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2691 known, it can just be copied to the lkb and the function will return
2692 0. If the rsb's nodeid is _not_ known, it needs to be looked up
2693 before it can be copied to the lkb.
2695 When the rsb nodeid is being looked up remotely, the initial lkb
2696 causing the lookup is kept on the ls_waiters list waiting for the
2697 lookup reply. Other lkb's waiting for the same rsb lookup are kept
2698 on the rsb's res_lookup list until the master is verified.
2701 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2702 1: the rsb master is not available and the lkb has been placed on
2706 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2708 int our_nodeid = dlm_our_nodeid();
2710 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2711 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2712 r->res_first_lkid = lkb->lkb_id;
2713 lkb->lkb_nodeid = r->res_nodeid;
2717 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2718 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2722 if (r->res_master_nodeid == our_nodeid) {
2723 lkb->lkb_nodeid = 0;
2727 if (r->res_master_nodeid) {
2728 lkb->lkb_nodeid = r->res_master_nodeid;
2732 if (dlm_dir_nodeid(r) == our_nodeid) {
2733 /* This is a somewhat unusual case; find_rsb will usually
2734 have set res_master_nodeid when dir nodeid is local, but
2735 there are cases where we become the dir node after we've
2736 past find_rsb and go through _request_lock again.
2737 confirm_master() or process_lookup_list() needs to be
2738 called after this. */
2739 log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2740 lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2742 r->res_master_nodeid = our_nodeid;
2744 lkb->lkb_nodeid = 0;
2748 wait_pending_remove(r);
2750 r->res_first_lkid = lkb->lkb_id;
2751 send_lookup(r, lkb);
2755 static void process_lookup_list(struct dlm_rsb *r)
2757 struct dlm_lkb *lkb, *safe;
2759 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2760 list_del_init(&lkb->lkb_rsb_lookup);
2761 _request_lock(r, lkb);
2766 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
2768 static void confirm_master(struct dlm_rsb *r, int error)
2770 struct dlm_lkb *lkb;
2772 if (!r->res_first_lkid)
2778 r->res_first_lkid = 0;
2779 process_lookup_list(r);
2785 /* the remote request failed and won't be retried (it was
2786 a NOQUEUE, or has been canceled/unlocked); make a waiting
2787 lkb the first_lkid */
2789 r->res_first_lkid = 0;
2791 if (!list_empty(&r->res_lookup)) {
2792 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2794 list_del_init(&lkb->lkb_rsb_lookup);
2795 r->res_first_lkid = lkb->lkb_id;
2796 _request_lock(r, lkb);
2801 log_error(r->res_ls, "confirm_master unknown error %d", error);
2805 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2806 int namelen, unsigned long timeout_cs,
2807 void (*ast) (void *astparam),
2809 void (*bast) (void *astparam, int mode),
2810 struct dlm_args *args)
2814 /* check for invalid arg usage */
2816 if (mode < 0 || mode > DLM_LOCK_EX)
2819 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2822 if (flags & DLM_LKF_CANCEL)
2825 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2828 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2831 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2834 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2837 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2840 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2843 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2849 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2852 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2855 /* these args will be copied to the lkb in validate_lock_args,
2856 it cannot be done now because when converting locks, fields in
2857 an active lkb cannot be modified before locking the rsb */
2859 args->flags = flags;
2861 args->astparam = astparam;
2862 args->bastfn = bast;
2863 args->timeout = timeout_cs;
2871 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2873 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2874 DLM_LKF_FORCEUNLOCK))
2877 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2880 args->flags = flags;
2881 args->astparam = astarg;
2885 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2886 struct dlm_args *args)
2890 if (args->flags & DLM_LKF_CONVERT) {
2891 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2894 if (args->flags & DLM_LKF_QUECVT &&
2895 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2899 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2902 if (lkb->lkb_wait_type)
2905 if (is_overlap(lkb))
2909 lkb->lkb_exflags = args->flags;
2910 lkb->lkb_sbflags = 0;
2911 lkb->lkb_astfn = args->astfn;
2912 lkb->lkb_astparam = args->astparam;
2913 lkb->lkb_bastfn = args->bastfn;
2914 lkb->lkb_rqmode = args->mode;
2915 lkb->lkb_lksb = args->lksb;
2916 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2917 lkb->lkb_ownpid = (int) current->pid;
2918 lkb->lkb_timeout_cs = args->timeout;
2922 log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2923 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2924 lkb->lkb_status, lkb->lkb_wait_type,
2925 lkb->lkb_resource->res_name);
2929 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2932 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2933 because there may be a lookup in progress and it's valid to do
2934 cancel/unlockf on it */
2936 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2938 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2941 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2942 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2947 /* an lkb may still exist even though the lock is EOL'ed due to a
2948 cancel, unlock or failed noqueue request; an app can't use these
2949 locks; return same error as if the lkid had not been found at all */
2951 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2952 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2957 /* an lkb may be waiting for an rsb lookup to complete where the
2958 lookup was initiated by another lock */
2960 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2961 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2962 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2963 list_del_init(&lkb->lkb_rsb_lookup);
2964 queue_cast(lkb->lkb_resource, lkb,
2965 args->flags & DLM_LKF_CANCEL ?
2966 -DLM_ECANCEL : -DLM_EUNLOCK);
2967 unhold_lkb(lkb); /* undoes create_lkb() */
2969 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2974 /* cancel not allowed with another cancel/unlock in progress */
2976 if (args->flags & DLM_LKF_CANCEL) {
2977 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2980 if (is_overlap(lkb))
2983 /* don't let scand try to do a cancel */
2986 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2987 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2992 /* there's nothing to cancel */
2993 if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2994 !lkb->lkb_wait_type) {
2999 switch (lkb->lkb_wait_type) {
3000 case DLM_MSG_LOOKUP:
3001 case DLM_MSG_REQUEST:
3002 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3005 case DLM_MSG_UNLOCK:
3006 case DLM_MSG_CANCEL:
3009 /* add_to_waiters() will set OVERLAP_CANCEL */
3013 /* do we need to allow a force-unlock if there's a normal unlock
3014 already in progress? in what conditions could the normal unlock
3015 fail such that we'd want to send a force-unlock to be sure? */
3017 if (args->flags & DLM_LKF_FORCEUNLOCK) {
3018 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3021 if (is_overlap_unlock(lkb))
3024 /* don't let scand try to do a cancel */
3027 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3028 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3033 switch (lkb->lkb_wait_type) {
3034 case DLM_MSG_LOOKUP:
3035 case DLM_MSG_REQUEST:
3036 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3039 case DLM_MSG_UNLOCK:
3042 /* add_to_waiters() will set OVERLAP_UNLOCK */
3046 /* normal unlock not allowed if there's any op in progress */
3048 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
3052 /* an overlapping op shouldn't blow away exflags from other op */
3053 lkb->lkb_exflags |= args->flags;
3054 lkb->lkb_sbflags = 0;
3055 lkb->lkb_astparam = args->astparam;
3059 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
3060 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3061 args->flags, lkb->lkb_wait_type,
3062 lkb->lkb_resource->res_name);
3067 * Four stage 4 varieties:
3068 * do_request(), do_convert(), do_unlock(), do_cancel()
3069 * These are called on the master node for the given lock and
3070 * from the central locking logic.
3073 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3077 if (can_be_granted(r, lkb, 1, 0, NULL)) {
3079 queue_cast(r, lkb, 0);
3083 if (can_be_queued(lkb)) {
3084 error = -EINPROGRESS;
3085 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3091 queue_cast(r, lkb, -EAGAIN);
3096 static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3101 if (force_blocking_asts(lkb))
3102 send_blocking_asts_all(r, lkb);
3105 send_blocking_asts(r, lkb);
3110 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3115 /* changing an existing lock may allow others to be granted */
3117 if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3119 queue_cast(r, lkb, 0);
3123 /* can_be_granted() detected that this lock would block in a conversion
3124 deadlock, so we leave it on the granted queue and return EDEADLK in
3125 the ast for the convert. */
3128 /* it's left on the granted queue */
3129 revert_lock(r, lkb);
3130 queue_cast(r, lkb, -EDEADLK);
3135 /* is_demoted() means the can_be_granted() above set the grmode
3136 to NL, and left us on the granted queue. This auto-demotion
3137 (due to CONVDEADLK) might mean other locks, and/or this lock, are
3138 now grantable. We have to try to grant other converting locks
3139 before we try again to grant this one. */
3141 if (is_demoted(lkb)) {
3142 grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3143 if (_can_be_granted(r, lkb, 1, 0)) {
3145 queue_cast(r, lkb, 0);
3148 /* else fall through and move to convert queue */
3151 if (can_be_queued(lkb)) {
3152 error = -EINPROGRESS;
3154 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3160 queue_cast(r, lkb, -EAGAIN);
3165 static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3170 grant_pending_locks(r, NULL);
3171 /* grant_pending_locks also sends basts */
3174 if (force_blocking_asts(lkb))
3175 send_blocking_asts_all(r, lkb);
3178 send_blocking_asts(r, lkb);
3183 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3185 remove_lock(r, lkb);
3186 queue_cast(r, lkb, -DLM_EUNLOCK);
3187 return -DLM_EUNLOCK;
3190 static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3193 grant_pending_locks(r, NULL);
3196 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3198 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3202 error = revert_lock(r, lkb);
3204 queue_cast(r, lkb, -DLM_ECANCEL);
3205 return -DLM_ECANCEL;
3210 static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3214 grant_pending_locks(r, NULL);
3218 * Four stage 3 varieties:
3219 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3222 /* add a new lkb to a possibly new rsb, called by requesting process */
3224 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3228 /* set_master: sets lkb nodeid from r */
3230 error = set_master(r, lkb);
3239 /* receive_request() calls do_request() on remote node */
3240 error = send_request(r, lkb);
3242 error = do_request(r, lkb);
3243 /* for remote locks the request_reply is sent
3244 between do_request and do_request_effects */
3245 do_request_effects(r, lkb, error);
3251 /* change some property of an existing lkb, e.g. mode */
3253 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3258 /* receive_convert() calls do_convert() on remote node */
3259 error = send_convert(r, lkb);
3261 error = do_convert(r, lkb);
3262 /* for remote locks the convert_reply is sent
3263 between do_convert and do_convert_effects */
3264 do_convert_effects(r, lkb, error);
3270 /* remove an existing lkb from the granted queue */
3272 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3277 /* receive_unlock() calls do_unlock() on remote node */
3278 error = send_unlock(r, lkb);
3280 error = do_unlock(r, lkb);
3281 /* for remote locks the unlock_reply is sent
3282 between do_unlock and do_unlock_effects */
3283 do_unlock_effects(r, lkb, error);
3289 /* remove an existing lkb from the convert or wait queue */
3291 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3296 /* receive_cancel() calls do_cancel() on remote node */
3297 error = send_cancel(r, lkb);
3299 error = do_cancel(r, lkb);
3300 /* for remote locks the cancel_reply is sent
3301 between do_cancel and do_cancel_effects */
3302 do_cancel_effects(r, lkb, error);
3309 * Four stage 2 varieties:
3310 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3313 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
3314 int len, struct dlm_args *args)
3319 error = validate_lock_args(ls, lkb, args);
3323 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3330 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3332 error = _request_lock(r, lkb);
3339 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3340 struct dlm_args *args)
3345 r = lkb->lkb_resource;
3350 error = validate_lock_args(ls, lkb, args);
3354 error = _convert_lock(r, lkb);
3361 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3362 struct dlm_args *args)
3367 r = lkb->lkb_resource;
3372 error = validate_unlock_args(lkb, args);
3376 error = _unlock_lock(r, lkb);
3383 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3384 struct dlm_args *args)
3389 r = lkb->lkb_resource;
3394 error = validate_unlock_args(lkb, args);
3398 error = _cancel_lock(r, lkb);
3406 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
3409 int dlm_lock(dlm_lockspace_t *lockspace,
3411 struct dlm_lksb *lksb,
3414 unsigned int namelen,
3415 uint32_t parent_lkid,
3416 void (*ast) (void *astarg),
3418 void (*bast) (void *astarg, int mode))
3421 struct dlm_lkb *lkb;
3422 struct dlm_args args;
3423 int error, convert = flags & DLM_LKF_CONVERT;
3425 ls = dlm_find_lockspace_local(lockspace);
3429 dlm_lock_recovery(ls);
3432 error = find_lkb(ls, lksb->sb_lkid, &lkb);
3434 error = create_lkb(ls, &lkb);
3439 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3440 astarg, bast, &args);
3445 error = convert_lock(ls, lkb, &args);
3447 error = request_lock(ls, lkb, name, namelen, &args);
3449 if (error == -EINPROGRESS)
3452 if (convert || error)
3454 if (error == -EAGAIN || error == -EDEADLK)
3457 dlm_unlock_recovery(ls);
3458 dlm_put_lockspace(ls);
3462 int dlm_unlock(dlm_lockspace_t *lockspace,
3465 struct dlm_lksb *lksb,
3469 struct dlm_lkb *lkb;
3470 struct dlm_args args;
3473 ls = dlm_find_lockspace_local(lockspace);
3477 dlm_lock_recovery(ls);
3479 error = find_lkb(ls, lkid, &lkb);
3483 error = set_unlock_args(flags, astarg, &args);
3487 if (flags & DLM_LKF_CANCEL)
3488 error = cancel_lock(ls, lkb, &args);
3490 error = unlock_lock(ls, lkb, &args);
3492 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3494 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3499 dlm_unlock_recovery(ls);
3500 dlm_put_lockspace(ls);
3505 * send/receive routines for remote operations and replies
3509 * send_request receive_request
3510 * send_convert receive_convert
3511 * send_unlock receive_unlock
3512 * send_cancel receive_cancel
3513 * send_grant receive_grant
3514 * send_bast receive_bast
3515 * send_lookup receive_lookup
3516 * send_remove receive_remove
3519 * receive_request_reply send_request_reply
3520 * receive_convert_reply send_convert_reply
3521 * receive_unlock_reply send_unlock_reply
3522 * receive_cancel_reply send_cancel_reply
3523 * receive_lookup_reply send_lookup_reply
3526 static int _create_message(struct dlm_ls *ls, int mb_len,
3527 int to_nodeid, int mstype,
3528 struct dlm_message **ms_ret,
3529 struct dlm_mhandle **mh_ret)
3531 struct dlm_message *ms;
3532 struct dlm_mhandle *mh;
3535 /* get_buffer gives us a message handle (mh) that we need to
3536 pass into lowcomms_commit and a message buffer (mb) that we
3537 write our data into */
3539 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
3543 memset(mb, 0, mb_len);
3545 ms = (struct dlm_message *) mb;
3547 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3548 ms->m_header.h_lockspace = ls->ls_global_id;
3549 ms->m_header.h_nodeid = dlm_our_nodeid();
3550 ms->m_header.h_length = mb_len;
3551 ms->m_header.h_cmd = DLM_MSG;
3553 ms->m_type = mstype;
3560 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3561 int to_nodeid, int mstype,
3562 struct dlm_message **ms_ret,
3563 struct dlm_mhandle **mh_ret)
3565 int mb_len = sizeof(struct dlm_message);
3568 case DLM_MSG_REQUEST:
3569 case DLM_MSG_LOOKUP:
3570 case DLM_MSG_REMOVE:
3571 mb_len += r->res_length;
3573 case DLM_MSG_CONVERT:
3574 case DLM_MSG_UNLOCK:
3575 case DLM_MSG_REQUEST_REPLY:
3576 case DLM_MSG_CONVERT_REPLY:
3578 if (lkb && lkb->lkb_lvbptr)
3579 mb_len += r->res_ls->ls_lvblen;
3583 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3587 /* further lowcomms enhancements or alternate implementations may make
3588 the return value from this function useful at some point */
3590 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
3592 dlm_message_out(ms);
3593 dlm_lowcomms_commit_buffer(mh);
3597 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3598 struct dlm_message *ms)
3600 ms->m_nodeid = lkb->lkb_nodeid;
3601 ms->m_pid = lkb->lkb_ownpid;
3602 ms->m_lkid = lkb->lkb_id;
3603 ms->m_remid = lkb->lkb_remid;
3604 ms->m_exflags = lkb->lkb_exflags;
3605 ms->m_sbflags = lkb->lkb_sbflags;
3606 ms->m_flags = lkb->lkb_flags;
3607 ms->m_lvbseq = lkb->lkb_lvbseq;
3608 ms->m_status = lkb->lkb_status;
3609 ms->m_grmode = lkb->lkb_grmode;
3610 ms->m_rqmode = lkb->lkb_rqmode;
3611 ms->m_hash = r->res_hash;
3613 /* m_result and m_bastmode are set from function args,
3614 not from lkb fields */
3616 if (lkb->lkb_bastfn)
3617 ms->m_asts |= DLM_CB_BAST;
3619 ms->m_asts |= DLM_CB_CAST;
3621 /* compare with switch in create_message; send_remove() doesn't
3624 switch (ms->m_type) {
3625 case DLM_MSG_REQUEST:
3626 case DLM_MSG_LOOKUP:
3627 memcpy(ms->m_extra, r->res_name, r->res_length);
3629 case DLM_MSG_CONVERT:
3630 case DLM_MSG_UNLOCK:
3631 case DLM_MSG_REQUEST_REPLY:
3632 case DLM_MSG_CONVERT_REPLY:
3634 if (!lkb->lkb_lvbptr)
3636 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3641 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3643 struct dlm_message *ms;
3644 struct dlm_mhandle *mh;
3645 int to_nodeid, error;
3647 to_nodeid = r->res_nodeid;
3649 error = add_to_waiters(lkb, mstype, to_nodeid);
3653 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3657 send_args(r, lkb, ms);
3659 error = send_message(mh, ms);
3665 remove_from_waiters(lkb, msg_reply_type(mstype));
3669 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3671 return send_common(r, lkb, DLM_MSG_REQUEST);
3674 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3678 error = send_common(r, lkb, DLM_MSG_CONVERT);
3680 /* down conversions go without a reply from the master */
3681 if (!error && down_conversion(lkb)) {
3682 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3683 r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS;
3684 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3685 r->res_ls->ls_stub_ms.m_result = 0;
3686 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3692 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
3693 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3694 that the master is still correct. */
3696 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3698 return send_common(r, lkb, DLM_MSG_UNLOCK);
3701 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3703 return send_common(r, lkb, DLM_MSG_CANCEL);
3706 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3708 struct dlm_message *ms;
3709 struct dlm_mhandle *mh;
3710 int to_nodeid, error;
3712 to_nodeid = lkb->lkb_nodeid;
3714 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3718 send_args(r, lkb, ms);
3722 error = send_message(mh, ms);
3727 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3729 struct dlm_message *ms;
3730 struct dlm_mhandle *mh;
3731 int to_nodeid, error;
3733 to_nodeid = lkb->lkb_nodeid;
3735 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3739 send_args(r, lkb, ms);
3741 ms->m_bastmode = mode;
3743 error = send_message(mh, ms);
3748 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3750 struct dlm_message *ms;
3751 struct dlm_mhandle *mh;
3752 int to_nodeid, error;
3754 to_nodeid = dlm_dir_nodeid(r);
3756 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3760 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3764 send_args(r, lkb, ms);
3766 error = send_message(mh, ms);
3772 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3776 static int send_remove(struct dlm_rsb *r)
3778 struct dlm_message *ms;
3779 struct dlm_mhandle *mh;
3780 int to_nodeid, error;
3782 to_nodeid = dlm_dir_nodeid(r);
3784 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3788 memcpy(ms->m_extra, r->res_name, r->res_length);
3789 ms->m_hash = r->res_hash;
3791 error = send_message(mh, ms);
3796 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3799 struct dlm_message *ms;
3800 struct dlm_mhandle *mh;
3801 int to_nodeid, error;
3803 to_nodeid = lkb->lkb_nodeid;
3805 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3809 send_args(r, lkb, ms);
3813 error = send_message(mh, ms);
3818 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3820 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3823 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3825 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3828 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3830 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3833 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3835 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3838 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3839 int ret_nodeid, int rv)
3841 struct dlm_rsb *r = &ls->ls_stub_rsb;
3842 struct dlm_message *ms;
3843 struct dlm_mhandle *mh;
3844 int error, nodeid = ms_in->m_header.h_nodeid;
3846 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3850 ms->m_lkid = ms_in->m_lkid;
3852 ms->m_nodeid = ret_nodeid;
3854 error = send_message(mh, ms);
3859 /* which args we save from a received message depends heavily on the type
3860 of message, unlike the send side where we can safely send everything about
3861 the lkb for any type of message */
3863 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3865 lkb->lkb_exflags = ms->m_exflags;
3866 lkb->lkb_sbflags = ms->m_sbflags;
3867 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3868 (ms->m_flags & 0x0000FFFF);
3871 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3873 if (ms->m_flags == DLM_IFL_STUB_MS)
3876 lkb->lkb_sbflags = ms->m_sbflags;
3877 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3878 (ms->m_flags & 0x0000FFFF);
3881 static int receive_extralen(struct dlm_message *ms)
3883 return (ms->m_header.h_length - sizeof(struct dlm_message));
3886 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3887 struct dlm_message *ms)
3891 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3892 if (!lkb->lkb_lvbptr)
3893 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3894 if (!lkb->lkb_lvbptr)
3896 len = receive_extralen(ms);
3897 if (len > ls->ls_lvblen)
3898 len = ls->ls_lvblen;
3899 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3904 static void fake_bastfn(void *astparam, int mode)
3906 log_print("fake_bastfn should not be called");
3909 static void fake_astfn(void *astparam)
3911 log_print("fake_astfn should not be called");
3914 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3915 struct dlm_message *ms)
3917 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3918 lkb->lkb_ownpid = ms->m_pid;
3919 lkb->lkb_remid = ms->m_lkid;
3920 lkb->lkb_grmode = DLM_LOCK_IV;
3921 lkb->lkb_rqmode = ms->m_rqmode;
3923 lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
3924 lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
3926 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3927 /* lkb was just created so there won't be an lvb yet */
3928 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3929 if (!lkb->lkb_lvbptr)
3936 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3937 struct dlm_message *ms)
3939 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3942 if (receive_lvb(ls, lkb, ms))
3945 lkb->lkb_rqmode = ms->m_rqmode;
3946 lkb->lkb_lvbseq = ms->m_lvbseq;
3951 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3952 struct dlm_message *ms)
3954 if (receive_lvb(ls, lkb, ms))
3959 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3960 uses to send a reply and that the remote end uses to process the reply. */
3962 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3964 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3965 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3966 lkb->lkb_remid = ms->m_lkid;
3969 /* This is called after the rsb is locked so that we can safely inspect
3970 fields in the lkb. */
3972 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3974 int from = ms->m_header.h_nodeid;
3977 /* currently mixing of user/kernel locks are not supported */
3978 if (ms->m_flags & DLM_IFL_USER && ~lkb->lkb_flags & DLM_IFL_USER) {
3979 log_error(lkb->lkb_resource->res_ls,
3980 "got user dlm message for a kernel lock");
3985 switch (ms->m_type) {
3986 case DLM_MSG_CONVERT:
3987 case DLM_MSG_UNLOCK:
3988 case DLM_MSG_CANCEL:
3989 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3993 case DLM_MSG_CONVERT_REPLY:
3994 case DLM_MSG_UNLOCK_REPLY:
3995 case DLM_MSG_CANCEL_REPLY:
3998 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
4002 case DLM_MSG_REQUEST_REPLY:
4003 if (!is_process_copy(lkb))
4005 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
4015 log_error(lkb->lkb_resource->res_ls,
4016 "ignore invalid message %d from %d %x %x %x %d",
4017 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
4018 lkb->lkb_flags, lkb->lkb_nodeid);
4022 static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
4024 char name[DLM_RESNAME_MAXLEN + 1];
4025 struct dlm_message *ms;
4026 struct dlm_mhandle *mh;
4031 memset(name, 0, sizeof(name));
4032 memcpy(name, ms_name, len);
4034 hash = jhash(name, len, 0);
4035 b = hash & (ls->ls_rsbtbl_size - 1);
4037 dir_nodeid = dlm_hash2nodeid(ls, hash);
4039 log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
4041 spin_lock(&ls->ls_rsbtbl[b].lock);
4042 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4044 spin_unlock(&ls->ls_rsbtbl[b].lock);
4045 log_error(ls, "repeat_remove on keep %s", name);
4049 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4051 spin_unlock(&ls->ls_rsbtbl[b].lock);
4052 log_error(ls, "repeat_remove on toss %s", name);
4056 /* use ls->remove_name2 to avoid conflict with shrink? */
4058 spin_lock(&ls->ls_remove_spin);
4059 ls->ls_remove_len = len;
4060 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
4061 spin_unlock(&ls->ls_remove_spin);
4062 spin_unlock(&ls->ls_rsbtbl[b].lock);
4064 rv = _create_message(ls, sizeof(struct dlm_message) + len,
4065 dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
4069 memcpy(ms->m_extra, name, len);
4072 send_message(mh, ms);
4074 spin_lock(&ls->ls_remove_spin);
4075 ls->ls_remove_len = 0;
4076 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
4077 spin_unlock(&ls->ls_remove_spin);
4080 static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4082 struct dlm_lkb *lkb;
4085 int error, namelen = 0;
4087 from_nodeid = ms->m_header.h_nodeid;
4089 error = create_lkb(ls, &lkb);
4093 receive_flags(lkb, ms);
4094 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4095 error = receive_request_args(ls, lkb, ms);
4101 /* The dir node is the authority on whether we are the master
4102 for this rsb or not, so if the master sends us a request, we should
4103 recreate the rsb if we've destroyed it. This race happens when we
4104 send a remove message to the dir node at the same time that the dir
4105 node sends us a request for the rsb. */
4107 namelen = receive_extralen(ms);
4109 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4110 R_RECEIVE_REQUEST, &r);
4118 if (r->res_master_nodeid != dlm_our_nodeid()) {
4119 error = validate_master_nodeid(ls, r, from_nodeid);
4129 error = do_request(r, lkb);
4130 send_request_reply(r, lkb, error);
4131 do_request_effects(r, lkb, error);
4136 if (error == -EINPROGRESS)
4143 /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4144 and do this receive_request again from process_lookup_list once
4145 we get the lookup reply. This would avoid a many repeated
4146 ENOTBLK request failures when the lookup reply designating us
4147 as master is delayed. */
4149 /* We could repeatedly return -EBADR here if our send_remove() is
4150 delayed in being sent/arriving/being processed on the dir node.
4151 Another node would repeatedly lookup up the master, and the dir
4152 node would continue returning our nodeid until our send_remove
4155 We send another remove message in case our previous send_remove
4156 was lost/ignored/missed somehow. */
4158 if (error != -ENOTBLK) {
4159 log_limit(ls, "receive_request %x from %d %d",
4160 ms->m_lkid, from_nodeid, error);
4163 if (namelen && error == -EBADR) {
4164 send_repeat_remove(ls, ms->m_extra, namelen);
4168 setup_stub_lkb(ls, ms);
4169 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4173 static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4175 struct dlm_lkb *lkb;
4177 int error, reply = 1;
4179 error = find_lkb(ls, ms->m_remid, &lkb);
4183 if (lkb->lkb_remid != ms->m_lkid) {
4184 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4185 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4186 (unsigned long long)lkb->lkb_recover_seq,
4187 ms->m_header.h_nodeid, ms->m_lkid);
4193 r = lkb->lkb_resource;
4198 error = validate_message(lkb, ms);
4202 receive_flags(lkb, ms);
4204 error = receive_convert_args(ls, lkb, ms);
4206 send_convert_reply(r, lkb, error);
4210 reply = !down_conversion(lkb);
4212 error = do_convert(r, lkb);
4214 send_convert_reply(r, lkb, error);
4215 do_convert_effects(r, lkb, error);
4223 setup_stub_lkb(ls, ms);
4224 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4228 static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4230 struct dlm_lkb *lkb;
4234 error = find_lkb(ls, ms->m_remid, &lkb);
4238 if (lkb->lkb_remid != ms->m_lkid) {
4239 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4240 lkb->lkb_id, lkb->lkb_remid,
4241 ms->m_header.h_nodeid, ms->m_lkid);
4247 r = lkb->lkb_resource;
4252 error = validate_message(lkb, ms);
4256 receive_flags(lkb, ms);
4258 error = receive_unlock_args(ls, lkb, ms);
4260 send_unlock_reply(r, lkb, error);
4264 error = do_unlock(r, lkb);
4265 send_unlock_reply(r, lkb, error);
4266 do_unlock_effects(r, lkb, error);
4274 setup_stub_lkb(ls, ms);
4275 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4279 static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4281 struct dlm_lkb *lkb;
4285 error = find_lkb(ls, ms->m_remid, &lkb);
4289 receive_flags(lkb, ms);
4291 r = lkb->lkb_resource;
4296 error = validate_message(lkb, ms);
4300 error = do_cancel(r, lkb);
4301 send_cancel_reply(r, lkb, error);
4302 do_cancel_effects(r, lkb, error);
4310 setup_stub_lkb(ls, ms);
4311 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4315 static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4317 struct dlm_lkb *lkb;
4321 error = find_lkb(ls, ms->m_remid, &lkb);
4325 r = lkb->lkb_resource;
4330 error = validate_message(lkb, ms);
4334 receive_flags_reply(lkb, ms);
4335 if (is_altmode(lkb))
4336 munge_altmode(lkb, ms);
4337 grant_lock_pc(r, lkb, ms);
4338 queue_cast(r, lkb, 0);
4346 static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4348 struct dlm_lkb *lkb;
4352 error = find_lkb(ls, ms->m_remid, &lkb);
4356 r = lkb->lkb_resource;
4361 error = validate_message(lkb, ms);
4365 queue_bast(r, lkb, ms->m_bastmode);
4366 lkb->lkb_highbast = ms->m_bastmode;
4374 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4376 int len, error, ret_nodeid, from_nodeid, our_nodeid;
4378 from_nodeid = ms->m_header.h_nodeid;
4379 our_nodeid = dlm_our_nodeid();
4381 len = receive_extralen(ms);
4383 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4386 /* Optimization: we're master so treat lookup as a request */
4387 if (!error && ret_nodeid == our_nodeid) {
4388 receive_request(ls, ms);
4391 send_lookup_reply(ls, ms, ret_nodeid, error);
4394 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4396 char name[DLM_RESNAME_MAXLEN+1];
4399 int rv, len, dir_nodeid, from_nodeid;
4401 from_nodeid = ms->m_header.h_nodeid;
4403 len = receive_extralen(ms);
4405 if (len > DLM_RESNAME_MAXLEN) {
4406 log_error(ls, "receive_remove from %d bad len %d",
4411 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
4412 if (dir_nodeid != dlm_our_nodeid()) {
4413 log_error(ls, "receive_remove from %d bad nodeid %d",
4414 from_nodeid, dir_nodeid);
4418 /* Look for name on rsbtbl.toss, if it's there, kill it.
4419 If it's on rsbtbl.keep, it's being used, and we should ignore this
4420 message. This is an expected race between the dir node sending a
4421 request to the master node at the same time as the master node sends
4422 a remove to the dir node. The resolution to that race is for the
4423 dir node to ignore the remove message, and the master node to
4424 recreate the master rsb when it gets a request from the dir node for
4425 an rsb it doesn't have. */
4427 memset(name, 0, sizeof(name));
4428 memcpy(name, ms->m_extra, len);
4430 hash = jhash(name, len, 0);
4431 b = hash & (ls->ls_rsbtbl_size - 1);
4433 spin_lock(&ls->ls_rsbtbl[b].lock);
4435 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4437 /* verify the rsb is on keep list per comment above */
4438 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4440 /* should not happen */
4441 log_error(ls, "receive_remove from %d not found %s",
4443 spin_unlock(&ls->ls_rsbtbl[b].lock);
4446 if (r->res_master_nodeid != from_nodeid) {
4447 /* should not happen */
4448 log_error(ls, "receive_remove keep from %d master %d",
4449 from_nodeid, r->res_master_nodeid);
4451 spin_unlock(&ls->ls_rsbtbl[b].lock);
4455 log_debug(ls, "receive_remove from %d master %d first %x %s",
4456 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4458 spin_unlock(&ls->ls_rsbtbl[b].lock);
4462 if (r->res_master_nodeid != from_nodeid) {
4463 log_error(ls, "receive_remove toss from %d master %d",
4464 from_nodeid, r->res_master_nodeid);
4466 spin_unlock(&ls->ls_rsbtbl[b].lock);
4470 if (kref_put(&r->res_ref, kill_rsb)) {
4471 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4472 spin_unlock(&ls->ls_rsbtbl[b].lock);
4475 log_error(ls, "receive_remove from %d rsb ref error",
4478 spin_unlock(&ls->ls_rsbtbl[b].lock);
4482 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4484 do_purge(ls, ms->m_nodeid, ms->m_pid);
4487 static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4489 struct dlm_lkb *lkb;
4491 int error, mstype, result;
4492 int from_nodeid = ms->m_header.h_nodeid;
4494 error = find_lkb(ls, ms->m_remid, &lkb);
4498 r = lkb->lkb_resource;
4502 error = validate_message(lkb, ms);
4506 mstype = lkb->lkb_wait_type;
4507 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4509 log_error(ls, "receive_request_reply %x remote %d %x result %d",
4510 lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result);
4515 /* Optimization: the dir node was also the master, so it took our
4516 lookup as a request and sent request reply instead of lookup reply */
4517 if (mstype == DLM_MSG_LOOKUP) {
4518 r->res_master_nodeid = from_nodeid;
4519 r->res_nodeid = from_nodeid;
4520 lkb->lkb_nodeid = from_nodeid;
4523 /* this is the value returned from do_request() on the master */
4524 result = ms->m_result;
4528 /* request would block (be queued) on remote master */
4529 queue_cast(r, lkb, -EAGAIN);
4530 confirm_master(r, -EAGAIN);
4531 unhold_lkb(lkb); /* undoes create_lkb() */
4536 /* request was queued or granted on remote master */
4537 receive_flags_reply(lkb, ms);
4538 lkb->lkb_remid = ms->m_lkid;
4539 if (is_altmode(lkb))
4540 munge_altmode(lkb, ms);
4542 add_lkb(r, lkb, DLM_LKSTS_WAITING);
4545 grant_lock_pc(r, lkb, ms);
4546 queue_cast(r, lkb, 0);
4548 confirm_master(r, result);
4553 /* find_rsb failed to find rsb or rsb wasn't master */
4554 log_limit(ls, "receive_request_reply %x from %d %d "
4555 "master %d dir %d first %x %s", lkb->lkb_id,
4556 from_nodeid, result, r->res_master_nodeid,
4557 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4559 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4560 r->res_master_nodeid != dlm_our_nodeid()) {
4561 /* cause _request_lock->set_master->send_lookup */
4562 r->res_master_nodeid = 0;
4564 lkb->lkb_nodeid = -1;
4567 if (is_overlap(lkb)) {
4568 /* we'll ignore error in cancel/unlock reply */
4569 queue_cast_overlap(r, lkb);
4570 confirm_master(r, result);
4571 unhold_lkb(lkb); /* undoes create_lkb() */
4573 _request_lock(r, lkb);
4575 if (r->res_master_nodeid == dlm_our_nodeid())
4576 confirm_master(r, 0);
4581 log_error(ls, "receive_request_reply %x error %d",
4582 lkb->lkb_id, result);
4585 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4586 log_debug(ls, "receive_request_reply %x result %d unlock",
4587 lkb->lkb_id, result);
4588 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4589 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4590 send_unlock(r, lkb);
4591 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4592 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4593 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4594 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4595 send_cancel(r, lkb);
4597 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4598 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4607 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4608 struct dlm_message *ms)
4610 /* this is the value returned from do_convert() on the master */
4611 switch (ms->m_result) {
4613 /* convert would block (be queued) on remote master */
4614 queue_cast(r, lkb, -EAGAIN);
4618 receive_flags_reply(lkb, ms);
4619 revert_lock_pc(r, lkb);
4620 queue_cast(r, lkb, -EDEADLK);
4624 /* convert was queued on remote master */
4625 receive_flags_reply(lkb, ms);
4626 if (is_demoted(lkb))
4629 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4634 /* convert was granted on remote master */
4635 receive_flags_reply(lkb, ms);
4636 if (is_demoted(lkb))
4638 grant_lock_pc(r, lkb, ms);
4639 queue_cast(r, lkb, 0);
4643 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4644 lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid,
4651 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4653 struct dlm_rsb *r = lkb->lkb_resource;
4659 error = validate_message(lkb, ms);
4663 /* stub reply can happen with waiters_mutex held */
4664 error = remove_from_waiters_ms(lkb, ms);
4668 __receive_convert_reply(r, lkb, ms);
4674 static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4676 struct dlm_lkb *lkb;
4679 error = find_lkb(ls, ms->m_remid, &lkb);
4683 _receive_convert_reply(lkb, ms);
4688 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4690 struct dlm_rsb *r = lkb->lkb_resource;
4696 error = validate_message(lkb, ms);
4700 /* stub reply can happen with waiters_mutex held */
4701 error = remove_from_waiters_ms(lkb, ms);
4705 /* this is the value returned from do_unlock() on the master */
4707 switch (ms->m_result) {
4709 receive_flags_reply(lkb, ms);
4710 remove_lock_pc(r, lkb);
4711 queue_cast(r, lkb, -DLM_EUNLOCK);
4716 log_error(r->res_ls, "receive_unlock_reply %x error %d",
4717 lkb->lkb_id, ms->m_result);
4724 static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4726 struct dlm_lkb *lkb;
4729 error = find_lkb(ls, ms->m_remid, &lkb);
4733 _receive_unlock_reply(lkb, ms);
4738 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4740 struct dlm_rsb *r = lkb->lkb_resource;
4746 error = validate_message(lkb, ms);
4750 /* stub reply can happen with waiters_mutex held */
4751 error = remove_from_waiters_ms(lkb, ms);
4755 /* this is the value returned from do_cancel() on the master */
4757 switch (ms->m_result) {
4759 receive_flags_reply(lkb, ms);
4760 revert_lock_pc(r, lkb);
4761 queue_cast(r, lkb, -DLM_ECANCEL);
4766 log_error(r->res_ls, "receive_cancel_reply %x error %d",
4767 lkb->lkb_id, ms->m_result);
4774 static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4776 struct dlm_lkb *lkb;
4779 error = find_lkb(ls, ms->m_remid, &lkb);
4783 _receive_cancel_reply(lkb, ms);
4788 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4790 struct dlm_lkb *lkb;
4792 int error, ret_nodeid;
4793 int do_lookup_list = 0;
4795 error = find_lkb(ls, ms->m_lkid, &lkb);
4797 log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid);
4801 /* ms->m_result is the value returned by dlm_master_lookup on dir node
4802 FIXME: will a non-zero error ever be returned? */
4804 r = lkb->lkb_resource;
4808 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4812 ret_nodeid = ms->m_nodeid;
4814 /* We sometimes receive a request from the dir node for this
4815 rsb before we've received the dir node's loookup_reply for it.
4816 The request from the dir node implies we're the master, so we set
4817 ourself as master in receive_request_reply, and verify here that
4818 we are indeed the master. */
4820 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4821 /* This should never happen */
4822 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4823 "master %d dir %d our %d first %x %s",
4824 lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid,
4825 r->res_master_nodeid, r->res_dir_nodeid,
4826 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4829 if (ret_nodeid == dlm_our_nodeid()) {
4830 r->res_master_nodeid = ret_nodeid;
4833 r->res_first_lkid = 0;
4834 } else if (ret_nodeid == -1) {
4835 /* the remote node doesn't believe it's the dir node */
4836 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4837 lkb->lkb_id, ms->m_header.h_nodeid);
4838 r->res_master_nodeid = 0;
4840 lkb->lkb_nodeid = -1;
4842 /* set_master() will set lkb_nodeid from r */
4843 r->res_master_nodeid = ret_nodeid;
4844 r->res_nodeid = ret_nodeid;
4847 if (is_overlap(lkb)) {
4848 log_debug(ls, "receive_lookup_reply %x unlock %x",
4849 lkb->lkb_id, lkb->lkb_flags);
4850 queue_cast_overlap(r, lkb);
4851 unhold_lkb(lkb); /* undoes create_lkb() */
4855 _request_lock(r, lkb);
4859 process_lookup_list(r);
4866 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4869 int error = 0, noent = 0;
4871 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
4872 log_limit(ls, "receive %d from non-member %d %x %x %d",
4873 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4874 ms->m_remid, ms->m_result);
4878 switch (ms->m_type) {
4880 /* messages sent to a master node */
4882 case DLM_MSG_REQUEST:
4883 error = receive_request(ls, ms);
4886 case DLM_MSG_CONVERT:
4887 error = receive_convert(ls, ms);
4890 case DLM_MSG_UNLOCK:
4891 error = receive_unlock(ls, ms);
4894 case DLM_MSG_CANCEL:
4896 error = receive_cancel(ls, ms);
4899 /* messages sent from a master node (replies to above) */
4901 case DLM_MSG_REQUEST_REPLY:
4902 error = receive_request_reply(ls, ms);
4905 case DLM_MSG_CONVERT_REPLY:
4906 error = receive_convert_reply(ls, ms);
4909 case DLM_MSG_UNLOCK_REPLY:
4910 error = receive_unlock_reply(ls, ms);
4913 case DLM_MSG_CANCEL_REPLY:
4914 error = receive_cancel_reply(ls, ms);
4917 /* messages sent from a master node (only two types of async msg) */
4921 error = receive_grant(ls, ms);
4926 error = receive_bast(ls, ms);
4929 /* messages sent to a dir node */
4931 case DLM_MSG_LOOKUP:
4932 receive_lookup(ls, ms);
4935 case DLM_MSG_REMOVE:
4936 receive_remove(ls, ms);
4939 /* messages sent from a dir node (remove has no reply) */
4941 case DLM_MSG_LOOKUP_REPLY:
4942 receive_lookup_reply(ls, ms);
4945 /* other messages */
4948 receive_purge(ls, ms);
4952 log_error(ls, "unknown message type %d", ms->m_type);
4956 * When checking for ENOENT, we're checking the result of
4957 * find_lkb(m_remid):
4959 * The lock id referenced in the message wasn't found. This may
4960 * happen in normal usage for the async messages and cancel, so
4961 * only use log_debug for them.
4963 * Some errors are expected and normal.
4966 if (error == -ENOENT && noent) {
4967 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4968 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4969 ms->m_lkid, saved_seq);
4970 } else if (error == -ENOENT) {
4971 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4972 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4973 ms->m_lkid, saved_seq);
4975 if (ms->m_type == DLM_MSG_CONVERT)
4976 dlm_dump_rsb_hash(ls, ms->m_hash);
4979 if (error == -EINVAL) {
4980 log_error(ls, "receive %d inval from %d lkid %x remid %x "
4982 ms->m_type, ms->m_header.h_nodeid,
4983 ms->m_lkid, ms->m_remid, saved_seq);
4987 /* If the lockspace is in recovery mode (locking stopped), then normal
4988 messages are saved on the requestqueue for processing after recovery is
4989 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4990 messages off the requestqueue before we process new ones. This occurs right
4991 after recovery completes when we transition from saving all messages on
4992 requestqueue, to processing all the saved messages, to processing new
4993 messages as they arrive. */
4995 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4998 if (dlm_locking_stopped(ls)) {
4999 /* If we were a member of this lockspace, left, and rejoined,
5000 other nodes may still be sending us messages from the
5001 lockspace generation before we left. */
5002 if (!ls->ls_generation) {
5003 log_limit(ls, "receive %d from %d ignore old gen",
5004 ms->m_type, nodeid);
5008 dlm_add_requestqueue(ls, nodeid, ms);
5010 dlm_wait_requestqueue(ls);
5011 _receive_message(ls, ms, 0);
5015 /* This is called by dlm_recoverd to process messages that were saved on
5016 the requestqueue. */
5018 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
5021 _receive_message(ls, ms, saved_seq);
5024 /* This is called by the midcomms layer when something is received for
5025 the lockspace. It could be either a MSG (normal message sent as part of
5026 standard locking activity) or an RCOM (recovery message sent as part of
5027 lockspace recovery). */
5029 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
5031 struct dlm_header *hd = &p->header;
5035 switch (hd->h_cmd) {
5037 dlm_message_in(&p->message);
5038 type = p->message.m_type;
5041 dlm_rcom_in(&p->rcom);
5042 type = p->rcom.rc_type;
5045 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5049 if (hd->h_nodeid != nodeid) {
5050 log_print("invalid h_nodeid %d from %d lockspace %x",
5051 hd->h_nodeid, nodeid, hd->h_lockspace);
5055 ls = dlm_find_lockspace_global(hd->h_lockspace);
5057 if (dlm_config.ci_log_debug) {
5058 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5059 "%u from %d cmd %d type %d\n",
5060 hd->h_lockspace, nodeid, hd->h_cmd, type);
5063 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5064 dlm_send_ls_not_ready(nodeid, &p->rcom);
5068 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5069 be inactive (in this ls) before transitioning to recovery mode */
5071 down_read(&ls->ls_recv_active);
5072 if (hd->h_cmd == DLM_MSG)
5073 dlm_receive_message(ls, &p->message, nodeid);
5075 dlm_receive_rcom(ls, &p->rcom, nodeid);
5076 up_read(&ls->ls_recv_active);
5078 dlm_put_lockspace(ls);
5081 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5082 struct dlm_message *ms_stub)
5084 if (middle_conversion(lkb)) {
5086 memset(ms_stub, 0, sizeof(struct dlm_message));
5087 ms_stub->m_flags = DLM_IFL_STUB_MS;
5088 ms_stub->m_type = DLM_MSG_CONVERT_REPLY;
5089 ms_stub->m_result = -EINPROGRESS;
5090 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5091 _receive_convert_reply(lkb, ms_stub);
5093 /* Same special case as in receive_rcom_lock_args() */
5094 lkb->lkb_grmode = DLM_LOCK_IV;
5095 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5098 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5099 lkb->lkb_flags |= DLM_IFL_RESEND;
5102 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5103 conversions are async; there's no reply from the remote master */
5106 /* A waiting lkb needs recovery if the master node has failed, or
5107 the master node is changing (only when no directory is used) */
5109 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5112 if (dlm_no_directory(ls))
5115 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5121 /* Recovery for locks that are waiting for replies from nodes that are now
5122 gone. We can just complete unlocks and cancels by faking a reply from the
5123 dead node. Requests and up-conversions we flag to be resent after
5124 recovery. Down-conversions can just be completed with a fake reply like
5125 unlocks. Conversions between PR and CW need special attention. */
5127 void dlm_recover_waiters_pre(struct dlm_ls *ls)
5129 struct dlm_lkb *lkb, *safe;
5130 struct dlm_message *ms_stub;
5131 int wait_type, stub_unlock_result, stub_cancel_result;
5134 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5138 mutex_lock(&ls->ls_waiters_mutex);
5140 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5142 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5144 /* exclude debug messages about unlocks because there can be so
5145 many and they aren't very interesting */
5147 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5148 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5149 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5153 lkb->lkb_resource->res_nodeid,
5155 lkb->lkb_wait_nodeid,
5159 /* all outstanding lookups, regardless of destination will be
5160 resent after recovery is done */
5162 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5163 lkb->lkb_flags |= DLM_IFL_RESEND;
5167 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5170 wait_type = lkb->lkb_wait_type;
5171 stub_unlock_result = -DLM_EUNLOCK;
5172 stub_cancel_result = -DLM_ECANCEL;
5174 /* Main reply may have been received leaving a zero wait_type,
5175 but a reply for the overlapping op may not have been
5176 received. In that case we need to fake the appropriate
5177 reply for the overlap op. */
5180 if (is_overlap_cancel(lkb)) {
5181 wait_type = DLM_MSG_CANCEL;
5182 if (lkb->lkb_grmode == DLM_LOCK_IV)
5183 stub_cancel_result = 0;
5185 if (is_overlap_unlock(lkb)) {
5186 wait_type = DLM_MSG_UNLOCK;
5187 if (lkb->lkb_grmode == DLM_LOCK_IV)
5188 stub_unlock_result = -ENOENT;
5191 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5192 lkb->lkb_id, lkb->lkb_flags, wait_type,
5193 stub_cancel_result, stub_unlock_result);
5196 switch (wait_type) {
5198 case DLM_MSG_REQUEST:
5199 lkb->lkb_flags |= DLM_IFL_RESEND;
5202 case DLM_MSG_CONVERT:
5203 recover_convert_waiter(ls, lkb, ms_stub);
5206 case DLM_MSG_UNLOCK:
5208 memset(ms_stub, 0, sizeof(struct dlm_message));
5209 ms_stub->m_flags = DLM_IFL_STUB_MS;
5210 ms_stub->m_type = DLM_MSG_UNLOCK_REPLY;
5211 ms_stub->m_result = stub_unlock_result;
5212 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5213 _receive_unlock_reply(lkb, ms_stub);
5217 case DLM_MSG_CANCEL:
5219 memset(ms_stub, 0, sizeof(struct dlm_message));
5220 ms_stub->m_flags = DLM_IFL_STUB_MS;
5221 ms_stub->m_type = DLM_MSG_CANCEL_REPLY;
5222 ms_stub->m_result = stub_cancel_result;
5223 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5224 _receive_cancel_reply(lkb, ms_stub);
5229 log_error(ls, "invalid lkb wait_type %d %d",
5230 lkb->lkb_wait_type, wait_type);
5234 mutex_unlock(&ls->ls_waiters_mutex);
5238 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5240 struct dlm_lkb *lkb;
5243 mutex_lock(&ls->ls_waiters_mutex);
5244 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
5245 if (lkb->lkb_flags & DLM_IFL_RESEND) {
5251 mutex_unlock(&ls->ls_waiters_mutex);
5258 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
5259 master or dir-node for r. Processing the lkb may result in it being placed
5262 /* We do this after normal locking has been enabled and any saved messages
5263 (in requestqueue) have been processed. We should be confident that at
5264 this point we won't get or process a reply to any of these waiting
5265 operations. But, new ops may be coming in on the rsbs/locks here from
5266 userspace or remotely. */
5268 /* there may have been an overlap unlock/cancel prior to recovery or after
5269 recovery. if before, the lkb may still have a pos wait_count; if after, the
5270 overlap flag would just have been set and nothing new sent. we can be
5271 confident here than any replies to either the initial op or overlap ops
5272 prior to recovery have been received. */
5274 int dlm_recover_waiters_post(struct dlm_ls *ls)
5276 struct dlm_lkb *lkb;
5278 int error = 0, mstype, err, oc, ou;
5281 if (dlm_locking_stopped(ls)) {
5282 log_debug(ls, "recover_waiters_post aborted");
5287 lkb = find_resend_waiter(ls);
5291 r = lkb->lkb_resource;
5295 mstype = lkb->lkb_wait_type;
5296 oc = is_overlap_cancel(lkb);
5297 ou = is_overlap_unlock(lkb);
5300 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5301 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5302 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5303 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5304 dlm_dir_nodeid(r), oc, ou);
5306 /* At this point we assume that we won't get a reply to any
5307 previous op or overlap op on this lock. First, do a big
5308 remove_from_waiters() for all previous ops. */
5310 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5311 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5312 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5313 lkb->lkb_wait_type = 0;
5314 lkb->lkb_wait_count = 0;
5315 mutex_lock(&ls->ls_waiters_mutex);
5316 list_del_init(&lkb->lkb_wait_reply);
5317 mutex_unlock(&ls->ls_waiters_mutex);
5318 unhold_lkb(lkb); /* for waiters list */
5321 /* do an unlock or cancel instead of resending */
5323 case DLM_MSG_LOOKUP:
5324 case DLM_MSG_REQUEST:
5325 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5327 unhold_lkb(lkb); /* undoes create_lkb() */
5329 case DLM_MSG_CONVERT:
5331 queue_cast(r, lkb, -DLM_ECANCEL);
5333 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5334 _unlock_lock(r, lkb);
5342 case DLM_MSG_LOOKUP:
5343 case DLM_MSG_REQUEST:
5344 _request_lock(r, lkb);
5346 confirm_master(r, 0);
5348 case DLM_MSG_CONVERT:
5349 _convert_lock(r, lkb);
5357 log_error(ls, "waiter %x msg %d r_nodeid %d "
5358 "dir_nodeid %d overlap %d %d",
5359 lkb->lkb_id, mstype, r->res_nodeid,
5360 dlm_dir_nodeid(r), oc, ou);
5370 static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5371 struct list_head *list)
5373 struct dlm_lkb *lkb, *safe;
5375 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5376 if (!is_master_copy(lkb))
5379 /* don't purge lkbs we've added in recover_master_copy for
5380 the current recovery seq */
5382 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5387 /* this put should free the lkb */
5388 if (!dlm_put_lkb(lkb))
5389 log_error(ls, "purged mstcpy lkb not released");
5393 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5395 struct dlm_ls *ls = r->res_ls;
5397 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5398 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5399 purge_mstcpy_list(ls, r, &r->res_waitqueue);
5402 static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5403 struct list_head *list,
5404 int nodeid_gone, unsigned int *count)
5406 struct dlm_lkb *lkb, *safe;
5408 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5409 if (!is_master_copy(lkb))
5412 if ((lkb->lkb_nodeid == nodeid_gone) ||
5413 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5415 /* tell recover_lvb to invalidate the lvb
5416 because a node holding EX/PW failed */
5417 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5418 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5419 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5424 /* this put should free the lkb */
5425 if (!dlm_put_lkb(lkb))
5426 log_error(ls, "purged dead lkb not released");
5428 rsb_set_flag(r, RSB_RECOVER_GRANT);
5435 /* Get rid of locks held by nodes that are gone. */
5437 void dlm_recover_purge(struct dlm_ls *ls)
5440 struct dlm_member *memb;
5441 int nodes_count = 0;
5442 int nodeid_gone = 0;
5443 unsigned int lkb_count = 0;
5445 /* cache one removed nodeid to optimize the common
5446 case of a single node removed */
5448 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5450 nodeid_gone = memb->nodeid;
5456 down_write(&ls->ls_root_sem);
5457 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5461 purge_dead_list(ls, r, &r->res_grantqueue,
5462 nodeid_gone, &lkb_count);
5463 purge_dead_list(ls, r, &r->res_convertqueue,
5464 nodeid_gone, &lkb_count);
5465 purge_dead_list(ls, r, &r->res_waitqueue,
5466 nodeid_gone, &lkb_count);
5472 up_write(&ls->ls_root_sem);
5475 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5476 lkb_count, nodes_count);
5479 static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5484 spin_lock(&ls->ls_rsbtbl[bucket].lock);
5485 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5486 r = rb_entry(n, struct dlm_rsb, res_hashnode);
5488 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5490 if (!is_master(r)) {
5491 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5495 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5498 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5503 * Attempt to grant locks on resources that we are the master of.
5504 * Locks may have become grantable during recovery because locks
5505 * from departed nodes have been purged (or not rebuilt), allowing
5506 * previously blocked locks to now be granted. The subset of rsb's
5507 * we are interested in are those with lkb's on either the convert or
5510 * Simplest would be to go through each master rsb and check for non-empty
5511 * convert or waiting queues, and attempt to grant on those rsbs.
5512 * Checking the queues requires lock_rsb, though, for which we'd need
5513 * to release the rsbtbl lock. This would make iterating through all
5514 * rsb's very inefficient. So, we rely on earlier recovery routines
5515 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5519 void dlm_recover_grant(struct dlm_ls *ls)
5523 unsigned int count = 0;
5524 unsigned int rsb_count = 0;
5525 unsigned int lkb_count = 0;
5528 r = find_grant_rsb(ls, bucket);
5530 if (bucket == ls->ls_rsbtbl_size - 1)
5538 /* the RECOVER_GRANT flag is checked in the grant path */
5539 grant_pending_locks(r, &count);
5540 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5542 confirm_master(r, 0);
5549 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5550 lkb_count, rsb_count);
5553 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5556 struct dlm_lkb *lkb;
5558 list_for_each_entry(lkb, head, lkb_statequeue) {
5559 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5565 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5568 struct dlm_lkb *lkb;
5570 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5573 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5576 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5582 /* needs at least dlm_rcom + rcom_lock */
5583 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5584 struct dlm_rsb *r, struct dlm_rcom *rc)
5586 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5588 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
5589 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5590 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5591 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5592 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5593 lkb->lkb_flags |= DLM_IFL_MSTCPY;
5594 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5595 lkb->lkb_rqmode = rl->rl_rqmode;
5596 lkb->lkb_grmode = rl->rl_grmode;
5597 /* don't set lkb_status because add_lkb wants to itself */
5599 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5600 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5602 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5603 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
5604 sizeof(struct rcom_lock);
5605 if (lvblen > ls->ls_lvblen)
5607 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5608 if (!lkb->lkb_lvbptr)
5610 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5613 /* Conversions between PR and CW (middle modes) need special handling.
5614 The real granted mode of these converting locks cannot be determined
5615 until all locks have been rebuilt on the rsb (recover_conversion) */
5617 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5618 middle_conversion(lkb)) {
5619 rl->rl_status = DLM_LKSTS_CONVERT;
5620 lkb->lkb_grmode = DLM_LOCK_IV;
5621 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5627 /* This lkb may have been recovered in a previous aborted recovery so we need
5628 to check if the rsb already has an lkb with the given remote nodeid/lkid.
5629 If so we just send back a standard reply. If not, we create a new lkb with
5630 the given values and send back our lkid. We send back our lkid by sending
5631 back the rcom_lock struct we got but with the remid field filled in. */
5633 /* needs at least dlm_rcom + rcom_lock */
5634 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5636 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5638 struct dlm_lkb *lkb;
5640 int from_nodeid = rc->rc_header.h_nodeid;
5643 if (rl->rl_parent_lkid) {
5644 error = -EOPNOTSUPP;
5648 remid = le32_to_cpu(rl->rl_lkid);
5650 /* In general we expect the rsb returned to be R_MASTER, but we don't
5651 have to require it. Recovery of masters on one node can overlap
5652 recovery of locks on another node, so one node can send us MSTCPY
5653 locks before we've made ourselves master of this rsb. We can still
5654 add new MSTCPY locks that we receive here without any harm; when
5655 we make ourselves master, dlm_recover_masters() won't touch the
5656 MSTCPY locks we've received early. */
5658 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5659 from_nodeid, R_RECEIVE_RECOVER, &r);
5665 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5666 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5667 from_nodeid, remid);
5672 lkb = search_remid(r, from_nodeid, remid);
5678 error = create_lkb(ls, &lkb);
5682 error = receive_rcom_lock_args(ls, lkb, r, rc);
5689 add_lkb(r, lkb, rl->rl_status);
5691 ls->ls_recover_locks_in++;
5693 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5694 rsb_set_flag(r, RSB_RECOVER_GRANT);
5697 /* this is the new value returned to the lock holder for
5698 saving in its process-copy lkb */
5699 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5701 lkb->lkb_recover_seq = ls->ls_recover_seq;
5707 if (error && error != -EEXIST)
5708 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5709 from_nodeid, remid, error);
5710 rl->rl_result = cpu_to_le32(error);
5714 /* needs at least dlm_rcom + rcom_lock */
5715 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5717 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5719 struct dlm_lkb *lkb;
5720 uint32_t lkid, remid;
5723 lkid = le32_to_cpu(rl->rl_lkid);
5724 remid = le32_to_cpu(rl->rl_remid);
5725 result = le32_to_cpu(rl->rl_result);
5727 error = find_lkb(ls, lkid, &lkb);
5729 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5730 lkid, rc->rc_header.h_nodeid, remid, result);
5734 r = lkb->lkb_resource;
5738 if (!is_process_copy(lkb)) {
5739 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5740 lkid, rc->rc_header.h_nodeid, remid, result);
5750 /* There's a chance the new master received our lock before
5751 dlm_recover_master_reply(), this wouldn't happen if we did
5752 a barrier between recover_masters and recover_locks. */
5754 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5755 lkid, rc->rc_header.h_nodeid, remid, result);
5757 dlm_send_rcom_lock(r, lkb);
5761 lkb->lkb_remid = remid;
5764 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5765 lkid, rc->rc_header.h_nodeid, remid, result);
5768 /* an ack for dlm_recover_locks() which waits for replies from
5769 all the locks it sends to new masters */
5770 dlm_recovered_lock(r);
5779 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5780 int mode, uint32_t flags, void *name, unsigned int namelen,
5781 unsigned long timeout_cs)
5783 struct dlm_lkb *lkb;
5784 struct dlm_args args;
5787 dlm_lock_recovery(ls);
5789 error = create_lkb(ls, &lkb);
5795 if (flags & DLM_LKF_VALBLK) {
5796 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5797 if (!ua->lksb.sb_lvbptr) {
5804 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5805 fake_astfn, ua, fake_bastfn, &args);
5807 kfree(ua->lksb.sb_lvbptr);
5808 ua->lksb.sb_lvbptr = NULL;
5814 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5815 When DLM_IFL_USER is set, the dlm knows that this is a userspace
5816 lock and that lkb_astparam is the dlm_user_args structure. */
5817 lkb->lkb_flags |= DLM_IFL_USER;
5818 error = request_lock(ls, lkb, name, namelen, &args);
5834 /* add this new lkb to the per-process list of locks */
5835 spin_lock(&ua->proc->locks_spin);
5837 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5838 spin_unlock(&ua->proc->locks_spin);
5840 dlm_unlock_recovery(ls);
5844 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5845 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5846 unsigned long timeout_cs)
5848 struct dlm_lkb *lkb;
5849 struct dlm_args args;
5850 struct dlm_user_args *ua;
5853 dlm_lock_recovery(ls);
5855 error = find_lkb(ls, lkid, &lkb);
5859 /* user can change the params on its lock when it converts it, or
5860 add an lvb that didn't exist before */
5864 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5865 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5866 if (!ua->lksb.sb_lvbptr) {
5871 if (lvb_in && ua->lksb.sb_lvbptr)
5872 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5874 ua->xid = ua_tmp->xid;
5875 ua->castparam = ua_tmp->castparam;
5876 ua->castaddr = ua_tmp->castaddr;
5877 ua->bastparam = ua_tmp->bastparam;
5878 ua->bastaddr = ua_tmp->bastaddr;
5879 ua->user_lksb = ua_tmp->user_lksb;
5881 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5882 fake_astfn, ua, fake_bastfn, &args);
5886 error = convert_lock(ls, lkb, &args);
5888 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5893 dlm_unlock_recovery(ls);
5899 * The caller asks for an orphan lock on a given resource with a given mode.
5900 * If a matching lock exists, it's moved to the owner's list of locks and
5901 * the lkid is returned.
5904 int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5905 int mode, uint32_t flags, void *name, unsigned int namelen,
5906 unsigned long timeout_cs, uint32_t *lkid)
5908 struct dlm_lkb *lkb;
5909 struct dlm_user_args *ua;
5910 int found_other_mode = 0;
5914 mutex_lock(&ls->ls_orphans_mutex);
5915 list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) {
5916 if (lkb->lkb_resource->res_length != namelen)
5918 if (memcmp(lkb->lkb_resource->res_name, name, namelen))
5920 if (lkb->lkb_grmode != mode) {
5921 found_other_mode = 1;
5926 list_del_init(&lkb->lkb_ownqueue);
5927 lkb->lkb_flags &= ~DLM_IFL_ORPHAN;
5928 *lkid = lkb->lkb_id;
5931 mutex_unlock(&ls->ls_orphans_mutex);
5933 if (!found && found_other_mode) {
5943 lkb->lkb_exflags = flags;
5944 lkb->lkb_ownpid = (int) current->pid;
5948 ua->proc = ua_tmp->proc;
5949 ua->xid = ua_tmp->xid;
5950 ua->castparam = ua_tmp->castparam;
5951 ua->castaddr = ua_tmp->castaddr;
5952 ua->bastparam = ua_tmp->bastparam;
5953 ua->bastaddr = ua_tmp->bastaddr;
5954 ua->user_lksb = ua_tmp->user_lksb;
5957 * The lkb reference from the ls_orphans list was not
5958 * removed above, and is now considered the reference
5959 * for the proc locks list.
5962 spin_lock(&ua->proc->locks_spin);
5963 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5964 spin_unlock(&ua->proc->locks_spin);
5970 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5971 uint32_t flags, uint32_t lkid, char *lvb_in)
5973 struct dlm_lkb *lkb;
5974 struct dlm_args args;
5975 struct dlm_user_args *ua;
5978 dlm_lock_recovery(ls);
5980 error = find_lkb(ls, lkid, &lkb);
5986 if (lvb_in && ua->lksb.sb_lvbptr)
5987 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5988 if (ua_tmp->castparam)
5989 ua->castparam = ua_tmp->castparam;
5990 ua->user_lksb = ua_tmp->user_lksb;
5992 error = set_unlock_args(flags, ua, &args);
5996 error = unlock_lock(ls, lkb, &args);
5998 if (error == -DLM_EUNLOCK)
6000 /* from validate_unlock_args() */
6001 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
6006 spin_lock(&ua->proc->locks_spin);
6007 /* dlm_user_add_cb() may have already taken lkb off the proc list */
6008 if (!list_empty(&lkb->lkb_ownqueue))
6009 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
6010 spin_unlock(&ua->proc->locks_spin);
6014 dlm_unlock_recovery(ls);
6019 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6020 uint32_t flags, uint32_t lkid)
6022 struct dlm_lkb *lkb;
6023 struct dlm_args args;
6024 struct dlm_user_args *ua;
6027 dlm_lock_recovery(ls);
6029 error = find_lkb(ls, lkid, &lkb);
6034 if (ua_tmp->castparam)
6035 ua->castparam = ua_tmp->castparam;
6036 ua->user_lksb = ua_tmp->user_lksb;
6038 error = set_unlock_args(flags, ua, &args);
6042 error = cancel_lock(ls, lkb, &args);
6044 if (error == -DLM_ECANCEL)
6046 /* from validate_unlock_args() */
6047 if (error == -EBUSY)
6052 dlm_unlock_recovery(ls);
6057 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6059 struct dlm_lkb *lkb;
6060 struct dlm_args args;
6061 struct dlm_user_args *ua;
6065 dlm_lock_recovery(ls);
6067 error = find_lkb(ls, lkid, &lkb);
6073 error = set_unlock_args(flags, ua, &args);
6077 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6079 r = lkb->lkb_resource;
6083 error = validate_unlock_args(lkb, &args);
6086 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6088 error = _cancel_lock(r, lkb);
6093 if (error == -DLM_ECANCEL)
6095 /* from validate_unlock_args() */
6096 if (error == -EBUSY)
6101 dlm_unlock_recovery(ls);
6105 /* lkb's that are removed from the waiters list by revert are just left on the
6106 orphans list with the granted orphan locks, to be freed by purge */
6108 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6110 struct dlm_args args;
6113 hold_lkb(lkb); /* reference for the ls_orphans list */
6114 mutex_lock(&ls->ls_orphans_mutex);
6115 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6116 mutex_unlock(&ls->ls_orphans_mutex);
6118 set_unlock_args(0, lkb->lkb_ua, &args);
6120 error = cancel_lock(ls, lkb, &args);
6121 if (error == -DLM_ECANCEL)
6126 /* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6127 granted. Regardless of what rsb queue the lock is on, it's removed and
6128 freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6129 if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6131 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6133 struct dlm_args args;
6136 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6137 lkb->lkb_ua, &args);
6139 error = unlock_lock(ls, lkb, &args);
6140 if (error == -DLM_EUNLOCK)
6145 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6146 (which does lock_rsb) due to deadlock with receiving a message that does
6147 lock_rsb followed by dlm_user_add_cb() */
6149 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6150 struct dlm_user_proc *proc)
6152 struct dlm_lkb *lkb = NULL;
6154 mutex_lock(&ls->ls_clear_proc_locks);
6155 if (list_empty(&proc->locks))
6158 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6159 list_del_init(&lkb->lkb_ownqueue);
6161 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6162 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6164 lkb->lkb_flags |= DLM_IFL_DEAD;
6166 mutex_unlock(&ls->ls_clear_proc_locks);
6170 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6171 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6172 which we clear here. */
6174 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
6175 list, and no more device_writes should add lkb's to proc->locks list; so we
6176 shouldn't need to take asts_spin or locks_spin here. this assumes that
6177 device reads/writes/closes are serialized -- FIXME: we may need to serialize
6180 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6182 struct dlm_lkb *lkb, *safe;
6184 dlm_lock_recovery(ls);
6187 lkb = del_proc_lock(ls, proc);
6191 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6192 orphan_proc_lock(ls, lkb);
6194 unlock_proc_lock(ls, lkb);
6196 /* this removes the reference for the proc->locks list
6197 added by dlm_user_request, it may result in the lkb
6203 mutex_lock(&ls->ls_clear_proc_locks);
6205 /* in-progress unlocks */
6206 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6207 list_del_init(&lkb->lkb_ownqueue);
6208 lkb->lkb_flags |= DLM_IFL_DEAD;
6212 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6213 memset(&lkb->lkb_callbacks, 0,
6214 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6215 list_del_init(&lkb->lkb_cb_list);
6219 mutex_unlock(&ls->ls_clear_proc_locks);
6220 dlm_unlock_recovery(ls);
6223 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6225 struct dlm_lkb *lkb, *safe;
6229 spin_lock(&proc->locks_spin);
6230 if (!list_empty(&proc->locks)) {
6231 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6233 list_del_init(&lkb->lkb_ownqueue);
6235 spin_unlock(&proc->locks_spin);
6240 lkb->lkb_flags |= DLM_IFL_DEAD;
6241 unlock_proc_lock(ls, lkb);
6242 dlm_put_lkb(lkb); /* ref from proc->locks list */
6245 spin_lock(&proc->locks_spin);
6246 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6247 list_del_init(&lkb->lkb_ownqueue);
6248 lkb->lkb_flags |= DLM_IFL_DEAD;
6251 spin_unlock(&proc->locks_spin);
6253 spin_lock(&proc->asts_spin);
6254 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6255 memset(&lkb->lkb_callbacks, 0,
6256 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6257 list_del_init(&lkb->lkb_cb_list);
6260 spin_unlock(&proc->asts_spin);
6263 /* pid of 0 means purge all orphans */
6265 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6267 struct dlm_lkb *lkb, *safe;
6269 mutex_lock(&ls->ls_orphans_mutex);
6270 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6271 if (pid && lkb->lkb_ownpid != pid)
6273 unlock_proc_lock(ls, lkb);
6274 list_del_init(&lkb->lkb_ownqueue);
6277 mutex_unlock(&ls->ls_orphans_mutex);
6280 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6282 struct dlm_message *ms;
6283 struct dlm_mhandle *mh;
6286 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6287 DLM_MSG_PURGE, &ms, &mh);
6290 ms->m_nodeid = nodeid;
6293 return send_message(mh, ms);
6296 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6297 int nodeid, int pid)
6301 if (nodeid && (nodeid != dlm_our_nodeid())) {
6302 error = send_purge(ls, nodeid, pid);
6304 dlm_lock_recovery(ls);
6305 if (pid == current->pid)
6306 purge_proc_locks(ls, proc);
6308 do_purge(ls, nodeid, pid);
6309 dlm_unlock_recovery(ls);