1 /* -*- mode: c; c-basic-offset: 8; -*-
2 * vim: noexpandtab sw=8 ts=8 sts=0:
6 * standalone DLM module
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/timer.h>
41 #include <linux/kthread.h>
42 #include <linux/delay.h>
45 #include "cluster/heartbeat.h"
46 #include "cluster/nodemanager.h"
47 #include "cluster/tcp.h"
50 #include "dlmcommon.h"
51 #include "dlmdomain.h"
53 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_THREAD)
54 #include "cluster/masklog.h"
56 static int dlm_thread(void *data
);
57 static void dlm_flush_asts(struct dlm_ctxt
*dlm
);
59 #define dlm_lock_is_remote(dlm, lock) ((lock)->ml.node != (dlm)->node_num)
61 /* will exit holding res->spinlock, but may drop in function */
62 /* waits until flags are cleared on res->state */
63 void __dlm_wait_on_lockres_flags(struct dlm_lock_resource
*res
, int flags
)
65 DECLARE_WAITQUEUE(wait
, current
);
67 assert_spin_locked(&res
->spinlock
);
69 add_wait_queue(&res
->wq
, &wait
);
71 set_current_state(TASK_UNINTERRUPTIBLE
);
72 if (res
->state
& flags
) {
73 spin_unlock(&res
->spinlock
);
75 spin_lock(&res
->spinlock
);
78 remove_wait_queue(&res
->wq
, &wait
);
79 __set_current_state(TASK_RUNNING
);
82 int __dlm_lockres_has_locks(struct dlm_lock_resource
*res
)
84 if (list_empty(&res
->granted
) &&
85 list_empty(&res
->converting
) &&
86 list_empty(&res
->blocked
))
91 /* "unused": the lockres has no locks, is not on the dirty list,
92 * has no inflight locks (in the gap between mastery and acquiring
93 * the first lock), and has no bits in its refmap.
94 * truly ready to be freed. */
95 int __dlm_lockres_unused(struct dlm_lock_resource
*res
)
97 if (!__dlm_lockres_has_locks(res
) &&
98 (list_empty(&res
->dirty
) && !(res
->state
& DLM_LOCK_RES_DIRTY
))) {
99 /* try not to scan the bitmap unless the first two
100 * conditions are already true */
101 int bit
= find_next_bit(res
->refmap
, O2NM_MAX_NODES
, 0);
102 if (bit
>= O2NM_MAX_NODES
) {
103 /* since the bit for dlm->node_num is not
104 * set, inflight_locks better be zero */
105 BUG_ON(res
->inflight_locks
!= 0);
113 /* Call whenever you may have added or deleted something from one of
114 * the lockres queue's. This will figure out whether it belongs on the
115 * unused list or not and does the appropriate thing. */
116 void __dlm_lockres_calc_usage(struct dlm_ctxt
*dlm
,
117 struct dlm_lock_resource
*res
)
119 mlog_entry("%.*s\n", res
->lockname
.len
, res
->lockname
.name
);
121 assert_spin_locked(&dlm
->spinlock
);
122 assert_spin_locked(&res
->spinlock
);
124 if (__dlm_lockres_unused(res
)){
125 if (list_empty(&res
->purge
)) {
126 mlog(0, "putting lockres %.*s:%p onto purge list\n",
127 res
->lockname
.len
, res
->lockname
.name
, res
);
129 res
->last_used
= jiffies
;
130 dlm_lockres_get(res
);
131 list_add_tail(&res
->purge
, &dlm
->purge_list
);
134 } else if (!list_empty(&res
->purge
)) {
135 mlog(0, "removing lockres %.*s:%p from purge list, owner=%u\n",
136 res
->lockname
.len
, res
->lockname
.name
, res
, res
->owner
);
138 list_del_init(&res
->purge
);
139 dlm_lockres_put(res
);
144 void dlm_lockres_calc_usage(struct dlm_ctxt
*dlm
,
145 struct dlm_lock_resource
*res
)
147 mlog_entry("%.*s\n", res
->lockname
.len
, res
->lockname
.name
);
148 spin_lock(&dlm
->spinlock
);
149 spin_lock(&res
->spinlock
);
151 __dlm_lockres_calc_usage(dlm
, res
);
153 spin_unlock(&res
->spinlock
);
154 spin_unlock(&dlm
->spinlock
);
157 static int dlm_purge_lockres(struct dlm_ctxt
*dlm
,
158 struct dlm_lock_resource
*res
)
163 spin_lock(&res
->spinlock
);
164 if (!__dlm_lockres_unused(res
)) {
165 spin_unlock(&res
->spinlock
);
166 mlog(0, "%s:%.*s: tried to purge but not unused\n",
167 dlm
->name
, res
->lockname
.len
, res
->lockname
.name
);
170 master
= (res
->owner
== dlm
->node_num
);
172 res
->state
|= DLM_LOCK_RES_DROPPING_REF
;
173 spin_unlock(&res
->spinlock
);
175 mlog(0, "purging lockres %.*s, master = %d\n", res
->lockname
.len
,
176 res
->lockname
.name
, master
);
179 /* drop spinlock... retake below */
180 spin_unlock(&dlm
->spinlock
);
182 spin_lock(&res
->spinlock
);
183 /* This ensures that clear refmap is sent after the set */
184 __dlm_wait_on_lockres_flags(res
, (DLM_LOCK_RES_SETREF_INPROG
|
185 DLM_LOCK_RES_MIGRATING
));
186 spin_unlock(&res
->spinlock
);
188 /* clear our bit from the master's refmap, ignore errors */
189 ret
= dlm_drop_lockres_ref(dlm
, res
);
192 if (!dlm_is_host_down(ret
))
195 mlog(0, "%s:%.*s: dlm_deref_lockres returned %d\n",
196 dlm
->name
, res
->lockname
.len
, res
->lockname
.name
, ret
);
197 spin_lock(&dlm
->spinlock
);
200 if (!list_empty(&res
->purge
)) {
201 mlog(0, "removing lockres %.*s:%p from purgelist, "
202 "master = %d\n", res
->lockname
.len
, res
->lockname
.name
,
204 list_del_init(&res
->purge
);
205 dlm_lockres_put(res
);
208 __dlm_unhash_lockres(res
);
210 /* lockres is not in the hash now. drop the flag and wake up
211 * any processes waiting in dlm_get_lock_resource. */
213 spin_lock(&res
->spinlock
);
214 res
->state
&= ~DLM_LOCK_RES_DROPPING_REF
;
215 spin_unlock(&res
->spinlock
);
221 static void dlm_run_purge_list(struct dlm_ctxt
*dlm
,
224 unsigned int run_max
, unused
;
225 unsigned long purge_jiffies
;
226 struct dlm_lock_resource
*lockres
;
228 spin_lock(&dlm
->spinlock
);
229 run_max
= dlm
->purge_count
;
231 while(run_max
&& !list_empty(&dlm
->purge_list
)) {
234 lockres
= list_entry(dlm
->purge_list
.next
,
235 struct dlm_lock_resource
, purge
);
237 /* Status of the lockres *might* change so double
238 * check. If the lockres is unused, holding the dlm
239 * spinlock will prevent people from getting and more
240 * refs on it -- there's no need to keep the lockres
242 spin_lock(&lockres
->spinlock
);
243 unused
= __dlm_lockres_unused(lockres
);
244 spin_unlock(&lockres
->spinlock
);
249 purge_jiffies
= lockres
->last_used
+
250 msecs_to_jiffies(DLM_PURGE_INTERVAL_MS
);
252 /* Make sure that we want to be processing this guy at
254 if (!purge_now
&& time_after(purge_jiffies
, jiffies
)) {
255 /* Since resources are added to the purge list
256 * in tail order, we can stop at the first
257 * unpurgable resource -- anyone added after
258 * him will have a greater last_used value */
262 dlm_lockres_get(lockres
);
264 /* This may drop and reacquire the dlm spinlock if it
265 * has to do migration. */
266 if (dlm_purge_lockres(dlm
, lockres
))
269 dlm_lockres_put(lockres
);
271 /* Avoid adding any scheduling latencies */
272 cond_resched_lock(&dlm
->spinlock
);
275 spin_unlock(&dlm
->spinlock
);
278 static void dlm_shuffle_lists(struct dlm_ctxt
*dlm
,
279 struct dlm_lock_resource
*res
)
281 struct dlm_lock
*lock
, *target
;
282 struct list_head
*iter
;
283 struct list_head
*head
;
286 //mlog(0, "res->lockname.len=%d\n", res->lockname.len);
287 //mlog(0, "res->lockname.name=%p\n", res->lockname.name);
288 //mlog(0, "shuffle res %.*s\n", res->lockname.len,
289 // res->lockname.name);
291 /* because this function is called with the lockres
292 * spinlock, and because we know that it is not migrating/
293 * recovering/in-progress, it is fine to reserve asts and
294 * basts right before queueing them all throughout */
295 assert_spin_locked(&res
->spinlock
);
296 BUG_ON((res
->state
& (DLM_LOCK_RES_MIGRATING
|
297 DLM_LOCK_RES_RECOVERING
|
298 DLM_LOCK_RES_IN_PROGRESS
)));
301 if (list_empty(&res
->converting
))
303 mlog(0, "res %.*s has locks on a convert queue\n", res
->lockname
.len
,
306 target
= list_entry(res
->converting
.next
, struct dlm_lock
, list
);
307 if (target
->ml
.convert_type
== LKM_IVMODE
) {
308 mlog(ML_ERROR
, "%.*s: converting a lock with no "
309 "convert_type!\n", res
->lockname
.len
, res
->lockname
.name
);
312 head
= &res
->granted
;
313 list_for_each(iter
, head
) {
314 lock
= list_entry(iter
, struct dlm_lock
, list
);
317 if (!dlm_lock_compatible(lock
->ml
.type
,
318 target
->ml
.convert_type
)) {
320 /* queue the BAST if not already */
321 if (lock
->ml
.highest_blocked
== LKM_IVMODE
) {
322 __dlm_lockres_reserve_ast(res
);
323 dlm_queue_bast(dlm
, lock
);
325 /* update the highest_blocked if needed */
326 if (lock
->ml
.highest_blocked
< target
->ml
.convert_type
)
327 lock
->ml
.highest_blocked
=
328 target
->ml
.convert_type
;
331 head
= &res
->converting
;
332 list_for_each(iter
, head
) {
333 lock
= list_entry(iter
, struct dlm_lock
, list
);
336 if (!dlm_lock_compatible(lock
->ml
.type
,
337 target
->ml
.convert_type
)) {
339 if (lock
->ml
.highest_blocked
== LKM_IVMODE
) {
340 __dlm_lockres_reserve_ast(res
);
341 dlm_queue_bast(dlm
, lock
);
343 if (lock
->ml
.highest_blocked
< target
->ml
.convert_type
)
344 lock
->ml
.highest_blocked
=
345 target
->ml
.convert_type
;
349 /* we can convert the lock */
351 spin_lock(&target
->spinlock
);
352 BUG_ON(target
->ml
.highest_blocked
!= LKM_IVMODE
);
354 mlog(0, "calling ast for converting lock: %.*s, have: %d, "
355 "granting: %d, node: %u\n", res
->lockname
.len
,
356 res
->lockname
.name
, target
->ml
.type
,
357 target
->ml
.convert_type
, target
->ml
.node
);
359 target
->ml
.type
= target
->ml
.convert_type
;
360 target
->ml
.convert_type
= LKM_IVMODE
;
361 list_move_tail(&target
->list
, &res
->granted
);
363 BUG_ON(!target
->lksb
);
364 target
->lksb
->status
= DLM_NORMAL
;
366 spin_unlock(&target
->spinlock
);
368 __dlm_lockres_reserve_ast(res
);
369 dlm_queue_ast(dlm
, target
);
370 /* go back and check for more */
375 if (list_empty(&res
->blocked
))
377 target
= list_entry(res
->blocked
.next
, struct dlm_lock
, list
);
379 head
= &res
->granted
;
380 list_for_each(iter
, head
) {
381 lock
= list_entry(iter
, struct dlm_lock
, list
);
384 if (!dlm_lock_compatible(lock
->ml
.type
, target
->ml
.type
)) {
386 if (lock
->ml
.highest_blocked
== LKM_IVMODE
) {
387 __dlm_lockres_reserve_ast(res
);
388 dlm_queue_bast(dlm
, lock
);
390 if (lock
->ml
.highest_blocked
< target
->ml
.type
)
391 lock
->ml
.highest_blocked
= target
->ml
.type
;
395 head
= &res
->converting
;
396 list_for_each(iter
, head
) {
397 lock
= list_entry(iter
, struct dlm_lock
, list
);
400 if (!dlm_lock_compatible(lock
->ml
.type
, target
->ml
.type
)) {
402 if (lock
->ml
.highest_blocked
== LKM_IVMODE
) {
403 __dlm_lockres_reserve_ast(res
);
404 dlm_queue_bast(dlm
, lock
);
406 if (lock
->ml
.highest_blocked
< target
->ml
.type
)
407 lock
->ml
.highest_blocked
= target
->ml
.type
;
411 /* we can grant the blocked lock (only
412 * possible if converting list empty) */
414 spin_lock(&target
->spinlock
);
415 BUG_ON(target
->ml
.highest_blocked
!= LKM_IVMODE
);
417 mlog(0, "calling ast for blocked lock: %.*s, granting: %d, "
418 "node: %u\n", res
->lockname
.len
, res
->lockname
.name
,
419 target
->ml
.type
, target
->ml
.node
);
421 // target->ml.type is already correct
422 list_move_tail(&target
->list
, &res
->granted
);
424 BUG_ON(!target
->lksb
);
425 target
->lksb
->status
= DLM_NORMAL
;
427 spin_unlock(&target
->spinlock
);
429 __dlm_lockres_reserve_ast(res
);
430 dlm_queue_ast(dlm
, target
);
431 /* go back and check for more */
439 /* must have NO locks when calling this with res !=NULL * */
440 void dlm_kick_thread(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
)
442 mlog_entry("dlm=%p, res=%p\n", dlm
, res
);
444 spin_lock(&dlm
->spinlock
);
445 spin_lock(&res
->spinlock
);
446 __dlm_dirty_lockres(dlm
, res
);
447 spin_unlock(&res
->spinlock
);
448 spin_unlock(&dlm
->spinlock
);
450 wake_up(&dlm
->dlm_thread_wq
);
453 void __dlm_dirty_lockres(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
)
455 mlog_entry("dlm=%p, res=%p\n", dlm
, res
);
457 assert_spin_locked(&dlm
->spinlock
);
458 assert_spin_locked(&res
->spinlock
);
460 /* don't shuffle secondary queues */
461 if ((res
->owner
== dlm
->node_num
)) {
462 if (res
->state
& (DLM_LOCK_RES_MIGRATING
|
463 DLM_LOCK_RES_BLOCK_DIRTY
))
466 if (list_empty(&res
->dirty
)) {
467 /* ref for dirty_list */
468 dlm_lockres_get(res
);
469 list_add_tail(&res
->dirty
, &dlm
->dirty_list
);
470 res
->state
|= DLM_LOCK_RES_DIRTY
;
476 /* Launch the NM thread for the mounted volume */
477 int dlm_launch_thread(struct dlm_ctxt
*dlm
)
479 mlog(0, "starting dlm thread...\n");
481 dlm
->dlm_thread_task
= kthread_run(dlm_thread
, dlm
, "dlm_thread");
482 if (IS_ERR(dlm
->dlm_thread_task
)) {
483 mlog_errno(PTR_ERR(dlm
->dlm_thread_task
));
484 dlm
->dlm_thread_task
= NULL
;
491 void dlm_complete_thread(struct dlm_ctxt
*dlm
)
493 if (dlm
->dlm_thread_task
) {
494 mlog(ML_KTHREAD
, "waiting for dlm thread to exit\n");
495 kthread_stop(dlm
->dlm_thread_task
);
496 dlm
->dlm_thread_task
= NULL
;
500 static int dlm_dirty_list_empty(struct dlm_ctxt
*dlm
)
504 spin_lock(&dlm
->spinlock
);
505 empty
= list_empty(&dlm
->dirty_list
);
506 spin_unlock(&dlm
->spinlock
);
511 static void dlm_flush_asts(struct dlm_ctxt
*dlm
)
514 struct dlm_lock
*lock
;
515 struct dlm_lock_resource
*res
;
518 spin_lock(&dlm
->ast_lock
);
519 while (!list_empty(&dlm
->pending_asts
)) {
520 lock
= list_entry(dlm
->pending_asts
.next
,
521 struct dlm_lock
, ast_list
);
522 /* get an extra ref on lock */
525 mlog(0, "delivering an ast for this lockres\n");
527 BUG_ON(!lock
->ast_pending
);
529 /* remove from list (including ref) */
530 list_del_init(&lock
->ast_list
);
532 spin_unlock(&dlm
->ast_lock
);
534 if (lock
->ml
.node
!= dlm
->node_num
) {
535 ret
= dlm_do_remote_ast(dlm
, res
, lock
);
539 dlm_do_local_ast(dlm
, res
, lock
);
541 spin_lock(&dlm
->ast_lock
);
543 /* possible that another ast was queued while
544 * we were delivering the last one */
545 if (!list_empty(&lock
->ast_list
)) {
546 mlog(0, "aha another ast got queued while "
547 "we were finishing the last one. will "
548 "keep the ast_pending flag set.\n");
550 lock
->ast_pending
= 0;
552 /* drop the extra ref.
553 * this may drop it completely. */
555 dlm_lockres_release_ast(dlm
, res
);
558 while (!list_empty(&dlm
->pending_basts
)) {
559 lock
= list_entry(dlm
->pending_basts
.next
,
560 struct dlm_lock
, bast_list
);
561 /* get an extra ref on lock */
565 BUG_ON(!lock
->bast_pending
);
567 /* get the highest blocked lock, and reset */
568 spin_lock(&lock
->spinlock
);
569 BUG_ON(lock
->ml
.highest_blocked
<= LKM_IVMODE
);
570 hi
= lock
->ml
.highest_blocked
;
571 lock
->ml
.highest_blocked
= LKM_IVMODE
;
572 spin_unlock(&lock
->spinlock
);
574 /* remove from list (including ref) */
575 list_del_init(&lock
->bast_list
);
577 spin_unlock(&dlm
->ast_lock
);
579 mlog(0, "delivering a bast for this lockres "
580 "(blocked = %d\n", hi
);
582 if (lock
->ml
.node
!= dlm
->node_num
) {
583 ret
= dlm_send_proxy_bast(dlm
, res
, lock
, hi
);
587 dlm_do_local_bast(dlm
, res
, lock
, hi
);
589 spin_lock(&dlm
->ast_lock
);
591 /* possible that another bast was queued while
592 * we were delivering the last one */
593 if (!list_empty(&lock
->bast_list
)) {
594 mlog(0, "aha another bast got queued while "
595 "we were finishing the last one. will "
596 "keep the bast_pending flag set.\n");
598 lock
->bast_pending
= 0;
600 /* drop the extra ref.
601 * this may drop it completely. */
603 dlm_lockres_release_ast(dlm
, res
);
605 wake_up(&dlm
->ast_wq
);
606 spin_unlock(&dlm
->ast_lock
);
610 #define DLM_THREAD_TIMEOUT_MS (4 * 1000)
611 #define DLM_THREAD_MAX_DIRTY 100
612 #define DLM_THREAD_MAX_ASTS 10
614 static int dlm_thread(void *data
)
616 struct dlm_lock_resource
*res
;
617 struct dlm_ctxt
*dlm
= data
;
618 unsigned long timeout
= msecs_to_jiffies(DLM_THREAD_TIMEOUT_MS
);
620 mlog(0, "dlm thread running for %s...\n", dlm
->name
);
622 while (!kthread_should_stop()) {
623 int n
= DLM_THREAD_MAX_DIRTY
;
625 /* dlm_shutting_down is very point-in-time, but that
626 * doesn't matter as we'll just loop back around if we
627 * get false on the leading edge of a state
629 dlm_run_purge_list(dlm
, dlm_shutting_down(dlm
));
631 /* We really don't want to hold dlm->spinlock while
632 * calling dlm_shuffle_lists on each lockres that
633 * needs to have its queues adjusted and AST/BASTs
634 * run. So let's pull each entry off the dirty_list
635 * and drop dlm->spinlock ASAP. Once off the list,
636 * res->spinlock needs to be taken again to protect
637 * the queues while calling dlm_shuffle_lists. */
638 spin_lock(&dlm
->spinlock
);
639 while (!list_empty(&dlm
->dirty_list
)) {
641 res
= list_entry(dlm
->dirty_list
.next
,
642 struct dlm_lock_resource
, dirty
);
644 /* peel a lockres off, remove it from the list,
645 * unset the dirty flag and drop the dlm lock */
647 dlm_lockres_get(res
);
649 spin_lock(&res
->spinlock
);
650 /* We clear the DLM_LOCK_RES_DIRTY state once we shuffle lists below */
651 list_del_init(&res
->dirty
);
652 spin_unlock(&res
->spinlock
);
653 spin_unlock(&dlm
->spinlock
);
654 /* Drop dirty_list ref */
655 dlm_lockres_put(res
);
657 /* lockres can be re-dirtied/re-added to the
658 * dirty_list in this gap, but that is ok */
660 spin_lock(&res
->spinlock
);
661 if (res
->owner
!= dlm
->node_num
) {
662 __dlm_print_one_lock_resource(res
);
663 mlog(ML_ERROR
, "inprog:%s, mig:%s, reco:%s, dirty:%s\n",
664 res
->state
& DLM_LOCK_RES_IN_PROGRESS
? "yes" : "no",
665 res
->state
& DLM_LOCK_RES_MIGRATING
? "yes" : "no",
666 res
->state
& DLM_LOCK_RES_RECOVERING
? "yes" : "no",
667 res
->state
& DLM_LOCK_RES_DIRTY
? "yes" : "no");
669 BUG_ON(res
->owner
!= dlm
->node_num
);
671 /* it is now ok to move lockreses in these states
672 * to the dirty list, assuming that they will only be
673 * dirty for a short while. */
674 BUG_ON(res
->state
& DLM_LOCK_RES_MIGRATING
);
675 if (res
->state
& (DLM_LOCK_RES_IN_PROGRESS
|
676 DLM_LOCK_RES_RECOVERING
)) {
677 /* move it to the tail and keep going */
678 res
->state
&= ~DLM_LOCK_RES_DIRTY
;
679 spin_unlock(&res
->spinlock
);
680 mlog(0, "delaying list shuffling for in-"
681 "progress lockres %.*s, state=%d\n",
682 res
->lockname
.len
, res
->lockname
.name
,
688 /* at this point the lockres is not migrating/
689 * recovering/in-progress. we have the lockres
690 * spinlock and do NOT have the dlm lock.
691 * safe to reserve/queue asts and run the lists. */
693 mlog(0, "calling dlm_shuffle_lists with dlm=%s, "
694 "res=%.*s\n", dlm
->name
,
695 res
->lockname
.len
, res
->lockname
.name
);
697 /* called while holding lockres lock */
698 dlm_shuffle_lists(dlm
, res
);
699 res
->state
&= ~DLM_LOCK_RES_DIRTY
;
700 spin_unlock(&res
->spinlock
);
702 dlm_lockres_calc_usage(dlm
, res
);
706 spin_lock(&dlm
->spinlock
);
707 /* if the lock was in-progress, stick
708 * it on the back of the list */
710 spin_lock(&res
->spinlock
);
711 __dlm_dirty_lockres(dlm
, res
);
712 spin_unlock(&res
->spinlock
);
714 dlm_lockres_put(res
);
716 /* unlikely, but we may need to give time to
719 mlog(0, "throttling dlm_thread\n");
724 spin_unlock(&dlm
->spinlock
);
727 /* yield and continue right away if there is more work to do */
733 wait_event_interruptible_timeout(dlm
->dlm_thread_wq
,
734 !dlm_dirty_list_empty(dlm
) ||
735 kthread_should_stop(),
739 mlog(0, "quitting DLM thread\n");