1 // SPDX-License-Identifier: GPL-2.0-only
2 /******************************************************************************
3 *******************************************************************************
5 ** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved.
8 *******************************************************************************
9 ******************************************************************************/
11 /* Central locking logic has four stages:
31 Stage 1 (lock, unlock) is mainly about checking input args and
32 splitting into one of the four main operations:
34 dlm_lock = request_lock
35 dlm_lock+CONVERT = convert_lock
36 dlm_unlock = unlock_lock
37 dlm_unlock+CANCEL = cancel_lock
39 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
40 provided to the next stage.
42 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
43 When remote, it calls send_xxxx(), when local it calls do_xxxx().
45 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
46 given rsb and lkb and queues callbacks.
48 For remote operations, send_xxxx() results in the corresponding do_xxxx()
49 function being executed on the remote node. The connecting send/receive
50 calls on local (L) and remote (R) nodes:
52 L: send_xxxx() -> R: receive_xxxx()
54 L: receive_xxxx_reply() <- R: send_xxxx_reply()
56 #include <linux/types.h>
57 #include <linux/rbtree.h>
58 #include <linux/slab.h>
59 #include "dlm_internal.h"
60 #include <linux/dlm_device.h>
63 #include "requestqueue.h"
67 #include "lockspace.h"
72 #include "lvb_table.h"
76 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
77 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
78 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
79 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
80 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
81 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
82 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
83 static int send_remove(struct dlm_rsb *r);
84 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
85 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
86 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
87 struct dlm_message *ms);
88 static int receive_extralen(struct dlm_message *ms);
89 static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
90 static void del_timeout(struct dlm_lkb *lkb);
91 static void toss_rsb(struct kref *kref);
94 * Lock compatibilty matrix - thanks Steve
95 * UN = Unlocked state. Not really a state, used as a flag
96 * PD = Padding. Used to make the matrix a nice power of two in size
97 * Other states are the same as the VMS DLM.
98 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
101 static const int __dlm_compat_matrix[8][8] = {
102 /* UN NL CR CW PR PW EX PD */
103 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
104 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
105 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
106 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
107 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
108 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
109 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
110 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
114 * This defines the direction of transfer of LVB data.
115 * Granted mode is the row; requested mode is the column.
116 * Usage: matrix[grmode+1][rqmode+1]
117 * 1 = LVB is returned to the caller
118 * 0 = LVB is written to the resource
119 * -1 = nothing happens to the LVB
122 const int dlm_lvb_operations[8][8] = {
123 /* UN NL CR CW PR PW EX PD*/
124 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
125 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
126 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
127 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
128 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
129 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
130 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
131 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
134 #define modes_compat(gr, rq) \
135 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
137 int dlm_modes_compat(int mode1, int mode2)
139 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
143 * Compatibility matrix for conversions with QUECVT set.
144 * Granted mode is the row; requested mode is the column.
145 * Usage: matrix[grmode+1][rqmode+1]
148 static const int __quecvt_compat_matrix[8][8] = {
149 /* UN NL CR CW PR PW EX PD */
150 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
151 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
152 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
153 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
154 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
155 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
156 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
157 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
160 void dlm_print_lkb(struct dlm_lkb *lkb)
162 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
163 "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
164 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
165 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
166 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
167 (unsigned long long)lkb->lkb_recover_seq);
170 static void dlm_print_rsb(struct dlm_rsb *r)
172 printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
174 r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
175 r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
179 void dlm_dump_rsb(struct dlm_rsb *r)
185 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
186 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
187 printk(KERN_ERR "rsb lookup list\n");
188 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
190 printk(KERN_ERR "rsb grant queue:\n");
191 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
193 printk(KERN_ERR "rsb convert queue:\n");
194 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
196 printk(KERN_ERR "rsb wait queue:\n");
197 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
201 /* Threads cannot use the lockspace while it's being recovered */
203 static inline void dlm_lock_recovery(struct dlm_ls *ls)
205 down_read(&ls->ls_in_recovery);
208 void dlm_unlock_recovery(struct dlm_ls *ls)
210 up_read(&ls->ls_in_recovery);
213 int dlm_lock_recovery_try(struct dlm_ls *ls)
215 return down_read_trylock(&ls->ls_in_recovery);
218 static inline int can_be_queued(struct dlm_lkb *lkb)
220 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
223 static inline int force_blocking_asts(struct dlm_lkb *lkb)
225 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
228 static inline int is_demoted(struct dlm_lkb *lkb)
230 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
233 static inline int is_altmode(struct dlm_lkb *lkb)
235 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
238 static inline int is_granted(struct dlm_lkb *lkb)
240 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
243 static inline int is_remote(struct dlm_rsb *r)
245 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
246 return !!r->res_nodeid;
249 static inline int is_process_copy(struct dlm_lkb *lkb)
251 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
254 static inline int is_master_copy(struct dlm_lkb *lkb)
256 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
259 static inline int middle_conversion(struct dlm_lkb *lkb)
261 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
262 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
267 static inline int down_conversion(struct dlm_lkb *lkb)
269 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
272 static inline int is_overlap_unlock(struct dlm_lkb *lkb)
274 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
277 static inline int is_overlap_cancel(struct dlm_lkb *lkb)
279 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
282 static inline int is_overlap(struct dlm_lkb *lkb)
284 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
285 DLM_IFL_OVERLAP_CANCEL));
288 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
290 if (is_master_copy(lkb))
295 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
297 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
298 timeout caused the cancel then return -ETIMEDOUT */
299 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
300 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
304 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
305 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
309 dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
312 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
315 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
318 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
320 if (is_master_copy(lkb)) {
321 send_bast(r, lkb, rqmode);
323 dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
328 * Basic operations on rsb's and lkb's
331 /* This is only called to add a reference when the code already holds
332 a valid reference to the rsb, so there's no need for locking. */
334 static inline void hold_rsb(struct dlm_rsb *r)
336 kref_get(&r->res_ref);
339 void dlm_hold_rsb(struct dlm_rsb *r)
344 /* When all references to the rsb are gone it's transferred to
345 the tossed list for later disposal. */
347 static void put_rsb(struct dlm_rsb *r)
349 struct dlm_ls *ls = r->res_ls;
350 uint32_t bucket = r->res_bucket;
352 spin_lock(&ls->ls_rsbtbl[bucket].lock);
353 kref_put(&r->res_ref, toss_rsb);
354 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
357 void dlm_put_rsb(struct dlm_rsb *r)
362 static int pre_rsb_struct(struct dlm_ls *ls)
364 struct dlm_rsb *r1, *r2;
367 spin_lock(&ls->ls_new_rsb_spin);
368 if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
369 spin_unlock(&ls->ls_new_rsb_spin);
372 spin_unlock(&ls->ls_new_rsb_spin);
374 r1 = dlm_allocate_rsb(ls);
375 r2 = dlm_allocate_rsb(ls);
377 spin_lock(&ls->ls_new_rsb_spin);
379 list_add(&r1->res_hashchain, &ls->ls_new_rsb);
380 ls->ls_new_rsb_count++;
383 list_add(&r2->res_hashchain, &ls->ls_new_rsb);
384 ls->ls_new_rsb_count++;
386 count = ls->ls_new_rsb_count;
387 spin_unlock(&ls->ls_new_rsb_spin);
394 /* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
395 unlock any spinlocks, go back and call pre_rsb_struct again.
396 Otherwise, take an rsb off the list and return it. */
398 static int get_rsb_struct(struct dlm_ls *ls, char *name, int len,
399 struct dlm_rsb **r_ret)
404 spin_lock(&ls->ls_new_rsb_spin);
405 if (list_empty(&ls->ls_new_rsb)) {
406 count = ls->ls_new_rsb_count;
407 spin_unlock(&ls->ls_new_rsb_spin);
408 log_debug(ls, "find_rsb retry %d %d %s",
409 count, dlm_config.ci_new_rsb_count, name);
413 r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
414 list_del(&r->res_hashchain);
415 /* Convert the empty list_head to a NULL rb_node for tree usage: */
416 memset(&r->res_hashnode, 0, sizeof(struct rb_node));
417 ls->ls_new_rsb_count--;
418 spin_unlock(&ls->ls_new_rsb_spin);
422 memcpy(r->res_name, name, len);
423 mutex_init(&r->res_mutex);
425 INIT_LIST_HEAD(&r->res_lookup);
426 INIT_LIST_HEAD(&r->res_grantqueue);
427 INIT_LIST_HEAD(&r->res_convertqueue);
428 INIT_LIST_HEAD(&r->res_waitqueue);
429 INIT_LIST_HEAD(&r->res_root_list);
430 INIT_LIST_HEAD(&r->res_recover_list);
436 static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
438 char maxname[DLM_RESNAME_MAXLEN];
440 memset(maxname, 0, DLM_RESNAME_MAXLEN);
441 memcpy(maxname, name, nlen);
442 return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
445 int dlm_search_rsb_tree(struct rb_root *tree, char *name, int len,
446 struct dlm_rsb **r_ret)
448 struct rb_node *node = tree->rb_node;
453 r = rb_entry(node, struct dlm_rsb, res_hashnode);
454 rc = rsb_cmp(r, name, len);
456 node = node->rb_left;
458 node = node->rb_right;
470 static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
472 struct rb_node **newn = &tree->rb_node;
473 struct rb_node *parent = NULL;
477 struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
481 rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
483 newn = &parent->rb_left;
485 newn = &parent->rb_right;
487 log_print("rsb_insert match");
494 rb_link_node(&rsb->res_hashnode, parent, newn);
495 rb_insert_color(&rsb->res_hashnode, tree);
500 * Find rsb in rsbtbl and potentially create/add one
502 * Delaying the release of rsb's has a similar benefit to applications keeping
503 * NL locks on an rsb, but without the guarantee that the cached master value
504 * will still be valid when the rsb is reused. Apps aren't always smart enough
505 * to keep NL locks on an rsb that they may lock again shortly; this can lead
506 * to excessive master lookups and removals if we don't delay the release.
508 * Searching for an rsb means looking through both the normal list and toss
509 * list. When found on the toss list the rsb is moved to the normal list with
510 * ref count of 1; when found on normal list the ref count is incremented.
512 * rsb's on the keep list are being used locally and refcounted.
513 * rsb's on the toss list are not being used locally, and are not refcounted.
515 * The toss list rsb's were either
516 * - previously used locally but not any more (were on keep list, then
517 * moved to toss list when last refcount dropped)
518 * - created and put on toss list as a directory record for a lookup
519 * (we are the dir node for the res, but are not using the res right now,
520 * but some other node is)
522 * The purpose of find_rsb() is to return a refcounted rsb for local use.
523 * So, if the given rsb is on the toss list, it is moved to the keep list
524 * before being returned.
526 * toss_rsb() happens when all local usage of the rsb is done, i.e. no
527 * more refcounts exist, so the rsb is moved from the keep list to the
530 * rsb's on both keep and toss lists are used for doing a name to master
531 * lookups. rsb's that are in use locally (and being refcounted) are on
532 * the keep list, rsb's that are not in use locally (not refcounted) and
533 * only exist for name/master lookups are on the toss list.
535 * rsb's on the toss list who's dir_nodeid is not local can have stale
536 * name/master mappings. So, remote requests on such rsb's can potentially
537 * return with an error, which means the mapping is stale and needs to
538 * be updated with a new lookup. (The idea behind MASTER UNCERTAIN and
539 * first_lkid is to keep only a single outstanding request on an rsb
540 * while that rsb has a potentially stale master.)
543 static int find_rsb_dir(struct dlm_ls *ls, char *name, int len,
544 uint32_t hash, uint32_t b,
545 int dir_nodeid, int from_nodeid,
546 unsigned int flags, struct dlm_rsb **r_ret)
548 struct dlm_rsb *r = NULL;
549 int our_nodeid = dlm_our_nodeid();
556 if (flags & R_RECEIVE_REQUEST) {
557 if (from_nodeid == dir_nodeid)
561 } else if (flags & R_REQUEST) {
566 * flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
567 * from_nodeid has sent us a lock in dlm_recover_locks, believing
568 * we're the new master. Our local recovery may not have set
569 * res_master_nodeid to our_nodeid yet, so allow either. Don't
570 * create the rsb; dlm_recover_process_copy() will handle EBADR
573 * If someone sends us a request, we are the dir node, and we do
574 * not find the rsb anywhere, then recreate it. This happens if
575 * someone sends us a request after we have removed/freed an rsb
576 * from our toss list. (They sent a request instead of lookup
577 * because they are using an rsb from their toss list.)
580 if (from_local || from_dir ||
581 (from_other && (dir_nodeid == our_nodeid))) {
587 error = pre_rsb_struct(ls);
592 spin_lock(&ls->ls_rsbtbl[b].lock);
594 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
599 * rsb is active, so we can't check master_nodeid without lock_rsb.
602 kref_get(&r->res_ref);
608 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
613 * rsb found inactive (master_nodeid may be out of date unless
614 * we are the dir_nodeid or were the master) No other thread
615 * is using this rsb because it's on the toss list, so we can
616 * look at or update res_master_nodeid without lock_rsb.
619 if ((r->res_master_nodeid != our_nodeid) && from_other) {
620 /* our rsb was not master, and another node (not the dir node)
621 has sent us a request */
622 log_debug(ls, "find_rsb toss from_other %d master %d dir %d %s",
623 from_nodeid, r->res_master_nodeid, dir_nodeid,
629 if ((r->res_master_nodeid != our_nodeid) && from_dir) {
630 /* don't think this should ever happen */
631 log_error(ls, "find_rsb toss from_dir %d master %d",
632 from_nodeid, r->res_master_nodeid);
634 /* fix it and go on */
635 r->res_master_nodeid = our_nodeid;
637 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
638 r->res_first_lkid = 0;
641 if (from_local && (r->res_master_nodeid != our_nodeid)) {
642 /* Because we have held no locks on this rsb,
643 res_master_nodeid could have become stale. */
644 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
645 r->res_first_lkid = 0;
648 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
649 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
658 if (error == -EBADR && !create)
661 error = get_rsb_struct(ls, name, len, &r);
662 if (error == -EAGAIN) {
663 spin_unlock(&ls->ls_rsbtbl[b].lock);
671 r->res_dir_nodeid = dir_nodeid;
672 kref_init(&r->res_ref);
675 /* want to see how often this happens */
676 log_debug(ls, "find_rsb new from_dir %d recreate %s",
677 from_nodeid, r->res_name);
678 r->res_master_nodeid = our_nodeid;
683 if (from_other && (dir_nodeid != our_nodeid)) {
684 /* should never happen */
685 log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
686 from_nodeid, dir_nodeid, our_nodeid, r->res_name);
694 log_debug(ls, "find_rsb new from_other %d dir %d %s",
695 from_nodeid, dir_nodeid, r->res_name);
698 if (dir_nodeid == our_nodeid) {
699 /* When we are the dir nodeid, we can set the master
701 r->res_master_nodeid = our_nodeid;
704 /* set_master will send_lookup to dir_nodeid */
705 r->res_master_nodeid = 0;
710 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
712 spin_unlock(&ls->ls_rsbtbl[b].lock);
718 /* During recovery, other nodes can send us new MSTCPY locks (from
719 dlm_recover_locks) before we've made ourself master (in
720 dlm_recover_masters). */
722 static int find_rsb_nodir(struct dlm_ls *ls, char *name, int len,
723 uint32_t hash, uint32_t b,
724 int dir_nodeid, int from_nodeid,
725 unsigned int flags, struct dlm_rsb **r_ret)
727 struct dlm_rsb *r = NULL;
728 int our_nodeid = dlm_our_nodeid();
729 int recover = (flags & R_RECEIVE_RECOVER);
733 error = pre_rsb_struct(ls);
737 spin_lock(&ls->ls_rsbtbl[b].lock);
739 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
744 * rsb is active, so we can't check master_nodeid without lock_rsb.
747 kref_get(&r->res_ref);
752 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
757 * rsb found inactive. No other thread is using this rsb because
758 * it's on the toss list, so we can look at or update
759 * res_master_nodeid without lock_rsb.
762 if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
763 /* our rsb is not master, and another node has sent us a
764 request; this should never happen */
765 log_error(ls, "find_rsb toss from_nodeid %d master %d dir %d",
766 from_nodeid, r->res_master_nodeid, dir_nodeid);
772 if (!recover && (r->res_master_nodeid != our_nodeid) &&
773 (dir_nodeid == our_nodeid)) {
774 /* our rsb is not master, and we are dir; may as well fix it;
775 this should never happen */
776 log_error(ls, "find_rsb toss our %d master %d dir %d",
777 our_nodeid, r->res_master_nodeid, dir_nodeid);
779 r->res_master_nodeid = our_nodeid;
783 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
784 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
793 error = get_rsb_struct(ls, name, len, &r);
794 if (error == -EAGAIN) {
795 spin_unlock(&ls->ls_rsbtbl[b].lock);
803 r->res_dir_nodeid = dir_nodeid;
804 r->res_master_nodeid = dir_nodeid;
805 r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
806 kref_init(&r->res_ref);
808 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
810 spin_unlock(&ls->ls_rsbtbl[b].lock);
816 static int find_rsb(struct dlm_ls *ls, char *name, int len, int from_nodeid,
817 unsigned int flags, struct dlm_rsb **r_ret)
822 if (len > DLM_RESNAME_MAXLEN)
825 hash = jhash(name, len, 0);
826 b = hash & (ls->ls_rsbtbl_size - 1);
828 dir_nodeid = dlm_hash2nodeid(ls, hash);
830 if (dlm_no_directory(ls))
831 return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
832 from_nodeid, flags, r_ret);
834 return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
835 from_nodeid, flags, r_ret);
838 /* we have received a request and found that res_master_nodeid != our_nodeid,
839 so we need to return an error or make ourself the master */
841 static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
844 if (dlm_no_directory(ls)) {
845 log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
846 from_nodeid, r->res_master_nodeid,
852 if (from_nodeid != r->res_dir_nodeid) {
853 /* our rsb is not master, and another node (not the dir node)
854 has sent us a request. this is much more common when our
855 master_nodeid is zero, so limit debug to non-zero. */
857 if (r->res_master_nodeid) {
858 log_debug(ls, "validate master from_other %d master %d "
859 "dir %d first %x %s", from_nodeid,
860 r->res_master_nodeid, r->res_dir_nodeid,
861 r->res_first_lkid, r->res_name);
865 /* our rsb is not master, but the dir nodeid has sent us a
866 request; this could happen with master 0 / res_nodeid -1 */
868 if (r->res_master_nodeid) {
869 log_error(ls, "validate master from_dir %d master %d "
871 from_nodeid, r->res_master_nodeid,
872 r->res_first_lkid, r->res_name);
875 r->res_master_nodeid = dlm_our_nodeid();
882 * We're the dir node for this res and another node wants to know the
883 * master nodeid. During normal operation (non recovery) this is only
884 * called from receive_lookup(); master lookups when the local node is
885 * the dir node are done by find_rsb().
887 * normal operation, we are the dir node for a resource
892 * . dlm_master_lookup flags 0
894 * recover directory, we are rebuilding dir for all resources
895 * . dlm_recover_directory
897 * remote node sends back the rsb names it is master of and we are dir of
898 * . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
899 * we either create new rsb setting remote node as master, or find existing
900 * rsb and set master to be the remote node.
902 * recover masters, we are finding the new master for resources
903 * . dlm_recover_masters
905 * . dlm_send_rcom_lookup
906 * . receive_rcom_lookup
907 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
910 int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
911 unsigned int flags, int *r_nodeid, int *result)
913 struct dlm_rsb *r = NULL;
915 int from_master = (flags & DLM_LU_RECOVER_DIR);
916 int fix_master = (flags & DLM_LU_RECOVER_MASTER);
917 int our_nodeid = dlm_our_nodeid();
918 int dir_nodeid, error, toss_list = 0;
920 if (len > DLM_RESNAME_MAXLEN)
923 if (from_nodeid == our_nodeid) {
924 log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
929 hash = jhash(name, len, 0);
930 b = hash & (ls->ls_rsbtbl_size - 1);
932 dir_nodeid = dlm_hash2nodeid(ls, hash);
933 if (dir_nodeid != our_nodeid) {
934 log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
935 from_nodeid, dir_nodeid, our_nodeid, hash,
942 error = pre_rsb_struct(ls);
946 spin_lock(&ls->ls_rsbtbl[b].lock);
947 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
949 /* because the rsb is active, we need to lock_rsb before
950 checking/changing re_master_nodeid */
953 spin_unlock(&ls->ls_rsbtbl[b].lock);
958 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
962 /* because the rsb is inactive (on toss list), it's not refcounted
963 and lock_rsb is not used, but is protected by the rsbtbl lock */
967 if (r->res_dir_nodeid != our_nodeid) {
968 /* should not happen, but may as well fix it and carry on */
969 log_error(ls, "dlm_master_lookup res_dir %d our %d %s",
970 r->res_dir_nodeid, our_nodeid, r->res_name);
971 r->res_dir_nodeid = our_nodeid;
974 if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
975 /* Recovery uses this function to set a new master when
976 the previous master failed. Setting NEW_MASTER will
977 force dlm_recover_masters to call recover_master on this
978 rsb even though the res_nodeid is no longer removed. */
980 r->res_master_nodeid = from_nodeid;
981 r->res_nodeid = from_nodeid;
982 rsb_set_flag(r, RSB_NEW_MASTER);
985 /* I don't think we should ever find it on toss list. */
986 log_error(ls, "dlm_master_lookup fix_master on toss");
991 if (from_master && (r->res_master_nodeid != from_nodeid)) {
992 /* this will happen if from_nodeid became master during
993 a previous recovery cycle, and we aborted the previous
994 cycle before recovering this master value */
996 log_limit(ls, "dlm_master_lookup from_master %d "
997 "master_nodeid %d res_nodeid %d first %x %s",
998 from_nodeid, r->res_master_nodeid, r->res_nodeid,
999 r->res_first_lkid, r->res_name);
1001 if (r->res_master_nodeid == our_nodeid) {
1002 log_error(ls, "from_master %d our_master", from_nodeid);
1007 r->res_master_nodeid = from_nodeid;
1008 r->res_nodeid = from_nodeid;
1009 rsb_set_flag(r, RSB_NEW_MASTER);
1012 if (!r->res_master_nodeid) {
1013 /* this will happen if recovery happens while we're looking
1014 up the master for this rsb */
1016 log_debug(ls, "dlm_master_lookup master 0 to %d first %x %s",
1017 from_nodeid, r->res_first_lkid, r->res_name);
1018 r->res_master_nodeid = from_nodeid;
1019 r->res_nodeid = from_nodeid;
1022 if (!from_master && !fix_master &&
1023 (r->res_master_nodeid == from_nodeid)) {
1024 /* this can happen when the master sends remove, the dir node
1025 finds the rsb on the keep list and ignores the remove,
1026 and the former master sends a lookup */
1028 log_limit(ls, "dlm_master_lookup from master %d flags %x "
1029 "first %x %s", from_nodeid, flags,
1030 r->res_first_lkid, r->res_name);
1034 *r_nodeid = r->res_master_nodeid;
1036 *result = DLM_LU_MATCH;
1039 r->res_toss_time = jiffies;
1040 /* the rsb was inactive (on toss list) */
1041 spin_unlock(&ls->ls_rsbtbl[b].lock);
1043 /* the rsb was active */
1050 error = get_rsb_struct(ls, name, len, &r);
1051 if (error == -EAGAIN) {
1052 spin_unlock(&ls->ls_rsbtbl[b].lock);
1060 r->res_dir_nodeid = our_nodeid;
1061 r->res_master_nodeid = from_nodeid;
1062 r->res_nodeid = from_nodeid;
1063 kref_init(&r->res_ref);
1064 r->res_toss_time = jiffies;
1066 error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1068 /* should never happen */
1070 spin_unlock(&ls->ls_rsbtbl[b].lock);
1075 *result = DLM_LU_ADD;
1076 *r_nodeid = from_nodeid;
1079 spin_unlock(&ls->ls_rsbtbl[b].lock);
1083 static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1089 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1090 spin_lock(&ls->ls_rsbtbl[i].lock);
1091 for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1092 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1093 if (r->res_hash == hash)
1096 spin_unlock(&ls->ls_rsbtbl[i].lock);
1100 void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
1102 struct dlm_rsb *r = NULL;
1106 hash = jhash(name, len, 0);
1107 b = hash & (ls->ls_rsbtbl_size - 1);
1109 spin_lock(&ls->ls_rsbtbl[b].lock);
1110 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1114 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1120 spin_unlock(&ls->ls_rsbtbl[b].lock);
1123 static void toss_rsb(struct kref *kref)
1125 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1126 struct dlm_ls *ls = r->res_ls;
1128 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1129 kref_init(&r->res_ref);
1130 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1131 rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
1132 r->res_toss_time = jiffies;
1133 ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
1134 if (r->res_lvbptr) {
1135 dlm_free_lvb(r->res_lvbptr);
1136 r->res_lvbptr = NULL;
1140 /* See comment for unhold_lkb */
1142 static void unhold_rsb(struct dlm_rsb *r)
1145 rv = kref_put(&r->res_ref, toss_rsb);
1146 DLM_ASSERT(!rv, dlm_dump_rsb(r););
1149 static void kill_rsb(struct kref *kref)
1151 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1153 /* All work is done after the return from kref_put() so we
1154 can release the write_lock before the remove and free. */
1156 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1157 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1158 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1159 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1160 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1161 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1164 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1165 The rsb must exist as long as any lkb's for it do. */
1167 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1170 lkb->lkb_resource = r;
1173 static void detach_lkb(struct dlm_lkb *lkb)
1175 if (lkb->lkb_resource) {
1176 put_rsb(lkb->lkb_resource);
1177 lkb->lkb_resource = NULL;
1181 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1183 struct dlm_lkb *lkb;
1186 lkb = dlm_allocate_lkb(ls);
1190 lkb->lkb_nodeid = -1;
1191 lkb->lkb_grmode = DLM_LOCK_IV;
1192 kref_init(&lkb->lkb_ref);
1193 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1194 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1195 INIT_LIST_HEAD(&lkb->lkb_time_list);
1196 INIT_LIST_HEAD(&lkb->lkb_cb_list);
1197 mutex_init(&lkb->lkb_cb_mutex);
1198 INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
1200 idr_preload(GFP_NOFS);
1201 spin_lock(&ls->ls_lkbidr_spin);
1202 rv = idr_alloc(&ls->ls_lkbidr, lkb, 1, 0, GFP_NOWAIT);
1205 spin_unlock(&ls->ls_lkbidr_spin);
1209 log_error(ls, "create_lkb idr error %d", rv);
1218 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1220 struct dlm_lkb *lkb;
1222 spin_lock(&ls->ls_lkbidr_spin);
1223 lkb = idr_find(&ls->ls_lkbidr, lkid);
1225 kref_get(&lkb->lkb_ref);
1226 spin_unlock(&ls->ls_lkbidr_spin);
1229 return lkb ? 0 : -ENOENT;
1232 static void kill_lkb(struct kref *kref)
1234 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1236 /* All work is done after the return from kref_put() so we
1237 can release the write_lock before the detach_lkb */
1239 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1242 /* __put_lkb() is used when an lkb may not have an rsb attached to
1243 it so we need to provide the lockspace explicitly */
1245 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
1247 uint32_t lkid = lkb->lkb_id;
1249 spin_lock(&ls->ls_lkbidr_spin);
1250 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
1251 idr_remove(&ls->ls_lkbidr, lkid);
1252 spin_unlock(&ls->ls_lkbidr_spin);
1256 /* for local/process lkbs, lvbptr points to caller's lksb */
1257 if (lkb->lkb_lvbptr && is_master_copy(lkb))
1258 dlm_free_lvb(lkb->lkb_lvbptr);
1262 spin_unlock(&ls->ls_lkbidr_spin);
1267 int dlm_put_lkb(struct dlm_lkb *lkb)
1271 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1272 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1274 ls = lkb->lkb_resource->res_ls;
1275 return __put_lkb(ls, lkb);
1278 /* This is only called to add a reference when the code already holds
1279 a valid reference to the lkb, so there's no need for locking. */
1281 static inline void hold_lkb(struct dlm_lkb *lkb)
1283 kref_get(&lkb->lkb_ref);
1286 /* This is called when we need to remove a reference and are certain
1287 it's not the last ref. e.g. del_lkb is always called between a
1288 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1289 put_lkb would work fine, but would involve unnecessary locking */
1291 static inline void unhold_lkb(struct dlm_lkb *lkb)
1294 rv = kref_put(&lkb->lkb_ref, kill_lkb);
1295 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
1298 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1301 struct dlm_lkb *lkb = NULL;
1303 list_for_each_entry(lkb, head, lkb_statequeue)
1304 if (lkb->lkb_rqmode < mode)
1307 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
1310 /* add/remove lkb to rsb's grant/convert/wait queue */
1312 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1314 kref_get(&lkb->lkb_ref);
1316 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1318 lkb->lkb_timestamp = ktime_get();
1320 lkb->lkb_status = status;
1323 case DLM_LKSTS_WAITING:
1324 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1325 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1327 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1329 case DLM_LKSTS_GRANTED:
1330 /* convention says granted locks kept in order of grmode */
1331 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1334 case DLM_LKSTS_CONVERT:
1335 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1336 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1338 list_add_tail(&lkb->lkb_statequeue,
1339 &r->res_convertqueue);
1342 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1346 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1348 lkb->lkb_status = 0;
1349 list_del(&lkb->lkb_statequeue);
1353 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1357 add_lkb(r, lkb, sts);
1361 static int msg_reply_type(int mstype)
1364 case DLM_MSG_REQUEST:
1365 return DLM_MSG_REQUEST_REPLY;
1366 case DLM_MSG_CONVERT:
1367 return DLM_MSG_CONVERT_REPLY;
1368 case DLM_MSG_UNLOCK:
1369 return DLM_MSG_UNLOCK_REPLY;
1370 case DLM_MSG_CANCEL:
1371 return DLM_MSG_CANCEL_REPLY;
1372 case DLM_MSG_LOOKUP:
1373 return DLM_MSG_LOOKUP_REPLY;
1378 static int nodeid_warned(int nodeid, int num_nodes, int *warned)
1382 for (i = 0; i < num_nodes; i++) {
1387 if (warned[i] == nodeid)
1393 void dlm_scan_waiters(struct dlm_ls *ls)
1395 struct dlm_lkb *lkb;
1397 s64 debug_maxus = 0;
1398 u32 debug_scanned = 0;
1399 u32 debug_expired = 0;
1403 if (!dlm_config.ci_waitwarn_us)
1406 mutex_lock(&ls->ls_waiters_mutex);
1408 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1409 if (!lkb->lkb_wait_time)
1414 us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time));
1416 if (us < dlm_config.ci_waitwarn_us)
1419 lkb->lkb_wait_time = 0;
1422 if (us > debug_maxus)
1426 num_nodes = ls->ls_num_nodes;
1427 warned = kcalloc(num_nodes, sizeof(int), GFP_KERNEL);
1431 if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned))
1434 log_error(ls, "waitwarn %x %lld %d us check connection to "
1435 "node %d", lkb->lkb_id, (long long)us,
1436 dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid);
1438 mutex_unlock(&ls->ls_waiters_mutex);
1442 log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us",
1443 debug_scanned, debug_expired,
1444 dlm_config.ci_waitwarn_us, (long long)debug_maxus);
1447 /* add/remove lkb from global waiters list of lkb's waiting for
1448 a reply from a remote node */
1450 static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1452 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1455 mutex_lock(&ls->ls_waiters_mutex);
1457 if (is_overlap_unlock(lkb) ||
1458 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1463 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1465 case DLM_MSG_UNLOCK:
1466 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1468 case DLM_MSG_CANCEL:
1469 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1475 lkb->lkb_wait_count++;
1478 log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1479 lkb->lkb_id, lkb->lkb_wait_type, mstype,
1480 lkb->lkb_wait_count, lkb->lkb_flags);
1484 DLM_ASSERT(!lkb->lkb_wait_count,
1486 printk("wait_count %d\n", lkb->lkb_wait_count););
1488 lkb->lkb_wait_count++;
1489 lkb->lkb_wait_type = mstype;
1490 lkb->lkb_wait_time = ktime_get();
1491 lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1493 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1496 log_error(ls, "addwait error %x %d flags %x %d %d %s",
1497 lkb->lkb_id, error, lkb->lkb_flags, mstype,
1498 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1499 mutex_unlock(&ls->ls_waiters_mutex);
1503 /* We clear the RESEND flag because we might be taking an lkb off the waiters
1504 list as part of process_requestqueue (e.g. a lookup that has an optimized
1505 request reply on the requestqueue) between dlm_recover_waiters_pre() which
1506 set RESEND and dlm_recover_waiters_post() */
1508 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1509 struct dlm_message *ms)
1511 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1512 int overlap_done = 0;
1514 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1515 log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1516 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1521 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1522 log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1523 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1528 /* Cancel state was preemptively cleared by a successful convert,
1529 see next comment, nothing to do. */
1531 if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1532 (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1533 log_debug(ls, "remwait %x cancel_reply wait_type %d",
1534 lkb->lkb_id, lkb->lkb_wait_type);
1538 /* Remove for the convert reply, and premptively remove for the
1539 cancel reply. A convert has been granted while there's still
1540 an outstanding cancel on it (the cancel is moot and the result
1541 in the cancel reply should be 0). We preempt the cancel reply
1542 because the app gets the convert result and then can follow up
1543 with another op, like convert. This subsequent op would see the
1544 lingering state of the cancel and fail with -EBUSY. */
1546 if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1547 (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1548 is_overlap_cancel(lkb) && ms && !ms->m_result) {
1549 log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1551 lkb->lkb_wait_type = 0;
1552 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1553 lkb->lkb_wait_count--;
1557 /* N.B. type of reply may not always correspond to type of original
1558 msg due to lookup->request optimization, verify others? */
1560 if (lkb->lkb_wait_type) {
1561 lkb->lkb_wait_type = 0;
1565 log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1566 lkb->lkb_id, ms ? ms->m_header.h_nodeid : 0, lkb->lkb_remid,
1567 mstype, lkb->lkb_flags);
1571 /* the force-unlock/cancel has completed and we haven't recvd a reply
1572 to the op that was in progress prior to the unlock/cancel; we
1573 give up on any reply to the earlier op. FIXME: not sure when/how
1574 this would happen */
1576 if (overlap_done && lkb->lkb_wait_type) {
1577 log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1578 lkb->lkb_id, mstype, lkb->lkb_wait_type);
1579 lkb->lkb_wait_count--;
1580 lkb->lkb_wait_type = 0;
1583 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1585 lkb->lkb_flags &= ~DLM_IFL_RESEND;
1586 lkb->lkb_wait_count--;
1587 if (!lkb->lkb_wait_count)
1588 list_del_init(&lkb->lkb_wait_reply);
1593 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1595 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1598 mutex_lock(&ls->ls_waiters_mutex);
1599 error = _remove_from_waiters(lkb, mstype, NULL);
1600 mutex_unlock(&ls->ls_waiters_mutex);
1604 /* Handles situations where we might be processing a "fake" or "stub" reply in
1605 which we can't try to take waiters_mutex again. */
1607 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1609 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1612 if (ms->m_flags != DLM_IFL_STUB_MS)
1613 mutex_lock(&ls->ls_waiters_mutex);
1614 error = _remove_from_waiters(lkb, ms->m_type, ms);
1615 if (ms->m_flags != DLM_IFL_STUB_MS)
1616 mutex_unlock(&ls->ls_waiters_mutex);
1620 /* If there's an rsb for the same resource being removed, ensure
1621 that the remove message is sent before the new lookup message.
1622 It should be rare to need a delay here, but if not, then it may
1623 be worthwhile to add a proper wait mechanism rather than a delay. */
1625 static void wait_pending_remove(struct dlm_rsb *r)
1627 struct dlm_ls *ls = r->res_ls;
1629 spin_lock(&ls->ls_remove_spin);
1630 if (ls->ls_remove_len &&
1631 !rsb_cmp(r, ls->ls_remove_name, ls->ls_remove_len)) {
1632 log_debug(ls, "delay lookup for remove dir %d %s",
1633 r->res_dir_nodeid, r->res_name);
1634 spin_unlock(&ls->ls_remove_spin);
1638 spin_unlock(&ls->ls_remove_spin);
1642 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1643 * read by other threads in wait_pending_remove. ls_remove_names
1644 * and ls_remove_lens are only used by the scan thread, so they do
1645 * not need protection.
1648 static void shrink_bucket(struct dlm_ls *ls, int b)
1650 struct rb_node *n, *next;
1653 int our_nodeid = dlm_our_nodeid();
1654 int remote_count = 0;
1655 int need_shrink = 0;
1658 memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1660 spin_lock(&ls->ls_rsbtbl[b].lock);
1662 if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1663 spin_unlock(&ls->ls_rsbtbl[b].lock);
1667 for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1669 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1671 /* If we're the directory record for this rsb, and
1672 we're not the master of it, then we need to wait
1673 for the master node to send us a dir remove for
1674 before removing the dir record. */
1676 if (!dlm_no_directory(ls) &&
1677 (r->res_master_nodeid != our_nodeid) &&
1678 (dlm_dir_nodeid(r) == our_nodeid)) {
1684 if (!time_after_eq(jiffies, r->res_toss_time +
1685 dlm_config.ci_toss_secs * HZ)) {
1689 if (!dlm_no_directory(ls) &&
1690 (r->res_master_nodeid == our_nodeid) &&
1691 (dlm_dir_nodeid(r) != our_nodeid)) {
1693 /* We're the master of this rsb but we're not
1694 the directory record, so we need to tell the
1695 dir node to remove the dir record. */
1697 ls->ls_remove_lens[remote_count] = r->res_length;
1698 memcpy(ls->ls_remove_names[remote_count], r->res_name,
1699 DLM_RESNAME_MAXLEN);
1702 if (remote_count >= DLM_REMOVE_NAMES_MAX)
1707 if (!kref_put(&r->res_ref, kill_rsb)) {
1708 log_error(ls, "tossed rsb in use %s", r->res_name);
1712 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1717 ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1719 ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1720 spin_unlock(&ls->ls_rsbtbl[b].lock);
1723 * While searching for rsb's to free, we found some that require
1724 * remote removal. We leave them in place and find them again here
1725 * so there is a very small gap between removing them from the toss
1726 * list and sending the removal. Keeping this gap small is
1727 * important to keep us (the master node) from being out of sync
1728 * with the remote dir node for very long.
1730 * From the time the rsb is removed from toss until just after
1731 * send_remove, the rsb name is saved in ls_remove_name. A new
1732 * lookup checks this to ensure that a new lookup message for the
1733 * same resource name is not sent just before the remove message.
1736 for (i = 0; i < remote_count; i++) {
1737 name = ls->ls_remove_names[i];
1738 len = ls->ls_remove_lens[i];
1740 spin_lock(&ls->ls_rsbtbl[b].lock);
1741 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1743 spin_unlock(&ls->ls_rsbtbl[b].lock);
1744 log_debug(ls, "remove_name not toss %s", name);
1748 if (r->res_master_nodeid != our_nodeid) {
1749 spin_unlock(&ls->ls_rsbtbl[b].lock);
1750 log_debug(ls, "remove_name master %d dir %d our %d %s",
1751 r->res_master_nodeid, r->res_dir_nodeid,
1756 if (r->res_dir_nodeid == our_nodeid) {
1757 /* should never happen */
1758 spin_unlock(&ls->ls_rsbtbl[b].lock);
1759 log_error(ls, "remove_name dir %d master %d our %d %s",
1760 r->res_dir_nodeid, r->res_master_nodeid,
1765 if (!time_after_eq(jiffies, r->res_toss_time +
1766 dlm_config.ci_toss_secs * HZ)) {
1767 spin_unlock(&ls->ls_rsbtbl[b].lock);
1768 log_debug(ls, "remove_name toss_time %lu now %lu %s",
1769 r->res_toss_time, jiffies, name);
1773 if (!kref_put(&r->res_ref, kill_rsb)) {
1774 spin_unlock(&ls->ls_rsbtbl[b].lock);
1775 log_error(ls, "remove_name in use %s", name);
1779 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1781 /* block lookup of same name until we've sent remove */
1782 spin_lock(&ls->ls_remove_spin);
1783 ls->ls_remove_len = len;
1784 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1785 spin_unlock(&ls->ls_remove_spin);
1786 spin_unlock(&ls->ls_rsbtbl[b].lock);
1790 /* allow lookup of name again */
1791 spin_lock(&ls->ls_remove_spin);
1792 ls->ls_remove_len = 0;
1793 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1794 spin_unlock(&ls->ls_remove_spin);
1800 void dlm_scan_rsbs(struct dlm_ls *ls)
1804 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1805 shrink_bucket(ls, i);
1806 if (dlm_locking_stopped(ls))
1812 static void add_timeout(struct dlm_lkb *lkb)
1814 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1816 if (is_master_copy(lkb))
1819 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1820 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1821 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1824 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1829 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1830 mutex_lock(&ls->ls_timeout_mutex);
1832 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1833 mutex_unlock(&ls->ls_timeout_mutex);
1836 static void del_timeout(struct dlm_lkb *lkb)
1838 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1840 mutex_lock(&ls->ls_timeout_mutex);
1841 if (!list_empty(&lkb->lkb_time_list)) {
1842 list_del_init(&lkb->lkb_time_list);
1845 mutex_unlock(&ls->ls_timeout_mutex);
1848 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1849 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1850 and then lock rsb because of lock ordering in add_timeout. We may need
1851 to specify some special timeout-related bits in the lkb that are just to
1852 be accessed under the timeout_mutex. */
1854 void dlm_scan_timeout(struct dlm_ls *ls)
1857 struct dlm_lkb *lkb;
1858 int do_cancel, do_warn;
1862 if (dlm_locking_stopped(ls))
1867 mutex_lock(&ls->ls_timeout_mutex);
1868 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1870 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1871 lkb->lkb_timestamp));
1873 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1874 wait_us >= (lkb->lkb_timeout_cs * 10000))
1877 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1878 wait_us >= dlm_config.ci_timewarn_cs * 10000)
1881 if (!do_cancel && !do_warn)
1886 mutex_unlock(&ls->ls_timeout_mutex);
1888 if (!do_cancel && !do_warn)
1891 r = lkb->lkb_resource;
1896 /* clear flag so we only warn once */
1897 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1898 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1900 dlm_timeout_warn(lkb);
1904 log_debug(ls, "timeout cancel %x node %d %s",
1905 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1906 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1907 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1909 _cancel_lock(r, lkb);
1918 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1919 dlm_recoverd before checking/setting ls_recover_begin. */
1921 void dlm_adjust_timeouts(struct dlm_ls *ls)
1923 struct dlm_lkb *lkb;
1924 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1926 ls->ls_recover_begin = 0;
1927 mutex_lock(&ls->ls_timeout_mutex);
1928 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1929 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1930 mutex_unlock(&ls->ls_timeout_mutex);
1932 if (!dlm_config.ci_waitwarn_us)
1935 mutex_lock(&ls->ls_waiters_mutex);
1936 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1937 if (ktime_to_us(lkb->lkb_wait_time))
1938 lkb->lkb_wait_time = ktime_get();
1940 mutex_unlock(&ls->ls_waiters_mutex);
1943 /* lkb is master or local copy */
1945 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1947 int b, len = r->res_ls->ls_lvblen;
1949 /* b=1 lvb returned to caller
1950 b=0 lvb written to rsb or invalidated
1953 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1956 if (!lkb->lkb_lvbptr)
1959 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1965 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1966 lkb->lkb_lvbseq = r->res_lvbseq;
1968 } else if (b == 0) {
1969 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1970 rsb_set_flag(r, RSB_VALNOTVALID);
1974 if (!lkb->lkb_lvbptr)
1977 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1981 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1986 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1988 lkb->lkb_lvbseq = r->res_lvbseq;
1989 rsb_clear_flag(r, RSB_VALNOTVALID);
1992 if (rsb_flag(r, RSB_VALNOTVALID))
1993 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1996 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1998 if (lkb->lkb_grmode < DLM_LOCK_PW)
2001 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
2002 rsb_set_flag(r, RSB_VALNOTVALID);
2006 if (!lkb->lkb_lvbptr)
2009 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2013 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
2018 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2020 rsb_clear_flag(r, RSB_VALNOTVALID);
2023 /* lkb is process copy (pc) */
2025 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2026 struct dlm_message *ms)
2030 if (!lkb->lkb_lvbptr)
2033 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2036 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
2038 int len = receive_extralen(ms);
2039 if (len > r->res_ls->ls_lvblen)
2040 len = r->res_ls->ls_lvblen;
2041 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2042 lkb->lkb_lvbseq = ms->m_lvbseq;
2046 /* Manipulate lkb's on rsb's convert/granted/waiting queues
2047 remove_lock -- used for unlock, removes lkb from granted
2048 revert_lock -- used for cancel, moves lkb from convert to granted
2049 grant_lock -- used for request and convert, adds lkb to granted or
2050 moves lkb from convert or waiting to granted
2052 Each of these is used for master or local copy lkb's. There is
2053 also a _pc() variation used to make the corresponding change on
2054 a process copy (pc) lkb. */
2056 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2059 lkb->lkb_grmode = DLM_LOCK_IV;
2060 /* this unhold undoes the original ref from create_lkb()
2061 so this leads to the lkb being freed */
2065 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2067 set_lvb_unlock(r, lkb);
2068 _remove_lock(r, lkb);
2071 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2073 _remove_lock(r, lkb);
2076 /* returns: 0 did nothing
2077 1 moved lock to granted
2080 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2084 lkb->lkb_rqmode = DLM_LOCK_IV;
2086 switch (lkb->lkb_status) {
2087 case DLM_LKSTS_GRANTED:
2089 case DLM_LKSTS_CONVERT:
2090 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2093 case DLM_LKSTS_WAITING:
2095 lkb->lkb_grmode = DLM_LOCK_IV;
2096 /* this unhold undoes the original ref from create_lkb()
2097 so this leads to the lkb being freed */
2102 log_print("invalid status for revert %d", lkb->lkb_status);
2107 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2109 return revert_lock(r, lkb);
2112 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2114 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2115 lkb->lkb_grmode = lkb->lkb_rqmode;
2116 if (lkb->lkb_status)
2117 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2119 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2122 lkb->lkb_rqmode = DLM_LOCK_IV;
2123 lkb->lkb_highbast = 0;
2126 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2128 set_lvb_lock(r, lkb);
2129 _grant_lock(r, lkb);
2132 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2133 struct dlm_message *ms)
2135 set_lvb_lock_pc(r, lkb, ms);
2136 _grant_lock(r, lkb);
2139 /* called by grant_pending_locks() which means an async grant message must
2140 be sent to the requesting node in addition to granting the lock if the
2141 lkb belongs to a remote node. */
2143 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2146 if (is_master_copy(lkb))
2149 queue_cast(r, lkb, 0);
2152 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2153 change the granted/requested modes. We're munging things accordingly in
2155 CONVDEADLK: our grmode may have been forced down to NL to resolve a
2157 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2158 compatible with other granted locks */
2160 static void munge_demoted(struct dlm_lkb *lkb)
2162 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2163 log_print("munge_demoted %x invalid modes gr %d rq %d",
2164 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2168 lkb->lkb_grmode = DLM_LOCK_NL;
2171 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2173 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
2174 ms->m_type != DLM_MSG_GRANT) {
2175 log_print("munge_altmode %x invalid reply type %d",
2176 lkb->lkb_id, ms->m_type);
2180 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2181 lkb->lkb_rqmode = DLM_LOCK_PR;
2182 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2183 lkb->lkb_rqmode = DLM_LOCK_CW;
2185 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2190 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2192 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2194 if (lkb->lkb_id == first->lkb_id)
2200 /* Check if the given lkb conflicts with another lkb on the queue. */
2202 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2204 struct dlm_lkb *this;
2206 list_for_each_entry(this, head, lkb_statequeue) {
2209 if (!modes_compat(this, lkb))
2216 * "A conversion deadlock arises with a pair of lock requests in the converting
2217 * queue for one resource. The granted mode of each lock blocks the requested
2218 * mode of the other lock."
2220 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2221 * convert queue from being granted, then deadlk/demote lkb.
2224 * Granted Queue: empty
2225 * Convert Queue: NL->EX (first lock)
2226 * PR->EX (second lock)
2228 * The first lock can't be granted because of the granted mode of the second
2229 * lock and the second lock can't be granted because it's not first in the
2230 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2231 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2232 * flag set and return DEMOTED in the lksb flags.
2234 * Originally, this function detected conv-deadlk in a more limited scope:
2235 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2236 * - if lkb1 was the first entry in the queue (not just earlier), and was
2237 * blocked by the granted mode of lkb2, and there was nothing on the
2238 * granted queue preventing lkb1 from being granted immediately, i.e.
2239 * lkb2 was the only thing preventing lkb1 from being granted.
2241 * That second condition meant we'd only say there was conv-deadlk if
2242 * resolving it (by demotion) would lead to the first lock on the convert
2243 * queue being granted right away. It allowed conversion deadlocks to exist
2244 * between locks on the convert queue while they couldn't be granted anyway.
2246 * Now, we detect and take action on conversion deadlocks immediately when
2247 * they're created, even if they may not be immediately consequential. If
2248 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2249 * mode that would prevent lkb1's conversion from being granted, we do a
2250 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2251 * I think this means that the lkb_is_ahead condition below should always
2252 * be zero, i.e. there will never be conv-deadlk between two locks that are
2253 * both already on the convert queue.
2256 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2258 struct dlm_lkb *lkb1;
2259 int lkb_is_ahead = 0;
2261 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2267 if (!lkb_is_ahead) {
2268 if (!modes_compat(lkb2, lkb1))
2271 if (!modes_compat(lkb2, lkb1) &&
2272 !modes_compat(lkb1, lkb2))
2280 * Return 1 if the lock can be granted, 0 otherwise.
2281 * Also detect and resolve conversion deadlocks.
2283 * lkb is the lock to be granted
2285 * now is 1 if the function is being called in the context of the
2286 * immediate request, it is 0 if called later, after the lock has been
2289 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2292 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2295 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2298 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2301 * 6-10: Version 5.4 introduced an option to address the phenomenon of
2302 * a new request for a NL mode lock being blocked.
2304 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2305 * request, then it would be granted. In essence, the use of this flag
2306 * tells the Lock Manager to expedite theis request by not considering
2307 * what may be in the CONVERTING or WAITING queues... As of this
2308 * writing, the EXPEDITE flag can be used only with new requests for NL
2309 * mode locks. This flag is not valid for conversion requests.
2311 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
2312 * conversion or used with a non-NL requested mode. We also know an
2313 * EXPEDITE request is always granted immediately, so now must always
2314 * be 1. The full condition to grant an expedite request: (now &&
2315 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2316 * therefore be shortened to just checking the flag.
2319 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2323 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2324 * added to the remaining conditions.
2327 if (queue_conflict(&r->res_grantqueue, lkb))
2331 * 6-3: By default, a conversion request is immediately granted if the
2332 * requested mode is compatible with the modes of all other granted
2336 if (queue_conflict(&r->res_convertqueue, lkb))
2340 * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2341 * locks for a recovered rsb, on which lkb's have been rebuilt.
2342 * The lkb's may have been rebuilt on the queues in a different
2343 * order than they were in on the previous master. So, granting
2344 * queued conversions in order after recovery doesn't make sense
2345 * since the order hasn't been preserved anyway. The new order
2346 * could also have created a new "in place" conversion deadlock.
2347 * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2348 * After recovery, there would be no granted locks, and possibly
2349 * NL->EX, PR->EX, an in-place conversion deadlock.) So, after
2350 * recovery, grant conversions without considering order.
2353 if (conv && recover)
2357 * 6-5: But the default algorithm for deciding whether to grant or
2358 * queue conversion requests does not by itself guarantee that such
2359 * requests are serviced on a "first come first serve" basis. This, in
2360 * turn, can lead to a phenomenon known as "indefinate postponement".
2362 * 6-7: This issue is dealt with by using the optional QUECVT flag with
2363 * the system service employed to request a lock conversion. This flag
2364 * forces certain conversion requests to be queued, even if they are
2365 * compatible with the granted modes of other locks on the same
2366 * resource. Thus, the use of this flag results in conversion requests
2367 * being ordered on a "first come first servce" basis.
2369 * DCT: This condition is all about new conversions being able to occur
2370 * "in place" while the lock remains on the granted queue (assuming
2371 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
2372 * doesn't _have_ to go onto the convert queue where it's processed in
2373 * order. The "now" variable is necessary to distinguish converts
2374 * being received and processed for the first time now, because once a
2375 * convert is moved to the conversion queue the condition below applies
2376 * requiring fifo granting.
2379 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2383 * Even if the convert is compat with all granted locks,
2384 * QUECVT forces it behind other locks on the convert queue.
2387 if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2388 if (list_empty(&r->res_convertqueue))
2395 * The NOORDER flag is set to avoid the standard vms rules on grant
2399 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2403 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2404 * granted until all other conversion requests ahead of it are granted
2408 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2412 * 6-4: By default, a new request is immediately granted only if all
2413 * three of the following conditions are satisfied when the request is
2415 * - The queue of ungranted conversion requests for the resource is
2417 * - The queue of ungranted new requests for the resource is empty.
2418 * - The mode of the new request is compatible with the most
2419 * restrictive mode of all granted locks on the resource.
2422 if (now && !conv && list_empty(&r->res_convertqueue) &&
2423 list_empty(&r->res_waitqueue))
2427 * 6-4: Once a lock request is in the queue of ungranted new requests,
2428 * it cannot be granted until the queue of ungranted conversion
2429 * requests is empty, all ungranted new requests ahead of it are
2430 * granted and/or canceled, and it is compatible with the granted mode
2431 * of the most restrictive lock granted on the resource.
2434 if (!now && !conv && list_empty(&r->res_convertqueue) &&
2435 first_in_list(lkb, &r->res_waitqueue))
2441 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2442 int recover, int *err)
2445 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2446 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2451 rv = _can_be_granted(r, lkb, now, recover);
2456 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2457 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2458 * cancels one of the locks.
2461 if (is_convert && can_be_queued(lkb) &&
2462 conversion_deadlock_detect(r, lkb)) {
2463 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2464 lkb->lkb_grmode = DLM_LOCK_NL;
2465 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2469 log_print("can_be_granted deadlock %x now %d",
2477 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2478 * to grant a request in a mode other than the normal rqmode. It's a
2479 * simple way to provide a big optimization to applications that can
2483 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2485 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2489 lkb->lkb_rqmode = alt;
2490 rv = _can_be_granted(r, lkb, now, 0);
2492 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2494 lkb->lkb_rqmode = rqmode;
2500 /* Returns the highest requested mode of all blocked conversions; sets
2501 cw if there's a blocked conversion to DLM_LOCK_CW. */
2503 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2504 unsigned int *count)
2506 struct dlm_lkb *lkb, *s;
2507 int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2508 int hi, demoted, quit, grant_restart, demote_restart;
2517 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2518 demoted = is_demoted(lkb);
2521 if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2522 grant_lock_pending(r, lkb);
2529 if (!demoted && is_demoted(lkb)) {
2530 log_print("WARN: pending demoted %x node %d %s",
2531 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2538 * If DLM_LKB_NODLKWT flag is set and conversion
2539 * deadlock is detected, we request blocking AST and
2540 * down (or cancel) conversion.
2542 if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
2543 if (lkb->lkb_highbast < lkb->lkb_rqmode) {
2544 queue_bast(r, lkb, lkb->lkb_rqmode);
2545 lkb->lkb_highbast = lkb->lkb_rqmode;
2548 log_print("WARN: pending deadlock %x node %d %s",
2549 lkb->lkb_id, lkb->lkb_nodeid,
2556 hi = max_t(int, lkb->lkb_rqmode, hi);
2558 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2564 if (demote_restart && !quit) {
2569 return max_t(int, high, hi);
2572 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2573 unsigned int *count)
2575 struct dlm_lkb *lkb, *s;
2577 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2578 if (can_be_granted(r, lkb, 0, 0, NULL)) {
2579 grant_lock_pending(r, lkb);
2583 high = max_t(int, lkb->lkb_rqmode, high);
2584 if (lkb->lkb_rqmode == DLM_LOCK_CW)
2592 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2593 on either the convert or waiting queue.
2594 high is the largest rqmode of all locks blocked on the convert or
2597 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2599 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2600 if (gr->lkb_highbast < DLM_LOCK_EX)
2605 if (gr->lkb_highbast < high &&
2606 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2611 static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2613 struct dlm_lkb *lkb, *s;
2614 int high = DLM_LOCK_IV;
2617 if (!is_master(r)) {
2618 log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2623 high = grant_pending_convert(r, high, &cw, count);
2624 high = grant_pending_wait(r, high, &cw, count);
2626 if (high == DLM_LOCK_IV)
2630 * If there are locks left on the wait/convert queue then send blocking
2631 * ASTs to granted locks based on the largest requested mode (high)
2635 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2636 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2637 if (cw && high == DLM_LOCK_PR &&
2638 lkb->lkb_grmode == DLM_LOCK_PR)
2639 queue_bast(r, lkb, DLM_LOCK_CW);
2641 queue_bast(r, lkb, high);
2642 lkb->lkb_highbast = high;
2647 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2649 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2650 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2651 if (gr->lkb_highbast < DLM_LOCK_EX)
2656 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2661 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2662 struct dlm_lkb *lkb)
2666 list_for_each_entry(gr, head, lkb_statequeue) {
2667 /* skip self when sending basts to convertqueue */
2670 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2671 queue_bast(r, gr, lkb->lkb_rqmode);
2672 gr->lkb_highbast = lkb->lkb_rqmode;
2677 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2679 send_bast_queue(r, &r->res_grantqueue, lkb);
2682 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2684 send_bast_queue(r, &r->res_grantqueue, lkb);
2685 send_bast_queue(r, &r->res_convertqueue, lkb);
2688 /* set_master(r, lkb) -- set the master nodeid of a resource
2690 The purpose of this function is to set the nodeid field in the given
2691 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2692 known, it can just be copied to the lkb and the function will return
2693 0. If the rsb's nodeid is _not_ known, it needs to be looked up
2694 before it can be copied to the lkb.
2696 When the rsb nodeid is being looked up remotely, the initial lkb
2697 causing the lookup is kept on the ls_waiters list waiting for the
2698 lookup reply. Other lkb's waiting for the same rsb lookup are kept
2699 on the rsb's res_lookup list until the master is verified.
2702 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2703 1: the rsb master is not available and the lkb has been placed on
2707 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2709 int our_nodeid = dlm_our_nodeid();
2711 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2712 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2713 r->res_first_lkid = lkb->lkb_id;
2714 lkb->lkb_nodeid = r->res_nodeid;
2718 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2719 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2723 if (r->res_master_nodeid == our_nodeid) {
2724 lkb->lkb_nodeid = 0;
2728 if (r->res_master_nodeid) {
2729 lkb->lkb_nodeid = r->res_master_nodeid;
2733 if (dlm_dir_nodeid(r) == our_nodeid) {
2734 /* This is a somewhat unusual case; find_rsb will usually
2735 have set res_master_nodeid when dir nodeid is local, but
2736 there are cases where we become the dir node after we've
2737 past find_rsb and go through _request_lock again.
2738 confirm_master() or process_lookup_list() needs to be
2739 called after this. */
2740 log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2741 lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2743 r->res_master_nodeid = our_nodeid;
2745 lkb->lkb_nodeid = 0;
2749 wait_pending_remove(r);
2751 r->res_first_lkid = lkb->lkb_id;
2752 send_lookup(r, lkb);
2756 static void process_lookup_list(struct dlm_rsb *r)
2758 struct dlm_lkb *lkb, *safe;
2760 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2761 list_del_init(&lkb->lkb_rsb_lookup);
2762 _request_lock(r, lkb);
2767 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
2769 static void confirm_master(struct dlm_rsb *r, int error)
2771 struct dlm_lkb *lkb;
2773 if (!r->res_first_lkid)
2779 r->res_first_lkid = 0;
2780 process_lookup_list(r);
2786 /* the remote request failed and won't be retried (it was
2787 a NOQUEUE, or has been canceled/unlocked); make a waiting
2788 lkb the first_lkid */
2790 r->res_first_lkid = 0;
2792 if (!list_empty(&r->res_lookup)) {
2793 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2795 list_del_init(&lkb->lkb_rsb_lookup);
2796 r->res_first_lkid = lkb->lkb_id;
2797 _request_lock(r, lkb);
2802 log_error(r->res_ls, "confirm_master unknown error %d", error);
2806 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2807 int namelen, unsigned long timeout_cs,
2808 void (*ast) (void *astparam),
2810 void (*bast) (void *astparam, int mode),
2811 struct dlm_args *args)
2815 /* check for invalid arg usage */
2817 if (mode < 0 || mode > DLM_LOCK_EX)
2820 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2823 if (flags & DLM_LKF_CANCEL)
2826 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2829 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2832 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2835 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2838 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2841 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2844 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2850 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2853 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2856 /* these args will be copied to the lkb in validate_lock_args,
2857 it cannot be done now because when converting locks, fields in
2858 an active lkb cannot be modified before locking the rsb */
2860 args->flags = flags;
2862 args->astparam = astparam;
2863 args->bastfn = bast;
2864 args->timeout = timeout_cs;
2872 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2874 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2875 DLM_LKF_FORCEUNLOCK))
2878 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2881 args->flags = flags;
2882 args->astparam = astarg;
2886 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2887 struct dlm_args *args)
2891 if (args->flags & DLM_LKF_CONVERT) {
2892 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2895 if (args->flags & DLM_LKF_QUECVT &&
2896 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2900 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2903 if (lkb->lkb_wait_type)
2906 if (is_overlap(lkb))
2910 lkb->lkb_exflags = args->flags;
2911 lkb->lkb_sbflags = 0;
2912 lkb->lkb_astfn = args->astfn;
2913 lkb->lkb_astparam = args->astparam;
2914 lkb->lkb_bastfn = args->bastfn;
2915 lkb->lkb_rqmode = args->mode;
2916 lkb->lkb_lksb = args->lksb;
2917 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2918 lkb->lkb_ownpid = (int) current->pid;
2919 lkb->lkb_timeout_cs = args->timeout;
2923 log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2924 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2925 lkb->lkb_status, lkb->lkb_wait_type,
2926 lkb->lkb_resource->res_name);
2930 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2933 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2934 because there may be a lookup in progress and it's valid to do
2935 cancel/unlockf on it */
2937 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2939 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2942 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2943 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2948 /* an lkb may still exist even though the lock is EOL'ed due to a
2949 cancel, unlock or failed noqueue request; an app can't use these
2950 locks; return same error as if the lkid had not been found at all */
2952 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2953 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2958 /* an lkb may be waiting for an rsb lookup to complete where the
2959 lookup was initiated by another lock */
2961 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2962 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2963 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2964 list_del_init(&lkb->lkb_rsb_lookup);
2965 queue_cast(lkb->lkb_resource, lkb,
2966 args->flags & DLM_LKF_CANCEL ?
2967 -DLM_ECANCEL : -DLM_EUNLOCK);
2968 unhold_lkb(lkb); /* undoes create_lkb() */
2970 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2975 /* cancel not allowed with another cancel/unlock in progress */
2977 if (args->flags & DLM_LKF_CANCEL) {
2978 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2981 if (is_overlap(lkb))
2984 /* don't let scand try to do a cancel */
2987 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2988 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2993 /* there's nothing to cancel */
2994 if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2995 !lkb->lkb_wait_type) {
3000 switch (lkb->lkb_wait_type) {
3001 case DLM_MSG_LOOKUP:
3002 case DLM_MSG_REQUEST:
3003 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3006 case DLM_MSG_UNLOCK:
3007 case DLM_MSG_CANCEL:
3010 /* add_to_waiters() will set OVERLAP_CANCEL */
3014 /* do we need to allow a force-unlock if there's a normal unlock
3015 already in progress? in what conditions could the normal unlock
3016 fail such that we'd want to send a force-unlock to be sure? */
3018 if (args->flags & DLM_LKF_FORCEUNLOCK) {
3019 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3022 if (is_overlap_unlock(lkb))
3025 /* don't let scand try to do a cancel */
3028 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3029 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3034 switch (lkb->lkb_wait_type) {
3035 case DLM_MSG_LOOKUP:
3036 case DLM_MSG_REQUEST:
3037 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3040 case DLM_MSG_UNLOCK:
3043 /* add_to_waiters() will set OVERLAP_UNLOCK */
3047 /* normal unlock not allowed if there's any op in progress */
3049 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
3053 /* an overlapping op shouldn't blow away exflags from other op */
3054 lkb->lkb_exflags |= args->flags;
3055 lkb->lkb_sbflags = 0;
3056 lkb->lkb_astparam = args->astparam;
3060 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
3061 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3062 args->flags, lkb->lkb_wait_type,
3063 lkb->lkb_resource->res_name);
3068 * Four stage 4 varieties:
3069 * do_request(), do_convert(), do_unlock(), do_cancel()
3070 * These are called on the master node for the given lock and
3071 * from the central locking logic.
3074 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3078 if (can_be_granted(r, lkb, 1, 0, NULL)) {
3080 queue_cast(r, lkb, 0);
3084 if (can_be_queued(lkb)) {
3085 error = -EINPROGRESS;
3086 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3092 queue_cast(r, lkb, -EAGAIN);
3097 static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3102 if (force_blocking_asts(lkb))
3103 send_blocking_asts_all(r, lkb);
3106 send_blocking_asts(r, lkb);
3111 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3116 /* changing an existing lock may allow others to be granted */
3118 if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3120 queue_cast(r, lkb, 0);
3124 /* can_be_granted() detected that this lock would block in a conversion
3125 deadlock, so we leave it on the granted queue and return EDEADLK in
3126 the ast for the convert. */
3128 if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
3129 /* it's left on the granted queue */
3130 revert_lock(r, lkb);
3131 queue_cast(r, lkb, -EDEADLK);
3136 /* is_demoted() means the can_be_granted() above set the grmode
3137 to NL, and left us on the granted queue. This auto-demotion
3138 (due to CONVDEADLK) might mean other locks, and/or this lock, are
3139 now grantable. We have to try to grant other converting locks
3140 before we try again to grant this one. */
3142 if (is_demoted(lkb)) {
3143 grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3144 if (_can_be_granted(r, lkb, 1, 0)) {
3146 queue_cast(r, lkb, 0);
3149 /* else fall through and move to convert queue */
3152 if (can_be_queued(lkb)) {
3153 error = -EINPROGRESS;
3155 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3161 queue_cast(r, lkb, -EAGAIN);
3166 static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3171 grant_pending_locks(r, NULL);
3172 /* grant_pending_locks also sends basts */
3175 if (force_blocking_asts(lkb))
3176 send_blocking_asts_all(r, lkb);
3179 send_blocking_asts(r, lkb);
3184 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3186 remove_lock(r, lkb);
3187 queue_cast(r, lkb, -DLM_EUNLOCK);
3188 return -DLM_EUNLOCK;
3191 static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3194 grant_pending_locks(r, NULL);
3197 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3199 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3203 error = revert_lock(r, lkb);
3205 queue_cast(r, lkb, -DLM_ECANCEL);
3206 return -DLM_ECANCEL;
3211 static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3215 grant_pending_locks(r, NULL);
3219 * Four stage 3 varieties:
3220 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3223 /* add a new lkb to a possibly new rsb, called by requesting process */
3225 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3229 /* set_master: sets lkb nodeid from r */
3231 error = set_master(r, lkb);
3240 /* receive_request() calls do_request() on remote node */
3241 error = send_request(r, lkb);
3243 error = do_request(r, lkb);
3244 /* for remote locks the request_reply is sent
3245 between do_request and do_request_effects */
3246 do_request_effects(r, lkb, error);
3252 /* change some property of an existing lkb, e.g. mode */
3254 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3259 /* receive_convert() calls do_convert() on remote node */
3260 error = send_convert(r, lkb);
3262 error = do_convert(r, lkb);
3263 /* for remote locks the convert_reply is sent
3264 between do_convert and do_convert_effects */
3265 do_convert_effects(r, lkb, error);
3271 /* remove an existing lkb from the granted queue */
3273 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3278 /* receive_unlock() calls do_unlock() on remote node */
3279 error = send_unlock(r, lkb);
3281 error = do_unlock(r, lkb);
3282 /* for remote locks the unlock_reply is sent
3283 between do_unlock and do_unlock_effects */
3284 do_unlock_effects(r, lkb, error);
3290 /* remove an existing lkb from the convert or wait queue */
3292 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3297 /* receive_cancel() calls do_cancel() on remote node */
3298 error = send_cancel(r, lkb);
3300 error = do_cancel(r, lkb);
3301 /* for remote locks the cancel_reply is sent
3302 between do_cancel and do_cancel_effects */
3303 do_cancel_effects(r, lkb, error);
3310 * Four stage 2 varieties:
3311 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3314 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
3315 int len, struct dlm_args *args)
3320 error = validate_lock_args(ls, lkb, args);
3324 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3331 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3333 error = _request_lock(r, lkb);
3340 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3341 struct dlm_args *args)
3346 r = lkb->lkb_resource;
3351 error = validate_lock_args(ls, lkb, args);
3355 error = _convert_lock(r, lkb);
3362 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3363 struct dlm_args *args)
3368 r = lkb->lkb_resource;
3373 error = validate_unlock_args(lkb, args);
3377 error = _unlock_lock(r, lkb);
3384 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3385 struct dlm_args *args)
3390 r = lkb->lkb_resource;
3395 error = validate_unlock_args(lkb, args);
3399 error = _cancel_lock(r, lkb);
3407 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
3410 int dlm_lock(dlm_lockspace_t *lockspace,
3412 struct dlm_lksb *lksb,
3415 unsigned int namelen,
3416 uint32_t parent_lkid,
3417 void (*ast) (void *astarg),
3419 void (*bast) (void *astarg, int mode))
3422 struct dlm_lkb *lkb;
3423 struct dlm_args args;
3424 int error, convert = flags & DLM_LKF_CONVERT;
3426 ls = dlm_find_lockspace_local(lockspace);
3430 dlm_lock_recovery(ls);
3433 error = find_lkb(ls, lksb->sb_lkid, &lkb);
3435 error = create_lkb(ls, &lkb);
3440 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3441 astarg, bast, &args);
3446 error = convert_lock(ls, lkb, &args);
3448 error = request_lock(ls, lkb, name, namelen, &args);
3450 if (error == -EINPROGRESS)
3453 if (convert || error)
3455 if (error == -EAGAIN || error == -EDEADLK)
3458 dlm_unlock_recovery(ls);
3459 dlm_put_lockspace(ls);
3463 int dlm_unlock(dlm_lockspace_t *lockspace,
3466 struct dlm_lksb *lksb,
3470 struct dlm_lkb *lkb;
3471 struct dlm_args args;
3474 ls = dlm_find_lockspace_local(lockspace);
3478 dlm_lock_recovery(ls);
3480 error = find_lkb(ls, lkid, &lkb);
3484 error = set_unlock_args(flags, astarg, &args);
3488 if (flags & DLM_LKF_CANCEL)
3489 error = cancel_lock(ls, lkb, &args);
3491 error = unlock_lock(ls, lkb, &args);
3493 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3495 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3500 dlm_unlock_recovery(ls);
3501 dlm_put_lockspace(ls);
3506 * send/receive routines for remote operations and replies
3510 * send_request receive_request
3511 * send_convert receive_convert
3512 * send_unlock receive_unlock
3513 * send_cancel receive_cancel
3514 * send_grant receive_grant
3515 * send_bast receive_bast
3516 * send_lookup receive_lookup
3517 * send_remove receive_remove
3520 * receive_request_reply send_request_reply
3521 * receive_convert_reply send_convert_reply
3522 * receive_unlock_reply send_unlock_reply
3523 * receive_cancel_reply send_cancel_reply
3524 * receive_lookup_reply send_lookup_reply
3527 static int _create_message(struct dlm_ls *ls, int mb_len,
3528 int to_nodeid, int mstype,
3529 struct dlm_message **ms_ret,
3530 struct dlm_mhandle **mh_ret)
3532 struct dlm_message *ms;
3533 struct dlm_mhandle *mh;
3536 /* get_buffer gives us a message handle (mh) that we need to
3537 pass into midcomms_commit and a message buffer (mb) that we
3538 write our data into */
3540 mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb);
3544 ms = (struct dlm_message *) mb;
3546 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3547 ms->m_header.u.h_lockspace = ls->ls_global_id;
3548 ms->m_header.h_nodeid = dlm_our_nodeid();
3549 ms->m_header.h_length = mb_len;
3550 ms->m_header.h_cmd = DLM_MSG;
3552 ms->m_type = mstype;
3559 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3560 int to_nodeid, int mstype,
3561 struct dlm_message **ms_ret,
3562 struct dlm_mhandle **mh_ret)
3564 int mb_len = sizeof(struct dlm_message);
3567 case DLM_MSG_REQUEST:
3568 case DLM_MSG_LOOKUP:
3569 case DLM_MSG_REMOVE:
3570 mb_len += r->res_length;
3572 case DLM_MSG_CONVERT:
3573 case DLM_MSG_UNLOCK:
3574 case DLM_MSG_REQUEST_REPLY:
3575 case DLM_MSG_CONVERT_REPLY:
3577 if (lkb && lkb->lkb_lvbptr)
3578 mb_len += r->res_ls->ls_lvblen;
3582 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3586 /* further lowcomms enhancements or alternate implementations may make
3587 the return value from this function useful at some point */
3589 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
3591 dlm_message_out(ms);
3592 dlm_midcomms_commit_mhandle(mh);
3596 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3597 struct dlm_message *ms)
3599 ms->m_nodeid = lkb->lkb_nodeid;
3600 ms->m_pid = lkb->lkb_ownpid;
3601 ms->m_lkid = lkb->lkb_id;
3602 ms->m_remid = lkb->lkb_remid;
3603 ms->m_exflags = lkb->lkb_exflags;
3604 ms->m_sbflags = lkb->lkb_sbflags;
3605 ms->m_flags = lkb->lkb_flags;
3606 ms->m_lvbseq = lkb->lkb_lvbseq;
3607 ms->m_status = lkb->lkb_status;
3608 ms->m_grmode = lkb->lkb_grmode;
3609 ms->m_rqmode = lkb->lkb_rqmode;
3610 ms->m_hash = r->res_hash;
3612 /* m_result and m_bastmode are set from function args,
3613 not from lkb fields */
3615 if (lkb->lkb_bastfn)
3616 ms->m_asts |= DLM_CB_BAST;
3618 ms->m_asts |= DLM_CB_CAST;
3620 /* compare with switch in create_message; send_remove() doesn't
3623 switch (ms->m_type) {
3624 case DLM_MSG_REQUEST:
3625 case DLM_MSG_LOOKUP:
3626 memcpy(ms->m_extra, r->res_name, r->res_length);
3628 case DLM_MSG_CONVERT:
3629 case DLM_MSG_UNLOCK:
3630 case DLM_MSG_REQUEST_REPLY:
3631 case DLM_MSG_CONVERT_REPLY:
3633 if (!lkb->lkb_lvbptr)
3635 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3640 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3642 struct dlm_message *ms;
3643 struct dlm_mhandle *mh;
3644 int to_nodeid, error;
3646 to_nodeid = r->res_nodeid;
3648 error = add_to_waiters(lkb, mstype, to_nodeid);
3652 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3656 send_args(r, lkb, ms);
3658 error = send_message(mh, ms);
3664 remove_from_waiters(lkb, msg_reply_type(mstype));
3668 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3670 return send_common(r, lkb, DLM_MSG_REQUEST);
3673 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3677 error = send_common(r, lkb, DLM_MSG_CONVERT);
3679 /* down conversions go without a reply from the master */
3680 if (!error && down_conversion(lkb)) {
3681 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3682 r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS;
3683 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3684 r->res_ls->ls_stub_ms.m_result = 0;
3685 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3691 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
3692 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3693 that the master is still correct. */
3695 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3697 return send_common(r, lkb, DLM_MSG_UNLOCK);
3700 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3702 return send_common(r, lkb, DLM_MSG_CANCEL);
3705 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3707 struct dlm_message *ms;
3708 struct dlm_mhandle *mh;
3709 int to_nodeid, error;
3711 to_nodeid = lkb->lkb_nodeid;
3713 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3717 send_args(r, lkb, ms);
3721 error = send_message(mh, ms);
3726 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3728 struct dlm_message *ms;
3729 struct dlm_mhandle *mh;
3730 int to_nodeid, error;
3732 to_nodeid = lkb->lkb_nodeid;
3734 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3738 send_args(r, lkb, ms);
3740 ms->m_bastmode = mode;
3742 error = send_message(mh, ms);
3747 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3749 struct dlm_message *ms;
3750 struct dlm_mhandle *mh;
3751 int to_nodeid, error;
3753 to_nodeid = dlm_dir_nodeid(r);
3755 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3759 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3763 send_args(r, lkb, ms);
3765 error = send_message(mh, ms);
3771 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3775 static int send_remove(struct dlm_rsb *r)
3777 struct dlm_message *ms;
3778 struct dlm_mhandle *mh;
3779 int to_nodeid, error;
3781 to_nodeid = dlm_dir_nodeid(r);
3783 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3787 memcpy(ms->m_extra, r->res_name, r->res_length);
3788 ms->m_hash = r->res_hash;
3790 error = send_message(mh, ms);
3795 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3798 struct dlm_message *ms;
3799 struct dlm_mhandle *mh;
3800 int to_nodeid, error;
3802 to_nodeid = lkb->lkb_nodeid;
3804 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3808 send_args(r, lkb, ms);
3812 error = send_message(mh, ms);
3817 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3819 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3822 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3824 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3827 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3829 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3832 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3834 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3837 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3838 int ret_nodeid, int rv)
3840 struct dlm_rsb *r = &ls->ls_stub_rsb;
3841 struct dlm_message *ms;
3842 struct dlm_mhandle *mh;
3843 int error, nodeid = ms_in->m_header.h_nodeid;
3845 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3849 ms->m_lkid = ms_in->m_lkid;
3851 ms->m_nodeid = ret_nodeid;
3853 error = send_message(mh, ms);
3858 /* which args we save from a received message depends heavily on the type
3859 of message, unlike the send side where we can safely send everything about
3860 the lkb for any type of message */
3862 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3864 lkb->lkb_exflags = ms->m_exflags;
3865 lkb->lkb_sbflags = ms->m_sbflags;
3866 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3867 (ms->m_flags & 0x0000FFFF);
3870 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3872 if (ms->m_flags == DLM_IFL_STUB_MS)
3875 lkb->lkb_sbflags = ms->m_sbflags;
3876 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3877 (ms->m_flags & 0x0000FFFF);
3880 static int receive_extralen(struct dlm_message *ms)
3882 return (ms->m_header.h_length - sizeof(struct dlm_message));
3885 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3886 struct dlm_message *ms)
3890 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3891 if (!lkb->lkb_lvbptr)
3892 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3893 if (!lkb->lkb_lvbptr)
3895 len = receive_extralen(ms);
3896 if (len > ls->ls_lvblen)
3897 len = ls->ls_lvblen;
3898 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3903 static void fake_bastfn(void *astparam, int mode)
3905 log_print("fake_bastfn should not be called");
3908 static void fake_astfn(void *astparam)
3910 log_print("fake_astfn should not be called");
3913 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3914 struct dlm_message *ms)
3916 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3917 lkb->lkb_ownpid = ms->m_pid;
3918 lkb->lkb_remid = ms->m_lkid;
3919 lkb->lkb_grmode = DLM_LOCK_IV;
3920 lkb->lkb_rqmode = ms->m_rqmode;
3922 lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
3923 lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
3925 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3926 /* lkb was just created so there won't be an lvb yet */
3927 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3928 if (!lkb->lkb_lvbptr)
3935 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3936 struct dlm_message *ms)
3938 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3941 if (receive_lvb(ls, lkb, ms))
3944 lkb->lkb_rqmode = ms->m_rqmode;
3945 lkb->lkb_lvbseq = ms->m_lvbseq;
3950 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3951 struct dlm_message *ms)
3953 if (receive_lvb(ls, lkb, ms))
3958 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3959 uses to send a reply and that the remote end uses to process the reply. */
3961 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3963 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3964 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3965 lkb->lkb_remid = ms->m_lkid;
3968 /* This is called after the rsb is locked so that we can safely inspect
3969 fields in the lkb. */
3971 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3973 int from = ms->m_header.h_nodeid;
3976 switch (ms->m_type) {
3977 case DLM_MSG_CONVERT:
3978 case DLM_MSG_UNLOCK:
3979 case DLM_MSG_CANCEL:
3980 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3984 case DLM_MSG_CONVERT_REPLY:
3985 case DLM_MSG_UNLOCK_REPLY:
3986 case DLM_MSG_CANCEL_REPLY:
3989 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3993 case DLM_MSG_REQUEST_REPLY:
3994 if (!is_process_copy(lkb))
3996 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
4005 log_error(lkb->lkb_resource->res_ls,
4006 "ignore invalid message %d from %d %x %x %x %d",
4007 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
4008 lkb->lkb_flags, lkb->lkb_nodeid);
4012 static void send_repeat_remove(struct dlm_ls *ls, char *ms_name, int len)
4014 char name[DLM_RESNAME_MAXLEN + 1];
4015 struct dlm_message *ms;
4016 struct dlm_mhandle *mh;
4021 memset(name, 0, sizeof(name));
4022 memcpy(name, ms_name, len);
4024 hash = jhash(name, len, 0);
4025 b = hash & (ls->ls_rsbtbl_size - 1);
4027 dir_nodeid = dlm_hash2nodeid(ls, hash);
4029 log_error(ls, "send_repeat_remove dir %d %s", dir_nodeid, name);
4031 spin_lock(&ls->ls_rsbtbl[b].lock);
4032 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4034 spin_unlock(&ls->ls_rsbtbl[b].lock);
4035 log_error(ls, "repeat_remove on keep %s", name);
4039 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4041 spin_unlock(&ls->ls_rsbtbl[b].lock);
4042 log_error(ls, "repeat_remove on toss %s", name);
4046 /* use ls->remove_name2 to avoid conflict with shrink? */
4048 spin_lock(&ls->ls_remove_spin);
4049 ls->ls_remove_len = len;
4050 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
4051 spin_unlock(&ls->ls_remove_spin);
4052 spin_unlock(&ls->ls_rsbtbl[b].lock);
4054 rv = _create_message(ls, sizeof(struct dlm_message) + len,
4055 dir_nodeid, DLM_MSG_REMOVE, &ms, &mh);
4059 memcpy(ms->m_extra, name, len);
4062 send_message(mh, ms);
4064 spin_lock(&ls->ls_remove_spin);
4065 ls->ls_remove_len = 0;
4066 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
4067 spin_unlock(&ls->ls_remove_spin);
4070 static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4072 struct dlm_lkb *lkb;
4075 int error, namelen = 0;
4077 from_nodeid = ms->m_header.h_nodeid;
4079 error = create_lkb(ls, &lkb);
4083 receive_flags(lkb, ms);
4084 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4085 error = receive_request_args(ls, lkb, ms);
4091 /* The dir node is the authority on whether we are the master
4092 for this rsb or not, so if the master sends us a request, we should
4093 recreate the rsb if we've destroyed it. This race happens when we
4094 send a remove message to the dir node at the same time that the dir
4095 node sends us a request for the rsb. */
4097 namelen = receive_extralen(ms);
4099 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4100 R_RECEIVE_REQUEST, &r);
4108 if (r->res_master_nodeid != dlm_our_nodeid()) {
4109 error = validate_master_nodeid(ls, r, from_nodeid);
4119 error = do_request(r, lkb);
4120 send_request_reply(r, lkb, error);
4121 do_request_effects(r, lkb, error);
4126 if (error == -EINPROGRESS)
4133 /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4134 and do this receive_request again from process_lookup_list once
4135 we get the lookup reply. This would avoid a many repeated
4136 ENOTBLK request failures when the lookup reply designating us
4137 as master is delayed. */
4139 /* We could repeatedly return -EBADR here if our send_remove() is
4140 delayed in being sent/arriving/being processed on the dir node.
4141 Another node would repeatedly lookup up the master, and the dir
4142 node would continue returning our nodeid until our send_remove
4145 We send another remove message in case our previous send_remove
4146 was lost/ignored/missed somehow. */
4148 if (error != -ENOTBLK) {
4149 log_limit(ls, "receive_request %x from %d %d",
4150 ms->m_lkid, from_nodeid, error);
4153 if (namelen && error == -EBADR) {
4154 send_repeat_remove(ls, ms->m_extra, namelen);
4158 setup_stub_lkb(ls, ms);
4159 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4163 static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4165 struct dlm_lkb *lkb;
4167 int error, reply = 1;
4169 error = find_lkb(ls, ms->m_remid, &lkb);
4173 if (lkb->lkb_remid != ms->m_lkid) {
4174 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4175 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4176 (unsigned long long)lkb->lkb_recover_seq,
4177 ms->m_header.h_nodeid, ms->m_lkid);
4183 r = lkb->lkb_resource;
4188 error = validate_message(lkb, ms);
4192 receive_flags(lkb, ms);
4194 error = receive_convert_args(ls, lkb, ms);
4196 send_convert_reply(r, lkb, error);
4200 reply = !down_conversion(lkb);
4202 error = do_convert(r, lkb);
4204 send_convert_reply(r, lkb, error);
4205 do_convert_effects(r, lkb, error);
4213 setup_stub_lkb(ls, ms);
4214 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4218 static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4220 struct dlm_lkb *lkb;
4224 error = find_lkb(ls, ms->m_remid, &lkb);
4228 if (lkb->lkb_remid != ms->m_lkid) {
4229 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4230 lkb->lkb_id, lkb->lkb_remid,
4231 ms->m_header.h_nodeid, ms->m_lkid);
4237 r = lkb->lkb_resource;
4242 error = validate_message(lkb, ms);
4246 receive_flags(lkb, ms);
4248 error = receive_unlock_args(ls, lkb, ms);
4250 send_unlock_reply(r, lkb, error);
4254 error = do_unlock(r, lkb);
4255 send_unlock_reply(r, lkb, error);
4256 do_unlock_effects(r, lkb, error);
4264 setup_stub_lkb(ls, ms);
4265 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4269 static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4271 struct dlm_lkb *lkb;
4275 error = find_lkb(ls, ms->m_remid, &lkb);
4279 receive_flags(lkb, ms);
4281 r = lkb->lkb_resource;
4286 error = validate_message(lkb, ms);
4290 error = do_cancel(r, lkb);
4291 send_cancel_reply(r, lkb, error);
4292 do_cancel_effects(r, lkb, error);
4300 setup_stub_lkb(ls, ms);
4301 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4305 static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4307 struct dlm_lkb *lkb;
4311 error = find_lkb(ls, ms->m_remid, &lkb);
4315 r = lkb->lkb_resource;
4320 error = validate_message(lkb, ms);
4324 receive_flags_reply(lkb, ms);
4325 if (is_altmode(lkb))
4326 munge_altmode(lkb, ms);
4327 grant_lock_pc(r, lkb, ms);
4328 queue_cast(r, lkb, 0);
4336 static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4338 struct dlm_lkb *lkb;
4342 error = find_lkb(ls, ms->m_remid, &lkb);
4346 r = lkb->lkb_resource;
4351 error = validate_message(lkb, ms);
4355 queue_bast(r, lkb, ms->m_bastmode);
4356 lkb->lkb_highbast = ms->m_bastmode;
4364 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4366 int len, error, ret_nodeid, from_nodeid, our_nodeid;
4368 from_nodeid = ms->m_header.h_nodeid;
4369 our_nodeid = dlm_our_nodeid();
4371 len = receive_extralen(ms);
4373 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4376 /* Optimization: we're master so treat lookup as a request */
4377 if (!error && ret_nodeid == our_nodeid) {
4378 receive_request(ls, ms);
4381 send_lookup_reply(ls, ms, ret_nodeid, error);
4384 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4386 char name[DLM_RESNAME_MAXLEN+1];
4389 int rv, len, dir_nodeid, from_nodeid;
4391 from_nodeid = ms->m_header.h_nodeid;
4393 len = receive_extralen(ms);
4395 if (len > DLM_RESNAME_MAXLEN) {
4396 log_error(ls, "receive_remove from %d bad len %d",
4401 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
4402 if (dir_nodeid != dlm_our_nodeid()) {
4403 log_error(ls, "receive_remove from %d bad nodeid %d",
4404 from_nodeid, dir_nodeid);
4408 /* Look for name on rsbtbl.toss, if it's there, kill it.
4409 If it's on rsbtbl.keep, it's being used, and we should ignore this
4410 message. This is an expected race between the dir node sending a
4411 request to the master node at the same time as the master node sends
4412 a remove to the dir node. The resolution to that race is for the
4413 dir node to ignore the remove message, and the master node to
4414 recreate the master rsb when it gets a request from the dir node for
4415 an rsb it doesn't have. */
4417 memset(name, 0, sizeof(name));
4418 memcpy(name, ms->m_extra, len);
4420 hash = jhash(name, len, 0);
4421 b = hash & (ls->ls_rsbtbl_size - 1);
4423 spin_lock(&ls->ls_rsbtbl[b].lock);
4425 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4427 /* verify the rsb is on keep list per comment above */
4428 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4430 /* should not happen */
4431 log_error(ls, "receive_remove from %d not found %s",
4433 spin_unlock(&ls->ls_rsbtbl[b].lock);
4436 if (r->res_master_nodeid != from_nodeid) {
4437 /* should not happen */
4438 log_error(ls, "receive_remove keep from %d master %d",
4439 from_nodeid, r->res_master_nodeid);
4441 spin_unlock(&ls->ls_rsbtbl[b].lock);
4445 log_debug(ls, "receive_remove from %d master %d first %x %s",
4446 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4448 spin_unlock(&ls->ls_rsbtbl[b].lock);
4452 if (r->res_master_nodeid != from_nodeid) {
4453 log_error(ls, "receive_remove toss from %d master %d",
4454 from_nodeid, r->res_master_nodeid);
4456 spin_unlock(&ls->ls_rsbtbl[b].lock);
4460 if (kref_put(&r->res_ref, kill_rsb)) {
4461 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4462 spin_unlock(&ls->ls_rsbtbl[b].lock);
4465 log_error(ls, "receive_remove from %d rsb ref error",
4468 spin_unlock(&ls->ls_rsbtbl[b].lock);
4472 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4474 do_purge(ls, ms->m_nodeid, ms->m_pid);
4477 static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4479 struct dlm_lkb *lkb;
4481 int error, mstype, result;
4482 int from_nodeid = ms->m_header.h_nodeid;
4484 error = find_lkb(ls, ms->m_remid, &lkb);
4488 r = lkb->lkb_resource;
4492 error = validate_message(lkb, ms);
4496 mstype = lkb->lkb_wait_type;
4497 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4499 log_error(ls, "receive_request_reply %x remote %d %x result %d",
4500 lkb->lkb_id, from_nodeid, ms->m_lkid, ms->m_result);
4505 /* Optimization: the dir node was also the master, so it took our
4506 lookup as a request and sent request reply instead of lookup reply */
4507 if (mstype == DLM_MSG_LOOKUP) {
4508 r->res_master_nodeid = from_nodeid;
4509 r->res_nodeid = from_nodeid;
4510 lkb->lkb_nodeid = from_nodeid;
4513 /* this is the value returned from do_request() on the master */
4514 result = ms->m_result;
4518 /* request would block (be queued) on remote master */
4519 queue_cast(r, lkb, -EAGAIN);
4520 confirm_master(r, -EAGAIN);
4521 unhold_lkb(lkb); /* undoes create_lkb() */
4526 /* request was queued or granted on remote master */
4527 receive_flags_reply(lkb, ms);
4528 lkb->lkb_remid = ms->m_lkid;
4529 if (is_altmode(lkb))
4530 munge_altmode(lkb, ms);
4532 add_lkb(r, lkb, DLM_LKSTS_WAITING);
4535 grant_lock_pc(r, lkb, ms);
4536 queue_cast(r, lkb, 0);
4538 confirm_master(r, result);
4543 /* find_rsb failed to find rsb or rsb wasn't master */
4544 log_limit(ls, "receive_request_reply %x from %d %d "
4545 "master %d dir %d first %x %s", lkb->lkb_id,
4546 from_nodeid, result, r->res_master_nodeid,
4547 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4549 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4550 r->res_master_nodeid != dlm_our_nodeid()) {
4551 /* cause _request_lock->set_master->send_lookup */
4552 r->res_master_nodeid = 0;
4554 lkb->lkb_nodeid = -1;
4557 if (is_overlap(lkb)) {
4558 /* we'll ignore error in cancel/unlock reply */
4559 queue_cast_overlap(r, lkb);
4560 confirm_master(r, result);
4561 unhold_lkb(lkb); /* undoes create_lkb() */
4563 _request_lock(r, lkb);
4565 if (r->res_master_nodeid == dlm_our_nodeid())
4566 confirm_master(r, 0);
4571 log_error(ls, "receive_request_reply %x error %d",
4572 lkb->lkb_id, result);
4575 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4576 log_debug(ls, "receive_request_reply %x result %d unlock",
4577 lkb->lkb_id, result);
4578 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4579 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4580 send_unlock(r, lkb);
4581 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4582 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4583 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4584 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4585 send_cancel(r, lkb);
4587 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4588 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4597 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4598 struct dlm_message *ms)
4600 /* this is the value returned from do_convert() on the master */
4601 switch (ms->m_result) {
4603 /* convert would block (be queued) on remote master */
4604 queue_cast(r, lkb, -EAGAIN);
4608 receive_flags_reply(lkb, ms);
4609 revert_lock_pc(r, lkb);
4610 queue_cast(r, lkb, -EDEADLK);
4614 /* convert was queued on remote master */
4615 receive_flags_reply(lkb, ms);
4616 if (is_demoted(lkb))
4619 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4624 /* convert was granted on remote master */
4625 receive_flags_reply(lkb, ms);
4626 if (is_demoted(lkb))
4628 grant_lock_pc(r, lkb, ms);
4629 queue_cast(r, lkb, 0);
4633 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4634 lkb->lkb_id, ms->m_header.h_nodeid, ms->m_lkid,
4641 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4643 struct dlm_rsb *r = lkb->lkb_resource;
4649 error = validate_message(lkb, ms);
4653 /* stub reply can happen with waiters_mutex held */
4654 error = remove_from_waiters_ms(lkb, ms);
4658 __receive_convert_reply(r, lkb, ms);
4664 static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4666 struct dlm_lkb *lkb;
4669 error = find_lkb(ls, ms->m_remid, &lkb);
4673 _receive_convert_reply(lkb, ms);
4678 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4680 struct dlm_rsb *r = lkb->lkb_resource;
4686 error = validate_message(lkb, ms);
4690 /* stub reply can happen with waiters_mutex held */
4691 error = remove_from_waiters_ms(lkb, ms);
4695 /* this is the value returned from do_unlock() on the master */
4697 switch (ms->m_result) {
4699 receive_flags_reply(lkb, ms);
4700 remove_lock_pc(r, lkb);
4701 queue_cast(r, lkb, -DLM_EUNLOCK);
4706 log_error(r->res_ls, "receive_unlock_reply %x error %d",
4707 lkb->lkb_id, ms->m_result);
4714 static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4716 struct dlm_lkb *lkb;
4719 error = find_lkb(ls, ms->m_remid, &lkb);
4723 _receive_unlock_reply(lkb, ms);
4728 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4730 struct dlm_rsb *r = lkb->lkb_resource;
4736 error = validate_message(lkb, ms);
4740 /* stub reply can happen with waiters_mutex held */
4741 error = remove_from_waiters_ms(lkb, ms);
4745 /* this is the value returned from do_cancel() on the master */
4747 switch (ms->m_result) {
4749 receive_flags_reply(lkb, ms);
4750 revert_lock_pc(r, lkb);
4751 queue_cast(r, lkb, -DLM_ECANCEL);
4756 log_error(r->res_ls, "receive_cancel_reply %x error %d",
4757 lkb->lkb_id, ms->m_result);
4764 static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4766 struct dlm_lkb *lkb;
4769 error = find_lkb(ls, ms->m_remid, &lkb);
4773 _receive_cancel_reply(lkb, ms);
4778 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4780 struct dlm_lkb *lkb;
4782 int error, ret_nodeid;
4783 int do_lookup_list = 0;
4785 error = find_lkb(ls, ms->m_lkid, &lkb);
4787 log_error(ls, "receive_lookup_reply no lkid %x", ms->m_lkid);
4791 /* ms->m_result is the value returned by dlm_master_lookup on dir node
4792 FIXME: will a non-zero error ever be returned? */
4794 r = lkb->lkb_resource;
4798 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4802 ret_nodeid = ms->m_nodeid;
4804 /* We sometimes receive a request from the dir node for this
4805 rsb before we've received the dir node's loookup_reply for it.
4806 The request from the dir node implies we're the master, so we set
4807 ourself as master in receive_request_reply, and verify here that
4808 we are indeed the master. */
4810 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4811 /* This should never happen */
4812 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4813 "master %d dir %d our %d first %x %s",
4814 lkb->lkb_id, ms->m_header.h_nodeid, ret_nodeid,
4815 r->res_master_nodeid, r->res_dir_nodeid,
4816 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4819 if (ret_nodeid == dlm_our_nodeid()) {
4820 r->res_master_nodeid = ret_nodeid;
4823 r->res_first_lkid = 0;
4824 } else if (ret_nodeid == -1) {
4825 /* the remote node doesn't believe it's the dir node */
4826 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4827 lkb->lkb_id, ms->m_header.h_nodeid);
4828 r->res_master_nodeid = 0;
4830 lkb->lkb_nodeid = -1;
4832 /* set_master() will set lkb_nodeid from r */
4833 r->res_master_nodeid = ret_nodeid;
4834 r->res_nodeid = ret_nodeid;
4837 if (is_overlap(lkb)) {
4838 log_debug(ls, "receive_lookup_reply %x unlock %x",
4839 lkb->lkb_id, lkb->lkb_flags);
4840 queue_cast_overlap(r, lkb);
4841 unhold_lkb(lkb); /* undoes create_lkb() */
4845 _request_lock(r, lkb);
4849 process_lookup_list(r);
4856 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4859 int error = 0, noent = 0;
4861 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
4862 log_limit(ls, "receive %d from non-member %d %x %x %d",
4863 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4864 ms->m_remid, ms->m_result);
4868 switch (ms->m_type) {
4870 /* messages sent to a master node */
4872 case DLM_MSG_REQUEST:
4873 error = receive_request(ls, ms);
4876 case DLM_MSG_CONVERT:
4877 error = receive_convert(ls, ms);
4880 case DLM_MSG_UNLOCK:
4881 error = receive_unlock(ls, ms);
4884 case DLM_MSG_CANCEL:
4886 error = receive_cancel(ls, ms);
4889 /* messages sent from a master node (replies to above) */
4891 case DLM_MSG_REQUEST_REPLY:
4892 error = receive_request_reply(ls, ms);
4895 case DLM_MSG_CONVERT_REPLY:
4896 error = receive_convert_reply(ls, ms);
4899 case DLM_MSG_UNLOCK_REPLY:
4900 error = receive_unlock_reply(ls, ms);
4903 case DLM_MSG_CANCEL_REPLY:
4904 error = receive_cancel_reply(ls, ms);
4907 /* messages sent from a master node (only two types of async msg) */
4911 error = receive_grant(ls, ms);
4916 error = receive_bast(ls, ms);
4919 /* messages sent to a dir node */
4921 case DLM_MSG_LOOKUP:
4922 receive_lookup(ls, ms);
4925 case DLM_MSG_REMOVE:
4926 receive_remove(ls, ms);
4929 /* messages sent from a dir node (remove has no reply) */
4931 case DLM_MSG_LOOKUP_REPLY:
4932 receive_lookup_reply(ls, ms);
4935 /* other messages */
4938 receive_purge(ls, ms);
4942 log_error(ls, "unknown message type %d", ms->m_type);
4946 * When checking for ENOENT, we're checking the result of
4947 * find_lkb(m_remid):
4949 * The lock id referenced in the message wasn't found. This may
4950 * happen in normal usage for the async messages and cancel, so
4951 * only use log_debug for them.
4953 * Some errors are expected and normal.
4956 if (error == -ENOENT && noent) {
4957 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4958 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4959 ms->m_lkid, saved_seq);
4960 } else if (error == -ENOENT) {
4961 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4962 ms->m_type, ms->m_remid, ms->m_header.h_nodeid,
4963 ms->m_lkid, saved_seq);
4965 if (ms->m_type == DLM_MSG_CONVERT)
4966 dlm_dump_rsb_hash(ls, ms->m_hash);
4969 if (error == -EINVAL) {
4970 log_error(ls, "receive %d inval from %d lkid %x remid %x "
4972 ms->m_type, ms->m_header.h_nodeid,
4973 ms->m_lkid, ms->m_remid, saved_seq);
4977 /* If the lockspace is in recovery mode (locking stopped), then normal
4978 messages are saved on the requestqueue for processing after recovery is
4979 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4980 messages off the requestqueue before we process new ones. This occurs right
4981 after recovery completes when we transition from saving all messages on
4982 requestqueue, to processing all the saved messages, to processing new
4983 messages as they arrive. */
4985 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4988 if (dlm_locking_stopped(ls)) {
4989 /* If we were a member of this lockspace, left, and rejoined,
4990 other nodes may still be sending us messages from the
4991 lockspace generation before we left. */
4992 if (!ls->ls_generation) {
4993 log_limit(ls, "receive %d from %d ignore old gen",
4994 ms->m_type, nodeid);
4998 dlm_add_requestqueue(ls, nodeid, ms);
5000 dlm_wait_requestqueue(ls);
5001 _receive_message(ls, ms, 0);
5005 /* This is called by dlm_recoverd to process messages that were saved on
5006 the requestqueue. */
5008 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
5011 _receive_message(ls, ms, saved_seq);
5014 /* This is called by the midcomms layer when something is received for
5015 the lockspace. It could be either a MSG (normal message sent as part of
5016 standard locking activity) or an RCOM (recovery message sent as part of
5017 lockspace recovery). */
5019 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
5021 struct dlm_header *hd = &p->header;
5025 switch (hd->h_cmd) {
5027 dlm_message_in(&p->message);
5028 type = p->message.m_type;
5031 dlm_rcom_in(&p->rcom);
5032 type = p->rcom.rc_type;
5035 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5039 if (hd->h_nodeid != nodeid) {
5040 log_print("invalid h_nodeid %d from %d lockspace %x",
5041 hd->h_nodeid, nodeid, hd->u.h_lockspace);
5045 ls = dlm_find_lockspace_global(hd->u.h_lockspace);
5047 if (dlm_config.ci_log_debug) {
5048 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5049 "%u from %d cmd %d type %d\n",
5050 hd->u.h_lockspace, nodeid, hd->h_cmd, type);
5053 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5054 dlm_send_ls_not_ready(nodeid, &p->rcom);
5058 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5059 be inactive (in this ls) before transitioning to recovery mode */
5061 down_read(&ls->ls_recv_active);
5062 if (hd->h_cmd == DLM_MSG)
5063 dlm_receive_message(ls, &p->message, nodeid);
5065 dlm_receive_rcom(ls, &p->rcom, nodeid);
5066 up_read(&ls->ls_recv_active);
5068 dlm_put_lockspace(ls);
5071 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5072 struct dlm_message *ms_stub)
5074 if (middle_conversion(lkb)) {
5076 memset(ms_stub, 0, sizeof(struct dlm_message));
5077 ms_stub->m_flags = DLM_IFL_STUB_MS;
5078 ms_stub->m_type = DLM_MSG_CONVERT_REPLY;
5079 ms_stub->m_result = -EINPROGRESS;
5080 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5081 _receive_convert_reply(lkb, ms_stub);
5083 /* Same special case as in receive_rcom_lock_args() */
5084 lkb->lkb_grmode = DLM_LOCK_IV;
5085 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5088 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5089 lkb->lkb_flags |= DLM_IFL_RESEND;
5092 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5093 conversions are async; there's no reply from the remote master */
5096 /* A waiting lkb needs recovery if the master node has failed, or
5097 the master node is changing (only when no directory is used) */
5099 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5102 if (dlm_no_directory(ls))
5105 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5111 /* Recovery for locks that are waiting for replies from nodes that are now
5112 gone. We can just complete unlocks and cancels by faking a reply from the
5113 dead node. Requests and up-conversions we flag to be resent after
5114 recovery. Down-conversions can just be completed with a fake reply like
5115 unlocks. Conversions between PR and CW need special attention. */
5117 void dlm_recover_waiters_pre(struct dlm_ls *ls)
5119 struct dlm_lkb *lkb, *safe;
5120 struct dlm_message *ms_stub;
5121 int wait_type, stub_unlock_result, stub_cancel_result;
5124 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5128 mutex_lock(&ls->ls_waiters_mutex);
5130 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5132 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5134 /* exclude debug messages about unlocks because there can be so
5135 many and they aren't very interesting */
5137 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5138 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5139 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5143 lkb->lkb_resource->res_nodeid,
5145 lkb->lkb_wait_nodeid,
5149 /* all outstanding lookups, regardless of destination will be
5150 resent after recovery is done */
5152 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5153 lkb->lkb_flags |= DLM_IFL_RESEND;
5157 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5160 wait_type = lkb->lkb_wait_type;
5161 stub_unlock_result = -DLM_EUNLOCK;
5162 stub_cancel_result = -DLM_ECANCEL;
5164 /* Main reply may have been received leaving a zero wait_type,
5165 but a reply for the overlapping op may not have been
5166 received. In that case we need to fake the appropriate
5167 reply for the overlap op. */
5170 if (is_overlap_cancel(lkb)) {
5171 wait_type = DLM_MSG_CANCEL;
5172 if (lkb->lkb_grmode == DLM_LOCK_IV)
5173 stub_cancel_result = 0;
5175 if (is_overlap_unlock(lkb)) {
5176 wait_type = DLM_MSG_UNLOCK;
5177 if (lkb->lkb_grmode == DLM_LOCK_IV)
5178 stub_unlock_result = -ENOENT;
5181 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5182 lkb->lkb_id, lkb->lkb_flags, wait_type,
5183 stub_cancel_result, stub_unlock_result);
5186 switch (wait_type) {
5188 case DLM_MSG_REQUEST:
5189 lkb->lkb_flags |= DLM_IFL_RESEND;
5192 case DLM_MSG_CONVERT:
5193 recover_convert_waiter(ls, lkb, ms_stub);
5196 case DLM_MSG_UNLOCK:
5198 memset(ms_stub, 0, sizeof(struct dlm_message));
5199 ms_stub->m_flags = DLM_IFL_STUB_MS;
5200 ms_stub->m_type = DLM_MSG_UNLOCK_REPLY;
5201 ms_stub->m_result = stub_unlock_result;
5202 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5203 _receive_unlock_reply(lkb, ms_stub);
5207 case DLM_MSG_CANCEL:
5209 memset(ms_stub, 0, sizeof(struct dlm_message));
5210 ms_stub->m_flags = DLM_IFL_STUB_MS;
5211 ms_stub->m_type = DLM_MSG_CANCEL_REPLY;
5212 ms_stub->m_result = stub_cancel_result;
5213 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
5214 _receive_cancel_reply(lkb, ms_stub);
5219 log_error(ls, "invalid lkb wait_type %d %d",
5220 lkb->lkb_wait_type, wait_type);
5224 mutex_unlock(&ls->ls_waiters_mutex);
5228 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5230 struct dlm_lkb *lkb;
5233 mutex_lock(&ls->ls_waiters_mutex);
5234 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
5235 if (lkb->lkb_flags & DLM_IFL_RESEND) {
5241 mutex_unlock(&ls->ls_waiters_mutex);
5248 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
5249 master or dir-node for r. Processing the lkb may result in it being placed
5252 /* We do this after normal locking has been enabled and any saved messages
5253 (in requestqueue) have been processed. We should be confident that at
5254 this point we won't get or process a reply to any of these waiting
5255 operations. But, new ops may be coming in on the rsbs/locks here from
5256 userspace or remotely. */
5258 /* there may have been an overlap unlock/cancel prior to recovery or after
5259 recovery. if before, the lkb may still have a pos wait_count; if after, the
5260 overlap flag would just have been set and nothing new sent. we can be
5261 confident here than any replies to either the initial op or overlap ops
5262 prior to recovery have been received. */
5264 int dlm_recover_waiters_post(struct dlm_ls *ls)
5266 struct dlm_lkb *lkb;
5268 int error = 0, mstype, err, oc, ou;
5271 if (dlm_locking_stopped(ls)) {
5272 log_debug(ls, "recover_waiters_post aborted");
5277 lkb = find_resend_waiter(ls);
5281 r = lkb->lkb_resource;
5285 mstype = lkb->lkb_wait_type;
5286 oc = is_overlap_cancel(lkb);
5287 ou = is_overlap_unlock(lkb);
5290 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5291 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5292 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5293 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5294 dlm_dir_nodeid(r), oc, ou);
5296 /* At this point we assume that we won't get a reply to any
5297 previous op or overlap op on this lock. First, do a big
5298 remove_from_waiters() for all previous ops. */
5300 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5301 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5302 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5303 lkb->lkb_wait_type = 0;
5304 lkb->lkb_wait_count = 0;
5305 mutex_lock(&ls->ls_waiters_mutex);
5306 list_del_init(&lkb->lkb_wait_reply);
5307 mutex_unlock(&ls->ls_waiters_mutex);
5308 unhold_lkb(lkb); /* for waiters list */
5311 /* do an unlock or cancel instead of resending */
5313 case DLM_MSG_LOOKUP:
5314 case DLM_MSG_REQUEST:
5315 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5317 unhold_lkb(lkb); /* undoes create_lkb() */
5319 case DLM_MSG_CONVERT:
5321 queue_cast(r, lkb, -DLM_ECANCEL);
5323 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5324 _unlock_lock(r, lkb);
5332 case DLM_MSG_LOOKUP:
5333 case DLM_MSG_REQUEST:
5334 _request_lock(r, lkb);
5336 confirm_master(r, 0);
5338 case DLM_MSG_CONVERT:
5339 _convert_lock(r, lkb);
5347 log_error(ls, "waiter %x msg %d r_nodeid %d "
5348 "dir_nodeid %d overlap %d %d",
5349 lkb->lkb_id, mstype, r->res_nodeid,
5350 dlm_dir_nodeid(r), oc, ou);
5360 static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5361 struct list_head *list)
5363 struct dlm_lkb *lkb, *safe;
5365 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5366 if (!is_master_copy(lkb))
5369 /* don't purge lkbs we've added in recover_master_copy for
5370 the current recovery seq */
5372 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5377 /* this put should free the lkb */
5378 if (!dlm_put_lkb(lkb))
5379 log_error(ls, "purged mstcpy lkb not released");
5383 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5385 struct dlm_ls *ls = r->res_ls;
5387 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5388 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5389 purge_mstcpy_list(ls, r, &r->res_waitqueue);
5392 static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5393 struct list_head *list,
5394 int nodeid_gone, unsigned int *count)
5396 struct dlm_lkb *lkb, *safe;
5398 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5399 if (!is_master_copy(lkb))
5402 if ((lkb->lkb_nodeid == nodeid_gone) ||
5403 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5405 /* tell recover_lvb to invalidate the lvb
5406 because a node holding EX/PW failed */
5407 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5408 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5409 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5414 /* this put should free the lkb */
5415 if (!dlm_put_lkb(lkb))
5416 log_error(ls, "purged dead lkb not released");
5418 rsb_set_flag(r, RSB_RECOVER_GRANT);
5425 /* Get rid of locks held by nodes that are gone. */
5427 void dlm_recover_purge(struct dlm_ls *ls)
5430 struct dlm_member *memb;
5431 int nodes_count = 0;
5432 int nodeid_gone = 0;
5433 unsigned int lkb_count = 0;
5435 /* cache one removed nodeid to optimize the common
5436 case of a single node removed */
5438 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5440 nodeid_gone = memb->nodeid;
5446 down_write(&ls->ls_root_sem);
5447 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5451 purge_dead_list(ls, r, &r->res_grantqueue,
5452 nodeid_gone, &lkb_count);
5453 purge_dead_list(ls, r, &r->res_convertqueue,
5454 nodeid_gone, &lkb_count);
5455 purge_dead_list(ls, r, &r->res_waitqueue,
5456 nodeid_gone, &lkb_count);
5462 up_write(&ls->ls_root_sem);
5465 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5466 lkb_count, nodes_count);
5469 static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5474 spin_lock(&ls->ls_rsbtbl[bucket].lock);
5475 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5476 r = rb_entry(n, struct dlm_rsb, res_hashnode);
5478 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5480 if (!is_master(r)) {
5481 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5485 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5488 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5493 * Attempt to grant locks on resources that we are the master of.
5494 * Locks may have become grantable during recovery because locks
5495 * from departed nodes have been purged (or not rebuilt), allowing
5496 * previously blocked locks to now be granted. The subset of rsb's
5497 * we are interested in are those with lkb's on either the convert or
5500 * Simplest would be to go through each master rsb and check for non-empty
5501 * convert or waiting queues, and attempt to grant on those rsbs.
5502 * Checking the queues requires lock_rsb, though, for which we'd need
5503 * to release the rsbtbl lock. This would make iterating through all
5504 * rsb's very inefficient. So, we rely on earlier recovery routines
5505 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5509 void dlm_recover_grant(struct dlm_ls *ls)
5513 unsigned int count = 0;
5514 unsigned int rsb_count = 0;
5515 unsigned int lkb_count = 0;
5518 r = find_grant_rsb(ls, bucket);
5520 if (bucket == ls->ls_rsbtbl_size - 1)
5528 /* the RECOVER_GRANT flag is checked in the grant path */
5529 grant_pending_locks(r, &count);
5530 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5532 confirm_master(r, 0);
5539 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5540 lkb_count, rsb_count);
5543 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5546 struct dlm_lkb *lkb;
5548 list_for_each_entry(lkb, head, lkb_statequeue) {
5549 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5555 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5558 struct dlm_lkb *lkb;
5560 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5563 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5566 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5572 /* needs at least dlm_rcom + rcom_lock */
5573 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5574 struct dlm_rsb *r, struct dlm_rcom *rc)
5576 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5578 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
5579 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5580 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5581 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5582 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5583 lkb->lkb_flags |= DLM_IFL_MSTCPY;
5584 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5585 lkb->lkb_rqmode = rl->rl_rqmode;
5586 lkb->lkb_grmode = rl->rl_grmode;
5587 /* don't set lkb_status because add_lkb wants to itself */
5589 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5590 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5592 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5593 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
5594 sizeof(struct rcom_lock);
5595 if (lvblen > ls->ls_lvblen)
5597 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5598 if (!lkb->lkb_lvbptr)
5600 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5603 /* Conversions between PR and CW (middle modes) need special handling.
5604 The real granted mode of these converting locks cannot be determined
5605 until all locks have been rebuilt on the rsb (recover_conversion) */
5607 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5608 middle_conversion(lkb)) {
5609 rl->rl_status = DLM_LKSTS_CONVERT;
5610 lkb->lkb_grmode = DLM_LOCK_IV;
5611 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5617 /* This lkb may have been recovered in a previous aborted recovery so we need
5618 to check if the rsb already has an lkb with the given remote nodeid/lkid.
5619 If so we just send back a standard reply. If not, we create a new lkb with
5620 the given values and send back our lkid. We send back our lkid by sending
5621 back the rcom_lock struct we got but with the remid field filled in. */
5623 /* needs at least dlm_rcom + rcom_lock */
5624 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5626 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5628 struct dlm_lkb *lkb;
5630 int from_nodeid = rc->rc_header.h_nodeid;
5633 if (rl->rl_parent_lkid) {
5634 error = -EOPNOTSUPP;
5638 remid = le32_to_cpu(rl->rl_lkid);
5640 /* In general we expect the rsb returned to be R_MASTER, but we don't
5641 have to require it. Recovery of masters on one node can overlap
5642 recovery of locks on another node, so one node can send us MSTCPY
5643 locks before we've made ourselves master of this rsb. We can still
5644 add new MSTCPY locks that we receive here without any harm; when
5645 we make ourselves master, dlm_recover_masters() won't touch the
5646 MSTCPY locks we've received early. */
5648 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5649 from_nodeid, R_RECEIVE_RECOVER, &r);
5655 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5656 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5657 from_nodeid, remid);
5662 lkb = search_remid(r, from_nodeid, remid);
5668 error = create_lkb(ls, &lkb);
5672 error = receive_rcom_lock_args(ls, lkb, r, rc);
5679 add_lkb(r, lkb, rl->rl_status);
5681 ls->ls_recover_locks_in++;
5683 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5684 rsb_set_flag(r, RSB_RECOVER_GRANT);
5687 /* this is the new value returned to the lock holder for
5688 saving in its process-copy lkb */
5689 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5691 lkb->lkb_recover_seq = ls->ls_recover_seq;
5697 if (error && error != -EEXIST)
5698 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5699 from_nodeid, remid, error);
5700 rl->rl_result = cpu_to_le32(error);
5704 /* needs at least dlm_rcom + rcom_lock */
5705 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5707 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5709 struct dlm_lkb *lkb;
5710 uint32_t lkid, remid;
5713 lkid = le32_to_cpu(rl->rl_lkid);
5714 remid = le32_to_cpu(rl->rl_remid);
5715 result = le32_to_cpu(rl->rl_result);
5717 error = find_lkb(ls, lkid, &lkb);
5719 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5720 lkid, rc->rc_header.h_nodeid, remid, result);
5724 r = lkb->lkb_resource;
5728 if (!is_process_copy(lkb)) {
5729 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5730 lkid, rc->rc_header.h_nodeid, remid, result);
5740 /* There's a chance the new master received our lock before
5741 dlm_recover_master_reply(), this wouldn't happen if we did
5742 a barrier between recover_masters and recover_locks. */
5744 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5745 lkid, rc->rc_header.h_nodeid, remid, result);
5747 dlm_send_rcom_lock(r, lkb);
5751 lkb->lkb_remid = remid;
5754 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5755 lkid, rc->rc_header.h_nodeid, remid, result);
5758 /* an ack for dlm_recover_locks() which waits for replies from
5759 all the locks it sends to new masters */
5760 dlm_recovered_lock(r);
5769 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5770 int mode, uint32_t flags, void *name, unsigned int namelen,
5771 unsigned long timeout_cs)
5773 struct dlm_lkb *lkb;
5774 struct dlm_args args;
5777 dlm_lock_recovery(ls);
5779 error = create_lkb(ls, &lkb);
5785 if (flags & DLM_LKF_VALBLK) {
5786 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5787 if (!ua->lksb.sb_lvbptr) {
5794 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5795 fake_astfn, ua, fake_bastfn, &args);
5797 kfree(ua->lksb.sb_lvbptr);
5798 ua->lksb.sb_lvbptr = NULL;
5804 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5805 When DLM_IFL_USER is set, the dlm knows that this is a userspace
5806 lock and that lkb_astparam is the dlm_user_args structure. */
5807 lkb->lkb_flags |= DLM_IFL_USER;
5808 error = request_lock(ls, lkb, name, namelen, &args);
5824 /* add this new lkb to the per-process list of locks */
5825 spin_lock(&ua->proc->locks_spin);
5827 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5828 spin_unlock(&ua->proc->locks_spin);
5830 dlm_unlock_recovery(ls);
5834 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5835 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5836 unsigned long timeout_cs)
5838 struct dlm_lkb *lkb;
5839 struct dlm_args args;
5840 struct dlm_user_args *ua;
5843 dlm_lock_recovery(ls);
5845 error = find_lkb(ls, lkid, &lkb);
5849 /* user can change the params on its lock when it converts it, or
5850 add an lvb that didn't exist before */
5854 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5855 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5856 if (!ua->lksb.sb_lvbptr) {
5861 if (lvb_in && ua->lksb.sb_lvbptr)
5862 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5864 ua->xid = ua_tmp->xid;
5865 ua->castparam = ua_tmp->castparam;
5866 ua->castaddr = ua_tmp->castaddr;
5867 ua->bastparam = ua_tmp->bastparam;
5868 ua->bastaddr = ua_tmp->bastaddr;
5869 ua->user_lksb = ua_tmp->user_lksb;
5871 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5872 fake_astfn, ua, fake_bastfn, &args);
5876 error = convert_lock(ls, lkb, &args);
5878 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5883 dlm_unlock_recovery(ls);
5889 * The caller asks for an orphan lock on a given resource with a given mode.
5890 * If a matching lock exists, it's moved to the owner's list of locks and
5891 * the lkid is returned.
5894 int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5895 int mode, uint32_t flags, void *name, unsigned int namelen,
5896 unsigned long timeout_cs, uint32_t *lkid)
5898 struct dlm_lkb *lkb;
5899 struct dlm_user_args *ua;
5900 int found_other_mode = 0;
5904 mutex_lock(&ls->ls_orphans_mutex);
5905 list_for_each_entry(lkb, &ls->ls_orphans, lkb_ownqueue) {
5906 if (lkb->lkb_resource->res_length != namelen)
5908 if (memcmp(lkb->lkb_resource->res_name, name, namelen))
5910 if (lkb->lkb_grmode != mode) {
5911 found_other_mode = 1;
5916 list_del_init(&lkb->lkb_ownqueue);
5917 lkb->lkb_flags &= ~DLM_IFL_ORPHAN;
5918 *lkid = lkb->lkb_id;
5921 mutex_unlock(&ls->ls_orphans_mutex);
5923 if (!found && found_other_mode) {
5933 lkb->lkb_exflags = flags;
5934 lkb->lkb_ownpid = (int) current->pid;
5938 ua->proc = ua_tmp->proc;
5939 ua->xid = ua_tmp->xid;
5940 ua->castparam = ua_tmp->castparam;
5941 ua->castaddr = ua_tmp->castaddr;
5942 ua->bastparam = ua_tmp->bastparam;
5943 ua->bastaddr = ua_tmp->bastaddr;
5944 ua->user_lksb = ua_tmp->user_lksb;
5947 * The lkb reference from the ls_orphans list was not
5948 * removed above, and is now considered the reference
5949 * for the proc locks list.
5952 spin_lock(&ua->proc->locks_spin);
5953 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5954 spin_unlock(&ua->proc->locks_spin);
5960 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5961 uint32_t flags, uint32_t lkid, char *lvb_in)
5963 struct dlm_lkb *lkb;
5964 struct dlm_args args;
5965 struct dlm_user_args *ua;
5968 dlm_lock_recovery(ls);
5970 error = find_lkb(ls, lkid, &lkb);
5976 if (lvb_in && ua->lksb.sb_lvbptr)
5977 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5978 if (ua_tmp->castparam)
5979 ua->castparam = ua_tmp->castparam;
5980 ua->user_lksb = ua_tmp->user_lksb;
5982 error = set_unlock_args(flags, ua, &args);
5986 error = unlock_lock(ls, lkb, &args);
5988 if (error == -DLM_EUNLOCK)
5990 /* from validate_unlock_args() */
5991 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
5996 spin_lock(&ua->proc->locks_spin);
5997 /* dlm_user_add_cb() may have already taken lkb off the proc list */
5998 if (!list_empty(&lkb->lkb_ownqueue))
5999 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
6000 spin_unlock(&ua->proc->locks_spin);
6004 dlm_unlock_recovery(ls);
6009 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6010 uint32_t flags, uint32_t lkid)
6012 struct dlm_lkb *lkb;
6013 struct dlm_args args;
6014 struct dlm_user_args *ua;
6017 dlm_lock_recovery(ls);
6019 error = find_lkb(ls, lkid, &lkb);
6024 if (ua_tmp->castparam)
6025 ua->castparam = ua_tmp->castparam;
6026 ua->user_lksb = ua_tmp->user_lksb;
6028 error = set_unlock_args(flags, ua, &args);
6032 error = cancel_lock(ls, lkb, &args);
6034 if (error == -DLM_ECANCEL)
6036 /* from validate_unlock_args() */
6037 if (error == -EBUSY)
6042 dlm_unlock_recovery(ls);
6047 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6049 struct dlm_lkb *lkb;
6050 struct dlm_args args;
6051 struct dlm_user_args *ua;
6055 dlm_lock_recovery(ls);
6057 error = find_lkb(ls, lkid, &lkb);
6063 error = set_unlock_args(flags, ua, &args);
6067 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6069 r = lkb->lkb_resource;
6073 error = validate_unlock_args(lkb, &args);
6076 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6078 error = _cancel_lock(r, lkb);
6083 if (error == -DLM_ECANCEL)
6085 /* from validate_unlock_args() */
6086 if (error == -EBUSY)
6091 dlm_unlock_recovery(ls);
6095 /* lkb's that are removed from the waiters list by revert are just left on the
6096 orphans list with the granted orphan locks, to be freed by purge */
6098 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6100 struct dlm_args args;
6103 hold_lkb(lkb); /* reference for the ls_orphans list */
6104 mutex_lock(&ls->ls_orphans_mutex);
6105 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6106 mutex_unlock(&ls->ls_orphans_mutex);
6108 set_unlock_args(0, lkb->lkb_ua, &args);
6110 error = cancel_lock(ls, lkb, &args);
6111 if (error == -DLM_ECANCEL)
6116 /* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6117 granted. Regardless of what rsb queue the lock is on, it's removed and
6118 freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6119 if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6121 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6123 struct dlm_args args;
6126 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6127 lkb->lkb_ua, &args);
6129 error = unlock_lock(ls, lkb, &args);
6130 if (error == -DLM_EUNLOCK)
6135 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6136 (which does lock_rsb) due to deadlock with receiving a message that does
6137 lock_rsb followed by dlm_user_add_cb() */
6139 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6140 struct dlm_user_proc *proc)
6142 struct dlm_lkb *lkb = NULL;
6144 mutex_lock(&ls->ls_clear_proc_locks);
6145 if (list_empty(&proc->locks))
6148 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6149 list_del_init(&lkb->lkb_ownqueue);
6151 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6152 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6154 lkb->lkb_flags |= DLM_IFL_DEAD;
6156 mutex_unlock(&ls->ls_clear_proc_locks);
6160 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6161 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6162 which we clear here. */
6164 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
6165 list, and no more device_writes should add lkb's to proc->locks list; so we
6166 shouldn't need to take asts_spin or locks_spin here. this assumes that
6167 device reads/writes/closes are serialized -- FIXME: we may need to serialize
6170 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6172 struct dlm_lkb *lkb, *safe;
6174 dlm_lock_recovery(ls);
6177 lkb = del_proc_lock(ls, proc);
6181 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6182 orphan_proc_lock(ls, lkb);
6184 unlock_proc_lock(ls, lkb);
6186 /* this removes the reference for the proc->locks list
6187 added by dlm_user_request, it may result in the lkb
6193 mutex_lock(&ls->ls_clear_proc_locks);
6195 /* in-progress unlocks */
6196 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6197 list_del_init(&lkb->lkb_ownqueue);
6198 lkb->lkb_flags |= DLM_IFL_DEAD;
6202 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6203 memset(&lkb->lkb_callbacks, 0,
6204 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6205 list_del_init(&lkb->lkb_cb_list);
6209 mutex_unlock(&ls->ls_clear_proc_locks);
6210 dlm_unlock_recovery(ls);
6213 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6215 struct dlm_lkb *lkb, *safe;
6219 spin_lock(&proc->locks_spin);
6220 if (!list_empty(&proc->locks)) {
6221 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6223 list_del_init(&lkb->lkb_ownqueue);
6225 spin_unlock(&proc->locks_spin);
6230 lkb->lkb_flags |= DLM_IFL_DEAD;
6231 unlock_proc_lock(ls, lkb);
6232 dlm_put_lkb(lkb); /* ref from proc->locks list */
6235 spin_lock(&proc->locks_spin);
6236 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6237 list_del_init(&lkb->lkb_ownqueue);
6238 lkb->lkb_flags |= DLM_IFL_DEAD;
6241 spin_unlock(&proc->locks_spin);
6243 spin_lock(&proc->asts_spin);
6244 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6245 memset(&lkb->lkb_callbacks, 0,
6246 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6247 list_del_init(&lkb->lkb_cb_list);
6250 spin_unlock(&proc->asts_spin);
6253 /* pid of 0 means purge all orphans */
6255 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6257 struct dlm_lkb *lkb, *safe;
6259 mutex_lock(&ls->ls_orphans_mutex);
6260 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6261 if (pid && lkb->lkb_ownpid != pid)
6263 unlock_proc_lock(ls, lkb);
6264 list_del_init(&lkb->lkb_ownqueue);
6267 mutex_unlock(&ls->ls_orphans_mutex);
6270 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6272 struct dlm_message *ms;
6273 struct dlm_mhandle *mh;
6276 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6277 DLM_MSG_PURGE, &ms, &mh);
6280 ms->m_nodeid = nodeid;
6283 return send_message(mh, ms);
6286 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6287 int nodeid, int pid)
6291 if (nodeid && (nodeid != dlm_our_nodeid())) {
6292 error = send_purge(ls, nodeid, pid);
6294 dlm_lock_recovery(ls);
6295 if (pid == current->pid)
6296 purge_proc_locks(ls, proc);
6298 do_purge(ls, nodeid, pid);
6299 dlm_unlock_recovery(ls);