4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
28 #include <asm/uaccess.h>
31 #include <linux/drbd.h>
33 #include <linux/file.h>
36 #include <linux/memcontrol.h>
37 #include <linux/mm_inline.h>
38 #include <linux/slab.h>
39 #include <linux/pkt_sched.h>
40 #define __KERNEL_SYSCALLS__
41 #include <linux/unistd.h>
42 #include <linux/vmalloc.h>
43 #include <linux/random.h>
44 #include <linux/string.h>
45 #include <linux/scatterlist.h>
57 static int drbd_do_handshake(struct drbd_conf
*mdev
);
58 static int drbd_do_auth(struct drbd_conf
*mdev
);
60 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*, struct drbd_epoch
*, enum epoch_event
);
61 static int e_end_block(struct drbd_conf
*, struct drbd_work
*, int);
64 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
67 * some helper functions to deal with single linked page lists,
68 * page->private being our "next" pointer.
71 /* If at least n pages are linked at head, get n pages off.
72 * Otherwise, don't modify head, and return NULL.
73 * Locking is the responsibility of the caller.
75 static struct page
*page_chain_del(struct page
**head
, int n
)
89 tmp
= page_chain_next(page
);
91 break; /* found sufficient pages */
93 /* insufficient pages, don't use any of them. */
98 /* add end of list marker for the returned list */
99 set_page_private(page
, 0);
100 /* actual return value, and adjustment of head */
106 /* may be used outside of locks to find the tail of a (usually short)
107 * "private" page chain, before adding it back to a global chain head
108 * with page_chain_add() under a spinlock. */
109 static struct page
*page_chain_tail(struct page
*page
, int *len
)
113 while ((tmp
= page_chain_next(page
)))
120 static int page_chain_free(struct page
*page
)
124 page_chain_for_each_safe(page
, tmp
) {
131 static void page_chain_add(struct page
**head
,
132 struct page
*chain_first
, struct page
*chain_last
)
136 tmp
= page_chain_tail(chain_first
, NULL
);
137 BUG_ON(tmp
!= chain_last
);
140 /* add chain to head */
141 set_page_private(chain_last
, (unsigned long)*head
);
145 static struct page
*drbd_pp_first_pages_or_try_alloc(struct drbd_conf
*mdev
, int number
)
147 struct page
*page
= NULL
;
148 struct page
*tmp
= NULL
;
151 /* Yes, testing drbd_pp_vacant outside the lock is racy.
152 * So what. It saves a spin_lock. */
153 if (drbd_pp_vacant
>= number
) {
154 spin_lock(&drbd_pp_lock
);
155 page
= page_chain_del(&drbd_pp_pool
, number
);
157 drbd_pp_vacant
-= number
;
158 spin_unlock(&drbd_pp_lock
);
163 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
164 * "criss-cross" setup, that might cause write-out on some other DRBD,
165 * which in turn might block on the other node at this very place. */
166 for (i
= 0; i
< number
; i
++) {
167 tmp
= alloc_page(GFP_TRY
);
170 set_page_private(tmp
, (unsigned long)page
);
177 /* Not enough pages immediately available this time.
178 * No need to jump around here, drbd_pp_alloc will retry this
179 * function "soon". */
181 tmp
= page_chain_tail(page
, NULL
);
182 spin_lock(&drbd_pp_lock
);
183 page_chain_add(&drbd_pp_pool
, page
, tmp
);
185 spin_unlock(&drbd_pp_lock
);
190 static void reclaim_net_ee(struct drbd_conf
*mdev
, struct list_head
*to_be_freed
)
192 struct drbd_epoch_entry
*e
;
193 struct list_head
*le
, *tle
;
195 /* The EEs are always appended to the end of the list. Since
196 they are sent in order over the wire, they have to finish
197 in order. As soon as we see the first not finished we can
198 stop to examine the list... */
200 list_for_each_safe(le
, tle
, &mdev
->net_ee
) {
201 e
= list_entry(le
, struct drbd_epoch_entry
, w
.list
);
202 if (drbd_ee_has_active_page(e
))
204 list_move(le
, to_be_freed
);
208 static void drbd_kick_lo_and_reclaim_net(struct drbd_conf
*mdev
)
210 LIST_HEAD(reclaimed
);
211 struct drbd_epoch_entry
*e
, *t
;
213 spin_lock_irq(&mdev
->req_lock
);
214 reclaim_net_ee(mdev
, &reclaimed
);
215 spin_unlock_irq(&mdev
->req_lock
);
217 list_for_each_entry_safe(e
, t
, &reclaimed
, w
.list
)
218 drbd_free_net_ee(mdev
, e
);
222 * drbd_pp_alloc() - Returns @number pages, retries forever (or until signalled)
223 * @mdev: DRBD device.
224 * @number: number of pages requested
225 * @retry: whether to retry, if not enough pages are available right now
227 * Tries to allocate number pages, first from our own page pool, then from
228 * the kernel, unless this allocation would exceed the max_buffers setting.
229 * Possibly retry until DRBD frees sufficient pages somewhere else.
231 * Returns a page chain linked via page->private.
233 static struct page
*drbd_pp_alloc(struct drbd_conf
*mdev
, unsigned number
, bool retry
)
235 struct page
*page
= NULL
;
238 /* Yes, we may run up to @number over max_buffers. If we
239 * follow it strictly, the admin will get it wrong anyways. */
240 if (atomic_read(&mdev
->pp_in_use
) < mdev
->net_conf
->max_buffers
)
241 page
= drbd_pp_first_pages_or_try_alloc(mdev
, number
);
243 while (page
== NULL
) {
244 prepare_to_wait(&drbd_pp_wait
, &wait
, TASK_INTERRUPTIBLE
);
246 drbd_kick_lo_and_reclaim_net(mdev
);
248 if (atomic_read(&mdev
->pp_in_use
) < mdev
->net_conf
->max_buffers
) {
249 page
= drbd_pp_first_pages_or_try_alloc(mdev
, number
);
257 if (signal_pending(current
)) {
258 dev_warn(DEV
, "drbd_pp_alloc interrupted!\n");
264 finish_wait(&drbd_pp_wait
, &wait
);
267 atomic_add(number
, &mdev
->pp_in_use
);
271 /* Must not be used from irq, as that may deadlock: see drbd_pp_alloc.
272 * Is also used from inside an other spin_lock_irq(&mdev->req_lock);
273 * Either links the page chain back to the global pool,
274 * or returns all pages to the system. */
275 static void drbd_pp_free(struct drbd_conf
*mdev
, struct page
*page
, int is_net
)
277 atomic_t
*a
= is_net
? &mdev
->pp_in_use_by_net
: &mdev
->pp_in_use
;
280 if (drbd_pp_vacant
> (DRBD_MAX_SEGMENT_SIZE
/PAGE_SIZE
)*minor_count
)
281 i
= page_chain_free(page
);
284 tmp
= page_chain_tail(page
, &i
);
285 spin_lock(&drbd_pp_lock
);
286 page_chain_add(&drbd_pp_pool
, page
, tmp
);
288 spin_unlock(&drbd_pp_lock
);
290 i
= atomic_sub_return(i
, a
);
292 dev_warn(DEV
, "ASSERTION FAILED: %s: %d < 0\n",
293 is_net
? "pp_in_use_by_net" : "pp_in_use", i
);
294 wake_up(&drbd_pp_wait
);
298 You need to hold the req_lock:
299 _drbd_wait_ee_list_empty()
301 You must not have the req_lock:
307 drbd_process_done_ee()
309 drbd_wait_ee_list_empty()
312 struct drbd_epoch_entry
*drbd_alloc_ee(struct drbd_conf
*mdev
,
315 unsigned int data_size
,
316 gfp_t gfp_mask
) __must_hold(local
)
318 struct drbd_epoch_entry
*e
;
320 unsigned nr_pages
= (data_size
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
322 if (FAULT_ACTIVE(mdev
, DRBD_FAULT_AL_EE
))
325 e
= mempool_alloc(drbd_ee_mempool
, gfp_mask
& ~__GFP_HIGHMEM
);
327 if (!(gfp_mask
& __GFP_NOWARN
))
328 dev_err(DEV
, "alloc_ee: Allocation of an EE failed\n");
332 page
= drbd_pp_alloc(mdev
, nr_pages
, (gfp_mask
& __GFP_WAIT
));
336 INIT_HLIST_NODE(&e
->colision
);
340 atomic_set(&e
->pending_bios
, 0);
349 mempool_free(e
, drbd_ee_mempool
);
353 void drbd_free_some_ee(struct drbd_conf
*mdev
, struct drbd_epoch_entry
*e
, int is_net
)
355 if (e
->flags
& EE_HAS_DIGEST
)
357 drbd_pp_free(mdev
, e
->pages
, is_net
);
358 D_ASSERT(atomic_read(&e
->pending_bios
) == 0);
359 D_ASSERT(hlist_unhashed(&e
->colision
));
360 mempool_free(e
, drbd_ee_mempool
);
363 int drbd_release_ee(struct drbd_conf
*mdev
, struct list_head
*list
)
365 LIST_HEAD(work_list
);
366 struct drbd_epoch_entry
*e
, *t
;
368 int is_net
= list
== &mdev
->net_ee
;
370 spin_lock_irq(&mdev
->req_lock
);
371 list_splice_init(list
, &work_list
);
372 spin_unlock_irq(&mdev
->req_lock
);
374 list_for_each_entry_safe(e
, t
, &work_list
, w
.list
) {
375 drbd_free_some_ee(mdev
, e
, is_net
);
383 * This function is called from _asender only_
384 * but see also comments in _req_mod(,barrier_acked)
385 * and receive_Barrier.
387 * Move entries from net_ee to done_ee, if ready.
388 * Grab done_ee, call all callbacks, free the entries.
389 * The callbacks typically send out ACKs.
391 static int drbd_process_done_ee(struct drbd_conf
*mdev
)
393 LIST_HEAD(work_list
);
394 LIST_HEAD(reclaimed
);
395 struct drbd_epoch_entry
*e
, *t
;
396 int ok
= (mdev
->state
.conn
>= C_WF_REPORT_PARAMS
);
398 spin_lock_irq(&mdev
->req_lock
);
399 reclaim_net_ee(mdev
, &reclaimed
);
400 list_splice_init(&mdev
->done_ee
, &work_list
);
401 spin_unlock_irq(&mdev
->req_lock
);
403 list_for_each_entry_safe(e
, t
, &reclaimed
, w
.list
)
404 drbd_free_net_ee(mdev
, e
);
406 /* possible callbacks here:
407 * e_end_block, and e_end_resync_block, e_send_discard_ack.
408 * all ignore the last argument.
410 list_for_each_entry_safe(e
, t
, &work_list
, w
.list
) {
411 /* list_del not necessary, next/prev members not touched */
412 ok
= e
->w
.cb(mdev
, &e
->w
, !ok
) && ok
;
413 drbd_free_ee(mdev
, e
);
415 wake_up(&mdev
->ee_wait
);
420 void _drbd_wait_ee_list_empty(struct drbd_conf
*mdev
, struct list_head
*head
)
424 /* avoids spin_lock/unlock
425 * and calling prepare_to_wait in the fast path */
426 while (!list_empty(head
)) {
427 prepare_to_wait(&mdev
->ee_wait
, &wait
, TASK_UNINTERRUPTIBLE
);
428 spin_unlock_irq(&mdev
->req_lock
);
430 finish_wait(&mdev
->ee_wait
, &wait
);
431 spin_lock_irq(&mdev
->req_lock
);
435 void drbd_wait_ee_list_empty(struct drbd_conf
*mdev
, struct list_head
*head
)
437 spin_lock_irq(&mdev
->req_lock
);
438 _drbd_wait_ee_list_empty(mdev
, head
);
439 spin_unlock_irq(&mdev
->req_lock
);
442 /* see also kernel_accept; which is only present since 2.6.18.
443 * also we want to log which part of it failed, exactly */
444 static int drbd_accept(struct drbd_conf
*mdev
, const char **what
,
445 struct socket
*sock
, struct socket
**newsock
)
447 struct sock
*sk
= sock
->sk
;
451 err
= sock
->ops
->listen(sock
, 5);
455 *what
= "sock_create_lite";
456 err
= sock_create_lite(sk
->sk_family
, sk
->sk_type
, sk
->sk_protocol
,
462 err
= sock
->ops
->accept(sock
, *newsock
, 0);
464 sock_release(*newsock
);
468 (*newsock
)->ops
= sock
->ops
;
474 static int drbd_recv_short(struct drbd_conf
*mdev
, struct socket
*sock
,
475 void *buf
, size_t size
, int flags
)
482 struct msghdr msg
= {
484 .msg_iov
= (struct iovec
*)&iov
,
485 .msg_flags
= (flags
? flags
: MSG_WAITALL
| MSG_NOSIGNAL
)
491 rv
= sock_recvmsg(sock
, &msg
, size
, msg
.msg_flags
);
497 static int drbd_recv(struct drbd_conf
*mdev
, void *buf
, size_t size
)
504 struct msghdr msg
= {
506 .msg_iov
= (struct iovec
*)&iov
,
507 .msg_flags
= MSG_WAITALL
| MSG_NOSIGNAL
515 rv
= sock_recvmsg(mdev
->data
.socket
, &msg
, size
, msg
.msg_flags
);
520 * ECONNRESET other side closed the connection
521 * ERESTARTSYS (on sock) we got a signal
525 if (rv
== -ECONNRESET
)
526 dev_info(DEV
, "sock was reset by peer\n");
527 else if (rv
!= -ERESTARTSYS
)
528 dev_err(DEV
, "sock_recvmsg returned %d\n", rv
);
530 } else if (rv
== 0) {
531 dev_info(DEV
, "sock was shut down by peer\n");
534 /* signal came in, or peer/link went down,
535 * after we read a partial message
537 /* D_ASSERT(signal_pending(current)); */
545 drbd_force_state(mdev
, NS(conn
, C_BROKEN_PIPE
));
551 * On individual connections, the socket buffer size must be set prior to the
552 * listen(2) or connect(2) calls in order to have it take effect.
553 * This is our wrapper to do so.
555 static void drbd_setbufsize(struct socket
*sock
, unsigned int snd
,
558 /* open coded SO_SNDBUF, SO_RCVBUF */
560 sock
->sk
->sk_sndbuf
= snd
;
561 sock
->sk
->sk_userlocks
|= SOCK_SNDBUF_LOCK
;
564 sock
->sk
->sk_rcvbuf
= rcv
;
565 sock
->sk
->sk_userlocks
|= SOCK_RCVBUF_LOCK
;
569 static struct socket
*drbd_try_connect(struct drbd_conf
*mdev
)
573 struct sockaddr_in6 src_in6
;
575 int disconnect_on_error
= 1;
577 if (!get_net_conf(mdev
))
580 what
= "sock_create_kern";
581 err
= sock_create_kern(((struct sockaddr
*)mdev
->net_conf
->my_addr
)->sa_family
,
582 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
588 sock
->sk
->sk_rcvtimeo
=
589 sock
->sk
->sk_sndtimeo
= mdev
->net_conf
->try_connect_int
*HZ
;
590 drbd_setbufsize(sock
, mdev
->net_conf
->sndbuf_size
,
591 mdev
->net_conf
->rcvbuf_size
);
593 /* explicitly bind to the configured IP as source IP
594 * for the outgoing connections.
595 * This is needed for multihomed hosts and to be
596 * able to use lo: interfaces for drbd.
597 * Make sure to use 0 as port number, so linux selects
598 * a free one dynamically.
600 memcpy(&src_in6
, mdev
->net_conf
->my_addr
,
601 min_t(int, mdev
->net_conf
->my_addr_len
, sizeof(src_in6
)));
602 if (((struct sockaddr
*)mdev
->net_conf
->my_addr
)->sa_family
== AF_INET6
)
603 src_in6
.sin6_port
= 0;
605 ((struct sockaddr_in
*)&src_in6
)->sin_port
= 0; /* AF_INET & AF_SCI */
607 what
= "bind before connect";
608 err
= sock
->ops
->bind(sock
,
609 (struct sockaddr
*) &src_in6
,
610 mdev
->net_conf
->my_addr_len
);
614 /* connect may fail, peer not yet available.
615 * stay C_WF_CONNECTION, don't go Disconnecting! */
616 disconnect_on_error
= 0;
618 err
= sock
->ops
->connect(sock
,
619 (struct sockaddr
*)mdev
->net_conf
->peer_addr
,
620 mdev
->net_conf
->peer_addr_len
, 0);
629 /* timeout, busy, signal pending */
630 case ETIMEDOUT
: case EAGAIN
: case EINPROGRESS
:
631 case EINTR
: case ERESTARTSYS
:
632 /* peer not (yet) available, network problem */
633 case ECONNREFUSED
: case ENETUNREACH
:
634 case EHOSTDOWN
: case EHOSTUNREACH
:
635 disconnect_on_error
= 0;
638 dev_err(DEV
, "%s failed, err = %d\n", what
, err
);
640 if (disconnect_on_error
)
641 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
647 static struct socket
*drbd_wait_for_connect(struct drbd_conf
*mdev
)
650 struct socket
*s_estab
= NULL
, *s_listen
;
653 if (!get_net_conf(mdev
))
656 what
= "sock_create_kern";
657 err
= sock_create_kern(((struct sockaddr
*)mdev
->net_conf
->my_addr
)->sa_family
,
658 SOCK_STREAM
, IPPROTO_TCP
, &s_listen
);
664 timeo
= mdev
->net_conf
->try_connect_int
* HZ
;
665 timeo
+= (random32() & 1) ? timeo
/ 7 : -timeo
/ 7; /* 28.5% random jitter */
667 s_listen
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
668 s_listen
->sk
->sk_rcvtimeo
= timeo
;
669 s_listen
->sk
->sk_sndtimeo
= timeo
;
670 drbd_setbufsize(s_listen
, mdev
->net_conf
->sndbuf_size
,
671 mdev
->net_conf
->rcvbuf_size
);
673 what
= "bind before listen";
674 err
= s_listen
->ops
->bind(s_listen
,
675 (struct sockaddr
*) mdev
->net_conf
->my_addr
,
676 mdev
->net_conf
->my_addr_len
);
680 err
= drbd_accept(mdev
, &what
, s_listen
, &s_estab
);
684 sock_release(s_listen
);
686 if (err
!= -EAGAIN
&& err
!= -EINTR
&& err
!= -ERESTARTSYS
) {
687 dev_err(DEV
, "%s failed, err = %d\n", what
, err
);
688 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
696 static int drbd_send_fp(struct drbd_conf
*mdev
,
697 struct socket
*sock
, enum drbd_packets cmd
)
699 struct p_header80
*h
= &mdev
->data
.sbuf
.header
.h80
;
701 return _drbd_send_cmd(mdev
, sock
, cmd
, h
, sizeof(*h
), 0);
704 static enum drbd_packets
drbd_recv_fp(struct drbd_conf
*mdev
, struct socket
*sock
)
706 struct p_header80
*h
= &mdev
->data
.rbuf
.header
.h80
;
709 rr
= drbd_recv_short(mdev
, sock
, h
, sizeof(*h
), 0);
711 if (rr
== sizeof(*h
) && h
->magic
== BE_DRBD_MAGIC
)
712 return be16_to_cpu(h
->command
);
718 * drbd_socket_okay() - Free the socket if its connection is not okay
719 * @mdev: DRBD device.
720 * @sock: pointer to the pointer to the socket.
722 static int drbd_socket_okay(struct drbd_conf
*mdev
, struct socket
**sock
)
730 rr
= drbd_recv_short(mdev
, *sock
, tb
, 4, MSG_DONTWAIT
| MSG_PEEK
);
732 if (rr
> 0 || rr
== -EAGAIN
) {
743 * 1 yes, we have a valid connection
744 * 0 oops, did not work out, please try again
745 * -1 peer talks different language,
746 * no point in trying again, please go standalone.
747 * -2 We do not have a network config...
749 static int drbd_connect(struct drbd_conf
*mdev
)
751 struct socket
*s
, *sock
, *msock
;
754 D_ASSERT(!mdev
->data
.socket
);
756 if (drbd_request_state(mdev
, NS(conn
, C_WF_CONNECTION
)) < SS_SUCCESS
)
759 clear_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
766 /* 3 tries, this should take less than a second! */
767 s
= drbd_try_connect(mdev
);
770 /* give the other side time to call bind() & listen() */
771 __set_current_state(TASK_INTERRUPTIBLE
);
772 schedule_timeout(HZ
/ 10);
777 drbd_send_fp(mdev
, s
, P_HAND_SHAKE_S
);
781 drbd_send_fp(mdev
, s
, P_HAND_SHAKE_M
);
785 dev_err(DEV
, "Logic error in drbd_connect()\n");
786 goto out_release_sockets
;
791 __set_current_state(TASK_INTERRUPTIBLE
);
792 schedule_timeout(HZ
/ 10);
793 ok
= drbd_socket_okay(mdev
, &sock
);
794 ok
= drbd_socket_okay(mdev
, &msock
) && ok
;
800 s
= drbd_wait_for_connect(mdev
);
802 try = drbd_recv_fp(mdev
, s
);
803 drbd_socket_okay(mdev
, &sock
);
804 drbd_socket_okay(mdev
, &msock
);
808 dev_warn(DEV
, "initial packet S crossed\n");
815 dev_warn(DEV
, "initial packet M crossed\n");
819 set_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
822 dev_warn(DEV
, "Error receiving initial packet\n");
829 if (mdev
->state
.conn
<= C_DISCONNECTING
)
830 goto out_release_sockets
;
831 if (signal_pending(current
)) {
832 flush_signals(current
);
834 if (get_t_state(&mdev
->receiver
) == Exiting
)
835 goto out_release_sockets
;
839 ok
= drbd_socket_okay(mdev
, &sock
);
840 ok
= drbd_socket_okay(mdev
, &msock
) && ok
;
846 msock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
847 sock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
849 sock
->sk
->sk_allocation
= GFP_NOIO
;
850 msock
->sk
->sk_allocation
= GFP_NOIO
;
852 sock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE_BULK
;
853 msock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE
;
856 * sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
857 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
858 * first set it to the P_HAND_SHAKE timeout,
859 * which we set to 4x the configured ping_timeout. */
860 sock
->sk
->sk_sndtimeo
=
861 sock
->sk
->sk_rcvtimeo
= mdev
->net_conf
->ping_timeo
*4*HZ
/10;
863 msock
->sk
->sk_sndtimeo
= mdev
->net_conf
->timeout
*HZ
/10;
864 msock
->sk
->sk_rcvtimeo
= mdev
->net_conf
->ping_int
*HZ
;
866 /* we don't want delays.
867 * we use TCP_CORK where apropriate, though */
868 drbd_tcp_nodelay(sock
);
869 drbd_tcp_nodelay(msock
);
871 mdev
->data
.socket
= sock
;
872 mdev
->meta
.socket
= msock
;
873 mdev
->last_received
= jiffies
;
875 D_ASSERT(mdev
->asender
.task
== NULL
);
877 h
= drbd_do_handshake(mdev
);
881 if (mdev
->cram_hmac_tfm
) {
882 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
883 switch (drbd_do_auth(mdev
)) {
885 dev_err(DEV
, "Authentication of peer failed\n");
888 dev_err(DEV
, "Authentication of peer failed, trying again.\n");
893 if (drbd_request_state(mdev
, NS(conn
, C_WF_REPORT_PARAMS
)) < SS_SUCCESS
)
896 sock
->sk
->sk_sndtimeo
= mdev
->net_conf
->timeout
*HZ
/10;
897 sock
->sk
->sk_rcvtimeo
= MAX_SCHEDULE_TIMEOUT
;
899 atomic_set(&mdev
->packet_seq
, 0);
902 drbd_thread_start(&mdev
->asender
);
904 if (mdev
->agreed_pro_version
< 95 && get_ldev(mdev
)) {
905 drbd_setup_queue_param(mdev
, DRBD_MAX_SIZE_H80_PACKET
);
909 if (!drbd_send_protocol(mdev
))
911 drbd_send_sync_param(mdev
, &mdev
->sync_conf
);
912 drbd_send_sizes(mdev
, 0, 0);
913 drbd_send_uuids(mdev
);
914 drbd_send_state(mdev
);
915 clear_bit(USE_DEGR_WFC_T
, &mdev
->flags
);
916 clear_bit(RESIZE_PENDING
, &mdev
->flags
);
928 static int drbd_recv_header(struct drbd_conf
*mdev
, enum drbd_packets
*cmd
, unsigned int *packet_size
)
930 union p_header
*h
= &mdev
->data
.rbuf
.header
;
933 r
= drbd_recv(mdev
, h
, sizeof(*h
));
934 if (unlikely(r
!= sizeof(*h
))) {
935 dev_err(DEV
, "short read expecting header on sock: r=%d\n", r
);
939 if (likely(h
->h80
.magic
== BE_DRBD_MAGIC
)) {
940 *cmd
= be16_to_cpu(h
->h80
.command
);
941 *packet_size
= be16_to_cpu(h
->h80
.length
);
942 } else if (h
->h95
.magic
== BE_DRBD_MAGIC_BIG
) {
943 *cmd
= be16_to_cpu(h
->h95
.command
);
944 *packet_size
= be32_to_cpu(h
->h95
.length
);
946 dev_err(DEV
, "magic?? on data m: 0x%08x c: %d l: %d\n",
947 be32_to_cpu(h
->h80
.magic
),
948 be16_to_cpu(h
->h80
.command
),
949 be16_to_cpu(h
->h80
.length
));
952 mdev
->last_received
= jiffies
;
957 static void drbd_flush(struct drbd_conf
*mdev
)
961 if (mdev
->write_ordering
>= WO_bdev_flush
&& get_ldev(mdev
)) {
962 rv
= blkdev_issue_flush(mdev
->ldev
->backing_bdev
, GFP_KERNEL
,
965 dev_err(DEV
, "local disk flush failed with status %d\n", rv
);
966 /* would rather check on EOPNOTSUPP, but that is not reliable.
967 * don't try again for ANY return value != 0
968 * if (rv == -EOPNOTSUPP) */
969 drbd_bump_write_ordering(mdev
, WO_drain_io
);
976 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
977 * @mdev: DRBD device.
978 * @epoch: Epoch object.
981 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*mdev
,
982 struct drbd_epoch
*epoch
,
986 struct drbd_epoch
*next_epoch
;
987 enum finish_epoch rv
= FE_STILL_LIVE
;
989 spin_lock(&mdev
->epoch_lock
);
993 epoch_size
= atomic_read(&epoch
->epoch_size
);
995 switch (ev
& ~EV_CLEANUP
) {
997 atomic_dec(&epoch
->active
);
999 case EV_GOT_BARRIER_NR
:
1000 set_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
);
1002 case EV_BECAME_LAST
:
1007 if (epoch_size
!= 0 &&
1008 atomic_read(&epoch
->active
) == 0 &&
1009 test_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
)) {
1010 if (!(ev
& EV_CLEANUP
)) {
1011 spin_unlock(&mdev
->epoch_lock
);
1012 drbd_send_b_ack(mdev
, epoch
->barrier_nr
, epoch_size
);
1013 spin_lock(&mdev
->epoch_lock
);
1017 if (mdev
->current_epoch
!= epoch
) {
1018 next_epoch
= list_entry(epoch
->list
.next
, struct drbd_epoch
, list
);
1019 list_del(&epoch
->list
);
1020 ev
= EV_BECAME_LAST
| (ev
& EV_CLEANUP
);
1024 if (rv
== FE_STILL_LIVE
)
1028 atomic_set(&epoch
->epoch_size
, 0);
1029 /* atomic_set(&epoch->active, 0); is already zero */
1030 if (rv
== FE_STILL_LIVE
)
1032 wake_up(&mdev
->ee_wait
);
1042 spin_unlock(&mdev
->epoch_lock
);
1048 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1049 * @mdev: DRBD device.
1050 * @wo: Write ordering method to try.
1052 void drbd_bump_write_ordering(struct drbd_conf
*mdev
, enum write_ordering_e wo
) __must_hold(local
)
1054 enum write_ordering_e pwo
;
1055 static char *write_ordering_str
[] = {
1057 [WO_drain_io
] = "drain",
1058 [WO_bdev_flush
] = "flush",
1061 pwo
= mdev
->write_ordering
;
1063 if (wo
== WO_bdev_flush
&& mdev
->ldev
->dc
.no_disk_flush
)
1065 if (wo
== WO_drain_io
&& mdev
->ldev
->dc
.no_disk_drain
)
1067 mdev
->write_ordering
= wo
;
1068 if (pwo
!= mdev
->write_ordering
|| wo
== WO_bdev_flush
)
1069 dev_info(DEV
, "Method to ensure write ordering: %s\n", write_ordering_str
[mdev
->write_ordering
]);
1074 * @mdev: DRBD device.
1076 * @rw: flag field, see bio->bi_rw
1078 /* TODO allocate from our own bio_set. */
1079 int drbd_submit_ee(struct drbd_conf
*mdev
, struct drbd_epoch_entry
*e
,
1080 const unsigned rw
, const int fault_type
)
1082 struct bio
*bios
= NULL
;
1084 struct page
*page
= e
->pages
;
1085 sector_t sector
= e
->sector
;
1086 unsigned ds
= e
->size
;
1087 unsigned n_bios
= 0;
1088 unsigned nr_pages
= (ds
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
1090 /* In most cases, we will only need one bio. But in case the lower
1091 * level restrictions happen to be different at this offset on this
1092 * side than those of the sending peer, we may need to submit the
1093 * request in more than one bio. */
1095 bio
= bio_alloc(GFP_NOIO
, nr_pages
);
1097 dev_err(DEV
, "submit_ee: Allocation of a bio failed\n");
1100 /* > e->sector, unless this is the first bio */
1101 bio
->bi_sector
= sector
;
1102 bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1103 /* we special case some flags in the multi-bio case, see below
1106 bio
->bi_private
= e
;
1107 bio
->bi_end_io
= drbd_endio_sec
;
1109 bio
->bi_next
= bios
;
1113 page_chain_for_each(page
) {
1114 unsigned len
= min_t(unsigned, ds
, PAGE_SIZE
);
1115 if (!bio_add_page(bio
, page
, len
, 0)) {
1116 /* a single page must always be possible! */
1117 BUG_ON(bio
->bi_vcnt
== 0);
1124 D_ASSERT(page
== NULL
);
1127 atomic_set(&e
->pending_bios
, n_bios
);
1130 bios
= bios
->bi_next
;
1131 bio
->bi_next
= NULL
;
1133 /* strip off REQ_UNPLUG unless it is the last bio */
1135 bio
->bi_rw
&= ~REQ_UNPLUG
;
1137 drbd_generic_make_request(mdev
, fault_type
, bio
);
1144 bios
= bios
->bi_next
;
1150 static int receive_Barrier(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
1153 struct p_barrier
*p
= &mdev
->data
.rbuf
.barrier
;
1154 struct drbd_epoch
*epoch
;
1158 mdev
->current_epoch
->barrier_nr
= p
->barrier
;
1159 rv
= drbd_may_finish_epoch(mdev
, mdev
->current_epoch
, EV_GOT_BARRIER_NR
);
1161 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1162 * the activity log, which means it would not be resynced in case the
1163 * R_PRIMARY crashes now.
1164 * Therefore we must send the barrier_ack after the barrier request was
1166 switch (mdev
->write_ordering
) {
1168 if (rv
== FE_RECYCLED
)
1171 /* receiver context, in the writeout path of the other node.
1172 * avoid potential distributed deadlock */
1173 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1177 dev_warn(DEV
, "Allocation of an epoch failed, slowing down\n");
1182 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1185 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1186 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1191 epoch
= mdev
->current_epoch
;
1192 wait_event(mdev
->ee_wait
, atomic_read(&epoch
->epoch_size
) == 0);
1194 D_ASSERT(atomic_read(&epoch
->active
) == 0);
1195 D_ASSERT(epoch
->flags
== 0);
1199 dev_err(DEV
, "Strangeness in mdev->write_ordering %d\n", mdev
->write_ordering
);
1204 atomic_set(&epoch
->epoch_size
, 0);
1205 atomic_set(&epoch
->active
, 0);
1207 spin_lock(&mdev
->epoch_lock
);
1208 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1209 list_add(&epoch
->list
, &mdev
->current_epoch
->list
);
1210 mdev
->current_epoch
= epoch
;
1213 /* The current_epoch got recycled while we allocated this one... */
1216 spin_unlock(&mdev
->epoch_lock
);
1221 /* used from receive_RSDataReply (recv_resync_read)
1222 * and from receive_Data */
1223 static struct drbd_epoch_entry
*
1224 read_in_block(struct drbd_conf
*mdev
, u64 id
, sector_t sector
, int data_size
) __must_hold(local
)
1226 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1227 struct drbd_epoch_entry
*e
;
1230 void *dig_in
= mdev
->int_dig_in
;
1231 void *dig_vv
= mdev
->int_dig_vv
;
1232 unsigned long *data
;
1234 dgs
= (mdev
->agreed_pro_version
>= 87 && mdev
->integrity_r_tfm
) ?
1235 crypto_hash_digestsize(mdev
->integrity_r_tfm
) : 0;
1238 rr
= drbd_recv(mdev
, dig_in
, dgs
);
1240 dev_warn(DEV
, "short read receiving data digest: read %d expected %d\n",
1248 ERR_IF(data_size
& 0x1ff) return NULL
;
1249 ERR_IF(data_size
> DRBD_MAX_SEGMENT_SIZE
) return NULL
;
1251 /* even though we trust out peer,
1252 * we sometimes have to double check. */
1253 if (sector
+ (data_size
>>9) > capacity
) {
1254 dev_err(DEV
, "capacity: %llus < sector: %llus + size: %u\n",
1255 (unsigned long long)capacity
,
1256 (unsigned long long)sector
, data_size
);
1260 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1261 * "criss-cross" setup, that might cause write-out on some other DRBD,
1262 * which in turn might block on the other node at this very place. */
1263 e
= drbd_alloc_ee(mdev
, id
, sector
, data_size
, GFP_NOIO
);
1269 page_chain_for_each(page
) {
1270 unsigned len
= min_t(int, ds
, PAGE_SIZE
);
1272 rr
= drbd_recv(mdev
, data
, len
);
1273 if (FAULT_ACTIVE(mdev
, DRBD_FAULT_RECEIVE
)) {
1274 dev_err(DEV
, "Fault injection: Corrupting data on receive\n");
1275 data
[0] = data
[0] ^ (unsigned long)-1;
1279 drbd_free_ee(mdev
, e
);
1280 dev_warn(DEV
, "short read receiving data: read %d expected %d\n",
1288 drbd_csum_ee(mdev
, mdev
->integrity_r_tfm
, e
, dig_vv
);
1289 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1290 dev_err(DEV
, "Digest integrity check FAILED.\n");
1291 drbd_bcast_ee(mdev
, "digest failed",
1292 dgs
, dig_in
, dig_vv
, e
);
1293 drbd_free_ee(mdev
, e
);
1297 mdev
->recv_cnt
+= data_size
>>9;
1301 /* drbd_drain_block() just takes a data block
1302 * out of the socket input buffer, and discards it.
1304 static int drbd_drain_block(struct drbd_conf
*mdev
, int data_size
)
1313 page
= drbd_pp_alloc(mdev
, 1, 1);
1317 rr
= drbd_recv(mdev
, data
, min_t(int, data_size
, PAGE_SIZE
));
1318 if (rr
!= min_t(int, data_size
, PAGE_SIZE
)) {
1320 dev_warn(DEV
, "short read receiving data: read %d expected %d\n",
1321 rr
, min_t(int, data_size
, PAGE_SIZE
));
1327 drbd_pp_free(mdev
, page
, 0);
1331 static int recv_dless_read(struct drbd_conf
*mdev
, struct drbd_request
*req
,
1332 sector_t sector
, int data_size
)
1334 struct bio_vec
*bvec
;
1336 int dgs
, rr
, i
, expect
;
1337 void *dig_in
= mdev
->int_dig_in
;
1338 void *dig_vv
= mdev
->int_dig_vv
;
1340 dgs
= (mdev
->agreed_pro_version
>= 87 && mdev
->integrity_r_tfm
) ?
1341 crypto_hash_digestsize(mdev
->integrity_r_tfm
) : 0;
1344 rr
= drbd_recv(mdev
, dig_in
, dgs
);
1346 dev_warn(DEV
, "short read receiving data reply digest: read %d expected %d\n",
1354 /* optimistically update recv_cnt. if receiving fails below,
1355 * we disconnect anyways, and counters will be reset. */
1356 mdev
->recv_cnt
+= data_size
>>9;
1358 bio
= req
->master_bio
;
1359 D_ASSERT(sector
== bio
->bi_sector
);
1361 bio_for_each_segment(bvec
, bio
, i
) {
1362 expect
= min_t(int, data_size
, bvec
->bv_len
);
1363 rr
= drbd_recv(mdev
,
1364 kmap(bvec
->bv_page
)+bvec
->bv_offset
,
1366 kunmap(bvec
->bv_page
);
1368 dev_warn(DEV
, "short read receiving data reply: "
1369 "read %d expected %d\n",
1377 drbd_csum_bio(mdev
, mdev
->integrity_r_tfm
, bio
, dig_vv
);
1378 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1379 dev_err(DEV
, "Digest integrity check FAILED. Broken NICs?\n");
1384 D_ASSERT(data_size
== 0);
1388 /* e_end_resync_block() is called via
1389 * drbd_process_done_ee() by asender only */
1390 static int e_end_resync_block(struct drbd_conf
*mdev
, struct drbd_work
*w
, int unused
)
1392 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1393 sector_t sector
= e
->sector
;
1396 D_ASSERT(hlist_unhashed(&e
->colision
));
1398 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
1399 drbd_set_in_sync(mdev
, sector
, e
->size
);
1400 ok
= drbd_send_ack(mdev
, P_RS_WRITE_ACK
, e
);
1402 /* Record failure to sync */
1403 drbd_rs_failed_io(mdev
, sector
, e
->size
);
1405 ok
= drbd_send_ack(mdev
, P_NEG_ACK
, e
);
1412 static int recv_resync_read(struct drbd_conf
*mdev
, sector_t sector
, int data_size
) __releases(local
)
1414 struct drbd_epoch_entry
*e
;
1416 e
= read_in_block(mdev
, ID_SYNCER
, sector
, data_size
);
1420 dec_rs_pending(mdev
);
1423 /* corresponding dec_unacked() in e_end_resync_block()
1424 * respective _drbd_clear_done_ee */
1426 e
->w
.cb
= e_end_resync_block
;
1428 spin_lock_irq(&mdev
->req_lock
);
1429 list_add(&e
->w
.list
, &mdev
->sync_ee
);
1430 spin_unlock_irq(&mdev
->req_lock
);
1432 atomic_add(data_size
>> 9, &mdev
->rs_sect_ev
);
1433 if (drbd_submit_ee(mdev
, e
, WRITE
, DRBD_FAULT_RS_WR
) == 0)
1436 /* drbd_submit_ee currently fails for one reason only:
1437 * not being able to allocate enough bios.
1438 * Is dropping the connection going to help? */
1439 spin_lock_irq(&mdev
->req_lock
);
1440 list_del(&e
->w
.list
);
1441 spin_unlock_irq(&mdev
->req_lock
);
1443 drbd_free_ee(mdev
, e
);
1449 static int receive_DataReply(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
1451 struct drbd_request
*req
;
1454 struct p_data
*p
= &mdev
->data
.rbuf
.data
;
1456 sector
= be64_to_cpu(p
->sector
);
1458 spin_lock_irq(&mdev
->req_lock
);
1459 req
= _ar_id_to_req(mdev
, p
->block_id
, sector
);
1460 spin_unlock_irq(&mdev
->req_lock
);
1461 if (unlikely(!req
)) {
1462 dev_err(DEV
, "Got a corrupt block_id/sector pair(1).\n");
1466 /* hlist_del(&req->colision) is done in _req_may_be_done, to avoid
1467 * special casing it there for the various failure cases.
1468 * still no race with drbd_fail_pending_reads */
1469 ok
= recv_dless_read(mdev
, req
, sector
, data_size
);
1472 req_mod(req
, data_received
);
1473 /* else: nothing. handled from drbd_disconnect...
1474 * I don't think we may complete this just yet
1475 * in case we are "on-disconnect: freeze" */
1480 static int receive_RSDataReply(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
1484 struct p_data
*p
= &mdev
->data
.rbuf
.data
;
1486 sector
= be64_to_cpu(p
->sector
);
1487 D_ASSERT(p
->block_id
== ID_SYNCER
);
1489 if (get_ldev(mdev
)) {
1490 /* data is submitted to disk within recv_resync_read.
1491 * corresponding put_ldev done below on error,
1492 * or in drbd_endio_write_sec. */
1493 ok
= recv_resync_read(mdev
, sector
, data_size
);
1495 if (__ratelimit(&drbd_ratelimit_state
))
1496 dev_err(DEV
, "Can not write resync data to local disk.\n");
1498 ok
= drbd_drain_block(mdev
, data_size
);
1500 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, data_size
);
1503 atomic_add(data_size
>> 9, &mdev
->rs_sect_in
);
1508 /* e_end_block() is called via drbd_process_done_ee().
1509 * this means this function only runs in the asender thread
1511 static int e_end_block(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1513 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1514 sector_t sector
= e
->sector
;
1517 if (mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
) {
1518 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
1519 pcmd
= (mdev
->state
.conn
>= C_SYNC_SOURCE
&&
1520 mdev
->state
.conn
<= C_PAUSED_SYNC_T
&&
1521 e
->flags
& EE_MAY_SET_IN_SYNC
) ?
1522 P_RS_WRITE_ACK
: P_WRITE_ACK
;
1523 ok
&= drbd_send_ack(mdev
, pcmd
, e
);
1524 if (pcmd
== P_RS_WRITE_ACK
)
1525 drbd_set_in_sync(mdev
, sector
, e
->size
);
1527 ok
= drbd_send_ack(mdev
, P_NEG_ACK
, e
);
1528 /* we expect it to be marked out of sync anyways...
1529 * maybe assert this? */
1533 /* we delete from the conflict detection hash _after_ we sent out the
1534 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1535 if (mdev
->net_conf
->two_primaries
) {
1536 spin_lock_irq(&mdev
->req_lock
);
1537 D_ASSERT(!hlist_unhashed(&e
->colision
));
1538 hlist_del_init(&e
->colision
);
1539 spin_unlock_irq(&mdev
->req_lock
);
1541 D_ASSERT(hlist_unhashed(&e
->colision
));
1544 drbd_may_finish_epoch(mdev
, e
->epoch
, EV_PUT
+ (cancel
? EV_CLEANUP
: 0));
1549 static int e_send_discard_ack(struct drbd_conf
*mdev
, struct drbd_work
*w
, int unused
)
1551 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1554 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
1555 ok
= drbd_send_ack(mdev
, P_DISCARD_ACK
, e
);
1557 spin_lock_irq(&mdev
->req_lock
);
1558 D_ASSERT(!hlist_unhashed(&e
->colision
));
1559 hlist_del_init(&e
->colision
);
1560 spin_unlock_irq(&mdev
->req_lock
);
1567 /* Called from receive_Data.
1568 * Synchronize packets on sock with packets on msock.
1570 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1571 * packet traveling on msock, they are still processed in the order they have
1574 * Note: we don't care for Ack packets overtaking P_DATA packets.
1576 * In case packet_seq is larger than mdev->peer_seq number, there are
1577 * outstanding packets on the msock. We wait for them to arrive.
1578 * In case we are the logically next packet, we update mdev->peer_seq
1579 * ourselves. Correctly handles 32bit wrap around.
1581 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1582 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1583 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1584 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1586 * returns 0 if we may process the packet,
1587 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1588 static int drbd_wait_peer_seq(struct drbd_conf
*mdev
, const u32 packet_seq
)
1594 spin_lock(&mdev
->peer_seq_lock
);
1596 prepare_to_wait(&mdev
->seq_wait
, &wait
, TASK_INTERRUPTIBLE
);
1597 if (seq_le(packet_seq
, mdev
->peer_seq
+1))
1599 if (signal_pending(current
)) {
1603 p_seq
= mdev
->peer_seq
;
1604 spin_unlock(&mdev
->peer_seq_lock
);
1605 timeout
= schedule_timeout(30*HZ
);
1606 spin_lock(&mdev
->peer_seq_lock
);
1607 if (timeout
== 0 && p_seq
== mdev
->peer_seq
) {
1609 dev_err(DEV
, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n");
1613 finish_wait(&mdev
->seq_wait
, &wait
);
1614 if (mdev
->peer_seq
+1 == packet_seq
)
1616 spin_unlock(&mdev
->peer_seq_lock
);
1620 static unsigned long write_flags_to_bio(struct drbd_conf
*mdev
, u32 dpf
)
1622 if (mdev
->agreed_pro_version
>= 95)
1623 return (dpf
& DP_RW_SYNC
? REQ_SYNC
: 0) |
1624 (dpf
& DP_UNPLUG
? REQ_UNPLUG
: 0) |
1625 (dpf
& DP_FUA
? REQ_FUA
: 0) |
1626 (dpf
& DP_FLUSH
? REQ_FUA
: 0) |
1627 (dpf
& DP_DISCARD
? REQ_DISCARD
: 0);
1629 return dpf
& DP_RW_SYNC
? (REQ_SYNC
| REQ_UNPLUG
) : 0;
1632 /* mirrored write */
1633 static int receive_Data(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
1636 struct drbd_epoch_entry
*e
;
1637 struct p_data
*p
= &mdev
->data
.rbuf
.data
;
1641 if (!get_ldev(mdev
)) {
1642 if (__ratelimit(&drbd_ratelimit_state
))
1643 dev_err(DEV
, "Can not write mirrored data block "
1644 "to local disk.\n");
1645 spin_lock(&mdev
->peer_seq_lock
);
1646 if (mdev
->peer_seq
+1 == be32_to_cpu(p
->seq_num
))
1648 spin_unlock(&mdev
->peer_seq_lock
);
1650 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, data_size
);
1651 atomic_inc(&mdev
->current_epoch
->epoch_size
);
1652 return drbd_drain_block(mdev
, data_size
);
1655 /* get_ldev(mdev) successful.
1656 * Corresponding put_ldev done either below (on various errors),
1657 * or in drbd_endio_write_sec, if we successfully submit the data at
1658 * the end of this function. */
1660 sector
= be64_to_cpu(p
->sector
);
1661 e
= read_in_block(mdev
, p
->block_id
, sector
, data_size
);
1667 e
->w
.cb
= e_end_block
;
1669 spin_lock(&mdev
->epoch_lock
);
1670 e
->epoch
= mdev
->current_epoch
;
1671 atomic_inc(&e
->epoch
->epoch_size
);
1672 atomic_inc(&e
->epoch
->active
);
1673 spin_unlock(&mdev
->epoch_lock
);
1675 dp_flags
= be32_to_cpu(p
->dp_flags
);
1676 rw
|= write_flags_to_bio(mdev
, dp_flags
);
1678 if (dp_flags
& DP_MAY_SET_IN_SYNC
)
1679 e
->flags
|= EE_MAY_SET_IN_SYNC
;
1681 /* I'm the receiver, I do hold a net_cnt reference. */
1682 if (!mdev
->net_conf
->two_primaries
) {
1683 spin_lock_irq(&mdev
->req_lock
);
1685 /* don't get the req_lock yet,
1686 * we may sleep in drbd_wait_peer_seq */
1687 const int size
= e
->size
;
1688 const int discard
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
1690 struct drbd_request
*i
;
1691 struct hlist_node
*n
;
1692 struct hlist_head
*slot
;
1695 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
1696 BUG_ON(mdev
->ee_hash
== NULL
);
1697 BUG_ON(mdev
->tl_hash
== NULL
);
1699 /* conflict detection and handling:
1700 * 1. wait on the sequence number,
1701 * in case this data packet overtook ACK packets.
1702 * 2. check our hash tables for conflicting requests.
1703 * we only need to walk the tl_hash, since an ee can not
1704 * have a conflict with an other ee: on the submitting
1705 * node, the corresponding req had already been conflicting,
1706 * and a conflicting req is never sent.
1708 * Note: for two_primaries, we are protocol C,
1709 * so there cannot be any request that is DONE
1710 * but still on the transfer log.
1712 * unconditionally add to the ee_hash.
1714 * if no conflicting request is found:
1717 * if any conflicting request is found
1718 * that has not yet been acked,
1719 * AND I have the "discard concurrent writes" flag:
1720 * queue (via done_ee) the P_DISCARD_ACK; OUT.
1722 * if any conflicting request is found:
1723 * block the receiver, waiting on misc_wait
1724 * until no more conflicting requests are there,
1725 * or we get interrupted (disconnect).
1727 * we do not just write after local io completion of those
1728 * requests, but only after req is done completely, i.e.
1729 * we wait for the P_DISCARD_ACK to arrive!
1731 * then proceed normally, i.e. submit.
1733 if (drbd_wait_peer_seq(mdev
, be32_to_cpu(p
->seq_num
)))
1734 goto out_interrupted
;
1736 spin_lock_irq(&mdev
->req_lock
);
1738 hlist_add_head(&e
->colision
, ee_hash_slot(mdev
, sector
));
1740 #define OVERLAPS overlaps(i->sector, i->size, sector, size)
1741 slot
= tl_hash_slot(mdev
, sector
);
1744 int have_unacked
= 0;
1745 int have_conflict
= 0;
1746 prepare_to_wait(&mdev
->misc_wait
, &wait
,
1747 TASK_INTERRUPTIBLE
);
1748 hlist_for_each_entry(i
, n
, slot
, colision
) {
1750 /* only ALERT on first iteration,
1751 * we may be woken up early... */
1753 dev_alert(DEV
, "%s[%u] Concurrent local write detected!"
1754 " new: %llus +%u; pending: %llus +%u\n",
1755 current
->comm
, current
->pid
,
1756 (unsigned long long)sector
, size
,
1757 (unsigned long long)i
->sector
, i
->size
);
1758 if (i
->rq_state
& RQ_NET_PENDING
)
1767 /* Discard Ack only for the _first_ iteration */
1768 if (first
&& discard
&& have_unacked
) {
1769 dev_alert(DEV
, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n",
1770 (unsigned long long)sector
);
1772 e
->w
.cb
= e_send_discard_ack
;
1773 list_add_tail(&e
->w
.list
, &mdev
->done_ee
);
1775 spin_unlock_irq(&mdev
->req_lock
);
1777 /* we could probably send that P_DISCARD_ACK ourselves,
1778 * but I don't like the receiver using the msock */
1782 finish_wait(&mdev
->misc_wait
, &wait
);
1786 if (signal_pending(current
)) {
1787 hlist_del_init(&e
->colision
);
1789 spin_unlock_irq(&mdev
->req_lock
);
1791 finish_wait(&mdev
->misc_wait
, &wait
);
1792 goto out_interrupted
;
1795 spin_unlock_irq(&mdev
->req_lock
);
1798 dev_alert(DEV
, "Concurrent write! [W AFTERWARDS] "
1799 "sec=%llus\n", (unsigned long long)sector
);
1800 } else if (discard
) {
1801 /* we had none on the first iteration.
1802 * there must be none now. */
1803 D_ASSERT(have_unacked
== 0);
1806 spin_lock_irq(&mdev
->req_lock
);
1808 finish_wait(&mdev
->misc_wait
, &wait
);
1811 list_add(&e
->w
.list
, &mdev
->active_ee
);
1812 spin_unlock_irq(&mdev
->req_lock
);
1814 switch (mdev
->net_conf
->wire_protocol
) {
1817 /* corresponding dec_unacked() in e_end_block()
1818 * respective _drbd_clear_done_ee */
1821 /* I really don't like it that the receiver thread
1822 * sends on the msock, but anyways */
1823 drbd_send_ack(mdev
, P_RECV_ACK
, e
);
1830 if (mdev
->state
.pdsk
< D_INCONSISTENT
) {
1831 /* In case we have the only disk of the cluster, */
1832 drbd_set_out_of_sync(mdev
, e
->sector
, e
->size
);
1833 e
->flags
|= EE_CALL_AL_COMPLETE_IO
;
1834 e
->flags
&= ~EE_MAY_SET_IN_SYNC
;
1835 drbd_al_begin_io(mdev
, e
->sector
);
1838 if (drbd_submit_ee(mdev
, e
, rw
, DRBD_FAULT_DT_WR
) == 0)
1841 /* drbd_submit_ee currently fails for one reason only:
1842 * not being able to allocate enough bios.
1843 * Is dropping the connection going to help? */
1844 spin_lock_irq(&mdev
->req_lock
);
1845 list_del(&e
->w
.list
);
1846 hlist_del_init(&e
->colision
);
1847 spin_unlock_irq(&mdev
->req_lock
);
1848 if (e
->flags
& EE_CALL_AL_COMPLETE_IO
)
1849 drbd_al_complete_io(mdev
, e
->sector
);
1852 /* yes, the epoch_size now is imbalanced.
1853 * but we drop the connection anyways, so we don't have a chance to
1854 * receive a barrier... atomic_inc(&mdev->epoch_size); */
1856 drbd_free_ee(mdev
, e
);
1860 /* We may throttle resync, if the lower device seems to be busy,
1861 * and current sync rate is above c_min_rate.
1863 * To decide whether or not the lower device is busy, we use a scheme similar
1864 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
1865 * (more than 64 sectors) of activity we cannot account for with our own resync
1866 * activity, it obviously is "busy".
1868 * The current sync rate used here uses only the most recent two step marks,
1869 * to have a short time average so we can react faster.
1871 int drbd_rs_should_slow_down(struct drbd_conf
*mdev
)
1873 struct gendisk
*disk
= mdev
->ldev
->backing_bdev
->bd_contains
->bd_disk
;
1874 unsigned long db
, dt
, dbdt
;
1878 /* feature disabled? */
1879 if (mdev
->sync_conf
.c_min_rate
== 0)
1882 curr_events
= (int)part_stat_read(&disk
->part0
, sectors
[0]) +
1883 (int)part_stat_read(&disk
->part0
, sectors
[1]) -
1884 atomic_read(&mdev
->rs_sect_ev
);
1885 if (!mdev
->rs_last_events
|| curr_events
- mdev
->rs_last_events
> 64) {
1886 unsigned long rs_left
;
1889 mdev
->rs_last_events
= curr_events
;
1891 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
1893 i
= (mdev
->rs_last_mark
+ DRBD_SYNC_MARKS
-2) % DRBD_SYNC_MARKS
;
1894 rs_left
= drbd_bm_total_weight(mdev
) - mdev
->rs_failed
;
1896 dt
= ((long)jiffies
- (long)mdev
->rs_mark_time
[i
]) / HZ
;
1899 db
= mdev
->rs_mark_left
[i
] - rs_left
;
1900 dbdt
= Bit2KB(db
/dt
);
1902 if (dbdt
> mdev
->sync_conf
.c_min_rate
)
1909 static int receive_DataRequest(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int digest_size
)
1912 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1913 struct drbd_epoch_entry
*e
;
1914 struct digest_info
*di
= NULL
;
1916 unsigned int fault_type
;
1917 struct p_block_req
*p
= &mdev
->data
.rbuf
.block_req
;
1919 sector
= be64_to_cpu(p
->sector
);
1920 size
= be32_to_cpu(p
->blksize
);
1922 if (size
<= 0 || (size
& 0x1ff) != 0 || size
> DRBD_MAX_SEGMENT_SIZE
) {
1923 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
1924 (unsigned long long)sector
, size
);
1927 if (sector
+ (size
>>9) > capacity
) {
1928 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
1929 (unsigned long long)sector
, size
);
1933 if (!get_ldev_if_state(mdev
, D_UP_TO_DATE
)) {
1936 case P_DATA_REQUEST
:
1937 drbd_send_ack_rp(mdev
, P_NEG_DREPLY
, p
);
1939 case P_RS_DATA_REQUEST
:
1940 case P_CSUM_RS_REQUEST
:
1942 drbd_send_ack_rp(mdev
, P_NEG_RS_DREPLY
, p
);
1946 dec_rs_pending(mdev
);
1947 drbd_send_ack_ex(mdev
, P_OV_RESULT
, sector
, size
, ID_IN_SYNC
);
1950 dev_err(DEV
, "unexpected command (%s) in receive_DataRequest\n",
1953 if (verb
&& __ratelimit(&drbd_ratelimit_state
))
1954 dev_err(DEV
, "Can not satisfy peer's read request, "
1955 "no local data.\n");
1957 /* drain possibly payload */
1958 return drbd_drain_block(mdev
, digest_size
);
1961 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1962 * "criss-cross" setup, that might cause write-out on some other DRBD,
1963 * which in turn might block on the other node at this very place. */
1964 e
= drbd_alloc_ee(mdev
, p
->block_id
, sector
, size
, GFP_NOIO
);
1971 case P_DATA_REQUEST
:
1972 e
->w
.cb
= w_e_end_data_req
;
1973 fault_type
= DRBD_FAULT_DT_RD
;
1974 /* application IO, don't drbd_rs_begin_io */
1977 case P_RS_DATA_REQUEST
:
1978 e
->w
.cb
= w_e_end_rsdata_req
;
1979 fault_type
= DRBD_FAULT_RS_RD
;
1983 case P_CSUM_RS_REQUEST
:
1984 fault_type
= DRBD_FAULT_RS_RD
;
1985 di
= kmalloc(sizeof(*di
) + digest_size
, GFP_NOIO
);
1989 di
->digest_size
= digest_size
;
1990 di
->digest
= (((char *)di
)+sizeof(struct digest_info
));
1993 e
->flags
|= EE_HAS_DIGEST
;
1995 if (drbd_recv(mdev
, di
->digest
, digest_size
) != digest_size
)
1998 if (cmd
== P_CSUM_RS_REQUEST
) {
1999 D_ASSERT(mdev
->agreed_pro_version
>= 89);
2000 e
->w
.cb
= w_e_end_csum_rs_req
;
2001 } else if (cmd
== P_OV_REPLY
) {
2002 e
->w
.cb
= w_e_end_ov_reply
;
2003 dec_rs_pending(mdev
);
2004 /* drbd_rs_begin_io done when we sent this request,
2005 * but accounting still needs to be done. */
2006 goto submit_for_resync
;
2011 if (mdev
->ov_start_sector
== ~(sector_t
)0 &&
2012 mdev
->agreed_pro_version
>= 90) {
2013 mdev
->ov_start_sector
= sector
;
2014 mdev
->ov_position
= sector
;
2015 mdev
->ov_left
= mdev
->rs_total
- BM_SECT_TO_BIT(sector
);
2016 dev_info(DEV
, "Online Verify start sector: %llu\n",
2017 (unsigned long long)sector
);
2019 e
->w
.cb
= w_e_end_ov_req
;
2020 fault_type
= DRBD_FAULT_RS_RD
;
2024 dev_err(DEV
, "unexpected command (%s) in receive_DataRequest\n",
2026 fault_type
= DRBD_FAULT_MAX
;
2030 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2031 * wrt the receiver, but it is not as straightforward as it may seem.
2032 * Various places in the resync start and stop logic assume resync
2033 * requests are processed in order, requeuing this on the worker thread
2034 * introduces a bunch of new code for synchronization between threads.
2036 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2037 * "forever", throttling after drbd_rs_begin_io will lock that extent
2038 * for application writes for the same time. For now, just throttle
2039 * here, where the rest of the code expects the receiver to sleep for
2043 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2044 * this defers syncer requests for some time, before letting at least
2045 * on request through. The resync controller on the receiving side
2046 * will adapt to the incoming rate accordingly.
2048 * We cannot throttle here if remote is Primary/SyncTarget:
2049 * we would also throttle its application reads.
2050 * In that case, throttling is done on the SyncTarget only.
2052 if (mdev
->state
.peer
!= R_PRIMARY
&& drbd_rs_should_slow_down(mdev
))
2054 if (drbd_rs_begin_io(mdev
, e
->sector
))
2058 atomic_add(size
>> 9, &mdev
->rs_sect_ev
);
2062 spin_lock_irq(&mdev
->req_lock
);
2063 list_add_tail(&e
->w
.list
, &mdev
->read_ee
);
2064 spin_unlock_irq(&mdev
->req_lock
);
2066 if (drbd_submit_ee(mdev
, e
, READ
, fault_type
) == 0)
2069 /* drbd_submit_ee currently fails for one reason only:
2070 * not being able to allocate enough bios.
2071 * Is dropping the connection going to help? */
2072 spin_lock_irq(&mdev
->req_lock
);
2073 list_del(&e
->w
.list
);
2074 spin_unlock_irq(&mdev
->req_lock
);
2075 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2079 drbd_free_ee(mdev
, e
);
2083 static int drbd_asb_recover_0p(struct drbd_conf
*mdev
) __must_hold(local
)
2085 int self
, peer
, rv
= -100;
2086 unsigned long ch_self
, ch_peer
;
2088 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2089 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2091 ch_peer
= mdev
->p_uuid
[UI_SIZE
];
2092 ch_self
= mdev
->comm_bm_set
;
2094 switch (mdev
->net_conf
->after_sb_0p
) {
2096 case ASB_DISCARD_SECONDARY
:
2097 case ASB_CALL_HELPER
:
2098 dev_err(DEV
, "Configuration error.\n");
2100 case ASB_DISCONNECT
:
2102 case ASB_DISCARD_YOUNGER_PRI
:
2103 if (self
== 0 && peer
== 1) {
2107 if (self
== 1 && peer
== 0) {
2111 /* Else fall through to one of the other strategies... */
2112 case ASB_DISCARD_OLDER_PRI
:
2113 if (self
== 0 && peer
== 1) {
2117 if (self
== 1 && peer
== 0) {
2121 /* Else fall through to one of the other strategies... */
2122 dev_warn(DEV
, "Discard younger/older primary did not find a decision\n"
2123 "Using discard-least-changes instead\n");
2124 case ASB_DISCARD_ZERO_CHG
:
2125 if (ch_peer
== 0 && ch_self
== 0) {
2126 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
)
2130 if (ch_peer
== 0) { rv
= 1; break; }
2131 if (ch_self
== 0) { rv
= -1; break; }
2133 if (mdev
->net_conf
->after_sb_0p
== ASB_DISCARD_ZERO_CHG
)
2135 case ASB_DISCARD_LEAST_CHG
:
2136 if (ch_self
< ch_peer
)
2138 else if (ch_self
> ch_peer
)
2140 else /* ( ch_self == ch_peer ) */
2141 /* Well, then use something else. */
2142 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
)
2145 case ASB_DISCARD_LOCAL
:
2148 case ASB_DISCARD_REMOTE
:
2155 static int drbd_asb_recover_1p(struct drbd_conf
*mdev
) __must_hold(local
)
2157 int self
, peer
, hg
, rv
= -100;
2159 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2160 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2162 switch (mdev
->net_conf
->after_sb_1p
) {
2163 case ASB_DISCARD_YOUNGER_PRI
:
2164 case ASB_DISCARD_OLDER_PRI
:
2165 case ASB_DISCARD_LEAST_CHG
:
2166 case ASB_DISCARD_LOCAL
:
2167 case ASB_DISCARD_REMOTE
:
2168 dev_err(DEV
, "Configuration error.\n");
2170 case ASB_DISCONNECT
:
2173 hg
= drbd_asb_recover_0p(mdev
);
2174 if (hg
== -1 && mdev
->state
.role
== R_SECONDARY
)
2176 if (hg
== 1 && mdev
->state
.role
== R_PRIMARY
)
2180 rv
= drbd_asb_recover_0p(mdev
);
2182 case ASB_DISCARD_SECONDARY
:
2183 return mdev
->state
.role
== R_PRIMARY
? 1 : -1;
2184 case ASB_CALL_HELPER
:
2185 hg
= drbd_asb_recover_0p(mdev
);
2186 if (hg
== -1 && mdev
->state
.role
== R_PRIMARY
) {
2187 self
= drbd_set_role(mdev
, R_SECONDARY
, 0);
2188 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2189 * we might be here in C_WF_REPORT_PARAMS which is transient.
2190 * we do not need to wait for the after state change work either. */
2191 self
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2192 if (self
!= SS_SUCCESS
) {
2193 drbd_khelper(mdev
, "pri-lost-after-sb");
2195 dev_warn(DEV
, "Successfully gave up primary role.\n");
2205 static int drbd_asb_recover_2p(struct drbd_conf
*mdev
) __must_hold(local
)
2207 int self
, peer
, hg
, rv
= -100;
2209 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2210 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2212 switch (mdev
->net_conf
->after_sb_2p
) {
2213 case ASB_DISCARD_YOUNGER_PRI
:
2214 case ASB_DISCARD_OLDER_PRI
:
2215 case ASB_DISCARD_LEAST_CHG
:
2216 case ASB_DISCARD_LOCAL
:
2217 case ASB_DISCARD_REMOTE
:
2219 case ASB_DISCARD_SECONDARY
:
2220 dev_err(DEV
, "Configuration error.\n");
2223 rv
= drbd_asb_recover_0p(mdev
);
2225 case ASB_DISCONNECT
:
2227 case ASB_CALL_HELPER
:
2228 hg
= drbd_asb_recover_0p(mdev
);
2230 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2231 * we might be here in C_WF_REPORT_PARAMS which is transient.
2232 * we do not need to wait for the after state change work either. */
2233 self
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2234 if (self
!= SS_SUCCESS
) {
2235 drbd_khelper(mdev
, "pri-lost-after-sb");
2237 dev_warn(DEV
, "Successfully gave up primary role.\n");
2247 static void drbd_uuid_dump(struct drbd_conf
*mdev
, char *text
, u64
*uuid
,
2248 u64 bits
, u64 flags
)
2251 dev_info(DEV
, "%s uuid info vanished while I was looking!\n", text
);
2254 dev_info(DEV
, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2256 (unsigned long long)uuid
[UI_CURRENT
],
2257 (unsigned long long)uuid
[UI_BITMAP
],
2258 (unsigned long long)uuid
[UI_HISTORY_START
],
2259 (unsigned long long)uuid
[UI_HISTORY_END
],
2260 (unsigned long long)bits
,
2261 (unsigned long long)flags
);
2265 100 after split brain try auto recover
2266 2 C_SYNC_SOURCE set BitMap
2267 1 C_SYNC_SOURCE use BitMap
2269 -1 C_SYNC_TARGET use BitMap
2270 -2 C_SYNC_TARGET set BitMap
2271 -100 after split brain, disconnect
2272 -1000 unrelated data
2274 static int drbd_uuid_compare(struct drbd_conf
*mdev
, int *rule_nr
) __must_hold(local
)
2279 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2280 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2283 if (self
== UUID_JUST_CREATED
&& peer
== UUID_JUST_CREATED
)
2287 if ((self
== UUID_JUST_CREATED
|| self
== (u64
)0) &&
2288 peer
!= UUID_JUST_CREATED
)
2292 if (self
!= UUID_JUST_CREATED
&&
2293 (peer
== UUID_JUST_CREATED
|| peer
== (u64
)0))
2297 int rct
, dc
; /* roles at crash time */
2299 if (mdev
->p_uuid
[UI_BITMAP
] == (u64
)0 && mdev
->ldev
->md
.uuid
[UI_BITMAP
] != (u64
)0) {
2301 if (mdev
->agreed_pro_version
< 91)
2304 if ((mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) &&
2305 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1))) {
2306 dev_info(DEV
, "was SyncSource, missed the resync finished event, corrected myself:\n");
2307 drbd_uuid_set_bm(mdev
, 0UL);
2309 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2310 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2313 dev_info(DEV
, "was SyncSource (peer failed to write sync_uuid)\n");
2320 if (mdev
->ldev
->md
.uuid
[UI_BITMAP
] == (u64
)0 && mdev
->p_uuid
[UI_BITMAP
] != (u64
)0) {
2322 if (mdev
->agreed_pro_version
< 91)
2325 if ((mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1)) &&
2326 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1))) {
2327 dev_info(DEV
, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2329 mdev
->p_uuid
[UI_HISTORY_START
+ 1] = mdev
->p_uuid
[UI_HISTORY_START
];
2330 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_BITMAP
];
2331 mdev
->p_uuid
[UI_BITMAP
] = 0UL;
2333 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2336 dev_info(DEV
, "was SyncTarget (failed to write sync_uuid)\n");
2343 /* Common power [off|failure] */
2344 rct
= (test_bit(CRASHED_PRIMARY
, &mdev
->flags
) ? 1 : 0) +
2345 (mdev
->p_uuid
[UI_FLAGS
] & 2);
2346 /* lowest bit is set when we were primary,
2347 * next bit (weight 2) is set when peer was primary */
2351 case 0: /* !self_pri && !peer_pri */ return 0;
2352 case 1: /* self_pri && !peer_pri */ return 1;
2353 case 2: /* !self_pri && peer_pri */ return -1;
2354 case 3: /* self_pri && peer_pri */
2355 dc
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
2361 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2366 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2368 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2369 peer
= mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1);
2371 /* The last P_SYNC_UUID did not get though. Undo the last start of
2372 resync as sync source modifications of the peer's UUIDs. */
2374 if (mdev
->agreed_pro_version
< 91)
2377 mdev
->p_uuid
[UI_BITMAP
] = mdev
->p_uuid
[UI_HISTORY_START
];
2378 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_HISTORY_START
+ 1];
2384 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2385 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2386 peer
= mdev
->p_uuid
[i
] & ~((u64
)1);
2392 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2393 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2398 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2400 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1);
2401 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2403 /* The last P_SYNC_UUID did not get though. Undo the last start of
2404 resync as sync source modifications of our UUIDs. */
2406 if (mdev
->agreed_pro_version
< 91)
2409 _drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
]);
2410 _drbd_uuid_set(mdev
, UI_HISTORY_START
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1]);
2412 dev_info(DEV
, "Undid last start of resync:\n");
2414 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2415 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2423 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2424 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2425 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2431 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2432 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2433 if (self
== peer
&& self
!= ((u64
)0))
2437 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2438 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2439 for (j
= UI_HISTORY_START
; j
<= UI_HISTORY_END
; j
++) {
2440 peer
= mdev
->p_uuid
[j
] & ~((u64
)1);
2449 /* drbd_sync_handshake() returns the new conn state on success, or
2450 CONN_MASK (-1) on failure.
2452 static enum drbd_conns
drbd_sync_handshake(struct drbd_conf
*mdev
, enum drbd_role peer_role
,
2453 enum drbd_disk_state peer_disk
) __must_hold(local
)
2456 enum drbd_conns rv
= C_MASK
;
2457 enum drbd_disk_state mydisk
;
2459 mydisk
= mdev
->state
.disk
;
2460 if (mydisk
== D_NEGOTIATING
)
2461 mydisk
= mdev
->new_state_tmp
.disk
;
2463 dev_info(DEV
, "drbd_sync_handshake:\n");
2464 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
, mdev
->comm_bm_set
, 0);
2465 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
,
2466 mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2468 hg
= drbd_uuid_compare(mdev
, &rule_nr
);
2470 dev_info(DEV
, "uuid_compare()=%d by rule %d\n", hg
, rule_nr
);
2473 dev_alert(DEV
, "Unrelated data, aborting!\n");
2477 dev_alert(DEV
, "To resolve this both sides have to support at least protocol\n");
2481 if ((mydisk
== D_INCONSISTENT
&& peer_disk
> D_INCONSISTENT
) ||
2482 (peer_disk
== D_INCONSISTENT
&& mydisk
> D_INCONSISTENT
)) {
2483 int f
= (hg
== -100) || abs(hg
) == 2;
2484 hg
= mydisk
> D_INCONSISTENT
? 1 : -1;
2487 dev_info(DEV
, "Becoming sync %s due to disk states.\n",
2488 hg
> 0 ? "source" : "target");
2492 drbd_khelper(mdev
, "initial-split-brain");
2494 if (hg
== 100 || (hg
== -100 && mdev
->net_conf
->always_asbp
)) {
2495 int pcount
= (mdev
->state
.role
== R_PRIMARY
)
2496 + (peer_role
== R_PRIMARY
);
2497 int forced
= (hg
== -100);
2501 hg
= drbd_asb_recover_0p(mdev
);
2504 hg
= drbd_asb_recover_1p(mdev
);
2507 hg
= drbd_asb_recover_2p(mdev
);
2510 if (abs(hg
) < 100) {
2511 dev_warn(DEV
, "Split-Brain detected, %d primaries, "
2512 "automatically solved. Sync from %s node\n",
2513 pcount
, (hg
< 0) ? "peer" : "this");
2515 dev_warn(DEV
, "Doing a full sync, since"
2516 " UUIDs where ambiguous.\n");
2523 if (mdev
->net_conf
->want_lose
&& !(mdev
->p_uuid
[UI_FLAGS
]&1))
2525 if (!mdev
->net_conf
->want_lose
&& (mdev
->p_uuid
[UI_FLAGS
]&1))
2529 dev_warn(DEV
, "Split-Brain detected, manually solved. "
2530 "Sync from %s node\n",
2531 (hg
< 0) ? "peer" : "this");
2535 /* FIXME this log message is not correct if we end up here
2536 * after an attempted attach on a diskless node.
2537 * We just refuse to attach -- well, we drop the "connection"
2538 * to that disk, in a way... */
2539 dev_alert(DEV
, "Split-Brain detected but unresolved, dropping connection!\n");
2540 drbd_khelper(mdev
, "split-brain");
2544 if (hg
> 0 && mydisk
<= D_INCONSISTENT
) {
2545 dev_err(DEV
, "I shall become SyncSource, but I am inconsistent!\n");
2549 if (hg
< 0 && /* by intention we do not use mydisk here. */
2550 mdev
->state
.role
== R_PRIMARY
&& mdev
->state
.disk
>= D_CONSISTENT
) {
2551 switch (mdev
->net_conf
->rr_conflict
) {
2552 case ASB_CALL_HELPER
:
2553 drbd_khelper(mdev
, "pri-lost");
2555 case ASB_DISCONNECT
:
2556 dev_err(DEV
, "I shall become SyncTarget, but I am primary!\n");
2559 dev_warn(DEV
, "Becoming SyncTarget, violating the stable-data"
2564 if (mdev
->net_conf
->dry_run
|| test_bit(CONN_DRY_RUN
, &mdev
->flags
)) {
2566 dev_info(DEV
, "dry-run connect: No resync, would become Connected immediately.\n");
2568 dev_info(DEV
, "dry-run connect: Would become %s, doing a %s resync.",
2569 drbd_conn_str(hg
> 0 ? C_SYNC_SOURCE
: C_SYNC_TARGET
),
2570 abs(hg
) >= 2 ? "full" : "bit-map based");
2575 dev_info(DEV
, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2576 if (drbd_bitmap_io(mdev
, &drbd_bmio_set_n_write
, "set_n_write from sync_handshake"))
2580 if (hg
> 0) { /* become sync source. */
2582 } else if (hg
< 0) { /* become sync target */
2586 if (drbd_bm_total_weight(mdev
)) {
2587 dev_info(DEV
, "No resync, but %lu bits in bitmap!\n",
2588 drbd_bm_total_weight(mdev
));
2595 /* returns 1 if invalid */
2596 static int cmp_after_sb(enum drbd_after_sb_p peer
, enum drbd_after_sb_p self
)
2598 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2599 if ((peer
== ASB_DISCARD_REMOTE
&& self
== ASB_DISCARD_LOCAL
) ||
2600 (self
== ASB_DISCARD_REMOTE
&& peer
== ASB_DISCARD_LOCAL
))
2603 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2604 if (peer
== ASB_DISCARD_REMOTE
|| peer
== ASB_DISCARD_LOCAL
||
2605 self
== ASB_DISCARD_REMOTE
|| self
== ASB_DISCARD_LOCAL
)
2608 /* everything else is valid if they are equal on both sides. */
2612 /* everything es is invalid. */
2616 static int receive_protocol(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
2618 struct p_protocol
*p
= &mdev
->data
.rbuf
.protocol
;
2619 int p_proto
, p_after_sb_0p
, p_after_sb_1p
, p_after_sb_2p
;
2620 int p_want_lose
, p_two_primaries
, cf
;
2621 char p_integrity_alg
[SHARED_SECRET_MAX
] = "";
2623 p_proto
= be32_to_cpu(p
->protocol
);
2624 p_after_sb_0p
= be32_to_cpu(p
->after_sb_0p
);
2625 p_after_sb_1p
= be32_to_cpu(p
->after_sb_1p
);
2626 p_after_sb_2p
= be32_to_cpu(p
->after_sb_2p
);
2627 p_two_primaries
= be32_to_cpu(p
->two_primaries
);
2628 cf
= be32_to_cpu(p
->conn_flags
);
2629 p_want_lose
= cf
& CF_WANT_LOSE
;
2631 clear_bit(CONN_DRY_RUN
, &mdev
->flags
);
2633 if (cf
& CF_DRY_RUN
)
2634 set_bit(CONN_DRY_RUN
, &mdev
->flags
);
2636 if (p_proto
!= mdev
->net_conf
->wire_protocol
) {
2637 dev_err(DEV
, "incompatible communication protocols\n");
2641 if (cmp_after_sb(p_after_sb_0p
, mdev
->net_conf
->after_sb_0p
)) {
2642 dev_err(DEV
, "incompatible after-sb-0pri settings\n");
2646 if (cmp_after_sb(p_after_sb_1p
, mdev
->net_conf
->after_sb_1p
)) {
2647 dev_err(DEV
, "incompatible after-sb-1pri settings\n");
2651 if (cmp_after_sb(p_after_sb_2p
, mdev
->net_conf
->after_sb_2p
)) {
2652 dev_err(DEV
, "incompatible after-sb-2pri settings\n");
2656 if (p_want_lose
&& mdev
->net_conf
->want_lose
) {
2657 dev_err(DEV
, "both sides have the 'want_lose' flag set\n");
2661 if (p_two_primaries
!= mdev
->net_conf
->two_primaries
) {
2662 dev_err(DEV
, "incompatible setting of the two-primaries options\n");
2666 if (mdev
->agreed_pro_version
>= 87) {
2667 unsigned char *my_alg
= mdev
->net_conf
->integrity_alg
;
2669 if (drbd_recv(mdev
, p_integrity_alg
, data_size
) != data_size
)
2672 p_integrity_alg
[SHARED_SECRET_MAX
-1] = 0;
2673 if (strcmp(p_integrity_alg
, my_alg
)) {
2674 dev_err(DEV
, "incompatible setting of the data-integrity-alg\n");
2677 dev_info(DEV
, "data-integrity-alg: %s\n",
2678 my_alg
[0] ? my_alg
: (unsigned char *)"<not-used>");
2684 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2689 * input: alg name, feature name
2690 * return: NULL (alg name was "")
2691 * ERR_PTR(error) if something goes wrong
2692 * or the crypto hash ptr, if it worked out ok. */
2693 struct crypto_hash
*drbd_crypto_alloc_digest_safe(const struct drbd_conf
*mdev
,
2694 const char *alg
, const char *name
)
2696 struct crypto_hash
*tfm
;
2701 tfm
= crypto_alloc_hash(alg
, 0, CRYPTO_ALG_ASYNC
);
2703 dev_err(DEV
, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2704 alg
, name
, PTR_ERR(tfm
));
2707 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm
))) {
2708 crypto_free_hash(tfm
);
2709 dev_err(DEV
, "\"%s\" is not a digest (%s)\n", alg
, name
);
2710 return ERR_PTR(-EINVAL
);
2715 static int receive_SyncParam(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int packet_size
)
2718 struct p_rs_param_95
*p
= &mdev
->data
.rbuf
.rs_param_95
;
2719 unsigned int header_size
, data_size
, exp_max_sz
;
2720 struct crypto_hash
*verify_tfm
= NULL
;
2721 struct crypto_hash
*csums_tfm
= NULL
;
2722 const int apv
= mdev
->agreed_pro_version
;
2723 int *rs_plan_s
= NULL
;
2726 exp_max_sz
= apv
<= 87 ? sizeof(struct p_rs_param
)
2727 : apv
== 88 ? sizeof(struct p_rs_param
)
2729 : apv
<= 94 ? sizeof(struct p_rs_param_89
)
2730 : /* apv >= 95 */ sizeof(struct p_rs_param_95
);
2732 if (packet_size
> exp_max_sz
) {
2733 dev_err(DEV
, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2734 packet_size
, exp_max_sz
);
2739 header_size
= sizeof(struct p_rs_param
) - sizeof(struct p_header80
);
2740 data_size
= packet_size
- header_size
;
2741 } else if (apv
<= 94) {
2742 header_size
= sizeof(struct p_rs_param_89
) - sizeof(struct p_header80
);
2743 data_size
= packet_size
- header_size
;
2744 D_ASSERT(data_size
== 0);
2746 header_size
= sizeof(struct p_rs_param_95
) - sizeof(struct p_header80
);
2747 data_size
= packet_size
- header_size
;
2748 D_ASSERT(data_size
== 0);
2751 /* initialize verify_alg and csums_alg */
2752 memset(p
->verify_alg
, 0, 2 * SHARED_SECRET_MAX
);
2754 if (drbd_recv(mdev
, &p
->head
.payload
, header_size
) != header_size
)
2757 mdev
->sync_conf
.rate
= be32_to_cpu(p
->rate
);
2761 if (data_size
> SHARED_SECRET_MAX
) {
2762 dev_err(DEV
, "verify-alg too long, "
2763 "peer wants %u, accepting only %u byte\n",
2764 data_size
, SHARED_SECRET_MAX
);
2768 if (drbd_recv(mdev
, p
->verify_alg
, data_size
) != data_size
)
2771 /* we expect NUL terminated string */
2772 /* but just in case someone tries to be evil */
2773 D_ASSERT(p
->verify_alg
[data_size
-1] == 0);
2774 p
->verify_alg
[data_size
-1] = 0;
2776 } else /* apv >= 89 */ {
2777 /* we still expect NUL terminated strings */
2778 /* but just in case someone tries to be evil */
2779 D_ASSERT(p
->verify_alg
[SHARED_SECRET_MAX
-1] == 0);
2780 D_ASSERT(p
->csums_alg
[SHARED_SECRET_MAX
-1] == 0);
2781 p
->verify_alg
[SHARED_SECRET_MAX
-1] = 0;
2782 p
->csums_alg
[SHARED_SECRET_MAX
-1] = 0;
2785 if (strcmp(mdev
->sync_conf
.verify_alg
, p
->verify_alg
)) {
2786 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
2787 dev_err(DEV
, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2788 mdev
->sync_conf
.verify_alg
, p
->verify_alg
);
2791 verify_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
2792 p
->verify_alg
, "verify-alg");
2793 if (IS_ERR(verify_tfm
)) {
2799 if (apv
>= 89 && strcmp(mdev
->sync_conf
.csums_alg
, p
->csums_alg
)) {
2800 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
2801 dev_err(DEV
, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2802 mdev
->sync_conf
.csums_alg
, p
->csums_alg
);
2805 csums_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
2806 p
->csums_alg
, "csums-alg");
2807 if (IS_ERR(csums_tfm
)) {
2814 mdev
->sync_conf
.rate
= be32_to_cpu(p
->rate
);
2815 mdev
->sync_conf
.c_plan_ahead
= be32_to_cpu(p
->c_plan_ahead
);
2816 mdev
->sync_conf
.c_delay_target
= be32_to_cpu(p
->c_delay_target
);
2817 mdev
->sync_conf
.c_fill_target
= be32_to_cpu(p
->c_fill_target
);
2818 mdev
->sync_conf
.c_max_rate
= be32_to_cpu(p
->c_max_rate
);
2820 fifo_size
= (mdev
->sync_conf
.c_plan_ahead
* 10 * SLEEP_TIME
) / HZ
;
2821 if (fifo_size
!= mdev
->rs_plan_s
.size
&& fifo_size
> 0) {
2822 rs_plan_s
= kzalloc(sizeof(int) * fifo_size
, GFP_KERNEL
);
2824 dev_err(DEV
, "kmalloc of fifo_buffer failed");
2830 spin_lock(&mdev
->peer_seq_lock
);
2831 /* lock against drbd_nl_syncer_conf() */
2833 strcpy(mdev
->sync_conf
.verify_alg
, p
->verify_alg
);
2834 mdev
->sync_conf
.verify_alg_len
= strlen(p
->verify_alg
) + 1;
2835 crypto_free_hash(mdev
->verify_tfm
);
2836 mdev
->verify_tfm
= verify_tfm
;
2837 dev_info(DEV
, "using verify-alg: \"%s\"\n", p
->verify_alg
);
2840 strcpy(mdev
->sync_conf
.csums_alg
, p
->csums_alg
);
2841 mdev
->sync_conf
.csums_alg_len
= strlen(p
->csums_alg
) + 1;
2842 crypto_free_hash(mdev
->csums_tfm
);
2843 mdev
->csums_tfm
= csums_tfm
;
2844 dev_info(DEV
, "using csums-alg: \"%s\"\n", p
->csums_alg
);
2846 if (fifo_size
!= mdev
->rs_plan_s
.size
) {
2847 kfree(mdev
->rs_plan_s
.values
);
2848 mdev
->rs_plan_s
.values
= rs_plan_s
;
2849 mdev
->rs_plan_s
.size
= fifo_size
;
2850 mdev
->rs_planed
= 0;
2852 spin_unlock(&mdev
->peer_seq_lock
);
2857 /* just for completeness: actually not needed,
2858 * as this is not reached if csums_tfm was ok. */
2859 crypto_free_hash(csums_tfm
);
2860 /* but free the verify_tfm again, if csums_tfm did not work out */
2861 crypto_free_hash(verify_tfm
);
2862 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2866 static void drbd_setup_order_type(struct drbd_conf
*mdev
, int peer
)
2868 /* sorry, we currently have no working implementation
2869 * of distributed TCQ */
2872 /* warn if the arguments differ by more than 12.5% */
2873 static void warn_if_differ_considerably(struct drbd_conf
*mdev
,
2874 const char *s
, sector_t a
, sector_t b
)
2877 if (a
== 0 || b
== 0)
2879 d
= (a
> b
) ? (a
- b
) : (b
- a
);
2880 if (d
> (a
>>3) || d
> (b
>>3))
2881 dev_warn(DEV
, "Considerable difference in %s: %llus vs. %llus\n", s
,
2882 (unsigned long long)a
, (unsigned long long)b
);
2885 static int receive_sizes(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
2887 struct p_sizes
*p
= &mdev
->data
.rbuf
.sizes
;
2888 enum determine_dev_size dd
= unchanged
;
2889 unsigned int max_seg_s
;
2890 sector_t p_size
, p_usize
, my_usize
;
2891 int ldsc
= 0; /* local disk size changed */
2892 enum dds_flags ddsf
;
2894 p_size
= be64_to_cpu(p
->d_size
);
2895 p_usize
= be64_to_cpu(p
->u_size
);
2897 if (p_size
== 0 && mdev
->state
.disk
== D_DISKLESS
) {
2898 dev_err(DEV
, "some backing storage is needed\n");
2899 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2903 /* just store the peer's disk size for now.
2904 * we still need to figure out whether we accept that. */
2905 mdev
->p_size
= p_size
;
2907 if (get_ldev(mdev
)) {
2908 warn_if_differ_considerably(mdev
, "lower level device sizes",
2909 p_size
, drbd_get_max_capacity(mdev
->ldev
));
2910 warn_if_differ_considerably(mdev
, "user requested size",
2911 p_usize
, mdev
->ldev
->dc
.disk_size
);
2913 /* if this is the first connect, or an otherwise expected
2914 * param exchange, choose the minimum */
2915 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
)
2916 p_usize
= min_not_zero((sector_t
)mdev
->ldev
->dc
.disk_size
,
2919 my_usize
= mdev
->ldev
->dc
.disk_size
;
2921 if (mdev
->ldev
->dc
.disk_size
!= p_usize
) {
2922 mdev
->ldev
->dc
.disk_size
= p_usize
;
2923 dev_info(DEV
, "Peer sets u_size to %lu sectors\n",
2924 (unsigned long)mdev
->ldev
->dc
.disk_size
);
2927 /* Never shrink a device with usable data during connect.
2928 But allow online shrinking if we are connected. */
2929 if (drbd_new_dev_size(mdev
, mdev
->ldev
, 0) <
2930 drbd_get_capacity(mdev
->this_bdev
) &&
2931 mdev
->state
.disk
>= D_OUTDATED
&&
2932 mdev
->state
.conn
< C_CONNECTED
) {
2933 dev_err(DEV
, "The peer's disk size is too small!\n");
2934 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2935 mdev
->ldev
->dc
.disk_size
= my_usize
;
2943 ddsf
= be16_to_cpu(p
->dds_flags
);
2944 if (get_ldev(mdev
)) {
2945 dd
= drbd_determin_dev_size(mdev
, ddsf
);
2947 if (dd
== dev_size_error
)
2951 /* I am diskless, need to accept the peer's size. */
2952 drbd_set_my_capacity(mdev
, p_size
);
2955 if (get_ldev(mdev
)) {
2956 if (mdev
->ldev
->known_size
!= drbd_get_capacity(mdev
->ldev
->backing_bdev
)) {
2957 mdev
->ldev
->known_size
= drbd_get_capacity(mdev
->ldev
->backing_bdev
);
2961 if (mdev
->agreed_pro_version
< 94)
2962 max_seg_s
= be32_to_cpu(p
->max_segment_size
);
2963 else if (mdev
->agreed_pro_version
== 94)
2964 max_seg_s
= DRBD_MAX_SIZE_H80_PACKET
;
2965 else /* drbd 8.3.8 onwards */
2966 max_seg_s
= DRBD_MAX_SEGMENT_SIZE
;
2968 if (max_seg_s
!= queue_max_segment_size(mdev
->rq_queue
))
2969 drbd_setup_queue_param(mdev
, max_seg_s
);
2971 drbd_setup_order_type(mdev
, be16_to_cpu(p
->queue_order_type
));
2975 if (mdev
->state
.conn
> C_WF_REPORT_PARAMS
) {
2976 if (be64_to_cpu(p
->c_size
) !=
2977 drbd_get_capacity(mdev
->this_bdev
) || ldsc
) {
2978 /* we have different sizes, probably peer
2979 * needs to know my new size... */
2980 drbd_send_sizes(mdev
, 0, ddsf
);
2982 if (test_and_clear_bit(RESIZE_PENDING
, &mdev
->flags
) ||
2983 (dd
== grew
&& mdev
->state
.conn
== C_CONNECTED
)) {
2984 if (mdev
->state
.pdsk
>= D_INCONSISTENT
&&
2985 mdev
->state
.disk
>= D_INCONSISTENT
) {
2986 if (ddsf
& DDSF_NO_RESYNC
)
2987 dev_info(DEV
, "Resync of new storage suppressed with --assume-clean\n");
2989 resync_after_online_grow(mdev
);
2991 set_bit(RESYNC_AFTER_NEG
, &mdev
->flags
);
2998 static int receive_uuids(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3000 struct p_uuids
*p
= &mdev
->data
.rbuf
.uuids
;
3004 p_uuid
= kmalloc(sizeof(u64
)*UI_EXTENDED_SIZE
, GFP_NOIO
);
3006 for (i
= UI_CURRENT
; i
< UI_EXTENDED_SIZE
; i
++)
3007 p_uuid
[i
] = be64_to_cpu(p
->uuid
[i
]);
3009 kfree(mdev
->p_uuid
);
3010 mdev
->p_uuid
= p_uuid
;
3012 if (mdev
->state
.conn
< C_CONNECTED
&&
3013 mdev
->state
.disk
< D_INCONSISTENT
&&
3014 mdev
->state
.role
== R_PRIMARY
&&
3015 (mdev
->ed_uuid
& ~((u64
)1)) != (p_uuid
[UI_CURRENT
] & ~((u64
)1))) {
3016 dev_err(DEV
, "Can only connect to data with current UUID=%016llX\n",
3017 (unsigned long long)mdev
->ed_uuid
);
3018 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3022 if (get_ldev(mdev
)) {
3023 int skip_initial_sync
=
3024 mdev
->state
.conn
== C_CONNECTED
&&
3025 mdev
->agreed_pro_version
>= 90 &&
3026 mdev
->ldev
->md
.uuid
[UI_CURRENT
] == UUID_JUST_CREATED
&&
3027 (p_uuid
[UI_FLAGS
] & 8);
3028 if (skip_initial_sync
) {
3029 dev_info(DEV
, "Accepted new current UUID, preparing to skip initial sync\n");
3030 drbd_bitmap_io(mdev
, &drbd_bmio_clear_n_write
,
3031 "clear_n_write from receive_uuids");
3032 _drbd_uuid_set(mdev
, UI_CURRENT
, p_uuid
[UI_CURRENT
]);
3033 _drbd_uuid_set(mdev
, UI_BITMAP
, 0);
3034 _drbd_set_state(_NS2(mdev
, disk
, D_UP_TO_DATE
, pdsk
, D_UP_TO_DATE
),
3039 } else if (mdev
->state
.disk
< D_INCONSISTENT
&&
3040 mdev
->state
.role
== R_PRIMARY
) {
3041 /* I am a diskless primary, the peer just created a new current UUID
3043 drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3046 /* Before we test for the disk state, we should wait until an eventually
3047 ongoing cluster wide state change is finished. That is important if
3048 we are primary and are detaching from our disk. We need to see the
3049 new disk state... */
3050 wait_event(mdev
->misc_wait
, !test_bit(CLUSTER_ST_CHANGE
, &mdev
->flags
));
3051 if (mdev
->state
.conn
>= C_CONNECTED
&& mdev
->state
.disk
< D_INCONSISTENT
)
3052 drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3058 * convert_state() - Converts the peer's view of the cluster state to our point of view
3059 * @ps: The state as seen by the peer.
3061 static union drbd_state
convert_state(union drbd_state ps
)
3063 union drbd_state ms
;
3065 static enum drbd_conns c_tab
[] = {
3066 [C_CONNECTED
] = C_CONNECTED
,
3068 [C_STARTING_SYNC_S
] = C_STARTING_SYNC_T
,
3069 [C_STARTING_SYNC_T
] = C_STARTING_SYNC_S
,
3070 [C_DISCONNECTING
] = C_TEAR_DOWN
, /* C_NETWORK_FAILURE, */
3071 [C_VERIFY_S
] = C_VERIFY_T
,
3077 ms
.conn
= c_tab
[ps
.conn
];
3082 ms
.peer_isp
= (ps
.aftr_isp
| ps
.user_isp
);
3087 static int receive_req_state(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3089 struct p_req_state
*p
= &mdev
->data
.rbuf
.req_state
;
3090 union drbd_state mask
, val
;
3093 mask
.i
= be32_to_cpu(p
->mask
);
3094 val
.i
= be32_to_cpu(p
->val
);
3096 if (test_bit(DISCARD_CONCURRENT
, &mdev
->flags
) &&
3097 test_bit(CLUSTER_ST_CHANGE
, &mdev
->flags
)) {
3098 drbd_send_sr_reply(mdev
, SS_CONCURRENT_ST_CHG
);
3102 mask
= convert_state(mask
);
3103 val
= convert_state(val
);
3105 rv
= drbd_change_state(mdev
, CS_VERBOSE
, mask
, val
);
3107 drbd_send_sr_reply(mdev
, rv
);
3113 static int receive_state(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3115 struct p_state
*p
= &mdev
->data
.rbuf
.state
;
3116 union drbd_state os
, ns
, peer_state
;
3117 enum drbd_disk_state real_peer_disk
;
3118 enum chg_state_flags cs_flags
;
3121 peer_state
.i
= be32_to_cpu(p
->state
);
3123 real_peer_disk
= peer_state
.disk
;
3124 if (peer_state
.disk
== D_NEGOTIATING
) {
3125 real_peer_disk
= mdev
->p_uuid
[UI_FLAGS
] & 4 ? D_INCONSISTENT
: D_CONSISTENT
;
3126 dev_info(DEV
, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk
));
3129 spin_lock_irq(&mdev
->req_lock
);
3131 os
= ns
= mdev
->state
;
3132 spin_unlock_irq(&mdev
->req_lock
);
3134 /* peer says his disk is uptodate, while we think it is inconsistent,
3135 * and this happens while we think we have a sync going on. */
3136 if (os
.pdsk
== D_INCONSISTENT
&& real_peer_disk
== D_UP_TO_DATE
&&
3137 os
.conn
> C_CONNECTED
&& os
.disk
== D_UP_TO_DATE
) {
3138 /* If we are (becoming) SyncSource, but peer is still in sync
3139 * preparation, ignore its uptodate-ness to avoid flapping, it
3140 * will change to inconsistent once the peer reaches active
3142 * It may have changed syncer-paused flags, however, so we
3143 * cannot ignore this completely. */
3144 if (peer_state
.conn
> C_CONNECTED
&&
3145 peer_state
.conn
< C_SYNC_SOURCE
)
3146 real_peer_disk
= D_INCONSISTENT
;
3148 /* if peer_state changes to connected at the same time,
3149 * it explicitly notifies us that it finished resync.
3150 * Maybe we should finish it up, too? */
3151 else if (os
.conn
>= C_SYNC_SOURCE
&&
3152 peer_state
.conn
== C_CONNECTED
) {
3153 if (drbd_bm_total_weight(mdev
) <= mdev
->rs_failed
)
3154 drbd_resync_finished(mdev
);
3159 /* peer says his disk is inconsistent, while we think it is uptodate,
3160 * and this happens while the peer still thinks we have a sync going on,
3161 * but we think we are already done with the sync.
3162 * We ignore this to avoid flapping pdsk.
3163 * This should not happen, if the peer is a recent version of drbd. */
3164 if (os
.pdsk
== D_UP_TO_DATE
&& real_peer_disk
== D_INCONSISTENT
&&
3165 os
.conn
== C_CONNECTED
&& peer_state
.conn
> C_SYNC_SOURCE
)
3166 real_peer_disk
= D_UP_TO_DATE
;
3168 if (ns
.conn
== C_WF_REPORT_PARAMS
)
3169 ns
.conn
= C_CONNECTED
;
3171 if (mdev
->p_uuid
&& peer_state
.disk
>= D_NEGOTIATING
&&
3172 get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3173 int cr
; /* consider resync */
3175 /* if we established a new connection */
3176 cr
= (os
.conn
< C_CONNECTED
);
3177 /* if we had an established connection
3178 * and one of the nodes newly attaches a disk */
3179 cr
|= (os
.conn
== C_CONNECTED
&&
3180 (peer_state
.disk
== D_NEGOTIATING
||
3181 os
.disk
== D_NEGOTIATING
));
3182 /* if we have both been inconsistent, and the peer has been
3183 * forced to be UpToDate with --overwrite-data */
3184 cr
|= test_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3185 /* if we had been plain connected, and the admin requested to
3186 * start a sync by "invalidate" or "invalidate-remote" */
3187 cr
|= (os
.conn
== C_CONNECTED
&&
3188 (peer_state
.conn
>= C_STARTING_SYNC_S
&&
3189 peer_state
.conn
<= C_WF_BITMAP_T
));
3192 ns
.conn
= drbd_sync_handshake(mdev
, peer_state
.role
, real_peer_disk
);
3195 if (ns
.conn
== C_MASK
) {
3196 ns
.conn
= C_CONNECTED
;
3197 if (mdev
->state
.disk
== D_NEGOTIATING
) {
3198 drbd_force_state(mdev
, NS(disk
, D_FAILED
));
3199 } else if (peer_state
.disk
== D_NEGOTIATING
) {
3200 dev_err(DEV
, "Disk attach process on the peer node was aborted.\n");
3201 peer_state
.disk
= D_DISKLESS
;
3202 real_peer_disk
= D_DISKLESS
;
3204 if (test_and_clear_bit(CONN_DRY_RUN
, &mdev
->flags
))
3206 D_ASSERT(os
.conn
== C_WF_REPORT_PARAMS
);
3207 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3213 spin_lock_irq(&mdev
->req_lock
);
3214 if (mdev
->state
.i
!= os
.i
)
3216 clear_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3217 ns
.peer
= peer_state
.role
;
3218 ns
.pdsk
= real_peer_disk
;
3219 ns
.peer_isp
= (peer_state
.aftr_isp
| peer_state
.user_isp
);
3220 if ((ns
.conn
== C_CONNECTED
|| ns
.conn
== C_WF_BITMAP_S
) && ns
.disk
== D_NEGOTIATING
)
3221 ns
.disk
= mdev
->new_state_tmp
.disk
;
3222 cs_flags
= CS_VERBOSE
+ (os
.conn
< C_CONNECTED
&& ns
.conn
>= C_CONNECTED
? 0 : CS_HARD
);
3223 if (ns
.pdsk
== D_CONSISTENT
&& is_susp(ns
) && ns
.conn
== C_CONNECTED
&& os
.conn
< C_CONNECTED
&&
3224 test_bit(NEW_CUR_UUID
, &mdev
->flags
)) {
3225 /* Do not allow tl_restart(resend) for a rebooted peer. We can only allow this
3226 for temporal network outages! */
3227 spin_unlock_irq(&mdev
->req_lock
);
3228 dev_err(DEV
, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3230 drbd_uuid_new_current(mdev
);
3231 clear_bit(NEW_CUR_UUID
, &mdev
->flags
);
3232 drbd_force_state(mdev
, NS2(conn
, C_PROTOCOL_ERROR
, susp
, 0));
3235 rv
= _drbd_set_state(mdev
, ns
, cs_flags
, NULL
);
3237 spin_unlock_irq(&mdev
->req_lock
);
3239 if (rv
< SS_SUCCESS
) {
3240 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3244 if (os
.conn
> C_WF_REPORT_PARAMS
) {
3245 if (ns
.conn
> C_CONNECTED
&& peer_state
.conn
<= C_CONNECTED
&&
3246 peer_state
.disk
!= D_NEGOTIATING
) {
3247 /* we want resync, peer has not yet decided to sync... */
3248 /* Nowadays only used when forcing a node into primary role and
3249 setting its disk to UpToDate with that */
3250 drbd_send_uuids(mdev
);
3251 drbd_send_state(mdev
);
3255 mdev
->net_conf
->want_lose
= 0;
3257 drbd_md_sync(mdev
); /* update connected indicator, la_size, ... */
3262 static int receive_sync_uuid(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3264 struct p_rs_uuid
*p
= &mdev
->data
.rbuf
.rs_uuid
;
3266 wait_event(mdev
->misc_wait
,
3267 mdev
->state
.conn
== C_WF_SYNC_UUID
||
3268 mdev
->state
.conn
< C_CONNECTED
||
3269 mdev
->state
.disk
< D_NEGOTIATING
);
3271 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3273 /* Here the _drbd_uuid_ functions are right, current should
3274 _not_ be rotated into the history */
3275 if (get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3276 _drbd_uuid_set(mdev
, UI_CURRENT
, be64_to_cpu(p
->uuid
));
3277 _drbd_uuid_set(mdev
, UI_BITMAP
, 0UL);
3279 drbd_start_resync(mdev
, C_SYNC_TARGET
);
3283 dev_err(DEV
, "Ignoring SyncUUID packet!\n");
3288 enum receive_bitmap_ret
{ OK
, DONE
, FAILED
};
3290 static enum receive_bitmap_ret
3291 receive_bitmap_plain(struct drbd_conf
*mdev
, unsigned int data_size
,
3292 unsigned long *buffer
, struct bm_xfer_ctx
*c
)
3294 unsigned num_words
= min_t(size_t, BM_PACKET_WORDS
, c
->bm_words
- c
->word_offset
);
3295 unsigned want
= num_words
* sizeof(long);
3297 if (want
!= data_size
) {
3298 dev_err(DEV
, "%s:want (%u) != data_size (%u)\n", __func__
, want
, data_size
);
3303 if (drbd_recv(mdev
, buffer
, want
) != want
)
3306 drbd_bm_merge_lel(mdev
, c
->word_offset
, num_words
, buffer
);
3308 c
->word_offset
+= num_words
;
3309 c
->bit_offset
= c
->word_offset
* BITS_PER_LONG
;
3310 if (c
->bit_offset
> c
->bm_bits
)
3311 c
->bit_offset
= c
->bm_bits
;
3316 static enum receive_bitmap_ret
3317 recv_bm_rle_bits(struct drbd_conf
*mdev
,
3318 struct p_compressed_bm
*p
,
3319 struct bm_xfer_ctx
*c
)
3321 struct bitstream bs
;
3325 unsigned long s
= c
->bit_offset
;
3327 int len
= be16_to_cpu(p
->head
.length
) - (sizeof(*p
) - sizeof(p
->head
));
3328 int toggle
= DCBP_get_start(p
);
3332 bitstream_init(&bs
, p
->code
, len
, DCBP_get_pad_bits(p
));
3334 bits
= bitstream_get_bits(&bs
, &look_ahead
, 64);
3338 for (have
= bits
; have
> 0; s
+= rl
, toggle
= !toggle
) {
3339 bits
= vli_decode_bits(&rl
, look_ahead
);
3345 if (e
>= c
->bm_bits
) {
3346 dev_err(DEV
, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e
);
3349 _drbd_bm_set_bits(mdev
, s
, e
);
3353 dev_err(DEV
, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3354 have
, bits
, look_ahead
,
3355 (unsigned int)(bs
.cur
.b
- p
->code
),
3356 (unsigned int)bs
.buf_len
);
3359 look_ahead
>>= bits
;
3362 bits
= bitstream_get_bits(&bs
, &tmp
, 64 - have
);
3365 look_ahead
|= tmp
<< have
;
3370 bm_xfer_ctx_bit_to_word_offset(c
);
3372 return (s
== c
->bm_bits
) ? DONE
: OK
;
3375 static enum receive_bitmap_ret
3376 decode_bitmap_c(struct drbd_conf
*mdev
,
3377 struct p_compressed_bm
*p
,
3378 struct bm_xfer_ctx
*c
)
3380 if (DCBP_get_code(p
) == RLE_VLI_Bits
)
3381 return recv_bm_rle_bits(mdev
, p
, c
);
3383 /* other variants had been implemented for evaluation,
3384 * but have been dropped as this one turned out to be "best"
3385 * during all our tests. */
3387 dev_err(DEV
, "receive_bitmap_c: unknown encoding %u\n", p
->encoding
);
3388 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3392 void INFO_bm_xfer_stats(struct drbd_conf
*mdev
,
3393 const char *direction
, struct bm_xfer_ctx
*c
)
3395 /* what would it take to transfer it "plaintext" */
3396 unsigned plain
= sizeof(struct p_header80
) *
3397 ((c
->bm_words
+BM_PACKET_WORDS
-1)/BM_PACKET_WORDS
+1)
3398 + c
->bm_words
* sizeof(long);
3399 unsigned total
= c
->bytes
[0] + c
->bytes
[1];
3402 /* total can not be zero. but just in case: */
3406 /* don't report if not compressed */
3410 /* total < plain. check for overflow, still */
3411 r
= (total
> UINT_MAX
/1000) ? (total
/ (plain
/1000))
3412 : (1000 * total
/ plain
);
3418 dev_info(DEV
, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3419 "total %u; compression: %u.%u%%\n",
3421 c
->bytes
[1], c
->packets
[1],
3422 c
->bytes
[0], c
->packets
[0],
3423 total
, r
/10, r
% 10);
3426 /* Since we are processing the bitfield from lower addresses to higher,
3427 it does not matter if the process it in 32 bit chunks or 64 bit
3428 chunks as long as it is little endian. (Understand it as byte stream,
3429 beginning with the lowest byte...) If we would use big endian
3430 we would need to process it from the highest address to the lowest,
3431 in order to be agnostic to the 32 vs 64 bits issue.
3433 returns 0 on failure, 1 if we successfully received it. */
3434 static int receive_bitmap(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3436 struct bm_xfer_ctx c
;
3438 enum receive_bitmap_ret ret
;
3440 struct p_header80
*h
= &mdev
->data
.rbuf
.header
.h80
;
3442 wait_event(mdev
->misc_wait
, !atomic_read(&mdev
->ap_bio_cnt
));
3444 drbd_bm_lock(mdev
, "receive bitmap");
3446 /* maybe we should use some per thread scratch page,
3447 * and allocate that during initial device creation? */
3448 buffer
= (unsigned long *) __get_free_page(GFP_NOIO
);
3450 dev_err(DEV
, "failed to allocate one page buffer in %s\n", __func__
);
3454 c
= (struct bm_xfer_ctx
) {
3455 .bm_bits
= drbd_bm_bits(mdev
),
3456 .bm_words
= drbd_bm_words(mdev
),
3460 if (cmd
== P_BITMAP
) {
3461 ret
= receive_bitmap_plain(mdev
, data_size
, buffer
, &c
);
3462 } else if (cmd
== P_COMPRESSED_BITMAP
) {
3463 /* MAYBE: sanity check that we speak proto >= 90,
3464 * and the feature is enabled! */
3465 struct p_compressed_bm
*p
;
3467 if (data_size
> BM_PACKET_PAYLOAD_BYTES
) {
3468 dev_err(DEV
, "ReportCBitmap packet too large\n");
3471 /* use the page buff */
3473 memcpy(p
, h
, sizeof(*h
));
3474 if (drbd_recv(mdev
, p
->head
.payload
, data_size
) != data_size
)
3476 if (data_size
<= (sizeof(*p
) - sizeof(p
->head
))) {
3477 dev_err(DEV
, "ReportCBitmap packet too small (l:%u)\n", data_size
);
3480 ret
= decode_bitmap_c(mdev
, p
, &c
);
3482 dev_warn(DEV
, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", cmd
);
3486 c
.packets
[cmd
== P_BITMAP
]++;
3487 c
.bytes
[cmd
== P_BITMAP
] += sizeof(struct p_header80
) + data_size
;
3492 if (!drbd_recv_header(mdev
, &cmd
, &data_size
))
3494 } while (ret
== OK
);
3498 INFO_bm_xfer_stats(mdev
, "receive", &c
);
3500 if (mdev
->state
.conn
== C_WF_BITMAP_T
) {
3501 ok
= !drbd_send_bitmap(mdev
);
3504 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
3505 ok
= _drbd_request_state(mdev
, NS(conn
, C_WF_SYNC_UUID
), CS_VERBOSE
);
3506 D_ASSERT(ok
== SS_SUCCESS
);
3507 } else if (mdev
->state
.conn
!= C_WF_BITMAP_S
) {
3508 /* admin may have requested C_DISCONNECTING,
3509 * other threads may have noticed network errors */
3510 dev_info(DEV
, "unexpected cstate (%s) in receive_bitmap\n",
3511 drbd_conn_str(mdev
->state
.conn
));
3516 drbd_bm_unlock(mdev
);
3517 if (ok
&& mdev
->state
.conn
== C_WF_BITMAP_S
)
3518 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
3519 free_page((unsigned long) buffer
);
3523 static int receive_skip(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3525 /* TODO zero copy sink :) */
3526 static char sink
[128];
3529 dev_warn(DEV
, "skipping unknown optional packet type %d, l: %d!\n",
3534 want
= min_t(int, size
, sizeof(sink
));
3535 r
= drbd_recv(mdev
, sink
, want
);
3536 ERR_IF(r
<= 0) break;
3542 static int receive_UnplugRemote(struct drbd_conf
*mdev
, enum drbd_packets cmd
, unsigned int data_size
)
3544 /* Make sure we've acked all the TCP data associated
3545 * with the data requests being unplugged */
3546 drbd_tcp_quickack(mdev
->data
.socket
);
3551 typedef int (*drbd_cmd_handler_f
)(struct drbd_conf
*, enum drbd_packets cmd
, unsigned int to_receive
);
3556 drbd_cmd_handler_f function
;
3559 static struct data_cmd drbd_cmd_handler
[] = {
3560 [P_DATA
] = { 1, sizeof(struct p_data
), receive_Data
},
3561 [P_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_DataReply
},
3562 [P_RS_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_RSDataReply
} ,
3563 [P_BARRIER
] = { 0, sizeof(struct p_barrier
), receive_Barrier
} ,
3564 [P_BITMAP
] = { 1, sizeof(struct p_header80
), receive_bitmap
} ,
3565 [P_COMPRESSED_BITMAP
] = { 1, sizeof(struct p_header80
), receive_bitmap
} ,
3566 [P_UNPLUG_REMOTE
] = { 0, sizeof(struct p_header80
), receive_UnplugRemote
},
3567 [P_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
3568 [P_RS_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
3569 [P_SYNC_PARAM
] = { 1, sizeof(struct p_header80
), receive_SyncParam
},
3570 [P_SYNC_PARAM89
] = { 1, sizeof(struct p_header80
), receive_SyncParam
},
3571 [P_PROTOCOL
] = { 1, sizeof(struct p_protocol
), receive_protocol
},
3572 [P_UUIDS
] = { 0, sizeof(struct p_uuids
), receive_uuids
},
3573 [P_SIZES
] = { 0, sizeof(struct p_sizes
), receive_sizes
},
3574 [P_STATE
] = { 0, sizeof(struct p_state
), receive_state
},
3575 [P_STATE_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_state
},
3576 [P_SYNC_UUID
] = { 0, sizeof(struct p_rs_uuid
), receive_sync_uuid
},
3577 [P_OV_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
3578 [P_OV_REPLY
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
3579 [P_CSUM_RS_REQUEST
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
3580 [P_DELAY_PROBE
] = { 0, sizeof(struct p_delay_probe93
), receive_skip
},
3581 /* anything missing from this table is in
3582 * the asender_tbl, see get_asender_cmd */
3583 [P_MAX_CMD
] = { 0, 0, NULL
},
3586 /* All handler functions that expect a sub-header get that sub-heder in
3587 mdev->data.rbuf.header.head.payload.
3589 Usually in mdev->data.rbuf.header.head the callback can find the usual
3590 p_header, but they may not rely on that. Since there is also p_header95 !
3593 static void drbdd(struct drbd_conf
*mdev
)
3595 union p_header
*header
= &mdev
->data
.rbuf
.header
;
3596 unsigned int packet_size
;
3597 enum drbd_packets cmd
;
3598 size_t shs
; /* sub header size */
3601 while (get_t_state(&mdev
->receiver
) == Running
) {
3602 drbd_thread_current_set_cpu(mdev
);
3603 if (!drbd_recv_header(mdev
, &cmd
, &packet_size
))
3606 if (unlikely(cmd
>= P_MAX_CMD
|| !drbd_cmd_handler
[cmd
].function
)) {
3607 dev_err(DEV
, "unknown packet type %d, l: %d!\n", cmd
, packet_size
);
3611 shs
= drbd_cmd_handler
[cmd
].pkt_size
- sizeof(union p_header
);
3612 if (packet_size
- shs
> 0 && !drbd_cmd_handler
[cmd
].expect_payload
) {
3613 dev_err(DEV
, "No payload expected %s l:%d\n", cmdname(cmd
), packet_size
);
3618 rv
= drbd_recv(mdev
, &header
->h80
.payload
, shs
);
3619 if (unlikely(rv
!= shs
)) {
3620 dev_err(DEV
, "short read while reading sub header: rv=%d\n", rv
);
3625 rv
= drbd_cmd_handler
[cmd
].function(mdev
, cmd
, packet_size
- shs
);
3627 if (unlikely(!rv
)) {
3628 dev_err(DEV
, "error receiving %s, l: %d!\n",
3629 cmdname(cmd
), packet_size
);
3636 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3638 /* If we leave here, we probably want to update at least the
3639 * "Connected" indicator on stable storage. Do so explicitly here. */
3643 void drbd_flush_workqueue(struct drbd_conf
*mdev
)
3645 struct drbd_wq_barrier barr
;
3647 barr
.w
.cb
= w_prev_work_done
;
3648 init_completion(&barr
.done
);
3649 drbd_queue_work(&mdev
->data
.work
, &barr
.w
);
3650 wait_for_completion(&barr
.done
);
3653 void drbd_free_tl_hash(struct drbd_conf
*mdev
)
3655 struct hlist_head
*h
;
3657 spin_lock_irq(&mdev
->req_lock
);
3659 if (!mdev
->tl_hash
|| mdev
->state
.conn
!= C_STANDALONE
) {
3660 spin_unlock_irq(&mdev
->req_lock
);
3664 for (h
= mdev
->ee_hash
; h
< mdev
->ee_hash
+ mdev
->ee_hash_s
; h
++)
3666 dev_err(DEV
, "ASSERT FAILED ee_hash[%u].first == %p, expected NULL\n",
3667 (int)(h
- mdev
->ee_hash
), h
->first
);
3668 kfree(mdev
->ee_hash
);
3669 mdev
->ee_hash
= NULL
;
3670 mdev
->ee_hash_s
= 0;
3673 for (h
= mdev
->tl_hash
; h
< mdev
->tl_hash
+ mdev
->tl_hash_s
; h
++)
3675 dev_err(DEV
, "ASSERT FAILED tl_hash[%u] == %p, expected NULL\n",
3676 (int)(h
- mdev
->tl_hash
), h
->first
);
3677 kfree(mdev
->tl_hash
);
3678 mdev
->tl_hash
= NULL
;
3679 mdev
->tl_hash_s
= 0;
3680 spin_unlock_irq(&mdev
->req_lock
);
3683 static void drbd_disconnect(struct drbd_conf
*mdev
)
3685 enum drbd_fencing_p fp
;
3686 union drbd_state os
, ns
;
3687 int rv
= SS_UNKNOWN_ERROR
;
3690 if (mdev
->state
.conn
== C_STANDALONE
)
3692 if (mdev
->state
.conn
>= C_WF_CONNECTION
)
3693 dev_err(DEV
, "ASSERT FAILED cstate = %s, expected < WFConnection\n",
3694 drbd_conn_str(mdev
->state
.conn
));
3696 /* asender does not clean up anything. it must not interfere, either */
3697 drbd_thread_stop(&mdev
->asender
);
3698 drbd_free_sock(mdev
);
3700 /* wait for current activity to cease. */
3701 spin_lock_irq(&mdev
->req_lock
);
3702 _drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
3703 _drbd_wait_ee_list_empty(mdev
, &mdev
->sync_ee
);
3704 _drbd_wait_ee_list_empty(mdev
, &mdev
->read_ee
);
3705 spin_unlock_irq(&mdev
->req_lock
);
3707 /* We do not have data structures that would allow us to
3708 * get the rs_pending_cnt down to 0 again.
3709 * * On C_SYNC_TARGET we do not have any data structures describing
3710 * the pending RSDataRequest's we have sent.
3711 * * On C_SYNC_SOURCE there is no data structure that tracks
3712 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
3713 * And no, it is not the sum of the reference counts in the
3714 * resync_LRU. The resync_LRU tracks the whole operation including
3715 * the disk-IO, while the rs_pending_cnt only tracks the blocks
3717 drbd_rs_cancel_all(mdev
);
3719 mdev
->rs_failed
= 0;
3720 atomic_set(&mdev
->rs_pending_cnt
, 0);
3721 wake_up(&mdev
->misc_wait
);
3723 /* make sure syncer is stopped and w_resume_next_sg queued */
3724 del_timer_sync(&mdev
->resync_timer
);
3725 resync_timer_fn((unsigned long)mdev
);
3727 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
3728 * w_make_resync_request etc. which may still be on the worker queue
3729 * to be "canceled" */
3730 drbd_flush_workqueue(mdev
);
3732 /* This also does reclaim_net_ee(). If we do this too early, we might
3733 * miss some resync ee and pages.*/
3734 drbd_process_done_ee(mdev
);
3736 kfree(mdev
->p_uuid
);
3737 mdev
->p_uuid
= NULL
;
3739 if (!is_susp(mdev
->state
))
3742 dev_info(DEV
, "Connection closed\n");
3747 if (get_ldev(mdev
)) {
3748 fp
= mdev
->ldev
->dc
.fencing
;
3752 if (mdev
->state
.role
== R_PRIMARY
&& fp
>= FP_RESOURCE
&& mdev
->state
.pdsk
>= D_UNKNOWN
)
3753 drbd_try_outdate_peer_async(mdev
);
3755 spin_lock_irq(&mdev
->req_lock
);
3757 if (os
.conn
>= C_UNCONNECTED
) {
3758 /* Do not restart in case we are C_DISCONNECTING */
3760 ns
.conn
= C_UNCONNECTED
;
3761 rv
= _drbd_set_state(mdev
, ns
, CS_VERBOSE
, NULL
);
3763 spin_unlock_irq(&mdev
->req_lock
);
3765 if (os
.conn
== C_DISCONNECTING
) {
3766 wait_event(mdev
->net_cnt_wait
, atomic_read(&mdev
->net_cnt
) == 0);
3768 if (!is_susp(mdev
->state
)) {
3769 /* we must not free the tl_hash
3770 * while application io is still on the fly */
3771 wait_event(mdev
->misc_wait
, !atomic_read(&mdev
->ap_bio_cnt
));
3772 drbd_free_tl_hash(mdev
);
3775 crypto_free_hash(mdev
->cram_hmac_tfm
);
3776 mdev
->cram_hmac_tfm
= NULL
;
3778 kfree(mdev
->net_conf
);
3779 mdev
->net_conf
= NULL
;
3780 drbd_request_state(mdev
, NS(conn
, C_STANDALONE
));
3783 /* tcp_close and release of sendpage pages can be deferred. I don't
3784 * want to use SO_LINGER, because apparently it can be deferred for
3785 * more than 20 seconds (longest time I checked).
3787 * Actually we don't care for exactly when the network stack does its
3788 * put_page(), but release our reference on these pages right here.
3790 i
= drbd_release_ee(mdev
, &mdev
->net_ee
);
3792 dev_info(DEV
, "net_ee not empty, killed %u entries\n", i
);
3793 i
= atomic_read(&mdev
->pp_in_use_by_net
);
3795 dev_info(DEV
, "pp_in_use_by_net = %d, expected 0\n", i
);
3796 i
= atomic_read(&mdev
->pp_in_use
);
3798 dev_info(DEV
, "pp_in_use = %d, expected 0\n", i
);
3800 D_ASSERT(list_empty(&mdev
->read_ee
));
3801 D_ASSERT(list_empty(&mdev
->active_ee
));
3802 D_ASSERT(list_empty(&mdev
->sync_ee
));
3803 D_ASSERT(list_empty(&mdev
->done_ee
));
3805 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
3806 atomic_set(&mdev
->current_epoch
->epoch_size
, 0);
3807 D_ASSERT(list_empty(&mdev
->current_epoch
->list
));
3811 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
3812 * we can agree on is stored in agreed_pro_version.
3814 * feature flags and the reserved array should be enough room for future
3815 * enhancements of the handshake protocol, and possible plugins...
3817 * for now, they are expected to be zero, but ignored.
3819 static int drbd_send_handshake(struct drbd_conf
*mdev
)
3821 /* ASSERT current == mdev->receiver ... */
3822 struct p_handshake
*p
= &mdev
->data
.sbuf
.handshake
;
3825 if (mutex_lock_interruptible(&mdev
->data
.mutex
)) {
3826 dev_err(DEV
, "interrupted during initial handshake\n");
3827 return 0; /* interrupted. not ok. */
3830 if (mdev
->data
.socket
== NULL
) {
3831 mutex_unlock(&mdev
->data
.mutex
);
3835 memset(p
, 0, sizeof(*p
));
3836 p
->protocol_min
= cpu_to_be32(PRO_VERSION_MIN
);
3837 p
->protocol_max
= cpu_to_be32(PRO_VERSION_MAX
);
3838 ok
= _drbd_send_cmd( mdev
, mdev
->data
.socket
, P_HAND_SHAKE
,
3839 (struct p_header80
*)p
, sizeof(*p
), 0 );
3840 mutex_unlock(&mdev
->data
.mutex
);
3846 * 1 yes, we have a valid connection
3847 * 0 oops, did not work out, please try again
3848 * -1 peer talks different language,
3849 * no point in trying again, please go standalone.
3851 static int drbd_do_handshake(struct drbd_conf
*mdev
)
3853 /* ASSERT current == mdev->receiver ... */
3854 struct p_handshake
*p
= &mdev
->data
.rbuf
.handshake
;
3855 const int expect
= sizeof(struct p_handshake
) - sizeof(struct p_header80
);
3856 unsigned int length
;
3857 enum drbd_packets cmd
;
3860 rv
= drbd_send_handshake(mdev
);
3864 rv
= drbd_recv_header(mdev
, &cmd
, &length
);
3868 if (cmd
!= P_HAND_SHAKE
) {
3869 dev_err(DEV
, "expected HandShake packet, received: %s (0x%04x)\n",
3874 if (length
!= expect
) {
3875 dev_err(DEV
, "expected HandShake length: %u, received: %u\n",
3880 rv
= drbd_recv(mdev
, &p
->head
.payload
, expect
);
3883 dev_err(DEV
, "short read receiving handshake packet: l=%u\n", rv
);
3887 p
->protocol_min
= be32_to_cpu(p
->protocol_min
);
3888 p
->protocol_max
= be32_to_cpu(p
->protocol_max
);
3889 if (p
->protocol_max
== 0)
3890 p
->protocol_max
= p
->protocol_min
;
3892 if (PRO_VERSION_MAX
< p
->protocol_min
||
3893 PRO_VERSION_MIN
> p
->protocol_max
)
3896 mdev
->agreed_pro_version
= min_t(int, PRO_VERSION_MAX
, p
->protocol_max
);
3898 dev_info(DEV
, "Handshake successful: "
3899 "Agreed network protocol version %d\n", mdev
->agreed_pro_version
);
3904 dev_err(DEV
, "incompatible DRBD dialects: "
3905 "I support %d-%d, peer supports %d-%d\n",
3906 PRO_VERSION_MIN
, PRO_VERSION_MAX
,
3907 p
->protocol_min
, p
->protocol_max
);
3911 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
3912 static int drbd_do_auth(struct drbd_conf
*mdev
)
3914 dev_err(DEV
, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
3915 dev_err(DEV
, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
3919 #define CHALLENGE_LEN 64
3923 0 - failed, try again (network error),
3924 -1 - auth failed, don't try again.
3927 static int drbd_do_auth(struct drbd_conf
*mdev
)
3929 char my_challenge
[CHALLENGE_LEN
]; /* 64 Bytes... */
3930 struct scatterlist sg
;
3931 char *response
= NULL
;
3932 char *right_response
= NULL
;
3933 char *peers_ch
= NULL
;
3934 unsigned int key_len
= strlen(mdev
->net_conf
->shared_secret
);
3935 unsigned int resp_size
;
3936 struct hash_desc desc
;
3937 enum drbd_packets cmd
;
3938 unsigned int length
;
3941 desc
.tfm
= mdev
->cram_hmac_tfm
;
3944 rv
= crypto_hash_setkey(mdev
->cram_hmac_tfm
,
3945 (u8
*)mdev
->net_conf
->shared_secret
, key_len
);
3947 dev_err(DEV
, "crypto_hash_setkey() failed with %d\n", rv
);
3952 get_random_bytes(my_challenge
, CHALLENGE_LEN
);
3954 rv
= drbd_send_cmd2(mdev
, P_AUTH_CHALLENGE
, my_challenge
, CHALLENGE_LEN
);
3958 rv
= drbd_recv_header(mdev
, &cmd
, &length
);
3962 if (cmd
!= P_AUTH_CHALLENGE
) {
3963 dev_err(DEV
, "expected AuthChallenge packet, received: %s (0x%04x)\n",
3969 if (length
> CHALLENGE_LEN
* 2) {
3970 dev_err(DEV
, "expected AuthChallenge payload too big.\n");
3975 peers_ch
= kmalloc(length
, GFP_NOIO
);
3976 if (peers_ch
== NULL
) {
3977 dev_err(DEV
, "kmalloc of peers_ch failed\n");
3982 rv
= drbd_recv(mdev
, peers_ch
, length
);
3985 dev_err(DEV
, "short read AuthChallenge: l=%u\n", rv
);
3990 resp_size
= crypto_hash_digestsize(mdev
->cram_hmac_tfm
);
3991 response
= kmalloc(resp_size
, GFP_NOIO
);
3992 if (response
== NULL
) {
3993 dev_err(DEV
, "kmalloc of response failed\n");
3998 sg_init_table(&sg
, 1);
3999 sg_set_buf(&sg
, peers_ch
, length
);
4001 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, response
);
4003 dev_err(DEV
, "crypto_hash_digest() failed with %d\n", rv
);
4008 rv
= drbd_send_cmd2(mdev
, P_AUTH_RESPONSE
, response
, resp_size
);
4012 rv
= drbd_recv_header(mdev
, &cmd
, &length
);
4016 if (cmd
!= P_AUTH_RESPONSE
) {
4017 dev_err(DEV
, "expected AuthResponse packet, received: %s (0x%04x)\n",
4023 if (length
!= resp_size
) {
4024 dev_err(DEV
, "expected AuthResponse payload of wrong size\n");
4029 rv
= drbd_recv(mdev
, response
, resp_size
);
4031 if (rv
!= resp_size
) {
4032 dev_err(DEV
, "short read receiving AuthResponse: l=%u\n", rv
);
4037 right_response
= kmalloc(resp_size
, GFP_NOIO
);
4038 if (right_response
== NULL
) {
4039 dev_err(DEV
, "kmalloc of right_response failed\n");
4044 sg_set_buf(&sg
, my_challenge
, CHALLENGE_LEN
);
4046 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, right_response
);
4048 dev_err(DEV
, "crypto_hash_digest() failed with %d\n", rv
);
4053 rv
= !memcmp(response
, right_response
, resp_size
);
4056 dev_info(DEV
, "Peer authenticated using %d bytes of '%s' HMAC\n",
4057 resp_size
, mdev
->net_conf
->cram_hmac_alg
);
4064 kfree(right_response
);
4070 int drbdd_init(struct drbd_thread
*thi
)
4072 struct drbd_conf
*mdev
= thi
->mdev
;
4073 unsigned int minor
= mdev_to_minor(mdev
);
4076 sprintf(current
->comm
, "drbd%d_receiver", minor
);
4078 dev_info(DEV
, "receiver (re)started\n");
4081 h
= drbd_connect(mdev
);
4083 drbd_disconnect(mdev
);
4084 __set_current_state(TASK_INTERRUPTIBLE
);
4085 schedule_timeout(HZ
);
4088 dev_warn(DEV
, "Discarding network configuration.\n");
4089 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
4094 if (get_net_conf(mdev
)) {
4100 drbd_disconnect(mdev
);
4102 dev_info(DEV
, "receiver terminated\n");
4106 /* ********* acknowledge sender ******** */
4108 static int got_RqSReply(struct drbd_conf
*mdev
, struct p_header80
*h
)
4110 struct p_req_state_reply
*p
= (struct p_req_state_reply
*)h
;
4112 int retcode
= be32_to_cpu(p
->retcode
);
4114 if (retcode
>= SS_SUCCESS
) {
4115 set_bit(CL_ST_CHG_SUCCESS
, &mdev
->flags
);
4117 set_bit(CL_ST_CHG_FAIL
, &mdev
->flags
);
4118 dev_err(DEV
, "Requested state change failed by peer: %s (%d)\n",
4119 drbd_set_st_err_str(retcode
), retcode
);
4121 wake_up(&mdev
->state_wait
);
4126 static int got_Ping(struct drbd_conf
*mdev
, struct p_header80
*h
)
4128 return drbd_send_ping_ack(mdev
);
4132 static int got_PingAck(struct drbd_conf
*mdev
, struct p_header80
*h
)
4134 /* restore idle timeout */
4135 mdev
->meta
.socket
->sk
->sk_rcvtimeo
= mdev
->net_conf
->ping_int
*HZ
;
4136 if (!test_and_set_bit(GOT_PING_ACK
, &mdev
->flags
))
4137 wake_up(&mdev
->misc_wait
);
4142 static int got_IsInSync(struct drbd_conf
*mdev
, struct p_header80
*h
)
4144 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4145 sector_t sector
= be64_to_cpu(p
->sector
);
4146 int blksize
= be32_to_cpu(p
->blksize
);
4148 D_ASSERT(mdev
->agreed_pro_version
>= 89);
4150 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4152 if (get_ldev(mdev
)) {
4153 drbd_rs_complete_io(mdev
, sector
);
4154 drbd_set_in_sync(mdev
, sector
, blksize
);
4155 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4156 mdev
->rs_same_csum
+= (blksize
>> BM_BLOCK_SHIFT
);
4159 dec_rs_pending(mdev
);
4160 atomic_add(blksize
>> 9, &mdev
->rs_sect_in
);
4165 /* when we receive the ACK for a write request,
4166 * verify that we actually know about it */
4167 static struct drbd_request
*_ack_id_to_req(struct drbd_conf
*mdev
,
4168 u64 id
, sector_t sector
)
4170 struct hlist_head
*slot
= tl_hash_slot(mdev
, sector
);
4171 struct hlist_node
*n
;
4172 struct drbd_request
*req
;
4174 hlist_for_each_entry(req
, n
, slot
, colision
) {
4175 if ((unsigned long)req
== (unsigned long)id
) {
4176 if (req
->sector
!= sector
) {
4177 dev_err(DEV
, "_ack_id_to_req: found req %p but it has "
4178 "wrong sector (%llus versus %llus)\n", req
,
4179 (unsigned long long)req
->sector
,
4180 (unsigned long long)sector
);
4186 dev_err(DEV
, "_ack_id_to_req: failed to find req %p, sector %llus in list\n",
4187 (void *)(unsigned long)id
, (unsigned long long)sector
);
4191 typedef struct drbd_request
*(req_validator_fn
)
4192 (struct drbd_conf
*mdev
, u64 id
, sector_t sector
);
4194 static int validate_req_change_req_state(struct drbd_conf
*mdev
,
4195 u64 id
, sector_t sector
, req_validator_fn validator
,
4196 const char *func
, enum drbd_req_event what
)
4198 struct drbd_request
*req
;
4199 struct bio_and_error m
;
4201 spin_lock_irq(&mdev
->req_lock
);
4202 req
= validator(mdev
, id
, sector
);
4203 if (unlikely(!req
)) {
4204 spin_unlock_irq(&mdev
->req_lock
);
4205 dev_err(DEV
, "%s: got a corrupt block_id/sector pair\n", func
);
4208 __req_mod(req
, what
, &m
);
4209 spin_unlock_irq(&mdev
->req_lock
);
4212 complete_master_bio(mdev
, &m
);
4216 static int got_BlockAck(struct drbd_conf
*mdev
, struct p_header80
*h
)
4218 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4219 sector_t sector
= be64_to_cpu(p
->sector
);
4220 int blksize
= be32_to_cpu(p
->blksize
);
4221 enum drbd_req_event what
;
4223 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4225 if (is_syncer_block_id(p
->block_id
)) {
4226 drbd_set_in_sync(mdev
, sector
, blksize
);
4227 dec_rs_pending(mdev
);
4230 switch (be16_to_cpu(h
->command
)) {
4231 case P_RS_WRITE_ACK
:
4232 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4233 what
= write_acked_by_peer_and_sis
;
4236 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4237 what
= write_acked_by_peer
;
4240 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_B
);
4241 what
= recv_acked_by_peer
;
4244 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4245 what
= conflict_discarded_by_peer
;
4252 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4253 _ack_id_to_req
, __func__
, what
);
4256 static int got_NegAck(struct drbd_conf
*mdev
, struct p_header80
*h
)
4258 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4259 sector_t sector
= be64_to_cpu(p
->sector
);
4261 if (__ratelimit(&drbd_ratelimit_state
))
4262 dev_warn(DEV
, "Got NegAck packet. Peer is in troubles?\n");
4264 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4266 if (is_syncer_block_id(p
->block_id
)) {
4267 int size
= be32_to_cpu(p
->blksize
);
4268 dec_rs_pending(mdev
);
4269 drbd_rs_failed_io(mdev
, sector
, size
);
4272 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4273 _ack_id_to_req
, __func__
, neg_acked
);
4276 static int got_NegDReply(struct drbd_conf
*mdev
, struct p_header80
*h
)
4278 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4279 sector_t sector
= be64_to_cpu(p
->sector
);
4281 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4282 dev_err(DEV
, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4283 (unsigned long long)sector
, be32_to_cpu(p
->blksize
));
4285 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4286 _ar_id_to_req
, __func__
, neg_acked
);
4289 static int got_NegRSDReply(struct drbd_conf
*mdev
, struct p_header80
*h
)
4293 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4295 sector
= be64_to_cpu(p
->sector
);
4296 size
= be32_to_cpu(p
->blksize
);
4298 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4300 dec_rs_pending(mdev
);
4302 if (get_ldev_if_state(mdev
, D_FAILED
)) {
4303 drbd_rs_complete_io(mdev
, sector
);
4304 drbd_rs_failed_io(mdev
, sector
, size
);
4311 static int got_BarrierAck(struct drbd_conf
*mdev
, struct p_header80
*h
)
4313 struct p_barrier_ack
*p
= (struct p_barrier_ack
*)h
;
4315 tl_release(mdev
, p
->barrier
, be32_to_cpu(p
->set_size
));
4320 static int got_OVResult(struct drbd_conf
*mdev
, struct p_header80
*h
)
4322 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4323 struct drbd_work
*w
;
4327 sector
= be64_to_cpu(p
->sector
);
4328 size
= be32_to_cpu(p
->blksize
);
4330 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4332 if (be64_to_cpu(p
->block_id
) == ID_OUT_OF_SYNC
)
4333 drbd_ov_oos_found(mdev
, sector
, size
);
4337 if (!get_ldev(mdev
))
4340 drbd_rs_complete_io(mdev
, sector
);
4341 dec_rs_pending(mdev
);
4343 if (--mdev
->ov_left
== 0) {
4344 w
= kmalloc(sizeof(*w
), GFP_NOIO
);
4346 w
->cb
= w_ov_finished
;
4347 drbd_queue_work_front(&mdev
->data
.work
, w
);
4349 dev_err(DEV
, "kmalloc(w) failed.");
4351 drbd_resync_finished(mdev
);
4358 static int got_skip(struct drbd_conf
*mdev
, struct p_header80
*h
)
4363 struct asender_cmd
{
4365 int (*process
)(struct drbd_conf
*mdev
, struct p_header80
*h
);
4368 static struct asender_cmd
*get_asender_cmd(int cmd
)
4370 static struct asender_cmd asender_tbl
[] = {
4371 /* anything missing from this table is in
4372 * the drbd_cmd_handler (drbd_default_handler) table,
4373 * see the beginning of drbdd() */
4374 [P_PING
] = { sizeof(struct p_header80
), got_Ping
},
4375 [P_PING_ACK
] = { sizeof(struct p_header80
), got_PingAck
},
4376 [P_RECV_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4377 [P_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4378 [P_RS_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4379 [P_DISCARD_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4380 [P_NEG_ACK
] = { sizeof(struct p_block_ack
), got_NegAck
},
4381 [P_NEG_DREPLY
] = { sizeof(struct p_block_ack
), got_NegDReply
},
4382 [P_NEG_RS_DREPLY
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
4383 [P_OV_RESULT
] = { sizeof(struct p_block_ack
), got_OVResult
},
4384 [P_BARRIER_ACK
] = { sizeof(struct p_barrier_ack
), got_BarrierAck
},
4385 [P_STATE_CHG_REPLY
] = { sizeof(struct p_req_state_reply
), got_RqSReply
},
4386 [P_RS_IS_IN_SYNC
] = { sizeof(struct p_block_ack
), got_IsInSync
},
4387 [P_DELAY_PROBE
] = { sizeof(struct p_delay_probe93
), got_skip
},
4388 [P_MAX_CMD
] = { 0, NULL
},
4390 if (cmd
> P_MAX_CMD
|| asender_tbl
[cmd
].process
== NULL
)
4392 return &asender_tbl
[cmd
];
4395 int drbd_asender(struct drbd_thread
*thi
)
4397 struct drbd_conf
*mdev
= thi
->mdev
;
4398 struct p_header80
*h
= &mdev
->meta
.rbuf
.header
.h80
;
4399 struct asender_cmd
*cmd
= NULL
;
4404 int expect
= sizeof(struct p_header80
);
4407 sprintf(current
->comm
, "drbd%d_asender", mdev_to_minor(mdev
));
4409 current
->policy
= SCHED_RR
; /* Make this a realtime task! */
4410 current
->rt_priority
= 2; /* more important than all other tasks */
4412 while (get_t_state(thi
) == Running
) {
4413 drbd_thread_current_set_cpu(mdev
);
4414 if (test_and_clear_bit(SEND_PING
, &mdev
->flags
)) {
4415 ERR_IF(!drbd_send_ping(mdev
)) goto reconnect
;
4416 mdev
->meta
.socket
->sk
->sk_rcvtimeo
=
4417 mdev
->net_conf
->ping_timeo
*HZ
/10;
4420 /* conditionally cork;
4421 * it may hurt latency if we cork without much to send */
4422 if (!mdev
->net_conf
->no_cork
&&
4423 3 < atomic_read(&mdev
->unacked_cnt
))
4424 drbd_tcp_cork(mdev
->meta
.socket
);
4426 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4427 flush_signals(current
);
4428 if (!drbd_process_done_ee(mdev
))
4430 /* to avoid race with newly queued ACKs */
4431 set_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4432 spin_lock_irq(&mdev
->req_lock
);
4433 empty
= list_empty(&mdev
->done_ee
);
4434 spin_unlock_irq(&mdev
->req_lock
);
4435 /* new ack may have been queued right here,
4436 * but then there is also a signal pending,
4437 * and we start over... */
4441 /* but unconditionally uncork unless disabled */
4442 if (!mdev
->net_conf
->no_cork
)
4443 drbd_tcp_uncork(mdev
->meta
.socket
);
4445 /* short circuit, recv_msg would return EINTR anyways. */
4446 if (signal_pending(current
))
4449 rv
= drbd_recv_short(mdev
, mdev
->meta
.socket
,
4450 buf
, expect
-received
, 0);
4451 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4453 flush_signals(current
);
4456 * -EINTR (on meta) we got a signal
4457 * -EAGAIN (on meta) rcvtimeo expired
4458 * -ECONNRESET other side closed the connection
4459 * -ERESTARTSYS (on data) we got a signal
4460 * rv < 0 other than above: unexpected error!
4461 * rv == expected: full header or command
4462 * rv < expected: "woken" by signal during receive
4463 * rv == 0 : "connection shut down by peer"
4465 if (likely(rv
> 0)) {
4468 } else if (rv
== 0) {
4469 dev_err(DEV
, "meta connection shut down by peer.\n");
4471 } else if (rv
== -EAGAIN
) {
4472 if (mdev
->meta
.socket
->sk
->sk_rcvtimeo
==
4473 mdev
->net_conf
->ping_timeo
*HZ
/10) {
4474 dev_err(DEV
, "PingAck did not arrive in time.\n");
4477 set_bit(SEND_PING
, &mdev
->flags
);
4479 } else if (rv
== -EINTR
) {
4482 dev_err(DEV
, "sock_recvmsg returned %d\n", rv
);
4486 if (received
== expect
&& cmd
== NULL
) {
4487 if (unlikely(h
->magic
!= BE_DRBD_MAGIC
)) {
4488 dev_err(DEV
, "magic?? on meta m: 0x%08x c: %d l: %d\n",
4489 be32_to_cpu(h
->magic
),
4490 be16_to_cpu(h
->command
),
4491 be16_to_cpu(h
->length
));
4494 cmd
= get_asender_cmd(be16_to_cpu(h
->command
));
4495 len
= be16_to_cpu(h
->length
);
4496 if (unlikely(cmd
== NULL
)) {
4497 dev_err(DEV
, "unknown command?? on meta m: 0x%08x c: %d l: %d\n",
4498 be32_to_cpu(h
->magic
),
4499 be16_to_cpu(h
->command
),
4500 be16_to_cpu(h
->length
));
4503 expect
= cmd
->pkt_size
;
4504 ERR_IF(len
!= expect
-sizeof(struct p_header80
))
4507 if (received
== expect
) {
4508 D_ASSERT(cmd
!= NULL
);
4509 if (!cmd
->process(mdev
, h
))
4514 expect
= sizeof(struct p_header80
);
4521 drbd_force_state(mdev
, NS(conn
, C_NETWORK_FAILURE
));
4526 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
4529 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4531 D_ASSERT(mdev
->state
.conn
< C_CONNECTED
);
4532 dev_info(DEV
, "asender terminated\n");