4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
28 #include <asm/uaccess.h>
31 #include <linux/drbd.h>
33 #include <linux/file.h>
36 #include <linux/memcontrol.h>
37 #include <linux/mm_inline.h>
38 #include <linux/slab.h>
39 #include <linux/smp_lock.h>
40 #include <linux/pkt_sched.h>
41 #define __KERNEL_SYSCALLS__
42 #include <linux/unistd.h>
43 #include <linux/vmalloc.h>
44 #include <linux/random.h>
45 #include <linux/string.h>
46 #include <linux/scatterlist.h>
58 static int drbd_do_handshake(struct drbd_conf *mdev);
59 static int drbd_do_auth(struct drbd_conf *mdev);
61 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *, struct drbd_epoch *, enum epoch_event);
62 static int e_end_block(struct drbd_conf *, struct drbd_work *, int);
65 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
68 * some helper functions to deal with single linked page lists,
69 * page->private being our "next" pointer.
72 /* If at least n pages are linked at head, get n pages off.
73 * Otherwise, don't modify head, and return NULL.
74 * Locking is the responsibility of the caller.
76 static struct page *page_chain_del(struct page **head, int n)
90 tmp = page_chain_next(page);
92 break; /* found sufficient pages */
94 /* insufficient pages, don't use any of them. */
99 /* add end of list marker for the returned list */
100 set_page_private(page, 0);
101 /* actual return value, and adjustment of head */
107 /* may be used outside of locks to find the tail of a (usually short)
108 * "private" page chain, before adding it back to a global chain head
109 * with page_chain_add() under a spinlock. */
110 static struct page *page_chain_tail(struct page *page, int *len)
114 while ((tmp = page_chain_next(page)))
121 static int page_chain_free(struct page *page)
125 page_chain_for_each_safe(page, tmp) {
132 static void page_chain_add(struct page **head,
133 struct page *chain_first, struct page *chain_last)
137 tmp = page_chain_tail(chain_first, NULL);
138 BUG_ON(tmp != chain_last);
141 /* add chain to head */
142 set_page_private(chain_last, (unsigned long)*head);
146 static struct page *drbd_pp_first_pages_or_try_alloc(struct drbd_conf *mdev, int number)
148 struct page *page = NULL;
149 struct page *tmp = NULL;
152 /* Yes, testing drbd_pp_vacant outside the lock is racy.
153 * So what. It saves a spin_lock. */
154 if (drbd_pp_vacant >= number) {
155 spin_lock(&drbd_pp_lock);
156 page = page_chain_del(&drbd_pp_pool, number);
158 drbd_pp_vacant -= number;
159 spin_unlock(&drbd_pp_lock);
164 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
165 * "criss-cross" setup, that might cause write-out on some other DRBD,
166 * which in turn might block on the other node at this very place. */
167 for (i = 0; i < number; i++) {
168 tmp = alloc_page(GFP_TRY);
171 set_page_private(tmp, (unsigned long)page);
178 /* Not enough pages immediately available this time.
179 * No need to jump around here, drbd_pp_alloc will retry this
180 * function "soon". */
182 tmp = page_chain_tail(page, NULL);
183 spin_lock(&drbd_pp_lock);
184 page_chain_add(&drbd_pp_pool, page, tmp);
186 spin_unlock(&drbd_pp_lock);
191 /* kick lower level device, if we have more than (arbitrary number)
192 * reference counts on it, which typically are locally submitted io
193 * requests. don't use unacked_cnt, so we speed up proto A and B, too. */
194 static void maybe_kick_lo(struct drbd_conf *mdev)
196 if (atomic_read(&mdev->local_cnt) >= mdev->net_conf->unplug_watermark)
200 static void reclaim_net_ee(struct drbd_conf *mdev, struct list_head *to_be_freed)
202 struct drbd_epoch_entry *e;
203 struct list_head *le, *tle;
205 /* The EEs are always appended to the end of the list. Since
206 they are sent in order over the wire, they have to finish
207 in order. As soon as we see the first not finished we can
208 stop to examine the list... */
210 list_for_each_safe(le, tle, &mdev->net_ee) {
211 e = list_entry(le, struct drbd_epoch_entry, w.list);
212 if (drbd_ee_has_active_page(e))
214 list_move(le, to_be_freed);
218 static void drbd_kick_lo_and_reclaim_net(struct drbd_conf *mdev)
220 LIST_HEAD(reclaimed);
221 struct drbd_epoch_entry *e, *t;
224 spin_lock_irq(&mdev->req_lock);
225 reclaim_net_ee(mdev, &reclaimed);
226 spin_unlock_irq(&mdev->req_lock);
228 list_for_each_entry_safe(e, t, &reclaimed, w.list)
229 drbd_free_net_ee(mdev, e);
233 * drbd_pp_alloc() - Returns @number pages, retries forever (or until signalled)
234 * @mdev: DRBD device.
235 * @number: number of pages requested
236 * @retry: whether to retry, if not enough pages are available right now
238 * Tries to allocate number pages, first from our own page pool, then from
239 * the kernel, unless this allocation would exceed the max_buffers setting.
240 * Possibly retry until DRBD frees sufficient pages somewhere else.
242 * Returns a page chain linked via page->private.
244 static struct page *drbd_pp_alloc(struct drbd_conf *mdev, unsigned number, bool retry)
246 struct page *page = NULL;
249 /* Yes, we may run up to @number over max_buffers. If we
250 * follow it strictly, the admin will get it wrong anyways. */
251 if (atomic_read(&mdev->pp_in_use) < mdev->net_conf->max_buffers)
252 page = drbd_pp_first_pages_or_try_alloc(mdev, number);
254 while (page == NULL) {
255 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
257 drbd_kick_lo_and_reclaim_net(mdev);
259 if (atomic_read(&mdev->pp_in_use) < mdev->net_conf->max_buffers) {
260 page = drbd_pp_first_pages_or_try_alloc(mdev, number);
268 if (signal_pending(current)) {
269 dev_warn(DEV, "drbd_pp_alloc interrupted!\n");
275 finish_wait(&drbd_pp_wait, &wait);
278 atomic_add(number, &mdev->pp_in_use);
282 /* Must not be used from irq, as that may deadlock: see drbd_pp_alloc.
283 * Is also used from inside an other spin_lock_irq(&mdev->req_lock);
284 * Either links the page chain back to the global pool,
285 * or returns all pages to the system. */
286 static void drbd_pp_free(struct drbd_conf *mdev, struct page *page, int is_net)
288 atomic_t *a = is_net ? &mdev->pp_in_use_by_net : &mdev->pp_in_use;
291 if (drbd_pp_vacant > (DRBD_MAX_SEGMENT_SIZE/PAGE_SIZE)*minor_count)
292 i = page_chain_free(page);
295 tmp = page_chain_tail(page, &i);
296 spin_lock(&drbd_pp_lock);
297 page_chain_add(&drbd_pp_pool, page, tmp);
299 spin_unlock(&drbd_pp_lock);
301 i = atomic_sub_return(i, a);
303 dev_warn(DEV, "ASSERTION FAILED: %s: %d < 0\n",
304 is_net ? "pp_in_use_by_net" : "pp_in_use", i);
305 wake_up(&drbd_pp_wait);
309 You need to hold the req_lock:
310 _drbd_wait_ee_list_empty()
312 You must not have the req_lock:
318 drbd_process_done_ee()
320 drbd_wait_ee_list_empty()
323 struct drbd_epoch_entry *drbd_alloc_ee(struct drbd_conf *mdev,
326 unsigned int data_size,
327 gfp_t gfp_mask) __must_hold(local)
329 struct drbd_epoch_entry *e;
331 unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
333 if (FAULT_ACTIVE(mdev, DRBD_FAULT_AL_EE))
336 e = mempool_alloc(drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
338 if (!(gfp_mask & __GFP_NOWARN))
339 dev_err(DEV, "alloc_ee: Allocation of an EE failed\n");
343 page = drbd_pp_alloc(mdev, nr_pages, (gfp_mask & __GFP_WAIT));
347 INIT_HLIST_NODE(&e->colision);
351 atomic_set(&e->pending_bios, 0);
360 mempool_free(e, drbd_ee_mempool);
364 void drbd_free_some_ee(struct drbd_conf *mdev, struct drbd_epoch_entry *e, int is_net)
366 if (e->flags & EE_HAS_DIGEST)
368 drbd_pp_free(mdev, e->pages, is_net);
369 D_ASSERT(atomic_read(&e->pending_bios) == 0);
370 D_ASSERT(hlist_unhashed(&e->colision));
371 mempool_free(e, drbd_ee_mempool);
374 int drbd_release_ee(struct drbd_conf *mdev, struct list_head *list)
376 LIST_HEAD(work_list);
377 struct drbd_epoch_entry *e, *t;
379 int is_net = list == &mdev->net_ee;
381 spin_lock_irq(&mdev->req_lock);
382 list_splice_init(list, &work_list);
383 spin_unlock_irq(&mdev->req_lock);
385 list_for_each_entry_safe(e, t, &work_list, w.list) {
386 drbd_free_some_ee(mdev, e, is_net);
394 * This function is called from _asender only_
395 * but see also comments in _req_mod(,barrier_acked)
396 * and receive_Barrier.
398 * Move entries from net_ee to done_ee, if ready.
399 * Grab done_ee, call all callbacks, free the entries.
400 * The callbacks typically send out ACKs.
402 static int drbd_process_done_ee(struct drbd_conf *mdev)
404 LIST_HEAD(work_list);
405 LIST_HEAD(reclaimed);
406 struct drbd_epoch_entry *e, *t;
407 int ok = (mdev->state.conn >= C_WF_REPORT_PARAMS);
409 spin_lock_irq(&mdev->req_lock);
410 reclaim_net_ee(mdev, &reclaimed);
411 list_splice_init(&mdev->done_ee, &work_list);
412 spin_unlock_irq(&mdev->req_lock);
414 list_for_each_entry_safe(e, t, &reclaimed, w.list)
415 drbd_free_net_ee(mdev, e);
417 /* possible callbacks here:
418 * e_end_block, and e_end_resync_block, e_send_discard_ack.
419 * all ignore the last argument.
421 list_for_each_entry_safe(e, t, &work_list, w.list) {
422 /* list_del not necessary, next/prev members not touched */
423 ok = e->w.cb(mdev, &e->w, !ok) && ok;
424 drbd_free_ee(mdev, e);
426 wake_up(&mdev->ee_wait);
431 void _drbd_wait_ee_list_empty(struct drbd_conf *mdev, struct list_head *head)
435 /* avoids spin_lock/unlock
436 * and calling prepare_to_wait in the fast path */
437 while (!list_empty(head)) {
438 prepare_to_wait(&mdev->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
439 spin_unlock_irq(&mdev->req_lock);
442 finish_wait(&mdev->ee_wait, &wait);
443 spin_lock_irq(&mdev->req_lock);
447 void drbd_wait_ee_list_empty(struct drbd_conf *mdev, struct list_head *head)
449 spin_lock_irq(&mdev->req_lock);
450 _drbd_wait_ee_list_empty(mdev, head);
451 spin_unlock_irq(&mdev->req_lock);
454 /* see also kernel_accept; which is only present since 2.6.18.
455 * also we want to log which part of it failed, exactly */
456 static int drbd_accept(struct drbd_conf *mdev, const char **what,
457 struct socket *sock, struct socket **newsock)
459 struct sock *sk = sock->sk;
463 err = sock->ops->listen(sock, 5);
467 *what = "sock_create_lite";
468 err = sock_create_lite(sk->sk_family, sk->sk_type, sk->sk_protocol,
474 err = sock->ops->accept(sock, *newsock, 0);
476 sock_release(*newsock);
480 (*newsock)->ops = sock->ops;
486 static int drbd_recv_short(struct drbd_conf *mdev, struct socket *sock,
487 void *buf, size_t size, int flags)
494 struct msghdr msg = {
496 .msg_iov = (struct iovec *)&iov,
497 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
503 rv = sock_recvmsg(sock, &msg, size, msg.msg_flags);
509 static int drbd_recv(struct drbd_conf *mdev, void *buf, size_t size)
516 struct msghdr msg = {
518 .msg_iov = (struct iovec *)&iov,
519 .msg_flags = MSG_WAITALL | MSG_NOSIGNAL
527 rv = sock_recvmsg(mdev->data.socket, &msg, size, msg.msg_flags);
532 * ECONNRESET other side closed the connection
533 * ERESTARTSYS (on sock) we got a signal
537 if (rv == -ECONNRESET)
538 dev_info(DEV, "sock was reset by peer\n");
539 else if (rv != -ERESTARTSYS)
540 dev_err(DEV, "sock_recvmsg returned %d\n", rv);
542 } else if (rv == 0) {
543 dev_info(DEV, "sock was shut down by peer\n");
546 /* signal came in, or peer/link went down,
547 * after we read a partial message
549 /* D_ASSERT(signal_pending(current)); */
557 drbd_force_state(mdev, NS(conn, C_BROKEN_PIPE));
563 * On individual connections, the socket buffer size must be set prior to the
564 * listen(2) or connect(2) calls in order to have it take effect.
565 * This is our wrapper to do so.
567 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
570 /* open coded SO_SNDBUF, SO_RCVBUF */
572 sock->sk->sk_sndbuf = snd;
573 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
576 sock->sk->sk_rcvbuf = rcv;
577 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
581 static struct socket *drbd_try_connect(struct drbd_conf *mdev)
585 struct sockaddr_in6 src_in6;
587 int disconnect_on_error = 1;
589 if (!get_net_conf(mdev))
592 what = "sock_create_kern";
593 err = sock_create_kern(((struct sockaddr *)mdev->net_conf->my_addr)->sa_family,
594 SOCK_STREAM, IPPROTO_TCP, &sock);
600 sock->sk->sk_rcvtimeo =
601 sock->sk->sk_sndtimeo = mdev->net_conf->try_connect_int*HZ;
602 drbd_setbufsize(sock, mdev->net_conf->sndbuf_size,
603 mdev->net_conf->rcvbuf_size);
605 /* explicitly bind to the configured IP as source IP
606 * for the outgoing connections.
607 * This is needed for multihomed hosts and to be
608 * able to use lo: interfaces for drbd.
609 * Make sure to use 0 as port number, so linux selects
610 * a free one dynamically.
612 memcpy(&src_in6, mdev->net_conf->my_addr,
613 min_t(int, mdev->net_conf->my_addr_len, sizeof(src_in6)));
614 if (((struct sockaddr *)mdev->net_conf->my_addr)->sa_family == AF_INET6)
615 src_in6.sin6_port = 0;
617 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
619 what = "bind before connect";
620 err = sock->ops->bind(sock,
621 (struct sockaddr *) &src_in6,
622 mdev->net_conf->my_addr_len);
626 /* connect may fail, peer not yet available.
627 * stay C_WF_CONNECTION, don't go Disconnecting! */
628 disconnect_on_error = 0;
630 err = sock->ops->connect(sock,
631 (struct sockaddr *)mdev->net_conf->peer_addr,
632 mdev->net_conf->peer_addr_len, 0);
641 /* timeout, busy, signal pending */
642 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
643 case EINTR: case ERESTARTSYS:
644 /* peer not (yet) available, network problem */
645 case ECONNREFUSED: case ENETUNREACH:
646 case EHOSTDOWN: case EHOSTUNREACH:
647 disconnect_on_error = 0;
650 dev_err(DEV, "%s failed, err = %d\n", what, err);
652 if (disconnect_on_error)
653 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
659 static struct socket *drbd_wait_for_connect(struct drbd_conf *mdev)
662 struct socket *s_estab = NULL, *s_listen;
665 if (!get_net_conf(mdev))
668 what = "sock_create_kern";
669 err = sock_create_kern(((struct sockaddr *)mdev->net_conf->my_addr)->sa_family,
670 SOCK_STREAM, IPPROTO_TCP, &s_listen);
676 timeo = mdev->net_conf->try_connect_int * HZ;
677 timeo += (random32() & 1) ? timeo / 7 : -timeo / 7; /* 28.5% random jitter */
679 s_listen->sk->sk_reuse = 1; /* SO_REUSEADDR */
680 s_listen->sk->sk_rcvtimeo = timeo;
681 s_listen->sk->sk_sndtimeo = timeo;
682 drbd_setbufsize(s_listen, mdev->net_conf->sndbuf_size,
683 mdev->net_conf->rcvbuf_size);
685 what = "bind before listen";
686 err = s_listen->ops->bind(s_listen,
687 (struct sockaddr *) mdev->net_conf->my_addr,
688 mdev->net_conf->my_addr_len);
692 err = drbd_accept(mdev, &what, s_listen, &s_estab);
696 sock_release(s_listen);
698 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
699 dev_err(DEV, "%s failed, err = %d\n", what, err);
700 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
708 static int drbd_send_fp(struct drbd_conf *mdev,
709 struct socket *sock, enum drbd_packets cmd)
711 struct p_header80 *h = &mdev->data.sbuf.header.h80;
713 return _drbd_send_cmd(mdev, sock, cmd, h, sizeof(*h), 0);
716 static enum drbd_packets drbd_recv_fp(struct drbd_conf *mdev, struct socket *sock)
718 struct p_header80 *h = &mdev->data.rbuf.header.h80;
721 rr = drbd_recv_short(mdev, sock, h, sizeof(*h), 0);
723 if (rr == sizeof(*h) && h->magic == BE_DRBD_MAGIC)
724 return be16_to_cpu(h->command);
730 * drbd_socket_okay() - Free the socket if its connection is not okay
731 * @mdev: DRBD device.
732 * @sock: pointer to the pointer to the socket.
734 static int drbd_socket_okay(struct drbd_conf *mdev, struct socket **sock)
742 rr = drbd_recv_short(mdev, *sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
744 if (rr > 0 || rr == -EAGAIN) {
755 * 1 yes, we have a valid connection
756 * 0 oops, did not work out, please try again
757 * -1 peer talks different language,
758 * no point in trying again, please go standalone.
759 * -2 We do not have a network config...
761 static int drbd_connect(struct drbd_conf *mdev)
763 struct socket *s, *sock, *msock;
766 D_ASSERT(!mdev->data.socket);
768 if (drbd_request_state(mdev, NS(conn, C_WF_CONNECTION)) < SS_SUCCESS)
771 clear_bit(DISCARD_CONCURRENT, &mdev->flags);
778 /* 3 tries, this should take less than a second! */
779 s = drbd_try_connect(mdev);
782 /* give the other side time to call bind() & listen() */
783 __set_current_state(TASK_INTERRUPTIBLE);
784 schedule_timeout(HZ / 10);
789 drbd_send_fp(mdev, s, P_HAND_SHAKE_S);
793 drbd_send_fp(mdev, s, P_HAND_SHAKE_M);
797 dev_err(DEV, "Logic error in drbd_connect()\n");
798 goto out_release_sockets;
803 __set_current_state(TASK_INTERRUPTIBLE);
804 schedule_timeout(HZ / 10);
805 ok = drbd_socket_okay(mdev, &sock);
806 ok = drbd_socket_okay(mdev, &msock) && ok;
812 s = drbd_wait_for_connect(mdev);
814 try = drbd_recv_fp(mdev, s);
815 drbd_socket_okay(mdev, &sock);
816 drbd_socket_okay(mdev, &msock);
820 dev_warn(DEV, "initial packet S crossed\n");
827 dev_warn(DEV, "initial packet M crossed\n");
831 set_bit(DISCARD_CONCURRENT, &mdev->flags);
834 dev_warn(DEV, "Error receiving initial packet\n");
841 if (mdev->state.conn <= C_DISCONNECTING)
842 goto out_release_sockets;
843 if (signal_pending(current)) {
844 flush_signals(current);
846 if (get_t_state(&mdev->receiver) == Exiting)
847 goto out_release_sockets;
851 ok = drbd_socket_okay(mdev, &sock);
852 ok = drbd_socket_okay(mdev, &msock) && ok;
858 msock->sk->sk_reuse = 1; /* SO_REUSEADDR */
859 sock->sk->sk_reuse = 1; /* SO_REUSEADDR */
861 sock->sk->sk_allocation = GFP_NOIO;
862 msock->sk->sk_allocation = GFP_NOIO;
864 sock->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
865 msock->sk->sk_priority = TC_PRIO_INTERACTIVE;
868 * sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
869 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
870 * first set it to the P_HAND_SHAKE timeout,
871 * which we set to 4x the configured ping_timeout. */
872 sock->sk->sk_sndtimeo =
873 sock->sk->sk_rcvtimeo = mdev->net_conf->ping_timeo*4*HZ/10;
875 msock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
876 msock->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ;
878 /* we don't want delays.
879 * we use TCP_CORK where apropriate, though */
880 drbd_tcp_nodelay(sock);
881 drbd_tcp_nodelay(msock);
883 mdev->data.socket = sock;
884 mdev->meta.socket = msock;
885 mdev->last_received = jiffies;
887 D_ASSERT(mdev->asender.task == NULL);
889 h = drbd_do_handshake(mdev);
893 if (mdev->cram_hmac_tfm) {
894 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
895 switch (drbd_do_auth(mdev)) {
897 dev_err(DEV, "Authentication of peer failed\n");
900 dev_err(DEV, "Authentication of peer failed, trying again.\n");
905 if (drbd_request_state(mdev, NS(conn, C_WF_REPORT_PARAMS)) < SS_SUCCESS)
908 sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
909 sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
911 atomic_set(&mdev->packet_seq, 0);
914 drbd_thread_start(&mdev->asender);
916 if (mdev->agreed_pro_version < 95 && get_ldev(mdev)) {
917 drbd_setup_queue_param(mdev, DRBD_MAX_SIZE_H80_PACKET);
921 if (!drbd_send_protocol(mdev))
923 drbd_send_sync_param(mdev, &mdev->sync_conf);
924 drbd_send_sizes(mdev, 0, 0);
925 drbd_send_uuids(mdev);
926 drbd_send_state(mdev);
927 clear_bit(USE_DEGR_WFC_T, &mdev->flags);
928 clear_bit(RESIZE_PENDING, &mdev->flags);
940 static int drbd_recv_header(struct drbd_conf *mdev, enum drbd_packets *cmd, unsigned int *packet_size)
942 union p_header *h = &mdev->data.rbuf.header;
945 r = drbd_recv(mdev, h, sizeof(*h));
946 if (unlikely(r != sizeof(*h))) {
947 dev_err(DEV, "short read expecting header on sock: r=%d\n", r);
951 if (likely(h->h80.magic == BE_DRBD_MAGIC)) {
952 *cmd = be16_to_cpu(h->h80.command);
953 *packet_size = be16_to_cpu(h->h80.length);
954 } else if (h->h95.magic == BE_DRBD_MAGIC_BIG) {
955 *cmd = be16_to_cpu(h->h95.command);
956 *packet_size = be32_to_cpu(h->h95.length);
958 dev_err(DEV, "magic?? on data m: 0x%08x c: %d l: %d\n",
959 be32_to_cpu(h->h80.magic),
960 be16_to_cpu(h->h80.command),
961 be16_to_cpu(h->h80.length));
964 mdev->last_received = jiffies;
969 static void drbd_flush(struct drbd_conf *mdev)
973 if (mdev->write_ordering >= WO_bdev_flush && get_ldev(mdev)) {
974 rv = blkdev_issue_flush(mdev->ldev->backing_bdev, GFP_KERNEL,
977 dev_err(DEV, "local disk flush failed with status %d\n", rv);
978 /* would rather check on EOPNOTSUPP, but that is not reliable.
979 * don't try again for ANY return value != 0
980 * if (rv == -EOPNOTSUPP) */
981 drbd_bump_write_ordering(mdev, WO_drain_io);
988 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
989 * @mdev: DRBD device.
990 * @epoch: Epoch object.
993 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *mdev,
994 struct drbd_epoch *epoch,
998 struct drbd_epoch *next_epoch;
999 enum finish_epoch rv = FE_STILL_LIVE;
1001 spin_lock(&mdev->epoch_lock);
1005 epoch_size = atomic_read(&epoch->epoch_size);
1007 switch (ev & ~EV_CLEANUP) {
1009 atomic_dec(&epoch->active);
1011 case EV_GOT_BARRIER_NR:
1012 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1014 case EV_BECAME_LAST:
1019 if (epoch_size != 0 &&
1020 atomic_read(&epoch->active) == 0 &&
1021 test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags)) {
1022 if (!(ev & EV_CLEANUP)) {
1023 spin_unlock(&mdev->epoch_lock);
1024 drbd_send_b_ack(mdev, epoch->barrier_nr, epoch_size);
1025 spin_lock(&mdev->epoch_lock);
1029 if (mdev->current_epoch != epoch) {
1030 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1031 list_del(&epoch->list);
1032 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1036 if (rv == FE_STILL_LIVE)
1040 atomic_set(&epoch->epoch_size, 0);
1041 /* atomic_set(&epoch->active, 0); is already zero */
1042 if (rv == FE_STILL_LIVE)
1044 wake_up(&mdev->ee_wait);
1054 spin_unlock(&mdev->epoch_lock);
1060 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1061 * @mdev: DRBD device.
1062 * @wo: Write ordering method to try.
1064 void drbd_bump_write_ordering(struct drbd_conf *mdev, enum write_ordering_e wo) __must_hold(local)
1066 enum write_ordering_e pwo;
1067 static char *write_ordering_str[] = {
1069 [WO_drain_io] = "drain",
1070 [WO_bdev_flush] = "flush",
1073 pwo = mdev->write_ordering;
1075 if (wo == WO_bdev_flush && mdev->ldev->dc.no_disk_flush)
1077 if (wo == WO_drain_io && mdev->ldev->dc.no_disk_drain)
1079 mdev->write_ordering = wo;
1080 if (pwo != mdev->write_ordering || wo == WO_bdev_flush)
1081 dev_info(DEV, "Method to ensure write ordering: %s\n", write_ordering_str[mdev->write_ordering]);
1086 * @mdev: DRBD device.
1088 * @rw: flag field, see bio->bi_rw
1090 /* TODO allocate from our own bio_set. */
1091 int drbd_submit_ee(struct drbd_conf *mdev, struct drbd_epoch_entry *e,
1092 const unsigned rw, const int fault_type)
1094 struct bio *bios = NULL;
1096 struct page *page = e->pages;
1097 sector_t sector = e->sector;
1098 unsigned ds = e->size;
1099 unsigned n_bios = 0;
1100 unsigned nr_pages = (ds + PAGE_SIZE -1) >> PAGE_SHIFT;
1102 /* In most cases, we will only need one bio. But in case the lower
1103 * level restrictions happen to be different at this offset on this
1104 * side than those of the sending peer, we may need to submit the
1105 * request in more than one bio. */
1107 bio = bio_alloc(GFP_NOIO, nr_pages);
1109 dev_err(DEV, "submit_ee: Allocation of a bio failed\n");
1112 /* > e->sector, unless this is the first bio */
1113 bio->bi_sector = sector;
1114 bio->bi_bdev = mdev->ldev->backing_bdev;
1115 /* we special case some flags in the multi-bio case, see below
1118 bio->bi_private = e;
1119 bio->bi_end_io = drbd_endio_sec;
1121 bio->bi_next = bios;
1125 page_chain_for_each(page) {
1126 unsigned len = min_t(unsigned, ds, PAGE_SIZE);
1127 if (!bio_add_page(bio, page, len, 0)) {
1128 /* a single page must always be possible! */
1129 BUG_ON(bio->bi_vcnt == 0);
1136 D_ASSERT(page == NULL);
1139 atomic_set(&e->pending_bios, n_bios);
1142 bios = bios->bi_next;
1143 bio->bi_next = NULL;
1145 /* strip off REQ_UNPLUG unless it is the last bio */
1147 bio->bi_rw &= ~REQ_UNPLUG;
1149 drbd_generic_make_request(mdev, fault_type, bio);
1151 maybe_kick_lo(mdev);
1157 bios = bios->bi_next;
1163 static int receive_Barrier(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
1166 struct p_barrier *p = &mdev->data.rbuf.barrier;
1167 struct drbd_epoch *epoch;
1171 if (mdev->net_conf->wire_protocol != DRBD_PROT_C)
1174 mdev->current_epoch->barrier_nr = p->barrier;
1175 rv = drbd_may_finish_epoch(mdev, mdev->current_epoch, EV_GOT_BARRIER_NR);
1177 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1178 * the activity log, which means it would not be resynced in case the
1179 * R_PRIMARY crashes now.
1180 * Therefore we must send the barrier_ack after the barrier request was
1182 switch (mdev->write_ordering) {
1184 if (rv == FE_RECYCLED)
1187 /* receiver context, in the writeout path of the other node.
1188 * avoid potential distributed deadlock */
1189 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1193 dev_warn(DEV, "Allocation of an epoch failed, slowing down\n");
1198 drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
1201 if (atomic_read(&mdev->current_epoch->epoch_size)) {
1202 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1207 epoch = mdev->current_epoch;
1208 wait_event(mdev->ee_wait, atomic_read(&epoch->epoch_size) == 0);
1210 D_ASSERT(atomic_read(&epoch->active) == 0);
1211 D_ASSERT(epoch->flags == 0);
1215 dev_err(DEV, "Strangeness in mdev->write_ordering %d\n", mdev->write_ordering);
1220 atomic_set(&epoch->epoch_size, 0);
1221 atomic_set(&epoch->active, 0);
1223 spin_lock(&mdev->epoch_lock);
1224 if (atomic_read(&mdev->current_epoch->epoch_size)) {
1225 list_add(&epoch->list, &mdev->current_epoch->list);
1226 mdev->current_epoch = epoch;
1229 /* The current_epoch got recycled while we allocated this one... */
1232 spin_unlock(&mdev->epoch_lock);
1237 /* used from receive_RSDataReply (recv_resync_read)
1238 * and from receive_Data */
1239 static struct drbd_epoch_entry *
1240 read_in_block(struct drbd_conf *mdev, u64 id, sector_t sector, int data_size) __must_hold(local)
1242 const sector_t capacity = drbd_get_capacity(mdev->this_bdev);
1243 struct drbd_epoch_entry *e;
1246 void *dig_in = mdev->int_dig_in;
1247 void *dig_vv = mdev->int_dig_vv;
1248 unsigned long *data;
1250 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
1251 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
1254 rr = drbd_recv(mdev, dig_in, dgs);
1256 dev_warn(DEV, "short read receiving data digest: read %d expected %d\n",
1264 ERR_IF(data_size & 0x1ff) return NULL;
1265 ERR_IF(data_size > DRBD_MAX_SEGMENT_SIZE) return NULL;
1267 /* even though we trust out peer,
1268 * we sometimes have to double check. */
1269 if (sector + (data_size>>9) > capacity) {
1270 dev_err(DEV, "capacity: %llus < sector: %llus + size: %u\n",
1271 (unsigned long long)capacity,
1272 (unsigned long long)sector, data_size);
1276 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1277 * "criss-cross" setup, that might cause write-out on some other DRBD,
1278 * which in turn might block on the other node at this very place. */
1279 e = drbd_alloc_ee(mdev, id, sector, data_size, GFP_NOIO);
1285 page_chain_for_each(page) {
1286 unsigned len = min_t(int, ds, PAGE_SIZE);
1288 rr = drbd_recv(mdev, data, len);
1289 if (FAULT_ACTIVE(mdev, DRBD_FAULT_RECEIVE)) {
1290 dev_err(DEV, "Fault injection: Corrupting data on receive\n");
1291 data[0] = data[0] ^ (unsigned long)-1;
1295 drbd_free_ee(mdev, e);
1296 dev_warn(DEV, "short read receiving data: read %d expected %d\n",
1304 drbd_csum_ee(mdev, mdev->integrity_r_tfm, e, dig_vv);
1305 if (memcmp(dig_in, dig_vv, dgs)) {
1306 dev_err(DEV, "Digest integrity check FAILED.\n");
1307 drbd_bcast_ee(mdev, "digest failed",
1308 dgs, dig_in, dig_vv, e);
1309 drbd_free_ee(mdev, e);
1313 mdev->recv_cnt += data_size>>9;
1317 /* drbd_drain_block() just takes a data block
1318 * out of the socket input buffer, and discards it.
1320 static int drbd_drain_block(struct drbd_conf *mdev, int data_size)
1329 page = drbd_pp_alloc(mdev, 1, 1);
1333 rr = drbd_recv(mdev, data, min_t(int, data_size, PAGE_SIZE));
1334 if (rr != min_t(int, data_size, PAGE_SIZE)) {
1336 dev_warn(DEV, "short read receiving data: read %d expected %d\n",
1337 rr, min_t(int, data_size, PAGE_SIZE));
1343 drbd_pp_free(mdev, page, 0);
1347 static int recv_dless_read(struct drbd_conf *mdev, struct drbd_request *req,
1348 sector_t sector, int data_size)
1350 struct bio_vec *bvec;
1352 int dgs, rr, i, expect;
1353 void *dig_in = mdev->int_dig_in;
1354 void *dig_vv = mdev->int_dig_vv;
1356 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
1357 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
1360 rr = drbd_recv(mdev, dig_in, dgs);
1362 dev_warn(DEV, "short read receiving data reply digest: read %d expected %d\n",
1370 /* optimistically update recv_cnt. if receiving fails below,
1371 * we disconnect anyways, and counters will be reset. */
1372 mdev->recv_cnt += data_size>>9;
1374 bio = req->master_bio;
1375 D_ASSERT(sector == bio->bi_sector);
1377 bio_for_each_segment(bvec, bio, i) {
1378 expect = min_t(int, data_size, bvec->bv_len);
1379 rr = drbd_recv(mdev,
1380 kmap(bvec->bv_page)+bvec->bv_offset,
1382 kunmap(bvec->bv_page);
1384 dev_warn(DEV, "short read receiving data reply: "
1385 "read %d expected %d\n",
1393 drbd_csum_bio(mdev, mdev->integrity_r_tfm, bio, dig_vv);
1394 if (memcmp(dig_in, dig_vv, dgs)) {
1395 dev_err(DEV, "Digest integrity check FAILED. Broken NICs?\n");
1400 D_ASSERT(data_size == 0);
1404 /* e_end_resync_block() is called via
1405 * drbd_process_done_ee() by asender only */
1406 static int e_end_resync_block(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1408 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1409 sector_t sector = e->sector;
1412 D_ASSERT(hlist_unhashed(&e->colision));
1414 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
1415 drbd_set_in_sync(mdev, sector, e->size);
1416 ok = drbd_send_ack(mdev, P_RS_WRITE_ACK, e);
1418 /* Record failure to sync */
1419 drbd_rs_failed_io(mdev, sector, e->size);
1421 ok = drbd_send_ack(mdev, P_NEG_ACK, e);
1428 static int recv_resync_read(struct drbd_conf *mdev, sector_t sector, int data_size) __releases(local)
1430 struct drbd_epoch_entry *e;
1432 e = read_in_block(mdev, ID_SYNCER, sector, data_size);
1436 dec_rs_pending(mdev);
1439 /* corresponding dec_unacked() in e_end_resync_block()
1440 * respective _drbd_clear_done_ee */
1442 e->w.cb = e_end_resync_block;
1444 spin_lock_irq(&mdev->req_lock);
1445 list_add(&e->w.list, &mdev->sync_ee);
1446 spin_unlock_irq(&mdev->req_lock);
1448 atomic_add(data_size >> 9, &mdev->rs_sect_ev);
1449 if (drbd_submit_ee(mdev, e, WRITE, DRBD_FAULT_RS_WR) == 0)
1452 /* drbd_submit_ee currently fails for one reason only:
1453 * not being able to allocate enough bios.
1454 * Is dropping the connection going to help? */
1455 spin_lock_irq(&mdev->req_lock);
1456 list_del(&e->w.list);
1457 spin_unlock_irq(&mdev->req_lock);
1459 drbd_free_ee(mdev, e);
1465 static int receive_DataReply(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
1467 struct drbd_request *req;
1470 struct p_data *p = &mdev->data.rbuf.data;
1472 sector = be64_to_cpu(p->sector);
1474 spin_lock_irq(&mdev->req_lock);
1475 req = _ar_id_to_req(mdev, p->block_id, sector);
1476 spin_unlock_irq(&mdev->req_lock);
1477 if (unlikely(!req)) {
1478 dev_err(DEV, "Got a corrupt block_id/sector pair(1).\n");
1482 /* hlist_del(&req->colision) is done in _req_may_be_done, to avoid
1483 * special casing it there for the various failure cases.
1484 * still no race with drbd_fail_pending_reads */
1485 ok = recv_dless_read(mdev, req, sector, data_size);
1488 req_mod(req, data_received);
1489 /* else: nothing. handled from drbd_disconnect...
1490 * I don't think we may complete this just yet
1491 * in case we are "on-disconnect: freeze" */
1496 static int receive_RSDataReply(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
1500 struct p_data *p = &mdev->data.rbuf.data;
1502 sector = be64_to_cpu(p->sector);
1503 D_ASSERT(p->block_id == ID_SYNCER);
1505 if (get_ldev(mdev)) {
1506 /* data is submitted to disk within recv_resync_read.
1507 * corresponding put_ldev done below on error,
1508 * or in drbd_endio_write_sec. */
1509 ok = recv_resync_read(mdev, sector, data_size);
1511 if (__ratelimit(&drbd_ratelimit_state))
1512 dev_err(DEV, "Can not write resync data to local disk.\n");
1514 ok = drbd_drain_block(mdev, data_size);
1516 drbd_send_ack_dp(mdev, P_NEG_ACK, p, data_size);
1519 atomic_add(data_size >> 9, &mdev->rs_sect_in);
1524 /* e_end_block() is called via drbd_process_done_ee().
1525 * this means this function only runs in the asender thread
1527 static int e_end_block(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1529 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1530 sector_t sector = e->sector;
1533 if (mdev->net_conf->wire_protocol == DRBD_PROT_C) {
1534 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
1535 pcmd = (mdev->state.conn >= C_SYNC_SOURCE &&
1536 mdev->state.conn <= C_PAUSED_SYNC_T &&
1537 e->flags & EE_MAY_SET_IN_SYNC) ?
1538 P_RS_WRITE_ACK : P_WRITE_ACK;
1539 ok &= drbd_send_ack(mdev, pcmd, e);
1540 if (pcmd == P_RS_WRITE_ACK)
1541 drbd_set_in_sync(mdev, sector, e->size);
1543 ok = drbd_send_ack(mdev, P_NEG_ACK, e);
1544 /* we expect it to be marked out of sync anyways...
1545 * maybe assert this? */
1549 /* we delete from the conflict detection hash _after_ we sent out the
1550 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1551 if (mdev->net_conf->two_primaries) {
1552 spin_lock_irq(&mdev->req_lock);
1553 D_ASSERT(!hlist_unhashed(&e->colision));
1554 hlist_del_init(&e->colision);
1555 spin_unlock_irq(&mdev->req_lock);
1557 D_ASSERT(hlist_unhashed(&e->colision));
1560 drbd_may_finish_epoch(mdev, e->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
1565 static int e_send_discard_ack(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1567 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1570 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
1571 ok = drbd_send_ack(mdev, P_DISCARD_ACK, e);
1573 spin_lock_irq(&mdev->req_lock);
1574 D_ASSERT(!hlist_unhashed(&e->colision));
1575 hlist_del_init(&e->colision);
1576 spin_unlock_irq(&mdev->req_lock);
1583 /* Called from receive_Data.
1584 * Synchronize packets on sock with packets on msock.
1586 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1587 * packet traveling on msock, they are still processed in the order they have
1590 * Note: we don't care for Ack packets overtaking P_DATA packets.
1592 * In case packet_seq is larger than mdev->peer_seq number, there are
1593 * outstanding packets on the msock. We wait for them to arrive.
1594 * In case we are the logically next packet, we update mdev->peer_seq
1595 * ourselves. Correctly handles 32bit wrap around.
1597 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1598 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1599 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1600 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1602 * returns 0 if we may process the packet,
1603 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1604 static int drbd_wait_peer_seq(struct drbd_conf *mdev, const u32 packet_seq)
1610 spin_lock(&mdev->peer_seq_lock);
1612 prepare_to_wait(&mdev->seq_wait, &wait, TASK_INTERRUPTIBLE);
1613 if (seq_le(packet_seq, mdev->peer_seq+1))
1615 if (signal_pending(current)) {
1619 p_seq = mdev->peer_seq;
1620 spin_unlock(&mdev->peer_seq_lock);
1621 timeout = schedule_timeout(30*HZ);
1622 spin_lock(&mdev->peer_seq_lock);
1623 if (timeout == 0 && p_seq == mdev->peer_seq) {
1625 dev_err(DEV, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n");
1629 finish_wait(&mdev->seq_wait, &wait);
1630 if (mdev->peer_seq+1 == packet_seq)
1632 spin_unlock(&mdev->peer_seq_lock);
1636 static unsigned long write_flags_to_bio(struct drbd_conf *mdev, u32 dpf)
1638 if (mdev->agreed_pro_version >= 95)
1639 return (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
1640 (dpf & DP_UNPLUG ? REQ_UNPLUG : 0) |
1641 (dpf & DP_FUA ? REQ_FUA : 0) |
1642 (dpf & DP_FLUSH ? REQ_FUA : 0) |
1643 (dpf & DP_DISCARD ? REQ_DISCARD : 0);
1645 return dpf & DP_RW_SYNC ? (REQ_SYNC | REQ_UNPLUG) : 0;
1648 /* mirrored write */
1649 static int receive_Data(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
1652 struct drbd_epoch_entry *e;
1653 struct p_data *p = &mdev->data.rbuf.data;
1657 if (!get_ldev(mdev)) {
1658 if (__ratelimit(&drbd_ratelimit_state))
1659 dev_err(DEV, "Can not write mirrored data block "
1660 "to local disk.\n");
1661 spin_lock(&mdev->peer_seq_lock);
1662 if (mdev->peer_seq+1 == be32_to_cpu(p->seq_num))
1664 spin_unlock(&mdev->peer_seq_lock);
1666 drbd_send_ack_dp(mdev, P_NEG_ACK, p, data_size);
1667 atomic_inc(&mdev->current_epoch->epoch_size);
1668 return drbd_drain_block(mdev, data_size);
1671 /* get_ldev(mdev) successful.
1672 * Corresponding put_ldev done either below (on various errors),
1673 * or in drbd_endio_write_sec, if we successfully submit the data at
1674 * the end of this function. */
1676 sector = be64_to_cpu(p->sector);
1677 e = read_in_block(mdev, p->block_id, sector, data_size);
1683 e->w.cb = e_end_block;
1685 spin_lock(&mdev->epoch_lock);
1686 e->epoch = mdev->current_epoch;
1687 atomic_inc(&e->epoch->epoch_size);
1688 atomic_inc(&e->epoch->active);
1689 spin_unlock(&mdev->epoch_lock);
1691 dp_flags = be32_to_cpu(p->dp_flags);
1692 rw |= write_flags_to_bio(mdev, dp_flags);
1694 if (dp_flags & DP_MAY_SET_IN_SYNC)
1695 e->flags |= EE_MAY_SET_IN_SYNC;
1697 /* I'm the receiver, I do hold a net_cnt reference. */
1698 if (!mdev->net_conf->two_primaries) {
1699 spin_lock_irq(&mdev->req_lock);
1701 /* don't get the req_lock yet,
1702 * we may sleep in drbd_wait_peer_seq */
1703 const int size = e->size;
1704 const int discard = test_bit(DISCARD_CONCURRENT, &mdev->flags);
1706 struct drbd_request *i;
1707 struct hlist_node *n;
1708 struct hlist_head *slot;
1711 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
1712 BUG_ON(mdev->ee_hash == NULL);
1713 BUG_ON(mdev->tl_hash == NULL);
1715 /* conflict detection and handling:
1716 * 1. wait on the sequence number,
1717 * in case this data packet overtook ACK packets.
1718 * 2. check our hash tables for conflicting requests.
1719 * we only need to walk the tl_hash, since an ee can not
1720 * have a conflict with an other ee: on the submitting
1721 * node, the corresponding req had already been conflicting,
1722 * and a conflicting req is never sent.
1724 * Note: for two_primaries, we are protocol C,
1725 * so there cannot be any request that is DONE
1726 * but still on the transfer log.
1728 * unconditionally add to the ee_hash.
1730 * if no conflicting request is found:
1733 * if any conflicting request is found
1734 * that has not yet been acked,
1735 * AND I have the "discard concurrent writes" flag:
1736 * queue (via done_ee) the P_DISCARD_ACK; OUT.
1738 * if any conflicting request is found:
1739 * block the receiver, waiting on misc_wait
1740 * until no more conflicting requests are there,
1741 * or we get interrupted (disconnect).
1743 * we do not just write after local io completion of those
1744 * requests, but only after req is done completely, i.e.
1745 * we wait for the P_DISCARD_ACK to arrive!
1747 * then proceed normally, i.e. submit.
1749 if (drbd_wait_peer_seq(mdev, be32_to_cpu(p->seq_num)))
1750 goto out_interrupted;
1752 spin_lock_irq(&mdev->req_lock);
1754 hlist_add_head(&e->colision, ee_hash_slot(mdev, sector));
1756 #define OVERLAPS overlaps(i->sector, i->size, sector, size)
1757 slot = tl_hash_slot(mdev, sector);
1760 int have_unacked = 0;
1761 int have_conflict = 0;
1762 prepare_to_wait(&mdev->misc_wait, &wait,
1763 TASK_INTERRUPTIBLE);
1764 hlist_for_each_entry(i, n, slot, colision) {
1766 /* only ALERT on first iteration,
1767 * we may be woken up early... */
1769 dev_alert(DEV, "%s[%u] Concurrent local write detected!"
1770 " new: %llus +%u; pending: %llus +%u\n",
1771 current->comm, current->pid,
1772 (unsigned long long)sector, size,
1773 (unsigned long long)i->sector, i->size);
1774 if (i->rq_state & RQ_NET_PENDING)
1783 /* Discard Ack only for the _first_ iteration */
1784 if (first && discard && have_unacked) {
1785 dev_alert(DEV, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n",
1786 (unsigned long long)sector);
1788 e->w.cb = e_send_discard_ack;
1789 list_add_tail(&e->w.list, &mdev->done_ee);
1791 spin_unlock_irq(&mdev->req_lock);
1793 /* we could probably send that P_DISCARD_ACK ourselves,
1794 * but I don't like the receiver using the msock */
1798 finish_wait(&mdev->misc_wait, &wait);
1802 if (signal_pending(current)) {
1803 hlist_del_init(&e->colision);
1805 spin_unlock_irq(&mdev->req_lock);
1807 finish_wait(&mdev->misc_wait, &wait);
1808 goto out_interrupted;
1811 spin_unlock_irq(&mdev->req_lock);
1814 dev_alert(DEV, "Concurrent write! [W AFTERWARDS] "
1815 "sec=%llus\n", (unsigned long long)sector);
1816 } else if (discard) {
1817 /* we had none on the first iteration.
1818 * there must be none now. */
1819 D_ASSERT(have_unacked == 0);
1822 spin_lock_irq(&mdev->req_lock);
1824 finish_wait(&mdev->misc_wait, &wait);
1827 list_add(&e->w.list, &mdev->active_ee);
1828 spin_unlock_irq(&mdev->req_lock);
1830 switch (mdev->net_conf->wire_protocol) {
1833 /* corresponding dec_unacked() in e_end_block()
1834 * respective _drbd_clear_done_ee */
1837 /* I really don't like it that the receiver thread
1838 * sends on the msock, but anyways */
1839 drbd_send_ack(mdev, P_RECV_ACK, e);
1846 if (mdev->state.pdsk < D_INCONSISTENT) {
1847 /* In case we have the only disk of the cluster, */
1848 drbd_set_out_of_sync(mdev, e->sector, e->size);
1849 e->flags |= EE_CALL_AL_COMPLETE_IO;
1850 e->flags &= ~EE_MAY_SET_IN_SYNC;
1851 drbd_al_begin_io(mdev, e->sector);
1854 if (drbd_submit_ee(mdev, e, rw, DRBD_FAULT_DT_WR) == 0)
1857 /* drbd_submit_ee currently fails for one reason only:
1858 * not being able to allocate enough bios.
1859 * Is dropping the connection going to help? */
1860 spin_lock_irq(&mdev->req_lock);
1861 list_del(&e->w.list);
1862 hlist_del_init(&e->colision);
1863 spin_unlock_irq(&mdev->req_lock);
1864 if (e->flags & EE_CALL_AL_COMPLETE_IO)
1865 drbd_al_complete_io(mdev, e->sector);
1868 /* yes, the epoch_size now is imbalanced.
1869 * but we drop the connection anyways, so we don't have a chance to
1870 * receive a barrier... atomic_inc(&mdev->epoch_size); */
1872 drbd_free_ee(mdev, e);
1876 /* We may throttle resync, if the lower device seems to be busy,
1877 * and current sync rate is above c_min_rate.
1879 * To decide whether or not the lower device is busy, we use a scheme similar
1880 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
1881 * (more than 64 sectors) of activity we cannot account for with our own resync
1882 * activity, it obviously is "busy".
1884 * The current sync rate used here uses only the most recent two step marks,
1885 * to have a short time average so we can react faster.
1887 int drbd_rs_should_slow_down(struct drbd_conf *mdev)
1889 struct gendisk *disk = mdev->ldev->backing_bdev->bd_contains->bd_disk;
1890 unsigned long db, dt, dbdt;
1894 /* feature disabled? */
1895 if (mdev->sync_conf.c_min_rate == 0)
1898 curr_events = (int)part_stat_read(&disk->part0, sectors[0]) +
1899 (int)part_stat_read(&disk->part0, sectors[1]) -
1900 atomic_read(&mdev->rs_sect_ev);
1901 if (!mdev->rs_last_events || curr_events - mdev->rs_last_events > 64) {
1902 unsigned long rs_left;
1905 mdev->rs_last_events = curr_events;
1907 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
1909 i = (mdev->rs_last_mark + DRBD_SYNC_MARKS-2) % DRBD_SYNC_MARKS;
1910 rs_left = drbd_bm_total_weight(mdev) - mdev->rs_failed;
1912 dt = ((long)jiffies - (long)mdev->rs_mark_time[i]) / HZ;
1915 db = mdev->rs_mark_left[i] - rs_left;
1916 dbdt = Bit2KB(db/dt);
1918 if (dbdt > mdev->sync_conf.c_min_rate)
1925 static int receive_DataRequest(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int digest_size)
1928 const sector_t capacity = drbd_get_capacity(mdev->this_bdev);
1929 struct drbd_epoch_entry *e;
1930 struct digest_info *di = NULL;
1932 unsigned int fault_type;
1933 struct p_block_req *p = &mdev->data.rbuf.block_req;
1935 sector = be64_to_cpu(p->sector);
1936 size = be32_to_cpu(p->blksize);
1938 if (size <= 0 || (size & 0x1ff) != 0 || size > DRBD_MAX_SEGMENT_SIZE) {
1939 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
1940 (unsigned long long)sector, size);
1943 if (sector + (size>>9) > capacity) {
1944 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
1945 (unsigned long long)sector, size);
1949 if (!get_ldev_if_state(mdev, D_UP_TO_DATE)) {
1952 case P_DATA_REQUEST:
1953 drbd_send_ack_rp(mdev, P_NEG_DREPLY, p);
1955 case P_RS_DATA_REQUEST:
1956 case P_CSUM_RS_REQUEST:
1958 drbd_send_ack_rp(mdev, P_NEG_RS_DREPLY , p);
1962 dec_rs_pending(mdev);
1963 drbd_send_ack_ex(mdev, P_OV_RESULT, sector, size, ID_IN_SYNC);
1966 dev_err(DEV, "unexpected command (%s) in receive_DataRequest\n",
1969 if (verb && __ratelimit(&drbd_ratelimit_state))
1970 dev_err(DEV, "Can not satisfy peer's read request, "
1971 "no local data.\n");
1973 /* drain possibly payload */
1974 return drbd_drain_block(mdev, digest_size);
1977 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1978 * "criss-cross" setup, that might cause write-out on some other DRBD,
1979 * which in turn might block on the other node at this very place. */
1980 e = drbd_alloc_ee(mdev, p->block_id, sector, size, GFP_NOIO);
1987 case P_DATA_REQUEST:
1988 e->w.cb = w_e_end_data_req;
1989 fault_type = DRBD_FAULT_DT_RD;
1990 /* application IO, don't drbd_rs_begin_io */
1993 case P_RS_DATA_REQUEST:
1994 e->w.cb = w_e_end_rsdata_req;
1995 fault_type = DRBD_FAULT_RS_RD;
1999 case P_CSUM_RS_REQUEST:
2000 fault_type = DRBD_FAULT_RS_RD;
2001 di = kmalloc(sizeof(*di) + digest_size, GFP_NOIO);
2005 di->digest_size = digest_size;
2006 di->digest = (((char *)di)+sizeof(struct digest_info));
2009 e->flags |= EE_HAS_DIGEST;
2011 if (drbd_recv(mdev, di->digest, digest_size) != digest_size)
2014 if (cmd == P_CSUM_RS_REQUEST) {
2015 D_ASSERT(mdev->agreed_pro_version >= 89);
2016 e->w.cb = w_e_end_csum_rs_req;
2017 } else if (cmd == P_OV_REPLY) {
2018 e->w.cb = w_e_end_ov_reply;
2019 dec_rs_pending(mdev);
2020 /* drbd_rs_begin_io done when we sent this request,
2021 * but accounting still needs to be done. */
2022 goto submit_for_resync;
2027 if (mdev->ov_start_sector == ~(sector_t)0 &&
2028 mdev->agreed_pro_version >= 90) {
2029 mdev->ov_start_sector = sector;
2030 mdev->ov_position = sector;
2031 mdev->ov_left = mdev->rs_total - BM_SECT_TO_BIT(sector);
2032 dev_info(DEV, "Online Verify start sector: %llu\n",
2033 (unsigned long long)sector);
2035 e->w.cb = w_e_end_ov_req;
2036 fault_type = DRBD_FAULT_RS_RD;
2040 dev_err(DEV, "unexpected command (%s) in receive_DataRequest\n",
2042 fault_type = DRBD_FAULT_MAX;
2046 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2047 * wrt the receiver, but it is not as straightforward as it may seem.
2048 * Various places in the resync start and stop logic assume resync
2049 * requests are processed in order, requeuing this on the worker thread
2050 * introduces a bunch of new code for synchronization between threads.
2052 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2053 * "forever", throttling after drbd_rs_begin_io will lock that extent
2054 * for application writes for the same time. For now, just throttle
2055 * here, where the rest of the code expects the receiver to sleep for
2059 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2060 * this defers syncer requests for some time, before letting at least
2061 * on request through. The resync controller on the receiving side
2062 * will adapt to the incoming rate accordingly.
2064 * We cannot throttle here if remote is Primary/SyncTarget:
2065 * we would also throttle its application reads.
2066 * In that case, throttling is done on the SyncTarget only.
2068 if (mdev->state.peer != R_PRIMARY && drbd_rs_should_slow_down(mdev))
2070 if (drbd_rs_begin_io(mdev, e->sector))
2074 atomic_add(size >> 9, &mdev->rs_sect_ev);
2078 spin_lock_irq(&mdev->req_lock);
2079 list_add_tail(&e->w.list, &mdev->read_ee);
2080 spin_unlock_irq(&mdev->req_lock);
2082 if (drbd_submit_ee(mdev, e, READ, fault_type) == 0)
2085 /* drbd_submit_ee currently fails for one reason only:
2086 * not being able to allocate enough bios.
2087 * Is dropping the connection going to help? */
2088 spin_lock_irq(&mdev->req_lock);
2089 list_del(&e->w.list);
2090 spin_unlock_irq(&mdev->req_lock);
2091 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2095 drbd_free_ee(mdev, e);
2099 static int drbd_asb_recover_0p(struct drbd_conf *mdev) __must_hold(local)
2101 int self, peer, rv = -100;
2102 unsigned long ch_self, ch_peer;
2104 self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2105 peer = mdev->p_uuid[UI_BITMAP] & 1;
2107 ch_peer = mdev->p_uuid[UI_SIZE];
2108 ch_self = mdev->comm_bm_set;
2110 switch (mdev->net_conf->after_sb_0p) {
2112 case ASB_DISCARD_SECONDARY:
2113 case ASB_CALL_HELPER:
2114 dev_err(DEV, "Configuration error.\n");
2116 case ASB_DISCONNECT:
2118 case ASB_DISCARD_YOUNGER_PRI:
2119 if (self == 0 && peer == 1) {
2123 if (self == 1 && peer == 0) {
2127 /* Else fall through to one of the other strategies... */
2128 case ASB_DISCARD_OLDER_PRI:
2129 if (self == 0 && peer == 1) {
2133 if (self == 1 && peer == 0) {
2137 /* Else fall through to one of the other strategies... */
2138 dev_warn(DEV, "Discard younger/older primary did not find a decision\n"
2139 "Using discard-least-changes instead\n");
2140 case ASB_DISCARD_ZERO_CHG:
2141 if (ch_peer == 0 && ch_self == 0) {
2142 rv = test_bit(DISCARD_CONCURRENT, &mdev->flags)
2146 if (ch_peer == 0) { rv = 1; break; }
2147 if (ch_self == 0) { rv = -1; break; }
2149 if (mdev->net_conf->after_sb_0p == ASB_DISCARD_ZERO_CHG)
2151 case ASB_DISCARD_LEAST_CHG:
2152 if (ch_self < ch_peer)
2154 else if (ch_self > ch_peer)
2156 else /* ( ch_self == ch_peer ) */
2157 /* Well, then use something else. */
2158 rv = test_bit(DISCARD_CONCURRENT, &mdev->flags)
2161 case ASB_DISCARD_LOCAL:
2164 case ASB_DISCARD_REMOTE:
2171 static int drbd_asb_recover_1p(struct drbd_conf *mdev) __must_hold(local)
2173 int self, peer, hg, rv = -100;
2175 self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2176 peer = mdev->p_uuid[UI_BITMAP] & 1;
2178 switch (mdev->net_conf->after_sb_1p) {
2179 case ASB_DISCARD_YOUNGER_PRI:
2180 case ASB_DISCARD_OLDER_PRI:
2181 case ASB_DISCARD_LEAST_CHG:
2182 case ASB_DISCARD_LOCAL:
2183 case ASB_DISCARD_REMOTE:
2184 dev_err(DEV, "Configuration error.\n");
2186 case ASB_DISCONNECT:
2189 hg = drbd_asb_recover_0p(mdev);
2190 if (hg == -1 && mdev->state.role == R_SECONDARY)
2192 if (hg == 1 && mdev->state.role == R_PRIMARY)
2196 rv = drbd_asb_recover_0p(mdev);
2198 case ASB_DISCARD_SECONDARY:
2199 return mdev->state.role == R_PRIMARY ? 1 : -1;
2200 case ASB_CALL_HELPER:
2201 hg = drbd_asb_recover_0p(mdev);
2202 if (hg == -1 && mdev->state.role == R_PRIMARY) {
2203 self = drbd_set_role(mdev, R_SECONDARY, 0);
2204 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2205 * we might be here in C_WF_REPORT_PARAMS which is transient.
2206 * we do not need to wait for the after state change work either. */
2207 self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY));
2208 if (self != SS_SUCCESS) {
2209 drbd_khelper(mdev, "pri-lost-after-sb");
2211 dev_warn(DEV, "Successfully gave up primary role.\n");
2221 static int drbd_asb_recover_2p(struct drbd_conf *mdev) __must_hold(local)
2223 int self, peer, hg, rv = -100;
2225 self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2226 peer = mdev->p_uuid[UI_BITMAP] & 1;
2228 switch (mdev->net_conf->after_sb_2p) {
2229 case ASB_DISCARD_YOUNGER_PRI:
2230 case ASB_DISCARD_OLDER_PRI:
2231 case ASB_DISCARD_LEAST_CHG:
2232 case ASB_DISCARD_LOCAL:
2233 case ASB_DISCARD_REMOTE:
2235 case ASB_DISCARD_SECONDARY:
2236 dev_err(DEV, "Configuration error.\n");
2239 rv = drbd_asb_recover_0p(mdev);
2241 case ASB_DISCONNECT:
2243 case ASB_CALL_HELPER:
2244 hg = drbd_asb_recover_0p(mdev);
2246 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2247 * we might be here in C_WF_REPORT_PARAMS which is transient.
2248 * we do not need to wait for the after state change work either. */
2249 self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY));
2250 if (self != SS_SUCCESS) {
2251 drbd_khelper(mdev, "pri-lost-after-sb");
2253 dev_warn(DEV, "Successfully gave up primary role.\n");
2263 static void drbd_uuid_dump(struct drbd_conf *mdev, char *text, u64 *uuid,
2264 u64 bits, u64 flags)
2267 dev_info(DEV, "%s uuid info vanished while I was looking!\n", text);
2270 dev_info(DEV, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2272 (unsigned long long)uuid[UI_CURRENT],
2273 (unsigned long long)uuid[UI_BITMAP],
2274 (unsigned long long)uuid[UI_HISTORY_START],
2275 (unsigned long long)uuid[UI_HISTORY_END],
2276 (unsigned long long)bits,
2277 (unsigned long long)flags);
2281 100 after split brain try auto recover
2282 2 C_SYNC_SOURCE set BitMap
2283 1 C_SYNC_SOURCE use BitMap
2285 -1 C_SYNC_TARGET use BitMap
2286 -2 C_SYNC_TARGET set BitMap
2287 -100 after split brain, disconnect
2288 -1000 unrelated data
2290 static int drbd_uuid_compare(struct drbd_conf *mdev, int *rule_nr) __must_hold(local)
2295 self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2296 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2299 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
2303 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
2304 peer != UUID_JUST_CREATED)
2308 if (self != UUID_JUST_CREATED &&
2309 (peer == UUID_JUST_CREATED || peer == (u64)0))
2313 int rct, dc; /* roles at crash time */
2315 if (mdev->p_uuid[UI_BITMAP] == (u64)0 && mdev->ldev->md.uuid[UI_BITMAP] != (u64)0) {
2317 if (mdev->agreed_pro_version < 91)
2320 if ((mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
2321 (mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
2322 dev_info(DEV, "was SyncSource, missed the resync finished event, corrected myself:\n");
2323 drbd_uuid_set_bm(mdev, 0UL);
2325 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid,
2326 mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0);
2329 dev_info(DEV, "was SyncSource (peer failed to write sync_uuid)\n");
2336 if (mdev->ldev->md.uuid[UI_BITMAP] == (u64)0 && mdev->p_uuid[UI_BITMAP] != (u64)0) {
2338 if (mdev->agreed_pro_version < 91)
2341 if ((mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_BITMAP] & ~((u64)1)) &&
2342 (mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
2343 dev_info(DEV, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2345 mdev->p_uuid[UI_HISTORY_START + 1] = mdev->p_uuid[UI_HISTORY_START];
2346 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_BITMAP];
2347 mdev->p_uuid[UI_BITMAP] = 0UL;
2349 drbd_uuid_dump(mdev, "peer", mdev->p_uuid, mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2352 dev_info(DEV, "was SyncTarget (failed to write sync_uuid)\n");
2359 /* Common power [off|failure] */
2360 rct = (test_bit(CRASHED_PRIMARY, &mdev->flags) ? 1 : 0) +
2361 (mdev->p_uuid[UI_FLAGS] & 2);
2362 /* lowest bit is set when we were primary,
2363 * next bit (weight 2) is set when peer was primary */
2367 case 0: /* !self_pri && !peer_pri */ return 0;
2368 case 1: /* self_pri && !peer_pri */ return 1;
2369 case 2: /* !self_pri && peer_pri */ return -1;
2370 case 3: /* self_pri && peer_pri */
2371 dc = test_bit(DISCARD_CONCURRENT, &mdev->flags);
2377 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2382 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1);
2384 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2385 peer = mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1);
2387 /* The last P_SYNC_UUID did not get though. Undo the last start of
2388 resync as sync source modifications of the peer's UUIDs. */
2390 if (mdev->agreed_pro_version < 91)
2393 mdev->p_uuid[UI_BITMAP] = mdev->p_uuid[UI_HISTORY_START];
2394 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_HISTORY_START + 1];
2400 self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2401 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2402 peer = mdev->p_uuid[i] & ~((u64)1);
2408 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2409 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2414 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2416 self = mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1);
2417 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1);
2419 /* The last P_SYNC_UUID did not get though. Undo the last start of
2420 resync as sync source modifications of our UUIDs. */
2422 if (mdev->agreed_pro_version < 91)
2425 _drbd_uuid_set(mdev, UI_BITMAP, mdev->ldev->md.uuid[UI_HISTORY_START]);
2426 _drbd_uuid_set(mdev, UI_HISTORY_START, mdev->ldev->md.uuid[UI_HISTORY_START + 1]);
2428 dev_info(DEV, "Undid last start of resync:\n");
2430 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid,
2431 mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0);
2439 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2440 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2441 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2447 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2448 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2449 if (self == peer && self != ((u64)0))
2453 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2454 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2455 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
2456 peer = mdev->p_uuid[j] & ~((u64)1);
2465 /* drbd_sync_handshake() returns the new conn state on success, or
2466 CONN_MASK (-1) on failure.
2468 static enum drbd_conns drbd_sync_handshake(struct drbd_conf *mdev, enum drbd_role peer_role,
2469 enum drbd_disk_state peer_disk) __must_hold(local)
2472 enum drbd_conns rv = C_MASK;
2473 enum drbd_disk_state mydisk;
2475 mydisk = mdev->state.disk;
2476 if (mydisk == D_NEGOTIATING)
2477 mydisk = mdev->new_state_tmp.disk;
2479 dev_info(DEV, "drbd_sync_handshake:\n");
2480 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, mdev->comm_bm_set, 0);
2481 drbd_uuid_dump(mdev, "peer", mdev->p_uuid,
2482 mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2484 hg = drbd_uuid_compare(mdev, &rule_nr);
2486 dev_info(DEV, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
2489 dev_alert(DEV, "Unrelated data, aborting!\n");
2493 dev_alert(DEV, "To resolve this both sides have to support at least protocol\n");
2497 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
2498 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
2499 int f = (hg == -100) || abs(hg) == 2;
2500 hg = mydisk > D_INCONSISTENT ? 1 : -1;
2503 dev_info(DEV, "Becoming sync %s due to disk states.\n",
2504 hg > 0 ? "source" : "target");
2508 drbd_khelper(mdev, "initial-split-brain");
2510 if (hg == 100 || (hg == -100 && mdev->net_conf->always_asbp)) {
2511 int pcount = (mdev->state.role == R_PRIMARY)
2512 + (peer_role == R_PRIMARY);
2513 int forced = (hg == -100);
2517 hg = drbd_asb_recover_0p(mdev);
2520 hg = drbd_asb_recover_1p(mdev);
2523 hg = drbd_asb_recover_2p(mdev);
2526 if (abs(hg) < 100) {
2527 dev_warn(DEV, "Split-Brain detected, %d primaries, "
2528 "automatically solved. Sync from %s node\n",
2529 pcount, (hg < 0) ? "peer" : "this");
2531 dev_warn(DEV, "Doing a full sync, since"
2532 " UUIDs where ambiguous.\n");
2539 if (mdev->net_conf->want_lose && !(mdev->p_uuid[UI_FLAGS]&1))
2541 if (!mdev->net_conf->want_lose && (mdev->p_uuid[UI_FLAGS]&1))
2545 dev_warn(DEV, "Split-Brain detected, manually solved. "
2546 "Sync from %s node\n",
2547 (hg < 0) ? "peer" : "this");
2551 /* FIXME this log message is not correct if we end up here
2552 * after an attempted attach on a diskless node.
2553 * We just refuse to attach -- well, we drop the "connection"
2554 * to that disk, in a way... */
2555 dev_alert(DEV, "Split-Brain detected but unresolved, dropping connection!\n");
2556 drbd_khelper(mdev, "split-brain");
2560 if (hg > 0 && mydisk <= D_INCONSISTENT) {
2561 dev_err(DEV, "I shall become SyncSource, but I am inconsistent!\n");
2565 if (hg < 0 && /* by intention we do not use mydisk here. */
2566 mdev->state.role == R_PRIMARY && mdev->state.disk >= D_CONSISTENT) {
2567 switch (mdev->net_conf->rr_conflict) {
2568 case ASB_CALL_HELPER:
2569 drbd_khelper(mdev, "pri-lost");
2571 case ASB_DISCONNECT:
2572 dev_err(DEV, "I shall become SyncTarget, but I am primary!\n");
2575 dev_warn(DEV, "Becoming SyncTarget, violating the stable-data"
2580 if (mdev->net_conf->dry_run || test_bit(CONN_DRY_RUN, &mdev->flags)) {
2582 dev_info(DEV, "dry-run connect: No resync, would become Connected immediately.\n");
2584 dev_info(DEV, "dry-run connect: Would become %s, doing a %s resync.",
2585 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
2586 abs(hg) >= 2 ? "full" : "bit-map based");
2591 dev_info(DEV, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2592 if (drbd_bitmap_io(mdev, &drbd_bmio_set_n_write, "set_n_write from sync_handshake"))
2596 if (hg > 0) { /* become sync source. */
2598 } else if (hg < 0) { /* become sync target */
2602 if (drbd_bm_total_weight(mdev)) {
2603 dev_info(DEV, "No resync, but %lu bits in bitmap!\n",
2604 drbd_bm_total_weight(mdev));
2611 /* returns 1 if invalid */
2612 static int cmp_after_sb(enum drbd_after_sb_p peer, enum drbd_after_sb_p self)
2614 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2615 if ((peer == ASB_DISCARD_REMOTE && self == ASB_DISCARD_LOCAL) ||
2616 (self == ASB_DISCARD_REMOTE && peer == ASB_DISCARD_LOCAL))
2619 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2620 if (peer == ASB_DISCARD_REMOTE || peer == ASB_DISCARD_LOCAL ||
2621 self == ASB_DISCARD_REMOTE || self == ASB_DISCARD_LOCAL)
2624 /* everything else is valid if they are equal on both sides. */
2628 /* everything es is invalid. */
2632 static int receive_protocol(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
2634 struct p_protocol *p = &mdev->data.rbuf.protocol;
2635 int p_proto, p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
2636 int p_want_lose, p_two_primaries, cf;
2637 char p_integrity_alg[SHARED_SECRET_MAX] = "";
2639 p_proto = be32_to_cpu(p->protocol);
2640 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
2641 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
2642 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
2643 p_two_primaries = be32_to_cpu(p->two_primaries);
2644 cf = be32_to_cpu(p->conn_flags);
2645 p_want_lose = cf & CF_WANT_LOSE;
2647 clear_bit(CONN_DRY_RUN, &mdev->flags);
2649 if (cf & CF_DRY_RUN)
2650 set_bit(CONN_DRY_RUN, &mdev->flags);
2652 if (p_proto != mdev->net_conf->wire_protocol) {
2653 dev_err(DEV, "incompatible communication protocols\n");
2657 if (cmp_after_sb(p_after_sb_0p, mdev->net_conf->after_sb_0p)) {
2658 dev_err(DEV, "incompatible after-sb-0pri settings\n");
2662 if (cmp_after_sb(p_after_sb_1p, mdev->net_conf->after_sb_1p)) {
2663 dev_err(DEV, "incompatible after-sb-1pri settings\n");
2667 if (cmp_after_sb(p_after_sb_2p, mdev->net_conf->after_sb_2p)) {
2668 dev_err(DEV, "incompatible after-sb-2pri settings\n");
2672 if (p_want_lose && mdev->net_conf->want_lose) {
2673 dev_err(DEV, "both sides have the 'want_lose' flag set\n");
2677 if (p_two_primaries != mdev->net_conf->two_primaries) {
2678 dev_err(DEV, "incompatible setting of the two-primaries options\n");
2682 if (mdev->agreed_pro_version >= 87) {
2683 unsigned char *my_alg = mdev->net_conf->integrity_alg;
2685 if (drbd_recv(mdev, p_integrity_alg, data_size) != data_size)
2688 p_integrity_alg[SHARED_SECRET_MAX-1] = 0;
2689 if (strcmp(p_integrity_alg, my_alg)) {
2690 dev_err(DEV, "incompatible setting of the data-integrity-alg\n");
2693 dev_info(DEV, "data-integrity-alg: %s\n",
2694 my_alg[0] ? my_alg : (unsigned char *)"<not-used>");
2700 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2705 * input: alg name, feature name
2706 * return: NULL (alg name was "")
2707 * ERR_PTR(error) if something goes wrong
2708 * or the crypto hash ptr, if it worked out ok. */
2709 struct crypto_hash *drbd_crypto_alloc_digest_safe(const struct drbd_conf *mdev,
2710 const char *alg, const char *name)
2712 struct crypto_hash *tfm;
2717 tfm = crypto_alloc_hash(alg, 0, CRYPTO_ALG_ASYNC);
2719 dev_err(DEV, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2720 alg, name, PTR_ERR(tfm));
2723 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm))) {
2724 crypto_free_hash(tfm);
2725 dev_err(DEV, "\"%s\" is not a digest (%s)\n", alg, name);
2726 return ERR_PTR(-EINVAL);
2731 static int receive_SyncParam(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int packet_size)
2734 struct p_rs_param_95 *p = &mdev->data.rbuf.rs_param_95;
2735 unsigned int header_size, data_size, exp_max_sz;
2736 struct crypto_hash *verify_tfm = NULL;
2737 struct crypto_hash *csums_tfm = NULL;
2738 const int apv = mdev->agreed_pro_version;
2739 int *rs_plan_s = NULL;
2742 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
2743 : apv == 88 ? sizeof(struct p_rs_param)
2745 : apv <= 94 ? sizeof(struct p_rs_param_89)
2746 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
2748 if (packet_size > exp_max_sz) {
2749 dev_err(DEV, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2750 packet_size, exp_max_sz);
2755 header_size = sizeof(struct p_rs_param) - sizeof(struct p_header80);
2756 data_size = packet_size - header_size;
2757 } else if (apv <= 94) {
2758 header_size = sizeof(struct p_rs_param_89) - sizeof(struct p_header80);
2759 data_size = packet_size - header_size;
2760 D_ASSERT(data_size == 0);
2762 header_size = sizeof(struct p_rs_param_95) - sizeof(struct p_header80);
2763 data_size = packet_size - header_size;
2764 D_ASSERT(data_size == 0);
2767 /* initialize verify_alg and csums_alg */
2768 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
2770 if (drbd_recv(mdev, &p->head.payload, header_size) != header_size)
2773 mdev->sync_conf.rate = be32_to_cpu(p->rate);
2777 if (data_size > SHARED_SECRET_MAX) {
2778 dev_err(DEV, "verify-alg too long, "
2779 "peer wants %u, accepting only %u byte\n",
2780 data_size, SHARED_SECRET_MAX);
2784 if (drbd_recv(mdev, p->verify_alg, data_size) != data_size)
2787 /* we expect NUL terminated string */
2788 /* but just in case someone tries to be evil */
2789 D_ASSERT(p->verify_alg[data_size-1] == 0);
2790 p->verify_alg[data_size-1] = 0;
2792 } else /* apv >= 89 */ {
2793 /* we still expect NUL terminated strings */
2794 /* but just in case someone tries to be evil */
2795 D_ASSERT(p->verify_alg[SHARED_SECRET_MAX-1] == 0);
2796 D_ASSERT(p->csums_alg[SHARED_SECRET_MAX-1] == 0);
2797 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
2798 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
2801 if (strcmp(mdev->sync_conf.verify_alg, p->verify_alg)) {
2802 if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2803 dev_err(DEV, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2804 mdev->sync_conf.verify_alg, p->verify_alg);
2807 verify_tfm = drbd_crypto_alloc_digest_safe(mdev,
2808 p->verify_alg, "verify-alg");
2809 if (IS_ERR(verify_tfm)) {
2815 if (apv >= 89 && strcmp(mdev->sync_conf.csums_alg, p->csums_alg)) {
2816 if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2817 dev_err(DEV, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2818 mdev->sync_conf.csums_alg, p->csums_alg);
2821 csums_tfm = drbd_crypto_alloc_digest_safe(mdev,
2822 p->csums_alg, "csums-alg");
2823 if (IS_ERR(csums_tfm)) {
2830 mdev->sync_conf.rate = be32_to_cpu(p->rate);
2831 mdev->sync_conf.c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
2832 mdev->sync_conf.c_delay_target = be32_to_cpu(p->c_delay_target);
2833 mdev->sync_conf.c_fill_target = be32_to_cpu(p->c_fill_target);
2834 mdev->sync_conf.c_max_rate = be32_to_cpu(p->c_max_rate);
2836 fifo_size = (mdev->sync_conf.c_plan_ahead * 10 * SLEEP_TIME) / HZ;
2837 if (fifo_size != mdev->rs_plan_s.size && fifo_size > 0) {
2838 rs_plan_s = kzalloc(sizeof(int) * fifo_size, GFP_KERNEL);
2840 dev_err(DEV, "kmalloc of fifo_buffer failed");
2846 spin_lock(&mdev->peer_seq_lock);
2847 /* lock against drbd_nl_syncer_conf() */
2849 strcpy(mdev->sync_conf.verify_alg, p->verify_alg);
2850 mdev->sync_conf.verify_alg_len = strlen(p->verify_alg) + 1;
2851 crypto_free_hash(mdev->verify_tfm);
2852 mdev->verify_tfm = verify_tfm;
2853 dev_info(DEV, "using verify-alg: \"%s\"\n", p->verify_alg);
2856 strcpy(mdev->sync_conf.csums_alg, p->csums_alg);
2857 mdev->sync_conf.csums_alg_len = strlen(p->csums_alg) + 1;
2858 crypto_free_hash(mdev->csums_tfm);
2859 mdev->csums_tfm = csums_tfm;
2860 dev_info(DEV, "using csums-alg: \"%s\"\n", p->csums_alg);
2862 if (fifo_size != mdev->rs_plan_s.size) {
2863 kfree(mdev->rs_plan_s.values);
2864 mdev->rs_plan_s.values = rs_plan_s;
2865 mdev->rs_plan_s.size = fifo_size;
2866 mdev->rs_planed = 0;
2868 spin_unlock(&mdev->peer_seq_lock);
2873 /* just for completeness: actually not needed,
2874 * as this is not reached if csums_tfm was ok. */
2875 crypto_free_hash(csums_tfm);
2876 /* but free the verify_tfm again, if csums_tfm did not work out */
2877 crypto_free_hash(verify_tfm);
2878 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2882 static void drbd_setup_order_type(struct drbd_conf *mdev, int peer)
2884 /* sorry, we currently have no working implementation
2885 * of distributed TCQ */
2888 /* warn if the arguments differ by more than 12.5% */
2889 static void warn_if_differ_considerably(struct drbd_conf *mdev,
2890 const char *s, sector_t a, sector_t b)
2893 if (a == 0 || b == 0)
2895 d = (a > b) ? (a - b) : (b - a);
2896 if (d > (a>>3) || d > (b>>3))
2897 dev_warn(DEV, "Considerable difference in %s: %llus vs. %llus\n", s,
2898 (unsigned long long)a, (unsigned long long)b);
2901 static int receive_sizes(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
2903 struct p_sizes *p = &mdev->data.rbuf.sizes;
2904 enum determine_dev_size dd = unchanged;
2905 unsigned int max_seg_s;
2906 sector_t p_size, p_usize, my_usize;
2907 int ldsc = 0; /* local disk size changed */
2908 enum dds_flags ddsf;
2910 p_size = be64_to_cpu(p->d_size);
2911 p_usize = be64_to_cpu(p->u_size);
2913 if (p_size == 0 && mdev->state.disk == D_DISKLESS) {
2914 dev_err(DEV, "some backing storage is needed\n");
2915 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2919 /* just store the peer's disk size for now.
2920 * we still need to figure out whether we accept that. */
2921 mdev->p_size = p_size;
2923 if (get_ldev(mdev)) {
2924 warn_if_differ_considerably(mdev, "lower level device sizes",
2925 p_size, drbd_get_max_capacity(mdev->ldev));
2926 warn_if_differ_considerably(mdev, "user requested size",
2927 p_usize, mdev->ldev->dc.disk_size);
2929 /* if this is the first connect, or an otherwise expected
2930 * param exchange, choose the minimum */
2931 if (mdev->state.conn == C_WF_REPORT_PARAMS)
2932 p_usize = min_not_zero((sector_t)mdev->ldev->dc.disk_size,
2935 my_usize = mdev->ldev->dc.disk_size;
2937 if (mdev->ldev->dc.disk_size != p_usize) {
2938 mdev->ldev->dc.disk_size = p_usize;
2939 dev_info(DEV, "Peer sets u_size to %lu sectors\n",
2940 (unsigned long)mdev->ldev->dc.disk_size);
2943 /* Never shrink a device with usable data during connect.
2944 But allow online shrinking if we are connected. */
2945 if (drbd_new_dev_size(mdev, mdev->ldev, 0) <
2946 drbd_get_capacity(mdev->this_bdev) &&
2947 mdev->state.disk >= D_OUTDATED &&
2948 mdev->state.conn < C_CONNECTED) {
2949 dev_err(DEV, "The peer's disk size is too small!\n");
2950 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2951 mdev->ldev->dc.disk_size = my_usize;
2959 ddsf = be16_to_cpu(p->dds_flags);
2960 if (get_ldev(mdev)) {
2961 dd = drbd_determin_dev_size(mdev, ddsf);
2963 if (dd == dev_size_error)
2967 /* I am diskless, need to accept the peer's size. */
2968 drbd_set_my_capacity(mdev, p_size);
2971 if (get_ldev(mdev)) {
2972 if (mdev->ldev->known_size != drbd_get_capacity(mdev->ldev->backing_bdev)) {
2973 mdev->ldev->known_size = drbd_get_capacity(mdev->ldev->backing_bdev);
2977 if (mdev->agreed_pro_version < 94)
2978 max_seg_s = be32_to_cpu(p->max_segment_size);
2979 else if (mdev->agreed_pro_version == 94)
2980 max_seg_s = DRBD_MAX_SIZE_H80_PACKET;
2981 else /* drbd 8.3.8 onwards */
2982 max_seg_s = DRBD_MAX_SEGMENT_SIZE;
2984 if (max_seg_s != queue_max_segment_size(mdev->rq_queue))
2985 drbd_setup_queue_param(mdev, max_seg_s);
2987 drbd_setup_order_type(mdev, be16_to_cpu(p->queue_order_type));
2991 if (mdev->state.conn > C_WF_REPORT_PARAMS) {
2992 if (be64_to_cpu(p->c_size) !=
2993 drbd_get_capacity(mdev->this_bdev) || ldsc) {
2994 /* we have different sizes, probably peer
2995 * needs to know my new size... */
2996 drbd_send_sizes(mdev, 0, ddsf);
2998 if (test_and_clear_bit(RESIZE_PENDING, &mdev->flags) ||
2999 (dd == grew && mdev->state.conn == C_CONNECTED)) {
3000 if (mdev->state.pdsk >= D_INCONSISTENT &&
3001 mdev->state.disk >= D_INCONSISTENT) {
3002 if (ddsf & DDSF_NO_RESYNC)
3003 dev_info(DEV, "Resync of new storage suppressed with --assume-clean\n");
3005 resync_after_online_grow(mdev);
3007 set_bit(RESYNC_AFTER_NEG, &mdev->flags);
3014 static int receive_uuids(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3016 struct p_uuids *p = &mdev->data.rbuf.uuids;
3020 p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
3022 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
3023 p_uuid[i] = be64_to_cpu(p->uuid[i]);
3025 kfree(mdev->p_uuid);
3026 mdev->p_uuid = p_uuid;
3028 if (mdev->state.conn < C_CONNECTED &&
3029 mdev->state.disk < D_INCONSISTENT &&
3030 mdev->state.role == R_PRIMARY &&
3031 (mdev->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
3032 dev_err(DEV, "Can only connect to data with current UUID=%016llX\n",
3033 (unsigned long long)mdev->ed_uuid);
3034 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3038 if (get_ldev(mdev)) {
3039 int skip_initial_sync =
3040 mdev->state.conn == C_CONNECTED &&
3041 mdev->agreed_pro_version >= 90 &&
3042 mdev->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
3043 (p_uuid[UI_FLAGS] & 8);
3044 if (skip_initial_sync) {
3045 dev_info(DEV, "Accepted new current UUID, preparing to skip initial sync\n");
3046 drbd_bitmap_io(mdev, &drbd_bmio_clear_n_write,
3047 "clear_n_write from receive_uuids");
3048 _drbd_uuid_set(mdev, UI_CURRENT, p_uuid[UI_CURRENT]);
3049 _drbd_uuid_set(mdev, UI_BITMAP, 0);
3050 _drbd_set_state(_NS2(mdev, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
3055 } else if (mdev->state.disk < D_INCONSISTENT &&
3056 mdev->state.role == R_PRIMARY) {
3057 /* I am a diskless primary, the peer just created a new current UUID
3059 drbd_set_ed_uuid(mdev, p_uuid[UI_CURRENT]);
3062 /* Before we test for the disk state, we should wait until an eventually
3063 ongoing cluster wide state change is finished. That is important if
3064 we are primary and are detaching from our disk. We need to see the
3065 new disk state... */
3066 wait_event(mdev->misc_wait, !test_bit(CLUSTER_ST_CHANGE, &mdev->flags));
3067 if (mdev->state.conn >= C_CONNECTED && mdev->state.disk < D_INCONSISTENT)
3068 drbd_set_ed_uuid(mdev, p_uuid[UI_CURRENT]);
3074 * convert_state() - Converts the peer's view of the cluster state to our point of view
3075 * @ps: The state as seen by the peer.
3077 static union drbd_state convert_state(union drbd_state ps)
3079 union drbd_state ms;
3081 static enum drbd_conns c_tab[] = {
3082 [C_CONNECTED] = C_CONNECTED,
3084 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
3085 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
3086 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
3087 [C_VERIFY_S] = C_VERIFY_T,
3093 ms.conn = c_tab[ps.conn];
3098 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
3103 static int receive_req_state(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3105 struct p_req_state *p = &mdev->data.rbuf.req_state;
3106 union drbd_state mask, val;
3109 mask.i = be32_to_cpu(p->mask);
3110 val.i = be32_to_cpu(p->val);
3112 if (test_bit(DISCARD_CONCURRENT, &mdev->flags) &&
3113 test_bit(CLUSTER_ST_CHANGE, &mdev->flags)) {
3114 drbd_send_sr_reply(mdev, SS_CONCURRENT_ST_CHG);
3118 mask = convert_state(mask);
3119 val = convert_state(val);
3121 rv = drbd_change_state(mdev, CS_VERBOSE, mask, val);
3123 drbd_send_sr_reply(mdev, rv);
3129 static int receive_state(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3131 struct p_state *p = &mdev->data.rbuf.state;
3132 union drbd_state os, ns, peer_state;
3133 enum drbd_disk_state real_peer_disk;
3134 enum chg_state_flags cs_flags;
3137 peer_state.i = be32_to_cpu(p->state);
3139 real_peer_disk = peer_state.disk;
3140 if (peer_state.disk == D_NEGOTIATING) {
3141 real_peer_disk = mdev->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
3142 dev_info(DEV, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
3145 spin_lock_irq(&mdev->req_lock);
3147 os = ns = mdev->state;
3148 spin_unlock_irq(&mdev->req_lock);
3150 /* peer says his disk is uptodate, while we think it is inconsistent,
3151 * and this happens while we think we have a sync going on. */
3152 if (os.pdsk == D_INCONSISTENT && real_peer_disk == D_UP_TO_DATE &&
3153 os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
3154 /* If we are (becoming) SyncSource, but peer is still in sync
3155 * preparation, ignore its uptodate-ness to avoid flapping, it
3156 * will change to inconsistent once the peer reaches active
3158 * It may have changed syncer-paused flags, however, so we
3159 * cannot ignore this completely. */
3160 if (peer_state.conn > C_CONNECTED &&
3161 peer_state.conn < C_SYNC_SOURCE)
3162 real_peer_disk = D_INCONSISTENT;
3164 /* if peer_state changes to connected at the same time,
3165 * it explicitly notifies us that it finished resync.
3166 * Maybe we should finish it up, too? */
3167 else if (os.conn >= C_SYNC_SOURCE &&
3168 peer_state.conn == C_CONNECTED) {
3169 if (drbd_bm_total_weight(mdev) <= mdev->rs_failed)
3170 drbd_resync_finished(mdev);
3175 /* peer says his disk is inconsistent, while we think it is uptodate,
3176 * and this happens while the peer still thinks we have a sync going on,
3177 * but we think we are already done with the sync.
3178 * We ignore this to avoid flapping pdsk.
3179 * This should not happen, if the peer is a recent version of drbd. */
3180 if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
3181 os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
3182 real_peer_disk = D_UP_TO_DATE;
3184 if (ns.conn == C_WF_REPORT_PARAMS)
3185 ns.conn = C_CONNECTED;
3187 if (mdev->p_uuid && peer_state.disk >= D_NEGOTIATING &&
3188 get_ldev_if_state(mdev, D_NEGOTIATING)) {
3189 int cr; /* consider resync */
3191 /* if we established a new connection */
3192 cr = (os.conn < C_CONNECTED);
3193 /* if we had an established connection
3194 * and one of the nodes newly attaches a disk */
3195 cr |= (os.conn == C_CONNECTED &&
3196 (peer_state.disk == D_NEGOTIATING ||
3197 os.disk == D_NEGOTIATING));
3198 /* if we have both been inconsistent, and the peer has been
3199 * forced to be UpToDate with --overwrite-data */
3200 cr |= test_bit(CONSIDER_RESYNC, &mdev->flags);
3201 /* if we had been plain connected, and the admin requested to
3202 * start a sync by "invalidate" or "invalidate-remote" */
3203 cr |= (os.conn == C_CONNECTED &&
3204 (peer_state.conn >= C_STARTING_SYNC_S &&
3205 peer_state.conn <= C_WF_BITMAP_T));
3208 ns.conn = drbd_sync_handshake(mdev, peer_state.role, real_peer_disk);
3211 if (ns.conn == C_MASK) {
3212 ns.conn = C_CONNECTED;
3213 if (mdev->state.disk == D_NEGOTIATING) {
3214 drbd_force_state(mdev, NS(disk, D_FAILED));
3215 } else if (peer_state.disk == D_NEGOTIATING) {
3216 dev_err(DEV, "Disk attach process on the peer node was aborted.\n");
3217 peer_state.disk = D_DISKLESS;
3218 real_peer_disk = D_DISKLESS;
3220 if (test_and_clear_bit(CONN_DRY_RUN, &mdev->flags))
3222 D_ASSERT(os.conn == C_WF_REPORT_PARAMS);
3223 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3229 spin_lock_irq(&mdev->req_lock);
3230 if (mdev->state.i != os.i)
3232 clear_bit(CONSIDER_RESYNC, &mdev->flags);
3233 ns.peer = peer_state.role;
3234 ns.pdsk = real_peer_disk;
3235 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
3236 if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
3237 ns.disk = mdev->new_state_tmp.disk;
3238 cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
3239 if (ns.pdsk == D_CONSISTENT && is_susp(ns) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
3240 test_bit(NEW_CUR_UUID, &mdev->flags)) {
3241 /* Do not allow tl_restart(resend) for a rebooted peer. We can only allow this
3242 for temporal network outages! */
3243 spin_unlock_irq(&mdev->req_lock);
3244 dev_err(DEV, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3246 drbd_uuid_new_current(mdev);
3247 clear_bit(NEW_CUR_UUID, &mdev->flags);
3248 drbd_force_state(mdev, NS2(conn, C_PROTOCOL_ERROR, susp, 0));
3251 rv = _drbd_set_state(mdev, ns, cs_flags, NULL);
3253 spin_unlock_irq(&mdev->req_lock);
3255 if (rv < SS_SUCCESS) {
3256 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3260 if (os.conn > C_WF_REPORT_PARAMS) {
3261 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
3262 peer_state.disk != D_NEGOTIATING ) {
3263 /* we want resync, peer has not yet decided to sync... */
3264 /* Nowadays only used when forcing a node into primary role and
3265 setting its disk to UpToDate with that */
3266 drbd_send_uuids(mdev);
3267 drbd_send_state(mdev);
3271 mdev->net_conf->want_lose = 0;
3273 drbd_md_sync(mdev); /* update connected indicator, la_size, ... */
3278 static int receive_sync_uuid(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3280 struct p_rs_uuid *p = &mdev->data.rbuf.rs_uuid;
3282 wait_event(mdev->misc_wait,
3283 mdev->state.conn == C_WF_SYNC_UUID ||
3284 mdev->state.conn < C_CONNECTED ||
3285 mdev->state.disk < D_NEGOTIATING);
3287 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3289 /* Here the _drbd_uuid_ functions are right, current should
3290 _not_ be rotated into the history */
3291 if (get_ldev_if_state(mdev, D_NEGOTIATING)) {
3292 _drbd_uuid_set(mdev, UI_CURRENT, be64_to_cpu(p->uuid));
3293 _drbd_uuid_set(mdev, UI_BITMAP, 0UL);
3295 drbd_start_resync(mdev, C_SYNC_TARGET);
3299 dev_err(DEV, "Ignoring SyncUUID packet!\n");
3304 enum receive_bitmap_ret { OK, DONE, FAILED };
3306 static enum receive_bitmap_ret
3307 receive_bitmap_plain(struct drbd_conf *mdev, unsigned int data_size,
3308 unsigned long *buffer, struct bm_xfer_ctx *c)
3310 unsigned num_words = min_t(size_t, BM_PACKET_WORDS, c->bm_words - c->word_offset);
3311 unsigned want = num_words * sizeof(long);
3313 if (want != data_size) {
3314 dev_err(DEV, "%s:want (%u) != data_size (%u)\n", __func__, want, data_size);
3319 if (drbd_recv(mdev, buffer, want) != want)
3322 drbd_bm_merge_lel(mdev, c->word_offset, num_words, buffer);
3324 c->word_offset += num_words;
3325 c->bit_offset = c->word_offset * BITS_PER_LONG;
3326 if (c->bit_offset > c->bm_bits)
3327 c->bit_offset = c->bm_bits;
3332 static enum receive_bitmap_ret
3333 recv_bm_rle_bits(struct drbd_conf *mdev,
3334 struct p_compressed_bm *p,
3335 struct bm_xfer_ctx *c)
3337 struct bitstream bs;
3341 unsigned long s = c->bit_offset;
3343 int len = be16_to_cpu(p->head.length) - (sizeof(*p) - sizeof(p->head));
3344 int toggle = DCBP_get_start(p);
3348 bitstream_init(&bs, p->code, len, DCBP_get_pad_bits(p));
3350 bits = bitstream_get_bits(&bs, &look_ahead, 64);
3354 for (have = bits; have > 0; s += rl, toggle = !toggle) {
3355 bits = vli_decode_bits(&rl, look_ahead);
3361 if (e >= c->bm_bits) {
3362 dev_err(DEV, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
3365 _drbd_bm_set_bits(mdev, s, e);
3369 dev_err(DEV, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3370 have, bits, look_ahead,
3371 (unsigned int)(bs.cur.b - p->code),
3372 (unsigned int)bs.buf_len);
3375 look_ahead >>= bits;
3378 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
3381 look_ahead |= tmp << have;
3386 bm_xfer_ctx_bit_to_word_offset(c);
3388 return (s == c->bm_bits) ? DONE : OK;
3391 static enum receive_bitmap_ret
3392 decode_bitmap_c(struct drbd_conf *mdev,
3393 struct p_compressed_bm *p,
3394 struct bm_xfer_ctx *c)
3396 if (DCBP_get_code(p) == RLE_VLI_Bits)
3397 return recv_bm_rle_bits(mdev, p, c);
3399 /* other variants had been implemented for evaluation,
3400 * but have been dropped as this one turned out to be "best"
3401 * during all our tests. */
3403 dev_err(DEV, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
3404 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3408 void INFO_bm_xfer_stats(struct drbd_conf *mdev,
3409 const char *direction, struct bm_xfer_ctx *c)
3411 /* what would it take to transfer it "plaintext" */
3412 unsigned plain = sizeof(struct p_header80) *
3413 ((c->bm_words+BM_PACKET_WORDS-1)/BM_PACKET_WORDS+1)
3414 + c->bm_words * sizeof(long);
3415 unsigned total = c->bytes[0] + c->bytes[1];
3418 /* total can not be zero. but just in case: */
3422 /* don't report if not compressed */
3426 /* total < plain. check for overflow, still */
3427 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
3428 : (1000 * total / plain);
3434 dev_info(DEV, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3435 "total %u; compression: %u.%u%%\n",
3437 c->bytes[1], c->packets[1],
3438 c->bytes[0], c->packets[0],
3439 total, r/10, r % 10);
3442 /* Since we are processing the bitfield from lower addresses to higher,
3443 it does not matter if the process it in 32 bit chunks or 64 bit
3444 chunks as long as it is little endian. (Understand it as byte stream,
3445 beginning with the lowest byte...) If we would use big endian
3446 we would need to process it from the highest address to the lowest,
3447 in order to be agnostic to the 32 vs 64 bits issue.
3449 returns 0 on failure, 1 if we successfully received it. */
3450 static int receive_bitmap(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3452 struct bm_xfer_ctx c;
3454 enum receive_bitmap_ret ret;
3456 struct p_header80 *h = &mdev->data.rbuf.header.h80;
3458 wait_event(mdev->misc_wait, !atomic_read(&mdev->ap_bio_cnt));
3460 drbd_bm_lock(mdev, "receive bitmap");
3462 /* maybe we should use some per thread scratch page,
3463 * and allocate that during initial device creation? */
3464 buffer = (unsigned long *) __get_free_page(GFP_NOIO);
3466 dev_err(DEV, "failed to allocate one page buffer in %s\n", __func__);
3470 c = (struct bm_xfer_ctx) {
3471 .bm_bits = drbd_bm_bits(mdev),
3472 .bm_words = drbd_bm_words(mdev),
3476 if (cmd == P_BITMAP) {
3477 ret = receive_bitmap_plain(mdev, data_size, buffer, &c);
3478 } else if (cmd == P_COMPRESSED_BITMAP) {
3479 /* MAYBE: sanity check that we speak proto >= 90,
3480 * and the feature is enabled! */
3481 struct p_compressed_bm *p;
3483 if (data_size > BM_PACKET_PAYLOAD_BYTES) {
3484 dev_err(DEV, "ReportCBitmap packet too large\n");
3487 /* use the page buff */
3489 memcpy(p, h, sizeof(*h));
3490 if (drbd_recv(mdev, p->head.payload, data_size) != data_size)
3492 if (data_size <= (sizeof(*p) - sizeof(p->head))) {
3493 dev_err(DEV, "ReportCBitmap packet too small (l:%u)\n", data_size);
3496 ret = decode_bitmap_c(mdev, p, &c);
3498 dev_warn(DEV, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", cmd);
3502 c.packets[cmd == P_BITMAP]++;
3503 c.bytes[cmd == P_BITMAP] += sizeof(struct p_header80) + data_size;
3508 if (!drbd_recv_header(mdev, &cmd, &data_size))
3510 } while (ret == OK);
3514 INFO_bm_xfer_stats(mdev, "receive", &c);
3516 if (mdev->state.conn == C_WF_BITMAP_T) {
3517 ok = !drbd_send_bitmap(mdev);
3520 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
3521 ok = _drbd_request_state(mdev, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
3522 D_ASSERT(ok == SS_SUCCESS);
3523 } else if (mdev->state.conn != C_WF_BITMAP_S) {
3524 /* admin may have requested C_DISCONNECTING,
3525 * other threads may have noticed network errors */
3526 dev_info(DEV, "unexpected cstate (%s) in receive_bitmap\n",
3527 drbd_conn_str(mdev->state.conn));
3532 drbd_bm_unlock(mdev);
3533 if (ok && mdev->state.conn == C_WF_BITMAP_S)
3534 drbd_start_resync(mdev, C_SYNC_SOURCE);
3535 free_page((unsigned long) buffer);
3539 static int receive_skip(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3541 /* TODO zero copy sink :) */
3542 static char sink[128];
3545 dev_warn(DEV, "skipping unknown optional packet type %d, l: %d!\n",
3550 want = min_t(int, size, sizeof(sink));
3551 r = drbd_recv(mdev, sink, want);
3552 ERR_IF(r <= 0) break;
3558 static int receive_UnplugRemote(struct drbd_conf *mdev, enum drbd_packets cmd, unsigned int data_size)
3560 if (mdev->state.disk >= D_INCONSISTENT)
3563 /* Make sure we've acked all the TCP data associated
3564 * with the data requests being unplugged */
3565 drbd_tcp_quickack(mdev->data.socket);
3570 typedef int (*drbd_cmd_handler_f)(struct drbd_conf *, enum drbd_packets cmd, unsigned int to_receive);
3575 drbd_cmd_handler_f function;
3578 static struct data_cmd drbd_cmd_handler[] = {
3579 [P_DATA] = { 1, sizeof(struct p_data), receive_Data },
3580 [P_DATA_REPLY] = { 1, sizeof(struct p_data), receive_DataReply },
3581 [P_RS_DATA_REPLY] = { 1, sizeof(struct p_data), receive_RSDataReply } ,
3582 [P_BARRIER] = { 0, sizeof(struct p_barrier), receive_Barrier } ,
3583 [P_BITMAP] = { 1, sizeof(struct p_header80), receive_bitmap } ,
3584 [P_COMPRESSED_BITMAP] = { 1, sizeof(struct p_header80), receive_bitmap } ,
3585 [P_UNPLUG_REMOTE] = { 0, sizeof(struct p_header80), receive_UnplugRemote },
3586 [P_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
3587 [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
3588 [P_SYNC_PARAM] = { 1, sizeof(struct p_header80), receive_SyncParam },
3589 [P_SYNC_PARAM89] = { 1, sizeof(struct p_header80), receive_SyncParam },
3590 [P_PROTOCOL] = { 1, sizeof(struct p_protocol), receive_protocol },
3591 [P_UUIDS] = { 0, sizeof(struct p_uuids), receive_uuids },
3592 [P_SIZES] = { 0, sizeof(struct p_sizes), receive_sizes },
3593 [P_STATE] = { 0, sizeof(struct p_state), receive_state },
3594 [P_STATE_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_state },
3595 [P_SYNC_UUID] = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
3596 [P_OV_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
3597 [P_OV_REPLY] = { 1, sizeof(struct p_block_req), receive_DataRequest },
3598 [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
3599 [P_DELAY_PROBE] = { 0, sizeof(struct p_delay_probe93), receive_skip },
3600 /* anything missing from this table is in
3601 * the asender_tbl, see get_asender_cmd */
3602 [P_MAX_CMD] = { 0, 0, NULL },
3605 /* All handler functions that expect a sub-header get that sub-heder in
3606 mdev->data.rbuf.header.head.payload.
3608 Usually in mdev->data.rbuf.header.head the callback can find the usual
3609 p_header, but they may not rely on that. Since there is also p_header95 !
3612 static void drbdd(struct drbd_conf *mdev)
3614 union p_header *header = &mdev->data.rbuf.header;
3615 unsigned int packet_size;
3616 enum drbd_packets cmd;
3617 size_t shs; /* sub header size */
3620 while (get_t_state(&mdev->receiver) == Running) {
3621 drbd_thread_current_set_cpu(mdev);
3622 if (!drbd_recv_header(mdev, &cmd, &packet_size))
3625 if (unlikely(cmd >= P_MAX_CMD || !drbd_cmd_handler[cmd].function)) {
3626 dev_err(DEV, "unknown packet type %d, l: %d!\n", cmd, packet_size);
3630 shs = drbd_cmd_handler[cmd].pkt_size - sizeof(union p_header);
3631 rv = drbd_recv(mdev, &header->h80.payload, shs);
3632 if (unlikely(rv != shs)) {
3633 dev_err(DEV, "short read while reading sub header: rv=%d\n", rv);
3637 if (packet_size - shs > 0 && !drbd_cmd_handler[cmd].expect_payload) {
3638 dev_err(DEV, "No payload expected %s l:%d\n", cmdname(cmd), packet_size);
3642 rv = drbd_cmd_handler[cmd].function(mdev, cmd, packet_size - shs);
3644 if (unlikely(!rv)) {
3645 dev_err(DEV, "error receiving %s, l: %d!\n",
3646 cmdname(cmd), packet_size);
3653 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3655 /* If we leave here, we probably want to update at least the
3656 * "Connected" indicator on stable storage. Do so explicitly here. */
3660 void drbd_flush_workqueue(struct drbd_conf *mdev)
3662 struct drbd_wq_barrier barr;
3664 barr.w.cb = w_prev_work_done;
3665 init_completion(&barr.done);
3666 drbd_queue_work(&mdev->data.work, &barr.w);
3667 wait_for_completion(&barr.done);
3670 void drbd_free_tl_hash(struct drbd_conf *mdev)
3672 struct hlist_head *h;
3674 spin_lock_irq(&mdev->req_lock);
3676 if (!mdev->tl_hash || mdev->state.conn != C_STANDALONE) {
3677 spin_unlock_irq(&mdev->req_lock);
3681 for (h = mdev->ee_hash; h < mdev->ee_hash + mdev->ee_hash_s; h++)
3683 dev_err(DEV, "ASSERT FAILED ee_hash[%u].first == %p, expected NULL\n",
3684 (int)(h - mdev->ee_hash), h->first);
3685 kfree(mdev->ee_hash);
3686 mdev->ee_hash = NULL;
3687 mdev->ee_hash_s = 0;
3690 for (h = mdev->tl_hash; h < mdev->tl_hash + mdev->tl_hash_s; h++)
3692 dev_err(DEV, "ASSERT FAILED tl_hash[%u] == %p, expected NULL\n",
3693 (int)(h - mdev->tl_hash), h->first);
3694 kfree(mdev->tl_hash);
3695 mdev->tl_hash = NULL;
3696 mdev->tl_hash_s = 0;
3697 spin_unlock_irq(&mdev->req_lock);
3700 static void drbd_disconnect(struct drbd_conf *mdev)
3702 enum drbd_fencing_p fp;
3703 union drbd_state os, ns;
3704 int rv = SS_UNKNOWN_ERROR;
3707 if (mdev->state.conn == C_STANDALONE)
3709 if (mdev->state.conn >= C_WF_CONNECTION)
3710 dev_err(DEV, "ASSERT FAILED cstate = %s, expected < WFConnection\n",
3711 drbd_conn_str(mdev->state.conn));
3713 /* asender does not clean up anything. it must not interfere, either */
3714 drbd_thread_stop(&mdev->asender);
3715 drbd_free_sock(mdev);
3717 /* wait for current activity to cease. */
3718 spin_lock_irq(&mdev->req_lock);
3719 _drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
3720 _drbd_wait_ee_list_empty(mdev, &mdev->sync_ee);
3721 _drbd_wait_ee_list_empty(mdev, &mdev->read_ee);
3722 spin_unlock_irq(&mdev->req_lock);
3724 /* We do not have data structures that would allow us to
3725 * get the rs_pending_cnt down to 0 again.
3726 * * On C_SYNC_TARGET we do not have any data structures describing
3727 * the pending RSDataRequest's we have sent.
3728 * * On C_SYNC_SOURCE there is no data structure that tracks
3729 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
3730 * And no, it is not the sum of the reference counts in the
3731 * resync_LRU. The resync_LRU tracks the whole operation including
3732 * the disk-IO, while the rs_pending_cnt only tracks the blocks
3734 drbd_rs_cancel_all(mdev);
3736 mdev->rs_failed = 0;
3737 atomic_set(&mdev->rs_pending_cnt, 0);
3738 wake_up(&mdev->misc_wait);
3740 /* make sure syncer is stopped and w_resume_next_sg queued */
3741 del_timer_sync(&mdev->resync_timer);
3742 resync_timer_fn((unsigned long)mdev);
3744 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
3745 * w_make_resync_request etc. which may still be on the worker queue
3746 * to be "canceled" */
3747 drbd_flush_workqueue(mdev);
3749 /* This also does reclaim_net_ee(). If we do this too early, we might
3750 * miss some resync ee and pages.*/
3751 drbd_process_done_ee(mdev);
3753 kfree(mdev->p_uuid);
3754 mdev->p_uuid = NULL;
3756 if (!is_susp(mdev->state))
3759 dev_info(DEV, "Connection closed\n");
3764 if (get_ldev(mdev)) {
3765 fp = mdev->ldev->dc.fencing;
3769 if (mdev->state.role == R_PRIMARY && fp >= FP_RESOURCE && mdev->state.pdsk >= D_UNKNOWN)
3770 drbd_try_outdate_peer_async(mdev);
3772 spin_lock_irq(&mdev->req_lock);
3774 if (os.conn >= C_UNCONNECTED) {
3775 /* Do not restart in case we are C_DISCONNECTING */
3777 ns.conn = C_UNCONNECTED;
3778 rv = _drbd_set_state(mdev, ns, CS_VERBOSE, NULL);
3780 spin_unlock_irq(&mdev->req_lock);
3782 if (os.conn == C_DISCONNECTING) {
3783 wait_event(mdev->net_cnt_wait, atomic_read(&mdev->net_cnt) == 0);
3785 if (!is_susp(mdev->state)) {
3786 /* we must not free the tl_hash
3787 * while application io is still on the fly */
3788 wait_event(mdev->misc_wait, !atomic_read(&mdev->ap_bio_cnt));
3789 drbd_free_tl_hash(mdev);
3792 crypto_free_hash(mdev->cram_hmac_tfm);
3793 mdev->cram_hmac_tfm = NULL;
3795 kfree(mdev->net_conf);
3796 mdev->net_conf = NULL;
3797 drbd_request_state(mdev, NS(conn, C_STANDALONE));
3800 /* tcp_close and release of sendpage pages can be deferred. I don't
3801 * want to use SO_LINGER, because apparently it can be deferred for
3802 * more than 20 seconds (longest time I checked).
3804 * Actually we don't care for exactly when the network stack does its
3805 * put_page(), but release our reference on these pages right here.
3807 i = drbd_release_ee(mdev, &mdev->net_ee);
3809 dev_info(DEV, "net_ee not empty, killed %u entries\n", i);
3810 i = atomic_read(&mdev->pp_in_use_by_net);
3812 dev_info(DEV, "pp_in_use_by_net = %d, expected 0\n", i);
3813 i = atomic_read(&mdev->pp_in_use);
3815 dev_info(DEV, "pp_in_use = %d, expected 0\n", i);
3817 D_ASSERT(list_empty(&mdev->read_ee));
3818 D_ASSERT(list_empty(&mdev->active_ee));
3819 D_ASSERT(list_empty(&mdev->sync_ee));
3820 D_ASSERT(list_empty(&mdev->done_ee));
3822 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
3823 atomic_set(&mdev->current_epoch->epoch_size, 0);
3824 D_ASSERT(list_empty(&mdev->current_epoch->list));
3828 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
3829 * we can agree on is stored in agreed_pro_version.
3831 * feature flags and the reserved array should be enough room for future
3832 * enhancements of the handshake protocol, and possible plugins...
3834 * for now, they are expected to be zero, but ignored.
3836 static int drbd_send_handshake(struct drbd_conf *mdev)
3838 /* ASSERT current == mdev->receiver ... */
3839 struct p_handshake *p = &mdev->data.sbuf.handshake;
3842 if (mutex_lock_interruptible(&mdev->data.mutex)) {
3843 dev_err(DEV, "interrupted during initial handshake\n");
3844 return 0; /* interrupted. not ok. */
3847 if (mdev->data.socket == NULL) {
3848 mutex_unlock(&mdev->data.mutex);
3852 memset(p, 0, sizeof(*p));
3853 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
3854 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
3855 ok = _drbd_send_cmd( mdev, mdev->data.socket, P_HAND_SHAKE,
3856 (struct p_header80 *)p, sizeof(*p), 0 );
3857 mutex_unlock(&mdev->data.mutex);
3863 * 1 yes, we have a valid connection
3864 * 0 oops, did not work out, please try again
3865 * -1 peer talks different language,
3866 * no point in trying again, please go standalone.
3868 static int drbd_do_handshake(struct drbd_conf *mdev)
3870 /* ASSERT current == mdev->receiver ... */
3871 struct p_handshake *p = &mdev->data.rbuf.handshake;
3872 const int expect = sizeof(struct p_handshake) - sizeof(struct p_header80);
3873 unsigned int length;
3874 enum drbd_packets cmd;
3877 rv = drbd_send_handshake(mdev);
3881 rv = drbd_recv_header(mdev, &cmd, &length);
3885 if (cmd != P_HAND_SHAKE) {
3886 dev_err(DEV, "expected HandShake packet, received: %s (0x%04x)\n",
3891 if (length != expect) {
3892 dev_err(DEV, "expected HandShake length: %u, received: %u\n",
3897 rv = drbd_recv(mdev, &p->head.payload, expect);
3900 dev_err(DEV, "short read receiving handshake packet: l=%u\n", rv);
3904 p->protocol_min = be32_to_cpu(p->protocol_min);
3905 p->protocol_max = be32_to_cpu(p->protocol_max);
3906 if (p->protocol_max == 0)
3907 p->protocol_max = p->protocol_min;
3909 if (PRO_VERSION_MAX < p->protocol_min ||
3910 PRO_VERSION_MIN > p->protocol_max)
3913 mdev->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
3915 dev_info(DEV, "Handshake successful: "
3916 "Agreed network protocol version %d\n", mdev->agreed_pro_version);
3921 dev_err(DEV, "incompatible DRBD dialects: "
3922 "I support %d-%d, peer supports %d-%d\n",
3923 PRO_VERSION_MIN, PRO_VERSION_MAX,
3924 p->protocol_min, p->protocol_max);
3928 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
3929 static int drbd_do_auth(struct drbd_conf *mdev)
3931 dev_err(DEV, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
3932 dev_err(DEV, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
3936 #define CHALLENGE_LEN 64
3940 0 - failed, try again (network error),
3941 -1 - auth failed, don't try again.
3944 static int drbd_do_auth(struct drbd_conf *mdev)
3946 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
3947 struct scatterlist sg;
3948 char *response = NULL;
3949 char *right_response = NULL;
3950 char *peers_ch = NULL;
3951 unsigned int key_len = strlen(mdev->net_conf->shared_secret);
3952 unsigned int resp_size;
3953 struct hash_desc desc;
3954 enum drbd_packets cmd;
3955 unsigned int length;
3958 desc.tfm = mdev->cram_hmac_tfm;
3961 rv = crypto_hash_setkey(mdev->cram_hmac_tfm,
3962 (u8 *)mdev->net_conf->shared_secret, key_len);
3964 dev_err(DEV, "crypto_hash_setkey() failed with %d\n", rv);
3969 get_random_bytes(my_challenge, CHALLENGE_LEN);
3971 rv = drbd_send_cmd2(mdev, P_AUTH_CHALLENGE, my_challenge, CHALLENGE_LEN);
3975 rv = drbd_recv_header(mdev, &cmd, &length);
3979 if (cmd != P_AUTH_CHALLENGE) {
3980 dev_err(DEV, "expected AuthChallenge packet, received: %s (0x%04x)\n",
3986 if (length > CHALLENGE_LEN * 2) {
3987 dev_err(DEV, "expected AuthChallenge payload too big.\n");
3992 peers_ch = kmalloc(length, GFP_NOIO);
3993 if (peers_ch == NULL) {
3994 dev_err(DEV, "kmalloc of peers_ch failed\n");
3999 rv = drbd_recv(mdev, peers_ch, length);
4002 dev_err(DEV, "short read AuthChallenge: l=%u\n", rv);
4007 resp_size = crypto_hash_digestsize(mdev->cram_hmac_tfm);
4008 response = kmalloc(resp_size, GFP_NOIO);
4009 if (response == NULL) {
4010 dev_err(DEV, "kmalloc of response failed\n");
4015 sg_init_table(&sg, 1);
4016 sg_set_buf(&sg, peers_ch, length);
4018 rv = crypto_hash_digest(&desc, &sg, sg.length, response);
4020 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv);
4025 rv = drbd_send_cmd2(mdev, P_AUTH_RESPONSE, response, resp_size);
4029 rv = drbd_recv_header(mdev, &cmd, &length);
4033 if (cmd != P_AUTH_RESPONSE) {
4034 dev_err(DEV, "expected AuthResponse packet, received: %s (0x%04x)\n",
4040 if (length != resp_size) {
4041 dev_err(DEV, "expected AuthResponse payload of wrong size\n");
4046 rv = drbd_recv(mdev, response , resp_size);
4048 if (rv != resp_size) {
4049 dev_err(DEV, "short read receiving AuthResponse: l=%u\n", rv);
4054 right_response = kmalloc(resp_size, GFP_NOIO);
4055 if (right_response == NULL) {
4056 dev_err(DEV, "kmalloc of right_response failed\n");
4061 sg_set_buf(&sg, my_challenge, CHALLENGE_LEN);
4063 rv = crypto_hash_digest(&desc, &sg, sg.length, right_response);
4065 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv);
4070 rv = !memcmp(response, right_response, resp_size);
4073 dev_info(DEV, "Peer authenticated using %d bytes of '%s' HMAC\n",
4074 resp_size, mdev->net_conf->cram_hmac_alg);
4081 kfree(right_response);
4087 int drbdd_init(struct drbd_thread *thi)
4089 struct drbd_conf *mdev = thi->mdev;
4090 unsigned int minor = mdev_to_minor(mdev);
4093 sprintf(current->comm, "drbd%d_receiver", minor);
4095 dev_info(DEV, "receiver (re)started\n");
4098 h = drbd_connect(mdev);
4100 drbd_disconnect(mdev);
4101 __set_current_state(TASK_INTERRUPTIBLE);
4102 schedule_timeout(HZ);
4105 dev_warn(DEV, "Discarding network configuration.\n");
4106 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
4111 if (get_net_conf(mdev)) {
4117 drbd_disconnect(mdev);
4119 dev_info(DEV, "receiver terminated\n");
4123 /* ********* acknowledge sender ******** */
4125 static int got_RqSReply(struct drbd_conf *mdev, struct p_header80 *h)
4127 struct p_req_state_reply *p = (struct p_req_state_reply *)h;
4129 int retcode = be32_to_cpu(p->retcode);
4131 if (retcode >= SS_SUCCESS) {
4132 set_bit(CL_ST_CHG_SUCCESS, &mdev->flags);
4134 set_bit(CL_ST_CHG_FAIL, &mdev->flags);
4135 dev_err(DEV, "Requested state change failed by peer: %s (%d)\n",
4136 drbd_set_st_err_str(retcode), retcode);
4138 wake_up(&mdev->state_wait);
4143 static int got_Ping(struct drbd_conf *mdev, struct p_header80 *h)
4145 return drbd_send_ping_ack(mdev);
4149 static int got_PingAck(struct drbd_conf *mdev, struct p_header80 *h)
4151 /* restore idle timeout */
4152 mdev->meta.socket->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ;
4153 if (!test_and_set_bit(GOT_PING_ACK, &mdev->flags))
4154 wake_up(&mdev->misc_wait);
4159 static int got_IsInSync(struct drbd_conf *mdev, struct p_header80 *h)
4161 struct p_block_ack *p = (struct p_block_ack *)h;
4162 sector_t sector = be64_to_cpu(p->sector);
4163 int blksize = be32_to_cpu(p->blksize);
4165 D_ASSERT(mdev->agreed_pro_version >= 89);
4167 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4169 if (get_ldev(mdev)) {
4170 drbd_rs_complete_io(mdev, sector);
4171 drbd_set_in_sync(mdev, sector, blksize);
4172 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4173 mdev->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
4176 dec_rs_pending(mdev);
4177 atomic_add(blksize >> 9, &mdev->rs_sect_in);
4182 /* when we receive the ACK for a write request,
4183 * verify that we actually know about it */
4184 static struct drbd_request *_ack_id_to_req(struct drbd_conf *mdev,
4185 u64 id, sector_t sector)
4187 struct hlist_head *slot = tl_hash_slot(mdev, sector);
4188 struct hlist_node *n;
4189 struct drbd_request *req;
4191 hlist_for_each_entry(req, n, slot, colision) {
4192 if ((unsigned long)req == (unsigned long)id) {
4193 if (req->sector != sector) {
4194 dev_err(DEV, "_ack_id_to_req: found req %p but it has "
4195 "wrong sector (%llus versus %llus)\n", req,
4196 (unsigned long long)req->sector,
4197 (unsigned long long)sector);
4203 dev_err(DEV, "_ack_id_to_req: failed to find req %p, sector %llus in list\n",
4204 (void *)(unsigned long)id, (unsigned long long)sector);
4208 typedef struct drbd_request *(req_validator_fn)
4209 (struct drbd_conf *mdev, u64 id, sector_t sector);
4211 static int validate_req_change_req_state(struct drbd_conf *mdev,
4212 u64 id, sector_t sector, req_validator_fn validator,
4213 const char *func, enum drbd_req_event what)
4215 struct drbd_request *req;
4216 struct bio_and_error m;
4218 spin_lock_irq(&mdev->req_lock);
4219 req = validator(mdev, id, sector);
4220 if (unlikely(!req)) {
4221 spin_unlock_irq(&mdev->req_lock);
4222 dev_err(DEV, "%s: got a corrupt block_id/sector pair\n", func);
4225 __req_mod(req, what, &m);
4226 spin_unlock_irq(&mdev->req_lock);
4229 complete_master_bio(mdev, &m);
4233 static int got_BlockAck(struct drbd_conf *mdev, struct p_header80 *h)
4235 struct p_block_ack *p = (struct p_block_ack *)h;
4236 sector_t sector = be64_to_cpu(p->sector);
4237 int blksize = be32_to_cpu(p->blksize);
4238 enum drbd_req_event what;
4240 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4242 if (is_syncer_block_id(p->block_id)) {
4243 drbd_set_in_sync(mdev, sector, blksize);
4244 dec_rs_pending(mdev);
4247 switch (be16_to_cpu(h->command)) {
4248 case P_RS_WRITE_ACK:
4249 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4250 what = write_acked_by_peer_and_sis;
4253 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4254 what = write_acked_by_peer;
4257 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_B);
4258 what = recv_acked_by_peer;
4261 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4262 what = conflict_discarded_by_peer;
4269 return validate_req_change_req_state(mdev, p->block_id, sector,
4270 _ack_id_to_req, __func__ , what);
4273 static int got_NegAck(struct drbd_conf *mdev, struct p_header80 *h)
4275 struct p_block_ack *p = (struct p_block_ack *)h;
4276 sector_t sector = be64_to_cpu(p->sector);
4278 if (__ratelimit(&drbd_ratelimit_state))
4279 dev_warn(DEV, "Got NegAck packet. Peer is in troubles?\n");
4281 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4283 if (is_syncer_block_id(p->block_id)) {
4284 int size = be32_to_cpu(p->blksize);
4285 dec_rs_pending(mdev);
4286 drbd_rs_failed_io(mdev, sector, size);
4289 return validate_req_change_req_state(mdev, p->block_id, sector,
4290 _ack_id_to_req, __func__ , neg_acked);
4293 static int got_NegDReply(struct drbd_conf *mdev, struct p_header80 *h)
4295 struct p_block_ack *p = (struct p_block_ack *)h;
4296 sector_t sector = be64_to_cpu(p->sector);
4298 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4299 dev_err(DEV, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4300 (unsigned long long)sector, be32_to_cpu(p->blksize));
4302 return validate_req_change_req_state(mdev, p->block_id, sector,
4303 _ar_id_to_req, __func__ , neg_acked);
4306 static int got_NegRSDReply(struct drbd_conf *mdev, struct p_header80 *h)
4310 struct p_block_ack *p = (struct p_block_ack *)h;
4312 sector = be64_to_cpu(p->sector);
4313 size = be32_to_cpu(p->blksize);
4315 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4317 dec_rs_pending(mdev);
4319 if (get_ldev_if_state(mdev, D_FAILED)) {
4320 drbd_rs_complete_io(mdev, sector);
4321 drbd_rs_failed_io(mdev, sector, size);
4328 static int got_BarrierAck(struct drbd_conf *mdev, struct p_header80 *h)
4330 struct p_barrier_ack *p = (struct p_barrier_ack *)h;
4332 tl_release(mdev, p->barrier, be32_to_cpu(p->set_size));
4337 static int got_OVResult(struct drbd_conf *mdev, struct p_header80 *h)
4339 struct p_block_ack *p = (struct p_block_ack *)h;
4340 struct drbd_work *w;
4344 sector = be64_to_cpu(p->sector);
4345 size = be32_to_cpu(p->blksize);
4347 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4349 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
4350 drbd_ov_oos_found(mdev, sector, size);
4354 if (!get_ldev(mdev))
4357 drbd_rs_complete_io(mdev, sector);
4358 dec_rs_pending(mdev);
4360 if (--mdev->ov_left == 0) {
4361 w = kmalloc(sizeof(*w), GFP_NOIO);
4363 w->cb = w_ov_finished;
4364 drbd_queue_work_front(&mdev->data.work, w);
4366 dev_err(DEV, "kmalloc(w) failed.");
4368 drbd_resync_finished(mdev);
4375 static int got_skip(struct drbd_conf *mdev, struct p_header80 *h)
4380 struct asender_cmd {
4382 int (*process)(struct drbd_conf *mdev, struct p_header80 *h);
4385 static struct asender_cmd *get_asender_cmd(int cmd)
4387 static struct asender_cmd asender_tbl[] = {
4388 /* anything missing from this table is in
4389 * the drbd_cmd_handler (drbd_default_handler) table,
4390 * see the beginning of drbdd() */
4391 [P_PING] = { sizeof(struct p_header80), got_Ping },
4392 [P_PING_ACK] = { sizeof(struct p_header80), got_PingAck },
4393 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4394 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4395 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4396 [P_DISCARD_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4397 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
4398 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
4399 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply},
4400 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
4401 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
4402 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
4403 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
4404 [P_DELAY_PROBE] = { sizeof(struct p_delay_probe93), got_skip },
4405 [P_MAX_CMD] = { 0, NULL },
4407 if (cmd > P_MAX_CMD || asender_tbl[cmd].process == NULL)
4409 return &asender_tbl[cmd];
4412 int drbd_asender(struct drbd_thread *thi)
4414 struct drbd_conf *mdev = thi->mdev;
4415 struct p_header80 *h = &mdev->meta.rbuf.header.h80;
4416 struct asender_cmd *cmd = NULL;
4421 int expect = sizeof(struct p_header80);
4424 sprintf(current->comm, "drbd%d_asender", mdev_to_minor(mdev));
4426 current->policy = SCHED_RR; /* Make this a realtime task! */
4427 current->rt_priority = 2; /* more important than all other tasks */
4429 while (get_t_state(thi) == Running) {
4430 drbd_thread_current_set_cpu(mdev);
4431 if (test_and_clear_bit(SEND_PING, &mdev->flags)) {
4432 ERR_IF(!drbd_send_ping(mdev)) goto reconnect;
4433 mdev->meta.socket->sk->sk_rcvtimeo =
4434 mdev->net_conf->ping_timeo*HZ/10;
4437 /* conditionally cork;
4438 * it may hurt latency if we cork without much to send */
4439 if (!mdev->net_conf->no_cork &&
4440 3 < atomic_read(&mdev->unacked_cnt))
4441 drbd_tcp_cork(mdev->meta.socket);
4443 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4444 flush_signals(current);
4445 if (!drbd_process_done_ee(mdev))
4447 /* to avoid race with newly queued ACKs */
4448 set_bit(SIGNAL_ASENDER, &mdev->flags);
4449 spin_lock_irq(&mdev->req_lock);
4450 empty = list_empty(&mdev->done_ee);
4451 spin_unlock_irq(&mdev->req_lock);
4452 /* new ack may have been queued right here,
4453 * but then there is also a signal pending,
4454 * and we start over... */
4458 /* but unconditionally uncork unless disabled */
4459 if (!mdev->net_conf->no_cork)
4460 drbd_tcp_uncork(mdev->meta.socket);
4462 /* short circuit, recv_msg would return EINTR anyways. */
4463 if (signal_pending(current))
4466 rv = drbd_recv_short(mdev, mdev->meta.socket,
4467 buf, expect-received, 0);
4468 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4470 flush_signals(current);
4473 * -EINTR (on meta) we got a signal
4474 * -EAGAIN (on meta) rcvtimeo expired
4475 * -ECONNRESET other side closed the connection
4476 * -ERESTARTSYS (on data) we got a signal
4477 * rv < 0 other than above: unexpected error!
4478 * rv == expected: full header or command
4479 * rv < expected: "woken" by signal during receive
4480 * rv == 0 : "connection shut down by peer"
4482 if (likely(rv > 0)) {
4485 } else if (rv == 0) {
4486 dev_err(DEV, "meta connection shut down by peer.\n");
4488 } else if (rv == -EAGAIN) {
4489 if (mdev->meta.socket->sk->sk_rcvtimeo ==
4490 mdev->net_conf->ping_timeo*HZ/10) {
4491 dev_err(DEV, "PingAck did not arrive in time.\n");
4494 set_bit(SEND_PING, &mdev->flags);
4496 } else if (rv == -EINTR) {
4499 dev_err(DEV, "sock_recvmsg returned %d\n", rv);
4503 if (received == expect && cmd == NULL) {
4504 if (unlikely(h->magic != BE_DRBD_MAGIC)) {
4505 dev_err(DEV, "magic?? on meta m: 0x%08x c: %d l: %d\n",
4506 be32_to_cpu(h->magic),
4507 be16_to_cpu(h->command),
4508 be16_to_cpu(h->length));
4511 cmd = get_asender_cmd(be16_to_cpu(h->command));
4512 len = be16_to_cpu(h->length);
4513 if (unlikely(cmd == NULL)) {
4514 dev_err(DEV, "unknown command?? on meta m: 0x%08x c: %d l: %d\n",
4515 be32_to_cpu(h->magic),
4516 be16_to_cpu(h->command),
4517 be16_to_cpu(h->length));
4520 expect = cmd->pkt_size;
4521 ERR_IF(len != expect-sizeof(struct p_header80))
4524 if (received == expect) {
4525 D_ASSERT(cmd != NULL);
4526 if (!cmd->process(mdev, h))
4531 expect = sizeof(struct p_header80);
4538 drbd_force_state(mdev, NS(conn, C_NETWORK_FAILURE));
4543 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
4546 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4548 D_ASSERT(mdev->state.conn < C_CONNECTED);
4549 dev_info(DEV, "asender terminated\n");