1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved.
6 ** This copyrighted material is made available to anyone wishing to use,
7 ** modify, copy, or redistribute it subject to the terms and conditions
8 ** of the GNU General Public License v.2.
10 *******************************************************************************
11 ******************************************************************************/
13 /* Central locking logic has four stages:
33 Stage 1 (lock, unlock) is mainly about checking input args and
34 splitting into one of the four main operations:
36 dlm_lock = request_lock
37 dlm_lock+CONVERT = convert_lock
38 dlm_unlock = unlock_lock
39 dlm_unlock+CANCEL = cancel_lock
41 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
42 provided to the next stage.
44 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
45 When remote, it calls send_xxxx(), when local it calls do_xxxx().
47 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
48 given rsb and lkb and queues callbacks.
50 For remote operations, send_xxxx() results in the corresponding do_xxxx()
51 function being executed on the remote node. The connecting send/receive
52 calls on local (L) and remote (R) nodes:
54 L: send_xxxx() -> R: receive_xxxx()
56 L: receive_xxxx_reply() <- R: send_xxxx_reply()
58 #include <linux/types.h>
59 #include <linux/rbtree.h>
60 #include <linux/slab.h>
61 #include "dlm_internal.h"
62 #include <linux/dlm_device.h>
65 #include "requestqueue.h"
69 #include "lockspace.h"
74 #include "lvb_table.h"
78 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
79 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
80 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
81 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
82 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
83 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
84 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
85 static int send_remove(struct dlm_rsb *r);
86 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
87 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
88 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
89 struct dlm_message *ms);
90 static int receive_extralen(struct dlm_message *ms);
91 static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
92 static void del_timeout(struct dlm_lkb *lkb);
95 * Lock compatibilty matrix - thanks Steve
96 * UN = Unlocked state. Not really a state, used as a flag
97 * PD = Padding. Used to make the matrix a nice power of two in size
98 * Other states are the same as the VMS DLM.
99 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
102 static const int __dlm_compat_matrix[8][8] = {
103 /* UN NL CR CW PR PW EX PD */
104 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
105 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
106 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
107 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
108 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
109 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
110 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
111 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
115 * This defines the direction of transfer of LVB data.
116 * Granted mode is the row; requested mode is the column.
117 * Usage: matrix[grmode+1][rqmode+1]
118 * 1 = LVB is returned to the caller
119 * 0 = LVB is written to the resource
120 * -1 = nothing happens to the LVB
123 const int dlm_lvb_operations[8][8] = {
124 /* UN NL CR CW PR PW EX PD*/
125 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
126 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
127 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
128 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
129 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
130 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
131 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
132 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
135 #define modes_compat(gr, rq) \
136 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
138 int dlm_modes_compat(int mode1, int mode2)
140 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
144 * Compatibility matrix for conversions with QUECVT set.
145 * Granted mode is the row; requested mode is the column.
146 * Usage: matrix[grmode+1][rqmode+1]
149 static const int __quecvt_compat_matrix[8][8] = {
150 /* UN NL CR CW PR PW EX PD */
151 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
152 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
153 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
154 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
155 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
156 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
157 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
158 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
161 void dlm_print_lkb(struct dlm_lkb *lkb)
163 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
164 " status %d rqmode %d grmode %d wait_type %d\n",
165 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
166 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
167 lkb->lkb_grmode, lkb->lkb_wait_type);
170 static void dlm_print_rsb(struct dlm_rsb *r)
172 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
173 r->res_nodeid, r->res_flags, r->res_first_lkid,
174 r->res_recover_locks_count, r->res_name);
177 void dlm_dump_rsb(struct dlm_rsb *r)
183 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
184 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
185 printk(KERN_ERR "rsb lookup list\n");
186 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
188 printk(KERN_ERR "rsb grant queue:\n");
189 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
191 printk(KERN_ERR "rsb convert queue:\n");
192 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
194 printk(KERN_ERR "rsb wait queue:\n");
195 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
199 /* Threads cannot use the lockspace while it's being recovered */
201 static inline void dlm_lock_recovery(struct dlm_ls *ls)
203 down_read(&ls->ls_in_recovery);
206 void dlm_unlock_recovery(struct dlm_ls *ls)
208 up_read(&ls->ls_in_recovery);
211 int dlm_lock_recovery_try(struct dlm_ls *ls)
213 return down_read_trylock(&ls->ls_in_recovery);
216 static inline int can_be_queued(struct dlm_lkb *lkb)
218 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
221 static inline int force_blocking_asts(struct dlm_lkb *lkb)
223 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
226 static inline int is_demoted(struct dlm_lkb *lkb)
228 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
231 static inline int is_altmode(struct dlm_lkb *lkb)
233 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
236 static inline int is_granted(struct dlm_lkb *lkb)
238 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
241 static inline int is_remote(struct dlm_rsb *r)
243 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
244 return !!r->res_nodeid;
247 static inline int is_process_copy(struct dlm_lkb *lkb)
249 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
252 static inline int is_master_copy(struct dlm_lkb *lkb)
254 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
255 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
256 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
259 static inline int middle_conversion(struct dlm_lkb *lkb)
261 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
262 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
267 static inline int down_conversion(struct dlm_lkb *lkb)
269 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
272 static inline int is_overlap_unlock(struct dlm_lkb *lkb)
274 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
277 static inline int is_overlap_cancel(struct dlm_lkb *lkb)
279 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
282 static inline int is_overlap(struct dlm_lkb *lkb)
284 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
285 DLM_IFL_OVERLAP_CANCEL));
288 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
290 if (is_master_copy(lkb))
295 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
297 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
298 timeout caused the cancel then return -ETIMEDOUT */
299 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
300 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
304 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
305 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
309 dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
312 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
315 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
318 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
320 if (is_master_copy(lkb)) {
321 send_bast(r, lkb, rqmode);
323 dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
328 * Basic operations on rsb's and lkb's
331 static int pre_rsb_struct(struct dlm_ls *ls)
333 struct dlm_rsb *r1, *r2;
336 spin_lock(&ls->ls_new_rsb_spin);
337 if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
338 spin_unlock(&ls->ls_new_rsb_spin);
341 spin_unlock(&ls->ls_new_rsb_spin);
343 r1 = dlm_allocate_rsb(ls);
344 r2 = dlm_allocate_rsb(ls);
346 spin_lock(&ls->ls_new_rsb_spin);
348 list_add(&r1->res_hashchain, &ls->ls_new_rsb);
349 ls->ls_new_rsb_count++;
352 list_add(&r2->res_hashchain, &ls->ls_new_rsb);
353 ls->ls_new_rsb_count++;
355 count = ls->ls_new_rsb_count;
356 spin_unlock(&ls->ls_new_rsb_spin);
363 /* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
364 unlock any spinlocks, go back and call pre_rsb_struct again.
365 Otherwise, take an rsb off the list and return it. */
367 static int get_rsb_struct(struct dlm_ls *ls, char *name, int len,
368 struct dlm_rsb **r_ret)
373 spin_lock(&ls->ls_new_rsb_spin);
374 if (list_empty(&ls->ls_new_rsb)) {
375 count = ls->ls_new_rsb_count;
376 spin_unlock(&ls->ls_new_rsb_spin);
377 log_debug(ls, "find_rsb retry %d %d %s",
378 count, dlm_config.ci_new_rsb_count, name);
382 r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
383 list_del(&r->res_hashchain);
384 /* Convert the empty list_head to a NULL rb_node for tree usage: */
385 memset(&r->res_hashnode, 0, sizeof(struct rb_node));
386 ls->ls_new_rsb_count--;
387 spin_unlock(&ls->ls_new_rsb_spin);
391 memcpy(r->res_name, name, len);
392 mutex_init(&r->res_mutex);
394 INIT_LIST_HEAD(&r->res_lookup);
395 INIT_LIST_HEAD(&r->res_grantqueue);
396 INIT_LIST_HEAD(&r->res_convertqueue);
397 INIT_LIST_HEAD(&r->res_waitqueue);
398 INIT_LIST_HEAD(&r->res_root_list);
399 INIT_LIST_HEAD(&r->res_recover_list);
405 static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
407 char maxname[DLM_RESNAME_MAXLEN];
409 memset(maxname, 0, DLM_RESNAME_MAXLEN);
410 memcpy(maxname, name, nlen);
411 return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
414 int dlm_search_rsb_tree(struct rb_root *tree, char *name, int len,
415 unsigned int flags, struct dlm_rsb **r_ret)
417 struct rb_node *node = tree->rb_node;
423 r = rb_entry(node, struct dlm_rsb, res_hashnode);
424 rc = rsb_cmp(r, name, len);
426 node = node->rb_left;
428 node = node->rb_right;
436 if (r->res_nodeid && (flags & R_MASTER))
442 static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
444 struct rb_node **newn = &tree->rb_node;
445 struct rb_node *parent = NULL;
449 struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
453 rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
455 newn = &parent->rb_left;
457 newn = &parent->rb_right;
459 log_print("rsb_insert match");
466 rb_link_node(&rsb->res_hashnode, parent, newn);
467 rb_insert_color(&rsb->res_hashnode, tree);
471 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
472 unsigned int flags, struct dlm_rsb **r_ret)
477 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, flags, &r);
479 kref_get(&r->res_ref);
482 error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
486 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
487 error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
491 if (dlm_no_directory(ls))
494 if (r->res_nodeid == -1) {
495 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
496 r->res_first_lkid = 0;
497 } else if (r->res_nodeid > 0) {
498 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
499 r->res_first_lkid = 0;
501 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
502 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
510 * Find rsb in rsbtbl and potentially create/add one
512 * Delaying the release of rsb's has a similar benefit to applications keeping
513 * NL locks on an rsb, but without the guarantee that the cached master value
514 * will still be valid when the rsb is reused. Apps aren't always smart enough
515 * to keep NL locks on an rsb that they may lock again shortly; this can lead
516 * to excessive master lookups and removals if we don't delay the release.
518 * Searching for an rsb means looking through both the normal list and toss
519 * list. When found on the toss list the rsb is moved to the normal list with
520 * ref count of 1; when found on normal list the ref count is incremented.
523 static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
524 unsigned int flags, struct dlm_rsb **r_ret)
526 struct dlm_rsb *r = NULL;
527 uint32_t hash, bucket;
530 if (namelen > DLM_RESNAME_MAXLEN) {
535 if (dlm_no_directory(ls))
538 hash = jhash(name, namelen, 0);
539 bucket = hash & (ls->ls_rsbtbl_size - 1);
542 if (flags & R_CREATE) {
543 error = pre_rsb_struct(ls);
548 spin_lock(&ls->ls_rsbtbl[bucket].lock);
550 error = _search_rsb(ls, name, namelen, bucket, flags, &r);
554 if (error == -EBADR && !(flags & R_CREATE))
557 /* the rsb was found but wasn't a master copy */
558 if (error == -ENOTBLK)
561 error = get_rsb_struct(ls, name, namelen, &r);
562 if (error == -EAGAIN) {
563 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
570 r->res_bucket = bucket;
572 kref_init(&r->res_ref);
574 /* With no directory, the master can be set immediately */
575 if (dlm_no_directory(ls)) {
576 int nodeid = dlm_dir_nodeid(r);
577 if (nodeid == dlm_our_nodeid())
579 r->res_nodeid = nodeid;
581 error = rsb_insert(r, &ls->ls_rsbtbl[bucket].keep);
583 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
589 /* This is only called to add a reference when the code already holds
590 a valid reference to the rsb, so there's no need for locking. */
592 static inline void hold_rsb(struct dlm_rsb *r)
594 kref_get(&r->res_ref);
597 void dlm_hold_rsb(struct dlm_rsb *r)
602 static void toss_rsb(struct kref *kref)
604 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
605 struct dlm_ls *ls = r->res_ls;
607 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
608 kref_init(&r->res_ref);
609 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
610 rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
611 r->res_toss_time = jiffies;
613 dlm_free_lvb(r->res_lvbptr);
614 r->res_lvbptr = NULL;
618 /* When all references to the rsb are gone it's transferred to
619 the tossed list for later disposal. */
621 static void put_rsb(struct dlm_rsb *r)
623 struct dlm_ls *ls = r->res_ls;
624 uint32_t bucket = r->res_bucket;
626 spin_lock(&ls->ls_rsbtbl[bucket].lock);
627 kref_put(&r->res_ref, toss_rsb);
628 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
631 void dlm_put_rsb(struct dlm_rsb *r)
636 /* See comment for unhold_lkb */
638 static void unhold_rsb(struct dlm_rsb *r)
641 rv = kref_put(&r->res_ref, toss_rsb);
642 DLM_ASSERT(!rv, dlm_dump_rsb(r););
645 static void kill_rsb(struct kref *kref)
647 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
649 /* All work is done after the return from kref_put() so we
650 can release the write_lock before the remove and free. */
652 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
653 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
654 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
655 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
656 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
657 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
660 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
661 The rsb must exist as long as any lkb's for it do. */
663 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
666 lkb->lkb_resource = r;
669 static void detach_lkb(struct dlm_lkb *lkb)
671 if (lkb->lkb_resource) {
672 put_rsb(lkb->lkb_resource);
673 lkb->lkb_resource = NULL;
677 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
682 lkb = dlm_allocate_lkb(ls);
686 lkb->lkb_nodeid = -1;
687 lkb->lkb_grmode = DLM_LOCK_IV;
688 kref_init(&lkb->lkb_ref);
689 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
690 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
691 INIT_LIST_HEAD(&lkb->lkb_time_list);
692 INIT_LIST_HEAD(&lkb->lkb_cb_list);
693 mutex_init(&lkb->lkb_cb_mutex);
694 INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
697 rv = idr_pre_get(&ls->ls_lkbidr, GFP_NOFS);
701 spin_lock(&ls->ls_lkbidr_spin);
702 rv = idr_get_new_above(&ls->ls_lkbidr, lkb, 1, &id);
705 spin_unlock(&ls->ls_lkbidr_spin);
711 log_error(ls, "create_lkb idr error %d", rv);
719 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
723 spin_lock(&ls->ls_lkbidr_spin);
724 lkb = idr_find(&ls->ls_lkbidr, lkid);
726 kref_get(&lkb->lkb_ref);
727 spin_unlock(&ls->ls_lkbidr_spin);
730 return lkb ? 0 : -ENOENT;
733 static void kill_lkb(struct kref *kref)
735 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
737 /* All work is done after the return from kref_put() so we
738 can release the write_lock before the detach_lkb */
740 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
743 /* __put_lkb() is used when an lkb may not have an rsb attached to
744 it so we need to provide the lockspace explicitly */
746 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
748 uint32_t lkid = lkb->lkb_id;
750 spin_lock(&ls->ls_lkbidr_spin);
751 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
752 idr_remove(&ls->ls_lkbidr, lkid);
753 spin_unlock(&ls->ls_lkbidr_spin);
757 /* for local/process lkbs, lvbptr points to caller's lksb */
758 if (lkb->lkb_lvbptr && is_master_copy(lkb))
759 dlm_free_lvb(lkb->lkb_lvbptr);
763 spin_unlock(&ls->ls_lkbidr_spin);
768 int dlm_put_lkb(struct dlm_lkb *lkb)
772 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
773 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
775 ls = lkb->lkb_resource->res_ls;
776 return __put_lkb(ls, lkb);
779 /* This is only called to add a reference when the code already holds
780 a valid reference to the lkb, so there's no need for locking. */
782 static inline void hold_lkb(struct dlm_lkb *lkb)
784 kref_get(&lkb->lkb_ref);
787 /* This is called when we need to remove a reference and are certain
788 it's not the last ref. e.g. del_lkb is always called between a
789 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
790 put_lkb would work fine, but would involve unnecessary locking */
792 static inline void unhold_lkb(struct dlm_lkb *lkb)
795 rv = kref_put(&lkb->lkb_ref, kill_lkb);
796 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
799 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
802 struct dlm_lkb *lkb = NULL;
804 list_for_each_entry(lkb, head, lkb_statequeue)
805 if (lkb->lkb_rqmode < mode)
808 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
811 /* add/remove lkb to rsb's grant/convert/wait queue */
813 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
815 kref_get(&lkb->lkb_ref);
817 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
819 lkb->lkb_timestamp = ktime_get();
821 lkb->lkb_status = status;
824 case DLM_LKSTS_WAITING:
825 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
826 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
828 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
830 case DLM_LKSTS_GRANTED:
831 /* convention says granted locks kept in order of grmode */
832 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
835 case DLM_LKSTS_CONVERT:
836 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
837 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
839 list_add_tail(&lkb->lkb_statequeue,
840 &r->res_convertqueue);
843 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
847 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
850 list_del(&lkb->lkb_statequeue);
854 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
858 add_lkb(r, lkb, sts);
862 static int msg_reply_type(int mstype)
865 case DLM_MSG_REQUEST:
866 return DLM_MSG_REQUEST_REPLY;
867 case DLM_MSG_CONVERT:
868 return DLM_MSG_CONVERT_REPLY;
870 return DLM_MSG_UNLOCK_REPLY;
872 return DLM_MSG_CANCEL_REPLY;
874 return DLM_MSG_LOOKUP_REPLY;
879 static int nodeid_warned(int nodeid, int num_nodes, int *warned)
883 for (i = 0; i < num_nodes; i++) {
888 if (warned[i] == nodeid)
894 void dlm_scan_waiters(struct dlm_ls *ls)
897 ktime_t zero = ktime_set(0, 0);
900 u32 debug_scanned = 0;
901 u32 debug_expired = 0;
905 if (!dlm_config.ci_waitwarn_us)
908 mutex_lock(&ls->ls_waiters_mutex);
910 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
911 if (ktime_equal(lkb->lkb_wait_time, zero))
916 us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time));
918 if (us < dlm_config.ci_waitwarn_us)
921 lkb->lkb_wait_time = zero;
924 if (us > debug_maxus)
928 num_nodes = ls->ls_num_nodes;
929 warned = kzalloc(num_nodes * sizeof(int), GFP_KERNEL);
933 if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned))
936 log_error(ls, "waitwarn %x %lld %d us check connection to "
937 "node %d", lkb->lkb_id, (long long)us,
938 dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid);
940 mutex_unlock(&ls->ls_waiters_mutex);
944 log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us",
945 debug_scanned, debug_expired,
946 dlm_config.ci_waitwarn_us, (long long)debug_maxus);
949 /* add/remove lkb from global waiters list of lkb's waiting for
950 a reply from a remote node */
952 static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
954 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
957 mutex_lock(&ls->ls_waiters_mutex);
959 if (is_overlap_unlock(lkb) ||
960 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
965 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
968 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
971 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
977 lkb->lkb_wait_count++;
980 log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
981 lkb->lkb_id, lkb->lkb_wait_type, mstype,
982 lkb->lkb_wait_count, lkb->lkb_flags);
986 DLM_ASSERT(!lkb->lkb_wait_count,
988 printk("wait_count %d\n", lkb->lkb_wait_count););
990 lkb->lkb_wait_count++;
991 lkb->lkb_wait_type = mstype;
992 lkb->lkb_wait_time = ktime_get();
993 lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
995 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
998 log_error(ls, "addwait error %x %d flags %x %d %d %s",
999 lkb->lkb_id, error, lkb->lkb_flags, mstype,
1000 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1001 mutex_unlock(&ls->ls_waiters_mutex);
1005 /* We clear the RESEND flag because we might be taking an lkb off the waiters
1006 list as part of process_requestqueue (e.g. a lookup that has an optimized
1007 request reply on the requestqueue) between dlm_recover_waiters_pre() which
1008 set RESEND and dlm_recover_waiters_post() */
1010 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1011 struct dlm_message *ms)
1013 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1014 int overlap_done = 0;
1016 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1017 log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1018 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1023 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1024 log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1025 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1030 /* Cancel state was preemptively cleared by a successful convert,
1031 see next comment, nothing to do. */
1033 if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1034 (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1035 log_debug(ls, "remwait %x cancel_reply wait_type %d",
1036 lkb->lkb_id, lkb->lkb_wait_type);
1040 /* Remove for the convert reply, and premptively remove for the
1041 cancel reply. A convert has been granted while there's still
1042 an outstanding cancel on it (the cancel is moot and the result
1043 in the cancel reply should be 0). We preempt the cancel reply
1044 because the app gets the convert result and then can follow up
1045 with another op, like convert. This subsequent op would see the
1046 lingering state of the cancel and fail with -EBUSY. */
1048 if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1049 (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1050 is_overlap_cancel(lkb) && ms && !ms->m_result) {
1051 log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1053 lkb->lkb_wait_type = 0;
1054 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1055 lkb->lkb_wait_count--;
1059 /* N.B. type of reply may not always correspond to type of original
1060 msg due to lookup->request optimization, verify others? */
1062 if (lkb->lkb_wait_type) {
1063 lkb->lkb_wait_type = 0;
1067 log_error(ls, "remwait error %x reply %d flags %x no wait_type",
1068 lkb->lkb_id, mstype, lkb->lkb_flags);
1072 /* the force-unlock/cancel has completed and we haven't recvd a reply
1073 to the op that was in progress prior to the unlock/cancel; we
1074 give up on any reply to the earlier op. FIXME: not sure when/how
1075 this would happen */
1077 if (overlap_done && lkb->lkb_wait_type) {
1078 log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1079 lkb->lkb_id, mstype, lkb->lkb_wait_type);
1080 lkb->lkb_wait_count--;
1081 lkb->lkb_wait_type = 0;
1084 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1086 lkb->lkb_flags &= ~DLM_IFL_RESEND;
1087 lkb->lkb_wait_count--;
1088 if (!lkb->lkb_wait_count)
1089 list_del_init(&lkb->lkb_wait_reply);
1094 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1096 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1099 mutex_lock(&ls->ls_waiters_mutex);
1100 error = _remove_from_waiters(lkb, mstype, NULL);
1101 mutex_unlock(&ls->ls_waiters_mutex);
1105 /* Handles situations where we might be processing a "fake" or "stub" reply in
1106 which we can't try to take waiters_mutex again. */
1108 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1110 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1113 if (ms->m_flags != DLM_IFL_STUB_MS)
1114 mutex_lock(&ls->ls_waiters_mutex);
1115 error = _remove_from_waiters(lkb, ms->m_type, ms);
1116 if (ms->m_flags != DLM_IFL_STUB_MS)
1117 mutex_unlock(&ls->ls_waiters_mutex);
1121 static void dir_remove(struct dlm_rsb *r)
1125 if (dlm_no_directory(r->res_ls))
1128 to_nodeid = dlm_dir_nodeid(r);
1129 if (to_nodeid != dlm_our_nodeid())
1132 dlm_dir_remove_entry(r->res_ls, to_nodeid,
1133 r->res_name, r->res_length);
1136 /* FIXME: make this more efficient */
1138 static int shrink_bucket(struct dlm_ls *ls, int b)
1142 int count = 0, found;
1146 spin_lock(&ls->ls_rsbtbl[b].lock);
1147 for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = rb_next(n)) {
1148 r = rb_entry(n, struct dlm_rsb, res_hashnode);
1149 if (!time_after_eq(jiffies, r->res_toss_time +
1150 dlm_config.ci_toss_secs * HZ))
1157 spin_unlock(&ls->ls_rsbtbl[b].lock);
1161 if (kref_put(&r->res_ref, kill_rsb)) {
1162 rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1163 spin_unlock(&ls->ls_rsbtbl[b].lock);
1170 spin_unlock(&ls->ls_rsbtbl[b].lock);
1171 log_error(ls, "tossed rsb in use %s", r->res_name);
1178 void dlm_scan_rsbs(struct dlm_ls *ls)
1182 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1183 shrink_bucket(ls, i);
1184 if (dlm_locking_stopped(ls))
1190 static void add_timeout(struct dlm_lkb *lkb)
1192 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1194 if (is_master_copy(lkb))
1197 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1198 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1199 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1202 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1207 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1208 mutex_lock(&ls->ls_timeout_mutex);
1210 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1211 mutex_unlock(&ls->ls_timeout_mutex);
1214 static void del_timeout(struct dlm_lkb *lkb)
1216 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1218 mutex_lock(&ls->ls_timeout_mutex);
1219 if (!list_empty(&lkb->lkb_time_list)) {
1220 list_del_init(&lkb->lkb_time_list);
1223 mutex_unlock(&ls->ls_timeout_mutex);
1226 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1227 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1228 and then lock rsb because of lock ordering in add_timeout. We may need
1229 to specify some special timeout-related bits in the lkb that are just to
1230 be accessed under the timeout_mutex. */
1232 void dlm_scan_timeout(struct dlm_ls *ls)
1235 struct dlm_lkb *lkb;
1236 int do_cancel, do_warn;
1240 if (dlm_locking_stopped(ls))
1245 mutex_lock(&ls->ls_timeout_mutex);
1246 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1248 wait_us = ktime_to_us(ktime_sub(ktime_get(),
1249 lkb->lkb_timestamp));
1251 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1252 wait_us >= (lkb->lkb_timeout_cs * 10000))
1255 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1256 wait_us >= dlm_config.ci_timewarn_cs * 10000)
1259 if (!do_cancel && !do_warn)
1264 mutex_unlock(&ls->ls_timeout_mutex);
1266 if (!do_cancel && !do_warn)
1269 r = lkb->lkb_resource;
1274 /* clear flag so we only warn once */
1275 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1276 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1278 dlm_timeout_warn(lkb);
1282 log_debug(ls, "timeout cancel %x node %d %s",
1283 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1284 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1285 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1287 _cancel_lock(r, lkb);
1296 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1297 dlm_recoverd before checking/setting ls_recover_begin. */
1299 void dlm_adjust_timeouts(struct dlm_ls *ls)
1301 struct dlm_lkb *lkb;
1302 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1304 ls->ls_recover_begin = 0;
1305 mutex_lock(&ls->ls_timeout_mutex);
1306 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1307 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1308 mutex_unlock(&ls->ls_timeout_mutex);
1310 if (!dlm_config.ci_waitwarn_us)
1313 mutex_lock(&ls->ls_waiters_mutex);
1314 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
1315 if (ktime_to_us(lkb->lkb_wait_time))
1316 lkb->lkb_wait_time = ktime_get();
1318 mutex_unlock(&ls->ls_waiters_mutex);
1321 /* lkb is master or local copy */
1323 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1325 int b, len = r->res_ls->ls_lvblen;
1327 /* b=1 lvb returned to caller
1328 b=0 lvb written to rsb or invalidated
1331 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1334 if (!lkb->lkb_lvbptr)
1337 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1343 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1344 lkb->lkb_lvbseq = r->res_lvbseq;
1346 } else if (b == 0) {
1347 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1348 rsb_set_flag(r, RSB_VALNOTVALID);
1352 if (!lkb->lkb_lvbptr)
1355 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1359 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1364 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1366 lkb->lkb_lvbseq = r->res_lvbseq;
1367 rsb_clear_flag(r, RSB_VALNOTVALID);
1370 if (rsb_flag(r, RSB_VALNOTVALID))
1371 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1374 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1376 if (lkb->lkb_grmode < DLM_LOCK_PW)
1379 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1380 rsb_set_flag(r, RSB_VALNOTVALID);
1384 if (!lkb->lkb_lvbptr)
1387 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1391 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1396 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1398 rsb_clear_flag(r, RSB_VALNOTVALID);
1401 /* lkb is process copy (pc) */
1403 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1404 struct dlm_message *ms)
1408 if (!lkb->lkb_lvbptr)
1411 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1414 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1416 int len = receive_extralen(ms);
1417 if (len > DLM_RESNAME_MAXLEN)
1418 len = DLM_RESNAME_MAXLEN;
1419 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1420 lkb->lkb_lvbseq = ms->m_lvbseq;
1424 /* Manipulate lkb's on rsb's convert/granted/waiting queues
1425 remove_lock -- used for unlock, removes lkb from granted
1426 revert_lock -- used for cancel, moves lkb from convert to granted
1427 grant_lock -- used for request and convert, adds lkb to granted or
1428 moves lkb from convert or waiting to granted
1430 Each of these is used for master or local copy lkb's. There is
1431 also a _pc() variation used to make the corresponding change on
1432 a process copy (pc) lkb. */
1434 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1437 lkb->lkb_grmode = DLM_LOCK_IV;
1438 /* this unhold undoes the original ref from create_lkb()
1439 so this leads to the lkb being freed */
1443 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1445 set_lvb_unlock(r, lkb);
1446 _remove_lock(r, lkb);
1449 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1451 _remove_lock(r, lkb);
1454 /* returns: 0 did nothing
1455 1 moved lock to granted
1458 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1462 lkb->lkb_rqmode = DLM_LOCK_IV;
1464 switch (lkb->lkb_status) {
1465 case DLM_LKSTS_GRANTED:
1467 case DLM_LKSTS_CONVERT:
1468 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1471 case DLM_LKSTS_WAITING:
1473 lkb->lkb_grmode = DLM_LOCK_IV;
1474 /* this unhold undoes the original ref from create_lkb()
1475 so this leads to the lkb being freed */
1480 log_print("invalid status for revert %d", lkb->lkb_status);
1485 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1487 return revert_lock(r, lkb);
1490 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1492 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1493 lkb->lkb_grmode = lkb->lkb_rqmode;
1494 if (lkb->lkb_status)
1495 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1497 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1500 lkb->lkb_rqmode = DLM_LOCK_IV;
1503 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1505 set_lvb_lock(r, lkb);
1506 _grant_lock(r, lkb);
1507 lkb->lkb_highbast = 0;
1510 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1511 struct dlm_message *ms)
1513 set_lvb_lock_pc(r, lkb, ms);
1514 _grant_lock(r, lkb);
1517 /* called by grant_pending_locks() which means an async grant message must
1518 be sent to the requesting node in addition to granting the lock if the
1519 lkb belongs to a remote node. */
1521 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1524 if (is_master_copy(lkb))
1527 queue_cast(r, lkb, 0);
1530 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1531 change the granted/requested modes. We're munging things accordingly in
1533 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1535 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1536 compatible with other granted locks */
1538 static void munge_demoted(struct dlm_lkb *lkb)
1540 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
1541 log_print("munge_demoted %x invalid modes gr %d rq %d",
1542 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
1546 lkb->lkb_grmode = DLM_LOCK_NL;
1549 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
1551 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
1552 ms->m_type != DLM_MSG_GRANT) {
1553 log_print("munge_altmode %x invalid reply type %d",
1554 lkb->lkb_id, ms->m_type);
1558 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
1559 lkb->lkb_rqmode = DLM_LOCK_PR;
1560 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
1561 lkb->lkb_rqmode = DLM_LOCK_CW;
1563 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
1568 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1570 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1572 if (lkb->lkb_id == first->lkb_id)
1578 /* Check if the given lkb conflicts with another lkb on the queue. */
1580 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1582 struct dlm_lkb *this;
1584 list_for_each_entry(this, head, lkb_statequeue) {
1587 if (!modes_compat(this, lkb))
1594 * "A conversion deadlock arises with a pair of lock requests in the converting
1595 * queue for one resource. The granted mode of each lock blocks the requested
1596 * mode of the other lock."
1598 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1599 * convert queue from being granted, then deadlk/demote lkb.
1602 * Granted Queue: empty
1603 * Convert Queue: NL->EX (first lock)
1604 * PR->EX (second lock)
1606 * The first lock can't be granted because of the granted mode of the second
1607 * lock and the second lock can't be granted because it's not first in the
1608 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1609 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1610 * flag set and return DEMOTED in the lksb flags.
1612 * Originally, this function detected conv-deadlk in a more limited scope:
1613 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1614 * - if lkb1 was the first entry in the queue (not just earlier), and was
1615 * blocked by the granted mode of lkb2, and there was nothing on the
1616 * granted queue preventing lkb1 from being granted immediately, i.e.
1617 * lkb2 was the only thing preventing lkb1 from being granted.
1619 * That second condition meant we'd only say there was conv-deadlk if
1620 * resolving it (by demotion) would lead to the first lock on the convert
1621 * queue being granted right away. It allowed conversion deadlocks to exist
1622 * between locks on the convert queue while they couldn't be granted anyway.
1624 * Now, we detect and take action on conversion deadlocks immediately when
1625 * they're created, even if they may not be immediately consequential. If
1626 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1627 * mode that would prevent lkb1's conversion from being granted, we do a
1628 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1629 * I think this means that the lkb_is_ahead condition below should always
1630 * be zero, i.e. there will never be conv-deadlk between two locks that are
1631 * both already on the convert queue.
1634 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
1636 struct dlm_lkb *lkb1;
1637 int lkb_is_ahead = 0;
1639 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
1645 if (!lkb_is_ahead) {
1646 if (!modes_compat(lkb2, lkb1))
1649 if (!modes_compat(lkb2, lkb1) &&
1650 !modes_compat(lkb1, lkb2))
1658 * Return 1 if the lock can be granted, 0 otherwise.
1659 * Also detect and resolve conversion deadlocks.
1661 * lkb is the lock to be granted
1663 * now is 1 if the function is being called in the context of the
1664 * immediate request, it is 0 if called later, after the lock has been
1667 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1670 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1672 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1675 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1676 * a new request for a NL mode lock being blocked.
1678 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1679 * request, then it would be granted. In essence, the use of this flag
1680 * tells the Lock Manager to expedite theis request by not considering
1681 * what may be in the CONVERTING or WAITING queues... As of this
1682 * writing, the EXPEDITE flag can be used only with new requests for NL
1683 * mode locks. This flag is not valid for conversion requests.
1685 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1686 * conversion or used with a non-NL requested mode. We also know an
1687 * EXPEDITE request is always granted immediately, so now must always
1688 * be 1. The full condition to grant an expedite request: (now &&
1689 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1690 * therefore be shortened to just checking the flag.
1693 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1697 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1698 * added to the remaining conditions.
1701 if (queue_conflict(&r->res_grantqueue, lkb))
1705 * 6-3: By default, a conversion request is immediately granted if the
1706 * requested mode is compatible with the modes of all other granted
1710 if (queue_conflict(&r->res_convertqueue, lkb))
1714 * 6-5: But the default algorithm for deciding whether to grant or
1715 * queue conversion requests does not by itself guarantee that such
1716 * requests are serviced on a "first come first serve" basis. This, in
1717 * turn, can lead to a phenomenon known as "indefinate postponement".
1719 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1720 * the system service employed to request a lock conversion. This flag
1721 * forces certain conversion requests to be queued, even if they are
1722 * compatible with the granted modes of other locks on the same
1723 * resource. Thus, the use of this flag results in conversion requests
1724 * being ordered on a "first come first servce" basis.
1726 * DCT: This condition is all about new conversions being able to occur
1727 * "in place" while the lock remains on the granted queue (assuming
1728 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1729 * doesn't _have_ to go onto the convert queue where it's processed in
1730 * order. The "now" variable is necessary to distinguish converts
1731 * being received and processed for the first time now, because once a
1732 * convert is moved to the conversion queue the condition below applies
1733 * requiring fifo granting.
1736 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1740 * Even if the convert is compat with all granted locks,
1741 * QUECVT forces it behind other locks on the convert queue.
1744 if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
1745 if (list_empty(&r->res_convertqueue))
1752 * The NOORDER flag is set to avoid the standard vms rules on grant
1756 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1760 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1761 * granted until all other conversion requests ahead of it are granted
1765 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1769 * 6-4: By default, a new request is immediately granted only if all
1770 * three of the following conditions are satisfied when the request is
1772 * - The queue of ungranted conversion requests for the resource is
1774 * - The queue of ungranted new requests for the resource is empty.
1775 * - The mode of the new request is compatible with the most
1776 * restrictive mode of all granted locks on the resource.
1779 if (now && !conv && list_empty(&r->res_convertqueue) &&
1780 list_empty(&r->res_waitqueue))
1784 * 6-4: Once a lock request is in the queue of ungranted new requests,
1785 * it cannot be granted until the queue of ungranted conversion
1786 * requests is empty, all ungranted new requests ahead of it are
1787 * granted and/or canceled, and it is compatible with the granted mode
1788 * of the most restrictive lock granted on the resource.
1791 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1792 first_in_list(lkb, &r->res_waitqueue))
1798 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
1802 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1803 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
1808 rv = _can_be_granted(r, lkb, now);
1813 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1814 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1815 * cancels one of the locks.
1818 if (is_convert && can_be_queued(lkb) &&
1819 conversion_deadlock_detect(r, lkb)) {
1820 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
1821 lkb->lkb_grmode = DLM_LOCK_NL;
1822 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1823 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1827 log_print("can_be_granted deadlock %x now %d",
1836 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1837 * to grant a request in a mode other than the normal rqmode. It's a
1838 * simple way to provide a big optimization to applications that can
1842 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
1844 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
1848 lkb->lkb_rqmode = alt;
1849 rv = _can_be_granted(r, lkb, now);
1851 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1853 lkb->lkb_rqmode = rqmode;
1859 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1860 for locks pending on the convert list. Once verified (watch for these
1861 log_prints), we should be able to just call _can_be_granted() and not
1862 bother with the demote/deadlk cases here (and there's no easy way to deal
1863 with a deadlk here, we'd have to generate something like grant_lock with
1864 the deadlk error.) */
1866 /* Returns the highest requested mode of all blocked conversions; sets
1867 cw if there's a blocked conversion to DLM_LOCK_CW. */
1869 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
1871 struct dlm_lkb *lkb, *s;
1872 int hi, demoted, quit, grant_restart, demote_restart;
1881 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1882 demoted = is_demoted(lkb);
1885 if (can_be_granted(r, lkb, 0, &deadlk)) {
1886 grant_lock_pending(r, lkb);
1891 if (!demoted && is_demoted(lkb)) {
1892 log_print("WARN: pending demoted %x node %d %s",
1893 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1899 log_print("WARN: pending deadlock %x node %d %s",
1900 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1905 hi = max_t(int, lkb->lkb_rqmode, hi);
1907 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
1913 if (demote_restart && !quit) {
1918 return max_t(int, high, hi);
1921 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
1923 struct dlm_lkb *lkb, *s;
1925 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1926 if (can_be_granted(r, lkb, 0, NULL))
1927 grant_lock_pending(r, lkb);
1929 high = max_t(int, lkb->lkb_rqmode, high);
1930 if (lkb->lkb_rqmode == DLM_LOCK_CW)
1938 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
1939 on either the convert or waiting queue.
1940 high is the largest rqmode of all locks blocked on the convert or
1943 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
1945 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
1946 if (gr->lkb_highbast < DLM_LOCK_EX)
1951 if (gr->lkb_highbast < high &&
1952 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
1957 static void grant_pending_locks(struct dlm_rsb *r)
1959 struct dlm_lkb *lkb, *s;
1960 int high = DLM_LOCK_IV;
1963 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
1965 high = grant_pending_convert(r, high, &cw);
1966 high = grant_pending_wait(r, high, &cw);
1968 if (high == DLM_LOCK_IV)
1972 * If there are locks left on the wait/convert queue then send blocking
1973 * ASTs to granted locks based on the largest requested mode (high)
1977 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1978 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
1979 if (cw && high == DLM_LOCK_PR &&
1980 lkb->lkb_grmode == DLM_LOCK_PR)
1981 queue_bast(r, lkb, DLM_LOCK_CW);
1983 queue_bast(r, lkb, high);
1984 lkb->lkb_highbast = high;
1989 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
1991 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
1992 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
1993 if (gr->lkb_highbast < DLM_LOCK_EX)
1998 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2003 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2004 struct dlm_lkb *lkb)
2008 list_for_each_entry(gr, head, lkb_statequeue) {
2009 /* skip self when sending basts to convertqueue */
2012 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2013 queue_bast(r, gr, lkb->lkb_rqmode);
2014 gr->lkb_highbast = lkb->lkb_rqmode;
2019 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2021 send_bast_queue(r, &r->res_grantqueue, lkb);
2024 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2026 send_bast_queue(r, &r->res_grantqueue, lkb);
2027 send_bast_queue(r, &r->res_convertqueue, lkb);
2030 /* set_master(r, lkb) -- set the master nodeid of a resource
2032 The purpose of this function is to set the nodeid field in the given
2033 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
2034 known, it can just be copied to the lkb and the function will return
2035 0. If the rsb's nodeid is _not_ known, it needs to be looked up
2036 before it can be copied to the lkb.
2038 When the rsb nodeid is being looked up remotely, the initial lkb
2039 causing the lookup is kept on the ls_waiters list waiting for the
2040 lookup reply. Other lkb's waiting for the same rsb lookup are kept
2041 on the rsb's res_lookup list until the master is verified.
2044 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2045 1: the rsb master is not available and the lkb has been placed on
2049 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2051 struct dlm_ls *ls = r->res_ls;
2052 int i, error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
2054 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2055 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2056 r->res_first_lkid = lkb->lkb_id;
2057 lkb->lkb_nodeid = r->res_nodeid;
2061 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2062 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2066 if (r->res_nodeid == 0) {
2067 lkb->lkb_nodeid = 0;
2071 if (r->res_nodeid > 0) {
2072 lkb->lkb_nodeid = r->res_nodeid;
2076 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
2078 dir_nodeid = dlm_dir_nodeid(r);
2080 if (dir_nodeid != our_nodeid) {
2081 r->res_first_lkid = lkb->lkb_id;
2082 send_lookup(r, lkb);
2086 for (i = 0; i < 2; i++) {
2087 /* It's possible for dlm_scand to remove an old rsb for
2088 this same resource from the toss list, us to create
2089 a new one, look up the master locally, and find it
2090 already exists just before dlm_scand does the
2091 dir_remove() on the previous rsb. */
2093 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
2094 r->res_length, &ret_nodeid);
2097 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
2100 if (error && error != -EEXIST)
2103 if (ret_nodeid == our_nodeid) {
2104 r->res_first_lkid = 0;
2106 lkb->lkb_nodeid = 0;
2108 r->res_first_lkid = lkb->lkb_id;
2109 r->res_nodeid = ret_nodeid;
2110 lkb->lkb_nodeid = ret_nodeid;
2115 static void process_lookup_list(struct dlm_rsb *r)
2117 struct dlm_lkb *lkb, *safe;
2119 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2120 list_del_init(&lkb->lkb_rsb_lookup);
2121 _request_lock(r, lkb);
2126 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
2128 static void confirm_master(struct dlm_rsb *r, int error)
2130 struct dlm_lkb *lkb;
2132 if (!r->res_first_lkid)
2138 r->res_first_lkid = 0;
2139 process_lookup_list(r);
2145 /* the remote request failed and won't be retried (it was
2146 a NOQUEUE, or has been canceled/unlocked); make a waiting
2147 lkb the first_lkid */
2149 r->res_first_lkid = 0;
2151 if (!list_empty(&r->res_lookup)) {
2152 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2154 list_del_init(&lkb->lkb_rsb_lookup);
2155 r->res_first_lkid = lkb->lkb_id;
2156 _request_lock(r, lkb);
2161 log_error(r->res_ls, "confirm_master unknown error %d", error);
2165 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2166 int namelen, unsigned long timeout_cs,
2167 void (*ast) (void *astparam),
2169 void (*bast) (void *astparam, int mode),
2170 struct dlm_args *args)
2174 /* check for invalid arg usage */
2176 if (mode < 0 || mode > DLM_LOCK_EX)
2179 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2182 if (flags & DLM_LKF_CANCEL)
2185 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2188 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2191 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2194 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2197 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2200 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2203 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2209 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2212 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2215 /* these args will be copied to the lkb in validate_lock_args,
2216 it cannot be done now because when converting locks, fields in
2217 an active lkb cannot be modified before locking the rsb */
2219 args->flags = flags;
2221 args->astparam = astparam;
2222 args->bastfn = bast;
2223 args->timeout = timeout_cs;
2231 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2233 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2234 DLM_LKF_FORCEUNLOCK))
2237 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2240 args->flags = flags;
2241 args->astparam = astarg;
2245 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2246 struct dlm_args *args)
2250 if (args->flags & DLM_LKF_CONVERT) {
2251 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2254 if (args->flags & DLM_LKF_QUECVT &&
2255 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2259 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2262 if (lkb->lkb_wait_type)
2265 if (is_overlap(lkb))
2269 lkb->lkb_exflags = args->flags;
2270 lkb->lkb_sbflags = 0;
2271 lkb->lkb_astfn = args->astfn;
2272 lkb->lkb_astparam = args->astparam;
2273 lkb->lkb_bastfn = args->bastfn;
2274 lkb->lkb_rqmode = args->mode;
2275 lkb->lkb_lksb = args->lksb;
2276 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2277 lkb->lkb_ownpid = (int) current->pid;
2278 lkb->lkb_timeout_cs = args->timeout;
2282 log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s",
2283 rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2284 lkb->lkb_status, lkb->lkb_wait_type,
2285 lkb->lkb_resource->res_name);
2289 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2292 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2293 because there may be a lookup in progress and it's valid to do
2294 cancel/unlockf on it */
2296 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2298 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2301 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2302 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2307 /* an lkb may still exist even though the lock is EOL'ed due to a
2308 cancel, unlock or failed noqueue request; an app can't use these
2309 locks; return same error as if the lkid had not been found at all */
2311 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2312 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2317 /* an lkb may be waiting for an rsb lookup to complete where the
2318 lookup was initiated by another lock */
2320 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2321 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2322 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2323 list_del_init(&lkb->lkb_rsb_lookup);
2324 queue_cast(lkb->lkb_resource, lkb,
2325 args->flags & DLM_LKF_CANCEL ?
2326 -DLM_ECANCEL : -DLM_EUNLOCK);
2327 unhold_lkb(lkb); /* undoes create_lkb() */
2329 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2334 /* cancel not allowed with another cancel/unlock in progress */
2336 if (args->flags & DLM_LKF_CANCEL) {
2337 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2340 if (is_overlap(lkb))
2343 /* don't let scand try to do a cancel */
2346 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2347 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2352 /* there's nothing to cancel */
2353 if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2354 !lkb->lkb_wait_type) {
2359 switch (lkb->lkb_wait_type) {
2360 case DLM_MSG_LOOKUP:
2361 case DLM_MSG_REQUEST:
2362 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2365 case DLM_MSG_UNLOCK:
2366 case DLM_MSG_CANCEL:
2369 /* add_to_waiters() will set OVERLAP_CANCEL */
2373 /* do we need to allow a force-unlock if there's a normal unlock
2374 already in progress? in what conditions could the normal unlock
2375 fail such that we'd want to send a force-unlock to be sure? */
2377 if (args->flags & DLM_LKF_FORCEUNLOCK) {
2378 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2381 if (is_overlap_unlock(lkb))
2384 /* don't let scand try to do a cancel */
2387 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2388 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2393 switch (lkb->lkb_wait_type) {
2394 case DLM_MSG_LOOKUP:
2395 case DLM_MSG_REQUEST:
2396 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2399 case DLM_MSG_UNLOCK:
2402 /* add_to_waiters() will set OVERLAP_UNLOCK */
2406 /* normal unlock not allowed if there's any op in progress */
2408 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
2412 /* an overlapping op shouldn't blow away exflags from other op */
2413 lkb->lkb_exflags |= args->flags;
2414 lkb->lkb_sbflags = 0;
2415 lkb->lkb_astparam = args->astparam;
2419 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
2420 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
2421 args->flags, lkb->lkb_wait_type,
2422 lkb->lkb_resource->res_name);
2427 * Four stage 4 varieties:
2428 * do_request(), do_convert(), do_unlock(), do_cancel()
2429 * These are called on the master node for the given lock and
2430 * from the central locking logic.
2433 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2437 if (can_be_granted(r, lkb, 1, NULL)) {
2439 queue_cast(r, lkb, 0);
2443 if (can_be_queued(lkb)) {
2444 error = -EINPROGRESS;
2445 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2451 queue_cast(r, lkb, -EAGAIN);
2456 static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2461 if (force_blocking_asts(lkb))
2462 send_blocking_asts_all(r, lkb);
2465 send_blocking_asts(r, lkb);
2470 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2475 /* changing an existing lock may allow others to be granted */
2477 if (can_be_granted(r, lkb, 1, &deadlk)) {
2479 queue_cast(r, lkb, 0);
2483 /* can_be_granted() detected that this lock would block in a conversion
2484 deadlock, so we leave it on the granted queue and return EDEADLK in
2485 the ast for the convert. */
2488 /* it's left on the granted queue */
2489 revert_lock(r, lkb);
2490 queue_cast(r, lkb, -EDEADLK);
2495 /* is_demoted() means the can_be_granted() above set the grmode
2496 to NL, and left us on the granted queue. This auto-demotion
2497 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2498 now grantable. We have to try to grant other converting locks
2499 before we try again to grant this one. */
2501 if (is_demoted(lkb)) {
2502 grant_pending_convert(r, DLM_LOCK_IV, NULL);
2503 if (_can_be_granted(r, lkb, 1)) {
2505 queue_cast(r, lkb, 0);
2508 /* else fall through and move to convert queue */
2511 if (can_be_queued(lkb)) {
2512 error = -EINPROGRESS;
2514 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2520 queue_cast(r, lkb, -EAGAIN);
2525 static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2530 grant_pending_locks(r);
2531 /* grant_pending_locks also sends basts */
2534 if (force_blocking_asts(lkb))
2535 send_blocking_asts_all(r, lkb);
2538 send_blocking_asts(r, lkb);
2543 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2545 remove_lock(r, lkb);
2546 queue_cast(r, lkb, -DLM_EUNLOCK);
2547 return -DLM_EUNLOCK;
2550 static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2553 grant_pending_locks(r);
2556 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
2558 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2562 error = revert_lock(r, lkb);
2564 queue_cast(r, lkb, -DLM_ECANCEL);
2565 return -DLM_ECANCEL;
2570 static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
2574 grant_pending_locks(r);
2578 * Four stage 3 varieties:
2579 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2582 /* add a new lkb to a possibly new rsb, called by requesting process */
2584 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2588 /* set_master: sets lkb nodeid from r */
2590 error = set_master(r, lkb);
2599 /* receive_request() calls do_request() on remote node */
2600 error = send_request(r, lkb);
2602 error = do_request(r, lkb);
2603 /* for remote locks the request_reply is sent
2604 between do_request and do_request_effects */
2605 do_request_effects(r, lkb, error);
2611 /* change some property of an existing lkb, e.g. mode */
2613 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2618 /* receive_convert() calls do_convert() on remote node */
2619 error = send_convert(r, lkb);
2621 error = do_convert(r, lkb);
2622 /* for remote locks the convert_reply is sent
2623 between do_convert and do_convert_effects */
2624 do_convert_effects(r, lkb, error);
2630 /* remove an existing lkb from the granted queue */
2632 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2637 /* receive_unlock() calls do_unlock() on remote node */
2638 error = send_unlock(r, lkb);
2640 error = do_unlock(r, lkb);
2641 /* for remote locks the unlock_reply is sent
2642 between do_unlock and do_unlock_effects */
2643 do_unlock_effects(r, lkb, error);
2649 /* remove an existing lkb from the convert or wait queue */
2651 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2656 /* receive_cancel() calls do_cancel() on remote node */
2657 error = send_cancel(r, lkb);
2659 error = do_cancel(r, lkb);
2660 /* for remote locks the cancel_reply is sent
2661 between do_cancel and do_cancel_effects */
2662 do_cancel_effects(r, lkb, error);
2669 * Four stage 2 varieties:
2670 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2673 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
2674 int len, struct dlm_args *args)
2679 error = validate_lock_args(ls, lkb, args);
2683 error = find_rsb(ls, name, len, R_CREATE, &r);
2690 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
2692 error = _request_lock(r, lkb);
2701 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2702 struct dlm_args *args)
2707 r = lkb->lkb_resource;
2712 error = validate_lock_args(ls, lkb, args);
2716 error = _convert_lock(r, lkb);
2723 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2724 struct dlm_args *args)
2729 r = lkb->lkb_resource;
2734 error = validate_unlock_args(lkb, args);
2738 error = _unlock_lock(r, lkb);
2745 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2746 struct dlm_args *args)
2751 r = lkb->lkb_resource;
2756 error = validate_unlock_args(lkb, args);
2760 error = _cancel_lock(r, lkb);
2768 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2771 int dlm_lock(dlm_lockspace_t *lockspace,
2773 struct dlm_lksb *lksb,
2776 unsigned int namelen,
2777 uint32_t parent_lkid,
2778 void (*ast) (void *astarg),
2780 void (*bast) (void *astarg, int mode))
2783 struct dlm_lkb *lkb;
2784 struct dlm_args args;
2785 int error, convert = flags & DLM_LKF_CONVERT;
2787 ls = dlm_find_lockspace_local(lockspace);
2791 dlm_lock_recovery(ls);
2794 error = find_lkb(ls, lksb->sb_lkid, &lkb);
2796 error = create_lkb(ls, &lkb);
2801 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
2802 astarg, bast, &args);
2807 error = convert_lock(ls, lkb, &args);
2809 error = request_lock(ls, lkb, name, namelen, &args);
2811 if (error == -EINPROGRESS)
2814 if (convert || error)
2816 if (error == -EAGAIN || error == -EDEADLK)
2819 dlm_unlock_recovery(ls);
2820 dlm_put_lockspace(ls);
2824 int dlm_unlock(dlm_lockspace_t *lockspace,
2827 struct dlm_lksb *lksb,
2831 struct dlm_lkb *lkb;
2832 struct dlm_args args;
2835 ls = dlm_find_lockspace_local(lockspace);
2839 dlm_lock_recovery(ls);
2841 error = find_lkb(ls, lkid, &lkb);
2845 error = set_unlock_args(flags, astarg, &args);
2849 if (flags & DLM_LKF_CANCEL)
2850 error = cancel_lock(ls, lkb, &args);
2852 error = unlock_lock(ls, lkb, &args);
2854 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2856 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
2861 dlm_unlock_recovery(ls);
2862 dlm_put_lockspace(ls);
2867 * send/receive routines for remote operations and replies
2871 * send_request receive_request
2872 * send_convert receive_convert
2873 * send_unlock receive_unlock
2874 * send_cancel receive_cancel
2875 * send_grant receive_grant
2876 * send_bast receive_bast
2877 * send_lookup receive_lookup
2878 * send_remove receive_remove
2881 * receive_request_reply send_request_reply
2882 * receive_convert_reply send_convert_reply
2883 * receive_unlock_reply send_unlock_reply
2884 * receive_cancel_reply send_cancel_reply
2885 * receive_lookup_reply send_lookup_reply
2888 static int _create_message(struct dlm_ls *ls, int mb_len,
2889 int to_nodeid, int mstype,
2890 struct dlm_message **ms_ret,
2891 struct dlm_mhandle **mh_ret)
2893 struct dlm_message *ms;
2894 struct dlm_mhandle *mh;
2897 /* get_buffer gives us a message handle (mh) that we need to
2898 pass into lowcomms_commit and a message buffer (mb) that we
2899 write our data into */
2901 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
2905 memset(mb, 0, mb_len);
2907 ms = (struct dlm_message *) mb;
2909 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2910 ms->m_header.h_lockspace = ls->ls_global_id;
2911 ms->m_header.h_nodeid = dlm_our_nodeid();
2912 ms->m_header.h_length = mb_len;
2913 ms->m_header.h_cmd = DLM_MSG;
2915 ms->m_type = mstype;
2922 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2923 int to_nodeid, int mstype,
2924 struct dlm_message **ms_ret,
2925 struct dlm_mhandle **mh_ret)
2927 int mb_len = sizeof(struct dlm_message);
2930 case DLM_MSG_REQUEST:
2931 case DLM_MSG_LOOKUP:
2932 case DLM_MSG_REMOVE:
2933 mb_len += r->res_length;
2935 case DLM_MSG_CONVERT:
2936 case DLM_MSG_UNLOCK:
2937 case DLM_MSG_REQUEST_REPLY:
2938 case DLM_MSG_CONVERT_REPLY:
2940 if (lkb && lkb->lkb_lvbptr)
2941 mb_len += r->res_ls->ls_lvblen;
2945 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
2949 /* further lowcomms enhancements or alternate implementations may make
2950 the return value from this function useful at some point */
2952 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2954 dlm_message_out(ms);
2955 dlm_lowcomms_commit_buffer(mh);
2959 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2960 struct dlm_message *ms)
2962 ms->m_nodeid = lkb->lkb_nodeid;
2963 ms->m_pid = lkb->lkb_ownpid;
2964 ms->m_lkid = lkb->lkb_id;
2965 ms->m_remid = lkb->lkb_remid;
2966 ms->m_exflags = lkb->lkb_exflags;
2967 ms->m_sbflags = lkb->lkb_sbflags;
2968 ms->m_flags = lkb->lkb_flags;
2969 ms->m_lvbseq = lkb->lkb_lvbseq;
2970 ms->m_status = lkb->lkb_status;
2971 ms->m_grmode = lkb->lkb_grmode;
2972 ms->m_rqmode = lkb->lkb_rqmode;
2973 ms->m_hash = r->res_hash;
2975 /* m_result and m_bastmode are set from function args,
2976 not from lkb fields */
2978 if (lkb->lkb_bastfn)
2979 ms->m_asts |= DLM_CB_BAST;
2981 ms->m_asts |= DLM_CB_CAST;
2983 /* compare with switch in create_message; send_remove() doesn't
2986 switch (ms->m_type) {
2987 case DLM_MSG_REQUEST:
2988 case DLM_MSG_LOOKUP:
2989 memcpy(ms->m_extra, r->res_name, r->res_length);
2991 case DLM_MSG_CONVERT:
2992 case DLM_MSG_UNLOCK:
2993 case DLM_MSG_REQUEST_REPLY:
2994 case DLM_MSG_CONVERT_REPLY:
2996 if (!lkb->lkb_lvbptr)
2998 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3003 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3005 struct dlm_message *ms;
3006 struct dlm_mhandle *mh;
3007 int to_nodeid, error;
3009 to_nodeid = r->res_nodeid;
3011 error = add_to_waiters(lkb, mstype, to_nodeid);
3015 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3019 send_args(r, lkb, ms);
3021 error = send_message(mh, ms);
3027 remove_from_waiters(lkb, msg_reply_type(mstype));
3031 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3033 return send_common(r, lkb, DLM_MSG_REQUEST);
3036 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3040 error = send_common(r, lkb, DLM_MSG_CONVERT);
3042 /* down conversions go without a reply from the master */
3043 if (!error && down_conversion(lkb)) {
3044 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3045 r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS;
3046 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3047 r->res_ls->ls_stub_ms.m_result = 0;
3048 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3054 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
3055 MASTER_UNCERTAIN to force the next request on the rsb to confirm
3056 that the master is still correct. */
3058 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3060 return send_common(r, lkb, DLM_MSG_UNLOCK);
3063 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3065 return send_common(r, lkb, DLM_MSG_CANCEL);
3068 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3070 struct dlm_message *ms;
3071 struct dlm_mhandle *mh;
3072 int to_nodeid, error;
3074 to_nodeid = lkb->lkb_nodeid;
3076 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
3080 send_args(r, lkb, ms);
3084 error = send_message(mh, ms);
3089 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3091 struct dlm_message *ms;
3092 struct dlm_mhandle *mh;
3093 int to_nodeid, error;
3095 to_nodeid = lkb->lkb_nodeid;
3097 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
3101 send_args(r, lkb, ms);
3103 ms->m_bastmode = mode;
3105 error = send_message(mh, ms);
3110 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3112 struct dlm_message *ms;
3113 struct dlm_mhandle *mh;
3114 int to_nodeid, error;
3116 to_nodeid = dlm_dir_nodeid(r);
3118 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3122 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
3126 send_args(r, lkb, ms);
3128 error = send_message(mh, ms);
3134 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3138 static int send_remove(struct dlm_rsb *r)
3140 struct dlm_message *ms;
3141 struct dlm_mhandle *mh;
3142 int to_nodeid, error;
3144 to_nodeid = dlm_dir_nodeid(r);
3146 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
3150 memcpy(ms->m_extra, r->res_name, r->res_length);
3151 ms->m_hash = r->res_hash;
3153 error = send_message(mh, ms);
3158 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3161 struct dlm_message *ms;
3162 struct dlm_mhandle *mh;
3163 int to_nodeid, error;
3165 to_nodeid = lkb->lkb_nodeid;
3167 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
3171 send_args(r, lkb, ms);
3175 error = send_message(mh, ms);
3180 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3182 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3185 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3187 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3190 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3192 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3195 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3197 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3200 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3201 int ret_nodeid, int rv)
3203 struct dlm_rsb *r = &ls->ls_stub_rsb;
3204 struct dlm_message *ms;
3205 struct dlm_mhandle *mh;
3206 int error, nodeid = ms_in->m_header.h_nodeid;
3208 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
3212 ms->m_lkid = ms_in->m_lkid;
3214 ms->m_nodeid = ret_nodeid;
3216 error = send_message(mh, ms);
3221 /* which args we save from a received message depends heavily on the type
3222 of message, unlike the send side where we can safely send everything about
3223 the lkb for any type of message */
3225 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3227 lkb->lkb_exflags = ms->m_exflags;
3228 lkb->lkb_sbflags = ms->m_sbflags;
3229 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3230 (ms->m_flags & 0x0000FFFF);
3233 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3235 if (ms->m_flags == DLM_IFL_STUB_MS)
3238 lkb->lkb_sbflags = ms->m_sbflags;
3239 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3240 (ms->m_flags & 0x0000FFFF);
3243 static int receive_extralen(struct dlm_message *ms)
3245 return (ms->m_header.h_length - sizeof(struct dlm_message));
3248 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3249 struct dlm_message *ms)
3253 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3254 if (!lkb->lkb_lvbptr)
3255 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3256 if (!lkb->lkb_lvbptr)
3258 len = receive_extralen(ms);
3259 if (len > DLM_RESNAME_MAXLEN)
3260 len = DLM_RESNAME_MAXLEN;
3261 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3266 static void fake_bastfn(void *astparam, int mode)
3268 log_print("fake_bastfn should not be called");
3271 static void fake_astfn(void *astparam)
3273 log_print("fake_astfn should not be called");
3276 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3277 struct dlm_message *ms)
3279 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3280 lkb->lkb_ownpid = ms->m_pid;
3281 lkb->lkb_remid = ms->m_lkid;
3282 lkb->lkb_grmode = DLM_LOCK_IV;
3283 lkb->lkb_rqmode = ms->m_rqmode;
3285 lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
3286 lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
3288 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3289 /* lkb was just created so there won't be an lvb yet */
3290 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3291 if (!lkb->lkb_lvbptr)
3298 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3299 struct dlm_message *ms)
3301 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3304 if (receive_lvb(ls, lkb, ms))
3307 lkb->lkb_rqmode = ms->m_rqmode;
3308 lkb->lkb_lvbseq = ms->m_lvbseq;
3313 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3314 struct dlm_message *ms)
3316 if (receive_lvb(ls, lkb, ms))
3321 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3322 uses to send a reply and that the remote end uses to process the reply. */
3324 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3326 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3327 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3328 lkb->lkb_remid = ms->m_lkid;
3331 /* This is called after the rsb is locked so that we can safely inspect
3332 fields in the lkb. */
3334 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3336 int from = ms->m_header.h_nodeid;
3339 switch (ms->m_type) {
3340 case DLM_MSG_CONVERT:
3341 case DLM_MSG_UNLOCK:
3342 case DLM_MSG_CANCEL:
3343 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3347 case DLM_MSG_CONVERT_REPLY:
3348 case DLM_MSG_UNLOCK_REPLY:
3349 case DLM_MSG_CANCEL_REPLY:
3352 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3356 case DLM_MSG_REQUEST_REPLY:
3357 if (!is_process_copy(lkb))
3359 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3368 log_error(lkb->lkb_resource->res_ls,
3369 "ignore invalid message %d from %d %x %x %x %d",
3370 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
3371 lkb->lkb_flags, lkb->lkb_nodeid);
3375 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3377 struct dlm_lkb *lkb;
3381 error = create_lkb(ls, &lkb);
3385 receive_flags(lkb, ms);
3386 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3387 error = receive_request_args(ls, lkb, ms);
3393 namelen = receive_extralen(ms);
3395 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
3404 error = do_request(r, lkb);
3405 send_request_reply(r, lkb, error);
3406 do_request_effects(r, lkb, error);
3411 if (error == -EINPROGRESS)
3418 setup_stub_lkb(ls, ms);
3419 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3422 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3424 struct dlm_lkb *lkb;
3426 int error, reply = 1;
3428 error = find_lkb(ls, ms->m_remid, &lkb);
3432 r = lkb->lkb_resource;
3437 error = validate_message(lkb, ms);
3441 receive_flags(lkb, ms);
3443 error = receive_convert_args(ls, lkb, ms);
3445 send_convert_reply(r, lkb, error);
3449 reply = !down_conversion(lkb);
3451 error = do_convert(r, lkb);
3453 send_convert_reply(r, lkb, error);
3454 do_convert_effects(r, lkb, error);
3462 setup_stub_lkb(ls, ms);
3463 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3466 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3468 struct dlm_lkb *lkb;
3472 error = find_lkb(ls, ms->m_remid, &lkb);
3476 r = lkb->lkb_resource;
3481 error = validate_message(lkb, ms);
3485 receive_flags(lkb, ms);
3487 error = receive_unlock_args(ls, lkb, ms);
3489 send_unlock_reply(r, lkb, error);
3493 error = do_unlock(r, lkb);
3494 send_unlock_reply(r, lkb, error);
3495 do_unlock_effects(r, lkb, error);
3503 setup_stub_lkb(ls, ms);
3504 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3507 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3509 struct dlm_lkb *lkb;
3513 error = find_lkb(ls, ms->m_remid, &lkb);
3517 receive_flags(lkb, ms);
3519 r = lkb->lkb_resource;
3524 error = validate_message(lkb, ms);
3528 error = do_cancel(r, lkb);
3529 send_cancel_reply(r, lkb, error);
3530 do_cancel_effects(r, lkb, error);
3538 setup_stub_lkb(ls, ms);
3539 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3542 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
3544 struct dlm_lkb *lkb;
3548 error = find_lkb(ls, ms->m_remid, &lkb);
3550 log_debug(ls, "receive_grant from %d no lkb %x",
3551 ms->m_header.h_nodeid, ms->m_remid);
3555 r = lkb->lkb_resource;
3560 error = validate_message(lkb, ms);
3564 receive_flags_reply(lkb, ms);
3565 if (is_altmode(lkb))
3566 munge_altmode(lkb, ms);
3567 grant_lock_pc(r, lkb, ms);
3568 queue_cast(r, lkb, 0);
3575 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
3577 struct dlm_lkb *lkb;
3581 error = find_lkb(ls, ms->m_remid, &lkb);
3583 log_debug(ls, "receive_bast from %d no lkb %x",
3584 ms->m_header.h_nodeid, ms->m_remid);
3588 r = lkb->lkb_resource;
3593 error = validate_message(lkb, ms);
3597 queue_bast(r, lkb, ms->m_bastmode);
3604 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
3606 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
3608 from_nodeid = ms->m_header.h_nodeid;
3609 our_nodeid = dlm_our_nodeid();
3611 len = receive_extralen(ms);
3613 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3614 if (dir_nodeid != our_nodeid) {
3615 log_error(ls, "lookup dir_nodeid %d from %d",
3616 dir_nodeid, from_nodeid);
3622 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
3624 /* Optimization: we're master so treat lookup as a request */
3625 if (!error && ret_nodeid == our_nodeid) {
3626 receive_request(ls, ms);
3630 send_lookup_reply(ls, ms, ret_nodeid, error);
3633 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
3635 int len, dir_nodeid, from_nodeid;
3637 from_nodeid = ms->m_header.h_nodeid;
3639 len = receive_extralen(ms);
3641 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3642 if (dir_nodeid != dlm_our_nodeid()) {
3643 log_error(ls, "remove dir entry dir_nodeid %d from %d",
3644 dir_nodeid, from_nodeid);
3648 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
3651 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
3653 do_purge(ls, ms->m_nodeid, ms->m_pid);
3656 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
3658 struct dlm_lkb *lkb;
3660 int error, mstype, result;
3662 error = find_lkb(ls, ms->m_remid, &lkb);
3664 log_debug(ls, "receive_request_reply from %d no lkb %x",
3665 ms->m_header.h_nodeid, ms->m_remid);
3669 r = lkb->lkb_resource;
3673 error = validate_message(lkb, ms);
3677 mstype = lkb->lkb_wait_type;
3678 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
3682 /* Optimization: the dir node was also the master, so it took our
3683 lookup as a request and sent request reply instead of lookup reply */
3684 if (mstype == DLM_MSG_LOOKUP) {
3685 r->res_nodeid = ms->m_header.h_nodeid;
3686 lkb->lkb_nodeid = r->res_nodeid;
3689 /* this is the value returned from do_request() on the master */
3690 result = ms->m_result;
3694 /* request would block (be queued) on remote master */
3695 queue_cast(r, lkb, -EAGAIN);
3696 confirm_master(r, -EAGAIN);
3697 unhold_lkb(lkb); /* undoes create_lkb() */
3702 /* request was queued or granted on remote master */
3703 receive_flags_reply(lkb, ms);
3704 lkb->lkb_remid = ms->m_lkid;
3705 if (is_altmode(lkb))
3706 munge_altmode(lkb, ms);
3708 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3711 grant_lock_pc(r, lkb, ms);
3712 queue_cast(r, lkb, 0);
3714 confirm_master(r, result);
3719 /* find_rsb failed to find rsb or rsb wasn't master */
3720 log_debug(ls, "receive_request_reply %x %x master diff %d %d",
3721 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result);
3723 lkb->lkb_nodeid = -1;
3725 if (is_overlap(lkb)) {
3726 /* we'll ignore error in cancel/unlock reply */
3727 queue_cast_overlap(r, lkb);
3728 confirm_master(r, result);
3729 unhold_lkb(lkb); /* undoes create_lkb() */
3731 _request_lock(r, lkb);
3735 log_error(ls, "receive_request_reply %x error %d",
3736 lkb->lkb_id, result);
3739 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
3740 log_debug(ls, "receive_request_reply %x result %d unlock",
3741 lkb->lkb_id, result);
3742 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3743 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3744 send_unlock(r, lkb);
3745 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
3746 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
3747 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3748 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3749 send_cancel(r, lkb);
3751 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3752 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3760 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3761 struct dlm_message *ms)
3763 /* this is the value returned from do_convert() on the master */
3764 switch (ms->m_result) {
3766 /* convert would block (be queued) on remote master */
3767 queue_cast(r, lkb, -EAGAIN);
3771 receive_flags_reply(lkb, ms);
3772 revert_lock_pc(r, lkb);
3773 queue_cast(r, lkb, -EDEADLK);
3777 /* convert was queued on remote master */
3778 receive_flags_reply(lkb, ms);
3779 if (is_demoted(lkb))
3782 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3787 /* convert was granted on remote master */
3788 receive_flags_reply(lkb, ms);
3789 if (is_demoted(lkb))
3791 grant_lock_pc(r, lkb, ms);
3792 queue_cast(r, lkb, 0);
3796 log_error(r->res_ls, "receive_convert_reply %x error %d",
3797 lkb->lkb_id, ms->m_result);
3801 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3803 struct dlm_rsb *r = lkb->lkb_resource;
3809 error = validate_message(lkb, ms);
3813 /* stub reply can happen with waiters_mutex held */
3814 error = remove_from_waiters_ms(lkb, ms);
3818 __receive_convert_reply(r, lkb, ms);
3824 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
3826 struct dlm_lkb *lkb;
3829 error = find_lkb(ls, ms->m_remid, &lkb);
3831 log_debug(ls, "receive_convert_reply from %d no lkb %x",
3832 ms->m_header.h_nodeid, ms->m_remid);
3836 _receive_convert_reply(lkb, ms);
3840 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3842 struct dlm_rsb *r = lkb->lkb_resource;
3848 error = validate_message(lkb, ms);
3852 /* stub reply can happen with waiters_mutex held */
3853 error = remove_from_waiters_ms(lkb, ms);
3857 /* this is the value returned from do_unlock() on the master */
3859 switch (ms->m_result) {
3861 receive_flags_reply(lkb, ms);
3862 remove_lock_pc(r, lkb);
3863 queue_cast(r, lkb, -DLM_EUNLOCK);
3868 log_error(r->res_ls, "receive_unlock_reply %x error %d",
3869 lkb->lkb_id, ms->m_result);
3876 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
3878 struct dlm_lkb *lkb;
3881 error = find_lkb(ls, ms->m_remid, &lkb);
3883 log_debug(ls, "receive_unlock_reply from %d no lkb %x",
3884 ms->m_header.h_nodeid, ms->m_remid);
3888 _receive_unlock_reply(lkb, ms);
3892 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3894 struct dlm_rsb *r = lkb->lkb_resource;
3900 error = validate_message(lkb, ms);
3904 /* stub reply can happen with waiters_mutex held */
3905 error = remove_from_waiters_ms(lkb, ms);
3909 /* this is the value returned from do_cancel() on the master */
3911 switch (ms->m_result) {
3913 receive_flags_reply(lkb, ms);
3914 revert_lock_pc(r, lkb);
3915 queue_cast(r, lkb, -DLM_ECANCEL);
3920 log_error(r->res_ls, "receive_cancel_reply %x error %d",
3921 lkb->lkb_id, ms->m_result);
3928 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
3930 struct dlm_lkb *lkb;
3933 error = find_lkb(ls, ms->m_remid, &lkb);
3935 log_debug(ls, "receive_cancel_reply from %d no lkb %x",
3936 ms->m_header.h_nodeid, ms->m_remid);
3940 _receive_cancel_reply(lkb, ms);
3944 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
3946 struct dlm_lkb *lkb;
3948 int error, ret_nodeid;
3950 error = find_lkb(ls, ms->m_lkid, &lkb);
3952 log_error(ls, "receive_lookup_reply no lkb");
3956 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
3957 FIXME: will a non-zero error ever be returned? */
3959 r = lkb->lkb_resource;
3963 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3967 ret_nodeid = ms->m_nodeid;
3968 if (ret_nodeid == dlm_our_nodeid()) {
3971 r->res_first_lkid = 0;
3973 /* set_master() will copy res_nodeid to lkb_nodeid */
3974 r->res_nodeid = ret_nodeid;
3977 if (is_overlap(lkb)) {
3978 log_debug(ls, "receive_lookup_reply %x unlock %x",
3979 lkb->lkb_id, lkb->lkb_flags);
3980 queue_cast_overlap(r, lkb);
3981 unhold_lkb(lkb); /* undoes create_lkb() */
3985 _request_lock(r, lkb);
3989 process_lookup_list(r);
3996 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms)
3998 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
3999 log_debug(ls, "ignore non-member message %d from %d %x %x %d",
4000 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
4001 ms->m_remid, ms->m_result);
4005 switch (ms->m_type) {
4007 /* messages sent to a master node */
4009 case DLM_MSG_REQUEST:
4010 receive_request(ls, ms);
4013 case DLM_MSG_CONVERT:
4014 receive_convert(ls, ms);
4017 case DLM_MSG_UNLOCK:
4018 receive_unlock(ls, ms);
4021 case DLM_MSG_CANCEL:
4022 receive_cancel(ls, ms);
4025 /* messages sent from a master node (replies to above) */
4027 case DLM_MSG_REQUEST_REPLY:
4028 receive_request_reply(ls, ms);
4031 case DLM_MSG_CONVERT_REPLY:
4032 receive_convert_reply(ls, ms);
4035 case DLM_MSG_UNLOCK_REPLY:
4036 receive_unlock_reply(ls, ms);
4039 case DLM_MSG_CANCEL_REPLY:
4040 receive_cancel_reply(ls, ms);
4043 /* messages sent from a master node (only two types of async msg) */
4046 receive_grant(ls, ms);
4050 receive_bast(ls, ms);
4053 /* messages sent to a dir node */
4055 case DLM_MSG_LOOKUP:
4056 receive_lookup(ls, ms);
4059 case DLM_MSG_REMOVE:
4060 receive_remove(ls, ms);
4063 /* messages sent from a dir node (remove has no reply) */
4065 case DLM_MSG_LOOKUP_REPLY:
4066 receive_lookup_reply(ls, ms);
4069 /* other messages */
4072 receive_purge(ls, ms);
4076 log_error(ls, "unknown message type %d", ms->m_type);
4080 /* If the lockspace is in recovery mode (locking stopped), then normal
4081 messages are saved on the requestqueue for processing after recovery is
4082 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
4083 messages off the requestqueue before we process new ones. This occurs right
4084 after recovery completes when we transition from saving all messages on
4085 requestqueue, to processing all the saved messages, to processing new
4086 messages as they arrive. */
4088 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4091 if (dlm_locking_stopped(ls)) {
4092 dlm_add_requestqueue(ls, nodeid, ms);
4094 dlm_wait_requestqueue(ls);
4095 _receive_message(ls, ms);
4099 /* This is called by dlm_recoverd to process messages that were saved on
4100 the requestqueue. */
4102 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms)
4104 _receive_message(ls, ms);
4107 /* This is called by the midcomms layer when something is received for
4108 the lockspace. It could be either a MSG (normal message sent as part of
4109 standard locking activity) or an RCOM (recovery message sent as part of
4110 lockspace recovery). */
4112 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
4114 struct dlm_header *hd = &p->header;
4118 switch (hd->h_cmd) {
4120 dlm_message_in(&p->message);
4121 type = p->message.m_type;
4124 dlm_rcom_in(&p->rcom);
4125 type = p->rcom.rc_type;
4128 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
4132 if (hd->h_nodeid != nodeid) {
4133 log_print("invalid h_nodeid %d from %d lockspace %x",
4134 hd->h_nodeid, nodeid, hd->h_lockspace);
4138 ls = dlm_find_lockspace_global(hd->h_lockspace);
4140 if (dlm_config.ci_log_debug)
4141 log_print("invalid lockspace %x from %d cmd %d type %d",
4142 hd->h_lockspace, nodeid, hd->h_cmd, type);
4144 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
4145 dlm_send_ls_not_ready(nodeid, &p->rcom);
4149 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
4150 be inactive (in this ls) before transitioning to recovery mode */
4152 down_read(&ls->ls_recv_active);
4153 if (hd->h_cmd == DLM_MSG)
4154 dlm_receive_message(ls, &p->message, nodeid);
4156 dlm_receive_rcom(ls, &p->rcom, nodeid);
4157 up_read(&ls->ls_recv_active);
4159 dlm_put_lockspace(ls);
4162 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
4163 struct dlm_message *ms_stub)
4165 if (middle_conversion(lkb)) {
4167 memset(ms_stub, 0, sizeof(struct dlm_message));
4168 ms_stub->m_flags = DLM_IFL_STUB_MS;
4169 ms_stub->m_type = DLM_MSG_CONVERT_REPLY;
4170 ms_stub->m_result = -EINPROGRESS;
4171 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
4172 _receive_convert_reply(lkb, ms_stub);
4174 /* Same special case as in receive_rcom_lock_args() */
4175 lkb->lkb_grmode = DLM_LOCK_IV;
4176 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
4179 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
4180 lkb->lkb_flags |= DLM_IFL_RESEND;
4183 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
4184 conversions are async; there's no reply from the remote master */
4187 /* A waiting lkb needs recovery if the master node has failed, or
4188 the master node is changing (only when no directory is used) */
4190 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
4192 if (dlm_is_removed(ls, lkb->lkb_nodeid))
4195 if (!dlm_no_directory(ls))
4198 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
4204 /* Recovery for locks that are waiting for replies from nodes that are now
4205 gone. We can just complete unlocks and cancels by faking a reply from the
4206 dead node. Requests and up-conversions we flag to be resent after
4207 recovery. Down-conversions can just be completed with a fake reply like
4208 unlocks. Conversions between PR and CW need special attention. */
4210 void dlm_recover_waiters_pre(struct dlm_ls *ls)
4212 struct dlm_lkb *lkb, *safe;
4213 struct dlm_message *ms_stub;
4214 int wait_type, stub_unlock_result, stub_cancel_result;
4216 ms_stub = kmalloc(sizeof(struct dlm_message), GFP_KERNEL);
4218 log_error(ls, "dlm_recover_waiters_pre no mem");
4222 mutex_lock(&ls->ls_waiters_mutex);
4224 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
4226 /* exclude debug messages about unlocks because there can be so
4227 many and they aren't very interesting */
4229 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
4230 log_debug(ls, "recover_waiter %x nodeid %d "
4231 "msg %d to %d", lkb->lkb_id, lkb->lkb_nodeid,
4232 lkb->lkb_wait_type, lkb->lkb_wait_nodeid);
4235 /* all outstanding lookups, regardless of destination will be
4236 resent after recovery is done */
4238 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
4239 lkb->lkb_flags |= DLM_IFL_RESEND;
4243 if (!waiter_needs_recovery(ls, lkb))
4246 wait_type = lkb->lkb_wait_type;
4247 stub_unlock_result = -DLM_EUNLOCK;
4248 stub_cancel_result = -DLM_ECANCEL;
4250 /* Main reply may have been received leaving a zero wait_type,
4251 but a reply for the overlapping op may not have been
4252 received. In that case we need to fake the appropriate
4253 reply for the overlap op. */
4256 if (is_overlap_cancel(lkb)) {
4257 wait_type = DLM_MSG_CANCEL;
4258 if (lkb->lkb_grmode == DLM_LOCK_IV)
4259 stub_cancel_result = 0;
4261 if (is_overlap_unlock(lkb)) {
4262 wait_type = DLM_MSG_UNLOCK;
4263 if (lkb->lkb_grmode == DLM_LOCK_IV)
4264 stub_unlock_result = -ENOENT;
4267 log_debug(ls, "rwpre overlap %x %x %d %d %d",
4268 lkb->lkb_id, lkb->lkb_flags, wait_type,
4269 stub_cancel_result, stub_unlock_result);
4272 switch (wait_type) {
4274 case DLM_MSG_REQUEST:
4275 lkb->lkb_flags |= DLM_IFL_RESEND;
4278 case DLM_MSG_CONVERT:
4279 recover_convert_waiter(ls, lkb, ms_stub);
4282 case DLM_MSG_UNLOCK:
4284 memset(ms_stub, 0, sizeof(struct dlm_message));
4285 ms_stub->m_flags = DLM_IFL_STUB_MS;
4286 ms_stub->m_type = DLM_MSG_UNLOCK_REPLY;
4287 ms_stub->m_result = stub_unlock_result;
4288 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
4289 _receive_unlock_reply(lkb, ms_stub);
4293 case DLM_MSG_CANCEL:
4295 memset(ms_stub, 0, sizeof(struct dlm_message));
4296 ms_stub->m_flags = DLM_IFL_STUB_MS;
4297 ms_stub->m_type = DLM_MSG_CANCEL_REPLY;
4298 ms_stub->m_result = stub_cancel_result;
4299 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid;
4300 _receive_cancel_reply(lkb, ms_stub);
4305 log_error(ls, "invalid lkb wait_type %d %d",
4306 lkb->lkb_wait_type, wait_type);
4310 mutex_unlock(&ls->ls_waiters_mutex);
4314 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
4316 struct dlm_lkb *lkb;
4319 mutex_lock(&ls->ls_waiters_mutex);
4320 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
4321 if (lkb->lkb_flags & DLM_IFL_RESEND) {
4327 mutex_unlock(&ls->ls_waiters_mutex);
4334 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
4335 master or dir-node for r. Processing the lkb may result in it being placed
4338 /* We do this after normal locking has been enabled and any saved messages
4339 (in requestqueue) have been processed. We should be confident that at
4340 this point we won't get or process a reply to any of these waiting
4341 operations. But, new ops may be coming in on the rsbs/locks here from
4342 userspace or remotely. */
4344 /* there may have been an overlap unlock/cancel prior to recovery or after
4345 recovery. if before, the lkb may still have a pos wait_count; if after, the
4346 overlap flag would just have been set and nothing new sent. we can be
4347 confident here than any replies to either the initial op or overlap ops
4348 prior to recovery have been received. */
4350 int dlm_recover_waiters_post(struct dlm_ls *ls)
4352 struct dlm_lkb *lkb;
4354 int error = 0, mstype, err, oc, ou;
4357 if (dlm_locking_stopped(ls)) {
4358 log_debug(ls, "recover_waiters_post aborted");
4363 lkb = find_resend_waiter(ls);
4367 r = lkb->lkb_resource;
4371 mstype = lkb->lkb_wait_type;
4372 oc = is_overlap_cancel(lkb);
4373 ou = is_overlap_unlock(lkb);
4376 log_debug(ls, "recover_waiter %x nodeid %d msg %d r_nodeid %d",
4377 lkb->lkb_id, lkb->lkb_nodeid, mstype, r->res_nodeid);
4379 /* At this point we assume that we won't get a reply to any
4380 previous op or overlap op on this lock. First, do a big
4381 remove_from_waiters() for all previous ops. */
4383 lkb->lkb_flags &= ~DLM_IFL_RESEND;
4384 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4385 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4386 lkb->lkb_wait_type = 0;
4387 lkb->lkb_wait_count = 0;
4388 mutex_lock(&ls->ls_waiters_mutex);
4389 list_del_init(&lkb->lkb_wait_reply);
4390 mutex_unlock(&ls->ls_waiters_mutex);
4391 unhold_lkb(lkb); /* for waiters list */
4394 /* do an unlock or cancel instead of resending */
4396 case DLM_MSG_LOOKUP:
4397 case DLM_MSG_REQUEST:
4398 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
4400 unhold_lkb(lkb); /* undoes create_lkb() */
4402 case DLM_MSG_CONVERT:
4404 queue_cast(r, lkb, -DLM_ECANCEL);
4406 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
4407 _unlock_lock(r, lkb);
4415 case DLM_MSG_LOOKUP:
4416 case DLM_MSG_REQUEST:
4417 _request_lock(r, lkb);
4419 confirm_master(r, 0);
4421 case DLM_MSG_CONVERT:
4422 _convert_lock(r, lkb);
4430 log_error(ls, "recover_waiters_post %x %d %x %d %d",
4431 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou);
4440 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
4441 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
4443 struct dlm_ls *ls = r->res_ls;
4444 struct dlm_lkb *lkb, *safe;
4446 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
4447 if (test(ls, lkb)) {
4448 rsb_set_flag(r, RSB_LOCKS_PURGED);
4450 /* this put should free the lkb */
4451 if (!dlm_put_lkb(lkb))
4452 log_error(ls, "purged lkb not released");
4457 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4459 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
4462 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4464 return is_master_copy(lkb);
4467 static void purge_dead_locks(struct dlm_rsb *r)
4469 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
4470 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
4471 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
4474 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
4476 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
4477 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
4478 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
4481 /* Get rid of locks held by nodes that are gone. */
4483 int dlm_purge_locks(struct dlm_ls *ls)
4487 log_debug(ls, "dlm_purge_locks");
4489 down_write(&ls->ls_root_sem);
4490 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
4494 purge_dead_locks(r);
4500 up_write(&ls->ls_root_sem);
4505 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
4508 struct dlm_rsb *r, *r_ret = NULL;
4510 spin_lock(&ls->ls_rsbtbl[bucket].lock);
4511 for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
4512 r = rb_entry(n, struct dlm_rsb, res_hashnode);
4513 if (!rsb_flag(r, RSB_LOCKS_PURGED))
4516 rsb_clear_flag(r, RSB_LOCKS_PURGED);
4520 spin_unlock(&ls->ls_rsbtbl[bucket].lock);
4524 void dlm_grant_after_purge(struct dlm_ls *ls)
4530 r = find_purged_rsb(ls, bucket);
4532 if (bucket == ls->ls_rsbtbl_size - 1)
4539 grant_pending_locks(r);
4540 confirm_master(r, 0);
4548 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
4551 struct dlm_lkb *lkb;
4553 list_for_each_entry(lkb, head, lkb_statequeue) {
4554 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
4560 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
4563 struct dlm_lkb *lkb;
4565 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
4568 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
4571 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
4577 /* needs at least dlm_rcom + rcom_lock */
4578 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
4579 struct dlm_rsb *r, struct dlm_rcom *rc)
4581 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4583 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
4584 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
4585 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
4586 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
4587 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
4588 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4589 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
4590 lkb->lkb_rqmode = rl->rl_rqmode;
4591 lkb->lkb_grmode = rl->rl_grmode;
4592 /* don't set lkb_status because add_lkb wants to itself */
4594 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
4595 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
4597 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
4598 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
4599 sizeof(struct rcom_lock);
4600 if (lvblen > ls->ls_lvblen)
4602 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
4603 if (!lkb->lkb_lvbptr)
4605 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
4608 /* Conversions between PR and CW (middle modes) need special handling.
4609 The real granted mode of these converting locks cannot be determined
4610 until all locks have been rebuilt on the rsb (recover_conversion) */
4612 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
4613 middle_conversion(lkb)) {
4614 rl->rl_status = DLM_LKSTS_CONVERT;
4615 lkb->lkb_grmode = DLM_LOCK_IV;
4616 rsb_set_flag(r, RSB_RECOVER_CONVERT);
4622 /* This lkb may have been recovered in a previous aborted recovery so we need
4623 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4624 If so we just send back a standard reply. If not, we create a new lkb with
4625 the given values and send back our lkid. We send back our lkid by sending
4626 back the rcom_lock struct we got but with the remid field filled in. */
4628 /* needs at least dlm_rcom + rcom_lock */
4629 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4631 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4633 struct dlm_lkb *lkb;
4636 if (rl->rl_parent_lkid) {
4637 error = -EOPNOTSUPP;
4641 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
4648 lkb = search_remid(r, rc->rc_header.h_nodeid, le32_to_cpu(rl->rl_lkid));
4654 error = create_lkb(ls, &lkb);
4658 error = receive_rcom_lock_args(ls, lkb, r, rc);
4665 add_lkb(r, lkb, rl->rl_status);
4669 /* this is the new value returned to the lock holder for
4670 saving in its process-copy lkb */
4671 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
4678 log_debug(ls, "recover_master_copy %d %x", error,
4679 le32_to_cpu(rl->rl_lkid));
4680 rl->rl_result = cpu_to_le32(error);
4684 /* needs at least dlm_rcom + rcom_lock */
4685 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4687 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4689 struct dlm_lkb *lkb;
4692 error = find_lkb(ls, le32_to_cpu(rl->rl_lkid), &lkb);
4694 log_error(ls, "recover_process_copy no lkid %x",
4695 le32_to_cpu(rl->rl_lkid));
4699 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
4701 error = le32_to_cpu(rl->rl_result);
4703 r = lkb->lkb_resource;
4709 /* There's a chance the new master received our lock before
4710 dlm_recover_master_reply(), this wouldn't happen if we did
4711 a barrier between recover_masters and recover_locks. */
4712 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
4713 (unsigned long)r, r->res_name);
4714 dlm_send_rcom_lock(r, lkb);
4717 log_debug(ls, "master copy exists %x", lkb->lkb_id);
4720 lkb->lkb_remid = le32_to_cpu(rl->rl_remid);
4723 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
4724 error, lkb->lkb_id);
4727 /* an ack for dlm_recover_locks() which waits for replies from
4728 all the locks it sends to new masters */
4729 dlm_recovered_lock(r);
4738 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
4739 int mode, uint32_t flags, void *name, unsigned int namelen,
4740 unsigned long timeout_cs)
4742 struct dlm_lkb *lkb;
4743 struct dlm_args args;
4746 dlm_lock_recovery(ls);
4748 error = create_lkb(ls, &lkb);
4754 if (flags & DLM_LKF_VALBLK) {
4755 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
4756 if (!ua->lksb.sb_lvbptr) {
4764 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
4765 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4766 lock and that lkb_astparam is the dlm_user_args structure. */
4768 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
4769 fake_astfn, ua, fake_bastfn, &args);
4770 lkb->lkb_flags |= DLM_IFL_USER;
4777 error = request_lock(ls, lkb, name, namelen, &args);
4793 /* add this new lkb to the per-process list of locks */
4794 spin_lock(&ua->proc->locks_spin);
4796 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
4797 spin_unlock(&ua->proc->locks_spin);
4799 dlm_unlock_recovery(ls);
4803 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4804 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
4805 unsigned long timeout_cs)
4807 struct dlm_lkb *lkb;
4808 struct dlm_args args;
4809 struct dlm_user_args *ua;
4812 dlm_lock_recovery(ls);
4814 error = find_lkb(ls, lkid, &lkb);
4818 /* user can change the params on its lock when it converts it, or
4819 add an lvb that didn't exist before */
4823 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
4824 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
4825 if (!ua->lksb.sb_lvbptr) {
4830 if (lvb_in && ua->lksb.sb_lvbptr)
4831 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4833 ua->xid = ua_tmp->xid;
4834 ua->castparam = ua_tmp->castparam;
4835 ua->castaddr = ua_tmp->castaddr;
4836 ua->bastparam = ua_tmp->bastparam;
4837 ua->bastaddr = ua_tmp->bastaddr;
4838 ua->user_lksb = ua_tmp->user_lksb;
4840 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
4841 fake_astfn, ua, fake_bastfn, &args);
4845 error = convert_lock(ls, lkb, &args);
4847 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
4852 dlm_unlock_recovery(ls);
4857 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4858 uint32_t flags, uint32_t lkid, char *lvb_in)
4860 struct dlm_lkb *lkb;
4861 struct dlm_args args;
4862 struct dlm_user_args *ua;
4865 dlm_lock_recovery(ls);
4867 error = find_lkb(ls, lkid, &lkb);
4873 if (lvb_in && ua->lksb.sb_lvbptr)
4874 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4875 if (ua_tmp->castparam)
4876 ua->castparam = ua_tmp->castparam;
4877 ua->user_lksb = ua_tmp->user_lksb;
4879 error = set_unlock_args(flags, ua, &args);
4883 error = unlock_lock(ls, lkb, &args);
4885 if (error == -DLM_EUNLOCK)
4887 /* from validate_unlock_args() */
4888 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
4893 spin_lock(&ua->proc->locks_spin);
4894 /* dlm_user_add_cb() may have already taken lkb off the proc list */
4895 if (!list_empty(&lkb->lkb_ownqueue))
4896 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
4897 spin_unlock(&ua->proc->locks_spin);
4901 dlm_unlock_recovery(ls);
4906 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4907 uint32_t flags, uint32_t lkid)
4909 struct dlm_lkb *lkb;
4910 struct dlm_args args;
4911 struct dlm_user_args *ua;
4914 dlm_lock_recovery(ls);
4916 error = find_lkb(ls, lkid, &lkb);
4921 if (ua_tmp->castparam)
4922 ua->castparam = ua_tmp->castparam;
4923 ua->user_lksb = ua_tmp->user_lksb;
4925 error = set_unlock_args(flags, ua, &args);
4929 error = cancel_lock(ls, lkb, &args);
4931 if (error == -DLM_ECANCEL)
4933 /* from validate_unlock_args() */
4934 if (error == -EBUSY)
4939 dlm_unlock_recovery(ls);
4944 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
4946 struct dlm_lkb *lkb;
4947 struct dlm_args args;
4948 struct dlm_user_args *ua;
4952 dlm_lock_recovery(ls);
4954 error = find_lkb(ls, lkid, &lkb);
4960 error = set_unlock_args(flags, ua, &args);
4964 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4966 r = lkb->lkb_resource;
4970 error = validate_unlock_args(lkb, &args);
4973 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
4975 error = _cancel_lock(r, lkb);
4980 if (error == -DLM_ECANCEL)
4982 /* from validate_unlock_args() */
4983 if (error == -EBUSY)
4988 dlm_unlock_recovery(ls);
4992 /* lkb's that are removed from the waiters list by revert are just left on the
4993 orphans list with the granted orphan locks, to be freed by purge */
4995 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4997 struct dlm_args args;
5001 mutex_lock(&ls->ls_orphans_mutex);
5002 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
5003 mutex_unlock(&ls->ls_orphans_mutex);
5005 set_unlock_args(0, lkb->lkb_ua, &args);
5007 error = cancel_lock(ls, lkb, &args);
5008 if (error == -DLM_ECANCEL)
5013 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
5014 Regardless of what rsb queue the lock is on, it's removed and freed. */
5016 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
5018 struct dlm_args args;
5021 set_unlock_args(DLM_LKF_FORCEUNLOCK, lkb->lkb_ua, &args);
5023 error = unlock_lock(ls, lkb, &args);
5024 if (error == -DLM_EUNLOCK)
5029 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
5030 (which does lock_rsb) due to deadlock with receiving a message that does
5031 lock_rsb followed by dlm_user_add_cb() */
5033 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
5034 struct dlm_user_proc *proc)
5036 struct dlm_lkb *lkb = NULL;
5038 mutex_lock(&ls->ls_clear_proc_locks);
5039 if (list_empty(&proc->locks))
5042 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
5043 list_del_init(&lkb->lkb_ownqueue);
5045 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
5046 lkb->lkb_flags |= DLM_IFL_ORPHAN;
5048 lkb->lkb_flags |= DLM_IFL_DEAD;
5050 mutex_unlock(&ls->ls_clear_proc_locks);
5054 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
5055 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
5056 which we clear here. */
5058 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
5059 list, and no more device_writes should add lkb's to proc->locks list; so we
5060 shouldn't need to take asts_spin or locks_spin here. this assumes that
5061 device reads/writes/closes are serialized -- FIXME: we may need to serialize
5064 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
5066 struct dlm_lkb *lkb, *safe;
5068 dlm_lock_recovery(ls);
5071 lkb = del_proc_lock(ls, proc);
5075 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
5076 orphan_proc_lock(ls, lkb);
5078 unlock_proc_lock(ls, lkb);
5080 /* this removes the reference for the proc->locks list
5081 added by dlm_user_request, it may result in the lkb
5087 mutex_lock(&ls->ls_clear_proc_locks);
5089 /* in-progress unlocks */
5090 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
5091 list_del_init(&lkb->lkb_ownqueue);
5092 lkb->lkb_flags |= DLM_IFL_DEAD;
5096 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
5097 memset(&lkb->lkb_callbacks, 0,
5098 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
5099 list_del_init(&lkb->lkb_cb_list);
5103 mutex_unlock(&ls->ls_clear_proc_locks);
5104 dlm_unlock_recovery(ls);
5107 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
5109 struct dlm_lkb *lkb, *safe;
5113 spin_lock(&proc->locks_spin);
5114 if (!list_empty(&proc->locks)) {
5115 lkb = list_entry(proc->locks.next, struct dlm_lkb,
5117 list_del_init(&lkb->lkb_ownqueue);
5119 spin_unlock(&proc->locks_spin);
5124 lkb->lkb_flags |= DLM_IFL_DEAD;
5125 unlock_proc_lock(ls, lkb);
5126 dlm_put_lkb(lkb); /* ref from proc->locks list */
5129 spin_lock(&proc->locks_spin);
5130 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
5131 list_del_init(&lkb->lkb_ownqueue);
5132 lkb->lkb_flags |= DLM_IFL_DEAD;
5135 spin_unlock(&proc->locks_spin);
5137 spin_lock(&proc->asts_spin);
5138 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
5139 memset(&lkb->lkb_callbacks, 0,
5140 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE);
5141 list_del_init(&lkb->lkb_cb_list);
5144 spin_unlock(&proc->asts_spin);
5147 /* pid of 0 means purge all orphans */
5149 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
5151 struct dlm_lkb *lkb, *safe;
5153 mutex_lock(&ls->ls_orphans_mutex);
5154 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
5155 if (pid && lkb->lkb_ownpid != pid)
5157 unlock_proc_lock(ls, lkb);
5158 list_del_init(&lkb->lkb_ownqueue);
5161 mutex_unlock(&ls->ls_orphans_mutex);
5164 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
5166 struct dlm_message *ms;
5167 struct dlm_mhandle *mh;
5170 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
5171 DLM_MSG_PURGE, &ms, &mh);
5174 ms->m_nodeid = nodeid;
5177 return send_message(mh, ms);
5180 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
5181 int nodeid, int pid)
5185 if (nodeid != dlm_our_nodeid()) {
5186 error = send_purge(ls, nodeid, pid);
5188 dlm_lock_recovery(ls);
5189 if (pid == current->pid)
5190 purge_proc_locks(ls, proc);
5192 do_purge(ls, nodeid, pid);
5193 dlm_unlock_recovery(ls);