1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005-2008 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 "dlm_internal.h"
60 #include <linux/dlm_device.h>
63 #include "requestqueue.h"
67 #include "lockspace.h"
72 #include "lvb_table.h"
76 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
77 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
78 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
79 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
80 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
81 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
82 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
83 static int send_remove(struct dlm_rsb *r);
84 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
85 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
86 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
87 struct dlm_message *ms);
88 static int receive_extralen(struct dlm_message *ms);
89 static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
90 static void del_timeout(struct dlm_lkb *lkb);
93 * Lock compatibilty matrix - thanks Steve
94 * UN = Unlocked state. Not really a state, used as a flag
95 * PD = Padding. Used to make the matrix a nice power of two in size
96 * Other states are the same as the VMS DLM.
97 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
100 static const int __dlm_compat_matrix[8][8] = {
101 /* UN NL CR CW PR PW EX PD */
102 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
103 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
104 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
105 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
106 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
107 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
108 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
109 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
113 * This defines the direction of transfer of LVB data.
114 * Granted mode is the row; requested mode is the column.
115 * Usage: matrix[grmode+1][rqmode+1]
116 * 1 = LVB is returned to the caller
117 * 0 = LVB is written to the resource
118 * -1 = nothing happens to the LVB
121 const int dlm_lvb_operations[8][8] = {
122 /* UN NL CR CW PR PW EX PD*/
123 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
124 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
125 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
126 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
127 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
128 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
129 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
130 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
133 #define modes_compat(gr, rq) \
134 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
136 int dlm_modes_compat(int mode1, int mode2)
138 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
142 * Compatibility matrix for conversions with QUECVT set.
143 * Granted mode is the row; requested mode is the column.
144 * Usage: matrix[grmode+1][rqmode+1]
147 static const int __quecvt_compat_matrix[8][8] = {
148 /* UN NL CR CW PR PW EX PD */
149 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
150 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
151 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
152 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
153 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
154 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
155 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
156 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
159 void dlm_print_lkb(struct dlm_lkb *lkb)
161 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
162 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
163 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
164 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
165 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
168 void dlm_print_rsb(struct dlm_rsb *r)
170 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
171 r->res_nodeid, r->res_flags, r->res_first_lkid,
172 r->res_recover_locks_count, r->res_name);
175 void dlm_dump_rsb(struct dlm_rsb *r)
181 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
182 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
183 printk(KERN_ERR "rsb lookup list\n");
184 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
186 printk(KERN_ERR "rsb grant queue:\n");
187 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
189 printk(KERN_ERR "rsb convert queue:\n");
190 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
192 printk(KERN_ERR "rsb wait queue:\n");
193 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
197 /* Threads cannot use the lockspace while it's being recovered */
199 static inline void dlm_lock_recovery(struct dlm_ls *ls)
201 down_read(&ls->ls_in_recovery);
204 void dlm_unlock_recovery(struct dlm_ls *ls)
206 up_read(&ls->ls_in_recovery);
209 int dlm_lock_recovery_try(struct dlm_ls *ls)
211 return down_read_trylock(&ls->ls_in_recovery);
214 static inline int can_be_queued(struct dlm_lkb *lkb)
216 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
219 static inline int force_blocking_asts(struct dlm_lkb *lkb)
221 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
224 static inline int is_demoted(struct dlm_lkb *lkb)
226 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
229 static inline int is_altmode(struct dlm_lkb *lkb)
231 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
234 static inline int is_granted(struct dlm_lkb *lkb)
236 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
239 static inline int is_remote(struct dlm_rsb *r)
241 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
242 return !!r->res_nodeid;
245 static inline int is_process_copy(struct dlm_lkb *lkb)
247 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
250 static inline int is_master_copy(struct dlm_lkb *lkb)
252 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
253 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
254 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
257 static inline int middle_conversion(struct dlm_lkb *lkb)
259 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
260 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
265 static inline int down_conversion(struct dlm_lkb *lkb)
267 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
270 static inline int is_overlap_unlock(struct dlm_lkb *lkb)
272 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
275 static inline int is_overlap_cancel(struct dlm_lkb *lkb)
277 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
280 static inline int is_overlap(struct dlm_lkb *lkb)
282 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
283 DLM_IFL_OVERLAP_CANCEL));
286 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
288 if (is_master_copy(lkb))
293 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
295 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
296 timeout caused the cancel then return -ETIMEDOUT */
297 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
298 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
302 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
303 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
307 lkb->lkb_lksb->sb_status = rv;
308 lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
310 dlm_add_ast(lkb, AST_COMP);
313 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
316 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
319 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
321 if (is_master_copy(lkb))
322 send_bast(r, lkb, rqmode);
324 lkb->lkb_bastmode = rqmode;
325 dlm_add_ast(lkb, AST_BAST);
330 * Basic operations on rsb's and lkb's
333 static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
337 r = dlm_allocate_rsb(ls, len);
343 memcpy(r->res_name, name, len);
344 mutex_init(&r->res_mutex);
346 INIT_LIST_HEAD(&r->res_lookup);
347 INIT_LIST_HEAD(&r->res_grantqueue);
348 INIT_LIST_HEAD(&r->res_convertqueue);
349 INIT_LIST_HEAD(&r->res_waitqueue);
350 INIT_LIST_HEAD(&r->res_root_list);
351 INIT_LIST_HEAD(&r->res_recover_list);
356 static int search_rsb_list(struct list_head *head, char *name, int len,
357 unsigned int flags, struct dlm_rsb **r_ret)
362 list_for_each_entry(r, head, res_hashchain) {
363 if (len == r->res_length && !memcmp(name, r->res_name, len))
369 if (r->res_nodeid && (flags & R_MASTER))
375 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
376 unsigned int flags, struct dlm_rsb **r_ret)
381 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
383 kref_get(&r->res_ref);
386 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
390 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
392 if (dlm_no_directory(ls))
395 if (r->res_nodeid == -1) {
396 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
397 r->res_first_lkid = 0;
398 } else if (r->res_nodeid > 0) {
399 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
400 r->res_first_lkid = 0;
402 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
403 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
410 static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
411 unsigned int flags, struct dlm_rsb **r_ret)
414 write_lock(&ls->ls_rsbtbl[b].lock);
415 error = _search_rsb(ls, name, len, b, flags, r_ret);
416 write_unlock(&ls->ls_rsbtbl[b].lock);
421 * Find rsb in rsbtbl and potentially create/add one
423 * Delaying the release of rsb's has a similar benefit to applications keeping
424 * NL locks on an rsb, but without the guarantee that the cached master value
425 * will still be valid when the rsb is reused. Apps aren't always smart enough
426 * to keep NL locks on an rsb that they may lock again shortly; this can lead
427 * to excessive master lookups and removals if we don't delay the release.
429 * Searching for an rsb means looking through both the normal list and toss
430 * list. When found on the toss list the rsb is moved to the normal list with
431 * ref count of 1; when found on normal list the ref count is incremented.
434 static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
435 unsigned int flags, struct dlm_rsb **r_ret)
437 struct dlm_rsb *r, *tmp;
438 uint32_t hash, bucket;
441 if (namelen > DLM_RESNAME_MAXLEN)
444 if (dlm_no_directory(ls))
448 hash = jhash(name, namelen, 0);
449 bucket = hash & (ls->ls_rsbtbl_size - 1);
451 error = search_rsb(ls, name, namelen, bucket, flags, &r);
455 if (error == -EBADR && !(flags & R_CREATE))
458 /* the rsb was found but wasn't a master copy */
459 if (error == -ENOTBLK)
463 r = create_rsb(ls, name, namelen);
468 r->res_bucket = bucket;
470 kref_init(&r->res_ref);
472 /* With no directory, the master can be set immediately */
473 if (dlm_no_directory(ls)) {
474 int nodeid = dlm_dir_nodeid(r);
475 if (nodeid == dlm_our_nodeid())
477 r->res_nodeid = nodeid;
480 write_lock(&ls->ls_rsbtbl[bucket].lock);
481 error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
483 write_unlock(&ls->ls_rsbtbl[bucket].lock);
488 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
489 write_unlock(&ls->ls_rsbtbl[bucket].lock);
496 /* This is only called to add a reference when the code already holds
497 a valid reference to the rsb, so there's no need for locking. */
499 static inline void hold_rsb(struct dlm_rsb *r)
501 kref_get(&r->res_ref);
504 void dlm_hold_rsb(struct dlm_rsb *r)
509 static void toss_rsb(struct kref *kref)
511 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
512 struct dlm_ls *ls = r->res_ls;
514 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
515 kref_init(&r->res_ref);
516 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
517 r->res_toss_time = jiffies;
519 dlm_free_lvb(r->res_lvbptr);
520 r->res_lvbptr = NULL;
524 /* When all references to the rsb are gone it's transfered to
525 the tossed list for later disposal. */
527 static void put_rsb(struct dlm_rsb *r)
529 struct dlm_ls *ls = r->res_ls;
530 uint32_t bucket = r->res_bucket;
532 write_lock(&ls->ls_rsbtbl[bucket].lock);
533 kref_put(&r->res_ref, toss_rsb);
534 write_unlock(&ls->ls_rsbtbl[bucket].lock);
537 void dlm_put_rsb(struct dlm_rsb *r)
542 /* See comment for unhold_lkb */
544 static void unhold_rsb(struct dlm_rsb *r)
547 rv = kref_put(&r->res_ref, toss_rsb);
548 DLM_ASSERT(!rv, dlm_dump_rsb(r););
551 static void kill_rsb(struct kref *kref)
553 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
555 /* All work is done after the return from kref_put() so we
556 can release the write_lock before the remove and free. */
558 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
559 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
560 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
561 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
562 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
563 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
566 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
567 The rsb must exist as long as any lkb's for it do. */
569 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
572 lkb->lkb_resource = r;
575 static void detach_lkb(struct dlm_lkb *lkb)
577 if (lkb->lkb_resource) {
578 put_rsb(lkb->lkb_resource);
579 lkb->lkb_resource = NULL;
583 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
585 struct dlm_lkb *lkb, *tmp;
589 lkb = dlm_allocate_lkb(ls);
593 lkb->lkb_nodeid = -1;
594 lkb->lkb_grmode = DLM_LOCK_IV;
595 kref_init(&lkb->lkb_ref);
596 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
597 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
598 INIT_LIST_HEAD(&lkb->lkb_time_list);
600 get_random_bytes(&bucket, sizeof(bucket));
601 bucket &= (ls->ls_lkbtbl_size - 1);
603 write_lock(&ls->ls_lkbtbl[bucket].lock);
605 /* counter can roll over so we must verify lkid is not in use */
608 lkid = (bucket << 16) | ls->ls_lkbtbl[bucket].counter++;
610 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
612 if (tmp->lkb_id != lkid)
620 list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
621 write_unlock(&ls->ls_lkbtbl[bucket].lock);
627 static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
630 uint16_t bucket = (lkid >> 16);
632 list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
633 if (lkb->lkb_id == lkid)
639 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
642 uint16_t bucket = (lkid >> 16);
644 if (bucket >= ls->ls_lkbtbl_size)
647 read_lock(&ls->ls_lkbtbl[bucket].lock);
648 lkb = __find_lkb(ls, lkid);
650 kref_get(&lkb->lkb_ref);
651 read_unlock(&ls->ls_lkbtbl[bucket].lock);
654 return lkb ? 0 : -ENOENT;
657 static void kill_lkb(struct kref *kref)
659 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
661 /* All work is done after the return from kref_put() so we
662 can release the write_lock before the detach_lkb */
664 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
667 /* __put_lkb() is used when an lkb may not have an rsb attached to
668 it so we need to provide the lockspace explicitly */
670 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
672 uint16_t bucket = (lkb->lkb_id >> 16);
674 write_lock(&ls->ls_lkbtbl[bucket].lock);
675 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
676 list_del(&lkb->lkb_idtbl_list);
677 write_unlock(&ls->ls_lkbtbl[bucket].lock);
681 /* for local/process lkbs, lvbptr points to caller's lksb */
682 if (lkb->lkb_lvbptr && is_master_copy(lkb))
683 dlm_free_lvb(lkb->lkb_lvbptr);
687 write_unlock(&ls->ls_lkbtbl[bucket].lock);
692 int dlm_put_lkb(struct dlm_lkb *lkb)
696 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
697 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
699 ls = lkb->lkb_resource->res_ls;
700 return __put_lkb(ls, lkb);
703 /* This is only called to add a reference when the code already holds
704 a valid reference to the lkb, so there's no need for locking. */
706 static inline void hold_lkb(struct dlm_lkb *lkb)
708 kref_get(&lkb->lkb_ref);
711 /* This is called when we need to remove a reference and are certain
712 it's not the last ref. e.g. del_lkb is always called between a
713 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
714 put_lkb would work fine, but would involve unnecessary locking */
716 static inline void unhold_lkb(struct dlm_lkb *lkb)
719 rv = kref_put(&lkb->lkb_ref, kill_lkb);
720 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
723 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
726 struct dlm_lkb *lkb = NULL;
728 list_for_each_entry(lkb, head, lkb_statequeue)
729 if (lkb->lkb_rqmode < mode)
733 list_add_tail(new, head);
735 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
738 /* add/remove lkb to rsb's grant/convert/wait queue */
740 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
742 kref_get(&lkb->lkb_ref);
744 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
746 lkb->lkb_status = status;
749 case DLM_LKSTS_WAITING:
750 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
751 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
753 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
755 case DLM_LKSTS_GRANTED:
756 /* convention says granted locks kept in order of grmode */
757 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
760 case DLM_LKSTS_CONVERT:
761 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
762 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
764 list_add_tail(&lkb->lkb_statequeue,
765 &r->res_convertqueue);
768 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
772 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
775 list_del(&lkb->lkb_statequeue);
779 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
783 add_lkb(r, lkb, sts);
787 static int msg_reply_type(int mstype)
790 case DLM_MSG_REQUEST:
791 return DLM_MSG_REQUEST_REPLY;
792 case DLM_MSG_CONVERT:
793 return DLM_MSG_CONVERT_REPLY;
795 return DLM_MSG_UNLOCK_REPLY;
797 return DLM_MSG_CANCEL_REPLY;
799 return DLM_MSG_LOOKUP_REPLY;
804 /* add/remove lkb from global waiters list of lkb's waiting for
805 a reply from a remote node */
807 static int add_to_waiters(struct dlm_lkb *lkb, int mstype)
809 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
812 mutex_lock(&ls->ls_waiters_mutex);
814 if (is_overlap_unlock(lkb) ||
815 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
820 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
823 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
826 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
832 lkb->lkb_wait_count++;
835 log_debug(ls, "add overlap %x cur %d new %d count %d flags %x",
836 lkb->lkb_id, lkb->lkb_wait_type, mstype,
837 lkb->lkb_wait_count, lkb->lkb_flags);
841 DLM_ASSERT(!lkb->lkb_wait_count,
843 printk("wait_count %d\n", lkb->lkb_wait_count););
845 lkb->lkb_wait_count++;
846 lkb->lkb_wait_type = mstype;
848 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
851 log_error(ls, "add_to_waiters %x error %d flags %x %d %d %s",
852 lkb->lkb_id, error, lkb->lkb_flags, mstype,
853 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
854 mutex_unlock(&ls->ls_waiters_mutex);
858 /* We clear the RESEND flag because we might be taking an lkb off the waiters
859 list as part of process_requestqueue (e.g. a lookup that has an optimized
860 request reply on the requestqueue) between dlm_recover_waiters_pre() which
861 set RESEND and dlm_recover_waiters_post() */
863 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype)
865 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
866 int overlap_done = 0;
868 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
869 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
874 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
875 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
880 /* N.B. type of reply may not always correspond to type of original
881 msg due to lookup->request optimization, verify others? */
883 if (lkb->lkb_wait_type) {
884 lkb->lkb_wait_type = 0;
888 log_error(ls, "remove_from_waiters lkid %x flags %x types %d %d",
889 lkb->lkb_id, lkb->lkb_flags, mstype, lkb->lkb_wait_type);
893 /* the force-unlock/cancel has completed and we haven't recvd a reply
894 to the op that was in progress prior to the unlock/cancel; we
895 give up on any reply to the earlier op. FIXME: not sure when/how
898 if (overlap_done && lkb->lkb_wait_type) {
899 log_error(ls, "remove_from_waiters %x reply %d give up on %d",
900 lkb->lkb_id, mstype, lkb->lkb_wait_type);
901 lkb->lkb_wait_count--;
902 lkb->lkb_wait_type = 0;
905 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
907 lkb->lkb_flags &= ~DLM_IFL_RESEND;
908 lkb->lkb_wait_count--;
909 if (!lkb->lkb_wait_count)
910 list_del_init(&lkb->lkb_wait_reply);
915 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
917 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
920 mutex_lock(&ls->ls_waiters_mutex);
921 error = _remove_from_waiters(lkb, mstype);
922 mutex_unlock(&ls->ls_waiters_mutex);
926 /* Handles situations where we might be processing a "fake" or "stub" reply in
927 which we can't try to take waiters_mutex again. */
929 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
931 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
934 if (ms != &ls->ls_stub_ms)
935 mutex_lock(&ls->ls_waiters_mutex);
936 error = _remove_from_waiters(lkb, ms->m_type);
937 if (ms != &ls->ls_stub_ms)
938 mutex_unlock(&ls->ls_waiters_mutex);
942 static void dir_remove(struct dlm_rsb *r)
946 if (dlm_no_directory(r->res_ls))
949 to_nodeid = dlm_dir_nodeid(r);
950 if (to_nodeid != dlm_our_nodeid())
953 dlm_dir_remove_entry(r->res_ls, to_nodeid,
954 r->res_name, r->res_length);
957 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
958 found since they are in order of newest to oldest? */
960 static int shrink_bucket(struct dlm_ls *ls, int b)
963 int count = 0, found;
967 write_lock(&ls->ls_rsbtbl[b].lock);
968 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
970 if (!time_after_eq(jiffies, r->res_toss_time +
971 dlm_config.ci_toss_secs * HZ))
978 write_unlock(&ls->ls_rsbtbl[b].lock);
982 if (kref_put(&r->res_ref, kill_rsb)) {
983 list_del(&r->res_hashchain);
984 write_unlock(&ls->ls_rsbtbl[b].lock);
991 write_unlock(&ls->ls_rsbtbl[b].lock);
992 log_error(ls, "tossed rsb in use %s", r->res_name);
999 void dlm_scan_rsbs(struct dlm_ls *ls)
1003 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1004 shrink_bucket(ls, i);
1005 if (dlm_locking_stopped(ls))
1011 static void add_timeout(struct dlm_lkb *lkb)
1013 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1015 if (is_master_copy(lkb)) {
1016 lkb->lkb_timestamp = jiffies;
1020 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1021 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1022 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1025 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1030 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1031 mutex_lock(&ls->ls_timeout_mutex);
1033 lkb->lkb_timestamp = jiffies;
1034 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1035 mutex_unlock(&ls->ls_timeout_mutex);
1038 static void del_timeout(struct dlm_lkb *lkb)
1040 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1042 mutex_lock(&ls->ls_timeout_mutex);
1043 if (!list_empty(&lkb->lkb_time_list)) {
1044 list_del_init(&lkb->lkb_time_list);
1047 mutex_unlock(&ls->ls_timeout_mutex);
1050 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1051 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1052 and then lock rsb because of lock ordering in add_timeout. We may need
1053 to specify some special timeout-related bits in the lkb that are just to
1054 be accessed under the timeout_mutex. */
1056 void dlm_scan_timeout(struct dlm_ls *ls)
1059 struct dlm_lkb *lkb;
1060 int do_cancel, do_warn;
1063 if (dlm_locking_stopped(ls))
1068 mutex_lock(&ls->ls_timeout_mutex);
1069 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1071 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1072 time_after_eq(jiffies, lkb->lkb_timestamp +
1073 lkb->lkb_timeout_cs * HZ/100))
1076 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1077 time_after_eq(jiffies, lkb->lkb_timestamp +
1078 dlm_config.ci_timewarn_cs * HZ/100))
1081 if (!do_cancel && !do_warn)
1086 mutex_unlock(&ls->ls_timeout_mutex);
1088 if (!do_cancel && !do_warn)
1091 r = lkb->lkb_resource;
1096 /* clear flag so we only warn once */
1097 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1098 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1100 dlm_timeout_warn(lkb);
1104 log_debug(ls, "timeout cancel %x node %d %s",
1105 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1106 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1107 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1109 _cancel_lock(r, lkb);
1118 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1119 dlm_recoverd before checking/setting ls_recover_begin. */
1121 void dlm_adjust_timeouts(struct dlm_ls *ls)
1123 struct dlm_lkb *lkb;
1124 long adj = jiffies - ls->ls_recover_begin;
1126 ls->ls_recover_begin = 0;
1127 mutex_lock(&ls->ls_timeout_mutex);
1128 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1129 lkb->lkb_timestamp += adj;
1130 mutex_unlock(&ls->ls_timeout_mutex);
1133 /* lkb is master or local copy */
1135 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1137 int b, len = r->res_ls->ls_lvblen;
1139 /* b=1 lvb returned to caller
1140 b=0 lvb written to rsb or invalidated
1143 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1146 if (!lkb->lkb_lvbptr)
1149 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1155 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1156 lkb->lkb_lvbseq = r->res_lvbseq;
1158 } else if (b == 0) {
1159 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1160 rsb_set_flag(r, RSB_VALNOTVALID);
1164 if (!lkb->lkb_lvbptr)
1167 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1171 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1176 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1178 lkb->lkb_lvbseq = r->res_lvbseq;
1179 rsb_clear_flag(r, RSB_VALNOTVALID);
1182 if (rsb_flag(r, RSB_VALNOTVALID))
1183 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1186 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1188 if (lkb->lkb_grmode < DLM_LOCK_PW)
1191 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1192 rsb_set_flag(r, RSB_VALNOTVALID);
1196 if (!lkb->lkb_lvbptr)
1199 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1203 r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1208 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1210 rsb_clear_flag(r, RSB_VALNOTVALID);
1213 /* lkb is process copy (pc) */
1215 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1216 struct dlm_message *ms)
1220 if (!lkb->lkb_lvbptr)
1223 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1226 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1228 int len = receive_extralen(ms);
1229 if (len > DLM_RESNAME_MAXLEN)
1230 len = DLM_RESNAME_MAXLEN;
1231 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1232 lkb->lkb_lvbseq = ms->m_lvbseq;
1236 /* Manipulate lkb's on rsb's convert/granted/waiting queues
1237 remove_lock -- used for unlock, removes lkb from granted
1238 revert_lock -- used for cancel, moves lkb from convert to granted
1239 grant_lock -- used for request and convert, adds lkb to granted or
1240 moves lkb from convert or waiting to granted
1242 Each of these is used for master or local copy lkb's. There is
1243 also a _pc() variation used to make the corresponding change on
1244 a process copy (pc) lkb. */
1246 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1249 lkb->lkb_grmode = DLM_LOCK_IV;
1250 /* this unhold undoes the original ref from create_lkb()
1251 so this leads to the lkb being freed */
1255 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1257 set_lvb_unlock(r, lkb);
1258 _remove_lock(r, lkb);
1261 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1263 _remove_lock(r, lkb);
1266 /* returns: 0 did nothing
1267 1 moved lock to granted
1270 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1274 lkb->lkb_rqmode = DLM_LOCK_IV;
1276 switch (lkb->lkb_status) {
1277 case DLM_LKSTS_GRANTED:
1279 case DLM_LKSTS_CONVERT:
1280 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1283 case DLM_LKSTS_WAITING:
1285 lkb->lkb_grmode = DLM_LOCK_IV;
1286 /* this unhold undoes the original ref from create_lkb()
1287 so this leads to the lkb being freed */
1292 log_print("invalid status for revert %d", lkb->lkb_status);
1297 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1299 return revert_lock(r, lkb);
1302 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1304 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1305 lkb->lkb_grmode = lkb->lkb_rqmode;
1306 if (lkb->lkb_status)
1307 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1309 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1312 lkb->lkb_rqmode = DLM_LOCK_IV;
1315 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1317 set_lvb_lock(r, lkb);
1318 _grant_lock(r, lkb);
1319 lkb->lkb_highbast = 0;
1322 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1323 struct dlm_message *ms)
1325 set_lvb_lock_pc(r, lkb, ms);
1326 _grant_lock(r, lkb);
1329 /* called by grant_pending_locks() which means an async grant message must
1330 be sent to the requesting node in addition to granting the lock if the
1331 lkb belongs to a remote node. */
1333 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1336 if (is_master_copy(lkb))
1339 queue_cast(r, lkb, 0);
1342 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1343 change the granted/requested modes. We're munging things accordingly in
1345 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1347 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1348 compatible with other granted locks */
1350 static void munge_demoted(struct dlm_lkb *lkb, struct dlm_message *ms)
1352 if (ms->m_type != DLM_MSG_CONVERT_REPLY) {
1353 log_print("munge_demoted %x invalid reply type %d",
1354 lkb->lkb_id, ms->m_type);
1358 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
1359 log_print("munge_demoted %x invalid modes gr %d rq %d",
1360 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
1364 lkb->lkb_grmode = DLM_LOCK_NL;
1367 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
1369 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
1370 ms->m_type != DLM_MSG_GRANT) {
1371 log_print("munge_altmode %x invalid reply type %d",
1372 lkb->lkb_id, ms->m_type);
1376 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
1377 lkb->lkb_rqmode = DLM_LOCK_PR;
1378 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
1379 lkb->lkb_rqmode = DLM_LOCK_CW;
1381 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
1386 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1388 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1390 if (lkb->lkb_id == first->lkb_id)
1396 /* Check if the given lkb conflicts with another lkb on the queue. */
1398 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1400 struct dlm_lkb *this;
1402 list_for_each_entry(this, head, lkb_statequeue) {
1405 if (!modes_compat(this, lkb))
1412 * "A conversion deadlock arises with a pair of lock requests in the converting
1413 * queue for one resource. The granted mode of each lock blocks the requested
1414 * mode of the other lock."
1416 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1417 * convert queue from being granted, then deadlk/demote lkb.
1420 * Granted Queue: empty
1421 * Convert Queue: NL->EX (first lock)
1422 * PR->EX (second lock)
1424 * The first lock can't be granted because of the granted mode of the second
1425 * lock and the second lock can't be granted because it's not first in the
1426 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1427 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1428 * flag set and return DEMOTED in the lksb flags.
1430 * Originally, this function detected conv-deadlk in a more limited scope:
1431 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1432 * - if lkb1 was the first entry in the queue (not just earlier), and was
1433 * blocked by the granted mode of lkb2, and there was nothing on the
1434 * granted queue preventing lkb1 from being granted immediately, i.e.
1435 * lkb2 was the only thing preventing lkb1 from being granted.
1437 * That second condition meant we'd only say there was conv-deadlk if
1438 * resolving it (by demotion) would lead to the first lock on the convert
1439 * queue being granted right away. It allowed conversion deadlocks to exist
1440 * between locks on the convert queue while they couldn't be granted anyway.
1442 * Now, we detect and take action on conversion deadlocks immediately when
1443 * they're created, even if they may not be immediately consequential. If
1444 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1445 * mode that would prevent lkb1's conversion from being granted, we do a
1446 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1447 * I think this means that the lkb_is_ahead condition below should always
1448 * be zero, i.e. there will never be conv-deadlk between two locks that are
1449 * both already on the convert queue.
1452 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
1454 struct dlm_lkb *lkb1;
1455 int lkb_is_ahead = 0;
1457 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
1463 if (!lkb_is_ahead) {
1464 if (!modes_compat(lkb2, lkb1))
1467 if (!modes_compat(lkb2, lkb1) &&
1468 !modes_compat(lkb1, lkb2))
1476 * Return 1 if the lock can be granted, 0 otherwise.
1477 * Also detect and resolve conversion deadlocks.
1479 * lkb is the lock to be granted
1481 * now is 1 if the function is being called in the context of the
1482 * immediate request, it is 0 if called later, after the lock has been
1485 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1488 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1490 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1493 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1494 * a new request for a NL mode lock being blocked.
1496 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1497 * request, then it would be granted. In essence, the use of this flag
1498 * tells the Lock Manager to expedite theis request by not considering
1499 * what may be in the CONVERTING or WAITING queues... As of this
1500 * writing, the EXPEDITE flag can be used only with new requests for NL
1501 * mode locks. This flag is not valid for conversion requests.
1503 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1504 * conversion or used with a non-NL requested mode. We also know an
1505 * EXPEDITE request is always granted immediately, so now must always
1506 * be 1. The full condition to grant an expedite request: (now &&
1507 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1508 * therefore be shortened to just checking the flag.
1511 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1515 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1516 * added to the remaining conditions.
1519 if (queue_conflict(&r->res_grantqueue, lkb))
1523 * 6-3: By default, a conversion request is immediately granted if the
1524 * requested mode is compatible with the modes of all other granted
1528 if (queue_conflict(&r->res_convertqueue, lkb))
1532 * 6-5: But the default algorithm for deciding whether to grant or
1533 * queue conversion requests does not by itself guarantee that such
1534 * requests are serviced on a "first come first serve" basis. This, in
1535 * turn, can lead to a phenomenon known as "indefinate postponement".
1537 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1538 * the system service employed to request a lock conversion. This flag
1539 * forces certain conversion requests to be queued, even if they are
1540 * compatible with the granted modes of other locks on the same
1541 * resource. Thus, the use of this flag results in conversion requests
1542 * being ordered on a "first come first servce" basis.
1544 * DCT: This condition is all about new conversions being able to occur
1545 * "in place" while the lock remains on the granted queue (assuming
1546 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1547 * doesn't _have_ to go onto the convert queue where it's processed in
1548 * order. The "now" variable is necessary to distinguish converts
1549 * being received and processed for the first time now, because once a
1550 * convert is moved to the conversion queue the condition below applies
1551 * requiring fifo granting.
1554 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1558 * The NOORDER flag is set to avoid the standard vms rules on grant
1562 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1566 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1567 * granted until all other conversion requests ahead of it are granted
1571 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1575 * 6-4: By default, a new request is immediately granted only if all
1576 * three of the following conditions are satisfied when the request is
1578 * - The queue of ungranted conversion requests for the resource is
1580 * - The queue of ungranted new requests for the resource is empty.
1581 * - The mode of the new request is compatible with the most
1582 * restrictive mode of all granted locks on the resource.
1585 if (now && !conv && list_empty(&r->res_convertqueue) &&
1586 list_empty(&r->res_waitqueue))
1590 * 6-4: Once a lock request is in the queue of ungranted new requests,
1591 * it cannot be granted until the queue of ungranted conversion
1592 * requests is empty, all ungranted new requests ahead of it are
1593 * granted and/or canceled, and it is compatible with the granted mode
1594 * of the most restrictive lock granted on the resource.
1597 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1598 first_in_list(lkb, &r->res_waitqueue))
1604 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
1608 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1609 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
1614 rv = _can_be_granted(r, lkb, now);
1619 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1620 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1621 * cancels one of the locks.
1624 if (is_convert && can_be_queued(lkb) &&
1625 conversion_deadlock_detect(r, lkb)) {
1626 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
1627 lkb->lkb_grmode = DLM_LOCK_NL;
1628 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1629 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1633 log_print("can_be_granted deadlock %x now %d",
1642 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1643 * to grant a request in a mode other than the normal rqmode. It's a
1644 * simple way to provide a big optimization to applications that can
1648 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
1650 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
1654 lkb->lkb_rqmode = alt;
1655 rv = _can_be_granted(r, lkb, now);
1657 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1659 lkb->lkb_rqmode = rqmode;
1665 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1666 for locks pending on the convert list. Once verified (watch for these
1667 log_prints), we should be able to just call _can_be_granted() and not
1668 bother with the demote/deadlk cases here (and there's no easy way to deal
1669 with a deadlk here, we'd have to generate something like grant_lock with
1670 the deadlk error.) */
1672 /* Returns the highest requested mode of all blocked conversions; sets
1673 cw if there's a blocked conversion to DLM_LOCK_CW. */
1675 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
1677 struct dlm_lkb *lkb, *s;
1678 int hi, demoted, quit, grant_restart, demote_restart;
1687 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1688 demoted = is_demoted(lkb);
1691 if (can_be_granted(r, lkb, 0, &deadlk)) {
1692 grant_lock_pending(r, lkb);
1697 if (!demoted && is_demoted(lkb)) {
1698 log_print("WARN: pending demoted %x node %d %s",
1699 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1705 log_print("WARN: pending deadlock %x node %d %s",
1706 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1711 hi = max_t(int, lkb->lkb_rqmode, hi);
1713 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
1719 if (demote_restart && !quit) {
1724 return max_t(int, high, hi);
1727 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
1729 struct dlm_lkb *lkb, *s;
1731 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1732 if (can_be_granted(r, lkb, 0, NULL))
1733 grant_lock_pending(r, lkb);
1735 high = max_t(int, lkb->lkb_rqmode, high);
1736 if (lkb->lkb_rqmode == DLM_LOCK_CW)
1744 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
1745 on either the convert or waiting queue.
1746 high is the largest rqmode of all locks blocked on the convert or
1749 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
1751 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
1752 if (gr->lkb_highbast < DLM_LOCK_EX)
1757 if (gr->lkb_highbast < high &&
1758 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
1763 static void grant_pending_locks(struct dlm_rsb *r)
1765 struct dlm_lkb *lkb, *s;
1766 int high = DLM_LOCK_IV;
1769 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
1771 high = grant_pending_convert(r, high, &cw);
1772 high = grant_pending_wait(r, high, &cw);
1774 if (high == DLM_LOCK_IV)
1778 * If there are locks left on the wait/convert queue then send blocking
1779 * ASTs to granted locks based on the largest requested mode (high)
1783 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1784 if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
1785 if (cw && high == DLM_LOCK_PR)
1786 queue_bast(r, lkb, DLM_LOCK_CW);
1788 queue_bast(r, lkb, high);
1789 lkb->lkb_highbast = high;
1794 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
1796 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
1797 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
1798 if (gr->lkb_highbast < DLM_LOCK_EX)
1803 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
1808 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1809 struct dlm_lkb *lkb)
1813 list_for_each_entry(gr, head, lkb_statequeue) {
1814 if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
1815 queue_bast(r, gr, lkb->lkb_rqmode);
1816 gr->lkb_highbast = lkb->lkb_rqmode;
1821 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1823 send_bast_queue(r, &r->res_grantqueue, lkb);
1826 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1828 send_bast_queue(r, &r->res_grantqueue, lkb);
1829 send_bast_queue(r, &r->res_convertqueue, lkb);
1832 /* set_master(r, lkb) -- set the master nodeid of a resource
1834 The purpose of this function is to set the nodeid field in the given
1835 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1836 known, it can just be copied to the lkb and the function will return
1837 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1838 before it can be copied to the lkb.
1840 When the rsb nodeid is being looked up remotely, the initial lkb
1841 causing the lookup is kept on the ls_waiters list waiting for the
1842 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1843 on the rsb's res_lookup list until the master is verified.
1846 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1847 1: the rsb master is not available and the lkb has been placed on
1851 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1853 struct dlm_ls *ls = r->res_ls;
1854 int i, error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
1856 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1857 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1858 r->res_first_lkid = lkb->lkb_id;
1859 lkb->lkb_nodeid = r->res_nodeid;
1863 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1864 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1868 if (r->res_nodeid == 0) {
1869 lkb->lkb_nodeid = 0;
1873 if (r->res_nodeid > 0) {
1874 lkb->lkb_nodeid = r->res_nodeid;
1878 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
1880 dir_nodeid = dlm_dir_nodeid(r);
1882 if (dir_nodeid != our_nodeid) {
1883 r->res_first_lkid = lkb->lkb_id;
1884 send_lookup(r, lkb);
1888 for (i = 0; i < 2; i++) {
1889 /* It's possible for dlm_scand to remove an old rsb for
1890 this same resource from the toss list, us to create
1891 a new one, look up the master locally, and find it
1892 already exists just before dlm_scand does the
1893 dir_remove() on the previous rsb. */
1895 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1896 r->res_length, &ret_nodeid);
1899 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1902 if (error && error != -EEXIST)
1905 if (ret_nodeid == our_nodeid) {
1906 r->res_first_lkid = 0;
1908 lkb->lkb_nodeid = 0;
1910 r->res_first_lkid = lkb->lkb_id;
1911 r->res_nodeid = ret_nodeid;
1912 lkb->lkb_nodeid = ret_nodeid;
1917 static void process_lookup_list(struct dlm_rsb *r)
1919 struct dlm_lkb *lkb, *safe;
1921 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
1922 list_del_init(&lkb->lkb_rsb_lookup);
1923 _request_lock(r, lkb);
1928 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1930 static void confirm_master(struct dlm_rsb *r, int error)
1932 struct dlm_lkb *lkb;
1934 if (!r->res_first_lkid)
1940 r->res_first_lkid = 0;
1941 process_lookup_list(r);
1947 /* the remote request failed and won't be retried (it was
1948 a NOQUEUE, or has been canceled/unlocked); make a waiting
1949 lkb the first_lkid */
1951 r->res_first_lkid = 0;
1953 if (!list_empty(&r->res_lookup)) {
1954 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1956 list_del_init(&lkb->lkb_rsb_lookup);
1957 r->res_first_lkid = lkb->lkb_id;
1958 _request_lock(r, lkb);
1964 log_error(r->res_ls, "confirm_master unknown error %d", error);
1968 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
1969 int namelen, unsigned long timeout_cs, void *ast,
1970 void *astarg, void *bast, struct dlm_args *args)
1974 /* check for invalid arg usage */
1976 if (mode < 0 || mode > DLM_LOCK_EX)
1979 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1982 if (flags & DLM_LKF_CANCEL)
1985 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1988 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1991 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1994 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1997 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2000 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2003 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2009 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2012 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2015 /* these args will be copied to the lkb in validate_lock_args,
2016 it cannot be done now because when converting locks, fields in
2017 an active lkb cannot be modified before locking the rsb */
2019 args->flags = flags;
2020 args->astaddr = ast;
2021 args->astparam = (long) astarg;
2022 args->bastaddr = bast;
2023 args->timeout = timeout_cs;
2031 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2033 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2034 DLM_LKF_FORCEUNLOCK))
2037 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2040 args->flags = flags;
2041 args->astparam = (long) astarg;
2045 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2046 struct dlm_args *args)
2050 if (args->flags & DLM_LKF_CONVERT) {
2051 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2054 if (args->flags & DLM_LKF_QUECVT &&
2055 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2059 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2062 if (lkb->lkb_wait_type)
2065 if (is_overlap(lkb))
2069 lkb->lkb_exflags = args->flags;
2070 lkb->lkb_sbflags = 0;
2071 lkb->lkb_astaddr = args->astaddr;
2072 lkb->lkb_astparam = args->astparam;
2073 lkb->lkb_bastaddr = args->bastaddr;
2074 lkb->lkb_rqmode = args->mode;
2075 lkb->lkb_lksb = args->lksb;
2076 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2077 lkb->lkb_ownpid = (int) current->pid;
2078 lkb->lkb_timeout_cs = args->timeout;
2084 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2087 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2088 because there may be a lookup in progress and it's valid to do
2089 cancel/unlockf on it */
2091 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2093 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2096 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2097 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2102 /* an lkb may still exist even though the lock is EOL'ed due to a
2103 cancel, unlock or failed noqueue request; an app can't use these
2104 locks; return same error as if the lkid had not been found at all */
2106 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2107 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2112 /* an lkb may be waiting for an rsb lookup to complete where the
2113 lookup was initiated by another lock */
2115 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2116 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2117 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2118 list_del_init(&lkb->lkb_rsb_lookup);
2119 queue_cast(lkb->lkb_resource, lkb,
2120 args->flags & DLM_LKF_CANCEL ?
2121 -DLM_ECANCEL : -DLM_EUNLOCK);
2122 unhold_lkb(lkb); /* undoes create_lkb() */
2124 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2129 /* cancel not allowed with another cancel/unlock in progress */
2131 if (args->flags & DLM_LKF_CANCEL) {
2132 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2135 if (is_overlap(lkb))
2138 /* don't let scand try to do a cancel */
2141 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2142 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2147 switch (lkb->lkb_wait_type) {
2148 case DLM_MSG_LOOKUP:
2149 case DLM_MSG_REQUEST:
2150 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2153 case DLM_MSG_UNLOCK:
2154 case DLM_MSG_CANCEL:
2157 /* add_to_waiters() will set OVERLAP_CANCEL */
2161 /* do we need to allow a force-unlock if there's a normal unlock
2162 already in progress? in what conditions could the normal unlock
2163 fail such that we'd want to send a force-unlock to be sure? */
2165 if (args->flags & DLM_LKF_FORCEUNLOCK) {
2166 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2169 if (is_overlap_unlock(lkb))
2172 /* don't let scand try to do a cancel */
2175 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2176 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2181 switch (lkb->lkb_wait_type) {
2182 case DLM_MSG_LOOKUP:
2183 case DLM_MSG_REQUEST:
2184 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2187 case DLM_MSG_UNLOCK:
2190 /* add_to_waiters() will set OVERLAP_UNLOCK */
2194 /* normal unlock not allowed if there's any op in progress */
2196 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
2200 /* an overlapping op shouldn't blow away exflags from other op */
2201 lkb->lkb_exflags |= args->flags;
2202 lkb->lkb_sbflags = 0;
2203 lkb->lkb_astparam = args->astparam;
2207 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
2208 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
2209 args->flags, lkb->lkb_wait_type,
2210 lkb->lkb_resource->res_name);
2215 * Four stage 4 varieties:
2216 * do_request(), do_convert(), do_unlock(), do_cancel()
2217 * These are called on the master node for the given lock and
2218 * from the central locking logic.
2221 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2225 if (can_be_granted(r, lkb, 1, NULL)) {
2227 queue_cast(r, lkb, 0);
2231 if (can_be_queued(lkb)) {
2232 error = -EINPROGRESS;
2233 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2234 send_blocking_asts(r, lkb);
2240 if (force_blocking_asts(lkb))
2241 send_blocking_asts_all(r, lkb);
2242 queue_cast(r, lkb, -EAGAIN);
2248 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2253 /* changing an existing lock may allow others to be granted */
2255 if (can_be_granted(r, lkb, 1, &deadlk)) {
2257 queue_cast(r, lkb, 0);
2258 grant_pending_locks(r);
2262 /* can_be_granted() detected that this lock would block in a conversion
2263 deadlock, so we leave it on the granted queue and return EDEADLK in
2264 the ast for the convert. */
2267 /* it's left on the granted queue */
2268 log_debug(r->res_ls, "deadlock %x node %d sts%d g%d r%d %s",
2269 lkb->lkb_id, lkb->lkb_nodeid, lkb->lkb_status,
2270 lkb->lkb_grmode, lkb->lkb_rqmode, r->res_name);
2271 revert_lock(r, lkb);
2272 queue_cast(r, lkb, -EDEADLK);
2277 /* is_demoted() means the can_be_granted() above set the grmode
2278 to NL, and left us on the granted queue. This auto-demotion
2279 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2280 now grantable. We have to try to grant other converting locks
2281 before we try again to grant this one. */
2283 if (is_demoted(lkb)) {
2284 grant_pending_convert(r, DLM_LOCK_IV, NULL);
2285 if (_can_be_granted(r, lkb, 1)) {
2287 queue_cast(r, lkb, 0);
2288 grant_pending_locks(r);
2291 /* else fall through and move to convert queue */
2294 if (can_be_queued(lkb)) {
2295 error = -EINPROGRESS;
2297 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2298 send_blocking_asts(r, lkb);
2304 if (force_blocking_asts(lkb))
2305 send_blocking_asts_all(r, lkb);
2306 queue_cast(r, lkb, -EAGAIN);
2312 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2314 remove_lock(r, lkb);
2315 queue_cast(r, lkb, -DLM_EUNLOCK);
2316 grant_pending_locks(r);
2317 return -DLM_EUNLOCK;
2320 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
2322 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2326 error = revert_lock(r, lkb);
2328 queue_cast(r, lkb, -DLM_ECANCEL);
2329 grant_pending_locks(r);
2330 return -DLM_ECANCEL;
2336 * Four stage 3 varieties:
2337 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2340 /* add a new lkb to a possibly new rsb, called by requesting process */
2342 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2346 /* set_master: sets lkb nodeid from r */
2348 error = set_master(r, lkb);
2357 /* receive_request() calls do_request() on remote node */
2358 error = send_request(r, lkb);
2360 error = do_request(r, lkb);
2365 /* change some property of an existing lkb, e.g. mode */
2367 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2372 /* receive_convert() calls do_convert() on remote node */
2373 error = send_convert(r, lkb);
2375 error = do_convert(r, lkb);
2380 /* remove an existing lkb from the granted queue */
2382 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2387 /* receive_unlock() calls do_unlock() on remote node */
2388 error = send_unlock(r, lkb);
2390 error = do_unlock(r, lkb);
2395 /* remove an existing lkb from the convert or wait queue */
2397 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2402 /* receive_cancel() calls do_cancel() on remote node */
2403 error = send_cancel(r, lkb);
2405 error = do_cancel(r, lkb);
2411 * Four stage 2 varieties:
2412 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2415 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
2416 int len, struct dlm_args *args)
2421 error = validate_lock_args(ls, lkb, args);
2425 error = find_rsb(ls, name, len, R_CREATE, &r);
2432 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
2434 error = _request_lock(r, lkb);
2443 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2444 struct dlm_args *args)
2449 r = lkb->lkb_resource;
2454 error = validate_lock_args(ls, lkb, args);
2458 error = _convert_lock(r, lkb);
2465 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2466 struct dlm_args *args)
2471 r = lkb->lkb_resource;
2476 error = validate_unlock_args(lkb, args);
2480 error = _unlock_lock(r, lkb);
2487 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2488 struct dlm_args *args)
2493 r = lkb->lkb_resource;
2498 error = validate_unlock_args(lkb, args);
2502 error = _cancel_lock(r, lkb);
2510 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2513 int dlm_lock(dlm_lockspace_t *lockspace,
2515 struct dlm_lksb *lksb,
2518 unsigned int namelen,
2519 uint32_t parent_lkid,
2520 void (*ast) (void *astarg),
2522 void (*bast) (void *astarg, int mode))
2525 struct dlm_lkb *lkb;
2526 struct dlm_args args;
2527 int error, convert = flags & DLM_LKF_CONVERT;
2529 ls = dlm_find_lockspace_local(lockspace);
2533 dlm_lock_recovery(ls);
2536 error = find_lkb(ls, lksb->sb_lkid, &lkb);
2538 error = create_lkb(ls, &lkb);
2543 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
2544 astarg, bast, &args);
2549 error = convert_lock(ls, lkb, &args);
2551 error = request_lock(ls, lkb, name, namelen, &args);
2553 if (error == -EINPROGRESS)
2556 if (convert || error)
2558 if (error == -EAGAIN || error == -EDEADLK)
2561 dlm_unlock_recovery(ls);
2562 dlm_put_lockspace(ls);
2566 int dlm_unlock(dlm_lockspace_t *lockspace,
2569 struct dlm_lksb *lksb,
2573 struct dlm_lkb *lkb;
2574 struct dlm_args args;
2577 ls = dlm_find_lockspace_local(lockspace);
2581 dlm_lock_recovery(ls);
2583 error = find_lkb(ls, lkid, &lkb);
2587 error = set_unlock_args(flags, astarg, &args);
2591 if (flags & DLM_LKF_CANCEL)
2592 error = cancel_lock(ls, lkb, &args);
2594 error = unlock_lock(ls, lkb, &args);
2596 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2598 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
2603 dlm_unlock_recovery(ls);
2604 dlm_put_lockspace(ls);
2609 * send/receive routines for remote operations and replies
2613 * send_request receive_request
2614 * send_convert receive_convert
2615 * send_unlock receive_unlock
2616 * send_cancel receive_cancel
2617 * send_grant receive_grant
2618 * send_bast receive_bast
2619 * send_lookup receive_lookup
2620 * send_remove receive_remove
2623 * receive_request_reply send_request_reply
2624 * receive_convert_reply send_convert_reply
2625 * receive_unlock_reply send_unlock_reply
2626 * receive_cancel_reply send_cancel_reply
2627 * receive_lookup_reply send_lookup_reply
2630 static int _create_message(struct dlm_ls *ls, int mb_len,
2631 int to_nodeid, int mstype,
2632 struct dlm_message **ms_ret,
2633 struct dlm_mhandle **mh_ret)
2635 struct dlm_message *ms;
2636 struct dlm_mhandle *mh;
2639 /* get_buffer gives us a message handle (mh) that we need to
2640 pass into lowcomms_commit and a message buffer (mb) that we
2641 write our data into */
2643 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
2647 memset(mb, 0, mb_len);
2649 ms = (struct dlm_message *) mb;
2651 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2652 ms->m_header.h_lockspace = ls->ls_global_id;
2653 ms->m_header.h_nodeid = dlm_our_nodeid();
2654 ms->m_header.h_length = mb_len;
2655 ms->m_header.h_cmd = DLM_MSG;
2657 ms->m_type = mstype;
2664 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2665 int to_nodeid, int mstype,
2666 struct dlm_message **ms_ret,
2667 struct dlm_mhandle **mh_ret)
2669 int mb_len = sizeof(struct dlm_message);
2672 case DLM_MSG_REQUEST:
2673 case DLM_MSG_LOOKUP:
2674 case DLM_MSG_REMOVE:
2675 mb_len += r->res_length;
2677 case DLM_MSG_CONVERT:
2678 case DLM_MSG_UNLOCK:
2679 case DLM_MSG_REQUEST_REPLY:
2680 case DLM_MSG_CONVERT_REPLY:
2682 if (lkb && lkb->lkb_lvbptr)
2683 mb_len += r->res_ls->ls_lvblen;
2687 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
2691 /* further lowcomms enhancements or alternate implementations may make
2692 the return value from this function useful at some point */
2694 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2696 dlm_message_out(ms);
2697 dlm_lowcomms_commit_buffer(mh);
2701 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2702 struct dlm_message *ms)
2704 ms->m_nodeid = lkb->lkb_nodeid;
2705 ms->m_pid = lkb->lkb_ownpid;
2706 ms->m_lkid = lkb->lkb_id;
2707 ms->m_remid = lkb->lkb_remid;
2708 ms->m_exflags = lkb->lkb_exflags;
2709 ms->m_sbflags = lkb->lkb_sbflags;
2710 ms->m_flags = lkb->lkb_flags;
2711 ms->m_lvbseq = lkb->lkb_lvbseq;
2712 ms->m_status = lkb->lkb_status;
2713 ms->m_grmode = lkb->lkb_grmode;
2714 ms->m_rqmode = lkb->lkb_rqmode;
2715 ms->m_hash = r->res_hash;
2717 /* m_result and m_bastmode are set from function args,
2718 not from lkb fields */
2720 if (lkb->lkb_bastaddr)
2721 ms->m_asts |= AST_BAST;
2722 if (lkb->lkb_astaddr)
2723 ms->m_asts |= AST_COMP;
2725 /* compare with switch in create_message; send_remove() doesn't
2728 switch (ms->m_type) {
2729 case DLM_MSG_REQUEST:
2730 case DLM_MSG_LOOKUP:
2731 memcpy(ms->m_extra, r->res_name, r->res_length);
2733 case DLM_MSG_CONVERT:
2734 case DLM_MSG_UNLOCK:
2735 case DLM_MSG_REQUEST_REPLY:
2736 case DLM_MSG_CONVERT_REPLY:
2738 if (!lkb->lkb_lvbptr)
2740 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2745 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2747 struct dlm_message *ms;
2748 struct dlm_mhandle *mh;
2749 int to_nodeid, error;
2751 error = add_to_waiters(lkb, mstype);
2755 to_nodeid = r->res_nodeid;
2757 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2761 send_args(r, lkb, ms);
2763 error = send_message(mh, ms);
2769 remove_from_waiters(lkb, msg_reply_type(mstype));
2773 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2775 return send_common(r, lkb, DLM_MSG_REQUEST);
2778 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2782 error = send_common(r, lkb, DLM_MSG_CONVERT);
2784 /* down conversions go without a reply from the master */
2785 if (!error && down_conversion(lkb)) {
2786 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
2787 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
2788 r->res_ls->ls_stub_ms.m_result = 0;
2789 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
2790 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2796 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2797 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2798 that the master is still correct. */
2800 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2802 return send_common(r, lkb, DLM_MSG_UNLOCK);
2805 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2807 return send_common(r, lkb, DLM_MSG_CANCEL);
2810 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2812 struct dlm_message *ms;
2813 struct dlm_mhandle *mh;
2814 int to_nodeid, error;
2816 to_nodeid = lkb->lkb_nodeid;
2818 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2822 send_args(r, lkb, ms);
2826 error = send_message(mh, ms);
2831 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2833 struct dlm_message *ms;
2834 struct dlm_mhandle *mh;
2835 int to_nodeid, error;
2837 to_nodeid = lkb->lkb_nodeid;
2839 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2843 send_args(r, lkb, ms);
2845 ms->m_bastmode = mode;
2847 error = send_message(mh, ms);
2852 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2854 struct dlm_message *ms;
2855 struct dlm_mhandle *mh;
2856 int to_nodeid, error;
2858 error = add_to_waiters(lkb, DLM_MSG_LOOKUP);
2862 to_nodeid = dlm_dir_nodeid(r);
2864 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2868 send_args(r, lkb, ms);
2870 error = send_message(mh, ms);
2876 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
2880 static int send_remove(struct dlm_rsb *r)
2882 struct dlm_message *ms;
2883 struct dlm_mhandle *mh;
2884 int to_nodeid, error;
2886 to_nodeid = dlm_dir_nodeid(r);
2888 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2892 memcpy(ms->m_extra, r->res_name, r->res_length);
2893 ms->m_hash = r->res_hash;
2895 error = send_message(mh, ms);
2900 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2903 struct dlm_message *ms;
2904 struct dlm_mhandle *mh;
2905 int to_nodeid, error;
2907 to_nodeid = lkb->lkb_nodeid;
2909 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2913 send_args(r, lkb, ms);
2917 error = send_message(mh, ms);
2922 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2924 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2927 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2929 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2932 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2934 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2937 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2939 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2942 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2943 int ret_nodeid, int rv)
2945 struct dlm_rsb *r = &ls->ls_stub_rsb;
2946 struct dlm_message *ms;
2947 struct dlm_mhandle *mh;
2948 int error, nodeid = ms_in->m_header.h_nodeid;
2950 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2954 ms->m_lkid = ms_in->m_lkid;
2956 ms->m_nodeid = ret_nodeid;
2958 error = send_message(mh, ms);
2963 /* which args we save from a received message depends heavily on the type
2964 of message, unlike the send side where we can safely send everything about
2965 the lkb for any type of message */
2967 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2969 lkb->lkb_exflags = ms->m_exflags;
2970 lkb->lkb_sbflags = ms->m_sbflags;
2971 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2972 (ms->m_flags & 0x0000FFFF);
2975 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2977 lkb->lkb_sbflags = ms->m_sbflags;
2978 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2979 (ms->m_flags & 0x0000FFFF);
2982 static int receive_extralen(struct dlm_message *ms)
2984 return (ms->m_header.h_length - sizeof(struct dlm_message));
2987 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2988 struct dlm_message *ms)
2992 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2993 if (!lkb->lkb_lvbptr)
2994 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
2995 if (!lkb->lkb_lvbptr)
2997 len = receive_extralen(ms);
2998 if (len > DLM_RESNAME_MAXLEN)
2999 len = DLM_RESNAME_MAXLEN;
3000 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3005 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3006 struct dlm_message *ms)
3008 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3009 lkb->lkb_ownpid = ms->m_pid;
3010 lkb->lkb_remid = ms->m_lkid;
3011 lkb->lkb_grmode = DLM_LOCK_IV;
3012 lkb->lkb_rqmode = ms->m_rqmode;
3013 lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
3014 lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
3016 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3017 /* lkb was just created so there won't be an lvb yet */
3018 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3019 if (!lkb->lkb_lvbptr)
3026 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3027 struct dlm_message *ms)
3029 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3032 if (receive_lvb(ls, lkb, ms))
3035 lkb->lkb_rqmode = ms->m_rqmode;
3036 lkb->lkb_lvbseq = ms->m_lvbseq;
3041 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3042 struct dlm_message *ms)
3044 if (receive_lvb(ls, lkb, ms))
3049 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3050 uses to send a reply and that the remote end uses to process the reply. */
3052 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3054 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3055 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3056 lkb->lkb_remid = ms->m_lkid;
3059 /* This is called after the rsb is locked so that we can safely inspect
3060 fields in the lkb. */
3062 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3064 int from = ms->m_header.h_nodeid;
3067 switch (ms->m_type) {
3068 case DLM_MSG_CONVERT:
3069 case DLM_MSG_UNLOCK:
3070 case DLM_MSG_CANCEL:
3071 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3075 case DLM_MSG_CONVERT_REPLY:
3076 case DLM_MSG_UNLOCK_REPLY:
3077 case DLM_MSG_CANCEL_REPLY:
3080 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3084 case DLM_MSG_REQUEST_REPLY:
3085 if (!is_process_copy(lkb))
3087 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3096 log_error(lkb->lkb_resource->res_ls,
3097 "ignore invalid message %d from %d %x %x %x %d",
3098 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
3099 lkb->lkb_flags, lkb->lkb_nodeid);
3103 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3105 struct dlm_lkb *lkb;
3109 error = create_lkb(ls, &lkb);
3113 receive_flags(lkb, ms);
3114 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3115 error = receive_request_args(ls, lkb, ms);
3121 namelen = receive_extralen(ms);
3123 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
3132 error = do_request(r, lkb);
3133 send_request_reply(r, lkb, error);
3138 if (error == -EINPROGRESS)
3145 setup_stub_lkb(ls, ms);
3146 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3149 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3151 struct dlm_lkb *lkb;
3153 int error, reply = 1;
3155 error = find_lkb(ls, ms->m_remid, &lkb);
3159 r = lkb->lkb_resource;
3164 error = validate_message(lkb, ms);
3168 receive_flags(lkb, ms);
3169 error = receive_convert_args(ls, lkb, ms);
3172 reply = !down_conversion(lkb);
3174 error = do_convert(r, lkb);
3177 send_convert_reply(r, lkb, error);
3185 setup_stub_lkb(ls, ms);
3186 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3189 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3191 struct dlm_lkb *lkb;
3195 error = find_lkb(ls, ms->m_remid, &lkb);
3199 r = lkb->lkb_resource;
3204 error = validate_message(lkb, ms);
3208 receive_flags(lkb, ms);
3209 error = receive_unlock_args(ls, lkb, ms);
3213 error = do_unlock(r, lkb);
3215 send_unlock_reply(r, lkb, error);
3223 setup_stub_lkb(ls, ms);
3224 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3227 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3229 struct dlm_lkb *lkb;
3233 error = find_lkb(ls, ms->m_remid, &lkb);
3237 receive_flags(lkb, ms);
3239 r = lkb->lkb_resource;
3244 error = validate_message(lkb, ms);
3248 error = do_cancel(r, lkb);
3249 send_cancel_reply(r, lkb, error);
3257 setup_stub_lkb(ls, ms);
3258 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3261 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
3263 struct dlm_lkb *lkb;
3267 error = find_lkb(ls, ms->m_remid, &lkb);
3269 log_debug(ls, "receive_grant from %d no lkb %x",
3270 ms->m_header.h_nodeid, ms->m_remid);
3274 r = lkb->lkb_resource;
3279 error = validate_message(lkb, ms);
3283 receive_flags_reply(lkb, ms);
3284 if (is_altmode(lkb))
3285 munge_altmode(lkb, ms);
3286 grant_lock_pc(r, lkb, ms);
3287 queue_cast(r, lkb, 0);
3294 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
3296 struct dlm_lkb *lkb;
3300 error = find_lkb(ls, ms->m_remid, &lkb);
3302 log_debug(ls, "receive_bast from %d no lkb %x",
3303 ms->m_header.h_nodeid, ms->m_remid);
3307 r = lkb->lkb_resource;
3312 error = validate_message(lkb, ms);
3316 queue_bast(r, lkb, ms->m_bastmode);
3323 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
3325 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
3327 from_nodeid = ms->m_header.h_nodeid;
3328 our_nodeid = dlm_our_nodeid();
3330 len = receive_extralen(ms);
3332 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3333 if (dir_nodeid != our_nodeid) {
3334 log_error(ls, "lookup dir_nodeid %d from %d",
3335 dir_nodeid, from_nodeid);
3341 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
3343 /* Optimization: we're master so treat lookup as a request */
3344 if (!error && ret_nodeid == our_nodeid) {
3345 receive_request(ls, ms);
3349 send_lookup_reply(ls, ms, ret_nodeid, error);
3352 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
3354 int len, dir_nodeid, from_nodeid;
3356 from_nodeid = ms->m_header.h_nodeid;
3358 len = receive_extralen(ms);
3360 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3361 if (dir_nodeid != dlm_our_nodeid()) {
3362 log_error(ls, "remove dir entry dir_nodeid %d from %d",
3363 dir_nodeid, from_nodeid);
3367 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
3370 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
3372 do_purge(ls, ms->m_nodeid, ms->m_pid);
3375 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
3377 struct dlm_lkb *lkb;
3379 int error, mstype, result;
3381 error = find_lkb(ls, ms->m_remid, &lkb);
3383 log_debug(ls, "receive_request_reply from %d no lkb %x",
3384 ms->m_header.h_nodeid, ms->m_remid);
3388 r = lkb->lkb_resource;
3392 error = validate_message(lkb, ms);
3396 mstype = lkb->lkb_wait_type;
3397 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
3401 /* Optimization: the dir node was also the master, so it took our
3402 lookup as a request and sent request reply instead of lookup reply */
3403 if (mstype == DLM_MSG_LOOKUP) {
3404 r->res_nodeid = ms->m_header.h_nodeid;
3405 lkb->lkb_nodeid = r->res_nodeid;
3408 /* this is the value returned from do_request() on the master */
3409 result = ms->m_result;
3413 /* request would block (be queued) on remote master */
3414 queue_cast(r, lkb, -EAGAIN);
3415 confirm_master(r, -EAGAIN);
3416 unhold_lkb(lkb); /* undoes create_lkb() */
3421 /* request was queued or granted on remote master */
3422 receive_flags_reply(lkb, ms);
3423 lkb->lkb_remid = ms->m_lkid;
3424 if (is_altmode(lkb))
3425 munge_altmode(lkb, ms);
3427 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3430 grant_lock_pc(r, lkb, ms);
3431 queue_cast(r, lkb, 0);
3433 confirm_master(r, result);
3438 /* find_rsb failed to find rsb or rsb wasn't master */
3439 log_debug(ls, "receive_request_reply %x %x master diff %d %d",
3440 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result);
3442 lkb->lkb_nodeid = -1;
3444 if (is_overlap(lkb)) {
3445 /* we'll ignore error in cancel/unlock reply */
3446 queue_cast_overlap(r, lkb);
3447 confirm_master(r, result);
3448 unhold_lkb(lkb); /* undoes create_lkb() */
3450 _request_lock(r, lkb);
3454 log_error(ls, "receive_request_reply %x error %d",
3455 lkb->lkb_id, result);
3458 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
3459 log_debug(ls, "receive_request_reply %x result %d unlock",
3460 lkb->lkb_id, result);
3461 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3462 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3463 send_unlock(r, lkb);
3464 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
3465 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
3466 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3467 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3468 send_cancel(r, lkb);
3470 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3471 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3479 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3480 struct dlm_message *ms)
3482 /* this is the value returned from do_convert() on the master */
3483 switch (ms->m_result) {
3485 /* convert would block (be queued) on remote master */
3486 queue_cast(r, lkb, -EAGAIN);
3490 receive_flags_reply(lkb, ms);
3491 revert_lock_pc(r, lkb);
3492 queue_cast(r, lkb, -EDEADLK);
3496 /* convert was queued on remote master */
3497 receive_flags_reply(lkb, ms);
3498 if (is_demoted(lkb))
3499 munge_demoted(lkb, ms);
3501 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3506 /* convert was granted on remote master */
3507 receive_flags_reply(lkb, ms);
3508 if (is_demoted(lkb))
3509 munge_demoted(lkb, ms);
3510 grant_lock_pc(r, lkb, ms);
3511 queue_cast(r, lkb, 0);
3515 log_error(r->res_ls, "receive_convert_reply %x error %d",
3516 lkb->lkb_id, ms->m_result);
3520 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3522 struct dlm_rsb *r = lkb->lkb_resource;
3528 error = validate_message(lkb, ms);
3532 /* stub reply can happen with waiters_mutex held */
3533 error = remove_from_waiters_ms(lkb, ms);
3537 __receive_convert_reply(r, lkb, ms);
3543 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
3545 struct dlm_lkb *lkb;
3548 error = find_lkb(ls, ms->m_remid, &lkb);
3550 log_debug(ls, "receive_convert_reply from %d no lkb %x",
3551 ms->m_header.h_nodeid, ms->m_remid);
3555 _receive_convert_reply(lkb, ms);
3559 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3561 struct dlm_rsb *r = lkb->lkb_resource;
3567 error = validate_message(lkb, ms);
3571 /* stub reply can happen with waiters_mutex held */
3572 error = remove_from_waiters_ms(lkb, ms);
3576 /* this is the value returned from do_unlock() on the master */
3578 switch (ms->m_result) {
3580 receive_flags_reply(lkb, ms);
3581 remove_lock_pc(r, lkb);
3582 queue_cast(r, lkb, -DLM_EUNLOCK);
3587 log_error(r->res_ls, "receive_unlock_reply %x error %d",
3588 lkb->lkb_id, ms->m_result);
3595 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
3597 struct dlm_lkb *lkb;
3600 error = find_lkb(ls, ms->m_remid, &lkb);
3602 log_debug(ls, "receive_unlock_reply from %d no lkb %x",
3603 ms->m_header.h_nodeid, ms->m_remid);
3607 _receive_unlock_reply(lkb, ms);
3611 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3613 struct dlm_rsb *r = lkb->lkb_resource;
3619 error = validate_message(lkb, ms);
3623 /* stub reply can happen with waiters_mutex held */
3624 error = remove_from_waiters_ms(lkb, ms);
3628 /* this is the value returned from do_cancel() on the master */
3630 switch (ms->m_result) {
3632 receive_flags_reply(lkb, ms);
3633 revert_lock_pc(r, lkb);
3634 queue_cast(r, lkb, -DLM_ECANCEL);
3639 log_error(r->res_ls, "receive_cancel_reply %x error %d",
3640 lkb->lkb_id, ms->m_result);
3647 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
3649 struct dlm_lkb *lkb;
3652 error = find_lkb(ls, ms->m_remid, &lkb);
3654 log_debug(ls, "receive_cancel_reply from %d no lkb %x",
3655 ms->m_header.h_nodeid, ms->m_remid);
3659 _receive_cancel_reply(lkb, ms);
3663 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
3665 struct dlm_lkb *lkb;
3667 int error, ret_nodeid;
3669 error = find_lkb(ls, ms->m_lkid, &lkb);
3671 log_error(ls, "receive_lookup_reply no lkb");
3675 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
3676 FIXME: will a non-zero error ever be returned? */
3678 r = lkb->lkb_resource;
3682 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3686 ret_nodeid = ms->m_nodeid;
3687 if (ret_nodeid == dlm_our_nodeid()) {
3690 r->res_first_lkid = 0;
3692 /* set_master() will copy res_nodeid to lkb_nodeid */
3693 r->res_nodeid = ret_nodeid;
3696 if (is_overlap(lkb)) {
3697 log_debug(ls, "receive_lookup_reply %x unlock %x",
3698 lkb->lkb_id, lkb->lkb_flags);
3699 queue_cast_overlap(r, lkb);
3700 unhold_lkb(lkb); /* undoes create_lkb() */
3704 _request_lock(r, lkb);
3708 process_lookup_list(r);
3715 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms)
3717 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
3718 log_debug(ls, "ignore non-member message %d from %d %x %x %d",
3719 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
3720 ms->m_remid, ms->m_result);
3724 switch (ms->m_type) {
3726 /* messages sent to a master node */
3728 case DLM_MSG_REQUEST:
3729 receive_request(ls, ms);
3732 case DLM_MSG_CONVERT:
3733 receive_convert(ls, ms);
3736 case DLM_MSG_UNLOCK:
3737 receive_unlock(ls, ms);
3740 case DLM_MSG_CANCEL:
3741 receive_cancel(ls, ms);
3744 /* messages sent from a master node (replies to above) */
3746 case DLM_MSG_REQUEST_REPLY:
3747 receive_request_reply(ls, ms);
3750 case DLM_MSG_CONVERT_REPLY:
3751 receive_convert_reply(ls, ms);
3754 case DLM_MSG_UNLOCK_REPLY:
3755 receive_unlock_reply(ls, ms);
3758 case DLM_MSG_CANCEL_REPLY:
3759 receive_cancel_reply(ls, ms);
3762 /* messages sent from a master node (only two types of async msg) */
3765 receive_grant(ls, ms);
3769 receive_bast(ls, ms);
3772 /* messages sent to a dir node */
3774 case DLM_MSG_LOOKUP:
3775 receive_lookup(ls, ms);
3778 case DLM_MSG_REMOVE:
3779 receive_remove(ls, ms);
3782 /* messages sent from a dir node (remove has no reply) */
3784 case DLM_MSG_LOOKUP_REPLY:
3785 receive_lookup_reply(ls, ms);
3788 /* other messages */
3791 receive_purge(ls, ms);
3795 log_error(ls, "unknown message type %d", ms->m_type);
3801 /* If the lockspace is in recovery mode (locking stopped), then normal
3802 messages are saved on the requestqueue for processing after recovery is
3803 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
3804 messages off the requestqueue before we process new ones. This occurs right
3805 after recovery completes when we transition from saving all messages on
3806 requestqueue, to processing all the saved messages, to processing new
3807 messages as they arrive. */
3809 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
3812 if (dlm_locking_stopped(ls)) {
3813 dlm_add_requestqueue(ls, nodeid, ms);
3815 dlm_wait_requestqueue(ls);
3816 _receive_message(ls, ms);
3820 /* This is called by dlm_recoverd to process messages that were saved on
3821 the requestqueue. */
3823 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms)
3825 _receive_message(ls, ms);
3828 /* This is called by the midcomms layer when something is received for
3829 the lockspace. It could be either a MSG (normal message sent as part of
3830 standard locking activity) or an RCOM (recovery message sent as part of
3831 lockspace recovery). */
3833 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
3835 struct dlm_header *hd = &p->header;
3839 switch (hd->h_cmd) {
3841 dlm_message_in(&p->message);
3842 type = p->message.m_type;
3845 dlm_rcom_in(&p->rcom);
3846 type = p->rcom.rc_type;
3849 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
3853 if (hd->h_nodeid != nodeid) {
3854 log_print("invalid h_nodeid %d from %d lockspace %x",
3855 hd->h_nodeid, nodeid, hd->h_lockspace);
3859 ls = dlm_find_lockspace_global(hd->h_lockspace);
3861 if (dlm_config.ci_log_debug)
3862 log_print("invalid lockspace %x from %d cmd %d type %d",
3863 hd->h_lockspace, nodeid, hd->h_cmd, type);
3865 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
3866 dlm_send_ls_not_ready(nodeid, &p->rcom);
3870 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
3871 be inactive (in this ls) before transitioning to recovery mode */
3873 down_read(&ls->ls_recv_active);
3874 if (hd->h_cmd == DLM_MSG)
3875 dlm_receive_message(ls, &p->message, nodeid);
3877 dlm_receive_rcom(ls, &p->rcom, nodeid);
3878 up_read(&ls->ls_recv_active);
3880 dlm_put_lockspace(ls);
3883 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3885 if (middle_conversion(lkb)) {
3887 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3888 ls->ls_stub_ms.m_result = -EINPROGRESS;
3889 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3890 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
3891 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3893 /* Same special case as in receive_rcom_lock_args() */
3894 lkb->lkb_grmode = DLM_LOCK_IV;
3895 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3898 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3899 lkb->lkb_flags |= DLM_IFL_RESEND;
3902 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3903 conversions are async; there's no reply from the remote master */
3906 /* A waiting lkb needs recovery if the master node has failed, or
3907 the master node is changing (only when no directory is used) */
3909 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3911 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3914 if (!dlm_no_directory(ls))
3917 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3923 /* Recovery for locks that are waiting for replies from nodes that are now
3924 gone. We can just complete unlocks and cancels by faking a reply from the
3925 dead node. Requests and up-conversions we flag to be resent after
3926 recovery. Down-conversions can just be completed with a fake reply like
3927 unlocks. Conversions between PR and CW need special attention. */
3929 void dlm_recover_waiters_pre(struct dlm_ls *ls)
3931 struct dlm_lkb *lkb, *safe;
3932 int wait_type, stub_unlock_result, stub_cancel_result;
3934 mutex_lock(&ls->ls_waiters_mutex);
3936 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3937 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3938 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3940 /* all outstanding lookups, regardless of destination will be
3941 resent after recovery is done */
3943 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3944 lkb->lkb_flags |= DLM_IFL_RESEND;
3948 if (!waiter_needs_recovery(ls, lkb))
3951 wait_type = lkb->lkb_wait_type;
3952 stub_unlock_result = -DLM_EUNLOCK;
3953 stub_cancel_result = -DLM_ECANCEL;
3955 /* Main reply may have been received leaving a zero wait_type,
3956 but a reply for the overlapping op may not have been
3957 received. In that case we need to fake the appropriate
3958 reply for the overlap op. */
3961 if (is_overlap_cancel(lkb)) {
3962 wait_type = DLM_MSG_CANCEL;
3963 if (lkb->lkb_grmode == DLM_LOCK_IV)
3964 stub_cancel_result = 0;
3966 if (is_overlap_unlock(lkb)) {
3967 wait_type = DLM_MSG_UNLOCK;
3968 if (lkb->lkb_grmode == DLM_LOCK_IV)
3969 stub_unlock_result = -ENOENT;
3972 log_debug(ls, "rwpre overlap %x %x %d %d %d",
3973 lkb->lkb_id, lkb->lkb_flags, wait_type,
3974 stub_cancel_result, stub_unlock_result);
3977 switch (wait_type) {
3979 case DLM_MSG_REQUEST:
3980 lkb->lkb_flags |= DLM_IFL_RESEND;
3983 case DLM_MSG_CONVERT:
3984 recover_convert_waiter(ls, lkb);
3987 case DLM_MSG_UNLOCK:
3989 ls->ls_stub_ms.m_type = DLM_MSG_UNLOCK_REPLY;
3990 ls->ls_stub_ms.m_result = stub_unlock_result;
3991 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3992 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
3993 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
3997 case DLM_MSG_CANCEL:
3999 ls->ls_stub_ms.m_type = DLM_MSG_CANCEL_REPLY;
4000 ls->ls_stub_ms.m_result = stub_cancel_result;
4001 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
4002 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
4003 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
4008 log_error(ls, "invalid lkb wait_type %d %d",
4009 lkb->lkb_wait_type, wait_type);
4013 mutex_unlock(&ls->ls_waiters_mutex);
4016 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
4018 struct dlm_lkb *lkb;
4021 mutex_lock(&ls->ls_waiters_mutex);
4022 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
4023 if (lkb->lkb_flags & DLM_IFL_RESEND) {
4029 mutex_unlock(&ls->ls_waiters_mutex);
4036 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
4037 master or dir-node for r. Processing the lkb may result in it being placed
4040 /* We do this after normal locking has been enabled and any saved messages
4041 (in requestqueue) have been processed. We should be confident that at
4042 this point we won't get or process a reply to any of these waiting
4043 operations. But, new ops may be coming in on the rsbs/locks here from
4044 userspace or remotely. */
4046 /* there may have been an overlap unlock/cancel prior to recovery or after
4047 recovery. if before, the lkb may still have a pos wait_count; if after, the
4048 overlap flag would just have been set and nothing new sent. we can be
4049 confident here than any replies to either the initial op or overlap ops
4050 prior to recovery have been received. */
4052 int dlm_recover_waiters_post(struct dlm_ls *ls)
4054 struct dlm_lkb *lkb;
4056 int error = 0, mstype, err, oc, ou;
4059 if (dlm_locking_stopped(ls)) {
4060 log_debug(ls, "recover_waiters_post aborted");
4065 lkb = find_resend_waiter(ls);
4069 r = lkb->lkb_resource;
4073 mstype = lkb->lkb_wait_type;
4074 oc = is_overlap_cancel(lkb);
4075 ou = is_overlap_unlock(lkb);
4078 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
4079 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
4081 /* At this point we assume that we won't get a reply to any
4082 previous op or overlap op on this lock. First, do a big
4083 remove_from_waiters() for all previous ops. */
4085 lkb->lkb_flags &= ~DLM_IFL_RESEND;
4086 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4087 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4088 lkb->lkb_wait_type = 0;
4089 lkb->lkb_wait_count = 0;
4090 mutex_lock(&ls->ls_waiters_mutex);
4091 list_del_init(&lkb->lkb_wait_reply);
4092 mutex_unlock(&ls->ls_waiters_mutex);
4093 unhold_lkb(lkb); /* for waiters list */
4096 /* do an unlock or cancel instead of resending */
4098 case DLM_MSG_LOOKUP:
4099 case DLM_MSG_REQUEST:
4100 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
4102 unhold_lkb(lkb); /* undoes create_lkb() */
4104 case DLM_MSG_CONVERT:
4106 queue_cast(r, lkb, -DLM_ECANCEL);
4108 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
4109 _unlock_lock(r, lkb);
4117 case DLM_MSG_LOOKUP:
4118 case DLM_MSG_REQUEST:
4119 _request_lock(r, lkb);
4121 confirm_master(r, 0);
4123 case DLM_MSG_CONVERT:
4124 _convert_lock(r, lkb);
4132 log_error(ls, "recover_waiters_post %x %d %x %d %d",
4133 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou);
4142 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
4143 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
4145 struct dlm_ls *ls = r->res_ls;
4146 struct dlm_lkb *lkb, *safe;
4148 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
4149 if (test(ls, lkb)) {
4150 rsb_set_flag(r, RSB_LOCKS_PURGED);
4152 /* this put should free the lkb */
4153 if (!dlm_put_lkb(lkb))
4154 log_error(ls, "purged lkb not released");
4159 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4161 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
4164 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4166 return is_master_copy(lkb);
4169 static void purge_dead_locks(struct dlm_rsb *r)
4171 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
4172 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
4173 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
4176 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
4178 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
4179 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
4180 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
4183 /* Get rid of locks held by nodes that are gone. */
4185 int dlm_purge_locks(struct dlm_ls *ls)
4189 log_debug(ls, "dlm_purge_locks");
4191 down_write(&ls->ls_root_sem);
4192 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
4196 purge_dead_locks(r);
4202 up_write(&ls->ls_root_sem);
4207 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
4209 struct dlm_rsb *r, *r_ret = NULL;
4211 read_lock(&ls->ls_rsbtbl[bucket].lock);
4212 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
4213 if (!rsb_flag(r, RSB_LOCKS_PURGED))
4216 rsb_clear_flag(r, RSB_LOCKS_PURGED);
4220 read_unlock(&ls->ls_rsbtbl[bucket].lock);
4224 void dlm_grant_after_purge(struct dlm_ls *ls)
4230 r = find_purged_rsb(ls, bucket);
4232 if (bucket == ls->ls_rsbtbl_size - 1)
4239 grant_pending_locks(r);
4240 confirm_master(r, 0);
4248 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
4251 struct dlm_lkb *lkb;
4253 list_for_each_entry(lkb, head, lkb_statequeue) {
4254 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
4260 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
4263 struct dlm_lkb *lkb;
4265 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
4268 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
4271 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
4277 /* needs at least dlm_rcom + rcom_lock */
4278 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
4279 struct dlm_rsb *r, struct dlm_rcom *rc)
4281 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4283 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
4284 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
4285 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
4286 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
4287 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
4288 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4289 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
4290 lkb->lkb_rqmode = rl->rl_rqmode;
4291 lkb->lkb_grmode = rl->rl_grmode;
4292 /* don't set lkb_status because add_lkb wants to itself */
4294 lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
4295 lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
4297 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
4298 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
4299 sizeof(struct rcom_lock);
4300 if (lvblen > ls->ls_lvblen)
4302 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
4303 if (!lkb->lkb_lvbptr)
4305 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
4308 /* Conversions between PR and CW (middle modes) need special handling.
4309 The real granted mode of these converting locks cannot be determined
4310 until all locks have been rebuilt on the rsb (recover_conversion) */
4312 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
4313 middle_conversion(lkb)) {
4314 rl->rl_status = DLM_LKSTS_CONVERT;
4315 lkb->lkb_grmode = DLM_LOCK_IV;
4316 rsb_set_flag(r, RSB_RECOVER_CONVERT);
4322 /* This lkb may have been recovered in a previous aborted recovery so we need
4323 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4324 If so we just send back a standard reply. If not, we create a new lkb with
4325 the given values and send back our lkid. We send back our lkid by sending
4326 back the rcom_lock struct we got but with the remid field filled in. */
4328 /* needs at least dlm_rcom + rcom_lock */
4329 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4331 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4333 struct dlm_lkb *lkb;
4336 if (rl->rl_parent_lkid) {
4337 error = -EOPNOTSUPP;
4341 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
4348 lkb = search_remid(r, rc->rc_header.h_nodeid, le32_to_cpu(rl->rl_lkid));
4354 error = create_lkb(ls, &lkb);
4358 error = receive_rcom_lock_args(ls, lkb, r, rc);
4365 add_lkb(r, lkb, rl->rl_status);
4369 /* this is the new value returned to the lock holder for
4370 saving in its process-copy lkb */
4371 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
4378 log_debug(ls, "recover_master_copy %d %x", error,
4379 le32_to_cpu(rl->rl_lkid));
4380 rl->rl_result = cpu_to_le32(error);
4384 /* needs at least dlm_rcom + rcom_lock */
4385 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4387 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4389 struct dlm_lkb *lkb;
4392 error = find_lkb(ls, le32_to_cpu(rl->rl_lkid), &lkb);
4394 log_error(ls, "recover_process_copy no lkid %x",
4395 le32_to_cpu(rl->rl_lkid));
4399 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
4401 error = le32_to_cpu(rl->rl_result);
4403 r = lkb->lkb_resource;
4409 /* There's a chance the new master received our lock before
4410 dlm_recover_master_reply(), this wouldn't happen if we did
4411 a barrier between recover_masters and recover_locks. */
4412 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
4413 (unsigned long)r, r->res_name);
4414 dlm_send_rcom_lock(r, lkb);
4417 log_debug(ls, "master copy exists %x", lkb->lkb_id);
4420 lkb->lkb_remid = le32_to_cpu(rl->rl_remid);
4423 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
4424 error, lkb->lkb_id);
4427 /* an ack for dlm_recover_locks() which waits for replies from
4428 all the locks it sends to new masters */
4429 dlm_recovered_lock(r);
4438 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
4439 int mode, uint32_t flags, void *name, unsigned int namelen,
4440 unsigned long timeout_cs)
4442 struct dlm_lkb *lkb;
4443 struct dlm_args args;
4446 dlm_lock_recovery(ls);
4448 error = create_lkb(ls, &lkb);
4454 if (flags & DLM_LKF_VALBLK) {
4455 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
4456 if (!ua->lksb.sb_lvbptr) {
4464 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
4465 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4466 lock and that lkb_astparam is the dlm_user_args structure. */
4468 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
4469 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
4470 lkb->lkb_flags |= DLM_IFL_USER;
4471 ua->old_mode = DLM_LOCK_IV;
4478 error = request_lock(ls, lkb, name, namelen, &args);
4494 /* add this new lkb to the per-process list of locks */
4495 spin_lock(&ua->proc->locks_spin);
4497 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
4498 spin_unlock(&ua->proc->locks_spin);
4500 dlm_unlock_recovery(ls);
4504 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4505 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
4506 unsigned long timeout_cs)
4508 struct dlm_lkb *lkb;
4509 struct dlm_args args;
4510 struct dlm_user_args *ua;
4513 dlm_lock_recovery(ls);
4515 error = find_lkb(ls, lkid, &lkb);
4519 /* user can change the params on its lock when it converts it, or
4520 add an lvb that didn't exist before */
4522 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4524 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
4525 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
4526 if (!ua->lksb.sb_lvbptr) {
4531 if (lvb_in && ua->lksb.sb_lvbptr)
4532 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4534 ua->xid = ua_tmp->xid;
4535 ua->castparam = ua_tmp->castparam;
4536 ua->castaddr = ua_tmp->castaddr;
4537 ua->bastparam = ua_tmp->bastparam;
4538 ua->bastaddr = ua_tmp->bastaddr;
4539 ua->user_lksb = ua_tmp->user_lksb;
4540 ua->old_mode = lkb->lkb_grmode;
4542 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
4543 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
4547 error = convert_lock(ls, lkb, &args);
4549 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
4554 dlm_unlock_recovery(ls);
4559 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4560 uint32_t flags, uint32_t lkid, char *lvb_in)
4562 struct dlm_lkb *lkb;
4563 struct dlm_args args;
4564 struct dlm_user_args *ua;
4567 dlm_lock_recovery(ls);
4569 error = find_lkb(ls, lkid, &lkb);
4573 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4575 if (lvb_in && ua->lksb.sb_lvbptr)
4576 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4577 if (ua_tmp->castparam)
4578 ua->castparam = ua_tmp->castparam;
4579 ua->user_lksb = ua_tmp->user_lksb;
4581 error = set_unlock_args(flags, ua, &args);
4585 error = unlock_lock(ls, lkb, &args);
4587 if (error == -DLM_EUNLOCK)
4589 /* from validate_unlock_args() */
4590 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
4595 spin_lock(&ua->proc->locks_spin);
4596 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4597 if (!list_empty(&lkb->lkb_ownqueue))
4598 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
4599 spin_unlock(&ua->proc->locks_spin);
4603 dlm_unlock_recovery(ls);
4608 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4609 uint32_t flags, uint32_t lkid)
4611 struct dlm_lkb *lkb;
4612 struct dlm_args args;
4613 struct dlm_user_args *ua;
4616 dlm_lock_recovery(ls);
4618 error = find_lkb(ls, lkid, &lkb);
4622 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4623 if (ua_tmp->castparam)
4624 ua->castparam = ua_tmp->castparam;
4625 ua->user_lksb = ua_tmp->user_lksb;
4627 error = set_unlock_args(flags, ua, &args);
4631 error = cancel_lock(ls, lkb, &args);
4633 if (error == -DLM_ECANCEL)
4635 /* from validate_unlock_args() */
4636 if (error == -EBUSY)
4641 dlm_unlock_recovery(ls);
4646 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
4648 struct dlm_lkb *lkb;
4649 struct dlm_args args;
4650 struct dlm_user_args *ua;
4654 dlm_lock_recovery(ls);
4656 error = find_lkb(ls, lkid, &lkb);
4660 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4662 error = set_unlock_args(flags, ua, &args);
4666 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4668 r = lkb->lkb_resource;
4672 error = validate_unlock_args(lkb, &args);
4675 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
4677 error = _cancel_lock(r, lkb);
4682 if (error == -DLM_ECANCEL)
4684 /* from validate_unlock_args() */
4685 if (error == -EBUSY)
4690 dlm_unlock_recovery(ls);
4694 /* lkb's that are removed from the waiters list by revert are just left on the
4695 orphans list with the granted orphan locks, to be freed by purge */
4697 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4699 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4700 struct dlm_args args;
4704 mutex_lock(&ls->ls_orphans_mutex);
4705 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
4706 mutex_unlock(&ls->ls_orphans_mutex);
4708 set_unlock_args(0, ua, &args);
4710 error = cancel_lock(ls, lkb, &args);
4711 if (error == -DLM_ECANCEL)
4716 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4717 Regardless of what rsb queue the lock is on, it's removed and freed. */
4719 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4721 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4722 struct dlm_args args;
4725 set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
4727 error = unlock_lock(ls, lkb, &args);
4728 if (error == -DLM_EUNLOCK)
4733 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4734 (which does lock_rsb) due to deadlock with receiving a message that does
4735 lock_rsb followed by dlm_user_add_ast() */
4737 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
4738 struct dlm_user_proc *proc)
4740 struct dlm_lkb *lkb = NULL;
4742 mutex_lock(&ls->ls_clear_proc_locks);
4743 if (list_empty(&proc->locks))
4746 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
4747 list_del_init(&lkb->lkb_ownqueue);
4749 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4750 lkb->lkb_flags |= DLM_IFL_ORPHAN;
4752 lkb->lkb_flags |= DLM_IFL_DEAD;
4754 mutex_unlock(&ls->ls_clear_proc_locks);
4758 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4759 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4760 which we clear here. */
4762 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
4763 list, and no more device_writes should add lkb's to proc->locks list; so we
4764 shouldn't need to take asts_spin or locks_spin here. this assumes that
4765 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4768 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4770 struct dlm_lkb *lkb, *safe;
4772 dlm_lock_recovery(ls);
4775 lkb = del_proc_lock(ls, proc);
4779 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4780 orphan_proc_lock(ls, lkb);
4782 unlock_proc_lock(ls, lkb);
4784 /* this removes the reference for the proc->locks list
4785 added by dlm_user_request, it may result in the lkb
4791 mutex_lock(&ls->ls_clear_proc_locks);
4793 /* in-progress unlocks */
4794 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4795 list_del_init(&lkb->lkb_ownqueue);
4796 lkb->lkb_flags |= DLM_IFL_DEAD;
4800 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4801 lkb->lkb_ast_type = 0;
4802 list_del(&lkb->lkb_astqueue);
4806 mutex_unlock(&ls->ls_clear_proc_locks);
4807 dlm_unlock_recovery(ls);
4810 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4812 struct dlm_lkb *lkb, *safe;
4816 spin_lock(&proc->locks_spin);
4817 if (!list_empty(&proc->locks)) {
4818 lkb = list_entry(proc->locks.next, struct dlm_lkb,
4820 list_del_init(&lkb->lkb_ownqueue);
4822 spin_unlock(&proc->locks_spin);
4827 lkb->lkb_flags |= DLM_IFL_DEAD;
4828 unlock_proc_lock(ls, lkb);
4829 dlm_put_lkb(lkb); /* ref from proc->locks list */
4832 spin_lock(&proc->locks_spin);
4833 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4834 list_del_init(&lkb->lkb_ownqueue);
4835 lkb->lkb_flags |= DLM_IFL_DEAD;
4838 spin_unlock(&proc->locks_spin);
4840 spin_lock(&proc->asts_spin);
4841 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4842 list_del(&lkb->lkb_astqueue);
4845 spin_unlock(&proc->asts_spin);
4848 /* pid of 0 means purge all orphans */
4850 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
4852 struct dlm_lkb *lkb, *safe;
4854 mutex_lock(&ls->ls_orphans_mutex);
4855 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
4856 if (pid && lkb->lkb_ownpid != pid)
4858 unlock_proc_lock(ls, lkb);
4859 list_del_init(&lkb->lkb_ownqueue);
4862 mutex_unlock(&ls->ls_orphans_mutex);
4865 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
4867 struct dlm_message *ms;
4868 struct dlm_mhandle *mh;
4871 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
4872 DLM_MSG_PURGE, &ms, &mh);
4875 ms->m_nodeid = nodeid;
4878 return send_message(mh, ms);
4881 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
4882 int nodeid, int pid)
4886 if (nodeid != dlm_our_nodeid()) {
4887 error = send_purge(ls, nodeid, pid);
4889 dlm_lock_recovery(ls);
4890 if (pid == current->pid)
4891 purge_proc_locks(ls, proc);
4893 do_purge(ls, nodeid, pid);
4894 dlm_unlock_recovery(ls);