1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005-2008 Red Hat, Inc. All rights reserved.
6 ** This copyrighted material is made available to anyone wishing to use,
7 ** modify, copy, or redistribute it subject to the terms and conditions
8 ** of the GNU General Public License v.2.
10 *******************************************************************************
11 ******************************************************************************/
13 /* Central locking logic has four stages:
33 Stage 1 (lock, unlock) is mainly about checking input args and
34 splitting into one of the four main operations:
36 dlm_lock = request_lock
37 dlm_lock+CONVERT = convert_lock
38 dlm_unlock = unlock_lock
39 dlm_unlock+CANCEL = cancel_lock
41 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
42 provided to the next stage.
44 Stage 3, _xxxx_lock(), determines if the operation is local or remote.
45 When remote, it calls send_xxxx(), when local it calls do_xxxx().
47 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the
48 given rsb and lkb and queues callbacks.
50 For remote operations, send_xxxx() results in the corresponding do_xxxx()
51 function being executed on the remote node. The connecting send/receive
52 calls on local (L) and remote (R) nodes:
54 L: send_xxxx() -> R: receive_xxxx()
56 L: receive_xxxx_reply() <- R: send_xxxx_reply()
58 #include <linux/types.h>
59 #include "dlm_internal.h"
60 #include <linux/dlm_device.h>
63 #include "requestqueue.h"
67 #include "lockspace.h"
72 #include "lvb_table.h"
76 static int send_request(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
77 static int send_convert(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
78 static int send_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
79 static int send_cancel(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
80 static int send_grant(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
81 static int send_bast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int mode
);
82 static int send_lookup(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
83 static int send_remove(struct dlm_rsb
*r
);
84 static int _request_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
85 static int _cancel_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
);
86 static void __receive_convert_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
87 struct dlm_message
*ms
);
88 static int receive_extralen(struct dlm_message
*ms
);
89 static void do_purge(struct dlm_ls
*ls
, int nodeid
, int pid
);
90 static void del_timeout(struct dlm_lkb
*lkb
);
93 * Lock compatibilty matrix - thanks Steve
94 * UN = Unlocked state. Not really a state, used as a flag
95 * PD = Padding. Used to make the matrix a nice power of two in size
96 * Other states are the same as the VMS DLM.
97 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same)
100 static const int __dlm_compat_matrix
[8][8] = {
101 /* UN NL CR CW PR PW EX PD */
102 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */
103 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */
104 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */
105 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */
106 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */
107 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */
108 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */
109 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
113 * This defines the direction of transfer of LVB data.
114 * Granted mode is the row; requested mode is the column.
115 * Usage: matrix[grmode+1][rqmode+1]
116 * 1 = LVB is returned to the caller
117 * 0 = LVB is written to the resource
118 * -1 = nothing happens to the LVB
121 const int dlm_lvb_operations
[8][8] = {
122 /* UN NL CR CW PR PW EX PD*/
123 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */
124 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */
125 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */
126 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */
127 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */
128 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */
129 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */
130 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */
133 #define modes_compat(gr, rq) \
134 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
136 int dlm_modes_compat(int mode1
, int mode2
)
138 return __dlm_compat_matrix
[mode1
+ 1][mode2
+ 1];
142 * Compatibility matrix for conversions with QUECVT set.
143 * Granted mode is the row; requested mode is the column.
144 * Usage: matrix[grmode+1][rqmode+1]
147 static const int __quecvt_compat_matrix
[8][8] = {
148 /* UN NL CR CW PR PW EX PD */
149 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */
150 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */
151 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */
152 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */
153 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */
154 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */
155 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */
156 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */
159 void dlm_print_lkb(struct dlm_lkb
*lkb
)
161 printk(KERN_ERR
"lkb: nodeid %d id %x remid %x exflags %x flags %x\n"
162 " status %d rqmode %d grmode %d wait_type %d ast_type %d\n",
163 lkb
->lkb_nodeid
, lkb
->lkb_id
, lkb
->lkb_remid
, lkb
->lkb_exflags
,
164 lkb
->lkb_flags
, lkb
->lkb_status
, lkb
->lkb_rqmode
,
165 lkb
->lkb_grmode
, lkb
->lkb_wait_type
, lkb
->lkb_ast_type
);
168 void dlm_print_rsb(struct dlm_rsb
*r
)
170 printk(KERN_ERR
"rsb: nodeid %d flags %lx first %x rlc %d name %s\n",
171 r
->res_nodeid
, r
->res_flags
, r
->res_first_lkid
,
172 r
->res_recover_locks_count
, r
->res_name
);
175 void dlm_dump_rsb(struct dlm_rsb
*r
)
181 printk(KERN_ERR
"rsb: root_list empty %d recover_list empty %d\n",
182 list_empty(&r
->res_root_list
), list_empty(&r
->res_recover_list
));
183 printk(KERN_ERR
"rsb lookup list\n");
184 list_for_each_entry(lkb
, &r
->res_lookup
, lkb_rsb_lookup
)
186 printk(KERN_ERR
"rsb grant queue:\n");
187 list_for_each_entry(lkb
, &r
->res_grantqueue
, lkb_statequeue
)
189 printk(KERN_ERR
"rsb convert queue:\n");
190 list_for_each_entry(lkb
, &r
->res_convertqueue
, lkb_statequeue
)
192 printk(KERN_ERR
"rsb wait queue:\n");
193 list_for_each_entry(lkb
, &r
->res_waitqueue
, lkb_statequeue
)
197 /* Threads cannot use the lockspace while it's being recovered */
199 static inline void dlm_lock_recovery(struct dlm_ls
*ls
)
201 down_read(&ls
->ls_in_recovery
);
204 void dlm_unlock_recovery(struct dlm_ls
*ls
)
206 up_read(&ls
->ls_in_recovery
);
209 int dlm_lock_recovery_try(struct dlm_ls
*ls
)
211 return down_read_trylock(&ls
->ls_in_recovery
);
214 static inline int can_be_queued(struct dlm_lkb
*lkb
)
216 return !(lkb
->lkb_exflags
& DLM_LKF_NOQUEUE
);
219 static inline int force_blocking_asts(struct dlm_lkb
*lkb
)
221 return (lkb
->lkb_exflags
& DLM_LKF_NOQUEUEBAST
);
224 static inline int is_demoted(struct dlm_lkb
*lkb
)
226 return (lkb
->lkb_sbflags
& DLM_SBF_DEMOTED
);
229 static inline int is_altmode(struct dlm_lkb
*lkb
)
231 return (lkb
->lkb_sbflags
& DLM_SBF_ALTMODE
);
234 static inline int is_granted(struct dlm_lkb
*lkb
)
236 return (lkb
->lkb_status
== DLM_LKSTS_GRANTED
);
239 static inline int is_remote(struct dlm_rsb
*r
)
241 DLM_ASSERT(r
->res_nodeid
>= 0, dlm_print_rsb(r
););
242 return !!r
->res_nodeid
;
245 static inline int is_process_copy(struct dlm_lkb
*lkb
)
247 return (lkb
->lkb_nodeid
&& !(lkb
->lkb_flags
& DLM_IFL_MSTCPY
));
250 static inline int is_master_copy(struct dlm_lkb
*lkb
)
252 if (lkb
->lkb_flags
& DLM_IFL_MSTCPY
)
253 DLM_ASSERT(lkb
->lkb_nodeid
, dlm_print_lkb(lkb
););
254 return (lkb
->lkb_flags
& DLM_IFL_MSTCPY
) ? 1 : 0;
257 static inline int middle_conversion(struct dlm_lkb
*lkb
)
259 if ((lkb
->lkb_grmode
==DLM_LOCK_PR
&& lkb
->lkb_rqmode
==DLM_LOCK_CW
) ||
260 (lkb
->lkb_rqmode
==DLM_LOCK_PR
&& lkb
->lkb_grmode
==DLM_LOCK_CW
))
265 static inline int down_conversion(struct dlm_lkb
*lkb
)
267 return (!middle_conversion(lkb
) && lkb
->lkb_rqmode
< lkb
->lkb_grmode
);
270 static inline int is_overlap_unlock(struct dlm_lkb
*lkb
)
272 return lkb
->lkb_flags
& DLM_IFL_OVERLAP_UNLOCK
;
275 static inline int is_overlap_cancel(struct dlm_lkb
*lkb
)
277 return lkb
->lkb_flags
& DLM_IFL_OVERLAP_CANCEL
;
280 static inline int is_overlap(struct dlm_lkb
*lkb
)
282 return (lkb
->lkb_flags
& (DLM_IFL_OVERLAP_UNLOCK
|
283 DLM_IFL_OVERLAP_CANCEL
));
286 static void queue_cast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
288 if (is_master_copy(lkb
))
293 DLM_ASSERT(lkb
->lkb_lksb
, dlm_print_lkb(lkb
););
295 /* if the operation was a cancel, then return -DLM_ECANCEL, if a
296 timeout caused the cancel then return -ETIMEDOUT */
297 if (rv
== -DLM_ECANCEL
&& (lkb
->lkb_flags
& DLM_IFL_TIMEOUT_CANCEL
)) {
298 lkb
->lkb_flags
&= ~DLM_IFL_TIMEOUT_CANCEL
;
302 if (rv
== -DLM_ECANCEL
&& (lkb
->lkb_flags
& DLM_IFL_DEADLOCK_CANCEL
)) {
303 lkb
->lkb_flags
&= ~DLM_IFL_DEADLOCK_CANCEL
;
307 lkb
->lkb_lksb
->sb_status
= rv
;
308 lkb
->lkb_lksb
->sb_flags
= lkb
->lkb_sbflags
;
310 dlm_add_ast(lkb
, AST_COMP
);
313 static inline void queue_cast_overlap(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
316 is_overlap_unlock(lkb
) ? -DLM_EUNLOCK
: -DLM_ECANCEL
);
319 static void queue_bast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rqmode
)
321 if (is_master_copy(lkb
))
322 send_bast(r
, lkb
, rqmode
);
324 lkb
->lkb_bastmode
= rqmode
;
325 dlm_add_ast(lkb
, AST_BAST
);
330 * Basic operations on rsb's and lkb's
333 static struct dlm_rsb
*create_rsb(struct dlm_ls
*ls
, char *name
, int len
)
337 r
= dlm_allocate_rsb(ls
, len
);
343 memcpy(r
->res_name
, name
, len
);
344 mutex_init(&r
->res_mutex
);
346 INIT_LIST_HEAD(&r
->res_lookup
);
347 INIT_LIST_HEAD(&r
->res_grantqueue
);
348 INIT_LIST_HEAD(&r
->res_convertqueue
);
349 INIT_LIST_HEAD(&r
->res_waitqueue
);
350 INIT_LIST_HEAD(&r
->res_root_list
);
351 INIT_LIST_HEAD(&r
->res_recover_list
);
356 static int search_rsb_list(struct list_head
*head
, char *name
, int len
,
357 unsigned int flags
, struct dlm_rsb
**r_ret
)
362 list_for_each_entry(r
, head
, res_hashchain
) {
363 if (len
== r
->res_length
&& !memcmp(name
, r
->res_name
, len
))
369 if (r
->res_nodeid
&& (flags
& R_MASTER
))
375 static int _search_rsb(struct dlm_ls
*ls
, char *name
, int len
, int b
,
376 unsigned int flags
, struct dlm_rsb
**r_ret
)
381 error
= search_rsb_list(&ls
->ls_rsbtbl
[b
].list
, name
, len
, flags
, &r
);
383 kref_get(&r
->res_ref
);
386 error
= search_rsb_list(&ls
->ls_rsbtbl
[b
].toss
, name
, len
, flags
, &r
);
390 list_move(&r
->res_hashchain
, &ls
->ls_rsbtbl
[b
].list
);
392 if (dlm_no_directory(ls
))
395 if (r
->res_nodeid
== -1) {
396 rsb_clear_flag(r
, RSB_MASTER_UNCERTAIN
);
397 r
->res_first_lkid
= 0;
398 } else if (r
->res_nodeid
> 0) {
399 rsb_set_flag(r
, RSB_MASTER_UNCERTAIN
);
400 r
->res_first_lkid
= 0;
402 DLM_ASSERT(r
->res_nodeid
== 0, dlm_print_rsb(r
););
403 DLM_ASSERT(!rsb_flag(r
, RSB_MASTER_UNCERTAIN
),);
410 static int search_rsb(struct dlm_ls
*ls
, char *name
, int len
, int b
,
411 unsigned int flags
, struct dlm_rsb
**r_ret
)
414 write_lock(&ls
->ls_rsbtbl
[b
].lock
);
415 error
= _search_rsb(ls
, name
, len
, b
, flags
, r_ret
);
416 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
421 * Find rsb in rsbtbl and potentially create/add one
423 * Delaying the release of rsb's has a similar benefit to applications keeping
424 * NL locks on an rsb, but without the guarantee that the cached master value
425 * will still be valid when the rsb is reused. Apps aren't always smart enough
426 * to keep NL locks on an rsb that they may lock again shortly; this can lead
427 * to excessive master lookups and removals if we don't delay the release.
429 * Searching for an rsb means looking through both the normal list and toss
430 * list. When found on the toss list the rsb is moved to the normal list with
431 * ref count of 1; when found on normal list the ref count is incremented.
434 static int find_rsb(struct dlm_ls
*ls
, char *name
, int namelen
,
435 unsigned int flags
, struct dlm_rsb
**r_ret
)
437 struct dlm_rsb
*r
, *tmp
;
438 uint32_t hash
, bucket
;
441 if (dlm_no_directory(ls
))
444 hash
= jhash(name
, namelen
, 0);
445 bucket
= hash
& (ls
->ls_rsbtbl_size
- 1);
447 error
= search_rsb(ls
, name
, namelen
, bucket
, flags
, &r
);
451 if (error
== -EBADR
&& !(flags
& R_CREATE
))
454 /* the rsb was found but wasn't a master copy */
455 if (error
== -ENOTBLK
)
459 r
= create_rsb(ls
, name
, namelen
);
464 r
->res_bucket
= bucket
;
466 kref_init(&r
->res_ref
);
468 /* With no directory, the master can be set immediately */
469 if (dlm_no_directory(ls
)) {
470 int nodeid
= dlm_dir_nodeid(r
);
471 if (nodeid
== dlm_our_nodeid())
473 r
->res_nodeid
= nodeid
;
476 write_lock(&ls
->ls_rsbtbl
[bucket
].lock
);
477 error
= _search_rsb(ls
, name
, namelen
, bucket
, 0, &tmp
);
479 write_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
484 list_add(&r
->res_hashchain
, &ls
->ls_rsbtbl
[bucket
].list
);
485 write_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
492 int dlm_find_rsb(struct dlm_ls
*ls
, char *name
, int namelen
,
493 unsigned int flags
, struct dlm_rsb
**r_ret
)
495 return find_rsb(ls
, name
, namelen
, flags
, r_ret
);
498 /* This is only called to add a reference when the code already holds
499 a valid reference to the rsb, so there's no need for locking. */
501 static inline void hold_rsb(struct dlm_rsb
*r
)
503 kref_get(&r
->res_ref
);
506 void dlm_hold_rsb(struct dlm_rsb
*r
)
511 static void toss_rsb(struct kref
*kref
)
513 struct dlm_rsb
*r
= container_of(kref
, struct dlm_rsb
, res_ref
);
514 struct dlm_ls
*ls
= r
->res_ls
;
516 DLM_ASSERT(list_empty(&r
->res_root_list
), dlm_print_rsb(r
););
517 kref_init(&r
->res_ref
);
518 list_move(&r
->res_hashchain
, &ls
->ls_rsbtbl
[r
->res_bucket
].toss
);
519 r
->res_toss_time
= jiffies
;
521 dlm_free_lvb(r
->res_lvbptr
);
522 r
->res_lvbptr
= NULL
;
526 /* When all references to the rsb are gone it's transfered to
527 the tossed list for later disposal. */
529 static void put_rsb(struct dlm_rsb
*r
)
531 struct dlm_ls
*ls
= r
->res_ls
;
532 uint32_t bucket
= r
->res_bucket
;
534 write_lock(&ls
->ls_rsbtbl
[bucket
].lock
);
535 kref_put(&r
->res_ref
, toss_rsb
);
536 write_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
539 void dlm_put_rsb(struct dlm_rsb
*r
)
544 /* See comment for unhold_lkb */
546 static void unhold_rsb(struct dlm_rsb
*r
)
549 rv
= kref_put(&r
->res_ref
, toss_rsb
);
550 DLM_ASSERT(!rv
, dlm_dump_rsb(r
););
553 static void kill_rsb(struct kref
*kref
)
555 struct dlm_rsb
*r
= container_of(kref
, struct dlm_rsb
, res_ref
);
557 /* All work is done after the return from kref_put() so we
558 can release the write_lock before the remove and free. */
560 DLM_ASSERT(list_empty(&r
->res_lookup
), dlm_dump_rsb(r
););
561 DLM_ASSERT(list_empty(&r
->res_grantqueue
), dlm_dump_rsb(r
););
562 DLM_ASSERT(list_empty(&r
->res_convertqueue
), dlm_dump_rsb(r
););
563 DLM_ASSERT(list_empty(&r
->res_waitqueue
), dlm_dump_rsb(r
););
564 DLM_ASSERT(list_empty(&r
->res_root_list
), dlm_dump_rsb(r
););
565 DLM_ASSERT(list_empty(&r
->res_recover_list
), dlm_dump_rsb(r
););
568 /* Attaching/detaching lkb's from rsb's is for rsb reference counting.
569 The rsb must exist as long as any lkb's for it do. */
571 static void attach_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
574 lkb
->lkb_resource
= r
;
577 static void detach_lkb(struct dlm_lkb
*lkb
)
579 if (lkb
->lkb_resource
) {
580 put_rsb(lkb
->lkb_resource
);
581 lkb
->lkb_resource
= NULL
;
585 static int create_lkb(struct dlm_ls
*ls
, struct dlm_lkb
**lkb_ret
)
587 struct dlm_lkb
*lkb
, *tmp
;
591 lkb
= dlm_allocate_lkb(ls
);
595 lkb
->lkb_nodeid
= -1;
596 lkb
->lkb_grmode
= DLM_LOCK_IV
;
597 kref_init(&lkb
->lkb_ref
);
598 INIT_LIST_HEAD(&lkb
->lkb_ownqueue
);
599 INIT_LIST_HEAD(&lkb
->lkb_rsb_lookup
);
600 INIT_LIST_HEAD(&lkb
->lkb_time_list
);
602 get_random_bytes(&bucket
, sizeof(bucket
));
603 bucket
&= (ls
->ls_lkbtbl_size
- 1);
605 write_lock(&ls
->ls_lkbtbl
[bucket
].lock
);
607 /* counter can roll over so we must verify lkid is not in use */
610 lkid
= (bucket
<< 16) | ls
->ls_lkbtbl
[bucket
].counter
++;
612 list_for_each_entry(tmp
, &ls
->ls_lkbtbl
[bucket
].list
,
614 if (tmp
->lkb_id
!= lkid
)
622 list_add(&lkb
->lkb_idtbl_list
, &ls
->ls_lkbtbl
[bucket
].list
);
623 write_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
629 static struct dlm_lkb
*__find_lkb(struct dlm_ls
*ls
, uint32_t lkid
)
632 uint16_t bucket
= (lkid
>> 16);
634 list_for_each_entry(lkb
, &ls
->ls_lkbtbl
[bucket
].list
, lkb_idtbl_list
) {
635 if (lkb
->lkb_id
== lkid
)
641 static int find_lkb(struct dlm_ls
*ls
, uint32_t lkid
, struct dlm_lkb
**lkb_ret
)
644 uint16_t bucket
= (lkid
>> 16);
646 if (bucket
>= ls
->ls_lkbtbl_size
)
649 read_lock(&ls
->ls_lkbtbl
[bucket
].lock
);
650 lkb
= __find_lkb(ls
, lkid
);
652 kref_get(&lkb
->lkb_ref
);
653 read_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
656 return lkb
? 0 : -ENOENT
;
659 static void kill_lkb(struct kref
*kref
)
661 struct dlm_lkb
*lkb
= container_of(kref
, struct dlm_lkb
, lkb_ref
);
663 /* All work is done after the return from kref_put() so we
664 can release the write_lock before the detach_lkb */
666 DLM_ASSERT(!lkb
->lkb_status
, dlm_print_lkb(lkb
););
669 /* __put_lkb() is used when an lkb may not have an rsb attached to
670 it so we need to provide the lockspace explicitly */
672 static int __put_lkb(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
674 uint16_t bucket
= (lkb
->lkb_id
>> 16);
676 write_lock(&ls
->ls_lkbtbl
[bucket
].lock
);
677 if (kref_put(&lkb
->lkb_ref
, kill_lkb
)) {
678 list_del(&lkb
->lkb_idtbl_list
);
679 write_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
683 /* for local/process lkbs, lvbptr points to caller's lksb */
684 if (lkb
->lkb_lvbptr
&& is_master_copy(lkb
))
685 dlm_free_lvb(lkb
->lkb_lvbptr
);
689 write_unlock(&ls
->ls_lkbtbl
[bucket
].lock
);
694 int dlm_put_lkb(struct dlm_lkb
*lkb
)
698 DLM_ASSERT(lkb
->lkb_resource
, dlm_print_lkb(lkb
););
699 DLM_ASSERT(lkb
->lkb_resource
->res_ls
, dlm_print_lkb(lkb
););
701 ls
= lkb
->lkb_resource
->res_ls
;
702 return __put_lkb(ls
, lkb
);
705 /* This is only called to add a reference when the code already holds
706 a valid reference to the lkb, so there's no need for locking. */
708 static inline void hold_lkb(struct dlm_lkb
*lkb
)
710 kref_get(&lkb
->lkb_ref
);
713 /* This is called when we need to remove a reference and are certain
714 it's not the last ref. e.g. del_lkb is always called between a
715 find_lkb/put_lkb and is always the inverse of a previous add_lkb.
716 put_lkb would work fine, but would involve unnecessary locking */
718 static inline void unhold_lkb(struct dlm_lkb
*lkb
)
721 rv
= kref_put(&lkb
->lkb_ref
, kill_lkb
);
722 DLM_ASSERT(!rv
, dlm_print_lkb(lkb
););
725 static void lkb_add_ordered(struct list_head
*new, struct list_head
*head
,
728 struct dlm_lkb
*lkb
= NULL
;
730 list_for_each_entry(lkb
, head
, lkb_statequeue
)
731 if (lkb
->lkb_rqmode
< mode
)
735 list_add_tail(new, head
);
737 __list_add(new, lkb
->lkb_statequeue
.prev
, &lkb
->lkb_statequeue
);
740 /* add/remove lkb to rsb's grant/convert/wait queue */
742 static void add_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int status
)
744 kref_get(&lkb
->lkb_ref
);
746 DLM_ASSERT(!lkb
->lkb_status
, dlm_print_lkb(lkb
););
748 lkb
->lkb_status
= status
;
751 case DLM_LKSTS_WAITING
:
752 if (lkb
->lkb_exflags
& DLM_LKF_HEADQUE
)
753 list_add(&lkb
->lkb_statequeue
, &r
->res_waitqueue
);
755 list_add_tail(&lkb
->lkb_statequeue
, &r
->res_waitqueue
);
757 case DLM_LKSTS_GRANTED
:
758 /* convention says granted locks kept in order of grmode */
759 lkb_add_ordered(&lkb
->lkb_statequeue
, &r
->res_grantqueue
,
762 case DLM_LKSTS_CONVERT
:
763 if (lkb
->lkb_exflags
& DLM_LKF_HEADQUE
)
764 list_add(&lkb
->lkb_statequeue
, &r
->res_convertqueue
);
766 list_add_tail(&lkb
->lkb_statequeue
,
767 &r
->res_convertqueue
);
770 DLM_ASSERT(0, dlm_print_lkb(lkb
); printk("sts=%d\n", status
););
774 static void del_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
777 list_del(&lkb
->lkb_statequeue
);
781 static void move_lkb(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int sts
)
785 add_lkb(r
, lkb
, sts
);
789 static int msg_reply_type(int mstype
)
792 case DLM_MSG_REQUEST
:
793 return DLM_MSG_REQUEST_REPLY
;
794 case DLM_MSG_CONVERT
:
795 return DLM_MSG_CONVERT_REPLY
;
797 return DLM_MSG_UNLOCK_REPLY
;
799 return DLM_MSG_CANCEL_REPLY
;
801 return DLM_MSG_LOOKUP_REPLY
;
806 /* add/remove lkb from global waiters list of lkb's waiting for
807 a reply from a remote node */
809 static int add_to_waiters(struct dlm_lkb
*lkb
, int mstype
)
811 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
814 mutex_lock(&ls
->ls_waiters_mutex
);
816 if (is_overlap_unlock(lkb
) ||
817 (is_overlap_cancel(lkb
) && (mstype
== DLM_MSG_CANCEL
))) {
822 if (lkb
->lkb_wait_type
|| is_overlap_cancel(lkb
)) {
825 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_UNLOCK
;
828 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_CANCEL
;
834 lkb
->lkb_wait_count
++;
837 log_debug(ls
, "add overlap %x cur %d new %d count %d flags %x",
838 lkb
->lkb_id
, lkb
->lkb_wait_type
, mstype
,
839 lkb
->lkb_wait_count
, lkb
->lkb_flags
);
843 DLM_ASSERT(!lkb
->lkb_wait_count
,
845 printk("wait_count %d\n", lkb
->lkb_wait_count
););
847 lkb
->lkb_wait_count
++;
848 lkb
->lkb_wait_type
= mstype
;
850 list_add(&lkb
->lkb_wait_reply
, &ls
->ls_waiters
);
853 log_error(ls
, "add_to_waiters %x error %d flags %x %d %d %s",
854 lkb
->lkb_id
, error
, lkb
->lkb_flags
, mstype
,
855 lkb
->lkb_wait_type
, lkb
->lkb_resource
->res_name
);
856 mutex_unlock(&ls
->ls_waiters_mutex
);
860 /* We clear the RESEND flag because we might be taking an lkb off the waiters
861 list as part of process_requestqueue (e.g. a lookup that has an optimized
862 request reply on the requestqueue) between dlm_recover_waiters_pre() which
863 set RESEND and dlm_recover_waiters_post() */
865 static int _remove_from_waiters(struct dlm_lkb
*lkb
, int mstype
)
867 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
868 int overlap_done
= 0;
870 if (is_overlap_unlock(lkb
) && (mstype
== DLM_MSG_UNLOCK_REPLY
)) {
871 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
876 if (is_overlap_cancel(lkb
) && (mstype
== DLM_MSG_CANCEL_REPLY
)) {
877 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
882 /* N.B. type of reply may not always correspond to type of original
883 msg due to lookup->request optimization, verify others? */
885 if (lkb
->lkb_wait_type
) {
886 lkb
->lkb_wait_type
= 0;
890 log_error(ls
, "remove_from_waiters lkid %x flags %x types %d %d",
891 lkb
->lkb_id
, lkb
->lkb_flags
, mstype
, lkb
->lkb_wait_type
);
895 /* the force-unlock/cancel has completed and we haven't recvd a reply
896 to the op that was in progress prior to the unlock/cancel; we
897 give up on any reply to the earlier op. FIXME: not sure when/how
900 if (overlap_done
&& lkb
->lkb_wait_type
) {
901 log_error(ls
, "remove_from_waiters %x reply %d give up on %d",
902 lkb
->lkb_id
, mstype
, lkb
->lkb_wait_type
);
903 lkb
->lkb_wait_count
--;
904 lkb
->lkb_wait_type
= 0;
907 DLM_ASSERT(lkb
->lkb_wait_count
, dlm_print_lkb(lkb
););
909 lkb
->lkb_flags
&= ~DLM_IFL_RESEND
;
910 lkb
->lkb_wait_count
--;
911 if (!lkb
->lkb_wait_count
)
912 list_del_init(&lkb
->lkb_wait_reply
);
917 static int remove_from_waiters(struct dlm_lkb
*lkb
, int mstype
)
919 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
922 mutex_lock(&ls
->ls_waiters_mutex
);
923 error
= _remove_from_waiters(lkb
, mstype
);
924 mutex_unlock(&ls
->ls_waiters_mutex
);
928 /* Handles situations where we might be processing a "fake" or "stub" reply in
929 which we can't try to take waiters_mutex again. */
931 static int remove_from_waiters_ms(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
933 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
936 if (ms
!= &ls
->ls_stub_ms
)
937 mutex_lock(&ls
->ls_waiters_mutex
);
938 error
= _remove_from_waiters(lkb
, ms
->m_type
);
939 if (ms
!= &ls
->ls_stub_ms
)
940 mutex_unlock(&ls
->ls_waiters_mutex
);
944 static void dir_remove(struct dlm_rsb
*r
)
948 if (dlm_no_directory(r
->res_ls
))
951 to_nodeid
= dlm_dir_nodeid(r
);
952 if (to_nodeid
!= dlm_our_nodeid())
955 dlm_dir_remove_entry(r
->res_ls
, to_nodeid
,
956 r
->res_name
, r
->res_length
);
959 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is
960 found since they are in order of newest to oldest? */
962 static int shrink_bucket(struct dlm_ls
*ls
, int b
)
965 int count
= 0, found
;
969 write_lock(&ls
->ls_rsbtbl
[b
].lock
);
970 list_for_each_entry_reverse(r
, &ls
->ls_rsbtbl
[b
].toss
,
972 if (!time_after_eq(jiffies
, r
->res_toss_time
+
973 dlm_config
.ci_toss_secs
* HZ
))
980 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
984 if (kref_put(&r
->res_ref
, kill_rsb
)) {
985 list_del(&r
->res_hashchain
);
986 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
993 write_unlock(&ls
->ls_rsbtbl
[b
].lock
);
994 log_error(ls
, "tossed rsb in use %s", r
->res_name
);
1001 void dlm_scan_rsbs(struct dlm_ls
*ls
)
1005 for (i
= 0; i
< ls
->ls_rsbtbl_size
; i
++) {
1006 shrink_bucket(ls
, i
);
1007 if (dlm_locking_stopped(ls
))
1013 static void add_timeout(struct dlm_lkb
*lkb
)
1015 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
1017 if (is_master_copy(lkb
)) {
1018 lkb
->lkb_timestamp
= jiffies
;
1022 if (test_bit(LSFL_TIMEWARN
, &ls
->ls_flags
) &&
1023 !(lkb
->lkb_exflags
& DLM_LKF_NODLCKWT
)) {
1024 lkb
->lkb_flags
|= DLM_IFL_WATCH_TIMEWARN
;
1027 if (lkb
->lkb_exflags
& DLM_LKF_TIMEOUT
)
1032 DLM_ASSERT(list_empty(&lkb
->lkb_time_list
), dlm_print_lkb(lkb
););
1033 mutex_lock(&ls
->ls_timeout_mutex
);
1035 lkb
->lkb_timestamp
= jiffies
;
1036 list_add_tail(&lkb
->lkb_time_list
, &ls
->ls_timeout
);
1037 mutex_unlock(&ls
->ls_timeout_mutex
);
1040 static void del_timeout(struct dlm_lkb
*lkb
)
1042 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
1044 mutex_lock(&ls
->ls_timeout_mutex
);
1045 if (!list_empty(&lkb
->lkb_time_list
)) {
1046 list_del_init(&lkb
->lkb_time_list
);
1049 mutex_unlock(&ls
->ls_timeout_mutex
);
1052 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1053 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex
1054 and then lock rsb because of lock ordering in add_timeout. We may need
1055 to specify some special timeout-related bits in the lkb that are just to
1056 be accessed under the timeout_mutex. */
1058 void dlm_scan_timeout(struct dlm_ls
*ls
)
1061 struct dlm_lkb
*lkb
;
1062 int do_cancel
, do_warn
;
1065 if (dlm_locking_stopped(ls
))
1070 mutex_lock(&ls
->ls_timeout_mutex
);
1071 list_for_each_entry(lkb
, &ls
->ls_timeout
, lkb_time_list
) {
1073 if ((lkb
->lkb_exflags
& DLM_LKF_TIMEOUT
) &&
1074 time_after_eq(jiffies
, lkb
->lkb_timestamp
+
1075 lkb
->lkb_timeout_cs
* HZ
/100))
1078 if ((lkb
->lkb_flags
& DLM_IFL_WATCH_TIMEWARN
) &&
1079 time_after_eq(jiffies
, lkb
->lkb_timestamp
+
1080 dlm_config
.ci_timewarn_cs
* HZ
/100))
1083 if (!do_cancel
&& !do_warn
)
1088 mutex_unlock(&ls
->ls_timeout_mutex
);
1090 if (!do_cancel
&& !do_warn
)
1093 r
= lkb
->lkb_resource
;
1098 /* clear flag so we only warn once */
1099 lkb
->lkb_flags
&= ~DLM_IFL_WATCH_TIMEWARN
;
1100 if (!(lkb
->lkb_exflags
& DLM_LKF_TIMEOUT
))
1102 dlm_timeout_warn(lkb
);
1106 log_debug(ls
, "timeout cancel %x node %d %s",
1107 lkb
->lkb_id
, lkb
->lkb_nodeid
, r
->res_name
);
1108 lkb
->lkb_flags
&= ~DLM_IFL_WATCH_TIMEWARN
;
1109 lkb
->lkb_flags
|= DLM_IFL_TIMEOUT_CANCEL
;
1111 _cancel_lock(r
, lkb
);
1120 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1121 dlm_recoverd before checking/setting ls_recover_begin. */
1123 void dlm_adjust_timeouts(struct dlm_ls
*ls
)
1125 struct dlm_lkb
*lkb
;
1126 long adj
= jiffies
- ls
->ls_recover_begin
;
1128 ls
->ls_recover_begin
= 0;
1129 mutex_lock(&ls
->ls_timeout_mutex
);
1130 list_for_each_entry(lkb
, &ls
->ls_timeout
, lkb_time_list
)
1131 lkb
->lkb_timestamp
+= adj
;
1132 mutex_unlock(&ls
->ls_timeout_mutex
);
1135 /* lkb is master or local copy */
1137 static void set_lvb_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1139 int b
, len
= r
->res_ls
->ls_lvblen
;
1141 /* b=1 lvb returned to caller
1142 b=0 lvb written to rsb or invalidated
1145 b
= dlm_lvb_operations
[lkb
->lkb_grmode
+ 1][lkb
->lkb_rqmode
+ 1];
1148 if (!lkb
->lkb_lvbptr
)
1151 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1157 memcpy(lkb
->lkb_lvbptr
, r
->res_lvbptr
, len
);
1158 lkb
->lkb_lvbseq
= r
->res_lvbseq
;
1160 } else if (b
== 0) {
1161 if (lkb
->lkb_exflags
& DLM_LKF_IVVALBLK
) {
1162 rsb_set_flag(r
, RSB_VALNOTVALID
);
1166 if (!lkb
->lkb_lvbptr
)
1169 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1173 r
->res_lvbptr
= dlm_allocate_lvb(r
->res_ls
);
1178 memcpy(r
->res_lvbptr
, lkb
->lkb_lvbptr
, len
);
1180 lkb
->lkb_lvbseq
= r
->res_lvbseq
;
1181 rsb_clear_flag(r
, RSB_VALNOTVALID
);
1184 if (rsb_flag(r
, RSB_VALNOTVALID
))
1185 lkb
->lkb_sbflags
|= DLM_SBF_VALNOTVALID
;
1188 static void set_lvb_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1190 if (lkb
->lkb_grmode
< DLM_LOCK_PW
)
1193 if (lkb
->lkb_exflags
& DLM_LKF_IVVALBLK
) {
1194 rsb_set_flag(r
, RSB_VALNOTVALID
);
1198 if (!lkb
->lkb_lvbptr
)
1201 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1205 r
->res_lvbptr
= dlm_allocate_lvb(r
->res_ls
);
1210 memcpy(r
->res_lvbptr
, lkb
->lkb_lvbptr
, r
->res_ls
->ls_lvblen
);
1212 rsb_clear_flag(r
, RSB_VALNOTVALID
);
1215 /* lkb is process copy (pc) */
1217 static void set_lvb_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
1218 struct dlm_message
*ms
)
1222 if (!lkb
->lkb_lvbptr
)
1225 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
1228 b
= dlm_lvb_operations
[lkb
->lkb_grmode
+ 1][lkb
->lkb_rqmode
+ 1];
1230 int len
= receive_extralen(ms
);
1231 memcpy(lkb
->lkb_lvbptr
, ms
->m_extra
, len
);
1232 lkb
->lkb_lvbseq
= ms
->m_lvbseq
;
1236 /* Manipulate lkb's on rsb's convert/granted/waiting queues
1237 remove_lock -- used for unlock, removes lkb from granted
1238 revert_lock -- used for cancel, moves lkb from convert to granted
1239 grant_lock -- used for request and convert, adds lkb to granted or
1240 moves lkb from convert or waiting to granted
1242 Each of these is used for master or local copy lkb's. There is
1243 also a _pc() variation used to make the corresponding change on
1244 a process copy (pc) lkb. */
1246 static void _remove_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1249 lkb
->lkb_grmode
= DLM_LOCK_IV
;
1250 /* this unhold undoes the original ref from create_lkb()
1251 so this leads to the lkb being freed */
1255 static void remove_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1257 set_lvb_unlock(r
, lkb
);
1258 _remove_lock(r
, lkb
);
1261 static void remove_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1263 _remove_lock(r
, lkb
);
1266 /* returns: 0 did nothing
1267 1 moved lock to granted
1270 static int revert_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1274 lkb
->lkb_rqmode
= DLM_LOCK_IV
;
1276 switch (lkb
->lkb_status
) {
1277 case DLM_LKSTS_GRANTED
:
1279 case DLM_LKSTS_CONVERT
:
1280 move_lkb(r
, lkb
, DLM_LKSTS_GRANTED
);
1283 case DLM_LKSTS_WAITING
:
1285 lkb
->lkb_grmode
= DLM_LOCK_IV
;
1286 /* this unhold undoes the original ref from create_lkb()
1287 so this leads to the lkb being freed */
1292 log_print("invalid status for revert %d", lkb
->lkb_status
);
1297 static int revert_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1299 return revert_lock(r
, lkb
);
1302 static void _grant_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1304 if (lkb
->lkb_grmode
!= lkb
->lkb_rqmode
) {
1305 lkb
->lkb_grmode
= lkb
->lkb_rqmode
;
1306 if (lkb
->lkb_status
)
1307 move_lkb(r
, lkb
, DLM_LKSTS_GRANTED
);
1309 add_lkb(r
, lkb
, DLM_LKSTS_GRANTED
);
1312 lkb
->lkb_rqmode
= DLM_LOCK_IV
;
1315 static void grant_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1317 set_lvb_lock(r
, lkb
);
1318 _grant_lock(r
, lkb
);
1319 lkb
->lkb_highbast
= 0;
1322 static void grant_lock_pc(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
1323 struct dlm_message
*ms
)
1325 set_lvb_lock_pc(r
, lkb
, ms
);
1326 _grant_lock(r
, lkb
);
1329 /* called by grant_pending_locks() which means an async grant message must
1330 be sent to the requesting node in addition to granting the lock if the
1331 lkb belongs to a remote node. */
1333 static void grant_lock_pending(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1336 if (is_master_copy(lkb
))
1339 queue_cast(r
, lkb
, 0);
1342 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
1343 change the granted/requested modes. We're munging things accordingly in
1345 CONVDEADLK: our grmode may have been forced down to NL to resolve a
1347 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
1348 compatible with other granted locks */
1350 static void munge_demoted(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
1352 if (ms
->m_type
!= DLM_MSG_CONVERT_REPLY
) {
1353 log_print("munge_demoted %x invalid reply type %d",
1354 lkb
->lkb_id
, ms
->m_type
);
1358 if (lkb
->lkb_rqmode
== DLM_LOCK_IV
|| lkb
->lkb_grmode
== DLM_LOCK_IV
) {
1359 log_print("munge_demoted %x invalid modes gr %d rq %d",
1360 lkb
->lkb_id
, lkb
->lkb_grmode
, lkb
->lkb_rqmode
);
1364 lkb
->lkb_grmode
= DLM_LOCK_NL
;
1367 static void munge_altmode(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
1369 if (ms
->m_type
!= DLM_MSG_REQUEST_REPLY
&&
1370 ms
->m_type
!= DLM_MSG_GRANT
) {
1371 log_print("munge_altmode %x invalid reply type %d",
1372 lkb
->lkb_id
, ms
->m_type
);
1376 if (lkb
->lkb_exflags
& DLM_LKF_ALTPR
)
1377 lkb
->lkb_rqmode
= DLM_LOCK_PR
;
1378 else if (lkb
->lkb_exflags
& DLM_LKF_ALTCW
)
1379 lkb
->lkb_rqmode
= DLM_LOCK_CW
;
1381 log_print("munge_altmode invalid exflags %x", lkb
->lkb_exflags
);
1386 static inline int first_in_list(struct dlm_lkb
*lkb
, struct list_head
*head
)
1388 struct dlm_lkb
*first
= list_entry(head
->next
, struct dlm_lkb
,
1390 if (lkb
->lkb_id
== first
->lkb_id
)
1396 /* Check if the given lkb conflicts with another lkb on the queue. */
1398 static int queue_conflict(struct list_head
*head
, struct dlm_lkb
*lkb
)
1400 struct dlm_lkb
*this;
1402 list_for_each_entry(this, head
, lkb_statequeue
) {
1405 if (!modes_compat(this, lkb
))
1412 * "A conversion deadlock arises with a pair of lock requests in the converting
1413 * queue for one resource. The granted mode of each lock blocks the requested
1414 * mode of the other lock."
1416 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
1417 * convert queue from being granted, then deadlk/demote lkb.
1420 * Granted Queue: empty
1421 * Convert Queue: NL->EX (first lock)
1422 * PR->EX (second lock)
1424 * The first lock can't be granted because of the granted mode of the second
1425 * lock and the second lock can't be granted because it's not first in the
1426 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
1427 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
1428 * flag set and return DEMOTED in the lksb flags.
1430 * Originally, this function detected conv-deadlk in a more limited scope:
1431 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
1432 * - if lkb1 was the first entry in the queue (not just earlier), and was
1433 * blocked by the granted mode of lkb2, and there was nothing on the
1434 * granted queue preventing lkb1 from being granted immediately, i.e.
1435 * lkb2 was the only thing preventing lkb1 from being granted.
1437 * That second condition meant we'd only say there was conv-deadlk if
1438 * resolving it (by demotion) would lead to the first lock on the convert
1439 * queue being granted right away. It allowed conversion deadlocks to exist
1440 * between locks on the convert queue while they couldn't be granted anyway.
1442 * Now, we detect and take action on conversion deadlocks immediately when
1443 * they're created, even if they may not be immediately consequential. If
1444 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
1445 * mode that would prevent lkb1's conversion from being granted, we do a
1446 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
1447 * I think this means that the lkb_is_ahead condition below should always
1448 * be zero, i.e. there will never be conv-deadlk between two locks that are
1449 * both already on the convert queue.
1452 static int conversion_deadlock_detect(struct dlm_rsb
*r
, struct dlm_lkb
*lkb2
)
1454 struct dlm_lkb
*lkb1
;
1455 int lkb_is_ahead
= 0;
1457 list_for_each_entry(lkb1
, &r
->res_convertqueue
, lkb_statequeue
) {
1463 if (!lkb_is_ahead
) {
1464 if (!modes_compat(lkb2
, lkb1
))
1467 if (!modes_compat(lkb2
, lkb1
) &&
1468 !modes_compat(lkb1
, lkb2
))
1476 * Return 1 if the lock can be granted, 0 otherwise.
1477 * Also detect and resolve conversion deadlocks.
1479 * lkb is the lock to be granted
1481 * now is 1 if the function is being called in the context of the
1482 * immediate request, it is 0 if called later, after the lock has been
1485 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
1488 static int _can_be_granted(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int now
)
1490 int8_t conv
= (lkb
->lkb_grmode
!= DLM_LOCK_IV
);
1493 * 6-10: Version 5.4 introduced an option to address the phenomenon of
1494 * a new request for a NL mode lock being blocked.
1496 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
1497 * request, then it would be granted. In essence, the use of this flag
1498 * tells the Lock Manager to expedite theis request by not considering
1499 * what may be in the CONVERTING or WAITING queues... As of this
1500 * writing, the EXPEDITE flag can be used only with new requests for NL
1501 * mode locks. This flag is not valid for conversion requests.
1503 * A shortcut. Earlier checks return an error if EXPEDITE is used in a
1504 * conversion or used with a non-NL requested mode. We also know an
1505 * EXPEDITE request is always granted immediately, so now must always
1506 * be 1. The full condition to grant an expedite request: (now &&
1507 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
1508 * therefore be shortened to just checking the flag.
1511 if (lkb
->lkb_exflags
& DLM_LKF_EXPEDITE
)
1515 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
1516 * added to the remaining conditions.
1519 if (queue_conflict(&r
->res_grantqueue
, lkb
))
1523 * 6-3: By default, a conversion request is immediately granted if the
1524 * requested mode is compatible with the modes of all other granted
1528 if (queue_conflict(&r
->res_convertqueue
, lkb
))
1532 * 6-5: But the default algorithm for deciding whether to grant or
1533 * queue conversion requests does not by itself guarantee that such
1534 * requests are serviced on a "first come first serve" basis. This, in
1535 * turn, can lead to a phenomenon known as "indefinate postponement".
1537 * 6-7: This issue is dealt with by using the optional QUECVT flag with
1538 * the system service employed to request a lock conversion. This flag
1539 * forces certain conversion requests to be queued, even if they are
1540 * compatible with the granted modes of other locks on the same
1541 * resource. Thus, the use of this flag results in conversion requests
1542 * being ordered on a "first come first servce" basis.
1544 * DCT: This condition is all about new conversions being able to occur
1545 * "in place" while the lock remains on the granted queue (assuming
1546 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion
1547 * doesn't _have_ to go onto the convert queue where it's processed in
1548 * order. The "now" variable is necessary to distinguish converts
1549 * being received and processed for the first time now, because once a
1550 * convert is moved to the conversion queue the condition below applies
1551 * requiring fifo granting.
1554 if (now
&& conv
&& !(lkb
->lkb_exflags
& DLM_LKF_QUECVT
))
1558 * The NOORDER flag is set to avoid the standard vms rules on grant
1562 if (lkb
->lkb_exflags
& DLM_LKF_NOORDER
)
1566 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
1567 * granted until all other conversion requests ahead of it are granted
1571 if (!now
&& conv
&& first_in_list(lkb
, &r
->res_convertqueue
))
1575 * 6-4: By default, a new request is immediately granted only if all
1576 * three of the following conditions are satisfied when the request is
1578 * - The queue of ungranted conversion requests for the resource is
1580 * - The queue of ungranted new requests for the resource is empty.
1581 * - The mode of the new request is compatible with the most
1582 * restrictive mode of all granted locks on the resource.
1585 if (now
&& !conv
&& list_empty(&r
->res_convertqueue
) &&
1586 list_empty(&r
->res_waitqueue
))
1590 * 6-4: Once a lock request is in the queue of ungranted new requests,
1591 * it cannot be granted until the queue of ungranted conversion
1592 * requests is empty, all ungranted new requests ahead of it are
1593 * granted and/or canceled, and it is compatible with the granted mode
1594 * of the most restrictive lock granted on the resource.
1597 if (!now
&& !conv
&& list_empty(&r
->res_convertqueue
) &&
1598 first_in_list(lkb
, &r
->res_waitqueue
))
1604 static int can_be_granted(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int now
,
1608 int8_t alt
= 0, rqmode
= lkb
->lkb_rqmode
;
1609 int8_t is_convert
= (lkb
->lkb_grmode
!= DLM_LOCK_IV
);
1614 rv
= _can_be_granted(r
, lkb
, now
);
1619 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
1620 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
1621 * cancels one of the locks.
1624 if (is_convert
&& can_be_queued(lkb
) &&
1625 conversion_deadlock_detect(r
, lkb
)) {
1626 if (lkb
->lkb_exflags
& DLM_LKF_CONVDEADLK
) {
1627 lkb
->lkb_grmode
= DLM_LOCK_NL
;
1628 lkb
->lkb_sbflags
|= DLM_SBF_DEMOTED
;
1629 } else if (!(lkb
->lkb_exflags
& DLM_LKF_NODLCKWT
)) {
1633 log_print("can_be_granted deadlock %x now %d",
1642 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
1643 * to grant a request in a mode other than the normal rqmode. It's a
1644 * simple way to provide a big optimization to applications that can
1648 if (rqmode
!= DLM_LOCK_PR
&& (lkb
->lkb_exflags
& DLM_LKF_ALTPR
))
1650 else if (rqmode
!= DLM_LOCK_CW
&& (lkb
->lkb_exflags
& DLM_LKF_ALTCW
))
1654 lkb
->lkb_rqmode
= alt
;
1655 rv
= _can_be_granted(r
, lkb
, now
);
1657 lkb
->lkb_sbflags
|= DLM_SBF_ALTMODE
;
1659 lkb
->lkb_rqmode
= rqmode
;
1665 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock
1666 for locks pending on the convert list. Once verified (watch for these
1667 log_prints), we should be able to just call _can_be_granted() and not
1668 bother with the demote/deadlk cases here (and there's no easy way to deal
1669 with a deadlk here, we'd have to generate something like grant_lock with
1670 the deadlk error.) */
1672 /* Returns the highest requested mode of all blocked conversions; sets
1673 cw if there's a blocked conversion to DLM_LOCK_CW. */
1675 static int grant_pending_convert(struct dlm_rsb
*r
, int high
, int *cw
)
1677 struct dlm_lkb
*lkb
, *s
;
1678 int hi
, demoted
, quit
, grant_restart
, demote_restart
;
1687 list_for_each_entry_safe(lkb
, s
, &r
->res_convertqueue
, lkb_statequeue
) {
1688 demoted
= is_demoted(lkb
);
1691 if (can_be_granted(r
, lkb
, 0, &deadlk
)) {
1692 grant_lock_pending(r
, lkb
);
1697 if (!demoted
&& is_demoted(lkb
)) {
1698 log_print("WARN: pending demoted %x node %d %s",
1699 lkb
->lkb_id
, lkb
->lkb_nodeid
, r
->res_name
);
1705 log_print("WARN: pending deadlock %x node %d %s",
1706 lkb
->lkb_id
, lkb
->lkb_nodeid
, r
->res_name
);
1711 hi
= max_t(int, lkb
->lkb_rqmode
, hi
);
1713 if (cw
&& lkb
->lkb_rqmode
== DLM_LOCK_CW
)
1719 if (demote_restart
&& !quit
) {
1724 return max_t(int, high
, hi
);
1727 static int grant_pending_wait(struct dlm_rsb
*r
, int high
, int *cw
)
1729 struct dlm_lkb
*lkb
, *s
;
1731 list_for_each_entry_safe(lkb
, s
, &r
->res_waitqueue
, lkb_statequeue
) {
1732 if (can_be_granted(r
, lkb
, 0, NULL
))
1733 grant_lock_pending(r
, lkb
);
1735 high
= max_t(int, lkb
->lkb_rqmode
, high
);
1736 if (lkb
->lkb_rqmode
== DLM_LOCK_CW
)
1744 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
1745 on either the convert or waiting queue.
1746 high is the largest rqmode of all locks blocked on the convert or
1749 static int lock_requires_bast(struct dlm_lkb
*gr
, int high
, int cw
)
1751 if (gr
->lkb_grmode
== DLM_LOCK_PR
&& cw
) {
1752 if (gr
->lkb_highbast
< DLM_LOCK_EX
)
1757 if (gr
->lkb_highbast
< high
&&
1758 !__dlm_compat_matrix
[gr
->lkb_grmode
+1][high
+1])
1763 static void grant_pending_locks(struct dlm_rsb
*r
)
1765 struct dlm_lkb
*lkb
, *s
;
1766 int high
= DLM_LOCK_IV
;
1769 DLM_ASSERT(is_master(r
), dlm_dump_rsb(r
););
1771 high
= grant_pending_convert(r
, high
, &cw
);
1772 high
= grant_pending_wait(r
, high
, &cw
);
1774 if (high
== DLM_LOCK_IV
)
1778 * If there are locks left on the wait/convert queue then send blocking
1779 * ASTs to granted locks based on the largest requested mode (high)
1783 list_for_each_entry_safe(lkb
, s
, &r
->res_grantqueue
, lkb_statequeue
) {
1784 if (lkb
->lkb_bastaddr
&& lock_requires_bast(lkb
, high
, cw
)) {
1785 if (cw
&& high
== DLM_LOCK_PR
)
1786 queue_bast(r
, lkb
, DLM_LOCK_CW
);
1788 queue_bast(r
, lkb
, high
);
1789 lkb
->lkb_highbast
= high
;
1794 static int modes_require_bast(struct dlm_lkb
*gr
, struct dlm_lkb
*rq
)
1796 if ((gr
->lkb_grmode
== DLM_LOCK_PR
&& rq
->lkb_rqmode
== DLM_LOCK_CW
) ||
1797 (gr
->lkb_grmode
== DLM_LOCK_CW
&& rq
->lkb_rqmode
== DLM_LOCK_PR
)) {
1798 if (gr
->lkb_highbast
< DLM_LOCK_EX
)
1803 if (gr
->lkb_highbast
< rq
->lkb_rqmode
&& !modes_compat(gr
, rq
))
1808 static void send_bast_queue(struct dlm_rsb
*r
, struct list_head
*head
,
1809 struct dlm_lkb
*lkb
)
1813 list_for_each_entry(gr
, head
, lkb_statequeue
) {
1814 if (gr
->lkb_bastaddr
&& modes_require_bast(gr
, lkb
)) {
1815 queue_bast(r
, gr
, lkb
->lkb_rqmode
);
1816 gr
->lkb_highbast
= lkb
->lkb_rqmode
;
1821 static void send_blocking_asts(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1823 send_bast_queue(r
, &r
->res_grantqueue
, lkb
);
1826 static void send_blocking_asts_all(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1828 send_bast_queue(r
, &r
->res_grantqueue
, lkb
);
1829 send_bast_queue(r
, &r
->res_convertqueue
, lkb
);
1832 /* set_master(r, lkb) -- set the master nodeid of a resource
1834 The purpose of this function is to set the nodeid field in the given
1835 lkb using the nodeid field in the given rsb. If the rsb's nodeid is
1836 known, it can just be copied to the lkb and the function will return
1837 0. If the rsb's nodeid is _not_ known, it needs to be looked up
1838 before it can be copied to the lkb.
1840 When the rsb nodeid is being looked up remotely, the initial lkb
1841 causing the lookup is kept on the ls_waiters list waiting for the
1842 lookup reply. Other lkb's waiting for the same rsb lookup are kept
1843 on the rsb's res_lookup list until the master is verified.
1846 0: nodeid is set in rsb/lkb and the caller should go ahead and use it
1847 1: the rsb master is not available and the lkb has been placed on
1851 static int set_master(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
1853 struct dlm_ls
*ls
= r
->res_ls
;
1854 int error
, dir_nodeid
, ret_nodeid
, our_nodeid
= dlm_our_nodeid();
1856 if (rsb_flag(r
, RSB_MASTER_UNCERTAIN
)) {
1857 rsb_clear_flag(r
, RSB_MASTER_UNCERTAIN
);
1858 r
->res_first_lkid
= lkb
->lkb_id
;
1859 lkb
->lkb_nodeid
= r
->res_nodeid
;
1863 if (r
->res_first_lkid
&& r
->res_first_lkid
!= lkb
->lkb_id
) {
1864 list_add_tail(&lkb
->lkb_rsb_lookup
, &r
->res_lookup
);
1868 if (r
->res_nodeid
== 0) {
1869 lkb
->lkb_nodeid
= 0;
1873 if (r
->res_nodeid
> 0) {
1874 lkb
->lkb_nodeid
= r
->res_nodeid
;
1878 DLM_ASSERT(r
->res_nodeid
== -1, dlm_dump_rsb(r
););
1880 dir_nodeid
= dlm_dir_nodeid(r
);
1882 if (dir_nodeid
!= our_nodeid
) {
1883 r
->res_first_lkid
= lkb
->lkb_id
;
1884 send_lookup(r
, lkb
);
1889 /* It's possible for dlm_scand to remove an old rsb for
1890 this same resource from the toss list, us to create
1891 a new one, look up the master locally, and find it
1892 already exists just before dlm_scand does the
1893 dir_remove() on the previous rsb. */
1895 error
= dlm_dir_lookup(ls
, our_nodeid
, r
->res_name
,
1896 r
->res_length
, &ret_nodeid
);
1899 log_debug(ls
, "dir_lookup error %d %s", error
, r
->res_name
);
1903 if (ret_nodeid
== our_nodeid
) {
1904 r
->res_first_lkid
= 0;
1906 lkb
->lkb_nodeid
= 0;
1908 r
->res_first_lkid
= lkb
->lkb_id
;
1909 r
->res_nodeid
= ret_nodeid
;
1910 lkb
->lkb_nodeid
= ret_nodeid
;
1915 static void process_lookup_list(struct dlm_rsb
*r
)
1917 struct dlm_lkb
*lkb
, *safe
;
1919 list_for_each_entry_safe(lkb
, safe
, &r
->res_lookup
, lkb_rsb_lookup
) {
1920 list_del_init(&lkb
->lkb_rsb_lookup
);
1921 _request_lock(r
, lkb
);
1926 /* confirm_master -- confirm (or deny) an rsb's master nodeid */
1928 static void confirm_master(struct dlm_rsb
*r
, int error
)
1930 struct dlm_lkb
*lkb
;
1932 if (!r
->res_first_lkid
)
1938 r
->res_first_lkid
= 0;
1939 process_lookup_list(r
);
1945 /* the remote request failed and won't be retried (it was
1946 a NOQUEUE, or has been canceled/unlocked); make a waiting
1947 lkb the first_lkid */
1949 r
->res_first_lkid
= 0;
1951 if (!list_empty(&r
->res_lookup
)) {
1952 lkb
= list_entry(r
->res_lookup
.next
, struct dlm_lkb
,
1954 list_del_init(&lkb
->lkb_rsb_lookup
);
1955 r
->res_first_lkid
= lkb
->lkb_id
;
1956 _request_lock(r
, lkb
);
1962 log_error(r
->res_ls
, "confirm_master unknown error %d", error
);
1966 static int set_lock_args(int mode
, struct dlm_lksb
*lksb
, uint32_t flags
,
1967 int namelen
, unsigned long timeout_cs
, void *ast
,
1968 void *astarg
, void *bast
, struct dlm_args
*args
)
1972 /* check for invalid arg usage */
1974 if (mode
< 0 || mode
> DLM_LOCK_EX
)
1977 if (!(flags
& DLM_LKF_CONVERT
) && (namelen
> DLM_RESNAME_MAXLEN
))
1980 if (flags
& DLM_LKF_CANCEL
)
1983 if (flags
& DLM_LKF_QUECVT
&& !(flags
& DLM_LKF_CONVERT
))
1986 if (flags
& DLM_LKF_CONVDEADLK
&& !(flags
& DLM_LKF_CONVERT
))
1989 if (flags
& DLM_LKF_CONVDEADLK
&& flags
& DLM_LKF_NOQUEUE
)
1992 if (flags
& DLM_LKF_EXPEDITE
&& flags
& DLM_LKF_CONVERT
)
1995 if (flags
& DLM_LKF_EXPEDITE
&& flags
& DLM_LKF_QUECVT
)
1998 if (flags
& DLM_LKF_EXPEDITE
&& flags
& DLM_LKF_NOQUEUE
)
2001 if (flags
& DLM_LKF_EXPEDITE
&& mode
!= DLM_LOCK_NL
)
2007 if (flags
& DLM_LKF_VALBLK
&& !lksb
->sb_lvbptr
)
2010 if (flags
& DLM_LKF_CONVERT
&& !lksb
->sb_lkid
)
2013 /* these args will be copied to the lkb in validate_lock_args,
2014 it cannot be done now because when converting locks, fields in
2015 an active lkb cannot be modified before locking the rsb */
2017 args
->flags
= flags
;
2018 args
->astaddr
= ast
;
2019 args
->astparam
= (long) astarg
;
2020 args
->bastaddr
= bast
;
2021 args
->timeout
= timeout_cs
;
2029 static int set_unlock_args(uint32_t flags
, void *astarg
, struct dlm_args
*args
)
2031 if (flags
& ~(DLM_LKF_CANCEL
| DLM_LKF_VALBLK
| DLM_LKF_IVVALBLK
|
2032 DLM_LKF_FORCEUNLOCK
))
2035 if (flags
& DLM_LKF_CANCEL
&& flags
& DLM_LKF_FORCEUNLOCK
)
2038 args
->flags
= flags
;
2039 args
->astparam
= (long) astarg
;
2043 static int validate_lock_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2044 struct dlm_args
*args
)
2048 if (args
->flags
& DLM_LKF_CONVERT
) {
2049 if (lkb
->lkb_flags
& DLM_IFL_MSTCPY
)
2052 if (args
->flags
& DLM_LKF_QUECVT
&&
2053 !__quecvt_compat_matrix
[lkb
->lkb_grmode
+1][args
->mode
+1])
2057 if (lkb
->lkb_status
!= DLM_LKSTS_GRANTED
)
2060 if (lkb
->lkb_wait_type
)
2063 if (is_overlap(lkb
))
2067 lkb
->lkb_exflags
= args
->flags
;
2068 lkb
->lkb_sbflags
= 0;
2069 lkb
->lkb_astaddr
= args
->astaddr
;
2070 lkb
->lkb_astparam
= args
->astparam
;
2071 lkb
->lkb_bastaddr
= args
->bastaddr
;
2072 lkb
->lkb_rqmode
= args
->mode
;
2073 lkb
->lkb_lksb
= args
->lksb
;
2074 lkb
->lkb_lvbptr
= args
->lksb
->sb_lvbptr
;
2075 lkb
->lkb_ownpid
= (int) current
->pid
;
2076 lkb
->lkb_timeout_cs
= args
->timeout
;
2082 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2085 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2086 because there may be a lookup in progress and it's valid to do
2087 cancel/unlockf on it */
2089 static int validate_unlock_args(struct dlm_lkb
*lkb
, struct dlm_args
*args
)
2091 struct dlm_ls
*ls
= lkb
->lkb_resource
->res_ls
;
2094 if (lkb
->lkb_flags
& DLM_IFL_MSTCPY
) {
2095 log_error(ls
, "unlock on MSTCPY %x", lkb
->lkb_id
);
2100 /* an lkb may still exist even though the lock is EOL'ed due to a
2101 cancel, unlock or failed noqueue request; an app can't use these
2102 locks; return same error as if the lkid had not been found at all */
2104 if (lkb
->lkb_flags
& DLM_IFL_ENDOFLIFE
) {
2105 log_debug(ls
, "unlock on ENDOFLIFE %x", lkb
->lkb_id
);
2110 /* an lkb may be waiting for an rsb lookup to complete where the
2111 lookup was initiated by another lock */
2113 if (args
->flags
& (DLM_LKF_CANCEL
| DLM_LKF_FORCEUNLOCK
)) {
2114 if (!list_empty(&lkb
->lkb_rsb_lookup
)) {
2115 log_debug(ls
, "unlock on rsb_lookup %x", lkb
->lkb_id
);
2116 list_del_init(&lkb
->lkb_rsb_lookup
);
2117 queue_cast(lkb
->lkb_resource
, lkb
,
2118 args
->flags
& DLM_LKF_CANCEL
?
2119 -DLM_ECANCEL
: -DLM_EUNLOCK
);
2120 unhold_lkb(lkb
); /* undoes create_lkb() */
2126 /* cancel not allowed with another cancel/unlock in progress */
2128 if (args
->flags
& DLM_LKF_CANCEL
) {
2129 if (lkb
->lkb_exflags
& DLM_LKF_CANCEL
)
2132 if (is_overlap(lkb
))
2135 /* don't let scand try to do a cancel */
2138 if (lkb
->lkb_flags
& DLM_IFL_RESEND
) {
2139 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_CANCEL
;
2144 switch (lkb
->lkb_wait_type
) {
2145 case DLM_MSG_LOOKUP
:
2146 case DLM_MSG_REQUEST
:
2147 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_CANCEL
;
2150 case DLM_MSG_UNLOCK
:
2151 case DLM_MSG_CANCEL
:
2154 /* add_to_waiters() will set OVERLAP_CANCEL */
2158 /* do we need to allow a force-unlock if there's a normal unlock
2159 already in progress? in what conditions could the normal unlock
2160 fail such that we'd want to send a force-unlock to be sure? */
2162 if (args
->flags
& DLM_LKF_FORCEUNLOCK
) {
2163 if (lkb
->lkb_exflags
& DLM_LKF_FORCEUNLOCK
)
2166 if (is_overlap_unlock(lkb
))
2169 /* don't let scand try to do a cancel */
2172 if (lkb
->lkb_flags
& DLM_IFL_RESEND
) {
2173 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_UNLOCK
;
2178 switch (lkb
->lkb_wait_type
) {
2179 case DLM_MSG_LOOKUP
:
2180 case DLM_MSG_REQUEST
:
2181 lkb
->lkb_flags
|= DLM_IFL_OVERLAP_UNLOCK
;
2184 case DLM_MSG_UNLOCK
:
2187 /* add_to_waiters() will set OVERLAP_UNLOCK */
2191 /* normal unlock not allowed if there's any op in progress */
2193 if (lkb
->lkb_wait_type
|| lkb
->lkb_wait_count
)
2197 /* an overlapping op shouldn't blow away exflags from other op */
2198 lkb
->lkb_exflags
|= args
->flags
;
2199 lkb
->lkb_sbflags
= 0;
2200 lkb
->lkb_astparam
= args
->astparam
;
2204 log_debug(ls
, "validate_unlock_args %d %x %x %x %x %d %s", rv
,
2205 lkb
->lkb_id
, lkb
->lkb_flags
, lkb
->lkb_exflags
,
2206 args
->flags
, lkb
->lkb_wait_type
,
2207 lkb
->lkb_resource
->res_name
);
2212 * Four stage 4 varieties:
2213 * do_request(), do_convert(), do_unlock(), do_cancel()
2214 * These are called on the master node for the given lock and
2215 * from the central locking logic.
2218 static int do_request(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2222 if (can_be_granted(r
, lkb
, 1, NULL
)) {
2224 queue_cast(r
, lkb
, 0);
2228 if (can_be_queued(lkb
)) {
2229 error
= -EINPROGRESS
;
2230 add_lkb(r
, lkb
, DLM_LKSTS_WAITING
);
2231 send_blocking_asts(r
, lkb
);
2237 if (force_blocking_asts(lkb
))
2238 send_blocking_asts_all(r
, lkb
);
2239 queue_cast(r
, lkb
, -EAGAIN
);
2245 static int do_convert(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2250 /* changing an existing lock may allow others to be granted */
2252 if (can_be_granted(r
, lkb
, 1, &deadlk
)) {
2254 queue_cast(r
, lkb
, 0);
2255 grant_pending_locks(r
);
2259 /* can_be_granted() detected that this lock would block in a conversion
2260 deadlock, so we leave it on the granted queue and return EDEADLK in
2261 the ast for the convert. */
2264 /* it's left on the granted queue */
2265 log_debug(r
->res_ls
, "deadlock %x node %d sts%d g%d r%d %s",
2266 lkb
->lkb_id
, lkb
->lkb_nodeid
, lkb
->lkb_status
,
2267 lkb
->lkb_grmode
, lkb
->lkb_rqmode
, r
->res_name
);
2268 revert_lock(r
, lkb
);
2269 queue_cast(r
, lkb
, -EDEADLK
);
2274 /* is_demoted() means the can_be_granted() above set the grmode
2275 to NL, and left us on the granted queue. This auto-demotion
2276 (due to CONVDEADLK) might mean other locks, and/or this lock, are
2277 now grantable. We have to try to grant other converting locks
2278 before we try again to grant this one. */
2280 if (is_demoted(lkb
)) {
2281 grant_pending_convert(r
, DLM_LOCK_IV
, NULL
);
2282 if (_can_be_granted(r
, lkb
, 1)) {
2284 queue_cast(r
, lkb
, 0);
2285 grant_pending_locks(r
);
2288 /* else fall through and move to convert queue */
2291 if (can_be_queued(lkb
)) {
2292 error
= -EINPROGRESS
;
2294 add_lkb(r
, lkb
, DLM_LKSTS_CONVERT
);
2295 send_blocking_asts(r
, lkb
);
2301 if (force_blocking_asts(lkb
))
2302 send_blocking_asts_all(r
, lkb
);
2303 queue_cast(r
, lkb
, -EAGAIN
);
2309 static int do_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2311 remove_lock(r
, lkb
);
2312 queue_cast(r
, lkb
, -DLM_EUNLOCK
);
2313 grant_pending_locks(r
);
2314 return -DLM_EUNLOCK
;
2317 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
2319 static int do_cancel(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2323 error
= revert_lock(r
, lkb
);
2325 queue_cast(r
, lkb
, -DLM_ECANCEL
);
2326 grant_pending_locks(r
);
2327 return -DLM_ECANCEL
;
2333 * Four stage 3 varieties:
2334 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
2337 /* add a new lkb to a possibly new rsb, called by requesting process */
2339 static int _request_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2343 /* set_master: sets lkb nodeid from r */
2345 error
= set_master(r
, lkb
);
2354 /* receive_request() calls do_request() on remote node */
2355 error
= send_request(r
, lkb
);
2357 error
= do_request(r
, lkb
);
2362 /* change some property of an existing lkb, e.g. mode */
2364 static int _convert_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2369 /* receive_convert() calls do_convert() on remote node */
2370 error
= send_convert(r
, lkb
);
2372 error
= do_convert(r
, lkb
);
2377 /* remove an existing lkb from the granted queue */
2379 static int _unlock_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2384 /* receive_unlock() calls do_unlock() on remote node */
2385 error
= send_unlock(r
, lkb
);
2387 error
= do_unlock(r
, lkb
);
2392 /* remove an existing lkb from the convert or wait queue */
2394 static int _cancel_lock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2399 /* receive_cancel() calls do_cancel() on remote node */
2400 error
= send_cancel(r
, lkb
);
2402 error
= do_cancel(r
, lkb
);
2408 * Four stage 2 varieties:
2409 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
2412 static int request_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
, char *name
,
2413 int len
, struct dlm_args
*args
)
2418 error
= validate_lock_args(ls
, lkb
, args
);
2422 error
= find_rsb(ls
, name
, len
, R_CREATE
, &r
);
2429 lkb
->lkb_lksb
->sb_lkid
= lkb
->lkb_id
;
2431 error
= _request_lock(r
, lkb
);
2440 static int convert_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2441 struct dlm_args
*args
)
2446 r
= lkb
->lkb_resource
;
2451 error
= validate_lock_args(ls
, lkb
, args
);
2455 error
= _convert_lock(r
, lkb
);
2462 static int unlock_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2463 struct dlm_args
*args
)
2468 r
= lkb
->lkb_resource
;
2473 error
= validate_unlock_args(lkb
, args
);
2477 error
= _unlock_lock(r
, lkb
);
2484 static int cancel_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2485 struct dlm_args
*args
)
2490 r
= lkb
->lkb_resource
;
2495 error
= validate_unlock_args(lkb
, args
);
2499 error
= _cancel_lock(r
, lkb
);
2507 * Two stage 1 varieties: dlm_lock() and dlm_unlock()
2510 int dlm_lock(dlm_lockspace_t
*lockspace
,
2512 struct dlm_lksb
*lksb
,
2515 unsigned int namelen
,
2516 uint32_t parent_lkid
,
2517 void (*ast
) (void *astarg
),
2519 void (*bast
) (void *astarg
, int mode
))
2522 struct dlm_lkb
*lkb
;
2523 struct dlm_args args
;
2524 int error
, convert
= flags
& DLM_LKF_CONVERT
;
2526 ls
= dlm_find_lockspace_local(lockspace
);
2530 dlm_lock_recovery(ls
);
2533 error
= find_lkb(ls
, lksb
->sb_lkid
, &lkb
);
2535 error
= create_lkb(ls
, &lkb
);
2540 error
= set_lock_args(mode
, lksb
, flags
, namelen
, 0, ast
,
2541 astarg
, bast
, &args
);
2546 error
= convert_lock(ls
, lkb
, &args
);
2548 error
= request_lock(ls
, lkb
, name
, namelen
, &args
);
2550 if (error
== -EINPROGRESS
)
2553 if (convert
|| error
)
2555 if (error
== -EAGAIN
|| error
== -EDEADLK
)
2558 dlm_unlock_recovery(ls
);
2559 dlm_put_lockspace(ls
);
2563 int dlm_unlock(dlm_lockspace_t
*lockspace
,
2566 struct dlm_lksb
*lksb
,
2570 struct dlm_lkb
*lkb
;
2571 struct dlm_args args
;
2574 ls
= dlm_find_lockspace_local(lockspace
);
2578 dlm_lock_recovery(ls
);
2580 error
= find_lkb(ls
, lkid
, &lkb
);
2584 error
= set_unlock_args(flags
, astarg
, &args
);
2588 if (flags
& DLM_LKF_CANCEL
)
2589 error
= cancel_lock(ls
, lkb
, &args
);
2591 error
= unlock_lock(ls
, lkb
, &args
);
2593 if (error
== -DLM_EUNLOCK
|| error
== -DLM_ECANCEL
)
2595 if (error
== -EBUSY
&& (flags
& (DLM_LKF_CANCEL
| DLM_LKF_FORCEUNLOCK
)))
2600 dlm_unlock_recovery(ls
);
2601 dlm_put_lockspace(ls
);
2606 * send/receive routines for remote operations and replies
2610 * send_request receive_request
2611 * send_convert receive_convert
2612 * send_unlock receive_unlock
2613 * send_cancel receive_cancel
2614 * send_grant receive_grant
2615 * send_bast receive_bast
2616 * send_lookup receive_lookup
2617 * send_remove receive_remove
2620 * receive_request_reply send_request_reply
2621 * receive_convert_reply send_convert_reply
2622 * receive_unlock_reply send_unlock_reply
2623 * receive_cancel_reply send_cancel_reply
2624 * receive_lookup_reply send_lookup_reply
2627 static int _create_message(struct dlm_ls
*ls
, int mb_len
,
2628 int to_nodeid
, int mstype
,
2629 struct dlm_message
**ms_ret
,
2630 struct dlm_mhandle
**mh_ret
)
2632 struct dlm_message
*ms
;
2633 struct dlm_mhandle
*mh
;
2636 /* get_buffer gives us a message handle (mh) that we need to
2637 pass into lowcomms_commit and a message buffer (mb) that we
2638 write our data into */
2640 mh
= dlm_lowcomms_get_buffer(to_nodeid
, mb_len
, ls
->ls_allocation
, &mb
);
2644 memset(mb
, 0, mb_len
);
2646 ms
= (struct dlm_message
*) mb
;
2648 ms
->m_header
.h_version
= (DLM_HEADER_MAJOR
| DLM_HEADER_MINOR
);
2649 ms
->m_header
.h_lockspace
= ls
->ls_global_id
;
2650 ms
->m_header
.h_nodeid
= dlm_our_nodeid();
2651 ms
->m_header
.h_length
= mb_len
;
2652 ms
->m_header
.h_cmd
= DLM_MSG
;
2654 ms
->m_type
= mstype
;
2661 static int create_message(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
2662 int to_nodeid
, int mstype
,
2663 struct dlm_message
**ms_ret
,
2664 struct dlm_mhandle
**mh_ret
)
2666 int mb_len
= sizeof(struct dlm_message
);
2669 case DLM_MSG_REQUEST
:
2670 case DLM_MSG_LOOKUP
:
2671 case DLM_MSG_REMOVE
:
2672 mb_len
+= r
->res_length
;
2674 case DLM_MSG_CONVERT
:
2675 case DLM_MSG_UNLOCK
:
2676 case DLM_MSG_REQUEST_REPLY
:
2677 case DLM_MSG_CONVERT_REPLY
:
2679 if (lkb
&& lkb
->lkb_lvbptr
)
2680 mb_len
+= r
->res_ls
->ls_lvblen
;
2684 return _create_message(r
->res_ls
, mb_len
, to_nodeid
, mstype
,
2688 /* further lowcomms enhancements or alternate implementations may make
2689 the return value from this function useful at some point */
2691 static int send_message(struct dlm_mhandle
*mh
, struct dlm_message
*ms
)
2693 dlm_message_out(ms
);
2694 dlm_lowcomms_commit_buffer(mh
);
2698 static void send_args(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
2699 struct dlm_message
*ms
)
2701 ms
->m_nodeid
= lkb
->lkb_nodeid
;
2702 ms
->m_pid
= lkb
->lkb_ownpid
;
2703 ms
->m_lkid
= lkb
->lkb_id
;
2704 ms
->m_remid
= lkb
->lkb_remid
;
2705 ms
->m_exflags
= lkb
->lkb_exflags
;
2706 ms
->m_sbflags
= lkb
->lkb_sbflags
;
2707 ms
->m_flags
= lkb
->lkb_flags
;
2708 ms
->m_lvbseq
= lkb
->lkb_lvbseq
;
2709 ms
->m_status
= lkb
->lkb_status
;
2710 ms
->m_grmode
= lkb
->lkb_grmode
;
2711 ms
->m_rqmode
= lkb
->lkb_rqmode
;
2712 ms
->m_hash
= r
->res_hash
;
2714 /* m_result and m_bastmode are set from function args,
2715 not from lkb fields */
2717 if (lkb
->lkb_bastaddr
)
2718 ms
->m_asts
|= AST_BAST
;
2719 if (lkb
->lkb_astaddr
)
2720 ms
->m_asts
|= AST_COMP
;
2722 /* compare with switch in create_message; send_remove() doesn't
2725 switch (ms
->m_type
) {
2726 case DLM_MSG_REQUEST
:
2727 case DLM_MSG_LOOKUP
:
2728 memcpy(ms
->m_extra
, r
->res_name
, r
->res_length
);
2730 case DLM_MSG_CONVERT
:
2731 case DLM_MSG_UNLOCK
:
2732 case DLM_MSG_REQUEST_REPLY
:
2733 case DLM_MSG_CONVERT_REPLY
:
2735 if (!lkb
->lkb_lvbptr
)
2737 memcpy(ms
->m_extra
, lkb
->lkb_lvbptr
, r
->res_ls
->ls_lvblen
);
2742 static int send_common(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int mstype
)
2744 struct dlm_message
*ms
;
2745 struct dlm_mhandle
*mh
;
2746 int to_nodeid
, error
;
2748 error
= add_to_waiters(lkb
, mstype
);
2752 to_nodeid
= r
->res_nodeid
;
2754 error
= create_message(r
, lkb
, to_nodeid
, mstype
, &ms
, &mh
);
2758 send_args(r
, lkb
, ms
);
2760 error
= send_message(mh
, ms
);
2766 remove_from_waiters(lkb
, msg_reply_type(mstype
));
2770 static int send_request(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2772 return send_common(r
, lkb
, DLM_MSG_REQUEST
);
2775 static int send_convert(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2779 error
= send_common(r
, lkb
, DLM_MSG_CONVERT
);
2781 /* down conversions go without a reply from the master */
2782 if (!error
&& down_conversion(lkb
)) {
2783 remove_from_waiters(lkb
, DLM_MSG_CONVERT_REPLY
);
2784 r
->res_ls
->ls_stub_ms
.m_type
= DLM_MSG_CONVERT_REPLY
;
2785 r
->res_ls
->ls_stub_ms
.m_result
= 0;
2786 r
->res_ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
2787 __receive_convert_reply(r
, lkb
, &r
->res_ls
->ls_stub_ms
);
2793 /* FIXME: if this lkb is the only lock we hold on the rsb, then set
2794 MASTER_UNCERTAIN to force the next request on the rsb to confirm
2795 that the master is still correct. */
2797 static int send_unlock(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2799 return send_common(r
, lkb
, DLM_MSG_UNLOCK
);
2802 static int send_cancel(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2804 return send_common(r
, lkb
, DLM_MSG_CANCEL
);
2807 static int send_grant(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2809 struct dlm_message
*ms
;
2810 struct dlm_mhandle
*mh
;
2811 int to_nodeid
, error
;
2813 to_nodeid
= lkb
->lkb_nodeid
;
2815 error
= create_message(r
, lkb
, to_nodeid
, DLM_MSG_GRANT
, &ms
, &mh
);
2819 send_args(r
, lkb
, ms
);
2823 error
= send_message(mh
, ms
);
2828 static int send_bast(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int mode
)
2830 struct dlm_message
*ms
;
2831 struct dlm_mhandle
*mh
;
2832 int to_nodeid
, error
;
2834 to_nodeid
= lkb
->lkb_nodeid
;
2836 error
= create_message(r
, NULL
, to_nodeid
, DLM_MSG_BAST
, &ms
, &mh
);
2840 send_args(r
, lkb
, ms
);
2842 ms
->m_bastmode
= mode
;
2844 error
= send_message(mh
, ms
);
2849 static int send_lookup(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
)
2851 struct dlm_message
*ms
;
2852 struct dlm_mhandle
*mh
;
2853 int to_nodeid
, error
;
2855 error
= add_to_waiters(lkb
, DLM_MSG_LOOKUP
);
2859 to_nodeid
= dlm_dir_nodeid(r
);
2861 error
= create_message(r
, NULL
, to_nodeid
, DLM_MSG_LOOKUP
, &ms
, &mh
);
2865 send_args(r
, lkb
, ms
);
2867 error
= send_message(mh
, ms
);
2873 remove_from_waiters(lkb
, DLM_MSG_LOOKUP_REPLY
);
2877 static int send_remove(struct dlm_rsb
*r
)
2879 struct dlm_message
*ms
;
2880 struct dlm_mhandle
*mh
;
2881 int to_nodeid
, error
;
2883 to_nodeid
= dlm_dir_nodeid(r
);
2885 error
= create_message(r
, NULL
, to_nodeid
, DLM_MSG_REMOVE
, &ms
, &mh
);
2889 memcpy(ms
->m_extra
, r
->res_name
, r
->res_length
);
2890 ms
->m_hash
= r
->res_hash
;
2892 error
= send_message(mh
, ms
);
2897 static int send_common_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
2900 struct dlm_message
*ms
;
2901 struct dlm_mhandle
*mh
;
2902 int to_nodeid
, error
;
2904 to_nodeid
= lkb
->lkb_nodeid
;
2906 error
= create_message(r
, lkb
, to_nodeid
, mstype
, &ms
, &mh
);
2910 send_args(r
, lkb
, ms
);
2914 error
= send_message(mh
, ms
);
2919 static int send_request_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2921 return send_common_reply(r
, lkb
, DLM_MSG_REQUEST_REPLY
, rv
);
2924 static int send_convert_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2926 return send_common_reply(r
, lkb
, DLM_MSG_CONVERT_REPLY
, rv
);
2929 static int send_unlock_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2931 return send_common_reply(r
, lkb
, DLM_MSG_UNLOCK_REPLY
, rv
);
2934 static int send_cancel_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
, int rv
)
2936 return send_common_reply(r
, lkb
, DLM_MSG_CANCEL_REPLY
, rv
);
2939 static int send_lookup_reply(struct dlm_ls
*ls
, struct dlm_message
*ms_in
,
2940 int ret_nodeid
, int rv
)
2942 struct dlm_rsb
*r
= &ls
->ls_stub_rsb
;
2943 struct dlm_message
*ms
;
2944 struct dlm_mhandle
*mh
;
2945 int error
, nodeid
= ms_in
->m_header
.h_nodeid
;
2947 error
= create_message(r
, NULL
, nodeid
, DLM_MSG_LOOKUP_REPLY
, &ms
, &mh
);
2951 ms
->m_lkid
= ms_in
->m_lkid
;
2953 ms
->m_nodeid
= ret_nodeid
;
2955 error
= send_message(mh
, ms
);
2960 /* which args we save from a received message depends heavily on the type
2961 of message, unlike the send side where we can safely send everything about
2962 the lkb for any type of message */
2964 static void receive_flags(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
2966 lkb
->lkb_exflags
= ms
->m_exflags
;
2967 lkb
->lkb_sbflags
= ms
->m_sbflags
;
2968 lkb
->lkb_flags
= (lkb
->lkb_flags
& 0xFFFF0000) |
2969 (ms
->m_flags
& 0x0000FFFF);
2972 static void receive_flags_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
2974 lkb
->lkb_sbflags
= ms
->m_sbflags
;
2975 lkb
->lkb_flags
= (lkb
->lkb_flags
& 0xFFFF0000) |
2976 (ms
->m_flags
& 0x0000FFFF);
2979 static int receive_extralen(struct dlm_message
*ms
)
2981 return (ms
->m_header
.h_length
- sizeof(struct dlm_message
));
2984 static int receive_lvb(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
2985 struct dlm_message
*ms
)
2989 if (lkb
->lkb_exflags
& DLM_LKF_VALBLK
) {
2990 if (!lkb
->lkb_lvbptr
)
2991 lkb
->lkb_lvbptr
= dlm_allocate_lvb(ls
);
2992 if (!lkb
->lkb_lvbptr
)
2994 len
= receive_extralen(ms
);
2995 memcpy(lkb
->lkb_lvbptr
, ms
->m_extra
, len
);
3000 static int receive_request_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
3001 struct dlm_message
*ms
)
3003 lkb
->lkb_nodeid
= ms
->m_header
.h_nodeid
;
3004 lkb
->lkb_ownpid
= ms
->m_pid
;
3005 lkb
->lkb_remid
= ms
->m_lkid
;
3006 lkb
->lkb_grmode
= DLM_LOCK_IV
;
3007 lkb
->lkb_rqmode
= ms
->m_rqmode
;
3008 lkb
->lkb_bastaddr
= (void *) (long) (ms
->m_asts
& AST_BAST
);
3009 lkb
->lkb_astaddr
= (void *) (long) (ms
->m_asts
& AST_COMP
);
3011 if (lkb
->lkb_exflags
& DLM_LKF_VALBLK
) {
3012 /* lkb was just created so there won't be an lvb yet */
3013 lkb
->lkb_lvbptr
= dlm_allocate_lvb(ls
);
3014 if (!lkb
->lkb_lvbptr
)
3021 static int receive_convert_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
3022 struct dlm_message
*ms
)
3024 if (lkb
->lkb_status
!= DLM_LKSTS_GRANTED
)
3027 if (receive_lvb(ls
, lkb
, ms
))
3030 lkb
->lkb_rqmode
= ms
->m_rqmode
;
3031 lkb
->lkb_lvbseq
= ms
->m_lvbseq
;
3036 static int receive_unlock_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
3037 struct dlm_message
*ms
)
3039 if (receive_lvb(ls
, lkb
, ms
))
3044 /* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3045 uses to send a reply and that the remote end uses to process the reply. */
3047 static void setup_stub_lkb(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3049 struct dlm_lkb
*lkb
= &ls
->ls_stub_lkb
;
3050 lkb
->lkb_nodeid
= ms
->m_header
.h_nodeid
;
3051 lkb
->lkb_remid
= ms
->m_lkid
;
3054 /* This is called after the rsb is locked so that we can safely inspect
3055 fields in the lkb. */
3057 static int validate_message(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3059 int from
= ms
->m_header
.h_nodeid
;
3062 switch (ms
->m_type
) {
3063 case DLM_MSG_CONVERT
:
3064 case DLM_MSG_UNLOCK
:
3065 case DLM_MSG_CANCEL
:
3066 if (!is_master_copy(lkb
) || lkb
->lkb_nodeid
!= from
)
3070 case DLM_MSG_CONVERT_REPLY
:
3071 case DLM_MSG_UNLOCK_REPLY
:
3072 case DLM_MSG_CANCEL_REPLY
:
3075 if (!is_process_copy(lkb
) || lkb
->lkb_nodeid
!= from
)
3079 case DLM_MSG_REQUEST_REPLY
:
3080 if (!is_process_copy(lkb
))
3082 else if (lkb
->lkb_nodeid
!= -1 && lkb
->lkb_nodeid
!= from
)
3091 log_error(lkb
->lkb_resource
->res_ls
,
3092 "ignore invalid message %d from %d %x %x %x %d",
3093 ms
->m_type
, from
, lkb
->lkb_id
, lkb
->lkb_remid
,
3094 lkb
->lkb_flags
, lkb
->lkb_nodeid
);
3098 static void receive_request(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3100 struct dlm_lkb
*lkb
;
3104 error
= create_lkb(ls
, &lkb
);
3108 receive_flags(lkb
, ms
);
3109 lkb
->lkb_flags
|= DLM_IFL_MSTCPY
;
3110 error
= receive_request_args(ls
, lkb
, ms
);
3116 namelen
= receive_extralen(ms
);
3118 error
= find_rsb(ls
, ms
->m_extra
, namelen
, R_MASTER
, &r
);
3127 error
= do_request(r
, lkb
);
3128 send_request_reply(r
, lkb
, error
);
3133 if (error
== -EINPROGRESS
)
3140 setup_stub_lkb(ls
, ms
);
3141 send_request_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
3144 static void receive_convert(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3146 struct dlm_lkb
*lkb
;
3148 int error
, reply
= 1;
3150 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3154 r
= lkb
->lkb_resource
;
3159 error
= validate_message(lkb
, ms
);
3163 receive_flags(lkb
, ms
);
3164 error
= receive_convert_args(ls
, lkb
, ms
);
3167 reply
= !down_conversion(lkb
);
3169 error
= do_convert(r
, lkb
);
3172 send_convert_reply(r
, lkb
, error
);
3180 setup_stub_lkb(ls
, ms
);
3181 send_convert_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
3184 static void receive_unlock(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3186 struct dlm_lkb
*lkb
;
3190 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3194 r
= lkb
->lkb_resource
;
3199 error
= validate_message(lkb
, ms
);
3203 receive_flags(lkb
, ms
);
3204 error
= receive_unlock_args(ls
, lkb
, ms
);
3208 error
= do_unlock(r
, lkb
);
3210 send_unlock_reply(r
, lkb
, error
);
3218 setup_stub_lkb(ls
, ms
);
3219 send_unlock_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
3222 static void receive_cancel(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3224 struct dlm_lkb
*lkb
;
3228 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3232 receive_flags(lkb
, ms
);
3234 r
= lkb
->lkb_resource
;
3239 error
= validate_message(lkb
, ms
);
3243 error
= do_cancel(r
, lkb
);
3244 send_cancel_reply(r
, lkb
, error
);
3252 setup_stub_lkb(ls
, ms
);
3253 send_cancel_reply(&ls
->ls_stub_rsb
, &ls
->ls_stub_lkb
, error
);
3256 static void receive_grant(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3258 struct dlm_lkb
*lkb
;
3262 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3264 log_debug(ls
, "receive_grant from %d no lkb %x",
3265 ms
->m_header
.h_nodeid
, ms
->m_remid
);
3269 r
= lkb
->lkb_resource
;
3274 error
= validate_message(lkb
, ms
);
3278 receive_flags_reply(lkb
, ms
);
3279 if (is_altmode(lkb
))
3280 munge_altmode(lkb
, ms
);
3281 grant_lock_pc(r
, lkb
, ms
);
3282 queue_cast(r
, lkb
, 0);
3289 static void receive_bast(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3291 struct dlm_lkb
*lkb
;
3295 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3297 log_debug(ls
, "receive_bast from %d no lkb %x",
3298 ms
->m_header
.h_nodeid
, ms
->m_remid
);
3302 r
= lkb
->lkb_resource
;
3307 error
= validate_message(lkb
, ms
);
3311 queue_bast(r
, lkb
, ms
->m_bastmode
);
3318 static void receive_lookup(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3320 int len
, error
, ret_nodeid
, dir_nodeid
, from_nodeid
, our_nodeid
;
3322 from_nodeid
= ms
->m_header
.h_nodeid
;
3323 our_nodeid
= dlm_our_nodeid();
3325 len
= receive_extralen(ms
);
3327 dir_nodeid
= dlm_hash2nodeid(ls
, ms
->m_hash
);
3328 if (dir_nodeid
!= our_nodeid
) {
3329 log_error(ls
, "lookup dir_nodeid %d from %d",
3330 dir_nodeid
, from_nodeid
);
3336 error
= dlm_dir_lookup(ls
, from_nodeid
, ms
->m_extra
, len
, &ret_nodeid
);
3338 /* Optimization: we're master so treat lookup as a request */
3339 if (!error
&& ret_nodeid
== our_nodeid
) {
3340 receive_request(ls
, ms
);
3344 send_lookup_reply(ls
, ms
, ret_nodeid
, error
);
3347 static void receive_remove(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3349 int len
, dir_nodeid
, from_nodeid
;
3351 from_nodeid
= ms
->m_header
.h_nodeid
;
3353 len
= receive_extralen(ms
);
3355 dir_nodeid
= dlm_hash2nodeid(ls
, ms
->m_hash
);
3356 if (dir_nodeid
!= dlm_our_nodeid()) {
3357 log_error(ls
, "remove dir entry dir_nodeid %d from %d",
3358 dir_nodeid
, from_nodeid
);
3362 dlm_dir_remove_entry(ls
, from_nodeid
, ms
->m_extra
, len
);
3365 static void receive_purge(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3367 do_purge(ls
, ms
->m_nodeid
, ms
->m_pid
);
3370 static void receive_request_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3372 struct dlm_lkb
*lkb
;
3374 int error
, mstype
, result
;
3376 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3378 log_debug(ls
, "receive_request_reply from %d no lkb %x",
3379 ms
->m_header
.h_nodeid
, ms
->m_remid
);
3383 r
= lkb
->lkb_resource
;
3387 error
= validate_message(lkb
, ms
);
3391 mstype
= lkb
->lkb_wait_type
;
3392 error
= remove_from_waiters(lkb
, DLM_MSG_REQUEST_REPLY
);
3396 /* Optimization: the dir node was also the master, so it took our
3397 lookup as a request and sent request reply instead of lookup reply */
3398 if (mstype
== DLM_MSG_LOOKUP
) {
3399 r
->res_nodeid
= ms
->m_header
.h_nodeid
;
3400 lkb
->lkb_nodeid
= r
->res_nodeid
;
3403 /* this is the value returned from do_request() on the master */
3404 result
= ms
->m_result
;
3408 /* request would block (be queued) on remote master */
3409 queue_cast(r
, lkb
, -EAGAIN
);
3410 confirm_master(r
, -EAGAIN
);
3411 unhold_lkb(lkb
); /* undoes create_lkb() */
3416 /* request was queued or granted on remote master */
3417 receive_flags_reply(lkb
, ms
);
3418 lkb
->lkb_remid
= ms
->m_lkid
;
3419 if (is_altmode(lkb
))
3420 munge_altmode(lkb
, ms
);
3422 add_lkb(r
, lkb
, DLM_LKSTS_WAITING
);
3425 grant_lock_pc(r
, lkb
, ms
);
3426 queue_cast(r
, lkb
, 0);
3428 confirm_master(r
, result
);
3433 /* find_rsb failed to find rsb or rsb wasn't master */
3434 log_debug(ls
, "receive_request_reply %x %x master diff %d %d",
3435 lkb
->lkb_id
, lkb
->lkb_flags
, r
->res_nodeid
, result
);
3437 lkb
->lkb_nodeid
= -1;
3439 if (is_overlap(lkb
)) {
3440 /* we'll ignore error in cancel/unlock reply */
3441 queue_cast_overlap(r
, lkb
);
3442 confirm_master(r
, result
);
3443 unhold_lkb(lkb
); /* undoes create_lkb() */
3445 _request_lock(r
, lkb
);
3449 log_error(ls
, "receive_request_reply %x error %d",
3450 lkb
->lkb_id
, result
);
3453 if (is_overlap_unlock(lkb
) && (result
== 0 || result
== -EINPROGRESS
)) {
3454 log_debug(ls
, "receive_request_reply %x result %d unlock",
3455 lkb
->lkb_id
, result
);
3456 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3457 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3458 send_unlock(r
, lkb
);
3459 } else if (is_overlap_cancel(lkb
) && (result
== -EINPROGRESS
)) {
3460 log_debug(ls
, "receive_request_reply %x cancel", lkb
->lkb_id
);
3461 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3462 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3463 send_cancel(r
, lkb
);
3465 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
3466 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
3474 static void __receive_convert_reply(struct dlm_rsb
*r
, struct dlm_lkb
*lkb
,
3475 struct dlm_message
*ms
)
3477 /* this is the value returned from do_convert() on the master */
3478 switch (ms
->m_result
) {
3480 /* convert would block (be queued) on remote master */
3481 queue_cast(r
, lkb
, -EAGAIN
);
3485 receive_flags_reply(lkb
, ms
);
3486 revert_lock_pc(r
, lkb
);
3487 queue_cast(r
, lkb
, -EDEADLK
);
3491 /* convert was queued on remote master */
3492 receive_flags_reply(lkb
, ms
);
3493 if (is_demoted(lkb
))
3494 munge_demoted(lkb
, ms
);
3496 add_lkb(r
, lkb
, DLM_LKSTS_CONVERT
);
3501 /* convert was granted on remote master */
3502 receive_flags_reply(lkb
, ms
);
3503 if (is_demoted(lkb
))
3504 munge_demoted(lkb
, ms
);
3505 grant_lock_pc(r
, lkb
, ms
);
3506 queue_cast(r
, lkb
, 0);
3510 log_error(r
->res_ls
, "receive_convert_reply %x error %d",
3511 lkb
->lkb_id
, ms
->m_result
);
3515 static void _receive_convert_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3517 struct dlm_rsb
*r
= lkb
->lkb_resource
;
3523 error
= validate_message(lkb
, ms
);
3527 /* stub reply can happen with waiters_mutex held */
3528 error
= remove_from_waiters_ms(lkb
, ms
);
3532 __receive_convert_reply(r
, lkb
, ms
);
3538 static void receive_convert_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3540 struct dlm_lkb
*lkb
;
3543 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3545 log_debug(ls
, "receive_convert_reply from %d no lkb %x",
3546 ms
->m_header
.h_nodeid
, ms
->m_remid
);
3550 _receive_convert_reply(lkb
, ms
);
3554 static void _receive_unlock_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3556 struct dlm_rsb
*r
= lkb
->lkb_resource
;
3562 error
= validate_message(lkb
, ms
);
3566 /* stub reply can happen with waiters_mutex held */
3567 error
= remove_from_waiters_ms(lkb
, ms
);
3571 /* this is the value returned from do_unlock() on the master */
3573 switch (ms
->m_result
) {
3575 receive_flags_reply(lkb
, ms
);
3576 remove_lock_pc(r
, lkb
);
3577 queue_cast(r
, lkb
, -DLM_EUNLOCK
);
3582 log_error(r
->res_ls
, "receive_unlock_reply %x error %d",
3583 lkb
->lkb_id
, ms
->m_result
);
3590 static void receive_unlock_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3592 struct dlm_lkb
*lkb
;
3595 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3597 log_debug(ls
, "receive_unlock_reply from %d no lkb %x",
3598 ms
->m_header
.h_nodeid
, ms
->m_remid
);
3602 _receive_unlock_reply(lkb
, ms
);
3606 static void _receive_cancel_reply(struct dlm_lkb
*lkb
, struct dlm_message
*ms
)
3608 struct dlm_rsb
*r
= lkb
->lkb_resource
;
3614 error
= validate_message(lkb
, ms
);
3618 /* stub reply can happen with waiters_mutex held */
3619 error
= remove_from_waiters_ms(lkb
, ms
);
3623 /* this is the value returned from do_cancel() on the master */
3625 switch (ms
->m_result
) {
3627 receive_flags_reply(lkb
, ms
);
3628 revert_lock_pc(r
, lkb
);
3629 queue_cast(r
, lkb
, -DLM_ECANCEL
);
3634 log_error(r
->res_ls
, "receive_cancel_reply %x error %d",
3635 lkb
->lkb_id
, ms
->m_result
);
3642 static void receive_cancel_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3644 struct dlm_lkb
*lkb
;
3647 error
= find_lkb(ls
, ms
->m_remid
, &lkb
);
3649 log_debug(ls
, "receive_cancel_reply from %d no lkb %x",
3650 ms
->m_header
.h_nodeid
, ms
->m_remid
);
3654 _receive_cancel_reply(lkb
, ms
);
3658 static void receive_lookup_reply(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3660 struct dlm_lkb
*lkb
;
3662 int error
, ret_nodeid
;
3664 error
= find_lkb(ls
, ms
->m_lkid
, &lkb
);
3666 log_error(ls
, "receive_lookup_reply no lkb");
3670 /* ms->m_result is the value returned by dlm_dir_lookup on dir node
3671 FIXME: will a non-zero error ever be returned? */
3673 r
= lkb
->lkb_resource
;
3677 error
= remove_from_waiters(lkb
, DLM_MSG_LOOKUP_REPLY
);
3681 ret_nodeid
= ms
->m_nodeid
;
3682 if (ret_nodeid
== dlm_our_nodeid()) {
3685 r
->res_first_lkid
= 0;
3687 /* set_master() will copy res_nodeid to lkb_nodeid */
3688 r
->res_nodeid
= ret_nodeid
;
3691 if (is_overlap(lkb
)) {
3692 log_debug(ls
, "receive_lookup_reply %x unlock %x",
3693 lkb
->lkb_id
, lkb
->lkb_flags
);
3694 queue_cast_overlap(r
, lkb
);
3695 unhold_lkb(lkb
); /* undoes create_lkb() */
3699 _request_lock(r
, lkb
);
3703 process_lookup_list(r
);
3710 static void _receive_message(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3712 if (!dlm_is_member(ls
, ms
->m_header
.h_nodeid
)) {
3713 log_debug(ls
, "ignore non-member message %d from %d %x %x %d",
3714 ms
->m_type
, ms
->m_header
.h_nodeid
, ms
->m_lkid
,
3715 ms
->m_remid
, ms
->m_result
);
3719 switch (ms
->m_type
) {
3721 /* messages sent to a master node */
3723 case DLM_MSG_REQUEST
:
3724 receive_request(ls
, ms
);
3727 case DLM_MSG_CONVERT
:
3728 receive_convert(ls
, ms
);
3731 case DLM_MSG_UNLOCK
:
3732 receive_unlock(ls
, ms
);
3735 case DLM_MSG_CANCEL
:
3736 receive_cancel(ls
, ms
);
3739 /* messages sent from a master node (replies to above) */
3741 case DLM_MSG_REQUEST_REPLY
:
3742 receive_request_reply(ls
, ms
);
3745 case DLM_MSG_CONVERT_REPLY
:
3746 receive_convert_reply(ls
, ms
);
3749 case DLM_MSG_UNLOCK_REPLY
:
3750 receive_unlock_reply(ls
, ms
);
3753 case DLM_MSG_CANCEL_REPLY
:
3754 receive_cancel_reply(ls
, ms
);
3757 /* messages sent from a master node (only two types of async msg) */
3760 receive_grant(ls
, ms
);
3764 receive_bast(ls
, ms
);
3767 /* messages sent to a dir node */
3769 case DLM_MSG_LOOKUP
:
3770 receive_lookup(ls
, ms
);
3773 case DLM_MSG_REMOVE
:
3774 receive_remove(ls
, ms
);
3777 /* messages sent from a dir node (remove has no reply) */
3779 case DLM_MSG_LOOKUP_REPLY
:
3780 receive_lookup_reply(ls
, ms
);
3783 /* other messages */
3786 receive_purge(ls
, ms
);
3790 log_error(ls
, "unknown message type %d", ms
->m_type
);
3796 /* If the lockspace is in recovery mode (locking stopped), then normal
3797 messages are saved on the requestqueue for processing after recovery is
3798 done. When not in recovery mode, we wait for dlm_recoverd to drain saved
3799 messages off the requestqueue before we process new ones. This occurs right
3800 after recovery completes when we transition from saving all messages on
3801 requestqueue, to processing all the saved messages, to processing new
3802 messages as they arrive. */
3804 static void dlm_receive_message(struct dlm_ls
*ls
, struct dlm_message
*ms
,
3807 if (dlm_locking_stopped(ls
)) {
3808 dlm_add_requestqueue(ls
, nodeid
, (struct dlm_header
*) ms
);
3810 dlm_wait_requestqueue(ls
);
3811 _receive_message(ls
, ms
);
3815 /* This is called by dlm_recoverd to process messages that were saved on
3816 the requestqueue. */
3818 void dlm_receive_message_saved(struct dlm_ls
*ls
, struct dlm_message
*ms
)
3820 _receive_message(ls
, ms
);
3823 /* This is called by the midcomms layer when something is received for
3824 the lockspace. It could be either a MSG (normal message sent as part of
3825 standard locking activity) or an RCOM (recovery message sent as part of
3826 lockspace recovery). */
3828 void dlm_receive_buffer(struct dlm_header
*hd
, int nodeid
)
3830 struct dlm_message
*ms
= (struct dlm_message
*) hd
;
3831 struct dlm_rcom
*rc
= (struct dlm_rcom
*) hd
;
3835 switch (hd
->h_cmd
) {
3845 log_print("invalid h_cmd %d from %u", hd
->h_cmd
, nodeid
);
3849 if (hd
->h_nodeid
!= nodeid
) {
3850 log_print("invalid h_nodeid %d from %d lockspace %x",
3851 hd
->h_nodeid
, nodeid
, hd
->h_lockspace
);
3855 ls
= dlm_find_lockspace_global(hd
->h_lockspace
);
3857 log_print("invalid h_lockspace %x from %d cmd %d type %d",
3858 hd
->h_lockspace
, nodeid
, hd
->h_cmd
, type
);
3860 if (hd
->h_cmd
== DLM_RCOM
&& type
== DLM_RCOM_STATUS
)
3861 dlm_send_ls_not_ready(nodeid
, rc
);
3865 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
3866 be inactive (in this ls) before transitioning to recovery mode */
3868 down_read(&ls
->ls_recv_active
);
3869 if (hd
->h_cmd
== DLM_MSG
)
3870 dlm_receive_message(ls
, ms
, nodeid
);
3872 dlm_receive_rcom(ls
, rc
, nodeid
);
3873 up_read(&ls
->ls_recv_active
);
3875 dlm_put_lockspace(ls
);
3878 static void recover_convert_waiter(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
3880 if (middle_conversion(lkb
)) {
3882 ls
->ls_stub_ms
.m_type
= DLM_MSG_CONVERT_REPLY
;
3883 ls
->ls_stub_ms
.m_result
= -EINPROGRESS
;
3884 ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
3885 ls
->ls_stub_ms
.m_header
.h_nodeid
= lkb
->lkb_nodeid
;
3886 _receive_convert_reply(lkb
, &ls
->ls_stub_ms
);
3888 /* Same special case as in receive_rcom_lock_args() */
3889 lkb
->lkb_grmode
= DLM_LOCK_IV
;
3890 rsb_set_flag(lkb
->lkb_resource
, RSB_RECOVER_CONVERT
);
3893 } else if (lkb
->lkb_rqmode
>= lkb
->lkb_grmode
) {
3894 lkb
->lkb_flags
|= DLM_IFL_RESEND
;
3897 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
3898 conversions are async; there's no reply from the remote master */
3901 /* A waiting lkb needs recovery if the master node has failed, or
3902 the master node is changing (only when no directory is used) */
3904 static int waiter_needs_recovery(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
3906 if (dlm_is_removed(ls
, lkb
->lkb_nodeid
))
3909 if (!dlm_no_directory(ls
))
3912 if (dlm_dir_nodeid(lkb
->lkb_resource
) != lkb
->lkb_nodeid
)
3918 /* Recovery for locks that are waiting for replies from nodes that are now
3919 gone. We can just complete unlocks and cancels by faking a reply from the
3920 dead node. Requests and up-conversions we flag to be resent after
3921 recovery. Down-conversions can just be completed with a fake reply like
3922 unlocks. Conversions between PR and CW need special attention. */
3924 void dlm_recover_waiters_pre(struct dlm_ls
*ls
)
3926 struct dlm_lkb
*lkb
, *safe
;
3927 int wait_type
, stub_unlock_result
, stub_cancel_result
;
3929 mutex_lock(&ls
->ls_waiters_mutex
);
3931 list_for_each_entry_safe(lkb
, safe
, &ls
->ls_waiters
, lkb_wait_reply
) {
3932 log_debug(ls
, "pre recover waiter lkid %x type %d flags %x",
3933 lkb
->lkb_id
, lkb
->lkb_wait_type
, lkb
->lkb_flags
);
3935 /* all outstanding lookups, regardless of destination will be
3936 resent after recovery is done */
3938 if (lkb
->lkb_wait_type
== DLM_MSG_LOOKUP
) {
3939 lkb
->lkb_flags
|= DLM_IFL_RESEND
;
3943 if (!waiter_needs_recovery(ls
, lkb
))
3946 wait_type
= lkb
->lkb_wait_type
;
3947 stub_unlock_result
= -DLM_EUNLOCK
;
3948 stub_cancel_result
= -DLM_ECANCEL
;
3950 /* Main reply may have been received leaving a zero wait_type,
3951 but a reply for the overlapping op may not have been
3952 received. In that case we need to fake the appropriate
3953 reply for the overlap op. */
3956 if (is_overlap_cancel(lkb
)) {
3957 wait_type
= DLM_MSG_CANCEL
;
3958 if (lkb
->lkb_grmode
== DLM_LOCK_IV
)
3959 stub_cancel_result
= 0;
3961 if (is_overlap_unlock(lkb
)) {
3962 wait_type
= DLM_MSG_UNLOCK
;
3963 if (lkb
->lkb_grmode
== DLM_LOCK_IV
)
3964 stub_unlock_result
= -ENOENT
;
3967 log_debug(ls
, "rwpre overlap %x %x %d %d %d",
3968 lkb
->lkb_id
, lkb
->lkb_flags
, wait_type
,
3969 stub_cancel_result
, stub_unlock_result
);
3972 switch (wait_type
) {
3974 case DLM_MSG_REQUEST
:
3975 lkb
->lkb_flags
|= DLM_IFL_RESEND
;
3978 case DLM_MSG_CONVERT
:
3979 recover_convert_waiter(ls
, lkb
);
3982 case DLM_MSG_UNLOCK
:
3984 ls
->ls_stub_ms
.m_type
= DLM_MSG_UNLOCK_REPLY
;
3985 ls
->ls_stub_ms
.m_result
= stub_unlock_result
;
3986 ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
3987 ls
->ls_stub_ms
.m_header
.h_nodeid
= lkb
->lkb_nodeid
;
3988 _receive_unlock_reply(lkb
, &ls
->ls_stub_ms
);
3992 case DLM_MSG_CANCEL
:
3994 ls
->ls_stub_ms
.m_type
= DLM_MSG_CANCEL_REPLY
;
3995 ls
->ls_stub_ms
.m_result
= stub_cancel_result
;
3996 ls
->ls_stub_ms
.m_flags
= lkb
->lkb_flags
;
3997 ls
->ls_stub_ms
.m_header
.h_nodeid
= lkb
->lkb_nodeid
;
3998 _receive_cancel_reply(lkb
, &ls
->ls_stub_ms
);
4003 log_error(ls
, "invalid lkb wait_type %d %d",
4004 lkb
->lkb_wait_type
, wait_type
);
4008 mutex_unlock(&ls
->ls_waiters_mutex
);
4011 static struct dlm_lkb
*find_resend_waiter(struct dlm_ls
*ls
)
4013 struct dlm_lkb
*lkb
;
4016 mutex_lock(&ls
->ls_waiters_mutex
);
4017 list_for_each_entry(lkb
, &ls
->ls_waiters
, lkb_wait_reply
) {
4018 if (lkb
->lkb_flags
& DLM_IFL_RESEND
) {
4024 mutex_unlock(&ls
->ls_waiters_mutex
);
4031 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the
4032 master or dir-node for r. Processing the lkb may result in it being placed
4035 /* We do this after normal locking has been enabled and any saved messages
4036 (in requestqueue) have been processed. We should be confident that at
4037 this point we won't get or process a reply to any of these waiting
4038 operations. But, new ops may be coming in on the rsbs/locks here from
4039 userspace or remotely. */
4041 /* there may have been an overlap unlock/cancel prior to recovery or after
4042 recovery. if before, the lkb may still have a pos wait_count; if after, the
4043 overlap flag would just have been set and nothing new sent. we can be
4044 confident here than any replies to either the initial op or overlap ops
4045 prior to recovery have been received. */
4047 int dlm_recover_waiters_post(struct dlm_ls
*ls
)
4049 struct dlm_lkb
*lkb
;
4051 int error
= 0, mstype
, err
, oc
, ou
;
4054 if (dlm_locking_stopped(ls
)) {
4055 log_debug(ls
, "recover_waiters_post aborted");
4060 lkb
= find_resend_waiter(ls
);
4064 r
= lkb
->lkb_resource
;
4068 mstype
= lkb
->lkb_wait_type
;
4069 oc
= is_overlap_cancel(lkb
);
4070 ou
= is_overlap_unlock(lkb
);
4073 log_debug(ls
, "recover_waiters_post %x type %d flags %x %s",
4074 lkb
->lkb_id
, mstype
, lkb
->lkb_flags
, r
->res_name
);
4076 /* At this point we assume that we won't get a reply to any
4077 previous op or overlap op on this lock. First, do a big
4078 remove_from_waiters() for all previous ops. */
4080 lkb
->lkb_flags
&= ~DLM_IFL_RESEND
;
4081 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_UNLOCK
;
4082 lkb
->lkb_flags
&= ~DLM_IFL_OVERLAP_CANCEL
;
4083 lkb
->lkb_wait_type
= 0;
4084 lkb
->lkb_wait_count
= 0;
4085 mutex_lock(&ls
->ls_waiters_mutex
);
4086 list_del_init(&lkb
->lkb_wait_reply
);
4087 mutex_unlock(&ls
->ls_waiters_mutex
);
4088 unhold_lkb(lkb
); /* for waiters list */
4091 /* do an unlock or cancel instead of resending */
4093 case DLM_MSG_LOOKUP
:
4094 case DLM_MSG_REQUEST
:
4095 queue_cast(r
, lkb
, ou
? -DLM_EUNLOCK
:
4097 unhold_lkb(lkb
); /* undoes create_lkb() */
4099 case DLM_MSG_CONVERT
:
4101 queue_cast(r
, lkb
, -DLM_ECANCEL
);
4103 lkb
->lkb_exflags
|= DLM_LKF_FORCEUNLOCK
;
4104 _unlock_lock(r
, lkb
);
4112 case DLM_MSG_LOOKUP
:
4113 case DLM_MSG_REQUEST
:
4114 _request_lock(r
, lkb
);
4116 confirm_master(r
, 0);
4118 case DLM_MSG_CONVERT
:
4119 _convert_lock(r
, lkb
);
4127 log_error(ls
, "recover_waiters_post %x %d %x %d %d",
4128 lkb
->lkb_id
, mstype
, lkb
->lkb_flags
, oc
, ou
);
4137 static void purge_queue(struct dlm_rsb
*r
, struct list_head
*queue
,
4138 int (*test
)(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
))
4140 struct dlm_ls
*ls
= r
->res_ls
;
4141 struct dlm_lkb
*lkb
, *safe
;
4143 list_for_each_entry_safe(lkb
, safe
, queue
, lkb_statequeue
) {
4144 if (test(ls
, lkb
)) {
4145 rsb_set_flag(r
, RSB_LOCKS_PURGED
);
4147 /* this put should free the lkb */
4148 if (!dlm_put_lkb(lkb
))
4149 log_error(ls
, "purged lkb not released");
4154 static int purge_dead_test(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
4156 return (is_master_copy(lkb
) && dlm_is_removed(ls
, lkb
->lkb_nodeid
));
4159 static int purge_mstcpy_test(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
4161 return is_master_copy(lkb
);
4164 static void purge_dead_locks(struct dlm_rsb
*r
)
4166 purge_queue(r
, &r
->res_grantqueue
, &purge_dead_test
);
4167 purge_queue(r
, &r
->res_convertqueue
, &purge_dead_test
);
4168 purge_queue(r
, &r
->res_waitqueue
, &purge_dead_test
);
4171 void dlm_purge_mstcpy_locks(struct dlm_rsb
*r
)
4173 purge_queue(r
, &r
->res_grantqueue
, &purge_mstcpy_test
);
4174 purge_queue(r
, &r
->res_convertqueue
, &purge_mstcpy_test
);
4175 purge_queue(r
, &r
->res_waitqueue
, &purge_mstcpy_test
);
4178 /* Get rid of locks held by nodes that are gone. */
4180 int dlm_purge_locks(struct dlm_ls
*ls
)
4184 log_debug(ls
, "dlm_purge_locks");
4186 down_write(&ls
->ls_root_sem
);
4187 list_for_each_entry(r
, &ls
->ls_root_list
, res_root_list
) {
4191 purge_dead_locks(r
);
4197 up_write(&ls
->ls_root_sem
);
4202 static struct dlm_rsb
*find_purged_rsb(struct dlm_ls
*ls
, int bucket
)
4204 struct dlm_rsb
*r
, *r_ret
= NULL
;
4206 read_lock(&ls
->ls_rsbtbl
[bucket
].lock
);
4207 list_for_each_entry(r
, &ls
->ls_rsbtbl
[bucket
].list
, res_hashchain
) {
4208 if (!rsb_flag(r
, RSB_LOCKS_PURGED
))
4211 rsb_clear_flag(r
, RSB_LOCKS_PURGED
);
4215 read_unlock(&ls
->ls_rsbtbl
[bucket
].lock
);
4219 void dlm_grant_after_purge(struct dlm_ls
*ls
)
4225 r
= find_purged_rsb(ls
, bucket
);
4227 if (bucket
== ls
->ls_rsbtbl_size
- 1)
4234 grant_pending_locks(r
);
4235 confirm_master(r
, 0);
4243 static struct dlm_lkb
*search_remid_list(struct list_head
*head
, int nodeid
,
4246 struct dlm_lkb
*lkb
;
4248 list_for_each_entry(lkb
, head
, lkb_statequeue
) {
4249 if (lkb
->lkb_nodeid
== nodeid
&& lkb
->lkb_remid
== remid
)
4255 static struct dlm_lkb
*search_remid(struct dlm_rsb
*r
, int nodeid
,
4258 struct dlm_lkb
*lkb
;
4260 lkb
= search_remid_list(&r
->res_grantqueue
, nodeid
, remid
);
4263 lkb
= search_remid_list(&r
->res_convertqueue
, nodeid
, remid
);
4266 lkb
= search_remid_list(&r
->res_waitqueue
, nodeid
, remid
);
4272 static int receive_rcom_lock_args(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
,
4273 struct dlm_rsb
*r
, struct dlm_rcom
*rc
)
4275 struct rcom_lock
*rl
= (struct rcom_lock
*) rc
->rc_buf
;
4278 lkb
->lkb_nodeid
= rc
->rc_header
.h_nodeid
;
4279 lkb
->lkb_ownpid
= rl
->rl_ownpid
;
4280 lkb
->lkb_remid
= rl
->rl_lkid
;
4281 lkb
->lkb_exflags
= rl
->rl_exflags
;
4282 lkb
->lkb_flags
= rl
->rl_flags
& 0x0000FFFF;
4283 lkb
->lkb_flags
|= DLM_IFL_MSTCPY
;
4284 lkb
->lkb_lvbseq
= rl
->rl_lvbseq
;
4285 lkb
->lkb_rqmode
= rl
->rl_rqmode
;
4286 lkb
->lkb_grmode
= rl
->rl_grmode
;
4287 /* don't set lkb_status because add_lkb wants to itself */
4289 lkb
->lkb_bastaddr
= (void *) (long) (rl
->rl_asts
& AST_BAST
);
4290 lkb
->lkb_astaddr
= (void *) (long) (rl
->rl_asts
& AST_COMP
);
4292 if (lkb
->lkb_exflags
& DLM_LKF_VALBLK
) {
4293 lkb
->lkb_lvbptr
= dlm_allocate_lvb(ls
);
4294 if (!lkb
->lkb_lvbptr
)
4296 lvblen
= rc
->rc_header
.h_length
- sizeof(struct dlm_rcom
) -
4297 sizeof(struct rcom_lock
);
4298 memcpy(lkb
->lkb_lvbptr
, rl
->rl_lvb
, lvblen
);
4301 /* Conversions between PR and CW (middle modes) need special handling.
4302 The real granted mode of these converting locks cannot be determined
4303 until all locks have been rebuilt on the rsb (recover_conversion) */
4305 if (rl
->rl_wait_type
== DLM_MSG_CONVERT
&& middle_conversion(lkb
)) {
4306 rl
->rl_status
= DLM_LKSTS_CONVERT
;
4307 lkb
->lkb_grmode
= DLM_LOCK_IV
;
4308 rsb_set_flag(r
, RSB_RECOVER_CONVERT
);
4314 /* This lkb may have been recovered in a previous aborted recovery so we need
4315 to check if the rsb already has an lkb with the given remote nodeid/lkid.
4316 If so we just send back a standard reply. If not, we create a new lkb with
4317 the given values and send back our lkid. We send back our lkid by sending
4318 back the rcom_lock struct we got but with the remid field filled in. */
4320 int dlm_recover_master_copy(struct dlm_ls
*ls
, struct dlm_rcom
*rc
)
4322 struct rcom_lock
*rl
= (struct rcom_lock
*) rc
->rc_buf
;
4324 struct dlm_lkb
*lkb
;
4327 if (rl
->rl_parent_lkid
) {
4328 error
= -EOPNOTSUPP
;
4332 error
= find_rsb(ls
, rl
->rl_name
, rl
->rl_namelen
, R_MASTER
, &r
);
4338 lkb
= search_remid(r
, rc
->rc_header
.h_nodeid
, rl
->rl_lkid
);
4344 error
= create_lkb(ls
, &lkb
);
4348 error
= receive_rcom_lock_args(ls
, lkb
, r
, rc
);
4355 add_lkb(r
, lkb
, rl
->rl_status
);
4359 /* this is the new value returned to the lock holder for
4360 saving in its process-copy lkb */
4361 rl
->rl_remid
= lkb
->lkb_id
;
4368 log_debug(ls
, "recover_master_copy %d %x", error
, rl
->rl_lkid
);
4369 rl
->rl_result
= error
;
4373 int dlm_recover_process_copy(struct dlm_ls
*ls
, struct dlm_rcom
*rc
)
4375 struct rcom_lock
*rl
= (struct rcom_lock
*) rc
->rc_buf
;
4377 struct dlm_lkb
*lkb
;
4380 error
= find_lkb(ls
, rl
->rl_lkid
, &lkb
);
4382 log_error(ls
, "recover_process_copy no lkid %x", rl
->rl_lkid
);
4386 DLM_ASSERT(is_process_copy(lkb
), dlm_print_lkb(lkb
););
4388 error
= rl
->rl_result
;
4390 r
= lkb
->lkb_resource
;
4396 /* There's a chance the new master received our lock before
4397 dlm_recover_master_reply(), this wouldn't happen if we did
4398 a barrier between recover_masters and recover_locks. */
4399 log_debug(ls
, "master copy not ready %x r %lx %s", lkb
->lkb_id
,
4400 (unsigned long)r
, r
->res_name
);
4401 dlm_send_rcom_lock(r
, lkb
);
4404 log_debug(ls
, "master copy exists %x", lkb
->lkb_id
);
4407 lkb
->lkb_remid
= rl
->rl_remid
;
4410 log_error(ls
, "dlm_recover_process_copy unknown error %d %x",
4411 error
, lkb
->lkb_id
);
4414 /* an ack for dlm_recover_locks() which waits for replies from
4415 all the locks it sends to new masters */
4416 dlm_recovered_lock(r
);
4425 int dlm_user_request(struct dlm_ls
*ls
, struct dlm_user_args
*ua
,
4426 int mode
, uint32_t flags
, void *name
, unsigned int namelen
,
4427 unsigned long timeout_cs
)
4429 struct dlm_lkb
*lkb
;
4430 struct dlm_args args
;
4433 dlm_lock_recovery(ls
);
4435 error
= create_lkb(ls
, &lkb
);
4441 if (flags
& DLM_LKF_VALBLK
) {
4442 ua
->lksb
.sb_lvbptr
= kzalloc(DLM_USER_LVB_LEN
, GFP_KERNEL
);
4443 if (!ua
->lksb
.sb_lvbptr
) {
4451 /* After ua is attached to lkb it will be freed by dlm_free_lkb().
4452 When DLM_IFL_USER is set, the dlm knows that this is a userspace
4453 lock and that lkb_astparam is the dlm_user_args structure. */
4455 error
= set_lock_args(mode
, &ua
->lksb
, flags
, namelen
, timeout_cs
,
4456 DLM_FAKE_USER_AST
, ua
, DLM_FAKE_USER_AST
, &args
);
4457 lkb
->lkb_flags
|= DLM_IFL_USER
;
4458 ua
->old_mode
= DLM_LOCK_IV
;
4465 error
= request_lock(ls
, lkb
, name
, namelen
, &args
);
4481 /* add this new lkb to the per-process list of locks */
4482 spin_lock(&ua
->proc
->locks_spin
);
4484 list_add_tail(&lkb
->lkb_ownqueue
, &ua
->proc
->locks
);
4485 spin_unlock(&ua
->proc
->locks_spin
);
4487 dlm_unlock_recovery(ls
);
4491 int dlm_user_convert(struct dlm_ls
*ls
, struct dlm_user_args
*ua_tmp
,
4492 int mode
, uint32_t flags
, uint32_t lkid
, char *lvb_in
,
4493 unsigned long timeout_cs
)
4495 struct dlm_lkb
*lkb
;
4496 struct dlm_args args
;
4497 struct dlm_user_args
*ua
;
4500 dlm_lock_recovery(ls
);
4502 error
= find_lkb(ls
, lkid
, &lkb
);
4506 /* user can change the params on its lock when it converts it, or
4507 add an lvb that didn't exist before */
4509 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4511 if (flags
& DLM_LKF_VALBLK
&& !ua
->lksb
.sb_lvbptr
) {
4512 ua
->lksb
.sb_lvbptr
= kzalloc(DLM_USER_LVB_LEN
, GFP_KERNEL
);
4513 if (!ua
->lksb
.sb_lvbptr
) {
4518 if (lvb_in
&& ua
->lksb
.sb_lvbptr
)
4519 memcpy(ua
->lksb
.sb_lvbptr
, lvb_in
, DLM_USER_LVB_LEN
);
4521 ua
->xid
= ua_tmp
->xid
;
4522 ua
->castparam
= ua_tmp
->castparam
;
4523 ua
->castaddr
= ua_tmp
->castaddr
;
4524 ua
->bastparam
= ua_tmp
->bastparam
;
4525 ua
->bastaddr
= ua_tmp
->bastaddr
;
4526 ua
->user_lksb
= ua_tmp
->user_lksb
;
4527 ua
->old_mode
= lkb
->lkb_grmode
;
4529 error
= set_lock_args(mode
, &ua
->lksb
, flags
, 0, timeout_cs
,
4530 DLM_FAKE_USER_AST
, ua
, DLM_FAKE_USER_AST
, &args
);
4534 error
= convert_lock(ls
, lkb
, &args
);
4536 if (error
== -EINPROGRESS
|| error
== -EAGAIN
|| error
== -EDEADLK
)
4541 dlm_unlock_recovery(ls
);
4546 int dlm_user_unlock(struct dlm_ls
*ls
, struct dlm_user_args
*ua_tmp
,
4547 uint32_t flags
, uint32_t lkid
, char *lvb_in
)
4549 struct dlm_lkb
*lkb
;
4550 struct dlm_args args
;
4551 struct dlm_user_args
*ua
;
4554 dlm_lock_recovery(ls
);
4556 error
= find_lkb(ls
, lkid
, &lkb
);
4560 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4562 if (lvb_in
&& ua
->lksb
.sb_lvbptr
)
4563 memcpy(ua
->lksb
.sb_lvbptr
, lvb_in
, DLM_USER_LVB_LEN
);
4564 if (ua_tmp
->castparam
)
4565 ua
->castparam
= ua_tmp
->castparam
;
4566 ua
->user_lksb
= ua_tmp
->user_lksb
;
4568 error
= set_unlock_args(flags
, ua
, &args
);
4572 error
= unlock_lock(ls
, lkb
, &args
);
4574 if (error
== -DLM_EUNLOCK
)
4576 /* from validate_unlock_args() */
4577 if (error
== -EBUSY
&& (flags
& DLM_LKF_FORCEUNLOCK
))
4582 spin_lock(&ua
->proc
->locks_spin
);
4583 /* dlm_user_add_ast() may have already taken lkb off the proc list */
4584 if (!list_empty(&lkb
->lkb_ownqueue
))
4585 list_move(&lkb
->lkb_ownqueue
, &ua
->proc
->unlocking
);
4586 spin_unlock(&ua
->proc
->locks_spin
);
4590 dlm_unlock_recovery(ls
);
4595 int dlm_user_cancel(struct dlm_ls
*ls
, struct dlm_user_args
*ua_tmp
,
4596 uint32_t flags
, uint32_t lkid
)
4598 struct dlm_lkb
*lkb
;
4599 struct dlm_args args
;
4600 struct dlm_user_args
*ua
;
4603 dlm_lock_recovery(ls
);
4605 error
= find_lkb(ls
, lkid
, &lkb
);
4609 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4610 if (ua_tmp
->castparam
)
4611 ua
->castparam
= ua_tmp
->castparam
;
4612 ua
->user_lksb
= ua_tmp
->user_lksb
;
4614 error
= set_unlock_args(flags
, ua
, &args
);
4618 error
= cancel_lock(ls
, lkb
, &args
);
4620 if (error
== -DLM_ECANCEL
)
4622 /* from validate_unlock_args() */
4623 if (error
== -EBUSY
)
4628 dlm_unlock_recovery(ls
);
4633 int dlm_user_deadlock(struct dlm_ls
*ls
, uint32_t flags
, uint32_t lkid
)
4635 struct dlm_lkb
*lkb
;
4636 struct dlm_args args
;
4637 struct dlm_user_args
*ua
;
4641 dlm_lock_recovery(ls
);
4643 error
= find_lkb(ls
, lkid
, &lkb
);
4647 ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4649 error
= set_unlock_args(flags
, ua
, &args
);
4653 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
4655 r
= lkb
->lkb_resource
;
4659 error
= validate_unlock_args(lkb
, &args
);
4662 lkb
->lkb_flags
|= DLM_IFL_DEADLOCK_CANCEL
;
4664 error
= _cancel_lock(r
, lkb
);
4669 if (error
== -DLM_ECANCEL
)
4671 /* from validate_unlock_args() */
4672 if (error
== -EBUSY
)
4677 dlm_unlock_recovery(ls
);
4681 /* lkb's that are removed from the waiters list by revert are just left on the
4682 orphans list with the granted orphan locks, to be freed by purge */
4684 static int orphan_proc_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
4686 struct dlm_user_args
*ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4687 struct dlm_args args
;
4691 mutex_lock(&ls
->ls_orphans_mutex
);
4692 list_add_tail(&lkb
->lkb_ownqueue
, &ls
->ls_orphans
);
4693 mutex_unlock(&ls
->ls_orphans_mutex
);
4695 set_unlock_args(0, ua
, &args
);
4697 error
= cancel_lock(ls
, lkb
, &args
);
4698 if (error
== -DLM_ECANCEL
)
4703 /* The force flag allows the unlock to go ahead even if the lkb isn't granted.
4704 Regardless of what rsb queue the lock is on, it's removed and freed. */
4706 static int unlock_proc_lock(struct dlm_ls
*ls
, struct dlm_lkb
*lkb
)
4708 struct dlm_user_args
*ua
= (struct dlm_user_args
*)lkb
->lkb_astparam
;
4709 struct dlm_args args
;
4712 set_unlock_args(DLM_LKF_FORCEUNLOCK
, ua
, &args
);
4714 error
= unlock_lock(ls
, lkb
, &args
);
4715 if (error
== -DLM_EUNLOCK
)
4720 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
4721 (which does lock_rsb) due to deadlock with receiving a message that does
4722 lock_rsb followed by dlm_user_add_ast() */
4724 static struct dlm_lkb
*del_proc_lock(struct dlm_ls
*ls
,
4725 struct dlm_user_proc
*proc
)
4727 struct dlm_lkb
*lkb
= NULL
;
4729 mutex_lock(&ls
->ls_clear_proc_locks
);
4730 if (list_empty(&proc
->locks
))
4733 lkb
= list_entry(proc
->locks
.next
, struct dlm_lkb
, lkb_ownqueue
);
4734 list_del_init(&lkb
->lkb_ownqueue
);
4736 if (lkb
->lkb_exflags
& DLM_LKF_PERSISTENT
)
4737 lkb
->lkb_flags
|= DLM_IFL_ORPHAN
;
4739 lkb
->lkb_flags
|= DLM_IFL_DEAD
;
4741 mutex_unlock(&ls
->ls_clear_proc_locks
);
4745 /* The ls_clear_proc_locks mutex protects against dlm_user_add_asts() which
4746 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
4747 which we clear here. */
4749 /* proc CLOSING flag is set so no more device_reads should look at proc->asts
4750 list, and no more device_writes should add lkb's to proc->locks list; so we
4751 shouldn't need to take asts_spin or locks_spin here. this assumes that
4752 device reads/writes/closes are serialized -- FIXME: we may need to serialize
4755 void dlm_clear_proc_locks(struct dlm_ls
*ls
, struct dlm_user_proc
*proc
)
4757 struct dlm_lkb
*lkb
, *safe
;
4759 dlm_lock_recovery(ls
);
4762 lkb
= del_proc_lock(ls
, proc
);
4766 if (lkb
->lkb_exflags
& DLM_LKF_PERSISTENT
)
4767 orphan_proc_lock(ls
, lkb
);
4769 unlock_proc_lock(ls
, lkb
);
4771 /* this removes the reference for the proc->locks list
4772 added by dlm_user_request, it may result in the lkb
4778 mutex_lock(&ls
->ls_clear_proc_locks
);
4780 /* in-progress unlocks */
4781 list_for_each_entry_safe(lkb
, safe
, &proc
->unlocking
, lkb_ownqueue
) {
4782 list_del_init(&lkb
->lkb_ownqueue
);
4783 lkb
->lkb_flags
|= DLM_IFL_DEAD
;
4787 list_for_each_entry_safe(lkb
, safe
, &proc
->asts
, lkb_astqueue
) {
4788 lkb
->lkb_ast_type
= 0;
4789 list_del(&lkb
->lkb_astqueue
);
4793 mutex_unlock(&ls
->ls_clear_proc_locks
);
4794 dlm_unlock_recovery(ls
);
4797 static void purge_proc_locks(struct dlm_ls
*ls
, struct dlm_user_proc
*proc
)
4799 struct dlm_lkb
*lkb
, *safe
;
4803 spin_lock(&proc
->locks_spin
);
4804 if (!list_empty(&proc
->locks
)) {
4805 lkb
= list_entry(proc
->locks
.next
, struct dlm_lkb
,
4807 list_del_init(&lkb
->lkb_ownqueue
);
4809 spin_unlock(&proc
->locks_spin
);
4814 lkb
->lkb_flags
|= DLM_IFL_DEAD
;
4815 unlock_proc_lock(ls
, lkb
);
4816 dlm_put_lkb(lkb
); /* ref from proc->locks list */
4819 spin_lock(&proc
->locks_spin
);
4820 list_for_each_entry_safe(lkb
, safe
, &proc
->unlocking
, lkb_ownqueue
) {
4821 list_del_init(&lkb
->lkb_ownqueue
);
4822 lkb
->lkb_flags
|= DLM_IFL_DEAD
;
4825 spin_unlock(&proc
->locks_spin
);
4827 spin_lock(&proc
->asts_spin
);
4828 list_for_each_entry_safe(lkb
, safe
, &proc
->asts
, lkb_astqueue
) {
4829 list_del(&lkb
->lkb_astqueue
);
4832 spin_unlock(&proc
->asts_spin
);
4835 /* pid of 0 means purge all orphans */
4837 static void do_purge(struct dlm_ls
*ls
, int nodeid
, int pid
)
4839 struct dlm_lkb
*lkb
, *safe
;
4841 mutex_lock(&ls
->ls_orphans_mutex
);
4842 list_for_each_entry_safe(lkb
, safe
, &ls
->ls_orphans
, lkb_ownqueue
) {
4843 if (pid
&& lkb
->lkb_ownpid
!= pid
)
4845 unlock_proc_lock(ls
, lkb
);
4846 list_del_init(&lkb
->lkb_ownqueue
);
4849 mutex_unlock(&ls
->ls_orphans_mutex
);
4852 static int send_purge(struct dlm_ls
*ls
, int nodeid
, int pid
)
4854 struct dlm_message
*ms
;
4855 struct dlm_mhandle
*mh
;
4858 error
= _create_message(ls
, sizeof(struct dlm_message
), nodeid
,
4859 DLM_MSG_PURGE
, &ms
, &mh
);
4862 ms
->m_nodeid
= nodeid
;
4865 return send_message(mh
, ms
);
4868 int dlm_user_purge(struct dlm_ls
*ls
, struct dlm_user_proc
*proc
,
4869 int nodeid
, int pid
)
4873 if (nodeid
!= dlm_our_nodeid()) {
4874 error
= send_purge(ls
, nodeid
, pid
);
4876 dlm_lock_recovery(ls
);
4877 if (pid
== current
->pid
)
4878 purge_proc_locks(ls
, proc
);
4880 do_purge(ls
, nodeid
, pid
);
4881 dlm_unlock_recovery(ls
);
This page took 0.152317 seconds and 5 git commands to generate.