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 <linux/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 <uapi/linux/sched/types.h>
40 #include <linux/sched/signal.h>
41 #include <linux/pkt_sched.h>
42 #define __KERNEL_SYSCALLS__
43 #include <linux/unistd.h>
44 #include <linux/vmalloc.h>
45 #include <linux/random.h>
46 #include <linux/string.h>
47 #include <linux/scatterlist.h>
49 #include "drbd_protocol.h"
53 #define PRO_FEATURES (DRBD_FF_TRIM|DRBD_FF_THIN_RESYNC|DRBD_FF_WSAME)
68 static int drbd_do_features(struct drbd_connection *connection);
69 static int drbd_do_auth(struct drbd_connection *connection);
70 static int drbd_disconnected(struct drbd_peer_device *);
71 static void conn_wait_active_ee_empty(struct drbd_connection *connection);
72 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
73 static int e_end_block(struct drbd_work *, int);
76 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
79 * some helper functions to deal with single linked page lists,
80 * page->private being our "next" pointer.
83 /* If at least n pages are linked at head, get n pages off.
84 * Otherwise, don't modify head, and return NULL.
85 * Locking is the responsibility of the caller.
87 static struct page *page_chain_del(struct page **head, int n)
101 tmp = page_chain_next(page);
103 break; /* found sufficient pages */
105 /* insufficient pages, don't use any of them. */
110 /* add end of list marker for the returned list */
111 set_page_private(page, 0);
112 /* actual return value, and adjustment of head */
118 /* may be used outside of locks to find the tail of a (usually short)
119 * "private" page chain, before adding it back to a global chain head
120 * with page_chain_add() under a spinlock. */
121 static struct page *page_chain_tail(struct page *page, int *len)
125 while ((tmp = page_chain_next(page)))
132 static int page_chain_free(struct page *page)
136 page_chain_for_each_safe(page, tmp) {
143 static void page_chain_add(struct page **head,
144 struct page *chain_first, struct page *chain_last)
148 tmp = page_chain_tail(chain_first, NULL);
149 BUG_ON(tmp != chain_last);
152 /* add chain to head */
153 set_page_private(chain_last, (unsigned long)*head);
157 static struct page *__drbd_alloc_pages(struct drbd_device *device,
160 struct page *page = NULL;
161 struct page *tmp = NULL;
164 /* Yes, testing drbd_pp_vacant outside the lock is racy.
165 * So what. It saves a spin_lock. */
166 if (drbd_pp_vacant >= number) {
167 spin_lock(&drbd_pp_lock);
168 page = page_chain_del(&drbd_pp_pool, number);
170 drbd_pp_vacant -= number;
171 spin_unlock(&drbd_pp_lock);
176 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
177 * "criss-cross" setup, that might cause write-out on some other DRBD,
178 * which in turn might block on the other node at this very place. */
179 for (i = 0; i < number; i++) {
180 tmp = alloc_page(GFP_TRY);
183 set_page_private(tmp, (unsigned long)page);
190 /* Not enough pages immediately available this time.
191 * No need to jump around here, drbd_alloc_pages will retry this
192 * function "soon". */
194 tmp = page_chain_tail(page, NULL);
195 spin_lock(&drbd_pp_lock);
196 page_chain_add(&drbd_pp_pool, page, tmp);
198 spin_unlock(&drbd_pp_lock);
203 static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
204 struct list_head *to_be_freed)
206 struct drbd_peer_request *peer_req, *tmp;
208 /* The EEs are always appended to the end of the list. Since
209 they are sent in order over the wire, they have to finish
210 in order. As soon as we see the first not finished we can
211 stop to examine the list... */
213 list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
214 if (drbd_peer_req_has_active_page(peer_req))
216 list_move(&peer_req->w.list, to_be_freed);
220 static void drbd_reclaim_net_peer_reqs(struct drbd_device *device)
222 LIST_HEAD(reclaimed);
223 struct drbd_peer_request *peer_req, *t;
225 spin_lock_irq(&device->resource->req_lock);
226 reclaim_finished_net_peer_reqs(device, &reclaimed);
227 spin_unlock_irq(&device->resource->req_lock);
228 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
229 drbd_free_net_peer_req(device, peer_req);
232 static void conn_reclaim_net_peer_reqs(struct drbd_connection *connection)
234 struct drbd_peer_device *peer_device;
238 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
239 struct drbd_device *device = peer_device->device;
240 if (!atomic_read(&device->pp_in_use_by_net))
243 kref_get(&device->kref);
245 drbd_reclaim_net_peer_reqs(device);
246 kref_put(&device->kref, drbd_destroy_device);
253 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
254 * @device: DRBD device.
255 * @number: number of pages requested
256 * @retry: whether to retry, if not enough pages are available right now
258 * Tries to allocate number pages, first from our own page pool, then from
260 * Possibly retry until DRBD frees sufficient pages somewhere else.
262 * If this allocation would exceed the max_buffers setting, we throttle
263 * allocation (schedule_timeout) to give the system some room to breathe.
265 * We do not use max-buffers as hard limit, because it could lead to
266 * congestion and further to a distributed deadlock during online-verify or
267 * (checksum based) resync, if the max-buffers, socket buffer sizes and
268 * resync-rate settings are mis-configured.
270 * Returns a page chain linked via page->private.
272 struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
275 struct drbd_device *device = peer_device->device;
276 struct page *page = NULL;
282 nc = rcu_dereference(peer_device->connection->net_conf);
283 mxb = nc ? nc->max_buffers : 1000000;
286 if (atomic_read(&device->pp_in_use) < mxb)
287 page = __drbd_alloc_pages(device, number);
289 /* Try to keep the fast path fast, but occasionally we need
290 * to reclaim the pages we lended to the network stack. */
291 if (page && atomic_read(&device->pp_in_use_by_net) > 512)
292 drbd_reclaim_net_peer_reqs(device);
294 while (page == NULL) {
295 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
297 drbd_reclaim_net_peer_reqs(device);
299 if (atomic_read(&device->pp_in_use) < mxb) {
300 page = __drbd_alloc_pages(device, number);
308 if (signal_pending(current)) {
309 drbd_warn(device, "drbd_alloc_pages interrupted!\n");
313 if (schedule_timeout(HZ/10) == 0)
316 finish_wait(&drbd_pp_wait, &wait);
319 atomic_add(number, &device->pp_in_use);
323 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
324 * Is also used from inside an other spin_lock_irq(&resource->req_lock);
325 * Either links the page chain back to the global pool,
326 * or returns all pages to the system. */
327 static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
329 atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
335 if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * drbd_minor_count)
336 i = page_chain_free(page);
339 tmp = page_chain_tail(page, &i);
340 spin_lock(&drbd_pp_lock);
341 page_chain_add(&drbd_pp_pool, page, tmp);
343 spin_unlock(&drbd_pp_lock);
345 i = atomic_sub_return(i, a);
347 drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
348 is_net ? "pp_in_use_by_net" : "pp_in_use", i);
349 wake_up(&drbd_pp_wait);
353 You need to hold the req_lock:
354 _drbd_wait_ee_list_empty()
356 You must not have the req_lock:
358 drbd_alloc_peer_req()
359 drbd_free_peer_reqs()
361 drbd_finish_peer_reqs()
363 drbd_wait_ee_list_empty()
366 /* normal: payload_size == request size (bi_size)
367 * w_same: payload_size == logical_block_size
368 * trim: payload_size == 0 */
369 struct drbd_peer_request *
370 drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
371 unsigned int request_size, unsigned int payload_size, gfp_t gfp_mask) __must_hold(local)
373 struct drbd_device *device = peer_device->device;
374 struct drbd_peer_request *peer_req;
375 struct page *page = NULL;
376 unsigned nr_pages = (payload_size + PAGE_SIZE -1) >> PAGE_SHIFT;
378 if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
381 peer_req = mempool_alloc(&drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
383 if (!(gfp_mask & __GFP_NOWARN))
384 drbd_err(device, "%s: allocation failed\n", __func__);
389 page = drbd_alloc_pages(peer_device, nr_pages,
390 gfpflags_allow_blocking(gfp_mask));
395 memset(peer_req, 0, sizeof(*peer_req));
396 INIT_LIST_HEAD(&peer_req->w.list);
397 drbd_clear_interval(&peer_req->i);
398 peer_req->i.size = request_size;
399 peer_req->i.sector = sector;
400 peer_req->submit_jif = jiffies;
401 peer_req->peer_device = peer_device;
402 peer_req->pages = page;
404 * The block_id is opaque to the receiver. It is not endianness
405 * converted, and sent back to the sender unchanged.
407 peer_req->block_id = id;
412 mempool_free(peer_req, &drbd_ee_mempool);
416 void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
420 if (peer_req->flags & EE_HAS_DIGEST)
421 kfree(peer_req->digest);
422 drbd_free_pages(device, peer_req->pages, is_net);
423 D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
424 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
425 if (!expect(!(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
426 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
427 drbd_al_complete_io(device, &peer_req->i);
429 mempool_free(peer_req, &drbd_ee_mempool);
432 int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
434 LIST_HEAD(work_list);
435 struct drbd_peer_request *peer_req, *t;
437 int is_net = list == &device->net_ee;
439 spin_lock_irq(&device->resource->req_lock);
440 list_splice_init(list, &work_list);
441 spin_unlock_irq(&device->resource->req_lock);
443 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
444 __drbd_free_peer_req(device, peer_req, is_net);
451 * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
453 static int drbd_finish_peer_reqs(struct drbd_device *device)
455 LIST_HEAD(work_list);
456 LIST_HEAD(reclaimed);
457 struct drbd_peer_request *peer_req, *t;
460 spin_lock_irq(&device->resource->req_lock);
461 reclaim_finished_net_peer_reqs(device, &reclaimed);
462 list_splice_init(&device->done_ee, &work_list);
463 spin_unlock_irq(&device->resource->req_lock);
465 list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
466 drbd_free_net_peer_req(device, peer_req);
468 /* possible callbacks here:
469 * e_end_block, and e_end_resync_block, e_send_superseded.
470 * all ignore the last argument.
472 list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
475 /* list_del not necessary, next/prev members not touched */
476 err2 = peer_req->w.cb(&peer_req->w, !!err);
479 drbd_free_peer_req(device, peer_req);
481 wake_up(&device->ee_wait);
486 static void _drbd_wait_ee_list_empty(struct drbd_device *device,
487 struct list_head *head)
491 /* avoids spin_lock/unlock
492 * and calling prepare_to_wait in the fast path */
493 while (!list_empty(head)) {
494 prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
495 spin_unlock_irq(&device->resource->req_lock);
497 finish_wait(&device->ee_wait, &wait);
498 spin_lock_irq(&device->resource->req_lock);
502 static void drbd_wait_ee_list_empty(struct drbd_device *device,
503 struct list_head *head)
505 spin_lock_irq(&device->resource->req_lock);
506 _drbd_wait_ee_list_empty(device, head);
507 spin_unlock_irq(&device->resource->req_lock);
510 static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
516 struct msghdr msg = {
517 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
519 iov_iter_kvec(&msg.msg_iter, READ, &iov, 1, size);
520 return sock_recvmsg(sock, &msg, msg.msg_flags);
523 static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
527 rv = drbd_recv_short(connection->data.socket, buf, size, 0);
530 if (rv == -ECONNRESET)
531 drbd_info(connection, "sock was reset by peer\n");
532 else if (rv != -ERESTARTSYS)
533 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
534 } else if (rv == 0) {
535 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
538 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
541 t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);
546 drbd_info(connection, "sock was shut down by peer\n");
550 conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
556 static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
560 err = drbd_recv(connection, buf, size);
569 static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
573 err = drbd_recv_all(connection, buf, size);
574 if (err && !signal_pending(current))
575 drbd_warn(connection, "short read (expected size %d)\n", (int)size);
580 * On individual connections, the socket buffer size must be set prior to the
581 * listen(2) or connect(2) calls in order to have it take effect.
582 * This is our wrapper to do so.
584 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
587 /* open coded SO_SNDBUF, SO_RCVBUF */
589 sock->sk->sk_sndbuf = snd;
590 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
593 sock->sk->sk_rcvbuf = rcv;
594 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
598 static struct socket *drbd_try_connect(struct drbd_connection *connection)
602 struct sockaddr_in6 src_in6;
603 struct sockaddr_in6 peer_in6;
605 int err, peer_addr_len, my_addr_len;
606 int sndbuf_size, rcvbuf_size, connect_int;
607 int disconnect_on_error = 1;
610 nc = rcu_dereference(connection->net_conf);
615 sndbuf_size = nc->sndbuf_size;
616 rcvbuf_size = nc->rcvbuf_size;
617 connect_int = nc->connect_int;
620 my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
621 memcpy(&src_in6, &connection->my_addr, my_addr_len);
623 if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
624 src_in6.sin6_port = 0;
626 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
628 peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
629 memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);
631 what = "sock_create_kern";
632 err = sock_create_kern(&init_net, ((struct sockaddr *)&src_in6)->sa_family,
633 SOCK_STREAM, IPPROTO_TCP, &sock);
639 sock->sk->sk_rcvtimeo =
640 sock->sk->sk_sndtimeo = connect_int * HZ;
641 drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);
643 /* explicitly bind to the configured IP as source IP
644 * for the outgoing connections.
645 * This is needed for multihomed hosts and to be
646 * able to use lo: interfaces for drbd.
647 * Make sure to use 0 as port number, so linux selects
648 * a free one dynamically.
650 what = "bind before connect";
651 err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
655 /* connect may fail, peer not yet available.
656 * stay C_WF_CONNECTION, don't go Disconnecting! */
657 disconnect_on_error = 0;
659 err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);
668 /* timeout, busy, signal pending */
669 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
670 case EINTR: case ERESTARTSYS:
671 /* peer not (yet) available, network problem */
672 case ECONNREFUSED: case ENETUNREACH:
673 case EHOSTDOWN: case EHOSTUNREACH:
674 disconnect_on_error = 0;
677 drbd_err(connection, "%s failed, err = %d\n", what, err);
679 if (disconnect_on_error)
680 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
686 struct accept_wait_data {
687 struct drbd_connection *connection;
688 struct socket *s_listen;
689 struct completion door_bell;
690 void (*original_sk_state_change)(struct sock *sk);
694 static void drbd_incoming_connection(struct sock *sk)
696 struct accept_wait_data *ad = sk->sk_user_data;
697 void (*state_change)(struct sock *sk);
699 state_change = ad->original_sk_state_change;
700 if (sk->sk_state == TCP_ESTABLISHED)
701 complete(&ad->door_bell);
705 static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
707 int err, sndbuf_size, rcvbuf_size, my_addr_len;
708 struct sockaddr_in6 my_addr;
709 struct socket *s_listen;
714 nc = rcu_dereference(connection->net_conf);
719 sndbuf_size = nc->sndbuf_size;
720 rcvbuf_size = nc->rcvbuf_size;
723 my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
724 memcpy(&my_addr, &connection->my_addr, my_addr_len);
726 what = "sock_create_kern";
727 err = sock_create_kern(&init_net, ((struct sockaddr *)&my_addr)->sa_family,
728 SOCK_STREAM, IPPROTO_TCP, &s_listen);
734 s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
735 drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);
737 what = "bind before listen";
738 err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
742 ad->s_listen = s_listen;
743 write_lock_bh(&s_listen->sk->sk_callback_lock);
744 ad->original_sk_state_change = s_listen->sk->sk_state_change;
745 s_listen->sk->sk_state_change = drbd_incoming_connection;
746 s_listen->sk->sk_user_data = ad;
747 write_unlock_bh(&s_listen->sk->sk_callback_lock);
750 err = s_listen->ops->listen(s_listen, 5);
757 sock_release(s_listen);
759 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
760 drbd_err(connection, "%s failed, err = %d\n", what, err);
761 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
768 static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
770 write_lock_bh(&sk->sk_callback_lock);
771 sk->sk_state_change = ad->original_sk_state_change;
772 sk->sk_user_data = NULL;
773 write_unlock_bh(&sk->sk_callback_lock);
776 static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
778 int timeo, connect_int, err = 0;
779 struct socket *s_estab = NULL;
783 nc = rcu_dereference(connection->net_conf);
788 connect_int = nc->connect_int;
791 timeo = connect_int * HZ;
792 /* 28.5% random jitter */
793 timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;
795 err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
799 err = kernel_accept(ad->s_listen, &s_estab, 0);
801 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
802 drbd_err(connection, "accept failed, err = %d\n", err);
803 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
808 unregister_state_change(s_estab->sk, ad);
813 static int decode_header(struct drbd_connection *, void *, struct packet_info *);
815 static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
816 enum drbd_packet cmd)
818 if (!conn_prepare_command(connection, sock))
820 return conn_send_command(connection, sock, cmd, 0, NULL, 0);
823 static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
825 unsigned int header_size = drbd_header_size(connection);
826 struct packet_info pi;
831 nc = rcu_dereference(connection->net_conf);
836 sock->sk->sk_rcvtimeo = nc->ping_timeo * 4 * HZ / 10;
839 err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
840 if (err != header_size) {
845 err = decode_header(connection, connection->data.rbuf, &pi);
852 * drbd_socket_okay() - Free the socket if its connection is not okay
853 * @sock: pointer to the pointer to the socket.
855 static bool drbd_socket_okay(struct socket **sock)
863 rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
865 if (rr > 0 || rr == -EAGAIN) {
874 static bool connection_established(struct drbd_connection *connection,
875 struct socket **sock1,
876 struct socket **sock2)
882 if (!*sock1 || !*sock2)
886 nc = rcu_dereference(connection->net_conf);
887 timeout = (nc->sock_check_timeo ?: nc->ping_timeo) * HZ / 10;
889 schedule_timeout_interruptible(timeout);
891 ok = drbd_socket_okay(sock1);
892 ok = drbd_socket_okay(sock2) && ok;
897 /* Gets called if a connection is established, or if a new minor gets created
899 int drbd_connected(struct drbd_peer_device *peer_device)
901 struct drbd_device *device = peer_device->device;
904 atomic_set(&device->packet_seq, 0);
905 device->peer_seq = 0;
907 device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
908 &peer_device->connection->cstate_mutex :
909 &device->own_state_mutex;
911 err = drbd_send_sync_param(peer_device);
913 err = drbd_send_sizes(peer_device, 0, 0);
915 err = drbd_send_uuids(peer_device);
917 err = drbd_send_current_state(peer_device);
918 clear_bit(USE_DEGR_WFC_T, &device->flags);
919 clear_bit(RESIZE_PENDING, &device->flags);
920 atomic_set(&device->ap_in_flight, 0);
921 mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
927 * 1 yes, we have a valid connection
928 * 0 oops, did not work out, please try again
929 * -1 peer talks different language,
930 * no point in trying again, please go standalone.
931 * -2 We do not have a network config...
933 static int conn_connect(struct drbd_connection *connection)
935 struct drbd_socket sock, msock;
936 struct drbd_peer_device *peer_device;
939 bool discard_my_data, ok;
940 enum drbd_state_rv rv;
941 struct accept_wait_data ad = {
942 .connection = connection,
943 .door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
946 clear_bit(DISCONNECT_SENT, &connection->flags);
947 if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
950 mutex_init(&sock.mutex);
951 sock.sbuf = connection->data.sbuf;
952 sock.rbuf = connection->data.rbuf;
954 mutex_init(&msock.mutex);
955 msock.sbuf = connection->meta.sbuf;
956 msock.rbuf = connection->meta.rbuf;
959 /* Assume that the peer only understands protocol 80 until we know better. */
960 connection->agreed_pro_version = 80;
962 if (prepare_listen_socket(connection, &ad))
968 s = drbd_try_connect(connection);
972 send_first_packet(connection, &sock, P_INITIAL_DATA);
973 } else if (!msock.socket) {
974 clear_bit(RESOLVE_CONFLICTS, &connection->flags);
976 send_first_packet(connection, &msock, P_INITIAL_META);
978 drbd_err(connection, "Logic error in conn_connect()\n");
979 goto out_release_sockets;
983 if (connection_established(connection, &sock.socket, &msock.socket))
987 s = drbd_wait_for_connect(connection, &ad);
989 int fp = receive_first_packet(connection, s);
990 drbd_socket_okay(&sock.socket);
991 drbd_socket_okay(&msock.socket);
995 drbd_warn(connection, "initial packet S crossed\n");
996 sock_release(sock.socket);
1002 case P_INITIAL_META:
1003 set_bit(RESOLVE_CONFLICTS, &connection->flags);
1005 drbd_warn(connection, "initial packet M crossed\n");
1006 sock_release(msock.socket);
1013 drbd_warn(connection, "Error receiving initial packet\n");
1016 if (prandom_u32() & 1)
1021 if (connection->cstate <= C_DISCONNECTING)
1022 goto out_release_sockets;
1023 if (signal_pending(current)) {
1024 flush_signals(current);
1026 if (get_t_state(&connection->receiver) == EXITING)
1027 goto out_release_sockets;
1030 ok = connection_established(connection, &sock.socket, &msock.socket);
1034 sock_release(ad.s_listen);
1036 sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1037 msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1039 sock.socket->sk->sk_allocation = GFP_NOIO;
1040 msock.socket->sk->sk_allocation = GFP_NOIO;
1042 sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
1043 msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
1046 * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
1047 * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1048 * first set it to the P_CONNECTION_FEATURES timeout,
1049 * which we set to 4x the configured ping_timeout. */
1051 nc = rcu_dereference(connection->net_conf);
1053 sock.socket->sk->sk_sndtimeo =
1054 sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
1056 msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
1057 timeout = nc->timeout * HZ / 10;
1058 discard_my_data = nc->discard_my_data;
1061 msock.socket->sk->sk_sndtimeo = timeout;
1063 /* we don't want delays.
1064 * we use TCP_CORK where appropriate, though */
1065 drbd_tcp_nodelay(sock.socket);
1066 drbd_tcp_nodelay(msock.socket);
1068 connection->data.socket = sock.socket;
1069 connection->meta.socket = msock.socket;
1070 connection->last_received = jiffies;
1072 h = drbd_do_features(connection);
1076 if (connection->cram_hmac_tfm) {
1077 /* drbd_request_state(device, NS(conn, WFAuth)); */
1078 switch (drbd_do_auth(connection)) {
1080 drbd_err(connection, "Authentication of peer failed\n");
1083 drbd_err(connection, "Authentication of peer failed, trying again.\n");
1088 connection->data.socket->sk->sk_sndtimeo = timeout;
1089 connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1091 if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1094 /* Prevent a race between resync-handshake and
1095 * being promoted to Primary.
1097 * Grab and release the state mutex, so we know that any current
1098 * drbd_set_role() is finished, and any incoming drbd_set_role
1099 * will see the STATE_SENT flag, and wait for it to be cleared.
1101 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1102 mutex_lock(peer_device->device->state_mutex);
1104 /* avoid a race with conn_request_state( C_DISCONNECTING ) */
1105 spin_lock_irq(&connection->resource->req_lock);
1106 set_bit(STATE_SENT, &connection->flags);
1107 spin_unlock_irq(&connection->resource->req_lock);
1109 idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1110 mutex_unlock(peer_device->device->state_mutex);
1113 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1114 struct drbd_device *device = peer_device->device;
1115 kref_get(&device->kref);
1118 if (discard_my_data)
1119 set_bit(DISCARD_MY_DATA, &device->flags);
1121 clear_bit(DISCARD_MY_DATA, &device->flags);
1123 drbd_connected(peer_device);
1124 kref_put(&device->kref, drbd_destroy_device);
1129 rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1130 if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1131 clear_bit(STATE_SENT, &connection->flags);
1135 drbd_thread_start(&connection->ack_receiver);
1136 /* opencoded create_singlethread_workqueue(),
1137 * to be able to use format string arguments */
1138 connection->ack_sender =
1139 alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
1140 if (!connection->ack_sender) {
1141 drbd_err(connection, "Failed to create workqueue ack_sender\n");
1145 mutex_lock(&connection->resource->conf_update);
1146 /* The discard_my_data flag is a single-shot modifier to the next
1147 * connection attempt, the handshake of which is now well underway.
1148 * No need for rcu style copying of the whole struct
1149 * just to clear a single value. */
1150 connection->net_conf->discard_my_data = 0;
1151 mutex_unlock(&connection->resource->conf_update);
1155 out_release_sockets:
1157 sock_release(ad.s_listen);
1159 sock_release(sock.socket);
1161 sock_release(msock.socket);
1165 static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1167 unsigned int header_size = drbd_header_size(connection);
1169 if (header_size == sizeof(struct p_header100) &&
1170 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1171 struct p_header100 *h = header;
1173 drbd_err(connection, "Header padding is not zero\n");
1176 pi->vnr = be16_to_cpu(h->volume);
1177 pi->cmd = be16_to_cpu(h->command);
1178 pi->size = be32_to_cpu(h->length);
1179 } else if (header_size == sizeof(struct p_header95) &&
1180 *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1181 struct p_header95 *h = header;
1182 pi->cmd = be16_to_cpu(h->command);
1183 pi->size = be32_to_cpu(h->length);
1185 } else if (header_size == sizeof(struct p_header80) &&
1186 *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1187 struct p_header80 *h = header;
1188 pi->cmd = be16_to_cpu(h->command);
1189 pi->size = be16_to_cpu(h->length);
1192 drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1193 be32_to_cpu(*(__be32 *)header),
1194 connection->agreed_pro_version);
1197 pi->data = header + header_size;
1201 static void drbd_unplug_all_devices(struct drbd_connection *connection)
1203 if (current->plug == &connection->receiver_plug) {
1204 blk_finish_plug(&connection->receiver_plug);
1205 blk_start_plug(&connection->receiver_plug);
1206 } /* else: maybe just schedule() ?? */
1209 static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1211 void *buffer = connection->data.rbuf;
1214 err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1218 err = decode_header(connection, buffer, pi);
1219 connection->last_received = jiffies;
1224 static int drbd_recv_header_maybe_unplug(struct drbd_connection *connection, struct packet_info *pi)
1226 void *buffer = connection->data.rbuf;
1227 unsigned int size = drbd_header_size(connection);
1230 err = drbd_recv_short(connection->data.socket, buffer, size, MSG_NOSIGNAL|MSG_DONTWAIT);
1232 /* If we have nothing in the receive buffer now, to reduce
1233 * application latency, try to drain the backend queues as
1234 * quickly as possible, and let remote TCP know what we have
1235 * received so far. */
1236 if (err == -EAGAIN) {
1237 drbd_tcp_quickack(connection->data.socket);
1238 drbd_unplug_all_devices(connection);
1244 err = drbd_recv_all_warn(connection, buffer, size);
1249 err = decode_header(connection, connection->data.rbuf, pi);
1250 connection->last_received = jiffies;
1254 /* This is blkdev_issue_flush, but asynchronous.
1255 * We want to submit to all component volumes in parallel,
1256 * then wait for all completions.
1258 struct issue_flush_context {
1261 struct completion done;
1263 struct one_flush_context {
1264 struct drbd_device *device;
1265 struct issue_flush_context *ctx;
1268 static void one_flush_endio(struct bio *bio)
1270 struct one_flush_context *octx = bio->bi_private;
1271 struct drbd_device *device = octx->device;
1272 struct issue_flush_context *ctx = octx->ctx;
1274 if (bio->bi_status) {
1275 ctx->error = blk_status_to_errno(bio->bi_status);
1276 drbd_info(device, "local disk FLUSH FAILED with status %d\n", bio->bi_status);
1281 clear_bit(FLUSH_PENDING, &device->flags);
1283 kref_put(&device->kref, drbd_destroy_device);
1285 if (atomic_dec_and_test(&ctx->pending))
1286 complete(&ctx->done);
1289 static void submit_one_flush(struct drbd_device *device, struct issue_flush_context *ctx)
1291 struct bio *bio = bio_alloc(GFP_NOIO, 0);
1292 struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);
1293 if (!bio || !octx) {
1294 drbd_warn(device, "Could not allocate a bio, CANNOT ISSUE FLUSH\n");
1295 /* FIXME: what else can I do now? disconnecting or detaching
1296 * really does not help to improve the state of the world, either.
1302 ctx->error = -ENOMEM;
1304 kref_put(&device->kref, drbd_destroy_device);
1308 octx->device = device;
1310 bio_set_dev(bio, device->ldev->backing_bdev);
1311 bio->bi_private = octx;
1312 bio->bi_end_io = one_flush_endio;
1313 bio->bi_opf = REQ_OP_FLUSH | REQ_PREFLUSH;
1315 device->flush_jif = jiffies;
1316 set_bit(FLUSH_PENDING, &device->flags);
1317 atomic_inc(&ctx->pending);
1321 static void drbd_flush(struct drbd_connection *connection)
1323 if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1324 struct drbd_peer_device *peer_device;
1325 struct issue_flush_context ctx;
1328 atomic_set(&ctx.pending, 1);
1330 init_completion(&ctx.done);
1333 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1334 struct drbd_device *device = peer_device->device;
1336 if (!get_ldev(device))
1338 kref_get(&device->kref);
1341 submit_one_flush(device, &ctx);
1347 /* Do we want to add a timeout,
1348 * if disk-timeout is set? */
1349 if (!atomic_dec_and_test(&ctx.pending))
1350 wait_for_completion(&ctx.done);
1353 /* would rather check on EOPNOTSUPP, but that is not reliable.
1354 * don't try again for ANY return value != 0
1355 * if (rv == -EOPNOTSUPP) */
1356 /* Any error is already reported by bio_endio callback. */
1357 drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1363 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1364 * @device: DRBD device.
1365 * @epoch: Epoch object.
1368 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1369 struct drbd_epoch *epoch,
1370 enum epoch_event ev)
1373 struct drbd_epoch *next_epoch;
1374 enum finish_epoch rv = FE_STILL_LIVE;
1376 spin_lock(&connection->epoch_lock);
1380 epoch_size = atomic_read(&epoch->epoch_size);
1382 switch (ev & ~EV_CLEANUP) {
1384 atomic_dec(&epoch->active);
1386 case EV_GOT_BARRIER_NR:
1387 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1389 case EV_BECAME_LAST:
1394 if (epoch_size != 0 &&
1395 atomic_read(&epoch->active) == 0 &&
1396 (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1397 if (!(ev & EV_CLEANUP)) {
1398 spin_unlock(&connection->epoch_lock);
1399 drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1400 spin_lock(&connection->epoch_lock);
1403 /* FIXME: dec unacked on connection, once we have
1404 * something to count pending connection packets in. */
1405 if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1406 dec_unacked(epoch->connection);
1409 if (connection->current_epoch != epoch) {
1410 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1411 list_del(&epoch->list);
1412 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1413 connection->epochs--;
1416 if (rv == FE_STILL_LIVE)
1420 atomic_set(&epoch->epoch_size, 0);
1421 /* atomic_set(&epoch->active, 0); is already zero */
1422 if (rv == FE_STILL_LIVE)
1433 spin_unlock(&connection->epoch_lock);
1438 static enum write_ordering_e
1439 max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1441 struct disk_conf *dc;
1443 dc = rcu_dereference(bdev->disk_conf);
1445 if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1447 if (wo == WO_DRAIN_IO && !dc->disk_drain)
1454 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1455 * @connection: DRBD connection.
1456 * @wo: Write ordering method to try.
1458 void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1459 enum write_ordering_e wo)
1461 struct drbd_device *device;
1462 enum write_ordering_e pwo;
1464 static char *write_ordering_str[] = {
1466 [WO_DRAIN_IO] = "drain",
1467 [WO_BDEV_FLUSH] = "flush",
1470 pwo = resource->write_ordering;
1471 if (wo != WO_BDEV_FLUSH)
1474 idr_for_each_entry(&resource->devices, device, vnr) {
1475 if (get_ldev(device)) {
1476 wo = max_allowed_wo(device->ldev, wo);
1477 if (device->ldev == bdev)
1484 wo = max_allowed_wo(bdev, wo);
1488 resource->write_ordering = wo;
1489 if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1490 drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1493 static void drbd_issue_peer_discard(struct drbd_device *device, struct drbd_peer_request *peer_req)
1495 struct block_device *bdev = device->ldev->backing_bdev;
1497 if (blkdev_issue_zeroout(bdev, peer_req->i.sector, peer_req->i.size >> 9,
1499 peer_req->flags |= EE_WAS_ERROR;
1501 drbd_endio_write_sec_final(peer_req);
1504 static void drbd_issue_peer_wsame(struct drbd_device *device,
1505 struct drbd_peer_request *peer_req)
1507 struct block_device *bdev = device->ldev->backing_bdev;
1508 sector_t s = peer_req->i.sector;
1509 sector_t nr = peer_req->i.size >> 9;
1510 if (blkdev_issue_write_same(bdev, s, nr, GFP_NOIO, peer_req->pages))
1511 peer_req->flags |= EE_WAS_ERROR;
1512 drbd_endio_write_sec_final(peer_req);
1517 * drbd_submit_peer_request()
1518 * @device: DRBD device.
1519 * @peer_req: peer request
1520 * @rw: flag field, see bio->bi_opf
1522 * May spread the pages to multiple bios,
1523 * depending on bio_add_page restrictions.
1525 * Returns 0 if all bios have been submitted,
1526 * -ENOMEM if we could not allocate enough bios,
1527 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1528 * single page to an empty bio (which should never happen and likely indicates
1529 * that the lower level IO stack is in some way broken). This has been observed
1530 * on certain Xen deployments.
1532 /* TODO allocate from our own bio_set. */
1533 int drbd_submit_peer_request(struct drbd_device *device,
1534 struct drbd_peer_request *peer_req,
1535 const unsigned op, const unsigned op_flags,
1536 const int fault_type)
1538 struct bio *bios = NULL;
1540 struct page *page = peer_req->pages;
1541 sector_t sector = peer_req->i.sector;
1542 unsigned data_size = peer_req->i.size;
1543 unsigned n_bios = 0;
1544 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
1547 /* TRIM/DISCARD: for now, always use the helper function
1548 * blkdev_issue_zeroout(..., discard=true).
1549 * It's synchronous, but it does the right thing wrt. bio splitting.
1550 * Correctness first, performance later. Next step is to code an
1551 * asynchronous variant of the same.
1553 if (peer_req->flags & (EE_IS_TRIM|EE_WRITE_SAME)) {
1554 /* wait for all pending IO completions, before we start
1555 * zeroing things out. */
1556 conn_wait_active_ee_empty(peer_req->peer_device->connection);
1557 /* add it to the active list now,
1558 * so we can find it to present it in debugfs */
1559 peer_req->submit_jif = jiffies;
1560 peer_req->flags |= EE_SUBMITTED;
1562 /* If this was a resync request from receive_rs_deallocated(),
1563 * it is already on the sync_ee list */
1564 if (list_empty(&peer_req->w.list)) {
1565 spin_lock_irq(&device->resource->req_lock);
1566 list_add_tail(&peer_req->w.list, &device->active_ee);
1567 spin_unlock_irq(&device->resource->req_lock);
1570 if (peer_req->flags & EE_IS_TRIM)
1571 drbd_issue_peer_discard(device, peer_req);
1572 else /* EE_WRITE_SAME */
1573 drbd_issue_peer_wsame(device, peer_req);
1577 /* In most cases, we will only need one bio. But in case the lower
1578 * level restrictions happen to be different at this offset on this
1579 * side than those of the sending peer, we may need to submit the
1580 * request in more than one bio.
1582 * Plain bio_alloc is good enough here, this is no DRBD internally
1583 * generated bio, but a bio allocated on behalf of the peer.
1586 bio = bio_alloc(GFP_NOIO, nr_pages);
1588 drbd_err(device, "submit_ee: Allocation of a bio failed (nr_pages=%u)\n", nr_pages);
1591 /* > peer_req->i.sector, unless this is the first bio */
1592 bio->bi_iter.bi_sector = sector;
1593 bio_set_dev(bio, device->ldev->backing_bdev);
1594 bio_set_op_attrs(bio, op, op_flags);
1595 bio->bi_private = peer_req;
1596 bio->bi_end_io = drbd_peer_request_endio;
1598 bio->bi_next = bios;
1602 page_chain_for_each(page) {
1603 unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1604 if (!bio_add_page(bio, page, len, 0))
1610 D_ASSERT(device, data_size == 0);
1611 D_ASSERT(device, page == NULL);
1613 atomic_set(&peer_req->pending_bios, n_bios);
1614 /* for debugfs: update timestamp, mark as submitted */
1615 peer_req->submit_jif = jiffies;
1616 peer_req->flags |= EE_SUBMITTED;
1619 bios = bios->bi_next;
1620 bio->bi_next = NULL;
1622 drbd_generic_make_request(device, fault_type, bio);
1629 bios = bios->bi_next;
1635 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1636 struct drbd_peer_request *peer_req)
1638 struct drbd_interval *i = &peer_req->i;
1640 drbd_remove_interval(&device->write_requests, i);
1641 drbd_clear_interval(i);
1643 /* Wake up any processes waiting for this peer request to complete. */
1645 wake_up(&device->misc_wait);
1648 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1650 struct drbd_peer_device *peer_device;
1654 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1655 struct drbd_device *device = peer_device->device;
1657 kref_get(&device->kref);
1659 drbd_wait_ee_list_empty(device, &device->active_ee);
1660 kref_put(&device->kref, drbd_destroy_device);
1666 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1669 struct p_barrier *p = pi->data;
1670 struct drbd_epoch *epoch;
1672 /* FIXME these are unacked on connection,
1673 * not a specific (peer)device.
1675 connection->current_epoch->barrier_nr = p->barrier;
1676 connection->current_epoch->connection = connection;
1677 rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1679 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1680 * the activity log, which means it would not be resynced in case the
1681 * R_PRIMARY crashes now.
1682 * Therefore we must send the barrier_ack after the barrier request was
1684 switch (connection->resource->write_ordering) {
1686 if (rv == FE_RECYCLED)
1689 /* receiver context, in the writeout path of the other node.
1690 * avoid potential distributed deadlock */
1691 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1695 drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1700 conn_wait_active_ee_empty(connection);
1701 drbd_flush(connection);
1703 if (atomic_read(&connection->current_epoch->epoch_size)) {
1704 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1711 drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1712 connection->resource->write_ordering);
1717 atomic_set(&epoch->epoch_size, 0);
1718 atomic_set(&epoch->active, 0);
1720 spin_lock(&connection->epoch_lock);
1721 if (atomic_read(&connection->current_epoch->epoch_size)) {
1722 list_add(&epoch->list, &connection->current_epoch->list);
1723 connection->current_epoch = epoch;
1724 connection->epochs++;
1726 /* The current_epoch got recycled while we allocated this one... */
1729 spin_unlock(&connection->epoch_lock);
1734 /* quick wrapper in case payload size != request_size (write same) */
1735 static void drbd_csum_ee_size(struct crypto_shash *h,
1736 struct drbd_peer_request *r, void *d,
1737 unsigned int payload_size)
1739 unsigned int tmp = r->i.size;
1740 r->i.size = payload_size;
1741 drbd_csum_ee(h, r, d);
1745 /* used from receive_RSDataReply (recv_resync_read)
1746 * and from receive_Data.
1747 * data_size: actual payload ("data in")
1748 * for normal writes that is bi_size.
1749 * for discards, that is zero.
1750 * for write same, it is logical_block_size.
1751 * both trim and write same have the bi_size ("data len to be affected")
1752 * as extra argument in the packet header.
1754 static struct drbd_peer_request *
1755 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1756 struct packet_info *pi) __must_hold(local)
1758 struct drbd_device *device = peer_device->device;
1759 const sector_t capacity = drbd_get_capacity(device->this_bdev);
1760 struct drbd_peer_request *peer_req;
1762 int digest_size, err;
1763 unsigned int data_size = pi->size, ds;
1764 void *dig_in = peer_device->connection->int_dig_in;
1765 void *dig_vv = peer_device->connection->int_dig_vv;
1766 unsigned long *data;
1767 struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1768 struct p_trim *wsame = (pi->cmd == P_WSAME) ? pi->data : NULL;
1771 if (!trim && peer_device->connection->peer_integrity_tfm) {
1772 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1774 * FIXME: Receive the incoming digest into the receive buffer
1775 * here, together with its struct p_data?
1777 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1780 data_size -= digest_size;
1783 /* assume request_size == data_size, but special case trim and wsame. */
1786 if (!expect(data_size == 0))
1788 ds = be32_to_cpu(trim->size);
1790 if (data_size != queue_logical_block_size(device->rq_queue)) {
1791 drbd_err(peer_device, "data size (%u) != drbd logical block size (%u)\n",
1792 data_size, queue_logical_block_size(device->rq_queue));
1795 if (data_size != bdev_logical_block_size(device->ldev->backing_bdev)) {
1796 drbd_err(peer_device, "data size (%u) != backend logical block size (%u)\n",
1797 data_size, bdev_logical_block_size(device->ldev->backing_bdev));
1800 ds = be32_to_cpu(wsame->size);
1803 if (!expect(IS_ALIGNED(ds, 512)))
1805 if (trim || wsame) {
1806 if (!expect(ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
1808 } else if (!expect(ds <= DRBD_MAX_BIO_SIZE))
1811 /* even though we trust out peer,
1812 * we sometimes have to double check. */
1813 if (sector + (ds>>9) > capacity) {
1814 drbd_err(device, "request from peer beyond end of local disk: "
1815 "capacity: %llus < sector: %llus + size: %u\n",
1816 (unsigned long long)capacity,
1817 (unsigned long long)sector, ds);
1821 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1822 * "criss-cross" setup, that might cause write-out on some other DRBD,
1823 * which in turn might block on the other node at this very place. */
1824 peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
1828 peer_req->flags |= EE_WRITE;
1830 peer_req->flags |= EE_IS_TRIM;
1834 peer_req->flags |= EE_WRITE_SAME;
1836 /* receive payload size bytes into page chain */
1838 page = peer_req->pages;
1839 page_chain_for_each(page) {
1840 unsigned len = min_t(int, ds, PAGE_SIZE);
1842 err = drbd_recv_all_warn(peer_device->connection, data, len);
1843 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1844 drbd_err(device, "Fault injection: Corrupting data on receive\n");
1845 data[0] = data[0] ^ (unsigned long)-1;
1849 drbd_free_peer_req(device, peer_req);
1856 drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
1857 if (memcmp(dig_in, dig_vv, digest_size)) {
1858 drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1859 (unsigned long long)sector, data_size);
1860 drbd_free_peer_req(device, peer_req);
1864 device->recv_cnt += data_size >> 9;
1868 /* drbd_drain_block() just takes a data block
1869 * out of the socket input buffer, and discards it.
1871 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1880 page = drbd_alloc_pages(peer_device, 1, 1);
1884 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1886 err = drbd_recv_all_warn(peer_device->connection, data, len);
1892 drbd_free_pages(peer_device->device, page, 0);
1896 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1897 sector_t sector, int data_size)
1899 struct bio_vec bvec;
1900 struct bvec_iter iter;
1902 int digest_size, err, expect;
1903 void *dig_in = peer_device->connection->int_dig_in;
1904 void *dig_vv = peer_device->connection->int_dig_vv;
1907 if (peer_device->connection->peer_integrity_tfm) {
1908 digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1909 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1912 data_size -= digest_size;
1915 /* optimistically update recv_cnt. if receiving fails below,
1916 * we disconnect anyways, and counters will be reset. */
1917 peer_device->device->recv_cnt += data_size>>9;
1919 bio = req->master_bio;
1920 D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
1922 bio_for_each_segment(bvec, bio, iter) {
1923 void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
1924 expect = min_t(int, data_size, bvec.bv_len);
1925 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
1926 kunmap(bvec.bv_page);
1929 data_size -= expect;
1933 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
1934 if (memcmp(dig_in, dig_vv, digest_size)) {
1935 drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
1940 D_ASSERT(peer_device->device, data_size == 0);
1945 * e_end_resync_block() is called in ack_sender context via
1946 * drbd_finish_peer_reqs().
1948 static int e_end_resync_block(struct drbd_work *w, int unused)
1950 struct drbd_peer_request *peer_req =
1951 container_of(w, struct drbd_peer_request, w);
1952 struct drbd_peer_device *peer_device = peer_req->peer_device;
1953 struct drbd_device *device = peer_device->device;
1954 sector_t sector = peer_req->i.sector;
1957 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1959 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1960 drbd_set_in_sync(device, sector, peer_req->i.size);
1961 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
1963 /* Record failure to sync */
1964 drbd_rs_failed_io(device, sector, peer_req->i.size);
1966 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1968 dec_unacked(device);
1973 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
1974 struct packet_info *pi) __releases(local)
1976 struct drbd_device *device = peer_device->device;
1977 struct drbd_peer_request *peer_req;
1979 peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
1983 dec_rs_pending(device);
1985 inc_unacked(device);
1986 /* corresponding dec_unacked() in e_end_resync_block()
1987 * respective _drbd_clear_done_ee */
1989 peer_req->w.cb = e_end_resync_block;
1990 peer_req->submit_jif = jiffies;
1992 spin_lock_irq(&device->resource->req_lock);
1993 list_add_tail(&peer_req->w.list, &device->sync_ee);
1994 spin_unlock_irq(&device->resource->req_lock);
1996 atomic_add(pi->size >> 9, &device->rs_sect_ev);
1997 if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE, 0,
1998 DRBD_FAULT_RS_WR) == 0)
2001 /* don't care for the reason here */
2002 drbd_err(device, "submit failed, triggering re-connect\n");
2003 spin_lock_irq(&device->resource->req_lock);
2004 list_del(&peer_req->w.list);
2005 spin_unlock_irq(&device->resource->req_lock);
2007 drbd_free_peer_req(device, peer_req);
2013 static struct drbd_request *
2014 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
2015 sector_t sector, bool missing_ok, const char *func)
2017 struct drbd_request *req;
2019 /* Request object according to our peer */
2020 req = (struct drbd_request *)(unsigned long)id;
2021 if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
2024 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
2025 (unsigned long)id, (unsigned long long)sector);
2030 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
2032 struct drbd_peer_device *peer_device;
2033 struct drbd_device *device;
2034 struct drbd_request *req;
2037 struct p_data *p = pi->data;
2039 peer_device = conn_peer_device(connection, pi->vnr);
2042 device = peer_device->device;
2044 sector = be64_to_cpu(p->sector);
2046 spin_lock_irq(&device->resource->req_lock);
2047 req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
2048 spin_unlock_irq(&device->resource->req_lock);
2052 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
2053 * special casing it there for the various failure cases.
2054 * still no race with drbd_fail_pending_reads */
2055 err = recv_dless_read(peer_device, req, sector, pi->size);
2057 req_mod(req, DATA_RECEIVED);
2058 /* else: nothing. handled from drbd_disconnect...
2059 * I don't think we may complete this just yet
2060 * in case we are "on-disconnect: freeze" */
2065 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
2067 struct drbd_peer_device *peer_device;
2068 struct drbd_device *device;
2071 struct p_data *p = pi->data;
2073 peer_device = conn_peer_device(connection, pi->vnr);
2076 device = peer_device->device;
2078 sector = be64_to_cpu(p->sector);
2079 D_ASSERT(device, p->block_id == ID_SYNCER);
2081 if (get_ldev(device)) {
2082 /* data is submitted to disk within recv_resync_read.
2083 * corresponding put_ldev done below on error,
2084 * or in drbd_peer_request_endio. */
2085 err = recv_resync_read(peer_device, sector, pi);
2087 if (__ratelimit(&drbd_ratelimit_state))
2088 drbd_err(device, "Can not write resync data to local disk.\n");
2090 err = drbd_drain_block(peer_device, pi->size);
2092 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2095 atomic_add(pi->size >> 9, &device->rs_sect_in);
2100 static void restart_conflicting_writes(struct drbd_device *device,
2101 sector_t sector, int size)
2103 struct drbd_interval *i;
2104 struct drbd_request *req;
2106 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2109 req = container_of(i, struct drbd_request, i);
2110 if (req->rq_state & RQ_LOCAL_PENDING ||
2111 !(req->rq_state & RQ_POSTPONED))
2113 /* as it is RQ_POSTPONED, this will cause it to
2114 * be queued on the retry workqueue. */
2115 __req_mod(req, CONFLICT_RESOLVED, NULL);
2120 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2122 static int e_end_block(struct drbd_work *w, int cancel)
2124 struct drbd_peer_request *peer_req =
2125 container_of(w, struct drbd_peer_request, w);
2126 struct drbd_peer_device *peer_device = peer_req->peer_device;
2127 struct drbd_device *device = peer_device->device;
2128 sector_t sector = peer_req->i.sector;
2131 if (peer_req->flags & EE_SEND_WRITE_ACK) {
2132 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2133 pcmd = (device->state.conn >= C_SYNC_SOURCE &&
2134 device->state.conn <= C_PAUSED_SYNC_T &&
2135 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
2136 P_RS_WRITE_ACK : P_WRITE_ACK;
2137 err = drbd_send_ack(peer_device, pcmd, peer_req);
2138 if (pcmd == P_RS_WRITE_ACK)
2139 drbd_set_in_sync(device, sector, peer_req->i.size);
2141 err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2142 /* we expect it to be marked out of sync anyways...
2143 * maybe assert this? */
2145 dec_unacked(device);
2148 /* we delete from the conflict detection hash _after_ we sent out the
2149 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
2150 if (peer_req->flags & EE_IN_INTERVAL_TREE) {
2151 spin_lock_irq(&device->resource->req_lock);
2152 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
2153 drbd_remove_epoch_entry_interval(device, peer_req);
2154 if (peer_req->flags & EE_RESTART_REQUESTS)
2155 restart_conflicting_writes(device, sector, peer_req->i.size);
2156 spin_unlock_irq(&device->resource->req_lock);
2158 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2160 drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2165 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2167 struct drbd_peer_request *peer_req =
2168 container_of(w, struct drbd_peer_request, w);
2169 struct drbd_peer_device *peer_device = peer_req->peer_device;
2172 err = drbd_send_ack(peer_device, ack, peer_req);
2173 dec_unacked(peer_device->device);
2178 static int e_send_superseded(struct drbd_work *w, int unused)
2180 return e_send_ack(w, P_SUPERSEDED);
2183 static int e_send_retry_write(struct drbd_work *w, int unused)
2185 struct drbd_peer_request *peer_req =
2186 container_of(w, struct drbd_peer_request, w);
2187 struct drbd_connection *connection = peer_req->peer_device->connection;
2189 return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2190 P_RETRY_WRITE : P_SUPERSEDED);
2193 static bool seq_greater(u32 a, u32 b)
2196 * We assume 32-bit wrap-around here.
2197 * For 24-bit wrap-around, we would have to shift:
2200 return (s32)a - (s32)b > 0;
2203 static u32 seq_max(u32 a, u32 b)
2205 return seq_greater(a, b) ? a : b;
2208 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2210 struct drbd_device *device = peer_device->device;
2211 unsigned int newest_peer_seq;
2213 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2214 spin_lock(&device->peer_seq_lock);
2215 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2216 device->peer_seq = newest_peer_seq;
2217 spin_unlock(&device->peer_seq_lock);
2218 /* wake up only if we actually changed device->peer_seq */
2219 if (peer_seq == newest_peer_seq)
2220 wake_up(&device->seq_wait);
2224 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2226 return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2229 /* maybe change sync_ee into interval trees as well? */
2230 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2232 struct drbd_peer_request *rs_req;
2235 spin_lock_irq(&device->resource->req_lock);
2236 list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2237 if (overlaps(peer_req->i.sector, peer_req->i.size,
2238 rs_req->i.sector, rs_req->i.size)) {
2243 spin_unlock_irq(&device->resource->req_lock);
2248 /* Called from receive_Data.
2249 * Synchronize packets on sock with packets on msock.
2251 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2252 * packet traveling on msock, they are still processed in the order they have
2255 * Note: we don't care for Ack packets overtaking P_DATA packets.
2257 * In case packet_seq is larger than device->peer_seq number, there are
2258 * outstanding packets on the msock. We wait for them to arrive.
2259 * In case we are the logically next packet, we update device->peer_seq
2260 * ourselves. Correctly handles 32bit wrap around.
2262 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2263 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2264 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2265 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2267 * returns 0 if we may process the packet,
2268 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2269 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2271 struct drbd_device *device = peer_device->device;
2276 if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2279 spin_lock(&device->peer_seq_lock);
2281 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2282 device->peer_seq = seq_max(device->peer_seq, peer_seq);
2286 if (signal_pending(current)) {
2292 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2298 /* Only need to wait if two_primaries is enabled */
2299 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2300 spin_unlock(&device->peer_seq_lock);
2302 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2304 timeout = schedule_timeout(timeout);
2305 spin_lock(&device->peer_seq_lock);
2308 drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2312 spin_unlock(&device->peer_seq_lock);
2313 finish_wait(&device->seq_wait, &wait);
2317 /* see also bio_flags_to_wire()
2318 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2319 * flags and back. We may replicate to other kernel versions. */
2320 static unsigned long wire_flags_to_bio_flags(u32 dpf)
2322 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2323 (dpf & DP_FUA ? REQ_FUA : 0) |
2324 (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2327 static unsigned long wire_flags_to_bio_op(u32 dpf)
2329 if (dpf & DP_DISCARD)
2330 return REQ_OP_WRITE_ZEROES;
2332 return REQ_OP_WRITE;
2335 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2338 struct drbd_interval *i;
2341 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2342 struct drbd_request *req;
2343 struct bio_and_error m;
2347 req = container_of(i, struct drbd_request, i);
2348 if (!(req->rq_state & RQ_POSTPONED))
2350 req->rq_state &= ~RQ_POSTPONED;
2351 __req_mod(req, NEG_ACKED, &m);
2352 spin_unlock_irq(&device->resource->req_lock);
2354 complete_master_bio(device, &m);
2355 spin_lock_irq(&device->resource->req_lock);
2360 static int handle_write_conflicts(struct drbd_device *device,
2361 struct drbd_peer_request *peer_req)
2363 struct drbd_connection *connection = peer_req->peer_device->connection;
2364 bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2365 sector_t sector = peer_req->i.sector;
2366 const unsigned int size = peer_req->i.size;
2367 struct drbd_interval *i;
2372 * Inserting the peer request into the write_requests tree will prevent
2373 * new conflicting local requests from being added.
2375 drbd_insert_interval(&device->write_requests, &peer_req->i);
2378 drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2379 if (i == &peer_req->i)
2386 * Our peer has sent a conflicting remote request; this
2387 * should not happen in a two-node setup. Wait for the
2388 * earlier peer request to complete.
2390 err = drbd_wait_misc(device, i);
2396 equal = i->sector == sector && i->size == size;
2397 if (resolve_conflicts) {
2399 * If the peer request is fully contained within the
2400 * overlapping request, it can be considered overwritten
2401 * and thus superseded; otherwise, it will be retried
2402 * once all overlapping requests have completed.
2404 bool superseded = i->sector <= sector && i->sector +
2405 (i->size >> 9) >= sector + (size >> 9);
2408 drbd_alert(device, "Concurrent writes detected: "
2409 "local=%llus +%u, remote=%llus +%u, "
2410 "assuming %s came first\n",
2411 (unsigned long long)i->sector, i->size,
2412 (unsigned long long)sector, size,
2413 superseded ? "local" : "remote");
2415 peer_req->w.cb = superseded ? e_send_superseded :
2417 list_add_tail(&peer_req->w.list, &device->done_ee);
2418 queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2423 struct drbd_request *req =
2424 container_of(i, struct drbd_request, i);
2427 drbd_alert(device, "Concurrent writes detected: "
2428 "local=%llus +%u, remote=%llus +%u\n",
2429 (unsigned long long)i->sector, i->size,
2430 (unsigned long long)sector, size);
2432 if (req->rq_state & RQ_LOCAL_PENDING ||
2433 !(req->rq_state & RQ_POSTPONED)) {
2435 * Wait for the node with the discard flag to
2436 * decide if this request has been superseded
2437 * or needs to be retried.
2438 * Requests that have been superseded will
2439 * disappear from the write_requests tree.
2441 * In addition, wait for the conflicting
2442 * request to finish locally before submitting
2443 * the conflicting peer request.
2445 err = drbd_wait_misc(device, &req->i);
2447 _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2448 fail_postponed_requests(device, sector, size);
2454 * Remember to restart the conflicting requests after
2455 * the new peer request has completed.
2457 peer_req->flags |= EE_RESTART_REQUESTS;
2464 drbd_remove_epoch_entry_interval(device, peer_req);
2468 /* mirrored write */
2469 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2471 struct drbd_peer_device *peer_device;
2472 struct drbd_device *device;
2473 struct net_conf *nc;
2475 struct drbd_peer_request *peer_req;
2476 struct p_data *p = pi->data;
2477 u32 peer_seq = be32_to_cpu(p->seq_num);
2482 peer_device = conn_peer_device(connection, pi->vnr);
2485 device = peer_device->device;
2487 if (!get_ldev(device)) {
2490 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2491 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2492 atomic_inc(&connection->current_epoch->epoch_size);
2493 err2 = drbd_drain_block(peer_device, pi->size);
2500 * Corresponding put_ldev done either below (on various errors), or in
2501 * drbd_peer_request_endio, if we successfully submit the data at the
2502 * end of this function.
2505 sector = be64_to_cpu(p->sector);
2506 peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2512 peer_req->w.cb = e_end_block;
2513 peer_req->submit_jif = jiffies;
2514 peer_req->flags |= EE_APPLICATION;
2516 dp_flags = be32_to_cpu(p->dp_flags);
2517 op = wire_flags_to_bio_op(dp_flags);
2518 op_flags = wire_flags_to_bio_flags(dp_flags);
2519 if (pi->cmd == P_TRIM) {
2520 D_ASSERT(peer_device, peer_req->i.size > 0);
2521 D_ASSERT(peer_device, op == REQ_OP_WRITE_ZEROES);
2522 D_ASSERT(peer_device, peer_req->pages == NULL);
2523 } else if (peer_req->pages == NULL) {
2524 D_ASSERT(device, peer_req->i.size == 0);
2525 D_ASSERT(device, dp_flags & DP_FLUSH);
2528 if (dp_flags & DP_MAY_SET_IN_SYNC)
2529 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2531 spin_lock(&connection->epoch_lock);
2532 peer_req->epoch = connection->current_epoch;
2533 atomic_inc(&peer_req->epoch->epoch_size);
2534 atomic_inc(&peer_req->epoch->active);
2535 spin_unlock(&connection->epoch_lock);
2538 nc = rcu_dereference(peer_device->connection->net_conf);
2539 tp = nc->two_primaries;
2540 if (peer_device->connection->agreed_pro_version < 100) {
2541 switch (nc->wire_protocol) {
2543 dp_flags |= DP_SEND_WRITE_ACK;
2546 dp_flags |= DP_SEND_RECEIVE_ACK;
2552 if (dp_flags & DP_SEND_WRITE_ACK) {
2553 peer_req->flags |= EE_SEND_WRITE_ACK;
2554 inc_unacked(device);
2555 /* corresponding dec_unacked() in e_end_block()
2556 * respective _drbd_clear_done_ee */
2559 if (dp_flags & DP_SEND_RECEIVE_ACK) {
2560 /* I really don't like it that the receiver thread
2561 * sends on the msock, but anyways */
2562 drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2566 /* two primaries implies protocol C */
2567 D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2568 peer_req->flags |= EE_IN_INTERVAL_TREE;
2569 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2571 goto out_interrupted;
2572 spin_lock_irq(&device->resource->req_lock);
2573 err = handle_write_conflicts(device, peer_req);
2575 spin_unlock_irq(&device->resource->req_lock);
2576 if (err == -ENOENT) {
2580 goto out_interrupted;
2583 update_peer_seq(peer_device, peer_seq);
2584 spin_lock_irq(&device->resource->req_lock);
2586 /* TRIM and WRITE_SAME are processed synchronously,
2587 * we wait for all pending requests, respectively wait for
2588 * active_ee to become empty in drbd_submit_peer_request();
2589 * better not add ourselves here. */
2590 if ((peer_req->flags & (EE_IS_TRIM|EE_WRITE_SAME)) == 0)
2591 list_add_tail(&peer_req->w.list, &device->active_ee);
2592 spin_unlock_irq(&device->resource->req_lock);
2594 if (device->state.conn == C_SYNC_TARGET)
2595 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2597 if (device->state.pdsk < D_INCONSISTENT) {
2598 /* In case we have the only disk of the cluster, */
2599 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2600 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2601 drbd_al_begin_io(device, &peer_req->i);
2602 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2605 err = drbd_submit_peer_request(device, peer_req, op, op_flags,
2610 /* don't care for the reason here */
2611 drbd_err(device, "submit failed, triggering re-connect\n");
2612 spin_lock_irq(&device->resource->req_lock);
2613 list_del(&peer_req->w.list);
2614 drbd_remove_epoch_entry_interval(device, peer_req);
2615 spin_unlock_irq(&device->resource->req_lock);
2616 if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2617 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2618 drbd_al_complete_io(device, &peer_req->i);
2622 drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
2624 drbd_free_peer_req(device, peer_req);
2628 /* We may throttle resync, if the lower device seems to be busy,
2629 * and current sync rate is above c_min_rate.
2631 * To decide whether or not the lower device is busy, we use a scheme similar
2632 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2633 * (more than 64 sectors) of activity we cannot account for with our own resync
2634 * activity, it obviously is "busy".
2636 * The current sync rate used here uses only the most recent two step marks,
2637 * to have a short time average so we can react faster.
2639 bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2640 bool throttle_if_app_is_waiting)
2642 struct lc_element *tmp;
2643 bool throttle = drbd_rs_c_min_rate_throttle(device);
2645 if (!throttle || throttle_if_app_is_waiting)
2648 spin_lock_irq(&device->al_lock);
2649 tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2651 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2652 if (test_bit(BME_PRIORITY, &bm_ext->flags))
2654 /* Do not slow down if app IO is already waiting for this extent,
2655 * and our progress is necessary for application IO to complete. */
2657 spin_unlock_irq(&device->al_lock);
2662 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2664 struct gendisk *disk = device->ldev->backing_bdev->bd_contains->bd_disk;
2665 unsigned long db, dt, dbdt;
2666 unsigned int c_min_rate;
2670 c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2673 /* feature disabled? */
2674 if (c_min_rate == 0)
2677 curr_events = (int)part_stat_read_accum(&disk->part0, sectors) -
2678 atomic_read(&device->rs_sect_ev);
2680 if (atomic_read(&device->ap_actlog_cnt)
2681 || curr_events - device->rs_last_events > 64) {
2682 unsigned long rs_left;
2685 device->rs_last_events = curr_events;
2687 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2689 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2691 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2692 rs_left = device->ov_left;
2694 rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2696 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2699 db = device->rs_mark_left[i] - rs_left;
2700 dbdt = Bit2KB(db/dt);
2702 if (dbdt > c_min_rate)
2708 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2710 struct drbd_peer_device *peer_device;
2711 struct drbd_device *device;
2714 struct drbd_peer_request *peer_req;
2715 struct digest_info *di = NULL;
2717 unsigned int fault_type;
2718 struct p_block_req *p = pi->data;
2720 peer_device = conn_peer_device(connection, pi->vnr);
2723 device = peer_device->device;
2724 capacity = drbd_get_capacity(device->this_bdev);
2726 sector = be64_to_cpu(p->sector);
2727 size = be32_to_cpu(p->blksize);
2729 if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2730 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2731 (unsigned long long)sector, size);
2734 if (sector + (size>>9) > capacity) {
2735 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2736 (unsigned long long)sector, size);
2740 if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2743 case P_DATA_REQUEST:
2744 drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2747 case P_RS_DATA_REQUEST:
2748 case P_CSUM_RS_REQUEST:
2750 drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2754 dec_rs_pending(device);
2755 drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2760 if (verb && __ratelimit(&drbd_ratelimit_state))
2761 drbd_err(device, "Can not satisfy peer's read request, "
2762 "no local data.\n");
2764 /* drain possibly payload */
2765 return drbd_drain_block(peer_device, pi->size);
2768 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2769 * "criss-cross" setup, that might cause write-out on some other DRBD,
2770 * which in turn might block on the other node at this very place. */
2771 peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2779 case P_DATA_REQUEST:
2780 peer_req->w.cb = w_e_end_data_req;
2781 fault_type = DRBD_FAULT_DT_RD;
2782 /* application IO, don't drbd_rs_begin_io */
2783 peer_req->flags |= EE_APPLICATION;
2787 /* If at some point in the future we have a smart way to
2788 find out if this data block is completely deallocated,
2789 then we would do something smarter here than reading
2791 peer_req->flags |= EE_RS_THIN_REQ;
2793 case P_RS_DATA_REQUEST:
2794 peer_req->w.cb = w_e_end_rsdata_req;
2795 fault_type = DRBD_FAULT_RS_RD;
2796 /* used in the sector offset progress display */
2797 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2801 case P_CSUM_RS_REQUEST:
2802 fault_type = DRBD_FAULT_RS_RD;
2803 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2807 di->digest_size = pi->size;
2808 di->digest = (((char *)di)+sizeof(struct digest_info));
2810 peer_req->digest = di;
2811 peer_req->flags |= EE_HAS_DIGEST;
2813 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2816 if (pi->cmd == P_CSUM_RS_REQUEST) {
2817 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2818 peer_req->w.cb = w_e_end_csum_rs_req;
2819 /* used in the sector offset progress display */
2820 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2821 /* remember to report stats in drbd_resync_finished */
2822 device->use_csums = true;
2823 } else if (pi->cmd == P_OV_REPLY) {
2824 /* track progress, we may need to throttle */
2825 atomic_add(size >> 9, &device->rs_sect_in);
2826 peer_req->w.cb = w_e_end_ov_reply;
2827 dec_rs_pending(device);
2828 /* drbd_rs_begin_io done when we sent this request,
2829 * but accounting still needs to be done. */
2830 goto submit_for_resync;
2835 if (device->ov_start_sector == ~(sector_t)0 &&
2836 peer_device->connection->agreed_pro_version >= 90) {
2837 unsigned long now = jiffies;
2839 device->ov_start_sector = sector;
2840 device->ov_position = sector;
2841 device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2842 device->rs_total = device->ov_left;
2843 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2844 device->rs_mark_left[i] = device->ov_left;
2845 device->rs_mark_time[i] = now;
2847 drbd_info(device, "Online Verify start sector: %llu\n",
2848 (unsigned long long)sector);
2850 peer_req->w.cb = w_e_end_ov_req;
2851 fault_type = DRBD_FAULT_RS_RD;
2858 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2859 * wrt the receiver, but it is not as straightforward as it may seem.
2860 * Various places in the resync start and stop logic assume resync
2861 * requests are processed in order, requeuing this on the worker thread
2862 * introduces a bunch of new code for synchronization between threads.
2864 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2865 * "forever", throttling after drbd_rs_begin_io will lock that extent
2866 * for application writes for the same time. For now, just throttle
2867 * here, where the rest of the code expects the receiver to sleep for
2871 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2872 * this defers syncer requests for some time, before letting at least
2873 * on request through. The resync controller on the receiving side
2874 * will adapt to the incoming rate accordingly.
2876 * We cannot throttle here if remote is Primary/SyncTarget:
2877 * we would also throttle its application reads.
2878 * In that case, throttling is done on the SyncTarget only.
2881 /* Even though this may be a resync request, we do add to "read_ee";
2882 * "sync_ee" is only used for resync WRITEs.
2883 * Add to list early, so debugfs can find this request
2884 * even if we have to sleep below. */
2885 spin_lock_irq(&device->resource->req_lock);
2886 list_add_tail(&peer_req->w.list, &device->read_ee);
2887 spin_unlock_irq(&device->resource->req_lock);
2889 update_receiver_timing_details(connection, drbd_rs_should_slow_down);
2890 if (device->state.peer != R_PRIMARY
2891 && drbd_rs_should_slow_down(device, sector, false))
2892 schedule_timeout_uninterruptible(HZ/10);
2893 update_receiver_timing_details(connection, drbd_rs_begin_io);
2894 if (drbd_rs_begin_io(device, sector))
2898 atomic_add(size >> 9, &device->rs_sect_ev);
2901 update_receiver_timing_details(connection, drbd_submit_peer_request);
2902 inc_unacked(device);
2903 if (drbd_submit_peer_request(device, peer_req, REQ_OP_READ, 0,
2907 /* don't care for the reason here */
2908 drbd_err(device, "submit failed, triggering re-connect\n");
2911 spin_lock_irq(&device->resource->req_lock);
2912 list_del(&peer_req->w.list);
2913 spin_unlock_irq(&device->resource->req_lock);
2914 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2917 drbd_free_peer_req(device, peer_req);
2922 * drbd_asb_recover_0p - Recover after split-brain with no remaining primaries
2924 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
2926 struct drbd_device *device = peer_device->device;
2927 int self, peer, rv = -100;
2928 unsigned long ch_self, ch_peer;
2929 enum drbd_after_sb_p after_sb_0p;
2931 self = device->ldev->md.uuid[UI_BITMAP] & 1;
2932 peer = device->p_uuid[UI_BITMAP] & 1;
2934 ch_peer = device->p_uuid[UI_SIZE];
2935 ch_self = device->comm_bm_set;
2938 after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
2940 switch (after_sb_0p) {
2942 case ASB_DISCARD_SECONDARY:
2943 case ASB_CALL_HELPER:
2945 drbd_err(device, "Configuration error.\n");
2947 case ASB_DISCONNECT:
2949 case ASB_DISCARD_YOUNGER_PRI:
2950 if (self == 0 && peer == 1) {
2954 if (self == 1 && peer == 0) {
2958 /* Else fall through to one of the other strategies... */
2959 case ASB_DISCARD_OLDER_PRI:
2960 if (self == 0 && peer == 1) {
2964 if (self == 1 && peer == 0) {
2968 /* Else fall through to one of the other strategies... */
2969 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
2970 "Using discard-least-changes instead\n");
2972 case ASB_DISCARD_ZERO_CHG:
2973 if (ch_peer == 0 && ch_self == 0) {
2974 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2978 if (ch_peer == 0) { rv = 1; break; }
2979 if (ch_self == 0) { rv = -1; break; }
2981 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
2983 /* else: fall through */
2984 case ASB_DISCARD_LEAST_CHG:
2985 if (ch_self < ch_peer)
2987 else if (ch_self > ch_peer)
2989 else /* ( ch_self == ch_peer ) */
2990 /* Well, then use something else. */
2991 rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2994 case ASB_DISCARD_LOCAL:
2997 case ASB_DISCARD_REMOTE:
3005 * drbd_asb_recover_1p - Recover after split-brain with one remaining primary
3007 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
3009 struct drbd_device *device = peer_device->device;
3011 enum drbd_after_sb_p after_sb_1p;
3014 after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
3016 switch (after_sb_1p) {
3017 case ASB_DISCARD_YOUNGER_PRI:
3018 case ASB_DISCARD_OLDER_PRI:
3019 case ASB_DISCARD_LEAST_CHG:
3020 case ASB_DISCARD_LOCAL:
3021 case ASB_DISCARD_REMOTE:
3022 case ASB_DISCARD_ZERO_CHG:
3023 drbd_err(device, "Configuration error.\n");
3025 case ASB_DISCONNECT:
3028 hg = drbd_asb_recover_0p(peer_device);
3029 if (hg == -1 && device->state.role == R_SECONDARY)
3031 if (hg == 1 && device->state.role == R_PRIMARY)
3035 rv = drbd_asb_recover_0p(peer_device);
3037 case ASB_DISCARD_SECONDARY:
3038 return device->state.role == R_PRIMARY ? 1 : -1;
3039 case ASB_CALL_HELPER:
3040 hg = drbd_asb_recover_0p(peer_device);
3041 if (hg == -1 && device->state.role == R_PRIMARY) {
3042 enum drbd_state_rv rv2;
3044 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3045 * we might be here in C_WF_REPORT_PARAMS which is transient.
3046 * we do not need to wait for the after state change work either. */
3047 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3048 if (rv2 != SS_SUCCESS) {
3049 drbd_khelper(device, "pri-lost-after-sb");
3051 drbd_warn(device, "Successfully gave up primary role.\n");
3062 * drbd_asb_recover_2p - Recover after split-brain with two remaining primaries
3064 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
3066 struct drbd_device *device = peer_device->device;
3068 enum drbd_after_sb_p after_sb_2p;
3071 after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
3073 switch (after_sb_2p) {
3074 case ASB_DISCARD_YOUNGER_PRI:
3075 case ASB_DISCARD_OLDER_PRI:
3076 case ASB_DISCARD_LEAST_CHG:
3077 case ASB_DISCARD_LOCAL:
3078 case ASB_DISCARD_REMOTE:
3080 case ASB_DISCARD_SECONDARY:
3081 case ASB_DISCARD_ZERO_CHG:
3082 drbd_err(device, "Configuration error.\n");
3085 rv = drbd_asb_recover_0p(peer_device);
3087 case ASB_DISCONNECT:
3089 case ASB_CALL_HELPER:
3090 hg = drbd_asb_recover_0p(peer_device);
3092 enum drbd_state_rv rv2;
3094 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3095 * we might be here in C_WF_REPORT_PARAMS which is transient.
3096 * we do not need to wait for the after state change work either. */
3097 rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3098 if (rv2 != SS_SUCCESS) {
3099 drbd_khelper(device, "pri-lost-after-sb");
3101 drbd_warn(device, "Successfully gave up primary role.\n");
3111 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
3112 u64 bits, u64 flags)
3115 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
3118 drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
3120 (unsigned long long)uuid[UI_CURRENT],
3121 (unsigned long long)uuid[UI_BITMAP],
3122 (unsigned long long)uuid[UI_HISTORY_START],
3123 (unsigned long long)uuid[UI_HISTORY_END],
3124 (unsigned long long)bits,
3125 (unsigned long long)flags);
3129 100 after split brain try auto recover
3130 2 C_SYNC_SOURCE set BitMap
3131 1 C_SYNC_SOURCE use BitMap
3133 -1 C_SYNC_TARGET use BitMap
3134 -2 C_SYNC_TARGET set BitMap
3135 -100 after split brain, disconnect
3136 -1000 unrelated data
3137 -1091 requires proto 91
3138 -1096 requires proto 96
3141 static int drbd_uuid_compare(struct drbd_device *const device, enum drbd_role const peer_role, int *rule_nr) __must_hold(local)
3143 struct drbd_peer_device *const peer_device = first_peer_device(device);
3144 struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
3148 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3149 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3152 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
3156 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
3157 peer != UUID_JUST_CREATED)
3161 if (self != UUID_JUST_CREATED &&
3162 (peer == UUID_JUST_CREATED || peer == (u64)0))
3166 int rct, dc; /* roles at crash time */
3168 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3170 if (connection->agreed_pro_version < 91)
3173 if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3174 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3175 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3176 drbd_uuid_move_history(device);
3177 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3178 device->ldev->md.uuid[UI_BITMAP] = 0;
3180 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3181 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3184 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3191 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3193 if (connection->agreed_pro_version < 91)
3196 if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3197 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3198 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3200 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3201 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3202 device->p_uuid[UI_BITMAP] = 0UL;
3204 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3207 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3214 /* Common power [off|failure] */
3215 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3216 (device->p_uuid[UI_FLAGS] & 2);
3217 /* lowest bit is set when we were primary,
3218 * next bit (weight 2) is set when peer was primary */
3221 /* Neither has the "crashed primary" flag set,
3222 * only a replication link hickup. */
3226 /* Current UUID equal and no bitmap uuid; does not necessarily
3227 * mean this was a "simultaneous hard crash", maybe IO was
3228 * frozen, so no UUID-bump happened.
3229 * This is a protocol change, overload DRBD_FF_WSAME as flag
3230 * for "new-enough" peer DRBD version. */
3231 if (device->state.role == R_PRIMARY || peer_role == R_PRIMARY) {
3233 if (!(connection->agreed_features & DRBD_FF_WSAME)) {
3234 drbd_warn(peer_device, "Equivalent unrotated UUIDs, but current primary present.\n");
3235 return -(0x10000 | PRO_VERSION_MAX | (DRBD_FF_WSAME << 8));
3237 if (device->state.role == R_PRIMARY && peer_role == R_PRIMARY) {
3238 /* At least one has the "crashed primary" bit set,
3239 * both are primary now, but neither has rotated its UUIDs?
3240 * "Can not happen." */
3241 drbd_err(peer_device, "Equivalent unrotated UUIDs, but both are primary. Can not resolve this.\n");
3244 if (device->state.role == R_PRIMARY)
3249 /* Both are secondary.
3250 * Really looks like recovery from simultaneous hard crash.
3251 * Check which had been primary before, and arbitrate. */
3253 case 0: /* !self_pri && !peer_pri */ return 0; /* already handled */
3254 case 1: /* self_pri && !peer_pri */ return 1;
3255 case 2: /* !self_pri && peer_pri */ return -1;
3256 case 3: /* self_pri && peer_pri */
3257 dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3263 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3268 peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3270 if (connection->agreed_pro_version < 96 ?
3271 (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3272 (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3273 peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3274 /* The last P_SYNC_UUID did not get though. Undo the last start of
3275 resync as sync source modifications of the peer's UUIDs. */
3277 if (connection->agreed_pro_version < 91)
3280 device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3281 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3283 drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3284 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3291 self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3292 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3293 peer = device->p_uuid[i] & ~((u64)1);
3299 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3300 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3305 self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3307 if (connection->agreed_pro_version < 96 ?
3308 (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3309 (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3310 self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3311 /* The last P_SYNC_UUID did not get though. Undo the last start of
3312 resync as sync source modifications of our UUIDs. */
3314 if (connection->agreed_pro_version < 91)
3317 __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3318 __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3320 drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3321 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3322 device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3330 peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3331 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3332 self = device->ldev->md.uuid[i] & ~((u64)1);
3338 self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3339 peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3340 if (self == peer && self != ((u64)0))
3344 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3345 self = device->ldev->md.uuid[i] & ~((u64)1);
3346 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3347 peer = device->p_uuid[j] & ~((u64)1);
3356 /* drbd_sync_handshake() returns the new conn state on success, or
3357 CONN_MASK (-1) on failure.
3359 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3360 enum drbd_role peer_role,
3361 enum drbd_disk_state peer_disk) __must_hold(local)
3363 struct drbd_device *device = peer_device->device;
3364 enum drbd_conns rv = C_MASK;
3365 enum drbd_disk_state mydisk;
3366 struct net_conf *nc;
3367 int hg, rule_nr, rr_conflict, tentative, always_asbp;
3369 mydisk = device->state.disk;
3370 if (mydisk == D_NEGOTIATING)
3371 mydisk = device->new_state_tmp.disk;
3373 drbd_info(device, "drbd_sync_handshake:\n");
3375 spin_lock_irq(&device->ldev->md.uuid_lock);
3376 drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3377 drbd_uuid_dump(device, "peer", device->p_uuid,
3378 device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3380 hg = drbd_uuid_compare(device, peer_role, &rule_nr);
3381 spin_unlock_irq(&device->ldev->md.uuid_lock);
3383 drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3386 drbd_alert(device, "Unrelated data, aborting!\n");
3389 if (hg < -0x10000) {
3393 fflags = (hg >> 8) & 0xff;
3394 drbd_alert(device, "To resolve this both sides have to support at least protocol %d and feature flags 0x%x\n",
3399 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3403 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3404 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
3405 int f = (hg == -100) || abs(hg) == 2;
3406 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3409 drbd_info(device, "Becoming sync %s due to disk states.\n",
3410 hg > 0 ? "source" : "target");
3414 drbd_khelper(device, "initial-split-brain");
3417 nc = rcu_dereference(peer_device->connection->net_conf);
3418 always_asbp = nc->always_asbp;
3419 rr_conflict = nc->rr_conflict;
3420 tentative = nc->tentative;
3423 if (hg == 100 || (hg == -100 && always_asbp)) {
3424 int pcount = (device->state.role == R_PRIMARY)
3425 + (peer_role == R_PRIMARY);
3426 int forced = (hg == -100);
3430 hg = drbd_asb_recover_0p(peer_device);
3433 hg = drbd_asb_recover_1p(peer_device);
3436 hg = drbd_asb_recover_2p(peer_device);
3439 if (abs(hg) < 100) {
3440 drbd_warn(device, "Split-Brain detected, %d primaries, "
3441 "automatically solved. Sync from %s node\n",
3442 pcount, (hg < 0) ? "peer" : "this");
3444 drbd_warn(device, "Doing a full sync, since"
3445 " UUIDs where ambiguous.\n");
3452 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3454 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3458 drbd_warn(device, "Split-Brain detected, manually solved. "
3459 "Sync from %s node\n",
3460 (hg < 0) ? "peer" : "this");
3464 /* FIXME this log message is not correct if we end up here
3465 * after an attempted attach on a diskless node.
3466 * We just refuse to attach -- well, we drop the "connection"
3467 * to that disk, in a way... */
3468 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3469 drbd_khelper(device, "split-brain");
3473 if (hg > 0 && mydisk <= D_INCONSISTENT) {
3474 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3478 if (hg < 0 && /* by intention we do not use mydisk here. */
3479 device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3480 switch (rr_conflict) {
3481 case ASB_CALL_HELPER:
3482 drbd_khelper(device, "pri-lost");
3484 case ASB_DISCONNECT:
3485 drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3488 drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3493 if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3495 drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3497 drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3498 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3499 abs(hg) >= 2 ? "full" : "bit-map based");
3504 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3505 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3506 BM_LOCKED_SET_ALLOWED))
3510 if (hg > 0) { /* become sync source. */
3512 } else if (hg < 0) { /* become sync target */
3516 if (drbd_bm_total_weight(device)) {
3517 drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3518 drbd_bm_total_weight(device));
3525 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3527 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3528 if (peer == ASB_DISCARD_REMOTE)
3529 return ASB_DISCARD_LOCAL;
3531 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3532 if (peer == ASB_DISCARD_LOCAL)
3533 return ASB_DISCARD_REMOTE;
3535 /* everything else is valid if they are equal on both sides. */
3539 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3541 struct p_protocol *p = pi->data;
3542 enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3543 int p_proto, p_discard_my_data, p_two_primaries, cf;
3544 struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3545 char integrity_alg[SHARED_SECRET_MAX] = "";
3546 struct crypto_shash *peer_integrity_tfm = NULL;
3547 void *int_dig_in = NULL, *int_dig_vv = NULL;
3549 p_proto = be32_to_cpu(p->protocol);
3550 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
3551 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
3552 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
3553 p_two_primaries = be32_to_cpu(p->two_primaries);
3554 cf = be32_to_cpu(p->conn_flags);
3555 p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3557 if (connection->agreed_pro_version >= 87) {
3560 if (pi->size > sizeof(integrity_alg))
3562 err = drbd_recv_all(connection, integrity_alg, pi->size);
3565 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3568 if (pi->cmd != P_PROTOCOL_UPDATE) {
3569 clear_bit(CONN_DRY_RUN, &connection->flags);
3571 if (cf & CF_DRY_RUN)
3572 set_bit(CONN_DRY_RUN, &connection->flags);
3575 nc = rcu_dereference(connection->net_conf);
3577 if (p_proto != nc->wire_protocol) {
3578 drbd_err(connection, "incompatible %s settings\n", "protocol");
3579 goto disconnect_rcu_unlock;
3582 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3583 drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3584 goto disconnect_rcu_unlock;
3587 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3588 drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3589 goto disconnect_rcu_unlock;
3592 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3593 drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3594 goto disconnect_rcu_unlock;
3597 if (p_discard_my_data && nc->discard_my_data) {
3598 drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3599 goto disconnect_rcu_unlock;
3602 if (p_two_primaries != nc->two_primaries) {
3603 drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3604 goto disconnect_rcu_unlock;
3607 if (strcmp(integrity_alg, nc->integrity_alg)) {
3608 drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3609 goto disconnect_rcu_unlock;
3615 if (integrity_alg[0]) {
3619 * We can only change the peer data integrity algorithm
3620 * here. Changing our own data integrity algorithm
3621 * requires that we send a P_PROTOCOL_UPDATE packet at
3622 * the same time; otherwise, the peer has no way to
3623 * tell between which packets the algorithm should
3627 peer_integrity_tfm = crypto_alloc_shash(integrity_alg, 0, CRYPTO_ALG_ASYNC);
3628 if (IS_ERR(peer_integrity_tfm)) {
3629 peer_integrity_tfm = NULL;
3630 drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3635 hash_size = crypto_shash_digestsize(peer_integrity_tfm);
3636 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3637 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3638 if (!(int_dig_in && int_dig_vv)) {
3639 drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3644 new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3645 if (!new_net_conf) {
3646 drbd_err(connection, "Allocation of new net_conf failed\n");
3650 mutex_lock(&connection->data.mutex);
3651 mutex_lock(&connection->resource->conf_update);
3652 old_net_conf = connection->net_conf;
3653 *new_net_conf = *old_net_conf;
3655 new_net_conf->wire_protocol = p_proto;
3656 new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3657 new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3658 new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3659 new_net_conf->two_primaries = p_two_primaries;
3661 rcu_assign_pointer(connection->net_conf, new_net_conf);
3662 mutex_unlock(&connection->resource->conf_update);
3663 mutex_unlock(&connection->data.mutex);
3665 crypto_free_shash(connection->peer_integrity_tfm);
3666 kfree(connection->int_dig_in);
3667 kfree(connection->int_dig_vv);
3668 connection->peer_integrity_tfm = peer_integrity_tfm;
3669 connection->int_dig_in = int_dig_in;
3670 connection->int_dig_vv = int_dig_vv;
3672 if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3673 drbd_info(connection, "peer data-integrity-alg: %s\n",
3674 integrity_alg[0] ? integrity_alg : "(none)");
3677 kfree(old_net_conf);
3680 disconnect_rcu_unlock:
3683 crypto_free_shash(peer_integrity_tfm);
3686 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3691 * input: alg name, feature name
3692 * return: NULL (alg name was "")
3693 * ERR_PTR(error) if something goes wrong
3694 * or the crypto hash ptr, if it worked out ok. */
3695 static struct crypto_shash *drbd_crypto_alloc_digest_safe(
3696 const struct drbd_device *device,
3697 const char *alg, const char *name)
3699 struct crypto_shash *tfm;
3704 tfm = crypto_alloc_shash(alg, 0, 0);
3706 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3707 alg, name, PTR_ERR(tfm));
3713 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3715 void *buffer = connection->data.rbuf;
3716 int size = pi->size;
3719 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3720 s = drbd_recv(connection, buffer, s);
3734 * config_unknown_volume - device configuration command for unknown volume
3736 * When a device is added to an existing connection, the node on which the
3737 * device is added first will send configuration commands to its peer but the
3738 * peer will not know about the device yet. It will warn and ignore these
3739 * commands. Once the device is added on the second node, the second node will
3740 * send the same device configuration commands, but in the other direction.
3742 * (We can also end up here if drbd is misconfigured.)
3744 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3746 drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3747 cmdname(pi->cmd), pi->vnr);
3748 return ignore_remaining_packet(connection, pi);
3751 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3753 struct drbd_peer_device *peer_device;
3754 struct drbd_device *device;
3755 struct p_rs_param_95 *p;
3756 unsigned int header_size, data_size, exp_max_sz;
3757 struct crypto_shash *verify_tfm = NULL;
3758 struct crypto_shash *csums_tfm = NULL;
3759 struct net_conf *old_net_conf, *new_net_conf = NULL;
3760 struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3761 const int apv = connection->agreed_pro_version;
3762 struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3766 peer_device = conn_peer_device(connection, pi->vnr);
3768 return config_unknown_volume(connection, pi);
3769 device = peer_device->device;
3771 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
3772 : apv == 88 ? sizeof(struct p_rs_param)
3774 : apv <= 94 ? sizeof(struct p_rs_param_89)
3775 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3777 if (pi->size > exp_max_sz) {
3778 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3779 pi->size, exp_max_sz);
3784 header_size = sizeof(struct p_rs_param);
3785 data_size = pi->size - header_size;
3786 } else if (apv <= 94) {
3787 header_size = sizeof(struct p_rs_param_89);
3788 data_size = pi->size - header_size;
3789 D_ASSERT(device, data_size == 0);
3791 header_size = sizeof(struct p_rs_param_95);
3792 data_size = pi->size - header_size;
3793 D_ASSERT(device, data_size == 0);
3796 /* initialize verify_alg and csums_alg */
3798 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
3800 err = drbd_recv_all(peer_device->connection, p, header_size);
3804 mutex_lock(&connection->resource->conf_update);
3805 old_net_conf = peer_device->connection->net_conf;
3806 if (get_ldev(device)) {
3807 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3808 if (!new_disk_conf) {
3810 mutex_unlock(&connection->resource->conf_update);
3811 drbd_err(device, "Allocation of new disk_conf failed\n");
3815 old_disk_conf = device->ldev->disk_conf;
3816 *new_disk_conf = *old_disk_conf;
3818 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3823 if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3824 drbd_err(device, "verify-alg of wrong size, "
3825 "peer wants %u, accepting only up to %u byte\n",
3826 data_size, SHARED_SECRET_MAX);
3831 err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3834 /* we expect NUL terminated string */
3835 /* but just in case someone tries to be evil */
3836 D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3837 p->verify_alg[data_size-1] = 0;
3839 } else /* apv >= 89 */ {
3840 /* we still expect NUL terminated strings */
3841 /* but just in case someone tries to be evil */
3842 D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3843 D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3844 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3845 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3848 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3849 if (device->state.conn == C_WF_REPORT_PARAMS) {
3850 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3851 old_net_conf->verify_alg, p->verify_alg);
3854 verify_tfm = drbd_crypto_alloc_digest_safe(device,
3855 p->verify_alg, "verify-alg");
3856 if (IS_ERR(verify_tfm)) {
3862 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3863 if (device->state.conn == C_WF_REPORT_PARAMS) {
3864 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3865 old_net_conf->csums_alg, p->csums_alg);
3868 csums_tfm = drbd_crypto_alloc_digest_safe(device,
3869 p->csums_alg, "csums-alg");
3870 if (IS_ERR(csums_tfm)) {
3876 if (apv > 94 && new_disk_conf) {
3877 new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3878 new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3879 new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3880 new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3882 fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3883 if (fifo_size != device->rs_plan_s->size) {
3884 new_plan = fifo_alloc(fifo_size);
3886 drbd_err(device, "kmalloc of fifo_buffer failed");
3893 if (verify_tfm || csums_tfm) {
3894 new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3895 if (!new_net_conf) {
3896 drbd_err(device, "Allocation of new net_conf failed\n");
3900 *new_net_conf = *old_net_conf;
3903 strcpy(new_net_conf->verify_alg, p->verify_alg);
3904 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3905 crypto_free_shash(peer_device->connection->verify_tfm);
3906 peer_device->connection->verify_tfm = verify_tfm;
3907 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3910 strcpy(new_net_conf->csums_alg, p->csums_alg);
3911 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3912 crypto_free_shash(peer_device->connection->csums_tfm);
3913 peer_device->connection->csums_tfm = csums_tfm;
3914 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
3916 rcu_assign_pointer(connection->net_conf, new_net_conf);
3920 if (new_disk_conf) {
3921 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3926 old_plan = device->rs_plan_s;
3927 rcu_assign_pointer(device->rs_plan_s, new_plan);
3930 mutex_unlock(&connection->resource->conf_update);
3933 kfree(old_net_conf);
3934 kfree(old_disk_conf);
3940 if (new_disk_conf) {
3942 kfree(new_disk_conf);
3944 mutex_unlock(&connection->resource->conf_update);
3949 if (new_disk_conf) {
3951 kfree(new_disk_conf);
3953 mutex_unlock(&connection->resource->conf_update);
3954 /* just for completeness: actually not needed,
3955 * as this is not reached if csums_tfm was ok. */
3956 crypto_free_shash(csums_tfm);
3957 /* but free the verify_tfm again, if csums_tfm did not work out */
3958 crypto_free_shash(verify_tfm);
3959 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3963 /* warn if the arguments differ by more than 12.5% */
3964 static void warn_if_differ_considerably(struct drbd_device *device,
3965 const char *s, sector_t a, sector_t b)
3968 if (a == 0 || b == 0)
3970 d = (a > b) ? (a - b) : (b - a);
3971 if (d > (a>>3) || d > (b>>3))
3972 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
3973 (unsigned long long)a, (unsigned long long)b);
3976 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
3978 struct drbd_peer_device *peer_device;
3979 struct drbd_device *device;
3980 struct p_sizes *p = pi->data;
3981 struct o_qlim *o = (connection->agreed_features & DRBD_FF_WSAME) ? p->qlim : NULL;
3982 enum determine_dev_size dd = DS_UNCHANGED;
3983 sector_t p_size, p_usize, p_csize, my_usize;
3984 sector_t new_size, cur_size;
3985 int ldsc = 0; /* local disk size changed */
3986 enum dds_flags ddsf;
3988 peer_device = conn_peer_device(connection, pi->vnr);
3990 return config_unknown_volume(connection, pi);
3991 device = peer_device->device;
3992 cur_size = drbd_get_capacity(device->this_bdev);
3994 p_size = be64_to_cpu(p->d_size);
3995 p_usize = be64_to_cpu(p->u_size);
3996 p_csize = be64_to_cpu(p->c_size);
3998 /* just store the peer's disk size for now.
3999 * we still need to figure out whether we accept that. */
4000 device->p_size = p_size;
4002 if (get_ldev(device)) {
4004 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
4007 warn_if_differ_considerably(device, "lower level device sizes",
4008 p_size, drbd_get_max_capacity(device->ldev));
4009 warn_if_differ_considerably(device, "user requested size",
4012 /* if this is the first connect, or an otherwise expected
4013 * param exchange, choose the minimum */
4014 if (device->state.conn == C_WF_REPORT_PARAMS)
4015 p_usize = min_not_zero(my_usize, p_usize);
4017 /* Never shrink a device with usable data during connect,
4018 * or "attach" on the peer.
4019 * But allow online shrinking if we are connected. */
4020 new_size = drbd_new_dev_size(device, device->ldev, p_usize, 0);
4021 if (new_size < cur_size &&
4022 device->state.disk >= D_OUTDATED &&
4023 (device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS)) {
4024 drbd_err(device, "The peer's disk size is too small! (%llu < %llu sectors)\n",
4025 (unsigned long long)new_size, (unsigned long long)cur_size);
4026 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4031 if (my_usize != p_usize) {
4032 struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
4034 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
4035 if (!new_disk_conf) {
4036 drbd_err(device, "Allocation of new disk_conf failed\n");
4041 mutex_lock(&connection->resource->conf_update);
4042 old_disk_conf = device->ldev->disk_conf;
4043 *new_disk_conf = *old_disk_conf;
4044 new_disk_conf->disk_size = p_usize;
4046 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
4047 mutex_unlock(&connection->resource->conf_update);
4049 kfree(old_disk_conf);
4051 drbd_info(device, "Peer sets u_size to %lu sectors (old: %lu)\n",
4052 (unsigned long)p_usize, (unsigned long)my_usize);
4058 device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
4059 /* Leave drbd_reconsider_queue_parameters() before drbd_determine_dev_size().
4060 In case we cleared the QUEUE_FLAG_DISCARD from our queue in
4061 drbd_reconsider_queue_parameters(), we can be sure that after
4062 drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
4064 ddsf = be16_to_cpu(p->dds_flags);
4065 if (get_ldev(device)) {
4066 drbd_reconsider_queue_parameters(device, device->ldev, o);
4067 dd = drbd_determine_dev_size(device, ddsf, NULL);
4071 drbd_md_sync(device);
4074 * I am diskless, need to accept the peer's *current* size.
4075 * I must NOT accept the peers backing disk size,
4076 * it may have been larger than mine all along...
4078 * At this point, the peer knows more about my disk, or at
4079 * least about what we last agreed upon, than myself.
4080 * So if his c_size is less than his d_size, the most likely
4081 * reason is that *my* d_size was smaller last time we checked.
4083 * However, if he sends a zero current size,
4084 * take his (user-capped or) backing disk size anyways.
4086 * Unless of course he does not have a disk himself.
4087 * In which case we ignore this completely.
4089 sector_t new_size = p_csize ?: p_usize ?: p_size;
4090 drbd_reconsider_queue_parameters(device, NULL, o);
4091 if (new_size == 0) {
4092 /* Ignore, peer does not know nothing. */
4093 } else if (new_size == cur_size) {
4095 } else if (cur_size != 0 && p_size == 0) {
4096 drbd_warn(device, "Ignored diskless peer device size (peer:%llu != me:%llu sectors)!\n",
4097 (unsigned long long)new_size, (unsigned long long)cur_size);
4098 } else if (new_size < cur_size && device->state.role == R_PRIMARY) {
4099 drbd_err(device, "The peer's device size is too small! (%llu < %llu sectors); demote me first!\n",
4100 (unsigned long long)new_size, (unsigned long long)cur_size);
4101 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4104 /* I believe the peer, if
4105 * - I don't have a current size myself
4106 * - we agree on the size anyways
4107 * - I do have a current size, am Secondary,
4108 * and he has the only disk
4109 * - I do have a current size, am Primary,
4110 * and he has the only disk,
4111 * which is larger than my current size
4113 drbd_set_my_capacity(device, new_size);
4117 if (get_ldev(device)) {
4118 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
4119 device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
4126 if (device->state.conn > C_WF_REPORT_PARAMS) {
4127 if (be64_to_cpu(p->c_size) !=
4128 drbd_get_capacity(device->this_bdev) || ldsc) {
4129 /* we have different sizes, probably peer
4130 * needs to know my new size... */
4131 drbd_send_sizes(peer_device, 0, ddsf);
4133 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
4134 (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
4135 if (device->state.pdsk >= D_INCONSISTENT &&
4136 device->state.disk >= D_INCONSISTENT) {
4137 if (ddsf & DDSF_NO_RESYNC)
4138 drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
4140 resync_after_online_grow(device);
4142 set_bit(RESYNC_AFTER_NEG, &device->flags);
4149 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
4151 struct drbd_peer_device *peer_device;
4152 struct drbd_device *device;
4153 struct p_uuids *p = pi->data;
4155 int i, updated_uuids = 0;
4157 peer_device = conn_peer_device(connection, pi->vnr);
4159 return config_unknown_volume(connection, pi);
4160 device = peer_device->device;
4162 p_uuid = kmalloc_array(UI_EXTENDED_SIZE, sizeof(*p_uuid), GFP_NOIO);
4164 drbd_err(device, "kmalloc of p_uuid failed\n");
4168 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
4169 p_uuid[i] = be64_to_cpu(p->uuid[i]);
4171 kfree(device->p_uuid);
4172 device->p_uuid = p_uuid;
4174 if ((device->state.conn < C_CONNECTED || device->state.pdsk == D_DISKLESS) &&
4175 device->state.disk < D_INCONSISTENT &&
4176 device->state.role == R_PRIMARY &&
4177 (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
4178 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
4179 (unsigned long long)device->ed_uuid);
4180 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4184 if (get_ldev(device)) {
4185 int skip_initial_sync =
4186 device->state.conn == C_CONNECTED &&
4187 peer_device->connection->agreed_pro_version >= 90 &&
4188 device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
4189 (p_uuid[UI_FLAGS] & 8);
4190 if (skip_initial_sync) {
4191 drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
4192 drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
4193 "clear_n_write from receive_uuids",
4194 BM_LOCKED_TEST_ALLOWED);
4195 _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
4196 _drbd_uuid_set(device, UI_BITMAP, 0);
4197 _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
4199 drbd_md_sync(device);
4203 } else if (device->state.disk < D_INCONSISTENT &&
4204 device->state.role == R_PRIMARY) {
4205 /* I am a diskless primary, the peer just created a new current UUID
4207 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4210 /* Before we test for the disk state, we should wait until an eventually
4211 ongoing cluster wide state change is finished. That is important if
4212 we are primary and are detaching from our disk. We need to see the
4213 new disk state... */
4214 mutex_lock(device->state_mutex);
4215 mutex_unlock(device->state_mutex);
4216 if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
4217 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
4220 drbd_print_uuids(device, "receiver updated UUIDs to");
4226 * convert_state() - Converts the peer's view of the cluster state to our point of view
4227 * @ps: The state as seen by the peer.
4229 static union drbd_state convert_state(union drbd_state ps)
4231 union drbd_state ms;
4233 static enum drbd_conns c_tab[] = {
4234 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
4235 [C_CONNECTED] = C_CONNECTED,
4237 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
4238 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
4239 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
4240 [C_VERIFY_S] = C_VERIFY_T,
4246 ms.conn = c_tab[ps.conn];
4251 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
4256 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4258 struct drbd_peer_device *peer_device;
4259 struct drbd_device *device;
4260 struct p_req_state *p = pi->data;
4261 union drbd_state mask, val;
4262 enum drbd_state_rv rv;
4264 peer_device = conn_peer_device(connection, pi->vnr);
4267 device = peer_device->device;
4269 mask.i = be32_to_cpu(p->mask);
4270 val.i = be32_to_cpu(p->val);
4272 if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4273 mutex_is_locked(device->state_mutex)) {
4274 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4278 mask = convert_state(mask);
4279 val = convert_state(val);
4281 rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4282 drbd_send_sr_reply(peer_device, rv);
4284 drbd_md_sync(device);
4289 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4291 struct p_req_state *p = pi->data;
4292 union drbd_state mask, val;
4293 enum drbd_state_rv rv;
4295 mask.i = be32_to_cpu(p->mask);
4296 val.i = be32_to_cpu(p->val);
4298 if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4299 mutex_is_locked(&connection->cstate_mutex)) {
4300 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4304 mask = convert_state(mask);
4305 val = convert_state(val);
4307 rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4308 conn_send_sr_reply(connection, rv);
4313 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4315 struct drbd_peer_device *peer_device;
4316 struct drbd_device *device;
4317 struct p_state *p = pi->data;
4318 union drbd_state os, ns, peer_state;
4319 enum drbd_disk_state real_peer_disk;
4320 enum chg_state_flags cs_flags;
4323 peer_device = conn_peer_device(connection, pi->vnr);
4325 return config_unknown_volume(connection, pi);
4326 device = peer_device->device;
4328 peer_state.i = be32_to_cpu(p->state);
4330 real_peer_disk = peer_state.disk;
4331 if (peer_state.disk == D_NEGOTIATING) {
4332 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4333 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4336 spin_lock_irq(&device->resource->req_lock);
4338 os = ns = drbd_read_state(device);
4339 spin_unlock_irq(&device->resource->req_lock);
4341 /* If some other part of the code (ack_receiver thread, timeout)
4342 * already decided to close the connection again,
4343 * we must not "re-establish" it here. */
4344 if (os.conn <= C_TEAR_DOWN)
4347 /* If this is the "end of sync" confirmation, usually the peer disk
4348 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4349 * set) resync started in PausedSyncT, or if the timing of pause-/
4350 * unpause-sync events has been "just right", the peer disk may
4351 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4353 if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4354 real_peer_disk == D_UP_TO_DATE &&
4355 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4356 /* If we are (becoming) SyncSource, but peer is still in sync
4357 * preparation, ignore its uptodate-ness to avoid flapping, it
4358 * will change to inconsistent once the peer reaches active
4360 * It may have changed syncer-paused flags, however, so we
4361 * cannot ignore this completely. */
4362 if (peer_state.conn > C_CONNECTED &&
4363 peer_state.conn < C_SYNC_SOURCE)
4364 real_peer_disk = D_INCONSISTENT;
4366 /* if peer_state changes to connected at the same time,
4367 * it explicitly notifies us that it finished resync.
4368 * Maybe we should finish it up, too? */
4369 else if (os.conn >= C_SYNC_SOURCE &&
4370 peer_state.conn == C_CONNECTED) {
4371 if (drbd_bm_total_weight(device) <= device->rs_failed)
4372 drbd_resync_finished(device);
4377 /* explicit verify finished notification, stop sector reached. */
4378 if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4379 peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4380 ov_out_of_sync_print(device);
4381 drbd_resync_finished(device);
4385 /* peer says his disk is inconsistent, while we think it is uptodate,
4386 * and this happens while the peer still thinks we have a sync going on,
4387 * but we think we are already done with the sync.
4388 * We ignore this to avoid flapping pdsk.
4389 * This should not happen, if the peer is a recent version of drbd. */
4390 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4391 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4392 real_peer_disk = D_UP_TO_DATE;
4394 if (ns.conn == C_WF_REPORT_PARAMS)
4395 ns.conn = C_CONNECTED;
4397 if (peer_state.conn == C_AHEAD)
4401 * if (primary and diskless and peer uuid != effective uuid)
4402 * abort attach on peer;
4404 * If this node does not have good data, was already connected, but
4405 * the peer did a late attach only now, trying to "negotiate" with me,
4406 * AND I am currently Primary, possibly frozen, with some specific
4407 * "effective" uuid, this should never be reached, really, because
4408 * we first send the uuids, then the current state.
4410 * In this scenario, we already dropped the connection hard
4411 * when we received the unsuitable uuids (receive_uuids().
4413 * Should we want to change this, that is: not drop the connection in
4414 * receive_uuids() already, then we would need to add a branch here
4415 * that aborts the attach of "unsuitable uuids" on the peer in case
4416 * this node is currently Diskless Primary.
4419 if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4420 get_ldev_if_state(device, D_NEGOTIATING)) {
4421 int cr; /* consider resync */
4423 /* if we established a new connection */
4424 cr = (os.conn < C_CONNECTED);
4425 /* if we had an established connection
4426 * and one of the nodes newly attaches a disk */
4427 cr |= (os.conn == C_CONNECTED &&
4428 (peer_state.disk == D_NEGOTIATING ||
4429 os.disk == D_NEGOTIATING));
4430 /* if we have both been inconsistent, and the peer has been
4431 * forced to be UpToDate with --force */
4432 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4433 /* if we had been plain connected, and the admin requested to
4434 * start a sync by "invalidate" or "invalidate-remote" */
4435 cr |= (os.conn == C_CONNECTED &&
4436 (peer_state.conn >= C_STARTING_SYNC_S &&
4437 peer_state.conn <= C_WF_BITMAP_T));
4440 ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4443 if (ns.conn == C_MASK) {
4444 ns.conn = C_CONNECTED;
4445 if (device->state.disk == D_NEGOTIATING) {
4446 drbd_force_state(device, NS(disk, D_FAILED));
4447 } else if (peer_state.disk == D_NEGOTIATING) {
4448 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4449 peer_state.disk = D_DISKLESS;
4450 real_peer_disk = D_DISKLESS;
4452 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4454 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4455 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4461 spin_lock_irq(&device->resource->req_lock);
4462 if (os.i != drbd_read_state(device).i)
4464 clear_bit(CONSIDER_RESYNC, &device->flags);
4465 ns.peer = peer_state.role;
4466 ns.pdsk = real_peer_disk;
4467 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4468 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4469 ns.disk = device->new_state_tmp.disk;
4470 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4471 if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4472 test_bit(NEW_CUR_UUID, &device->flags)) {
4473 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4474 for temporal network outages! */
4475 spin_unlock_irq(&device->resource->req_lock);
4476 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4477 tl_clear(peer_device->connection);
4478 drbd_uuid_new_current(device);
4479 clear_bit(NEW_CUR_UUID, &device->flags);
4480 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4483 rv = _drbd_set_state(device, ns, cs_flags, NULL);
4484 ns = drbd_read_state(device);
4485 spin_unlock_irq(&device->resource->req_lock);
4487 if (rv < SS_SUCCESS) {
4488 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4492 if (os.conn > C_WF_REPORT_PARAMS) {
4493 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4494 peer_state.disk != D_NEGOTIATING ) {
4495 /* we want resync, peer has not yet decided to sync... */
4496 /* Nowadays only used when forcing a node into primary role and
4497 setting its disk to UpToDate with that */
4498 drbd_send_uuids(peer_device);
4499 drbd_send_current_state(peer_device);
4503 clear_bit(DISCARD_MY_DATA, &device->flags);
4505 drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4510 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4512 struct drbd_peer_device *peer_device;
4513 struct drbd_device *device;
4514 struct p_rs_uuid *p = pi->data;
4516 peer_device = conn_peer_device(connection, pi->vnr);
4519 device = peer_device->device;
4521 wait_event(device->misc_wait,
4522 device->state.conn == C_WF_SYNC_UUID ||
4523 device->state.conn == C_BEHIND ||
4524 device->state.conn < C_CONNECTED ||
4525 device->state.disk < D_NEGOTIATING);
4527 /* D_ASSERT(device, device->state.conn == C_WF_SYNC_UUID ); */
4529 /* Here the _drbd_uuid_ functions are right, current should
4530 _not_ be rotated into the history */
4531 if (get_ldev_if_state(device, D_NEGOTIATING)) {
4532 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4533 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4535 drbd_print_uuids(device, "updated sync uuid");
4536 drbd_start_resync(device, C_SYNC_TARGET);
4540 drbd_err(device, "Ignoring SyncUUID packet!\n");
4546 * receive_bitmap_plain
4548 * Return 0 when done, 1 when another iteration is needed, and a negative error
4549 * code upon failure.
4552 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4553 unsigned long *p, struct bm_xfer_ctx *c)
4555 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4556 drbd_header_size(peer_device->connection);
4557 unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4558 c->bm_words - c->word_offset);
4559 unsigned int want = num_words * sizeof(*p);
4563 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4568 err = drbd_recv_all(peer_device->connection, p, want);
4572 drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4574 c->word_offset += num_words;
4575 c->bit_offset = c->word_offset * BITS_PER_LONG;
4576 if (c->bit_offset > c->bm_bits)
4577 c->bit_offset = c->bm_bits;
4582 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4584 return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4587 static int dcbp_get_start(struct p_compressed_bm *p)
4589 return (p->encoding & 0x80) != 0;
4592 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4594 return (p->encoding >> 4) & 0x7;
4600 * Return 0 when done, 1 when another iteration is needed, and a negative error
4601 * code upon failure.
4604 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4605 struct p_compressed_bm *p,
4606 struct bm_xfer_ctx *c,
4609 struct bitstream bs;
4613 unsigned long s = c->bit_offset;
4615 int toggle = dcbp_get_start(p);
4619 bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4621 bits = bitstream_get_bits(&bs, &look_ahead, 64);
4625 for (have = bits; have > 0; s += rl, toggle = !toggle) {
4626 bits = vli_decode_bits(&rl, look_ahead);
4632 if (e >= c->bm_bits) {
4633 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4636 _drbd_bm_set_bits(peer_device->device, s, e);
4640 drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4641 have, bits, look_ahead,
4642 (unsigned int)(bs.cur.b - p->code),
4643 (unsigned int)bs.buf_len);
4646 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4647 if (likely(bits < 64))
4648 look_ahead >>= bits;
4653 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4656 look_ahead |= tmp << have;
4661 bm_xfer_ctx_bit_to_word_offset(c);
4663 return (s != c->bm_bits);
4669 * Return 0 when done, 1 when another iteration is needed, and a negative error
4670 * code upon failure.
4673 decode_bitmap_c(struct drbd_peer_device *peer_device,
4674 struct p_compressed_bm *p,
4675 struct bm_xfer_ctx *c,
4678 if (dcbp_get_code(p) == RLE_VLI_Bits)
4679 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4681 /* other variants had been implemented for evaluation,
4682 * but have been dropped as this one turned out to be "best"
4683 * during all our tests. */
4685 drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4686 conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4690 void INFO_bm_xfer_stats(struct drbd_device *device,
4691 const char *direction, struct bm_xfer_ctx *c)
4693 /* what would it take to transfer it "plaintext" */
4694 unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4695 unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4696 unsigned int plain =
4697 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4698 c->bm_words * sizeof(unsigned long);
4699 unsigned int total = c->bytes[0] + c->bytes[1];
4702 /* total can not be zero. but just in case: */
4706 /* don't report if not compressed */
4710 /* total < plain. check for overflow, still */
4711 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4712 : (1000 * total / plain);
4718 drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4719 "total %u; compression: %u.%u%%\n",
4721 c->bytes[1], c->packets[1],
4722 c->bytes[0], c->packets[0],
4723 total, r/10, r % 10);
4726 /* Since we are processing the bitfield from lower addresses to higher,
4727 it does not matter if the process it in 32 bit chunks or 64 bit
4728 chunks as long as it is little endian. (Understand it as byte stream,
4729 beginning with the lowest byte...) If we would use big endian
4730 we would need to process it from the highest address to the lowest,
4731 in order to be agnostic to the 32 vs 64 bits issue.
4733 returns 0 on failure, 1 if we successfully received it. */
4734 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4736 struct drbd_peer_device *peer_device;
4737 struct drbd_device *device;
4738 struct bm_xfer_ctx c;
4741 peer_device = conn_peer_device(connection, pi->vnr);
4744 device = peer_device->device;
4746 drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4747 /* you are supposed to send additional out-of-sync information
4748 * if you actually set bits during this phase */
4750 c = (struct bm_xfer_ctx) {
4751 .bm_bits = drbd_bm_bits(device),
4752 .bm_words = drbd_bm_words(device),
4756 if (pi->cmd == P_BITMAP)
4757 err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4758 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4759 /* MAYBE: sanity check that we speak proto >= 90,
4760 * and the feature is enabled! */
4761 struct p_compressed_bm *p = pi->data;
4763 if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4764 drbd_err(device, "ReportCBitmap packet too large\n");
4768 if (pi->size <= sizeof(*p)) {
4769 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4773 err = drbd_recv_all(peer_device->connection, p, pi->size);
4776 err = decode_bitmap_c(peer_device, p, &c, pi->size);
4778 drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4783 c.packets[pi->cmd == P_BITMAP]++;
4784 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4791 err = drbd_recv_header(peer_device->connection, pi);
4796 INFO_bm_xfer_stats(device, "receive", &c);
4798 if (device->state.conn == C_WF_BITMAP_T) {
4799 enum drbd_state_rv rv;
4801 err = drbd_send_bitmap(device);
4804 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4805 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4806 D_ASSERT(device, rv == SS_SUCCESS);
4807 } else if (device->state.conn != C_WF_BITMAP_S) {
4808 /* admin may have requested C_DISCONNECTING,
4809 * other threads may have noticed network errors */
4810 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4811 drbd_conn_str(device->state.conn));
4816 drbd_bm_unlock(device);
4817 if (!err && device->state.conn == C_WF_BITMAP_S)
4818 drbd_start_resync(device, C_SYNC_SOURCE);
4822 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4824 drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4827 return ignore_remaining_packet(connection, pi);
4830 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4832 /* Make sure we've acked all the TCP data associated
4833 * with the data requests being unplugged */
4834 drbd_tcp_quickack(connection->data.socket);
4839 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4841 struct drbd_peer_device *peer_device;
4842 struct drbd_device *device;
4843 struct p_block_desc *p = pi->data;
4845 peer_device = conn_peer_device(connection, pi->vnr);
4848 device = peer_device->device;
4850 switch (device->state.conn) {
4851 case C_WF_SYNC_UUID:
4856 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4857 drbd_conn_str(device->state.conn));
4860 drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4865 static int receive_rs_deallocated(struct drbd_connection *connection, struct packet_info *pi)
4867 struct drbd_peer_device *peer_device;
4868 struct p_block_desc *p = pi->data;
4869 struct drbd_device *device;
4873 peer_device = conn_peer_device(connection, pi->vnr);
4876 device = peer_device->device;
4878 sector = be64_to_cpu(p->sector);
4879 size = be32_to_cpu(p->blksize);
4881 dec_rs_pending(device);
4883 if (get_ldev(device)) {
4884 struct drbd_peer_request *peer_req;
4885 const int op = REQ_OP_WRITE_ZEROES;
4887 peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER, sector,
4894 peer_req->w.cb = e_end_resync_block;
4895 peer_req->submit_jif = jiffies;
4896 peer_req->flags |= EE_IS_TRIM;
4898 spin_lock_irq(&device->resource->req_lock);
4899 list_add_tail(&peer_req->w.list, &device->sync_ee);
4900 spin_unlock_irq(&device->resource->req_lock);
4902 atomic_add(pi->size >> 9, &device->rs_sect_ev);
4903 err = drbd_submit_peer_request(device, peer_req, op, 0, DRBD_FAULT_RS_WR);
4906 spin_lock_irq(&device->resource->req_lock);
4907 list_del(&peer_req->w.list);
4908 spin_unlock_irq(&device->resource->req_lock);
4910 drbd_free_peer_req(device, peer_req);
4916 inc_unacked(device);
4918 /* No put_ldev() here. Gets called in drbd_endio_write_sec_final(),
4919 as well as drbd_rs_complete_io() */
4922 drbd_rs_complete_io(device, sector);
4923 drbd_send_ack_ex(peer_device, P_NEG_ACK, sector, size, ID_SYNCER);
4926 atomic_add(size >> 9, &device->rs_sect_in);
4933 unsigned int pkt_size;
4934 int (*fn)(struct drbd_connection *, struct packet_info *);
4937 static struct data_cmd drbd_cmd_handler[] = {
4938 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
4939 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
4940 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
4941 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
4942 [P_BITMAP] = { 1, 0, receive_bitmap } ,
4943 [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
4944 [P_UNPLUG_REMOTE] = { 0, 0, receive_UnplugRemote },
4945 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4946 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4947 [P_SYNC_PARAM] = { 1, 0, receive_SyncParam },
4948 [P_SYNC_PARAM89] = { 1, 0, receive_SyncParam },
4949 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
4950 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
4951 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
4952 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
4953 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
4954 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
4955 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4956 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4957 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4958 [P_RS_THIN_REQ] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4959 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
4960 [P_OUT_OF_SYNC] = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
4961 [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
4962 [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
4963 [P_TRIM] = { 0, sizeof(struct p_trim), receive_Data },
4964 [P_RS_DEALLOCATED] = { 0, sizeof(struct p_block_desc), receive_rs_deallocated },
4965 [P_WSAME] = { 1, sizeof(struct p_wsame), receive_Data },
4968 static void drbdd(struct drbd_connection *connection)
4970 struct packet_info pi;
4971 size_t shs; /* sub header size */
4974 while (get_t_state(&connection->receiver) == RUNNING) {
4975 struct data_cmd const *cmd;
4977 drbd_thread_current_set_cpu(&connection->receiver);
4978 update_receiver_timing_details(connection, drbd_recv_header_maybe_unplug);
4979 if (drbd_recv_header_maybe_unplug(connection, &pi))
4982 cmd = &drbd_cmd_handler[pi.cmd];
4983 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
4984 drbd_err(connection, "Unexpected data packet %s (0x%04x)",
4985 cmdname(pi.cmd), pi.cmd);
4989 shs = cmd->pkt_size;
4990 if (pi.cmd == P_SIZES && connection->agreed_features & DRBD_FF_WSAME)
4991 shs += sizeof(struct o_qlim);
4992 if (pi.size > shs && !cmd->expect_payload) {
4993 drbd_err(connection, "No payload expected %s l:%d\n",
4994 cmdname(pi.cmd), pi.size);
4997 if (pi.size < shs) {
4998 drbd_err(connection, "%s: unexpected packet size, expected:%d received:%d\n",
4999 cmdname(pi.cmd), (int)shs, pi.size);
5004 update_receiver_timing_details(connection, drbd_recv_all_warn);
5005 err = drbd_recv_all_warn(connection, pi.data, shs);
5011 update_receiver_timing_details(connection, cmd->fn);
5012 err = cmd->fn(connection, &pi);
5014 drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
5015 cmdname(pi.cmd), err, pi.size);
5022 conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
5025 static void conn_disconnect(struct drbd_connection *connection)
5027 struct drbd_peer_device *peer_device;
5031 if (connection->cstate == C_STANDALONE)
5034 /* We are about to start the cleanup after connection loss.
5035 * Make sure drbd_make_request knows about that.
5036 * Usually we should be in some network failure state already,
5037 * but just in case we are not, we fix it up here.
5039 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5041 /* ack_receiver does not clean up anything. it must not interfere, either */
5042 drbd_thread_stop(&connection->ack_receiver);
5043 if (connection->ack_sender) {
5044 destroy_workqueue(connection->ack_sender);
5045 connection->ack_sender = NULL;
5047 drbd_free_sock(connection);
5050 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5051 struct drbd_device *device = peer_device->device;
5052 kref_get(&device->kref);
5054 drbd_disconnected(peer_device);
5055 kref_put(&device->kref, drbd_destroy_device);
5060 if (!list_empty(&connection->current_epoch->list))
5061 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
5062 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
5063 atomic_set(&connection->current_epoch->epoch_size, 0);
5064 connection->send.seen_any_write_yet = false;
5066 drbd_info(connection, "Connection closed\n");
5068 if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
5069 conn_try_outdate_peer_async(connection);
5071 spin_lock_irq(&connection->resource->req_lock);
5072 oc = connection->cstate;
5073 if (oc >= C_UNCONNECTED)
5074 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
5076 spin_unlock_irq(&connection->resource->req_lock);
5078 if (oc == C_DISCONNECTING)
5079 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
5082 static int drbd_disconnected(struct drbd_peer_device *peer_device)
5084 struct drbd_device *device = peer_device->device;
5087 /* wait for current activity to cease. */
5088 spin_lock_irq(&device->resource->req_lock);
5089 _drbd_wait_ee_list_empty(device, &device->active_ee);
5090 _drbd_wait_ee_list_empty(device, &device->sync_ee);
5091 _drbd_wait_ee_list_empty(device, &device->read_ee);
5092 spin_unlock_irq(&device->resource->req_lock);
5094 /* We do not have data structures that would allow us to
5095 * get the rs_pending_cnt down to 0 again.
5096 * * On C_SYNC_TARGET we do not have any data structures describing
5097 * the pending RSDataRequest's we have sent.
5098 * * On C_SYNC_SOURCE there is no data structure that tracks
5099 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
5100 * And no, it is not the sum of the reference counts in the
5101 * resync_LRU. The resync_LRU tracks the whole operation including
5102 * the disk-IO, while the rs_pending_cnt only tracks the blocks
5104 drbd_rs_cancel_all(device);
5105 device->rs_total = 0;
5106 device->rs_failed = 0;
5107 atomic_set(&device->rs_pending_cnt, 0);
5108 wake_up(&device->misc_wait);
5110 del_timer_sync(&device->resync_timer);
5111 resync_timer_fn(&device->resync_timer);
5113 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
5114 * w_make_resync_request etc. which may still be on the worker queue
5115 * to be "canceled" */
5116 drbd_flush_workqueue(&peer_device->connection->sender_work);
5118 drbd_finish_peer_reqs(device);
5120 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
5121 might have issued a work again. The one before drbd_finish_peer_reqs() is
5122 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
5123 drbd_flush_workqueue(&peer_device->connection->sender_work);
5125 /* need to do it again, drbd_finish_peer_reqs() may have populated it
5126 * again via drbd_try_clear_on_disk_bm(). */
5127 drbd_rs_cancel_all(device);
5129 kfree(device->p_uuid);
5130 device->p_uuid = NULL;
5132 if (!drbd_suspended(device))
5133 tl_clear(peer_device->connection);
5135 drbd_md_sync(device);
5137 if (get_ldev(device)) {
5138 drbd_bitmap_io(device, &drbd_bm_write_copy_pages,
5139 "write from disconnected", BM_LOCKED_CHANGE_ALLOWED);
5143 /* tcp_close and release of sendpage pages can be deferred. I don't
5144 * want to use SO_LINGER, because apparently it can be deferred for
5145 * more than 20 seconds (longest time I checked).
5147 * Actually we don't care for exactly when the network stack does its
5148 * put_page(), but release our reference on these pages right here.
5150 i = drbd_free_peer_reqs(device, &device->net_ee);
5152 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
5153 i = atomic_read(&device->pp_in_use_by_net);
5155 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
5156 i = atomic_read(&device->pp_in_use);
5158 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
5160 D_ASSERT(device, list_empty(&device->read_ee));
5161 D_ASSERT(device, list_empty(&device->active_ee));
5162 D_ASSERT(device, list_empty(&device->sync_ee));
5163 D_ASSERT(device, list_empty(&device->done_ee));
5169 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
5170 * we can agree on is stored in agreed_pro_version.
5172 * feature flags and the reserved array should be enough room for future
5173 * enhancements of the handshake protocol, and possible plugins...
5175 * for now, they are expected to be zero, but ignored.
5177 static int drbd_send_features(struct drbd_connection *connection)
5179 struct drbd_socket *sock;
5180 struct p_connection_features *p;
5182 sock = &connection->data;
5183 p = conn_prepare_command(connection, sock);
5186 memset(p, 0, sizeof(*p));
5187 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
5188 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
5189 p->feature_flags = cpu_to_be32(PRO_FEATURES);
5190 return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
5195 * 1 yes, we have a valid connection
5196 * 0 oops, did not work out, please try again
5197 * -1 peer talks different language,
5198 * no point in trying again, please go standalone.
5200 static int drbd_do_features(struct drbd_connection *connection)
5202 /* ASSERT current == connection->receiver ... */
5203 struct p_connection_features *p;
5204 const int expect = sizeof(struct p_connection_features);
5205 struct packet_info pi;
5208 err = drbd_send_features(connection);
5212 err = drbd_recv_header(connection, &pi);
5216 if (pi.cmd != P_CONNECTION_FEATURES) {
5217 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
5218 cmdname(pi.cmd), pi.cmd);
5222 if (pi.size != expect) {
5223 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
5229 err = drbd_recv_all_warn(connection, p, expect);
5233 p->protocol_min = be32_to_cpu(p->protocol_min);
5234 p->protocol_max = be32_to_cpu(p->protocol_max);
5235 if (p->protocol_max == 0)
5236 p->protocol_max = p->protocol_min;
5238 if (PRO_VERSION_MAX < p->protocol_min ||
5239 PRO_VERSION_MIN > p->protocol_max)
5242 connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
5243 connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
5245 drbd_info(connection, "Handshake successful: "
5246 "Agreed network protocol version %d\n", connection->agreed_pro_version);
5248 drbd_info(connection, "Feature flags enabled on protocol level: 0x%x%s%s%s.\n",
5249 connection->agreed_features,
5250 connection->agreed_features & DRBD_FF_TRIM ? " TRIM" : "",
5251 connection->agreed_features & DRBD_FF_THIN_RESYNC ? " THIN_RESYNC" : "",
5252 connection->agreed_features & DRBD_FF_WSAME ? " WRITE_SAME" :
5253 connection->agreed_features ? "" : " none");
5258 drbd_err(connection, "incompatible DRBD dialects: "
5259 "I support %d-%d, peer supports %d-%d\n",
5260 PRO_VERSION_MIN, PRO_VERSION_MAX,
5261 p->protocol_min, p->protocol_max);
5265 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
5266 static int drbd_do_auth(struct drbd_connection *connection)
5268 drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
5269 drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
5273 #define CHALLENGE_LEN 64
5277 0 - failed, try again (network error),
5278 -1 - auth failed, don't try again.
5281 static int drbd_do_auth(struct drbd_connection *connection)
5283 struct drbd_socket *sock;
5284 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
5285 char *response = NULL;
5286 char *right_response = NULL;
5287 char *peers_ch = NULL;
5288 unsigned int key_len;
5289 char secret[SHARED_SECRET_MAX]; /* 64 byte */
5290 unsigned int resp_size;
5291 SHASH_DESC_ON_STACK(desc, connection->cram_hmac_tfm);
5292 struct packet_info pi;
5293 struct net_conf *nc;
5296 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
5299 nc = rcu_dereference(connection->net_conf);
5300 key_len = strlen(nc->shared_secret);
5301 memcpy(secret, nc->shared_secret, key_len);
5304 desc->tfm = connection->cram_hmac_tfm;
5307 rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
5309 drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
5314 get_random_bytes(my_challenge, CHALLENGE_LEN);
5316 sock = &connection->data;
5317 if (!conn_prepare_command(connection, sock)) {
5321 rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
5322 my_challenge, CHALLENGE_LEN);
5326 err = drbd_recv_header(connection, &pi);
5332 if (pi.cmd != P_AUTH_CHALLENGE) {
5333 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
5334 cmdname(pi.cmd), pi.cmd);
5339 if (pi.size > CHALLENGE_LEN * 2) {
5340 drbd_err(connection, "expected AuthChallenge payload too big.\n");
5345 if (pi.size < CHALLENGE_LEN) {
5346 drbd_err(connection, "AuthChallenge payload too small.\n");
5351 peers_ch = kmalloc(pi.size, GFP_NOIO);
5352 if (peers_ch == NULL) {
5353 drbd_err(connection, "kmalloc of peers_ch failed\n");
5358 err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5364 if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5365 drbd_err(connection, "Peer presented the same challenge!\n");
5370 resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5371 response = kmalloc(resp_size, GFP_NOIO);
5372 if (response == NULL) {
5373 drbd_err(connection, "kmalloc of response failed\n");
5378 rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5380 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5385 if (!conn_prepare_command(connection, sock)) {
5389 rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5390 response, resp_size);
5394 err = drbd_recv_header(connection, &pi);
5400 if (pi.cmd != P_AUTH_RESPONSE) {
5401 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5402 cmdname(pi.cmd), pi.cmd);
5407 if (pi.size != resp_size) {
5408 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5413 err = drbd_recv_all_warn(connection, response , resp_size);
5419 right_response = kmalloc(resp_size, GFP_NOIO);
5420 if (right_response == NULL) {
5421 drbd_err(connection, "kmalloc of right_response failed\n");
5426 rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5429 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5434 rv = !memcmp(response, right_response, resp_size);
5437 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5445 kfree(right_response);
5446 shash_desc_zero(desc);
5452 int drbd_receiver(struct drbd_thread *thi)
5454 struct drbd_connection *connection = thi->connection;
5457 drbd_info(connection, "receiver (re)started\n");
5460 h = conn_connect(connection);
5462 conn_disconnect(connection);
5463 schedule_timeout_interruptible(HZ);
5466 drbd_warn(connection, "Discarding network configuration.\n");
5467 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5472 blk_start_plug(&connection->receiver_plug);
5474 blk_finish_plug(&connection->receiver_plug);
5477 conn_disconnect(connection);
5479 drbd_info(connection, "receiver terminated\n");
5483 /* ********* acknowledge sender ******** */
5485 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5487 struct p_req_state_reply *p = pi->data;
5488 int retcode = be32_to_cpu(p->retcode);
5490 if (retcode >= SS_SUCCESS) {
5491 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5493 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5494 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5495 drbd_set_st_err_str(retcode), retcode);
5497 wake_up(&connection->ping_wait);
5502 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5504 struct drbd_peer_device *peer_device;
5505 struct drbd_device *device;
5506 struct p_req_state_reply *p = pi->data;
5507 int retcode = be32_to_cpu(p->retcode);
5509 peer_device = conn_peer_device(connection, pi->vnr);
5512 device = peer_device->device;
5514 if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5515 D_ASSERT(device, connection->agreed_pro_version < 100);
5516 return got_conn_RqSReply(connection, pi);
5519 if (retcode >= SS_SUCCESS) {
5520 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5522 set_bit(CL_ST_CHG_FAIL, &device->flags);
5523 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5524 drbd_set_st_err_str(retcode), retcode);
5526 wake_up(&device->state_wait);
5531 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5533 return drbd_send_ping_ack(connection);
5537 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5539 /* restore idle timeout */
5540 connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5541 if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5542 wake_up(&connection->ping_wait);
5547 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5549 struct drbd_peer_device *peer_device;
5550 struct drbd_device *device;
5551 struct p_block_ack *p = pi->data;
5552 sector_t sector = be64_to_cpu(p->sector);
5553 int blksize = be32_to_cpu(p->blksize);
5555 peer_device = conn_peer_device(connection, pi->vnr);
5558 device = peer_device->device;
5560 D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5562 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5564 if (get_ldev(device)) {
5565 drbd_rs_complete_io(device, sector);
5566 drbd_set_in_sync(device, sector, blksize);
5567 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5568 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5571 dec_rs_pending(device);
5572 atomic_add(blksize >> 9, &device->rs_sect_in);
5578 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5579 struct rb_root *root, const char *func,
5580 enum drbd_req_event what, bool missing_ok)
5582 struct drbd_request *req;
5583 struct bio_and_error m;
5585 spin_lock_irq(&device->resource->req_lock);
5586 req = find_request(device, root, id, sector, missing_ok, func);
5587 if (unlikely(!req)) {
5588 spin_unlock_irq(&device->resource->req_lock);
5591 __req_mod(req, what, &m);
5592 spin_unlock_irq(&device->resource->req_lock);
5595 complete_master_bio(device, &m);
5599 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5601 struct drbd_peer_device *peer_device;
5602 struct drbd_device *device;
5603 struct p_block_ack *p = pi->data;
5604 sector_t sector = be64_to_cpu(p->sector);
5605 int blksize = be32_to_cpu(p->blksize);
5606 enum drbd_req_event what;
5608 peer_device = conn_peer_device(connection, pi->vnr);
5611 device = peer_device->device;
5613 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5615 if (p->block_id == ID_SYNCER) {
5616 drbd_set_in_sync(device, sector, blksize);
5617 dec_rs_pending(device);
5621 case P_RS_WRITE_ACK:
5622 what = WRITE_ACKED_BY_PEER_AND_SIS;
5625 what = WRITE_ACKED_BY_PEER;
5628 what = RECV_ACKED_BY_PEER;
5631 what = CONFLICT_RESOLVED;
5634 what = POSTPONE_WRITE;
5640 return validate_req_change_req_state(device, p->block_id, sector,
5641 &device->write_requests, __func__,
5645 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5647 struct drbd_peer_device *peer_device;
5648 struct drbd_device *device;
5649 struct p_block_ack *p = pi->data;
5650 sector_t sector = be64_to_cpu(p->sector);
5651 int size = be32_to_cpu(p->blksize);
5654 peer_device = conn_peer_device(connection, pi->vnr);
5657 device = peer_device->device;
5659 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5661 if (p->block_id == ID_SYNCER) {
5662 dec_rs_pending(device);
5663 drbd_rs_failed_io(device, sector, size);
5667 err = validate_req_change_req_state(device, p->block_id, sector,
5668 &device->write_requests, __func__,
5671 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5672 The master bio might already be completed, therefore the
5673 request is no longer in the collision hash. */
5674 /* In Protocol B we might already have got a P_RECV_ACK
5675 but then get a P_NEG_ACK afterwards. */
5676 drbd_set_out_of_sync(device, sector, size);
5681 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5683 struct drbd_peer_device *peer_device;
5684 struct drbd_device *device;
5685 struct p_block_ack *p = pi->data;
5686 sector_t sector = be64_to_cpu(p->sector);
5688 peer_device = conn_peer_device(connection, pi->vnr);
5691 device = peer_device->device;
5693 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5695 drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5696 (unsigned long long)sector, be32_to_cpu(p->blksize));
5698 return validate_req_change_req_state(device, p->block_id, sector,
5699 &device->read_requests, __func__,
5703 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5705 struct drbd_peer_device *peer_device;
5706 struct drbd_device *device;
5709 struct p_block_ack *p = pi->data;
5711 peer_device = conn_peer_device(connection, pi->vnr);
5714 device = peer_device->device;
5716 sector = be64_to_cpu(p->sector);
5717 size = be32_to_cpu(p->blksize);
5719 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5721 dec_rs_pending(device);
5723 if (get_ldev_if_state(device, D_FAILED)) {
5724 drbd_rs_complete_io(device, sector);
5726 case P_NEG_RS_DREPLY:
5727 drbd_rs_failed_io(device, sector, size);
5739 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5741 struct p_barrier_ack *p = pi->data;
5742 struct drbd_peer_device *peer_device;
5745 tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5748 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5749 struct drbd_device *device = peer_device->device;
5751 if (device->state.conn == C_AHEAD &&
5752 atomic_read(&device->ap_in_flight) == 0 &&
5753 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5754 device->start_resync_timer.expires = jiffies + HZ;
5755 add_timer(&device->start_resync_timer);
5763 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5765 struct drbd_peer_device *peer_device;
5766 struct drbd_device *device;
5767 struct p_block_ack *p = pi->data;
5768 struct drbd_device_work *dw;
5772 peer_device = conn_peer_device(connection, pi->vnr);
5775 device = peer_device->device;
5777 sector = be64_to_cpu(p->sector);
5778 size = be32_to_cpu(p->blksize);
5780 update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5782 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5783 drbd_ov_out_of_sync_found(device, sector, size);
5785 ov_out_of_sync_print(device);
5787 if (!get_ldev(device))
5790 drbd_rs_complete_io(device, sector);
5791 dec_rs_pending(device);
5795 /* let's advance progress step marks only for every other megabyte */
5796 if ((device->ov_left & 0x200) == 0x200)
5797 drbd_advance_rs_marks(device, device->ov_left);
5799 if (device->ov_left == 0) {
5800 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5802 dw->w.cb = w_ov_finished;
5803 dw->device = device;
5804 drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5806 drbd_err(device, "kmalloc(dw) failed.");
5807 ov_out_of_sync_print(device);
5808 drbd_resync_finished(device);
5815 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5820 struct meta_sock_cmd {
5822 int (*fn)(struct drbd_connection *connection, struct packet_info *);
5825 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5828 struct net_conf *nc;
5831 nc = rcu_dereference(connection->net_conf);
5832 t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5839 connection->meta.socket->sk->sk_rcvtimeo = t;
5842 static void set_ping_timeout(struct drbd_connection *connection)
5844 set_rcvtimeo(connection, 1);
5847 static void set_idle_timeout(struct drbd_connection *connection)
5849 set_rcvtimeo(connection, 0);
5852 static struct meta_sock_cmd ack_receiver_tbl[] = {
5853 [P_PING] = { 0, got_Ping },
5854 [P_PING_ACK] = { 0, got_PingAck },
5855 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5856 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5857 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
5858 [P_SUPERSEDED] = { sizeof(struct p_block_ack), got_BlockAck },
5859 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
5860 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
5861 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply },
5862 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
5863 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
5864 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5865 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
5866 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
5867 [P_RS_CANCEL] = { sizeof(struct p_block_ack), got_NegRSDReply },
5868 [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5869 [P_RETRY_WRITE] = { sizeof(struct p_block_ack), got_BlockAck },
5872 int drbd_ack_receiver(struct drbd_thread *thi)
5874 struct drbd_connection *connection = thi->connection;
5875 struct meta_sock_cmd *cmd = NULL;
5876 struct packet_info pi;
5877 unsigned long pre_recv_jif;
5879 void *buf = connection->meta.rbuf;
5881 unsigned int header_size = drbd_header_size(connection);
5882 int expect = header_size;
5883 bool ping_timeout_active = false;
5884 struct sched_param param = { .sched_priority = 2 };
5886 rv = sched_setscheduler(current, SCHED_RR, ¶m);
5888 drbd_err(connection, "drbd_ack_receiver: ERROR set priority, ret=%d\n", rv);
5890 while (get_t_state(thi) == RUNNING) {
5891 drbd_thread_current_set_cpu(thi);
5893 conn_reclaim_net_peer_reqs(connection);
5895 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5896 if (drbd_send_ping(connection)) {
5897 drbd_err(connection, "drbd_send_ping has failed\n");
5900 set_ping_timeout(connection);
5901 ping_timeout_active = true;
5904 pre_recv_jif = jiffies;
5905 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5908 * -EINTR (on meta) we got a signal
5909 * -EAGAIN (on meta) rcvtimeo expired
5910 * -ECONNRESET other side closed the connection
5911 * -ERESTARTSYS (on data) we got a signal
5912 * rv < 0 other than above: unexpected error!
5913 * rv == expected: full header or command
5914 * rv < expected: "woken" by signal during receive
5915 * rv == 0 : "connection shut down by peer"
5917 if (likely(rv > 0)) {
5920 } else if (rv == 0) {
5921 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
5924 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
5927 t = wait_event_timeout(connection->ping_wait,
5928 connection->cstate < C_WF_REPORT_PARAMS,
5933 drbd_err(connection, "meta connection shut down by peer.\n");
5935 } else if (rv == -EAGAIN) {
5936 /* If the data socket received something meanwhile,
5937 * that is good enough: peer is still alive. */
5938 if (time_after(connection->last_received, pre_recv_jif))
5940 if (ping_timeout_active) {
5941 drbd_err(connection, "PingAck did not arrive in time.\n");
5944 set_bit(SEND_PING, &connection->flags);
5946 } else if (rv == -EINTR) {
5947 /* maybe drbd_thread_stop(): the while condition will notice.
5948 * maybe woken for send_ping: we'll send a ping above,
5949 * and change the rcvtimeo */
5950 flush_signals(current);
5953 drbd_err(connection, "sock_recvmsg returned %d\n", rv);
5957 if (received == expect && cmd == NULL) {
5958 if (decode_header(connection, connection->meta.rbuf, &pi))
5960 cmd = &ack_receiver_tbl[pi.cmd];
5961 if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
5962 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
5963 cmdname(pi.cmd), pi.cmd);
5966 expect = header_size + cmd->pkt_size;
5967 if (pi.size != expect - header_size) {
5968 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
5973 if (received == expect) {
5976 err = cmd->fn(connection, &pi);
5978 drbd_err(connection, "%pf failed\n", cmd->fn);
5982 connection->last_received = jiffies;
5984 if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
5985 set_idle_timeout(connection);
5986 ping_timeout_active = false;
5989 buf = connection->meta.rbuf;
5991 expect = header_size;
5998 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5999 conn_md_sync(connection);
6003 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
6006 drbd_info(connection, "ack_receiver terminated\n");
6011 void drbd_send_acks_wf(struct work_struct *ws)
6013 struct drbd_peer_device *peer_device =
6014 container_of(ws, struct drbd_peer_device, send_acks_work);
6015 struct drbd_connection *connection = peer_device->connection;
6016 struct drbd_device *device = peer_device->device;
6017 struct net_conf *nc;
6021 nc = rcu_dereference(connection->net_conf);
6022 tcp_cork = nc->tcp_cork;
6026 drbd_tcp_cork(connection->meta.socket);
6028 err = drbd_finish_peer_reqs(device);
6029 kref_put(&device->kref, drbd_destroy_device);
6030 /* get is in drbd_endio_write_sec_final(). That is necessary to keep the
6031 struct work_struct send_acks_work alive, which is in the peer_device object */
6034 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
6039 drbd_tcp_uncork(connection->meta.socket);