1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005 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 void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
86 struct dlm_message *ms);
87 static int receive_extralen(struct dlm_message *ms);
89 #define FAKE_USER_AST (void*)0xff00ff00
92 * Lock compatibilty matrix - thanks Steve
93 * UN = Unlocked state. Not really a state, used as a flag
94 * PD = Padding. Used to make the matrix a nice power of two in size
95 * Other states are the same as the VMS DLM.
96 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
99 static const int __dlm_compat_matrix[8][8] = {
100 /* UN NL CR CW PR PW EX PD */
101 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
102 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
103 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
104 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
105 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
106 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
107 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
108 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
112 * This defines the direction of transfer of LVB data.
113 * Granted mode is the row; requested mode is the column.
114 * Usage: matrix[grmode+1][rqmode+1]
115 * 1 = LVB is returned to the caller
116 * 0 = LVB is written to the resource
117 * -1 = nothing happens to the LVB
120 const int dlm_lvb_operations[8][8] = {
121 /* UN NL CR CW PR PW EX PD*/
122 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
123 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
124 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
125 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
126 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
127 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
128 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
129 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
132 #define modes_compat(gr, rq) \
133 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
135 int dlm_modes_compat(int mode1, int mode2)
137 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
141 * Compatibility matrix for conversions with QUECVT set.
142 * Granted mode is the row; requested mode is the column.
143 * Usage: matrix[grmode+1][rqmode+1]
146 static const int __quecvt_compat_matrix[8][8] = {
147 /* UN NL CR CW PR PW EX PD */
148 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
149 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
150 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
151 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
152 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
153 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
154 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
155 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
158 void dlm_print_lkb(struct dlm_lkb *lkb)
160 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
161 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
162 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
163 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
164 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
167 void dlm_print_rsb(struct dlm_rsb *r)
169 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
170 r->res_nodeid, r->res_flags, r->res_first_lkid,
171 r->res_recover_locks_count, r->res_name);
174 /* Threads cannot use the lockspace while it's being recovered */
176 static inline void lock_recovery(struct dlm_ls *ls)
178 down_read(&ls->ls_in_recovery);
181 static inline void unlock_recovery(struct dlm_ls *ls)
183 up_read(&ls->ls_in_recovery);
186 static inline int lock_recovery_try(struct dlm_ls *ls)
188 return down_read_trylock(&ls->ls_in_recovery);
191 static inline int can_be_queued(struct dlm_lkb *lkb)
193 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
196 static inline int force_blocking_asts(struct dlm_lkb *lkb)
198 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
201 static inline int is_demoted(struct dlm_lkb *lkb)
203 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
206 static inline int is_remote(struct dlm_rsb *r)
208 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
209 return !!r->res_nodeid;
212 static inline int is_process_copy(struct dlm_lkb *lkb)
214 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
217 static inline int is_master_copy(struct dlm_lkb *lkb)
219 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
220 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
221 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
224 static inline int middle_conversion(struct dlm_lkb *lkb)
226 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
227 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
232 static inline int down_conversion(struct dlm_lkb *lkb)
234 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
237 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
239 if (is_master_copy(lkb))
242 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
244 lkb->lkb_lksb->sb_status = rv;
245 lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
247 dlm_add_ast(lkb, AST_COMP);
250 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
252 if (is_master_copy(lkb))
253 send_bast(r, lkb, rqmode);
255 lkb->lkb_bastmode = rqmode;
256 dlm_add_ast(lkb, AST_BAST);
261 * Basic operations on rsb's and lkb's
264 static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
268 r = allocate_rsb(ls, len);
274 memcpy(r->res_name, name, len);
275 mutex_init(&r->res_mutex);
277 INIT_LIST_HEAD(&r->res_lookup);
278 INIT_LIST_HEAD(&r->res_grantqueue);
279 INIT_LIST_HEAD(&r->res_convertqueue);
280 INIT_LIST_HEAD(&r->res_waitqueue);
281 INIT_LIST_HEAD(&r->res_root_list);
282 INIT_LIST_HEAD(&r->res_recover_list);
287 static int search_rsb_list(struct list_head *head, char *name, int len,
288 unsigned int flags, struct dlm_rsb **r_ret)
293 list_for_each_entry(r, head, res_hashchain) {
294 if (len == r->res_length && !memcmp(name, r->res_name, len))
300 if (r->res_nodeid && (flags & R_MASTER))
306 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
307 unsigned int flags, struct dlm_rsb **r_ret)
312 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
314 kref_get(&r->res_ref);
317 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
321 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
323 if (dlm_no_directory(ls))
326 if (r->res_nodeid == -1) {
327 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
328 r->res_first_lkid = 0;
329 } else if (r->res_nodeid > 0) {
330 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
331 r->res_first_lkid = 0;
333 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
334 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
341 static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
342 unsigned int flags, struct dlm_rsb **r_ret)
345 write_lock(&ls->ls_rsbtbl[b].lock);
346 error = _search_rsb(ls, name, len, b, flags, r_ret);
347 write_unlock(&ls->ls_rsbtbl[b].lock);
352 * Find rsb in rsbtbl and potentially create/add one
354 * Delaying the release of rsb's has a similar benefit to applications keeping
355 * NL locks on an rsb, but without the guarantee that the cached master value
356 * will still be valid when the rsb is reused. Apps aren't always smart enough
357 * to keep NL locks on an rsb that they may lock again shortly; this can lead
358 * to excessive master lookups and removals if we don't delay the release.
360 * Searching for an rsb means looking through both the normal list and toss
361 * list. When found on the toss list the rsb is moved to the normal list with
362 * ref count of 1; when found on normal list the ref count is incremented.
365 static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
366 unsigned int flags, struct dlm_rsb **r_ret)
368 struct dlm_rsb *r, *tmp;
369 uint32_t hash, bucket;
372 if (dlm_no_directory(ls))
375 hash = jhash(name, namelen, 0);
376 bucket = hash & (ls->ls_rsbtbl_size - 1);
378 error = search_rsb(ls, name, namelen, bucket, flags, &r);
382 if (error == -EBADR && !(flags & R_CREATE))
385 /* the rsb was found but wasn't a master copy */
386 if (error == -ENOTBLK)
390 r = create_rsb(ls, name, namelen);
395 r->res_bucket = bucket;
397 kref_init(&r->res_ref);
399 /* With no directory, the master can be set immediately */
400 if (dlm_no_directory(ls)) {
401 int nodeid = dlm_dir_nodeid(r);
402 if (nodeid == dlm_our_nodeid())
404 r->res_nodeid = nodeid;
407 write_lock(&ls->ls_rsbtbl[bucket].lock);
408 error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
410 write_unlock(&ls->ls_rsbtbl[bucket].lock);
415 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
416 write_unlock(&ls->ls_rsbtbl[bucket].lock);
423 int dlm_find_rsb(struct dlm_ls *ls, char *name, int namelen,
424 unsigned int flags, struct dlm_rsb **r_ret)
426 return find_rsb(ls, name, namelen, flags, r_ret);
429 /* This is only called to add a reference when the code already holds
430 a valid reference to the rsb, so there's no need for locking. */
432 static inline void hold_rsb(struct dlm_rsb *r)
434 kref_get(&r->res_ref);
437 void dlm_hold_rsb(struct dlm_rsb *r)
442 static void toss_rsb(struct kref *kref)
444 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
445 struct dlm_ls *ls = r->res_ls;
447 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
448 kref_init(&r->res_ref);
449 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
450 r->res_toss_time = jiffies;
452 free_lvb(r->res_lvbptr);
453 r->res_lvbptr = NULL;
457 /* When all references to the rsb are gone it's transfered to
458 the tossed list for later disposal. */
460 static void put_rsb(struct dlm_rsb *r)
462 struct dlm_ls *ls = r->res_ls;
463 uint32_t bucket = r->res_bucket;
465 write_lock(&ls->ls_rsbtbl[bucket].lock);
466 kref_put(&r->res_ref, toss_rsb);
467 write_unlock(&ls->ls_rsbtbl[bucket].lock);
470 void dlm_put_rsb(struct dlm_rsb *r)
475 /* See comment for unhold_lkb */
477 static void unhold_rsb(struct dlm_rsb *r)
480 rv = kref_put(&r->res_ref, toss_rsb);
481 DLM_ASSERT(!rv, dlm_print_rsb(r););
484 static void kill_rsb(struct kref *kref)
486 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
488 /* All work is done after the return from kref_put() so we
489 can release the write_lock before the remove and free. */
491 DLM_ASSERT(list_empty(&r->res_lookup),);
492 DLM_ASSERT(list_empty(&r->res_grantqueue),);
493 DLM_ASSERT(list_empty(&r->res_convertqueue),);
494 DLM_ASSERT(list_empty(&r->res_waitqueue),);
495 DLM_ASSERT(list_empty(&r->res_root_list),);
496 DLM_ASSERT(list_empty(&r->res_recover_list),);
499 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
500 The rsb must exist as long as any lkb's for it do. */
502 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
505 lkb->lkb_resource = r;
508 static void detach_lkb(struct dlm_lkb *lkb)
510 if (lkb->lkb_resource) {
511 put_rsb(lkb->lkb_resource);
512 lkb->lkb_resource = NULL;
516 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
518 struct dlm_lkb *lkb, *tmp;
522 lkb = allocate_lkb(ls);
526 lkb->lkb_nodeid = -1;
527 lkb->lkb_grmode = DLM_LOCK_IV;
528 kref_init(&lkb->lkb_ref);
529 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
531 get_random_bytes(&bucket, sizeof(bucket));
532 bucket &= (ls->ls_lkbtbl_size - 1);
534 write_lock(&ls->ls_lkbtbl[bucket].lock);
536 /* counter can roll over so we must verify lkid is not in use */
539 lkid = bucket | (ls->ls_lkbtbl[bucket].counter++ << 16);
541 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
543 if (tmp->lkb_id != lkid)
551 list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
552 write_unlock(&ls->ls_lkbtbl[bucket].lock);
558 static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
560 uint16_t bucket = lkid & 0xFFFF;
563 list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
564 if (lkb->lkb_id == lkid)
570 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
573 uint16_t bucket = lkid & 0xFFFF;
575 if (bucket >= ls->ls_lkbtbl_size)
578 read_lock(&ls->ls_lkbtbl[bucket].lock);
579 lkb = __find_lkb(ls, lkid);
581 kref_get(&lkb->lkb_ref);
582 read_unlock(&ls->ls_lkbtbl[bucket].lock);
585 return lkb ? 0 : -ENOENT;
588 static void kill_lkb(struct kref *kref)
590 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
592 /* All work is done after the return from kref_put() so we
593 can release the write_lock before the detach_lkb */
595 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
598 /* __put_lkb() is used when an lkb may not have an rsb attached to
599 it so we need to provide the lockspace explicitly */
601 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
603 uint16_t bucket = lkb->lkb_id & 0xFFFF;
605 write_lock(&ls->ls_lkbtbl[bucket].lock);
606 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
607 list_del(&lkb->lkb_idtbl_list);
608 write_unlock(&ls->ls_lkbtbl[bucket].lock);
612 /* for local/process lkbs, lvbptr points to caller's lksb */
613 if (lkb->lkb_lvbptr && is_master_copy(lkb))
614 free_lvb(lkb->lkb_lvbptr);
618 write_unlock(&ls->ls_lkbtbl[bucket].lock);
623 int dlm_put_lkb(struct dlm_lkb *lkb)
627 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
628 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
630 ls = lkb->lkb_resource->res_ls;
631 return __put_lkb(ls, lkb);
634 /* This is only called to add a reference when the code already holds
635 a valid reference to the lkb, so there's no need for locking. */
637 static inline void hold_lkb(struct dlm_lkb *lkb)
639 kref_get(&lkb->lkb_ref);
642 /* This is called when we need to remove a reference and are certain
643 it's not the last ref. e.g. del_lkb is always called between a
644 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
645 put_lkb would work fine, but would involve unnecessary locking */
647 static inline void unhold_lkb(struct dlm_lkb *lkb)
650 rv = kref_put(&lkb->lkb_ref, kill_lkb);
651 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
654 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
657 struct dlm_lkb *lkb = NULL;
659 list_for_each_entry(lkb, head, lkb_statequeue)
660 if (lkb->lkb_rqmode < mode)
664 list_add_tail(new, head);
666 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
669 /* add/remove lkb to rsb's grant/convert/wait queue */
671 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
673 kref_get(&lkb->lkb_ref);
675 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
677 lkb->lkb_status = status;
680 case DLM_LKSTS_WAITING:
681 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
682 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
684 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
686 case DLM_LKSTS_GRANTED:
687 /* convention says granted locks kept in order of grmode */
688 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
691 case DLM_LKSTS_CONVERT:
692 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
693 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
695 list_add_tail(&lkb->lkb_statequeue,
696 &r->res_convertqueue);
699 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
703 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
706 list_del(&lkb->lkb_statequeue);
710 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
714 add_lkb(r, lkb, sts);
718 /* add/remove lkb from global waiters list of lkb's waiting for
719 a reply from a remote node */
721 static void add_to_waiters(struct dlm_lkb *lkb, int mstype)
723 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
725 mutex_lock(&ls->ls_waiters_mutex);
726 if (lkb->lkb_wait_type) {
727 log_print("add_to_waiters error %d", lkb->lkb_wait_type);
730 lkb->lkb_wait_type = mstype;
731 kref_get(&lkb->lkb_ref);
732 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
734 mutex_unlock(&ls->ls_waiters_mutex);
737 static int _remove_from_waiters(struct dlm_lkb *lkb)
741 if (!lkb->lkb_wait_type) {
742 log_print("remove_from_waiters error");
746 lkb->lkb_wait_type = 0;
747 list_del(&lkb->lkb_wait_reply);
753 static int remove_from_waiters(struct dlm_lkb *lkb)
755 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
758 mutex_lock(&ls->ls_waiters_mutex);
759 error = _remove_from_waiters(lkb);
760 mutex_unlock(&ls->ls_waiters_mutex);
764 static void dir_remove(struct dlm_rsb *r)
768 if (dlm_no_directory(r->res_ls))
771 to_nodeid = dlm_dir_nodeid(r);
772 if (to_nodeid != dlm_our_nodeid())
775 dlm_dir_remove_entry(r->res_ls, to_nodeid,
776 r->res_name, r->res_length);
779 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
780 found since they are in order of newest to oldest? */
782 static int shrink_bucket(struct dlm_ls *ls, int b)
785 int count = 0, found;
789 write_lock(&ls->ls_rsbtbl[b].lock);
790 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
792 if (!time_after_eq(jiffies, r->res_toss_time +
793 dlm_config.toss_secs * HZ))
800 write_unlock(&ls->ls_rsbtbl[b].lock);
804 if (kref_put(&r->res_ref, kill_rsb)) {
805 list_del(&r->res_hashchain);
806 write_unlock(&ls->ls_rsbtbl[b].lock);
813 write_unlock(&ls->ls_rsbtbl[b].lock);
814 log_error(ls, "tossed rsb in use %s", r->res_name);
821 void dlm_scan_rsbs(struct dlm_ls *ls)
825 if (dlm_locking_stopped(ls))
828 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
829 shrink_bucket(ls, i);
834 /* lkb is master or local copy */
836 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
838 int b, len = r->res_ls->ls_lvblen;
840 /* b=1 lvb returned to caller
841 b=0 lvb written to rsb or invalidated
844 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
847 if (!lkb->lkb_lvbptr)
850 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
856 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
857 lkb->lkb_lvbseq = r->res_lvbseq;
860 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
861 rsb_set_flag(r, RSB_VALNOTVALID);
865 if (!lkb->lkb_lvbptr)
868 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
872 r->res_lvbptr = allocate_lvb(r->res_ls);
877 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
879 lkb->lkb_lvbseq = r->res_lvbseq;
880 rsb_clear_flag(r, RSB_VALNOTVALID);
883 if (rsb_flag(r, RSB_VALNOTVALID))
884 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
887 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
889 if (lkb->lkb_grmode < DLM_LOCK_PW)
892 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
893 rsb_set_flag(r, RSB_VALNOTVALID);
897 if (!lkb->lkb_lvbptr)
900 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
904 r->res_lvbptr = allocate_lvb(r->res_ls);
909 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
911 rsb_clear_flag(r, RSB_VALNOTVALID);
914 /* lkb is process copy (pc) */
916 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
917 struct dlm_message *ms)
921 if (!lkb->lkb_lvbptr)
924 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
927 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
929 int len = receive_extralen(ms);
930 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
931 lkb->lkb_lvbseq = ms->m_lvbseq;
935 /* Manipulate lkb's on rsb's convert/granted/waiting queues
936 remove_lock -- used for unlock, removes lkb from granted
937 revert_lock -- used for cancel, moves lkb from convert to granted
938 grant_lock -- used for request and convert, adds lkb to granted or
939 moves lkb from convert or waiting to granted
941 Each of these is used for master or local copy lkb's. There is
942 also a _pc() variation used to make the corresponding change on
943 a process copy (pc) lkb. */
945 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
948 lkb->lkb_grmode = DLM_LOCK_IV;
949 /* this unhold undoes the original ref from create_lkb()
950 so this leads to the lkb being freed */
954 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
956 set_lvb_unlock(r, lkb);
957 _remove_lock(r, lkb);
960 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
962 _remove_lock(r, lkb);
965 static void revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
967 lkb->lkb_rqmode = DLM_LOCK_IV;
969 switch (lkb->lkb_status) {
970 case DLM_LKSTS_GRANTED:
972 case DLM_LKSTS_CONVERT:
973 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
975 case DLM_LKSTS_WAITING:
977 lkb->lkb_grmode = DLM_LOCK_IV;
978 /* this unhold undoes the original ref from create_lkb()
979 so this leads to the lkb being freed */
983 log_print("invalid status for revert %d", lkb->lkb_status);
987 static void revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
992 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
994 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
995 lkb->lkb_grmode = lkb->lkb_rqmode;
997 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
999 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1002 lkb->lkb_rqmode = DLM_LOCK_IV;
1005 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1007 set_lvb_lock(r, lkb);
1008 _grant_lock(r, lkb);
1009 lkb->lkb_highbast = 0;
1012 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1013 struct dlm_message *ms)
1015 set_lvb_lock_pc(r, lkb, ms);
1016 _grant_lock(r, lkb);
1019 /* called by grant_pending_locks() which means an async grant message must
1020 be sent to the requesting node in addition to granting the lock if the
1021 lkb belongs to a remote node. */
1023 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1026 if (is_master_copy(lkb))
1029 queue_cast(r, lkb, 0);
1032 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1034 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1036 if (lkb->lkb_id == first->lkb_id)
1042 /* Check if the given lkb conflicts with another lkb on the queue. */
1044 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1046 struct dlm_lkb *this;
1048 list_for_each_entry(this, head, lkb_statequeue) {
1051 if (!modes_compat(this, lkb))
1058 * "A conversion deadlock arises with a pair of lock requests in the converting
1059 * queue for one resource. The granted mode of each lock blocks the requested
1060 * mode of the other lock."
1062 * Part 2: if the granted mode of lkb is preventing the first lkb in the
1063 * convert queue from being granted, then demote lkb (set grmode to NL).
1064 * This second form requires that we check for conv-deadlk even when
1065 * now == 0 in _can_be_granted().
1068 * Granted Queue: empty
1069 * Convert Queue: NL->EX (first lock)
1070 * PR->EX (second lock)
1072 * The first lock can't be granted because of the granted mode of the second
1073 * lock and the second lock can't be granted because it's not first in the
1074 * list. We demote the granted mode of the second lock (the lkb passed to this
1077 * After the resolution, the "grant pending" function needs to go back and try
1078 * to grant locks on the convert queue again since the first lock can now be
1082 static int conversion_deadlock_detect(struct dlm_rsb *rsb, struct dlm_lkb *lkb)
1084 struct dlm_lkb *this, *first = NULL, *self = NULL;
1086 list_for_each_entry(this, &rsb->res_convertqueue, lkb_statequeue) {
1094 if (!modes_compat(this, lkb) && !modes_compat(lkb, this))
1098 /* if lkb is on the convert queue and is preventing the first
1099 from being granted, then there's deadlock and we demote lkb.
1100 multiple converting locks may need to do this before the first
1101 converting lock can be granted. */
1103 if (self && self != first) {
1104 if (!modes_compat(lkb, first) &&
1105 !queue_conflict(&rsb->res_grantqueue, first))
1113 * Return 1 if the lock can be granted, 0 otherwise.
1114 * Also detect and resolve conversion deadlocks.
1116 * lkb is the lock to be granted
1118 * now is 1 if the function is being called in the context of the
1119 * immediate request, it is 0 if called later, after the lock has been
1122 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1125 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1127 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1130 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1131 * a new request for a NL mode lock being blocked.
1133 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1134 * request, then it would be granted. In essence, the use of this flag
1135 * tells the Lock Manager to expedite theis request by not considering
1136 * what may be in the CONVERTING or WAITING queues... As of this
1137 * writing, the EXPEDITE flag can be used only with new requests for NL
1138 * mode locks. This flag is not valid for conversion requests.
1140 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1141 * conversion or used with a non-NL requested mode. We also know an
1142 * EXPEDITE request is always granted immediately, so now must always
1143 * be 1. The full condition to grant an expedite request: (now &&
1144 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1145 * therefore be shortened to just checking the flag.
1148 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1152 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1153 * added to the remaining conditions.
1156 if (queue_conflict(&r->res_grantqueue, lkb))
1160 * 6-3: By default, a conversion request is immediately granted if the
1161 * requested mode is compatible with the modes of all other granted
1165 if (queue_conflict(&r->res_convertqueue, lkb))
1169 * 6-5: But the default algorithm for deciding whether to grant or
1170 * queue conversion requests does not by itself guarantee that such
1171 * requests are serviced on a "first come first serve" basis. This, in
1172 * turn, can lead to a phenomenon known as "indefinate postponement".
1174 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1175 * the system service employed to request a lock conversion. This flag
1176 * forces certain conversion requests to be queued, even if they are
1177 * compatible with the granted modes of other locks on the same
1178 * resource. Thus, the use of this flag results in conversion requests
1179 * being ordered on a "first come first servce" basis.
1181 * DCT: This condition is all about new conversions being able to occur
1182 * "in place" while the lock remains on the granted queue (assuming
1183 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1184 * doesn't _have_ to go onto the convert queue where it's processed in
1185 * order. The "now" variable is necessary to distinguish converts
1186 * being received and processed for the first time now, because once a
1187 * convert is moved to the conversion queue the condition below applies
1188 * requiring fifo granting.
1191 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1195 * The NOORDER flag is set to avoid the standard vms rules on grant
1199 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1203 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1204 * granted until all other conversion requests ahead of it are granted
1208 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1212 * 6-4: By default, a new request is immediately granted only if all
1213 * three of the following conditions are satisfied when the request is
1215 * - The queue of ungranted conversion requests for the resource is
1217 * - The queue of ungranted new requests for the resource is empty.
1218 * - The mode of the new request is compatible with the most
1219 * restrictive mode of all granted locks on the resource.
1222 if (now && !conv && list_empty(&r->res_convertqueue) &&
1223 list_empty(&r->res_waitqueue))
1227 * 6-4: Once a lock request is in the queue of ungranted new requests,
1228 * it cannot be granted until the queue of ungranted conversion
1229 * requests is empty, all ungranted new requests ahead of it are
1230 * granted and/or canceled, and it is compatible with the granted mode
1231 * of the most restrictive lock granted on the resource.
1234 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1235 first_in_list(lkb, &r->res_waitqueue))
1240 * The following, enabled by CONVDEADLK, departs from VMS.
1243 if (conv && (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) &&
1244 conversion_deadlock_detect(r, lkb)) {
1245 lkb->lkb_grmode = DLM_LOCK_NL;
1246 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1253 * The ALTPR and ALTCW flags aren't traditional lock manager flags, but are a
1254 * simple way to provide a big optimization to applications that can use them.
1257 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1259 uint32_t flags = lkb->lkb_exflags;
1261 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1263 rv = _can_be_granted(r, lkb, now);
1267 if (lkb->lkb_sbflags & DLM_SBF_DEMOTED)
1270 if (rqmode != DLM_LOCK_PR && flags & DLM_LKF_ALTPR)
1272 else if (rqmode != DLM_LOCK_CW && flags & DLM_LKF_ALTCW)
1276 lkb->lkb_rqmode = alt;
1277 rv = _can_be_granted(r, lkb, now);
1279 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1281 lkb->lkb_rqmode = rqmode;
1287 static int grant_pending_convert(struct dlm_rsb *r, int high)
1289 struct dlm_lkb *lkb, *s;
1290 int hi, demoted, quit, grant_restart, demote_restart;
1298 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1299 demoted = is_demoted(lkb);
1300 if (can_be_granted(r, lkb, 0)) {
1301 grant_lock_pending(r, lkb);
1304 hi = max_t(int, lkb->lkb_rqmode, hi);
1305 if (!demoted && is_demoted(lkb))
1312 if (demote_restart && !quit) {
1317 return max_t(int, high, hi);
1320 static int grant_pending_wait(struct dlm_rsb *r, int high)
1322 struct dlm_lkb *lkb, *s;
1324 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1325 if (can_be_granted(r, lkb, 0))
1326 grant_lock_pending(r, lkb);
1328 high = max_t(int, lkb->lkb_rqmode, high);
1334 static void grant_pending_locks(struct dlm_rsb *r)
1336 struct dlm_lkb *lkb, *s;
1337 int high = DLM_LOCK_IV;
1339 DLM_ASSERT(is_master(r), dlm_print_rsb(r););
1341 high = grant_pending_convert(r, high);
1342 high = grant_pending_wait(r, high);
1344 if (high == DLM_LOCK_IV)
1348 * If there are locks left on the wait/convert queue then send blocking
1349 * ASTs to granted locks based on the largest requested mode (high)
1350 * found above. FIXME: highbast < high comparison not valid for PR/CW.
1353 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1354 if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
1355 !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
1356 queue_bast(r, lkb, high);
1357 lkb->lkb_highbast = high;
1362 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1363 struct dlm_lkb *lkb)
1367 list_for_each_entry(gr, head, lkb_statequeue) {
1368 if (gr->lkb_bastaddr &&
1369 gr->lkb_highbast < lkb->lkb_rqmode &&
1370 !modes_compat(gr, lkb)) {
1371 queue_bast(r, gr, lkb->lkb_rqmode);
1372 gr->lkb_highbast = lkb->lkb_rqmode;
1377 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1379 send_bast_queue(r, &r->res_grantqueue, lkb);
1382 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1384 send_bast_queue(r, &r->res_grantqueue, lkb);
1385 send_bast_queue(r, &r->res_convertqueue, lkb);
1388 /* set_master(r, lkb) -- set the master nodeid of a resource
1390 The purpose of this function is to set the nodeid field in the given
1391 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1392 known, it can just be copied to the lkb and the function will return
1393 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1394 before it can be copied to the lkb.
1396 When the rsb nodeid is being looked up remotely, the initial lkb
1397 causing the lookup is kept on the ls_waiters list waiting for the
1398 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1399 on the rsb's res_lookup list until the master is verified.
1402 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1403 1: the rsb master is not available and the lkb has been placed on
1407 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1409 struct dlm_ls *ls = r->res_ls;
1410 int error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
1412 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1413 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1414 r->res_first_lkid = lkb->lkb_id;
1415 lkb->lkb_nodeid = r->res_nodeid;
1419 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1420 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1424 if (r->res_nodeid == 0) {
1425 lkb->lkb_nodeid = 0;
1429 if (r->res_nodeid > 0) {
1430 lkb->lkb_nodeid = r->res_nodeid;
1434 DLM_ASSERT(r->res_nodeid == -1, dlm_print_rsb(r););
1436 dir_nodeid = dlm_dir_nodeid(r);
1438 if (dir_nodeid != our_nodeid) {
1439 r->res_first_lkid = lkb->lkb_id;
1440 send_lookup(r, lkb);
1445 /* It's possible for dlm_scand to remove an old rsb for
1446 this same resource from the toss list, us to create
1447 a new one, look up the master locally, and find it
1448 already exists just before dlm_scand does the
1449 dir_remove() on the previous rsb. */
1451 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1452 r->res_length, &ret_nodeid);
1455 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1459 if (ret_nodeid == our_nodeid) {
1460 r->res_first_lkid = 0;
1462 lkb->lkb_nodeid = 0;
1464 r->res_first_lkid = lkb->lkb_id;
1465 r->res_nodeid = ret_nodeid;
1466 lkb->lkb_nodeid = ret_nodeid;
1471 static void process_lookup_list(struct dlm_rsb *r)
1473 struct dlm_lkb *lkb, *safe;
1475 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
1476 list_del(&lkb->lkb_rsb_lookup);
1477 _request_lock(r, lkb);
1482 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1484 static void confirm_master(struct dlm_rsb *r, int error)
1486 struct dlm_lkb *lkb;
1488 if (!r->res_first_lkid)
1494 r->res_first_lkid = 0;
1495 process_lookup_list(r);
1499 /* the remote master didn't queue our NOQUEUE request;
1500 make a waiting lkb the first_lkid */
1502 r->res_first_lkid = 0;
1504 if (!list_empty(&r->res_lookup)) {
1505 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1507 list_del(&lkb->lkb_rsb_lookup);
1508 r->res_first_lkid = lkb->lkb_id;
1509 _request_lock(r, lkb);
1515 log_error(r->res_ls, "confirm_master unknown error %d", error);
1519 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
1520 int namelen, uint32_t parent_lkid, void *ast,
1521 void *astarg, void *bast, struct dlm_args *args)
1525 /* check for invalid arg usage */
1527 if (mode < 0 || mode > DLM_LOCK_EX)
1530 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1533 if (flags & DLM_LKF_CANCEL)
1536 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1539 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1542 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1545 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1548 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
1551 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
1554 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
1560 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
1563 /* parent/child locks not yet supported */
1567 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
1570 /* these args will be copied to the lkb in validate_lock_args,
1571 it cannot be done now because when converting locks, fields in
1572 an active lkb cannot be modified before locking the rsb */
1574 args->flags = flags;
1575 args->astaddr = ast;
1576 args->astparam = (long) astarg;
1577 args->bastaddr = bast;
1585 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
1587 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
1588 DLM_LKF_FORCEUNLOCK))
1591 args->flags = flags;
1592 args->astparam = (long) astarg;
1596 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
1597 struct dlm_args *args)
1601 if (args->flags & DLM_LKF_CONVERT) {
1602 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
1605 if (args->flags & DLM_LKF_QUECVT &&
1606 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
1610 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
1613 if (lkb->lkb_wait_type)
1617 lkb->lkb_exflags = args->flags;
1618 lkb->lkb_sbflags = 0;
1619 lkb->lkb_astaddr = args->astaddr;
1620 lkb->lkb_astparam = args->astparam;
1621 lkb->lkb_bastaddr = args->bastaddr;
1622 lkb->lkb_rqmode = args->mode;
1623 lkb->lkb_lksb = args->lksb;
1624 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
1625 lkb->lkb_ownpid = (int) current->pid;
1631 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
1635 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
1638 if (args->flags & DLM_LKF_FORCEUNLOCK)
1641 if (args->flags & DLM_LKF_CANCEL &&
1642 lkb->lkb_status == DLM_LKSTS_GRANTED)
1645 if (!(args->flags & DLM_LKF_CANCEL) &&
1646 lkb->lkb_status != DLM_LKSTS_GRANTED)
1650 if (lkb->lkb_wait_type)
1654 lkb->lkb_exflags = args->flags;
1655 lkb->lkb_sbflags = 0;
1656 lkb->lkb_astparam = args->astparam;
1664 * Four stage 4 varieties:
1665 * do_request(), do_convert(), do_unlock(), do_cancel()
1666 * These are called on the master node for the given lock and
1667 * from the central locking logic.
1670 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
1674 if (can_be_granted(r, lkb, 1)) {
1676 queue_cast(r, lkb, 0);
1680 if (can_be_queued(lkb)) {
1681 error = -EINPROGRESS;
1682 add_lkb(r, lkb, DLM_LKSTS_WAITING);
1683 send_blocking_asts(r, lkb);
1688 if (force_blocking_asts(lkb))
1689 send_blocking_asts_all(r, lkb);
1690 queue_cast(r, lkb, -EAGAIN);
1696 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
1700 /* changing an existing lock may allow others to be granted */
1702 if (can_be_granted(r, lkb, 1)) {
1704 queue_cast(r, lkb, 0);
1705 grant_pending_locks(r);
1709 if (can_be_queued(lkb)) {
1710 if (is_demoted(lkb))
1711 grant_pending_locks(r);
1712 error = -EINPROGRESS;
1714 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
1715 send_blocking_asts(r, lkb);
1720 if (force_blocking_asts(lkb))
1721 send_blocking_asts_all(r, lkb);
1722 queue_cast(r, lkb, -EAGAIN);
1728 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1730 remove_lock(r, lkb);
1731 queue_cast(r, lkb, -DLM_EUNLOCK);
1732 grant_pending_locks(r);
1733 return -DLM_EUNLOCK;
1736 /* FIXME: if revert_lock() finds that the lkb is granted, we should
1737 skip the queue_cast(ECANCEL). It indicates that the request/convert
1738 completed (and queued a normal ast) just before the cancel; we don't
1739 want to clobber the sb_result for the normal ast with ECANCEL. */
1741 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
1743 revert_lock(r, lkb);
1744 queue_cast(r, lkb, -DLM_ECANCEL);
1745 grant_pending_locks(r);
1746 return -DLM_ECANCEL;
1750 * Four stage 3 varieties:
1751 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
1754 /* add a new lkb to a possibly new rsb, called by requesting process */
1756 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1760 /* set_master: sets lkb nodeid from r */
1762 error = set_master(r, lkb);
1771 /* receive_request() calls do_request() on remote node */
1772 error = send_request(r, lkb);
1774 error = do_request(r, lkb);
1779 /* change some property of an existing lkb, e.g. mode */
1781 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1786 /* receive_convert() calls do_convert() on remote node */
1787 error = send_convert(r, lkb);
1789 error = do_convert(r, lkb);
1794 /* remove an existing lkb from the granted queue */
1796 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1801 /* receive_unlock() calls do_unlock() on remote node */
1802 error = send_unlock(r, lkb);
1804 error = do_unlock(r, lkb);
1809 /* remove an existing lkb from the convert or wait queue */
1811 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1816 /* receive_cancel() calls do_cancel() on remote node */
1817 error = send_cancel(r, lkb);
1819 error = do_cancel(r, lkb);
1825 * Four stage 2 varieties:
1826 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
1829 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
1830 int len, struct dlm_args *args)
1835 error = validate_lock_args(ls, lkb, args);
1839 error = find_rsb(ls, name, len, R_CREATE, &r);
1846 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
1848 error = _request_lock(r, lkb);
1857 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1858 struct dlm_args *args)
1863 r = lkb->lkb_resource;
1868 error = validate_lock_args(ls, lkb, args);
1872 error = _convert_lock(r, lkb);
1879 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1880 struct dlm_args *args)
1885 r = lkb->lkb_resource;
1890 error = validate_unlock_args(lkb, args);
1894 error = _unlock_lock(r, lkb);
1901 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1902 struct dlm_args *args)
1907 r = lkb->lkb_resource;
1912 error = validate_unlock_args(lkb, args);
1916 error = _cancel_lock(r, lkb);
1924 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
1927 int dlm_lock(dlm_lockspace_t *lockspace,
1929 struct dlm_lksb *lksb,
1932 unsigned int namelen,
1933 uint32_t parent_lkid,
1934 void (*ast) (void *astarg),
1936 void (*bast) (void *astarg, int mode))
1939 struct dlm_lkb *lkb;
1940 struct dlm_args args;
1941 int error, convert = flags & DLM_LKF_CONVERT;
1943 ls = dlm_find_lockspace_local(lockspace);
1950 error = find_lkb(ls, lksb->sb_lkid, &lkb);
1952 error = create_lkb(ls, &lkb);
1957 error = set_lock_args(mode, lksb, flags, namelen, parent_lkid, ast,
1958 astarg, bast, &args);
1963 error = convert_lock(ls, lkb, &args);
1965 error = request_lock(ls, lkb, name, namelen, &args);
1967 if (error == -EINPROGRESS)
1970 if (convert || error)
1972 if (error == -EAGAIN)
1975 unlock_recovery(ls);
1976 dlm_put_lockspace(ls);
1980 int dlm_unlock(dlm_lockspace_t *lockspace,
1983 struct dlm_lksb *lksb,
1987 struct dlm_lkb *lkb;
1988 struct dlm_args args;
1991 ls = dlm_find_lockspace_local(lockspace);
1997 error = find_lkb(ls, lkid, &lkb);
2001 error = set_unlock_args(flags, astarg, &args);
2005 if (flags & DLM_LKF_CANCEL)
2006 error = cancel_lock(ls, lkb, &args);
2008 error = unlock_lock(ls, lkb, &args);
2010 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2015 unlock_recovery(ls);
2016 dlm_put_lockspace(ls);
2021 * send/receive routines for remote operations and replies
2025 * send_request receive_request
2026 * send_convert receive_convert
2027 * send_unlock receive_unlock
2028 * send_cancel receive_cancel
2029 * send_grant receive_grant
2030 * send_bast receive_bast
2031 * send_lookup receive_lookup
2032 * send_remove receive_remove
2035 * receive_request_reply send_request_reply
2036 * receive_convert_reply send_convert_reply
2037 * receive_unlock_reply send_unlock_reply
2038 * receive_cancel_reply send_cancel_reply
2039 * receive_lookup_reply send_lookup_reply
2042 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2043 int to_nodeid, int mstype,
2044 struct dlm_message **ms_ret,
2045 struct dlm_mhandle **mh_ret)
2047 struct dlm_message *ms;
2048 struct dlm_mhandle *mh;
2050 int mb_len = sizeof(struct dlm_message);
2053 case DLM_MSG_REQUEST:
2054 case DLM_MSG_LOOKUP:
2055 case DLM_MSG_REMOVE:
2056 mb_len += r->res_length;
2058 case DLM_MSG_CONVERT:
2059 case DLM_MSG_UNLOCK:
2060 case DLM_MSG_REQUEST_REPLY:
2061 case DLM_MSG_CONVERT_REPLY:
2063 if (lkb && lkb->lkb_lvbptr)
2064 mb_len += r->res_ls->ls_lvblen;
2068 /* get_buffer gives us a message handle (mh) that we need to
2069 pass into lowcomms_commit and a message buffer (mb) that we
2070 write our data into */
2072 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_KERNEL, &mb);
2076 memset(mb, 0, mb_len);
2078 ms = (struct dlm_message *) mb;
2080 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2081 ms->m_header.h_lockspace = r->res_ls->ls_global_id;
2082 ms->m_header.h_nodeid = dlm_our_nodeid();
2083 ms->m_header.h_length = mb_len;
2084 ms->m_header.h_cmd = DLM_MSG;
2086 ms->m_type = mstype;
2093 /* further lowcomms enhancements or alternate implementations may make
2094 the return value from this function useful at some point */
2096 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2098 dlm_message_out(ms);
2099 dlm_lowcomms_commit_buffer(mh);
2103 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2104 struct dlm_message *ms)
2106 ms->m_nodeid = lkb->lkb_nodeid;
2107 ms->m_pid = lkb->lkb_ownpid;
2108 ms->m_lkid = lkb->lkb_id;
2109 ms->m_remid = lkb->lkb_remid;
2110 ms->m_exflags = lkb->lkb_exflags;
2111 ms->m_sbflags = lkb->lkb_sbflags;
2112 ms->m_flags = lkb->lkb_flags;
2113 ms->m_lvbseq = lkb->lkb_lvbseq;
2114 ms->m_status = lkb->lkb_status;
2115 ms->m_grmode = lkb->lkb_grmode;
2116 ms->m_rqmode = lkb->lkb_rqmode;
2117 ms->m_hash = r->res_hash;
2119 /* m_result and m_bastmode are set from function args,
2120 not from lkb fields */
2122 if (lkb->lkb_bastaddr)
2123 ms->m_asts |= AST_BAST;
2124 if (lkb->lkb_astaddr)
2125 ms->m_asts |= AST_COMP;
2127 if (ms->m_type == DLM_MSG_REQUEST || ms->m_type == DLM_MSG_LOOKUP)
2128 memcpy(ms->m_extra, r->res_name, r->res_length);
2130 else if (lkb->lkb_lvbptr)
2131 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2135 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2137 struct dlm_message *ms;
2138 struct dlm_mhandle *mh;
2139 int to_nodeid, error;
2141 add_to_waiters(lkb, mstype);
2143 to_nodeid = r->res_nodeid;
2145 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2149 send_args(r, lkb, ms);
2151 error = send_message(mh, ms);
2157 remove_from_waiters(lkb);
2161 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2163 return send_common(r, lkb, DLM_MSG_REQUEST);
2166 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2170 error = send_common(r, lkb, DLM_MSG_CONVERT);
2172 /* down conversions go without a reply from the master */
2173 if (!error && down_conversion(lkb)) {
2174 remove_from_waiters(lkb);
2175 r->res_ls->ls_stub_ms.m_result = 0;
2176 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2182 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2183 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2184 that the master is still correct. */
2186 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2188 return send_common(r, lkb, DLM_MSG_UNLOCK);
2191 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2193 return send_common(r, lkb, DLM_MSG_CANCEL);
2196 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2198 struct dlm_message *ms;
2199 struct dlm_mhandle *mh;
2200 int to_nodeid, error;
2202 to_nodeid = lkb->lkb_nodeid;
2204 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2208 send_args(r, lkb, ms);
2212 error = send_message(mh, ms);
2217 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2219 struct dlm_message *ms;
2220 struct dlm_mhandle *mh;
2221 int to_nodeid, error;
2223 to_nodeid = lkb->lkb_nodeid;
2225 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2229 send_args(r, lkb, ms);
2231 ms->m_bastmode = mode;
2233 error = send_message(mh, ms);
2238 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2240 struct dlm_message *ms;
2241 struct dlm_mhandle *mh;
2242 int to_nodeid, error;
2244 add_to_waiters(lkb, DLM_MSG_LOOKUP);
2246 to_nodeid = dlm_dir_nodeid(r);
2248 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2252 send_args(r, lkb, ms);
2254 error = send_message(mh, ms);
2260 remove_from_waiters(lkb);
2264 static int send_remove(struct dlm_rsb *r)
2266 struct dlm_message *ms;
2267 struct dlm_mhandle *mh;
2268 int to_nodeid, error;
2270 to_nodeid = dlm_dir_nodeid(r);
2272 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2276 memcpy(ms->m_extra, r->res_name, r->res_length);
2277 ms->m_hash = r->res_hash;
2279 error = send_message(mh, ms);
2284 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2287 struct dlm_message *ms;
2288 struct dlm_mhandle *mh;
2289 int to_nodeid, error;
2291 to_nodeid = lkb->lkb_nodeid;
2293 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2297 send_args(r, lkb, ms);
2301 error = send_message(mh, ms);
2306 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2308 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2311 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2313 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2316 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2318 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2321 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2323 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2326 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2327 int ret_nodeid, int rv)
2329 struct dlm_rsb *r = &ls->ls_stub_rsb;
2330 struct dlm_message *ms;
2331 struct dlm_mhandle *mh;
2332 int error, nodeid = ms_in->m_header.h_nodeid;
2334 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2338 ms->m_lkid = ms_in->m_lkid;
2340 ms->m_nodeid = ret_nodeid;
2342 error = send_message(mh, ms);
2347 /* which args we save from a received message depends heavily on the type
2348 of message, unlike the send side where we can safely send everything about
2349 the lkb for any type of message */
2351 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2353 lkb->lkb_exflags = ms->m_exflags;
2354 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2355 (ms->m_flags & 0x0000FFFF);
2358 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2360 lkb->lkb_sbflags = ms->m_sbflags;
2361 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2362 (ms->m_flags & 0x0000FFFF);
2365 static int receive_extralen(struct dlm_message *ms)
2367 return (ms->m_header.h_length - sizeof(struct dlm_message));
2370 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2371 struct dlm_message *ms)
2375 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2376 if (!lkb->lkb_lvbptr)
2377 lkb->lkb_lvbptr = allocate_lvb(ls);
2378 if (!lkb->lkb_lvbptr)
2380 len = receive_extralen(ms);
2381 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2386 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2387 struct dlm_message *ms)
2389 lkb->lkb_nodeid = ms->m_header.h_nodeid;
2390 lkb->lkb_ownpid = ms->m_pid;
2391 lkb->lkb_remid = ms->m_lkid;
2392 lkb->lkb_grmode = DLM_LOCK_IV;
2393 lkb->lkb_rqmode = ms->m_rqmode;
2394 lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
2395 lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
2397 DLM_ASSERT(is_master_copy(lkb), dlm_print_lkb(lkb););
2399 if (receive_lvb(ls, lkb, ms))
2405 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2406 struct dlm_message *ms)
2408 if (lkb->lkb_nodeid != ms->m_header.h_nodeid) {
2409 log_error(ls, "convert_args nodeid %d %d lkid %x %x",
2410 lkb->lkb_nodeid, ms->m_header.h_nodeid,
2411 lkb->lkb_id, lkb->lkb_remid);
2415 if (!is_master_copy(lkb))
2418 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2421 if (receive_lvb(ls, lkb, ms))
2424 lkb->lkb_rqmode = ms->m_rqmode;
2425 lkb->lkb_lvbseq = ms->m_lvbseq;
2430 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2431 struct dlm_message *ms)
2433 if (!is_master_copy(lkb))
2435 if (receive_lvb(ls, lkb, ms))
2440 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
2441 uses to send a reply and that the remote end uses to process the reply. */
2443 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
2445 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
2446 lkb->lkb_nodeid = ms->m_header.h_nodeid;
2447 lkb->lkb_remid = ms->m_lkid;
2450 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
2452 struct dlm_lkb *lkb;
2456 error = create_lkb(ls, &lkb);
2460 receive_flags(lkb, ms);
2461 lkb->lkb_flags |= DLM_IFL_MSTCPY;
2462 error = receive_request_args(ls, lkb, ms);
2468 namelen = receive_extralen(ms);
2470 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
2479 error = do_request(r, lkb);
2480 send_request_reply(r, lkb, error);
2485 if (error == -EINPROGRESS)
2492 setup_stub_lkb(ls, ms);
2493 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2496 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
2498 struct dlm_lkb *lkb;
2500 int error, reply = 1;
2502 error = find_lkb(ls, ms->m_remid, &lkb);
2506 r = lkb->lkb_resource;
2511 receive_flags(lkb, ms);
2512 error = receive_convert_args(ls, lkb, ms);
2515 reply = !down_conversion(lkb);
2517 error = do_convert(r, lkb);
2520 send_convert_reply(r, lkb, error);
2528 setup_stub_lkb(ls, ms);
2529 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2532 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
2534 struct dlm_lkb *lkb;
2538 error = find_lkb(ls, ms->m_remid, &lkb);
2542 r = lkb->lkb_resource;
2547 receive_flags(lkb, ms);
2548 error = receive_unlock_args(ls, lkb, ms);
2552 error = do_unlock(r, lkb);
2554 send_unlock_reply(r, lkb, error);
2562 setup_stub_lkb(ls, ms);
2563 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2566 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
2568 struct dlm_lkb *lkb;
2572 error = find_lkb(ls, ms->m_remid, &lkb);
2576 receive_flags(lkb, ms);
2578 r = lkb->lkb_resource;
2583 error = do_cancel(r, lkb);
2584 send_cancel_reply(r, lkb, error);
2592 setup_stub_lkb(ls, ms);
2593 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2596 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
2598 struct dlm_lkb *lkb;
2602 error = find_lkb(ls, ms->m_remid, &lkb);
2604 log_error(ls, "receive_grant no lkb");
2607 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2609 r = lkb->lkb_resource;
2614 receive_flags_reply(lkb, ms);
2615 grant_lock_pc(r, lkb, ms);
2616 queue_cast(r, lkb, 0);
2623 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
2625 struct dlm_lkb *lkb;
2629 error = find_lkb(ls, ms->m_remid, &lkb);
2631 log_error(ls, "receive_bast no lkb");
2634 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2636 r = lkb->lkb_resource;
2641 queue_bast(r, lkb, ms->m_bastmode);
2648 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
2650 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
2652 from_nodeid = ms->m_header.h_nodeid;
2653 our_nodeid = dlm_our_nodeid();
2655 len = receive_extralen(ms);
2657 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
2658 if (dir_nodeid != our_nodeid) {
2659 log_error(ls, "lookup dir_nodeid %d from %d",
2660 dir_nodeid, from_nodeid);
2666 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
2668 /* Optimization: we're master so treat lookup as a request */
2669 if (!error && ret_nodeid == our_nodeid) {
2670 receive_request(ls, ms);
2674 send_lookup_reply(ls, ms, ret_nodeid, error);
2677 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
2679 int len, dir_nodeid, from_nodeid;
2681 from_nodeid = ms->m_header.h_nodeid;
2683 len = receive_extralen(ms);
2685 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
2686 if (dir_nodeid != dlm_our_nodeid()) {
2687 log_error(ls, "remove dir entry dir_nodeid %d from %d",
2688 dir_nodeid, from_nodeid);
2692 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
2695 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
2697 struct dlm_lkb *lkb;
2701 error = find_lkb(ls, ms->m_remid, &lkb);
2703 log_error(ls, "receive_request_reply no lkb");
2706 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2708 mstype = lkb->lkb_wait_type;
2709 error = remove_from_waiters(lkb);
2711 log_error(ls, "receive_request_reply not on waiters");
2715 /* this is the value returned from do_request() on the master */
2716 error = ms->m_result;
2718 r = lkb->lkb_resource;
2722 /* Optimization: the dir node was also the master, so it took our
2723 lookup as a request and sent request reply instead of lookup reply */
2724 if (mstype == DLM_MSG_LOOKUP) {
2725 r->res_nodeid = ms->m_header.h_nodeid;
2726 lkb->lkb_nodeid = r->res_nodeid;
2731 /* request would block (be queued) on remote master;
2732 the unhold undoes the original ref from create_lkb()
2733 so it leads to the lkb being freed */
2734 queue_cast(r, lkb, -EAGAIN);
2735 confirm_master(r, -EAGAIN);
2741 /* request was queued or granted on remote master */
2742 receive_flags_reply(lkb, ms);
2743 lkb->lkb_remid = ms->m_lkid;
2745 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2747 grant_lock_pc(r, lkb, ms);
2748 queue_cast(r, lkb, 0);
2750 confirm_master(r, error);
2755 /* find_rsb failed to find rsb or rsb wasn't master */
2757 lkb->lkb_nodeid = -1;
2758 _request_lock(r, lkb);
2762 log_error(ls, "receive_request_reply error %d", error);
2771 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2772 struct dlm_message *ms)
2774 int error = ms->m_result;
2776 /* this is the value returned from do_convert() on the master */
2780 /* convert would block (be queued) on remote master */
2781 queue_cast(r, lkb, -EAGAIN);
2785 /* convert was queued on remote master */
2787 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2791 /* convert was granted on remote master */
2792 receive_flags_reply(lkb, ms);
2793 grant_lock_pc(r, lkb, ms);
2794 queue_cast(r, lkb, 0);
2798 log_error(r->res_ls, "receive_convert_reply error %d", error);
2802 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2804 struct dlm_rsb *r = lkb->lkb_resource;
2809 __receive_convert_reply(r, lkb, ms);
2815 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
2817 struct dlm_lkb *lkb;
2820 error = find_lkb(ls, ms->m_remid, &lkb);
2822 log_error(ls, "receive_convert_reply no lkb");
2825 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2827 error = remove_from_waiters(lkb);
2829 log_error(ls, "receive_convert_reply not on waiters");
2833 _receive_convert_reply(lkb, ms);
2838 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2840 struct dlm_rsb *r = lkb->lkb_resource;
2841 int error = ms->m_result;
2846 /* this is the value returned from do_unlock() on the master */
2850 receive_flags_reply(lkb, ms);
2851 remove_lock_pc(r, lkb);
2852 queue_cast(r, lkb, -DLM_EUNLOCK);
2855 log_error(r->res_ls, "receive_unlock_reply error %d", error);
2862 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
2864 struct dlm_lkb *lkb;
2867 error = find_lkb(ls, ms->m_remid, &lkb);
2869 log_error(ls, "receive_unlock_reply no lkb");
2872 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2874 error = remove_from_waiters(lkb);
2876 log_error(ls, "receive_unlock_reply not on waiters");
2880 _receive_unlock_reply(lkb, ms);
2885 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2887 struct dlm_rsb *r = lkb->lkb_resource;
2888 int error = ms->m_result;
2893 /* this is the value returned from do_cancel() on the master */
2897 receive_flags_reply(lkb, ms);
2898 revert_lock_pc(r, lkb);
2899 queue_cast(r, lkb, -DLM_ECANCEL);
2902 log_error(r->res_ls, "receive_cancel_reply error %d", error);
2909 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
2911 struct dlm_lkb *lkb;
2914 error = find_lkb(ls, ms->m_remid, &lkb);
2916 log_error(ls, "receive_cancel_reply no lkb");
2919 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2921 error = remove_from_waiters(lkb);
2923 log_error(ls, "receive_cancel_reply not on waiters");
2927 _receive_cancel_reply(lkb, ms);
2932 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
2934 struct dlm_lkb *lkb;
2936 int error, ret_nodeid;
2938 error = find_lkb(ls, ms->m_lkid, &lkb);
2940 log_error(ls, "receive_lookup_reply no lkb");
2944 error = remove_from_waiters(lkb);
2946 log_error(ls, "receive_lookup_reply not on waiters");
2950 /* this is the value returned by dlm_dir_lookup on dir node
2951 FIXME: will a non-zero error ever be returned? */
2952 error = ms->m_result;
2954 r = lkb->lkb_resource;
2958 ret_nodeid = ms->m_nodeid;
2959 if (ret_nodeid == dlm_our_nodeid()) {
2962 r->res_first_lkid = 0;
2964 /* set_master() will copy res_nodeid to lkb_nodeid */
2965 r->res_nodeid = ret_nodeid;
2968 _request_lock(r, lkb);
2971 process_lookup_list(r);
2979 int dlm_receive_message(struct dlm_header *hd, int nodeid, int recovery)
2981 struct dlm_message *ms = (struct dlm_message *) hd;
2988 ls = dlm_find_lockspace_global(hd->h_lockspace);
2990 log_print("drop message %d from %d for unknown lockspace %d",
2991 ms->m_type, nodeid, hd->h_lockspace);
2995 /* recovery may have just ended leaving a bunch of backed-up requests
2996 in the requestqueue; wait while dlm_recoverd clears them */
2999 dlm_wait_requestqueue(ls);
3001 /* recovery may have just started while there were a bunch of
3002 in-flight requests -- save them in requestqueue to be processed
3003 after recovery. we can't let dlm_recvd block on the recovery
3004 lock. if dlm_recoverd is calling this function to clear the
3005 requestqueue, it needs to be interrupted (-EINTR) if another
3006 recovery operation is starting. */
3009 if (dlm_locking_stopped(ls)) {
3011 dlm_add_requestqueue(ls, nodeid, hd);
3016 if (lock_recovery_try(ls))
3021 switch (ms->m_type) {
3023 /* messages sent to a master node */
3025 case DLM_MSG_REQUEST:
3026 receive_request(ls, ms);
3029 case DLM_MSG_CONVERT:
3030 receive_convert(ls, ms);
3033 case DLM_MSG_UNLOCK:
3034 receive_unlock(ls, ms);
3037 case DLM_MSG_CANCEL:
3038 receive_cancel(ls, ms);
3041 /* messages sent from a master node (replies to above) */
3043 case DLM_MSG_REQUEST_REPLY:
3044 receive_request_reply(ls, ms);
3047 case DLM_MSG_CONVERT_REPLY:
3048 receive_convert_reply(ls, ms);
3051 case DLM_MSG_UNLOCK_REPLY:
3052 receive_unlock_reply(ls, ms);
3055 case DLM_MSG_CANCEL_REPLY:
3056 receive_cancel_reply(ls, ms);
3059 /* messages sent from a master node (only two types of async msg) */
3062 receive_grant(ls, ms);
3066 receive_bast(ls, ms);
3069 /* messages sent to a dir node */
3071 case DLM_MSG_LOOKUP:
3072 receive_lookup(ls, ms);
3075 case DLM_MSG_REMOVE:
3076 receive_remove(ls, ms);
3079 /* messages sent from a dir node (remove has no reply) */
3081 case DLM_MSG_LOOKUP_REPLY:
3082 receive_lookup_reply(ls, ms);
3086 log_error(ls, "unknown message type %d", ms->m_type);
3089 unlock_recovery(ls);
3091 dlm_put_lockspace(ls);
3101 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3103 if (middle_conversion(lkb)) {
3105 ls->ls_stub_ms.m_result = -EINPROGRESS;
3106 _remove_from_waiters(lkb);
3107 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3109 /* Same special case as in receive_rcom_lock_args() */
3110 lkb->lkb_grmode = DLM_LOCK_IV;
3111 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3114 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3115 lkb->lkb_flags |= DLM_IFL_RESEND;
3118 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3119 conversions are async; there's no reply from the remote master */
3122 /* A waiting lkb needs recovery if the master node has failed, or
3123 the master node is changing (only when no directory is used) */
3125 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3127 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3130 if (!dlm_no_directory(ls))
3133 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3139 /* Recovery for locks that are waiting for replies from nodes that are now
3140 gone. We can just complete unlocks and cancels by faking a reply from the
3141 dead node. Requests and up-conversions we flag to be resent after
3142 recovery. Down-conversions can just be completed with a fake reply like
3143 unlocks. Conversions between PR and CW need special attention. */
3145 void dlm_recover_waiters_pre(struct dlm_ls *ls)
3147 struct dlm_lkb *lkb, *safe;
3149 mutex_lock(&ls->ls_waiters_mutex);
3151 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3152 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3153 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3155 /* all outstanding lookups, regardless of destination will be
3156 resent after recovery is done */
3158 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3159 lkb->lkb_flags |= DLM_IFL_RESEND;
3163 if (!waiter_needs_recovery(ls, lkb))
3166 switch (lkb->lkb_wait_type) {
3168 case DLM_MSG_REQUEST:
3169 lkb->lkb_flags |= DLM_IFL_RESEND;
3172 case DLM_MSG_CONVERT:
3173 recover_convert_waiter(ls, lkb);
3176 case DLM_MSG_UNLOCK:
3178 ls->ls_stub_ms.m_result = -DLM_EUNLOCK;
3179 _remove_from_waiters(lkb);
3180 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
3184 case DLM_MSG_CANCEL:
3186 ls->ls_stub_ms.m_result = -DLM_ECANCEL;
3187 _remove_from_waiters(lkb);
3188 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
3193 log_error(ls, "invalid lkb wait_type %d",
3194 lkb->lkb_wait_type);
3198 mutex_unlock(&ls->ls_waiters_mutex);
3201 static int remove_resend_waiter(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
3203 struct dlm_lkb *lkb;
3206 mutex_lock(&ls->ls_waiters_mutex);
3207 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
3208 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3209 rv = lkb->lkb_wait_type;
3210 _remove_from_waiters(lkb);
3211 lkb->lkb_flags &= ~DLM_IFL_RESEND;
3215 mutex_unlock(&ls->ls_waiters_mutex);
3223 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
3224 master or dir-node for r. Processing the lkb may result in it being placed
3227 int dlm_recover_waiters_post(struct dlm_ls *ls)
3229 struct dlm_lkb *lkb;
3231 int error = 0, mstype;
3234 if (dlm_locking_stopped(ls)) {
3235 log_debug(ls, "recover_waiters_post aborted");
3240 mstype = remove_resend_waiter(ls, &lkb);
3244 r = lkb->lkb_resource;
3246 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
3247 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
3251 case DLM_MSG_LOOKUP:
3254 _request_lock(r, lkb);
3256 confirm_master(r, 0);
3261 case DLM_MSG_REQUEST:
3264 _request_lock(r, lkb);
3269 case DLM_MSG_CONVERT:
3272 _convert_lock(r, lkb);
3278 log_error(ls, "recover_waiters_post type %d", mstype);
3285 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
3286 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
3288 struct dlm_ls *ls = r->res_ls;
3289 struct dlm_lkb *lkb, *safe;
3291 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
3292 if (test(ls, lkb)) {
3293 rsb_set_flag(r, RSB_LOCKS_PURGED);
3295 /* this put should free the lkb */
3296 if (!dlm_put_lkb(lkb))
3297 log_error(ls, "purged lkb not released");
3302 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3304 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
3307 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3309 return is_master_copy(lkb);
3312 static void purge_dead_locks(struct dlm_rsb *r)
3314 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
3315 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
3316 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
3319 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
3321 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
3322 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
3323 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
3326 /* Get rid of locks held by nodes that are gone. */
3328 int dlm_purge_locks(struct dlm_ls *ls)
3332 log_debug(ls, "dlm_purge_locks");
3334 down_write(&ls->ls_root_sem);
3335 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
3339 purge_dead_locks(r);
3345 up_write(&ls->ls_root_sem);
3350 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
3352 struct dlm_rsb *r, *r_ret = NULL;
3354 read_lock(&ls->ls_rsbtbl[bucket].lock);
3355 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
3356 if (!rsb_flag(r, RSB_LOCKS_PURGED))
3359 rsb_clear_flag(r, RSB_LOCKS_PURGED);
3363 read_unlock(&ls->ls_rsbtbl[bucket].lock);
3367 void dlm_grant_after_purge(struct dlm_ls *ls)
3373 r = find_purged_rsb(ls, bucket);
3375 if (bucket == ls->ls_rsbtbl_size - 1)
3382 grant_pending_locks(r);
3383 confirm_master(r, 0);
3391 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
3394 struct dlm_lkb *lkb;
3396 list_for_each_entry(lkb, head, lkb_statequeue) {
3397 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
3403 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
3406 struct dlm_lkb *lkb;
3408 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
3411 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
3414 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
3420 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3421 struct dlm_rsb *r, struct dlm_rcom *rc)
3423 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3426 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
3427 lkb->lkb_ownpid = rl->rl_ownpid;
3428 lkb->lkb_remid = rl->rl_lkid;
3429 lkb->lkb_exflags = rl->rl_exflags;
3430 lkb->lkb_flags = rl->rl_flags & 0x0000FFFF;
3431 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3432 lkb->lkb_lvbseq = rl->rl_lvbseq;
3433 lkb->lkb_rqmode = rl->rl_rqmode;
3434 lkb->lkb_grmode = rl->rl_grmode;
3435 /* don't set lkb_status because add_lkb wants to itself */
3437 lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
3438 lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
3440 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3441 lkb->lkb_lvbptr = allocate_lvb(ls);
3442 if (!lkb->lkb_lvbptr)
3444 lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
3445 sizeof(struct rcom_lock);
3446 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
3449 /* Conversions between PR and CW (middle modes) need special handling.
3450 The real granted mode of these converting locks cannot be determined
3451 until all locks have been rebuilt on the rsb (recover_conversion) */
3453 if (rl->rl_wait_type == DLM_MSG_CONVERT && middle_conversion(lkb)) {
3454 rl->rl_status = DLM_LKSTS_CONVERT;
3455 lkb->lkb_grmode = DLM_LOCK_IV;
3456 rsb_set_flag(r, RSB_RECOVER_CONVERT);
3462 /* This lkb may have been recovered in a previous aborted recovery so we need
3463 to check if the rsb already has an lkb with the given remote nodeid/lkid.
3464 If so we just send back a standard reply. If not, we create a new lkb with
3465 the given values and send back our lkid. We send back our lkid by sending
3466 back the rcom_lock struct we got but with the remid field filled in. */
3468 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
3470 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3472 struct dlm_lkb *lkb;
3475 if (rl->rl_parent_lkid) {
3476 error = -EOPNOTSUPP;
3480 error = find_rsb(ls, rl->rl_name, rl->rl_namelen, R_MASTER, &r);
3486 lkb = search_remid(r, rc->rc_header.h_nodeid, rl->rl_lkid);
3492 error = create_lkb(ls, &lkb);
3496 error = receive_rcom_lock_args(ls, lkb, r, rc);
3503 add_lkb(r, lkb, rl->rl_status);
3507 /* this is the new value returned to the lock holder for
3508 saving in its process-copy lkb */
3509 rl->rl_remid = lkb->lkb_id;
3516 log_print("recover_master_copy %d %x", error, rl->rl_lkid);
3517 rl->rl_result = error;
3521 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
3523 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3525 struct dlm_lkb *lkb;
3528 error = find_lkb(ls, rl->rl_lkid, &lkb);
3530 log_error(ls, "recover_process_copy no lkid %x", rl->rl_lkid);
3534 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3536 error = rl->rl_result;
3538 r = lkb->lkb_resource;
3544 log_debug(ls, "master copy exists %x", lkb->lkb_id);
3547 lkb->lkb_remid = rl->rl_remid;
3550 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
3551 error, lkb->lkb_id);
3554 /* an ack for dlm_recover_locks() which waits for replies from
3555 all the locks it sends to new masters */
3556 dlm_recovered_lock(r);
3565 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
3566 int mode, uint32_t flags, void *name, unsigned int namelen,
3567 uint32_t parent_lkid)
3569 struct dlm_lkb *lkb;
3570 struct dlm_args args;
3575 error = create_lkb(ls, &lkb);
3581 if (flags & DLM_LKF_VALBLK) {
3582 ua->lksb.sb_lvbptr = kmalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
3583 if (!ua->lksb.sb_lvbptr) {
3591 /* After ua is attached to lkb it will be freed by free_lkb().
3592 When DLM_IFL_USER is set, the dlm knows that this is a userspace
3593 lock and that lkb_astparam is the dlm_user_args structure. */
3595 error = set_lock_args(mode, &ua->lksb, flags, namelen, parent_lkid,
3596 FAKE_USER_AST, ua, FAKE_USER_AST, &args);
3597 lkb->lkb_flags |= DLM_IFL_USER;
3598 ua->old_mode = DLM_LOCK_IV;
3605 error = request_lock(ls, lkb, name, namelen, &args);
3621 /* add this new lkb to the per-process list of locks */
3622 spin_lock(&ua->proc->locks_spin);
3623 kref_get(&lkb->lkb_ref);
3624 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
3625 spin_unlock(&ua->proc->locks_spin);
3627 unlock_recovery(ls);
3631 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3632 int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
3634 struct dlm_lkb *lkb;
3635 struct dlm_args args;
3636 struct dlm_user_args *ua;
3641 error = find_lkb(ls, lkid, &lkb);
3645 /* user can change the params on its lock when it converts it, or
3646 add an lvb that didn't exist before */
3648 ua = (struct dlm_user_args *)lkb->lkb_astparam;
3650 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
3651 ua->lksb.sb_lvbptr = kmalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
3652 if (!ua->lksb.sb_lvbptr) {
3657 if (lvb_in && ua->lksb.sb_lvbptr)
3658 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
3660 ua->castparam = ua_tmp->castparam;
3661 ua->castaddr = ua_tmp->castaddr;
3662 ua->bastparam = ua_tmp->bastparam;
3663 ua->bastaddr = ua_tmp->bastaddr;
3664 ua->old_mode = lkb->lkb_grmode;
3666 error = set_lock_args(mode, &ua->lksb, flags, 0, 0, FAKE_USER_AST, ua,
3667 FAKE_USER_AST, &args);
3671 error = convert_lock(ls, lkb, &args);
3673 if (error == -EINPROGRESS || error == -EAGAIN)
3678 unlock_recovery(ls);
3683 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3684 uint32_t flags, uint32_t lkid, char *lvb_in)
3686 struct dlm_lkb *lkb;
3687 struct dlm_args args;
3688 struct dlm_user_args *ua;
3693 error = find_lkb(ls, lkid, &lkb);
3697 ua = (struct dlm_user_args *)lkb->lkb_astparam;
3699 if (lvb_in && ua->lksb.sb_lvbptr)
3700 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
3701 ua->castparam = ua_tmp->castparam;
3703 error = set_unlock_args(flags, ua, &args);
3707 error = unlock_lock(ls, lkb, &args);
3709 if (error == -DLM_EUNLOCK)
3714 spin_lock(&ua->proc->locks_spin);
3715 list_del_init(&lkb->lkb_ownqueue);
3716 spin_unlock(&ua->proc->locks_spin);
3718 /* this removes the reference for the proc->locks list added by
3724 unlock_recovery(ls);
3728 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3729 uint32_t flags, uint32_t lkid)
3731 struct dlm_lkb *lkb;
3732 struct dlm_args args;
3733 struct dlm_user_args *ua;
3738 error = find_lkb(ls, lkid, &lkb);
3742 ua = (struct dlm_user_args *)lkb->lkb_astparam;
3743 ua->castparam = ua_tmp->castparam;
3745 error = set_unlock_args(flags, ua, &args);
3749 error = cancel_lock(ls, lkb, &args);
3751 if (error == -DLM_ECANCEL)
3756 /* this lkb was removed from the WAITING queue */
3757 if (lkb->lkb_grmode == DLM_LOCK_IV) {
3758 spin_lock(&ua->proc->locks_spin);
3759 list_del_init(&lkb->lkb_ownqueue);
3760 spin_unlock(&ua->proc->locks_spin);
3766 unlock_recovery(ls);
3770 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
3772 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
3774 if (ua->lksb.sb_lvbptr)
3775 kfree(ua->lksb.sb_lvbptr);
3777 lkb->lkb_astparam = (long)NULL;
3779 /* TODO: propogate to master if needed */
3783 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
3784 Regardless of what rsb queue the lock is on, it's removed and freed. */
3786 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
3788 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
3789 struct dlm_args args;
3792 /* FIXME: we need to handle the case where the lkb is in limbo
3793 while the rsb is being looked up, currently we assert in
3794 _unlock_lock/is_remote because rsb nodeid is -1. */
3796 set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
3798 error = unlock_lock(ls, lkb, &args);
3799 if (error == -DLM_EUNLOCK)
3804 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
3805 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
3806 which we clear here. */
3808 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
3809 list, and no more device_writes should add lkb's to proc->locks list; so we
3810 shouldn't need to take asts_spin or locks_spin here. this assumes that
3811 device reads/writes/closes are serialized -- FIXME: we may need to serialize
3814 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
3816 struct dlm_lkb *lkb, *safe;
3819 mutex_lock(&ls->ls_clear_proc_locks);
3821 list_for_each_entry_safe(lkb, safe, &proc->locks, lkb_ownqueue) {
3822 if (lkb->lkb_ast_type) {
3823 list_del(&lkb->lkb_astqueue);
3827 list_del_init(&lkb->lkb_ownqueue);
3829 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT) {
3830 lkb->lkb_flags |= DLM_IFL_ORPHAN;
3831 orphan_proc_lock(ls, lkb);
3833 lkb->lkb_flags |= DLM_IFL_DEAD;
3834 unlock_proc_lock(ls, lkb);
3837 /* this removes the reference for the proc->locks list
3838 added by dlm_user_request, it may result in the lkb
3843 mutex_unlock(&ls->ls_clear_proc_locks);
3844 unlock_recovery(ls);