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/highmem.h>
32 #include <linux/init.h>
33 #include <linux/sysctl.h>
34 #include <linux/random.h>
35 #include <linux/blkdev.h>
36 #include <linux/socket.h>
37 #include <linux/inet.h>
38 #include <linux/spinlock.h>
39 #include <linux/delay.h>
41 #include "cluster/heartbeat.h"
42 #include "cluster/nodemanager.h"
43 #include "cluster/tcp.h"
46 #include "dlmcommon.h"
48 #define MLOG_MASK_PREFIX ML_DLM
49 #include "cluster/masklog.h"
51 #define DLM_UNLOCK_FREE_LOCK 0x00000001
52 #define DLM_UNLOCK_CALL_AST 0x00000002
53 #define DLM_UNLOCK_REMOVE_LOCK 0x00000004
54 #define DLM_UNLOCK_REGRANT_LOCK 0x00000008
55 #define DLM_UNLOCK_CLEAR_CONVERT_TYPE 0x00000010
58 static enum dlm_status
dlm_get_cancel_actions(struct dlm_ctxt
*dlm
,
59 struct dlm_lock_resource
*res
,
60 struct dlm_lock
*lock
,
61 struct dlm_lockstatus
*lksb
,
63 static enum dlm_status
dlm_get_unlock_actions(struct dlm_ctxt
*dlm
,
64 struct dlm_lock_resource
*res
,
65 struct dlm_lock
*lock
,
66 struct dlm_lockstatus
*lksb
,
69 static enum dlm_status
dlm_send_remote_unlock_request(struct dlm_ctxt
*dlm
,
70 struct dlm_lock_resource
*res
,
71 struct dlm_lock
*lock
,
72 struct dlm_lockstatus
*lksb
,
78 * according to the spec:
79 * http://opendlm.sourceforge.net/cvsmirror/opendlm/docs/dlmbook_final.pdf
81 * flags & LKM_CANCEL != 0: must be converting or blocked
82 * flags & LKM_CANCEL == 0: must be granted
84 * So to unlock a converting lock, you must first cancel the
85 * convert (passing LKM_CANCEL in flags), then call the unlock
86 * again (with no LKM_CANCEL in flags).
93 * taken: res->spinlock and lock->spinlock taken and dropped
95 * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
96 * all callers should have taken an extra ref on lock coming in
98 static enum dlm_status
dlmunlock_common(struct dlm_ctxt
*dlm
,
99 struct dlm_lock_resource
*res
,
100 struct dlm_lock
*lock
,
101 struct dlm_lockstatus
*lksb
,
102 int flags
, int *call_ast
,
105 enum dlm_status status
;
110 mlog(0, "master_node = %d, valblk = %d\n", master_node
,
114 BUG_ON(res
->owner
!= dlm
->node_num
);
116 BUG_ON(res
->owner
== dlm
->node_num
);
118 spin_lock(&dlm
->ast_lock
);
119 /* We want to be sure that we're not freeing a lock
120 * that still has AST's pending... */
121 in_use
= !list_empty(&lock
->ast_list
);
122 spin_unlock(&dlm
->ast_lock
);
123 if (in_use
&& !(flags
& LKM_CANCEL
)) {
124 mlog(ML_ERROR
, "lockres %.*s: Someone is calling dlmunlock "
125 "while waiting for an ast!", res
->lockname
.len
,
130 spin_lock(&res
->spinlock
);
131 if (res
->state
& DLM_LOCK_RES_IN_PROGRESS
) {
132 if (master_node
&& !(flags
& LKM_CANCEL
)) {
133 mlog(ML_ERROR
, "lockres in progress!\n");
134 spin_unlock(&res
->spinlock
);
137 /* ok for this to sleep if not in a network handler */
138 __dlm_wait_on_lockres(res
);
139 res
->state
|= DLM_LOCK_RES_IN_PROGRESS
;
141 spin_lock(&lock
->spinlock
);
143 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
144 status
= DLM_RECOVERING
;
148 if (res
->state
& DLM_LOCK_RES_MIGRATING
) {
149 status
= DLM_MIGRATING
;
153 /* see above for what the spec says about
154 * LKM_CANCEL and the lock queue state */
155 if (flags
& LKM_CANCEL
)
156 status
= dlm_get_cancel_actions(dlm
, res
, lock
, lksb
, &actions
);
158 status
= dlm_get_unlock_actions(dlm
, res
, lock
, lksb
, &actions
);
160 if (status
!= DLM_NORMAL
&& (status
!= DLM_CANCELGRANT
|| !master_node
))
163 /* By now this has been masked out of cancel requests. */
164 if (flags
& LKM_VALBLK
) {
165 /* make the final update to the lvb */
167 memcpy(res
->lvb
, lksb
->lvb
, DLM_LVB_LEN
);
169 flags
|= LKM_PUT_LVB
; /* let the send function
175 /* drop locks and send message */
176 if (flags
& LKM_CANCEL
)
177 lock
->cancel_pending
= 1;
179 lock
->unlock_pending
= 1;
180 spin_unlock(&lock
->spinlock
);
181 spin_unlock(&res
->spinlock
);
182 status
= dlm_send_remote_unlock_request(dlm
, res
, lock
, lksb
,
184 spin_lock(&res
->spinlock
);
185 spin_lock(&lock
->spinlock
);
186 /* if the master told us the lock was already granted,
187 * let the ast handle all of these actions */
188 if (status
== DLM_CANCELGRANT
) {
189 actions
&= ~(DLM_UNLOCK_REMOVE_LOCK
|
190 DLM_UNLOCK_REGRANT_LOCK
|
191 DLM_UNLOCK_CLEAR_CONVERT_TYPE
);
192 } else if (status
== DLM_RECOVERING
||
193 status
== DLM_MIGRATING
||
194 status
== DLM_FORWARD
) {
195 /* must clear the actions because this unlock
196 * is about to be retried. cannot free or do
197 * any list manipulation. */
198 mlog(0, "%s:%.*s: clearing actions, %s\n",
199 dlm
->name
, res
->lockname
.len
,
201 status
==DLM_RECOVERING
?"recovering":
202 (status
==DLM_MIGRATING
?"migrating":
206 if (flags
& LKM_CANCEL
)
207 lock
->cancel_pending
= 0;
209 lock
->unlock_pending
= 0;
213 /* get an extra ref on lock. if we are just switching
214 * lists here, we dont want the lock to go away. */
217 if (actions
& DLM_UNLOCK_REMOVE_LOCK
) {
218 list_del_init(&lock
->list
);
221 if (actions
& DLM_UNLOCK_REGRANT_LOCK
) {
223 list_add_tail(&lock
->list
, &res
->granted
);
225 if (actions
& DLM_UNLOCK_CLEAR_CONVERT_TYPE
) {
226 mlog(0, "clearing convert_type at %smaster node\n",
227 master_node
? "" : "non-");
228 lock
->ml
.convert_type
= LKM_IVMODE
;
231 /* remove the extra ref on lock */
235 res
->state
&= ~DLM_LOCK_RES_IN_PROGRESS
;
236 if (!dlm_lock_on_list(&res
->converting
, lock
))
237 BUG_ON(lock
->ml
.convert_type
!= LKM_IVMODE
);
239 BUG_ON(lock
->ml
.convert_type
== LKM_IVMODE
);
240 spin_unlock(&lock
->spinlock
);
241 spin_unlock(&res
->spinlock
);
244 /* let the caller's final dlm_lock_put handle the actual kfree */
245 if (actions
& DLM_UNLOCK_FREE_LOCK
) {
246 /* this should always be coupled with list removal */
247 BUG_ON(!(actions
& DLM_UNLOCK_REMOVE_LOCK
));
248 mlog(0, "lock %u:%llu should be gone now! refs=%d\n",
249 dlm_get_lock_cookie_node(be64_to_cpu(lock
->ml
.cookie
)),
250 dlm_get_lock_cookie_seq(be64_to_cpu(lock
->ml
.cookie
)),
251 atomic_read(&lock
->lock_refs
.refcount
)-1);
254 if (actions
& DLM_UNLOCK_CALL_AST
)
257 /* if cancel or unlock succeeded, lvb work is done */
258 if (status
== DLM_NORMAL
)
259 lksb
->flags
&= ~(DLM_LKSB_PUT_LVB
|DLM_LKSB_GET_LVB
);
264 void dlm_commit_pending_unlock(struct dlm_lock_resource
*res
,
265 struct dlm_lock
*lock
)
267 /* leave DLM_LKSB_PUT_LVB on the lksb so any final
268 * update of the lvb will be sent to the new master */
269 list_del_init(&lock
->list
);
272 void dlm_commit_pending_cancel(struct dlm_lock_resource
*res
,
273 struct dlm_lock
*lock
)
275 list_move_tail(&lock
->list
, &res
->granted
);
276 lock
->ml
.convert_type
= LKM_IVMODE
;
280 static inline enum dlm_status
dlmunlock_master(struct dlm_ctxt
*dlm
,
281 struct dlm_lock_resource
*res
,
282 struct dlm_lock
*lock
,
283 struct dlm_lockstatus
*lksb
,
287 return dlmunlock_common(dlm
, res
, lock
, lksb
, flags
, call_ast
, 1);
290 static inline enum dlm_status
dlmunlock_remote(struct dlm_ctxt
*dlm
,
291 struct dlm_lock_resource
*res
,
292 struct dlm_lock
*lock
,
293 struct dlm_lockstatus
*lksb
,
294 int flags
, int *call_ast
)
296 return dlmunlock_common(dlm
, res
, lock
, lksb
, flags
, call_ast
, 0);
304 * returns: DLM_NORMAL, DLM_NOLOCKMGR, status from network
306 static enum dlm_status
dlm_send_remote_unlock_request(struct dlm_ctxt
*dlm
,
307 struct dlm_lock_resource
*res
,
308 struct dlm_lock
*lock
,
309 struct dlm_lockstatus
*lksb
,
313 struct dlm_unlock_lock unlock
;
320 mlog(0, "%.*s\n", res
->lockname
.len
, res
->lockname
.name
);
322 if (owner
== dlm
->node_num
) {
323 /* ended up trying to contact ourself. this means
324 * that the lockres had been remote but became local
325 * via a migration. just retry it, now as local */
326 mlog(0, "%s:%.*s: this node became the master due to a "
327 "migration, re-evaluate now\n", dlm
->name
,
328 res
->lockname
.len
, res
->lockname
.name
);
332 memset(&unlock
, 0, sizeof(unlock
));
333 unlock
.node_idx
= dlm
->node_num
;
334 unlock
.flags
= cpu_to_be32(flags
);
335 unlock
.cookie
= lock
->ml
.cookie
;
336 unlock
.namelen
= res
->lockname
.len
;
337 memcpy(unlock
.name
, res
->lockname
.name
, unlock
.namelen
);
339 vec
[0].iov_len
= sizeof(struct dlm_unlock_lock
);
340 vec
[0].iov_base
= &unlock
;
342 if (flags
& LKM_PUT_LVB
) {
343 /* extra data to send if we are updating lvb */
344 vec
[1].iov_len
= DLM_LVB_LEN
;
345 vec
[1].iov_base
= lock
->lksb
->lvb
;
349 tmpret
= o2net_send_message_vec(DLM_UNLOCK_LOCK_MSG
, dlm
->key
,
350 vec
, veclen
, owner
, &status
);
352 // successfully sent and received
353 if (status
== DLM_FORWARD
)
354 mlog(0, "master was in-progress. retry\n");
357 mlog(ML_ERROR
, "Error %d when sending message %u (key 0x%x) to "
358 "node %u\n", tmpret
, DLM_UNLOCK_LOCK_MSG
, dlm
->key
, owner
);
359 if (dlm_is_host_down(tmpret
)) {
360 /* NOTE: this seems strange, but it is what we want.
361 * when the master goes down during a cancel or
362 * unlock, the recovery code completes the operation
363 * as if the master had not died, then passes the
364 * updated state to the recovery master. this thread
365 * just needs to finish out the operation and call
369 /* something bad. this will BUG in ocfs2 */
370 ret
= dlm_err_to_dlm_status(tmpret
);
380 * taken: takes and drops res->spinlock
382 * returns: DLM_NORMAL, DLM_BADARGS, DLM_IVLOCKID,
383 * return value from dlmunlock_master
385 int dlm_unlock_lock_handler(struct o2net_msg
*msg
, u32 len
, void *data
,
388 struct dlm_ctxt
*dlm
= data
;
389 struct dlm_unlock_lock
*unlock
= (struct dlm_unlock_lock
*)msg
->buf
;
390 struct dlm_lock_resource
*res
= NULL
;
391 struct dlm_lock
*lock
= NULL
;
392 enum dlm_status status
= DLM_NORMAL
;
394 struct dlm_lockstatus
*lksb
= NULL
;
397 struct list_head
*queue
;
399 flags
= be32_to_cpu(unlock
->flags
);
401 if (flags
& LKM_GET_LVB
) {
402 mlog(ML_ERROR
, "bad args! GET_LVB specified on unlock!\n");
406 if ((flags
& (LKM_PUT_LVB
|LKM_CANCEL
)) == (LKM_PUT_LVB
|LKM_CANCEL
)) {
407 mlog(ML_ERROR
, "bad args! cannot modify lvb on a CANCEL "
412 if (unlock
->namelen
> DLM_LOCKID_NAME_MAX
) {
413 mlog(ML_ERROR
, "Invalid name length in unlock handler!\n");
420 mlog_bug_on_msg(!dlm_domain_fully_joined(dlm
),
421 "Domain %s not fully joined!\n", dlm
->name
);
423 mlog(0, "lvb: %s\n", flags
& LKM_PUT_LVB
? "put lvb" : "none");
425 res
= dlm_lookup_lockres(dlm
, unlock
->name
, unlock
->namelen
);
427 /* We assume here that a no lock resource simply means
428 * it was migrated away and destroyed before the other
429 * node could detect it. */
430 mlog(0, "returning DLM_FORWARD -- res no longer exists\n");
431 status
= DLM_FORWARD
;
437 spin_lock(&res
->spinlock
);
438 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
439 spin_unlock(&res
->spinlock
);
440 mlog(0, "returning DLM_RECOVERING\n");
441 status
= DLM_RECOVERING
;
445 if (res
->state
& DLM_LOCK_RES_MIGRATING
) {
446 spin_unlock(&res
->spinlock
);
447 mlog(0, "returning DLM_MIGRATING\n");
448 status
= DLM_MIGRATING
;
452 if (res
->owner
!= dlm
->node_num
) {
453 spin_unlock(&res
->spinlock
);
454 mlog(0, "returning DLM_FORWARD -- not master\n");
455 status
= DLM_FORWARD
;
459 for (i
=0; i
<3; i
++) {
460 list_for_each_entry(lock
, queue
, list
) {
461 if (lock
->ml
.cookie
== unlock
->cookie
&&
462 lock
->ml
.node
== unlock
->node_idx
) {
470 /* scan granted -> converting -> blocked queues */
473 spin_unlock(&res
->spinlock
);
475 status
= DLM_IVLOCKID
;
479 /* lock was found on queue */
481 if (flags
& (LKM_VALBLK
|LKM_PUT_LVB
) &&
482 lock
->ml
.type
!= LKM_EXMODE
)
483 flags
&= ~(LKM_VALBLK
|LKM_PUT_LVB
);
485 /* unlockast only called on originating node */
486 if (flags
& LKM_PUT_LVB
) {
487 lksb
->flags
|= DLM_LKSB_PUT_LVB
;
488 memcpy(&lksb
->lvb
[0], &unlock
->lvb
[0], DLM_LVB_LEN
);
491 /* if this is in-progress, propagate the DLM_FORWARD
492 * all the way back out */
493 status
= dlmunlock_master(dlm
, res
, lock
, lksb
, flags
, &ignore
);
494 if (status
== DLM_FORWARD
)
495 mlog(0, "lockres is in progress\n");
497 if (flags
& LKM_PUT_LVB
)
498 lksb
->flags
&= ~DLM_LKSB_PUT_LVB
;
500 dlm_lockres_calc_usage(dlm
, res
);
501 dlm_kick_thread(dlm
, res
);
505 mlog(ML_ERROR
, "failed to find lock to unlock! "
507 dlm_get_lock_cookie_node(be64_to_cpu(unlock
->cookie
)),
508 dlm_get_lock_cookie_seq(be64_to_cpu(unlock
->cookie
)));
514 dlm_lockres_put(res
);
522 static enum dlm_status
dlm_get_cancel_actions(struct dlm_ctxt
*dlm
,
523 struct dlm_lock_resource
*res
,
524 struct dlm_lock
*lock
,
525 struct dlm_lockstatus
*lksb
,
528 enum dlm_status status
;
530 if (dlm_lock_on_list(&res
->blocked
, lock
)) {
531 /* cancel this outright */
533 *actions
= (DLM_UNLOCK_CALL_AST
|
534 DLM_UNLOCK_REMOVE_LOCK
);
535 } else if (dlm_lock_on_list(&res
->converting
, lock
)) {
536 /* cancel the request, put back on granted */
538 *actions
= (DLM_UNLOCK_CALL_AST
|
539 DLM_UNLOCK_REMOVE_LOCK
|
540 DLM_UNLOCK_REGRANT_LOCK
|
541 DLM_UNLOCK_CLEAR_CONVERT_TYPE
);
542 } else if (dlm_lock_on_list(&res
->granted
, lock
)) {
543 /* too late, already granted. */
544 status
= DLM_CANCELGRANT
;
545 *actions
= DLM_UNLOCK_CALL_AST
;
547 mlog(ML_ERROR
, "lock to cancel is not on any list!\n");
548 status
= DLM_IVLOCKID
;
554 static enum dlm_status
dlm_get_unlock_actions(struct dlm_ctxt
*dlm
,
555 struct dlm_lock_resource
*res
,
556 struct dlm_lock
*lock
,
557 struct dlm_lockstatus
*lksb
,
560 enum dlm_status status
;
563 if (!dlm_lock_on_list(&res
->granted
, lock
)) {
568 /* unlock granted lock */
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
;
590 dlm_error(DLM_BADARGS
);
594 if (flags
& ~(LKM_CANCEL
| LKM_VALBLK
| LKM_INVVALBLK
)) {
595 dlm_error(DLM_BADPARAM
);
599 if ((flags
& (LKM_VALBLK
| LKM_CANCEL
)) == (LKM_VALBLK
| LKM_CANCEL
)) {
600 mlog(0, "VALBLK given with CANCEL: ignoring VALBLK\n");
601 flags
&= ~LKM_VALBLK
;
604 if (!lksb
->lockid
|| !lksb
->lockid
->lockres
) {
605 dlm_error(DLM_BADPARAM
);
615 dlm_lockres_get(res
);
618 /* need to retry up here because owner may have changed */
619 mlog(0, "lock=%p res=%p\n", lock
, res
);
621 spin_lock(&res
->spinlock
);
622 is_master
= (res
->owner
== dlm
->node_num
);
623 if (flags
& LKM_VALBLK
&& lock
->ml
.type
!= LKM_EXMODE
)
624 flags
&= ~LKM_VALBLK
;
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
, 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
, status
);
673 if (status
== DLM_CANCELGRANT
)
676 if (status
== DLM_NORMAL
) {
677 mlog(0, "kicking the thread\n");
678 dlm_kick_thread(dlm
, res
);
682 dlm_lockres_calc_usage(dlm
, res
);
683 dlm_lockres_put(res
);
686 mlog(0, "returning status=%d!\n", status
);
689 EXPORT_SYMBOL_GPL(dlmunlock
);