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"
51 #define PRO_FEATURES (FF_TRIM)
66 static int drbd_do_features(struct drbd_connection *connection);
67 static int drbd_do_auth(struct drbd_connection *connection);
68 static int drbd_disconnected(struct drbd_peer_device *);
69 static void conn_wait_active_ee_empty(struct drbd_connection *connection);
70 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
71 static int e_end_block(struct drbd_work *, int);
74 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
77 * some helper functions to deal with single linked page lists,
78 * page->private being our "next" pointer.
81 /* If at least n pages are linked at head, get n pages off.
82 * Otherwise, don't modify head, and return NULL.
83 * Locking is the responsibility of the caller.
85 static struct page *page_chain_del(struct page **head, int n)
99 tmp = page_chain_next(page);
101 break; /* found sufficient pages */
103 /* insufficient pages, don't use any of them. */
108 /* add end of list marker for the returned list */
109 set_page_private(page, 0);
110 /* actual return value, and adjustment of head */
116 /* may be used outside of locks to find the tail of a (usually short)
117 * "private" page chain, before adding it back to a global chain head
118 * with page_chain_add() under a spinlock. */
119 static struct page *page_chain_tail(struct page *page, int *len)
123 while ((tmp = page_chain_next(page)))
130 static int page_chain_free(struct page *page)
134 page_chain_for_each_safe(page, tmp) {
141 static void page_chain_add(struct page **head,
142 struct page *chain_first, struct page *chain_last)
146 tmp = page_chain_tail(chain_first, NULL);
147 BUG_ON(tmp != chain_last);
150 /* add chain to head */
151 set_page_private(chain_last, (unsigned long)*head);
155 static struct page *__drbd_alloc_pages(struct drbd_device *device,
158 struct page *page = NULL;
159 struct page *tmp = NULL;
162 /* Yes, testing drbd_pp_vacant outside the lock is racy.
163 * So what. It saves a spin_lock. */
164 if (drbd_pp_vacant >= number) {
165 spin_lock(&drbd_pp_lock);
166 page = page_chain_del(&drbd_pp_pool, number);
168 drbd_pp_vacant -= number;
169 spin_unlock(&drbd_pp_lock);
174 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
175 * "criss-cross" setup, that might cause write-out on some other DRBD,
176 * which in turn might block on the other node at this very place. */
177 for (i = 0; i < number; i++) {
178 tmp = alloc_page(GFP_TRY);
181 set_page_private(tmp, (unsigned long)page);
188 /* Not enough pages immediately available this time.
189 * No need to jump around here, drbd_alloc_pages will retry this
190 * function "soon". */
192 tmp = page_chain_tail(page, NULL);
193 spin_lock(&drbd_pp_lock);
194 page_chain_add(&drbd_pp_pool, page, tmp);
196 spin_unlock(&drbd_pp_lock);
201 static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
202 struct list_head *to_be_freed)
204 struct drbd_peer_request *peer_req, *tmp;
206 /* The EEs are always appended to the end of the list. Since
207 they are sent in order over the wire, they have to finish
208 in order. As soon as we see the first not finished we can
209 stop to examine the list... */
211 list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
212 if (drbd_peer_req_has_active_page(peer_req))
214 list_move(&peer_req->w.list, to_be_freed);
218 static void drbd_reclaim_net_peer_reqs(struct drbd_device *device)
220 LIST_HEAD(reclaimed);
221 struct drbd_peer_request *peer_req, *t;
223 spin_lock_irq(&device->resource->req_lock);
224 reclaim_finished_net_peer_reqs(device, &reclaimed);
225 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);
230 static void conn_reclaim_net_peer_reqs(struct drbd_connection *connection)
232 struct drbd_peer_device *peer_device;
236 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
237 struct drbd_device *device = peer_device->device;
238 if (!atomic_read(&device->pp_in_use_by_net))
241 kref_get(&device->kref);
243 drbd_reclaim_net_peer_reqs(device);
244 kref_put(&device->kref, drbd_destroy_device);
251 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
252 * @device: DRBD device.
253 * @number: number of pages requested
254 * @retry: whether to retry, if not enough pages are available right now
256 * Tries to allocate number pages, first from our own page pool, then from
258 * Possibly retry until DRBD frees sufficient pages somewhere else.
260 * If this allocation would exceed the max_buffers setting, we throttle
261 * allocation (schedule_timeout) to give the system some room to breathe.
263 * We do not use max-buffers as hard limit, because it could lead to
264 * congestion and further to a distributed deadlock during online-verify or
265 * (checksum based) resync, if the max-buffers, socket buffer sizes and
266 * resync-rate settings are mis-configured.
268 * Returns a page chain linked via page->private.
270 struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
273 struct drbd_device *device = peer_device->device;
274 struct page *page = NULL;
280 nc = rcu_dereference(peer_device->connection->net_conf);
281 mxb = nc ? nc->max_buffers : 1000000;
284 if (atomic_read(&device->pp_in_use) < mxb)
285 page = __drbd_alloc_pages(device, number);
287 /* Try to keep the fast path fast, but occasionally we need
288 * to reclaim the pages we lended to the network stack. */
289 if (page && atomic_read(&device->pp_in_use_by_net) > 512)
290 drbd_reclaim_net_peer_reqs(device);
292 while (page == NULL) {
293 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
295 drbd_reclaim_net_peer_reqs(device);
297 if (atomic_read(&device->pp_in_use) < mxb) {
298 page = __drbd_alloc_pages(device, number);
306 if (signal_pending(current)) {
307 drbd_warn(device, "drbd_alloc_pages interrupted!\n");
311 if (schedule_timeout(HZ/10) == 0)
314 finish_wait(&drbd_pp_wait, &wait);
317 atomic_add(number, &device->pp_in_use);
321 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
322 * Is also used from inside an other spin_lock_irq(&resource->req_lock);
323 * Either links the page chain back to the global pool,
324 * or returns all pages to the system. */
325 static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
327 atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
333 if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * minor_count)
334 i = page_chain_free(page);
337 tmp = page_chain_tail(page, &i);
338 spin_lock(&drbd_pp_lock);
339 page_chain_add(&drbd_pp_pool, page, tmp);
341 spin_unlock(&drbd_pp_lock);
343 i = atomic_sub_return(i, a);
345 drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
346 is_net ? "pp_in_use_by_net" : "pp_in_use", i);
347 wake_up(&drbd_pp_wait);
351 You need to hold the req_lock:
352 _drbd_wait_ee_list_empty()
354 You must not have the req_lock:
356 drbd_alloc_peer_req()
357 drbd_free_peer_reqs()
359 drbd_finish_peer_reqs()
361 drbd_wait_ee_list_empty()
364 struct drbd_peer_request *
365 drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
366 unsigned int data_size, bool has_payload, gfp_t gfp_mask) __must_hold(local)
368 struct drbd_device *device = peer_device->device;
369 struct drbd_peer_request *peer_req;
370 struct page *page = NULL;
371 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
373 if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
376 peer_req = mempool_alloc(drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
378 if (!(gfp_mask & __GFP_NOWARN))
379 drbd_err(device, "%s: allocation failed\n", __func__);
383 if (has_payload && data_size) {
384 page = drbd_alloc_pages(peer_device, nr_pages,
385 gfpflags_allow_blocking(gfp_mask));
390 memset(peer_req, 0, sizeof(*peer_req));
391 INIT_LIST_HEAD(&peer_req->w.list);
392 drbd_clear_interval(&peer_req->i);
393 peer_req->i.size = data_size;
394 peer_req->i.sector = sector;
395 peer_req->submit_jif = jiffies;
396 peer_req->peer_device = peer_device;
397 peer_req->pages = page;
399 * The block_id is opaque to the receiver. It is not endianness
400 * converted, and sent back to the sender unchanged.
402 peer_req->block_id = id;
407 mempool_free(peer_req, drbd_ee_mempool);
411 void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
415 if (peer_req->flags & EE_HAS_DIGEST)
416 kfree(peer_req->digest);
417 drbd_free_pages(device, peer_req->pages, is_net);
418 D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
419 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
420 if (!expect(!(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
421 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
422 drbd_al_complete_io(device, &peer_req->i);
424 mempool_free(peer_req, drbd_ee_mempool);
427 int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
429 LIST_HEAD(work_list);
430 struct drbd_peer_request *peer_req, *t;
432 int is_net = list == &device->net_ee;
434 spin_lock_irq(&device->resource->req_lock);
435 list_splice_init(list, &work_list);
436 spin_unlock_irq(&device->resource->req_lock);
438 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
439 __drbd_free_peer_req(device, peer_req, is_net);
446 * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
448 static int drbd_finish_peer_reqs(struct drbd_device *device)
450 LIST_HEAD(work_list);
451 LIST_HEAD(reclaimed);
452 struct drbd_peer_request *peer_req, *t;
455 spin_lock_irq(&device->resource->req_lock);
456 reclaim_finished_net_peer_reqs(device, &reclaimed);
457 list_splice_init(&device->done_ee, &work_list);
458 spin_unlock_irq(&device->resource->req_lock);
460 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
461 drbd_free_net_peer_req(device, peer_req);
463 /* possible callbacks here:
464 * e_end_block, and e_end_resync_block, e_send_superseded.
465 * all ignore the last argument.
467 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
470 /* list_del not necessary, next/prev members not touched */
471 err2 = peer_req->w.cb(&peer_req->w, !!err);
474 drbd_free_peer_req(device, peer_req);
476 wake_up(&device->ee_wait);
481 static void _drbd_wait_ee_list_empty(struct drbd_device *device,
482 struct list_head *head)
486 /* avoids spin_lock/unlock
487 * and calling prepare_to_wait in the fast path */
488 while (!list_empty(head)) {
489 prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
490 spin_unlock_irq(&device->resource->req_lock);
492 finish_wait(&device->ee_wait, &wait);
493 spin_lock_irq(&device->resource->req_lock);
497 static void drbd_wait_ee_list_empty(struct drbd_device *device,
498 struct list_head *head)
500 spin_lock_irq(&device->resource->req_lock);
501 _drbd_wait_ee_list_empty(device, head);
502 spin_unlock_irq(&device->resource->req_lock);
505 static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
511 struct msghdr msg = {
512 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
514 return kernel_recvmsg(sock, &msg, &iov, 1, size, msg.msg_flags);
517 static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
521 rv = drbd_recv_short(connection->data.socket, buf, size, 0);
524 if (rv == -ECONNRESET)
525 drbd_info(connection, "sock was reset by peer\n");
526 else if (rv != -ERESTARTSYS)
527 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
528 } else if (rv == 0) {
529 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
532 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
535 t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);
540 drbd_info(connection, "sock was shut down by peer\n");
544 conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
550 static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
554 err = drbd_recv(connection, buf, size);
563 static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
567 err = drbd_recv_all(connection, buf, size);
568 if (err && !signal_pending(current))
569 drbd_warn(connection, "short read (expected size %d)\n", (int)size);
574 * On individual connections, the socket buffer size must be set prior to the
575 * listen(2) or connect(2) calls in order to have it take effect.
576 * This is our wrapper to do so.
578 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
581 /* open coded SO_SNDBUF, SO_RCVBUF */
583 sock->sk->sk_sndbuf = snd;
584 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
587 sock->sk->sk_rcvbuf = rcv;
588 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
592 static struct socket *drbd_try_connect(struct drbd_connection *connection)
596 struct sockaddr_in6 src_in6;
597 struct sockaddr_in6 peer_in6;
599 int err, peer_addr_len, my_addr_len;
600 int sndbuf_size, rcvbuf_size, connect_int;
601 int disconnect_on_error = 1;
604 nc = rcu_dereference(connection->net_conf);
609 sndbuf_size = nc->sndbuf_size;
610 rcvbuf_size = nc->rcvbuf_size;
611 connect_int = nc->connect_int;
614 my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
615 memcpy(&src_in6, &connection->my_addr, my_addr_len);
617 if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
618 src_in6.sin6_port = 0;
620 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
622 peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
623 memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);
625 what = "sock_create_kern";
626 err = sock_create_kern(&init_net, ((struct sockaddr *)&src_in6)->sa_family,
627 SOCK_STREAM, IPPROTO_TCP, &sock);
633 sock->sk->sk_rcvtimeo =
634 sock->sk->sk_sndtimeo = connect_int * HZ;
635 drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);
637 /* explicitly bind to the configured IP as source IP
638 * for the outgoing connections.
639 * This is needed for multihomed hosts and to be
640 * able to use lo: interfaces for drbd.
641 * Make sure to use 0 as port number, so linux selects
642 * a free one dynamically.
644 what = "bind before connect";
645 err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
649 /* connect may fail, peer not yet available.
650 * stay C_WF_CONNECTION, don't go Disconnecting! */
651 disconnect_on_error = 0;
653 err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);
662 /* timeout, busy, signal pending */
663 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
664 case EINTR: case ERESTARTSYS:
665 /* peer not (yet) available, network problem */
666 case ECONNREFUSED: case ENETUNREACH:
667 case EHOSTDOWN: case EHOSTUNREACH:
668 disconnect_on_error = 0;
671 drbd_err(connection, "%s failed, err = %d\n", what, err);
673 if (disconnect_on_error)
674 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
680 struct accept_wait_data {
681 struct drbd_connection *connection;
682 struct socket *s_listen;
683 struct completion door_bell;
684 void (*original_sk_state_change)(struct sock *sk);
688 static void drbd_incoming_connection(struct sock *sk)
690 struct accept_wait_data *ad = sk->sk_user_data;
691 void (*state_change)(struct sock *sk);
693 state_change = ad->original_sk_state_change;
694 if (sk->sk_state == TCP_ESTABLISHED)
695 complete(&ad->door_bell);
699 static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
701 int err, sndbuf_size, rcvbuf_size, my_addr_len;
702 struct sockaddr_in6 my_addr;
703 struct socket *s_listen;
708 nc = rcu_dereference(connection->net_conf);
713 sndbuf_size = nc->sndbuf_size;
714 rcvbuf_size = nc->rcvbuf_size;
717 my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
718 memcpy(&my_addr, &connection->my_addr, my_addr_len);
720 what = "sock_create_kern";
721 err = sock_create_kern(&init_net, ((struct sockaddr *)&my_addr)->sa_family,
722 SOCK_STREAM, IPPROTO_TCP, &s_listen);
728 s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
729 drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);
731 what = "bind before listen";
732 err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
736 ad->s_listen = s_listen;
737 write_lock_bh(&s_listen->sk->sk_callback_lock);
738 ad->original_sk_state_change = s_listen->sk->sk_state_change;
739 s_listen->sk->sk_state_change = drbd_incoming_connection;
740 s_listen->sk->sk_user_data = ad;
741 write_unlock_bh(&s_listen->sk->sk_callback_lock);
744 err = s_listen->ops->listen(s_listen, 5);
751 sock_release(s_listen);
753 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
754 drbd_err(connection, "%s failed, err = %d\n", what, err);
755 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
762 static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
764 write_lock_bh(&sk->sk_callback_lock);
765 sk->sk_state_change = ad->original_sk_state_change;
766 sk->sk_user_data = NULL;
767 write_unlock_bh(&sk->sk_callback_lock);
770 static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
772 int timeo, connect_int, err = 0;
773 struct socket *s_estab = NULL;
777 nc = rcu_dereference(connection->net_conf);
782 connect_int = nc->connect_int;
785 timeo = connect_int * HZ;
786 /* 28.5% random jitter */
787 timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;
789 err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
793 err = kernel_accept(ad->s_listen, &s_estab, 0);
795 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
796 drbd_err(connection, "accept failed, err = %d\n", err);
797 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
802 unregister_state_change(s_estab->sk, ad);
807 static int decode_header(struct drbd_connection *, void *, struct packet_info *);
809 static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
810 enum drbd_packet cmd)
812 if (!conn_prepare_command(connection, sock))
814 return conn_send_command(connection, sock, cmd, 0, NULL, 0);
817 static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
819 unsigned int header_size = drbd_header_size(connection);
820 struct packet_info pi;
825 nc = rcu_dereference(connection->net_conf);
830 sock->sk->sk_rcvtimeo = nc->ping_timeo * 4 * HZ / 10;
833 err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
834 if (err != header_size) {
839 err = decode_header(connection, connection->data.rbuf, &pi);
846 * drbd_socket_okay() - Free the socket if its connection is not okay
847 * @sock: pointer to the pointer to the socket.
849 static bool drbd_socket_okay(struct socket **sock)
857 rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
859 if (rr > 0 || rr == -EAGAIN) {
868 static bool connection_established(struct drbd_connection *connection,
869 struct socket **sock1,
870 struct socket **sock2)
876 if (!*sock1 || !*sock2)
880 nc = rcu_dereference(connection->net_conf);
881 timeout = (nc->sock_check_timeo ?: nc->ping_timeo) * HZ / 10;
883 schedule_timeout_interruptible(timeout);
885 ok = drbd_socket_okay(sock1);
886 ok = drbd_socket_okay(sock2) && ok;
891 /* Gets called if a connection is established, or if a new minor gets created
893 int drbd_connected(struct drbd_peer_device *peer_device)
895 struct drbd_device *device = peer_device->device;
898 atomic_set(&device->packet_seq, 0);
899 device->peer_seq = 0;
901 device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
902 &peer_device->connection->cstate_mutex :
903 &device->own_state_mutex;
905 err = drbd_send_sync_param(peer_device);
907 err = drbd_send_sizes(peer_device, 0, 0);
909 err = drbd_send_uuids(peer_device);
911 err = drbd_send_current_state(peer_device);
912 clear_bit(USE_DEGR_WFC_T, &device->flags);
913 clear_bit(RESIZE_PENDING, &device->flags);
914 atomic_set(&device->ap_in_flight, 0);
915 mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
921 * 1 yes, we have a valid connection
922 * 0 oops, did not work out, please try again
923 * -1 peer talks different language,
924 * no point in trying again, please go standalone.
925 * -2 We do not have a network config...
927 static int conn_connect(struct drbd_connection *connection)
929 struct drbd_socket sock, msock;
930 struct drbd_peer_device *peer_device;
933 bool discard_my_data, ok;
934 enum drbd_state_rv rv;
935 struct accept_wait_data ad = {
936 .connection = connection,
937 .door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
940 clear_bit(DISCONNECT_SENT, &connection->flags);
941 if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
944 mutex_init(&sock.mutex);
945 sock.sbuf = connection->data.sbuf;
946 sock.rbuf = connection->data.rbuf;
948 mutex_init(&msock.mutex);
949 msock.sbuf = connection->meta.sbuf;
950 msock.rbuf = connection->meta.rbuf;
953 /* Assume that the peer only understands protocol 80 until we know better. */
954 connection->agreed_pro_version = 80;
956 if (prepare_listen_socket(connection, &ad))
962 s = drbd_try_connect(connection);
966 send_first_packet(connection, &sock, P_INITIAL_DATA);
967 } else if (!msock.socket) {
968 clear_bit(RESOLVE_CONFLICTS, &connection->flags);
970 send_first_packet(connection, &msock, P_INITIAL_META);
972 drbd_err(connection, "Logic error in conn_connect()\n");
973 goto out_release_sockets;
977 if (connection_established(connection, &sock.socket, &msock.socket))
981 s = drbd_wait_for_connect(connection, &ad);
983 int fp = receive_first_packet(connection, s);
984 drbd_socket_okay(&sock.socket);
985 drbd_socket_okay(&msock.socket);
989 drbd_warn(connection, "initial packet S crossed\n");
990 sock_release(sock.socket);
997 set_bit(RESOLVE_CONFLICTS, &connection->flags);
999 drbd_warn(connection, "initial packet M crossed\n");
1000 sock_release(msock.socket);
1007 drbd_warn(connection, "Error receiving initial packet\n");
1010 if (prandom_u32() & 1)
1015 if (connection->cstate <= C_DISCONNECTING)
1016 goto out_release_sockets;
1017 if (signal_pending(current)) {
1018 flush_signals(current);
1020 if (get_t_state(&connection->receiver) == EXITING)
1021 goto out_release_sockets;
1024 ok = connection_established(connection, &sock.socket, &msock.socket);
1028 sock_release(ad.s_listen);
1030 sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1031 msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1033 sock.socket->sk->sk_allocation = GFP_NOIO;
1034 msock.socket->sk->sk_allocation = GFP_NOIO;
1036 sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
1037 msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
1040 * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
1041 * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1042 * first set it to the P_CONNECTION_FEATURES timeout,
1043 * which we set to 4x the configured ping_timeout. */
1045 nc = rcu_dereference(connection->net_conf);
1047 sock.socket->sk->sk_sndtimeo =
1048 sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
1050 msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
1051 timeout = nc->timeout * HZ / 10;
1052 discard_my_data = nc->discard_my_data;
1055 msock.socket->sk->sk_sndtimeo = timeout;
1057 /* we don't want delays.
1058 * we use TCP_CORK where appropriate, though */
1059 drbd_tcp_nodelay(sock.socket);
1060 drbd_tcp_nodelay(msock.socket);
1062 connection->data.socket = sock.socket;
1063 connection->meta.socket = msock.socket;
1064 connection->last_received = jiffies;
1066 h = drbd_do_features(connection);
1070 if (connection->cram_hmac_tfm) {
1071 /* drbd_request_state(device, NS(conn, WFAuth)); */
1072 switch (drbd_do_auth(connection)) {
1074 drbd_err(connection, "Authentication of peer failed\n");
1077 drbd_err(connection, "Authentication of peer failed, trying again.\n");
1082 connection->data.socket->sk->sk_sndtimeo = timeout;
1083 connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1085 if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1088 /* Prevent a race between resync-handshake and
1089 * being promoted to Primary.
1091 * Grab and release the state mutex, so we know that any current
1092 * drbd_set_role() is finished, and any incoming drbd_set_role
1093 * will see the STATE_SENT flag, and wait for it to be cleared.
1095 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1096 mutex_lock(peer_device->device->state_mutex);
1098 set_bit(STATE_SENT, &connection->flags);
1100 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1101 mutex_unlock(peer_device->device->state_mutex);
1104 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1105 struct drbd_device *device = peer_device->device;
1106 kref_get(&device->kref);
1109 if (discard_my_data)
1110 set_bit(DISCARD_MY_DATA, &device->flags);
1112 clear_bit(DISCARD_MY_DATA, &device->flags);
1114 drbd_connected(peer_device);
1115 kref_put(&device->kref, drbd_destroy_device);
1120 rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1121 if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1122 clear_bit(STATE_SENT, &connection->flags);
1126 drbd_thread_start(&connection->ack_receiver);
1127 /* opencoded create_singlethread_workqueue(),
1128 * to be able to use format string arguments */
1129 connection->ack_sender =
1130 alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
1131 if (!connection->ack_sender) {
1132 drbd_err(connection, "Failed to create workqueue ack_sender\n");
1136 mutex_lock(&connection->resource->conf_update);
1137 /* The discard_my_data flag is a single-shot modifier to the next
1138 * connection attempt, the handshake of which is now well underway.
1139 * No need for rcu style copying of the whole struct
1140 * just to clear a single value. */
1141 connection->net_conf->discard_my_data = 0;
1142 mutex_unlock(&connection->resource->conf_update);
1146 out_release_sockets:
1148 sock_release(ad.s_listen);
1150 sock_release(sock.socket);
1152 sock_release(msock.socket);
1156 static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1158 unsigned int header_size = drbd_header_size(connection);
1160 if (header_size == sizeof(struct p_header100) &&
1161 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1162 struct p_header100 *h = header;
1164 drbd_err(connection, "Header padding is not zero\n");
1167 pi->vnr = be16_to_cpu(h->volume);
1168 pi->cmd = be16_to_cpu(h->command);
1169 pi->size = be32_to_cpu(h->length);
1170 } else if (header_size == sizeof(struct p_header95) &&
1171 *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1172 struct p_header95 *h = header;
1173 pi->cmd = be16_to_cpu(h->command);
1174 pi->size = be32_to_cpu(h->length);
1176 } else if (header_size == sizeof(struct p_header80) &&
1177 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1178 struct p_header80 *h = header;
1179 pi->cmd = be16_to_cpu(h->command);
1180 pi->size = be16_to_cpu(h->length);
1183 drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1184 be32_to_cpu(*(__be32 *)header),
1185 connection->agreed_pro_version);
1188 pi->data = header + header_size;
1192 static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1194 void *buffer = connection->data.rbuf;
1197 err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1201 err = decode_header(connection, buffer, pi);
1202 connection->last_received = jiffies;
1207 static void drbd_flush(struct drbd_connection *connection)
1210 struct drbd_peer_device *peer_device;
1213 if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1215 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1216 struct drbd_device *device = peer_device->device;
1218 if (!get_ldev(device))
1220 kref_get(&device->kref);
1223 /* Right now, we have only this one synchronous code path
1224 * for flushes between request epochs.
1225 * We may want to make those asynchronous,
1226 * or at least parallelize the flushes to the volume devices.
1228 device->flush_jif = jiffies;
1229 set_bit(FLUSH_PENDING, &device->flags);
1230 rv = blkdev_issue_flush(device->ldev->backing_bdev,
1232 clear_bit(FLUSH_PENDING, &device->flags);
1234 drbd_info(device, "local disk flush failed with status %d\n", rv);
1235 /* would rather check on EOPNOTSUPP, but that is not reliable.
1236 * don't try again for ANY return value != 0
1237 * if (rv == -EOPNOTSUPP) */
1238 drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1241 kref_put(&device->kref, drbd_destroy_device);
1252 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1253 * @device: DRBD device.
1254 * @epoch: Epoch object.
1257 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1258 struct drbd_epoch *epoch,
1259 enum epoch_event ev)
1262 struct drbd_epoch *next_epoch;
1263 enum finish_epoch rv = FE_STILL_LIVE;
1265 spin_lock(&connection->epoch_lock);
1269 epoch_size = atomic_read(&epoch->epoch_size);
1271 switch (ev & ~EV_CLEANUP) {
1273 atomic_dec(&epoch->active);
1275 case EV_GOT_BARRIER_NR:
1276 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1278 case EV_BECAME_LAST:
1283 if (epoch_size != 0 &&
1284 atomic_read(&epoch->active) == 0 &&
1285 (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1286 if (!(ev & EV_CLEANUP)) {
1287 spin_unlock(&connection->epoch_lock);
1288 drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1289 spin_lock(&connection->epoch_lock);
1292 /* FIXME: dec unacked on connection, once we have
1293 * something to count pending connection packets in. */
1294 if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1295 dec_unacked(epoch->connection);
1298 if (connection->current_epoch != epoch) {
1299 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1300 list_del(&epoch->list);
1301 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1302 connection->epochs--;
1305 if (rv == FE_STILL_LIVE)
1309 atomic_set(&epoch->epoch_size, 0);
1310 /* atomic_set(&epoch->active, 0); is already zero */
1311 if (rv == FE_STILL_LIVE)
1322 spin_unlock(&connection->epoch_lock);
1327 static enum write_ordering_e
1328 max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1330 struct disk_conf *dc;
1332 dc = rcu_dereference(bdev->disk_conf);
1334 if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1336 if (wo == WO_DRAIN_IO && !dc->disk_drain)
1343 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1344 * @connection: DRBD connection.
1345 * @wo: Write ordering method to try.
1347 void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1348 enum write_ordering_e wo)
1350 struct drbd_device *device;
1351 enum write_ordering_e pwo;
1353 static char *write_ordering_str[] = {
1355 [WO_DRAIN_IO] = "drain",
1356 [WO_BDEV_FLUSH] = "flush",
1359 pwo = resource->write_ordering;
1360 if (wo != WO_BDEV_FLUSH)
1363 idr_for_each_entry(&resource->devices, device, vnr) {
1364 if (get_ldev(device)) {
1365 wo = max_allowed_wo(device->ldev, wo);
1366 if (device->ldev == bdev)
1373 wo = max_allowed_wo(bdev, wo);
1377 resource->write_ordering = wo;
1378 if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1379 drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1383 * drbd_submit_peer_request()
1384 * @device: DRBD device.
1385 * @peer_req: peer request
1386 * @rw: flag field, see bio->bi_rw
1388 * May spread the pages to multiple bios,
1389 * depending on bio_add_page restrictions.
1391 * Returns 0 if all bios have been submitted,
1392 * -ENOMEM if we could not allocate enough bios,
1393 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1394 * single page to an empty bio (which should never happen and likely indicates
1395 * that the lower level IO stack is in some way broken). This has been observed
1396 * on certain Xen deployments.
1398 /* TODO allocate from our own bio_set. */
1399 int drbd_submit_peer_request(struct drbd_device *device,
1400 struct drbd_peer_request *peer_req,
1401 const unsigned op, const unsigned op_flags,
1402 const int fault_type)
1404 struct bio *bios = NULL;
1406 struct page *page = peer_req->pages;
1407 sector_t sector = peer_req->i.sector;
1408 unsigned data_size = peer_req->i.size;
1409 unsigned n_bios = 0;
1410 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
1413 if (peer_req->flags & EE_IS_TRIM_USE_ZEROOUT) {
1414 /* wait for all pending IO completions, before we start
1415 * zeroing things out. */
1416 conn_wait_active_ee_empty(peer_req->peer_device->connection);
1417 /* add it to the active list now,
1418 * so we can find it to present it in debugfs */
1419 peer_req->submit_jif = jiffies;
1420 peer_req->flags |= EE_SUBMITTED;
1421 spin_lock_irq(&device->resource->req_lock);
1422 list_add_tail(&peer_req->w.list, &device->active_ee);
1423 spin_unlock_irq(&device->resource->req_lock);
1424 if (blkdev_issue_zeroout(device->ldev->backing_bdev,
1425 sector, data_size >> 9, GFP_NOIO, false))
1426 peer_req->flags |= EE_WAS_ERROR;
1427 drbd_endio_write_sec_final(peer_req);
1431 /* Discards don't have any payload.
1432 * But the scsi layer still expects a bio_vec it can use internally,
1433 * see sd_setup_discard_cmnd() and blk_add_request_payload(). */
1434 if (peer_req->flags & EE_IS_TRIM)
1437 /* In most cases, we will only need one bio. But in case the lower
1438 * level restrictions happen to be different at this offset on this
1439 * side than those of the sending peer, we may need to submit the
1440 * request in more than one bio.
1442 * Plain bio_alloc is good enough here, this is no DRBD internally
1443 * generated bio, but a bio allocated on behalf of the peer.
1446 bio = bio_alloc(GFP_NOIO, nr_pages);
1448 drbd_err(device, "submit_ee: Allocation of a bio failed (nr_pages=%u)\n", nr_pages);
1451 /* > peer_req->i.sector, unless this is the first bio */
1452 bio->bi_iter.bi_sector = sector;
1453 bio->bi_bdev = device->ldev->backing_bdev;
1454 bio_set_op_attrs(bio, op, op_flags);
1455 bio->bi_private = peer_req;
1456 bio->bi_end_io = drbd_peer_request_endio;
1458 bio->bi_next = bios;
1462 if (op == REQ_OP_DISCARD) {
1463 bio->bi_iter.bi_size = data_size;
1467 page_chain_for_each(page) {
1468 unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1469 if (!bio_add_page(bio, page, len, 0)) {
1470 /* A single page must always be possible!
1471 * But in case it fails anyways,
1472 * we deal with it, and complain (below). */
1473 if (bio->bi_vcnt == 0) {
1475 "bio_add_page failed for len=%u, "
1476 "bi_vcnt=0 (bi_sector=%llu)\n",
1477 len, (uint64_t)bio->bi_iter.bi_sector);
1487 D_ASSERT(device, data_size == 0);
1489 D_ASSERT(device, page == NULL);
1491 atomic_set(&peer_req->pending_bios, n_bios);
1492 /* for debugfs: update timestamp, mark as submitted */
1493 peer_req->submit_jif = jiffies;
1494 peer_req->flags |= EE_SUBMITTED;
1497 bios = bios->bi_next;
1498 bio->bi_next = NULL;
1500 drbd_generic_make_request(device, fault_type, bio);
1507 bios = bios->bi_next;
1513 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1514 struct drbd_peer_request *peer_req)
1516 struct drbd_interval *i = &peer_req->i;
1518 drbd_remove_interval(&device->write_requests, i);
1519 drbd_clear_interval(i);
1521 /* Wake up any processes waiting for this peer request to complete. */
1523 wake_up(&device->misc_wait);
1526 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1528 struct drbd_peer_device *peer_device;
1532 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1533 struct drbd_device *device = peer_device->device;
1535 kref_get(&device->kref);
1537 drbd_wait_ee_list_empty(device, &device->active_ee);
1538 kref_put(&device->kref, drbd_destroy_device);
1544 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1547 struct p_barrier *p = pi->data;
1548 struct drbd_epoch *epoch;
1550 /* FIXME these are unacked on connection,
1551 * not a specific (peer)device.
1553 connection->current_epoch->barrier_nr = p->barrier;
1554 connection->current_epoch->connection = connection;
1555 rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1557 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1558 * the activity log, which means it would not be resynced in case the
1559 * R_PRIMARY crashes now.
1560 * Therefore we must send the barrier_ack after the barrier request was
1562 switch (connection->resource->write_ordering) {
1564 if (rv == FE_RECYCLED)
1567 /* receiver context, in the writeout path of the other node.
1568 * avoid potential distributed deadlock */
1569 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1573 drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1578 conn_wait_active_ee_empty(connection);
1579 drbd_flush(connection);
1581 if (atomic_read(&connection->current_epoch->epoch_size)) {
1582 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1589 drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1590 connection->resource->write_ordering);
1595 atomic_set(&epoch->epoch_size, 0);
1596 atomic_set(&epoch->active, 0);
1598 spin_lock(&connection->epoch_lock);
1599 if (atomic_read(&connection->current_epoch->epoch_size)) {
1600 list_add(&epoch->list, &connection->current_epoch->list);
1601 connection->current_epoch = epoch;
1602 connection->epochs++;
1604 /* The current_epoch got recycled while we allocated this one... */
1607 spin_unlock(&connection->epoch_lock);
1612 /* used from receive_RSDataReply (recv_resync_read)
1613 * and from receive_Data */
1614 static struct drbd_peer_request *
1615 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1616 struct packet_info *pi) __must_hold(local)
1618 struct drbd_device *device = peer_device->device;
1619 const sector_t capacity = drbd_get_capacity(device->this_bdev);
1620 struct drbd_peer_request *peer_req;
1622 int digest_size, err;
1623 unsigned int data_size = pi->size, ds;
1624 void *dig_in = peer_device->connection->int_dig_in;
1625 void *dig_vv = peer_device->connection->int_dig_vv;
1626 unsigned long *data;
1627 struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1630 if (!trim && peer_device->connection->peer_integrity_tfm) {
1631 digest_size = crypto_ahash_digestsize(peer_device->connection->peer_integrity_tfm);
1633 * FIXME: Receive the incoming digest into the receive buffer
1634 * here, together with its struct p_data?
1636 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1639 data_size -= digest_size;
1643 D_ASSERT(peer_device, data_size == 0);
1644 data_size = be32_to_cpu(trim->size);
1647 if (!expect(IS_ALIGNED(data_size, 512)))
1649 /* prepare for larger trim requests. */
1650 if (!trim && !expect(data_size <= DRBD_MAX_BIO_SIZE))
1653 /* even though we trust out peer,
1654 * we sometimes have to double check. */
1655 if (sector + (data_size>>9) > capacity) {
1656 drbd_err(device, "request from peer beyond end of local disk: "
1657 "capacity: %llus < sector: %llus + size: %u\n",
1658 (unsigned long long)capacity,
1659 (unsigned long long)sector, data_size);
1663 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1664 * "criss-cross" setup, that might cause write-out on some other DRBD,
1665 * which in turn might block on the other node at this very place. */
1666 peer_req = drbd_alloc_peer_req(peer_device, id, sector, data_size, trim == NULL, GFP_NOIO);
1670 peer_req->flags |= EE_WRITE;
1675 page = peer_req->pages;
1676 page_chain_for_each(page) {
1677 unsigned len = min_t(int, ds, PAGE_SIZE);
1679 err = drbd_recv_all_warn(peer_device->connection, data, len);
1680 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1681 drbd_err(device, "Fault injection: Corrupting data on receive\n");
1682 data[0] = data[0] ^ (unsigned long)-1;
1686 drbd_free_peer_req(device, peer_req);
1693 drbd_csum_ee(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv);
1694 if (memcmp(dig_in, dig_vv, digest_size)) {
1695 drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1696 (unsigned long long)sector, data_size);
1697 drbd_free_peer_req(device, peer_req);
1701 device->recv_cnt += data_size >> 9;
1705 /* drbd_drain_block() just takes a data block
1706 * out of the socket input buffer, and discards it.
1708 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1717 page = drbd_alloc_pages(peer_device, 1, 1);
1721 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1723 err = drbd_recv_all_warn(peer_device->connection, data, len);
1729 drbd_free_pages(peer_device->device, page, 0);
1733 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1734 sector_t sector, int data_size)
1736 struct bio_vec bvec;
1737 struct bvec_iter iter;
1739 int digest_size, err, expect;
1740 void *dig_in = peer_device->connection->int_dig_in;
1741 void *dig_vv = peer_device->connection->int_dig_vv;
1744 if (peer_device->connection->peer_integrity_tfm) {
1745 digest_size = crypto_ahash_digestsize(peer_device->connection->peer_integrity_tfm);
1746 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1749 data_size -= digest_size;
1752 /* optimistically update recv_cnt. if receiving fails below,
1753 * we disconnect anyways, and counters will be reset. */
1754 peer_device->device->recv_cnt += data_size>>9;
1756 bio = req->master_bio;
1757 D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
1759 bio_for_each_segment(bvec, bio, iter) {
1760 void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
1761 expect = min_t(int, data_size, bvec.bv_len);
1762 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
1763 kunmap(bvec.bv_page);
1766 data_size -= expect;
1770 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
1771 if (memcmp(dig_in, dig_vv, digest_size)) {
1772 drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
1777 D_ASSERT(peer_device->device, data_size == 0);
1782 * e_end_resync_block() is called in ack_sender context via
1783 * drbd_finish_peer_reqs().
1785 static int e_end_resync_block(struct drbd_work *w, int unused)
1787 struct drbd_peer_request *peer_req =
1788 container_of(w, struct drbd_peer_request, w);
1789 struct drbd_peer_device *peer_device = peer_req->peer_device;
1790 struct drbd_device *device = peer_device->device;
1791 sector_t sector = peer_req->i.sector;
1794 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1796 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1797 drbd_set_in_sync(device, sector, peer_req->i.size);
1798 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
1800 /* Record failure to sync */
1801 drbd_rs_failed_io(device, sector, peer_req->i.size);
1803 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1805 dec_unacked(device);
1810 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
1811 struct packet_info *pi) __releases(local)
1813 struct drbd_device *device = peer_device->device;
1814 struct drbd_peer_request *peer_req;
1816 peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
1820 dec_rs_pending(device);
1822 inc_unacked(device);
1823 /* corresponding dec_unacked() in e_end_resync_block()
1824 * respective _drbd_clear_done_ee */
1826 peer_req->w.cb = e_end_resync_block;
1827 peer_req->submit_jif = jiffies;
1829 spin_lock_irq(&device->resource->req_lock);
1830 list_add_tail(&peer_req->w.list, &device->sync_ee);
1831 spin_unlock_irq(&device->resource->req_lock);
1833 atomic_add(pi->size >> 9, &device->rs_sect_ev);
1834 if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE, 0,
1835 DRBD_FAULT_RS_WR) == 0)
1838 /* don't care for the reason here */
1839 drbd_err(device, "submit failed, triggering re-connect\n");
1840 spin_lock_irq(&device->resource->req_lock);
1841 list_del(&peer_req->w.list);
1842 spin_unlock_irq(&device->resource->req_lock);
1844 drbd_free_peer_req(device, peer_req);
1850 static struct drbd_request *
1851 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
1852 sector_t sector, bool missing_ok, const char *func)
1854 struct drbd_request *req;
1856 /* Request object according to our peer */
1857 req = (struct drbd_request *)(unsigned long)id;
1858 if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
1861 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
1862 (unsigned long)id, (unsigned long long)sector);
1867 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
1869 struct drbd_peer_device *peer_device;
1870 struct drbd_device *device;
1871 struct drbd_request *req;
1874 struct p_data *p = pi->data;
1876 peer_device = conn_peer_device(connection, pi->vnr);
1879 device = peer_device->device;
1881 sector = be64_to_cpu(p->sector);
1883 spin_lock_irq(&device->resource->req_lock);
1884 req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
1885 spin_unlock_irq(&device->resource->req_lock);
1889 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1890 * special casing it there for the various failure cases.
1891 * still no race with drbd_fail_pending_reads */
1892 err = recv_dless_read(peer_device, req, sector, pi->size);
1894 req_mod(req, DATA_RECEIVED);
1895 /* else: nothing. handled from drbd_disconnect...
1896 * I don't think we may complete this just yet
1897 * in case we are "on-disconnect: freeze" */
1902 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
1904 struct drbd_peer_device *peer_device;
1905 struct drbd_device *device;
1908 struct p_data *p = pi->data;
1910 peer_device = conn_peer_device(connection, pi->vnr);
1913 device = peer_device->device;
1915 sector = be64_to_cpu(p->sector);
1916 D_ASSERT(device, p->block_id == ID_SYNCER);
1918 if (get_ldev(device)) {
1919 /* data is submitted to disk within recv_resync_read.
1920 * corresponding put_ldev done below on error,
1921 * or in drbd_peer_request_endio. */
1922 err = recv_resync_read(peer_device, sector, pi);
1924 if (__ratelimit(&drbd_ratelimit_state))
1925 drbd_err(device, "Can not write resync data to local disk.\n");
1927 err = drbd_drain_block(peer_device, pi->size);
1929 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
1932 atomic_add(pi->size >> 9, &device->rs_sect_in);
1937 static void restart_conflicting_writes(struct drbd_device *device,
1938 sector_t sector, int size)
1940 struct drbd_interval *i;
1941 struct drbd_request *req;
1943 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
1946 req = container_of(i, struct drbd_request, i);
1947 if (req->rq_state & RQ_LOCAL_PENDING ||
1948 !(req->rq_state & RQ_POSTPONED))
1950 /* as it is RQ_POSTPONED, this will cause it to
1951 * be queued on the retry workqueue. */
1952 __req_mod(req, CONFLICT_RESOLVED, NULL);
1957 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
1959 static int e_end_block(struct drbd_work *w, int cancel)
1961 struct drbd_peer_request *peer_req =
1962 container_of(w, struct drbd_peer_request, w);
1963 struct drbd_peer_device *peer_device = peer_req->peer_device;
1964 struct drbd_device *device = peer_device->device;
1965 sector_t sector = peer_req->i.sector;
1968 if (peer_req->flags & EE_SEND_WRITE_ACK) {
1969 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1970 pcmd = (device->state.conn >= C_SYNC_SOURCE &&
1971 device->state.conn <= C_PAUSED_SYNC_T &&
1972 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
1973 P_RS_WRITE_ACK : P_WRITE_ACK;
1974 err = drbd_send_ack(peer_device, pcmd, peer_req);
1975 if (pcmd == P_RS_WRITE_ACK)
1976 drbd_set_in_sync(device, sector, peer_req->i.size);
1978 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1979 /* we expect it to be marked out of sync anyways...
1980 * maybe assert this? */
1982 dec_unacked(device);
1985 /* we delete from the conflict detection hash _after_ we sent out the
1986 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1987 if (peer_req->flags & EE_IN_INTERVAL_TREE) {
1988 spin_lock_irq(&device->resource->req_lock);
1989 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
1990 drbd_remove_epoch_entry_interval(device, peer_req);
1991 if (peer_req->flags & EE_RESTART_REQUESTS)
1992 restart_conflicting_writes(device, sector, peer_req->i.size);
1993 spin_unlock_irq(&device->resource->req_lock);
1995 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1997 drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2002 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2004 struct drbd_peer_request *peer_req =
2005 container_of(w, struct drbd_peer_request, w);
2006 struct drbd_peer_device *peer_device = peer_req->peer_device;
2009 err = drbd_send_ack(peer_device, ack, peer_req);
2010 dec_unacked(peer_device->device);
2015 static int e_send_superseded(struct drbd_work *w, int unused)
2017 return e_send_ack(w, P_SUPERSEDED);
2020 static int e_send_retry_write(struct drbd_work *w, int unused)
2022 struct drbd_peer_request *peer_req =
2023 container_of(w, struct drbd_peer_request, w);
2024 struct drbd_connection *connection = peer_req->peer_device->connection;
2026 return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2027 P_RETRY_WRITE : P_SUPERSEDED);
2030 static bool seq_greater(u32 a, u32 b)
2033 * We assume 32-bit wrap-around here.
2034 * For 24-bit wrap-around, we would have to shift:
2037 return (s32)a - (s32)b > 0;
2040 static u32 seq_max(u32 a, u32 b)
2042 return seq_greater(a, b) ? a : b;
2045 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2047 struct drbd_device *device = peer_device->device;
2048 unsigned int newest_peer_seq;
2050 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2051 spin_lock(&device->peer_seq_lock);
2052 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2053 device->peer_seq = newest_peer_seq;
2054 spin_unlock(&device->peer_seq_lock);
2055 /* wake up only if we actually changed device->peer_seq */
2056 if (peer_seq == newest_peer_seq)
2057 wake_up(&device->seq_wait);
2061 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2063 return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2066 /* maybe change sync_ee into interval trees as well? */
2067 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2069 struct drbd_peer_request *rs_req;
2072 spin_lock_irq(&device->resource->req_lock);
2073 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2074 if (overlaps(peer_req->i.sector, peer_req->i.size,
2075 rs_req->i.sector, rs_req->i.size)) {
2080 spin_unlock_irq(&device->resource->req_lock);
2085 /* Called from receive_Data.
2086 * Synchronize packets on sock with packets on msock.
2088 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2089 * packet traveling on msock, they are still processed in the order they have
2092 * Note: we don't care for Ack packets overtaking P_DATA packets.
2094 * In case packet_seq is larger than device->peer_seq number, there are
2095 * outstanding packets on the msock. We wait for them to arrive.
2096 * In case we are the logically next packet, we update device->peer_seq
2097 * ourselves. Correctly handles 32bit wrap around.
2099 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2100 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2101 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2102 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2104 * returns 0 if we may process the packet,
2105 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2106 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2108 struct drbd_device *device = peer_device->device;
2113 if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2116 spin_lock(&device->peer_seq_lock);
2118 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2119 device->peer_seq = seq_max(device->peer_seq, peer_seq);
2123 if (signal_pending(current)) {
2129 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2135 /* Only need to wait if two_primaries is enabled */
2136 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2137 spin_unlock(&device->peer_seq_lock);
2139 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2141 timeout = schedule_timeout(timeout);
2142 spin_lock(&device->peer_seq_lock);
2145 drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2149 spin_unlock(&device->peer_seq_lock);
2150 finish_wait(&device->seq_wait, &wait);
2154 /* see also bio_flags_to_wire()
2155 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2156 * flags and back. We may replicate to other kernel versions. */
2157 static unsigned long wire_flags_to_bio_flags(u32 dpf)
2159 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2160 (dpf & DP_FUA ? REQ_FUA : 0) |
2161 (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2164 static unsigned long wire_flags_to_bio_op(u32 dpf)
2166 if (dpf & DP_DISCARD)
2167 return REQ_OP_DISCARD;
2169 return REQ_OP_WRITE;
2172 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2175 struct drbd_interval *i;
2178 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2179 struct drbd_request *req;
2180 struct bio_and_error m;
2184 req = container_of(i, struct drbd_request, i);
2185 if (!(req->rq_state & RQ_POSTPONED))
2187 req->rq_state &= ~RQ_POSTPONED;
2188 __req_mod(req, NEG_ACKED, &m);
2189 spin_unlock_irq(&device->resource->req_lock);
2191 complete_master_bio(device, &m);
2192 spin_lock_irq(&device->resource->req_lock);
2197 static int handle_write_conflicts(struct drbd_device *device,
2198 struct drbd_peer_request *peer_req)
2200 struct drbd_connection *connection = peer_req->peer_device->connection;
2201 bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2202 sector_t sector = peer_req->i.sector;
2203 const unsigned int size = peer_req->i.size;
2204 struct drbd_interval *i;
2209 * Inserting the peer request into the write_requests tree will prevent
2210 * new conflicting local requests from being added.
2212 drbd_insert_interval(&device->write_requests, &peer_req->i);
2215 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2216 if (i == &peer_req->i)
2223 * Our peer has sent a conflicting remote request; this
2224 * should not happen in a two-node setup. Wait for the
2225 * earlier peer request to complete.
2227 err = drbd_wait_misc(device, i);
2233 equal = i->sector == sector && i->size == size;
2234 if (resolve_conflicts) {
2236 * If the peer request is fully contained within the
2237 * overlapping request, it can be considered overwritten
2238 * and thus superseded; otherwise, it will be retried
2239 * once all overlapping requests have completed.
2241 bool superseded = i->sector <= sector && i->sector +
2242 (i->size >> 9) >= sector + (size >> 9);
2245 drbd_alert(device, "Concurrent writes detected: "
2246 "local=%llus +%u, remote=%llus +%u, "
2247 "assuming %s came first\n",
2248 (unsigned long long)i->sector, i->size,
2249 (unsigned long long)sector, size,
2250 superseded ? "local" : "remote");
2252 peer_req->w.cb = superseded ? e_send_superseded :
2254 list_add_tail(&peer_req->w.list, &device->done_ee);
2255 queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2260 struct drbd_request *req =
2261 container_of(i, struct drbd_request, i);
2264 drbd_alert(device, "Concurrent writes detected: "
2265 "local=%llus +%u, remote=%llus +%u\n",
2266 (unsigned long long)i->sector, i->size,
2267 (unsigned long long)sector, size);
2269 if (req->rq_state & RQ_LOCAL_PENDING ||
2270 !(req->rq_state & RQ_POSTPONED)) {
2272 * Wait for the node with the discard flag to
2273 * decide if this request has been superseded
2274 * or needs to be retried.
2275 * Requests that have been superseded will
2276 * disappear from the write_requests tree.
2278 * In addition, wait for the conflicting
2279 * request to finish locally before submitting
2280 * the conflicting peer request.
2282 err = drbd_wait_misc(device, &req->i);
2284 _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2285 fail_postponed_requests(device, sector, size);
2291 * Remember to restart the conflicting requests after
2292 * the new peer request has completed.
2294 peer_req->flags |= EE_RESTART_REQUESTS;
2301 drbd_remove_epoch_entry_interval(device, peer_req);
2305 /* mirrored write */
2306 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2308 struct drbd_peer_device *peer_device;
2309 struct drbd_device *device;
2310 struct net_conf *nc;
2312 struct drbd_peer_request *peer_req;
2313 struct p_data *p = pi->data;
2314 u32 peer_seq = be32_to_cpu(p->seq_num);
2319 peer_device = conn_peer_device(connection, pi->vnr);
2322 device = peer_device->device;
2324 if (!get_ldev(device)) {
2327 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2328 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2329 atomic_inc(&connection->current_epoch->epoch_size);
2330 err2 = drbd_drain_block(peer_device, pi->size);
2337 * Corresponding put_ldev done either below (on various errors), or in
2338 * drbd_peer_request_endio, if we successfully submit the data at the
2339 * end of this function.
2342 sector = be64_to_cpu(p->sector);
2343 peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2349 peer_req->w.cb = e_end_block;
2350 peer_req->submit_jif = jiffies;
2351 peer_req->flags |= EE_APPLICATION;
2353 dp_flags = be32_to_cpu(p->dp_flags);
2354 op = wire_flags_to_bio_op(dp_flags);
2355 op_flags = wire_flags_to_bio_flags(dp_flags);
2356 if (pi->cmd == P_TRIM) {
2357 struct request_queue *q = bdev_get_queue(device->ldev->backing_bdev);
2358 peer_req->flags |= EE_IS_TRIM;
2359 if (!blk_queue_discard(q))
2360 peer_req->flags |= EE_IS_TRIM_USE_ZEROOUT;
2361 D_ASSERT(peer_device, peer_req->i.size > 0);
2362 D_ASSERT(peer_device, op == REQ_OP_DISCARD);
2363 D_ASSERT(peer_device, peer_req->pages == NULL);
2364 } else if (peer_req->pages == NULL) {
2365 D_ASSERT(device, peer_req->i.size == 0);
2366 D_ASSERT(device, dp_flags & DP_FLUSH);
2369 if (dp_flags & DP_MAY_SET_IN_SYNC)
2370 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2372 spin_lock(&connection->epoch_lock);
2373 peer_req->epoch = connection->current_epoch;
2374 atomic_inc(&peer_req->epoch->epoch_size);
2375 atomic_inc(&peer_req->epoch->active);
2376 spin_unlock(&connection->epoch_lock);
2379 nc = rcu_dereference(peer_device->connection->net_conf);
2380 tp = nc->two_primaries;
2381 if (peer_device->connection->agreed_pro_version < 100) {
2382 switch (nc->wire_protocol) {
2384 dp_flags |= DP_SEND_WRITE_ACK;
2387 dp_flags |= DP_SEND_RECEIVE_ACK;
2393 if (dp_flags & DP_SEND_WRITE_ACK) {
2394 peer_req->flags |= EE_SEND_WRITE_ACK;
2395 inc_unacked(device);
2396 /* corresponding dec_unacked() in e_end_block()
2397 * respective _drbd_clear_done_ee */
2400 if (dp_flags & DP_SEND_RECEIVE_ACK) {
2401 /* I really don't like it that the receiver thread
2402 * sends on the msock, but anyways */
2403 drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2407 /* two primaries implies protocol C */
2408 D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2409 peer_req->flags |= EE_IN_INTERVAL_TREE;
2410 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2412 goto out_interrupted;
2413 spin_lock_irq(&device->resource->req_lock);
2414 err = handle_write_conflicts(device, peer_req);
2416 spin_unlock_irq(&device->resource->req_lock);
2417 if (err == -ENOENT) {
2421 goto out_interrupted;
2424 update_peer_seq(peer_device, peer_seq);
2425 spin_lock_irq(&device->resource->req_lock);
2427 /* if we use the zeroout fallback code, we process synchronously
2428 * and we wait for all pending requests, respectively wait for
2429 * active_ee to become empty in drbd_submit_peer_request();
2430 * better not add ourselves here. */
2431 if ((peer_req->flags & EE_IS_TRIM_USE_ZEROOUT) == 0)
2432 list_add_tail(&peer_req->w.list, &device->active_ee);
2433 spin_unlock_irq(&device->resource->req_lock);
2435 if (device->state.conn == C_SYNC_TARGET)
2436 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2438 if (device->state.pdsk < D_INCONSISTENT) {
2439 /* In case we have the only disk of the cluster, */
2440 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2441 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2442 drbd_al_begin_io(device, &peer_req->i);
2443 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2446 err = drbd_submit_peer_request(device, peer_req, op, op_flags,
2451 /* don't care for the reason here */
2452 drbd_err(device, "submit failed, triggering re-connect\n");
2453 spin_lock_irq(&device->resource->req_lock);
2454 list_del(&peer_req->w.list);
2455 drbd_remove_epoch_entry_interval(device, peer_req);
2456 spin_unlock_irq(&device->resource->req_lock);
2457 if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2458 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2459 drbd_al_complete_io(device, &peer_req->i);
2463 drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT + EV_CLEANUP);
2465 drbd_free_peer_req(device, peer_req);
2469 /* We may throttle resync, if the lower device seems to be busy,
2470 * and current sync rate is above c_min_rate.
2472 * To decide whether or not the lower device is busy, we use a scheme similar
2473 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2474 * (more than 64 sectors) of activity we cannot account for with our own resync
2475 * activity, it obviously is "busy".
2477 * The current sync rate used here uses only the most recent two step marks,
2478 * to have a short time average so we can react faster.
2480 bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2481 bool throttle_if_app_is_waiting)
2483 struct lc_element *tmp;
2484 bool throttle = drbd_rs_c_min_rate_throttle(device);
2486 if (!throttle || throttle_if_app_is_waiting)
2489 spin_lock_irq(&device->al_lock);
2490 tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2492 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2493 if (test_bit(BME_PRIORITY, &bm_ext->flags))
2495 /* Do not slow down if app IO is already waiting for this extent,
2496 * and our progress is necessary for application IO to complete. */
2498 spin_unlock_irq(&device->al_lock);
2503 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2505 struct gendisk *disk = device->ldev->backing_bdev->bd_contains->bd_disk;
2506 unsigned long db, dt, dbdt;
2507 unsigned int c_min_rate;
2511 c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2514 /* feature disabled? */
2515 if (c_min_rate == 0)
2518 curr_events = (int)part_stat_read(&disk->part0, sectors[0]) +
2519 (int)part_stat_read(&disk->part0, sectors[1]) -
2520 atomic_read(&device->rs_sect_ev);
2522 if (atomic_read(&device->ap_actlog_cnt)
2523 || curr_events - device->rs_last_events > 64) {
2524 unsigned long rs_left;
2527 device->rs_last_events = curr_events;
2529 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2531 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2533 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2534 rs_left = device->ov_left;
2536 rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2538 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2541 db = device->rs_mark_left[i] - rs_left;
2542 dbdt = Bit2KB(db/dt);
2544 if (dbdt > c_min_rate)
2550 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2552 struct drbd_peer_device *peer_device;
2553 struct drbd_device *device;
2556 struct drbd_peer_request *peer_req;
2557 struct digest_info *di = NULL;
2559 unsigned int fault_type;
2560 struct p_block_req *p = pi->data;
2562 peer_device = conn_peer_device(connection, pi->vnr);
2565 device = peer_device->device;
2566 capacity = drbd_get_capacity(device->this_bdev);
2568 sector = be64_to_cpu(p->sector);
2569 size = be32_to_cpu(p->blksize);
2571 if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2572 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2573 (unsigned long long)sector, size);
2576 if (sector + (size>>9) > capacity) {
2577 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2578 (unsigned long long)sector, size);
2582 if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2585 case P_DATA_REQUEST:
2586 drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2588 case P_RS_DATA_REQUEST:
2589 case P_CSUM_RS_REQUEST:
2591 drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2595 dec_rs_pending(device);
2596 drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2601 if (verb && __ratelimit(&drbd_ratelimit_state))
2602 drbd_err(device, "Can not satisfy peer's read request, "
2603 "no local data.\n");
2605 /* drain possibly payload */
2606 return drbd_drain_block(peer_device, pi->size);
2609 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2610 * "criss-cross" setup, that might cause write-out on some other DRBD,
2611 * which in turn might block on the other node at this very place. */
2612 peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2613 true /* has real payload */, GFP_NOIO);
2620 case P_DATA_REQUEST:
2621 peer_req->w.cb = w_e_end_data_req;
2622 fault_type = DRBD_FAULT_DT_RD;
2623 /* application IO, don't drbd_rs_begin_io */
2624 peer_req->flags |= EE_APPLICATION;
2627 case P_RS_DATA_REQUEST:
2628 peer_req->w.cb = w_e_end_rsdata_req;
2629 fault_type = DRBD_FAULT_RS_RD;
2630 /* used in the sector offset progress display */
2631 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2635 case P_CSUM_RS_REQUEST:
2636 fault_type = DRBD_FAULT_RS_RD;
2637 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2641 di->digest_size = pi->size;
2642 di->digest = (((char *)di)+sizeof(struct digest_info));
2644 peer_req->digest = di;
2645 peer_req->flags |= EE_HAS_DIGEST;
2647 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2650 if (pi->cmd == P_CSUM_RS_REQUEST) {
2651 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2652 peer_req->w.cb = w_e_end_csum_rs_req;
2653 /* used in the sector offset progress display */
2654 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2655 /* remember to report stats in drbd_resync_finished */
2656 device->use_csums = true;
2657 } else if (pi->cmd == P_OV_REPLY) {
2658 /* track progress, we may need to throttle */
2659 atomic_add(size >> 9, &device->rs_sect_in);
2660 peer_req->w.cb = w_e_end_ov_reply;
2661 dec_rs_pending(device);
2662 /* drbd_rs_begin_io done when we sent this request,
2663 * but accounting still needs to be done. */
2664 goto submit_for_resync;
2669 if (device->ov_start_sector == ~(sector_t)0 &&
2670 peer_device->connection->agreed_pro_version >= 90) {
2671 unsigned long now = jiffies;
2673 device->ov_start_sector = sector;
2674 device->ov_position = sector;
2675 device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2676 device->rs_total = device->ov_left;
2677 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2678 device->rs_mark_left[i] = device->ov_left;
2679 device->rs_mark_time[i] = now;
2681 drbd_info(device, "Online Verify start sector: %llu\n",
2682 (unsigned long long)sector);
2684 peer_req->w.cb = w_e_end_ov_req;
2685 fault_type = DRBD_FAULT_RS_RD;
2692 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2693 * wrt the receiver, but it is not as straightforward as it may seem.
2694 * Various places in the resync start and stop logic assume resync
2695 * requests are processed in order, requeuing this on the worker thread
2696 * introduces a bunch of new code for synchronization between threads.
2698 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2699 * "forever", throttling after drbd_rs_begin_io will lock that extent
2700 * for application writes for the same time. For now, just throttle
2701 * here, where the rest of the code expects the receiver to sleep for
2705 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2706 * this defers syncer requests for some time, before letting at least
2707 * on request through. The resync controller on the receiving side
2708 * will adapt to the incoming rate accordingly.
2710 * We cannot throttle here if remote is Primary/SyncTarget:
2711 * we would also throttle its application reads.
2712 * In that case, throttling is done on the SyncTarget only.
2715 /* Even though this may be a resync request, we do add to "read_ee";
2716 * "sync_ee" is only used for resync WRITEs.
2717 * Add to list early, so debugfs can find this request
2718 * even if we have to sleep below. */
2719 spin_lock_irq(&device->resource->req_lock);
2720 list_add_tail(&peer_req->w.list, &device->read_ee);
2721 spin_unlock_irq(&device->resource->req_lock);
2723 update_receiver_timing_details(connection, drbd_rs_should_slow_down);
2724 if (device->state.peer != R_PRIMARY
2725 && drbd_rs_should_slow_down(device, sector, false))
2726 schedule_timeout_uninterruptible(HZ/10);
2727 update_receiver_timing_details(connection, drbd_rs_begin_io);
2728 if (drbd_rs_begin_io(device, sector))
2732 atomic_add(size >> 9, &device->rs_sect_ev);
2735 update_receiver_timing_details(connection, drbd_submit_peer_request);
2736 inc_unacked(device);
2737 if (drbd_submit_peer_request(device, peer_req, REQ_OP_READ, 0,
2741 /* don't care for the reason here */
2742 drbd_err(device, "submit failed, triggering re-connect\n");
2745 spin_lock_irq(&device->resource->req_lock);
2746 list_del(&peer_req->w.list);
2747 spin_unlock_irq(&device->resource->req_lock);
2748 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2751 drbd_free_peer_req(device, peer_req);
2756 * drbd_asb_recover_0p - Recover after split-brain with no remaining primaries
2758 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
2760 struct drbd_device *device = peer_device->device;
2761 int self, peer, rv = -100;
2762 unsigned long ch_self, ch_peer;
2763 enum drbd_after_sb_p after_sb_0p;
2765 self = device->ldev->md.uuid[UI_BITMAP] & 1;
2766 peer = device->p_uuid[UI_BITMAP] & 1;
2768 ch_peer = device->p_uuid[UI_SIZE];
2769 ch_self = device->comm_bm_set;
2772 after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
2774 switch (after_sb_0p) {
2776 case ASB_DISCARD_SECONDARY:
2777 case ASB_CALL_HELPER:
2779 drbd_err(device, "Configuration error.\n");
2781 case ASB_DISCONNECT:
2783 case ASB_DISCARD_YOUNGER_PRI:
2784 if (self == 0 && peer == 1) {
2788 if (self == 1 && peer == 0) {
2792 /* Else fall through to one of the other strategies... */
2793 case ASB_DISCARD_OLDER_PRI:
2794 if (self == 0 && peer == 1) {
2798 if (self == 1 && peer == 0) {
2802 /* Else fall through to one of the other strategies... */
2803 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
2804 "Using discard-least-changes instead\n");
2805 case ASB_DISCARD_ZERO_CHG:
2806 if (ch_peer == 0 && ch_self == 0) {
2807 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2811 if (ch_peer == 0) { rv = 1; break; }
2812 if (ch_self == 0) { rv = -1; break; }
2814 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
2816 case ASB_DISCARD_LEAST_CHG:
2817 if (ch_self < ch_peer)
2819 else if (ch_self > ch_peer)
2821 else /* ( ch_self == ch_peer ) */
2822 /* Well, then use something else. */
2823 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2826 case ASB_DISCARD_LOCAL:
2829 case ASB_DISCARD_REMOTE:
2837 * drbd_asb_recover_1p - Recover after split-brain with one remaining primary
2839 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
2841 struct drbd_device *device = peer_device->device;
2843 enum drbd_after_sb_p after_sb_1p;
2846 after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
2848 switch (after_sb_1p) {
2849 case ASB_DISCARD_YOUNGER_PRI:
2850 case ASB_DISCARD_OLDER_PRI:
2851 case ASB_DISCARD_LEAST_CHG:
2852 case ASB_DISCARD_LOCAL:
2853 case ASB_DISCARD_REMOTE:
2854 case ASB_DISCARD_ZERO_CHG:
2855 drbd_err(device, "Configuration error.\n");
2857 case ASB_DISCONNECT:
2860 hg = drbd_asb_recover_0p(peer_device);
2861 if (hg == -1 && device->state.role == R_SECONDARY)
2863 if (hg == 1 && device->state.role == R_PRIMARY)
2867 rv = drbd_asb_recover_0p(peer_device);
2869 case ASB_DISCARD_SECONDARY:
2870 return device->state.role == R_PRIMARY ? 1 : -1;
2871 case ASB_CALL_HELPER:
2872 hg = drbd_asb_recover_0p(peer_device);
2873 if (hg == -1 && device->state.role == R_PRIMARY) {
2874 enum drbd_state_rv rv2;
2876 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2877 * we might be here in C_WF_REPORT_PARAMS which is transient.
2878 * we do not need to wait for the after state change work either. */
2879 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
2880 if (rv2 != SS_SUCCESS) {
2881 drbd_khelper(device, "pri-lost-after-sb");
2883 drbd_warn(device, "Successfully gave up primary role.\n");
2894 * drbd_asb_recover_2p - Recover after split-brain with two remaining primaries
2896 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
2898 struct drbd_device *device = peer_device->device;
2900 enum drbd_after_sb_p after_sb_2p;
2903 after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
2905 switch (after_sb_2p) {
2906 case ASB_DISCARD_YOUNGER_PRI:
2907 case ASB_DISCARD_OLDER_PRI:
2908 case ASB_DISCARD_LEAST_CHG:
2909 case ASB_DISCARD_LOCAL:
2910 case ASB_DISCARD_REMOTE:
2912 case ASB_DISCARD_SECONDARY:
2913 case ASB_DISCARD_ZERO_CHG:
2914 drbd_err(device, "Configuration error.\n");
2917 rv = drbd_asb_recover_0p(peer_device);
2919 case ASB_DISCONNECT:
2921 case ASB_CALL_HELPER:
2922 hg = drbd_asb_recover_0p(peer_device);
2924 enum drbd_state_rv rv2;
2926 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2927 * we might be here in C_WF_REPORT_PARAMS which is transient.
2928 * we do not need to wait for the after state change work either. */
2929 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
2930 if (rv2 != SS_SUCCESS) {
2931 drbd_khelper(device, "pri-lost-after-sb");
2933 drbd_warn(device, "Successfully gave up primary role.\n");
2943 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
2944 u64 bits, u64 flags)
2947 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
2950 drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2952 (unsigned long long)uuid[UI_CURRENT],
2953 (unsigned long long)uuid[UI_BITMAP],
2954 (unsigned long long)uuid[UI_HISTORY_START],
2955 (unsigned long long)uuid[UI_HISTORY_END],
2956 (unsigned long long)bits,
2957 (unsigned long long)flags);
2961 100 after split brain try auto recover
2962 2 C_SYNC_SOURCE set BitMap
2963 1 C_SYNC_SOURCE use BitMap
2965 -1 C_SYNC_TARGET use BitMap
2966 -2 C_SYNC_TARGET set BitMap
2967 -100 after split brain, disconnect
2968 -1000 unrelated data
2969 -1091 requires proto 91
2970 -1096 requires proto 96
2972 static int drbd_uuid_compare(struct drbd_device *const device, int *rule_nr) __must_hold(local)
2974 struct drbd_peer_device *const peer_device = first_peer_device(device);
2975 struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
2979 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2980 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
2983 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
2987 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
2988 peer != UUID_JUST_CREATED)
2992 if (self != UUID_JUST_CREATED &&
2993 (peer == UUID_JUST_CREATED || peer == (u64)0))
2997 int rct, dc; /* roles at crash time */
2999 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3001 if (connection->agreed_pro_version < 91)
3004 if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3005 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3006 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3007 drbd_uuid_move_history(device);
3008 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3009 device->ldev->md.uuid[UI_BITMAP] = 0;
3011 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3012 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3015 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3022 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3024 if (connection->agreed_pro_version < 91)
3027 if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3028 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3029 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3031 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3032 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3033 device->p_uuid[UI_BITMAP] = 0UL;
3035 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3038 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3045 /* Common power [off|failure] */
3046 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3047 (device->p_uuid[UI_FLAGS] & 2);
3048 /* lowest bit is set when we were primary,
3049 * next bit (weight 2) is set when peer was primary */
3053 case 0: /* !self_pri && !peer_pri */ return 0;
3054 case 1: /* self_pri && !peer_pri */ return 1;
3055 case 2: /* !self_pri && peer_pri */ return -1;
3056 case 3: /* self_pri && peer_pri */
3057 dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3063 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3068 peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3070 if (connection->agreed_pro_version < 96 ?
3071 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3072 (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3073 peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3074 /* The last P_SYNC_UUID did not get though. Undo the last start of
3075 resync as sync source modifications of the peer's UUIDs. */
3077 if (connection->agreed_pro_version < 91)
3080 device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3081 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3083 drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3084 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3091 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3092 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3093 peer = device->p_uuid[i] & ~((u64)1);
3099 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3100 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3105 self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3107 if (connection->agreed_pro_version < 96 ?
3108 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3109 (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3110 self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3111 /* The last P_SYNC_UUID did not get though. Undo the last start of
3112 resync as sync source modifications of our UUIDs. */
3114 if (connection->agreed_pro_version < 91)
3117 __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3118 __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3120 drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3121 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3122 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3130 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3131 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3132 self = device->ldev->md.uuid[i] & ~((u64)1);
3138 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3139 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3140 if (self == peer && self != ((u64)0))
3144 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3145 self = device->ldev->md.uuid[i] & ~((u64)1);
3146 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3147 peer = device->p_uuid[j] & ~((u64)1);
3156 /* drbd_sync_handshake() returns the new conn state on success, or
3157 CONN_MASK (-1) on failure.
3159 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3160 enum drbd_role peer_role,
3161 enum drbd_disk_state peer_disk) __must_hold(local)
3163 struct drbd_device *device = peer_device->device;
3164 enum drbd_conns rv = C_MASK;
3165 enum drbd_disk_state mydisk;
3166 struct net_conf *nc;
3167 int hg, rule_nr, rr_conflict, tentative;
3169 mydisk = device->state.disk;
3170 if (mydisk == D_NEGOTIATING)
3171 mydisk = device->new_state_tmp.disk;
3173 drbd_info(device, "drbd_sync_handshake:\n");
3175 spin_lock_irq(&device->ldev->md.uuid_lock);
3176 drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3177 drbd_uuid_dump(device, "peer", device->p_uuid,
3178 device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3180 hg = drbd_uuid_compare(device, &rule_nr);
3181 spin_unlock_irq(&device->ldev->md.uuid_lock);
3183 drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3186 drbd_alert(device, "Unrelated data, aborting!\n");
3190 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3194 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3195 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
3196 int f = (hg == -100) || abs(hg) == 2;
3197 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3200 drbd_info(device, "Becoming sync %s due to disk states.\n",
3201 hg > 0 ? "source" : "target");
3205 drbd_khelper(device, "initial-split-brain");
3208 nc = rcu_dereference(peer_device->connection->net_conf);
3210 if (hg == 100 || (hg == -100 && nc->always_asbp)) {
3211 int pcount = (device->state.role == R_PRIMARY)
3212 + (peer_role == R_PRIMARY);
3213 int forced = (hg == -100);
3217 hg = drbd_asb_recover_0p(peer_device);
3220 hg = drbd_asb_recover_1p(peer_device);
3223 hg = drbd_asb_recover_2p(peer_device);
3226 if (abs(hg) < 100) {
3227 drbd_warn(device, "Split-Brain detected, %d primaries, "
3228 "automatically solved. Sync from %s node\n",
3229 pcount, (hg < 0) ? "peer" : "this");
3231 drbd_warn(device, "Doing a full sync, since"
3232 " UUIDs where ambiguous.\n");
3239 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3241 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3245 drbd_warn(device, "Split-Brain detected, manually solved. "
3246 "Sync from %s node\n",
3247 (hg < 0) ? "peer" : "this");
3249 rr_conflict = nc->rr_conflict;
3250 tentative = nc->tentative;
3254 /* FIXME this log message is not correct if we end up here
3255 * after an attempted attach on a diskless node.
3256 * We just refuse to attach -- well, we drop the "connection"
3257 * to that disk, in a way... */
3258 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3259 drbd_khelper(device, "split-brain");
3263 if (hg > 0 && mydisk <= D_INCONSISTENT) {
3264 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3268 if (hg < 0 && /* by intention we do not use mydisk here. */
3269 device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3270 switch (rr_conflict) {
3271 case ASB_CALL_HELPER:
3272 drbd_khelper(device, "pri-lost");
3274 case ASB_DISCONNECT:
3275 drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3278 drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3283 if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3285 drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3287 drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3288 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3289 abs(hg) >= 2 ? "full" : "bit-map based");
3294 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3295 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3296 BM_LOCKED_SET_ALLOWED))
3300 if (hg > 0) { /* become sync source. */
3302 } else if (hg < 0) { /* become sync target */
3306 if (drbd_bm_total_weight(device)) {
3307 drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3308 drbd_bm_total_weight(device));
3315 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3317 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3318 if (peer == ASB_DISCARD_REMOTE)
3319 return ASB_DISCARD_LOCAL;
3321 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3322 if (peer == ASB_DISCARD_LOCAL)
3323 return ASB_DISCARD_REMOTE;
3325 /* everything else is valid if they are equal on both sides. */
3329 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3331 struct p_protocol *p = pi->data;
3332 enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3333 int p_proto, p_discard_my_data, p_two_primaries, cf;
3334 struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3335 char integrity_alg[SHARED_SECRET_MAX] = "";
3336 struct crypto_ahash *peer_integrity_tfm = NULL;
3337 void *int_dig_in = NULL, *int_dig_vv = NULL;
3339 p_proto = be32_to_cpu(p->protocol);
3340 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
3341 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
3342 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
3343 p_two_primaries = be32_to_cpu(p->two_primaries);
3344 cf = be32_to_cpu(p->conn_flags);
3345 p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3347 if (connection->agreed_pro_version >= 87) {
3350 if (pi->size > sizeof(integrity_alg))
3352 err = drbd_recv_all(connection, integrity_alg, pi->size);
3355 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3358 if (pi->cmd != P_PROTOCOL_UPDATE) {
3359 clear_bit(CONN_DRY_RUN, &connection->flags);
3361 if (cf & CF_DRY_RUN)
3362 set_bit(CONN_DRY_RUN, &connection->flags);
3365 nc = rcu_dereference(connection->net_conf);
3367 if (p_proto != nc->wire_protocol) {
3368 drbd_err(connection, "incompatible %s settings\n", "protocol");
3369 goto disconnect_rcu_unlock;
3372 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3373 drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3374 goto disconnect_rcu_unlock;
3377 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3378 drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3379 goto disconnect_rcu_unlock;
3382 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3383 drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3384 goto disconnect_rcu_unlock;
3387 if (p_discard_my_data && nc->discard_my_data) {
3388 drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3389 goto disconnect_rcu_unlock;
3392 if (p_two_primaries != nc->two_primaries) {
3393 drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3394 goto disconnect_rcu_unlock;
3397 if (strcmp(integrity_alg, nc->integrity_alg)) {
3398 drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3399 goto disconnect_rcu_unlock;
3405 if (integrity_alg[0]) {
3409 * We can only change the peer data integrity algorithm
3410 * here. Changing our own data integrity algorithm
3411 * requires that we send a P_PROTOCOL_UPDATE packet at
3412 * the same time; otherwise, the peer has no way to
3413 * tell between which packets the algorithm should
3417 peer_integrity_tfm = crypto_alloc_ahash(integrity_alg, 0, CRYPTO_ALG_ASYNC);
3418 if (!peer_integrity_tfm) {
3419 drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3424 hash_size = crypto_ahash_digestsize(peer_integrity_tfm);
3425 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3426 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3427 if (!(int_dig_in && int_dig_vv)) {
3428 drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3433 new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3434 if (!new_net_conf) {
3435 drbd_err(connection, "Allocation of new net_conf failed\n");
3439 mutex_lock(&connection->data.mutex);
3440 mutex_lock(&connection->resource->conf_update);
3441 old_net_conf = connection->net_conf;
3442 *new_net_conf = *old_net_conf;
3444 new_net_conf->wire_protocol = p_proto;
3445 new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3446 new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3447 new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3448 new_net_conf->two_primaries = p_two_primaries;
3450 rcu_assign_pointer(connection->net_conf, new_net_conf);
3451 mutex_unlock(&connection->resource->conf_update);
3452 mutex_unlock(&connection->data.mutex);
3454 crypto_free_ahash(connection->peer_integrity_tfm);
3455 kfree(connection->int_dig_in);
3456 kfree(connection->int_dig_vv);
3457 connection->peer_integrity_tfm = peer_integrity_tfm;
3458 connection->int_dig_in = int_dig_in;
3459 connection->int_dig_vv = int_dig_vv;
3461 if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3462 drbd_info(connection, "peer data-integrity-alg: %s\n",
3463 integrity_alg[0] ? integrity_alg : "(none)");
3466 kfree(old_net_conf);
3469 disconnect_rcu_unlock:
3472 crypto_free_ahash(peer_integrity_tfm);
3475 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3480 * input: alg name, feature name
3481 * return: NULL (alg name was "")
3482 * ERR_PTR(error) if something goes wrong
3483 * or the crypto hash ptr, if it worked out ok. */
3484 static struct crypto_ahash *drbd_crypto_alloc_digest_safe(const struct drbd_device *device,
3485 const char *alg, const char *name)
3487 struct crypto_ahash *tfm;
3492 tfm = crypto_alloc_ahash(alg, 0, CRYPTO_ALG_ASYNC);
3494 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3495 alg, name, PTR_ERR(tfm));
3501 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3503 void *buffer = connection->data.rbuf;
3504 int size = pi->size;
3507 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3508 s = drbd_recv(connection, buffer, s);
3522 * config_unknown_volume - device configuration command for unknown volume
3524 * When a device is added to an existing connection, the node on which the
3525 * device is added first will send configuration commands to its peer but the
3526 * peer will not know about the device yet. It will warn and ignore these
3527 * commands. Once the device is added on the second node, the second node will
3528 * send the same device configuration commands, but in the other direction.
3530 * (We can also end up here if drbd is misconfigured.)
3532 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3534 drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3535 cmdname(pi->cmd), pi->vnr);
3536 return ignore_remaining_packet(connection, pi);
3539 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3541 struct drbd_peer_device *peer_device;
3542 struct drbd_device *device;
3543 struct p_rs_param_95 *p;
3544 unsigned int header_size, data_size, exp_max_sz;
3545 struct crypto_ahash *verify_tfm = NULL;
3546 struct crypto_ahash *csums_tfm = NULL;
3547 struct net_conf *old_net_conf, *new_net_conf = NULL;
3548 struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3549 const int apv = connection->agreed_pro_version;
3550 struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3554 peer_device = conn_peer_device(connection, pi->vnr);
3556 return config_unknown_volume(connection, pi);
3557 device = peer_device->device;
3559 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
3560 : apv == 88 ? sizeof(struct p_rs_param)
3562 : apv <= 94 ? sizeof(struct p_rs_param_89)
3563 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3565 if (pi->size > exp_max_sz) {
3566 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3567 pi->size, exp_max_sz);
3572 header_size = sizeof(struct p_rs_param);
3573 data_size = pi->size - header_size;
3574 } else if (apv <= 94) {
3575 header_size = sizeof(struct p_rs_param_89);
3576 data_size = pi->size - header_size;
3577 D_ASSERT(device, data_size == 0);
3579 header_size = sizeof(struct p_rs_param_95);
3580 data_size = pi->size - header_size;
3581 D_ASSERT(device, data_size == 0);
3584 /* initialize verify_alg and csums_alg */
3586 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
3588 err = drbd_recv_all(peer_device->connection, p, header_size);
3592 mutex_lock(&connection->resource->conf_update);
3593 old_net_conf = peer_device->connection->net_conf;
3594 if (get_ldev(device)) {
3595 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3596 if (!new_disk_conf) {
3598 mutex_unlock(&connection->resource->conf_update);
3599 drbd_err(device, "Allocation of new disk_conf failed\n");
3603 old_disk_conf = device->ldev->disk_conf;
3604 *new_disk_conf = *old_disk_conf;
3606 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3611 if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3612 drbd_err(device, "verify-alg of wrong size, "
3613 "peer wants %u, accepting only up to %u byte\n",
3614 data_size, SHARED_SECRET_MAX);
3619 err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3622 /* we expect NUL terminated string */
3623 /* but just in case someone tries to be evil */
3624 D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3625 p->verify_alg[data_size-1] = 0;
3627 } else /* apv >= 89 */ {
3628 /* we still expect NUL terminated strings */
3629 /* but just in case someone tries to be evil */
3630 D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3631 D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3632 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3633 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3636 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3637 if (device->state.conn == C_WF_REPORT_PARAMS) {
3638 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3639 old_net_conf->verify_alg, p->verify_alg);
3642 verify_tfm = drbd_crypto_alloc_digest_safe(device,
3643 p->verify_alg, "verify-alg");
3644 if (IS_ERR(verify_tfm)) {
3650 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3651 if (device->state.conn == C_WF_REPORT_PARAMS) {
3652 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3653 old_net_conf->csums_alg, p->csums_alg);
3656 csums_tfm = drbd_crypto_alloc_digest_safe(device,
3657 p->csums_alg, "csums-alg");
3658 if (IS_ERR(csums_tfm)) {
3664 if (apv > 94 && new_disk_conf) {
3665 new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3666 new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3667 new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3668 new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3670 fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3671 if (fifo_size != device->rs_plan_s->size) {
3672 new_plan = fifo_alloc(fifo_size);
3674 drbd_err(device, "kmalloc of fifo_buffer failed");
3681 if (verify_tfm || csums_tfm) {
3682 new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3683 if (!new_net_conf) {
3684 drbd_err(device, "Allocation of new net_conf failed\n");
3688 *new_net_conf = *old_net_conf;
3691 strcpy(new_net_conf->verify_alg, p->verify_alg);
3692 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3693 crypto_free_ahash(peer_device->connection->verify_tfm);
3694 peer_device->connection->verify_tfm = verify_tfm;
3695 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3698 strcpy(new_net_conf->csums_alg, p->csums_alg);
3699 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3700 crypto_free_ahash(peer_device->connection->csums_tfm);
3701 peer_device->connection->csums_tfm = csums_tfm;
3702 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
3704 rcu_assign_pointer(connection->net_conf, new_net_conf);
3708 if (new_disk_conf) {
3709 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3714 old_plan = device->rs_plan_s;
3715 rcu_assign_pointer(device->rs_plan_s, new_plan);
3718 mutex_unlock(&connection->resource->conf_update);
3721 kfree(old_net_conf);
3722 kfree(old_disk_conf);
3728 if (new_disk_conf) {
3730 kfree(new_disk_conf);
3732 mutex_unlock(&connection->resource->conf_update);
3737 if (new_disk_conf) {
3739 kfree(new_disk_conf);
3741 mutex_unlock(&connection->resource->conf_update);
3742 /* just for completeness: actually not needed,
3743 * as this is not reached if csums_tfm was ok. */
3744 crypto_free_ahash(csums_tfm);
3745 /* but free the verify_tfm again, if csums_tfm did not work out */
3746 crypto_free_ahash(verify_tfm);
3747 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3751 /* warn if the arguments differ by more than 12.5% */
3752 static void warn_if_differ_considerably(struct drbd_device *device,
3753 const char *s, sector_t a, sector_t b)
3756 if (a == 0 || b == 0)
3758 d = (a > b) ? (a - b) : (b - a);
3759 if (d > (a>>3) || d > (b>>3))
3760 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
3761 (unsigned long long)a, (unsigned long long)b);
3764 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
3766 struct drbd_peer_device *peer_device;
3767 struct drbd_device *device;
3768 struct p_sizes *p = pi->data;
3769 enum determine_dev_size dd = DS_UNCHANGED;
3770 sector_t p_size, p_usize, p_csize, my_usize;
3771 int ldsc = 0; /* local disk size changed */
3772 enum dds_flags ddsf;
3774 peer_device = conn_peer_device(connection, pi->vnr);
3776 return config_unknown_volume(connection, pi);
3777 device = peer_device->device;
3779 p_size = be64_to_cpu(p->d_size);
3780 p_usize = be64_to_cpu(p->u_size);
3781 p_csize = be64_to_cpu(p->c_size);
3783 /* just store the peer's disk size for now.
3784 * we still need to figure out whether we accept that. */
3785 device->p_size = p_size;
3787 if (get_ldev(device)) {
3789 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
3792 warn_if_differ_considerably(device, "lower level device sizes",
3793 p_size, drbd_get_max_capacity(device->ldev));
3794 warn_if_differ_considerably(device, "user requested size",
3797 /* if this is the first connect, or an otherwise expected
3798 * param exchange, choose the minimum */
3799 if (device->state.conn == C_WF_REPORT_PARAMS)
3800 p_usize = min_not_zero(my_usize, p_usize);
3802 /* Never shrink a device with usable data during connect.
3803 But allow online shrinking if we are connected. */
3804 if (drbd_new_dev_size(device, device->ldev, p_usize, 0) <
3805 drbd_get_capacity(device->this_bdev) &&
3806 device->state.disk >= D_OUTDATED &&
3807 device->state.conn < C_CONNECTED) {
3808 drbd_err(device, "The peer's disk size is too small!\n");
3809 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3814 if (my_usize != p_usize) {
3815 struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
3817 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3818 if (!new_disk_conf) {
3819 drbd_err(device, "Allocation of new disk_conf failed\n");
3824 mutex_lock(&connection->resource->conf_update);
3825 old_disk_conf = device->ldev->disk_conf;
3826 *new_disk_conf = *old_disk_conf;
3827 new_disk_conf->disk_size = p_usize;
3829 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3830 mutex_unlock(&connection->resource->conf_update);
3832 kfree(old_disk_conf);
3834 drbd_info(device, "Peer sets u_size to %lu sectors\n",
3835 (unsigned long)my_usize);
3841 device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
3842 /* Leave drbd_reconsider_max_bio_size() before drbd_determine_dev_size().
3843 In case we cleared the QUEUE_FLAG_DISCARD from our queue in
3844 drbd_reconsider_max_bio_size(), we can be sure that after
3845 drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
3847 ddsf = be16_to_cpu(p->dds_flags);
3848 if (get_ldev(device)) {
3849 drbd_reconsider_max_bio_size(device, device->ldev);
3850 dd = drbd_determine_dev_size(device, ddsf, NULL);
3854 drbd_md_sync(device);
3857 * I am diskless, need to accept the peer's *current* size.
3858 * I must NOT accept the peers backing disk size,
3859 * it may have been larger than mine all along...
3861 * At this point, the peer knows more about my disk, or at
3862 * least about what we last agreed upon, than myself.
3863 * So if his c_size is less than his d_size, the most likely
3864 * reason is that *my* d_size was smaller last time we checked.
3866 * However, if he sends a zero current size,
3867 * take his (user-capped or) backing disk size anyways.
3869 drbd_reconsider_max_bio_size(device, NULL);
3870 drbd_set_my_capacity(device, p_csize ?: p_usize ?: p_size);
3873 if (get_ldev(device)) {
3874 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
3875 device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
3882 if (device->state.conn > C_WF_REPORT_PARAMS) {
3883 if (be64_to_cpu(p->c_size) !=
3884 drbd_get_capacity(device->this_bdev) || ldsc) {
3885 /* we have different sizes, probably peer
3886 * needs to know my new size... */
3887 drbd_send_sizes(peer_device, 0, ddsf);
3889 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
3890 (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
3891 if (device->state.pdsk >= D_INCONSISTENT &&
3892 device->state.disk >= D_INCONSISTENT) {
3893 if (ddsf & DDSF_NO_RESYNC)
3894 drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
3896 resync_after_online_grow(device);
3898 set_bit(RESYNC_AFTER_NEG, &device->flags);
3905 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
3907 struct drbd_peer_device *peer_device;
3908 struct drbd_device *device;
3909 struct p_uuids *p = pi->data;
3911 int i, updated_uuids = 0;
3913 peer_device = conn_peer_device(connection, pi->vnr);
3915 return config_unknown_volume(connection, pi);
3916 device = peer_device->device;
3918 p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
3920 drbd_err(device, "kmalloc of p_uuid failed\n");
3924 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
3925 p_uuid[i] = be64_to_cpu(p->uuid[i]);
3927 kfree(device->p_uuid);
3928 device->p_uuid = p_uuid;
3930 if (device->state.conn < C_CONNECTED &&
3931 device->state.disk < D_INCONSISTENT &&
3932 device->state.role == R_PRIMARY &&
3933 (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
3934 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
3935 (unsigned long long)device->ed_uuid);
3936 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3940 if (get_ldev(device)) {
3941 int skip_initial_sync =
3942 device->state.conn == C_CONNECTED &&
3943 peer_device->connection->agreed_pro_version >= 90 &&
3944 device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
3945 (p_uuid[UI_FLAGS] & 8);
3946 if (skip_initial_sync) {
3947 drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
3948 drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
3949 "clear_n_write from receive_uuids",
3950 BM_LOCKED_TEST_ALLOWED);
3951 _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
3952 _drbd_uuid_set(device, UI_BITMAP, 0);
3953 _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
3955 drbd_md_sync(device);
3959 } else if (device->state.disk < D_INCONSISTENT &&
3960 device->state.role == R_PRIMARY) {
3961 /* I am a diskless primary, the peer just created a new current UUID
3963 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
3966 /* Before we test for the disk state, we should wait until an eventually
3967 ongoing cluster wide state change is finished. That is important if
3968 we are primary and are detaching from our disk. We need to see the
3969 new disk state... */
3970 mutex_lock(device->state_mutex);
3971 mutex_unlock(device->state_mutex);
3972 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
3973 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
3976 drbd_print_uuids(device, "receiver updated UUIDs to");
3982 * convert_state() - Converts the peer's view of the cluster state to our point of view
3983 * @ps: The state as seen by the peer.
3985 static union drbd_state convert_state(union drbd_state ps)
3987 union drbd_state ms;
3989 static enum drbd_conns c_tab[] = {
3990 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
3991 [C_CONNECTED] = C_CONNECTED,
3993 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
3994 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
3995 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
3996 [C_VERIFY_S] = C_VERIFY_T,
4002 ms.conn = c_tab[ps.conn];
4007 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
4012 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4014 struct drbd_peer_device *peer_device;
4015 struct drbd_device *device;
4016 struct p_req_state *p = pi->data;
4017 union drbd_state mask, val;
4018 enum drbd_state_rv rv;
4020 peer_device = conn_peer_device(connection, pi->vnr);
4023 device = peer_device->device;
4025 mask.i = be32_to_cpu(p->mask);
4026 val.i = be32_to_cpu(p->val);
4028 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4029 mutex_is_locked(device->state_mutex)) {
4030 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4034 mask = convert_state(mask);
4035 val = convert_state(val);
4037 rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4038 drbd_send_sr_reply(peer_device, rv);
4040 drbd_md_sync(device);
4045 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4047 struct p_req_state *p = pi->data;
4048 union drbd_state mask, val;
4049 enum drbd_state_rv rv;
4051 mask.i = be32_to_cpu(p->mask);
4052 val.i = be32_to_cpu(p->val);
4054 if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4055 mutex_is_locked(&connection->cstate_mutex)) {
4056 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4060 mask = convert_state(mask);
4061 val = convert_state(val);
4063 rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4064 conn_send_sr_reply(connection, rv);
4069 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4071 struct drbd_peer_device *peer_device;
4072 struct drbd_device *device;
4073 struct p_state *p = pi->data;
4074 union drbd_state os, ns, peer_state;
4075 enum drbd_disk_state real_peer_disk;
4076 enum chg_state_flags cs_flags;
4079 peer_device = conn_peer_device(connection, pi->vnr);
4081 return config_unknown_volume(connection, pi);
4082 device = peer_device->device;
4084 peer_state.i = be32_to_cpu(p->state);
4086 real_peer_disk = peer_state.disk;
4087 if (peer_state.disk == D_NEGOTIATING) {
4088 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4089 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4092 spin_lock_irq(&device->resource->req_lock);
4094 os = ns = drbd_read_state(device);
4095 spin_unlock_irq(&device->resource->req_lock);
4097 /* If some other part of the code (ack_receiver thread, timeout)
4098 * already decided to close the connection again,
4099 * we must not "re-establish" it here. */
4100 if (os.conn <= C_TEAR_DOWN)
4103 /* If this is the "end of sync" confirmation, usually the peer disk
4104 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4105 * set) resync started in PausedSyncT, or if the timing of pause-/
4106 * unpause-sync events has been "just right", the peer disk may
4107 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4109 if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4110 real_peer_disk == D_UP_TO_DATE &&
4111 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4112 /* If we are (becoming) SyncSource, but peer is still in sync
4113 * preparation, ignore its uptodate-ness to avoid flapping, it
4114 * will change to inconsistent once the peer reaches active
4116 * It may have changed syncer-paused flags, however, so we
4117 * cannot ignore this completely. */
4118 if (peer_state.conn > C_CONNECTED &&
4119 peer_state.conn < C_SYNC_SOURCE)
4120 real_peer_disk = D_INCONSISTENT;
4122 /* if peer_state changes to connected at the same time,
4123 * it explicitly notifies us that it finished resync.
4124 * Maybe we should finish it up, too? */
4125 else if (os.conn >= C_SYNC_SOURCE &&
4126 peer_state.conn == C_CONNECTED) {
4127 if (drbd_bm_total_weight(device) <= device->rs_failed)
4128 drbd_resync_finished(device);
4133 /* explicit verify finished notification, stop sector reached. */
4134 if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4135 peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4136 ov_out_of_sync_print(device);
4137 drbd_resync_finished(device);
4141 /* peer says his disk is inconsistent, while we think it is uptodate,
4142 * and this happens while the peer still thinks we have a sync going on,
4143 * but we think we are already done with the sync.
4144 * We ignore this to avoid flapping pdsk.
4145 * This should not happen, if the peer is a recent version of drbd. */
4146 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4147 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4148 real_peer_disk = D_UP_TO_DATE;
4150 if (ns.conn == C_WF_REPORT_PARAMS)
4151 ns.conn = C_CONNECTED;
4153 if (peer_state.conn == C_AHEAD)
4156 if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4157 get_ldev_if_state(device, D_NEGOTIATING)) {
4158 int cr; /* consider resync */
4160 /* if we established a new connection */
4161 cr = (os.conn < C_CONNECTED);
4162 /* if we had an established connection
4163 * and one of the nodes newly attaches a disk */
4164 cr |= (os.conn == C_CONNECTED &&
4165 (peer_state.disk == D_NEGOTIATING ||
4166 os.disk == D_NEGOTIATING));
4167 /* if we have both been inconsistent, and the peer has been
4168 * forced to be UpToDate with --overwrite-data */
4169 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4170 /* if we had been plain connected, and the admin requested to
4171 * start a sync by "invalidate" or "invalidate-remote" */
4172 cr |= (os.conn == C_CONNECTED &&
4173 (peer_state.conn >= C_STARTING_SYNC_S &&
4174 peer_state.conn <= C_WF_BITMAP_T));
4177 ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4180 if (ns.conn == C_MASK) {
4181 ns.conn = C_CONNECTED;
4182 if (device->state.disk == D_NEGOTIATING) {
4183 drbd_force_state(device, NS(disk, D_FAILED));
4184 } else if (peer_state.disk == D_NEGOTIATING) {
4185 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4186 peer_state.disk = D_DISKLESS;
4187 real_peer_disk = D_DISKLESS;
4189 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4191 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4192 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4198 spin_lock_irq(&device->resource->req_lock);
4199 if (os.i != drbd_read_state(device).i)
4201 clear_bit(CONSIDER_RESYNC, &device->flags);
4202 ns.peer = peer_state.role;
4203 ns.pdsk = real_peer_disk;
4204 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4205 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4206 ns.disk = device->new_state_tmp.disk;
4207 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4208 if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4209 test_bit(NEW_CUR_UUID, &device->flags)) {
4210 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4211 for temporal network outages! */
4212 spin_unlock_irq(&device->resource->req_lock);
4213 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4214 tl_clear(peer_device->connection);
4215 drbd_uuid_new_current(device);
4216 clear_bit(NEW_CUR_UUID, &device->flags);
4217 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4220 rv = _drbd_set_state(device, ns, cs_flags, NULL);
4221 ns = drbd_read_state(device);
4222 spin_unlock_irq(&device->resource->req_lock);
4224 if (rv < SS_SUCCESS) {
4225 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4229 if (os.conn > C_WF_REPORT_PARAMS) {
4230 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4231 peer_state.disk != D_NEGOTIATING ) {
4232 /* we want resync, peer has not yet decided to sync... */
4233 /* Nowadays only used when forcing a node into primary role and
4234 setting its disk to UpToDate with that */
4235 drbd_send_uuids(peer_device);
4236 drbd_send_current_state(peer_device);
4240 clear_bit(DISCARD_MY_DATA, &device->flags);
4242 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4247 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4249 struct drbd_peer_device *peer_device;
4250 struct drbd_device *device;
4251 struct p_rs_uuid *p = pi->data;
4253 peer_device = conn_peer_device(connection, pi->vnr);
4256 device = peer_device->device;
4258 wait_event(device->misc_wait,
4259 device->state.conn == C_WF_SYNC_UUID ||
4260 device->state.conn == C_BEHIND ||
4261 device->state.conn < C_CONNECTED ||
4262 device->state.disk < D_NEGOTIATING);
4264 /* D_ASSERT(device, device->state.conn == C_WF_SYNC_UUID ); */
4266 /* Here the _drbd_uuid_ functions are right, current should
4267 _not_ be rotated into the history */
4268 if (get_ldev_if_state(device, D_NEGOTIATING)) {
4269 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4270 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4272 drbd_print_uuids(device, "updated sync uuid");
4273 drbd_start_resync(device, C_SYNC_TARGET);
4277 drbd_err(device, "Ignoring SyncUUID packet!\n");
4283 * receive_bitmap_plain
4285 * Return 0 when done, 1 when another iteration is needed, and a negative error
4286 * code upon failure.
4289 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4290 unsigned long *p, struct bm_xfer_ctx *c)
4292 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4293 drbd_header_size(peer_device->connection);
4294 unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4295 c->bm_words - c->word_offset);
4296 unsigned int want = num_words * sizeof(*p);
4300 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4305 err = drbd_recv_all(peer_device->connection, p, want);
4309 drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4311 c->word_offset += num_words;
4312 c->bit_offset = c->word_offset * BITS_PER_LONG;
4313 if (c->bit_offset > c->bm_bits)
4314 c->bit_offset = c->bm_bits;
4319 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4321 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4324 static int dcbp_get_start(struct p_compressed_bm *p)
4326 return (p->encoding & 0x80) != 0;
4329 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4331 return (p->encoding >> 4) & 0x7;
4337 * Return 0 when done, 1 when another iteration is needed, and a negative error
4338 * code upon failure.
4341 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4342 struct p_compressed_bm *p,
4343 struct bm_xfer_ctx *c,
4346 struct bitstream bs;
4350 unsigned long s = c->bit_offset;
4352 int toggle = dcbp_get_start(p);
4356 bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4358 bits = bitstream_get_bits(&bs, &look_ahead, 64);
4362 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4363 bits = vli_decode_bits(&rl, look_ahead);
4369 if (e >= c->bm_bits) {
4370 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4373 _drbd_bm_set_bits(peer_device->device, s, e);
4377 drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4378 have, bits, look_ahead,
4379 (unsigned int)(bs.cur.b - p->code),
4380 (unsigned int)bs.buf_len);
4383 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4384 if (likely(bits < 64))
4385 look_ahead >>= bits;
4390 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4393 look_ahead |= tmp << have;
4398 bm_xfer_ctx_bit_to_word_offset(c);
4400 return (s != c->bm_bits);
4406 * Return 0 when done, 1 when another iteration is needed, and a negative error
4407 * code upon failure.
4410 decode_bitmap_c(struct drbd_peer_device *peer_device,
4411 struct p_compressed_bm *p,
4412 struct bm_xfer_ctx *c,
4415 if (dcbp_get_code(p) == RLE_VLI_Bits)
4416 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4418 /* other variants had been implemented for evaluation,
4419 * but have been dropped as this one turned out to be "best"
4420 * during all our tests. */
4422 drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4423 conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4427 void INFO_bm_xfer_stats(struct drbd_device *device,
4428 const char *direction, struct bm_xfer_ctx *c)
4430 /* what would it take to transfer it "plaintext" */
4431 unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4432 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4433 unsigned int plain =
4434 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4435 c->bm_words * sizeof(unsigned long);
4436 unsigned int total = c->bytes[0] + c->bytes[1];
4439 /* total can not be zero. but just in case: */
4443 /* don't report if not compressed */
4447 /* total < plain. check for overflow, still */
4448 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4449 : (1000 * total / plain);
4455 drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4456 "total %u; compression: %u.%u%%\n",
4458 c->bytes[1], c->packets[1],
4459 c->bytes[0], c->packets[0],
4460 total, r/10, r % 10);
4463 /* Since we are processing the bitfield from lower addresses to higher,
4464 it does not matter if the process it in 32 bit chunks or 64 bit
4465 chunks as long as it is little endian. (Understand it as byte stream,
4466 beginning with the lowest byte...) If we would use big endian
4467 we would need to process it from the highest address to the lowest,
4468 in order to be agnostic to the 32 vs 64 bits issue.
4470 returns 0 on failure, 1 if we successfully received it. */
4471 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4473 struct drbd_peer_device *peer_device;
4474 struct drbd_device *device;
4475 struct bm_xfer_ctx c;
4478 peer_device = conn_peer_device(connection, pi->vnr);
4481 device = peer_device->device;
4483 drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4484 /* you are supposed to send additional out-of-sync information
4485 * if you actually set bits during this phase */
4487 c = (struct bm_xfer_ctx) {
4488 .bm_bits = drbd_bm_bits(device),
4489 .bm_words = drbd_bm_words(device),
4493 if (pi->cmd == P_BITMAP)
4494 err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4495 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4496 /* MAYBE: sanity check that we speak proto >= 90,
4497 * and the feature is enabled! */
4498 struct p_compressed_bm *p = pi->data;
4500 if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4501 drbd_err(device, "ReportCBitmap packet too large\n");
4505 if (pi->size <= sizeof(*p)) {
4506 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4510 err = drbd_recv_all(peer_device->connection, p, pi->size);
4513 err = decode_bitmap_c(peer_device, p, &c, pi->size);
4515 drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4520 c.packets[pi->cmd == P_BITMAP]++;
4521 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4528 err = drbd_recv_header(peer_device->connection, pi);
4533 INFO_bm_xfer_stats(device, "receive", &c);
4535 if (device->state.conn == C_WF_BITMAP_T) {
4536 enum drbd_state_rv rv;
4538 err = drbd_send_bitmap(device);
4541 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4542 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4543 D_ASSERT(device, rv == SS_SUCCESS);
4544 } else if (device->state.conn != C_WF_BITMAP_S) {
4545 /* admin may have requested C_DISCONNECTING,
4546 * other threads may have noticed network errors */
4547 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4548 drbd_conn_str(device->state.conn));
4553 drbd_bm_unlock(device);
4554 if (!err && device->state.conn == C_WF_BITMAP_S)
4555 drbd_start_resync(device, C_SYNC_SOURCE);
4559 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4561 drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4564 return ignore_remaining_packet(connection, pi);
4567 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4569 /* Make sure we've acked all the TCP data associated
4570 * with the data requests being unplugged */
4571 drbd_tcp_quickack(connection->data.socket);
4576 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4578 struct drbd_peer_device *peer_device;
4579 struct drbd_device *device;
4580 struct p_block_desc *p = pi->data;
4582 peer_device = conn_peer_device(connection, pi->vnr);
4585 device = peer_device->device;
4587 switch (device->state.conn) {
4588 case C_WF_SYNC_UUID:
4593 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4594 drbd_conn_str(device->state.conn));
4597 drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4605 int (*fn)(struct drbd_connection *, struct packet_info *);
4608 static struct data_cmd drbd_cmd_handler[] = {
4609 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
4610 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
4611 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
4612 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
4613 [P_BITMAP] = { 1, 0, receive_bitmap } ,
4614 [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
4615 [P_UNPLUG_REMOTE] = { 0, 0, receive_UnplugRemote },
4616 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4617 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4618 [P_SYNC_PARAM] = { 1, 0, receive_SyncParam },
4619 [P_SYNC_PARAM89] = { 1, 0, receive_SyncParam },
4620 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
4621 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
4622 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
4623 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
4624 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
4625 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
4626 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4627 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4628 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4629 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
4630 [P_OUT_OF_SYNC] = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
4631 [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
4632 [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
4633 [P_TRIM] = { 0, sizeof(struct p_trim), receive_Data },
4636 static void drbdd(struct drbd_connection *connection)
4638 struct packet_info pi;
4639 size_t shs; /* sub header size */
4642 while (get_t_state(&connection->receiver) == RUNNING) {
4643 struct data_cmd *cmd;
4645 drbd_thread_current_set_cpu(&connection->receiver);
4646 update_receiver_timing_details(connection, drbd_recv_header);
4647 if (drbd_recv_header(connection, &pi))
4650 cmd = &drbd_cmd_handler[pi.cmd];
4651 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
4652 drbd_err(connection, "Unexpected data packet %s (0x%04x)",
4653 cmdname(pi.cmd), pi.cmd);
4657 shs = cmd->pkt_size;
4658 if (pi.size > shs && !cmd->expect_payload) {
4659 drbd_err(connection, "No payload expected %s l:%d\n",
4660 cmdname(pi.cmd), pi.size);
4665 update_receiver_timing_details(connection, drbd_recv_all_warn);
4666 err = drbd_recv_all_warn(connection, pi.data, shs);
4672 update_receiver_timing_details(connection, cmd->fn);
4673 err = cmd->fn(connection, &pi);
4675 drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
4676 cmdname(pi.cmd), err, pi.size);
4683 conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4686 static void conn_disconnect(struct drbd_connection *connection)
4688 struct drbd_peer_device *peer_device;
4692 if (connection->cstate == C_STANDALONE)
4695 /* We are about to start the cleanup after connection loss.
4696 * Make sure drbd_make_request knows about that.
4697 * Usually we should be in some network failure state already,
4698 * but just in case we are not, we fix it up here.
4700 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
4702 /* ack_receiver does not clean up anything. it must not interfere, either */
4703 drbd_thread_stop(&connection->ack_receiver);
4704 if (connection->ack_sender) {
4705 destroy_workqueue(connection->ack_sender);
4706 connection->ack_sender = NULL;
4708 drbd_free_sock(connection);
4711 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
4712 struct drbd_device *device = peer_device->device;
4713 kref_get(&device->kref);
4715 drbd_disconnected(peer_device);
4716 kref_put(&device->kref, drbd_destroy_device);
4721 if (!list_empty(&connection->current_epoch->list))
4722 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
4723 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
4724 atomic_set(&connection->current_epoch->epoch_size, 0);
4725 connection->send.seen_any_write_yet = false;
4727 drbd_info(connection, "Connection closed\n");
4729 if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
4730 conn_try_outdate_peer_async(connection);
4732 spin_lock_irq(&connection->resource->req_lock);
4733 oc = connection->cstate;
4734 if (oc >= C_UNCONNECTED)
4735 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
4737 spin_unlock_irq(&connection->resource->req_lock);
4739 if (oc == C_DISCONNECTING)
4740 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
4743 static int drbd_disconnected(struct drbd_peer_device *peer_device)
4745 struct drbd_device *device = peer_device->device;
4748 /* wait for current activity to cease. */
4749 spin_lock_irq(&device->resource->req_lock);
4750 _drbd_wait_ee_list_empty(device, &device->active_ee);
4751 _drbd_wait_ee_list_empty(device, &device->sync_ee);
4752 _drbd_wait_ee_list_empty(device, &device->read_ee);
4753 spin_unlock_irq(&device->resource->req_lock);
4755 /* We do not have data structures that would allow us to
4756 * get the rs_pending_cnt down to 0 again.
4757 * * On C_SYNC_TARGET we do not have any data structures describing
4758 * the pending RSDataRequest's we have sent.
4759 * * On C_SYNC_SOURCE there is no data structure that tracks
4760 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
4761 * And no, it is not the sum of the reference counts in the
4762 * resync_LRU. The resync_LRU tracks the whole operation including
4763 * the disk-IO, while the rs_pending_cnt only tracks the blocks
4765 drbd_rs_cancel_all(device);
4766 device->rs_total = 0;
4767 device->rs_failed = 0;
4768 atomic_set(&device->rs_pending_cnt, 0);
4769 wake_up(&device->misc_wait);
4771 del_timer_sync(&device->resync_timer);
4772 resync_timer_fn((unsigned long)device);
4774 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
4775 * w_make_resync_request etc. which may still be on the worker queue
4776 * to be "canceled" */
4777 drbd_flush_workqueue(&peer_device->connection->sender_work);
4779 drbd_finish_peer_reqs(device);
4781 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
4782 might have issued a work again. The one before drbd_finish_peer_reqs() is
4783 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
4784 drbd_flush_workqueue(&peer_device->connection->sender_work);
4786 /* need to do it again, drbd_finish_peer_reqs() may have populated it
4787 * again via drbd_try_clear_on_disk_bm(). */
4788 drbd_rs_cancel_all(device);
4790 kfree(device->p_uuid);
4791 device->p_uuid = NULL;
4793 if (!drbd_suspended(device))
4794 tl_clear(peer_device->connection);
4796 drbd_md_sync(device);
4798 /* serialize with bitmap writeout triggered by the state change,
4800 wait_event(device->misc_wait, !test_bit(BITMAP_IO, &device->flags));
4802 /* tcp_close and release of sendpage pages can be deferred. I don't
4803 * want to use SO_LINGER, because apparently it can be deferred for
4804 * more than 20 seconds (longest time I checked).
4806 * Actually we don't care for exactly when the network stack does its
4807 * put_page(), but release our reference on these pages right here.
4809 i = drbd_free_peer_reqs(device, &device->net_ee);
4811 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
4812 i = atomic_read(&device->pp_in_use_by_net);
4814 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
4815 i = atomic_read(&device->pp_in_use);
4817 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
4819 D_ASSERT(device, list_empty(&device->read_ee));
4820 D_ASSERT(device, list_empty(&device->active_ee));
4821 D_ASSERT(device, list_empty(&device->sync_ee));
4822 D_ASSERT(device, list_empty(&device->done_ee));
4828 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
4829 * we can agree on is stored in agreed_pro_version.
4831 * feature flags and the reserved array should be enough room for future
4832 * enhancements of the handshake protocol, and possible plugins...
4834 * for now, they are expected to be zero, but ignored.
4836 static int drbd_send_features(struct drbd_connection *connection)
4838 struct drbd_socket *sock;
4839 struct p_connection_features *p;
4841 sock = &connection->data;
4842 p = conn_prepare_command(connection, sock);
4845 memset(p, 0, sizeof(*p));
4846 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
4847 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
4848 p->feature_flags = cpu_to_be32(PRO_FEATURES);
4849 return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
4854 * 1 yes, we have a valid connection
4855 * 0 oops, did not work out, please try again
4856 * -1 peer talks different language,
4857 * no point in trying again, please go standalone.
4859 static int drbd_do_features(struct drbd_connection *connection)
4861 /* ASSERT current == connection->receiver ... */
4862 struct p_connection_features *p;
4863 const int expect = sizeof(struct p_connection_features);
4864 struct packet_info pi;
4867 err = drbd_send_features(connection);
4871 err = drbd_recv_header(connection, &pi);
4875 if (pi.cmd != P_CONNECTION_FEATURES) {
4876 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
4877 cmdname(pi.cmd), pi.cmd);
4881 if (pi.size != expect) {
4882 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
4888 err = drbd_recv_all_warn(connection, p, expect);
4892 p->protocol_min = be32_to_cpu(p->protocol_min);
4893 p->protocol_max = be32_to_cpu(p->protocol_max);
4894 if (p->protocol_max == 0)
4895 p->protocol_max = p->protocol_min;
4897 if (PRO_VERSION_MAX < p->protocol_min ||
4898 PRO_VERSION_MIN > p->protocol_max)
4901 connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
4902 connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
4904 drbd_info(connection, "Handshake successful: "
4905 "Agreed network protocol version %d\n", connection->agreed_pro_version);
4907 drbd_info(connection, "Agreed to%ssupport TRIM on protocol level\n",
4908 connection->agreed_features & FF_TRIM ? " " : " not ");
4913 drbd_err(connection, "incompatible DRBD dialects: "
4914 "I support %d-%d, peer supports %d-%d\n",
4915 PRO_VERSION_MIN, PRO_VERSION_MAX,
4916 p->protocol_min, p->protocol_max);
4920 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4921 static int drbd_do_auth(struct drbd_connection *connection)
4923 drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4924 drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4928 #define CHALLENGE_LEN 64
4932 0 - failed, try again (network error),
4933 -1 - auth failed, don't try again.
4936 static int drbd_do_auth(struct drbd_connection *connection)
4938 struct drbd_socket *sock;
4939 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
4940 char *response = NULL;
4941 char *right_response = NULL;
4942 char *peers_ch = NULL;
4943 unsigned int key_len;
4944 char secret[SHARED_SECRET_MAX]; /* 64 byte */
4945 unsigned int resp_size;
4946 SHASH_DESC_ON_STACK(desc, connection->cram_hmac_tfm);
4947 struct packet_info pi;
4948 struct net_conf *nc;
4951 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
4954 nc = rcu_dereference(connection->net_conf);
4955 key_len = strlen(nc->shared_secret);
4956 memcpy(secret, nc->shared_secret, key_len);
4959 desc->tfm = connection->cram_hmac_tfm;
4962 rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
4964 drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
4969 get_random_bytes(my_challenge, CHALLENGE_LEN);
4971 sock = &connection->data;
4972 if (!conn_prepare_command(connection, sock)) {
4976 rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
4977 my_challenge, CHALLENGE_LEN);
4981 err = drbd_recv_header(connection, &pi);
4987 if (pi.cmd != P_AUTH_CHALLENGE) {
4988 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4989 cmdname(pi.cmd), pi.cmd);
4994 if (pi.size > CHALLENGE_LEN * 2) {
4995 drbd_err(connection, "expected AuthChallenge payload too big.\n");
5000 if (pi.size < CHALLENGE_LEN) {
5001 drbd_err(connection, "AuthChallenge payload too small.\n");
5006 peers_ch = kmalloc(pi.size, GFP_NOIO);
5007 if (peers_ch == NULL) {
5008 drbd_err(connection, "kmalloc of peers_ch failed\n");
5013 err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5019 if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5020 drbd_err(connection, "Peer presented the same challenge!\n");
5025 resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5026 response = kmalloc(resp_size, GFP_NOIO);
5027 if (response == NULL) {
5028 drbd_err(connection, "kmalloc of response failed\n");
5033 rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5035 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5040 if (!conn_prepare_command(connection, sock)) {
5044 rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5045 response, resp_size);
5049 err = drbd_recv_header(connection, &pi);
5055 if (pi.cmd != P_AUTH_RESPONSE) {
5056 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5057 cmdname(pi.cmd), pi.cmd);
5062 if (pi.size != resp_size) {
5063 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5068 err = drbd_recv_all_warn(connection, response , resp_size);
5074 right_response = kmalloc(resp_size, GFP_NOIO);
5075 if (right_response == NULL) {
5076 drbd_err(connection, "kmalloc of right_response failed\n");
5081 rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5084 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5089 rv = !memcmp(response, right_response, resp_size);
5092 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5100 kfree(right_response);
5101 shash_desc_zero(desc);
5107 int drbd_receiver(struct drbd_thread *thi)
5109 struct drbd_connection *connection = thi->connection;
5112 drbd_info(connection, "receiver (re)started\n");
5115 h = conn_connect(connection);
5117 conn_disconnect(connection);
5118 schedule_timeout_interruptible(HZ);
5121 drbd_warn(connection, "Discarding network configuration.\n");
5122 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5129 conn_disconnect(connection);
5131 drbd_info(connection, "receiver terminated\n");
5135 /* ********* acknowledge sender ******** */
5137 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5139 struct p_req_state_reply *p = pi->data;
5140 int retcode = be32_to_cpu(p->retcode);
5142 if (retcode >= SS_SUCCESS) {
5143 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5145 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5146 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5147 drbd_set_st_err_str(retcode), retcode);
5149 wake_up(&connection->ping_wait);
5154 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5156 struct drbd_peer_device *peer_device;
5157 struct drbd_device *device;
5158 struct p_req_state_reply *p = pi->data;
5159 int retcode = be32_to_cpu(p->retcode);
5161 peer_device = conn_peer_device(connection, pi->vnr);
5164 device = peer_device->device;
5166 if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5167 D_ASSERT(device, connection->agreed_pro_version < 100);
5168 return got_conn_RqSReply(connection, pi);
5171 if (retcode >= SS_SUCCESS) {
5172 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5174 set_bit(CL_ST_CHG_FAIL, &device->flags);
5175 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5176 drbd_set_st_err_str(retcode), retcode);
5178 wake_up(&device->state_wait);
5183 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5185 return drbd_send_ping_ack(connection);
5189 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5191 /* restore idle timeout */
5192 connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5193 if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5194 wake_up(&connection->ping_wait);
5199 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5201 struct drbd_peer_device *peer_device;
5202 struct drbd_device *device;
5203 struct p_block_ack *p = pi->data;
5204 sector_t sector = be64_to_cpu(p->sector);
5205 int blksize = be32_to_cpu(p->blksize);
5207 peer_device = conn_peer_device(connection, pi->vnr);
5210 device = peer_device->device;
5212 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5214 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5216 if (get_ldev(device)) {
5217 drbd_rs_complete_io(device, sector);
5218 drbd_set_in_sync(device, sector, blksize);
5219 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5220 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5223 dec_rs_pending(device);
5224 atomic_add(blksize >> 9, &device->rs_sect_in);
5230 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5231 struct rb_root *root, const char *func,
5232 enum drbd_req_event what, bool missing_ok)
5234 struct drbd_request *req;
5235 struct bio_and_error m;
5237 spin_lock_irq(&device->resource->req_lock);
5238 req = find_request(device, root, id, sector, missing_ok, func);
5239 if (unlikely(!req)) {
5240 spin_unlock_irq(&device->resource->req_lock);
5243 __req_mod(req, what, &m);
5244 spin_unlock_irq(&device->resource->req_lock);
5247 complete_master_bio(device, &m);
5251 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5253 struct drbd_peer_device *peer_device;
5254 struct drbd_device *device;
5255 struct p_block_ack *p = pi->data;
5256 sector_t sector = be64_to_cpu(p->sector);
5257 int blksize = be32_to_cpu(p->blksize);
5258 enum drbd_req_event what;
5260 peer_device = conn_peer_device(connection, pi->vnr);
5263 device = peer_device->device;
5265 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5267 if (p->block_id == ID_SYNCER) {
5268 drbd_set_in_sync(device, sector, blksize);
5269 dec_rs_pending(device);
5273 case P_RS_WRITE_ACK:
5274 what = WRITE_ACKED_BY_PEER_AND_SIS;
5277 what = WRITE_ACKED_BY_PEER;
5280 what = RECV_ACKED_BY_PEER;
5283 what = CONFLICT_RESOLVED;
5286 what = POSTPONE_WRITE;
5292 return validate_req_change_req_state(device, p->block_id, sector,
5293 &device->write_requests, __func__,
5297 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5299 struct drbd_peer_device *peer_device;
5300 struct drbd_device *device;
5301 struct p_block_ack *p = pi->data;
5302 sector_t sector = be64_to_cpu(p->sector);
5303 int size = be32_to_cpu(p->blksize);
5306 peer_device = conn_peer_device(connection, pi->vnr);
5309 device = peer_device->device;
5311 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5313 if (p->block_id == ID_SYNCER) {
5314 dec_rs_pending(device);
5315 drbd_rs_failed_io(device, sector, size);
5319 err = validate_req_change_req_state(device, p->block_id, sector,
5320 &device->write_requests, __func__,
5323 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5324 The master bio might already be completed, therefore the
5325 request is no longer in the collision hash. */
5326 /* In Protocol B we might already have got a P_RECV_ACK
5327 but then get a P_NEG_ACK afterwards. */
5328 drbd_set_out_of_sync(device, sector, size);
5333 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5335 struct drbd_peer_device *peer_device;
5336 struct drbd_device *device;
5337 struct p_block_ack *p = pi->data;
5338 sector_t sector = be64_to_cpu(p->sector);
5340 peer_device = conn_peer_device(connection, pi->vnr);
5343 device = peer_device->device;
5345 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5347 drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5348 (unsigned long long)sector, be32_to_cpu(p->blksize));
5350 return validate_req_change_req_state(device, p->block_id, sector,
5351 &device->read_requests, __func__,
5355 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5357 struct drbd_peer_device *peer_device;
5358 struct drbd_device *device;
5361 struct p_block_ack *p = pi->data;
5363 peer_device = conn_peer_device(connection, pi->vnr);
5366 device = peer_device->device;
5368 sector = be64_to_cpu(p->sector);
5369 size = be32_to_cpu(p->blksize);
5371 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5373 dec_rs_pending(device);
5375 if (get_ldev_if_state(device, D_FAILED)) {
5376 drbd_rs_complete_io(device, sector);
5378 case P_NEG_RS_DREPLY:
5379 drbd_rs_failed_io(device, sector, size);
5391 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5393 struct p_barrier_ack *p = pi->data;
5394 struct drbd_peer_device *peer_device;
5397 tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5400 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5401 struct drbd_device *device = peer_device->device;
5403 if (device->state.conn == C_AHEAD &&
5404 atomic_read(&device->ap_in_flight) == 0 &&
5405 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5406 device->start_resync_timer.expires = jiffies + HZ;
5407 add_timer(&device->start_resync_timer);
5415 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5417 struct drbd_peer_device *peer_device;
5418 struct drbd_device *device;
5419 struct p_block_ack *p = pi->data;
5420 struct drbd_device_work *dw;
5424 peer_device = conn_peer_device(connection, pi->vnr);
5427 device = peer_device->device;
5429 sector = be64_to_cpu(p->sector);
5430 size = be32_to_cpu(p->blksize);
5432 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5434 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5435 drbd_ov_out_of_sync_found(device, sector, size);
5437 ov_out_of_sync_print(device);
5439 if (!get_ldev(device))
5442 drbd_rs_complete_io(device, sector);
5443 dec_rs_pending(device);
5447 /* let's advance progress step marks only for every other megabyte */
5448 if ((device->ov_left & 0x200) == 0x200)
5449 drbd_advance_rs_marks(device, device->ov_left);
5451 if (device->ov_left == 0) {
5452 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5454 dw->w.cb = w_ov_finished;
5455 dw->device = device;
5456 drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5458 drbd_err(device, "kmalloc(dw) failed.");
5459 ov_out_of_sync_print(device);
5460 drbd_resync_finished(device);
5467 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5472 struct meta_sock_cmd {
5474 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5477 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5480 struct net_conf *nc;
5483 nc = rcu_dereference(connection->net_conf);
5484 t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5491 connection->meta.socket->sk->sk_rcvtimeo = t;
5494 static void set_ping_timeout(struct drbd_connection *connection)
5496 set_rcvtimeo(connection, 1);
5499 static void set_idle_timeout(struct drbd_connection *connection)
5501 set_rcvtimeo(connection, 0);
5504 static struct meta_sock_cmd ack_receiver_tbl[] = {
5505 [P_PING] = { 0, got_Ping },
5506 [P_PING_ACK] = { 0, got_PingAck },
5507 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5508 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5509 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5510 [P_SUPERSEDED] = { sizeof(struct p_block_ack), got_BlockAck },
5511 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
5512 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
5513 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply },
5514 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
5515 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
5516 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5517 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
5518 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
5519 [P_RS_CANCEL] = { sizeof(struct p_block_ack), got_NegRSDReply },
5520 [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5521 [P_RETRY_WRITE] = { sizeof(struct p_block_ack), got_BlockAck },
5524 int drbd_ack_receiver(struct drbd_thread *thi)
5526 struct drbd_connection *connection = thi->connection;
5527 struct meta_sock_cmd *cmd = NULL;
5528 struct packet_info pi;
5529 unsigned long pre_recv_jif;
5531 void *buf = connection->meta.rbuf;
5533 unsigned int header_size = drbd_header_size(connection);
5534 int expect = header_size;
5535 bool ping_timeout_active = false;
5536 struct sched_param param = { .sched_priority = 2 };
5538 rv = sched_setscheduler(current, SCHED_RR, ¶m);
5540 drbd_err(connection, "drbd_ack_receiver: ERROR set priority, ret=%d\n", rv);
5542 while (get_t_state(thi) == RUNNING) {
5543 drbd_thread_current_set_cpu(thi);
5545 conn_reclaim_net_peer_reqs(connection);
5547 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5548 if (drbd_send_ping(connection)) {
5549 drbd_err(connection, "drbd_send_ping has failed\n");
5552 set_ping_timeout(connection);
5553 ping_timeout_active = true;
5556 pre_recv_jif = jiffies;
5557 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5560 * -EINTR (on meta) we got a signal
5561 * -EAGAIN (on meta) rcvtimeo expired
5562 * -ECONNRESET other side closed the connection
5563 * -ERESTARTSYS (on data) we got a signal
5564 * rv < 0 other than above: unexpected error!
5565 * rv == expected: full header or command
5566 * rv < expected: "woken" by signal during receive
5567 * rv == 0 : "connection shut down by peer"
5569 if (likely(rv > 0)) {
5572 } else if (rv == 0) {
5573 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
5576 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
5579 t = wait_event_timeout(connection->ping_wait,
5580 connection->cstate < C_WF_REPORT_PARAMS,
5585 drbd_err(connection, "meta connection shut down by peer.\n");
5587 } else if (rv == -EAGAIN) {
5588 /* If the data socket received something meanwhile,
5589 * that is good enough: peer is still alive. */
5590 if (time_after(connection->last_received, pre_recv_jif))
5592 if (ping_timeout_active) {
5593 drbd_err(connection, "PingAck did not arrive in time.\n");
5596 set_bit(SEND_PING, &connection->flags);
5598 } else if (rv == -EINTR) {
5599 /* maybe drbd_thread_stop(): the while condition will notice.
5600 * maybe woken for send_ping: we'll send a ping above,
5601 * and change the rcvtimeo */
5602 flush_signals(current);
5605 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
5609 if (received == expect && cmd == NULL) {
5610 if (decode_header(connection, connection->meta.rbuf, &pi))
5612 cmd = &ack_receiver_tbl[pi.cmd];
5613 if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
5614 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
5615 cmdname(pi.cmd), pi.cmd);
5618 expect = header_size + cmd->pkt_size;
5619 if (pi.size != expect - header_size) {
5620 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
5625 if (received == expect) {
5628 err = cmd->fn(connection, &pi);
5630 drbd_err(connection, "%pf failed\n", cmd->fn);
5634 connection->last_received = jiffies;
5636 if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
5637 set_idle_timeout(connection);
5638 ping_timeout_active = false;
5641 buf = connection->meta.rbuf;
5643 expect = header_size;
5650 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5651 conn_md_sync(connection);
5655 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5658 drbd_info(connection, "ack_receiver terminated\n");
5663 void drbd_send_acks_wf(struct work_struct *ws)
5665 struct drbd_peer_device *peer_device =
5666 container_of(ws, struct drbd_peer_device, send_acks_work);
5667 struct drbd_connection *connection = peer_device->connection;
5668 struct drbd_device *device = peer_device->device;
5669 struct net_conf *nc;
5673 nc = rcu_dereference(connection->net_conf);
5674 tcp_cork = nc->tcp_cork;
5678 drbd_tcp_cork(connection->meta.socket);
5680 err = drbd_finish_peer_reqs(device);
5681 kref_put(&device->kref, drbd_destroy_device);
5682 /* get is in drbd_endio_write_sec_final(). That is necessary to keep the
5683 struct work_struct send_acks_work alive, which is in the peer_device object */
5686 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5691 drbd_tcp_uncork(connection->meta.socket);