1 /* -*- mode: c; c-basic-offset: 8; -*-
2 * vim: noexpandtab sw=8 ts=8 sts=0:
6 * underlying calls for unlocking locks
8 * Copyright (C) 2004 Oracle. All rights reserved.
10 * This program is free software; you can redistribute it and/or
11 * modify it under the terms of the GNU General Public
12 * License as published by the Free Software Foundation; either
13 * version 2 of the License, or (at your option) any later version.
15 * This program is distributed in the hope that it will be useful,
16 * but WITHOUT ANY WARRANTY; without even the implied warranty of
17 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
18 * General Public License for more details.
20 * You should have received a copy of the GNU General Public
21 * License along with this program; if not, write to the
22 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
23 * Boston, MA 021110-1307, USA.
28 #include <linux/module.h>
30 #include <linux/types.h>
31 #include <linux/slab.h>
32 #include <linux/highmem.h>
33 #include <linux/utsname.h>
34 #include <linux/init.h>
35 #include <linux/sysctl.h>
36 #include <linux/random.h>
37 #include <linux/blkdev.h>
38 #include <linux/socket.h>
39 #include <linux/inet.h>
40 #include <linux/spinlock.h>
41 #include <linux/delay.h>
43 #include "cluster/heartbeat.h"
44 #include "cluster/nodemanager.h"
45 #include "cluster/tcp.h"
48 #include "dlmcommon.h"
50 #define MLOG_MASK_PREFIX ML_DLM
51 #include "cluster/masklog.h"
53 #define DLM_UNLOCK_FREE_LOCK 0x00000001
54 #define DLM_UNLOCK_CALL_AST 0x00000002
55 #define DLM_UNLOCK_REMOVE_LOCK 0x00000004
56 #define DLM_UNLOCK_REGRANT_LOCK 0x00000008
57 #define DLM_UNLOCK_CLEAR_CONVERT_TYPE 0x00000010
60 static enum dlm_status
dlm_get_cancel_actions(struct dlm_ctxt
*dlm
,
61 struct dlm_lock_resource
*res
,
62 struct dlm_lock
*lock
,
63 struct dlm_lockstatus
*lksb
,
65 static enum dlm_status
dlm_get_unlock_actions(struct dlm_ctxt
*dlm
,
66 struct dlm_lock_resource
*res
,
67 struct dlm_lock
*lock
,
68 struct dlm_lockstatus
*lksb
,
71 static enum dlm_status
dlm_send_remote_unlock_request(struct dlm_ctxt
*dlm
,
72 struct dlm_lock_resource
*res
,
73 struct dlm_lock
*lock
,
74 struct dlm_lockstatus
*lksb
,
80 * according to the spec:
81 * http://opendlm.sourceforge.net/cvsmirror/opendlm/docs/dlmbook_final.pdf
83 * flags & LKM_CANCEL != 0: must be converting or blocked
84 * flags & LKM_CANCEL == 0: must be granted
86 * So to unlock a converting lock, you must first cancel the
87 * convert (passing LKM_CANCEL in flags), then call the unlock
88 * again (with no LKM_CANCEL in flags).
95 * taken: res->spinlock and lock->spinlock taken and dropped
97 * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
98 * all callers should have taken an extra ref on lock coming in
100 static enum dlm_status
dlmunlock_common(struct dlm_ctxt
*dlm
,
101 struct dlm_lock_resource
*res
,
102 struct dlm_lock
*lock
,
103 struct dlm_lockstatus
*lksb
,
104 int flags
, int *call_ast
,
107 enum dlm_status status
;
112 mlog(0, "master_node = %d, valblk = %d\n", master_node
,
116 BUG_ON(res
->owner
!= dlm
->node_num
);
118 BUG_ON(res
->owner
== dlm
->node_num
);
120 spin_lock(&dlm
->spinlock
);
121 /* We want to be sure that we're not freeing a lock
122 * that still has AST's pending... */
123 in_use
= !list_empty(&lock
->ast_list
);
124 spin_unlock(&dlm
->spinlock
);
126 mlog(ML_ERROR
, "lockres %.*s: Someone is calling dlmunlock "
127 "while waiting for an ast!", res
->lockname
.len
,
132 spin_lock(&res
->spinlock
);
133 if (res
->state
& DLM_LOCK_RES_IN_PROGRESS
) {
135 mlog(ML_ERROR
, "lockres in progress!\n");
136 spin_unlock(&res
->spinlock
);
139 /* ok for this to sleep if not in a network handler */
140 __dlm_wait_on_lockres(res
);
141 res
->state
|= DLM_LOCK_RES_IN_PROGRESS
;
143 spin_lock(&lock
->spinlock
);
145 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
146 status
= DLM_RECOVERING
;
151 /* see above for what the spec says about
152 * LKM_CANCEL and the lock queue state */
153 if (flags
& LKM_CANCEL
)
154 status
= dlm_get_cancel_actions(dlm
, res
, lock
, lksb
, &actions
);
156 status
= dlm_get_unlock_actions(dlm
, res
, lock
, lksb
, &actions
);
158 if (status
!= DLM_NORMAL
)
161 /* By now this has been masked out of cancel requests. */
162 if (flags
& LKM_VALBLK
) {
163 /* make the final update to the lvb */
165 memcpy(res
->lvb
, lksb
->lvb
, DLM_LVB_LEN
);
167 flags
|= LKM_PUT_LVB
; /* let the send function
173 /* drop locks and send message */
174 if (flags
& LKM_CANCEL
)
175 lock
->cancel_pending
= 1;
177 lock
->unlock_pending
= 1;
178 spin_unlock(&lock
->spinlock
);
179 spin_unlock(&res
->spinlock
);
180 status
= dlm_send_remote_unlock_request(dlm
, res
, lock
, lksb
,
182 spin_lock(&res
->spinlock
);
183 spin_lock(&lock
->spinlock
);
184 /* if the master told us the lock was already granted,
185 * let the ast handle all of these actions */
186 if (status
== DLM_NORMAL
&&
187 lksb
->status
== DLM_CANCELGRANT
) {
188 actions
&= ~(DLM_UNLOCK_REMOVE_LOCK
|
189 DLM_UNLOCK_REGRANT_LOCK
|
190 DLM_UNLOCK_CLEAR_CONVERT_TYPE
);
191 } else if (status
== DLM_RECOVERING
||
192 status
== DLM_MIGRATING
||
193 status
== DLM_FORWARD
) {
194 /* must clear the actions because this unlock
195 * is about to be retried. cannot free or do
196 * any list manipulation. */
197 mlog(0, "%s:%.*s: clearing actions, %s\n",
198 dlm
->name
, res
->lockname
.len
,
200 status
==DLM_RECOVERING
?"recovering":
201 (status
==DLM_MIGRATING
?"migrating":
205 if (flags
& LKM_CANCEL
)
206 lock
->cancel_pending
= 0;
208 lock
->unlock_pending
= 0;
212 /* get an extra ref on lock. if we are just switching
213 * lists here, we dont want the lock to go away. */
216 if (actions
& DLM_UNLOCK_REMOVE_LOCK
) {
217 list_del_init(&lock
->list
);
220 if (actions
& DLM_UNLOCK_REGRANT_LOCK
) {
222 list_add_tail(&lock
->list
, &res
->granted
);
224 if (actions
& DLM_UNLOCK_CLEAR_CONVERT_TYPE
) {
225 mlog(0, "clearing convert_type at %smaster node\n",
226 master_node
? "" : "non-");
227 lock
->ml
.convert_type
= LKM_IVMODE
;
230 /* remove the extra ref on lock */
234 res
->state
&= ~DLM_LOCK_RES_IN_PROGRESS
;
235 if (!dlm_lock_on_list(&res
->converting
, lock
))
236 BUG_ON(lock
->ml
.convert_type
!= LKM_IVMODE
);
238 BUG_ON(lock
->ml
.convert_type
== LKM_IVMODE
);
239 spin_unlock(&lock
->spinlock
);
240 spin_unlock(&res
->spinlock
);
243 /* let the caller's final dlm_lock_put handle the actual kfree */
244 if (actions
& DLM_UNLOCK_FREE_LOCK
) {
245 /* this should always be coupled with list removal */
246 BUG_ON(!(actions
& DLM_UNLOCK_REMOVE_LOCK
));
247 mlog(0, "lock %u:%llu should be gone now! refs=%d\n",
248 dlm_get_lock_cookie_node(lock
->ml
.cookie
),
249 dlm_get_lock_cookie_seq(lock
->ml
.cookie
),
250 atomic_read(&lock
->lock_refs
.refcount
)-1);
253 if (actions
& DLM_UNLOCK_CALL_AST
)
256 /* if cancel or unlock succeeded, lvb work is done */
257 if (status
== DLM_NORMAL
)
258 lksb
->flags
&= ~(DLM_LKSB_PUT_LVB
|DLM_LKSB_GET_LVB
);
263 void dlm_commit_pending_unlock(struct dlm_lock_resource
*res
,
264 struct dlm_lock
*lock
)
266 /* leave DLM_LKSB_PUT_LVB on the lksb so any final
267 * update of the lvb will be sent to the new master */
268 list_del_init(&lock
->list
);
271 void dlm_commit_pending_cancel(struct dlm_lock_resource
*res
,
272 struct dlm_lock
*lock
)
274 list_move_tail(&lock
->list
, &res
->granted
);
275 lock
->ml
.convert_type
= LKM_IVMODE
;
279 static inline enum dlm_status
dlmunlock_master(struct dlm_ctxt
*dlm
,
280 struct dlm_lock_resource
*res
,
281 struct dlm_lock
*lock
,
282 struct dlm_lockstatus
*lksb
,
286 return dlmunlock_common(dlm
, res
, lock
, lksb
, flags
, call_ast
, 1);
289 static inline enum dlm_status
dlmunlock_remote(struct dlm_ctxt
*dlm
,
290 struct dlm_lock_resource
*res
,
291 struct dlm_lock
*lock
,
292 struct dlm_lockstatus
*lksb
,
293 int flags
, int *call_ast
)
295 return dlmunlock_common(dlm
, res
, lock
, lksb
, flags
, call_ast
, 0);
303 * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
305 static enum dlm_status
dlm_send_remote_unlock_request(struct dlm_ctxt
*dlm
,
306 struct dlm_lock_resource
*res
,
307 struct dlm_lock
*lock
,
308 struct dlm_lockstatus
*lksb
,
312 struct dlm_unlock_lock unlock
;
319 mlog_entry("%.*s\n", res
->lockname
.len
, res
->lockname
.name
);
321 memset(&unlock
, 0, sizeof(unlock
));
322 unlock
.node_idx
= dlm
->node_num
;
323 unlock
.flags
= cpu_to_be32(flags
);
324 unlock
.cookie
= lock
->ml
.cookie
;
325 unlock
.namelen
= res
->lockname
.len
;
326 memcpy(unlock
.name
, res
->lockname
.name
, unlock
.namelen
);
328 vec
[0].iov_len
= sizeof(struct dlm_unlock_lock
);
329 vec
[0].iov_base
= &unlock
;
331 if (flags
& LKM_PUT_LVB
) {
332 /* extra data to send if we are updating lvb */
333 vec
[1].iov_len
= DLM_LVB_LEN
;
334 vec
[1].iov_base
= lock
->lksb
->lvb
;
338 tmpret
= o2net_send_message_vec(DLM_UNLOCK_LOCK_MSG
, dlm
->key
,
339 vec
, veclen
, owner
, &status
);
341 // successfully sent and received
342 if (status
== DLM_CANCELGRANT
)
344 else if (status
== DLM_FORWARD
) {
345 mlog(0, "master was in-progress. retry\n");
349 lksb
->status
= status
;
352 if (dlm_is_host_down(tmpret
)) {
353 /* NOTE: this seems strange, but it is what we want.
354 * when the master goes down during a cancel or
355 * unlock, the recovery code completes the operation
356 * as if the master had not died, then passes the
357 * updated state to the recovery master. this thread
358 * just needs to finish out the operation and call
362 /* something bad. this will BUG in ocfs2 */
363 ret
= dlm_err_to_dlm_status(tmpret
);
374 * taken: takes and drops res->spinlock
376 * returns: DLM_NORMAL, DLM_BADARGS, DLM_IVLOCKID,
377 * return value from dlmunlock_master
379 int dlm_unlock_lock_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
381 struct dlm_ctxt
*dlm
= data
;
382 struct dlm_unlock_lock
*unlock
= (struct dlm_unlock_lock
*)msg
->buf
;
383 struct dlm_lock_resource
*res
= NULL
;
384 struct list_head
*iter
;
385 struct dlm_lock
*lock
= NULL
;
386 enum dlm_status status
= DLM_NORMAL
;
388 struct dlm_lockstatus
*lksb
= NULL
;
391 struct list_head
*queue
;
393 flags
= be32_to_cpu(unlock
->flags
);
395 if (flags
& LKM_GET_LVB
) {
396 mlog(ML_ERROR
, "bad args! GET_LVB specified on unlock!\n");
400 if ((flags
& (LKM_PUT_LVB
|LKM_CANCEL
)) == (LKM_PUT_LVB
|LKM_CANCEL
)) {
401 mlog(ML_ERROR
, "bad args! cannot modify lvb on a CANCEL "
406 if (unlock
->namelen
> DLM_LOCKID_NAME_MAX
) {
407 mlog(ML_ERROR
, "Invalid name length in unlock handler!\n");
414 mlog_bug_on_msg(!dlm_domain_fully_joined(dlm
),
415 "Domain %s not fully joined!\n", dlm
->name
);
417 mlog(0, "lvb: %s\n", flags
& LKM_PUT_LVB
? "put lvb" : "none");
419 res
= dlm_lookup_lockres(dlm
, unlock
->name
, unlock
->namelen
);
421 /* We assume here that a no lock resource simply means
422 * it was migrated away and destroyed before the other
423 * node could detect it. */
424 mlog(0, "returning DLM_FORWARD -- res no longer exists\n");
425 status
= DLM_FORWARD
;
431 spin_lock(&res
->spinlock
);
432 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
433 spin_unlock(&res
->spinlock
);
434 mlog(0, "returning DLM_RECOVERING\n");
435 status
= DLM_RECOVERING
;
439 if (res
->state
& DLM_LOCK_RES_MIGRATING
) {
440 spin_unlock(&res
->spinlock
);
441 mlog(0, "returning DLM_MIGRATING\n");
442 status
= DLM_MIGRATING
;
446 if (res
->owner
!= dlm
->node_num
) {
447 spin_unlock(&res
->spinlock
);
448 mlog(0, "returning DLM_FORWARD -- not master\n");
449 status
= DLM_FORWARD
;
453 for (i
=0; i
<3; i
++) {
454 list_for_each(iter
, queue
) {
455 lock
= list_entry(iter
, struct dlm_lock
, list
);
456 if (lock
->ml
.cookie
== unlock
->cookie
&&
457 lock
->ml
.node
== unlock
->node_idx
) {
465 /* scan granted -> converting -> blocked queues */
468 spin_unlock(&res
->spinlock
);
470 status
= DLM_IVLOCKID
;
474 /* lock was found on queue */
476 /* unlockast only called on originating node */
477 if (flags
& LKM_PUT_LVB
) {
478 lksb
->flags
|= DLM_LKSB_PUT_LVB
;
479 memcpy(&lksb
->lvb
[0], &unlock
->lvb
[0], DLM_LVB_LEN
);
482 /* if this is in-progress, propagate the DLM_FORWARD
483 * all the way back out */
484 status
= dlmunlock_master(dlm
, res
, lock
, lksb
, flags
, &ignore
);
485 if (status
== DLM_FORWARD
)
486 mlog(0, "lockres is in progress\n");
488 if (flags
& LKM_PUT_LVB
)
489 lksb
->flags
&= ~DLM_LKSB_PUT_LVB
;
491 dlm_lockres_calc_usage(dlm
, res
);
492 dlm_kick_thread(dlm
, res
);
496 mlog(ML_ERROR
, "failed to find lock to unlock! "
498 dlm_get_lock_cookie_node(unlock
->cookie
),
499 dlm_get_lock_cookie_seq(unlock
->cookie
));
501 /* send the lksb->status back to the other node */
502 status
= lksb
->status
;
508 dlm_lockres_put(res
);
516 static enum dlm_status
dlm_get_cancel_actions(struct dlm_ctxt
*dlm
,
517 struct dlm_lock_resource
*res
,
518 struct dlm_lock
*lock
,
519 struct dlm_lockstatus
*lksb
,
522 enum dlm_status status
;
524 if (dlm_lock_on_list(&res
->blocked
, lock
)) {
525 /* cancel this outright */
526 lksb
->status
= DLM_NORMAL
;
528 *actions
= (DLM_UNLOCK_CALL_AST
|
529 DLM_UNLOCK_REMOVE_LOCK
);
530 } else if (dlm_lock_on_list(&res
->converting
, lock
)) {
531 /* cancel the request, put back on granted */
532 lksb
->status
= DLM_NORMAL
;
534 *actions
= (DLM_UNLOCK_CALL_AST
|
535 DLM_UNLOCK_REMOVE_LOCK
|
536 DLM_UNLOCK_REGRANT_LOCK
|
537 DLM_UNLOCK_CLEAR_CONVERT_TYPE
);
538 } else if (dlm_lock_on_list(&res
->granted
, lock
)) {
539 /* too late, already granted. DLM_CANCELGRANT */
540 lksb
->status
= DLM_CANCELGRANT
;
542 *actions
= DLM_UNLOCK_CALL_AST
;
544 mlog(ML_ERROR
, "lock to cancel is not on any list!\n");
545 lksb
->status
= DLM_IVLOCKID
;
546 status
= DLM_IVLOCKID
;
552 static enum dlm_status
dlm_get_unlock_actions(struct dlm_ctxt
*dlm
,
553 struct dlm_lock_resource
*res
,
554 struct dlm_lock
*lock
,
555 struct dlm_lockstatus
*lksb
,
558 enum dlm_status status
;
561 if (!dlm_lock_on_list(&res
->granted
, lock
)) {
562 lksb
->status
= DLM_DENIED
;
567 /* unlock granted lock */
568 lksb
->status
= DLM_NORMAL
;
570 *actions
= (DLM_UNLOCK_FREE_LOCK
|
571 DLM_UNLOCK_CALL_AST
|
572 DLM_UNLOCK_REMOVE_LOCK
);
577 /* there seems to be no point in doing this async
578 * since (even for the remote case) there is really
579 * no work to queue up... so just do it and fire the
580 * unlockast by hand when done... */
581 enum dlm_status
dlmunlock(struct dlm_ctxt
*dlm
, struct dlm_lockstatus
*lksb
,
582 int flags
, dlm_astunlockfunc_t
*unlockast
, void *data
)
584 enum dlm_status status
;
585 struct dlm_lock_resource
*res
;
586 struct dlm_lock
*lock
= NULL
;
587 int call_ast
, is_master
;
592 dlm_error(DLM_BADARGS
);
596 if (flags
& ~(LKM_CANCEL
| LKM_VALBLK
| LKM_INVVALBLK
)) {
597 dlm_error(DLM_BADPARAM
);
601 if ((flags
& (LKM_VALBLK
| LKM_CANCEL
)) == (LKM_VALBLK
| LKM_CANCEL
)) {
602 mlog(0, "VALBLK given with CANCEL: ignoring VALBLK\n");
603 flags
&= ~LKM_VALBLK
;
606 if (!lksb
->lockid
|| !lksb
->lockid
->lockres
) {
607 dlm_error(DLM_BADPARAM
);
617 dlm_lockres_get(res
);
620 /* need to retry up here because owner may have changed */
621 mlog(0, "lock=%p res=%p\n", lock
, res
);
623 spin_lock(&res
->spinlock
);
624 is_master
= (res
->owner
== dlm
->node_num
);
625 spin_unlock(&res
->spinlock
);
628 status
= dlmunlock_master(dlm
, res
, lock
, lksb
, flags
,
630 mlog(0, "done calling dlmunlock_master: returned %d, "
631 "call_ast is %d\n", status
, call_ast
);
633 status
= dlmunlock_remote(dlm
, res
, lock
, lksb
, flags
,
635 mlog(0, "done calling dlmunlock_remote: returned %d, "
636 "call_ast is %d\n", status
, call_ast
);
639 if (status
== DLM_RECOVERING
||
640 status
== DLM_MIGRATING
||
641 status
== DLM_FORWARD
) {
642 /* We want to go away for a tiny bit to allow recovery
643 * / migration to complete on this resource. I don't
644 * know of any wait queue we could sleep on as this
645 * may be happening on another node. Perhaps the
646 * proper solution is to queue up requests on the
649 /* do we want to yield(); ?? */
652 mlog(0, "retrying unlock due to pending recovery/"
653 "migration/in-progress\n");
658 mlog(0, "calling unlockast(%p, %d)\n", data
, lksb
->status
);
660 /* it is possible that there is one last bast
661 * pending. make sure it is flushed, then
662 * call the unlockast.
663 * not an issue if this is a mastered remotely,
664 * since this lock has been removed from the
665 * lockres queues and cannot be found. */
666 dlm_kick_thread(dlm
, NULL
);
667 wait_event(dlm
->ast_wq
,
668 dlm_lock_basts_flushed(dlm
, lock
));
670 (*unlockast
)(data
, lksb
->status
);
673 if (status
== DLM_NORMAL
) {
674 mlog(0, "kicking the thread\n");
675 dlm_kick_thread(dlm
, res
);
679 dlm_lockres_calc_usage(dlm
, res
);
680 dlm_lockres_put(res
);
683 mlog(0, "returning status=%d!\n", status
);
686 EXPORT_SYMBOL_GPL(dlmunlock
);