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)
3616 dlm_midcomms_commit_mhandle(mh);
3620 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3621 struct dlm_message *ms)
3623 ms->m_nodeid = cpu_to_le32(lkb->lkb_nodeid);
3624 ms->m_pid = cpu_to_le32(lkb->lkb_ownpid);
3625 ms->m_lkid = cpu_to_le32(lkb->lkb_id);
3626 ms->m_remid = cpu_to_le32(lkb->lkb_remid);
3627 ms->m_exflags = cpu_to_le32(lkb->lkb_exflags);
3628 ms->m_sbflags = cpu_to_le32(lkb->lkb_sbflags);
3629 ms->m_flags = cpu_to_le32(lkb->lkb_flags);
3630 ms->m_lvbseq = cpu_to_le32(lkb->lkb_lvbseq);
3631 ms->m_status = cpu_to_le32(lkb->lkb_status);
3632 ms->m_grmode = cpu_to_le32(lkb->lkb_grmode);
3633 ms->m_rqmode = cpu_to_le32(lkb->lkb_rqmode);
3634 ms->m_hash = cpu_to_le32(r->res_hash);
3636 /* m_result and m_bastmode are set from function args,
3637 not from lkb fields */
3639 if (lkb->lkb_bastfn)
3640 ms->m_asts |= cpu_to_le32(DLM_CB_BAST);
3642 ms->m_asts |= cpu_to_le32(DLM_CB_CAST);
3644 /* compare with switch in create_message; send_remove() doesn't
3647 switch (ms->m_type) {
3648 case cpu_to_le32(DLM_MSG_REQUEST):
3649 case cpu_to_le32(DLM_MSG_LOOKUP):
3650 memcpy(ms->m_extra, r->res_name, r->res_length);
3652 case cpu_to_le32(DLM_MSG_CONVERT):
3653 case cpu_to_le32(DLM_MSG_UNLOCK):
3654 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3655 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3656 case cpu_to_le32(DLM_MSG_GRANT):
3657 if (!lkb->lkb_lvbptr || !(lkb->lkb_exflags & DLM_LKF_VALBLK))
3659 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3664 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3666 struct dlm_message *ms;
3667 struct dlm_mhandle *mh;
3668 int to_nodeid, error;
3670 to_nodeid = r->res_nodeid;
3672 error = add_to_waiters(lkb, mstype, to_nodeid);
3676 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3680 send_args(r, lkb, ms);
3682 error = send_message(mh, ms);
3688 remove_from_waiters(lkb, msg_reply_type(mstype));
3692 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3694 return send_common(r, lkb, DLM_MSG_REQUEST);
3697 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3701 error = send_common(r, lkb, DLM_MSG_CONVERT);
3703 /* down conversions go without a reply from the master */
3704 if (!error && down_conversion(lkb)) {
3705 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3706 r->res_ls->ls_stub_ms.m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
3707 r->res_ls->ls_stub_ms.m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
3708 r->res_ls->ls_stub_ms.m_result = 0;
3709 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3715 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
3716 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3717 that the master is still correct. */
3719 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3721 return send_common(r, lkb, DLM_MSG_UNLOCK);
3724 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3726 return send_common(r, lkb, DLM_MSG_CANCEL);
3729 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3731 struct dlm_message *ms;
3732 struct dlm_mhandle *mh;
3733 int to_nodeid, error;
3735 to_nodeid = lkb->lkb_nodeid;
3737 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3741 send_args(r, lkb, ms);
3745 error = send_message(mh, ms);
3750 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3752 struct dlm_message *ms;
3753 struct dlm_mhandle *mh;
3754 int to_nodeid, error;
3756 to_nodeid = lkb->lkb_nodeid;
3758 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3762 send_args(r, lkb, ms);
3764 ms->m_bastmode = cpu_to_le32(mode);
3766 error = send_message(mh, ms);
3771 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3773 struct dlm_message *ms;
3774 struct dlm_mhandle *mh;
3775 int to_nodeid, error;
3777 to_nodeid = dlm_dir_nodeid(r);
3779 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3783 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3787 send_args(r, lkb, ms);
3789 error = send_message(mh, ms);
3795 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3799 static int send_remove(struct dlm_rsb *r)
3801 struct dlm_message *ms;
3802 struct dlm_mhandle *mh;
3803 int to_nodeid, error;
3805 to_nodeid = dlm_dir_nodeid(r);
3807 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3811 memcpy(ms->m_extra, r->res_name, r->res_length);
3812 ms->m_hash = cpu_to_le32(r->res_hash);
3814 error = send_message(mh, ms);
3819 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3822 struct dlm_message *ms;
3823 struct dlm_mhandle *mh;
3824 int to_nodeid, error;
3826 to_nodeid = lkb->lkb_nodeid;
3828 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3832 send_args(r, lkb, ms);
3834 ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3836 error = send_message(mh, ms);
3841 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3843 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3846 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3848 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3851 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3853 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3856 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3858 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3861 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3862 int ret_nodeid, int rv)
3864 struct dlm_rsb *r = &ls->ls_stub_rsb;
3865 struct dlm_message *ms;
3866 struct dlm_mhandle *mh;
3867 int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid);
3869 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3873 ms->m_lkid = ms_in->m_lkid;
3874 ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3875 ms->m_nodeid = cpu_to_le32(ret_nodeid);
3877 error = send_message(mh, ms);
3882 /* which args we save from a received message depends heavily on the type
3883 of message, unlike the send side where we can safely send everything about
3884 the lkb for any type of message */
3886 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3888 lkb->lkb_exflags = le32_to_cpu(ms->m_exflags);
3889 lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
3890 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3891 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
3894 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3896 if (ms->m_flags == cpu_to_le32(DLM_IFL_STUB_MS))
3899 lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
3900 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3901 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
3904 static int receive_extralen(struct dlm_message *ms)
3906 return (le16_to_cpu(ms->m_header.h_length) -
3907 sizeof(struct dlm_message));
3910 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3911 struct dlm_message *ms)
3915 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3916 if (!lkb->lkb_lvbptr)
3917 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3918 if (!lkb->lkb_lvbptr)
3920 len = receive_extralen(ms);
3921 if (len > ls->ls_lvblen)
3922 len = ls->ls_lvblen;
3923 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3928 static void fake_bastfn(void *astparam, int mode)
3930 log_print("fake_bastfn should not be called");
3933 static void fake_astfn(void *astparam)
3935 log_print("fake_astfn should not be called");
3938 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3939 struct dlm_message *ms)
3941 lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3942 lkb->lkb_ownpid = le32_to_cpu(ms->m_pid);
3943 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3944 lkb->lkb_grmode = DLM_LOCK_IV;
3945 lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3947 lkb->lkb_bastfn = (ms->m_asts & cpu_to_le32(DLM_CB_BAST)) ? &fake_bastfn : NULL;
3948 lkb->lkb_astfn = (ms->m_asts & cpu_to_le32(DLM_CB_CAST)) ? &fake_astfn : NULL;
3950 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3951 /* lkb was just created so there won't be an lvb yet */
3952 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3953 if (!lkb->lkb_lvbptr)
3960 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3961 struct dlm_message *ms)
3963 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3966 if (receive_lvb(ls, lkb, ms))
3969 lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3970 lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
3975 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3976 struct dlm_message *ms)
3978 if (receive_lvb(ls, lkb, ms))
3983 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3984 uses to send a reply and that the remote end uses to process the reply. */
3986 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3988 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3989 lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3990 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3993 /* This is called after the rsb is locked so that we can safely inspect
3994 fields in the lkb. */
3996 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3998 int from = le32_to_cpu(ms->m_header.h_nodeid);
4001 /* currently mixing of user/kernel locks are not supported */
4002 if (ms->m_flags & cpu_to_le32(DLM_IFL_USER) &&
4003 ~lkb->lkb_flags & DLM_IFL_USER) {
4004 log_error(lkb->lkb_resource->res_ls,
4005 "got user dlm message for a kernel lock");
4010 switch (ms->m_type) {
4011 case cpu_to_le32(DLM_MSG_CONVERT):
4012 case cpu_to_le32(DLM_MSG_UNLOCK):
4013 case cpu_to_le32(DLM_MSG_CANCEL):
4014 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
4018 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4019 case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4020 case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4021 case cpu_to_le32(DLM_MSG_GRANT):
4022 case cpu_to_le32(DLM_MSG_BAST):
4023 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
4027 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
4028 if (!is_process_copy(lkb))
4030 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
4040 log_error(lkb->lkb_resource->res_ls,
4041 "ignore invalid message %d from %d %x %x %x %d",
4042 le32_to_cpu(ms->m_type), from, lkb->lkb_id,
4043 lkb->lkb_remid, lkb->lkb_flags, lkb->lkb_nodeid);
4047 static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4049 struct dlm_lkb *lkb;
4052 int error, namelen = 0;
4054 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4056 error = create_lkb(ls, &lkb);
4060 receive_flags(lkb, ms);
4061 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4062 error = receive_request_args(ls, lkb, ms);
4068 /* The dir node is the authority on whether we are the master
4069 for this rsb or not, so if the master sends us a request, we should
4070 recreate the rsb if we've destroyed it. This race happens when we
4071 send a remove message to the dir node at the same time that the dir
4072 node sends us a request for the rsb. */
4074 namelen = receive_extralen(ms);
4076 error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4077 R_RECEIVE_REQUEST, &r);
4085 if (r->res_master_nodeid != dlm_our_nodeid()) {
4086 error = validate_master_nodeid(ls, r, from_nodeid);
4096 error = do_request(r, lkb);
4097 send_request_reply(r, lkb, error);
4098 do_request_effects(r, lkb, error);
4103 if (error == -EINPROGRESS)
4110 /* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4111 and do this receive_request again from process_lookup_list once
4112 we get the lookup reply. This would avoid a many repeated
4113 ENOTBLK request failures when the lookup reply designating us
4114 as master is delayed. */
4116 if (error != -ENOTBLK) {
4117 log_limit(ls, "receive_request %x from %d %d",
4118 le32_to_cpu(ms->m_lkid), from_nodeid, error);
4121 setup_stub_lkb(ls, ms);
4122 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4126 static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4128 struct dlm_lkb *lkb;
4130 int error, reply = 1;
4132 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4136 if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4137 log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4138 "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4139 (unsigned long long)lkb->lkb_recover_seq,
4140 le32_to_cpu(ms->m_header.h_nodeid),
4141 le32_to_cpu(ms->m_lkid));
4147 r = lkb->lkb_resource;
4152 error = validate_message(lkb, ms);
4156 receive_flags(lkb, ms);
4158 error = receive_convert_args(ls, lkb, ms);
4160 send_convert_reply(r, lkb, error);
4164 reply = !down_conversion(lkb);
4166 error = do_convert(r, lkb);
4168 send_convert_reply(r, lkb, error);
4169 do_convert_effects(r, lkb, error);
4177 setup_stub_lkb(ls, ms);
4178 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4182 static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4184 struct dlm_lkb *lkb;
4188 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4192 if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4193 log_error(ls, "receive_unlock %x remid %x remote %d %x",
4194 lkb->lkb_id, lkb->lkb_remid,
4195 le32_to_cpu(ms->m_header.h_nodeid),
4196 le32_to_cpu(ms->m_lkid));
4202 r = lkb->lkb_resource;
4207 error = validate_message(lkb, ms);
4211 receive_flags(lkb, ms);
4213 error = receive_unlock_args(ls, lkb, ms);
4215 send_unlock_reply(r, lkb, error);
4219 error = do_unlock(r, lkb);
4220 send_unlock_reply(r, lkb, error);
4221 do_unlock_effects(r, lkb, error);
4229 setup_stub_lkb(ls, ms);
4230 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4234 static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4236 struct dlm_lkb *lkb;
4240 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4244 receive_flags(lkb, ms);
4246 r = lkb->lkb_resource;
4251 error = validate_message(lkb, ms);
4255 error = do_cancel(r, lkb);
4256 send_cancel_reply(r, lkb, error);
4257 do_cancel_effects(r, lkb, error);
4265 setup_stub_lkb(ls, ms);
4266 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4270 static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4272 struct dlm_lkb *lkb;
4276 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4280 r = lkb->lkb_resource;
4285 error = validate_message(lkb, ms);
4289 receive_flags_reply(lkb, ms);
4290 if (is_altmode(lkb))
4291 munge_altmode(lkb, ms);
4292 grant_lock_pc(r, lkb, ms);
4293 queue_cast(r, lkb, 0);
4301 static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4303 struct dlm_lkb *lkb;
4307 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4311 r = lkb->lkb_resource;
4316 error = validate_message(lkb, ms);
4320 queue_bast(r, lkb, le32_to_cpu(ms->m_bastmode));
4321 lkb->lkb_highbast = le32_to_cpu(ms->m_bastmode);
4329 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4331 int len, error, ret_nodeid, from_nodeid, our_nodeid;
4333 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4334 our_nodeid = dlm_our_nodeid();
4336 len = receive_extralen(ms);
4338 error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4341 /* Optimization: we're master so treat lookup as a request */
4342 if (!error && ret_nodeid == our_nodeid) {
4343 receive_request(ls, ms);
4346 send_lookup_reply(ls, ms, ret_nodeid, error);
4349 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4351 char name[DLM_RESNAME_MAXLEN+1];
4354 int rv, len, dir_nodeid, from_nodeid;
4356 from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4358 len = receive_extralen(ms);
4360 if (len > DLM_RESNAME_MAXLEN) {
4361 log_error(ls, "receive_remove from %d bad len %d",
4366 dir_nodeid = dlm_hash2nodeid(ls, le32_to_cpu(ms->m_hash));
4367 if (dir_nodeid != dlm_our_nodeid()) {
4368 log_error(ls, "receive_remove from %d bad nodeid %d",
4369 from_nodeid, dir_nodeid);
4373 /* Look for name on rsbtbl.toss, if it's there, kill it.
4374 If it's on rsbtbl.keep, it's being used, and we should ignore this
4375 message. This is an expected race between the dir node sending a
4376 request to the master node at the same time as the master node sends
4377 a remove to the dir node. The resolution to that race is for the
4378 dir node to ignore the remove message, and the master node to
4379 recreate the master rsb when it gets a request from the dir node for
4380 an rsb it doesn't have. */
4382 memset(name, 0, sizeof(name));
4383 memcpy(name, ms->m_extra, len);
4385 hash = jhash(name, len, 0);
4386 b = hash & (ls->ls_rsbtbl_size - 1);
4388 spin_lock(&ls->ls_rsbtbl[b].lock);
4390 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4392 /* verify the rsb is on keep list per comment above */
4393 rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4395 /* should not happen */
4396 log_error(ls, "receive_remove from %d not found %s",
4398 spin_unlock(&ls->ls_rsbtbl[b].lock);
4401 if (r->res_master_nodeid != from_nodeid) {
4402 /* should not happen */
4403 log_error(ls, "receive_remove keep from %d master %d",
4404 from_nodeid, r->res_master_nodeid);
4406 spin_unlock(&ls->ls_rsbtbl[b].lock);
4410 log_debug(ls, "receive_remove from %d master %d first %x %s",
4411 from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4413 spin_unlock(&ls->ls_rsbtbl[b].lock);
4417 if (r->res_master_nodeid != from_nodeid) {
4418 log_error(ls, "receive_remove toss from %d master %d",
4419 from_nodeid, r->res_master_nodeid);
4421 spin_unlock(&ls->ls_rsbtbl[b].lock);
4425 if (kref_put(&r->res_ref, kill_rsb)) {
4426 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4427 spin_unlock(&ls->ls_rsbtbl[b].lock);
4430 log_error(ls, "receive_remove from %d rsb ref error",
4433 spin_unlock(&ls->ls_rsbtbl[b].lock);
4437 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4439 do_purge(ls, le32_to_cpu(ms->m_nodeid), le32_to_cpu(ms->m_pid));
4442 static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4444 struct dlm_lkb *lkb;
4446 int error, mstype, result;
4447 int from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4449 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4453 r = lkb->lkb_resource;
4457 error = validate_message(lkb, ms);
4461 mstype = lkb->lkb_wait_type;
4462 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4464 log_error(ls, "receive_request_reply %x remote %d %x result %d",
4465 lkb->lkb_id, from_nodeid, le32_to_cpu(ms->m_lkid),
4466 from_dlm_errno(le32_to_cpu(ms->m_result)));
4471 /* Optimization: the dir node was also the master, so it took our
4472 lookup as a request and sent request reply instead of lookup reply */
4473 if (mstype == DLM_MSG_LOOKUP) {
4474 r->res_master_nodeid = from_nodeid;
4475 r->res_nodeid = from_nodeid;
4476 lkb->lkb_nodeid = from_nodeid;
4479 /* this is the value returned from do_request() on the master */
4480 result = from_dlm_errno(le32_to_cpu(ms->m_result));
4484 /* request would block (be queued) on remote master */
4485 queue_cast(r, lkb, -EAGAIN);
4486 confirm_master(r, -EAGAIN);
4487 unhold_lkb(lkb); /* undoes create_lkb() */
4492 /* request was queued or granted on remote master */
4493 receive_flags_reply(lkb, ms);
4494 lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
4495 if (is_altmode(lkb))
4496 munge_altmode(lkb, ms);
4498 add_lkb(r, lkb, DLM_LKSTS_WAITING);
4501 grant_lock_pc(r, lkb, ms);
4502 queue_cast(r, lkb, 0);
4504 confirm_master(r, result);
4509 /* find_rsb failed to find rsb or rsb wasn't master */
4510 log_limit(ls, "receive_request_reply %x from %d %d "
4511 "master %d dir %d first %x %s", lkb->lkb_id,
4512 from_nodeid, result, r->res_master_nodeid,
4513 r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4515 if (r->res_dir_nodeid != dlm_our_nodeid() &&
4516 r->res_master_nodeid != dlm_our_nodeid()) {
4517 /* cause _request_lock->set_master->send_lookup */
4518 r->res_master_nodeid = 0;
4520 lkb->lkb_nodeid = -1;
4523 if (is_overlap(lkb)) {
4524 /* we'll ignore error in cancel/unlock reply */
4525 queue_cast_overlap(r, lkb);
4526 confirm_master(r, result);
4527 unhold_lkb(lkb); /* undoes create_lkb() */
4529 _request_lock(r, lkb);
4531 if (r->res_master_nodeid == dlm_our_nodeid())
4532 confirm_master(r, 0);
4537 log_error(ls, "receive_request_reply %x error %d",
4538 lkb->lkb_id, result);
4541 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4542 log_debug(ls, "receive_request_reply %x result %d unlock",
4543 lkb->lkb_id, result);
4544 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4545 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4546 send_unlock(r, lkb);
4547 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4548 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4549 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4550 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4551 send_cancel(r, lkb);
4553 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4554 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4563 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4564 struct dlm_message *ms)
4566 /* this is the value returned from do_convert() on the master */
4567 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4569 /* convert would block (be queued) on remote master */
4570 queue_cast(r, lkb, -EAGAIN);
4574 receive_flags_reply(lkb, ms);
4575 revert_lock_pc(r, lkb);
4576 queue_cast(r, lkb, -EDEADLK);
4580 /* convert was queued on remote master */
4581 receive_flags_reply(lkb, ms);
4582 if (is_demoted(lkb))
4585 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4590 /* convert was granted on remote master */
4591 receive_flags_reply(lkb, ms);
4592 if (is_demoted(lkb))
4594 grant_lock_pc(r, lkb, ms);
4595 queue_cast(r, lkb, 0);
4599 log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4600 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4601 le32_to_cpu(ms->m_lkid),
4602 from_dlm_errno(le32_to_cpu(ms->m_result)));
4608 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4610 struct dlm_rsb *r = lkb->lkb_resource;
4616 error = validate_message(lkb, ms);
4620 /* stub reply can happen with waiters_mutex held */
4621 error = remove_from_waiters_ms(lkb, ms);
4625 __receive_convert_reply(r, lkb, ms);
4631 static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4633 struct dlm_lkb *lkb;
4636 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4640 _receive_convert_reply(lkb, ms);
4645 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4647 struct dlm_rsb *r = lkb->lkb_resource;
4653 error = validate_message(lkb, ms);
4657 /* stub reply can happen with waiters_mutex held */
4658 error = remove_from_waiters_ms(lkb, ms);
4662 /* this is the value returned from do_unlock() on the master */
4664 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4666 receive_flags_reply(lkb, ms);
4667 remove_lock_pc(r, lkb);
4668 queue_cast(r, lkb, -DLM_EUNLOCK);
4673 log_error(r->res_ls, "receive_unlock_reply %x error %d",
4674 lkb->lkb_id, from_dlm_errno(le32_to_cpu(ms->m_result)));
4681 static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4683 struct dlm_lkb *lkb;
4686 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4690 _receive_unlock_reply(lkb, ms);
4695 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4697 struct dlm_rsb *r = lkb->lkb_resource;
4703 error = validate_message(lkb, ms);
4707 /* stub reply can happen with waiters_mutex held */
4708 error = remove_from_waiters_ms(lkb, ms);
4712 /* this is the value returned from do_cancel() on the master */
4714 switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4716 receive_flags_reply(lkb, ms);
4717 revert_lock_pc(r, lkb);
4718 queue_cast(r, lkb, -DLM_ECANCEL);
4723 log_error(r->res_ls, "receive_cancel_reply %x error %d",
4725 from_dlm_errno(le32_to_cpu(ms->m_result)));
4732 static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4734 struct dlm_lkb *lkb;
4737 error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4741 _receive_cancel_reply(lkb, ms);
4746 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4748 struct dlm_lkb *lkb;
4750 int error, ret_nodeid;
4751 int do_lookup_list = 0;
4753 error = find_lkb(ls, le32_to_cpu(ms->m_lkid), &lkb);
4755 log_error(ls, "%s no lkid %x", __func__,
4756 le32_to_cpu(ms->m_lkid));
4760 /* ms->m_result is the value returned by dlm_master_lookup on dir node
4761 FIXME: will a non-zero error ever be returned? */
4763 r = lkb->lkb_resource;
4767 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4771 ret_nodeid = le32_to_cpu(ms->m_nodeid);
4773 /* We sometimes receive a request from the dir node for this
4774 rsb before we've received the dir node's loookup_reply for it.
4775 The request from the dir node implies we're the master, so we set
4776 ourself as master in receive_request_reply, and verify here that
4777 we are indeed the master. */
4779 if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4780 /* This should never happen */
4781 log_error(ls, "receive_lookup_reply %x from %d ret %d "
4782 "master %d dir %d our %d first %x %s",
4783 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4784 ret_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
4785 dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4788 if (ret_nodeid == dlm_our_nodeid()) {
4789 r->res_master_nodeid = ret_nodeid;
4792 r->res_first_lkid = 0;
4793 } else if (ret_nodeid == -1) {
4794 /* the remote node doesn't believe it's the dir node */
4795 log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4796 lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid));
4797 r->res_master_nodeid = 0;
4799 lkb->lkb_nodeid = -1;
4801 /* set_master() will set lkb_nodeid from r */
4802 r->res_master_nodeid = ret_nodeid;
4803 r->res_nodeid = ret_nodeid;
4806 if (is_overlap(lkb)) {
4807 log_debug(ls, "receive_lookup_reply %x unlock %x",
4808 lkb->lkb_id, lkb->lkb_flags);
4809 queue_cast_overlap(r, lkb);
4810 unhold_lkb(lkb); /* undoes create_lkb() */
4814 _request_lock(r, lkb);
4818 process_lookup_list(r);
4825 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4828 int error = 0, noent = 0;
4830 if (!dlm_is_member(ls, le32_to_cpu(ms->m_header.h_nodeid))) {
4831 log_limit(ls, "receive %d from non-member %d %x %x %d",
4832 le32_to_cpu(ms->m_type),
4833 le32_to_cpu(ms->m_header.h_nodeid),
4834 le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4835 from_dlm_errno(le32_to_cpu(ms->m_result)));
4839 switch (ms->m_type) {
4841 /* messages sent to a master node */
4843 case cpu_to_le32(DLM_MSG_REQUEST):
4844 error = receive_request(ls, ms);
4847 case cpu_to_le32(DLM_MSG_CONVERT):
4848 error = receive_convert(ls, ms);
4851 case cpu_to_le32(DLM_MSG_UNLOCK):
4852 error = receive_unlock(ls, ms);
4855 case cpu_to_le32(DLM_MSG_CANCEL):
4857 error = receive_cancel(ls, ms);
4860 /* messages sent from a master node (replies to above) */
4862 case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
4863 error = receive_request_reply(ls, ms);
4866 case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4867 error = receive_convert_reply(ls, ms);
4870 case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4871 error = receive_unlock_reply(ls, ms);
4874 case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4875 error = receive_cancel_reply(ls, ms);
4878 /* messages sent from a master node (only two types of async msg) */
4880 case cpu_to_le32(DLM_MSG_GRANT):
4882 error = receive_grant(ls, ms);
4885 case cpu_to_le32(DLM_MSG_BAST):
4887 error = receive_bast(ls, ms);
4890 /* messages sent to a dir node */
4892 case cpu_to_le32(DLM_MSG_LOOKUP):
4893 receive_lookup(ls, ms);
4896 case cpu_to_le32(DLM_MSG_REMOVE):
4897 receive_remove(ls, ms);
4900 /* messages sent from a dir node (remove has no reply) */
4902 case cpu_to_le32(DLM_MSG_LOOKUP_REPLY):
4903 receive_lookup_reply(ls, ms);
4906 /* other messages */
4908 case cpu_to_le32(DLM_MSG_PURGE):
4909 receive_purge(ls, ms);
4913 log_error(ls, "unknown message type %d",
4914 le32_to_cpu(ms->m_type));
4918 * When checking for ENOENT, we're checking the result of
4919 * find_lkb(m_remid):
4921 * The lock id referenced in the message wasn't found. This may
4922 * happen in normal usage for the async messages and cancel, so
4923 * only use log_debug for them.
4925 * Some errors are expected and normal.
4928 if (error == -ENOENT && noent) {
4929 log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4930 le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4931 le32_to_cpu(ms->m_header.h_nodeid),
4932 le32_to_cpu(ms->m_lkid), saved_seq);
4933 } else if (error == -ENOENT) {
4934 log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4935 le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4936 le32_to_cpu(ms->m_header.h_nodeid),
4937 le32_to_cpu(ms->m_lkid), saved_seq);
4939 if (ms->m_type == cpu_to_le32(DLM_MSG_CONVERT))
4940 dlm_dump_rsb_hash(ls, le32_to_cpu(ms->m_hash));
4943 if (error == -EINVAL) {
4944 log_error(ls, "receive %d inval from %d lkid %x remid %x "
4946 le32_to_cpu(ms->m_type),
4947 le32_to_cpu(ms->m_header.h_nodeid),
4948 le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4953 /* If the lockspace is in recovery mode (locking stopped), then normal
4954 messages are saved on the requestqueue for processing after recovery is
4955 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4956 messages off the requestqueue before we process new ones. This occurs right
4957 after recovery completes when we transition from saving all messages on
4958 requestqueue, to processing all the saved messages, to processing new
4959 messages as they arrive. */
4961 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4964 if (dlm_locking_stopped(ls)) {
4965 /* If we were a member of this lockspace, left, and rejoined,
4966 other nodes may still be sending us messages from the
4967 lockspace generation before we left. */
4968 if (!ls->ls_generation) {
4969 log_limit(ls, "receive %d from %d ignore old gen",
4970 le32_to_cpu(ms->m_type), nodeid);
4974 dlm_add_requestqueue(ls, nodeid, ms);
4976 dlm_wait_requestqueue(ls);
4977 _receive_message(ls, ms, 0);
4981 /* This is called by dlm_recoverd to process messages that were saved on
4982 the requestqueue. */
4984 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
4987 _receive_message(ls, ms, saved_seq);
4990 /* This is called by the midcomms layer when something is received for
4991 the lockspace. It could be either a MSG (normal message sent as part of
4992 standard locking activity) or an RCOM (recovery message sent as part of
4993 lockspace recovery). */
4995 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
4997 struct dlm_header *hd = &p->header;
5001 switch (hd->h_cmd) {
5003 type = le32_to_cpu(p->message.m_type);
5006 type = le32_to_cpu(p->rcom.rc_type);
5009 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
5013 if (le32_to_cpu(hd->h_nodeid) != nodeid) {
5014 log_print("invalid h_nodeid %d from %d lockspace %x",
5015 le32_to_cpu(hd->h_nodeid), nodeid,
5016 le32_to_cpu(hd->u.h_lockspace));
5020 ls = dlm_find_lockspace_global(le32_to_cpu(hd->u.h_lockspace));
5022 if (dlm_config.ci_log_debug) {
5023 printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
5024 "%u from %d cmd %d type %d\n",
5025 le32_to_cpu(hd->u.h_lockspace), nodeid,
5029 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
5030 dlm_send_ls_not_ready(nodeid, &p->rcom);
5034 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
5035 be inactive (in this ls) before transitioning to recovery mode */
5037 down_read(&ls->ls_recv_active);
5038 if (hd->h_cmd == DLM_MSG)
5039 dlm_receive_message(ls, &p->message, nodeid);
5040 else if (hd->h_cmd == DLM_RCOM)
5041 dlm_receive_rcom(ls, &p->rcom, nodeid);
5043 log_error(ls, "invalid h_cmd %d from %d lockspace %x",
5044 hd->h_cmd, nodeid, le32_to_cpu(hd->u.h_lockspace));
5045 up_read(&ls->ls_recv_active);
5047 dlm_put_lockspace(ls);
5050 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5051 struct dlm_message *ms_stub)
5053 if (middle_conversion(lkb)) {
5055 memset(ms_stub, 0, sizeof(struct dlm_message));
5056 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5057 ms_stub->m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
5058 ms_stub->m_result = cpu_to_le32(to_dlm_errno(-EINPROGRESS));
5059 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5060 _receive_convert_reply(lkb, ms_stub);
5062 /* Same special case as in receive_rcom_lock_args() */
5063 lkb->lkb_grmode = DLM_LOCK_IV;
5064 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5067 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5068 lkb->lkb_flags |= DLM_IFL_RESEND;
5071 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5072 conversions are async; there's no reply from the remote master */
5075 /* A waiting lkb needs recovery if the master node has failed, or
5076 the master node is changing (only when no directory is used) */
5078 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5081 if (dlm_no_directory(ls))
5084 if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5090 /* Recovery for locks that are waiting for replies from nodes that are now
5091 gone. We can just complete unlocks and cancels by faking a reply from the
5092 dead node. Requests and up-conversions we flag to be resent after
5093 recovery. Down-conversions can just be completed with a fake reply like
5094 unlocks. Conversions between PR and CW need special attention. */
5096 void dlm_recover_waiters_pre(struct dlm_ls *ls)
5098 struct dlm_lkb *lkb, *safe;
5099 struct dlm_message *ms_stub;
5100 int wait_type, stub_unlock_result, stub_cancel_result;
5103 ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5107 mutex_lock(&ls->ls_waiters_mutex);
5109 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5111 dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5113 /* exclude debug messages about unlocks because there can be so
5114 many and they aren't very interesting */
5116 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5117 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5118 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5122 lkb->lkb_resource->res_nodeid,
5124 lkb->lkb_wait_nodeid,
5128 /* all outstanding lookups, regardless of destination will be
5129 resent after recovery is done */
5131 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5132 lkb->lkb_flags |= DLM_IFL_RESEND;
5136 if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5139 wait_type = lkb->lkb_wait_type;
5140 stub_unlock_result = -DLM_EUNLOCK;
5141 stub_cancel_result = -DLM_ECANCEL;
5143 /* Main reply may have been received leaving a zero wait_type,
5144 but a reply for the overlapping op may not have been
5145 received. In that case we need to fake the appropriate
5146 reply for the overlap op. */
5149 if (is_overlap_cancel(lkb)) {
5150 wait_type = DLM_MSG_CANCEL;
5151 if (lkb->lkb_grmode == DLM_LOCK_IV)
5152 stub_cancel_result = 0;
5154 if (is_overlap_unlock(lkb)) {
5155 wait_type = DLM_MSG_UNLOCK;
5156 if (lkb->lkb_grmode == DLM_LOCK_IV)
5157 stub_unlock_result = -ENOENT;
5160 log_debug(ls, "rwpre overlap %x %x %d %d %d",
5161 lkb->lkb_id, lkb->lkb_flags, wait_type,
5162 stub_cancel_result, stub_unlock_result);
5165 switch (wait_type) {
5167 case DLM_MSG_REQUEST:
5168 lkb->lkb_flags |= DLM_IFL_RESEND;
5171 case DLM_MSG_CONVERT:
5172 recover_convert_waiter(ls, lkb, ms_stub);
5175 case DLM_MSG_UNLOCK:
5177 memset(ms_stub, 0, sizeof(struct dlm_message));
5178 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5179 ms_stub->m_type = cpu_to_le32(DLM_MSG_UNLOCK_REPLY);
5180 ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_unlock_result));
5181 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5182 _receive_unlock_reply(lkb, ms_stub);
5186 case DLM_MSG_CANCEL:
5188 memset(ms_stub, 0, sizeof(struct dlm_message));
5189 ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5190 ms_stub->m_type = cpu_to_le32(DLM_MSG_CANCEL_REPLY);
5191 ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_cancel_result));
5192 ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5193 _receive_cancel_reply(lkb, ms_stub);
5198 log_error(ls, "invalid lkb wait_type %d %d",
5199 lkb->lkb_wait_type, wait_type);
5203 mutex_unlock(&ls->ls_waiters_mutex);
5207 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5209 struct dlm_lkb *lkb = NULL, *iter;
5211 mutex_lock(&ls->ls_waiters_mutex);
5212 list_for_each_entry(iter, &ls->ls_waiters, lkb_wait_reply) {
5213 if (iter->lkb_flags & DLM_IFL_RESEND) {
5219 mutex_unlock(&ls->ls_waiters_mutex);
5224 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
5225 master or dir-node for r. Processing the lkb may result in it being placed
5228 /* We do this after normal locking has been enabled and any saved messages
5229 (in requestqueue) have been processed. We should be confident that at
5230 this point we won't get or process a reply to any of these waiting
5231 operations. But, new ops may be coming in on the rsbs/locks here from
5232 userspace or remotely. */
5234 /* there may have been an overlap unlock/cancel prior to recovery or after
5235 recovery. if before, the lkb may still have a pos wait_count; if after, the
5236 overlap flag would just have been set and nothing new sent. we can be
5237 confident here than any replies to either the initial op or overlap ops
5238 prior to recovery have been received. */
5240 int dlm_recover_waiters_post(struct dlm_ls *ls)
5242 struct dlm_lkb *lkb;
5244 int error = 0, mstype, err, oc, ou;
5247 if (dlm_locking_stopped(ls)) {
5248 log_debug(ls, "recover_waiters_post aborted");
5253 lkb = find_resend_waiter(ls);
5257 r = lkb->lkb_resource;
5261 mstype = lkb->lkb_wait_type;
5262 oc = is_overlap_cancel(lkb);
5263 ou = is_overlap_unlock(lkb);
5266 log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5267 "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5268 "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5269 r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5270 dlm_dir_nodeid(r), oc, ou);
5272 /* At this point we assume that we won't get a reply to any
5273 previous op or overlap op on this lock. First, do a big
5274 remove_from_waiters() for all previous ops. */
5276 lkb->lkb_flags &= ~DLM_IFL_RESEND;
5277 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5278 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5279 lkb->lkb_wait_type = 0;
5280 /* drop all wait_count references we still
5281 * hold a reference for this iteration.
5283 while (lkb->lkb_wait_count) {
5284 lkb->lkb_wait_count--;
5287 mutex_lock(&ls->ls_waiters_mutex);
5288 list_del_init(&lkb->lkb_wait_reply);
5289 mutex_unlock(&ls->ls_waiters_mutex);
5292 /* do an unlock or cancel instead of resending */
5294 case DLM_MSG_LOOKUP:
5295 case DLM_MSG_REQUEST:
5296 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5298 unhold_lkb(lkb); /* undoes create_lkb() */
5300 case DLM_MSG_CONVERT:
5302 queue_cast(r, lkb, -DLM_ECANCEL);
5304 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5305 _unlock_lock(r, lkb);
5313 case DLM_MSG_LOOKUP:
5314 case DLM_MSG_REQUEST:
5315 _request_lock(r, lkb);
5317 confirm_master(r, 0);
5319 case DLM_MSG_CONVERT:
5320 _convert_lock(r, lkb);
5328 log_error(ls, "waiter %x msg %d r_nodeid %d "
5329 "dir_nodeid %d overlap %d %d",
5330 lkb->lkb_id, mstype, r->res_nodeid,
5331 dlm_dir_nodeid(r), oc, ou);
5341 static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5342 struct list_head *list)
5344 struct dlm_lkb *lkb, *safe;
5346 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5347 if (!is_master_copy(lkb))
5350 /* don't purge lkbs we've added in recover_master_copy for
5351 the current recovery seq */
5353 if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5358 /* this put should free the lkb */
5359 if (!dlm_put_lkb(lkb))
5360 log_error(ls, "purged mstcpy lkb not released");
5364 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5366 struct dlm_ls *ls = r->res_ls;
5368 purge_mstcpy_list(ls, r, &r->res_grantqueue);
5369 purge_mstcpy_list(ls, r, &r->res_convertqueue);
5370 purge_mstcpy_list(ls, r, &r->res_waitqueue);
5373 static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5374 struct list_head *list,
5375 int nodeid_gone, unsigned int *count)
5377 struct dlm_lkb *lkb, *safe;
5379 list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5380 if (!is_master_copy(lkb))
5383 if ((lkb->lkb_nodeid == nodeid_gone) ||
5384 dlm_is_removed(ls, lkb->lkb_nodeid)) {
5386 /* tell recover_lvb to invalidate the lvb
5387 because a node holding EX/PW failed */
5388 if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5389 (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5390 rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5395 /* this put should free the lkb */
5396 if (!dlm_put_lkb(lkb))
5397 log_error(ls, "purged dead lkb not released");
5399 rsb_set_flag(r, RSB_RECOVER_GRANT);
5406 /* Get rid of locks held by nodes that are gone. */
5408 void dlm_recover_purge(struct dlm_ls *ls)
5411 struct dlm_member *memb;
5412 int nodes_count = 0;
5413 int nodeid_gone = 0;
5414 unsigned int lkb_count = 0;
5416 /* cache one removed nodeid to optimize the common
5417 case of a single node removed */
5419 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5421 nodeid_gone = memb->nodeid;
5427 down_write(&ls->ls_root_sem);
5428 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5432 purge_dead_list(ls, r, &r->res_grantqueue,
5433 nodeid_gone, &lkb_count);
5434 purge_dead_list(ls, r, &r->res_convertqueue,
5435 nodeid_gone, &lkb_count);
5436 purge_dead_list(ls, r, &r->res_waitqueue,
5437 nodeid_gone, &lkb_count);
5443 up_write(&ls->ls_root_sem);
5446 log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5447 lkb_count, nodes_count);
5450 static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5455 spin_lock(&ls->ls_rsbtbl[bucket].lock);
5456 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5457 r = rb_entry(n, struct dlm_rsb, res_hashnode);
5459 if (!rsb_flag(r, RSB_RECOVER_GRANT))
5461 if (!is_master(r)) {
5462 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5466 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5469 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5474 * Attempt to grant locks on resources that we are the master of.
5475 * Locks may have become grantable during recovery because locks
5476 * from departed nodes have been purged (or not rebuilt), allowing
5477 * previously blocked locks to now be granted. The subset of rsb's
5478 * we are interested in are those with lkb's on either the convert or
5481 * Simplest would be to go through each master rsb and check for non-empty
5482 * convert or waiting queues, and attempt to grant on those rsbs.
5483 * Checking the queues requires lock_rsb, though, for which we'd need
5484 * to release the rsbtbl lock. This would make iterating through all
5485 * rsb's very inefficient. So, we rely on earlier recovery routines
5486 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5490 void dlm_recover_grant(struct dlm_ls *ls)
5494 unsigned int count = 0;
5495 unsigned int rsb_count = 0;
5496 unsigned int lkb_count = 0;
5499 r = find_grant_rsb(ls, bucket);
5501 if (bucket == ls->ls_rsbtbl_size - 1)
5509 /* the RECOVER_GRANT flag is checked in the grant path */
5510 grant_pending_locks(r, &count);
5511 rsb_clear_flag(r, RSB_RECOVER_GRANT);
5513 confirm_master(r, 0);
5520 log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5521 lkb_count, rsb_count);
5524 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5527 struct dlm_lkb *lkb;
5529 list_for_each_entry(lkb, head, lkb_statequeue) {
5530 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5536 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5539 struct dlm_lkb *lkb;
5541 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5544 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5547 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5553 /* needs at least dlm_rcom + rcom_lock */
5554 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5555 struct dlm_rsb *r, struct dlm_rcom *rc)
5557 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5559 lkb->lkb_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5560 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5561 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5562 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5563 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5564 lkb->lkb_flags |= DLM_IFL_MSTCPY;
5565 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5566 lkb->lkb_rqmode = rl->rl_rqmode;
5567 lkb->lkb_grmode = rl->rl_grmode;
5568 /* don't set lkb_status because add_lkb wants to itself */
5570 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5571 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5573 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5574 int lvblen = le16_to_cpu(rc->rc_header.h_length) -
5575 sizeof(struct dlm_rcom) - sizeof(struct rcom_lock);
5576 if (lvblen > ls->ls_lvblen)
5578 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5579 if (!lkb->lkb_lvbptr)
5581 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5584 /* Conversions between PR and CW (middle modes) need special handling.
5585 The real granted mode of these converting locks cannot be determined
5586 until all locks have been rebuilt on the rsb (recover_conversion) */
5588 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5589 middle_conversion(lkb)) {
5590 rl->rl_status = DLM_LKSTS_CONVERT;
5591 lkb->lkb_grmode = DLM_LOCK_IV;
5592 rsb_set_flag(r, RSB_RECOVER_CONVERT);
5598 /* This lkb may have been recovered in a previous aborted recovery so we need
5599 to check if the rsb already has an lkb with the given remote nodeid/lkid.
5600 If so we just send back a standard reply. If not, we create a new lkb with
5601 the given values and send back our lkid. We send back our lkid by sending
5602 back the rcom_lock struct we got but with the remid field filled in. */
5604 /* needs at least dlm_rcom + rcom_lock */
5605 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5607 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5609 struct dlm_lkb *lkb;
5611 int from_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5614 if (rl->rl_parent_lkid) {
5615 error = -EOPNOTSUPP;
5619 remid = le32_to_cpu(rl->rl_lkid);
5621 /* In general we expect the rsb returned to be R_MASTER, but we don't
5622 have to require it. Recovery of masters on one node can overlap
5623 recovery of locks on another node, so one node can send us MSTCPY
5624 locks before we've made ourselves master of this rsb. We can still
5625 add new MSTCPY locks that we receive here without any harm; when
5626 we make ourselves master, dlm_recover_masters() won't touch the
5627 MSTCPY locks we've received early. */
5629 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5630 from_nodeid, R_RECEIVE_RECOVER, &r);
5636 if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5637 log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5638 from_nodeid, remid);
5643 lkb = search_remid(r, from_nodeid, remid);
5649 error = create_lkb(ls, &lkb);
5653 error = receive_rcom_lock_args(ls, lkb, r, rc);
5660 add_lkb(r, lkb, rl->rl_status);
5661 ls->ls_recover_locks_in++;
5663 if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5664 rsb_set_flag(r, RSB_RECOVER_GRANT);
5667 /* this is the new value returned to the lock holder for
5668 saving in its process-copy lkb */
5669 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5671 lkb->lkb_recover_seq = ls->ls_recover_seq;
5677 if (error && error != -EEXIST)
5678 log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5679 from_nodeid, remid, error);
5680 rl->rl_result = cpu_to_le32(error);
5684 /* needs at least dlm_rcom + rcom_lock */
5685 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5687 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5689 struct dlm_lkb *lkb;
5690 uint32_t lkid, remid;
5693 lkid = le32_to_cpu(rl->rl_lkid);
5694 remid = le32_to_cpu(rl->rl_remid);
5695 result = le32_to_cpu(rl->rl_result);
5697 error = find_lkb(ls, lkid, &lkb);
5699 log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5700 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5705 r = lkb->lkb_resource;
5709 if (!is_process_copy(lkb)) {
5710 log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5711 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5722 /* There's a chance the new master received our lock before
5723 dlm_recover_master_reply(), this wouldn't happen if we did
5724 a barrier between recover_masters and recover_locks. */
5726 log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5727 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5730 dlm_send_rcom_lock(r, lkb);
5734 lkb->lkb_remid = remid;
5737 log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5738 lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5742 /* an ack for dlm_recover_locks() which waits for replies from
5743 all the locks it sends to new masters */
5744 dlm_recovered_lock(r);
5753 #ifdef CONFIG_DLM_DEPRECATED_API
5754 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5755 int mode, uint32_t flags, void *name, unsigned int namelen,
5756 unsigned long timeout_cs)
5758 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5759 int mode, uint32_t flags, void *name, unsigned int namelen)
5762 struct dlm_lkb *lkb;
5763 struct dlm_args args;
5767 dlm_lock_recovery(ls);
5769 error = create_lkb(ls, &lkb);
5775 trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
5777 if (flags & DLM_LKF_VALBLK) {
5778 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5779 if (!ua->lksb.sb_lvbptr) {
5785 #ifdef CONFIG_DLM_DEPRECATED_API
5786 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5787 fake_astfn, ua, fake_bastfn, &args);
5789 error = set_lock_args(mode, &ua->lksb, flags, namelen, fake_astfn, ua,
5790 fake_bastfn, &args);
5793 kfree(ua->lksb.sb_lvbptr);
5794 ua->lksb.sb_lvbptr = NULL;
5799 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
5800 When DLM_IFL_USER is set, the dlm knows that this is a userspace
5801 lock and that lkb_astparam is the dlm_user_args structure. */
5802 lkb->lkb_flags |= DLM_IFL_USER;
5803 error = request_lock(ls, lkb, name, namelen, &args);
5818 /* add this new lkb to the per-process list of locks */
5819 spin_lock(&ua->proc->locks_spin);
5821 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5822 spin_unlock(&ua->proc->locks_spin);
5825 trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, false);
5829 dlm_unlock_recovery(ls);
5833 #ifdef CONFIG_DLM_DEPRECATED_API
5834 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5835 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5836 unsigned long timeout_cs)
5838 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5839 int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
5842 struct dlm_lkb *lkb;
5843 struct dlm_args args;
5844 struct dlm_user_args *ua;
5847 dlm_lock_recovery(ls);
5849 error = find_lkb(ls, lkid, &lkb);
5853 trace_dlm_lock_start(ls, lkb, NULL, 0, mode, flags);
5855 /* user can change the params on its lock when it converts it, or
5856 add an lvb that didn't exist before */
5860 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5861 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5862 if (!ua->lksb.sb_lvbptr) {
5867 if (lvb_in && ua->lksb.sb_lvbptr)
5868 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5870 ua->xid = ua_tmp->xid;
5871 ua->castparam = ua_tmp->castparam;
5872 ua->castaddr = ua_tmp->castaddr;
5873 ua->bastparam = ua_tmp->bastparam;
5874 ua->bastaddr = ua_tmp->bastaddr;
5875 ua->user_lksb = ua_tmp->user_lksb;
5877 #ifdef CONFIG_DLM_DEPRECATED_API
5878 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5879 fake_astfn, ua, fake_bastfn, &args);
5881 error = set_lock_args(mode, &ua->lksb, flags, 0, fake_astfn, ua,
5882 fake_bastfn, &args);
5887 error = convert_lock(ls, lkb, &args);
5889 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5892 trace_dlm_lock_end(ls, lkb, NULL, 0, mode, flags, error, false);
5895 dlm_unlock_recovery(ls);
5901 * The caller asks for an orphan lock on a given resource with a given mode.
5902 * If a matching lock exists, it's moved to the owner's list of locks and
5903 * the lkid is returned.
5906 int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5907 int mode, uint32_t flags, void *name, unsigned int namelen,
5910 struct dlm_lkb *lkb = NULL, *iter;
5911 struct dlm_user_args *ua;
5912 int found_other_mode = 0;
5915 mutex_lock(&ls->ls_orphans_mutex);
5916 list_for_each_entry(iter, &ls->ls_orphans, lkb_ownqueue) {
5917 if (iter->lkb_resource->res_length != namelen)
5919 if (memcmp(iter->lkb_resource->res_name, name, namelen))
5921 if (iter->lkb_grmode != mode) {
5922 found_other_mode = 1;
5927 list_del_init(&iter->lkb_ownqueue);
5928 iter->lkb_flags &= ~DLM_IFL_ORPHAN;
5929 *lkid = iter->lkb_id;
5932 mutex_unlock(&ls->ls_orphans_mutex);
5934 if (!lkb && found_other_mode) {
5944 lkb->lkb_exflags = flags;
5945 lkb->lkb_ownpid = (int) current->pid;
5949 ua->proc = ua_tmp->proc;
5950 ua->xid = ua_tmp->xid;
5951 ua->castparam = ua_tmp->castparam;
5952 ua->castaddr = ua_tmp->castaddr;
5953 ua->bastparam = ua_tmp->bastparam;
5954 ua->bastaddr = ua_tmp->bastaddr;
5955 ua->user_lksb = ua_tmp->user_lksb;
5958 * The lkb reference from the ls_orphans list was not
5959 * removed above, and is now considered the reference
5960 * for the proc locks list.
5963 spin_lock(&ua->proc->locks_spin);
5964 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5965 spin_unlock(&ua->proc->locks_spin);
5971 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5972 uint32_t flags, uint32_t lkid, char *lvb_in)
5974 struct dlm_lkb *lkb;
5975 struct dlm_args args;
5976 struct dlm_user_args *ua;
5979 dlm_lock_recovery(ls);
5981 error = find_lkb(ls, lkid, &lkb);
5985 trace_dlm_unlock_start(ls, lkb, flags);
5989 if (lvb_in && ua->lksb.sb_lvbptr)
5990 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5991 if (ua_tmp->castparam)
5992 ua->castparam = ua_tmp->castparam;
5993 ua->user_lksb = ua_tmp->user_lksb;
5995 error = set_unlock_args(flags, ua, &args);
5999 error = unlock_lock(ls, lkb, &args);
6001 if (error == -DLM_EUNLOCK)
6003 /* from validate_unlock_args() */
6004 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
6009 spin_lock(&ua->proc->locks_spin);
6010 /* dlm_user_add_cb() may have already taken lkb off the proc list */
6011 if (!list_empty(&lkb->lkb_ownqueue))
6012 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
6013 spin_unlock(&ua->proc->locks_spin);
6015 trace_dlm_unlock_end(ls, lkb, flags, error);
6018 dlm_unlock_recovery(ls);
6023 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
6024 uint32_t flags, uint32_t lkid)
6026 struct dlm_lkb *lkb;
6027 struct dlm_args args;
6028 struct dlm_user_args *ua;
6031 dlm_lock_recovery(ls);
6033 error = find_lkb(ls, lkid, &lkb);
6037 trace_dlm_unlock_start(ls, lkb, flags);
6040 if (ua_tmp->castparam)
6041 ua->castparam = ua_tmp->castparam;
6042 ua->user_lksb = ua_tmp->user_lksb;
6044 error = set_unlock_args(flags, ua, &args);
6048 error = cancel_lock(ls, lkb, &args);
6050 if (error == -DLM_ECANCEL)
6052 /* from validate_unlock_args() */
6053 if (error == -EBUSY)
6056 trace_dlm_unlock_end(ls, lkb, flags, error);
6059 dlm_unlock_recovery(ls);
6064 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6066 struct dlm_lkb *lkb;
6067 struct dlm_args args;
6068 struct dlm_user_args *ua;
6072 dlm_lock_recovery(ls);
6074 error = find_lkb(ls, lkid, &lkb);
6078 trace_dlm_unlock_start(ls, lkb, flags);
6082 error = set_unlock_args(flags, ua, &args);
6086 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6088 r = lkb->lkb_resource;
6092 error = validate_unlock_args(lkb, &args);
6095 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6097 error = _cancel_lock(r, lkb);
6102 if (error == -DLM_ECANCEL)
6104 /* from validate_unlock_args() */
6105 if (error == -EBUSY)
6108 trace_dlm_unlock_end(ls, lkb, flags, error);
6111 dlm_unlock_recovery(ls);
6115 /* lkb's that are removed from the waiters list by revert are just left on the
6116 orphans list with the granted orphan locks, to be freed by purge */
6118 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6120 struct dlm_args args;
6123 hold_lkb(lkb); /* reference for the ls_orphans list */
6124 mutex_lock(&ls->ls_orphans_mutex);
6125 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6126 mutex_unlock(&ls->ls_orphans_mutex);
6128 set_unlock_args(0, lkb->lkb_ua, &args);
6130 error = cancel_lock(ls, lkb, &args);
6131 if (error == -DLM_ECANCEL)
6136 /* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6137 granted. Regardless of what rsb queue the lock is on, it's removed and
6138 freed. The IVVALBLK flag causes the lvb on the resource to be invalidated
6139 if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6141 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6143 struct dlm_args args;
6146 set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6147 lkb->lkb_ua, &args);
6149 error = unlock_lock(ls, lkb, &args);
6150 if (error == -DLM_EUNLOCK)
6155 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6156 (which does lock_rsb) due to deadlock with receiving a message that does
6157 lock_rsb followed by dlm_user_add_cb() */
6159 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6160 struct dlm_user_proc *proc)
6162 struct dlm_lkb *lkb = NULL;
6164 spin_lock(&ls->ls_clear_proc_locks);
6165 if (list_empty(&proc->locks))
6168 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6169 list_del_init(&lkb->lkb_ownqueue);
6171 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6172 lkb->lkb_flags |= DLM_IFL_ORPHAN;
6174 lkb->lkb_flags |= DLM_IFL_DEAD;
6176 spin_unlock(&ls->ls_clear_proc_locks);
6180 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6181 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6182 which we clear here. */
6184 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
6185 list, and no more device_writes should add lkb's to proc->locks list; so we
6186 shouldn't need to take asts_spin or locks_spin here. this assumes that
6187 device reads/writes/closes are serialized -- FIXME: we may need to serialize
6190 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6192 struct dlm_lkb *lkb, *safe;
6194 dlm_lock_recovery(ls);
6197 lkb = del_proc_lock(ls, proc);
6201 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6202 orphan_proc_lock(ls, lkb);
6204 unlock_proc_lock(ls, lkb);
6206 /* this removes the reference for the proc->locks list
6207 added by dlm_user_request, it may result in the lkb
6213 spin_lock(&ls->ls_clear_proc_locks);
6215 /* in-progress unlocks */
6216 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6217 list_del_init(&lkb->lkb_ownqueue);
6218 lkb->lkb_flags |= DLM_IFL_DEAD;
6222 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6223 memset(&lkb->lkb_callbacks, 0,
6224 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6225 list_del_init(&lkb->lkb_cb_list);
6229 spin_unlock(&ls->ls_clear_proc_locks);
6230 dlm_unlock_recovery(ls);
6233 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6235 struct dlm_lkb *lkb, *safe;
6239 spin_lock(&proc->locks_spin);
6240 if (!list_empty(&proc->locks)) {
6241 lkb = list_entry(proc->locks.next, struct dlm_lkb,
6243 list_del_init(&lkb->lkb_ownqueue);
6245 spin_unlock(&proc->locks_spin);
6250 lkb->lkb_flags |= DLM_IFL_DEAD;
6251 unlock_proc_lock(ls, lkb);
6252 dlm_put_lkb(lkb); /* ref from proc->locks list */
6255 spin_lock(&proc->locks_spin);
6256 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6257 list_del_init(&lkb->lkb_ownqueue);
6258 lkb->lkb_flags |= DLM_IFL_DEAD;
6261 spin_unlock(&proc->locks_spin);
6263 spin_lock(&proc->asts_spin);
6264 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6265 memset(&lkb->lkb_callbacks, 0,
6266 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
6267 list_del_init(&lkb->lkb_cb_list);
6270 spin_unlock(&proc->asts_spin);
6273 /* pid of 0 means purge all orphans */
6275 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6277 struct dlm_lkb *lkb, *safe;
6279 mutex_lock(&ls->ls_orphans_mutex);
6280 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6281 if (pid && lkb->lkb_ownpid != pid)
6283 unlock_proc_lock(ls, lkb);
6284 list_del_init(&lkb->lkb_ownqueue);
6287 mutex_unlock(&ls->ls_orphans_mutex);
6290 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6292 struct dlm_message *ms;
6293 struct dlm_mhandle *mh;
6296 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6297 DLM_MSG_PURGE, &ms, &mh);
6300 ms->m_nodeid = cpu_to_le32(nodeid);
6301 ms->m_pid = cpu_to_le32(pid);
6303 return send_message(mh, ms);
6306 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6307 int nodeid, int pid)
6311 if (nodeid && (nodeid != dlm_our_nodeid())) {
6312 error = send_purge(ls, nodeid, pid);
6314 dlm_lock_recovery(ls);
6315 if (pid == current->pid)
6316 purge_proc_locks(ls, proc);
6318 do_purge(ls, nodeid, pid);
6319 dlm_unlock_recovery(ls);
6324 /* debug functionality */
6325 int dlm_debug_add_lkb(struct dlm_ls *ls, uint32_t lkb_id, char *name, int len,
6326 int lkb_nodeid, unsigned int lkb_flags, int lkb_status)
6328 struct dlm_lksb *lksb;
6329 struct dlm_lkb *lkb;
6333 /* we currently can't set a valid user lock */
6334 if (lkb_flags & DLM_IFL_USER)
6337 lksb = kzalloc(sizeof(*lksb), GFP_NOFS);
6341 error = _create_lkb(ls, &lkb, lkb_id, lkb_id + 1);
6347 lkb->lkb_flags = lkb_flags;
6348 lkb->lkb_nodeid = lkb_nodeid;
6349 lkb->lkb_lksb = lksb;
6350 /* user specific pointer, just don't have it NULL for kernel locks */
6351 if (~lkb_flags & DLM_IFL_USER)
6352 lkb->lkb_astparam = (void *)0xDEADBEEF;
6354 error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
6363 add_lkb(r, lkb, lkb_status);
6370 int dlm_debug_add_lkb_to_waiters(struct dlm_ls *ls, uint32_t lkb_id,
6371 int mstype, int to_nodeid)
6373 struct dlm_lkb *lkb;
6376 error = find_lkb(ls, lkb_id, &lkb);
6380 error = add_to_waiters(lkb, mstype, to_nodeid);