]>
git.proxmox.com Git - mirror_ubuntu-bionic-kernel.git/blob - fs/dlm/lock.c
b865a46059ddd27aac86dc05eec54c62865e00f6
1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005-2007 Red Hat, Inc. All rights reserved.
6 ** This copyrighted material is made available to anyone wishing to use,
7 ** modify, copy, or redistribute it subject to the terms and conditions
8 ** of the GNU General Public License v.2.
10 *******************************************************************************
11 ******************************************************************************/
13 /* Central locking logic has four stages:
33 Stage 1 (lock, unlock) is mainly about checking input args and
34 splitting into one of the four main operations:
36 dlm_lock = request_lock
37 dlm_lock+CONVERT = convert_lock
38 dlm_unlock = unlock_lock
39 dlm_unlock+CANCEL = cancel_lock
41 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
42 provided to the next stage.
44 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
45 When remote, it calls send_xxxx(), when local it calls do_xxxx().
47 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
48 given rsb and lkb and queues callbacks.
50 For remote operations, send_xxxx() results in the corresponding do_xxxx()
51 function being executed on the remote node. The connecting send/receive
52 calls on local (L) and remote (R) nodes:
54 L: send_xxxx() -> R: receive_xxxx()
56 L: receive_xxxx_reply() <- R: send_xxxx_reply()
58 #include <linux/types.h>
59 #include "dlm_internal.h"
60 #include <linux/dlm_device.h>
63 #include "requestqueue.h"
67 #include "lockspace.h"
72 #include "lvb_table.h"
76 static int send_request(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
77 static int send_convert(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
78 static int send_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
79 static int send_cancel(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
80 static int send_grant(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
81 static int send_bast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int mode
);
82 static int send_lookup(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
83 static int send_remove(struct dlm_rsb
*r
);
84 static int _request_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
85 static 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 inline int is_overlap_unlock(struct dlm_lkb
*lkb
)
259 return lkb
->lkb_flags
& DLM_IFL_OVERLAP_UNLOCK
;
262 static inline int is_overlap_cancel(struct dlm_lkb
*lkb
)
264 return lkb
->lkb_flags
& DLM_IFL_OVERLAP_CANCEL
;
267 static inline int is_overlap(struct dlm_lkb
*lkb
)
269 return (lkb
->lkb_flags
& (DLM_IFL_OVERLAP_UNLOCK
|
270 DLM_IFL_OVERLAP_CANCEL
));
273 static void queue_cast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
275 if (is_master_copy(lkb
))
278 DLM_ASSERT(lkb
->lkb_lksb
, dlm_print_lkb(lkb
););
280 lkb
->lkb_lksb
->sb_status
= rv
;
281 lkb
->lkb_lksb
->sb_flags
= lkb
->lkb_sbflags
;
283 dlm_add_ast(lkb
, AST_COMP
);
286 static inline void queue_cast_overlap(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
289 is_overlap_unlock(lkb
) ? -DLM_EUNLOCK
: -DLM_ECANCEL
);
292 static void queue_bast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rqmode
)
294 if (is_master_copy(lkb
))
295 send_bast(r
, lkb
, rqmode
);
297 lkb
->lkb_bastmode
= rqmode
;
298 dlm_add_ast(lkb
, AST_BAST
);
303 * Basic operations on rsb's and lkb's
306 static struct dlm_rsb
*create_rsb(struct dlm_ls
*ls
, char *name
, int len
)
310 r
= allocate_rsb(ls
, len
);
316 memcpy(r
->res_name
, name
, len
);
317 mutex_init(&r
->res_mutex
);
319 INIT_LIST_HEAD(&r
->res_lookup
);
320 INIT_LIST_HEAD(&r
->res_grantqueue
);
321 INIT_LIST_HEAD(&r
->res_convertqueue
);
322 INIT_LIST_HEAD(&r
->res_waitqueue
);
323 INIT_LIST_HEAD(&r
->res_root_list
);
324 INIT_LIST_HEAD(&r
->res_recover_list
);
329 static int search_rsb_list(struct list_head
*head
, char *name
, int len
,
330 unsigned int flags
, struct dlm_rsb
**r_ret
)
335 list_for_each_entry(r
, head
, res_hashchain
) {
336 if (len
== r
->res_length
&& !memcmp(name
, r
->res_name
, len
))
342 if (r
->res_nodeid
&& (flags
& R_MASTER
))
348 static int _search_rsb(struct dlm_ls
*ls
, char *name
, int len
, int b
,
349 unsigned int flags
, struct dlm_rsb
**r_ret
)
354 error
= search_rsb_list(&ls
->ls_rsbtbl
[b
].list
, name
, len
, flags
, &r
);
356 kref_get(&r
->res_ref
);
359 error
= search_rsb_list(&ls
->ls_rsbtbl
[b
].toss
, name
, len
, flags
, &r
);
363 list_move(&r
->res_hashchain
, &ls
->ls_rsbtbl
[b
].list
);
365 if (dlm_no_directory(ls
))
368 if (r
->res_nodeid
== -1) {
369 rsb_clear_flag(r
, RSB_MASTER_UNCERTAIN
);
370 r
->res_first_lkid
= 0;
371 } else if (r
->res_nodeid
> 0) {
372 rsb_set_flag(r
, RSB_MASTER_UNCERTAIN
);
373 r
->res_first_lkid
= 0;
375 DLM_ASSERT(r
->res_nodeid
== 0, dlm_print_rsb(r
););
376 DLM_ASSERT(!rsb_flag(r
, RSB_MASTER_UNCERTAIN
),);
383 static int search_rsb(struct dlm_ls
*ls
, char *name
, int len
, int b
,
384 unsigned int flags
, struct dlm_rsb
**r_ret
)
387 write_lock(&ls
->ls_rsbtbl
[b
].lock
);
388 error
= _search_rsb(ls
, name
, len
, b
, flags
, r_ret
);
389 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
394 * Find rsb in rsbtbl and potentially create/add one
396 * Delaying the release of rsb's has a similar benefit to applications keeping
397 * NL locks on an rsb, but without the guarantee that the cached master value
398 * will still be valid when the rsb is reused. Apps aren't always smart enough
399 * to keep NL locks on an rsb that they may lock again shortly; this can lead
400 * to excessive master lookups and removals if we don't delay the release.
402 * Searching for an rsb means looking through both the normal list and toss
403 * list. When found on the toss list the rsb is moved to the normal list with
404 * ref count of 1; when found on normal list the ref count is incremented.
407 static int find_rsb(struct dlm_ls
*ls
, char *name
, int namelen
,
408 unsigned int flags
, struct dlm_rsb
**r_ret
)
410 struct dlm_rsb
*r
, *tmp
;
411 uint32_t hash
, bucket
;
414 if (dlm_no_directory(ls
))
417 hash
= jhash(name
, namelen
, 0);
418 bucket
= hash
& (ls
->ls_rsbtbl_size
- 1);
420 error
= search_rsb(ls
, name
, namelen
, bucket
, flags
, &r
);
424 if (error
== -EBADR
&& !(flags
& R_CREATE
))
427 /* the rsb was found but wasn't a master copy */
428 if (error
== -ENOTBLK
)
432 r
= create_rsb(ls
, name
, namelen
);
437 r
->res_bucket
= bucket
;
439 kref_init(&r
->res_ref
);
441 /* With no directory, the master can be set immediately */
442 if (dlm_no_directory(ls
)) {
443 int nodeid
= dlm_dir_nodeid(r
);
444 if (nodeid
== dlm_our_nodeid())
446 r
->res_nodeid
= nodeid
;
449 write_lock(&ls
->ls_rsbtbl
[bucket
].lock
);
450 error
= _search_rsb(ls
, name
, namelen
, bucket
, 0, &tmp
);
452 write_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
457 list_add(&r
->res_hashchain
, &ls
->ls_rsbtbl
[bucket
].list
);
458 write_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
465 int dlm_find_rsb(struct dlm_ls
*ls
, char *name
, int namelen
,
466 unsigned int flags
, struct dlm_rsb
**r_ret
)
468 return find_rsb(ls
, name
, namelen
, flags
, r_ret
);
471 /* This is only called to add a reference when the code already holds
472 a valid reference to the rsb, so there's no need for locking. */
474 static inline void hold_rsb(struct dlm_rsb
*r
)
476 kref_get(&r
->res_ref
);
479 void dlm_hold_rsb(struct dlm_rsb
*r
)
484 static void toss_rsb(struct kref
*kref
)
486 struct dlm_rsb
*r
= container_of(kref
, struct dlm_rsb
, res_ref
);
487 struct dlm_ls
*ls
= r
->res_ls
;
489 DLM_ASSERT(list_empty(&r
->res_root_list
), dlm_print_rsb(r
););
490 kref_init(&r
->res_ref
);
491 list_move(&r
->res_hashchain
, &ls
->ls_rsbtbl
[r
->res_bucket
].toss
);
492 r
->res_toss_time
= jiffies
;
494 free_lvb(r
->res_lvbptr
);
495 r
->res_lvbptr
= NULL
;
499 /* When all references to the rsb are gone it's transfered to
500 the tossed list for later disposal. */
502 static void put_rsb(struct dlm_rsb
*r
)
504 struct dlm_ls
*ls
= r
->res_ls
;
505 uint32_t bucket
= r
->res_bucket
;
507 write_lock(&ls
->ls_rsbtbl
[bucket
].lock
);
508 kref_put(&r
->res_ref
, toss_rsb
);
509 write_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
512 void dlm_put_rsb(struct dlm_rsb
*r
)
517 /* See comment for unhold_lkb */
519 static void unhold_rsb(struct dlm_rsb
*r
)
522 rv
= kref_put(&r
->res_ref
, toss_rsb
);
523 DLM_ASSERT(!rv
, dlm_dump_rsb(r
););
526 static void kill_rsb(struct kref
*kref
)
528 struct dlm_rsb
*r
= container_of(kref
, struct dlm_rsb
, res_ref
);
530 /* All work is done after the return from kref_put() so we
531 can release the write_lock before the remove and free. */
533 DLM_ASSERT(list_empty(&r
->res_lookup
), dlm_dump_rsb(r
););
534 DLM_ASSERT(list_empty(&r
->res_grantqueue
), dlm_dump_rsb(r
););
535 DLM_ASSERT(list_empty(&r
->res_convertqueue
), dlm_dump_rsb(r
););
536 DLM_ASSERT(list_empty(&r
->res_waitqueue
), dlm_dump_rsb(r
););
537 DLM_ASSERT(list_empty(&r
->res_root_list
), dlm_dump_rsb(r
););
538 DLM_ASSERT(list_empty(&r
->res_recover_list
), dlm_dump_rsb(r
););
541 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
542 The rsb must exist as long as any lkb's for it do. */
544 static void attach_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
547 lkb
->lkb_resource
= r
;
550 static void detach_lkb(struct dlm_lkb
*lkb
)
552 if (lkb
->lkb_resource
) {
553 put_rsb(lkb
->lkb_resource
);
554 lkb
->lkb_resource
= NULL
;
558 static int create_lkb(struct dlm_ls
*ls
, struct dlm_lkb
**lkb_ret
)
560 struct dlm_lkb
*lkb
, *tmp
;
564 lkb
= allocate_lkb(ls
);
568 lkb
->lkb_nodeid
= -1;
569 lkb
->lkb_grmode
= DLM_LOCK_IV
;
570 kref_init(&lkb
->lkb_ref
);
571 INIT_LIST_HEAD(&lkb
->lkb_ownqueue
);
572 INIT_LIST_HEAD(&lkb
->lkb_rsb_lookup
);
574 get_random_bytes(&bucket
, sizeof(bucket
));
575 bucket
&= (ls
->ls_lkbtbl_size
- 1);
577 write_lock(&ls
->ls_lkbtbl
[bucket
].lock
);
579 /* counter can roll over so we must verify lkid is not in use */
582 lkid
= bucket
| (ls
->ls_lkbtbl
[bucket
].counter
++ << 16);
584 list_for_each_entry(tmp
, &ls
->ls_lkbtbl
[bucket
].list
,
586 if (tmp
->lkb_id
!= lkid
)
594 list_add(&lkb
->lkb_idtbl_list
, &ls
->ls_lkbtbl
[bucket
].list
);
595 write_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
601 static struct dlm_lkb
*__find_lkb(struct dlm_ls
*ls
, uint32_t lkid
)
603 uint16_t bucket
= lkid
& 0xFFFF;
606 list_for_each_entry(lkb
, &ls
->ls_lkbtbl
[bucket
].list
, lkb_idtbl_list
) {
607 if (lkb
->lkb_id
== lkid
)
613 static int find_lkb(struct dlm_ls
*ls
, uint32_t lkid
, struct dlm_lkb
**lkb_ret
)
616 uint16_t bucket
= lkid
& 0xFFFF;
618 if (bucket
>= ls
->ls_lkbtbl_size
)
621 read_lock(&ls
->ls_lkbtbl
[bucket
].lock
);
622 lkb
= __find_lkb(ls
, lkid
);
624 kref_get(&lkb
->lkb_ref
);
625 read_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
628 return lkb
? 0 : -ENOENT
;
631 static void kill_lkb(struct kref
*kref
)
633 struct dlm_lkb
*lkb
= container_of(kref
, struct dlm_lkb
, lkb_ref
);
635 /* All work is done after the return from kref_put() so we
636 can release the write_lock before the detach_lkb */
638 DLM_ASSERT(!lkb
->lkb_status
, dlm_print_lkb(lkb
););
641 /* __put_lkb() is used when an lkb may not have an rsb attached to
642 it so we need to provide the lockspace explicitly */
644 static int __put_lkb(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
646 uint16_t bucket
= lkb
->lkb_id
& 0xFFFF;
648 write_lock(&ls
->ls_lkbtbl
[bucket
].lock
);
649 if (kref_put(&lkb
->lkb_ref
, kill_lkb
)) {
650 list_del(&lkb
->lkb_idtbl_list
);
651 write_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
655 /* for local/process lkbs, lvbptr points to caller's lksb */
656 if (lkb
->lkb_lvbptr
&& is_master_copy(lkb
))
657 free_lvb(lkb
->lkb_lvbptr
);
661 write_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
666 int dlm_put_lkb(struct dlm_lkb
*lkb
)
670 DLM_ASSERT(lkb
->lkb_resource
, dlm_print_lkb(lkb
););
671 DLM_ASSERT(lkb
->lkb_resource
->res_ls
, dlm_print_lkb(lkb
););
673 ls
= lkb
->lkb_resource
->res_ls
;
674 return __put_lkb(ls
, lkb
);
677 /* This is only called to add a reference when the code already holds
678 a valid reference to the lkb, so there's no need for locking. */
680 static inline void hold_lkb(struct dlm_lkb
*lkb
)
682 kref_get(&lkb
->lkb_ref
);
685 /* This is called when we need to remove a reference and are certain
686 it's not the last ref. e.g. del_lkb is always called between a
687 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
688 put_lkb would work fine, but would involve unnecessary locking */
690 static inline void unhold_lkb(struct dlm_lkb
*lkb
)
693 rv
= kref_put(&lkb
->lkb_ref
, kill_lkb
);
694 DLM_ASSERT(!rv
, dlm_print_lkb(lkb
););
697 static void lkb_add_ordered(struct list_head
*new, struct list_head
*head
,
700 struct dlm_lkb
*lkb
= NULL
;
702 list_for_each_entry(lkb
, head
, lkb_statequeue
)
703 if (lkb
->lkb_rqmode
< mode
)
707 list_add_tail(new, head
);
709 __list_add(new, lkb
->lkb_statequeue
.prev
, &lkb
->lkb_statequeue
);
712 /* add/remove lkb to rsb's grant/convert/wait queue */
714 static void add_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int status
)
716 kref_get(&lkb
->lkb_ref
);
718 DLM_ASSERT(!lkb
->lkb_status
, dlm_print_lkb(lkb
););
720 lkb
->lkb_status
= status
;
723 case DLM_LKSTS_WAITING
:
724 if (lkb
->lkb_exflags
& DLM_LKF_HEADQUE
)
725 list_add(&lkb
->lkb_statequeue
, &r
->res_waitqueue
);
727 list_add_tail(&lkb
->lkb_statequeue
, &r
->res_waitqueue
);
729 case DLM_LKSTS_GRANTED
:
730 /* convention says granted locks kept in order of grmode */
731 lkb_add_ordered(&lkb
->lkb_statequeue
, &r
->res_grantqueue
,
734 case DLM_LKSTS_CONVERT
:
735 if (lkb
->lkb_exflags
& DLM_LKF_HEADQUE
)
736 list_add(&lkb
->lkb_statequeue
, &r
->res_convertqueue
);
738 list_add_tail(&lkb
->lkb_statequeue
,
739 &r
->res_convertqueue
);
742 DLM_ASSERT(0, dlm_print_lkb(lkb
); printk("sts=%d\n", status
););
746 static void del_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
749 list_del(&lkb
->lkb_statequeue
);
753 static void move_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int sts
)
757 add_lkb(r
, lkb
, sts
);
761 static int msg_reply_type(int mstype
)
764 case DLM_MSG_REQUEST
:
765 return DLM_MSG_REQUEST_REPLY
;
766 case DLM_MSG_CONVERT
:
767 return DLM_MSG_CONVERT_REPLY
;
769 return DLM_MSG_UNLOCK_REPLY
;
771 return DLM_MSG_CANCEL_REPLY
;
773 return DLM_MSG_LOOKUP_REPLY
;
778 /* add/remove lkb from global waiters list of lkb's waiting for
779 a reply from a remote node */
781 static int add_to_waiters(struct dlm_lkb
*lkb
, int mstype
)
783 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
786 mutex_lock(&ls
->ls_waiters_mutex
);
788 if (is_overlap_unlock(lkb
) ||
789 (is_overlap_cancel(lkb
) && (mstype
== DLM_MSG_CANCEL
))) {
794 if (lkb
->lkb_wait_type
|| is_overlap_cancel(lkb
)) {
797 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_UNLOCK
;
800 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_CANCEL
;
806 lkb
->lkb_wait_count
++;
809 log_debug(ls
, "add overlap %x cur %d new %d count %d flags %x",
810 lkb
->lkb_id
, lkb
->lkb_wait_type
, mstype
,
811 lkb
->lkb_wait_count
, lkb
->lkb_flags
);
815 DLM_ASSERT(!lkb
->lkb_wait_count
,
817 printk("wait_count %d\n", lkb
->lkb_wait_count
););
819 lkb
->lkb_wait_count
++;
820 lkb
->lkb_wait_type
= mstype
;
822 list_add(&lkb
->lkb_wait_reply
, &ls
->ls_waiters
);
825 log_error(ls
, "add_to_waiters %x error %d flags %x %d %d %s",
826 lkb
->lkb_id
, error
, lkb
->lkb_flags
, mstype
,
827 lkb
->lkb_wait_type
, lkb
->lkb_resource
->res_name
);
828 mutex_unlock(&ls
->ls_waiters_mutex
);
832 /* We clear the RESEND flag because we might be taking an lkb off the waiters
833 list as part of process_requestqueue (e.g. a lookup that has an optimized
834 request reply on the requestqueue) between dlm_recover_waiters_pre() which
835 set RESEND and dlm_recover_waiters_post() */
837 static int _remove_from_waiters(struct dlm_lkb
*lkb
, int mstype
)
839 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
840 int overlap_done
= 0;
842 if (is_overlap_unlock(lkb
) && (mstype
== DLM_MSG_UNLOCK_REPLY
)) {
843 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
848 if (is_overlap_cancel(lkb
) && (mstype
== DLM_MSG_CANCEL_REPLY
)) {
849 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
854 /* N.B. type of reply may not always correspond to type of original
855 msg due to lookup->request optimization, verify others? */
857 if (lkb
->lkb_wait_type
) {
858 lkb
->lkb_wait_type
= 0;
862 log_error(ls
, "remove_from_waiters lkid %x flags %x types %d %d",
863 lkb
->lkb_id
, lkb
->lkb_flags
, mstype
, lkb
->lkb_wait_type
);
867 /* the force-unlock/cancel has completed and we haven't recvd a reply
868 to the op that was in progress prior to the unlock/cancel; we
869 give up on any reply to the earlier op. FIXME: not sure when/how
872 if (overlap_done
&& lkb
->lkb_wait_type
) {
873 log_error(ls
, "remove_from_waiters %x reply %d give up on %d",
874 lkb
->lkb_id
, mstype
, lkb
->lkb_wait_type
);
875 lkb
->lkb_wait_count
--;
876 lkb
->lkb_wait_type
= 0;
879 DLM_ASSERT(lkb
->lkb_wait_count
, dlm_print_lkb(lkb
););
881 lkb
->lkb_flags
&= ~DLM_IFL_RESEND
;
882 lkb
->lkb_wait_count
--;
883 if (!lkb
->lkb_wait_count
)
884 list_del_init(&lkb
->lkb_wait_reply
);
889 static int remove_from_waiters(struct dlm_lkb
*lkb
, int mstype
)
891 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
894 mutex_lock(&ls
->ls_waiters_mutex
);
895 error
= _remove_from_waiters(lkb
, mstype
);
896 mutex_unlock(&ls
->ls_waiters_mutex
);
900 /* Handles situations where we might be processing a "fake" or "stub" reply in
901 which we can't try to take waiters_mutex again. */
903 static int remove_from_waiters_ms(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
905 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
908 if (ms
!= &ls
->ls_stub_ms
)
909 mutex_lock(&ls
->ls_waiters_mutex
);
910 error
= _remove_from_waiters(lkb
, ms
->m_type
);
911 if (ms
!= &ls
->ls_stub_ms
)
912 mutex_unlock(&ls
->ls_waiters_mutex
);
916 static void dir_remove(struct dlm_rsb
*r
)
920 if (dlm_no_directory(r
->res_ls
))
923 to_nodeid
= dlm_dir_nodeid(r
);
924 if (to_nodeid
!= dlm_our_nodeid())
927 dlm_dir_remove_entry(r
->res_ls
, to_nodeid
,
928 r
->res_name
, r
->res_length
);
931 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
932 found since they are in order of newest to oldest? */
934 static int shrink_bucket(struct dlm_ls
*ls
, int b
)
937 int count
= 0, found
;
941 write_lock(&ls
->ls_rsbtbl
[b
].lock
);
942 list_for_each_entry_reverse(r
, &ls
->ls_rsbtbl
[b
].toss
,
944 if (!time_after_eq(jiffies
, r
->res_toss_time
+
945 dlm_config
.ci_toss_secs
* HZ
))
952 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
956 if (kref_put(&r
->res_ref
, kill_rsb
)) {
957 list_del(&r
->res_hashchain
);
958 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
965 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
966 log_error(ls
, "tossed rsb in use %s", r
->res_name
);
973 void dlm_scan_rsbs(struct dlm_ls
*ls
)
977 if (dlm_locking_stopped(ls
))
980 for (i
= 0; i
< ls
->ls_rsbtbl_size
; i
++) {
981 shrink_bucket(ls
, i
);
986 /* lkb is master or local copy */
988 static void set_lvb_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
990 int b
, len
= r
->res_ls
->ls_lvblen
;
992 /* b=1 lvb returned to caller
993 b=0 lvb written to rsb or invalidated
996 b
= dlm_lvb_operations
[lkb
->lkb_grmode
+ 1][lkb
->lkb_rqmode
+ 1];
999 if (!lkb
->lkb_lvbptr
)
1002 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1008 memcpy(lkb
->lkb_lvbptr
, r
->res_lvbptr
, len
);
1009 lkb
->lkb_lvbseq
= r
->res_lvbseq
;
1011 } else if (b
== 0) {
1012 if (lkb
->lkb_exflags
& DLM_LKF_IVVALBLK
) {
1013 rsb_set_flag(r
, RSB_VALNOTVALID
);
1017 if (!lkb
->lkb_lvbptr
)
1020 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1024 r
->res_lvbptr
= allocate_lvb(r
->res_ls
);
1029 memcpy(r
->res_lvbptr
, lkb
->lkb_lvbptr
, len
);
1031 lkb
->lkb_lvbseq
= r
->res_lvbseq
;
1032 rsb_clear_flag(r
, RSB_VALNOTVALID
);
1035 if (rsb_flag(r
, RSB_VALNOTVALID
))
1036 lkb
->lkb_sbflags
|= DLM_SBF_VALNOTVALID
;
1039 static void set_lvb_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1041 if (lkb
->lkb_grmode
< DLM_LOCK_PW
)
1044 if (lkb
->lkb_exflags
& DLM_LKF_IVVALBLK
) {
1045 rsb_set_flag(r
, RSB_VALNOTVALID
);
1049 if (!lkb
->lkb_lvbptr
)
1052 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1056 r
->res_lvbptr
= allocate_lvb(r
->res_ls
);
1061 memcpy(r
->res_lvbptr
, lkb
->lkb_lvbptr
, r
->res_ls
->ls_lvblen
);
1063 rsb_clear_flag(r
, RSB_VALNOTVALID
);
1066 /* lkb is process copy (pc) */
1068 static void set_lvb_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
1069 struct dlm_message
*ms
)
1073 if (!lkb
->lkb_lvbptr
)
1076 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1079 b
= dlm_lvb_operations
[lkb
->lkb_grmode
+ 1][lkb
->lkb_rqmode
+ 1];
1081 int len
= receive_extralen(ms
);
1082 memcpy(lkb
->lkb_lvbptr
, ms
->m_extra
, len
);
1083 lkb
->lkb_lvbseq
= ms
->m_lvbseq
;
1087 /* Manipulate lkb's on rsb's convert/granted/waiting queues
1088 remove_lock -- used for unlock, removes lkb from granted
1089 revert_lock -- used for cancel, moves lkb from convert to granted
1090 grant_lock -- used for request and convert, adds lkb to granted or
1091 moves lkb from convert or waiting to granted
1093 Each of these is used for master or local copy lkb's. There is
1094 also a _pc() variation used to make the corresponding change on
1095 a process copy (pc) lkb. */
1097 static void _remove_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1100 lkb
->lkb_grmode
= DLM_LOCK_IV
;
1101 /* this unhold undoes the original ref from create_lkb()
1102 so this leads to the lkb being freed */
1106 static void remove_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1108 set_lvb_unlock(r
, lkb
);
1109 _remove_lock(r
, lkb
);
1112 static void remove_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1114 _remove_lock(r
, lkb
);
1117 /* returns: 0 did nothing
1118 1 moved lock to granted
1121 static int revert_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1125 lkb
->lkb_rqmode
= DLM_LOCK_IV
;
1127 switch (lkb
->lkb_status
) {
1128 case DLM_LKSTS_GRANTED
:
1130 case DLM_LKSTS_CONVERT
:
1131 move_lkb(r
, lkb
, DLM_LKSTS_GRANTED
);
1134 case DLM_LKSTS_WAITING
:
1136 lkb
->lkb_grmode
= DLM_LOCK_IV
;
1137 /* this unhold undoes the original ref from create_lkb()
1138 so this leads to the lkb being freed */
1143 log_print("invalid status for revert %d", lkb
->lkb_status
);
1148 static int revert_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1150 return revert_lock(r
, lkb
);
1153 static void _grant_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1155 if (lkb
->lkb_grmode
!= lkb
->lkb_rqmode
) {
1156 lkb
->lkb_grmode
= lkb
->lkb_rqmode
;
1157 if (lkb
->lkb_status
)
1158 move_lkb(r
, lkb
, DLM_LKSTS_GRANTED
);
1160 add_lkb(r
, lkb
, DLM_LKSTS_GRANTED
);
1163 lkb
->lkb_rqmode
= DLM_LOCK_IV
;
1166 static void grant_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1168 set_lvb_lock(r
, lkb
);
1169 _grant_lock(r
, lkb
);
1170 lkb
->lkb_highbast
= 0;
1173 static void grant_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
1174 struct dlm_message
*ms
)
1176 set_lvb_lock_pc(r
, lkb
, ms
);
1177 _grant_lock(r
, lkb
);
1180 /* called by grant_pending_locks() which means an async grant message must
1181 be sent to the requesting node in addition to granting the lock if the
1182 lkb belongs to a remote node. */
1184 static void grant_lock_pending(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1187 if (is_master_copy(lkb
))
1190 queue_cast(r
, lkb
, 0);
1193 static inline int first_in_list(struct dlm_lkb
*lkb
, struct list_head
*head
)
1195 struct dlm_lkb
*first
= list_entry(head
->next
, struct dlm_lkb
,
1197 if (lkb
->lkb_id
== first
->lkb_id
)
1203 /* Check if the given lkb conflicts with another lkb on the queue. */
1205 static int queue_conflict(struct list_head
*head
, struct dlm_lkb
*lkb
)
1207 struct dlm_lkb
*this;
1209 list_for_each_entry(this, head
, lkb_statequeue
) {
1212 if (!modes_compat(this, lkb
))
1219 * "A conversion deadlock arises with a pair of lock requests in the converting
1220 * queue for one resource. The granted mode of each lock blocks the requested
1221 * mode of the other lock."
1223 * Part 2: if the granted mode of lkb is preventing the first lkb in the
1224 * convert queue from being granted, then demote lkb (set grmode to NL).
1225 * This second form requires that we check for conv-deadlk even when
1226 * now == 0 in _can_be_granted().
1229 * Granted Queue: empty
1230 * Convert Queue: NL->EX (first lock)
1231 * PR->EX (second lock)
1233 * The first lock can't be granted because of the granted mode of the second
1234 * lock and the second lock can't be granted because it's not first in the
1235 * list. We demote the granted mode of the second lock (the lkb passed to this
1238 * After the resolution, the "grant pending" function needs to go back and try
1239 * to grant locks on the convert queue again since the first lock can now be
1243 static int conversion_deadlock_detect(struct dlm_rsb
*rsb
, struct dlm_lkb
*lkb
)
1245 struct dlm_lkb
*this, *first
= NULL
, *self
= NULL
;
1247 list_for_each_entry(this, &rsb
->res_convertqueue
, lkb_statequeue
) {
1255 if (!modes_compat(this, lkb
) && !modes_compat(lkb
, this))
1259 /* if lkb is on the convert queue and is preventing the first
1260 from being granted, then there's deadlock and we demote lkb.
1261 multiple converting locks may need to do this before the first
1262 converting lock can be granted. */
1264 if (self
&& self
!= first
) {
1265 if (!modes_compat(lkb
, first
) &&
1266 !queue_conflict(&rsb
->res_grantqueue
, first
))
1274 * Return 1 if the lock can be granted, 0 otherwise.
1275 * Also detect and resolve conversion deadlocks.
1277 * lkb is the lock to be granted
1279 * now is 1 if the function is being called in the context of the
1280 * immediate request, it is 0 if called later, after the lock has been
1283 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1286 static int _can_be_granted(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int now
)
1288 int8_t conv
= (lkb
->lkb_grmode
!= DLM_LOCK_IV
);
1291 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1292 * a new request for a NL mode lock being blocked.
1294 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1295 * request, then it would be granted. In essence, the use of this flag
1296 * tells the Lock Manager to expedite theis request by not considering
1297 * what may be in the CONVERTING or WAITING queues... As of this
1298 * writing, the EXPEDITE flag can be used only with new requests for NL
1299 * mode locks. This flag is not valid for conversion requests.
1301 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1302 * conversion or used with a non-NL requested mode. We also know an
1303 * EXPEDITE request is always granted immediately, so now must always
1304 * be 1. The full condition to grant an expedite request: (now &&
1305 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1306 * therefore be shortened to just checking the flag.
1309 if (lkb
->lkb_exflags
& DLM_LKF_EXPEDITE
)
1313 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1314 * added to the remaining conditions.
1317 if (queue_conflict(&r
->res_grantqueue
, lkb
))
1321 * 6-3: By default, a conversion request is immediately granted if the
1322 * requested mode is compatible with the modes of all other granted
1326 if (queue_conflict(&r
->res_convertqueue
, lkb
))
1330 * 6-5: But the default algorithm for deciding whether to grant or
1331 * queue conversion requests does not by itself guarantee that such
1332 * requests are serviced on a "first come first serve" basis. This, in
1333 * turn, can lead to a phenomenon known as "indefinate postponement".
1335 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1336 * the system service employed to request a lock conversion. This flag
1337 * forces certain conversion requests to be queued, even if they are
1338 * compatible with the granted modes of other locks on the same
1339 * resource. Thus, the use of this flag results in conversion requests
1340 * being ordered on a "first come first servce" basis.
1342 * DCT: This condition is all about new conversions being able to occur
1343 * "in place" while the lock remains on the granted queue (assuming
1344 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1345 * doesn't _have_ to go onto the convert queue where it's processed in
1346 * order. The "now" variable is necessary to distinguish converts
1347 * being received and processed for the first time now, because once a
1348 * convert is moved to the conversion queue the condition below applies
1349 * requiring fifo granting.
1352 if (now
&& conv
&& !(lkb
->lkb_exflags
& DLM_LKF_QUECVT
))
1356 * The NOORDER flag is set to avoid the standard vms rules on grant
1360 if (lkb
->lkb_exflags
& DLM_LKF_NOORDER
)
1364 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1365 * granted until all other conversion requests ahead of it are granted
1369 if (!now
&& conv
&& first_in_list(lkb
, &r
->res_convertqueue
))
1373 * 6-4: By default, a new request is immediately granted only if all
1374 * three of the following conditions are satisfied when the request is
1376 * - The queue of ungranted conversion requests for the resource is
1378 * - The queue of ungranted new requests for the resource is empty.
1379 * - The mode of the new request is compatible with the most
1380 * restrictive mode of all granted locks on the resource.
1383 if (now
&& !conv
&& list_empty(&r
->res_convertqueue
) &&
1384 list_empty(&r
->res_waitqueue
))
1388 * 6-4: Once a lock request is in the queue of ungranted new requests,
1389 * it cannot be granted until the queue of ungranted conversion
1390 * requests is empty, all ungranted new requests ahead of it are
1391 * granted and/or canceled, and it is compatible with the granted mode
1392 * of the most restrictive lock granted on the resource.
1395 if (!now
&& !conv
&& list_empty(&r
->res_convertqueue
) &&
1396 first_in_list(lkb
, &r
->res_waitqueue
))
1401 * The following, enabled by CONVDEADLK, departs from VMS.
1404 if (conv
&& (lkb
->lkb_exflags
& DLM_LKF_CONVDEADLK
) &&
1405 conversion_deadlock_detect(r
, lkb
)) {
1406 lkb
->lkb_grmode
= DLM_LOCK_NL
;
1407 lkb
->lkb_sbflags
|= DLM_SBF_DEMOTED
;
1414 * The ALTPR and ALTCW flags aren't traditional lock manager flags, but are a
1415 * simple way to provide a big optimization to applications that can use them.
1418 static int can_be_granted(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int now
)
1420 uint32_t flags
= lkb
->lkb_exflags
;
1422 int8_t alt
= 0, rqmode
= lkb
->lkb_rqmode
;
1424 rv
= _can_be_granted(r
, lkb
, now
);
1428 if (lkb
->lkb_sbflags
& DLM_SBF_DEMOTED
)
1431 if (rqmode
!= DLM_LOCK_PR
&& flags
& DLM_LKF_ALTPR
)
1433 else if (rqmode
!= DLM_LOCK_CW
&& flags
& DLM_LKF_ALTCW
)
1437 lkb
->lkb_rqmode
= alt
;
1438 rv
= _can_be_granted(r
, lkb
, now
);
1440 lkb
->lkb_sbflags
|= DLM_SBF_ALTMODE
;
1442 lkb
->lkb_rqmode
= rqmode
;
1448 static int grant_pending_convert(struct dlm_rsb
*r
, int high
)
1450 struct dlm_lkb
*lkb
, *s
;
1451 int hi
, demoted
, quit
, grant_restart
, demote_restart
;
1459 list_for_each_entry_safe(lkb
, s
, &r
->res_convertqueue
, lkb_statequeue
) {
1460 demoted
= is_demoted(lkb
);
1461 if (can_be_granted(r
, lkb
, 0)) {
1462 grant_lock_pending(r
, lkb
);
1465 hi
= max_t(int, lkb
->lkb_rqmode
, hi
);
1466 if (!demoted
&& is_demoted(lkb
))
1473 if (demote_restart
&& !quit
) {
1478 return max_t(int, high
, hi
);
1481 static int grant_pending_wait(struct dlm_rsb
*r
, int high
)
1483 struct dlm_lkb
*lkb
, *s
;
1485 list_for_each_entry_safe(lkb
, s
, &r
->res_waitqueue
, lkb_statequeue
) {
1486 if (can_be_granted(r
, lkb
, 0))
1487 grant_lock_pending(r
, lkb
);
1489 high
= max_t(int, lkb
->lkb_rqmode
, high
);
1495 static void grant_pending_locks(struct dlm_rsb
*r
)
1497 struct dlm_lkb
*lkb
, *s
;
1498 int high
= DLM_LOCK_IV
;
1500 DLM_ASSERT(is_master(r
), dlm_dump_rsb(r
););
1502 high
= grant_pending_convert(r
, high
);
1503 high
= grant_pending_wait(r
, high
);
1505 if (high
== DLM_LOCK_IV
)
1509 * If there are locks left on the wait/convert queue then send blocking
1510 * ASTs to granted locks based on the largest requested mode (high)
1511 * found above. FIXME: highbast < high comparison not valid for PR/CW.
1514 list_for_each_entry_safe(lkb
, s
, &r
->res_grantqueue
, lkb_statequeue
) {
1515 if (lkb
->lkb_bastaddr
&& (lkb
->lkb_highbast
< high
) &&
1516 !__dlm_compat_matrix
[lkb
->lkb_grmode
+1][high
+1]) {
1517 queue_bast(r
, lkb
, high
);
1518 lkb
->lkb_highbast
= high
;
1523 static void send_bast_queue(struct dlm_rsb
*r
, struct list_head
*head
,
1524 struct dlm_lkb
*lkb
)
1528 list_for_each_entry(gr
, head
, lkb_statequeue
) {
1529 if (gr
->lkb_bastaddr
&&
1530 gr
->lkb_highbast
< lkb
->lkb_rqmode
&&
1531 !modes_compat(gr
, lkb
)) {
1532 queue_bast(r
, gr
, lkb
->lkb_rqmode
);
1533 gr
->lkb_highbast
= lkb
->lkb_rqmode
;
1538 static void send_blocking_asts(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1540 send_bast_queue(r
, &r
->res_grantqueue
, lkb
);
1543 static void send_blocking_asts_all(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1545 send_bast_queue(r
, &r
->res_grantqueue
, lkb
);
1546 send_bast_queue(r
, &r
->res_convertqueue
, lkb
);
1549 /* set_master(r, lkb) -- set the master nodeid of a resource
1551 The purpose of this function is to set the nodeid field in the given
1552 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1553 known, it can just be copied to the lkb and the function will return
1554 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1555 before it can be copied to the lkb.
1557 When the rsb nodeid is being looked up remotely, the initial lkb
1558 causing the lookup is kept on the ls_waiters list waiting for the
1559 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1560 on the rsb's res_lookup list until the master is verified.
1563 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1564 1: the rsb master is not available and the lkb has been placed on
1568 static int set_master(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1570 struct dlm_ls
*ls
= r
->res_ls
;
1571 int error
, dir_nodeid
, ret_nodeid
, our_nodeid
= dlm_our_nodeid();
1573 if (rsb_flag(r
, RSB_MASTER_UNCERTAIN
)) {
1574 rsb_clear_flag(r
, RSB_MASTER_UNCERTAIN
);
1575 r
->res_first_lkid
= lkb
->lkb_id
;
1576 lkb
->lkb_nodeid
= r
->res_nodeid
;
1580 if (r
->res_first_lkid
&& r
->res_first_lkid
!= lkb
->lkb_id
) {
1581 list_add_tail(&lkb
->lkb_rsb_lookup
, &r
->res_lookup
);
1585 if (r
->res_nodeid
== 0) {
1586 lkb
->lkb_nodeid
= 0;
1590 if (r
->res_nodeid
> 0) {
1591 lkb
->lkb_nodeid
= r
->res_nodeid
;
1595 DLM_ASSERT(r
->res_nodeid
== -1, dlm_dump_rsb(r
););
1597 dir_nodeid
= dlm_dir_nodeid(r
);
1599 if (dir_nodeid
!= our_nodeid
) {
1600 r
->res_first_lkid
= lkb
->lkb_id
;
1601 send_lookup(r
, lkb
);
1606 /* It's possible for dlm_scand to remove an old rsb for
1607 this same resource from the toss list, us to create
1608 a new one, look up the master locally, and find it
1609 already exists just before dlm_scand does the
1610 dir_remove() on the previous rsb. */
1612 error
= dlm_dir_lookup(ls
, our_nodeid
, r
->res_name
,
1613 r
->res_length
, &ret_nodeid
);
1616 log_debug(ls
, "dir_lookup error %d %s", error
, r
->res_name
);
1620 if (ret_nodeid
== our_nodeid
) {
1621 r
->res_first_lkid
= 0;
1623 lkb
->lkb_nodeid
= 0;
1625 r
->res_first_lkid
= lkb
->lkb_id
;
1626 r
->res_nodeid
= ret_nodeid
;
1627 lkb
->lkb_nodeid
= ret_nodeid
;
1632 static void process_lookup_list(struct dlm_rsb
*r
)
1634 struct dlm_lkb
*lkb
, *safe
;
1636 list_for_each_entry_safe(lkb
, safe
, &r
->res_lookup
, lkb_rsb_lookup
) {
1637 list_del_init(&lkb
->lkb_rsb_lookup
);
1638 _request_lock(r
, lkb
);
1643 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1645 static void confirm_master(struct dlm_rsb
*r
, int error
)
1647 struct dlm_lkb
*lkb
;
1649 if (!r
->res_first_lkid
)
1655 r
->res_first_lkid
= 0;
1656 process_lookup_list(r
);
1660 /* the remote master didn't queue our NOQUEUE request;
1661 make a waiting lkb the first_lkid */
1663 r
->res_first_lkid
= 0;
1665 if (!list_empty(&r
->res_lookup
)) {
1666 lkb
= list_entry(r
->res_lookup
.next
, struct dlm_lkb
,
1668 list_del_init(&lkb
->lkb_rsb_lookup
);
1669 r
->res_first_lkid
= lkb
->lkb_id
;
1670 _request_lock(r
, lkb
);
1676 log_error(r
->res_ls
, "confirm_master unknown error %d", error
);
1680 static int set_lock_args(int mode
, struct dlm_lksb
*lksb
, uint32_t flags
,
1681 int namelen
, uint32_t parent_lkid
, void *ast
,
1682 void *astarg
, void *bast
, struct dlm_args
*args
)
1686 /* check for invalid arg usage */
1688 if (mode
< 0 || mode
> DLM_LOCK_EX
)
1691 if (!(flags
& DLM_LKF_CONVERT
) && (namelen
> DLM_RESNAME_MAXLEN
))
1694 if (flags
& DLM_LKF_CANCEL
)
1697 if (flags
& DLM_LKF_QUECVT
&& !(flags
& DLM_LKF_CONVERT
))
1700 if (flags
& DLM_LKF_CONVDEADLK
&& !(flags
& DLM_LKF_CONVERT
))
1703 if (flags
& DLM_LKF_CONVDEADLK
&& flags
& DLM_LKF_NOQUEUE
)
1706 if (flags
& DLM_LKF_EXPEDITE
&& flags
& DLM_LKF_CONVERT
)
1709 if (flags
& DLM_LKF_EXPEDITE
&& flags
& DLM_LKF_QUECVT
)
1712 if (flags
& DLM_LKF_EXPEDITE
&& flags
& DLM_LKF_NOQUEUE
)
1715 if (flags
& DLM_LKF_EXPEDITE
&& mode
!= DLM_LOCK_NL
)
1721 if (flags
& DLM_LKF_VALBLK
&& !lksb
->sb_lvbptr
)
1724 /* parent/child locks not yet supported */
1728 if (flags
& DLM_LKF_CONVERT
&& !lksb
->sb_lkid
)
1731 /* these args will be copied to the lkb in validate_lock_args,
1732 it cannot be done now because when converting locks, fields in
1733 an active lkb cannot be modified before locking the rsb */
1735 args
->flags
= flags
;
1736 args
->astaddr
= ast
;
1737 args
->astparam
= (long) astarg
;
1738 args
->bastaddr
= bast
;
1746 static int set_unlock_args(uint32_t flags
, void *astarg
, struct dlm_args
*args
)
1748 if (flags
& ~(DLM_LKF_CANCEL
| DLM_LKF_VALBLK
| DLM_LKF_IVVALBLK
|
1749 DLM_LKF_FORCEUNLOCK
))
1752 if (flags
& DLM_LKF_CANCEL
&& flags
& DLM_LKF_FORCEUNLOCK
)
1755 args
->flags
= flags
;
1756 args
->astparam
= (long) astarg
;
1760 static int validate_lock_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
1761 struct dlm_args
*args
)
1765 if (args
->flags
& DLM_LKF_CONVERT
) {
1766 if (lkb
->lkb_flags
& DLM_IFL_MSTCPY
)
1769 if (args
->flags
& DLM_LKF_QUECVT
&&
1770 !__quecvt_compat_matrix
[lkb
->lkb_grmode
+1][args
->mode
+1])
1774 if (lkb
->lkb_status
!= DLM_LKSTS_GRANTED
)
1777 if (lkb
->lkb_wait_type
)
1780 if (is_overlap(lkb
))
1784 lkb
->lkb_exflags
= args
->flags
;
1785 lkb
->lkb_sbflags
= 0;
1786 lkb
->lkb_astaddr
= args
->astaddr
;
1787 lkb
->lkb_astparam
= args
->astparam
;
1788 lkb
->lkb_bastaddr
= args
->bastaddr
;
1789 lkb
->lkb_rqmode
= args
->mode
;
1790 lkb
->lkb_lksb
= args
->lksb
;
1791 lkb
->lkb_lvbptr
= args
->lksb
->sb_lvbptr
;
1792 lkb
->lkb_ownpid
= (int) current
->pid
;
1798 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
1801 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
1802 because there may be a lookup in progress and it's valid to do
1803 cancel/unlockf on it */
1805 static int validate_unlock_args(struct dlm_lkb
*lkb
, struct dlm_args
*args
)
1807 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
1810 if (lkb
->lkb_flags
& DLM_IFL_MSTCPY
) {
1811 log_error(ls
, "unlock on MSTCPY %x", lkb
->lkb_id
);
1816 /* an lkb may still exist even though the lock is EOL'ed due to a
1817 cancel, unlock or failed noqueue request; an app can't use these
1818 locks; return same error as if the lkid had not been found at all */
1820 if (lkb
->lkb_flags
& DLM_IFL_ENDOFLIFE
) {
1821 log_debug(ls
, "unlock on ENDOFLIFE %x", lkb
->lkb_id
);
1826 /* an lkb may be waiting for an rsb lookup to complete where the
1827 lookup was initiated by another lock */
1829 if (args
->flags
& (DLM_LKF_CANCEL
| DLM_LKF_FORCEUNLOCK
)) {
1830 if (!list_empty(&lkb
->lkb_rsb_lookup
)) {
1831 log_debug(ls
, "unlock on rsb_lookup %x", lkb
->lkb_id
);
1832 list_del_init(&lkb
->lkb_rsb_lookup
);
1833 queue_cast(lkb
->lkb_resource
, lkb
,
1834 args
->flags
& DLM_LKF_CANCEL
?
1835 -DLM_ECANCEL
: -DLM_EUNLOCK
);
1836 unhold_lkb(lkb
); /* undoes create_lkb() */
1842 /* cancel not allowed with another cancel/unlock in progress */
1844 if (args
->flags
& DLM_LKF_CANCEL
) {
1845 if (lkb
->lkb_exflags
& DLM_LKF_CANCEL
)
1848 if (is_overlap(lkb
))
1851 if (lkb
->lkb_flags
& DLM_IFL_RESEND
) {
1852 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_CANCEL
;
1857 switch (lkb
->lkb_wait_type
) {
1858 case DLM_MSG_LOOKUP
:
1859 case DLM_MSG_REQUEST
:
1860 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_CANCEL
;
1863 case DLM_MSG_UNLOCK
:
1864 case DLM_MSG_CANCEL
:
1867 /* add_to_waiters() will set OVERLAP_CANCEL */
1871 /* do we need to allow a force-unlock if there's a normal unlock
1872 already in progress? in what conditions could the normal unlock
1873 fail such that we'd want to send a force-unlock to be sure? */
1875 if (args
->flags
& DLM_LKF_FORCEUNLOCK
) {
1876 if (lkb
->lkb_exflags
& DLM_LKF_FORCEUNLOCK
)
1879 if (is_overlap_unlock(lkb
))
1882 if (lkb
->lkb_flags
& DLM_IFL_RESEND
) {
1883 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_UNLOCK
;
1888 switch (lkb
->lkb_wait_type
) {
1889 case DLM_MSG_LOOKUP
:
1890 case DLM_MSG_REQUEST
:
1891 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_UNLOCK
;
1894 case DLM_MSG_UNLOCK
:
1897 /* add_to_waiters() will set OVERLAP_UNLOCK */
1901 /* normal unlock not allowed if there's any op in progress */
1903 if (lkb
->lkb_wait_type
|| lkb
->lkb_wait_count
)
1907 /* an overlapping op shouldn't blow away exflags from other op */
1908 lkb
->lkb_exflags
|= args
->flags
;
1909 lkb
->lkb_sbflags
= 0;
1910 lkb
->lkb_astparam
= args
->astparam
;
1914 log_debug(ls
, "validate_unlock_args %d %x %x %x %x %d %s", rv
,
1915 lkb
->lkb_id
, lkb
->lkb_flags
, lkb
->lkb_exflags
,
1916 args
->flags
, lkb
->lkb_wait_type
,
1917 lkb
->lkb_resource
->res_name
);
1922 * Four stage 4 varieties:
1923 * do_request(), do_convert(), do_unlock(), do_cancel()
1924 * These are called on the master node for the given lock and
1925 * from the central locking logic.
1928 static int do_request(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1932 if (can_be_granted(r
, lkb
, 1)) {
1934 queue_cast(r
, lkb
, 0);
1938 if (can_be_queued(lkb
)) {
1939 error
= -EINPROGRESS
;
1940 add_lkb(r
, lkb
, DLM_LKSTS_WAITING
);
1941 send_blocking_asts(r
, lkb
);
1946 if (force_blocking_asts(lkb
))
1947 send_blocking_asts_all(r
, lkb
);
1948 queue_cast(r
, lkb
, -EAGAIN
);
1954 static int do_convert(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1958 /* changing an existing lock may allow others to be granted */
1960 if (can_be_granted(r
, lkb
, 1)) {
1962 queue_cast(r
, lkb
, 0);
1963 grant_pending_locks(r
);
1967 if (can_be_queued(lkb
)) {
1968 if (is_demoted(lkb
))
1969 grant_pending_locks(r
);
1970 error
= -EINPROGRESS
;
1972 add_lkb(r
, lkb
, DLM_LKSTS_CONVERT
);
1973 send_blocking_asts(r
, lkb
);
1978 if (force_blocking_asts(lkb
))
1979 send_blocking_asts_all(r
, lkb
);
1980 queue_cast(r
, lkb
, -EAGAIN
);
1986 static int do_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1988 remove_lock(r
, lkb
);
1989 queue_cast(r
, lkb
, -DLM_EUNLOCK
);
1990 grant_pending_locks(r
);
1991 return -DLM_EUNLOCK
;
1994 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
1996 static int do_cancel(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2000 error
= revert_lock(r
, lkb
);
2002 queue_cast(r
, lkb
, -DLM_ECANCEL
);
2003 grant_pending_locks(r
);
2004 return -DLM_ECANCEL
;
2010 * Four stage 3 varieties:
2011 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2014 /* add a new lkb to a possibly new rsb, called by requesting process */
2016 static int _request_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2020 /* set_master: sets lkb nodeid from r */
2022 error
= set_master(r
, lkb
);
2031 /* receive_request() calls do_request() on remote node */
2032 error
= send_request(r
, lkb
);
2034 error
= do_request(r
, lkb
);
2039 /* change some property of an existing lkb, e.g. mode */
2041 static int _convert_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2046 /* receive_convert() calls do_convert() on remote node */
2047 error
= send_convert(r
, lkb
);
2049 error
= do_convert(r
, lkb
);
2054 /* remove an existing lkb from the granted queue */
2056 static int _unlock_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2061 /* receive_unlock() calls do_unlock() on remote node */
2062 error
= send_unlock(r
, lkb
);
2064 error
= do_unlock(r
, lkb
);
2069 /* remove an existing lkb from the convert or wait queue */
2071 static int _cancel_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2076 /* receive_cancel() calls do_cancel() on remote node */
2077 error
= send_cancel(r
, lkb
);
2079 error
= do_cancel(r
, lkb
);
2085 * Four stage 2 varieties:
2086 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2089 static int request_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
, char *name
,
2090 int len
, struct dlm_args
*args
)
2095 error
= validate_lock_args(ls
, lkb
, args
);
2099 error
= find_rsb(ls
, name
, len
, R_CREATE
, &r
);
2106 lkb
->lkb_lksb
->sb_lkid
= lkb
->lkb_id
;
2108 error
= _request_lock(r
, lkb
);
2117 static int convert_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2118 struct dlm_args
*args
)
2123 r
= lkb
->lkb_resource
;
2128 error
= validate_lock_args(ls
, lkb
, args
);
2132 error
= _convert_lock(r
, lkb
);
2139 static int unlock_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2140 struct dlm_args
*args
)
2145 r
= lkb
->lkb_resource
;
2150 error
= validate_unlock_args(lkb
, args
);
2154 error
= _unlock_lock(r
, lkb
);
2161 static int cancel_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2162 struct dlm_args
*args
)
2167 r
= lkb
->lkb_resource
;
2172 error
= validate_unlock_args(lkb
, args
);
2176 error
= _cancel_lock(r
, lkb
);
2184 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2187 int dlm_lock(dlm_lockspace_t
*lockspace
,
2189 struct dlm_lksb
*lksb
,
2192 unsigned int namelen
,
2193 uint32_t parent_lkid
,
2194 void (*ast
) (void *astarg
),
2196 void (*bast
) (void *astarg
, int mode
))
2199 struct dlm_lkb
*lkb
;
2200 struct dlm_args args
;
2201 int error
, convert
= flags
& DLM_LKF_CONVERT
;
2203 ls
= dlm_find_lockspace_local(lockspace
);
2210 error
= find_lkb(ls
, lksb
->sb_lkid
, &lkb
);
2212 error
= create_lkb(ls
, &lkb
);
2217 error
= set_lock_args(mode
, lksb
, flags
, namelen
, parent_lkid
, ast
,
2218 astarg
, bast
, &args
);
2223 error
= convert_lock(ls
, lkb
, &args
);
2225 error
= request_lock(ls
, lkb
, name
, namelen
, &args
);
2227 if (error
== -EINPROGRESS
)
2230 if (convert
|| error
)
2232 if (error
== -EAGAIN
)
2235 unlock_recovery(ls
);
2236 dlm_put_lockspace(ls
);
2240 int dlm_unlock(dlm_lockspace_t
*lockspace
,
2243 struct dlm_lksb
*lksb
,
2247 struct dlm_lkb
*lkb
;
2248 struct dlm_args args
;
2251 ls
= dlm_find_lockspace_local(lockspace
);
2257 error
= find_lkb(ls
, lkid
, &lkb
);
2261 error
= set_unlock_args(flags
, astarg
, &args
);
2265 if (flags
& DLM_LKF_CANCEL
)
2266 error
= cancel_lock(ls
, lkb
, &args
);
2268 error
= unlock_lock(ls
, lkb
, &args
);
2270 if (error
== -DLM_EUNLOCK
|| error
== -DLM_ECANCEL
)
2272 if (error
== -EBUSY
&& (flags
& (DLM_LKF_CANCEL
| DLM_LKF_FORCEUNLOCK
)))
2277 unlock_recovery(ls
);
2278 dlm_put_lockspace(ls
);
2283 * send/receive routines for remote operations and replies
2287 * send_request receive_request
2288 * send_convert receive_convert
2289 * send_unlock receive_unlock
2290 * send_cancel receive_cancel
2291 * send_grant receive_grant
2292 * send_bast receive_bast
2293 * send_lookup receive_lookup
2294 * send_remove receive_remove
2297 * receive_request_reply send_request_reply
2298 * receive_convert_reply send_convert_reply
2299 * receive_unlock_reply send_unlock_reply
2300 * receive_cancel_reply send_cancel_reply
2301 * receive_lookup_reply send_lookup_reply
2304 static int create_message(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
2305 int to_nodeid
, int mstype
,
2306 struct dlm_message
**ms_ret
,
2307 struct dlm_mhandle
**mh_ret
)
2309 struct dlm_message
*ms
;
2310 struct dlm_mhandle
*mh
;
2312 int mb_len
= sizeof(struct dlm_message
);
2315 case DLM_MSG_REQUEST
:
2316 case DLM_MSG_LOOKUP
:
2317 case DLM_MSG_REMOVE
:
2318 mb_len
+= r
->res_length
;
2320 case DLM_MSG_CONVERT
:
2321 case DLM_MSG_UNLOCK
:
2322 case DLM_MSG_REQUEST_REPLY
:
2323 case DLM_MSG_CONVERT_REPLY
:
2325 if (lkb
&& lkb
->lkb_lvbptr
)
2326 mb_len
+= r
->res_ls
->ls_lvblen
;
2330 /* get_buffer gives us a message handle (mh) that we need to
2331 pass into lowcomms_commit and a message buffer (mb) that we
2332 write our data into */
2334 mh
= dlm_lowcomms_get_buffer(to_nodeid
, mb_len
, GFP_KERNEL
, &mb
);
2338 memset(mb
, 0, mb_len
);
2340 ms
= (struct dlm_message
*) mb
;
2342 ms
->m_header
.h_version
= (DLM_HEADER_MAJOR
| DLM_HEADER_MINOR
);
2343 ms
->m_header
.h_lockspace
= r
->res_ls
->ls_global_id
;
2344 ms
->m_header
.h_nodeid
= dlm_our_nodeid();
2345 ms
->m_header
.h_length
= mb_len
;
2346 ms
->m_header
.h_cmd
= DLM_MSG
;
2348 ms
->m_type
= mstype
;
2355 /* further lowcomms enhancements or alternate implementations may make
2356 the return value from this function useful at some point */
2358 static int send_message(struct dlm_mhandle
*mh
, struct dlm_message
*ms
)
2360 dlm_message_out(ms
);
2361 dlm_lowcomms_commit_buffer(mh
);
2365 static void send_args(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
2366 struct dlm_message
*ms
)
2368 ms
->m_nodeid
= lkb
->lkb_nodeid
;
2369 ms
->m_pid
= lkb
->lkb_ownpid
;
2370 ms
->m_lkid
= lkb
->lkb_id
;
2371 ms
->m_remid
= lkb
->lkb_remid
;
2372 ms
->m_exflags
= lkb
->lkb_exflags
;
2373 ms
->m_sbflags
= lkb
->lkb_sbflags
;
2374 ms
->m_flags
= lkb
->lkb_flags
;
2375 ms
->m_lvbseq
= lkb
->lkb_lvbseq
;
2376 ms
->m_status
= lkb
->lkb_status
;
2377 ms
->m_grmode
= lkb
->lkb_grmode
;
2378 ms
->m_rqmode
= lkb
->lkb_rqmode
;
2379 ms
->m_hash
= r
->res_hash
;
2381 /* m_result and m_bastmode are set from function args,
2382 not from lkb fields */
2384 if (lkb
->lkb_bastaddr
)
2385 ms
->m_asts
|= AST_BAST
;
2386 if (lkb
->lkb_astaddr
)
2387 ms
->m_asts
|= AST_COMP
;
2389 /* compare with switch in create_message; send_remove() doesn't
2392 switch (ms
->m_type
) {
2393 case DLM_MSG_REQUEST
:
2394 case DLM_MSG_LOOKUP
:
2395 memcpy(ms
->m_extra
, r
->res_name
, r
->res_length
);
2397 case DLM_MSG_CONVERT
:
2398 case DLM_MSG_UNLOCK
:
2399 case DLM_MSG_REQUEST_REPLY
:
2400 case DLM_MSG_CONVERT_REPLY
:
2402 if (!lkb
->lkb_lvbptr
)
2404 memcpy(ms
->m_extra
, lkb
->lkb_lvbptr
, r
->res_ls
->ls_lvblen
);
2409 static int send_common(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int mstype
)
2411 struct dlm_message
*ms
;
2412 struct dlm_mhandle
*mh
;
2413 int to_nodeid
, error
;
2415 error
= add_to_waiters(lkb
, mstype
);
2419 to_nodeid
= r
->res_nodeid
;
2421 error
= create_message(r
, lkb
, to_nodeid
, mstype
, &ms
, &mh
);
2425 send_args(r
, lkb
, ms
);
2427 error
= send_message(mh
, ms
);
2433 remove_from_waiters(lkb
, msg_reply_type(mstype
));
2437 static int send_request(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2439 return send_common(r
, lkb
, DLM_MSG_REQUEST
);
2442 static int send_convert(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2446 error
= send_common(r
, lkb
, DLM_MSG_CONVERT
);
2448 /* down conversions go without a reply from the master */
2449 if (!error
&& down_conversion(lkb
)) {
2450 remove_from_waiters(lkb
, DLM_MSG_CONVERT_REPLY
);
2451 r
->res_ls
->ls_stub_ms
.m_type
= DLM_MSG_CONVERT_REPLY
;
2452 r
->res_ls
->ls_stub_ms
.m_result
= 0;
2453 r
->res_ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
2454 __receive_convert_reply(r
, lkb
, &r
->res_ls
->ls_stub_ms
);
2460 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2461 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2462 that the master is still correct. */
2464 static int send_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2466 return send_common(r
, lkb
, DLM_MSG_UNLOCK
);
2469 static int send_cancel(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2471 return send_common(r
, lkb
, DLM_MSG_CANCEL
);
2474 static int send_grant(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2476 struct dlm_message
*ms
;
2477 struct dlm_mhandle
*mh
;
2478 int to_nodeid
, error
;
2480 to_nodeid
= lkb
->lkb_nodeid
;
2482 error
= create_message(r
, lkb
, to_nodeid
, DLM_MSG_GRANT
, &ms
, &mh
);
2486 send_args(r
, lkb
, ms
);
2490 error
= send_message(mh
, ms
);
2495 static int send_bast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int mode
)
2497 struct dlm_message
*ms
;
2498 struct dlm_mhandle
*mh
;
2499 int to_nodeid
, error
;
2501 to_nodeid
= lkb
->lkb_nodeid
;
2503 error
= create_message(r
, NULL
, to_nodeid
, DLM_MSG_BAST
, &ms
, &mh
);
2507 send_args(r
, lkb
, ms
);
2509 ms
->m_bastmode
= mode
;
2511 error
= send_message(mh
, ms
);
2516 static int send_lookup(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2518 struct dlm_message
*ms
;
2519 struct dlm_mhandle
*mh
;
2520 int to_nodeid
, error
;
2522 error
= add_to_waiters(lkb
, DLM_MSG_LOOKUP
);
2526 to_nodeid
= dlm_dir_nodeid(r
);
2528 error
= create_message(r
, NULL
, to_nodeid
, DLM_MSG_LOOKUP
, &ms
, &mh
);
2532 send_args(r
, lkb
, ms
);
2534 error
= send_message(mh
, ms
);
2540 remove_from_waiters(lkb
, DLM_MSG_LOOKUP_REPLY
);
2544 static int send_remove(struct dlm_rsb
*r
)
2546 struct dlm_message
*ms
;
2547 struct dlm_mhandle
*mh
;
2548 int to_nodeid
, error
;
2550 to_nodeid
= dlm_dir_nodeid(r
);
2552 error
= create_message(r
, NULL
, to_nodeid
, DLM_MSG_REMOVE
, &ms
, &mh
);
2556 memcpy(ms
->m_extra
, r
->res_name
, r
->res_length
);
2557 ms
->m_hash
= r
->res_hash
;
2559 error
= send_message(mh
, ms
);
2564 static int send_common_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
2567 struct dlm_message
*ms
;
2568 struct dlm_mhandle
*mh
;
2569 int to_nodeid
, error
;
2571 to_nodeid
= lkb
->lkb_nodeid
;
2573 error
= create_message(r
, lkb
, to_nodeid
, mstype
, &ms
, &mh
);
2577 send_args(r
, lkb
, ms
);
2581 error
= send_message(mh
, ms
);
2586 static int send_request_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2588 return send_common_reply(r
, lkb
, DLM_MSG_REQUEST_REPLY
, rv
);
2591 static int send_convert_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2593 return send_common_reply(r
, lkb
, DLM_MSG_CONVERT_REPLY
, rv
);
2596 static int send_unlock_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2598 return send_common_reply(r
, lkb
, DLM_MSG_UNLOCK_REPLY
, rv
);
2601 static int send_cancel_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2603 return send_common_reply(r
, lkb
, DLM_MSG_CANCEL_REPLY
, rv
);
2606 static int send_lookup_reply(struct dlm_ls
*ls
, struct dlm_message
*ms_in
,
2607 int ret_nodeid
, int rv
)
2609 struct dlm_rsb
*r
= &ls
->ls_stub_rsb
;
2610 struct dlm_message
*ms
;
2611 struct dlm_mhandle
*mh
;
2612 int error
, nodeid
= ms_in
->m_header
.h_nodeid
;
2614 error
= create_message(r
, NULL
, nodeid
, DLM_MSG_LOOKUP_REPLY
, &ms
, &mh
);
2618 ms
->m_lkid
= ms_in
->m_lkid
;
2620 ms
->m_nodeid
= ret_nodeid
;
2622 error
= send_message(mh
, ms
);
2627 /* which args we save from a received message depends heavily on the type
2628 of message, unlike the send side where we can safely send everything about
2629 the lkb for any type of message */
2631 static void receive_flags(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
2633 lkb
->lkb_exflags
= ms
->m_exflags
;
2634 lkb
->lkb_sbflags
= ms
->m_sbflags
;
2635 lkb
->lkb_flags
= (lkb
->lkb_flags
& 0xFFFF0000) |
2636 (ms
->m_flags
& 0x0000FFFF);
2639 static void receive_flags_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
2641 lkb
->lkb_sbflags
= ms
->m_sbflags
;
2642 lkb
->lkb_flags
= (lkb
->lkb_flags
& 0xFFFF0000) |
2643 (ms
->m_flags
& 0x0000FFFF);
2646 static int receive_extralen(struct dlm_message
*ms
)
2648 return (ms
->m_header
.h_length
- sizeof(struct dlm_message
));
2651 static int receive_lvb(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2652 struct dlm_message
*ms
)
2656 if (lkb
->lkb_exflags
& DLM_LKF_VALBLK
) {
2657 if (!lkb
->lkb_lvbptr
)
2658 lkb
->lkb_lvbptr
= allocate_lvb(ls
);
2659 if (!lkb
->lkb_lvbptr
)
2661 len
= receive_extralen(ms
);
2662 memcpy(lkb
->lkb_lvbptr
, ms
->m_extra
, len
);
2667 static int receive_request_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2668 struct dlm_message
*ms
)
2670 lkb
->lkb_nodeid
= ms
->m_header
.h_nodeid
;
2671 lkb
->lkb_ownpid
= ms
->m_pid
;
2672 lkb
->lkb_remid
= ms
->m_lkid
;
2673 lkb
->lkb_grmode
= DLM_LOCK_IV
;
2674 lkb
->lkb_rqmode
= ms
->m_rqmode
;
2675 lkb
->lkb_bastaddr
= (void *) (long) (ms
->m_asts
& AST_BAST
);
2676 lkb
->lkb_astaddr
= (void *) (long) (ms
->m_asts
& AST_COMP
);
2678 DLM_ASSERT(is_master_copy(lkb
), dlm_print_lkb(lkb
););
2680 if (lkb
->lkb_exflags
& DLM_LKF_VALBLK
) {
2681 /* lkb was just created so there won't be an lvb yet */
2682 lkb
->lkb_lvbptr
= allocate_lvb(ls
);
2683 if (!lkb
->lkb_lvbptr
)
2690 static int receive_convert_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2691 struct dlm_message
*ms
)
2693 if (lkb
->lkb_nodeid
!= ms
->m_header
.h_nodeid
) {
2694 log_error(ls
, "convert_args nodeid %d %d lkid %x %x",
2695 lkb
->lkb_nodeid
, ms
->m_header
.h_nodeid
,
2696 lkb
->lkb_id
, lkb
->lkb_remid
);
2700 if (!is_master_copy(lkb
))
2703 if (lkb
->lkb_status
!= DLM_LKSTS_GRANTED
)
2706 if (receive_lvb(ls
, lkb
, ms
))
2709 lkb
->lkb_rqmode
= ms
->m_rqmode
;
2710 lkb
->lkb_lvbseq
= ms
->m_lvbseq
;
2715 static int receive_unlock_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2716 struct dlm_message
*ms
)
2718 if (!is_master_copy(lkb
))
2720 if (receive_lvb(ls
, lkb
, ms
))
2725 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
2726 uses to send a reply and that the remote end uses to process the reply. */
2728 static void setup_stub_lkb(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2730 struct dlm_lkb
*lkb
= &ls
->ls_stub_lkb
;
2731 lkb
->lkb_nodeid
= ms
->m_header
.h_nodeid
;
2732 lkb
->lkb_remid
= ms
->m_lkid
;
2735 static void receive_request(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2737 struct dlm_lkb
*lkb
;
2741 error
= create_lkb(ls
, &lkb
);
2745 receive_flags(lkb
, ms
);
2746 lkb
->lkb_flags
|= DLM_IFL_MSTCPY
;
2747 error
= receive_request_args(ls
, lkb
, ms
);
2753 namelen
= receive_extralen(ms
);
2755 error
= find_rsb(ls
, ms
->m_extra
, namelen
, R_MASTER
, &r
);
2764 error
= do_request(r
, lkb
);
2765 send_request_reply(r
, lkb
, error
);
2770 if (error
== -EINPROGRESS
)
2777 setup_stub_lkb(ls
, ms
);
2778 send_request_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
2781 static void receive_convert(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2783 struct dlm_lkb
*lkb
;
2785 int error
, reply
= 1;
2787 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
2791 r
= lkb
->lkb_resource
;
2796 receive_flags(lkb
, ms
);
2797 error
= receive_convert_args(ls
, lkb
, ms
);
2800 reply
= !down_conversion(lkb
);
2802 error
= do_convert(r
, lkb
);
2805 send_convert_reply(r
, lkb
, error
);
2813 setup_stub_lkb(ls
, ms
);
2814 send_convert_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
2817 static void receive_unlock(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2819 struct dlm_lkb
*lkb
;
2823 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
2827 r
= lkb
->lkb_resource
;
2832 receive_flags(lkb
, ms
);
2833 error
= receive_unlock_args(ls
, lkb
, ms
);
2837 error
= do_unlock(r
, lkb
);
2839 send_unlock_reply(r
, lkb
, error
);
2847 setup_stub_lkb(ls
, ms
);
2848 send_unlock_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
2851 static void receive_cancel(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2853 struct dlm_lkb
*lkb
;
2857 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
2861 receive_flags(lkb
, ms
);
2863 r
= lkb
->lkb_resource
;
2868 error
= do_cancel(r
, lkb
);
2869 send_cancel_reply(r
, lkb
, error
);
2877 setup_stub_lkb(ls
, ms
);
2878 send_cancel_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
2881 static void receive_grant(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2883 struct dlm_lkb
*lkb
;
2887 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
2889 log_error(ls
, "receive_grant no lkb");
2892 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
2894 r
= lkb
->lkb_resource
;
2899 receive_flags_reply(lkb
, ms
);
2900 grant_lock_pc(r
, lkb
, ms
);
2901 queue_cast(r
, lkb
, 0);
2908 static void receive_bast(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2910 struct dlm_lkb
*lkb
;
2914 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
2916 log_error(ls
, "receive_bast no lkb");
2919 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
2921 r
= lkb
->lkb_resource
;
2926 queue_bast(r
, lkb
, ms
->m_bastmode
);
2933 static void receive_lookup(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2935 int len
, error
, ret_nodeid
, dir_nodeid
, from_nodeid
, our_nodeid
;
2937 from_nodeid
= ms
->m_header
.h_nodeid
;
2938 our_nodeid
= dlm_our_nodeid();
2940 len
= receive_extralen(ms
);
2942 dir_nodeid
= dlm_hash2nodeid(ls
, ms
->m_hash
);
2943 if (dir_nodeid
!= our_nodeid
) {
2944 log_error(ls
, "lookup dir_nodeid %d from %d",
2945 dir_nodeid
, from_nodeid
);
2951 error
= dlm_dir_lookup(ls
, from_nodeid
, ms
->m_extra
, len
, &ret_nodeid
);
2953 /* Optimization: we're master so treat lookup as a request */
2954 if (!error
&& ret_nodeid
== our_nodeid
) {
2955 receive_request(ls
, ms
);
2959 send_lookup_reply(ls
, ms
, ret_nodeid
, error
);
2962 static void receive_remove(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2964 int len
, dir_nodeid
, from_nodeid
;
2966 from_nodeid
= ms
->m_header
.h_nodeid
;
2968 len
= receive_extralen(ms
);
2970 dir_nodeid
= dlm_hash2nodeid(ls
, ms
->m_hash
);
2971 if (dir_nodeid
!= dlm_our_nodeid()) {
2972 log_error(ls
, "remove dir entry dir_nodeid %d from %d",
2973 dir_nodeid
, from_nodeid
);
2977 dlm_dir_remove_entry(ls
, from_nodeid
, ms
->m_extra
, len
);
2980 static void receive_request_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
2982 struct dlm_lkb
*lkb
;
2984 int error
, mstype
, result
;
2986 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
2988 log_error(ls
, "receive_request_reply no lkb");
2991 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
2993 r
= lkb
->lkb_resource
;
2997 mstype
= lkb
->lkb_wait_type
;
2998 error
= remove_from_waiters(lkb
, DLM_MSG_REQUEST_REPLY
);
3002 /* Optimization: the dir node was also the master, so it took our
3003 lookup as a request and sent request reply instead of lookup reply */
3004 if (mstype
== DLM_MSG_LOOKUP
) {
3005 r
->res_nodeid
= ms
->m_header
.h_nodeid
;
3006 lkb
->lkb_nodeid
= r
->res_nodeid
;
3009 /* this is the value returned from do_request() on the master */
3010 result
= ms
->m_result
;
3014 /* request would block (be queued) on remote master */
3015 queue_cast(r
, lkb
, -EAGAIN
);
3016 confirm_master(r
, -EAGAIN
);
3017 unhold_lkb(lkb
); /* undoes create_lkb() */
3022 /* request was queued or granted on remote master */
3023 receive_flags_reply(lkb
, ms
);
3024 lkb
->lkb_remid
= ms
->m_lkid
;
3026 add_lkb(r
, lkb
, DLM_LKSTS_WAITING
);
3028 grant_lock_pc(r
, lkb
, ms
);
3029 queue_cast(r
, lkb
, 0);
3031 confirm_master(r
, result
);
3036 /* find_rsb failed to find rsb or rsb wasn't master */
3037 log_debug(ls
, "receive_request_reply %x %x master diff %d %d",
3038 lkb
->lkb_id
, lkb
->lkb_flags
, r
->res_nodeid
, result
);
3040 lkb
->lkb_nodeid
= -1;
3042 if (is_overlap(lkb
)) {
3043 /* we'll ignore error in cancel/unlock reply */
3044 queue_cast_overlap(r
, lkb
);
3045 unhold_lkb(lkb
); /* undoes create_lkb() */
3047 _request_lock(r
, lkb
);
3051 log_error(ls
, "receive_request_reply %x error %d",
3052 lkb
->lkb_id
, result
);
3055 if (is_overlap_unlock(lkb
) && (result
== 0 || result
== -EINPROGRESS
)) {
3056 log_debug(ls
, "receive_request_reply %x result %d unlock",
3057 lkb
->lkb_id
, result
);
3058 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3059 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3060 send_unlock(r
, lkb
);
3061 } else if (is_overlap_cancel(lkb
) && (result
== -EINPROGRESS
)) {
3062 log_debug(ls
, "receive_request_reply %x cancel", lkb
->lkb_id
);
3063 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3064 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3065 send_cancel(r
, lkb
);
3067 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3068 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3076 static void __receive_convert_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
3077 struct dlm_message
*ms
)
3079 /* this is the value returned from do_convert() on the master */
3080 switch (ms
->m_result
) {
3082 /* convert would block (be queued) on remote master */
3083 queue_cast(r
, lkb
, -EAGAIN
);
3087 /* convert was queued on remote master */
3089 add_lkb(r
, lkb
, DLM_LKSTS_CONVERT
);
3093 /* convert was granted on remote master */
3094 receive_flags_reply(lkb
, ms
);
3095 grant_lock_pc(r
, lkb
, ms
);
3096 queue_cast(r
, lkb
, 0);
3100 log_error(r
->res_ls
, "receive_convert_reply %x error %d",
3101 lkb
->lkb_id
, ms
->m_result
);
3105 static void _receive_convert_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3107 struct dlm_rsb
*r
= lkb
->lkb_resource
;
3113 /* stub reply can happen with waiters_mutex held */
3114 error
= remove_from_waiters_ms(lkb
, ms
);
3118 __receive_convert_reply(r
, lkb
, ms
);
3124 static void receive_convert_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3126 struct dlm_lkb
*lkb
;
3129 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3131 log_error(ls
, "receive_convert_reply no lkb");
3134 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
3136 _receive_convert_reply(lkb
, ms
);
3140 static void _receive_unlock_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3142 struct dlm_rsb
*r
= lkb
->lkb_resource
;
3148 /* stub reply can happen with waiters_mutex held */
3149 error
= remove_from_waiters_ms(lkb
, ms
);
3153 /* this is the value returned from do_unlock() on the master */
3155 switch (ms
->m_result
) {
3157 receive_flags_reply(lkb
, ms
);
3158 remove_lock_pc(r
, lkb
);
3159 queue_cast(r
, lkb
, -DLM_EUNLOCK
);
3164 log_error(r
->res_ls
, "receive_unlock_reply %x error %d",
3165 lkb
->lkb_id
, ms
->m_result
);
3172 static void receive_unlock_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3174 struct dlm_lkb
*lkb
;
3177 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3179 log_error(ls
, "receive_unlock_reply no lkb");
3182 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
3184 _receive_unlock_reply(lkb
, ms
);
3188 static void _receive_cancel_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3190 struct dlm_rsb
*r
= lkb
->lkb_resource
;
3196 /* stub reply can happen with waiters_mutex held */
3197 error
= remove_from_waiters_ms(lkb
, ms
);
3201 /* this is the value returned from do_cancel() on the master */
3203 switch (ms
->m_result
) {
3205 receive_flags_reply(lkb
, ms
);
3206 revert_lock_pc(r
, lkb
);
3208 queue_cast(r
, lkb
, -DLM_ECANCEL
);
3213 log_error(r
->res_ls
, "receive_cancel_reply %x error %d",
3214 lkb
->lkb_id
, ms
->m_result
);
3221 static void receive_cancel_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3223 struct dlm_lkb
*lkb
;
3226 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3228 log_error(ls
, "receive_cancel_reply no lkb");
3231 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
3233 _receive_cancel_reply(lkb
, ms
);
3237 static void receive_lookup_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3239 struct dlm_lkb
*lkb
;
3241 int error
, ret_nodeid
;
3243 error
= find_lkb(ls
, ms
->m_lkid
, &lkb
);
3245 log_error(ls
, "receive_lookup_reply no lkb");
3249 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
3250 FIXME: will a non-zero error ever be returned? */
3252 r
= lkb
->lkb_resource
;
3256 error
= remove_from_waiters(lkb
, DLM_MSG_LOOKUP_REPLY
);
3260 ret_nodeid
= ms
->m_nodeid
;
3261 if (ret_nodeid
== dlm_our_nodeid()) {
3264 r
->res_first_lkid
= 0;
3266 /* set_master() will copy res_nodeid to lkb_nodeid */
3267 r
->res_nodeid
= ret_nodeid
;
3270 if (is_overlap(lkb
)) {
3271 log_debug(ls
, "receive_lookup_reply %x unlock %x",
3272 lkb
->lkb_id
, lkb
->lkb_flags
);
3273 queue_cast_overlap(r
, lkb
);
3274 unhold_lkb(lkb
); /* undoes create_lkb() */
3278 _request_lock(r
, lkb
);
3282 process_lookup_list(r
);
3289 int dlm_receive_message(struct dlm_header
*hd
, int nodeid
, int recovery
)
3291 struct dlm_message
*ms
= (struct dlm_message
*) hd
;
3298 ls
= dlm_find_lockspace_global(hd
->h_lockspace
);
3300 log_print("drop message %d from %d for unknown lockspace %d",
3301 ms
->m_type
, nodeid
, hd
->h_lockspace
);
3305 /* recovery may have just ended leaving a bunch of backed-up requests
3306 in the requestqueue; wait while dlm_recoverd clears them */
3309 dlm_wait_requestqueue(ls
);
3311 /* recovery may have just started while there were a bunch of
3312 in-flight requests -- save them in requestqueue to be processed
3313 after recovery. we can't let dlm_recvd block on the recovery
3314 lock. if dlm_recoverd is calling this function to clear the
3315 requestqueue, it needs to be interrupted (-EINTR) if another
3316 recovery operation is starting. */
3319 if (dlm_locking_stopped(ls
)) {
3324 error
= dlm_add_requestqueue(ls
, nodeid
, hd
);
3325 if (error
== -EAGAIN
)
3333 if (lock_recovery_try(ls
))
3338 switch (ms
->m_type
) {
3340 /* messages sent to a master node */
3342 case DLM_MSG_REQUEST
:
3343 receive_request(ls
, ms
);
3346 case DLM_MSG_CONVERT
:
3347 receive_convert(ls
, ms
);
3350 case DLM_MSG_UNLOCK
:
3351 receive_unlock(ls
, ms
);
3354 case DLM_MSG_CANCEL
:
3355 receive_cancel(ls
, ms
);
3358 /* messages sent from a master node (replies to above) */
3360 case DLM_MSG_REQUEST_REPLY
:
3361 receive_request_reply(ls
, ms
);
3364 case DLM_MSG_CONVERT_REPLY
:
3365 receive_convert_reply(ls
, ms
);
3368 case DLM_MSG_UNLOCK_REPLY
:
3369 receive_unlock_reply(ls
, ms
);
3372 case DLM_MSG_CANCEL_REPLY
:
3373 receive_cancel_reply(ls
, ms
);
3376 /* messages sent from a master node (only two types of async msg) */
3379 receive_grant(ls
, ms
);
3383 receive_bast(ls
, ms
);
3386 /* messages sent to a dir node */
3388 case DLM_MSG_LOOKUP
:
3389 receive_lookup(ls
, ms
);
3392 case DLM_MSG_REMOVE
:
3393 receive_remove(ls
, ms
);
3396 /* messages sent from a dir node (remove has no reply) */
3398 case DLM_MSG_LOOKUP_REPLY
:
3399 receive_lookup_reply(ls
, ms
);
3403 log_error(ls
, "unknown message type %d", ms
->m_type
);
3406 unlock_recovery(ls
);
3408 dlm_put_lockspace(ls
);
3418 static void recover_convert_waiter(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
3420 if (middle_conversion(lkb
)) {
3422 ls
->ls_stub_ms
.m_type
= DLM_MSG_CONVERT_REPLY
;
3423 ls
->ls_stub_ms
.m_result
= -EINPROGRESS
;
3424 ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
3425 _receive_convert_reply(lkb
, &ls
->ls_stub_ms
);
3427 /* Same special case as in receive_rcom_lock_args() */
3428 lkb
->lkb_grmode
= DLM_LOCK_IV
;
3429 rsb_set_flag(lkb
->lkb_resource
, RSB_RECOVER_CONVERT
);
3432 } else if (lkb
->lkb_rqmode
>= lkb
->lkb_grmode
) {
3433 lkb
->lkb_flags
|= DLM_IFL_RESEND
;
3436 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3437 conversions are async; there's no reply from the remote master */
3440 /* A waiting lkb needs recovery if the master node has failed, or
3441 the master node is changing (only when no directory is used) */
3443 static int waiter_needs_recovery(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
3445 if (dlm_is_removed(ls
, lkb
->lkb_nodeid
))
3448 if (!dlm_no_directory(ls
))
3451 if (dlm_dir_nodeid(lkb
->lkb_resource
) != lkb
->lkb_nodeid
)
3457 /* Recovery for locks that are waiting for replies from nodes that are now
3458 gone. We can just complete unlocks and cancels by faking a reply from the
3459 dead node. Requests and up-conversions we flag to be resent after
3460 recovery. Down-conversions can just be completed with a fake reply like
3461 unlocks. Conversions between PR and CW need special attention. */
3463 void dlm_recover_waiters_pre(struct dlm_ls
*ls
)
3465 struct dlm_lkb
*lkb
, *safe
;
3467 mutex_lock(&ls
->ls_waiters_mutex
);
3469 list_for_each_entry_safe(lkb
, safe
, &ls
->ls_waiters
, lkb_wait_reply
) {
3470 log_debug(ls
, "pre recover waiter lkid %x type %d flags %x",
3471 lkb
->lkb_id
, lkb
->lkb_wait_type
, lkb
->lkb_flags
);
3473 /* all outstanding lookups, regardless of destination will be
3474 resent after recovery is done */
3476 if (lkb
->lkb_wait_type
== DLM_MSG_LOOKUP
) {
3477 lkb
->lkb_flags
|= DLM_IFL_RESEND
;
3481 if (!waiter_needs_recovery(ls
, lkb
))
3484 switch (lkb
->lkb_wait_type
) {
3486 case DLM_MSG_REQUEST
:
3487 lkb
->lkb_flags
|= DLM_IFL_RESEND
;
3490 case DLM_MSG_CONVERT
:
3491 recover_convert_waiter(ls
, lkb
);
3494 case DLM_MSG_UNLOCK
:
3496 ls
->ls_stub_ms
.m_type
= DLM_MSG_UNLOCK_REPLY
;
3497 ls
->ls_stub_ms
.m_result
= -DLM_EUNLOCK
;
3498 ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
3499 _receive_unlock_reply(lkb
, &ls
->ls_stub_ms
);
3503 case DLM_MSG_CANCEL
:
3505 ls
->ls_stub_ms
.m_type
= DLM_MSG_CANCEL_REPLY
;
3506 ls
->ls_stub_ms
.m_result
= -DLM_ECANCEL
;
3507 ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
3508 _receive_cancel_reply(lkb
, &ls
->ls_stub_ms
);
3513 log_error(ls
, "invalid lkb wait_type %d",
3514 lkb
->lkb_wait_type
);
3518 mutex_unlock(&ls
->ls_waiters_mutex
);
3521 static struct dlm_lkb
*find_resend_waiter(struct dlm_ls
*ls
)
3523 struct dlm_lkb
*lkb
;
3526 mutex_lock(&ls
->ls_waiters_mutex
);
3527 list_for_each_entry(lkb
, &ls
->ls_waiters
, lkb_wait_reply
) {
3528 if (lkb
->lkb_flags
& DLM_IFL_RESEND
) {
3534 mutex_unlock(&ls
->ls_waiters_mutex
);
3541 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
3542 master or dir-node for r. Processing the lkb may result in it being placed
3545 /* We do this after normal locking has been enabled and any saved messages
3546 (in requestqueue) have been processed. We should be confident that at
3547 this point we won't get or process a reply to any of these waiting
3548 operations. But, new ops may be coming in on the rsbs/locks here from
3549 userspace or remotely. */
3551 /* there may have been an overlap unlock/cancel prior to recovery or after
3552 recovery. if before, the lkb may still have a pos wait_count; if after, the
3553 overlap flag would just have been set and nothing new sent. we can be
3554 confident here than any replies to either the initial op or overlap ops
3555 prior to recovery have been received. */
3557 int dlm_recover_waiters_post(struct dlm_ls
*ls
)
3559 struct dlm_lkb
*lkb
;
3561 int error
= 0, mstype
, err
, oc
, ou
;
3564 if (dlm_locking_stopped(ls
)) {
3565 log_debug(ls
, "recover_waiters_post aborted");
3570 lkb
= find_resend_waiter(ls
);
3574 r
= lkb
->lkb_resource
;
3578 mstype
= lkb
->lkb_wait_type
;
3579 oc
= is_overlap_cancel(lkb
);
3580 ou
= is_overlap_unlock(lkb
);
3583 log_debug(ls
, "recover_waiters_post %x type %d flags %x %s",
3584 lkb
->lkb_id
, mstype
, lkb
->lkb_flags
, r
->res_name
);
3586 /* At this point we assume that we won't get a reply to any
3587 previous op or overlap op on this lock. First, do a big
3588 remove_from_waiters() for all previous ops. */
3590 lkb
->lkb_flags
&= ~DLM_IFL_RESEND
;
3591 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3592 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3593 lkb
->lkb_wait_type
= 0;
3594 lkb
->lkb_wait_count
= 0;
3595 mutex_lock(&ls
->ls_waiters_mutex
);
3596 list_del_init(&lkb
->lkb_wait_reply
);
3597 mutex_unlock(&ls
->ls_waiters_mutex
);
3598 unhold_lkb(lkb
); /* for waiters list */
3601 /* do an unlock or cancel instead of resending */
3603 case DLM_MSG_LOOKUP
:
3604 case DLM_MSG_REQUEST
:
3605 queue_cast(r
, lkb
, ou
? -DLM_EUNLOCK
:
3607 unhold_lkb(lkb
); /* undoes create_lkb() */
3609 case DLM_MSG_CONVERT
:
3611 queue_cast(r
, lkb
, -DLM_ECANCEL
);
3613 lkb
->lkb_exflags
|= DLM_LKF_FORCEUNLOCK
;
3614 _unlock_lock(r
, lkb
);
3622 case DLM_MSG_LOOKUP
:
3623 case DLM_MSG_REQUEST
:
3624 _request_lock(r
, lkb
);
3626 confirm_master(r
, 0);
3628 case DLM_MSG_CONVERT
:
3629 _convert_lock(r
, lkb
);
3637 log_error(ls
, "recover_waiters_post %x %d %x %d %d",
3638 lkb
->lkb_id
, mstype
, lkb
->lkb_flags
, oc
, ou
);
3647 static void purge_queue(struct dlm_rsb
*r
, struct list_head
*queue
,
3648 int (*test
)(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
))
3650 struct dlm_ls
*ls
= r
->res_ls
;
3651 struct dlm_lkb
*lkb
, *safe
;
3653 list_for_each_entry_safe(lkb
, safe
, queue
, lkb_statequeue
) {
3654 if (test(ls
, lkb
)) {
3655 rsb_set_flag(r
, RSB_LOCKS_PURGED
);
3657 /* this put should free the lkb */
3658 if (!dlm_put_lkb(lkb
))
3659 log_error(ls
, "purged lkb not released");
3664 static int purge_dead_test(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
3666 return (is_master_copy(lkb
) && dlm_is_removed(ls
, lkb
->lkb_nodeid
));
3669 static int purge_mstcpy_test(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
3671 return is_master_copy(lkb
);
3674 static void purge_dead_locks(struct dlm_rsb
*r
)
3676 purge_queue(r
, &r
->res_grantqueue
, &purge_dead_test
);
3677 purge_queue(r
, &r
->res_convertqueue
, &purge_dead_test
);
3678 purge_queue(r
, &r
->res_waitqueue
, &purge_dead_test
);
3681 void dlm_purge_mstcpy_locks(struct dlm_rsb
*r
)
3683 purge_queue(r
, &r
->res_grantqueue
, &purge_mstcpy_test
);
3684 purge_queue(r
, &r
->res_convertqueue
, &purge_mstcpy_test
);
3685 purge_queue(r
, &r
->res_waitqueue
, &purge_mstcpy_test
);
3688 /* Get rid of locks held by nodes that are gone. */
3690 int dlm_purge_locks(struct dlm_ls
*ls
)
3694 log_debug(ls
, "dlm_purge_locks");
3696 down_write(&ls
->ls_root_sem
);
3697 list_for_each_entry(r
, &ls
->ls_root_list
, res_root_list
) {
3701 purge_dead_locks(r
);
3707 up_write(&ls
->ls_root_sem
);
3712 static struct dlm_rsb
*find_purged_rsb(struct dlm_ls
*ls
, int bucket
)
3714 struct dlm_rsb
*r
, *r_ret
= NULL
;
3716 read_lock(&ls
->ls_rsbtbl
[bucket
].lock
);
3717 list_for_each_entry(r
, &ls
->ls_rsbtbl
[bucket
].list
, res_hashchain
) {
3718 if (!rsb_flag(r
, RSB_LOCKS_PURGED
))
3721 rsb_clear_flag(r
, RSB_LOCKS_PURGED
);
3725 read_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
3729 void dlm_grant_after_purge(struct dlm_ls
*ls
)
3735 r
= find_purged_rsb(ls
, bucket
);
3737 if (bucket
== ls
->ls_rsbtbl_size
- 1)
3744 grant_pending_locks(r
);
3745 confirm_master(r
, 0);
3753 static struct dlm_lkb
*search_remid_list(struct list_head
*head
, int nodeid
,
3756 struct dlm_lkb
*lkb
;
3758 list_for_each_entry(lkb
, head
, lkb_statequeue
) {
3759 if (lkb
->lkb_nodeid
== nodeid
&& lkb
->lkb_remid
== remid
)
3765 static struct dlm_lkb
*search_remid(struct dlm_rsb
*r
, int nodeid
,
3768 struct dlm_lkb
*lkb
;
3770 lkb
= search_remid_list(&r
->res_grantqueue
, nodeid
, remid
);
3773 lkb
= search_remid_list(&r
->res_convertqueue
, nodeid
, remid
);
3776 lkb
= search_remid_list(&r
->res_waitqueue
, nodeid
, remid
);
3782 static int receive_rcom_lock_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
3783 struct dlm_rsb
*r
, struct dlm_rcom
*rc
)
3785 struct rcom_lock
*rl
= (struct rcom_lock
*) rc
->rc_buf
;
3788 lkb
->lkb_nodeid
= rc
->rc_header
.h_nodeid
;
3789 lkb
->lkb_ownpid
= rl
->rl_ownpid
;
3790 lkb
->lkb_remid
= rl
->rl_lkid
;
3791 lkb
->lkb_exflags
= rl
->rl_exflags
;
3792 lkb
->lkb_flags
= rl
->rl_flags
& 0x0000FFFF;
3793 lkb
->lkb_flags
|= DLM_IFL_MSTCPY
;
3794 lkb
->lkb_lvbseq
= rl
->rl_lvbseq
;
3795 lkb
->lkb_rqmode
= rl
->rl_rqmode
;
3796 lkb
->lkb_grmode
= rl
->rl_grmode
;
3797 /* don't set lkb_status because add_lkb wants to itself */
3799 lkb
->lkb_bastaddr
= (void *) (long) (rl
->rl_asts
& AST_BAST
);
3800 lkb
->lkb_astaddr
= (void *) (long) (rl
->rl_asts
& AST_COMP
);
3802 if (lkb
->lkb_exflags
& DLM_LKF_VALBLK
) {
3803 lkb
->lkb_lvbptr
= allocate_lvb(ls
);
3804 if (!lkb
->lkb_lvbptr
)
3806 lvblen
= rc
->rc_header
.h_length
- sizeof(struct dlm_rcom
) -
3807 sizeof(struct rcom_lock
);
3808 memcpy(lkb
->lkb_lvbptr
, rl
->rl_lvb
, lvblen
);
3811 /* Conversions between PR and CW (middle modes) need special handling.
3812 The real granted mode of these converting locks cannot be determined
3813 until all locks have been rebuilt on the rsb (recover_conversion) */
3815 if (rl
->rl_wait_type
== DLM_MSG_CONVERT
&& middle_conversion(lkb
)) {
3816 rl
->rl_status
= DLM_LKSTS_CONVERT
;
3817 lkb
->lkb_grmode
= DLM_LOCK_IV
;
3818 rsb_set_flag(r
, RSB_RECOVER_CONVERT
);
3824 /* This lkb may have been recovered in a previous aborted recovery so we need
3825 to check if the rsb already has an lkb with the given remote nodeid/lkid.
3826 If so we just send back a standard reply. If not, we create a new lkb with
3827 the given values and send back our lkid. We send back our lkid by sending
3828 back the rcom_lock struct we got but with the remid field filled in. */
3830 int dlm_recover_master_copy(struct dlm_ls
*ls
, struct dlm_rcom
*rc
)
3832 struct rcom_lock
*rl
= (struct rcom_lock
*) rc
->rc_buf
;
3834 struct dlm_lkb
*lkb
;
3837 if (rl
->rl_parent_lkid
) {
3838 error
= -EOPNOTSUPP
;
3842 error
= find_rsb(ls
, rl
->rl_name
, rl
->rl_namelen
, R_MASTER
, &r
);
3848 lkb
= search_remid(r
, rc
->rc_header
.h_nodeid
, rl
->rl_lkid
);
3854 error
= create_lkb(ls
, &lkb
);
3858 error
= receive_rcom_lock_args(ls
, lkb
, r
, rc
);
3865 add_lkb(r
, lkb
, rl
->rl_status
);
3869 /* this is the new value returned to the lock holder for
3870 saving in its process-copy lkb */
3871 rl
->rl_remid
= lkb
->lkb_id
;
3878 log_print("recover_master_copy %d %x", error
, rl
->rl_lkid
);
3879 rl
->rl_result
= error
;
3883 int dlm_recover_process_copy(struct dlm_ls
*ls
, struct dlm_rcom
*rc
)
3885 struct rcom_lock
*rl
= (struct rcom_lock
*) rc
->rc_buf
;
3887 struct dlm_lkb
*lkb
;
3890 error
= find_lkb(ls
, rl
->rl_lkid
, &lkb
);
3892 log_error(ls
, "recover_process_copy no lkid %x", rl
->rl_lkid
);
3896 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
3898 error
= rl
->rl_result
;
3900 r
= lkb
->lkb_resource
;
3906 /* There's a chance the new master received our lock before
3907 dlm_recover_master_reply(), this wouldn't happen if we did
3908 a barrier between recover_masters and recover_locks. */
3909 log_debug(ls
, "master copy not ready %x r %lx %s", lkb
->lkb_id
,
3910 (unsigned long)r
, r
->res_name
);
3911 dlm_send_rcom_lock(r
, lkb
);
3914 log_debug(ls
, "master copy exists %x", lkb
->lkb_id
);
3917 lkb
->lkb_remid
= rl
->rl_remid
;
3920 log_error(ls
, "dlm_recover_process_copy unknown error %d %x",
3921 error
, lkb
->lkb_id
);
3924 /* an ack for dlm_recover_locks() which waits for replies from
3925 all the locks it sends to new masters */
3926 dlm_recovered_lock(r
);
3935 int dlm_user_request(struct dlm_ls
*ls
, struct dlm_user_args
*ua
,
3936 int mode
, uint32_t flags
, void *name
, unsigned int namelen
,
3937 uint32_t parent_lkid
)
3939 struct dlm_lkb
*lkb
;
3940 struct dlm_args args
;
3945 error
= create_lkb(ls
, &lkb
);
3951 if (flags
& DLM_LKF_VALBLK
) {
3952 ua
->lksb
.sb_lvbptr
= kzalloc(DLM_USER_LVB_LEN
, GFP_KERNEL
);
3953 if (!ua
->lksb
.sb_lvbptr
) {
3961 /* After ua is attached to lkb it will be freed by free_lkb().
3962 When DLM_IFL_USER is set, the dlm knows that this is a userspace
3963 lock and that lkb_astparam is the dlm_user_args structure. */
3965 error
= set_lock_args(mode
, &ua
->lksb
, flags
, namelen
, parent_lkid
,
3966 DLM_FAKE_USER_AST
, ua
, DLM_FAKE_USER_AST
, &args
);
3967 lkb
->lkb_flags
|= DLM_IFL_USER
;
3968 ua
->old_mode
= DLM_LOCK_IV
;
3975 error
= request_lock(ls
, lkb
, name
, namelen
, &args
);
3991 /* add this new lkb to the per-process list of locks */
3992 spin_lock(&ua
->proc
->locks_spin
);
3994 list_add_tail(&lkb
->lkb_ownqueue
, &ua
->proc
->locks
);
3995 spin_unlock(&ua
->proc
->locks_spin
);
3997 unlock_recovery(ls
);
4001 int dlm_user_convert(struct dlm_ls
*ls
, struct dlm_user_args
*ua_tmp
,
4002 int mode
, uint32_t flags
, uint32_t lkid
, char *lvb_in
)
4004 struct dlm_lkb
*lkb
;
4005 struct dlm_args args
;
4006 struct dlm_user_args
*ua
;
4011 error
= find_lkb(ls
, lkid
, &lkb
);
4015 /* user can change the params on its lock when it converts it, or
4016 add an lvb that didn't exist before */
4018 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4020 if (flags
& DLM_LKF_VALBLK
&& !ua
->lksb
.sb_lvbptr
) {
4021 ua
->lksb
.sb_lvbptr
= kzalloc(DLM_USER_LVB_LEN
, GFP_KERNEL
);
4022 if (!ua
->lksb
.sb_lvbptr
) {
4027 if (lvb_in
&& ua
->lksb
.sb_lvbptr
)
4028 memcpy(ua
->lksb
.sb_lvbptr
, lvb_in
, DLM_USER_LVB_LEN
);
4030 ua
->castparam
= ua_tmp
->castparam
;
4031 ua
->castaddr
= ua_tmp
->castaddr
;
4032 ua
->bastparam
= ua_tmp
->bastparam
;
4033 ua
->bastaddr
= ua_tmp
->bastaddr
;
4034 ua
->user_lksb
= ua_tmp
->user_lksb
;
4035 ua
->old_mode
= lkb
->lkb_grmode
;
4037 error
= set_lock_args(mode
, &ua
->lksb
, flags
, 0, 0, DLM_FAKE_USER_AST
,
4038 ua
, DLM_FAKE_USER_AST
, &args
);
4042 error
= convert_lock(ls
, lkb
, &args
);
4044 if (error
== -EINPROGRESS
|| error
== -EAGAIN
)
4049 unlock_recovery(ls
);
4054 int dlm_user_unlock(struct dlm_ls
*ls
, struct dlm_user_args
*ua_tmp
,
4055 uint32_t flags
, uint32_t lkid
, char *lvb_in
)
4057 struct dlm_lkb
*lkb
;
4058 struct dlm_args args
;
4059 struct dlm_user_args
*ua
;
4064 error
= find_lkb(ls
, lkid
, &lkb
);
4068 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4070 if (lvb_in
&& ua
->lksb
.sb_lvbptr
)
4071 memcpy(ua
->lksb
.sb_lvbptr
, lvb_in
, DLM_USER_LVB_LEN
);
4072 ua
->castparam
= ua_tmp
->castparam
;
4073 ua
->user_lksb
= ua_tmp
->user_lksb
;
4075 error
= set_unlock_args(flags
, ua
, &args
);
4079 error
= unlock_lock(ls
, lkb
, &args
);
4081 if (error
== -DLM_EUNLOCK
)
4083 /* from validate_unlock_args() */
4084 if (error
== -EBUSY
&& (flags
& DLM_LKF_FORCEUNLOCK
))
4089 spin_lock(&ua
->proc
->locks_spin
);
4090 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4091 if (!list_empty(&lkb
->lkb_ownqueue
))
4092 list_move(&lkb
->lkb_ownqueue
, &ua
->proc
->unlocking
);
4093 spin_unlock(&ua
->proc
->locks_spin
);
4097 unlock_recovery(ls
);
4102 int dlm_user_cancel(struct dlm_ls
*ls
, struct dlm_user_args
*ua_tmp
,
4103 uint32_t flags
, uint32_t lkid
)
4105 struct dlm_lkb
*lkb
;
4106 struct dlm_args args
;
4107 struct dlm_user_args
*ua
;
4112 error
= find_lkb(ls
, lkid
, &lkb
);
4116 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4117 ua
->castparam
= ua_tmp
->castparam
;
4118 ua
->user_lksb
= ua_tmp
->user_lksb
;
4120 error
= set_unlock_args(flags
, ua
, &args
);
4124 error
= cancel_lock(ls
, lkb
, &args
);
4126 if (error
== -DLM_ECANCEL
)
4128 /* from validate_unlock_args() */
4129 if (error
== -EBUSY
)
4134 unlock_recovery(ls
);
4139 /* lkb's that are removed from the waiters list by revert are just left on the
4140 orphans list with the granted orphan locks, to be freed by purge */
4142 static int orphan_proc_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
4144 struct dlm_user_args
*ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4145 struct dlm_args args
;
4149 mutex_lock(&ls
->ls_orphans_mutex
);
4150 list_add_tail(&lkb
->lkb_ownqueue
, &ls
->ls_orphans
);
4151 mutex_unlock(&ls
->ls_orphans_mutex
);
4153 set_unlock_args(0, ua
, &args
);
4155 error
= cancel_lock(ls
, lkb
, &args
);
4156 if (error
== -DLM_ECANCEL
)
4161 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4162 Regardless of what rsb queue the lock is on, it's removed and freed. */
4164 static int unlock_proc_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
4166 struct dlm_user_args
*ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4167 struct dlm_args args
;
4170 set_unlock_args(DLM_LKF_FORCEUNLOCK
, ua
, &args
);
4172 error
= unlock_lock(ls
, lkb
, &args
);
4173 if (error
== -DLM_EUNLOCK
)
4178 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4179 (which does lock_rsb) due to deadlock with receiving a message that does
4180 lock_rsb followed by dlm_user_add_ast() */
4182 static struct dlm_lkb
*del_proc_lock(struct dlm_ls
*ls
,
4183 struct dlm_user_proc
*proc
)
4185 struct dlm_lkb
*lkb
= NULL
;
4187 mutex_lock(&ls
->ls_clear_proc_locks
);
4188 if (list_empty(&proc
->locks
))
4191 lkb
= list_entry(proc
->locks
.next
, struct dlm_lkb
, lkb_ownqueue
);
4192 list_del_init(&lkb
->lkb_ownqueue
);
4194 if (lkb
->lkb_exflags
& DLM_LKF_PERSISTENT
)
4195 lkb
->lkb_flags
|= DLM_IFL_ORPHAN
;
4197 lkb
->lkb_flags
|= DLM_IFL_DEAD
;
4199 mutex_unlock(&ls
->ls_clear_proc_locks
);
4203 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4204 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4205 which we clear here. */
4207 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
4208 list, and no more device_writes should add lkb's to proc->locks list; so we
4209 shouldn't need to take asts_spin or locks_spin here. this assumes that
4210 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4213 void dlm_clear_proc_locks(struct dlm_ls
*ls
, struct dlm_user_proc
*proc
)
4215 struct dlm_lkb
*lkb
, *safe
;
4220 lkb
= del_proc_lock(ls
, proc
);
4223 if (lkb
->lkb_exflags
& DLM_LKF_PERSISTENT
)
4224 orphan_proc_lock(ls
, lkb
);
4226 unlock_proc_lock(ls
, lkb
);
4228 /* this removes the reference for the proc->locks list
4229 added by dlm_user_request, it may result in the lkb
4235 mutex_lock(&ls
->ls_clear_proc_locks
);
4237 /* in-progress unlocks */
4238 list_for_each_entry_safe(lkb
, safe
, &proc
->unlocking
, lkb_ownqueue
) {
4239 list_del_init(&lkb
->lkb_ownqueue
);
4240 lkb
->lkb_flags
|= DLM_IFL_DEAD
;
4244 list_for_each_entry_safe(lkb
, safe
, &proc
->asts
, lkb_astqueue
) {
4245 list_del(&lkb
->lkb_astqueue
);
4249 mutex_unlock(&ls
->ls_clear_proc_locks
);
4250 unlock_recovery(ls
);