drbd_req.c 41.2 KB
Newer Older
P
Philipp Reisner 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33
/*
   drbd_req.c

   This file is part of DRBD by Philipp Reisner and Lars Ellenberg.

   Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
   Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
   Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.

   drbd is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.

   drbd is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with drbd; see the file COPYING.  If not, write to
   the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.

 */

#include <linux/module.h>

#include <linux/slab.h>
#include <linux/drbd.h>
#include "drbd_int.h"
#include "drbd_req.h"


34 35
static bool drbd_may_do_local_read(struct drbd_conf *mdev, sector_t sector, int size);

P
Philipp Reisner 已提交
36
/* Update disk stats at start of I/O request */
37
static void _drbd_start_io_acct(struct drbd_conf *mdev, struct drbd_request *req)
P
Philipp Reisner 已提交
38
{
39
	const int rw = bio_data_dir(req->master_bio);
P
Philipp Reisner 已提交
40 41
	int cpu;
	cpu = part_stat_lock();
42
	part_round_stats(cpu, &mdev->vdisk->part0);
P
Philipp Reisner 已提交
43
	part_stat_inc(cpu, &mdev->vdisk->part0, ios[rw]);
44
	part_stat_add(cpu, &mdev->vdisk->part0, sectors[rw], req->i.size >> 9);
P
Philipp Reisner 已提交
45 46
	(void) cpu; /* The macro invocations above want the cpu argument, I do not like
		       the compiler warning about cpu only assigned but never used... */
47
	part_inc_in_flight(&mdev->vdisk->part0, rw);
P
Philipp Reisner 已提交
48 49 50 51 52 53 54 55 56 57 58 59
	part_stat_unlock();
}

/* Update disk stats when completing request upwards */
static void _drbd_end_io_acct(struct drbd_conf *mdev, struct drbd_request *req)
{
	int rw = bio_data_dir(req->master_bio);
	unsigned long duration = jiffies - req->start_time;
	int cpu;
	cpu = part_stat_lock();
	part_stat_add(cpu, &mdev->vdisk->part0, ticks[rw], duration);
	part_round_stats(cpu, &mdev->vdisk->part0);
60
	part_dec_in_flight(&mdev->vdisk->part0, rw);
P
Philipp Reisner 已提交
61 62 63
	part_stat_unlock();
}

64 65 66 67 68 69 70 71 72 73 74
static struct drbd_request *drbd_req_new(struct drbd_conf *mdev,
					       struct bio *bio_src)
{
	struct drbd_request *req;

	req = mempool_alloc(drbd_request_mempool, GFP_NOIO);
	if (!req)
		return NULL;

	drbd_req_make_private_bio(req, bio_src);
	req->rq_state    = bio_data_dir(bio_src) == WRITE ? RQ_WRITE : 0;
75
	req->w.mdev      = mdev;
76 77
	req->master_bio  = bio_src;
	req->epoch       = 0;
78

79 80 81
	drbd_clear_interval(&req->i);
	req->i.sector     = bio_src->bi_sector;
	req->i.size      = bio_src->bi_size;
82
	req->i.local = true;
83 84
	req->i.waiting = false;

85 86 87
	INIT_LIST_HEAD(&req->tl_requests);
	INIT_LIST_HEAD(&req->w.list);

88
	/* one reference to be put by __drbd_make_request */
89
	atomic_set(&req->completion_ref, 1);
90
	/* one kref as long as completion_ref > 0 */
91
	kref_init(&req->kref);
92 93 94
	return req;
}

95
void drbd_req_destroy(struct kref *kref)
P
Philipp Reisner 已提交
96
{
97 98
	struct drbd_request *req = container_of(kref, struct drbd_request, kref);
	struct drbd_conf *mdev = req->w.mdev;
99 100 101 102 103 104 105 106 107 108
	const unsigned s = req->rq_state;

	if ((req->master_bio && !(s & RQ_POSTPONED)) ||
		atomic_read(&req->completion_ref) ||
		(s & RQ_LOCAL_PENDING) ||
		((s & RQ_NET_MASK) && !(s & RQ_NET_DONE))) {
		dev_err(DEV, "drbd_req_destroy: Logic BUG rq_state = 0x%x, completion_ref = %d\n",
				s, atomic_read(&req->completion_ref));
		return;
	}
109 110 111 112 113 114 115

	/* remove it from the transfer log.
	 * well, only if it had been there in the first
	 * place... if it had not (local only or conflicting
	 * and never sent), it should still be "empty" as
	 * initialized in drbd_req_new(), so we can list_del() it
	 * here unconditionally */
116
	list_del_init(&req->tl_requests);
117

P
Philipp Reisner 已提交
118 119 120
	/* if it was a write, we may have to set the corresponding
	 * bit(s) out-of-sync first. If it had a local part, we need to
	 * release the reference to the activity log. */
121
	if (s & RQ_WRITE) {
P
Philipp Reisner 已提交
122 123 124 125 126
		/* Set out-of-sync unless both OK flags are set
		 * (local only or remote failed).
		 * Other places where we set out-of-sync:
		 * READ with local io-error */

127 128 129 130 131 132 133
		/* There is a special case:
		 * we may notice late that IO was suspended,
		 * and postpone, or schedule for retry, a write,
		 * before it even was submitted or sent.
		 * In that case we do not want to touch the bitmap at all.
		 */
		if ((s & (RQ_POSTPONED|RQ_LOCAL_MASK|RQ_NET_MASK)) != RQ_POSTPONED) {
P
Philipp Reisner 已提交
134 135
			if (!(s & RQ_NET_OK) || !(s & RQ_LOCAL_OK))
				drbd_set_out_of_sync(mdev, req->i.sector, req->i.size);
P
Philipp Reisner 已提交
136

P
Philipp Reisner 已提交
137 138 139
			if ((s & RQ_NET_OK) && (s & RQ_LOCAL_OK) && (s & RQ_NET_SIS))
				drbd_set_in_sync(mdev, req->i.sector, req->i.size);
		}
P
Philipp Reisner 已提交
140 141

		/* one might be tempted to move the drbd_al_complete_io
142
		 * to the local io completion callback drbd_request_endio.
P
Philipp Reisner 已提交
143 144 145 146 147 148 149 150
		 * but, if this was a mirror write, we may only
		 * drbd_al_complete_io after this is RQ_NET_DONE,
		 * otherwise the extent could be dropped from the al
		 * before it has actually been written on the peer.
		 * if we crash before our peer knows about the request,
		 * but after the extent has been dropped from the al,
		 * we would forget to resync the corresponding extent.
		 */
P
Philipp Reisner 已提交
151
		if (s & RQ_IN_ACT_LOG) {
P
Philipp Reisner 已提交
152
			if (get_ldev_if_state(mdev, D_FAILED)) {
P
Philipp Reisner 已提交
153
				drbd_al_complete_io(mdev, &req->i);
P
Philipp Reisner 已提交
154 155
				put_ldev(mdev);
			} else if (__ratelimit(&drbd_ratelimit_state)) {
156 157 158
				dev_warn(DEV, "Should have called drbd_al_complete_io(, %llu, %u), "
					 "but my Disk seems to have failed :(\n",
					 (unsigned long long) req->i.sector, req->i.size);
P
Philipp Reisner 已提交
159 160 161 162
			}
		}
	}

163
	mempool_free(req, drbd_request_mempool);
P
Philipp Reisner 已提交
164 165
}

166 167
static void wake_all_senders(struct drbd_tconn *tconn) {
	wake_up(&tconn->sender_work.q_wait);
P
Philipp Reisner 已提交
168 169
}

