1 /* 2 * Copyright (C) 2003 Sistina Software Limited. 3 * Copyright (C) 2005-2008 Red Hat, Inc. All rights reserved. 4 * 5 * This file is released under the GPL. 6 */ 7 8 #include "dm-bio-record.h" 9 10 #include <linux/init.h> 11 #include <linux/mempool.h> 12 #include <linux/module.h> 13 #include <linux/pagemap.h> 14 #include <linux/slab.h> 15 #include <linux/workqueue.h> 16 #include <linux/device-mapper.h> 17 #include <linux/dm-io.h> 18 #include <linux/dm-dirty-log.h> 19 #include <linux/dm-kcopyd.h> 20 #include <linux/dm-region-hash.h> 21 22 #define DM_MSG_PREFIX "raid1" 23 24 #define MAX_RECOVERY 1 /* Maximum number of regions recovered in parallel. */ 25 26 #define DM_RAID1_HANDLE_ERRORS 0x01 27 #define errors_handled(p) ((p)->features & DM_RAID1_HANDLE_ERRORS) 28 29 static DECLARE_WAIT_QUEUE_HEAD(_kmirrord_recovery_stopped); 30 31 /*----------------------------------------------------------------- 32 * Mirror set structures. 33 *---------------------------------------------------------------*/ 34 enum dm_raid1_error { 35 DM_RAID1_WRITE_ERROR, 36 DM_RAID1_FLUSH_ERROR, 37 DM_RAID1_SYNC_ERROR, 38 DM_RAID1_READ_ERROR 39 }; 40 41 struct mirror { 42 struct mirror_set *ms; 43 atomic_t error_count; 44 unsigned long error_type; 45 struct dm_dev *dev; 46 sector_t offset; 47 }; 48 49 struct mirror_set { 50 struct dm_target *ti; 51 struct list_head list; 52 53 uint64_t features; 54 55 spinlock_t lock; /* protects the lists */ 56 struct bio_list reads; 57 struct bio_list writes; 58 struct bio_list failures; 59 struct bio_list holds; /* bios are waiting until suspend */ 60 61 struct dm_region_hash *rh; 62 struct dm_kcopyd_client *kcopyd_client; 63 struct dm_io_client *io_client; 64 mempool_t *read_record_pool; 65 66 /* recovery */ 67 region_t nr_regions; 68 int in_sync; 69 int log_failure; 70 int leg_failure; 71 atomic_t suspend; 72 73 atomic_t default_mirror; /* Default mirror */ 74 75 struct workqueue_struct *kmirrord_wq; 76 struct work_struct kmirrord_work; 77 struct timer_list timer; 78 unsigned long timer_pending; 79 80 struct work_struct trigger_event; 81 82 unsigned nr_mirrors; 83 struct mirror mirror[0]; 84 }; 85 86 static void wakeup_mirrord(void *context) 87 { 88 struct mirror_set *ms = context; 89 90 queue_work(ms->kmirrord_wq, &ms->kmirrord_work); 91 } 92 93 static void delayed_wake_fn(unsigned long data) 94 { 95 struct mirror_set *ms = (struct mirror_set *) data; 96 97 clear_bit(0, &ms->timer_pending); 98 wakeup_mirrord(ms); 99 } 100 101 static void delayed_wake(struct mirror_set *ms) 102 { 103 if (test_and_set_bit(0, &ms->timer_pending)) 104 return; 105 106 ms->timer.expires = jiffies + HZ / 5; 107 ms->timer.data = (unsigned long) ms; 108 ms->timer.function = delayed_wake_fn; 109 add_timer(&ms->timer); 110 } 111 112 static void wakeup_all_recovery_waiters(void *context) 113 { 114 wake_up_all(&_kmirrord_recovery_stopped); 115 } 116 117 static void queue_bio(struct mirror_set *ms, struct bio *bio, int rw) 118 { 119 unsigned long flags; 120 int should_wake = 0; 121 struct bio_list *bl; 122 123 bl = (rw == WRITE) ? &ms->writes : &ms->reads; 124 spin_lock_irqsave(&ms->lock, flags); 125 should_wake = !(bl->head); 126 bio_list_add(bl, bio); 127 spin_unlock_irqrestore(&ms->lock, flags); 128 129 if (should_wake) 130 wakeup_mirrord(ms); 131 } 132 133 static void dispatch_bios(void *context, struct bio_list *bio_list) 134 { 135 struct mirror_set *ms = context; 136 struct bio *bio; 137 138 while ((bio = bio_list_pop(bio_list))) 139 queue_bio(ms, bio, WRITE); 140 } 141 142 #define MIN_READ_RECORDS 20 143 struct dm_raid1_read_record { 144 struct mirror *m; 145 struct dm_bio_details details; 146 }; 147 148 static struct kmem_cache *_dm_raid1_read_record_cache; 149 150 /* 151 * Every mirror should look like this one. 152 */ 153 #define DEFAULT_MIRROR 0 154 155 /* 156 * This is yucky. We squirrel the mirror struct away inside 157 * bi_next for read/write buffers. This is safe since the bh 158 * doesn't get submitted to the lower levels of block layer. 159 */ 160 static struct mirror *bio_get_m(struct bio *bio) 161 { 162 return (struct mirror *) bio->bi_next; 163 } 164 165 static void bio_set_m(struct bio *bio, struct mirror *m) 166 { 167 bio->bi_next = (struct bio *) m; 168 } 169 170 static struct mirror *get_default_mirror(struct mirror_set *ms) 171 { 172 return &ms->mirror[atomic_read(&ms->default_mirror)]; 173 } 174 175 static void set_default_mirror(struct mirror *m) 176 { 177 struct mirror_set *ms = m->ms; 178 struct mirror *m0 = &(ms->mirror[0]); 179 180 atomic_set(&ms->default_mirror, m - m0); 181 } 182 183 static struct mirror *get_valid_mirror(struct mirror_set *ms) 184 { 185 struct mirror *m; 186 187 for (m = ms->mirror; m < ms->mirror + ms->nr_mirrors; m++) 188 if (!atomic_read(&m->error_count)) 189 return m; 190 191 return NULL; 192 } 193 194 /* fail_mirror 195 * @m: mirror device to fail 196 * @error_type: one of the enum's, DM_RAID1_*_ERROR 197 * 198 * If errors are being handled, record the type of 199 * error encountered for this device. If this type 200 * of error has already been recorded, we can return; 201 * otherwise, we must signal userspace by triggering 202 * an event. Additionally, if the device is the 203 * primary device, we must choose a new primary, but 204 * only if the mirror is in-sync. 205 * 206 * This function must not block. 207 */ 208 static void fail_mirror(struct mirror *m, enum dm_raid1_error error_type) 209 { 210 struct mirror_set *ms = m->ms; 211 struct mirror *new; 212 213 ms->leg_failure = 1; 214 215 /* 216 * error_count is used for nothing more than a 217 * simple way to tell if a device has encountered 218 * errors. 219 */ 220 atomic_inc(&m->error_count); 221 222 if (test_and_set_bit(error_type, &m->error_type)) 223 return; 224 225 if (!errors_handled(ms)) 226 return; 227 228 if (m != get_default_mirror(ms)) 229 goto out; 230 231 if (!ms->in_sync) { 232 /* 233 * Better to issue requests to same failing device 234 * than to risk returning corrupt data. 235 */ 236 DMERR("Primary mirror (%s) failed while out-of-sync: " 237 "Reads may fail.", m->dev->name); 238 goto out; 239 } 240 241 new = get_valid_mirror(ms); 242 if (new) 243 set_default_mirror(new); 244 else 245 DMWARN("All sides of mirror have failed."); 246 247 out: 248 schedule_work(&ms->trigger_event); 249 } 250 251 static int mirror_flush(struct dm_target *ti) 252 { 253 struct mirror_set *ms = ti->private; 254 unsigned long error_bits; 255 256 unsigned int i; 257 struct dm_io_region io[ms->nr_mirrors]; 258 struct mirror *m; 259 struct dm_io_request io_req = { 260 .bi_rw = WRITE_FLUSH, 261 .mem.type = DM_IO_KMEM, 262 .mem.ptr.addr = NULL, 263 .client = ms->io_client, 264 }; 265 266 for (i = 0, m = ms->mirror; i < ms->nr_mirrors; i++, m++) { 267 io[i].bdev = m->dev->bdev; 268 io[i].sector = 0; 269 io[i].count = 0; 270 } 271 272 error_bits = -1; 273 dm_io(&io_req, ms->nr_mirrors, io, &error_bits); 274 if (unlikely(error_bits != 0)) { 275 for (i = 0; i < ms->nr_mirrors; i++) 276 if (test_bit(i, &error_bits)) 277 fail_mirror(ms->mirror + i, 278 DM_RAID1_FLUSH_ERROR); 279 return -EIO; 280 } 281 282 return 0; 283 } 284 285 /*----------------------------------------------------------------- 286 * Recovery. 287 * 288 * When a mirror is first activated we may find that some regions 289 * are in the no-sync state. We have to recover these by 290 * recopying from the default mirror to all the others. 291 *---------------------------------------------------------------*/ 292 static void recovery_complete(int read_err, unsigned long write_err, 293 void *context) 294 { 295 struct dm_region *reg = context; 296 struct mirror_set *ms = dm_rh_region_context(reg); 297 int m, bit = 0; 298 299 if (read_err) { 300 /* Read error means the failure of default mirror. */ 301 DMERR_LIMIT("Unable to read primary mirror during recovery"); 302 fail_mirror(get_default_mirror(ms), DM_RAID1_SYNC_ERROR); 303 } 304 305 if (write_err) { 306 DMERR_LIMIT("Write error during recovery (error = 0x%lx)", 307 write_err); 308 /* 309 * Bits correspond to devices (excluding default mirror). 310 * The default mirror cannot change during recovery. 311 */ 312 for (m = 0; m < ms->nr_mirrors; m++) { 313 if (&ms->mirror[m] == get_default_mirror(ms)) 314 continue; 315 if (test_bit(bit, &write_err)) 316 fail_mirror(ms->mirror + m, 317 DM_RAID1_SYNC_ERROR); 318 bit++; 319 } 320 } 321 322 dm_rh_recovery_end(reg, !(read_err || write_err)); 323 } 324 325 static int recover(struct mirror_set *ms, struct dm_region *reg) 326 { 327 int r; 328 unsigned i; 329 struct dm_io_region from, to[DM_KCOPYD_MAX_REGIONS], *dest; 330 struct mirror *m; 331 unsigned long flags = 0; 332 region_t key = dm_rh_get_region_key(reg); 333 sector_t region_size = dm_rh_get_region_size(ms->rh); 334 335 /* fill in the source */ 336 m = get_default_mirror(ms); 337 from.bdev = m->dev->bdev; 338 from.sector = m->offset + dm_rh_region_to_sector(ms->rh, key); 339 if (key == (ms->nr_regions - 1)) { 340 /* 341 * The final region may be smaller than 342 * region_size. 343 */ 344 from.count = ms->ti->len & (region_size - 1); 345 if (!from.count) 346 from.count = region_size; 347 } else 348 from.count = region_size; 349 350 /* fill in the destinations */ 351 for (i = 0, dest = to; i < ms->nr_mirrors; i++) { 352 if (&ms->mirror[i] == get_default_mirror(ms)) 353 continue; 354 355 m = ms->mirror + i; 356 dest->bdev = m->dev->bdev; 357 dest->sector = m->offset + dm_rh_region_to_sector(ms->rh, key); 358 dest->count = from.count; 359 dest++; 360 } 361 362 /* hand to kcopyd */ 363 if (!errors_handled(ms)) 364 set_bit(DM_KCOPYD_IGNORE_ERROR, &flags); 365 366 r = dm_kcopyd_copy(ms->kcopyd_client, &from, ms->nr_mirrors - 1, to, 367 flags, recovery_complete, reg); 368 369 return r; 370 } 371 372 static void do_recovery(struct mirror_set *ms) 373 { 374 struct dm_region *reg; 375 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 376 int r; 377 378 /* 379 * Start quiescing some regions. 380 */ 381 dm_rh_recovery_prepare(ms->rh); 382 383 /* 384 * Copy any already quiesced regions. 385 */ 386 while ((reg = dm_rh_recovery_start(ms->rh))) { 387 r = recover(ms, reg); 388 if (r) 389 dm_rh_recovery_end(reg, 0); 390 } 391 392 /* 393 * Update the in sync flag. 394 */ 395 if (!ms->in_sync && 396 (log->type->get_sync_count(log) == ms->nr_regions)) { 397 /* the sync is complete */ 398 dm_table_event(ms->ti->table); 399 ms->in_sync = 1; 400 } 401 } 402 403 /*----------------------------------------------------------------- 404 * Reads 405 *---------------------------------------------------------------*/ 406 static struct mirror *choose_mirror(struct mirror_set *ms, sector_t sector) 407 { 408 struct mirror *m = get_default_mirror(ms); 409 410 do { 411 if (likely(!atomic_read(&m->error_count))) 412 return m; 413 414 if (m-- == ms->mirror) 415 m += ms->nr_mirrors; 416 } while (m != get_default_mirror(ms)); 417 418 return NULL; 419 } 420 421 static int default_ok(struct mirror *m) 422 { 423 struct mirror *default_mirror = get_default_mirror(m->ms); 424 425 return !atomic_read(&default_mirror->error_count); 426 } 427 428 static int mirror_available(struct mirror_set *ms, struct bio *bio) 429 { 430 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 431 region_t region = dm_rh_bio_to_region(ms->rh, bio); 432 433 if (log->type->in_sync(log, region, 0)) 434 return choose_mirror(ms, bio->bi_sector) ? 1 : 0; 435 436 return 0; 437 } 438 439 /* 440 * remap a buffer to a particular mirror. 441 */ 442 static sector_t map_sector(struct mirror *m, struct bio *bio) 443 { 444 if (unlikely(!bio->bi_size)) 445 return 0; 446 return m->offset + dm_target_offset(m->ms->ti, bio->bi_sector); 447 } 448 449 static void map_bio(struct mirror *m, struct bio *bio) 450 { 451 bio->bi_bdev = m->dev->bdev; 452 bio->bi_sector = map_sector(m, bio); 453 } 454 455 static void map_region(struct dm_io_region *io, struct mirror *m, 456 struct bio *bio) 457 { 458 io->bdev = m->dev->bdev; 459 io->sector = map_sector(m, bio); 460 io->count = bio->bi_size >> 9; 461 } 462 463 static void hold_bio(struct mirror_set *ms, struct bio *bio) 464 { 465 /* 466 * Lock is required to avoid race condition during suspend 467 * process. 468 */ 469 spin_lock_irq(&ms->lock); 470 471 if (atomic_read(&ms->suspend)) { 472 spin_unlock_irq(&ms->lock); 473 474 /* 475 * If device is suspended, complete the bio. 476 */ 477 if (dm_noflush_suspending(ms->ti)) 478 bio_endio(bio, DM_ENDIO_REQUEUE); 479 else 480 bio_endio(bio, -EIO); 481 return; 482 } 483 484 /* 485 * Hold bio until the suspend is complete. 486 */ 487 bio_list_add(&ms->holds, bio); 488 spin_unlock_irq(&ms->lock); 489 } 490 491 /*----------------------------------------------------------------- 492 * Reads 493 *---------------------------------------------------------------*/ 494 static void read_callback(unsigned long error, void *context) 495 { 496 struct bio *bio = context; 497 struct mirror *m; 498 499 m = bio_get_m(bio); 500 bio_set_m(bio, NULL); 501 502 if (likely(!error)) { 503 bio_endio(bio, 0); 504 return; 505 } 506 507 fail_mirror(m, DM_RAID1_READ_ERROR); 508 509 if (likely(default_ok(m)) || mirror_available(m->ms, bio)) { 510 DMWARN_LIMIT("Read failure on mirror device %s. " 511 "Trying alternative device.", 512 m->dev->name); 513 queue_bio(m->ms, bio, bio_rw(bio)); 514 return; 515 } 516 517 DMERR_LIMIT("Read failure on mirror device %s. Failing I/O.", 518 m->dev->name); 519 bio_endio(bio, -EIO); 520 } 521 522 /* Asynchronous read. */ 523 static void read_async_bio(struct mirror *m, struct bio *bio) 524 { 525 struct dm_io_region io; 526 struct dm_io_request io_req = { 527 .bi_rw = READ, 528 .mem.type = DM_IO_BVEC, 529 .mem.ptr.bvec = bio->bi_io_vec + bio->bi_idx, 530 .notify.fn = read_callback, 531 .notify.context = bio, 532 .client = m->ms->io_client, 533 }; 534 535 map_region(&io, m, bio); 536 bio_set_m(bio, m); 537 BUG_ON(dm_io(&io_req, 1, &io, NULL)); 538 } 539 540 static inline int region_in_sync(struct mirror_set *ms, region_t region, 541 int may_block) 542 { 543 int state = dm_rh_get_state(ms->rh, region, may_block); 544 return state == DM_RH_CLEAN || state == DM_RH_DIRTY; 545 } 546 547 static void do_reads(struct mirror_set *ms, struct bio_list *reads) 548 { 549 region_t region; 550 struct bio *bio; 551 struct mirror *m; 552 553 while ((bio = bio_list_pop(reads))) { 554 region = dm_rh_bio_to_region(ms->rh, bio); 555 m = get_default_mirror(ms); 556 557 /* 558 * We can only read balance if the region is in sync. 559 */ 560 if (likely(region_in_sync(ms, region, 1))) 561 m = choose_mirror(ms, bio->bi_sector); 562 else if (m && atomic_read(&m->error_count)) 563 m = NULL; 564 565 if (likely(m)) 566 read_async_bio(m, bio); 567 else 568 bio_endio(bio, -EIO); 569 } 570 } 571 572 /*----------------------------------------------------------------- 573 * Writes. 574 * 575 * We do different things with the write io depending on the 576 * state of the region that it's in: 577 * 578 * SYNC: increment pending, use kcopyd to write to *all* mirrors 579 * RECOVERING: delay the io until recovery completes 580 * NOSYNC: increment pending, just write to the default mirror 581 *---------------------------------------------------------------*/ 582 583 584 static void write_callback(unsigned long error, void *context) 585 { 586 unsigned i, ret = 0; 587 struct bio *bio = (struct bio *) context; 588 struct mirror_set *ms; 589 int should_wake = 0; 590 unsigned long flags; 591 592 ms = bio_get_m(bio)->ms; 593 bio_set_m(bio, NULL); 594 595 /* 596 * NOTE: We don't decrement the pending count here, 597 * instead it is done by the targets endio function. 598 * This way we handle both writes to SYNC and NOSYNC 599 * regions with the same code. 600 */ 601 if (likely(!error)) { 602 bio_endio(bio, ret); 603 return; 604 } 605 606 for (i = 0; i < ms->nr_mirrors; i++) 607 if (test_bit(i, &error)) 608 fail_mirror(ms->mirror + i, DM_RAID1_WRITE_ERROR); 609 610 /* 611 * Need to raise event. Since raising 612 * events can block, we need to do it in 613 * the main thread. 614 */ 615 spin_lock_irqsave(&ms->lock, flags); 616 if (!ms->failures.head) 617 should_wake = 1; 618 bio_list_add(&ms->failures, bio); 619 spin_unlock_irqrestore(&ms->lock, flags); 620 if (should_wake) 621 wakeup_mirrord(ms); 622 } 623 624 static void do_write(struct mirror_set *ms, struct bio *bio) 625 { 626 unsigned int i; 627 struct dm_io_region io[ms->nr_mirrors], *dest = io; 628 struct mirror *m; 629 struct dm_io_request io_req = { 630 .bi_rw = WRITE | (bio->bi_rw & WRITE_FLUSH_FUA), 631 .mem.type = DM_IO_BVEC, 632 .mem.ptr.bvec = bio->bi_io_vec + bio->bi_idx, 633 .notify.fn = write_callback, 634 .notify.context = bio, 635 .client = ms->io_client, 636 }; 637 638 if (bio->bi_rw & REQ_DISCARD) { 639 io_req.bi_rw |= REQ_DISCARD; 640 io_req.mem.type = DM_IO_KMEM; 641 io_req.mem.ptr.addr = NULL; 642 } 643 644 for (i = 0, m = ms->mirror; i < ms->nr_mirrors; i++, m++) 645 map_region(dest++, m, bio); 646 647 /* 648 * Use default mirror because we only need it to retrieve the reference 649 * to the mirror set in write_callback(). 650 */ 651 bio_set_m(bio, get_default_mirror(ms)); 652 653 BUG_ON(dm_io(&io_req, ms->nr_mirrors, io, NULL)); 654 } 655 656 static void do_writes(struct mirror_set *ms, struct bio_list *writes) 657 { 658 int state; 659 struct bio *bio; 660 struct bio_list sync, nosync, recover, *this_list = NULL; 661 struct bio_list requeue; 662 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 663 region_t region; 664 665 if (!writes->head) 666 return; 667 668 /* 669 * Classify each write. 670 */ 671 bio_list_init(&sync); 672 bio_list_init(&nosync); 673 bio_list_init(&recover); 674 bio_list_init(&requeue); 675 676 while ((bio = bio_list_pop(writes))) { 677 if ((bio->bi_rw & REQ_FLUSH) || 678 (bio->bi_rw & REQ_DISCARD)) { 679 bio_list_add(&sync, bio); 680 continue; 681 } 682 683 region = dm_rh_bio_to_region(ms->rh, bio); 684 685 if (log->type->is_remote_recovering && 686 log->type->is_remote_recovering(log, region)) { 687 bio_list_add(&requeue, bio); 688 continue; 689 } 690 691 state = dm_rh_get_state(ms->rh, region, 1); 692 switch (state) { 693 case DM_RH_CLEAN: 694 case DM_RH_DIRTY: 695 this_list = &sync; 696 break; 697 698 case DM_RH_NOSYNC: 699 this_list = &nosync; 700 break; 701 702 case DM_RH_RECOVERING: 703 this_list = &recover; 704 break; 705 } 706 707 bio_list_add(this_list, bio); 708 } 709 710 /* 711 * Add bios that are delayed due to remote recovery 712 * back on to the write queue 713 */ 714 if (unlikely(requeue.head)) { 715 spin_lock_irq(&ms->lock); 716 bio_list_merge(&ms->writes, &requeue); 717 spin_unlock_irq(&ms->lock); 718 delayed_wake(ms); 719 } 720 721 /* 722 * Increment the pending counts for any regions that will 723 * be written to (writes to recover regions are going to 724 * be delayed). 725 */ 726 dm_rh_inc_pending(ms->rh, &sync); 727 dm_rh_inc_pending(ms->rh, &nosync); 728 729 /* 730 * If the flush fails on a previous call and succeeds here, 731 * we must not reset the log_failure variable. We need 732 * userspace interaction to do that. 733 */ 734 ms->log_failure = dm_rh_flush(ms->rh) ? 1 : ms->log_failure; 735 736 /* 737 * Dispatch io. 738 */ 739 if (unlikely(ms->log_failure) && errors_handled(ms)) { 740 spin_lock_irq(&ms->lock); 741 bio_list_merge(&ms->failures, &sync); 742 spin_unlock_irq(&ms->lock); 743 wakeup_mirrord(ms); 744 } else 745 while ((bio = bio_list_pop(&sync))) 746 do_write(ms, bio); 747 748 while ((bio = bio_list_pop(&recover))) 749 dm_rh_delay(ms->rh, bio); 750 751 while ((bio = bio_list_pop(&nosync))) { 752 if (unlikely(ms->leg_failure) && errors_handled(ms)) { 753 spin_lock_irq(&ms->lock); 754 bio_list_add(&ms->failures, bio); 755 spin_unlock_irq(&ms->lock); 756 wakeup_mirrord(ms); 757 } else { 758 map_bio(get_default_mirror(ms), bio); 759 generic_make_request(bio); 760 } 761 } 762 } 763 764 static void do_failures(struct mirror_set *ms, struct bio_list *failures) 765 { 766 struct bio *bio; 767 768 if (likely(!failures->head)) 769 return; 770 771 /* 772 * If the log has failed, unattempted writes are being 773 * put on the holds list. We can't issue those writes 774 * until a log has been marked, so we must store them. 775 * 776 * If a 'noflush' suspend is in progress, we can requeue 777 * the I/O's to the core. This give userspace a chance 778 * to reconfigure the mirror, at which point the core 779 * will reissue the writes. If the 'noflush' flag is 780 * not set, we have no choice but to return errors. 781 * 782 * Some writes on the failures list may have been 783 * submitted before the log failure and represent a 784 * failure to write to one of the devices. It is ok 785 * for us to treat them the same and requeue them 786 * as well. 787 */ 788 while ((bio = bio_list_pop(failures))) { 789 if (!ms->log_failure) { 790 ms->in_sync = 0; 791 dm_rh_mark_nosync(ms->rh, bio); 792 } 793 794 /* 795 * If all the legs are dead, fail the I/O. 796 * If we have been told to handle errors, hold the bio 797 * and wait for userspace to deal with the problem. 798 * Otherwise pretend that the I/O succeeded. (This would 799 * be wrong if the failed leg returned after reboot and 800 * got replicated back to the good legs.) 801 */ 802 if (!get_valid_mirror(ms)) 803 bio_endio(bio, -EIO); 804 else if (errors_handled(ms)) 805 hold_bio(ms, bio); 806 else 807 bio_endio(bio, 0); 808 } 809 } 810 811 static void trigger_event(struct work_struct *work) 812 { 813 struct mirror_set *ms = 814 container_of(work, struct mirror_set, trigger_event); 815 816 dm_table_event(ms->ti->table); 817 } 818 819 /*----------------------------------------------------------------- 820 * kmirrord 821 *---------------------------------------------------------------*/ 822 static void do_mirror(struct work_struct *work) 823 { 824 struct mirror_set *ms = container_of(work, struct mirror_set, 825 kmirrord_work); 826 struct bio_list reads, writes, failures; 827 unsigned long flags; 828 829 spin_lock_irqsave(&ms->lock, flags); 830 reads = ms->reads; 831 writes = ms->writes; 832 failures = ms->failures; 833 bio_list_init(&ms->reads); 834 bio_list_init(&ms->writes); 835 bio_list_init(&ms->failures); 836 spin_unlock_irqrestore(&ms->lock, flags); 837 838 dm_rh_update_states(ms->rh, errors_handled(ms)); 839 do_recovery(ms); 840 do_reads(ms, &reads); 841 do_writes(ms, &writes); 842 do_failures(ms, &failures); 843 } 844 845 /*----------------------------------------------------------------- 846 * Target functions 847 *---------------------------------------------------------------*/ 848 static struct mirror_set *alloc_context(unsigned int nr_mirrors, 849 uint32_t region_size, 850 struct dm_target *ti, 851 struct dm_dirty_log *dl) 852 { 853 size_t len; 854 struct mirror_set *ms = NULL; 855 856 len = sizeof(*ms) + (sizeof(ms->mirror[0]) * nr_mirrors); 857 858 ms = kzalloc(len, GFP_KERNEL); 859 if (!ms) { 860 ti->error = "Cannot allocate mirror context"; 861 return NULL; 862 } 863 864 spin_lock_init(&ms->lock); 865 bio_list_init(&ms->reads); 866 bio_list_init(&ms->writes); 867 bio_list_init(&ms->failures); 868 bio_list_init(&ms->holds); 869 870 ms->ti = ti; 871 ms->nr_mirrors = nr_mirrors; 872 ms->nr_regions = dm_sector_div_up(ti->len, region_size); 873 ms->in_sync = 0; 874 ms->log_failure = 0; 875 ms->leg_failure = 0; 876 atomic_set(&ms->suspend, 0); 877 atomic_set(&ms->default_mirror, DEFAULT_MIRROR); 878 879 ms->read_record_pool = mempool_create_slab_pool(MIN_READ_RECORDS, 880 _dm_raid1_read_record_cache); 881 882 if (!ms->read_record_pool) { 883 ti->error = "Error creating mirror read_record_pool"; 884 kfree(ms); 885 return NULL; 886 } 887 888 ms->io_client = dm_io_client_create(); 889 if (IS_ERR(ms->io_client)) { 890 ti->error = "Error creating dm_io client"; 891 mempool_destroy(ms->read_record_pool); 892 kfree(ms); 893 return NULL; 894 } 895 896 ms->rh = dm_region_hash_create(ms, dispatch_bios, wakeup_mirrord, 897 wakeup_all_recovery_waiters, 898 ms->ti->begin, MAX_RECOVERY, 899 dl, region_size, ms->nr_regions); 900 if (IS_ERR(ms->rh)) { 901 ti->error = "Error creating dirty region hash"; 902 dm_io_client_destroy(ms->io_client); 903 mempool_destroy(ms->read_record_pool); 904 kfree(ms); 905 return NULL; 906 } 907 908 return ms; 909 } 910 911 static void free_context(struct mirror_set *ms, struct dm_target *ti, 912 unsigned int m) 913 { 914 while (m--) 915 dm_put_device(ti, ms->mirror[m].dev); 916 917 dm_io_client_destroy(ms->io_client); 918 dm_region_hash_destroy(ms->rh); 919 mempool_destroy(ms->read_record_pool); 920 kfree(ms); 921 } 922 923 static int get_mirror(struct mirror_set *ms, struct dm_target *ti, 924 unsigned int mirror, char **argv) 925 { 926 unsigned long long offset; 927 928 if (sscanf(argv[1], "%llu", &offset) != 1) { 929 ti->error = "Invalid offset"; 930 return -EINVAL; 931 } 932 933 if (dm_get_device(ti, argv[0], dm_table_get_mode(ti->table), 934 &ms->mirror[mirror].dev)) { 935 ti->error = "Device lookup failure"; 936 return -ENXIO; 937 } 938 939 ms->mirror[mirror].ms = ms; 940 atomic_set(&(ms->mirror[mirror].error_count), 0); 941 ms->mirror[mirror].error_type = 0; 942 ms->mirror[mirror].offset = offset; 943 944 return 0; 945 } 946 947 /* 948 * Create dirty log: log_type #log_params <log_params> 949 */ 950 static struct dm_dirty_log *create_dirty_log(struct dm_target *ti, 951 unsigned argc, char **argv, 952 unsigned *args_used) 953 { 954 unsigned param_count; 955 struct dm_dirty_log *dl; 956 957 if (argc < 2) { 958 ti->error = "Insufficient mirror log arguments"; 959 return NULL; 960 } 961 962 if (sscanf(argv[1], "%u", ¶m_count) != 1) { 963 ti->error = "Invalid mirror log argument count"; 964 return NULL; 965 } 966 967 *args_used = 2 + param_count; 968 969 if (argc < *args_used) { 970 ti->error = "Insufficient mirror log arguments"; 971 return NULL; 972 } 973 974 dl = dm_dirty_log_create(argv[0], ti, mirror_flush, param_count, 975 argv + 2); 976 if (!dl) { 977 ti->error = "Error creating mirror dirty log"; 978 return NULL; 979 } 980 981 return dl; 982 } 983 984 static int parse_features(struct mirror_set *ms, unsigned argc, char **argv, 985 unsigned *args_used) 986 { 987 unsigned num_features; 988 struct dm_target *ti = ms->ti; 989 990 *args_used = 0; 991 992 if (!argc) 993 return 0; 994 995 if (sscanf(argv[0], "%u", &num_features) != 1) { 996 ti->error = "Invalid number of features"; 997 return -EINVAL; 998 } 999 1000 argc--; 1001 argv++; 1002 (*args_used)++; 1003 1004 if (num_features > argc) { 1005 ti->error = "Not enough arguments to support feature count"; 1006 return -EINVAL; 1007 } 1008 1009 if (!strcmp("handle_errors", argv[0])) 1010 ms->features |= DM_RAID1_HANDLE_ERRORS; 1011 else { 1012 ti->error = "Unrecognised feature requested"; 1013 return -EINVAL; 1014 } 1015 1016 (*args_used)++; 1017 1018 return 0; 1019 } 1020 1021 /* 1022 * Construct a mirror mapping: 1023 * 1024 * log_type #log_params <log_params> 1025 * #mirrors [mirror_path offset]{2,} 1026 * [#features <features>] 1027 * 1028 * log_type is "core" or "disk" 1029 * #log_params is between 1 and 3 1030 * 1031 * If present, features must be "handle_errors". 1032 */ 1033 static int mirror_ctr(struct dm_target *ti, unsigned int argc, char **argv) 1034 { 1035 int r; 1036 unsigned int nr_mirrors, m, args_used; 1037 struct mirror_set *ms; 1038 struct dm_dirty_log *dl; 1039 1040 dl = create_dirty_log(ti, argc, argv, &args_used); 1041 if (!dl) 1042 return -EINVAL; 1043 1044 argv += args_used; 1045 argc -= args_used; 1046 1047 if (!argc || sscanf(argv[0], "%u", &nr_mirrors) != 1 || 1048 nr_mirrors < 2 || nr_mirrors > DM_KCOPYD_MAX_REGIONS + 1) { 1049 ti->error = "Invalid number of mirrors"; 1050 dm_dirty_log_destroy(dl); 1051 return -EINVAL; 1052 } 1053 1054 argv++, argc--; 1055 1056 if (argc < nr_mirrors * 2) { 1057 ti->error = "Too few mirror arguments"; 1058 dm_dirty_log_destroy(dl); 1059 return -EINVAL; 1060 } 1061 1062 ms = alloc_context(nr_mirrors, dl->type->get_region_size(dl), ti, dl); 1063 if (!ms) { 1064 dm_dirty_log_destroy(dl); 1065 return -ENOMEM; 1066 } 1067 1068 /* Get the mirror parameter sets */ 1069 for (m = 0; m < nr_mirrors; m++) { 1070 r = get_mirror(ms, ti, m, argv); 1071 if (r) { 1072 free_context(ms, ti, m); 1073 return r; 1074 } 1075 argv += 2; 1076 argc -= 2; 1077 } 1078 1079 ti->private = ms; 1080 ti->split_io = dm_rh_get_region_size(ms->rh); 1081 ti->num_flush_requests = 1; 1082 ti->num_discard_requests = 1; 1083 1084 ms->kmirrord_wq = alloc_workqueue("kmirrord", 1085 WQ_NON_REENTRANT | WQ_MEM_RECLAIM, 0); 1086 if (!ms->kmirrord_wq) { 1087 DMERR("couldn't start kmirrord"); 1088 r = -ENOMEM; 1089 goto err_free_context; 1090 } 1091 INIT_WORK(&ms->kmirrord_work, do_mirror); 1092 init_timer(&ms->timer); 1093 ms->timer_pending = 0; 1094 INIT_WORK(&ms->trigger_event, trigger_event); 1095 1096 r = parse_features(ms, argc, argv, &args_used); 1097 if (r) 1098 goto err_destroy_wq; 1099 1100 argv += args_used; 1101 argc -= args_used; 1102 1103 /* 1104 * Any read-balancing addition depends on the 1105 * DM_RAID1_HANDLE_ERRORS flag being present. 1106 * This is because the decision to balance depends 1107 * on the sync state of a region. If the above 1108 * flag is not present, we ignore errors; and 1109 * the sync state may be inaccurate. 1110 */ 1111 1112 if (argc) { 1113 ti->error = "Too many mirror arguments"; 1114 r = -EINVAL; 1115 goto err_destroy_wq; 1116 } 1117 1118 ms->kcopyd_client = dm_kcopyd_client_create(); 1119 if (IS_ERR(ms->kcopyd_client)) { 1120 r = PTR_ERR(ms->kcopyd_client); 1121 goto err_destroy_wq; 1122 } 1123 1124 wakeup_mirrord(ms); 1125 return 0; 1126 1127 err_destroy_wq: 1128 destroy_workqueue(ms->kmirrord_wq); 1129 err_free_context: 1130 free_context(ms, ti, ms->nr_mirrors); 1131 return r; 1132 } 1133 1134 static void mirror_dtr(struct dm_target *ti) 1135 { 1136 struct mirror_set *ms = (struct mirror_set *) ti->private; 1137 1138 del_timer_sync(&ms->timer); 1139 flush_workqueue(ms->kmirrord_wq); 1140 flush_work_sync(&ms->trigger_event); 1141 dm_kcopyd_client_destroy(ms->kcopyd_client); 1142 destroy_workqueue(ms->kmirrord_wq); 1143 free_context(ms, ti, ms->nr_mirrors); 1144 } 1145 1146 /* 1147 * Mirror mapping function 1148 */ 1149 static int mirror_map(struct dm_target *ti, struct bio *bio, 1150 union map_info *map_context) 1151 { 1152 int r, rw = bio_rw(bio); 1153 struct mirror *m; 1154 struct mirror_set *ms = ti->private; 1155 struct dm_raid1_read_record *read_record = NULL; 1156 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 1157 1158 if (rw == WRITE) { 1159 /* Save region for mirror_end_io() handler */ 1160 map_context->ll = dm_rh_bio_to_region(ms->rh, bio); 1161 queue_bio(ms, bio, rw); 1162 return DM_MAPIO_SUBMITTED; 1163 } 1164 1165 r = log->type->in_sync(log, dm_rh_bio_to_region(ms->rh, bio), 0); 1166 if (r < 0 && r != -EWOULDBLOCK) 1167 return r; 1168 1169 /* 1170 * If region is not in-sync queue the bio. 1171 */ 1172 if (!r || (r == -EWOULDBLOCK)) { 1173 if (rw == READA) 1174 return -EWOULDBLOCK; 1175 1176 queue_bio(ms, bio, rw); 1177 return DM_MAPIO_SUBMITTED; 1178 } 1179 1180 /* 1181 * The region is in-sync and we can perform reads directly. 1182 * Store enough information so we can retry if it fails. 1183 */ 1184 m = choose_mirror(ms, bio->bi_sector); 1185 if (unlikely(!m)) 1186 return -EIO; 1187 1188 read_record = mempool_alloc(ms->read_record_pool, GFP_NOIO); 1189 if (likely(read_record)) { 1190 dm_bio_record(&read_record->details, bio); 1191 map_context->ptr = read_record; 1192 read_record->m = m; 1193 } 1194 1195 map_bio(m, bio); 1196 1197 return DM_MAPIO_REMAPPED; 1198 } 1199 1200 static int mirror_end_io(struct dm_target *ti, struct bio *bio, 1201 int error, union map_info *map_context) 1202 { 1203 int rw = bio_rw(bio); 1204 struct mirror_set *ms = (struct mirror_set *) ti->private; 1205 struct mirror *m = NULL; 1206 struct dm_bio_details *bd = NULL; 1207 struct dm_raid1_read_record *read_record = map_context->ptr; 1208 1209 /* 1210 * We need to dec pending if this was a write. 1211 */ 1212 if (rw == WRITE) { 1213 if (!(bio->bi_rw & REQ_FLUSH)) 1214 dm_rh_dec(ms->rh, map_context->ll); 1215 return error; 1216 } 1217 1218 if (error == -EOPNOTSUPP) 1219 goto out; 1220 1221 if ((error == -EWOULDBLOCK) && (bio->bi_rw & REQ_RAHEAD)) 1222 goto out; 1223 1224 if (unlikely(error)) { 1225 if (!read_record) { 1226 /* 1227 * There wasn't enough memory to record necessary 1228 * information for a retry or there was no other 1229 * mirror in-sync. 1230 */ 1231 DMERR_LIMIT("Mirror read failed."); 1232 return -EIO; 1233 } 1234 1235 m = read_record->m; 1236 1237 DMERR("Mirror read failed from %s. Trying alternative device.", 1238 m->dev->name); 1239 1240 fail_mirror(m, DM_RAID1_READ_ERROR); 1241 1242 /* 1243 * A failed read is requeued for another attempt using an intact 1244 * mirror. 1245 */ 1246 if (default_ok(m) || mirror_available(ms, bio)) { 1247 bd = &read_record->details; 1248 1249 dm_bio_restore(bd, bio); 1250 mempool_free(read_record, ms->read_record_pool); 1251 map_context->ptr = NULL; 1252 queue_bio(ms, bio, rw); 1253 return 1; 1254 } 1255 DMERR("All replicated volumes dead, failing I/O"); 1256 } 1257 1258 out: 1259 if (read_record) { 1260 mempool_free(read_record, ms->read_record_pool); 1261 map_context->ptr = NULL; 1262 } 1263 1264 return error; 1265 } 1266 1267 static void mirror_presuspend(struct dm_target *ti) 1268 { 1269 struct mirror_set *ms = (struct mirror_set *) ti->private; 1270 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 1271 1272 struct bio_list holds; 1273 struct bio *bio; 1274 1275 atomic_set(&ms->suspend, 1); 1276 1277 /* 1278 * Process bios in the hold list to start recovery waiting 1279 * for bios in the hold list. After the process, no bio has 1280 * a chance to be added in the hold list because ms->suspend 1281 * is set. 1282 */ 1283 spin_lock_irq(&ms->lock); 1284 holds = ms->holds; 1285 bio_list_init(&ms->holds); 1286 spin_unlock_irq(&ms->lock); 1287 1288 while ((bio = bio_list_pop(&holds))) 1289 hold_bio(ms, bio); 1290 1291 /* 1292 * We must finish up all the work that we've 1293 * generated (i.e. recovery work). 1294 */ 1295 dm_rh_stop_recovery(ms->rh); 1296 1297 wait_event(_kmirrord_recovery_stopped, 1298 !dm_rh_recovery_in_flight(ms->rh)); 1299 1300 if (log->type->presuspend && log->type->presuspend(log)) 1301 /* FIXME: need better error handling */ 1302 DMWARN("log presuspend failed"); 1303 1304 /* 1305 * Now that recovery is complete/stopped and the 1306 * delayed bios are queued, we need to wait for 1307 * the worker thread to complete. This way, 1308 * we know that all of our I/O has been pushed. 1309 */ 1310 flush_workqueue(ms->kmirrord_wq); 1311 } 1312 1313 static void mirror_postsuspend(struct dm_target *ti) 1314 { 1315 struct mirror_set *ms = ti->private; 1316 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 1317 1318 if (log->type->postsuspend && log->type->postsuspend(log)) 1319 /* FIXME: need better error handling */ 1320 DMWARN("log postsuspend failed"); 1321 } 1322 1323 static void mirror_resume(struct dm_target *ti) 1324 { 1325 struct mirror_set *ms = ti->private; 1326 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 1327 1328 atomic_set(&ms->suspend, 0); 1329 if (log->type->resume && log->type->resume(log)) 1330 /* FIXME: need better error handling */ 1331 DMWARN("log resume failed"); 1332 dm_rh_start_recovery(ms->rh); 1333 } 1334 1335 /* 1336 * device_status_char 1337 * @m: mirror device/leg we want the status of 1338 * 1339 * We return one character representing the most severe error 1340 * we have encountered. 1341 * A => Alive - No failures 1342 * D => Dead - A write failure occurred leaving mirror out-of-sync 1343 * S => Sync - A sychronization failure occurred, mirror out-of-sync 1344 * R => Read - A read failure occurred, mirror data unaffected 1345 * 1346 * Returns: <char> 1347 */ 1348 static char device_status_char(struct mirror *m) 1349 { 1350 if (!atomic_read(&(m->error_count))) 1351 return 'A'; 1352 1353 return (test_bit(DM_RAID1_FLUSH_ERROR, &(m->error_type))) ? 'F' : 1354 (test_bit(DM_RAID1_WRITE_ERROR, &(m->error_type))) ? 'D' : 1355 (test_bit(DM_RAID1_SYNC_ERROR, &(m->error_type))) ? 'S' : 1356 (test_bit(DM_RAID1_READ_ERROR, &(m->error_type))) ? 'R' : 'U'; 1357 } 1358 1359 1360 static int mirror_status(struct dm_target *ti, status_type_t type, 1361 char *result, unsigned int maxlen) 1362 { 1363 unsigned int m, sz = 0; 1364 struct mirror_set *ms = (struct mirror_set *) ti->private; 1365 struct dm_dirty_log *log = dm_rh_dirty_log(ms->rh); 1366 char buffer[ms->nr_mirrors + 1]; 1367 1368 switch (type) { 1369 case STATUSTYPE_INFO: 1370 DMEMIT("%d ", ms->nr_mirrors); 1371 for (m = 0; m < ms->nr_mirrors; m++) { 1372 DMEMIT("%s ", ms->mirror[m].dev->name); 1373 buffer[m] = device_status_char(&(ms->mirror[m])); 1374 } 1375 buffer[m] = '\0'; 1376 1377 DMEMIT("%llu/%llu 1 %s ", 1378 (unsigned long long)log->type->get_sync_count(log), 1379 (unsigned long long)ms->nr_regions, buffer); 1380 1381 sz += log->type->status(log, type, result+sz, maxlen-sz); 1382 1383 break; 1384 1385 case STATUSTYPE_TABLE: 1386 sz = log->type->status(log, type, result, maxlen); 1387 1388 DMEMIT("%d", ms->nr_mirrors); 1389 for (m = 0; m < ms->nr_mirrors; m++) 1390 DMEMIT(" %s %llu", ms->mirror[m].dev->name, 1391 (unsigned long long)ms->mirror[m].offset); 1392 1393 if (ms->features & DM_RAID1_HANDLE_ERRORS) 1394 DMEMIT(" 1 handle_errors"); 1395 } 1396 1397 return 0; 1398 } 1399 1400 static int mirror_iterate_devices(struct dm_target *ti, 1401 iterate_devices_callout_fn fn, void *data) 1402 { 1403 struct mirror_set *ms = ti->private; 1404 int ret = 0; 1405 unsigned i; 1406 1407 for (i = 0; !ret && i < ms->nr_mirrors; i++) 1408 ret = fn(ti, ms->mirror[i].dev, 1409 ms->mirror[i].offset, ti->len, data); 1410 1411 return ret; 1412 } 1413 1414 static struct target_type mirror_target = { 1415 .name = "mirror", 1416 .version = {1, 12, 1}, 1417 .module = THIS_MODULE, 1418 .ctr = mirror_ctr, 1419 .dtr = mirror_dtr, 1420 .map = mirror_map, 1421 .end_io = mirror_end_io, 1422 .presuspend = mirror_presuspend, 1423 .postsuspend = mirror_postsuspend, 1424 .resume = mirror_resume, 1425 .status = mirror_status, 1426 .iterate_devices = mirror_iterate_devices, 1427 }; 1428 1429 static int __init dm_mirror_init(void) 1430 { 1431 int r; 1432 1433 _dm_raid1_read_record_cache = KMEM_CACHE(dm_raid1_read_record, 0); 1434 if (!_dm_raid1_read_record_cache) { 1435 DMERR("Can't allocate dm_raid1_read_record cache"); 1436 r = -ENOMEM; 1437 goto bad_cache; 1438 } 1439 1440 r = dm_register_target(&mirror_target); 1441 if (r < 0) { 1442 DMERR("Failed to register mirror target"); 1443 goto bad_target; 1444 } 1445 1446 return 0; 1447 1448 bad_target: 1449 kmem_cache_destroy(_dm_raid1_read_record_cache); 1450 bad_cache: 1451 return r; 1452 } 1453 1454 static void __exit dm_mirror_exit(void) 1455 { 1456 dm_unregister_target(&mirror_target); 1457 kmem_cache_destroy(_dm_raid1_read_record_cache); 1458 } 1459 1460 /* Module hooks */ 1461 module_init(dm_mirror_init); 1462 module_exit(dm_mirror_exit); 1463 1464 MODULE_DESCRIPTION(DM_NAME " mirror target"); 1465 MODULE_AUTHOR("Joe Thornber"); 1466 MODULE_LICENSE("GPL"); 1467