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
43struct packet_info {
44 enum drbd_packet cmd;
45 unsigned int size;
46 unsigned int vnr;
47 void *data;
48};
49
50enum finish_epoch {
51 FE_STILL_LIVE,
52 FE_DESTROYED,
53 FE_RECYCLED,
54};
55
56static int drbd_do_features(struct drbd_connection *connection);
57static int drbd_do_auth(struct drbd_connection *connection);
58static int drbd_disconnected(struct drbd_peer_device *);
59static void conn_wait_active_ee_empty(struct drbd_connection *connection);
60static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
61static 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 */
75static 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, private: 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. */
109static 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
122static 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
133static 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(page: chain_first, NULL);
139 BUG_ON(tmp != chain_last);
140#endif
141
142 /* add chain to head */
143 set_page_private(page: chain_last, private: (unsigned long)*head);
144 *head = chain_first;
145}
146
147static 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(lock: &drbd_pp_lock);
158 page = page_chain_del(head: &drbd_pp_pool, n: number);
159 if (page)
160 drbd_pp_vacant -= number;
161 spin_unlock(lock: &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(page: tmp, private: (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(lock: &drbd_pp_lock);
186 page_chain_add(head: &drbd_pp_pool, chain_first: page, chain_last: tmp);
187 drbd_pp_vacant += i;
188 spin_unlock(lock: &drbd_pp_lock);
189 }
190 return NULL;
191}
192
193static 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(list: &peer_req->w.list, head: to_be_freed);
207 }
208}
209
210static 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(lock: &device->resource->req_lock);
216 reclaim_finished_net_peer_reqs(device, to_be_freed: &reclaimed);
217 spin_unlock_irq(lock: &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
222static 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(v: &device->pp_in_use_by_net))
231 continue;
232
233 kref_get(kref: &device->kref);
234 rcu_read_unlock();
235 drbd_reclaim_net_peer_reqs(device);
236 kref_put(kref: &device->kref, release: 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 */
262struct 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(v: &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(v: &device->pp_in_use_by_net) > 512)
282 drbd_reclaim_net_peer_reqs(device);
283
284 while (page == NULL) {
285 prepare_to_wait(wq_head: &drbd_pp_wait, wq_entry: &wait, TASK_INTERRUPTIBLE);
286
287 drbd_reclaim_net_peer_reqs(device);
288
289 if (atomic_read(v: &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(wq_head: &drbd_pp_wait, wq_entry: &wait);
307
308 if (page)
309 atomic_add(i: number, v: &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. */
317static 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, len: &i);
330 spin_lock(lock: &drbd_pp_lock);
331 page_chain_add(head: &drbd_pp_pool, chain_first: page, chain_last: tmp);
332 drbd_pp_vacant += i;
333 spin_unlock(lock: &drbd_pp_lock);
334 }
335 i = atomic_sub_return(i, v: 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/*
343You need to hold the req_lock:
344 _drbd_wait_ee_list_empty()
345
346You 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 */
359struct drbd_peer_request *
360drbd_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, type: DRBD_FAULT_AL_EE))
369 return NULL;
370
371 peer_req = mempool_alloc(pool: &drbd_ee_mempool, gfp_mask: 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, number: nr_pages,
380 retry: gfpflags_allow_blocking(gfp_flags: gfp_mask));
381 if (!page)
382 goto fail;
383 }
384
385 memset(peer_req, 0, sizeof(*peer_req));
386 INIT_LIST_HEAD(list: &peer_req->w.list);
387 drbd_clear_interval(i: &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(element: peer_req, pool: &drbd_ee_mempool);
403 return NULL;
404}
405
406void __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(objp: peer_req->digest);
412 drbd_free_pages(device, page: 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, i: &peer_req->i);
418 }
419 mempool_free(element: peer_req, pool: &drbd_ee_mempool);
420}
421
422int 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(lock: &device->resource->req_lock);
430 list_splice_init(list, head: &work_list);
431 spin_unlock_irq(lock: &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 */
443static 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(lock: &device->resource->req_lock);
451 reclaim_finished_net_peer_reqs(device, to_be_freed: &reclaimed);
452 list_splice_init(list: &device->done_ee, head: &work_list);
453 spin_unlock_irq(lock: &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
476static 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(wq_head: &device->ee_wait, wq_entry: &wait, TASK_UNINTERRUPTIBLE);
485 spin_unlock_irq(lock: &device->resource->req_lock);
486 io_schedule();
487 finish_wait(wq_head: &device->ee_wait, wq_entry: &wait);
488 spin_lock_irq(lock: &device->resource->req_lock);
489 }
490}
491
492static void drbd_wait_ee_list_empty(struct drbd_device *device,
493 struct list_head *head)
494{
495 spin_lock_irq(lock: &device->resource->req_lock);
496 _drbd_wait_ee_list_empty(device, head);
497 spin_unlock_irq(lock: &device->resource->req_lock);
498}
499
500static 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(i: &msg.msg_iter, ITER_DEST, kvec: &iov, nr_segs: 1, count: size);
510 return sock_recvmsg(sock, msg: &msg, flags: msg.msg_flags);
511}
512
513static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
514{
515 int rv;
516
517 rv = drbd_recv_short(sock: connection->data.socket, buf, size, flags: 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), flags: CS_HARD);
541
542out:
543 return rv;
544}
545
546static 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
559static 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 */
574static 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
588static 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(net: &init_net, family: ((struct sockaddr *)&src_in6)->sa_family,
623 type: SOCK_STREAM, IPPROTO_TCP, res: &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, snd: sndbuf_size, rcv: 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
651out:
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), flags: CS_HARD);
671 }
672
673 return sock;
674}
675
676struct 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
684static 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
695static 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(net: &init_net, family: ((struct sockaddr *)&my_addr)->sa_family,
718 type: SOCK_STREAM, IPPROTO_TCP, res: &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(sock: s_listen, snd: sndbuf_size, rcv: 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;
745out:
746 if (s_listen)
747 sock_release(sock: 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), flags: CS_HARD);
752 }
753 }
754
755 return -EIO;
756}
757
758static 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
766static 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(ceil: 2) ? timeo / 7 : -timeo / 7;
784
785 err = wait_for_completion_interruptible_timeout(x: &ad->door_bell, timeout: timeo);
786 if (err <= 0)
787 return NULL;
788
789 err = kernel_accept(sock: ad->s_listen, newsock: &s_estab, flags: 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), flags: CS_HARD);
794 }
795 }
796
797 if (s_estab)
798 unregister_state_change(sk: s_estab->sk, ad);
799
800 return s_estab;
801}
802
803static int decode_header(struct drbd_connection *, void *, struct packet_info *);
804
805static 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
813static 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, buf: connection->data.rbuf, size: header_size, flags: 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 */
845static 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: *sock, buf: tb, size: 4, MSG_DONTWAIT | MSG_PEEK);
854
855 if (rr > 0 || rr == -EAGAIN) {
856 return true;
857 } else {
858 sock_release(sock: *sock);
859 *sock = NULL;
860 return false;
861 }
862}
863
864static 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(sock: sock1);
882 ok = drbd_socket_okay(sock: 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 */
889int drbd_connected(struct drbd_peer_device *peer_device)
890{
891 struct drbd_device *device = peer_device->device;
892 int err;
893
894 atomic_set(v: &device->packet_seq, i: 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, trigger_reply: 0, flags: 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(nr: USE_DEGR_WFC_T, addr: &device->flags);
909 clear_bit(nr: RESIZE_PENDING, addr: &device->flags);
910 atomic_set(v: &device->ap_in_flight, i: 0);
911 mod_timer(timer: &device->request_timer, expires: 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 */
923static 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(nr: DISCONNECT_SENT, addr: &connection->flags);
937 if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), flags: 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: &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: &sock, cmd: P_INITIAL_DATA);
963 } else if (!msock.socket) {
964 clear_bit(nr: RESOLVE_CONFLICTS, addr: &connection->flags);
965 msock.socket = s;
966 send_first_packet(connection, sock: &msock, cmd: 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, sock1: &sock.socket, sock2: &msock.socket))
974 break;
975
976retry:
977 s = drbd_wait_for_connect(connection, ad: &ad);
978 if (s) {
979 int fp = receive_first_packet(connection, sock: s);
980 drbd_socket_okay(sock: &sock.socket);
981 drbd_socket_okay(sock: &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: sock.socket);
987 sock.socket = s;
988 goto randomize;
989 }
990 sock.socket = s;
991 break;
992 case P_INITIAL_META:
993 set_bit(nr: RESOLVE_CONFLICTS, addr: &connection->flags);
994 if (msock.socket) {
995 drbd_warn(connection, "initial packet M crossed\n");
996 sock_release(sock: 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(sock: s);
1005randomize:
1006 if (get_random_u32_below(ceil: 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(thi: &connection->receiver) == EXITING)
1017 goto out_release_sockets;
1018 }
1019
1020 ok = connection_established(connection, sock1: &sock.socket, sock2: &msock.socket);
1021 } while (!ok);
1022
1023 if (ad.s_listen)
1024 sock_release(sock: 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(sk: sock.socket->sk);
1059 tcp_sock_set_nodelay(sk: 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(lock: &connection->resource->req_lock);
1099 set_bit(nr: STATE_SENT, addr: &connection->flags);
1100 spin_unlock_irq(lock: &connection->resource->req_lock);
1101
1102 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1103 mutex_unlock(lock: 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(kref: &device->kref);
1109 rcu_read_unlock();
1110
1111 if (discard_my_data)
1112 set_bit(nr: DISCARD_MY_DATA, addr: &device->flags);
1113 else
1114 clear_bit(nr: DISCARD_MY_DATA, addr: &device->flags);
1115
1116 drbd_connected(peer_device);
1117 kref_put(kref: &device->kref, release: 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), flags: CS_VERBOSE);
1123 if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1124 clear_bit(nr: STATE_SENT, addr: &connection->flags);
1125 return 0;
1126 }
1127
1128 drbd_thread_start(thi: &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(lock: &connection->resource->conf_update);
1145
1146 return h;
1147
1148out_release_sockets:
1149 if (ad.s_listen)
1150 sock_release(sock: ad.s_listen);
1151 if (sock.socket)
1152 sock_release(sock: sock.socket);
1153 if (msock.socket)
1154 sock_release(sock: msock.socket);
1155 return -1;
1156}
1157
1158static 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
1194static 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
1202static 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, buf: buffer, size: drbd_header_size(connection));
1208 if (err)
1209 return err;
1210
1211 err = decode_header(connection, header: buffer, pi);
1212 connection->last_received = jiffies;
1213
1214 return err;
1215}
1216
1217static 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(sock: connection->data.socket, buf: 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(sk: connection->data.socket->sk, val: 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, buf: buffer, size);
1238 if (err)
1239 return err;
1240 }
1241
1242 err = decode_header(connection, header: 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 */
1251struct issue_flush_context {
1252 atomic_t pending;
1253 int error;
1254 struct completion done;
1255};
1256struct one_flush_context {
1257 struct drbd_device *device;
1258 struct issue_flush_context *ctx;
1259};
1260
1261static 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(status: bio->bi_status);
1269 drbd_info(device, "local disk FLUSH FAILED with status %d\n", bio->bi_status);
1270 }
1271 kfree(objp: octx);
1272 bio_put(bio);
1273
1274 clear_bit(nr: FLUSH_PENDING, addr: &device->flags);
1275 put_ldev(device);
1276 kref_put(kref: &device->kref, release: drbd_destroy_device);
1277
1278 if (atomic_dec_and_test(v: &ctx->pending))
1279 complete(&ctx->done);
1280}
1281
1282static void submit_one_flush(struct drbd_device *device, struct issue_flush_context *ctx)
1283{
1284 struct bio *bio = bio_alloc(bdev: device->ldev->backing_bdev, nr_vecs: 0,
1285 opf: REQ_OP_WRITE | REQ_PREFLUSH, GFP_NOIO);
1286 struct one_flush_context *octx = kmalloc(size: 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(kref: &device->kref, release: 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(nr: FLUSH_PENDING, addr: &device->flags);
1308 atomic_inc(v: &ctx->pending);
1309 submit_bio(bio);
1310}
1311
1312static 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(v: &ctx.pending, i: 1);
1320 ctx.error = 0;
1321 init_completion(x: &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(kref: &device->kref);
1330 rcu_read_unlock();
1331
1332 submit_one_flush(device, ctx: &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(v: &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(resource: connection->resource, NULL, wo: 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 */
1359static 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(lock: &connection->epoch_lock);
1368 do {
1369 next_epoch = NULL;
1370
1371 epoch_size = atomic_read(v: &epoch->epoch_size);
1372
1373 switch (ev & ~EV_CLEANUP) {
1374 case EV_PUT:
1375 atomic_dec(v: &epoch->active);
1376 break;
1377 case EV_GOT_BARRIER_NR:
1378 set_bit(nr: DE_HAVE_BARRIER_NUMBER, addr: &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(v: &epoch->active) == 0 &&
1387 (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1388 if (!(ev & EV_CLEANUP)) {
1389 spin_unlock(lock: &connection->epoch_lock);
1390 drbd_send_b_ack(connection: epoch->connection, barrier_nr: epoch->barrier_nr, set_size: epoch_size);
1391 spin_lock(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(entry: &epoch->list);
1403 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1404 connection->epochs--;
1405 kfree(objp: epoch);
1406
1407 if (rv == FE_STILL_LIVE)
1408 rv = FE_DESTROYED;
1409 } else {
1410 epoch->flags = 0;
1411 atomic_set(v: &epoch->epoch_size, i: 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(lock: &connection->epoch_lock);
1425
1426 return rv;
1427}
1428
1429static enum write_ordering_e
1430max_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 */
1448void 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(bdev: 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 */
1513int 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, sector: start, nr_sects: nr, GFP_NOIO, flags: 0);
1548 nr_sectors -= nr;
1549 start = tmp;
1550 }
1551 while (nr_sectors >= max_discard_sectors) {
1552 err |= blkdev_issue_discard(bdev, sector: start, nr_sects: 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, sector: start, nr_sects: 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, sector: start, nr_sects: nr_sectors, GFP_NOIO,
1573 flags: (flags & EE_TRIM) ? 0 : BLKDEV_ZERO_NOUNMAP);
1574 }
1575 return err != 0;
1576}
1577
1578static 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(bdev: 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
1593static 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, start: peer_req->i.sector,
1603 nr_sectors: peer_req->i.size >> 9, flags: peer_req->flags & (EE_ZEROOUT|EE_TRIM)))
1604 peer_req->flags |= EE_WAS_ERROR;
1605 drbd_endio_write_sec_final(peer_req);
1606}
1607
1608static 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. */
1634int 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(connection: 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(head: &peer_req->w.list)) {
1663 spin_lock_irq(lock: &device->resource->req_lock);
1664 list_add_tail(new: &peer_req->w.list, head: &device->active_ee);
1665 spin_unlock_irq(lock: &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 */
1680next_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(bdev: device->ldev->backing_bdev, nr_vecs: nr_pages, opf: 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, off: 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(v: &peer_req->pending_bios, i: 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, fault_type: peer_request_fault_type(peer_req), bio);
1723 } while (bios);
1724 return 0;
1725}
1726
1727static 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
1740static 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(kref: &device->kref);
1750 rcu_read_unlock();
1751 drbd_wait_ee_list_empty(device, head: &device->active_ee);
1752 kref_put(kref: &device->kref, release: drbd_destroy_device);
1753 rcu_read_lock();
1754 }
1755 rcu_read_unlock();
1756}
1757
1758static 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, epoch: connection->current_epoch, ev: 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(size: 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(v: &connection->current_epoch->epoch_size)) {
1796 epoch = kmalloc(size: 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(v: &epoch->epoch_size, i: 0);
1810 atomic_set(v: &epoch->active, i: 0);
1811
1812 spin_lock(lock: &connection->epoch_lock);
1813 if (atomic_read(v: &connection->current_epoch->epoch_size)) {
1814 list_add(new: &epoch->list, head: &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(objp: epoch);
1820 }
1821 spin_unlock(lock: &connection->epoch_lock);
1822
1823 return 0;
1824}
1825
1826/* quick wrapper in case payload size != request_size (write same) */
1827static 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 */
1846static struct drbd_peer_request *
1847read_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(disk: 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(tfm: 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(connection: peer_device->connection, buf: dig_in, size: 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, request_size: ds, payload_size: 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(connection: peer_device->connection, buf: data, size: len);
1929 if (drbd_insert_fault(device, type: 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(h: peer_device->connection->peer_integrity_tfm, r: peer_req, d: dig_vv, payload_size: data_size);
1943 if (memcmp(p: dig_in, q: dig_vv, size: 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 */
1957static 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, number: 1, retry: 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(connection: peer_device->connection, buf: data, size: len);
1973 if (err)
1974 break;
1975 data_size -= len;
1976 }
1977 kunmap(page);
1978 drbd_free_pages(device: peer_device->device, page, is_net: 0);
1979 return err;
1980}
1981
1982static 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(tfm: peer_device->connection->peer_integrity_tfm);
1995 err = drbd_recv_all_warn(connection: peer_device->connection, buf: dig_in, size: 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: &bvec);
2010 expect = min_t(int, data_size, bvec.bv_len);
2011 err = drbd_recv_all_warn(connection: peer_device->connection, buf: mapped, size: 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(p: dig_in, q: dig_vv, size: 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 */
2034static 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
2059static 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(lock: &device->resource->req_lock);
2080 list_add_tail(new: &peer_req->w.list, head: &device->sync_ee);
2081 spin_unlock_irq(lock: &device->resource->req_lock);
2082
2083 atomic_add(i: pi->size >> 9, v: &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(lock: &device->resource->req_lock);
2090 list_del(entry: &peer_req->w.list);
2091 spin_unlock_irq(lock: &device->resource->req_lock);
2092
2093 drbd_free_peer_req(device, peer_req);
2094fail:
2095 put_ldev(device);
2096 return -EIO;
2097}
2098
2099static struct drbd_request *
2100find_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
2116static 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, volume_number: 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(lock: &device->resource->req_lock);
2133 req = find_request(device, root: &device->read_requests, id: p->block_id, sector, missing_ok: false, func: __func__);
2134 spin_unlock_irq(lock: &device->resource->req_lock);
2135 if (unlikely(!req))
2136 return -EIO;
2137
2138 err = recv_dless_read(peer_device, req, sector, data_size: pi->size);
2139 if (!err)
2140 req_mod(req, what: 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
2148static 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, volume_number: 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, data_size: pi->size);
2174
2175 drbd_send_ack_dp(peer_device, P_NEG_ACK, dp: p, data_size: pi->size);
2176 }
2177
2178 atomic_add(i: pi->size >> 9, v: &device->rs_sect_in);
2179
2180 return err;
2181}
2182
2183static 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, what: CONFLICT_RESOLVED, NULL, NULL);
2199 }
2200}
2201
2202/*
2203 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2204 */
2205static 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(lock: &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, size: peer_req->i.size);
2239 spin_unlock_irq(lock: &device->resource->req_lock);
2240 } else
2241 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2242
2243 drbd_may_finish_epoch(connection: peer_device->connection, epoch: peer_req->epoch, ev: EV_PUT + (cancel ? EV_CLEANUP : 0));
2244
2245 return err;
2246}
2247
2248static 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
2261static int e_send_superseded(struct drbd_work *w, int unused)
2262{
2263 return e_send_ack(w, ack: P_SUPERSEDED);
2264}
2265
2266static 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, ack: connection->agreed_pro_version >= 100 ?
2273 P_RETRY_WRITE : P_SUPERSEDED);
2274}
2275
2276static 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
2286static u32 seq_max(u32 a, u32 b)
2287{
2288 return seq_greater(a, b) ? a : b;
2289}
2290
2291static 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(lock: &device->peer_seq_lock);
2298 newest_peer_seq = seq_max(a: device->peer_seq, b: peer_seq);
2299 device->peer_seq = newest_peer_seq;
2300 spin_unlock(lock: &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
2307static 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? */
2313static 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(lock: &device->resource->req_lock);
2319 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2320 if (overlaps(s1: peer_req->i.sector, l1: peer_req->i.size,
2321 s2: rs_req->i.sector, l2: rs_req->i.size)) {
2322 rv = true;
2323 break;
2324 }
2325 }
2326 spin_unlock_irq(lock: &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). */
2352static 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(lock: &device->peer_seq_lock);
2363 for (;;) {
2364 if (!seq_greater(a: peer_seq - 1, b: device->peer_seq)) {
2365 device->peer_seq = seq_max(a: device->peer_seq, b: 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(wq_head: &device->seq_wait, wq_entry: &wait, TASK_INTERRUPTIBLE);
2383 spin_unlock(lock: &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(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(lock: &device->peer_seq_lock);
2396 finish_wait(wq_head: &device->seq_wait, wq_entry: &wait);
2397 return ret;
2398}
2399
2400static 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() */
2411static 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
2419static 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, what: NEG_ACKED, peer_device, m: &m);
2437 spin_unlock_irq(lock: &device->resource->req_lock);
2438 if (m.bio)
2439 complete_master_bio(device, m: &m);
2440 spin_lock_irq(lock: &device->resource->req_lock);
2441 goto repeat;
2442 }
2443}
2444
2445static 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(new: &peer_req->w.list, head: &device->done_ee);
2503 queue_work(wq: connection->ack_sender, work: &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), flags: 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 */
2554static 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, volume_number: 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, dp: p, data_size: pi->size);
2576 atomic_inc(v: &connection->current_epoch->epoch_size);
2577 err2 = drbd_drain_block(peer_device, data_size: 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, id: 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, dpf: 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(lock: &connection->epoch_lock);
2626 peer_req->epoch = connection->current_epoch;
2627 atomic_inc(v: &peer_req->epoch->epoch_size);
2628 atomic_inc(v: &peer_req->epoch->active);
2629 spin_unlock(lock: &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(lock: &device->resource->req_lock);
2667 err = handle_write_conflicts(device, peer_req);
2668 if (err) {
2669 spin_unlock_irq(lock: &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(lock: &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(new: &peer_req->w.list, head: &device->active_ee);
2686 spin_unlock_irq(lock: &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, i: &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(lock: &device->resource->req_lock);
2706 list_del(entry: &peer_req->w.list);
2707 drbd_remove_epoch_entry_interval(device, peer_req);
2708 spin_unlock_irq(lock: &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, i: &peer_req->i);
2712 }
2713
2714out_interrupted:
2715 drbd_may_finish_epoch(connection, epoch: peer_req->epoch, ev: 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 */
2732bool 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(lock: &device->al_lock);
2743 tmp = lc_find(lc: 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(lock: &device->al_lock);
2752
2753 return throttle;
2754}
2755
2756bool 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(v: &device->rs_sect_ev);
2773
2774 if (atomic_read(v: &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
2802static 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, volume_number: pi->vnr);
2814 if (!peer_device)
2815 return -EIO;
2816 device = peer_device->device;
2817 capacity = get_capacity(disk: 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, rp: 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 , rp: 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, blksize: 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, data_size: 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, id: p->block_id, sector, request_size: size,
2865 payload_size: 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(size: 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(connection: peer_device->connection, buf: di->digest, size: 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(i: size >> 9, v: &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(lock: &device->resource->req_lock);
2976 list_add_tail(new: &peer_req->w.list, head: &device->read_ee);
2977 spin_unlock_irq(lock: &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, throttle_if_app_is_waiting: 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
2987submit_for_resync:
2988 atomic_add(i: size >> 9, v: &device->rs_sect_ev);
2989
2990submit:
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
2999out_free_e:
3000 spin_lock_irq(lock: &device->resource->req_lock);
3001 list_del(entry: &peer_req->w.list);
3002 spin_unlock_irq(lock: &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 */
3013static 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 */
3096static 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, f: CS_VERBOSE, NS(role, R_SECONDARY));
3137 if (rv2 != SS_SUCCESS) {
3138 drbd_khelper(device, cmd: "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 */
3153static 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, f: CS_VERBOSE, NS(role, R_SECONDARY));
3187 if (rv2 != SS_SUCCESS) {
3188 drbd_khelper(device, cmd: "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
3200static 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
3230static 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, text: "self", uuid: device->ldev->md.uuid,
3271 bits: device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, flags: 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, text: "peer", uuid: device->p_uuid, bits: device->p_uuid[UI_SIZE], flags: 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, text: "peer", uuid: device->p_uuid, bits: device->p_uuid[UI_SIZE], flags: 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, idx: UI_BITMAP, val: device->ldev->md.uuid[UI_HISTORY_START]);
3408 __drbd_uuid_set(device, idx: UI_HISTORY_START, val: 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, text: "self", uuid: device->ldev->md.uuid,
3412 bits: device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, flags: 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 */
3449static 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(lock: &device->ldev->md.uuid_lock);
3466 drbd_uuid_dump(device, text: "self", uuid: device->ldev->md.uuid, bits: device->comm_bm_set, flags: 0);
3467 drbd_uuid_dump(device, text: "peer", uuid: device->p_uuid,
3468 bits: device->p_uuid[UI_SIZE], flags: device->p_uuid[UI_FLAGS]);
3469
3470 hg = drbd_uuid_compare(peer_device, peer_role, rule_nr: &rule_nr);
3471 spin_unlock_irq(lock: &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, cmd: "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, cmd: "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, cmd: "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, io_fn: &drbd_bmio_set_n_write, why: "set_n_write from sync_handshake",
3596 flags: 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
3615static 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
3629static 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, buf: integrity_alg, size: 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(nr: CONN_DRY_RUN, addr: &connection->flags);
3660
3661 if (cf & CF_DRY_RUN)
3662 set_bit(nr: CONN_DRY_RUN, addr: &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(peer: 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(peer: 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(peer: 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(alg_name: integrity_alg, type: 0, mask: 0);
3718 if (IS_ERR(ptr: 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(tfm: peer_integrity_tfm);
3726 int_dig_in = kmalloc(size: hash_size, GFP_KERNEL);
3727 int_dig_vv = kmalloc(size: 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(size: 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(peer: p_after_sb_0p);
3745 new_net_conf->after_sb_1p = convert_after_sb(peer: p_after_sb_1p);
3746 new_net_conf->after_sb_2p = convert_after_sb(peer: 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(lock: &connection->resource->conf_update);
3751 mutex_unlock(lock: &connection->data.mutex);
3752
3753 crypto_free_shash(tfm: connection->peer_integrity_tfm);
3754 kfree(objp: connection->int_dig_in);
3755 kfree(objp: 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
3767disconnect_rcu_unlock:
3768 rcu_read_unlock();
3769disconnect:
3770 crypto_free_shash(tfm: peer_integrity_tfm);
3771 kfree(objp: int_dig_in);
3772 kfree(objp: int_dig_vv);
3773 conn_request_state(connection, NS(conn, C_DISCONNECTING), flags: 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. */
3782static 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_name: alg, type: 0, mask: 0);
3792 if (IS_ERR(ptr: 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
3800static 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, buf: buffer, size: 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 */
3831static 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
3838static 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, volume_number: 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(connection: peer_device->connection, buf: p, size: 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(size: sizeof(struct disk_conf), GFP_KERNEL);
3896 if (!new_disk_conf) {
3897 put_ldev(device);
3898 mutex_unlock(lock: &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(connection: peer_device->connection, buf: p->verify_alg, size: 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 alg: p->verify_alg, name: "verify-alg");
3943 if (IS_ERR(ptr: 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 alg: p->csums_alg, name: "csums-alg");
3957 if (IS_ERR(ptr: 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(size: 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(p: new_net_conf->verify_alg, q: p->verify_alg);
3989 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3990 crypto_free_shash(tfm: 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(p: new_net_conf->csums_alg, q: p->csums_alg);
3996 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3997 crypto_free_shash(tfm: 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(lock: &connection->resource->conf_update);
4016 synchronize_rcu();
4017 if (new_net_conf)
4018 kfree(objp: old_net_conf);
4019 kfree(objp: old_disk_conf);
4020 kfree(objp: old_plan);
4021
4022 return 0;
4023
4024reconnect:
4025 if (new_disk_conf) {
4026 put_ldev(device);
4027 kfree(objp: new_disk_conf);
4028 }
4029 mutex_unlock(lock: &connection->resource->conf_update);
4030 return -EIO;
4031
4032disconnect:
4033 kfree(objp: new_plan);
4034 if (new_disk_conf) {
4035 put_ldev(device);
4036 kfree(objp: new_disk_conf);
4037 }
4038 mutex_unlock(lock: &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(tfm: csums_tfm);
4042 /* but free the verify_tfm again, if csums_tfm did not work out */
4043 crypto_free_shash(tfm: verify_tfm);
4044 conn_request_state(connection: peer_device->connection, NS(conn, C_DISCONNECTING), flags: CS_HARD);
4045 return -EIO;
4046}
4047
4048/* warn if the arguments differ by more than 12.5% */
4049static 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
4061static 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, volume_number: pi->vnr);
4074 if (!peer_device)
4075 return config_unknown_volume(connection, pi);
4076 device = peer_device->device;
4077 cur_size = get_capacity(disk: 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, s: "lower level device sizes",
4093 a: p_size, b: drbd_get_max_capacity(bdev: device->ldev));
4094 warn_if_differ_considerably(device, s: "user requested size",
4095 a: p_usize, b: 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(connection: peer_device->connection, NS(conn, C_DISCONNECTING), flags: 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(size: 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(lock: &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, bdev: 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(connection: peer_device->connection, NS(conn, C_DISCONNECTING), flags: 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, size: new_size);
4197 }
4198 }
4199
4200 if (get_ldev(device)) {
4201 if (device->ldev->known_size != drbd_get_capacity(bdev: device->ldev->backing_bdev)) {
4202 device->ldev->known_size = drbd_get_capacity(bdev: 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(disk: device->vdisk) ||
4211 ldsc) {
4212 /* we have different sizes, probably peer
4213 * needs to know my new size... */
4214 drbd_send_sizes(peer_device, trigger_reply: 0, flags: ddsf);
4215 }
4216 if (test_and_clear_bit(nr: RESIZE_PENDING, addr: &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(nr: RESYNC_AFTER_NEG, addr: &device->flags);
4226 }
4227 }
4228
4229 return 0;
4230}
4231
4232static 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, volume_number: pi->vnr);
4241 if (!peer_device)
4242 return config_unknown_volume(connection, pi);
4243 device = peer_device->device;
4244
4245 p_uuid = kmalloc_array(n: UI_EXTENDED_SIZE, 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(objp: 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(connection: peer_device->connection, NS(conn, C_DISCONNECTING), flags: 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, io_fn: &drbd_bmio_clear_n_write,
4274 why: "clear_n_write from receive_uuids",
4275 flags: BM_LOCKED_TEST_ALLOWED, NULL);
4276 _drbd_uuid_set(device, idx: UI_CURRENT, val: p_uuid[UI_CURRENT]);
4277 _drbd_uuid_set(device, idx: UI_BITMAP, val: 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, val: 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(lock: device->state_mutex);
4297 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
4298 updated_uuids |= drbd_set_ed_uuid(device, val: p_uuid[UI_CURRENT]);
4299
4300 if (updated_uuids)
4301 drbd_print_uuids(device, text: "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 */
4310static 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
4337static 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, volume_number: 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(lock: device->state_mutex)) {
4355 drbd_send_sr_reply(peer_device, retcode: SS_CONCURRENT_ST_CHG);
4356 return 0;
4357 }
4358
4359 mask = convert_state(ps: mask);
4360 val = convert_state(ps: val);
4361
4362 rv = drbd_change_state(device, f: CS_VERBOSE, mask, val);
4363 drbd_send_sr_reply(peer_device, retcode: rv);
4364
4365 drbd_md_sync(device);
4366
4367 return 0;
4368}
4369
4370static 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(lock: &connection->cstate_mutex)) {
4381 conn_send_sr_reply(connection, retcode: SS_CONCURRENT_ST_CHG);
4382 return 0;
4383 }
4384
4385 mask = convert_state(ps: mask);
4386 val = convert_state(ps: val);
4387
4388 rv = conn_request_state(connection, mask, val, flags: CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4389 conn_send_sr_reply(connection, retcode: rv);
4390
4391 return 0;
4392}
4393
4394static 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, volume_number: 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(lock: &device->resource->req_lock);
4418 retry:
4419 os = ns = drbd_read_state(device);
4420 spin_unlock_irq(lock: &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_role: peer_state.role, peer_disk: 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(nr: CONN_DRY_RUN, addr: &peer_device->connection->flags))
4534 return -EIO;
4535 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4536 conn_request_state(connection: peer_device->connection, NS(conn, C_DISCONNECTING), flags: CS_HARD);
4537 return -EIO;
4538 }
4539 }
4540 }
4541
4542 spin_lock_irq(lock: &device->resource->req_lock);
4543 if (os.i != drbd_read_state(device).i)
4544 goto retry;
4545 clear_bit(nr: CONSIDER_RESYNC, addr: &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(lock: &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(nr: NEW_CUR_UUID, addr: &device->flags);
4561 conn_request_state(connection: peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), flags: 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(lock: &device->resource->req_lock);
4567
4568 if (rv < SS_SUCCESS) {
4569 conn_request_state(connection: peer_device->connection, NS(conn, C_DISCONNECTING), flags: 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(nr: DISCARD_MY_DATA, addr: &device->flags);
4585
4586 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4587
4588 return 0;
4589}
4590
4591static 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, volume_number: 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, idx: UI_CURRENT, be64_to_cpu(p->uuid));
4614 _drbd_uuid_set(device, idx: UI_BITMAP, val: 0UL);
4615
4616 drbd_print_uuids(device, text: "updated sync uuid");
4617 drbd_start_resync(device, side: 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 */
4632static int
4633receive_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(connection: 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(connection: peer_device->connection, buf: p, size: want);
4650 if (err)
4651 return err;
4652
4653 drbd_bm_merge_lel(device: peer_device->device, offset: c->word_offset, number: num_words, buffer: 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
4663static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4664{
4665 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4666}
4667
4668static int dcbp_get_start(struct p_compressed_bm *p)
4669{
4670 return (p->encoding & 0x80) != 0;
4671}
4672
4673static 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 */
4684static int
4685recv_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: &bs, s: p->code, len, pad_bits: dcbp_get_pad_bits(p));
4701
4702 bits = bitstream_get_bits(bs: &bs, out: &look_ahead, bits: 64);
4703 if (bits < 0)
4704 return -EIO;
4705
4706 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4707 bits = vli_decode_bits(out: &rl, in: 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(device: 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: &bs, out: &tmp, bits: 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 */
4753static int
4754decode_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: 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(connection: peer_device->connection, NS(conn, C_PROTOCOL_ERROR), flags: CS_HARD);
4768 return -EIO;
4769}
4770
4771void 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(connection: 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. */
4815static 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, volume_number: pi->vnr);
4823 if (!peer_device)
4824 return -EIO;
4825 device = peer_device->device;
4826
4827 drbd_bm_lock(device, why: "receive bitmap", flags: 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, size: pi->size, p: pi->data, c: &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(connection: peer_device->connection, buf: p, size: pi->size);
4855 if (err)
4856 goto out;
4857 err = decode_bitmap_c(peer_device, p, c: &c, len: 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(connection: peer_device->connection, pi);
4873 if (err)
4874 goto out;
4875 }
4876
4877 INFO_bm_xfer_stats(peer_device, direction: "receive", c: &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, side: C_SYNC_SOURCE);
4900 return err;
4901}
4902
4903static 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
4911static 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(sk: connection->data.socket->sk, val: 2);
4916 return 0;
4917}
4918
4919static 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, volume_number: 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
4945static 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, volume_number: 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 request_size: size, payload_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(lock: &device->resource->req_lock);
4979 list_add_tail(new: &peer_req->w.list, head: &device->sync_ee);
4980 spin_unlock_irq(lock: &device->resource->req_lock);
4981
4982 atomic_add(i: pi->size >> 9, v: &device->rs_sect_ev);
4983 err = drbd_submit_peer_request(peer_req);
4984
4985 if (err) {
4986 spin_lock_irq(lock: &device->resource->req_lock);
4987 list_del(entry: &peer_req->w.list);
4988 spin_unlock_irq(lock: &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, blksize: size, ID_SYNCER);
5004 }
5005
5006 atomic_add(i: size >> 9, v: &device->rs_sect_in);
5007
5008 return err;
5009}
5010
5011struct data_cmd {
5012 int expect_payload;
5013 unsigned int pkt_size;
5014 int (*fn)(struct drbd_connection *, struct packet_info *);
5015};
5016
5017static struct data_cmd drbd_cmd_handler[] = {
5018 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
5019 [P_DATA_REPLY] = { .expect_payload: 1, .pkt_size: sizeof(struct p_data), .fn: receive_DataReply },
5020 [P_RS_DATA_REPLY] = { .expect_payload: 1, .pkt_size: sizeof(struct p_data), .fn: receive_RSDataReply } ,
5021 [P_BARRIER] = { .expect_payload: 0, .pkt_size: sizeof(struct p_barrier), .fn: receive_Barrier } ,
5022 [P_BITMAP] = { .expect_payload: 1, .pkt_size: 0, .fn: receive_bitmap } ,
5023 [P_COMPRESSED_BITMAP] = { .expect_payload: 1, .pkt_size: 0, .fn: receive_bitmap } ,
5024 [P_UNPLUG_REMOTE] = { .expect_payload: 0, .pkt_size: 0, .fn: receive_UnplugRemote },
5025 [P_DATA_REQUEST] = { .expect_payload: 0, .pkt_size: sizeof(struct p_block_req), .fn: receive_DataRequest },
5026 [P_RS_DATA_REQUEST] = { .expect_payload: 0, .pkt_size: sizeof(struct p_block_req), .fn: receive_DataRequest },
5027 [P_SYNC_PARAM] = { .expect_payload: 1, .pkt_size: 0, .fn: receive_SyncParam },
5028 [P_SYNC_PARAM89] = { .expect_payload: 1, .pkt_size: 0, .fn: receive_SyncParam },
5029 [P_PROTOCOL] = { .expect_payload: 1, .pkt_size: sizeof(struct p_protocol), .fn: receive_protocol },
5030 [P_UUIDS] = { .expect_payload: 0, .pkt_size: sizeof(struct p_uuids), .fn: receive_uuids },
5031 [P_SIZES] = { .expect_payload: 0, .pkt_size: sizeof(struct p_sizes), .fn: receive_sizes },
5032 [P_STATE] = { .expect_payload: 0, .pkt_size: sizeof(struct p_state), .fn: receive_state },
5033 [P_STATE_CHG_REQ] = { .expect_payload: 0, .pkt_size: sizeof(struct p_req_state), .fn: receive_req_state },
5034 [P_SYNC_UUID] = { .expect_payload: 0, .pkt_size: sizeof(struct p_rs_uuid), .fn: receive_sync_uuid },
5035 [P_OV_REQUEST] = { .expect_payload: 0, .pkt_size: sizeof(struct p_block_req), .fn: receive_DataRequest },
5036 [P_OV_REPLY] = { .expect_payload: 1, .pkt_size: sizeof(struct p_block_req), .fn: receive_DataRequest },
5037 [P_CSUM_RS_REQUEST] = { .expect_payload: 1, .pkt_size: sizeof(struct p_block_req), .fn: receive_DataRequest },
5038 [P_RS_THIN_REQ] = { .expect_payload: 0, .pkt_size: sizeof(struct p_block_req), .fn: receive_DataRequest },
5039 [P_DELAY_PROBE] = { .expect_payload: 0, .pkt_size: sizeof(struct p_delay_probe93), .fn: receive_skip },
5040 [P_OUT_OF_SYNC] = { .expect_payload: 0, .pkt_size: sizeof(struct p_block_desc), .fn: receive_out_of_sync },
5041 [P_CONN_ST_CHG_REQ] = { .expect_payload: 0, .pkt_size: sizeof(struct p_req_state), .fn: receive_req_conn_state },
5042 [P_PROTOCOL_UPDATE] = { .expect_payload: 1, .pkt_size: sizeof(struct p_protocol), .fn: receive_protocol },
5043 [P_TRIM] = { .expect_payload: 0, .pkt_size: sizeof(struct p_trim), .fn: receive_Data },
5044 [P_ZEROES] = { .expect_payload: 0, .pkt_size: sizeof(struct p_trim), .fn: receive_Data },
5045 [P_RS_DEALLOCATED] = { .expect_payload: 0, .pkt_size: sizeof(struct p_block_desc), .fn: receive_rs_deallocated },
5046};
5047
5048static 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(thi: &connection->receiver) == RUNNING) {
5055 struct data_cmd const *cmd;
5056
5057 drbd_thread_current_set_cpu(thi: &connection->receiver);
5058 update_receiver_timing_details(connection, drbd_recv_header_maybe_unplug);
5059 if (drbd_recv_header_maybe_unplug(connection, pi: &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, buf: pi.data, size: 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), flags: CS_HARD);
5103}
5104
5105static 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), flags: CS_HARD);
5120
5121 /* ack_receiver does not clean up anything. it must not interfere, either */
5122 drbd_thread_stop(thi: &connection->ack_receiver);
5123 if (connection->ack_sender) {
5124 destroy_workqueue(wq: 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(kref: &device->kref);
5133 rcu_read_unlock();
5134 drbd_disconnected(peer_device);
5135 kref_put(kref: &device->kref, release: drbd_destroy_device);
5136 rcu_read_lock();
5137 }
5138 rcu_read_unlock();
5139
5140 if (!list_empty(head: &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(v: &connection->current_epoch->epoch_size, i: 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(lock: &connection->resource->req_lock);
5152 oc = connection->cstate;
5153 if (oc >= C_UNCONNECTED)
5154 _conn_request_state(connection, NS(conn, C_UNCONNECTED), flags: CS_VERBOSE);
5155
5156 spin_unlock_irq(lock: &connection->resource->req_lock);
5157
5158 if (oc == C_DISCONNECTING)
5159 conn_request_state(connection, NS(conn, C_STANDALONE), flags: CS_VERBOSE | CS_HARD);
5160}
5161
5162static 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(lock: &device->resource->req_lock);
5169 _drbd_wait_ee_list_empty(device, head: &device->active_ee);
5170 _drbd_wait_ee_list_empty(device, head: &device->sync_ee);
5171 _drbd_wait_ee_list_empty(device, head: &device->read_ee);
5172 spin_unlock_irq(lock: &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(v: &device->rs_pending_cnt, i: 0);
5188 wake_up(&device->misc_wait);
5189
5190 del_timer_sync(timer: &device->resync_timer);
5191 resync_timer_fn(t: &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(work_queue: &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(work_queue: &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(objp: 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, io_fn: &drbd_bm_write_copy_pages,
5219 why: "write from disconnected", flags: 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, list: &device->net_ee);
5231 if (i)
5232 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
5233 i = atomic_read(v: &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(v: &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 */
5257static 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 */
5280static 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: &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, buf: p, size: 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)
5347static 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
5362static 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(size: sizeof(struct shash_desc) +
5386 crypto_shash_descsize(tfm: 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(tfm: connection->cram_hmac_tfm, key: (u8 *)secret, keylen: 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(buf: 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: &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(size: pi.size, GFP_NOIO);
5439 if (!peers_ch) {
5440 rv = -1;
5441 goto fail;
5442 }
5443
5444 err = drbd_recv_all_warn(connection, buf: peers_ch, size: pi.size);
5445 if (err) {
5446 rv = 0;
5447 goto fail;
5448 }
5449
5450 if (!memcmp(p: my_challenge, q: 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(tfm: connection->cram_hmac_tfm);
5457 response = kmalloc(size: resp_size, GFP_NOIO);
5458 if (!response) {
5459 rv = -1;
5460 goto fail;
5461 }
5462
5463 rv = crypto_shash_digest(desc, data: peers_ch, len: pi.size, out: 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: &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, buf: response , size: resp_size);
5499 if (err) {
5500 rv = 0;
5501 goto fail;
5502 }
5503
5504 right_response = kmalloc(size: resp_size, GFP_NOIO);
5505 if (!right_response) {
5506 rv = -1;
5507 goto fail;
5508 }
5509
5510 rv = crypto_shash_digest(desc, data: my_challenge, CHALLENGE_LEN,
5511 out: 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(p: response, q: right_response, size: 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(objp: peers_ch);
5528 kfree(objp: response);
5529 kfree(objp: right_response);
5530 if (desc) {
5531 shash_desc_zero(desc);
5532 kfree(objp: desc);
5533 }
5534
5535 return rv;
5536}
5537#endif
5538
5539int 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), flags: 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
5572static 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(nr: CONN_WD_ST_CHG_OKAY, addr: &connection->flags);
5579 } else {
5580 set_bit(nr: CONN_WD_ST_CHG_FAIL, addr: &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
5589static 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, volume_number: 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(nr: CL_ST_CHG_SUCCESS, addr: &device->flags);
5608 } else {
5609 set_bit(nr: CL_ST_CHG_FAIL, addr: &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
5618static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5619{
5620 return drbd_send_ping_ack(connection);
5621
5622}
5623
5624static 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(nr: GOT_PING_ACK, addr: &connection->flags))
5629 wake_up(&connection->ping_wait);
5630
5631 return 0;
5632}
5633
5634static 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, volume_number: 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(i: blksize >> 9, v: &device->rs_sect_in);
5660
5661 return 0;
5662}
5663
5664static int
5665validate_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(lock: &device->resource->req_lock);
5674 req = find_request(device, root, id, sector, missing_ok, func);
5675 if (unlikely(!req)) {
5676 spin_unlock_irq(lock: &device->resource->req_lock);
5677 return -EIO;
5678 }
5679 __req_mod(req, what, peer_device, m: &m);
5680 spin_unlock_irq(lock: &device->resource->req_lock);
5681
5682 if (m.bio)
5683 complete_master_bio(device, m: &m);
5684 return 0;
5685}
5686
5687static 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, volume_number: 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, id: p->block_id, sector,
5729 root: &device->write_requests, func: __func__,
5730 what, missing_ok: false);
5731}
5732
5733static 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, volume_number: 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, id: p->block_id, sector,
5756 root: &device->write_requests, func: __func__,
5757 what: NEG_ACKED, missing_ok: 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
5769static 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, volume_number: 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, id: p->block_id, sector,
5787 root: &device->read_requests, func: __func__,
5788 what: NEG_ACKED, missing_ok: false);
5789}
5790
5791static 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, volume_number: 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
5828static 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, barrier_nr: 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(v: &device->ap_in_flight) == 0 &&
5842 !test_and_set_bit(nr: AHEAD_TO_SYNC_SOURCE, addr: &device->flags)) {
5843 device->start_resync_timer.expires = jiffies + HZ;
5844 add_timer(timer: &device->start_resync_timer);
5845 }
5846 }
5847 rcu_read_unlock();
5848
5849 return 0;
5850}
5851
5852static 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, volume_number: 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, still_to_go: device->ov_left);
5887
5888 if (device->ov_left == 0) {
5889 dw = kmalloc(size: sizeof(*dw), GFP_NOIO);
5890 if (dw) {
5891 dw->w.cb = w_ov_finished;
5892 dw->device = device;
5893 drbd_queue_work(q: &peer_device->connection->sender_work, w: &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
5904static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5905{
5906 return 0;
5907}
5908
5909struct meta_sock_cmd {
5910 size_t pkt_size;
5911 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5912};
5913
5914static 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
5931static void set_ping_timeout(struct drbd_connection *connection)
5932{
5933 set_rcvtimeo(connection, ping_timeout: 1);
5934}
5935
5936static void set_idle_timeout(struct drbd_connection *connection)
5937{
5938 set_rcvtimeo(connection, ping_timeout: 0);
5939}
5940
5941static struct meta_sock_cmd ack_receiver_tbl[] = {
5942 [P_PING] = { 0, got_Ping },
5943 [P_PING_ACK] = { .pkt_size: 0, .fn: got_PingAck },
5944 [P_RECV_ACK] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_BlockAck },
5945 [P_WRITE_ACK] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_BlockAck },
5946 [P_RS_WRITE_ACK] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_BlockAck },
5947 [P_SUPERSEDED] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_BlockAck },
5948 [P_NEG_ACK] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_NegAck },
5949 [P_NEG_DREPLY] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_NegDReply },
5950 [P_NEG_RS_DREPLY] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_NegRSDReply },
5951 [P_OV_RESULT] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_OVResult },
5952 [P_BARRIER_ACK] = { .pkt_size: sizeof(struct p_barrier_ack), .fn: got_BarrierAck },
5953 [P_STATE_CHG_REPLY] = { .pkt_size: sizeof(struct p_req_state_reply), .fn: got_RqSReply },
5954 [P_RS_IS_IN_SYNC] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_IsInSync },
5955 [P_DELAY_PROBE] = { .pkt_size: sizeof(struct p_delay_probe93), .fn: got_skip },
5956 [P_RS_CANCEL] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_NegRSDReply },
5957 [P_CONN_ST_CHG_REPLY]={ .pkt_size: sizeof(struct p_req_state_reply), .fn: got_conn_RqSReply },
5958 [P_RETRY_WRITE] = { .pkt_size: sizeof(struct p_block_ack), .fn: got_BlockAck },
5959};
5960
5961int 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(nr: SEND_PING, addr: &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(sock: connection->meta.socket, buf, size: expect-received, flags: 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(nr: SEND_PING, addr: &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, header: connection->meta.rbuf, pi: &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) {
6083reconnect:
6084 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), flags: CS_HARD);
6085 conn_md_sync(connection);
6086 }
6087 if (0) {
6088disconnect:
6089 conn_request_state(connection, NS(conn, C_DISCONNECTING), flags: CS_HARD);
6090 }
6091
6092 drbd_info(connection, "ack_receiver terminated\n");
6093
6094 return 0;
6095}
6096
6097void 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(sk: connection->meta.socket->sk, on: true);
6113
6114 err = drbd_finish_peer_reqs(device);
6115 kref_put(kref: &device->kref, release: 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), flags: CS_HARD);
6121 return;
6122 }
6123
6124 if (tcp_cork)
6125 tcp_sock_set_cork(sk: connection->meta.socket->sk, on: false);
6126
6127 return;
6128}
6129

source code of linux/drivers/block/drbd/drbd_receiver.c