170
/* must hold resource->req_lock */
171
void start_new_tl_epoch(struct drbd_tconn *tconn)
P
Philipp Reisner 已提交
172
{
173 174 175
	/* no point closing an epoch, if it is empty, anyways. */
	if (tconn->current_tle_writes == 0)
		return;
P
Philipp Reisner 已提交
176

177 178 179
	tconn->current_tle_writes = 0;
	atomic_inc(&tconn->current_tle_nr);
	wake_all_senders(tconn);
P
Philipp Reisner 已提交
180 181 182 183 184 185 186 187 188
}

void complete_master_bio(struct drbd_conf *mdev,
		struct bio_and_error *m)
{
	bio_endio(m->bio, m->error);
	dec_ap_bio(mdev);
}

189 190 191 192

static void drbd_remove_request_interval(struct rb_root *root,
					 struct drbd_request *req)
{
193
	struct drbd_conf *mdev = req->w.mdev;
194 195 196 197 198 199 200 201 202
	struct drbd_interval *i = &req->i;

	drbd_remove_interval(root, i);

	/* Wake up any processes waiting for this request to complete.  */
	if (i->waiting)
		wake_up(&mdev->misc_wait);
}

P
Philipp Reisner 已提交
203 204 205 206 207 208
/* Helper for __req_mod().
 * Set m->bio to the master bio, if it is fit to be completed,
 * or leave it alone (it is initialized to NULL in __req_mod),
 * if it has already been completed, or cannot be completed yet.
 * If m->bio is set, the error status to be returned is placed in m->error.
 */
209
static
210
void drbd_req_complete(struct drbd_request *req, struct bio_and_error *m)
P
Philipp Reisner 已提交
211
{
212
	const unsigned s = req->rq_state;
213
	struct drbd_conf *mdev = req->w.mdev;
214 215
	int rw;
	int error, ok;
P
Philipp Reisner 已提交
216 217 218 219 220 221 222 223 224 225

	/* we must not complete the master bio, while it is
	 *	still being processed by _drbd_send_zc_bio (drbd_send_dblock)
	 *	not yet acknowledged by the peer
	 *	not yet completed by the local io subsystem
	 * these flags may get cleared in any order by
	 *	the worker,
	 *	the receiver,
	 *	the bio_endio completion callbacks.
	 */
226 227 228 229
	if ((s & RQ_LOCAL_PENDING && !(s & RQ_LOCAL_ABORTED)) ||
	    (s & RQ_NET_QUEUED) || (s & RQ_NET_PENDING) ||
	    (s & RQ_COMPLETION_SUSP)) {
		dev_err(DEV, "drbd_req_complete: Logic BUG rq_state = 0x%x\n", s);
P
Philipp Reisner 已提交
230
		return;
231 232 233 234
	}

	if (!req->master_bio) {
		dev_err(DEV, "drbd_req_complete: Logic BUG, master_bio == NULL!\n");
P
Philipp Reisner 已提交
235
		return;
236
	}
P
Philipp Reisner 已提交
237

238
	rw = bio_rw(req->master_bio);
P
Philipp Reisner 已提交
239

240 241 242 243 244 245 246 247 248 249 250 251 252 253 254
	/*
	 * figure out whether to report success or failure.
	 *
	 * report success when at least one of the operations succeeded.
	 * or, to put the other way,
	 * only report failure, when both operations failed.
	 *
	 * what to do about the failures is handled elsewhere.
	 * what we need to do here is just: complete the master_bio.
	 *
	 * local completion error, if any, has been stored as ERR_PTR
	 * in private_bio within drbd_request_endio.
	 */
	ok = (s & RQ_LOCAL_OK) || (s & RQ_NET_OK);
	error = PTR_ERR(req->private_bio);
P
Philipp Reisner 已提交
255

256 257 258 259
	/* remove the request from the conflict detection
	 * respective block_id verification hash */
	if (!drbd_interval_empty(&req->i)) {
		struct rb_root *root;
P
Philipp Reisner 已提交
260 261

		if (rw == WRITE)
262 263 264 265
			root = &mdev->write_requests;
		else
			root = &mdev->read_requests;
		drbd_remove_request_interval(root, req);
266
	}
P
Philipp Reisner 已提交
267

268 269 270 271 272 273 274 275 276 277
	/* Before we can signal completion to the upper layers,
	 * we may need to close the current transfer log epoch.
	 * We are within the request lock, so we can simply compare
	 * the request epoch number with the current transfer log
	 * epoch number.  If they match, increase the current_tle_nr,
	 * and reset the transfer log epoch write_cnt.
	 */
	if (rw == WRITE &&
	    req->epoch == atomic_read(&mdev->tconn->current_tle_nr))
		start_new_tl_epoch(mdev->tconn);
P
Philipp Reisner 已提交
278

279 280
	/* Update disk stats */
	_drbd_end_io_acct(mdev, req);
P
Philipp Reisner 已提交
281

282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297
	/* If READ failed,
	 * have it be pushed back to the retry work queue,
	 * so it will re-enter __drbd_make_request(),
	 * and be re-assigned to a suitable local or remote path,
	 * or failed if we do not have access to good data anymore.
	 *
	 * Unless it was failed early by __drbd_make_request(),
	 * because no path was available, in which case
	 * it was not even added to the transfer_log.
	 *
	 * READA may fail, and will not be retried.
	 *
	 * WRITE should have used all available paths already.
	 */
	if (!ok && rw == READ && !list_empty(&req->tl_requests))
		req->rq_state |= RQ_POSTPONED;
P
Philipp Reisner 已提交
298

299
	if (!(req->rq_state & RQ_POSTPONED)) {
P
Philipp Reisner 已提交
300 301 302 303 304 305
		m->error = ok ? 0 : (error ?: -EIO);
		m->bio = req->master_bio;
		req->master_bio = NULL;
	}
}

306
static int drbd_req_put_completion_ref(struct drbd_request *req, struct bio_and_error *m, int put)
307
{
308
	struct drbd_conf *mdev = req->w.mdev;
309 310 311 312
	D_ASSERT(m || (req->rq_state & RQ_POSTPONED));

	if (!atomic_sub_and_test(put, &req->completion_ref))
		return 0;
313

314
	drbd_req_complete(req, m);
315 316 317 318 319 320

	if (req->rq_state & RQ_POSTPONED) {
		/* don't destroy the req object just yet,
		 * but queue it for retry */
		drbd_restart_request(req);
		return 0;
P
Philipp Reisner 已提交
321
	}
322

323
	return 1;
P
Philipp Reisner 已提交
324 325
}

326 327 328 329
/* I'd like this to be the only place that manipulates
 * req->completion_ref and req->kref. */
static void mod_rq_state(struct drbd_request *req, struct bio_and_error *m,
		int clear, int set)
