1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005-2007 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);
91 void dlm_timeout_warn(struct dlm_lkb *lkb);
94 * Lock compatibilty matrix - thanks Steve
95 * UN = Unlocked state. Not really a state, used as a flag
96 * PD = Padding. Used to make the matrix a nice power of two in size
97 * Other states are the same as the VMS DLM.
98 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
101 static const int __dlm_compat_matrix[8][8] = {
102 /* UN NL CR CW PR PW EX PD */
103 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
104 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
105 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
106 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
107 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
108 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
109 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
110 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
114 * This defines the direction of transfer of LVB data.
115 * Granted mode is the row; requested mode is the column.
116 * Usage: matrix[grmode+1][rqmode+1]
117 * 1 = LVB is returned to the caller
118 * 0 = LVB is written to the resource
119 * -1 = nothing happens to the LVB
122 const int dlm_lvb_operations[8][8] = {
123 /* UN NL CR CW PR PW EX PD*/
124 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
125 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
126 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
127 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
128 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
129 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
130 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
131 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
134 #define modes_compat(gr, rq) \
135 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
137 int dlm_modes_compat(int mode1, int mode2)
139 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
143 * Compatibility matrix for conversions with QUECVT set.
144 * Granted mode is the row; requested mode is the column.
145 * Usage: matrix[grmode+1][rqmode+1]
148 static const int __quecvt_compat_matrix[8][8] = {
149 /* UN NL CR CW PR PW EX PD */
150 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
151 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
152 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
153 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
154 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
155 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
156 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
157 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
160 void dlm_print_lkb(struct dlm_lkb *lkb)
162 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
163 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
164 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
165 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
166 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
169 void dlm_print_rsb(struct dlm_rsb *r)
171 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
172 r->res_nodeid, r->res_flags, r->res_first_lkid,
173 r->res_recover_locks_count, r->res_name);
176 void dlm_dump_rsb(struct dlm_rsb *r)
182 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
183 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
184 printk(KERN_ERR "rsb lookup list\n");
185 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
187 printk(KERN_ERR "rsb grant queue:\n");
188 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
190 printk(KERN_ERR "rsb convert queue:\n");
191 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
193 printk(KERN_ERR "rsb wait queue:\n");
194 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
198 /* Threads cannot use the lockspace while it's being recovered */
200 static inline void dlm_lock_recovery(struct dlm_ls *ls)
202 down_read(&ls->ls_in_recovery);
205 void dlm_unlock_recovery(struct dlm_ls *ls)
207 up_read(&ls->ls_in_recovery);
210 int dlm_lock_recovery_try(struct dlm_ls *ls)
212 return down_read_trylock(&ls->ls_in_recovery);
215 static inline int can_be_queued(struct dlm_lkb *lkb)
217 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
220 static inline int force_blocking_asts(struct dlm_lkb *lkb)
222 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
225 static inline int is_demoted(struct dlm_lkb *lkb)
227 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
230 static inline int is_altmode(struct dlm_lkb *lkb)
232 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
235 static inline int is_granted(struct dlm_lkb *lkb)
237 return (lkb->lkb_status == DLM_LKSTS_GRANTED);
240 static inline int is_remote(struct dlm_rsb *r)
242 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
243 return !!r->res_nodeid;
246 static inline int is_process_copy(struct dlm_lkb *lkb)
248 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
251 static inline int is_master_copy(struct dlm_lkb *lkb)
253 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
254 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
255 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
258 static inline int middle_conversion(struct dlm_lkb *lkb)
260 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
261 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
266 static inline int down_conversion(struct dlm_lkb *lkb)
268 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
271 static inline int is_overlap_unlock(struct dlm_lkb *lkb)
273 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
276 static inline int is_overlap_cancel(struct dlm_lkb *lkb)
278 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
281 static inline int is_overlap(struct dlm_lkb *lkb)
283 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
284 DLM_IFL_OVERLAP_CANCEL));
287 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
289 if (is_master_copy(lkb))
294 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
296 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
297 timeout caused the cancel then return -ETIMEDOUT */
298 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
299 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
303 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
304 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
308 lkb->lkb_lksb->sb_status = rv;
309 lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
311 dlm_add_ast(lkb, AST_COMP);
314 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
317 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
320 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
322 if (is_master_copy(lkb))
323 send_bast(r, lkb, rqmode);
325 lkb->lkb_bastmode = rqmode;
326 dlm_add_ast(lkb, AST_BAST);
331 * Basic operations on rsb's and lkb's
334 static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
338 r = allocate_rsb(ls, len);
344 memcpy(r->res_name, name, len);
345 mutex_init(&r->res_mutex);
347 INIT_LIST_HEAD(&r->res_lookup);
348 INIT_LIST_HEAD(&r->res_grantqueue);
349 INIT_LIST_HEAD(&r->res_convertqueue);
350 INIT_LIST_HEAD(&r->res_waitqueue);
351 INIT_LIST_HEAD(&r->res_root_list);
352 INIT_LIST_HEAD(&r->res_recover_list);
357 static int search_rsb_list(struct list_head *head, char *name, int len,
358 unsigned int flags, struct dlm_rsb **r_ret)
363 list_for_each_entry(r, head, res_hashchain) {
364 if (len == r->res_length && !memcmp(name, r->res_name, len))
370 if (r->res_nodeid && (flags & R_MASTER))
376 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
377 unsigned int flags, struct dlm_rsb **r_ret)
382 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
384 kref_get(&r->res_ref);
387 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
391 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
393 if (dlm_no_directory(ls))
396 if (r->res_nodeid == -1) {
397 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
398 r->res_first_lkid = 0;
399 } else if (r->res_nodeid > 0) {
400 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
401 r->res_first_lkid = 0;
403 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
404 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
411 static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
412 unsigned int flags, struct dlm_rsb **r_ret)
415 write_lock(&ls->ls_rsbtbl[b].lock);
416 error = _search_rsb(ls, name, len, b, flags, r_ret);
417 write_unlock(&ls->ls_rsbtbl[b].lock);
422 * Find rsb in rsbtbl and potentially create/add one
424 * Delaying the release of rsb's has a similar benefit to applications keeping
425 * NL locks on an rsb, but without the guarantee that the cached master value
426 * will still be valid when the rsb is reused. Apps aren't always smart enough
427 * to keep NL locks on an rsb that they may lock again shortly; this can lead
428 * to excessive master lookups and removals if we don't delay the release.
430 * Searching for an rsb means looking through both the normal list and toss
431 * list. When found on the toss list the rsb is moved to the normal list with
432 * ref count of 1; when found on normal list the ref count is incremented.
435 static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
436 unsigned int flags, struct dlm_rsb **r_ret)
438 struct dlm_rsb *r, *tmp;
439 uint32_t hash, bucket;
442 if (dlm_no_directory(ls))
445 hash = jhash(name, namelen, 0);
446 bucket = hash & (ls->ls_rsbtbl_size - 1);
448 error = search_rsb(ls, name, namelen, bucket, flags, &r);
452 if (error == -EBADR && !(flags & R_CREATE))
455 /* the rsb was found but wasn't a master copy */
456 if (error == -ENOTBLK)
460 r = create_rsb(ls, name, namelen);
465 r->res_bucket = bucket;
467 kref_init(&r->res_ref);
469 /* With no directory, the master can be set immediately */
470 if (dlm_no_directory(ls)) {
471 int nodeid = dlm_dir_nodeid(r);
472 if (nodeid == dlm_our_nodeid())
474 r->res_nodeid = nodeid;
477 write_lock(&ls->ls_rsbtbl[bucket].lock);
478 error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
480 write_unlock(&ls->ls_rsbtbl[bucket].lock);
485 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
486 write_unlock(&ls->ls_rsbtbl[bucket].lock);
493 int dlm_find_rsb(struct dlm_ls *ls, char *name, int namelen,
494 unsigned int flags, struct dlm_rsb **r_ret)
496 return find_rsb(ls, name, namelen, flags, r_ret);
499 /* This is only called to add a reference when the code already holds
500 a valid reference to the rsb, so there's no need for locking. */
502 static inline void hold_rsb(struct dlm_rsb *r)
504 kref_get(&r->res_ref);
507 void dlm_hold_rsb(struct dlm_rsb *r)
512 static void toss_rsb(struct kref *kref)
514 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
515 struct dlm_ls *ls = r->res_ls;
517 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
518 kref_init(&r->res_ref);
519 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
520 r->res_toss_time = jiffies;
522 free_lvb(r->res_lvbptr);
523 r->res_lvbptr = NULL;
527 /* When all references to the rsb are gone it's transfered to
528 the tossed list for later disposal. */
530 static void put_rsb(struct dlm_rsb *r)
532 struct dlm_ls *ls = r->res_ls;
533 uint32_t bucket = r->res_bucket;
535 write_lock(&ls->ls_rsbtbl[bucket].lock);
536 kref_put(&r->res_ref, toss_rsb);
537 write_unlock(&ls->ls_rsbtbl[bucket].lock);
540 void dlm_put_rsb(struct dlm_rsb *r)
545 /* See comment for unhold_lkb */
547 static void unhold_rsb(struct dlm_rsb *r)
550 rv = kref_put(&r->res_ref, toss_rsb);
551 DLM_ASSERT(!rv, dlm_dump_rsb(r););
554 static void kill_rsb(struct kref *kref)
556 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
558 /* All work is done after the return from kref_put() so we
559 can release the write_lock before the remove and free. */
561 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
562 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
563 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
564 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
565 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
566 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
569 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
570 The rsb must exist as long as any lkb's for it do. */
572 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
575 lkb->lkb_resource = r;
578 static void detach_lkb(struct dlm_lkb *lkb)
580 if (lkb->lkb_resource) {
581 put_rsb(lkb->lkb_resource);
582 lkb->lkb_resource = NULL;
586 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
588 struct dlm_lkb *lkb, *tmp;
592 lkb = allocate_lkb(ls);
596 lkb->lkb_nodeid = -1;
597 lkb->lkb_grmode = DLM_LOCK_IV;
598 kref_init(&lkb->lkb_ref);
599 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
600 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
601 INIT_LIST_HEAD(&lkb->lkb_time_list);
603 get_random_bytes(&bucket, sizeof(bucket));
604 bucket &= (ls->ls_lkbtbl_size - 1);
606 write_lock(&ls->ls_lkbtbl[bucket].lock);
608 /* counter can roll over so we must verify lkid is not in use */
611 lkid = (bucket << 16) | ls->ls_lkbtbl[bucket].counter++;
613 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
615 if (tmp->lkb_id != lkid)
623 list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
624 write_unlock(&ls->ls_lkbtbl[bucket].lock);
630 static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
633 uint16_t bucket = (lkid >> 16);
635 list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
636 if (lkb->lkb_id == lkid)
642 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
645 uint16_t bucket = (lkid >> 16);
647 if (bucket >= ls->ls_lkbtbl_size)
650 read_lock(&ls->ls_lkbtbl[bucket].lock);
651 lkb = __find_lkb(ls, lkid);
653 kref_get(&lkb->lkb_ref);
654 read_unlock(&ls->ls_lkbtbl[bucket].lock);
657 return lkb ? 0 : -ENOENT;
660 static void kill_lkb(struct kref *kref)
662 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
664 /* All work is done after the return from kref_put() so we
665 can release the write_lock before the detach_lkb */
667 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
670 /* __put_lkb() is used when an lkb may not have an rsb attached to
671 it so we need to provide the lockspace explicitly */
673 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
675 uint16_t bucket = (lkb->lkb_id >> 16);
677 write_lock(&ls->ls_lkbtbl[bucket].lock);
678 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
679 list_del(&lkb->lkb_idtbl_list);
680 write_unlock(&ls->ls_lkbtbl[bucket].lock);
684 /* for local/process lkbs, lvbptr points to caller's lksb */
685 if (lkb->lkb_lvbptr && is_master_copy(lkb))
686 free_lvb(lkb->lkb_lvbptr);
690 write_unlock(&ls->ls_lkbtbl[bucket].lock);
695 int dlm_put_lkb(struct dlm_lkb *lkb)
699 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
700 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
702 ls = lkb->lkb_resource->res_ls;
703 return __put_lkb(ls, lkb);
706 /* This is only called to add a reference when the code already holds
707 a valid reference to the lkb, so there's no need for locking. */
709 static inline void hold_lkb(struct dlm_lkb *lkb)
711 kref_get(&lkb->lkb_ref);
714 /* This is called when we need to remove a reference and are certain
715 it's not the last ref. e.g. del_lkb is always called between a
716 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
717 put_lkb would work fine, but would involve unnecessary locking */
719 static inline void unhold_lkb(struct dlm_lkb *lkb)
722 rv = kref_put(&lkb->lkb_ref, kill_lkb);
723 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
726 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
729 struct dlm_lkb *lkb = NULL;
731 list_for_each_entry(lkb, head, lkb_statequeue)
732 if (lkb->lkb_rqmode < mode)
736 list_add_tail(new, head);
738 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
741 /* add/remove lkb to rsb's grant/convert/wait queue */
743 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
745 kref_get(&lkb->lkb_ref);
747 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
749 lkb->lkb_status = status;
752 case DLM_LKSTS_WAITING:
753 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
754 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
756 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
758 case DLM_LKSTS_GRANTED:
759 /* convention says granted locks kept in order of grmode */
760 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
763 case DLM_LKSTS_CONVERT:
764 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
765 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
767 list_add_tail(&lkb->lkb_statequeue,
768 &r->res_convertqueue);
771 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
775 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
778 list_del(&lkb->lkb_statequeue);
782 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
786 add_lkb(r, lkb, sts);
790 static int msg_reply_type(int mstype)
793 case DLM_MSG_REQUEST:
794 return DLM_MSG_REQUEST_REPLY;
795 case DLM_MSG_CONVERT:
796 return DLM_MSG_CONVERT_REPLY;
798 return DLM_MSG_UNLOCK_REPLY;
800 return DLM_MSG_CANCEL_REPLY;
802 return DLM_MSG_LOOKUP_REPLY;
807 /* add/remove lkb from global waiters list of lkb's waiting for
808 a reply from a remote node */
810 static int add_to_waiters(struct dlm_lkb *lkb, int mstype)
812 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
815 mutex_lock(&ls->ls_waiters_mutex);
817 if (is_overlap_unlock(lkb) ||
818 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
823 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
826 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
829 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
835 lkb->lkb_wait_count++;
838 log_debug(ls, "add overlap %x cur %d new %d count %d flags %x",
839 lkb->lkb_id, lkb->lkb_wait_type, mstype,
840 lkb->lkb_wait_count, lkb->lkb_flags);
844 DLM_ASSERT(!lkb->lkb_wait_count,
846 printk("wait_count %d\n", lkb->lkb_wait_count););
848 lkb->lkb_wait_count++;
849 lkb->lkb_wait_type = mstype;
851 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
854 log_error(ls, "add_to_waiters %x error %d flags %x %d %d %s",
855 lkb->lkb_id, error, lkb->lkb_flags, mstype,
856 lkb->lkb_wait_type, lkb->lkb_resource->res_name);
857 mutex_unlock(&ls->ls_waiters_mutex);
861 /* We clear the RESEND flag because we might be taking an lkb off the waiters
862 list as part of process_requestqueue (e.g. a lookup that has an optimized
863 request reply on the requestqueue) between dlm_recover_waiters_pre() which
864 set RESEND and dlm_recover_waiters_post() */
866 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype)
868 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
869 int overlap_done = 0;
871 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
872 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
877 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
878 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
883 /* N.B. type of reply may not always correspond to type of original
884 msg due to lookup->request optimization, verify others? */
886 if (lkb->lkb_wait_type) {
887 lkb->lkb_wait_type = 0;
891 log_error(ls, "remove_from_waiters lkid %x flags %x types %d %d",
892 lkb->lkb_id, lkb->lkb_flags, mstype, lkb->lkb_wait_type);
896 /* the force-unlock/cancel has completed and we haven't recvd a reply
897 to the op that was in progress prior to the unlock/cancel; we
898 give up on any reply to the earlier op. FIXME: not sure when/how
901 if (overlap_done && lkb->lkb_wait_type) {
902 log_error(ls, "remove_from_waiters %x reply %d give up on %d",
903 lkb->lkb_id, mstype, lkb->lkb_wait_type);
904 lkb->lkb_wait_count--;
905 lkb->lkb_wait_type = 0;
908 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
910 lkb->lkb_flags &= ~DLM_IFL_RESEND;
911 lkb->lkb_wait_count--;
912 if (!lkb->lkb_wait_count)
913 list_del_init(&lkb->lkb_wait_reply);
918 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
920 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
923 mutex_lock(&ls->ls_waiters_mutex);
924 error = _remove_from_waiters(lkb, mstype);
925 mutex_unlock(&ls->ls_waiters_mutex);
929 /* Handles situations where we might be processing a "fake" or "stub" reply in
930 which we can't try to take waiters_mutex again. */
932 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
934 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
937 if (ms != &ls->ls_stub_ms)
938 mutex_lock(&ls->ls_waiters_mutex);
939 error = _remove_from_waiters(lkb, ms->m_type);
940 if (ms != &ls->ls_stub_ms)
941 mutex_unlock(&ls->ls_waiters_mutex);
945 static void dir_remove(struct dlm_rsb *r)
949 if (dlm_no_directory(r->res_ls))
952 to_nodeid = dlm_dir_nodeid(r);
953 if (to_nodeid != dlm_our_nodeid())
956 dlm_dir_remove_entry(r->res_ls, to_nodeid,
957 r->res_name, r->res_length);
960 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
961 found since they are in order of newest to oldest? */
963 static int shrink_bucket(struct dlm_ls *ls, int b)
966 int count = 0, found;
970 write_lock(&ls->ls_rsbtbl[b].lock);
971 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
973 if (!time_after_eq(jiffies, r->res_toss_time +
974 dlm_config.ci_toss_secs * HZ))
981 write_unlock(&ls->ls_rsbtbl[b].lock);
985 if (kref_put(&r->res_ref, kill_rsb)) {
986 list_del(&r->res_hashchain);
987 write_unlock(&ls->ls_rsbtbl[b].lock);
994 write_unlock(&ls->ls_rsbtbl[b].lock);
995 log_error(ls, "tossed rsb in use %s", r->res_name);
1002 void dlm_scan_rsbs(struct dlm_ls *ls)
1006 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1007 shrink_bucket(ls, i);
1008 if (dlm_locking_stopped(ls))
1014 static void add_timeout(struct dlm_lkb *lkb)
1016 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1018 if (is_master_copy(lkb)) {
1019 lkb->lkb_timestamp = jiffies;
1023 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1024 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1025 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1028 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1033 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1034 mutex_lock(&ls->ls_timeout_mutex);
1036 lkb->lkb_timestamp = jiffies;
1037 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1038 mutex_unlock(&ls->ls_timeout_mutex);
1041 static void del_timeout(struct dlm_lkb *lkb)
1043 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1045 mutex_lock(&ls->ls_timeout_mutex);
1046 if (!list_empty(&lkb->lkb_time_list)) {
1047 list_del_init(&lkb->lkb_time_list);
1050 mutex_unlock(&ls->ls_timeout_mutex);
1053 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1054 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1055 and then lock rsb because of lock ordering in add_timeout. We may need
1056 to specify some special timeout-related bits in the lkb that are just to
1057 be accessed under the timeout_mutex. */
1059 void dlm_scan_timeout(struct dlm_ls *ls)
1062 struct dlm_lkb *lkb;
1063 int do_cancel, do_warn;
1066 if (dlm_locking_stopped(ls))
1071 mutex_lock(&ls->ls_timeout_mutex);
1072 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) {
1074 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) &&
1075 time_after_eq(jiffies, lkb->lkb_timestamp +
1076 lkb->lkb_timeout_cs * HZ/100))
1079 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1080 time_after_eq(jiffies, lkb->lkb_timestamp +
1081 dlm_config.ci_timewarn_cs * HZ/100))
1084 if (!do_cancel && !do_warn)
1089 mutex_unlock(&ls->ls_timeout_mutex);
1091 if (!do_cancel && !do_warn)
1094 r = lkb->lkb_resource;
1099 /* clear flag so we only warn once */
1100 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1101 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1103 dlm_timeout_warn(lkb);
1107 log_debug(ls, "timeout cancel %x node %d %s",
1108 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1109 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1110 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1112 _cancel_lock(r, lkb);
1121 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1122 dlm_recoverd before checking/setting ls_recover_begin. */
1124 void dlm_adjust_timeouts(struct dlm_ls *ls)
1126 struct dlm_lkb *lkb;
1127 long adj = jiffies - ls->ls_recover_begin;
1129 ls->ls_recover_begin = 0;
1130 mutex_lock(&ls->ls_timeout_mutex);
1131 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1132 lkb->lkb_timestamp += adj;
1133 mutex_unlock(&ls->ls_timeout_mutex);
1136 /* lkb is master or local copy */
1138 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1140 int b, len = r->res_ls->ls_lvblen;
1142 /* b=1 lvb returned to caller
1143 b=0 lvb written to rsb or invalidated
1146 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1149 if (!lkb->lkb_lvbptr)
1152 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1158 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1159 lkb->lkb_lvbseq = r->res_lvbseq;
1161 } else if (b == 0) {
1162 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1163 rsb_set_flag(r, RSB_VALNOTVALID);
1167 if (!lkb->lkb_lvbptr)
1170 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1174 r->res_lvbptr = allocate_lvb(r->res_ls);
1179 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1181 lkb->lkb_lvbseq = r->res_lvbseq;
1182 rsb_clear_flag(r, RSB_VALNOTVALID);
1185 if (rsb_flag(r, RSB_VALNOTVALID))
1186 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1189 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1191 if (lkb->lkb_grmode < DLM_LOCK_PW)
1194 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1195 rsb_set_flag(r, RSB_VALNOTVALID);
1199 if (!lkb->lkb_lvbptr)
1202 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1206 r->res_lvbptr = allocate_lvb(r->res_ls);
1211 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1213 rsb_clear_flag(r, RSB_VALNOTVALID);
1216 /* lkb is process copy (pc) */
1218 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1219 struct dlm_message *ms)
1223 if (!lkb->lkb_lvbptr)
1226 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1229 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1231 int len = receive_extralen(ms);
1232 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1233 lkb->lkb_lvbseq = ms->m_lvbseq;
1237 /* Manipulate lkb's on rsb's convert/granted/waiting queues
1238 remove_lock -- used for unlock, removes lkb from granted
1239 revert_lock -- used for cancel, moves lkb from convert to granted
1240 grant_lock -- used for request and convert, adds lkb to granted or
1241 moves lkb from convert or waiting to granted
1243 Each of these is used for master or local copy lkb's. There is
1244 also a _pc() variation used to make the corresponding change on
1245 a process copy (pc) lkb. */
1247 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1250 lkb->lkb_grmode = DLM_LOCK_IV;
1251 /* this unhold undoes the original ref from create_lkb()
1252 so this leads to the lkb being freed */
1256 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1258 set_lvb_unlock(r, lkb);
1259 _remove_lock(r, lkb);
1262 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1264 _remove_lock(r, lkb);
1267 /* returns: 0 did nothing
1268 1 moved lock to granted
1271 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1275 lkb->lkb_rqmode = DLM_LOCK_IV;
1277 switch (lkb->lkb_status) {
1278 case DLM_LKSTS_GRANTED:
1280 case DLM_LKSTS_CONVERT:
1281 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1284 case DLM_LKSTS_WAITING:
1286 lkb->lkb_grmode = DLM_LOCK_IV;
1287 /* this unhold undoes the original ref from create_lkb()
1288 so this leads to the lkb being freed */
1293 log_print("invalid status for revert %d", lkb->lkb_status);
1298 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1300 return revert_lock(r, lkb);
1303 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1305 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1306 lkb->lkb_grmode = lkb->lkb_rqmode;
1307 if (lkb->lkb_status)
1308 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1310 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1313 lkb->lkb_rqmode = DLM_LOCK_IV;
1316 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1318 set_lvb_lock(r, lkb);
1319 _grant_lock(r, lkb);
1320 lkb->lkb_highbast = 0;
1323 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1324 struct dlm_message *ms)
1326 set_lvb_lock_pc(r, lkb, ms);
1327 _grant_lock(r, lkb);
1330 /* called by grant_pending_locks() which means an async grant message must
1331 be sent to the requesting node in addition to granting the lock if the
1332 lkb belongs to a remote node. */
1334 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1337 if (is_master_copy(lkb))
1340 queue_cast(r, lkb, 0);
1343 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1344 change the granted/requested modes. We're munging things accordingly in
1346 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1348 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1349 compatible with other granted locks */
1351 static void munge_demoted(struct dlm_lkb *lkb, struct dlm_message *ms)
1353 if (ms->m_type != DLM_MSG_CONVERT_REPLY) {
1354 log_print("munge_demoted %x invalid reply type %d",
1355 lkb->lkb_id, ms->m_type);
1359 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
1360 log_print("munge_demoted %x invalid modes gr %d rq %d",
1361 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
1365 lkb->lkb_grmode = DLM_LOCK_NL;
1368 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
1370 if (ms->m_type != DLM_MSG_REQUEST_REPLY &&
1371 ms->m_type != DLM_MSG_GRANT) {
1372 log_print("munge_altmode %x invalid reply type %d",
1373 lkb->lkb_id, ms->m_type);
1377 if (lkb->lkb_exflags & DLM_LKF_ALTPR)
1378 lkb->lkb_rqmode = DLM_LOCK_PR;
1379 else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
1380 lkb->lkb_rqmode = DLM_LOCK_CW;
1382 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
1387 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1389 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1391 if (lkb->lkb_id == first->lkb_id)
1397 /* Check if the given lkb conflicts with another lkb on the queue. */
1399 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1401 struct dlm_lkb *this;
1403 list_for_each_entry(this, head, lkb_statequeue) {
1406 if (!modes_compat(this, lkb))
1413 * "A conversion deadlock arises with a pair of lock requests in the converting
1414 * queue for one resource. The granted mode of each lock blocks the requested
1415 * mode of the other lock."
1417 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1418 * convert queue from being granted, then deadlk/demote lkb.
1421 * Granted Queue: empty
1422 * Convert Queue: NL->EX (first lock)
1423 * PR->EX (second lock)
1425 * The first lock can't be granted because of the granted mode of the second
1426 * lock and the second lock can't be granted because it's not first in the
1427 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1428 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1429 * flag set and return DEMOTED in the lksb flags.
1431 * Originally, this function detected conv-deadlk in a more limited scope:
1432 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1433 * - if lkb1 was the first entry in the queue (not just earlier), and was
1434 * blocked by the granted mode of lkb2, and there was nothing on the
1435 * granted queue preventing lkb1 from being granted immediately, i.e.
1436 * lkb2 was the only thing preventing lkb1 from being granted.
1438 * That second condition meant we'd only say there was conv-deadlk if
1439 * resolving it (by demotion) would lead to the first lock on the convert
1440 * queue being granted right away. It allowed conversion deadlocks to exist
1441 * between locks on the convert queue while they couldn't be granted anyway.
1443 * Now, we detect and take action on conversion deadlocks immediately when
1444 * they're created, even if they may not be immediately consequential. If
1445 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1446 * mode that would prevent lkb1's conversion from being granted, we do a
1447 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1448 * I think this means that the lkb_is_ahead condition below should always
1449 * be zero, i.e. there will never be conv-deadlk between two locks that are
1450 * both already on the convert queue.
1453 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
1455 struct dlm_lkb *lkb1;
1456 int lkb_is_ahead = 0;
1458 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
1464 if (!lkb_is_ahead) {
1465 if (!modes_compat(lkb2, lkb1))
1468 if (!modes_compat(lkb2, lkb1) &&
1469 !modes_compat(lkb1, lkb2))
1477 * Return 1 if the lock can be granted, 0 otherwise.
1478 * Also detect and resolve conversion deadlocks.
1480 * lkb is the lock to be granted
1482 * now is 1 if the function is being called in the context of the
1483 * immediate request, it is 0 if called later, after the lock has been
1486 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1489 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1491 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1494 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1495 * a new request for a NL mode lock being blocked.
1497 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1498 * request, then it would be granted. In essence, the use of this flag
1499 * tells the Lock Manager to expedite theis request by not considering
1500 * what may be in the CONVERTING or WAITING queues... As of this
1501 * writing, the EXPEDITE flag can be used only with new requests for NL
1502 * mode locks. This flag is not valid for conversion requests.
1504 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1505 * conversion or used with a non-NL requested mode. We also know an
1506 * EXPEDITE request is always granted immediately, so now must always
1507 * be 1. The full condition to grant an expedite request: (now &&
1508 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1509 * therefore be shortened to just checking the flag.
1512 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1516 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1517 * added to the remaining conditions.
1520 if (queue_conflict(&r->res_grantqueue, lkb))
1524 * 6-3: By default, a conversion request is immediately granted if the
1525 * requested mode is compatible with the modes of all other granted
1529 if (queue_conflict(&r->res_convertqueue, lkb))
1533 * 6-5: But the default algorithm for deciding whether to grant or
1534 * queue conversion requests does not by itself guarantee that such
1535 * requests are serviced on a "first come first serve" basis. This, in
1536 * turn, can lead to a phenomenon known as "indefinate postponement".
1538 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1539 * the system service employed to request a lock conversion. This flag
1540 * forces certain conversion requests to be queued, even if they are
1541 * compatible with the granted modes of other locks on the same
1542 * resource. Thus, the use of this flag results in conversion requests
1543 * being ordered on a "first come first servce" basis.
1545 * DCT: This condition is all about new conversions being able to occur
1546 * "in place" while the lock remains on the granted queue (assuming
1547 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1548 * doesn't _have_ to go onto the convert queue where it's processed in
1549 * order. The "now" variable is necessary to distinguish converts
1550 * being received and processed for the first time now, because once a
1551 * convert is moved to the conversion queue the condition below applies
1552 * requiring fifo granting.
1555 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1559 * The NOORDER flag is set to avoid the standard vms rules on grant
1563 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1567 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1568 * granted until all other conversion requests ahead of it are granted
1572 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1576 * 6-4: By default, a new request is immediately granted only if all
1577 * three of the following conditions are satisfied when the request is
1579 * - The queue of ungranted conversion requests for the resource is
1581 * - The queue of ungranted new requests for the resource is empty.
1582 * - The mode of the new request is compatible with the most
1583 * restrictive mode of all granted locks on the resource.
1586 if (now && !conv && list_empty(&r->res_convertqueue) &&
1587 list_empty(&r->res_waitqueue))
1591 * 6-4: Once a lock request is in the queue of ungranted new requests,
1592 * it cannot be granted until the queue of ungranted conversion
1593 * requests is empty, all ungranted new requests ahead of it are
1594 * granted and/or canceled, and it is compatible with the granted mode
1595 * of the most restrictive lock granted on the resource.
1598 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1599 first_in_list(lkb, &r->res_waitqueue))
1605 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
1609 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1610 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
1615 rv = _can_be_granted(r, lkb, now);
1620 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1621 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1622 * cancels one of the locks.
1625 if (is_convert && can_be_queued(lkb) &&
1626 conversion_deadlock_detect(r, lkb)) {
1627 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
1628 lkb->lkb_grmode = DLM_LOCK_NL;
1629 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1630 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1634 log_print("can_be_granted deadlock %x now %d",
1643 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1644 * to grant a request in a mode other than the normal rqmode. It's a
1645 * simple way to provide a big optimization to applications that can
1649 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
1651 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
1655 lkb->lkb_rqmode = alt;
1656 rv = _can_be_granted(r, lkb, now);
1658 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1660 lkb->lkb_rqmode = rqmode;
1666 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1667 for locks pending on the convert list. Once verified (watch for these
1668 log_prints), we should be able to just call _can_be_granted() and not
1669 bother with the demote/deadlk cases here (and there's no easy way to deal
1670 with a deadlk here, we'd have to generate something like grant_lock with
1671 the deadlk error.) */
1673 /* returns the highest requested mode of all blocked conversions */
1675 static int grant_pending_convert(struct dlm_rsb *r, int high)
1677 struct dlm_lkb *lkb, *s;
1678 int hi, demoted, quit, grant_restart, demote_restart;
1687 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1688 demoted = is_demoted(lkb);
1691 if (can_be_granted(r, lkb, 0, &deadlk)) {
1692 grant_lock_pending(r, lkb);
1697 if (!demoted && is_demoted(lkb)) {
1698 log_print("WARN: pending demoted %x node %d %s",
1699 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1705 log_print("WARN: pending deadlock %x node %d %s",
1706 lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1711 hi = max_t(int, lkb->lkb_rqmode, hi);
1716 if (demote_restart && !quit) {
1721 return max_t(int, high, hi);
1724 static int grant_pending_wait(struct dlm_rsb *r, int high)
1726 struct dlm_lkb *lkb, *s;
1728 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1729 if (can_be_granted(r, lkb, 0, NULL))
1730 grant_lock_pending(r, lkb);
1732 high = max_t(int, lkb->lkb_rqmode, high);
1738 static void grant_pending_locks(struct dlm_rsb *r)
1740 struct dlm_lkb *lkb, *s;
1741 int high = DLM_LOCK_IV;
1743 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
1745 high = grant_pending_convert(r, high);
1746 high = grant_pending_wait(r, high);
1748 if (high == DLM_LOCK_IV)
1752 * If there are locks left on the wait/convert queue then send blocking
1753 * ASTs to granted locks based on the largest requested mode (high)
1754 * found above. FIXME: highbast < high comparison not valid for PR/CW.
1757 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1758 if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
1759 !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
1760 queue_bast(r, lkb, high);
1761 lkb->lkb_highbast = high;
1766 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1767 struct dlm_lkb *lkb)
1771 list_for_each_entry(gr, head, lkb_statequeue) {
1772 if (gr->lkb_bastaddr &&
1773 gr->lkb_highbast < lkb->lkb_rqmode &&
1774 !modes_compat(gr, lkb)) {
1775 queue_bast(r, gr, lkb->lkb_rqmode);
1776 gr->lkb_highbast = lkb->lkb_rqmode;
1781 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1783 send_bast_queue(r, &r->res_grantqueue, lkb);
1786 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1788 send_bast_queue(r, &r->res_grantqueue, lkb);
1789 send_bast_queue(r, &r->res_convertqueue, lkb);
1792 /* set_master(r, lkb) -- set the master nodeid of a resource
1794 The purpose of this function is to set the nodeid field in the given
1795 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1796 known, it can just be copied to the lkb and the function will return
1797 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1798 before it can be copied to the lkb.
1800 When the rsb nodeid is being looked up remotely, the initial lkb
1801 causing the lookup is kept on the ls_waiters list waiting for the
1802 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1803 on the rsb's res_lookup list until the master is verified.
1806 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1807 1: the rsb master is not available and the lkb has been placed on
1811 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1813 struct dlm_ls *ls = r->res_ls;
1814 int error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
1816 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1817 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1818 r->res_first_lkid = lkb->lkb_id;
1819 lkb->lkb_nodeid = r->res_nodeid;
1823 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1824 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1828 if (r->res_nodeid == 0) {
1829 lkb->lkb_nodeid = 0;
1833 if (r->res_nodeid > 0) {
1834 lkb->lkb_nodeid = r->res_nodeid;
1838 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
1840 dir_nodeid = dlm_dir_nodeid(r);
1842 if (dir_nodeid != our_nodeid) {
1843 r->res_first_lkid = lkb->lkb_id;
1844 send_lookup(r, lkb);
1849 /* It's possible for dlm_scand to remove an old rsb for
1850 this same resource from the toss list, us to create
1851 a new one, look up the master locally, and find it
1852 already exists just before dlm_scand does the
1853 dir_remove() on the previous rsb. */
1855 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1856 r->res_length, &ret_nodeid);
1859 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1863 if (ret_nodeid == our_nodeid) {
1864 r->res_first_lkid = 0;
1866 lkb->lkb_nodeid = 0;
1868 r->res_first_lkid = lkb->lkb_id;
1869 r->res_nodeid = ret_nodeid;
1870 lkb->lkb_nodeid = ret_nodeid;
1875 static void process_lookup_list(struct dlm_rsb *r)
1877 struct dlm_lkb *lkb, *safe;
1879 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
1880 list_del_init(&lkb->lkb_rsb_lookup);
1881 _request_lock(r, lkb);
1886 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1888 static void confirm_master(struct dlm_rsb *r, int error)
1890 struct dlm_lkb *lkb;
1892 if (!r->res_first_lkid)
1898 r->res_first_lkid = 0;
1899 process_lookup_list(r);
1903 /* the remote master didn't queue our NOQUEUE request;
1904 make a waiting lkb the first_lkid */
1906 r->res_first_lkid = 0;
1908 if (!list_empty(&r->res_lookup)) {
1909 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1911 list_del_init(&lkb->lkb_rsb_lookup);
1912 r->res_first_lkid = lkb->lkb_id;
1913 _request_lock(r, lkb);
1919 log_error(r->res_ls, "confirm_master unknown error %d", error);
1923 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
1924 int namelen, unsigned long timeout_cs, void *ast,
1925 void *astarg, void *bast, struct dlm_args *args)
1929 /* check for invalid arg usage */
1931 if (mode < 0 || mode > DLM_LOCK_EX)
1934 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1937 if (flags & DLM_LKF_CANCEL)
1940 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1943 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1946 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1949 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1952 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
1955 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
1958 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
1964 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
1967 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
1970 /* these args will be copied to the lkb in validate_lock_args,
1971 it cannot be done now because when converting locks, fields in
1972 an active lkb cannot be modified before locking the rsb */
1974 args->flags = flags;
1975 args->astaddr = ast;
1976 args->astparam = (long) astarg;
1977 args->bastaddr = bast;
1978 args->timeout = timeout_cs;
1986 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
1988 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
1989 DLM_LKF_FORCEUNLOCK))
1992 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
1995 args->flags = flags;
1996 args->astparam = (long) astarg;
2000 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2001 struct dlm_args *args)
2005 if (args->flags & DLM_LKF_CONVERT) {
2006 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2009 if (args->flags & DLM_LKF_QUECVT &&
2010 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2014 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2017 if (lkb->lkb_wait_type)
2020 if (is_overlap(lkb))
2024 lkb->lkb_exflags = args->flags;
2025 lkb->lkb_sbflags = 0;
2026 lkb->lkb_astaddr = args->astaddr;
2027 lkb->lkb_astparam = args->astparam;
2028 lkb->lkb_bastaddr = args->bastaddr;
2029 lkb->lkb_rqmode = args->mode;
2030 lkb->lkb_lksb = args->lksb;
2031 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2032 lkb->lkb_ownpid = (int) current->pid;
2033 lkb->lkb_timeout_cs = args->timeout;
2039 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2042 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2043 because there may be a lookup in progress and it's valid to do
2044 cancel/unlockf on it */
2046 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2048 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2051 if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2052 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2057 /* an lkb may still exist even though the lock is EOL'ed due to a
2058 cancel, unlock or failed noqueue request; an app can't use these
2059 locks; return same error as if the lkid had not been found at all */
2061 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2062 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2067 /* an lkb may be waiting for an rsb lookup to complete where the
2068 lookup was initiated by another lock */
2070 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2071 if (!list_empty(&lkb->lkb_rsb_lookup)) {
2072 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2073 list_del_init(&lkb->lkb_rsb_lookup);
2074 queue_cast(lkb->lkb_resource, lkb,
2075 args->flags & DLM_LKF_CANCEL ?
2076 -DLM_ECANCEL : -DLM_EUNLOCK);
2077 unhold_lkb(lkb); /* undoes create_lkb() */
2083 /* cancel not allowed with another cancel/unlock in progress */
2085 if (args->flags & DLM_LKF_CANCEL) {
2086 if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2089 if (is_overlap(lkb))
2092 /* don't let scand try to do a cancel */
2095 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2096 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2101 switch (lkb->lkb_wait_type) {
2102 case DLM_MSG_LOOKUP:
2103 case DLM_MSG_REQUEST:
2104 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2107 case DLM_MSG_UNLOCK:
2108 case DLM_MSG_CANCEL:
2111 /* add_to_waiters() will set OVERLAP_CANCEL */
2115 /* do we need to allow a force-unlock if there's a normal unlock
2116 already in progress? in what conditions could the normal unlock
2117 fail such that we'd want to send a force-unlock to be sure? */
2119 if (args->flags & DLM_LKF_FORCEUNLOCK) {
2120 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2123 if (is_overlap_unlock(lkb))
2126 /* don't let scand try to do a cancel */
2129 if (lkb->lkb_flags & DLM_IFL_RESEND) {
2130 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2135 switch (lkb->lkb_wait_type) {
2136 case DLM_MSG_LOOKUP:
2137 case DLM_MSG_REQUEST:
2138 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2141 case DLM_MSG_UNLOCK:
2144 /* add_to_waiters() will set OVERLAP_UNLOCK */
2148 /* normal unlock not allowed if there's any op in progress */
2150 if (lkb->lkb_wait_type || lkb->lkb_wait_count)
2154 /* an overlapping op shouldn't blow away exflags from other op */
2155 lkb->lkb_exflags |= args->flags;
2156 lkb->lkb_sbflags = 0;
2157 lkb->lkb_astparam = args->astparam;
2161 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv,
2162 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
2163 args->flags, lkb->lkb_wait_type,
2164 lkb->lkb_resource->res_name);
2169 * Four stage 4 varieties:
2170 * do_request(), do_convert(), do_unlock(), do_cancel()
2171 * These are called on the master node for the given lock and
2172 * from the central locking logic.
2175 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2179 if (can_be_granted(r, lkb, 1, NULL)) {
2181 queue_cast(r, lkb, 0);
2185 if (can_be_queued(lkb)) {
2186 error = -EINPROGRESS;
2187 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2188 send_blocking_asts(r, lkb);
2194 if (force_blocking_asts(lkb))
2195 send_blocking_asts_all(r, lkb);
2196 queue_cast(r, lkb, -EAGAIN);
2202 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2207 /* changing an existing lock may allow others to be granted */
2209 if (can_be_granted(r, lkb, 1, &deadlk)) {
2211 queue_cast(r, lkb, 0);
2212 grant_pending_locks(r);
2216 /* can_be_granted() detected that this lock would block in a conversion
2217 deadlock, so we leave it on the granted queue and return EDEADLK in
2218 the ast for the convert. */
2221 /* it's left on the granted queue */
2222 log_debug(r->res_ls, "deadlock %x node %d sts%d g%d r%d %s",
2223 lkb->lkb_id, lkb->lkb_nodeid, lkb->lkb_status,
2224 lkb->lkb_grmode, lkb->lkb_rqmode, r->res_name);
2225 revert_lock(r, lkb);
2226 queue_cast(r, lkb, -EDEADLK);
2231 /* is_demoted() means the can_be_granted() above set the grmode
2232 to NL, and left us on the granted queue. This auto-demotion
2233 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2234 now grantable. We have to try to grant other converting locks
2235 before we try again to grant this one. */
2237 if (is_demoted(lkb)) {
2238 grant_pending_convert(r, DLM_LOCK_IV);
2239 if (_can_be_granted(r, lkb, 1)) {
2241 queue_cast(r, lkb, 0);
2242 grant_pending_locks(r);
2245 /* else fall through and move to convert queue */
2248 if (can_be_queued(lkb)) {
2249 error = -EINPROGRESS;
2251 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2252 send_blocking_asts(r, lkb);
2258 if (force_blocking_asts(lkb))
2259 send_blocking_asts_all(r, lkb);
2260 queue_cast(r, lkb, -EAGAIN);
2266 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2268 remove_lock(r, lkb);
2269 queue_cast(r, lkb, -DLM_EUNLOCK);
2270 grant_pending_locks(r);
2271 return -DLM_EUNLOCK;
2274 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
2276 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2280 error = revert_lock(r, lkb);
2282 queue_cast(r, lkb, -DLM_ECANCEL);
2283 grant_pending_locks(r);
2284 return -DLM_ECANCEL;
2290 * Four stage 3 varieties:
2291 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2294 /* add a new lkb to a possibly new rsb, called by requesting process */
2296 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2300 /* set_master: sets lkb nodeid from r */
2302 error = set_master(r, lkb);
2311 /* receive_request() calls do_request() on remote node */
2312 error = send_request(r, lkb);
2314 error = do_request(r, lkb);
2319 /* change some property of an existing lkb, e.g. mode */
2321 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2326 /* receive_convert() calls do_convert() on remote node */
2327 error = send_convert(r, lkb);
2329 error = do_convert(r, lkb);
2334 /* remove an existing lkb from the granted queue */
2336 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2341 /* receive_unlock() calls do_unlock() on remote node */
2342 error = send_unlock(r, lkb);
2344 error = do_unlock(r, lkb);
2349 /* remove an existing lkb from the convert or wait queue */
2351 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2356 /* receive_cancel() calls do_cancel() on remote node */
2357 error = send_cancel(r, lkb);
2359 error = do_cancel(r, lkb);
2365 * Four stage 2 varieties:
2366 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2369 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
2370 int len, struct dlm_args *args)
2375 error = validate_lock_args(ls, lkb, args);
2379 error = find_rsb(ls, name, len, R_CREATE, &r);
2386 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
2388 error = _request_lock(r, lkb);
2397 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2398 struct dlm_args *args)
2403 r = lkb->lkb_resource;
2408 error = validate_lock_args(ls, lkb, args);
2412 error = _convert_lock(r, lkb);
2419 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2420 struct dlm_args *args)
2425 r = lkb->lkb_resource;
2430 error = validate_unlock_args(lkb, args);
2434 error = _unlock_lock(r, lkb);
2441 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
2442 struct dlm_args *args)
2447 r = lkb->lkb_resource;
2452 error = validate_unlock_args(lkb, args);
2456 error = _cancel_lock(r, lkb);
2464 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2467 int dlm_lock(dlm_lockspace_t *lockspace,
2469 struct dlm_lksb *lksb,
2472 unsigned int namelen,
2473 uint32_t parent_lkid,
2474 void (*ast) (void *astarg),
2476 void (*bast) (void *astarg, int mode))
2479 struct dlm_lkb *lkb;
2480 struct dlm_args args;
2481 int error, convert = flags & DLM_LKF_CONVERT;
2483 ls = dlm_find_lockspace_local(lockspace);
2487 dlm_lock_recovery(ls);
2490 error = find_lkb(ls, lksb->sb_lkid, &lkb);
2492 error = create_lkb(ls, &lkb);
2497 error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
2498 astarg, bast, &args);
2503 error = convert_lock(ls, lkb, &args);
2505 error = request_lock(ls, lkb, name, namelen, &args);
2507 if (error == -EINPROGRESS)
2510 if (convert || error)
2512 if (error == -EAGAIN || error == -EDEADLK)
2515 dlm_unlock_recovery(ls);
2516 dlm_put_lockspace(ls);
2520 int dlm_unlock(dlm_lockspace_t *lockspace,
2523 struct dlm_lksb *lksb,
2527 struct dlm_lkb *lkb;
2528 struct dlm_args args;
2531 ls = dlm_find_lockspace_local(lockspace);
2535 dlm_lock_recovery(ls);
2537 error = find_lkb(ls, lkid, &lkb);
2541 error = set_unlock_args(flags, astarg, &args);
2545 if (flags & DLM_LKF_CANCEL)
2546 error = cancel_lock(ls, lkb, &args);
2548 error = unlock_lock(ls, lkb, &args);
2550 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2552 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
2557 dlm_unlock_recovery(ls);
2558 dlm_put_lockspace(ls);
2563 * send/receive routines for remote operations and replies
2567 * send_request receive_request
2568 * send_convert receive_convert
2569 * send_unlock receive_unlock
2570 * send_cancel receive_cancel
2571 * send_grant receive_grant
2572 * send_bast receive_bast
2573 * send_lookup receive_lookup
2574 * send_remove receive_remove
2577 * receive_request_reply send_request_reply
2578 * receive_convert_reply send_convert_reply
2579 * receive_unlock_reply send_unlock_reply
2580 * receive_cancel_reply send_cancel_reply
2581 * receive_lookup_reply send_lookup_reply
2584 static int _create_message(struct dlm_ls *ls, int mb_len,
2585 int to_nodeid, int mstype,
2586 struct dlm_message **ms_ret,
2587 struct dlm_mhandle **mh_ret)
2589 struct dlm_message *ms;
2590 struct dlm_mhandle *mh;
2593 /* get_buffer gives us a message handle (mh) that we need to
2594 pass into lowcomms_commit and a message buffer (mb) that we
2595 write our data into */
2597 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
2601 memset(mb, 0, mb_len);
2603 ms = (struct dlm_message *) mb;
2605 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2606 ms->m_header.h_lockspace = ls->ls_global_id;
2607 ms->m_header.h_nodeid = dlm_our_nodeid();
2608 ms->m_header.h_length = mb_len;
2609 ms->m_header.h_cmd = DLM_MSG;
2611 ms->m_type = mstype;
2618 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2619 int to_nodeid, int mstype,
2620 struct dlm_message **ms_ret,
2621 struct dlm_mhandle **mh_ret)
2623 int mb_len = sizeof(struct dlm_message);
2626 case DLM_MSG_REQUEST:
2627 case DLM_MSG_LOOKUP:
2628 case DLM_MSG_REMOVE:
2629 mb_len += r->res_length;
2631 case DLM_MSG_CONVERT:
2632 case DLM_MSG_UNLOCK:
2633 case DLM_MSG_REQUEST_REPLY:
2634 case DLM_MSG_CONVERT_REPLY:
2636 if (lkb && lkb->lkb_lvbptr)
2637 mb_len += r->res_ls->ls_lvblen;
2641 return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
2645 /* further lowcomms enhancements or alternate implementations may make
2646 the return value from this function useful at some point */
2648 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2650 dlm_message_out(ms);
2651 dlm_lowcomms_commit_buffer(mh);
2655 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2656 struct dlm_message *ms)
2658 ms->m_nodeid = lkb->lkb_nodeid;
2659 ms->m_pid = lkb->lkb_ownpid;
2660 ms->m_lkid = lkb->lkb_id;
2661 ms->m_remid = lkb->lkb_remid;
2662 ms->m_exflags = lkb->lkb_exflags;
2663 ms->m_sbflags = lkb->lkb_sbflags;
2664 ms->m_flags = lkb->lkb_flags;
2665 ms->m_lvbseq = lkb->lkb_lvbseq;
2666 ms->m_status = lkb->lkb_status;
2667 ms->m_grmode = lkb->lkb_grmode;
2668 ms->m_rqmode = lkb->lkb_rqmode;
2669 ms->m_hash = r->res_hash;
2671 /* m_result and m_bastmode are set from function args,
2672 not from lkb fields */
2674 if (lkb->lkb_bastaddr)
2675 ms->m_asts |= AST_BAST;
2676 if (lkb->lkb_astaddr)
2677 ms->m_asts |= AST_COMP;
2679 /* compare with switch in create_message; send_remove() doesn't
2682 switch (ms->m_type) {
2683 case DLM_MSG_REQUEST:
2684 case DLM_MSG_LOOKUP:
2685 memcpy(ms->m_extra, r->res_name, r->res_length);
2687 case DLM_MSG_CONVERT:
2688 case DLM_MSG_UNLOCK:
2689 case DLM_MSG_REQUEST_REPLY:
2690 case DLM_MSG_CONVERT_REPLY:
2692 if (!lkb->lkb_lvbptr)
2694 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2699 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2701 struct dlm_message *ms;
2702 struct dlm_mhandle *mh;
2703 int to_nodeid, error;
2705 error = add_to_waiters(lkb, mstype);
2709 to_nodeid = r->res_nodeid;
2711 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2715 send_args(r, lkb, ms);
2717 error = send_message(mh, ms);
2723 remove_from_waiters(lkb, msg_reply_type(mstype));
2727 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2729 return send_common(r, lkb, DLM_MSG_REQUEST);
2732 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2736 error = send_common(r, lkb, DLM_MSG_CONVERT);
2738 /* down conversions go without a reply from the master */
2739 if (!error && down_conversion(lkb)) {
2740 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
2741 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
2742 r->res_ls->ls_stub_ms.m_result = 0;
2743 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
2744 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2750 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2751 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2752 that the master is still correct. */
2754 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2756 return send_common(r, lkb, DLM_MSG_UNLOCK);
2759 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2761 return send_common(r, lkb, DLM_MSG_CANCEL);
2764 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2766 struct dlm_message *ms;
2767 struct dlm_mhandle *mh;
2768 int to_nodeid, error;
2770 to_nodeid = lkb->lkb_nodeid;
2772 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2776 send_args(r, lkb, ms);
2780 error = send_message(mh, ms);
2785 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2787 struct dlm_message *ms;
2788 struct dlm_mhandle *mh;
2789 int to_nodeid, error;
2791 to_nodeid = lkb->lkb_nodeid;
2793 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2797 send_args(r, lkb, ms);
2799 ms->m_bastmode = mode;
2801 error = send_message(mh, ms);
2806 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2808 struct dlm_message *ms;
2809 struct dlm_mhandle *mh;
2810 int to_nodeid, error;
2812 error = add_to_waiters(lkb, DLM_MSG_LOOKUP);
2816 to_nodeid = dlm_dir_nodeid(r);
2818 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2822 send_args(r, lkb, ms);
2824 error = send_message(mh, ms);
2830 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
2834 static int send_remove(struct dlm_rsb *r)
2836 struct dlm_message *ms;
2837 struct dlm_mhandle *mh;
2838 int to_nodeid, error;
2840 to_nodeid = dlm_dir_nodeid(r);
2842 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2846 memcpy(ms->m_extra, r->res_name, r->res_length);
2847 ms->m_hash = r->res_hash;
2849 error = send_message(mh, ms);
2854 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2857 struct dlm_message *ms;
2858 struct dlm_mhandle *mh;
2859 int to_nodeid, error;
2861 to_nodeid = lkb->lkb_nodeid;
2863 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2867 send_args(r, lkb, ms);
2871 error = send_message(mh, ms);
2876 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2878 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2881 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2883 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2886 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2888 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2891 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2893 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2896 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2897 int ret_nodeid, int rv)
2899 struct dlm_rsb *r = &ls->ls_stub_rsb;
2900 struct dlm_message *ms;
2901 struct dlm_mhandle *mh;
2902 int error, nodeid = ms_in->m_header.h_nodeid;
2904 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2908 ms->m_lkid = ms_in->m_lkid;
2910 ms->m_nodeid = ret_nodeid;
2912 error = send_message(mh, ms);
2917 /* which args we save from a received message depends heavily on the type
2918 of message, unlike the send side where we can safely send everything about
2919 the lkb for any type of message */
2921 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2923 lkb->lkb_exflags = ms->m_exflags;
2924 lkb->lkb_sbflags = ms->m_sbflags;
2925 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2926 (ms->m_flags & 0x0000FFFF);
2929 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2931 lkb->lkb_sbflags = ms->m_sbflags;
2932 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2933 (ms->m_flags & 0x0000FFFF);
2936 static int receive_extralen(struct dlm_message *ms)
2938 return (ms->m_header.h_length - sizeof(struct dlm_message));
2941 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2942 struct dlm_message *ms)
2946 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2947 if (!lkb->lkb_lvbptr)
2948 lkb->lkb_lvbptr = allocate_lvb(ls);
2949 if (!lkb->lkb_lvbptr)
2951 len = receive_extralen(ms);
2952 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2957 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2958 struct dlm_message *ms)
2960 lkb->lkb_nodeid = ms->m_header.h_nodeid;
2961 lkb->lkb_ownpid = ms->m_pid;
2962 lkb->lkb_remid = ms->m_lkid;
2963 lkb->lkb_grmode = DLM_LOCK_IV;
2964 lkb->lkb_rqmode = ms->m_rqmode;
2965 lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
2966 lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
2968 DLM_ASSERT(is_master_copy(lkb), dlm_print_lkb(lkb););
2970 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2971 /* lkb was just created so there won't be an lvb yet */
2972 lkb->lkb_lvbptr = allocate_lvb(ls);
2973 if (!lkb->lkb_lvbptr)
2980 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2981 struct dlm_message *ms)
2983 if (lkb->lkb_nodeid != ms->m_header.h_nodeid) {
2984 log_error(ls, "convert_args nodeid %d %d lkid %x %x",
2985 lkb->lkb_nodeid, ms->m_header.h_nodeid,
2986 lkb->lkb_id, lkb->lkb_remid);
2990 if (!is_master_copy(lkb))
2993 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2996 if (receive_lvb(ls, lkb, ms))
2999 lkb->lkb_rqmode = ms->m_rqmode;
3000 lkb->lkb_lvbseq = ms->m_lvbseq;
3005 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3006 struct dlm_message *ms)
3008 if (!is_master_copy(lkb))
3010 if (receive_lvb(ls, lkb, ms))
3015 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3016 uses to send a reply and that the remote end uses to process the reply. */
3018 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3020 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3021 lkb->lkb_nodeid = ms->m_header.h_nodeid;
3022 lkb->lkb_remid = ms->m_lkid;
3025 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
3027 struct dlm_lkb *lkb;
3031 error = create_lkb(ls, &lkb);
3035 receive_flags(lkb, ms);
3036 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3037 error = receive_request_args(ls, lkb, ms);
3043 namelen = receive_extralen(ms);
3045 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
3054 error = do_request(r, lkb);
3055 send_request_reply(r, lkb, error);
3060 if (error == -EINPROGRESS)
3067 setup_stub_lkb(ls, ms);
3068 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3071 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
3073 struct dlm_lkb *lkb;
3075 int error, reply = 1;
3077 error = find_lkb(ls, ms->m_remid, &lkb);
3081 r = lkb->lkb_resource;
3086 receive_flags(lkb, ms);
3087 error = receive_convert_args(ls, lkb, ms);
3090 reply = !down_conversion(lkb);
3092 error = do_convert(r, lkb);
3095 send_convert_reply(r, lkb, error);
3103 setup_stub_lkb(ls, ms);
3104 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3107 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
3109 struct dlm_lkb *lkb;
3113 error = find_lkb(ls, ms->m_remid, &lkb);
3117 r = lkb->lkb_resource;
3122 receive_flags(lkb, ms);
3123 error = receive_unlock_args(ls, lkb, ms);
3127 error = do_unlock(r, lkb);
3129 send_unlock_reply(r, lkb, error);
3137 setup_stub_lkb(ls, ms);
3138 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3141 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
3143 struct dlm_lkb *lkb;
3147 error = find_lkb(ls, ms->m_remid, &lkb);
3151 receive_flags(lkb, ms);
3153 r = lkb->lkb_resource;
3158 error = do_cancel(r, lkb);
3159 send_cancel_reply(r, lkb, error);
3167 setup_stub_lkb(ls, ms);
3168 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
3171 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
3173 struct dlm_lkb *lkb;
3177 error = find_lkb(ls, ms->m_remid, &lkb);
3179 log_error(ls, "receive_grant no lkb");
3182 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3184 r = lkb->lkb_resource;
3189 receive_flags_reply(lkb, ms);
3190 if (is_altmode(lkb))
3191 munge_altmode(lkb, ms);
3192 grant_lock_pc(r, lkb, ms);
3193 queue_cast(r, lkb, 0);
3200 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
3202 struct dlm_lkb *lkb;
3206 error = find_lkb(ls, ms->m_remid, &lkb);
3208 log_error(ls, "receive_bast no lkb");
3211 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3213 r = lkb->lkb_resource;
3218 queue_bast(r, lkb, ms->m_bastmode);
3225 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
3227 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
3229 from_nodeid = ms->m_header.h_nodeid;
3230 our_nodeid = dlm_our_nodeid();
3232 len = receive_extralen(ms);
3234 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3235 if (dir_nodeid != our_nodeid) {
3236 log_error(ls, "lookup dir_nodeid %d from %d",
3237 dir_nodeid, from_nodeid);
3243 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
3245 /* Optimization: we're master so treat lookup as a request */
3246 if (!error && ret_nodeid == our_nodeid) {
3247 receive_request(ls, ms);
3251 send_lookup_reply(ls, ms, ret_nodeid, error);
3254 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
3256 int len, dir_nodeid, from_nodeid;
3258 from_nodeid = ms->m_header.h_nodeid;
3260 len = receive_extralen(ms);
3262 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
3263 if (dir_nodeid != dlm_our_nodeid()) {
3264 log_error(ls, "remove dir entry dir_nodeid %d from %d",
3265 dir_nodeid, from_nodeid);
3269 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
3272 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
3274 do_purge(ls, ms->m_nodeid, ms->m_pid);
3277 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
3279 struct dlm_lkb *lkb;
3281 int error, mstype, result;
3283 error = find_lkb(ls, ms->m_remid, &lkb);
3285 log_error(ls, "receive_request_reply no lkb");
3288 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3290 r = lkb->lkb_resource;
3294 mstype = lkb->lkb_wait_type;
3295 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
3299 /* Optimization: the dir node was also the master, so it took our
3300 lookup as a request and sent request reply instead of lookup reply */
3301 if (mstype == DLM_MSG_LOOKUP) {
3302 r->res_nodeid = ms->m_header.h_nodeid;
3303 lkb->lkb_nodeid = r->res_nodeid;
3306 /* this is the value returned from do_request() on the master */
3307 result = ms->m_result;
3311 /* request would block (be queued) on remote master */
3312 queue_cast(r, lkb, -EAGAIN);
3313 confirm_master(r, -EAGAIN);
3314 unhold_lkb(lkb); /* undoes create_lkb() */
3319 /* request was queued or granted on remote master */
3320 receive_flags_reply(lkb, ms);
3321 lkb->lkb_remid = ms->m_lkid;
3322 if (is_altmode(lkb))
3323 munge_altmode(lkb, ms);
3325 add_lkb(r, lkb, DLM_LKSTS_WAITING);
3328 grant_lock_pc(r, lkb, ms);
3329 queue_cast(r, lkb, 0);
3331 confirm_master(r, result);
3336 /* find_rsb failed to find rsb or rsb wasn't master */
3337 log_debug(ls, "receive_request_reply %x %x master diff %d %d",
3338 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result);
3340 lkb->lkb_nodeid = -1;
3342 if (is_overlap(lkb)) {
3343 /* we'll ignore error in cancel/unlock reply */
3344 queue_cast_overlap(r, lkb);
3345 unhold_lkb(lkb); /* undoes create_lkb() */
3347 _request_lock(r, lkb);
3351 log_error(ls, "receive_request_reply %x error %d",
3352 lkb->lkb_id, result);
3355 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
3356 log_debug(ls, "receive_request_reply %x result %d unlock",
3357 lkb->lkb_id, result);
3358 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3359 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3360 send_unlock(r, lkb);
3361 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
3362 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
3363 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3364 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3365 send_cancel(r, lkb);
3367 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3368 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3376 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3377 struct dlm_message *ms)
3379 /* this is the value returned from do_convert() on the master */
3380 switch (ms->m_result) {
3382 /* convert would block (be queued) on remote master */
3383 queue_cast(r, lkb, -EAGAIN);
3387 receive_flags_reply(lkb, ms);
3388 revert_lock_pc(r, lkb);
3389 queue_cast(r, lkb, -EDEADLK);
3393 /* convert was queued on remote master */
3394 receive_flags_reply(lkb, ms);
3395 if (is_demoted(lkb))
3396 munge_demoted(lkb, ms);
3398 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3403 /* convert was granted on remote master */
3404 receive_flags_reply(lkb, ms);
3405 if (is_demoted(lkb))
3406 munge_demoted(lkb, ms);
3407 grant_lock_pc(r, lkb, ms);
3408 queue_cast(r, lkb, 0);
3412 log_error(r->res_ls, "receive_convert_reply %x error %d",
3413 lkb->lkb_id, ms->m_result);
3417 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3419 struct dlm_rsb *r = lkb->lkb_resource;
3425 /* stub reply can happen with waiters_mutex held */
3426 error = remove_from_waiters_ms(lkb, ms);
3430 __receive_convert_reply(r, lkb, ms);
3436 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
3438 struct dlm_lkb *lkb;
3441 error = find_lkb(ls, ms->m_remid, &lkb);
3443 log_error(ls, "receive_convert_reply no lkb");
3446 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3448 _receive_convert_reply(lkb, ms);
3452 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3454 struct dlm_rsb *r = lkb->lkb_resource;
3460 /* stub reply can happen with waiters_mutex held */
3461 error = remove_from_waiters_ms(lkb, ms);
3465 /* this is the value returned from do_unlock() on the master */
3467 switch (ms->m_result) {
3469 receive_flags_reply(lkb, ms);
3470 remove_lock_pc(r, lkb);
3471 queue_cast(r, lkb, -DLM_EUNLOCK);
3476 log_error(r->res_ls, "receive_unlock_reply %x error %d",
3477 lkb->lkb_id, ms->m_result);
3484 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
3486 struct dlm_lkb *lkb;
3489 error = find_lkb(ls, ms->m_remid, &lkb);
3491 log_error(ls, "receive_unlock_reply no lkb");
3494 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3496 _receive_unlock_reply(lkb, ms);
3500 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3502 struct dlm_rsb *r = lkb->lkb_resource;
3508 /* stub reply can happen with waiters_mutex held */
3509 error = remove_from_waiters_ms(lkb, ms);
3513 /* this is the value returned from do_cancel() on the master */
3515 switch (ms->m_result) {
3517 receive_flags_reply(lkb, ms);
3518 revert_lock_pc(r, lkb);
3519 queue_cast(r, lkb, -DLM_ECANCEL);
3524 log_error(r->res_ls, "receive_cancel_reply %x error %d",
3525 lkb->lkb_id, ms->m_result);
3532 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
3534 struct dlm_lkb *lkb;
3537 error = find_lkb(ls, ms->m_remid, &lkb);
3539 log_error(ls, "receive_cancel_reply no lkb");
3542 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3544 _receive_cancel_reply(lkb, ms);
3548 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
3550 struct dlm_lkb *lkb;
3552 int error, ret_nodeid;
3554 error = find_lkb(ls, ms->m_lkid, &lkb);
3556 log_error(ls, "receive_lookup_reply no lkb");
3560 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
3561 FIXME: will a non-zero error ever be returned? */
3563 r = lkb->lkb_resource;
3567 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3571 ret_nodeid = ms->m_nodeid;
3572 if (ret_nodeid == dlm_our_nodeid()) {
3575 r->res_first_lkid = 0;
3577 /* set_master() will copy res_nodeid to lkb_nodeid */
3578 r->res_nodeid = ret_nodeid;
3581 if (is_overlap(lkb)) {
3582 log_debug(ls, "receive_lookup_reply %x unlock %x",
3583 lkb->lkb_id, lkb->lkb_flags);
3584 queue_cast_overlap(r, lkb);
3585 unhold_lkb(lkb); /* undoes create_lkb() */
3589 _request_lock(r, lkb);
3593 process_lookup_list(r);
3600 int dlm_receive_message(struct dlm_header *hd, int nodeid, int recovery)
3602 struct dlm_message *ms = (struct dlm_message *) hd;
3609 ls = dlm_find_lockspace_global(hd->h_lockspace);
3611 log_print("drop message %d from %d for unknown lockspace %d",
3612 ms->m_type, nodeid, hd->h_lockspace);
3616 /* recovery may have just ended leaving a bunch of backed-up requests
3617 in the requestqueue; wait while dlm_recoverd clears them */
3620 dlm_wait_requestqueue(ls);
3622 /* recovery may have just started while there were a bunch of
3623 in-flight requests -- save them in requestqueue to be processed
3624 after recovery. we can't let dlm_recvd block on the recovery
3625 lock. if dlm_recoverd is calling this function to clear the
3626 requestqueue, it needs to be interrupted (-EINTR) if another
3627 recovery operation is starting. */
3630 if (dlm_locking_stopped(ls)) {
3635 error = dlm_add_requestqueue(ls, nodeid, hd);
3636 if (error == -EAGAIN)
3644 if (dlm_lock_recovery_try(ls))
3649 switch (ms->m_type) {
3651 /* messages sent to a master node */
3653 case DLM_MSG_REQUEST:
3654 receive_request(ls, ms);
3657 case DLM_MSG_CONVERT:
3658 receive_convert(ls, ms);
3661 case DLM_MSG_UNLOCK:
3662 receive_unlock(ls, ms);
3665 case DLM_MSG_CANCEL:
3666 receive_cancel(ls, ms);
3669 /* messages sent from a master node (replies to above) */
3671 case DLM_MSG_REQUEST_REPLY:
3672 receive_request_reply(ls, ms);
3675 case DLM_MSG_CONVERT_REPLY:
3676 receive_convert_reply(ls, ms);
3679 case DLM_MSG_UNLOCK_REPLY:
3680 receive_unlock_reply(ls, ms);
3683 case DLM_MSG_CANCEL_REPLY:
3684 receive_cancel_reply(ls, ms);
3687 /* messages sent from a master node (only two types of async msg) */
3690 receive_grant(ls, ms);
3694 receive_bast(ls, ms);
3697 /* messages sent to a dir node */
3699 case DLM_MSG_LOOKUP:
3700 receive_lookup(ls, ms);
3703 case DLM_MSG_REMOVE:
3704 receive_remove(ls, ms);
3707 /* messages sent from a dir node (remove has no reply) */
3709 case DLM_MSG_LOOKUP_REPLY:
3710 receive_lookup_reply(ls, ms);
3713 /* other messages */
3716 receive_purge(ls, ms);
3720 log_error(ls, "unknown message type %d", ms->m_type);
3723 dlm_unlock_recovery(ls);
3725 dlm_put_lockspace(ls);
3735 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3737 if (middle_conversion(lkb)) {
3739 ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY;
3740 ls->ls_stub_ms.m_result = -EINPROGRESS;
3741 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3742 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3744 /* Same special case as in receive_rcom_lock_args() */
3745 lkb->lkb_grmode = DLM_LOCK_IV;
3746 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3749 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3750 lkb->lkb_flags |= DLM_IFL_RESEND;
3753 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3754 conversions are async; there's no reply from the remote master */
3757 /* A waiting lkb needs recovery if the master node has failed, or
3758 the master node is changing (only when no directory is used) */
3760 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3762 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3765 if (!dlm_no_directory(ls))
3768 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3774 /* Recovery for locks that are waiting for replies from nodes that are now
3775 gone. We can just complete unlocks and cancels by faking a reply from the
3776 dead node. Requests and up-conversions we flag to be resent after
3777 recovery. Down-conversions can just be completed with a fake reply like
3778 unlocks. Conversions between PR and CW need special attention. */
3780 void dlm_recover_waiters_pre(struct dlm_ls *ls)
3782 struct dlm_lkb *lkb, *safe;
3784 mutex_lock(&ls->ls_waiters_mutex);
3786 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3787 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3788 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3790 /* all outstanding lookups, regardless of destination will be
3791 resent after recovery is done */
3793 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3794 lkb->lkb_flags |= DLM_IFL_RESEND;
3798 if (!waiter_needs_recovery(ls, lkb))
3801 switch (lkb->lkb_wait_type) {
3803 case DLM_MSG_REQUEST:
3804 lkb->lkb_flags |= DLM_IFL_RESEND;
3807 case DLM_MSG_CONVERT:
3808 recover_convert_waiter(ls, lkb);
3811 case DLM_MSG_UNLOCK:
3813 ls->ls_stub_ms.m_type = DLM_MSG_UNLOCK_REPLY;
3814 ls->ls_stub_ms.m_result = -DLM_EUNLOCK;
3815 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3816 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
3820 case DLM_MSG_CANCEL:
3822 ls->ls_stub_ms.m_type = DLM_MSG_CANCEL_REPLY;
3823 ls->ls_stub_ms.m_result = -DLM_ECANCEL;
3824 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3825 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
3830 log_error(ls, "invalid lkb wait_type %d",
3831 lkb->lkb_wait_type);
3835 mutex_unlock(&ls->ls_waiters_mutex);
3838 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
3840 struct dlm_lkb *lkb;
3843 mutex_lock(&ls->ls_waiters_mutex);
3844 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
3845 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3851 mutex_unlock(&ls->ls_waiters_mutex);
3858 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
3859 master or dir-node for r. Processing the lkb may result in it being placed
3862 /* We do this after normal locking has been enabled and any saved messages
3863 (in requestqueue) have been processed. We should be confident that at
3864 this point we won't get or process a reply to any of these waiting
3865 operations. But, new ops may be coming in on the rsbs/locks here from
3866 userspace or remotely. */
3868 /* there may have been an overlap unlock/cancel prior to recovery or after
3869 recovery. if before, the lkb may still have a pos wait_count; if after, the
3870 overlap flag would just have been set and nothing new sent. we can be
3871 confident here than any replies to either the initial op or overlap ops
3872 prior to recovery have been received. */
3874 int dlm_recover_waiters_post(struct dlm_ls *ls)
3876 struct dlm_lkb *lkb;
3878 int error = 0, mstype, err, oc, ou;
3881 if (dlm_locking_stopped(ls)) {
3882 log_debug(ls, "recover_waiters_post aborted");
3887 lkb = find_resend_waiter(ls);
3891 r = lkb->lkb_resource;
3895 mstype = lkb->lkb_wait_type;
3896 oc = is_overlap_cancel(lkb);
3897 ou = is_overlap_unlock(lkb);
3900 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
3901 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
3903 /* At this point we assume that we won't get a reply to any
3904 previous op or overlap op on this lock. First, do a big
3905 remove_from_waiters() for all previous ops. */
3907 lkb->lkb_flags &= ~DLM_IFL_RESEND;
3908 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
3909 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
3910 lkb->lkb_wait_type = 0;
3911 lkb->lkb_wait_count = 0;
3912 mutex_lock(&ls->ls_waiters_mutex);
3913 list_del_init(&lkb->lkb_wait_reply);
3914 mutex_unlock(&ls->ls_waiters_mutex);
3915 unhold_lkb(lkb); /* for waiters list */
3918 /* do an unlock or cancel instead of resending */
3920 case DLM_MSG_LOOKUP:
3921 case DLM_MSG_REQUEST:
3922 queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
3924 unhold_lkb(lkb); /* undoes create_lkb() */
3926 case DLM_MSG_CONVERT:
3928 queue_cast(r, lkb, -DLM_ECANCEL);
3930 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
3931 _unlock_lock(r, lkb);
3939 case DLM_MSG_LOOKUP:
3940 case DLM_MSG_REQUEST:
3941 _request_lock(r, lkb);
3943 confirm_master(r, 0);
3945 case DLM_MSG_CONVERT:
3946 _convert_lock(r, lkb);
3954 log_error(ls, "recover_waiters_post %x %d %x %d %d",
3955 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou);
3964 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
3965 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
3967 struct dlm_ls *ls = r->res_ls;
3968 struct dlm_lkb *lkb, *safe;
3970 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
3971 if (test(ls, lkb)) {
3972 rsb_set_flag(r, RSB_LOCKS_PURGED);
3974 /* this put should free the lkb */
3975 if (!dlm_put_lkb(lkb))
3976 log_error(ls, "purged lkb not released");
3981 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3983 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
3986 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3988 return is_master_copy(lkb);
3991 static void purge_dead_locks(struct dlm_rsb *r)
3993 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
3994 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
3995 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
3998 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
4000 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
4001 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
4002 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
4005 /* Get rid of locks held by nodes that are gone. */
4007 int dlm_purge_locks(struct dlm_ls *ls)
4011 log_debug(ls, "dlm_purge_locks");
4013 down_write(&ls->ls_root_sem);
4014 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
4018 purge_dead_locks(r);
4024 up_write(&ls->ls_root_sem);
4029 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
4031 struct dlm_rsb *r, *r_ret = NULL;
4033 read_lock(&ls->ls_rsbtbl[bucket].lock);
4034 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
4035 if (!rsb_flag(r, RSB_LOCKS_PURGED))
4038 rsb_clear_flag(r, RSB_LOCKS_PURGED);
4042 read_unlock(&ls->ls_rsbtbl[bucket].lock);
4046 void dlm_grant_after_purge(struct dlm_ls *ls)
4052 r = find_purged_rsb(ls, bucket);
4054 if (bucket == ls->ls_rsbtbl_size - 1)
4061 grant_pending_locks(r);
4062 confirm_master(r, 0);
4070 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
4073 struct dlm_lkb *lkb;
4075 list_for_each_entry(lkb, head, lkb_statequeue) {
4076 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
4082 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
4085 struct dlm_lkb *lkb;
4087 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
4090 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
4093 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
4099 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
4100 struct dlm_rsb *r, struct dlm_rcom *rc)
4102 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4105 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
4106 lkb->lkb_ownpid = rl->rl_ownpid;
4107 lkb->lkb_remid = rl->rl_lkid;
4108 lkb->lkb_exflags = rl->rl_exflags;
4109 lkb->lkb_flags = rl->rl_flags & 0x0000FFFF;
4110 lkb->lkb_flags |= DLM_IFL_MSTCPY;
4111 lkb->lkb_lvbseq = rl->rl_lvbseq;
4112 lkb->lkb_rqmode = rl->rl_rqmode;
4113 lkb->lkb_grmode = rl->rl_grmode;
4114 /* don't set lkb_status because add_lkb wants to itself */
4116 lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
4117 lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
4119 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
4120 lkb->lkb_lvbptr = allocate_lvb(ls);
4121 if (!lkb->lkb_lvbptr)
4123 lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
4124 sizeof(struct rcom_lock);
4125 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
4128 /* Conversions between PR and CW (middle modes) need special handling.
4129 The real granted mode of these converting locks cannot be determined
4130 until all locks have been rebuilt on the rsb (recover_conversion) */
4132 if (rl->rl_wait_type == DLM_MSG_CONVERT && middle_conversion(lkb)) {
4133 rl->rl_status = DLM_LKSTS_CONVERT;
4134 lkb->lkb_grmode = DLM_LOCK_IV;
4135 rsb_set_flag(r, RSB_RECOVER_CONVERT);
4141 /* This lkb may have been recovered in a previous aborted recovery so we need
4142 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4143 If so we just send back a standard reply. If not, we create a new lkb with
4144 the given values and send back our lkid. We send back our lkid by sending
4145 back the rcom_lock struct we got but with the remid field filled in. */
4147 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4149 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4151 struct dlm_lkb *lkb;
4154 if (rl->rl_parent_lkid) {
4155 error = -EOPNOTSUPP;
4159 error = find_rsb(ls, rl->rl_name, rl->rl_namelen, R_MASTER, &r);
4165 lkb = search_remid(r, rc->rc_header.h_nodeid, rl->rl_lkid);
4171 error = create_lkb(ls, &lkb);
4175 error = receive_rcom_lock_args(ls, lkb, r, rc);
4182 add_lkb(r, lkb, rl->rl_status);
4186 /* this is the new value returned to the lock holder for
4187 saving in its process-copy lkb */
4188 rl->rl_remid = lkb->lkb_id;
4195 log_print("recover_master_copy %d %x", error, rl->rl_lkid);
4196 rl->rl_result = error;
4200 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
4202 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
4204 struct dlm_lkb *lkb;
4207 error = find_lkb(ls, rl->rl_lkid, &lkb);
4209 log_error(ls, "recover_process_copy no lkid %x", rl->rl_lkid);
4213 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
4215 error = rl->rl_result;
4217 r = lkb->lkb_resource;
4223 /* There's a chance the new master received our lock before
4224 dlm_recover_master_reply(), this wouldn't happen if we did
4225 a barrier between recover_masters and recover_locks. */
4226 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
4227 (unsigned long)r, r->res_name);
4228 dlm_send_rcom_lock(r, lkb);
4231 log_debug(ls, "master copy exists %x", lkb->lkb_id);
4234 lkb->lkb_remid = rl->rl_remid;
4237 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
4238 error, lkb->lkb_id);
4241 /* an ack for dlm_recover_locks() which waits for replies from
4242 all the locks it sends to new masters */
4243 dlm_recovered_lock(r);
4252 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
4253 int mode, uint32_t flags, void *name, unsigned int namelen,
4254 unsigned long timeout_cs)
4256 struct dlm_lkb *lkb;
4257 struct dlm_args args;
4260 dlm_lock_recovery(ls);
4262 error = create_lkb(ls, &lkb);
4268 if (flags & DLM_LKF_VALBLK) {
4269 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
4270 if (!ua->lksb.sb_lvbptr) {
4278 /* After ua is attached to lkb it will be freed by free_lkb().
4279 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4280 lock and that lkb_astparam is the dlm_user_args structure. */
4282 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
4283 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
4284 lkb->lkb_flags |= DLM_IFL_USER;
4285 ua->old_mode = DLM_LOCK_IV;
4292 error = request_lock(ls, lkb, name, namelen, &args);
4308 /* add this new lkb to the per-process list of locks */
4309 spin_lock(&ua->proc->locks_spin);
4311 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
4312 spin_unlock(&ua->proc->locks_spin);
4314 dlm_unlock_recovery(ls);
4318 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4319 int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
4320 unsigned long timeout_cs)
4322 struct dlm_lkb *lkb;
4323 struct dlm_args args;
4324 struct dlm_user_args *ua;
4327 dlm_lock_recovery(ls);
4329 error = find_lkb(ls, lkid, &lkb);
4333 /* user can change the params on its lock when it converts it, or
4334 add an lvb that didn't exist before */
4336 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4338 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
4339 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
4340 if (!ua->lksb.sb_lvbptr) {
4345 if (lvb_in && ua->lksb.sb_lvbptr)
4346 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4348 ua->xid = ua_tmp->xid;
4349 ua->castparam = ua_tmp->castparam;
4350 ua->castaddr = ua_tmp->castaddr;
4351 ua->bastparam = ua_tmp->bastparam;
4352 ua->bastaddr = ua_tmp->bastaddr;
4353 ua->user_lksb = ua_tmp->user_lksb;
4354 ua->old_mode = lkb->lkb_grmode;
4356 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
4357 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
4361 error = convert_lock(ls, lkb, &args);
4363 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
4368 dlm_unlock_recovery(ls);
4373 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4374 uint32_t flags, uint32_t lkid, char *lvb_in)
4376 struct dlm_lkb *lkb;
4377 struct dlm_args args;
4378 struct dlm_user_args *ua;
4381 dlm_lock_recovery(ls);
4383 error = find_lkb(ls, lkid, &lkb);
4387 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4389 if (lvb_in && ua->lksb.sb_lvbptr)
4390 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
4391 ua->castparam = ua_tmp->castparam;
4392 ua->user_lksb = ua_tmp->user_lksb;
4394 error = set_unlock_args(flags, ua, &args);
4398 error = unlock_lock(ls, lkb, &args);
4400 if (error == -DLM_EUNLOCK)
4402 /* from validate_unlock_args() */
4403 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
4408 spin_lock(&ua->proc->locks_spin);
4409 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4410 if (!list_empty(&lkb->lkb_ownqueue))
4411 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
4412 spin_unlock(&ua->proc->locks_spin);
4416 dlm_unlock_recovery(ls);
4421 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
4422 uint32_t flags, uint32_t lkid)
4424 struct dlm_lkb *lkb;
4425 struct dlm_args args;
4426 struct dlm_user_args *ua;
4429 dlm_lock_recovery(ls);
4431 error = find_lkb(ls, lkid, &lkb);
4435 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4436 ua->castparam = ua_tmp->castparam;
4437 ua->user_lksb = ua_tmp->user_lksb;
4439 error = set_unlock_args(flags, ua, &args);
4443 error = cancel_lock(ls, lkb, &args);
4445 if (error == -DLM_ECANCEL)
4447 /* from validate_unlock_args() */
4448 if (error == -EBUSY)
4453 dlm_unlock_recovery(ls);
4458 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
4460 struct dlm_lkb *lkb;
4461 struct dlm_args args;
4462 struct dlm_user_args *ua;
4466 dlm_lock_recovery(ls);
4468 error = find_lkb(ls, lkid, &lkb);
4472 ua = (struct dlm_user_args *)lkb->lkb_astparam;
4474 error = set_unlock_args(flags, ua, &args);
4478 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4480 r = lkb->lkb_resource;
4484 error = validate_unlock_args(lkb, &args);
4487 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
4489 error = _cancel_lock(r, lkb);
4494 if (error == -DLM_ECANCEL)
4496 /* from validate_unlock_args() */
4497 if (error == -EBUSY)
4502 dlm_unlock_recovery(ls);
4506 /* lkb's that are removed from the waiters list by revert are just left on the
4507 orphans list with the granted orphan locks, to be freed by purge */
4509 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4511 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4512 struct dlm_args args;
4516 mutex_lock(&ls->ls_orphans_mutex);
4517 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
4518 mutex_unlock(&ls->ls_orphans_mutex);
4520 set_unlock_args(0, ua, &args);
4522 error = cancel_lock(ls, lkb, &args);
4523 if (error == -DLM_ECANCEL)
4528 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4529 Regardless of what rsb queue the lock is on, it's removed and freed. */
4531 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
4533 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
4534 struct dlm_args args;
4537 set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
4539 error = unlock_lock(ls, lkb, &args);
4540 if (error == -DLM_EUNLOCK)
4545 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4546 (which does lock_rsb) due to deadlock with receiving a message that does
4547 lock_rsb followed by dlm_user_add_ast() */
4549 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
4550 struct dlm_user_proc *proc)
4552 struct dlm_lkb *lkb = NULL;
4554 mutex_lock(&ls->ls_clear_proc_locks);
4555 if (list_empty(&proc->locks))
4558 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
4559 list_del_init(&lkb->lkb_ownqueue);
4561 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4562 lkb->lkb_flags |= DLM_IFL_ORPHAN;
4564 lkb->lkb_flags |= DLM_IFL_DEAD;
4566 mutex_unlock(&ls->ls_clear_proc_locks);
4570 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4571 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4572 which we clear here. */
4574 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
4575 list, and no more device_writes should add lkb's to proc->locks list; so we
4576 shouldn't need to take asts_spin or locks_spin here. this assumes that
4577 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4580 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4582 struct dlm_lkb *lkb, *safe;
4584 dlm_lock_recovery(ls);
4587 lkb = del_proc_lock(ls, proc);
4591 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
4592 orphan_proc_lock(ls, lkb);
4594 unlock_proc_lock(ls, lkb);
4596 /* this removes the reference for the proc->locks list
4597 added by dlm_user_request, it may result in the lkb
4603 mutex_lock(&ls->ls_clear_proc_locks);
4605 /* in-progress unlocks */
4606 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4607 list_del_init(&lkb->lkb_ownqueue);
4608 lkb->lkb_flags |= DLM_IFL_DEAD;
4612 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4613 list_del(&lkb->lkb_astqueue);
4617 mutex_unlock(&ls->ls_clear_proc_locks);
4618 dlm_unlock_recovery(ls);
4621 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
4623 struct dlm_lkb *lkb, *safe;
4627 spin_lock(&proc->locks_spin);
4628 if (!list_empty(&proc->locks)) {
4629 lkb = list_entry(proc->locks.next, struct dlm_lkb,
4631 list_del_init(&lkb->lkb_ownqueue);
4633 spin_unlock(&proc->locks_spin);
4638 lkb->lkb_flags |= DLM_IFL_DEAD;
4639 unlock_proc_lock(ls, lkb);
4640 dlm_put_lkb(lkb); /* ref from proc->locks list */
4643 spin_lock(&proc->locks_spin);
4644 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
4645 list_del_init(&lkb->lkb_ownqueue);
4646 lkb->lkb_flags |= DLM_IFL_DEAD;
4649 spin_unlock(&proc->locks_spin);
4651 spin_lock(&proc->asts_spin);
4652 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
4653 list_del(&lkb->lkb_astqueue);
4656 spin_unlock(&proc->asts_spin);
4659 /* pid of 0 means purge all orphans */
4661 static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
4663 struct dlm_lkb *lkb, *safe;
4665 mutex_lock(&ls->ls_orphans_mutex);
4666 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
4667 if (pid && lkb->lkb_ownpid != pid)
4669 unlock_proc_lock(ls, lkb);
4670 list_del_init(&lkb->lkb_ownqueue);
4673 mutex_unlock(&ls->ls_orphans_mutex);
4676 static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
4678 struct dlm_message *ms;
4679 struct dlm_mhandle *mh;
4682 error = _create_message(ls, sizeof(struct dlm_message), nodeid,
4683 DLM_MSG_PURGE, &ms, &mh);
4686 ms->m_nodeid = nodeid;
4689 return send_message(mh, ms);
4692 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
4693 int nodeid, int pid)
4697 if (nodeid != dlm_our_nodeid()) {
4698 error = send_purge(ls, nodeid, pid);
4700 dlm_lock_recovery(ls);
4701 if (pid == current->pid)
4702 purge_proc_locks(ls, proc);
4704 do_purge(ls, nodeid, pid);
4705 dlm_unlock_recovery(ls);