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 <trace/events/dlm.h>
58 #include <linux/types.h>
59 #include <linux/rbtree.h>
60 #include <linux/slab.h>
61 #include "dlm_internal.h"
62 #include <linux/dlm_device.h>
65 #include "requestqueue.h"
69 #include "lockspace.h"
74 #include "lvb_table.h"
78 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
79 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
80 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
81 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
82 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
83 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
84 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
85 static int send_remove(struct dlm_rsb *r);
86 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
87 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
88 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
89 struct dlm_message *ms);
90 static int receive_extralen(struct dlm_message *ms);
91 static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
92 static void del_timeout(struct dlm_lkb *lkb);
93 static void toss_rsb(struct kref *kref);
96 * Lock compatibilty matrix - thanks Steve
97 * UN = Unlocked state. Not really a state, used as a flag
98 * PD = Padding. Used to make the matrix a nice power of two in size
99 * Other states are the same as the VMS DLM.
100 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
103 static const int __dlm_compat_matrix[8][8] = {
104 /* UN NL CR CW PR PW EX PD */
105 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
106 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
107 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
108 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
109 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
110 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
111 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
112 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
116 * This defines the direction of transfer of LVB data.
117 * Granted mode is the row; requested mode is the column.
118 * Usage: matrix[grmode+1][rqmode+1]
119 * 1 = LVB is returned to the caller
120 * 0 = LVB is written to the resource
121 * -1 = nothing happens to the LVB
124 const int dlm_lvb_operations[8][8] = {
125 /* UN NL CR CW PR PW EX PD*/
126 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
127 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
128 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
129 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
130 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
131 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
132 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
133 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
136 #define modes_compat(gr, rq) \
137 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
139 int dlm_modes_compat(int mode1, int mode2)
141 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
145 * Compatibility matrix for conversions with QUECVT set.
146 * Granted mode is the row; requested mode is the column.
147 * Usage: matrix[grmode+1][rqmode+1]
150 static const int __quecvt_compat_matrix[8][8] = {
151 /* UN NL CR CW PR PW EX PD */
152 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
153 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
154 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
155 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
156 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
157 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
158 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
159 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
162 void dlm_print_lkb(struct dlm_lkb *lkb)
164 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
165 "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
166 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
167 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
168 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
169 (unsigned long long)lkb->lkb_recover_seq);
172 static void dlm_print_rsb(struct dlm_rsb *r)
174 printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
176 r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
177 r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
181 void dlm_dump_rsb(struct dlm_rsb *r)
187 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
188 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
189 printk(KERN_ERR "rsb lookup list\n");
190 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
192 printk(KERN_ERR "rsb grant queue:\n");
193 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
195 printk(KERN_ERR "rsb convert queue:\n");
196 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
198 printk(KERN_ERR "rsb wait queue:\n");
199 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
203 /* Threads cannot use the lockspace while it's being recovered */
205 static inline void dlm_lock_recovery(struct dlm_ls *ls)
207 down_read(&ls->ls_in_recovery);
210 void dlm_unlock_recovery(struct dlm_ls *ls)
212 up_read(&ls->ls_in_recovery);
215 int dlm_lock_recovery_try(struct dlm_ls *ls)
217 return down_read_trylock(&ls->ls_in_recovery);
220 static inline int can_be_queued(struct dlm_lkb *lkb)
222 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
225 static inline int force_blocking_asts(struct dlm_lkb *lkb)
227 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
230 static inline int is_demoted(struct dlm_lkb *lkb)
232 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
235 static inline int is_altmode(struct dlm_lkb *lkb)
237 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
240 static inline int is_granted(struct dlm_lkb *lkb)
242 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
245 static inline int is_remote(struct dlm_rsb *r)
247 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
248 return !!r->res_nodeid;
251 static inline int is_process_copy(struct dlm_lkb *lkb)
253 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
256 static inline int is_master_copy(struct dlm_lkb *lkb)
258 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
261 static inline int middle_conversion(struct dlm_lkb *lkb)
263 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
264 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
269 static inline int down_conversion(struct dlm_lkb *lkb)
271 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
274 static inline int is_overlap_unlock(struct dlm_lkb *lkb)
276 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
279 static inline int is_overlap_cancel(struct dlm_lkb *lkb)
281 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
284 static inline int is_overlap(struct dlm_lkb *lkb)
286 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
287 DLM_IFL_OVERLAP_CANCEL));
290 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
292 if (is_master_copy(lkb))
297 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
299 #ifdef CONFIG_DLM_DEPRECATED_API
300 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
301 timeout caused the cancel then return -ETIMEDOUT */
302 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
303 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
308 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
309 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
313 dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
316 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
319 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
322 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
324 if (is_master_copy(lkb)) {
325 send_bast(r, lkb, rqmode);
327 dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
332 * Basic operations on rsb's and lkb's
335 /* This is only called to add a reference when the code already holds
336 a valid reference to the rsb, so there's no need for locking. */
338 static inline void hold_rsb(struct dlm_rsb *r)
340 kref_get(&r->res_ref);
343 void dlm_hold_rsb(struct dlm_rsb *r)
348 /* When all references to the rsb are gone it's transferred to
349 the tossed list for later disposal. */
351 static void put_rsb(struct dlm_rsb *r)
353 struct dlm_ls *ls = r->res_ls;
354 uint32_t bucket = r->res_bucket;
357 rv = kref_put_lock(&r->res_ref, toss_rsb,
358 &ls->ls_rsbtbl[bucket].lock);
360 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
363 void dlm_put_rsb(struct dlm_rsb *r)
368 static int pre_rsb_struct(struct dlm_ls *ls)
370 struct dlm_rsb *r1, *r2;
373 spin_lock(&ls->ls_new_rsb_spin);
374 if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
375 spin_unlock(&ls->ls_new_rsb_spin);
378 spin_unlock(&ls->ls_new_rsb_spin);
380 r1 = dlm_allocate_rsb(ls);
381 r2 = dlm_allocate_rsb(ls);
383 spin_lock(&ls->ls_new_rsb_spin);
385 list_add(&r1->res_hashchain, &ls->ls_new_rsb);
386 ls->ls_new_rsb_count++;
389 list_add(&r2->res_hashchain, &ls->ls_new_rsb);
390 ls->ls_new_rsb_count++;
392 count = ls->ls_new_rsb_count;
393 spin_unlock(&ls->ls_new_rsb_spin);
400 /* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
401 unlock any spinlocks, go back and call pre_rsb_struct again.
402 Otherwise, take an rsb off the list and return it. */
404 static int get_rsb_struct(struct dlm_ls *ls, const void *name, int len,
405 struct dlm_rsb **r_ret)
410 spin_lock(&ls->ls_new_rsb_spin);
411 if (list_empty(&ls->ls_new_rsb)) {
412 count = ls->ls_new_rsb_count;
413 spin_unlock(&ls->ls_new_rsb_spin);
414 log_debug(ls, "find_rsb retry %d %d %s",
415 count, dlm_config.ci_new_rsb_count,
420 r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
421 list_del(&r->res_hashchain);
422 /* Convert the empty list_head to a NULL rb_node for tree usage: */
423 memset(&r->res_hashnode, 0, sizeof(struct rb_node));
424 ls->ls_new_rsb_count--;
425 spin_unlock(&ls->ls_new_rsb_spin);
429 memcpy(r->res_name, name, len);
430 mutex_init(&r->res_mutex);
432 INIT_LIST_HEAD(&r->res_lookup);
433 INIT_LIST_HEAD(&r->res_grantqueue);
434 INIT_LIST_HEAD(&r->res_convertqueue);
435 INIT_LIST_HEAD(&r->res_waitqueue);
436 INIT_LIST_HEAD(&r->res_root_list);
437 INIT_LIST_HEAD(&r->res_recover_list);
443 static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
445 char maxname[DLM_RESNAME_MAXLEN];
447 memset(maxname, 0, DLM_RESNAME_MAXLEN);
448 memcpy(maxname, name, nlen);
449 return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
452 int dlm_search_rsb_tree(struct rb_root *tree, const void *name, int len,
453 struct dlm_rsb **r_ret)
455 struct rb_node *node = tree->rb_node;
460 r = rb_entry(node, struct dlm_rsb, res_hashnode);
461 rc = rsb_cmp(r, name, len);
463 node = node->rb_left;
465 node = node->rb_right;
477 static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
479 struct rb_node **newn = &tree->rb_node;
480 struct rb_node *parent = NULL;
484 struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
488 rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
490 newn = &parent->rb_left;
492 newn = &parent->rb_right;
494 log_print("rsb_insert match");
501 rb_link_node(&rsb->res_hashnode, parent, newn);
502 rb_insert_color(&rsb->res_hashnode, tree);
507 * Find rsb in rsbtbl and potentially create/add one
509 * Delaying the release of rsb's has a similar benefit to applications keeping
510 * NL locks on an rsb, but without the guarantee that the cached master value
511 * will still be valid when the rsb is reused. Apps aren't always smart enough
512 * to keep NL locks on an rsb that they may lock again shortly; this can lead
513 * to excessive master lookups and removals if we don't delay the release.
515 * Searching for an rsb means looking through both the normal list and toss
516 * list. When found on the toss list the rsb is moved to the normal list with
517 * ref count of 1; when found on normal list the ref count is incremented.
519 * rsb's on the keep list are being used locally and refcounted.
520 * rsb's on the toss list are not being used locally, and are not refcounted.
522 * The toss list rsb's were either
523 * - previously used locally but not any more (were on keep list, then
524 * moved to toss list when last refcount dropped)
525 * - created and put on toss list as a directory record for a lookup
526 * (we are the dir node for the res, but are not using the res right now,
527 * but some other node is)
529 * The purpose of find_rsb() is to return a refcounted rsb for local use.
530 * So, if the given rsb is on the toss list, it is moved to the keep list
531 * before being returned.
533 * toss_rsb() happens when all local usage of the rsb is done, i.e. no
534 * more refcounts exist, so the rsb is moved from the keep list to the
537 * rsb's on both keep and toss lists are used for doing a name to master
538 * lookups. rsb's that are in use locally (and being refcounted) are on
539 * the keep list, rsb's that are not in use locally (not refcounted) and
540 * only exist for name/master lookups are on the toss list.
542 * rsb's on the toss list who's dir_nodeid is not local can have stale
543 * name/master mappings. So, remote requests on such rsb's can potentially
544 * return with an error, which means the mapping is stale and needs to
545 * be updated with a new lookup. (The idea behind MASTER UNCERTAIN and
546 * first_lkid is to keep only a single outstanding request on an rsb
547 * while that rsb has a potentially stale master.)
550 static int find_rsb_dir(struct dlm_ls *ls, const void *name, int len,
551 uint32_t hash, uint32_t b,
552 int dir_nodeid, int from_nodeid,
553 unsigned int flags, struct dlm_rsb **r_ret)
555 struct dlm_rsb *r = NULL;
556 int our_nodeid = dlm_our_nodeid();
563 if (flags & R_RECEIVE_REQUEST) {
564 if (from_nodeid == dir_nodeid)
568 } else if (flags & R_REQUEST) {
573 * flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
574 * from_nodeid has sent us a lock in dlm_recover_locks, believing
575 * we're the new master. Our local recovery may not have set
576 * res_master_nodeid to our_nodeid yet, so allow either. Don't
577 * create the rsb; dlm_recover_process_copy() will handle EBADR
580 * If someone sends us a request, we are the dir node, and we do
581 * not find the rsb anywhere, then recreate it. This happens if
582 * someone sends us a request after we have removed/freed an rsb
583 * from our toss list. (They sent a request instead of lookup
584 * because they are using an rsb from their toss list.)
587 if (from_local || from_dir ||
588 (from_other && (dir_nodeid == our_nodeid))) {
594 error = pre_rsb_struct(ls);
599 spin_lock(&ls->ls_rsbtbl[b].lock);
601 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
606 * rsb is active, so we can't check master_nodeid without lock_rsb.
609 kref_get(&r->res_ref);
614 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
619 * rsb found inactive (master_nodeid may be out of date unless
620 * we are the dir_nodeid or were the master) No other thread
621 * is using this rsb because it's on the toss list, so we can
622 * look at or update res_master_nodeid without lock_rsb.
625 if ((r->res_master_nodeid != our_nodeid) && from_other) {
626 /* our rsb was not master, and another node (not the dir node)
627 has sent us a request */
628 log_debug(ls, "find_rsb toss from_other %d master %d dir %d %s",
629 from_nodeid, r->res_master_nodeid, dir_nodeid,
635 if ((r->res_master_nodeid != our_nodeid) && from_dir) {
636 /* don't think this should ever happen */
637 log_error(ls, "find_rsb toss from_dir %d master %d",
638 from_nodeid, r->res_master_nodeid);
640 /* fix it and go on */
641 r->res_master_nodeid = our_nodeid;
643 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
644 r->res_first_lkid = 0;
647 if (from_local && (r->res_master_nodeid != our_nodeid)) {
648 /* Because we have held no locks on this rsb,
649 res_master_nodeid could have become stale. */
650 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
651 r->res_first_lkid = 0;
654 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
655 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
664 if (error == -EBADR && !create)
667 error = get_rsb_struct(ls, name, len, &r);
668 if (error == -EAGAIN) {
669 spin_unlock(&ls->ls_rsbtbl[b].lock);
677 r->res_dir_nodeid = dir_nodeid;
678 kref_init(&r->res_ref);
681 /* want to see how often this happens */
682 log_debug(ls, "find_rsb new from_dir %d recreate %s",
683 from_nodeid, r->res_name);
684 r->res_master_nodeid = our_nodeid;
689 if (from_other && (dir_nodeid != our_nodeid)) {
690 /* should never happen */
691 log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
692 from_nodeid, dir_nodeid, our_nodeid, r->res_name);
700 log_debug(ls, "find_rsb new from_other %d dir %d %s",
701 from_nodeid, dir_nodeid, r->res_name);
704 if (dir_nodeid == our_nodeid) {
705 /* When we are the dir nodeid, we can set the master
707 r->res_master_nodeid = our_nodeid;
710 /* set_master will send_lookup to dir_nodeid */
711 r->res_master_nodeid = 0;
716 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
718 spin_unlock(&ls->ls_rsbtbl[b].lock);
724 /* During recovery, other nodes can send us new MSTCPY locks (from
725 dlm_recover_locks) before we've made ourself master (in
726 dlm_recover_masters). */
728 static int find_rsb_nodir(struct dlm_ls *ls, const void *name, int len,
729 uint32_t hash, uint32_t b,
730 int dir_nodeid, int from_nodeid,
731 unsigned int flags, struct dlm_rsb **r_ret)
733 struct dlm_rsb *r = NULL;
734 int our_nodeid = dlm_our_nodeid();
735 int recover = (flags & R_RECEIVE_RECOVER);
739 error = pre_rsb_struct(ls);
743 spin_lock(&ls->ls_rsbtbl[b].lock);
745 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
750 * rsb is active, so we can't check master_nodeid without lock_rsb.
753 kref_get(&r->res_ref);
758 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
763 * rsb found inactive. No other thread is using this rsb because
764 * it's on the toss list, so we can look at or update
765 * res_master_nodeid without lock_rsb.
768 if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
769 /* our rsb is not master, and another node has sent us a
770 request; this should never happen */
771 log_error(ls, "find_rsb toss from_nodeid %d master %d dir %d",
772 from_nodeid, r->res_master_nodeid, dir_nodeid);
778 if (!recover && (r->res_master_nodeid != our_nodeid) &&
779 (dir_nodeid == our_nodeid)) {
780 /* our rsb is not master, and we are dir; may as well fix it;
781 this should never happen */
782 log_error(ls, "find_rsb toss our %d master %d dir %d",
783 our_nodeid, r->res_master_nodeid, dir_nodeid);
785 r->res_master_nodeid = our_nodeid;
789 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
790 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
799 error = get_rsb_struct(ls, name, len, &r);
800 if (error == -EAGAIN) {
801 spin_unlock(&ls->ls_rsbtbl[b].lock);
809 r->res_dir_nodeid = dir_nodeid;
810 r->res_master_nodeid = dir_nodeid;
811 r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
812 kref_init(&r->res_ref);
814 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
816 spin_unlock(&ls->ls_rsbtbl[b].lock);
822 static int find_rsb(struct dlm_ls *ls, const void *name, int len,
823 int from_nodeid, unsigned int flags,
824 struct dlm_rsb **r_ret)
829 if (len > DLM_RESNAME_MAXLEN)
832 hash = jhash(name, len, 0);
833 b = hash & (ls->ls_rsbtbl_size - 1);
835 dir_nodeid = dlm_hash2nodeid(ls, hash);
837 if (dlm_no_directory(ls))
838 return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
839 from_nodeid, flags, r_ret);
841 return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
842 from_nodeid, flags, r_ret);
845 /* we have received a request and found that res_master_nodeid != our_nodeid,
846 so we need to return an error or make ourself the master */
848 static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
851 if (dlm_no_directory(ls)) {
852 log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
853 from_nodeid, r->res_master_nodeid,
859 if (from_nodeid != r->res_dir_nodeid) {
860 /* our rsb is not master, and another node (not the dir node)
861 has sent us a request. this is much more common when our
862 master_nodeid is zero, so limit debug to non-zero. */
864 if (r->res_master_nodeid) {
865 log_debug(ls, "validate master from_other %d master %d "
866 "dir %d first %x %s", from_nodeid,
867 r->res_master_nodeid, r->res_dir_nodeid,
868 r->res_first_lkid, r->res_name);
872 /* our rsb is not master, but the dir nodeid has sent us a
873 request; this could happen with master 0 / res_nodeid -1 */
875 if (r->res_master_nodeid) {
876 log_error(ls, "validate master from_dir %d master %d "
878 from_nodeid, r->res_master_nodeid,
879 r->res_first_lkid, r->res_name);
882 r->res_master_nodeid = dlm_our_nodeid();
888 static void __dlm_master_lookup(struct dlm_ls *ls, struct dlm_rsb *r, int our_nodeid,
889 int from_nodeid, bool toss_list, unsigned int flags,
890 int *r_nodeid, int *result)
892 int fix_master = (flags & DLM_LU_RECOVER_MASTER);
893 int from_master = (flags & DLM_LU_RECOVER_DIR);
895 if (r->res_dir_nodeid != our_nodeid) {
896 /* should not happen, but may as well fix it and carry on */
897 log_error(ls, "%s res_dir %d our %d %s", __func__,
898 r->res_dir_nodeid, our_nodeid, r->res_name);
899 r->res_dir_nodeid = our_nodeid;
902 if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
903 /* Recovery uses this function to set a new master when
904 * the previous master failed. Setting NEW_MASTER will
905 * force dlm_recover_masters to call recover_master on this
906 * rsb even though the res_nodeid is no longer removed.
909 r->res_master_nodeid = from_nodeid;
910 r->res_nodeid = from_nodeid;
911 rsb_set_flag(r, RSB_NEW_MASTER);
914 /* I don't think we should ever find it on toss list. */
915 log_error(ls, "%s fix_master on toss", __func__);
920 if (from_master && (r->res_master_nodeid != from_nodeid)) {
921 /* this will happen if from_nodeid became master during
922 * a previous recovery cycle, and we aborted the previous
923 * cycle before recovering this master value
926 log_limit(ls, "%s from_master %d master_nodeid %d res_nodeid %d first %x %s",
927 __func__, from_nodeid, r->res_master_nodeid,
928 r->res_nodeid, r->res_first_lkid, r->res_name);
930 if (r->res_master_nodeid == our_nodeid) {
931 log_error(ls, "from_master %d our_master", from_nodeid);
936 r->res_master_nodeid = from_nodeid;
937 r->res_nodeid = from_nodeid;
938 rsb_set_flag(r, RSB_NEW_MASTER);
941 if (!r->res_master_nodeid) {
942 /* this will happen if recovery happens while we're looking
943 * up the master for this rsb
946 log_debug(ls, "%s master 0 to %d first %x %s", __func__,
947 from_nodeid, r->res_first_lkid, r->res_name);
948 r->res_master_nodeid = from_nodeid;
949 r->res_nodeid = from_nodeid;
952 if (!from_master && !fix_master &&
953 (r->res_master_nodeid == from_nodeid)) {
954 /* this can happen when the master sends remove, the dir node
955 * finds the rsb on the keep list and ignores the remove,
956 * and the former master sends a lookup
959 log_limit(ls, "%s from master %d flags %x first %x %s",
960 __func__, from_nodeid, flags, r->res_first_lkid,
965 *r_nodeid = r->res_master_nodeid;
967 *result = DLM_LU_MATCH;
971 * We're the dir node for this res and another node wants to know the
972 * master nodeid. During normal operation (non recovery) this is only
973 * called from receive_lookup(); master lookups when the local node is
974 * the dir node are done by find_rsb().
976 * normal operation, we are the dir node for a resource
981 * . dlm_master_lookup flags 0
983 * recover directory, we are rebuilding dir for all resources
984 * . dlm_recover_directory
986 * remote node sends back the rsb names it is master of and we are dir of
987 * . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
988 * we either create new rsb setting remote node as master, or find existing
989 * rsb and set master to be the remote node.
991 * recover masters, we are finding the new master for resources
992 * . dlm_recover_masters
994 * . dlm_send_rcom_lookup
995 * . receive_rcom_lookup
996 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
999 int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
1000 unsigned int flags, int *r_nodeid, int *result)
1002 struct dlm_rsb *r = NULL;
1004 int our_nodeid = dlm_our_nodeid();
1005 int dir_nodeid, error;
1007 if (len > DLM_RESNAME_MAXLEN)
1010 if (from_nodeid == our_nodeid) {
1011 log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
1016 hash = jhash(name, len, 0);
1017 b = hash & (ls->ls_rsbtbl_size - 1);
1019 dir_nodeid = dlm_hash2nodeid(ls, hash);
1020 if (dir_nodeid != our_nodeid) {
1021 log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
1022 from_nodeid, dir_nodeid, our_nodeid, hash,
1029 error = pre_rsb_struct(ls);
1033 spin_lock(&ls->ls_rsbtbl[b].lock);
1034 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1036 /* because the rsb is active, we need to lock_rsb before
1037 * checking/changing re_master_nodeid
1041 spin_unlock(&ls->ls_rsbtbl[b].lock);
1044 __dlm_master_lookup(ls, r, our_nodeid, from_nodeid, false,
1045 flags, r_nodeid, result);
1047 /* the rsb was active */
1054 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1058 /* because the rsb is inactive (on toss list), it's not refcounted
1059 * and lock_rsb is not used, but is protected by the rsbtbl lock
1062 __dlm_master_lookup(ls, r, our_nodeid, from_nodeid, true, flags,
1065 r->res_toss_time = jiffies;
1066 /* the rsb was inactive (on toss list) */
1067 spin_unlock(&ls->ls_rsbtbl[b].lock);
1072 error = get_rsb_struct(ls, name, len, &r);
1073 if (error == -EAGAIN) {
1074 spin_unlock(&ls->ls_rsbtbl[b].lock);
1082 r->res_dir_nodeid = our_nodeid;
1083 r->res_master_nodeid = from_nodeid;
1084 r->res_nodeid = from_nodeid;
1085 kref_init(&r->res_ref);
1086 r->res_toss_time = jiffies;
1088 error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1090 /* should never happen */
1092 spin_unlock(&ls->ls_rsbtbl[b].lock);
1097 *result = DLM_LU_ADD;
1098 *r_nodeid = from_nodeid;
1100 spin_unlock(&ls->ls_rsbtbl[b].lock);
1104 static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1110 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1111 spin_lock(&ls->ls_rsbtbl[i].lock);
1112 for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1113 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1114 if (r->res_hash == hash)
1117 spin_unlock(&ls->ls_rsbtbl[i].lock);
1121 void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
1123 struct dlm_rsb *r = NULL;
1127 hash = jhash(name, len, 0);
1128 b = hash & (ls->ls_rsbtbl_size - 1);
1130 spin_lock(&ls->ls_rsbtbl[b].lock);
1131 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1135 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1141 spin_unlock(&ls->ls_rsbtbl[b].lock);
1144 static void toss_rsb(struct kref *kref)
1146 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1147 struct dlm_ls *ls = r->res_ls;
1149 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1150 kref_init(&r->res_ref);
1151 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1152 rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
1153 r->res_toss_time = jiffies;
1154 ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
1155 if (r->res_lvbptr) {
1156 dlm_free_lvb(r->res_lvbptr);
1157 r->res_lvbptr = NULL;
1161 /* See comment for unhold_lkb */
1163 static void unhold_rsb(struct dlm_rsb *r)
1166 rv = kref_put(&r->res_ref, toss_rsb);
1167 DLM_ASSERT(!rv, dlm_dump_rsb(r););
1170 static void kill_rsb(struct kref *kref)
1172 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1174 /* All work is done after the return from kref_put() so we
1175 can release the write_lock before the remove and free. */
1177 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1178 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1179 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1180 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1181 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1182 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1185 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1186 The rsb must exist as long as any lkb's for it do. */
1188 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1191 lkb->lkb_resource = r;
1194 static void detach_lkb(struct dlm_lkb *lkb)
1196 if (lkb->lkb_resource) {
1197 put_rsb(lkb->lkb_resource);
1198 lkb->lkb_resource = NULL;
1202 static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret,
1205 struct dlm_lkb *lkb;
1208 lkb = dlm_allocate_lkb(ls);
1212 lkb->lkb_nodeid = -1;
1213 lkb->lkb_grmode = DLM_LOCK_IV;
1214 kref_init(&lkb->lkb_ref);
1215 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1216 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1217 #ifdef CONFIG_DLM_DEPRECATED_API
1218 INIT_LIST_HEAD(&lkb->lkb_time_list);
1220 INIT_LIST_HEAD(&lkb->lkb_cb_list);
1221 mutex_init(&lkb->lkb_cb_mutex);
1222 INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
1224 idr_preload(GFP_NOFS);
1225 spin_lock(&ls->ls_lkbidr_spin);
1226 rv = idr_alloc(&ls->ls_lkbidr, lkb, start, end, GFP_NOWAIT);
1229 spin_unlock(&ls->ls_lkbidr_spin);
1233 log_error(ls, "create_lkb idr error %d", rv);
1242 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1244 return _create_lkb(ls, lkb_ret, 1, 0);
1247 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1249 struct dlm_lkb *lkb;
1251 spin_lock(&ls->ls_lkbidr_spin);
1252 lkb = idr_find(&ls->ls_lkbidr, lkid);
1254 kref_get(&lkb->lkb_ref);
1255 spin_unlock(&ls->ls_lkbidr_spin);
1258 return lkb ? 0 : -ENOENT;
1261 static void kill_lkb(struct kref *kref)
1263 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1265 /* All work is done after the return from kref_put() so we
1266 can release the write_lock before the detach_lkb */
1268 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1271 /* __put_lkb() is used when an lkb may not have an rsb attached to
1272 it so we need to provide the lockspace explicitly */
1274 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
1276 uint32_t lkid = lkb->lkb_id;
1279 rv = kref_put_lock(&lkb->lkb_ref, kill_lkb,
1280 &ls->ls_lkbidr_spin);
1282 idr_remove(&ls->ls_lkbidr, lkid);
1283 spin_unlock(&ls->ls_lkbidr_spin);
1287 /* for local/process lkbs, lvbptr points to caller's lksb */
1288 if (lkb->lkb_lvbptr && is_master_copy(lkb))
1289 dlm_free_lvb(lkb->lkb_lvbptr);
1296 int dlm_put_lkb(struct dlm_lkb *lkb)
1300 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1301 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1303 ls = lkb->lkb_resource->res_ls;
1304 return __put_lkb(ls, lkb);
1307 /* This is only called to add a reference when the code already holds
1308 a valid reference to the lkb, so there's no need for locking. */
1310 static inline void hold_lkb(struct dlm_lkb *lkb)
1312 kref_get(&lkb->lkb_ref);
1315 static void unhold_lkb_assert(struct kref *kref)
1317 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1319 DLM_ASSERT(false, dlm_print_lkb(lkb););
1322 /* This is called when we need to remove a reference and are certain
1323 it's not the last ref. e.g. del_lkb is always called between a
1324 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1325 put_lkb would work fine, but would involve unnecessary locking */
1327 static inline void unhold_lkb(struct dlm_lkb *lkb)
1329 kref_put(&lkb->lkb_ref, unhold_lkb_assert);
1332 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1335 struct dlm_lkb *lkb = NULL, *iter;
1337 list_for_each_entry(iter, head, lkb_statequeue)
1338 if (iter->lkb_rqmode < mode) {
1340 list_add_tail(new, &iter->lkb_statequeue);
1345 list_add_tail(new, head);
1348 /* add/remove lkb to rsb's grant/convert/wait queue */
1350 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1352 kref_get(&lkb->lkb_ref);
1354 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1356 lkb->lkb_timestamp = ktime_get();
1358 lkb->lkb_status = status;
1361 case DLM_LKSTS_WAITING:
1362 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1363 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1365 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1367 case DLM_LKSTS_GRANTED:
1368 /* convention says granted locks kept in order of grmode */
1369 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1372 case DLM_LKSTS_CONVERT:
1373 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1374 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1376 list_add_tail(&lkb->lkb_statequeue,
1377 &r->res_convertqueue);
1380 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1384 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1386 lkb->lkb_status = 0;
1387 list_del(&lkb->lkb_statequeue);
1391 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1395 add_lkb(r, lkb, sts);
1399 static int msg_reply_type(int mstype)
1402 case DLM_MSG_REQUEST:
1403 return DLM_MSG_REQUEST_REPLY;
1404 case DLM_MSG_CONVERT:
1405 return DLM_MSG_CONVERT_REPLY;
1406 case DLM_MSG_UNLOCK:
1407 return DLM_MSG_UNLOCK_REPLY;
1408 case DLM_MSG_CANCEL:
1409 return DLM_MSG_CANCEL_REPLY;
1410 case DLM_MSG_LOOKUP:
1411 return DLM_MSG_LOOKUP_REPLY;
1416 /* add/remove lkb from global waiters list of lkb's waiting for
1417 a reply from a remote node */
1419 static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1421 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1424 mutex_lock(&ls->ls_waiters_mutex);
1426 if (is_overlap_unlock(lkb) ||
1427 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1432 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1434 case DLM_MSG_UNLOCK:
1435 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1437 case DLM_MSG_CANCEL:
1438 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1444 lkb->lkb_wait_count++;
1447 log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1448 lkb->lkb_id, lkb->lkb_wait_type, mstype,
1449 lkb->lkb_wait_count, lkb->lkb_flags);
1453 DLM_ASSERT(!lkb->lkb_wait_count,
1455 printk("wait_count %d\n", lkb->lkb_wait_count););
1457 lkb->lkb_wait_count++;
1458 lkb->lkb_wait_type = mstype;
1459 lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1461 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1464 log_error(ls, "addwait error %x %d flags %x %d %d %s",
1465 lkb->lkb_id, error, lkb->lkb_flags, mstype,
1466 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1467 mutex_unlock(&ls->ls_waiters_mutex);
1471 /* We clear the RESEND flag because we might be taking an lkb off the waiters
1472 list as part of process_requestqueue (e.g. a lookup that has an optimized
1473 request reply on the requestqueue) between dlm_recover_waiters_pre() which
1474 set RESEND and dlm_recover_waiters_post() */
1476 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1477 struct dlm_message *ms)
1479 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1480 int overlap_done = 0;
1482 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1483 log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1484 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1489 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1490 log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1491 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1496 /* Cancel state was preemptively cleared by a successful convert,
1497 see next comment, nothing to do. */
1499 if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1500 (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1501 log_debug(ls, "remwait %x cancel_reply wait_type %d",
1502 lkb->lkb_id, lkb->lkb_wait_type);
1506 /* Remove for the convert reply, and premptively remove for the
1507 cancel reply. A convert has been granted while there's still
1508 an outstanding cancel on it (the cancel is moot and the result
1509 in the cancel reply should be 0). We preempt the cancel reply
1510 because the app gets the convert result and then can follow up
1511 with another op, like convert. This subsequent op would see the
1512 lingering state of the cancel and fail with -EBUSY. */
1514 if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1515 (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1516 is_overlap_cancel(lkb) && ms && !ms->m_result) {
1517 log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1519 lkb->lkb_wait_type = 0;
1520 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1521 lkb->lkb_wait_count--;
1526 /* N.B. type of reply may not always correspond to type of original
1527 msg due to lookup->request optimization, verify others? */
1529 if (lkb->lkb_wait_type) {
1530 lkb->lkb_wait_type = 0;
1534 log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1535 lkb->lkb_id, ms ? le32_to_cpu(ms->m_header.h_nodeid) : 0,
1536 lkb->lkb_remid, mstype, lkb->lkb_flags);
1540 /* the force-unlock/cancel has completed and we haven't recvd a reply
1541 to the op that was in progress prior to the unlock/cancel; we
1542 give up on any reply to the earlier op. FIXME: not sure when/how
1543 this would happen */
1545 if (overlap_done && lkb->lkb_wait_type) {
1546 log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1547 lkb->lkb_id, mstype, lkb->lkb_wait_type);
1548 lkb->lkb_wait_count--;
1550 lkb->lkb_wait_type = 0;
1553 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1555 lkb->lkb_flags &= ~DLM_IFL_RESEND;
1556 lkb->lkb_wait_count--;
1557 if (!lkb->lkb_wait_count)
1558 list_del_init(&lkb->lkb_wait_reply);
1563 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1565 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1568 mutex_lock(&ls->ls_waiters_mutex);
1569 error = _remove_from_waiters(lkb, mstype, NULL);
1570 mutex_unlock(&ls->ls_waiters_mutex);
1574 /* Handles situations where we might be processing a "fake" or "stub" reply in
1575 which we can't try to take waiters_mutex again. */
1577 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1579 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1582 if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS))
1583 mutex_lock(&ls->ls_waiters_mutex);
1584 error = _remove_from_waiters(lkb, le32_to_cpu(ms->m_type), ms);
1585 if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS))
1586 mutex_unlock(&ls->ls_waiters_mutex);
1590 /* If there's an rsb for the same resource being removed, ensure
1591 * that the remove message is sent before the new lookup message.
1594 #define DLM_WAIT_PENDING_COND(ls, r) \
1595 (ls->ls_remove_len && \
1596 !rsb_cmp(r, ls->ls_remove_name, \
1599 static void wait_pending_remove(struct dlm_rsb *r)
1601 struct dlm_ls *ls = r->res_ls;
1603 spin_lock(&ls->ls_remove_spin);
1604 if (DLM_WAIT_PENDING_COND(ls, r)) {
1605 log_debug(ls, "delay lookup for remove dir %d %s",
1606 r->res_dir_nodeid, r->res_name);
1607 spin_unlock(&ls->ls_remove_spin);
1608 wait_event(ls->ls_remove_wait, !DLM_WAIT_PENDING_COND(ls, r));
1611 spin_unlock(&ls->ls_remove_spin);
1615 * ls_remove_spin protects ls_remove_name and ls_remove_len which are
1616 * read by other threads in wait_pending_remove. ls_remove_names
1617 * and ls_remove_lens are only used by the scan thread, so they do
1618 * not need protection.
1621 static void shrink_bucket(struct dlm_ls *ls, int b)
1623 struct rb_node *n, *next;
1626 int our_nodeid = dlm_our_nodeid();
1627 int remote_count = 0;
1628 int need_shrink = 0;
1631 memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1633 spin_lock(&ls->ls_rsbtbl[b].lock);
1635 if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1636 spin_unlock(&ls->ls_rsbtbl[b].lock);
1640 for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1642 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1644 /* If we're the directory record for this rsb, and
1645 we're not the master of it, then we need to wait
1646 for the master node to send us a dir remove for
1647 before removing the dir record. */
1649 if (!dlm_no_directory(ls) &&
1650 (r->res_master_nodeid != our_nodeid) &&
1651 (dlm_dir_nodeid(r) == our_nodeid)) {
1657 if (!time_after_eq(jiffies, r->res_toss_time +
1658 dlm_config.ci_toss_secs * HZ)) {
1662 if (!dlm_no_directory(ls) &&
1663 (r->res_master_nodeid == our_nodeid) &&
1664 (dlm_dir_nodeid(r) != our_nodeid)) {
1666 /* We're the master of this rsb but we're not
1667 the directory record, so we need to tell the
1668 dir node to remove the dir record. */
1670 ls->ls_remove_lens[remote_count] = r->res_length;
1671 memcpy(ls->ls_remove_names[remote_count], r->res_name,
1672 DLM_RESNAME_MAXLEN);
1675 if (remote_count >= DLM_REMOVE_NAMES_MAX)
1680 if (!kref_put(&r->res_ref, kill_rsb)) {
1681 log_error(ls, "tossed rsb in use %s", r->res_name);
1685 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1690 ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1692 ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1693 spin_unlock(&ls->ls_rsbtbl[b].lock);
1696 * While searching for rsb's to free, we found some that require
1697 * remote removal. We leave them in place and find them again here
1698 * so there is a very small gap between removing them from the toss
1699 * list and sending the removal. Keeping this gap small is
1700 * important to keep us (the master node) from being out of sync
1701 * with the remote dir node for very long.
1703 * From the time the rsb is removed from toss until just after
1704 * send_remove, the rsb name is saved in ls_remove_name. A new
1705 * lookup checks this to ensure that a new lookup message for the
1706 * same resource name is not sent just before the remove message.
1709 for (i = 0; i < remote_count; i++) {
1710 name = ls->ls_remove_names[i];
1711 len = ls->ls_remove_lens[i];
1713 spin_lock(&ls->ls_rsbtbl[b].lock);
1714 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1716 spin_unlock(&ls->ls_rsbtbl[b].lock);
1717 log_debug(ls, "remove_name not toss %s", name);
1721 if (r->res_master_nodeid != our_nodeid) {
1722 spin_unlock(&ls->ls_rsbtbl[b].lock);
1723 log_debug(ls, "remove_name master %d dir %d our %d %s",
1724 r->res_master_nodeid, r->res_dir_nodeid,
1729 if (r->res_dir_nodeid == our_nodeid) {
1730 /* should never happen */
1731 spin_unlock(&ls->ls_rsbtbl[b].lock);
1732 log_error(ls, "remove_name dir %d master %d our %d %s",
1733 r->res_dir_nodeid, r->res_master_nodeid,
1738 if (!time_after_eq(jiffies, r->res_toss_time +
1739 dlm_config.ci_toss_secs * HZ)) {
1740 spin_unlock(&ls->ls_rsbtbl[b].lock);
1741 log_debug(ls, "remove_name toss_time %lu now %lu %s",
1742 r->res_toss_time, jiffies, name);
1746 if (!kref_put(&r->res_ref, kill_rsb)) {
1747 spin_unlock(&ls->ls_rsbtbl[b].lock);
1748 log_error(ls, "remove_name in use %s", name);
1752 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1754 /* block lookup of same name until we've sent remove */
1755 spin_lock(&ls->ls_remove_spin);
1756 ls->ls_remove_len = len;
1757 memcpy(ls->ls_remove_name, name, DLM_RESNAME_MAXLEN);
1758 spin_unlock(&ls->ls_remove_spin);
1759 spin_unlock(&ls->ls_rsbtbl[b].lock);
1763 /* allow lookup of name again */
1764 spin_lock(&ls->ls_remove_spin);
1765 ls->ls_remove_len = 0;
1766 memset(ls->ls_remove_name, 0, DLM_RESNAME_MAXLEN);
1767 spin_unlock(&ls->ls_remove_spin);
1768 wake_up(&ls->ls_remove_wait);
1774 void dlm_scan_rsbs(struct dlm_ls *ls)
1778 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1779 shrink_bucket(ls, i);
1780 if (dlm_locking_stopped(ls))
1786 #ifdef CONFIG_DLM_DEPRECATED_API
1787 static void add_timeout(struct dlm_lkb *lkb)
1789 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1791 if (is_master_copy(lkb))
1794 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1795 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1796 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1799 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1804 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1805 mutex_lock(&ls->ls_timeout_mutex);
1807 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1808 mutex_unlock(&ls->ls_timeout_mutex);
1811 static void del_timeout(struct dlm_lkb *lkb)
1813 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1815 mutex_lock(&ls->ls_timeout_mutex);
1816 if (!list_empty(&lkb->lkb_time_list)) {
1817 list_del_init(&lkb->lkb_time_list);
1820 mutex_unlock(&ls->ls_timeout_mutex);
1823 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1824 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1825 and then lock rsb because of lock ordering in add_timeout. We may need
1826 to specify some special timeout-related bits in the lkb that are just to
1827 be accessed under the timeout_mutex. */
1829 void dlm_scan_timeout(struct dlm_ls *ls)
1832 struct dlm_lkb *lkb = NULL, *iter;
1833 int do_cancel, do_warn;
1837 if (dlm_locking_stopped(ls))
1842 mutex_lock(&ls->ls_timeout_mutex);
1843 list_for_each_entry(iter, &ls->ls_timeout, lkb_time_list) {
1845 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1846 iter->lkb_timestamp));
1848 if ((iter->lkb_exflags & DLM_LKF_TIMEOUT) &&
1849 wait_us >= (iter->lkb_timeout_cs * 10000))
1852 if ((iter->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1853 wait_us >= dlm_config.ci_timewarn_cs * 10000)
1856 if (!do_cancel && !do_warn)
1862 mutex_unlock(&ls->ls_timeout_mutex);
1867 r = lkb->lkb_resource;
1872 /* clear flag so we only warn once */
1873 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1874 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1876 dlm_timeout_warn(lkb);
1880 log_debug(ls, "timeout cancel %x node %d %s",
1881 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1882 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1883 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1885 _cancel_lock(r, lkb);
1894 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1895 dlm_recoverd before checking/setting ls_recover_begin. */
1897 void dlm_adjust_timeouts(struct dlm_ls *ls)
1899 struct dlm_lkb *lkb;
1900 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1902 ls->ls_recover_begin = 0;
1903 mutex_lock(&ls->ls_timeout_mutex);
1904 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1905 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1906 mutex_unlock(&ls->ls_timeout_mutex);
1909 static void add_timeout(struct dlm_lkb *lkb) { }
1910 static void del_timeout(struct dlm_lkb *lkb) { }
1913 /* lkb is master or local copy */
1915 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1917 int b, len = r->res_ls->ls_lvblen;
1919 /* b=1 lvb returned to caller
1920 b=0 lvb written to rsb or invalidated
1923 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1926 if (!lkb->lkb_lvbptr)
1929 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1935 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1936 lkb->lkb_lvbseq = r->res_lvbseq;
1938 } else if (b == 0) {
1939 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1940 rsb_set_flag(r, RSB_VALNOTVALID);
1944 if (!lkb->lkb_lvbptr)
1947 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1951 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1956 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1958 lkb->lkb_lvbseq = r->res_lvbseq;
1959 rsb_clear_flag(r, RSB_VALNOTVALID);
1962 if (rsb_flag(r, RSB_VALNOTVALID))
1963 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1966 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1968 if (lkb->lkb_grmode < DLM_LOCK_PW)
1971 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1972 rsb_set_flag(r, RSB_VALNOTVALID);
1976 if (!lkb->lkb_lvbptr)
1979 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1983 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1988 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1990 rsb_clear_flag(r, RSB_VALNOTVALID);
1993 /* lkb is process copy (pc) */
1995 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1996 struct dlm_message *ms)
2000 if (!lkb->lkb_lvbptr)
2003 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
2006 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
2008 int len = receive_extralen(ms);
2009 if (len > r->res_ls->ls_lvblen)
2010 len = r->res_ls->ls_lvblen;
2011 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2012 lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
2016 /* Manipulate lkb's on rsb's convert/granted/waiting queues
2017 remove_lock -- used for unlock, removes lkb from granted
2018 revert_lock -- used for cancel, moves lkb from convert to granted
2019 grant_lock -- used for request and convert, adds lkb to granted or
2020 moves lkb from convert or waiting to granted
2022 Each of these is used for master or local copy lkb's. There is
2023 also a _pc() variation used to make the corresponding change on
2024 a process copy (pc) lkb. */
2026 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2029 lkb->lkb_grmode = DLM_LOCK_IV;
2030 /* this unhold undoes the original ref from create_lkb()
2031 so this leads to the lkb being freed */
2035 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2037 set_lvb_unlock(r, lkb);
2038 _remove_lock(r, lkb);
2041 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2043 _remove_lock(r, lkb);
2046 /* returns: 0 did nothing
2047 1 moved lock to granted
2050 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2054 lkb->lkb_rqmode = DLM_LOCK_IV;
2056 switch (lkb->lkb_status) {
2057 case DLM_LKSTS_GRANTED:
2059 case DLM_LKSTS_CONVERT:
2060 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2063 case DLM_LKSTS_WAITING:
2065 lkb->lkb_grmode = DLM_LOCK_IV;
2066 /* this unhold undoes the original ref from create_lkb()
2067 so this leads to the lkb being freed */
2072 log_print("invalid status for revert %d", lkb->lkb_status);
2077 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2079 return revert_lock(r, lkb);
2082 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2084 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2085 lkb->lkb_grmode = lkb->lkb_rqmode;
2086 if (lkb->lkb_status)
2087 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2089 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2092 lkb->lkb_rqmode = DLM_LOCK_IV;
2093 lkb->lkb_highbast = 0;
2096 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2098 set_lvb_lock(r, lkb);
2099 _grant_lock(r, lkb);
2102 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2103 struct dlm_message *ms)
2105 set_lvb_lock_pc(r, lkb, ms);
2106 _grant_lock(r, lkb);
2109 /* called by grant_pending_locks() which means an async grant message must
2110 be sent to the requesting node in addition to granting the lock if the
2111 lkb belongs to a remote node. */
2113 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2116 if (is_master_copy(lkb))
2119 queue_cast(r, lkb, 0);
2122 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2123 change the granted/requested modes. We're munging things accordingly in
2125 CONVDEADLK: our grmode may have been forced down to NL to resolve a
2127 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2128 compatible with other granted locks */
2130 static void munge_demoted(struct dlm_lkb *lkb)
2132 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2133 log_print("munge_demoted %x invalid modes gr %d rq %d",
2134 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2138 lkb->lkb_grmode = DLM_LOCK_NL;
2141 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2143 if (ms->m_type != cpu_to_le32(DLM_MSG_REQUEST_REPLY) &&
2144 ms->m_type != cpu_to_le32(DLM_MSG_GRANT)) {
2145 log_print("munge_altmode %x invalid reply type %d",
2146 lkb->lkb_id, le32_to_cpu(ms->m_type));
2150 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2151 lkb->lkb_rqmode = DLM_LOCK_PR;
2152 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2153 lkb->lkb_rqmode = DLM_LOCK_CW;
2155 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2160 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2162 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2164 if (lkb->lkb_id == first->lkb_id)
2170 /* Check if the given lkb conflicts with another lkb on the queue. */
2172 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2174 struct dlm_lkb *this;
2176 list_for_each_entry(this, head, lkb_statequeue) {
2179 if (!modes_compat(this, lkb))
2186 * "A conversion deadlock arises with a pair of lock requests in the converting
2187 * queue for one resource. The granted mode of each lock blocks the requested
2188 * mode of the other lock."
2190 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2191 * convert queue from being granted, then deadlk/demote lkb.
2194 * Granted Queue: empty
2195 * Convert Queue: NL->EX (first lock)
2196 * PR->EX (second lock)
2198 * The first lock can't be granted because of the granted mode of the second
2199 * lock and the second lock can't be granted because it's not first in the
2200 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2201 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2202 * flag set and return DEMOTED in the lksb flags.
2204 * Originally, this function detected conv-deadlk in a more limited scope:
2205 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2206 * - if lkb1 was the first entry in the queue (not just earlier), and was
2207 * blocked by the granted mode of lkb2, and there was nothing on the
2208 * granted queue preventing lkb1 from being granted immediately, i.e.
2209 * lkb2 was the only thing preventing lkb1 from being granted.
2211 * That second condition meant we'd only say there was conv-deadlk if
2212 * resolving it (by demotion) would lead to the first lock on the convert
2213 * queue being granted right away. It allowed conversion deadlocks to exist
2214 * between locks on the convert queue while they couldn't be granted anyway.
2216 * Now, we detect and take action on conversion deadlocks immediately when
2217 * they're created, even if they may not be immediately consequential. If
2218 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2219 * mode that would prevent lkb1's conversion from being granted, we do a
2220 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2221 * I think this means that the lkb_is_ahead condition below should always
2222 * be zero, i.e. there will never be conv-deadlk between two locks that are
2223 * both already on the convert queue.
2226 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2228 struct dlm_lkb *lkb1;
2229 int lkb_is_ahead = 0;
2231 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2237 if (!lkb_is_ahead) {
2238 if (!modes_compat(lkb2, lkb1))
2241 if (!modes_compat(lkb2, lkb1) &&
2242 !modes_compat(lkb1, lkb2))
2250 * Return 1 if the lock can be granted, 0 otherwise.
2251 * Also detect and resolve conversion deadlocks.
2253 * lkb is the lock to be granted
2255 * now is 1 if the function is being called in the context of the
2256 * immediate request, it is 0 if called later, after the lock has been
2259 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2262 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2265 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2268 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2271 * 6-10: Version 5.4 introduced an option to address the phenomenon of
2272 * a new request for a NL mode lock being blocked.
2274 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2275 * request, then it would be granted. In essence, the use of this flag
2276 * tells the Lock Manager to expedite theis request by not considering
2277 * what may be in the CONVERTING or WAITING queues... As of this
2278 * writing, the EXPEDITE flag can be used only with new requests for NL
2279 * mode locks. This flag is not valid for conversion requests.
2281 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
2282 * conversion or used with a non-NL requested mode. We also know an
2283 * EXPEDITE request is always granted immediately, so now must always
2284 * be 1. The full condition to grant an expedite request: (now &&
2285 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2286 * therefore be shortened to just checking the flag.
2289 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2293 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2294 * added to the remaining conditions.
2297 if (queue_conflict(&r->res_grantqueue, lkb))
2301 * 6-3: By default, a conversion request is immediately granted if the
2302 * requested mode is compatible with the modes of all other granted
2306 if (queue_conflict(&r->res_convertqueue, lkb))
2310 * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2311 * locks for a recovered rsb, on which lkb's have been rebuilt.
2312 * The lkb's may have been rebuilt on the queues in a different
2313 * order than they were in on the previous master. So, granting
2314 * queued conversions in order after recovery doesn't make sense
2315 * since the order hasn't been preserved anyway. The new order
2316 * could also have created a new "in place" conversion deadlock.
2317 * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2318 * After recovery, there would be no granted locks, and possibly
2319 * NL->EX, PR->EX, an in-place conversion deadlock.) So, after
2320 * recovery, grant conversions without considering order.
2323 if (conv && recover)
2327 * 6-5: But the default algorithm for deciding whether to grant or
2328 * queue conversion requests does not by itself guarantee that such
2329 * requests are serviced on a "first come first serve" basis. This, in
2330 * turn, can lead to a phenomenon known as "indefinate postponement".
2332 * 6-7: This issue is dealt with by using the optional QUECVT flag with
2333 * the system service employed to request a lock conversion. This flag
2334 * forces certain conversion requests to be queued, even if they are
2335 * compatible with the granted modes of other locks on the same
2336 * resource. Thus, the use of this flag results in conversion requests
2337 * being ordered on a "first come first servce" basis.
2339 * DCT: This condition is all about new conversions being able to occur
2340 * "in place" while the lock remains on the granted queue (assuming
2341 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
2342 * doesn't _have_ to go onto the convert queue where it's processed in
2343 * order. The "now" variable is necessary to distinguish converts
2344 * being received and processed for the first time now, because once a
2345 * convert is moved to the conversion queue the condition below applies
2346 * requiring fifo granting.
2349 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2353 * Even if the convert is compat with all granted locks,
2354 * QUECVT forces it behind other locks on the convert queue.
2357 if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2358 if (list_empty(&r->res_convertqueue))
2365 * The NOORDER flag is set to avoid the standard vms rules on grant
2369 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2373 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2374 * granted until all other conversion requests ahead of it are granted
2378 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2382 * 6-4: By default, a new request is immediately granted only if all
2383 * three of the following conditions are satisfied when the request is
2385 * - The queue of ungranted conversion requests for the resource is
2387 * - The queue of ungranted new requests for the resource is empty.
2388 * - The mode of the new request is compatible with the most
2389 * restrictive mode of all granted locks on the resource.
2392 if (now && !conv && list_empty(&r->res_convertqueue) &&
2393 list_empty(&r->res_waitqueue))
2397 * 6-4: Once a lock request is in the queue of ungranted new requests,
2398 * it cannot be granted until the queue of ungranted conversion
2399 * requests is empty, all ungranted new requests ahead of it are
2400 * granted and/or canceled, and it is compatible with the granted mode
2401 * of the most restrictive lock granted on the resource.
2404 if (!now && !conv && list_empty(&r->res_convertqueue) &&
2405 first_in_list(lkb, &r->res_waitqueue))
2411 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2412 int recover, int *err)
2415 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2416 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2421 rv = _can_be_granted(r, lkb, now, recover);
2426 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2427 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2428 * cancels one of the locks.
2431 if (is_convert && can_be_queued(lkb) &&
2432 conversion_deadlock_detect(r, lkb)) {
2433 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2434 lkb->lkb_grmode = DLM_LOCK_NL;
2435 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2439 log_print("can_be_granted deadlock %x now %d",
2447 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2448 * to grant a request in a mode other than the normal rqmode. It's a
2449 * simple way to provide a big optimization to applications that can
2453 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2455 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2459 lkb->lkb_rqmode = alt;
2460 rv = _can_be_granted(r, lkb, now, 0);
2462 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2464 lkb->lkb_rqmode = rqmode;
2470 /* Returns the highest requested mode of all blocked conversions; sets
2471 cw if there's a blocked conversion to DLM_LOCK_CW. */
2473 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2474 unsigned int *count)
2476 struct dlm_lkb *lkb, *s;
2477 int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2478 int hi, demoted, quit, grant_restart, demote_restart;
2487 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2488 demoted = is_demoted(lkb);
2491 if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2492 grant_lock_pending(r, lkb);
2499 if (!demoted && is_demoted(lkb)) {
2500 log_print("WARN: pending demoted %x node %d %s",
2501 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2508 * If DLM_LKB_NODLKWT flag is set and conversion
2509 * deadlock is detected, we request blocking AST and
2510 * down (or cancel) conversion.
2512 if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
2513 if (lkb->lkb_highbast < lkb->lkb_rqmode) {
2514 queue_bast(r, lkb, lkb->lkb_rqmode);
2515 lkb->lkb_highbast = lkb->lkb_rqmode;
2518 log_print("WARN: pending deadlock %x node %d %s",
2519 lkb->lkb_id, lkb->lkb_nodeid,
2526 hi = max_t(int, lkb->lkb_rqmode, hi);
2528 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2534 if (demote_restart && !quit) {
2539 return max_t(int, high, hi);
2542 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2543 unsigned int *count)
2545 struct dlm_lkb *lkb, *s;
2547 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2548 if (can_be_granted(r, lkb, 0, 0, NULL)) {
2549 grant_lock_pending(r, lkb);
2553 high = max_t(int, lkb->lkb_rqmode, high);
2554 if (lkb->lkb_rqmode == DLM_LOCK_CW)
2562 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2563 on either the convert or waiting queue.
2564 high is the largest rqmode of all locks blocked on the convert or
2567 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2569 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2570 if (gr->lkb_highbast < DLM_LOCK_EX)
2575 if (gr->lkb_highbast < high &&
2576 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2581 static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2583 struct dlm_lkb *lkb, *s;
2584 int high = DLM_LOCK_IV;
2587 if (!is_master(r)) {
2588 log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2593 high = grant_pending_convert(r, high, &cw, count);
2594 high = grant_pending_wait(r, high, &cw, count);
2596 if (high == DLM_LOCK_IV)
2600 * If there are locks left on the wait/convert queue then send blocking
2601 * ASTs to granted locks based on the largest requested mode (high)
2605 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2606 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2607 if (cw && high == DLM_LOCK_PR &&
2608 lkb->lkb_grmode == DLM_LOCK_PR)
2609 queue_bast(r, lkb, DLM_LOCK_CW);
2611 queue_bast(r, lkb, high);
2612 lkb->lkb_highbast = high;
2617 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2619 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2620 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2621 if (gr->lkb_highbast < DLM_LOCK_EX)
2626 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2631 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2632 struct dlm_lkb *lkb)
2636 list_for_each_entry(gr, head, lkb_statequeue) {
2637 /* skip self when sending basts to convertqueue */
2640 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2641 queue_bast(r, gr, lkb->lkb_rqmode);
2642 gr->lkb_highbast = lkb->lkb_rqmode;
2647 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2649 send_bast_queue(r, &r->res_grantqueue, lkb);
2652 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2654 send_bast_queue(r, &r->res_grantqueue, lkb);
2655 send_bast_queue(r, &r->res_convertqueue, lkb);
2658 /* set_master(r, lkb) -- set the master nodeid of a resource
2660 The purpose of this function is to set the nodeid field in the given
2661 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2662 known, it can just be copied to the lkb and the function will return
2663 0. If the rsb's nodeid is _not_ known, it needs to be looked up
2664 before it can be copied to the lkb.
2666 When the rsb nodeid is being looked up remotely, the initial lkb
2667 causing the lookup is kept on the ls_waiters list waiting for the
2668 lookup reply. Other lkb's waiting for the same rsb lookup are kept
2669 on the rsb's res_lookup list until the master is verified.
2672 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2673 1: the rsb master is not available and the lkb has been placed on
2677 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2679 int our_nodeid = dlm_our_nodeid();
2681 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2682 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2683 r->res_first_lkid = lkb->lkb_id;
2684 lkb->lkb_nodeid = r->res_nodeid;
2688 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2689 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2693 if (r->res_master_nodeid == our_nodeid) {
2694 lkb->lkb_nodeid = 0;
2698 if (r->res_master_nodeid) {
2699 lkb->lkb_nodeid = r->res_master_nodeid;
2703 if (dlm_dir_nodeid(r) == our_nodeid) {
2704 /* This is a somewhat unusual case; find_rsb will usually
2705 have set res_master_nodeid when dir nodeid is local, but
2706 there are cases where we become the dir node after we've
2707 past find_rsb and go through _request_lock again.
2708 confirm_master() or process_lookup_list() needs to be
2709 called after this. */
2710 log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2711 lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2713 r->res_master_nodeid = our_nodeid;
2715 lkb->lkb_nodeid = 0;
2719 wait_pending_remove(r);
2721 r->res_first_lkid = lkb->lkb_id;
2722 send_lookup(r, lkb);
2726 static void process_lookup_list(struct dlm_rsb *r)
2728 struct dlm_lkb *lkb, *safe;
2730 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2731 list_del_init(&lkb->lkb_rsb_lookup);
2732 _request_lock(r, lkb);
2737 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
2739 static void confirm_master(struct dlm_rsb *r, int error)
2741 struct dlm_lkb *lkb;
2743 if (!r->res_first_lkid)
2749 r->res_first_lkid = 0;
2750 process_lookup_list(r);
2756 /* the remote request failed and won't be retried (it was
2757 a NOQUEUE, or has been canceled/unlocked); make a waiting
2758 lkb the first_lkid */
2760 r->res_first_lkid = 0;
2762 if (!list_empty(&r->res_lookup)) {
2763 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2765 list_del_init(&lkb->lkb_rsb_lookup);
2766 r->res_first_lkid = lkb->lkb_id;
2767 _request_lock(r, lkb);
2772 log_error(r->res_ls, "confirm_master unknown error %d", error);
2776 #ifdef CONFIG_DLM_DEPRECATED_API
2777 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2778 int namelen, unsigned long timeout_cs,
2779 void (*ast) (void *astparam),
2781 void (*bast) (void *astparam, int mode),
2782 struct dlm_args *args)
2784 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2785 int namelen, void (*ast)(void *astparam),
2787 void (*bast)(void *astparam, int mode),
2788 struct dlm_args *args)
2793 /* check for invalid arg usage */
2795 if (mode < 0 || mode > DLM_LOCK_EX)
2798 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2801 if (flags & DLM_LKF_CANCEL)
2804 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2807 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2810 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2813 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2816 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2819 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2822 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2828 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2831 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2834 /* these args will be copied to the lkb in validate_lock_args,
2835 it cannot be done now because when converting locks, fields in
2836 an active lkb cannot be modified before locking the rsb */
2838 args->flags = flags;
2840 args->astparam = astparam;
2841 args->bastfn = bast;
2842 #ifdef CONFIG_DLM_DEPRECATED_API
2843 args->timeout = timeout_cs;
2852 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2854 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2855 DLM_LKF_FORCEUNLOCK))
2858 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2861 args->flags = flags;
2862 args->astparam = astarg;
2866 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2867 struct dlm_args *args)
2871 if (args->flags & DLM_LKF_CONVERT) {
2872 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2875 /* lock not allowed if there's any op in progress */
2876 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
2879 if (is_overlap(lkb))
2883 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2886 if (args->flags & DLM_LKF_QUECVT &&
2887 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2891 lkb->lkb_exflags = args->flags;
2892 lkb->lkb_sbflags = 0;
2893 lkb->lkb_astfn = args->astfn;
2894 lkb->lkb_astparam = args->astparam;
2895 lkb->lkb_bastfn = args->bastfn;
2896 lkb->lkb_rqmode = args->mode;
2897 lkb->lkb_lksb = args->lksb;
2898 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2899 lkb->lkb_ownpid = (int) current->pid;
2900 #ifdef CONFIG_DLM_DEPRECATED_API
2901 lkb->lkb_timeout_cs = args->timeout;
2909 /* annoy the user because dlm usage is wrong */
2911 log_error(ls, "%s %d %x %x %x %d %d %s", __func__,
2912 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2913 lkb->lkb_status, lkb->lkb_wait_type,
2914 lkb->lkb_resource->res_name);
2917 log_debug(ls, "%s %d %x %x %x %d %d %s", __func__,
2918 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2919 lkb->lkb_status, lkb->lkb_wait_type,
2920 lkb->lkb_resource->res_name);
2927 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2930 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2931 because there may be a lookup in progress and it's valid to do
2932 cancel/unlockf on it */
2934 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2936 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2939 /* normal unlock not allowed if there's any op in progress */
2940 if (!(args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) &&
2941 (lkb->lkb_wait_type || lkb->lkb_wait_count))
2944 /* an lkb may be waiting for an rsb lookup to complete where the
2945 lookup was initiated by another lock */
2947 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2948 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2949 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2950 list_del_init(&lkb->lkb_rsb_lookup);
2951 queue_cast(lkb->lkb_resource, lkb,
2952 args->flags & DLM_LKF_CANCEL ?
2953 -DLM_ECANCEL : -DLM_EUNLOCK);
2954 unhold_lkb(lkb); /* undoes create_lkb() */
2956 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2961 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2962 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2967 /* an lkb may still exist even though the lock is EOL'ed due to a
2968 * cancel, unlock or failed noqueue request; an app can't use these
2969 * locks; return same error as if the lkid had not been found at all
2972 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2973 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2978 /* cancel not allowed with another cancel/unlock in progress */
2980 if (args->flags & DLM_LKF_CANCEL) {
2981 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2984 if (is_overlap(lkb))
2987 /* don't let scand try to do a cancel */
2990 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2991 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2996 /* there's nothing to cancel */
2997 if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2998 !lkb->lkb_wait_type) {
3003 switch (lkb->lkb_wait_type) {
3004 case DLM_MSG_LOOKUP:
3005 case DLM_MSG_REQUEST:
3006 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
3009 case DLM_MSG_UNLOCK:
3010 case DLM_MSG_CANCEL:
3013 /* add_to_waiters() will set OVERLAP_CANCEL */
3017 /* do we need to allow a force-unlock if there's a normal unlock
3018 already in progress? in what conditions could the normal unlock
3019 fail such that we'd want to send a force-unlock to be sure? */
3021 if (args->flags & DLM_LKF_FORCEUNLOCK) {
3022 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
3025 if (is_overlap_unlock(lkb))
3028 /* don't let scand try to do a cancel */
3031 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3032 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3037 switch (lkb->lkb_wait_type) {
3038 case DLM_MSG_LOOKUP:
3039 case DLM_MSG_REQUEST:
3040 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
3043 case DLM_MSG_UNLOCK:
3046 /* add_to_waiters() will set OVERLAP_UNLOCK */
3050 /* an overlapping op shouldn't blow away exflags from other op */
3051 lkb->lkb_exflags |= args->flags;
3052 lkb->lkb_sbflags = 0;
3053 lkb->lkb_astparam = args->astparam;
3060 /* annoy the user because dlm usage is wrong */
3062 log_error(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
3063 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3064 args->flags, lkb->lkb_wait_type,
3065 lkb->lkb_resource->res_name);
3068 log_debug(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
3069 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3070 args->flags, lkb->lkb_wait_type,
3071 lkb->lkb_resource->res_name);
3079 * Four stage 4 varieties:
3080 * do_request(), do_convert(), do_unlock(), do_cancel()
3081 * These are called on the master node for the given lock and
3082 * from the central locking logic.
3085 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3089 if (can_be_granted(r, lkb, 1, 0, NULL)) {
3091 queue_cast(r, lkb, 0);
3095 if (can_be_queued(lkb)) {
3096 error = -EINPROGRESS;
3097 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3103 queue_cast(r, lkb, -EAGAIN);
3108 static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3113 if (force_blocking_asts(lkb))
3114 send_blocking_asts_all(r, lkb);
3117 send_blocking_asts(r, lkb);
3122 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3127 /* changing an existing lock may allow others to be granted */
3129 if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3131 queue_cast(r, lkb, 0);
3135 /* can_be_granted() detected that this lock would block in a conversion
3136 deadlock, so we leave it on the granted queue and return EDEADLK in
3137 the ast for the convert. */
3139 if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
3140 /* it's left on the granted queue */
3141 revert_lock(r, lkb);
3142 queue_cast(r, lkb, -EDEADLK);
3147 /* is_demoted() means the can_be_granted() above set the grmode
3148 to NL, and left us on the granted queue. This auto-demotion
3149 (due to CONVDEADLK) might mean other locks, and/or this lock, are
3150 now grantable. We have to try to grant other converting locks
3151 before we try again to grant this one. */
3153 if (is_demoted(lkb)) {
3154 grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3155 if (_can_be_granted(r, lkb, 1, 0)) {
3157 queue_cast(r, lkb, 0);
3160 /* else fall through and move to convert queue */
3163 if (can_be_queued(lkb)) {
3164 error = -EINPROGRESS;
3166 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3172 queue_cast(r, lkb, -EAGAIN);
3177 static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3182 grant_pending_locks(r, NULL);
3183 /* grant_pending_locks also sends basts */
3186 if (force_blocking_asts(lkb))
3187 send_blocking_asts_all(r, lkb);
3190 send_blocking_asts(r, lkb);
3195 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3197 remove_lock(r, lkb);
3198 queue_cast(r, lkb, -DLM_EUNLOCK);
3199 return -DLM_EUNLOCK;
3202 static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3205 grant_pending_locks(r, NULL);
3208 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3210 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3214 error = revert_lock(r, lkb);
3216 queue_cast(r, lkb, -DLM_ECANCEL);
3217 return -DLM_ECANCEL;
3222 static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3226 grant_pending_locks(r, NULL);
3230 * Four stage 3 varieties:
3231 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3234 /* add a new lkb to a possibly new rsb, called by requesting process */
3236 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3240 /* set_master: sets lkb nodeid from r */
3242 error = set_master(r, lkb);
3251 /* receive_request() calls do_request() on remote node */
3252 error = send_request(r, lkb);
3254 error = do_request(r, lkb);
3255 /* for remote locks the request_reply is sent
3256 between do_request and do_request_effects */
3257 do_request_effects(r, lkb, error);
3263 /* change some property of an existing lkb, e.g. mode */
3265 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3270 /* receive_convert() calls do_convert() on remote node */
3271 error = send_convert(r, lkb);
3273 error = do_convert(r, lkb);
3274 /* for remote locks the convert_reply is sent
3275 between do_convert and do_convert_effects */
3276 do_convert_effects(r, lkb, error);
3282 /* remove an existing lkb from the granted queue */
3284 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3289 /* receive_unlock() calls do_unlock() on remote node */
3290 error = send_unlock(r, lkb);
3292 error = do_unlock(r, lkb);
3293 /* for remote locks the unlock_reply is sent
3294 between do_unlock and do_unlock_effects */
3295 do_unlock_effects(r, lkb, error);
3301 /* remove an existing lkb from the convert or wait queue */
3303 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3308 /* receive_cancel() calls do_cancel() on remote node */
3309 error = send_cancel(r, lkb);
3311 error = do_cancel(r, lkb);
3312 /* for remote locks the cancel_reply is sent
3313 between do_cancel and do_cancel_effects */
3314 do_cancel_effects(r, lkb, error);
3321 * Four stage 2 varieties:
3322 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3325 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3326 const void *name, int len,
3327 struct dlm_args *args)
3332 error = validate_lock_args(ls, lkb, args);
3336 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3343 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3345 error = _request_lock(r, lkb);
3352 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3353 struct dlm_args *args)
3358 r = lkb->lkb_resource;
3363 error = validate_lock_args(ls, lkb, args);
3367 error = _convert_lock(r, lkb);
3374 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3375 struct dlm_args *args)
3380 r = lkb->lkb_resource;
3385 error = validate_unlock_args(lkb, args);
3389 error = _unlock_lock(r, lkb);
3396 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3397 struct dlm_args *args)
3402 r = lkb->lkb_resource;
3407 error = validate_unlock_args(lkb, args);
3411 error = _cancel_lock(r, lkb);
3419 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
3422 int dlm_lock(dlm_lockspace_t *lockspace,
3424 struct dlm_lksb *lksb,
3427 unsigned int namelen,
3428 uint32_t parent_lkid,
3429 void (*ast) (void *astarg),
3431 void (*bast) (void *astarg, int mode))
3434 struct dlm_lkb *lkb;
3435 struct dlm_args args;
3436 int error, convert = flags & DLM_LKF_CONVERT;
3438 ls = dlm_find_lockspace_local(lockspace);
3442 dlm_lock_recovery(ls);
3445 error = find_lkb(ls, lksb->sb_lkid, &lkb);
3447 error = create_lkb(ls, &lkb);
3452 trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
3454 #ifdef CONFIG_DLM_DEPRECATED_API
3455 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3456 astarg, bast, &args);
3458 error = set_lock_args(mode, lksb, flags, namelen, ast, astarg, bast,
3465 error = convert_lock(ls, lkb, &args);
3467 error = request_lock(ls, lkb, name, namelen, &args);
3469 if (error == -EINPROGRESS)
3472 trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, true);
3474 if (convert || error)
3476 if (error == -EAGAIN || error == -EDEADLK)
3479 dlm_unlock_recovery(ls);
3480 dlm_put_lockspace(ls);
3484 int dlm_unlock(dlm_lockspace_t *lockspace,
3487 struct dlm_lksb *lksb,
3491 struct dlm_lkb *lkb;
3492 struct dlm_args args;
3495 ls = dlm_find_lockspace_local(lockspace);
3499 dlm_lock_recovery(ls);
3501 error = find_lkb(ls, lkid, &lkb);
3505 trace_dlm_unlock_start(ls, lkb, flags);
3507 error = set_unlock_args(flags, astarg, &args);
3511 if (flags & DLM_LKF_CANCEL)
3512 error = cancel_lock(ls, lkb, &args);
3514 error = unlock_lock(ls, lkb, &args);
3516 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3518 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3521 trace_dlm_unlock_end(ls, lkb, flags, error);
3525 dlm_unlock_recovery(ls);
3526 dlm_put_lockspace(ls);
3531 * send/receive routines for remote operations and replies
3535 * send_request receive_request
3536 * send_convert receive_convert
3537 * send_unlock receive_unlock
3538 * send_cancel receive_cancel
3539 * send_grant receive_grant
3540 * send_bast receive_bast
3541 * send_lookup receive_lookup
3542 * send_remove receive_remove
3545 * receive_request_reply send_request_reply
3546 * receive_convert_reply send_convert_reply
3547 * receive_unlock_reply send_unlock_reply
3548 * receive_cancel_reply send_cancel_reply
3549 * receive_lookup_reply send_lookup_reply
3552 static int _create_message(struct dlm_ls *ls, int mb_len,
3553 int to_nodeid, int mstype,
3554 struct dlm_message **ms_ret,
3555 struct dlm_mhandle **mh_ret)
3557 struct dlm_message *ms;
3558 struct dlm_mhandle *mh;
3561 /* get_buffer gives us a message handle (mh) that we need to
3562 pass into midcomms_commit and a message buffer (mb) that we
3563 write our data into */
3565 mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb);
3569 ms = (struct dlm_message *) mb;
3571 ms->m_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3572 ms->m_header.u.h_lockspace = cpu_to_le32(ls->ls_global_id);
3573 ms->m_header.h_nodeid = cpu_to_le32(dlm_our_nodeid());
3574 ms->m_header.h_length = cpu_to_le16(mb_len);
3575 ms->m_header.h_cmd = DLM_MSG;
3577 ms->m_type = cpu_to_le32(mstype);
3584 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3585 int to_nodeid, int mstype,
3586 struct dlm_message **ms_ret,
3587 struct dlm_mhandle **mh_ret)
3589 int mb_len = sizeof(struct dlm_message);
3592 case DLM_MSG_REQUEST:
3593 case DLM_MSG_LOOKUP:
3594 case DLM_MSG_REMOVE:
3595 mb_len += r->res_length;
3597 case DLM_MSG_CONVERT:
3598 case DLM_MSG_UNLOCK:
3599 case DLM_MSG_REQUEST_REPLY:
3600 case DLM_MSG_CONVERT_REPLY:
3602 if (lkb && lkb->lkb_lvbptr)
3603 mb_len += r->res_ls->ls_lvblen;
3607 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3611 /* further lowcomms enhancements or alternate implementations may make
3612 the return value from this function useful at some point */
3614 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms,
3615 const void *name, int namelen)
3617 dlm_midcomms_commit_mhandle(mh, name, namelen);
3621 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3622 struct dlm_message *ms)
3624 ms->m_nodeid = cpu_to_le32(lkb->lkb_nodeid);
3625 ms->m_pid = cpu_to_le32(lkb->lkb_ownpid);
3626 ms->m_lkid = cpu_to_le32(lkb->lkb_id);
3627 ms->m_remid = cpu_to_le32(lkb->lkb_remid);
3628 ms->m_exflags = cpu_to_le32(lkb->lkb_exflags);
3629 ms->m_sbflags = cpu_to_le32(lkb->lkb_sbflags);
3630 ms->m_flags = cpu_to_le32(lkb->lkb_flags);
3631 ms->m_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
3632 ms->m_status = cpu_to_le32(lkb->lkb_status);
3633 ms->m_grmode = cpu_to_le32(lkb->lkb_grmode);
3634 ms->m_rqmode = cpu_to_le32(lkb->lkb_rqmode);
3635 ms->m_hash = cpu_to_le32(r->res_hash);
3637 /* m_result and m_bastmode are set from function args,
3638 not from lkb fields */
3640 if (lkb->lkb_bastfn)
3641 ms->m_asts |= cpu_to_le32(DLM_CB_BAST);
3643 ms->m_asts |= cpu_to_le32(DLM_CB_CAST);
3645 /* compare with switch in create_message; send_remove() doesn't
3648 switch (ms->m_type) {
3649 case cpu_to_le32(DLM_MSG_REQUEST):
3650 case cpu_to_le32(DLM_MSG_LOOKUP):
3651 memcpy(ms->m_extra, r->res_name, r->res_length);
3653 case cpu_to_le32(DLM_MSG_CONVERT):
3654 case cpu_to_le32(DLM_MSG_UNLOCK):
3655 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3656 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3657 case cpu_to_le32(DLM_MSG_GRANT):
3658 if (!lkb->lkb_lvbptr || !(lkb->lkb_exflags & DLM_LKF_VALBLK))
3660 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3665 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3667 struct dlm_message *ms;
3668 struct dlm_mhandle *mh;
3669 int to_nodeid, error;
3671 to_nodeid = r->res_nodeid;
3673 error = add_to_waiters(lkb, mstype, to_nodeid);
3677 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3681 send_args(r, lkb, ms);
3683 error = send_message(mh, ms, r->res_name, r->res_length);
3689 remove_from_waiters(lkb, msg_reply_type(mstype));
3693 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3695 return send_common(r, lkb, DLM_MSG_REQUEST);
3698 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3702 error = send_common(r, lkb, DLM_MSG_CONVERT);
3704 /* down conversions go without a reply from the master */
3705 if (!error && down_conversion(lkb)) {
3706 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3707 r->res_ls->ls_stub_ms.m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
3708 r->res_ls->ls_stub_ms.m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
3709 r->res_ls->ls_stub_ms.m_result = 0;
3710 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3716 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
3717 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3718 that the master is still correct. */
3720 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3722 return send_common(r, lkb, DLM_MSG_UNLOCK);
3725 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3727 return send_common(r, lkb, DLM_MSG_CANCEL);
3730 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3732 struct dlm_message *ms;
3733 struct dlm_mhandle *mh;
3734 int to_nodeid, error;
3736 to_nodeid = lkb->lkb_nodeid;
3738 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3742 send_args(r, lkb, ms);
3746 error = send_message(mh, ms, r->res_name, r->res_length);
3751 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3753 struct dlm_message *ms;
3754 struct dlm_mhandle *mh;
3755 int to_nodeid, error;
3757 to_nodeid = lkb->lkb_nodeid;
3759 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3763 send_args(r, lkb, ms);
3765 ms->m_bastmode = cpu_to_le32(mode);
3767 error = send_message(mh, ms, r->res_name, r->res_length);
3772 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3774 struct dlm_message *ms;
3775 struct dlm_mhandle *mh;
3776 int to_nodeid, error;
3778 to_nodeid = dlm_dir_nodeid(r);
3780 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3784 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3788 send_args(r, lkb, ms);
3790 error = send_message(mh, ms, r->res_name, r->res_length);
3796 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3800 static int send_remove(struct dlm_rsb *r)
3802 struct dlm_message *ms;
3803 struct dlm_mhandle *mh;
3804 int to_nodeid, error;
3806 to_nodeid = dlm_dir_nodeid(r);
3808 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3812 memcpy(ms->m_extra, r->res_name, r->res_length);
3813 ms->m_hash = cpu_to_le32(r->res_hash);
3815 error = send_message(mh, ms, r->res_name, r->res_length);
3820 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3823 struct dlm_message *ms;
3824 struct dlm_mhandle *mh;
3825 int to_nodeid, error;
3827 to_nodeid = lkb->lkb_nodeid;
3829 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3833 send_args(r, lkb, ms);
3835 ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3837 error = send_message(mh, ms, r->res_name, r->res_length);
3842 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3844 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3847 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3849 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3852 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3854 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3857 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3859 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3862 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3863 int ret_nodeid, int rv)
3865 struct dlm_rsb *r = &ls->ls_stub_rsb;
3866 struct dlm_message *ms;
3867 struct dlm_mhandle *mh;
3868 int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid);
3870 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3874 ms->m_lkid = ms_in->m_lkid;
3875 ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3876 ms->m_nodeid = cpu_to_le32(ret_nodeid);
3878 error = send_message(mh, ms, ms_in->m_extra, receive_extralen(ms_in));
3883 /* which args we save from a received message depends heavily on the type
3884 of message, unlike the send side where we can safely send everything about
3885 the lkb for any type of message */
3887 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3889 lkb->lkb_exflags = le32_to_cpu(ms->m_exflags);
3890 lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
3891 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3892 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
3895 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3897 if (ms->m_flags == cpu_to_le32(DLM_IFL_STUB_MS))
3900 lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
3901 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3902 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
3905 static int receive_extralen(struct dlm_message *ms)
3907 return (le16_to_cpu(ms->m_header.h_length) -
3908 sizeof(struct dlm_message));
3911 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3912 struct dlm_message *ms)
3916 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3917 if (!lkb->lkb_lvbptr)
3918 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3919 if (!lkb->lkb_lvbptr)
3921 len = receive_extralen(ms);
3922 if (len > ls->ls_lvblen)
3923 len = ls->ls_lvblen;
3924 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3929 static void fake_bastfn(void *astparam, int mode)
3931 log_print("fake_bastfn should not be called");
3934 static void fake_astfn(void *astparam)
3936 log_print("fake_astfn should not be called");
3939 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3940 struct dlm_message *ms)
3942 lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3943 lkb->lkb_ownpid = le32_to_cpu(ms->m_pid);
3944 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3945 lkb->lkb_grmode = DLM_LOCK_IV;
3946 lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3948 lkb->lkb_bastfn = (ms->m_asts & cpu_to_le32(DLM_CB_BAST)) ? &fake_bastfn : NULL;
3949 lkb->lkb_astfn = (ms->m_asts & cpu_to_le32(DLM_CB_CAST)) ? &fake_astfn : NULL;
3951 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3952 /* lkb was just created so there won't be an lvb yet */
3953 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3954 if (!lkb->lkb_lvbptr)
3961 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3962 struct dlm_message *ms)
3964 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3967 if (receive_lvb(ls, lkb, ms))
3970 lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3971 lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
3976 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3977 struct dlm_message *ms)
3979 if (receive_lvb(ls, lkb, ms))
3984 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3985 uses to send a reply and that the remote end uses to process the reply. */
3987 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3989 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3990 lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3991 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3994 /* This is called after the rsb is locked so that we can safely inspect
3995 fields in the lkb. */
3997 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3999 int from = le32_to_cpu(ms->m_header.h_nodeid);
4002 /* currently mixing of user/kernel locks are not supported */
4003 if (ms->m_flags & cpu_to_le32(DLM_IFL_USER) &&
4004 ~lkb->lkb_flags & DLM_IFL_USER) {
4005 log_error(lkb->lkb_resource->res_ls,
4006 "got user dlm message for a kernel lock");
4011 switch (ms->m_type) {
4012 case cpu_to_le32(DLM_MSG_CONVERT):
4013 case cpu_to_le32(DLM_MSG_UNLOCK):
4014 case cpu_to_le32(DLM_MSG_CANCEL):
4015 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
4019 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4020 case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4021 case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4022 case cpu_to_le32(DLM_MSG_GRANT):
4023 case cpu_to_le32(DLM_MSG_BAST):
4024 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
4028 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
4029 if (!is_process_copy(lkb))
4031 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
4041 log_error(lkb->lkb_resource->res_ls,
4042 "ignore invalid message %d from %d %x %x %x %d",
4043 le32_to_cpu(ms->m_type), from, lkb->lkb_id,
4044 lkb->lkb_remid, lkb->lkb_flags, lkb->lkb_nodeid);
4048 static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4050 struct dlm_lkb *lkb;
4053 int error, namelen = 0;
4055 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4057 error = create_lkb(ls, &lkb);
4061 receive_flags(lkb, ms);
4062 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4063 error = receive_request_args(ls, lkb, ms);
4069 /* The dir node is the authority on whether we are the master
4070 for this rsb or not, so if the master sends us a request, we should
4071 recreate the rsb if we've destroyed it. This race happens when we
4072 send a remove message to the dir node at the same time that the dir
4073 node sends us a request for the rsb. */
4075 namelen = receive_extralen(ms);
4077 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4078 R_RECEIVE_REQUEST, &r);
4086 if (r->res_master_nodeid != dlm_our_nodeid()) {
4087 error = validate_master_nodeid(ls, r, from_nodeid);
4097 error = do_request(r, lkb);
4098 send_request_reply(r, lkb, error);
4099 do_request_effects(r, lkb, error);
4104 if (error == -EINPROGRESS)
4111 /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4112 and do this receive_request again from process_lookup_list once
4113 we get the lookup reply. This would avoid a many repeated
4114 ENOTBLK request failures when the lookup reply designating us
4115 as master is delayed. */
4117 if (error != -ENOTBLK) {
4118 log_limit(ls, "receive_request %x from %d %d",
4119 le32_to_cpu(ms->m_lkid), from_nodeid, error);
4122 setup_stub_lkb(ls, ms);
4123 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4127 static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4129 struct dlm_lkb *lkb;
4131 int error, reply = 1;
4133 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4137 if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4138 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4139 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4140 (unsigned long long)lkb->lkb_recover_seq,
4141 le32_to_cpu(ms->m_header.h_nodeid),
4142 le32_to_cpu(ms->m_lkid));
4148 r = lkb->lkb_resource;
4153 error = validate_message(lkb, ms);
4157 receive_flags(lkb, ms);
4159 error = receive_convert_args(ls, lkb, ms);
4161 send_convert_reply(r, lkb, error);
4165 reply = !down_conversion(lkb);
4167 error = do_convert(r, lkb);
4169 send_convert_reply(r, lkb, error);
4170 do_convert_effects(r, lkb, error);
4178 setup_stub_lkb(ls, ms);
4179 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4183 static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4185 struct dlm_lkb *lkb;
4189 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4193 if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4194 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4195 lkb->lkb_id, lkb->lkb_remid,
4196 le32_to_cpu(ms->m_header.h_nodeid),
4197 le32_to_cpu(ms->m_lkid));
4203 r = lkb->lkb_resource;
4208 error = validate_message(lkb, ms);
4212 receive_flags(lkb, ms);
4214 error = receive_unlock_args(ls, lkb, ms);
4216 send_unlock_reply(r, lkb, error);
4220 error = do_unlock(r, lkb);
4221 send_unlock_reply(r, lkb, error);
4222 do_unlock_effects(r, lkb, error);
4230 setup_stub_lkb(ls, ms);
4231 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4235 static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4237 struct dlm_lkb *lkb;
4241 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4245 receive_flags(lkb, ms);
4247 r = lkb->lkb_resource;
4252 error = validate_message(lkb, ms);
4256 error = do_cancel(r, lkb);
4257 send_cancel_reply(r, lkb, error);
4258 do_cancel_effects(r, lkb, error);
4266 setup_stub_lkb(ls, ms);
4267 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4271 static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4273 struct dlm_lkb *lkb;
4277 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4281 r = lkb->lkb_resource;
4286 error = validate_message(lkb, ms);
4290 receive_flags_reply(lkb, ms);
4291 if (is_altmode(lkb))
4292 munge_altmode(lkb, ms);
4293 grant_lock_pc(r, lkb, ms);
4294 queue_cast(r, lkb, 0);
4302 static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4304 struct dlm_lkb *lkb;
4308 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4312 r = lkb->lkb_resource;
4317 error = validate_message(lkb, ms);
4321 queue_bast(r, lkb, le32_to_cpu(ms->m_bastmode));
4322 lkb->lkb_highbast = le32_to_cpu(ms->m_bastmode);
4330 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4332 int len, error, ret_nodeid, from_nodeid, our_nodeid;
4334 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4335 our_nodeid = dlm_our_nodeid();
4337 len = receive_extralen(ms);
4339 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4342 /* Optimization: we're master so treat lookup as a request */
4343 if (!error && ret_nodeid == our_nodeid) {
4344 receive_request(ls, ms);
4347 send_lookup_reply(ls, ms, ret_nodeid, error);
4350 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4352 char name[DLM_RESNAME_MAXLEN+1];
4355 int rv, len, dir_nodeid, from_nodeid;
4357 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4359 len = receive_extralen(ms);
4361 if (len > DLM_RESNAME_MAXLEN) {
4362 log_error(ls, "receive_remove from %d bad len %d",
4367 dir_nodeid = dlm_hash2nodeid(ls, le32_to_cpu(ms->m_hash));
4368 if (dir_nodeid != dlm_our_nodeid()) {
4369 log_error(ls, "receive_remove from %d bad nodeid %d",
4370 from_nodeid, dir_nodeid);
4374 /* Look for name on rsbtbl.toss, if it's there, kill it.
4375 If it's on rsbtbl.keep, it's being used, and we should ignore this
4376 message. This is an expected race between the dir node sending a
4377 request to the master node at the same time as the master node sends
4378 a remove to the dir node. The resolution to that race is for the
4379 dir node to ignore the remove message, and the master node to
4380 recreate the master rsb when it gets a request from the dir node for
4381 an rsb it doesn't have. */
4383 memset(name, 0, sizeof(name));
4384 memcpy(name, ms->m_extra, len);
4386 hash = jhash(name, len, 0);
4387 b = hash & (ls->ls_rsbtbl_size - 1);
4389 spin_lock(&ls->ls_rsbtbl[b].lock);
4391 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4393 /* verify the rsb is on keep list per comment above */
4394 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4396 /* should not happen */
4397 log_error(ls, "receive_remove from %d not found %s",
4399 spin_unlock(&ls->ls_rsbtbl[b].lock);
4402 if (r->res_master_nodeid != from_nodeid) {
4403 /* should not happen */
4404 log_error(ls, "receive_remove keep from %d master %d",
4405 from_nodeid, r->res_master_nodeid);
4407 spin_unlock(&ls->ls_rsbtbl[b].lock);
4411 log_debug(ls, "receive_remove from %d master %d first %x %s",
4412 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4414 spin_unlock(&ls->ls_rsbtbl[b].lock);
4418 if (r->res_master_nodeid != from_nodeid) {
4419 log_error(ls, "receive_remove toss from %d master %d",
4420 from_nodeid, r->res_master_nodeid);
4422 spin_unlock(&ls->ls_rsbtbl[b].lock);
4426 if (kref_put(&r->res_ref, kill_rsb)) {
4427 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4428 spin_unlock(&ls->ls_rsbtbl[b].lock);
4431 log_error(ls, "receive_remove from %d rsb ref error",
4434 spin_unlock(&ls->ls_rsbtbl[b].lock);
4438 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4440 do_purge(ls, le32_to_cpu(ms->m_nodeid), le32_to_cpu(ms->m_pid));
4443 static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4445 struct dlm_lkb *lkb;
4447 int error, mstype, result;
4448 int from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4450 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4454 r = lkb->lkb_resource;
4458 error = validate_message(lkb, ms);
4462 mstype = lkb->lkb_wait_type;
4463 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4465 log_error(ls, "receive_request_reply %x remote %d %x result %d",
4466 lkb->lkb_id, from_nodeid, le32_to_cpu(ms->m_lkid),
4467 from_dlm_errno(le32_to_cpu(ms->m_result)));
4472 /* Optimization: the dir node was also the master, so it took our
4473 lookup as a request and sent request reply instead of lookup reply */
4474 if (mstype == DLM_MSG_LOOKUP) {
4475 r->res_master_nodeid = from_nodeid;
4476 r->res_nodeid = from_nodeid;
4477 lkb->lkb_nodeid = from_nodeid;
4480 /* this is the value returned from do_request() on the master */
4481 result = from_dlm_errno(le32_to_cpu(ms->m_result));
4485 /* request would block (be queued) on remote master */
4486 queue_cast(r, lkb, -EAGAIN);
4487 confirm_master(r, -EAGAIN);
4488 unhold_lkb(lkb); /* undoes create_lkb() */
4493 /* request was queued or granted on remote master */
4494 receive_flags_reply(lkb, ms);
4495 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
4496 if (is_altmode(lkb))
4497 munge_altmode(lkb, ms);
4499 add_lkb(r, lkb, DLM_LKSTS_WAITING);
4502 grant_lock_pc(r, lkb, ms);
4503 queue_cast(r, lkb, 0);
4505 confirm_master(r, result);
4510 /* find_rsb failed to find rsb or rsb wasn't master */
4511 log_limit(ls, "receive_request_reply %x from %d %d "
4512 "master %d dir %d first %x %s", lkb->lkb_id,
4513 from_nodeid, result, r->res_master_nodeid,
4514 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4516 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4517 r->res_master_nodeid != dlm_our_nodeid()) {
4518 /* cause _request_lock->set_master->send_lookup */
4519 r->res_master_nodeid = 0;
4521 lkb->lkb_nodeid = -1;
4524 if (is_overlap(lkb)) {
4525 /* we'll ignore error in cancel/unlock reply */
4526 queue_cast_overlap(r, lkb);
4527 confirm_master(r, result);
4528 unhold_lkb(lkb); /* undoes create_lkb() */
4530 _request_lock(r, lkb);
4532 if (r->res_master_nodeid == dlm_our_nodeid())
4533 confirm_master(r, 0);
4538 log_error(ls, "receive_request_reply %x error %d",
4539 lkb->lkb_id, result);
4542 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4543 log_debug(ls, "receive_request_reply %x result %d unlock",
4544 lkb->lkb_id, result);
4545 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4546 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4547 send_unlock(r, lkb);
4548 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4549 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4550 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4551 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4552 send_cancel(r, lkb);
4554 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4555 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4564 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4565 struct dlm_message *ms)
4567 /* this is the value returned from do_convert() on the master */
4568 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4570 /* convert would block (be queued) on remote master */
4571 queue_cast(r, lkb, -EAGAIN);
4575 receive_flags_reply(lkb, ms);
4576 revert_lock_pc(r, lkb);
4577 queue_cast(r, lkb, -EDEADLK);
4581 /* convert was queued on remote master */
4582 receive_flags_reply(lkb, ms);
4583 if (is_demoted(lkb))
4586 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4591 /* convert was granted on remote master */
4592 receive_flags_reply(lkb, ms);
4593 if (is_demoted(lkb))
4595 grant_lock_pc(r, lkb, ms);
4596 queue_cast(r, lkb, 0);
4600 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4601 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4602 le32_to_cpu(ms->m_lkid),
4603 from_dlm_errno(le32_to_cpu(ms->m_result)));
4609 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4611 struct dlm_rsb *r = lkb->lkb_resource;
4617 error = validate_message(lkb, ms);
4621 /* stub reply can happen with waiters_mutex held */
4622 error = remove_from_waiters_ms(lkb, ms);
4626 __receive_convert_reply(r, lkb, ms);
4632 static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4634 struct dlm_lkb *lkb;
4637 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4641 _receive_convert_reply(lkb, ms);
4646 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4648 struct dlm_rsb *r = lkb->lkb_resource;
4654 error = validate_message(lkb, ms);
4658 /* stub reply can happen with waiters_mutex held */
4659 error = remove_from_waiters_ms(lkb, ms);
4663 /* this is the value returned from do_unlock() on the master */
4665 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4667 receive_flags_reply(lkb, ms);
4668 remove_lock_pc(r, lkb);
4669 queue_cast(r, lkb, -DLM_EUNLOCK);
4674 log_error(r->res_ls, "receive_unlock_reply %x error %d",
4675 lkb->lkb_id, from_dlm_errno(le32_to_cpu(ms->m_result)));
4682 static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4684 struct dlm_lkb *lkb;
4687 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4691 _receive_unlock_reply(lkb, ms);
4696 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4698 struct dlm_rsb *r = lkb->lkb_resource;
4704 error = validate_message(lkb, ms);
4708 /* stub reply can happen with waiters_mutex held */
4709 error = remove_from_waiters_ms(lkb, ms);
4713 /* this is the value returned from do_cancel() on the master */
4715 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4717 receive_flags_reply(lkb, ms);
4718 revert_lock_pc(r, lkb);
4719 queue_cast(r, lkb, -DLM_ECANCEL);
4724 log_error(r->res_ls, "receive_cancel_reply %x error %d",
4726 from_dlm_errno(le32_to_cpu(ms->m_result)));
4733 static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4735 struct dlm_lkb *lkb;
4738 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4742 _receive_cancel_reply(lkb, ms);
4747 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4749 struct dlm_lkb *lkb;
4751 int error, ret_nodeid;
4752 int do_lookup_list = 0;
4754 error = find_lkb(ls, le32_to_cpu(ms->m_lkid), &lkb);
4756 log_error(ls, "%s no lkid %x", __func__,
4757 le32_to_cpu(ms->m_lkid));
4761 /* ms->m_result is the value returned by dlm_master_lookup on dir node
4762 FIXME: will a non-zero error ever be returned? */
4764 r = lkb->lkb_resource;
4768 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4772 ret_nodeid = le32_to_cpu(ms->m_nodeid);
4774 /* We sometimes receive a request from the dir node for this
4775 rsb before we've received the dir node's loookup_reply for it.
4776 The request from the dir node implies we're the master, so we set
4777 ourself as master in receive_request_reply, and verify here that
4778 we are indeed the master. */
4780 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4781 /* This should never happen */
4782 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4783 "master %d dir %d our %d first %x %s",
4784 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4785 ret_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
4786 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4789 if (ret_nodeid == dlm_our_nodeid()) {
4790 r->res_master_nodeid = ret_nodeid;
4793 r->res_first_lkid = 0;
4794 } else if (ret_nodeid == -1) {
4795 /* the remote node doesn't believe it's the dir node */
4796 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4797 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid));
4798 r->res_master_nodeid = 0;
4800 lkb->lkb_nodeid = -1;
4802 /* set_master() will set lkb_nodeid from r */
4803 r->res_master_nodeid = ret_nodeid;
4804 r->res_nodeid = ret_nodeid;
4807 if (is_overlap(lkb)) {
4808 log_debug(ls, "receive_lookup_reply %x unlock %x",
4809 lkb->lkb_id, lkb->lkb_flags);
4810 queue_cast_overlap(r, lkb);
4811 unhold_lkb(lkb); /* undoes create_lkb() */
4815 _request_lock(r, lkb);
4819 process_lookup_list(r);
4826 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4829 int error = 0, noent = 0;
4831 if (!dlm_is_member(ls, le32_to_cpu(ms->m_header.h_nodeid))) {
4832 log_limit(ls, "receive %d from non-member %d %x %x %d",
4833 le32_to_cpu(ms->m_type),
4834 le32_to_cpu(ms->m_header.h_nodeid),
4835 le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4836 from_dlm_errno(le32_to_cpu(ms->m_result)));
4840 switch (ms->m_type) {
4842 /* messages sent to a master node */
4844 case cpu_to_le32(DLM_MSG_REQUEST):
4845 error = receive_request(ls, ms);
4848 case cpu_to_le32(DLM_MSG_CONVERT):
4849 error = receive_convert(ls, ms);
4852 case cpu_to_le32(DLM_MSG_UNLOCK):
4853 error = receive_unlock(ls, ms);
4856 case cpu_to_le32(DLM_MSG_CANCEL):
4858 error = receive_cancel(ls, ms);
4861 /* messages sent from a master node (replies to above) */
4863 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
4864 error = receive_request_reply(ls, ms);
4867 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4868 error = receive_convert_reply(ls, ms);
4871 case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4872 error = receive_unlock_reply(ls, ms);
4875 case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4876 error = receive_cancel_reply(ls, ms);
4879 /* messages sent from a master node (only two types of async msg) */
4881 case cpu_to_le32(DLM_MSG_GRANT):
4883 error = receive_grant(ls, ms);
4886 case cpu_to_le32(DLM_MSG_BAST):
4888 error = receive_bast(ls, ms);
4891 /* messages sent to a dir node */
4893 case cpu_to_le32(DLM_MSG_LOOKUP):
4894 receive_lookup(ls, ms);
4897 case cpu_to_le32(DLM_MSG_REMOVE):
4898 receive_remove(ls, ms);
4901 /* messages sent from a dir node (remove has no reply) */
4903 case cpu_to_le32(DLM_MSG_LOOKUP_REPLY):
4904 receive_lookup_reply(ls, ms);
4907 /* other messages */
4909 case cpu_to_le32(DLM_MSG_PURGE):
4910 receive_purge(ls, ms);
4914 log_error(ls, "unknown message type %d",
4915 le32_to_cpu(ms->m_type));
4919 * When checking for ENOENT, we're checking the result of
4920 * find_lkb(m_remid):
4922 * The lock id referenced in the message wasn't found. This may
4923 * happen in normal usage for the async messages and cancel, so
4924 * only use log_debug for them.
4926 * Some errors are expected and normal.
4929 if (error == -ENOENT && noent) {
4930 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4931 le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4932 le32_to_cpu(ms->m_header.h_nodeid),
4933 le32_to_cpu(ms->m_lkid), saved_seq);
4934 } else if (error == -ENOENT) {
4935 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4936 le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4937 le32_to_cpu(ms->m_header.h_nodeid),
4938 le32_to_cpu(ms->m_lkid), saved_seq);
4940 if (ms->m_type == cpu_to_le32(DLM_MSG_CONVERT))
4941 dlm_dump_rsb_hash(ls, le32_to_cpu(ms->m_hash));
4944 if (error == -EINVAL) {
4945 log_error(ls, "receive %d inval from %d lkid %x remid %x "
4947 le32_to_cpu(ms->m_type),
4948 le32_to_cpu(ms->m_header.h_nodeid),
4949 le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4954 /* If the lockspace is in recovery mode (locking stopped), then normal
4955 messages are saved on the requestqueue for processing after recovery is
4956 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4957 messages off the requestqueue before we process new ones. This occurs right
4958 after recovery completes when we transition from saving all messages on
4959 requestqueue, to processing all the saved messages, to processing new
4960 messages as they arrive. */
4962 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4965 if (dlm_locking_stopped(ls)) {
4966 /* If we were a member of this lockspace, left, and rejoined,
4967 other nodes may still be sending us messages from the
4968 lockspace generation before we left. */
4969 if (!ls->ls_generation) {
4970 log_limit(ls, "receive %d from %d ignore old gen",
4971 le32_to_cpu(ms->m_type), nodeid);
4975 dlm_add_requestqueue(ls, nodeid, ms);
4977 dlm_wait_requestqueue(ls);
4978 _receive_message(ls, ms, 0);
4982 /* This is called by dlm_recoverd to process messages that were saved on
4983 the requestqueue. */
4985 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
4988 _receive_message(ls, ms, saved_seq);
4991 /* This is called by the midcomms layer when something is received for
4992 the lockspace. It could be either a MSG (normal message sent as part of
4993 standard locking activity) or an RCOM (recovery message sent as part of
4994 lockspace recovery). */
4996 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
4998 struct dlm_header *hd = &p->header;
5002 switch (hd->h_cmd) {
5004 type = le32_to_cpu(p->message.m_type);
5007 type = le32_to_cpu(p->rcom.rc_type);
5010 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5014 if (le32_to_cpu(hd->h_nodeid) != nodeid) {
5015 log_print("invalid h_nodeid %d from %d lockspace %x",
5016 le32_to_cpu(hd->h_nodeid), nodeid,
5017 le32_to_cpu(hd->u.h_lockspace));
5021 ls = dlm_find_lockspace_global(le32_to_cpu(hd->u.h_lockspace));
5023 if (dlm_config.ci_log_debug) {
5024 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5025 "%u from %d cmd %d type %d\n",
5026 le32_to_cpu(hd->u.h_lockspace), nodeid,
5030 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5031 dlm_send_ls_not_ready(nodeid, &p->rcom);
5035 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5036 be inactive (in this ls) before transitioning to recovery mode */
5038 down_read(&ls->ls_recv_active);
5039 if (hd->h_cmd == DLM_MSG)
5040 dlm_receive_message(ls, &p->message, nodeid);
5041 else if (hd->h_cmd == DLM_RCOM)
5042 dlm_receive_rcom(ls, &p->rcom, nodeid);
5044 log_error(ls, "invalid h_cmd %d from %d lockspace %x",
5045 hd->h_cmd, nodeid, le32_to_cpu(hd->u.h_lockspace));
5046 up_read(&ls->ls_recv_active);
5048 dlm_put_lockspace(ls);
5051 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5052 struct dlm_message *ms_stub)
5054 if (middle_conversion(lkb)) {
5056 memset(ms_stub, 0, sizeof(struct dlm_message));
5057 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5058 ms_stub->m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
5059 ms_stub->m_result = cpu_to_le32(to_dlm_errno(-EINPROGRESS));
5060 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5061 _receive_convert_reply(lkb, ms_stub);
5063 /* Same special case as in receive_rcom_lock_args() */
5064 lkb->lkb_grmode = DLM_LOCK_IV;
5065 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5068 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5069 lkb->lkb_flags |= DLM_IFL_RESEND;
5072 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5073 conversions are async; there's no reply from the remote master */
5076 /* A waiting lkb needs recovery if the master node has failed, or
5077 the master node is changing (only when no directory is used) */
5079 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5082 if (dlm_no_directory(ls))
5085 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5091 /* Recovery for locks that are waiting for replies from nodes that are now
5092 gone. We can just complete unlocks and cancels by faking a reply from the
5093 dead node. Requests and up-conversions we flag to be resent after
5094 recovery. Down-conversions can just be completed with a fake reply like
5095 unlocks. Conversions between PR and CW need special attention. */
5097 void dlm_recover_waiters_pre(struct dlm_ls *ls)
5099 struct dlm_lkb *lkb, *safe;
5100 struct dlm_message *ms_stub;
5101 int wait_type, stub_unlock_result, stub_cancel_result;
5104 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5108 mutex_lock(&ls->ls_waiters_mutex);
5110 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5112 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5114 /* exclude debug messages about unlocks because there can be so
5115 many and they aren't very interesting */
5117 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5118 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5119 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5123 lkb->lkb_resource->res_nodeid,
5125 lkb->lkb_wait_nodeid,
5129 /* all outstanding lookups, regardless of destination will be
5130 resent after recovery is done */
5132 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5133 lkb->lkb_flags |= DLM_IFL_RESEND;
5137 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5140 wait_type = lkb->lkb_wait_type;
5141 stub_unlock_result = -DLM_EUNLOCK;
5142 stub_cancel_result = -DLM_ECANCEL;
5144 /* Main reply may have been received leaving a zero wait_type,
5145 but a reply for the overlapping op may not have been
5146 received. In that case we need to fake the appropriate
5147 reply for the overlap op. */
5150 if (is_overlap_cancel(lkb)) {
5151 wait_type = DLM_MSG_CANCEL;
5152 if (lkb->lkb_grmode == DLM_LOCK_IV)
5153 stub_cancel_result = 0;
5155 if (is_overlap_unlock(lkb)) {
5156 wait_type = DLM_MSG_UNLOCK;
5157 if (lkb->lkb_grmode == DLM_LOCK_IV)
5158 stub_unlock_result = -ENOENT;
5161 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5162 lkb->lkb_id, lkb->lkb_flags, wait_type,
5163 stub_cancel_result, stub_unlock_result);
5166 switch (wait_type) {
5168 case DLM_MSG_REQUEST:
5169 lkb->lkb_flags |= DLM_IFL_RESEND;
5172 case DLM_MSG_CONVERT:
5173 recover_convert_waiter(ls, lkb, ms_stub);
5176 case DLM_MSG_UNLOCK:
5178 memset(ms_stub, 0, sizeof(struct dlm_message));
5179 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5180 ms_stub->m_type = cpu_to_le32(DLM_MSG_UNLOCK_REPLY);
5181 ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_unlock_result));
5182 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5183 _receive_unlock_reply(lkb, ms_stub);
5187 case DLM_MSG_CANCEL:
5189 memset(ms_stub, 0, sizeof(struct dlm_message));
5190 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5191 ms_stub->m_type = cpu_to_le32(DLM_MSG_CANCEL_REPLY);
5192 ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_cancel_result));
5193 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5194 _receive_cancel_reply(lkb, ms_stub);
5199 log_error(ls, "invalid lkb wait_type %d %d",
5200 lkb->lkb_wait_type, wait_type);
5204 mutex_unlock(&ls->ls_waiters_mutex);
5208 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5210 struct dlm_lkb *lkb = NULL, *iter;
5212 mutex_lock(&ls->ls_waiters_mutex);
5213 list_for_each_entry(iter, &ls->ls_waiters, lkb_wait_reply) {
5214 if (iter->lkb_flags & DLM_IFL_RESEND) {
5220 mutex_unlock(&ls->ls_waiters_mutex);
5225 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
5226 master or dir-node for r. Processing the lkb may result in it being placed
5229 /* We do this after normal locking has been enabled and any saved messages
5230 (in requestqueue) have been processed. We should be confident that at
5231 this point we won't get or process a reply to any of these waiting
5232 operations. But, new ops may be coming in on the rsbs/locks here from
5233 userspace or remotely. */
5235 /* there may have been an overlap unlock/cancel prior to recovery or after
5236 recovery. if before, the lkb may still have a pos wait_count; if after, the
5237 overlap flag would just have been set and nothing new sent. we can be
5238 confident here than any replies to either the initial op or overlap ops
5239 prior to recovery have been received. */
5241 int dlm_recover_waiters_post(struct dlm_ls *ls)
5243 struct dlm_lkb *lkb;
5245 int error = 0, mstype, err, oc, ou;
5248 if (dlm_locking_stopped(ls)) {
5249 log_debug(ls, "recover_waiters_post aborted");
5254 lkb = find_resend_waiter(ls);
5258 r = lkb->lkb_resource;
5262 mstype = lkb->lkb_wait_type;
5263 oc = is_overlap_cancel(lkb);
5264 ou = is_overlap_unlock(lkb);
5267 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5268 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5269 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5270 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5271 dlm_dir_nodeid(r), oc, ou);
5273 /* At this point we assume that we won't get a reply to any
5274 previous op or overlap op on this lock. First, do a big
5275 remove_from_waiters() for all previous ops. */
5277 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5278 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5279 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5280 lkb->lkb_wait_type = 0;
5281 /* drop all wait_count references we still
5282 * hold a reference for this iteration.
5284 while (lkb->lkb_wait_count) {
5285 lkb->lkb_wait_count--;
5288 mutex_lock(&ls->ls_waiters_mutex);
5289 list_del_init(&lkb->lkb_wait_reply);
5290 mutex_unlock(&ls->ls_waiters_mutex);
5293 /* do an unlock or cancel instead of resending */
5295 case DLM_MSG_LOOKUP:
5296 case DLM_MSG_REQUEST:
5297 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5299 unhold_lkb(lkb); /* undoes create_lkb() */
5301 case DLM_MSG_CONVERT:
5303 queue_cast(r, lkb, -DLM_ECANCEL);
5305 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5306 _unlock_lock(r, lkb);
5314 case DLM_MSG_LOOKUP:
5315 case DLM_MSG_REQUEST:
5316 _request_lock(r, lkb);
5318 confirm_master(r, 0);
5320 case DLM_MSG_CONVERT:
5321 _convert_lock(r, lkb);
5329 log_error(ls, "waiter %x msg %d r_nodeid %d "
5330 "dir_nodeid %d overlap %d %d",
5331 lkb->lkb_id, mstype, r->res_nodeid,
5332 dlm_dir_nodeid(r), oc, ou);
5342 static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5343 struct list_head *list)
5345 struct dlm_lkb *lkb, *safe;
5347 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5348 if (!is_master_copy(lkb))
5351 /* don't purge lkbs we've added in recover_master_copy for
5352 the current recovery seq */
5354 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5359 /* this put should free the lkb */
5360 if (!dlm_put_lkb(lkb))
5361 log_error(ls, "purged mstcpy lkb not released");
5365 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5367 struct dlm_ls *ls = r->res_ls;
5369 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5370 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5371 purge_mstcpy_list(ls, r, &r->res_waitqueue);
5374 static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5375 struct list_head *list,
5376 int nodeid_gone, unsigned int *count)
5378 struct dlm_lkb *lkb, *safe;
5380 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5381 if (!is_master_copy(lkb))
5384 if ((lkb->lkb_nodeid == nodeid_gone) ||
5385 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5387 /* tell recover_lvb to invalidate the lvb
5388 because a node holding EX/PW failed */
5389 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5390 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5391 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5396 /* this put should free the lkb */
5397 if (!dlm_put_lkb(lkb))
5398 log_error(ls, "purged dead lkb not released");
5400 rsb_set_flag(r, RSB_RECOVER_GRANT);
5407 /* Get rid of locks held by nodes that are gone. */
5409 void dlm_recover_purge(struct dlm_ls *ls)
5412 struct dlm_member *memb;
5413 int nodes_count = 0;
5414 int nodeid_gone = 0;
5415 unsigned int lkb_count = 0;
5417 /* cache one removed nodeid to optimize the common
5418 case of a single node removed */
5420 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5422 nodeid_gone = memb->nodeid;
5428 down_write(&ls->ls_root_sem);
5429 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5433 purge_dead_list(ls, r, &r->res_grantqueue,
5434 nodeid_gone, &lkb_count);
5435 purge_dead_list(ls, r, &r->res_convertqueue,
5436 nodeid_gone, &lkb_count);
5437 purge_dead_list(ls, r, &r->res_waitqueue,
5438 nodeid_gone, &lkb_count);
5444 up_write(&ls->ls_root_sem);
5447 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5448 lkb_count, nodes_count);
5451 static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5456 spin_lock(&ls->ls_rsbtbl[bucket].lock);
5457 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5458 r = rb_entry(n, struct dlm_rsb, res_hashnode);
5460 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5462 if (!is_master(r)) {
5463 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5467 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5470 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5475 * Attempt to grant locks on resources that we are the master of.
5476 * Locks may have become grantable during recovery because locks
5477 * from departed nodes have been purged (or not rebuilt), allowing
5478 * previously blocked locks to now be granted. The subset of rsb's
5479 * we are interested in are those with lkb's on either the convert or
5482 * Simplest would be to go through each master rsb and check for non-empty
5483 * convert or waiting queues, and attempt to grant on those rsbs.
5484 * Checking the queues requires lock_rsb, though, for which we'd need
5485 * to release the rsbtbl lock. This would make iterating through all
5486 * rsb's very inefficient. So, we rely on earlier recovery routines
5487 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5491 void dlm_recover_grant(struct dlm_ls *ls)
5495 unsigned int count = 0;
5496 unsigned int rsb_count = 0;
5497 unsigned int lkb_count = 0;
5500 r = find_grant_rsb(ls, bucket);
5502 if (bucket == ls->ls_rsbtbl_size - 1)
5510 /* the RECOVER_GRANT flag is checked in the grant path */
5511 grant_pending_locks(r, &count);
5512 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5514 confirm_master(r, 0);
5521 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5522 lkb_count, rsb_count);
5525 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5528 struct dlm_lkb *lkb;
5530 list_for_each_entry(lkb, head, lkb_statequeue) {
5531 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5537 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5540 struct dlm_lkb *lkb;
5542 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5545 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5548 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5554 /* needs at least dlm_rcom + rcom_lock */
5555 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5556 struct dlm_rsb *r, struct dlm_rcom *rc)
5558 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5560 lkb->lkb_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5561 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5562 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5563 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5564 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5565 lkb->lkb_flags |= DLM_IFL_MSTCPY;
5566 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5567 lkb->lkb_rqmode = rl->rl_rqmode;
5568 lkb->lkb_grmode = rl->rl_grmode;
5569 /* don't set lkb_status because add_lkb wants to itself */
5571 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5572 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5574 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5575 int lvblen = le16_to_cpu(rc->rc_header.h_length) -
5576 sizeof(struct dlm_rcom) - sizeof(struct rcom_lock);
5577 if (lvblen > ls->ls_lvblen)
5579 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5580 if (!lkb->lkb_lvbptr)
5582 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5585 /* Conversions between PR and CW (middle modes) need special handling.
5586 The real granted mode of these converting locks cannot be determined
5587 until all locks have been rebuilt on the rsb (recover_conversion) */
5589 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5590 middle_conversion(lkb)) {
5591 rl->rl_status = DLM_LKSTS_CONVERT;
5592 lkb->lkb_grmode = DLM_LOCK_IV;
5593 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5599 /* This lkb may have been recovered in a previous aborted recovery so we need
5600 to check if the rsb already has an lkb with the given remote nodeid/lkid.
5601 If so we just send back a standard reply. If not, we create a new lkb with
5602 the given values and send back our lkid. We send back our lkid by sending
5603 back the rcom_lock struct we got but with the remid field filled in. */
5605 /* needs at least dlm_rcom + rcom_lock */
5606 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5608 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5610 struct dlm_lkb *lkb;
5612 int from_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5615 if (rl->rl_parent_lkid) {
5616 error = -EOPNOTSUPP;
5620 remid = le32_to_cpu(rl->rl_lkid);
5622 /* In general we expect the rsb returned to be R_MASTER, but we don't
5623 have to require it. Recovery of masters on one node can overlap
5624 recovery of locks on another node, so one node can send us MSTCPY
5625 locks before we've made ourselves master of this rsb. We can still
5626 add new MSTCPY locks that we receive here without any harm; when
5627 we make ourselves master, dlm_recover_masters() won't touch the
5628 MSTCPY locks we've received early. */
5630 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5631 from_nodeid, R_RECEIVE_RECOVER, &r);
5637 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5638 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5639 from_nodeid, remid);
5644 lkb = search_remid(r, from_nodeid, remid);
5650 error = create_lkb(ls, &lkb);
5654 error = receive_rcom_lock_args(ls, lkb, r, rc);
5661 add_lkb(r, lkb, rl->rl_status);
5662 ls->ls_recover_locks_in++;
5664 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5665 rsb_set_flag(r, RSB_RECOVER_GRANT);
5668 /* this is the new value returned to the lock holder for
5669 saving in its process-copy lkb */
5670 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5672 lkb->lkb_recover_seq = ls->ls_recover_seq;
5678 if (error && error != -EEXIST)
5679 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5680 from_nodeid, remid, error);
5681 rl->rl_result = cpu_to_le32(error);
5685 /* needs at least dlm_rcom + rcom_lock */
5686 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5688 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5690 struct dlm_lkb *lkb;
5691 uint32_t lkid, remid;
5694 lkid = le32_to_cpu(rl->rl_lkid);
5695 remid = le32_to_cpu(rl->rl_remid);
5696 result = le32_to_cpu(rl->rl_result);
5698 error = find_lkb(ls, lkid, &lkb);
5700 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5701 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5706 r = lkb->lkb_resource;
5710 if (!is_process_copy(lkb)) {
5711 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5712 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5723 /* There's a chance the new master received our lock before
5724 dlm_recover_master_reply(), this wouldn't happen if we did
5725 a barrier between recover_masters and recover_locks. */
5727 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5728 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5731 dlm_send_rcom_lock(r, lkb);
5735 lkb->lkb_remid = remid;
5738 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5739 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5743 /* an ack for dlm_recover_locks() which waits for replies from
5744 all the locks it sends to new masters */
5745 dlm_recovered_lock(r);
5754 #ifdef CONFIG_DLM_DEPRECATED_API
5755 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5756 int mode, uint32_t flags, void *name, unsigned int namelen,
5757 unsigned long timeout_cs)
5759 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5760 int mode, uint32_t flags, void *name, unsigned int namelen)
5763 struct dlm_lkb *lkb;
5764 struct dlm_args args;
5768 dlm_lock_recovery(ls);
5770 error = create_lkb(ls, &lkb);
5776 trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
5778 if (flags & DLM_LKF_VALBLK) {
5779 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5780 if (!ua->lksb.sb_lvbptr) {
5786 #ifdef CONFIG_DLM_DEPRECATED_API
5787 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5788 fake_astfn, ua, fake_bastfn, &args);
5790 error = set_lock_args(mode, &ua->lksb, flags, namelen, fake_astfn, ua,
5791 fake_bastfn, &args);
5794 kfree(ua->lksb.sb_lvbptr);
5795 ua->lksb.sb_lvbptr = NULL;
5800 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5801 When DLM_IFL_USER is set, the dlm knows that this is a userspace
5802 lock and that lkb_astparam is the dlm_user_args structure. */
5803 lkb->lkb_flags |= DLM_IFL_USER;
5804 error = request_lock(ls, lkb, name, namelen, &args);
5819 /* add this new lkb to the per-process list of locks */
5820 spin_lock(&ua->proc->locks_spin);
5822 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5823 spin_unlock(&ua->proc->locks_spin);
5826 trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, false);
5830 dlm_unlock_recovery(ls);
5834 #ifdef CONFIG_DLM_DEPRECATED_API
5835 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5836 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5837 unsigned long timeout_cs)
5839 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5840 int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
5843 struct dlm_lkb *lkb;
5844 struct dlm_args args;
5845 struct dlm_user_args *ua;
5848 dlm_lock_recovery(ls);
5850 error = find_lkb(ls, lkid, &lkb);
5854 trace_dlm_lock_start(ls, lkb, NULL, 0, mode, flags);
5856 /* user can change the params on its lock when it converts it, or
5857 add an lvb that didn't exist before */
5861 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5862 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5863 if (!ua->lksb.sb_lvbptr) {
5868 if (lvb_in && ua->lksb.sb_lvbptr)
5869 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5871 ua->xid = ua_tmp->xid;
5872 ua->castparam = ua_tmp->castparam;
5873 ua->castaddr = ua_tmp->castaddr;
5874 ua->bastparam = ua_tmp->bastparam;
5875 ua->bastaddr = ua_tmp->bastaddr;
5876 ua->user_lksb = ua_tmp->user_lksb;
5878 #ifdef CONFIG_DLM_DEPRECATED_API
5879 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5880 fake_astfn, ua, fake_bastfn, &args);
5882 error = set_lock_args(mode, &ua->lksb, flags, 0, fake_astfn, ua,
5883 fake_bastfn, &args);
5888 error = convert_lock(ls, lkb, &args);
5890 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5893 trace_dlm_lock_end(ls, lkb, NULL, 0, mode, flags, error, false);
5896 dlm_unlock_recovery(ls);
5902 * The caller asks for an orphan lock on a given resource with a given mode.
5903 * If a matching lock exists, it's moved to the owner's list of locks and
5904 * the lkid is returned.
5907 int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5908 int mode, uint32_t flags, void *name, unsigned int namelen,
5911 struct dlm_lkb *lkb = NULL, *iter;
5912 struct dlm_user_args *ua;
5913 int found_other_mode = 0;
5916 mutex_lock(&ls->ls_orphans_mutex);
5917 list_for_each_entry(iter, &ls->ls_orphans, lkb_ownqueue) {
5918 if (iter->lkb_resource->res_length != namelen)
5920 if (memcmp(iter->lkb_resource->res_name, name, namelen))
5922 if (iter->lkb_grmode != mode) {
5923 found_other_mode = 1;
5928 list_del_init(&iter->lkb_ownqueue);
5929 iter->lkb_flags &= ~DLM_IFL_ORPHAN;
5930 *lkid = iter->lkb_id;
5933 mutex_unlock(&ls->ls_orphans_mutex);
5935 if (!lkb && found_other_mode) {
5945 lkb->lkb_exflags = flags;
5946 lkb->lkb_ownpid = (int) current->pid;
5950 ua->proc = ua_tmp->proc;
5951 ua->xid = ua_tmp->xid;
5952 ua->castparam = ua_tmp->castparam;
5953 ua->castaddr = ua_tmp->castaddr;
5954 ua->bastparam = ua_tmp->bastparam;
5955 ua->bastaddr = ua_tmp->bastaddr;
5956 ua->user_lksb = ua_tmp->user_lksb;
5959 * The lkb reference from the ls_orphans list was not
5960 * removed above, and is now considered the reference
5961 * for the proc locks list.
5964 spin_lock(&ua->proc->locks_spin);
5965 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5966 spin_unlock(&ua->proc->locks_spin);
5972 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5973 uint32_t flags, uint32_t lkid, char *lvb_in)
5975 struct dlm_lkb *lkb;
5976 struct dlm_args args;
5977 struct dlm_user_args *ua;
5980 dlm_lock_recovery(ls);
5982 error = find_lkb(ls, lkid, &lkb);
5986 trace_dlm_unlock_start(ls, lkb, flags);
5990 if (lvb_in && ua->lksb.sb_lvbptr)
5991 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5992 if (ua_tmp->castparam)
5993 ua->castparam = ua_tmp->castparam;
5994 ua->user_lksb = ua_tmp->user_lksb;
5996 error = set_unlock_args(flags, ua, &args);
6000 error = unlock_lock(ls, lkb, &args);
6002 if (error == -DLM_EUNLOCK)
6004 /* from validate_unlock_args() */
6005 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
6010 spin_lock(&ua->proc->locks_spin);
6011 /* dlm_user_add_cb() may have already taken lkb off the proc list */
6012 if (!list_empty(&lkb->lkb_ownqueue))
6013 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
6014 spin_unlock(&ua->proc->locks_spin);
6016 trace_dlm_unlock_end(ls, lkb, flags, error);
6019 dlm_unlock_recovery(ls);
6024 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6025 uint32_t flags, uint32_t lkid)
6027 struct dlm_lkb *lkb;
6028 struct dlm_args args;
6029 struct dlm_user_args *ua;
6032 dlm_lock_recovery(ls);
6034 error = find_lkb(ls, lkid, &lkb);
6038 trace_dlm_unlock_start(ls, lkb, flags);
6041 if (ua_tmp->castparam)
6042 ua->castparam = ua_tmp->castparam;
6043 ua->user_lksb = ua_tmp->user_lksb;
6045 error = set_unlock_args(flags, ua, &args);
6049 error = cancel_lock(ls, lkb, &args);
6051 if (error == -DLM_ECANCEL)
6053 /* from validate_unlock_args() */
6054 if (error == -EBUSY)
6057 trace_dlm_unlock_end(ls, lkb, flags, error);
6060 dlm_unlock_recovery(ls);
6065 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6067 struct dlm_lkb *lkb;
6068 struct dlm_args args;
6069 struct dlm_user_args *ua;
6073 dlm_lock_recovery(ls);
6075 error = find_lkb(ls, lkid, &lkb);
6079 trace_dlm_unlock_start(ls, lkb, flags);
6083 error = set_unlock_args(flags, ua, &args);
6087 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6089 r = lkb->lkb_resource;
6093 error = validate_unlock_args(lkb, &args);
6096 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6098 error = _cancel_lock(r, lkb);
6103 if (error == -DLM_ECANCEL)
6105 /* from validate_unlock_args() */
6106 if (error == -EBUSY)
6109 trace_dlm_unlock_end(ls, lkb, flags, error);
6112 dlm_unlock_recovery(ls);
6116 /* lkb's that are removed from the waiters list by revert are just left on the
6117 orphans list with the granted orphan locks, to be freed by purge */
6119 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6121 struct dlm_args args;
6124 hold_lkb(lkb); /* reference for the ls_orphans list */
6125 mutex_lock(&ls->ls_orphans_mutex);
6126 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6127 mutex_unlock(&ls->ls_orphans_mutex);
6129 set_unlock_args(0, lkb->lkb_ua, &args);
6131 error = cancel_lock(ls, lkb, &args);
6132 if (error == -DLM_ECANCEL)
6137 /* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6138 granted. Regardless of what rsb queue the lock is on, it's removed and
6139 freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6140 if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6142 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6144 struct dlm_args args;
6147 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6148 lkb->lkb_ua, &args);
6150 error = unlock_lock(ls, lkb, &args);
6151 if (error == -DLM_EUNLOCK)
6156 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6157 (which does lock_rsb) due to deadlock with receiving a message that does
6158 lock_rsb followed by dlm_user_add_cb() */
6160 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6161 struct dlm_user_proc *proc)
6163 struct dlm_lkb *lkb = NULL;
6165 spin_lock(&ls->ls_clear_proc_locks);
6166 if (list_empty(&proc->locks))
6169 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6170 list_del_init(&lkb->lkb_ownqueue);
6172 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6173 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6175 lkb->lkb_flags |= DLM_IFL_DEAD;
6177 spin_unlock(&ls->ls_clear_proc_locks);
6181 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6182 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6183 which we clear here. */
6185 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
6186 list, and no more device_writes should add lkb's to proc->locks list; so we
6187 shouldn't need to take asts_spin or locks_spin here. this assumes that
6188 device reads/writes/closes are serialized -- FIXME: we may need to serialize
6191 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6193 struct dlm_lkb *lkb, *safe;
6195 dlm_lock_recovery(ls);
6198 lkb = del_proc_lock(ls, proc);
6202 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6203 orphan_proc_lock(ls, lkb);
6205 unlock_proc_lock(ls, lkb);
6207 /* this removes the reference for the proc->locks list
6208 added by dlm_user_request, it may result in the lkb
6214 spin_lock(&ls->ls_clear_proc_locks);
6216 /* in-progress unlocks */
6217 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6218 list_del_init(&lkb->lkb_ownqueue);
6219 lkb->lkb_flags |= DLM_IFL_DEAD;
6223 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6224 memset(&lkb->lkb_callbacks, 0,
6225 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6226 list_del_init(&lkb->lkb_cb_list);
6230 spin_unlock(&ls->ls_clear_proc_locks);
6231 dlm_unlock_recovery(ls);
6234 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6236 struct dlm_lkb *lkb, *safe;
6240 spin_lock(&proc->locks_spin);
6241 if (!list_empty(&proc->locks)) {
6242 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6244 list_del_init(&lkb->lkb_ownqueue);
6246 spin_unlock(&proc->locks_spin);
6251 lkb->lkb_flags |= DLM_IFL_DEAD;
6252 unlock_proc_lock(ls, lkb);
6253 dlm_put_lkb(lkb); /* ref from proc->locks list */
6256 spin_lock(&proc->locks_spin);
6257 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6258 list_del_init(&lkb->lkb_ownqueue);
6259 lkb->lkb_flags |= DLM_IFL_DEAD;
6262 spin_unlock(&proc->locks_spin);
6264 spin_lock(&proc->asts_spin);
6265 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6266 memset(&lkb->lkb_callbacks, 0,
6267 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6268 list_del_init(&lkb->lkb_cb_list);
6271 spin_unlock(&proc->asts_spin);
6274 /* pid of 0 means purge all orphans */
6276 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6278 struct dlm_lkb *lkb, *safe;
6280 mutex_lock(&ls->ls_orphans_mutex);
6281 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6282 if (pid && lkb->lkb_ownpid != pid)
6284 unlock_proc_lock(ls, lkb);
6285 list_del_init(&lkb->lkb_ownqueue);
6288 mutex_unlock(&ls->ls_orphans_mutex);
6291 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6293 struct dlm_message *ms;
6294 struct dlm_mhandle *mh;
6297 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6298 DLM_MSG_PURGE, &ms, &mh);
6301 ms->m_nodeid = cpu_to_le32(nodeid);
6302 ms->m_pid = cpu_to_le32(pid);
6304 return send_message(mh, ms, NULL, 0);
6307 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6308 int nodeid, int pid)
6312 if (nodeid && (nodeid != dlm_our_nodeid())) {
6313 error = send_purge(ls, nodeid, pid);
6315 dlm_lock_recovery(ls);
6316 if (pid == current->pid)
6317 purge_proc_locks(ls, proc);
6319 do_purge(ls, nodeid, pid);
6320 dlm_unlock_recovery(ls);
6325 /* debug functionality */
6326 int dlm_debug_add_lkb(struct dlm_ls *ls, uint32_t lkb_id, char *name, int len,
6327 int lkb_nodeid, unsigned int lkb_flags, int lkb_status)
6329 struct dlm_lksb *lksb;
6330 struct dlm_lkb *lkb;
6334 /* we currently can't set a valid user lock */
6335 if (lkb_flags & DLM_IFL_USER)
6338 lksb = kzalloc(sizeof(*lksb), GFP_NOFS);
6342 error = _create_lkb(ls, &lkb, lkb_id, lkb_id + 1);
6348 lkb->lkb_flags = lkb_flags;
6349 lkb->lkb_nodeid = lkb_nodeid;
6350 lkb->lkb_lksb = lksb;
6351 /* user specific pointer, just don't have it NULL for kernel locks */
6352 if (~lkb_flags & DLM_IFL_USER)
6353 lkb->lkb_astparam = (void *)0xDEADBEEF;
6355 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
6364 add_lkb(r, lkb, lkb_status);
6371 int dlm_debug_add_lkb_to_waiters(struct dlm_ls *ls, uint32_t lkb_id,
6372 int mstype, int to_nodeid)
6374 struct dlm_lkb *lkb;
6377 error = find_lkb(ls, lkb_id, &lkb);
6381 error = add_to_waiters(lkb, mstype, to_nodeid);