330
{
331 332 333 334
	struct drbd_conf *mdev = req->w.mdev;
	unsigned s = req->rq_state;
	int c_put = 0;
	int k_put = 0;
335

336 337
	if (drbd_suspended(mdev) && !((s | clear) & RQ_COMPLETION_SUSP))
		set |= RQ_COMPLETION_SUSP;
338

339
	/* apply */
P
Philipp Reisner 已提交
340

341 342
	req->rq_state &= ~clear;
	req->rq_state |= set;
P
Philipp Reisner 已提交
343

344 345 346
	/* no change? */
	if (req->rq_state == s)
		return;
P
Philipp Reisner 已提交
347

348 349 350 351 352 353 354 355
	/* intent: get references */

	if (!(s & RQ_LOCAL_PENDING) && (set & RQ_LOCAL_PENDING))
		atomic_inc(&req->completion_ref);

	if (!(s & RQ_NET_PENDING) && (set & RQ_NET_PENDING)) {
		inc_ap_pending(mdev);
		atomic_inc(&req->completion_ref);
P
Philipp Reisner 已提交
356 357
	}

358 359 360 361 362 363 364 365 366
	if (!(s & RQ_NET_QUEUED) && (set & RQ_NET_QUEUED))
		atomic_inc(&req->completion_ref);

	if (!(s & RQ_EXP_BARR_ACK) && (set & RQ_EXP_BARR_ACK))
		kref_get(&req->kref); /* wait for the DONE */

	if (!(s & RQ_NET_SENT) && (set & RQ_NET_SENT))
		atomic_add(req->i.size >> 9, &mdev->ap_in_flight);

367 368 369
	if (!(s & RQ_COMPLETION_SUSP) && (set & RQ_COMPLETION_SUSP))
		atomic_inc(&req->completion_ref);

370 371 372 373 374 375 376 377 378 379 380
	/* progress: put references */

	if ((s & RQ_COMPLETION_SUSP) && (clear & RQ_COMPLETION_SUSP))
		++c_put;

	if (!(s & RQ_LOCAL_ABORTED) && (set & RQ_LOCAL_ABORTED)) {
		D_ASSERT(req->rq_state & RQ_LOCAL_PENDING);
		/* local completion may still come in later,
		 * we need to keep the req object around. */
		kref_get(&req->kref);
		++c_put;
P
Philipp Reisner 已提交
381 382
	}

383 384 385 386 387 388
	if ((s & RQ_LOCAL_PENDING) && (clear & RQ_LOCAL_PENDING)) {
		if (req->rq_state & RQ_LOCAL_ABORTED)
			++k_put;
		else
			++c_put;
	}
P
Philipp Reisner 已提交
389

390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424
	if ((s & RQ_NET_PENDING) && (clear & RQ_NET_PENDING)) {
		dec_ap_pending(mdev);
		++c_put;
	}

	if ((s & RQ_NET_QUEUED) && (clear & RQ_NET_QUEUED))
		++c_put;

	if ((s & RQ_EXP_BARR_ACK) && !(s & RQ_NET_DONE) && (set & RQ_NET_DONE)) {
		if (req->rq_state & RQ_NET_SENT)
			atomic_sub(req->i.size >> 9, &mdev->ap_in_flight);
		++k_put;
	}

	/* potentially complete and destroy */

	if (k_put || c_put) {
		/* Completion does it's own kref_put.  If we are going to
		 * kref_sub below, we need req to be still around then. */
		int at_least = k_put + !!c_put;
		int refcount = atomic_read(&req->kref.refcount);
		if (refcount < at_least)
			dev_err(DEV,
				"mod_rq_state: Logic BUG: %x -> %x: refcount = %d, should be >= %d\n",
				s, req->rq_state, refcount, at_least);
	}

	/* If we made progress, retry conflicting peer requests, if any. */
	if (req->i.waiting)
		wake_up(&mdev->misc_wait);

	if (c_put)
		k_put += drbd_req_put_completion_ref(req, m, c_put);
	if (k_put)
		kref_sub(&req->kref, k_put, drbd_req_destroy);
P
Philipp Reisner 已提交
425 426
}

427 428 429 430
static void drbd_report_io_error(struct drbd_conf *mdev, struct drbd_request *req)
{
        char b[BDEVNAME_SIZE];

431
	if (!__ratelimit(&drbd_ratelimit_state))
432 433 434 435
		return;

	dev_warn(DEV, "local %s IO error sector %llu+%u on %s\n",
			(req->rq_state & RQ_WRITE) ? "WRITE" : "READ",
436 437
			(unsigned long long)req->i.sector,
			req->i.size >> 9,
438 439 440
			bdevname(mdev->ldev->backing_bdev, b));
}

P
Philipp Reisner 已提交
441 442 443 444 445 446 447 448 449 450 451 452
/* obviously this could be coded as many single functions
 * instead of one huge switch,
 * or by putting the code directly in the respective locations
 * (as it has been before).
 *
 * but having it this way
 *  enforces that it is all in this one place, where it is easier to audit,
 *  it makes it obvious that whatever "event" "happens" to a request should
 *  happen "atomically" within the req_lock,
 *  and it enforces that we have to think in a very structured manner
 *  about the "events" that may happen to a request during its life time ...
 */
453
int __req_mod(struct drbd_request *req, enum drbd_req_event what,
P
Philipp Reisner 已提交
454 455
		struct bio_and_error *m)
{
456
	struct drbd_conf *mdev = req->w.mdev;
457
	struct net_conf *nc;
458
	int p, rv = 0;
459 460 461

	if (m)
		m->bio = NULL;
P
Philipp Reisner 已提交
462 463 464 465 466 467 468 469

