xref: /openbmc/linux/drivers/block/drbd/drbd_req.c (revision e8f6f3b4)
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 static bool drbd_may_do_local_read(struct drbd_device *device, sector_t sector, int size);
35 
36 /* Update disk stats at start of I/O request */
37 static void _drbd_start_io_acct(struct drbd_device *device, struct drbd_request *req)
38 {
39 	generic_start_io_acct(bio_data_dir(req->master_bio), req->i.size >> 9,
40 			      &device->vdisk->part0);
41 }
42 
43 /* Update disk stats when completing request upwards */
44 static void _drbd_end_io_acct(struct drbd_device *device, struct drbd_request *req)
45 {
46 	generic_end_io_acct(bio_data_dir(req->master_bio),
47 			    &device->vdisk->part0, req->start_jif);
48 }
49 
50 static struct drbd_request *drbd_req_new(struct drbd_device *device,
51 					       struct bio *bio_src)
52 {
53 	struct drbd_request *req;
54 
55 	req = mempool_alloc(drbd_request_mempool, GFP_NOIO | __GFP_ZERO);
56 	if (!req)
57 		return NULL;
58 
59 	drbd_req_make_private_bio(req, bio_src);
60 	req->rq_state    = bio_data_dir(bio_src) == WRITE ? RQ_WRITE : 0;
61 	req->device   = device;
62 	req->master_bio  = bio_src;
63 	req->epoch       = 0;
64 
65 	drbd_clear_interval(&req->i);
66 	req->i.sector     = bio_src->bi_iter.bi_sector;
67 	req->i.size      = bio_src->bi_iter.bi_size;
68 	req->i.local = true;
69 	req->i.waiting = false;
70 
71 	INIT_LIST_HEAD(&req->tl_requests);
72 	INIT_LIST_HEAD(&req->w.list);
73 	INIT_LIST_HEAD(&req->req_pending_master_completion);
74 	INIT_LIST_HEAD(&req->req_pending_local);
75 
76 	/* one reference to be put by __drbd_make_request */
77 	atomic_set(&req->completion_ref, 1);
78 	/* one kref as long as completion_ref > 0 */
79 	kref_init(&req->kref);
80 	return req;
81 }
82 
83 static void drbd_remove_request_interval(struct rb_root *root,
84 					 struct drbd_request *req)
85 {
86 	struct drbd_device *device = req->device;
87 	struct drbd_interval *i = &req->i;
88 
89 	drbd_remove_interval(root, i);
90 
91 	/* Wake up any processes waiting for this request to complete.  */
92 	if (i->waiting)
93 		wake_up(&device->misc_wait);
94 }
95 
96 void drbd_req_destroy(struct kref *kref)
97 {
98 	struct drbd_request *req = container_of(kref, struct drbd_request, kref);
99 	struct drbd_device *device = req->device;
100 	const unsigned s = req->rq_state;
101 
102 	if ((req->master_bio && !(s & RQ_POSTPONED)) ||
103 		atomic_read(&req->completion_ref) ||
104 		(s & RQ_LOCAL_PENDING) ||
105 		((s & RQ_NET_MASK) && !(s & RQ_NET_DONE))) {
106 		drbd_err(device, "drbd_req_destroy: Logic BUG rq_state = 0x%x, completion_ref = %d\n",
107 				s, atomic_read(&req->completion_ref));
108 		return;
109 	}
110 
111 	/* If called from mod_rq_state (expected normal case) or
112 	 * drbd_send_and_submit (the less likely normal path), this holds the
113 	 * req_lock, and req->tl_requests will typicaly be on ->transfer_log,
114 	 * though it may be still empty (never added to the transfer log).
115 	 *
116 	 * If called from do_retry(), we do NOT hold the req_lock, but we are
117 	 * still allowed to unconditionally list_del(&req->tl_requests),
118 	 * because it will be on a local on-stack list only. */
119 	list_del_init(&req->tl_requests);
120 
121 	/* finally remove the request from the conflict detection
122 	 * respective block_id verification interval tree. */
123 	if (!drbd_interval_empty(&req->i)) {
124 		struct rb_root *root;
125 
126 		if (s & RQ_WRITE)
127 			root = &device->write_requests;
128 		else
129 			root = &device->read_requests;
130 		drbd_remove_request_interval(root, req);
131 	} else if (s & (RQ_NET_MASK & ~RQ_NET_DONE) && req->i.size != 0)
132 		drbd_err(device, "drbd_req_destroy: Logic BUG: interval empty, but: rq_state=0x%x, sect=%llu, size=%u\n",
133 			s, (unsigned long long)req->i.sector, req->i.size);
134 
135 	/* if it was a write, we may have to set the corresponding
136 	 * bit(s) out-of-sync first. If it had a local part, we need to
137 	 * release the reference to the activity log. */
138 	if (s & RQ_WRITE) {
139 		/* Set out-of-sync unless both OK flags are set
140 		 * (local only or remote failed).
141 		 * Other places where we set out-of-sync:
142 		 * READ with local io-error */
143 
144 		/* There is a special case:
145 		 * we may notice late that IO was suspended,
146 		 * and postpone, or schedule for retry, a write,
147 		 * before it even was submitted or sent.
148 		 * In that case we do not want to touch the bitmap at all.
149 		 */
150 		if ((s & (RQ_POSTPONED|RQ_LOCAL_MASK|RQ_NET_MASK)) != RQ_POSTPONED) {
151 			if (!(s & RQ_NET_OK) || !(s & RQ_LOCAL_OK))
152 				drbd_set_out_of_sync(device, req->i.sector, req->i.size);
153 
154 			if ((s & RQ_NET_OK) && (s & RQ_LOCAL_OK) && (s & RQ_NET_SIS))
155 				drbd_set_in_sync(device, req->i.sector, req->i.size);
156 		}
157 
158 		/* one might be tempted to move the drbd_al_complete_io
159 		 * to the local io completion callback drbd_request_endio.
160 		 * but, if this was a mirror write, we may only
161 		 * drbd_al_complete_io after this is RQ_NET_DONE,
162 		 * otherwise the extent could be dropped from the al
163 		 * before it has actually been written on the peer.
164 		 * if we crash before our peer knows about the request,
165 		 * but after the extent has been dropped from the al,
166 		 * we would forget to resync the corresponding extent.
167 		 */
168 		if (s & RQ_IN_ACT_LOG) {
169 			if (get_ldev_if_state(device, D_FAILED)) {
170 				drbd_al_complete_io(device, &req->i);
171 				put_ldev(device);
172 			} else if (__ratelimit(&drbd_ratelimit_state)) {
173 				drbd_warn(device, "Should have called drbd_al_complete_io(, %llu, %u), "
174 					 "but my Disk seems to have failed :(\n",
175 					 (unsigned long long) req->i.sector, req->i.size);
176 			}
177 		}
178 	}
179 
180 	mempool_free(req, drbd_request_mempool);
181 }
182 
183 static void wake_all_senders(struct drbd_connection *connection)
184 {
185 	wake_up(&connection->sender_work.q_wait);
186 }
187 
188 /* must hold resource->req_lock */
189 void start_new_tl_epoch(struct drbd_connection *connection)
190 {
191 	/* no point closing an epoch, if it is empty, anyways. */
192 	if (connection->current_tle_writes == 0)
193 		return;
194 
195 	connection->current_tle_writes = 0;
196 	atomic_inc(&connection->current_tle_nr);
197 	wake_all_senders(connection);
198 }
199 
200 void complete_master_bio(struct drbd_device *device,
201 		struct bio_and_error *m)
202 {
203 	bio_endio(m->bio, m->error);
204 	dec_ap_bio(device);
205 }
206 
207 
208 /* Helper for __req_mod().
209  * Set m->bio to the master bio, if it is fit to be completed,
210  * or leave it alone (it is initialized to NULL in __req_mod),
211  * if it has already been completed, or cannot be completed yet.
212  * If m->bio is set, the error status to be returned is placed in m->error.
213  */
214 static
215 void drbd_req_complete(struct drbd_request *req, struct bio_and_error *m)
216 {
217 	const unsigned s = req->rq_state;
218 	struct drbd_device *device = req->device;
219 	int rw;
220 	int error, ok;
221 
222 	/* we must not complete the master bio, while it is
223 	 *	still being processed by _drbd_send_zc_bio (drbd_send_dblock)
224 	 *	not yet acknowledged by the peer
225 	 *	not yet completed by the local io subsystem
226 	 * these flags may get cleared in any order by
227 	 *	the worker,
228 	 *	the receiver,
229 	 *	the bio_endio completion callbacks.
230 	 */
231 	if ((s & RQ_LOCAL_PENDING && !(s & RQ_LOCAL_ABORTED)) ||
232 	    (s & RQ_NET_QUEUED) || (s & RQ_NET_PENDING) ||
233 	    (s & RQ_COMPLETION_SUSP)) {
234 		drbd_err(device, "drbd_req_complete: Logic BUG rq_state = 0x%x\n", s);
235 		return;
236 	}
237 
238 	if (!req->master_bio) {
239 		drbd_err(device, "drbd_req_complete: Logic BUG, master_bio == NULL!\n");
240 		return;
241 	}
242 
243 	rw = bio_rw(req->master_bio);
244 
245 	/*
246 	 * figure out whether to report success or failure.
247 	 *
248 	 * report success when at least one of the operations succeeded.
249 	 * or, to put the other way,
250 	 * only report failure, when both operations failed.
251 	 *
252 	 * what to do about the failures is handled elsewhere.
253 	 * what we need to do here is just: complete the master_bio.
254 	 *
255 	 * local completion error, if any, has been stored as ERR_PTR
256 	 * in private_bio within drbd_request_endio.
257 	 */
258 	ok = (s & RQ_LOCAL_OK) || (s & RQ_NET_OK);
259 	error = PTR_ERR(req->private_bio);
260 
261 	/* Before we can signal completion to the upper layers,
262 	 * we may need to close the current transfer log epoch.
263 	 * We are within the request lock, so we can simply compare
264 	 * the request epoch number with the current transfer log
265 	 * epoch number.  If they match, increase the current_tle_nr,
266 	 * and reset the transfer log epoch write_cnt.
267 	 */
268 	if (rw == WRITE &&
269 	    req->epoch == atomic_read(&first_peer_device(device)->connection->current_tle_nr))
270 		start_new_tl_epoch(first_peer_device(device)->connection);
271 
272 	/* Update disk stats */
273 	_drbd_end_io_acct(device, req);
274 
275 	/* If READ failed,
276 	 * have it be pushed back to the retry work queue,
277 	 * so it will re-enter __drbd_make_request(),
278 	 * and be re-assigned to a suitable local or remote path,
279 	 * or failed if we do not have access to good data anymore.
280 	 *
281 	 * Unless it was failed early by __drbd_make_request(),
282 	 * because no path was available, in which case
283 	 * it was not even added to the transfer_log.
284 	 *
285 	 * READA may fail, and will not be retried.
286 	 *
287 	 * WRITE should have used all available paths already.
288 	 */
289 	if (!ok && rw == READ && !list_empty(&req->tl_requests))
290 		req->rq_state |= RQ_POSTPONED;
291 
292 	if (!(req->rq_state & RQ_POSTPONED)) {
293 		m->error = ok ? 0 : (error ?: -EIO);
294 		m->bio = req->master_bio;
295 		req->master_bio = NULL;
296 		/* We leave it in the tree, to be able to verify later
297 		 * write-acks in protocol != C during resync.
298 		 * But we mark it as "complete", so it won't be counted as
299 		 * conflict in a multi-primary setup. */
300 		req->i.completed = true;
301 	}
302 
303 	if (req->i.waiting)
304 		wake_up(&device->misc_wait);
305 
306 	/* Either we are about to complete to upper layers,
307 	 * or we will restart this request.
308 	 * In either case, the request object will be destroyed soon,
309 	 * so better remove it from all lists. */
310 	list_del_init(&req->req_pending_master_completion);
311 }
312 
313 /* still holds resource->req_lock */
314 static int drbd_req_put_completion_ref(struct drbd_request *req, struct bio_and_error *m, int put)
315 {
316 	struct drbd_device *device = req->device;
317 	D_ASSERT(device, m || (req->rq_state & RQ_POSTPONED));
318 
319 	if (!atomic_sub_and_test(put, &req->completion_ref))
320 		return 0;
321 
322 	drbd_req_complete(req, m);
323 
324 	if (req->rq_state & RQ_POSTPONED) {
325 		/* don't destroy the req object just yet,
326 		 * but queue it for retry */
327 		drbd_restart_request(req);
328 		return 0;
329 	}
330 
331 	return 1;
332 }
333 
334 static void set_if_null_req_next(struct drbd_peer_device *peer_device, struct drbd_request *req)
335 {
336 	struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
337 	if (!connection)
338 		return;
339 	if (connection->req_next == NULL)
340 		connection->req_next = req;
341 }
342 
343 static void advance_conn_req_next(struct drbd_peer_device *peer_device, struct drbd_request *req)
344 {
345 	struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
346 	if (!connection)
347 		return;
348 	if (connection->req_next != req)
349 		return;
350 	list_for_each_entry_continue(req, &connection->transfer_log, tl_requests) {
351 		const unsigned s = req->rq_state;
352 		if (s & RQ_NET_QUEUED)
353 			break;
354 	}
355 	if (&req->tl_requests == &connection->transfer_log)
356 		req = NULL;
357 	connection->req_next = req;
358 }
359 
360 static void set_if_null_req_ack_pending(struct drbd_peer_device *peer_device, struct drbd_request *req)
361 {
362 	struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
363 	if (!connection)
364 		return;
365 	if (connection->req_ack_pending == NULL)
366 		connection->req_ack_pending = req;
367 }
368 
369 static void advance_conn_req_ack_pending(struct drbd_peer_device *peer_device, struct drbd_request *req)
370 {
371 	struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
372 	if (!connection)
373 		return;
374 	if (connection->req_ack_pending != req)
375 		return;
376 	list_for_each_entry_continue(req, &connection->transfer_log, tl_requests) {
377 		const unsigned s = req->rq_state;
378 		if ((s & RQ_NET_SENT) && (s & RQ_NET_PENDING))
379 			break;
380 	}
381 	if (&req->tl_requests == &connection->transfer_log)
382 		req = NULL;
383 	connection->req_ack_pending = req;
384 }
385 
386 static void set_if_null_req_not_net_done(struct drbd_peer_device *peer_device, struct drbd_request *req)
387 {
388 	struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
389 	if (!connection)
390 		return;
391 	if (connection->req_not_net_done == NULL)
392 		connection->req_not_net_done = req;
393 }
394 
395 static void advance_conn_req_not_net_done(struct drbd_peer_device *peer_device, struct drbd_request *req)
396 {
397 	struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
398 	if (!connection)
399 		return;
400 	if (connection->req_not_net_done != req)
401 		return;
402 	list_for_each_entry_continue(req, &connection->transfer_log, tl_requests) {
403 		const unsigned s = req->rq_state;
404 		if ((s & RQ_NET_SENT) && !(s & RQ_NET_DONE))
405 			break;
406 	}
407 	if (&req->tl_requests == &connection->transfer_log)
408 		req = NULL;
409 	connection->req_not_net_done = req;
410 }
411 
412 /* I'd like this to be the only place that manipulates
413  * req->completion_ref and req->kref. */
414 static void mod_rq_state(struct drbd_request *req, struct bio_and_error *m,
415 		int clear, int set)
416 {
417 	struct drbd_device *device = req->device;
418 	struct drbd_peer_device *peer_device = first_peer_device(device);
419 	unsigned s = req->rq_state;
420 	int c_put = 0;
421 	int k_put = 0;
422 
423 	if (drbd_suspended(device) && !((s | clear) & RQ_COMPLETION_SUSP))
424 		set |= RQ_COMPLETION_SUSP;
425 
426 	/* apply */
427 
428 	req->rq_state &= ~clear;
429 	req->rq_state |= set;
430 
431 	/* no change? */
432 	if (req->rq_state == s)
433 		return;
434 
435 	/* intent: get references */
436 
437 	if (!(s & RQ_LOCAL_PENDING) && (set & RQ_LOCAL_PENDING))
438 		atomic_inc(&req->completion_ref);
439 
440 	if (!(s & RQ_NET_PENDING) && (set & RQ_NET_PENDING)) {
441 		inc_ap_pending(device);
442 		atomic_inc(&req->completion_ref);
443 	}
444 
445 	if (!(s & RQ_NET_QUEUED) && (set & RQ_NET_QUEUED)) {
446 		atomic_inc(&req->completion_ref);
447 		set_if_null_req_next(peer_device, req);
448 	}
449 
450 	if (!(s & RQ_EXP_BARR_ACK) && (set & RQ_EXP_BARR_ACK))
451 		kref_get(&req->kref); /* wait for the DONE */
452 
453 	if (!(s & RQ_NET_SENT) && (set & RQ_NET_SENT)) {
454 		/* potentially already completed in the asender thread */
455 		if (!(s & RQ_NET_DONE)) {
456 			atomic_add(req->i.size >> 9, &device->ap_in_flight);
457 			set_if_null_req_not_net_done(peer_device, req);
458 		}
459 		if (s & RQ_NET_PENDING)
460 			set_if_null_req_ack_pending(peer_device, req);
461 	}
462 
463 	if (!(s & RQ_COMPLETION_SUSP) && (set & RQ_COMPLETION_SUSP))
464 		atomic_inc(&req->completion_ref);
465 
466 	/* progress: put references */
467 
468 	if ((s & RQ_COMPLETION_SUSP) && (clear & RQ_COMPLETION_SUSP))
469 		++c_put;
470 
471 	if (!(s & RQ_LOCAL_ABORTED) && (set & RQ_LOCAL_ABORTED)) {
472 		D_ASSERT(device, req->rq_state & RQ_LOCAL_PENDING);
473 		/* local completion may still come in later,
474 		 * we need to keep the req object around. */
475 		kref_get(&req->kref);
476 		++c_put;
477 	}
478 
479 	if ((s & RQ_LOCAL_PENDING) && (clear & RQ_LOCAL_PENDING)) {
480 		if (req->rq_state & RQ_LOCAL_ABORTED)
481 			++k_put;
482 		else
483 			++c_put;
484 		list_del_init(&req->req_pending_local);
485 	}
486 
487 	if ((s & RQ_NET_PENDING) && (clear & RQ_NET_PENDING)) {
488 		dec_ap_pending(device);
489 		++c_put;
490 		req->acked_jif = jiffies;
491 		advance_conn_req_ack_pending(peer_device, req);
492 	}
493 
494 	if ((s & RQ_NET_QUEUED) && (clear & RQ_NET_QUEUED)) {
495 		++c_put;
496 		advance_conn_req_next(peer_device, req);
497 	}
498 
499 	if (!(s & RQ_NET_DONE) && (set & RQ_NET_DONE)) {
500 		if (s & RQ_NET_SENT)
501 			atomic_sub(req->i.size >> 9, &device->ap_in_flight);
502 		if (s & RQ_EXP_BARR_ACK)
503 			++k_put;
504 		req->net_done_jif = jiffies;
505 
506 		/* in ahead/behind mode, or just in case,
507 		 * before we finally destroy this request,
508 		 * the caching pointers must not reference it anymore */
509 		advance_conn_req_next(peer_device, req);
510 		advance_conn_req_ack_pending(peer_device, req);
511 		advance_conn_req_not_net_done(peer_device, req);
512 	}
513 
514 	/* potentially complete and destroy */
515 
516 	if (k_put || c_put) {
517 		/* Completion does it's own kref_put.  If we are going to
518 		 * kref_sub below, we need req to be still around then. */
519 		int at_least = k_put + !!c_put;
520 		int refcount = atomic_read(&req->kref.refcount);
521 		if (refcount < at_least)
522 			drbd_err(device,
523 				"mod_rq_state: Logic BUG: %x -> %x: refcount = %d, should be >= %d\n",
524 				s, req->rq_state, refcount, at_least);
525 	}
526 
527 	/* If we made progress, retry conflicting peer requests, if any. */
528 	if (req->i.waiting)
529 		wake_up(&device->misc_wait);
530 
531 	if (c_put)
532 		k_put += drbd_req_put_completion_ref(req, m, c_put);
533 	if (k_put)
534 		kref_sub(&req->kref, k_put, drbd_req_destroy);
535 }
536 
537 static void drbd_report_io_error(struct drbd_device *device, struct drbd_request *req)
538 {
539         char b[BDEVNAME_SIZE];
540 
541 	if (!__ratelimit(&drbd_ratelimit_state))
542 		return;
543 
544 	drbd_warn(device, "local %s IO error sector %llu+%u on %s\n",
545 			(req->rq_state & RQ_WRITE) ? "WRITE" : "READ",
546 			(unsigned long long)req->i.sector,
547 			req->i.size >> 9,
548 			bdevname(device->ldev->backing_bdev, b));
549 }
550 
551 /* Helper for HANDED_OVER_TO_NETWORK.
552  * Is this a protocol A write (neither WRITE_ACK nor RECEIVE_ACK expected)?
553  * Is it also still "PENDING"?
554  * --> If so, clear PENDING and set NET_OK below.
555  * If it is a protocol A write, but not RQ_PENDING anymore, neg-ack was faster
556  * (and we must not set RQ_NET_OK) */
557 static inline bool is_pending_write_protocol_A(struct drbd_request *req)
558 {
559 	return (req->rq_state &
560 		   (RQ_WRITE|RQ_NET_PENDING|RQ_EXP_WRITE_ACK|RQ_EXP_RECEIVE_ACK))
561 		== (RQ_WRITE|RQ_NET_PENDING);
562 }
563 
564 /* obviously this could be coded as many single functions
565  * instead of one huge switch,
566  * or by putting the code directly in the respective locations
567  * (as it has been before).
568  *
569  * but having it this way
570  *  enforces that it is all in this one place, where it is easier to audit,
571  *  it makes it obvious that whatever "event" "happens" to a request should
572  *  happen "atomically" within the req_lock,
573  *  and it enforces that we have to think in a very structured manner
574  *  about the "events" that may happen to a request during its life time ...
575  */
576 int __req_mod(struct drbd_request *req, enum drbd_req_event what,
577 		struct bio_and_error *m)
578 {
579 	struct drbd_device *const device = req->device;
580 	struct drbd_peer_device *const peer_device = first_peer_device(device);
581 	struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
582 	struct net_conf *nc;
583 	int p, rv = 0;
584 
585 	if (m)
586 		m->bio = NULL;
587 
588 	switch (what) {
589 	default:
590 		drbd_err(device, "LOGIC BUG in %s:%u\n", __FILE__ , __LINE__);
591 		break;
592 
593 	/* does not happen...
594 	 * initialization done in drbd_req_new
595 	case CREATED:
596 		break;
597 		*/
598 
599 	case TO_BE_SENT: /* via network */
600 		/* reached via __drbd_make_request
601 		 * and from w_read_retry_remote */
602 		D_ASSERT(device, !(req->rq_state & RQ_NET_MASK));
603 		rcu_read_lock();
604 		nc = rcu_dereference(connection->net_conf);
605 		p = nc->wire_protocol;
606 		rcu_read_unlock();
607 		req->rq_state |=
608 			p == DRBD_PROT_C ? RQ_EXP_WRITE_ACK :
609 			p == DRBD_PROT_B ? RQ_EXP_RECEIVE_ACK : 0;
610 		mod_rq_state(req, m, 0, RQ_NET_PENDING);
611 		break;
612 
613 	case TO_BE_SUBMITTED: /* locally */
614 		/* reached via __drbd_make_request */
615 		D_ASSERT(device, !(req->rq_state & RQ_LOCAL_MASK));
616 		mod_rq_state(req, m, 0, RQ_LOCAL_PENDING);
617 		break;
618 
619 	case COMPLETED_OK:
620 		if (req->rq_state & RQ_WRITE)
621 			device->writ_cnt += req->i.size >> 9;
622 		else
623 			device->read_cnt += req->i.size >> 9;
624 
625 		mod_rq_state(req, m, RQ_LOCAL_PENDING,
626 				RQ_LOCAL_COMPLETED|RQ_LOCAL_OK);
627 		break;
628 
629 	case ABORT_DISK_IO:
630 		mod_rq_state(req, m, 0, RQ_LOCAL_ABORTED);
631 		break;
632 
633 	case WRITE_COMPLETED_WITH_ERROR:
634 		drbd_report_io_error(device, req);
635 		__drbd_chk_io_error(device, DRBD_WRITE_ERROR);
636 		mod_rq_state(req, m, RQ_LOCAL_PENDING, RQ_LOCAL_COMPLETED);
637 		break;
638 
639 	case READ_COMPLETED_WITH_ERROR:
640 		drbd_set_out_of_sync(device, req->i.sector, req->i.size);
641 		drbd_report_io_error(device, req);
642 		__drbd_chk_io_error(device, DRBD_READ_ERROR);
643 		/* fall through. */
644 	case READ_AHEAD_COMPLETED_WITH_ERROR:
645 		/* it is legal to fail READA, no __drbd_chk_io_error in that case. */
646 		mod_rq_state(req, m, RQ_LOCAL_PENDING, RQ_LOCAL_COMPLETED);
647 		break;
648 
649 	case DISCARD_COMPLETED_NOTSUPP:
650 	case DISCARD_COMPLETED_WITH_ERROR:
651 		/* I'd rather not detach from local disk just because it
652 		 * failed a REQ_DISCARD. */
653 		mod_rq_state(req, m, RQ_LOCAL_PENDING, RQ_LOCAL_COMPLETED);
654 		break;
655 
656 	case QUEUE_FOR_NET_READ:
657 		/* READ or READA, and
658 		 * no local disk,
659 		 * or target area marked as invalid,
660 		 * or just got an io-error. */
661 		/* from __drbd_make_request
662 		 * or from bio_endio during read io-error recovery */
663 
664 		/* So we can verify the handle in the answer packet.
665 		 * Corresponding drbd_remove_request_interval is in
666 		 * drbd_req_complete() */
667 		D_ASSERT(device, drbd_interval_empty(&req->i));
668 		drbd_insert_interval(&device->read_requests, &req->i);
669 
670 		set_bit(UNPLUG_REMOTE, &device->flags);
671 
672 		D_ASSERT(device, req->rq_state & RQ_NET_PENDING);
673 		D_ASSERT(device, (req->rq_state & RQ_LOCAL_MASK) == 0);
674 		mod_rq_state(req, m, 0, RQ_NET_QUEUED);
675 		req->w.cb = w_send_read_req;
676 		drbd_queue_work(&connection->sender_work,
677 				&req->w);
678 		break;
679 
680 	case QUEUE_FOR_NET_WRITE:
681 		/* assert something? */
682 		/* from __drbd_make_request only */
683 
684 		/* Corresponding drbd_remove_request_interval is in
685 		 * drbd_req_complete() */
686 		D_ASSERT(device, drbd_interval_empty(&req->i));
687 		drbd_insert_interval(&device->write_requests, &req->i);
688 
689 		/* NOTE
690 		 * In case the req ended up on the transfer log before being
691 		 * queued on the worker, it could lead to this request being
692 		 * missed during cleanup after connection loss.
693 		 * So we have to do both operations here,
694 		 * within the same lock that protects the transfer log.
695 		 *
696 		 * _req_add_to_epoch(req); this has to be after the
697 		 * _maybe_start_new_epoch(req); which happened in
698 		 * __drbd_make_request, because we now may set the bit
699 		 * again ourselves to close the current epoch.
700 		 *
701 		 * Add req to the (now) current epoch (barrier). */
702 
703 		/* otherwise we may lose an unplug, which may cause some remote
704 		 * io-scheduler timeout to expire, increasing maximum latency,
705 		 * hurting performance. */
706 		set_bit(UNPLUG_REMOTE, &device->flags);
707 
708 		/* queue work item to send data */
709 		D_ASSERT(device, req->rq_state & RQ_NET_PENDING);
710 		mod_rq_state(req, m, 0, RQ_NET_QUEUED|RQ_EXP_BARR_ACK);
711 		req->w.cb =  w_send_dblock;
712 		drbd_queue_work(&connection->sender_work,
713 				&req->w);
714 
715 		/* close the epoch, in case it outgrew the limit */
716 		rcu_read_lock();
717 		nc = rcu_dereference(connection->net_conf);
718 		p = nc->max_epoch_size;
719 		rcu_read_unlock();
720 		if (connection->current_tle_writes >= p)
721 			start_new_tl_epoch(connection);
722 
723 		break;
724 
725 	case QUEUE_FOR_SEND_OOS:
726 		mod_rq_state(req, m, 0, RQ_NET_QUEUED);
727 		req->w.cb =  w_send_out_of_sync;
728 		drbd_queue_work(&connection->sender_work,
729 				&req->w);
730 		break;
731 
732 	case READ_RETRY_REMOTE_CANCELED:
733 	case SEND_CANCELED:
734 	case SEND_FAILED:
735 		/* real cleanup will be done from tl_clear.  just update flags
736 		 * so it is no longer marked as on the worker queue */
737 		mod_rq_state(req, m, RQ_NET_QUEUED, 0);
738 		break;
739 
740 	case HANDED_OVER_TO_NETWORK:
741 		/* assert something? */
742 		if (is_pending_write_protocol_A(req))
743 			/* this is what is dangerous about protocol A:
744 			 * pretend it was successfully written on the peer. */
745 			mod_rq_state(req, m, RQ_NET_QUEUED|RQ_NET_PENDING,
746 						RQ_NET_SENT|RQ_NET_OK);
747 		else
748 			mod_rq_state(req, m, RQ_NET_QUEUED, RQ_NET_SENT);
749 		/* It is still not yet RQ_NET_DONE until the
750 		 * corresponding epoch barrier got acked as well,
751 		 * so we know what to dirty on connection loss. */
752 		break;
753 
754 	case OOS_HANDED_TO_NETWORK:
755 		/* Was not set PENDING, no longer QUEUED, so is now DONE
756 		 * as far as this connection is concerned. */
757 		mod_rq_state(req, m, RQ_NET_QUEUED, RQ_NET_DONE);
758 		break;
759 
760 	case CONNECTION_LOST_WHILE_PENDING:
761 		/* transfer log cleanup after connection loss */
762 		mod_rq_state(req, m,
763 				RQ_NET_OK|RQ_NET_PENDING|RQ_COMPLETION_SUSP,
764 				RQ_NET_DONE);
765 		break;
766 
767 	case CONFLICT_RESOLVED:
768 		/* for superseded conflicting writes of multiple primaries,
769 		 * there is no need to keep anything in the tl, potential
770 		 * node crashes are covered by the activity log.
771 		 *
772 		 * If this request had been marked as RQ_POSTPONED before,
773 		 * it will actually not be completed, but "restarted",
774 		 * resubmitted from the retry worker context. */
775 		D_ASSERT(device, req->rq_state & RQ_NET_PENDING);
776 		D_ASSERT(device, req->rq_state & RQ_EXP_WRITE_ACK);
777 		mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_DONE|RQ_NET_OK);
778 		break;
779 
780 	case WRITE_ACKED_BY_PEER_AND_SIS:
781 		req->rq_state |= RQ_NET_SIS;
782 	case WRITE_ACKED_BY_PEER:
783 		/* Normal operation protocol C: successfully written on peer.
784 		 * During resync, even in protocol != C,
785 		 * we requested an explicit write ack anyways.
786 		 * Which means we cannot even assert anything here.
787 		 * Nothing more to do here.
788 		 * We want to keep the tl in place for all protocols, to cater
789 		 * for volatile write-back caches on lower level devices. */
790 		goto ack_common;
791 	case RECV_ACKED_BY_PEER:
792 		D_ASSERT(device, req->rq_state & RQ_EXP_RECEIVE_ACK);
793 		/* protocol B; pretends to be successfully written on peer.
794 		 * see also notes above in HANDED_OVER_TO_NETWORK about
795 		 * protocol != C */
796 	ack_common:
797 		mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_OK);
798 		break;
799 
800 	case POSTPONE_WRITE:
801 		D_ASSERT(device, req->rq_state & RQ_EXP_WRITE_ACK);
802 		/* If this node has already detected the write conflict, the
803 		 * worker will be waiting on misc_wait.  Wake it up once this
804 		 * request has completed locally.
805 		 */
806 		D_ASSERT(device, req->rq_state & RQ_NET_PENDING);
807 		req->rq_state |= RQ_POSTPONED;
808 		if (req->i.waiting)
809 			wake_up(&device->misc_wait);
810 		/* Do not clear RQ_NET_PENDING. This request will make further
811 		 * progress via restart_conflicting_writes() or
812 		 * fail_postponed_requests(). Hopefully. */
813 		break;
814 
815 	case NEG_ACKED:
816 		mod_rq_state(req, m, RQ_NET_OK|RQ_NET_PENDING, 0);
817 		break;
818 
819 	case FAIL_FROZEN_DISK_IO:
820 		if (!(req->rq_state & RQ_LOCAL_COMPLETED))
821 			break;
822 		mod_rq_state(req, m, RQ_COMPLETION_SUSP, 0);
823 		break;
824 
825 	case RESTART_FROZEN_DISK_IO:
826 		if (!(req->rq_state & RQ_LOCAL_COMPLETED))
827 			break;
828 
829 		mod_rq_state(req, m,
830 				RQ_COMPLETION_SUSP|RQ_LOCAL_COMPLETED,
831 				RQ_LOCAL_PENDING);
832 
833 		rv = MR_READ;
834 		if (bio_data_dir(req->master_bio) == WRITE)
835 			rv = MR_WRITE;
836 
837 		get_ldev(device); /* always succeeds in this call path */
838 		req->w.cb = w_restart_disk_io;
839 		drbd_queue_work(&connection->sender_work,
840 				&req->w);
841 		break;
842 
843 	case RESEND:
844 		/* Simply complete (local only) READs. */
845 		if (!(req->rq_state & RQ_WRITE) && !req->w.cb) {
846 			mod_rq_state(req, m, RQ_COMPLETION_SUSP, 0);
847 			break;
848 		}
849 
850 		/* If RQ_NET_OK is already set, we got a P_WRITE_ACK or P_RECV_ACK
851 		   before the connection loss (B&C only); only P_BARRIER_ACK
852 		   (or the local completion?) was missing when we suspended.
853 		   Throwing them out of the TL here by pretending we got a BARRIER_ACK.
854 		   During connection handshake, we ensure that the peer was not rebooted. */
855 		if (!(req->rq_state & RQ_NET_OK)) {
856 			/* FIXME could this possibly be a req->dw.cb == w_send_out_of_sync?
857 			 * in that case we must not set RQ_NET_PENDING. */
858 
859 			mod_rq_state(req, m, RQ_COMPLETION_SUSP, RQ_NET_QUEUED|RQ_NET_PENDING);
860 			if (req->w.cb) {
861 				/* w.cb expected to be w_send_dblock, or w_send_read_req */
862 				drbd_queue_work(&connection->sender_work,
863 						&req->w);
864 				rv = req->rq_state & RQ_WRITE ? MR_WRITE : MR_READ;
865 			} /* else: FIXME can this happen? */
866 			break;
867 		}
868 		/* else, fall through to BARRIER_ACKED */
869 
870 	case BARRIER_ACKED:
871 		/* barrier ack for READ requests does not make sense */
872 		if (!(req->rq_state & RQ_WRITE))
873 			break;
874 
875 		if (req->rq_state & RQ_NET_PENDING) {
876 			/* barrier came in before all requests were acked.
877 			 * this is bad, because if the connection is lost now,
878 			 * we won't be able to clean them up... */
879 			drbd_err(device, "FIXME (BARRIER_ACKED but pending)\n");
880 		}
881 		/* Allowed to complete requests, even while suspended.
882 		 * As this is called for all requests within a matching epoch,
883 		 * we need to filter, and only set RQ_NET_DONE for those that
884 		 * have actually been on the wire. */
885 		mod_rq_state(req, m, RQ_COMPLETION_SUSP,
886 				(req->rq_state & RQ_NET_MASK) ? RQ_NET_DONE : 0);
887 		break;
888 
889 	case DATA_RECEIVED:
890 		D_ASSERT(device, req->rq_state & RQ_NET_PENDING);
891 		mod_rq_state(req, m, RQ_NET_PENDING, RQ_NET_OK|RQ_NET_DONE);
892 		break;
893 
894 	case QUEUE_AS_DRBD_BARRIER:
895 		start_new_tl_epoch(connection);
896 		mod_rq_state(req, m, 0, RQ_NET_OK|RQ_NET_DONE);
897 		break;
898 	};
899 
900 	return rv;
901 }
902 
903 /* we may do a local read if:
904  * - we are consistent (of course),
905  * - or we are generally inconsistent,
906  *   BUT we are still/already IN SYNC for this area.
907  *   since size may be bigger than BM_BLOCK_SIZE,
908  *   we may need to check several bits.
909  */
910 static bool drbd_may_do_local_read(struct drbd_device *device, sector_t sector, int size)
911 {
912 	unsigned long sbnr, ebnr;
913 	sector_t esector, nr_sectors;
914 
915 	if (device->state.disk == D_UP_TO_DATE)
916 		return true;
917 	if (device->state.disk != D_INCONSISTENT)
918 		return false;
919 	esector = sector + (size >> 9) - 1;
920 	nr_sectors = drbd_get_capacity(device->this_bdev);
921 	D_ASSERT(device, sector  < nr_sectors);
922 	D_ASSERT(device, esector < nr_sectors);
923 
924 	sbnr = BM_SECT_TO_BIT(sector);
925 	ebnr = BM_SECT_TO_BIT(esector);
926 
927 	return drbd_bm_count_bits(device, sbnr, ebnr) == 0;
928 }
929 
930 static bool remote_due_to_read_balancing(struct drbd_device *device, sector_t sector,
931 		enum drbd_read_balancing rbm)
932 {
933 	struct backing_dev_info *bdi;
934 	int stripe_shift;
935 
936 	switch (rbm) {
937 	case RB_CONGESTED_REMOTE:
938 		bdi = &device->ldev->backing_bdev->bd_disk->queue->backing_dev_info;
939 		return bdi_read_congested(bdi);
940 	case RB_LEAST_PENDING:
941 		return atomic_read(&device->local_cnt) >
942 			atomic_read(&device->ap_pending_cnt) + atomic_read(&device->rs_pending_cnt);
943 	case RB_32K_STRIPING:  /* stripe_shift = 15 */
944 	case RB_64K_STRIPING:
945 	case RB_128K_STRIPING:
946 	case RB_256K_STRIPING:
947 	case RB_512K_STRIPING:
948 	case RB_1M_STRIPING:   /* stripe_shift = 20 */
949 		stripe_shift = (rbm - RB_32K_STRIPING + 15);
950 		return (sector >> (stripe_shift - 9)) & 1;
951 	case RB_ROUND_ROBIN:
952 		return test_and_change_bit(READ_BALANCE_RR, &device->flags);
953 	case RB_PREFER_REMOTE:
954 		return true;
955 	case RB_PREFER_LOCAL:
956 	default:
957 		return false;
958 	}
959 }
960 
961 /*
962  * complete_conflicting_writes  -  wait for any conflicting write requests
963  *
964  * The write_requests tree contains all active write requests which we
965  * currently know about.  Wait for any requests to complete which conflict with
966  * the new one.
967  *
968  * Only way out: remove the conflicting intervals from the tree.
969  */
970 static void complete_conflicting_writes(struct drbd_request *req)
971 {
972 	DEFINE_WAIT(wait);
973 	struct drbd_device *device = req->device;
974 	struct drbd_interval *i;
975 	sector_t sector = req->i.sector;
976 	int size = req->i.size;
977 
978 	i = drbd_find_overlap(&device->write_requests, sector, size);
979 	if (!i)
980 		return;
981 
982 	for (;;) {
983 		prepare_to_wait(&device->misc_wait, &wait, TASK_UNINTERRUPTIBLE);
984 		i = drbd_find_overlap(&device->write_requests, sector, size);
985 		if (!i)
986 			break;
987 		/* Indicate to wake up device->misc_wait on progress.  */
988 		i->waiting = true;
989 		spin_unlock_irq(&device->resource->req_lock);
990 		schedule();
991 		spin_lock_irq(&device->resource->req_lock);
992 	}
993 	finish_wait(&device->misc_wait, &wait);
994 }
995 
996 /* called within req_lock and rcu_read_lock() */
997 static void maybe_pull_ahead(struct drbd_device *device)
998 {
999 	struct drbd_connection *connection = first_peer_device(device)->connection;
1000 	struct net_conf *nc;
1001 	bool congested = false;
1002 	enum drbd_on_congestion on_congestion;
1003 
1004 	rcu_read_lock();
1005 	nc = rcu_dereference(connection->net_conf);
1006 	on_congestion = nc ? nc->on_congestion : OC_BLOCK;
1007 	rcu_read_unlock();
1008 	if (on_congestion == OC_BLOCK ||
1009 	    connection->agreed_pro_version < 96)
1010 		return;
1011 
1012 	if (on_congestion == OC_PULL_AHEAD && device->state.conn == C_AHEAD)
1013 		return; /* nothing to do ... */
1014 
1015 	/* If I don't even have good local storage, we can not reasonably try
1016 	 * to pull ahead of the peer. We also need the local reference to make
1017 	 * sure device->act_log is there.
1018 	 */
1019 	if (!get_ldev_if_state(device, D_UP_TO_DATE))
1020 		return;
1021 
1022 	if (nc->cong_fill &&
1023 	    atomic_read(&device->ap_in_flight) >= nc->cong_fill) {
1024 		drbd_info(device, "Congestion-fill threshold reached\n");
1025 		congested = true;
1026 	}
1027 
1028 	if (device->act_log->used >= nc->cong_extents) {
1029 		drbd_info(device, "Congestion-extents threshold reached\n");
1030 		congested = true;
1031 	}
1032 
1033 	if (congested) {
1034 		/* start a new epoch for non-mirrored writes */
1035 		start_new_tl_epoch(first_peer_device(device)->connection);
1036 
1037 		if (on_congestion == OC_PULL_AHEAD)
1038 			_drbd_set_state(_NS(device, conn, C_AHEAD), 0, NULL);
1039 		else  /*nc->on_congestion == OC_DISCONNECT */
1040 			_drbd_set_state(_NS(device, conn, C_DISCONNECTING), 0, NULL);
1041 	}
1042 	put_ldev(device);
1043 }
1044 
1045 /* If this returns false, and req->private_bio is still set,
1046  * this should be submitted locally.
1047  *
1048  * If it returns false, but req->private_bio is not set,
1049  * we do not have access to good data :(
1050  *
1051  * Otherwise, this destroys req->private_bio, if any,
1052  * and returns true.
1053  */
1054 static bool do_remote_read(struct drbd_request *req)
1055 {
1056 	struct drbd_device *device = req->device;
1057 	enum drbd_read_balancing rbm;
1058 
1059 	if (req->private_bio) {
1060 		if (!drbd_may_do_local_read(device,
1061 					req->i.sector, req->i.size)) {
1062 			bio_put(req->private_bio);
1063 			req->private_bio = NULL;
1064 			put_ldev(device);
1065 		}
1066 	}
1067 
1068 	if (device->state.pdsk != D_UP_TO_DATE)
1069 		return false;
1070 
1071 	if (req->private_bio == NULL)
1072 		return true;
1073 
1074 	/* TODO: improve read balancing decisions, take into account drbd
1075 	 * protocol, pending requests etc. */
1076 
1077 	rcu_read_lock();
1078 	rbm = rcu_dereference(device->ldev->disk_conf)->read_balancing;
1079 	rcu_read_unlock();
1080 
1081 	if (rbm == RB_PREFER_LOCAL && req->private_bio)
1082 		return false; /* submit locally */
1083 
1084 	if (remote_due_to_read_balancing(device, req->i.sector, rbm)) {
1085 		if (req->private_bio) {
1086 			bio_put(req->private_bio);
1087 			req->private_bio = NULL;
1088 			put_ldev(device);
1089 		}
1090 		return true;
1091 	}
1092 
1093 	return false;
1094 }
1095 
1096 /* returns number of connections (== 1, for drbd 8.4)
1097  * expected to actually write this data,
1098  * which does NOT include those that we are L_AHEAD for. */
1099 static int drbd_process_write_request(struct drbd_request *req)
1100 {
1101 	struct drbd_device *device = req->device;
1102 	int remote, send_oos;
1103 
1104 	remote = drbd_should_do_remote(device->state);
1105 	send_oos = drbd_should_send_out_of_sync(device->state);
1106 
1107 	/* Need to replicate writes.  Unless it is an empty flush,
1108 	 * which is better mapped to a DRBD P_BARRIER packet,
1109 	 * also for drbd wire protocol compatibility reasons.
1110 	 * If this was a flush, just start a new epoch.
1111 	 * Unless the current epoch was empty anyways, or we are not currently
1112 	 * replicating, in which case there is no point. */
1113 	if (unlikely(req->i.size == 0)) {
1114 		/* The only size==0 bios we expect are empty flushes. */
1115 		D_ASSERT(device, req->master_bio->bi_rw & REQ_FLUSH);
1116 		if (remote)
1117 			_req_mod(req, QUEUE_AS_DRBD_BARRIER);
1118 		return remote;
1119 	}
1120 
1121 	if (!remote && !send_oos)
1122 		return 0;
1123 
1124 	D_ASSERT(device, !(remote && send_oos));
1125 
1126 	if (remote) {
1127 		_req_mod(req, TO_BE_SENT);
1128 		_req_mod(req, QUEUE_FOR_NET_WRITE);
1129 	} else if (drbd_set_out_of_sync(device, req->i.sector, req->i.size))
1130 		_req_mod(req, QUEUE_FOR_SEND_OOS);
1131 
1132 	return remote;
1133 }
1134 
1135 static void
1136 drbd_submit_req_private_bio(struct drbd_request *req)
1137 {
1138 	struct drbd_device *device = req->device;
1139 	struct bio *bio = req->private_bio;
1140 	const int rw = bio_rw(bio);
1141 
1142 	bio->bi_bdev = device->ldev->backing_bdev;
1143 
1144 	/* State may have changed since we grabbed our reference on the
1145 	 * ->ldev member. Double check, and short-circuit to endio.
1146 	 * In case the last activity log transaction failed to get on
1147 	 * stable storage, and this is a WRITE, we may not even submit
1148 	 * this bio. */
1149 	if (get_ldev(device)) {
1150 		req->pre_submit_jif = jiffies;
1151 		if (drbd_insert_fault(device,
1152 				      rw == WRITE ? DRBD_FAULT_DT_WR
1153 				    : rw == READ  ? DRBD_FAULT_DT_RD
1154 				    :               DRBD_FAULT_DT_RA))
1155 			bio_endio(bio, -EIO);
1156 		else
1157 			generic_make_request(bio);
1158 		put_ldev(device);
1159 	} else
1160 		bio_endio(bio, -EIO);
1161 }
1162 
1163 static void drbd_queue_write(struct drbd_device *device, struct drbd_request *req)
1164 {
1165 	spin_lock_irq(&device->resource->req_lock);
1166 	list_add_tail(&req->tl_requests, &device->submit.writes);
1167 	list_add_tail(&req->req_pending_master_completion,
1168 			&device->pending_master_completion[1 /* WRITE */]);
1169 	spin_unlock_irq(&device->resource->req_lock);
1170 	queue_work(device->submit.wq, &device->submit.worker);
1171 	/* do_submit() may sleep internally on al_wait, too */
1172 	wake_up(&device->al_wait);
1173 }
1174 
1175 /* returns the new drbd_request pointer, if the caller is expected to
1176  * drbd_send_and_submit() it (to save latency), or NULL if we queued the
1177  * request on the submitter thread.
1178  * Returns ERR_PTR(-ENOMEM) if we cannot allocate a drbd_request.
1179  */
1180 static struct drbd_request *
1181 drbd_request_prepare(struct drbd_device *device, struct bio *bio, unsigned long start_jif)
1182 {
1183 	const int rw = bio_data_dir(bio);
1184 	struct drbd_request *req;
1185 
1186 	/* allocate outside of all locks; */
1187 	req = drbd_req_new(device, bio);
1188 	if (!req) {
1189 		dec_ap_bio(device);
1190 		/* only pass the error to the upper layers.
1191 		 * if user cannot handle io errors, that's not our business. */
1192 		drbd_err(device, "could not kmalloc() req\n");
1193 		bio_endio(bio, -ENOMEM);
1194 		return ERR_PTR(-ENOMEM);
1195 	}
1196 	req->start_jif = start_jif;
1197 
1198 	if (!get_ldev(device)) {
1199 		bio_put(req->private_bio);
1200 		req->private_bio = NULL;
1201 	}
1202 
1203 	/* Update disk stats */
1204 	_drbd_start_io_acct(device, req);
1205 
1206 	if (rw == WRITE && req->private_bio && req->i.size
1207 	&& !test_bit(AL_SUSPENDED, &device->flags)) {
1208 		if (!drbd_al_begin_io_fastpath(device, &req->i)) {
1209 			atomic_inc(&device->ap_actlog_cnt);
1210 			drbd_queue_write(device, req);
1211 			return NULL;
1212 		}
1213 		req->rq_state |= RQ_IN_ACT_LOG;
1214 		req->in_actlog_jif = jiffies;
1215 	}
1216 
1217 	return req;
1218 }
1219 
1220 static void drbd_send_and_submit(struct drbd_device *device, struct drbd_request *req)
1221 {
1222 	struct drbd_resource *resource = device->resource;
1223 	const int rw = bio_rw(req->master_bio);
1224 	struct bio_and_error m = { NULL, };
1225 	bool no_remote = false;
1226 	bool submit_private_bio = false;
1227 
1228 	spin_lock_irq(&resource->req_lock);
1229 	if (rw == WRITE) {
1230 		/* This may temporarily give up the req_lock,
1231 		 * but will re-aquire it before it returns here.
1232 		 * Needs to be before the check on drbd_suspended() */
1233 		complete_conflicting_writes(req);
1234 		/* no more giving up req_lock from now on! */
1235 
1236 		/* check for congestion, and potentially stop sending
1237 		 * full data updates, but start sending "dirty bits" only. */
1238 		maybe_pull_ahead(device);
1239 	}
1240 
1241 
1242 	if (drbd_suspended(device)) {
1243 		/* push back and retry: */
1244 		req->rq_state |= RQ_POSTPONED;
1245 		if (req->private_bio) {
1246 			bio_put(req->private_bio);
1247 			req->private_bio = NULL;
1248 			put_ldev(device);
1249 		}
1250 		goto out;
1251 	}
1252 
1253 	/* We fail READ/READA early, if we can not serve it.
1254 	 * We must do this before req is registered on any lists.
1255 	 * Otherwise, drbd_req_complete() will queue failed READ for retry. */
1256 	if (rw != WRITE) {
1257 		if (!do_remote_read(req) && !req->private_bio)
1258 			goto nodata;
1259 	}
1260 
1261 	/* which transfer log epoch does this belong to? */
1262 	req->epoch = atomic_read(&first_peer_device(device)->connection->current_tle_nr);
1263 
1264 	/* no point in adding empty flushes to the transfer log,
1265 	 * they are mapped to drbd barriers already. */
1266 	if (likely(req->i.size!=0)) {
1267 		if (rw == WRITE)
1268 			first_peer_device(device)->connection->current_tle_writes++;
1269 
1270 		list_add_tail(&req->tl_requests, &first_peer_device(device)->connection->transfer_log);
1271 	}
1272 
1273 	if (rw == WRITE) {
1274 		if (!drbd_process_write_request(req))
1275 			no_remote = true;
1276 	} else {
1277 		/* We either have a private_bio, or we can read from remote.
1278 		 * Otherwise we had done the goto nodata above. */
1279 		if (req->private_bio == NULL) {
1280 			_req_mod(req, TO_BE_SENT);
1281 			_req_mod(req, QUEUE_FOR_NET_READ);
1282 		} else
1283 			no_remote = true;
1284 	}
1285 
1286 	/* If it took the fast path in drbd_request_prepare, add it here.
1287 	 * The slow path has added it already. */
1288 	if (list_empty(&req->req_pending_master_completion))
1289 		list_add_tail(&req->req_pending_master_completion,
1290 			&device->pending_master_completion[rw == WRITE]);
1291 	if (req->private_bio) {
1292 		/* needs to be marked within the same spinlock */
1293 		list_add_tail(&req->req_pending_local,
1294 			&device->pending_completion[rw == WRITE]);
1295 		_req_mod(req, TO_BE_SUBMITTED);
1296 		/* but we need to give up the spinlock to submit */
1297 		submit_private_bio = true;
1298 	} else if (no_remote) {
1299 nodata:
1300 		if (__ratelimit(&drbd_ratelimit_state))
1301 			drbd_err(device, "IO ERROR: neither local nor remote data, sector %llu+%u\n",
1302 					(unsigned long long)req->i.sector, req->i.size >> 9);
1303 		/* A write may have been queued for send_oos, however.
1304 		 * So we can not simply free it, we must go through drbd_req_put_completion_ref() */
1305 	}
1306 
1307 out:
1308 	if (drbd_req_put_completion_ref(req, &m, 1))
1309 		kref_put(&req->kref, drbd_req_destroy);
1310 	spin_unlock_irq(&resource->req_lock);
1311 
1312 	/* Even though above is a kref_put(), this is safe.
1313 	 * As long as we still need to submit our private bio,
1314 	 * we hold a completion ref, and the request cannot disappear.
1315 	 * If however this request did not even have a private bio to submit
1316 	 * (e.g. remote read), req may already be invalid now.
1317 	 * That's why we cannot check on req->private_bio. */
1318 	if (submit_private_bio)
1319 		drbd_submit_req_private_bio(req);
1320 	if (m.bio)
1321 		complete_master_bio(device, &m);
1322 }
1323 
1324 void __drbd_make_request(struct drbd_device *device, struct bio *bio, unsigned long start_jif)
1325 {
1326 	struct drbd_request *req = drbd_request_prepare(device, bio, start_jif);
1327 	if (IS_ERR_OR_NULL(req))
1328 		return;
1329 	drbd_send_and_submit(device, req);
1330 }
1331 
1332 static void submit_fast_path(struct drbd_device *device, struct list_head *incoming)
1333 {
1334 	struct drbd_request *req, *tmp;
1335 	list_for_each_entry_safe(req, tmp, incoming, tl_requests) {
1336 		const int rw = bio_data_dir(req->master_bio);
1337 
1338 		if (rw == WRITE /* rw != WRITE should not even end up here! */
1339 		&& req->private_bio && req->i.size
1340 		&& !test_bit(AL_SUSPENDED, &device->flags)) {
1341 			if (!drbd_al_begin_io_fastpath(device, &req->i))
1342 				continue;
1343 
1344 			req->rq_state |= RQ_IN_ACT_LOG;
1345 			req->in_actlog_jif = jiffies;
1346 			atomic_dec(&device->ap_actlog_cnt);
1347 		}
1348 
1349 		list_del_init(&req->tl_requests);
1350 		drbd_send_and_submit(device, req);
1351 	}
1352 }
1353 
1354 static bool prepare_al_transaction_nonblock(struct drbd_device *device,
1355 					    struct list_head *incoming,
1356 					    struct list_head *pending,
1357 					    struct list_head *later)
1358 {
1359 	struct drbd_request *req, *tmp;
1360 	int wake = 0;
1361 	int err;
1362 
1363 	spin_lock_irq(&device->al_lock);
1364 	list_for_each_entry_safe(req, tmp, incoming, tl_requests) {
1365 		err = drbd_al_begin_io_nonblock(device, &req->i);
1366 		if (err == -ENOBUFS)
1367 			break;
1368 		if (err == -EBUSY)
1369 			wake = 1;
1370 		if (err)
1371 			list_move_tail(&req->tl_requests, later);
1372 		else
1373 			list_move_tail(&req->tl_requests, pending);
1374 	}
1375 	spin_unlock_irq(&device->al_lock);
1376 	if (wake)
1377 		wake_up(&device->al_wait);
1378 	return !list_empty(pending);
1379 }
1380 
1381 void send_and_submit_pending(struct drbd_device *device, struct list_head *pending)
1382 {
1383 	struct drbd_request *req, *tmp;
1384 
1385 	list_for_each_entry_safe(req, tmp, pending, tl_requests) {
1386 		req->rq_state |= RQ_IN_ACT_LOG;
1387 		req->in_actlog_jif = jiffies;
1388 		atomic_dec(&device->ap_actlog_cnt);
1389 		list_del_init(&req->tl_requests);
1390 		drbd_send_and_submit(device, req);
1391 	}
1392 }
1393 
1394 void do_submit(struct work_struct *ws)
1395 {
1396 	struct drbd_device *device = container_of(ws, struct drbd_device, submit.worker);
1397 	LIST_HEAD(incoming);	/* from drbd_make_request() */
1398 	LIST_HEAD(pending);	/* to be submitted after next AL-transaction commit */
1399 	LIST_HEAD(busy);	/* blocked by resync requests */
1400 
1401 	/* grab new incoming requests */
1402 	spin_lock_irq(&device->resource->req_lock);
1403 	list_splice_tail_init(&device->submit.writes, &incoming);
1404 	spin_unlock_irq(&device->resource->req_lock);
1405 
1406 	for (;;) {
1407 		DEFINE_WAIT(wait);
1408 
1409 		/* move used-to-be-busy back to front of incoming */
1410 		list_splice_init(&busy, &incoming);
1411 		submit_fast_path(device, &incoming);
1412 		if (list_empty(&incoming))
1413 			break;
1414 
1415 		for (;;) {
1416 			prepare_to_wait(&device->al_wait, &wait, TASK_UNINTERRUPTIBLE);
1417 
1418 			list_splice_init(&busy, &incoming);
1419 			prepare_al_transaction_nonblock(device, &incoming, &pending, &busy);
1420 			if (!list_empty(&pending))
1421 				break;
1422 
1423 			schedule();
1424 
1425 			/* If all currently "hot" activity log extents are kept busy by
1426 			 * incoming requests, we still must not totally starve new
1427 			 * requests to "cold" extents.
1428 			 * Something left on &incoming means there had not been
1429 			 * enough update slots available, and the activity log
1430 			 * has been marked as "starving".
1431 			 *
1432 			 * Try again now, without looking for new requests,
1433 			 * effectively blocking all new requests until we made
1434 			 * at least _some_ progress with what we currently have.
1435 			 */
1436 			if (!list_empty(&incoming))
1437 				continue;
1438 
1439 			/* Nothing moved to pending, but nothing left
1440 			 * on incoming: all moved to busy!
1441 			 * Grab new and iterate. */
1442 			spin_lock_irq(&device->resource->req_lock);
1443 			list_splice_tail_init(&device->submit.writes, &incoming);
1444 			spin_unlock_irq(&device->resource->req_lock);
1445 		}
1446 		finish_wait(&device->al_wait, &wait);
1447 
1448 		/* If the transaction was full, before all incoming requests
1449 		 * had been processed, skip ahead to commit, and iterate
1450 		 * without splicing in more incoming requests from upper layers.
1451 		 *
1452 		 * Else, if all incoming have been processed,
1453 		 * they have become either "pending" (to be submitted after
1454 		 * next transaction commit) or "busy" (blocked by resync).
1455 		 *
1456 		 * Maybe more was queued, while we prepared the transaction?
1457 		 * Try to stuff those into this transaction as well.
1458 		 * Be strictly non-blocking here,
1459 		 * we already have something to commit.
1460 		 *
1461 		 * Commit if we don't make any more progres.
1462 		 */
1463 
1464 		while (list_empty(&incoming)) {
1465 			LIST_HEAD(more_pending);
1466 			LIST_HEAD(more_incoming);
1467 			bool made_progress;
1468 
1469 			/* It is ok to look outside the lock,
1470 			 * it's only an optimization anyways */
1471 			if (list_empty(&device->submit.writes))
1472 				break;
1473 
1474 			spin_lock_irq(&device->resource->req_lock);
1475 			list_splice_tail_init(&device->submit.writes, &more_incoming);
1476 			spin_unlock_irq(&device->resource->req_lock);
1477 
1478 			if (list_empty(&more_incoming))
1479 				break;
1480 
1481 			made_progress = prepare_al_transaction_nonblock(device, &more_incoming, &more_pending, &busy);
1482 
1483 			list_splice_tail_init(&more_pending, &pending);
1484 			list_splice_tail_init(&more_incoming, &incoming);
1485 			if (!made_progress)
1486 				break;
1487 		}
1488 
1489 		drbd_al_begin_io_commit(device);
1490 		send_and_submit_pending(device, &pending);
1491 	}
1492 }
1493 
1494 void drbd_make_request(struct request_queue *q, struct bio *bio)
1495 {
1496 	struct drbd_device *device = (struct drbd_device *) q->queuedata;
1497 	unsigned long start_jif;
1498 
1499 	start_jif = jiffies;
1500 
1501 	/*
1502 	 * what we "blindly" assume:
1503 	 */
1504 	D_ASSERT(device, IS_ALIGNED(bio->bi_iter.bi_size, 512));
1505 
1506 	inc_ap_bio(device);
1507 	__drbd_make_request(device, bio, start_jif);
1508 }
1509 
1510 /* This is called by bio_add_page().
1511  *
1512  * q->max_hw_sectors and other global limits are already enforced there.
1513  *
1514  * We need to call down to our lower level device,
1515  * in case it has special restrictions.
1516  *
1517  * We also may need to enforce configured max-bio-bvecs limits.
1518  *
1519  * As long as the BIO is empty we have to allow at least one bvec,
1520  * regardless of size and offset, so no need to ask lower levels.
1521  */
1522 int drbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bvm, struct bio_vec *bvec)
1523 {
1524 	struct drbd_device *device = (struct drbd_device *) q->queuedata;
1525 	unsigned int bio_size = bvm->bi_size;
1526 	int limit = DRBD_MAX_BIO_SIZE;
1527 	int backing_limit;
1528 
1529 	if (bio_size && get_ldev(device)) {
1530 		unsigned int max_hw_sectors = queue_max_hw_sectors(q);
1531 		struct request_queue * const b =
1532 			device->ldev->backing_bdev->bd_disk->queue;
1533 		if (b->merge_bvec_fn) {
1534 			bvm->bi_bdev = device->ldev->backing_bdev;
1535 			backing_limit = b->merge_bvec_fn(b, bvm, bvec);
1536 			limit = min(limit, backing_limit);
1537 		}
1538 		put_ldev(device);
1539 		if ((limit >> 9) > max_hw_sectors)
1540 			limit = max_hw_sectors << 9;
1541 	}
1542 	return limit;
1543 }
1544 
1545 void request_timer_fn(unsigned long data)
1546 {
1547 	struct drbd_device *device = (struct drbd_device *) data;
1548 	struct drbd_connection *connection = first_peer_device(device)->connection;
1549 	struct drbd_request *req_read, *req_write, *req_peer; /* oldest request */
1550 	struct net_conf *nc;
1551 	unsigned long oldest_submit_jif;
1552 	unsigned long ent = 0, dt = 0, et, nt; /* effective timeout = ko_count * timeout */
1553 	unsigned long now;
1554 
1555 	rcu_read_lock();
1556 	nc = rcu_dereference(connection->net_conf);
1557 	if (nc && device->state.conn >= C_WF_REPORT_PARAMS)
1558 		ent = nc->timeout * HZ/10 * nc->ko_count;
1559 
1560 	if (get_ldev(device)) { /* implicit state.disk >= D_INCONSISTENT */
1561 		dt = rcu_dereference(device->ldev->disk_conf)->disk_timeout * HZ / 10;
1562 		put_ldev(device);
1563 	}
1564 	rcu_read_unlock();
1565 
1566 	et = min_not_zero(dt, ent);
1567 
1568 	if (!et)
1569 		return; /* Recurring timer stopped */
1570 
1571 	now = jiffies;
1572 	nt = now + et;
1573 
1574 	spin_lock_irq(&device->resource->req_lock);
1575 	req_read = list_first_entry_or_null(&device->pending_completion[0], struct drbd_request, req_pending_local);
1576 	req_write = list_first_entry_or_null(&device->pending_completion[1], struct drbd_request, req_pending_local);
1577 	req_peer = connection->req_not_net_done;
1578 	/* maybe the oldest request waiting for the peer is in fact still
1579 	 * blocking in tcp sendmsg */
1580 	if (!req_peer && connection->req_next && connection->req_next->pre_send_jif)
1581 		req_peer = connection->req_next;
1582 
1583 	/* evaluate the oldest peer request only in one timer! */
1584 	if (req_peer && req_peer->device != device)
1585 		req_peer = NULL;
1586 
1587 	/* do we have something to evaluate? */
1588 	if (req_peer == NULL && req_write == NULL && req_read == NULL)
1589 		goto out;
1590 
1591 	oldest_submit_jif =
1592 		(req_write && req_read)
1593 		? ( time_before(req_write->pre_submit_jif, req_read->pre_submit_jif)
1594 		  ? req_write->pre_submit_jif : req_read->pre_submit_jif )
1595 		: req_write ? req_write->pre_submit_jif
1596 		: req_read ? req_read->pre_submit_jif : now;
1597 
1598 	/* The request is considered timed out, if
1599 	 * - we have some effective timeout from the configuration,
1600 	 *   with above state restrictions applied,
1601 	 * - the oldest request is waiting for a response from the network
1602 	 *   resp. the local disk,
1603 	 * - the oldest request is in fact older than the effective timeout,
1604 	 * - the connection was established (resp. disk was attached)
1605 	 *   for longer than the timeout already.
1606 	 * Note that for 32bit jiffies and very stable connections/disks,
1607 	 * we may have a wrap around, which is catched by
1608 	 *   !time_in_range(now, last_..._jif, last_..._jif + timeout).
1609 	 *
1610 	 * Side effect: once per 32bit wrap-around interval, which means every
1611 	 * ~198 days with 250 HZ, we have a window where the timeout would need
1612 	 * to expire twice (worst case) to become effective. Good enough.
1613 	 */
1614 	if (ent && req_peer &&
1615 		 time_after(now, req_peer->pre_send_jif + ent) &&
1616 		!time_in_range(now, connection->last_reconnect_jif, connection->last_reconnect_jif + ent)) {
1617 		drbd_warn(device, "Remote failed to finish a request within ko-count * timeout\n");
1618 		_conn_request_state(connection, NS(conn, C_TIMEOUT), CS_VERBOSE | CS_HARD);
1619 	}
1620 	if (dt && oldest_submit_jif != now &&
1621 		 time_after(now, oldest_submit_jif + dt) &&
1622 		!time_in_range(now, device->last_reattach_jif, device->last_reattach_jif + dt)) {
1623 		drbd_warn(device, "Local backing device failed to meet the disk-timeout\n");
1624 		__drbd_chk_io_error(device, DRBD_FORCE_DETACH);
1625 	}
1626 
1627 	/* Reschedule timer for the nearest not already expired timeout.
1628 	 * Fallback to now + min(effective network timeout, disk timeout). */
1629 	ent = (ent && req_peer && time_before(now, req_peer->pre_send_jif + ent))
1630 		? req_peer->pre_send_jif + ent : now + et;
1631 	dt = (dt && oldest_submit_jif != now && time_before(now, oldest_submit_jif + dt))
1632 		? oldest_submit_jif + dt : now + et;
1633 	nt = time_before(ent, dt) ? ent : dt;
1634 out:
1635 	spin_unlock_irq(&device->resource->req_lock);
1636 	mod_timer(&device->request_timer, nt);
1637 }
1638