drivers/block/drbd/drbd_req.c

   1 /*
   2    drbd_req.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
  28 #include <linux/slab.h>
  29 #include <linux/drbd.h>
  30 #include "drbd_int.h"
  31 #include "drbd_req.h"
  32
  33
  34 /* Update disk stats at start of I/O request */
  35 static void _drbd_start_io_acct(struct drbd_conf *mdev, struct drbd_request *req, struct bio *bio)
  36 {
  37         const int rw = bio_data_dir(bio);
  38         int cpu;
  39         cpu = part_stat_lock();
  40         part_stat_inc(cpu, &mdev->vdisk->part0, ios[rw]);
  41         part_stat_add(cpu, &mdev->vdisk->part0, sectors[rw], bio_sectors(bio));
  42         part_inc_in_flight(&mdev->vdisk->part0, rw);
  43         part_stat_unlock();
  44 }
  45
  46 /* Update disk stats when completing request upwards */
  47 static void _drbd_end_io_acct(struct drbd_conf *mdev, struct drbd_request *req)
  48 {
  49         int rw = bio_data_dir(req->master_bio);
  50         unsigned long duration = jiffies - req->start_time;
  51         int cpu;
  52         cpu = part_stat_lock();
  53         part_stat_add(cpu, &mdev->vdisk->part0, ticks[rw], duration);
  54         part_round_stats(cpu, &mdev->vdisk->part0);
  55         part_dec_in_flight(&mdev->vdisk->part0, rw);
  56         part_stat_unlock();
  57 }
  58
  59 static struct drbd_request *drbd_req_new(struct drbd_conf *mdev,
  60                                                struct bio *bio_src)
  61 {
  62         struct drbd_request *req;
  63
  64         req = mempool_alloc(drbd_request_mempool, GFP_NOIO);
  65         if (!req)
  66                 return NULL;
  67
  68         drbd_req_make_private_bio(req, bio_src);
  69         req->rq_state    = bio_data_dir(bio_src) == WRITE ? RQ_WRITE : 0;
  70         req->w.mdev      = mdev;
  71         req->master_bio  = bio_src;
  72         req->epoch       = 0;
  73
  74         drbd_clear_interval(&req->i);
  75         req->i.sector     = bio_src->bi_sector;
  76         req->i.size      = bio_src->bi_size;
  77         req->i.local = true;
  78         req->i.waiting = false;
  79
  80         INIT_LIST_HEAD(&req->tl_requests);
  81         INIT_LIST_HEAD(&req->w.list);
  82
  83         return req;
  84 }
  85
  86 static void drbd_req_free(struct drbd_request *req)
  87 {
  88         mempool_free(req, drbd_request_mempool);
  89 }
  90
  91 /* rw is bio_data_dir(), only READ or WRITE */
  92 static void _req_is_done(struct drbd_conf *mdev, struct drbd_request *req, const int rw)
  93 {
  94         const unsigned long s = req->rq_state;
  95
  96         /* remove it from the transfer log.
  97          * well, only if it had been there in the first
  98          * place... if it had not (local only or conflicting
  99          * and never sent), it should still be "empty" as
 100          * initialized in drbd_req_new(), so we can list_del() it
 101          * here unconditionally */
 102         list_del(&req->tl_requests);
 103
 104         /* if it was a write, we may have to set the corresponding
 105          * bit(s) out-of-sync first. If it had a local part, we need to
 106          * release the reference to the activity log. */
 107         if (rw == WRITE) {
 108                 /* Set out-of-sync unless both OK flags are set
 109                  * (local only or remote failed).
 110                  * Other places where we set out-of-sync:
 111                  * READ with local io-error */
 112                 if (!(s & RQ_NET_OK) || !(s & RQ_LOCAL_OK))
 113                         drbd_set_out_of_sync(mdev, req->i.sector, req->i.size);
 114
 115                 if ((s & RQ_NET_OK) && (s & RQ_LOCAL_OK) && (s & RQ_NET_SIS))
 116                         drbd_set_in_sync(mdev, req->i.sector, req->i.size);
 117
 118                 /* one might be tempted to move the drbd_al_complete_io
 119                  * to the local io completion callback drbd_request_endio.
 120                  * but, if this was a mirror write, we may only
 121                  * drbd_al_complete_io after this is RQ_NET_DONE,
 122                  * otherwise the extent could be dropped from the al
 123                  * before it has actually been written on the peer.
 124                  * if we crash before our peer knows about the request,
 125                  * but after the extent has been dropped from the al,
 126                  * we would forget to resync the corresponding extent.
 127                  */
 128                 if (s & RQ_LOCAL_MASK) {
 129                         if (get_ldev_if_state(mdev, D_FAILED)) {
 130                                 if (s & RQ_IN_ACT_LOG)
 131                                         drbd_al_complete_io(mdev, &req->i);
 132                                 put_ldev(mdev);
 133                         } else if (__ratelimit(&drbd_ratelimit_state)) {
 134                                 dev_warn(DEV, "Should have called drbd_al_complete_io(, %llu, %u), "
 135                                          "but my Disk seems to have failed :(\n",
 136                                          (unsigned long long) req->i.sector, req->i.size);
 137                         }
 138                 }
 139         }
 140
 141         drbd_req_free(req);
 142 }
 143
 144 static void queue_barrier(struct drbd_conf *mdev)
 145 {
 146         struct drbd_tl_epoch *b;
 147
 148         /* We are within the req_lock. Once we queued the barrier for sending,
 149          * we set the CREATE_BARRIER bit. It is cleared as soon as a new
 150          * barrier/epoch object is added. This is the only place this bit is
 151          * set. It indicates that the barrier for this epoch is already queued,
 152          * and no new epoch has been created yet. */
 153         if (test_bit(CREATE_BARRIER, &mdev->flags))
 154                 return;
 155
 156         b = mdev->tconn->newest_tle;
 157         b->w.cb = w_send_barrier;
 158         b->w.mdev = mdev;
 159         /* inc_ap_pending done here, so we won't
 160          * get imbalanced on connection loss.
 161          * dec_ap_pending will be done in got_BarrierAck
 162          * or (on connection loss) in tl_clear.  */
 163         inc_ap_pending(mdev);
 164         drbd_queue_work(&mdev->tconn->data.work, &b->w);
 165         set_bit(CREATE_BARRIER, &mdev->flags);
 166 }
 167
 168 static void _about_to_complete_local_write(struct drbd_conf *mdev,
 169         struct drbd_request *req)
 170 {
 171         const unsigned long s = req->rq_state;
 172
 173         /* Before we can signal completion to the upper layers,
 174          * we may need to close the current epoch.
 175          * We can skip this, if this request has not even been sent, because we
 176          * did not have a fully established connection yet/anymore, during
 177          * bitmap exchange, or while we are C_AHEAD due to congestion policy.
 178          */
 179         if (mdev->state.conn >= C_CONNECTED &&
 180             (s & RQ_NET_SENT) != 0 &&
 181             req->epoch == mdev->tconn->newest_tle->br_number)
 182                 queue_barrier(mdev);
 183 }
 184
 185 void complete_master_bio(struct drbd_conf *mdev,
 186                 struct bio_and_error *m)
 187 {
 188         bio_endio(m->bio, m->error);
 189         dec_ap_bio(mdev);
 190 }
 191
 192
 193 static void drbd_remove_request_interval(struct rb_root *root,
 194                                          struct drbd_request *req)
 195 {
 196         struct drbd_conf *mdev = req->w.mdev;
 197         struct drbd_interval *i = &req->i;
 198
 199         drbd_remove_interval(root, i);
 200
 201         /* Wake up any processes waiting for this request to complete.  */
 202         if (i->waiting)
 203                 wake_up(&mdev->misc_wait);
 204 }
 205
 206 /* Helper for __req_mod().
 207  * Set m->bio to the master bio, if it is fit to be completed,
 208  * or leave it alone (it is initialized to NULL in __req_mod),
 209  * if it has already been completed, or cannot be completed yet.
 210  * If m->bio is set, the error status to be returned is placed in m->error.
 211  */
 212 void _req_may_be_done(struct drbd_request *req, struct bio_and_error *m)
 213 {
 214         const unsigned long s = req->rq_state;
 215         struct drbd_conf *mdev = req->w.mdev;
 216         /* only WRITES may end up here without a master bio (on barrier ack) */
 217         int rw = req->master_bio ? bio_data_dir(req->master_bio) : WRITE;
 218
 219         /* we must not complete the master bio, while it is
 220          *      still being processed by _drbd_send_zc_bio (drbd_send_dblock)
 221          *      not yet acknowledged by the peer
 222          *      not yet completed by the local io subsystem
 223          * these flags may get cleared in any order by
 224          *      the worker,
 225          *      the receiver,
 226          *      the bio_endio completion callbacks.
 227          */
 228         if (s & RQ_LOCAL_PENDING)
 229                 return;
 230         if (req->i.waiting) {
 231                 /* Retry all conflicting peer requests.  */
 232                 wake_up(&mdev->misc_wait);
 233         }
 234         if (s & RQ_NET_QUEUED)
 235                 return;
 236         if (s & RQ_NET_PENDING)
 237                 return;
 238
 239         if (req->master_bio) {
 240                 /* this is DATA_RECEIVED (remote read)
 241                  * or protocol C P_WRITE_ACK
 242                  * or protocol B P_RECV_ACK
 243                  * or protocol A "HANDED_OVER_TO_NETWORK" (SendAck)
 244                  * or canceled or failed,
 245                  * or killed from the transfer log due to connection loss.
 246                  */
 247
 248                 /*
 249                  * figure out whether to report success or failure.
 250                  *
 251                  * report success when at least one of the operations succeeded.
 252                  * or, to put the other way,
 253                  * only report failure, when both operations failed.
 254                  *
 255                  * what to do about the failures is handled elsewhere.
 256                  * what we need to do here is just: complete the master_bio.
 257                  *
 258                  * local completion error, if any, has been stored as ERR_PTR
 259                  * in private_bio within drbd_request_endio.
 260                  */
 261                 int ok = (s & RQ_LOCAL_OK) || (s & RQ_NET_OK);
 262                 int error = PTR_ERR(req->private_bio);
 263
 264                 /* remove the request from the conflict detection
 265                  * respective block_id verification hash */
 266                 if (!drbd_interval_empty(&req->i)) {
 267                         struct rb_root *root;
 268
 269                         if (rw == WRITE)
 270                                 root = &mdev->write_requests;
 271                         else
 272                                 root = &mdev->read_requests;
 273                         drbd_remove_request_interval(root, req);
 274                 } else if (!(s & RQ_POSTPONED))
 275                         D_ASSERT((s & (RQ_NET_MASK & ~RQ_NET_DONE)) == 0);
 276
 277                 /* for writes we need to do some extra housekeeping */
 278                 if (rw == WRITE)
 279                         _about_to_complete_local_write(mdev, req);
 280
 281                 /* Update disk stats */
 282                 _drbd_end_io_acct(mdev, req);
 283
 284                 if (!(s & RQ_POSTPONED)) {
 285                         m->error = ok ? 0 : (error ?: -EIO);
 286                         m->bio = req->master_bio;
 287                 }
 288                 req->master_bio = NULL;
 289         }
 290
 291         if ((s & RQ_NET_MASK) == 0 || (s & RQ_NET_DONE)) {
 292                 /* this is disconnected (local only) operation,
 293                  * or protocol C P_WRITE_ACK,
 294                  * or protocol A or B P_BARRIER_ACK,
 295                  * or killed from the transfer log due to connection loss. */
 296                 _req_is_done(mdev, req, rw);
 297         }
 298         /* else: network part and not DONE yet. that is
 299          * protocol A or B, barrier ack still pending... */
 300 }
 301
 302 static void _req_may_be_done_not_susp(struct drbd_request *req, struct bio_and_error *m)
 303 {
 304         struct drbd_conf *mdev = req->w.mdev;
 305
 306         if (!is_susp(mdev->state))
 307                 _req_may_be_done(req, m);
 308 }
 309
 310 /* obviously this could be coded as many single functions
 311  * instead of one huge switch,
 312  * or by putting the code directly in the respective locations
 313  * (as it has been before).
 314  *
 315  * but having it this way
 316  *  enforces that it is all in this one place, where it is easier to audit,
 317  *  it makes it obvious that whatever "event" "happens" to a request should
 318  *  happen "atomically" within the req_lock,
 319  *  and it enforces that we have to think in a very structured manner
 320  *  about the "events" that may happen to a request during its life time ...
 321  */
 322 int __req_mod(struct drbd_request *req, enum drbd_req_event what,
 323                 struct bio_and_error *m)
 324 {
 325         struct drbd_conf *mdev = req->w.mdev;
 326         int rv = 0;
 327
 328         if (m)
 329                 m->bio = NULL;
 330
 331         switch (what) {
 332         default:
 333                 dev_err(DEV, "LOGIC BUG in %s:%u\n", __FILE__ , __LINE__);
 334                 break;
 335
 336         /* does not happen...
 337          * initialization done in drbd_req_new
 338         case CREATED:
 339                 break;
 340                 */
 341
 342         case TO_BE_SENT: /* via network */
 343                 /* reached via __drbd_make_request
 344                  * and from w_read_retry_remote */
 345                 D_ASSERT(!(req->rq_state & RQ_NET_MASK));
 346                 req->rq_state |= RQ_NET_PENDING;
 347                 inc_ap_pending(mdev);
 348                 break;
 349
 350         case TO_BE_SUBMITTED: /* locally */
 351                 /* reached via __drbd_make_request */
 352                 D_ASSERT(!(req->rq_state & RQ_LOCAL_MASK));
 353                 req->rq_state |= RQ_LOCAL_PENDING;
 354                 break;
 355
 356         case COMPLETED_OK:
 357                 if (bio_data_dir(req->master_bio) == WRITE)
 358                         mdev->writ_cnt += req->i.size >> 9;
 359                 else
 360                         mdev->read_cnt += req->i.size >> 9;
 361
 362                 req->rq_state |= (RQ_LOCAL_COMPLETED|RQ_LOCAL_OK);
 363                 req->rq_state &= ~RQ_LOCAL_PENDING;
 364
 365                 _req_may_be_done_not_susp(req, m);
 366                 put_ldev(mdev);
 367                 break;
 368
 369         case WRITE_COMPLETED_WITH_ERROR:
 370                 req->rq_state |= RQ_LOCAL_COMPLETED;
 371                 req->rq_state &= ~RQ_LOCAL_PENDING;
 372
 373                 __drbd_chk_io_error(mdev, false);
 374                 _req_may_be_done_not_susp(req, m);
 375                 put_ldev(mdev);
 376                 break;
 377
 378         case READ_AHEAD_COMPLETED_WITH_ERROR:
 379                 /* it is legal to fail READA */
 380                 req->rq_state |= RQ_LOCAL_COMPLETED;
 381                 req->rq_state &= ~RQ_LOCAL_PENDING;
 382                 _req_may_be_done_not_susp(req, m);
 383                 put_ldev(mdev);
 384                 break;
 385
 386         case READ_COMPLETED_WITH_ERROR:
 387                 drbd_set_out_of_sync(mdev, req->i.sector, req->i.size);
 388
 389                 req->rq_state |= RQ_LOCAL_COMPLETED;
 390                 req->rq_state &= ~RQ_LOCAL_PENDING;
 391
 392                 D_ASSERT(!(req->rq_state & RQ_NET_MASK));
 393
 394                 __drbd_chk_io_error(mdev, false);
 395                 put_ldev(mdev);
 396
 397                 /* no point in retrying if there is no good remote data,
 398                  * or we have no connection. */
 399                 if (mdev->state.pdsk != D_UP_TO_DATE) {
 400                         _req_may_be_done_not_susp(req, m);
 401                         break;
 402                 }
 403
 404                 /* _req_mod(req,TO_BE_SENT); oops, recursion... */
 405                 req->rq_state |= RQ_NET_PENDING;
 406                 inc_ap_pending(mdev);
 407                 /* fall through: _req_mod(req,QUEUE_FOR_NET_READ); */
 408
 409         case QUEUE_FOR_NET_READ:
 410                 /* READ or READA, and
 411                  * no local disk,
 412                  * or target area marked as invalid,
 413                  * or just got an io-error. */
 414                 /* from __drbd_make_request
 415                  * or from bio_endio during read io-error recovery */
 416
 417                 /* so we can verify the handle in the answer packet
 418                  * corresponding hlist_del is in _req_may_be_done() */
 419                 drbd_insert_interval(&mdev->read_requests, &req->i);
 420
 421                 set_bit(UNPLUG_REMOTE, &mdev->flags);
 422
 423                 D_ASSERT(req->rq_state & RQ_NET_PENDING);
 424                 req->rq_state |= RQ_NET_QUEUED;
 425                 req->w.cb = (req->rq_state & RQ_LOCAL_MASK)
 426                         ? w_read_retry_remote
 427                         : w_send_read_req;
 428                 drbd_queue_work(&mdev->tconn->data.work, &req->w);
 429                 break;
 430
 431         case QUEUE_FOR_NET_WRITE:
 432                 /* assert something? */
 433                 /* from __drbd_make_request only */
 434
 435                 /* corresponding hlist_del is in _req_may_be_done() */
 436                 drbd_insert_interval(&mdev->write_requests, &req->i);
 437
 438                 /* NOTE
 439                  * In case the req ended up on the transfer log before being
 440                  * queued on the worker, it could lead to this request being
 441                  * missed during cleanup after connection loss.
 442                  * So we have to do both operations here,
 443                  * within the same lock that protects the transfer log.
 444                  *
 445                  * _req_add_to_epoch(req); this has to be after the
 446                  * _maybe_start_new_epoch(req); which happened in
 447                  * __drbd_make_request, because we now may set the bit
 448                  * again ourselves to close the current epoch.
 449                  *
 450                  * Add req to the (now) current epoch (barrier). */
 451
 452                 /* otherwise we may lose an unplug, which may cause some remote
 453                  * io-scheduler timeout to expire, increasing maximum latency,
 454                  * hurting performance. */
 455                 set_bit(UNPLUG_REMOTE, &mdev->flags);
 456
 457                 /* see __drbd_make_request,
 458                  * just after it grabs the req_lock */
 459                 D_ASSERT(test_bit(CREATE_BARRIER, &mdev->flags) == 0);
 460
 461                 req->epoch = mdev->tconn->newest_tle->br_number;
 462
 463                 /* increment size of current epoch */
 464                 mdev->tconn->newest_tle->n_writes++;
 465
 466                 /* queue work item to send data */
 467                 D_ASSERT(req->rq_state & RQ_NET_PENDING);
 468                 req->rq_state |= RQ_NET_QUEUED;
 469                 req->w.cb =  w_send_dblock;
 470                 drbd_queue_work(&mdev->tconn->data.work, &req->w);
 471
 472                 /* close the epoch, in case it outgrew the limit */
 473                 if (mdev->tconn->newest_tle->n_writes >= mdev->tconn->net_conf->max_epoch_size)
 474                         queue_barrier(mdev);
 475
 476                 break;
 477
 478         case QUEUE_FOR_SEND_OOS:
 479                 req->rq_state |= RQ_NET_QUEUED;
 480                 req->w.cb =  w_send_out_of_sync;
 481                 drbd_queue_work(&mdev->tconn->data.work, &req->w);
 482                 break;
 483
 484         case OOS_HANDED_TO_NETWORK:
 485                 /* actually the same */
 486         case SEND_CANCELED:
 487                 /* treat it the same */
 488         case SEND_FAILED:
 489                 /* real cleanup will be done from tl_clear.  just update flags
 490                  * so it is no longer marked as on the worker queue */
 491                 req->rq_state &= ~RQ_NET_QUEUED;
 492                 /* if we did it right, tl_clear should be scheduled only after
 493                  * this, so this should not be necessary! */
 494                 _req_may_be_done_not_susp(req, m);
 495                 break;
 496
 497         case HANDED_OVER_TO_NETWORK:
 498                 /* assert something? */
 499                 if (bio_data_dir(req->master_bio) == WRITE)
 500                         atomic_add(req->i.size >> 9, &mdev->ap_in_flight);
 501
 502                 if (bio_data_dir(req->master_bio) == WRITE &&
 503                     mdev->tconn->net_conf->wire_protocol == DRBD_PROT_A) {
 504                         /* this is what is dangerous about protocol A:
 505                          * pretend it was successfully written on the peer. */
 506                         if (req->rq_state & RQ_NET_PENDING) {
 507                                 dec_ap_pending(mdev);
 508                                 req->rq_state &= ~RQ_NET_PENDING;
 509                                 req->rq_state |= RQ_NET_OK;
 510                         } /* else: neg-ack was faster... */
 511                         /* it is still not yet RQ_NET_DONE until the
 512                          * corresponding epoch barrier got acked as well,
 513                          * so we know what to dirty on connection loss */
 514                 }
 515                 req->rq_state &= ~RQ_NET_QUEUED;
 516                 req->rq_state |= RQ_NET_SENT;
 517                 /* because _drbd_send_zc_bio could sleep, and may want to
 518                  * dereference the bio even after the "WRITE_ACKED_BY_PEER" and
 519                  * "COMPLETED_OK" events came in, once we return from
 520                  * _drbd_send_zc_bio (drbd_send_dblock), we have to check
 521                  * whether it is done already, and end it.  */
 522                 _req_may_be_done_not_susp(req, m);
 523                 break;
 524
 525         case READ_RETRY_REMOTE_CANCELED:
 526                 req->rq_state &= ~RQ_NET_QUEUED;
 527                 /* fall through, in case we raced with drbd_disconnect */
 528         case CONNECTION_LOST_WHILE_PENDING:
 529                 /* transfer log cleanup after connection loss */
 530                 /* assert something? */
 531                 if (req->rq_state & RQ_NET_PENDING)
 532                         dec_ap_pending(mdev);
 533                 req->rq_state &= ~(RQ_NET_OK|RQ_NET_PENDING);
 534                 req->rq_state |= RQ_NET_DONE;
 535                 if (req->rq_state & RQ_NET_SENT && req->rq_state & RQ_WRITE)
 536                         atomic_sub(req->i.size >> 9, &mdev->ap_in_flight);
 537
 538                 /* if it is still queued, we may not complete it here.
 539                  * it will be canceled soon. */
 540                 if (!(req->rq_state & RQ_NET_QUEUED))
 541                         _req_may_be_done(req, m); /* Allowed while state.susp */
 542                 break;
 543
 544         case WRITE_ACKED_BY_PEER_AND_SIS:
 545                 req->rq_state |= RQ_NET_SIS;
 546         case DISCARD_WRITE:
 547                 /* for discarded conflicting writes of multiple primaries,
 548                  * there is no need to keep anything in the tl, potential
 549                  * node crashes are covered by the activity log. */
 550                 req->rq_state |= RQ_NET_DONE;
 551                 /* fall through */
 552         case WRITE_ACKED_BY_PEER:
 553                 /* protocol C; successfully written on peer.
 554                  * Nothing to do here.
 555                  * We want to keep the tl in place for all protocols, to cater
 556                  * for volatile write-back caches on lower level devices.
 557                  *
 558                  * A barrier request is expected to have forced all prior
 559                  * requests onto stable storage, so completion of a barrier
 560                  * request could set NET_DONE right here, and not wait for the
 561                  * P_BARRIER_ACK, but that is an unnecessary optimization. */
 562
 563                 /* this makes it effectively the same as for: */
 564         case RECV_ACKED_BY_PEER:
 565                 /* protocol B; pretends to be successfully written on peer.
 566                  * see also notes above in HANDED_OVER_TO_NETWORK about
 567                  * protocol != C */
 568                 req->rq_state |= RQ_NET_OK;
 569                 D_ASSERT(req->rq_state & RQ_NET_PENDING);
 570                 dec_ap_pending(mdev);
 571                 atomic_sub(req->i.size >> 9, &mdev->ap_in_flight);
 572                 req->rq_state &= ~RQ_NET_PENDING;
 573                 _req_may_be_done_not_susp(req, m);
 574                 break;
 575
 576         case POSTPONE_WRITE:
 577                 /*
 578                  * If this node has already detected the write conflict, the
 579                  * worker will be waiting on misc_wait.  Wake it up once this
 580                  * request has completed locally.
 581                  */
 582                 D_ASSERT(req->rq_state & RQ_NET_PENDING);
 583                 req->rq_state |= RQ_POSTPONED;
 584                 _req_may_be_done_not_susp(req, m);
 585                 break;
 586
 587         case NEG_ACKED:
 588                 /* assert something? */
 589                 if (req->rq_state & RQ_NET_PENDING) {
 590                         dec_ap_pending(mdev);
 591                         atomic_sub(req->i.size >> 9, &mdev->ap_in_flight);
 592                 }
 593                 req->rq_state &= ~(RQ_NET_OK|RQ_NET_PENDING);
 594
 595                 req->rq_state |= RQ_NET_DONE;
 596                 _req_may_be_done_not_susp(req, m);
 597                 /* else: done by HANDED_OVER_TO_NETWORK */
 598                 break;
 599
 600         case FAIL_FROZEN_DISK_IO:
 601                 if (!(req->rq_state & RQ_LOCAL_COMPLETED))
 602                         break;
 603
 604                 _req_may_be_done(req, m); /* Allowed while state.susp */
 605                 break;
 606
 607         case RESTART_FROZEN_DISK_IO:
 608                 if (!(req->rq_state & RQ_LOCAL_COMPLETED))
 609                         break;
 610
 611                 req->rq_state &= ~RQ_LOCAL_COMPLETED;
 612
 613                 rv = MR_READ;
 614                 if (bio_data_dir(req->master_bio) == WRITE)
 615                         rv = MR_WRITE;
 616
 617                 get_ldev(mdev);
 618                 req->w.cb = w_restart_disk_io;
 619                 drbd_queue_work(&mdev->tconn->data.work, &req->w);
 620                 break;
 621
 622         case RESEND:
 623                 /* If RQ_NET_OK is already set, we got a P_WRITE_ACK or P_RECV_ACK
 624                    before the connection loss (B&C only); only P_BARRIER_ACK was missing.
 625                    Trowing them out of the TL here by pretending we got a BARRIER_ACK
 626                    We ensure that the peer was not rebooted */
 627                 if (!(req->rq_state & RQ_NET_OK)) {
 628                         if (req->w.cb) {
 629                                 drbd_queue_work(&mdev->tconn->data.work, &req->w);
 630                                 rv = req->rq_state & RQ_WRITE ? MR_WRITE : MR_READ;
 631                         }
 632                         break;
 633                 }
 634                 /* else, fall through to BARRIER_ACKED */
 635
 636         case BARRIER_ACKED:
 637                 if (!(req->rq_state & RQ_WRITE))
 638                         break;
 639
 640                 if (req->rq_state & RQ_NET_PENDING) {
 641                         /* barrier came in before all requests have been acked.
 642                          * this is bad, because if the connection is lost now,
 643                          * we won't be able to clean them up... */
 644                         dev_err(DEV, "FIXME (BARRIER_ACKED but pending)\n");
 645                         list_move(&req->tl_requests, &mdev->tconn->out_of_sequence_requests);
 646                 }
 647                 if ((req->rq_state & RQ_NET_MASK) != 0) {
 648                         req->rq_state |= RQ_NET_DONE;
 649                         if (mdev->tconn->net_conf->wire_protocol == DRBD_PROT_A)
 650                                 atomic_sub(req->i.size>>9, &mdev->ap_in_flight);
 651                 }
 652                 _req_may_be_done(req, m); /* Allowed while state.susp */
 653                 break;
 654
 655         case DATA_RECEIVED:
 656                 D_ASSERT(req->rq_state & RQ_NET_PENDING);
 657                 dec_ap_pending(mdev);
 658                 req->rq_state &= ~RQ_NET_PENDING;
 659                 req->rq_state |= (RQ_NET_OK|RQ_NET_DONE);
 660                 _req_may_be_done_not_susp(req, m);
 661                 break;
 662         };
 663
 664         return rv;
 665 }
 666
 667 /* we may do a local read if:
 668  * - we are consistent (of course),
 669  * - or we are generally inconsistent,
 670  *   BUT we are still/already IN SYNC for this area.
 671  *   since size may be bigger than BM_BLOCK_SIZE,
 672  *   we may need to check several bits.
 673  */
 674 static bool drbd_may_do_local_read(struct drbd_conf *mdev, sector_t sector, int size)
 675 {
 676         unsigned long sbnr, ebnr;
 677         sector_t esector, nr_sectors;
 678
 679         if (mdev->state.disk == D_UP_TO_DATE)
 680                 return true;
 681         if (mdev->state.disk != D_INCONSISTENT)
 682                 return false;
 683         esector = sector + (size >> 9) - 1;
 684         nr_sectors = drbd_get_capacity(mdev->this_bdev);
 685         D_ASSERT(sector  < nr_sectors);
 686         D_ASSERT(esector < nr_sectors);
 687
 688         sbnr = BM_SECT_TO_BIT(sector);
 689         ebnr = BM_SECT_TO_BIT(esector);
 690
 691         return drbd_bm_count_bits(mdev, sbnr, ebnr) == 0;
 692 }
 693
 694 /*
 695  * complete_conflicting_writes  -  wait for any conflicting write requests
 696  *
 697  * The write_requests tree contains all active write requests which we
 698  * currently know about.  Wait for any requests to complete which conflict with
 699  * the new one.
 700  */
 701 static int complete_conflicting_writes(struct drbd_conf *mdev,
 702                                        sector_t sector, int size)
 703 {
 704         for(;;) {
 705                 struct drbd_interval *i;
 706                 int err;
 707
 708                 i = drbd_find_overlap(&mdev->write_requests, sector, size);
 709                 if (!i)
 710                         return 0;
 711                 err = drbd_wait_misc(mdev, i);
 712                 if (err)
 713                         return err;
 714         }
 715 }
 716
 717 int __drbd_make_request(struct drbd_conf *mdev, struct bio *bio, unsigned long start_time)
 718 {
 719         const int rw = bio_rw(bio);
 720         const int size = bio->bi_size;
 721         const sector_t sector = bio->bi_sector;
 722         struct drbd_tl_epoch *b = NULL;
 723         struct drbd_request *req;
 724         int local, remote, send_oos = 0;
 725         int err;
 726         int ret = 0;
 727
 728         /* allocate outside of all locks; */
 729         req = drbd_req_new(mdev, bio);
 730         if (!req) {
 731                 dec_ap_bio(mdev);
 732                 /* only pass the error to the upper layers.
 733                  * if user cannot handle io errors, that's not our business. */
 734                 dev_err(DEV, "could not kmalloc() req\n");
 735                 bio_endio(bio, -ENOMEM);
 736                 return 0;
 737         }
 738         req->start_time = start_time;
 739
 740         local = get_ldev(mdev);
 741         if (!local) {
 742                 bio_put(req->private_bio); /* or we get a bio leak */
 743                 req->private_bio = NULL;
 744         }
 745         if (rw == WRITE) {
 746                 remote = 1;
 747         } else {
 748                 /* READ || READA */
 749                 if (local) {
 750                         if (!drbd_may_do_local_read(mdev, sector, size)) {
 751                                 /* we could kick the syncer to
 752                                  * sync this extent asap, wait for
 753                                  * it, then continue locally.
 754                                  * Or just issue the request remotely.
 755                                  */
 756                                 local = 0;
 757                                 bio_put(req->private_bio);
 758                                 req->private_bio = NULL;
 759                                 put_ldev(mdev);
 760                         }
 761                 }
 762                 remote = !local && mdev->state.pdsk >= D_UP_TO_DATE;
 763         }
 764
 765         /* If we have a disk, but a READA request is mapped to remote,
 766          * we are R_PRIMARY, D_INCONSISTENT, SyncTarget.
 767          * Just fail that READA request right here.
 768          *
 769          * THINK: maybe fail all READA when not local?
 770          *        or make this configurable...
 771          *        if network is slow, READA won't do any good.
 772          */
 773         if (rw == READA && mdev->state.disk >= D_INCONSISTENT && !local) {
 774                 err = -EWOULDBLOCK;
 775                 goto fail_and_free_req;
 776         }
 777
 778         /* For WRITES going to the local disk, grab a reference on the target
 779          * extent.  This waits for any resync activity in the corresponding
 780          * resync extent to finish, and, if necessary, pulls in the target
 781          * extent into the activity log, which involves further disk io because
 782          * of transactional on-disk meta data updates. */
 783         if (rw == WRITE && local && !test_bit(AL_SUSPENDED, &mdev->flags)) {
 784                 req->rq_state |= RQ_IN_ACT_LOG;
 785                 drbd_al_begin_io(mdev, &req->i);
 786         }
 787
 788         remote = remote && drbd_should_do_remote(mdev->state);
 789         send_oos = rw == WRITE && drbd_should_send_out_of_sync(mdev->state);
 790         D_ASSERT(!(remote && send_oos));
 791
 792         if (!(local || remote) && !is_susp(mdev->state)) {
 793                 if (__ratelimit(&drbd_ratelimit_state))
 794                         dev_err(DEV, "IO ERROR: neither local nor remote disk\n");
 795                 err = -EIO;
 796                 goto fail_free_complete;
 797         }
 798
 799         /* For WRITE request, we have to make sure that we have an
 800          * unused_spare_tle, in case we need to start a new epoch.
 801          * I try to be smart and avoid to pre-allocate always "just in case",
 802          * but there is a race between testing the bit and pointer outside the
 803          * spinlock, and grabbing the spinlock.
 804          * if we lost that race, we retry.  */
 805         if (rw == WRITE && (remote || send_oos) &&
 806             mdev->tconn->unused_spare_tle == NULL &&
 807             test_bit(CREATE_BARRIER, &mdev->flags)) {
 808 allocate_barrier:
 809                 b = kmalloc(sizeof(struct drbd_tl_epoch), GFP_NOIO);
 810                 if (!b) {
 811                         dev_err(DEV, "Failed to alloc barrier.\n");
 812                         err = -ENOMEM;
 813                         goto fail_free_complete;
 814                 }
 815         }
 816
 817         /* GOOD, everything prepared, grab the spin_lock */
 818         spin_lock_irq(&mdev->tconn->req_lock);
 819
 820         if (rw == WRITE) {
 821                 err = complete_conflicting_writes(mdev, sector, size);
 822                 if (err) {
 823                         if (err != -ERESTARTSYS)
 824                                 _conn_request_state(mdev->tconn,
 825                                                     NS(conn, C_TIMEOUT),
 826                                                     CS_HARD);
 827                         spin_unlock_irq(&mdev->tconn->req_lock);
 828                         err = -EIO;
 829                         goto fail_free_complete;
 830                 }
 831         }
 832
 833         if (is_susp(mdev->state)) {
 834                 /* If we got suspended, use the retry mechanism of
 835                    generic_make_request() to restart processing of this
 836                    bio. In the next call to drbd_make_request
 837                    we sleep in inc_ap_bio() */
 838                 ret = 1;
 839                 spin_unlock_irq(&mdev->tconn->req_lock);
 840                 goto fail_free_complete;
 841         }
 842
 843         if (remote || send_oos) {
 844                 remote = drbd_should_do_remote(mdev->state);
 845                 send_oos = rw == WRITE && drbd_should_send_out_of_sync(mdev->state);
 846                 D_ASSERT(!(remote && send_oos));
 847
 848                 if (!(remote || send_oos))
 849                         dev_warn(DEV, "lost connection while grabbing the req_lock!\n");
 850                 if (!(local || remote)) {
 851                         dev_err(DEV, "IO ERROR: neither local nor remote disk\n");
 852                         spin_unlock_irq(&mdev->tconn->req_lock);
 853                         err = -EIO;
 854                         goto fail_free_complete;
 855                 }
 856         }
 857
 858         if (b && mdev->tconn->unused_spare_tle == NULL) {
 859                 mdev->tconn->unused_spare_tle = b;
 860                 b = NULL;
 861         }
 862         if (rw == WRITE && (remote || send_oos) &&
 863             mdev->tconn->unused_spare_tle == NULL &&
 864             test_bit(CREATE_BARRIER, &mdev->flags)) {
 865                 /* someone closed the current epoch
 866                  * while we were grabbing the spinlock */
 867                 spin_unlock_irq(&mdev->tconn->req_lock);
 868                 goto allocate_barrier;
 869         }
 870
 871
 872         /* Update disk stats */
 873         _drbd_start_io_acct(mdev, req, bio);
 874
 875         /* _maybe_start_new_epoch(mdev);
 876          * If we need to generate a write barrier packet, we have to add the
 877          * new epoch (barrier) object, and queue the barrier packet for sending,
 878          * and queue the req's data after it _within the same lock_, otherwise
 879          * we have race conditions were the reorder domains could be mixed up.
 880          *
 881          * Even read requests may start a new epoch and queue the corresponding
 882          * barrier packet.  To get the write ordering right, we only have to
 883          * make sure that, if this is a write request and it triggered a
 884          * barrier packet, this request is queued within the same spinlock. */
 885         if ((remote || send_oos) && mdev->tconn->unused_spare_tle &&
 886             test_and_clear_bit(CREATE_BARRIER, &mdev->flags)) {
 887                 _tl_add_barrier(mdev->tconn, mdev->tconn->unused_spare_tle);
 888                 mdev->tconn->unused_spare_tle = NULL;
 889         } else {
 890                 D_ASSERT(!(remote && rw == WRITE &&
 891                            test_bit(CREATE_BARRIER, &mdev->flags)));
 892         }
 893
 894         /* NOTE
 895          * Actually, 'local' may be wrong here already, since we may have failed
 896          * to write to the meta data, and may become wrong anytime because of
 897          * local io-error for some other request, which would lead to us
 898          * "detaching" the local disk.
 899          *
 900          * 'remote' may become wrong any time because the network could fail.
 901          *
 902          * This is a harmless race condition, though, since it is handled
 903          * correctly at the appropriate places; so it just defers the failure
 904          * of the respective operation.
 905          */
 906
 907         /* mark them early for readability.
 908          * this just sets some state flags. */
 909         if (remote)
 910                 _req_mod(req, TO_BE_SENT);
 911         if (local)
 912                 _req_mod(req, TO_BE_SUBMITTED);
 913
 914         list_add_tail(&req->tl_requests, &mdev->tconn->newest_tle->requests);
 915
 916         /* NOTE remote first: to get the concurrent write detection right,
 917          * we must register the request before start of local IO.  */
 918         if (remote) {
 919                 /* either WRITE and C_CONNECTED,
 920                  * or READ, and no local disk,
 921                  * or READ, but not in sync.
 922                  */
 923                 _req_mod(req, (rw == WRITE)
 924                                 ? QUEUE_FOR_NET_WRITE
 925                                 : QUEUE_FOR_NET_READ);
 926         }
 927         if (send_oos && drbd_set_out_of_sync(mdev, sector, size))
 928                 _req_mod(req, QUEUE_FOR_SEND_OOS);
 929
 930         if (remote &&
 931             mdev->tconn->net_conf->on_congestion != OC_BLOCK && mdev->tconn->agreed_pro_version >= 96) {
 932                 int congested = 0;
 933
 934                 if (mdev->tconn->net_conf->cong_fill &&
 935                     atomic_read(&mdev->ap_in_flight) >= mdev->tconn->net_conf->cong_fill) {
 936                         dev_info(DEV, "Congestion-fill threshold reached\n");
 937                         congested = 1;
 938                 }
 939
 940                 if (mdev->act_log->used >= mdev->tconn->net_conf->cong_extents) {
 941                         dev_info(DEV, "Congestion-extents threshold reached\n");
 942                         congested = 1;
 943                 }
 944
 945                 if (congested) {
 946                         queue_barrier(mdev); /* last barrier, after mirrored writes */
 947
 948                         if (mdev->tconn->net_conf->on_congestion == OC_PULL_AHEAD)
 949                                 _drbd_set_state(_NS(mdev, conn, C_AHEAD), 0, NULL);
 950                         else  /*mdev->tconn->net_conf->on_congestion == OC_DISCONNECT */
 951                                 _drbd_set_state(_NS(mdev, conn, C_DISCONNECTING), 0, NULL);
 952                 }
 953         }
 954
 955         spin_unlock_irq(&mdev->tconn->req_lock);
 956         kfree(b); /* if someone else has beaten us to it... */
 957
 958         if (local) {
 959                 req->private_bio->bi_bdev = mdev->ldev->backing_bdev;
 960
 961                 /* State may have changed since we grabbed our reference on the
 962                  * mdev->ldev member. Double check, and short-circuit to endio.
 963                  * In case the last activity log transaction failed to get on
 964                  * stable storage, and this is a WRITE, we may not even submit
 965                  * this bio. */
 966                 if (get_ldev(mdev)) {
 967                         if (drbd_insert_fault(mdev,   rw == WRITE ? DRBD_FAULT_DT_WR
 968                                                     : rw == READ  ? DRBD_FAULT_DT_RD
 969                                                     :               DRBD_FAULT_DT_RA))
 970                                 bio_endio(req->private_bio, -EIO);
 971                         else
 972                                 generic_make_request(req->private_bio);
 973                         put_ldev(mdev);
 974                 } else
 975                         bio_endio(req->private_bio, -EIO);
 976         }
 977
 978         return 0;
 979
 980 fail_free_complete:
 981         if (req->rq_state & RQ_IN_ACT_LOG)
 982                 drbd_al_complete_io(mdev, &req->i);
 983 fail_and_free_req:
 984         if (local) {
 985                 bio_put(req->private_bio);
 986                 req->private_bio = NULL;
 987                 put_ldev(mdev);
 988         }
 989         if (!ret)
 990                 bio_endio(bio, err);
 991
 992         drbd_req_free(req);
 993         dec_ap_bio(mdev);
 994         kfree(b);
 995
 996         return ret;
 997 }
 998
 999 int drbd_make_request(struct request_queue *q, struct bio *bio)