	switch (what) {
	default:
		dev_err(DEV, "LOGIC BUG in %s:%u\n", __FILE__ , __LINE__);
		break;

	/* does not happen...
	 * initialization done in drbd_req_new
470
	case CREATED:
P
Philipp Reisner 已提交
471 472 473
		break;
		*/

474
	case TO_BE_SENT: /* via network */
475
		/* reached via __drbd_make_request
P
Philipp Reisner 已提交
476 477
		 * and from w_read_retry_remote */
		D_ASSERT(!(req->rq_state & RQ_NET_MASK));
478 479 480 481
		rcu_read_lock();
		nc = rcu_dereference(mdev->tconn->net_conf);
		p = nc->wire_protocol;
		rcu_read_unlock();
482 483 484
		req->rq_state |=
			p == DRBD_PROT_C ? RQ_EXP_WRITE_ACK :
			p == DRBD_PROT_B ? RQ_EXP_RECEIVE_ACK : 0;
485
		mod_rq_state(req, m, 0, RQ_NET_PENDING);
P
Philipp Reisner 已提交
486 487
		break;

488
	case TO_BE_SUBMITTED: /* locally */
489
		/* reached via __drbd_make_request */
P
Philipp Reisner 已提交
490
		D_ASSERT(!(req->rq_state & RQ_LOCAL_MASK));
491
		mod_rq_state(req, m, 0, RQ_LOCAL_PENDING);
P
Philipp Reisner 已提交
492 493
		break;

494
	case COMPLETED_OK:
495
		if (req->rq_state & RQ_WRITE)
496
			mdev->writ_cnt += req->i.size >> 9;
P
Philipp Reisner 已提交
497
		else
498
			mdev->read_cnt += req->i.size >> 9;
P
Philipp Reisner 已提交
499

500 501
		mod_rq_state(req, m, RQ_LOCAL_PENDING,
				RQ_LOCAL_COMPLETED|RQ_LOCAL_OK);
P
Philipp Reisner 已提交
502 503
		break;

504
	case ABORT_DISK_IO:
505
		mod_rq_state(req, m, 0, RQ_LOCAL_ABORTED);
506 507
		break;

508
	case WRITE_COMPLETED_WITH_ERROR:
509
		drbd_report_io_error(mdev, req);
510
		__drbd_chk_io_error(mdev, DRBD_WRITE_ERROR);
511
		mod_rq_state(req, m, RQ_LOCAL_PENDING, RQ_LOCAL_COMPLETED);
P
Philipp Reisner 已提交
512 513
		break;

514
	case READ_COMPLETED_WITH_ERROR:
515
		drbd_set_out_of_sync(mdev, req->i.sector, req->i.size);
516
		drbd_report_io_error(mdev, req);
517
		__drbd_chk_io_error(mdev, DRBD_READ_ERROR);
518 519 520 521
		/* fall through. */
	case READ_AHEAD_COMPLETED_WITH_ERROR:
		/* it is legal to fail READA, no __drbd_chk_io_error in that case. */
		mod_rq_state(req, m, RQ_LOCAL_PENDING, RQ_LOCAL_COMPLETED);
522
		break;
P
Philipp Reisner 已提交
523

524
	case QUEUE_FOR_NET_READ:
P
Philipp Reisner 已提交
525 526 527 528
		/* READ or READA, and
		 * no local disk,
		 * or target area marked as invalid,
		 * or just got an io-error. */
529
		/* from __drbd_make_request
P
Philipp Reisner 已提交
530 531
		 * or from bio_endio during read io-error recovery */

532 533
		/* So we can verify the handle in the answer packet.
		 * Corresponding drbd_remove_request_interval is in
534
		 * drbd_req_complete() */
535
		D_ASSERT(drbd_interval_empty(&req->i));
536
		drbd_insert_interval(&mdev->read_requests, &req->i);
P
Philipp Reisner 已提交
537

538
		set_bit(UNPLUG_REMOTE, &mdev->flags);
P
Philipp Reisner 已提交
539 540

		D_ASSERT(req->rq_state & RQ_NET_PENDING);
541
		D_ASSERT((req->rq_state & RQ_LOCAL_MASK) == 0);
542
		mod_rq_state(req, m, 0, RQ_NET_QUEUED);
543
		req->w.cb = w_send_read_req;
544
		drbd_queue_work(&mdev->tconn->sender_work, &req->w);
P
Philipp Reisner 已提交
545 546
		break;

547
	case QUEUE_FOR_NET_WRITE:
P
Philipp Reisner 已提交
548
		/* assert something? */
549
		/* from __drbd_make_request only */
P
Philipp Reisner 已提交
550

551
		/* Corresponding drbd_remove_request_interval is in
552
		 * drbd_req_complete() */
553
		D_ASSERT(drbd_interval_empty(&req->i));
554
		drbd_insert_interval(&mdev->write_requests, &req->i);
P
Philipp Reisner 已提交
555 556 557 558 559 560 561 562 563 564

		/* NOTE
		 * In case the req ended up on the transfer log before being
		 * queued on the worker, it could lead to this request being
		 * missed during cleanup after connection loss.
		 * So we have to do both operations here,
		 * within the same lock that protects the transfer log.
		 *
		 * _req_add_to_epoch(req); this has to be after the
		 * _maybe_start_new_epoch(req); which happened in
565
		 * __drbd_make_request, because we now may set the bit
P
Philipp Reisner 已提交
566 567 568 569
		 * again ourselves to close the current epoch.
		 *
		 * Add req to the (now) current epoch (barrier). */

570 571 572 573
		/* otherwise we may lose an unplug, which may cause some remote
		 * io-scheduler timeout to expire, increasing maximum latency,
		 * hurting performance. */
		set_bit(UNPLUG_REMOTE, &mdev->flags);
P
Philipp Reisner 已提交
574 575 576

		/* queue work item to send data */
		D_ASSERT(req->rq_state & RQ_NET_PENDING);
577
		mod_rq_state(req, m, 0, RQ_NET_QUEUED|RQ_EXP_BARR_ACK);
P
Philipp Reisner 已提交
578
		req->w.cb =  w_send_dblock;
579
		drbd_queue_work(&mdev->tconn->sender_work, &req->w);
P
Philipp Reisner 已提交
580 581

		/* close the epoch, in case it outgrew the limit */
582 583 584 585
		rcu_read_lock();
		nc = rcu_dereference(mdev->tconn->net_conf);
		p = nc->max_epoch_size;
		rcu_read_unlock();
586 587
		if (mdev->tconn->current_tle_writes >= p)
			start_new_tl_epoch(mdev->tconn);
P
Philipp Reisner 已提交
588 589 590

		break;

591
	case QUEUE_FOR_SEND_OOS:
592
		mod_rq_state(req, m, 0, RQ_NET_QUEUED);
593
		req->w.cb =  w_send_out_of_sync;
594
		drbd_queue_work(&mdev->tconn->sender_work, &req->w);
595 596
		break;

597
	case READ_RETRY_REMOTE_CANCELED:
598 599
	case SEND_CANCELED:
	case SEND_FAILED:
P
Philipp Reisner 已提交
600 601
		/* real cleanup will be done from tl_clear.  just update flags
		 * so it is no longer marked as on the worker queue */
602
		mod_rq_state(req, m, RQ_NET_QUEUED, 0);
P
Philipp Reisner 已提交
603 604
		break;

605
	case HANDED_OVER_TO_NETWORK:
P
Philipp Reisner 已提交
606 607
		/* assert something? */
		if (bio_data_dir(req->master_bio) == WRITE &&
608
		    !(req->rq_state & (RQ_EXP_RECEIVE_ACK | RQ_EXP_WRITE_ACK))) {
P
Philipp Reisner 已提交
609 610
			/* this is what is dangerous about protocol A:
			 * pretend it was successfully written on the peer. */
611 612 613
			if (req->rq_state & RQ_NET_PENDING)
				mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_OK);
			/* else: neg-ack was faster... */
P
Philipp Reisner 已提交
614 615 616 617
			/* it is still not yet RQ_NET_DONE until the
			 * corresponding epoch barrier got acked as well,
			 * so we know what to dirty on connection loss */
		}
618
		mod_rq_state(req, m, RQ_NET_QUEUED, RQ_NET_SENT);
619 620
		break;

621
	case OOS_HANDED_TO_NETWORK:
622 623
		/* Was not set PENDING, no longer QUEUED, so is now DONE
		 * as far as this connection is concerned. */
624
		mod_rq_state(req, m, RQ_NET_QUEUED, RQ_NET_DONE);
P
Philipp Reisner 已提交
625 626
		break;

627
	case CONNECTION_LOST_WHILE_PENDING:
P
Philipp Reisner 已提交
628
		/* transfer log cleanup after connection loss */
629 630 631
		mod_rq_state(req, m,
				RQ_NET_OK|RQ_NET_PENDING|RQ_COMPLETION_SUSP,
				RQ_NET_DONE);
P
Philipp Reisner 已提交
632 633
		break;

634 635
	case CONFLICT_RESOLVED:
		/* for superseded conflicting writes of multiple primaries,
P
Philipp Reisner 已提交
636
		 * there is no need to keep anything in the tl, potential
637 638 639
		 * node crashes are covered by the activity log.
		 *
		 * If this request had been marked as RQ_POSTPONED before,
640
		 * it will actually not be completed, but "restarted",
641 642 643 644 645 646
		 * resubmitted from the retry worker context. */
		D_ASSERT(req->rq_state & RQ_NET_PENDING);
		D_ASSERT(req->rq_state & RQ_EXP_WRITE_ACK);
		mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_DONE|RQ_NET_OK);
		break;

647
	case WRITE_ACKED_BY_PEER_AND_SIS:
648
		req->rq_state |= RQ_NET_SIS;
649
	case WRITE_ACKED_BY_PEER:
650
		D_ASSERT(req->rq_state & RQ_EXP_WRITE_ACK);
P
Philipp Reisner 已提交
651
		/* protocol C; successfully written on peer.
652
		 * Nothing more to do here.
P
Philipp Reisner 已提交
653
		 * We want to keep the tl in place for all protocols, to cater
654
		 * for volatile write-back caches on lower level devices. */
P
Philipp Reisner 已提交
655

656
		goto ack_common;
657
	case RECV_ACKED_BY_PEER:
658
		D_ASSERT(req->rq_state & RQ_EXP_RECEIVE_ACK);
P
Philipp Reisner 已提交
659
		/* protocol B; pretends to be successfully written on peer.
660
		 * see also notes above in HANDED_OVER_TO_NETWORK about
P
Philipp Reisner 已提交
661
		 * protocol != C */
662
	ack_common:
P
Philipp Reisner 已提交
663
		D_ASSERT(req->rq_state & RQ_NET_PENDING);
664
		mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_OK);
P
Philipp Reisner 已提交
665 666
		break;

667
	case POSTPONE_WRITE:
668 669
		D_ASSERT(req->rq_state & RQ_EXP_WRITE_ACK);
		/* If this node has already detected the write conflict, the
670 671 672 673 674
		 * worker will be waiting on misc_wait.  Wake it up once this
		 * request has completed locally.
		 */
		D_ASSERT(req->rq_state & RQ_NET_PENDING);
		req->rq_state |= RQ_POSTPONED;
