1 /* 2 drbd_worker.c 3 4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg. 5 6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH. 7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>. 8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>. 9 10 drbd is free software; you can redistribute it and/or modify 11 it under the terms of the GNU General Public License as published by 12 the Free Software Foundation; either version 2, or (at your option) 13 any later version. 14 15 drbd is distributed in the hope that it will be useful, 16 but WITHOUT ANY WARRANTY; without even the implied warranty of 17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 18 GNU General Public License for more details. 19 20 You should have received a copy of the GNU General Public License 21 along with drbd; see the file COPYING. If not, write to 22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA. 23 24 */ 25 26 #include <linux/module.h> 27 #include <linux/drbd.h> 28 #include <linux/sched.h> 29 #include <linux/wait.h> 30 #include <linux/mm.h> 31 #include <linux/memcontrol.h> 32 #include <linux/mm_inline.h> 33 #include <linux/slab.h> 34 #include <linux/random.h> 35 #include <linux/string.h> 36 #include <linux/scatterlist.h> 37 38 #include "drbd_int.h" 39 #include "drbd_req.h" 40 41 static int w_make_ov_request(struct drbd_conf *mdev, struct drbd_work *w, int cancel); 42 43 44 45 /* defined here: 46 drbd_md_io_complete 47 drbd_endio_sec 48 drbd_endio_pri 49 50 * more endio handlers: 51 atodb_endio in drbd_actlog.c 52 drbd_bm_async_io_complete in drbd_bitmap.c 53 54 * For all these callbacks, note the following: 55 * The callbacks will be called in irq context by the IDE drivers, 56 * and in Softirqs/Tasklets/BH context by the SCSI drivers. 57 * Try to get the locking right :) 58 * 59 */ 60 61 62 /* About the global_state_lock 63 Each state transition on an device holds a read lock. In case we have 64 to evaluate the sync after dependencies, we grab a write lock, because 65 we need stable states on all devices for that. */ 66 rwlock_t global_state_lock; 67 68 /* used for synchronous meta data and bitmap IO 69 * submitted by drbd_md_sync_page_io() 70 */ 71 void drbd_md_io_complete(struct bio *bio, int error) 72 { 73 struct drbd_md_io *md_io; 74 75 md_io = (struct drbd_md_io *)bio->bi_private; 76 md_io->error = error; 77 78 complete(&md_io->event); 79 } 80 81 /* reads on behalf of the partner, 82 * "submitted" by the receiver 83 */ 84 void drbd_endio_read_sec_final(struct drbd_epoch_entry *e) __releases(local) 85 { 86 unsigned long flags = 0; 87 struct drbd_conf *mdev = e->mdev; 88 89 D_ASSERT(e->block_id != ID_VACANT); 90 91 spin_lock_irqsave(&mdev->req_lock, flags); 92 mdev->read_cnt += e->size >> 9; 93 list_del(&e->w.list); 94 if (list_empty(&mdev->read_ee)) 95 wake_up(&mdev->ee_wait); 96 if (test_bit(__EE_WAS_ERROR, &e->flags)) 97 __drbd_chk_io_error(mdev, FALSE); 98 spin_unlock_irqrestore(&mdev->req_lock, flags); 99 100 drbd_queue_work(&mdev->data.work, &e->w); 101 put_ldev(mdev); 102 } 103 104 /* writes on behalf of the partner, or resync writes, 105 * "submitted" by the receiver, final stage. */ 106 static void drbd_endio_write_sec_final(struct drbd_epoch_entry *e) __releases(local) 107 { 108 unsigned long flags = 0; 109 struct drbd_conf *mdev = e->mdev; 110 sector_t e_sector; 111 int do_wake; 112 int is_syncer_req; 113 int do_al_complete_io; 114 115 D_ASSERT(e->block_id != ID_VACANT); 116 117 /* after we moved e to done_ee, 118 * we may no longer access it, 119 * it may be freed/reused already! 120 * (as soon as we release the req_lock) */ 121 e_sector = e->sector; 122 do_al_complete_io = e->flags & EE_CALL_AL_COMPLETE_IO; 123 is_syncer_req = is_syncer_block_id(e->block_id); 124 125 spin_lock_irqsave(&mdev->req_lock, flags); 126 mdev->writ_cnt += e->size >> 9; 127 list_del(&e->w.list); /* has been on active_ee or sync_ee */ 128 list_add_tail(&e->w.list, &mdev->done_ee); 129 130 /* No hlist_del_init(&e->colision) here, we did not send the Ack yet, 131 * neither did we wake possibly waiting conflicting requests. 132 * done from "drbd_process_done_ee" within the appropriate w.cb 133 * (e_end_block/e_end_resync_block) or from _drbd_clear_done_ee */ 134 135 do_wake = is_syncer_req 136 ? list_empty(&mdev->sync_ee) 137 : list_empty(&mdev->active_ee); 138 139 if (test_bit(__EE_WAS_ERROR, &e->flags)) 140 __drbd_chk_io_error(mdev, FALSE); 141 spin_unlock_irqrestore(&mdev->req_lock, flags); 142 143 if (is_syncer_req) 144 drbd_rs_complete_io(mdev, e_sector); 145 146 if (do_wake) 147 wake_up(&mdev->ee_wait); 148 149 if (do_al_complete_io) 150 drbd_al_complete_io(mdev, e_sector); 151 152 wake_asender(mdev); 153 put_ldev(mdev); 154 } 155 156 /* writes on behalf of the partner, or resync writes, 157 * "submitted" by the receiver. 158 */ 159 void drbd_endio_sec(struct bio *bio, int error) 160 { 161 struct drbd_epoch_entry *e = bio->bi_private; 162 struct drbd_conf *mdev = e->mdev; 163 int uptodate = bio_flagged(bio, BIO_UPTODATE); 164 int is_write = bio_data_dir(bio) == WRITE; 165 166 if (error) 167 dev_warn(DEV, "%s: error=%d s=%llus\n", 168 is_write ? "write" : "read", error, 169 (unsigned long long)e->sector); 170 if (!error && !uptodate) { 171 dev_warn(DEV, "%s: setting error to -EIO s=%llus\n", 172 is_write ? "write" : "read", 173 (unsigned long long)e->sector); 174 /* strange behavior of some lower level drivers... 175 * fail the request by clearing the uptodate flag, 176 * but do not return any error?! */ 177 error = -EIO; 178 } 179 180 if (error) 181 set_bit(__EE_WAS_ERROR, &e->flags); 182 183 bio_put(bio); /* no need for the bio anymore */ 184 if (atomic_dec_and_test(&e->pending_bios)) { 185 if (is_write) 186 drbd_endio_write_sec_final(e); 187 else 188 drbd_endio_read_sec_final(e); 189 } 190 } 191 192 /* read, readA or write requests on R_PRIMARY coming from drbd_make_request 193 */ 194 void drbd_endio_pri(struct bio *bio, int error) 195 { 196 unsigned long flags; 197 struct drbd_request *req = bio->bi_private; 198 struct drbd_conf *mdev = req->mdev; 199 struct bio_and_error m; 200 enum drbd_req_event what; 201 int uptodate = bio_flagged(bio, BIO_UPTODATE); 202 203 if (!error && !uptodate) { 204 dev_warn(DEV, "p %s: setting error to -EIO\n", 205 bio_data_dir(bio) == WRITE ? "write" : "read"); 206 /* strange behavior of some lower level drivers... 207 * fail the request by clearing the uptodate flag, 208 * but do not return any error?! */ 209 error = -EIO; 210 } 211 212 /* to avoid recursion in __req_mod */ 213 if (unlikely(error)) { 214 what = (bio_data_dir(bio) == WRITE) 215 ? write_completed_with_error 216 : (bio_rw(bio) == READ) 217 ? read_completed_with_error 218 : read_ahead_completed_with_error; 219 } else 220 what = completed_ok; 221 222 bio_put(req->private_bio); 223 req->private_bio = ERR_PTR(error); 224 225 /* not req_mod(), we need irqsave here! */ 226 spin_lock_irqsave(&mdev->req_lock, flags); 227 __req_mod(req, what, &m); 228 spin_unlock_irqrestore(&mdev->req_lock, flags); 229 230 if (m.bio) 231 complete_master_bio(mdev, &m); 232 } 233 234 int w_read_retry_remote(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 235 { 236 struct drbd_request *req = container_of(w, struct drbd_request, w); 237 238 /* We should not detach for read io-error, 239 * but try to WRITE the P_DATA_REPLY to the failed location, 240 * to give the disk the chance to relocate that block */ 241 242 spin_lock_irq(&mdev->req_lock); 243 if (cancel || mdev->state.pdsk != D_UP_TO_DATE) { 244 _req_mod(req, read_retry_remote_canceled); 245 spin_unlock_irq(&mdev->req_lock); 246 return 1; 247 } 248 spin_unlock_irq(&mdev->req_lock); 249 250 return w_send_read_req(mdev, w, 0); 251 } 252 253 int w_resync_inactive(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 254 { 255 ERR_IF(cancel) return 1; 256 dev_err(DEV, "resync inactive, but callback triggered??\n"); 257 return 1; /* Simply ignore this! */ 258 } 259 260 void drbd_csum_ee(struct drbd_conf *mdev, struct crypto_hash *tfm, struct drbd_epoch_entry *e, void *digest) 261 { 262 struct hash_desc desc; 263 struct scatterlist sg; 264 struct page *page = e->pages; 265 struct page *tmp; 266 unsigned len; 267 268 desc.tfm = tfm; 269 desc.flags = 0; 270 271 sg_init_table(&sg, 1); 272 crypto_hash_init(&desc); 273 274 while ((tmp = page_chain_next(page))) { 275 /* all but the last page will be fully used */ 276 sg_set_page(&sg, page, PAGE_SIZE, 0); 277 crypto_hash_update(&desc, &sg, sg.length); 278 page = tmp; 279 } 280 /* and now the last, possibly only partially used page */ 281 len = e->size & (PAGE_SIZE - 1); 282 sg_set_page(&sg, page, len ?: PAGE_SIZE, 0); 283 crypto_hash_update(&desc, &sg, sg.length); 284 crypto_hash_final(&desc, digest); 285 } 286 287 void drbd_csum_bio(struct drbd_conf *mdev, struct crypto_hash *tfm, struct bio *bio, void *digest) 288 { 289 struct hash_desc desc; 290 struct scatterlist sg; 291 struct bio_vec *bvec; 292 int i; 293 294 desc.tfm = tfm; 295 desc.flags = 0; 296 297 sg_init_table(&sg, 1); 298 crypto_hash_init(&desc); 299 300 __bio_for_each_segment(bvec, bio, i, 0) { 301 sg_set_page(&sg, bvec->bv_page, bvec->bv_len, bvec->bv_offset); 302 crypto_hash_update(&desc, &sg, sg.length); 303 } 304 crypto_hash_final(&desc, digest); 305 } 306 307 static int w_e_send_csum(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 308 { 309 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w); 310 int digest_size; 311 void *digest; 312 int ok; 313 314 D_ASSERT(e->block_id == DRBD_MAGIC + 0xbeef); 315 316 if (unlikely(cancel)) { 317 drbd_free_ee(mdev, e); 318 return 1; 319 } 320 321 if (likely((e->flags & EE_WAS_ERROR) == 0)) { 322 digest_size = crypto_hash_digestsize(mdev->csums_tfm); 323 digest = kmalloc(digest_size, GFP_NOIO); 324 if (digest) { 325 drbd_csum_ee(mdev, mdev->csums_tfm, e, digest); 326 327 inc_rs_pending(mdev); 328 ok = drbd_send_drequest_csum(mdev, 329 e->sector, 330 e->size, 331 digest, 332 digest_size, 333 P_CSUM_RS_REQUEST); 334 kfree(digest); 335 } else { 336 dev_err(DEV, "kmalloc() of digest failed.\n"); 337 ok = 0; 338 } 339 } else 340 ok = 1; 341 342 drbd_free_ee(mdev, e); 343 344 if (unlikely(!ok)) 345 dev_err(DEV, "drbd_send_drequest(..., csum) failed\n"); 346 return ok; 347 } 348 349 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN) 350 351 static int read_for_csum(struct drbd_conf *mdev, sector_t sector, int size) 352 { 353 struct drbd_epoch_entry *e; 354 355 if (!get_ldev(mdev)) 356 return -EIO; 357 358 if (drbd_rs_should_slow_down(mdev)) 359 goto defer; 360 361 /* GFP_TRY, because if there is no memory available right now, this may 362 * be rescheduled for later. It is "only" background resync, after all. */ 363 e = drbd_alloc_ee(mdev, DRBD_MAGIC+0xbeef, sector, size, GFP_TRY); 364 if (!e) 365 goto defer; 366 367 e->w.cb = w_e_send_csum; 368 spin_lock_irq(&mdev->req_lock); 369 list_add(&e->w.list, &mdev->read_ee); 370 spin_unlock_irq(&mdev->req_lock); 371 372 atomic_add(size >> 9, &mdev->rs_sect_ev); 373 if (drbd_submit_ee(mdev, e, READ, DRBD_FAULT_RS_RD) == 0) 374 return 0; 375 376 /* drbd_submit_ee currently fails for one reason only: 377 * not being able to allocate enough bios. 378 * Is dropping the connection going to help? */ 379 spin_lock_irq(&mdev->req_lock); 380 list_del(&e->w.list); 381 spin_unlock_irq(&mdev->req_lock); 382 383 drbd_free_ee(mdev, e); 384 defer: 385 put_ldev(mdev); 386 return -EAGAIN; 387 } 388 389 void resync_timer_fn(unsigned long data) 390 { 391 struct drbd_conf *mdev = (struct drbd_conf *) data; 392 int queue; 393 394 queue = 1; 395 switch (mdev->state.conn) { 396 case C_VERIFY_S: 397 mdev->resync_work.cb = w_make_ov_request; 398 break; 399 case C_SYNC_TARGET: 400 mdev->resync_work.cb = w_make_resync_request; 401 break; 402 default: 403 queue = 0; 404 mdev->resync_work.cb = w_resync_inactive; 405 } 406 407 /* harmless race: list_empty outside data.work.q_lock */ 408 if (list_empty(&mdev->resync_work.list) && queue) 409 drbd_queue_work(&mdev->data.work, &mdev->resync_work); 410 } 411 412 static void fifo_set(struct fifo_buffer *fb, int value) 413 { 414 int i; 415 416 for (i = 0; i < fb->size; i++) 417 fb->values[i] = value; 418 } 419 420 static int fifo_push(struct fifo_buffer *fb, int value) 421 { 422 int ov; 423 424 ov = fb->values[fb->head_index]; 425 fb->values[fb->head_index++] = value; 426 427 if (fb->head_index >= fb->size) 428 fb->head_index = 0; 429 430 return ov; 431 } 432 433 static void fifo_add_val(struct fifo_buffer *fb, int value) 434 { 435 int i; 436 437 for (i = 0; i < fb->size; i++) 438 fb->values[i] += value; 439 } 440 441 int drbd_rs_controller(struct drbd_conf *mdev) 442 { 443 unsigned int sect_in; /* Number of sectors that came in since the last turn */ 444 unsigned int want; /* The number of sectors we want in the proxy */ 445 int req_sect; /* Number of sectors to request in this turn */ 446 int correction; /* Number of sectors more we need in the proxy*/ 447 int cps; /* correction per invocation of drbd_rs_controller() */ 448 int steps; /* Number of time steps to plan ahead */ 449 int curr_corr; 450 int max_sect; 451 452 sect_in = atomic_xchg(&mdev->rs_sect_in, 0); /* Number of sectors that came in */ 453 mdev->rs_in_flight -= sect_in; 454 455 spin_lock(&mdev->peer_seq_lock); /* get an atomic view on mdev->rs_plan_s */ 456 457 steps = mdev->rs_plan_s.size; /* (mdev->sync_conf.c_plan_ahead * 10 * SLEEP_TIME) / HZ; */ 458 459 if (mdev->rs_in_flight + sect_in == 0) { /* At start of resync */ 460 want = ((mdev->sync_conf.rate * 2 * SLEEP_TIME) / HZ) * steps; 461 } else { /* normal path */ 462 want = mdev->sync_conf.c_fill_target ? mdev->sync_conf.c_fill_target : 463 sect_in * mdev->sync_conf.c_delay_target * HZ / (SLEEP_TIME * 10); 464 } 465 466 correction = want - mdev->rs_in_flight - mdev->rs_planed; 467 468 /* Plan ahead */ 469 cps = correction / steps; 470 fifo_add_val(&mdev->rs_plan_s, cps); 471 mdev->rs_planed += cps * steps; 472 473 /* What we do in this step */ 474 curr_corr = fifo_push(&mdev->rs_plan_s, 0); 475 spin_unlock(&mdev->peer_seq_lock); 476 mdev->rs_planed -= curr_corr; 477 478 req_sect = sect_in + curr_corr; 479 if (req_sect < 0) 480 req_sect = 0; 481 482 max_sect = (mdev->sync_conf.c_max_rate * 2 * SLEEP_TIME) / HZ; 483 if (req_sect > max_sect) 484 req_sect = max_sect; 485 486 /* 487 dev_warn(DEV, "si=%u if=%d wa=%u co=%d st=%d cps=%d pl=%d cc=%d rs=%d\n", 488 sect_in, mdev->rs_in_flight, want, correction, 489 steps, cps, mdev->rs_planed, curr_corr, req_sect); 490 */ 491 492 return req_sect; 493 } 494 495 int drbd_rs_number_requests(struct drbd_conf *mdev) 496 { 497 int number; 498 if (mdev->rs_plan_s.size) { /* mdev->sync_conf.c_plan_ahead */ 499 number = drbd_rs_controller(mdev) >> (BM_BLOCK_SHIFT - 9); 500 mdev->c_sync_rate = number * HZ * (BM_BLOCK_SIZE / 1024) / SLEEP_TIME; 501 } else { 502 mdev->c_sync_rate = mdev->sync_conf.rate; 503 number = SLEEP_TIME * mdev->c_sync_rate / ((BM_BLOCK_SIZE / 1024) * HZ); 504 } 505 506 /* Throttle resync on lower level disk activity, which may also be 507 * caused by application IO on Primary/SyncTarget. 508 * Keep this after the call to drbd_rs_controller, as that assumes 509 * to be called as precisely as possible every SLEEP_TIME, 510 * and would be confused otherwise. */ 511 if (number && drbd_rs_should_slow_down(mdev)) { 512 mdev->c_sync_rate = 1; 513 number = 0; 514 } 515 516 /* ignore the amount of pending requests, the resync controller should 517 * throttle down to incoming reply rate soon enough anyways. */ 518 return number; 519 } 520 521 int w_make_resync_request(struct drbd_conf *mdev, 522 struct drbd_work *w, int cancel) 523 { 524 unsigned long bit; 525 sector_t sector; 526 const sector_t capacity = drbd_get_capacity(mdev->this_bdev); 527 int max_bio_size; 528 int number, rollback_i, size; 529 int align, queued, sndbuf; 530 int i = 0; 531 532 if (unlikely(cancel)) 533 return 1; 534 535 if (unlikely(mdev->state.conn < C_CONNECTED)) { 536 dev_err(DEV, "Confused in w_make_resync_request()! cstate < Connected"); 537 return 0; 538 } 539 540 if (mdev->state.conn != C_SYNC_TARGET) 541 dev_err(DEV, "%s in w_make_resync_request\n", 542 drbd_conn_str(mdev->state.conn)); 543 544 if (mdev->rs_total == 0) { 545 /* empty resync? */ 546 drbd_resync_finished(mdev); 547 return 1; 548 } 549 550 if (!get_ldev(mdev)) { 551 /* Since we only need to access mdev->rsync a 552 get_ldev_if_state(mdev,D_FAILED) would be sufficient, but 553 to continue resync with a broken disk makes no sense at 554 all */ 555 dev_err(DEV, "Disk broke down during resync!\n"); 556 mdev->resync_work.cb = w_resync_inactive; 557 return 1; 558 } 559 560 /* starting with drbd 8.3.8, we can handle multi-bio EEs, 561 * if it should be necessary */ 562 max_bio_size = 563 mdev->agreed_pro_version < 94 ? queue_max_hw_sectors(mdev->rq_queue) << 9 : 564 mdev->agreed_pro_version < 95 ? DRBD_MAX_SIZE_H80_PACKET : DRBD_MAX_BIO_SIZE; 565 566 number = drbd_rs_number_requests(mdev); 567 if (number == 0) 568 goto requeue; 569 570 for (i = 0; i < number; i++) { 571 /* Stop generating RS requests, when half of the send buffer is filled */ 572 mutex_lock(&mdev->data.mutex); 573 if (mdev->data.socket) { 574 queued = mdev->data.socket->sk->sk_wmem_queued; 575 sndbuf = mdev->data.socket->sk->sk_sndbuf; 576 } else { 577 queued = 1; 578 sndbuf = 0; 579 } 580 mutex_unlock(&mdev->data.mutex); 581 if (queued > sndbuf / 2) 582 goto requeue; 583 584 next_sector: 585 size = BM_BLOCK_SIZE; 586 bit = drbd_bm_find_next(mdev, mdev->bm_resync_fo); 587 588 if (bit == -1UL) { 589 mdev->bm_resync_fo = drbd_bm_bits(mdev); 590 mdev->resync_work.cb = w_resync_inactive; 591 put_ldev(mdev); 592 return 1; 593 } 594 595 sector = BM_BIT_TO_SECT(bit); 596 597 if (drbd_try_rs_begin_io(mdev, sector)) { 598 mdev->bm_resync_fo = bit; 599 goto requeue; 600 } 601 mdev->bm_resync_fo = bit + 1; 602 603 if (unlikely(drbd_bm_test_bit(mdev, bit) == 0)) { 604 drbd_rs_complete_io(mdev, sector); 605 goto next_sector; 606 } 607 608 #if DRBD_MAX_BIO_SIZE > BM_BLOCK_SIZE 609 /* try to find some adjacent bits. 610 * we stop if we have already the maximum req size. 611 * 612 * Additionally always align bigger requests, in order to 613 * be prepared for all stripe sizes of software RAIDs. 614 */ 615 align = 1; 616 rollback_i = i; 617 for (;;) { 618 if (size + BM_BLOCK_SIZE > max_bio_size) 619 break; 620 621 /* Be always aligned */ 622 if (sector & ((1<<(align+3))-1)) 623 break; 624 625 /* do not cross extent boundaries */ 626 if (((bit+1) & BM_BLOCKS_PER_BM_EXT_MASK) == 0) 627 break; 628 /* now, is it actually dirty, after all? 629 * caution, drbd_bm_test_bit is tri-state for some 630 * obscure reason; ( b == 0 ) would get the out-of-band 631 * only accidentally right because of the "oddly sized" 632 * adjustment below */ 633 if (drbd_bm_test_bit(mdev, bit+1) != 1) 634 break; 635 bit++; 636 size += BM_BLOCK_SIZE; 637 if ((BM_BLOCK_SIZE << align) <= size) 638 align++; 639 i++; 640 } 641 /* if we merged some, 642 * reset the offset to start the next drbd_bm_find_next from */ 643 if (size > BM_BLOCK_SIZE) 644 mdev->bm_resync_fo = bit + 1; 645 #endif 646 647 /* adjust very last sectors, in case we are oddly sized */ 648 if (sector + (size>>9) > capacity) 649 size = (capacity-sector)<<9; 650 if (mdev->agreed_pro_version >= 89 && mdev->csums_tfm) { 651 switch (read_for_csum(mdev, sector, size)) { 652 case -EIO: /* Disk failure */ 653 put_ldev(mdev); 654 return 0; 655 case -EAGAIN: /* allocation failed, or ldev busy */ 656 drbd_rs_complete_io(mdev, sector); 657 mdev->bm_resync_fo = BM_SECT_TO_BIT(sector); 658 i = rollback_i; 659 goto requeue; 660 case 0: 661 /* everything ok */ 662 break; 663 default: 664 BUG(); 665 } 666 } else { 667 inc_rs_pending(mdev); 668 if (!drbd_send_drequest(mdev, P_RS_DATA_REQUEST, 669 sector, size, ID_SYNCER)) { 670 dev_err(DEV, "drbd_send_drequest() failed, aborting...\n"); 671 dec_rs_pending(mdev); 672 put_ldev(mdev); 673 return 0; 674 } 675 } 676 } 677 678 if (mdev->bm_resync_fo >= drbd_bm_bits(mdev)) { 679 /* last syncer _request_ was sent, 680 * but the P_RS_DATA_REPLY not yet received. sync will end (and 681 * next sync group will resume), as soon as we receive the last 682 * resync data block, and the last bit is cleared. 683 * until then resync "work" is "inactive" ... 684 */ 685 mdev->resync_work.cb = w_resync_inactive; 686 put_ldev(mdev); 687 return 1; 688 } 689 690 requeue: 691 mdev->rs_in_flight += (i << (BM_BLOCK_SHIFT - 9)); 692 mod_timer(&mdev->resync_timer, jiffies + SLEEP_TIME); 693 put_ldev(mdev); 694 return 1; 695 } 696 697 static int w_make_ov_request(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 698 { 699 int number, i, size; 700 sector_t sector; 701 const sector_t capacity = drbd_get_capacity(mdev->this_bdev); 702 703 if (unlikely(cancel)) 704 return 1; 705 706 if (unlikely(mdev->state.conn < C_CONNECTED)) { 707 dev_err(DEV, "Confused in w_make_ov_request()! cstate < Connected"); 708 return 0; 709 } 710 711 number = drbd_rs_number_requests(mdev); 712 713 sector = mdev->ov_position; 714 for (i = 0; i < number; i++) { 715 if (sector >= capacity) { 716 mdev->resync_work.cb = w_resync_inactive; 717 return 1; 718 } 719 720 size = BM_BLOCK_SIZE; 721 722 if (drbd_try_rs_begin_io(mdev, sector)) { 723 mdev->ov_position = sector; 724 goto requeue; 725 } 726 727 if (sector + (size>>9) > capacity) 728 size = (capacity-sector)<<9; 729 730 inc_rs_pending(mdev); 731 if (!drbd_send_ov_request(mdev, sector, size)) { 732 dec_rs_pending(mdev); 733 return 0; 734 } 735 sector += BM_SECT_PER_BIT; 736 } 737 mdev->ov_position = sector; 738 739 requeue: 740 mdev->rs_in_flight += (i << (BM_BLOCK_SHIFT - 9)); 741 mod_timer(&mdev->resync_timer, jiffies + SLEEP_TIME); 742 return 1; 743 } 744 745 int w_ov_finished(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 746 { 747 kfree(w); 748 ov_oos_print(mdev); 749 drbd_resync_finished(mdev); 750 751 return 1; 752 } 753 754 static int w_resync_finished(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 755 { 756 kfree(w); 757 758 drbd_resync_finished(mdev); 759 760 return 1; 761 } 762 763 static void ping_peer(struct drbd_conf *mdev) 764 { 765 clear_bit(GOT_PING_ACK, &mdev->flags); 766 request_ping(mdev); 767 wait_event(mdev->misc_wait, 768 test_bit(GOT_PING_ACK, &mdev->flags) || mdev->state.conn < C_CONNECTED); 769 } 770 771 int drbd_resync_finished(struct drbd_conf *mdev) 772 { 773 unsigned long db, dt, dbdt; 774 unsigned long n_oos; 775 union drbd_state os, ns; 776 struct drbd_work *w; 777 char *khelper_cmd = NULL; 778 int verify_done = 0; 779 780 /* Remove all elements from the resync LRU. Since future actions 781 * might set bits in the (main) bitmap, then the entries in the 782 * resync LRU would be wrong. */ 783 if (drbd_rs_del_all(mdev)) { 784 /* In case this is not possible now, most probably because 785 * there are P_RS_DATA_REPLY Packets lingering on the worker's 786 * queue (or even the read operations for those packets 787 * is not finished by now). Retry in 100ms. */ 788 789 __set_current_state(TASK_INTERRUPTIBLE); 790 schedule_timeout(HZ / 10); 791 w = kmalloc(sizeof(struct drbd_work), GFP_ATOMIC); 792 if (w) { 793 w->cb = w_resync_finished; 794 drbd_queue_work(&mdev->data.work, w); 795 return 1; 796 } 797 dev_err(DEV, "Warn failed to drbd_rs_del_all() and to kmalloc(w).\n"); 798 } 799 800 dt = (jiffies - mdev->rs_start - mdev->rs_paused) / HZ; 801 if (dt <= 0) 802 dt = 1; 803 db = mdev->rs_total; 804 dbdt = Bit2KB(db/dt); 805 mdev->rs_paused /= HZ; 806 807 if (!get_ldev(mdev)) 808 goto out; 809 810 ping_peer(mdev); 811 812 spin_lock_irq(&mdev->req_lock); 813 os = mdev->state; 814 815 verify_done = (os.conn == C_VERIFY_S || os.conn == C_VERIFY_T); 816 817 /* This protects us against multiple calls (that can happen in the presence 818 of application IO), and against connectivity loss just before we arrive here. */ 819 if (os.conn <= C_CONNECTED) 820 goto out_unlock; 821 822 ns = os; 823 ns.conn = C_CONNECTED; 824 825 dev_info(DEV, "%s done (total %lu sec; paused %lu sec; %lu K/sec)\n", 826 verify_done ? "Online verify " : "Resync", 827 dt + mdev->rs_paused, mdev->rs_paused, dbdt); 828 829 n_oos = drbd_bm_total_weight(mdev); 830 831 if (os.conn == C_VERIFY_S || os.conn == C_VERIFY_T) { 832 if (n_oos) { 833 dev_alert(DEV, "Online verify found %lu %dk block out of sync!\n", 834 n_oos, Bit2KB(1)); 835 khelper_cmd = "out-of-sync"; 836 } 837 } else { 838 D_ASSERT((n_oos - mdev->rs_failed) == 0); 839 840 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) 841 khelper_cmd = "after-resync-target"; 842 843 if (mdev->csums_tfm && mdev->rs_total) { 844 const unsigned long s = mdev->rs_same_csum; 845 const unsigned long t = mdev->rs_total; 846 const int ratio = 847 (t == 0) ? 0 : 848 (t < 100000) ? ((s*100)/t) : (s/(t/100)); 849 dev_info(DEV, "%u %% had equal check sums, eliminated: %luK; " 850 "transferred %luK total %luK\n", 851 ratio, 852 Bit2KB(mdev->rs_same_csum), 853 Bit2KB(mdev->rs_total - mdev->rs_same_csum), 854 Bit2KB(mdev->rs_total)); 855 } 856 } 857 858 if (mdev->rs_failed) { 859 dev_info(DEV, " %lu failed blocks\n", mdev->rs_failed); 860 861 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) { 862 ns.disk = D_INCONSISTENT; 863 ns.pdsk = D_UP_TO_DATE; 864 } else { 865 ns.disk = D_UP_TO_DATE; 866 ns.pdsk = D_INCONSISTENT; 867 } 868 } else { 869 ns.disk = D_UP_TO_DATE; 870 ns.pdsk = D_UP_TO_DATE; 871 872 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) { 873 if (mdev->p_uuid) { 874 int i; 875 for (i = UI_BITMAP ; i <= UI_HISTORY_END ; i++) 876 _drbd_uuid_set(mdev, i, mdev->p_uuid[i]); 877 drbd_uuid_set(mdev, UI_BITMAP, mdev->ldev->md.uuid[UI_CURRENT]); 878 _drbd_uuid_set(mdev, UI_CURRENT, mdev->p_uuid[UI_CURRENT]); 879 } else { 880 dev_err(DEV, "mdev->p_uuid is NULL! BUG\n"); 881 } 882 } 883 884 drbd_uuid_set_bm(mdev, 0UL); 885 886 if (mdev->p_uuid) { 887 /* Now the two UUID sets are equal, update what we 888 * know of the peer. */ 889 int i; 890 for (i = UI_CURRENT ; i <= UI_HISTORY_END ; i++) 891 mdev->p_uuid[i] = mdev->ldev->md.uuid[i]; 892 } 893 } 894 895 _drbd_set_state(mdev, ns, CS_VERBOSE, NULL); 896 out_unlock: 897 spin_unlock_irq(&mdev->req_lock); 898 put_ldev(mdev); 899 out: 900 mdev->rs_total = 0; 901 mdev->rs_failed = 0; 902 mdev->rs_paused = 0; 903 if (verify_done) 904 mdev->ov_start_sector = 0; 905 906 drbd_md_sync(mdev); 907 908 if (test_and_clear_bit(WRITE_BM_AFTER_RESYNC, &mdev->flags)) { 909 dev_info(DEV, "Writing the whole bitmap\n"); 910 drbd_queue_bitmap_io(mdev, &drbd_bm_write, NULL, "write from resync_finished"); 911 } 912 913 if (khelper_cmd) 914 drbd_khelper(mdev, khelper_cmd); 915 916 return 1; 917 } 918 919 /* helper */ 920 static void move_to_net_ee_or_free(struct drbd_conf *mdev, struct drbd_epoch_entry *e) 921 { 922 if (drbd_ee_has_active_page(e)) { 923 /* This might happen if sendpage() has not finished */ 924 int i = (e->size + PAGE_SIZE -1) >> PAGE_SHIFT; 925 atomic_add(i, &mdev->pp_in_use_by_net); 926 atomic_sub(i, &mdev->pp_in_use); 927 spin_lock_irq(&mdev->req_lock); 928 list_add_tail(&e->w.list, &mdev->net_ee); 929 spin_unlock_irq(&mdev->req_lock); 930 wake_up(&drbd_pp_wait); 931 } else 932 drbd_free_ee(mdev, e); 933 } 934 935 /** 936 * w_e_end_data_req() - Worker callback, to send a P_DATA_REPLY packet in response to a P_DATA_REQUEST 937 * @mdev: DRBD device. 938 * @w: work object. 939 * @cancel: The connection will be closed anyways 940 */ 941 int w_e_end_data_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 942 { 943 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w); 944 int ok; 945 946 if (unlikely(cancel)) { 947 drbd_free_ee(mdev, e); 948 dec_unacked(mdev); 949 return 1; 950 } 951 952 if (likely((e->flags & EE_WAS_ERROR) == 0)) { 953 ok = drbd_send_block(mdev, P_DATA_REPLY, e); 954 } else { 955 if (__ratelimit(&drbd_ratelimit_state)) 956 dev_err(DEV, "Sending NegDReply. sector=%llus.\n", 957 (unsigned long long)e->sector); 958 959 ok = drbd_send_ack(mdev, P_NEG_DREPLY, e); 960 } 961 962 dec_unacked(mdev); 963 964 move_to_net_ee_or_free(mdev, e); 965 966 if (unlikely(!ok)) 967 dev_err(DEV, "drbd_send_block() failed\n"); 968 return ok; 969 } 970 971 /** 972 * w_e_end_rsdata_req() - Worker callback to send a P_RS_DATA_REPLY packet in response to a P_RS_DATA_REQUESTRS 973 * @mdev: DRBD device. 974 * @w: work object. 975 * @cancel: The connection will be closed anyways 976 */ 977 int w_e_end_rsdata_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 978 { 979 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w); 980 int ok; 981 982 if (unlikely(cancel)) { 983 drbd_free_ee(mdev, e); 984 dec_unacked(mdev); 985 return 1; 986 } 987 988 if (get_ldev_if_state(mdev, D_FAILED)) { 989 drbd_rs_complete_io(mdev, e->sector); 990 put_ldev(mdev); 991 } 992 993 if (likely((e->flags & EE_WAS_ERROR) == 0)) { 994 if (likely(mdev->state.pdsk >= D_INCONSISTENT)) { 995 inc_rs_pending(mdev); 996 ok = drbd_send_block(mdev, P_RS_DATA_REPLY, e); 997 } else { 998 if (__ratelimit(&drbd_ratelimit_state)) 999 dev_err(DEV, "Not sending RSDataReply, " 1000 "partner DISKLESS!\n"); 1001 ok = 1; 1002 } 1003 } else { 1004 if (__ratelimit(&drbd_ratelimit_state)) 1005 dev_err(DEV, "Sending NegRSDReply. sector %llus.\n", 1006 (unsigned long long)e->sector); 1007 1008 ok = drbd_send_ack(mdev, P_NEG_RS_DREPLY, e); 1009 1010 /* update resync data with failure */ 1011 drbd_rs_failed_io(mdev, e->sector, e->size); 1012 } 1013 1014 dec_unacked(mdev); 1015 1016 move_to_net_ee_or_free(mdev, e); 1017 1018 if (unlikely(!ok)) 1019 dev_err(DEV, "drbd_send_block() failed\n"); 1020 return ok; 1021 } 1022 1023 int w_e_end_csum_rs_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1024 { 1025 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w); 1026 struct digest_info *di; 1027 int digest_size; 1028 void *digest = NULL; 1029 int ok, eq = 0; 1030 1031 if (unlikely(cancel)) { 1032 drbd_free_ee(mdev, e); 1033 dec_unacked(mdev); 1034 return 1; 1035 } 1036 1037 if (get_ldev(mdev)) { 1038 drbd_rs_complete_io(mdev, e->sector); 1039 put_ldev(mdev); 1040 } 1041 1042 di = e->digest; 1043 1044 if (likely((e->flags & EE_WAS_ERROR) == 0)) { 1045 /* quick hack to try to avoid a race against reconfiguration. 1046 * a real fix would be much more involved, 1047 * introducing more locking mechanisms */ 1048 if (mdev->csums_tfm) { 1049 digest_size = crypto_hash_digestsize(mdev->csums_tfm); 1050 D_ASSERT(digest_size == di->digest_size); 1051 digest = kmalloc(digest_size, GFP_NOIO); 1052 } 1053 if (digest) { 1054 drbd_csum_ee(mdev, mdev->csums_tfm, e, digest); 1055 eq = !memcmp(digest, di->digest, digest_size); 1056 kfree(digest); 1057 } 1058 1059 if (eq) { 1060 drbd_set_in_sync(mdev, e->sector, e->size); 1061 /* rs_same_csums unit is BM_BLOCK_SIZE */ 1062 mdev->rs_same_csum += e->size >> BM_BLOCK_SHIFT; 1063 ok = drbd_send_ack(mdev, P_RS_IS_IN_SYNC, e); 1064 } else { 1065 inc_rs_pending(mdev); 1066 e->block_id = ID_SYNCER; /* By setting block_id, digest pointer becomes invalid! */ 1067 e->flags &= ~EE_HAS_DIGEST; /* This e no longer has a digest pointer */ 1068 kfree(di); 1069 ok = drbd_send_block(mdev, P_RS_DATA_REPLY, e); 1070 } 1071 } else { 1072 ok = drbd_send_ack(mdev, P_NEG_RS_DREPLY, e); 1073 if (__ratelimit(&drbd_ratelimit_state)) 1074 dev_err(DEV, "Sending NegDReply. I guess it gets messy.\n"); 1075 } 1076 1077 dec_unacked(mdev); 1078 move_to_net_ee_or_free(mdev, e); 1079 1080 if (unlikely(!ok)) 1081 dev_err(DEV, "drbd_send_block/ack() failed\n"); 1082 return ok; 1083 } 1084 1085 int w_e_end_ov_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1086 { 1087 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w); 1088 int digest_size; 1089 void *digest; 1090 int ok = 1; 1091 1092 if (unlikely(cancel)) 1093 goto out; 1094 1095 if (unlikely((e->flags & EE_WAS_ERROR) != 0)) 1096 goto out; 1097 1098 digest_size = crypto_hash_digestsize(mdev->verify_tfm); 1099 /* FIXME if this allocation fails, online verify will not terminate! */ 1100 digest = kmalloc(digest_size, GFP_NOIO); 1101 if (digest) { 1102 drbd_csum_ee(mdev, mdev->verify_tfm, e, digest); 1103 inc_rs_pending(mdev); 1104 ok = drbd_send_drequest_csum(mdev, e->sector, e->size, 1105 digest, digest_size, P_OV_REPLY); 1106 if (!ok) 1107 dec_rs_pending(mdev); 1108 kfree(digest); 1109 } 1110 1111 out: 1112 drbd_free_ee(mdev, e); 1113 1114 dec_unacked(mdev); 1115 1116 return ok; 1117 } 1118 1119 void drbd_ov_oos_found(struct drbd_conf *mdev, sector_t sector, int size) 1120 { 1121 if (mdev->ov_last_oos_start + mdev->ov_last_oos_size == sector) { 1122 mdev->ov_last_oos_size += size>>9; 1123 } else { 1124 mdev->ov_last_oos_start = sector; 1125 mdev->ov_last_oos_size = size>>9; 1126 } 1127 drbd_set_out_of_sync(mdev, sector, size); 1128 set_bit(WRITE_BM_AFTER_RESYNC, &mdev->flags); 1129 } 1130 1131 int w_e_end_ov_reply(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1132 { 1133 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w); 1134 struct digest_info *di; 1135 int digest_size; 1136 void *digest; 1137 int ok, eq = 0; 1138 1139 if (unlikely(cancel)) { 1140 drbd_free_ee(mdev, e); 1141 dec_unacked(mdev); 1142 return 1; 1143 } 1144 1145 /* after "cancel", because after drbd_disconnect/drbd_rs_cancel_all 1146 * the resync lru has been cleaned up already */ 1147 if (get_ldev(mdev)) { 1148 drbd_rs_complete_io(mdev, e->sector); 1149 put_ldev(mdev); 1150 } 1151 1152 di = e->digest; 1153 1154 if (likely((e->flags & EE_WAS_ERROR) == 0)) { 1155 digest_size = crypto_hash_digestsize(mdev->verify_tfm); 1156 digest = kmalloc(digest_size, GFP_NOIO); 1157 if (digest) { 1158 drbd_csum_ee(mdev, mdev->verify_tfm, e, digest); 1159 1160 D_ASSERT(digest_size == di->digest_size); 1161 eq = !memcmp(digest, di->digest, digest_size); 1162 kfree(digest); 1163 } 1164 } else { 1165 ok = drbd_send_ack(mdev, P_NEG_RS_DREPLY, e); 1166 if (__ratelimit(&drbd_ratelimit_state)) 1167 dev_err(DEV, "Sending NegDReply. I guess it gets messy.\n"); 1168 } 1169 1170 dec_unacked(mdev); 1171 if (!eq) 1172 drbd_ov_oos_found(mdev, e->sector, e->size); 1173 else 1174 ov_oos_print(mdev); 1175 1176 ok = drbd_send_ack_ex(mdev, P_OV_RESULT, e->sector, e->size, 1177 eq ? ID_IN_SYNC : ID_OUT_OF_SYNC); 1178 1179 drbd_free_ee(mdev, e); 1180 1181 --mdev->ov_left; 1182 1183 /* let's advance progress step marks only for every other megabyte */ 1184 if ((mdev->ov_left & 0x200) == 0x200) 1185 drbd_advance_rs_marks(mdev, mdev->ov_left); 1186 1187 if (mdev->ov_left == 0) { 1188 ov_oos_print(mdev); 1189 drbd_resync_finished(mdev); 1190 } 1191 1192 return ok; 1193 } 1194 1195 int w_prev_work_done(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1196 { 1197 struct drbd_wq_barrier *b = container_of(w, struct drbd_wq_barrier, w); 1198 complete(&b->done); 1199 return 1; 1200 } 1201 1202 int w_send_barrier(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1203 { 1204 struct drbd_tl_epoch *b = container_of(w, struct drbd_tl_epoch, w); 1205 struct p_barrier *p = &mdev->data.sbuf.barrier; 1206 int ok = 1; 1207 1208 /* really avoid racing with tl_clear. w.cb may have been referenced 1209 * just before it was reassigned and re-queued, so double check that. 1210 * actually, this race was harmless, since we only try to send the 1211 * barrier packet here, and otherwise do nothing with the object. 1212 * but compare with the head of w_clear_epoch */ 1213 spin_lock_irq(&mdev->req_lock); 1214 if (w->cb != w_send_barrier || mdev->state.conn < C_CONNECTED) 1215 cancel = 1; 1216 spin_unlock_irq(&mdev->req_lock); 1217 if (cancel) 1218 return 1; 1219 1220 if (!drbd_get_data_sock(mdev)) 1221 return 0; 1222 p->barrier = b->br_number; 1223 /* inc_ap_pending was done where this was queued. 1224 * dec_ap_pending will be done in got_BarrierAck 1225 * or (on connection loss) in w_clear_epoch. */ 1226 ok = _drbd_send_cmd(mdev, mdev->data.socket, P_BARRIER, 1227 (struct p_header80 *)p, sizeof(*p), 0); 1228 drbd_put_data_sock(mdev); 1229 1230 return ok; 1231 } 1232 1233 int w_send_write_hint(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1234 { 1235 if (cancel) 1236 return 1; 1237 return drbd_send_short_cmd(mdev, P_UNPLUG_REMOTE); 1238 } 1239 1240 int w_send_oos(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1241 { 1242 struct drbd_request *req = container_of(w, struct drbd_request, w); 1243 int ok; 1244 1245 if (unlikely(cancel)) { 1246 req_mod(req, send_canceled); 1247 return 1; 1248 } 1249 1250 ok = drbd_send_oos(mdev, req); 1251 req_mod(req, oos_handed_to_network); 1252 1253 return ok; 1254 } 1255 1256 /** 1257 * w_send_dblock() - Worker callback to send a P_DATA packet in order to mirror a write request 1258 * @mdev: DRBD device. 1259 * @w: work object. 1260 * @cancel: The connection will be closed anyways 1261 */ 1262 int w_send_dblock(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1263 { 1264 struct drbd_request *req = container_of(w, struct drbd_request, w); 1265 int ok; 1266 1267 if (unlikely(cancel)) { 1268 req_mod(req, send_canceled); 1269 return 1; 1270 } 1271 1272 ok = drbd_send_dblock(mdev, req); 1273 req_mod(req, ok ? handed_over_to_network : send_failed); 1274 1275 return ok; 1276 } 1277 1278 /** 1279 * w_send_read_req() - Worker callback to send a read request (P_DATA_REQUEST) packet 1280 * @mdev: DRBD device. 1281 * @w: work object. 1282 * @cancel: The connection will be closed anyways 1283 */ 1284 int w_send_read_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1285 { 1286 struct drbd_request *req = container_of(w, struct drbd_request, w); 1287 int ok; 1288 1289 if (unlikely(cancel)) { 1290 req_mod(req, send_canceled); 1291 return 1; 1292 } 1293 1294 ok = drbd_send_drequest(mdev, P_DATA_REQUEST, req->sector, req->size, 1295 (unsigned long)req); 1296 1297 if (!ok) { 1298 /* ?? we set C_TIMEOUT or C_BROKEN_PIPE in drbd_send(); 1299 * so this is probably redundant */ 1300 if (mdev->state.conn >= C_CONNECTED) 1301 drbd_force_state(mdev, NS(conn, C_NETWORK_FAILURE)); 1302 } 1303 req_mod(req, ok ? handed_over_to_network : send_failed); 1304 1305 return ok; 1306 } 1307 1308 int w_restart_disk_io(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1309 { 1310 struct drbd_request *req = container_of(w, struct drbd_request, w); 1311 1312 if (bio_data_dir(req->master_bio) == WRITE && req->rq_state & RQ_IN_ACT_LOG) 1313 drbd_al_begin_io(mdev, req->sector); 1314 /* Calling drbd_al_begin_io() out of the worker might deadlocks 1315 theoretically. Practically it can not deadlock, since this is 1316 only used when unfreezing IOs. All the extents of the requests 1317 that made it into the TL are already active */ 1318 1319 drbd_req_make_private_bio(req, req->master_bio); 1320 req->private_bio->bi_bdev = mdev->ldev->backing_bdev; 1321 generic_make_request(req->private_bio); 1322 1323 return 1; 1324 } 1325 1326 static int _drbd_may_sync_now(struct drbd_conf *mdev) 1327 { 1328 struct drbd_conf *odev = mdev; 1329 1330 while (1) { 1331 if (odev->sync_conf.after == -1) 1332 return 1; 1333 odev = minor_to_mdev(odev->sync_conf.after); 1334 ERR_IF(!odev) return 1; 1335 if ((odev->state.conn >= C_SYNC_SOURCE && 1336 odev->state.conn <= C_PAUSED_SYNC_T) || 1337 odev->state.aftr_isp || odev->state.peer_isp || 1338 odev->state.user_isp) 1339 return 0; 1340 } 1341 } 1342 1343 /** 1344 * _drbd_pause_after() - Pause resync on all devices that may not resync now 1345 * @mdev: DRBD device. 1346 * 1347 * Called from process context only (admin command and after_state_ch). 1348 */ 1349 static int _drbd_pause_after(struct drbd_conf *mdev) 1350 { 1351 struct drbd_conf *odev; 1352 int i, rv = 0; 1353 1354 for (i = 0; i < minor_count; i++) { 1355 odev = minor_to_mdev(i); 1356 if (!odev) 1357 continue; 1358 if (odev->state.conn == C_STANDALONE && odev->state.disk == D_DISKLESS) 1359 continue; 1360 if (!_drbd_may_sync_now(odev)) 1361 rv |= (__drbd_set_state(_NS(odev, aftr_isp, 1), CS_HARD, NULL) 1362 != SS_NOTHING_TO_DO); 1363 } 1364 1365 return rv; 1366 } 1367 1368 /** 1369 * _drbd_resume_next() - Resume resync on all devices that may resync now 1370 * @mdev: DRBD device. 1371 * 1372 * Called from process context only (admin command and worker). 1373 */ 1374 static int _drbd_resume_next(struct drbd_conf *mdev) 1375 { 1376 struct drbd_conf *odev; 1377 int i, rv = 0; 1378 1379 for (i = 0; i < minor_count; i++) { 1380 odev = minor_to_mdev(i); 1381 if (!odev) 1382 continue; 1383 if (odev->state.conn == C_STANDALONE && odev->state.disk == D_DISKLESS) 1384 continue; 1385 if (odev->state.aftr_isp) { 1386 if (_drbd_may_sync_now(odev)) 1387 rv |= (__drbd_set_state(_NS(odev, aftr_isp, 0), 1388 CS_HARD, NULL) 1389 != SS_NOTHING_TO_DO) ; 1390 } 1391 } 1392 return rv; 1393 } 1394 1395 void resume_next_sg(struct drbd_conf *mdev) 1396 { 1397 write_lock_irq(&global_state_lock); 1398 _drbd_resume_next(mdev); 1399 write_unlock_irq(&global_state_lock); 1400 } 1401 1402 void suspend_other_sg(struct drbd_conf *mdev) 1403 { 1404 write_lock_irq(&global_state_lock); 1405 _drbd_pause_after(mdev); 1406 write_unlock_irq(&global_state_lock); 1407 } 1408 1409 static int sync_after_error(struct drbd_conf *mdev, int o_minor) 1410 { 1411 struct drbd_conf *odev; 1412 1413 if (o_minor == -1) 1414 return NO_ERROR; 1415 if (o_minor < -1 || minor_to_mdev(o_minor) == NULL) 1416 return ERR_SYNC_AFTER; 1417 1418 /* check for loops */ 1419 odev = minor_to_mdev(o_minor); 1420 while (1) { 1421 if (odev == mdev) 1422 return ERR_SYNC_AFTER_CYCLE; 1423 1424 /* dependency chain ends here, no cycles. */ 1425 if (odev->sync_conf.after == -1) 1426 return NO_ERROR; 1427 1428 /* follow the dependency chain */ 1429 odev = minor_to_mdev(odev->sync_conf.after); 1430 } 1431 } 1432 1433 int drbd_alter_sa(struct drbd_conf *mdev, int na) 1434 { 1435 int changes; 1436 int retcode; 1437 1438 write_lock_irq(&global_state_lock); 1439 retcode = sync_after_error(mdev, na); 1440 if (retcode == NO_ERROR) { 1441 mdev->sync_conf.after = na; 1442 do { 1443 changes = _drbd_pause_after(mdev); 1444 changes |= _drbd_resume_next(mdev); 1445 } while (changes); 1446 } 1447 write_unlock_irq(&global_state_lock); 1448 return retcode; 1449 } 1450 1451 void drbd_rs_controller_reset(struct drbd_conf *mdev) 1452 { 1453 atomic_set(&mdev->rs_sect_in, 0); 1454 atomic_set(&mdev->rs_sect_ev, 0); 1455 mdev->rs_in_flight = 0; 1456 mdev->rs_planed = 0; 1457 spin_lock(&mdev->peer_seq_lock); 1458 fifo_set(&mdev->rs_plan_s, 0); 1459 spin_unlock(&mdev->peer_seq_lock); 1460 } 1461 1462 /** 1463 * drbd_start_resync() - Start the resync process 1464 * @mdev: DRBD device. 1465 * @side: Either C_SYNC_SOURCE or C_SYNC_TARGET 1466 * 1467 * This function might bring you directly into one of the 1468 * C_PAUSED_SYNC_* states. 1469 */ 1470 void drbd_start_resync(struct drbd_conf *mdev, enum drbd_conns side) 1471 { 1472 union drbd_state ns; 1473 int r; 1474 1475 if (mdev->state.conn >= C_SYNC_SOURCE) { 1476 dev_err(DEV, "Resync already running!\n"); 1477 return; 1478 } 1479 1480 /* In case a previous resync run was aborted by an IO error/detach on the peer. */ 1481 drbd_rs_cancel_all(mdev); 1482 1483 if (side == C_SYNC_TARGET) { 1484 /* Since application IO was locked out during C_WF_BITMAP_T and 1485 C_WF_SYNC_UUID we are still unmodified. Before going to C_SYNC_TARGET 1486 we check that we might make the data inconsistent. */ 1487 r = drbd_khelper(mdev, "before-resync-target"); 1488 r = (r >> 8) & 0xff; 1489 if (r > 0) { 1490 dev_info(DEV, "before-resync-target handler returned %d, " 1491 "dropping connection.\n", r); 1492 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 1493 return; 1494 } 1495 } 1496 1497 drbd_state_lock(mdev); 1498 1499 if (!get_ldev_if_state(mdev, D_NEGOTIATING)) { 1500 drbd_state_unlock(mdev); 1501 return; 1502 } 1503 1504 if (side == C_SYNC_TARGET) { 1505 mdev->bm_resync_fo = 0; 1506 } else /* side == C_SYNC_SOURCE */ { 1507 u64 uuid; 1508 1509 get_random_bytes(&uuid, sizeof(u64)); 1510 drbd_uuid_set(mdev, UI_BITMAP, uuid); 1511 drbd_send_sync_uuid(mdev, uuid); 1512 1513 D_ASSERT(mdev->state.disk == D_UP_TO_DATE); 1514 } 1515 1516 write_lock_irq(&global_state_lock); 1517 ns = mdev->state; 1518 1519 ns.aftr_isp = !_drbd_may_sync_now(mdev); 1520 1521 ns.conn = side; 1522 1523 if (side == C_SYNC_TARGET) 1524 ns.disk = D_INCONSISTENT; 1525 else /* side == C_SYNC_SOURCE */ 1526 ns.pdsk = D_INCONSISTENT; 1527 1528 r = __drbd_set_state(mdev, ns, CS_VERBOSE, NULL); 1529 ns = mdev->state; 1530 1531 if (ns.conn < C_CONNECTED) 1532 r = SS_UNKNOWN_ERROR; 1533 1534 if (r == SS_SUCCESS) { 1535 unsigned long tw = drbd_bm_total_weight(mdev); 1536 unsigned long now = jiffies; 1537 int i; 1538 1539 mdev->rs_failed = 0; 1540 mdev->rs_paused = 0; 1541 mdev->rs_same_csum = 0; 1542 mdev->rs_last_events = 0; 1543 mdev->rs_last_sect_ev = 0; 1544 mdev->rs_total = tw; 1545 mdev->rs_start = now; 1546 for (i = 0; i < DRBD_SYNC_MARKS; i++) { 1547 mdev->rs_mark_left[i] = tw; 1548 mdev->rs_mark_time[i] = now; 1549 } 1550 _drbd_pause_after(mdev); 1551 } 1552 write_unlock_irq(&global_state_lock); 1553 put_ldev(mdev); 1554 1555 if (r == SS_SUCCESS) { 1556 dev_info(DEV, "Began resync as %s (will sync %lu KB [%lu bits set]).\n", 1557 drbd_conn_str(ns.conn), 1558 (unsigned long) mdev->rs_total << (BM_BLOCK_SHIFT-10), 1559 (unsigned long) mdev->rs_total); 1560 1561 if (mdev->agreed_pro_version < 95 && mdev->rs_total == 0) { 1562 /* This still has a race (about when exactly the peers 1563 * detect connection loss) that can lead to a full sync 1564 * on next handshake. In 8.3.9 we fixed this with explicit 1565 * resync-finished notifications, but the fix 1566 * introduces a protocol change. Sleeping for some 1567 * time longer than the ping interval + timeout on the 1568 * SyncSource, to give the SyncTarget the chance to 1569 * detect connection loss, then waiting for a ping 1570 * response (implicit in drbd_resync_finished) reduces 1571 * the race considerably, but does not solve it. */ 1572 if (side == C_SYNC_SOURCE) 1573 schedule_timeout_interruptible( 1574 mdev->net_conf->ping_int * HZ + 1575 mdev->net_conf->ping_timeo*HZ/9); 1576 drbd_resync_finished(mdev); 1577 } 1578 1579 drbd_rs_controller_reset(mdev); 1580 /* ns.conn may already be != mdev->state.conn, 1581 * we may have been paused in between, or become paused until 1582 * the timer triggers. 1583 * No matter, that is handled in resync_timer_fn() */ 1584 if (ns.conn == C_SYNC_TARGET) 1585 mod_timer(&mdev->resync_timer, jiffies); 1586 1587 drbd_md_sync(mdev); 1588 } 1589 drbd_state_unlock(mdev); 1590 } 1591 1592 int drbd_worker(struct drbd_thread *thi) 1593 { 1594 struct drbd_conf *mdev = thi->mdev; 1595 struct drbd_work *w = NULL; 1596 LIST_HEAD(work_list); 1597 int intr = 0, i; 1598 1599 sprintf(current->comm, "drbd%d_worker", mdev_to_minor(mdev)); 1600 1601 while (get_t_state(thi) == Running) { 1602 drbd_thread_current_set_cpu(mdev); 1603 1604 if (down_trylock(&mdev->data.work.s)) { 1605 mutex_lock(&mdev->data.mutex); 1606 if (mdev->data.socket && !mdev->net_conf->no_cork) 1607 drbd_tcp_uncork(mdev->data.socket); 1608 mutex_unlock(&mdev->data.mutex); 1609 1610 intr = down_interruptible(&mdev->data.work.s); 1611 1612 mutex_lock(&mdev->data.mutex); 1613 if (mdev->data.socket && !mdev->net_conf->no_cork) 1614 drbd_tcp_cork(mdev->data.socket); 1615 mutex_unlock(&mdev->data.mutex); 1616 } 1617 1618 if (intr) { 1619 D_ASSERT(intr == -EINTR); 1620 flush_signals(current); 1621 ERR_IF (get_t_state(thi) == Running) 1622 continue; 1623 break; 1624 } 1625 1626 if (get_t_state(thi) != Running) 1627 break; 1628 /* With this break, we have done a down() but not consumed 1629 the entry from the list. The cleanup code takes care of 1630 this... */ 1631 1632 w = NULL; 1633 spin_lock_irq(&mdev->data.work.q_lock); 1634 ERR_IF(list_empty(&mdev->data.work.q)) { 1635 /* something terribly wrong in our logic. 1636 * we were able to down() the semaphore, 1637 * but the list is empty... doh. 1638 * 1639 * what is the best thing to do now? 1640 * try again from scratch, restarting the receiver, 1641 * asender, whatnot? could break even more ugly, 1642 * e.g. when we are primary, but no good local data. 1643 * 1644 * I'll try to get away just starting over this loop. 1645 */ 1646 spin_unlock_irq(&mdev->data.work.q_lock); 1647 continue; 1648 } 1649 w = list_entry(mdev->data.work.q.next, struct drbd_work, list); 1650 list_del_init(&w->list); 1651 spin_unlock_irq(&mdev->data.work.q_lock); 1652 1653 if (!w->cb(mdev, w, mdev->state.conn < C_CONNECTED)) { 1654 /* dev_warn(DEV, "worker: a callback failed! \n"); */ 1655 if (mdev->state.conn >= C_CONNECTED) 1656 drbd_force_state(mdev, 1657 NS(conn, C_NETWORK_FAILURE)); 1658 } 1659 } 1660 D_ASSERT(test_bit(DEVICE_DYING, &mdev->flags)); 1661 D_ASSERT(test_bit(CONFIG_PENDING, &mdev->flags)); 1662 1663 spin_lock_irq(&mdev->data.work.q_lock); 1664 i = 0; 1665 while (!list_empty(&mdev->data.work.q)) { 1666 list_splice_init(&mdev->data.work.q, &work_list); 1667 spin_unlock_irq(&mdev->data.work.q_lock); 1668 1669 while (!list_empty(&work_list)) { 1670 w = list_entry(work_list.next, struct drbd_work, list); 1671 list_del_init(&w->list); 1672 w->cb(mdev, w, 1); 1673 i++; /* dead debugging code */ 1674 } 1675 1676 spin_lock_irq(&mdev->data.work.q_lock); 1677 } 1678 sema_init(&mdev->data.work.s, 0); 1679 /* DANGEROUS race: if someone did queue his work within the spinlock, 1680 * but up() ed outside the spinlock, we could get an up() on the 1681 * semaphore without corresponding list entry. 1682 * So don't do that. 1683 */ 1684 spin_unlock_irq(&mdev->data.work.q_lock); 1685 1686 D_ASSERT(mdev->state.disk == D_DISKLESS && mdev->state.conn == C_STANDALONE); 1687 /* _drbd_set_state only uses stop_nowait. 1688 * wait here for the Exiting receiver. */ 1689 drbd_thread_stop(&mdev->receiver); 1690 drbd_mdev_cleanup(mdev); 1691 1692 dev_info(DEV, "worker terminated\n"); 1693 1694 clear_bit(DEVICE_DYING, &mdev->flags); 1695 clear_bit(CONFIG_PENDING, &mdev->flags); 1696 wake_up(&mdev->state_wait); 1697 1698 return 0; 1699 } 1700