1 // SPDX-License-Identifier: GPL-2.0-only
2 /*
3    drbd_receiver.c
4 
5    This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 
7    Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
8    Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
9    Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 
11  */
12 
13 
14 #include <linux/module.h>
15 
16 #include <linux/uaccess.h>
17 #include <net/sock.h>
18 
19 #include <linux/drbd.h>
20 #include <linux/fs.h>
21 #include <linux/file.h>
22 #include <linux/in.h>
23 #include <linux/mm.h>
24 #include <linux/memcontrol.h>
25 #include <linux/mm_inline.h>
26 #include <linux/slab.h>
27 #include <uapi/linux/sched/types.h>
28 #include <linux/sched/signal.h>
29 #include <linux/pkt_sched.h>
30 #include <linux/unistd.h>
31 #include <linux/vmalloc.h>
32 #include <linux/random.h>
33 #include <linux/string.h>
34 #include <linux/scatterlist.h>
35 #include <linux/part_stat.h>
36 #include "drbd_int.h"
37 #include "drbd_protocol.h"
38 #include "drbd_req.h"
39 #include "drbd_vli.h"
40 
41 #define PRO_FEATURES (DRBD_FF_TRIM|DRBD_FF_THIN_RESYNC|DRBD_FF_WSAME|DRBD_FF_WZEROES)
42 
43 struct packet_info {
44 	enum drbd_packet cmd;
45 	unsigned int size;
46 	unsigned int vnr;
47 	void *data;
48 };
49 
50 enum finish_epoch {
51 	FE_STILL_LIVE,
52 	FE_DESTROYED,
53 	FE_RECYCLED,
54 };
55 
56 static int drbd_do_features(struct drbd_connection *connection);
57 static int drbd_do_auth(struct drbd_connection *connection);
58 static int drbd_disconnected(struct drbd_peer_device *);
59 static void conn_wait_active_ee_empty(struct drbd_connection *connection);
60 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
61 static int e_end_block(struct drbd_work *, int);
62 
63 
64 #define GFP_TRY	(__GFP_HIGHMEM | __GFP_NOWARN)
65 
66 /*
67  * some helper functions to deal with single linked page lists,
68  * page->private being our "next" pointer.
69  */
70 
71 /* If at least n pages are linked at head, get n pages off.
72  * Otherwise, don't modify head, and return NULL.
73  * Locking is the responsibility of the caller.
74  */
75 static struct page *page_chain_del(struct page **head, int n)
76 {
77 	struct page *page;
78 	struct page *tmp;
79 
80 	BUG_ON(!n);
81 	BUG_ON(!head);
82 
83 	page = *head;
84 
85 	if (!page)
86 		return NULL;
87 
88 	while (page) {
89 		tmp = page_chain_next(page);
90 		if (--n == 0)
91 			break; /* found sufficient pages */
92 		if (tmp == NULL)
93 			/* insufficient pages, don't use any of them. */
94 			return NULL;
95 		page = tmp;
96 	}
97 
98 	/* add end of list marker for the returned list */
99 	set_page_private(page, 0);
100 	/* actual return value, and adjustment of head */
101 	page = *head;
102 	*head = tmp;
103 	return page;
104 }
105 
106 /* may be used outside of locks to find the tail of a (usually short)
107  * "private" page chain, before adding it back to a global chain head
108  * with page_chain_add() under a spinlock. */
109 static struct page *page_chain_tail(struct page *page, int *len)
110 {
111 	struct page *tmp;
112 	int i = 1;
113 	while ((tmp = page_chain_next(page))) {
114 		++i;
115 		page = tmp;
116 	}
117 	if (len)
118 		*len = i;
119 	return page;
120 }
121 
122 static int page_chain_free(struct page *page)
123 {
124 	struct page *tmp;
125 	int i = 0;
126 	page_chain_for_each_safe(page, tmp) {
127 		put_page(page);
128 		++i;
129 	}
130 	return i;
131 }
132 
133 static void page_chain_add(struct page **head,
134 		struct page *chain_first, struct page *chain_last)
135 {
136 #if 1
137 	struct page *tmp;
138 	tmp = page_chain_tail(chain_first, NULL);
139 	BUG_ON(tmp != chain_last);
140 #endif
141 
142 	/* add chain to head */
143 	set_page_private(chain_last, (unsigned long)*head);
144 	*head = chain_first;
145 }
146 
147 static struct page *__drbd_alloc_pages(struct drbd_device *device,
148 				       unsigned int number)
149 {
150 	struct page *page = NULL;
151 	struct page *tmp = NULL;
152 	unsigned int i = 0;
153 
154 	/* Yes, testing drbd_pp_vacant outside the lock is racy.
155 	 * So what. It saves a spin_lock. */
156 	if (drbd_pp_vacant >= number) {
157 		spin_lock(&drbd_pp_lock);
158 		page = page_chain_del(&drbd_pp_pool, number);
159 		if (page)
160 			drbd_pp_vacant -= number;
161 		spin_unlock(&drbd_pp_lock);
162 		if (page)
163 			return page;
164 	}
165 
166 	/* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
167 	 * "criss-cross" setup, that might cause write-out on some other DRBD,
168 	 * which in turn might block on the other node at this very place.  */
169 	for (i = 0; i < number; i++) {
170 		tmp = alloc_page(GFP_TRY);
171 		if (!tmp)
172 			break;
173 		set_page_private(tmp, (unsigned long)page);
174 		page = tmp;
175 	}
176 
177 	if (i == number)
178 		return page;
179 
180 	/* Not enough pages immediately available this time.
181 	 * No need to jump around here, drbd_alloc_pages will retry this
182 	 * function "soon". */
183 	if (page) {
184 		tmp = page_chain_tail(page, NULL);
185 		spin_lock(&drbd_pp_lock);
186 		page_chain_add(&drbd_pp_pool, page, tmp);
187 		drbd_pp_vacant += i;
188 		spin_unlock(&drbd_pp_lock);
189 	}
190 	return NULL;
191 }
192 
193 static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
194 					   struct list_head *to_be_freed)
195 {
196 	struct drbd_peer_request *peer_req, *tmp;
197 
198 	/* The EEs are always appended to the end of the list. Since
199 	   they are sent in order over the wire, they have to finish
200 	   in order. As soon as we see the first not finished we can
201 	   stop to examine the list... */
202 
203 	list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
204 		if (drbd_peer_req_has_active_page(peer_req))
205 			break;
206 		list_move(&peer_req->w.list, to_be_freed);
207 	}
208 }
209 
210 static void drbd_reclaim_net_peer_reqs(struct drbd_device *device)
211 {
212 	LIST_HEAD(reclaimed);
213 	struct drbd_peer_request *peer_req, *t;
214 
215 	spin_lock_irq(&device->resource->req_lock);
216 	reclaim_finished_net_peer_reqs(device, &reclaimed);
217 	spin_unlock_irq(&device->resource->req_lock);
218 	list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
219 		drbd_free_net_peer_req(device, peer_req);
220 }
221 
222 static void conn_reclaim_net_peer_reqs(struct drbd_connection *connection)
223 {
224 	struct drbd_peer_device *peer_device;
225 	int vnr;
226 
227 	rcu_read_lock();
228 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
229 		struct drbd_device *device = peer_device->device;
230 		if (!atomic_read(&device->pp_in_use_by_net))
231 			continue;
232 
233 		kref_get(&device->kref);
234 		rcu_read_unlock();
235 		drbd_reclaim_net_peer_reqs(device);
236 		kref_put(&device->kref, drbd_destroy_device);
237 		rcu_read_lock();
238 	}
239 	rcu_read_unlock();
240 }
241 
242 /**
243  * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
244  * @peer_device:	DRBD device.
245  * @number:		number of pages requested
246  * @retry:		whether to retry, if not enough pages are available right now
247  *
248  * Tries to allocate number pages, first from our own page pool, then from
249  * the kernel.
250  * Possibly retry until DRBD frees sufficient pages somewhere else.
251  *
252  * If this allocation would exceed the max_buffers setting, we throttle
253  * allocation (schedule_timeout) to give the system some room to breathe.
254  *
255  * We do not use max-buffers as hard limit, because it could lead to
256  * congestion and further to a distributed deadlock during online-verify or
257  * (checksum based) resync, if the max-buffers, socket buffer sizes and
258  * resync-rate settings are mis-configured.
259  *
260  * Returns a page chain linked via page->private.
261  */
262 struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
263 			      bool retry)
264 {
265 	struct drbd_device *device = peer_device->device;
266 	struct page *page = NULL;
267 	struct net_conf *nc;
268 	DEFINE_WAIT(wait);
269 	unsigned int mxb;
270 
271 	rcu_read_lock();
272 	nc = rcu_dereference(peer_device->connection->net_conf);
273 	mxb = nc ? nc->max_buffers : 1000000;
274 	rcu_read_unlock();
275 
276 	if (atomic_read(&device->pp_in_use) < mxb)
277 		page = __drbd_alloc_pages(device, number);
278 
279 	/* Try to keep the fast path fast, but occasionally we need
280 	 * to reclaim the pages we lended to the network stack. */
281 	if (page && atomic_read(&device->pp_in_use_by_net) > 512)
282 		drbd_reclaim_net_peer_reqs(device);
283 
284 	while (page == NULL) {
285 		prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
286 
287 		drbd_reclaim_net_peer_reqs(device);
288 
289 		if (atomic_read(&device->pp_in_use) < mxb) {
290 			page = __drbd_alloc_pages(device, number);
291 			if (page)
292 				break;
293 		}
294 
295 		if (!retry)
296 			break;
297 
298 		if (signal_pending(current)) {
299 			drbd_warn(device, "drbd_alloc_pages interrupted!\n");
300 			break;
301 		}
302 
303 		if (schedule_timeout(HZ/10) == 0)
304 			mxb = UINT_MAX;
305 	}
306 	finish_wait(&drbd_pp_wait, &wait);
307 
308 	if (page)
309 		atomic_add(number, &device->pp_in_use);
310 	return page;
311 }
312 
313 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
314  * Is also used from inside an other spin_lock_irq(&resource->req_lock);
315  * Either links the page chain back to the global pool,
316  * or returns all pages to the system. */
317 static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
318 {
319 	atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
320 	int i;
321 
322 	if (page == NULL)
323 		return;
324 
325 	if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * drbd_minor_count)
326 		i = page_chain_free(page);
327 	else {
328 		struct page *tmp;
329 		tmp = page_chain_tail(page, &i);
330 		spin_lock(&drbd_pp_lock);
331 		page_chain_add(&drbd_pp_pool, page, tmp);
332 		drbd_pp_vacant += i;
333 		spin_unlock(&drbd_pp_lock);
334 	}
335 	i = atomic_sub_return(i, a);
336 	if (i < 0)
337 		drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
338 			is_net ? "pp_in_use_by_net" : "pp_in_use", i);
339 	wake_up(&drbd_pp_wait);
340 }
341 
342 /*
343 You need to hold the req_lock:
344  _drbd_wait_ee_list_empty()
345 
346 You must not have the req_lock:
347  drbd_free_peer_req()
348  drbd_alloc_peer_req()
349  drbd_free_peer_reqs()
350  drbd_ee_fix_bhs()
351  drbd_finish_peer_reqs()
352  drbd_clear_done_ee()
353  drbd_wait_ee_list_empty()
354 */
355 
356 /* normal: payload_size == request size (bi_size)
357  * w_same: payload_size == logical_block_size
358  * trim: payload_size == 0 */
359 struct drbd_peer_request *
360 drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
361 		    unsigned int request_size, unsigned int payload_size, gfp_t gfp_mask) __must_hold(local)
362 {
363 	struct drbd_device *device = peer_device->device;
364 	struct drbd_peer_request *peer_req;
365 	struct page *page = NULL;
366 	unsigned int nr_pages = PFN_UP(payload_size);
367 
368 	if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
369 		return NULL;
370 
371 	peer_req = mempool_alloc(&drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
372 	if (!peer_req) {
373 		if (!(gfp_mask & __GFP_NOWARN))
374 			drbd_err(device, "%s: allocation failed\n", __func__);
375 		return NULL;
376 	}
377 
378 	if (nr_pages) {
379 		page = drbd_alloc_pages(peer_device, nr_pages,
380 					gfpflags_allow_blocking(gfp_mask));
381 		if (!page)
382 			goto fail;
383 	}
384 
385 	memset(peer_req, 0, sizeof(*peer_req));
386 	INIT_LIST_HEAD(&peer_req->w.list);
387 	drbd_clear_interval(&peer_req->i);
388 	peer_req->i.size = request_size;
389 	peer_req->i.sector = sector;
390 	peer_req->submit_jif = jiffies;
391 	peer_req->peer_device = peer_device;
392 	peer_req->pages = page;
393 	/*
394 	 * The block_id is opaque to the receiver.  It is not endianness
395 	 * converted, and sent back to the sender unchanged.
396 	 */
397 	peer_req->block_id = id;
398 
399 	return peer_req;
400 
401  fail:
402 	mempool_free(peer_req, &drbd_ee_mempool);
403 	return NULL;
404 }
405 
406 void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
407 		       int is_net)
408 {
409 	might_sleep();
410 	if (peer_req->flags & EE_HAS_DIGEST)
411 		kfree(peer_req->digest);
412 	drbd_free_pages(device, peer_req->pages, is_net);
413 	D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
414 	D_ASSERT(device, drbd_interval_empty(&peer_req->i));
415 	if (!expect(device, !(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
416 		peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
417 		drbd_al_complete_io(device, &peer_req->i);
418 	}
419 	mempool_free(peer_req, &drbd_ee_mempool);
420 }
421 
422 int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
423 {
424 	LIST_HEAD(work_list);
425 	struct drbd_peer_request *peer_req, *t;
426 	int count = 0;
427 	int is_net = list == &device->net_ee;
428 
429 	spin_lock_irq(&device->resource->req_lock);
430 	list_splice_init(list, &work_list);
431 	spin_unlock_irq(&device->resource->req_lock);
432 
433 	list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
434 		__drbd_free_peer_req(device, peer_req, is_net);
435 		count++;
436 	}
437 	return count;
438 }
439 
440 /*
441  * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
442  */
443 static int drbd_finish_peer_reqs(struct drbd_device *device)
444 {
445 	LIST_HEAD(work_list);
446 	LIST_HEAD(reclaimed);
447 	struct drbd_peer_request *peer_req, *t;
448 	int err = 0;
449 
450 	spin_lock_irq(&device->resource->req_lock);
451 	reclaim_finished_net_peer_reqs(device, &reclaimed);
452 	list_splice_init(&device->done_ee, &work_list);
453 	spin_unlock_irq(&device->resource->req_lock);
454 
455 	list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
456 		drbd_free_net_peer_req(device, peer_req);
457 
458 	/* possible callbacks here:
459 	 * e_end_block, and e_end_resync_block, e_send_superseded.
460 	 * all ignore the last argument.
461 	 */
462 	list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
463 		int err2;
464 
465 		/* list_del not necessary, next/prev members not touched */
466 		err2 = peer_req->w.cb(&peer_req->w, !!err);
467 		if (!err)
468 			err = err2;
469 		drbd_free_peer_req(device, peer_req);
470 	}
471 	wake_up(&device->ee_wait);
472 
473 	return err;
474 }
475 
476 static void _drbd_wait_ee_list_empty(struct drbd_device *device,
477 				     struct list_head *head)
478 {
479 	DEFINE_WAIT(wait);
480 
481 	/* avoids spin_lock/unlock
482 	 * and calling prepare_to_wait in the fast path */
483 	while (!list_empty(head)) {
484 		prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
485 		spin_unlock_irq(&device->resource->req_lock);
486 		io_schedule();
487 		finish_wait(&device->ee_wait, &wait);
488 		spin_lock_irq(&device->resource->req_lock);
489 	}
490 }
491 
492 static void drbd_wait_ee_list_empty(struct drbd_device *device,
493 				    struct list_head *head)
494 {
495 	spin_lock_irq(&device->resource->req_lock);
496 	_drbd_wait_ee_list_empty(device, head);
497 	spin_unlock_irq(&device->resource->req_lock);
498 }
499 
500 static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
501 {
502 	struct kvec iov = {
503 		.iov_base = buf,
504 		.iov_len = size,
505 	};
506 	struct msghdr msg = {
507 		.msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
508 	};
509 	iov_iter_kvec(&msg.msg_iter, ITER_DEST, &iov, 1, size);
510 	return sock_recvmsg(sock, &msg, msg.msg_flags);
511 }
512 
513 static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
514 {
515 	int rv;
516 
517 	rv = drbd_recv_short(connection->data.socket, buf, size, 0);
518 
519 	if (rv < 0) {
520 		if (rv == -ECONNRESET)
521 			drbd_info(connection, "sock was reset by peer\n");
522 		else if (rv != -ERESTARTSYS)
523 			drbd_err(connection, "sock_recvmsg returned %d\n", rv);
524 	} else if (rv == 0) {
525 		if (test_bit(DISCONNECT_SENT, &connection->flags)) {
526 			long t;
527 			rcu_read_lock();
528 			t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
529 			rcu_read_unlock();
530 
531 			t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);
532 
533 			if (t)
534 				goto out;
535 		}
536 		drbd_info(connection, "sock was shut down by peer\n");
537 	}
538 
539 	if (rv != size)
540 		conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
541 
542 out:
543 	return rv;
544 }
545 
546 static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
547 {
548 	int err;
549 
550 	err = drbd_recv(connection, buf, size);
551 	if (err != size) {
552 		if (err >= 0)
553 			err = -EIO;
554 	} else
555 		err = 0;
556 	return err;
557 }
558 
559 static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
560 {
561 	int err;
562 
563 	err = drbd_recv_all(connection, buf, size);
564 	if (err && !signal_pending(current))
565 		drbd_warn(connection, "short read (expected size %d)\n", (int)size);
566 	return err;
567 }
568 
569 /* quoting tcp(7):
570  *   On individual connections, the socket buffer size must be set prior to the
571  *   listen(2) or connect(2) calls in order to have it take effect.
572  * This is our wrapper to do so.
573  */
574 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
575 		unsigned int rcv)
576 {
577 	/* open coded SO_SNDBUF, SO_RCVBUF */
578 	if (snd) {
579 		sock->sk->sk_sndbuf = snd;
580 		sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
581 	}
582 	if (rcv) {
583 		sock->sk->sk_rcvbuf = rcv;
584 		sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
585 	}
586 }
587 
588 static struct socket *drbd_try_connect(struct drbd_connection *connection)
589 {
590 	const char *what;
591 	struct socket *sock;
592 	struct sockaddr_in6 src_in6;
593 	struct sockaddr_in6 peer_in6;
594 	struct net_conf *nc;
595 	int err, peer_addr_len, my_addr_len;
596 	int sndbuf_size, rcvbuf_size, connect_int;
597 	int disconnect_on_error = 1;
598 
599 	rcu_read_lock();
600 	nc = rcu_dereference(connection->net_conf);
601 	if (!nc) {
602 		rcu_read_unlock();
603 		return NULL;
604 	}
605 	sndbuf_size = nc->sndbuf_size;
606 	rcvbuf_size = nc->rcvbuf_size;
607 	connect_int = nc->connect_int;
608 	rcu_read_unlock();
609 
610 	my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
611 	memcpy(&src_in6, &connection->my_addr, my_addr_len);
612 
613 	if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
614 		src_in6.sin6_port = 0;
615 	else
616 		((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
617 
618 	peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
619 	memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);
620 
621 	what = "sock_create_kern";
622 	err = sock_create_kern(&init_net, ((struct sockaddr *)&src_in6)->sa_family,
623 			       SOCK_STREAM, IPPROTO_TCP, &sock);
624 	if (err < 0) {
625 		sock = NULL;
626 		goto out;
627 	}
628 
629 	sock->sk->sk_rcvtimeo =
630 	sock->sk->sk_sndtimeo = connect_int * HZ;
631 	drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);
632 
633        /* explicitly bind to the configured IP as source IP
634 	*  for the outgoing connections.
635 	*  This is needed for multihomed hosts and to be
636 	*  able to use lo: interfaces for drbd.
637 	* Make sure to use 0 as port number, so linux selects
638 	*  a free one dynamically.
639 	*/
640 	what = "bind before connect";
641 	err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
642 	if (err < 0)
643 		goto out;
644 
645 	/* connect may fail, peer not yet available.
646 	 * stay C_WF_CONNECTION, don't go Disconnecting! */
647 	disconnect_on_error = 0;
648 	what = "connect";
649 	err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);
650 
651 out:
652 	if (err < 0) {
653 		if (sock) {
654 			sock_release(sock);
655 			sock = NULL;
656 		}
657 		switch (-err) {
658 			/* timeout, busy, signal pending */
659 		case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
660 		case EINTR: case ERESTARTSYS:
661 			/* peer not (yet) available, network problem */
662 		case ECONNREFUSED: case ENETUNREACH:
663 		case EHOSTDOWN:    case EHOSTUNREACH:
664 			disconnect_on_error = 0;
665 			break;
666 		default:
667 			drbd_err(connection, "%s failed, err = %d\n", what, err);
668 		}
669 		if (disconnect_on_error)
670 			conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
671 	}
672 
673 	return sock;
674 }
675 
676 struct accept_wait_data {
677 	struct drbd_connection *connection;
678 	struct socket *s_listen;
679 	struct completion door_bell;
680 	void (*original_sk_state_change)(struct sock *sk);
681 
682 };
683 
684 static void drbd_incoming_connection(struct sock *sk)
685 {
686 	struct accept_wait_data *ad = sk->sk_user_data;
687 	void (*state_change)(struct sock *sk);
688 
689 	state_change = ad->original_sk_state_change;
690 	if (sk->sk_state == TCP_ESTABLISHED)
691 		complete(&ad->door_bell);
692 	state_change(sk);
693 }
694 
695 static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
696 {
697 	int err, sndbuf_size, rcvbuf_size, my_addr_len;
698 	struct sockaddr_in6 my_addr;
699 	struct socket *s_listen;
700 	struct net_conf *nc;
701 	const char *what;
702 
703 	rcu_read_lock();
704 	nc = rcu_dereference(connection->net_conf);
705 	if (!nc) {
706 		rcu_read_unlock();
707 		return -EIO;
708 	}
709 	sndbuf_size = nc->sndbuf_size;
710 	rcvbuf_size = nc->rcvbuf_size;
711 	rcu_read_unlock();
712 
713 	my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
714 	memcpy(&my_addr, &connection->my_addr, my_addr_len);
715 
716 	what = "sock_create_kern";
717 	err = sock_create_kern(&init_net, ((struct sockaddr *)&my_addr)->sa_family,
718 			       SOCK_STREAM, IPPROTO_TCP, &s_listen);
719 	if (err) {
720 		s_listen = NULL;
721 		goto out;
722 	}
723 
724 	s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
725 	drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);
726 
727 	what = "bind before listen";
728 	err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
729 	if (err < 0)
730 		goto out;
731 
732 	ad->s_listen = s_listen;
733 	write_lock_bh(&s_listen->sk->sk_callback_lock);
734 	ad->original_sk_state_change = s_listen->sk->sk_state_change;
735 	s_listen->sk->sk_state_change = drbd_incoming_connection;
736 	s_listen->sk->sk_user_data = ad;
737 	write_unlock_bh(&s_listen->sk->sk_callback_lock);
738 
739 	what = "listen";
740 	err = s_listen->ops->listen(s_listen, 5);
741 	if (err < 0)
742 		goto out;
743 
744 	return 0;
745 out:
746 	if (s_listen)
747 		sock_release(s_listen);
748 	if (err < 0) {
749 		if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
750 			drbd_err(connection, "%s failed, err = %d\n", what, err);
751 			conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
752 		}
753 	}
754 
755 	return -EIO;
756 }
757 
758 static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
759 {
760 	write_lock_bh(&sk->sk_callback_lock);
761 	sk->sk_state_change = ad->original_sk_state_change;
762 	sk->sk_user_data = NULL;
763 	write_unlock_bh(&sk->sk_callback_lock);
764 }
765 
766 static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
767 {
768 	int timeo, connect_int, err = 0;
769 	struct socket *s_estab = NULL;
770 	struct net_conf *nc;
771 
772 	rcu_read_lock();
773 	nc = rcu_dereference(connection->net_conf);
774 	if (!nc) {
775 		rcu_read_unlock();
776 		return NULL;
777 	}
778 	connect_int = nc->connect_int;
779 	rcu_read_unlock();
780 
781 	timeo = connect_int * HZ;
782 	/* 28.5% random jitter */
783 	timeo += get_random_u32_below(2) ? timeo / 7 : -timeo / 7;
784 
785 	err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
786 	if (err <= 0)
787 		return NULL;
788 
789 	err = kernel_accept(ad->s_listen, &s_estab, 0);
790 	if (err < 0) {
791 		if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
792 			drbd_err(connection, "accept failed, err = %d\n", err);
793 			conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
794 		}
795 	}
796 
797 	if (s_estab)
798 		unregister_state_change(s_estab->sk, ad);
799 
800 	return s_estab;
801 }
802 
803 static int decode_header(struct drbd_connection *, void *, struct packet_info *);
804 
805 static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
806 			     enum drbd_packet cmd)
807 {
808 	if (!conn_prepare_command(connection, sock))
809 		return -EIO;
810 	return conn_send_command(connection, sock, cmd, 0, NULL, 0);
811 }
812 
813 static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
814 {
815 	unsigned int header_size = drbd_header_size(connection);
816 	struct packet_info pi;
817 	struct net_conf *nc;
818 	int err;
819 
820 	rcu_read_lock();
821 	nc = rcu_dereference(connection->net_conf);
822 	if (!nc) {
823 		rcu_read_unlock();
824 		return -EIO;
825 	}
826 	sock->sk->sk_rcvtimeo = nc->ping_timeo * 4 * HZ / 10;
827 	rcu_read_unlock();
828 
829 	err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
830 	if (err != header_size) {
831 		if (err >= 0)
832 			err = -EIO;
833 		return err;
834 	}
835 	err = decode_header(connection, connection->data.rbuf, &pi);
836 	if (err)
837 		return err;
838 	return pi.cmd;
839 }
840 
841 /**
842  * drbd_socket_okay() - Free the socket if its connection is not okay
843  * @sock:	pointer to the pointer to the socket.
844  */
845 static bool drbd_socket_okay(struct socket **sock)
846 {
847 	int rr;
848 	char tb[4];
849 
850 	if (!*sock)
851 		return false;
852 
853 	rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
854 
855 	if (rr > 0 || rr == -EAGAIN) {
856 		return true;
857 	} else {
858 		sock_release(*sock);
859 		*sock = NULL;
860 		return false;
861 	}
862 }
863 
864 static bool connection_established(struct drbd_connection *connection,
865 				   struct socket **sock1,
866 				   struct socket **sock2)
867 {
868 	struct net_conf *nc;
869 	int timeout;
870 	bool ok;
871 
872 	if (!*sock1 || !*sock2)
873 		return false;
874 
875 	rcu_read_lock();
876 	nc = rcu_dereference(connection->net_conf);
877 	timeout = (nc->sock_check_timeo ?: nc->ping_timeo) * HZ / 10;
878 	rcu_read_unlock();
879 	schedule_timeout_interruptible(timeout);
880 
881 	ok = drbd_socket_okay(sock1);
882 	ok = drbd_socket_okay(sock2) && ok;
883 
884 	return ok;
885 }
886 
887 /* Gets called if a connection is established, or if a new minor gets created
888    in a connection */
889 int drbd_connected(struct drbd_peer_device *peer_device)
890 {
891 	struct drbd_device *device = peer_device->device;
892 	int err;
893 
894 	atomic_set(&device->packet_seq, 0);
895 	device->peer_seq = 0;
896 
897 	device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
898 		&peer_device->connection->cstate_mutex :
899 		&device->own_state_mutex;
900 
901 	err = drbd_send_sync_param(peer_device);
902 	if (!err)
903 		err = drbd_send_sizes(peer_device, 0, 0);
904 	if (!err)
905 		err = drbd_send_uuids(peer_device);
906 	if (!err)
907 		err = drbd_send_current_state(peer_device);
908 	clear_bit(USE_DEGR_WFC_T, &device->flags);
909 	clear_bit(RESIZE_PENDING, &device->flags);
910 	atomic_set(&device->ap_in_flight, 0);
911 	mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
912 	return err;
913 }
914 
915 /*
916  * return values:
917  *   1 yes, we have a valid connection
918  *   0 oops, did not work out, please try again
919  *  -1 peer talks different language,
920  *     no point in trying again, please go standalone.
921  *  -2 We do not have a network config...
922  */
923 static int conn_connect(struct drbd_connection *connection)
924 {
925 	struct drbd_socket sock, msock;
926 	struct drbd_peer_device *peer_device;
927 	struct net_conf *nc;
928 	int vnr, timeout, h;
929 	bool discard_my_data, ok;
930 	enum drbd_state_rv rv;
931 	struct accept_wait_data ad = {
932 		.connection = connection,
933 		.door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
934 	};
935 
936 	clear_bit(DISCONNECT_SENT, &connection->flags);
937 	if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
938 		return -2;
939 
940 	mutex_init(&sock.mutex);
941 	sock.sbuf = connection->data.sbuf;
942 	sock.rbuf = connection->data.rbuf;
943 	sock.socket = NULL;
944 	mutex_init(&msock.mutex);
945 	msock.sbuf = connection->meta.sbuf;
946 	msock.rbuf = connection->meta.rbuf;
947 	msock.socket = NULL;
948 
949 	/* Assume that the peer only understands protocol 80 until we know better.  */
950 	connection->agreed_pro_version = 80;
951 
952 	if (prepare_listen_socket(connection, &ad))
953 		return 0;
954 
955 	do {
956 		struct socket *s;
957 
958 		s = drbd_try_connect(connection);
959 		if (s) {
960 			if (!sock.socket) {
961 				sock.socket = s;
962 				send_first_packet(connection, &sock, P_INITIAL_DATA);
963 			} else if (!msock.socket) {
964 				clear_bit(RESOLVE_CONFLICTS, &connection->flags);
965 				msock.socket = s;
966 				send_first_packet(connection, &msock, P_INITIAL_META);
967 			} else {
968 				drbd_err(connection, "Logic error in conn_connect()\n");
969 				goto out_release_sockets;
970 			}
971 		}
972 
973 		if (connection_established(connection, &sock.socket, &msock.socket))
974 			break;
975 
976 retry:
977 		s = drbd_wait_for_connect(connection, &ad);
978 		if (s) {
979 			int fp = receive_first_packet(connection, s);
980 			drbd_socket_okay(&sock.socket);
981 			drbd_socket_okay(&msock.socket);
982 			switch (fp) {
983 			case P_INITIAL_DATA:
984 				if (sock.socket) {
985 					drbd_warn(connection, "initial packet S crossed\n");
986 					sock_release(sock.socket);
987 					sock.socket = s;
988 					goto randomize;
989 				}
990 				sock.socket = s;
991 				break;
992 			case P_INITIAL_META:
993 				set_bit(RESOLVE_CONFLICTS, &connection->flags);
994 				if (msock.socket) {
995 					drbd_warn(connection, "initial packet M crossed\n");
996 					sock_release(msock.socket);
997 					msock.socket = s;
998 					goto randomize;
999 				}
1000 				msock.socket = s;
1001 				break;
1002 			default:
1003 				drbd_warn(connection, "Error receiving initial packet\n");
1004 				sock_release(s);
1005 randomize:
1006 				if (get_random_u32_below(2))
1007 					goto retry;
1008 			}
1009 		}
1010 
1011 		if (connection->cstate <= C_DISCONNECTING)
1012 			goto out_release_sockets;
1013 		if (signal_pending(current)) {
1014 			flush_signals(current);
1015 			smp_rmb();
1016 			if (get_t_state(&connection->receiver) == EXITING)
1017 				goto out_release_sockets;
1018 		}
1019 
1020 		ok = connection_established(connection, &sock.socket, &msock.socket);
1021 	} while (!ok);
1022 
1023 	if (ad.s_listen)
1024 		sock_release(ad.s_listen);
1025 
1026 	sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1027 	msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1028 
1029 	sock.socket->sk->sk_allocation = GFP_NOIO;
1030 	msock.socket->sk->sk_allocation = GFP_NOIO;
1031 
1032 	sock.socket->sk->sk_use_task_frag = false;
1033 	msock.socket->sk->sk_use_task_frag = false;
1034 
1035 	sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
1036 	msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
1037 
1038 	/* NOT YET ...
1039 	 * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
1040 	 * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1041 	 * first set it to the P_CONNECTION_FEATURES timeout,
1042 	 * which we set to 4x the configured ping_timeout. */
1043 	rcu_read_lock();
1044 	nc = rcu_dereference(connection->net_conf);
1045 
1046 	sock.socket->sk->sk_sndtimeo =
1047 	sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
1048 
1049 	msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
1050 	timeout = nc->timeout * HZ / 10;
1051 	discard_my_data = nc->discard_my_data;
1052 	rcu_read_unlock();
1053 
1054 	msock.socket->sk->sk_sndtimeo = timeout;
1055 
1056 	/* we don't want delays.
1057 	 * we use TCP_CORK where appropriate, though */
1058 	tcp_sock_set_nodelay(sock.socket->sk);
1059 	tcp_sock_set_nodelay(msock.socket->sk);
1060 
1061 	connection->data.socket = sock.socket;
1062 	connection->meta.socket = msock.socket;
1063 	connection->last_received = jiffies;
1064 
1065 	h = drbd_do_features(connection);
1066 	if (h <= 0)
1067 		return h;
1068 
1069 	if (connection->cram_hmac_tfm) {
1070 		/* drbd_request_state(device, NS(conn, WFAuth)); */
1071 		switch (drbd_do_auth(connection)) {
1072 		case -1:
1073 			drbd_err(connection, "Authentication of peer failed\n");
1074 			return -1;
1075 		case 0:
1076 			drbd_err(connection, "Authentication of peer failed, trying again.\n");
1077 			return 0;
1078 		}
1079 	}
1080 
1081 	connection->data.socket->sk->sk_sndtimeo = timeout;
1082 	connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1083 
1084 	if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1085 		return -1;
1086 
1087 	/* Prevent a race between resync-handshake and
1088 	 * being promoted to Primary.
1089 	 *
1090 	 * Grab and release the state mutex, so we know that any current
1091 	 * drbd_set_role() is finished, and any incoming drbd_set_role
1092 	 * will see the STATE_SENT flag, and wait for it to be cleared.
1093 	 */
1094 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1095 		mutex_lock(peer_device->device->state_mutex);
1096 
1097 	/* avoid a race with conn_request_state( C_DISCONNECTING ) */
1098 	spin_lock_irq(&connection->resource->req_lock);
1099 	set_bit(STATE_SENT, &connection->flags);
1100 	spin_unlock_irq(&connection->resource->req_lock);
1101 
1102 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1103 		mutex_unlock(peer_device->device->state_mutex);
1104 
1105 	rcu_read_lock();
1106 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1107 		struct drbd_device *device = peer_device->device;
1108 		kref_get(&device->kref);
1109 		rcu_read_unlock();
1110 
1111 		if (discard_my_data)
1112 			set_bit(DISCARD_MY_DATA, &device->flags);
1113 		else
1114 			clear_bit(DISCARD_MY_DATA, &device->flags);
1115 
1116 		drbd_connected(peer_device);
1117 		kref_put(&device->kref, drbd_destroy_device);
1118 		rcu_read_lock();
1119 	}
1120 	rcu_read_unlock();
1121 
1122 	rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1123 	if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1124 		clear_bit(STATE_SENT, &connection->flags);
1125 		return 0;
1126 	}
1127 
1128 	drbd_thread_start(&connection->ack_receiver);
1129 	/* opencoded create_singlethread_workqueue(),
1130 	 * to be able to use format string arguments */
1131 	connection->ack_sender =
1132 		alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
1133 	if (!connection->ack_sender) {
1134 		drbd_err(connection, "Failed to create workqueue ack_sender\n");
1135 		return 0;
1136 	}
1137 
1138 	mutex_lock(&connection->resource->conf_update);
1139 	/* The discard_my_data flag is a single-shot modifier to the next
1140 	 * connection attempt, the handshake of which is now well underway.
1141 	 * No need for rcu style copying of the whole struct
1142 	 * just to clear a single value. */
1143 	connection->net_conf->discard_my_data = 0;
1144 	mutex_unlock(&connection->resource->conf_update);
1145 
1146 	return h;
1147 
1148 out_release_sockets:
1149 	if (ad.s_listen)
1150 		sock_release(ad.s_listen);
1151 	if (sock.socket)
1152 		sock_release(sock.socket);
1153 	if (msock.socket)
1154 		sock_release(msock.socket);
1155 	return -1;
1156 }
1157 
1158 static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1159 {
1160 	unsigned int header_size = drbd_header_size(connection);
1161 
1162 	if (header_size == sizeof(struct p_header100) &&
1163 	    *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1164 		struct p_header100 *h = header;
1165 		if (h->pad != 0) {
1166 			drbd_err(connection, "Header padding is not zero\n");
1167 			return -EINVAL;
1168 		}
1169 		pi->vnr = be16_to_cpu(h->volume);
1170 		pi->cmd = be16_to_cpu(h->command);
1171 		pi->size = be32_to_cpu(h->length);
1172 	} else if (header_size == sizeof(struct p_header95) &&
1173 		   *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1174 		struct p_header95 *h = header;
1175 		pi->cmd = be16_to_cpu(h->command);
1176 		pi->size = be32_to_cpu(h->length);
1177 		pi->vnr = 0;
1178 	} else if (header_size == sizeof(struct p_header80) &&
1179 		   *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1180 		struct p_header80 *h = header;
1181 		pi->cmd = be16_to_cpu(h->command);
1182 		pi->size = be16_to_cpu(h->length);
1183 		pi->vnr = 0;
1184 	} else {
1185 		drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1186 			 be32_to_cpu(*(__be32 *)header),
1187 			 connection->agreed_pro_version);
1188 		return -EINVAL;
1189 	}
1190 	pi->data = header + header_size;
1191 	return 0;
1192 }
1193 
1194 static void drbd_unplug_all_devices(struct drbd_connection *connection)
1195 {
1196 	if (current->plug == &connection->receiver_plug) {
1197 		blk_finish_plug(&connection->receiver_plug);
1198 		blk_start_plug(&connection->receiver_plug);
1199 	} /* else: maybe just schedule() ?? */
1200 }
1201 
1202 static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1203 {
1204 	void *buffer = connection->data.rbuf;
1205 	int err;
1206 
1207 	err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1208 	if (err)
1209 		return err;
1210 
1211 	err = decode_header(connection, buffer, pi);
1212 	connection->last_received = jiffies;
1213 
1214 	return err;
1215 }
1216 
1217 static int drbd_recv_header_maybe_unplug(struct drbd_connection *connection, struct packet_info *pi)
1218 {
1219 	void *buffer = connection->data.rbuf;
1220 	unsigned int size = drbd_header_size(connection);
1221 	int err;
1222 
1223 	err = drbd_recv_short(connection->data.socket, buffer, size, MSG_NOSIGNAL|MSG_DONTWAIT);
1224 	if (err != size) {
1225 		/* If we have nothing in the receive buffer now, to reduce
1226 		 * application latency, try to drain the backend queues as
1227 		 * quickly as possible, and let remote TCP know what we have
1228 		 * received so far. */
1229 		if (err == -EAGAIN) {
1230 			tcp_sock_set_quickack(connection->data.socket->sk, 2);
1231 			drbd_unplug_all_devices(connection);
1232 		}
1233 		if (err > 0) {
1234 			buffer += err;
1235 			size -= err;
1236 		}
1237 		err = drbd_recv_all_warn(connection, buffer, size);
1238 		if (err)
1239 			return err;
1240 	}
1241 
1242 	err = decode_header(connection, connection->data.rbuf, pi);
1243 	connection->last_received = jiffies;
1244 
1245 	return err;
1246 }
1247 /* This is blkdev_issue_flush, but asynchronous.
1248  * We want to submit to all component volumes in parallel,
1249  * then wait for all completions.
1250  */
1251 struct issue_flush_context {
1252 	atomic_t pending;
1253 	int error;
1254 	struct completion done;
1255 };
1256 struct one_flush_context {
1257 	struct drbd_device *device;
1258 	struct issue_flush_context *ctx;
1259 };
1260 
1261 static void one_flush_endio(struct bio *bio)
1262 {
1263 	struct one_flush_context *octx = bio->bi_private;
1264 	struct drbd_device *device = octx->device;
1265 	struct issue_flush_context *ctx = octx->ctx;
1266 
1267 	if (bio->bi_status) {
1268 		ctx->error = blk_status_to_errno(bio->bi_status);
1269 		drbd_info(device, "local disk FLUSH FAILED with status %d\n", bio->bi_status);
1270 	}
1271 	kfree(octx);
1272 	bio_put(bio);
1273 
1274 	clear_bit(FLUSH_PENDING, &device->flags);
1275 	put_ldev(device);
1276 	kref_put(&device->kref, drbd_destroy_device);
1277 
1278 	if (atomic_dec_and_test(&ctx->pending))
1279 		complete(&ctx->done);
1280 }
1281 
1282 static void submit_one_flush(struct drbd_device *device, struct issue_flush_context *ctx)
1283 {
1284 	struct bio *bio = bio_alloc(device->ldev->backing_bdev, 0,
1285 				    REQ_OP_WRITE | REQ_PREFLUSH, GFP_NOIO);
1286 	struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);
1287 
1288 	if (!octx) {
1289 		drbd_warn(device, "Could not allocate a octx, CANNOT ISSUE FLUSH\n");
1290 		/* FIXME: what else can I do now?  disconnecting or detaching
1291 		 * really does not help to improve the state of the world, either.
1292 		 */
1293 		bio_put(bio);
1294 
1295 		ctx->error = -ENOMEM;
1296 		put_ldev(device);
1297 		kref_put(&device->kref, drbd_destroy_device);
1298 		return;
1299 	}
1300 
1301 	octx->device = device;
1302 	octx->ctx = ctx;
1303 	bio->bi_private = octx;
1304 	bio->bi_end_io = one_flush_endio;
1305 
1306 	device->flush_jif = jiffies;
1307 	set_bit(FLUSH_PENDING, &device->flags);
1308 	atomic_inc(&ctx->pending);
1309 	submit_bio(bio);
1310 }
1311 
1312 static void drbd_flush(struct drbd_connection *connection)
1313 {
1314 	if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1315 		struct drbd_peer_device *peer_device;
1316 		struct issue_flush_context ctx;
1317 		int vnr;
1318 
1319 		atomic_set(&ctx.pending, 1);
1320 		ctx.error = 0;
1321 		init_completion(&ctx.done);
1322 
1323 		rcu_read_lock();
1324 		idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1325 			struct drbd_device *device = peer_device->device;
1326 
1327 			if (!get_ldev(device))
1328 				continue;
1329 			kref_get(&device->kref);
1330 			rcu_read_unlock();
1331 
1332 			submit_one_flush(device, &ctx);
1333 
1334 			rcu_read_lock();
1335 		}
1336 		rcu_read_unlock();
1337 
1338 		/* Do we want to add a timeout,
1339 		 * if disk-timeout is set? */
1340 		if (!atomic_dec_and_test(&ctx.pending))
1341 			wait_for_completion(&ctx.done);
1342 
1343 		if (ctx.error) {
1344 			/* would rather check on EOPNOTSUPP, but that is not reliable.
1345 			 * don't try again for ANY return value != 0
1346 			 * if (rv == -EOPNOTSUPP) */
1347 			/* Any error is already reported by bio_endio callback. */
1348 			drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1349 		}
1350 	}
1351 }
1352 
1353 /**
1354  * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1355  * @connection:	DRBD connection.
1356  * @epoch:	Epoch object.
1357  * @ev:		Epoch event.
1358  */
1359 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1360 					       struct drbd_epoch *epoch,
1361 					       enum epoch_event ev)
1362 {
1363 	int epoch_size;
1364 	struct drbd_epoch *next_epoch;
1365 	enum finish_epoch rv = FE_STILL_LIVE;
1366 
1367 	spin_lock(&connection->epoch_lock);
1368 	do {
1369 		next_epoch = NULL;
1370 
1371 		epoch_size = atomic_read(&epoch->epoch_size);
1372 
1373 		switch (ev & ~EV_CLEANUP) {
1374 		case EV_PUT:
1375 			atomic_dec(&epoch->active);
1376 			break;
1377 		case EV_GOT_BARRIER_NR:
1378 			set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1379 			break;
1380 		case EV_BECAME_LAST:
1381 			/* nothing to do*/
1382 			break;
1383 		}
1384 
1385 		if (epoch_size != 0 &&
1386 		    atomic_read(&epoch->active) == 0 &&
1387 		    (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1388 			if (!(ev & EV_CLEANUP)) {
1389 				spin_unlock(&connection->epoch_lock);
1390 				drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1391 				spin_lock(&connection->epoch_lock);
1392 			}
1393 #if 0
1394 			/* FIXME: dec unacked on connection, once we have
1395 			 * something to count pending connection packets in. */
1396 			if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1397 				dec_unacked(epoch->connection);
1398 #endif
1399 
1400 			if (connection->current_epoch != epoch) {
1401 				next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1402 				list_del(&epoch->list);
1403 				ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1404 				connection->epochs--;
1405 				kfree(epoch);
1406 
1407 				if (rv == FE_STILL_LIVE)
1408 					rv = FE_DESTROYED;
1409 			} else {
1410 				epoch->flags = 0;
1411 				atomic_set(&epoch->epoch_size, 0);
1412 				/* atomic_set(&epoch->active, 0); is already zero */
1413 				if (rv == FE_STILL_LIVE)
1414 					rv = FE_RECYCLED;
1415 			}
1416 		}
1417 
1418 		if (!next_epoch)
1419 			break;
1420 
1421 		epoch = next_epoch;
1422 	} while (1);
1423 
1424 	spin_unlock(&connection->epoch_lock);
1425 
1426 	return rv;
1427 }
1428 
1429 static enum write_ordering_e
1430 max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1431 {
1432 	struct disk_conf *dc;
1433 
1434 	dc = rcu_dereference(bdev->disk_conf);
1435 
1436 	if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1437 		wo = WO_DRAIN_IO;
1438 	if (wo == WO_DRAIN_IO && !dc->disk_drain)
1439 		wo = WO_NONE;
1440 
1441 	return wo;
1442 }
1443 
1444 /*
1445  * drbd_bump_write_ordering() - Fall back to an other write ordering method
1446  * @wo:		Write ordering method to try.
1447  */
1448 void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1449 			      enum write_ordering_e wo)
1450 {
1451 	struct drbd_device *device;
1452 	enum write_ordering_e pwo;
1453 	int vnr;
1454 	static char *write_ordering_str[] = {
1455 		[WO_NONE] = "none",
1456 		[WO_DRAIN_IO] = "drain",
1457 		[WO_BDEV_FLUSH] = "flush",
1458 	};
1459 
1460 	pwo = resource->write_ordering;
1461 	if (wo != WO_BDEV_FLUSH)
1462 		wo = min(pwo, wo);
1463 	rcu_read_lock();
1464 	idr_for_each_entry(&resource->devices, device, vnr) {
1465 		if (get_ldev(device)) {
1466 			wo = max_allowed_wo(device->ldev, wo);
1467 			if (device->ldev == bdev)
1468 				bdev = NULL;
1469 			put_ldev(device);
1470 		}
1471 	}
1472 
1473 	if (bdev)
1474 		wo = max_allowed_wo(bdev, wo);
1475 
1476 	rcu_read_unlock();
1477 
1478 	resource->write_ordering = wo;
1479 	if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1480 		drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1481 }
1482 
1483 /*
1484  * Mapping "discard" to ZEROOUT with UNMAP does not work for us:
1485  * Drivers have to "announce" q->limits.max_write_zeroes_sectors, or it
1486  * will directly go to fallback mode, submitting normal writes, and
1487  * never even try to UNMAP.
1488  *
1489  * And dm-thin does not do this (yet), mostly because in general it has
1490  * to assume that "skip_block_zeroing" is set.  See also:
1491  * https://www.mail-archive.com/dm-devel%40redhat.com/msg07965.html
1492  * https://www.redhat.com/archives/dm-devel/2018-January/msg00271.html
1493  *
1494  * We *may* ignore the discard-zeroes-data setting, if so configured.
1495  *
1496  * Assumption is that this "discard_zeroes_data=0" is only because the backend
1497  * may ignore partial unaligned discards.
1498  *
1499  * LVM/DM thin as of at least
1500  *   LVM version:     2.02.115(2)-RHEL7 (2015-01-28)
1501  *   Library version: 1.02.93-RHEL7 (2015-01-28)
1502  *   Driver version:  4.29.0
1503  * still behaves this way.
1504  *
1505  * For unaligned (wrt. alignment and granularity) or too small discards,
1506  * we zero-out the initial (and/or) trailing unaligned partial chunks,
1507  * but discard all the aligned full chunks.
1508  *
1509  * At least for LVM/DM thin, with skip_block_zeroing=false,
1510  * the result is effectively "discard_zeroes_data=1".
1511  */
1512 /* flags: EE_TRIM|EE_ZEROOUT */
1513 int drbd_issue_discard_or_zero_out(struct drbd_device *device, sector_t start, unsigned int nr_sectors, int flags)
1514 {
1515 	struct block_device *bdev = device->ldev->backing_bdev;
1516 	sector_t tmp, nr;
1517 	unsigned int max_discard_sectors, granularity;
1518 	int alignment;
1519 	int err = 0;
1520 
1521 	if ((flags & EE_ZEROOUT) || !(flags & EE_TRIM))
1522 		goto zero_out;
1523 
1524 	/* Zero-sector (unknown) and one-sector granularities are the same.  */
1525 	granularity = max(bdev_discard_granularity(bdev) >> 9, 1U);
1526 	alignment = (bdev_discard_alignment(bdev) >> 9) % granularity;
1527 
1528 	max_discard_sectors = min(bdev_max_discard_sectors(bdev), (1U << 22));
1529 	max_discard_sectors -= max_discard_sectors % granularity;
1530 	if (unlikely(!max_discard_sectors))
1531 		goto zero_out;
1532 
1533 	if (nr_sectors < granularity)
1534 		goto zero_out;
1535 
1536 	tmp = start;
1537 	if (sector_div(tmp, granularity) != alignment) {
1538 		if (nr_sectors < 2*granularity)
1539 			goto zero_out;
1540 		/* start + gran - (start + gran - align) % gran */
1541 		tmp = start + granularity - alignment;
1542 		tmp = start + granularity - sector_div(tmp, granularity);
1543 
1544 		nr = tmp - start;
1545 		/* don't flag BLKDEV_ZERO_NOUNMAP, we don't know how many
1546 		 * layers are below us, some may have smaller granularity */
1547 		err |= blkdev_issue_zeroout(bdev, start, nr, GFP_NOIO, 0);
1548 		nr_sectors -= nr;
1549 		start = tmp;
1550 	}
1551 	while (nr_sectors >= max_discard_sectors) {
1552 		err |= blkdev_issue_discard(bdev, start, max_discard_sectors,
1553 					    GFP_NOIO);
1554 		nr_sectors -= max_discard_sectors;
1555 		start += max_discard_sectors;
1556 	}
1557 	if (nr_sectors) {
1558 		/* max_discard_sectors is unsigned int (and a multiple of
1559 		 * granularity, we made sure of that above already);
1560 		 * nr is < max_discard_sectors;
1561 		 * I don't need sector_div here, even though nr is sector_t */
1562 		nr = nr_sectors;
1563 		nr -= (unsigned int)nr % granularity;
1564 		if (nr) {
1565 			err |= blkdev_issue_discard(bdev, start, nr, GFP_NOIO);
1566 			nr_sectors -= nr;
1567 			start += nr;
1568 		}
1569 	}
1570  zero_out:
1571 	if (nr_sectors) {
1572 		err |= blkdev_issue_zeroout(bdev, start, nr_sectors, GFP_NOIO,
1573 				(flags & EE_TRIM) ? 0 : BLKDEV_ZERO_NOUNMAP);
1574 	}
1575 	return err != 0;
1576 }
1577 
1578 static bool can_do_reliable_discards(struct drbd_device *device)
1579 {
1580 	struct disk_conf *dc;
1581 	bool can_do;
1582 
1583 	if (!bdev_max_discard_sectors(device->ldev->backing_bdev))
1584 		return false;
1585 
1586 	rcu_read_lock();
1587 	dc = rcu_dereference(device->ldev->disk_conf);
1588 	can_do = dc->discard_zeroes_if_aligned;
1589 	rcu_read_unlock();
1590 	return can_do;
1591 }
1592 
1593 static void drbd_issue_peer_discard_or_zero_out(struct drbd_device *device, struct drbd_peer_request *peer_req)
1594 {
1595 	/* If the backend cannot discard, or does not guarantee
1596 	 * read-back zeroes in discarded ranges, we fall back to
1597 	 * zero-out.  Unless configuration specifically requested
1598 	 * otherwise. */
1599 	if (!can_do_reliable_discards(device))
1600 		peer_req->flags |= EE_ZEROOUT;
1601 
1602 	if (drbd_issue_discard_or_zero_out(device, peer_req->i.sector,
1603 	    peer_req->i.size >> 9, peer_req->flags & (EE_ZEROOUT|EE_TRIM)))
1604 		peer_req->flags |= EE_WAS_ERROR;
1605 	drbd_endio_write_sec_final(peer_req);
1606 }
1607 
1608 static int peer_request_fault_type(struct drbd_peer_request *peer_req)
1609 {
1610 	if (peer_req_op(peer_req) == REQ_OP_READ) {
1611 		return peer_req->flags & EE_APPLICATION ?
1612 			DRBD_FAULT_DT_RD : DRBD_FAULT_RS_RD;
1613 	} else {
1614 		return peer_req->flags & EE_APPLICATION ?
1615 			DRBD_FAULT_DT_WR : DRBD_FAULT_RS_WR;
1616 	}
1617 }
1618 
1619 /**
1620  * drbd_submit_peer_request()
1621  * @peer_req:	peer request
1622  *
1623  * May spread the pages to multiple bios,
1624  * depending on bio_add_page restrictions.
1625  *
1626  * Returns 0 if all bios have been submitted,
1627  * -ENOMEM if we could not allocate enough bios,
1628  * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1629  *  single page to an empty bio (which should never happen and likely indicates
1630  *  that the lower level IO stack is in some way broken). This has been observed
1631  *  on certain Xen deployments.
1632  */
1633 /* TODO allocate from our own bio_set. */
1634 int drbd_submit_peer_request(struct drbd_peer_request *peer_req)
1635 {
1636 	struct drbd_device *device = peer_req->peer_device->device;
1637 	struct bio *bios = NULL;
1638 	struct bio *bio;
1639 	struct page *page = peer_req->pages;
1640 	sector_t sector = peer_req->i.sector;
1641 	unsigned int data_size = peer_req->i.size;
1642 	unsigned int n_bios = 0;
1643 	unsigned int nr_pages = PFN_UP(data_size);
1644 
1645 	/* TRIM/DISCARD: for now, always use the helper function
1646 	 * blkdev_issue_zeroout(..., discard=true).
1647 	 * It's synchronous, but it does the right thing wrt. bio splitting.
1648 	 * Correctness first, performance later.  Next step is to code an
1649 	 * asynchronous variant of the same.
1650 	 */
1651 	if (peer_req->flags & (EE_TRIM | EE_ZEROOUT)) {
1652 		/* wait for all pending IO completions, before we start
1653 		 * zeroing things out. */
1654 		conn_wait_active_ee_empty(peer_req->peer_device->connection);
1655 		/* add it to the active list now,
1656 		 * so we can find it to present it in debugfs */
1657 		peer_req->submit_jif = jiffies;
1658 		peer_req->flags |= EE_SUBMITTED;
1659 
1660 		/* If this was a resync request from receive_rs_deallocated(),
1661 		 * it is already on the sync_ee list */
1662 		if (list_empty(&peer_req->w.list)) {
1663 			spin_lock_irq(&device->resource->req_lock);
1664 			list_add_tail(&peer_req->w.list, &device->active_ee);
1665 			spin_unlock_irq(&device->resource->req_lock);
1666 		}
1667 
1668 		drbd_issue_peer_discard_or_zero_out(device, peer_req);
1669 		return 0;
1670 	}
1671 
1672 	/* In most cases, we will only need one bio.  But in case the lower
1673 	 * level restrictions happen to be different at this offset on this
1674 	 * side than those of the sending peer, we may need to submit the
1675 	 * request in more than one bio.
1676 	 *
1677 	 * Plain bio_alloc is good enough here, this is no DRBD internally
1678 	 * generated bio, but a bio allocated on behalf of the peer.
1679 	 */
1680 next_bio:
1681 	/* _DISCARD, _WRITE_ZEROES handled above.
1682 	 * REQ_OP_FLUSH (empty flush) not expected,
1683 	 * should have been mapped to a "drbd protocol barrier".
1684 	 * REQ_OP_SECURE_ERASE: I don't see how we could ever support that.
1685 	 */
1686 	if (!(peer_req_op(peer_req) == REQ_OP_WRITE ||
1687 				peer_req_op(peer_req) == REQ_OP_READ)) {
1688 		drbd_err(device, "Invalid bio op received: 0x%x\n", peer_req->opf);
1689 		return -EINVAL;
1690 	}
1691 
1692 	bio = bio_alloc(device->ldev->backing_bdev, nr_pages, peer_req->opf, GFP_NOIO);
1693 	/* > peer_req->i.sector, unless this is the first bio */
1694 	bio->bi_iter.bi_sector = sector;
1695 	bio->bi_private = peer_req;
1696 	bio->bi_end_io = drbd_peer_request_endio;
1697 
1698 	bio->bi_next = bios;
1699 	bios = bio;
1700 	++n_bios;
1701 
1702 	page_chain_for_each(page) {
1703 		unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1704 		if (!bio_add_page(bio, page, len, 0))
1705 			goto next_bio;
1706 		data_size -= len;
1707 		sector += len >> 9;
1708 		--nr_pages;
1709 	}
1710 	D_ASSERT(device, data_size == 0);
1711 	D_ASSERT(device, page == NULL);
1712 
1713 	atomic_set(&peer_req->pending_bios, n_bios);
1714 	/* for debugfs: update timestamp, mark as submitted */
1715 	peer_req->submit_jif = jiffies;
1716 	peer_req->flags |= EE_SUBMITTED;
1717 	do {
1718 		bio = bios;
1719 		bios = bios->bi_next;
1720 		bio->bi_next = NULL;
1721 
1722 		drbd_submit_bio_noacct(device, peer_request_fault_type(peer_req), bio);
1723 	} while (bios);
1724 	return 0;
1725 }
1726 
1727 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1728 					     struct drbd_peer_request *peer_req)
1729 {
1730 	struct drbd_interval *i = &peer_req->i;
1731 
1732 	drbd_remove_interval(&device->write_requests, i);
1733 	drbd_clear_interval(i);
1734 
1735 	/* Wake up any processes waiting for this peer request to complete.  */
1736 	if (i->waiting)
1737 		wake_up(&device->misc_wait);
1738 }
1739 
1740 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1741 {
1742 	struct drbd_peer_device *peer_device;
1743 	int vnr;
1744 
1745 	rcu_read_lock();
1746 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1747 		struct drbd_device *device = peer_device->device;
1748 
1749 		kref_get(&device->kref);
1750 		rcu_read_unlock();
1751 		drbd_wait_ee_list_empty(device, &device->active_ee);
1752 		kref_put(&device->kref, drbd_destroy_device);
1753 		rcu_read_lock();
1754 	}
1755 	rcu_read_unlock();
1756 }
1757 
1758 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1759 {
1760 	int rv;
1761 	struct p_barrier *p = pi->data;
1762 	struct drbd_epoch *epoch;
1763 
1764 	/* FIXME these are unacked on connection,
1765 	 * not a specific (peer)device.
1766 	 */
1767 	connection->current_epoch->barrier_nr = p->barrier;
1768 	connection->current_epoch->connection = connection;
1769 	rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1770 
1771 	/* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1772 	 * the activity log, which means it would not be resynced in case the
1773 	 * R_PRIMARY crashes now.
1774 	 * Therefore we must send the barrier_ack after the barrier request was
1775 	 * completed. */
1776 	switch (connection->resource->write_ordering) {
1777 	case WO_NONE:
1778 		if (rv == FE_RECYCLED)
1779 			return 0;
1780 
1781 		/* receiver context, in the writeout path of the other node.
1782 		 * avoid potential distributed deadlock */
1783 		epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1784 		if (epoch)
1785 			break;
1786 		else
1787 			drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1788 		fallthrough;
1789 
1790 	case WO_BDEV_FLUSH:
1791 	case WO_DRAIN_IO:
1792 		conn_wait_active_ee_empty(connection);
1793 		drbd_flush(connection);
1794 
1795 		if (atomic_read(&connection->current_epoch->epoch_size)) {
1796 			epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1797 			if (epoch)
1798 				break;
1799 		}
1800 
1801 		return 0;
1802 	default:
1803 		drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1804 			 connection->resource->write_ordering);
1805 		return -EIO;
1806 	}
1807 
1808 	epoch->flags = 0;
1809 	atomic_set(&epoch->epoch_size, 0);
1810 	atomic_set(&epoch->active, 0);
1811 
1812 	spin_lock(&connection->epoch_lock);
1813 	if (atomic_read(&connection->current_epoch->epoch_size)) {
1814 		list_add(&epoch->list, &connection->current_epoch->list);
1815 		connection->current_epoch = epoch;
1816 		connection->epochs++;
1817 	} else {
1818 		/* The current_epoch got recycled while we allocated this one... */
1819 		kfree(epoch);
1820 	}
1821 	spin_unlock(&connection->epoch_lock);
1822 
1823 	return 0;
1824 }
1825 
1826 /* quick wrapper in case payload size != request_size (write same) */
1827 static void drbd_csum_ee_size(struct crypto_shash *h,
1828 			      struct drbd_peer_request *r, void *d,
1829 			      unsigned int payload_size)
1830 {
1831 	unsigned int tmp = r->i.size;
1832 	r->i.size = payload_size;
1833 	drbd_csum_ee(h, r, d);
1834 	r->i.size = tmp;
1835 }
1836 
1837 /* used from receive_RSDataReply (recv_resync_read)
1838  * and from receive_Data.
1839  * data_size: actual payload ("data in")
1840  * 	for normal writes that is bi_size.
1841  * 	for discards, that is zero.
1842  * 	for write same, it is logical_block_size.
1843  * both trim and write same have the bi_size ("data len to be affected")
1844  * as extra argument in the packet header.
1845  */
1846 static struct drbd_peer_request *
1847 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1848 	      struct packet_info *pi) __must_hold(local)
1849 {
1850 	struct drbd_device *device = peer_device->device;
1851 	const sector_t capacity = get_capacity(device->vdisk);
1852 	struct drbd_peer_request *peer_req;
1853 	struct page *page;
1854 	int digest_size, err;
1855 	unsigned int data_size = pi->size, ds;
1856 	void *dig_in = peer_device->connection->int_dig_in;
1857 	void *dig_vv = peer_device->connection->int_dig_vv;
1858 	unsigned long *data;
1859 	struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1860 	struct p_trim *zeroes = (pi->cmd == P_ZEROES) ? pi->data : NULL;
1861 
1862 	digest_size = 0;
1863 	if (!trim && peer_device->connection->peer_integrity_tfm) {
1864 		digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1865 		/*
1866 		 * FIXME: Receive the incoming digest into the receive buffer
1867 		 *	  here, together with its struct p_data?
1868 		 */
1869 		err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1870 		if (err)
1871 			return NULL;
1872 		data_size -= digest_size;
1873 	}
1874 
1875 	/* assume request_size == data_size, but special case trim. */
1876 	ds = data_size;
1877 	if (trim) {
1878 		if (!expect(peer_device, data_size == 0))
1879 			return NULL;
1880 		ds = be32_to_cpu(trim->size);
1881 	} else if (zeroes) {
1882 		if (!expect(peer_device, data_size == 0))
1883 			return NULL;
1884 		ds = be32_to_cpu(zeroes->size);
1885 	}
1886 
1887 	if (!expect(peer_device, IS_ALIGNED(ds, 512)))
1888 		return NULL;
1889 	if (trim || zeroes) {
1890 		if (!expect(peer_device, ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
1891 			return NULL;
1892 	} else if (!expect(peer_device, ds <= DRBD_MAX_BIO_SIZE))
1893 		return NULL;
1894 
1895 	/* even though we trust out peer,
1896 	 * we sometimes have to double check. */
1897 	if (sector + (ds>>9) > capacity) {
1898 		drbd_err(device, "request from peer beyond end of local disk: "
1899 			"capacity: %llus < sector: %llus + size: %u\n",
1900 			(unsigned long long)capacity,
1901 			(unsigned long long)sector, ds);
1902 		return NULL;
1903 	}
1904 
1905 	/* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1906 	 * "criss-cross" setup, that might cause write-out on some other DRBD,
1907 	 * which in turn might block on the other node at this very place.  */
1908 	peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
1909 	if (!peer_req)
1910 		return NULL;
1911 
1912 	peer_req->flags |= EE_WRITE;
1913 	if (trim) {
1914 		peer_req->flags |= EE_TRIM;
1915 		return peer_req;
1916 	}
1917 	if (zeroes) {
1918 		peer_req->flags |= EE_ZEROOUT;
1919 		return peer_req;
1920 	}
1921 
1922 	/* receive payload size bytes into page chain */
1923 	ds = data_size;
1924 	page = peer_req->pages;
1925 	page_chain_for_each(page) {
1926 		unsigned len = min_t(int, ds, PAGE_SIZE);
1927 		data = kmap(page);
1928 		err = drbd_recv_all_warn(peer_device->connection, data, len);
1929 		if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1930 			drbd_err(device, "Fault injection: Corrupting data on receive\n");
1931 			data[0] = data[0] ^ (unsigned long)-1;
1932 		}
1933 		kunmap(page);
1934 		if (err) {
1935 			drbd_free_peer_req(device, peer_req);
1936 			return NULL;
1937 		}
1938 		ds -= len;
1939 	}
1940 
1941 	if (digest_size) {
1942 		drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
1943 		if (memcmp(dig_in, dig_vv, digest_size)) {
1944 			drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1945 				(unsigned long long)sector, data_size);
1946 			drbd_free_peer_req(device, peer_req);
1947 			return NULL;
1948 		}
1949 	}
1950 	device->recv_cnt += data_size >> 9;
1951 	return peer_req;
1952 }
1953 
1954 /* drbd_drain_block() just takes a data block
1955  * out of the socket input buffer, and discards it.
1956  */
1957 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1958 {
1959 	struct page *page;
1960 	int err = 0;
1961 	void *data;
1962 
1963 	if (!data_size)
1964 		return 0;
1965 
1966 	page = drbd_alloc_pages(peer_device, 1, 1);
1967 
1968 	data = kmap(page);
1969 	while (data_size) {
1970 		unsigned int len = min_t(int, data_size, PAGE_SIZE);
1971 
1972 		err = drbd_recv_all_warn(peer_device->connection, data, len);
1973 		if (err)
1974 			break;
1975 		data_size -= len;
1976 	}
1977 	kunmap(page);
1978 	drbd_free_pages(peer_device->device, page, 0);
1979 	return err;
1980 }
1981 
1982 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1983 			   sector_t sector, int data_size)
1984 {
1985 	struct bio_vec bvec;
1986 	struct bvec_iter iter;
1987 	struct bio *bio;
1988 	int digest_size, err, expect;
1989 	void *dig_in = peer_device->connection->int_dig_in;
1990 	void *dig_vv = peer_device->connection->int_dig_vv;
1991 
1992 	digest_size = 0;
1993 	if (peer_device->connection->peer_integrity_tfm) {
1994 		digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1995 		err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1996 		if (err)
1997 			return err;
1998 		data_size -= digest_size;
1999 	}
2000 
2001 	/* optimistically update recv_cnt.  if receiving fails below,
2002 	 * we disconnect anyways, and counters will be reset. */
2003 	peer_device->device->recv_cnt += data_size>>9;
2004 
2005 	bio = req->master_bio;
2006 	D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
2007 
2008 	bio_for_each_segment(bvec, bio, iter) {
2009 		void *mapped = bvec_kmap_local(&bvec);
2010 		expect = min_t(int, data_size, bvec.bv_len);
2011 		err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
2012 		kunmap_local(mapped);
2013 		if (err)
2014 			return err;
2015 		data_size -= expect;
2016 	}
2017 
2018 	if (digest_size) {
2019 		drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
2020 		if (memcmp(dig_in, dig_vv, digest_size)) {
2021 			drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
2022 			return -EINVAL;
2023 		}
2024 	}
2025 
2026 	D_ASSERT(peer_device->device, data_size == 0);
2027 	return 0;
2028 }
2029 
2030 /*
2031  * e_end_resync_block() is called in ack_sender context via
2032  * drbd_finish_peer_reqs().
2033  */
2034 static int e_end_resync_block(struct drbd_work *w, int unused)
2035 {
2036 	struct drbd_peer_request *peer_req =
2037 		container_of(w, struct drbd_peer_request, w);
2038 	struct drbd_peer_device *peer_device = peer_req->peer_device;
2039 	struct drbd_device *device = peer_device->device;
2040 	sector_t sector = peer_req->i.sector;
2041 	int err;
2042 
2043 	D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2044 
2045 	if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2046 		drbd_set_in_sync(peer_device, sector, peer_req->i.size);
2047 		err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
2048 	} else {
2049 		/* Record failure to sync */
2050 		drbd_rs_failed_io(peer_device, sector, peer_req->i.size);
2051 
2052 		err  = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2053 	}
2054 	dec_unacked(device);
2055 
2056 	return err;
2057 }
2058 
2059 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
2060 			    struct packet_info *pi) __releases(local)
2061 {
2062 	struct drbd_device *device = peer_device->device;
2063 	struct drbd_peer_request *peer_req;
2064 
2065 	peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
2066 	if (!peer_req)
2067 		goto fail;
2068 
2069 	dec_rs_pending(peer_device);
2070 
2071 	inc_unacked(device);
2072 	/* corresponding dec_unacked() in e_end_resync_block()
2073 	 * respective _drbd_clear_done_ee */
2074 
2075 	peer_req->w.cb = e_end_resync_block;
2076 	peer_req->opf = REQ_OP_WRITE;
2077 	peer_req->submit_jif = jiffies;
2078 
2079 	spin_lock_irq(&device->resource->req_lock);
2080 	list_add_tail(&peer_req->w.list, &device->sync_ee);
2081 	spin_unlock_irq(&device->resource->req_lock);
2082 
2083 	atomic_add(pi->size >> 9, &device->rs_sect_ev);
2084 	if (drbd_submit_peer_request(peer_req) == 0)
2085 		return 0;
2086 
2087 	/* don't care for the reason here */
2088 	drbd_err(device, "submit failed, triggering re-connect\n");
2089 	spin_lock_irq(&device->resource->req_lock);
2090 	list_del(&peer_req->w.list);
2091 	spin_unlock_irq(&device->resource->req_lock);
2092 
2093 	drbd_free_peer_req(device, peer_req);
2094 fail:
2095 	put_ldev(device);
2096 	return -EIO;
2097 }
2098 
2099 static struct drbd_request *
2100 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
2101 	     sector_t sector, bool missing_ok, const char *func)
2102 {
2103 	struct drbd_request *req;
2104 
2105 	/* Request object according to our peer */
2106 	req = (struct drbd_request *)(unsigned long)id;
2107 	if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
2108 		return req;
2109 	if (!missing_ok) {
2110 		drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
2111 			(unsigned long)id, (unsigned long long)sector);
2112 	}
2113 	return NULL;
2114 }
2115 
2116 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
2117 {
2118 	struct drbd_peer_device *peer_device;
2119 	struct drbd_device *device;
2120 	struct drbd_request *req;
2121 	sector_t sector;
2122 	int err;
2123 	struct p_data *p = pi->data;
2124 
2125 	peer_device = conn_peer_device(connection, pi->vnr);
2126 	if (!peer_device)
2127 		return -EIO;
2128 	device = peer_device->device;
2129 
2130 	sector = be64_to_cpu(p->sector);
2131 
2132 	spin_lock_irq(&device->resource->req_lock);
2133 	req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
2134 	spin_unlock_irq(&device->resource->req_lock);
2135 	if (unlikely(!req))
2136 		return -EIO;
2137 
2138 	err = recv_dless_read(peer_device, req, sector, pi->size);
2139 	if (!err)
2140 		req_mod(req, DATA_RECEIVED, peer_device);
2141 	/* else: nothing. handled from drbd_disconnect...
2142 	 * I don't think we may complete this just yet
2143 	 * in case we are "on-disconnect: freeze" */
2144 
2145 	return err;
2146 }
2147 
2148 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
2149 {
2150 	struct drbd_peer_device *peer_device;
2151 	struct drbd_device *device;
2152 	sector_t sector;
2153 	int err;
2154 	struct p_data *p = pi->data;
2155 
2156 	peer_device = conn_peer_device(connection, pi->vnr);
2157 	if (!peer_device)
2158 		return -EIO;
2159 	device = peer_device->device;
2160 
2161 	sector = be64_to_cpu(p->sector);
2162 	D_ASSERT(device, p->block_id == ID_SYNCER);
2163 
2164 	if (get_ldev(device)) {
2165 		/* data is submitted to disk within recv_resync_read.
2166 		 * corresponding put_ldev done below on error,
2167 		 * or in drbd_peer_request_endio. */
2168 		err = recv_resync_read(peer_device, sector, pi);
2169 	} else {
2170 		if (drbd_ratelimit())
2171 			drbd_err(device, "Can not write resync data to local disk.\n");
2172 
2173 		err = drbd_drain_block(peer_device, pi->size);
2174 
2175 		drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2176 	}
2177 
2178 	atomic_add(pi->size >> 9, &device->rs_sect_in);
2179 
2180 	return err;
2181 }
2182 
2183 static void restart_conflicting_writes(struct drbd_device *device,
2184 				       sector_t sector, int size)
2185 {
2186 	struct drbd_interval *i;
2187 	struct drbd_request *req;
2188 
2189 	drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2190 		if (!i->local)
2191 			continue;
2192 		req = container_of(i, struct drbd_request, i);
2193 		if (req->rq_state & RQ_LOCAL_PENDING ||
2194 		    !(req->rq_state & RQ_POSTPONED))
2195 			continue;
2196 		/* as it is RQ_POSTPONED, this will cause it to
2197 		 * be queued on the retry workqueue. */
2198 		__req_mod(req, CONFLICT_RESOLVED, NULL, NULL);
2199 	}
2200 }
2201 
2202 /*
2203  * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2204  */
2205 static int e_end_block(struct drbd_work *w, int cancel)
2206 {
2207 	struct drbd_peer_request *peer_req =
2208 		container_of(w, struct drbd_peer_request, w);
2209 	struct drbd_peer_device *peer_device = peer_req->peer_device;
2210 	struct drbd_device *device = peer_device->device;
2211 	sector_t sector = peer_req->i.sector;
2212 	int err = 0, pcmd;
2213 
2214 	if (peer_req->flags & EE_SEND_WRITE_ACK) {
2215 		if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2216 			pcmd = (device->state.conn >= C_SYNC_SOURCE &&
2217 				device->state.conn <= C_PAUSED_SYNC_T &&
2218 				peer_req->flags & EE_MAY_SET_IN_SYNC) ?
2219 				P_RS_WRITE_ACK : P_WRITE_ACK;
2220 			err = drbd_send_ack(peer_device, pcmd, peer_req);
2221 			if (pcmd == P_RS_WRITE_ACK)
2222 				drbd_set_in_sync(peer_device, sector, peer_req->i.size);
2223 		} else {
2224 			err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2225 			/* we expect it to be marked out of sync anyways...
2226 			 * maybe assert this?  */
2227 		}
2228 		dec_unacked(device);
2229 	}
2230 
2231 	/* we delete from the conflict detection hash _after_ we sent out the
2232 	 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right.  */
2233 	if (peer_req->flags & EE_IN_INTERVAL_TREE) {
2234 		spin_lock_irq(&device->resource->req_lock);
2235 		D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
2236 		drbd_remove_epoch_entry_interval(device, peer_req);
2237 		if (peer_req->flags & EE_RESTART_REQUESTS)
2238 			restart_conflicting_writes(device, sector, peer_req->i.size);
2239 		spin_unlock_irq(&device->resource->req_lock);
2240 	} else
2241 		D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2242 
2243 	drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2244 
2245 	return err;
2246 }
2247 
2248 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2249 {
2250 	struct drbd_peer_request *peer_req =
2251 		container_of(w, struct drbd_peer_request, w);
2252 	struct drbd_peer_device *peer_device = peer_req->peer_device;
2253 	int err;
2254 
2255 	err = drbd_send_ack(peer_device, ack, peer_req);
2256 	dec_unacked(peer_device->device);
2257 
2258 	return err;
2259 }
2260 
2261 static int e_send_superseded(struct drbd_work *w, int unused)
2262 {
2263 	return e_send_ack(w, P_SUPERSEDED);
2264 }
2265 
2266 static int e_send_retry_write(struct drbd_work *w, int unused)
2267 {
2268 	struct drbd_peer_request *peer_req =
2269 		container_of(w, struct drbd_peer_request, w);
2270 	struct drbd_connection *connection = peer_req->peer_device->connection;
2271 
2272 	return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2273 			     P_RETRY_WRITE : P_SUPERSEDED);
2274 }
2275 
2276 static bool seq_greater(u32 a, u32 b)
2277 {
2278 	/*
2279 	 * We assume 32-bit wrap-around here.
2280 	 * For 24-bit wrap-around, we would have to shift:
2281 	 *  a <<= 8; b <<= 8;
2282 	 */
2283 	return (s32)a - (s32)b > 0;
2284 }
2285 
2286 static u32 seq_max(u32 a, u32 b)
2287 {
2288 	return seq_greater(a, b) ? a : b;
2289 }
2290 
2291 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2292 {
2293 	struct drbd_device *device = peer_device->device;
2294 	unsigned int newest_peer_seq;
2295 
2296 	if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2297 		spin_lock(&device->peer_seq_lock);
2298 		newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2299 		device->peer_seq = newest_peer_seq;
2300 		spin_unlock(&device->peer_seq_lock);
2301 		/* wake up only if we actually changed device->peer_seq */
2302 		if (peer_seq == newest_peer_seq)
2303 			wake_up(&device->seq_wait);
2304 	}
2305 }
2306 
2307 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2308 {
2309 	return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2310 }
2311 
2312 /* maybe change sync_ee into interval trees as well? */
2313 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2314 {
2315 	struct drbd_peer_request *rs_req;
2316 	bool rv = false;
2317 
2318 	spin_lock_irq(&device->resource->req_lock);
2319 	list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2320 		if (overlaps(peer_req->i.sector, peer_req->i.size,
2321 			     rs_req->i.sector, rs_req->i.size)) {
2322 			rv = true;
2323 			break;
2324 		}
2325 	}
2326 	spin_unlock_irq(&device->resource->req_lock);
2327 
2328 	return rv;
2329 }
2330 
2331 /* Called from receive_Data.
2332  * Synchronize packets on sock with packets on msock.
2333  *
2334  * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2335  * packet traveling on msock, they are still processed in the order they have
2336  * been sent.
2337  *
2338  * Note: we don't care for Ack packets overtaking P_DATA packets.
2339  *
2340  * In case packet_seq is larger than device->peer_seq number, there are
2341  * outstanding packets on the msock. We wait for them to arrive.
2342  * In case we are the logically next packet, we update device->peer_seq
2343  * ourselves. Correctly handles 32bit wrap around.
2344  *
2345  * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2346  * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2347  * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2348  * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2349  *
2350  * returns 0 if we may process the packet,
2351  * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2352 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2353 {
2354 	struct drbd_device *device = peer_device->device;
2355 	DEFINE_WAIT(wait);
2356 	long timeout;
2357 	int ret = 0, tp;
2358 
2359 	if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2360 		return 0;
2361 
2362 	spin_lock(&device->peer_seq_lock);
2363 	for (;;) {
2364 		if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2365 			device->peer_seq = seq_max(device->peer_seq, peer_seq);
2366 			break;
2367 		}
2368 
2369 		if (signal_pending(current)) {
2370 			ret = -ERESTARTSYS;
2371 			break;
2372 		}
2373 
2374 		rcu_read_lock();
2375 		tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2376 		rcu_read_unlock();
2377 
2378 		if (!tp)
2379 			break;
2380 
2381 		/* Only need to wait if two_primaries is enabled */
2382 		prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2383 		spin_unlock(&device->peer_seq_lock);
2384 		rcu_read_lock();
2385 		timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2386 		rcu_read_unlock();
2387 		timeout = schedule_timeout(timeout);
2388 		spin_lock(&device->peer_seq_lock);
2389 		if (!timeout) {
2390 			ret = -ETIMEDOUT;
2391 			drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2392 			break;
2393 		}
2394 	}
2395 	spin_unlock(&device->peer_seq_lock);
2396 	finish_wait(&device->seq_wait, &wait);
2397 	return ret;
2398 }
2399 
2400 static enum req_op wire_flags_to_bio_op(u32 dpf)
2401 {
2402 	if (dpf & DP_ZEROES)
2403 		return REQ_OP_WRITE_ZEROES;
2404 	if (dpf & DP_DISCARD)
2405 		return REQ_OP_DISCARD;
2406 	else
2407 		return REQ_OP_WRITE;
2408 }
2409 
2410 /* see also bio_flags_to_wire() */
2411 static blk_opf_t wire_flags_to_bio(struct drbd_connection *connection, u32 dpf)
2412 {
2413 	return wire_flags_to_bio_op(dpf) |
2414 		(dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2415 		(dpf & DP_FUA ? REQ_FUA : 0) |
2416 		(dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2417 }
2418 
2419 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2420 				    unsigned int size)
2421 {
2422 	struct drbd_peer_device *peer_device = first_peer_device(device);
2423 	struct drbd_interval *i;
2424 
2425     repeat:
2426 	drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2427 		struct drbd_request *req;
2428 		struct bio_and_error m;
2429 
2430 		if (!i->local)
2431 			continue;
2432 		req = container_of(i, struct drbd_request, i);
2433 		if (!(req->rq_state & RQ_POSTPONED))
2434 			continue;
2435 		req->rq_state &= ~RQ_POSTPONED;
2436 		__req_mod(req, NEG_ACKED, peer_device, &m);
2437 		spin_unlock_irq(&device->resource->req_lock);
2438 		if (m.bio)
2439 			complete_master_bio(device, &m);
2440 		spin_lock_irq(&device->resource->req_lock);
2441 		goto repeat;
2442 	}
2443 }
2444 
2445 static int handle_write_conflicts(struct drbd_device *device,
2446 				  struct drbd_peer_request *peer_req)
2447 {
2448 	struct drbd_connection *connection = peer_req->peer_device->connection;
2449 	bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2450 	sector_t sector = peer_req->i.sector;
2451 	const unsigned int size = peer_req->i.size;
2452 	struct drbd_interval *i;
2453 	bool equal;
2454 	int err;
2455 
2456 	/*
2457 	 * Inserting the peer request into the write_requests tree will prevent
2458 	 * new conflicting local requests from being added.
2459 	 */
2460 	drbd_insert_interval(&device->write_requests, &peer_req->i);
2461 
2462     repeat:
2463 	drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2464 		if (i == &peer_req->i)
2465 			continue;
2466 		if (i->completed)
2467 			continue;
2468 
2469 		if (!i->local) {
2470 			/*
2471 			 * Our peer has sent a conflicting remote request; this
2472 			 * should not happen in a two-node setup.  Wait for the
2473 			 * earlier peer request to complete.
2474 			 */
2475 			err = drbd_wait_misc(device, i);
2476 			if (err)
2477 				goto out;
2478 			goto repeat;
2479 		}
2480 
2481 		equal = i->sector == sector && i->size == size;
2482 		if (resolve_conflicts) {
2483 			/*
2484 			 * If the peer request is fully contained within the
2485 			 * overlapping request, it can be considered overwritten
2486 			 * and thus superseded; otherwise, it will be retried
2487 			 * once all overlapping requests have completed.
2488 			 */
2489 			bool superseded = i->sector <= sector && i->sector +
2490 				       (i->size >> 9) >= sector + (size >> 9);
2491 
2492 			if (!equal)
2493 				drbd_alert(device, "Concurrent writes detected: "
2494 					       "local=%llus +%u, remote=%llus +%u, "
2495 					       "assuming %s came first\n",
2496 					  (unsigned long long)i->sector, i->size,
2497 					  (unsigned long long)sector, size,
2498 					  superseded ? "local" : "remote");
2499 
2500 			peer_req->w.cb = superseded ? e_send_superseded :
2501 						   e_send_retry_write;
2502 			list_add_tail(&peer_req->w.list, &device->done_ee);
2503 			queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2504 
2505 			err = -ENOENT;
2506 			goto out;
2507 		} else {
2508 			struct drbd_request *req =
2509 				container_of(i, struct drbd_request, i);
2510 
2511 			if (!equal)
2512 				drbd_alert(device, "Concurrent writes detected: "
2513 					       "local=%llus +%u, remote=%llus +%u\n",
2514 					  (unsigned long long)i->sector, i->size,
2515 					  (unsigned long long)sector, size);
2516 
2517 			if (req->rq_state & RQ_LOCAL_PENDING ||
2518 			    !(req->rq_state & RQ_POSTPONED)) {
2519 				/*
2520 				 * Wait for the node with the discard flag to
2521 				 * decide if this request has been superseded
2522 				 * or needs to be retried.
2523 				 * Requests that have been superseded will
2524 				 * disappear from the write_requests tree.
2525 				 *
2526 				 * In addition, wait for the conflicting
2527 				 * request to finish locally before submitting
2528 				 * the conflicting peer request.
2529 				 */
2530 				err = drbd_wait_misc(device, &req->i);
2531 				if (err) {
2532 					_conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2533 					fail_postponed_requests(device, sector, size);
2534 					goto out;
2535 				}
2536 				goto repeat;
2537 			}
2538 			/*
2539 			 * Remember to restart the conflicting requests after
2540 			 * the new peer request has completed.
2541 			 */
2542 			peer_req->flags |= EE_RESTART_REQUESTS;
2543 		}
2544 	}
2545 	err = 0;
2546 
2547     out:
2548 	if (err)
2549 		drbd_remove_epoch_entry_interval(device, peer_req);
2550 	return err;
2551 }
2552 
2553 /* mirrored write */
2554 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2555 {
2556 	struct drbd_peer_device *peer_device;
2557 	struct drbd_device *device;
2558 	struct net_conf *nc;
2559 	sector_t sector;
2560 	struct drbd_peer_request *peer_req;
2561 	struct p_data *p = pi->data;
2562 	u32 peer_seq = be32_to_cpu(p->seq_num);
2563 	u32 dp_flags;
2564 	int err, tp;
2565 
2566 	peer_device = conn_peer_device(connection, pi->vnr);
2567 	if (!peer_device)
2568 		return -EIO;
2569 	device = peer_device->device;
2570 
2571 	if (!get_ldev(device)) {
2572 		int err2;
2573 
2574 		err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2575 		drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2576 		atomic_inc(&connection->current_epoch->epoch_size);
2577 		err2 = drbd_drain_block(peer_device, pi->size);
2578 		if (!err)
2579 			err = err2;
2580 		return err;
2581 	}
2582 
2583 	/*
2584 	 * Corresponding put_ldev done either below (on various errors), or in
2585 	 * drbd_peer_request_endio, if we successfully submit the data at the
2586 	 * end of this function.
2587 	 */
2588 
2589 	sector = be64_to_cpu(p->sector);
2590 	peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2591 	if (!peer_req) {
2592 		put_ldev(device);
2593 		return -EIO;
2594 	}
2595 
2596 	peer_req->w.cb = e_end_block;
2597 	peer_req->submit_jif = jiffies;
2598 	peer_req->flags |= EE_APPLICATION;
2599 
2600 	dp_flags = be32_to_cpu(p->dp_flags);
2601 	peer_req->opf = wire_flags_to_bio(connection, dp_flags);
2602 	if (pi->cmd == P_TRIM) {
2603 		D_ASSERT(peer_device, peer_req->i.size > 0);
2604 		D_ASSERT(peer_device, peer_req_op(peer_req) == REQ_OP_DISCARD);
2605 		D_ASSERT(peer_device, peer_req->pages == NULL);
2606 		/* need to play safe: an older DRBD sender
2607 		 * may mean zero-out while sending P_TRIM. */
2608 		if (0 == (connection->agreed_features & DRBD_FF_WZEROES))
2609 			peer_req->flags |= EE_ZEROOUT;
2610 	} else if (pi->cmd == P_ZEROES) {
2611 		D_ASSERT(peer_device, peer_req->i.size > 0);
2612 		D_ASSERT(peer_device, peer_req_op(peer_req) == REQ_OP_WRITE_ZEROES);
2613 		D_ASSERT(peer_device, peer_req->pages == NULL);
2614 		/* Do (not) pass down BLKDEV_ZERO_NOUNMAP? */
2615 		if (dp_flags & DP_DISCARD)
2616 			peer_req->flags |= EE_TRIM;
2617 	} else if (peer_req->pages == NULL) {
2618 		D_ASSERT(device, peer_req->i.size == 0);
2619 		D_ASSERT(device, dp_flags & DP_FLUSH);
2620 	}
2621 
2622 	if (dp_flags & DP_MAY_SET_IN_SYNC)
2623 		peer_req->flags |= EE_MAY_SET_IN_SYNC;
2624 
2625 	spin_lock(&connection->epoch_lock);
2626 	peer_req->epoch = connection->current_epoch;
2627 	atomic_inc(&peer_req->epoch->epoch_size);
2628 	atomic_inc(&peer_req->epoch->active);
2629 	spin_unlock(&connection->epoch_lock);
2630 
2631 	rcu_read_lock();
2632 	nc = rcu_dereference(peer_device->connection->net_conf);
2633 	tp = nc->two_primaries;
2634 	if (peer_device->connection->agreed_pro_version < 100) {
2635 		switch (nc->wire_protocol) {
2636 		case DRBD_PROT_C:
2637 			dp_flags |= DP_SEND_WRITE_ACK;
2638 			break;
2639 		case DRBD_PROT_B:
2640 			dp_flags |= DP_SEND_RECEIVE_ACK;
2641 			break;
2642 		}
2643 	}
2644 	rcu_read_unlock();
2645 
2646 	if (dp_flags & DP_SEND_WRITE_ACK) {
2647 		peer_req->flags |= EE_SEND_WRITE_ACK;
2648 		inc_unacked(device);
2649 		/* corresponding dec_unacked() in e_end_block()
2650 		 * respective _drbd_clear_done_ee */
2651 	}
2652 
2653 	if (dp_flags & DP_SEND_RECEIVE_ACK) {
2654 		/* I really don't like it that the receiver thread
2655 		 * sends on the msock, but anyways */
2656 		drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2657 	}
2658 
2659 	if (tp) {
2660 		/* two primaries implies protocol C */
2661 		D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2662 		peer_req->flags |= EE_IN_INTERVAL_TREE;
2663 		err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2664 		if (err)
2665 			goto out_interrupted;
2666 		spin_lock_irq(&device->resource->req_lock);
2667 		err = handle_write_conflicts(device, peer_req);
2668 		if (err) {
2669 			spin_unlock_irq(&device->resource->req_lock);
2670 			if (err == -ENOENT) {
2671 				put_ldev(device);
2672 				return 0;
2673 			}
2674 			goto out_interrupted;
2675 		}
2676 	} else {
2677 		update_peer_seq(peer_device, peer_seq);
2678 		spin_lock_irq(&device->resource->req_lock);
2679 	}
2680 	/* TRIM and is processed synchronously,
2681 	 * we wait for all pending requests, respectively wait for
2682 	 * active_ee to become empty in drbd_submit_peer_request();
2683 	 * better not add ourselves here. */
2684 	if ((peer_req->flags & (EE_TRIM | EE_ZEROOUT)) == 0)
2685 		list_add_tail(&peer_req->w.list, &device->active_ee);
2686 	spin_unlock_irq(&device->resource->req_lock);
2687 
2688 	if (device->state.conn == C_SYNC_TARGET)
2689 		wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2690 
2691 	if (device->state.pdsk < D_INCONSISTENT) {
2692 		/* In case we have the only disk of the cluster, */
2693 		drbd_set_out_of_sync(peer_device, peer_req->i.sector, peer_req->i.size);
2694 		peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2695 		drbd_al_begin_io(device, &peer_req->i);
2696 		peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2697 	}
2698 
2699 	err = drbd_submit_peer_request(peer_req);
2700 	if (!err)
2701 		return 0;
2702 
2703 	/* don't care for the reason here */
2704 	drbd_err(device, "submit failed, triggering re-connect\n");
2705 	spin_lock_irq(&device->resource->req_lock);
2706 	list_del(&peer_req->w.list);
2707 	drbd_remove_epoch_entry_interval(device, peer_req);
2708 	spin_unlock_irq(&device->resource->req_lock);
2709 	if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2710 		peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2711 		drbd_al_complete_io(device, &peer_req->i);
2712 	}
2713 
2714 out_interrupted:
2715 	drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
2716 	put_ldev(device);
2717 	drbd_free_peer_req(device, peer_req);
2718 	return err;
2719 }
2720 
2721 /* We may throttle resync, if the lower device seems to be busy,
2722  * and current sync rate is above c_min_rate.
2723  *
2724  * To decide whether or not the lower device is busy, we use a scheme similar
2725  * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2726  * (more than 64 sectors) of activity we cannot account for with our own resync
2727  * activity, it obviously is "busy".
2728  *
2729  * The current sync rate used here uses only the most recent two step marks,
2730  * to have a short time average so we can react faster.
2731  */
2732 bool drbd_rs_should_slow_down(struct drbd_peer_device *peer_device, sector_t sector,
2733 		bool throttle_if_app_is_waiting)
2734 {
2735 	struct drbd_device *device = peer_device->device;
2736 	struct lc_element *tmp;
2737 	bool throttle = drbd_rs_c_min_rate_throttle(device);
2738 
2739 	if (!throttle || throttle_if_app_is_waiting)
2740 		return throttle;
2741 
2742 	spin_lock_irq(&device->al_lock);
2743 	tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2744 	if (tmp) {
2745 		struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2746 		if (test_bit(BME_PRIORITY, &bm_ext->flags))
2747 			throttle = false;
2748 		/* Do not slow down if app IO is already waiting for this extent,
2749 		 * and our progress is necessary for application IO to complete. */
2750 	}
2751 	spin_unlock_irq(&device->al_lock);
2752 
2753 	return throttle;
2754 }
2755 
2756 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2757 {
2758 	struct gendisk *disk = device->ldev->backing_bdev->bd_disk;
2759 	unsigned long db, dt, dbdt;
2760 	unsigned int c_min_rate;
2761 	int curr_events;
2762 
2763 	rcu_read_lock();
2764 	c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2765 	rcu_read_unlock();
2766 
2767 	/* feature disabled? */
2768 	if (c_min_rate == 0)
2769 		return false;
2770 
2771 	curr_events = (int)part_stat_read_accum(disk->part0, sectors) -
2772 			atomic_read(&device->rs_sect_ev);
2773 
2774 	if (atomic_read(&device->ap_actlog_cnt)
2775 	    || curr_events - device->rs_last_events > 64) {
2776 		unsigned long rs_left;
2777 		int i;
2778 
2779 		device->rs_last_events = curr_events;
2780 
2781 		/* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2782 		 * approx. */
2783 		i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2784 
2785 		if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2786 			rs_left = device->ov_left;
2787 		else
2788 			rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2789 
2790 		dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2791 		if (!dt)
2792 			dt++;
2793 		db = device->rs_mark_left[i] - rs_left;
2794 		dbdt = Bit2KB(db/dt);
2795 
2796 		if (dbdt > c_min_rate)
2797 			return true;
2798 	}
2799 	return false;
2800 }
2801 
2802 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2803 {
2804 	struct drbd_peer_device *peer_device;
2805 	struct drbd_device *device;
2806 	sector_t sector;
2807 	sector_t capacity;
2808 	struct drbd_peer_request *peer_req;
2809 	struct digest_info *di = NULL;
2810 	int size, verb;
2811 	struct p_block_req *p =	pi->data;
2812 
2813 	peer_device = conn_peer_device(connection, pi->vnr);
2814 	if (!peer_device)
2815 		return -EIO;
2816 	device = peer_device->device;
2817 	capacity = get_capacity(device->vdisk);
2818 
2819 	sector = be64_to_cpu(p->sector);
2820 	size   = be32_to_cpu(p->blksize);
2821 
2822 	if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2823 		drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2824 				(unsigned long long)sector, size);
2825 		return -EINVAL;
2826 	}
2827 	if (sector + (size>>9) > capacity) {
2828 		drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2829 				(unsigned long long)sector, size);
2830 		return -EINVAL;
2831 	}
2832 
2833 	if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2834 		verb = 1;
2835 		switch (pi->cmd) {
2836 		case P_DATA_REQUEST:
2837 			drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2838 			break;
2839 		case P_RS_THIN_REQ:
2840 		case P_RS_DATA_REQUEST:
2841 		case P_CSUM_RS_REQUEST:
2842 		case P_OV_REQUEST:
2843 			drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2844 			break;
2845 		case P_OV_REPLY:
2846 			verb = 0;
2847 			dec_rs_pending(peer_device);
2848 			drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2849 			break;
2850 		default:
2851 			BUG();
2852 		}
2853 		if (verb && drbd_ratelimit())
2854 			drbd_err(device, "Can not satisfy peer's read request, "
2855 			    "no local data.\n");
2856 
2857 		/* drain possibly payload */
2858 		return drbd_drain_block(peer_device, pi->size);
2859 	}
2860 
2861 	/* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2862 	 * "criss-cross" setup, that might cause write-out on some other DRBD,
2863 	 * which in turn might block on the other node at this very place.  */
2864 	peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2865 			size, GFP_NOIO);
2866 	if (!peer_req) {
2867 		put_ldev(device);
2868 		return -ENOMEM;
2869 	}
2870 	peer_req->opf = REQ_OP_READ;
2871 
2872 	switch (pi->cmd) {
2873 	case P_DATA_REQUEST:
2874 		peer_req->w.cb = w_e_end_data_req;
2875 		/* application IO, don't drbd_rs_begin_io */
2876 		peer_req->flags |= EE_APPLICATION;
2877 		goto submit;
2878 
2879 	case P_RS_THIN_REQ:
2880 		/* If at some point in the future we have a smart way to
2881 		   find out if this data block is completely deallocated,
2882 		   then we would do something smarter here than reading
2883 		   the block... */
2884 		peer_req->flags |= EE_RS_THIN_REQ;
2885 		fallthrough;
2886 	case P_RS_DATA_REQUEST:
2887 		peer_req->w.cb = w_e_end_rsdata_req;
2888 		/* used in the sector offset progress display */
2889 		device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2890 		break;
2891 
2892 	case P_OV_REPLY:
2893 	case P_CSUM_RS_REQUEST:
2894 		di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2895 		if (!di)
2896 			goto out_free_e;
2897 
2898 		di->digest_size = pi->size;
2899 		di->digest = (((char *)di)+sizeof(struct digest_info));
2900 
2901 		peer_req->digest = di;
2902 		peer_req->flags |= EE_HAS_DIGEST;
2903 
2904 		if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2905 			goto out_free_e;
2906 
2907 		if (pi->cmd == P_CSUM_RS_REQUEST) {
2908 			D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2909 			peer_req->w.cb = w_e_end_csum_rs_req;
2910 			/* used in the sector offset progress display */
2911 			device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2912 			/* remember to report stats in drbd_resync_finished */
2913 			device->use_csums = true;
2914 		} else if (pi->cmd == P_OV_REPLY) {
2915 			/* track progress, we may need to throttle */
2916 			atomic_add(size >> 9, &device->rs_sect_in);
2917 			peer_req->w.cb = w_e_end_ov_reply;
2918 			dec_rs_pending(peer_device);
2919 			/* drbd_rs_begin_io done when we sent this request,
2920 			 * but accounting still needs to be done. */
2921 			goto submit_for_resync;
2922 		}
2923 		break;
2924 
2925 	case P_OV_REQUEST:
2926 		if (device->ov_start_sector == ~(sector_t)0 &&
2927 		    peer_device->connection->agreed_pro_version >= 90) {
2928 			unsigned long now = jiffies;
2929 			int i;
2930 			device->ov_start_sector = sector;
2931 			device->ov_position = sector;
2932 			device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2933 			device->rs_total = device->ov_left;
2934 			for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2935 				device->rs_mark_left[i] = device->ov_left;
2936 				device->rs_mark_time[i] = now;
2937 			}
2938 			drbd_info(device, "Online Verify start sector: %llu\n",
2939 					(unsigned long long)sector);
2940 		}
2941 		peer_req->w.cb = w_e_end_ov_req;
2942 		break;
2943 
2944 	default:
2945 		BUG();
2946 	}
2947 
2948 	/* Throttle, drbd_rs_begin_io and submit should become asynchronous
2949 	 * wrt the receiver, but it is not as straightforward as it may seem.
2950 	 * Various places in the resync start and stop logic assume resync
2951 	 * requests are processed in order, requeuing this on the worker thread
2952 	 * introduces a bunch of new code for synchronization between threads.
2953 	 *
2954 	 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2955 	 * "forever", throttling after drbd_rs_begin_io will lock that extent
2956 	 * for application writes for the same time.  For now, just throttle
2957 	 * here, where the rest of the code expects the receiver to sleep for
2958 	 * a while, anyways.
2959 	 */
2960 
2961 	/* Throttle before drbd_rs_begin_io, as that locks out application IO;
2962 	 * this defers syncer requests for some time, before letting at least
2963 	 * on request through.  The resync controller on the receiving side
2964 	 * will adapt to the incoming rate accordingly.
2965 	 *
2966 	 * We cannot throttle here if remote is Primary/SyncTarget:
2967 	 * we would also throttle its application reads.
2968 	 * In that case, throttling is done on the SyncTarget only.
2969 	 */
2970 
2971 	/* Even though this may be a resync request, we do add to "read_ee";
2972 	 * "sync_ee" is only used for resync WRITEs.
2973 	 * Add to list early, so debugfs can find this request
2974 	 * even if we have to sleep below. */
2975 	spin_lock_irq(&device->resource->req_lock);
2976 	list_add_tail(&peer_req->w.list, &device->read_ee);
2977 	spin_unlock_irq(&device->resource->req_lock);
2978 
2979 	update_receiver_timing_details(connection, drbd_rs_should_slow_down);
2980 	if (device->state.peer != R_PRIMARY
2981 	&& drbd_rs_should_slow_down(peer_device, sector, false))
2982 		schedule_timeout_uninterruptible(HZ/10);
2983 	update_receiver_timing_details(connection, drbd_rs_begin_io);
2984 	if (drbd_rs_begin_io(device, sector))
2985 		goto out_free_e;
2986 
2987 submit_for_resync:
2988 	atomic_add(size >> 9, &device->rs_sect_ev);
2989 
2990 submit:
2991 	update_receiver_timing_details(connection, drbd_submit_peer_request);
2992 	inc_unacked(device);
2993 	if (drbd_submit_peer_request(peer_req) == 0)
2994 		return 0;
2995 
2996 	/* don't care for the reason here */
2997 	drbd_err(device, "submit failed, triggering re-connect\n");
2998 
2999 out_free_e:
3000 	spin_lock_irq(&device->resource->req_lock);
3001 	list_del(&peer_req->w.list);
3002 	spin_unlock_irq(&device->resource->req_lock);
3003 	/* no drbd_rs_complete_io(), we are dropping the connection anyways */
3004 
3005 	put_ldev(device);
3006 	drbd_free_peer_req(device, peer_req);
3007 	return -EIO;
3008 }
3009 
3010 /*
3011  * drbd_asb_recover_0p  -  Recover after split-brain with no remaining primaries
3012  */
3013 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
3014 {
3015 	struct drbd_device *device = peer_device->device;
3016 	int self, peer, rv = -100;
3017 	unsigned long ch_self, ch_peer;
3018 	enum drbd_after_sb_p after_sb_0p;
3019 
3020 	self = device->ldev->md.uuid[UI_BITMAP] & 1;
3021 	peer = device->p_uuid[UI_BITMAP] & 1;
3022 
3023 	ch_peer = device->p_uuid[UI_SIZE];
3024 	ch_self = device->comm_bm_set;
3025 
3026 	rcu_read_lock();
3027 	after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
3028 	rcu_read_unlock();
3029 	switch (after_sb_0p) {
3030 	case ASB_CONSENSUS:
3031 	case ASB_DISCARD_SECONDARY:
3032 	case ASB_CALL_HELPER:
3033 	case ASB_VIOLENTLY:
3034 		drbd_err(device, "Configuration error.\n");
3035 		break;
3036 	case ASB_DISCONNECT:
3037 		break;
3038 	case ASB_DISCARD_YOUNGER_PRI:
3039 		if (self == 0 && peer == 1) {
3040 			rv = -1;
3041 			break;
3042 		}
3043 		if (self == 1 && peer == 0) {
3044 			rv =  1;
3045 			break;
3046 		}
3047 		fallthrough;	/* to one of the other strategies */
3048 	case ASB_DISCARD_OLDER_PRI:
3049 		if (self == 0 && peer == 1) {
3050 			rv = 1;
3051 			break;
3052 		}
3053 		if (self == 1 && peer == 0) {
3054 			rv = -1;
3055 			break;
3056 		}
3057 		/* Else fall through to one of the other strategies... */
3058 		drbd_warn(device, "Discard younger/older primary did not find a decision\n"
3059 		     "Using discard-least-changes instead\n");
3060 		fallthrough;
3061 	case ASB_DISCARD_ZERO_CHG:
3062 		if (ch_peer == 0 && ch_self == 0) {
3063 			rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3064 				? -1 : 1;
3065 			break;
3066 		} else {
3067 			if (ch_peer == 0) { rv =  1; break; }
3068 			if (ch_self == 0) { rv = -1; break; }
3069 		}
3070 		if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
3071 			break;
3072 		fallthrough;
3073 	case ASB_DISCARD_LEAST_CHG:
3074 		if	(ch_self < ch_peer)
3075 			rv = -1;
3076 		else if (ch_self > ch_peer)
3077 			rv =  1;
3078 		else /* ( ch_self == ch_peer ) */
3079 		     /* Well, then use something else. */
3080 			rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3081 				? -1 : 1;
3082 		break;
3083 	case ASB_DISCARD_LOCAL:
3084 		rv = -1;
3085 		break;
3086 	case ASB_DISCARD_REMOTE:
3087 		rv =  1;
3088 	}
3089 
3090 	return rv;
3091 }
3092 
3093 /*
3094  * drbd_asb_recover_1p  -  Recover after split-brain with one remaining primary
3095  */
3096 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
3097 {
3098 	struct drbd_device *device = peer_device->device;
3099 	int hg, rv = -100;
3100 	enum drbd_after_sb_p after_sb_1p;
3101 
3102 	rcu_read_lock();
3103 	after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
3104 	rcu_read_unlock();
3105 	switch (after_sb_1p) {
3106 	case ASB_DISCARD_YOUNGER_PRI:
3107 	case ASB_DISCARD_OLDER_PRI:
3108 	case ASB_DISCARD_LEAST_CHG:
3109 	case ASB_DISCARD_LOCAL:
3110 	case ASB_DISCARD_REMOTE:
3111 	case ASB_DISCARD_ZERO_CHG:
3112 		drbd_err(device, "Configuration error.\n");
3113 		break;
3114 	case ASB_DISCONNECT:
3115 		break;
3116 	case ASB_CONSENSUS:
3117 		hg = drbd_asb_recover_0p(peer_device);
3118 		if (hg == -1 && device->state.role == R_SECONDARY)
3119 			rv = hg;
3120 		if (hg == 1  && device->state.role == R_PRIMARY)
3121 			rv = hg;
3122 		break;
3123 	case ASB_VIOLENTLY:
3124 		rv = drbd_asb_recover_0p(peer_device);
3125 		break;
3126 	case ASB_DISCARD_SECONDARY:
3127 		return device->state.role == R_PRIMARY ? 1 : -1;
3128 	case ASB_CALL_HELPER:
3129 		hg = drbd_asb_recover_0p(peer_device);
3130 		if (hg == -1 && device->state.role == R_PRIMARY) {
3131 			enum drbd_state_rv rv2;
3132 
3133 			 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3134 			  * we might be here in C_WF_REPORT_PARAMS which is transient.
3135 			  * we do not need to wait for the after state change work either. */
3136 			rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3137 			if (rv2 != SS_SUCCESS) {
3138 				drbd_khelper(device, "pri-lost-after-sb");
3139 			} else {
3140 				drbd_warn(device, "Successfully gave up primary role.\n");
3141 				rv = hg;
3142 			}
3143 		} else
3144 			rv = hg;
3145 	}
3146 
3147 	return rv;
3148 }
3149 
3150 /*
3151  * drbd_asb_recover_2p  -  Recover after split-brain with two remaining primaries
3152  */
3153 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
3154 {
3155 	struct drbd_device *device = peer_device->device;
3156 	int hg, rv = -100;
3157 	enum drbd_after_sb_p after_sb_2p;
3158 
3159 	rcu_read_lock();
3160 	after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
3161 	rcu_read_unlock();
3162 	switch (after_sb_2p) {
3163 	case ASB_DISCARD_YOUNGER_PRI:
3164 	case ASB_DISCARD_OLDER_PRI:
3165 	case ASB_DISCARD_LEAST_CHG:
3166 	case ASB_DISCARD_LOCAL:
3167 	case ASB_DISCARD_REMOTE:
3168 	case ASB_CONSENSUS:
3169 	case ASB_DISCARD_SECONDARY:
3170 	case ASB_DISCARD_ZERO_CHG:
3171 		drbd_err(device, "Configuration error.\n");
3172 		break;
3173 	case ASB_VIOLENTLY:
3174 		rv = drbd_asb_recover_0p(peer_device);
3175 		break;
3176 	case ASB_DISCONNECT:
3177 		break;
3178 	case ASB_CALL_HELPER:
3179 		hg = drbd_asb_recover_0p(peer_device);
3180 		if (hg == -1) {
3181 			enum drbd_state_rv rv2;
3182 
3183 			 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3184 			  * we might be here in C_WF_REPORT_PARAMS which is transient.
3185 			  * we do not need to wait for the after state change work either. */
3186 			rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3187 			if (rv2 != SS_SUCCESS) {
3188 				drbd_khelper(device, "pri-lost-after-sb");
3189 			} else {
3190 				drbd_warn(device, "Successfully gave up primary role.\n");
3191 				rv = hg;
3192 			}
3193 		} else
3194 			rv = hg;
3195 	}
3196 
3197 	return rv;
3198 }
3199 
3200 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
3201 			   u64 bits, u64 flags)
3202 {
3203 	if (!uuid) {
3204 		drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
3205 		return;
3206 	}
3207 	drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
3208 	     text,
3209 	     (unsigned long long)uuid[UI_CURRENT],
3210 	     (unsigned long long)uuid[UI_BITMAP],
3211 	     (unsigned long long)uuid[UI_HISTORY_START],
3212 	     (unsigned long long)uuid[UI_HISTORY_END],
3213 	     (unsigned long long)bits,
3214 	     (unsigned long long)flags);
3215 }
3216 
3217 /*
3218   100	after split brain try auto recover
3219     2	C_SYNC_SOURCE set BitMap
3220     1	C_SYNC_SOURCE use BitMap
3221     0	no Sync
3222    -1	C_SYNC_TARGET use BitMap
3223    -2	C_SYNC_TARGET set BitMap
3224  -100	after split brain, disconnect
3225 -1000	unrelated data
3226 -1091   requires proto 91
3227 -1096   requires proto 96
3228  */
3229 
3230 static int drbd_uuid_compare(struct drbd_peer_device *const peer_device,
3231 		enum drbd_role const peer_role, int *rule_nr) __must_hold(local)
3232 {
3233 	struct drbd_connection *const connection = peer_device->connection;
3234 	struct drbd_device *device = peer_device->device;
3235 	u64 self, peer;
3236 	int i, j;
3237 
3238 	self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3239 	peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3240 
3241 	*rule_nr = 10;
3242 	if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
3243 		return 0;
3244 
3245 	*rule_nr = 20;
3246 	if ((self == UUID_JUST_CREATED || self == (u64)0) &&
3247 	     peer != UUID_JUST_CREATED)
3248 		return -2;
3249 
3250 	*rule_nr = 30;
3251 	if (self != UUID_JUST_CREATED &&
3252 	    (peer == UUID_JUST_CREATED || peer == (u64)0))
3253 		return 2;
3254 
3255 	if (self == peer) {
3256 		int rct, dc; /* roles at crash time */
3257 
3258 		if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3259 
3260 			if (connection->agreed_pro_version < 91)
3261 				return -1091;
3262 
3263 			if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3264 			    (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3265 				drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3266 				drbd_uuid_move_history(device);
3267 				device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3268 				device->ldev->md.uuid[UI_BITMAP] = 0;
3269 
3270 				drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3271 					       device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3272 				*rule_nr = 34;
3273 			} else {
3274 				drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3275 				*rule_nr = 36;
3276 			}
3277 
3278 			return 1;
3279 		}
3280 
3281 		if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3282 
3283 			if (connection->agreed_pro_version < 91)
3284 				return -1091;
3285 
3286 			if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3287 			    (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3288 				drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3289 
3290 				device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3291 				device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3292 				device->p_uuid[UI_BITMAP] = 0UL;
3293 
3294 				drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3295 				*rule_nr = 35;
3296 			} else {
3297 				drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3298 				*rule_nr = 37;
3299 			}
3300 
3301 			return -1;
3302 		}
3303 
3304 		/* Common power [off|failure] */
3305 		rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3306 			(device->p_uuid[UI_FLAGS] & 2);
3307 		/* lowest bit is set when we were primary,
3308 		 * next bit (weight 2) is set when peer was primary */
3309 		*rule_nr = 40;
3310 
3311 		/* Neither has the "crashed primary" flag set,
3312 		 * only a replication link hickup. */
3313 		if (rct == 0)
3314 			return 0;
3315 
3316 		/* Current UUID equal and no bitmap uuid; does not necessarily
3317 		 * mean this was a "simultaneous hard crash", maybe IO was
3318 		 * frozen, so no UUID-bump happened.
3319 		 * This is a protocol change, overload DRBD_FF_WSAME as flag
3320 		 * for "new-enough" peer DRBD version. */
3321 		if (device->state.role == R_PRIMARY || peer_role == R_PRIMARY) {
3322 			*rule_nr = 41;
3323 			if (!(connection->agreed_features & DRBD_FF_WSAME)) {
3324 				drbd_warn(peer_device, "Equivalent unrotated UUIDs, but current primary present.\n");
3325 				return -(0x10000 | PRO_VERSION_MAX | (DRBD_FF_WSAME << 8));
3326 			}
3327 			if (device->state.role == R_PRIMARY && peer_role == R_PRIMARY) {
3328 				/* At least one has the "crashed primary" bit set,
3329 				 * both are primary now, but neither has rotated its UUIDs?
3330 				 * "Can not happen." */
3331 				drbd_err(peer_device, "Equivalent unrotated UUIDs, but both are primary. Can not resolve this.\n");
3332 				return -100;
3333 			}
3334 			if (device->state.role == R_PRIMARY)
3335 				return 1;
3336 			return -1;
3337 		}
3338 
3339 		/* Both are secondary.
3340 		 * Really looks like recovery from simultaneous hard crash.
3341 		 * Check which had been primary before, and arbitrate. */
3342 		switch (rct) {
3343 		case 0: /* !self_pri && !peer_pri */ return 0; /* already handled */
3344 		case 1: /*  self_pri && !peer_pri */ return 1;
3345 		case 2: /* !self_pri &&  peer_pri */ return -1;
3346 		case 3: /*  self_pri &&  peer_pri */
3347 			dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3348 			return dc ? -1 : 1;
3349 		}
3350 	}
3351 
3352 	*rule_nr = 50;
3353 	peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3354 	if (self == peer)
3355 		return -1;
3356 
3357 	*rule_nr = 51;
3358 	peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3359 	if (self == peer) {
3360 		if (connection->agreed_pro_version < 96 ?
3361 		    (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3362 		    (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3363 		    peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3364 			/* The last P_SYNC_UUID did not get though. Undo the last start of
3365 			   resync as sync source modifications of the peer's UUIDs. */
3366 
3367 			if (connection->agreed_pro_version < 91)
3368 				return -1091;
3369 
3370 			device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3371 			device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3372 
3373 			drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3374 			drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3375 
3376 			return -1;
3377 		}
3378 	}
3379 
3380 	*rule_nr = 60;
3381 	self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3382 	for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3383 		peer = device->p_uuid[i] & ~((u64)1);
3384 		if (self == peer)
3385 			return -2;
3386 	}
3387 
3388 	*rule_nr = 70;
3389 	self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3390 	peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3391 	if (self == peer)
3392 		return 1;
3393 
3394 	*rule_nr = 71;
3395 	self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3396 	if (self == peer) {
3397 		if (connection->agreed_pro_version < 96 ?
3398 		    (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3399 		    (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3400 		    self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3401 			/* The last P_SYNC_UUID did not get though. Undo the last start of
3402 			   resync as sync source modifications of our UUIDs. */
3403 
3404 			if (connection->agreed_pro_version < 91)
3405 				return -1091;
3406 
3407 			__drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3408 			__drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3409 
3410 			drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3411 			drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3412 				       device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3413 
3414 			return 1;
3415 		}
3416 	}
3417 
3418 
3419 	*rule_nr = 80;
3420 	peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3421 	for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3422 		self = device->ldev->md.uuid[i] & ~((u64)1);
3423 		if (self == peer)
3424 			return 2;
3425 	}
3426 
3427 	*rule_nr = 90;
3428 	self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3429 	peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3430 	if (self == peer && self != ((u64)0))
3431 		return 100;
3432 
3433 	*rule_nr = 100;
3434 	for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3435 		self = device->ldev->md.uuid[i] & ~((u64)1);
3436 		for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3437 			peer = device->p_uuid[j] & ~((u64)1);
3438 			if (self == peer)
3439 				return -100;
3440 		}
3441 	}
3442 
3443 	return -1000;
3444 }
3445 
3446 /* drbd_sync_handshake() returns the new conn state on success, or
3447    CONN_MASK (-1) on failure.
3448  */
3449 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3450 					   enum drbd_role peer_role,
3451 					   enum drbd_disk_state peer_disk) __must_hold(local)
3452 {
3453 	struct drbd_device *device = peer_device->device;
3454 	enum drbd_conns rv = C_MASK;
3455 	enum drbd_disk_state mydisk;
3456 	struct net_conf *nc;
3457 	int hg, rule_nr, rr_conflict, tentative, always_asbp;
3458 
3459 	mydisk = device->state.disk;
3460 	if (mydisk == D_NEGOTIATING)
3461 		mydisk = device->new_state_tmp.disk;
3462 
3463 	drbd_info(device, "drbd_sync_handshake:\n");
3464 
3465 	spin_lock_irq(&device->ldev->md.uuid_lock);
3466 	drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3467 	drbd_uuid_dump(device, "peer", device->p_uuid,
3468 		       device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3469 
3470 	hg = drbd_uuid_compare(peer_device, peer_role, &rule_nr);
3471 	spin_unlock_irq(&device->ldev->md.uuid_lock);
3472 
3473 	drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3474 
3475 	if (hg == -1000) {
3476 		drbd_alert(device, "Unrelated data, aborting!\n");
3477 		return C_MASK;
3478 	}
3479 	if (hg < -0x10000) {
3480 		int proto, fflags;
3481 		hg = -hg;
3482 		proto = hg & 0xff;
3483 		fflags = (hg >> 8) & 0xff;
3484 		drbd_alert(device, "To resolve this both sides have to support at least protocol %d and feature flags 0x%x\n",
3485 					proto, fflags);
3486 		return C_MASK;
3487 	}
3488 	if (hg < -1000) {
3489 		drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3490 		return C_MASK;
3491 	}
3492 
3493 	if    ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3494 	    (peer_disk == D_INCONSISTENT && mydisk    > D_INCONSISTENT)) {
3495 		int f = (hg == -100) || abs(hg) == 2;
3496 		hg = mydisk > D_INCONSISTENT ? 1 : -1;
3497 		if (f)
3498 			hg = hg*2;
3499 		drbd_info(device, "Becoming sync %s due to disk states.\n",
3500 		     hg > 0 ? "source" : "target");
3501 	}
3502 
3503 	if (abs(hg) == 100)
3504 		drbd_khelper(device, "initial-split-brain");
3505 
3506 	rcu_read_lock();
3507 	nc = rcu_dereference(peer_device->connection->net_conf);
3508 	always_asbp = nc->always_asbp;
3509 	rr_conflict = nc->rr_conflict;
3510 	tentative = nc->tentative;
3511 	rcu_read_unlock();
3512 
3513 	if (hg == 100 || (hg == -100 && always_asbp)) {
3514 		int pcount = (device->state.role == R_PRIMARY)
3515 			   + (peer_role == R_PRIMARY);
3516 		int forced = (hg == -100);
3517 
3518 		switch (pcount) {
3519 		case 0:
3520 			hg = drbd_asb_recover_0p(peer_device);
3521 			break;
3522 		case 1:
3523 			hg = drbd_asb_recover_1p(peer_device);
3524 			break;
3525 		case 2:
3526 			hg = drbd_asb_recover_2p(peer_device);
3527 			break;
3528 		}
3529 		if (abs(hg) < 100) {
3530 			drbd_warn(device, "Split-Brain detected, %d primaries, "
3531 			     "automatically solved. Sync from %s node\n",
3532 			     pcount, (hg < 0) ? "peer" : "this");
3533 			if (forced) {
3534 				drbd_warn(device, "Doing a full sync, since"
3535 				     " UUIDs where ambiguous.\n");
3536 				hg = hg*2;
3537 			}
3538 		}
3539 	}
3540 
3541 	if (hg == -100) {
3542 		if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3543 			hg = -1;
3544 		if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3545 			hg = 1;
3546 
3547 		if (abs(hg) < 100)
3548 			drbd_warn(device, "Split-Brain detected, manually solved. "
3549 			     "Sync from %s node\n",
3550 			     (hg < 0) ? "peer" : "this");
3551 	}
3552 
3553 	if (hg == -100) {
3554 		/* FIXME this log message is not correct if we end up here
3555 		 * after an attempted attach on a diskless node.
3556 		 * We just refuse to attach -- well, we drop the "connection"
3557 		 * to that disk, in a way... */
3558 		drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3559 		drbd_khelper(device, "split-brain");
3560 		return C_MASK;
3561 	}
3562 
3563 	if (hg > 0 && mydisk <= D_INCONSISTENT) {
3564 		drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3565 		return C_MASK;
3566 	}
3567 
3568 	if (hg < 0 && /* by intention we do not use mydisk here. */
3569 	    device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3570 		switch (rr_conflict) {
3571 		case ASB_CALL_HELPER:
3572 			drbd_khelper(device, "pri-lost");
3573 			fallthrough;
3574 		case ASB_DISCONNECT:
3575 			drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3576 			return C_MASK;
3577 		case ASB_VIOLENTLY:
3578 			drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3579 			     "assumption\n");
3580 		}
3581 	}
3582 
3583 	if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3584 		if (hg == 0)
3585 			drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3586 		else
3587 			drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3588 				 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3589 				 abs(hg) >= 2 ? "full" : "bit-map based");
3590 		return C_MASK;
3591 	}
3592 
3593 	if (abs(hg) >= 2) {
3594 		drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3595 		if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3596 					BM_LOCKED_SET_ALLOWED, NULL))
3597 			return C_MASK;
3598 	}
3599 
3600 	if (hg > 0) { /* become sync source. */
3601 		rv = C_WF_BITMAP_S;
3602 	} else if (hg < 0) { /* become sync target */
3603 		rv = C_WF_BITMAP_T;
3604 	} else {
3605 		rv = C_CONNECTED;
3606 		if (drbd_bm_total_weight(device)) {
3607 			drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3608 			     drbd_bm_total_weight(device));
3609 		}
3610 	}
3611 
3612 	return rv;
3613 }
3614 
3615 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3616 {
3617 	/* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3618 	if (peer == ASB_DISCARD_REMOTE)
3619 		return ASB_DISCARD_LOCAL;
3620 
3621 	/* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3622 	if (peer == ASB_DISCARD_LOCAL)
3623 		return ASB_DISCARD_REMOTE;
3624 
3625 	/* everything else is valid if they are equal on both sides. */
3626 	return peer;
3627 }
3628 
3629 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3630 {
3631 	struct p_protocol *p = pi->data;
3632 	enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3633 	int p_proto, p_discard_my_data, p_two_primaries, cf;
3634 	struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3635 	char integrity_alg[SHARED_SECRET_MAX] = "";
3636 	struct crypto_shash *peer_integrity_tfm = NULL;
3637 	void *int_dig_in = NULL, *int_dig_vv = NULL;
3638 
3639 	p_proto		= be32_to_cpu(p->protocol);
3640 	p_after_sb_0p	= be32_to_cpu(p->after_sb_0p);
3641 	p_after_sb_1p	= be32_to_cpu(p->after_sb_1p);
3642 	p_after_sb_2p	= be32_to_cpu(p->after_sb_2p);
3643 	p_two_primaries = be32_to_cpu(p->two_primaries);
3644 	cf		= be32_to_cpu(p->conn_flags);
3645 	p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3646 
3647 	if (connection->agreed_pro_version >= 87) {
3648 		int err;
3649 
3650 		if (pi->size > sizeof(integrity_alg))
3651 			return -EIO;
3652 		err = drbd_recv_all(connection, integrity_alg, pi->size);
3653 		if (err)
3654 			return err;
3655 		integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3656 	}
3657 
3658 	if (pi->cmd != P_PROTOCOL_UPDATE) {
3659 		clear_bit(CONN_DRY_RUN, &connection->flags);
3660 
3661 		if (cf & CF_DRY_RUN)
3662 			set_bit(CONN_DRY_RUN, &connection->flags);
3663 
3664 		rcu_read_lock();
3665 		nc = rcu_dereference(connection->net_conf);
3666 
3667 		if (p_proto != nc->wire_protocol) {
3668 			drbd_err(connection, "incompatible %s settings\n", "protocol");
3669 			goto disconnect_rcu_unlock;
3670 		}
3671 
3672 		if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3673 			drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3674 			goto disconnect_rcu_unlock;
3675 		}
3676 
3677 		if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3678 			drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3679 			goto disconnect_rcu_unlock;
3680 		}
3681 
3682 		if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3683 			drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3684 			goto disconnect_rcu_unlock;
3685 		}
3686 
3687 		if (p_discard_my_data && nc->discard_my_data) {
3688 			drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3689 			goto disconnect_rcu_unlock;
3690 		}
3691 
3692 		if (p_two_primaries != nc->two_primaries) {
3693 			drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3694 			goto disconnect_rcu_unlock;
3695 		}
3696 
3697 		if (strcmp(integrity_alg, nc->integrity_alg)) {
3698 			drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3699 			goto disconnect_rcu_unlock;
3700 		}
3701 
3702 		rcu_read_unlock();
3703 	}
3704 
3705 	if (integrity_alg[0]) {
3706 		int hash_size;
3707 
3708 		/*
3709 		 * We can only change the peer data integrity algorithm
3710 		 * here.  Changing our own data integrity algorithm
3711 		 * requires that we send a P_PROTOCOL_UPDATE packet at
3712 		 * the same time; otherwise, the peer has no way to
3713 		 * tell between which packets the algorithm should
3714 		 * change.
3715 		 */
3716 
3717 		peer_integrity_tfm = crypto_alloc_shash(integrity_alg, 0, 0);
3718 		if (IS_ERR(peer_integrity_tfm)) {
3719 			peer_integrity_tfm = NULL;
3720 			drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3721 				 integrity_alg);
3722 			goto disconnect;
3723 		}
3724 
3725 		hash_size = crypto_shash_digestsize(peer_integrity_tfm);
3726 		int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3727 		int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3728 		if (!(int_dig_in && int_dig_vv)) {
3729 			drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3730 			goto disconnect;
3731 		}
3732 	}
3733 
3734 	new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3735 	if (!new_net_conf)
3736 		goto disconnect;
3737 
3738 	mutex_lock(&connection->data.mutex);
3739 	mutex_lock(&connection->resource->conf_update);
3740 	old_net_conf = connection->net_conf;
3741 	*new_net_conf = *old_net_conf;
3742 
3743 	new_net_conf->wire_protocol = p_proto;
3744 	new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3745 	new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3746 	new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3747 	new_net_conf->two_primaries = p_two_primaries;
3748 
3749 	rcu_assign_pointer(connection->net_conf, new_net_conf);
3750 	mutex_unlock(&connection->resource->conf_update);
3751 	mutex_unlock(&connection->data.mutex);
3752 
3753 	crypto_free_shash(connection->peer_integrity_tfm);
3754 	kfree(connection->int_dig_in);
3755 	kfree(connection->int_dig_vv);
3756 	connection->peer_integrity_tfm = peer_integrity_tfm;
3757 	connection->int_dig_in = int_dig_in;
3758 	connection->int_dig_vv = int_dig_vv;
3759 
3760 	if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3761 		drbd_info(connection, "peer data-integrity-alg: %s\n",
3762 			  integrity_alg[0] ? integrity_alg : "(none)");
3763 
3764 	kvfree_rcu_mightsleep(old_net_conf);
3765 	return 0;
3766 
3767 disconnect_rcu_unlock:
3768 	rcu_read_unlock();
3769 disconnect:
3770 	crypto_free_shash(peer_integrity_tfm);
3771 	kfree(int_dig_in);
3772 	kfree(int_dig_vv);
3773 	conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3774 	return -EIO;
3775 }
3776 
3777 /* helper function
3778  * input: alg name, feature name
3779  * return: NULL (alg name was "")
3780  *         ERR_PTR(error) if something goes wrong
3781  *         or the crypto hash ptr, if it worked out ok. */
3782 static struct crypto_shash *drbd_crypto_alloc_digest_safe(
3783 		const struct drbd_device *device,
3784 		const char *alg, const char *name)
3785 {
3786 	struct crypto_shash *tfm;
3787 
3788 	if (!alg[0])
3789 		return NULL;
3790 
3791 	tfm = crypto_alloc_shash(alg, 0, 0);
3792 	if (IS_ERR(tfm)) {
3793 		drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3794 			alg, name, PTR_ERR(tfm));
3795 		return tfm;
3796 	}
3797 	return tfm;
3798 }
3799 
3800 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3801 {
3802 	void *buffer = connection->data.rbuf;
3803 	int size = pi->size;
3804 
3805 	while (size) {
3806 		int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3807 		s = drbd_recv(connection, buffer, s);
3808 		if (s <= 0) {
3809 			if (s < 0)
3810 				return s;
3811 			break;
3812 		}
3813 		size -= s;
3814 	}
3815 	if (size)
3816 		return -EIO;
3817 	return 0;
3818 }
3819 
3820 /*
3821  * config_unknown_volume  -  device configuration command for unknown volume
3822  *
3823  * When a device is added to an existing connection, the node on which the
3824  * device is added first will send configuration commands to its peer but the
3825  * peer will not know about the device yet.  It will warn and ignore these
3826  * commands.  Once the device is added on the second node, the second node will
3827  * send the same device configuration commands, but in the other direction.
3828  *
3829  * (We can also end up here if drbd is misconfigured.)
3830  */
3831 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3832 {
3833 	drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3834 		  cmdname(pi->cmd), pi->vnr);
3835 	return ignore_remaining_packet(connection, pi);
3836 }
3837 
3838 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3839 {
3840 	struct drbd_peer_device *peer_device;
3841 	struct drbd_device *device;
3842 	struct p_rs_param_95 *p;
3843 	unsigned int header_size, data_size, exp_max_sz;
3844 	struct crypto_shash *verify_tfm = NULL;
3845 	struct crypto_shash *csums_tfm = NULL;
3846 	struct net_conf *old_net_conf, *new_net_conf = NULL;
3847 	struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3848 	const int apv = connection->agreed_pro_version;
3849 	struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3850 	unsigned int fifo_size = 0;
3851 	int err;
3852 
3853 	peer_device = conn_peer_device(connection, pi->vnr);
3854 	if (!peer_device)
3855 		return config_unknown_volume(connection, pi);
3856 	device = peer_device->device;
3857 
3858 	exp_max_sz  = apv <= 87 ? sizeof(struct p_rs_param)
3859 		    : apv == 88 ? sizeof(struct p_rs_param)
3860 					+ SHARED_SECRET_MAX
3861 		    : apv <= 94 ? sizeof(struct p_rs_param_89)
3862 		    : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3863 
3864 	if (pi->size > exp_max_sz) {
3865 		drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3866 		    pi->size, exp_max_sz);
3867 		return -EIO;
3868 	}
3869 
3870 	if (apv <= 88) {
3871 		header_size = sizeof(struct p_rs_param);
3872 		data_size = pi->size - header_size;
3873 	} else if (apv <= 94) {
3874 		header_size = sizeof(struct p_rs_param_89);
3875 		data_size = pi->size - header_size;
3876 		D_ASSERT(device, data_size == 0);
3877 	} else {
3878 		header_size = sizeof(struct p_rs_param_95);
3879 		data_size = pi->size - header_size;
3880 		D_ASSERT(device, data_size == 0);
3881 	}
3882 
3883 	/* initialize verify_alg and csums_alg */
3884 	p = pi->data;
3885 	BUILD_BUG_ON(sizeof(p->algs) != 2 * SHARED_SECRET_MAX);
3886 	memset(&p->algs, 0, sizeof(p->algs));
3887 
3888 	err = drbd_recv_all(peer_device->connection, p, header_size);
3889 	if (err)
3890 		return err;
3891 
3892 	mutex_lock(&connection->resource->conf_update);
3893 	old_net_conf = peer_device->connection->net_conf;
3894 	if (get_ldev(device)) {
3895 		new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3896 		if (!new_disk_conf) {
3897 			put_ldev(device);
3898 			mutex_unlock(&connection->resource->conf_update);
3899 			drbd_err(device, "Allocation of new disk_conf failed\n");
3900 			return -ENOMEM;
3901 		}
3902 
3903 		old_disk_conf = device->ldev->disk_conf;
3904 		*new_disk_conf = *old_disk_conf;
3905 
3906 		new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3907 	}
3908 
3909 	if (apv >= 88) {
3910 		if (apv == 88) {
3911 			if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3912 				drbd_err(device, "verify-alg of wrong size, "
3913 					"peer wants %u, accepting only up to %u byte\n",
3914 					data_size, SHARED_SECRET_MAX);
3915 				goto reconnect;
3916 			}
3917 
3918 			err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3919 			if (err)
3920 				goto reconnect;
3921 			/* we expect NUL terminated string */
3922 			/* but just in case someone tries to be evil */
3923 			D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3924 			p->verify_alg[data_size-1] = 0;
3925 
3926 		} else /* apv >= 89 */ {
3927 			/* we still expect NUL terminated strings */
3928 			/* but just in case someone tries to be evil */
3929 			D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3930 			D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3931 			p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3932 			p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3933 		}
3934 
3935 		if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3936 			if (device->state.conn == C_WF_REPORT_PARAMS) {
3937 				drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3938 				    old_net_conf->verify_alg, p->verify_alg);
3939 				goto disconnect;
3940 			}
3941 			verify_tfm = drbd_crypto_alloc_digest_safe(device,
3942 					p->verify_alg, "verify-alg");
3943 			if (IS_ERR(verify_tfm)) {
3944 				verify_tfm = NULL;
3945 				goto disconnect;
3946 			}
3947 		}
3948 
3949 		if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3950 			if (device->state.conn == C_WF_REPORT_PARAMS) {
3951 				drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3952 				    old_net_conf->csums_alg, p->csums_alg);
3953 				goto disconnect;
3954 			}
3955 			csums_tfm = drbd_crypto_alloc_digest_safe(device,
3956 					p->csums_alg, "csums-alg");
3957 			if (IS_ERR(csums_tfm)) {
3958 				csums_tfm = NULL;
3959 				goto disconnect;
3960 			}
3961 		}
3962 
3963 		if (apv > 94 && new_disk_conf) {
3964 			new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3965 			new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3966 			new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3967 			new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3968 
3969 			fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3970 			if (fifo_size != device->rs_plan_s->size) {
3971 				new_plan = fifo_alloc(fifo_size);
3972 				if (!new_plan) {
3973 					drbd_err(device, "kmalloc of fifo_buffer failed");
3974 					put_ldev(device);
3975 					goto disconnect;
3976 				}
3977 			}
3978 		}
3979 
3980 		if (verify_tfm || csums_tfm) {
3981 			new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3982 			if (!new_net_conf)
3983 				goto disconnect;
3984 
3985 			*new_net_conf = *old_net_conf;
3986 
3987 			if (verify_tfm) {
3988 				strcpy(new_net_conf->verify_alg, p->verify_alg);
3989 				new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3990 				crypto_free_shash(peer_device->connection->verify_tfm);
3991 				peer_device->connection->verify_tfm = verify_tfm;
3992 				drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3993 			}
3994 			if (csums_tfm) {
3995 				strcpy(new_net_conf->csums_alg, p->csums_alg);
3996 				new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3997 				crypto_free_shash(peer_device->connection->csums_tfm);
3998 				peer_device->connection->csums_tfm = csums_tfm;
3999 				drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
4000 			}
4001 			rcu_assign_pointer(connection->net_conf, new_net_conf);
4002 		}
4003 	}
4004 
4005 	if (new_disk_conf) {
4006 		rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4007 		put_ldev(device);
4008 	}
4009 
4010 	if (new_plan) {
4011 		old_plan = device->rs_plan_s;
4012 		rcu_assign_pointer(device->rs_plan_s, new_plan);
4013 	}
4014 
4015 	mutex_unlock(&connection->resource->conf_update);
4016 	synchronize_rcu();
4017 	if (new_net_conf)
4018 		kfree(old_net_conf);
4019 	kfree(old_disk_conf);
4020 	kfree(old_plan);
4021 
4022 	return 0;
4023 
4024 reconnect:
4025 	if (new_disk_conf) {
4026 		put_ldev(device);
4027 		kfree(new_disk_conf);
4028 	}
4029 	mutex_unlock(&connection->resource->conf_update);
4030 	return -EIO;
4031 
4032 disconnect:
4033 	kfree(new_plan);
4034 	if (new_disk_conf) {
4035 		put_ldev(device);
4036 		kfree(new_disk_conf);
4037 	}
4038 	mutex_unlock(&connection->resource->conf_update);
4039 	/* just for completeness: actually not needed,
4040 	 * as this is not reached if csums_tfm was ok. */
4041 	crypto_free_shash(csums_tfm);
4042 	/* but free the verify_tfm again, if csums_tfm did not work out */
4043 	crypto_free_shash(verify_tfm);
4044 	conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4045 	return -EIO;
4046 }
4047 
4048 /* warn if the arguments differ by more than 12.5% */
4049 static void warn_if_differ_considerably(struct drbd_device *device,
4050 	const char *s, sector_t a, sector_t b)
4051 {
4052 	sector_t d;
4053 	if (a == 0 || b == 0)
4054 		return;
4055 	d = (a > b) ? (a - b) : (b - a);
4056 	if (d > (a>>3) || d > (b>>3))
4057 		drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
4058 		     (unsigned long long)a, (unsigned long long)b);
4059 }
4060 
4061 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
4062 {
4063 	struct drbd_peer_device *peer_device;
4064 	struct drbd_device *device;
4065 	struct p_sizes *p = pi->data;
4066 	struct o_qlim *o = (connection->agreed_features & DRBD_FF_WSAME) ? p->qlim : NULL;
4067 	enum determine_dev_size dd = DS_UNCHANGED;
4068 	sector_t p_size, p_usize, p_csize, my_usize;
4069 	sector_t new_size, cur_size;
4070 	int ldsc = 0; /* local disk size changed */
4071 	enum dds_flags ddsf;
4072 
4073 	peer_device = conn_peer_device(connection, pi->vnr);
4074 	if (!peer_device)
4075 		return config_unknown_volume(connection, pi);
4076 	device = peer_device->device;
4077 	cur_size = get_capacity(device->vdisk);
4078 
4079 	p_size = be64_to_cpu(p->d_size);
4080 	p_usize = be64_to_cpu(p->u_size);
4081 	p_csize = be64_to_cpu(p->c_size);
4082 
4083 	/* just store the peer's disk size for now.
4084 	 * we still need to figure out whether we accept that. */
4085 	device->p_size = p_size;
4086 
4087 	if (get_ldev(device)) {
4088 		rcu_read_lock();
4089 		my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
4090 		rcu_read_unlock();
4091 
4092 		warn_if_differ_considerably(device, "lower level device sizes",
4093 			   p_size, drbd_get_max_capacity(device->ldev));
4094 		warn_if_differ_considerably(device, "user requested size",
4095 					    p_usize, my_usize);
4096 
4097 		/* if this is the first connect, or an otherwise expected
4098 		 * param exchange, choose the minimum */
4099 		if (device->state.conn == C_WF_REPORT_PARAMS)
4100 			p_usize = min_not_zero(my_usize, p_usize);
4101 
4102 		/* Never shrink a device with usable data during connect,
4103 		 * or "attach" on the peer.
4104 		 * But allow online shrinking if we are connected. */
4105 		new_size = drbd_new_dev_size(device, device->ldev, p_usize, 0);
4106 		if (new_size < cur_size &&
4107 		    device->state.disk >= D_OUTDATED &&
4108 		    (device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS)) {
4109 			drbd_err(device, "The peer's disk size is too small! (%llu < %llu sectors)\n",
4110 					(unsigned long long)new_size, (unsigned long long)cur_size);
4111 			conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4112 			put_ldev(device);
4113 			return -EIO;
4114 		}
4115 
4116 		if (my_usize != p_usize) {
4117 			struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
4118 
4119 			new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
4120 			if (!new_disk_conf) {
4121 				put_ldev(device);
4122 				return -ENOMEM;
4123 			}
4124 
4125 			mutex_lock(&connection->resource->conf_update);
4126 			old_disk_conf = device->ldev->disk_conf;
4127 			*new_disk_conf = *old_disk_conf;
4128 			new_disk_conf->disk_size = p_usize;
4129 
4130 			rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4131 			mutex_unlock(&connection->resource->conf_update);
4132 			kvfree_rcu_mightsleep(old_disk_conf);
4133 
4134 			drbd_info(device, "Peer sets u_size to %lu sectors (old: %lu)\n",
4135 				 (unsigned long)p_usize, (unsigned long)my_usize);
4136 		}
4137 
4138 		put_ldev(device);
4139 	}
4140 
4141 	device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
4142 	/* Leave drbd_reconsider_queue_parameters() before drbd_determine_dev_size().
4143 	   In case we cleared the QUEUE_FLAG_DISCARD from our queue in
4144 	   drbd_reconsider_queue_parameters(), we can be sure that after
4145 	   drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
4146 
4147 	ddsf = be16_to_cpu(p->dds_flags);
4148 	if (get_ldev(device)) {
4149 		drbd_reconsider_queue_parameters(device, device->ldev, o);
4150 		dd = drbd_determine_dev_size(device, ddsf, NULL);
4151 		put_ldev(device);
4152 		if (dd == DS_ERROR)
4153 			return -EIO;
4154 		drbd_md_sync(device);
4155 	} else {
4156 		/*
4157 		 * I am diskless, need to accept the peer's *current* size.
4158 		 * I must NOT accept the peers backing disk size,
4159 		 * it may have been larger than mine all along...
4160 		 *
4161 		 * At this point, the peer knows more about my disk, or at
4162 		 * least about what we last agreed upon, than myself.
4163 		 * So if his c_size is less than his d_size, the most likely
4164 		 * reason is that *my* d_size was smaller last time we checked.
4165 		 *
4166 		 * However, if he sends a zero current size,
4167 		 * take his (user-capped or) backing disk size anyways.
4168 		 *
4169 		 * Unless of course he does not have a disk himself.
4170 		 * In which case we ignore this completely.
4171 		 */
4172 		sector_t new_size = p_csize ?: p_usize ?: p_size;
4173 		drbd_reconsider_queue_parameters(device, NULL, o);
4174 		if (new_size == 0) {
4175 			/* Ignore, peer does not know nothing. */
4176 		} else if (new_size == cur_size) {
4177 			/* nothing to do */
4178 		} else if (cur_size != 0 && p_size == 0) {
4179 			drbd_warn(device, "Ignored diskless peer device size (peer:%llu != me:%llu sectors)!\n",
4180 					(unsigned long long)new_size, (unsigned long long)cur_size);
4181 		} else if (new_size < cur_size && device->state.role == R_PRIMARY) {
4182 			drbd_err(device, "The peer's device size is too small! (%llu < %llu sectors); demote me first!\n",
4183 					(unsigned long long)new_size, (unsigned long long)cur_size);
4184 			conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4185 			return -EIO;
4186 		} else {
4187 			/* I believe the peer, if
4188 			 *  - I don't have a current size myself
4189 			 *  - we agree on the size anyways
4190 			 *  - I do have a current size, am Secondary,
4191 			 *    and he has the only disk
4192 			 *  - I do have a current size, am Primary,
4193 			 *    and he has the only disk,
4194 			 *    which is larger than my current size
4195 			 */
4196 			drbd_set_my_capacity(device, new_size);
4197 		}
4198 	}
4199 
4200 	if (get_ldev(device)) {
4201 		if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
4202 			device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
4203 			ldsc = 1;
4204 		}
4205 
4206 		put_ldev(device);
4207 	}
4208 
4209 	if (device->state.conn > C_WF_REPORT_PARAMS) {
4210 		if (be64_to_cpu(p->c_size) != get_capacity(device->vdisk) ||
4211 		    ldsc) {
4212 			/* we have different sizes, probably peer
4213 			 * needs to know my new size... */
4214 			drbd_send_sizes(peer_device, 0, ddsf);
4215 		}
4216 		if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
4217 		    (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
4218 			if (device->state.pdsk >= D_INCONSISTENT &&
4219 			    device->state.disk >= D_INCONSISTENT) {
4220 				if (ddsf & DDSF_NO_RESYNC)
4221 					drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
4222 				else
4223 					resync_after_online_grow(device);
4224 			} else
4225 				set_bit(RESYNC_AFTER_NEG, &device->flags);
4226 		}
4227 	}
4228 
4229 	return 0;
4230 }
4231 
4232 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
4233 {
4234 	struct drbd_peer_device *peer_device;
4235 	struct drbd_device *device;
4236 	struct p_uuids *p = pi->data;
4237 	u64 *p_uuid;
4238 	int i, updated_uuids = 0;
4239 
4240 	peer_device = conn_peer_device(connection, pi->vnr);
4241 	if (!peer_device)
4242 		return config_unknown_volume(connection, pi);
4243 	device = peer_device->device;
4244 
4245 	p_uuid = kmalloc_array(UI_EXTENDED_SIZE, sizeof(*p_uuid), GFP_NOIO);
4246 	if (!p_uuid)
4247 		return false;
4248 
4249 	for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
4250 		p_uuid[i] = be64_to_cpu(p->uuid[i]);
4251 
4252 	kfree(device->p_uuid);
4253 	device->p_uuid = p_uuid;
4254 
4255 	if ((device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS) &&
4256 	    device->state.disk < D_INCONSISTENT &&
4257 	    device->state.role == R_PRIMARY &&
4258 	    (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
4259 		drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
4260 		    (unsigned long long)device->ed_uuid);
4261 		conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4262 		return -EIO;
4263 	}
4264 
4265 	if (get_ldev(device)) {
4266 		int skip_initial_sync =
4267 			device->state.conn == C_CONNECTED &&
4268 			peer_device->connection->agreed_pro_version >= 90 &&
4269 			device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
4270 			(p_uuid[UI_FLAGS] & 8);
4271 		if (skip_initial_sync) {
4272 			drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
4273 			drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
4274 					"clear_n_write from receive_uuids",
4275 					BM_LOCKED_TEST_ALLOWED, NULL);
4276 			_drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
4277 			_drbd_uuid_set(device, UI_BITMAP, 0);
4278 			_drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
4279 					CS_VERBOSE, NULL);
4280 			drbd_md_sync(device);
4281 			updated_uuids = 1;
4282 		}
4283 		put_ldev(device);
4284 	} else if (device->state.disk < D_INCONSISTENT &&
4285 		   device->state.role == R_PRIMARY) {
4286 		/* I am a diskless primary, the peer just created a new current UUID
4287 		   for me. */
4288 		updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4289 	}
4290 
4291 	/* Before we test for the disk state, we should wait until an eventually
4292 	   ongoing cluster wide state change is finished. That is important if
4293 	   we are primary and are detaching from our disk. We need to see the
4294 	   new disk state... */
4295 	mutex_lock(device->state_mutex);
4296 	mutex_unlock(device->state_mutex);
4297 	if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
4298 		updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4299 
4300 	if (updated_uuids)
4301 		drbd_print_uuids(device, "receiver updated UUIDs to");
4302 
4303 	return 0;
4304 }
4305 
4306 /**
4307  * convert_state() - Converts the peer's view of the cluster state to our point of view
4308  * @ps:		The state as seen by the peer.
4309  */
4310 static union drbd_state convert_state(union drbd_state ps)
4311 {
4312 	union drbd_state ms;
4313 
4314 	static enum drbd_conns c_tab[] = {
4315 		[C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
4316 		[C_CONNECTED] = C_CONNECTED,
4317 
4318 		[C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
4319 		[C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
4320 		[C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
4321 		[C_VERIFY_S]       = C_VERIFY_T,
4322 		[C_MASK]   = C_MASK,
4323 	};
4324 
4325 	ms.i = ps.i;
4326 
4327 	ms.conn = c_tab[ps.conn];
4328 	ms.peer = ps.role;
4329 	ms.role = ps.peer;
4330 	ms.pdsk = ps.disk;
4331 	ms.disk = ps.pdsk;
4332 	ms.peer_isp = (ps.aftr_isp | ps.user_isp);
4333 
4334 	return ms;
4335 }
4336 
4337 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4338 {
4339 	struct drbd_peer_device *peer_device;
4340 	struct drbd_device *device;
4341 	struct p_req_state *p = pi->data;
4342 	union drbd_state mask, val;
4343 	enum drbd_state_rv rv;
4344 
4345 	peer_device = conn_peer_device(connection, pi->vnr);
4346 	if (!peer_device)
4347 		return -EIO;
4348 	device = peer_device->device;
4349 
4350 	mask.i = be32_to_cpu(p->mask);
4351 	val.i = be32_to_cpu(p->val);
4352 
4353 	if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4354 	    mutex_is_locked(device->state_mutex)) {
4355 		drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4356 		return 0;
4357 	}
4358 
4359 	mask = convert_state(mask);
4360 	val = convert_state(val);
4361 
4362 	rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4363 	drbd_send_sr_reply(peer_device, rv);
4364 
4365 	drbd_md_sync(device);
4366 
4367 	return 0;
4368 }
4369 
4370 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4371 {
4372 	struct p_req_state *p = pi->data;
4373 	union drbd_state mask, val;
4374 	enum drbd_state_rv rv;
4375 
4376 	mask.i = be32_to_cpu(p->mask);
4377 	val.i = be32_to_cpu(p->val);
4378 
4379 	if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4380 	    mutex_is_locked(&connection->cstate_mutex)) {
4381 		conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4382 		return 0;
4383 	}
4384 
4385 	mask = convert_state(mask);
4386 	val = convert_state(val);
4387 
4388 	rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4389 	conn_send_sr_reply(connection, rv);
4390 
4391 	return 0;
4392 }
4393 
4394 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4395 {
4396 	struct drbd_peer_device *peer_device;
4397 	struct drbd_device *device;
4398 	struct p_state *p = pi->data;
4399 	union drbd_state os, ns, peer_state;
4400 	enum drbd_disk_state real_peer_disk;
4401 	enum chg_state_flags cs_flags;
4402 	int rv;
4403 
4404 	peer_device = conn_peer_device(connection, pi->vnr);
4405 	if (!peer_device)
4406 		return config_unknown_volume(connection, pi);
4407 	device = peer_device->device;
4408 
4409 	peer_state.i = be32_to_cpu(p->state);
4410 
4411 	real_peer_disk = peer_state.disk;
4412 	if (peer_state.disk == D_NEGOTIATING) {
4413 		real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4414 		drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4415 	}
4416 
4417 	spin_lock_irq(&device->resource->req_lock);
4418  retry:
4419 	os = ns = drbd_read_state(device);
4420 	spin_unlock_irq(&device->resource->req_lock);
4421 
4422 	/* If some other part of the code (ack_receiver thread, timeout)
4423 	 * already decided to close the connection again,
4424 	 * we must not "re-establish" it here. */
4425 	if (os.conn <= C_TEAR_DOWN)
4426 		return -ECONNRESET;
4427 
4428 	/* If this is the "end of sync" confirmation, usually the peer disk
4429 	 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4430 	 * set) resync started in PausedSyncT, or if the timing of pause-/
4431 	 * unpause-sync events has been "just right", the peer disk may
4432 	 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4433 	 */
4434 	if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4435 	    real_peer_disk == D_UP_TO_DATE &&
4436 	    os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4437 		/* If we are (becoming) SyncSource, but peer is still in sync
4438 		 * preparation, ignore its uptodate-ness to avoid flapping, it
4439 		 * will change to inconsistent once the peer reaches active
4440 		 * syncing states.
4441 		 * It may have changed syncer-paused flags, however, so we
4442 		 * cannot ignore this completely. */
4443 		if (peer_state.conn > C_CONNECTED &&
4444 		    peer_state.conn < C_SYNC_SOURCE)
4445 			real_peer_disk = D_INCONSISTENT;
4446 
4447 		/* if peer_state changes to connected at the same time,
4448 		 * it explicitly notifies us that it finished resync.
4449 		 * Maybe we should finish it up, too? */
4450 		else if (os.conn >= C_SYNC_SOURCE &&
4451 			 peer_state.conn == C_CONNECTED) {
4452 			if (drbd_bm_total_weight(device) <= device->rs_failed)
4453 				drbd_resync_finished(peer_device);
4454 			return 0;
4455 		}
4456 	}
4457 
4458 	/* explicit verify finished notification, stop sector reached. */
4459 	if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4460 	    peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4461 		ov_out_of_sync_print(peer_device);
4462 		drbd_resync_finished(peer_device);
4463 		return 0;
4464 	}
4465 
4466 	/* peer says his disk is inconsistent, while we think it is uptodate,
4467 	 * and this happens while the peer still thinks we have a sync going on,
4468 	 * but we think we are already done with the sync.
4469 	 * We ignore this to avoid flapping pdsk.
4470 	 * This should not happen, if the peer is a recent version of drbd. */
4471 	if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4472 	    os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4473 		real_peer_disk = D_UP_TO_DATE;
4474 
4475 	if (ns.conn == C_WF_REPORT_PARAMS)
4476 		ns.conn = C_CONNECTED;
4477 
4478 	if (peer_state.conn == C_AHEAD)
4479 		ns.conn = C_BEHIND;
4480 
4481 	/* TODO:
4482 	 * if (primary and diskless and peer uuid != effective uuid)
4483 	 *     abort attach on peer;
4484 	 *
4485 	 * If this node does not have good data, was already connected, but
4486 	 * the peer did a late attach only now, trying to "negotiate" with me,
4487 	 * AND I am currently Primary, possibly frozen, with some specific
4488 	 * "effective" uuid, this should never be reached, really, because
4489 	 * we first send the uuids, then the current state.
4490 	 *
4491 	 * In this scenario, we already dropped the connection hard
4492 	 * when we received the unsuitable uuids (receive_uuids().
4493 	 *
4494 	 * Should we want to change this, that is: not drop the connection in
4495 	 * receive_uuids() already, then we would need to add a branch here
4496 	 * that aborts the attach of "unsuitable uuids" on the peer in case
4497 	 * this node is currently Diskless Primary.
4498 	 */
4499 
4500 	if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4501 	    get_ldev_if_state(device, D_NEGOTIATING)) {
4502 		int cr; /* consider resync */
4503 
4504 		/* if we established a new connection */
4505 		cr  = (os.conn < C_CONNECTED);
4506 		/* if we had an established connection
4507 		 * and one of the nodes newly attaches a disk */
4508 		cr |= (os.conn == C_CONNECTED &&
4509 		       (peer_state.disk == D_NEGOTIATING ||
4510 			os.disk == D_NEGOTIATING));
4511 		/* if we have both been inconsistent, and the peer has been
4512 		 * forced to be UpToDate with --force */
4513 		cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4514 		/* if we had been plain connected, and the admin requested to
4515 		 * start a sync by "invalidate" or "invalidate-remote" */
4516 		cr |= (os.conn == C_CONNECTED &&
4517 				(peer_state.conn >= C_STARTING_SYNC_S &&
4518 				 peer_state.conn <= C_WF_BITMAP_T));
4519 
4520 		if (cr)
4521 			ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4522 
4523 		put_ldev(device);
4524 		if (ns.conn == C_MASK) {
4525 			ns.conn = C_CONNECTED;
4526 			if (device->state.disk == D_NEGOTIATING) {
4527 				drbd_force_state(device, NS(disk, D_FAILED));
4528 			} else if (peer_state.disk == D_NEGOTIATING) {
4529 				drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4530 				peer_state.disk = D_DISKLESS;
4531 				real_peer_disk = D_DISKLESS;
4532 			} else {
4533 				if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4534 					return -EIO;
4535 				D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4536 				conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4537 				return -EIO;
4538 			}
4539 		}
4540 	}
4541 
4542 	spin_lock_irq(&device->resource->req_lock);
4543 	if (os.i != drbd_read_state(device).i)
4544 		goto retry;
4545 	clear_bit(CONSIDER_RESYNC, &device->flags);
4546 	ns.peer = peer_state.role;
4547 	ns.pdsk = real_peer_disk;
4548 	ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4549 	if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4550 		ns.disk = device->new_state_tmp.disk;
4551 	cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4552 	if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4553 	    test_bit(NEW_CUR_UUID, &device->flags)) {
4554 		/* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4555 		   for temporal network outages! */
4556 		spin_unlock_irq(&device->resource->req_lock);
4557 		drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4558 		tl_clear(peer_device->connection);
4559 		drbd_uuid_new_current(device);
4560 		clear_bit(NEW_CUR_UUID, &device->flags);
4561 		conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4562 		return -EIO;
4563 	}
4564 	rv = _drbd_set_state(device, ns, cs_flags, NULL);
4565 	ns = drbd_read_state(device);
4566 	spin_unlock_irq(&device->resource->req_lock);
4567 
4568 	if (rv < SS_SUCCESS) {
4569 		conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4570 		return -EIO;
4571 	}
4572 
4573 	if (os.conn > C_WF_REPORT_PARAMS) {
4574 		if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4575 		    peer_state.disk != D_NEGOTIATING ) {
4576 			/* we want resync, peer has not yet decided to sync... */
4577 			/* Nowadays only used when forcing a node into primary role and
4578 			   setting its disk to UpToDate with that */
4579 			drbd_send_uuids(peer_device);
4580 			drbd_send_current_state(peer_device);
4581 		}
4582 	}
4583 
4584 	clear_bit(DISCARD_MY_DATA, &device->flags);
4585 
4586 	drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4587 
4588 	return 0;
4589 }
4590 
4591 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4592 {
4593 	struct drbd_peer_device *peer_device;
4594 	struct drbd_device *device;
4595 	struct p_rs_uuid *p = pi->data;
4596 
4597 	peer_device = conn_peer_device(connection, pi->vnr);
4598 	if (!peer_device)
4599 		return -EIO;
4600 	device = peer_device->device;
4601 
4602 	wait_event(device->misc_wait,
4603 		   device->state.conn == C_WF_SYNC_UUID ||
4604 		   device->state.conn == C_BEHIND ||
4605 		   device->state.conn < C_CONNECTED ||
4606 		   device->state.disk < D_NEGOTIATING);
4607 
4608 	/* D_ASSERT(device,  device->state.conn == C_WF_SYNC_UUID ); */
4609 
4610 	/* Here the _drbd_uuid_ functions are right, current should
4611 	   _not_ be rotated into the history */
4612 	if (get_ldev_if_state(device, D_NEGOTIATING)) {
4613 		_drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4614 		_drbd_uuid_set(device, UI_BITMAP, 0UL);
4615 
4616 		drbd_print_uuids(device, "updated sync uuid");
4617 		drbd_start_resync(device, C_SYNC_TARGET);
4618 
4619 		put_ldev(device);
4620 	} else
4621 		drbd_err(device, "Ignoring SyncUUID packet!\n");
4622 
4623 	return 0;
4624 }
4625 
4626 /*
4627  * receive_bitmap_plain
4628  *
4629  * Return 0 when done, 1 when another iteration is needed, and a negative error
4630  * code upon failure.
4631  */
4632 static int
4633 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4634 		     unsigned long *p, struct bm_xfer_ctx *c)
4635 {
4636 	unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4637 				 drbd_header_size(peer_device->connection);
4638 	unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4639 				       c->bm_words - c->word_offset);
4640 	unsigned int want = num_words * sizeof(*p);
4641 	int err;
4642 
4643 	if (want != size) {
4644 		drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4645 		return -EIO;
4646 	}
4647 	if (want == 0)
4648 		return 0;
4649 	err = drbd_recv_all(peer_device->connection, p, want);
4650 	if (err)
4651 		return err;
4652 
4653 	drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4654 
4655 	c->word_offset += num_words;
4656 	c->bit_offset = c->word_offset * BITS_PER_LONG;
4657 	if (c->bit_offset > c->bm_bits)
4658 		c->bit_offset = c->bm_bits;
4659 
4660 	return 1;
4661 }
4662 
4663 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4664 {
4665 	return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4666 }
4667 
4668 static int dcbp_get_start(struct p_compressed_bm *p)
4669 {
4670 	return (p->encoding & 0x80) != 0;
4671 }
4672 
4673 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4674 {
4675 	return (p->encoding >> 4) & 0x7;
4676 }
4677 
4678 /*
4679  * recv_bm_rle_bits
4680  *
4681  * Return 0 when done, 1 when another iteration is needed, and a negative error
4682  * code upon failure.
4683  */
4684 static int
4685 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4686 		struct p_compressed_bm *p,
4687 		 struct bm_xfer_ctx *c,
4688 		 unsigned int len)
4689 {
4690 	struct bitstream bs;
4691 	u64 look_ahead;
4692 	u64 rl;
4693 	u64 tmp;
4694 	unsigned long s = c->bit_offset;
4695 	unsigned long e;
4696 	int toggle = dcbp_get_start(p);
4697 	int have;
4698 	int bits;
4699 
4700 	bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4701 
4702 	bits = bitstream_get_bits(&bs, &look_ahead, 64);
4703 	if (bits < 0)
4704 		return -EIO;
4705 
4706 	for (have = bits; have > 0; s += rl, toggle = !toggle) {
4707 		bits = vli_decode_bits(&rl, look_ahead);
4708 		if (bits <= 0)
4709 			return -EIO;
4710 
4711 		if (toggle) {
4712 			e = s + rl -1;
4713 			if (e >= c->bm_bits) {
4714 				drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4715 				return -EIO;
4716 			}
4717 			_drbd_bm_set_bits(peer_device->device, s, e);
4718 		}
4719 
4720 		if (have < bits) {
4721 			drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4722 				have, bits, look_ahead,
4723 				(unsigned int)(bs.cur.b - p->code),
4724 				(unsigned int)bs.buf_len);
4725 			return -EIO;
4726 		}
4727 		/* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4728 		if (likely(bits < 64))
4729 			look_ahead >>= bits;
4730 		else
4731 			look_ahead = 0;
4732 		have -= bits;
4733 
4734 		bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4735 		if (bits < 0)
4736 			return -EIO;
4737 		look_ahead |= tmp << have;
4738 		have += bits;
4739 	}
4740 
4741 	c->bit_offset = s;
4742 	bm_xfer_ctx_bit_to_word_offset(c);
4743 
4744 	return (s != c->bm_bits);
4745 }
4746 
4747 /*
4748  * decode_bitmap_c
4749  *
4750  * Return 0 when done, 1 when another iteration is needed, and a negative error
4751  * code upon failure.
4752  */
4753 static int
4754 decode_bitmap_c(struct drbd_peer_device *peer_device,
4755 		struct p_compressed_bm *p,
4756 		struct bm_xfer_ctx *c,
4757 		unsigned int len)
4758 {
4759 	if (dcbp_get_code(p) == RLE_VLI_Bits)
4760 		return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4761 
4762 	/* other variants had been implemented for evaluation,
4763 	 * but have been dropped as this one turned out to be "best"
4764 	 * during all our tests. */
4765 
4766 	drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4767 	conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4768 	return -EIO;
4769 }
4770 
4771 void INFO_bm_xfer_stats(struct drbd_peer_device *peer_device,
4772 		const char *direction, struct bm_xfer_ctx *c)
4773 {
4774 	/* what would it take to transfer it "plaintext" */
4775 	unsigned int header_size = drbd_header_size(peer_device->connection);
4776 	unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4777 	unsigned int plain =
4778 		header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4779 		c->bm_words * sizeof(unsigned long);
4780 	unsigned int total = c->bytes[0] + c->bytes[1];
4781 	unsigned int r;
4782 
4783 	/* total can not be zero. but just in case: */
4784 	if (total == 0)
4785 		return;
4786 
4787 	/* don't report if not compressed */
4788 	if (total >= plain)
4789 		return;
4790 
4791 	/* total < plain. check for overflow, still */
4792 	r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4793 		                    : (1000 * total / plain);
4794 
4795 	if (r > 1000)
4796 		r = 1000;
4797 
4798 	r = 1000 - r;
4799 	drbd_info(peer_device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4800 	     "total %u; compression: %u.%u%%\n",
4801 			direction,
4802 			c->bytes[1], c->packets[1],
4803 			c->bytes[0], c->packets[0],
4804 			total, r/10, r % 10);
4805 }
4806 
4807 /* Since we are processing the bitfield from lower addresses to higher,
4808    it does not matter if the process it in 32 bit chunks or 64 bit
4809    chunks as long as it is little endian. (Understand it as byte stream,
4810    beginning with the lowest byte...) If we would use big endian
4811    we would need to process it from the highest address to the lowest,
4812    in order to be agnostic to the 32 vs 64 bits issue.
4813 
4814    returns 0 on failure, 1 if we successfully received it. */
4815 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4816 {
4817 	struct drbd_peer_device *peer_device;
4818 	struct drbd_device *device;
4819 	struct bm_xfer_ctx c;
4820 	int err;
4821 
4822 	peer_device = conn_peer_device(connection, pi->vnr);
4823 	if (!peer_device)
4824 		return -EIO;
4825 	device = peer_device->device;
4826 
4827 	drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4828 	/* you are supposed to send additional out-of-sync information
4829 	 * if you actually set bits during this phase */
4830 
4831 	c = (struct bm_xfer_ctx) {
4832 		.bm_bits = drbd_bm_bits(device),
4833 		.bm_words = drbd_bm_words(device),
4834 	};
4835 
4836 	for(;;) {
4837 		if (pi->cmd == P_BITMAP)
4838 			err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4839 		else if (pi->cmd == P_COMPRESSED_BITMAP) {
4840 			/* MAYBE: sanity check that we speak proto >= 90,
4841 			 * and the feature is enabled! */
4842 			struct p_compressed_bm *p = pi->data;
4843 
4844 			if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4845 				drbd_err(device, "ReportCBitmap packet too large\n");
4846 				err = -EIO;
4847 				goto out;
4848 			}
4849 			if (pi->size <= sizeof(*p)) {
4850 				drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4851 				err = -EIO;
4852 				goto out;
4853 			}
4854 			err = drbd_recv_all(peer_device->connection, p, pi->size);
4855 			if (err)
4856 			       goto out;
4857 			err = decode_bitmap_c(peer_device, p, &c, pi->size);
4858 		} else {
4859 			drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4860 			err = -EIO;
4861 			goto out;
4862 		}
4863 
4864 		c.packets[pi->cmd == P_BITMAP]++;
4865 		c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4866 
4867 		if (err <= 0) {
4868 			if (err < 0)
4869 				goto out;
4870 			break;
4871 		}
4872 		err = drbd_recv_header(peer_device->connection, pi);
4873 		if (err)
4874 			goto out;
4875 	}
4876 
4877 	INFO_bm_xfer_stats(peer_device, "receive", &c);
4878 
4879 	if (device->state.conn == C_WF_BITMAP_T) {
4880 		enum drbd_state_rv rv;
4881 
4882 		err = drbd_send_bitmap(device, peer_device);
4883 		if (err)
4884 			goto out;
4885 		/* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4886 		rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4887 		D_ASSERT(device, rv == SS_SUCCESS);
4888 	} else if (device->state.conn != C_WF_BITMAP_S) {
4889 		/* admin may have requested C_DISCONNECTING,
4890 		 * other threads may have noticed network errors */
4891 		drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4892 		    drbd_conn_str(device->state.conn));
4893 	}
4894 	err = 0;
4895 
4896  out:
4897 	drbd_bm_unlock(device);
4898 	if (!err && device->state.conn == C_WF_BITMAP_S)
4899 		drbd_start_resync(device, C_SYNC_SOURCE);
4900 	return err;
4901 }
4902 
4903 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4904 {
4905 	drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4906 		 pi->cmd, pi->size);
4907 
4908 	return ignore_remaining_packet(connection, pi);
4909 }
4910 
4911 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4912 {
4913 	/* Make sure we've acked all the TCP data associated
4914 	 * with the data requests being unplugged */
4915 	tcp_sock_set_quickack(connection->data.socket->sk, 2);
4916 	return 0;
4917 }
4918 
4919 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4920 {
4921 	struct drbd_peer_device *peer_device;
4922 	struct drbd_device *device;
4923 	struct p_block_desc *p = pi->data;
4924 
4925 	peer_device = conn_peer_device(connection, pi->vnr);
4926 	if (!peer_device)
4927 		return -EIO;
4928 	device = peer_device->device;
4929 
4930 	switch (device->state.conn) {
4931 	case C_WF_SYNC_UUID:
4932 	case C_WF_BITMAP_T:
4933 	case C_BEHIND:
4934 			break;
4935 	default:
4936 		drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4937 				drbd_conn_str(device->state.conn));
4938 	}
4939 
4940 	drbd_set_out_of_sync(peer_device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4941 
4942 	return 0;
4943 }
4944 
4945 static int receive_rs_deallocated(struct drbd_connection *connection, struct packet_info *pi)
4946 {
4947 	struct drbd_peer_device *peer_device;
4948 	struct p_block_desc *p = pi->data;
4949 	struct drbd_device *device;
4950 	sector_t sector;
4951 	int size, err = 0;
4952 
4953 	peer_device = conn_peer_device(connection, pi->vnr);
4954 	if (!peer_device)
4955 		return -EIO;
4956 	device = peer_device->device;
4957 
4958 	sector = be64_to_cpu(p->sector);
4959 	size = be32_to_cpu(p->blksize);
4960 
4961 	dec_rs_pending(peer_device);
4962 
4963 	if (get_ldev(device)) {
4964 		struct drbd_peer_request *peer_req;
4965 
4966 		peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER, sector,
4967 					       size, 0, GFP_NOIO);
4968 		if (!peer_req) {
4969 			put_ldev(device);
4970 			return -ENOMEM;
4971 		}
4972 
4973 		peer_req->w.cb = e_end_resync_block;
4974 		peer_req->opf = REQ_OP_DISCARD;
4975 		peer_req->submit_jif = jiffies;
4976 		peer_req->flags |= EE_TRIM;
4977 
4978 		spin_lock_irq(&device->resource->req_lock);
4979 		list_add_tail(&peer_req->w.list, &device->sync_ee);
4980 		spin_unlock_irq(&device->resource->req_lock);
4981 
4982 		atomic_add(pi->size >> 9, &device->rs_sect_ev);
4983 		err = drbd_submit_peer_request(peer_req);
4984 
4985 		if (err) {
4986 			spin_lock_irq(&device->resource->req_lock);
4987 			list_del(&peer_req->w.list);
4988 			spin_unlock_irq(&device->resource->req_lock);
4989 
4990 			drbd_free_peer_req(device, peer_req);
4991 			put_ldev(device);
4992 			err = 0;
4993 			goto fail;
4994 		}
4995 
4996 		inc_unacked(device);
4997 
4998 		/* No put_ldev() here. Gets called in drbd_endio_write_sec_final(),
4999 		   as well as drbd_rs_complete_io() */
5000 	} else {
5001 	fail:
5002 		drbd_rs_complete_io(device, sector);
5003 		drbd_send_ack_ex(peer_device, P_NEG_ACK, sector, size, ID_SYNCER);
5004 	}
5005 
5006 	atomic_add(size >> 9, &device->rs_sect_in);
5007 
5008 	return err;
5009 }
5010 
5011 struct data_cmd {
5012 	int expect_payload;
5013 	unsigned int pkt_size;
5014 	int (*fn)(struct drbd_connection *, struct packet_info *);
5015 };
5016 
5017 static struct data_cmd drbd_cmd_handler[] = {
5018 	[P_DATA]	    = { 1, sizeof(struct p_data), receive_Data },
5019 	[P_DATA_REPLY]	    = { 1, sizeof(struct p_data), receive_DataReply },
5020 	[P_RS_DATA_REPLY]   = { 1, sizeof(struct p_data), receive_RSDataReply } ,
5021 	[P_BARRIER]	    = { 0, sizeof(struct p_barrier), receive_Barrier } ,
5022 	[P_BITMAP]	    = { 1, 0, receive_bitmap } ,
5023 	[P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
5024 	[P_UNPLUG_REMOTE]   = { 0, 0, receive_UnplugRemote },
5025 	[P_DATA_REQUEST]    = { 0, sizeof(struct p_block_req), receive_DataRequest },
5026 	[P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5027 	[P_SYNC_PARAM]	    = { 1, 0, receive_SyncParam },
5028 	[P_SYNC_PARAM89]    = { 1, 0, receive_SyncParam },
5029 	[P_PROTOCOL]        = { 1, sizeof(struct p_protocol), receive_protocol },
5030 	[P_UUIDS]	    = { 0, sizeof(struct p_uuids), receive_uuids },
5031 	[P_SIZES]	    = { 0, sizeof(struct p_sizes), receive_sizes },
5032 	[P_STATE]	    = { 0, sizeof(struct p_state), receive_state },
5033 	[P_STATE_CHG_REQ]   = { 0, sizeof(struct p_req_state), receive_req_state },
5034 	[P_SYNC_UUID]       = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
5035 	[P_OV_REQUEST]      = { 0, sizeof(struct p_block_req), receive_DataRequest },
5036 	[P_OV_REPLY]        = { 1, sizeof(struct p_block_req), receive_DataRequest },
5037 	[P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5038 	[P_RS_THIN_REQ]     = { 0, sizeof(struct p_block_req), receive_DataRequest },
5039 	[P_DELAY_PROBE]     = { 0, sizeof(struct p_delay_probe93), receive_skip },
5040 	[P_OUT_OF_SYNC]     = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
5041 	[P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
5042 	[P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
5043 	[P_TRIM]	    = { 0, sizeof(struct p_trim), receive_Data },
5044 	[P_ZEROES]	    = { 0, sizeof(struct p_trim), receive_Data },
5045 	[P_RS_DEALLOCATED]  = { 0, sizeof(struct p_block_desc), receive_rs_deallocated },
5046 };
5047 
5048 static void drbdd(struct drbd_connection *connection)
5049 {
5050 	struct packet_info pi;
5051 	size_t shs; /* sub header size */
5052 	int err;
5053 
5054 	while (get_t_state(&connection->receiver) == RUNNING) {
5055 		struct data_cmd const *cmd;
5056 
5057 		drbd_thread_current_set_cpu(&connection->receiver);
5058 		update_receiver_timing_details(connection, drbd_recv_header_maybe_unplug);
5059 		if (drbd_recv_header_maybe_unplug(connection, &pi))
5060 			goto err_out;
5061 
5062 		cmd = &drbd_cmd_handler[pi.cmd];
5063 		if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
5064 			drbd_err(connection, "Unexpected data packet %s (0x%04x)",
5065 				 cmdname(pi.cmd), pi.cmd);
5066 			goto err_out;
5067 		}
5068 
5069 		shs = cmd->pkt_size;
5070 		if (pi.cmd == P_SIZES && connection->agreed_features & DRBD_FF_WSAME)
5071 			shs += sizeof(struct o_qlim);
5072 		if (pi.size > shs && !cmd->expect_payload) {
5073 			drbd_err(connection, "No payload expected %s l:%d\n",
5074 				 cmdname(pi.cmd), pi.size);
5075 			goto err_out;
5076 		}
5077 		if (pi.size < shs) {
5078 			drbd_err(connection, "%s: unexpected packet size, expected:%d received:%d\n",
5079 				 cmdname(pi.cmd), (int)shs, pi.size);
5080 			goto err_out;
5081 		}
5082 
5083 		if (shs) {
5084 			update_receiver_timing_details(connection, drbd_recv_all_warn);
5085 			err = drbd_recv_all_warn(connection, pi.data, shs);
5086 			if (err)
5087 				goto err_out;
5088 			pi.size -= shs;
5089 		}
5090 
5091 		update_receiver_timing_details(connection, cmd->fn);
5092 		err = cmd->fn(connection, &pi);
5093 		if (err) {
5094 			drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
5095 				 cmdname(pi.cmd), err, pi.size);
5096 			goto err_out;
5097 		}
5098 	}
5099 	return;
5100 
5101     err_out:
5102 	conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
5103 }
5104 
5105 static void conn_disconnect(struct drbd_connection *connection)
5106 {
5107 	struct drbd_peer_device *peer_device;
5108 	enum drbd_conns oc;
5109 	int vnr;
5110 
5111 	if (connection->cstate == C_STANDALONE)
5112 		return;
5113 
5114 	/* We are about to start the cleanup after connection loss.
5115 	 * Make sure drbd_make_request knows about that.
5116 	 * Usually we should be in some network failure state already,
5117 	 * but just in case we are not, we fix it up here.
5118 	 */
5119 	conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5120 
5121 	/* ack_receiver does not clean up anything. it must not interfere, either */
5122 	drbd_thread_stop(&connection->ack_receiver);
5123 	if (connection->ack_sender) {
5124 		destroy_workqueue(connection->ack_sender);
5125 		connection->ack_sender = NULL;
5126 	}
5127 	drbd_free_sock(connection);
5128 
5129 	rcu_read_lock();
5130 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5131 		struct drbd_device *device = peer_device->device;
5132 		kref_get(&device->kref);
5133 		rcu_read_unlock();
5134 		drbd_disconnected(peer_device);
5135 		kref_put(&device->kref, drbd_destroy_device);
5136 		rcu_read_lock();
5137 	}
5138 	rcu_read_unlock();
5139 
5140 	if (!list_empty(&connection->current_epoch->list))
5141 		drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
5142 	/* ok, no more ee's on the fly, it is safe to reset the epoch_size */
5143 	atomic_set(&connection->current_epoch->epoch_size, 0);
5144 	connection->send.seen_any_write_yet = false;
5145 
5146 	drbd_info(connection, "Connection closed\n");
5147 
5148 	if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
5149 		conn_try_outdate_peer_async(connection);
5150 
5151 	spin_lock_irq(&connection->resource->req_lock);
5152 	oc = connection->cstate;
5153 	if (oc >= C_UNCONNECTED)
5154 		_conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
5155 
5156 	spin_unlock_irq(&connection->resource->req_lock);
5157 
5158 	if (oc == C_DISCONNECTING)
5159 		conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
5160 }
5161 
5162 static int drbd_disconnected(struct drbd_peer_device *peer_device)
5163 {
5164 	struct drbd_device *device = peer_device->device;
5165 	unsigned int i;
5166 
5167 	/* wait for current activity to cease. */
5168 	spin_lock_irq(&device->resource->req_lock);
5169 	_drbd_wait_ee_list_empty(device, &device->active_ee);
5170 	_drbd_wait_ee_list_empty(device, &device->sync_ee);
5171 	_drbd_wait_ee_list_empty(device, &device->read_ee);
5172 	spin_unlock_irq(&device->resource->req_lock);
5173 
5174 	/* We do not have data structures that would allow us to
5175 	 * get the rs_pending_cnt down to 0 again.
5176 	 *  * On C_SYNC_TARGET we do not have any data structures describing
5177 	 *    the pending RSDataRequest's we have sent.
5178 	 *  * On C_SYNC_SOURCE there is no data structure that tracks
5179 	 *    the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
5180 	 *  And no, it is not the sum of the reference counts in the
5181 	 *  resync_LRU. The resync_LRU tracks the whole operation including
5182 	 *  the disk-IO, while the rs_pending_cnt only tracks the blocks
5183 	 *  on the fly. */
5184 	drbd_rs_cancel_all(device);
5185 	device->rs_total = 0;
5186 	device->rs_failed = 0;
5187 	atomic_set(&device->rs_pending_cnt, 0);
5188 	wake_up(&device->misc_wait);
5189 
5190 	del_timer_sync(&device->resync_timer);
5191 	resync_timer_fn(&device->resync_timer);
5192 
5193 	/* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
5194 	 * w_make_resync_request etc. which may still be on the worker queue
5195 	 * to be "canceled" */
5196 	drbd_flush_workqueue(&peer_device->connection->sender_work);
5197 
5198 	drbd_finish_peer_reqs(device);
5199 
5200 	/* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
5201 	   might have issued a work again. The one before drbd_finish_peer_reqs() is
5202 	   necessary to reclain net_ee in drbd_finish_peer_reqs(). */
5203 	drbd_flush_workqueue(&peer_device->connection->sender_work);
5204 
5205 	/* need to do it again, drbd_finish_peer_reqs() may have populated it
5206 	 * again via drbd_try_clear_on_disk_bm(). */
5207 	drbd_rs_cancel_all(device);
5208 
5209 	kfree(device->p_uuid);
5210 	device->p_uuid = NULL;
5211 
5212 	if (!drbd_suspended(device))
5213 		tl_clear(peer_device->connection);
5214 
5215 	drbd_md_sync(device);
5216 
5217 	if (get_ldev(device)) {
5218 		drbd_bitmap_io(device, &drbd_bm_write_copy_pages,
5219 				"write from disconnected", BM_LOCKED_CHANGE_ALLOWED, NULL);
5220 		put_ldev(device);
5221 	}
5222 
5223 	/* tcp_close and release of sendpage pages can be deferred.  I don't
5224 	 * want to use SO_LINGER, because apparently it can be deferred for
5225 	 * more than 20 seconds (longest time I checked).
5226 	 *
5227 	 * Actually we don't care for exactly when the network stack does its
5228 	 * put_page(), but release our reference on these pages right here.
5229 	 */
5230 	i = drbd_free_peer_reqs(device, &device->net_ee);
5231 	if (i)
5232 		drbd_info(device, "net_ee not empty, killed %u entries\n", i);
5233 	i = atomic_read(&device->pp_in_use_by_net);
5234 	if (i)
5235 		drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
5236 	i = atomic_read(&device->pp_in_use);
5237 	if (i)
5238 		drbd_info(device, "pp_in_use = %d, expected 0\n", i);
5239 
5240 	D_ASSERT(device, list_empty(&device->read_ee));
5241 	D_ASSERT(device, list_empty(&device->active_ee));
5242 	D_ASSERT(device, list_empty(&device->sync_ee));
5243 	D_ASSERT(device, list_empty(&device->done_ee));
5244 
5245 	return 0;
5246 }
5247 
5248 /*
5249  * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
5250  * we can agree on is stored in agreed_pro_version.
5251  *
5252  * feature flags and the reserved array should be enough room for future
5253  * enhancements of the handshake protocol, and possible plugins...
5254  *
5255  * for now, they are expected to be zero, but ignored.
5256  */
5257 static int drbd_send_features(struct drbd_connection *connection)
5258 {
5259 	struct drbd_socket *sock;
5260 	struct p_connection_features *p;
5261 
5262 	sock = &connection->data;
5263 	p = conn_prepare_command(connection, sock);
5264 	if (!p)
5265 		return -EIO;
5266 	memset(p, 0, sizeof(*p));
5267 	p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
5268 	p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
5269 	p->feature_flags = cpu_to_be32(PRO_FEATURES);
5270 	return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
5271 }
5272 
5273 /*
5274  * return values:
5275  *   1 yes, we have a valid connection
5276  *   0 oops, did not work out, please try again
5277  *  -1 peer talks different language,
5278  *     no point in trying again, please go standalone.
5279  */
5280 static int drbd_do_features(struct drbd_connection *connection)
5281 {
5282 	/* ASSERT current == connection->receiver ... */
5283 	struct p_connection_features *p;
5284 	const int expect = sizeof(struct p_connection_features);
5285 	struct packet_info pi;
5286 	int err;
5287 
5288 	err = drbd_send_features(connection);
5289 	if (err)
5290 		return 0;
5291 
5292 	err = drbd_recv_header(connection, &pi);
5293 	if (err)
5294 		return 0;
5295 
5296 	if (pi.cmd != P_CONNECTION_FEATURES) {
5297 		drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
5298 			 cmdname(pi.cmd), pi.cmd);
5299 		return -1;
5300 	}
5301 
5302 	if (pi.size != expect) {
5303 		drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
5304 		     expect, pi.size);
5305 		return -1;
5306 	}
5307 
5308 	p = pi.data;
5309 	err = drbd_recv_all_warn(connection, p, expect);
5310 	if (err)
5311 		return 0;
5312 
5313 	p->protocol_min = be32_to_cpu(p->protocol_min);
5314 	p->protocol_max = be32_to_cpu(p->protocol_max);
5315 	if (p->protocol_max == 0)
5316 		p->protocol_max = p->protocol_min;
5317 
5318 	if (PRO_VERSION_MAX < p->protocol_min ||
5319 	    PRO_VERSION_MIN > p->protocol_max)
5320 		goto incompat;
5321 
5322 	connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
5323 	connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
5324 
5325 	drbd_info(connection, "Handshake successful: "
5326 	     "Agreed network protocol version %d\n", connection->agreed_pro_version);
5327 
5328 	drbd_info(connection, "Feature flags enabled on protocol level: 0x%x%s%s%s%s.\n",
5329 		  connection->agreed_features,
5330 		  connection->agreed_features & DRBD_FF_TRIM ? " TRIM" : "",
5331 		  connection->agreed_features & DRBD_FF_THIN_RESYNC ? " THIN_RESYNC" : "",
5332 		  connection->agreed_features & DRBD_FF_WSAME ? " WRITE_SAME" : "",
5333 		  connection->agreed_features & DRBD_FF_WZEROES ? " WRITE_ZEROES" :
5334 		  connection->agreed_features ? "" : " none");
5335 
5336 	return 1;
5337 
5338  incompat:
5339 	drbd_err(connection, "incompatible DRBD dialects: "
5340 	    "I support %d-%d, peer supports %d-%d\n",
5341 	    PRO_VERSION_MIN, PRO_VERSION_MAX,
5342 	    p->protocol_min, p->protocol_max);
5343 	return -1;
5344 }
5345 
5346 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
5347 static int drbd_do_auth(struct drbd_connection *connection)
5348 {
5349 	drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
5350 	drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
5351 	return -1;
5352 }
5353 #else
5354 #define CHALLENGE_LEN 64
5355 
5356 /* Return value:
5357 	1 - auth succeeded,
5358 	0 - failed, try again (network error),
5359 	-1 - auth failed, don't try again.
5360 */
5361 
5362 static int drbd_do_auth(struct drbd_connection *connection)
5363 {
5364 	struct drbd_socket *sock;
5365 	char my_challenge[CHALLENGE_LEN];  /* 64 Bytes... */
5366 	char *response = NULL;
5367 	char *right_response = NULL;
5368 	char *peers_ch = NULL;
5369 	unsigned int key_len;
5370 	char secret[SHARED_SECRET_MAX]; /* 64 byte */
5371 	unsigned int resp_size;
5372 	struct shash_desc *desc;
5373 	struct packet_info pi;
5374 	struct net_conf *nc;
5375 	int err, rv;
5376 
5377 	/* FIXME: Put the challenge/response into the preallocated socket buffer.  */
5378 
5379 	rcu_read_lock();
5380 	nc = rcu_dereference(connection->net_conf);
5381 	key_len = strlen(nc->shared_secret);
5382 	memcpy(secret, nc->shared_secret, key_len);
5383 	rcu_read_unlock();
5384 
5385 	desc = kmalloc(sizeof(struct shash_desc) +
5386 		       crypto_shash_descsize(connection->cram_hmac_tfm),
5387 		       GFP_KERNEL);
5388 	if (!desc) {
5389 		rv = -1;
5390 		goto fail;
5391 	}
5392 	desc->tfm = connection->cram_hmac_tfm;
5393 
5394 	rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
5395 	if (rv) {
5396 		drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
5397 		rv = -1;
5398 		goto fail;
5399 	}
5400 
5401 	get_random_bytes(my_challenge, CHALLENGE_LEN);
5402 
5403 	sock = &connection->data;
5404 	if (!conn_prepare_command(connection, sock)) {
5405 		rv = 0;
5406 		goto fail;
5407 	}
5408 	rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
5409 				my_challenge, CHALLENGE_LEN);
5410 	if (!rv)
5411 		goto fail;
5412 
5413 	err = drbd_recv_header(connection, &pi);
5414 	if (err) {
5415 		rv = 0;
5416 		goto fail;
5417 	}
5418 
5419 	if (pi.cmd != P_AUTH_CHALLENGE) {
5420 		drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
5421 			 cmdname(pi.cmd), pi.cmd);
5422 		rv = -1;
5423 		goto fail;
5424 	}
5425 
5426 	if (pi.size > CHALLENGE_LEN * 2) {
5427 		drbd_err(connection, "expected AuthChallenge payload too big.\n");
5428 		rv = -1;
5429 		goto fail;
5430 	}
5431 
5432 	if (pi.size < CHALLENGE_LEN) {
5433 		drbd_err(connection, "AuthChallenge payload too small.\n");
5434 		rv = -1;
5435 		goto fail;
5436 	}
5437 
5438 	peers_ch = kmalloc(pi.size, GFP_NOIO);
5439 	if (!peers_ch) {
5440 		rv = -1;
5441 		goto fail;
5442 	}
5443 
5444 	err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5445 	if (err) {
5446 		rv = 0;
5447 		goto fail;
5448 	}
5449 
5450 	if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5451 		drbd_err(connection, "Peer presented the same challenge!\n");
5452 		rv = -1;
5453 		goto fail;
5454 	}
5455 
5456 	resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5457 	response = kmalloc(resp_size, GFP_NOIO);
5458 	if (!response) {
5459 		rv = -1;
5460 		goto fail;
5461 	}
5462 
5463 	rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5464 	if (rv) {
5465 		drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5466 		rv = -1;
5467 		goto fail;
5468 	}
5469 
5470 	if (!conn_prepare_command(connection, sock)) {
5471 		rv = 0;
5472 		goto fail;
5473 	}
5474 	rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5475 				response, resp_size);
5476 	if (!rv)
5477 		goto fail;
5478 
5479 	err = drbd_recv_header(connection, &pi);
5480 	if (err) {
5481 		rv = 0;
5482 		goto fail;
5483 	}
5484 
5485 	if (pi.cmd != P_AUTH_RESPONSE) {
5486 		drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5487 			 cmdname(pi.cmd), pi.cmd);
5488 		rv = 0;
5489 		goto fail;
5490 	}
5491 
5492 	if (pi.size != resp_size) {
5493 		drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5494 		rv = 0;
5495 		goto fail;
5496 	}
5497 
5498 	err = drbd_recv_all_warn(connection, response , resp_size);
5499 	if (err) {
5500 		rv = 0;
5501 		goto fail;
5502 	}
5503 
5504 	right_response = kmalloc(resp_size, GFP_NOIO);
5505 	if (!right_response) {
5506 		rv = -1;
5507 		goto fail;
5508 	}
5509 
5510 	rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5511 				 right_response);
5512 	if (rv) {
5513 		drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5514 		rv = -1;
5515 		goto fail;
5516 	}
5517 
5518 	rv = !memcmp(response, right_response, resp_size);
5519 
5520 	if (rv)
5521 		drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5522 		     resp_size);
5523 	else
5524 		rv = -1;
5525 
5526  fail:
5527 	kfree(peers_ch);
5528 	kfree(response);
5529 	kfree(right_response);
5530 	if (desc) {
5531 		shash_desc_zero(desc);
5532 		kfree(desc);
5533 	}
5534 
5535 	return rv;
5536 }
5537 #endif
5538 
5539 int drbd_receiver(struct drbd_thread *thi)
5540 {
5541 	struct drbd_connection *connection = thi->connection;
5542 	int h;
5543 
5544 	drbd_info(connection, "receiver (re)started\n");
5545 
5546 	do {
5547 		h = conn_connect(connection);
5548 		if (h == 0) {
5549 			conn_disconnect(connection);
5550 			schedule_timeout_interruptible(HZ);
5551 		}
5552 		if (h == -1) {
5553 			drbd_warn(connection, "Discarding network configuration.\n");
5554 			conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5555 		}
5556 	} while (h == 0);
5557 
5558 	if (h > 0) {
5559 		blk_start_plug(&connection->receiver_plug);
5560 		drbdd(connection);
5561 		blk_finish_plug(&connection->receiver_plug);
5562 	}
5563 
5564 	conn_disconnect(connection);
5565 
5566 	drbd_info(connection, "receiver terminated\n");
5567 	return 0;
5568 }
5569 
5570 /* ********* acknowledge sender ******** */
5571 
5572 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5573 {
5574 	struct p_req_state_reply *p = pi->data;
5575 	int retcode = be32_to_cpu(p->retcode);
5576 
5577 	if (retcode >= SS_SUCCESS) {
5578 		set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5579 	} else {
5580 		set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5581 		drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5582 			 drbd_set_st_err_str(retcode), retcode);
5583 	}
5584 	wake_up(&connection->ping_wait);
5585 
5586 	return 0;
5587 }
5588 
5589 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5590 {
5591 	struct drbd_peer_device *peer_device;
5592 	struct drbd_device *device;
5593 	struct p_req_state_reply *p = pi->data;
5594 	int retcode = be32_to_cpu(p->retcode);
5595 
5596 	peer_device = conn_peer_device(connection, pi->vnr);
5597 	if (!peer_device)
5598 		return -EIO;
5599 	device = peer_device->device;
5600 
5601 	if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5602 		D_ASSERT(device, connection->agreed_pro_version < 100);
5603 		return got_conn_RqSReply(connection, pi);
5604 	}
5605 
5606 	if (retcode >= SS_SUCCESS) {
5607 		set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5608 	} else {
5609 		set_bit(CL_ST_CHG_FAIL, &device->flags);
5610 		drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5611 			drbd_set_st_err_str(retcode), retcode);
5612 	}
5613 	wake_up(&device->state_wait);
5614 
5615 	return 0;
5616 }
5617 
5618 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5619 {
5620 	return drbd_send_ping_ack(connection);
5621 
5622 }
5623 
5624 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5625 {
5626 	/* restore idle timeout */
5627 	connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5628 	if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5629 		wake_up(&connection->ping_wait);
5630 
5631 	return 0;
5632 }
5633 
5634 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5635 {
5636 	struct drbd_peer_device *peer_device;
5637 	struct drbd_device *device;
5638 	struct p_block_ack *p = pi->data;
5639 	sector_t sector = be64_to_cpu(p->sector);
5640 	int blksize = be32_to_cpu(p->blksize);
5641 
5642 	peer_device = conn_peer_device(connection, pi->vnr);
5643 	if (!peer_device)
5644 		return -EIO;
5645 	device = peer_device->device;
5646 
5647 	D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5648 
5649 	update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5650 
5651 	if (get_ldev(device)) {
5652 		drbd_rs_complete_io(device, sector);
5653 		drbd_set_in_sync(peer_device, sector, blksize);
5654 		/* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5655 		device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5656 		put_ldev(device);
5657 	}
5658 	dec_rs_pending(peer_device);
5659 	atomic_add(blksize >> 9, &device->rs_sect_in);
5660 
5661 	return 0;
5662 }
5663 
5664 static int
5665 validate_req_change_req_state(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
5666 			      struct rb_root *root, const char *func,
5667 			      enum drbd_req_event what, bool missing_ok)
5668 {
5669 	struct drbd_device *device = peer_device->device;
5670 	struct drbd_request *req;
5671 	struct bio_and_error m;
5672 
5673 	spin_lock_irq(&device->resource->req_lock);
5674 	req = find_request(device, root, id, sector, missing_ok, func);
5675 	if (unlikely(!req)) {
5676 		spin_unlock_irq(&device->resource->req_lock);
5677 		return -EIO;
5678 	}
5679 	__req_mod(req, what, peer_device, &m);
5680 	spin_unlock_irq(&device->resource->req_lock);
5681 
5682 	if (m.bio)
5683 		complete_master_bio(device, &m);
5684 	return 0;
5685 }
5686 
5687 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5688 {
5689 	struct drbd_peer_device *peer_device;
5690 	struct drbd_device *device;
5691 	struct p_block_ack *p = pi->data;
5692 	sector_t sector = be64_to_cpu(p->sector);
5693 	int blksize = be32_to_cpu(p->blksize);
5694 	enum drbd_req_event what;
5695 
5696 	peer_device = conn_peer_device(connection, pi->vnr);
5697 	if (!peer_device)
5698 		return -EIO;
5699 	device = peer_device->device;
5700 
5701 	update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5702 
5703 	if (p->block_id == ID_SYNCER) {
5704 		drbd_set_in_sync(peer_device, sector, blksize);
5705 		dec_rs_pending(peer_device);
5706 		return 0;
5707 	}
5708 	switch (pi->cmd) {
5709 	case P_RS_WRITE_ACK:
5710 		what = WRITE_ACKED_BY_PEER_AND_SIS;
5711 		break;
5712 	case P_WRITE_ACK:
5713 		what = WRITE_ACKED_BY_PEER;
5714 		break;
5715 	case P_RECV_ACK:
5716 		what = RECV_ACKED_BY_PEER;
5717 		break;
5718 	case P_SUPERSEDED:
5719 		what = CONFLICT_RESOLVED;
5720 		break;
5721 	case P_RETRY_WRITE:
5722 		what = POSTPONE_WRITE;
5723 		break;
5724 	default:
5725 		BUG();
5726 	}
5727 
5728 	return validate_req_change_req_state(peer_device, p->block_id, sector,
5729 					     &device->write_requests, __func__,
5730 					     what, false);
5731 }
5732 
5733 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5734 {
5735 	struct drbd_peer_device *peer_device;
5736 	struct drbd_device *device;
5737 	struct p_block_ack *p = pi->data;
5738 	sector_t sector = be64_to_cpu(p->sector);
5739 	int size = be32_to_cpu(p->blksize);
5740 	int err;
5741 
5742 	peer_device = conn_peer_device(connection, pi->vnr);
5743 	if (!peer_device)
5744 		return -EIO;
5745 	device = peer_device->device;
5746 
5747 	update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5748 
5749 	if (p->block_id == ID_SYNCER) {
5750 		dec_rs_pending(peer_device);
5751 		drbd_rs_failed_io(peer_device, sector, size);
5752 		return 0;
5753 	}
5754 
5755 	err = validate_req_change_req_state(peer_device, p->block_id, sector,
5756 					    &device->write_requests, __func__,
5757 					    NEG_ACKED, true);
5758 	if (err) {
5759 		/* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5760 		   The master bio might already be completed, therefore the
5761 		   request is no longer in the collision hash. */
5762 		/* In Protocol B we might already have got a P_RECV_ACK
5763 		   but then get a P_NEG_ACK afterwards. */
5764 		drbd_set_out_of_sync(peer_device, sector, size);
5765 	}
5766 	return 0;
5767 }
5768 
5769 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5770 {
5771 	struct drbd_peer_device *peer_device;
5772 	struct drbd_device *device;
5773 	struct p_block_ack *p = pi->data;
5774 	sector_t sector = be64_to_cpu(p->sector);
5775 
5776 	peer_device = conn_peer_device(connection, pi->vnr);
5777 	if (!peer_device)
5778 		return -EIO;
5779 	device = peer_device->device;
5780 
5781 	update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5782 
5783 	drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5784 	    (unsigned long long)sector, be32_to_cpu(p->blksize));
5785 
5786 	return validate_req_change_req_state(peer_device, p->block_id, sector,
5787 					     &device->read_requests, __func__,
5788 					     NEG_ACKED, false);
5789 }
5790 
5791 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5792 {
5793 	struct drbd_peer_device *peer_device;
5794 	struct drbd_device *device;
5795 	sector_t sector;
5796 	int size;
5797 	struct p_block_ack *p = pi->data;
5798 
5799 	peer_device = conn_peer_device(connection, pi->vnr);
5800 	if (!peer_device)
5801 		return -EIO;
5802 	device = peer_device->device;
5803 
5804 	sector = be64_to_cpu(p->sector);
5805 	size = be32_to_cpu(p->blksize);
5806 
5807 	update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5808 
5809 	dec_rs_pending(peer_device);
5810 
5811 	if (get_ldev_if_state(device, D_FAILED)) {
5812 		drbd_rs_complete_io(device, sector);
5813 		switch (pi->cmd) {
5814 		case P_NEG_RS_DREPLY:
5815 			drbd_rs_failed_io(peer_device, sector, size);
5816 			break;
5817 		case P_RS_CANCEL:
5818 			break;
5819 		default:
5820 			BUG();
5821 		}
5822 		put_ldev(device);
5823 	}
5824 
5825 	return 0;
5826 }
5827 
5828 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5829 {
5830 	struct p_barrier_ack *p = pi->data;
5831 	struct drbd_peer_device *peer_device;
5832 	int vnr;
5833 
5834 	tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5835 
5836 	rcu_read_lock();
5837 	idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5838 		struct drbd_device *device = peer_device->device;
5839 
5840 		if (device->state.conn == C_AHEAD &&
5841 		    atomic_read(&device->ap_in_flight) == 0 &&
5842 		    !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5843 			device->start_resync_timer.expires = jiffies + HZ;
5844 			add_timer(&device->start_resync_timer);
5845 		}
5846 	}
5847 	rcu_read_unlock();
5848 
5849 	return 0;
5850 }
5851 
5852 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5853 {
5854 	struct drbd_peer_device *peer_device;
5855 	struct drbd_device *device;
5856 	struct p_block_ack *p = pi->data;
5857 	struct drbd_device_work *dw;
5858 	sector_t sector;
5859 	int size;
5860 
5861 	peer_device = conn_peer_device(connection, pi->vnr);
5862 	if (!peer_device)
5863 		return -EIO;
5864 	device = peer_device->device;
5865 
5866 	sector = be64_to_cpu(p->sector);
5867 	size = be32_to_cpu(p->blksize);
5868 
5869 	update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5870 
5871 	if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5872 		drbd_ov_out_of_sync_found(peer_device, sector, size);
5873 	else
5874 		ov_out_of_sync_print(peer_device);
5875 
5876 	if (!get_ldev(device))
5877 		return 0;
5878 
5879 	drbd_rs_complete_io(device, sector);
5880 	dec_rs_pending(peer_device);
5881 
5882 	--device->ov_left;
5883 
5884 	/* let's advance progress step marks only for every other megabyte */
5885 	if ((device->ov_left & 0x200) == 0x200)
5886 		drbd_advance_rs_marks(peer_device, device->ov_left);
5887 
5888 	if (device->ov_left == 0) {
5889 		dw = kmalloc(sizeof(*dw), GFP_NOIO);
5890 		if (dw) {
5891 			dw->w.cb = w_ov_finished;
5892 			dw->device = device;
5893 			drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5894 		} else {
5895 			drbd_err(device, "kmalloc(dw) failed.");
5896 			ov_out_of_sync_print(peer_device);
5897 			drbd_resync_finished(peer_device);
5898 		}
5899 	}
5900 	put_ldev(device);
5901 	return 0;
5902 }
5903 
5904 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5905 {
5906 	return 0;
5907 }
5908 
5909 struct meta_sock_cmd {
5910 	size_t pkt_size;
5911 	int (*fn)(struct drbd_connection *connection, struct packet_info *);
5912 };
5913 
5914 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5915 {
5916 	long t;
5917 	struct net_conf *nc;
5918 
5919 	rcu_read_lock();
5920 	nc = rcu_dereference(connection->net_conf);
5921 	t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5922 	rcu_read_unlock();
5923 
5924 	t *= HZ;
5925 	if (ping_timeout)
5926 		t /= 10;
5927 
5928 	connection->meta.socket->sk->sk_rcvtimeo = t;
5929 }
5930 
5931 static void set_ping_timeout(struct drbd_connection *connection)
5932 {
5933 	set_rcvtimeo(connection, 1);
5934 }
5935 
5936 static void set_idle_timeout(struct drbd_connection *connection)
5937 {
5938 	set_rcvtimeo(connection, 0);
5939 }
5940 
5941 static struct meta_sock_cmd ack_receiver_tbl[] = {
5942 	[P_PING]	    = { 0, got_Ping },
5943 	[P_PING_ACK]	    = { 0, got_PingAck },
5944 	[P_RECV_ACK]	    = { sizeof(struct p_block_ack), got_BlockAck },
5945 	[P_WRITE_ACK]	    = { sizeof(struct p_block_ack), got_BlockAck },
5946 	[P_RS_WRITE_ACK]    = { sizeof(struct p_block_ack), got_BlockAck },
5947 	[P_SUPERSEDED]   = { sizeof(struct p_block_ack), got_BlockAck },
5948 	[P_NEG_ACK]	    = { sizeof(struct p_block_ack), got_NegAck },
5949 	[P_NEG_DREPLY]	    = { sizeof(struct p_block_ack), got_NegDReply },
5950 	[P_NEG_RS_DREPLY]   = { sizeof(struct p_block_ack), got_NegRSDReply },
5951 	[P_OV_RESULT]	    = { sizeof(struct p_block_ack), got_OVResult },
5952 	[P_BARRIER_ACK]	    = { sizeof(struct p_barrier_ack), got_BarrierAck },
5953 	[P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5954 	[P_RS_IS_IN_SYNC]   = { sizeof(struct p_block_ack), got_IsInSync },
5955 	[P_DELAY_PROBE]     = { sizeof(struct p_delay_probe93), got_skip },
5956 	[P_RS_CANCEL]       = { sizeof(struct p_block_ack), got_NegRSDReply },
5957 	[P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5958 	[P_RETRY_WRITE]	    = { sizeof(struct p_block_ack), got_BlockAck },
5959 };
5960 
5961 int drbd_ack_receiver(struct drbd_thread *thi)
5962 {
5963 	struct drbd_connection *connection = thi->connection;
5964 	struct meta_sock_cmd *cmd = NULL;
5965 	struct packet_info pi;
5966 	unsigned long pre_recv_jif;
5967 	int rv;
5968 	void *buf    = connection->meta.rbuf;
5969 	int received = 0;
5970 	unsigned int header_size = drbd_header_size(connection);
5971 	int expect   = header_size;
5972 	bool ping_timeout_active = false;
5973 
5974 	sched_set_fifo_low(current);
5975 
5976 	while (get_t_state(thi) == RUNNING) {
5977 		drbd_thread_current_set_cpu(thi);
5978 
5979 		conn_reclaim_net_peer_reqs(connection);
5980 
5981 		if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5982 			if (drbd_send_ping(connection)) {
5983 				drbd_err(connection, "drbd_send_ping has failed\n");
5984 				goto reconnect;
5985 			}
5986 			set_ping_timeout(connection);
5987 			ping_timeout_active = true;
5988 		}
5989 
5990 		pre_recv_jif = jiffies;
5991 		rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5992 
5993 		/* Note:
5994 		 * -EINTR	 (on meta) we got a signal
5995 		 * -EAGAIN	 (on meta) rcvtimeo expired
5996 		 * -ECONNRESET	 other side closed the connection
5997 		 * -ERESTARTSYS  (on data) we got a signal
5998 		 * rv <  0	 other than above: unexpected error!
5999 		 * rv == expected: full header or command
6000 		 * rv <  expected: "woken" by signal during receive
6001 		 * rv == 0	 : "connection shut down by peer"
6002 		 */
6003 		if (likely(rv > 0)) {
6004 			received += rv;
6005 			buf	 += rv;
6006 		} else if (rv == 0) {
6007 			if (test_bit(DISCONNECT_SENT, &connection->flags)) {
6008 				long t;
6009 				rcu_read_lock();
6010 				t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
6011 				rcu_read_unlock();
6012 
6013 				t = wait_event_timeout(connection->ping_wait,
6014 						       connection->cstate < C_WF_REPORT_PARAMS,
6015 						       t);
6016 				if (t)
6017 					break;
6018 			}
6019 			drbd_err(connection, "meta connection shut down by peer.\n");
6020 			goto reconnect;
6021 		} else if (rv == -EAGAIN) {
6022 			/* If the data socket received something meanwhile,
6023 			 * that is good enough: peer is still alive. */
6024 			if (time_after(connection->last_received, pre_recv_jif))
6025 				continue;
6026 			if (ping_timeout_active) {
6027 				drbd_err(connection, "PingAck did not arrive in time.\n");
6028 				goto reconnect;
6029 			}
6030 			set_bit(SEND_PING, &connection->flags);
6031 			continue;
6032 		} else if (rv == -EINTR) {
6033 			/* maybe drbd_thread_stop(): the while condition will notice.
6034 			 * maybe woken for send_ping: we'll send a ping above,
6035 			 * and change the rcvtimeo */
6036 			flush_signals(current);
6037 			continue;
6038 		} else {
6039 			drbd_err(connection, "sock_recvmsg returned %d\n", rv);
6040 			goto reconnect;
6041 		}
6042 
6043 		if (received == expect && cmd == NULL) {
6044 			if (decode_header(connection, connection->meta.rbuf, &pi))
6045 				goto reconnect;
6046 			cmd = &ack_receiver_tbl[pi.cmd];
6047 			if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
6048 				drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
6049 					 cmdname(pi.cmd), pi.cmd);
6050 				goto disconnect;
6051 			}
6052 			expect = header_size + cmd->pkt_size;
6053 			if (pi.size != expect - header_size) {
6054 				drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
6055 					pi.cmd, pi.size);
6056 				goto reconnect;
6057 			}
6058 		}
6059 		if (received == expect) {
6060 			bool err;
6061 
6062 			err = cmd->fn(connection, &pi);
6063 			if (err) {
6064 				drbd_err(connection, "%ps failed\n", cmd->fn);
6065 				goto reconnect;
6066 			}
6067 
6068 			connection->last_received = jiffies;
6069 
6070 			if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
6071 				set_idle_timeout(connection);
6072 				ping_timeout_active = false;
6073 			}
6074 
6075 			buf	 = connection->meta.rbuf;
6076 			received = 0;
6077 			expect	 = header_size;
6078 			cmd	 = NULL;
6079 		}
6080 	}
6081 
6082 	if (0) {
6083 reconnect:
6084 		conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6085 		conn_md_sync(connection);
6086 	}
6087 	if (0) {
6088 disconnect:
6089 		conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
6090 	}
6091 
6092 	drbd_info(connection, "ack_receiver terminated\n");
6093 
6094 	return 0;
6095 }
6096 
6097 void drbd_send_acks_wf(struct work_struct *ws)
6098 {
6099 	struct drbd_peer_device *peer_device =
6100 		container_of(ws, struct drbd_peer_device, send_acks_work);
6101 	struct drbd_connection *connection = peer_device->connection;
6102 	struct drbd_device *device = peer_device->device;
6103 	struct net_conf *nc;
6104 	int tcp_cork, err;
6105 
6106 	rcu_read_lock();
6107 	nc = rcu_dereference(connection->net_conf);
6108 	tcp_cork = nc->tcp_cork;
6109 	rcu_read_unlock();
6110 
6111 	if (tcp_cork)
6112 		tcp_sock_set_cork(connection->meta.socket->sk, true);
6113 
6114 	err = drbd_finish_peer_reqs(device);
6115 	kref_put(&device->kref, drbd_destroy_device);
6116 	/* get is in drbd_endio_write_sec_final(). That is necessary to keep the
6117 	   struct work_struct send_acks_work alive, which is in the peer_device object */
6118 
6119 	if (err) {
6120 		conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6121 		return;
6122 	}
6123 
6124 	if (tcp_cork)
6125 		tcp_sock_set_cork(connection->meta.socket->sk, false);
6126 
6127 	return;
6128 }
6129