675 676 677 678 679
		if (req->i.waiting)
			wake_up(&mdev->misc_wait);
		/* Do not clear RQ_NET_PENDING. This request will make further
		 * progress via restart_conflicting_writes() or
		 * fail_postponed_requests(). Hopefully. */
680
		break;
P
Philipp Reisner 已提交
681

682
	case NEG_ACKED:
683
		mod_rq_state(req, m, RQ_NET_OK|RQ_NET_PENDING, 0);
P
Philipp Reisner 已提交
684 685
		break;

686
	case FAIL_FROZEN_DISK_IO:
687 688
		if (!(req->rq_state & RQ_LOCAL_COMPLETED))
			break;
689
		mod_rq_state(req, m, RQ_COMPLETION_SUSP, 0);
690 691
		break;

692
	case RESTART_FROZEN_DISK_IO:
693 694 695
		if (!(req->rq_state & RQ_LOCAL_COMPLETED))
			break;

696 697 698
		mod_rq_state(req, m,
				RQ_COMPLETION_SUSP|RQ_LOCAL_COMPLETED,
				RQ_LOCAL_PENDING);
699 700 701 702 703

		rv = MR_READ;
		if (bio_data_dir(req->master_bio) == WRITE)
			rv = MR_WRITE;

704
		get_ldev(mdev); /* always succeeds in this call path */
705
		req->w.cb = w_restart_disk_io;
706
		drbd_queue_work(&mdev->tconn->sender_work, &req->w);
707 708
		break;

709
	case RESEND:
710 711
		/* Simply complete (local only) READs. */
		if (!(req->rq_state & RQ_WRITE) && !req->w.cb) {
712
			mod_rq_state(req, m, RQ_COMPLETION_SUSP, 0);
713 714 715
			break;
		}

716
		/* If RQ_NET_OK is already set, we got a P_WRITE_ACK or P_RECV_ACK
717 718
		   before the connection loss (B&C only); only P_BARRIER_ACK
		   (or the local completion?) was missing when we suspended.
719 720
		   Throwing them out of the TL here by pretending we got a BARRIER_ACK.
		   During connection handshake, we ensure that the peer was not rebooted. */
721
		if (!(req->rq_state & RQ_NET_OK)) {
722 723 724 725
			/* FIXME could this possibly be a req->w.cb == w_send_out_of_sync?
			 * in that case we must not set RQ_NET_PENDING. */

			mod_rq_state(req, m, RQ_COMPLETION_SUSP, RQ_NET_QUEUED|RQ_NET_PENDING);
726
			if (req->w.cb) {
727
				drbd_queue_work(&mdev->tconn->sender_work, &req->w);
728
				rv = req->rq_state & RQ_WRITE ? MR_WRITE : MR_READ;
729
			} /* else: FIXME can this happen? */
730 731
			break;
		}
732
		/* else, fall through to BARRIER_ACKED */
733

734
	case BARRIER_ACKED:
735
		/* barrier ack for READ requests does not make sense */
736 737 738
		if (!(req->rq_state & RQ_WRITE))
			break;

P
Philipp Reisner 已提交
739
		if (req->rq_state & RQ_NET_PENDING) {
740
			/* barrier came in before all requests were acked.
P
Philipp Reisner 已提交
741 742
			 * this is bad, because if the connection is lost now,
			 * we won't be able to clean them up... */
743
			dev_err(DEV, "FIXME (BARRIER_ACKED but pending)\n");
P
Philipp Reisner 已提交
744
		}
745 746 747 748 749 750
		/* Allowed to complete requests, even while suspended.
		 * As this is called for all requests within a matching epoch,
		 * we need to filter, and only set RQ_NET_DONE for those that
		 * have actually been on the wire. */
		mod_rq_state(req, m, RQ_COMPLETION_SUSP,
				(req->rq_state & RQ_NET_MASK) ? RQ_NET_DONE : 0);
P
Philipp Reisner 已提交
751 752
		break;

753
	case DATA_RECEIVED:
P
Philipp Reisner 已提交
754
		D_ASSERT(req->rq_state & RQ_NET_PENDING);
755
		mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_OK|RQ_NET_DONE);
P
Philipp Reisner 已提交
756
		break;
757 758 759 760 761

	case QUEUE_AS_DRBD_BARRIER:
		start_new_tl_epoch(mdev->tconn);
		mod_rq_state(req, m, 0, RQ_NET_OK|RQ_NET_DONE);
		break;
P
Philipp Reisner 已提交
762
	};
763 764

	return rv;
P
Philipp Reisner 已提交
765 766 767 768 769 770 771 772 773
}

/* we may do a local read if:
 * - we are consistent (of course),
 * - or we are generally inconsistent,
 *   BUT we are still/already IN SYNC for this area.
 *   since size may be bigger than BM_BLOCK_SIZE,
 *   we may need to check several bits.
 */
774
static bool drbd_may_do_local_read(struct drbd_conf *mdev, sector_t sector, int size)
P
Philipp Reisner 已提交
775 776 777 778 779
{
	unsigned long sbnr, ebnr;
	sector_t esector, nr_sectors;

	if (mdev->state.disk == D_UP_TO_DATE)
780
		return true;
781
	if (mdev->state.disk != D_INCONSISTENT)
782
		return false;
P
Philipp Reisner 已提交
783
	esector = sector + (size >> 9) - 1;
784
	nr_sectors = drbd_get_capacity(mdev->this_bdev);
P
Philipp Reisner 已提交
785 786 787 788 789 790
	D_ASSERT(sector  < nr_sectors);
	D_ASSERT(esector < nr_sectors);

	sbnr = BM_SECT_TO_BIT(sector);
	ebnr = BM_SECT_TO_BIT(esector);

791
	return drbd_bm_count_bits(mdev, sbnr, ebnr) == 0;
P
Philipp Reisner 已提交
792 793
}

794 795
static bool remote_due_to_read_balancing(struct drbd_conf *mdev, sector_t sector,
		enum drbd_read_balancing rbm)
796 797
{
	struct backing_dev_info *bdi;
798
	int stripe_shift;
799 800 801 802 803 804 805 806

	switch (rbm) {
	case RB_CONGESTED_REMOTE:
		bdi = &mdev->ldev->backing_bdev->bd_disk->queue->backing_dev_info;
		return bdi_read_congested(bdi);
	case RB_LEAST_PENDING:
		return atomic_read(&mdev->local_cnt) >
			atomic_read(&mdev->ap_pending_cnt) + atomic_read(&mdev->rs_pending_cnt);
807 808 809 810 811 812 813 814
	case RB_32K_STRIPING:  /* stripe_shift = 15 */
	case RB_64K_STRIPING:
	case RB_128K_STRIPING:
	case RB_256K_STRIPING:
	case RB_512K_STRIPING:
	case RB_1M_STRIPING:   /* stripe_shift = 20 */
		stripe_shift = (rbm - RB_32K_STRIPING + 15);
		return (sector >> (stripe_shift - 9)) & 1;
815 816 817 818 819 820 821 822 823 824
	case RB_ROUND_ROBIN:
		return test_and_change_bit(READ_BALANCE_RR, &mdev->flags);
	case RB_PREFER_REMOTE:
		return true;
	case RB_PREFER_LOCAL:
	default:
		return false;
	}
}

825 826 827 828 829 830
/*
 * complete_conflicting_writes  -  wait for any conflicting write requests
 *
 * The write_requests tree contains all active write requests which we
 * currently know about.  Wait for any requests to complete which conflict with
 * the new one.
831 832
 *
 * Only way out: remove the conflicting intervals from the tree.
833
 */
