drivers/block/drbd/drbd_worker.c

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