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 320 /* 321 * When the lockspace is released, all remaining glocks will be 322 * unlocked automatically. This is more efficient than unlocking them 323 * individually, but when the lock is held in DLM_LOCK_EX or 324 * DLM_LOCK_PW mode, the lock value block (LVB) will be lost. 325 */ 326 327 if (test_bit(SDF_SKIP_DLM_UNLOCK, &sdp->sd_flags) && 328 (!gl->gl_lksb.sb_lvbptr || gl->gl_state != LM_ST_EXCLUSIVE)) { 329 gfs2_glock_free_later(gl); 330 return; 331 } 332 333again: 334 error = dlm_unlock(ls->ls_dlm, gl->gl_lksb.sb_lkid, DLM_LKF_VALBLK, 335 NULL, gl); 336 if (error == -EBUSY) { 337 msleep(20); 338 goto again; 339 } 340 341 if (error) { 342 fs_err(sdp, "gdlm_unlock %x,%llx err=%d\n", 343 gl->gl_name.ln_type, 344 (unsigned long long)gl->gl_name.ln_number, error); 345 } 346} 347 348static void gdlm_cancel(struct gfs2_glock *gl) 349{ 350 struct lm_lockstruct *ls = &gl->gl_name.ln_sbd->sd_lockstruct; 351 dlm_unlock(ls->ls_dlm, gl->gl_lksb.sb_lkid, DLM_LKF_CANCEL, NULL, gl); 352} 353 354/* 355 * dlm/gfs2 recovery coordination using dlm_recover callbacks 356 * 357 * 0. gfs2 checks for another cluster node withdraw, needing journal replay 358 * 1. dlm_controld sees lockspace members change 359 * 2. dlm_controld blocks dlm-kernel locking activity 360 * 3. dlm_controld within dlm-kernel notifies gfs2 (recover_prep) 361 * 4. dlm_controld starts and finishes its own user level recovery 362 * 5. dlm_controld starts dlm-kernel dlm_recoverd to do kernel recovery 363 * 6. dlm_recoverd notifies gfs2 of failed nodes (recover_slot) 364 * 7. dlm_recoverd does its own lock recovery 365 * 8. dlm_recoverd unblocks dlm-kernel locking activity 366 * 9. dlm_recoverd notifies gfs2 when done (recover_done with new generation) 367 * 10. gfs2_control updates control_lock lvb with new generation and jid bits 368 * 11. gfs2_control enqueues journals for gfs2_recover to recover (maybe none) 369 * 12. gfs2_recover dequeues and recovers journals of failed nodes 370 * 13. gfs2_recover provides recovery results to gfs2_control (recovery_result) 371 * 14. gfs2_control updates control_lock lvb jid bits for recovered journals 372 * 15. gfs2_control unblocks normal locking when all journals are recovered 373 * 374 * - failures during recovery 375 * 376 * recover_prep() may set BLOCK_LOCKS (step 3) again before gfs2_control 377 * clears BLOCK_LOCKS (step 15), e.g. another node fails while still 378 * recovering for a prior failure. gfs2_control needs a way to detect 379 * this so it can leave BLOCK_LOCKS set in step 15. This is managed using 380 * the recover_block and recover_start values. 381 * 382 * recover_done() provides a new lockspace generation number each time it 383 * is called (step 9). This generation number is saved as recover_start. 384 * When recover_prep() is called, it sets BLOCK_LOCKS and sets 385 * recover_block = recover_start. So, while recover_block is equal to 386 * recover_start, BLOCK_LOCKS should remain set. (recover_spin must 387 * be held around the BLOCK_LOCKS/recover_block/recover_start logic.) 388 * 389 * - more specific gfs2 steps in sequence above 390 * 391 * 3. recover_prep sets BLOCK_LOCKS and sets recover_block = recover_start 392 * 6. recover_slot records any failed jids (maybe none) 393 * 9. recover_done sets recover_start = new generation number 394 * 10. gfs2_control sets control_lock lvb = new gen + bits for failed jids 395 * 12. gfs2_recover does journal recoveries for failed jids identified above 396 * 14. gfs2_control clears control_lock lvb bits for recovered jids 397 * 15. gfs2_control checks if recover_block == recover_start (step 3 occured 398 * again) then do nothing, otherwise if recover_start > recover_block 399 * then clear BLOCK_LOCKS. 400 * 401 * - parallel recovery steps across all nodes 402 * 403 * All nodes attempt to update the control_lock lvb with the new generation 404 * number and jid bits, but only the first to get the control_lock EX will 405 * do so; others will see that it's already done (lvb already contains new 406 * generation number.) 407 * 408 * . All nodes get the same recover_prep/recover_slot/recover_done callbacks 409 * . All nodes attempt to set control_lock lvb gen + bits for the new gen 410 * . One node gets control_lock first and writes the lvb, others see it's done 411 * . All nodes attempt to recover jids for which they see control_lock bits set 412 * . One node succeeds for a jid, and that one clears the jid bit in the lvb 413 * . All nodes will eventually see all lvb bits clear and unblock locks 414 * 415 * - is there a problem with clearing an lvb bit that should be set 416 * and missing a journal recovery? 417 * 418 * 1. jid fails 419 * 2. lvb bit set for step 1 420 * 3. jid recovered for step 1 421 * 4. jid taken again (new mount) 422 * 5. jid fails (for step 4) 423 * 6. lvb bit set for step 5 (will already be set) 424 * 7. lvb bit cleared for step 3 425 * 426 * This is not a problem because the failure in step 5 does not 427 * require recovery, because the mount in step 4 could not have 428 * progressed far enough to unblock locks and access the fs. The 429 * control_mount() function waits for all recoveries to be complete 430 * for the latest lockspace generation before ever unblocking locks 431 * and returning. The mount in step 4 waits until the recovery in 432 * step 1 is done. 433 * 434 * - special case of first mounter: first node to mount the fs 435 * 436 * The first node to mount a gfs2 fs needs to check all the journals 437 * and recover any that need recovery before other nodes are allowed 438 * to mount the fs. (Others may begin mounting, but they must wait 439 * for the first mounter to be done before taking locks on the fs 440 * or accessing the fs.) This has two parts: 441 * 442 * 1. The mounted_lock tells a node it's the first to mount the fs. 443 * Each node holds the mounted_lock in PR while it's mounted. 444 * Each node tries to acquire the mounted_lock in EX when it mounts. 445 * If a node is granted the mounted_lock EX it means there are no 446 * other mounted nodes (no PR locks exist), and it is the first mounter. 447 * The mounted_lock is demoted to PR when first recovery is done, so 448 * others will fail to get an EX lock, but will get a PR lock. 449 * 450 * 2. The control_lock blocks others in control_mount() while the first 451 * mounter is doing first mount recovery of all journals. 452 * A mounting node needs to acquire control_lock in EX mode before 453 * it can proceed. The first mounter holds control_lock in EX while doing 454 * the first mount recovery, blocking mounts from other nodes, then demotes 455 * control_lock to NL when it's done (others_may_mount/first_done), 456 * allowing other nodes to continue mounting. 457 * 458 * first mounter: 459 * control_lock EX/NOQUEUE success 460 * mounted_lock EX/NOQUEUE success (no other PR, so no other mounters) 461 * set first=1 462 * do first mounter recovery 463 * mounted_lock EX->PR 464 * control_lock EX->NL, write lvb generation 465 * 466 * other mounter: 467 * control_lock EX/NOQUEUE success (if fail -EAGAIN, retry) 468 * mounted_lock EX/NOQUEUE fail -EAGAIN (expected due to other mounters PR) 469 * mounted_lock PR/NOQUEUE success 470 * read lvb generation 471 * control_lock EX->NL 472 * set first=0 473 * 474 * - mount during recovery 475 * 476 * If a node mounts while others are doing recovery (not first mounter), 477 * the mounting node will get its initial recover_done() callback without 478 * having seen any previous failures/callbacks. 479 * 480 * It must wait for all recoveries preceding its mount to be finished 481 * before it unblocks locks. It does this by repeating the "other mounter" 482 * steps above until the lvb generation number is >= its mount generation 483 * number (from initial recover_done) and all lvb bits are clear. 484 * 485 * - control_lock lvb format 486 * 487 * 4 bytes generation number: the latest dlm lockspace generation number 488 * from recover_done callback. Indicates the jid bitmap has been updated 489 * to reflect all slot failures through that generation. 490 * 4 bytes unused. 491 * GDLM_LVB_SIZE-8 bytes of jid bit map. If bit N is set, it indicates 492 * that jid N needs recovery. 493 */ 494 495#define JID_BITMAP_OFFSET 8 /* 4 byte generation number + 4 byte unused */ 496 497static void control_lvb_read(struct lm_lockstruct *ls, uint32_t *lvb_gen, 498 char *lvb_bits) 499{ 500 __le32 gen; 501 memcpy(lvb_bits, ls->ls_control_lvb, GDLM_LVB_SIZE); 502 memcpy(&gen, lvb_bits, sizeof(__le32)); 503 *lvb_gen = le32_to_cpu(gen); 504} 505 506static void control_lvb_write(struct lm_lockstruct *ls, uint32_t lvb_gen, 507 char *lvb_bits) 508{ 509 __le32 gen; 510 memcpy(ls->ls_control_lvb, lvb_bits, GDLM_LVB_SIZE); 511 gen = cpu_to_le32(lvb_gen); 512 memcpy(ls->ls_control_lvb, &gen, sizeof(__le32)); 513} 514 515static int all_jid_bits_clear(char *lvb) 516{ 517 return !memchr_inv(lvb + JID_BITMAP_OFFSET, 0, 518 GDLM_LVB_SIZE - JID_BITMAP_OFFSET); 519} 520 521static void sync_wait_cb(void *arg) 522{ 523 struct lm_lockstruct *ls = arg; 524 complete(&ls->ls_sync_wait); 525} 526 527static int sync_unlock(struct gfs2_sbd *sdp, struct dlm_lksb *lksb, char *name) 528{ 529 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 530 int error; 531 532 error = dlm_unlock(ls->ls_dlm, lksb->sb_lkid, 0, lksb, ls); 533 if (error) { 534 fs_err(sdp, "%s lkid %x error %d\n", 535 name, lksb->sb_lkid, error); 536 return error; 537 } 538 539 wait_for_completion(&ls->ls_sync_wait); 540 541 if (lksb->sb_status != -DLM_EUNLOCK) { 542 fs_err(sdp, "%s lkid %x status %d\n", 543 name, lksb->sb_lkid, lksb->sb_status); 544 return -1; 545 } 546 return 0; 547} 548 549static int sync_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags, 550 unsigned int num, struct dlm_lksb *lksb, char *name) 551{ 552 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 553 char strname[GDLM_STRNAME_BYTES]; 554 int error, status; 555 556 memset(strname, 0, GDLM_STRNAME_BYTES); 557 snprintf(strname, GDLM_STRNAME_BYTES, "%8x%16x", LM_TYPE_NONDISK, num); 558 559 error = dlm_lock(ls->ls_dlm, mode, lksb, flags, 560 strname, GDLM_STRNAME_BYTES - 1, 561 0, sync_wait_cb, ls, NULL); 562 if (error) { 563 fs_err(sdp, "%s lkid %x flags %x mode %d error %d\n", 564 name, lksb->sb_lkid, flags, mode, error); 565 return error; 566 } 567 568 wait_for_completion(&ls->ls_sync_wait); 569 570 status = lksb->sb_status; 571 572 if (status && status != -EAGAIN) { 573 fs_err(sdp, "%s lkid %x flags %x mode %d status %d\n", 574 name, lksb->sb_lkid, flags, mode, status); 575 } 576 577 return status; 578} 579 580static int mounted_unlock(struct gfs2_sbd *sdp) 581{ 582 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 583 return sync_unlock(sdp, &ls->ls_mounted_lksb, "mounted_lock"); 584} 585 586static int mounted_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags) 587{ 588 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 589 return sync_lock(sdp, mode, flags, GFS2_MOUNTED_LOCK, 590 &ls->ls_mounted_lksb, "mounted_lock"); 591} 592 593static int control_unlock(struct gfs2_sbd *sdp) 594{ 595 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 596 return sync_unlock(sdp, &ls->ls_control_lksb, "control_lock"); 597} 598 599static int control_lock(struct gfs2_sbd *sdp, int mode, uint32_t flags) 600{ 601 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 602 return sync_lock(sdp, mode, flags, GFS2_CONTROL_LOCK, 603 &ls->ls_control_lksb, "control_lock"); 604} 605 606/** 607 * remote_withdraw - react to a node withdrawing from the file system 608 * @sdp: The superblock 609 */ 610static void remote_withdraw(struct gfs2_sbd *sdp) 611{ 612 struct gfs2_jdesc *jd; 613 int ret = 0, count = 0; 614 615 list_for_each_entry(jd, &sdp->sd_jindex_list, jd_list) { 616 if (jd->jd_jid == sdp->sd_lockstruct.ls_jid) 617 continue; 618 ret = gfs2_recover_journal(jd, true); 619 if (ret) 620 break; 621 count++; 622 } 623 624 /* Now drop the additional reference we acquired */ 625 fs_err(sdp, "Journals checked: %d, ret = %d.\n", count, ret); 626} 627 628static void gfs2_control_func(struct work_struct *work) 629{ 630 struct gfs2_sbd *sdp = container_of(work, struct gfs2_sbd, sd_control_work.work); 631 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 632 uint32_t block_gen, start_gen, lvb_gen, flags; 633 int recover_set = 0; 634 int write_lvb = 0; 635 int recover_size; 636 int i, error; 637 638 /* First check for other nodes that may have done a withdraw. */ 639 if (test_bit(SDF_REMOTE_WITHDRAW, &sdp->sd_flags)) { 640 remote_withdraw(sdp); 641 clear_bit(SDF_REMOTE_WITHDRAW, &sdp->sd_flags); 642 return; 643 } 644 645 spin_lock(&ls->ls_recover_spin); 646 /* 647 * No MOUNT_DONE means we're still mounting; control_mount() 648 * will set this flag, after which this thread will take over 649 * all further clearing of BLOCK_LOCKS. 650 * 651 * FIRST_MOUNT means this node is doing first mounter recovery, 652 * for which recovery control is handled by 653 * control_mount()/control_first_done(), not this thread. 654 */ 655 if (!test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) || 656 test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) { 657 spin_unlock(&ls->ls_recover_spin); 658 return; 659 } 660 block_gen = ls->ls_recover_block; 661 start_gen = ls->ls_recover_start; 662 spin_unlock(&ls->ls_recover_spin); 663 664 /* 665 * Equal block_gen and start_gen implies we are between 666 * recover_prep and recover_done callbacks, which means 667 * dlm recovery is in progress and dlm locking is blocked. 668 * There's no point trying to do any work until recover_done. 669 */ 670 671 if (block_gen == start_gen) 672 return; 673 674 /* 675 * Propagate recover_submit[] and recover_result[] to lvb: 676 * dlm_recoverd adds to recover_submit[] jids needing recovery 677 * gfs2_recover adds to recover_result[] journal recovery results 678 * 679 * set lvb bit for jids in recover_submit[] if the lvb has not 680 * yet been updated for the generation of the failure 681 * 682 * clear lvb bit for jids in recover_result[] if the result of 683 * the journal recovery is SUCCESS 684 */ 685 686 error = control_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_VALBLK); 687 if (error) { 688 fs_err(sdp, "control lock EX error %d\n", error); 689 return; 690 } 691 692 control_lvb_read(ls, &lvb_gen, ls->ls_lvb_bits); 693 694 spin_lock(&ls->ls_recover_spin); 695 if (block_gen != ls->ls_recover_block || 696 start_gen != ls->ls_recover_start) { 697 fs_info(sdp, "recover generation %u block1 %u %u\n", 698 start_gen, block_gen, ls->ls_recover_block); 699 spin_unlock(&ls->ls_recover_spin); 700 control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT); 701 return; 702 } 703 704 recover_size = ls->ls_recover_size; 705 706 if (lvb_gen <= start_gen) { 707 /* 708 * Clear lvb bits for jids we've successfully recovered. 709 * Because all nodes attempt to recover failed journals, 710 * a journal can be recovered multiple times successfully 711 * in succession. Only the first will really do recovery, 712 * the others find it clean, but still report a successful 713 * recovery. So, another node may have already recovered 714 * the jid and cleared the lvb bit for it. 715 */ 716 for (i = 0; i < recover_size; i++) { 717 if (ls->ls_recover_result[i] != LM_RD_SUCCESS) 718 continue; 719 720 ls->ls_recover_result[i] = 0; 721 722 if (!test_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET)) 723 continue; 724 725 __clear_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET); 726 write_lvb = 1; 727 } 728 } 729 730 if (lvb_gen == start_gen) { 731 /* 732 * Failed slots before start_gen are already set in lvb. 733 */ 734 for (i = 0; i < recover_size; i++) { 735 if (!ls->ls_recover_submit[i]) 736 continue; 737 if (ls->ls_recover_submit[i] < lvb_gen) 738 ls->ls_recover_submit[i] = 0; 739 } 740 } else if (lvb_gen < start_gen) { 741 /* 742 * Failed slots before start_gen are not yet set in lvb. 743 */ 744 for (i = 0; i < recover_size; i++) { 745 if (!ls->ls_recover_submit[i]) 746 continue; 747 if (ls->ls_recover_submit[i] < start_gen) { 748 ls->ls_recover_submit[i] = 0; 749 __set_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET); 750 } 751 } 752 /* even if there are no bits to set, we need to write the 753 latest generation to the lvb */ 754 write_lvb = 1; 755 } else { 756 /* 757 * we should be getting a recover_done() for lvb_gen soon 758 */ 759 } 760 spin_unlock(&ls->ls_recover_spin); 761 762 if (write_lvb) { 763 control_lvb_write(ls, start_gen, ls->ls_lvb_bits); 764 flags = DLM_LKF_CONVERT | DLM_LKF_VALBLK; 765 } else { 766 flags = DLM_LKF_CONVERT; 767 } 768 769 error = control_lock(sdp, DLM_LOCK_NL, flags); 770 if (error) { 771 fs_err(sdp, "control lock NL error %d\n", error); 772 return; 773 } 774 775 /* 776 * Everyone will see jid bits set in the lvb, run gfs2_recover_set(), 777 * and clear a jid bit in the lvb if the recovery is a success. 778 * Eventually all journals will be recovered, all jid bits will 779 * be cleared in the lvb, and everyone will clear BLOCK_LOCKS. 780 */ 781 782 for (i = 0; i < recover_size; i++) { 783 if (test_bit_le(i, ls->ls_lvb_bits + JID_BITMAP_OFFSET)) { 784 fs_info(sdp, "recover generation %u jid %d\n", 785 start_gen, i); 786 gfs2_recover_set(sdp, i); 787 recover_set++; 788 } 789 } 790 if (recover_set) 791 return; 792 793 /* 794 * No more jid bits set in lvb, all recovery is done, unblock locks 795 * (unless a new recover_prep callback has occured blocking locks 796 * again while working above) 797 */ 798 799 spin_lock(&ls->ls_recover_spin); 800 if (ls->ls_recover_block == block_gen && 801 ls->ls_recover_start == start_gen) { 802 clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags); 803 spin_unlock(&ls->ls_recover_spin); 804 fs_info(sdp, "recover generation %u done\n", start_gen); 805 gfs2_glock_thaw(sdp); 806 } else { 807 fs_info(sdp, "recover generation %u block2 %u %u\n", 808 start_gen, block_gen, ls->ls_recover_block); 809 spin_unlock(&ls->ls_recover_spin); 810 } 811} 812 813static int control_mount(struct gfs2_sbd *sdp) 814{ 815 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 816 uint32_t start_gen, block_gen, mount_gen, lvb_gen; 817 int mounted_mode; 818 int retries = 0; 819 int error; 820 821 memset(&ls->ls_mounted_lksb, 0, sizeof(struct dlm_lksb)); 822 memset(&ls->ls_control_lksb, 0, sizeof(struct dlm_lksb)); 823 memset(&ls->ls_control_lvb, 0, GDLM_LVB_SIZE); 824 ls->ls_control_lksb.sb_lvbptr = ls->ls_control_lvb; 825 init_completion(&ls->ls_sync_wait); 826 827 set_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags); 828 829 error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_VALBLK); 830 if (error) { 831 fs_err(sdp, "control_mount control_lock NL error %d\n", error); 832 return error; 833 } 834 835 error = mounted_lock(sdp, DLM_LOCK_NL, 0); 836 if (error) { 837 fs_err(sdp, "control_mount mounted_lock NL error %d\n", error); 838 control_unlock(sdp); 839 return error; 840 } 841 mounted_mode = DLM_LOCK_NL; 842 843restart: 844 if (retries++ && signal_pending(current)) { 845 error = -EINTR; 846 goto fail; 847 } 848 849 /* 850 * We always start with both locks in NL. control_lock is 851 * demoted to NL below so we don't need to do it here. 852 */ 853 854 if (mounted_mode != DLM_LOCK_NL) { 855 error = mounted_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT); 856 if (error) 857 goto fail; 858 mounted_mode = DLM_LOCK_NL; 859 } 860 861 /* 862 * Other nodes need to do some work in dlm recovery and gfs2_control 863 * before the recover_done and control_lock will be ready for us below. 864 * A delay here is not required but often avoids having to retry. 865 */ 866 867 msleep_interruptible(500); 868 869 /* 870 * Acquire control_lock in EX and mounted_lock in either EX or PR. 871 * control_lock lvb keeps track of any pending journal recoveries. 872 * mounted_lock indicates if any other nodes have the fs mounted. 873 */ 874 875 error = control_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE|DLM_LKF_VALBLK); 876 if (error == -EAGAIN) { 877 goto restart; 878 } else if (error) { 879 fs_err(sdp, "control_mount control_lock EX error %d\n", error); 880 goto fail; 881 } 882 883 /** 884 * If we're a spectator, we don't want to take the lock in EX because 885 * we cannot do the first-mount responsibility it implies: recovery. 886 */ 887 if (sdp->sd_args.ar_spectator) 888 goto locks_done; 889 890 error = mounted_lock(sdp, DLM_LOCK_EX, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE); 891 if (!error) { 892 mounted_mode = DLM_LOCK_EX; 893 goto locks_done; 894 } else if (error != -EAGAIN) { 895 fs_err(sdp, "control_mount mounted_lock EX error %d\n", error); 896 goto fail; 897 } 898 899 error = mounted_lock(sdp, DLM_LOCK_PR, DLM_LKF_CONVERT|DLM_LKF_NOQUEUE); 900 if (!error) { 901 mounted_mode = DLM_LOCK_PR; 902 goto locks_done; 903 } else { 904 /* not even -EAGAIN should happen here */ 905 fs_err(sdp, "control_mount mounted_lock PR error %d\n", error); 906 goto fail; 907 } 908 909locks_done: 910 /* 911 * If we got both locks above in EX, then we're the first mounter. 912 * If not, then we need to wait for the control_lock lvb to be 913 * updated by other mounted nodes to reflect our mount generation. 914 * 915 * In simple first mounter cases, first mounter will see zero lvb_gen, 916 * but in cases where all existing nodes leave/fail before mounting 917 * nodes finish control_mount, then all nodes will be mounting and 918 * lvb_gen will be non-zero. 919 */ 920 921 control_lvb_read(ls, &lvb_gen, ls->ls_lvb_bits); 922 923 if (lvb_gen == 0xFFFFFFFF) { 924 /* special value to force mount attempts to fail */ 925 fs_err(sdp, "control_mount control_lock disabled\n"); 926 error = -EINVAL; 927 goto fail; 928 } 929 930 if (mounted_mode == DLM_LOCK_EX) { 931 /* first mounter, keep both EX while doing first recovery */ 932 spin_lock(&ls->ls_recover_spin); 933 clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags); 934 set_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags); 935 set_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags); 936 spin_unlock(&ls->ls_recover_spin); 937 fs_info(sdp, "first mounter control generation %u\n", lvb_gen); 938 return 0; 939 } 940 941 error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT); 942 if (error) 943 goto fail; 944 945 /* 946 * We are not first mounter, now we need to wait for the control_lock 947 * lvb generation to be >= the generation from our first recover_done 948 * and all lvb bits to be clear (no pending journal recoveries.) 949 */ 950 951 if (!all_jid_bits_clear(ls->ls_lvb_bits)) { 952 /* journals need recovery, wait until all are clear */ 953 fs_info(sdp, "control_mount wait for journal recovery\n"); 954 goto restart; 955 } 956 957 spin_lock(&ls->ls_recover_spin); 958 block_gen = ls->ls_recover_block; 959 start_gen = ls->ls_recover_start; 960 mount_gen = ls->ls_recover_mount; 961 962 if (lvb_gen < mount_gen) { 963 /* wait for mounted nodes to update control_lock lvb to our 964 generation, which might include new recovery bits set */ 965 if (sdp->sd_args.ar_spectator) { 966 fs_info(sdp, "Recovery is required. Waiting for a " 967 "non-spectator to mount.\n"); 968 msleep_interruptible(1000); 969 } else { 970 fs_info(sdp, "control_mount wait1 block %u start %u " 971 "mount %u lvb %u flags %lx\n", block_gen, 972 start_gen, mount_gen, lvb_gen, 973 ls->ls_recover_flags); 974 } 975 spin_unlock(&ls->ls_recover_spin); 976 goto restart; 977 } 978 979 if (lvb_gen != start_gen) { 980 /* wait for mounted nodes to update control_lock lvb to the 981 latest recovery generation */ 982 fs_info(sdp, "control_mount wait2 block %u start %u mount %u " 983 "lvb %u flags %lx\n", block_gen, start_gen, mount_gen, 984 lvb_gen, ls->ls_recover_flags); 985 spin_unlock(&ls->ls_recover_spin); 986 goto restart; 987 } 988 989 if (block_gen == start_gen) { 990 /* dlm recovery in progress, wait for it to finish */ 991 fs_info(sdp, "control_mount wait3 block %u start %u mount %u " 992 "lvb %u flags %lx\n", block_gen, start_gen, mount_gen, 993 lvb_gen, ls->ls_recover_flags); 994 spin_unlock(&ls->ls_recover_spin); 995 goto restart; 996 } 997 998 clear_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags); 999 set_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags); 1000 memset(ls->ls_recover_submit, 0, ls->ls_recover_size*sizeof(uint32_t)); 1001 memset(ls->ls_recover_result, 0, ls->ls_recover_size*sizeof(uint32_t)); 1002 spin_unlock(&ls->ls_recover_spin); 1003 return 0; 1004 1005fail: 1006 mounted_unlock(sdp); 1007 control_unlock(sdp); 1008 return error; 1009} 1010 1011static int control_first_done(struct gfs2_sbd *sdp) 1012{ 1013 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1014 uint32_t start_gen, block_gen; 1015 int error; 1016 1017restart: 1018 spin_lock(&ls->ls_recover_spin); 1019 start_gen = ls->ls_recover_start; 1020 block_gen = ls->ls_recover_block; 1021 1022 if (test_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags) || 1023 !test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) || 1024 !test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) { 1025 /* sanity check, should not happen */ 1026 fs_err(sdp, "control_first_done start %u block %u flags %lx\n", 1027 start_gen, block_gen, ls->ls_recover_flags); 1028 spin_unlock(&ls->ls_recover_spin); 1029 control_unlock(sdp); 1030 return -1; 1031 } 1032 1033 if (start_gen == block_gen) { 1034 /* 1035 * Wait for the end of a dlm recovery cycle to switch from 1036 * first mounter recovery. We can ignore any recover_slot 1037 * callbacks between the recover_prep and next recover_done 1038 * because we are still the first mounter and any failed nodes 1039 * have not fully mounted, so they don't need recovery. 1040 */ 1041 spin_unlock(&ls->ls_recover_spin); 1042 fs_info(sdp, "control_first_done wait gen %u\n", start_gen); 1043 1044 wait_on_bit(&ls->ls_recover_flags, DFL_DLM_RECOVERY, 1045 TASK_UNINTERRUPTIBLE); 1046 goto restart; 1047 } 1048 1049 clear_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags); 1050 set_bit(DFL_FIRST_MOUNT_DONE, &ls->ls_recover_flags); 1051 memset(ls->ls_recover_submit, 0, ls->ls_recover_size*sizeof(uint32_t)); 1052 memset(ls->ls_recover_result, 0, ls->ls_recover_size*sizeof(uint32_t)); 1053 spin_unlock(&ls->ls_recover_spin); 1054 1055 memset(ls->ls_lvb_bits, 0, GDLM_LVB_SIZE); 1056 control_lvb_write(ls, start_gen, ls->ls_lvb_bits); 1057 1058 error = mounted_lock(sdp, DLM_LOCK_PR, DLM_LKF_CONVERT); 1059 if (error) 1060 fs_err(sdp, "control_first_done mounted PR error %d\n", error); 1061 1062 error = control_lock(sdp, DLM_LOCK_NL, DLM_LKF_CONVERT|DLM_LKF_VALBLK); 1063 if (error) 1064 fs_err(sdp, "control_first_done control NL error %d\n", error); 1065 1066 return error; 1067} 1068 1069/* 1070 * Expand static jid arrays if necessary (by increments of RECOVER_SIZE_INC) 1071 * to accommodate the largest slot number. (NB dlm slot numbers start at 1, 1072 * gfs2 jids start at 0, so jid = slot - 1) 1073 */ 1074 1075#define RECOVER_SIZE_INC 16 1076 1077static int set_recover_size(struct gfs2_sbd *sdp, struct dlm_slot *slots, 1078 int num_slots) 1079{ 1080 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1081 uint32_t *submit = NULL; 1082 uint32_t *result = NULL; 1083 uint32_t old_size, new_size; 1084 int i, max_jid; 1085 1086 if (!ls->ls_lvb_bits) { 1087 ls->ls_lvb_bits = kzalloc(GDLM_LVB_SIZE, GFP_NOFS); 1088 if (!ls->ls_lvb_bits) 1089 return -ENOMEM; 1090 } 1091 1092 max_jid = 0; 1093 for (i = 0; i < num_slots; i++) { 1094 if (max_jid < slots[i].slot - 1) 1095 max_jid = slots[i].slot - 1; 1096 } 1097 1098 old_size = ls->ls_recover_size; 1099 new_size = old_size; 1100 while (new_size < max_jid + 1) 1101 new_size += RECOVER_SIZE_INC; 1102 if (new_size == old_size) 1103 return 0; 1104 1105 submit = kcalloc(new_size, sizeof(uint32_t), GFP_NOFS); 1106 result = kcalloc(new_size, sizeof(uint32_t), GFP_NOFS); 1107 if (!submit || !result) { 1108 kfree(submit); 1109 kfree(result); 1110 return -ENOMEM; 1111 } 1112 1113 spin_lock(&ls->ls_recover_spin); 1114 memcpy(submit, ls->ls_recover_submit, old_size * sizeof(uint32_t)); 1115 memcpy(result, ls->ls_recover_result, old_size * sizeof(uint32_t)); 1116 kfree(ls->ls_recover_submit); 1117 kfree(ls->ls_recover_result); 1118 ls->ls_recover_submit = submit; 1119 ls->ls_recover_result = result; 1120 ls->ls_recover_size = new_size; 1121 spin_unlock(&ls->ls_recover_spin); 1122 return 0; 1123} 1124 1125static void free_recover_size(struct lm_lockstruct *ls) 1126{ 1127 kfree(ls->ls_lvb_bits); 1128 kfree(ls->ls_recover_submit); 1129 kfree(ls->ls_recover_result); 1130 ls->ls_recover_submit = NULL; 1131 ls->ls_recover_result = NULL; 1132 ls->ls_recover_size = 0; 1133 ls->ls_lvb_bits = NULL; 1134} 1135 1136/* dlm calls before it does lock recovery */ 1137 1138static void gdlm_recover_prep(void *arg) 1139{ 1140 struct gfs2_sbd *sdp = arg; 1141 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1142 1143 if (gfs2_withdrawing_or_withdrawn(sdp)) { 1144 fs_err(sdp, "recover_prep ignored due to withdraw.\n"); 1145 return; 1146 } 1147 spin_lock(&ls->ls_recover_spin); 1148 ls->ls_recover_block = ls->ls_recover_start; 1149 set_bit(DFL_DLM_RECOVERY, &ls->ls_recover_flags); 1150 1151 if (!test_bit(DFL_MOUNT_DONE, &ls->ls_recover_flags) || 1152 test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) { 1153 spin_unlock(&ls->ls_recover_spin); 1154 return; 1155 } 1156 set_bit(DFL_BLOCK_LOCKS, &ls->ls_recover_flags); 1157 spin_unlock(&ls->ls_recover_spin); 1158} 1159 1160/* dlm calls after recover_prep has been completed on all lockspace members; 1161 identifies slot/jid of failed member */ 1162 1163static void gdlm_recover_slot(void *arg, struct dlm_slot *slot) 1164{ 1165 struct gfs2_sbd *sdp = arg; 1166 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1167 int jid = slot->slot - 1; 1168 1169 if (gfs2_withdrawing_or_withdrawn(sdp)) { 1170 fs_err(sdp, "recover_slot jid %d ignored due to withdraw.\n", 1171 jid); 1172 return; 1173 } 1174 spin_lock(&ls->ls_recover_spin); 1175 if (ls->ls_recover_size < jid + 1) { 1176 fs_err(sdp, "recover_slot jid %d gen %u short size %d\n", 1177 jid, ls->ls_recover_block, ls->ls_recover_size); 1178 spin_unlock(&ls->ls_recover_spin); 1179 return; 1180 } 1181 1182 if (ls->ls_recover_submit[jid]) { 1183 fs_info(sdp, "recover_slot jid %d gen %u prev %u\n", 1184 jid, ls->ls_recover_block, ls->ls_recover_submit[jid]); 1185 } 1186 ls->ls_recover_submit[jid] = ls->ls_recover_block; 1187 spin_unlock(&ls->ls_recover_spin); 1188} 1189 1190/* dlm calls after recover_slot and after it completes lock recovery */ 1191 1192static void gdlm_recover_done(void *arg, struct dlm_slot *slots, int num_slots, 1193 int our_slot, uint32_t generation) 1194{ 1195 struct gfs2_sbd *sdp = arg; 1196 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1197 1198 if (gfs2_withdrawing_or_withdrawn(sdp)) { 1199 fs_err(sdp, "recover_done ignored due to withdraw.\n"); 1200 return; 1201 } 1202 /* ensure the ls jid arrays are large enough */ 1203 set_recover_size(sdp, slots, num_slots); 1204 1205 spin_lock(&ls->ls_recover_spin); 1206 ls->ls_recover_start = generation; 1207 1208 if (!ls->ls_recover_mount) { 1209 ls->ls_recover_mount = generation; 1210 ls->ls_jid = our_slot - 1; 1211 } 1212 1213 if (!test_bit(DFL_UNMOUNT, &ls->ls_recover_flags)) 1214 queue_delayed_work(gfs2_control_wq, &sdp->sd_control_work, 0); 1215 1216 clear_bit(DFL_DLM_RECOVERY, &ls->ls_recover_flags); 1217 smp_mb__after_atomic(); 1218 wake_up_bit(&ls->ls_recover_flags, DFL_DLM_RECOVERY); 1219 spin_unlock(&ls->ls_recover_spin); 1220} 1221 1222/* gfs2_recover thread has a journal recovery result */ 1223 1224static void gdlm_recovery_result(struct gfs2_sbd *sdp, unsigned int jid, 1225 unsigned int result) 1226{ 1227 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1228 1229 if (gfs2_withdrawing_or_withdrawn(sdp)) { 1230 fs_err(sdp, "recovery_result jid %d ignored due to withdraw.\n", 1231 jid); 1232 return; 1233 } 1234 if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags)) 1235 return; 1236 1237 /* don't care about the recovery of own journal during mount */ 1238 if (jid == ls->ls_jid) 1239 return; 1240 1241 spin_lock(&ls->ls_recover_spin); 1242 if (test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags)) { 1243 spin_unlock(&ls->ls_recover_spin); 1244 return; 1245 } 1246 if (ls->ls_recover_size < jid + 1) { 1247 fs_err(sdp, "recovery_result jid %d short size %d\n", 1248 jid, ls->ls_recover_size); 1249 spin_unlock(&ls->ls_recover_spin); 1250 return; 1251 } 1252 1253 fs_info(sdp, "recover jid %d result %s\n", jid, 1254 result == LM_RD_GAVEUP ? "busy" : "success"); 1255 1256 ls->ls_recover_result[jid] = result; 1257 1258 /* GAVEUP means another node is recovering the journal; delay our 1259 next attempt to recover it, to give the other node a chance to 1260 finish before trying again */ 1261 1262 if (!test_bit(DFL_UNMOUNT, &ls->ls_recover_flags)) 1263 queue_delayed_work(gfs2_control_wq, &sdp->sd_control_work, 1264 result == LM_RD_GAVEUP ? HZ : 0); 1265 spin_unlock(&ls->ls_recover_spin); 1266} 1267 1268static const struct dlm_lockspace_ops gdlm_lockspace_ops = { 1269 .recover_prep = gdlm_recover_prep, 1270 .recover_slot = gdlm_recover_slot, 1271 .recover_done = gdlm_recover_done, 1272}; 1273 1274static int gdlm_mount(struct gfs2_sbd *sdp, const char *table) 1275{ 1276 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1277 char cluster[GFS2_LOCKNAME_LEN]; 1278 const char *fsname; 1279 uint32_t flags; 1280 int error, ops_result; 1281 1282 /* 1283 * initialize everything 1284 */ 1285 1286 INIT_DELAYED_WORK(&sdp->sd_control_work, gfs2_control_func); 1287 spin_lock_init(&ls->ls_recover_spin); 1288 ls->ls_recover_flags = 0; 1289 ls->ls_recover_mount = 0; 1290 ls->ls_recover_start = 0; 1291 ls->ls_recover_block = 0; 1292 ls->ls_recover_size = 0; 1293 ls->ls_recover_submit = NULL; 1294 ls->ls_recover_result = NULL; 1295 ls->ls_lvb_bits = NULL; 1296 1297 error = set_recover_size(sdp, NULL, 0); 1298 if (error) 1299 goto fail; 1300 1301 /* 1302 * prepare dlm_new_lockspace args 1303 */ 1304 1305 fsname = strchr(table, ':'); 1306 if (!fsname) { 1307 fs_info(sdp, "no fsname found\n"); 1308 error = -EINVAL; 1309 goto fail_free; 1310 } 1311 memset(cluster, 0, sizeof(cluster)); 1312 memcpy(cluster, table, strlen(table) - strlen(fsname)); 1313 fsname++; 1314 1315 flags = DLM_LSFL_NEWEXCL; 1316 1317 /* 1318 * create/join lockspace 1319 */ 1320 1321 error = dlm_new_lockspace(fsname, cluster, flags, GDLM_LVB_SIZE, 1322 &gdlm_lockspace_ops, sdp, &ops_result, 1323 &ls->ls_dlm); 1324 if (error) { 1325 fs_err(sdp, "dlm_new_lockspace error %d\n", error); 1326 goto fail_free; 1327 } 1328 1329 if (ops_result < 0) { 1330 /* 1331 * dlm does not support ops callbacks, 1332 * old dlm_controld/gfs_controld are used, try without ops. 1333 */ 1334 fs_info(sdp, "dlm lockspace ops not used\n"); 1335 free_recover_size(ls); 1336 set_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags); 1337 return 0; 1338 } 1339 1340 if (!test_bit(SDF_NOJOURNALID, &sdp->sd_flags)) { 1341 fs_err(sdp, "dlm lockspace ops disallow jid preset\n"); 1342 error = -EINVAL; 1343 goto fail_release; 1344 } 1345 1346 /* 1347 * control_mount() uses control_lock to determine first mounter, 1348 * and for later mounts, waits for any recoveries to be cleared. 1349 */ 1350 1351 error = control_mount(sdp); 1352 if (error) { 1353 fs_err(sdp, "mount control error %d\n", error); 1354 goto fail_release; 1355 } 1356 1357 ls->ls_first = !!test_bit(DFL_FIRST_MOUNT, &ls->ls_recover_flags); 1358 clear_bit(SDF_NOJOURNALID, &sdp->sd_flags); 1359 smp_mb__after_atomic(); 1360 wake_up_bit(&sdp->sd_flags, SDF_NOJOURNALID); 1361 return 0; 1362 1363fail_release: 1364 dlm_release_lockspace(ls->ls_dlm, 2); 1365fail_free: 1366 free_recover_size(ls); 1367fail: 1368 return error; 1369} 1370 1371static void gdlm_first_done(struct gfs2_sbd *sdp) 1372{ 1373 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1374 int error; 1375 1376 if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags)) 1377 return; 1378 1379 error = control_first_done(sdp); 1380 if (error) 1381 fs_err(sdp, "mount first_done error %d\n", error); 1382} 1383 1384static void gdlm_unmount(struct gfs2_sbd *sdp) 1385{ 1386 struct lm_lockstruct *ls = &sdp->sd_lockstruct; 1387 1388 if (test_bit(DFL_NO_DLM_OPS, &ls->ls_recover_flags)) 1389 goto release; 1390 1391 /* wait for gfs2_control_wq to be done with this mount */ 1392 1393 spin_lock(&ls->ls_recover_spin); 1394 set_bit(DFL_UNMOUNT, &ls->ls_recover_flags); 1395 spin_unlock(&ls->ls_recover_spin); 1396 flush_delayed_work(&sdp->sd_control_work); 1397 1398 /* mounted_lock and control_lock will be purged in dlm recovery */ 1399release: 1400 if (ls->ls_dlm) { 1401 dlm_release_lockspace(ls->ls_dlm, 2); 1402 ls->ls_dlm = NULL; 1403 } 1404 1405 free_recover_size(ls); 1406} 1407 1408static const match_table_t dlm_tokens = { 1409 { Opt_jid, "jid=%d"}, 1410 { Opt_id, "id=%d"}, 1411 { Opt_first, "first=%d"}, 1412 { Opt_nodir, "nodir=%d"}, 1413 { Opt_err, NULL }, 1414}; 1415 1416const struct lm_lockops gfs2_dlm_ops = { 1417 .lm_proto_name = "lock_dlm", 1418 .lm_mount = gdlm_mount, 1419 .lm_first_done = gdlm_first_done, 1420 .lm_recovery_result = gdlm_recovery_result, 1421 .lm_unmount = gdlm_unmount, 1422 .lm_put_lock = gdlm_put_lock, 1423 .lm_lock = gdlm_lock, 1424 .lm_cancel = gdlm_cancel, 1425 .lm_tokens = &dlm_tokens, 1426}; 1427 1428