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(GFP_NOIO, 0);
1283 struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);
1284 if (!bio || !octx) {
1285 drbd_warn(device, "Could not allocate a bio, CANNOT ISSUE FLUSH\n");
1286 /* FIXME: what else can I do now? disconnecting or detaching
1287 * 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_set_dev(bio, device->ldev->backing_bdev);
1302 bio->bi_private = octx;
1303 bio->bi_end_io = one_flush_endio;
1304 bio->bi_opf = REQ_OP_FLUSH | REQ_PREFLUSH;
1306 device->flush_jif = jiffies;
1307 set_bit(FLUSH_PENDING, &device->flags);
1308 atomic_inc(&ctx->pending);
1312 static void drbd_flush(struct drbd_connection *connection)
1314 if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1315 struct drbd_peer_device *peer_device;
1316 struct issue_flush_context ctx;
1319 atomic_set(&ctx.pending, 1);
1321 init_completion(&ctx.done);
1324 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1325 struct drbd_device *device = peer_device->device;
1327 if (!get_ldev(device))
1329 kref_get(&device->kref);
1332 submit_one_flush(device, &ctx);
1338 /* Do we want to add a timeout,
1339 * if disk-timeout is set? */
1340 if (!atomic_dec_and_test(&ctx.pending))
1341 wait_for_completion(&ctx.done);
1344 /* would rather check on EOPNOTSUPP, but that is not reliable.
1345 * don't try again for ANY return value != 0
1346 * if (rv == -EOPNOTSUPP) */
1347 /* Any error is already reported by bio_endio callback. */
1348 drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1354 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1355 * @connection: DRBD connection.
1356 * @epoch: Epoch object.
1359 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1360 struct drbd_epoch *epoch,
1361 enum epoch_event ev)
1364 struct drbd_epoch *next_epoch;
1365 enum finish_epoch rv = FE_STILL_LIVE;
1367 spin_lock(&connection->epoch_lock);
1371 epoch_size = atomic_read(&epoch->epoch_size);
1373 switch (ev & ~EV_CLEANUP) {
1375 atomic_dec(&epoch->active);
1377 case EV_GOT_BARRIER_NR:
1378 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1380 case EV_BECAME_LAST:
1385 if (epoch_size != 0 &&
1386 atomic_read(&epoch->active) == 0 &&
1387 (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1388 if (!(ev & EV_CLEANUP)) {
1389 spin_unlock(&connection->epoch_lock);
1390 drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1391 spin_lock(&connection->epoch_lock);
1394 /* FIXME: dec unacked on connection, once we have
1395 * something to count pending connection packets in. */
1396 if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1397 dec_unacked(epoch->connection);
1400 if (connection->current_epoch != epoch) {
1401 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1402 list_del(&epoch->list);
1403 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1404 connection->epochs--;
1407 if (rv == FE_STILL_LIVE)
1411 atomic_set(&epoch->epoch_size, 0);
1412 /* atomic_set(&epoch->active, 0); is already zero */
1413 if (rv == FE_STILL_LIVE)
1424 spin_unlock(&connection->epoch_lock);
1429 static enum write_ordering_e
1430 max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1432 struct disk_conf *dc;
1434 dc = rcu_dereference(bdev->disk_conf);
1436 if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1438 if (wo == WO_DRAIN_IO && !dc->disk_drain)
1445 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1446 * @wo: Write ordering method to try.
1448 void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1449 enum write_ordering_e wo)
1451 struct drbd_device *device;
1452 enum write_ordering_e pwo;
1454 static char *write_ordering_str[] = {
1456 [WO_DRAIN_IO] = "drain",
1457 [WO_BDEV_FLUSH] = "flush",
1460 pwo = resource->write_ordering;
1461 if (wo != WO_BDEV_FLUSH)
1464 idr_for_each_entry(&resource->devices, device, vnr) {
1465 if (get_ldev(device)) {
1466 wo = max_allowed_wo(device->ldev, wo);
1467 if (device->ldev == bdev)
1474 wo = max_allowed_wo(bdev, wo);
1478 resource->write_ordering = wo;
1479 if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1480 drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1484 * Mapping "discard" to ZEROOUT with UNMAP does not work for us:
1485 * Drivers have to "announce" q->limits.max_write_zeroes_sectors, or it
1486 * will directly go to fallback mode, submitting normal writes, and
1487 * never even try to UNMAP.
1489 * And dm-thin does not do this (yet), mostly because in general it has
1490 * to assume that "skip_block_zeroing" is set. See also:
1491 * https://www.mail-archive.com/dm-devel%40redhat.com/msg07965.html
1492 * https://www.redhat.com/archives/dm-devel/2018-January/msg00271.html
1494 * We *may* ignore the discard-zeroes-data setting, if so configured.
1496 * Assumption is that this "discard_zeroes_data=0" is only because the backend
1497 * may ignore partial unaligned discards.
1499 * LVM/DM thin as of at least
1500 * LVM version: 2.02.115(2)-RHEL7 (2015-01-28)
1501 * Library version: 1.02.93-RHEL7 (2015-01-28)
1502 * Driver version: 4.29.0
1503 * still behaves this way.
1505 * For unaligned (wrt. alignment and granularity) or too small discards,
1506 * we zero-out the initial (and/or) trailing unaligned partial chunks,
1507 * but discard all the aligned full chunks.
1509 * At least for LVM/DM thin, with skip_block_zeroing=false,
1510 * the result is effectively "discard_zeroes_data=1".
1512 /* flags: EE_TRIM|EE_ZEROOUT */
1513 int drbd_issue_discard_or_zero_out(struct drbd_device *device, sector_t start, unsigned int nr_sectors, int flags)
1515 struct block_device *bdev = device->ldev->backing_bdev;
1516 struct request_queue *q = bdev_get_queue(bdev);
1518 unsigned int max_discard_sectors, granularity;
1522 if ((flags & EE_ZEROOUT) || !(flags & EE_TRIM))
1525 /* Zero-sector (unknown) and one-sector granularities are the same. */
1526 granularity = max(q->limits.discard_granularity >> 9, 1U);
1527 alignment = (bdev_discard_alignment(bdev) >> 9) % granularity;
1529 max_discard_sectors = min(q->limits.max_discard_sectors, (1U << 22));
1530 max_discard_sectors -= max_discard_sectors % granularity;
1531 if (unlikely(!max_discard_sectors))
1534 if (nr_sectors < granularity)
1538 if (sector_div(tmp, granularity) != alignment) {
1539 if (nr_sectors < 2*granularity)
1541 /* start + gran - (start + gran - align) % gran */
1542 tmp = start + granularity - alignment;
1543 tmp = start + granularity - sector_div(tmp, granularity);
1546 /* don't flag BLKDEV_ZERO_NOUNMAP, we don't know how many
1547 * layers are below us, some may have smaller granularity */
1548 err |= blkdev_issue_zeroout(bdev, start, nr, GFP_NOIO, 0);
1552 while (nr_sectors >= max_discard_sectors) {
1553 err |= blkdev_issue_discard(bdev, start, max_discard_sectors, GFP_NOIO, 0);
1554 nr_sectors -= max_discard_sectors;
1555 start += max_discard_sectors;
1558 /* max_discard_sectors is unsigned int (and a multiple of
1559 * granularity, we made sure of that above already);
1560 * nr is < max_discard_sectors;
1561 * I don't need sector_div here, even though nr is sector_t */
1563 nr -= (unsigned int)nr % granularity;
1565 err |= blkdev_issue_discard(bdev, start, nr, GFP_NOIO, 0);
1572 err |= blkdev_issue_zeroout(bdev, start, nr_sectors, GFP_NOIO,
1573 (flags & EE_TRIM) ? 0 : BLKDEV_ZERO_NOUNMAP);
1578 static bool can_do_reliable_discards(struct drbd_device *device)
1580 struct request_queue *q = bdev_get_queue(device->ldev->backing_bdev);
1581 struct disk_conf *dc;
1584 if (!blk_queue_discard(q))
1588 dc = rcu_dereference(device->ldev->disk_conf);
1589 can_do = dc->discard_zeroes_if_aligned;
1594 static void drbd_issue_peer_discard_or_zero_out(struct drbd_device *device, struct drbd_peer_request *peer_req)
1596 /* If the backend cannot discard, or does not guarantee
1597 * read-back zeroes in discarded ranges, we fall back to
1598 * zero-out. Unless configuration specifically requested
1600 if (!can_do_reliable_discards(device))
1601 peer_req->flags |= EE_ZEROOUT;
1603 if (drbd_issue_discard_or_zero_out(device, peer_req->i.sector,
1604 peer_req->i.size >> 9, peer_req->flags & (EE_ZEROOUT|EE_TRIM)))
1605 peer_req->flags |= EE_WAS_ERROR;
1606 drbd_endio_write_sec_final(peer_req);
1609 static void drbd_issue_peer_wsame(struct drbd_device *device,
1610 struct drbd_peer_request *peer_req)
1612 struct block_device *bdev = device->ldev->backing_bdev;
1613 sector_t s = peer_req->i.sector;
1614 sector_t nr = peer_req->i.size >> 9;
1615 if (blkdev_issue_write_same(bdev, s, nr, GFP_NOIO, peer_req->pages))
1616 peer_req->flags |= EE_WAS_ERROR;
1617 drbd_endio_write_sec_final(peer_req);
1622 * drbd_submit_peer_request()
1623 * @device: DRBD device.
1624 * @peer_req: peer request
1626 * May spread the pages to multiple bios,
1627 * depending on bio_add_page restrictions.
1629 * Returns 0 if all bios have been submitted,
1630 * -ENOMEM if we could not allocate enough bios,
1631 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1632 * single page to an empty bio (which should never happen and likely indicates
1633 * that the lower level IO stack is in some way broken). This has been observed
1634 * on certain Xen deployments.
1636 /* TODO allocate from our own bio_set. */
1637 int drbd_submit_peer_request(struct drbd_device *device,
1638 struct drbd_peer_request *peer_req,
1639 const unsigned op, const unsigned op_flags,
1640 const int fault_type)
1642 struct bio *bios = NULL;
1644 struct page *page = peer_req->pages;
1645 sector_t sector = peer_req->i.sector;
1646 unsigned data_size = peer_req->i.size;
1647 unsigned n_bios = 0;
1648 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
1651 /* TRIM/DISCARD: for now, always use the helper function
1652 * blkdev_issue_zeroout(..., discard=true).
1653 * It's synchronous, but it does the right thing wrt. bio splitting.
1654 * Correctness first, performance later. Next step is to code an
1655 * asynchronous variant of the same.
1657 if (peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) {
1658 /* wait for all pending IO completions, before we start
1659 * zeroing things out. */
1660 conn_wait_active_ee_empty(peer_req->peer_device->connection);
1661 /* add it to the active list now,
1662 * so we can find it to present it in debugfs */
1663 peer_req->submit_jif = jiffies;
1664 peer_req->flags |= EE_SUBMITTED;
1666 /* If this was a resync request from receive_rs_deallocated(),
1667 * it is already on the sync_ee list */
1668 if (list_empty(&peer_req->w.list)) {
1669 spin_lock_irq(&device->resource->req_lock);
1670 list_add_tail(&peer_req->w.list, &device->active_ee);
1671 spin_unlock_irq(&device->resource->req_lock);
1674 if (peer_req->flags & (EE_TRIM|EE_ZEROOUT))
1675 drbd_issue_peer_discard_or_zero_out(device, peer_req);
1676 else /* EE_WRITE_SAME */
1677 drbd_issue_peer_wsame(device, peer_req);
1681 /* In most cases, we will only need one bio. But in case the lower
1682 * level restrictions happen to be different at this offset on this
1683 * side than those of the sending peer, we may need to submit the
1684 * request in more than one bio.
1686 * Plain bio_alloc is good enough here, this is no DRBD internally
1687 * generated bio, but a bio allocated on behalf of the peer.
1690 bio = bio_alloc(GFP_NOIO, nr_pages);
1692 drbd_err(device, "submit_ee: Allocation of a bio failed (nr_pages=%u)\n", nr_pages);
1695 /* > peer_req->i.sector, unless this is the first bio */
1696 bio->bi_iter.bi_sector = sector;
1697 bio_set_dev(bio, device->ldev->backing_bdev);
1698 bio_set_op_attrs(bio, op, op_flags);
1699 bio->bi_private = peer_req;
1700 bio->bi_end_io = drbd_peer_request_endio;
1702 bio->bi_next = bios;
1706 page_chain_for_each(page) {
1707 unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1708 if (!bio_add_page(bio, page, len, 0))
1714 D_ASSERT(device, data_size == 0);
1715 D_ASSERT(device, page == NULL);
1717 atomic_set(&peer_req->pending_bios, n_bios);
1718 /* for debugfs: update timestamp, mark as submitted */
1719 peer_req->submit_jif = jiffies;
1720 peer_req->flags |= EE_SUBMITTED;
1723 bios = bios->bi_next;
1724 bio->bi_next = NULL;
1726 drbd_submit_bio_noacct(device, fault_type, bio);
1733 bios = bios->bi_next;
1739 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1740 struct drbd_peer_request *peer_req)
1742 struct drbd_interval *i = &peer_req->i;
1744 drbd_remove_interval(&device->write_requests, i);
1745 drbd_clear_interval(i);
1747 /* Wake up any processes waiting for this peer request to complete. */
1749 wake_up(&device->misc_wait);
1752 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1754 struct drbd_peer_device *peer_device;
1758 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1759 struct drbd_device *device = peer_device->device;
1761 kref_get(&device->kref);
1763 drbd_wait_ee_list_empty(device, &device->active_ee);
1764 kref_put(&device->kref, drbd_destroy_device);
1770 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1773 struct p_barrier *p = pi->data;
1774 struct drbd_epoch *epoch;
1776 /* FIXME these are unacked on connection,
1777 * not a specific (peer)device.
1779 connection->current_epoch->barrier_nr = p->barrier;
1780 connection->current_epoch->connection = connection;
1781 rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1783 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1784 * the activity log, which means it would not be resynced in case the
1785 * R_PRIMARY crashes now.
1786 * Therefore we must send the barrier_ack after the barrier request was
1788 switch (connection->resource->write_ordering) {
1790 if (rv == FE_RECYCLED)
1793 /* receiver context, in the writeout path of the other node.
1794 * avoid potential distributed deadlock */
1795 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1799 drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1804 conn_wait_active_ee_empty(connection);
1805 drbd_flush(connection);
1807 if (atomic_read(&connection->current_epoch->epoch_size)) {
1808 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1815 drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1816 connection->resource->write_ordering);
1821 atomic_set(&epoch->epoch_size, 0);
1822 atomic_set(&epoch->active, 0);
1824 spin_lock(&connection->epoch_lock);
1825 if (atomic_read(&connection->current_epoch->epoch_size)) {
1826 list_add(&epoch->list, &connection->current_epoch->list);
1827 connection->current_epoch = epoch;
1828 connection->epochs++;
1830 /* The current_epoch got recycled while we allocated this one... */
1833 spin_unlock(&connection->epoch_lock);
1838 /* quick wrapper in case payload size != request_size (write same) */
1839 static void drbd_csum_ee_size(struct crypto_shash *h,
1840 struct drbd_peer_request *r, void *d,
1841 unsigned int payload_size)
1843 unsigned int tmp = r->i.size;
1844 r->i.size = payload_size;
1845 drbd_csum_ee(h, r, d);
1849 /* used from receive_RSDataReply (recv_resync_read)
1850 * and from receive_Data.
1851 * data_size: actual payload ("data in")
1852 * for normal writes that is bi_size.
1853 * for discards, that is zero.
1854 * for write same, it is logical_block_size.
1855 * both trim and write same have the bi_size ("data len to be affected")
1856 * as extra argument in the packet header.
1858 static struct drbd_peer_request *
1859 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1860 struct packet_info *pi) __must_hold(local)
1862 struct drbd_device *device = peer_device->device;
1863 const sector_t capacity = get_capacity(device->vdisk);
1864 struct drbd_peer_request *peer_req;
1866 int digest_size, err;
1867 unsigned int data_size = pi->size, ds;
1868 void *dig_in = peer_device->connection->int_dig_in;
1869 void *dig_vv = peer_device->connection->int_dig_vv;
1870 unsigned long *data;
1871 struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1872 struct p_trim *zeroes = (pi->cmd == P_ZEROES) ? pi->data : NULL;
1873 struct p_trim *wsame = (pi->cmd == P_WSAME) ? pi->data : NULL;
1876 if (!trim && peer_device->connection->peer_integrity_tfm) {
1877 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1879 * FIXME: Receive the incoming digest into the receive buffer
1880 * here, together with its struct p_data?
1882 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1885 data_size -= digest_size;
1888 /* assume request_size == data_size, but special case trim and wsame. */
1891 if (!expect(data_size == 0))
1893 ds = be32_to_cpu(trim->size);
1894 } else if (zeroes) {
1895 if (!expect(data_size == 0))
1897 ds = be32_to_cpu(zeroes->size);
1899 if (data_size != queue_logical_block_size(device->rq_queue)) {
1900 drbd_err(peer_device, "data size (%u) != drbd logical block size (%u)\n",
1901 data_size, queue_logical_block_size(device->rq_queue));
1904 if (data_size != bdev_logical_block_size(device->ldev->backing_bdev)) {
1905 drbd_err(peer_device, "data size (%u) != backend logical block size (%u)\n",
1906 data_size, bdev_logical_block_size(device->ldev->backing_bdev));
1909 ds = be32_to_cpu(wsame->size);
1912 if (!expect(IS_ALIGNED(ds, 512)))
1914 if (trim || wsame || zeroes) {
1915 if (!expect(ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
1917 } else if (!expect(ds <= DRBD_MAX_BIO_SIZE))
1920 /* even though we trust out peer,
1921 * we sometimes have to double check. */
1922 if (sector + (ds>>9) > capacity) {
1923 drbd_err(device, "request from peer beyond end of local disk: "
1924 "capacity: %llus < sector: %llus + size: %u\n",
1925 (unsigned long long)capacity,
1926 (unsigned long long)sector, ds);
1930 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1931 * "criss-cross" setup, that might cause write-out on some other DRBD,
1932 * which in turn might block on the other node at this very place. */
1933 peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
1937 peer_req->flags |= EE_WRITE;
1939 peer_req->flags |= EE_TRIM;
1943 peer_req->flags |= EE_ZEROOUT;
1947 peer_req->flags |= EE_WRITE_SAME;
1949 /* receive payload size bytes into page chain */
1951 page = peer_req->pages;
1952 page_chain_for_each(page) {
1953 unsigned len = min_t(int, ds, PAGE_SIZE);
1955 err = drbd_recv_all_warn(peer_device->connection, data, len);
1956 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1957 drbd_err(device, "Fault injection: Corrupting data on receive\n");
1958 data[0] = data[0] ^ (unsigned long)-1;
1962 drbd_free_peer_req(device, peer_req);
1969 drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
1970 if (memcmp(dig_in, dig_vv, digest_size)) {
1971 drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1972 (unsigned long long)sector, data_size);
1973 drbd_free_peer_req(device, peer_req);
1977 device->recv_cnt += data_size >> 9;
1981 /* drbd_drain_block() just takes a data block
1982 * out of the socket input buffer, and discards it.
1984 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1993 page = drbd_alloc_pages(peer_device, 1, 1);
1997 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1999 err = drbd_recv_all_warn(peer_device->connection, data, len);
2005 drbd_free_pages(peer_device->device, page, 0);
2009 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
2010 sector_t sector, int data_size)
2012 struct bio_vec bvec;
2013 struct bvec_iter iter;
2015 int digest_size, err, expect;
2016 void *dig_in = peer_device->connection->int_dig_in;
2017 void *dig_vv = peer_device->connection->int_dig_vv;
2020 if (peer_device->connection->peer_integrity_tfm) {
2021 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
2022 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
2025 data_size -= digest_size;
2028 /* optimistically update recv_cnt. if receiving fails below,
2029 * we disconnect anyways, and counters will be reset. */
2030 peer_device->device->recv_cnt += data_size>>9;
2032 bio = req->master_bio;
2033 D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
2035 bio_for_each_segment(bvec, bio, iter) {
2036 void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
2037 expect = min_t(int, data_size, bvec.bv_len);
2038 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
2039 kunmap(bvec.bv_page);
2042 data_size -= expect;
2046 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
2047 if (memcmp(dig_in, dig_vv, digest_size)) {
2048 drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
2053 D_ASSERT(peer_device->device, data_size == 0);
2058 * e_end_resync_block() is called in ack_sender context via
2059 * drbd_finish_peer_reqs().
2061 static int e_end_resync_block(struct drbd_work *w, int unused)
2063 struct drbd_peer_request *peer_req =
2064 container_of(w, struct drbd_peer_request, w);
2065 struct drbd_peer_device *peer_device = peer_req->peer_device;
2066 struct drbd_device *device = peer_device->device;
2067 sector_t sector = peer_req->i.sector;
2070 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2072 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2073 drbd_set_in_sync(device, sector, peer_req->i.size);
2074 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
2076 /* Record failure to sync */
2077 drbd_rs_failed_io(device, sector, peer_req->i.size);
2079 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2081 dec_unacked(device);
2086 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
2087 struct packet_info *pi) __releases(local)
2089 struct drbd_device *device = peer_device->device;
2090 struct drbd_peer_request *peer_req;
2092 peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
2096 dec_rs_pending(device);
2098 inc_unacked(device);
2099 /* corresponding dec_unacked() in e_end_resync_block()
2100 * respective _drbd_clear_done_ee */
2102 peer_req->w.cb = e_end_resync_block;
2103 peer_req->submit_jif = jiffies;
2105 spin_lock_irq(&device->resource->req_lock);
2106 list_add_tail(&peer_req->w.list, &device->sync_ee);
2107 spin_unlock_irq(&device->resource->req_lock);
2109 atomic_add(pi->size >> 9, &device->rs_sect_ev);
2110 if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE, 0,
2111 DRBD_FAULT_RS_WR) == 0)
2114 /* don't care for the reason here */
2115 drbd_err(device, "submit failed, triggering re-connect\n");
2116 spin_lock_irq(&device->resource->req_lock);
2117 list_del(&peer_req->w.list);
2118 spin_unlock_irq(&device->resource->req_lock);
2120 drbd_free_peer_req(device, peer_req);
2126 static struct drbd_request *
2127 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
2128 sector_t sector, bool missing_ok, const char *func)
2130 struct drbd_request *req;
2132 /* Request object according to our peer */
2133 req = (struct drbd_request *)(unsigned long)id;
2134 if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
2137 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
2138 (unsigned long)id, (unsigned long long)sector);
2143 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
2145 struct drbd_peer_device *peer_device;
2146 struct drbd_device *device;
2147 struct drbd_request *req;
2150 struct p_data *p = pi->data;
2152 peer_device = conn_peer_device(connection, pi->vnr);
2155 device = peer_device->device;
2157 sector = be64_to_cpu(p->sector);
2159 spin_lock_irq(&device->resource->req_lock);
2160 req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
2161 spin_unlock_irq(&device->resource->req_lock);
2165 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
2166 * special casing it there for the various failure cases.
2167 * still no race with drbd_fail_pending_reads */
2168 err = recv_dless_read(peer_device, req, sector, pi->size);
2170 req_mod(req, DATA_RECEIVED);
2171 /* else: nothing. handled from drbd_disconnect...
2172 * I don't think we may complete this just yet
2173 * in case we are "on-disconnect: freeze" */
2178 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
2180 struct drbd_peer_device *peer_device;
2181 struct drbd_device *device;
2184 struct p_data *p = pi->data;
2186 peer_device = conn_peer_device(connection, pi->vnr);
2189 device = peer_device->device;
2191 sector = be64_to_cpu(p->sector);
2192 D_ASSERT(device, p->block_id == ID_SYNCER);
2194 if (get_ldev(device)) {
2195 /* data is submitted to disk within recv_resync_read.
2196 * corresponding put_ldev done below on error,
2197 * or in drbd_peer_request_endio. */
2198 err = recv_resync_read(peer_device, sector, pi);
2200 if (__ratelimit(&drbd_ratelimit_state))
2201 drbd_err(device, "Can not write resync data to local disk.\n");
2203 err = drbd_drain_block(peer_device, pi->size);
2205 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2208 atomic_add(pi->size >> 9, &device->rs_sect_in);
2213 static void restart_conflicting_writes(struct drbd_device *device,
2214 sector_t sector, int size)
2216 struct drbd_interval *i;
2217 struct drbd_request *req;
2219 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2222 req = container_of(i, struct drbd_request, i);
2223 if (req->rq_state & RQ_LOCAL_PENDING ||
2224 !(req->rq_state & RQ_POSTPONED))
2226 /* as it is RQ_POSTPONED, this will cause it to
2227 * be queued on the retry workqueue. */
2228 __req_mod(req, CONFLICT_RESOLVED, NULL);
2233 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2235 static int e_end_block(struct drbd_work *w, int cancel)
2237 struct drbd_peer_request *peer_req =
2238 container_of(w, struct drbd_peer_request, w);
2239 struct drbd_peer_device *peer_device = peer_req->peer_device;
2240 struct drbd_device *device = peer_device->device;
2241 sector_t sector = peer_req->i.sector;
2244 if (peer_req->flags & EE_SEND_WRITE_ACK) {
2245 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2246 pcmd = (device->state.conn >= C_SYNC_SOURCE &&
2247 device->state.conn <= C_PAUSED_SYNC_T &&
2248 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
2249 P_RS_WRITE_ACK : P_WRITE_ACK;
2250 err = drbd_send_ack(peer_device, pcmd, peer_req);
2251 if (pcmd == P_RS_WRITE_ACK)
2252 drbd_set_in_sync(device, sector, peer_req->i.size);
2254 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2255 /* we expect it to be marked out of sync anyways...
2256 * maybe assert this? */
2258 dec_unacked(device);
2261 /* we delete from the conflict detection hash _after_ we sent out the
2262 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
2263 if (peer_req->flags & EE_IN_INTERVAL_TREE) {
2264 spin_lock_irq(&device->resource->req_lock);
2265 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
2266 drbd_remove_epoch_entry_interval(device, peer_req);
2267 if (peer_req->flags & EE_RESTART_REQUESTS)
2268 restart_conflicting_writes(device, sector, peer_req->i.size);
2269 spin_unlock_irq(&device->resource->req_lock);
2271 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2273 drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2278 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2280 struct drbd_peer_request *peer_req =
2281 container_of(w, struct drbd_peer_request, w);
2282 struct drbd_peer_device *peer_device = peer_req->peer_device;
2285 err = drbd_send_ack(peer_device, ack, peer_req);
2286 dec_unacked(peer_device->device);
2291 static int e_send_superseded(struct drbd_work *w, int unused)
2293 return e_send_ack(w, P_SUPERSEDED);
2296 static int e_send_retry_write(struct drbd_work *w, int unused)
2298 struct drbd_peer_request *peer_req =
2299 container_of(w, struct drbd_peer_request, w);
2300 struct drbd_connection *connection = peer_req->peer_device->connection;
2302 return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2303 P_RETRY_WRITE : P_SUPERSEDED);
2306 static bool seq_greater(u32 a, u32 b)
2309 * We assume 32-bit wrap-around here.
2310 * For 24-bit wrap-around, we would have to shift:
2313 return (s32)a - (s32)b > 0;
2316 static u32 seq_max(u32 a, u32 b)
2318 return seq_greater(a, b) ? a : b;
2321 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2323 struct drbd_device *device = peer_device->device;
2324 unsigned int newest_peer_seq;
2326 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2327 spin_lock(&device->peer_seq_lock);
2328 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2329 device->peer_seq = newest_peer_seq;
2330 spin_unlock(&device->peer_seq_lock);
2331 /* wake up only if we actually changed device->peer_seq */
2332 if (peer_seq == newest_peer_seq)
2333 wake_up(&device->seq_wait);
2337 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2339 return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2342 /* maybe change sync_ee into interval trees as well? */
2343 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2345 struct drbd_peer_request *rs_req;
2348 spin_lock_irq(&device->resource->req_lock);
2349 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2350 if (overlaps(peer_req->i.sector, peer_req->i.size,
2351 rs_req->i.sector, rs_req->i.size)) {
2356 spin_unlock_irq(&device->resource->req_lock);
2361 /* Called from receive_Data.
2362 * Synchronize packets on sock with packets on msock.
2364 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2365 * packet traveling on msock, they are still processed in the order they have
2368 * Note: we don't care for Ack packets overtaking P_DATA packets.
2370 * In case packet_seq is larger than device->peer_seq number, there are
2371 * outstanding packets on the msock. We wait for them to arrive.
2372 * In case we are the logically next packet, we update device->peer_seq
2373 * ourselves. Correctly handles 32bit wrap around.
2375 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2376 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2377 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2378 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2380 * returns 0 if we may process the packet,
2381 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2382 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2384 struct drbd_device *device = peer_device->device;
2389 if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2392 spin_lock(&device->peer_seq_lock);
2394 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2395 device->peer_seq = seq_max(device->peer_seq, peer_seq);
2399 if (signal_pending(current)) {
2405 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2411 /* Only need to wait if two_primaries is enabled */
2412 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2413 spin_unlock(&device->peer_seq_lock);
2415 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2417 timeout = schedule_timeout(timeout);
2418 spin_lock(&device->peer_seq_lock);
2421 drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2425 spin_unlock(&device->peer_seq_lock);
2426 finish_wait(&device->seq_wait, &wait);
2430 /* see also bio_flags_to_wire()
2431 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2432 * flags and back. We may replicate to other kernel versions. */
2433 static unsigned long wire_flags_to_bio_flags(u32 dpf)
2435 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2436 (dpf & DP_FUA ? REQ_FUA : 0) |
2437 (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2440 static unsigned long wire_flags_to_bio_op(u32 dpf)
2442 if (dpf & DP_ZEROES)
2443 return REQ_OP_WRITE_ZEROES;
2444 if (dpf & DP_DISCARD)
2445 return REQ_OP_DISCARD;
2447 return REQ_OP_WRITE_SAME;
2449 return REQ_OP_WRITE;
2452 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2455 struct drbd_interval *i;
2458 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2459 struct drbd_request *req;
2460 struct bio_and_error m;
2464 req = container_of(i, struct drbd_request, i);
2465 if (!(req->rq_state & RQ_POSTPONED))
2467 req->rq_state &= ~RQ_POSTPONED;
2468 __req_mod(req, NEG_ACKED, &m);
2469 spin_unlock_irq(&device->resource->req_lock);
2471 complete_master_bio(device, &m);
2472 spin_lock_irq(&device->resource->req_lock);
2477 static int handle_write_conflicts(struct drbd_device *device,
2478 struct drbd_peer_request *peer_req)
2480 struct drbd_connection *connection = peer_req->peer_device->connection;
2481 bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2482 sector_t sector = peer_req->i.sector;
2483 const unsigned int size = peer_req->i.size;
2484 struct drbd_interval *i;
2489 * Inserting the peer request into the write_requests tree will prevent
2490 * new conflicting local requests from being added.
2492 drbd_insert_interval(&device->write_requests, &peer_req->i);
2495 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2496 if (i == &peer_req->i)
2503 * Our peer has sent a conflicting remote request; this
2504 * should not happen in a two-node setup. Wait for the
2505 * earlier peer request to complete.
2507 err = drbd_wait_misc(device, i);
2513 equal = i->sector == sector && i->size == size;
2514 if (resolve_conflicts) {
2516 * If the peer request is fully contained within the
2517 * overlapping request, it can be considered overwritten
2518 * and thus superseded; otherwise, it will be retried
2519 * once all overlapping requests have completed.
2521 bool superseded = i->sector <= sector && i->sector +
2522 (i->size >> 9) >= sector + (size >> 9);
2525 drbd_alert(device, "Concurrent writes detected: "
2526 "local=%llus +%u, remote=%llus +%u, "
2527 "assuming %s came first\n",
2528 (unsigned long long)i->sector, i->size,
2529 (unsigned long long)sector, size,
2530 superseded ? "local" : "remote");
2532 peer_req->w.cb = superseded ? e_send_superseded :
2534 list_add_tail(&peer_req->w.list, &device->done_ee);
2535 queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2540 struct drbd_request *req =
2541 container_of(i, struct drbd_request, i);
2544 drbd_alert(device, "Concurrent writes detected: "
2545 "local=%llus +%u, remote=%llus +%u\n",
2546 (unsigned long long)i->sector, i->size,
2547 (unsigned long long)sector, size);
2549 if (req->rq_state & RQ_LOCAL_PENDING ||
2550 !(req->rq_state & RQ_POSTPONED)) {
2552 * Wait for the node with the discard flag to
2553 * decide if this request has been superseded
2554 * or needs to be retried.
2555 * Requests that have been superseded will
2556 * disappear from the write_requests tree.
2558 * In addition, wait for the conflicting
2559 * request to finish locally before submitting
2560 * the conflicting peer request.
2562 err = drbd_wait_misc(device, &req->i);
2564 _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2565 fail_postponed_requests(device, sector, size);
2571 * Remember to restart the conflicting requests after
2572 * the new peer request has completed.
2574 peer_req->flags |= EE_RESTART_REQUESTS;
2581 drbd_remove_epoch_entry_interval(device, peer_req);
2585 /* mirrored write */
2586 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2588 struct drbd_peer_device *peer_device;
2589 struct drbd_device *device;
2590 struct net_conf *nc;
2592 struct drbd_peer_request *peer_req;
2593 struct p_data *p = pi->data;
2594 u32 peer_seq = be32_to_cpu(p->seq_num);
2599 peer_device = conn_peer_device(connection, pi->vnr);
2602 device = peer_device->device;
2604 if (!get_ldev(device)) {
2607 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2608 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2609 atomic_inc(&connection->current_epoch->epoch_size);
2610 err2 = drbd_drain_block(peer_device, pi->size);
2617 * Corresponding put_ldev done either below (on various errors), or in
2618 * drbd_peer_request_endio, if we successfully submit the data at the
2619 * end of this function.
2622 sector = be64_to_cpu(p->sector);
2623 peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2629 peer_req->w.cb = e_end_block;
2630 peer_req->submit_jif = jiffies;
2631 peer_req->flags |= EE_APPLICATION;
2633 dp_flags = be32_to_cpu(p->dp_flags);
2634 op = wire_flags_to_bio_op(dp_flags);
2635 op_flags = wire_flags_to_bio_flags(dp_flags);
2636 if (pi->cmd == P_TRIM) {
2637 D_ASSERT(peer_device, peer_req->i.size > 0);
2638 D_ASSERT(peer_device, op == REQ_OP_DISCARD);
2639 D_ASSERT(peer_device, peer_req->pages == NULL);
2640 /* need to play safe: an older DRBD sender
2641 * may mean zero-out while sending P_TRIM. */
2642 if (0 == (connection->agreed_features & DRBD_FF_WZEROES))
2643 peer_req->flags |= EE_ZEROOUT;
2644 } else if (pi->cmd == P_ZEROES) {
2645 D_ASSERT(peer_device, peer_req->i.size > 0);
2646 D_ASSERT(peer_device, op == REQ_OP_WRITE_ZEROES);
2647 D_ASSERT(peer_device, peer_req->pages == NULL);
2648 /* Do (not) pass down BLKDEV_ZERO_NOUNMAP? */
2649 if (dp_flags & DP_DISCARD)
2650 peer_req->flags |= EE_TRIM;
2651 } else if (peer_req->pages == NULL) {
2652 D_ASSERT(device, peer_req->i.size == 0);
2653 D_ASSERT(device, dp_flags & DP_FLUSH);
2656 if (dp_flags & DP_MAY_SET_IN_SYNC)
2657 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2659 spin_lock(&connection->epoch_lock);
2660 peer_req->epoch = connection->current_epoch;
2661 atomic_inc(&peer_req->epoch->epoch_size);
2662 atomic_inc(&peer_req->epoch->active);
2663 spin_unlock(&connection->epoch_lock);
2666 nc = rcu_dereference(peer_device->connection->net_conf);
2667 tp = nc->two_primaries;
2668 if (peer_device->connection->agreed_pro_version < 100) {
2669 switch (nc->wire_protocol) {
2671 dp_flags |= DP_SEND_WRITE_ACK;
2674 dp_flags |= DP_SEND_RECEIVE_ACK;
2680 if (dp_flags & DP_SEND_WRITE_ACK) {
2681 peer_req->flags |= EE_SEND_WRITE_ACK;
2682 inc_unacked(device);
2683 /* corresponding dec_unacked() in e_end_block()
2684 * respective _drbd_clear_done_ee */
2687 if (dp_flags & DP_SEND_RECEIVE_ACK) {
2688 /* I really don't like it that the receiver thread
2689 * sends on the msock, but anyways */
2690 drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2694 /* two primaries implies protocol C */
2695 D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2696 peer_req->flags |= EE_IN_INTERVAL_TREE;
2697 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2699 goto out_interrupted;
2700 spin_lock_irq(&device->resource->req_lock);
2701 err = handle_write_conflicts(device, peer_req);
2703 spin_unlock_irq(&device->resource->req_lock);
2704 if (err == -ENOENT) {
2708 goto out_interrupted;
2711 update_peer_seq(peer_device, peer_seq);
2712 spin_lock_irq(&device->resource->req_lock);
2714 /* TRIM and WRITE_SAME are processed synchronously,
2715 * we wait for all pending requests, respectively wait for
2716 * active_ee to become empty in drbd_submit_peer_request();
2717 * better not add ourselves here. */
2718 if ((peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) == 0)
2719 list_add_tail(&peer_req->w.list, &device->active_ee);
2720 spin_unlock_irq(&device->resource->req_lock);
2722 if (device->state.conn == C_SYNC_TARGET)
2723 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2725 if (device->state.pdsk < D_INCONSISTENT) {
2726 /* In case we have the only disk of the cluster, */
2727 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2728 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2729 drbd_al_begin_io(device, &peer_req->i);
2730 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2733 err = drbd_submit_peer_request(device, peer_req, op, op_flags,
2738 /* don't care for the reason here */
2739 drbd_err(device, "submit failed, triggering re-connect\n");
2740 spin_lock_irq(&device->resource->req_lock);
2741 list_del(&peer_req->w.list);
2742 drbd_remove_epoch_entry_interval(device, peer_req);
2743 spin_unlock_irq(&device->resource->req_lock);
2744 if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2745 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2746 drbd_al_complete_io(device, &peer_req->i);
2750 drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
2752 drbd_free_peer_req(device, peer_req);
2756 /* We may throttle resync, if the lower device seems to be busy,
2757 * and current sync rate is above c_min_rate.
2759 * To decide whether or not the lower device is busy, we use a scheme similar
2760 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2761 * (more than 64 sectors) of activity we cannot account for with our own resync
2762 * activity, it obviously is "busy".
2764 * The current sync rate used here uses only the most recent two step marks,
2765 * to have a short time average so we can react faster.
2767 bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2768 bool throttle_if_app_is_waiting)
2770 struct lc_element *tmp;
2771 bool throttle = drbd_rs_c_min_rate_throttle(device);
2773 if (!throttle || throttle_if_app_is_waiting)
2776 spin_lock_irq(&device->al_lock);
2777 tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2779 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2780 if (test_bit(BME_PRIORITY, &bm_ext->flags))
2782 /* Do not slow down if app IO is already waiting for this extent,
2783 * and our progress is necessary for application IO to complete. */
2785 spin_unlock_irq(&device->al_lock);
2790 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2792 struct gendisk *disk = device->ldev->backing_bdev->bd_disk;
2793 unsigned long db, dt, dbdt;
2794 unsigned int c_min_rate;
2798 c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2801 /* feature disabled? */
2802 if (c_min_rate == 0)
2805 curr_events = (int)part_stat_read_accum(disk->part0, sectors) -
2806 atomic_read(&device->rs_sect_ev);
2808 if (atomic_read(&device->ap_actlog_cnt)
2809 || curr_events - device->rs_last_events > 64) {
2810 unsigned long rs_left;
2813 device->rs_last_events = curr_events;
2815 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2817 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2819 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2820 rs_left = device->ov_left;
2822 rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2824 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2827 db = device->rs_mark_left[i] - rs_left;
2828 dbdt = Bit2KB(db/dt);
2830 if (dbdt > c_min_rate)
2836 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2838 struct drbd_peer_device *peer_device;
2839 struct drbd_device *device;
2842 struct drbd_peer_request *peer_req;
2843 struct digest_info *di = NULL;
2845 unsigned int fault_type;
2846 struct p_block_req *p = pi->data;
2848 peer_device = conn_peer_device(connection, pi->vnr);
2851 device = peer_device->device;
2852 capacity = get_capacity(device->vdisk);
2854 sector = be64_to_cpu(p->sector);
2855 size = be32_to_cpu(p->blksize);
2857 if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2858 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2859 (unsigned long long)sector, size);
2862 if (sector + (size>>9) > capacity) {
2863 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2864 (unsigned long long)sector, size);
2868 if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2871 case P_DATA_REQUEST:
2872 drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2875 case P_RS_DATA_REQUEST:
2876 case P_CSUM_RS_REQUEST:
2878 drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2882 dec_rs_pending(device);
2883 drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2888 if (verb && __ratelimit(&drbd_ratelimit_state))
2889 drbd_err(device, "Can not satisfy peer's read request, "
2890 "no local data.\n");
2892 /* drain possibly payload */
2893 return drbd_drain_block(peer_device, pi->size);
2896 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2897 * "criss-cross" setup, that might cause write-out on some other DRBD,
2898 * which in turn might block on the other node at this very place. */
2899 peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2907 case P_DATA_REQUEST:
2908 peer_req->w.cb = w_e_end_data_req;
2909 fault_type = DRBD_FAULT_DT_RD;
2910 /* application IO, don't drbd_rs_begin_io */
2911 peer_req->flags |= EE_APPLICATION;
2915 /* If at some point in the future we have a smart way to
2916 find out if this data block is completely deallocated,
2917 then we would do something smarter here than reading
2919 peer_req->flags |= EE_RS_THIN_REQ;
2921 case P_RS_DATA_REQUEST:
2922 peer_req->w.cb = w_e_end_rsdata_req;
2923 fault_type = DRBD_FAULT_RS_RD;
2924 /* used in the sector offset progress display */
2925 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2929 case P_CSUM_RS_REQUEST:
2930 fault_type = DRBD_FAULT_RS_RD;
2931 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2935 di->digest_size = pi->size;
2936 di->digest = (((char *)di)+sizeof(struct digest_info));
2938 peer_req->digest = di;
2939 peer_req->flags |= EE_HAS_DIGEST;
2941 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2944 if (pi->cmd == P_CSUM_RS_REQUEST) {
2945 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2946 peer_req->w.cb = w_e_end_csum_rs_req;
2947 /* used in the sector offset progress display */
2948 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2949 /* remember to report stats in drbd_resync_finished */
2950 device->use_csums = true;
2951 } else if (pi->cmd == P_OV_REPLY) {
2952 /* track progress, we may need to throttle */
2953 atomic_add(size >> 9, &device->rs_sect_in);
2954 peer_req->w.cb = w_e_end_ov_reply;
2955 dec_rs_pending(device);
2956 /* drbd_rs_begin_io done when we sent this request,
2957 * but accounting still needs to be done. */
2958 goto submit_for_resync;
2963 if (device->ov_start_sector == ~(sector_t)0 &&
2964 peer_device->connection->agreed_pro_version >= 90) {
2965 unsigned long now = jiffies;
2967 device->ov_start_sector = sector;
2968 device->ov_position = sector;
2969 device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2970 device->rs_total = device->ov_left;
2971 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2972 device->rs_mark_left[i] = device->ov_left;
2973 device->rs_mark_time[i] = now;
2975 drbd_info(device, "Online Verify start sector: %llu\n",
2976 (unsigned long long)sector);
2978 peer_req->w.cb = w_e_end_ov_req;
2979 fault_type = DRBD_FAULT_RS_RD;
2986 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2987 * wrt the receiver, but it is not as straightforward as it may seem.
2988 * Various places in the resync start and stop logic assume resync
2989 * requests are processed in order, requeuing this on the worker thread
2990 * introduces a bunch of new code for synchronization between threads.
2992 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2993 * "forever", throttling after drbd_rs_begin_io will lock that extent
2994 * for application writes for the same time. For now, just throttle
2995 * here, where the rest of the code expects the receiver to sleep for
2999 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
3000 * this defers syncer requests for some time, before letting at least
3001 * on request through. The resync controller on the receiving side
3002 * will adapt to the incoming rate accordingly.
3004 * We cannot throttle here if remote is Primary/SyncTarget:
3005 * we would also throttle its application reads.
3006 * In that case, throttling is done on the SyncTarget only.
3009 /* Even though this may be a resync request, we do add to "read_ee";
3010 * "sync_ee" is only used for resync WRITEs.
3011 * Add to list early, so debugfs can find this request
3012 * even if we have to sleep below. */
3013 spin_lock_irq(&device->resource->req_lock);
3014 list_add_tail(&peer_req->w.list, &device->read_ee);
3015 spin_unlock_irq(&device->resource->req_lock);
3017 update_receiver_timing_details(connection, drbd_rs_should_slow_down);
3018 if (device->state.peer != R_PRIMARY
3019 && drbd_rs_should_slow_down(device, sector, false))
3020 schedule_timeout_uninterruptible(HZ/10);
3021 update_receiver_timing_details(connection, drbd_rs_begin_io);
3022 if (drbd_rs_begin_io(device, sector))
3026 atomic_add(size >> 9, &device->rs_sect_ev);
3029 update_receiver_timing_details(connection, drbd_submit_peer_request);
3030 inc_unacked(device);
3031 if (drbd_submit_peer_request(device, peer_req, REQ_OP_READ, 0,
3035 /* don't care for the reason here */
3036 drbd_err(device, "submit failed, triggering re-connect\n");
3039 spin_lock_irq(&device->resource->req_lock);
3040 list_del(&peer_req->w.list);
3041 spin_unlock_irq(&device->resource->req_lock);
3042 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
3045 drbd_free_peer_req(device, peer_req);
3050 * drbd_asb_recover_0p - Recover after split-brain with no remaining primaries
3052 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
3054 struct drbd_device *device = peer_device->device;
3055 int self, peer, rv = -100;
3056 unsigned long ch_self, ch_peer;
3057 enum drbd_after_sb_p after_sb_0p;
3059 self = device->ldev->md.uuid[UI_BITMAP] & 1;
3060 peer = device->p_uuid[UI_BITMAP] & 1;
3062 ch_peer = device->p_uuid[UI_SIZE];
3063 ch_self = device->comm_bm_set;
3066 after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
3068 switch (after_sb_0p) {
3070 case ASB_DISCARD_SECONDARY:
3071 case ASB_CALL_HELPER:
3073 drbd_err(device, "Configuration error.\n");
3075 case ASB_DISCONNECT:
3077 case ASB_DISCARD_YOUNGER_PRI:
3078 if (self == 0 && peer == 1) {
3082 if (self == 1 && peer == 0) {
3086 fallthrough; /* to one of the other strategies */
3087 case ASB_DISCARD_OLDER_PRI:
3088 if (self == 0 && peer == 1) {
3092 if (self == 1 && peer == 0) {
3096 /* Else fall through to one of the other strategies... */
3097 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
3098 "Using discard-least-changes instead\n");
3100 case ASB_DISCARD_ZERO_CHG:
3101 if (ch_peer == 0 && ch_self == 0) {
3102 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3106 if (ch_peer == 0) { rv = 1; break; }
3107 if (ch_self == 0) { rv = -1; break; }
3109 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
3112 case ASB_DISCARD_LEAST_CHG:
3113 if (ch_self < ch_peer)
3115 else if (ch_self > ch_peer)
3117 else /* ( ch_self == ch_peer ) */
3118 /* Well, then use something else. */
3119 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3122 case ASB_DISCARD_LOCAL:
3125 case ASB_DISCARD_REMOTE:
3133 * drbd_asb_recover_1p - Recover after split-brain with one remaining primary
3135 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
3137 struct drbd_device *device = peer_device->device;
3139 enum drbd_after_sb_p after_sb_1p;
3142 after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
3144 switch (after_sb_1p) {
3145 case ASB_DISCARD_YOUNGER_PRI:
3146 case ASB_DISCARD_OLDER_PRI:
3147 case ASB_DISCARD_LEAST_CHG:
3148 case ASB_DISCARD_LOCAL:
3149 case ASB_DISCARD_REMOTE:
3150 case ASB_DISCARD_ZERO_CHG:
3151 drbd_err(device, "Configuration error.\n");
3153 case ASB_DISCONNECT:
3156 hg = drbd_asb_recover_0p(peer_device);
3157 if (hg == -1 && device->state.role == R_SECONDARY)
3159 if (hg == 1 && device->state.role == R_PRIMARY)
3163 rv = drbd_asb_recover_0p(peer_device);
3165 case ASB_DISCARD_SECONDARY:
3166 return device->state.role == R_PRIMARY ? 1 : -1;
3167 case ASB_CALL_HELPER:
3168 hg = drbd_asb_recover_0p(peer_device);
3169 if (hg == -1 && device->state.role == R_PRIMARY) {
3170 enum drbd_state_rv rv2;
3172 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3173 * we might be here in C_WF_REPORT_PARAMS which is transient.
3174 * we do not need to wait for the after state change work either. */
3175 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3176 if (rv2 != SS_SUCCESS) {
3177 drbd_khelper(device, "pri-lost-after-sb");
3179 drbd_warn(device, "Successfully gave up primary role.\n");
3190 * drbd_asb_recover_2p - Recover after split-brain with two remaining primaries
3192 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
3194 struct drbd_device *device = peer_device->device;
3196 enum drbd_after_sb_p after_sb_2p;
3199 after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
3201 switch (after_sb_2p) {
3202 case ASB_DISCARD_YOUNGER_PRI:
3203 case ASB_DISCARD_OLDER_PRI:
3204 case ASB_DISCARD_LEAST_CHG:
3205 case ASB_DISCARD_LOCAL:
3206 case ASB_DISCARD_REMOTE:
3208 case ASB_DISCARD_SECONDARY:
3209 case ASB_DISCARD_ZERO_CHG:
3210 drbd_err(device, "Configuration error.\n");
3213 rv = drbd_asb_recover_0p(peer_device);
3215 case ASB_DISCONNECT:
3217 case ASB_CALL_HELPER:
3218 hg = drbd_asb_recover_0p(peer_device);
3220 enum drbd_state_rv rv2;
3222 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3223 * we might be here in C_WF_REPORT_PARAMS which is transient.
3224 * we do not need to wait for the after state change work either. */
3225 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3226 if (rv2 != SS_SUCCESS) {
3227 drbd_khelper(device, "pri-lost-after-sb");
3229 drbd_warn(device, "Successfully gave up primary role.\n");
3239 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
3240 u64 bits, u64 flags)
3243 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
3246 drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
3248 (unsigned long long)uuid[UI_CURRENT],
3249 (unsigned long long)uuid[UI_BITMAP],
3250 (unsigned long long)uuid[UI_HISTORY_START],
3251 (unsigned long long)uuid[UI_HISTORY_END],
3252 (unsigned long long)bits,
3253 (unsigned long long)flags);
3257 100 after split brain try auto recover
3258 2 C_SYNC_SOURCE set BitMap
3259 1 C_SYNC_SOURCE use BitMap
3261 -1 C_SYNC_TARGET use BitMap
3262 -2 C_SYNC_TARGET set BitMap
3263 -100 after split brain, disconnect
3264 -1000 unrelated data
3265 -1091 requires proto 91
3266 -1096 requires proto 96
3269 static int drbd_uuid_compare(struct drbd_device *const device, enum drbd_role const peer_role, int *rule_nr) __must_hold(local)
3271 struct drbd_peer_device *const peer_device = first_peer_device(device);
3272 struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
3276 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3277 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3280 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
3284 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
3285 peer != UUID_JUST_CREATED)
3289 if (self != UUID_JUST_CREATED &&
3290 (peer == UUID_JUST_CREATED || peer == (u64)0))
3294 int rct, dc; /* roles at crash time */
3296 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3298 if (connection->agreed_pro_version < 91)
3301 if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3302 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3303 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3304 drbd_uuid_move_history(device);
3305 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3306 device->ldev->md.uuid[UI_BITMAP] = 0;
3308 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3309 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3312 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3319 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3321 if (connection->agreed_pro_version < 91)
3324 if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3325 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3326 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3328 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3329 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3330 device->p_uuid[UI_BITMAP] = 0UL;
3332 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3335 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3342 /* Common power [off|failure] */
3343 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3344 (device->p_uuid[UI_FLAGS] & 2);
3345 /* lowest bit is set when we were primary,
3346 * next bit (weight 2) is set when peer was primary */
3349 /* Neither has the "crashed primary" flag set,
3350 * only a replication link hickup. */
3354 /* Current UUID equal and no bitmap uuid; does not necessarily
3355 * mean this was a "simultaneous hard crash", maybe IO was
3356 * frozen, so no UUID-bump happened.
3357 * This is a protocol change, overload DRBD_FF_WSAME as flag
3358 * for "new-enough" peer DRBD version. */
3359 if (device->state.role == R_PRIMARY || peer_role == R_PRIMARY) {
3361 if (!(connection->agreed_features & DRBD_FF_WSAME)) {
3362 drbd_warn(peer_device, "Equivalent unrotated UUIDs, but current primary present.\n");
3363 return -(0x10000 | PRO_VERSION_MAX | (DRBD_FF_WSAME << 8));
3365 if (device->state.role == R_PRIMARY && peer_role == R_PRIMARY) {
3366 /* At least one has the "crashed primary" bit set,
3367 * both are primary now, but neither has rotated its UUIDs?
3368 * "Can not happen." */
3369 drbd_err(peer_device, "Equivalent unrotated UUIDs, but both are primary. Can not resolve this.\n");
3372 if (device->state.role == R_PRIMARY)
3377 /* Both are secondary.
3378 * Really looks like recovery from simultaneous hard crash.
3379 * Check which had been primary before, and arbitrate. */
3381 case 0: /* !self_pri && !peer_pri */ return 0; /* already handled */
3382 case 1: /* self_pri && !peer_pri */ return 1;
3383 case 2: /* !self_pri && peer_pri */ return -1;
3384 case 3: /* self_pri && peer_pri */
3385 dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3391 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3396 peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3398 if (connection->agreed_pro_version < 96 ?
3399 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3400 (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3401 peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3402 /* The last P_SYNC_UUID did not get though. Undo the last start of
3403 resync as sync source modifications of the peer's UUIDs. */
3405 if (connection->agreed_pro_version < 91)
3408 device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3409 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3411 drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3412 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3419 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3420 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3421 peer = device->p_uuid[i] & ~((u64)1);
3427 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3428 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3433 self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3435 if (connection->agreed_pro_version < 96 ?
3436 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3437 (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3438 self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3439 /* The last P_SYNC_UUID did not get though. Undo the last start of
3440 resync as sync source modifications of our UUIDs. */
3442 if (connection->agreed_pro_version < 91)
3445 __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3446 __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3448 drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3449 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3450 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3458 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3459 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3460 self = device->ldev->md.uuid[i] & ~((u64)1);
3466 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3467 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3468 if (self == peer && self != ((u64)0))
3472 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3473 self = device->ldev->md.uuid[i] & ~((u64)1);
3474 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3475 peer = device->p_uuid[j] & ~((u64)1);
3484 /* drbd_sync_handshake() returns the new conn state on success, or
3485 CONN_MASK (-1) on failure.
3487 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3488 enum drbd_role peer_role,
3489 enum drbd_disk_state peer_disk) __must_hold(local)
3491 struct drbd_device *device = peer_device->device;
3492 enum drbd_conns rv = C_MASK;
3493 enum drbd_disk_state mydisk;
3494 struct net_conf *nc;
3495 int hg, rule_nr, rr_conflict, tentative, always_asbp;
3497 mydisk = device->state.disk;
3498 if (mydisk == D_NEGOTIATING)
3499 mydisk = device->new_state_tmp.disk;
3501 drbd_info(device, "drbd_sync_handshake:\n");
3503 spin_lock_irq(&device->ldev->md.uuid_lock);
3504 drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3505 drbd_uuid_dump(device, "peer", device->p_uuid,
3506 device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3508 hg = drbd_uuid_compare(device, peer_role, &rule_nr);
3509 spin_unlock_irq(&device->ldev->md.uuid_lock);
3511 drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3514 drbd_alert(device, "Unrelated data, aborting!\n");
3517 if (hg < -0x10000) {
3521 fflags = (hg >> 8) & 0xff;
3522 drbd_alert(device, "To resolve this both sides have to support at least protocol %d and feature flags 0x%x\n",
3527 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3531 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3532 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
3533 int f = (hg == -100) || abs(hg) == 2;
3534 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3537 drbd_info(device, "Becoming sync %s due to disk states.\n",
3538 hg > 0 ? "source" : "target");
3542 drbd_khelper(device, "initial-split-brain");
3545 nc = rcu_dereference(peer_device->connection->net_conf);
3546 always_asbp = nc->always_asbp;
3547 rr_conflict = nc->rr_conflict;
3548 tentative = nc->tentative;
3551 if (hg == 100 || (hg == -100 && always_asbp)) {
3552 int pcount = (device->state.role == R_PRIMARY)
3553 + (peer_role == R_PRIMARY);
3554 int forced = (hg == -100);
3558 hg = drbd_asb_recover_0p(peer_device);
3561 hg = drbd_asb_recover_1p(peer_device);
3564 hg = drbd_asb_recover_2p(peer_device);
3567 if (abs(hg) < 100) {
3568 drbd_warn(device, "Split-Brain detected, %d primaries, "
3569 "automatically solved. Sync from %s node\n",
3570 pcount, (hg < 0) ? "peer" : "this");
3572 drbd_warn(device, "Doing a full sync, since"
3573 " UUIDs where ambiguous.\n");
3580 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3582 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3586 drbd_warn(device, "Split-Brain detected, manually solved. "
3587 "Sync from %s node\n",
3588 (hg < 0) ? "peer" : "this");
3592 /* FIXME this log message is not correct if we end up here
3593 * after an attempted attach on a diskless node.
3594 * We just refuse to attach -- well, we drop the "connection"
3595 * to that disk, in a way... */
3596 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3597 drbd_khelper(device, "split-brain");
3601 if (hg > 0 && mydisk <= D_INCONSISTENT) {
3602 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3606 if (hg < 0 && /* by intention we do not use mydisk here. */
3607 device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3608 switch (rr_conflict) {
3609 case ASB_CALL_HELPER:
3610 drbd_khelper(device, "pri-lost");
3612 case ASB_DISCONNECT:
3613 drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3616 drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3621 if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3623 drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3625 drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3626 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3627 abs(hg) >= 2 ? "full" : "bit-map based");
3632 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3633 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3634 BM_LOCKED_SET_ALLOWED))
3638 if (hg > 0) { /* become sync source. */
3640 } else if (hg < 0) { /* become sync target */
3644 if (drbd_bm_total_weight(device)) {
3645 drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3646 drbd_bm_total_weight(device));
3653 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3655 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3656 if (peer == ASB_DISCARD_REMOTE)
3657 return ASB_DISCARD_LOCAL;
3659 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3660 if (peer == ASB_DISCARD_LOCAL)
3661 return ASB_DISCARD_REMOTE;
3663 /* everything else is valid if they are equal on both sides. */
3667 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3669 struct p_protocol *p = pi->data;
3670 enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3671 int p_proto, p_discard_my_data, p_two_primaries, cf;
3672 struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3673 char integrity_alg[SHARED_SECRET_MAX] = "";
3674 struct crypto_shash *peer_integrity_tfm = NULL;
3675 void *int_dig_in = NULL, *int_dig_vv = NULL;
3677 p_proto = be32_to_cpu(p->protocol);
3678 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
3679 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
3680 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
3681 p_two_primaries = be32_to_cpu(p->two_primaries);
3682 cf = be32_to_cpu(p->conn_flags);
3683 p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3685 if (connection->agreed_pro_version >= 87) {
3688 if (pi->size > sizeof(integrity_alg))
3690 err = drbd_recv_all(connection, integrity_alg, pi->size);
3693 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3696 if (pi->cmd != P_PROTOCOL_UPDATE) {
3697 clear_bit(CONN_DRY_RUN, &connection->flags);
3699 if (cf & CF_DRY_RUN)
3700 set_bit(CONN_DRY_RUN, &connection->flags);
3703 nc = rcu_dereference(connection->net_conf);
3705 if (p_proto != nc->wire_protocol) {
3706 drbd_err(connection, "incompatible %s settings\n", "protocol");
3707 goto disconnect_rcu_unlock;
3710 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3711 drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3712 goto disconnect_rcu_unlock;
3715 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3716 drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3717 goto disconnect_rcu_unlock;
3720 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3721 drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3722 goto disconnect_rcu_unlock;
3725 if (p_discard_my_data && nc->discard_my_data) {
3726 drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3727 goto disconnect_rcu_unlock;
3730 if (p_two_primaries != nc->two_primaries) {
3731 drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3732 goto disconnect_rcu_unlock;
3735 if (strcmp(integrity_alg, nc->integrity_alg)) {
3736 drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3737 goto disconnect_rcu_unlock;
3743 if (integrity_alg[0]) {
3747 * We can only change the peer data integrity algorithm
3748 * here. Changing our own data integrity algorithm
3749 * requires that we send a P_PROTOCOL_UPDATE packet at
3750 * the same time; otherwise, the peer has no way to
3751 * tell between which packets the algorithm should
3755 peer_integrity_tfm = crypto_alloc_shash(integrity_alg, 0, 0);
3756 if (IS_ERR(peer_integrity_tfm)) {
3757 peer_integrity_tfm = NULL;
3758 drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3763 hash_size = crypto_shash_digestsize(peer_integrity_tfm);
3764 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3765 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3766 if (!(int_dig_in && int_dig_vv)) {
3767 drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3772 new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3776 mutex_lock(&connection->data.mutex);
3777 mutex_lock(&connection->resource->conf_update);
3778 old_net_conf = connection->net_conf;
3779 *new_net_conf = *old_net_conf;
3781 new_net_conf->wire_protocol = p_proto;
3782 new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3783 new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3784 new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3785 new_net_conf->two_primaries = p_two_primaries;
3787 rcu_assign_pointer(connection->net_conf, new_net_conf);
3788 mutex_unlock(&connection->resource->conf_update);
3789 mutex_unlock(&connection->data.mutex);
3791 crypto_free_shash(connection->peer_integrity_tfm);
3792 kfree(connection->int_dig_in);
3793 kfree(connection->int_dig_vv);
3794 connection->peer_integrity_tfm = peer_integrity_tfm;
3795 connection->int_dig_in = int_dig_in;
3796 connection->int_dig_vv = int_dig_vv;
3798 if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3799 drbd_info(connection, "peer data-integrity-alg: %s\n",
3800 integrity_alg[0] ? integrity_alg : "(none)");
3803 kfree(old_net_conf);
3806 disconnect_rcu_unlock:
3809 crypto_free_shash(peer_integrity_tfm);
3812 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3817 * input: alg name, feature name
3818 * return: NULL (alg name was "")
3819 * ERR_PTR(error) if something goes wrong
3820 * or the crypto hash ptr, if it worked out ok. */
3821 static struct crypto_shash *drbd_crypto_alloc_digest_safe(
3822 const struct drbd_device *device,
3823 const char *alg, const char *name)
3825 struct crypto_shash *tfm;
3830 tfm = crypto_alloc_shash(alg, 0, 0);
3832 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3833 alg, name, PTR_ERR(tfm));
3839 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3841 void *buffer = connection->data.rbuf;
3842 int size = pi->size;
3845 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3846 s = drbd_recv(connection, buffer, s);
3860 * config_unknown_volume - device configuration command for unknown volume
3862 * When a device is added to an existing connection, the node on which the
3863 * device is added first will send configuration commands to its peer but the
3864 * peer will not know about the device yet. It will warn and ignore these
3865 * commands. Once the device is added on the second node, the second node will
3866 * send the same device configuration commands, but in the other direction.
3868 * (We can also end up here if drbd is misconfigured.)
3870 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3872 drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3873 cmdname(pi->cmd), pi->vnr);
3874 return ignore_remaining_packet(connection, pi);
3877 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3879 struct drbd_peer_device *peer_device;
3880 struct drbd_device *device;
3881 struct p_rs_param_95 *p;
3882 unsigned int header_size, data_size, exp_max_sz;
3883 struct crypto_shash *verify_tfm = NULL;
3884 struct crypto_shash *csums_tfm = NULL;
3885 struct net_conf *old_net_conf, *new_net_conf = NULL;
3886 struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3887 const int apv = connection->agreed_pro_version;
3888 struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3889 unsigned int fifo_size = 0;
3892 peer_device = conn_peer_device(connection, pi->vnr);
3894 return config_unknown_volume(connection, pi);
3895 device = peer_device->device;
3897 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
3898 : apv == 88 ? sizeof(struct p_rs_param)
3900 : apv <= 94 ? sizeof(struct p_rs_param_89)
3901 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3903 if (pi->size > exp_max_sz) {
3904 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3905 pi->size, exp_max_sz);
3910 header_size = sizeof(struct p_rs_param);
3911 data_size = pi->size - header_size;
3912 } else if (apv <= 94) {
3913 header_size = sizeof(struct p_rs_param_89);
3914 data_size = pi->size - header_size;
3915 D_ASSERT(device, data_size == 0);
3917 header_size = sizeof(struct p_rs_param_95);
3918 data_size = pi->size - header_size;
3919 D_ASSERT(device, data_size == 0);
3922 /* initialize verify_alg and csums_alg */
3924 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
3926 err = drbd_recv_all(peer_device->connection, p, header_size);
3930 mutex_lock(&connection->resource->conf_update);
3931 old_net_conf = peer_device->connection->net_conf;
3932 if (get_ldev(device)) {
3933 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3934 if (!new_disk_conf) {
3936 mutex_unlock(&connection->resource->conf_update);
3937 drbd_err(device, "Allocation of new disk_conf failed\n");
3941 old_disk_conf = device->ldev->disk_conf;
3942 *new_disk_conf = *old_disk_conf;
3944 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3949 if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3950 drbd_err(device, "verify-alg of wrong size, "
3951 "peer wants %u, accepting only up to %u byte\n",
3952 data_size, SHARED_SECRET_MAX);
3957 err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3960 /* we expect NUL terminated string */
3961 /* but just in case someone tries to be evil */
3962 D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3963 p->verify_alg[data_size-1] = 0;
3965 } else /* apv >= 89 */ {
3966 /* we still expect NUL terminated strings */
3967 /* but just in case someone tries to be evil */
3968 D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3969 D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3970 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3971 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3974 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3975 if (device->state.conn == C_WF_REPORT_PARAMS) {
3976 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3977 old_net_conf->verify_alg, p->verify_alg);
3980 verify_tfm = drbd_crypto_alloc_digest_safe(device,
3981 p->verify_alg, "verify-alg");
3982 if (IS_ERR(verify_tfm)) {
3988 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3989 if (device->state.conn == C_WF_REPORT_PARAMS) {
3990 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3991 old_net_conf->csums_alg, p->csums_alg);
3994 csums_tfm = drbd_crypto_alloc_digest_safe(device,
3995 p->csums_alg, "csums-alg");
3996 if (IS_ERR(csums_tfm)) {
4002 if (apv > 94 && new_disk_conf) {
4003 new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
4004 new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
4005 new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
4006 new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
4008 fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
4009 if (fifo_size != device->rs_plan_s->size) {
4010 new_plan = fifo_alloc(fifo_size);
4012 drbd_err(device, "kmalloc of fifo_buffer failed");
4019 if (verify_tfm || csums_tfm) {
4020 new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
4024 *new_net_conf = *old_net_conf;
4027 strcpy(new_net_conf->verify_alg, p->verify_alg);
4028 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
4029 crypto_free_shash(peer_device->connection->verify_tfm);
4030 peer_device->connection->verify_tfm = verify_tfm;
4031 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
4034 strcpy(new_net_conf->csums_alg, p->csums_alg);
4035 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
4036 crypto_free_shash(peer_device->connection->csums_tfm);
4037 peer_device->connection->csums_tfm = csums_tfm;
4038 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
4040 rcu_assign_pointer(connection->net_conf, new_net_conf);
4044 if (new_disk_conf) {
4045 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4050 old_plan = device->rs_plan_s;
4051 rcu_assign_pointer(device->rs_plan_s, new_plan);
4054 mutex_unlock(&connection->resource->conf_update);
4057 kfree(old_net_conf);
4058 kfree(old_disk_conf);
4064 if (new_disk_conf) {
4066 kfree(new_disk_conf);
4068 mutex_unlock(&connection->resource->conf_update);
4073 if (new_disk_conf) {
4075 kfree(new_disk_conf);
4077 mutex_unlock(&connection->resource->conf_update);
4078 /* just for completeness: actually not needed,
4079 * as this is not reached if csums_tfm was ok. */
4080 crypto_free_shash(csums_tfm);
4081 /* but free the verify_tfm again, if csums_tfm did not work out */
4082 crypto_free_shash(verify_tfm);
4083 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4087 /* warn if the arguments differ by more than 12.5% */
4088 static void warn_if_differ_considerably(struct drbd_device *device,
4089 const char *s, sector_t a, sector_t b)
4092 if (a == 0 || b == 0)
4094 d = (a > b) ? (a - b) : (b - a);
4095 if (d > (a>>3) || d > (b>>3))
4096 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
4097 (unsigned long long)a, (unsigned long long)b);
4100 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
4102 struct drbd_peer_device *peer_device;
4103 struct drbd_device *device;
4104 struct p_sizes *p = pi->data;
4105 struct o_qlim *o = (connection->agreed_features & DRBD_FF_WSAME) ? p->qlim : NULL;
4106 enum determine_dev_size dd = DS_UNCHANGED;
4107 sector_t p_size, p_usize, p_csize, my_usize;
4108 sector_t new_size, cur_size;
4109 int ldsc = 0; /* local disk size changed */
4110 enum dds_flags ddsf;
4112 peer_device = conn_peer_device(connection, pi->vnr);
4114 return config_unknown_volume(connection, pi);
4115 device = peer_device->device;
4116 cur_size = get_capacity(device->vdisk);
4118 p_size = be64_to_cpu(p->d_size);
4119 p_usize = be64_to_cpu(p->u_size);
4120 p_csize = be64_to_cpu(p->c_size);
4122 /* just store the peer's disk size for now.
4123 * we still need to figure out whether we accept that. */
4124 device->p_size = p_size;
4126 if (get_ldev(device)) {
4128 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
4131 warn_if_differ_considerably(device, "lower level device sizes",
4132 p_size, drbd_get_max_capacity(device->ldev));
4133 warn_if_differ_considerably(device, "user requested size",
4136 /* if this is the first connect, or an otherwise expected
4137 * param exchange, choose the minimum */
4138 if (device->state.conn == C_WF_REPORT_PARAMS)
4139 p_usize = min_not_zero(my_usize, p_usize);
4141 /* Never shrink a device with usable data during connect,
4142 * or "attach" on the peer.
4143 * But allow online shrinking if we are connected. */
4144 new_size = drbd_new_dev_size(device, device->ldev, p_usize, 0);
4145 if (new_size < cur_size &&
4146 device->state.disk >= D_OUTDATED &&
4147 (device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS)) {
4148 drbd_err(device, "The peer's disk size is too small! (%llu < %llu sectors)\n",
4149 (unsigned long long)new_size, (unsigned long long)cur_size);
4150 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4155 if (my_usize != p_usize) {
4156 struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
4158 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
4159 if (!new_disk_conf) {
4164 mutex_lock(&connection->resource->conf_update);
4165 old_disk_conf = device->ldev->disk_conf;
4166 *new_disk_conf = *old_disk_conf;
4167 new_disk_conf->disk_size = p_usize;
4169 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4170 mutex_unlock(&connection->resource->conf_update);
4172 kfree(old_disk_conf);
4174 drbd_info(device, "Peer sets u_size to %lu sectors (old: %lu)\n",
4175 (unsigned long)p_usize, (unsigned long)my_usize);
4181 device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
4182 /* Leave drbd_reconsider_queue_parameters() before drbd_determine_dev_size().
4183 In case we cleared the QUEUE_FLAG_DISCARD from our queue in
4184 drbd_reconsider_queue_parameters(), we can be sure that after
4185 drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
4187 ddsf = be16_to_cpu(p->dds_flags);
4188 if (get_ldev(device)) {
4189 drbd_reconsider_queue_parameters(device, device->ldev, o);
4190 dd = drbd_determine_dev_size(device, ddsf, NULL);
4194 drbd_md_sync(device);
4197 * I am diskless, need to accept the peer's *current* size.
4198 * I must NOT accept the peers backing disk size,
4199 * it may have been larger than mine all along...
4201 * At this point, the peer knows more about my disk, or at
4202 * least about what we last agreed upon, than myself.
4203 * So if his c_size is less than his d_size, the most likely
4204 * reason is that *my* d_size was smaller last time we checked.
4206 * However, if he sends a zero current size,
4207 * take his (user-capped or) backing disk size anyways.
4209 * Unless of course he does not have a disk himself.
4210 * In which case we ignore this completely.
4212 sector_t new_size = p_csize ?: p_usize ?: p_size;
4213 drbd_reconsider_queue_parameters(device, NULL, o);
4214 if (new_size == 0) {
4215 /* Ignore, peer does not know nothing. */
4216 } else if (new_size == cur_size) {
4218 } else if (cur_size != 0 && p_size == 0) {
4219 drbd_warn(device, "Ignored diskless peer device size (peer:%llu != me:%llu sectors)!\n",
4220 (unsigned long long)new_size, (unsigned long long)cur_size);
4221 } else if (new_size < cur_size && device->state.role == R_PRIMARY) {
4222 drbd_err(device, "The peer's device size is too small! (%llu < %llu sectors); demote me first!\n",
4223 (unsigned long long)new_size, (unsigned long long)cur_size);
4224 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4227 /* I believe the peer, if
4228 * - I don't have a current size myself
4229 * - we agree on the size anyways
4230 * - I do have a current size, am Secondary,
4231 * and he has the only disk
4232 * - I do have a current size, am Primary,
4233 * and he has the only disk,
4234 * which is larger than my current size
4236 drbd_set_my_capacity(device, new_size);
4240 if (get_ldev(device)) {
4241 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
4242 device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
4249 if (device->state.conn > C_WF_REPORT_PARAMS) {
4250 if (be64_to_cpu(p->c_size) != get_capacity(device->vdisk) ||
4252 /* we have different sizes, probably peer
4253 * needs to know my new size... */
4254 drbd_send_sizes(peer_device, 0, ddsf);
4256 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
4257 (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
4258 if (device->state.pdsk >= D_INCONSISTENT &&
4259 device->state.disk >= D_INCONSISTENT) {
4260 if (ddsf & DDSF_NO_RESYNC)
4261 drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
4263 resync_after_online_grow(device);
4265 set_bit(RESYNC_AFTER_NEG, &device->flags);
4272 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
4274 struct drbd_peer_device *peer_device;
4275 struct drbd_device *device;
4276 struct p_uuids *p = pi->data;
4278 int i, updated_uuids = 0;
4280 peer_device = conn_peer_device(connection, pi->vnr);
4282 return config_unknown_volume(connection, pi);
4283 device = peer_device->device;
4285 p_uuid = kmalloc_array(UI_EXTENDED_SIZE, sizeof(*p_uuid), GFP_NOIO);
4289 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
4290 p_uuid[i] = be64_to_cpu(p->uuid[i]);
4292 kfree(device->p_uuid);
4293 device->p_uuid = p_uuid;
4295 if ((device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS) &&
4296 device->state.disk < D_INCONSISTENT &&
4297 device->state.role == R_PRIMARY &&
4298 (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
4299 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
4300 (unsigned long long)device->ed_uuid);
4301 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4305 if (get_ldev(device)) {
4306 int skip_initial_sync =
4307 device->state.conn == C_CONNECTED &&
4308 peer_device->connection->agreed_pro_version >= 90 &&
4309 device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
4310 (p_uuid[UI_FLAGS] & 8);
4311 if (skip_initial_sync) {
4312 drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
4313 drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
4314 "clear_n_write from receive_uuids",
4315 BM_LOCKED_TEST_ALLOWED);
4316 _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
4317 _drbd_uuid_set(device, UI_BITMAP, 0);
4318 _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
4320 drbd_md_sync(device);
4324 } else if (device->state.disk < D_INCONSISTENT &&
4325 device->state.role == R_PRIMARY) {
4326 /* I am a diskless primary, the peer just created a new current UUID
4328 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4331 /* Before we test for the disk state, we should wait until an eventually
4332 ongoing cluster wide state change is finished. That is important if
4333 we are primary and are detaching from our disk. We need to see the
4334 new disk state... */
4335 mutex_lock(device->state_mutex);
4336 mutex_unlock(device->state_mutex);
4337 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
4338 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4341 drbd_print_uuids(device, "receiver updated UUIDs to");
4347 * convert_state() - Converts the peer's view of the cluster state to our point of view
4348 * @ps: The state as seen by the peer.
4350 static union drbd_state convert_state(union drbd_state ps)
4352 union drbd_state ms;
4354 static enum drbd_conns c_tab[] = {
4355 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
4356 [C_CONNECTED] = C_CONNECTED,
4358 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
4359 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
4360 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
4361 [C_VERIFY_S] = C_VERIFY_T,
4367 ms.conn = c_tab[ps.conn];
4372 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
4377 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4379 struct drbd_peer_device *peer_device;
4380 struct drbd_device *device;
4381 struct p_req_state *p = pi->data;
4382 union drbd_state mask, val;
4383 enum drbd_state_rv rv;
4385 peer_device = conn_peer_device(connection, pi->vnr);
4388 device = peer_device->device;
4390 mask.i = be32_to_cpu(p->mask);
4391 val.i = be32_to_cpu(p->val);
4393 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4394 mutex_is_locked(device->state_mutex)) {
4395 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4399 mask = convert_state(mask);
4400 val = convert_state(val);
4402 rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4403 drbd_send_sr_reply(peer_device, rv);
4405 drbd_md_sync(device);
4410 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4412 struct p_req_state *p = pi->data;
4413 union drbd_state mask, val;
4414 enum drbd_state_rv rv;
4416 mask.i = be32_to_cpu(p->mask);
4417 val.i = be32_to_cpu(p->val);
4419 if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4420 mutex_is_locked(&connection->cstate_mutex)) {
4421 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4425 mask = convert_state(mask);
4426 val = convert_state(val);
4428 rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4429 conn_send_sr_reply(connection, rv);
4434 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4436 struct drbd_peer_device *peer_device;
4437 struct drbd_device *device;
4438 struct p_state *p = pi->data;
4439 union drbd_state os, ns, peer_state;
4440 enum drbd_disk_state real_peer_disk;
4441 enum chg_state_flags cs_flags;
4444 peer_device = conn_peer_device(connection, pi->vnr);
4446 return config_unknown_volume(connection, pi);
4447 device = peer_device->device;
4449 peer_state.i = be32_to_cpu(p->state);
4451 real_peer_disk = peer_state.disk;
4452 if (peer_state.disk == D_NEGOTIATING) {
4453 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4454 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4457 spin_lock_irq(&device->resource->req_lock);
4459 os = ns = drbd_read_state(device);
4460 spin_unlock_irq(&device->resource->req_lock);
4462 /* If some other part of the code (ack_receiver thread, timeout)
4463 * already decided to close the connection again,
4464 * we must not "re-establish" it here. */
4465 if (os.conn <= C_TEAR_DOWN)
4468 /* If this is the "end of sync" confirmation, usually the peer disk
4469 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4470 * set) resync started in PausedSyncT, or if the timing of pause-/
4471 * unpause-sync events has been "just right", the peer disk may
4472 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4474 if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4475 real_peer_disk == D_UP_TO_DATE &&
4476 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4477 /* If we are (becoming) SyncSource, but peer is still in sync
4478 * preparation, ignore its uptodate-ness to avoid flapping, it
4479 * will change to inconsistent once the peer reaches active
4481 * It may have changed syncer-paused flags, however, so we
4482 * cannot ignore this completely. */
4483 if (peer_state.conn > C_CONNECTED &&
4484 peer_state.conn < C_SYNC_SOURCE)
4485 real_peer_disk = D_INCONSISTENT;
4487 /* if peer_state changes to connected at the same time,
4488 * it explicitly notifies us that it finished resync.
4489 * Maybe we should finish it up, too? */
4490 else if (os.conn >= C_SYNC_SOURCE &&
4491 peer_state.conn == C_CONNECTED) {
4492 if (drbd_bm_total_weight(device) <= device->rs_failed)
4493 drbd_resync_finished(device);
4498 /* explicit verify finished notification, stop sector reached. */
4499 if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4500 peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4501 ov_out_of_sync_print(device);
4502 drbd_resync_finished(device);
4506 /* peer says his disk is inconsistent, while we think it is uptodate,
4507 * and this happens while the peer still thinks we have a sync going on,
4508 * but we think we are already done with the sync.
4509 * We ignore this to avoid flapping pdsk.
4510 * This should not happen, if the peer is a recent version of drbd. */
4511 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4512 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4513 real_peer_disk = D_UP_TO_DATE;
4515 if (ns.conn == C_WF_REPORT_PARAMS)
4516 ns.conn = C_CONNECTED;
4518 if (peer_state.conn == C_AHEAD)
4522 * if (primary and diskless and peer uuid != effective uuid)
4523 * abort attach on peer;
4525 * If this node does not have good data, was already connected, but
4526 * the peer did a late attach only now, trying to "negotiate" with me,
4527 * AND I am currently Primary, possibly frozen, with some specific
4528 * "effective" uuid, this should never be reached, really, because
4529 * we first send the uuids, then the current state.
4531 * In this scenario, we already dropped the connection hard
4532 * when we received the unsuitable uuids (receive_uuids().
4534 * Should we want to change this, that is: not drop the connection in
4535 * receive_uuids() already, then we would need to add a branch here
4536 * that aborts the attach of "unsuitable uuids" on the peer in case
4537 * this node is currently Diskless Primary.
4540 if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4541 get_ldev_if_state(device, D_NEGOTIATING)) {
4542 int cr; /* consider resync */
4544 /* if we established a new connection */
4545 cr = (os.conn < C_CONNECTED);
4546 /* if we had an established connection
4547 * and one of the nodes newly attaches a disk */
4548 cr |= (os.conn == C_CONNECTED &&
4549 (peer_state.disk == D_NEGOTIATING ||
4550 os.disk == D_NEGOTIATING));
4551 /* if we have both been inconsistent, and the peer has been
4552 * forced to be UpToDate with --force */
4553 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4554 /* if we had been plain connected, and the admin requested to
4555 * start a sync by "invalidate" or "invalidate-remote" */
4556 cr |= (os.conn == C_CONNECTED &&
4557 (peer_state.conn >= C_STARTING_SYNC_S &&
4558 peer_state.conn <= C_WF_BITMAP_T));
4561 ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4564 if (ns.conn == C_MASK) {
4565 ns.conn = C_CONNECTED;
4566 if (device->state.disk == D_NEGOTIATING) {
4567 drbd_force_state(device, NS(disk, D_FAILED));
4568 } else if (peer_state.disk == D_NEGOTIATING) {
4569 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4570 peer_state.disk = D_DISKLESS;
4571 real_peer_disk = D_DISKLESS;
4573 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4575 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4576 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4582 spin_lock_irq(&device->resource->req_lock);
4583 if (os.i != drbd_read_state(device).i)
4585 clear_bit(CONSIDER_RESYNC, &device->flags);
4586 ns.peer = peer_state.role;
4587 ns.pdsk = real_peer_disk;
4588 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4589 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4590 ns.disk = device->new_state_tmp.disk;
4591 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4592 if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4593 test_bit(NEW_CUR_UUID, &device->flags)) {
4594 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4595 for temporal network outages! */
4596 spin_unlock_irq(&device->resource->req_lock);
4597 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4598 tl_clear(peer_device->connection);
4599 drbd_uuid_new_current(device);
4600 clear_bit(NEW_CUR_UUID, &device->flags);
4601 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4604 rv = _drbd_set_state(device, ns, cs_flags, NULL);
4605 ns = drbd_read_state(device);
4606 spin_unlock_irq(&device->resource->req_lock);
4608 if (rv < SS_SUCCESS) {
4609 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4613 if (os.conn > C_WF_REPORT_PARAMS) {
4614 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4615 peer_state.disk != D_NEGOTIATING ) {
4616 /* we want resync, peer has not yet decided to sync... */
4617 /* Nowadays only used when forcing a node into primary role and
4618 setting its disk to UpToDate with that */
4619 drbd_send_uuids(peer_device);
4620 drbd_send_current_state(peer_device);
4624 clear_bit(DISCARD_MY_DATA, &device->flags);
4626 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4631 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4633 struct drbd_peer_device *peer_device;
4634 struct drbd_device *device;
4635 struct p_rs_uuid *p = pi->data;
4637 peer_device = conn_peer_device(connection, pi->vnr);
4640 device = peer_device->device;
4642 wait_event(device->misc_wait,
4643 device->state.conn == C_WF_SYNC_UUID ||
4644 device->state.conn == C_BEHIND ||
4645 device->state.conn < C_CONNECTED ||
4646 device->state.disk < D_NEGOTIATING);
4648 /* D_ASSERT(device, device->state.conn == C_WF_SYNC_UUID ); */
4650 /* Here the _drbd_uuid_ functions are right, current should
4651 _not_ be rotated into the history */
4652 if (get_ldev_if_state(device, D_NEGOTIATING)) {
4653 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4654 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4656 drbd_print_uuids(device, "updated sync uuid");
4657 drbd_start_resync(device, C_SYNC_TARGET);
4661 drbd_err(device, "Ignoring SyncUUID packet!\n");
4667 * receive_bitmap_plain
4669 * Return 0 when done, 1 when another iteration is needed, and a negative error
4670 * code upon failure.
4673 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4674 unsigned long *p, struct bm_xfer_ctx *c)
4676 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4677 drbd_header_size(peer_device->connection);
4678 unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4679 c->bm_words - c->word_offset);
4680 unsigned int want = num_words * sizeof(*p);
4684 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4689 err = drbd_recv_all(peer_device->connection, p, want);
4693 drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4695 c->word_offset += num_words;
4696 c->bit_offset = c->word_offset * BITS_PER_LONG;
4697 if (c->bit_offset > c->bm_bits)
4698 c->bit_offset = c->bm_bits;
4703 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4705 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4708 static int dcbp_get_start(struct p_compressed_bm *p)
4710 return (p->encoding & 0x80) != 0;
4713 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4715 return (p->encoding >> 4) & 0x7;
4721 * Return 0 when done, 1 when another iteration is needed, and a negative error
4722 * code upon failure.
4725 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4726 struct p_compressed_bm *p,
4727 struct bm_xfer_ctx *c,
4730 struct bitstream bs;
4734 unsigned long s = c->bit_offset;
4736 int toggle = dcbp_get_start(p);
4740 bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4742 bits = bitstream_get_bits(&bs, &look_ahead, 64);
4746 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4747 bits = vli_decode_bits(&rl, look_ahead);
4753 if (e >= c->bm_bits) {
4754 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4757 _drbd_bm_set_bits(peer_device->device, s, e);
4761 drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4762 have, bits, look_ahead,
4763 (unsigned int)(bs.cur.b - p->code),
4764 (unsigned int)bs.buf_len);
4767 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4768 if (likely(bits < 64))
4769 look_ahead >>= bits;
4774 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4777 look_ahead |= tmp << have;
4782 bm_xfer_ctx_bit_to_word_offset(c);
4784 return (s != c->bm_bits);
4790 * Return 0 when done, 1 when another iteration is needed, and a negative error
4791 * code upon failure.
4794 decode_bitmap_c(struct drbd_peer_device *peer_device,
4795 struct p_compressed_bm *p,
4796 struct bm_xfer_ctx *c,
4799 if (dcbp_get_code(p) == RLE_VLI_Bits)
4800 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4802 /* other variants had been implemented for evaluation,
4803 * but have been dropped as this one turned out to be "best"
4804 * during all our tests. */
4806 drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4807 conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4811 void INFO_bm_xfer_stats(struct drbd_device *device,
4812 const char *direction, struct bm_xfer_ctx *c)
4814 /* what would it take to transfer it "plaintext" */
4815 unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4816 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4817 unsigned int plain =
4818 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4819 c->bm_words * sizeof(unsigned long);
4820 unsigned int total = c->bytes[0] + c->bytes[1];
4823 /* total can not be zero. but just in case: */
4827 /* don't report if not compressed */
4831 /* total < plain. check for overflow, still */
4832 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4833 : (1000 * total / plain);
4839 drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4840 "total %u; compression: %u.%u%%\n",
4842 c->bytes[1], c->packets[1],
4843 c->bytes[0], c->packets[0],
4844 total, r/10, r % 10);
4847 /* Since we are processing the bitfield from lower addresses to higher,
4848 it does not matter if the process it in 32 bit chunks or 64 bit
4849 chunks as long as it is little endian. (Understand it as byte stream,
4850 beginning with the lowest byte...) If we would use big endian
4851 we would need to process it from the highest address to the lowest,
4852 in order to be agnostic to the 32 vs 64 bits issue.
4854 returns 0 on failure, 1 if we successfully received it. */
4855 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4857 struct drbd_peer_device *peer_device;
4858 struct drbd_device *device;
4859 struct bm_xfer_ctx c;
4862 peer_device = conn_peer_device(connection, pi->vnr);
4865 device = peer_device->device;
4867 drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4868 /* you are supposed to send additional out-of-sync information
4869 * if you actually set bits during this phase */
4871 c = (struct bm_xfer_ctx) {
4872 .bm_bits = drbd_bm_bits(device),
4873 .bm_words = drbd_bm_words(device),
4877 if (pi->cmd == P_BITMAP)
4878 err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4879 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4880 /* MAYBE: sanity check that we speak proto >= 90,
4881 * and the feature is enabled! */
4882 struct p_compressed_bm *p = pi->data;
4884 if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4885 drbd_err(device, "ReportCBitmap packet too large\n");
4889 if (pi->size <= sizeof(*p)) {
4890 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4894 err = drbd_recv_all(peer_device->connection, p, pi->size);
4897 err = decode_bitmap_c(peer_device, p, &c, pi->size);
4899 drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4904 c.packets[pi->cmd == P_BITMAP]++;
4905 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4912 err = drbd_recv_header(peer_device->connection, pi);
4917 INFO_bm_xfer_stats(device, "receive", &c);
4919 if (device->state.conn == C_WF_BITMAP_T) {
4920 enum drbd_state_rv rv;
4922 err = drbd_send_bitmap(device);
4925 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4926 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4927 D_ASSERT(device, rv == SS_SUCCESS);
4928 } else if (device->state.conn != C_WF_BITMAP_S) {
4929 /* admin may have requested C_DISCONNECTING,
4930 * other threads may have noticed network errors */
4931 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4932 drbd_conn_str(device->state.conn));
4937 drbd_bm_unlock(device);
4938 if (!err && device->state.conn == C_WF_BITMAP_S)
4939 drbd_start_resync(device, C_SYNC_SOURCE);
4943 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4945 drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4948 return ignore_remaining_packet(connection, pi);
4951 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4953 /* Make sure we've acked all the TCP data associated
4954 * with the data requests being unplugged */
4955 tcp_sock_set_quickack(connection->data.socket->sk, 2);
4959 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4961 struct drbd_peer_device *peer_device;
4962 struct drbd_device *device;
4963 struct p_block_desc *p = pi->data;
4965 peer_device = conn_peer_device(connection, pi->vnr);
4968 device = peer_device->device;
4970 switch (device->state.conn) {
4971 case C_WF_SYNC_UUID:
4976 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4977 drbd_conn_str(device->state.conn));
4980 drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4985 static int receive_rs_deallocated(struct drbd_connection *connection, struct packet_info *pi)
4987 struct drbd_peer_device *peer_device;
4988 struct p_block_desc *p = pi->data;
4989 struct drbd_device *device;
4993 peer_device = conn_peer_device(connection, pi->vnr);
4996 device = peer_device->device;
4998 sector = be64_to_cpu(p->sector);
4999 size = be32_to_cpu(p->blksize);
5001 dec_rs_pending(device);
5003 if (get_ldev(device)) {
5004 struct drbd_peer_request *peer_req;
5005 const int op = REQ_OP_WRITE_ZEROES;
5007 peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER, sector,
5014 peer_req->w.cb = e_end_resync_block;
5015 peer_req->submit_jif = jiffies;
5016 peer_req->flags |= EE_TRIM;
5018 spin_lock_irq(&device->resource->req_lock);
5019 list_add_tail(&peer_req->w.list, &device->sync_ee);
5020 spin_unlock_irq(&device->resource->req_lock);
5022 atomic_add(pi->size >> 9, &device->rs_sect_ev);
5023 err = drbd_submit_peer_request(device, peer_req, op, 0, DRBD_FAULT_RS_WR);
5026 spin_lock_irq(&device->resource->req_lock);
5027 list_del(&peer_req->w.list);
5028 spin_unlock_irq(&device->resource->req_lock);
5030 drbd_free_peer_req(device, peer_req);
5036 inc_unacked(device);
5038 /* No put_ldev() here. Gets called in drbd_endio_write_sec_final(),
5039 as well as drbd_rs_complete_io() */
5042 drbd_rs_complete_io(device, sector);
5043 drbd_send_ack_ex(peer_device, P_NEG_ACK, sector, size, ID_SYNCER);
5046 atomic_add(size >> 9, &device->rs_sect_in);
5053 unsigned int pkt_size;
5054 int (*fn)(struct drbd_connection *, struct packet_info *);
5057 static struct data_cmd drbd_cmd_handler[] = {
5058 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
5059 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
5060 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
5061 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
5062 [P_BITMAP] = { 1, 0, receive_bitmap } ,
5063 [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
5064 [P_UNPLUG_REMOTE] = { 0, 0, receive_UnplugRemote },
5065 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5066 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5067 [P_SYNC_PARAM] = { 1, 0, receive_SyncParam },
5068 [P_SYNC_PARAM89] = { 1, 0, receive_SyncParam },
5069 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
5070 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
5071 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
5072 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
5073 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
5074 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
5075 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5076 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5077 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
5078 [P_RS_THIN_REQ] = { 0, sizeof(struct p_block_req), receive_DataRequest },
5079 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
5080 [P_OUT_OF_SYNC] = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
5081 [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
5082 [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
5083 [P_TRIM] = { 0, sizeof(struct p_trim), receive_Data },
5084 [P_ZEROES] = { 0, sizeof(struct p_trim), receive_Data },
5085 [P_RS_DEALLOCATED] = { 0, sizeof(struct p_block_desc), receive_rs_deallocated },
5086 [P_WSAME] = { 1, sizeof(struct p_wsame), receive_Data },
5089 static void drbdd(struct drbd_connection *connection)
5091 struct packet_info pi;
5092 size_t shs; /* sub header size */
5095 while (get_t_state(&connection->receiver) == RUNNING) {
5096 struct data_cmd const *cmd;
5098 drbd_thread_current_set_cpu(&connection->receiver);
5099 update_receiver_timing_details(connection, drbd_recv_header_maybe_unplug);
5100 if (drbd_recv_header_maybe_unplug(connection, &pi))
5103 cmd = &drbd_cmd_handler[pi.cmd];
5104 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
5105 drbd_err(connection, "Unexpected data packet %s (0x%04x)",
5106 cmdname(pi.cmd), pi.cmd);
5110 shs = cmd->pkt_size;
5111 if (pi.cmd == P_SIZES && connection->agreed_features & DRBD_FF_WSAME)
5112 shs += sizeof(struct o_qlim);
5113 if (pi.size > shs && !cmd->expect_payload) {
5114 drbd_err(connection, "No payload expected %s l:%d\n",
5115 cmdname(pi.cmd), pi.size);
5118 if (pi.size < shs) {
5119 drbd_err(connection, "%s: unexpected packet size, expected:%d received:%d\n",
5120 cmdname(pi.cmd), (int)shs, pi.size);
5125 update_receiver_timing_details(connection, drbd_recv_all_warn);
5126 err = drbd_recv_all_warn(connection, pi.data, shs);
5132 update_receiver_timing_details(connection, cmd->fn);
5133 err = cmd->fn(connection, &pi);
5135 drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
5136 cmdname(pi.cmd), err, pi.size);
5143 conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
5146 static void conn_disconnect(struct drbd_connection *connection)
5148 struct drbd_peer_device *peer_device;
5152 if (connection->cstate == C_STANDALONE)
5155 /* We are about to start the cleanup after connection loss.
5156 * Make sure drbd_make_request knows about that.
5157 * Usually we should be in some network failure state already,
5158 * but just in case we are not, we fix it up here.
5160 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5162 /* ack_receiver does not clean up anything. it must not interfere, either */
5163 drbd_thread_stop(&connection->ack_receiver);
5164 if (connection->ack_sender) {
5165 destroy_workqueue(connection->ack_sender);
5166 connection->ack_sender = NULL;
5168 drbd_free_sock(connection);
5171 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5172 struct drbd_device *device = peer_device->device;
5173 kref_get(&device->kref);
5175 drbd_disconnected(peer_device);
5176 kref_put(&device->kref, drbd_destroy_device);
5181 if (!list_empty(&connection->current_epoch->list))
5182 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
5183 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
5184 atomic_set(&connection->current_epoch->epoch_size, 0);
5185 connection->send.seen_any_write_yet = false;
5187 drbd_info(connection, "Connection closed\n");
5189 if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
5190 conn_try_outdate_peer_async(connection);
5192 spin_lock_irq(&connection->resource->req_lock);
5193 oc = connection->cstate;
5194 if (oc >= C_UNCONNECTED)
5195 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
5197 spin_unlock_irq(&connection->resource->req_lock);
5199 if (oc == C_DISCONNECTING)
5200 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
5203 static int drbd_disconnected(struct drbd_peer_device *peer_device)
5205 struct drbd_device *device = peer_device->device;
5208 /* wait for current activity to cease. */
5209 spin_lock_irq(&device->resource->req_lock);
5210 _drbd_wait_ee_list_empty(device, &device->active_ee);
5211 _drbd_wait_ee_list_empty(device, &device->sync_ee);
5212 _drbd_wait_ee_list_empty(device, &device->read_ee);
5213 spin_unlock_irq(&device->resource->req_lock);
5215 /* We do not have data structures that would allow us to
5216 * get the rs_pending_cnt down to 0 again.
5217 * * On C_SYNC_TARGET we do not have any data structures describing
5218 * the pending RSDataRequest's we have sent.
5219 * * On C_SYNC_SOURCE there is no data structure that tracks
5220 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
5221 * And no, it is not the sum of the reference counts in the
5222 * resync_LRU. The resync_LRU tracks the whole operation including
5223 * the disk-IO, while the rs_pending_cnt only tracks the blocks
5225 drbd_rs_cancel_all(device);
5226 device->rs_total = 0;
5227 device->rs_failed = 0;
5228 atomic_set(&device->rs_pending_cnt, 0);
5229 wake_up(&device->misc_wait);
5231 del_timer_sync(&device->resync_timer);
5232 resync_timer_fn(&device->resync_timer);
5234 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
5235 * w_make_resync_request etc. which may still be on the worker queue
5236 * to be "canceled" */
5237 drbd_flush_workqueue(&peer_device->connection->sender_work);
5239 drbd_finish_peer_reqs(device);
5241 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
5242 might have issued a work again. The one before drbd_finish_peer_reqs() is
5243 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
5244 drbd_flush_workqueue(&peer_device->connection->sender_work);
5246 /* need to do it again, drbd_finish_peer_reqs() may have populated it
5247 * again via drbd_try_clear_on_disk_bm(). */
5248 drbd_rs_cancel_all(device);
5250 kfree(device->p_uuid);
5251 device->p_uuid = NULL;
5253 if (!drbd_suspended(device))
5254 tl_clear(peer_device->connection);
5256 drbd_md_sync(device);
5258 if (get_ldev(device)) {
5259 drbd_bitmap_io(device, &drbd_bm_write_copy_pages,
5260 "write from disconnected", BM_LOCKED_CHANGE_ALLOWED);
5264 /* tcp_close and release of sendpage pages can be deferred. I don't
5265 * want to use SO_LINGER, because apparently it can be deferred for
5266 * more than 20 seconds (longest time I checked).
5268 * Actually we don't care for exactly when the network stack does its
5269 * put_page(), but release our reference on these pages right here.
5271 i = drbd_free_peer_reqs(device, &device->net_ee);
5273 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
5274 i = atomic_read(&device->pp_in_use_by_net);
5276 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
5277 i = atomic_read(&device->pp_in_use);
5279 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
5281 D_ASSERT(device, list_empty(&device->read_ee));
5282 D_ASSERT(device, list_empty(&device->active_ee));
5283 D_ASSERT(device, list_empty(&device->sync_ee));
5284 D_ASSERT(device, list_empty(&device->done_ee));
5290 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
5291 * we can agree on is stored in agreed_pro_version.
5293 * feature flags and the reserved array should be enough room for future
5294 * enhancements of the handshake protocol, and possible plugins...
5296 * for now, they are expected to be zero, but ignored.
5298 static int drbd_send_features(struct drbd_connection *connection)
5300 struct drbd_socket *sock;
5301 struct p_connection_features *p;
5303 sock = &connection->data;
5304 p = conn_prepare_command(connection, sock);
5307 memset(p, 0, sizeof(*p));
5308 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
5309 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
5310 p->feature_flags = cpu_to_be32(PRO_FEATURES);
5311 return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
5316 * 1 yes, we have a valid connection
5317 * 0 oops, did not work out, please try again
5318 * -1 peer talks different language,
5319 * no point in trying again, please go standalone.
5321 static int drbd_do_features(struct drbd_connection *connection)
5323 /* ASSERT current == connection->receiver ... */
5324 struct p_connection_features *p;
5325 const int expect = sizeof(struct p_connection_features);
5326 struct packet_info pi;
5329 err = drbd_send_features(connection);
5333 err = drbd_recv_header(connection, &pi);
5337 if (pi.cmd != P_CONNECTION_FEATURES) {
5338 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
5339 cmdname(pi.cmd), pi.cmd);
5343 if (pi.size != expect) {
5344 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
5350 err = drbd_recv_all_warn(connection, p, expect);
5354 p->protocol_min = be32_to_cpu(p->protocol_min);
5355 p->protocol_max = be32_to_cpu(p->protocol_max);
5356 if (p->protocol_max == 0)
5357 p->protocol_max = p->protocol_min;
5359 if (PRO_VERSION_MAX < p->protocol_min ||
5360 PRO_VERSION_MIN > p->protocol_max)
5363 connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
5364 connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
5366 drbd_info(connection, "Handshake successful: "
5367 "Agreed network protocol version %d\n", connection->agreed_pro_version);
5369 drbd_info(connection, "Feature flags enabled on protocol level: 0x%x%s%s%s%s.\n",
5370 connection->agreed_features,
5371 connection->agreed_features & DRBD_FF_TRIM ? " TRIM" : "",
5372 connection->agreed_features & DRBD_FF_THIN_RESYNC ? " THIN_RESYNC" : "",
5373 connection->agreed_features & DRBD_FF_WSAME ? " WRITE_SAME" : "",
5374 connection->agreed_features & DRBD_FF_WZEROES ? " WRITE_ZEROES" :
5375 connection->agreed_features ? "" : " none");
5380 drbd_err(connection, "incompatible DRBD dialects: "
5381 "I support %d-%d, peer supports %d-%d\n",
5382 PRO_VERSION_MIN, PRO_VERSION_MAX,
5383 p->protocol_min, p->protocol_max);
5387 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
5388 static int drbd_do_auth(struct drbd_connection *connection)
5390 drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
5391 drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
5395 #define CHALLENGE_LEN 64
5399 0 - failed, try again (network error),
5400 -1 - auth failed, don't try again.
5403 static int drbd_do_auth(struct drbd_connection *connection)
5405 struct drbd_socket *sock;
5406 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
5407 char *response = NULL;
5408 char *right_response = NULL;
5409 char *peers_ch = NULL;
5410 unsigned int key_len;
5411 char secret[SHARED_SECRET_MAX]; /* 64 byte */
5412 unsigned int resp_size;
5413 struct shash_desc *desc;
5414 struct packet_info pi;
5415 struct net_conf *nc;
5418 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
5421 nc = rcu_dereference(connection->net_conf);
5422 key_len = strlen(nc->shared_secret);
5423 memcpy(secret, nc->shared_secret, key_len);
5426 desc = kmalloc(sizeof(struct shash_desc) +
5427 crypto_shash_descsize(connection->cram_hmac_tfm),
5433 desc->tfm = connection->cram_hmac_tfm;
5435 rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
5437 drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
5442 get_random_bytes(my_challenge, CHALLENGE_LEN);
5444 sock = &connection->data;
5445 if (!conn_prepare_command(connection, sock)) {
5449 rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
5450 my_challenge, CHALLENGE_LEN);
5454 err = drbd_recv_header(connection, &pi);
5460 if (pi.cmd != P_AUTH_CHALLENGE) {
5461 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
5462 cmdname(pi.cmd), pi.cmd);
5467 if (pi.size > CHALLENGE_LEN * 2) {
5468 drbd_err(connection, "expected AuthChallenge payload too big.\n");
5473 if (pi.size < CHALLENGE_LEN) {
5474 drbd_err(connection, "AuthChallenge payload too small.\n");
5479 peers_ch = kmalloc(pi.size, GFP_NOIO);
5485 err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5491 if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5492 drbd_err(connection, "Peer presented the same challenge!\n");
5497 resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5498 response = kmalloc(resp_size, GFP_NOIO);
5504 rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5506 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5511 if (!conn_prepare_command(connection, sock)) {
5515 rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5516 response, resp_size);
5520 err = drbd_recv_header(connection, &pi);
5526 if (pi.cmd != P_AUTH_RESPONSE) {
5527 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5528 cmdname(pi.cmd), pi.cmd);
5533 if (pi.size != resp_size) {
5534 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5539 err = drbd_recv_all_warn(connection, response , resp_size);
5545 right_response = kmalloc(resp_size, GFP_NOIO);
5546 if (!right_response) {
5551 rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5554 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5559 rv = !memcmp(response, right_response, resp_size);
5562 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5570 kfree(right_response);
5572 shash_desc_zero(desc);
5580 int drbd_receiver(struct drbd_thread *thi)
5582 struct drbd_connection *connection = thi->connection;
5585 drbd_info(connection, "receiver (re)started\n");
5588 h = conn_connect(connection);
5590 conn_disconnect(connection);
5591 schedule_timeout_interruptible(HZ);
5594 drbd_warn(connection, "Discarding network configuration.\n");
5595 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5600 blk_start_plug(&connection->receiver_plug);
5602 blk_finish_plug(&connection->receiver_plug);
5605 conn_disconnect(connection);
5607 drbd_info(connection, "receiver terminated\n");
5611 /* ********* acknowledge sender ******** */
5613 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5615 struct p_req_state_reply *p = pi->data;
5616 int retcode = be32_to_cpu(p->retcode);
5618 if (retcode >= SS_SUCCESS) {
5619 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5621 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5622 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5623 drbd_set_st_err_str(retcode), retcode);
5625 wake_up(&connection->ping_wait);
5630 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5632 struct drbd_peer_device *peer_device;
5633 struct drbd_device *device;
5634 struct p_req_state_reply *p = pi->data;
5635 int retcode = be32_to_cpu(p->retcode);
5637 peer_device = conn_peer_device(connection, pi->vnr);
5640 device = peer_device->device;
5642 if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5643 D_ASSERT(device, connection->agreed_pro_version < 100);
5644 return got_conn_RqSReply(connection, pi);
5647 if (retcode >= SS_SUCCESS) {
5648 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5650 set_bit(CL_ST_CHG_FAIL, &device->flags);
5651 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5652 drbd_set_st_err_str(retcode), retcode);
5654 wake_up(&device->state_wait);
5659 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5661 return drbd_send_ping_ack(connection);
5665 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5667 /* restore idle timeout */
5668 connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5669 if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5670 wake_up(&connection->ping_wait);
5675 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5677 struct drbd_peer_device *peer_device;
5678 struct drbd_device *device;
5679 struct p_block_ack *p = pi->data;
5680 sector_t sector = be64_to_cpu(p->sector);
5681 int blksize = be32_to_cpu(p->blksize);
5683 peer_device = conn_peer_device(connection, pi->vnr);
5686 device = peer_device->device;
5688 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5690 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5692 if (get_ldev(device)) {
5693 drbd_rs_complete_io(device, sector);
5694 drbd_set_in_sync(device, sector, blksize);
5695 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5696 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5699 dec_rs_pending(device);
5700 atomic_add(blksize >> 9, &device->rs_sect_in);
5706 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5707 struct rb_root *root, const char *func,
5708 enum drbd_req_event what, bool missing_ok)
5710 struct drbd_request *req;
5711 struct bio_and_error m;
5713 spin_lock_irq(&device->resource->req_lock);
5714 req = find_request(device, root, id, sector, missing_ok, func);
5715 if (unlikely(!req)) {
5716 spin_unlock_irq(&device->resource->req_lock);
5719 __req_mod(req, what, &m);
5720 spin_unlock_irq(&device->resource->req_lock);
5723 complete_master_bio(device, &m);
5727 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5729 struct drbd_peer_device *peer_device;
5730 struct drbd_device *device;
5731 struct p_block_ack *p = pi->data;
5732 sector_t sector = be64_to_cpu(p->sector);
5733 int blksize = be32_to_cpu(p->blksize);
5734 enum drbd_req_event what;
5736 peer_device = conn_peer_device(connection, pi->vnr);
5739 device = peer_device->device;
5741 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5743 if (p->block_id == ID_SYNCER) {
5744 drbd_set_in_sync(device, sector, blksize);
5745 dec_rs_pending(device);
5749 case P_RS_WRITE_ACK:
5750 what = WRITE_ACKED_BY_PEER_AND_SIS;
5753 what = WRITE_ACKED_BY_PEER;
5756 what = RECV_ACKED_BY_PEER;
5759 what = CONFLICT_RESOLVED;
5762 what = POSTPONE_WRITE;
5768 return validate_req_change_req_state(device, p->block_id, sector,
5769 &device->write_requests, __func__,
5773 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5775 struct drbd_peer_device *peer_device;
5776 struct drbd_device *device;
5777 struct p_block_ack *p = pi->data;
5778 sector_t sector = be64_to_cpu(p->sector);
5779 int size = be32_to_cpu(p->blksize);
5782 peer_device = conn_peer_device(connection, pi->vnr);
5785 device = peer_device->device;
5787 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5789 if (p->block_id == ID_SYNCER) {
5790 dec_rs_pending(device);
5791 drbd_rs_failed_io(device, sector, size);
5795 err = validate_req_change_req_state(device, p->block_id, sector,
5796 &device->write_requests, __func__,
5799 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5800 The master bio might already be completed, therefore the
5801 request is no longer in the collision hash. */
5802 /* In Protocol B we might already have got a P_RECV_ACK
5803 but then get a P_NEG_ACK afterwards. */
5804 drbd_set_out_of_sync(device, sector, size);
5809 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5811 struct drbd_peer_device *peer_device;
5812 struct drbd_device *device;
5813 struct p_block_ack *p = pi->data;
5814 sector_t sector = be64_to_cpu(p->sector);
5816 peer_device = conn_peer_device(connection, pi->vnr);
5819 device = peer_device->device;
5821 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5823 drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5824 (unsigned long long)sector, be32_to_cpu(p->blksize));
5826 return validate_req_change_req_state(device, p->block_id, sector,
5827 &device->read_requests, __func__,
5831 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5833 struct drbd_peer_device *peer_device;
5834 struct drbd_device *device;
5837 struct p_block_ack *p = pi->data;
5839 peer_device = conn_peer_device(connection, pi->vnr);
5842 device = peer_device->device;
5844 sector = be64_to_cpu(p->sector);
5845 size = be32_to_cpu(p->blksize);
5847 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5849 dec_rs_pending(device);
5851 if (get_ldev_if_state(device, D_FAILED)) {
5852 drbd_rs_complete_io(device, sector);
5854 case P_NEG_RS_DREPLY:
5855 drbd_rs_failed_io(device, sector, size);
5868 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5870 struct p_barrier_ack *p = pi->data;
5871 struct drbd_peer_device *peer_device;
5874 tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5877 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5878 struct drbd_device *device = peer_device->device;
5880 if (device->state.conn == C_AHEAD &&
5881 atomic_read(&device->ap_in_flight) == 0 &&
5882 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5883 device->start_resync_timer.expires = jiffies + HZ;
5884 add_timer(&device->start_resync_timer);
5892 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5894 struct drbd_peer_device *peer_device;
5895 struct drbd_device *device;
5896 struct p_block_ack *p = pi->data;
5897 struct drbd_device_work *dw;
5901 peer_device = conn_peer_device(connection, pi->vnr);
5904 device = peer_device->device;
5906 sector = be64_to_cpu(p->sector);
5907 size = be32_to_cpu(p->blksize);
5909 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5911 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5912 drbd_ov_out_of_sync_found(device, sector, size);
5914 ov_out_of_sync_print(device);
5916 if (!get_ldev(device))
5919 drbd_rs_complete_io(device, sector);
5920 dec_rs_pending(device);
5924 /* let's advance progress step marks only for every other megabyte */
5925 if ((device->ov_left & 0x200) == 0x200)
5926 drbd_advance_rs_marks(device, device->ov_left);
5928 if (device->ov_left == 0) {
5929 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5931 dw->w.cb = w_ov_finished;
5932 dw->device = device;
5933 drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5935 drbd_err(device, "kmalloc(dw) failed.");
5936 ov_out_of_sync_print(device);
5937 drbd_resync_finished(device);
5944 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5949 struct meta_sock_cmd {
5951 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5954 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5957 struct net_conf *nc;
5960 nc = rcu_dereference(connection->net_conf);
5961 t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5968 connection->meta.socket->sk->sk_rcvtimeo = t;
5971 static void set_ping_timeout(struct drbd_connection *connection)
5973 set_rcvtimeo(connection, 1);
5976 static void set_idle_timeout(struct drbd_connection *connection)
5978 set_rcvtimeo(connection, 0);
5981 static struct meta_sock_cmd ack_receiver_tbl[] = {
5982 [P_PING] = { 0, got_Ping },
5983 [P_PING_ACK] = { 0, got_PingAck },
5984 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5985 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5986 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5987 [P_SUPERSEDED] = { sizeof(struct p_block_ack), got_BlockAck },
5988 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
5989 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
5990 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply },
5991 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
5992 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
5993 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5994 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
5995 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
5996 [P_RS_CANCEL] = { sizeof(struct p_block_ack), got_NegRSDReply },
5997 [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5998 [P_RETRY_WRITE] = { sizeof(struct p_block_ack), got_BlockAck },
6001 int drbd_ack_receiver(struct drbd_thread *thi)
6003 struct drbd_connection *connection = thi->connection;
6004 struct meta_sock_cmd *cmd = NULL;
6005 struct packet_info pi;
6006 unsigned long pre_recv_jif;
6008 void *buf = connection->meta.rbuf;
6010 unsigned int header_size = drbd_header_size(connection);
6011 int expect = header_size;
6012 bool ping_timeout_active = false;
6014 sched_set_fifo_low(current);
6016 while (get_t_state(thi) == RUNNING) {
6017 drbd_thread_current_set_cpu(thi);
6019 conn_reclaim_net_peer_reqs(connection);
6021 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
6022 if (drbd_send_ping(connection)) {
6023 drbd_err(connection, "drbd_send_ping has failed\n");
6026 set_ping_timeout(connection);
6027 ping_timeout_active = true;
6030 pre_recv_jif = jiffies;
6031 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
6034 * -EINTR (on meta) we got a signal
6035 * -EAGAIN (on meta) rcvtimeo expired
6036 * -ECONNRESET other side closed the connection
6037 * -ERESTARTSYS (on data) we got a signal
6038 * rv < 0 other than above: unexpected error!
6039 * rv == expected: full header or command
6040 * rv < expected: "woken" by signal during receive
6041 * rv == 0 : "connection shut down by peer"
6043 if (likely(rv > 0)) {
6046 } else if (rv == 0) {
6047 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
6050 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
6053 t = wait_event_timeout(connection->ping_wait,
6054 connection->cstate < C_WF_REPORT_PARAMS,
6059 drbd_err(connection, "meta connection shut down by peer.\n");
6061 } else if (rv == -EAGAIN) {
6062 /* If the data socket received something meanwhile,
6063 * that is good enough: peer is still alive. */
6064 if (time_after(connection->last_received, pre_recv_jif))
6066 if (ping_timeout_active) {
6067 drbd_err(connection, "PingAck did not arrive in time.\n");
6070 set_bit(SEND_PING, &connection->flags);
6072 } else if (rv == -EINTR) {
6073 /* maybe drbd_thread_stop(): the while condition will notice.
6074 * maybe woken for send_ping: we'll send a ping above,
6075 * and change the rcvtimeo */
6076 flush_signals(current);
6079 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
6083 if (received == expect && cmd == NULL) {
6084 if (decode_header(connection, connection->meta.rbuf, &pi))
6086 cmd = &ack_receiver_tbl[pi.cmd];
6087 if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
6088 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
6089 cmdname(pi.cmd), pi.cmd);
6092 expect = header_size + cmd->pkt_size;
6093 if (pi.size != expect - header_size) {
6094 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
6099 if (received == expect) {
6102 err = cmd->fn(connection, &pi);
6104 drbd_err(connection, "%ps failed\n", cmd->fn);
6108 connection->last_received = jiffies;
6110 if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
6111 set_idle_timeout(connection);
6112 ping_timeout_active = false;
6115 buf = connection->meta.rbuf;
6117 expect = header_size;
6124 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6125 conn_md_sync(connection);
6129 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
6132 drbd_info(connection, "ack_receiver terminated\n");
6137 void drbd_send_acks_wf(struct work_struct *ws)
6139 struct drbd_peer_device *peer_device =
6140 container_of(ws, struct drbd_peer_device, send_acks_work);
6141 struct drbd_connection *connection = peer_device->connection;
6142 struct drbd_device *device = peer_device->device;
6143 struct net_conf *nc;
6147 nc = rcu_dereference(connection->net_conf);
6148 tcp_cork = nc->tcp_cork;
6152 tcp_sock_set_cork(connection->meta.socket->sk, true);
6154 err = drbd_finish_peer_reqs(device);
6155 kref_put(&device->kref, drbd_destroy_device);
6156 /* get is in drbd_endio_write_sec_final(). That is necessary to keep the
6157 struct work_struct send_acks_work alive, which is in the peer_device object */
6160 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6165 tcp_sock_set_cork(connection->meta.socket->sk, false);