1// SPDX-License-Identifier: GPL-2.0-only
   2/*
   3 * Copyright (C) Sistina Software, Inc.  1997-2003 All rights reserved.
   4 * Copyright 2004-2011 Red Hat, Inc.
   5 */
   6
   7#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
   8
   9#include <linux/fs.h>
  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>
  16
  17#include "incore.h"
  18#include "glock.h"
  19#include "glops.h"
  20#include "recovery.h"
  21#include "util.h"
  22#include "sys.h"
  23#include "trace_gfs2.h"
  24
  25/**
  26 * gfs2_update_stats - Update time based stats
  27 * @s: The stats to update (local or global)
  28 * @index: The index inside @s
  29 * @sample: New data to include
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
  30 */
 
  31static inline void gfs2_update_stats(struct gfs2_lkstats *s, unsigned index,
  32				     s64 sample)
  33{
  34	/*
  35	 * @delta is the difference between the current rtt sample and the
  36	 * running average srtt. We add 1/8 of that to the srtt in order to
  37	 * update the current srtt estimate. The variance estimate is a bit
  38	 * more complicated. We subtract the current variance estimate from
  39	 * the abs value of the @delta and add 1/4 of that to the running
  40	 * total.  That's equivalent to 3/4 of the current variance
  41	 * estimate plus 1/4 of the abs of @delta.
  42	 *
  43	 * Note that the index points at the array entry containing the
  44	 * smoothed mean value, and the variance is always in the following
  45	 * entry
  46	 *
  47	 * Reference: TCP/IP Illustrated, vol 2, p. 831,832
  48	 * All times are in units of integer nanoseconds. Unlike the TCP/IP
  49	 * case, they are not scaled fixed point.
  50	 */
  51
  52	s64 delta = sample - s->stats[index];
  53	s->stats[index] += (delta >> 3);
  54	index++;
  55	s->stats[index] += (s64)(abs(delta) - s->stats[index]) >> 2;
  56}
  57
  58/**
  59 * gfs2_update_reply_times - Update locking statistics
  60 * @gl: The glock to update
  61 *
  62 * This assumes that gl->gl_dstamp has been set earlier.
  63 *
  64 * The rtt (lock round trip time) is an estimate of the time
  65 * taken to perform a dlm lock request. We update it on each
  66 * reply from the dlm.
  67 *
  68 * The blocking flag is set on the glock for all dlm requests
  69 * which may potentially block due to lock requests from other nodes.
  70 * DLM requests where the current lock state is exclusive, the
  71 * requested state is null (or unlocked) or where the TRY or
  72 * TRY_1CB flags are set are classified as non-blocking. All
  73 * other DLM requests are counted as (potentially) blocking.
  74 */
  75static inline void gfs2_update_reply_times(struct gfs2_glock *gl)
  76{
  77	struct gfs2_pcpu_lkstats *lks;
  78	const unsigned gltype = gl->gl_name.ln_type;
  79	unsigned index = test_bit(GLF_BLOCKING, &gl->gl_flags) ?
  80			 GFS2_LKS_SRTTB : GFS2_LKS_SRTT;
  81	s64 rtt;
  82
  83	preempt_disable();
  84	rtt = ktime_to_ns(ktime_sub(ktime_get_real(), gl->gl_dstamp));
  85	lks = this_cpu_ptr(gl->gl_name.ln_sbd->sd_lkstats);
  86	gfs2_update_stats(&gl->gl_stats, index, rtt);		/* Local */
  87	gfs2_update_stats(&lks->lkstats[gltype], index, rtt);	/* Global */
  88	preempt_enable();
  89
  90	trace_gfs2_glock_lock_time(gl, rtt);
  91}
  92
  93/**
  94 * gfs2_update_request_times - Update locking statistics
  95 * @gl: The glock to update
  96 *
  97 * The irt (lock inter-request times) measures the average time
  98 * between requests to the dlm. It is updated immediately before
  99 * each dlm call.
 100 */
 101
 102static inline void gfs2_update_request_times(struct gfs2_glock *gl)
 103{
 104	struct gfs2_pcpu_lkstats *lks;
 105	const unsigned gltype = gl->gl_name.ln_type;
 106	ktime_t dstamp;
 107	s64 irt;
 108
 109	preempt_disable();
 110	dstamp = gl->gl_dstamp;
 111	gl->gl_dstamp = ktime_get_real();
 112	irt = ktime_to_ns(ktime_sub(gl->gl_dstamp, dstamp));
 113	lks = this_cpu_ptr(gl->gl_name.ln_sbd->sd_lkstats);
 114	gfs2_update_stats(&gl->gl_stats, GFS2_LKS_SIRT, irt);		/* Local */
 115	gfs2_update_stats(&lks->lkstats[gltype], GFS2_LKS_SIRT, irt);	/* Global */
 116	preempt_enable();
 117}
 118 
 119static void gdlm_ast(void *arg)
 120{
 121	struct gfs2_glock *gl = arg;
 122	unsigned ret = gl->gl_state;
 123
 124	/* If the glock is dead, we only react to a dlm_unlock() reply. */
 125	if (__lockref_is_dead(&gl->gl_lockref) &&
 126	    gl->gl_lksb.sb_status != -DLM_EUNLOCK)
 127		return;
 128
 129	gfs2_update_reply_times(gl);
 130	BUG_ON(gl->gl_lksb.sb_flags & DLM_SBF_DEMOTED);
 131
 132	if ((gl->gl_lksb.sb_flags & DLM_SBF_VALNOTVALID) && gl->gl_lksb.sb_lvbptr)
 133		memset(gl->gl_lksb.sb_lvbptr, 0, GDLM_LVB_SIZE);
 134
 135	switch (gl->gl_lksb.sb_status) {
 136	case -DLM_EUNLOCK: /* Unlocked, so glock can be freed */
 137		if (gl->gl_ops->go_free)
 138			gl->gl_ops->go_free(gl);
 139		gfs2_glock_free(gl);
 140		return;
 141	case -DLM_ECANCEL: /* Cancel while getting lock */
 142		ret |= LM_OUT_CANCELED;
 143		goto out;
 144	case -EAGAIN: /* Try lock fails */
 145	case -EDEADLK: /* Deadlock detected */
 146		goto out;
 147	case -ETIMEDOUT: /* Canceled due to timeout */
 148		ret |= LM_OUT_ERROR;
 149		goto out;
 150	case 0: /* Success */
 151		break;
 152	default: /* Something unexpected */
 153		BUG();
 154	}
 155
 156	ret = gl->gl_req;
 157	if (gl->gl_lksb.sb_flags & DLM_SBF_ALTMODE) {
 158		if (gl->gl_req == LM_ST_SHARED)
 159			ret = LM_ST_DEFERRED;
 160		else if (gl->gl_req == LM_ST_DEFERRED)
 161			ret = LM_ST_SHARED;
 162		else
 163			BUG();
 164	}
 165
 166	set_bit(GLF_INITIAL, &gl->gl_flags);
 167	gfs2_glock_complete(gl, ret);
 168	return;
 169out:
 170	if (!test_bit(GLF_INITIAL, &gl->gl_flags))
 171		gl->gl_lksb.sb_lkid = 0;
 172	gfs2_glock_complete(gl, ret);
 173}
 174
 175static void gdlm_bast(void *arg, int mode)
 176{
 177	struct gfs2_glock *gl = arg;
 178
 179	if (__lockref_is_dead(&gl->gl_lockref))
 180		return;
 181
 182	switch (mode) {
 183	case DLM_LOCK_EX:
 184		gfs2_glock_cb(gl, LM_ST_UNLOCKED);
 185		break;
 186	case DLM_LOCK_CW:
 187		gfs2_glock_cb(gl, LM_ST_DEFERRED);
 188		break;
 189	case DLM_LOCK_PR:
 190		gfs2_glock_cb(gl, LM_ST_SHARED);
 191		break;
 192	default:
 193		fs_err(gl->gl_name.ln_sbd, "unknown bast mode %d\n", mode);
 194		BUG();
 195	}
 196}
 197
 198/* convert gfs lock-state to dlm lock-mode */
 199
 200static int make_mode(struct gfs2_sbd *sdp, const unsigned int lmstate)
 201{
 202	switch (lmstate) {
 203	case LM_ST_UNLOCKED:
 204		return DLM_LOCK_NL;
 205	case LM_ST_EXCLUSIVE:
 206		return DLM_LOCK_EX;
 207	case LM_ST_DEFERRED:
 208		return DLM_LOCK_CW;
 209	case LM_ST_SHARED:
 210		return DLM_LOCK_PR;
 211	}
 212	fs_err(sdp, "unknown LM state %d\n", lmstate);
 213	BUG();
 214	return -1;
 215}
 216
 217static u32 make_flags(struct gfs2_glock *gl, const unsigned int gfs_flags,
 218		      const int req)
 219{
 220	u32 lkf = 0;
 221
 222	if (gl->gl_lksb.sb_lvbptr)
 223		lkf |= DLM_LKF_VALBLK;
 224
 225	if (gfs_flags & LM_FLAG_TRY)
 226		lkf |= DLM_LKF_NOQUEUE;
 227
 228	if (gfs_flags & LM_FLAG_TRY_1CB) {
 229		lkf |= DLM_LKF_NOQUEUE;
 230		lkf |= DLM_LKF_NOQUEUEBAST;
 231	}
 232
 
 
 
 
 
 233	if (gfs_flags & LM_FLAG_ANY) {
 234		if (req == DLM_LOCK_PR)
 235			lkf |= DLM_LKF_ALTCW;
 236		else if (req == DLM_LOCK_CW)
 237			lkf |= DLM_LKF_ALTPR;
 238		else
 239			BUG();
 240	}
 241
 242	if (gl->gl_lksb.sb_lkid != 0) {
 243		lkf |= DLM_LKF_CONVERT;
 244		if (test_bit(GLF_BLOCKING, &gl->gl_flags))
 245			lkf |= DLM_LKF_QUECVT;
 246	}
 247
 248	return lkf;
 249}
 250
 251static void gfs2_reverse_hex(char *c, u64 value)
 252{
 253	*c = '0';
 254	while (value) {
 255		*c-- = hex_asc[value & 0x0f];
 256		value >>= 4;
 257	}
 258}
 259
 260static int gdlm_lock(struct gfs2_glock *gl, unsigned int req_state,
 261		     unsigned int flags)
 262{
 263	struct lm_lockstruct *ls = &gl->gl_name.ln_sbd->sd_lockstruct;
 264	int req;
 265	u32 lkf;
 266	char strname[GDLM_STRNAME_BYTES] = "";
 267	int error;
 268
 269	req = make_mode(gl->gl_name.ln_sbd, req_state);
 270	lkf = make_flags(gl, flags, req);
 271	gfs2_glstats_inc(gl, GFS2_LKS_DCOUNT);
 272	gfs2_sbstats_inc(gl, GFS2_LKS_DCOUNT);
 273	if (gl->gl_lksb.sb_lkid) {
 274		gfs2_update_request_times(gl);
 275	} else {
 276		memset(strname, ' ', GDLM_STRNAME_BYTES - 1);
 277		strname[GDLM_STRNAME_BYTES - 1] = '\0';
 278		gfs2_reverse_hex(strname + 7, gl->gl_name.ln_type);
 279		gfs2_reverse_hex(strname + 23, gl->gl_name.ln_number);
 280		gl->gl_dstamp = ktime_get_real();
 281	}
 282	/*
 283	 * Submit the actual lock request.
 284	 */
 285
 286again:
 287	error = dlm_lock(ls->ls_dlm, req, &gl->gl_lksb, lkf, strname,
 288			GDLM_STRNAME_BYTES - 1, 0, gdlm_ast, gl, gdlm_bast);
 289	if (error == -EBUSY) {
 290		msleep(20);
 291		goto again;
 292	}
 293	return error;
 294}
 295
 296static void gdlm_put_lock(struct gfs2_glock *gl)
 297{
 298	struct gfs2_sbd *sdp = gl->gl_name.ln_sbd;
 299	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 
 300	int error;
 301
 302	BUG_ON(!__lockref_is_dead(&gl->gl_lockref));
 303
 304	if (gl->gl_lksb.sb_lkid == 0) {
 305		gfs2_glock_free(gl);
 306		return;
 307	}
 308
 309	clear_bit(GLF_BLOCKING, &gl->gl_flags);
 310	gfs2_glstats_inc(gl, GFS2_LKS_DCOUNT);
 311	gfs2_sbstats_inc(gl, GFS2_LKS_DCOUNT);
 312	gfs2_update_request_times(gl);
 313
 314	/* don't want to call dlm if we've unmounted the lock protocol */
 315	if (test_bit(DFL_UNMOUNT, &ls->ls_recover_flags)) {
 316		gfs2_glock_free(gl);
 317		return;
 318	}
 319	/* don't want to skip dlm_unlock writing the lvb when lock has one */
 320
 321	if (test_bit(SDF_SKIP_DLM_UNLOCK, &sdp->sd_flags) &&
 322	    !gl->gl_lksb.sb_lvbptr) {
 323		gfs2_glock_free_later(gl);
 324		return;
 325	}
 326
 327again:
 328	error = dlm_unlock(ls->ls_dlm, gl->gl_lksb.sb_lkid, DLM_LKF_VALBLK,
 329			   NULL, gl);
 330	if (error == -EBUSY) {
 331		msleep(20);
 332		goto again;
 333	}
 334
 335	if (error) {
 336		fs_err(sdp, "gdlm_unlock %x,%llx err=%d\n",
 337		       gl->gl_name.ln_type,
 338		       (unsigned long long)gl->gl_name.ln_number, error);
 
 339	}
 340}
 341
 342static void gdlm_cancel(struct gfs2_glock *gl)
 343{
 344	struct lm_lockstruct *ls = &gl->gl_name.ln_sbd->sd_lockstruct;
 345	dlm_unlock(ls->ls_dlm, gl->gl_lksb.sb_lkid, DLM_LKF_CANCEL, NULL, gl);
 346}
 347
 348/*
 349 * dlm/gfs2 recovery coordination using dlm_recover callbacks
 350 *
 351 *  0. gfs2 checks for another cluster node withdraw, needing journal replay
 352 *  1. dlm_controld sees lockspace members change
 353 *  2. dlm_controld blocks dlm-kernel locking activity
 354 *  3. dlm_controld within dlm-kernel notifies gfs2 (recover_prep)
 355 *  4. dlm_controld starts and finishes its own user level recovery
 356 *  5. dlm_controld starts dlm-kernel dlm_recoverd to do kernel recovery
 357 *  6. dlm_recoverd notifies gfs2 of failed nodes (recover_slot)
 358 *  7. dlm_recoverd does its own lock recovery
 359 *  8. dlm_recoverd unblocks dlm-kernel locking activity
 360 *  9. dlm_recoverd notifies gfs2 when done (recover_done with new generation)
 361 * 10. gfs2_control updates control_lock lvb with new generation and jid bits
 362 * 11. gfs2_control enqueues journals for gfs2_recover to recover (maybe none)
 363 * 12. gfs2_recover dequeues and recovers journals of failed nodes
 364 * 13. gfs2_recover provides recovery results to gfs2_control (recovery_result)
 365 * 14. gfs2_control updates control_lock lvb jid bits for recovered journals
 366 * 15. gfs2_control unblocks normal locking when all journals are recovered
 367 *
 368 * - failures during recovery
 369 *
 370 * recover_prep() may set BLOCK_LOCKS (step 3) again before gfs2_control
 371 * clears BLOCK_LOCKS (step 15), e.g. another node fails while still
 372 * recovering for a prior failure.  gfs2_control needs a way to detect
 373 * this so it can leave BLOCK_LOCKS set in step 15.  This is managed using
 374 * the recover_block and recover_start values.
 375 *
 376 * recover_done() provides a new lockspace generation number each time it
 377 * is called (step 9).  This generation number is saved as recover_start.
 378 * When recover_prep() is called, it sets BLOCK_LOCKS and sets
 379 * recover_block = recover_start.  So, while recover_block is equal to
 380 * recover_start, BLOCK_LOCKS should remain set.  (recover_spin must
 381 * be held around the BLOCK_LOCKS/recover_block/recover_start logic.)
 382 *
 383 * - more specific gfs2 steps in sequence above
 384 *
 385 *  3. recover_prep sets BLOCK_LOCKS and sets recover_block = recover_start
 386 *  6. recover_slot records any failed jids (maybe none)
 387 *  9. recover_done sets recover_start = new generation number
 388 * 10. gfs2_control sets control_lock lvb = new gen + bits for failed jids
 389 * 12. gfs2_recover does journal recoveries for failed jids identified above
 390 * 14. gfs2_control clears control_lock lvb bits for recovered jids
 391 * 15. gfs2_control checks if recover_block == recover_start (step 3 occured
 392 *     again) then do nothing, otherwise if recover_start > recover_block
 393 *     then clear BLOCK_LOCKS.
 394 *
 395 * - parallel recovery steps across all nodes
 396 *
 397 * All nodes attempt to update the control_lock lvb with the new generation
 398 * number and jid bits, but only the first to get the control_lock EX will
 399 * do so; others will see that it's already done (lvb already contains new
 400 * generation number.)
 401 *
 402 * . All nodes get the same recover_prep/recover_slot/recover_done callbacks
 403 * . All nodes attempt to set control_lock lvb gen + bits for the new gen
 404 * . One node gets control_lock first and writes the lvb, others see it's done
 405 * . All nodes attempt to recover jids for which they see control_lock bits set
 406 * . One node succeeds for a jid, and that one clears the jid bit in the lvb
 407 * . All nodes will eventually see all lvb bits clear and unblock locks
 408 *
 409 * - is there a problem with clearing an lvb bit that should be set
 410 *   and missing a journal recovery?
 411 *
 412 * 1. jid fails
 413 * 2. lvb bit set for step 1
 414 * 3. jid recovered for step 1
 415 * 4. jid taken again (new mount)
 416 * 5. jid fails (for step 4)
 417 * 6. lvb bit set for step 5 (will already be set)
 418 * 7. lvb bit cleared for step 3
 419 *
 420 * This is not a problem because the failure in step 5 does not
 421 * require recovery, because the mount in step 4 could not have
 422 * progressed far enough to unblock locks and access the fs.  The
 423 * control_mount() function waits for all recoveries to be complete
 424 * for the latest lockspace generation before ever unblocking locks
 425 * and returning.  The mount in step 4 waits until the recovery in
 426 * step 1 is done.
 427 *
 428 * - special case of first mounter: first node to mount the fs
 429 *
 430 * The first node to mount a gfs2 fs needs to check all the journals
 431 * and recover any that need recovery before other nodes are allowed
 432 * to mount the fs.  (Others may begin mounting, but they must wait
 433 * for the first mounter to be done before taking locks on the fs
 434 * or accessing the fs.)  This has two parts:
 435 *
 436 * 1. The mounted_lock tells a node it's the first to mount the fs.
 437 * Each node holds the mounted_lock in PR while it's mounted.
 438 * Each node tries to acquire the mounted_lock in EX when it mounts.
 439 * If a node is granted the mounted_lock EX it means there are no
 440 * other mounted nodes (no PR locks exist), and it is the first mounter.
 441 * The mounted_lock is demoted to PR when first recovery is done, so
 442 * others will fail to get an EX lock, but will get a PR lock.
 443 *
 444 * 2. The control_lock blocks others in control_mount() while the first
 445 * mounter is doing first mount recovery of all journals.
 446 * A mounting node needs to acquire control_lock in EX mode before
 447 * it can proceed.  The first mounter holds control_lock in EX while doing
 448 * the first mount recovery, blocking mounts from other nodes, then demotes
 449 * control_lock to NL when it's done (others_may_mount/first_done),
 450 * allowing other nodes to continue mounting.
 451 *
 452 * first mounter:
 453 * control_lock EX/NOQUEUE success
 454 * mounted_lock EX/NOQUEUE success (no other PR, so no other mounters)
 455 * set first=1
 456 * do first mounter recovery
 457 * mounted_lock EX->PR
 458 * control_lock EX->NL, write lvb generation
 459 *
 460 * other mounter:
 461 * control_lock EX/NOQUEUE success (if fail -EAGAIN, retry)
 462 * mounted_lock EX/NOQUEUE fail -EAGAIN (expected due to other mounters PR)
 463 * mounted_lock PR/NOQUEUE success
 464 * read lvb generation
 465 * control_lock EX->NL
 466 * set first=0
 467 *
 468 * - mount during recovery
 469 *
 470 * If a node mounts while others are doing recovery (not first mounter),
 471 * the mounting node will get its initial recover_done() callback without
 472 * having seen any previous failures/callbacks.
 473 *
 474 * It must wait for all recoveries preceding its mount to be finished
 475 * before it unblocks locks.  It does this by repeating the "other mounter"
 476 * steps above until the lvb generation number is >= its mount generation
 477 * number (from initial recover_done) and all lvb bits are clear.
 478 *
 479 * - control_lock lvb format
 480 *
 481 * 4 bytes generation number: the latest dlm lockspace generation number
 482 * from recover_done callback.  Indicates the jid bitmap has been updated
 483 * to reflect all slot failures through that generation.
 484 * 4 bytes unused.
 485 * GDLM_LVB_SIZE-8 bytes of jid bit map. If bit N is set, it indicates
 486 * that jid N needs recovery.
 487 */
 488
 489#define JID_BITMAP_OFFSET 8 /* 4 byte generation number + 4 byte unused */
 490
 491static void control_lvb_read(struct lm_lockstruct *ls, uint32_t *lvb_gen,
 492			     char *lvb_bits)
 493{
 494	__le32 gen;
 495	memcpy(lvb_bits, ls->ls_control_lvb, GDLM_LVB_SIZE);
 496	memcpy(&gen, lvb_bits, sizeof(__le32));
 497	*lvb_gen = le32_to_cpu(gen);
 498}
 499
 500static void control_lvb_write(struct lm_lockstruct *ls, uint32_t lvb_gen,
 501			      char *lvb_bits)
 502{
 503	__le32 gen;
 504	memcpy(ls->ls_control_lvb, lvb_bits, GDLM_LVB_SIZE);
 505	gen = cpu_to_le32(lvb_gen);
 506	memcpy(ls->ls_control_lvb, &gen, sizeof(__le32));
 507}
 508
 509static int all_jid_bits_clear(char *lvb)
 510{
 511	return !memchr_inv(lvb + JID_BITMAP_OFFSET, 0,
 512			GDLM_LVB_SIZE - JID_BITMAP_OFFSET);
 513}
 514
 515static void sync_wait_cb(void *arg)
 516{
 517	struct lm_lockstruct *ls = arg;
 518	complete(&ls->ls_sync_wait);
 519}
 520
 521static int sync_unlock(struct gfs2_sbd *sdp, struct dlm_lksb *lksb, char *name)
 522{
 523	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 524	int error;
 525
 526	error = dlm_unlock(ls->ls_dlm, lksb->sb_lkid, 0, lksb, ls);
 527	if (error) {
 528		fs_err(sdp, "%s lkid %x error %d\n",
 529		       name, lksb->sb_lkid, error);
 530		return error;
 531	}
 532
 533	wait_for_completion(&ls->ls_sync_wait);
 534
 535	if (lksb->sb_status != -DLM_EUNLOCK) {
 536		fs_err(sdp, "%s lkid %x status %d\n",
 537		       name, lksb->sb_lkid, lksb->sb_status);
 538		return -1;
 539	}
 540	return 0;
 541}
 542
 543static int sync_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags,
 544		     unsigned int num, struct dlm_lksb *lksb, char *name)
 545{
 546	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 547	char strname[GDLM_STRNAME_BYTES];
 548	int error, status;
 549
 550	memset(strname, 0, GDLM_STRNAME_BYTES);
 551	snprintf(strname, GDLM_STRNAME_BYTES, "%8x%16x", LM_TYPE_NONDISK, num);
 552
 553	error = dlm_lock(ls->ls_dlm, mode, lksb, flags,
 554			 strname, GDLM_STRNAME_BYTES - 1,
 555			 0, sync_wait_cb, ls, NULL);
 556	if (error) {
 557		fs_err(sdp, "%s lkid %x flags %x mode %d error %d\n",
 558		       name, lksb->sb_lkid, flags, mode, error);
 559		return error;
 560	}
 561
 562	wait_for_completion(&ls->ls_sync_wait);
 563
 564	status = lksb->sb_status;
 565
 566	if (status && status != -EAGAIN) {
 567		fs_err(sdp, "%s lkid %x flags %x mode %d status %d\n",
 568		       name, lksb->sb_lkid, flags, mode, status);
 569	}
 570
 571	return status;
 572}
 573
 574static int mounted_unlock(struct gfs2_sbd *sdp)
 575{
 576	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 577	return sync_unlock(sdp, &ls->ls_mounted_lksb, "mounted_lock");
 578}
 579
 580static int mounted_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags)
 581{
 582	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 583	return sync_lock(sdp, mode, flags, GFS2_MOUNTED_LOCK,
 584			 &ls->ls_mounted_lksb, "mounted_lock");
 585}
 586
 587static int control_unlock(struct gfs2_sbd *sdp)
 588{
 589	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 590	return sync_unlock(sdp, &ls->ls_control_lksb, "control_lock");
 591}
 592
 593static int control_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags)
 594{
 595	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 596	return sync_lock(sdp, mode, flags, GFS2_CONTROL_LOCK,
 597			 &ls->ls_control_lksb, "control_lock");
 598}
 599
 600/**
 601 * remote_withdraw - react to a node withdrawing from the file system
 602 * @sdp: The superblock
 603 */
 604static void remote_withdraw(struct gfs2_sbd *sdp)
 605{
 606	struct gfs2_jdesc *jd;
 607	int ret = 0, count = 0;
 608
 609	list_for_each_entry(jd, &sdp->sd_jindex_list, jd_list) {
 610		if (jd->jd_jid == sdp->sd_lockstruct.ls_jid)
 611			continue;
 612		ret = gfs2_recover_journal(jd, true);
 613		if (ret)
 614			break;
 615		count++;
 616	}
 617
 618	/* Now drop the additional reference we acquired */
 619	fs_err(sdp, "Journals checked: %d, ret = %d.\n", count, ret);
 620}
 621
 622static void gfs2_control_func(struct work_struct *work)
 623{
 624	struct gfs2_sbd *sdp = container_of(work, struct gfs2_sbd, sd_control_work.work);
 625	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 626	uint32_t block_gen, start_gen, lvb_gen, flags;
 627	int recover_set = 0;
 628	int write_lvb = 0;
 629	int recover_size;
 630	int i, error;
 631
 632	/* First check for other nodes that may have done a withdraw. */
 633	if (test_bit(SDF_REMOTE_WITHDRAW, &sdp->sd_flags)) {
 634		remote_withdraw(sdp);
 635		clear_bit(SDF_REMOTE_WITHDRAW, &sdp->sd_flags);
 636		return;
 637	}
 638
 639	spin_lock(&ls->ls_recover_spin);
 640	/*
 641	 * No MOUNT_DONE means we're still mounting; control_mount()
 642	 * will set this flag, after which this thread will take over
 643	 * all further clearing of BLOCK_LOCKS.
 644	 *
 645	 * FIRST_MOUNT means this node is doing first mounter recovery,
 646	 * for which recovery control is handled by
 647	 * control_mount()/control_first_done(), not this thread.
 648	 */
 649	if (!test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) ||
 650	     test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
 651		spin_unlock(&ls->ls_recover_spin);
 652		return;
 653	}
 654	block_gen = ls->ls_recover_block;
 655	start_gen = ls->ls_recover_start;
 656	spin_unlock(&ls->ls_recover_spin);
 657
 658	/*
 659	 * Equal block_gen and start_gen implies we are between
 660	 * recover_prep and recover_done callbacks, which means
 661	 * dlm recovery is in progress and dlm locking is blocked.
 662	 * There's no point trying to do any work until recover_done.
 663	 */
 664
 665	if (block_gen == start_gen)
 666		return;
 667
 668	/*
 669	 * Propagate recover_submit[] and recover_result[] to lvb:
 670	 * dlm_recoverd adds to recover_submit[] jids needing recovery
 671	 * gfs2_recover adds to recover_result[] journal recovery results
 672	 *
 673	 * set lvb bit for jids in recover_submit[] if the lvb has not
 674	 * yet been updated for the generation of the failure
 675	 *
 676	 * clear lvb bit for jids in recover_result[] if the result of
 677	 * the journal recovery is SUCCESS
 678	 */
 679
 680	error = control_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_VALBLK);
 681	if (error) {
 682		fs_err(sdp, "control lock EX error %d\n", error);
 683		return;
 684	}
 685
 686	control_lvb_read(ls, &lvb_gen, ls->ls_lvb_bits);
 687
 688	spin_lock(&ls->ls_recover_spin);
 689	if (block_gen != ls->ls_recover_block ||
 690	    start_gen != ls->ls_recover_start) {
 691		fs_info(sdp, "recover generation %u block1 %u %u\n",
 692			start_gen, block_gen, ls->ls_recover_block);
 693		spin_unlock(&ls->ls_recover_spin);
 694		control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT);
 695		return;
 696	}
 697
 698	recover_size = ls->ls_recover_size;
 699
 700	if (lvb_gen <= start_gen) {
 701		/*
 702		 * Clear lvb bits for jids we've successfully recovered.
 703		 * Because all nodes attempt to recover failed journals,
 704		 * a journal can be recovered multiple times successfully
 705		 * in succession.  Only the first will really do recovery,
 706		 * the others find it clean, but still report a successful
 707		 * recovery.  So, another node may have already recovered
 708		 * the jid and cleared the lvb bit for it.
 709		 */
 710		for (i = 0; i < recover_size; i++) {
 711			if (ls->ls_recover_result[i] != LM_RD_SUCCESS)
 712				continue;
 713
 714			ls->ls_recover_result[i] = 0;
 715
 716			if (!test_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET))
 717				continue;
 718
 719			__clear_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET);
 720			write_lvb = 1;
 721		}
 722	}
 723
 724	if (lvb_gen == start_gen) {
 725		/*
 726		 * Failed slots before start_gen are already set in lvb.
 727		 */
 728		for (i = 0; i < recover_size; i++) {
 729			if (!ls->ls_recover_submit[i])
 730				continue;
 731			if (ls->ls_recover_submit[i] < lvb_gen)
 732				ls->ls_recover_submit[i] = 0;
 733		}
 734	} else if (lvb_gen < start_gen) {
 735		/*
 736		 * Failed slots before start_gen are not yet set in lvb.
 737		 */
 738		for (i = 0; i < recover_size; i++) {
 739			if (!ls->ls_recover_submit[i])
 740				continue;
 741			if (ls->ls_recover_submit[i] < start_gen) {
 742				ls->ls_recover_submit[i] = 0;
 743				__set_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET);
 744			}
 745		}
 746		/* even if there are no bits to set, we need to write the
 747		   latest generation to the lvb */
 748		write_lvb = 1;
 749	} else {
 750		/*
 751		 * we should be getting a recover_done() for lvb_gen soon
 752		 */
 753	}
 754	spin_unlock(&ls->ls_recover_spin);
 755
 756	if (write_lvb) {
 757		control_lvb_write(ls, start_gen, ls->ls_lvb_bits);
 758		flags = DLM_LKF_CONVERT | DLM_LKF_VALBLK;
 759	} else {
 760		flags = DLM_LKF_CONVERT;
 761	}
 762
 763	error = control_lock(sdp, DLM_LOCK_NL, flags);
 764	if (error) {
 765		fs_err(sdp, "control lock NL error %d\n", error);
 766		return;
 767	}
 768
 769	/*
 770	 * Everyone will see jid bits set in the lvb, run gfs2_recover_set(),
 771	 * and clear a jid bit in the lvb if the recovery is a success.
 772	 * Eventually all journals will be recovered, all jid bits will
 773	 * be cleared in the lvb, and everyone will clear BLOCK_LOCKS.
 774	 */
 775
 776	for (i = 0; i < recover_size; i++) {
 777		if (test_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET)) {
 778			fs_info(sdp, "recover generation %u jid %d\n",
 779				start_gen, i);
 780			gfs2_recover_set(sdp, i);
 781			recover_set++;
 782		}
 783	}
 784	if (recover_set)
 785		return;
 786
 787	/*
 788	 * No more jid bits set in lvb, all recovery is done, unblock locks
 789	 * (unless a new recover_prep callback has occured blocking locks
 790	 * again while working above)
 791	 */
 792
 793	spin_lock(&ls->ls_recover_spin);
 794	if (ls->ls_recover_block == block_gen &&
 795	    ls->ls_recover_start == start_gen) {
 796		clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
 797		spin_unlock(&ls->ls_recover_spin);
 798		fs_info(sdp, "recover generation %u done\n", start_gen);
 799		gfs2_glock_thaw(sdp);
 800	} else {
 801		fs_info(sdp, "recover generation %u block2 %u %u\n",
 802			start_gen, block_gen, ls->ls_recover_block);
 803		spin_unlock(&ls->ls_recover_spin);
 804	}
 805}
 806
 807static int control_mount(struct gfs2_sbd *sdp)
 808{
 809	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
 810	uint32_t start_gen, block_gen, mount_gen, lvb_gen;
 811	int mounted_mode;
 812	int retries = 0;
 813	int error;
 814
 815	memset(&ls->ls_mounted_lksb, 0, sizeof(struct dlm_lksb));
 816	memset(&ls->ls_control_lksb, 0, sizeof(struct dlm_lksb));
 817	memset(&ls->ls_control_lvb, 0, GDLM_LVB_SIZE);
 818	ls->ls_control_lksb.sb_lvbptr = ls->ls_control_lvb;
 819	init_completion(&ls->ls_sync_wait);
 820
 821	set_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
 822
 823	error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_VALBLK);
 824	if (error) {
 825		fs_err(sdp, "control_mount control_lock NL error %d\n", error);
 826		return error;
 827	}
 828
 829	error = mounted_lock(sdp, DLM_LOCK_NL, 0);
 830	if (error) {
 831		fs_err(sdp, "control_mount mounted_lock NL error %d\n", error);
 832		control_unlock(sdp);
 833		return error;
 834	}
 835	mounted_mode = DLM_LOCK_NL;
 836
 837restart:
 838	if (retries++ && signal_pending(current)) {
 839		error = -EINTR;
 840		goto fail;
 841	}
 842
 843	/*
 844	 * We always start with both locks in NL. control_lock is
 845	 * demoted to NL below so we don't need to do it here.
 846	 */
 847
 848	if (mounted_mode != DLM_LOCK_NL) {
 849		error = mounted_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT);
 850		if (error)
 851			goto fail;
 852		mounted_mode = DLM_LOCK_NL;
 853	}
 854
 855	/*
 856	 * Other nodes need to do some work in dlm recovery and gfs2_control
 857	 * before the recover_done and control_lock will be ready for us below.
 858	 * A delay here is not required but often avoids having to retry.
 859	 */
 860
 861	msleep_interruptible(500);
 862
 863	/*
 864	 * Acquire control_lock in EX and mounted_lock in either EX or PR.
 865	 * control_lock lvb keeps track of any pending journal recoveries.
 866	 * mounted_lock indicates if any other nodes have the fs mounted.
 867	 */
 868
 869	error = control_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE|DLM_LKF_VALBLK);
 870	if (error == -EAGAIN) {
 871		goto restart;
 872	} else if (error) {
 873		fs_err(sdp, "control_mount control_lock EX error %d\n", error);
 874		goto fail;
 875	}
 876
 877	/**
 878	 * If we're a spectator, we don't want to take the lock in EX because
 879	 * we cannot do the first-mount responsibility it implies: recovery.
 880	 */
 881	if (sdp->sd_args.ar_spectator)
 882		goto locks_done;
 883
 884	error = mounted_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE);
 885	if (!error) {
 886		mounted_mode = DLM_LOCK_EX;
 887		goto locks_done;
 888	} else if (error != -EAGAIN) {
 889		fs_err(sdp, "control_mount mounted_lock EX error %d\n", error);
 890		goto fail;
 891	}
 892
 893	error = mounted_lock(sdp, DLM_LOCK_PR, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE);
 894	if (!error) {
 895		mounted_mode = DLM_LOCK_PR;
 896		goto locks_done;
 897	} else {
 898		/* not even -EAGAIN should happen here */
 899		fs_err(sdp, "control_mount mounted_lock PR error %d\n", error);
 900		goto fail;
 901	}
 902
 903locks_done:
 904	/*
 905	 * If we got both locks above in EX, then we're the first mounter.
 906	 * If not, then we need to wait for the control_lock lvb to be
 907	 * updated by other mounted nodes to reflect our mount generation.
 908	 *
 909	 * In simple first mounter cases, first mounter will see zero lvb_gen,
 910	 * but in cases where all existing nodes leave/fail before mounting
 911	 * nodes finish control_mount, then all nodes will be mounting and
 912	 * lvb_gen will be non-zero.
 913	 */
 914
 915	control_lvb_read(ls, &lvb_gen, ls->ls_lvb_bits);
 916
 917	if (lvb_gen == 0xFFFFFFFF) {
 918		/* special value to force mount attempts to fail */
 919		fs_err(sdp, "control_mount control_lock disabled\n");
 920		error = -EINVAL;
 921		goto fail;
 922	}
 923
 924	if (mounted_mode == DLM_LOCK_EX) {
 925		/* first mounter, keep both EX while doing first recovery */
 926		spin_lock(&ls->ls_recover_spin);
 927		clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
 928		set_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags);
 929		set_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags);
 930		spin_unlock(&ls->ls_recover_spin);
 931		fs_info(sdp, "first mounter control generation %u\n", lvb_gen);
 932		return 0;
 933	}
 934
 935	error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT);
 936	if (error)
 937		goto fail;
 938
 939	/*
 940	 * We are not first mounter, now we need to wait for the control_lock
 941	 * lvb generation to be >= the generation from our first recover_done
 942	 * and all lvb bits to be clear (no pending journal recoveries.)
 943	 */
 944
 945	if (!all_jid_bits_clear(ls->ls_lvb_bits)) {
 946		/* journals need recovery, wait until all are clear */
 947		fs_info(sdp, "control_mount wait for journal recovery\n");
 948		goto restart;
 949	}
 950
 951	spin_lock(&ls->ls_recover_spin);
 952	block_gen = ls->ls_recover_block;
 953	start_gen = ls->ls_recover_start;
 954	mount_gen = ls->ls_recover_mount;
 955
 956	if (lvb_gen < mount_gen) {
 957		/* wait for mounted nodes to update control_lock lvb to our
 958		   generation, which might include new recovery bits set */
 959		if (sdp->sd_args.ar_spectator) {
 960			fs_info(sdp, "Recovery is required. Waiting for a "
 961				"non-spectator to mount.\n");
 962			msleep_interruptible(1000);
 963		} else {
 964			fs_info(sdp, "control_mount wait1 block %u start %u "
 965				"mount %u lvb %u flags %lx\n", block_gen,
 966				start_gen, mount_gen, lvb_gen,
 967				ls->ls_recover_flags);
 968		}
 969		spin_unlock(&ls->ls_recover_spin);
 970		goto restart;
 971	}
 972
 973	if (lvb_gen != start_gen) {
 974		/* wait for mounted nodes to update control_lock lvb to the
 975		   latest recovery generation */
 976		fs_info(sdp, "control_mount wait2 block %u start %u mount %u "
 977			"lvb %u flags %lx\n", block_gen, start_gen, mount_gen,
 978			lvb_gen, ls->ls_recover_flags);
 979		spin_unlock(&ls->ls_recover_spin);
 980		goto restart;
 981	}
 982
 983	if (block_gen == start_gen) {
 984		/* dlm recovery in progress, wait for it to finish */
 985		fs_info(sdp, "control_mount wait3 block %u start %u mount %u "
 986			"lvb %u flags %lx\n", block_gen, start_gen, mount_gen,
 987			lvb_gen, ls->ls_recover_flags);
 988		spin_unlock(&ls->ls_recover_spin);
 989		goto restart;
 990	}
 991
 992	clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
 993	set_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags);
 994	memset(ls->ls_recover_submit, 0, ls->ls_recover_size*sizeof(uint32_t));
 995	memset(ls->ls_recover_result, 0, ls->ls_recover_size*sizeof(uint32_t));
 996	spin_unlock(&ls->ls_recover_spin);
 997	return 0;
 998
 999fail:
1000	mounted_unlock(sdp);
1001	control_unlock(sdp);
1002	return error;
1003}
1004
1005static int control_first_done(struct gfs2_sbd *sdp)
1006{
1007	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1008	uint32_t start_gen, block_gen;
1009	int error;
1010
1011restart:
1012	spin_lock(&ls->ls_recover_spin);
1013	start_gen = ls->ls_recover_start;
1014	block_gen = ls->ls_recover_block;
1015
1016	if (test_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags) ||
1017	    !test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) ||
1018	    !test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
1019		/* sanity check, should not happen */
1020		fs_err(sdp, "control_first_done start %u block %u flags %lx\n",
1021		       start_gen, block_gen, ls->ls_recover_flags);
1022		spin_unlock(&ls->ls_recover_spin);
1023		control_unlock(sdp);
1024		return -1;
1025	}
1026
1027	if (start_gen == block_gen) {
1028		/*
1029		 * Wait for the end of a dlm recovery cycle to switch from
1030		 * first mounter recovery.  We can ignore any recover_slot
1031		 * callbacks between the recover_prep and next recover_done
1032		 * because we are still the first mounter and any failed nodes
1033		 * have not fully mounted, so they don't need recovery.
1034		 */
1035		spin_unlock(&ls->ls_recover_spin);
1036		fs_info(sdp, "control_first_done wait gen %u\n", start_gen);
1037
1038		wait_on_bit(&ls->ls_recover_flags, DFL_DLM_RECOVERY,
1039			    TASK_UNINTERRUPTIBLE);
1040		goto restart;
1041	}
1042
1043	clear_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags);
1044	set_bit(DFL_FIRST_MOUNT_DONE, &ls->ls_recover_flags);
1045	memset(ls->ls_recover_submit, 0, ls->ls_recover_size*sizeof(uint32_t));
1046	memset(ls->ls_recover_result, 0, ls->ls_recover_size*sizeof(uint32_t));
1047	spin_unlock(&ls->ls_recover_spin);
1048
1049	memset(ls->ls_lvb_bits, 0, GDLM_LVB_SIZE);
1050	control_lvb_write(ls, start_gen, ls->ls_lvb_bits);
1051
1052	error = mounted_lock(sdp, DLM_LOCK_PR, DLM_LKF_CONVERT);
1053	if (error)
1054		fs_err(sdp, "control_first_done mounted PR error %d\n", error);
1055
1056	error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT|DLM_LKF_VALBLK);
1057	if (error)
1058		fs_err(sdp, "control_first_done control NL error %d\n", error);
1059
1060	return error;
1061}
1062
1063/*
1064 * Expand static jid arrays if necessary (by increments of RECOVER_SIZE_INC)
1065 * to accommodate the largest slot number.  (NB dlm slot numbers start at 1,
1066 * gfs2 jids start at 0, so jid = slot - 1)
1067 */
1068
1069#define RECOVER_SIZE_INC 16
1070
1071static int set_recover_size(struct gfs2_sbd *sdp, struct dlm_slot *slots,
1072			    int num_slots)
1073{
1074	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1075	uint32_t *submit = NULL;
1076	uint32_t *result = NULL;
1077	uint32_t old_size, new_size;
1078	int i, max_jid;
1079
1080	if (!ls->ls_lvb_bits) {
1081		ls->ls_lvb_bits = kzalloc(GDLM_LVB_SIZE, GFP_NOFS);
1082		if (!ls->ls_lvb_bits)
1083			return -ENOMEM;
1084	}
1085
1086	max_jid = 0;
1087	for (i = 0; i < num_slots; i++) {
1088		if (max_jid < slots[i].slot - 1)
1089			max_jid = slots[i].slot - 1;
1090	}
1091
1092	old_size = ls->ls_recover_size;
1093	new_size = old_size;
1094	while (new_size < max_jid + 1)
1095		new_size += RECOVER_SIZE_INC;
1096	if (new_size == old_size)
1097		return 0;
1098
1099	submit = kcalloc(new_size, sizeof(uint32_t), GFP_NOFS);
1100	result = kcalloc(new_size, sizeof(uint32_t), GFP_NOFS);
1101	if (!submit || !result) {
1102		kfree(submit);
1103		kfree(result);
1104		return -ENOMEM;
1105	}
1106
1107	spin_lock(&ls->ls_recover_spin);
1108	memcpy(submit, ls->ls_recover_submit, old_size * sizeof(uint32_t));
1109	memcpy(result, ls->ls_recover_result, old_size * sizeof(uint32_t));
1110	kfree(ls->ls_recover_submit);
1111	kfree(ls->ls_recover_result);
1112	ls->ls_recover_submit = submit;
1113	ls->ls_recover_result = result;
1114	ls->ls_recover_size = new_size;
1115	spin_unlock(&ls->ls_recover_spin);
1116	return 0;
1117}
1118
1119static void free_recover_size(struct lm_lockstruct *ls)
1120{
1121	kfree(ls->ls_lvb_bits);
1122	kfree(ls->ls_recover_submit);
1123	kfree(ls->ls_recover_result);
1124	ls->ls_recover_submit = NULL;
1125	ls->ls_recover_result = NULL;
1126	ls->ls_recover_size = 0;
1127	ls->ls_lvb_bits = NULL;
1128}
1129
1130/* dlm calls before it does lock recovery */
1131
1132static void gdlm_recover_prep(void *arg)
1133{
1134	struct gfs2_sbd *sdp = arg;
1135	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1136
1137	if (gfs2_withdrawing_or_withdrawn(sdp)) {
1138		fs_err(sdp, "recover_prep ignored due to withdraw.\n");
1139		return;
1140	}
1141	spin_lock(&ls->ls_recover_spin);
1142	ls->ls_recover_block = ls->ls_recover_start;
1143	set_bit(DFL_DLM_RECOVERY, &ls->ls_recover_flags);
1144
1145	if (!test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) ||
1146	     test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
1147		spin_unlock(&ls->ls_recover_spin);
1148		return;
1149	}
1150	set_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags);
1151	spin_unlock(&ls->ls_recover_spin);
1152}
1153
1154/* dlm calls after recover_prep has been completed on all lockspace members;
1155   identifies slot/jid of failed member */
1156
1157static void gdlm_recover_slot(void *arg, struct dlm_slot *slot)
1158{
1159	struct gfs2_sbd *sdp = arg;
1160	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1161	int jid = slot->slot - 1;
1162
1163	if (gfs2_withdrawing_or_withdrawn(sdp)) {
1164		fs_err(sdp, "recover_slot jid %d ignored due to withdraw.\n",
1165		       jid);
1166		return;
1167	}
1168	spin_lock(&ls->ls_recover_spin);
1169	if (ls->ls_recover_size < jid + 1) {
1170		fs_err(sdp, "recover_slot jid %d gen %u short size %d\n",
1171		       jid, ls->ls_recover_block, ls->ls_recover_size);
1172		spin_unlock(&ls->ls_recover_spin);
1173		return;
1174	}
1175
1176	if (ls->ls_recover_submit[jid]) {
1177		fs_info(sdp, "recover_slot jid %d gen %u prev %u\n",
1178			jid, ls->ls_recover_block, ls->ls_recover_submit[jid]);
1179	}
1180	ls->ls_recover_submit[jid] = ls->ls_recover_block;
1181	spin_unlock(&ls->ls_recover_spin);
1182}
1183
1184/* dlm calls after recover_slot and after it completes lock recovery */
1185
1186static void gdlm_recover_done(void *arg, struct dlm_slot *slots, int num_slots,
1187			      int our_slot, uint32_t generation)
1188{
1189	struct gfs2_sbd *sdp = arg;
1190	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1191
1192	if (gfs2_withdrawing_or_withdrawn(sdp)) {
1193		fs_err(sdp, "recover_done ignored due to withdraw.\n");
1194		return;
1195	}
1196	/* ensure the ls jid arrays are large enough */
1197	set_recover_size(sdp, slots, num_slots);
1198
1199	spin_lock(&ls->ls_recover_spin);
1200	ls->ls_recover_start = generation;
1201
1202	if (!ls->ls_recover_mount) {
1203		ls->ls_recover_mount = generation;
1204		ls->ls_jid = our_slot - 1;
1205	}
1206
1207	if (!test_bit(DFL_UNMOUNT, &ls->ls_recover_flags))
1208		queue_delayed_work(gfs2_control_wq, &sdp->sd_control_work, 0);
1209
1210	clear_bit(DFL_DLM_RECOVERY, &ls->ls_recover_flags);
1211	smp_mb__after_atomic();
1212	wake_up_bit(&ls->ls_recover_flags, DFL_DLM_RECOVERY);
1213	spin_unlock(&ls->ls_recover_spin);
1214}
1215
1216/* gfs2_recover thread has a journal recovery result */
1217
1218static void gdlm_recovery_result(struct gfs2_sbd *sdp, unsigned int jid,
1219				 unsigned int result)
1220{
1221	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1222
1223	if (gfs2_withdrawing_or_withdrawn(sdp)) {
1224		fs_err(sdp, "recovery_result jid %d ignored due to withdraw.\n",
1225		       jid);
1226		return;
1227	}
1228	if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags))
1229		return;
1230
1231	/* don't care about the recovery of own journal during mount */
1232	if (jid == ls->ls_jid)
1233		return;
1234
1235	spin_lock(&ls->ls_recover_spin);
1236	if (test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) {
1237		spin_unlock(&ls->ls_recover_spin);
1238		return;
1239	}
1240	if (ls->ls_recover_size < jid + 1) {
1241		fs_err(sdp, "recovery_result jid %d short size %d\n",
1242		       jid, ls->ls_recover_size);
1243		spin_unlock(&ls->ls_recover_spin);
1244		return;
1245	}
1246
1247	fs_info(sdp, "recover jid %d result %s\n", jid,
1248		result == LM_RD_GAVEUP ? "busy" : "success");
1249
1250	ls->ls_recover_result[jid] = result;
1251
1252	/* GAVEUP means another node is recovering the journal; delay our
1253	   next attempt to recover it, to give the other node a chance to
1254	   finish before trying again */
1255
1256	if (!test_bit(DFL_UNMOUNT, &ls->ls_recover_flags))
1257		queue_delayed_work(gfs2_control_wq, &sdp->sd_control_work,
1258				   result == LM_RD_GAVEUP ? HZ : 0);
1259	spin_unlock(&ls->ls_recover_spin);
1260}
1261
1262static const struct dlm_lockspace_ops gdlm_lockspace_ops = {
1263	.recover_prep = gdlm_recover_prep,
1264	.recover_slot = gdlm_recover_slot,
1265	.recover_done = gdlm_recover_done,
1266};
1267
1268static int gdlm_mount(struct gfs2_sbd *sdp, const char *table)
1269{
1270	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1271	char cluster[GFS2_LOCKNAME_LEN];
1272	const char *fsname;
1273	uint32_t flags;
1274	int error, ops_result;
1275
1276	/*
1277	 * initialize everything
1278	 */
1279
1280	INIT_DELAYED_WORK(&sdp->sd_control_work, gfs2_control_func);
1281	spin_lock_init(&ls->ls_recover_spin);
1282	ls->ls_recover_flags = 0;
1283	ls->ls_recover_mount = 0;
1284	ls->ls_recover_start = 0;
1285	ls->ls_recover_block = 0;
1286	ls->ls_recover_size = 0;
1287	ls->ls_recover_submit = NULL;
1288	ls->ls_recover_result = NULL;
1289	ls->ls_lvb_bits = NULL;
1290
1291	error = set_recover_size(sdp, NULL, 0);
1292	if (error)
1293		goto fail;
1294
1295	/*
1296	 * prepare dlm_new_lockspace args
1297	 */
1298
1299	fsname = strchr(table, ':');
1300	if (!fsname) {
1301		fs_info(sdp, "no fsname found\n");
1302		error = -EINVAL;
1303		goto fail_free;
1304	}
1305	memset(cluster, 0, sizeof(cluster));
1306	memcpy(cluster, table, strlen(table) - strlen(fsname));
1307	fsname++;
1308
1309	flags = DLM_LSFL_NEWEXCL;
1310
1311	/*
1312	 * create/join lockspace
1313	 */
1314
1315	error = dlm_new_lockspace(fsname, cluster, flags, GDLM_LVB_SIZE,
1316				  &gdlm_lockspace_ops, sdp, &ops_result,
1317				  &ls->ls_dlm);
1318	if (error) {
1319		fs_err(sdp, "dlm_new_lockspace error %d\n", error);
1320		goto fail_free;
1321	}
1322
1323	if (ops_result < 0) {
1324		/*
1325		 * dlm does not support ops callbacks,
1326		 * old dlm_controld/gfs_controld are used, try without ops.
1327		 */
1328		fs_info(sdp, "dlm lockspace ops not used\n");
1329		free_recover_size(ls);
1330		set_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags);
1331		return 0;
1332	}
1333
1334	if (!test_bit(SDF_NOJOURNALID, &sdp->sd_flags)) {
1335		fs_err(sdp, "dlm lockspace ops disallow jid preset\n");
1336		error = -EINVAL;
1337		goto fail_release;
1338	}
1339
1340	/*
1341	 * control_mount() uses control_lock to determine first mounter,
1342	 * and for later mounts, waits for any recoveries to be cleared.
1343	 */
1344
1345	error = control_mount(sdp);
1346	if (error) {
1347		fs_err(sdp, "mount control error %d\n", error);
1348		goto fail_release;
1349	}
1350
1351	ls->ls_first = !!test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags);
1352	clear_bit(SDF_NOJOURNALID, &sdp->sd_flags);
1353	smp_mb__after_atomic();
1354	wake_up_bit(&sdp->sd_flags, SDF_NOJOURNALID);
1355	return 0;
1356
1357fail_release:
1358	dlm_release_lockspace(ls->ls_dlm, 2);
1359fail_free:
1360	free_recover_size(ls);
1361fail:
1362	return error;
1363}
1364
1365static void gdlm_first_done(struct gfs2_sbd *sdp)
1366{
1367	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1368	int error;
1369
1370	if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags))
1371		return;
1372
1373	error = control_first_done(sdp);
1374	if (error)
1375		fs_err(sdp, "mount first_done error %d\n", error);
1376}
1377
1378static void gdlm_unmount(struct gfs2_sbd *sdp)
1379{
1380	struct lm_lockstruct *ls = &sdp->sd_lockstruct;
1381
1382	if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags))
1383		goto release;
1384
1385	/* wait for gfs2_control_wq to be done with this mount */
1386
1387	spin_lock(&ls->ls_recover_spin);
1388	set_bit(DFL_UNMOUNT, &ls->ls_recover_flags);
1389	spin_unlock(&ls->ls_recover_spin);
1390	flush_delayed_work(&sdp->sd_control_work);
1391
1392	/* mounted_lock and control_lock will be purged in dlm recovery */
1393release:
1394	if (ls->ls_dlm) {
1395		dlm_release_lockspace(ls->ls_dlm, 2);
1396		ls->ls_dlm = NULL;
1397	}
1398
1399	free_recover_size(ls);
1400}
1401
1402static const match_table_t dlm_tokens = {
1403	{ Opt_jid, "jid=%d"},
1404	{ Opt_id, "id=%d"},
1405	{ Opt_first, "first=%d"},
1406	{ Opt_nodir, "nodir=%d"},
1407	{ Opt_err, NULL },
1408};
1409
1410const struct lm_lockops gfs2_dlm_ops = {
1411	.lm_proto_name = "lock_dlm",
1412	.lm_mount = gdlm_mount,
1413	.lm_first_done = gdlm_first_done,
1414	.lm_recovery_result = gdlm_recovery_result,
1415	.lm_unmount = gdlm_unmount,
1416	.lm_put_lock = gdlm_put_lock,
1417	.lm_lock = gdlm_lock,
1418	.lm_cancel = gdlm_cancel,
1419	.lm_tokens = &dlm_tokens,
1420};
1421