4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
28 #include <asm/uaccess.h>
31 #include <linux/drbd.h>
33 #include <linux/file.h>
36 #include <linux/memcontrol.h>
37 #include <linux/mm_inline.h>
38 #include <linux/slab.h>
39 #include <linux/pkt_sched.h>
40 #define __KERNEL_SYSCALLS__
41 #include <linux/unistd.h>
42 #include <linux/vmalloc.h>
43 #include <linux/random.h>
44 #include <linux/string.h>
45 #include <linux/scatterlist.h>
47 #include "drbd_protocol.h"
65 static int drbd_do_features(struct drbd_connection *connection);
66 static int drbd_do_auth(struct drbd_connection *connection);
67 static int drbd_disconnected(struct drbd_peer_device *);
69 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
70 static int e_end_block(struct drbd_work *, int);
73 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
76 * some helper functions to deal with single linked page lists,
77 * page->private being our "next" pointer.
80 /* If at least n pages are linked at head, get n pages off.
81 * Otherwise, don't modify head, and return NULL.
82 * Locking is the responsibility of the caller.
84 static struct page *page_chain_del(struct page **head, int n)
98 tmp = page_chain_next(page);
100 break; /* found sufficient pages */
102 /* insufficient pages, don't use any of them. */
107 /* add end of list marker for the returned list */
108 set_page_private(page, 0);
109 /* actual return value, and adjustment of head */
115 /* may be used outside of locks to find the tail of a (usually short)
116 * "private" page chain, before adding it back to a global chain head
117 * with page_chain_add() under a spinlock. */
118 static struct page *page_chain_tail(struct page *page, int *len)
122 while ((tmp = page_chain_next(page)))
129 static int page_chain_free(struct page *page)
133 page_chain_for_each_safe(page, tmp) {
140 static void page_chain_add(struct page **head,
141 struct page *chain_first, struct page *chain_last)
145 tmp = page_chain_tail(chain_first, NULL);
146 BUG_ON(tmp != chain_last);
149 /* add chain to head */
150 set_page_private(chain_last, (unsigned long)*head);
154 static struct page *__drbd_alloc_pages(struct drbd_device *device,
157 struct page *page = NULL;
158 struct page *tmp = NULL;
161 /* Yes, testing drbd_pp_vacant outside the lock is racy.
162 * So what. It saves a spin_lock. */
163 if (drbd_pp_vacant >= number) {
164 spin_lock(&drbd_pp_lock);
165 page = page_chain_del(&drbd_pp_pool, number);
167 drbd_pp_vacant -= number;
168 spin_unlock(&drbd_pp_lock);
173 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
174 * "criss-cross" setup, that might cause write-out on some other DRBD,
175 * which in turn might block on the other node at this very place. */
176 for (i = 0; i < number; i++) {
177 tmp = alloc_page(GFP_TRY);
180 set_page_private(tmp, (unsigned long)page);
187 /* Not enough pages immediately available this time.
188 * No need to jump around here, drbd_alloc_pages will retry this
189 * function "soon". */
191 tmp = page_chain_tail(page, NULL);
192 spin_lock(&drbd_pp_lock);
193 page_chain_add(&drbd_pp_pool, page, tmp);
195 spin_unlock(&drbd_pp_lock);
200 static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
201 struct list_head *to_be_freed)
203 struct drbd_peer_request *peer_req, *tmp;
205 /* The EEs are always appended to the end of the list. Since
206 they are sent in order over the wire, they have to finish
207 in order. As soon as we see the first not finished we can
208 stop to examine the list... */
210 list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
211 if (drbd_peer_req_has_active_page(peer_req))
213 list_move(&peer_req->w.list, to_be_freed);
217 static void drbd_kick_lo_and_reclaim_net(struct drbd_device *device)
219 LIST_HEAD(reclaimed);
220 struct drbd_peer_request *peer_req, *t;
222 spin_lock_irq(&device->resource->req_lock);
223 reclaim_finished_net_peer_reqs(device, &reclaimed);
224 spin_unlock_irq(&device->resource->req_lock);
226 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
227 drbd_free_net_peer_req(device, peer_req);
231 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
232 * @device: DRBD device.
233 * @number: number of pages requested
234 * @retry: whether to retry, if not enough pages are available right now
236 * Tries to allocate number pages, first from our own page pool, then from
237 * the kernel, unless this allocation would exceed the max_buffers setting.
238 * Possibly retry until DRBD frees sufficient pages somewhere else.
240 * Returns a page chain linked via page->private.
242 struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
245 struct drbd_device *device = peer_device->device;
246 struct page *page = NULL;
251 /* Yes, we may run up to @number over max_buffers. If we
252 * follow it strictly, the admin will get it wrong anyways. */
254 nc = rcu_dereference(peer_device->connection->net_conf);
255 mxb = nc ? nc->max_buffers : 1000000;
258 if (atomic_read(&device->pp_in_use) < mxb)
259 page = __drbd_alloc_pages(device, number);
261 while (page == NULL) {
262 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
264 drbd_kick_lo_and_reclaim_net(device);
266 if (atomic_read(&device->pp_in_use) < mxb) {
267 page = __drbd_alloc_pages(device, number);
275 if (signal_pending(current)) {
276 drbd_warn(device, "drbd_alloc_pages interrupted!\n");
282 finish_wait(&drbd_pp_wait, &wait);
285 atomic_add(number, &device->pp_in_use);
289 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
290 * Is also used from inside an other spin_lock_irq(&resource->req_lock);
291 * Either links the page chain back to the global pool,
292 * or returns all pages to the system. */
293 static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
295 atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
301 if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * minor_count)
302 i = page_chain_free(page);
305 tmp = page_chain_tail(page, &i);
306 spin_lock(&drbd_pp_lock);
307 page_chain_add(&drbd_pp_pool, page, tmp);
309 spin_unlock(&drbd_pp_lock);
311 i = atomic_sub_return(i, a);
313 drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
314 is_net ? "pp_in_use_by_net" : "pp_in_use", i);
315 wake_up(&drbd_pp_wait);
319 You need to hold the req_lock:
320 _drbd_wait_ee_list_empty()
322 You must not have the req_lock:
324 drbd_alloc_peer_req()
325 drbd_free_peer_reqs()
327 drbd_finish_peer_reqs()
329 drbd_wait_ee_list_empty()
332 struct drbd_peer_request *
333 drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
334 unsigned int data_size, gfp_t gfp_mask) __must_hold(local)
336 struct drbd_device *device = peer_device->device;
337 struct drbd_peer_request *peer_req;
338 struct page *page = NULL;
339 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
341 if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
344 peer_req = mempool_alloc(drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
346 if (!(gfp_mask & __GFP_NOWARN))
347 drbd_err(device, "%s: allocation failed\n", __func__);
352 page = drbd_alloc_pages(peer_device, nr_pages, (gfp_mask & __GFP_WAIT));
357 drbd_clear_interval(&peer_req->i);
358 peer_req->i.size = data_size;
359 peer_req->i.sector = sector;
360 peer_req->i.local = false;
361 peer_req->i.waiting = false;
363 peer_req->epoch = NULL;
364 peer_req->peer_device = peer_device;
365 peer_req->pages = page;
366 atomic_set(&peer_req->pending_bios, 0);
369 * The block_id is opaque to the receiver. It is not endianness
370 * converted, and sent back to the sender unchanged.
372 peer_req->block_id = id;
377 mempool_free(peer_req, drbd_ee_mempool);
381 void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
384 if (peer_req->flags & EE_HAS_DIGEST)
385 kfree(peer_req->digest);
386 drbd_free_pages(device, peer_req->pages, is_net);
387 D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
388 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
389 mempool_free(peer_req, drbd_ee_mempool);
392 int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
394 LIST_HEAD(work_list);
395 struct drbd_peer_request *peer_req, *t;
397 int is_net = list == &device->net_ee;
399 spin_lock_irq(&device->resource->req_lock);
400 list_splice_init(list, &work_list);
401 spin_unlock_irq(&device->resource->req_lock);
403 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
404 __drbd_free_peer_req(device, peer_req, is_net);
411 * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
413 static int drbd_finish_peer_reqs(struct drbd_device *device)
415 LIST_HEAD(work_list);
416 LIST_HEAD(reclaimed);
417 struct drbd_peer_request *peer_req, *t;
420 spin_lock_irq(&device->resource->req_lock);
421 reclaim_finished_net_peer_reqs(device, &reclaimed);
422 list_splice_init(&device->done_ee, &work_list);
423 spin_unlock_irq(&device->resource->req_lock);
425 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
426 drbd_free_net_peer_req(device, peer_req);
428 /* possible callbacks here:
429 * e_end_block, and e_end_resync_block, e_send_superseded.
430 * all ignore the last argument.
432 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
435 /* list_del not necessary, next/prev members not touched */
436 err2 = peer_req->w.cb(&peer_req->w, !!err);
439 drbd_free_peer_req(device, peer_req);
441 wake_up(&device->ee_wait);
446 static void _drbd_wait_ee_list_empty(struct drbd_device *device,
447 struct list_head *head)
451 /* avoids spin_lock/unlock
452 * and calling prepare_to_wait in the fast path */
453 while (!list_empty(head)) {
454 prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
455 spin_unlock_irq(&device->resource->req_lock);
457 finish_wait(&device->ee_wait, &wait);
458 spin_lock_irq(&device->resource->req_lock);
462 static void drbd_wait_ee_list_empty(struct drbd_device *device,
463 struct list_head *head)
465 spin_lock_irq(&device->resource->req_lock);
466 _drbd_wait_ee_list_empty(device, head);
467 spin_unlock_irq(&device->resource->req_lock);
470 static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
476 struct msghdr msg = {
477 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
479 return kernel_recvmsg(sock, &msg, &iov, 1, size, msg.msg_flags);
482 static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
486 rv = drbd_recv_short(connection->data.socket, buf, size, 0);
489 if (rv == -ECONNRESET)
490 drbd_info(connection, "sock was reset by peer\n");
491 else if (rv != -ERESTARTSYS)
492 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
493 } else if (rv == 0) {
494 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
497 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
500 t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);
505 drbd_info(connection, "sock was shut down by peer\n");
509 conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
515 static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
519 err = drbd_recv(connection, buf, size);
528 static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
532 err = drbd_recv_all(connection, buf, size);
533 if (err && !signal_pending(current))
534 drbd_warn(connection, "short read (expected size %d)\n", (int)size);
539 * On individual connections, the socket buffer size must be set prior to the
540 * listen(2) or connect(2) calls in order to have it take effect.
541 * This is our wrapper to do so.
543 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
546 /* open coded SO_SNDBUF, SO_RCVBUF */
548 sock->sk->sk_sndbuf = snd;
549 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
552 sock->sk->sk_rcvbuf = rcv;
553 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
557 static struct socket *drbd_try_connect(struct drbd_connection *connection)
561 struct sockaddr_in6 src_in6;
562 struct sockaddr_in6 peer_in6;
564 int err, peer_addr_len, my_addr_len;
565 int sndbuf_size, rcvbuf_size, connect_int;
566 int disconnect_on_error = 1;
569 nc = rcu_dereference(connection->net_conf);
574 sndbuf_size = nc->sndbuf_size;
575 rcvbuf_size = nc->rcvbuf_size;
576 connect_int = nc->connect_int;
579 my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
580 memcpy(&src_in6, &connection->my_addr, my_addr_len);
582 if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
583 src_in6.sin6_port = 0;
585 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
587 peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
588 memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);
590 what = "sock_create_kern";
591 err = sock_create_kern(((struct sockaddr *)&src_in6)->sa_family,
592 SOCK_STREAM, IPPROTO_TCP, &sock);
598 sock->sk->sk_rcvtimeo =
599 sock->sk->sk_sndtimeo = connect_int * HZ;
600 drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);
602 /* explicitly bind to the configured IP as source IP
603 * for the outgoing connections.
604 * This is needed for multihomed hosts and to be
605 * able to use lo: interfaces for drbd.
606 * Make sure to use 0 as port number, so linux selects
607 * a free one dynamically.
609 what = "bind before connect";
610 err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
614 /* connect may fail, peer not yet available.
615 * stay C_WF_CONNECTION, don't go Disconnecting! */
616 disconnect_on_error = 0;
618 err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);
627 /* timeout, busy, signal pending */
628 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
629 case EINTR: case ERESTARTSYS:
630 /* peer not (yet) available, network problem */
631 case ECONNREFUSED: case ENETUNREACH:
632 case EHOSTDOWN: case EHOSTUNREACH:
633 disconnect_on_error = 0;
636 drbd_err(connection, "%s failed, err = %d\n", what, err);
638 if (disconnect_on_error)
639 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
645 struct accept_wait_data {
646 struct drbd_connection *connection;
647 struct socket *s_listen;
648 struct completion door_bell;
649 void (*original_sk_state_change)(struct sock *sk);
653 static void drbd_incoming_connection(struct sock *sk)
655 struct accept_wait_data *ad = sk->sk_user_data;
656 void (*state_change)(struct sock *sk);
658 state_change = ad->original_sk_state_change;
659 if (sk->sk_state == TCP_ESTABLISHED)
660 complete(&ad->door_bell);
664 static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
666 int err, sndbuf_size, rcvbuf_size, my_addr_len;
667 struct sockaddr_in6 my_addr;
668 struct socket *s_listen;
673 nc = rcu_dereference(connection->net_conf);
678 sndbuf_size = nc->sndbuf_size;
679 rcvbuf_size = nc->rcvbuf_size;
682 my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
683 memcpy(&my_addr, &connection->my_addr, my_addr_len);
685 what = "sock_create_kern";
686 err = sock_create_kern(((struct sockaddr *)&my_addr)->sa_family,
687 SOCK_STREAM, IPPROTO_TCP, &s_listen);
693 s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
694 drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);
696 what = "bind before listen";
697 err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
701 ad->s_listen = s_listen;
702 write_lock_bh(&s_listen->sk->sk_callback_lock);
703 ad->original_sk_state_change = s_listen->sk->sk_state_change;
704 s_listen->sk->sk_state_change = drbd_incoming_connection;
705 s_listen->sk->sk_user_data = ad;
706 write_unlock_bh(&s_listen->sk->sk_callback_lock);
709 err = s_listen->ops->listen(s_listen, 5);
716 sock_release(s_listen);
718 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
719 drbd_err(connection, "%s failed, err = %d\n", what, err);
720 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
727 static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
729 write_lock_bh(&sk->sk_callback_lock);
730 sk->sk_state_change = ad->original_sk_state_change;
731 sk->sk_user_data = NULL;
732 write_unlock_bh(&sk->sk_callback_lock);
735 static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
737 int timeo, connect_int, err = 0;
738 struct socket *s_estab = NULL;
742 nc = rcu_dereference(connection->net_conf);
747 connect_int = nc->connect_int;
750 timeo = connect_int * HZ;
751 /* 28.5% random jitter */
752 timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;
754 err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
758 err = kernel_accept(ad->s_listen, &s_estab, 0);
760 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
761 drbd_err(connection, "accept failed, err = %d\n", err);
762 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
767 unregister_state_change(s_estab->sk, ad);
772 static int decode_header(struct drbd_connection *, void *, struct packet_info *);
774 static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
775 enum drbd_packet cmd)
777 if (!conn_prepare_command(connection, sock))
779 return conn_send_command(connection, sock, cmd, 0, NULL, 0);
782 static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
784 unsigned int header_size = drbd_header_size(connection);
785 struct packet_info pi;
788 err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
789 if (err != header_size) {
794 err = decode_header(connection, connection->data.rbuf, &pi);
801 * drbd_socket_okay() - Free the socket if its connection is not okay
802 * @sock: pointer to the pointer to the socket.
804 static int drbd_socket_okay(struct socket **sock)
812 rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
814 if (rr > 0 || rr == -EAGAIN) {
822 /* Gets called if a connection is established, or if a new minor gets created
824 int drbd_connected(struct drbd_peer_device *peer_device)
826 struct drbd_device *device = peer_device->device;
829 atomic_set(&device->packet_seq, 0);
830 device->peer_seq = 0;
832 device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
833 &peer_device->connection->cstate_mutex :
834 &device->own_state_mutex;
836 err = drbd_send_sync_param(peer_device);
838 err = drbd_send_sizes(peer_device, 0, 0);
840 err = drbd_send_uuids(peer_device);
842 err = drbd_send_current_state(peer_device);
843 clear_bit(USE_DEGR_WFC_T, &device->flags);
844 clear_bit(RESIZE_PENDING, &device->flags);
845 atomic_set(&device->ap_in_flight, 0);
846 mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
852 * 1 yes, we have a valid connection
853 * 0 oops, did not work out, please try again
854 * -1 peer talks different language,
855 * no point in trying again, please go standalone.
856 * -2 We do not have a network config...
858 static int conn_connect(struct drbd_connection *connection)
860 struct drbd_socket sock, msock;
861 struct drbd_peer_device *peer_device;
863 int vnr, timeout, h, ok;
864 bool discard_my_data;
865 enum drbd_state_rv rv;
866 struct accept_wait_data ad = {
867 .connection = connection,
868 .door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
871 clear_bit(DISCONNECT_SENT, &connection->flags);
872 if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
875 mutex_init(&sock.mutex);
876 sock.sbuf = connection->data.sbuf;
877 sock.rbuf = connection->data.rbuf;
879 mutex_init(&msock.mutex);
880 msock.sbuf = connection->meta.sbuf;
881 msock.rbuf = connection->meta.rbuf;
884 /* Assume that the peer only understands protocol 80 until we know better. */
885 connection->agreed_pro_version = 80;
887 if (prepare_listen_socket(connection, &ad))
893 s = drbd_try_connect(connection);
897 send_first_packet(connection, &sock, P_INITIAL_DATA);
898 } else if (!msock.socket) {
899 clear_bit(RESOLVE_CONFLICTS, &connection->flags);
901 send_first_packet(connection, &msock, P_INITIAL_META);
903 drbd_err(connection, "Logic error in conn_connect()\n");
904 goto out_release_sockets;
908 if (sock.socket && msock.socket) {
910 nc = rcu_dereference(connection->net_conf);
911 timeout = nc->ping_timeo * HZ / 10;
913 schedule_timeout_interruptible(timeout);
914 ok = drbd_socket_okay(&sock.socket);
915 ok = drbd_socket_okay(&msock.socket) && ok;
921 s = drbd_wait_for_connect(connection, &ad);
923 int fp = receive_first_packet(connection, s);
924 drbd_socket_okay(&sock.socket);
925 drbd_socket_okay(&msock.socket);
929 drbd_warn(connection, "initial packet S crossed\n");
930 sock_release(sock.socket);
937 set_bit(RESOLVE_CONFLICTS, &connection->flags);
939 drbd_warn(connection, "initial packet M crossed\n");
940 sock_release(msock.socket);
947 drbd_warn(connection, "Error receiving initial packet\n");
950 if (prandom_u32() & 1)
955 if (connection->cstate <= C_DISCONNECTING)
956 goto out_release_sockets;
957 if (signal_pending(current)) {
958 flush_signals(current);
960 if (get_t_state(&connection->receiver) == EXITING)
961 goto out_release_sockets;
964 ok = drbd_socket_okay(&sock.socket);
965 ok = drbd_socket_okay(&msock.socket) && ok;
969 sock_release(ad.s_listen);
971 sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
972 msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
974 sock.socket->sk->sk_allocation = GFP_NOIO;
975 msock.socket->sk->sk_allocation = GFP_NOIO;
977 sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
978 msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
981 * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
982 * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
983 * first set it to the P_CONNECTION_FEATURES timeout,
984 * which we set to 4x the configured ping_timeout. */
986 nc = rcu_dereference(connection->net_conf);
988 sock.socket->sk->sk_sndtimeo =
989 sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
991 msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
992 timeout = nc->timeout * HZ / 10;
993 discard_my_data = nc->discard_my_data;
996 msock.socket->sk->sk_sndtimeo = timeout;
998 /* we don't want delays.
999 * we use TCP_CORK where appropriate, though */
1000 drbd_tcp_nodelay(sock.socket);
1001 drbd_tcp_nodelay(msock.socket);
1003 connection->data.socket = sock.socket;
1004 connection->meta.socket = msock.socket;
1005 connection->last_received = jiffies;
1007 h = drbd_do_features(connection);
1011 if (connection->cram_hmac_tfm) {
1012 /* drbd_request_state(device, NS(conn, WFAuth)); */
1013 switch (drbd_do_auth(connection)) {
1015 drbd_err(connection, "Authentication of peer failed\n");
1018 drbd_err(connection, "Authentication of peer failed, trying again.\n");
1023 connection->data.socket->sk->sk_sndtimeo = timeout;
1024 connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1026 if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1029 set_bit(STATE_SENT, &connection->flags);
1032 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1033 struct drbd_device *device = peer_device->device;
1034 kref_get(&device->kref);
1037 /* Prevent a race between resync-handshake and
1038 * being promoted to Primary.
1040 * Grab and release the state mutex, so we know that any current
1041 * drbd_set_role() is finished, and any incoming drbd_set_role
1042 * will see the STATE_SENT flag, and wait for it to be cleared.
1044 mutex_lock(device->state_mutex);
1045 mutex_unlock(device->state_mutex);
1047 if (discard_my_data)
1048 set_bit(DISCARD_MY_DATA, &device->flags);
1050 clear_bit(DISCARD_MY_DATA, &device->flags);
1052 drbd_connected(peer_device);
1053 kref_put(&device->kref, drbd_destroy_device);
1058 rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1059 if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1060 clear_bit(STATE_SENT, &connection->flags);
1064 drbd_thread_start(&connection->asender);
1066 mutex_lock(&connection->resource->conf_update);
1067 /* The discard_my_data flag is a single-shot modifier to the next
1068 * connection attempt, the handshake of which is now well underway.
1069 * No need for rcu style copying of the whole struct
1070 * just to clear a single value. */
1071 connection->net_conf->discard_my_data = 0;
1072 mutex_unlock(&connection->resource->conf_update);
1076 out_release_sockets:
1078 sock_release(ad.s_listen);
1080 sock_release(sock.socket);
1082 sock_release(msock.socket);
1086 static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1088 unsigned int header_size = drbd_header_size(connection);
1090 if (header_size == sizeof(struct p_header100) &&
1091 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1092 struct p_header100 *h = header;
1094 drbd_err(connection, "Header padding is not zero\n");
1097 pi->vnr = be16_to_cpu(h->volume);
1098 pi->cmd = be16_to_cpu(h->command);
1099 pi->size = be32_to_cpu(h->length);
1100 } else if (header_size == sizeof(struct p_header95) &&
1101 *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1102 struct p_header95 *h = header;
1103 pi->cmd = be16_to_cpu(h->command);
1104 pi->size = be32_to_cpu(h->length);
1106 } else if (header_size == sizeof(struct p_header80) &&
1107 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1108 struct p_header80 *h = header;
1109 pi->cmd = be16_to_cpu(h->command);
1110 pi->size = be16_to_cpu(h->length);
1113 drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1114 be32_to_cpu(*(__be32 *)header),
1115 connection->agreed_pro_version);
1118 pi->data = header + header_size;
1122 static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1124 void *buffer = connection->data.rbuf;
1127 err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1131 err = decode_header(connection, buffer, pi);
1132 connection->last_received = jiffies;
1137 static void drbd_flush(struct drbd_connection *connection)
1140 struct drbd_peer_device *peer_device;
1143 if (connection->write_ordering >= WO_bdev_flush) {
1145 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1146 struct drbd_device *device = peer_device->device;
1148 if (!get_ldev(device))
1150 kref_get(&device->kref);
1153 rv = blkdev_issue_flush(device->ldev->backing_bdev,
1156 drbd_info(device, "local disk flush failed with status %d\n", rv);
1157 /* would rather check on EOPNOTSUPP, but that is not reliable.
1158 * don't try again for ANY return value != 0
1159 * if (rv == -EOPNOTSUPP) */
1160 drbd_bump_write_ordering(connection, WO_drain_io);
1163 kref_put(&device->kref, drbd_destroy_device);
1174 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1175 * @device: DRBD device.
1176 * @epoch: Epoch object.
1179 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1180 struct drbd_epoch *epoch,
1181 enum epoch_event ev)
1184 struct drbd_epoch *next_epoch;
1185 enum finish_epoch rv = FE_STILL_LIVE;
1187 spin_lock(&connection->epoch_lock);
1191 epoch_size = atomic_read(&epoch->epoch_size);
1193 switch (ev & ~EV_CLEANUP) {
1195 atomic_dec(&epoch->active);
1197 case EV_GOT_BARRIER_NR:
1198 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1200 case EV_BECAME_LAST:
1205 if (epoch_size != 0 &&
1206 atomic_read(&epoch->active) == 0 &&
1207 (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1208 if (!(ev & EV_CLEANUP)) {
1209 spin_unlock(&connection->epoch_lock);
1210 drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1211 spin_lock(&connection->epoch_lock);
1214 /* FIXME: dec unacked on connection, once we have
1215 * something to count pending connection packets in. */
1216 if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1217 dec_unacked(epoch->connection);
1220 if (connection->current_epoch != epoch) {
1221 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1222 list_del(&epoch->list);
1223 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1224 connection->epochs--;
1227 if (rv == FE_STILL_LIVE)
1231 atomic_set(&epoch->epoch_size, 0);
1232 /* atomic_set(&epoch->active, 0); is already zero */
1233 if (rv == FE_STILL_LIVE)
1244 spin_unlock(&connection->epoch_lock);
1250 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1251 * @connection: DRBD connection.
1252 * @wo: Write ordering method to try.
1254 void drbd_bump_write_ordering(struct drbd_connection *connection, enum write_ordering_e wo)
1256 struct disk_conf *dc;
1257 struct drbd_peer_device *peer_device;
1258 enum write_ordering_e pwo;
1260 static char *write_ordering_str[] = {
1262 [WO_drain_io] = "drain",
1263 [WO_bdev_flush] = "flush",
1266 pwo = connection->write_ordering;
1269 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1270 struct drbd_device *device = peer_device->device;
1272 if (!get_ldev_if_state(device, D_ATTACHING))
1274 dc = rcu_dereference(device->ldev->disk_conf);
1276 if (wo == WO_bdev_flush && !dc->disk_flushes)
1278 if (wo == WO_drain_io && !dc->disk_drain)
1283 connection->write_ordering = wo;
1284 if (pwo != connection->write_ordering || wo == WO_bdev_flush)
1285 drbd_info(connection, "Method to ensure write ordering: %s\n", write_ordering_str[connection->write_ordering]);
1289 * drbd_submit_peer_request()
1290 * @device: DRBD device.
1291 * @peer_req: peer request
1292 * @rw: flag field, see bio->bi_rw
1294 * May spread the pages to multiple bios,
1295 * depending on bio_add_page restrictions.
1297 * Returns 0 if all bios have been submitted,
1298 * -ENOMEM if we could not allocate enough bios,
1299 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1300 * single page to an empty bio (which should never happen and likely indicates
1301 * that the lower level IO stack is in some way broken). This has been observed
1302 * on certain Xen deployments.
1304 /* TODO allocate from our own bio_set. */
1305 int drbd_submit_peer_request(struct drbd_device *device,
1306 struct drbd_peer_request *peer_req,
1307 const unsigned rw, const int fault_type)
1309 struct bio *bios = NULL;
1311 struct page *page = peer_req->pages;
1312 sector_t sector = peer_req->i.sector;
1313 unsigned ds = peer_req->i.size;
1314 unsigned n_bios = 0;
1315 unsigned nr_pages = (ds + PAGE_SIZE -1) >> PAGE_SHIFT;
1318 /* In most cases, we will only need one bio. But in case the lower
1319 * level restrictions happen to be different at this offset on this
1320 * side than those of the sending peer, we may need to submit the
1321 * request in more than one bio.
1323 * Plain bio_alloc is good enough here, this is no DRBD internally
1324 * generated bio, but a bio allocated on behalf of the peer.
1327 bio = bio_alloc(GFP_NOIO, nr_pages);
1329 drbd_err(device, "submit_ee: Allocation of a bio failed\n");
1332 /* > peer_req->i.sector, unless this is the first bio */
1333 bio->bi_iter.bi_sector = sector;
1334 bio->bi_bdev = device->ldev->backing_bdev;
1336 bio->bi_private = peer_req;
1337 bio->bi_end_io = drbd_peer_request_endio;
1339 bio->bi_next = bios;
1343 page_chain_for_each(page) {
1344 unsigned len = min_t(unsigned, ds, PAGE_SIZE);
1345 if (!bio_add_page(bio, page, len, 0)) {
1346 /* A single page must always be possible!
1347 * But in case it fails anyways,
1348 * we deal with it, and complain (below). */
1349 if (bio->bi_vcnt == 0) {
1351 "bio_add_page failed for len=%u, "
1352 "bi_vcnt=0 (bi_sector=%llu)\n",
1353 len, (uint64_t)bio->bi_iter.bi_sector);
1363 D_ASSERT(device, page == NULL);
1364 D_ASSERT(device, ds == 0);
1366 atomic_set(&peer_req->pending_bios, n_bios);
1369 bios = bios->bi_next;
1370 bio->bi_next = NULL;
1372 drbd_generic_make_request(device, fault_type, bio);
1379 bios = bios->bi_next;
1385 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1386 struct drbd_peer_request *peer_req)
1388 struct drbd_interval *i = &peer_req->i;
1390 drbd_remove_interval(&device->write_requests, i);
1391 drbd_clear_interval(i);
1393 /* Wake up any processes waiting for this peer request to complete. */
1395 wake_up(&device->misc_wait);
1398 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1400 struct drbd_peer_device *peer_device;
1404 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1405 struct drbd_device *device = peer_device->device;
1407 kref_get(&device->kref);
1409 drbd_wait_ee_list_empty(device, &device->active_ee);
1410 kref_put(&device->kref, drbd_destroy_device);
1416 static struct drbd_peer_device *
1417 conn_peer_device(struct drbd_connection *connection, int volume_number)
1419 return idr_find(&connection->peer_devices, volume_number);
1422 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1425 struct p_barrier *p = pi->data;
1426 struct drbd_epoch *epoch;
1428 /* FIXME these are unacked on connection,
1429 * not a specific (peer)device.
1431 connection->current_epoch->barrier_nr = p->barrier;
1432 connection->current_epoch->connection = connection;
1433 rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1435 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1436 * the activity log, which means it would not be resynced in case the
1437 * R_PRIMARY crashes now.
1438 * Therefore we must send the barrier_ack after the barrier request was
1440 switch (connection->write_ordering) {
1442 if (rv == FE_RECYCLED)
1445 /* receiver context, in the writeout path of the other node.
1446 * avoid potential distributed deadlock */
1447 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1451 drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1456 conn_wait_active_ee_empty(connection);
1457 drbd_flush(connection);
1459 if (atomic_read(&connection->current_epoch->epoch_size)) {
1460 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1467 drbd_err(connection, "Strangeness in connection->write_ordering %d\n", connection->write_ordering);
1472 atomic_set(&epoch->epoch_size, 0);
1473 atomic_set(&epoch->active, 0);
1475 spin_lock(&connection->epoch_lock);
1476 if (atomic_read(&connection->current_epoch->epoch_size)) {
1477 list_add(&epoch->list, &connection->current_epoch->list);
1478 connection->current_epoch = epoch;
1479 connection->epochs++;
1481 /* The current_epoch got recycled while we allocated this one... */
1484 spin_unlock(&connection->epoch_lock);
1489 /* used from receive_RSDataReply (recv_resync_read)
1490 * and from receive_Data */
1491 static struct drbd_peer_request *
1492 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1493 int data_size) __must_hold(local)
1495 struct drbd_device *device = peer_device->device;
1496 const sector_t capacity = drbd_get_capacity(device->this_bdev);
1497 struct drbd_peer_request *peer_req;
1500 void *dig_in = peer_device->connection->int_dig_in;
1501 void *dig_vv = peer_device->connection->int_dig_vv;
1502 unsigned long *data;
1505 if (peer_device->connection->peer_integrity_tfm) {
1506 dgs = crypto_hash_digestsize(peer_device->connection->peer_integrity_tfm);
1508 * FIXME: Receive the incoming digest into the receive buffer
1509 * here, together with its struct p_data?
1511 err = drbd_recv_all_warn(peer_device->connection, dig_in, dgs);
1517 if (!expect(IS_ALIGNED(data_size, 512)))
1519 if (!expect(data_size <= DRBD_MAX_BIO_SIZE))
1522 /* even though we trust out peer,
1523 * we sometimes have to double check. */
1524 if (sector + (data_size>>9) > capacity) {
1525 drbd_err(device, "request from peer beyond end of local disk: "
1526 "capacity: %llus < sector: %llus + size: %u\n",
1527 (unsigned long long)capacity,
1528 (unsigned long long)sector, data_size);
1532 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1533 * "criss-cross" setup, that might cause write-out on some other DRBD,
1534 * which in turn might block on the other node at this very place. */
1535 peer_req = drbd_alloc_peer_req(peer_device, id, sector, data_size, GFP_NOIO);
1543 page = peer_req->pages;
1544 page_chain_for_each(page) {
1545 unsigned len = min_t(int, ds, PAGE_SIZE);
1547 err = drbd_recv_all_warn(peer_device->connection, data, len);
1548 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1549 drbd_err(device, "Fault injection: Corrupting data on receive\n");
1550 data[0] = data[0] ^ (unsigned long)-1;
1554 drbd_free_peer_req(device, peer_req);
1561 drbd_csum_ee(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv);
1562 if (memcmp(dig_in, dig_vv, dgs)) {
1563 drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1564 (unsigned long long)sector, data_size);
1565 drbd_free_peer_req(device, peer_req);
1569 device->recv_cnt += data_size>>9;
1573 /* drbd_drain_block() just takes a data block
1574 * out of the socket input buffer, and discards it.
1576 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1585 page = drbd_alloc_pages(peer_device, 1, 1);
1589 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1591 err = drbd_recv_all_warn(peer_device->connection, data, len);
1597 drbd_free_pages(peer_device->device, page, 0);
1601 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1602 sector_t sector, int data_size)
1604 struct bio_vec bvec;
1605 struct bvec_iter iter;
1607 int dgs, err, expect;
1608 void *dig_in = peer_device->connection->int_dig_in;
1609 void *dig_vv = peer_device->connection->int_dig_vv;
1612 if (peer_device->connection->peer_integrity_tfm) {
1613 dgs = crypto_hash_digestsize(peer_device->connection->peer_integrity_tfm);
1614 err = drbd_recv_all_warn(peer_device->connection, dig_in, dgs);
1620 /* optimistically update recv_cnt. if receiving fails below,
1621 * we disconnect anyways, and counters will be reset. */
1622 peer_device->device->recv_cnt += data_size>>9;
1624 bio = req->master_bio;
1625 D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
1627 bio_for_each_segment(bvec, bio, iter) {
1628 void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
1629 expect = min_t(int, data_size, bvec.bv_len);
1630 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
1631 kunmap(bvec.bv_page);
1634 data_size -= expect;
1638 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
1639 if (memcmp(dig_in, dig_vv, dgs)) {
1640 drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
1645 D_ASSERT(peer_device->device, data_size == 0);
1650 * e_end_resync_block() is called in asender context via
1651 * drbd_finish_peer_reqs().
1653 static int e_end_resync_block(struct drbd_work *w, int unused)
1655 struct drbd_peer_request *peer_req =
1656 container_of(w, struct drbd_peer_request, w);
1657 struct drbd_peer_device *peer_device = peer_req->peer_device;
1658 struct drbd_device *device = peer_device->device;
1659 sector_t sector = peer_req->i.sector;
1662 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1664 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1665 drbd_set_in_sync(device, sector, peer_req->i.size);
1666 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
1668 /* Record failure to sync */
1669 drbd_rs_failed_io(device, sector, peer_req->i.size);
1671 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1673 dec_unacked(device);
1678 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
1679 int data_size) __releases(local)
1681 struct drbd_device *device = peer_device->device;
1682 struct drbd_peer_request *peer_req;
1684 peer_req = read_in_block(peer_device, ID_SYNCER, sector, data_size);
1688 dec_rs_pending(device);
1690 inc_unacked(device);
1691 /* corresponding dec_unacked() in e_end_resync_block()
1692 * respective _drbd_clear_done_ee */
1694 peer_req->w.cb = e_end_resync_block;
1696 spin_lock_irq(&device->resource->req_lock);
1697 list_add(&peer_req->w.list, &device->sync_ee);
1698 spin_unlock_irq(&device->resource->req_lock);
1700 atomic_add(data_size >> 9, &device->rs_sect_ev);
1701 if (drbd_submit_peer_request(device, peer_req, WRITE, DRBD_FAULT_RS_WR) == 0)
1704 /* don't care for the reason here */
1705 drbd_err(device, "submit failed, triggering re-connect\n");
1706 spin_lock_irq(&device->resource->req_lock);
1707 list_del(&peer_req->w.list);
1708 spin_unlock_irq(&device->resource->req_lock);
1710 drbd_free_peer_req(device, peer_req);
1716 static struct drbd_request *
1717 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
1718 sector_t sector, bool missing_ok, const char *func)
1720 struct drbd_request *req;
1722 /* Request object according to our peer */
1723 req = (struct drbd_request *)(unsigned long)id;
1724 if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
1727 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
1728 (unsigned long)id, (unsigned long long)sector);
1733 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
1735 struct drbd_peer_device *peer_device;
1736 struct drbd_device *device;
1737 struct drbd_request *req;
1740 struct p_data *p = pi->data;
1742 peer_device = conn_peer_device(connection, pi->vnr);
1745 device = peer_device->device;
1747 sector = be64_to_cpu(p->sector);
1749 spin_lock_irq(&device->resource->req_lock);
1750 req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
1751 spin_unlock_irq(&device->resource->req_lock);
1755 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1756 * special casing it there for the various failure cases.
1757 * still no race with drbd_fail_pending_reads */
1758 err = recv_dless_read(peer_device, req, sector, pi->size);
1760 req_mod(req, DATA_RECEIVED);
1761 /* else: nothing. handled from drbd_disconnect...
1762 * I don't think we may complete this just yet
1763 * in case we are "on-disconnect: freeze" */
1768 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
1770 struct drbd_peer_device *peer_device;
1771 struct drbd_device *device;
1774 struct p_data *p = pi->data;
1776 peer_device = conn_peer_device(connection, pi->vnr);
1779 device = peer_device->device;
1781 sector = be64_to_cpu(p->sector);
1782 D_ASSERT(device, p->block_id == ID_SYNCER);
1784 if (get_ldev(device)) {
1785 /* data is submitted to disk within recv_resync_read.
1786 * corresponding put_ldev done below on error,
1787 * or in drbd_peer_request_endio. */
1788 err = recv_resync_read(peer_device, sector, pi->size);
1790 if (__ratelimit(&drbd_ratelimit_state))
1791 drbd_err(device, "Can not write resync data to local disk.\n");
1793 err = drbd_drain_block(peer_device, pi->size);
1795 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
1798 atomic_add(pi->size >> 9, &device->rs_sect_in);
1803 static void restart_conflicting_writes(struct drbd_device *device,
1804 sector_t sector, int size)
1806 struct drbd_interval *i;
1807 struct drbd_request *req;
1809 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
1812 req = container_of(i, struct drbd_request, i);
1813 if (req->rq_state & RQ_LOCAL_PENDING ||
1814 !(req->rq_state & RQ_POSTPONED))
1816 /* as it is RQ_POSTPONED, this will cause it to
1817 * be queued on the retry workqueue. */
1818 __req_mod(req, CONFLICT_RESOLVED, NULL);
1823 * e_end_block() is called in asender context via drbd_finish_peer_reqs().
1825 static int e_end_block(struct drbd_work *w, int cancel)
1827 struct drbd_peer_request *peer_req =
1828 container_of(w, struct drbd_peer_request, w);
1829 struct drbd_peer_device *peer_device = peer_req->peer_device;
1830 struct drbd_device *device = peer_device->device;
1831 sector_t sector = peer_req->i.sector;
1834 if (peer_req->flags & EE_SEND_WRITE_ACK) {
1835 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1836 pcmd = (device->state.conn >= C_SYNC_SOURCE &&
1837 device->state.conn <= C_PAUSED_SYNC_T &&
1838 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
1839 P_RS_WRITE_ACK : P_WRITE_ACK;
1840 err = drbd_send_ack(peer_device, pcmd, peer_req);
1841 if (pcmd == P_RS_WRITE_ACK)
1842 drbd_set_in_sync(device, sector, peer_req->i.size);
1844 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1845 /* we expect it to be marked out of sync anyways...
1846 * maybe assert this? */
1848 dec_unacked(device);
1850 /* we delete from the conflict detection hash _after_ we sent out the
1851 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1852 if (peer_req->flags & EE_IN_INTERVAL_TREE) {
1853 spin_lock_irq(&device->resource->req_lock);
1854 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
1855 drbd_remove_epoch_entry_interval(device, peer_req);
1856 if (peer_req->flags & EE_RESTART_REQUESTS)
1857 restart_conflicting_writes(device, sector, peer_req->i.size);
1858 spin_unlock_irq(&device->resource->req_lock);
1860 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1862 drbd_may_finish_epoch(first_peer_device(device)->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
1867 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
1869 struct drbd_peer_request *peer_req =
1870 container_of(w, struct drbd_peer_request, w);
1871 struct drbd_peer_device *peer_device = peer_req->peer_device;
1874 err = drbd_send_ack(peer_device, ack, peer_req);
1875 dec_unacked(peer_device->device);
1880 static int e_send_superseded(struct drbd_work *w, int unused)
1882 return e_send_ack(w, P_SUPERSEDED);
1885 static int e_send_retry_write(struct drbd_work *w, int unused)
1887 struct drbd_peer_request *peer_req =
1888 container_of(w, struct drbd_peer_request, w);
1889 struct drbd_connection *connection = peer_req->peer_device->connection;
1891 return e_send_ack(w, connection->agreed_pro_version >= 100 ?
1892 P_RETRY_WRITE : P_SUPERSEDED);
1895 static bool seq_greater(u32 a, u32 b)
1898 * We assume 32-bit wrap-around here.
1899 * For 24-bit wrap-around, we would have to shift:
1902 return (s32)a - (s32)b > 0;
1905 static u32 seq_max(u32 a, u32 b)
1907 return seq_greater(a, b) ? a : b;
1910 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
1912 struct drbd_device *device = peer_device->device;
1913 unsigned int newest_peer_seq;
1915 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
1916 spin_lock(&device->peer_seq_lock);
1917 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
1918 device->peer_seq = newest_peer_seq;
1919 spin_unlock(&device->peer_seq_lock);
1920 /* wake up only if we actually changed device->peer_seq */
1921 if (peer_seq == newest_peer_seq)
1922 wake_up(&device->seq_wait);
1926 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
1928 return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
1931 /* maybe change sync_ee into interval trees as well? */
1932 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
1934 struct drbd_peer_request *rs_req;
1937 spin_lock_irq(&device->resource->req_lock);
1938 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
1939 if (overlaps(peer_req->i.sector, peer_req->i.size,
1940 rs_req->i.sector, rs_req->i.size)) {
1945 spin_unlock_irq(&device->resource->req_lock);
1950 /* Called from receive_Data.
1951 * Synchronize packets on sock with packets on msock.
1953 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1954 * packet traveling on msock, they are still processed in the order they have
1957 * Note: we don't care for Ack packets overtaking P_DATA packets.
1959 * In case packet_seq is larger than device->peer_seq number, there are
1960 * outstanding packets on the msock. We wait for them to arrive.
1961 * In case we are the logically next packet, we update device->peer_seq
1962 * ourselves. Correctly handles 32bit wrap around.
1964 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1965 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1966 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1967 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1969 * returns 0 if we may process the packet,
1970 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1971 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
1973 struct drbd_device *device = peer_device->device;
1978 if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
1981 spin_lock(&device->peer_seq_lock);
1983 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
1984 device->peer_seq = seq_max(device->peer_seq, peer_seq);
1988 if (signal_pending(current)) {
1994 tp = rcu_dereference(first_peer_device(device)->connection->net_conf)->two_primaries;
2000 /* Only need to wait if two_primaries is enabled */
2001 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2002 spin_unlock(&device->peer_seq_lock);
2004 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2006 timeout = schedule_timeout(timeout);
2007 spin_lock(&device->peer_seq_lock);
2010 drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2014 spin_unlock(&device->peer_seq_lock);
2015 finish_wait(&device->seq_wait, &wait);
2019 /* see also bio_flags_to_wire()
2020 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2021 * flags and back. We may replicate to other kernel versions. */
2022 static unsigned long wire_flags_to_bio(u32 dpf)
2024 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2025 (dpf & DP_FUA ? REQ_FUA : 0) |
2026 (dpf & DP_FLUSH ? REQ_FLUSH : 0) |
2027 (dpf & DP_DISCARD ? REQ_DISCARD : 0);
2030 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2033 struct drbd_interval *i;
2036 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2037 struct drbd_request *req;
2038 struct bio_and_error m;
2042 req = container_of(i, struct drbd_request, i);
2043 if (!(req->rq_state & RQ_POSTPONED))
2045 req->rq_state &= ~RQ_POSTPONED;
2046 __req_mod(req, NEG_ACKED, &m);
2047 spin_unlock_irq(&device->resource->req_lock);
2049 complete_master_bio(device, &m);
2050 spin_lock_irq(&device->resource->req_lock);
2055 static int handle_write_conflicts(struct drbd_device *device,
2056 struct drbd_peer_request *peer_req)
2058 struct drbd_connection *connection = peer_req->peer_device->connection;
2059 bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2060 sector_t sector = peer_req->i.sector;
2061 const unsigned int size = peer_req->i.size;
2062 struct drbd_interval *i;
2067 * Inserting the peer request into the write_requests tree will prevent
2068 * new conflicting local requests from being added.
2070 drbd_insert_interval(&device->write_requests, &peer_req->i);
2073 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2074 if (i == &peer_req->i)
2079 * Our peer has sent a conflicting remote request; this
2080 * should not happen in a two-node setup. Wait for the
2081 * earlier peer request to complete.
2083 err = drbd_wait_misc(device, i);
2089 equal = i->sector == sector && i->size == size;
2090 if (resolve_conflicts) {
2092 * If the peer request is fully contained within the
2093 * overlapping request, it can be considered overwritten
2094 * and thus superseded; otherwise, it will be retried
2095 * once all overlapping requests have completed.
2097 bool superseded = i->sector <= sector && i->sector +
2098 (i->size >> 9) >= sector + (size >> 9);
2101 drbd_alert(device, "Concurrent writes detected: "
2102 "local=%llus +%u, remote=%llus +%u, "
2103 "assuming %s came first\n",
2104 (unsigned long long)i->sector, i->size,
2105 (unsigned long long)sector, size,
2106 superseded ? "local" : "remote");
2108 inc_unacked(device);
2109 peer_req->w.cb = superseded ? e_send_superseded :
2111 list_add_tail(&peer_req->w.list, &device->done_ee);
2112 wake_asender(connection);
2117 struct drbd_request *req =
2118 container_of(i, struct drbd_request, i);
2121 drbd_alert(device, "Concurrent writes detected: "
2122 "local=%llus +%u, remote=%llus +%u\n",
2123 (unsigned long long)i->sector, i->size,
2124 (unsigned long long)sector, size);
2126 if (req->rq_state & RQ_LOCAL_PENDING ||
2127 !(req->rq_state & RQ_POSTPONED)) {
2129 * Wait for the node with the discard flag to
2130 * decide if this request has been superseded
2131 * or needs to be retried.
2132 * Requests that have been superseded will
2133 * disappear from the write_requests tree.
2135 * In addition, wait for the conflicting
2136 * request to finish locally before submitting
2137 * the conflicting peer request.
2139 err = drbd_wait_misc(device, &req->i);
2141 _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2142 fail_postponed_requests(device, sector, size);
2148 * Remember to restart the conflicting requests after
2149 * the new peer request has completed.
2151 peer_req->flags |= EE_RESTART_REQUESTS;
2158 drbd_remove_epoch_entry_interval(device, peer_req);
2162 /* mirrored write */
2163 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2165 struct drbd_peer_device *peer_device;
2166 struct drbd_device *device;
2168 struct drbd_peer_request *peer_req;
2169 struct p_data *p = pi->data;
2170 u32 peer_seq = be32_to_cpu(p->seq_num);
2175 peer_device = conn_peer_device(connection, pi->vnr);
2178 device = peer_device->device;
2180 if (!get_ldev(device)) {
2183 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2184 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2185 atomic_inc(&connection->current_epoch->epoch_size);
2186 err2 = drbd_drain_block(peer_device, pi->size);
2193 * Corresponding put_ldev done either below (on various errors), or in
2194 * drbd_peer_request_endio, if we successfully submit the data at the
2195 * end of this function.
2198 sector = be64_to_cpu(p->sector);
2199 peer_req = read_in_block(peer_device, p->block_id, sector, pi->size);
2205 peer_req->w.cb = e_end_block;
2207 dp_flags = be32_to_cpu(p->dp_flags);
2208 rw |= wire_flags_to_bio(dp_flags);
2209 if (peer_req->pages == NULL) {
2210 D_ASSERT(device, peer_req->i.size == 0);
2211 D_ASSERT(device, dp_flags & DP_FLUSH);
2214 if (dp_flags & DP_MAY_SET_IN_SYNC)
2215 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2217 spin_lock(&connection->epoch_lock);
2218 peer_req->epoch = connection->current_epoch;
2219 atomic_inc(&peer_req->epoch->epoch_size);
2220 atomic_inc(&peer_req->epoch->active);
2221 spin_unlock(&connection->epoch_lock);
2224 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2227 peer_req->flags |= EE_IN_INTERVAL_TREE;
2228 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2230 goto out_interrupted;
2231 spin_lock_irq(&device->resource->req_lock);
2232 err = handle_write_conflicts(device, peer_req);
2234 spin_unlock_irq(&device->resource->req_lock);
2235 if (err == -ENOENT) {
2239 goto out_interrupted;
2242 update_peer_seq(peer_device, peer_seq);
2243 spin_lock_irq(&device->resource->req_lock);
2245 list_add(&peer_req->w.list, &device->active_ee);
2246 spin_unlock_irq(&device->resource->req_lock);
2248 if (device->state.conn == C_SYNC_TARGET)
2249 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2251 if (peer_device->connection->agreed_pro_version < 100) {
2253 switch (rcu_dereference(peer_device->connection->net_conf)->wire_protocol) {
2255 dp_flags |= DP_SEND_WRITE_ACK;
2258 dp_flags |= DP_SEND_RECEIVE_ACK;
2264 if (dp_flags & DP_SEND_WRITE_ACK) {
2265 peer_req->flags |= EE_SEND_WRITE_ACK;
2266 inc_unacked(device);
2267 /* corresponding dec_unacked() in e_end_block()
2268 * respective _drbd_clear_done_ee */
2271 if (dp_flags & DP_SEND_RECEIVE_ACK) {
2272 /* I really don't like it that the receiver thread
2273 * sends on the msock, but anyways */
2274 drbd_send_ack(first_peer_device(device), P_RECV_ACK, peer_req);
2277 if (device->state.pdsk < D_INCONSISTENT) {
2278 /* In case we have the only disk of the cluster, */
2279 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2280 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2281 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2282 drbd_al_begin_io(device, &peer_req->i, true);
2285 err = drbd_submit_peer_request(device, peer_req, rw, DRBD_FAULT_DT_WR);
2289 /* don't care for the reason here */
2290 drbd_err(device, "submit failed, triggering re-connect\n");
2291 spin_lock_irq(&device->resource->req_lock);
2292 list_del(&peer_req->w.list);
2293 drbd_remove_epoch_entry_interval(device, peer_req);
2294 spin_unlock_irq(&device->resource->req_lock);
2295 if (peer_req->flags & EE_CALL_AL_COMPLETE_IO)
2296 drbd_al_complete_io(device, &peer_req->i);
2299 drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT + EV_CLEANUP);
2301 drbd_free_peer_req(device, peer_req);
2305 /* We may throttle resync, if the lower device seems to be busy,
2306 * and current sync rate is above c_min_rate.
2308 * To decide whether or not the lower device is busy, we use a scheme similar
2309 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2310 * (more than 64 sectors) of activity we cannot account for with our own resync
2311 * activity, it obviously is "busy".
2313 * The current sync rate used here uses only the most recent two step marks,
2314 * to have a short time average so we can react faster.
2316 int drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector)
2318 struct gendisk *disk = device->ldev->backing_bdev->bd_contains->bd_disk;
2319 unsigned long db, dt, dbdt;
2320 struct lc_element *tmp;
2323 unsigned int c_min_rate;
2326 c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2329 /* feature disabled? */
2330 if (c_min_rate == 0)
2333 spin_lock_irq(&device->al_lock);
2334 tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2336 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2337 if (test_bit(BME_PRIORITY, &bm_ext->flags)) {
2338 spin_unlock_irq(&device->al_lock);
2341 /* Do not slow down if app IO is already waiting for this extent */
2343 spin_unlock_irq(&device->al_lock);
2345 curr_events = (int)part_stat_read(&disk->part0, sectors[0]) +
2346 (int)part_stat_read(&disk->part0, sectors[1]) -
2347 atomic_read(&device->rs_sect_ev);
2349 if (!device->rs_last_events || curr_events - device->rs_last_events > 64) {
2350 unsigned long rs_left;
2353 device->rs_last_events = curr_events;
2355 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2357 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2359 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2360 rs_left = device->ov_left;
2362 rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2364 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2367 db = device->rs_mark_left[i] - rs_left;
2368 dbdt = Bit2KB(db/dt);
2370 if (dbdt > c_min_rate)
2377 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2379 struct drbd_peer_device *peer_device;
2380 struct drbd_device *device;
2383 struct drbd_peer_request *peer_req;
2384 struct digest_info *di = NULL;
2386 unsigned int fault_type;
2387 struct p_block_req *p = pi->data;
2389 peer_device = conn_peer_device(connection, pi->vnr);
2392 device = peer_device->device;
2393 capacity = drbd_get_capacity(device->this_bdev);
2395 sector = be64_to_cpu(p->sector);
2396 size = be32_to_cpu(p->blksize);
2398 if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2399 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2400 (unsigned long long)sector, size);
2403 if (sector + (size>>9) > capacity) {
2404 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2405 (unsigned long long)sector, size);
2409 if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2412 case P_DATA_REQUEST:
2413 drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2415 case P_RS_DATA_REQUEST:
2416 case P_CSUM_RS_REQUEST:
2418 drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2422 dec_rs_pending(device);
2423 drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2428 if (verb && __ratelimit(&drbd_ratelimit_state))
2429 drbd_err(device, "Can not satisfy peer's read request, "
2430 "no local data.\n");
2432 /* drain possibly payload */
2433 return drbd_drain_block(peer_device, pi->size);
2436 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2437 * "criss-cross" setup, that might cause write-out on some other DRBD,
2438 * which in turn might block on the other node at this very place. */
2439 peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size, GFP_NOIO);
2446 case P_DATA_REQUEST:
2447 peer_req->w.cb = w_e_end_data_req;
2448 fault_type = DRBD_FAULT_DT_RD;
2449 /* application IO, don't drbd_rs_begin_io */
2452 case P_RS_DATA_REQUEST:
2453 peer_req->w.cb = w_e_end_rsdata_req;
2454 fault_type = DRBD_FAULT_RS_RD;
2455 /* used in the sector offset progress display */
2456 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2460 case P_CSUM_RS_REQUEST:
2461 fault_type = DRBD_FAULT_RS_RD;
2462 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2466 di->digest_size = pi->size;
2467 di->digest = (((char *)di)+sizeof(struct digest_info));
2469 peer_req->digest = di;
2470 peer_req->flags |= EE_HAS_DIGEST;
2472 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2475 if (pi->cmd == P_CSUM_RS_REQUEST) {
2476 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2477 peer_req->w.cb = w_e_end_csum_rs_req;
2478 /* used in the sector offset progress display */
2479 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2480 } else if (pi->cmd == P_OV_REPLY) {
2481 /* track progress, we may need to throttle */
2482 atomic_add(size >> 9, &device->rs_sect_in);
2483 peer_req->w.cb = w_e_end_ov_reply;
2484 dec_rs_pending(device);
2485 /* drbd_rs_begin_io done when we sent this request,
2486 * but accounting still needs to be done. */
2487 goto submit_for_resync;
2492 if (device->ov_start_sector == ~(sector_t)0 &&
2493 peer_device->connection->agreed_pro_version >= 90) {
2494 unsigned long now = jiffies;
2496 device->ov_start_sector = sector;
2497 device->ov_position = sector;
2498 device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2499 device->rs_total = device->ov_left;
2500 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2501 device->rs_mark_left[i] = device->ov_left;
2502 device->rs_mark_time[i] = now;
2504 drbd_info(device, "Online Verify start sector: %llu\n",
2505 (unsigned long long)sector);
2507 peer_req->w.cb = w_e_end_ov_req;
2508 fault_type = DRBD_FAULT_RS_RD;
2515 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2516 * wrt the receiver, but it is not as straightforward as it may seem.
2517 * Various places in the resync start and stop logic assume resync
2518 * requests are processed in order, requeuing this on the worker thread
2519 * introduces a bunch of new code for synchronization between threads.
2521 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2522 * "forever", throttling after drbd_rs_begin_io will lock that extent
2523 * for application writes for the same time. For now, just throttle
2524 * here, where the rest of the code expects the receiver to sleep for
2528 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2529 * this defers syncer requests for some time, before letting at least
2530 * on request through. The resync controller on the receiving side
2531 * will adapt to the incoming rate accordingly.
2533 * We cannot throttle here if remote is Primary/SyncTarget:
2534 * we would also throttle its application reads.
2535 * In that case, throttling is done on the SyncTarget only.
2537 if (device->state.peer != R_PRIMARY && drbd_rs_should_slow_down(device, sector))
2538 schedule_timeout_uninterruptible(HZ/10);
2539 if (drbd_rs_begin_io(device, sector))
2543 atomic_add(size >> 9, &device->rs_sect_ev);
2546 inc_unacked(device);
2547 spin_lock_irq(&device->resource->req_lock);
2548 list_add_tail(&peer_req->w.list, &device->read_ee);
2549 spin_unlock_irq(&device->resource->req_lock);
2551 if (drbd_submit_peer_request(device, peer_req, READ, fault_type) == 0)
2554 /* don't care for the reason here */
2555 drbd_err(device, "submit failed, triggering re-connect\n");
2556 spin_lock_irq(&device->resource->req_lock);
2557 list_del(&peer_req->w.list);
2558 spin_unlock_irq(&device->resource->req_lock);
2559 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2563 drbd_free_peer_req(device, peer_req);
2568 * drbd_asb_recover_0p - Recover after split-brain with no remaining primaries
2570 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
2572 struct drbd_device *device = peer_device->device;
2573 int self, peer, rv = -100;
2574 unsigned long ch_self, ch_peer;
2575 enum drbd_after_sb_p after_sb_0p;
2577 self = device->ldev->md.uuid[UI_BITMAP] & 1;
2578 peer = device->p_uuid[UI_BITMAP] & 1;
2580 ch_peer = device->p_uuid[UI_SIZE];
2581 ch_self = device->comm_bm_set;
2584 after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
2586 switch (after_sb_0p) {
2588 case ASB_DISCARD_SECONDARY:
2589 case ASB_CALL_HELPER:
2591 drbd_err(device, "Configuration error.\n");
2593 case ASB_DISCONNECT:
2595 case ASB_DISCARD_YOUNGER_PRI:
2596 if (self == 0 && peer == 1) {
2600 if (self == 1 && peer == 0) {
2604 /* Else fall through to one of the other strategies... */
2605 case ASB_DISCARD_OLDER_PRI:
2606 if (self == 0 && peer == 1) {
2610 if (self == 1 && peer == 0) {
2614 /* Else fall through to one of the other strategies... */
2615 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
2616 "Using discard-least-changes instead\n");
2617 case ASB_DISCARD_ZERO_CHG:
2618 if (ch_peer == 0 && ch_self == 0) {
2619 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2623 if (ch_peer == 0) { rv = 1; break; }
2624 if (ch_self == 0) { rv = -1; break; }
2626 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
2628 case ASB_DISCARD_LEAST_CHG:
2629 if (ch_self < ch_peer)
2631 else if (ch_self > ch_peer)
2633 else /* ( ch_self == ch_peer ) */
2634 /* Well, then use something else. */
2635 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2638 case ASB_DISCARD_LOCAL:
2641 case ASB_DISCARD_REMOTE:
2649 * drbd_asb_recover_1p - Recover after split-brain with one remaining primary
2651 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
2653 struct drbd_device *device = peer_device->device;
2655 enum drbd_after_sb_p after_sb_1p;
2658 after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
2660 switch (after_sb_1p) {
2661 case ASB_DISCARD_YOUNGER_PRI:
2662 case ASB_DISCARD_OLDER_PRI:
2663 case ASB_DISCARD_LEAST_CHG:
2664 case ASB_DISCARD_LOCAL:
2665 case ASB_DISCARD_REMOTE:
2666 case ASB_DISCARD_ZERO_CHG:
2667 drbd_err(device, "Configuration error.\n");
2669 case ASB_DISCONNECT:
2672 hg = drbd_asb_recover_0p(peer_device);
2673 if (hg == -1 && device->state.role == R_SECONDARY)
2675 if (hg == 1 && device->state.role == R_PRIMARY)
2679 rv = drbd_asb_recover_0p(peer_device);
2681 case ASB_DISCARD_SECONDARY:
2682 return device->state.role == R_PRIMARY ? 1 : -1;
2683 case ASB_CALL_HELPER:
2684 hg = drbd_asb_recover_0p(peer_device);
2685 if (hg == -1 && device->state.role == R_PRIMARY) {
2686 enum drbd_state_rv rv2;
2688 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2689 * we might be here in C_WF_REPORT_PARAMS which is transient.
2690 * we do not need to wait for the after state change work either. */
2691 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
2692 if (rv2 != SS_SUCCESS) {
2693 drbd_khelper(device, "pri-lost-after-sb");
2695 drbd_warn(device, "Successfully gave up primary role.\n");
2706 * drbd_asb_recover_2p - Recover after split-brain with two remaining primaries
2708 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
2710 struct drbd_device *device = peer_device->device;
2712 enum drbd_after_sb_p after_sb_2p;
2715 after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
2717 switch (after_sb_2p) {
2718 case ASB_DISCARD_YOUNGER_PRI:
2719 case ASB_DISCARD_OLDER_PRI:
2720 case ASB_DISCARD_LEAST_CHG:
2721 case ASB_DISCARD_LOCAL:
2722 case ASB_DISCARD_REMOTE:
2724 case ASB_DISCARD_SECONDARY:
2725 case ASB_DISCARD_ZERO_CHG:
2726 drbd_err(device, "Configuration error.\n");
2729 rv = drbd_asb_recover_0p(peer_device);
2731 case ASB_DISCONNECT:
2733 case ASB_CALL_HELPER:
2734 hg = drbd_asb_recover_0p(peer_device);
2736 enum drbd_state_rv rv2;
2738 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2739 * we might be here in C_WF_REPORT_PARAMS which is transient.
2740 * we do not need to wait for the after state change work either. */
2741 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
2742 if (rv2 != SS_SUCCESS) {
2743 drbd_khelper(device, "pri-lost-after-sb");
2745 drbd_warn(device, "Successfully gave up primary role.\n");
2755 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
2756 u64 bits, u64 flags)
2759 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
2762 drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2764 (unsigned long long)uuid[UI_CURRENT],
2765 (unsigned long long)uuid[UI_BITMAP],
2766 (unsigned long long)uuid[UI_HISTORY_START],
2767 (unsigned long long)uuid[UI_HISTORY_END],
2768 (unsigned long long)bits,
2769 (unsigned long long)flags);
2773 100 after split brain try auto recover
2774 2 C_SYNC_SOURCE set BitMap
2775 1 C_SYNC_SOURCE use BitMap
2777 -1 C_SYNC_TARGET use BitMap
2778 -2 C_SYNC_TARGET set BitMap
2779 -100 after split brain, disconnect
2780 -1000 unrelated data
2781 -1091 requires proto 91
2782 -1096 requires proto 96
2784 static int drbd_uuid_compare(struct drbd_device *device, int *rule_nr) __must_hold(local)
2789 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2790 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
2793 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
2797 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
2798 peer != UUID_JUST_CREATED)
2802 if (self != UUID_JUST_CREATED &&
2803 (peer == UUID_JUST_CREATED || peer == (u64)0))
2807 int rct, dc; /* roles at crash time */
2809 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
2811 if (first_peer_device(device)->connection->agreed_pro_version < 91)
2814 if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
2815 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
2816 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
2817 drbd_uuid_move_history(device);
2818 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
2819 device->ldev->md.uuid[UI_BITMAP] = 0;
2821 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
2822 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
2825 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
2832 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
2834 if (first_peer_device(device)->connection->agreed_pro_version < 91)
2837 if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
2838 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
2839 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2841 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
2842 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
2843 device->p_uuid[UI_BITMAP] = 0UL;
2845 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
2848 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
2855 /* Common power [off|failure] */
2856 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
2857 (device->p_uuid[UI_FLAGS] & 2);
2858 /* lowest bit is set when we were primary,
2859 * next bit (weight 2) is set when peer was primary */
2863 case 0: /* !self_pri && !peer_pri */ return 0;
2864 case 1: /* self_pri && !peer_pri */ return 1;
2865 case 2: /* !self_pri && peer_pri */ return -1;
2866 case 3: /* self_pri && peer_pri */
2867 dc = test_bit(RESOLVE_CONFLICTS, &first_peer_device(device)->connection->flags);
2873 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
2878 peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
2880 if (first_peer_device(device)->connection->agreed_pro_version < 96 ?
2881 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
2882 (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
2883 peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
2884 /* The last P_SYNC_UUID did not get though. Undo the last start of
2885 resync as sync source modifications of the peer's UUIDs. */
2887 if (first_peer_device(device)->connection->agreed_pro_version < 91)
2890 device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
2891 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
2893 drbd_info(device, "Lost last syncUUID packet, corrected:\n");
2894 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
2901 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2902 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2903 peer = device->p_uuid[i] & ~((u64)1);
2909 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2910 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
2915 self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2917 if (first_peer_device(device)->connection->agreed_pro_version < 96 ?
2918 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
2919 (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
2920 self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
2921 /* The last P_SYNC_UUID did not get though. Undo the last start of
2922 resync as sync source modifications of our UUIDs. */
2924 if (first_peer_device(device)->connection->agreed_pro_version < 91)
2927 __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
2928 __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
2930 drbd_info(device, "Last syncUUID did not get through, corrected:\n");
2931 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
2932 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
2940 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
2941 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2942 self = device->ldev->md.uuid[i] & ~((u64)1);
2948 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2949 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
2950 if (self == peer && self != ((u64)0))
2954 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2955 self = device->ldev->md.uuid[i] & ~((u64)1);
2956 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
2957 peer = device->p_uuid[j] & ~((u64)1);
2966 /* drbd_sync_handshake() returns the new conn state on success, or
2967 CONN_MASK (-1) on failure.
2969 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
2970 enum drbd_role peer_role,
2971 enum drbd_disk_state peer_disk) __must_hold(local)
2973 struct drbd_device *device = peer_device->device;
2974 enum drbd_conns rv = C_MASK;
2975 enum drbd_disk_state mydisk;
2976 struct net_conf *nc;
2977 int hg, rule_nr, rr_conflict, tentative;
2979 mydisk = device->state.disk;
2980 if (mydisk == D_NEGOTIATING)
2981 mydisk = device->new_state_tmp.disk;
2983 drbd_info(device, "drbd_sync_handshake:\n");
2985 spin_lock_irq(&device->ldev->md.uuid_lock);
2986 drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
2987 drbd_uuid_dump(device, "peer", device->p_uuid,
2988 device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
2990 hg = drbd_uuid_compare(device, &rule_nr);
2991 spin_unlock_irq(&device->ldev->md.uuid_lock);
2993 drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
2996 drbd_alert(device, "Unrelated data, aborting!\n");
3000 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3004 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3005 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
3006 int f = (hg == -100) || abs(hg) == 2;
3007 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3010 drbd_info(device, "Becoming sync %s due to disk states.\n",
3011 hg > 0 ? "source" : "target");
3015 drbd_khelper(device, "initial-split-brain");
3018 nc = rcu_dereference(peer_device->connection->net_conf);
3020 if (hg == 100 || (hg == -100 && nc->always_asbp)) {
3021 int pcount = (device->state.role == R_PRIMARY)
3022 + (peer_role == R_PRIMARY);
3023 int forced = (hg == -100);
3027 hg = drbd_asb_recover_0p(peer_device);
3030 hg = drbd_asb_recover_1p(peer_device);
3033 hg = drbd_asb_recover_2p(peer_device);
3036 if (abs(hg) < 100) {
3037 drbd_warn(device, "Split-Brain detected, %d primaries, "
3038 "automatically solved. Sync from %s node\n",
3039 pcount, (hg < 0) ? "peer" : "this");
3041 drbd_warn(device, "Doing a full sync, since"
3042 " UUIDs where ambiguous.\n");
3049 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3051 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3055 drbd_warn(device, "Split-Brain detected, manually solved. "
3056 "Sync from %s node\n",
3057 (hg < 0) ? "peer" : "this");
3059 rr_conflict = nc->rr_conflict;
3060 tentative = nc->tentative;
3064 /* FIXME this log message is not correct if we end up here
3065 * after an attempted attach on a diskless node.
3066 * We just refuse to attach -- well, we drop the "connection"
3067 * to that disk, in a way... */
3068 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3069 drbd_khelper(device, "split-brain");
3073 if (hg > 0 && mydisk <= D_INCONSISTENT) {
3074 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3078 if (hg < 0 && /* by intention we do not use mydisk here. */
3079 device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3080 switch (rr_conflict) {
3081 case ASB_CALL_HELPER:
3082 drbd_khelper(device, "pri-lost");
3084 case ASB_DISCONNECT:
3085 drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3088 drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3093 if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3095 drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3097 drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3098 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3099 abs(hg) >= 2 ? "full" : "bit-map based");
3104 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3105 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3106 BM_LOCKED_SET_ALLOWED))
3110 if (hg > 0) { /* become sync source. */
3112 } else if (hg < 0) { /* become sync target */
3116 if (drbd_bm_total_weight(device)) {
3117 drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3118 drbd_bm_total_weight(device));
3125 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3127 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3128 if (peer == ASB_DISCARD_REMOTE)
3129 return ASB_DISCARD_LOCAL;
3131 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3132 if (peer == ASB_DISCARD_LOCAL)
3133 return ASB_DISCARD_REMOTE;
3135 /* everything else is valid if they are equal on both sides. */
3139 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3141 struct p_protocol *p = pi->data;
3142 enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3143 int p_proto, p_discard_my_data, p_two_primaries, cf;
3144 struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3145 char integrity_alg[SHARED_SECRET_MAX] = "";
3146 struct crypto_hash *peer_integrity_tfm = NULL;
3147 void *int_dig_in = NULL, *int_dig_vv = NULL;
3149 p_proto = be32_to_cpu(p->protocol);
3150 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
3151 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
3152 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
3153 p_two_primaries = be32_to_cpu(p->two_primaries);
3154 cf = be32_to_cpu(p->conn_flags);
3155 p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3157 if (connection->agreed_pro_version >= 87) {
3160 if (pi->size > sizeof(integrity_alg))
3162 err = drbd_recv_all(connection, integrity_alg, pi->size);
3165 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3168 if (pi->cmd != P_PROTOCOL_UPDATE) {
3169 clear_bit(CONN_DRY_RUN, &connection->flags);
3171 if (cf & CF_DRY_RUN)
3172 set_bit(CONN_DRY_RUN, &connection->flags);
3175 nc = rcu_dereference(connection->net_conf);
3177 if (p_proto != nc->wire_protocol) {
3178 drbd_err(connection, "incompatible %s settings\n", "protocol");
3179 goto disconnect_rcu_unlock;
3182 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3183 drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3184 goto disconnect_rcu_unlock;
3187 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3188 drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3189 goto disconnect_rcu_unlock;
3192 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3193 drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3194 goto disconnect_rcu_unlock;
3197 if (p_discard_my_data && nc->discard_my_data) {
3198 drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3199 goto disconnect_rcu_unlock;
3202 if (p_two_primaries != nc->two_primaries) {
3203 drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3204 goto disconnect_rcu_unlock;
3207 if (strcmp(integrity_alg, nc->integrity_alg)) {
3208 drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3209 goto disconnect_rcu_unlock;
3215 if (integrity_alg[0]) {
3219 * We can only change the peer data integrity algorithm
3220 * here. Changing our own data integrity algorithm
3221 * requires that we send a P_PROTOCOL_UPDATE packet at
3222 * the same time; otherwise, the peer has no way to
3223 * tell between which packets the algorithm should
3227 peer_integrity_tfm = crypto_alloc_hash(integrity_alg, 0, CRYPTO_ALG_ASYNC);
3228 if (!peer_integrity_tfm) {
3229 drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3234 hash_size = crypto_hash_digestsize(peer_integrity_tfm);
3235 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3236 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3237 if (!(int_dig_in && int_dig_vv)) {
3238 drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3243 new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3244 if (!new_net_conf) {
3245 drbd_err(connection, "Allocation of new net_conf failed\n");
3249 mutex_lock(&connection->data.mutex);
3250 mutex_lock(&connection->resource->conf_update);
3251 old_net_conf = connection->net_conf;
3252 *new_net_conf = *old_net_conf;
3254 new_net_conf->wire_protocol = p_proto;
3255 new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3256 new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3257 new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3258 new_net_conf->two_primaries = p_two_primaries;
3260 rcu_assign_pointer(connection->net_conf, new_net_conf);
3261 mutex_unlock(&connection->resource->conf_update);
3262 mutex_unlock(&connection->data.mutex);
3264 crypto_free_hash(connection->peer_integrity_tfm);
3265 kfree(connection->int_dig_in);
3266 kfree(connection->int_dig_vv);
3267 connection->peer_integrity_tfm = peer_integrity_tfm;
3268 connection->int_dig_in = int_dig_in;
3269 connection->int_dig_vv = int_dig_vv;
3271 if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3272 drbd_info(connection, "peer data-integrity-alg: %s\n",
3273 integrity_alg[0] ? integrity_alg : "(none)");
3276 kfree(old_net_conf);
3279 disconnect_rcu_unlock:
3282 crypto_free_hash(peer_integrity_tfm);
3285 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3290 * input: alg name, feature name
3291 * return: NULL (alg name was "")
3292 * ERR_PTR(error) if something goes wrong
3293 * or the crypto hash ptr, if it worked out ok. */
3295 struct crypto_hash *drbd_crypto_alloc_digest_safe(const struct drbd_device *device,
3296 const char *alg, const char *name)
3298 struct crypto_hash *tfm;
3303 tfm = crypto_alloc_hash(alg, 0, CRYPTO_ALG_ASYNC);
3305 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3306 alg, name, PTR_ERR(tfm));
3312 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3314 void *buffer = connection->data.rbuf;
3315 int size = pi->size;
3318 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3319 s = drbd_recv(connection, buffer, s);
3333 * config_unknown_volume - device configuration command for unknown volume
3335 * When a device is added to an existing connection, the node on which the
3336 * device is added first will send configuration commands to its peer but the
3337 * peer will not know about the device yet. It will warn and ignore these
3338 * commands. Once the device is added on the second node, the second node will
3339 * send the same device configuration commands, but in the other direction.
3341 * (We can also end up here if drbd is misconfigured.)
3343 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3345 drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3346 cmdname(pi->cmd), pi->vnr);
3347 return ignore_remaining_packet(connection, pi);
3350 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3352 struct drbd_peer_device *peer_device;
3353 struct drbd_device *device;
3354 struct p_rs_param_95 *p;
3355 unsigned int header_size, data_size, exp_max_sz;
3356 struct crypto_hash *verify_tfm = NULL;
3357 struct crypto_hash *csums_tfm = NULL;
3358 struct net_conf *old_net_conf, *new_net_conf = NULL;
3359 struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3360 const int apv = connection->agreed_pro_version;
3361 struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3365 peer_device = conn_peer_device(connection, pi->vnr);
3367 return config_unknown_volume(connection, pi);
3368 device = peer_device->device;
3370 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
3371 : apv == 88 ? sizeof(struct p_rs_param)
3373 : apv <= 94 ? sizeof(struct p_rs_param_89)
3374 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3376 if (pi->size > exp_max_sz) {
3377 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3378 pi->size, exp_max_sz);
3383 header_size = sizeof(struct p_rs_param);
3384 data_size = pi->size - header_size;
3385 } else if (apv <= 94) {
3386 header_size = sizeof(struct p_rs_param_89);
3387 data_size = pi->size - header_size;
3388 D_ASSERT(device, data_size == 0);
3390 header_size = sizeof(struct p_rs_param_95);
3391 data_size = pi->size - header_size;
3392 D_ASSERT(device, data_size == 0);
3395 /* initialize verify_alg and csums_alg */
3397 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
3399 err = drbd_recv_all(peer_device->connection, p, header_size);
3403 mutex_lock(&connection->resource->conf_update);
3404 old_net_conf = peer_device->connection->net_conf;
3405 if (get_ldev(device)) {
3406 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3407 if (!new_disk_conf) {
3409 mutex_unlock(&connection->resource->conf_update);
3410 drbd_err(device, "Allocation of new disk_conf failed\n");
3414 old_disk_conf = device->ldev->disk_conf;
3415 *new_disk_conf = *old_disk_conf;
3417 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3422 if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3423 drbd_err(device, "verify-alg of wrong size, "
3424 "peer wants %u, accepting only up to %u byte\n",
3425 data_size, SHARED_SECRET_MAX);
3430 err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3433 /* we expect NUL terminated string */
3434 /* but just in case someone tries to be evil */
3435 D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3436 p->verify_alg[data_size-1] = 0;
3438 } else /* apv >= 89 */ {
3439 /* we still expect NUL terminated strings */
3440 /* but just in case someone tries to be evil */
3441 D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3442 D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3443 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3444 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3447 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3448 if (device->state.conn == C_WF_REPORT_PARAMS) {
3449 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3450 old_net_conf->verify_alg, p->verify_alg);
3453 verify_tfm = drbd_crypto_alloc_digest_safe(device,
3454 p->verify_alg, "verify-alg");
3455 if (IS_ERR(verify_tfm)) {
3461 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3462 if (device->state.conn == C_WF_REPORT_PARAMS) {
3463 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3464 old_net_conf->csums_alg, p->csums_alg);
3467 csums_tfm = drbd_crypto_alloc_digest_safe(device,
3468 p->csums_alg, "csums-alg");
3469 if (IS_ERR(csums_tfm)) {
3475 if (apv > 94 && new_disk_conf) {
3476 new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3477 new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3478 new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3479 new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3481 fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3482 if (fifo_size != device->rs_plan_s->size) {
3483 new_plan = fifo_alloc(fifo_size);
3485 drbd_err(device, "kmalloc of fifo_buffer failed");
3492 if (verify_tfm || csums_tfm) {
3493 new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3494 if (!new_net_conf) {
3495 drbd_err(device, "Allocation of new net_conf failed\n");
3499 *new_net_conf = *old_net_conf;
3502 strcpy(new_net_conf->verify_alg, p->verify_alg);
3503 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3504 crypto_free_hash(peer_device->connection->verify_tfm);
3505 peer_device->connection->verify_tfm = verify_tfm;
3506 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3509 strcpy(new_net_conf->csums_alg, p->csums_alg);
3510 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3511 crypto_free_hash(peer_device->connection->csums_tfm);
3512 peer_device->connection->csums_tfm = csums_tfm;
3513 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
3515 rcu_assign_pointer(connection->net_conf, new_net_conf);
3519 if (new_disk_conf) {
3520 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3525 old_plan = device->rs_plan_s;
3526 rcu_assign_pointer(device->rs_plan_s, new_plan);
3529 mutex_unlock(&connection->resource->conf_update);
3532 kfree(old_net_conf);
3533 kfree(old_disk_conf);
3539 if (new_disk_conf) {
3541 kfree(new_disk_conf);
3543 mutex_unlock(&connection->resource->conf_update);
3548 if (new_disk_conf) {
3550 kfree(new_disk_conf);
3552 mutex_unlock(&connection->resource->conf_update);
3553 /* just for completeness: actually not needed,
3554 * as this is not reached if csums_tfm was ok. */
3555 crypto_free_hash(csums_tfm);
3556 /* but free the verify_tfm again, if csums_tfm did not work out */
3557 crypto_free_hash(verify_tfm);
3558 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3562 /* warn if the arguments differ by more than 12.5% */
3563 static void warn_if_differ_considerably(struct drbd_device *device,
3564 const char *s, sector_t a, sector_t b)
3567 if (a == 0 || b == 0)
3569 d = (a > b) ? (a - b) : (b - a);
3570 if (d > (a>>3) || d > (b>>3))
3571 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
3572 (unsigned long long)a, (unsigned long long)b);
3575 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
3577 struct drbd_peer_device *peer_device;
3578 struct drbd_device *device;
3579 struct p_sizes *p = pi->data;
3580 enum determine_dev_size dd = DS_UNCHANGED;
3581 sector_t p_size, p_usize, my_usize;
3582 int ldsc = 0; /* local disk size changed */
3583 enum dds_flags ddsf;
3585 peer_device = conn_peer_device(connection, pi->vnr);
3587 return config_unknown_volume(connection, pi);
3588 device = peer_device->device;
3590 p_size = be64_to_cpu(p->d_size);
3591 p_usize = be64_to_cpu(p->u_size);
3593 /* just store the peer's disk size for now.
3594 * we still need to figure out whether we accept that. */
3595 device->p_size = p_size;
3597 if (get_ldev(device)) {
3599 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
3602 warn_if_differ_considerably(device, "lower level device sizes",
3603 p_size, drbd_get_max_capacity(device->ldev));
3604 warn_if_differ_considerably(device, "user requested size",
3607 /* if this is the first connect, or an otherwise expected
3608 * param exchange, choose the minimum */
3609 if (device->state.conn == C_WF_REPORT_PARAMS)
3610 p_usize = min_not_zero(my_usize, p_usize);
3612 /* Never shrink a device with usable data during connect.
3613 But allow online shrinking if we are connected. */
3614 if (drbd_new_dev_size(device, device->ldev, p_usize, 0) <
3615 drbd_get_capacity(device->this_bdev) &&
3616 device->state.disk >= D_OUTDATED &&
3617 device->state.conn < C_CONNECTED) {
3618 drbd_err(device, "The peer's disk size is too small!\n");
3619 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3624 if (my_usize != p_usize) {
3625 struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
3627 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3628 if (!new_disk_conf) {
3629 drbd_err(device, "Allocation of new disk_conf failed\n");
3634 mutex_lock(&connection->resource->conf_update);
3635 old_disk_conf = device->ldev->disk_conf;
3636 *new_disk_conf = *old_disk_conf;
3637 new_disk_conf->disk_size = p_usize;
3639 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3640 mutex_unlock(&connection->resource->conf_update);
3642 kfree(old_disk_conf);
3644 drbd_info(device, "Peer sets u_size to %lu sectors\n",
3645 (unsigned long)my_usize);
3651 ddsf = be16_to_cpu(p->dds_flags);
3652 if (get_ldev(device)) {
3653 dd = drbd_determine_dev_size(device, ddsf, NULL);
3657 drbd_md_sync(device);
3659 /* I am diskless, need to accept the peer's size. */
3660 drbd_set_my_capacity(device, p_size);
3663 device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
3664 drbd_reconsider_max_bio_size(device);
3666 if (get_ldev(device)) {
3667 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
3668 device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
3675 if (device->state.conn > C_WF_REPORT_PARAMS) {
3676 if (be64_to_cpu(p->c_size) !=
3677 drbd_get_capacity(device->this_bdev) || ldsc) {
3678 /* we have different sizes, probably peer
3679 * needs to know my new size... */
3680 drbd_send_sizes(peer_device, 0, ddsf);
3682 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
3683 (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
3684 if (device->state.pdsk >= D_INCONSISTENT &&
3685 device->state.disk >= D_INCONSISTENT) {
3686 if (ddsf & DDSF_NO_RESYNC)
3687 drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
3689 resync_after_online_grow(device);
3691 set_bit(RESYNC_AFTER_NEG, &device->flags);
3698 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
3700 struct drbd_peer_device *peer_device;
3701 struct drbd_device *device;
3702 struct p_uuids *p = pi->data;
3704 int i, updated_uuids = 0;
3706 peer_device = conn_peer_device(connection, pi->vnr);
3708 return config_unknown_volume(connection, pi);
3709 device = peer_device->device;
3711 p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
3713 drbd_err(device, "kmalloc of p_uuid failed\n");
3717 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
3718 p_uuid[i] = be64_to_cpu(p->uuid[i]);
3720 kfree(device->p_uuid);
3721 device->p_uuid = p_uuid;
3723 if (device->state.conn < C_CONNECTED &&
3724 device->state.disk < D_INCONSISTENT &&
3725 device->state.role == R_PRIMARY &&
3726 (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
3727 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
3728 (unsigned long long)device->ed_uuid);
3729 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3733 if (get_ldev(device)) {
3734 int skip_initial_sync =
3735 device->state.conn == C_CONNECTED &&
3736 peer_device->connection->agreed_pro_version >= 90 &&
3737 device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
3738 (p_uuid[UI_FLAGS] & 8);
3739 if (skip_initial_sync) {
3740 drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
3741 drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
3742 "clear_n_write from receive_uuids",
3743 BM_LOCKED_TEST_ALLOWED);
3744 _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
3745 _drbd_uuid_set(device, UI_BITMAP, 0);
3746 _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
3748 drbd_md_sync(device);
3752 } else if (device->state.disk < D_INCONSISTENT &&
3753 device->state.role == R_PRIMARY) {
3754 /* I am a diskless primary, the peer just created a new current UUID
3756 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
3759 /* Before we test for the disk state, we should wait until an eventually
3760 ongoing cluster wide state change is finished. That is important if
3761 we are primary and are detaching from our disk. We need to see the
3762 new disk state... */
3763 mutex_lock(device->state_mutex);
3764 mutex_unlock(device->state_mutex);
3765 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
3766 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
3769 drbd_print_uuids(device, "receiver updated UUIDs to");
3775 * convert_state() - Converts the peer's view of the cluster state to our point of view
3776 * @ps: The state as seen by the peer.
3778 static union drbd_state convert_state(union drbd_state ps)
3780 union drbd_state ms;
3782 static enum drbd_conns c_tab[] = {
3783 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
3784 [C_CONNECTED] = C_CONNECTED,
3786 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
3787 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
3788 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
3789 [C_VERIFY_S] = C_VERIFY_T,
3795 ms.conn = c_tab[ps.conn];
3800 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
3805 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
3807 struct drbd_peer_device *peer_device;
3808 struct drbd_device *device;
3809 struct p_req_state *p = pi->data;
3810 union drbd_state mask, val;
3811 enum drbd_state_rv rv;
3813 peer_device = conn_peer_device(connection, pi->vnr);
3816 device = peer_device->device;
3818 mask.i = be32_to_cpu(p->mask);
3819 val.i = be32_to_cpu(p->val);
3821 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
3822 mutex_is_locked(device->state_mutex)) {
3823 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
3827 mask = convert_state(mask);
3828 val = convert_state(val);
3830 rv = drbd_change_state(device, CS_VERBOSE, mask, val);
3831 drbd_send_sr_reply(peer_device, rv);
3833 drbd_md_sync(device);
3838 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
3840 struct p_req_state *p = pi->data;
3841 union drbd_state mask, val;
3842 enum drbd_state_rv rv;
3844 mask.i = be32_to_cpu(p->mask);
3845 val.i = be32_to_cpu(p->val);
3847 if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
3848 mutex_is_locked(&connection->cstate_mutex)) {
3849 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
3853 mask = convert_state(mask);
3854 val = convert_state(val);
3856 rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
3857 conn_send_sr_reply(connection, rv);
3862 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
3864 struct drbd_peer_device *peer_device;
3865 struct drbd_device *device;
3866 struct p_state *p = pi->data;
3867 union drbd_state os, ns, peer_state;
3868 enum drbd_disk_state real_peer_disk;
3869 enum chg_state_flags cs_flags;
3872 peer_device = conn_peer_device(connection, pi->vnr);
3874 return config_unknown_volume(connection, pi);
3875 device = peer_device->device;
3877 peer_state.i = be32_to_cpu(p->state);
3879 real_peer_disk = peer_state.disk;
3880 if (peer_state.disk == D_NEGOTIATING) {
3881 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
3882 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
3885 spin_lock_irq(&device->resource->req_lock);
3887 os = ns = drbd_read_state(device);
3888 spin_unlock_irq(&device->resource->req_lock);
3890 /* If some other part of the code (asender thread, timeout)
3891 * already decided to close the connection again,
3892 * we must not "re-establish" it here. */
3893 if (os.conn <= C_TEAR_DOWN)
3896 /* If this is the "end of sync" confirmation, usually the peer disk
3897 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
3898 * set) resync started in PausedSyncT, or if the timing of pause-/
3899 * unpause-sync events has been "just right", the peer disk may
3900 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
3902 if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
3903 real_peer_disk == D_UP_TO_DATE &&
3904 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
3905 /* If we are (becoming) SyncSource, but peer is still in sync
3906 * preparation, ignore its uptodate-ness to avoid flapping, it
3907 * will change to inconsistent once the peer reaches active
3909 * It may have changed syncer-paused flags, however, so we
3910 * cannot ignore this completely. */
3911 if (peer_state.conn > C_CONNECTED &&
3912 peer_state.conn < C_SYNC_SOURCE)
3913 real_peer_disk = D_INCONSISTENT;
3915 /* if peer_state changes to connected at the same time,
3916 * it explicitly notifies us that it finished resync.
3917 * Maybe we should finish it up, too? */
3918 else if (os.conn >= C_SYNC_SOURCE &&
3919 peer_state.conn == C_CONNECTED) {
3920 if (drbd_bm_total_weight(device) <= device->rs_failed)
3921 drbd_resync_finished(device);
3926 /* explicit verify finished notification, stop sector reached. */
3927 if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
3928 peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
3929 ov_out_of_sync_print(device);
3930 drbd_resync_finished(device);
3934 /* peer says his disk is inconsistent, while we think it is uptodate,
3935 * and this happens while the peer still thinks we have a sync going on,
3936 * but we think we are already done with the sync.
3937 * We ignore this to avoid flapping pdsk.
3938 * This should not happen, if the peer is a recent version of drbd. */
3939 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
3940 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
3941 real_peer_disk = D_UP_TO_DATE;
3943 if (ns.conn == C_WF_REPORT_PARAMS)
3944 ns.conn = C_CONNECTED;
3946 if (peer_state.conn == C_AHEAD)
3949 if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
3950 get_ldev_if_state(device, D_NEGOTIATING)) {
3951 int cr; /* consider resync */
3953 /* if we established a new connection */
3954 cr = (os.conn < C_CONNECTED);
3955 /* if we had an established connection
3956 * and one of the nodes newly attaches a disk */
3957 cr |= (os.conn == C_CONNECTED &&
3958 (peer_state.disk == D_NEGOTIATING ||
3959 os.disk == D_NEGOTIATING));
3960 /* if we have both been inconsistent, and the peer has been
3961 * forced to be UpToDate with --overwrite-data */
3962 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
3963 /* if we had been plain connected, and the admin requested to
3964 * start a sync by "invalidate" or "invalidate-remote" */
3965 cr |= (os.conn == C_CONNECTED &&
3966 (peer_state.conn >= C_STARTING_SYNC_S &&
3967 peer_state.conn <= C_WF_BITMAP_T));
3970 ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
3973 if (ns.conn == C_MASK) {
3974 ns.conn = C_CONNECTED;
3975 if (device->state.disk == D_NEGOTIATING) {
3976 drbd_force_state(device, NS(disk, D_FAILED));
3977 } else if (peer_state.disk == D_NEGOTIATING) {
3978 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
3979 peer_state.disk = D_DISKLESS;
3980 real_peer_disk = D_DISKLESS;
3982 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
3984 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
3985 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3991 spin_lock_irq(&device->resource->req_lock);
3992 if (os.i != drbd_read_state(device).i)
3994 clear_bit(CONSIDER_RESYNC, &device->flags);
3995 ns.peer = peer_state.role;
3996 ns.pdsk = real_peer_disk;
3997 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
3998 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
3999 ns.disk = device->new_state_tmp.disk;
4000 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4001 if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4002 test_bit(NEW_CUR_UUID, &device->flags)) {
4003 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4004 for temporal network outages! */
4005 spin_unlock_irq(&device->resource->req_lock);
4006 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4007 tl_clear(peer_device->connection);
4008 drbd_uuid_new_current(device);
4009 clear_bit(NEW_CUR_UUID, &device->flags);
4010 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4013 rv = _drbd_set_state(device, ns, cs_flags, NULL);
4014 ns = drbd_read_state(device);
4015 spin_unlock_irq(&device->resource->req_lock);
4017 if (rv < SS_SUCCESS) {
4018 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4022 if (os.conn > C_WF_REPORT_PARAMS) {
4023 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4024 peer_state.disk != D_NEGOTIATING ) {
4025 /* we want resync, peer has not yet decided to sync... */
4026 /* Nowadays only used when forcing a node into primary role and
4027 setting its disk to UpToDate with that */
4028 drbd_send_uuids(peer_device);
4029 drbd_send_current_state(peer_device);
4033 clear_bit(DISCARD_MY_DATA, &device->flags);
4035 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4040 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4042 struct drbd_peer_device *peer_device;
4043 struct drbd_device *device;
4044 struct p_rs_uuid *p = pi->data;
4046 peer_device = conn_peer_device(connection, pi->vnr);
4049 device = peer_device->device;
4051 wait_event(device->misc_wait,
4052 device->state.conn == C_WF_SYNC_UUID ||
4053 device->state.conn == C_BEHIND ||
4054 device->state.conn < C_CONNECTED ||
4055 device->state.disk < D_NEGOTIATING);
4057 /* D_ASSERT(device, device->state.conn == C_WF_SYNC_UUID ); */
4059 /* Here the _drbd_uuid_ functions are right, current should
4060 _not_ be rotated into the history */
4061 if (get_ldev_if_state(device, D_NEGOTIATING)) {
4062 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4063 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4065 drbd_print_uuids(device, "updated sync uuid");
4066 drbd_start_resync(device, C_SYNC_TARGET);
4070 drbd_err(device, "Ignoring SyncUUID packet!\n");
4076 * receive_bitmap_plain
4078 * Return 0 when done, 1 when another iteration is needed, and a negative error
4079 * code upon failure.
4082 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4083 unsigned long *p, struct bm_xfer_ctx *c)
4085 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4086 drbd_header_size(peer_device->connection);
4087 unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4088 c->bm_words - c->word_offset);
4089 unsigned int want = num_words * sizeof(*p);
4093 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4098 err = drbd_recv_all(peer_device->connection, p, want);
4102 drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4104 c->word_offset += num_words;
4105 c->bit_offset = c->word_offset * BITS_PER_LONG;
4106 if (c->bit_offset > c->bm_bits)
4107 c->bit_offset = c->bm_bits;
4112 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4114 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4117 static int dcbp_get_start(struct p_compressed_bm *p)
4119 return (p->encoding & 0x80) != 0;
4122 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4124 return (p->encoding >> 4) & 0x7;
4130 * Return 0 when done, 1 when another iteration is needed, and a negative error
4131 * code upon failure.
4134 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4135 struct p_compressed_bm *p,
4136 struct bm_xfer_ctx *c,
4139 struct bitstream bs;
4143 unsigned long s = c->bit_offset;
4145 int toggle = dcbp_get_start(p);
4149 bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4151 bits = bitstream_get_bits(&bs, &look_ahead, 64);
4155 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4156 bits = vli_decode_bits(&rl, look_ahead);
4162 if (e >= c->bm_bits) {
4163 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4166 _drbd_bm_set_bits(peer_device->device, s, e);
4170 drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4171 have, bits, look_ahead,
4172 (unsigned int)(bs.cur.b - p->code),
4173 (unsigned int)bs.buf_len);
4176 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4177 if (likely(bits < 64))
4178 look_ahead >>= bits;
4183 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4186 look_ahead |= tmp << have;
4191 bm_xfer_ctx_bit_to_word_offset(c);
4193 return (s != c->bm_bits);
4199 * Return 0 when done, 1 when another iteration is needed, and a negative error
4200 * code upon failure.
4203 decode_bitmap_c(struct drbd_peer_device *peer_device,
4204 struct p_compressed_bm *p,
4205 struct bm_xfer_ctx *c,
4208 if (dcbp_get_code(p) == RLE_VLI_Bits)
4209 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4211 /* other variants had been implemented for evaluation,
4212 * but have been dropped as this one turned out to be "best"
4213 * during all our tests. */
4215 drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4216 conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4220 void INFO_bm_xfer_stats(struct drbd_device *device,
4221 const char *direction, struct bm_xfer_ctx *c)
4223 /* what would it take to transfer it "plaintext" */
4224 unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4225 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4226 unsigned int plain =
4227 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4228 c->bm_words * sizeof(unsigned long);
4229 unsigned int total = c->bytes[0] + c->bytes[1];
4232 /* total can not be zero. but just in case: */
4236 /* don't report if not compressed */
4240 /* total < plain. check for overflow, still */
4241 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4242 : (1000 * total / plain);
4248 drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4249 "total %u; compression: %u.%u%%\n",
4251 c->bytes[1], c->packets[1],
4252 c->bytes[0], c->packets[0],
4253 total, r/10, r % 10);
4256 /* Since we are processing the bitfield from lower addresses to higher,
4257 it does not matter if the process it in 32 bit chunks or 64 bit
4258 chunks as long as it is little endian. (Understand it as byte stream,
4259 beginning with the lowest byte...) If we would use big endian
4260 we would need to process it from the highest address to the lowest,
4261 in order to be agnostic to the 32 vs 64 bits issue.
4263 returns 0 on failure, 1 if we successfully received it. */
4264 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4266 struct drbd_peer_device *peer_device;
4267 struct drbd_device *device;
4268 struct bm_xfer_ctx c;
4271 peer_device = conn_peer_device(connection, pi->vnr);
4274 device = peer_device->device;
4276 drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4277 /* you are supposed to send additional out-of-sync information
4278 * if you actually set bits during this phase */
4280 c = (struct bm_xfer_ctx) {
4281 .bm_bits = drbd_bm_bits(device),
4282 .bm_words = drbd_bm_words(device),
4286 if (pi->cmd == P_BITMAP)
4287 err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4288 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4289 /* MAYBE: sanity check that we speak proto >= 90,
4290 * and the feature is enabled! */
4291 struct p_compressed_bm *p = pi->data;
4293 if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4294 drbd_err(device, "ReportCBitmap packet too large\n");
4298 if (pi->size <= sizeof(*p)) {
4299 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4303 err = drbd_recv_all(peer_device->connection, p, pi->size);
4306 err = decode_bitmap_c(peer_device, p, &c, pi->size);
4308 drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4313 c.packets[pi->cmd == P_BITMAP]++;
4314 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4321 err = drbd_recv_header(peer_device->connection, pi);
4326 INFO_bm_xfer_stats(device, "receive", &c);
4328 if (device->state.conn == C_WF_BITMAP_T) {
4329 enum drbd_state_rv rv;
4331 err = drbd_send_bitmap(device);
4334 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4335 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4336 D_ASSERT(device, rv == SS_SUCCESS);
4337 } else if (device->state.conn != C_WF_BITMAP_S) {
4338 /* admin may have requested C_DISCONNECTING,
4339 * other threads may have noticed network errors */
4340 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4341 drbd_conn_str(device->state.conn));
4346 drbd_bm_unlock(device);
4347 if (!err && device->state.conn == C_WF_BITMAP_S)
4348 drbd_start_resync(device, C_SYNC_SOURCE);
4352 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4354 drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4357 return ignore_remaining_packet(connection, pi);
4360 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4362 /* Make sure we've acked all the TCP data associated
4363 * with the data requests being unplugged */
4364 drbd_tcp_quickack(connection->data.socket);
4369 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4371 struct drbd_peer_device *peer_device;
4372 struct drbd_device *device;
4373 struct p_block_desc *p = pi->data;
4375 peer_device = conn_peer_device(connection, pi->vnr);
4378 device = peer_device->device;
4380 switch (device->state.conn) {
4381 case C_WF_SYNC_UUID:
4386 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4387 drbd_conn_str(device->state.conn));
4390 drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4398 int (*fn)(struct drbd_connection *, struct packet_info *);
4401 static struct data_cmd drbd_cmd_handler[] = {
4402 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
4403 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
4404 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
4405 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
4406 [P_BITMAP] = { 1, 0, receive_bitmap } ,
4407 [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
4408 [P_UNPLUG_REMOTE] = { 0, 0, receive_UnplugRemote },
4409 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4410 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4411 [P_SYNC_PARAM] = { 1, 0, receive_SyncParam },
4412 [P_SYNC_PARAM89] = { 1, 0, receive_SyncParam },
4413 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
4414 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
4415 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
4416 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
4417 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
4418 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
4419 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4420 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4421 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4422 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
4423 [P_OUT_OF_SYNC] = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
4424 [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
4425 [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
4428 static void drbdd(struct drbd_connection *connection)
4430 struct packet_info pi;
4431 size_t shs; /* sub header size */
4434 while (get_t_state(&connection->receiver) == RUNNING) {
4435 struct data_cmd *cmd;
4437 drbd_thread_current_set_cpu(&connection->receiver);
4438 if (drbd_recv_header(connection, &pi))
4441 cmd = &drbd_cmd_handler[pi.cmd];
4442 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
4443 drbd_err(connection, "Unexpected data packet %s (0x%04x)",
4444 cmdname(pi.cmd), pi.cmd);
4448 shs = cmd->pkt_size;
4449 if (pi.size > shs && !cmd->expect_payload) {
4450 drbd_err(connection, "No payload expected %s l:%d\n",
4451 cmdname(pi.cmd), pi.size);
4456 err = drbd_recv_all_warn(connection, pi.data, shs);
4462 err = cmd->fn(connection, &pi);
4464 drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
4465 cmdname(pi.cmd), err, pi.size);
4472 conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4475 static void conn_disconnect(struct drbd_connection *connection)
4477 struct drbd_peer_device *peer_device;
4481 if (connection->cstate == C_STANDALONE)
4484 /* We are about to start the cleanup after connection loss.
4485 * Make sure drbd_make_request knows about that.
4486 * Usually we should be in some network failure state already,
4487 * but just in case we are not, we fix it up here.
4489 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
4491 /* asender does not clean up anything. it must not interfere, either */
4492 drbd_thread_stop(&connection->asender);
4493 drbd_free_sock(connection);
4496 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
4497 struct drbd_device *device = peer_device->device;
4498 kref_get(&device->kref);
4500 drbd_disconnected(peer_device);
4501 kref_put(&device->kref, drbd_destroy_device);
4506 if (!list_empty(&connection->current_epoch->list))
4507 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
4508 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
4509 atomic_set(&connection->current_epoch->epoch_size, 0);
4510 connection->send.seen_any_write_yet = false;
4512 drbd_info(connection, "Connection closed\n");
4514 if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
4515 conn_try_outdate_peer_async(connection);
4517 spin_lock_irq(&connection->resource->req_lock);
4518 oc = connection->cstate;
4519 if (oc >= C_UNCONNECTED)
4520 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
4522 spin_unlock_irq(&connection->resource->req_lock);
4524 if (oc == C_DISCONNECTING)
4525 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
4528 static int drbd_disconnected(struct drbd_peer_device *peer_device)
4530 struct drbd_device *device = peer_device->device;
4533 /* wait for current activity to cease. */
4534 spin_lock_irq(&device->resource->req_lock);
4535 _drbd_wait_ee_list_empty(device, &device->active_ee);
4536 _drbd_wait_ee_list_empty(device, &device->sync_ee);
4537 _drbd_wait_ee_list_empty(device, &device->read_ee);
4538 spin_unlock_irq(&device->resource->req_lock);
4540 /* We do not have data structures that would allow us to
4541 * get the rs_pending_cnt down to 0 again.
4542 * * On C_SYNC_TARGET we do not have any data structures describing
4543 * the pending RSDataRequest's we have sent.
4544 * * On C_SYNC_SOURCE there is no data structure that tracks
4545 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
4546 * And no, it is not the sum of the reference counts in the
4547 * resync_LRU. The resync_LRU tracks the whole operation including
4548 * the disk-IO, while the rs_pending_cnt only tracks the blocks
4550 drbd_rs_cancel_all(device);
4551 device->rs_total = 0;
4552 device->rs_failed = 0;
4553 atomic_set(&device->rs_pending_cnt, 0);
4554 wake_up(&device->misc_wait);
4556 del_timer_sync(&device->resync_timer);
4557 resync_timer_fn((unsigned long)device);
4559 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
4560 * w_make_resync_request etc. which may still be on the worker queue
4561 * to be "canceled" */
4562 drbd_flush_workqueue(&peer_device->connection->sender_work);
4564 drbd_finish_peer_reqs(device);
4566 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
4567 might have issued a work again. The one before drbd_finish_peer_reqs() is
4568 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
4569 drbd_flush_workqueue(&peer_device->connection->sender_work);
4571 /* need to do it again, drbd_finish_peer_reqs() may have populated it
4572 * again via drbd_try_clear_on_disk_bm(). */
4573 drbd_rs_cancel_all(device);
4575 kfree(device->p_uuid);
4576 device->p_uuid = NULL;
4578 if (!drbd_suspended(device))
4579 tl_clear(peer_device->connection);
4581 drbd_md_sync(device);
4583 /* serialize with bitmap writeout triggered by the state change,
4585 wait_event(device->misc_wait, !test_bit(BITMAP_IO, &device->flags));
4587 /* tcp_close and release of sendpage pages can be deferred. I don't
4588 * want to use SO_LINGER, because apparently it can be deferred for
4589 * more than 20 seconds (longest time I checked).
4591 * Actually we don't care for exactly when the network stack does its
4592 * put_page(), but release our reference on these pages right here.
4594 i = drbd_free_peer_reqs(device, &device->net_ee);
4596 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
4597 i = atomic_read(&device->pp_in_use_by_net);
4599 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
4600 i = atomic_read(&device->pp_in_use);
4602 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
4604 D_ASSERT(device, list_empty(&device->read_ee));
4605 D_ASSERT(device, list_empty(&device->active_ee));
4606 D_ASSERT(device, list_empty(&device->sync_ee));
4607 D_ASSERT(device, list_empty(&device->done_ee));
4613 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
4614 * we can agree on is stored in agreed_pro_version.
4616 * feature flags and the reserved array should be enough room for future
4617 * enhancements of the handshake protocol, and possible plugins...
4619 * for now, they are expected to be zero, but ignored.
4621 static int drbd_send_features(struct drbd_connection *connection)
4623 struct drbd_socket *sock;
4624 struct p_connection_features *p;
4626 sock = &connection->data;
4627 p = conn_prepare_command(connection, sock);
4630 memset(p, 0, sizeof(*p));
4631 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
4632 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
4633 return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
4638 * 1 yes, we have a valid connection
4639 * 0 oops, did not work out, please try again
4640 * -1 peer talks different language,
4641 * no point in trying again, please go standalone.
4643 static int drbd_do_features(struct drbd_connection *connection)
4645 /* ASSERT current == connection->receiver ... */
4646 struct p_connection_features *p;
4647 const int expect = sizeof(struct p_connection_features);
4648 struct packet_info pi;
4651 err = drbd_send_features(connection);
4655 err = drbd_recv_header(connection, &pi);
4659 if (pi.cmd != P_CONNECTION_FEATURES) {
4660 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
4661 cmdname(pi.cmd), pi.cmd);
4665 if (pi.size != expect) {
4666 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
4672 err = drbd_recv_all_warn(connection, p, expect);
4676 p->protocol_min = be32_to_cpu(p->protocol_min);
4677 p->protocol_max = be32_to_cpu(p->protocol_max);
4678 if (p->protocol_max == 0)
4679 p->protocol_max = p->protocol_min;
4681 if (PRO_VERSION_MAX < p->protocol_min ||
4682 PRO_VERSION_MIN > p->protocol_max)
4685 connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
4687 drbd_info(connection, "Handshake successful: "
4688 "Agreed network protocol version %d\n", connection->agreed_pro_version);
4693 drbd_err(connection, "incompatible DRBD dialects: "
4694 "I support %d-%d, peer supports %d-%d\n",
4695 PRO_VERSION_MIN, PRO_VERSION_MAX,
4696 p->protocol_min, p->protocol_max);
4700 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4701 static int drbd_do_auth(struct drbd_connection *connection)
4703 drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4704 drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4708 #define CHALLENGE_LEN 64
4712 0 - failed, try again (network error),
4713 -1 - auth failed, don't try again.
4716 static int drbd_do_auth(struct drbd_connection *connection)
4718 struct drbd_socket *sock;
4719 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
4720 struct scatterlist sg;
4721 char *response = NULL;
4722 char *right_response = NULL;
4723 char *peers_ch = NULL;
4724 unsigned int key_len;
4725 char secret[SHARED_SECRET_MAX]; /* 64 byte */
4726 unsigned int resp_size;
4727 struct hash_desc desc;
4728 struct packet_info pi;
4729 struct net_conf *nc;
4732 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
4735 nc = rcu_dereference(connection->net_conf);
4736 key_len = strlen(nc->shared_secret);
4737 memcpy(secret, nc->shared_secret, key_len);
4740 desc.tfm = connection->cram_hmac_tfm;
4743 rv = crypto_hash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
4745 drbd_err(connection, "crypto_hash_setkey() failed with %d\n", rv);
4750 get_random_bytes(my_challenge, CHALLENGE_LEN);
4752 sock = &connection->data;
4753 if (!conn_prepare_command(connection, sock)) {
4757 rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
4758 my_challenge, CHALLENGE_LEN);
4762 err = drbd_recv_header(connection, &pi);
4768 if (pi.cmd != P_AUTH_CHALLENGE) {
4769 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4770 cmdname(pi.cmd), pi.cmd);
4775 if (pi.size > CHALLENGE_LEN * 2) {
4776 drbd_err(connection, "expected AuthChallenge payload too big.\n");
4781 peers_ch = kmalloc(pi.size, GFP_NOIO);
4782 if (peers_ch == NULL) {
4783 drbd_err(connection, "kmalloc of peers_ch failed\n");
4788 err = drbd_recv_all_warn(connection, peers_ch, pi.size);
4794 resp_size = crypto_hash_digestsize(connection->cram_hmac_tfm);
4795 response = kmalloc(resp_size, GFP_NOIO);
4796 if (response == NULL) {
4797 drbd_err(connection, "kmalloc of response failed\n");
4802 sg_init_table(&sg, 1);
4803 sg_set_buf(&sg, peers_ch, pi.size);
4805 rv = crypto_hash_digest(&desc, &sg, sg.length, response);
4807 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
4812 if (!conn_prepare_command(connection, sock)) {
4816 rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
4817 response, resp_size);
4821 err = drbd_recv_header(connection, &pi);
4827 if (pi.cmd != P_AUTH_RESPONSE) {
4828 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
4829 cmdname(pi.cmd), pi.cmd);
4834 if (pi.size != resp_size) {
4835 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
4840 err = drbd_recv_all_warn(connection, response , resp_size);
4846 right_response = kmalloc(resp_size, GFP_NOIO);
4847 if (right_response == NULL) {
4848 drbd_err(connection, "kmalloc of right_response failed\n");
4853 sg_set_buf(&sg, my_challenge, CHALLENGE_LEN);
4855 rv = crypto_hash_digest(&desc, &sg, sg.length, right_response);
4857 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
4862 rv = !memcmp(response, right_response, resp_size);
4865 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
4873 kfree(right_response);
4879 int drbd_receiver(struct drbd_thread *thi)
4881 struct drbd_connection *connection = thi->connection;
4884 drbd_info(connection, "receiver (re)started\n");
4887 h = conn_connect(connection);
4889 conn_disconnect(connection);
4890 schedule_timeout_interruptible(HZ);
4893 drbd_warn(connection, "Discarding network configuration.\n");
4894 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
4901 conn_disconnect(connection);
4903 drbd_info(connection, "receiver terminated\n");
4907 /* ********* acknowledge sender ******** */
4909 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
4911 struct p_req_state_reply *p = pi->data;
4912 int retcode = be32_to_cpu(p->retcode);
4914 if (retcode >= SS_SUCCESS) {
4915 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
4917 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
4918 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
4919 drbd_set_st_err_str(retcode), retcode);
4921 wake_up(&connection->ping_wait);
4926 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
4928 struct drbd_peer_device *peer_device;
4929 struct drbd_device *device;
4930 struct p_req_state_reply *p = pi->data;
4931 int retcode = be32_to_cpu(p->retcode);
4933 peer_device = conn_peer_device(connection, pi->vnr);
4936 device = peer_device->device;
4938 if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
4939 D_ASSERT(device, connection->agreed_pro_version < 100);
4940 return got_conn_RqSReply(connection, pi);
4943 if (retcode >= SS_SUCCESS) {
4944 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
4946 set_bit(CL_ST_CHG_FAIL, &device->flags);
4947 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
4948 drbd_set_st_err_str(retcode), retcode);
4950 wake_up(&device->state_wait);
4955 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
4957 return drbd_send_ping_ack(connection);
4961 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
4963 /* restore idle timeout */
4964 connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
4965 if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
4966 wake_up(&connection->ping_wait);
4971 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
4973 struct drbd_peer_device *peer_device;
4974 struct drbd_device *device;
4975 struct p_block_ack *p = pi->data;
4976 sector_t sector = be64_to_cpu(p->sector);
4977 int blksize = be32_to_cpu(p->blksize);
4979 peer_device = conn_peer_device(connection, pi->vnr);
4982 device = peer_device->device;
4984 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
4986 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
4988 if (get_ldev(device)) {
4989 drbd_rs_complete_io(device, sector);
4990 drbd_set_in_sync(device, sector, blksize);
4991 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4992 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
4995 dec_rs_pending(device);
4996 atomic_add(blksize >> 9, &device->rs_sect_in);
5002 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5003 struct rb_root *root, const char *func,
5004 enum drbd_req_event what, bool missing_ok)
5006 struct drbd_request *req;
5007 struct bio_and_error m;
5009 spin_lock_irq(&device->resource->req_lock);
5010 req = find_request(device, root, id, sector, missing_ok, func);
5011 if (unlikely(!req)) {
5012 spin_unlock_irq(&device->resource->req_lock);
5015 __req_mod(req, what, &m);
5016 spin_unlock_irq(&device->resource->req_lock);
5019 complete_master_bio(device, &m);
5023 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5025 struct drbd_peer_device *peer_device;
5026 struct drbd_device *device;
5027 struct p_block_ack *p = pi->data;
5028 sector_t sector = be64_to_cpu(p->sector);
5029 int blksize = be32_to_cpu(p->blksize);
5030 enum drbd_req_event what;
5032 peer_device = conn_peer_device(connection, pi->vnr);
5035 device = peer_device->device;
5037 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5039 if (p->block_id == ID_SYNCER) {
5040 drbd_set_in_sync(device, sector, blksize);
5041 dec_rs_pending(device);
5045 case P_RS_WRITE_ACK:
5046 what = WRITE_ACKED_BY_PEER_AND_SIS;
5049 what = WRITE_ACKED_BY_PEER;
5052 what = RECV_ACKED_BY_PEER;
5055 what = CONFLICT_RESOLVED;
5058 what = POSTPONE_WRITE;
5064 return validate_req_change_req_state(device, p->block_id, sector,
5065 &device->write_requests, __func__,
5069 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5071 struct drbd_peer_device *peer_device;
5072 struct drbd_device *device;
5073 struct p_block_ack *p = pi->data;
5074 sector_t sector = be64_to_cpu(p->sector);
5075 int size = be32_to_cpu(p->blksize);
5078 peer_device = conn_peer_device(connection, pi->vnr);
5081 device = peer_device->device;
5083 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5085 if (p->block_id == ID_SYNCER) {
5086 dec_rs_pending(device);
5087 drbd_rs_failed_io(device, sector, size);
5091 err = validate_req_change_req_state(device, p->block_id, sector,
5092 &device->write_requests, __func__,
5095 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5096 The master bio might already be completed, therefore the
5097 request is no longer in the collision hash. */
5098 /* In Protocol B we might already have got a P_RECV_ACK
5099 but then get a P_NEG_ACK afterwards. */
5100 drbd_set_out_of_sync(device, sector, size);
5105 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5107 struct drbd_peer_device *peer_device;
5108 struct drbd_device *device;
5109 struct p_block_ack *p = pi->data;
5110 sector_t sector = be64_to_cpu(p->sector);
5112 peer_device = conn_peer_device(connection, pi->vnr);
5115 device = peer_device->device;
5117 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5119 drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5120 (unsigned long long)sector, be32_to_cpu(p->blksize));
5122 return validate_req_change_req_state(device, p->block_id, sector,
5123 &device->read_requests, __func__,
5127 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5129 struct drbd_peer_device *peer_device;
5130 struct drbd_device *device;
5133 struct p_block_ack *p = pi->data;
5135 peer_device = conn_peer_device(connection, pi->vnr);
5138 device = peer_device->device;
5140 sector = be64_to_cpu(p->sector);
5141 size = be32_to_cpu(p->blksize);
5143 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5145 dec_rs_pending(device);
5147 if (get_ldev_if_state(device, D_FAILED)) {
5148 drbd_rs_complete_io(device, sector);
5150 case P_NEG_RS_DREPLY:
5151 drbd_rs_failed_io(device, sector, size);
5163 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5165 struct p_barrier_ack *p = pi->data;
5166 struct drbd_peer_device *peer_device;
5169 tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5172 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5173 struct drbd_device *device = peer_device->device;
5175 if (device->state.conn == C_AHEAD &&
5176 atomic_read(&device->ap_in_flight) == 0 &&
5177 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5178 device->start_resync_timer.expires = jiffies + HZ;
5179 add_timer(&device->start_resync_timer);
5187 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5189 struct drbd_peer_device *peer_device;
5190 struct drbd_device *device;
5191 struct p_block_ack *p = pi->data;
5192 struct drbd_device_work *dw;
5196 peer_device = conn_peer_device(connection, pi->vnr);
5199 device = peer_device->device;
5201 sector = be64_to_cpu(p->sector);
5202 size = be32_to_cpu(p->blksize);
5204 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5206 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5207 drbd_ov_out_of_sync_found(device, sector, size);
5209 ov_out_of_sync_print(device);
5211 if (!get_ldev(device))
5214 drbd_rs_complete_io(device, sector);
5215 dec_rs_pending(device);
5219 /* let's advance progress step marks only for every other megabyte */
5220 if ((device->ov_left & 0x200) == 0x200)
5221 drbd_advance_rs_marks(device, device->ov_left);
5223 if (device->ov_left == 0) {
5224 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5226 dw->w.cb = w_ov_finished;
5227 dw->device = device;
5228 drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5230 drbd_err(device, "kmalloc(dw) failed.");
5231 ov_out_of_sync_print(device);
5232 drbd_resync_finished(device);
5239 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5244 static int connection_finish_peer_reqs(struct drbd_connection *connection)
5246 struct drbd_peer_device *peer_device;
5247 int vnr, not_empty = 0;
5250 clear_bit(SIGNAL_ASENDER, &connection->flags);
5251 flush_signals(current);
5254 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5255 struct drbd_device *device = peer_device->device;
5256 kref_get(&device->kref);
5258 if (drbd_finish_peer_reqs(device)) {
5259 kref_put(&device->kref, drbd_destroy_device);
5262 kref_put(&device->kref, drbd_destroy_device);
5265 set_bit(SIGNAL_ASENDER, &connection->flags);
5267 spin_lock_irq(&connection->resource->req_lock);
5268 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5269 struct drbd_device *device = peer_device->device;
5270 not_empty = !list_empty(&device->done_ee);
5274 spin_unlock_irq(&connection->resource->req_lock);
5276 } while (not_empty);
5281 struct asender_cmd {
5283 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5286 static struct asender_cmd asender_tbl[] = {
5287 [P_PING] = { 0, got_Ping },
5288 [P_PING_ACK] = { 0, got_PingAck },
5289 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5290 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5291 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5292 [P_SUPERSEDED] = { sizeof(struct p_block_ack), got_BlockAck },
5293 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
5294 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
5295 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply },
5296 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
5297 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
5298 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5299 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
5300 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
5301 [P_RS_CANCEL] = { sizeof(struct p_block_ack), got_NegRSDReply },
5302 [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5303 [P_RETRY_WRITE] = { sizeof(struct p_block_ack), got_BlockAck },
5306 int drbd_asender(struct drbd_thread *thi)
5308 struct drbd_connection *connection = thi->connection;
5309 struct asender_cmd *cmd = NULL;
5310 struct packet_info pi;
5312 void *buf = connection->meta.rbuf;
5314 unsigned int header_size = drbd_header_size(connection);
5315 int expect = header_size;
5316 bool ping_timeout_active = false;
5317 struct net_conf *nc;
5318 int ping_timeo, tcp_cork, ping_int;
5319 struct sched_param param = { .sched_priority = 2 };
5321 rv = sched_setscheduler(current, SCHED_RR, ¶m);
5323 drbd_err(connection, "drbd_asender: ERROR set priority, ret=%d\n", rv);
5325 while (get_t_state(thi) == RUNNING) {
5326 drbd_thread_current_set_cpu(thi);
5329 nc = rcu_dereference(connection->net_conf);
5330 ping_timeo = nc->ping_timeo;
5331 tcp_cork = nc->tcp_cork;
5332 ping_int = nc->ping_int;
5335 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5336 if (drbd_send_ping(connection)) {
5337 drbd_err(connection, "drbd_send_ping has failed\n");
5340 connection->meta.socket->sk->sk_rcvtimeo = ping_timeo * HZ / 10;
5341 ping_timeout_active = true;
5344 /* TODO: conditionally cork; it may hurt latency if we cork without
5347 drbd_tcp_cork(connection->meta.socket);
5348 if (connection_finish_peer_reqs(connection)) {
5349 drbd_err(connection, "connection_finish_peer_reqs() failed\n");
5352 /* but unconditionally uncork unless disabled */
5354 drbd_tcp_uncork(connection->meta.socket);
5356 /* short circuit, recv_msg would return EINTR anyways. */
5357 if (signal_pending(current))
5360 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5361 clear_bit(SIGNAL_ASENDER, &connection->flags);
5363 flush_signals(current);
5366 * -EINTR (on meta) we got a signal
5367 * -EAGAIN (on meta) rcvtimeo expired
5368 * -ECONNRESET other side closed the connection
5369 * -ERESTARTSYS (on data) we got a signal
5370 * rv < 0 other than above: unexpected error!
5371 * rv == expected: full header or command
5372 * rv < expected: "woken" by signal during receive
5373 * rv == 0 : "connection shut down by peer"
5375 if (likely(rv > 0)) {
5378 } else if (rv == 0) {
5379 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
5382 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
5385 t = wait_event_timeout(connection->ping_wait,
5386 connection->cstate < C_WF_REPORT_PARAMS,
5391 drbd_err(connection, "meta connection shut down by peer.\n");
5393 } else if (rv == -EAGAIN) {
5394 /* If the data socket received something meanwhile,
5395 * that is good enough: peer is still alive. */
5396 if (time_after(connection->last_received,
5397 jiffies - connection->meta.socket->sk->sk_rcvtimeo))
5399 if (ping_timeout_active) {
5400 drbd_err(connection, "PingAck did not arrive in time.\n");
5403 set_bit(SEND_PING, &connection->flags);
5405 } else if (rv == -EINTR) {
5408 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
5412 if (received == expect && cmd == NULL) {
5413 if (decode_header(connection, connection->meta.rbuf, &pi))
5415 cmd = &asender_tbl[pi.cmd];
5416 if (pi.cmd >= ARRAY_SIZE(asender_tbl) || !cmd->fn) {
5417 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
5418 cmdname(pi.cmd), pi.cmd);
5421 expect = header_size + cmd->pkt_size;
5422 if (pi.size != expect - header_size) {
5423 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
5428 if (received == expect) {
5431 err = cmd->fn(connection, &pi);
5433 drbd_err(connection, "%pf failed\n", cmd->fn);
5437 connection->last_received = jiffies;
5439 if (cmd == &asender_tbl[P_PING_ACK]) {
5440 /* restore idle timeout */
5441 connection->meta.socket->sk->sk_rcvtimeo = ping_int * HZ;
5442 ping_timeout_active = false;
5445 buf = connection->meta.rbuf;
5447 expect = header_size;
5454 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5455 conn_md_sync(connection);
5459 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5461 clear_bit(SIGNAL_ASENDER, &connection->flags);
5463 drbd_info(connection, "asender terminated\n");