834
static void complete_conflicting_writes(struct drbd_request *req)
835
{
836 837 838 839 840 841 842 843 844
	DEFINE_WAIT(wait);
	struct drbd_conf *mdev = req->w.mdev;
	struct drbd_interval *i;
	sector_t sector = req->i.sector;
	int size = req->i.size;

	i = drbd_find_overlap(&mdev->write_requests, sector, size);
	if (!i)
		return;
845

846 847
	for (;;) {
		prepare_to_wait(&mdev->misc_wait, &wait, TASK_UNINTERRUPTIBLE);
848 849
		i = drbd_find_overlap(&mdev->write_requests, sector, size);
		if (!i)
850 851 852 853 854 855
			break;
		/* Indicate to wake up device->misc_wait on progress.  */
		i->waiting = true;
		spin_unlock_irq(&mdev->tconn->req_lock);
		schedule();
		spin_lock_irq(&mdev->tconn->req_lock);
856
	}
857
	finish_wait(&mdev->misc_wait, &wait);
P
Philipp Reisner 已提交
858 859
}

860
/* called within req_lock and rcu_read_lock() */
861 862
static void maybe_pull_ahead(struct drbd_conf *mdev)
{
863 864 865 866 867
	struct drbd_tconn *tconn = mdev->tconn;
	struct net_conf *nc;
	bool congested = false;
	enum drbd_on_congestion on_congestion;

868
	rcu_read_lock();
869 870
	nc = rcu_dereference(tconn->net_conf);
	on_congestion = nc ? nc->on_congestion : OC_BLOCK;
871
	rcu_read_unlock();
872 873
	if (on_congestion == OC_BLOCK ||
	    tconn->agreed_pro_version < 96)
874
		return;
875 876 877 878 879 880 881 882

	/* If I don't even have good local storage, we can not reasonably try
	 * to pull ahead of the peer. We also need the local reference to make
	 * sure mdev->act_log is there.
	 */
	if (!get_ldev_if_state(mdev, D_UP_TO_DATE))
		return;

883 884
	if (nc->cong_fill &&
	    atomic_read(&mdev->ap_in_flight) >= nc->cong_fill) {
885
		dev_info(DEV, "Congestion-fill threshold reached\n");
886
		congested = true;
887 888
	}

889
	if (mdev->act_log->used >= nc->cong_extents) {
890
		dev_info(DEV, "Congestion-extents threshold reached\n");
891
		congested = true;
892 893 894
	}

	if (congested) {
895 896
		/* start a new epoch for non-mirrored writes */
		start_new_tl_epoch(mdev->tconn);
897

898
		if (on_congestion == OC_PULL_AHEAD)
899
			_drbd_set_state(_NS(mdev, conn, C_AHEAD), 0, NULL);
900
		else  /*nc->on_congestion == OC_DISCONNECT */
901 902 903 904 905
			_drbd_set_state(_NS(mdev, conn, C_DISCONNECTING), 0, NULL);
	}
	put_ldev(mdev);
}

906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931
/* If this returns false, and req->private_bio is still set,
 * this should be submitted locally.
 *
 * If it returns false, but req->private_bio is not set,
 * we do not have access to good data :(
 *
 * Otherwise, this destroys req->private_bio, if any,
 * and returns true.
 */
static bool do_remote_read(struct drbd_request *req)
{
	struct drbd_conf *mdev = req->w.mdev;
	enum drbd_read_balancing rbm;

	if (req->private_bio) {
		if (!drbd_may_do_local_read(mdev,
					req->i.sector, req->i.size)) {
			bio_put(req->private_bio);
			req->private_bio = NULL;
			put_ldev(mdev);
		}
	}

	if (mdev->state.pdsk != D_UP_TO_DATE)
		return false;

932 933 934
	if (req->private_bio == NULL)
		return true;

935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967
	/* TODO: improve read balancing decisions, take into account drbd
	 * protocol, pending requests etc. */

	rcu_read_lock();
	rbm = rcu_dereference(mdev->ldev->disk_conf)->read_balancing;
	rcu_read_unlock();

	if (rbm == RB_PREFER_LOCAL && req->private_bio)
		return false; /* submit locally */

	if (remote_due_to_read_balancing(mdev, req->i.sector, rbm)) {
		if (req->private_bio) {
			bio_put(req->private_bio);
			req->private_bio = NULL;
			put_ldev(mdev);
		}
		return true;
	}

	return false;
}

/* returns number of connections (== 1, for drbd 8.4)
 * expected to actually write this data,
 * which does NOT include those that we are L_AHEAD for. */
static int drbd_process_write_request(struct drbd_request *req)
{
	struct drbd_conf *mdev = req->w.mdev;
	int remote, send_oos;

	remote = drbd_should_do_remote(mdev->state);
	send_oos = drbd_should_send_out_of_sync(mdev->state);

968 969 970 971 972 973 974 975 976
	/* Need to replicate writes.  Unless it is an empty flush,
	 * which is better mapped to a DRBD P_BARRIER packet,
	 * also for drbd wire protocol compatibility reasons.
	 * If this was a flush, just start a new epoch.
	 * Unless the current epoch was empty anyways, or we are not currently
	 * replicating, in which case there is no point. */
	if (unlikely(req->i.size == 0)) {
		/* The only size==0 bios we expect are empty flushes. */
		D_ASSERT(req->master_bio->bi_rw & REQ_FLUSH);
977
		if (remote)
978 979
			_req_mod(req, QUEUE_AS_DRBD_BARRIER);
		return remote;
980 981
	}

982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022
	if (!remote && !send_oos)
		return 0;

	D_ASSERT(!(remote && send_oos));

	if (remote) {
		_req_mod(req, TO_BE_SENT);
		_req_mod(req, QUEUE_FOR_NET_WRITE);
	} else if (drbd_set_out_of_sync(mdev, req->i.sector, req->i.size))
		_req_mod(req, QUEUE_FOR_SEND_OOS);

	return remote;
}

static void
drbd_submit_req_private_bio(struct drbd_request *req)
{
	struct drbd_conf *mdev = req->w.mdev;
	struct bio *bio = req->private_bio;
	const int rw = bio_rw(bio);

	bio->bi_bdev = mdev->ldev->backing_bdev;

	/* State may have changed since we grabbed our reference on the
	 * ->ldev member. Double check, and short-circuit to endio.
	 * In case the last activity log transaction failed to get on
	 * stable storage, and this is a WRITE, we may not even submit
	 * this bio. */
	if (get_ldev(mdev)) {
		if (drbd_insert_fault(mdev,
				      rw == WRITE ? DRBD_FAULT_DT_WR
				    : rw == READ  ? DRBD_FAULT_DT_RD
				    :               DRBD_FAULT_DT_RA))
			bio_endio(bio, -EIO);
		else
			generic_make_request(bio);
		put_ldev(mdev);
	} else
		bio_endio(bio, -EIO);
}

1023 1024 1025 1026 1027 1028 1029 1030
static void drbd_queue_write(struct drbd_conf *mdev, struct drbd_request *req)
{
	spin_lock(&mdev->submit.lock);
	list_add_tail(&req->tl_requests, &mdev->submit.writes);
	spin_unlock(&mdev->submit.lock);
	queue_work(mdev->submit.wq, &mdev->submit.worker);
}

1031 1032 1033 1034 1035 1036 1037
/* returns the new drbd_request pointer, if the caller is expected to
 * drbd_send_and_submit() it (to save latency), or NULL if we queued the
 * request on the submitter thread.
 * Returns ERR_PTR(-ENOMEM) if we cannot allocate a drbd_request.
 */