1000 {
1001         unsigned int s_enr, e_enr;
1002         struct drbd_conf *mdev = (struct drbd_conf *) q->queuedata;
1003         unsigned long start_time;
1004
1005         start_time = jiffies;
1006
1007         /*
1008          * what we "blindly" assume:
1009          */
1010         D_ASSERT(bio->bi_size > 0);
1011         D_ASSERT(IS_ALIGNED(bio->bi_size, 512));
1012
1013         /* to make some things easier, force alignment of requests within the
1014          * granularity of our hash tables */
1015         s_enr = bio->bi_sector >> HT_SHIFT;
1016         e_enr = (bio->bi_sector+(bio->bi_size>>9)-1) >> HT_SHIFT;
1017
1018         if (likely(s_enr == e_enr)) {
1019                 inc_ap_bio(mdev, 1);
1020                 return __drbd_make_request(mdev, bio, start_time);
1021         }
1022
1023         /* can this bio be split generically?
1024          * Maybe add our own split-arbitrary-bios function. */
1025         if (bio->bi_vcnt != 1 || bio->bi_idx != 0 || bio->bi_size > DRBD_MAX_BIO_SIZE) {
1026                 /* rather error out here than BUG in bio_split */
1027                 dev_err(DEV, "bio would need to, but cannot, be split: "
1028                     "(vcnt=%u,idx=%u,size=%u,sector=%llu)\n",
1029                     bio->bi_vcnt, bio->bi_idx, bio->bi_size,
1030                     (unsigned long long)bio->bi_sector);
1031                 bio_endio(bio, -EINVAL);
1032         } else {
1033                 /* This bio crosses some boundary, so we have to split it. */
1034                 struct bio_pair *bp;
1035                 /* works for the "do not cross hash slot boundaries" case
1036                  * e.g. sector 262269, size 4096
1037                  * s_enr = 262269 >> 6 = 4097
1038                  * e_enr = (262269+8-1) >> 6 = 4098
1039                  * HT_SHIFT = 6
1040                  * sps = 64, mask = 63
1041                  * first_sectors = 64 - (262269 & 63) = 3
1042                  */
1043                 const sector_t sect = bio->bi_sector;
1044                 const int sps = 1 << HT_SHIFT; /* sectors per slot */
1045                 const int mask = sps - 1;
1046                 const sector_t first_sectors = sps - (sect & mask);
1047                 bp = bio_split(bio, first_sectors);
1048
1049                 /* we need to get a "reference count" (ap_bio_cnt)
1050                  * to avoid races with the disconnect/reconnect/suspend code.
1051                  * In case we need to split the bio here, we need to get three references
1052                  * atomically, otherwise we might deadlock when trying to submit the
1053                  * second one! */
1054                 inc_ap_bio(mdev, 3);
1055
1056                 D_ASSERT(e_enr == s_enr + 1);
1057
1058                 while (__drbd_make_request(mdev, &bp->bio1, start_time))
1059                         inc_ap_bio(mdev, 1);
1060
1061                 while (__drbd_make_request(mdev, &bp->bio2, start_time))
1062                         inc_ap_bio(mdev, 1);
1063
1064                 dec_ap_bio(mdev);
1065
1066                 bio_pair_release(bp);
1067         }
1068         return 0;
1069 }
1070
1071 /* This is called by bio_add_page().  With this function we reduce
1072  * the number of BIOs that span over multiple DRBD_MAX_BIO_SIZEs
1073  * units (was AL_EXTENTs).
1074  *
1075  * we do the calculation within the lower 32bit of the byte offsets,
1076  * since we don't care for actual offset, but only check whether it
1077  * would cross "activity log extent" boundaries.
1078  *
1079  * As long as the BIO is empty we have to allow at least one bvec,
1080  * regardless of size and offset.  so the resulting bio may still
1081  * cross extent boundaries.  those are dealt with (bio_split) in
1082  * drbd_make_request.
1083  */
1084 int drbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bvm, struct bio_vec *bvec)
1085 {
1086         struct drbd_conf *mdev = (struct drbd_conf *) q->queuedata;
1087         unsigned int bio_offset =
1088                 (unsigned int)bvm->bi_sector << 9; /* 32 bit */
1089         unsigned int bio_size = bvm->bi_size;
1090         int limit, backing_limit;
1091
1092         limit = DRBD_MAX_BIO_SIZE
1093               - ((bio_offset & (DRBD_MAX_BIO_SIZE-1)) + bio_size);
1094         if (limit < 0)
1095                 limit = 0;
1096         if (bio_size == 0) {
1097                 if (limit <= bvec->bv_len)
1098                         limit = bvec->bv_len;
1099         } else if (limit && get_ldev(mdev)) {
1100                 struct request_queue * const b =
1101                         mdev->ldev->backing_bdev->bd_disk->queue;
1102                 if (b->merge_bvec_fn) {
1103                         backing_limit = b->merge_bvec_fn(b, bvm, bvec);
1104                         limit = min(limit, backing_limit);
1105                 }
1106                 put_ldev(mdev);
1107         }
1108         return limit;
1109 }
1110
1111 void request_timer_fn(unsigned long data)
1112 {
1113         struct drbd_conf *mdev = (struct drbd_conf *) data;
1114         struct drbd_request *req; /* oldest request */
1115         struct list_head *le;
1116         unsigned long et = 0; /* effective timeout = ko_count * timeout */
1117
1118         if (get_net_conf(mdev->tconn)) {
1119                 et = mdev->tconn->net_conf->timeout*HZ/10 * mdev->tconn->net_conf->ko_count;
1120                 put_net_conf(mdev->tconn);
1121         }
1122         if (!et || mdev->state.conn < C_WF_REPORT_PARAMS)
1123                 return; /* Recurring timer stopped */
1124
1125         spin_lock_irq(&mdev->tconn->req_lock);
1126         le = &mdev->tconn->oldest_tle->requests;
1127         if (list_empty(le)) {
1128                 spin_unlock_irq(&mdev->tconn->req_lock);
1129                 mod_timer(&mdev->request_timer, jiffies + et);
1130                 return;
1131         }
1132
1133         le = le->prev;
1134         req = list_entry(le, struct drbd_request, tl_requests);
1135         if (time_is_before_eq_jiffies(req->start_time + et)) {
1136                 if (req->rq_state & RQ_NET_PENDING) {
1137                         dev_warn(DEV, "Remote failed to finish a request within ko-count * timeout\n");
1138                         _drbd_set_state(_NS(mdev, conn, C_TIMEOUT), CS_VERBOSE, NULL);
1139                 } else {
1140                         dev_warn(DEV, "Local backing block device frozen?\n");
1141                         mod_timer(&mdev->request_timer, jiffies + et);
1142                 }
1143         } else {
1144                 mod_timer(&mdev->request_timer, req->start_time + et);
1145         }
1146
1147         spin_unlock_irq(&mdev->tconn->req_lock);
1148 }