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);
90 * Lock compatibilty matrix - thanks Steve
91 * UN = Unlocked state. Not really a state, used as a flag
92 * PD = Padding. Used to make the matrix a nice power of two in size
93 * Other states are the same as the VMS DLM.
94 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
97 static const int __dlm_compat_matrix[8][8] = {
98 /* UN NL CR CW PR PW EX PD */
99 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
100 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
101 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
102 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
103 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
104 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
105 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
106 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
110 * This defines the direction of transfer of LVB data.
111 * Granted mode is the row; requested mode is the column.
112 * Usage: matrix[grmode+1][rqmode+1]
113 * 1 = LVB is returned to the caller
114 * 0 = LVB is written to the resource
115 * -1 = nothing happens to the LVB
118 const int dlm_lvb_operations[8][8] = {
119 /* UN NL CR CW PR PW EX PD*/
120 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
121 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
122 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
123 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
124 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
125 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
126 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
127 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
130 #define modes_compat(gr, rq) \
131 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
133 int dlm_modes_compat(int mode1, int mode2)
135 return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
139 * Compatibility matrix for conversions with QUECVT set.
140 * Granted mode is the row; requested mode is the column.
141 * Usage: matrix[grmode+1][rqmode+1]
144 static const int __quecvt_compat_matrix[8][8] = {
145 /* UN NL CR CW PR PW EX PD */
146 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
147 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
148 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
149 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
150 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
151 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
152 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
153 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
156 void dlm_print_lkb(struct dlm_lkb *lkb)
158 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
159 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
160 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
161 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
162 lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_ast_type);
165 void dlm_print_rsb(struct dlm_rsb *r)
167 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
168 r->res_nodeid, r->res_flags, r->res_first_lkid,
169 r->res_recover_locks_count, r->res_name);
172 void dlm_dump_rsb(struct dlm_rsb *r)
178 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
179 list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
180 printk(KERN_ERR "rsb lookup list\n");
181 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
183 printk(KERN_ERR "rsb grant queue:\n");
184 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
186 printk(KERN_ERR "rsb convert queue:\n");
187 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
189 printk(KERN_ERR "rsb wait queue:\n");
190 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
194 /* Threads cannot use the lockspace while it's being recovered */
196 static inline void lock_recovery(struct dlm_ls *ls)
198 down_read(&ls->ls_in_recovery);
201 static inline void unlock_recovery(struct dlm_ls *ls)
203 up_read(&ls->ls_in_recovery);
206 static inline int lock_recovery_try(struct dlm_ls *ls)
208 return down_read_trylock(&ls->ls_in_recovery);
211 static inline int can_be_queued(struct dlm_lkb *lkb)
213 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
216 static inline int force_blocking_asts(struct dlm_lkb *lkb)
218 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
221 static inline int is_demoted(struct dlm_lkb *lkb)
223 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
226 static inline int is_remote(struct dlm_rsb *r)
228 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
229 return !!r->res_nodeid;
232 static inline int is_process_copy(struct dlm_lkb *lkb)
234 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
237 static inline int is_master_copy(struct dlm_lkb *lkb)
239 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
240 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb););
241 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
244 static inline int middle_conversion(struct dlm_lkb *lkb)
246 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
247 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
252 static inline int down_conversion(struct dlm_lkb *lkb)
254 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
257 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
259 if (is_master_copy(lkb))
262 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
264 lkb->lkb_lksb->sb_status = rv;
265 lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
267 dlm_add_ast(lkb, AST_COMP);
270 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
272 if (is_master_copy(lkb))
273 send_bast(r, lkb, rqmode);
275 lkb->lkb_bastmode = rqmode;
276 dlm_add_ast(lkb, AST_BAST);
281 * Basic operations on rsb's and lkb's
284 static struct dlm_rsb *create_rsb(struct dlm_ls *ls, char *name, int len)
288 r = allocate_rsb(ls, len);
294 memcpy(r->res_name, name, len);
295 mutex_init(&r->res_mutex);
297 INIT_LIST_HEAD(&r->res_lookup);
298 INIT_LIST_HEAD(&r->res_grantqueue);
299 INIT_LIST_HEAD(&r->res_convertqueue);
300 INIT_LIST_HEAD(&r->res_waitqueue);
301 INIT_LIST_HEAD(&r->res_root_list);
302 INIT_LIST_HEAD(&r->res_recover_list);
307 static int search_rsb_list(struct list_head *head, char *name, int len,
308 unsigned int flags, struct dlm_rsb **r_ret)
313 list_for_each_entry(r, head, res_hashchain) {
314 if (len == r->res_length && !memcmp(name, r->res_name, len))
320 if (r->res_nodeid && (flags & R_MASTER))
326 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b,
327 unsigned int flags, struct dlm_rsb **r_ret)
332 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r);
334 kref_get(&r->res_ref);
337 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r);
341 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list);
343 if (dlm_no_directory(ls))
346 if (r->res_nodeid == -1) {
347 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
348 r->res_first_lkid = 0;
349 } else if (r->res_nodeid > 0) {
350 rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
351 r->res_first_lkid = 0;
353 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r););
354 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),);
361 static int search_rsb(struct dlm_ls *ls, char *name, int len, int b,
362 unsigned int flags, struct dlm_rsb **r_ret)
365 write_lock(&ls->ls_rsbtbl[b].lock);
366 error = _search_rsb(ls, name, len, b, flags, r_ret);
367 write_unlock(&ls->ls_rsbtbl[b].lock);
372 * Find rsb in rsbtbl and potentially create/add one
374 * Delaying the release of rsb's has a similar benefit to applications keeping
375 * NL locks on an rsb, but without the guarantee that the cached master value
376 * will still be valid when the rsb is reused. Apps aren't always smart enough
377 * to keep NL locks on an rsb that they may lock again shortly; this can lead
378 * to excessive master lookups and removals if we don't delay the release.
380 * Searching for an rsb means looking through both the normal list and toss
381 * list. When found on the toss list the rsb is moved to the normal list with
382 * ref count of 1; when found on normal list the ref count is incremented.
385 static int find_rsb(struct dlm_ls *ls, char *name, int namelen,
386 unsigned int flags, struct dlm_rsb **r_ret)
388 struct dlm_rsb *r, *tmp;
389 uint32_t hash, bucket;
392 if (dlm_no_directory(ls))
395 hash = jhash(name, namelen, 0);
396 bucket = hash & (ls->ls_rsbtbl_size - 1);
398 error = search_rsb(ls, name, namelen, bucket, flags, &r);
402 if (error == -EBADR && !(flags & R_CREATE))
405 /* the rsb was found but wasn't a master copy */
406 if (error == -ENOTBLK)
410 r = create_rsb(ls, name, namelen);
415 r->res_bucket = bucket;
417 kref_init(&r->res_ref);
419 /* With no directory, the master can be set immediately */
420 if (dlm_no_directory(ls)) {
421 int nodeid = dlm_dir_nodeid(r);
422 if (nodeid == dlm_our_nodeid())
424 r->res_nodeid = nodeid;
427 write_lock(&ls->ls_rsbtbl[bucket].lock);
428 error = _search_rsb(ls, name, namelen, bucket, 0, &tmp);
430 write_unlock(&ls->ls_rsbtbl[bucket].lock);
435 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list);
436 write_unlock(&ls->ls_rsbtbl[bucket].lock);
443 int dlm_find_rsb(struct dlm_ls *ls, char *name, int namelen,
444 unsigned int flags, struct dlm_rsb **r_ret)
446 return find_rsb(ls, name, namelen, flags, r_ret);
449 /* This is only called to add a reference when the code already holds
450 a valid reference to the rsb, so there's no need for locking. */
452 static inline void hold_rsb(struct dlm_rsb *r)
454 kref_get(&r->res_ref);
457 void dlm_hold_rsb(struct dlm_rsb *r)
462 static void toss_rsb(struct kref *kref)
464 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
465 struct dlm_ls *ls = r->res_ls;
467 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
468 kref_init(&r->res_ref);
469 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss);
470 r->res_toss_time = jiffies;
472 free_lvb(r->res_lvbptr);
473 r->res_lvbptr = NULL;
477 /* When all references to the rsb are gone it's transfered to
478 the tossed list for later disposal. */
480 static void put_rsb(struct dlm_rsb *r)
482 struct dlm_ls *ls = r->res_ls;
483 uint32_t bucket = r->res_bucket;
485 write_lock(&ls->ls_rsbtbl[bucket].lock);
486 kref_put(&r->res_ref, toss_rsb);
487 write_unlock(&ls->ls_rsbtbl[bucket].lock);
490 void dlm_put_rsb(struct dlm_rsb *r)
495 /* See comment for unhold_lkb */
497 static void unhold_rsb(struct dlm_rsb *r)
500 rv = kref_put(&r->res_ref, toss_rsb);
501 DLM_ASSERT(!rv, dlm_dump_rsb(r););
504 static void kill_rsb(struct kref *kref)
506 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
508 /* All work is done after the return from kref_put() so we
509 can release the write_lock before the remove and free. */
511 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
512 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
513 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
514 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
515 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
516 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
519 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
520 The rsb must exist as long as any lkb's for it do. */
522 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
525 lkb->lkb_resource = r;
528 static void detach_lkb(struct dlm_lkb *lkb)
530 if (lkb->lkb_resource) {
531 put_rsb(lkb->lkb_resource);
532 lkb->lkb_resource = NULL;
536 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
538 struct dlm_lkb *lkb, *tmp;
542 lkb = allocate_lkb(ls);
546 lkb->lkb_nodeid = -1;
547 lkb->lkb_grmode = DLM_LOCK_IV;
548 kref_init(&lkb->lkb_ref);
549 INIT_LIST_HEAD(&lkb->lkb_ownqueue);
551 get_random_bytes(&bucket, sizeof(bucket));
552 bucket &= (ls->ls_lkbtbl_size - 1);
554 write_lock(&ls->ls_lkbtbl[bucket].lock);
556 /* counter can roll over so we must verify lkid is not in use */
559 lkid = bucket | (ls->ls_lkbtbl[bucket].counter++ << 16);
561 list_for_each_entry(tmp, &ls->ls_lkbtbl[bucket].list,
563 if (tmp->lkb_id != lkid)
571 list_add(&lkb->lkb_idtbl_list, &ls->ls_lkbtbl[bucket].list);
572 write_unlock(&ls->ls_lkbtbl[bucket].lock);
578 static struct dlm_lkb *__find_lkb(struct dlm_ls *ls, uint32_t lkid)
580 uint16_t bucket = lkid & 0xFFFF;
583 list_for_each_entry(lkb, &ls->ls_lkbtbl[bucket].list, lkb_idtbl_list) {
584 if (lkb->lkb_id == lkid)
590 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
593 uint16_t bucket = lkid & 0xFFFF;
595 if (bucket >= ls->ls_lkbtbl_size)
598 read_lock(&ls->ls_lkbtbl[bucket].lock);
599 lkb = __find_lkb(ls, lkid);
601 kref_get(&lkb->lkb_ref);
602 read_unlock(&ls->ls_lkbtbl[bucket].lock);
605 return lkb ? 0 : -ENOENT;
608 static void kill_lkb(struct kref *kref)
610 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
612 /* All work is done after the return from kref_put() so we
613 can release the write_lock before the detach_lkb */
615 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
618 /* __put_lkb() is used when an lkb may not have an rsb attached to
619 it so we need to provide the lockspace explicitly */
621 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
623 uint16_t bucket = lkb->lkb_id & 0xFFFF;
625 write_lock(&ls->ls_lkbtbl[bucket].lock);
626 if (kref_put(&lkb->lkb_ref, kill_lkb)) {
627 list_del(&lkb->lkb_idtbl_list);
628 write_unlock(&ls->ls_lkbtbl[bucket].lock);
632 /* for local/process lkbs, lvbptr points to caller's lksb */
633 if (lkb->lkb_lvbptr && is_master_copy(lkb))
634 free_lvb(lkb->lkb_lvbptr);
638 write_unlock(&ls->ls_lkbtbl[bucket].lock);
643 int dlm_put_lkb(struct dlm_lkb *lkb)
647 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
648 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
650 ls = lkb->lkb_resource->res_ls;
651 return __put_lkb(ls, lkb);
654 /* This is only called to add a reference when the code already holds
655 a valid reference to the lkb, so there's no need for locking. */
657 static inline void hold_lkb(struct dlm_lkb *lkb)
659 kref_get(&lkb->lkb_ref);
662 /* This is called when we need to remove a reference and are certain
663 it's not the last ref. e.g. del_lkb is always called between a
664 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
665 put_lkb would work fine, but would involve unnecessary locking */
667 static inline void unhold_lkb(struct dlm_lkb *lkb)
670 rv = kref_put(&lkb->lkb_ref, kill_lkb);
671 DLM_ASSERT(!rv, dlm_print_lkb(lkb););
674 static void lkb_add_ordered(struct list_head *new, struct list_head *head,
677 struct dlm_lkb *lkb = NULL;
679 list_for_each_entry(lkb, head, lkb_statequeue)
680 if (lkb->lkb_rqmode < mode)
684 list_add_tail(new, head);
686 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
689 /* add/remove lkb to rsb's grant/convert/wait queue */
691 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
693 kref_get(&lkb->lkb_ref);
695 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
697 lkb->lkb_status = status;
700 case DLM_LKSTS_WAITING:
701 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
702 list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
704 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
706 case DLM_LKSTS_GRANTED:
707 /* convention says granted locks kept in order of grmode */
708 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
711 case DLM_LKSTS_CONVERT:
712 if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
713 list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
715 list_add_tail(&lkb->lkb_statequeue,
716 &r->res_convertqueue);
719 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
723 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
726 list_del(&lkb->lkb_statequeue);
730 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
734 add_lkb(r, lkb, sts);
738 /* add/remove lkb from global waiters list of lkb's waiting for
739 a reply from a remote node */
741 static void add_to_waiters(struct dlm_lkb *lkb, int mstype)
743 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
745 mutex_lock(&ls->ls_waiters_mutex);
746 if (lkb->lkb_wait_type) {
747 log_print("add_to_waiters error %d", lkb->lkb_wait_type);
750 lkb->lkb_wait_type = mstype;
751 kref_get(&lkb->lkb_ref);
752 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
754 mutex_unlock(&ls->ls_waiters_mutex);
757 static int _remove_from_waiters(struct dlm_lkb *lkb)
761 if (!lkb->lkb_wait_type) {
762 log_print("remove_from_waiters error");
766 lkb->lkb_wait_type = 0;
767 list_del(&lkb->lkb_wait_reply);
773 static int remove_from_waiters(struct dlm_lkb *lkb)
775 struct dlm_ls *ls = lkb->lkb_resource->res_ls;
778 mutex_lock(&ls->ls_waiters_mutex);
779 error = _remove_from_waiters(lkb);
780 mutex_unlock(&ls->ls_waiters_mutex);
784 static void dir_remove(struct dlm_rsb *r)
788 if (dlm_no_directory(r->res_ls))
791 to_nodeid = dlm_dir_nodeid(r);
792 if (to_nodeid != dlm_our_nodeid())
795 dlm_dir_remove_entry(r->res_ls, to_nodeid,
796 r->res_name, r->res_length);
799 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
800 found since they are in order of newest to oldest? */
802 static int shrink_bucket(struct dlm_ls *ls, int b)
805 int count = 0, found;
809 write_lock(&ls->ls_rsbtbl[b].lock);
810 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss,
812 if (!time_after_eq(jiffies, r->res_toss_time +
813 dlm_config.ci_toss_secs * HZ))
820 write_unlock(&ls->ls_rsbtbl[b].lock);
824 if (kref_put(&r->res_ref, kill_rsb)) {
825 list_del(&r->res_hashchain);
826 write_unlock(&ls->ls_rsbtbl[b].lock);
833 write_unlock(&ls->ls_rsbtbl[b].lock);
834 log_error(ls, "tossed rsb in use %s", r->res_name);
841 void dlm_scan_rsbs(struct dlm_ls *ls)
845 if (dlm_locking_stopped(ls))
848 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
849 shrink_bucket(ls, i);
854 /* lkb is master or local copy */
856 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
858 int b, len = r->res_ls->ls_lvblen;
860 /* b=1 lvb returned to caller
861 b=0 lvb written to rsb or invalidated
864 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
867 if (!lkb->lkb_lvbptr)
870 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
876 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
877 lkb->lkb_lvbseq = r->res_lvbseq;
880 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
881 rsb_set_flag(r, RSB_VALNOTVALID);
885 if (!lkb->lkb_lvbptr)
888 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
892 r->res_lvbptr = allocate_lvb(r->res_ls);
897 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
899 lkb->lkb_lvbseq = r->res_lvbseq;
900 rsb_clear_flag(r, RSB_VALNOTVALID);
903 if (rsb_flag(r, RSB_VALNOTVALID))
904 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
907 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
909 if (lkb->lkb_grmode < DLM_LOCK_PW)
912 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
913 rsb_set_flag(r, RSB_VALNOTVALID);
917 if (!lkb->lkb_lvbptr)
920 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
924 r->res_lvbptr = allocate_lvb(r->res_ls);
929 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
931 rsb_clear_flag(r, RSB_VALNOTVALID);
934 /* lkb is process copy (pc) */
936 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
937 struct dlm_message *ms)
941 if (!lkb->lkb_lvbptr)
944 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
947 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
949 int len = receive_extralen(ms);
950 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
951 lkb->lkb_lvbseq = ms->m_lvbseq;
955 /* Manipulate lkb's on rsb's convert/granted/waiting queues
956 remove_lock -- used for unlock, removes lkb from granted
957 revert_lock -- used for cancel, moves lkb from convert to granted
958 grant_lock -- used for request and convert, adds lkb to granted or
959 moves lkb from convert or waiting to granted
961 Each of these is used for master or local copy lkb's. There is
962 also a _pc() variation used to make the corresponding change on
963 a process copy (pc) lkb. */
965 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
968 lkb->lkb_grmode = DLM_LOCK_IV;
969 /* this unhold undoes the original ref from create_lkb()
970 so this leads to the lkb being freed */
974 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
976 set_lvb_unlock(r, lkb);
977 _remove_lock(r, lkb);
980 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
982 _remove_lock(r, lkb);
985 static void revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
987 lkb->lkb_rqmode = DLM_LOCK_IV;
989 switch (lkb->lkb_status) {
990 case DLM_LKSTS_GRANTED:
992 case DLM_LKSTS_CONVERT:
993 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
995 case DLM_LKSTS_WAITING:
997 lkb->lkb_grmode = DLM_LOCK_IV;
998 /* this unhold undoes the original ref from create_lkb()
999 so this leads to the lkb being freed */
1003 log_print("invalid status for revert %d", lkb->lkb_status);
1007 static void revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1009 revert_lock(r, lkb);
1012 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1014 if (lkb->lkb_grmode != lkb->lkb_rqmode) {
1015 lkb->lkb_grmode = lkb->lkb_rqmode;
1016 if (lkb->lkb_status)
1017 move_lkb(r, lkb, DLM_LKSTS_GRANTED);
1019 add_lkb(r, lkb, DLM_LKSTS_GRANTED);
1022 lkb->lkb_rqmode = DLM_LOCK_IV;
1025 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1027 set_lvb_lock(r, lkb);
1028 _grant_lock(r, lkb);
1029 lkb->lkb_highbast = 0;
1032 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1033 struct dlm_message *ms)
1035 set_lvb_lock_pc(r, lkb, ms);
1036 _grant_lock(r, lkb);
1039 /* called by grant_pending_locks() which means an async grant message must
1040 be sent to the requesting node in addition to granting the lock if the
1041 lkb belongs to a remote node. */
1043 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
1046 if (is_master_copy(lkb))
1049 queue_cast(r, lkb, 0);
1052 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
1054 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
1056 if (lkb->lkb_id == first->lkb_id)
1062 /* Check if the given lkb conflicts with another lkb on the queue. */
1064 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
1066 struct dlm_lkb *this;
1068 list_for_each_entry(this, head, lkb_statequeue) {
1071 if (!modes_compat(this, lkb))
1078 * "A conversion deadlock arises with a pair of lock requests in the converting
1079 * queue for one resource. The granted mode of each lock blocks the requested
1080 * mode of the other lock."
1082 * Part 2: if the granted mode of lkb is preventing the first lkb in the
1083 * convert queue from being granted, then demote lkb (set grmode to NL).
1084 * This second form requires that we check for conv-deadlk even when
1085 * now == 0 in _can_be_granted().
1088 * Granted Queue: empty
1089 * Convert Queue: NL->EX (first lock)
1090 * PR->EX (second lock)
1092 * The first lock can't be granted because of the granted mode of the second
1093 * lock and the second lock can't be granted because it's not first in the
1094 * list. We demote the granted mode of the second lock (the lkb passed to this
1097 * After the resolution, the "grant pending" function needs to go back and try
1098 * to grant locks on the convert queue again since the first lock can now be
1102 static int conversion_deadlock_detect(struct dlm_rsb *rsb, struct dlm_lkb *lkb)
1104 struct dlm_lkb *this, *first = NULL, *self = NULL;
1106 list_for_each_entry(this, &rsb->res_convertqueue, lkb_statequeue) {
1114 if (!modes_compat(this, lkb) && !modes_compat(lkb, this))
1118 /* if lkb is on the convert queue and is preventing the first
1119 from being granted, then there's deadlock and we demote lkb.
1120 multiple converting locks may need to do this before the first
1121 converting lock can be granted. */
1123 if (self && self != first) {
1124 if (!modes_compat(lkb, first) &&
1125 !queue_conflict(&rsb->res_grantqueue, first))
1133 * Return 1 if the lock can be granted, 0 otherwise.
1134 * Also detect and resolve conversion deadlocks.
1136 * lkb is the lock to be granted
1138 * now is 1 if the function is being called in the context of the
1139 * immediate request, it is 0 if called later, after the lock has been
1142 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1145 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1147 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
1150 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1151 * a new request for a NL mode lock being blocked.
1153 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1154 * request, then it would be granted. In essence, the use of this flag
1155 * tells the Lock Manager to expedite theis request by not considering
1156 * what may be in the CONVERTING or WAITING queues... As of this
1157 * writing, the EXPEDITE flag can be used only with new requests for NL
1158 * mode locks. This flag is not valid for conversion requests.
1160 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1161 * conversion or used with a non-NL requested mode. We also know an
1162 * EXPEDITE request is always granted immediately, so now must always
1163 * be 1. The full condition to grant an expedite request: (now &&
1164 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1165 * therefore be shortened to just checking the flag.
1168 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
1172 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1173 * added to the remaining conditions.
1176 if (queue_conflict(&r->res_grantqueue, lkb))
1180 * 6-3: By default, a conversion request is immediately granted if the
1181 * requested mode is compatible with the modes of all other granted
1185 if (queue_conflict(&r->res_convertqueue, lkb))
1189 * 6-5: But the default algorithm for deciding whether to grant or
1190 * queue conversion requests does not by itself guarantee that such
1191 * requests are serviced on a "first come first serve" basis. This, in
1192 * turn, can lead to a phenomenon known as "indefinate postponement".
1194 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1195 * the system service employed to request a lock conversion. This flag
1196 * forces certain conversion requests to be queued, even if they are
1197 * compatible with the granted modes of other locks on the same
1198 * resource. Thus, the use of this flag results in conversion requests
1199 * being ordered on a "first come first servce" basis.
1201 * DCT: This condition is all about new conversions being able to occur
1202 * "in place" while the lock remains on the granted queue (assuming
1203 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1204 * doesn't _have_ to go onto the convert queue where it's processed in
1205 * order. The "now" variable is necessary to distinguish converts
1206 * being received and processed for the first time now, because once a
1207 * convert is moved to the conversion queue the condition below applies
1208 * requiring fifo granting.
1211 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
1215 * The NOORDER flag is set to avoid the standard vms rules on grant
1219 if (lkb->lkb_exflags & DLM_LKF_NOORDER)
1223 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1224 * granted until all other conversion requests ahead of it are granted
1228 if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
1232 * 6-4: By default, a new request is immediately granted only if all
1233 * three of the following conditions are satisfied when the request is
1235 * - The queue of ungranted conversion requests for the resource is
1237 * - The queue of ungranted new requests for the resource is empty.
1238 * - The mode of the new request is compatible with the most
1239 * restrictive mode of all granted locks on the resource.
1242 if (now && !conv && list_empty(&r->res_convertqueue) &&
1243 list_empty(&r->res_waitqueue))
1247 * 6-4: Once a lock request is in the queue of ungranted new requests,
1248 * it cannot be granted until the queue of ungranted conversion
1249 * requests is empty, all ungranted new requests ahead of it are
1250 * granted and/or canceled, and it is compatible with the granted mode
1251 * of the most restrictive lock granted on the resource.
1254 if (!now && !conv && list_empty(&r->res_convertqueue) &&
1255 first_in_list(lkb, &r->res_waitqueue))
1260 * The following, enabled by CONVDEADLK, departs from VMS.
1263 if (conv && (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) &&
1264 conversion_deadlock_detect(r, lkb)) {
1265 lkb->lkb_grmode = DLM_LOCK_NL;
1266 lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
1273 * The ALTPR and ALTCW flags aren't traditional lock manager flags, but are a
1274 * simple way to provide a big optimization to applications that can use them.
1277 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now)
1279 uint32_t flags = lkb->lkb_exflags;
1281 int8_t alt = 0, rqmode = lkb->lkb_rqmode;
1283 rv = _can_be_granted(r, lkb, now);
1287 if (lkb->lkb_sbflags & DLM_SBF_DEMOTED)
1290 if (rqmode != DLM_LOCK_PR && flags & DLM_LKF_ALTPR)
1292 else if (rqmode != DLM_LOCK_CW && flags & DLM_LKF_ALTCW)
1296 lkb->lkb_rqmode = alt;
1297 rv = _can_be_granted(r, lkb, now);
1299 lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
1301 lkb->lkb_rqmode = rqmode;
1307 static int grant_pending_convert(struct dlm_rsb *r, int high)
1309 struct dlm_lkb *lkb, *s;
1310 int hi, demoted, quit, grant_restart, demote_restart;
1318 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
1319 demoted = is_demoted(lkb);
1320 if (can_be_granted(r, lkb, 0)) {
1321 grant_lock_pending(r, lkb);
1324 hi = max_t(int, lkb->lkb_rqmode, hi);
1325 if (!demoted && is_demoted(lkb))
1332 if (demote_restart && !quit) {
1337 return max_t(int, high, hi);
1340 static int grant_pending_wait(struct dlm_rsb *r, int high)
1342 struct dlm_lkb *lkb, *s;
1344 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
1345 if (can_be_granted(r, lkb, 0))
1346 grant_lock_pending(r, lkb);
1348 high = max_t(int, lkb->lkb_rqmode, high);
1354 static void grant_pending_locks(struct dlm_rsb *r)
1356 struct dlm_lkb *lkb, *s;
1357 int high = DLM_LOCK_IV;
1359 DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
1361 high = grant_pending_convert(r, high);
1362 high = grant_pending_wait(r, high);
1364 if (high == DLM_LOCK_IV)
1368 * If there are locks left on the wait/convert queue then send blocking
1369 * ASTs to granted locks based on the largest requested mode (high)
1370 * found above. FIXME: highbast < high comparison not valid for PR/CW.
1373 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
1374 if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
1375 !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
1376 queue_bast(r, lkb, high);
1377 lkb->lkb_highbast = high;
1382 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
1383 struct dlm_lkb *lkb)
1387 list_for_each_entry(gr, head, lkb_statequeue) {
1388 if (gr->lkb_bastaddr &&
1389 gr->lkb_highbast < lkb->lkb_rqmode &&
1390 !modes_compat(gr, lkb)) {
1391 queue_bast(r, gr, lkb->lkb_rqmode);
1392 gr->lkb_highbast = lkb->lkb_rqmode;
1397 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
1399 send_bast_queue(r, &r->res_grantqueue, lkb);
1402 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
1404 send_bast_queue(r, &r->res_grantqueue, lkb);
1405 send_bast_queue(r, &r->res_convertqueue, lkb);
1408 /* set_master(r, lkb) -- set the master nodeid of a resource
1410 The purpose of this function is to set the nodeid field in the given
1411 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1412 known, it can just be copied to the lkb and the function will return
1413 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1414 before it can be copied to the lkb.
1416 When the rsb nodeid is being looked up remotely, the initial lkb
1417 causing the lookup is kept on the ls_waiters list waiting for the
1418 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1419 on the rsb's res_lookup list until the master is verified.
1422 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1423 1: the rsb master is not available and the lkb has been placed on
1427 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
1429 struct dlm_ls *ls = r->res_ls;
1430 int error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid();
1432 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
1433 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
1434 r->res_first_lkid = lkb->lkb_id;
1435 lkb->lkb_nodeid = r->res_nodeid;
1439 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
1440 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
1444 if (r->res_nodeid == 0) {
1445 lkb->lkb_nodeid = 0;
1449 if (r->res_nodeid > 0) {
1450 lkb->lkb_nodeid = r->res_nodeid;
1454 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r););
1456 dir_nodeid = dlm_dir_nodeid(r);
1458 if (dir_nodeid != our_nodeid) {
1459 r->res_first_lkid = lkb->lkb_id;
1460 send_lookup(r, lkb);
1465 /* It's possible for dlm_scand to remove an old rsb for
1466 this same resource from the toss list, us to create
1467 a new one, look up the master locally, and find it
1468 already exists just before dlm_scand does the
1469 dir_remove() on the previous rsb. */
1471 error = dlm_dir_lookup(ls, our_nodeid, r->res_name,
1472 r->res_length, &ret_nodeid);
1475 log_debug(ls, "dir_lookup error %d %s", error, r->res_name);
1479 if (ret_nodeid == our_nodeid) {
1480 r->res_first_lkid = 0;
1482 lkb->lkb_nodeid = 0;
1484 r->res_first_lkid = lkb->lkb_id;
1485 r->res_nodeid = ret_nodeid;
1486 lkb->lkb_nodeid = ret_nodeid;
1491 static void process_lookup_list(struct dlm_rsb *r)
1493 struct dlm_lkb *lkb, *safe;
1495 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
1496 list_del(&lkb->lkb_rsb_lookup);
1497 _request_lock(r, lkb);
1502 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1504 static void confirm_master(struct dlm_rsb *r, int error)
1506 struct dlm_lkb *lkb;
1508 if (!r->res_first_lkid)
1514 r->res_first_lkid = 0;
1515 process_lookup_list(r);
1519 /* the remote master didn't queue our NOQUEUE request;
1520 make a waiting lkb the first_lkid */
1522 r->res_first_lkid = 0;
1524 if (!list_empty(&r->res_lookup)) {
1525 lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
1527 list_del(&lkb->lkb_rsb_lookup);
1528 r->res_first_lkid = lkb->lkb_id;
1529 _request_lock(r, lkb);
1535 log_error(r->res_ls, "confirm_master unknown error %d", error);
1539 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
1540 int namelen, uint32_t parent_lkid, void *ast,
1541 void *astarg, void *bast, struct dlm_args *args)
1545 /* check for invalid arg usage */
1547 if (mode < 0 || mode > DLM_LOCK_EX)
1550 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
1553 if (flags & DLM_LKF_CANCEL)
1556 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
1559 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
1562 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
1565 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
1568 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
1571 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
1574 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
1580 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
1583 /* parent/child locks not yet supported */
1587 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
1590 /* these args will be copied to the lkb in validate_lock_args,
1591 it cannot be done now because when converting locks, fields in
1592 an active lkb cannot be modified before locking the rsb */
1594 args->flags = flags;
1595 args->astaddr = ast;
1596 args->astparam = (long) astarg;
1597 args->bastaddr = bast;
1605 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
1607 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
1608 DLM_LKF_FORCEUNLOCK))
1611 args->flags = flags;
1612 args->astparam = (long) astarg;
1616 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
1617 struct dlm_args *args)
1621 if (args->flags & DLM_LKF_CONVERT) {
1622 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
1625 if (args->flags & DLM_LKF_QUECVT &&
1626 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
1630 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
1633 if (lkb->lkb_wait_type)
1637 lkb->lkb_exflags = args->flags;
1638 lkb->lkb_sbflags = 0;
1639 lkb->lkb_astaddr = args->astaddr;
1640 lkb->lkb_astparam = args->astparam;
1641 lkb->lkb_bastaddr = args->bastaddr;
1642 lkb->lkb_rqmode = args->mode;
1643 lkb->lkb_lksb = args->lksb;
1644 lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
1645 lkb->lkb_ownpid = (int) current->pid;
1651 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
1655 if (lkb->lkb_flags & DLM_IFL_MSTCPY)
1658 if (args->flags & DLM_LKF_FORCEUNLOCK)
1661 if (args->flags & DLM_LKF_CANCEL &&
1662 lkb->lkb_status == DLM_LKSTS_GRANTED)
1665 if (!(args->flags & DLM_LKF_CANCEL) &&
1666 lkb->lkb_status != DLM_LKSTS_GRANTED)
1670 if (lkb->lkb_wait_type)
1674 lkb->lkb_exflags = args->flags;
1675 lkb->lkb_sbflags = 0;
1676 lkb->lkb_astparam = args->astparam;
1684 * Four stage 4 varieties:
1685 * do_request(), do_convert(), do_unlock(), do_cancel()
1686 * These are called on the master node for the given lock and
1687 * from the central locking logic.
1690 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
1694 if (can_be_granted(r, lkb, 1)) {
1696 queue_cast(r, lkb, 0);
1700 if (can_be_queued(lkb)) {
1701 error = -EINPROGRESS;
1702 add_lkb(r, lkb, DLM_LKSTS_WAITING);
1703 send_blocking_asts(r, lkb);
1708 if (force_blocking_asts(lkb))
1709 send_blocking_asts_all(r, lkb);
1710 queue_cast(r, lkb, -EAGAIN);
1716 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
1720 /* changing an existing lock may allow others to be granted */
1722 if (can_be_granted(r, lkb, 1)) {
1724 queue_cast(r, lkb, 0);
1725 grant_pending_locks(r);
1729 if (can_be_queued(lkb)) {
1730 if (is_demoted(lkb))
1731 grant_pending_locks(r);
1732 error = -EINPROGRESS;
1734 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
1735 send_blocking_asts(r, lkb);
1740 if (force_blocking_asts(lkb))
1741 send_blocking_asts_all(r, lkb);
1742 queue_cast(r, lkb, -EAGAIN);
1748 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1750 remove_lock(r, lkb);
1751 queue_cast(r, lkb, -DLM_EUNLOCK);
1752 grant_pending_locks(r);
1753 return -DLM_EUNLOCK;
1756 /* FIXME: if revert_lock() finds that the lkb is granted, we should
1757 skip the queue_cast(ECANCEL). It indicates that the request/convert
1758 completed (and queued a normal ast) just before the cancel; we don't
1759 want to clobber the sb_result for the normal ast with ECANCEL. */
1761 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
1763 revert_lock(r, lkb);
1764 queue_cast(r, lkb, -DLM_ECANCEL);
1765 grant_pending_locks(r);
1766 return -DLM_ECANCEL;
1770 * Four stage 3 varieties:
1771 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
1774 /* add a new lkb to a possibly new rsb, called by requesting process */
1776 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1780 /* set_master: sets lkb nodeid from r */
1782 error = set_master(r, lkb);
1791 /* receive_request() calls do_request() on remote node */
1792 error = send_request(r, lkb);
1794 error = do_request(r, lkb);
1799 /* change some property of an existing lkb, e.g. mode */
1801 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1806 /* receive_convert() calls do_convert() on remote node */
1807 error = send_convert(r, lkb);
1809 error = do_convert(r, lkb);
1814 /* remove an existing lkb from the granted queue */
1816 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1821 /* receive_unlock() calls do_unlock() on remote node */
1822 error = send_unlock(r, lkb);
1824 error = do_unlock(r, lkb);
1829 /* remove an existing lkb from the convert or wait queue */
1831 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1836 /* receive_cancel() calls do_cancel() on remote node */
1837 error = send_cancel(r, lkb);
1839 error = do_cancel(r, lkb);
1845 * Four stage 2 varieties:
1846 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
1849 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name,
1850 int len, struct dlm_args *args)
1855 error = validate_lock_args(ls, lkb, args);
1859 error = find_rsb(ls, name, len, R_CREATE, &r);
1866 lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
1868 error = _request_lock(r, lkb);
1877 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1878 struct dlm_args *args)
1883 r = lkb->lkb_resource;
1888 error = validate_lock_args(ls, lkb, args);
1892 error = _convert_lock(r, lkb);
1899 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1900 struct dlm_args *args)
1905 r = lkb->lkb_resource;
1910 error = validate_unlock_args(lkb, args);
1914 error = _unlock_lock(r, lkb);
1921 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
1922 struct dlm_args *args)
1927 r = lkb->lkb_resource;
1932 error = validate_unlock_args(lkb, args);
1936 error = _cancel_lock(r, lkb);
1944 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
1947 int dlm_lock(dlm_lockspace_t *lockspace,
1949 struct dlm_lksb *lksb,
1952 unsigned int namelen,
1953 uint32_t parent_lkid,
1954 void (*ast) (void *astarg),
1956 void (*bast) (void *astarg, int mode))
1959 struct dlm_lkb *lkb;
1960 struct dlm_args args;
1961 int error, convert = flags & DLM_LKF_CONVERT;
1963 ls = dlm_find_lockspace_local(lockspace);
1970 error = find_lkb(ls, lksb->sb_lkid, &lkb);
1972 error = create_lkb(ls, &lkb);
1977 error = set_lock_args(mode, lksb, flags, namelen, parent_lkid, ast,
1978 astarg, bast, &args);
1983 error = convert_lock(ls, lkb, &args);
1985 error = request_lock(ls, lkb, name, namelen, &args);
1987 if (error == -EINPROGRESS)
1990 if (convert || error)
1992 if (error == -EAGAIN)
1995 unlock_recovery(ls);
1996 dlm_put_lockspace(ls);
2000 int dlm_unlock(dlm_lockspace_t *lockspace,
2003 struct dlm_lksb *lksb,
2007 struct dlm_lkb *lkb;
2008 struct dlm_args args;
2011 ls = dlm_find_lockspace_local(lockspace);
2017 error = find_lkb(ls, lkid, &lkb);
2021 error = set_unlock_args(flags, astarg, &args);
2025 if (flags & DLM_LKF_CANCEL)
2026 error = cancel_lock(ls, lkb, &args);
2028 error = unlock_lock(ls, lkb, &args);
2030 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
2035 unlock_recovery(ls);
2036 dlm_put_lockspace(ls);
2041 * send/receive routines for remote operations and replies
2045 * send_request receive_request
2046 * send_convert receive_convert
2047 * send_unlock receive_unlock
2048 * send_cancel receive_cancel
2049 * send_grant receive_grant
2050 * send_bast receive_bast
2051 * send_lookup receive_lookup
2052 * send_remove receive_remove
2055 * receive_request_reply send_request_reply
2056 * receive_convert_reply send_convert_reply
2057 * receive_unlock_reply send_unlock_reply
2058 * receive_cancel_reply send_cancel_reply
2059 * receive_lookup_reply send_lookup_reply
2062 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
2063 int to_nodeid, int mstype,
2064 struct dlm_message **ms_ret,
2065 struct dlm_mhandle **mh_ret)
2067 struct dlm_message *ms;
2068 struct dlm_mhandle *mh;
2070 int mb_len = sizeof(struct dlm_message);
2073 case DLM_MSG_REQUEST:
2074 case DLM_MSG_LOOKUP:
2075 case DLM_MSG_REMOVE:
2076 mb_len += r->res_length;
2078 case DLM_MSG_CONVERT:
2079 case DLM_MSG_UNLOCK:
2080 case DLM_MSG_REQUEST_REPLY:
2081 case DLM_MSG_CONVERT_REPLY:
2083 if (lkb && lkb->lkb_lvbptr)
2084 mb_len += r->res_ls->ls_lvblen;
2088 /* get_buffer gives us a message handle (mh) that we need to
2089 pass into lowcomms_commit and a message buffer (mb) that we
2090 write our data into */
2092 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_KERNEL, &mb);
2096 memset(mb, 0, mb_len);
2098 ms = (struct dlm_message *) mb;
2100 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
2101 ms->m_header.h_lockspace = r->res_ls->ls_global_id;
2102 ms->m_header.h_nodeid = dlm_our_nodeid();
2103 ms->m_header.h_length = mb_len;
2104 ms->m_header.h_cmd = DLM_MSG;
2106 ms->m_type = mstype;
2113 /* further lowcomms enhancements or alternate implementations may make
2114 the return value from this function useful at some point */
2116 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms)
2118 dlm_message_out(ms);
2119 dlm_lowcomms_commit_buffer(mh);
2123 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
2124 struct dlm_message *ms)
2126 ms->m_nodeid = lkb->lkb_nodeid;
2127 ms->m_pid = lkb->lkb_ownpid;
2128 ms->m_lkid = lkb->lkb_id;
2129 ms->m_remid = lkb->lkb_remid;
2130 ms->m_exflags = lkb->lkb_exflags;
2131 ms->m_sbflags = lkb->lkb_sbflags;
2132 ms->m_flags = lkb->lkb_flags;
2133 ms->m_lvbseq = lkb->lkb_lvbseq;
2134 ms->m_status = lkb->lkb_status;
2135 ms->m_grmode = lkb->lkb_grmode;
2136 ms->m_rqmode = lkb->lkb_rqmode;
2137 ms->m_hash = r->res_hash;
2139 /* m_result and m_bastmode are set from function args,
2140 not from lkb fields */
2142 if (lkb->lkb_bastaddr)
2143 ms->m_asts |= AST_BAST;
2144 if (lkb->lkb_astaddr)
2145 ms->m_asts |= AST_COMP;
2147 /* compare with switch in create_message; send_remove() doesn't
2150 switch (ms->m_type) {
2151 case DLM_MSG_REQUEST:
2152 case DLM_MSG_LOOKUP:
2153 memcpy(ms->m_extra, r->res_name, r->res_length);
2155 case DLM_MSG_CONVERT:
2156 case DLM_MSG_UNLOCK:
2157 case DLM_MSG_REQUEST_REPLY:
2158 case DLM_MSG_CONVERT_REPLY:
2160 if (!lkb->lkb_lvbptr)
2162 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
2167 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
2169 struct dlm_message *ms;
2170 struct dlm_mhandle *mh;
2171 int to_nodeid, error;
2173 add_to_waiters(lkb, mstype);
2175 to_nodeid = r->res_nodeid;
2177 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2181 send_args(r, lkb, ms);
2183 error = send_message(mh, ms);
2189 remove_from_waiters(lkb);
2193 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
2195 return send_common(r, lkb, DLM_MSG_REQUEST);
2198 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
2202 error = send_common(r, lkb, DLM_MSG_CONVERT);
2204 /* down conversions go without a reply from the master */
2205 if (!error && down_conversion(lkb)) {
2206 remove_from_waiters(lkb);
2207 r->res_ls->ls_stub_ms.m_result = 0;
2208 r->res_ls->ls_stub_ms.m_flags = lkb->lkb_flags;
2209 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
2215 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2216 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2217 that the master is still correct. */
2219 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2221 return send_common(r, lkb, DLM_MSG_UNLOCK);
2224 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
2226 return send_common(r, lkb, DLM_MSG_CANCEL);
2229 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
2231 struct dlm_message *ms;
2232 struct dlm_mhandle *mh;
2233 int to_nodeid, error;
2235 to_nodeid = lkb->lkb_nodeid;
2237 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh);
2241 send_args(r, lkb, ms);
2245 error = send_message(mh, ms);
2250 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
2252 struct dlm_message *ms;
2253 struct dlm_mhandle *mh;
2254 int to_nodeid, error;
2256 to_nodeid = lkb->lkb_nodeid;
2258 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh);
2262 send_args(r, lkb, ms);
2264 ms->m_bastmode = mode;
2266 error = send_message(mh, ms);
2271 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
2273 struct dlm_message *ms;
2274 struct dlm_mhandle *mh;
2275 int to_nodeid, error;
2277 add_to_waiters(lkb, DLM_MSG_LOOKUP);
2279 to_nodeid = dlm_dir_nodeid(r);
2281 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh);
2285 send_args(r, lkb, ms);
2287 error = send_message(mh, ms);
2293 remove_from_waiters(lkb);
2297 static int send_remove(struct dlm_rsb *r)
2299 struct dlm_message *ms;
2300 struct dlm_mhandle *mh;
2301 int to_nodeid, error;
2303 to_nodeid = dlm_dir_nodeid(r);
2305 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh);
2309 memcpy(ms->m_extra, r->res_name, r->res_length);
2310 ms->m_hash = r->res_hash;
2312 error = send_message(mh, ms);
2317 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2320 struct dlm_message *ms;
2321 struct dlm_mhandle *mh;
2322 int to_nodeid, error;
2324 to_nodeid = lkb->lkb_nodeid;
2326 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh);
2330 send_args(r, lkb, ms);
2334 error = send_message(mh, ms);
2339 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2341 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
2344 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2346 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
2349 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2351 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
2354 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
2356 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
2359 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
2360 int ret_nodeid, int rv)
2362 struct dlm_rsb *r = &ls->ls_stub_rsb;
2363 struct dlm_message *ms;
2364 struct dlm_mhandle *mh;
2365 int error, nodeid = ms_in->m_header.h_nodeid;
2367 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh);
2371 ms->m_lkid = ms_in->m_lkid;
2373 ms->m_nodeid = ret_nodeid;
2375 error = send_message(mh, ms);
2380 /* which args we save from a received message depends heavily on the type
2381 of message, unlike the send side where we can safely send everything about
2382 the lkb for any type of message */
2384 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
2386 lkb->lkb_exflags = ms->m_exflags;
2387 lkb->lkb_sbflags = ms->m_sbflags;
2388 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2389 (ms->m_flags & 0x0000FFFF);
2392 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2394 lkb->lkb_sbflags = ms->m_sbflags;
2395 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
2396 (ms->m_flags & 0x0000FFFF);
2399 static int receive_extralen(struct dlm_message *ms)
2401 return (ms->m_header.h_length - sizeof(struct dlm_message));
2404 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
2405 struct dlm_message *ms)
2409 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2410 if (!lkb->lkb_lvbptr)
2411 lkb->lkb_lvbptr = allocate_lvb(ls);
2412 if (!lkb->lkb_lvbptr)
2414 len = receive_extralen(ms);
2415 memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
2420 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2421 struct dlm_message *ms)
2423 lkb->lkb_nodeid = ms->m_header.h_nodeid;
2424 lkb->lkb_ownpid = ms->m_pid;
2425 lkb->lkb_remid = ms->m_lkid;
2426 lkb->lkb_grmode = DLM_LOCK_IV;
2427 lkb->lkb_rqmode = ms->m_rqmode;
2428 lkb->lkb_bastaddr = (void *) (long) (ms->m_asts & AST_BAST);
2429 lkb->lkb_astaddr = (void *) (long) (ms->m_asts & AST_COMP);
2431 DLM_ASSERT(is_master_copy(lkb), dlm_print_lkb(lkb););
2433 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
2434 /* lkb was just created so there won't be an lvb yet */
2435 lkb->lkb_lvbptr = allocate_lvb(ls);
2436 if (!lkb->lkb_lvbptr)
2443 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2444 struct dlm_message *ms)
2446 if (lkb->lkb_nodeid != ms->m_header.h_nodeid) {
2447 log_error(ls, "convert_args nodeid %d %d lkid %x %x",
2448 lkb->lkb_nodeid, ms->m_header.h_nodeid,
2449 lkb->lkb_id, lkb->lkb_remid);
2453 if (!is_master_copy(lkb))
2456 if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2459 if (receive_lvb(ls, lkb, ms))
2462 lkb->lkb_rqmode = ms->m_rqmode;
2463 lkb->lkb_lvbseq = ms->m_lvbseq;
2468 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2469 struct dlm_message *ms)
2471 if (!is_master_copy(lkb))
2473 if (receive_lvb(ls, lkb, ms))
2478 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
2479 uses to send a reply and that the remote end uses to process the reply. */
2481 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
2483 struct dlm_lkb *lkb = &ls->ls_stub_lkb;
2484 lkb->lkb_nodeid = ms->m_header.h_nodeid;
2485 lkb->lkb_remid = ms->m_lkid;
2488 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
2490 struct dlm_lkb *lkb;
2494 error = create_lkb(ls, &lkb);
2498 receive_flags(lkb, ms);
2499 lkb->lkb_flags |= DLM_IFL_MSTCPY;
2500 error = receive_request_args(ls, lkb, ms);
2506 namelen = receive_extralen(ms);
2508 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r);
2517 error = do_request(r, lkb);
2518 send_request_reply(r, lkb, error);
2523 if (error == -EINPROGRESS)
2530 setup_stub_lkb(ls, ms);
2531 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2534 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
2536 struct dlm_lkb *lkb;
2538 int error, reply = 1;
2540 error = find_lkb(ls, ms->m_remid, &lkb);
2544 r = lkb->lkb_resource;
2549 receive_flags(lkb, ms);
2550 error = receive_convert_args(ls, lkb, ms);
2553 reply = !down_conversion(lkb);
2555 error = do_convert(r, lkb);
2558 send_convert_reply(r, lkb, error);
2566 setup_stub_lkb(ls, ms);
2567 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2570 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
2572 struct dlm_lkb *lkb;
2576 error = find_lkb(ls, ms->m_remid, &lkb);
2580 r = lkb->lkb_resource;
2585 receive_flags(lkb, ms);
2586 error = receive_unlock_args(ls, lkb, ms);
2590 error = do_unlock(r, lkb);
2592 send_unlock_reply(r, lkb, error);
2600 setup_stub_lkb(ls, ms);
2601 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2604 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
2606 struct dlm_lkb *lkb;
2610 error = find_lkb(ls, ms->m_remid, &lkb);
2614 receive_flags(lkb, ms);
2616 r = lkb->lkb_resource;
2621 error = do_cancel(r, lkb);
2622 send_cancel_reply(r, lkb, error);
2630 setup_stub_lkb(ls, ms);
2631 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
2634 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
2636 struct dlm_lkb *lkb;
2640 error = find_lkb(ls, ms->m_remid, &lkb);
2642 log_error(ls, "receive_grant no lkb");
2645 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2647 r = lkb->lkb_resource;
2652 receive_flags_reply(lkb, ms);
2653 grant_lock_pc(r, lkb, ms);
2654 queue_cast(r, lkb, 0);
2661 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
2663 struct dlm_lkb *lkb;
2667 error = find_lkb(ls, ms->m_remid, &lkb);
2669 log_error(ls, "receive_bast no lkb");
2672 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2674 r = lkb->lkb_resource;
2679 queue_bast(r, lkb, ms->m_bastmode);
2686 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
2688 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid;
2690 from_nodeid = ms->m_header.h_nodeid;
2691 our_nodeid = dlm_our_nodeid();
2693 len = receive_extralen(ms);
2695 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
2696 if (dir_nodeid != our_nodeid) {
2697 log_error(ls, "lookup dir_nodeid %d from %d",
2698 dir_nodeid, from_nodeid);
2704 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid);
2706 /* Optimization: we're master so treat lookup as a request */
2707 if (!error && ret_nodeid == our_nodeid) {
2708 receive_request(ls, ms);
2712 send_lookup_reply(ls, ms, ret_nodeid, error);
2715 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
2717 int len, dir_nodeid, from_nodeid;
2719 from_nodeid = ms->m_header.h_nodeid;
2721 len = receive_extralen(ms);
2723 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash);
2724 if (dir_nodeid != dlm_our_nodeid()) {
2725 log_error(ls, "remove dir entry dir_nodeid %d from %d",
2726 dir_nodeid, from_nodeid);
2730 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len);
2733 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
2735 struct dlm_lkb *lkb;
2739 error = find_lkb(ls, ms->m_remid, &lkb);
2741 log_error(ls, "receive_request_reply no lkb");
2744 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2746 mstype = lkb->lkb_wait_type;
2747 error = remove_from_waiters(lkb);
2749 log_error(ls, "receive_request_reply not on waiters");
2753 /* this is the value returned from do_request() on the master */
2754 error = ms->m_result;
2756 r = lkb->lkb_resource;
2760 /* Optimization: the dir node was also the master, so it took our
2761 lookup as a request and sent request reply instead of lookup reply */
2762 if (mstype == DLM_MSG_LOOKUP) {
2763 r->res_nodeid = ms->m_header.h_nodeid;
2764 lkb->lkb_nodeid = r->res_nodeid;
2769 /* request would block (be queued) on remote master;
2770 the unhold undoes the original ref from create_lkb()
2771 so it leads to the lkb being freed */
2772 queue_cast(r, lkb, -EAGAIN);
2773 confirm_master(r, -EAGAIN);
2779 /* request was queued or granted on remote master */
2780 receive_flags_reply(lkb, ms);
2781 lkb->lkb_remid = ms->m_lkid;
2783 add_lkb(r, lkb, DLM_LKSTS_WAITING);
2785 grant_lock_pc(r, lkb, ms);
2786 queue_cast(r, lkb, 0);
2788 confirm_master(r, error);
2793 /* find_rsb failed to find rsb or rsb wasn't master */
2795 lkb->lkb_nodeid = -1;
2796 _request_lock(r, lkb);
2800 log_error(ls, "receive_request_reply error %d", error);
2809 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
2810 struct dlm_message *ms)
2812 int error = ms->m_result;
2814 /* this is the value returned from do_convert() on the master */
2818 /* convert would block (be queued) on remote master */
2819 queue_cast(r, lkb, -EAGAIN);
2823 /* convert was queued on remote master */
2825 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
2829 /* convert was granted on remote master */
2830 receive_flags_reply(lkb, ms);
2831 grant_lock_pc(r, lkb, ms);
2832 queue_cast(r, lkb, 0);
2836 log_error(r->res_ls, "receive_convert_reply error %d", error);
2840 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2842 struct dlm_rsb *r = lkb->lkb_resource;
2847 __receive_convert_reply(r, lkb, ms);
2853 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
2855 struct dlm_lkb *lkb;
2858 error = find_lkb(ls, ms->m_remid, &lkb);
2860 log_error(ls, "receive_convert_reply no lkb");
2863 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2865 error = remove_from_waiters(lkb);
2867 log_error(ls, "receive_convert_reply not on waiters");
2871 _receive_convert_reply(lkb, ms);
2876 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2878 struct dlm_rsb *r = lkb->lkb_resource;
2879 int error = ms->m_result;
2884 /* this is the value returned from do_unlock() on the master */
2888 receive_flags_reply(lkb, ms);
2889 remove_lock_pc(r, lkb);
2890 queue_cast(r, lkb, -DLM_EUNLOCK);
2893 log_error(r->res_ls, "receive_unlock_reply error %d", error);
2900 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
2902 struct dlm_lkb *lkb;
2905 error = find_lkb(ls, ms->m_remid, &lkb);
2907 log_error(ls, "receive_unlock_reply no lkb");
2910 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2912 error = remove_from_waiters(lkb);
2914 log_error(ls, "receive_unlock_reply not on waiters");
2918 _receive_unlock_reply(lkb, ms);
2923 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
2925 struct dlm_rsb *r = lkb->lkb_resource;
2926 int error = ms->m_result;
2931 /* this is the value returned from do_cancel() on the master */
2935 receive_flags_reply(lkb, ms);
2936 revert_lock_pc(r, lkb);
2937 queue_cast(r, lkb, -DLM_ECANCEL);
2940 log_error(r->res_ls, "receive_cancel_reply error %d", error);
2947 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
2949 struct dlm_lkb *lkb;
2952 error = find_lkb(ls, ms->m_remid, &lkb);
2954 log_error(ls, "receive_cancel_reply no lkb");
2957 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
2959 error = remove_from_waiters(lkb);
2961 log_error(ls, "receive_cancel_reply not on waiters");
2965 _receive_cancel_reply(lkb, ms);
2970 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
2972 struct dlm_lkb *lkb;
2974 int error, ret_nodeid;
2976 error = find_lkb(ls, ms->m_lkid, &lkb);
2978 log_error(ls, "receive_lookup_reply no lkb");
2982 error = remove_from_waiters(lkb);
2984 log_error(ls, "receive_lookup_reply not on waiters");
2988 /* this is the value returned by dlm_dir_lookup on dir node
2989 FIXME: will a non-zero error ever be returned? */
2990 error = ms->m_result;
2992 r = lkb->lkb_resource;
2996 ret_nodeid = ms->m_nodeid;
2997 if (ret_nodeid == dlm_our_nodeid()) {
3000 r->res_first_lkid = 0;
3002 /* set_master() will copy res_nodeid to lkb_nodeid */
3003 r->res_nodeid = ret_nodeid;
3006 _request_lock(r, lkb);
3009 process_lookup_list(r);
3017 int dlm_receive_message(struct dlm_header *hd, int nodeid, int recovery)
3019 struct dlm_message *ms = (struct dlm_message *) hd;
3026 ls = dlm_find_lockspace_global(hd->h_lockspace);
3028 log_print("drop message %d from %d for unknown lockspace %d",
3029 ms->m_type, nodeid, hd->h_lockspace);
3033 /* recovery may have just ended leaving a bunch of backed-up requests
3034 in the requestqueue; wait while dlm_recoverd clears them */
3037 dlm_wait_requestqueue(ls);
3039 /* recovery may have just started while there were a bunch of
3040 in-flight requests -- save them in requestqueue to be processed
3041 after recovery. we can't let dlm_recvd block on the recovery
3042 lock. if dlm_recoverd is calling this function to clear the
3043 requestqueue, it needs to be interrupted (-EINTR) if another
3044 recovery operation is starting. */
3047 if (dlm_locking_stopped(ls)) {
3052 error = dlm_add_requestqueue(ls, nodeid, hd);
3053 if (error == -EAGAIN)
3061 if (lock_recovery_try(ls))
3066 switch (ms->m_type) {
3068 /* messages sent to a master node */
3070 case DLM_MSG_REQUEST:
3071 receive_request(ls, ms);
3074 case DLM_MSG_CONVERT:
3075 receive_convert(ls, ms);
3078 case DLM_MSG_UNLOCK:
3079 receive_unlock(ls, ms);
3082 case DLM_MSG_CANCEL:
3083 receive_cancel(ls, ms);
3086 /* messages sent from a master node (replies to above) */
3088 case DLM_MSG_REQUEST_REPLY:
3089 receive_request_reply(ls, ms);
3092 case DLM_MSG_CONVERT_REPLY:
3093 receive_convert_reply(ls, ms);
3096 case DLM_MSG_UNLOCK_REPLY:
3097 receive_unlock_reply(ls, ms);
3100 case DLM_MSG_CANCEL_REPLY:
3101 receive_cancel_reply(ls, ms);
3104 /* messages sent from a master node (only two types of async msg) */
3107 receive_grant(ls, ms);
3111 receive_bast(ls, ms);
3114 /* messages sent to a dir node */
3116 case DLM_MSG_LOOKUP:
3117 receive_lookup(ls, ms);
3120 case DLM_MSG_REMOVE:
3121 receive_remove(ls, ms);
3124 /* messages sent from a dir node (remove has no reply) */
3126 case DLM_MSG_LOOKUP_REPLY:
3127 receive_lookup_reply(ls, ms);
3131 log_error(ls, "unknown message type %d", ms->m_type);
3134 unlock_recovery(ls);
3136 dlm_put_lockspace(ls);
3146 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb)
3148 if (middle_conversion(lkb)) {
3150 ls->ls_stub_ms.m_result = -EINPROGRESS;
3151 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3152 _remove_from_waiters(lkb);
3153 _receive_convert_reply(lkb, &ls->ls_stub_ms);
3155 /* Same special case as in receive_rcom_lock_args() */
3156 lkb->lkb_grmode = DLM_LOCK_IV;
3157 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
3160 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
3161 lkb->lkb_flags |= DLM_IFL_RESEND;
3164 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3165 conversions are async; there's no reply from the remote master */
3168 /* A waiting lkb needs recovery if the master node has failed, or
3169 the master node is changing (only when no directory is used) */
3171 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb)
3173 if (dlm_is_removed(ls, lkb->lkb_nodeid))
3176 if (!dlm_no_directory(ls))
3179 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid)
3185 /* Recovery for locks that are waiting for replies from nodes that are now
3186 gone. We can just complete unlocks and cancels by faking a reply from the
3187 dead node. Requests and up-conversions we flag to be resent after
3188 recovery. Down-conversions can just be completed with a fake reply like
3189 unlocks. Conversions between PR and CW need special attention. */
3191 void dlm_recover_waiters_pre(struct dlm_ls *ls)
3193 struct dlm_lkb *lkb, *safe;
3195 mutex_lock(&ls->ls_waiters_mutex);
3197 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
3198 log_debug(ls, "pre recover waiter lkid %x type %d flags %x",
3199 lkb->lkb_id, lkb->lkb_wait_type, lkb->lkb_flags);
3201 /* all outstanding lookups, regardless of destination will be
3202 resent after recovery is done */
3204 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
3205 lkb->lkb_flags |= DLM_IFL_RESEND;
3209 if (!waiter_needs_recovery(ls, lkb))
3212 switch (lkb->lkb_wait_type) {
3214 case DLM_MSG_REQUEST:
3215 lkb->lkb_flags |= DLM_IFL_RESEND;
3218 case DLM_MSG_CONVERT:
3219 recover_convert_waiter(ls, lkb);
3222 case DLM_MSG_UNLOCK:
3224 ls->ls_stub_ms.m_result = -DLM_EUNLOCK;
3225 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3226 _remove_from_waiters(lkb);
3227 _receive_unlock_reply(lkb, &ls->ls_stub_ms);
3231 case DLM_MSG_CANCEL:
3233 ls->ls_stub_ms.m_result = -DLM_ECANCEL;
3234 ls->ls_stub_ms.m_flags = lkb->lkb_flags;
3235 _remove_from_waiters(lkb);
3236 _receive_cancel_reply(lkb, &ls->ls_stub_ms);
3241 log_error(ls, "invalid lkb wait_type %d",
3242 lkb->lkb_wait_type);
3246 mutex_unlock(&ls->ls_waiters_mutex);
3249 static int remove_resend_waiter(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
3251 struct dlm_lkb *lkb;
3254 mutex_lock(&ls->ls_waiters_mutex);
3255 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) {
3256 if (lkb->lkb_flags & DLM_IFL_RESEND) {
3257 rv = lkb->lkb_wait_type;
3258 _remove_from_waiters(lkb);
3259 lkb->lkb_flags &= ~DLM_IFL_RESEND;
3263 mutex_unlock(&ls->ls_waiters_mutex);
3271 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
3272 master or dir-node for r. Processing the lkb may result in it being placed
3275 int dlm_recover_waiters_post(struct dlm_ls *ls)
3277 struct dlm_lkb *lkb;
3279 int error = 0, mstype;
3282 if (dlm_locking_stopped(ls)) {
3283 log_debug(ls, "recover_waiters_post aborted");
3288 mstype = remove_resend_waiter(ls, &lkb);
3292 r = lkb->lkb_resource;
3294 log_debug(ls, "recover_waiters_post %x type %d flags %x %s",
3295 lkb->lkb_id, mstype, lkb->lkb_flags, r->res_name);
3299 case DLM_MSG_LOOKUP:
3302 _request_lock(r, lkb);
3304 confirm_master(r, 0);
3309 case DLM_MSG_REQUEST:
3312 _request_lock(r, lkb);
3314 confirm_master(r, 0);
3319 case DLM_MSG_CONVERT:
3322 _convert_lock(r, lkb);
3328 log_error(ls, "recover_waiters_post type %d", mstype);
3335 static void purge_queue(struct dlm_rsb *r, struct list_head *queue,
3336 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb))
3338 struct dlm_ls *ls = r->res_ls;
3339 struct dlm_lkb *lkb, *safe;
3341 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) {
3342 if (test(ls, lkb)) {
3343 rsb_set_flag(r, RSB_LOCKS_PURGED);
3345 /* this put should free the lkb */
3346 if (!dlm_put_lkb(lkb))
3347 log_error(ls, "purged lkb not released");
3352 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3354 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid));
3357 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb)
3359 return is_master_copy(lkb);
3362 static void purge_dead_locks(struct dlm_rsb *r)
3364 purge_queue(r, &r->res_grantqueue, &purge_dead_test);
3365 purge_queue(r, &r->res_convertqueue, &purge_dead_test);
3366 purge_queue(r, &r->res_waitqueue, &purge_dead_test);
3369 void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
3371 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test);
3372 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test);
3373 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test);
3376 /* Get rid of locks held by nodes that are gone. */
3378 int dlm_purge_locks(struct dlm_ls *ls)
3382 log_debug(ls, "dlm_purge_locks");
3384 down_write(&ls->ls_root_sem);
3385 list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
3389 purge_dead_locks(r);
3395 up_write(&ls->ls_root_sem);
3400 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket)
3402 struct dlm_rsb *r, *r_ret = NULL;
3404 read_lock(&ls->ls_rsbtbl[bucket].lock);
3405 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) {
3406 if (!rsb_flag(r, RSB_LOCKS_PURGED))
3409 rsb_clear_flag(r, RSB_LOCKS_PURGED);
3413 read_unlock(&ls->ls_rsbtbl[bucket].lock);
3417 void dlm_grant_after_purge(struct dlm_ls *ls)
3423 r = find_purged_rsb(ls, bucket);
3425 if (bucket == ls->ls_rsbtbl_size - 1)
3432 grant_pending_locks(r);
3433 confirm_master(r, 0);
3441 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
3444 struct dlm_lkb *lkb;
3446 list_for_each_entry(lkb, head, lkb_statequeue) {
3447 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
3453 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
3456 struct dlm_lkb *lkb;
3458 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
3461 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
3464 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
3470 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3471 struct dlm_rsb *r, struct dlm_rcom *rc)
3473 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3476 lkb->lkb_nodeid = rc->rc_header.h_nodeid;
3477 lkb->lkb_ownpid = rl->rl_ownpid;
3478 lkb->lkb_remid = rl->rl_lkid;
3479 lkb->lkb_exflags = rl->rl_exflags;
3480 lkb->lkb_flags = rl->rl_flags & 0x0000FFFF;
3481 lkb->lkb_flags |= DLM_IFL_MSTCPY;
3482 lkb->lkb_lvbseq = rl->rl_lvbseq;
3483 lkb->lkb_rqmode = rl->rl_rqmode;
3484 lkb->lkb_grmode = rl->rl_grmode;
3485 /* don't set lkb_status because add_lkb wants to itself */
3487 lkb->lkb_bastaddr = (void *) (long) (rl->rl_asts & AST_BAST);
3488 lkb->lkb_astaddr = (void *) (long) (rl->rl_asts & AST_COMP);
3490 if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3491 lkb->lkb_lvbptr = allocate_lvb(ls);
3492 if (!lkb->lkb_lvbptr)
3494 lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) -
3495 sizeof(struct rcom_lock);
3496 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
3499 /* Conversions between PR and CW (middle modes) need special handling.
3500 The real granted mode of these converting locks cannot be determined
3501 until all locks have been rebuilt on the rsb (recover_conversion) */
3503 if (rl->rl_wait_type == DLM_MSG_CONVERT && middle_conversion(lkb)) {
3504 rl->rl_status = DLM_LKSTS_CONVERT;
3505 lkb->lkb_grmode = DLM_LOCK_IV;
3506 rsb_set_flag(r, RSB_RECOVER_CONVERT);
3512 /* This lkb may have been recovered in a previous aborted recovery so we need
3513 to check if the rsb already has an lkb with the given remote nodeid/lkid.
3514 If so we just send back a standard reply. If not, we create a new lkb with
3515 the given values and send back our lkid. We send back our lkid by sending
3516 back the rcom_lock struct we got but with the remid field filled in. */
3518 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
3520 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3522 struct dlm_lkb *lkb;
3525 if (rl->rl_parent_lkid) {
3526 error = -EOPNOTSUPP;
3530 error = find_rsb(ls, rl->rl_name, rl->rl_namelen, R_MASTER, &r);
3536 lkb = search_remid(r, rc->rc_header.h_nodeid, rl->rl_lkid);
3542 error = create_lkb(ls, &lkb);
3546 error = receive_rcom_lock_args(ls, lkb, r, rc);
3553 add_lkb(r, lkb, rl->rl_status);
3557 /* this is the new value returned to the lock holder for
3558 saving in its process-copy lkb */
3559 rl->rl_remid = lkb->lkb_id;
3566 log_print("recover_master_copy %d %x", error, rl->rl_lkid);
3567 rl->rl_result = error;
3571 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
3573 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
3575 struct dlm_lkb *lkb;
3578 error = find_lkb(ls, rl->rl_lkid, &lkb);
3580 log_error(ls, "recover_process_copy no lkid %x", rl->rl_lkid);
3584 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb););
3586 error = rl->rl_result;
3588 r = lkb->lkb_resource;
3594 /* There's a chance the new master received our lock before
3595 dlm_recover_master_reply(), this wouldn't happen if we did
3596 a barrier between recover_masters and recover_locks. */
3597 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id,
3598 (unsigned long)r, r->res_name);
3599 dlm_send_rcom_lock(r, lkb);
3602 log_debug(ls, "master copy exists %x", lkb->lkb_id);
3605 lkb->lkb_remid = rl->rl_remid;
3608 log_error(ls, "dlm_recover_process_copy unknown error %d %x",
3609 error, lkb->lkb_id);
3612 /* an ack for dlm_recover_locks() which waits for replies from
3613 all the locks it sends to new masters */
3614 dlm_recovered_lock(r);
3623 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
3624 int mode, uint32_t flags, void *name, unsigned int namelen,
3625 uint32_t parent_lkid)
3627 struct dlm_lkb *lkb;
3628 struct dlm_args args;
3633 error = create_lkb(ls, &lkb);
3639 if (flags & DLM_LKF_VALBLK) {
3640 ua->lksb.sb_lvbptr = kmalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
3641 if (!ua->lksb.sb_lvbptr) {
3649 /* After ua is attached to lkb it will be freed by free_lkb().
3650 When DLM_IFL_USER is set, the dlm knows that this is a userspace
3651 lock and that lkb_astparam is the dlm_user_args structure. */
3653 error = set_lock_args(mode, &ua->lksb, flags, namelen, parent_lkid,
3654 DLM_FAKE_USER_AST, ua, DLM_FAKE_USER_AST, &args);
3655 lkb->lkb_flags |= DLM_IFL_USER;
3656 ua->old_mode = DLM_LOCK_IV;
3663 error = request_lock(ls, lkb, name, namelen, &args);
3679 /* add this new lkb to the per-process list of locks */
3680 spin_lock(&ua->proc->locks_spin);
3681 kref_get(&lkb->lkb_ref);
3682 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
3683 spin_unlock(&ua->proc->locks_spin);
3685 unlock_recovery(ls);
3689 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3690 int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
3692 struct dlm_lkb *lkb;
3693 struct dlm_args args;
3694 struct dlm_user_args *ua;
3699 error = find_lkb(ls, lkid, &lkb);
3703 /* user can change the params on its lock when it converts it, or
3704 add an lvb that didn't exist before */
3706 ua = (struct dlm_user_args *)lkb->lkb_astparam;
3708 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
3709 ua->lksb.sb_lvbptr = kmalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
3710 if (!ua->lksb.sb_lvbptr) {
3715 if (lvb_in && ua->lksb.sb_lvbptr)
3716 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
3718 ua->castparam = ua_tmp->castparam;
3719 ua->castaddr = ua_tmp->castaddr;
3720 ua->bastparam = ua_tmp->bastparam;
3721 ua->bastaddr = ua_tmp->bastaddr;
3722 ua->user_lksb = ua_tmp->user_lksb;
3723 ua->old_mode = lkb->lkb_grmode;
3725 error = set_lock_args(mode, &ua->lksb, flags, 0, 0, DLM_FAKE_USER_AST,
3726 ua, DLM_FAKE_USER_AST, &args);
3730 error = convert_lock(ls, lkb, &args);
3732 if (error == -EINPROGRESS || error == -EAGAIN)
3737 unlock_recovery(ls);
3742 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3743 uint32_t flags, uint32_t lkid, char *lvb_in)
3745 struct dlm_lkb *lkb;
3746 struct dlm_args args;
3747 struct dlm_user_args *ua;
3752 error = find_lkb(ls, lkid, &lkb);
3756 ua = (struct dlm_user_args *)lkb->lkb_astparam;
3758 if (lvb_in && ua->lksb.sb_lvbptr)
3759 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
3760 ua->castparam = ua_tmp->castparam;
3761 ua->user_lksb = ua_tmp->user_lksb;
3763 error = set_unlock_args(flags, ua, &args);
3767 error = unlock_lock(ls, lkb, &args);
3769 if (error == -DLM_EUNLOCK)
3774 spin_lock(&ua->proc->locks_spin);
3775 /* dlm_user_add_ast() may have already taken lkb off the proc list */
3776 if (!list_empty(&lkb->lkb_ownqueue))
3777 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
3778 spin_unlock(&ua->proc->locks_spin);
3782 unlock_recovery(ls);
3786 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
3787 uint32_t flags, uint32_t lkid)
3789 struct dlm_lkb *lkb;
3790 struct dlm_args args;
3791 struct dlm_user_args *ua;
3796 error = find_lkb(ls, lkid, &lkb);
3800 ua = (struct dlm_user_args *)lkb->lkb_astparam;
3801 ua->castparam = ua_tmp->castparam;
3802 ua->user_lksb = ua_tmp->user_lksb;
3804 error = set_unlock_args(flags, ua, &args);
3808 error = cancel_lock(ls, lkb, &args);
3810 if (error == -DLM_ECANCEL)
3815 /* this lkb was removed from the WAITING queue */
3816 if (lkb->lkb_grmode == DLM_LOCK_IV) {
3817 spin_lock(&ua->proc->locks_spin);
3818 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
3819 spin_unlock(&ua->proc->locks_spin);
3824 unlock_recovery(ls);
3828 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
3830 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
3832 if (ua->lksb.sb_lvbptr)
3833 kfree(ua->lksb.sb_lvbptr);
3835 lkb->lkb_astparam = (long)NULL;
3837 /* TODO: propogate to master if needed */
3841 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
3842 Regardless of what rsb queue the lock is on, it's removed and freed. */
3844 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
3846 struct dlm_user_args *ua = (struct dlm_user_args *)lkb->lkb_astparam;
3847 struct dlm_args args;
3850 /* FIXME: we need to handle the case where the lkb is in limbo
3851 while the rsb is being looked up, currently we assert in
3852 _unlock_lock/is_remote because rsb nodeid is -1. */
3854 set_unlock_args(DLM_LKF_FORCEUNLOCK, ua, &args);
3856 error = unlock_lock(ls, lkb, &args);
3857 if (error == -DLM_EUNLOCK)
3862 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
3863 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
3864 which we clear here. */
3866 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
3867 list, and no more device_writes should add lkb's to proc->locks list; so we
3868 shouldn't need to take asts_spin or locks_spin here. this assumes that
3869 device reads/writes/closes are serialized -- FIXME: we may need to serialize
3872 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
3874 struct dlm_lkb *lkb, *safe;
3877 mutex_lock(&ls->ls_clear_proc_locks);
3879 list_for_each_entry_safe(lkb, safe, &proc->locks, lkb_ownqueue) {
3880 list_del_init(&lkb->lkb_ownqueue);
3882 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT) {
3883 lkb->lkb_flags |= DLM_IFL_ORPHAN;
3884 orphan_proc_lock(ls, lkb);
3886 lkb->lkb_flags |= DLM_IFL_DEAD;
3887 unlock_proc_lock(ls, lkb);
3890 /* this removes the reference for the proc->locks list
3891 added by dlm_user_request, it may result in the lkb
3897 /* in-progress unlocks */
3898 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
3899 list_del_init(&lkb->lkb_ownqueue);
3900 lkb->lkb_flags |= DLM_IFL_DEAD;
3904 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_astqueue) {
3905 list_del(&lkb->lkb_astqueue);
3909 mutex_unlock(&ls->ls_clear_proc_locks);
3910 unlock_recovery(ls);