struct drbd_request *
drbd_request_prepare(struct drbd_conf *mdev, struct bio *bio, unsigned long start_time)
P
Philipp Reisner 已提交
1038
{
1039
	const int rw = bio_data_dir(bio);
P
Philipp Reisner 已提交
1040 1041 1042 1043 1044 1045 1046 1047 1048 1049
	struct drbd_request *req;

	/* allocate outside of all locks; */
	req = drbd_req_new(mdev, bio);
	if (!req) {
		dec_ap_bio(mdev);
		/* only pass the error to the upper layers.
		 * if user cannot handle io errors, that's not our business. */
		dev_err(DEV, "could not kmalloc() req\n");
		bio_endio(bio, -ENOMEM);
1050
		return ERR_PTR(-ENOMEM);
P
Philipp Reisner 已提交
1051
	}
1052
	req->start_time = start_time;
P
Philipp Reisner 已提交
1053

1054 1055
	if (!get_ldev(mdev)) {
		bio_put(req->private_bio);
P
Philipp Reisner 已提交
1056 1057 1058
		req->private_bio = NULL;
	}

1059 1060 1061
	/* Update disk stats */
	_drbd_start_io_acct(mdev, req);

1062
	if (rw == WRITE && req->private_bio && req->i.size
1063
	&& !test_bit(AL_SUSPENDED, &mdev->flags)) {
1064 1065 1066 1067
		if (!drbd_al_begin_io_fastpath(mdev, &req->i)) {
			drbd_queue_write(mdev, req);
			return NULL;
		}
1068 1069
		req->rq_state |= RQ_IN_ACT_LOG;
	}
P
Philipp Reisner 已提交
1070

1071 1072 1073 1074 1075 1076 1077 1078 1079
	return req;
}

static void drbd_send_and_submit(struct drbd_conf *mdev, struct drbd_request *req)
{
	const int rw = bio_rw(req->master_bio);
	struct bio_and_error m = { NULL, };
	bool no_remote = false;

1080
	spin_lock_irq(&mdev->tconn->req_lock);
1081
	if (rw == WRITE) {
1082 1083 1084 1085
		/* This may temporarily give up the req_lock,
		 * but will re-aquire it before it returns here.
		 * Needs to be before the check on drbd_suspended() */
		complete_conflicting_writes(req);
1086 1087 1088 1089 1090
		/* no more giving up req_lock from now on! */

		/* check for congestion, and potentially stop sending
		 * full data updates, but start sending "dirty bits" only. */
		maybe_pull_ahead(mdev);
P
Philipp Reisner 已提交
1091 1092
	}

1093

1094 1095 1096 1097 1098 1099
	if (drbd_suspended(mdev)) {
		/* push back and retry: */
		req->rq_state |= RQ_POSTPONED;
		if (req->private_bio) {
			bio_put(req->private_bio);
			req->private_bio = NULL;
P
Philipp Reisner 已提交
1100
			put_ldev(mdev);
P
Philipp Reisner 已提交
1101
		}
1102
		goto out;
P
Philipp Reisner 已提交
1103 1104
	}

1105 1106
	/* We fail READ/READA early, if we can not serve it.
	 * We must do this before req is registered on any lists.
1107
	 * Otherwise, drbd_req_complete() will queue failed READ for retry. */
1108 1109 1110
	if (rw != WRITE) {
		if (!do_remote_read(req) && !req->private_bio)
			goto nodata;
P
Philipp Reisner 已提交
1111 1112
	}

1113 1114
	/* which transfer log epoch does this belong to? */
	req->epoch = atomic_read(&mdev->tconn->current_tle_nr);
1115

1116 1117
	/* no point in adding empty flushes to the transfer log,
	 * they are mapped to drbd barriers already. */
1118 1119 1120
	if (likely(req->i.size!=0)) {
		if (rw == WRITE)
			mdev->tconn->current_tle_writes++;
1121

1122
		list_add_tail(&req->tl_requests, &mdev->tconn->transfer_log);
P
Philipp Reisner 已提交
1123
	}
1124

1125 1126 1127 1128 1129 1130 1131 1132 1133
	if (rw == WRITE) {
		if (!drbd_process_write_request(req))
			no_remote = true;
	} else {
		/* We either have a private_bio, or we can read from remote.
		 * Otherwise we had done the goto nodata above. */
		if (req->private_bio == NULL) {
			_req_mod(req, TO_BE_SENT);
			_req_mod(req, QUEUE_FOR_NET_READ);
1134
		} else
1135
			no_remote = true;
P
Philipp Reisner 已提交
1136 1137
	}

1138 1139 1140 1141 1142 1143
	if (req->private_bio) {
		/* needs to be marked within the same spinlock */
		_req_mod(req, TO_BE_SUBMITTED);
		/* but we need to give up the spinlock to submit */
		spin_unlock_irq(&mdev->tconn->req_lock);
		drbd_submit_req_private_bio(req);
1144
		spin_lock_irq(&mdev->tconn->req_lock);
1145 1146 1147
	} else if (no_remote) {
nodata:
		if (__ratelimit(&drbd_ratelimit_state))
1148 1149
			dev_err(DEV, "IO ERROR: neither local nor remote data, sector %llu+%u\n",
					(unsigned long long)req->i.sector, req->i.size >> 9);
1150
		/* A write may have been queued for send_oos, however.
1151
		 * So we can not simply free it, we must go through drbd_req_put_completion_ref() */
P
Philipp Reisner 已提交
1152 1153
	}

1154
out:
1155 1156
	if (drbd_req_put_completion_ref(req, &m, 1))
		kref_put(&req->kref, drbd_req_destroy);
1157
	spin_unlock_irq(&mdev->tconn->req_lock);
P
Philipp Reisner 已提交
1158

1159 1160
	if (m.bio)
		complete_master_bio(mdev, &m);
1161 1162 1163 1164 1165 1166 1167 1168
}

void __drbd_make_request(struct drbd_conf *mdev, struct bio *bio, unsigned long start_time)
{
	struct drbd_request *req = drbd_request_prepare(mdev, bio, start_time);
	if (IS_ERR_OR_NULL(req))
		return;
	drbd_send_and_submit(mdev, req);
P
Philipp Reisner 已提交
1169 1170
}

1171
static void submit_fast_path(struct drbd_conf *mdev, struct list_head *incoming)
1172
{
1173 1174 1175
	struct drbd_request *req, *tmp;
	list_for_each_entry_safe(req, tmp, incoming, tl_requests) {
		const int rw = bio_data_dir(req->master_bio);
1176

1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187
		if (rw == WRITE /* rw != WRITE should not even end up here! */
		&& req->private_bio && req->i.size
		&& !test_bit(AL_SUSPENDED, &mdev->flags)) {
			if (!drbd_al_begin_io_fastpath(mdev, &req->i))
				continue;

			req->rq_state |= RQ_IN_ACT_LOG;
		}

		list_del_init(&req->tl_requests);
		drbd_send_and_submit(mdev, req);
1188 1189 1190
	}
}

1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214
static bool prepare_al_transaction_nonblock(struct drbd_conf *mdev,
					    struct list_head *incoming,
					    struct list_head *pending)
{
	struct drbd_request *req, *tmp;
	int wake = 0;
	int err;

	spin_lock_irq(&mdev->al_lock);
	list_for_each_entry_safe(req, tmp, incoming, tl_requests) {
		err = drbd_al_begin_io_nonblock(mdev, &req->i);
		if (err == -EBUSY)
			wake = 1;
		if (err)
			continue;
		req->rq_state |= RQ_IN_ACT_LOG;
		list_move_tail(&req->tl_requests, pending);
	}
	spin_unlock_irq(&mdev->al_lock);
	if (wake)
		wake_up(&mdev->al_wait);

	return !list_empty(pending);
}
1215 1216 1217 1218

