1 // SPDX-License-Identifier: GPL-2.0-only
3 * Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
4 * Copyright 2004-2011 Red Hat, Inc.
7 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
10 #include <linux/dlm.h>
11 #include <linux/slab.h>
12 #include <linux/types.h>
13 #include <linux/delay.h>
14 #include <linux/gfs2_ondisk.h>
15 #include <linux/sched/signal.h>
23 #include "trace_gfs2.h"
26 * gfs2_update_stats - Update time based stats
27 * @mv: Pointer to mean/variance structure to update
28 * @sample: New data to include
30 * @delta is the difference between the current rtt sample and the
31 * running average srtt. We add 1/8 of that to the srtt in order to
32 * update the current srtt estimate. The variance estimate is a bit
33 * more complicated. We subtract the current variance estimate from
34 * the abs value of the @delta and add 1/4 of that to the running
35 * total. That's equivalent to 3/4 of the current variance
36 * estimate plus 1/4 of the abs of @delta.
38 * Note that the index points at the array entry containing the smoothed
39 * mean value, and the variance is always in the following entry
41 * Reference: TCP/IP Illustrated, vol 2, p. 831,832
42 * All times are in units of integer nanoseconds. Unlike the TCP/IP case,
43 * they are not scaled fixed point.
46 static inline void gfs2_update_stats(struct gfs2_lkstats *s, unsigned index,
49 s64 delta = sample - s->stats[index];
50 s->stats[index] += (delta >> 3);
52 s->stats[index] += (s64)(abs(delta) - s->stats[index]) >> 2;
56 * gfs2_update_reply_times - Update locking statistics
57 * @gl: The glock to update
59 * This assumes that gl->gl_dstamp has been set earlier.
61 * The rtt (lock round trip time) is an estimate of the time
62 * taken to perform a dlm lock request. We update it on each
65 * The blocking flag is set on the glock for all dlm requests
66 * which may potentially block due to lock requests from other nodes.
67 * DLM requests where the current lock state is exclusive, the
68 * requested state is null (or unlocked) or where the TRY or
69 * TRY_1CB flags are set are classified as non-blocking. All
70 * other DLM requests are counted as (potentially) blocking.
72 static inline void gfs2_update_reply_times(struct gfs2_glock *gl)
74 struct gfs2_pcpu_lkstats *lks;
75 const unsigned gltype = gl->gl_name.ln_type;
76 unsigned index = test_bit(GLF_BLOCKING, &gl->gl_flags) ?
77 GFS2_LKS_SRTTB : GFS2_LKS_SRTT;
81 rtt = ktime_to_ns(ktime_sub(ktime_get_real(), gl->gl_dstamp));
82 lks = this_cpu_ptr(gl->gl_name.ln_sbd->sd_lkstats);
83 gfs2_update_stats(&gl->gl_stats, index, rtt); /* Local */
84 gfs2_update_stats(&lks->lkstats[gltype], index, rtt); /* Global */
87 trace_gfs2_glock_lock_time(gl, rtt);
91 * gfs2_update_request_times - Update locking statistics
92 * @gl: The glock to update
94 * The irt (lock inter-request times) measures the average time
95 * between requests to the dlm. It is updated immediately before
99 static inline void gfs2_update_request_times(struct gfs2_glock *gl)
101 struct gfs2_pcpu_lkstats *lks;
102 const unsigned gltype = gl->gl_name.ln_type;
107 dstamp = gl->gl_dstamp;
108 gl->gl_dstamp = ktime_get_real();
109 irt = ktime_to_ns(ktime_sub(gl->gl_dstamp, dstamp));
110 lks = this_cpu_ptr(gl->gl_name.ln_sbd->sd_lkstats);
111 gfs2_update_stats(&gl->gl_stats, GFS2_LKS_SIRT, irt); /* Local */
112 gfs2_update_stats(&lks->lkstats[gltype], GFS2_LKS_SIRT, irt); /* Global */
116 static void gdlm_ast(void *arg)
118 struct gfs2_glock *gl = arg;
119 unsigned ret = gl->gl_state;
121 gfs2_update_reply_times(gl);
122 BUG_ON(gl->gl_lksb.sb_flags & DLM_SBF_DEMOTED);
124 if ((gl->gl_lksb.sb_flags & DLM_SBF_VALNOTVALID) && gl->gl_lksb.sb_lvbptr)
125 memset(gl->gl_lksb.sb_lvbptr, 0, GDLM_LVB_SIZE);
127 switch (gl->gl_lksb.sb_status) {
128 case -DLM_EUNLOCK: /* Unlocked, so glock can be freed */
129 if (gl->gl_ops->go_free)
130 gl->gl_ops->go_free(gl);
133 case -DLM_ECANCEL: /* Cancel while getting lock */
134 ret |= LM_OUT_CANCELED;
136 case -EAGAIN: /* Try lock fails */
137 case -EDEADLK: /* Deadlock detected */
139 case -ETIMEDOUT: /* Canceled due to timeout */
142 case 0: /* Success */
144 default: /* Something unexpected */
149 if (gl->gl_lksb.sb_flags & DLM_SBF_ALTMODE) {
150 if (gl->gl_req == LM_ST_SHARED)
151 ret = LM_ST_DEFERRED;
152 else if (gl->gl_req == LM_ST_DEFERRED)
158 set_bit(GLF_INITIAL, &gl->gl_flags);
159 gfs2_glock_complete(gl, ret);
162 if (!test_bit(GLF_INITIAL, &gl->gl_flags))
163 gl->gl_lksb.sb_lkid = 0;
164 gfs2_glock_complete(gl, ret);
167 static void gdlm_bast(void *arg, int mode)
169 struct gfs2_glock *gl = arg;
173 gfs2_glock_cb(gl, LM_ST_UNLOCKED);
176 gfs2_glock_cb(gl, LM_ST_DEFERRED);
179 gfs2_glock_cb(gl, LM_ST_SHARED);
182 fs_err(gl->gl_name.ln_sbd, "unknown bast mode %d\n", mode);
187 /* convert gfs lock-state to dlm lock-mode */
189 static int make_mode(struct gfs2_sbd *sdp, const unsigned int lmstate)
194 case LM_ST_EXCLUSIVE:
201 fs_err(sdp, "unknown LM state %d\n", lmstate);
206 static u32 make_flags(struct gfs2_glock *gl, const unsigned int gfs_flags,
211 if (gl->gl_lksb.sb_lvbptr)
212 lkf |= DLM_LKF_VALBLK;
214 if (gfs_flags & LM_FLAG_TRY)
215 lkf |= DLM_LKF_NOQUEUE;
217 if (gfs_flags & LM_FLAG_TRY_1CB) {
218 lkf |= DLM_LKF_NOQUEUE;
219 lkf |= DLM_LKF_NOQUEUEBAST;
222 if (gfs_flags & LM_FLAG_PRIORITY) {
223 lkf |= DLM_LKF_NOORDER;
224 lkf |= DLM_LKF_HEADQUE;
227 if (gfs_flags & LM_FLAG_ANY) {
228 if (req == DLM_LOCK_PR)
229 lkf |= DLM_LKF_ALTCW;
230 else if (req == DLM_LOCK_CW)
231 lkf |= DLM_LKF_ALTPR;
236 if (gl->gl_lksb.sb_lkid != 0) {
237 lkf |= DLM_LKF_CONVERT;
238 if (test_bit(GLF_BLOCKING, &gl->gl_flags))
239 lkf |= DLM_LKF_QUECVT;
245 static void gfs2_reverse_hex(char *c, u64 value)
249 *c-- = hex_asc[value & 0x0f];
254 static int gdlm_lock(struct gfs2_glock *gl, unsigned int req_state,
257 struct lm_lockstruct *ls = &gl->gl_name.ln_sbd->sd_lockstruct;
260 char strname[GDLM_STRNAME_BYTES] = "";
262 req = make_mode(gl->gl_name.ln_sbd, req_state);
263 lkf = make_flags(gl, flags, req);
264 gfs2_glstats_inc(gl, GFS2_LKS_DCOUNT);
265 gfs2_sbstats_inc(gl, GFS2_LKS_DCOUNT);
266 if (gl->gl_lksb.sb_lkid) {
267 gfs2_update_request_times(gl);
269 memset(strname, ' ', GDLM_STRNAME_BYTES - 1);
270 strname[GDLM_STRNAME_BYTES - 1] = '\0';
271 gfs2_reverse_hex(strname + 7, gl->gl_name.ln_type);
272 gfs2_reverse_hex(strname + 23, gl->gl_name.ln_number);
273 gl->gl_dstamp = ktime_get_real();
276 * Submit the actual lock request.
279 return dlm_lock(ls->ls_dlm, req, &gl->gl_lksb, lkf, strname,
280 GDLM_STRNAME_BYTES - 1, 0, gdlm_ast, gl, gdlm_bast);
283 static void gdlm_put_lock(struct gfs2_glock *gl)
285 struct gfs2_sbd *sdp = gl->gl_name.ln_sbd;
286 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
289 if (gl->gl_lksb.sb_lkid == 0) {
294 clear_bit(GLF_BLOCKING, &gl->gl_flags);
295 gfs2_glstats_inc(gl, GFS2_LKS_DCOUNT);
296 gfs2_sbstats_inc(gl, GFS2_LKS_DCOUNT);
297 gfs2_update_request_times(gl);
299 /* don't want to skip dlm_unlock writing the lvb when lock has one */
301 if (test_bit(SDF_SKIP_DLM_UNLOCK, &sdp->sd_flags) &&
302 !gl->gl_lksb.sb_lvbptr) {
307 error = dlm_unlock(ls->ls_dlm, gl->gl_lksb.sb_lkid, DLM_LKF_VALBLK,
310 fs_err(sdp, "gdlm_unlock %x,%llx err=%d\n",
312 (unsigned long long)gl->gl_name.ln_number, error);
317 static void gdlm_cancel(struct gfs2_glock *gl)
319 struct lm_lockstruct *ls = &gl->gl_name.ln_sbd->sd_lockstruct;
320 dlm_unlock(ls->ls_dlm, gl->gl_lksb.sb_lkid, DLM_LKF_CANCEL, NULL, gl);
324 * dlm/gfs2 recovery coordination using dlm_recover callbacks
326 * 0. gfs2 checks for another cluster node withdraw, needing journal replay
327 * 1. dlm_controld sees lockspace members change
328 * 2. dlm_controld blocks dlm-kernel locking activity
329 * 3. dlm_controld within dlm-kernel notifies gfs2 (recover_prep)
330 * 4. dlm_controld starts and finishes its own user level recovery
331 * 5. dlm_controld starts dlm-kernel dlm_recoverd to do kernel recovery
332 * 6. dlm_recoverd notifies gfs2 of failed nodes (recover_slot)
333 * 7. dlm_recoverd does its own lock recovery
334 * 8. dlm_recoverd unblocks dlm-kernel locking activity
335 * 9. dlm_recoverd notifies gfs2 when done (recover_done with new generation)
336 * 10. gfs2_control updates control_lock lvb with new generation and jid bits
337 * 11. gfs2_control enqueues journals for gfs2_recover to recover (maybe none)
338 * 12. gfs2_recover dequeues and recovers journals of failed nodes
339 * 13. gfs2_recover provides recovery results to gfs2_control (recovery_result)
340 * 14. gfs2_control updates control_lock lvb jid bits for recovered journals
341 * 15. gfs2_control unblocks normal locking when all journals are recovered
343 * - failures during recovery
345 * recover_prep() may set BLOCK_LOCKS (step 3) again before gfs2_control
346 * clears BLOCK_LOCKS (step 15), e.g. another node fails while still
347 * recovering for a prior failure. gfs2_control needs a way to detect
348 * this so it can leave BLOCK_LOCKS set in step 15. This is managed using
349 * the recover_block and recover_start values.
351 * recover_done() provides a new lockspace generation number each time it
352 * is called (step 9). This generation number is saved as recover_start.
353 * When recover_prep() is called, it sets BLOCK_LOCKS and sets
354 * recover_block = recover_start. So, while recover_block is equal to
355 * recover_start, BLOCK_LOCKS should remain set. (recover_spin must
356 * be held around the BLOCK_LOCKS/recover_block/recover_start logic.)
358 * - more specific gfs2 steps in sequence above
360 * 3. recover_prep sets BLOCK_LOCKS and sets recover_block = recover_start
361 * 6. recover_slot records any failed jids (maybe none)
362 * 9. recover_done sets recover_start = new generation number
363 * 10. gfs2_control sets control_lock lvb = new gen + bits for failed jids
364 * 12. gfs2_recover does journal recoveries for failed jids identified above
365 * 14. gfs2_control clears control_lock lvb bits for recovered jids
366 * 15. gfs2_control checks if recover_block == recover_start (step 3 occured
367 * again) then do nothing, otherwise if recover_start > recover_block
368 * then clear BLOCK_LOCKS.
370 * - parallel recovery steps across all nodes
372 * All nodes attempt to update the control_lock lvb with the new generation
373 * number and jid bits, but only the first to get the control_lock EX will
374 * do so; others will see that it's already done (lvb already contains new
375 * generation number.)
377 * . All nodes get the same recover_prep/recover_slot/recover_done callbacks
378 * . All nodes attempt to set control_lock lvb gen + bits for the new gen
379 * . One node gets control_lock first and writes the lvb, others see it's done
380 * . All nodes attempt to recover jids for which they see control_lock bits set
381 * . One node succeeds for a jid, and that one clears the jid bit in the lvb
382 * . All nodes will eventually see all lvb bits clear and unblock locks
384 * - is there a problem with clearing an lvb bit that should be set
385 * and missing a journal recovery?
388 * 2. lvb bit set for step 1
389 * 3. jid recovered for step 1
390 * 4. jid taken again (new mount)
391 * 5. jid fails (for step 4)
392 * 6. lvb bit set for step 5 (will already be set)
393 * 7. lvb bit cleared for step 3
395 * This is not a problem because the failure in step 5 does not
396 * require recovery, because the mount in step 4 could not have
397 * progressed far enough to unblock locks and access the fs. The
398 * control_mount() function waits for all recoveries to be complete
399 * for the latest lockspace generation before ever unblocking locks
400 * and returning. The mount in step 4 waits until the recovery in
403 * - special case of first mounter: first node to mount the fs
405 * The first node to mount a gfs2 fs needs to check all the journals
406 * and recover any that need recovery before other nodes are allowed
407 * to mount the fs. (Others may begin mounting, but they must wait
408 * for the first mounter to be done before taking locks on the fs
409 * or accessing the fs.) This has two parts:
411 * 1. The mounted_lock tells a node it's the first to mount the fs.
412 * Each node holds the mounted_lock in PR while it's mounted.
413 * Each node tries to acquire the mounted_lock in EX when it mounts.
414 * If a node is granted the mounted_lock EX it means there are no
415 * other mounted nodes (no PR locks exist), and it is the first mounter.
416 * The mounted_lock is demoted to PR when first recovery is done, so
417 * others will fail to get an EX lock, but will get a PR lock.
419 * 2. The control_lock blocks others in control_mount() while the first
420 * mounter is doing first mount recovery of all journals.
421 * A mounting node needs to acquire control_lock in EX mode before
422 * it can proceed. The first mounter holds control_lock in EX while doing
423 * the first mount recovery, blocking mounts from other nodes, then demotes
424 * control_lock to NL when it's done (others_may_mount/first_done),
425 * allowing other nodes to continue mounting.
428 * control_lock EX/NOQUEUE success
429 * mounted_lock EX/NOQUEUE success (no other PR, so no other mounters)
431 * do first mounter recovery
432 * mounted_lock EX->PR
433 * control_lock EX->NL, write lvb generation
436 * control_lock EX/NOQUEUE success (if fail -EAGAIN, retry)
437 * mounted_lock EX/NOQUEUE fail -EAGAIN (expected due to other mounters PR)
438 * mounted_lock PR/NOQUEUE success
439 * read lvb generation
440 * control_lock EX->NL
443 * - mount during recovery
445 * If a node mounts while others are doing recovery (not first mounter),
446 * the mounting node will get its initial recover_done() callback without
447 * having seen any previous failures/callbacks.
449 * It must wait for all recoveries preceding its mount to be finished
450 * before it unblocks locks. It does this by repeating the "other mounter"
451 * steps above until the lvb generation number is >= its mount generation
452 * number (from initial recover_done) and all lvb bits are clear.
454 * - control_lock lvb format
456 * 4 bytes generation number: the latest dlm lockspace generation number
457 * from recover_done callback. Indicates the jid bitmap has been updated
458 * to reflect all slot failures through that generation.
460 * GDLM_LVB_SIZE-8 bytes of jid bit map. If bit N is set, it indicates
461 * that jid N needs recovery.
464 #define JID_BITMAP_OFFSET 8 /* 4 byte generation number + 4 byte unused */
466 static void control_lvb_read(struct lm_lockstruct *ls, uint32_t *lvb_gen,
470 memcpy(lvb_bits, ls->ls_control_lvb, GDLM_LVB_SIZE);
471 memcpy(&gen, lvb_bits, sizeof(__le32));
472 *lvb_gen = le32_to_cpu(gen);
475 static void control_lvb_write(struct lm_lockstruct *ls, uint32_t lvb_gen,
479 memcpy(ls->ls_control_lvb, lvb_bits, GDLM_LVB_SIZE);
480 gen = cpu_to_le32(lvb_gen);
481 memcpy(ls->ls_control_lvb, &gen, sizeof(__le32));
484 static int all_jid_bits_clear(char *lvb)
486 return !memchr_inv(lvb + JID_BITMAP_OFFSET, 0,
487 GDLM_LVB_SIZE - JID_BITMAP_OFFSET);
490 static void sync_wait_cb(void *arg)
492 struct lm_lockstruct *ls = arg;
493 complete(&ls->ls_sync_wait);
496 static int sync_unlock(struct gfs2_sbd *sdp, struct dlm_lksb *lksb, char *name)
498 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
501 error = dlm_unlock(ls->ls_dlm, lksb->sb_lkid, 0, lksb, ls);
503 fs_err(sdp, "%s lkid %x error %d\n",
504 name, lksb->sb_lkid, error);
508 wait_for_completion(&ls->ls_sync_wait);
510 if (lksb->sb_status != -DLM_EUNLOCK) {
511 fs_err(sdp, "%s lkid %x status %d\n",
512 name, lksb->sb_lkid, lksb->sb_status);
518 static int sync_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags,
519 unsigned int num, struct dlm_lksb *lksb, char *name)
521 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
522 char strname[GDLM_STRNAME_BYTES];
525 memset(strname, 0, GDLM_STRNAME_BYTES);
526 snprintf(strname, GDLM_STRNAME_BYTES, "%8x%16x", LM_TYPE_NONDISK, num);
528 error = dlm_lock(ls->ls_dlm, mode, lksb, flags,
529 strname, GDLM_STRNAME_BYTES - 1,
530 0, sync_wait_cb, ls, NULL);
532 fs_err(sdp, "%s lkid %x flags %x mode %d error %d\n",
533 name, lksb->sb_lkid, flags, mode, error);
537 wait_for_completion(&ls->ls_sync_wait);
539 status = lksb->sb_status;
541 if (status && status != -EAGAIN) {
542 fs_err(sdp, "%s lkid %x flags %x mode %d status %d\n",
543 name, lksb->sb_lkid, flags, mode, status);
549 static int mounted_unlock(struct gfs2_sbd *sdp)
551 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
552 return sync_unlock(sdp, &ls->ls_mounted_lksb, "mounted_lock");
555 static int mounted_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags)
557 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
558 return sync_lock(sdp, mode, flags, GFS2_MOUNTED_LOCK,
559 &ls->ls_mounted_lksb, "mounted_lock");
562 static int control_unlock(struct gfs2_sbd *sdp)
564 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
565 return sync_unlock(sdp, &ls->ls_control_lksb, "control_lock");
568 static int control_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags)
570 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
571 return sync_lock(sdp, mode, flags, GFS2_CONTROL_LOCK,
572 &ls->ls_control_lksb, "control_lock");
576 * remote_withdraw - react to a node withdrawing from the file system
577 * @sdp: The superblock
579 static void remote_withdraw(struct gfs2_sbd *sdp)
581 struct gfs2_jdesc *jd;
582 int ret = 0, count = 0;
584 list_for_each_entry(jd, &sdp->sd_jindex_list, jd_list) {
585 if (jd->jd_jid == sdp->sd_lockstruct.ls_jid)
587 ret = gfs2_recover_journal(jd, true);
593 /* Now drop the additional reference we acquired */
594 fs_err(sdp, "Journals checked: %d, ret = %d.\n", count, ret);
597 static void gfs2_control_func(struct work_struct *work)
599 struct gfs2_sbd *sdp = container_of(work, struct gfs2_sbd, sd_control_work.work);
600 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
601 uint32_t block_gen, start_gen, lvb_gen, flags;
607 /* First check for other nodes that may have done a withdraw. */
608 if (test_bit(SDF_REMOTE_WITHDRAW, &sdp->sd_flags)) {
609 remote_withdraw(sdp);
610 clear_bit(SDF_REMOTE_WITHDRAW, &sdp->sd_flags);
614 spin_lock(&ls->ls_recover_spin);
616 * No MOUNT_DONE means we're still mounting; control_mount()
617 * will set this flag, after which this thread will take over
618 * all further clearing of BLOCK_LOCKS.
620 * FIRST_MOUNT means this node is doing first mounter recovery,
621 * for which recovery control is handled by
622 * control_mount()/control_first_done(), not this thread.
624 if (!test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) ||
625 test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
626 spin_unlock(&ls->ls_recover_spin);
629 block_gen = ls->ls_recover_block;
630 start_gen = ls->ls_recover_start;
631 spin_unlock(&ls->ls_recover_spin);
634 * Equal block_gen and start_gen implies we are between
635 * recover_prep and recover_done callbacks, which means
636 * dlm recovery is in progress and dlm locking is blocked.
637 * There's no point trying to do any work until recover_done.
640 if (block_gen == start_gen)
644 * Propagate recover_submit[] and recover_result[] to lvb:
645 * dlm_recoverd adds to recover_submit[] jids needing recovery
646 * gfs2_recover adds to recover_result[] journal recovery results
648 * set lvb bit for jids in recover_submit[] if the lvb has not
649 * yet been updated for the generation of the failure
651 * clear lvb bit for jids in recover_result[] if the result of
652 * the journal recovery is SUCCESS
655 error = control_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_VALBLK);
657 fs_err(sdp, "control lock EX error %d\n", error);
661 control_lvb_read(ls, &lvb_gen, ls->ls_lvb_bits);
663 spin_lock(&ls->ls_recover_spin);
664 if (block_gen != ls->ls_recover_block ||
665 start_gen != ls->ls_recover_start) {
666 fs_info(sdp, "recover generation %u block1 %u %u\n",
667 start_gen, block_gen, ls->ls_recover_block);
668 spin_unlock(&ls->ls_recover_spin);
669 control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT);
673 recover_size = ls->ls_recover_size;
675 if (lvb_gen <= start_gen) {
677 * Clear lvb bits for jids we've successfully recovered.
678 * Because all nodes attempt to recover failed journals,
679 * a journal can be recovered multiple times successfully
680 * in succession. Only the first will really do recovery,
681 * the others find it clean, but still report a successful
682 * recovery. So, another node may have already recovered
683 * the jid and cleared the lvb bit for it.
685 for (i = 0; i < recover_size; i++) {
686 if (ls->ls_recover_result[i] != LM_RD_SUCCESS)
689 ls->ls_recover_result[i] = 0;
691 if (!test_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET))
694 __clear_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET);
699 if (lvb_gen == start_gen) {
701 * Failed slots before start_gen are already set in lvb.
703 for (i = 0; i < recover_size; i++) {
704 if (!ls->ls_recover_submit[i])
706 if (ls->ls_recover_submit[i] < lvb_gen)
707 ls->ls_recover_submit[i] = 0;
709 } else if (lvb_gen < start_gen) {
711 * Failed slots before start_gen are not yet set in lvb.
713 for (i = 0; i < recover_size; i++) {
714 if (!ls->ls_recover_submit[i])
716 if (ls->ls_recover_submit[i] < start_gen) {
717 ls->ls_recover_submit[i] = 0;
718 __set_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET);
721 /* even if there are no bits to set, we need to write the
722 latest generation to the lvb */
726 * we should be getting a recover_done() for lvb_gen soon
729 spin_unlock(&ls->ls_recover_spin);
732 control_lvb_write(ls, start_gen, ls->ls_lvb_bits);
733 flags = DLM_LKF_CONVERT | DLM_LKF_VALBLK;
735 flags = DLM_LKF_CONVERT;
738 error = control_lock(sdp, DLM_LOCK_NL, flags);
740 fs_err(sdp, "control lock NL error %d\n", error);
745 * Everyone will see jid bits set in the lvb, run gfs2_recover_set(),
746 * and clear a jid bit in the lvb if the recovery is a success.
747 * Eventually all journals will be recovered, all jid bits will
748 * be cleared in the lvb, and everyone will clear BLOCK_LOCKS.
751 for (i = 0; i < recover_size; i++) {
752 if (test_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET)) {
753 fs_info(sdp, "recover generation %u jid %d\n",
755 gfs2_recover_set(sdp, i);
763 * No more jid bits set in lvb, all recovery is done, unblock locks
764 * (unless a new recover_prep callback has occured blocking locks
765 * again while working above)
768 spin_lock(&ls->ls_recover_spin);
769 if (ls->ls_recover_block == block_gen &&
770 ls->ls_recover_start == start_gen) {
771 clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
772 spin_unlock(&ls->ls_recover_spin);
773 fs_info(sdp, "recover generation %u done\n", start_gen);
774 gfs2_glock_thaw(sdp);
776 fs_info(sdp, "recover generation %u block2 %u %u\n",
777 start_gen, block_gen, ls->ls_recover_block);
778 spin_unlock(&ls->ls_recover_spin);
782 static int control_mount(struct gfs2_sbd *sdp)
784 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
785 uint32_t start_gen, block_gen, mount_gen, lvb_gen;
790 memset(&ls->ls_mounted_lksb, 0, sizeof(struct dlm_lksb));
791 memset(&ls->ls_control_lksb, 0, sizeof(struct dlm_lksb));
792 memset(&ls->ls_control_lvb, 0, GDLM_LVB_SIZE);
793 ls->ls_control_lksb.sb_lvbptr = ls->ls_control_lvb;
794 init_completion(&ls->ls_sync_wait);
796 set_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
798 error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_VALBLK);
800 fs_err(sdp, "control_mount control_lock NL error %d\n", error);
804 error = mounted_lock(sdp, DLM_LOCK_NL, 0);
806 fs_err(sdp, "control_mount mounted_lock NL error %d\n", error);
810 mounted_mode = DLM_LOCK_NL;
813 if (retries++ && signal_pending(current)) {
819 * We always start with both locks in NL. control_lock is
820 * demoted to NL below so we don't need to do it here.
823 if (mounted_mode != DLM_LOCK_NL) {
824 error = mounted_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT);
827 mounted_mode = DLM_LOCK_NL;
831 * Other nodes need to do some work in dlm recovery and gfs2_control
832 * before the recover_done and control_lock will be ready for us below.
833 * A delay here is not required but often avoids having to retry.
836 msleep_interruptible(500);
839 * Acquire control_lock in EX and mounted_lock in either EX or PR.
840 * control_lock lvb keeps track of any pending journal recoveries.
841 * mounted_lock indicates if any other nodes have the fs mounted.
844 error = control_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE|DLM_LKF_VALBLK);
845 if (error == -EAGAIN) {
848 fs_err(sdp, "control_mount control_lock EX error %d\n", error);
853 * If we're a spectator, we don't want to take the lock in EX because
854 * we cannot do the first-mount responsibility it implies: recovery.
856 if (sdp->sd_args.ar_spectator)
859 error = mounted_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE);
861 mounted_mode = DLM_LOCK_EX;
863 } else if (error != -EAGAIN) {
864 fs_err(sdp, "control_mount mounted_lock EX error %d\n", error);
868 error = mounted_lock(sdp, DLM_LOCK_PR, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE);
870 mounted_mode = DLM_LOCK_PR;
873 /* not even -EAGAIN should happen here */
874 fs_err(sdp, "control_mount mounted_lock PR error %d\n", error);
880 * If we got both locks above in EX, then we're the first mounter.
881 * If not, then we need to wait for the control_lock lvb to be
882 * updated by other mounted nodes to reflect our mount generation.
884 * In simple first mounter cases, first mounter will see zero lvb_gen,
885 * but in cases where all existing nodes leave/fail before mounting
886 * nodes finish control_mount, then all nodes will be mounting and
887 * lvb_gen will be non-zero.
890 control_lvb_read(ls, &lvb_gen, ls->ls_lvb_bits);
892 if (lvb_gen == 0xFFFFFFFF) {
893 /* special value to force mount attempts to fail */
894 fs_err(sdp, "control_mount control_lock disabled\n");
899 if (mounted_mode == DLM_LOCK_EX) {
900 /* first mounter, keep both EX while doing first recovery */
901 spin_lock(&ls->ls_recover_spin);
902 clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
903 set_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags);
904 set_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags);
905 spin_unlock(&ls->ls_recover_spin);
906 fs_info(sdp, "first mounter control generation %u\n", lvb_gen);
910 error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT);
915 * We are not first mounter, now we need to wait for the control_lock
916 * lvb generation to be >= the generation from our first recover_done
917 * and all lvb bits to be clear (no pending journal recoveries.)
920 if (!all_jid_bits_clear(ls->ls_lvb_bits)) {
921 /* journals need recovery, wait until all are clear */
922 fs_info(sdp, "control_mount wait for journal recovery\n");
926 spin_lock(&ls->ls_recover_spin);
927 block_gen = ls->ls_recover_block;
928 start_gen = ls->ls_recover_start;
929 mount_gen = ls->ls_recover_mount;
931 if (lvb_gen < mount_gen) {
932 /* wait for mounted nodes to update control_lock lvb to our
933 generation, which might include new recovery bits set */
934 if (sdp->sd_args.ar_spectator) {
935 fs_info(sdp, "Recovery is required. Waiting for a "
936 "non-spectator to mount.\n");
937 msleep_interruptible(1000);
939 fs_info(sdp, "control_mount wait1 block %u start %u "
940 "mount %u lvb %u flags %lx\n", block_gen,
941 start_gen, mount_gen, lvb_gen,
942 ls->ls_recover_flags);
944 spin_unlock(&ls->ls_recover_spin);
948 if (lvb_gen != start_gen) {
949 /* wait for mounted nodes to update control_lock lvb to the
950 latest recovery generation */
951 fs_info(sdp, "control_mount wait2 block %u start %u mount %u "
952 "lvb %u flags %lx\n", block_gen, start_gen, mount_gen,
953 lvb_gen, ls->ls_recover_flags);
954 spin_unlock(&ls->ls_recover_spin);
958 if (block_gen == start_gen) {
959 /* dlm recovery in progress, wait for it to finish */
960 fs_info(sdp, "control_mount wait3 block %u start %u mount %u "
961 "lvb %u flags %lx\n", block_gen, start_gen, mount_gen,
962 lvb_gen, ls->ls_recover_flags);
963 spin_unlock(&ls->ls_recover_spin);
967 clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
968 set_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags);
969 memset(ls->ls_recover_submit, 0, ls->ls_recover_size*sizeof(uint32_t));
970 memset(ls->ls_recover_result, 0, ls->ls_recover_size*sizeof(uint32_t));
971 spin_unlock(&ls->ls_recover_spin);
980 static int control_first_done(struct gfs2_sbd *sdp)
982 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
983 uint32_t start_gen, block_gen;
987 spin_lock(&ls->ls_recover_spin);
988 start_gen = ls->ls_recover_start;
989 block_gen = ls->ls_recover_block;
991 if (test_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags) ||
992 !test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) ||
993 !test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
994 /* sanity check, should not happen */
995 fs_err(sdp, "control_first_done start %u block %u flags %lx\n",
996 start_gen, block_gen, ls->ls_recover_flags);
997 spin_unlock(&ls->ls_recover_spin);
1002 if (start_gen == block_gen) {
1004 * Wait for the end of a dlm recovery cycle to switch from
1005 * first mounter recovery. We can ignore any recover_slot
1006 * callbacks between the recover_prep and next recover_done
1007 * because we are still the first mounter and any failed nodes
1008 * have not fully mounted, so they don't need recovery.
1010 spin_unlock(&ls->ls_recover_spin);
1011 fs_info(sdp, "control_first_done wait gen %u\n", start_gen);
1013 wait_on_bit(&ls->ls_recover_flags, DFL_DLM_RECOVERY,
1014 TASK_UNINTERRUPTIBLE);
1018 clear_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags);
1019 set_bit(DFL_FIRST_MOUNT_DONE, &ls->ls_recover_flags);
1020 memset(ls->ls_recover_submit, 0, ls->ls_recover_size*sizeof(uint32_t));
1021 memset(ls->ls_recover_result, 0, ls->ls_recover_size*sizeof(uint32_t));
1022 spin_unlock(&ls->ls_recover_spin);
1024 memset(ls->ls_lvb_bits, 0, GDLM_LVB_SIZE);
1025 control_lvb_write(ls, start_gen, ls->ls_lvb_bits);
1027 error = mounted_lock(sdp, DLM_LOCK_PR, DLM_LKF_CONVERT);
1029 fs_err(sdp, "control_first_done mounted PR error %d\n", error);
1031 error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT|DLM_LKF_VALBLK);
1033 fs_err(sdp, "control_first_done control NL error %d\n", error);
1039 * Expand static jid arrays if necessary (by increments of RECOVER_SIZE_INC)
1040 * to accomodate the largest slot number. (NB dlm slot numbers start at 1,
1041 * gfs2 jids start at 0, so jid = slot - 1)
1044 #define RECOVER_SIZE_INC 16
1046 static int set_recover_size(struct gfs2_sbd *sdp, struct dlm_slot *slots,
1049 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1050 uint32_t *submit = NULL;
1051 uint32_t *result = NULL;
1052 uint32_t old_size, new_size;
1055 if (!ls->ls_lvb_bits) {
1056 ls->ls_lvb_bits = kzalloc(GDLM_LVB_SIZE, GFP_NOFS);
1057 if (!ls->ls_lvb_bits)
1062 for (i = 0; i < num_slots; i++) {
1063 if (max_jid < slots[i].slot - 1)
1064 max_jid = slots[i].slot - 1;
1067 old_size = ls->ls_recover_size;
1068 new_size = old_size;
1069 while (new_size < max_jid + 1)
1070 new_size += RECOVER_SIZE_INC;
1071 if (new_size == old_size)
1074 submit = kcalloc(new_size, sizeof(uint32_t), GFP_NOFS);
1075 result = kcalloc(new_size, sizeof(uint32_t), GFP_NOFS);
1076 if (!submit || !result) {
1082 spin_lock(&ls->ls_recover_spin);
1083 memcpy(submit, ls->ls_recover_submit, old_size * sizeof(uint32_t));
1084 memcpy(result, ls->ls_recover_result, old_size * sizeof(uint32_t));
1085 kfree(ls->ls_recover_submit);
1086 kfree(ls->ls_recover_result);
1087 ls->ls_recover_submit = submit;
1088 ls->ls_recover_result = result;
1089 ls->ls_recover_size = new_size;
1090 spin_unlock(&ls->ls_recover_spin);
1094 static void free_recover_size(struct lm_lockstruct *ls)
1096 kfree(ls->ls_lvb_bits);
1097 kfree(ls->ls_recover_submit);
1098 kfree(ls->ls_recover_result);
1099 ls->ls_recover_submit = NULL;
1100 ls->ls_recover_result = NULL;
1101 ls->ls_recover_size = 0;
1102 ls->ls_lvb_bits = NULL;
1105 /* dlm calls before it does lock recovery */
1107 static void gdlm_recover_prep(void *arg)
1109 struct gfs2_sbd *sdp = arg;
1110 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1112 if (gfs2_withdrawn(sdp)) {
1113 fs_err(sdp, "recover_prep ignored due to withdraw.\n");
1116 spin_lock(&ls->ls_recover_spin);
1117 ls->ls_recover_block = ls->ls_recover_start;
1118 set_bit(DFL_DLM_RECOVERY, &ls->ls_recover_flags);
1120 if (!test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) ||
1121 test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
1122 spin_unlock(&ls->ls_recover_spin);
1125 set_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
1126 spin_unlock(&ls->ls_recover_spin);
1129 /* dlm calls after recover_prep has been completed on all lockspace members;
1130 identifies slot/jid of failed member */
1132 static void gdlm_recover_slot(void *arg, struct dlm_slot *slot)
1134 struct gfs2_sbd *sdp = arg;
1135 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1136 int jid = slot->slot - 1;
1138 if (gfs2_withdrawn(sdp)) {
1139 fs_err(sdp, "recover_slot jid %d ignored due to withdraw.\n",
1143 spin_lock(&ls->ls_recover_spin);
1144 if (ls->ls_recover_size < jid + 1) {
1145 fs_err(sdp, "recover_slot jid %d gen %u short size %d\n",
1146 jid, ls->ls_recover_block, ls->ls_recover_size);
1147 spin_unlock(&ls->ls_recover_spin);
1151 if (ls->ls_recover_submit[jid]) {
1152 fs_info(sdp, "recover_slot jid %d gen %u prev %u\n",
1153 jid, ls->ls_recover_block, ls->ls_recover_submit[jid]);
1155 ls->ls_recover_submit[jid] = ls->ls_recover_block;
1156 spin_unlock(&ls->ls_recover_spin);
1159 /* dlm calls after recover_slot and after it completes lock recovery */
1161 static void gdlm_recover_done(void *arg, struct dlm_slot *slots, int num_slots,
1162 int our_slot, uint32_t generation)
1164 struct gfs2_sbd *sdp = arg;
1165 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1167 if (gfs2_withdrawn(sdp)) {
1168 fs_err(sdp, "recover_done ignored due to withdraw.\n");
1171 /* ensure the ls jid arrays are large enough */
1172 set_recover_size(sdp, slots, num_slots);
1174 spin_lock(&ls->ls_recover_spin);
1175 ls->ls_recover_start = generation;
1177 if (!ls->ls_recover_mount) {
1178 ls->ls_recover_mount = generation;
1179 ls->ls_jid = our_slot - 1;
1182 if (!test_bit(DFL_UNMOUNT, &ls->ls_recover_flags))
1183 queue_delayed_work(gfs2_control_wq, &sdp->sd_control_work, 0);
1185 clear_bit(DFL_DLM_RECOVERY, &ls->ls_recover_flags);
1186 smp_mb__after_atomic();
1187 wake_up_bit(&ls->ls_recover_flags, DFL_DLM_RECOVERY);
1188 spin_unlock(&ls->ls_recover_spin);
1191 /* gfs2_recover thread has a journal recovery result */
1193 static void gdlm_recovery_result(struct gfs2_sbd *sdp, unsigned int jid,
1194 unsigned int result)
1196 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1198 if (gfs2_withdrawn(sdp)) {
1199 fs_err(sdp, "recovery_result jid %d ignored due to withdraw.\n",
1203 if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags))
1206 /* don't care about the recovery of own journal during mount */
1207 if (jid == ls->ls_jid)
1210 spin_lock(&ls->ls_recover_spin);
1211 if (test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
1212 spin_unlock(&ls->ls_recover_spin);
1215 if (ls->ls_recover_size < jid + 1) {
1216 fs_err(sdp, "recovery_result jid %d short size %d\n",
1217 jid, ls->ls_recover_size);
1218 spin_unlock(&ls->ls_recover_spin);
1222 fs_info(sdp, "recover jid %d result %s\n", jid,
1223 result == LM_RD_GAVEUP ? "busy" : "success");
1225 ls->ls_recover_result[jid] = result;
1227 /* GAVEUP means another node is recovering the journal; delay our
1228 next attempt to recover it, to give the other node a chance to
1229 finish before trying again */
1231 if (!test_bit(DFL_UNMOUNT, &ls->ls_recover_flags))
1232 queue_delayed_work(gfs2_control_wq, &sdp->sd_control_work,
1233 result == LM_RD_GAVEUP ? HZ : 0);
1234 spin_unlock(&ls->ls_recover_spin);
1237 static const struct dlm_lockspace_ops gdlm_lockspace_ops = {
1238 .recover_prep = gdlm_recover_prep,
1239 .recover_slot = gdlm_recover_slot,
1240 .recover_done = gdlm_recover_done,
1243 static int gdlm_mount(struct gfs2_sbd *sdp, const char *table)
1245 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1246 char cluster[GFS2_LOCKNAME_LEN];
1249 int error, ops_result;
1252 * initialize everything
1255 INIT_DELAYED_WORK(&sdp->sd_control_work, gfs2_control_func);
1256 spin_lock_init(&ls->ls_recover_spin);
1257 ls->ls_recover_flags = 0;
1258 ls->ls_recover_mount = 0;
1259 ls->ls_recover_start = 0;
1260 ls->ls_recover_block = 0;
1261 ls->ls_recover_size = 0;
1262 ls->ls_recover_submit = NULL;
1263 ls->ls_recover_result = NULL;
1264 ls->ls_lvb_bits = NULL;
1266 error = set_recover_size(sdp, NULL, 0);
1271 * prepare dlm_new_lockspace args
1274 fsname = strchr(table, ':');
1276 fs_info(sdp, "no fsname found\n");
1280 memset(cluster, 0, sizeof(cluster));
1281 memcpy(cluster, table, strlen(table) - strlen(fsname));
1284 flags = DLM_LSFL_FS | DLM_LSFL_NEWEXCL;
1287 * create/join lockspace
1290 error = dlm_new_lockspace(fsname, cluster, flags, GDLM_LVB_SIZE,
1291 &gdlm_lockspace_ops, sdp, &ops_result,
1294 fs_err(sdp, "dlm_new_lockspace error %d\n", error);
1298 if (ops_result < 0) {
1300 * dlm does not support ops callbacks,
1301 * old dlm_controld/gfs_controld are used, try without ops.
1303 fs_info(sdp, "dlm lockspace ops not used\n");
1304 free_recover_size(ls);
1305 set_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags);
1309 if (!test_bit(SDF_NOJOURNALID, &sdp->sd_flags)) {
1310 fs_err(sdp, "dlm lockspace ops disallow jid preset\n");
1316 * control_mount() uses control_lock to determine first mounter,
1317 * and for later mounts, waits for any recoveries to be cleared.
1320 error = control_mount(sdp);
1322 fs_err(sdp, "mount control error %d\n", error);
1326 ls->ls_first = !!test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags);
1327 clear_bit(SDF_NOJOURNALID, &sdp->sd_flags);
1328 smp_mb__after_atomic();
1329 wake_up_bit(&sdp->sd_flags, SDF_NOJOURNALID);
1333 dlm_release_lockspace(ls->ls_dlm, 2);
1335 free_recover_size(ls);
1340 static void gdlm_first_done(struct gfs2_sbd *sdp)
1342 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1345 if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags))
1348 error = control_first_done(sdp);
1350 fs_err(sdp, "mount first_done error %d\n", error);
1353 static void gdlm_unmount(struct gfs2_sbd *sdp)
1355 struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1357 if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags))
1360 /* wait for gfs2_control_wq to be done with this mount */
1362 spin_lock(&ls->ls_recover_spin);
1363 set_bit(DFL_UNMOUNT, &ls->ls_recover_flags);
1364 spin_unlock(&ls->ls_recover_spin);
1365 flush_delayed_work(&sdp->sd_control_work);
1367 /* mounted_lock and control_lock will be purged in dlm recovery */
1370 dlm_release_lockspace(ls->ls_dlm, 2);
1374 free_recover_size(ls);
1377 static const match_table_t dlm_tokens = {
1378 { Opt_jid, "jid=%d"},
1380 { Opt_first, "first=%d"},
1381 { Opt_nodir, "nodir=%d"},
1385 const struct lm_lockops gfs2_dlm_ops = {
1386 .lm_proto_name = "lock_dlm",
1387 .lm_mount = gdlm_mount,
1388 .lm_first_done = gdlm_first_done,
1389 .lm_recovery_result = gdlm_recovery_result,
1390 .lm_unmount = gdlm_unmount,
1391 .lm_put_lock = gdlm_put_lock,
1392 .lm_lock = gdlm_lock,
1393 .lm_cancel = gdlm_cancel,
1394 .lm_tokens = &dlm_tokens,