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 nr_pages = (payload_size + PAGE_SIZE -1) >> PAGE_SHIFT;
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 unsigned op, const unsigned op_flags,
1625 const int fault_type)
1627 struct bio *bios = NULL;
1629 struct page *page = peer_req->pages;
1630 sector_t sector = peer_req->i.sector;
1631 unsigned data_size = peer_req->i.size;
1632 unsigned n_bios = 0;
1633 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
1635 /* TRIM/DISCARD: for now, always use the helper function
1636 * blkdev_issue_zeroout(..., discard=true).
1637 * It's synchronous, but it does the right thing wrt. bio splitting.
1638 * Correctness first, performance later. Next step is to code an
1639 * asynchronous variant of the same.
1641 if (peer_req->flags & (EE_TRIM | EE_ZEROOUT)) {
1642 /* wait for all pending IO completions, before we start
1643 * zeroing things out. */
1644 conn_wait_active_ee_empty(peer_req->peer_device->connection);
1645 /* add it to the active list now,
1646 * so we can find it to present it in debugfs */
1647 peer_req->submit_jif = jiffies;
1648 peer_req->flags |= EE_SUBMITTED;
1650 /* If this was a resync request from receive_rs_deallocated(),
1651 * it is already on the sync_ee list */
1652 if (list_empty(&peer_req->w.list)) {
1653 spin_lock_irq(&device->resource->req_lock);
1654 list_add_tail(&peer_req->w.list, &device->active_ee);
1655 spin_unlock_irq(&device->resource->req_lock);
1658 drbd_issue_peer_discard_or_zero_out(device, peer_req);
1662 /* In most cases, we will only need one bio. But in case the lower
1663 * level restrictions happen to be different at this offset on this
1664 * side than those of the sending peer, we may need to submit the
1665 * request in more than one bio.
1667 * Plain bio_alloc is good enough here, this is no DRBD internally
1668 * generated bio, but a bio allocated on behalf of the peer.
1671 bio = bio_alloc(device->ldev->backing_bdev, nr_pages, op | op_flags,
1673 /* > peer_req->i.sector, unless this is the first bio */
1674 bio->bi_iter.bi_sector = sector;
1675 bio->bi_private = peer_req;
1676 bio->bi_end_io = drbd_peer_request_endio;
1678 bio->bi_next = bios;
1682 page_chain_for_each(page) {
1683 unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1684 if (!bio_add_page(bio, page, len, 0))
1690 D_ASSERT(device, data_size == 0);
1691 D_ASSERT(device, page == NULL);
1693 atomic_set(&peer_req->pending_bios, n_bios);
1694 /* for debugfs: update timestamp, mark as submitted */
1695 peer_req->submit_jif = jiffies;
1696 peer_req->flags |= EE_SUBMITTED;
1699 bios = bios->bi_next;
1700 bio->bi_next = NULL;
1702 drbd_submit_bio_noacct(device, fault_type, bio);
1707 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1708 struct drbd_peer_request *peer_req)
1710 struct drbd_interval *i = &peer_req->i;
1712 drbd_remove_interval(&device->write_requests, i);
1713 drbd_clear_interval(i);
1715 /* Wake up any processes waiting for this peer request to complete. */
1717 wake_up(&device->misc_wait);
1720 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1722 struct drbd_peer_device *peer_device;
1726 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1727 struct drbd_device *device = peer_device->device;
1729 kref_get(&device->kref);
1731 drbd_wait_ee_list_empty(device, &device->active_ee);
1732 kref_put(&device->kref, drbd_destroy_device);
1738 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1741 struct p_barrier *p = pi->data;
1742 struct drbd_epoch *epoch;
1744 /* FIXME these are unacked on connection,
1745 * not a specific (peer)device.
1747 connection->current_epoch->barrier_nr = p->barrier;
1748 connection->current_epoch->connection = connection;
1749 rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1751 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1752 * the activity log, which means it would not be resynced in case the
1753 * R_PRIMARY crashes now.
1754 * Therefore we must send the barrier_ack after the barrier request was
1756 switch (connection->resource->write_ordering) {
1758 if (rv == FE_RECYCLED)
1761 /* receiver context, in the writeout path of the other node.
1762 * avoid potential distributed deadlock */
1763 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1767 drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1772 conn_wait_active_ee_empty(connection);
1773 drbd_flush(connection);
1775 if (atomic_read(&connection->current_epoch->epoch_size)) {
1776 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1783 drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1784 connection->resource->write_ordering);
1789 atomic_set(&epoch->epoch_size, 0);
1790 atomic_set(&epoch->active, 0);
1792 spin_lock(&connection->epoch_lock);
1793 if (atomic_read(&connection->current_epoch->epoch_size)) {
1794 list_add(&epoch->list, &connection->current_epoch->list);
1795 connection->current_epoch = epoch;
1796 connection->epochs++;
1798 /* The current_epoch got recycled while we allocated this one... */
1801 spin_unlock(&connection->epoch_lock);
1806 /* quick wrapper in case payload size != request_size (write same) */
1807 static void drbd_csum_ee_size(struct crypto_shash *h,
1808 struct drbd_peer_request *r, void *d,
1809 unsigned int payload_size)
1811 unsigned int tmp = r->i.size;
1812 r->i.size = payload_size;
1813 drbd_csum_ee(h, r, d);
1817 /* used from receive_RSDataReply (recv_resync_read)
1818 * and from receive_Data.
1819 * data_size: actual payload ("data in")
1820 * for normal writes that is bi_size.
1821 * for discards, that is zero.
1822 * for write same, it is logical_block_size.
1823 * both trim and write same have the bi_size ("data len to be affected")
1824 * as extra argument in the packet header.
1826 static struct drbd_peer_request *
1827 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1828 struct packet_info *pi) __must_hold(local)
1830 struct drbd_device *device = peer_device->device;
1831 const sector_t capacity = get_capacity(device->vdisk);
1832 struct drbd_peer_request *peer_req;
1834 int digest_size, err;
1835 unsigned int data_size = pi->size, ds;
1836 void *dig_in = peer_device->connection->int_dig_in;
1837 void *dig_vv = peer_device->connection->int_dig_vv;
1838 unsigned long *data;
1839 struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1840 struct p_trim *zeroes = (pi->cmd == P_ZEROES) ? pi->data : NULL;
1843 if (!trim && peer_device->connection->peer_integrity_tfm) {
1844 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1846 * FIXME: Receive the incoming digest into the receive buffer
1847 * here, together with its struct p_data?
1849 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1852 data_size -= digest_size;
1855 /* assume request_size == data_size, but special case trim. */
1858 if (!expect(data_size == 0))
1860 ds = be32_to_cpu(trim->size);
1861 } else if (zeroes) {
1862 if (!expect(data_size == 0))
1864 ds = be32_to_cpu(zeroes->size);
1867 if (!expect(IS_ALIGNED(ds, 512)))
1869 if (trim || zeroes) {
1870 if (!expect(ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
1872 } else if (!expect(ds <= DRBD_MAX_BIO_SIZE))
1875 /* even though we trust out peer,
1876 * we sometimes have to double check. */
1877 if (sector + (ds>>9) > capacity) {
1878 drbd_err(device, "request from peer beyond end of local disk: "
1879 "capacity: %llus < sector: %llus + size: %u\n",
1880 (unsigned long long)capacity,
1881 (unsigned long long)sector, ds);
1885 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1886 * "criss-cross" setup, that might cause write-out on some other DRBD,
1887 * which in turn might block on the other node at this very place. */
1888 peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
1892 peer_req->flags |= EE_WRITE;
1894 peer_req->flags |= EE_TRIM;
1898 peer_req->flags |= EE_ZEROOUT;
1902 /* receive payload size bytes into page chain */
1904 page = peer_req->pages;
1905 page_chain_for_each(page) {
1906 unsigned len = min_t(int, ds, PAGE_SIZE);
1908 err = drbd_recv_all_warn(peer_device->connection, data, len);
1909 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1910 drbd_err(device, "Fault injection: Corrupting data on receive\n");
1911 data[0] = data[0] ^ (unsigned long)-1;
1915 drbd_free_peer_req(device, peer_req);
1922 drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
1923 if (memcmp(dig_in, dig_vv, digest_size)) {
1924 drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1925 (unsigned long long)sector, data_size);
1926 drbd_free_peer_req(device, peer_req);
1930 device->recv_cnt += data_size >> 9;
1934 /* drbd_drain_block() just takes a data block
1935 * out of the socket input buffer, and discards it.
1937 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1946 page = drbd_alloc_pages(peer_device, 1, 1);
1950 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1952 err = drbd_recv_all_warn(peer_device->connection, data, len);
1958 drbd_free_pages(peer_device->device, page, 0);
1962 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1963 sector_t sector, int data_size)
1965 struct bio_vec bvec;
1966 struct bvec_iter iter;
1968 int digest_size, err, expect;
1969 void *dig_in = peer_device->connection->int_dig_in;
1970 void *dig_vv = peer_device->connection->int_dig_vv;
1973 if (peer_device->connection->peer_integrity_tfm) {
1974 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1975 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1978 data_size -= digest_size;
1981 /* optimistically update recv_cnt. if receiving fails below,
1982 * we disconnect anyways, and counters will be reset. */
1983 peer_device->device->recv_cnt += data_size>>9;
1985 bio = req->master_bio;
1986 D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
1988 bio_for_each_segment(bvec, bio, iter) {
1989 void *mapped = bvec_kmap_local(&bvec);
1990 expect = min_t(int, data_size, bvec.bv_len);
1991 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
1992 kunmap_local(mapped);
1995 data_size -= expect;
1999 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
2000 if (memcmp(dig_in, dig_vv, digest_size)) {
2001 drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
2006 D_ASSERT(peer_device->device, data_size == 0);
2011 * e_end_resync_block() is called in ack_sender context via
2012 * drbd_finish_peer_reqs().
2014 static int e_end_resync_block(struct drbd_work *w, int unused)
2016 struct drbd_peer_request *peer_req =
2017 container_of(w, struct drbd_peer_request, w);
2018 struct drbd_peer_device *peer_device = peer_req->peer_device;
2019 struct drbd_device *device = peer_device->device;
2020 sector_t sector = peer_req->i.sector;
2023 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2025 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2026 drbd_set_in_sync(device, sector, peer_req->i.size);
2027 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
2029 /* Record failure to sync */
2030 drbd_rs_failed_io(device, sector, peer_req->i.size);
2032 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2034 dec_unacked(device);
2039 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
2040 struct packet_info *pi) __releases(local)
2042 struct drbd_device *device = peer_device->device;
2043 struct drbd_peer_request *peer_req;
2045 peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
2049 dec_rs_pending(device);
2051 inc_unacked(device);
2052 /* corresponding dec_unacked() in e_end_resync_block()
2053 * respective _drbd_clear_done_ee */
2055 peer_req->w.cb = e_end_resync_block;
2056 peer_req->submit_jif = jiffies;
2058 spin_lock_irq(&device->resource->req_lock);
2059 list_add_tail(&peer_req->w.list, &device->sync_ee);
2060 spin_unlock_irq(&device->resource->req_lock);
2062 atomic_add(pi->size >> 9, &device->rs_sect_ev);
2063 if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE, 0,
2064 DRBD_FAULT_RS_WR) == 0)
2067 /* don't care for the reason here */
2068 drbd_err(device, "submit failed, triggering re-connect\n");
2069 spin_lock_irq(&device->resource->req_lock);
2070 list_del(&peer_req->w.list);
2071 spin_unlock_irq(&device->resource->req_lock);
2073 drbd_free_peer_req(device, peer_req);
2079 static struct drbd_request *
2080 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
2081 sector_t sector, bool missing_ok, const char *func)
2083 struct drbd_request *req;
2085 /* Request object according to our peer */
2086 req = (struct drbd_request *)(unsigned long)id;
2087 if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
2090 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
2091 (unsigned long)id, (unsigned long long)sector);
2096 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
2098 struct drbd_peer_device *peer_device;
2099 struct drbd_device *device;
2100 struct drbd_request *req;
2103 struct p_data *p = pi->data;
2105 peer_device = conn_peer_device(connection, pi->vnr);
2108 device = peer_device->device;
2110 sector = be64_to_cpu(p->sector);
2112 spin_lock_irq(&device->resource->req_lock);
2113 req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
2114 spin_unlock_irq(&device->resource->req_lock);
2118 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
2119 * special casing it there for the various failure cases.
2120 * still no race with drbd_fail_pending_reads */
2121 err = recv_dless_read(peer_device, req, sector, pi->size);
2123 req_mod(req, DATA_RECEIVED);
2124 /* else: nothing. handled from drbd_disconnect...
2125 * I don't think we may complete this just yet
2126 * in case we are "on-disconnect: freeze" */
2131 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
2133 struct drbd_peer_device *peer_device;
2134 struct drbd_device *device;
2137 struct p_data *p = pi->data;
2139 peer_device = conn_peer_device(connection, pi->vnr);
2142 device = peer_device->device;
2144 sector = be64_to_cpu(p->sector);
2145 D_ASSERT(device, p->block_id == ID_SYNCER);
2147 if (get_ldev(device)) {
2148 /* data is submitted to disk within recv_resync_read.
2149 * corresponding put_ldev done below on error,
2150 * or in drbd_peer_request_endio. */
2151 err = recv_resync_read(peer_device, sector, pi);
2153 if (__ratelimit(&drbd_ratelimit_state))
2154 drbd_err(device, "Can not write resync data to local disk.\n");
2156 err = drbd_drain_block(peer_device, pi->size);
2158 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2161 atomic_add(pi->size >> 9, &device->rs_sect_in);
2166 static void restart_conflicting_writes(struct drbd_device *device,
2167 sector_t sector, int size)
2169 struct drbd_interval *i;
2170 struct drbd_request *req;
2172 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2175 req = container_of(i, struct drbd_request, i);
2176 if (req->rq_state & RQ_LOCAL_PENDING ||
2177 !(req->rq_state & RQ_POSTPONED))
2179 /* as it is RQ_POSTPONED, this will cause it to
2180 * be queued on the retry workqueue. */
2181 __req_mod(req, CONFLICT_RESOLVED, NULL);
2186 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2188 static int e_end_block(struct drbd_work *w, int cancel)
2190 struct drbd_peer_request *peer_req =
2191 container_of(w, struct drbd_peer_request, w);
2192 struct drbd_peer_device *peer_device = peer_req->peer_device;
2193 struct drbd_device *device = peer_device->device;
2194 sector_t sector = peer_req->i.sector;
2197 if (peer_req->flags & EE_SEND_WRITE_ACK) {
2198 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2199 pcmd = (device->state.conn >= C_SYNC_SOURCE &&
2200 device->state.conn <= C_PAUSED_SYNC_T &&
2201 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
2202 P_RS_WRITE_ACK : P_WRITE_ACK;
2203 err = drbd_send_ack(peer_device, pcmd, peer_req);
2204 if (pcmd == P_RS_WRITE_ACK)
2205 drbd_set_in_sync(device, sector, peer_req->i.size);
2207 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2208 /* we expect it to be marked out of sync anyways...
2209 * maybe assert this? */
2211 dec_unacked(device);
2214 /* we delete from the conflict detection hash _after_ we sent out the
2215 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
2216 if (peer_req->flags & EE_IN_INTERVAL_TREE) {
2217 spin_lock_irq(&device->resource->req_lock);
2218 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
2219 drbd_remove_epoch_entry_interval(device, peer_req);
2220 if (peer_req->flags & EE_RESTART_REQUESTS)
2221 restart_conflicting_writes(device, sector, peer_req->i.size);
2222 spin_unlock_irq(&device->resource->req_lock);
2224 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2226 drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2231 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2233 struct drbd_peer_request *peer_req =
2234 container_of(w, struct drbd_peer_request, w);
2235 struct drbd_peer_device *peer_device = peer_req->peer_device;
2238 err = drbd_send_ack(peer_device, ack, peer_req);
2239 dec_unacked(peer_device->device);
2244 static int e_send_superseded(struct drbd_work *w, int unused)
2246 return e_send_ack(w, P_SUPERSEDED);
2249 static int e_send_retry_write(struct drbd_work *w, int unused)
2251 struct drbd_peer_request *peer_req =
2252 container_of(w, struct drbd_peer_request, w);
2253 struct drbd_connection *connection = peer_req->peer_device->connection;
2255 return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2256 P_RETRY_WRITE : P_SUPERSEDED);
2259 static bool seq_greater(u32 a, u32 b)
2262 * We assume 32-bit wrap-around here.
2263 * For 24-bit wrap-around, we would have to shift:
2266 return (s32)a - (s32)b > 0;
2269 static u32 seq_max(u32 a, u32 b)
2271 return seq_greater(a, b) ? a : b;
2274 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2276 struct drbd_device *device = peer_device->device;
2277 unsigned int newest_peer_seq;
2279 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2280 spin_lock(&device->peer_seq_lock);
2281 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2282 device->peer_seq = newest_peer_seq;
2283 spin_unlock(&device->peer_seq_lock);
2284 /* wake up only if we actually changed device->peer_seq */
2285 if (peer_seq == newest_peer_seq)
2286 wake_up(&device->seq_wait);
2290 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2292 return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2295 /* maybe change sync_ee into interval trees as well? */
2296 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2298 struct drbd_peer_request *rs_req;
2301 spin_lock_irq(&device->resource->req_lock);
2302 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2303 if (overlaps(peer_req->i.sector, peer_req->i.size,
2304 rs_req->i.sector, rs_req->i.size)) {
2309 spin_unlock_irq(&device->resource->req_lock);
2314 /* Called from receive_Data.
2315 * Synchronize packets on sock with packets on msock.
2317 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2318 * packet traveling on msock, they are still processed in the order they have
2321 * Note: we don't care for Ack packets overtaking P_DATA packets.
2323 * In case packet_seq is larger than device->peer_seq number, there are
2324 * outstanding packets on the msock. We wait for them to arrive.
2325 * In case we are the logically next packet, we update device->peer_seq
2326 * ourselves. Correctly handles 32bit wrap around.
2328 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2329 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2330 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2331 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2333 * returns 0 if we may process the packet,
2334 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2335 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2337 struct drbd_device *device = peer_device->device;
2342 if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2345 spin_lock(&device->peer_seq_lock);
2347 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2348 device->peer_seq = seq_max(device->peer_seq, peer_seq);
2352 if (signal_pending(current)) {
2358 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2364 /* Only need to wait if two_primaries is enabled */
2365 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2366 spin_unlock(&device->peer_seq_lock);
2368 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2370 timeout = schedule_timeout(timeout);
2371 spin_lock(&device->peer_seq_lock);
2374 drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2378 spin_unlock(&device->peer_seq_lock);
2379 finish_wait(&device->seq_wait, &wait);
2383 /* see also bio_flags_to_wire()
2384 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2385 * flags and back. We may replicate to other kernel versions. */
2386 static unsigned long wire_flags_to_bio_flags(u32 dpf)
2388 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2389 (dpf & DP_FUA ? REQ_FUA : 0) |
2390 (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2393 static unsigned long wire_flags_to_bio_op(u32 dpf)
2395 if (dpf & DP_ZEROES)
2396 return REQ_OP_WRITE_ZEROES;
2397 if (dpf & DP_DISCARD)
2398 return REQ_OP_DISCARD;
2400 return REQ_OP_WRITE;
2403 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2406 struct drbd_interval *i;
2409 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2410 struct drbd_request *req;
2411 struct bio_and_error m;
2415 req = container_of(i, struct drbd_request, i);
2416 if (!(req->rq_state & RQ_POSTPONED))
2418 req->rq_state &= ~RQ_POSTPONED;
2419 __req_mod(req, NEG_ACKED, &m);
2420 spin_unlock_irq(&device->resource->req_lock);
2422 complete_master_bio(device, &m);
2423 spin_lock_irq(&device->resource->req_lock);
2428 static int handle_write_conflicts(struct drbd_device *device,
2429 struct drbd_peer_request *peer_req)
2431 struct drbd_connection *connection = peer_req->peer_device->connection;
2432 bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2433 sector_t sector = peer_req->i.sector;
2434 const unsigned int size = peer_req->i.size;
2435 struct drbd_interval *i;
2440 * Inserting the peer request into the write_requests tree will prevent
2441 * new conflicting local requests from being added.
2443 drbd_insert_interval(&device->write_requests, &peer_req->i);
2446 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2447 if (i == &peer_req->i)
2454 * Our peer has sent a conflicting remote request; this
2455 * should not happen in a two-node setup. Wait for the
2456 * earlier peer request to complete.
2458 err = drbd_wait_misc(device, i);
2464 equal = i->sector == sector && i->size == size;
2465 if (resolve_conflicts) {
2467 * If the peer request is fully contained within the
2468 * overlapping request, it can be considered overwritten
2469 * and thus superseded; otherwise, it will be retried
2470 * once all overlapping requests have completed.
2472 bool superseded = i->sector <= sector && i->sector +
2473 (i->size >> 9) >= sector + (size >> 9);
2476 drbd_alert(device, "Concurrent writes detected: "
2477 "local=%llus +%u, remote=%llus +%u, "
2478 "assuming %s came first\n",
2479 (unsigned long long)i->sector, i->size,
2480 (unsigned long long)sector, size,
2481 superseded ? "local" : "remote");
2483 peer_req->w.cb = superseded ? e_send_superseded :
2485 list_add_tail(&peer_req->w.list, &device->done_ee);
2486 queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2491 struct drbd_request *req =
2492 container_of(i, struct drbd_request, i);
2495 drbd_alert(device, "Concurrent writes detected: "
2496 "local=%llus +%u, remote=%llus +%u\n",
2497 (unsigned long long)i->sector, i->size,
2498 (unsigned long long)sector, size);
2500 if (req->rq_state & RQ_LOCAL_PENDING ||
2501 !(req->rq_state & RQ_POSTPONED)) {
2503 * Wait for the node with the discard flag to
2504 * decide if this request has been superseded
2505 * or needs to be retried.
2506 * Requests that have been superseded will
2507 * disappear from the write_requests tree.
2509 * In addition, wait for the conflicting
2510 * request to finish locally before submitting
2511 * the conflicting peer request.
2513 err = drbd_wait_misc(device, &req->i);
2515 _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2516 fail_postponed_requests(device, sector, size);
2522 * Remember to restart the conflicting requests after
2523 * the new peer request has completed.
2525 peer_req->flags |= EE_RESTART_REQUESTS;
2532 drbd_remove_epoch_entry_interval(device, peer_req);
2536 /* mirrored write */
2537 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2539 struct drbd_peer_device *peer_device;
2540 struct drbd_device *device;
2541 struct net_conf *nc;
2543 struct drbd_peer_request *peer_req;
2544 struct p_data *p = pi->data;
2545 u32 peer_seq = be32_to_cpu(p->seq_num);
2550 peer_device = conn_peer_device(connection, pi->vnr);
2553 device = peer_device->device;
2555 if (!get_ldev(device)) {
2558 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2559 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2560 atomic_inc(&connection->current_epoch->epoch_size);
2561 err2 = drbd_drain_block(peer_device, pi->size);
2568 * Corresponding put_ldev done either below (on various errors), or in
2569 * drbd_peer_request_endio, if we successfully submit the data at the
2570 * end of this function.
2573 sector = be64_to_cpu(p->sector);
2574 peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2580 peer_req->w.cb = e_end_block;
2581 peer_req->submit_jif = jiffies;
2582 peer_req->flags |= EE_APPLICATION;
2584 dp_flags = be32_to_cpu(p->dp_flags);
2585 op = wire_flags_to_bio_op(dp_flags);
2586 op_flags = wire_flags_to_bio_flags(dp_flags);
2587 if (pi->cmd == P_TRIM) {
2588 D_ASSERT(peer_device, peer_req->i.size > 0);
2589 D_ASSERT(peer_device, op == REQ_OP_DISCARD);
2590 D_ASSERT(peer_device, peer_req->pages == NULL);
2591 /* need to play safe: an older DRBD sender
2592 * may mean zero-out while sending P_TRIM. */
2593 if (0 == (connection->agreed_features & DRBD_FF_WZEROES))
2594 peer_req->flags |= EE_ZEROOUT;
2595 } else if (pi->cmd == P_ZEROES) {
2596 D_ASSERT(peer_device, peer_req->i.size > 0);
2597 D_ASSERT(peer_device, op == REQ_OP_WRITE_ZEROES);
2598 D_ASSERT(peer_device, peer_req->pages == NULL);
2599 /* Do (not) pass down BLKDEV_ZERO_NOUNMAP? */
2600 if (dp_flags & DP_DISCARD)
2601 peer_req->flags |= EE_TRIM;
2602 } else if (peer_req->pages == NULL) {
2603 D_ASSERT(device, peer_req->i.size == 0);
2604 D_ASSERT(device, dp_flags & DP_FLUSH);
2607 if (dp_flags & DP_MAY_SET_IN_SYNC)
2608 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2610 spin_lock(&connection->epoch_lock);
2611 peer_req->epoch = connection->current_epoch;
2612 atomic_inc(&peer_req->epoch->epoch_size);
2613 atomic_inc(&peer_req->epoch->active);
2614 spin_unlock(&connection->epoch_lock);
2617 nc = rcu_dereference(peer_device->connection->net_conf);
2618 tp = nc->two_primaries;
2619 if (peer_device->connection->agreed_pro_version < 100) {
2620 switch (nc->wire_protocol) {
2622 dp_flags |= DP_SEND_WRITE_ACK;
2625 dp_flags |= DP_SEND_RECEIVE_ACK;
2631 if (dp_flags & DP_SEND_WRITE_ACK) {
2632 peer_req->flags |= EE_SEND_WRITE_ACK;
2633 inc_unacked(device);
2634 /* corresponding dec_unacked() in e_end_block()
2635 * respective _drbd_clear_done_ee */
2638 if (dp_flags & DP_SEND_RECEIVE_ACK) {
2639 /* I really don't like it that the receiver thread
2640 * sends on the msock, but anyways */
2641 drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2645 /* two primaries implies protocol C */
2646 D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2647 peer_req->flags |= EE_IN_INTERVAL_TREE;
2648 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2650 goto out_interrupted;
2651 spin_lock_irq(&device->resource->req_lock);
2652 err = handle_write_conflicts(device, peer_req);
2654 spin_unlock_irq(&device->resource->req_lock);
2655 if (err == -ENOENT) {
2659 goto out_interrupted;
2662 update_peer_seq(peer_device, peer_seq);
2663 spin_lock_irq(&device->resource->req_lock);
2665 /* TRIM and is processed synchronously,
2666 * we wait for all pending requests, respectively wait for
2667 * active_ee to become empty in drbd_submit_peer_request();
2668 * better not add ourselves here. */
2669 if ((peer_req->flags & (EE_TRIM | EE_ZEROOUT)) == 0)
2670 list_add_tail(&peer_req->w.list, &device->active_ee);
2671 spin_unlock_irq(&device->resource->req_lock);
2673 if (device->state.conn == C_SYNC_TARGET)
2674 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2676 if (device->state.pdsk < D_INCONSISTENT) {
2677 /* In case we have the only disk of the cluster, */
2678 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2679 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2680 drbd_al_begin_io(device, &peer_req->i);
2681 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2684 err = drbd_submit_peer_request(device, peer_req, op, op_flags,
2689 /* don't care for the reason here */
2690 drbd_err(device, "submit failed, triggering re-connect\n");
2691 spin_lock_irq(&device->resource->req_lock);
2692 list_del(&peer_req->w.list);
2693 drbd_remove_epoch_entry_interval(device, peer_req);
2694 spin_unlock_irq(&device->resource->req_lock);
2695 if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2696 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2697 drbd_al_complete_io(device, &peer_req->i);
2701 drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
2703 drbd_free_peer_req(device, peer_req);
2707 /* We may throttle resync, if the lower device seems to be busy,
2708 * and current sync rate is above c_min_rate.
2710 * To decide whether or not the lower device is busy, we use a scheme similar
2711 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2712 * (more than 64 sectors) of activity we cannot account for with our own resync
2713 * activity, it obviously is "busy".
2715 * The current sync rate used here uses only the most recent two step marks,
2716 * to have a short time average so we can react faster.
2718 bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2719 bool throttle_if_app_is_waiting)
2721 struct lc_element *tmp;
2722 bool throttle = drbd_rs_c_min_rate_throttle(device);
2724 if (!throttle || throttle_if_app_is_waiting)
2727 spin_lock_irq(&device->al_lock);
2728 tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2730 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2731 if (test_bit(BME_PRIORITY, &bm_ext->flags))
2733 /* Do not slow down if app IO is already waiting for this extent,
2734 * and our progress is necessary for application IO to complete. */
2736 spin_unlock_irq(&device->al_lock);
2741 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2743 struct gendisk *disk = device->ldev->backing_bdev->bd_disk;
2744 unsigned long db, dt, dbdt;
2745 unsigned int c_min_rate;
2749 c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2752 /* feature disabled? */
2753 if (c_min_rate == 0)
2756 curr_events = (int)part_stat_read_accum(disk->part0, sectors) -
2757 atomic_read(&device->rs_sect_ev);
2759 if (atomic_read(&device->ap_actlog_cnt)
2760 || curr_events - device->rs_last_events > 64) {
2761 unsigned long rs_left;
2764 device->rs_last_events = curr_events;
2766 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2768 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2770 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2771 rs_left = device->ov_left;
2773 rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2775 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2778 db = device->rs_mark_left[i] - rs_left;
2779 dbdt = Bit2KB(db/dt);
2781 if (dbdt > c_min_rate)
2787 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2789 struct drbd_peer_device *peer_device;
2790 struct drbd_device *device;
2793 struct drbd_peer_request *peer_req;
2794 struct digest_info *di = NULL;
2796 unsigned int fault_type;
2797 struct p_block_req *p = pi->data;
2799 peer_device = conn_peer_device(connection, pi->vnr);
2802 device = peer_device->device;
2803 capacity = get_capacity(device->vdisk);
2805 sector = be64_to_cpu(p->sector);
2806 size = be32_to_cpu(p->blksize);
2808 if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2809 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2810 (unsigned long long)sector, size);
2813 if (sector + (size>>9) > capacity) {
2814 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2815 (unsigned long long)sector, size);
2819 if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2822 case P_DATA_REQUEST:
2823 drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2826 case P_RS_DATA_REQUEST:
2827 case P_CSUM_RS_REQUEST:
2829 drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2833 dec_rs_pending(device);
2834 drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2839 if (verb && __ratelimit(&drbd_ratelimit_state))
2840 drbd_err(device, "Can not satisfy peer's read request, "
2841 "no local data.\n");
2843 /* drain possibly payload */
2844 return drbd_drain_block(peer_device, pi->size);
2847 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2848 * "criss-cross" setup, that might cause write-out on some other DRBD,
2849 * which in turn might block on the other node at this very place. */
2850 peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2858 case P_DATA_REQUEST:
2859 peer_req->w.cb = w_e_end_data_req;
2860 fault_type = DRBD_FAULT_DT_RD;
2861 /* application IO, don't drbd_rs_begin_io */
2862 peer_req->flags |= EE_APPLICATION;
2866 /* If at some point in the future we have a smart way to
2867 find out if this data block is completely deallocated,
2868 then we would do something smarter here than reading
2870 peer_req->flags |= EE_RS_THIN_REQ;
2872 case P_RS_DATA_REQUEST:
2873 peer_req->w.cb = w_e_end_rsdata_req;
2874 fault_type = DRBD_FAULT_RS_RD;
2875 /* used in the sector offset progress display */
2876 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2880 case P_CSUM_RS_REQUEST:
2881 fault_type = DRBD_FAULT_RS_RD;
2882 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2886 di->digest_size = pi->size;
2887 di->digest = (((char *)di)+sizeof(struct digest_info));
2889 peer_req->digest = di;
2890 peer_req->flags |= EE_HAS_DIGEST;
2892 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2895 if (pi->cmd == P_CSUM_RS_REQUEST) {
2896 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2897 peer_req->w.cb = w_e_end_csum_rs_req;
2898 /* used in the sector offset progress display */
2899 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2900 /* remember to report stats in drbd_resync_finished */
2901 device->use_csums = true;
2902 } else if (pi->cmd == P_OV_REPLY) {
2903 /* track progress, we may need to throttle */
2904 atomic_add(size >> 9, &device->rs_sect_in);
2905 peer_req->w.cb = w_e_end_ov_reply;
2906 dec_rs_pending(device);
2907 /* drbd_rs_begin_io done when we sent this request,
2908 * but accounting still needs to be done. */
2909 goto submit_for_resync;
2914 if (device->ov_start_sector == ~(sector_t)0 &&
2915 peer_device->connection->agreed_pro_version >= 90) {
2916 unsigned long now = jiffies;
2918 device->ov_start_sector = sector;
2919 device->ov_position = sector;
2920 device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2921 device->rs_total = device->ov_left;
2922 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2923 device->rs_mark_left[i] = device->ov_left;
2924 device->rs_mark_time[i] = now;
2926 drbd_info(device, "Online Verify start sector: %llu\n",
2927 (unsigned long long)sector);
2929 peer_req->w.cb = w_e_end_ov_req;
2930 fault_type = DRBD_FAULT_RS_RD;
2937 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2938 * wrt the receiver, but it is not as straightforward as it may seem.
2939 * Various places in the resync start and stop logic assume resync
2940 * requests are processed in order, requeuing this on the worker thread
2941 * introduces a bunch of new code for synchronization between threads.
2943 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2944 * "forever", throttling after drbd_rs_begin_io will lock that extent
2945 * for application writes for the same time. For now, just throttle
2946 * here, where the rest of the code expects the receiver to sleep for
2950 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2951 * this defers syncer requests for some time, before letting at least
2952 * on request through. The resync controller on the receiving side
2953 * will adapt to the incoming rate accordingly.
2955 * We cannot throttle here if remote is Primary/SyncTarget:
2956 * we would also throttle its application reads.
2957 * In that case, throttling is done on the SyncTarget only.
2960 /* Even though this may be a resync request, we do add to "read_ee";
2961 * "sync_ee" is only used for resync WRITEs.
2962 * Add to list early, so debugfs can find this request
2963 * even if we have to sleep below. */
2964 spin_lock_irq(&device->resource->req_lock);
2965 list_add_tail(&peer_req->w.list, &device->read_ee);
2966 spin_unlock_irq(&device->resource->req_lock);
2968 update_receiver_timing_details(connection, drbd_rs_should_slow_down);
2969 if (device->state.peer != R_PRIMARY
2970 && drbd_rs_should_slow_down(device, sector, false))
2971 schedule_timeout_uninterruptible(HZ/10);
2972 update_receiver_timing_details(connection, drbd_rs_begin_io);
2973 if (drbd_rs_begin_io(device, sector))
2977 atomic_add(size >> 9, &device->rs_sect_ev);
2980 update_receiver_timing_details(connection, drbd_submit_peer_request);
2981 inc_unacked(device);
2982 if (drbd_submit_peer_request(device, peer_req, REQ_OP_READ, 0,
2986 /* don't care for the reason here */
2987 drbd_err(device, "submit failed, triggering re-connect\n");
2990 spin_lock_irq(&device->resource->req_lock);
2991 list_del(&peer_req->w.list);
2992 spin_unlock_irq(&device->resource->req_lock);
2993 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2996 drbd_free_peer_req(device, peer_req);
3001 * drbd_asb_recover_0p - Recover after split-brain with no remaining primaries
3003 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
3005 struct drbd_device *device = peer_device->device;
3006 int self, peer, rv = -100;
3007 unsigned long ch_self, ch_peer;
3008 enum drbd_after_sb_p after_sb_0p;
3010 self = device->ldev->md.uuid[UI_BITMAP] & 1;
3011 peer = device->p_uuid[UI_BITMAP] & 1;
3013 ch_peer = device->p_uuid[UI_SIZE];
3014 ch_self = device->comm_bm_set;
3017 after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
3019 switch (after_sb_0p) {
3021 case ASB_DISCARD_SECONDARY:
3022 case ASB_CALL_HELPER:
3024 drbd_err(device, "Configuration error.\n");
3026 case ASB_DISCONNECT:
3028 case ASB_DISCARD_YOUNGER_PRI:
3029 if (self == 0 && peer == 1) {
3033 if (self == 1 && peer == 0) {
3037 fallthrough; /* to one of the other strategies */
3038 case ASB_DISCARD_OLDER_PRI:
3039 if (self == 0 && peer == 1) {
3043 if (self == 1 && peer == 0) {
3047 /* Else fall through to one of the other strategies... */
3048 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
3049 "Using discard-least-changes instead\n");
3051 case ASB_DISCARD_ZERO_CHG:
3052 if (ch_peer == 0 && ch_self == 0) {
3053 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3057 if (ch_peer == 0) { rv = 1; break; }
3058 if (ch_self == 0) { rv = -1; break; }
3060 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
3063 case ASB_DISCARD_LEAST_CHG:
3064 if (ch_self < ch_peer)
3066 else if (ch_self > ch_peer)
3068 else /* ( ch_self == ch_peer ) */
3069 /* Well, then use something else. */
3070 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3073 case ASB_DISCARD_LOCAL:
3076 case ASB_DISCARD_REMOTE:
3084 * drbd_asb_recover_1p - Recover after split-brain with one remaining primary
3086 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
3088 struct drbd_device *device = peer_device->device;
3090 enum drbd_after_sb_p after_sb_1p;
3093 after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
3095 switch (after_sb_1p) {
3096 case ASB_DISCARD_YOUNGER_PRI:
3097 case ASB_DISCARD_OLDER_PRI:
3098 case ASB_DISCARD_LEAST_CHG:
3099 case ASB_DISCARD_LOCAL:
3100 case ASB_DISCARD_REMOTE:
3101 case ASB_DISCARD_ZERO_CHG:
3102 drbd_err(device, "Configuration error.\n");
3104 case ASB_DISCONNECT:
3107 hg = drbd_asb_recover_0p(peer_device);
3108 if (hg == -1 && device->state.role == R_SECONDARY)
3110 if (hg == 1 && device->state.role == R_PRIMARY)
3114 rv = drbd_asb_recover_0p(peer_device);
3116 case ASB_DISCARD_SECONDARY:
3117 return device->state.role == R_PRIMARY ? 1 : -1;
3118 case ASB_CALL_HELPER:
3119 hg = drbd_asb_recover_0p(peer_device);
3120 if (hg == -1 && device->state.role == R_PRIMARY) {
3121 enum drbd_state_rv rv2;
3123 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3124 * we might be here in C_WF_REPORT_PARAMS which is transient.
3125 * we do not need to wait for the after state change work either. */
3126 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3127 if (rv2 != SS_SUCCESS) {
3128 drbd_khelper(device, "pri-lost-after-sb");
3130 drbd_warn(device, "Successfully gave up primary role.\n");
3141 * drbd_asb_recover_2p - Recover after split-brain with two remaining primaries
3143 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
3145 struct drbd_device *device = peer_device->device;
3147 enum drbd_after_sb_p after_sb_2p;
3150 after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
3152 switch (after_sb_2p) {
3153 case ASB_DISCARD_YOUNGER_PRI:
3154 case ASB_DISCARD_OLDER_PRI:
3155 case ASB_DISCARD_LEAST_CHG:
3156 case ASB_DISCARD_LOCAL:
3157 case ASB_DISCARD_REMOTE:
3159 case ASB_DISCARD_SECONDARY:
3160 case ASB_DISCARD_ZERO_CHG:
3161 drbd_err(device, "Configuration error.\n");
3164 rv = drbd_asb_recover_0p(peer_device);
3166 case ASB_DISCONNECT:
3168 case ASB_CALL_HELPER:
3169 hg = drbd_asb_recover_0p(peer_device);
3171 enum drbd_state_rv rv2;
3173 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3174 * we might be here in C_WF_REPORT_PARAMS which is transient.
3175 * we do not need to wait for the after state change work either. */
3176 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3177 if (rv2 != SS_SUCCESS) {
3178 drbd_khelper(device, "pri-lost-after-sb");
3180 drbd_warn(device, "Successfully gave up primary role.\n");
3190 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
3191 u64 bits, u64 flags)
3194 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
3197 drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
3199 (unsigned long long)uuid[UI_CURRENT],
3200 (unsigned long long)uuid[UI_BITMAP],
3201 (unsigned long long)uuid[UI_HISTORY_START],
3202 (unsigned long long)uuid[UI_HISTORY_END],
3203 (unsigned long long)bits,
3204 (unsigned long long)flags);
3208 100 after split brain try auto recover
3209 2 C_SYNC_SOURCE set BitMap
3210 1 C_SYNC_SOURCE use BitMap
3212 -1 C_SYNC_TARGET use BitMap
3213 -2 C_SYNC_TARGET set BitMap
3214 -100 after split brain, disconnect
3215 -1000 unrelated data
3216 -1091 requires proto 91
3217 -1096 requires proto 96
3220 static int drbd_uuid_compare(struct drbd_device *const device, enum drbd_role const peer_role, int *rule_nr) __must_hold(local)
3222 struct drbd_peer_device *const peer_device = first_peer_device(device);
3223 struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
3227 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3228 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3231 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
3235 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
3236 peer != UUID_JUST_CREATED)
3240 if (self != UUID_JUST_CREATED &&
3241 (peer == UUID_JUST_CREATED || peer == (u64)0))
3245 int rct, dc; /* roles at crash time */
3247 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3249 if (connection->agreed_pro_version < 91)
3252 if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3253 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3254 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3255 drbd_uuid_move_history(device);
3256 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3257 device->ldev->md.uuid[UI_BITMAP] = 0;
3259 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3260 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3263 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3270 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3272 if (connection->agreed_pro_version < 91)
3275 if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3276 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3277 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3279 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3280 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3281 device->p_uuid[UI_BITMAP] = 0UL;
3283 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3286 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3293 /* Common power [off|failure] */
3294 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3295 (device->p_uuid[UI_FLAGS] & 2);
3296 /* lowest bit is set when we were primary,
3297 * next bit (weight 2) is set when peer was primary */
3300 /* Neither has the "crashed primary" flag set,
3301 * only a replication link hickup. */
3305 /* Current UUID equal and no bitmap uuid; does not necessarily
3306 * mean this was a "simultaneous hard crash", maybe IO was
3307 * frozen, so no UUID-bump happened.
3308 * This is a protocol change, overload DRBD_FF_WSAME as flag
3309 * for "new-enough" peer DRBD version. */
3310 if (device->state.role == R_PRIMARY || peer_role == R_PRIMARY) {
3312 if (!(connection->agreed_features & DRBD_FF_WSAME)) {
3313 drbd_warn(peer_device, "Equivalent unrotated UUIDs, but current primary present.\n");
3314 return -(0x10000 | PRO_VERSION_MAX | (DRBD_FF_WSAME << 8));
3316 if (device->state.role == R_PRIMARY && peer_role == R_PRIMARY) {
3317 /* At least one has the "crashed primary" bit set,
3318 * both are primary now, but neither has rotated its UUIDs?
3319 * "Can not happen." */
3320 drbd_err(peer_device, "Equivalent unrotated UUIDs, but both are primary. Can not resolve this.\n");
3323 if (device->state.role == R_PRIMARY)
3328 /* Both are secondary.
3329 * Really looks like recovery from simultaneous hard crash.
3330 * Check which had been primary before, and arbitrate. */
3332 case 0: /* !self_pri && !peer_pri */ return 0; /* already handled */
3333 case 1: /* self_pri && !peer_pri */ return 1;
3334 case 2: /* !self_pri && peer_pri */ return -1;
3335 case 3: /* self_pri && peer_pri */
3336 dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3342 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3347 peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3349 if (connection->agreed_pro_version < 96 ?
3350 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3351 (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3352 peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3353 /* The last P_SYNC_UUID did not get though. Undo the last start of
3354 resync as sync source modifications of the peer's UUIDs. */
3356 if (connection->agreed_pro_version < 91)
3359 device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3360 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3362 drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3363 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3370 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3371 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3372 peer = device->p_uuid[i] & ~((u64)1);
3378 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3379 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3384 self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3386 if (connection->agreed_pro_version < 96 ?
3387 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3388 (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3389 self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3390 /* The last P_SYNC_UUID did not get though. Undo the last start of
3391 resync as sync source modifications of our UUIDs. */
3393 if (connection->agreed_pro_version < 91)
3396 __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3397 __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3399 drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3400 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3401 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3409 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3410 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3411 self = device->ldev->md.uuid[i] & ~((u64)1);
3417 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3418 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3419 if (self == peer && self != ((u64)0))
3423 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3424 self = device->ldev->md.uuid[i] & ~((u64)1);
3425 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3426 peer = device->p_uuid[j] & ~((u64)1);
3435 /* drbd_sync_handshake() returns the new conn state on success, or
3436 CONN_MASK (-1) on failure.
3438 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3439 enum drbd_role peer_role,
3440 enum drbd_disk_state peer_disk) __must_hold(local)
3442 struct drbd_device *device = peer_device->device;
3443 enum drbd_conns rv = C_MASK;
3444 enum drbd_disk_state mydisk;
3445 struct net_conf *nc;
3446 int hg, rule_nr, rr_conflict, tentative, always_asbp;
3448 mydisk = device->state.disk;
3449 if (mydisk == D_NEGOTIATING)
3450 mydisk = device->new_state_tmp.disk;
3452 drbd_info(device, "drbd_sync_handshake:\n");
3454 spin_lock_irq(&device->ldev->md.uuid_lock);
3455 drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3456 drbd_uuid_dump(device, "peer", device->p_uuid,
3457 device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3459 hg = drbd_uuid_compare(device, peer_role, &rule_nr);
3460 spin_unlock_irq(&device->ldev->md.uuid_lock);
3462 drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3465 drbd_alert(device, "Unrelated data, aborting!\n");
3468 if (hg < -0x10000) {
3472 fflags = (hg >> 8) & 0xff;
3473 drbd_alert(device, "To resolve this both sides have to support at least protocol %d and feature flags 0x%x\n",
3478 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3482 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3483 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
3484 int f = (hg == -100) || abs(hg) == 2;
3485 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3488 drbd_info(device, "Becoming sync %s due to disk states.\n",
3489 hg > 0 ? "source" : "target");
3493 drbd_khelper(device, "initial-split-brain");
3496 nc = rcu_dereference(peer_device->connection->net_conf);
3497 always_asbp = nc->always_asbp;
3498 rr_conflict = nc->rr_conflict;
3499 tentative = nc->tentative;
3502 if (hg == 100 || (hg == -100 && always_asbp)) {
3503 int pcount = (device->state.role == R_PRIMARY)
3504 + (peer_role == R_PRIMARY);
3505 int forced = (hg == -100);
3509 hg = drbd_asb_recover_0p(peer_device);
3512 hg = drbd_asb_recover_1p(peer_device);
3515 hg = drbd_asb_recover_2p(peer_device);
3518 if (abs(hg) < 100) {
3519 drbd_warn(device, "Split-Brain detected, %d primaries, "
3520 "automatically solved. Sync from %s node\n",
3521 pcount, (hg < 0) ? "peer" : "this");
3523 drbd_warn(device, "Doing a full sync, since"
3524 " UUIDs where ambiguous.\n");
3531 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3533 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3537 drbd_warn(device, "Split-Brain detected, manually solved. "
3538 "Sync from %s node\n",
3539 (hg < 0) ? "peer" : "this");
3543 /* FIXME this log message is not correct if we end up here
3544 * after an attempted attach on a diskless node.
3545 * We just refuse to attach -- well, we drop the "connection"
3546 * to that disk, in a way... */
3547 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3548 drbd_khelper(device, "split-brain");
3552 if (hg > 0 && mydisk <= D_INCONSISTENT) {
3553 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3557 if (hg < 0 && /* by intention we do not use mydisk here. */
3558 device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3559 switch (rr_conflict) {
3560 case ASB_CALL_HELPER:
3561 drbd_khelper(device, "pri-lost");
3563 case ASB_DISCONNECT:
3564 drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3567 drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3572 if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3574 drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3576 drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3577 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3578 abs(hg) >= 2 ? "full" : "bit-map based");
3583 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3584 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3585 BM_LOCKED_SET_ALLOWED))
3589 if (hg > 0) { /* become sync source. */
3591 } else if (hg < 0) { /* become sync target */
3595 if (drbd_bm_total_weight(device)) {
3596 drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3597 drbd_bm_total_weight(device));
3604 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3606 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3607 if (peer == ASB_DISCARD_REMOTE)
3608 return ASB_DISCARD_LOCAL;
3610 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3611 if (peer == ASB_DISCARD_LOCAL)
3612 return ASB_DISCARD_REMOTE;
3614 /* everything else is valid if they are equal on both sides. */
3618 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3620 struct p_protocol *p = pi->data;
3621 enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3622 int p_proto, p_discard_my_data, p_two_primaries, cf;
3623 struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3624 char integrity_alg[SHARED_SECRET_MAX] = "";
3625 struct crypto_shash *peer_integrity_tfm = NULL;
3626 void *int_dig_in = NULL, *int_dig_vv = NULL;
3628 p_proto = be32_to_cpu(p->protocol);
3629 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
3630 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
3631 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
3632 p_two_primaries = be32_to_cpu(p->two_primaries);
3633 cf = be32_to_cpu(p->conn_flags);
3634 p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3636 if (connection->agreed_pro_version >= 87) {
3639 if (pi->size > sizeof(integrity_alg))
3641 err = drbd_recv_all(connection, integrity_alg, pi->size);
3644 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3647 if (pi->cmd != P_PROTOCOL_UPDATE) {
3648 clear_bit(CONN_DRY_RUN, &connection->flags);
3650 if (cf & CF_DRY_RUN)
3651 set_bit(CONN_DRY_RUN, &connection->flags);
3654 nc = rcu_dereference(connection->net_conf);
3656 if (p_proto != nc->wire_protocol) {
3657 drbd_err(connection, "incompatible %s settings\n", "protocol");
3658 goto disconnect_rcu_unlock;
3661 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3662 drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3663 goto disconnect_rcu_unlock;
3666 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3667 drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3668 goto disconnect_rcu_unlock;
3671 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3672 drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3673 goto disconnect_rcu_unlock;
3676 if (p_discard_my_data && nc->discard_my_data) {
3677 drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3678 goto disconnect_rcu_unlock;
3681 if (p_two_primaries != nc->two_primaries) {
3682 drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3683 goto disconnect_rcu_unlock;
3686 if (strcmp(integrity_alg, nc->integrity_alg)) {
3687 drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3688 goto disconnect_rcu_unlock;
3694 if (integrity_alg[0]) {
3698 * We can only change the peer data integrity algorithm
3699 * here. Changing our own data integrity algorithm
3700 * requires that we send a P_PROTOCOL_UPDATE packet at
3701 * the same time; otherwise, the peer has no way to
3702 * tell between which packets the algorithm should
3706 peer_integrity_tfm = crypto_alloc_shash(integrity_alg, 0, 0);
3707 if (IS_ERR(peer_integrity_tfm)) {
3708 peer_integrity_tfm = NULL;
3709 drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3714 hash_size = crypto_shash_digestsize(peer_integrity_tfm);
3715 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3716 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3717 if (!(int_dig_in && int_dig_vv)) {
3718 drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3723 new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3727 mutex_lock(&connection->data.mutex);
3728 mutex_lock(&connection->resource->conf_update);
3729 old_net_conf = connection->net_conf;
3730 *new_net_conf = *old_net_conf;
3732 new_net_conf->wire_protocol = p_proto;
3733 new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3734 new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3735 new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3736 new_net_conf->two_primaries = p_two_primaries;
3738 rcu_assign_pointer(connection->net_conf, new_net_conf);
3739 mutex_unlock(&connection->resource->conf_update);
3740 mutex_unlock(&connection->data.mutex);
3742 crypto_free_shash(connection->peer_integrity_tfm);
3743 kfree(connection->int_dig_in);
3744 kfree(connection->int_dig_vv);
3745 connection->peer_integrity_tfm = peer_integrity_tfm;
3746 connection->int_dig_in = int_dig_in;
3747 connection->int_dig_vv = int_dig_vv;
3749 if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3750 drbd_info(connection, "peer data-integrity-alg: %s\n",
3751 integrity_alg[0] ? integrity_alg : "(none)");
3754 kfree(old_net_conf);
3757 disconnect_rcu_unlock:
3760 crypto_free_shash(peer_integrity_tfm);
3763 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3768 * input: alg name, feature name
3769 * return: NULL (alg name was "")
3770 * ERR_PTR(error) if something goes wrong
3771 * or the crypto hash ptr, if it worked out ok. */
3772 static struct crypto_shash *drbd_crypto_alloc_digest_safe(
3773 const struct drbd_device *device,
3774 const char *alg, const char *name)
3776 struct crypto_shash *tfm;
3781 tfm = crypto_alloc_shash(alg, 0, 0);
3783 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3784 alg, name, PTR_ERR(tfm));
3790 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3792 void *buffer = connection->data.rbuf;
3793 int size = pi->size;
3796 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3797 s = drbd_recv(connection, buffer, s);
3811 * config_unknown_volume - device configuration command for unknown volume
3813 * When a device is added to an existing connection, the node on which the
3814 * device is added first will send configuration commands to its peer but the
3815 * peer will not know about the device yet. It will warn and ignore these
3816 * commands. Once the device is added on the second node, the second node will
3817 * send the same device configuration commands, but in the other direction.
3819 * (We can also end up here if drbd is misconfigured.)
3821 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3823 drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3824 cmdname(pi->cmd), pi->vnr);
3825 return ignore_remaining_packet(connection, pi);
3828 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3830 struct drbd_peer_device *peer_device;
3831 struct drbd_device *device;
3832 struct p_rs_param_95 *p;
3833 unsigned int header_size, data_size, exp_max_sz;
3834 struct crypto_shash *verify_tfm = NULL;
3835 struct crypto_shash *csums_tfm = NULL;
3836 struct net_conf *old_net_conf, *new_net_conf = NULL;
3837 struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3838 const int apv = connection->agreed_pro_version;
3839 struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3840 unsigned int fifo_size = 0;
3843 peer_device = conn_peer_device(connection, pi->vnr);
3845 return config_unknown_volume(connection, pi);
3846 device = peer_device->device;
3848 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
3849 : apv == 88 ? sizeof(struct p_rs_param)
3851 : apv <= 94 ? sizeof(struct p_rs_param_89)
3852 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3854 if (pi->size > exp_max_sz) {
3855 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3856 pi->size, exp_max_sz);
3861 header_size = sizeof(struct p_rs_param);
3862 data_size = pi->size - header_size;
3863 } else if (apv <= 94) {
3864 header_size = sizeof(struct p_rs_param_89);
3865 data_size = pi->size - header_size;
3866 D_ASSERT(device, data_size == 0);
3868 header_size = sizeof(struct p_rs_param_95);
3869 data_size = pi->size - header_size;
3870 D_ASSERT(device, data_size == 0);
3873 /* initialize verify_alg and csums_alg */
3875 BUILD_BUG_ON(sizeof(p->algs) != 2 * SHARED_SECRET_MAX);
3876 memset(&p->algs, 0, sizeof(p->algs));
3878 err = drbd_recv_all(peer_device->connection, p, header_size);
3882 mutex_lock(&connection->resource->conf_update);
3883 old_net_conf = peer_device->connection->net_conf;
3884 if (get_ldev(device)) {
3885 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3886 if (!new_disk_conf) {
3888 mutex_unlock(&connection->resource->conf_update);
3889 drbd_err(device, "Allocation of new disk_conf failed\n");
3893 old_disk_conf = device->ldev->disk_conf;
3894 *new_disk_conf = *old_disk_conf;
3896 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3901 if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3902 drbd_err(device, "verify-alg of wrong size, "
3903 "peer wants %u, accepting only up to %u byte\n",
3904 data_size, SHARED_SECRET_MAX);
3908 err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3911 /* we expect NUL terminated string */
3912 /* but just in case someone tries to be evil */
3913 D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3914 p->verify_alg[data_size-1] = 0;
3916 } else /* apv >= 89 */ {
3917 /* we still expect NUL terminated strings */
3918 /* but just in case someone tries to be evil */
3919 D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3920 D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3921 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3922 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3925 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3926 if (device->state.conn == C_WF_REPORT_PARAMS) {
3927 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3928 old_net_conf->verify_alg, p->verify_alg);
3931 verify_tfm = drbd_crypto_alloc_digest_safe(device,
3932 p->verify_alg, "verify-alg");
3933 if (IS_ERR(verify_tfm)) {
3939 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3940 if (device->state.conn == C_WF_REPORT_PARAMS) {
3941 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3942 old_net_conf->csums_alg, p->csums_alg);
3945 csums_tfm = drbd_crypto_alloc_digest_safe(device,
3946 p->csums_alg, "csums-alg");
3947 if (IS_ERR(csums_tfm)) {
3953 if (apv > 94 && new_disk_conf) {
3954 new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3955 new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3956 new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3957 new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3959 fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3960 if (fifo_size != device->rs_plan_s->size) {
3961 new_plan = fifo_alloc(fifo_size);
3963 drbd_err(device, "kmalloc of fifo_buffer failed");
3970 if (verify_tfm || csums_tfm) {
3971 new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3975 *new_net_conf = *old_net_conf;
3978 strcpy(new_net_conf->verify_alg, p->verify_alg);
3979 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3980 crypto_free_shash(peer_device->connection->verify_tfm);
3981 peer_device->connection->verify_tfm = verify_tfm;
3982 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3985 strcpy(new_net_conf->csums_alg, p->csums_alg);
3986 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3987 crypto_free_shash(peer_device->connection->csums_tfm);
3988 peer_device->connection->csums_tfm = csums_tfm;
3989 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
3991 rcu_assign_pointer(connection->net_conf, new_net_conf);
3995 if (new_disk_conf) {
3996 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4001 old_plan = device->rs_plan_s;
4002 rcu_assign_pointer(device->rs_plan_s, new_plan);
4005 mutex_unlock(&connection->resource->conf_update);
4008 kfree(old_net_conf);
4009 kfree(old_disk_conf);
4015 if (new_disk_conf) {
4017 kfree(new_disk_conf);
4019 mutex_unlock(&connection->resource->conf_update);
4024 if (new_disk_conf) {
4026 kfree(new_disk_conf);
4028 mutex_unlock(&connection->resource->conf_update);
4029 /* just for completeness: actually not needed,
4030 * as this is not reached if csums_tfm was ok. */
4031 crypto_free_shash(csums_tfm);
4032 /* but free the verify_tfm again, if csums_tfm did not work out */
4033 crypto_free_shash(verify_tfm);
4034 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4038 /* warn if the arguments differ by more than 12.5% */
4039 static void warn_if_differ_considerably(struct drbd_device *device,
4040 const char *s, sector_t a, sector_t b)
4043 if (a == 0 || b == 0)
4045 d = (a > b) ? (a - b) : (b - a);
4046 if (d > (a>>3) || d > (b>>3))
4047 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
4048 (unsigned long long)a, (unsigned long long)b);
4051 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
4053 struct drbd_peer_device *peer_device;
4054 struct drbd_device *device;
4055 struct p_sizes *p = pi->data;
4056 struct o_qlim *o = (connection->agreed_features & DRBD_FF_WSAME) ? p->qlim : NULL;
4057 enum determine_dev_size dd = DS_UNCHANGED;
4058 sector_t p_size, p_usize, p_csize, my_usize;
4059 sector_t new_size, cur_size;
4060 int ldsc = 0; /* local disk size changed */
4061 enum dds_flags ddsf;
4063 peer_device = conn_peer_device(connection, pi->vnr);
4065 return config_unknown_volume(connection, pi);
4066 device = peer_device->device;
4067 cur_size = get_capacity(device->vdisk);
4069 p_size = be64_to_cpu(p->d_size);
4070 p_usize = be64_to_cpu(p->u_size);
4071 p_csize = be64_to_cpu(p->c_size);
4073 /* just store the peer's disk size for now.
4074 * we still need to figure out whether we accept that. */
4075 device->p_size = p_size;
4077 if (get_ldev(device)) {
4079 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
4082 warn_if_differ_considerably(device, "lower level device sizes",
4083 p_size, drbd_get_max_capacity(device->ldev));
4084 warn_if_differ_considerably(device, "user requested size",
4087 /* if this is the first connect, or an otherwise expected
4088 * param exchange, choose the minimum */
4089 if (device->state.conn == C_WF_REPORT_PARAMS)
4090 p_usize = min_not_zero(my_usize, p_usize);
4092 /* Never shrink a device with usable data during connect,
4093 * or "attach" on the peer.
4094 * But allow online shrinking if we are connected. */
4095 new_size = drbd_new_dev_size(device, device->ldev, p_usize, 0);
4096 if (new_size < cur_size &&
4097 device->state.disk >= D_OUTDATED &&
4098 (device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS)) {
4099 drbd_err(device, "The peer's disk size is too small! (%llu < %llu sectors)\n",
4100 (unsigned long long)new_size, (unsigned long long)cur_size);
4101 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4106 if (my_usize != p_usize) {
4107 struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
4109 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
4110 if (!new_disk_conf) {
4115 mutex_lock(&connection->resource->conf_update);
4116 old_disk_conf = device->ldev->disk_conf;
4117 *new_disk_conf = *old_disk_conf;
4118 new_disk_conf->disk_size = p_usize;
4120 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4121 mutex_unlock(&connection->resource->conf_update);
4123 kfree(old_disk_conf);
4125 drbd_info(device, "Peer sets u_size to %lu sectors (old: %lu)\n",
4126 (unsigned long)p_usize, (unsigned long)my_usize);
4132 device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
4133 /* Leave drbd_reconsider_queue_parameters() before drbd_determine_dev_size().
4134 In case we cleared the QUEUE_FLAG_DISCARD from our queue in
4135 drbd_reconsider_queue_parameters(), we can be sure that after
4136 drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
4138 ddsf = be16_to_cpu(p->dds_flags);
4139 if (get_ldev(device)) {
4140 drbd_reconsider_queue_parameters(device, device->ldev, o);
4141 dd = drbd_determine_dev_size(device, ddsf, NULL);
4145 drbd_md_sync(device);
4148 * I am diskless, need to accept the peer's *current* size.
4149 * I must NOT accept the peers backing disk size,
4150 * it may have been larger than mine all along...
4152 * At this point, the peer knows more about my disk, or at
4153 * least about what we last agreed upon, than myself.
4154 * So if his c_size is less than his d_size, the most likely
4155 * reason is that *my* d_size was smaller last time we checked.
4157 * However, if he sends a zero current size,
4158 * take his (user-capped or) backing disk size anyways.
4160 * Unless of course he does not have a disk himself.
4161 * In which case we ignore this completely.
4163 sector_t new_size = p_csize ?: p_usize ?: p_size;
4164 drbd_reconsider_queue_parameters(device, NULL, o);
4165 if (new_size == 0) {
4166 /* Ignore, peer does not know nothing. */
4167 } else if (new_size == cur_size) {
4169 } else if (cur_size != 0 && p_size == 0) {
4170 drbd_warn(device, "Ignored diskless peer device size (peer:%llu != me:%llu sectors)!\n",
4171 (unsigned long long)new_size, (unsigned long long)cur_size);
4172 } else if (new_size < cur_size && device->state.role == R_PRIMARY) {
4173 drbd_err(device, "The peer's device size is too small! (%llu < %llu sectors); demote me first!\n",
4174 (unsigned long long)new_size, (unsigned long long)cur_size);
4175 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4178 /* I believe the peer, if
4179 * - I don't have a current size myself
4180 * - we agree on the size anyways
4181 * - I do have a current size, am Secondary,
4182 * and he has the only disk
4183 * - I do have a current size, am Primary,
4184 * and he has the only disk,
4185 * which is larger than my current size
4187 drbd_set_my_capacity(device, new_size);
4191 if (get_ldev(device)) {
4192 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
4193 device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
4200 if (device->state.conn > C_WF_REPORT_PARAMS) {
4201 if (be64_to_cpu(p->c_size) != get_capacity(device->vdisk) ||
4203 /* we have different sizes, probably peer
4204 * needs to know my new size... */
4205 drbd_send_sizes(peer_device, 0, ddsf);
4207 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
4208 (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
4209 if (device->state.pdsk >= D_INCONSISTENT &&
4210 device->state.disk >= D_INCONSISTENT) {
4211 if (ddsf & DDSF_NO_RESYNC)
4212 drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
4214 resync_after_online_grow(device);
4216 set_bit(RESYNC_AFTER_NEG, &device->flags);
4223 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
4225 struct drbd_peer_device *peer_device;
4226 struct drbd_device *device;
4227 struct p_uuids *p = pi->data;
4229 int i, updated_uuids = 0;
4231 peer_device = conn_peer_device(connection, pi->vnr);
4233 return config_unknown_volume(connection, pi);
4234 device = peer_device->device;
4236 p_uuid = kmalloc_array(UI_EXTENDED_SIZE, sizeof(*p_uuid), GFP_NOIO);
4240 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
4241 p_uuid[i] = be64_to_cpu(p->uuid[i]);
4243 kfree(device->p_uuid);
4244 device->p_uuid = p_uuid;
4246 if ((device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS) &&
4247 device->state.disk < D_INCONSISTENT &&
4248 device->state.role == R_PRIMARY &&
4249 (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
4250 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
4251 (unsigned long long)device->ed_uuid);
4252 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4256 if (get_ldev(device)) {
4257 int skip_initial_sync =
4258 device->state.conn == C_CONNECTED &&
4259 peer_device->connection->agreed_pro_version >= 90 &&
4260 device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
4261 (p_uuid[UI_FLAGS] & 8);
4262 if (skip_initial_sync) {
4263 drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
4264 drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
4265 "clear_n_write from receive_uuids",
4266 BM_LOCKED_TEST_ALLOWED);
4267 _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
4268 _drbd_uuid_set(device, UI_BITMAP, 0);
4269 _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
4271 drbd_md_sync(device);
4275 } else if (device->state.disk < D_INCONSISTENT &&
4276 device->state.role == R_PRIMARY) {
4277 /* I am a diskless primary, the peer just created a new current UUID
4279 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4282 /* Before we test for the disk state, we should wait until an eventually
4283 ongoing cluster wide state change is finished. That is important if
4284 we are primary and are detaching from our disk. We need to see the
4285 new disk state... */
4286 mutex_lock(device->state_mutex);
4287 mutex_unlock(device->state_mutex);
4288 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
4289 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4292 drbd_print_uuids(device, "receiver updated UUIDs to");
4298 * convert_state() - Converts the peer's view of the cluster state to our point of view
4299 * @ps: The state as seen by the peer.
4301 static union drbd_state convert_state(union drbd_state ps)
4303 union drbd_state ms;
4305 static enum drbd_conns c_tab[] = {
4306 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
4307 [C_CONNECTED] = C_CONNECTED,
4309 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
4310 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
4311 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
4312 [C_VERIFY_S] = C_VERIFY_T,
4318 ms.conn = c_tab[ps.conn];
4323 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
4328 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4330 struct drbd_peer_device *peer_device;
4331 struct drbd_device *device;
4332 struct p_req_state *p = pi->data;
4333 union drbd_state mask, val;
4334 enum drbd_state_rv rv;
4336 peer_device = conn_peer_device(connection, pi->vnr);
4339 device = peer_device->device;
4341 mask.i = be32_to_cpu(p->mask);
4342 val.i = be32_to_cpu(p->val);
4344 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4345 mutex_is_locked(device->state_mutex)) {
4346 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4350 mask = convert_state(mask);
4351 val = convert_state(val);
4353 rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4354 drbd_send_sr_reply(peer_device, rv);
4356 drbd_md_sync(device);
4361 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4363 struct p_req_state *p = pi->data;
4364 union drbd_state mask, val;
4365 enum drbd_state_rv rv;
4367 mask.i = be32_to_cpu(p->mask);
4368 val.i = be32_to_cpu(p->val);
4370 if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4371 mutex_is_locked(&connection->cstate_mutex)) {
4372 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4376 mask = convert_state(mask);
4377 val = convert_state(val);
4379 rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4380 conn_send_sr_reply(connection, rv);
4385 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4387 struct drbd_peer_device *peer_device;
4388 struct drbd_device *device;
4389 struct p_state *p = pi->data;
4390 union drbd_state os, ns, peer_state;
4391 enum drbd_disk_state real_peer_disk;
4392 enum chg_state_flags cs_flags;
4395 peer_device = conn_peer_device(connection, pi->vnr);
4397 return config_unknown_volume(connection, pi);
4398 device = peer_device->device;
4400 peer_state.i = be32_to_cpu(p->state);
4402 real_peer_disk = peer_state.disk;
4403 if (peer_state.disk == D_NEGOTIATING) {
4404 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4405 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4408 spin_lock_irq(&device->resource->req_lock);
4410 os = ns = drbd_read_state(device);
4411 spin_unlock_irq(&device->resource->req_lock);
4413 /* If some other part of the code (ack_receiver thread, timeout)
4414 * already decided to close the connection again,
4415 * we must not "re-establish" it here. */
4416 if (os.conn <= C_TEAR_DOWN)
4419 /* If this is the "end of sync" confirmation, usually the peer disk
4420 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4421 * set) resync started in PausedSyncT, or if the timing of pause-/
4422 * unpause-sync events has been "just right", the peer disk may
4423 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4425 if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4426 real_peer_disk == D_UP_TO_DATE &&
4427 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4428 /* If we are (becoming) SyncSource, but peer is still in sync
4429 * preparation, ignore its uptodate-ness to avoid flapping, it
4430 * will change to inconsistent once the peer reaches active
4432 * It may have changed syncer-paused flags, however, so we
4433 * cannot ignore this completely. */
4434 if (peer_state.conn > C_CONNECTED &&
4435 peer_state.conn < C_SYNC_SOURCE)
4436 real_peer_disk = D_INCONSISTENT;
4438 /* if peer_state changes to connected at the same time,
4439 * it explicitly notifies us that it finished resync.
4440 * Maybe we should finish it up, too? */
4441 else if (os.conn >= C_SYNC_SOURCE &&
4442 peer_state.conn == C_CONNECTED) {
4443 if (drbd_bm_total_weight(device) <= device->rs_failed)
4444 drbd_resync_finished(device);
4449 /* explicit verify finished notification, stop sector reached. */
4450 if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4451 peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4452 ov_out_of_sync_print(device);
4453 drbd_resync_finished(device);
4457 /* peer says his disk is inconsistent, while we think it is uptodate,
4458 * and this happens while the peer still thinks we have a sync going on,
4459 * but we think we are already done with the sync.
4460 * We ignore this to avoid flapping pdsk.
4461 * This should not happen, if the peer is a recent version of drbd. */
4462 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4463 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4464 real_peer_disk = D_UP_TO_DATE;
4466 if (ns.conn == C_WF_REPORT_PARAMS)
4467 ns.conn = C_CONNECTED;
4469 if (peer_state.conn == C_AHEAD)
4473 * if (primary and diskless and peer uuid != effective uuid)
4474 * abort attach on peer;
4476 * If this node does not have good data, was already connected, but
4477 * the peer did a late attach only now, trying to "negotiate" with me,
4478 * AND I am currently Primary, possibly frozen, with some specific
4479 * "effective" uuid, this should never be reached, really, because
4480 * we first send the uuids, then the current state.
4482 * In this scenario, we already dropped the connection hard
4483 * when we received the unsuitable uuids (receive_uuids().
4485 * Should we want to change this, that is: not drop the connection in
4486 * receive_uuids() already, then we would need to add a branch here
4487 * that aborts the attach of "unsuitable uuids" on the peer in case
4488 * this node is currently Diskless Primary.
4491 if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4492 get_ldev_if_state(device, D_NEGOTIATING)) {
4493 int cr; /* consider resync */
4495 /* if we established a new connection */
4496 cr = (os.conn < C_CONNECTED);
4497 /* if we had an established connection
4498 * and one of the nodes newly attaches a disk */
4499 cr |= (os.conn == C_CONNECTED &&
4500 (peer_state.disk == D_NEGOTIATING ||
4501 os.disk == D_NEGOTIATING));
4502 /* if we have both been inconsistent, and the peer has been
4503 * forced to be UpToDate with --force */
4504 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4505 /* if we had been plain connected, and the admin requested to
4506 * start a sync by "invalidate" or "invalidate-remote" */
4507 cr |= (os.conn == C_CONNECTED &&
4508 (peer_state.conn >= C_STARTING_SYNC_S &&
4509 peer_state.conn <= C_WF_BITMAP_T));
4512 ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4515 if (ns.conn == C_MASK) {
4516 ns.conn = C_CONNECTED;
4517 if (device->state.disk == D_NEGOTIATING) {
4518 drbd_force_state(device, NS(disk, D_FAILED));
4519 } else if (peer_state.disk == D_NEGOTIATING) {
4520 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4521 peer_state.disk = D_DISKLESS;
4522 real_peer_disk = D_DISKLESS;
4524 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4526 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4527 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4533 spin_lock_irq(&device->resource->req_lock);
4534 if (os.i != drbd_read_state(device).i)
4536 clear_bit(CONSIDER_RESYNC, &device->flags);
4537 ns.peer = peer_state.role;
4538 ns.pdsk = real_peer_disk;
4539 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4540 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4541 ns.disk = device->new_state_tmp.disk;
4542 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4543 if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4544 test_bit(NEW_CUR_UUID, &device->flags)) {
4545 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4546 for temporal network outages! */
4547 spin_unlock_irq(&device->resource->req_lock);
4548 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4549 tl_clear(peer_device->connection);
4550 drbd_uuid_new_current(device);
4551 clear_bit(NEW_CUR_UUID, &device->flags);
4552 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4555 rv = _drbd_set_state(device, ns, cs_flags, NULL);
4556 ns = drbd_read_state(device);
4557 spin_unlock_irq(&device->resource->req_lock);
4559 if (rv < SS_SUCCESS) {
4560 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4564 if (os.conn > C_WF_REPORT_PARAMS) {
4565 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4566 peer_state.disk != D_NEGOTIATING ) {
4567 /* we want resync, peer has not yet decided to sync... */
4568 /* Nowadays only used when forcing a node into primary role and
4569 setting its disk to UpToDate with that */
4570 drbd_send_uuids(peer_device);
4571 drbd_send_current_state(peer_device);
4575 clear_bit(DISCARD_MY_DATA, &device->flags);
4577 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4582 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4584 struct drbd_peer_device *peer_device;
4585 struct drbd_device *device;
4586 struct p_rs_uuid *p = pi->data;
4588 peer_device = conn_peer_device(connection, pi->vnr);
4591 device = peer_device->device;
4593 wait_event(device->misc_wait,
4594 device->state.conn == C_WF_SYNC_UUID ||
4595 device->state.conn == C_BEHIND ||
4596 device->state.conn < C_CONNECTED ||
4597 device->state.disk < D_NEGOTIATING);
4599 /* D_ASSERT(device, device->state.conn == C_WF_SYNC_UUID ); */
4601 /* Here the _drbd_uuid_ functions are right, current should
4602 _not_ be rotated into the history */
4603 if (get_ldev_if_state(device, D_NEGOTIATING)) {
4604 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4605 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4607 drbd_print_uuids(device, "updated sync uuid");
4608 drbd_start_resync(device, C_SYNC_TARGET);
4612 drbd_err(device, "Ignoring SyncUUID packet!\n");
4618 * receive_bitmap_plain
4620 * Return 0 when done, 1 when another iteration is needed, and a negative error
4621 * code upon failure.
4624 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4625 unsigned long *p, struct bm_xfer_ctx *c)
4627 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4628 drbd_header_size(peer_device->connection);
4629 unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4630 c->bm_words - c->word_offset);
4631 unsigned int want = num_words * sizeof(*p);
4635 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4640 err = drbd_recv_all(peer_device->connection, p, want);
4644 drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4646 c->word_offset += num_words;
4647 c->bit_offset = c->word_offset * BITS_PER_LONG;
4648 if (c->bit_offset > c->bm_bits)
4649 c->bit_offset = c->bm_bits;
4654 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4656 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4659 static int dcbp_get_start(struct p_compressed_bm *p)
4661 return (p->encoding & 0x80) != 0;
4664 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4666 return (p->encoding >> 4) & 0x7;
4672 * Return 0 when done, 1 when another iteration is needed, and a negative error
4673 * code upon failure.
4676 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4677 struct p_compressed_bm *p,
4678 struct bm_xfer_ctx *c,
4681 struct bitstream bs;
4685 unsigned long s = c->bit_offset;
4687 int toggle = dcbp_get_start(p);
4691 bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4693 bits = bitstream_get_bits(&bs, &look_ahead, 64);
4697 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4698 bits = vli_decode_bits(&rl, look_ahead);
4704 if (e >= c->bm_bits) {
4705 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4708 _drbd_bm_set_bits(peer_device->device, s, e);
4712 drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4713 have, bits, look_ahead,
4714 (unsigned int)(bs.cur.b - p->code),
4715 (unsigned int)bs.buf_len);
4718 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4719 if (likely(bits < 64))
4720 look_ahead >>= bits;
4725 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4728 look_ahead |= tmp << have;
4733 bm_xfer_ctx_bit_to_word_offset(c);
4735 return (s != c->bm_bits);
4741 * Return 0 when done, 1 when another iteration is needed, and a negative error
4742 * code upon failure.
4745 decode_bitmap_c(struct drbd_peer_device *peer_device,
4746 struct p_compressed_bm *p,
4747 struct bm_xfer_ctx *c,
4750 if (dcbp_get_code(p) == RLE_VLI_Bits)
4751 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4753 /* other variants had been implemented for evaluation,
4754 * but have been dropped as this one turned out to be "best"
4755 * during all our tests. */
4757 drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4758 conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4762 void INFO_bm_xfer_stats(struct drbd_device *device,
4763 const char *direction, struct bm_xfer_ctx *c)
4765 /* what would it take to transfer it "plaintext" */
4766 unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4767 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4768 unsigned int plain =
4769 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4770 c->bm_words * sizeof(unsigned long);
4771 unsigned int total = c->bytes[0] + c->bytes[1];
4774 /* total can not be zero. but just in case: */
4778 /* don't report if not compressed */
4782 /* total < plain. check for overflow, still */
4783 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4784 : (1000 * total / plain);
4790 drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4791 "total %u; compression: %u.%u%%\n",
4793 c->bytes[1], c->packets[1],
4794 c->bytes[0], c->packets[0],
4795 total, r/10, r % 10);
4798 /* Since we are processing the bitfield from lower addresses to higher,
4799 it does not matter if the process it in 32 bit chunks or 64 bit
4800 chunks as long as it is little endian. (Understand it as byte stream,
4801 beginning with the lowest byte...) If we would use big endian
4802 we would need to process it from the highest address to the lowest,
4803 in order to be agnostic to the 32 vs 64 bits issue.
4805 returns 0 on failure, 1 if we successfully received it. */
4806 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4808 struct drbd_peer_device *peer_device;
4809 struct drbd_device *device;
4810 struct bm_xfer_ctx c;
4813 peer_device = conn_peer_device(connection, pi->vnr);
4816 device = peer_device->device;
4818 drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4819 /* you are supposed to send additional out-of-sync information
4820 * if you actually set bits during this phase */
4822 c = (struct bm_xfer_ctx) {
4823 .bm_bits = drbd_bm_bits(device),
4824 .bm_words = drbd_bm_words(device),
4828 if (pi->cmd == P_BITMAP)
4829 err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4830 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4831 /* MAYBE: sanity check that we speak proto >= 90,
4832 * and the feature is enabled! */
4833 struct p_compressed_bm *p = pi->data;
4835 if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4836 drbd_err(device, "ReportCBitmap packet too large\n");
4840 if (pi->size <= sizeof(*p)) {
4841 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4845 err = drbd_recv_all(peer_device->connection, p, pi->size);
4848 err = decode_bitmap_c(peer_device, p, &c, pi->size);
4850 drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4855 c.packets[pi->cmd == P_BITMAP]++;
4856 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4863 err = drbd_recv_header(peer_device->connection, pi);
4868 INFO_bm_xfer_stats(device, "receive", &c);
4870 if (device->state.conn == C_WF_BITMAP_T) {
4871 enum drbd_state_rv rv;
4873 err = drbd_send_bitmap(device);
4876 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4877 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4878 D_ASSERT(device, rv == SS_SUCCESS);
4879 } else if (device->state.conn != C_WF_BITMAP_S) {
4880 /* admin may have requested C_DISCONNECTING,
4881 * other threads may have noticed network errors */
4882 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4883 drbd_conn_str(device->state.conn));
4888 drbd_bm_unlock(device);
4889 if (!err && device->state.conn == C_WF_BITMAP_S)
4890 drbd_start_resync(device, C_SYNC_SOURCE);
4894 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4896 drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4899 return ignore_remaining_packet(connection, pi);
4902 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4904 /* Make sure we've acked all the TCP data associated
4905 * with the data requests being unplugged */
4906 tcp_sock_set_quickack(connection->data.socket->sk, 2);
4910 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4912 struct drbd_peer_device *peer_device;
4913 struct drbd_device *device;
4914 struct p_block_desc *p = pi->data;
4916 peer_device = conn_peer_device(connection, pi->vnr);
4919 device = peer_device->device;
4921 switch (device->state.conn) {
4922 case C_WF_SYNC_UUID:
4927 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4928 drbd_conn_str(device->state.conn));
4931 drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4936 static int receive_rs_deallocated(struct drbd_connection *connection, struct packet_info *pi)
4938 struct drbd_peer_device *peer_device;
4939 struct p_block_desc *p = pi->data;
4940 struct drbd_device *device;
4944 peer_device = conn_peer_device(connection, pi->vnr);
4947 device = peer_device->device;
4949 sector = be64_to_cpu(p->sector);
4950 size = be32_to_cpu(p->blksize);
4952 dec_rs_pending(device);
4954 if (get_ldev(device)) {
4955 struct drbd_peer_request *peer_req;
4956 const int op = REQ_OP_WRITE_ZEROES;
4958 peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER, sector,
4965 peer_req->w.cb = e_end_resync_block;
4966 peer_req->submit_jif = jiffies;
4967 peer_req->flags |= EE_TRIM;
4969 spin_lock_irq(&device->resource->req_lock);
4970 list_add_tail(&peer_req->w.list, &device->sync_ee);
4971 spin_unlock_irq(&device->resource->req_lock);
4973 atomic_add(pi->size >> 9, &device->rs_sect_ev);
4974 err = drbd_submit_peer_request(device, peer_req, op, 0, DRBD_FAULT_RS_WR);
4977 spin_lock_irq(&device->resource->req_lock);
4978 list_del(&peer_req->w.list);
4979 spin_unlock_irq(&device->resource->req_lock);
4981 drbd_free_peer_req(device, peer_req);
4987 inc_unacked(device);
4989 /* No put_ldev() here. Gets called in drbd_endio_write_sec_final(),
4990 as well as drbd_rs_complete_io() */
4993 drbd_rs_complete_io(device, sector);
4994 drbd_send_ack_ex(peer_device, P_NEG_ACK, sector, size, ID_SYNCER);
4997 atomic_add(size >> 9, &device->rs_sect_in);
5004 unsigned int pkt_size;
5005 int (*fn)(struct drbd_connection *, struct packet_info *);
5008 static struct data_cmd drbd_cmd_handler[] = {
5009 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
5010 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
5011 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
5012 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
5013 [P_BITMAP] = { 1, 0, receive_bitmap } ,
5014 [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
5015 [P_UNPLUG_REMOTE] = { 0, 0, receive_UnplugRemote },
5016 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5017 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5018 [P_SYNC_PARAM] = { 1, 0, receive_SyncParam },
5019 [P_SYNC_PARAM89] = { 1, 0, receive_SyncParam },
5020 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
5021 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
5022 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
5023 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
5024 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
5025 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
5026 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5027 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5028 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5029 [P_RS_THIN_REQ] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5030 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
5031 [P_OUT_OF_SYNC] = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
5032 [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
5033 [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
5034 [P_TRIM] = { 0, sizeof(struct p_trim), receive_Data },
5035 [P_ZEROES] = { 0, sizeof(struct p_trim), receive_Data },
5036 [P_RS_DEALLOCATED] = { 0, sizeof(struct p_block_desc), receive_rs_deallocated },
5039 static void drbdd(struct drbd_connection *connection)
5041 struct packet_info pi;
5042 size_t shs; /* sub header size */
5045 while (get_t_state(&connection->receiver) == RUNNING) {
5046 struct data_cmd const *cmd;
5048 drbd_thread_current_set_cpu(&connection->receiver);
5049 update_receiver_timing_details(connection, drbd_recv_header_maybe_unplug);
5050 if (drbd_recv_header_maybe_unplug(connection, &pi))
5053 cmd = &drbd_cmd_handler[pi.cmd];
5054 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
5055 drbd_err(connection, "Unexpected data packet %s (0x%04x)",
5056 cmdname(pi.cmd), pi.cmd);
5060 shs = cmd->pkt_size;
5061 if (pi.cmd == P_SIZES && connection->agreed_features & DRBD_FF_WSAME)
5062 shs += sizeof(struct o_qlim);
5063 if (pi.size > shs && !cmd->expect_payload) {
5064 drbd_err(connection, "No payload expected %s l:%d\n",
5065 cmdname(pi.cmd), pi.size);
5068 if (pi.size < shs) {
5069 drbd_err(connection, "%s: unexpected packet size, expected:%d received:%d\n",
5070 cmdname(pi.cmd), (int)shs, pi.size);
5075 update_receiver_timing_details(connection, drbd_recv_all_warn);
5076 err = drbd_recv_all_warn(connection, pi.data, shs);
5082 update_receiver_timing_details(connection, cmd->fn);
5083 err = cmd->fn(connection, &pi);
5085 drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
5086 cmdname(pi.cmd), err, pi.size);
5093 conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
5096 static void conn_disconnect(struct drbd_connection *connection)
5098 struct drbd_peer_device *peer_device;
5102 if (connection->cstate == C_STANDALONE)
5105 /* We are about to start the cleanup after connection loss.
5106 * Make sure drbd_make_request knows about that.
5107 * Usually we should be in some network failure state already,
5108 * but just in case we are not, we fix it up here.
5110 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5112 /* ack_receiver does not clean up anything. it must not interfere, either */
5113 drbd_thread_stop(&connection->ack_receiver);
5114 if (connection->ack_sender) {
5115 destroy_workqueue(connection->ack_sender);
5116 connection->ack_sender = NULL;
5118 drbd_free_sock(connection);
5121 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5122 struct drbd_device *device = peer_device->device;
5123 kref_get(&device->kref);
5125 drbd_disconnected(peer_device);
5126 kref_put(&device->kref, drbd_destroy_device);
5131 if (!list_empty(&connection->current_epoch->list))
5132 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
5133 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
5134 atomic_set(&connection->current_epoch->epoch_size, 0);
5135 connection->send.seen_any_write_yet = false;
5137 drbd_info(connection, "Connection closed\n");
5139 if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
5140 conn_try_outdate_peer_async(connection);
5142 spin_lock_irq(&connection->resource->req_lock);
5143 oc = connection->cstate;
5144 if (oc >= C_UNCONNECTED)
5145 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
5147 spin_unlock_irq(&connection->resource->req_lock);
5149 if (oc == C_DISCONNECTING)
5150 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
5153 static int drbd_disconnected(struct drbd_peer_device *peer_device)
5155 struct drbd_device *device = peer_device->device;
5158 /* wait for current activity to cease. */
5159 spin_lock_irq(&device->resource->req_lock);
5160 _drbd_wait_ee_list_empty(device, &device->active_ee);
5161 _drbd_wait_ee_list_empty(device, &device->sync_ee);
5162 _drbd_wait_ee_list_empty(device, &device->read_ee);
5163 spin_unlock_irq(&device->resource->req_lock);
5165 /* We do not have data structures that would allow us to
5166 * get the rs_pending_cnt down to 0 again.
5167 * * On C_SYNC_TARGET we do not have any data structures describing
5168 * the pending RSDataRequest's we have sent.
5169 * * On C_SYNC_SOURCE there is no data structure that tracks
5170 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
5171 * And no, it is not the sum of the reference counts in the
5172 * resync_LRU. The resync_LRU tracks the whole operation including
5173 * the disk-IO, while the rs_pending_cnt only tracks the blocks
5175 drbd_rs_cancel_all(device);
5176 device->rs_total = 0;
5177 device->rs_failed = 0;
5178 atomic_set(&device->rs_pending_cnt, 0);
5179 wake_up(&device->misc_wait);
5181 del_timer_sync(&device->resync_timer);
5182 resync_timer_fn(&device->resync_timer);
5184 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
5185 * w_make_resync_request etc. which may still be on the worker queue
5186 * to be "canceled" */
5187 drbd_flush_workqueue(&peer_device->connection->sender_work);
5189 drbd_finish_peer_reqs(device);
5191 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
5192 might have issued a work again. The one before drbd_finish_peer_reqs() is
5193 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
5194 drbd_flush_workqueue(&peer_device->connection->sender_work);
5196 /* need to do it again, drbd_finish_peer_reqs() may have populated it
5197 * again via drbd_try_clear_on_disk_bm(). */
5198 drbd_rs_cancel_all(device);
5200 kfree(device->p_uuid);
5201 device->p_uuid = NULL;
5203 if (!drbd_suspended(device))
5204 tl_clear(peer_device->connection);
5206 drbd_md_sync(device);
5208 if (get_ldev(device)) {
5209 drbd_bitmap_io(device, &drbd_bm_write_copy_pages,
5210 "write from disconnected", BM_LOCKED_CHANGE_ALLOWED);
5214 /* tcp_close and release of sendpage pages can be deferred. I don't
5215 * want to use SO_LINGER, because apparently it can be deferred for
5216 * more than 20 seconds (longest time I checked).
5218 * Actually we don't care for exactly when the network stack does its
5219 * put_page(), but release our reference on these pages right here.
5221 i = drbd_free_peer_reqs(device, &device->net_ee);
5223 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
5224 i = atomic_read(&device->pp_in_use_by_net);
5226 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
5227 i = atomic_read(&device->pp_in_use);
5229 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
5231 D_ASSERT(device, list_empty(&device->read_ee));
5232 D_ASSERT(device, list_empty(&device->active_ee));
5233 D_ASSERT(device, list_empty(&device->sync_ee));
5234 D_ASSERT(device, list_empty(&device->done_ee));
5240 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
5241 * we can agree on is stored in agreed_pro_version.
5243 * feature flags and the reserved array should be enough room for future
5244 * enhancements of the handshake protocol, and possible plugins...
5246 * for now, they are expected to be zero, but ignored.
5248 static int drbd_send_features(struct drbd_connection *connection)
5250 struct drbd_socket *sock;
5251 struct p_connection_features *p;
5253 sock = &connection->data;
5254 p = conn_prepare_command(connection, sock);
5257 memset(p, 0, sizeof(*p));
5258 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
5259 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
5260 p->feature_flags = cpu_to_be32(PRO_FEATURES);
5261 return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
5266 * 1 yes, we have a valid connection
5267 * 0 oops, did not work out, please try again
5268 * -1 peer talks different language,
5269 * no point in trying again, please go standalone.
5271 static int drbd_do_features(struct drbd_connection *connection)
5273 /* ASSERT current == connection->receiver ... */
5274 struct p_connection_features *p;
5275 const int expect = sizeof(struct p_connection_features);
5276 struct packet_info pi;
5279 err = drbd_send_features(connection);
5283 err = drbd_recv_header(connection, &pi);
5287 if (pi.cmd != P_CONNECTION_FEATURES) {
5288 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
5289 cmdname(pi.cmd), pi.cmd);
5293 if (pi.size != expect) {
5294 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
5300 err = drbd_recv_all_warn(connection, p, expect);
5304 p->protocol_min = be32_to_cpu(p->protocol_min);
5305 p->protocol_max = be32_to_cpu(p->protocol_max);
5306 if (p->protocol_max == 0)
5307 p->protocol_max = p->protocol_min;
5309 if (PRO_VERSION_MAX < p->protocol_min ||
5310 PRO_VERSION_MIN > p->protocol_max)
5313 connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
5314 connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
5316 drbd_info(connection, "Handshake successful: "
5317 "Agreed network protocol version %d\n", connection->agreed_pro_version);
5319 drbd_info(connection, "Feature flags enabled on protocol level: 0x%x%s%s%s%s.\n",
5320 connection->agreed_features,
5321 connection->agreed_features & DRBD_FF_TRIM ? " TRIM" : "",
5322 connection->agreed_features & DRBD_FF_THIN_RESYNC ? " THIN_RESYNC" : "",
5323 connection->agreed_features & DRBD_FF_WSAME ? " WRITE_SAME" : "",
5324 connection->agreed_features & DRBD_FF_WZEROES ? " WRITE_ZEROES" :
5325 connection->agreed_features ? "" : " none");
5330 drbd_err(connection, "incompatible DRBD dialects: "
5331 "I support %d-%d, peer supports %d-%d\n",
5332 PRO_VERSION_MIN, PRO_VERSION_MAX,
5333 p->protocol_min, p->protocol_max);
5337 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
5338 static int drbd_do_auth(struct drbd_connection *connection)
5340 drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
5341 drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
5345 #define CHALLENGE_LEN 64
5349 0 - failed, try again (network error),
5350 -1 - auth failed, don't try again.
5353 static int drbd_do_auth(struct drbd_connection *connection)
5355 struct drbd_socket *sock;
5356 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
5357 char *response = NULL;
5358 char *right_response = NULL;
5359 char *peers_ch = NULL;
5360 unsigned int key_len;
5361 char secret[SHARED_SECRET_MAX]; /* 64 byte */
5362 unsigned int resp_size;
5363 struct shash_desc *desc;
5364 struct packet_info pi;
5365 struct net_conf *nc;
5368 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
5371 nc = rcu_dereference(connection->net_conf);
5372 key_len = strlen(nc->shared_secret);
5373 memcpy(secret, nc->shared_secret, key_len);
5376 desc = kmalloc(sizeof(struct shash_desc) +
5377 crypto_shash_descsize(connection->cram_hmac_tfm),
5383 desc->tfm = connection->cram_hmac_tfm;
5385 rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
5387 drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
5392 get_random_bytes(my_challenge, CHALLENGE_LEN);
5394 sock = &connection->data;
5395 if (!conn_prepare_command(connection, sock)) {
5399 rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
5400 my_challenge, CHALLENGE_LEN);
5404 err = drbd_recv_header(connection, &pi);
5410 if (pi.cmd != P_AUTH_CHALLENGE) {
5411 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
5412 cmdname(pi.cmd), pi.cmd);
5417 if (pi.size > CHALLENGE_LEN * 2) {
5418 drbd_err(connection, "expected AuthChallenge payload too big.\n");
5423 if (pi.size < CHALLENGE_LEN) {
5424 drbd_err(connection, "AuthChallenge payload too small.\n");
5429 peers_ch = kmalloc(pi.size, GFP_NOIO);
5435 err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5441 if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5442 drbd_err(connection, "Peer presented the same challenge!\n");
5447 resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5448 response = kmalloc(resp_size, GFP_NOIO);
5454 rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5456 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5461 if (!conn_prepare_command(connection, sock)) {
5465 rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5466 response, resp_size);
5470 err = drbd_recv_header(connection, &pi);
5476 if (pi.cmd != P_AUTH_RESPONSE) {
5477 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5478 cmdname(pi.cmd), pi.cmd);
5483 if (pi.size != resp_size) {
5484 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5489 err = drbd_recv_all_warn(connection, response , resp_size);
5495 right_response = kmalloc(resp_size, GFP_NOIO);
5496 if (!right_response) {
5501 rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5504 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5509 rv = !memcmp(response, right_response, resp_size);
5512 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5520 kfree(right_response);
5522 shash_desc_zero(desc);
5530 int drbd_receiver(struct drbd_thread *thi)
5532 struct drbd_connection *connection = thi->connection;
5535 drbd_info(connection, "receiver (re)started\n");
5538 h = conn_connect(connection);
5540 conn_disconnect(connection);
5541 schedule_timeout_interruptible(HZ);
5544 drbd_warn(connection, "Discarding network configuration.\n");
5545 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5550 blk_start_plug(&connection->receiver_plug);
5552 blk_finish_plug(&connection->receiver_plug);
5555 conn_disconnect(connection);
5557 drbd_info(connection, "receiver terminated\n");
5561 /* ********* acknowledge sender ******** */
5563 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5565 struct p_req_state_reply *p = pi->data;
5566 int retcode = be32_to_cpu(p->retcode);
5568 if (retcode >= SS_SUCCESS) {
5569 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5571 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5572 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5573 drbd_set_st_err_str(retcode), retcode);
5575 wake_up(&connection->ping_wait);
5580 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5582 struct drbd_peer_device *peer_device;
5583 struct drbd_device *device;
5584 struct p_req_state_reply *p = pi->data;
5585 int retcode = be32_to_cpu(p->retcode);
5587 peer_device = conn_peer_device(connection, pi->vnr);
5590 device = peer_device->device;
5592 if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5593 D_ASSERT(device, connection->agreed_pro_version < 100);
5594 return got_conn_RqSReply(connection, pi);
5597 if (retcode >= SS_SUCCESS) {
5598 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5600 set_bit(CL_ST_CHG_FAIL, &device->flags);
5601 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5602 drbd_set_st_err_str(retcode), retcode);
5604 wake_up(&device->state_wait);
5609 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5611 return drbd_send_ping_ack(connection);
5615 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5617 /* restore idle timeout */
5618 connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5619 if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5620 wake_up(&connection->ping_wait);
5625 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5627 struct drbd_peer_device *peer_device;
5628 struct drbd_device *device;
5629 struct p_block_ack *p = pi->data;
5630 sector_t sector = be64_to_cpu(p->sector);
5631 int blksize = be32_to_cpu(p->blksize);
5633 peer_device = conn_peer_device(connection, pi->vnr);
5636 device = peer_device->device;
5638 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5640 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5642 if (get_ldev(device)) {
5643 drbd_rs_complete_io(device, sector);
5644 drbd_set_in_sync(device, sector, blksize);
5645 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5646 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5649 dec_rs_pending(device);
5650 atomic_add(blksize >> 9, &device->rs_sect_in);
5656 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5657 struct rb_root *root, const char *func,
5658 enum drbd_req_event what, bool missing_ok)
5660 struct drbd_request *req;
5661 struct bio_and_error m;
5663 spin_lock_irq(&device->resource->req_lock);
5664 req = find_request(device, root, id, sector, missing_ok, func);
5665 if (unlikely(!req)) {
5666 spin_unlock_irq(&device->resource->req_lock);
5669 __req_mod(req, what, &m);
5670 spin_unlock_irq(&device->resource->req_lock);
5673 complete_master_bio(device, &m);
5677 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5679 struct drbd_peer_device *peer_device;
5680 struct drbd_device *device;
5681 struct p_block_ack *p = pi->data;
5682 sector_t sector = be64_to_cpu(p->sector);
5683 int blksize = be32_to_cpu(p->blksize);
5684 enum drbd_req_event what;
5686 peer_device = conn_peer_device(connection, pi->vnr);
5689 device = peer_device->device;
5691 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5693 if (p->block_id == ID_SYNCER) {
5694 drbd_set_in_sync(device, sector, blksize);
5695 dec_rs_pending(device);
5699 case P_RS_WRITE_ACK:
5700 what = WRITE_ACKED_BY_PEER_AND_SIS;
5703 what = WRITE_ACKED_BY_PEER;
5706 what = RECV_ACKED_BY_PEER;
5709 what = CONFLICT_RESOLVED;
5712 what = POSTPONE_WRITE;
5718 return validate_req_change_req_state(device, p->block_id, sector,
5719 &device->write_requests, __func__,
5723 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5725 struct drbd_peer_device *peer_device;
5726 struct drbd_device *device;
5727 struct p_block_ack *p = pi->data;
5728 sector_t sector = be64_to_cpu(p->sector);
5729 int size = be32_to_cpu(p->blksize);
5732 peer_device = conn_peer_device(connection, pi->vnr);
5735 device = peer_device->device;
5737 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5739 if (p->block_id == ID_SYNCER) {
5740 dec_rs_pending(device);
5741 drbd_rs_failed_io(device, sector, size);
5745 err = validate_req_change_req_state(device, p->block_id, sector,
5746 &device->write_requests, __func__,
5749 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5750 The master bio might already be completed, therefore the
5751 request is no longer in the collision hash. */
5752 /* In Protocol B we might already have got a P_RECV_ACK
5753 but then get a P_NEG_ACK afterwards. */
5754 drbd_set_out_of_sync(device, sector, size);
5759 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5761 struct drbd_peer_device *peer_device;
5762 struct drbd_device *device;
5763 struct p_block_ack *p = pi->data;
5764 sector_t sector = be64_to_cpu(p->sector);
5766 peer_device = conn_peer_device(connection, pi->vnr);
5769 device = peer_device->device;
5771 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5773 drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5774 (unsigned long long)sector, be32_to_cpu(p->blksize));
5776 return validate_req_change_req_state(device, p->block_id, sector,
5777 &device->read_requests, __func__,
5781 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5783 struct drbd_peer_device *peer_device;
5784 struct drbd_device *device;
5787 struct p_block_ack *p = pi->data;
5789 peer_device = conn_peer_device(connection, pi->vnr);
5792 device = peer_device->device;
5794 sector = be64_to_cpu(p->sector);
5795 size = be32_to_cpu(p->blksize);
5797 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5799 dec_rs_pending(device);
5801 if (get_ldev_if_state(device, D_FAILED)) {
5802 drbd_rs_complete_io(device, sector);
5804 case P_NEG_RS_DREPLY:
5805 drbd_rs_failed_io(device, sector, size);
5818 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5820 struct p_barrier_ack *p = pi->data;
5821 struct drbd_peer_device *peer_device;
5824 tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5827 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5828 struct drbd_device *device = peer_device->device;
5830 if (device->state.conn == C_AHEAD &&
5831 atomic_read(&device->ap_in_flight) == 0 &&
5832 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5833 device->start_resync_timer.expires = jiffies + HZ;
5834 add_timer(&device->start_resync_timer);
5842 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5844 struct drbd_peer_device *peer_device;
5845 struct drbd_device *device;
5846 struct p_block_ack *p = pi->data;
5847 struct drbd_device_work *dw;
5851 peer_device = conn_peer_device(connection, pi->vnr);
5854 device = peer_device->device;
5856 sector = be64_to_cpu(p->sector);
5857 size = be32_to_cpu(p->blksize);
5859 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5861 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5862 drbd_ov_out_of_sync_found(device, sector, size);
5864 ov_out_of_sync_print(device);
5866 if (!get_ldev(device))
5869 drbd_rs_complete_io(device, sector);
5870 dec_rs_pending(device);
5874 /* let's advance progress step marks only for every other megabyte */
5875 if ((device->ov_left & 0x200) == 0x200)
5876 drbd_advance_rs_marks(device, device->ov_left);
5878 if (device->ov_left == 0) {
5879 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5881 dw->w.cb = w_ov_finished;
5882 dw->device = device;
5883 drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5885 drbd_err(device, "kmalloc(dw) failed.");
5886 ov_out_of_sync_print(device);
5887 drbd_resync_finished(device);
5894 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5899 struct meta_sock_cmd {
5901 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5904 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5907 struct net_conf *nc;
5910 nc = rcu_dereference(connection->net_conf);
5911 t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5918 connection->meta.socket->sk->sk_rcvtimeo = t;
5921 static void set_ping_timeout(struct drbd_connection *connection)
5923 set_rcvtimeo(connection, 1);
5926 static void set_idle_timeout(struct drbd_connection *connection)
5928 set_rcvtimeo(connection, 0);
5931 static struct meta_sock_cmd ack_receiver_tbl[] = {
5932 [P_PING] = { 0, got_Ping },
5933 [P_PING_ACK] = { 0, got_PingAck },
5934 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5935 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5936 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5937 [P_SUPERSEDED] = { sizeof(struct p_block_ack), got_BlockAck },
5938 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
5939 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
5940 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply },
5941 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
5942 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
5943 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5944 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
5945 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
5946 [P_RS_CANCEL] = { sizeof(struct p_block_ack), got_NegRSDReply },
5947 [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5948 [P_RETRY_WRITE] = { sizeof(struct p_block_ack), got_BlockAck },
5951 int drbd_ack_receiver(struct drbd_thread *thi)
5953 struct drbd_connection *connection = thi->connection;
5954 struct meta_sock_cmd *cmd = NULL;
5955 struct packet_info pi;
5956 unsigned long pre_recv_jif;
5958 void *buf = connection->meta.rbuf;
5960 unsigned int header_size = drbd_header_size(connection);
5961 int expect = header_size;
5962 bool ping_timeout_active = false;
5964 sched_set_fifo_low(current);
5966 while (get_t_state(thi) == RUNNING) {
5967 drbd_thread_current_set_cpu(thi);
5969 conn_reclaim_net_peer_reqs(connection);
5971 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5972 if (drbd_send_ping(connection)) {
5973 drbd_err(connection, "drbd_send_ping has failed\n");
5976 set_ping_timeout(connection);
5977 ping_timeout_active = true;
5980 pre_recv_jif = jiffies;
5981 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5984 * -EINTR (on meta) we got a signal
5985 * -EAGAIN (on meta) rcvtimeo expired
5986 * -ECONNRESET other side closed the connection
5987 * -ERESTARTSYS (on data) we got a signal
5988 * rv < 0 other than above: unexpected error!
5989 * rv == expected: full header or command
5990 * rv < expected: "woken" by signal during receive
5991 * rv == 0 : "connection shut down by peer"
5993 if (likely(rv > 0)) {
5996 } else if (rv == 0) {
5997 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
6000 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
6003 t = wait_event_timeout(connection->ping_wait,
6004 connection->cstate < C_WF_REPORT_PARAMS,
6009 drbd_err(connection, "meta connection shut down by peer.\n");
6011 } else if (rv == -EAGAIN) {
6012 /* If the data socket received something meanwhile,
6013 * that is good enough: peer is still alive. */
6014 if (time_after(connection->last_received, pre_recv_jif))
6016 if (ping_timeout_active) {
6017 drbd_err(connection, "PingAck did not arrive in time.\n");
6020 set_bit(SEND_PING, &connection->flags);
6022 } else if (rv == -EINTR) {
6023 /* maybe drbd_thread_stop(): the while condition will notice.
6024 * maybe woken for send_ping: we'll send a ping above,
6025 * and change the rcvtimeo */
6026 flush_signals(current);
6029 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
6033 if (received == expect && cmd == NULL) {
6034 if (decode_header(connection, connection->meta.rbuf, &pi))
6036 cmd = &ack_receiver_tbl[pi.cmd];
6037 if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
6038 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
6039 cmdname(pi.cmd), pi.cmd);
6042 expect = header_size + cmd->pkt_size;
6043 if (pi.size != expect - header_size) {
6044 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
6049 if (received == expect) {
6052 err = cmd->fn(connection, &pi);
6054 drbd_err(connection, "%ps failed\n", cmd->fn);
6058 connection->last_received = jiffies;
6060 if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
6061 set_idle_timeout(connection);
6062 ping_timeout_active = false;
6065 buf = connection->meta.rbuf;
6067 expect = header_size;
6074 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6075 conn_md_sync(connection);
6079 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
6082 drbd_info(connection, "ack_receiver terminated\n");
6087 void drbd_send_acks_wf(struct work_struct *ws)
6089 struct drbd_peer_device *peer_device =
6090 container_of(ws, struct drbd_peer_device, send_acks_work);
6091 struct drbd_connection *connection = peer_device->connection;
6092 struct drbd_device *device = peer_device->device;
6093 struct net_conf *nc;
6097 nc = rcu_dereference(connection->net_conf);
6098 tcp_cork = nc->tcp_cork;
6102 tcp_sock_set_cork(connection->meta.socket->sk, true);
6104 err = drbd_finish_peer_reqs(device);
6105 kref_put(&device->kref, drbd_destroy_device);
6106 /* get is in drbd_endio_write_sec_final(). That is necessary to keep the
6107 struct work_struct send_acks_work alive, which is in the peer_device object */
6110 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6115 tcp_sock_set_cork(connection->meta.socket->sk, false);