void do_submit(struct work_struct *ws)
{
	struct drbd_conf *mdev = container_of(ws, struct drbd_conf, submit.worker);
1219 1220
	LIST_HEAD(incoming);
	LIST_HEAD(pending);
1221 1222
	struct drbd_request *req, *tmp;

1223 1224 1225 1226
	for (;;) {
		spin_lock(&mdev->submit.lock);
		list_splice_tail_init(&mdev->submit.writes, &incoming);
		spin_unlock(&mdev->submit.lock);
1227

1228 1229 1230 1231 1232
		submit_fast_path(mdev, &incoming);
		if (list_empty(&incoming))
			break;

		wait_event(mdev->al_wait, prepare_al_transaction_nonblock(mdev, &incoming, &pending));
1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263
		/* Maybe more was queued, while we prepared the transaction?
		 * Try to stuff them into this transaction as well.
		 * Be strictly non-blocking here, no wait_event, we already
		 * have something to commit.
		 * Stop if we don't make any more progres.
		 */
		for (;;) {
			LIST_HEAD(more_pending);
			LIST_HEAD(more_incoming);
			bool made_progress;

			/* It is ok to look outside the lock,
			 * it's only an optimization anyways */
			if (list_empty(&mdev->submit.writes))
				break;

			spin_lock(&mdev->submit.lock);
			list_splice_tail_init(&mdev->submit.writes, &more_incoming);
			spin_unlock(&mdev->submit.lock);

			if (list_empty(&more_incoming))
				break;

			made_progress = prepare_al_transaction_nonblock(mdev, &more_incoming, &more_pending);

			list_splice_tail_init(&more_pending, &pending);
			list_splice_tail_init(&more_incoming, &incoming);

			if (!made_progress)
				break;
		}
1264 1265 1266 1267 1268 1269
		drbd_al_begin_io_commit(mdev, false);

		list_for_each_entry_safe(req, tmp, &pending, tl_requests) {
			list_del_init(&req->tl_requests);
			drbd_send_and_submit(mdev, req);
		}
1270 1271 1272
	}
}

1273
void drbd_make_request(struct request_queue *q, struct bio *bio)
P
Philipp Reisner 已提交
1274 1275
{
	struct drbd_conf *mdev = (struct drbd_conf *) q->queuedata;
1276
	unsigned long start_time;
P
Philipp Reisner 已提交
1277

1278 1279
	start_time = jiffies;

P
Philipp Reisner 已提交
1280 1281 1282
	/*
	 * what we "blindly" assume:
	 */
1283
	D_ASSERT(IS_ALIGNED(bio->bi_size, 512));
P
Philipp Reisner 已提交
1284

1285 1286
	inc_ap_bio(mdev);
	__drbd_make_request(mdev, bio, start_time);
P
Philipp Reisner 已提交
1287 1288
}

1289 1290 1291
/* This is called by bio_add_page().
 *
 * q->max_hw_sectors and other global limits are already enforced there.
P
Philipp Reisner 已提交
1292
 *
1293 1294 1295 1296
 * We need to call down to our lower level device,
 * in case it has special restrictions.
 *
 * We also may need to enforce configured max-bio-bvecs limits.
P
Philipp Reisner 已提交
1297 1298
 *
 * As long as the BIO is empty we have to allow at least one bvec,
1299
 * regardless of size and offset, so no need to ask lower levels.
P
Philipp Reisner 已提交
1300 1301 1302 1303 1304
 */
int drbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bvm, struct bio_vec *bvec)
{
	struct drbd_conf *mdev = (struct drbd_conf *) q->queuedata;
	unsigned int bio_size = bvm->bi_size;
1305 1306 1307 1308
	int limit = DRBD_MAX_BIO_SIZE;
	int backing_limit;

	if (bio_size && get_ldev(mdev)) {
P
Philipp Reisner 已提交
1309 1310
		struct request_queue * const b =
			mdev->ldev->backing_bdev->bd_disk->queue;
1311
		if (b->merge_bvec_fn) {
P
Philipp Reisner 已提交
1312 1313 1314 1315 1316 1317 1318
			backing_limit = b->merge_bvec_fn(b, bvm, bvec);
			limit = min(limit, backing_limit);
		}
		put_ldev(mdev);
	}
	return limit;
}
1319

1320 1321 1322 1323 1324 1325
struct drbd_request *find_oldest_request(struct drbd_tconn *tconn)
{
	/* Walk the transfer log,
	 * and find the oldest not yet completed request */
	struct drbd_request *r;
	list_for_each_entry(r, &tconn->transfer_log, tl_requests) {
1326
		if (atomic_read(&r->completion_ref))
1327 1328 1329 1330 1331
			return r;
	}
	return NULL;
}

1332 1333 1334
void request_timer_fn(unsigned long data)
{
	struct drbd_conf *mdev = (struct drbd_conf *) data;
1335
	struct drbd_tconn *tconn = mdev->tconn;
1336
	struct drbd_request *req; /* oldest request */
1337
	struct net_conf *nc;
1338
	unsigned long ent = 0, dt = 0, et, nt; /* effective timeout = ko_count * timeout */
1339
	unsigned long now;
1340

1341 1342
	rcu_read_lock();
	nc = rcu_dereference(tconn->net_conf);
1343 1344
	if (nc && mdev->state.conn >= C_WF_REPORT_PARAMS)
		ent = nc->timeout * HZ/10 * nc->ko_count;
1345

1346
	if (get_ldev(mdev)) { /* implicit state.disk >= D_INCONSISTENT */
1347
		dt = rcu_dereference(mdev->ldev->disk_conf)->disk_timeout * HZ / 10;
1348 1349
		put_ldev(mdev);
	}
1350
	rcu_read_unlock();
1351

1352 1353
	et = min_not_zero(dt, ent);

1354
	if (!et)
1355 1356
		return; /* Recurring timer stopped */

1357 1358
	now = jiffies;

1359
	spin_lock_irq(&tconn->req_lock);
1360 1361
	req = find_oldest_request(tconn);
	if (!req) {
1362
		spin_unlock_irq(&tconn->req_lock);
1363
		mod_timer(&mdev->request_timer, now + et);
1364 1365 1366
		return;
	}

1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384
	/* The request is considered timed out, if
	 * - we have some effective timeout from the configuration,
	 *   with above state restrictions applied,
	 * - the oldest request is waiting for a response from the network
	 *   resp. the local disk,
	 * - the oldest request is in fact older than the effective timeout,
	 * - the connection was established (resp. disk was attached)
	 *   for longer than the timeout already.
	 * Note that for 32bit jiffies and very stable connections/disks,
	 * we may have a wrap around, which is catched by
	 *   !time_in_range(now, last_..._jif, last_..._jif + timeout).
	 *
	 * Side effect: once per 32bit wrap-around interval, which means every
	 * ~198 days with 250 HZ, we have a window where the timeout would need
	 * to expire twice (worst case) to become effective. Good enough.
	 */
	if (ent && req->rq_state & RQ_NET_PENDING &&
		 time_after(now, req->start_time + ent) &&
1385
		!time_in_range(now, tconn->last_reconnect_jif, tconn->last_reconnect_jif + ent)) {
1386 1387
		dev_warn(DEV, "Remote failed to finish a request within ko-count * timeout\n");
		_drbd_set_state(_NS(mdev, conn, C_TIMEOUT), CS_VERBOSE | CS_HARD, NULL);
1388
	}
1389
	if (dt && req->rq_state & RQ_LOCAL_PENDING && req->w.mdev == mdev &&
1390 1391 1392
		 time_after(now, req->start_time + dt) &&
		!time_in_range(now, mdev->last_reattach_jif, mdev->last_reattach_jif + dt)) {
		dev_warn(DEV, "Local backing device failed to meet the disk-timeout\n");
1393
		__drbd_chk_io_error(mdev, DRBD_FORCE_DETACH);
1394
	}
1395
	nt = (time_after(now, req->start_time + et) ? now : req->start_time) + et;
1396
	spin_unlock_irq(&tconn->req_lock);
1397
	mod_timer(&mdev->request_timer, nt);
1398
}