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 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1230 lkb->lkb_lvbseq = ms->m_lvbseq;
1234 /* Manipulate lkb's on rsb's convert/granted/waiting queues
1235 remove_lock -- used for unlock, removes lkb from granted
1236 revert_lock -- used for cancel, moves lkb from convert to granted
1237 grant_lock -- used for request and convert, adds lkb to granted or
1238 moves lkb from convert or waiting to granted
1240 Each of these is used for master or local copy lkb's. There is
1241 also a _pc() variation used to make the corresponding change on
1242 a process copy (pc) lkb. */
1244 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1247 lkb->lkb_grmode = DLM_LOCK_IV;
1248 /* this unhold undoes the original ref from create_lkb()
1249 so this leads to the lkb being freed */
1253 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1255 set_lvb_unlock(r, lkb);
1256 _remove_lock(r, lkb);
1259 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1261 _remove_lock(r, lkb);
1264 /* returns: 0 did nothing
1265 1 moved lock to granted
1268 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1272 lkb->lkb_rqmode = DLM_LOCK_IV;
1274 switch (lkb->lkb_status) {
1275 case DLM_LKSTS_GRANTED:
1277 case DLM_LKSTS_CONVERT:
1278 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1281 case DLM_LKSTS_WAITING:
1283 lkb->lkb_grmode = DLM_LOCK_IV;
1284 /* this unhold undoes the original ref from create_lkb()
1285 so this leads to the lkb being freed */
1290 log_print("invalid status for revert %d", lkb->lkb_status);
1295 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1297 return revert_lock(r, lkb);
1300 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1302 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1303 lkb->lkb_grmode = lkb->lkb_rqmode;
1304 if (lkb->lkb_status)
1305 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1307 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1310 lkb->lkb_rqmode = DLM_LOCK_IV;
1313 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1315 set_lvb_lock(r, lkb);
1316 _grant_lock(r, lkb);
1317 lkb->lkb_highbast = 0;
1320 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1321 struct dlm_message *ms)
1323 set_lvb_lock_pc(r, lkb, ms);
1324 _grant_lock(r, lkb);
1327 /* called by grant_pending_locks() which means an async grant message must
1328 be sent to the requesting node in addition to granting the lock if the
1329 lkb belongs to a remote node. */
1331 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1334 if (is_master_copy(lkb))
1337 queue_cast(r, lkb, 0);
1340 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1341 change the granted/requested modes. We're munging things accordingly in
1343 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1345 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1346 compatible with other granted locks */
1348 static void munge_demoted(struct dlm_lkb *lkb, struct dlm_message *ms)
1350 if (ms->m_type != DLM_MSG_CONVERT_REPLY) {
1351 log_print("munge_demoted %x invalid reply type %d",
1352 lkb->lkb_id, ms->m_type);
1356 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
1357 log_print("munge_demoted %x invalid modes gr %d rq %d",
1358 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
1362 lkb->lkb_grmode = DLM_LOCK_NL;
1365 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
1367 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
1368 ms->m_type != DLM_MSG_GRANT) {
1369 log_print("munge_altmode %x invalid reply type %d",
1370 lkb->lkb_id, ms->m_type);
1374 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
1375 lkb->lkb_rqmode = DLM_LOCK_PR;
1376 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
1377 lkb->lkb_rqmode = DLM_LOCK_CW;
1379 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
1384 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1386 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1388 if (lkb->lkb_id == first->lkb_id)
1394 /* Check if the given lkb conflicts with another lkb on the queue. */
1396 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1398 struct dlm_lkb *this;
1400 list_for_each_entry(this, head, lkb_statequeue) {
1403 if (!modes_compat(this, lkb))
1410 * "A conversion deadlock arises with a pair of lock requests in the converting
1411 * queue for one resource. The granted mode of each lock blocks the requested
1412 * mode of the other lock."
1414 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1415 * convert queue from being granted, then deadlk/demote lkb.
1418 * Granted Queue: empty
1419 * Convert Queue: NL->EX (first lock)
1420 * PR->EX (second lock)
1422 * The first lock can't be granted because of the granted mode of the second
1423 * lock and the second lock can't be granted because it's not first in the
1424 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1425 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1426 * flag set and return DEMOTED in the lksb flags.
1428 * Originally, this function detected conv-deadlk in a more limited scope:
1429 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1430 * - if lkb1 was the first entry in the queue (not just earlier), and was
1431 * blocked by the granted mode of lkb2, and there was nothing on the
1432 * granted queue preventing lkb1 from being granted immediately, i.e.
1433 * lkb2 was the only thing preventing lkb1 from being granted.
1435 * That second condition meant we'd only say there was conv-deadlk if
1436 * resolving it (by demotion) would lead to the first lock on the convert
1437 * queue being granted right away. It allowed conversion deadlocks to exist
1438 * between locks on the convert queue while they couldn't be granted anyway.
1440 * Now, we detect and take action on conversion deadlocks immediately when
1441 * they're created, even if they may not be immediately consequential. If
1442 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1443 * mode that would prevent lkb1's conversion from being granted, we do a
1444 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1445 * I think this means that the lkb_is_ahead condition below should always
1446 * be zero, i.e. there will never be conv-deadlk between two locks that are
1447 * both already on the convert queue.
1450 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
1452 struct dlm_lkb *lkb1;
1453 int lkb_is_ahead = 0;
1455 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
1461 if (!lkb_is_ahead) {
1462 if (!modes_compat(lkb2, lkb1))
1465 if (!modes_compat(lkb2, lkb1) &&
1466 !modes_compat(lkb1, lkb2))
1474 * Return 1 if the lock can be granted, 0 otherwise.
1475 * Also detect and resolve conversion deadlocks.
1477 * lkb is the lock to be granted
1479 * now is 1 if the function is being called in the context of the
1480 * immediate request, it is 0 if called later, after the lock has been
1483 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1486 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1488 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1491 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1492 * a new request for a NL mode lock being blocked.
1494 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1495 * request, then it would be granted. In essence, the use of this flag
1496 * tells the Lock Manager to expedite theis request by not considering
1497 * what may be in the CONVERTING or WAITING queues... As of this
1498 * writing, the EXPEDITE flag can be used only with new requests for NL
1499 * mode locks. This flag is not valid for conversion requests.
1501 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1502 * conversion or used with a non-NL requested mode. We also know an
1503 * EXPEDITE request is always granted immediately, so now must always
1504 * be 1. The full condition to grant an expedite request: (now &&
1505 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1506 * therefore be shortened to just checking the flag.
1509 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1513 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1514 * added to the remaining conditions.
1517 if (queue_conflict(&r->res_grantqueue, lkb))
1521 * 6-3: By default, a conversion request is immediately granted if the
1522 * requested mode is compatible with the modes of all other granted
1526 if (queue_conflict(&r->res_convertqueue, lkb))
1530 * 6-5: But the default algorithm for deciding whether to grant or
1531 * queue conversion requests does not by itself guarantee that such
1532 * requests are serviced on a "first come first serve" basis. This, in
1533 * turn, can lead to a phenomenon known as "indefinate postponement".
1535 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1536 * the system service employed to request a lock conversion. This flag
1537 * forces certain conversion requests to be queued, even if they are
1538 * compatible with the granted modes of other locks on the same
1539 * resource. Thus, the use of this flag results in conversion requests
1540 * being ordered on a "first come first servce" basis.
1542 * DCT: This condition is all about new conversions being able to occur
1543 * "in place" while the lock remains on the granted queue (assuming
1544 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1545 * doesn't _have_ to go onto the convert queue where it's processed in
1546 * order. The "now" variable is necessary to distinguish converts
1547 * being received and processed for the first time now, because once a
1548 * convert is moved to the conversion queue the condition below applies
1549 * requiring fifo granting.
1552 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1556 * The NOORDER flag is set to avoid the standard vms rules on grant
1560 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1564 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1565 * granted until all other conversion requests ahead of it are granted
1569 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1573 * 6-4: By default, a new request is immediately granted only if all
1574 * three of the following conditions are satisfied when the request is
1576 * - The queue of ungranted conversion requests for the resource is
1578 * - The queue of ungranted new requests for the resource is empty.
1579 * - The mode of the new request is compatible with the most
1580 * restrictive mode of all granted locks on the resource.
1583 if (now && !conv && list_empty(&r->res_convertqueue) &&
1584 list_empty(&r->res_waitqueue))
1588 * 6-4: Once a lock request is in the queue of ungranted new requests,
1589 * it cannot be granted until the queue of ungranted conversion
1590 * requests is empty, all ungranted new requests ahead of it are
1591 * granted and/or canceled, and it is compatible with the granted mode
1592 * of the most restrictive lock granted on the resource.
1595 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1596 first_in_list(lkb, &r->res_waitqueue))
1602 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
1606 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1607 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
1612 rv = _can_be_granted(r, lkb, now);
1617 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1618 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1619 * cancels one of the locks.
1622 if (is_convert && can_be_queued(lkb) &&
1623 conversion_deadlock_detect(r, lkb)) {
1624 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
1625 lkb->lkb_grmode = DLM_LOCK_NL;
1626 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1627 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1631 log_print("can_be_granted deadlock %x now %d",
1640 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1641 * to grant a request in a mode other than the normal rqmode. It's a
1642 * simple way to provide a big optimization to applications that can
1646 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
1648 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
1652 lkb->lkb_rqmode = alt;
1653 rv = _can_be_granted(r, lkb, now);
1655 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1657 lkb->lkb_rqmode = rqmode;
1663 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1664 for locks pending on the convert list. Once verified (watch for these
1665 log_prints), we should be able to just call _can_be_granted() and not
1666 bother with the demote/deadlk cases here (and there's no easy way to deal
1667 with a deadlk here, we'd have to generate something like grant_lock with
1668 the deadlk error.) */
1670 /* Returns the highest requested mode of all blocked conversions; sets
1671 cw if there's a blocked conversion to DLM_LOCK_CW. */
1673 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
1675 struct dlm_lkb *lkb, *s;
1676 int hi, demoted, quit, grant_restart, demote_restart;
1685 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1686 demoted = is_demoted(lkb);
1689 if (can_be_granted(r, lkb, 0, &deadlk)) {
1690 grant_lock_pending(r, lkb);
1695 if (!demoted && is_demoted(lkb)) {
1696 log_print("WARN: pending demoted %x node %d %s",
1697 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1703 log_print("WARN: pending deadlock %x node %d %s",
1704 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1709 hi = max_t(int, lkb->lkb_rqmode, hi);
1711 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
1717 if (demote_restart && !quit) {
1722 return max_t(int, high, hi);
1725 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
1727 struct dlm_lkb *lkb, *s;
1729 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1730 if (can_be_granted(r, lkb, 0, NULL))
1731 grant_lock_pending(r, lkb);
1733 high = max_t(int, lkb->lkb_rqmode, high);
1734 if (lkb->lkb_rqmode == DLM_LOCK_CW)
1742 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
1743 on either the convert or waiting queue.
1744 high is the largest rqmode of all locks blocked on the convert or
1747 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
1749 if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
1750 if (gr->lkb_highbast < DLM_LOCK_EX)
1755 if (gr->lkb_highbast < high &&
1756 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
1761 static void grant_pending_locks(struct dlm_rsb *r)
1763 struct dlm_lkb *lkb, *s;
1764 int high = DLM_LOCK_IV;
1767 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
1769 high = grant_pending_convert(r, high, &cw);
1770 high = grant_pending_wait(r, high, &cw);
1772 if (high == DLM_LOCK_IV)
1776 * If there are locks left on the wait/convert queue then send blocking
1777 * ASTs to granted locks based on the largest requested mode (high)
1781 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1782 if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
1783 if (cw && high == DLM_LOCK_PR)
1784 queue_bast(r, lkb, DLM_LOCK_CW);
1786 queue_bast(r, lkb, high);
1787 lkb->lkb_highbast = high;
1792 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
1794 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
1795 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
1796 if (gr->lkb_highbast < DLM_LOCK_EX)
1801 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
1806 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1807 struct dlm_lkb *lkb)
1811 list_for_each_entry(gr, head, lkb_statequeue) {
1812 if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
1813 queue_bast(r, gr, lkb->lkb_rqmode);
1814 gr->lkb_highbast = lkb->lkb_rqmode;
1819 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1821 send_bast_queue(r, &r->res_grantqueue, lkb);
1824 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1826 send_bast_queue(r, &r->res_grantqueue, lkb);
1827 send_bast_queue(r, &r->res_convertqueue, lkb);
1830 /* set_master(r, lkb) -- set the master nodeid of a resource
1832 The purpose of this function is to set the nodeid field in the given
1833 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1834 known, it can just be copied to the lkb and the function will return
1835 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1836 before it can be copied to the lkb.
1838 When the rsb nodeid is being looked up remotely, the initial lkb
1839 causing the lookup is kept on the ls_waiters list waiting for the
1840 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1841 on the rsb's res_lookup list until the master is verified.
1844 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1845 1: the rsb master is not available and the lkb has been placed on
1849 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1851 struct dlm_ls *ls = r->res_ls;
1852 int i, error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
1854 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1855 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1856 r->res_first_lkid = lkb->lkb_id;
1857 lkb->lkb_nodeid = r->res_nodeid;
1861 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1862 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1866 if (r->res_nodeid == 0) {
1867 lkb->lkb_nodeid = 0;
1871 if (r->res_nodeid > 0) {
1872 lkb->lkb_nodeid = r->res_nodeid;
1876 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
1878 dir_nodeid = dlm_dir_nodeid(r);
1880 if (dir_nodeid != our_nodeid) {
1881 r->res_first_lkid = lkb->lkb_id;
1882 send_lookup(r, lkb);
1886 for (i = 0; i < 2; i++) {
1887 /* It's possible for dlm_scand to remove an old rsb for
1888 this same resource from the toss list, us to create
1889 a new one, look up the master locally, and find it
1890 already exists just before dlm_scand does the
1891 dir_remove() on the previous rsb. */
1893 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1894 r->res_length, &ret_nodeid);
1897 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1900 if (error && error != -EEXIST)
1903 if (ret_nodeid == our_nodeid) {
1904 r->res_first_lkid = 0;
1906 lkb->lkb_nodeid = 0;
1908 r->res_first_lkid = lkb->lkb_id;
1909 r->res_nodeid = ret_nodeid;
1910 lkb->lkb_nodeid = ret_nodeid;
1915 static void process_lookup_list(struct dlm_rsb *r)
1917 struct dlm_lkb *lkb, *safe;
1919 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
1920 list_del_init(&lkb->lkb_rsb_lookup);
1921 _request_lock(r, lkb);
1926 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1928 static void confirm_master(struct dlm_rsb *r, int error)
1930 struct dlm_lkb *lkb;
1932 if (!r->res_first_lkid)
1938 r->res_first_lkid = 0;
1939 process_lookup_list(r);
1945 /* the remote request failed and won't be retried (it was
1946 a NOQUEUE, or has been canceled/unlocked); make a waiting
1947 lkb the first_lkid */
1949 r->res_first_lkid = 0;
1951 if (!list_empty(&r->res_lookup)) {
1952 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1954 list_del_init(&lkb->lkb_rsb_lookup);
1955 r->res_first_lkid = lkb->lkb_id;
1956 _request_lock(r, lkb);
1962 log_error(r->res_ls, "confirm_master unknown error %d", error);
1966 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
1967 int namelen, unsigned long timeout_cs, void *ast,
1968 void *astarg, void *bast, struct dlm_args *args)
1972 /* check for invalid arg usage */
1974 if (mode < 0 || mode > DLM_LOCK_EX)
1977 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1980 if (flags & DLM_LKF_CANCEL)
1983 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1986 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1989 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1992 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1995 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
1998 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2001 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2007 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2010 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2013 /* these args will be copied to the lkb in validate_lock_args,
2014 it cannot be done now because when converting locks, fields in
2015 an active lkb cannot be modified before locking the rsb */
2017 args->flags = flags;
2018 args->astaddr = ast;
2019 args->astparam = (long) astarg;
2020 args->bastaddr = bast;
2021 args->timeout = timeout_cs;
2029 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2031 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2032 DLM_LKF_FORCEUNLOCK))
2035 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2038 args->flags = flags;
2039 args->astparam = (long) astarg;
2043 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2044 struct dlm_args *args)
2048 if (args->flags & DLM_LKF_CONVERT) {
2049 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2052 if (args->flags & DLM_LKF_QUECVT &&
2053 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2057 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2060 if (lkb->lkb_wait_type)
2063 if (is_overlap(lkb))
2067 lkb->lkb_exflags = args->flags;
2068 lkb->lkb_sbflags = 0;
2069 lkb->lkb_astaddr = args->astaddr;
2070 lkb->lkb_astparam = args->astparam;
2071 lkb->lkb_bastaddr = args->bastaddr;
2072 lkb->lkb_rqmode = args->mode;
2073 lkb->lkb_lksb = args->lksb;
2074 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2075 lkb->lkb_ownpid = (int) current->pid;
2076 lkb->lkb_timeout_cs = args->timeout;
2082 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2085 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2086 because there may be a lookup in progress and it's valid to do
2087 cancel/unlockf on it */
2089 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2091 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2094 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2095 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2100 /* an lkb may still exist even though the lock is EOL'ed due to a
2101 cancel, unlock or failed noqueue request; an app can't use these
2102 locks; return same error as if the lkid had not been found at all */
2104 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2105 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2110 /* an lkb may be waiting for an rsb lookup to complete where the
2111 lookup was initiated by another lock */
2113 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2114 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2115 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2116 list_del_init(&lkb->lkb_rsb_lookup);
2117 queue_cast(lkb->lkb_resource, lkb,
2118 args->flags & DLM_LKF_CANCEL ?
2119 -DLM_ECANCEL : -DLM_EUNLOCK);
2120 unhold_lkb(lkb); /* undoes create_lkb() */
2122 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2127 /* cancel not allowed with another cancel/unlock in progress */
2129 if (args->flags & DLM_LKF_CANCEL) {
2130 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2133 if (is_overlap(lkb))
2136 /* don't let scand try to do a cancel */
2139 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2140 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2145 switch (lkb->lkb_wait_type) {
2146 case DLM_MSG_LOOKUP:
2147 case DLM_MSG_REQUEST:
2148 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2151 case DLM_MSG_UNLOCK:
2152 case DLM_MSG_CANCEL:
2155 /* add_to_waiters() will set OVERLAP_CANCEL */
2159 /* do we need to allow a force-unlock if there's a normal unlock
2160 already in progress? in what conditions could the normal unlock
2161 fail such that we'd want to send a force-unlock to be sure? */
2163 if (args->flags & DLM_LKF_FORCEUNLOCK) {
2164 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2167 if (is_overlap_unlock(lkb))
2170 /* don't let scand try to do a cancel */
2173 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2174 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2179 switch (lkb->lkb_wait_type) {
2180 case DLM_MSG_LOOKUP:
2181 case DLM_MSG_REQUEST:
2182 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2185 case DLM_MSG_UNLOCK:
2188 /* add_to_waiters() will set OVERLAP_UNLOCK */
2192 /* normal unlock not allowed if there's any op in progress */
2194 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
2198 /* an overlapping op shouldn't blow away exflags from other op */
2199 lkb->lkb_exflags |= args->flags;
2200 lkb->lkb_sbflags = 0;
2201 lkb->lkb_astparam = args->astparam;
2205 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
2206 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
2207 args->flags, lkb->lkb_wait_type,
2208 lkb->lkb_resource->res_name);
2213 * Four stage 4 varieties:
2214 * do_request(), do_convert(), do_unlock(), do_cancel()
2215 * These are called on the master node for the given lock and
2216 * from the central locking logic.
2219 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2223 if (can_be_granted(r, lkb, 1, NULL)) {
2225 queue_cast(r, lkb, 0);
2229 if (can_be_queued(lkb)) {
2230 error = -EINPROGRESS;
2231 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2232 send_blocking_asts(r, lkb);
2238 if (force_blocking_asts(lkb))
2239 send_blocking_asts_all(r, lkb);
2240 queue_cast(r, lkb, -EAGAIN);
2246 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2251 /* changing an existing lock may allow others to be granted */
2253 if (can_be_granted(r, lkb, 1, &deadlk)) {
2255 queue_cast(r, lkb, 0);
2256 grant_pending_locks(r);
2260 /* can_be_granted() detected that this lock would block in a conversion
2261 deadlock, so we leave it on the granted queue and return EDEADLK in
2262 the ast for the convert. */
2265 /* it's left on the granted queue */
2266 log_debug(r->res_ls, "deadlock %x node %d sts%d g%d r%d %s",
2267 lkb->lkb_id, lkb->lkb_nodeid, lkb->lkb_status,
2268 lkb->lkb_grmode, lkb->lkb_rqmode, r->res_name);
2269 revert_lock(r, lkb);
2270 queue_cast(r, lkb, -EDEADLK);
2275 /* is_demoted() means the can_be_granted() above set the grmode
2276 to NL, and left us on the granted queue. This auto-demotion
2277 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2278 now grantable. We have to try to grant other converting locks
2279 before we try again to grant this one. */
2281 if (is_demoted(lkb)) {
2282 grant_pending_convert(r, DLM_LOCK_IV, NULL);
2283 if (_can_be_granted(r, lkb, 1)) {
2285 queue_cast(r, lkb, 0);
2286 grant_pending_locks(r);
2289 /* else fall through and move to convert queue */
2292 if (can_be_queued(lkb)) {
2293 error = -EINPROGRESS;
2295 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2296 send_blocking_asts(r, lkb);
2302 if (force_blocking_asts(lkb))
2303 send_blocking_asts_all(r, lkb);
2304 queue_cast(r, lkb, -EAGAIN);
2310 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2312 remove_lock(r, lkb);
2313 queue_cast(r, lkb, -DLM_EUNLOCK);
2314 grant_pending_locks(r);
2315 return -DLM_EUNLOCK;
2318 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
2320 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2324 error = revert_lock(r, lkb);
2326 queue_cast(r, lkb, -DLM_ECANCEL);
2327 grant_pending_locks(r);
2328 return -DLM_ECANCEL;
2334 * Four stage 3 varieties:
2335 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2338 /* add a new lkb to a possibly new rsb, called by requesting process */
2340 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2344 /* set_master: sets lkb nodeid from r */
2346 error = set_master(r, lkb);
2355 /* receive_request() calls do_request() on remote node */
2356 error = send_request(r, lkb);
2358 error = do_request(r, lkb);
2363 /* change some property of an existing lkb, e.g. mode */
2365 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2370 /* receive_convert() calls do_convert() on remote node */
2371 error = send_convert(r, lkb);
2373 error = do_convert(r, lkb);
2378 /* remove an existing lkb from the granted queue */
2380 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2385 /* receive_unlock() calls do_unlock() on remote node */
2386 error = send_unlock(r, lkb);
2388 error = do_unlock(r, lkb);
2393 /* remove an existing lkb from the convert or wait queue */
2395 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2400 /* receive_cancel() calls do_cancel() on remote node */
2401 error = send_cancel(r, lkb);
2403 error = do_cancel(r, lkb);
2409 * Four stage 2 varieties:
2410 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2413 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
2414 int len, struct dlm_args *args)
2419 error = validate_lock_args(ls, lkb, args);
2423 error = find_rsb(ls, name, len, R_CREATE, &r);
2430 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
2432 error = _request_lock(r, lkb);
2441 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2442 struct dlm_args *args)
2447 r = lkb->lkb_resource;
2452 error = validate_lock_args(ls, lkb, args);
2456 error = _convert_lock(r, lkb);
2463 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2464 struct dlm_args *args)
2469 r = lkb->lkb_resource;
2474 error = validate_unlock_args(lkb, args);
2478 error = _unlock_lock(r, lkb);
2485 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2486 struct dlm_args *args)
2491 r = lkb->lkb_resource;
2496 error = validate_unlock_args(lkb, args);
2500 error = _cancel_lock(r, lkb);
2508 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2511 int dlm_lock(dlm_lockspace_t *lockspace,
2513 struct dlm_lksb *lksb,
2516 unsigned int namelen,
2517 uint32_t parent_lkid,
2518 void (*ast) (void *astarg),
2520 void (*bast) (void *astarg, int mode))
2523 struct dlm_lkb *lkb;
2524 struct dlm_args args;
2525 int error, convert = flags & DLM_LKF_CONVERT;
2527 ls = dlm_find_lockspace_local(lockspace);
2531 dlm_lock_recovery(ls);
2534 error = find_lkb(ls, lksb->sb_lkid, &lkb);
2536 error = create_lkb(ls, &lkb);
2541 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
2542 astarg, bast, &args);
2547 error = convert_lock(ls, lkb, &args);
2549 error = request_lock(ls, lkb, name, namelen, &args);
2551 if (error == -EINPROGRESS)
2554 if (convert || error)
2556 if (error == -EAGAIN || error == -EDEADLK)
2559 dlm_unlock_recovery(ls);
2560 dlm_put_lockspace(ls);
2564 int dlm_unlock(dlm_lockspace_t *lockspace,
2567 struct dlm_lksb *lksb,
2571 struct dlm_lkb *lkb;
2572 struct dlm_args args;
2575 ls = dlm_find_lockspace_local(lockspace);
2579 dlm_lock_recovery(ls);
2581 error = find_lkb(ls, lkid, &lkb);
2585 error = set_unlock_args(flags, astarg, &args);
2589 if (flags & DLM_LKF_CANCEL)
2590 error = cancel_lock(ls, lkb, &args);
2592 error = unlock_lock(ls, lkb, &args);
2594 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2596 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
2601 dlm_unlock_recovery(ls);
2602 dlm_put_lockspace(ls);
2607 * send/receive routines for remote operations and replies
2611 * send_request receive_request
2612 * send_convert receive_convert
2613 * send_unlock receive_unlock
2614 * send_cancel receive_cancel
2615 * send_grant receive_grant
2616 * send_bast receive_bast
2617 * send_lookup receive_lookup
2618 * send_remove receive_remove
2621 * receive_request_reply send_request_reply
2622 * receive_convert_reply send_convert_reply
2623 * receive_unlock_reply send_unlock_reply
2624 * receive_cancel_reply send_cancel_reply
2625 * receive_lookup_reply send_lookup_reply
2628 static int _create_message(struct dlm_ls *ls, int mb_len,
2629 int to_nodeid, int mstype,
2630 struct dlm_message **ms_ret,
2631 struct dlm_mhandle **mh_ret)
2633 struct dlm_message *ms;
2634 struct dlm_mhandle *mh;
2637 /* get_buffer gives us a message handle (mh) that we need to
2638 pass into lowcomms_commit and a message buffer (mb) that we
2639 write our data into */
2641 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
2645 memset(mb, 0, mb_len);
2647 ms = (struct dlm_message *) mb;
2649 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2650 ms->m_header.h_lockspace = ls->ls_global_id;
2651 ms->m_header.h_nodeid = dlm_our_nodeid();
2652 ms->m_header.h_length = mb_len;
2653 ms->m_header.h_cmd = DLM_MSG;
2655 ms->m_type = mstype;
2662 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2663 int to_nodeid, int mstype,
2664 struct dlm_message **ms_ret,
2665 struct dlm_mhandle **mh_ret)
2667 int mb_len = sizeof(struct dlm_message);
2670 case DLM_MSG_REQUEST:
2671 case DLM_MSG_LOOKUP:
2672 case DLM_MSG_REMOVE:
2673 mb_len += r->res_length;
2675 case DLM_MSG_CONVERT:
2676 case DLM_MSG_UNLOCK:
2677 case DLM_MSG_REQUEST_REPLY:
2678 case DLM_MSG_CONVERT_REPLY:
2680 if (lkb && lkb->lkb_lvbptr)
2681 mb_len += r->res_ls->ls_lvblen;
2685 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
2689 /* further lowcomms enhancements or alternate implementations may make
2690 the return value from this function useful at some point */
2692 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2694 dlm_message_out(ms);
2695 dlm_lowcomms_commit_buffer(mh);
2699 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2700 struct dlm_message *ms)
2702 ms->m_nodeid = lkb->lkb_nodeid;
2703 ms->m_pid = lkb->lkb_ownpid;
2704 ms->m_lkid = lkb->lkb_id;
2705 ms->m_remid = lkb->lkb_remid;
2706 ms->m_exflags = lkb->lkb_exflags;
2707 ms->m_sbflags = lkb->lkb_sbflags;
2708 ms->m_flags = lkb->lkb_flags;
2709 ms->m_lvbseq = lkb->lkb_lvbseq;
2710 ms->m_status = lkb->lkb_status;
2711 ms->m_grmode = lkb->lkb_grmode;
2712 ms->m_rqmode = lkb->lkb_rqmode;
2713 ms->m_hash = r->res_hash;
2715 /* m_result and m_bastmode are set from function args,
2716 not from lkb fields */
2718 if (lkb->lkb_bastaddr)
2719 ms->m_asts |= AST_BAST;
2720 if (lkb->lkb_astaddr)
2721 ms->m_asts |= AST_COMP;
2723 /* compare with switch in create_message; send_remove() doesn't
2726 switch (ms->m_type) {
2727 case DLM_MSG_REQUEST:
2728 case DLM_MSG_LOOKUP:
2729 memcpy(ms->m_extra, r->res_name, r->res_length);
2731 case DLM_MSG_CONVERT:
2732 case DLM_MSG_UNLOCK:
2733 case DLM_MSG_REQUEST_REPLY:
2734 case DLM_MSG_CONVERT_REPLY:
2736 if (!lkb->lkb_lvbptr)
2738 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2743 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2745 struct dlm_message *ms;
2746 struct dlm_mhandle *mh;
2747 int to_nodeid, error;
2749 error = add_to_waiters(lkb, mstype);
2753 to_nodeid = r->res_nodeid;
2755 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2759 send_args(r, lkb, ms);
2761 error = send_message(mh, ms);
2767 remove_from_waiters(lkb, msg_reply_type(mstype));
2771 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2773 return send_common(r, lkb, DLM_MSG_REQUEST);
2776 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2780 error = send_common(r, lkb, DLM_MSG_CONVERT);
2782 /* down conversions go without a reply from the master */
2783 if (!error && down_conversion(lkb)) {
2784 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
2785 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
2786 r->res_ls->ls_stub_ms.m_result = 0;
2787 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
2788 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2794 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2795 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2796 that the master is still correct. */
2798 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2800 return send_common(r, lkb, DLM_MSG_UNLOCK);
2803 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2805 return send_common(r, lkb, DLM_MSG_CANCEL);
2808 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2810 struct dlm_message *ms;
2811 struct dlm_mhandle *mh;
2812 int to_nodeid, error;
2814 to_nodeid = lkb->lkb_nodeid;
2816 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2820 send_args(r, lkb, ms);
2824 error = send_message(mh, ms);
2829 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2831 struct dlm_message *ms;
2832 struct dlm_mhandle *mh;
2833 int to_nodeid, error;
2835 to_nodeid = lkb->lkb_nodeid;
2837 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2841 send_args(r, lkb, ms);
2843 ms->m_bastmode = mode;
2845 error = send_message(mh, ms);
2850 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2852 struct dlm_message *ms;
2853 struct dlm_mhandle *mh;
2854 int to_nodeid, error;
2856 error = add_to_waiters(lkb, DLM_MSG_LOOKUP);
2860 to_nodeid = dlm_dir_nodeid(r);
2862 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2866 send_args(r, lkb, ms);
2868 error = send_message(mh, ms);
2874 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
2878 static int send_remove(struct dlm_rsb *r)
2880 struct dlm_message *ms;
2881 struct dlm_mhandle *mh;
2882 int to_nodeid, error;
2884 to_nodeid = dlm_dir_nodeid(r);
2886 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2890 memcpy(ms->m_extra, r->res_name, r->res_length);
2891 ms->m_hash = r->res_hash;
2893 error = send_message(mh, ms);
2898 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2901 struct dlm_message *ms;
2902 struct dlm_mhandle *mh;
2903 int to_nodeid, error;
2905 to_nodeid = lkb->lkb_nodeid;
2907 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2911 send_args(r, lkb, ms);
2915 error = send_message(mh, ms);
2920 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2922 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2925 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2927 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2930 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2932 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2935 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2937 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2940 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2941 int ret_nodeid, int rv)
2943 struct dlm_rsb *r = &ls->ls_stub_rsb;
2944 struct dlm_message *ms;
2945 struct dlm_mhandle *mh;
2946 int error, nodeid = ms_in->m_header.h_nodeid;
2948 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2952 ms->m_lkid = ms_in->m_lkid;
2954 ms->m_nodeid = ret_nodeid;
2956 error = send_message(mh, ms);
2961 /* which args we save from a received message depends heavily on the type
2962 of message, unlike the send side where we can safely send everything about
2963 the lkb for any type of message */
2965 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2967 lkb->lkb_exflags = ms->m_exflags;
2968 lkb->lkb_sbflags = ms->m_sbflags;
2969 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2970 (ms->m_flags & 0x0000FFFF);
2973 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2975 lkb->lkb_sbflags = ms->m_sbflags;
2976 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2977 (ms->m_flags & 0x0000FFFF);
2980 static int receive_extralen(struct dlm_message *ms)
2982 return (ms->m_header.h_length - sizeof(struct dlm_message));
2985 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2986 struct dlm_message *ms)
2990 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2991 if (!lkb->lkb_lvbptr)
2992 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
2993 if (!lkb->lkb_lvbptr)
2995 len = receive_extralen(ms);
2996 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3001 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3002 struct dlm_message *ms)
3004 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3005 lkb->lkb_ownpid = ms->m_pid;
3006 lkb->lkb_remid = ms->m_lkid;
3007 lkb->lkb_grmode = DLM_LOCK_IV;
3008 lkb->lkb_rqmode = ms->m_rqmode;
3009 lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
3010 lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
3012 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3013 /* lkb was just created so there won't be an lvb yet */
3014 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3015 if (!lkb->lkb_lvbptr)
3022 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3023 struct dlm_message *ms)
3025 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3028 if (receive_lvb(ls, lkb, ms))
3031 lkb->lkb_rqmode = ms->m_rqmode;
3032 lkb->lkb_lvbseq = ms->m_lvbseq;
3037 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3038 struct dlm_message *ms)
3040 if (receive_lvb(ls, lkb, ms))
3045 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3046 uses to send a reply and that the remote end uses to process the reply. */
3048 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3050 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3051 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3052 lkb->lkb_remid = ms->m_lkid;
3055 /* This is called after the rsb is locked so that we can safely inspect
3056 fields in the lkb. */
3058 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3060 int from = ms->m_header.h_nodeid;
3063 switch (ms->m_type) {
3064 case DLM_MSG_CONVERT:
3065 case DLM_MSG_UNLOCK:
3066 case DLM_MSG_CANCEL:
3067 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3071 case DLM_MSG_CONVERT_REPLY:
3072 case DLM_MSG_UNLOCK_REPLY:
3073 case DLM_MSG_CANCEL_REPLY:
3076 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3080 case DLM_MSG_REQUEST_REPLY:
3081 if (!is_process_copy(lkb))
3083 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3092 log_error(lkb->lkb_resource->res_ls,
3093 "ignore invalid message %d from %d %x %x %x %d",
3094 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid,
3095 lkb->lkb_flags, lkb->lkb_nodeid);
3099 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3101 struct dlm_lkb *lkb;
3105 error = create_lkb(ls, &lkb);
3109 receive_flags(lkb, ms);
3110 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3111 error = receive_request_args(ls, lkb, ms);
3117 namelen = receive_extralen(ms);
3119 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
3128 error = do_request(r, lkb);
3129 send_request_reply(r, lkb, error);
3134 if (error == -EINPROGRESS)
3141 setup_stub_lkb(ls, ms);
3142 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3145 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3147 struct dlm_lkb *lkb;
3149 int error, reply = 1;
3151 error = find_lkb(ls, ms->m_remid, &lkb);
3155 r = lkb->lkb_resource;
3160 error = validate_message(lkb, ms);
3164 receive_flags(lkb, ms);
3165 error = receive_convert_args(ls, lkb, ms);
3168 reply = !down_conversion(lkb);
3170 error = do_convert(r, lkb);
3173 send_convert_reply(r, lkb, error);
3181 setup_stub_lkb(ls, ms);
3182 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3185 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3187 struct dlm_lkb *lkb;
3191 error = find_lkb(ls, ms->m_remid, &lkb);
3195 r = lkb->lkb_resource;
3200 error = validate_message(lkb, ms);
3204 receive_flags(lkb, ms);
3205 error = receive_unlock_args(ls, lkb, ms);
3209 error = do_unlock(r, lkb);
3211 send_unlock_reply(r, lkb, error);
3219 setup_stub_lkb(ls, ms);
3220 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3223 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3225 struct dlm_lkb *lkb;
3229 error = find_lkb(ls, ms->m_remid, &lkb);
3233 receive_flags(lkb, ms);
3235 r = lkb->lkb_resource;
3240 error = validate_message(lkb, ms);
3244 error = do_cancel(r, lkb);
3245 send_cancel_reply(r, lkb, error);
3253 setup_stub_lkb(ls, ms);
3254 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3257 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
3259 struct dlm_lkb *lkb;
3263 error = find_lkb(ls, ms->m_remid, &lkb);
3265 log_debug(ls, "receive_grant from %d no lkb %x",
3266 ms->m_header.h_nodeid, ms->m_remid);
3270 r = lkb->lkb_resource;
3275 error = validate_message(lkb, ms);
3279 receive_flags_reply(lkb, ms);
3280 if (is_altmode(lkb))
3281 munge_altmode(lkb, ms);
3282 grant_lock_pc(r, lkb, ms);
3283 queue_cast(r, lkb, 0);
3290 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
3292 struct dlm_lkb *lkb;
3296 error = find_lkb(ls, ms->m_remid, &lkb);
3298 log_debug(ls, "receive_bast from %d no lkb %x",
3299 ms->m_header.h_nodeid, ms->m_remid);
3303 r = lkb->lkb_resource;
3308 error = validate_message(lkb, ms);
3312 queue_bast(r, lkb, ms->m_bastmode);
3319 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
3321 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
3323 from_nodeid = ms->m_header.h_nodeid;
3324 our_nodeid = dlm_our_nodeid();
3326 len = receive_extralen(ms);
3328 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3329 if (dir_nodeid != our_nodeid) {
3330 log_error(ls, "lookup dir_nodeid %d from %d",
3331 dir_nodeid, from_nodeid);
3337 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
3339 /* Optimization: we're master so treat lookup as a request */
3340 if (!error && ret_nodeid == our_nodeid) {
3341 receive_request(ls, ms);
3345 send_lookup_reply(ls, ms, ret_nodeid, error);
3348 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
3350 int len, dir_nodeid, from_nodeid;
3352 from_nodeid = ms->m_header.h_nodeid;
3354 len = receive_extralen(ms);
3356 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3357 if (dir_nodeid != dlm_our_nodeid()) {
3358 log_error(ls, "remove dir entry dir_nodeid %d from %d",
3359 dir_nodeid, from_nodeid);
3363 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
3366 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
3368 do_purge(ls, ms->m_nodeid, ms->m_pid);
3371 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
3373 struct dlm_lkb *lkb;
3375 int error, mstype, result;
3377 error = find_lkb(ls, ms->m_remid, &lkb);
3379 log_debug(ls, "receive_request_reply from %d no lkb %x",
3380 ms->m_header.h_nodeid, ms->m_remid);
3384 r = lkb->lkb_resource;
3388 error = validate_message(lkb, ms);
3392 mstype = lkb->lkb_wait_type;
3393 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
3397 /* Optimization: the dir node was also the master, so it took our
3398 lookup as a request and sent request reply instead of lookup reply */
3399 if (mstype == DLM_MSG_LOOKUP) {
3400 r->res_nodeid = ms->m_header.h_nodeid;
3401 lkb->lkb_nodeid = r->res_nodeid;
3404 /* this is the value returned from do_request() on the master */
3405 result = ms->m_result;
3409 /* request would block (be queued) on remote master */
3410 queue_cast(r, lkb, -EAGAIN);
3411 confirm_master(r, -EAGAIN);
3412 unhold_lkb(lkb); /* undoes create_lkb() */
3417 /* request was queued or granted on remote master */
3418 receive_flags_reply(lkb, ms);
3419 lkb->lkb_remid = ms->m_lkid;
3420 if (is_altmode(lkb))
3421 munge_altmode(lkb, ms);
3423 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3426 grant_lock_pc(r, lkb, ms);
3427 queue_cast(r, lkb, 0);
3429 confirm_master(r, result);
3434 /* find_rsb failed to find rsb or rsb wasn't master */
3435 log_debug(ls, "receive_request_reply %x %x master diff %d %d",
3436 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result);
3438 lkb->lkb_nodeid = -1;
3440 if (is_overlap(lkb)) {
3441 /* we'll ignore error in cancel/unlock reply */
3442 queue_cast_overlap(r, lkb);
3443 confirm_master(r, result);
3444 unhold_lkb(lkb); /* undoes create_lkb() */
3446 _request_lock(r, lkb);
3450 log_error(ls, "receive_request_reply %x error %d",
3451 lkb->lkb_id, result);
3454 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
3455 log_debug(ls, "receive_request_reply %x result %d unlock",
3456 lkb->lkb_id, result);
3457 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3458 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3459 send_unlock(r, lkb);
3460 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
3461 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
3462 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3463 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3464 send_cancel(r, lkb);
3466 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3467 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3475 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3476 struct dlm_message *ms)
3478 /* this is the value returned from do_convert() on the master */
3479 switch (ms->m_result) {
3481 /* convert would block (be queued) on remote master */
3482 queue_cast(r, lkb, -EAGAIN);
3486 receive_flags_reply(lkb, ms);
3487 revert_lock_pc(r, lkb);
3488 queue_cast(r, lkb, -EDEADLK);
3492 /* convert was queued on remote master */
3493 receive_flags_reply(lkb, ms);
3494 if (is_demoted(lkb))
3495 munge_demoted(lkb, ms);
3497 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3502 /* convert was granted on remote master */
3503 receive_flags_reply(lkb, ms);
3504 if (is_demoted(lkb))
3505 munge_demoted(lkb, ms);
3506 grant_lock_pc(r, lkb, ms);
3507 queue_cast(r, lkb, 0);
3511 log_error(r->res_ls, "receive_convert_reply %x error %d",
3512 lkb->lkb_id, ms->m_result);
3516 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3518 struct dlm_rsb *r = lkb->lkb_resource;
3524 error = validate_message(lkb, ms);
3528 /* stub reply can happen with waiters_mutex held */
3529 error = remove_from_waiters_ms(lkb, ms);
3533 __receive_convert_reply(r, lkb, ms);
3539 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
3541 struct dlm_lkb *lkb;
3544 error = find_lkb(ls, ms->m_remid, &lkb);
3546 log_debug(ls, "receive_convert_reply from %d no lkb %x",
3547 ms->m_header.h_nodeid, ms->m_remid);
3551 _receive_convert_reply(lkb, ms);
3555 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3557 struct dlm_rsb *r = lkb->lkb_resource;
3563 error = validate_message(lkb, ms);
3567 /* stub reply can happen with waiters_mutex held */
3568 error = remove_from_waiters_ms(lkb, ms);
3572 /* this is the value returned from do_unlock() on the master */
3574 switch (ms->m_result) {
3576 receive_flags_reply(lkb, ms);
3577 remove_lock_pc(r, lkb);
3578 queue_cast(r, lkb, -DLM_EUNLOCK);
3583 log_error(r->res_ls, "receive_unlock_reply %x error %d",
3584 lkb->lkb_id, ms->m_result);
3591 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
3593 struct dlm_lkb *lkb;
3596 error = find_lkb(ls, ms->m_remid, &lkb);
3598 log_debug(ls, "receive_unlock_reply from %d no lkb %x",
3599 ms->m_header.h_nodeid, ms->m_remid);
3603 _receive_unlock_reply(lkb, ms);
3607 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3609 struct dlm_rsb *r = lkb->lkb_resource;
3615 error = validate_message(lkb, ms);
3619 /* stub reply can happen with waiters_mutex held */
3620 error = remove_from_waiters_ms(lkb, ms);
3624 /* this is the value returned from do_cancel() on the master */
3626 switch (ms->m_result) {
3628 receive_flags_reply(lkb, ms);
3629 revert_lock_pc(r, lkb);
3630 queue_cast(r, lkb, -DLM_ECANCEL);
3635 log_error(r->res_ls, "receive_cancel_reply %x error %d",
3636 lkb->lkb_id, ms->m_result);
3643 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
3645 struct dlm_lkb *lkb;
3648 error = find_lkb(ls, ms->m_remid, &lkb);
3650 log_debug(ls, "receive_cancel_reply from %d no lkb %x",
3651 ms->m_header.h_nodeid, ms->m_remid);
3655 _receive_cancel_reply(lkb, ms);
3659 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
3661 struct dlm_lkb *lkb;
3663 int error, ret_nodeid;
3665 error = find_lkb(ls, ms->m_lkid, &lkb);
3667 log_error(ls, "receive_lookup_reply no lkb");
3671 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
3672 FIXME: will a non-zero error ever be returned? */
3674 r = lkb->lkb_resource;
3678 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3682 ret_nodeid = ms->m_nodeid;
3683 if (ret_nodeid == dlm_our_nodeid()) {
3686 r->res_first_lkid = 0;
3688 /* set_master() will copy res_nodeid to lkb_nodeid */
3689 r->res_nodeid = ret_nodeid;
3692 if (is_overlap(lkb)) {
3693 log_debug(ls, "receive_lookup_reply %x unlock %x",
3694 lkb->lkb_id, lkb->lkb_flags);
3695 queue_cast_overlap(r, lkb);
3696 unhold_lkb(lkb); /* undoes create_lkb() */
3700 _request_lock(r, lkb);
3704 process_lookup_list(r);
3711 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms)
3713 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) {
3714 log_debug(ls, "ignore non-member message %d from %d %x %x %d",
3715 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid,
3716 ms->m_remid, ms->m_result);
3720 switch (ms->m_type) {
3722 /* messages sent to a master node */
3724 case DLM_MSG_REQUEST:
3725 receive_request(ls, ms);
3728 case DLM_MSG_CONVERT:
3729 receive_convert(ls, ms);
3732 case DLM_MSG_UNLOCK:
3733 receive_unlock(ls, ms);
3736 case DLM_MSG_CANCEL:
3737 receive_cancel(ls, ms);
3740 /* messages sent from a master node (replies to above) */
3742 case DLM_MSG_REQUEST_REPLY:
3743 receive_request_reply(ls, ms);
3746 case DLM_MSG_CONVERT_REPLY:
3747 receive_convert_reply(ls, ms);
3750 case DLM_MSG_UNLOCK_REPLY:
3751 receive_unlock_reply(ls, ms);
3754 case DLM_MSG_CANCEL_REPLY:
3755 receive_cancel_reply(ls, ms);
3758 /* messages sent from a master node (only two types of async msg) */
3761 receive_grant(ls, ms);
3765 receive_bast(ls, ms);
3768 /* messages sent to a dir node */
3770 case DLM_MSG_LOOKUP:
3771 receive_lookup(ls, ms);
3774 case DLM_MSG_REMOVE:
3775 receive_remove(ls, ms);
3778 /* messages sent from a dir node (remove has no reply) */
3780 case DLM_MSG_LOOKUP_REPLY:
3781 receive_lookup_reply(ls, ms);
3784 /* other messages */
3787 receive_purge(ls, ms);
3791 log_error(ls, "unknown message type %d", ms->m_type);
3797 /* If the lockspace is in recovery mode (locking stopped), then normal
3798 messages are saved on the requestqueue for processing after recovery is
3799 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
3800 messages off the requestqueue before we process new ones. This occurs right
3801 after recovery completes when we transition from saving all messages on
3802 requestqueue, to processing all the saved messages, to processing new
3803 messages as they arrive. */
3805 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
3808 if (dlm_locking_stopped(ls)) {
3809 dlm_add_requestqueue(ls, nodeid, ms);
3811 dlm_wait_requestqueue(ls);
3812 _receive_message(ls, ms);
3816 /* This is called by dlm_recoverd to process messages that were saved on
3817 the requestqueue. */
3819 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms)
3821 _receive_message(ls, ms);
3824 /* This is called by the midcomms layer when something is received for
3825 the lockspace. It could be either a MSG (normal message sent as part of
3826 standard locking activity) or an RCOM (recovery message sent as part of
3827 lockspace recovery). */
3829 void dlm_receive_buffer(union dlm_packet *p, int nodeid)
3831 struct dlm_header *hd = &p->header;
3835 switch (hd->h_cmd) {
3837 dlm_message_in(&p->message);
3838 type = p->message.m_type;
3841 dlm_rcom_in(&p->rcom);
3842 type = p->rcom.rc_type;
3845 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
3849 if (hd->h_nodeid != nodeid) {
3850 log_print("invalid h_nodeid %d from %d lockspace %x",
3851 hd->h_nodeid, nodeid, hd->h_lockspace);
3855 ls = dlm_find_lockspace_global(hd->h_lockspace);
3857 if (dlm_config.ci_log_debug)
3858 log_print("invalid lockspace %x from %d cmd %d type %d",
3859 hd->h_lockspace, nodeid, hd->h_cmd, type);
3861 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
3862 dlm_send_ls_not_ready(nodeid, &p->rcom);
3866 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
3867 be inactive (in this ls) before transitioning to recovery mode */
3869 down_read(&ls->ls_recv_active);
3870 if (hd->h_cmd == DLM_MSG)
3871 dlm_receive_message(ls, &p->message, nodeid);
3873 dlm_receive_rcom(ls, &p->rcom, nodeid);
3874 up_read(&ls->ls_recv_active);
3876 dlm_put_lockspace(ls);
3879 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3881 if (middle_conversion(lkb)) {
3883 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3884 ls->ls_stub_ms.m_result = -EINPROGRESS;
3885 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3886 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
3887 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3889 /* Same special case as in receive_rcom_lock_args() */
3890 lkb->lkb_grmode = DLM_LOCK_IV;
3891 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3894 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3895 lkb->lkb_flags |= DLM_IFL_RESEND;
3898 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3899 conversions are async; there's no reply from the remote master */
3902 /* A waiting lkb needs recovery if the master node has failed, or
3903 the master node is changing (only when no directory is used) */
3905 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3907 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3910 if (!dlm_no_directory(ls))
3913 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3919 /* Recovery for locks that are waiting for replies from nodes that are now
3920 gone. We can just complete unlocks and cancels by faking a reply from the
3921 dead node. Requests and up-conversions we flag to be resent after
3922 recovery. Down-conversions can just be completed with a fake reply like
3923 unlocks. Conversions between PR and CW need special attention. */
3925 void dlm_recover_waiters_pre(struct dlm_ls *ls)
3927 struct dlm_lkb *lkb, *safe;
3928 int wait_type, stub_unlock_result, stub_cancel_result;
3930 mutex_lock(&ls->ls_waiters_mutex);
3932 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3933 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3934 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3936 /* all outstanding lookups, regardless of destination will be
3937 resent after recovery is done */
3939 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3940 lkb->lkb_flags |= DLM_IFL_RESEND;
3944 if (!waiter_needs_recovery(ls, lkb))
3947 wait_type = lkb->lkb_wait_type;
3948 stub_unlock_result = -DLM_EUNLOCK;
3949 stub_cancel_result = -DLM_ECANCEL;
3951 /* Main reply may have been received leaving a zero wait_type,
3952 but a reply for the overlapping op may not have been
3953 received. In that case we need to fake the appropriate
3954 reply for the overlap op. */
3957 if (is_overlap_cancel(lkb)) {
3958 wait_type = DLM_MSG_CANCEL;
3959 if (lkb->lkb_grmode == DLM_LOCK_IV)
3960 stub_cancel_result = 0;
3962 if (is_overlap_unlock(lkb)) {
3963 wait_type = DLM_MSG_UNLOCK;
3964 if (lkb->lkb_grmode == DLM_LOCK_IV)
3965 stub_unlock_result = -ENOENT;
3968 log_debug(ls, "rwpre overlap %x %x %d %d %d",
3969 lkb->lkb_id, lkb->lkb_flags, wait_type,
3970 stub_cancel_result, stub_unlock_result);
3973 switch (wait_type) {
3975 case DLM_MSG_REQUEST:
3976 lkb->lkb_flags |= DLM_IFL_RESEND;
3979 case DLM_MSG_CONVERT:
3980 recover_convert_waiter(ls, lkb);
3983 case DLM_MSG_UNLOCK:
3985 ls->ls_stub_ms.m_type = DLM_MSG_UNLOCK_REPLY;
3986 ls->ls_stub_ms.m_result = stub_unlock_result;
3987 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3988 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
3989 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
3993 case DLM_MSG_CANCEL:
3995 ls->ls_stub_ms.m_type = DLM_MSG_CANCEL_REPLY;
3996 ls->ls_stub_ms.m_result = stub_cancel_result;
3997 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3998 ls->ls_stub_ms.m_header.h_nodeid = lkb->lkb_nodeid;
3999 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
4004 log_error(ls, "invalid lkb wait_type %d %d",
4005 lkb->lkb_wait_type, wait_type);
4009 mutex_unlock(&ls->ls_waiters_mutex);
4012 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
4014 struct dlm_lkb *lkb;
4017 mutex_lock(&ls->ls_waiters_mutex);
4018 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
4019 if (lkb->lkb_flags & DLM_IFL_RESEND) {
4025 mutex_unlock(&ls->ls_waiters_mutex);
4032 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
4033 master or dir-node for r. Processing the lkb may result in it being placed
4036 /* We do this after normal locking has been enabled and any saved messages
4037 (in requestqueue) have been processed. We should be confident that at
4038 this point we won't get or process a reply to any of these waiting
4039 operations. But, new ops may be coming in on the rsbs/locks here from
4040 userspace or remotely. */
4042 /* there may have been an overlap unlock/cancel prior to recovery or after
4043 recovery. if before, the lkb may still have a pos wait_count; if after, the
4044 overlap flag would just have been set and nothing new sent. we can be
4045 confident here than any replies to either the initial op or overlap ops
4046 prior to recovery have been received. */
4048 int dlm_recover_waiters_post(struct dlm_ls *ls)
4050 struct dlm_lkb *lkb;
4052 int error = 0, mstype, err, oc, ou;
4055 if (dlm_locking_stopped(ls)) {
4056 log_debug(ls, "recover_waiters_post aborted");
4061 lkb = find_resend_waiter(ls);
4065 r = lkb->lkb_resource;
4069 mstype = lkb->lkb_wait_type;
4070 oc = is_overlap_cancel(lkb);
4071 ou = is_overlap_unlock(lkb);
4074 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
4075 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
4077 /* At this point we assume that we won't get a reply to any
4078 previous op or overlap op on this lock. First, do a big
4079 remove_from_waiters() for all previous ops. */
4081 lkb->lkb_flags &= ~DLM_IFL_RESEND;
4082 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4083 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4084 lkb->lkb_wait_type = 0;
4085 lkb->lkb_wait_count = 0;
4086 mutex_lock(&ls->ls_waiters_mutex);
4087 list_del_init(&lkb->lkb_wait_reply);
4088 mutex_unlock(&ls->ls_waiters_mutex);
4089 unhold_lkb(lkb); /* for waiters list */
4092 /* do an unlock or cancel instead of resending */
4094 case DLM_MSG_LOOKUP:
4095 case DLM_MSG_REQUEST:
4096 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
4098 unhold_lkb(lkb); /* undoes create_lkb() */
4100 case DLM_MSG_CONVERT:
4102 queue_cast(r, lkb, -DLM_ECANCEL);
4104 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
4105 _unlock_lock(r, lkb);
4113 case DLM_MSG_LOOKUP:
4114 case DLM_MSG_REQUEST:
4115 _request_lock(r, lkb);
4117 confirm_master(r, 0);
4119 case DLM_MSG_CONVERT:
4120 _convert_lock(r, lkb);
4128 log_error(ls, "recover_waiters_post %x %d %x %d %d",
4129 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou);
4138 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
4139 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
4141 struct dlm_ls *ls = r->res_ls;
4142 struct dlm_lkb *lkb, *safe;
4144 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
4145 if (test(ls, lkb)) {
4146 rsb_set_flag(r, RSB_LOCKS_PURGED);
4148 /* this put should free the lkb */
4149 if (!dlm_put_lkb(lkb))
4150 log_error(ls, "purged lkb not released");
4155 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4157 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
4160 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
4162 return is_master_copy(lkb);
4165 static void purge_dead_locks(struct dlm_rsb *r)
4167 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
4168 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
4169 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
4172 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
4174 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
4175 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
4176 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
4179 /* Get rid of locks held by nodes that are gone. */
4181 int dlm_purge_locks(struct dlm_ls *ls)
4185 log_debug(ls, "dlm_purge_locks");
4187 down_write(&ls->ls_root_sem);
4188 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
4192 purge_dead_locks(r);
4198 up_write(&ls->ls_root_sem);
4203 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
4205 struct dlm_rsb *r, *r_ret = NULL;
4207 read_lock(&ls->ls_rsbtbl[bucket].lock);
4208 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
4209 if (!rsb_flag(r, RSB_LOCKS_PURGED))
4212 rsb_clear_flag(r, RSB_LOCKS_PURGED);
4216 read_unlock(&ls->ls_rsbtbl[bucket].lock);
4220 void dlm_grant_after_purge(struct dlm_ls *ls)
4226 r = find_purged_rsb(ls, bucket);
4228 if (bucket == ls->ls_rsbtbl_size - 1)
4235 grant_pending_locks(r);
4236 confirm_master(r, 0);
4244 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
4247 struct dlm_lkb *lkb;
4249 list_for_each_entry(lkb, head, lkb_statequeue) {
4250 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
4256 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
4259 struct dlm_lkb *lkb;
4261 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
4264 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
4267 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
4273 /* needs at least dlm_rcom + rcom_lock */
4274 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
4275 struct dlm_rsb *r, struct dlm_rcom *rc)
4277 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4279 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
4280 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
4281 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
4282 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
4283 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
4284 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4285 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
4286 lkb->lkb_rqmode = rl->rl_rqmode;
4287 lkb->lkb_grmode = rl->rl_grmode;
4288 /* don't set lkb_status because add_lkb wants to itself */
4290 lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
4291 lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
4293 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
4294 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
4295 sizeof(struct rcom_lock);
4296 if (lvblen > ls->ls_lvblen)
4298 lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
4299 if (!lkb->lkb_lvbptr)
4301 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
4304 /* Conversions between PR and CW (middle modes) need special handling.
4305 The real granted mode of these converting locks cannot be determined
4306 until all locks have been rebuilt on the rsb (recover_conversion) */
4308 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
4309 middle_conversion(lkb)) {
4310 rl->rl_status = DLM_LKSTS_CONVERT;
4311 lkb->lkb_grmode = DLM_LOCK_IV;
4312 rsb_set_flag(r, RSB_RECOVER_CONVERT);
4318 /* This lkb may have been recovered in a previous aborted recovery so we need
4319 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4320 If so we just send back a standard reply. If not, we create a new lkb with
4321 the given values and send back our lkid. We send back our lkid by sending
4322 back the rcom_lock struct we got but with the remid field filled in. */
4324 /* needs at least dlm_rcom + rcom_lock */
4325 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4327 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4329 struct dlm_lkb *lkb;
4332 if (rl->rl_parent_lkid) {
4333 error = -EOPNOTSUPP;
4337 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
4344 lkb = search_remid(r, rc->rc_header.h_nodeid, le32_to_cpu(rl->rl_lkid));
4350 error = create_lkb(ls, &lkb);
4354 error = receive_rcom_lock_args(ls, lkb, r, rc);
4361 add_lkb(r, lkb, rl->rl_status);
4365 /* this is the new value returned to the lock holder for
4366 saving in its process-copy lkb */
4367 rl->rl_remid = cpu_to_le32(lkb->lkb_id);
4374 log_debug(ls, "recover_master_copy %d %x", error,
4375 le32_to_cpu(rl->rl_lkid));
4376 rl->rl_result = cpu_to_le32(error);
4380 /* needs at least dlm_rcom + rcom_lock */
4381 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4383 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4385 struct dlm_lkb *lkb;
4388 error = find_lkb(ls, le32_to_cpu(rl->rl_lkid), &lkb);
4390 log_error(ls, "recover_process_copy no lkid %x",
4391 le32_to_cpu(rl->rl_lkid));
4395 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
4397 error = le32_to_cpu(rl->rl_result);
4399 r = lkb->lkb_resource;
4405 /* There's a chance the new master received our lock before
4406 dlm_recover_master_reply(), this wouldn't happen if we did
4407 a barrier between recover_masters and recover_locks. */
4408 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
4409 (unsigned long)r, r->res_name);
4410 dlm_send_rcom_lock(r, lkb);
4413 log_debug(ls, "master copy exists %x", lkb->lkb_id);
4416 lkb->lkb_remid = le32_to_cpu(rl->rl_remid);
4419 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
4420 error, lkb->lkb_id);
4423 /* an ack for dlm_recover_locks() which waits for replies from
4424 all the locks it sends to new masters */
4425 dlm_recovered_lock(r);
4434 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
4435 int mode, uint32_t flags, void *name, unsigned int namelen,
4436 unsigned long timeout_cs)
4438 struct dlm_lkb *lkb;
4439 struct dlm_args args;
4442 dlm_lock_recovery(ls);
4444 error = create_lkb(ls, &lkb);
4450 if (flags & DLM_LKF_VALBLK) {
4451 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
4452 if (!ua->lksb.sb_lvbptr) {
4460 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
4461 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4462 lock and that lkb_astparam is the dlm_user_args structure. */
4464 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
4465 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
4466 lkb->lkb_flags |= DLM_IFL_USER;
4467 ua->old_mode = DLM_LOCK_IV;
4474 error = request_lock(ls, lkb, name, namelen, &args);
4490 /* add this new lkb to the per-process list of locks */
4491 spin_lock(&ua->proc->locks_spin);
4493 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
4494 spin_unlock(&ua->proc->locks_spin);
4496 dlm_unlock_recovery(ls);
4500 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4501 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
4502 unsigned long timeout_cs)
4504 struct dlm_lkb *lkb;
4505 struct dlm_args args;
4506 struct dlm_user_args *ua;
4509 dlm_lock_recovery(ls);
4511 error = find_lkb(ls, lkid, &lkb);
4515 /* user can change the params on its lock when it converts it, or
4516 add an lvb that didn't exist before */
4518 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4520 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
4521 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
4522 if (!ua->lksb.sb_lvbptr) {
4527 if (lvb_in && ua->lksb.sb_lvbptr)
4528 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4530 ua->xid = ua_tmp->xid;
4531 ua->castparam = ua_tmp->castparam;
4532 ua->castaddr = ua_tmp->castaddr;
4533 ua->bastparam = ua_tmp->bastparam;
4534 ua->bastaddr = ua_tmp->bastaddr;
4535 ua->user_lksb = ua_tmp->user_lksb;
4536 ua->old_mode = lkb->lkb_grmode;
4538 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
4539 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
4543 error = convert_lock(ls, lkb, &args);
4545 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
4550 dlm_unlock_recovery(ls);
4555 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4556 uint32_t flags, uint32_t lkid, char *lvb_in)
4558 struct dlm_lkb *lkb;
4559 struct dlm_args args;
4560 struct dlm_user_args *ua;
4563 dlm_lock_recovery(ls);
4565 error = find_lkb(ls, lkid, &lkb);
4569 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4571 if (lvb_in && ua->lksb.sb_lvbptr)
4572 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4573 if (ua_tmp->castparam)
4574 ua->castparam = ua_tmp->castparam;
4575 ua->user_lksb = ua_tmp->user_lksb;
4577 error = set_unlock_args(flags, ua, &args);
4581 error = unlock_lock(ls, lkb, &args);
4583 if (error == -DLM_EUNLOCK)
4585 /* from validate_unlock_args() */
4586 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
4591 spin_lock(&ua->proc->locks_spin);
4592 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4593 if (!list_empty(&lkb->lkb_ownqueue))
4594 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
4595 spin_unlock(&ua->proc->locks_spin);
4599 dlm_unlock_recovery(ls);
4604 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4605 uint32_t flags, uint32_t lkid)
4607 struct dlm_lkb *lkb;
4608 struct dlm_args args;
4609 struct dlm_user_args *ua;
4612 dlm_lock_recovery(ls);
4614 error = find_lkb(ls, lkid, &lkb);
4618 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4619 if (ua_tmp->castparam)
4620 ua->castparam = ua_tmp->castparam;
4621 ua->user_lksb = ua_tmp->user_lksb;
4623 error = set_unlock_args(flags, ua, &args);
4627 error = cancel_lock(ls, lkb, &args);
4629 if (error == -DLM_ECANCEL)
4631 /* from validate_unlock_args() */
4632 if (error == -EBUSY)
4637 dlm_unlock_recovery(ls);
4642 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
4644 struct dlm_lkb *lkb;
4645 struct dlm_args args;
4646 struct dlm_user_args *ua;
4650 dlm_lock_recovery(ls);
4652 error = find_lkb(ls, lkid, &lkb);
4656 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4658 error = set_unlock_args(flags, ua, &args);
4662 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4664 r = lkb->lkb_resource;
4668 error = validate_unlock_args(lkb, &args);
4671 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
4673 error = _cancel_lock(r, lkb);
4678 if (error == -DLM_ECANCEL)
4680 /* from validate_unlock_args() */
4681 if (error == -EBUSY)
4686 dlm_unlock_recovery(ls);
4690 /* lkb's that are removed from the waiters list by revert are just left on the
4691 orphans list with the granted orphan locks, to be freed by purge */
4693 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4695 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4696 struct dlm_args args;
4700 mutex_lock(&ls->ls_orphans_mutex);
4701 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
4702 mutex_unlock(&ls->ls_orphans_mutex);
4704 set_unlock_args(0, ua, &args);
4706 error = cancel_lock(ls, lkb, &args);
4707 if (error == -DLM_ECANCEL)
4712 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4713 Regardless of what rsb queue the lock is on, it's removed and freed. */
4715 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4717 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4718 struct dlm_args args;
4721 set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
4723 error = unlock_lock(ls, lkb, &args);
4724 if (error == -DLM_EUNLOCK)
4729 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4730 (which does lock_rsb) due to deadlock with receiving a message that does
4731 lock_rsb followed by dlm_user_add_ast() */
4733 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
4734 struct dlm_user_proc *proc)
4736 struct dlm_lkb *lkb = NULL;
4738 mutex_lock(&ls->ls_clear_proc_locks);
4739 if (list_empty(&proc->locks))
4742 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
4743 list_del_init(&lkb->lkb_ownqueue);
4745 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4746 lkb->lkb_flags |= DLM_IFL_ORPHAN;
4748 lkb->lkb_flags |= DLM_IFL_DEAD;
4750 mutex_unlock(&ls->ls_clear_proc_locks);
4754 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4755 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4756 which we clear here. */
4758 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
4759 list, and no more device_writes should add lkb's to proc->locks list; so we
4760 shouldn't need to take asts_spin or locks_spin here. this assumes that
4761 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4764 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4766 struct dlm_lkb *lkb, *safe;
4768 dlm_lock_recovery(ls);
4771 lkb = del_proc_lock(ls, proc);
4775 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4776 orphan_proc_lock(ls, lkb);
4778 unlock_proc_lock(ls, lkb);
4780 /* this removes the reference for the proc->locks list
4781 added by dlm_user_request, it may result in the lkb
4787 mutex_lock(&ls->ls_clear_proc_locks);
4789 /* in-progress unlocks */
4790 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4791 list_del_init(&lkb->lkb_ownqueue);
4792 lkb->lkb_flags |= DLM_IFL_DEAD;
4796 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4797 lkb->lkb_ast_type = 0;
4798 list_del(&lkb->lkb_astqueue);
4802 mutex_unlock(&ls->ls_clear_proc_locks);
4803 dlm_unlock_recovery(ls);
4806 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4808 struct dlm_lkb *lkb, *safe;
4812 spin_lock(&proc->locks_spin);
4813 if (!list_empty(&proc->locks)) {
4814 lkb = list_entry(proc->locks.next, struct dlm_lkb,
4816 list_del_init(&lkb->lkb_ownqueue);
4818 spin_unlock(&proc->locks_spin);
4823 lkb->lkb_flags |= DLM_IFL_DEAD;
4824 unlock_proc_lock(ls, lkb);
4825 dlm_put_lkb(lkb); /* ref from proc->locks list */
4828 spin_lock(&proc->locks_spin);
4829 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4830 list_del_init(&lkb->lkb_ownqueue);
4831 lkb->lkb_flags |= DLM_IFL_DEAD;
4834 spin_unlock(&proc->locks_spin);
4836 spin_lock(&proc->asts_spin);
4837 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4838 list_del(&lkb->lkb_astqueue);
4841 spin_unlock(&proc->asts_spin);
4844 /* pid of 0 means purge all orphans */
4846 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
4848 struct dlm_lkb *lkb, *safe;
4850 mutex_lock(&ls->ls_orphans_mutex);
4851 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
4852 if (pid && lkb->lkb_ownpid != pid)
4854 unlock_proc_lock(ls, lkb);
4855 list_del_init(&lkb->lkb_ownqueue);
4858 mutex_unlock(&ls->ls_orphans_mutex);
4861 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
4863 struct dlm_message *ms;
4864 struct dlm_mhandle *mh;
4867 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
4868 DLM_MSG_PURGE, &ms, &mh);
4871 ms->m_nodeid = nodeid;
4874 return send_message(mh, ms);
4877 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
4878 int nodeid, int pid)
4882 if (nodeid != dlm_our_nodeid()) {
4883 error = send_purge(ls, nodeid, pid);
4885 dlm_lock_recovery(ls);
4886 if (pid == current->pid)
4887 purge_proc_locks(ls, proc);
4889 do_purge(ls, nodeid, pid);
4890 dlm_unlock_recovery(ls);