1 // SPDX-License-Identifier: GPL-2.0-or-later
5 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
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>.
14 #include <linux/module.h>
16 #include <linux/uaccess.h>
19 #include <linux/drbd.h>
21 #include <linux/file.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 #define __KERNEL_SYSCALLS__
31 #include <linux/unistd.h>
32 #include <linux/vmalloc.h>
33 #include <linux/random.h>
34 #include <linux/string.h>
35 #include <linux/scatterlist.h>
36 #include <linux/part_stat.h>
38 #include "drbd_protocol.h"
42 #define PRO_FEATURES (DRBD_FF_TRIM|DRBD_FF_THIN_RESYNC|DRBD_FF_WSAME|DRBD_FF_WZEROES)
57 static int drbd_do_features(struct drbd_connection *connection);
58 static int drbd_do_auth(struct drbd_connection *connection);
59 static int drbd_disconnected(struct drbd_peer_device *);
60 static void conn_wait_active_ee_empty(struct drbd_connection *connection);
61 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
62 static int e_end_block(struct drbd_work *, int);
65 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
68 * some helper functions to deal with single linked page lists,
69 * page->private being our "next" pointer.
72 /* If at least n pages are linked at head, get n pages off.
73 * Otherwise, don't modify head, and return NULL.
74 * Locking is the responsibility of the caller.
76 static struct page *page_chain_del(struct page **head, int n)
90 tmp = page_chain_next(page);
92 break; /* found sufficient pages */
94 /* insufficient pages, don't use any of them. */
99 /* add end of list marker for the returned list */
100 set_page_private(page, 0);
101 /* actual return value, and adjustment of head */
107 /* may be used outside of locks to find the tail of a (usually short)
108 * "private" page chain, before adding it back to a global chain head
109 * with page_chain_add() under a spinlock. */
110 static struct page *page_chain_tail(struct page *page, int *len)
114 while ((tmp = page_chain_next(page))) {
123 static int page_chain_free(struct page *page)
127 page_chain_for_each_safe(page, tmp) {
134 static void page_chain_add(struct page **head,
135 struct page *chain_first, struct page *chain_last)
139 tmp = page_chain_tail(chain_first, NULL);
140 BUG_ON(tmp != chain_last);
143 /* add chain to head */
144 set_page_private(chain_last, (unsigned long)*head);
148 static struct page *__drbd_alloc_pages(struct drbd_device *device,
151 struct page *page = NULL;
152 struct page *tmp = NULL;
155 /* Yes, testing drbd_pp_vacant outside the lock is racy.
156 * So what. It saves a spin_lock. */
157 if (drbd_pp_vacant >= number) {
158 spin_lock(&drbd_pp_lock);
159 page = page_chain_del(&drbd_pp_pool, number);
161 drbd_pp_vacant -= number;
162 spin_unlock(&drbd_pp_lock);
167 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
168 * "criss-cross" setup, that might cause write-out on some other DRBD,
169 * which in turn might block on the other node at this very place. */
170 for (i = 0; i < number; i++) {
171 tmp = alloc_page(GFP_TRY);
174 set_page_private(tmp, (unsigned long)page);
181 /* Not enough pages immediately available this time.
182 * No need to jump around here, drbd_alloc_pages will retry this
183 * function "soon". */
185 tmp = page_chain_tail(page, NULL);
186 spin_lock(&drbd_pp_lock);
187 page_chain_add(&drbd_pp_pool, page, tmp);
189 spin_unlock(&drbd_pp_lock);
194 static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
195 struct list_head *to_be_freed)
197 struct drbd_peer_request *peer_req, *tmp;
199 /* The EEs are always appended to the end of the list. Since
200 they are sent in order over the wire, they have to finish
201 in order. As soon as we see the first not finished we can
202 stop to examine the list... */
204 list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
205 if (drbd_peer_req_has_active_page(peer_req))
207 list_move(&peer_req->w.list, to_be_freed);
211 static void drbd_reclaim_net_peer_reqs(struct drbd_device *device)
213 LIST_HEAD(reclaimed);
214 struct drbd_peer_request *peer_req, *t;
216 spin_lock_irq(&device->resource->req_lock);
217 reclaim_finished_net_peer_reqs(device, &reclaimed);
218 spin_unlock_irq(&device->resource->req_lock);
219 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
220 drbd_free_net_peer_req(device, peer_req);
223 static void conn_reclaim_net_peer_reqs(struct drbd_connection *connection)
225 struct drbd_peer_device *peer_device;
229 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
230 struct drbd_device *device = peer_device->device;
231 if (!atomic_read(&device->pp_in_use_by_net))
234 kref_get(&device->kref);
236 drbd_reclaim_net_peer_reqs(device);
237 kref_put(&device->kref, drbd_destroy_device);
244 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
245 * @peer_device: DRBD device.
246 * @number: number of pages requested
247 * @retry: whether to retry, if not enough pages are available right now
249 * Tries to allocate number pages, first from our own page pool, then from
251 * Possibly retry until DRBD frees sufficient pages somewhere else.
253 * If this allocation would exceed the max_buffers setting, we throttle
254 * allocation (schedule_timeout) to give the system some room to breathe.
256 * We do not use max-buffers as hard limit, because it could lead to
257 * congestion and further to a distributed deadlock during online-verify or
258 * (checksum based) resync, if the max-buffers, socket buffer sizes and
259 * resync-rate settings are mis-configured.
261 * Returns a page chain linked via page->private.
263 struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
266 struct drbd_device *device = peer_device->device;
267 struct page *page = NULL;
273 nc = rcu_dereference(peer_device->connection->net_conf);
274 mxb = nc ? nc->max_buffers : 1000000;
277 if (atomic_read(&device->pp_in_use) < mxb)
278 page = __drbd_alloc_pages(device, number);
280 /* Try to keep the fast path fast, but occasionally we need
281 * to reclaim the pages we lended to the network stack. */
282 if (page && atomic_read(&device->pp_in_use_by_net) > 512)
283 drbd_reclaim_net_peer_reqs(device);
285 while (page == NULL) {
286 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
288 drbd_reclaim_net_peer_reqs(device);
290 if (atomic_read(&device->pp_in_use) < mxb) {
291 page = __drbd_alloc_pages(device, number);
299 if (signal_pending(current)) {
300 drbd_warn(device, "drbd_alloc_pages interrupted!\n");
304 if (schedule_timeout(HZ/10) == 0)
307 finish_wait(&drbd_pp_wait, &wait);
310 atomic_add(number, &device->pp_in_use);
314 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
315 * Is also used from inside an other spin_lock_irq(&resource->req_lock);
316 * Either links the page chain back to the global pool,
317 * or returns all pages to the system. */
318 static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
320 atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
326 if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * drbd_minor_count)
327 i = page_chain_free(page);
330 tmp = page_chain_tail(page, &i);
331 spin_lock(&drbd_pp_lock);
332 page_chain_add(&drbd_pp_pool, page, tmp);
334 spin_unlock(&drbd_pp_lock);
336 i = atomic_sub_return(i, a);
338 drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
339 is_net ? "pp_in_use_by_net" : "pp_in_use", i);
340 wake_up(&drbd_pp_wait);
344 You need to hold the req_lock:
345 _drbd_wait_ee_list_empty()
347 You must not have the req_lock:
349 drbd_alloc_peer_req()
350 drbd_free_peer_reqs()
352 drbd_finish_peer_reqs()
354 drbd_wait_ee_list_empty()
357 /* normal: payload_size == request size (bi_size)
358 * w_same: payload_size == logical_block_size
359 * trim: payload_size == 0 */
360 struct drbd_peer_request *
361 drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
362 unsigned int request_size, unsigned int payload_size, gfp_t gfp_mask) __must_hold(local)
364 struct drbd_device *device = peer_device->device;
365 struct drbd_peer_request *peer_req;
366 struct page *page = NULL;
367 unsigned int nr_pages = PFN_UP(payload_size);
369 if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
372 peer_req = mempool_alloc(&drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
374 if (!(gfp_mask & __GFP_NOWARN))
375 drbd_err(device, "%s: allocation failed\n", __func__);
380 page = drbd_alloc_pages(peer_device, nr_pages,
381 gfpflags_allow_blocking(gfp_mask));
386 memset(peer_req, 0, sizeof(*peer_req));
387 INIT_LIST_HEAD(&peer_req->w.list);
388 drbd_clear_interval(&peer_req->i);
389 peer_req->i.size = request_size;
390 peer_req->i.sector = sector;
391 peer_req->submit_jif = jiffies;
392 peer_req->peer_device = peer_device;
393 peer_req->pages = page;
395 * The block_id is opaque to the receiver. It is not endianness
396 * converted, and sent back to the sender unchanged.
398 peer_req->block_id = id;
403 mempool_free(peer_req, &drbd_ee_mempool);
407 void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
411 if (peer_req->flags & EE_HAS_DIGEST)
412 kfree(peer_req->digest);
413 drbd_free_pages(device, peer_req->pages, is_net);
414 D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
415 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
416 if (!expect(!(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
417 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
418 drbd_al_complete_io(device, &peer_req->i);
420 mempool_free(peer_req, &drbd_ee_mempool);
423 int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
425 LIST_HEAD(work_list);
426 struct drbd_peer_request *peer_req, *t;
428 int is_net = list == &device->net_ee;
430 spin_lock_irq(&device->resource->req_lock);
431 list_splice_init(list, &work_list);
432 spin_unlock_irq(&device->resource->req_lock);
434 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
435 __drbd_free_peer_req(device, peer_req, is_net);
442 * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
444 static int drbd_finish_peer_reqs(struct drbd_device *device)
446 LIST_HEAD(work_list);
447 LIST_HEAD(reclaimed);
448 struct drbd_peer_request *peer_req, *t;
451 spin_lock_irq(&device->resource->req_lock);
452 reclaim_finished_net_peer_reqs(device, &reclaimed);
453 list_splice_init(&device->done_ee, &work_list);
454 spin_unlock_irq(&device->resource->req_lock);
456 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
457 drbd_free_net_peer_req(device, peer_req);
459 /* possible callbacks here:
460 * e_end_block, and e_end_resync_block, e_send_superseded.
461 * all ignore the last argument.
463 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
466 /* list_del not necessary, next/prev members not touched */
467 err2 = peer_req->w.cb(&peer_req->w, !!err);
470 drbd_free_peer_req(device, peer_req);
472 wake_up(&device->ee_wait);
477 static void _drbd_wait_ee_list_empty(struct drbd_device *device,
478 struct list_head *head)
482 /* avoids spin_lock/unlock
483 * and calling prepare_to_wait in the fast path */
484 while (!list_empty(head)) {
485 prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
486 spin_unlock_irq(&device->resource->req_lock);
488 finish_wait(&device->ee_wait, &wait);
489 spin_lock_irq(&device->resource->req_lock);
493 static void drbd_wait_ee_list_empty(struct drbd_device *device,
494 struct list_head *head)
496 spin_lock_irq(&device->resource->req_lock);
497 _drbd_wait_ee_list_empty(device, head);
498 spin_unlock_irq(&device->resource->req_lock);
501 static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
507 struct msghdr msg = {
508 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
510 iov_iter_kvec(&msg.msg_iter, READ, &iov, 1, size);
511 return sock_recvmsg(sock, &msg, msg.msg_flags);
514 static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
518 rv = drbd_recv_short(connection->data.socket, buf, size, 0);
521 if (rv == -ECONNRESET)
522 drbd_info(connection, "sock was reset by peer\n");
523 else if (rv != -ERESTARTSYS)
524 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
525 } else if (rv == 0) {
526 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
529 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
532 t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);
537 drbd_info(connection, "sock was shut down by peer\n");
541 conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
547 static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
551 err = drbd_recv(connection, buf, size);
560 static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
564 err = drbd_recv_all(connection, buf, size);
565 if (err && !signal_pending(current))
566 drbd_warn(connection, "short read (expected size %d)\n", (int)size);
571 * On individual connections, the socket buffer size must be set prior to the
572 * listen(2) or connect(2) calls in order to have it take effect.
573 * This is our wrapper to do so.
575 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
578 /* open coded SO_SNDBUF, SO_RCVBUF */
580 sock->sk->sk_sndbuf = snd;
581 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
584 sock->sk->sk_rcvbuf = rcv;
585 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
589 static struct socket *drbd_try_connect(struct drbd_connection *connection)
593 struct sockaddr_in6 src_in6;
594 struct sockaddr_in6 peer_in6;
596 int err, peer_addr_len, my_addr_len;
597 int sndbuf_size, rcvbuf_size, connect_int;
598 int disconnect_on_error = 1;
601 nc = rcu_dereference(connection->net_conf);
606 sndbuf_size = nc->sndbuf_size;
607 rcvbuf_size = nc->rcvbuf_size;
608 connect_int = nc->connect_int;
611 my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
612 memcpy(&src_in6, &connection->my_addr, my_addr_len);
614 if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
615 src_in6.sin6_port = 0;
617 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
619 peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
620 memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);
622 what = "sock_create_kern";
623 err = sock_create_kern(&init_net, ((struct sockaddr *)&src_in6)->sa_family,
624 SOCK_STREAM, IPPROTO_TCP, &sock);
630 sock->sk->sk_rcvtimeo =
631 sock->sk->sk_sndtimeo = connect_int * HZ;
632 drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);
634 /* explicitly bind to the configured IP as source IP
635 * for the outgoing connections.
636 * This is needed for multihomed hosts and to be
637 * able to use lo: interfaces for drbd.
638 * Make sure to use 0 as port number, so linux selects
639 * a free one dynamically.
641 what = "bind before connect";
642 err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
646 /* connect may fail, peer not yet available.
647 * stay C_WF_CONNECTION, don't go Disconnecting! */
648 disconnect_on_error = 0;
650 err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);
659 /* timeout, busy, signal pending */
660 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
661 case EINTR: case ERESTARTSYS:
662 /* peer not (yet) available, network problem */
663 case ECONNREFUSED: case ENETUNREACH:
664 case EHOSTDOWN: case EHOSTUNREACH:
665 disconnect_on_error = 0;
668 drbd_err(connection, "%s failed, err = %d\n", what, err);
670 if (disconnect_on_error)
671 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
677 struct accept_wait_data {
678 struct drbd_connection *connection;
679 struct socket *s_listen;
680 struct completion door_bell;
681 void (*original_sk_state_change)(struct sock *sk);
685 static void drbd_incoming_connection(struct sock *sk)
687 struct accept_wait_data *ad = sk->sk_user_data;
688 void (*state_change)(struct sock *sk);
690 state_change = ad->original_sk_state_change;
691 if (sk->sk_state == TCP_ESTABLISHED)
692 complete(&ad->door_bell);
696 static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
698 int err, sndbuf_size, rcvbuf_size, my_addr_len;
699 struct sockaddr_in6 my_addr;
700 struct socket *s_listen;
705 nc = rcu_dereference(connection->net_conf);
710 sndbuf_size = nc->sndbuf_size;
711 rcvbuf_size = nc->rcvbuf_size;
714 my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
715 memcpy(&my_addr, &connection->my_addr, my_addr_len);
717 what = "sock_create_kern";
718 err = sock_create_kern(&init_net, ((struct sockaddr *)&my_addr)->sa_family,
719 SOCK_STREAM, IPPROTO_TCP, &s_listen);
725 s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
726 drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);
728 what = "bind before listen";
729 err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
733 ad->s_listen = s_listen;
734 write_lock_bh(&s_listen->sk->sk_callback_lock);
735 ad->original_sk_state_change = s_listen->sk->sk_state_change;
736 s_listen->sk->sk_state_change = drbd_incoming_connection;
737 s_listen->sk->sk_user_data = ad;
738 write_unlock_bh(&s_listen->sk->sk_callback_lock);
741 err = s_listen->ops->listen(s_listen, 5);
748 sock_release(s_listen);
750 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
751 drbd_err(connection, "%s failed, err = %d\n", what, err);
752 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
759 static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
761 write_lock_bh(&sk->sk_callback_lock);
762 sk->sk_state_change = ad->original_sk_state_change;
763 sk->sk_user_data = NULL;
764 write_unlock_bh(&sk->sk_callback_lock);
767 static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
769 int timeo, connect_int, err = 0;
770 struct socket *s_estab = NULL;
774 nc = rcu_dereference(connection->net_conf);
779 connect_int = nc->connect_int;
782 timeo = connect_int * HZ;
783 /* 28.5% random jitter */
784 timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;
786 err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
790 err = kernel_accept(ad->s_listen, &s_estab, 0);
792 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
793 drbd_err(connection, "accept failed, err = %d\n", err);
794 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
799 unregister_state_change(s_estab->sk, ad);
804 static int decode_header(struct drbd_connection *, void *, struct packet_info *);
806 static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
807 enum drbd_packet cmd)
809 if (!conn_prepare_command(connection, sock))
811 return conn_send_command(connection, sock, cmd, 0, NULL, 0);
814 static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
816 unsigned int header_size = drbd_header_size(connection);
817 struct packet_info pi;
822 nc = rcu_dereference(connection->net_conf);
827 sock->sk->sk_rcvtimeo = nc->ping_timeo * 4 * HZ / 10;
830 err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
831 if (err != header_size) {
836 err = decode_header(connection, connection->data.rbuf, &pi);
843 * drbd_socket_okay() - Free the socket if its connection is not okay
844 * @sock: pointer to the pointer to the socket.
846 static bool drbd_socket_okay(struct socket **sock)
854 rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
856 if (rr > 0 || rr == -EAGAIN) {
865 static bool connection_established(struct drbd_connection *connection,
866 struct socket **sock1,
867 struct socket **sock2)
873 if (!*sock1 || !*sock2)
877 nc = rcu_dereference(connection->net_conf);
878 timeout = (nc->sock_check_timeo ?: nc->ping_timeo) * HZ / 10;
880 schedule_timeout_interruptible(timeout);
882 ok = drbd_socket_okay(sock1);
883 ok = drbd_socket_okay(sock2) && ok;
888 /* Gets called if a connection is established, or if a new minor gets created
890 int drbd_connected(struct drbd_peer_device *peer_device)
892 struct drbd_device *device = peer_device->device;
895 atomic_set(&device->packet_seq, 0);
896 device->peer_seq = 0;
898 device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
899 &peer_device->connection->cstate_mutex :
900 &device->own_state_mutex;
902 err = drbd_send_sync_param(peer_device);
904 err = drbd_send_sizes(peer_device, 0, 0);
906 err = drbd_send_uuids(peer_device);
908 err = drbd_send_current_state(peer_device);
909 clear_bit(USE_DEGR_WFC_T, &device->flags);
910 clear_bit(RESIZE_PENDING, &device->flags);
911 atomic_set(&device->ap_in_flight, 0);
912 mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
918 * 1 yes, we have a valid connection
919 * 0 oops, did not work out, please try again
920 * -1 peer talks different language,
921 * no point in trying again, please go standalone.
922 * -2 We do not have a network config...
924 static int conn_connect(struct drbd_connection *connection)
926 struct drbd_socket sock, msock;
927 struct drbd_peer_device *peer_device;
930 bool discard_my_data, ok;
931 enum drbd_state_rv rv;
932 struct accept_wait_data ad = {
933 .connection = connection,
934 .door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
937 clear_bit(DISCONNECT_SENT, &connection->flags);
938 if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
941 mutex_init(&sock.mutex);
942 sock.sbuf = connection->data.sbuf;
943 sock.rbuf = connection->data.rbuf;
945 mutex_init(&msock.mutex);
946 msock.sbuf = connection->meta.sbuf;
947 msock.rbuf = connection->meta.rbuf;
950 /* Assume that the peer only understands protocol 80 until we know better. */
951 connection->agreed_pro_version = 80;
953 if (prepare_listen_socket(connection, &ad))
959 s = drbd_try_connect(connection);
963 send_first_packet(connection, &sock, P_INITIAL_DATA);
964 } else if (!msock.socket) {
965 clear_bit(RESOLVE_CONFLICTS, &connection->flags);
967 send_first_packet(connection, &msock, P_INITIAL_META);
969 drbd_err(connection, "Logic error in conn_connect()\n");
970 goto out_release_sockets;
974 if (connection_established(connection, &sock.socket, &msock.socket))
978 s = drbd_wait_for_connect(connection, &ad);
980 int fp = receive_first_packet(connection, s);
981 drbd_socket_okay(&sock.socket);
982 drbd_socket_okay(&msock.socket);
986 drbd_warn(connection, "initial packet S crossed\n");
987 sock_release(sock.socket);
994 set_bit(RESOLVE_CONFLICTS, &connection->flags);
996 drbd_warn(connection, "initial packet M crossed\n");
997 sock_release(msock.socket);
1004 drbd_warn(connection, "Error receiving initial packet\n");
1007 if (prandom_u32() & 1)
1012 if (connection->cstate <= C_DISCONNECTING)
1013 goto out_release_sockets;
1014 if (signal_pending(current)) {
1015 flush_signals(current);
1017 if (get_t_state(&connection->receiver) == EXITING)
1018 goto out_release_sockets;
1021 ok = connection_established(connection, &sock.socket, &msock.socket);
1025 sock_release(ad.s_listen);
1027 sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1028 msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1030 sock.socket->sk->sk_allocation = GFP_NOIO;
1031 msock.socket->sk->sk_allocation = GFP_NOIO;
1033 sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
1034 msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
1037 * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
1038 * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1039 * first set it to the P_CONNECTION_FEATURES timeout,
1040 * which we set to 4x the configured ping_timeout. */
1042 nc = rcu_dereference(connection->net_conf);
1044 sock.socket->sk->sk_sndtimeo =
1045 sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
1047 msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
1048 timeout = nc->timeout * HZ / 10;
1049 discard_my_data = nc->discard_my_data;
1052 msock.socket->sk->sk_sndtimeo = timeout;
1054 /* we don't want delays.
1055 * we use TCP_CORK where appropriate, though */
1056 tcp_sock_set_nodelay(sock.socket->sk);
1057 tcp_sock_set_nodelay(msock.socket->sk);
1059 connection->data.socket = sock.socket;
1060 connection->meta.socket = msock.socket;
1061 connection->last_received = jiffies;
1063 h = drbd_do_features(connection);
1067 if (connection->cram_hmac_tfm) {
1068 /* drbd_request_state(device, NS(conn, WFAuth)); */
1069 switch (drbd_do_auth(connection)) {
1071 drbd_err(connection, "Authentication of peer failed\n");
1074 drbd_err(connection, "Authentication of peer failed, trying again.\n");
1079 connection->data.socket->sk->sk_sndtimeo = timeout;
1080 connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1082 if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1085 /* Prevent a race between resync-handshake and
1086 * being promoted to Primary.
1088 * Grab and release the state mutex, so we know that any current
1089 * drbd_set_role() is finished, and any incoming drbd_set_role
1090 * will see the STATE_SENT flag, and wait for it to be cleared.
1092 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1093 mutex_lock(peer_device->device->state_mutex);
1095 /* avoid a race with conn_request_state( C_DISCONNECTING ) */
1096 spin_lock_irq(&connection->resource->req_lock);
1097 set_bit(STATE_SENT, &connection->flags);
1098 spin_unlock_irq(&connection->resource->req_lock);
1100 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1101 mutex_unlock(peer_device->device->state_mutex);
1104 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1105 struct drbd_device *device = peer_device->device;
1106 kref_get(&device->kref);
1109 if (discard_my_data)
1110 set_bit(DISCARD_MY_DATA, &device->flags);
1112 clear_bit(DISCARD_MY_DATA, &device->flags);
1114 drbd_connected(peer_device);
1115 kref_put(&device->kref, drbd_destroy_device);
1120 rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1121 if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1122 clear_bit(STATE_SENT, &connection->flags);
1126 drbd_thread_start(&connection->ack_receiver);
1127 /* opencoded create_singlethread_workqueue(),
1128 * to be able to use format string arguments */
1129 connection->ack_sender =
1130 alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
1131 if (!connection->ack_sender) {
1132 drbd_err(connection, "Failed to create workqueue ack_sender\n");
1136 mutex_lock(&connection->resource->conf_update);
1137 /* The discard_my_data flag is a single-shot modifier to the next
1138 * connection attempt, the handshake of which is now well underway.
1139 * No need for rcu style copying of the whole struct
1140 * just to clear a single value. */
1141 connection->net_conf->discard_my_data = 0;
1142 mutex_unlock(&connection->resource->conf_update);
1146 out_release_sockets:
1148 sock_release(ad.s_listen);
1150 sock_release(sock.socket);
1152 sock_release(msock.socket);
1156 static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1158 unsigned int header_size = drbd_header_size(connection);
1160 if (header_size == sizeof(struct p_header100) &&
1161 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1162 struct p_header100 *h = header;
1164 drbd_err(connection, "Header padding is not zero\n");
1167 pi->vnr = be16_to_cpu(h->volume);
1168 pi->cmd = be16_to_cpu(h->command);
1169 pi->size = be32_to_cpu(h->length);
1170 } else if (header_size == sizeof(struct p_header95) &&
1171 *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1172 struct p_header95 *h = header;
1173 pi->cmd = be16_to_cpu(h->command);
1174 pi->size = be32_to_cpu(h->length);
1176 } else if (header_size == sizeof(struct p_header80) &&
1177 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1178 struct p_header80 *h = header;
1179 pi->cmd = be16_to_cpu(h->command);
1180 pi->size = be16_to_cpu(h->length);
1183 drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1184 be32_to_cpu(*(__be32 *)header),
1185 connection->agreed_pro_version);
1188 pi->data = header + header_size;
1192 static void drbd_unplug_all_devices(struct drbd_connection *connection)
1194 if (current->plug == &connection->receiver_plug) {
1195 blk_finish_plug(&connection->receiver_plug);
1196 blk_start_plug(&connection->receiver_plug);
1197 } /* else: maybe just schedule() ?? */
1200 static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1202 void *buffer = connection->data.rbuf;
1205 err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1209 err = decode_header(connection, buffer, pi);
1210 connection->last_received = jiffies;
1215 static int drbd_recv_header_maybe_unplug(struct drbd_connection *connection, struct packet_info *pi)
1217 void *buffer = connection->data.rbuf;
1218 unsigned int size = drbd_header_size(connection);
1221 err = drbd_recv_short(connection->data.socket, buffer, size, MSG_NOSIGNAL|MSG_DONTWAIT);
1223 /* If we have nothing in the receive buffer now, to reduce
1224 * application latency, try to drain the backend queues as
1225 * quickly as possible, and let remote TCP know what we have
1226 * received so far. */
1227 if (err == -EAGAIN) {
1228 tcp_sock_set_quickack(connection->data.socket->sk, 2);
1229 drbd_unplug_all_devices(connection);
1235 err = drbd_recv_all_warn(connection, buffer, size);
1240 err = decode_header(connection, connection->data.rbuf, pi);
1241 connection->last_received = jiffies;
1245 /* This is blkdev_issue_flush, but asynchronous.
1246 * We want to submit to all component volumes in parallel,
1247 * then wait for all completions.
1249 struct issue_flush_context {
1252 struct completion done;
1254 struct one_flush_context {
1255 struct drbd_device *device;
1256 struct issue_flush_context *ctx;
1259 static void one_flush_endio(struct bio *bio)
1261 struct one_flush_context *octx = bio->bi_private;
1262 struct drbd_device *device = octx->device;
1263 struct issue_flush_context *ctx = octx->ctx;
1265 if (bio->bi_status) {
1266 ctx->error = blk_status_to_errno(bio->bi_status);
1267 drbd_info(device, "local disk FLUSH FAILED with status %d\n", bio->bi_status);
1272 clear_bit(FLUSH_PENDING, &device->flags);
1274 kref_put(&device->kref, drbd_destroy_device);
1276 if (atomic_dec_and_test(&ctx->pending))
1277 complete(&ctx->done);
1280 static void submit_one_flush(struct drbd_device *device, struct issue_flush_context *ctx)
1282 struct bio *bio = bio_alloc(device->ldev->backing_bdev, 0,
1283 REQ_OP_FLUSH | REQ_PREFLUSH, GFP_NOIO);
1284 struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);
1287 drbd_warn(device, "Could not allocate a octx, CANNOT ISSUE FLUSH\n");
1288 /* FIXME: what else can I do now? disconnecting or detaching
1289 * really does not help to improve the state of the world, either.
1293 ctx->error = -ENOMEM;
1295 kref_put(&device->kref, drbd_destroy_device);
1299 octx->device = device;
1301 bio->bi_private = octx;
1302 bio->bi_end_io = one_flush_endio;
1304 device->flush_jif = jiffies;
1305 set_bit(FLUSH_PENDING, &device->flags);
1306 atomic_inc(&ctx->pending);
1310 static void drbd_flush(struct drbd_connection *connection)
1312 if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1313 struct drbd_peer_device *peer_device;
1314 struct issue_flush_context ctx;
1317 atomic_set(&ctx.pending, 1);
1319 init_completion(&ctx.done);
1322 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1323 struct drbd_device *device = peer_device->device;
1325 if (!get_ldev(device))
1327 kref_get(&device->kref);
1330 submit_one_flush(device, &ctx);
1336 /* Do we want to add a timeout,
1337 * if disk-timeout is set? */
1338 if (!atomic_dec_and_test(&ctx.pending))
1339 wait_for_completion(&ctx.done);
1342 /* would rather check on EOPNOTSUPP, but that is not reliable.
1343 * don't try again for ANY return value != 0
1344 * if (rv == -EOPNOTSUPP) */
1345 /* Any error is already reported by bio_endio callback. */
1346 drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1352 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1353 * @connection: DRBD connection.
1354 * @epoch: Epoch object.
1357 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1358 struct drbd_epoch *epoch,
1359 enum epoch_event ev)
1362 struct drbd_epoch *next_epoch;
1363 enum finish_epoch rv = FE_STILL_LIVE;
1365 spin_lock(&connection->epoch_lock);
1369 epoch_size = atomic_read(&epoch->epoch_size);
1371 switch (ev & ~EV_CLEANUP) {
1373 atomic_dec(&epoch->active);
1375 case EV_GOT_BARRIER_NR:
1376 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1378 case EV_BECAME_LAST:
1383 if (epoch_size != 0 &&
1384 atomic_read(&epoch->active) == 0 &&
1385 (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1386 if (!(ev & EV_CLEANUP)) {
1387 spin_unlock(&connection->epoch_lock);
1388 drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1389 spin_lock(&connection->epoch_lock);
1392 /* FIXME: dec unacked on connection, once we have
1393 * something to count pending connection packets in. */
1394 if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1395 dec_unacked(epoch->connection);
1398 if (connection->current_epoch != epoch) {
1399 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1400 list_del(&epoch->list);
1401 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1402 connection->epochs--;
1405 if (rv == FE_STILL_LIVE)
1409 atomic_set(&epoch->epoch_size, 0);
1410 /* atomic_set(&epoch->active, 0); is already zero */
1411 if (rv == FE_STILL_LIVE)
1422 spin_unlock(&connection->epoch_lock);
1427 static enum write_ordering_e
1428 max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1430 struct disk_conf *dc;
1432 dc = rcu_dereference(bdev->disk_conf);
1434 if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1436 if (wo == WO_DRAIN_IO && !dc->disk_drain)
1443 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1444 * @wo: Write ordering method to try.
1446 void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1447 enum write_ordering_e wo)
1449 struct drbd_device *device;
1450 enum write_ordering_e pwo;
1452 static char *write_ordering_str[] = {
1454 [WO_DRAIN_IO] = "drain",
1455 [WO_BDEV_FLUSH] = "flush",
1458 pwo = resource->write_ordering;
1459 if (wo != WO_BDEV_FLUSH)
1462 idr_for_each_entry(&resource->devices, device, vnr) {
1463 if (get_ldev(device)) {
1464 wo = max_allowed_wo(device->ldev, wo);
1465 if (device->ldev == bdev)
1472 wo = max_allowed_wo(bdev, wo);
1476 resource->write_ordering = wo;
1477 if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1478 drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1482 * Mapping "discard" to ZEROOUT with UNMAP does not work for us:
1483 * Drivers have to "announce" q->limits.max_write_zeroes_sectors, or it
1484 * will directly go to fallback mode, submitting normal writes, and
1485 * never even try to UNMAP.
1487 * And dm-thin does not do this (yet), mostly because in general it has
1488 * to assume that "skip_block_zeroing" is set. See also:
1489 * https://www.mail-archive.com/dm-devel%40redhat.com/msg07965.html
1490 * https://www.redhat.com/archives/dm-devel/2018-January/msg00271.html
1492 * We *may* ignore the discard-zeroes-data setting, if so configured.
1494 * Assumption is that this "discard_zeroes_data=0" is only because the backend
1495 * may ignore partial unaligned discards.
1497 * LVM/DM thin as of at least
1498 * LVM version: 2.02.115(2)-RHEL7 (2015-01-28)
1499 * Library version: 1.02.93-RHEL7 (2015-01-28)
1500 * Driver version: 4.29.0
1501 * still behaves this way.
1503 * For unaligned (wrt. alignment and granularity) or too small discards,
1504 * we zero-out the initial (and/or) trailing unaligned partial chunks,
1505 * but discard all the aligned full chunks.
1507 * At least for LVM/DM thin, with skip_block_zeroing=false,
1508 * the result is effectively "discard_zeroes_data=1".
1510 /* flags: EE_TRIM|EE_ZEROOUT */
1511 int drbd_issue_discard_or_zero_out(struct drbd_device *device, sector_t start, unsigned int nr_sectors, int flags)
1513 struct block_device *bdev = device->ldev->backing_bdev;
1515 unsigned int max_discard_sectors, granularity;
1519 if ((flags & EE_ZEROOUT) || !(flags & EE_TRIM))
1522 /* Zero-sector (unknown) and one-sector granularities are the same. */
1523 granularity = max(bdev_discard_granularity(bdev) >> 9, 1U);
1524 alignment = (bdev_discard_alignment(bdev) >> 9) % granularity;
1526 max_discard_sectors = min(bdev_max_discard_sectors(bdev), (1U << 22));
1527 max_discard_sectors -= max_discard_sectors % granularity;
1528 if (unlikely(!max_discard_sectors))
1531 if (nr_sectors < granularity)
1535 if (sector_div(tmp, granularity) != alignment) {
1536 if (nr_sectors < 2*granularity)
1538 /* start + gran - (start + gran - align) % gran */
1539 tmp = start + granularity - alignment;
1540 tmp = start + granularity - sector_div(tmp, granularity);
1543 /* don't flag BLKDEV_ZERO_NOUNMAP, we don't know how many
1544 * layers are below us, some may have smaller granularity */
1545 err |= blkdev_issue_zeroout(bdev, start, nr, GFP_NOIO, 0);
1549 while (nr_sectors >= max_discard_sectors) {
1550 err |= blkdev_issue_discard(bdev, start, max_discard_sectors,
1552 nr_sectors -= max_discard_sectors;
1553 start += max_discard_sectors;
1556 /* max_discard_sectors is unsigned int (and a multiple of
1557 * granularity, we made sure of that above already);
1558 * nr is < max_discard_sectors;
1559 * I don't need sector_div here, even though nr is sector_t */
1561 nr -= (unsigned int)nr % granularity;
1563 err |= blkdev_issue_discard(bdev, start, nr, GFP_NOIO);
1570 err |= blkdev_issue_zeroout(bdev, start, nr_sectors, GFP_NOIO,
1571 (flags & EE_TRIM) ? 0 : BLKDEV_ZERO_NOUNMAP);
1576 static bool can_do_reliable_discards(struct drbd_device *device)
1578 struct disk_conf *dc;
1581 if (!bdev_max_discard_sectors(device->ldev->backing_bdev))
1585 dc = rcu_dereference(device->ldev->disk_conf);
1586 can_do = dc->discard_zeroes_if_aligned;
1591 static void drbd_issue_peer_discard_or_zero_out(struct drbd_device *device, struct drbd_peer_request *peer_req)
1593 /* If the backend cannot discard, or does not guarantee
1594 * read-back zeroes in discarded ranges, we fall back to
1595 * zero-out. Unless configuration specifically requested
1597 if (!can_do_reliable_discards(device))
1598 peer_req->flags |= EE_ZEROOUT;
1600 if (drbd_issue_discard_or_zero_out(device, peer_req->i.sector,
1601 peer_req->i.size >> 9, peer_req->flags & (EE_ZEROOUT|EE_TRIM)))
1602 peer_req->flags |= EE_WAS_ERROR;
1603 drbd_endio_write_sec_final(peer_req);
1607 * drbd_submit_peer_request()
1608 * @device: DRBD device.
1609 * @peer_req: peer request
1611 * May spread the pages to multiple bios,
1612 * depending on bio_add_page restrictions.
1614 * Returns 0 if all bios have been submitted,
1615 * -ENOMEM if we could not allocate enough bios,
1616 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1617 * single page to an empty bio (which should never happen and likely indicates
1618 * that the lower level IO stack is in some way broken). This has been observed
1619 * on certain Xen deployments.
1621 /* TODO allocate from our own bio_set. */
1622 int drbd_submit_peer_request(struct drbd_device *device,
1623 struct drbd_peer_request *peer_req,
1624 const blk_opf_t opf, const int fault_type)
1626 struct bio *bios = NULL;
1628 struct page *page = peer_req->pages;
1629 sector_t sector = peer_req->i.sector;
1630 unsigned int data_size = peer_req->i.size;
1631 unsigned int n_bios = 0;
1632 unsigned int nr_pages = PFN_UP(data_size);
1634 /* TRIM/DISCARD: for now, always use the helper function
1635 * blkdev_issue_zeroout(..., discard=true).
1636 * It's synchronous, but it does the right thing wrt. bio splitting.
1637 * Correctness first, performance later. Next step is to code an
1638 * asynchronous variant of the same.
1640 if (peer_req->flags & (EE_TRIM | EE_ZEROOUT)) {
1641 /* wait for all pending IO completions, before we start
1642 * zeroing things out. */
1643 conn_wait_active_ee_empty(peer_req->peer_device->connection);
1644 /* add it to the active list now,
1645 * so we can find it to present it in debugfs */
1646 peer_req->submit_jif = jiffies;
1647 peer_req->flags |= EE_SUBMITTED;
1649 /* If this was a resync request from receive_rs_deallocated(),
1650 * it is already on the sync_ee list */
1651 if (list_empty(&peer_req->w.list)) {
1652 spin_lock_irq(&device->resource->req_lock);
1653 list_add_tail(&peer_req->w.list, &device->active_ee);
1654 spin_unlock_irq(&device->resource->req_lock);
1657 drbd_issue_peer_discard_or_zero_out(device, peer_req);
1661 /* In most cases, we will only need one bio. But in case the lower
1662 * level restrictions happen to be different at this offset on this
1663 * side than those of the sending peer, we may need to submit the
1664 * request in more than one bio.
1666 * Plain bio_alloc is good enough here, this is no DRBD internally
1667 * generated bio, but a bio allocated on behalf of the peer.
1670 bio = bio_alloc(device->ldev->backing_bdev, nr_pages, opf, GFP_NOIO);
1671 /* > peer_req->i.sector, unless this is the first bio */
1672 bio->bi_iter.bi_sector = sector;
1673 bio->bi_private = peer_req;
1674 bio->bi_end_io = drbd_peer_request_endio;
1676 bio->bi_next = bios;
1680 page_chain_for_each(page) {
1681 unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1682 if (!bio_add_page(bio, page, len, 0))
1688 D_ASSERT(device, data_size == 0);
1689 D_ASSERT(device, page == NULL);
1691 atomic_set(&peer_req->pending_bios, n_bios);
1692 /* for debugfs: update timestamp, mark as submitted */
1693 peer_req->submit_jif = jiffies;
1694 peer_req->flags |= EE_SUBMITTED;
1697 bios = bios->bi_next;
1698 bio->bi_next = NULL;
1700 drbd_submit_bio_noacct(device, fault_type, bio);
1705 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1706 struct drbd_peer_request *peer_req)
1708 struct drbd_interval *i = &peer_req->i;
1710 drbd_remove_interval(&device->write_requests, i);
1711 drbd_clear_interval(i);
1713 /* Wake up any processes waiting for this peer request to complete. */
1715 wake_up(&device->misc_wait);
1718 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1720 struct drbd_peer_device *peer_device;
1724 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1725 struct drbd_device *device = peer_device->device;
1727 kref_get(&device->kref);
1729 drbd_wait_ee_list_empty(device, &device->active_ee);
1730 kref_put(&device->kref, drbd_destroy_device);
1736 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1739 struct p_barrier *p = pi->data;
1740 struct drbd_epoch *epoch;
1742 /* FIXME these are unacked on connection,
1743 * not a specific (peer)device.
1745 connection->current_epoch->barrier_nr = p->barrier;
1746 connection->current_epoch->connection = connection;
1747 rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1749 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1750 * the activity log, which means it would not be resynced in case the
1751 * R_PRIMARY crashes now.
1752 * Therefore we must send the barrier_ack after the barrier request was
1754 switch (connection->resource->write_ordering) {
1756 if (rv == FE_RECYCLED)
1759 /* receiver context, in the writeout path of the other node.
1760 * avoid potential distributed deadlock */
1761 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1765 drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1770 conn_wait_active_ee_empty(connection);
1771 drbd_flush(connection);
1773 if (atomic_read(&connection->current_epoch->epoch_size)) {
1774 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1781 drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1782 connection->resource->write_ordering);
1787 atomic_set(&epoch->epoch_size, 0);
1788 atomic_set(&epoch->active, 0);
1790 spin_lock(&connection->epoch_lock);
1791 if (atomic_read(&connection->current_epoch->epoch_size)) {
1792 list_add(&epoch->list, &connection->current_epoch->list);
1793 connection->current_epoch = epoch;
1794 connection->epochs++;
1796 /* The current_epoch got recycled while we allocated this one... */
1799 spin_unlock(&connection->epoch_lock);
1804 /* quick wrapper in case payload size != request_size (write same) */
1805 static void drbd_csum_ee_size(struct crypto_shash *h,
1806 struct drbd_peer_request *r, void *d,
1807 unsigned int payload_size)
1809 unsigned int tmp = r->i.size;
1810 r->i.size = payload_size;
1811 drbd_csum_ee(h, r, d);
1815 /* used from receive_RSDataReply (recv_resync_read)
1816 * and from receive_Data.
1817 * data_size: actual payload ("data in")
1818 * for normal writes that is bi_size.
1819 * for discards, that is zero.
1820 * for write same, it is logical_block_size.
1821 * both trim and write same have the bi_size ("data len to be affected")
1822 * as extra argument in the packet header.
1824 static struct drbd_peer_request *
1825 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1826 struct packet_info *pi) __must_hold(local)
1828 struct drbd_device *device = peer_device->device;
1829 const sector_t capacity = get_capacity(device->vdisk);
1830 struct drbd_peer_request *peer_req;
1832 int digest_size, err;
1833 unsigned int data_size = pi->size, ds;
1834 void *dig_in = peer_device->connection->int_dig_in;
1835 void *dig_vv = peer_device->connection->int_dig_vv;
1836 unsigned long *data;
1837 struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1838 struct p_trim *zeroes = (pi->cmd == P_ZEROES) ? pi->data : NULL;
1841 if (!trim && peer_device->connection->peer_integrity_tfm) {
1842 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1844 * FIXME: Receive the incoming digest into the receive buffer
1845 * here, together with its struct p_data?
1847 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1850 data_size -= digest_size;
1853 /* assume request_size == data_size, but special case trim. */
1856 if (!expect(data_size == 0))
1858 ds = be32_to_cpu(trim->size);
1859 } else if (zeroes) {
1860 if (!expect(data_size == 0))
1862 ds = be32_to_cpu(zeroes->size);
1865 if (!expect(IS_ALIGNED(ds, 512)))
1867 if (trim || zeroes) {
1868 if (!expect(ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
1870 } else if (!expect(ds <= DRBD_MAX_BIO_SIZE))
1873 /* even though we trust out peer,
1874 * we sometimes have to double check. */
1875 if (sector + (ds>>9) > capacity) {
1876 drbd_err(device, "request from peer beyond end of local disk: "
1877 "capacity: %llus < sector: %llus + size: %u\n",
1878 (unsigned long long)capacity,
1879 (unsigned long long)sector, ds);
1883 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1884 * "criss-cross" setup, that might cause write-out on some other DRBD,
1885 * which in turn might block on the other node at this very place. */
1886 peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
1890 peer_req->flags |= EE_WRITE;
1892 peer_req->flags |= EE_TRIM;
1896 peer_req->flags |= EE_ZEROOUT;
1900 /* receive payload size bytes into page chain */
1902 page = peer_req->pages;
1903 page_chain_for_each(page) {
1904 unsigned len = min_t(int, ds, PAGE_SIZE);
1906 err = drbd_recv_all_warn(peer_device->connection, data, len);
1907 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1908 drbd_err(device, "Fault injection: Corrupting data on receive\n");
1909 data[0] = data[0] ^ (unsigned long)-1;
1913 drbd_free_peer_req(device, peer_req);
1920 drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
1921 if (memcmp(dig_in, dig_vv, digest_size)) {
1922 drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1923 (unsigned long long)sector, data_size);
1924 drbd_free_peer_req(device, peer_req);
1928 device->recv_cnt += data_size >> 9;
1932 /* drbd_drain_block() just takes a data block
1933 * out of the socket input buffer, and discards it.
1935 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1944 page = drbd_alloc_pages(peer_device, 1, 1);
1948 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1950 err = drbd_recv_all_warn(peer_device->connection, data, len);
1956 drbd_free_pages(peer_device->device, page, 0);
1960 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1961 sector_t sector, int data_size)
1963 struct bio_vec bvec;
1964 struct bvec_iter iter;
1966 int digest_size, err, expect;
1967 void *dig_in = peer_device->connection->int_dig_in;
1968 void *dig_vv = peer_device->connection->int_dig_vv;
1971 if (peer_device->connection->peer_integrity_tfm) {
1972 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1973 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1976 data_size -= digest_size;
1979 /* optimistically update recv_cnt. if receiving fails below,
1980 * we disconnect anyways, and counters will be reset. */
1981 peer_device->device->recv_cnt += data_size>>9;
1983 bio = req->master_bio;
1984 D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
1986 bio_for_each_segment(bvec, bio, iter) {
1987 void *mapped = bvec_kmap_local(&bvec);
1988 expect = min_t(int, data_size, bvec.bv_len);
1989 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
1990 kunmap_local(mapped);
1993 data_size -= expect;
1997 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
1998 if (memcmp(dig_in, dig_vv, digest_size)) {
1999 drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
2004 D_ASSERT(peer_device->device, data_size == 0);
2009 * e_end_resync_block() is called in ack_sender context via
2010 * drbd_finish_peer_reqs().
2012 static int e_end_resync_block(struct drbd_work *w, int unused)
2014 struct drbd_peer_request *peer_req =
2015 container_of(w, struct drbd_peer_request, w);
2016 struct drbd_peer_device *peer_device = peer_req->peer_device;
2017 struct drbd_device *device = peer_device->device;
2018 sector_t sector = peer_req->i.sector;
2021 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2023 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2024 drbd_set_in_sync(device, sector, peer_req->i.size);
2025 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
2027 /* Record failure to sync */
2028 drbd_rs_failed_io(device, sector, peer_req->i.size);
2030 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2032 dec_unacked(device);
2037 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
2038 struct packet_info *pi) __releases(local)
2040 struct drbd_device *device = peer_device->device;
2041 struct drbd_peer_request *peer_req;
2043 peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
2047 dec_rs_pending(device);
2049 inc_unacked(device);
2050 /* corresponding dec_unacked() in e_end_resync_block()
2051 * respective _drbd_clear_done_ee */
2053 peer_req->w.cb = e_end_resync_block;
2054 peer_req->submit_jif = jiffies;
2056 spin_lock_irq(&device->resource->req_lock);
2057 list_add_tail(&peer_req->w.list, &device->sync_ee);
2058 spin_unlock_irq(&device->resource->req_lock);
2060 atomic_add(pi->size >> 9, &device->rs_sect_ev);
2061 if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE,
2062 DRBD_FAULT_RS_WR) == 0)
2065 /* don't care for the reason here */
2066 drbd_err(device, "submit failed, triggering re-connect\n");
2067 spin_lock_irq(&device->resource->req_lock);
2068 list_del(&peer_req->w.list);
2069 spin_unlock_irq(&device->resource->req_lock);
2071 drbd_free_peer_req(device, peer_req);
2077 static struct drbd_request *
2078 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
2079 sector_t sector, bool missing_ok, const char *func)
2081 struct drbd_request *req;
2083 /* Request object according to our peer */
2084 req = (struct drbd_request *)(unsigned long)id;
2085 if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
2088 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
2089 (unsigned long)id, (unsigned long long)sector);
2094 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
2096 struct drbd_peer_device *peer_device;
2097 struct drbd_device *device;
2098 struct drbd_request *req;
2101 struct p_data *p = pi->data;
2103 peer_device = conn_peer_device(connection, pi->vnr);
2106 device = peer_device->device;
2108 sector = be64_to_cpu(p->sector);
2110 spin_lock_irq(&device->resource->req_lock);
2111 req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
2112 spin_unlock_irq(&device->resource->req_lock);
2116 err = recv_dless_read(peer_device, req, sector, pi->size);
2118 req_mod(req, DATA_RECEIVED);
2119 /* else: nothing. handled from drbd_disconnect...
2120 * I don't think we may complete this just yet
2121 * in case we are "on-disconnect: freeze" */
2126 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
2128 struct drbd_peer_device *peer_device;
2129 struct drbd_device *device;
2132 struct p_data *p = pi->data;
2134 peer_device = conn_peer_device(connection, pi->vnr);
2137 device = peer_device->device;
2139 sector = be64_to_cpu(p->sector);
2140 D_ASSERT(device, p->block_id == ID_SYNCER);
2142 if (get_ldev(device)) {
2143 /* data is submitted to disk within recv_resync_read.
2144 * corresponding put_ldev done below on error,
2145 * or in drbd_peer_request_endio. */
2146 err = recv_resync_read(peer_device, sector, pi);
2148 if (__ratelimit(&drbd_ratelimit_state))
2149 drbd_err(device, "Can not write resync data to local disk.\n");
2151 err = drbd_drain_block(peer_device, pi->size);
2153 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2156 atomic_add(pi->size >> 9, &device->rs_sect_in);
2161 static void restart_conflicting_writes(struct drbd_device *device,
2162 sector_t sector, int size)
2164 struct drbd_interval *i;
2165 struct drbd_request *req;
2167 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2170 req = container_of(i, struct drbd_request, i);
2171 if (req->rq_state & RQ_LOCAL_PENDING ||
2172 !(req->rq_state & RQ_POSTPONED))
2174 /* as it is RQ_POSTPONED, this will cause it to
2175 * be queued on the retry workqueue. */
2176 __req_mod(req, CONFLICT_RESOLVED, NULL);
2181 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2183 static int e_end_block(struct drbd_work *w, int cancel)
2185 struct drbd_peer_request *peer_req =
2186 container_of(w, struct drbd_peer_request, w);
2187 struct drbd_peer_device *peer_device = peer_req->peer_device;
2188 struct drbd_device *device = peer_device->device;
2189 sector_t sector = peer_req->i.sector;
2192 if (peer_req->flags & EE_SEND_WRITE_ACK) {
2193 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2194 pcmd = (device->state.conn >= C_SYNC_SOURCE &&
2195 device->state.conn <= C_PAUSED_SYNC_T &&
2196 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
2197 P_RS_WRITE_ACK : P_WRITE_ACK;
2198 err = drbd_send_ack(peer_device, pcmd, peer_req);
2199 if (pcmd == P_RS_WRITE_ACK)
2200 drbd_set_in_sync(device, sector, peer_req->i.size);
2202 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2203 /* we expect it to be marked out of sync anyways...
2204 * maybe assert this? */
2206 dec_unacked(device);
2209 /* we delete from the conflict detection hash _after_ we sent out the
2210 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
2211 if (peer_req->flags & EE_IN_INTERVAL_TREE) {
2212 spin_lock_irq(&device->resource->req_lock);
2213 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
2214 drbd_remove_epoch_entry_interval(device, peer_req);
2215 if (peer_req->flags & EE_RESTART_REQUESTS)
2216 restart_conflicting_writes(device, sector, peer_req->i.size);
2217 spin_unlock_irq(&device->resource->req_lock);
2219 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2221 drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2226 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2228 struct drbd_peer_request *peer_req =
2229 container_of(w, struct drbd_peer_request, w);
2230 struct drbd_peer_device *peer_device = peer_req->peer_device;
2233 err = drbd_send_ack(peer_device, ack, peer_req);
2234 dec_unacked(peer_device->device);
2239 static int e_send_superseded(struct drbd_work *w, int unused)
2241 return e_send_ack(w, P_SUPERSEDED);
2244 static int e_send_retry_write(struct drbd_work *w, int unused)
2246 struct drbd_peer_request *peer_req =
2247 container_of(w, struct drbd_peer_request, w);
2248 struct drbd_connection *connection = peer_req->peer_device->connection;
2250 return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2251 P_RETRY_WRITE : P_SUPERSEDED);
2254 static bool seq_greater(u32 a, u32 b)
2257 * We assume 32-bit wrap-around here.
2258 * For 24-bit wrap-around, we would have to shift:
2261 return (s32)a - (s32)b > 0;
2264 static u32 seq_max(u32 a, u32 b)
2266 return seq_greater(a, b) ? a : b;
2269 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2271 struct drbd_device *device = peer_device->device;
2272 unsigned int newest_peer_seq;
2274 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2275 spin_lock(&device->peer_seq_lock);
2276 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2277 device->peer_seq = newest_peer_seq;
2278 spin_unlock(&device->peer_seq_lock);
2279 /* wake up only if we actually changed device->peer_seq */
2280 if (peer_seq == newest_peer_seq)
2281 wake_up(&device->seq_wait);
2285 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2287 return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2290 /* maybe change sync_ee into interval trees as well? */
2291 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2293 struct drbd_peer_request *rs_req;
2296 spin_lock_irq(&device->resource->req_lock);
2297 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2298 if (overlaps(peer_req->i.sector, peer_req->i.size,
2299 rs_req->i.sector, rs_req->i.size)) {
2304 spin_unlock_irq(&device->resource->req_lock);
2309 /* Called from receive_Data.
2310 * Synchronize packets on sock with packets on msock.
2312 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2313 * packet traveling on msock, they are still processed in the order they have
2316 * Note: we don't care for Ack packets overtaking P_DATA packets.
2318 * In case packet_seq is larger than device->peer_seq number, there are
2319 * outstanding packets on the msock. We wait for them to arrive.
2320 * In case we are the logically next packet, we update device->peer_seq
2321 * ourselves. Correctly handles 32bit wrap around.
2323 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2324 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2325 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2326 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2328 * returns 0 if we may process the packet,
2329 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2330 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2332 struct drbd_device *device = peer_device->device;
2337 if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2340 spin_lock(&device->peer_seq_lock);
2342 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2343 device->peer_seq = seq_max(device->peer_seq, peer_seq);
2347 if (signal_pending(current)) {
2353 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2359 /* Only need to wait if two_primaries is enabled */
2360 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2361 spin_unlock(&device->peer_seq_lock);
2363 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2365 timeout = schedule_timeout(timeout);
2366 spin_lock(&device->peer_seq_lock);
2369 drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2373 spin_unlock(&device->peer_seq_lock);
2374 finish_wait(&device->seq_wait, &wait);
2378 /* see also bio_flags_to_wire()
2379 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2380 * flags and back. We may replicate to other kernel versions. */
2381 static blk_opf_t wire_flags_to_bio_flags(u32 dpf)
2383 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2384 (dpf & DP_FUA ? REQ_FUA : 0) |
2385 (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2388 static enum req_op wire_flags_to_bio_op(u32 dpf)
2390 if (dpf & DP_ZEROES)
2391 return REQ_OP_WRITE_ZEROES;
2392 if (dpf & DP_DISCARD)
2393 return REQ_OP_DISCARD;
2395 return REQ_OP_WRITE;
2398 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2401 struct drbd_interval *i;
2404 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2405 struct drbd_request *req;
2406 struct bio_and_error m;
2410 req = container_of(i, struct drbd_request, i);
2411 if (!(req->rq_state & RQ_POSTPONED))
2413 req->rq_state &= ~RQ_POSTPONED;
2414 __req_mod(req, NEG_ACKED, &m);
2415 spin_unlock_irq(&device->resource->req_lock);
2417 complete_master_bio(device, &m);
2418 spin_lock_irq(&device->resource->req_lock);
2423 static int handle_write_conflicts(struct drbd_device *device,
2424 struct drbd_peer_request *peer_req)
2426 struct drbd_connection *connection = peer_req->peer_device->connection;
2427 bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2428 sector_t sector = peer_req->i.sector;
2429 const unsigned int size = peer_req->i.size;
2430 struct drbd_interval *i;
2435 * Inserting the peer request into the write_requests tree will prevent
2436 * new conflicting local requests from being added.
2438 drbd_insert_interval(&device->write_requests, &peer_req->i);
2441 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2442 if (i == &peer_req->i)
2449 * Our peer has sent a conflicting remote request; this
2450 * should not happen in a two-node setup. Wait for the
2451 * earlier peer request to complete.
2453 err = drbd_wait_misc(device, i);
2459 equal = i->sector == sector && i->size == size;
2460 if (resolve_conflicts) {
2462 * If the peer request is fully contained within the
2463 * overlapping request, it can be considered overwritten
2464 * and thus superseded; otherwise, it will be retried
2465 * once all overlapping requests have completed.
2467 bool superseded = i->sector <= sector && i->sector +
2468 (i->size >> 9) >= sector + (size >> 9);
2471 drbd_alert(device, "Concurrent writes detected: "
2472 "local=%llus +%u, remote=%llus +%u, "
2473 "assuming %s came first\n",
2474 (unsigned long long)i->sector, i->size,
2475 (unsigned long long)sector, size,
2476 superseded ? "local" : "remote");
2478 peer_req->w.cb = superseded ? e_send_superseded :
2480 list_add_tail(&peer_req->w.list, &device->done_ee);
2481 queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2486 struct drbd_request *req =
2487 container_of(i, struct drbd_request, i);
2490 drbd_alert(device, "Concurrent writes detected: "
2491 "local=%llus +%u, remote=%llus +%u\n",
2492 (unsigned long long)i->sector, i->size,
2493 (unsigned long long)sector, size);
2495 if (req->rq_state & RQ_LOCAL_PENDING ||
2496 !(req->rq_state & RQ_POSTPONED)) {
2498 * Wait for the node with the discard flag to
2499 * decide if this request has been superseded
2500 * or needs to be retried.
2501 * Requests that have been superseded will
2502 * disappear from the write_requests tree.
2504 * In addition, wait for the conflicting
2505 * request to finish locally before submitting
2506 * the conflicting peer request.
2508 err = drbd_wait_misc(device, &req->i);
2510 _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2511 fail_postponed_requests(device, sector, size);
2517 * Remember to restart the conflicting requests after
2518 * the new peer request has completed.
2520 peer_req->flags |= EE_RESTART_REQUESTS;
2527 drbd_remove_epoch_entry_interval(device, peer_req);
2531 /* mirrored write */
2532 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2534 struct drbd_peer_device *peer_device;
2535 struct drbd_device *device;
2536 struct net_conf *nc;
2538 struct drbd_peer_request *peer_req;
2539 struct p_data *p = pi->data;
2540 u32 peer_seq = be32_to_cpu(p->seq_num);
2546 peer_device = conn_peer_device(connection, pi->vnr);
2549 device = peer_device->device;
2551 if (!get_ldev(device)) {
2554 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2555 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2556 atomic_inc(&connection->current_epoch->epoch_size);
2557 err2 = drbd_drain_block(peer_device, pi->size);
2564 * Corresponding put_ldev done either below (on various errors), or in
2565 * drbd_peer_request_endio, if we successfully submit the data at the
2566 * end of this function.
2569 sector = be64_to_cpu(p->sector);
2570 peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2576 peer_req->w.cb = e_end_block;
2577 peer_req->submit_jif = jiffies;
2578 peer_req->flags |= EE_APPLICATION;
2580 dp_flags = be32_to_cpu(p->dp_flags);
2581 op = wire_flags_to_bio_op(dp_flags);
2582 op_flags = wire_flags_to_bio_flags(dp_flags);
2583 if (pi->cmd == P_TRIM) {
2584 D_ASSERT(peer_device, peer_req->i.size > 0);
2585 D_ASSERT(peer_device, op == REQ_OP_DISCARD);
2586 D_ASSERT(peer_device, peer_req->pages == NULL);
2587 /* need to play safe: an older DRBD sender
2588 * may mean zero-out while sending P_TRIM. */
2589 if (0 == (connection->agreed_features & DRBD_FF_WZEROES))
2590 peer_req->flags |= EE_ZEROOUT;
2591 } else if (pi->cmd == P_ZEROES) {
2592 D_ASSERT(peer_device, peer_req->i.size > 0);
2593 D_ASSERT(peer_device, op == REQ_OP_WRITE_ZEROES);
2594 D_ASSERT(peer_device, peer_req->pages == NULL);
2595 /* Do (not) pass down BLKDEV_ZERO_NOUNMAP? */
2596 if (dp_flags & DP_DISCARD)
2597 peer_req->flags |= EE_TRIM;
2598 } else if (peer_req->pages == NULL) {
2599 D_ASSERT(device, peer_req->i.size == 0);
2600 D_ASSERT(device, dp_flags & DP_FLUSH);
2603 if (dp_flags & DP_MAY_SET_IN_SYNC)
2604 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2606 spin_lock(&connection->epoch_lock);
2607 peer_req->epoch = connection->current_epoch;
2608 atomic_inc(&peer_req->epoch->epoch_size);
2609 atomic_inc(&peer_req->epoch->active);
2610 spin_unlock(&connection->epoch_lock);
2613 nc = rcu_dereference(peer_device->connection->net_conf);
2614 tp = nc->two_primaries;
2615 if (peer_device->connection->agreed_pro_version < 100) {
2616 switch (nc->wire_protocol) {
2618 dp_flags |= DP_SEND_WRITE_ACK;
2621 dp_flags |= DP_SEND_RECEIVE_ACK;
2627 if (dp_flags & DP_SEND_WRITE_ACK) {
2628 peer_req->flags |= EE_SEND_WRITE_ACK;
2629 inc_unacked(device);
2630 /* corresponding dec_unacked() in e_end_block()
2631 * respective _drbd_clear_done_ee */
2634 if (dp_flags & DP_SEND_RECEIVE_ACK) {
2635 /* I really don't like it that the receiver thread
2636 * sends on the msock, but anyways */
2637 drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2641 /* two primaries implies protocol C */
2642 D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2643 peer_req->flags |= EE_IN_INTERVAL_TREE;
2644 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2646 goto out_interrupted;
2647 spin_lock_irq(&device->resource->req_lock);
2648 err = handle_write_conflicts(device, peer_req);
2650 spin_unlock_irq(&device->resource->req_lock);
2651 if (err == -ENOENT) {
2655 goto out_interrupted;
2658 update_peer_seq(peer_device, peer_seq);
2659 spin_lock_irq(&device->resource->req_lock);
2661 /* TRIM and is processed synchronously,
2662 * we wait for all pending requests, respectively wait for
2663 * active_ee to become empty in drbd_submit_peer_request();
2664 * better not add ourselves here. */
2665 if ((peer_req->flags & (EE_TRIM | EE_ZEROOUT)) == 0)
2666 list_add_tail(&peer_req->w.list, &device->active_ee);
2667 spin_unlock_irq(&device->resource->req_lock);
2669 if (device->state.conn == C_SYNC_TARGET)
2670 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2672 if (device->state.pdsk < D_INCONSISTENT) {
2673 /* In case we have the only disk of the cluster, */
2674 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2675 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2676 drbd_al_begin_io(device, &peer_req->i);
2677 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2680 err = drbd_submit_peer_request(device, peer_req, op | op_flags,
2685 /* don't care for the reason here */
2686 drbd_err(device, "submit failed, triggering re-connect\n");
2687 spin_lock_irq(&device->resource->req_lock);
2688 list_del(&peer_req->w.list);
2689 drbd_remove_epoch_entry_interval(device, peer_req);
2690 spin_unlock_irq(&device->resource->req_lock);
2691 if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2692 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2693 drbd_al_complete_io(device, &peer_req->i);
2697 drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
2699 drbd_free_peer_req(device, peer_req);
2703 /* We may throttle resync, if the lower device seems to be busy,
2704 * and current sync rate is above c_min_rate.
2706 * To decide whether or not the lower device is busy, we use a scheme similar
2707 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2708 * (more than 64 sectors) of activity we cannot account for with our own resync
2709 * activity, it obviously is "busy".
2711 * The current sync rate used here uses only the most recent two step marks,
2712 * to have a short time average so we can react faster.
2714 bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2715 bool throttle_if_app_is_waiting)
2717 struct lc_element *tmp;
2718 bool throttle = drbd_rs_c_min_rate_throttle(device);
2720 if (!throttle || throttle_if_app_is_waiting)
2723 spin_lock_irq(&device->al_lock);
2724 tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2726 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2727 if (test_bit(BME_PRIORITY, &bm_ext->flags))
2729 /* Do not slow down if app IO is already waiting for this extent,
2730 * and our progress is necessary for application IO to complete. */
2732 spin_unlock_irq(&device->al_lock);
2737 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2739 struct gendisk *disk = device->ldev->backing_bdev->bd_disk;
2740 unsigned long db, dt, dbdt;
2741 unsigned int c_min_rate;
2745 c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2748 /* feature disabled? */
2749 if (c_min_rate == 0)
2752 curr_events = (int)part_stat_read_accum(disk->part0, sectors) -
2753 atomic_read(&device->rs_sect_ev);
2755 if (atomic_read(&device->ap_actlog_cnt)
2756 || curr_events - device->rs_last_events > 64) {
2757 unsigned long rs_left;
2760 device->rs_last_events = curr_events;
2762 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2764 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2766 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2767 rs_left = device->ov_left;
2769 rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2771 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2774 db = device->rs_mark_left[i] - rs_left;
2775 dbdt = Bit2KB(db/dt);
2777 if (dbdt > c_min_rate)
2783 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2785 struct drbd_peer_device *peer_device;
2786 struct drbd_device *device;
2789 struct drbd_peer_request *peer_req;
2790 struct digest_info *di = NULL;
2792 unsigned int fault_type;
2793 struct p_block_req *p = pi->data;
2795 peer_device = conn_peer_device(connection, pi->vnr);
2798 device = peer_device->device;
2799 capacity = get_capacity(device->vdisk);
2801 sector = be64_to_cpu(p->sector);
2802 size = be32_to_cpu(p->blksize);
2804 if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2805 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2806 (unsigned long long)sector, size);
2809 if (sector + (size>>9) > capacity) {
2810 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2811 (unsigned long long)sector, size);
2815 if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2818 case P_DATA_REQUEST:
2819 drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2822 case P_RS_DATA_REQUEST:
2823 case P_CSUM_RS_REQUEST:
2825 drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2829 dec_rs_pending(device);
2830 drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2835 if (verb && __ratelimit(&drbd_ratelimit_state))
2836 drbd_err(device, "Can not satisfy peer's read request, "
2837 "no local data.\n");
2839 /* drain possibly payload */
2840 return drbd_drain_block(peer_device, pi->size);
2843 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2844 * "criss-cross" setup, that might cause write-out on some other DRBD,
2845 * which in turn might block on the other node at this very place. */
2846 peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2854 case P_DATA_REQUEST:
2855 peer_req->w.cb = w_e_end_data_req;
2856 fault_type = DRBD_FAULT_DT_RD;
2857 /* application IO, don't drbd_rs_begin_io */
2858 peer_req->flags |= EE_APPLICATION;
2862 /* If at some point in the future we have a smart way to
2863 find out if this data block is completely deallocated,
2864 then we would do something smarter here than reading
2866 peer_req->flags |= EE_RS_THIN_REQ;
2868 case P_RS_DATA_REQUEST:
2869 peer_req->w.cb = w_e_end_rsdata_req;
2870 fault_type = DRBD_FAULT_RS_RD;
2871 /* used in the sector offset progress display */
2872 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2876 case P_CSUM_RS_REQUEST:
2877 fault_type = DRBD_FAULT_RS_RD;
2878 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2882 di->digest_size = pi->size;
2883 di->digest = (((char *)di)+sizeof(struct digest_info));
2885 peer_req->digest = di;
2886 peer_req->flags |= EE_HAS_DIGEST;
2888 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2891 if (pi->cmd == P_CSUM_RS_REQUEST) {
2892 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2893 peer_req->w.cb = w_e_end_csum_rs_req;
2894 /* used in the sector offset progress display */
2895 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2896 /* remember to report stats in drbd_resync_finished */
2897 device->use_csums = true;
2898 } else if (pi->cmd == P_OV_REPLY) {
2899 /* track progress, we may need to throttle */
2900 atomic_add(size >> 9, &device->rs_sect_in);
2901 peer_req->w.cb = w_e_end_ov_reply;
2902 dec_rs_pending(device);
2903 /* drbd_rs_begin_io done when we sent this request,
2904 * but accounting still needs to be done. */
2905 goto submit_for_resync;
2910 if (device->ov_start_sector == ~(sector_t)0 &&
2911 peer_device->connection->agreed_pro_version >= 90) {
2912 unsigned long now = jiffies;
2914 device->ov_start_sector = sector;
2915 device->ov_position = sector;
2916 device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2917 device->rs_total = device->ov_left;
2918 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2919 device->rs_mark_left[i] = device->ov_left;
2920 device->rs_mark_time[i] = now;
2922 drbd_info(device, "Online Verify start sector: %llu\n",
2923 (unsigned long long)sector);
2925 peer_req->w.cb = w_e_end_ov_req;
2926 fault_type = DRBD_FAULT_RS_RD;
2933 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2934 * wrt the receiver, but it is not as straightforward as it may seem.
2935 * Various places in the resync start and stop logic assume resync
2936 * requests are processed in order, requeuing this on the worker thread
2937 * introduces a bunch of new code for synchronization between threads.
2939 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2940 * "forever", throttling after drbd_rs_begin_io will lock that extent
2941 * for application writes for the same time. For now, just throttle
2942 * here, where the rest of the code expects the receiver to sleep for
2946 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2947 * this defers syncer requests for some time, before letting at least
2948 * on request through. The resync controller on the receiving side
2949 * will adapt to the incoming rate accordingly.
2951 * We cannot throttle here if remote is Primary/SyncTarget:
2952 * we would also throttle its application reads.
2953 * In that case, throttling is done on the SyncTarget only.
2956 /* Even though this may be a resync request, we do add to "read_ee";
2957 * "sync_ee" is only used for resync WRITEs.
2958 * Add to list early, so debugfs can find this request
2959 * even if we have to sleep below. */
2960 spin_lock_irq(&device->resource->req_lock);
2961 list_add_tail(&peer_req->w.list, &device->read_ee);
2962 spin_unlock_irq(&device->resource->req_lock);
2964 update_receiver_timing_details(connection, drbd_rs_should_slow_down);
2965 if (device->state.peer != R_PRIMARY
2966 && drbd_rs_should_slow_down(device, sector, false))
2967 schedule_timeout_uninterruptible(HZ/10);
2968 update_receiver_timing_details(connection, drbd_rs_begin_io);
2969 if (drbd_rs_begin_io(device, sector))
2973 atomic_add(size >> 9, &device->rs_sect_ev);
2976 update_receiver_timing_details(connection, drbd_submit_peer_request);
2977 inc_unacked(device);
2978 if (drbd_submit_peer_request(device, peer_req, REQ_OP_READ,
2982 /* don't care for the reason here */
2983 drbd_err(device, "submit failed, triggering re-connect\n");
2986 spin_lock_irq(&device->resource->req_lock);
2987 list_del(&peer_req->w.list);
2988 spin_unlock_irq(&device->resource->req_lock);
2989 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2992 drbd_free_peer_req(device, peer_req);
2997 * drbd_asb_recover_0p - Recover after split-brain with no remaining primaries
2999 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
3001 struct drbd_device *device = peer_device->device;
3002 int self, peer, rv = -100;
3003 unsigned long ch_self, ch_peer;
3004 enum drbd_after_sb_p after_sb_0p;
3006 self = device->ldev->md.uuid[UI_BITMAP] & 1;
3007 peer = device->p_uuid[UI_BITMAP] & 1;
3009 ch_peer = device->p_uuid[UI_SIZE];
3010 ch_self = device->comm_bm_set;
3013 after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
3015 switch (after_sb_0p) {
3017 case ASB_DISCARD_SECONDARY:
3018 case ASB_CALL_HELPER:
3020 drbd_err(device, "Configuration error.\n");
3022 case ASB_DISCONNECT:
3024 case ASB_DISCARD_YOUNGER_PRI:
3025 if (self == 0 && peer == 1) {
3029 if (self == 1 && peer == 0) {
3033 fallthrough; /* to one of the other strategies */
3034 case ASB_DISCARD_OLDER_PRI:
3035 if (self == 0 && peer == 1) {
3039 if (self == 1 && peer == 0) {
3043 /* Else fall through to one of the other strategies... */
3044 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
3045 "Using discard-least-changes instead\n");
3047 case ASB_DISCARD_ZERO_CHG:
3048 if (ch_peer == 0 && ch_self == 0) {
3049 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3053 if (ch_peer == 0) { rv = 1; break; }
3054 if (ch_self == 0) { rv = -1; break; }
3056 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
3059 case ASB_DISCARD_LEAST_CHG:
3060 if (ch_self < ch_peer)
3062 else if (ch_self > ch_peer)
3064 else /* ( ch_self == ch_peer ) */
3065 /* Well, then use something else. */
3066 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3069 case ASB_DISCARD_LOCAL:
3072 case ASB_DISCARD_REMOTE:
3080 * drbd_asb_recover_1p - Recover after split-brain with one remaining primary
3082 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
3084 struct drbd_device *device = peer_device->device;
3086 enum drbd_after_sb_p after_sb_1p;
3089 after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
3091 switch (after_sb_1p) {
3092 case ASB_DISCARD_YOUNGER_PRI:
3093 case ASB_DISCARD_OLDER_PRI:
3094 case ASB_DISCARD_LEAST_CHG:
3095 case ASB_DISCARD_LOCAL:
3096 case ASB_DISCARD_REMOTE:
3097 case ASB_DISCARD_ZERO_CHG:
3098 drbd_err(device, "Configuration error.\n");
3100 case ASB_DISCONNECT:
3103 hg = drbd_asb_recover_0p(peer_device);
3104 if (hg == -1 && device->state.role == R_SECONDARY)
3106 if (hg == 1 && device->state.role == R_PRIMARY)
3110 rv = drbd_asb_recover_0p(peer_device);
3112 case ASB_DISCARD_SECONDARY:
3113 return device->state.role == R_PRIMARY ? 1 : -1;
3114 case ASB_CALL_HELPER:
3115 hg = drbd_asb_recover_0p(peer_device);
3116 if (hg == -1 && device->state.role == R_PRIMARY) {
3117 enum drbd_state_rv rv2;
3119 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3120 * we might be here in C_WF_REPORT_PARAMS which is transient.
3121 * we do not need to wait for the after state change work either. */
3122 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3123 if (rv2 != SS_SUCCESS) {
3124 drbd_khelper(device, "pri-lost-after-sb");
3126 drbd_warn(device, "Successfully gave up primary role.\n");
3137 * drbd_asb_recover_2p - Recover after split-brain with two remaining primaries
3139 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
3141 struct drbd_device *device = peer_device->device;
3143 enum drbd_after_sb_p after_sb_2p;
3146 after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
3148 switch (after_sb_2p) {
3149 case ASB_DISCARD_YOUNGER_PRI:
3150 case ASB_DISCARD_OLDER_PRI:
3151 case ASB_DISCARD_LEAST_CHG:
3152 case ASB_DISCARD_LOCAL:
3153 case ASB_DISCARD_REMOTE:
3155 case ASB_DISCARD_SECONDARY:
3156 case ASB_DISCARD_ZERO_CHG:
3157 drbd_err(device, "Configuration error.\n");
3160 rv = drbd_asb_recover_0p(peer_device);
3162 case ASB_DISCONNECT:
3164 case ASB_CALL_HELPER:
3165 hg = drbd_asb_recover_0p(peer_device);
3167 enum drbd_state_rv rv2;
3169 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3170 * we might be here in C_WF_REPORT_PARAMS which is transient.
3171 * we do not need to wait for the after state change work either. */
3172 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3173 if (rv2 != SS_SUCCESS) {
3174 drbd_khelper(device, "pri-lost-after-sb");
3176 drbd_warn(device, "Successfully gave up primary role.\n");
3186 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
3187 u64 bits, u64 flags)
3190 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
3193 drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
3195 (unsigned long long)uuid[UI_CURRENT],
3196 (unsigned long long)uuid[UI_BITMAP],
3197 (unsigned long long)uuid[UI_HISTORY_START],
3198 (unsigned long long)uuid[UI_HISTORY_END],
3199 (unsigned long long)bits,
3200 (unsigned long long)flags);
3204 100 after split brain try auto recover
3205 2 C_SYNC_SOURCE set BitMap
3206 1 C_SYNC_SOURCE use BitMap
3208 -1 C_SYNC_TARGET use BitMap
3209 -2 C_SYNC_TARGET set BitMap
3210 -100 after split brain, disconnect
3211 -1000 unrelated data
3212 -1091 requires proto 91
3213 -1096 requires proto 96
3216 static int drbd_uuid_compare(struct drbd_device *const device, enum drbd_role const peer_role, int *rule_nr) __must_hold(local)
3218 struct drbd_peer_device *const peer_device = first_peer_device(device);
3219 struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
3223 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3224 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3227 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
3231 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
3232 peer != UUID_JUST_CREATED)
3236 if (self != UUID_JUST_CREATED &&
3237 (peer == UUID_JUST_CREATED || peer == (u64)0))
3241 int rct, dc; /* roles at crash time */
3243 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3245 if (connection->agreed_pro_version < 91)
3248 if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3249 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3250 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3251 drbd_uuid_move_history(device);
3252 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3253 device->ldev->md.uuid[UI_BITMAP] = 0;
3255 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3256 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3259 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3266 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3268 if (connection->agreed_pro_version < 91)
3271 if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3272 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3273 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3275 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3276 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3277 device->p_uuid[UI_BITMAP] = 0UL;
3279 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3282 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3289 /* Common power [off|failure] */
3290 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3291 (device->p_uuid[UI_FLAGS] & 2);
3292 /* lowest bit is set when we were primary,
3293 * next bit (weight 2) is set when peer was primary */
3296 /* Neither has the "crashed primary" flag set,
3297 * only a replication link hickup. */
3301 /* Current UUID equal and no bitmap uuid; does not necessarily
3302 * mean this was a "simultaneous hard crash", maybe IO was
3303 * frozen, so no UUID-bump happened.
3304 * This is a protocol change, overload DRBD_FF_WSAME as flag
3305 * for "new-enough" peer DRBD version. */
3306 if (device->state.role == R_PRIMARY || peer_role == R_PRIMARY) {
3308 if (!(connection->agreed_features & DRBD_FF_WSAME)) {
3309 drbd_warn(peer_device, "Equivalent unrotated UUIDs, but current primary present.\n");
3310 return -(0x10000 | PRO_VERSION_MAX | (DRBD_FF_WSAME << 8));
3312 if (device->state.role == R_PRIMARY && peer_role == R_PRIMARY) {
3313 /* At least one has the "crashed primary" bit set,
3314 * both are primary now, but neither has rotated its UUIDs?
3315 * "Can not happen." */
3316 drbd_err(peer_device, "Equivalent unrotated UUIDs, but both are primary. Can not resolve this.\n");
3319 if (device->state.role == R_PRIMARY)
3324 /* Both are secondary.
3325 * Really looks like recovery from simultaneous hard crash.
3326 * Check which had been primary before, and arbitrate. */
3328 case 0: /* !self_pri && !peer_pri */ return 0; /* already handled */
3329 case 1: /* self_pri && !peer_pri */ return 1;
3330 case 2: /* !self_pri && peer_pri */ return -1;
3331 case 3: /* self_pri && peer_pri */
3332 dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3338 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3343 peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3345 if (connection->agreed_pro_version < 96 ?
3346 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3347 (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3348 peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3349 /* The last P_SYNC_UUID did not get though. Undo the last start of
3350 resync as sync source modifications of the peer's UUIDs. */
3352 if (connection->agreed_pro_version < 91)
3355 device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3356 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3358 drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3359 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3366 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3367 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3368 peer = device->p_uuid[i] & ~((u64)1);
3374 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3375 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3380 self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3382 if (connection->agreed_pro_version < 96 ?
3383 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3384 (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3385 self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3386 /* The last P_SYNC_UUID did not get though. Undo the last start of
3387 resync as sync source modifications of our UUIDs. */
3389 if (connection->agreed_pro_version < 91)
3392 __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3393 __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3395 drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3396 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3397 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3405 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3406 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3407 self = device->ldev->md.uuid[i] & ~((u64)1);
3413 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3414 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3415 if (self == peer && self != ((u64)0))
3419 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3420 self = device->ldev->md.uuid[i] & ~((u64)1);
3421 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3422 peer = device->p_uuid[j] & ~((u64)1);
3431 /* drbd_sync_handshake() returns the new conn state on success, or
3432 CONN_MASK (-1) on failure.
3434 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3435 enum drbd_role peer_role,
3436 enum drbd_disk_state peer_disk) __must_hold(local)
3438 struct drbd_device *device = peer_device->device;
3439 enum drbd_conns rv = C_MASK;
3440 enum drbd_disk_state mydisk;
3441 struct net_conf *nc;
3442 int hg, rule_nr, rr_conflict, tentative, always_asbp;
3444 mydisk = device->state.disk;
3445 if (mydisk == D_NEGOTIATING)
3446 mydisk = device->new_state_tmp.disk;
3448 drbd_info(device, "drbd_sync_handshake:\n");
3450 spin_lock_irq(&device->ldev->md.uuid_lock);
3451 drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3452 drbd_uuid_dump(device, "peer", device->p_uuid,
3453 device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3455 hg = drbd_uuid_compare(device, peer_role, &rule_nr);
3456 spin_unlock_irq(&device->ldev->md.uuid_lock);
3458 drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3461 drbd_alert(device, "Unrelated data, aborting!\n");
3464 if (hg < -0x10000) {
3468 fflags = (hg >> 8) & 0xff;
3469 drbd_alert(device, "To resolve this both sides have to support at least protocol %d and feature flags 0x%x\n",
3474 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3478 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3479 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
3480 int f = (hg == -100) || abs(hg) == 2;
3481 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3484 drbd_info(device, "Becoming sync %s due to disk states.\n",
3485 hg > 0 ? "source" : "target");
3489 drbd_khelper(device, "initial-split-brain");
3492 nc = rcu_dereference(peer_device->connection->net_conf);
3493 always_asbp = nc->always_asbp;
3494 rr_conflict = nc->rr_conflict;
3495 tentative = nc->tentative;
3498 if (hg == 100 || (hg == -100 && always_asbp)) {
3499 int pcount = (device->state.role == R_PRIMARY)
3500 + (peer_role == R_PRIMARY);
3501 int forced = (hg == -100);
3505 hg = drbd_asb_recover_0p(peer_device);
3508 hg = drbd_asb_recover_1p(peer_device);
3511 hg = drbd_asb_recover_2p(peer_device);
3514 if (abs(hg) < 100) {
3515 drbd_warn(device, "Split-Brain detected, %d primaries, "
3516 "automatically solved. Sync from %s node\n",
3517 pcount, (hg < 0) ? "peer" : "this");
3519 drbd_warn(device, "Doing a full sync, since"
3520 " UUIDs where ambiguous.\n");
3527 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3529 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3533 drbd_warn(device, "Split-Brain detected, manually solved. "
3534 "Sync from %s node\n",
3535 (hg < 0) ? "peer" : "this");
3539 /* FIXME this log message is not correct if we end up here
3540 * after an attempted attach on a diskless node.
3541 * We just refuse to attach -- well, we drop the "connection"
3542 * to that disk, in a way... */
3543 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3544 drbd_khelper(device, "split-brain");
3548 if (hg > 0 && mydisk <= D_INCONSISTENT) {
3549 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3553 if (hg < 0 && /* by intention we do not use mydisk here. */
3554 device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3555 switch (rr_conflict) {
3556 case ASB_CALL_HELPER:
3557 drbd_khelper(device, "pri-lost");
3559 case ASB_DISCONNECT:
3560 drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3563 drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3568 if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3570 drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3572 drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3573 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3574 abs(hg) >= 2 ? "full" : "bit-map based");
3579 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3580 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3581 BM_LOCKED_SET_ALLOWED))
3585 if (hg > 0) { /* become sync source. */
3587 } else if (hg < 0) { /* become sync target */
3591 if (drbd_bm_total_weight(device)) {
3592 drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3593 drbd_bm_total_weight(device));
3600 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3602 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3603 if (peer == ASB_DISCARD_REMOTE)
3604 return ASB_DISCARD_LOCAL;
3606 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3607 if (peer == ASB_DISCARD_LOCAL)
3608 return ASB_DISCARD_REMOTE;
3610 /* everything else is valid if they are equal on both sides. */
3614 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3616 struct p_protocol *p = pi->data;
3617 enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3618 int p_proto, p_discard_my_data, p_two_primaries, cf;
3619 struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3620 char integrity_alg[SHARED_SECRET_MAX] = "";
3621 struct crypto_shash *peer_integrity_tfm = NULL;
3622 void *int_dig_in = NULL, *int_dig_vv = NULL;
3624 p_proto = be32_to_cpu(p->protocol);
3625 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
3626 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
3627 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
3628 p_two_primaries = be32_to_cpu(p->two_primaries);
3629 cf = be32_to_cpu(p->conn_flags);
3630 p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3632 if (connection->agreed_pro_version >= 87) {
3635 if (pi->size > sizeof(integrity_alg))
3637 err = drbd_recv_all(connection, integrity_alg, pi->size);
3640 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3643 if (pi->cmd != P_PROTOCOL_UPDATE) {
3644 clear_bit(CONN_DRY_RUN, &connection->flags);
3646 if (cf & CF_DRY_RUN)
3647 set_bit(CONN_DRY_RUN, &connection->flags);
3650 nc = rcu_dereference(connection->net_conf);
3652 if (p_proto != nc->wire_protocol) {
3653 drbd_err(connection, "incompatible %s settings\n", "protocol");
3654 goto disconnect_rcu_unlock;
3657 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3658 drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3659 goto disconnect_rcu_unlock;
3662 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3663 drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3664 goto disconnect_rcu_unlock;
3667 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3668 drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3669 goto disconnect_rcu_unlock;
3672 if (p_discard_my_data && nc->discard_my_data) {
3673 drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3674 goto disconnect_rcu_unlock;
3677 if (p_two_primaries != nc->two_primaries) {
3678 drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3679 goto disconnect_rcu_unlock;
3682 if (strcmp(integrity_alg, nc->integrity_alg)) {
3683 drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3684 goto disconnect_rcu_unlock;
3690 if (integrity_alg[0]) {
3694 * We can only change the peer data integrity algorithm
3695 * here. Changing our own data integrity algorithm
3696 * requires that we send a P_PROTOCOL_UPDATE packet at
3697 * the same time; otherwise, the peer has no way to
3698 * tell between which packets the algorithm should
3702 peer_integrity_tfm = crypto_alloc_shash(integrity_alg, 0, 0);
3703 if (IS_ERR(peer_integrity_tfm)) {
3704 peer_integrity_tfm = NULL;
3705 drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3710 hash_size = crypto_shash_digestsize(peer_integrity_tfm);
3711 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3712 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3713 if (!(int_dig_in && int_dig_vv)) {
3714 drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3719 new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3723 mutex_lock(&connection->data.mutex);
3724 mutex_lock(&connection->resource->conf_update);
3725 old_net_conf = connection->net_conf;
3726 *new_net_conf = *old_net_conf;
3728 new_net_conf->wire_protocol = p_proto;
3729 new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3730 new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3731 new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3732 new_net_conf->two_primaries = p_two_primaries;
3734 rcu_assign_pointer(connection->net_conf, new_net_conf);
3735 mutex_unlock(&connection->resource->conf_update);
3736 mutex_unlock(&connection->data.mutex);
3738 crypto_free_shash(connection->peer_integrity_tfm);
3739 kfree(connection->int_dig_in);
3740 kfree(connection->int_dig_vv);
3741 connection->peer_integrity_tfm = peer_integrity_tfm;
3742 connection->int_dig_in = int_dig_in;
3743 connection->int_dig_vv = int_dig_vv;
3745 if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3746 drbd_info(connection, "peer data-integrity-alg: %s\n",
3747 integrity_alg[0] ? integrity_alg : "(none)");
3749 kvfree_rcu(old_net_conf);
3752 disconnect_rcu_unlock:
3755 crypto_free_shash(peer_integrity_tfm);
3758 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3763 * input: alg name, feature name
3764 * return: NULL (alg name was "")
3765 * ERR_PTR(error) if something goes wrong
3766 * or the crypto hash ptr, if it worked out ok. */
3767 static struct crypto_shash *drbd_crypto_alloc_digest_safe(
3768 const struct drbd_device *device,
3769 const char *alg, const char *name)
3771 struct crypto_shash *tfm;
3776 tfm = crypto_alloc_shash(alg, 0, 0);
3778 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3779 alg, name, PTR_ERR(tfm));
3785 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3787 void *buffer = connection->data.rbuf;
3788 int size = pi->size;
3791 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3792 s = drbd_recv(connection, buffer, s);
3806 * config_unknown_volume - device configuration command for unknown volume
3808 * When a device is added to an existing connection, the node on which the
3809 * device is added first will send configuration commands to its peer but the
3810 * peer will not know about the device yet. It will warn and ignore these
3811 * commands. Once the device is added on the second node, the second node will
3812 * send the same device configuration commands, but in the other direction.
3814 * (We can also end up here if drbd is misconfigured.)
3816 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3818 drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3819 cmdname(pi->cmd), pi->vnr);
3820 return ignore_remaining_packet(connection, pi);
3823 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3825 struct drbd_peer_device *peer_device;
3826 struct drbd_device *device;
3827 struct p_rs_param_95 *p;
3828 unsigned int header_size, data_size, exp_max_sz;
3829 struct crypto_shash *verify_tfm = NULL;
3830 struct crypto_shash *csums_tfm = NULL;
3831 struct net_conf *old_net_conf, *new_net_conf = NULL;
3832 struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3833 const int apv = connection->agreed_pro_version;
3834 struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3835 unsigned int fifo_size = 0;
3838 peer_device = conn_peer_device(connection, pi->vnr);
3840 return config_unknown_volume(connection, pi);
3841 device = peer_device->device;
3843 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
3844 : apv == 88 ? sizeof(struct p_rs_param)
3846 : apv <= 94 ? sizeof(struct p_rs_param_89)
3847 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3849 if (pi->size > exp_max_sz) {
3850 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3851 pi->size, exp_max_sz);
3856 header_size = sizeof(struct p_rs_param);
3857 data_size = pi->size - header_size;
3858 } else if (apv <= 94) {
3859 header_size = sizeof(struct p_rs_param_89);
3860 data_size = pi->size - header_size;
3861 D_ASSERT(device, data_size == 0);
3863 header_size = sizeof(struct p_rs_param_95);
3864 data_size = pi->size - header_size;
3865 D_ASSERT(device, data_size == 0);
3868 /* initialize verify_alg and csums_alg */
3870 BUILD_BUG_ON(sizeof(p->algs) != 2 * SHARED_SECRET_MAX);
3871 memset(&p->algs, 0, sizeof(p->algs));
3873 err = drbd_recv_all(peer_device->connection, p, header_size);
3877 mutex_lock(&connection->resource->conf_update);
3878 old_net_conf = peer_device->connection->net_conf;
3879 if (get_ldev(device)) {
3880 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3881 if (!new_disk_conf) {
3883 mutex_unlock(&connection->resource->conf_update);
3884 drbd_err(device, "Allocation of new disk_conf failed\n");
3888 old_disk_conf = device->ldev->disk_conf;
3889 *new_disk_conf = *old_disk_conf;
3891 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3896 if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3897 drbd_err(device, "verify-alg of wrong size, "
3898 "peer wants %u, accepting only up to %u byte\n",
3899 data_size, SHARED_SECRET_MAX);
3903 err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3906 /* we expect NUL terminated string */
3907 /* but just in case someone tries to be evil */
3908 D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3909 p->verify_alg[data_size-1] = 0;
3911 } else /* apv >= 89 */ {
3912 /* we still expect NUL terminated strings */
3913 /* but just in case someone tries to be evil */
3914 D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3915 D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3916 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3917 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3920 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3921 if (device->state.conn == C_WF_REPORT_PARAMS) {
3922 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3923 old_net_conf->verify_alg, p->verify_alg);
3926 verify_tfm = drbd_crypto_alloc_digest_safe(device,
3927 p->verify_alg, "verify-alg");
3928 if (IS_ERR(verify_tfm)) {
3934 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3935 if (device->state.conn == C_WF_REPORT_PARAMS) {
3936 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3937 old_net_conf->csums_alg, p->csums_alg);
3940 csums_tfm = drbd_crypto_alloc_digest_safe(device,
3941 p->csums_alg, "csums-alg");
3942 if (IS_ERR(csums_tfm)) {
3948 if (apv > 94 && new_disk_conf) {
3949 new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3950 new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3951 new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3952 new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3954 fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3955 if (fifo_size != device->rs_plan_s->size) {
3956 new_plan = fifo_alloc(fifo_size);
3958 drbd_err(device, "kmalloc of fifo_buffer failed");
3965 if (verify_tfm || csums_tfm) {
3966 new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3970 *new_net_conf = *old_net_conf;
3973 strcpy(new_net_conf->verify_alg, p->verify_alg);
3974 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3975 crypto_free_shash(peer_device->connection->verify_tfm);
3976 peer_device->connection->verify_tfm = verify_tfm;
3977 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3980 strcpy(new_net_conf->csums_alg, p->csums_alg);
3981 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3982 crypto_free_shash(peer_device->connection->csums_tfm);
3983 peer_device->connection->csums_tfm = csums_tfm;
3984 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
3986 rcu_assign_pointer(connection->net_conf, new_net_conf);
3990 if (new_disk_conf) {
3991 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3996 old_plan = device->rs_plan_s;
3997 rcu_assign_pointer(device->rs_plan_s, new_plan);
4000 mutex_unlock(&connection->resource->conf_update);
4003 kfree(old_net_conf);
4004 kfree(old_disk_conf);
4010 if (new_disk_conf) {
4012 kfree(new_disk_conf);
4014 mutex_unlock(&connection->resource->conf_update);
4019 if (new_disk_conf) {
4021 kfree(new_disk_conf);
4023 mutex_unlock(&connection->resource->conf_update);
4024 /* just for completeness: actually not needed,
4025 * as this is not reached if csums_tfm was ok. */
4026 crypto_free_shash(csums_tfm);
4027 /* but free the verify_tfm again, if csums_tfm did not work out */
4028 crypto_free_shash(verify_tfm);
4029 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4033 /* warn if the arguments differ by more than 12.5% */
4034 static void warn_if_differ_considerably(struct drbd_device *device,
4035 const char *s, sector_t a, sector_t b)
4038 if (a == 0 || b == 0)
4040 d = (a > b) ? (a - b) : (b - a);
4041 if (d > (a>>3) || d > (b>>3))
4042 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
4043 (unsigned long long)a, (unsigned long long)b);
4046 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
4048 struct drbd_peer_device *peer_device;
4049 struct drbd_device *device;
4050 struct p_sizes *p = pi->data;
4051 struct o_qlim *o = (connection->agreed_features & DRBD_FF_WSAME) ? p->qlim : NULL;
4052 enum determine_dev_size dd = DS_UNCHANGED;
4053 sector_t p_size, p_usize, p_csize, my_usize;
4054 sector_t new_size, cur_size;
4055 int ldsc = 0; /* local disk size changed */
4056 enum dds_flags ddsf;
4058 peer_device = conn_peer_device(connection, pi->vnr);
4060 return config_unknown_volume(connection, pi);
4061 device = peer_device->device;
4062 cur_size = get_capacity(device->vdisk);
4064 p_size = be64_to_cpu(p->d_size);
4065 p_usize = be64_to_cpu(p->u_size);
4066 p_csize = be64_to_cpu(p->c_size);
4068 /* just store the peer's disk size for now.
4069 * we still need to figure out whether we accept that. */
4070 device->p_size = p_size;
4072 if (get_ldev(device)) {
4074 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
4077 warn_if_differ_considerably(device, "lower level device sizes",
4078 p_size, drbd_get_max_capacity(device->ldev));
4079 warn_if_differ_considerably(device, "user requested size",
4082 /* if this is the first connect, or an otherwise expected
4083 * param exchange, choose the minimum */
4084 if (device->state.conn == C_WF_REPORT_PARAMS)
4085 p_usize = min_not_zero(my_usize, p_usize);
4087 /* Never shrink a device with usable data during connect,
4088 * or "attach" on the peer.
4089 * But allow online shrinking if we are connected. */
4090 new_size = drbd_new_dev_size(device, device->ldev, p_usize, 0);
4091 if (new_size < cur_size &&
4092 device->state.disk >= D_OUTDATED &&
4093 (device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS)) {
4094 drbd_err(device, "The peer's disk size is too small! (%llu < %llu sectors)\n",
4095 (unsigned long long)new_size, (unsigned long long)cur_size);
4096 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4101 if (my_usize != p_usize) {
4102 struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
4104 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
4105 if (!new_disk_conf) {
4110 mutex_lock(&connection->resource->conf_update);
4111 old_disk_conf = device->ldev->disk_conf;
4112 *new_disk_conf = *old_disk_conf;
4113 new_disk_conf->disk_size = p_usize;
4115 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4116 mutex_unlock(&connection->resource->conf_update);
4117 kvfree_rcu(old_disk_conf);
4119 drbd_info(device, "Peer sets u_size to %lu sectors (old: %lu)\n",
4120 (unsigned long)p_usize, (unsigned long)my_usize);
4126 device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
4127 /* Leave drbd_reconsider_queue_parameters() before drbd_determine_dev_size().
4128 In case we cleared the QUEUE_FLAG_DISCARD from our queue in
4129 drbd_reconsider_queue_parameters(), we can be sure that after
4130 drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
4132 ddsf = be16_to_cpu(p->dds_flags);
4133 if (get_ldev(device)) {
4134 drbd_reconsider_queue_parameters(device, device->ldev, o);
4135 dd = drbd_determine_dev_size(device, ddsf, NULL);
4139 drbd_md_sync(device);
4142 * I am diskless, need to accept the peer's *current* size.
4143 * I must NOT accept the peers backing disk size,
4144 * it may have been larger than mine all along...
4146 * At this point, the peer knows more about my disk, or at
4147 * least about what we last agreed upon, than myself.
4148 * So if his c_size is less than his d_size, the most likely
4149 * reason is that *my* d_size was smaller last time we checked.
4151 * However, if he sends a zero current size,
4152 * take his (user-capped or) backing disk size anyways.
4154 * Unless of course he does not have a disk himself.
4155 * In which case we ignore this completely.
4157 sector_t new_size = p_csize ?: p_usize ?: p_size;
4158 drbd_reconsider_queue_parameters(device, NULL, o);
4159 if (new_size == 0) {
4160 /* Ignore, peer does not know nothing. */
4161 } else if (new_size == cur_size) {
4163 } else if (cur_size != 0 && p_size == 0) {
4164 drbd_warn(device, "Ignored diskless peer device size (peer:%llu != me:%llu sectors)!\n",
4165 (unsigned long long)new_size, (unsigned long long)cur_size);
4166 } else if (new_size < cur_size && device->state.role == R_PRIMARY) {
4167 drbd_err(device, "The peer's device size is too small! (%llu < %llu sectors); demote me first!\n",
4168 (unsigned long long)new_size, (unsigned long long)cur_size);
4169 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4172 /* I believe the peer, if
4173 * - I don't have a current size myself
4174 * - we agree on the size anyways
4175 * - I do have a current size, am Secondary,
4176 * and he has the only disk
4177 * - I do have a current size, am Primary,
4178 * and he has the only disk,
4179 * which is larger than my current size
4181 drbd_set_my_capacity(device, new_size);
4185 if (get_ldev(device)) {
4186 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
4187 device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
4194 if (device->state.conn > C_WF_REPORT_PARAMS) {
4195 if (be64_to_cpu(p->c_size) != get_capacity(device->vdisk) ||
4197 /* we have different sizes, probably peer
4198 * needs to know my new size... */
4199 drbd_send_sizes(peer_device, 0, ddsf);
4201 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
4202 (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
4203 if (device->state.pdsk >= D_INCONSISTENT &&
4204 device->state.disk >= D_INCONSISTENT) {
4205 if (ddsf & DDSF_NO_RESYNC)
4206 drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
4208 resync_after_online_grow(device);
4210 set_bit(RESYNC_AFTER_NEG, &device->flags);
4217 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
4219 struct drbd_peer_device *peer_device;
4220 struct drbd_device *device;
4221 struct p_uuids *p = pi->data;
4223 int i, updated_uuids = 0;
4225 peer_device = conn_peer_device(connection, pi->vnr);
4227 return config_unknown_volume(connection, pi);
4228 device = peer_device->device;
4230 p_uuid = kmalloc_array(UI_EXTENDED_SIZE, sizeof(*p_uuid), GFP_NOIO);
4234 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
4235 p_uuid[i] = be64_to_cpu(p->uuid[i]);
4237 kfree(device->p_uuid);
4238 device->p_uuid = p_uuid;
4240 if ((device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS) &&
4241 device->state.disk < D_INCONSISTENT &&
4242 device->state.role == R_PRIMARY &&
4243 (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
4244 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
4245 (unsigned long long)device->ed_uuid);
4246 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4250 if (get_ldev(device)) {
4251 int skip_initial_sync =
4252 device->state.conn == C_CONNECTED &&
4253 peer_device->connection->agreed_pro_version >= 90 &&
4254 device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
4255 (p_uuid[UI_FLAGS] & 8);
4256 if (skip_initial_sync) {
4257 drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
4258 drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
4259 "clear_n_write from receive_uuids",
4260 BM_LOCKED_TEST_ALLOWED);
4261 _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
4262 _drbd_uuid_set(device, UI_BITMAP, 0);
4263 _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
4265 drbd_md_sync(device);
4269 } else if (device->state.disk < D_INCONSISTENT &&
4270 device->state.role == R_PRIMARY) {
4271 /* I am a diskless primary, the peer just created a new current UUID
4273 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4276 /* Before we test for the disk state, we should wait until an eventually
4277 ongoing cluster wide state change is finished. That is important if
4278 we are primary and are detaching from our disk. We need to see the
4279 new disk state... */
4280 mutex_lock(device->state_mutex);
4281 mutex_unlock(device->state_mutex);
4282 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
4283 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4286 drbd_print_uuids(device, "receiver updated UUIDs to");
4292 * convert_state() - Converts the peer's view of the cluster state to our point of view
4293 * @ps: The state as seen by the peer.
4295 static union drbd_state convert_state(union drbd_state ps)
4297 union drbd_state ms;
4299 static enum drbd_conns c_tab[] = {
4300 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
4301 [C_CONNECTED] = C_CONNECTED,
4303 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
4304 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
4305 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
4306 [C_VERIFY_S] = C_VERIFY_T,
4312 ms.conn = c_tab[ps.conn];
4317 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
4322 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4324 struct drbd_peer_device *peer_device;
4325 struct drbd_device *device;
4326 struct p_req_state *p = pi->data;
4327 union drbd_state mask, val;
4328 enum drbd_state_rv rv;
4330 peer_device = conn_peer_device(connection, pi->vnr);
4333 device = peer_device->device;
4335 mask.i = be32_to_cpu(p->mask);
4336 val.i = be32_to_cpu(p->val);
4338 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4339 mutex_is_locked(device->state_mutex)) {
4340 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4344 mask = convert_state(mask);
4345 val = convert_state(val);
4347 rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4348 drbd_send_sr_reply(peer_device, rv);
4350 drbd_md_sync(device);
4355 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4357 struct p_req_state *p = pi->data;
4358 union drbd_state mask, val;
4359 enum drbd_state_rv rv;
4361 mask.i = be32_to_cpu(p->mask);
4362 val.i = be32_to_cpu(p->val);
4364 if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4365 mutex_is_locked(&connection->cstate_mutex)) {
4366 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4370 mask = convert_state(mask);
4371 val = convert_state(val);
4373 rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4374 conn_send_sr_reply(connection, rv);
4379 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4381 struct drbd_peer_device *peer_device;
4382 struct drbd_device *device;
4383 struct p_state *p = pi->data;
4384 union drbd_state os, ns, peer_state;
4385 enum drbd_disk_state real_peer_disk;
4386 enum chg_state_flags cs_flags;
4389 peer_device = conn_peer_device(connection, pi->vnr);
4391 return config_unknown_volume(connection, pi);
4392 device = peer_device->device;
4394 peer_state.i = be32_to_cpu(p->state);
4396 real_peer_disk = peer_state.disk;
4397 if (peer_state.disk == D_NEGOTIATING) {
4398 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4399 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4402 spin_lock_irq(&device->resource->req_lock);
4404 os = ns = drbd_read_state(device);
4405 spin_unlock_irq(&device->resource->req_lock);
4407 /* If some other part of the code (ack_receiver thread, timeout)
4408 * already decided to close the connection again,
4409 * we must not "re-establish" it here. */
4410 if (os.conn <= C_TEAR_DOWN)
4413 /* If this is the "end of sync" confirmation, usually the peer disk
4414 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4415 * set) resync started in PausedSyncT, or if the timing of pause-/
4416 * unpause-sync events has been "just right", the peer disk may
4417 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4419 if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4420 real_peer_disk == D_UP_TO_DATE &&
4421 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4422 /* If we are (becoming) SyncSource, but peer is still in sync
4423 * preparation, ignore its uptodate-ness to avoid flapping, it
4424 * will change to inconsistent once the peer reaches active
4426 * It may have changed syncer-paused flags, however, so we
4427 * cannot ignore this completely. */
4428 if (peer_state.conn > C_CONNECTED &&
4429 peer_state.conn < C_SYNC_SOURCE)
4430 real_peer_disk = D_INCONSISTENT;
4432 /* if peer_state changes to connected at the same time,
4433 * it explicitly notifies us that it finished resync.
4434 * Maybe we should finish it up, too? */
4435 else if (os.conn >= C_SYNC_SOURCE &&
4436 peer_state.conn == C_CONNECTED) {
4437 if (drbd_bm_total_weight(device) <= device->rs_failed)
4438 drbd_resync_finished(device);
4443 /* explicit verify finished notification, stop sector reached. */
4444 if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4445 peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4446 ov_out_of_sync_print(device);
4447 drbd_resync_finished(device);
4451 /* peer says his disk is inconsistent, while we think it is uptodate,
4452 * and this happens while the peer still thinks we have a sync going on,
4453 * but we think we are already done with the sync.
4454 * We ignore this to avoid flapping pdsk.
4455 * This should not happen, if the peer is a recent version of drbd. */
4456 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4457 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4458 real_peer_disk = D_UP_TO_DATE;
4460 if (ns.conn == C_WF_REPORT_PARAMS)
4461 ns.conn = C_CONNECTED;
4463 if (peer_state.conn == C_AHEAD)
4467 * if (primary and diskless and peer uuid != effective uuid)
4468 * abort attach on peer;
4470 * If this node does not have good data, was already connected, but
4471 * the peer did a late attach only now, trying to "negotiate" with me,
4472 * AND I am currently Primary, possibly frozen, with some specific
4473 * "effective" uuid, this should never be reached, really, because
4474 * we first send the uuids, then the current state.
4476 * In this scenario, we already dropped the connection hard
4477 * when we received the unsuitable uuids (receive_uuids().
4479 * Should we want to change this, that is: not drop the connection in
4480 * receive_uuids() already, then we would need to add a branch here
4481 * that aborts the attach of "unsuitable uuids" on the peer in case
4482 * this node is currently Diskless Primary.
4485 if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4486 get_ldev_if_state(device, D_NEGOTIATING)) {
4487 int cr; /* consider resync */
4489 /* if we established a new connection */
4490 cr = (os.conn < C_CONNECTED);
4491 /* if we had an established connection
4492 * and one of the nodes newly attaches a disk */
4493 cr |= (os.conn == C_CONNECTED &&
4494 (peer_state.disk == D_NEGOTIATING ||
4495 os.disk == D_NEGOTIATING));
4496 /* if we have both been inconsistent, and the peer has been
4497 * forced to be UpToDate with --force */
4498 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4499 /* if we had been plain connected, and the admin requested to
4500 * start a sync by "invalidate" or "invalidate-remote" */
4501 cr |= (os.conn == C_CONNECTED &&
4502 (peer_state.conn >= C_STARTING_SYNC_S &&
4503 peer_state.conn <= C_WF_BITMAP_T));
4506 ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4509 if (ns.conn == C_MASK) {
4510 ns.conn = C_CONNECTED;
4511 if (device->state.disk == D_NEGOTIATING) {
4512 drbd_force_state(device, NS(disk, D_FAILED));
4513 } else if (peer_state.disk == D_NEGOTIATING) {
4514 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4515 peer_state.disk = D_DISKLESS;
4516 real_peer_disk = D_DISKLESS;
4518 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4520 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4521 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4527 spin_lock_irq(&device->resource->req_lock);
4528 if (os.i != drbd_read_state(device).i)
4530 clear_bit(CONSIDER_RESYNC, &device->flags);
4531 ns.peer = peer_state.role;
4532 ns.pdsk = real_peer_disk;
4533 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4534 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4535 ns.disk = device->new_state_tmp.disk;
4536 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4537 if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4538 test_bit(NEW_CUR_UUID, &device->flags)) {
4539 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4540 for temporal network outages! */
4541 spin_unlock_irq(&device->resource->req_lock);
4542 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4543 tl_clear(peer_device->connection);
4544 drbd_uuid_new_current(device);
4545 clear_bit(NEW_CUR_UUID, &device->flags);
4546 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4549 rv = _drbd_set_state(device, ns, cs_flags, NULL);
4550 ns = drbd_read_state(device);
4551 spin_unlock_irq(&device->resource->req_lock);
4553 if (rv < SS_SUCCESS) {
4554 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4558 if (os.conn > C_WF_REPORT_PARAMS) {
4559 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4560 peer_state.disk != D_NEGOTIATING ) {
4561 /* we want resync, peer has not yet decided to sync... */
4562 /* Nowadays only used when forcing a node into primary role and
4563 setting its disk to UpToDate with that */
4564 drbd_send_uuids(peer_device);
4565 drbd_send_current_state(peer_device);
4569 clear_bit(DISCARD_MY_DATA, &device->flags);
4571 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4576 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4578 struct drbd_peer_device *peer_device;
4579 struct drbd_device *device;
4580 struct p_rs_uuid *p = pi->data;
4582 peer_device = conn_peer_device(connection, pi->vnr);
4585 device = peer_device->device;
4587 wait_event(device->misc_wait,
4588 device->state.conn == C_WF_SYNC_UUID ||
4589 device->state.conn == C_BEHIND ||
4590 device->state.conn < C_CONNECTED ||
4591 device->state.disk < D_NEGOTIATING);
4593 /* D_ASSERT(device, device->state.conn == C_WF_SYNC_UUID ); */
4595 /* Here the _drbd_uuid_ functions are right, current should
4596 _not_ be rotated into the history */
4597 if (get_ldev_if_state(device, D_NEGOTIATING)) {
4598 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4599 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4601 drbd_print_uuids(device, "updated sync uuid");
4602 drbd_start_resync(device, C_SYNC_TARGET);
4606 drbd_err(device, "Ignoring SyncUUID packet!\n");
4612 * receive_bitmap_plain
4614 * Return 0 when done, 1 when another iteration is needed, and a negative error
4615 * code upon failure.
4618 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4619 unsigned long *p, struct bm_xfer_ctx *c)
4621 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4622 drbd_header_size(peer_device->connection);
4623 unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4624 c->bm_words - c->word_offset);
4625 unsigned int want = num_words * sizeof(*p);
4629 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4634 err = drbd_recv_all(peer_device->connection, p, want);
4638 drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4640 c->word_offset += num_words;
4641 c->bit_offset = c->word_offset * BITS_PER_LONG;
4642 if (c->bit_offset > c->bm_bits)
4643 c->bit_offset = c->bm_bits;
4648 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4650 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4653 static int dcbp_get_start(struct p_compressed_bm *p)
4655 return (p->encoding & 0x80) != 0;
4658 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4660 return (p->encoding >> 4) & 0x7;
4666 * Return 0 when done, 1 when another iteration is needed, and a negative error
4667 * code upon failure.
4670 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4671 struct p_compressed_bm *p,
4672 struct bm_xfer_ctx *c,
4675 struct bitstream bs;
4679 unsigned long s = c->bit_offset;
4681 int toggle = dcbp_get_start(p);
4685 bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4687 bits = bitstream_get_bits(&bs, &look_ahead, 64);
4691 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4692 bits = vli_decode_bits(&rl, look_ahead);
4698 if (e >= c->bm_bits) {
4699 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4702 _drbd_bm_set_bits(peer_device->device, s, e);
4706 drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4707 have, bits, look_ahead,
4708 (unsigned int)(bs.cur.b - p->code),
4709 (unsigned int)bs.buf_len);
4712 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4713 if (likely(bits < 64))
4714 look_ahead >>= bits;
4719 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4722 look_ahead |= tmp << have;
4727 bm_xfer_ctx_bit_to_word_offset(c);
4729 return (s != c->bm_bits);
4735 * Return 0 when done, 1 when another iteration is needed, and a negative error
4736 * code upon failure.
4739 decode_bitmap_c(struct drbd_peer_device *peer_device,
4740 struct p_compressed_bm *p,
4741 struct bm_xfer_ctx *c,
4744 if (dcbp_get_code(p) == RLE_VLI_Bits)
4745 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4747 /* other variants had been implemented for evaluation,
4748 * but have been dropped as this one turned out to be "best"
4749 * during all our tests. */
4751 drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4752 conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4756 void INFO_bm_xfer_stats(struct drbd_device *device,
4757 const char *direction, struct bm_xfer_ctx *c)
4759 /* what would it take to transfer it "plaintext" */
4760 unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4761 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4762 unsigned int plain =
4763 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4764 c->bm_words * sizeof(unsigned long);
4765 unsigned int total = c->bytes[0] + c->bytes[1];
4768 /* total can not be zero. but just in case: */
4772 /* don't report if not compressed */
4776 /* total < plain. check for overflow, still */
4777 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4778 : (1000 * total / plain);
4784 drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4785 "total %u; compression: %u.%u%%\n",
4787 c->bytes[1], c->packets[1],
4788 c->bytes[0], c->packets[0],
4789 total, r/10, r % 10);
4792 /* Since we are processing the bitfield from lower addresses to higher,
4793 it does not matter if the process it in 32 bit chunks or 64 bit
4794 chunks as long as it is little endian. (Understand it as byte stream,
4795 beginning with the lowest byte...) If we would use big endian
4796 we would need to process it from the highest address to the lowest,
4797 in order to be agnostic to the 32 vs 64 bits issue.
4799 returns 0 on failure, 1 if we successfully received it. */
4800 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4802 struct drbd_peer_device *peer_device;
4803 struct drbd_device *device;
4804 struct bm_xfer_ctx c;
4807 peer_device = conn_peer_device(connection, pi->vnr);
4810 device = peer_device->device;
4812 drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4813 /* you are supposed to send additional out-of-sync information
4814 * if you actually set bits during this phase */
4816 c = (struct bm_xfer_ctx) {
4817 .bm_bits = drbd_bm_bits(device),
4818 .bm_words = drbd_bm_words(device),
4822 if (pi->cmd == P_BITMAP)
4823 err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4824 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4825 /* MAYBE: sanity check that we speak proto >= 90,
4826 * and the feature is enabled! */
4827 struct p_compressed_bm *p = pi->data;
4829 if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4830 drbd_err(device, "ReportCBitmap packet too large\n");
4834 if (pi->size <= sizeof(*p)) {
4835 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4839 err = drbd_recv_all(peer_device->connection, p, pi->size);
4842 err = decode_bitmap_c(peer_device, p, &c, pi->size);
4844 drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4849 c.packets[pi->cmd == P_BITMAP]++;
4850 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4857 err = drbd_recv_header(peer_device->connection, pi);
4862 INFO_bm_xfer_stats(device, "receive", &c);
4864 if (device->state.conn == C_WF_BITMAP_T) {
4865 enum drbd_state_rv rv;
4867 err = drbd_send_bitmap(device);
4870 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4871 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4872 D_ASSERT(device, rv == SS_SUCCESS);
4873 } else if (device->state.conn != C_WF_BITMAP_S) {
4874 /* admin may have requested C_DISCONNECTING,
4875 * other threads may have noticed network errors */
4876 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4877 drbd_conn_str(device->state.conn));
4882 drbd_bm_unlock(device);
4883 if (!err && device->state.conn == C_WF_BITMAP_S)
4884 drbd_start_resync(device, C_SYNC_SOURCE);
4888 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4890 drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4893 return ignore_remaining_packet(connection, pi);
4896 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4898 /* Make sure we've acked all the TCP data associated
4899 * with the data requests being unplugged */
4900 tcp_sock_set_quickack(connection->data.socket->sk, 2);
4904 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4906 struct drbd_peer_device *peer_device;
4907 struct drbd_device *device;
4908 struct p_block_desc *p = pi->data;
4910 peer_device = conn_peer_device(connection, pi->vnr);
4913 device = peer_device->device;
4915 switch (device->state.conn) {
4916 case C_WF_SYNC_UUID:
4921 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4922 drbd_conn_str(device->state.conn));
4925 drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4930 static int receive_rs_deallocated(struct drbd_connection *connection, struct packet_info *pi)
4932 struct drbd_peer_device *peer_device;
4933 struct p_block_desc *p = pi->data;
4934 struct drbd_device *device;
4938 peer_device = conn_peer_device(connection, pi->vnr);
4941 device = peer_device->device;
4943 sector = be64_to_cpu(p->sector);
4944 size = be32_to_cpu(p->blksize);
4946 dec_rs_pending(device);
4948 if (get_ldev(device)) {
4949 struct drbd_peer_request *peer_req;
4950 const enum req_op op = REQ_OP_WRITE_ZEROES;
4952 peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER, sector,
4959 peer_req->w.cb = e_end_resync_block;
4960 peer_req->submit_jif = jiffies;
4961 peer_req->flags |= EE_TRIM;
4963 spin_lock_irq(&device->resource->req_lock);
4964 list_add_tail(&peer_req->w.list, &device->sync_ee);
4965 spin_unlock_irq(&device->resource->req_lock);
4967 atomic_add(pi->size >> 9, &device->rs_sect_ev);
4968 err = drbd_submit_peer_request(device, peer_req, op,
4972 spin_lock_irq(&device->resource->req_lock);
4973 list_del(&peer_req->w.list);
4974 spin_unlock_irq(&device->resource->req_lock);
4976 drbd_free_peer_req(device, peer_req);
4982 inc_unacked(device);
4984 /* No put_ldev() here. Gets called in drbd_endio_write_sec_final(),
4985 as well as drbd_rs_complete_io() */
4988 drbd_rs_complete_io(device, sector);
4989 drbd_send_ack_ex(peer_device, P_NEG_ACK, sector, size, ID_SYNCER);
4992 atomic_add(size >> 9, &device->rs_sect_in);
4999 unsigned int pkt_size;
5000 int (*fn)(struct drbd_connection *, struct packet_info *);
5003 static struct data_cmd drbd_cmd_handler[] = {
5004 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
5005 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
5006 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
5007 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
5008 [P_BITMAP] = { 1, 0, receive_bitmap } ,
5009 [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
5010 [P_UNPLUG_REMOTE] = { 0, 0, receive_UnplugRemote },
5011 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5012 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5013 [P_SYNC_PARAM] = { 1, 0, receive_SyncParam },
5014 [P_SYNC_PARAM89] = { 1, 0, receive_SyncParam },
5015 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
5016 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
5017 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
5018 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
5019 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
5020 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
5021 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5022 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5023 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5024 [P_RS_THIN_REQ] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5025 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
5026 [P_OUT_OF_SYNC] = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
5027 [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
5028 [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
5029 [P_TRIM] = { 0, sizeof(struct p_trim), receive_Data },
5030 [P_ZEROES] = { 0, sizeof(struct p_trim), receive_Data },
5031 [P_RS_DEALLOCATED] = { 0, sizeof(struct p_block_desc), receive_rs_deallocated },
5034 static void drbdd(struct drbd_connection *connection)
5036 struct packet_info pi;
5037 size_t shs; /* sub header size */
5040 while (get_t_state(&connection->receiver) == RUNNING) {
5041 struct data_cmd const *cmd;
5043 drbd_thread_current_set_cpu(&connection->receiver);
5044 update_receiver_timing_details(connection, drbd_recv_header_maybe_unplug);
5045 if (drbd_recv_header_maybe_unplug(connection, &pi))
5048 cmd = &drbd_cmd_handler[pi.cmd];
5049 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
5050 drbd_err(connection, "Unexpected data packet %s (0x%04x)",
5051 cmdname(pi.cmd), pi.cmd);
5055 shs = cmd->pkt_size;
5056 if (pi.cmd == P_SIZES && connection->agreed_features & DRBD_FF_WSAME)
5057 shs += sizeof(struct o_qlim);
5058 if (pi.size > shs && !cmd->expect_payload) {
5059 drbd_err(connection, "No payload expected %s l:%d\n",
5060 cmdname(pi.cmd), pi.size);
5063 if (pi.size < shs) {
5064 drbd_err(connection, "%s: unexpected packet size, expected:%d received:%d\n",
5065 cmdname(pi.cmd), (int)shs, pi.size);
5070 update_receiver_timing_details(connection, drbd_recv_all_warn);
5071 err = drbd_recv_all_warn(connection, pi.data, shs);
5077 update_receiver_timing_details(connection, cmd->fn);
5078 err = cmd->fn(connection, &pi);
5080 drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
5081 cmdname(pi.cmd), err, pi.size);
5088 conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
5091 static void conn_disconnect(struct drbd_connection *connection)
5093 struct drbd_peer_device *peer_device;
5097 if (connection->cstate == C_STANDALONE)
5100 /* We are about to start the cleanup after connection loss.
5101 * Make sure drbd_make_request knows about that.
5102 * Usually we should be in some network failure state already,
5103 * but just in case we are not, we fix it up here.
5105 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5107 /* ack_receiver does not clean up anything. it must not interfere, either */
5108 drbd_thread_stop(&connection->ack_receiver);
5109 if (connection->ack_sender) {
5110 destroy_workqueue(connection->ack_sender);
5111 connection->ack_sender = NULL;
5113 drbd_free_sock(connection);
5116 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5117 struct drbd_device *device = peer_device->device;
5118 kref_get(&device->kref);
5120 drbd_disconnected(peer_device);
5121 kref_put(&device->kref, drbd_destroy_device);
5126 if (!list_empty(&connection->current_epoch->list))
5127 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
5128 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
5129 atomic_set(&connection->current_epoch->epoch_size, 0);
5130 connection->send.seen_any_write_yet = false;
5132 drbd_info(connection, "Connection closed\n");
5134 if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
5135 conn_try_outdate_peer_async(connection);
5137 spin_lock_irq(&connection->resource->req_lock);
5138 oc = connection->cstate;
5139 if (oc >= C_UNCONNECTED)
5140 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
5142 spin_unlock_irq(&connection->resource->req_lock);
5144 if (oc == C_DISCONNECTING)
5145 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
5148 static int drbd_disconnected(struct drbd_peer_device *peer_device)
5150 struct drbd_device *device = peer_device->device;
5153 /* wait for current activity to cease. */
5154 spin_lock_irq(&device->resource->req_lock);
5155 _drbd_wait_ee_list_empty(device, &device->active_ee);
5156 _drbd_wait_ee_list_empty(device, &device->sync_ee);
5157 _drbd_wait_ee_list_empty(device, &device->read_ee);
5158 spin_unlock_irq(&device->resource->req_lock);
5160 /* We do not have data structures that would allow us to
5161 * get the rs_pending_cnt down to 0 again.
5162 * * On C_SYNC_TARGET we do not have any data structures describing
5163 * the pending RSDataRequest's we have sent.
5164 * * On C_SYNC_SOURCE there is no data structure that tracks
5165 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
5166 * And no, it is not the sum of the reference counts in the
5167 * resync_LRU. The resync_LRU tracks the whole operation including
5168 * the disk-IO, while the rs_pending_cnt only tracks the blocks
5170 drbd_rs_cancel_all(device);
5171 device->rs_total = 0;
5172 device->rs_failed = 0;
5173 atomic_set(&device->rs_pending_cnt, 0);
5174 wake_up(&device->misc_wait);
5176 del_timer_sync(&device->resync_timer);
5177 resync_timer_fn(&device->resync_timer);
5179 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
5180 * w_make_resync_request etc. which may still be on the worker queue
5181 * to be "canceled" */
5182 drbd_flush_workqueue(&peer_device->connection->sender_work);
5184 drbd_finish_peer_reqs(device);
5186 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
5187 might have issued a work again. The one before drbd_finish_peer_reqs() is
5188 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
5189 drbd_flush_workqueue(&peer_device->connection->sender_work);
5191 /* need to do it again, drbd_finish_peer_reqs() may have populated it
5192 * again via drbd_try_clear_on_disk_bm(). */
5193 drbd_rs_cancel_all(device);
5195 kfree(device->p_uuid);
5196 device->p_uuid = NULL;
5198 if (!drbd_suspended(device))
5199 tl_clear(peer_device->connection);
5201 drbd_md_sync(device);
5203 if (get_ldev(device)) {
5204 drbd_bitmap_io(device, &drbd_bm_write_copy_pages,
5205 "write from disconnected", BM_LOCKED_CHANGE_ALLOWED);
5209 /* tcp_close and release of sendpage pages can be deferred. I don't
5210 * want to use SO_LINGER, because apparently it can be deferred for
5211 * more than 20 seconds (longest time I checked).
5213 * Actually we don't care for exactly when the network stack does its
5214 * put_page(), but release our reference on these pages right here.
5216 i = drbd_free_peer_reqs(device, &device->net_ee);
5218 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
5219 i = atomic_read(&device->pp_in_use_by_net);
5221 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
5222 i = atomic_read(&device->pp_in_use);
5224 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
5226 D_ASSERT(device, list_empty(&device->read_ee));
5227 D_ASSERT(device, list_empty(&device->active_ee));
5228 D_ASSERT(device, list_empty(&device->sync_ee));
5229 D_ASSERT(device, list_empty(&device->done_ee));
5235 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
5236 * we can agree on is stored in agreed_pro_version.
5238 * feature flags and the reserved array should be enough room for future
5239 * enhancements of the handshake protocol, and possible plugins...
5241 * for now, they are expected to be zero, but ignored.
5243 static int drbd_send_features(struct drbd_connection *connection)
5245 struct drbd_socket *sock;
5246 struct p_connection_features *p;
5248 sock = &connection->data;
5249 p = conn_prepare_command(connection, sock);
5252 memset(p, 0, sizeof(*p));
5253 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
5254 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
5255 p->feature_flags = cpu_to_be32(PRO_FEATURES);
5256 return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
5261 * 1 yes, we have a valid connection
5262 * 0 oops, did not work out, please try again
5263 * -1 peer talks different language,
5264 * no point in trying again, please go standalone.
5266 static int drbd_do_features(struct drbd_connection *connection)
5268 /* ASSERT current == connection->receiver ... */
5269 struct p_connection_features *p;
5270 const int expect = sizeof(struct p_connection_features);
5271 struct packet_info pi;
5274 err = drbd_send_features(connection);
5278 err = drbd_recv_header(connection, &pi);
5282 if (pi.cmd != P_CONNECTION_FEATURES) {
5283 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
5284 cmdname(pi.cmd), pi.cmd);
5288 if (pi.size != expect) {
5289 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
5295 err = drbd_recv_all_warn(connection, p, expect);
5299 p->protocol_min = be32_to_cpu(p->protocol_min);
5300 p->protocol_max = be32_to_cpu(p->protocol_max);
5301 if (p->protocol_max == 0)
5302 p->protocol_max = p->protocol_min;
5304 if (PRO_VERSION_MAX < p->protocol_min ||
5305 PRO_VERSION_MIN > p->protocol_max)
5308 connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
5309 connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
5311 drbd_info(connection, "Handshake successful: "
5312 "Agreed network protocol version %d\n", connection->agreed_pro_version);
5314 drbd_info(connection, "Feature flags enabled on protocol level: 0x%x%s%s%s%s.\n",
5315 connection->agreed_features,
5316 connection->agreed_features & DRBD_FF_TRIM ? " TRIM" : "",
5317 connection->agreed_features & DRBD_FF_THIN_RESYNC ? " THIN_RESYNC" : "",
5318 connection->agreed_features & DRBD_FF_WSAME ? " WRITE_SAME" : "",
5319 connection->agreed_features & DRBD_FF_WZEROES ? " WRITE_ZEROES" :
5320 connection->agreed_features ? "" : " none");
5325 drbd_err(connection, "incompatible DRBD dialects: "
5326 "I support %d-%d, peer supports %d-%d\n",
5327 PRO_VERSION_MIN, PRO_VERSION_MAX,
5328 p->protocol_min, p->protocol_max);
5332 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
5333 static int drbd_do_auth(struct drbd_connection *connection)
5335 drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
5336 drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
5340 #define CHALLENGE_LEN 64
5344 0 - failed, try again (network error),
5345 -1 - auth failed, don't try again.
5348 static int drbd_do_auth(struct drbd_connection *connection)
5350 struct drbd_socket *sock;
5351 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
5352 char *response = NULL;
5353 char *right_response = NULL;
5354 char *peers_ch = NULL;
5355 unsigned int key_len;
5356 char secret[SHARED_SECRET_MAX]; /* 64 byte */
5357 unsigned int resp_size;
5358 struct shash_desc *desc;
5359 struct packet_info pi;
5360 struct net_conf *nc;
5363 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
5366 nc = rcu_dereference(connection->net_conf);
5367 key_len = strlen(nc->shared_secret);
5368 memcpy(secret, nc->shared_secret, key_len);
5371 desc = kmalloc(sizeof(struct shash_desc) +
5372 crypto_shash_descsize(connection->cram_hmac_tfm),
5378 desc->tfm = connection->cram_hmac_tfm;
5380 rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
5382 drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
5387 get_random_bytes(my_challenge, CHALLENGE_LEN);
5389 sock = &connection->data;
5390 if (!conn_prepare_command(connection, sock)) {
5394 rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
5395 my_challenge, CHALLENGE_LEN);
5399 err = drbd_recv_header(connection, &pi);
5405 if (pi.cmd != P_AUTH_CHALLENGE) {
5406 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
5407 cmdname(pi.cmd), pi.cmd);
5412 if (pi.size > CHALLENGE_LEN * 2) {
5413 drbd_err(connection, "expected AuthChallenge payload too big.\n");
5418 if (pi.size < CHALLENGE_LEN) {
5419 drbd_err(connection, "AuthChallenge payload too small.\n");
5424 peers_ch = kmalloc(pi.size, GFP_NOIO);
5430 err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5436 if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5437 drbd_err(connection, "Peer presented the same challenge!\n");
5442 resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5443 response = kmalloc(resp_size, GFP_NOIO);
5449 rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5451 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5456 if (!conn_prepare_command(connection, sock)) {
5460 rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5461 response, resp_size);
5465 err = drbd_recv_header(connection, &pi);
5471 if (pi.cmd != P_AUTH_RESPONSE) {
5472 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5473 cmdname(pi.cmd), pi.cmd);
5478 if (pi.size != resp_size) {
5479 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5484 err = drbd_recv_all_warn(connection, response , resp_size);
5490 right_response = kmalloc(resp_size, GFP_NOIO);
5491 if (!right_response) {
5496 rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5499 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5504 rv = !memcmp(response, right_response, resp_size);
5507 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5515 kfree(right_response);
5517 shash_desc_zero(desc);
5525 int drbd_receiver(struct drbd_thread *thi)
5527 struct drbd_connection *connection = thi->connection;
5530 drbd_info(connection, "receiver (re)started\n");
5533 h = conn_connect(connection);
5535 conn_disconnect(connection);
5536 schedule_timeout_interruptible(HZ);
5539 drbd_warn(connection, "Discarding network configuration.\n");
5540 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5545 blk_start_plug(&connection->receiver_plug);
5547 blk_finish_plug(&connection->receiver_plug);
5550 conn_disconnect(connection);
5552 drbd_info(connection, "receiver terminated\n");
5556 /* ********* acknowledge sender ******** */
5558 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5560 struct p_req_state_reply *p = pi->data;
5561 int retcode = be32_to_cpu(p->retcode);
5563 if (retcode >= SS_SUCCESS) {
5564 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5566 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5567 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5568 drbd_set_st_err_str(retcode), retcode);
5570 wake_up(&connection->ping_wait);
5575 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5577 struct drbd_peer_device *peer_device;
5578 struct drbd_device *device;
5579 struct p_req_state_reply *p = pi->data;
5580 int retcode = be32_to_cpu(p->retcode);
5582 peer_device = conn_peer_device(connection, pi->vnr);
5585 device = peer_device->device;
5587 if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5588 D_ASSERT(device, connection->agreed_pro_version < 100);
5589 return got_conn_RqSReply(connection, pi);
5592 if (retcode >= SS_SUCCESS) {
5593 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5595 set_bit(CL_ST_CHG_FAIL, &device->flags);
5596 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5597 drbd_set_st_err_str(retcode), retcode);
5599 wake_up(&device->state_wait);
5604 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5606 return drbd_send_ping_ack(connection);
5610 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5612 /* restore idle timeout */
5613 connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5614 if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5615 wake_up(&connection->ping_wait);
5620 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5622 struct drbd_peer_device *peer_device;
5623 struct drbd_device *device;
5624 struct p_block_ack *p = pi->data;
5625 sector_t sector = be64_to_cpu(p->sector);
5626 int blksize = be32_to_cpu(p->blksize);
5628 peer_device = conn_peer_device(connection, pi->vnr);
5631 device = peer_device->device;
5633 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5635 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5637 if (get_ldev(device)) {
5638 drbd_rs_complete_io(device, sector);
5639 drbd_set_in_sync(device, sector, blksize);
5640 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5641 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5644 dec_rs_pending(device);
5645 atomic_add(blksize >> 9, &device->rs_sect_in);
5651 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5652 struct rb_root *root, const char *func,
5653 enum drbd_req_event what, bool missing_ok)
5655 struct drbd_request *req;
5656 struct bio_and_error m;
5658 spin_lock_irq(&device->resource->req_lock);
5659 req = find_request(device, root, id, sector, missing_ok, func);
5660 if (unlikely(!req)) {
5661 spin_unlock_irq(&device->resource->req_lock);
5664 __req_mod(req, what, &m);
5665 spin_unlock_irq(&device->resource->req_lock);
5668 complete_master_bio(device, &m);
5672 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5674 struct drbd_peer_device *peer_device;
5675 struct drbd_device *device;
5676 struct p_block_ack *p = pi->data;
5677 sector_t sector = be64_to_cpu(p->sector);
5678 int blksize = be32_to_cpu(p->blksize);
5679 enum drbd_req_event what;
5681 peer_device = conn_peer_device(connection, pi->vnr);
5684 device = peer_device->device;
5686 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5688 if (p->block_id == ID_SYNCER) {
5689 drbd_set_in_sync(device, sector, blksize);
5690 dec_rs_pending(device);
5694 case P_RS_WRITE_ACK:
5695 what = WRITE_ACKED_BY_PEER_AND_SIS;
5698 what = WRITE_ACKED_BY_PEER;
5701 what = RECV_ACKED_BY_PEER;
5704 what = CONFLICT_RESOLVED;
5707 what = POSTPONE_WRITE;
5713 return validate_req_change_req_state(device, p->block_id, sector,
5714 &device->write_requests, __func__,
5718 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5720 struct drbd_peer_device *peer_device;
5721 struct drbd_device *device;
5722 struct p_block_ack *p = pi->data;
5723 sector_t sector = be64_to_cpu(p->sector);
5724 int size = be32_to_cpu(p->blksize);
5727 peer_device = conn_peer_device(connection, pi->vnr);
5730 device = peer_device->device;
5732 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5734 if (p->block_id == ID_SYNCER) {
5735 dec_rs_pending(device);
5736 drbd_rs_failed_io(device, sector, size);
5740 err = validate_req_change_req_state(device, p->block_id, sector,
5741 &device->write_requests, __func__,
5744 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5745 The master bio might already be completed, therefore the
5746 request is no longer in the collision hash. */
5747 /* In Protocol B we might already have got a P_RECV_ACK
5748 but then get a P_NEG_ACK afterwards. */
5749 drbd_set_out_of_sync(device, sector, size);
5754 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5756 struct drbd_peer_device *peer_device;
5757 struct drbd_device *device;
5758 struct p_block_ack *p = pi->data;
5759 sector_t sector = be64_to_cpu(p->sector);
5761 peer_device = conn_peer_device(connection, pi->vnr);
5764 device = peer_device->device;
5766 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5768 drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5769 (unsigned long long)sector, be32_to_cpu(p->blksize));
5771 return validate_req_change_req_state(device, p->block_id, sector,
5772 &device->read_requests, __func__,
5776 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5778 struct drbd_peer_device *peer_device;
5779 struct drbd_device *device;
5782 struct p_block_ack *p = pi->data;
5784 peer_device = conn_peer_device(connection, pi->vnr);
5787 device = peer_device->device;
5789 sector = be64_to_cpu(p->sector);
5790 size = be32_to_cpu(p->blksize);
5792 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5794 dec_rs_pending(device);
5796 if (get_ldev_if_state(device, D_FAILED)) {
5797 drbd_rs_complete_io(device, sector);
5799 case P_NEG_RS_DREPLY:
5800 drbd_rs_failed_io(device, sector, size);
5813 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5815 struct p_barrier_ack *p = pi->data;
5816 struct drbd_peer_device *peer_device;
5819 tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5822 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5823 struct drbd_device *device = peer_device->device;
5825 if (device->state.conn == C_AHEAD &&
5826 atomic_read(&device->ap_in_flight) == 0 &&
5827 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5828 device->start_resync_timer.expires = jiffies + HZ;
5829 add_timer(&device->start_resync_timer);
5837 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5839 struct drbd_peer_device *peer_device;
5840 struct drbd_device *device;
5841 struct p_block_ack *p = pi->data;
5842 struct drbd_device_work *dw;
5846 peer_device = conn_peer_device(connection, pi->vnr);
5849 device = peer_device->device;
5851 sector = be64_to_cpu(p->sector);
5852 size = be32_to_cpu(p->blksize);
5854 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5856 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5857 drbd_ov_out_of_sync_found(device, sector, size);
5859 ov_out_of_sync_print(device);
5861 if (!get_ldev(device))
5864 drbd_rs_complete_io(device, sector);
5865 dec_rs_pending(device);
5869 /* let's advance progress step marks only for every other megabyte */
5870 if ((device->ov_left & 0x200) == 0x200)
5871 drbd_advance_rs_marks(device, device->ov_left);
5873 if (device->ov_left == 0) {
5874 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5876 dw->w.cb = w_ov_finished;
5877 dw->device = device;
5878 drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5880 drbd_err(device, "kmalloc(dw) failed.");
5881 ov_out_of_sync_print(device);
5882 drbd_resync_finished(device);
5889 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5894 struct meta_sock_cmd {
5896 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5899 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5902 struct net_conf *nc;
5905 nc = rcu_dereference(connection->net_conf);
5906 t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5913 connection->meta.socket->sk->sk_rcvtimeo = t;
5916 static void set_ping_timeout(struct drbd_connection *connection)
5918 set_rcvtimeo(connection, 1);
5921 static void set_idle_timeout(struct drbd_connection *connection)
5923 set_rcvtimeo(connection, 0);
5926 static struct meta_sock_cmd ack_receiver_tbl[] = {
5927 [P_PING] = { 0, got_Ping },
5928 [P_PING_ACK] = { 0, got_PingAck },
5929 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5930 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5931 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5932 [P_SUPERSEDED] = { sizeof(struct p_block_ack), got_BlockAck },
5933 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
5934 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
5935 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply },
5936 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
5937 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
5938 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5939 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
5940 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
5941 [P_RS_CANCEL] = { sizeof(struct p_block_ack), got_NegRSDReply },
5942 [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5943 [P_RETRY_WRITE] = { sizeof(struct p_block_ack), got_BlockAck },
5946 int drbd_ack_receiver(struct drbd_thread *thi)
5948 struct drbd_connection *connection = thi->connection;
5949 struct meta_sock_cmd *cmd = NULL;
5950 struct packet_info pi;
5951 unsigned long pre_recv_jif;
5953 void *buf = connection->meta.rbuf;
5955 unsigned int header_size = drbd_header_size(connection);
5956 int expect = header_size;
5957 bool ping_timeout_active = false;
5959 sched_set_fifo_low(current);
5961 while (get_t_state(thi) == RUNNING) {
5962 drbd_thread_current_set_cpu(thi);
5964 conn_reclaim_net_peer_reqs(connection);
5966 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5967 if (drbd_send_ping(connection)) {
5968 drbd_err(connection, "drbd_send_ping has failed\n");
5971 set_ping_timeout(connection);
5972 ping_timeout_active = true;
5975 pre_recv_jif = jiffies;
5976 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5979 * -EINTR (on meta) we got a signal
5980 * -EAGAIN (on meta) rcvtimeo expired
5981 * -ECONNRESET other side closed the connection
5982 * -ERESTARTSYS (on data) we got a signal
5983 * rv < 0 other than above: unexpected error!
5984 * rv == expected: full header or command
5985 * rv < expected: "woken" by signal during receive
5986 * rv == 0 : "connection shut down by peer"
5988 if (likely(rv > 0)) {
5991 } else if (rv == 0) {
5992 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
5995 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
5998 t = wait_event_timeout(connection->ping_wait,
5999 connection->cstate < C_WF_REPORT_PARAMS,
6004 drbd_err(connection, "meta connection shut down by peer.\n");
6006 } else if (rv == -EAGAIN) {
6007 /* If the data socket received something meanwhile,
6008 * that is good enough: peer is still alive. */
6009 if (time_after(connection->last_received, pre_recv_jif))
6011 if (ping_timeout_active) {
6012 drbd_err(connection, "PingAck did not arrive in time.\n");
6015 set_bit(SEND_PING, &connection->flags);
6017 } else if (rv == -EINTR) {
6018 /* maybe drbd_thread_stop(): the while condition will notice.
6019 * maybe woken for send_ping: we'll send a ping above,
6020 * and change the rcvtimeo */
6021 flush_signals(current);
6024 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
6028 if (received == expect && cmd == NULL) {
6029 if (decode_header(connection, connection->meta.rbuf, &pi))
6031 cmd = &ack_receiver_tbl[pi.cmd];
6032 if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
6033 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
6034 cmdname(pi.cmd), pi.cmd);
6037 expect = header_size + cmd->pkt_size;
6038 if (pi.size != expect - header_size) {
6039 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
6044 if (received == expect) {
6047 err = cmd->fn(connection, &pi);
6049 drbd_err(connection, "%ps failed\n", cmd->fn);
6053 connection->last_received = jiffies;
6055 if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
6056 set_idle_timeout(connection);
6057 ping_timeout_active = false;
6060 buf = connection->meta.rbuf;
6062 expect = header_size;
6069 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6070 conn_md_sync(connection);
6074 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
6077 drbd_info(connection, "ack_receiver terminated\n");
6082 void drbd_send_acks_wf(struct work_struct *ws)
6084 struct drbd_peer_device *peer_device =
6085 container_of(ws, struct drbd_peer_device, send_acks_work);
6086 struct drbd_connection *connection = peer_device->connection;
6087 struct drbd_device *device = peer_device->device;
6088 struct net_conf *nc;
6092 nc = rcu_dereference(connection->net_conf);
6093 tcp_cork = nc->tcp_cork;
6097 tcp_sock_set_cork(connection->meta.socket->sk, true);
6099 err = drbd_finish_peer_reqs(device);
6100 kref_put(&device->kref, drbd_destroy_device);
6101 /* get is in drbd_endio_write_sec_final(). That is necessary to keep the
6102 struct work_struct send_acks_work alive, which is in the peer_device object */
6105 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6110 tcp_sock_set_cork(connection->meta.socket->sk, false);