4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
28 #include <asm/uaccess.h>
31 #include <linux/drbd.h>
33 #include <linux/file.h>
36 #include <linux/memcontrol.h>
37 #include <linux/mm_inline.h>
38 #include <linux/slab.h>
39 #include <linux/smp_lock.h>
40 #include <linux/pkt_sched.h>
41 #define __KERNEL_SYSCALLS__
42 #include <linux/unistd.h>
43 #include <linux/vmalloc.h>
44 #include <linux/random.h>
45 #include <linux/string.h>
46 #include <linux/scatterlist.h>
54 struct drbd_epoch
*epoch
;
63 static int drbd_do_handshake(struct drbd_conf
*mdev
);
64 static int drbd_do_auth(struct drbd_conf
*mdev
);
66 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*, struct drbd_epoch
*, enum epoch_event
);
67 static int e_end_block(struct drbd_conf
*, struct drbd_work
*, int);
69 static struct drbd_epoch
*previous_epoch(struct drbd_conf
*mdev
, struct drbd_epoch
*epoch
)
71 struct drbd_epoch
*prev
;
72 spin_lock(&mdev
->epoch_lock
);
73 prev
= list_entry(epoch
->list
.prev
, struct drbd_epoch
, list
);
74 if (prev
== epoch
|| prev
== mdev
->current_epoch
)
76 spin_unlock(&mdev
->epoch_lock
);
80 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
83 * some helper functions to deal with single linked page lists,
84 * page->private being our "next" pointer.
87 /* If at least n pages are linked at head, get n pages off.
88 * Otherwise, don't modify head, and return NULL.
89 * Locking is the responsibility of the caller.
91 static struct page
*page_chain_del(struct page
**head
, int n
)
105 tmp
= page_chain_next(page
);
107 break; /* found sufficient pages */
109 /* insufficient pages, don't use any of them. */
114 /* add end of list marker for the returned list */
115 set_page_private(page
, 0);
116 /* actual return value, and adjustment of head */
122 /* may be used outside of locks to find the tail of a (usually short)
123 * "private" page chain, before adding it back to a global chain head
124 * with page_chain_add() under a spinlock. */
125 static struct page
*page_chain_tail(struct page
*page
, int *len
)
129 while ((tmp
= page_chain_next(page
)))
136 static int page_chain_free(struct page
*page
)
140 page_chain_for_each_safe(page
, tmp
) {
147 static void page_chain_add(struct page
**head
,
148 struct page
*chain_first
, struct page
*chain_last
)
152 tmp
= page_chain_tail(chain_first
, NULL
);
153 BUG_ON(tmp
!= chain_last
);
156 /* add chain to head */
157 set_page_private(chain_last
, (unsigned long)*head
);
161 static struct page
*drbd_pp_first_pages_or_try_alloc(struct drbd_conf
*mdev
, int number
)
163 struct page
*page
= NULL
;
164 struct page
*tmp
= NULL
;
167 /* Yes, testing drbd_pp_vacant outside the lock is racy.
168 * So what. It saves a spin_lock. */
169 if (drbd_pp_vacant
>= number
) {
170 spin_lock(&drbd_pp_lock
);
171 page
= page_chain_del(&drbd_pp_pool
, number
);
173 drbd_pp_vacant
-= number
;
174 spin_unlock(&drbd_pp_lock
);
179 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
180 * "criss-cross" setup, that might cause write-out on some other DRBD,
181 * which in turn might block on the other node at this very place. */
182 for (i
= 0; i
< number
; i
++) {
183 tmp
= alloc_page(GFP_TRY
);
186 set_page_private(tmp
, (unsigned long)page
);
193 /* Not enough pages immediately available this time.
194 * No need to jump around here, drbd_pp_alloc will retry this
195 * function "soon". */
197 tmp
= page_chain_tail(page
, NULL
);
198 spin_lock(&drbd_pp_lock
);
199 page_chain_add(&drbd_pp_pool
, page
, tmp
);
201 spin_unlock(&drbd_pp_lock
);
206 /* kick lower level device, if we have more than (arbitrary number)
207 * reference counts on it, which typically are locally submitted io
208 * requests. don't use unacked_cnt, so we speed up proto A and B, too. */
209 static void maybe_kick_lo(struct drbd_conf
*mdev
)
211 if (atomic_read(&mdev
->local_cnt
) >= mdev
->net_conf
->unplug_watermark
)
215 static void reclaim_net_ee(struct drbd_conf
*mdev
, struct list_head
*to_be_freed
)
217 struct drbd_epoch_entry
*e
;
218 struct list_head
*le
, *tle
;
220 /* The EEs are always appended to the end of the list. Since
221 they are sent in order over the wire, they have to finish
222 in order. As soon as we see the first not finished we can
223 stop to examine the list... */
225 list_for_each_safe(le
, tle
, &mdev
->net_ee
) {
226 e
= list_entry(le
, struct drbd_epoch_entry
, w
.list
);
227 if (drbd_ee_has_active_page(e
))
229 list_move(le
, to_be_freed
);
233 static void drbd_kick_lo_and_reclaim_net(struct drbd_conf
*mdev
)
235 LIST_HEAD(reclaimed
);
236 struct drbd_epoch_entry
*e
, *t
;
239 spin_lock_irq(&mdev
->req_lock
);
240 reclaim_net_ee(mdev
, &reclaimed
);
241 spin_unlock_irq(&mdev
->req_lock
);
243 list_for_each_entry_safe(e
, t
, &reclaimed
, w
.list
)
244 drbd_free_ee(mdev
, e
);
248 * drbd_pp_alloc() - Returns @number pages, retries forever (or until signalled)
249 * @mdev: DRBD device.
250 * @number: number of pages requested
251 * @retry: whether to retry, if not enough pages are available right now
253 * Tries to allocate number pages, first from our own page pool, then from
254 * the kernel, unless this allocation would exceed the max_buffers setting.
255 * Possibly retry until DRBD frees sufficient pages somewhere else.
257 * Returns a page chain linked via page->private.
259 static struct page
*drbd_pp_alloc(struct drbd_conf
*mdev
, unsigned number
, bool retry
)
261 struct page
*page
= NULL
;
264 /* Yes, we may run up to @number over max_buffers. If we
265 * follow it strictly, the admin will get it wrong anyways. */
266 if (atomic_read(&mdev
->pp_in_use
) < mdev
->net_conf
->max_buffers
)
267 page
= drbd_pp_first_pages_or_try_alloc(mdev
, number
);
269 while (page
== NULL
) {
270 prepare_to_wait(&drbd_pp_wait
, &wait
, TASK_INTERRUPTIBLE
);
272 drbd_kick_lo_and_reclaim_net(mdev
);
274 if (atomic_read(&mdev
->pp_in_use
) < mdev
->net_conf
->max_buffers
) {
275 page
= drbd_pp_first_pages_or_try_alloc(mdev
, number
);
283 if (signal_pending(current
)) {
284 dev_warn(DEV
, "drbd_pp_alloc interrupted!\n");
290 finish_wait(&drbd_pp_wait
, &wait
);
293 atomic_add(number
, &mdev
->pp_in_use
);
297 /* Must not be used from irq, as that may deadlock: see drbd_pp_alloc.
298 * Is also used from inside an other spin_lock_irq(&mdev->req_lock);
299 * Either links the page chain back to the global pool,
300 * or returns all pages to the system. */
301 static void drbd_pp_free(struct drbd_conf
*mdev
, struct page
*page
)
304 if (drbd_pp_vacant
> (DRBD_MAX_SEGMENT_SIZE
/PAGE_SIZE
)*minor_count
)
305 i
= page_chain_free(page
);
308 tmp
= page_chain_tail(page
, &i
);
309 spin_lock(&drbd_pp_lock
);
310 page_chain_add(&drbd_pp_pool
, page
, tmp
);
312 spin_unlock(&drbd_pp_lock
);
314 atomic_sub(i
, &mdev
->pp_in_use
);
315 i
= atomic_read(&mdev
->pp_in_use
);
317 dev_warn(DEV
, "ASSERTION FAILED: pp_in_use: %d < 0\n", i
);
318 wake_up(&drbd_pp_wait
);
322 You need to hold the req_lock:
323 _drbd_wait_ee_list_empty()
325 You must not have the req_lock:
331 drbd_process_done_ee()
333 drbd_wait_ee_list_empty()
336 struct drbd_epoch_entry
*drbd_alloc_ee(struct drbd_conf
*mdev
,
339 unsigned int data_size
,
340 gfp_t gfp_mask
) __must_hold(local
)
342 struct drbd_epoch_entry
*e
;
344 unsigned nr_pages
= (data_size
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
346 if (FAULT_ACTIVE(mdev
, DRBD_FAULT_AL_EE
))
349 e
= mempool_alloc(drbd_ee_mempool
, gfp_mask
& ~__GFP_HIGHMEM
);
351 if (!(gfp_mask
& __GFP_NOWARN
))
352 dev_err(DEV
, "alloc_ee: Allocation of an EE failed\n");
356 page
= drbd_pp_alloc(mdev
, nr_pages
, (gfp_mask
& __GFP_WAIT
));
360 INIT_HLIST_NODE(&e
->colision
);
364 atomic_set(&e
->pending_bios
, 0);
374 mempool_free(e
, drbd_ee_mempool
);
378 void drbd_free_ee(struct drbd_conf
*mdev
, struct drbd_epoch_entry
*e
)
380 drbd_pp_free(mdev
, e
->pages
);
381 D_ASSERT(atomic_read(&e
->pending_bios
) == 0);
382 D_ASSERT(hlist_unhashed(&e
->colision
));
383 mempool_free(e
, drbd_ee_mempool
);
386 int drbd_release_ee(struct drbd_conf
*mdev
, struct list_head
*list
)
388 LIST_HEAD(work_list
);
389 struct drbd_epoch_entry
*e
, *t
;
392 spin_lock_irq(&mdev
->req_lock
);
393 list_splice_init(list
, &work_list
);
394 spin_unlock_irq(&mdev
->req_lock
);
396 list_for_each_entry_safe(e
, t
, &work_list
, w
.list
) {
397 drbd_free_ee(mdev
, e
);
405 * This function is called from _asender only_
406 * but see also comments in _req_mod(,barrier_acked)
407 * and receive_Barrier.
409 * Move entries from net_ee to done_ee, if ready.
410 * Grab done_ee, call all callbacks, free the entries.
411 * The callbacks typically send out ACKs.
413 static int drbd_process_done_ee(struct drbd_conf
*mdev
)
415 LIST_HEAD(work_list
);
416 LIST_HEAD(reclaimed
);
417 struct drbd_epoch_entry
*e
, *t
;
418 int ok
= (mdev
->state
.conn
>= C_WF_REPORT_PARAMS
);
420 spin_lock_irq(&mdev
->req_lock
);
421 reclaim_net_ee(mdev
, &reclaimed
);
422 list_splice_init(&mdev
->done_ee
, &work_list
);
423 spin_unlock_irq(&mdev
->req_lock
);
425 list_for_each_entry_safe(e
, t
, &reclaimed
, w
.list
)
426 drbd_free_ee(mdev
, e
);
428 /* possible callbacks here:
429 * e_end_block, and e_end_resync_block, e_send_discard_ack.
430 * all ignore the last argument.
432 list_for_each_entry_safe(e
, t
, &work_list
, w
.list
) {
433 /* list_del not necessary, next/prev members not touched */
434 ok
= e
->w
.cb(mdev
, &e
->w
, !ok
) && ok
;
435 drbd_free_ee(mdev
, e
);
437 wake_up(&mdev
->ee_wait
);
442 void _drbd_wait_ee_list_empty(struct drbd_conf
*mdev
, struct list_head
*head
)
446 /* avoids spin_lock/unlock
447 * and calling prepare_to_wait in the fast path */
448 while (!list_empty(head
)) {
449 prepare_to_wait(&mdev
->ee_wait
, &wait
, TASK_UNINTERRUPTIBLE
);
450 spin_unlock_irq(&mdev
->req_lock
);
453 finish_wait(&mdev
->ee_wait
, &wait
);
454 spin_lock_irq(&mdev
->req_lock
);
458 void drbd_wait_ee_list_empty(struct drbd_conf
*mdev
, struct list_head
*head
)
460 spin_lock_irq(&mdev
->req_lock
);
461 _drbd_wait_ee_list_empty(mdev
, head
);
462 spin_unlock_irq(&mdev
->req_lock
);
465 /* see also kernel_accept; which is only present since 2.6.18.
466 * also we want to log which part of it failed, exactly */
467 static int drbd_accept(struct drbd_conf
*mdev
, const char **what
,
468 struct socket
*sock
, struct socket
**newsock
)
470 struct sock
*sk
= sock
->sk
;
474 err
= sock
->ops
->listen(sock
, 5);
478 *what
= "sock_create_lite";
479 err
= sock_create_lite(sk
->sk_family
, sk
->sk_type
, sk
->sk_protocol
,
485 err
= sock
->ops
->accept(sock
, *newsock
, 0);
487 sock_release(*newsock
);
491 (*newsock
)->ops
= sock
->ops
;
497 static int drbd_recv_short(struct drbd_conf
*mdev
, struct socket
*sock
,
498 void *buf
, size_t size
, int flags
)
505 struct msghdr msg
= {
507 .msg_iov
= (struct iovec
*)&iov
,
508 .msg_flags
= (flags
? flags
: MSG_WAITALL
| MSG_NOSIGNAL
)
514 rv
= sock_recvmsg(sock
, &msg
, size
, msg
.msg_flags
);
520 static int drbd_recv(struct drbd_conf
*mdev
, void *buf
, size_t size
)
527 struct msghdr msg
= {
529 .msg_iov
= (struct iovec
*)&iov
,
530 .msg_flags
= MSG_WAITALL
| MSG_NOSIGNAL
538 rv
= sock_recvmsg(mdev
->data
.socket
, &msg
, size
, msg
.msg_flags
);
543 * ECONNRESET other side closed the connection
544 * ERESTARTSYS (on sock) we got a signal
548 if (rv
== -ECONNRESET
)
549 dev_info(DEV
, "sock was reset by peer\n");
550 else if (rv
!= -ERESTARTSYS
)
551 dev_err(DEV
, "sock_recvmsg returned %d\n", rv
);
553 } else if (rv
== 0) {
554 dev_info(DEV
, "sock was shut down by peer\n");
557 /* signal came in, or peer/link went down,
558 * after we read a partial message
560 /* D_ASSERT(signal_pending(current)); */
568 drbd_force_state(mdev
, NS(conn
, C_BROKEN_PIPE
));
574 * On individual connections, the socket buffer size must be set prior to the
575 * listen(2) or connect(2) calls in order to have it take effect.
576 * This is our wrapper to do so.
578 static void drbd_setbufsize(struct socket
*sock
, unsigned int snd
,
581 /* open coded SO_SNDBUF, SO_RCVBUF */
583 sock
->sk
->sk_sndbuf
= snd
;
584 sock
->sk
->sk_userlocks
|= SOCK_SNDBUF_LOCK
;
587 sock
->sk
->sk_rcvbuf
= rcv
;
588 sock
->sk
->sk_userlocks
|= SOCK_RCVBUF_LOCK
;
592 static struct socket
*drbd_try_connect(struct drbd_conf
*mdev
)
596 struct sockaddr_in6 src_in6
;
598 int disconnect_on_error
= 1;
600 if (!get_net_conf(mdev
))
603 what
= "sock_create_kern";
604 err
= sock_create_kern(((struct sockaddr
*)mdev
->net_conf
->my_addr
)->sa_family
,
605 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
611 sock
->sk
->sk_rcvtimeo
=
612 sock
->sk
->sk_sndtimeo
= mdev
->net_conf
->try_connect_int
*HZ
;
613 drbd_setbufsize(sock
, mdev
->net_conf
->sndbuf_size
,
614 mdev
->net_conf
->rcvbuf_size
);
616 /* explicitly bind to the configured IP as source IP
617 * for the outgoing connections.
618 * This is needed for multihomed hosts and to be
619 * able to use lo: interfaces for drbd.
620 * Make sure to use 0 as port number, so linux selects
621 * a free one dynamically.
623 memcpy(&src_in6
, mdev
->net_conf
->my_addr
,
624 min_t(int, mdev
->net_conf
->my_addr_len
, sizeof(src_in6
)));
625 if (((struct sockaddr
*)mdev
->net_conf
->my_addr
)->sa_family
== AF_INET6
)
626 src_in6
.sin6_port
= 0;
628 ((struct sockaddr_in
*)&src_in6
)->sin_port
= 0; /* AF_INET & AF_SCI */
630 what
= "bind before connect";
631 err
= sock
->ops
->bind(sock
,
632 (struct sockaddr
*) &src_in6
,
633 mdev
->net_conf
->my_addr_len
);
637 /* connect may fail, peer not yet available.
638 * stay C_WF_CONNECTION, don't go Disconnecting! */
639 disconnect_on_error
= 0;
641 err
= sock
->ops
->connect(sock
,
642 (struct sockaddr
*)mdev
->net_conf
->peer_addr
,
643 mdev
->net_conf
->peer_addr_len
, 0);
652 /* timeout, busy, signal pending */
653 case ETIMEDOUT
: case EAGAIN
: case EINPROGRESS
:
654 case EINTR
: case ERESTARTSYS
:
655 /* peer not (yet) available, network problem */
656 case ECONNREFUSED
: case ENETUNREACH
:
657 case EHOSTDOWN
: case EHOSTUNREACH
:
658 disconnect_on_error
= 0;
661 dev_err(DEV
, "%s failed, err = %d\n", what
, err
);
663 if (disconnect_on_error
)
664 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
670 static struct socket
*drbd_wait_for_connect(struct drbd_conf
*mdev
)
673 struct socket
*s_estab
= NULL
, *s_listen
;
676 if (!get_net_conf(mdev
))
679 what
= "sock_create_kern";
680 err
= sock_create_kern(((struct sockaddr
*)mdev
->net_conf
->my_addr
)->sa_family
,
681 SOCK_STREAM
, IPPROTO_TCP
, &s_listen
);
687 timeo
= mdev
->net_conf
->try_connect_int
* HZ
;
688 timeo
+= (random32() & 1) ? timeo
/ 7 : -timeo
/ 7; /* 28.5% random jitter */
690 s_listen
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
691 s_listen
->sk
->sk_rcvtimeo
= timeo
;
692 s_listen
->sk
->sk_sndtimeo
= timeo
;
693 drbd_setbufsize(s_listen
, mdev
->net_conf
->sndbuf_size
,
694 mdev
->net_conf
->rcvbuf_size
);
696 what
= "bind before listen";
697 err
= s_listen
->ops
->bind(s_listen
,
698 (struct sockaddr
*) mdev
->net_conf
->my_addr
,
699 mdev
->net_conf
->my_addr_len
);
703 err
= drbd_accept(mdev
, &what
, s_listen
, &s_estab
);
707 sock_release(s_listen
);
709 if (err
!= -EAGAIN
&& err
!= -EINTR
&& err
!= -ERESTARTSYS
) {
710 dev_err(DEV
, "%s failed, err = %d\n", what
, err
);
711 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
719 static int drbd_send_fp(struct drbd_conf
*mdev
,
720 struct socket
*sock
, enum drbd_packets cmd
)
722 struct p_header
*h
= (struct p_header
*) &mdev
->data
.sbuf
.header
;
724 return _drbd_send_cmd(mdev
, sock
, cmd
, h
, sizeof(*h
), 0);
727 static enum drbd_packets
drbd_recv_fp(struct drbd_conf
*mdev
, struct socket
*sock
)
729 struct p_header
*h
= (struct p_header
*) &mdev
->data
.sbuf
.header
;
732 rr
= drbd_recv_short(mdev
, sock
, h
, sizeof(*h
), 0);
734 if (rr
== sizeof(*h
) && h
->magic
== BE_DRBD_MAGIC
)
735 return be16_to_cpu(h
->command
);
741 * drbd_socket_okay() - Free the socket if its connection is not okay
742 * @mdev: DRBD device.
743 * @sock: pointer to the pointer to the socket.
745 static int drbd_socket_okay(struct drbd_conf
*mdev
, struct socket
**sock
)
753 rr
= drbd_recv_short(mdev
, *sock
, tb
, 4, MSG_DONTWAIT
| MSG_PEEK
);
755 if (rr
> 0 || rr
== -EAGAIN
) {
766 * 1 yes, we have a valid connection
767 * 0 oops, did not work out, please try again
768 * -1 peer talks different language,
769 * no point in trying again, please go standalone.
770 * -2 We do not have a network config...
772 static int drbd_connect(struct drbd_conf
*mdev
)
774 struct socket
*s
, *sock
, *msock
;
777 D_ASSERT(!mdev
->data
.socket
);
779 if (drbd_request_state(mdev
, NS(conn
, C_WF_CONNECTION
)) < SS_SUCCESS
)
782 clear_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
789 /* 3 tries, this should take less than a second! */
790 s
= drbd_try_connect(mdev
);
793 /* give the other side time to call bind() & listen() */
794 __set_current_state(TASK_INTERRUPTIBLE
);
795 schedule_timeout(HZ
/ 10);
800 drbd_send_fp(mdev
, s
, P_HAND_SHAKE_S
);
804 drbd_send_fp(mdev
, s
, P_HAND_SHAKE_M
);
808 dev_err(DEV
, "Logic error in drbd_connect()\n");
809 goto out_release_sockets
;
814 __set_current_state(TASK_INTERRUPTIBLE
);
815 schedule_timeout(HZ
/ 10);
816 ok
= drbd_socket_okay(mdev
, &sock
);
817 ok
= drbd_socket_okay(mdev
, &msock
) && ok
;
823 s
= drbd_wait_for_connect(mdev
);
825 try = drbd_recv_fp(mdev
, s
);
826 drbd_socket_okay(mdev
, &sock
);
827 drbd_socket_okay(mdev
, &msock
);
831 dev_warn(DEV
, "initial packet S crossed\n");
838 dev_warn(DEV
, "initial packet M crossed\n");
842 set_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
845 dev_warn(DEV
, "Error receiving initial packet\n");
852 if (mdev
->state
.conn
<= C_DISCONNECTING
)
853 goto out_release_sockets
;
854 if (signal_pending(current
)) {
855 flush_signals(current
);
857 if (get_t_state(&mdev
->receiver
) == Exiting
)
858 goto out_release_sockets
;
862 ok
= drbd_socket_okay(mdev
, &sock
);
863 ok
= drbd_socket_okay(mdev
, &msock
) && ok
;
869 msock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
870 sock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
872 sock
->sk
->sk_allocation
= GFP_NOIO
;
873 msock
->sk
->sk_allocation
= GFP_NOIO
;
875 sock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE_BULK
;
876 msock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE
;
879 * sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
880 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
881 * first set it to the P_HAND_SHAKE timeout,
882 * which we set to 4x the configured ping_timeout. */
883 sock
->sk
->sk_sndtimeo
=
884 sock
->sk
->sk_rcvtimeo
= mdev
->net_conf
->ping_timeo
*4*HZ
/10;
886 msock
->sk
->sk_sndtimeo
= mdev
->net_conf
->timeout
*HZ
/10;
887 msock
->sk
->sk_rcvtimeo
= mdev
->net_conf
->ping_int
*HZ
;
889 /* we don't want delays.
890 * we use TCP_CORK where apropriate, though */
891 drbd_tcp_nodelay(sock
);
892 drbd_tcp_nodelay(msock
);
894 mdev
->data
.socket
= sock
;
895 mdev
->meta
.socket
= msock
;
896 mdev
->last_received
= jiffies
;
898 D_ASSERT(mdev
->asender
.task
== NULL
);
900 h
= drbd_do_handshake(mdev
);
904 if (mdev
->cram_hmac_tfm
) {
905 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
906 switch (drbd_do_auth(mdev
)) {
908 dev_err(DEV
, "Authentication of peer failed\n");
911 dev_err(DEV
, "Authentication of peer failed, trying again.\n");
916 if (drbd_request_state(mdev
, NS(conn
, C_WF_REPORT_PARAMS
)) < SS_SUCCESS
)
919 sock
->sk
->sk_sndtimeo
= mdev
->net_conf
->timeout
*HZ
/10;
920 sock
->sk
->sk_rcvtimeo
= MAX_SCHEDULE_TIMEOUT
;
922 atomic_set(&mdev
->packet_seq
, 0);
925 drbd_thread_start(&mdev
->asender
);
927 if (!drbd_send_protocol(mdev
))
929 drbd_send_sync_param(mdev
, &mdev
->sync_conf
);
930 drbd_send_sizes(mdev
, 0, 0);
931 drbd_send_uuids(mdev
);
932 drbd_send_state(mdev
);
933 clear_bit(USE_DEGR_WFC_T
, &mdev
->flags
);
934 clear_bit(RESIZE_PENDING
, &mdev
->flags
);
946 static int drbd_recv_header(struct drbd_conf
*mdev
, struct p_header
*h
)
950 r
= drbd_recv(mdev
, h
, sizeof(*h
));
952 if (unlikely(r
!= sizeof(*h
))) {
953 dev_err(DEV
, "short read expecting header on sock: r=%d\n", r
);
956 h
->command
= be16_to_cpu(h
->command
);
957 h
->length
= be16_to_cpu(h
->length
);
958 if (unlikely(h
->magic
!= BE_DRBD_MAGIC
)) {
959 dev_err(DEV
, "magic?? on data m: 0x%lx c: %d l: %d\n",
960 (long)be32_to_cpu(h
->magic
),
961 h
->command
, h
->length
);
964 mdev
->last_received
= jiffies
;
969 static enum finish_epoch
drbd_flush_after_epoch(struct drbd_conf
*mdev
, struct drbd_epoch
*epoch
)
973 if (mdev
->write_ordering
>= WO_bdev_flush
&& get_ldev(mdev
)) {
974 rv
= blkdev_issue_flush(mdev
->ldev
->backing_bdev
, GFP_KERNEL
,
975 NULL
, BLKDEV_IFL_WAIT
);
977 dev_err(DEV
, "local disk flush failed with status %d\n", rv
);
978 /* would rather check on EOPNOTSUPP, but that is not reliable.
979 * don't try again for ANY return value != 0
980 * if (rv == -EOPNOTSUPP) */
981 drbd_bump_write_ordering(mdev
, WO_drain_io
);
986 return drbd_may_finish_epoch(mdev
, epoch
, EV_BARRIER_DONE
);
989 static int w_flush(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
991 struct flush_work
*fw
= (struct flush_work
*)w
;
992 struct drbd_epoch
*epoch
= fw
->epoch
;
996 if (!test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED
, &epoch
->flags
))
997 drbd_flush_after_epoch(mdev
, epoch
);
999 drbd_may_finish_epoch(mdev
, epoch
, EV_PUT
|
1000 (mdev
->state
.conn
< C_CONNECTED
? EV_CLEANUP
: 0));
1006 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1007 * @mdev: DRBD device.
1008 * @epoch: Epoch object.
1011 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*mdev
,
1012 struct drbd_epoch
*epoch
,
1013 enum epoch_event ev
)
1015 int finish
, epoch_size
;
1016 struct drbd_epoch
*next_epoch
;
1017 int schedule_flush
= 0;
1018 enum finish_epoch rv
= FE_STILL_LIVE
;
1020 spin_lock(&mdev
->epoch_lock
);
1025 epoch_size
= atomic_read(&epoch
->epoch_size
);
1027 switch (ev
& ~EV_CLEANUP
) {
1029 atomic_dec(&epoch
->active
);
1031 case EV_GOT_BARRIER_NR
:
1032 set_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
);
1034 /* Special case: If we just switched from WO_bio_barrier to
1035 WO_bdev_flush we should not finish the current epoch */
1036 if (test_bit(DE_CONTAINS_A_BARRIER
, &epoch
->flags
) && epoch_size
== 1 &&
1037 mdev
->write_ordering
!= WO_bio_barrier
&&
1038 epoch
== mdev
->current_epoch
)
1039 clear_bit(DE_CONTAINS_A_BARRIER
, &epoch
->flags
);
1041 case EV_BARRIER_DONE
:
1042 set_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE
, &epoch
->flags
);
1044 case EV_BECAME_LAST
:
1049 if (epoch_size
!= 0 &&
1050 atomic_read(&epoch
->active
) == 0 &&
1051 test_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
) &&
1052 epoch
->list
.prev
== &mdev
->current_epoch
->list
&&
1053 !test_bit(DE_IS_FINISHING
, &epoch
->flags
)) {
1054 /* Nearly all conditions are met to finish that epoch... */
1055 if (test_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE
, &epoch
->flags
) ||
1056 mdev
->write_ordering
== WO_none
||
1057 (epoch_size
== 1 && test_bit(DE_CONTAINS_A_BARRIER
, &epoch
->flags
)) ||
1060 set_bit(DE_IS_FINISHING
, &epoch
->flags
);
1061 } else if (!test_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED
, &epoch
->flags
) &&
1062 mdev
->write_ordering
== WO_bio_barrier
) {
1063 atomic_inc(&epoch
->active
);
1068 if (!(ev
& EV_CLEANUP
)) {
1069 spin_unlock(&mdev
->epoch_lock
);
1070 drbd_send_b_ack(mdev
, epoch
->barrier_nr
, epoch_size
);
1071 spin_lock(&mdev
->epoch_lock
);
1075 if (mdev
->current_epoch
!= epoch
) {
1076 next_epoch
= list_entry(epoch
->list
.next
, struct drbd_epoch
, list
);
1077 list_del(&epoch
->list
);
1078 ev
= EV_BECAME_LAST
| (ev
& EV_CLEANUP
);
1082 if (rv
== FE_STILL_LIVE
)
1086 atomic_set(&epoch
->epoch_size
, 0);
1087 /* atomic_set(&epoch->active, 0); is already zero */
1088 if (rv
== FE_STILL_LIVE
)
1099 spin_unlock(&mdev
->epoch_lock
);
1101 if (schedule_flush
) {
1102 struct flush_work
*fw
;
1103 fw
= kmalloc(sizeof(*fw
), GFP_ATOMIC
);
1107 drbd_queue_work(&mdev
->data
.work
, &fw
->w
);
1109 dev_warn(DEV
, "Could not kmalloc a flush_work obj\n");
1110 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED
, &epoch
->flags
);
1111 /* That is not a recursion, only one level */
1112 drbd_may_finish_epoch(mdev
, epoch
, EV_BARRIER_DONE
);
1113 drbd_may_finish_epoch(mdev
, epoch
, EV_PUT
);
1121 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1122 * @mdev: DRBD device.
1123 * @wo: Write ordering method to try.
1125 void drbd_bump_write_ordering(struct drbd_conf
*mdev
, enum write_ordering_e wo
) __must_hold(local
)
1127 enum write_ordering_e pwo
;
1128 static char *write_ordering_str
[] = {
1130 [WO_drain_io
] = "drain",
1131 [WO_bdev_flush
] = "flush",
1132 [WO_bio_barrier
] = "barrier",
1135 pwo
= mdev
->write_ordering
;
1137 if (wo
== WO_bio_barrier
&& mdev
->ldev
->dc
.no_disk_barrier
)
1139 if (wo
== WO_bdev_flush
&& mdev
->ldev
->dc
.no_disk_flush
)
1141 if (wo
== WO_drain_io
&& mdev
->ldev
->dc
.no_disk_drain
)
1143 mdev
->write_ordering
= wo
;
1144 if (pwo
!= mdev
->write_ordering
|| wo
== WO_bio_barrier
)
1145 dev_info(DEV
, "Method to ensure write ordering: %s\n", write_ordering_str
[mdev
->write_ordering
]);
1150 * @mdev: DRBD device.
1152 * @rw: flag field, see bio->bi_rw
1154 /* TODO allocate from our own bio_set. */
1155 int drbd_submit_ee(struct drbd_conf
*mdev
, struct drbd_epoch_entry
*e
,
1156 const unsigned rw
, const int fault_type
)
1158 struct bio
*bios
= NULL
;
1160 struct page
*page
= e
->pages
;
1161 sector_t sector
= e
->sector
;
1162 unsigned ds
= e
->size
;
1163 unsigned n_bios
= 0;
1164 unsigned nr_pages
= (ds
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
1166 /* In most cases, we will only need one bio. But in case the lower
1167 * level restrictions happen to be different at this offset on this
1168 * side than those of the sending peer, we may need to submit the
1169 * request in more than one bio. */
1171 bio
= bio_alloc(GFP_NOIO
, nr_pages
);
1173 dev_err(DEV
, "submit_ee: Allocation of a bio failed\n");
1176 /* > e->sector, unless this is the first bio */
1177 bio
->bi_sector
= sector
;
1178 bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1179 /* we special case some flags in the multi-bio case, see below
1180 * (REQ_UNPLUG, REQ_HARDBARRIER) */
1182 bio
->bi_private
= e
;
1183 bio
->bi_end_io
= drbd_endio_sec
;
1185 bio
->bi_next
= bios
;
1189 page_chain_for_each(page
) {
1190 unsigned len
= min_t(unsigned, ds
, PAGE_SIZE
);
1191 if (!bio_add_page(bio
, page
, len
, 0)) {
1192 /* a single page must always be possible! */
1193 BUG_ON(bio
->bi_vcnt
== 0);
1200 D_ASSERT(page
== NULL
);
1203 atomic_set(&e
->pending_bios
, n_bios
);
1206 bios
= bios
->bi_next
;
1207 bio
->bi_next
= NULL
;
1209 /* strip off REQ_UNPLUG unless it is the last bio */
1211 bio
->bi_rw
&= ~REQ_UNPLUG
;
1213 drbd_generic_make_request(mdev
, fault_type
, bio
);
1215 /* strip off REQ_HARDBARRIER,
1216 * unless it is the first or last bio */
1217 if (bios
&& bios
->bi_next
)
1218 bios
->bi_rw
&= ~REQ_HARDBARRIER
;
1220 maybe_kick_lo(mdev
);
1226 bios
= bios
->bi_next
;
1233 * w_e_reissue() - Worker callback; Resubmit a bio, without REQ_HARDBARRIER set
1234 * @mdev: DRBD device.
1236 * @cancel: The connection will be closed anyways (unused in this callback)
1238 int w_e_reissue(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
) __releases(local
)
1240 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1241 /* We leave DE_CONTAINS_A_BARRIER and EE_IS_BARRIER in place,
1242 (and DE_BARRIER_IN_NEXT_EPOCH_ISSUED in the previous Epoch)
1243 so that we can finish that epoch in drbd_may_finish_epoch().
1244 That is necessary if we already have a long chain of Epochs, before
1245 we realize that REQ_HARDBARRIER is actually not supported */
1247 /* As long as the -ENOTSUPP on the barrier is reported immediately
1248 that will never trigger. If it is reported late, we will just
1249 print that warning and continue correctly for all future requests
1250 with WO_bdev_flush */
1251 if (previous_epoch(mdev
, e
->epoch
))
1252 dev_warn(DEV
, "Write ordering was not enforced (one time event)\n");
1254 /* we still have a local reference,
1255 * get_ldev was done in receive_Data. */
1257 e
->w
.cb
= e_end_block
;
1258 if (drbd_submit_ee(mdev
, e
, WRITE
, DRBD_FAULT_DT_WR
) != 0) {
1259 /* drbd_submit_ee fails for one reason only:
1260 * if was not able to allocate sufficient bios.
1261 * requeue, try again later. */
1262 e
->w
.cb
= w_e_reissue
;
1263 drbd_queue_work(&mdev
->data
.work
, &e
->w
);
1268 static int receive_Barrier(struct drbd_conf
*mdev
, struct p_header
*h
)
1270 int rv
, issue_flush
;
1271 struct p_barrier
*p
= (struct p_barrier
*)h
;
1272 struct drbd_epoch
*epoch
;
1274 ERR_IF(h
->length
!= (sizeof(*p
)-sizeof(*h
))) return FALSE
;
1276 rv
= drbd_recv(mdev
, h
->payload
, h
->length
);
1277 ERR_IF(rv
!= h
->length
) return FALSE
;
1281 if (mdev
->net_conf
->wire_protocol
!= DRBD_PROT_C
)
1284 mdev
->current_epoch
->barrier_nr
= p
->barrier
;
1285 rv
= drbd_may_finish_epoch(mdev
, mdev
->current_epoch
, EV_GOT_BARRIER_NR
);
1287 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1288 * the activity log, which means it would not be resynced in case the
1289 * R_PRIMARY crashes now.
1290 * Therefore we must send the barrier_ack after the barrier request was
1292 switch (mdev
->write_ordering
) {
1293 case WO_bio_barrier
:
1295 if (rv
== FE_RECYCLED
)
1301 if (rv
== FE_STILL_LIVE
) {
1302 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED
, &mdev
->current_epoch
->flags
);
1303 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1304 rv
= drbd_flush_after_epoch(mdev
, mdev
->current_epoch
);
1306 if (rv
== FE_RECYCLED
)
1309 /* The asender will send all the ACKs and barrier ACKs out, since
1310 all EEs moved from the active_ee to the done_ee. We need to
1311 provide a new epoch object for the EEs that come in soon */
1315 /* receiver context, in the writeout path of the other node.
1316 * avoid potential distributed deadlock */
1317 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1319 dev_warn(DEV
, "Allocation of an epoch failed, slowing down\n");
1320 issue_flush
= !test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED
, &mdev
->current_epoch
->flags
);
1321 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1323 rv
= drbd_flush_after_epoch(mdev
, mdev
->current_epoch
);
1324 if (rv
== FE_RECYCLED
)
1328 drbd_wait_ee_list_empty(mdev
, &mdev
->done_ee
);
1334 atomic_set(&epoch
->epoch_size
, 0);
1335 atomic_set(&epoch
->active
, 0);
1337 spin_lock(&mdev
->epoch_lock
);
1338 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1339 list_add(&epoch
->list
, &mdev
->current_epoch
->list
);
1340 mdev
->current_epoch
= epoch
;
1343 /* The current_epoch got recycled while we allocated this one... */
1346 spin_unlock(&mdev
->epoch_lock
);
1351 /* used from receive_RSDataReply (recv_resync_read)
1352 * and from receive_Data */
1353 static struct drbd_epoch_entry
*
1354 read_in_block(struct drbd_conf
*mdev
, u64 id
, sector_t sector
, int data_size
) __must_hold(local
)
1356 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1357 struct drbd_epoch_entry
*e
;
1360 void *dig_in
= mdev
->int_dig_in
;
1361 void *dig_vv
= mdev
->int_dig_vv
;
1362 unsigned long *data
;
1364 dgs
= (mdev
->agreed_pro_version
>= 87 && mdev
->integrity_r_tfm
) ?
1365 crypto_hash_digestsize(mdev
->integrity_r_tfm
) : 0;
1368 rr
= drbd_recv(mdev
, dig_in
, dgs
);
1370 dev_warn(DEV
, "short read receiving data digest: read %d expected %d\n",
1378 ERR_IF(data_size
& 0x1ff) return NULL
;
1379 ERR_IF(data_size
> DRBD_MAX_SEGMENT_SIZE
) return NULL
;
1381 /* even though we trust out peer,
1382 * we sometimes have to double check. */
1383 if (sector
+ (data_size
>>9) > capacity
) {
1384 dev_err(DEV
, "capacity: %llus < sector: %llus + size: %u\n",
1385 (unsigned long long)capacity
,
1386 (unsigned long long)sector
, data_size
);
1390 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1391 * "criss-cross" setup, that might cause write-out on some other DRBD,
1392 * which in turn might block on the other node at this very place. */
1393 e
= drbd_alloc_ee(mdev
, id
, sector
, data_size
, GFP_NOIO
);
1399 page_chain_for_each(page
) {
1400 unsigned len
= min_t(int, ds
, PAGE_SIZE
);
1402 rr
= drbd_recv(mdev
, data
, len
);
1403 if (FAULT_ACTIVE(mdev
, DRBD_FAULT_RECEIVE
)) {
1404 dev_err(DEV
, "Fault injection: Corrupting data on receive\n");
1405 data
[0] = data
[0] ^ (unsigned long)-1;
1409 drbd_free_ee(mdev
, e
);
1410 dev_warn(DEV
, "short read receiving data: read %d expected %d\n",
1418 drbd_csum_ee(mdev
, mdev
->integrity_r_tfm
, e
, dig_vv
);
1419 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1420 dev_err(DEV
, "Digest integrity check FAILED.\n");
1421 drbd_bcast_ee(mdev
, "digest failed",
1422 dgs
, dig_in
, dig_vv
, e
);
1423 drbd_free_ee(mdev
, e
);
1427 mdev
->recv_cnt
+= data_size
>>9;
1431 /* drbd_drain_block() just takes a data block
1432 * out of the socket input buffer, and discards it.
1434 static int drbd_drain_block(struct drbd_conf
*mdev
, int data_size
)
1443 page
= drbd_pp_alloc(mdev
, 1, 1);
1447 rr
= drbd_recv(mdev
, data
, min_t(int, data_size
, PAGE_SIZE
));
1448 if (rr
!= min_t(int, data_size
, PAGE_SIZE
)) {
1450 dev_warn(DEV
, "short read receiving data: read %d expected %d\n",
1451 rr
, min_t(int, data_size
, PAGE_SIZE
));
1457 drbd_pp_free(mdev
, page
);
1461 static int recv_dless_read(struct drbd_conf
*mdev
, struct drbd_request
*req
,
1462 sector_t sector
, int data_size
)
1464 struct bio_vec
*bvec
;
1466 int dgs
, rr
, i
, expect
;
1467 void *dig_in
= mdev
->int_dig_in
;
1468 void *dig_vv
= mdev
->int_dig_vv
;
1470 dgs
= (mdev
->agreed_pro_version
>= 87 && mdev
->integrity_r_tfm
) ?
1471 crypto_hash_digestsize(mdev
->integrity_r_tfm
) : 0;
1474 rr
= drbd_recv(mdev
, dig_in
, dgs
);
1476 dev_warn(DEV
, "short read receiving data reply digest: read %d expected %d\n",
1484 /* optimistically update recv_cnt. if receiving fails below,
1485 * we disconnect anyways, and counters will be reset. */
1486 mdev
->recv_cnt
+= data_size
>>9;
1488 bio
= req
->master_bio
;
1489 D_ASSERT(sector
== bio
->bi_sector
);
1491 bio_for_each_segment(bvec
, bio
, i
) {
1492 expect
= min_t(int, data_size
, bvec
->bv_len
);
1493 rr
= drbd_recv(mdev
,
1494 kmap(bvec
->bv_page
)+bvec
->bv_offset
,
1496 kunmap(bvec
->bv_page
);
1498 dev_warn(DEV
, "short read receiving data reply: "
1499 "read %d expected %d\n",
1507 drbd_csum_bio(mdev
, mdev
->integrity_r_tfm
, bio
, dig_vv
);
1508 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1509 dev_err(DEV
, "Digest integrity check FAILED. Broken NICs?\n");
1514 D_ASSERT(data_size
== 0);
1518 /* e_end_resync_block() is called via
1519 * drbd_process_done_ee() by asender only */
1520 static int e_end_resync_block(struct drbd_conf
*mdev
, struct drbd_work
*w
, int unused
)
1522 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1523 sector_t sector
= e
->sector
;
1526 D_ASSERT(hlist_unhashed(&e
->colision
));
1528 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
1529 drbd_set_in_sync(mdev
, sector
, e
->size
);
1530 ok
= drbd_send_ack(mdev
, P_RS_WRITE_ACK
, e
);
1532 /* Record failure to sync */
1533 drbd_rs_failed_io(mdev
, sector
, e
->size
);
1535 ok
= drbd_send_ack(mdev
, P_NEG_ACK
, e
);
1542 static int recv_resync_read(struct drbd_conf
*mdev
, sector_t sector
, int data_size
) __releases(local
)
1544 struct drbd_epoch_entry
*e
;
1546 e
= read_in_block(mdev
, ID_SYNCER
, sector
, data_size
);
1550 dec_rs_pending(mdev
);
1553 /* corresponding dec_unacked() in e_end_resync_block()
1554 * respective _drbd_clear_done_ee */
1556 e
->w
.cb
= e_end_resync_block
;
1558 spin_lock_irq(&mdev
->req_lock
);
1559 list_add(&e
->w
.list
, &mdev
->sync_ee
);
1560 spin_unlock_irq(&mdev
->req_lock
);
1562 if (drbd_submit_ee(mdev
, e
, WRITE
, DRBD_FAULT_RS_WR
) == 0)
1565 drbd_free_ee(mdev
, e
);
1571 static int receive_DataReply(struct drbd_conf
*mdev
, struct p_header
*h
)
1573 struct drbd_request
*req
;
1575 unsigned int header_size
, data_size
;
1577 struct p_data
*p
= (struct p_data
*)h
;
1579 header_size
= sizeof(*p
) - sizeof(*h
);
1580 data_size
= h
->length
- header_size
;
1582 ERR_IF(data_size
== 0) return FALSE
;
1584 if (drbd_recv(mdev
, h
->payload
, header_size
) != header_size
)
1587 sector
= be64_to_cpu(p
->sector
);
1589 spin_lock_irq(&mdev
->req_lock
);
1590 req
= _ar_id_to_req(mdev
, p
->block_id
, sector
);
1591 spin_unlock_irq(&mdev
->req_lock
);
1592 if (unlikely(!req
)) {
1593 dev_err(DEV
, "Got a corrupt block_id/sector pair(1).\n");
1597 /* hlist_del(&req->colision) is done in _req_may_be_done, to avoid
1598 * special casing it there for the various failure cases.
1599 * still no race with drbd_fail_pending_reads */
1600 ok
= recv_dless_read(mdev
, req
, sector
, data_size
);
1603 req_mod(req
, data_received
);
1604 /* else: nothing. handled from drbd_disconnect...
1605 * I don't think we may complete this just yet
1606 * in case we are "on-disconnect: freeze" */
1611 static int receive_RSDataReply(struct drbd_conf
*mdev
, struct p_header
*h
)
1614 unsigned int header_size
, data_size
;
1616 struct p_data
*p
= (struct p_data
*)h
;
1618 header_size
= sizeof(*p
) - sizeof(*h
);
1619 data_size
= h
->length
- header_size
;
1621 ERR_IF(data_size
== 0) return FALSE
;
1623 if (drbd_recv(mdev
, h
->payload
, header_size
) != header_size
)
1626 sector
= be64_to_cpu(p
->sector
);
1627 D_ASSERT(p
->block_id
== ID_SYNCER
);
1629 if (get_ldev(mdev
)) {
1630 /* data is submitted to disk within recv_resync_read.
1631 * corresponding put_ldev done below on error,
1632 * or in drbd_endio_write_sec. */
1633 ok
= recv_resync_read(mdev
, sector
, data_size
);
1635 if (__ratelimit(&drbd_ratelimit_state
))
1636 dev_err(DEV
, "Can not write resync data to local disk.\n");
1638 ok
= drbd_drain_block(mdev
, data_size
);
1640 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
);
1643 atomic_add(data_size
>> 9, &mdev
->rs_sect_in
);
1648 /* e_end_block() is called via drbd_process_done_ee().
1649 * this means this function only runs in the asender thread
1651 static int e_end_block(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1653 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1654 sector_t sector
= e
->sector
;
1655 struct drbd_epoch
*epoch
;
1658 if (e
->flags
& EE_IS_BARRIER
) {
1659 epoch
= previous_epoch(mdev
, e
->epoch
);
1661 drbd_may_finish_epoch(mdev
, epoch
, EV_BARRIER_DONE
+ (cancel
? EV_CLEANUP
: 0));
1664 if (mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
) {
1665 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
1666 pcmd
= (mdev
->state
.conn
>= C_SYNC_SOURCE
&&
1667 mdev
->state
.conn
<= C_PAUSED_SYNC_T
&&
1668 e
->flags
& EE_MAY_SET_IN_SYNC
) ?
1669 P_RS_WRITE_ACK
: P_WRITE_ACK
;
1670 ok
&= drbd_send_ack(mdev
, pcmd
, e
);
1671 if (pcmd
== P_RS_WRITE_ACK
)
1672 drbd_set_in_sync(mdev
, sector
, e
->size
);
1674 ok
= drbd_send_ack(mdev
, P_NEG_ACK
, e
);
1675 /* we expect it to be marked out of sync anyways...
1676 * maybe assert this? */
1680 /* we delete from the conflict detection hash _after_ we sent out the
1681 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1682 if (mdev
->net_conf
->two_primaries
) {
1683 spin_lock_irq(&mdev
->req_lock
);
1684 D_ASSERT(!hlist_unhashed(&e
->colision
));
1685 hlist_del_init(&e
->colision
);
1686 spin_unlock_irq(&mdev
->req_lock
);
1688 D_ASSERT(hlist_unhashed(&e
->colision
));
1691 drbd_may_finish_epoch(mdev
, e
->epoch
, EV_PUT
+ (cancel
? EV_CLEANUP
: 0));
1696 static int e_send_discard_ack(struct drbd_conf
*mdev
, struct drbd_work
*w
, int unused
)
1698 struct drbd_epoch_entry
*e
= (struct drbd_epoch_entry
*)w
;
1701 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
1702 ok
= drbd_send_ack(mdev
, P_DISCARD_ACK
, e
);
1704 spin_lock_irq(&mdev
->req_lock
);
1705 D_ASSERT(!hlist_unhashed(&e
->colision
));
1706 hlist_del_init(&e
->colision
);
1707 spin_unlock_irq(&mdev
->req_lock
);
1714 /* Called from receive_Data.
1715 * Synchronize packets on sock with packets on msock.
1717 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1718 * packet traveling on msock, they are still processed in the order they have
1721 * Note: we don't care for Ack packets overtaking P_DATA packets.
1723 * In case packet_seq is larger than mdev->peer_seq number, there are
1724 * outstanding packets on the msock. We wait for them to arrive.
1725 * In case we are the logically next packet, we update mdev->peer_seq
1726 * ourselves. Correctly handles 32bit wrap around.
1728 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1729 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1730 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1731 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1733 * returns 0 if we may process the packet,
1734 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1735 static int drbd_wait_peer_seq(struct drbd_conf
*mdev
, const u32 packet_seq
)
1741 spin_lock(&mdev
->peer_seq_lock
);
1743 prepare_to_wait(&mdev
->seq_wait
, &wait
, TASK_INTERRUPTIBLE
);
1744 if (seq_le(packet_seq
, mdev
->peer_seq
+1))
1746 if (signal_pending(current
)) {
1750 p_seq
= mdev
->peer_seq
;
1751 spin_unlock(&mdev
->peer_seq_lock
);
1752 timeout
= schedule_timeout(30*HZ
);
1753 spin_lock(&mdev
->peer_seq_lock
);
1754 if (timeout
== 0 && p_seq
== mdev
->peer_seq
) {
1756 dev_err(DEV
, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n");
1760 finish_wait(&mdev
->seq_wait
, &wait
);
1761 if (mdev
->peer_seq
+1 == packet_seq
)
1763 spin_unlock(&mdev
->peer_seq_lock
);
1767 /* mirrored write */
1768 static int receive_Data(struct drbd_conf
*mdev
, struct p_header
*h
)
1771 struct drbd_epoch_entry
*e
;
1772 struct p_data
*p
= (struct p_data
*)h
;
1773 int header_size
, data_size
;
1777 header_size
= sizeof(*p
) - sizeof(*h
);
1778 data_size
= h
->length
- header_size
;
1780 ERR_IF(data_size
== 0) return FALSE
;
1782 if (drbd_recv(mdev
, h
->payload
, header_size
) != header_size
)
1785 if (!get_ldev(mdev
)) {
1786 if (__ratelimit(&drbd_ratelimit_state
))
1787 dev_err(DEV
, "Can not write mirrored data block "
1788 "to local disk.\n");
1789 spin_lock(&mdev
->peer_seq_lock
);
1790 if (mdev
->peer_seq
+1 == be32_to_cpu(p
->seq_num
))
1792 spin_unlock(&mdev
->peer_seq_lock
);
1794 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
);
1795 atomic_inc(&mdev
->current_epoch
->epoch_size
);
1796 return drbd_drain_block(mdev
, data_size
);
1799 /* get_ldev(mdev) successful.
1800 * Corresponding put_ldev done either below (on various errors),
1801 * or in drbd_endio_write_sec, if we successfully submit the data at
1802 * the end of this function. */
1804 sector
= be64_to_cpu(p
->sector
);
1805 e
= read_in_block(mdev
, p
->block_id
, sector
, data_size
);
1811 e
->w
.cb
= e_end_block
;
1813 spin_lock(&mdev
->epoch_lock
);
1814 e
->epoch
= mdev
->current_epoch
;
1815 atomic_inc(&e
->epoch
->epoch_size
);
1816 atomic_inc(&e
->epoch
->active
);
1818 if (mdev
->write_ordering
== WO_bio_barrier
&& atomic_read(&e
->epoch
->epoch_size
) == 1) {
1819 struct drbd_epoch
*epoch
;
1820 /* Issue a barrier if we start a new epoch, and the previous epoch
1821 was not a epoch containing a single request which already was
1823 epoch
= list_entry(e
->epoch
->list
.prev
, struct drbd_epoch
, list
);
1824 if (epoch
== e
->epoch
) {
1825 set_bit(DE_CONTAINS_A_BARRIER
, &e
->epoch
->flags
);
1826 rw
|= REQ_HARDBARRIER
;
1827 e
->flags
|= EE_IS_BARRIER
;
1829 if (atomic_read(&epoch
->epoch_size
) > 1 ||
1830 !test_bit(DE_CONTAINS_A_BARRIER
, &epoch
->flags
)) {
1831 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED
, &epoch
->flags
);
1832 set_bit(DE_CONTAINS_A_BARRIER
, &e
->epoch
->flags
);
1833 rw
|= REQ_HARDBARRIER
;
1834 e
->flags
|= EE_IS_BARRIER
;
1838 spin_unlock(&mdev
->epoch_lock
);
1840 dp_flags
= be32_to_cpu(p
->dp_flags
);
1841 if (dp_flags
& DP_HARDBARRIER
) {
1842 dev_err(DEV
, "ASSERT FAILED would have submitted barrier request\n");
1843 /* rw |= REQ_HARDBARRIER; */
1845 if (dp_flags
& DP_RW_SYNC
)
1846 rw
|= REQ_SYNC
| REQ_UNPLUG
;
1847 if (dp_flags
& DP_MAY_SET_IN_SYNC
)
1848 e
->flags
|= EE_MAY_SET_IN_SYNC
;
1850 /* I'm the receiver, I do hold a net_cnt reference. */
1851 if (!mdev
->net_conf
->two_primaries
) {
1852 spin_lock_irq(&mdev
->req_lock
);
1854 /* don't get the req_lock yet,
1855 * we may sleep in drbd_wait_peer_seq */
1856 const int size
= e
->size
;
1857 const int discard
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
1859 struct drbd_request
*i
;
1860 struct hlist_node
*n
;
1861 struct hlist_head
*slot
;
1864 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
1865 BUG_ON(mdev
->ee_hash
== NULL
);
1866 BUG_ON(mdev
->tl_hash
== NULL
);
1868 /* conflict detection and handling:
1869 * 1. wait on the sequence number,
1870 * in case this data packet overtook ACK packets.
1871 * 2. check our hash tables for conflicting requests.
1872 * we only need to walk the tl_hash, since an ee can not
1873 * have a conflict with an other ee: on the submitting
1874 * node, the corresponding req had already been conflicting,
1875 * and a conflicting req is never sent.
1877 * Note: for two_primaries, we are protocol C,
1878 * so there cannot be any request that is DONE
1879 * but still on the transfer log.
1881 * unconditionally add to the ee_hash.
1883 * if no conflicting request is found:
1886 * if any conflicting request is found
1887 * that has not yet been acked,
1888 * AND I have the "discard concurrent writes" flag:
1889 * queue (via done_ee) the P_DISCARD_ACK; OUT.
1891 * if any conflicting request is found:
1892 * block the receiver, waiting on misc_wait
1893 * until no more conflicting requests are there,
1894 * or we get interrupted (disconnect).
1896 * we do not just write after local io completion of those
1897 * requests, but only after req is done completely, i.e.
1898 * we wait for the P_DISCARD_ACK to arrive!
1900 * then proceed normally, i.e. submit.
1902 if (drbd_wait_peer_seq(mdev
, be32_to_cpu(p
->seq_num
)))
1903 goto out_interrupted
;
1905 spin_lock_irq(&mdev
->req_lock
);
1907 hlist_add_head(&e
->colision
, ee_hash_slot(mdev
, sector
));
1909 #define OVERLAPS overlaps(i->sector, i->size, sector, size)
1910 slot
= tl_hash_slot(mdev
, sector
);
1913 int have_unacked
= 0;
1914 int have_conflict
= 0;
1915 prepare_to_wait(&mdev
->misc_wait
, &wait
,
1916 TASK_INTERRUPTIBLE
);
1917 hlist_for_each_entry(i
, n
, slot
, colision
) {
1919 /* only ALERT on first iteration,
1920 * we may be woken up early... */
1922 dev_alert(DEV
, "%s[%u] Concurrent local write detected!"
1923 " new: %llus +%u; pending: %llus +%u\n",
1924 current
->comm
, current
->pid
,
1925 (unsigned long long)sector
, size
,
1926 (unsigned long long)i
->sector
, i
->size
);
1927 if (i
->rq_state
& RQ_NET_PENDING
)
1936 /* Discard Ack only for the _first_ iteration */
1937 if (first
&& discard
&& have_unacked
) {
1938 dev_alert(DEV
, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n",
1939 (unsigned long long)sector
);
1941 e
->w
.cb
= e_send_discard_ack
;
1942 list_add_tail(&e
->w
.list
, &mdev
->done_ee
);
1944 spin_unlock_irq(&mdev
->req_lock
);
1946 /* we could probably send that P_DISCARD_ACK ourselves,
1947 * but I don't like the receiver using the msock */
1951 finish_wait(&mdev
->misc_wait
, &wait
);
1955 if (signal_pending(current
)) {
1956 hlist_del_init(&e
->colision
);
1958 spin_unlock_irq(&mdev
->req_lock
);
1960 finish_wait(&mdev
->misc_wait
, &wait
);
1961 goto out_interrupted
;
1964 spin_unlock_irq(&mdev
->req_lock
);
1967 dev_alert(DEV
, "Concurrent write! [W AFTERWARDS] "
1968 "sec=%llus\n", (unsigned long long)sector
);
1969 } else if (discard
) {
1970 /* we had none on the first iteration.
1971 * there must be none now. */
1972 D_ASSERT(have_unacked
== 0);
1975 spin_lock_irq(&mdev
->req_lock
);
1977 finish_wait(&mdev
->misc_wait
, &wait
);
1980 list_add(&e
->w
.list
, &mdev
->active_ee
);
1981 spin_unlock_irq(&mdev
->req_lock
);
1983 switch (mdev
->net_conf
->wire_protocol
) {
1986 /* corresponding dec_unacked() in e_end_block()
1987 * respective _drbd_clear_done_ee */
1990 /* I really don't like it that the receiver thread
1991 * sends on the msock, but anyways */
1992 drbd_send_ack(mdev
, P_RECV_ACK
, e
);
1999 if (mdev
->state
.pdsk
== D_DISKLESS
) {
2000 /* In case we have the only disk of the cluster, */
2001 drbd_set_out_of_sync(mdev
, e
->sector
, e
->size
);
2002 e
->flags
|= EE_CALL_AL_COMPLETE_IO
;
2003 drbd_al_begin_io(mdev
, e
->sector
);
2006 if (drbd_submit_ee(mdev
, e
, rw
, DRBD_FAULT_DT_WR
) == 0)
2010 /* yes, the epoch_size now is imbalanced.
2011 * but we drop the connection anyways, so we don't have a chance to
2012 * receive a barrier... atomic_inc(&mdev->epoch_size); */
2014 drbd_free_ee(mdev
, e
);
2018 static int receive_DataRequest(struct drbd_conf
*mdev
, struct p_header
*h
)
2021 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
2022 struct drbd_epoch_entry
*e
;
2023 struct digest_info
*di
= NULL
;
2024 int size
, digest_size
;
2025 unsigned int fault_type
;
2026 struct p_block_req
*p
=
2027 (struct p_block_req
*)h
;
2028 const int brps
= sizeof(*p
)-sizeof(*h
);
2030 if (drbd_recv(mdev
, h
->payload
, brps
) != brps
)
2033 sector
= be64_to_cpu(p
->sector
);
2034 size
= be32_to_cpu(p
->blksize
);
2036 if (size
<= 0 || (size
& 0x1ff) != 0 || size
> DRBD_MAX_SEGMENT_SIZE
) {
2037 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2038 (unsigned long long)sector
, size
);
2041 if (sector
+ (size
>>9) > capacity
) {
2042 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2043 (unsigned long long)sector
, size
);
2047 if (!get_ldev_if_state(mdev
, D_UP_TO_DATE
)) {
2048 if (__ratelimit(&drbd_ratelimit_state
))
2049 dev_err(DEV
, "Can not satisfy peer's read request, "
2050 "no local data.\n");
2051 drbd_send_ack_rp(mdev
, h
->command
== P_DATA_REQUEST
? P_NEG_DREPLY
:
2052 P_NEG_RS_DREPLY
, p
);
2053 return drbd_drain_block(mdev
, h
->length
- brps
);
2056 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2057 * "criss-cross" setup, that might cause write-out on some other DRBD,
2058 * which in turn might block on the other node at this very place. */
2059 e
= drbd_alloc_ee(mdev
, p
->block_id
, sector
, size
, GFP_NOIO
);
2065 switch (h
->command
) {
2066 case P_DATA_REQUEST
:
2067 e
->w
.cb
= w_e_end_data_req
;
2068 fault_type
= DRBD_FAULT_DT_RD
;
2070 case P_RS_DATA_REQUEST
:
2071 e
->w
.cb
= w_e_end_rsdata_req
;
2072 fault_type
= DRBD_FAULT_RS_RD
;
2073 /* Eventually this should become asynchronously. Currently it
2074 * blocks the whole receiver just to delay the reading of a
2075 * resync data block.
2076 * the drbd_work_queue mechanism is made for this...
2078 if (!drbd_rs_begin_io(mdev
, sector
)) {
2079 /* we have been interrupted,
2080 * probably connection lost! */
2081 D_ASSERT(signal_pending(current
));
2087 case P_CSUM_RS_REQUEST
:
2088 fault_type
= DRBD_FAULT_RS_RD
;
2089 digest_size
= h
->length
- brps
;
2090 di
= kmalloc(sizeof(*di
) + digest_size
, GFP_NOIO
);
2094 di
->digest_size
= digest_size
;
2095 di
->digest
= (((char *)di
)+sizeof(struct digest_info
));
2097 if (drbd_recv(mdev
, di
->digest
, digest_size
) != digest_size
)
2101 if (h
->command
== P_CSUM_RS_REQUEST
) {
2102 D_ASSERT(mdev
->agreed_pro_version
>= 89);
2103 e
->w
.cb
= w_e_end_csum_rs_req
;
2104 } else if (h
->command
== P_OV_REPLY
) {
2105 e
->w
.cb
= w_e_end_ov_reply
;
2106 dec_rs_pending(mdev
);
2110 if (!drbd_rs_begin_io(mdev
, sector
)) {
2111 /* we have been interrupted, probably connection lost! */
2112 D_ASSERT(signal_pending(current
));
2118 if (mdev
->state
.conn
>= C_CONNECTED
&&
2119 mdev
->state
.conn
!= C_VERIFY_T
)
2120 dev_warn(DEV
, "ASSERT FAILED: got P_OV_REQUEST while being %s\n",
2121 drbd_conn_str(mdev
->state
.conn
));
2122 if (mdev
->ov_start_sector
== ~(sector_t
)0 &&
2123 mdev
->agreed_pro_version
>= 90) {
2124 mdev
->ov_start_sector
= sector
;
2125 mdev
->ov_position
= sector
;
2126 mdev
->ov_left
= mdev
->rs_total
- BM_SECT_TO_BIT(sector
);
2127 dev_info(DEV
, "Online Verify start sector: %llu\n",
2128 (unsigned long long)sector
);
2130 e
->w
.cb
= w_e_end_ov_req
;
2131 fault_type
= DRBD_FAULT_RS_RD
;
2132 /* Eventually this should become asynchronous. Currently it
2133 * blocks the whole receiver just to delay the reading of a
2134 * resync data block.
2135 * the drbd_work_queue mechanism is made for this...
2137 if (!drbd_rs_begin_io(mdev
, sector
)) {
2138 /* we have been interrupted,
2139 * probably connection lost! */
2140 D_ASSERT(signal_pending(current
));
2147 dev_err(DEV
, "unexpected command (%s) in receive_DataRequest\n",
2148 cmdname(h
->command
));
2149 fault_type
= DRBD_FAULT_MAX
;
2152 spin_lock_irq(&mdev
->req_lock
);
2153 list_add(&e
->w
.list
, &mdev
->read_ee
);
2154 spin_unlock_irq(&mdev
->req_lock
);
2158 if (drbd_submit_ee(mdev
, e
, READ
, fault_type
) == 0)
2164 drbd_free_ee(mdev
, e
);
2168 static int drbd_asb_recover_0p(struct drbd_conf
*mdev
) __must_hold(local
)
2170 int self
, peer
, rv
= -100;
2171 unsigned long ch_self
, ch_peer
;
2173 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2174 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2176 ch_peer
= mdev
->p_uuid
[UI_SIZE
];
2177 ch_self
= mdev
->comm_bm_set
;
2179 switch (mdev
->net_conf
->after_sb_0p
) {
2181 case ASB_DISCARD_SECONDARY
:
2182 case ASB_CALL_HELPER
:
2183 dev_err(DEV
, "Configuration error.\n");
2185 case ASB_DISCONNECT
:
2187 case ASB_DISCARD_YOUNGER_PRI
:
2188 if (self
== 0 && peer
== 1) {
2192 if (self
== 1 && peer
== 0) {
2196 /* Else fall through to one of the other strategies... */
2197 case ASB_DISCARD_OLDER_PRI
:
2198 if (self
== 0 && peer
== 1) {
2202 if (self
== 1 && peer
== 0) {
2206 /* Else fall through to one of the other strategies... */
2207 dev_warn(DEV
, "Discard younger/older primary did not find a decision\n"
2208 "Using discard-least-changes instead\n");
2209 case ASB_DISCARD_ZERO_CHG
:
2210 if (ch_peer
== 0 && ch_self
== 0) {
2211 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
)
2215 if (ch_peer
== 0) { rv
= 1; break; }
2216 if (ch_self
== 0) { rv
= -1; break; }
2218 if (mdev
->net_conf
->after_sb_0p
== ASB_DISCARD_ZERO_CHG
)
2220 case ASB_DISCARD_LEAST_CHG
:
2221 if (ch_self
< ch_peer
)
2223 else if (ch_self
> ch_peer
)
2225 else /* ( ch_self == ch_peer ) */
2226 /* Well, then use something else. */
2227 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
)
2230 case ASB_DISCARD_LOCAL
:
2233 case ASB_DISCARD_REMOTE
:
2240 static int drbd_asb_recover_1p(struct drbd_conf
*mdev
) __must_hold(local
)
2242 int self
, peer
, hg
, rv
= -100;
2244 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2245 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2247 switch (mdev
->net_conf
->after_sb_1p
) {
2248 case ASB_DISCARD_YOUNGER_PRI
:
2249 case ASB_DISCARD_OLDER_PRI
:
2250 case ASB_DISCARD_LEAST_CHG
:
2251 case ASB_DISCARD_LOCAL
:
2252 case ASB_DISCARD_REMOTE
:
2253 dev_err(DEV
, "Configuration error.\n");
2255 case ASB_DISCONNECT
:
2258 hg
= drbd_asb_recover_0p(mdev
);
2259 if (hg
== -1 && mdev
->state
.role
== R_SECONDARY
)
2261 if (hg
== 1 && mdev
->state
.role
== R_PRIMARY
)
2265 rv
= drbd_asb_recover_0p(mdev
);
2267 case ASB_DISCARD_SECONDARY
:
2268 return mdev
->state
.role
== R_PRIMARY
? 1 : -1;
2269 case ASB_CALL_HELPER
:
2270 hg
= drbd_asb_recover_0p(mdev
);
2271 if (hg
== -1 && mdev
->state
.role
== R_PRIMARY
) {
2272 self
= drbd_set_role(mdev
, R_SECONDARY
, 0);
2273 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2274 * we might be here in C_WF_REPORT_PARAMS which is transient.
2275 * we do not need to wait for the after state change work either. */
2276 self
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2277 if (self
!= SS_SUCCESS
) {
2278 drbd_khelper(mdev
, "pri-lost-after-sb");
2280 dev_warn(DEV
, "Successfully gave up primary role.\n");
2290 static int drbd_asb_recover_2p(struct drbd_conf
*mdev
) __must_hold(local
)
2292 int self
, peer
, hg
, rv
= -100;
2294 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2295 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2297 switch (mdev
->net_conf
->after_sb_2p
) {
2298 case ASB_DISCARD_YOUNGER_PRI
:
2299 case ASB_DISCARD_OLDER_PRI
:
2300 case ASB_DISCARD_LEAST_CHG
:
2301 case ASB_DISCARD_LOCAL
:
2302 case ASB_DISCARD_REMOTE
:
2304 case ASB_DISCARD_SECONDARY
:
2305 dev_err(DEV
, "Configuration error.\n");
2308 rv
= drbd_asb_recover_0p(mdev
);
2310 case ASB_DISCONNECT
:
2312 case ASB_CALL_HELPER
:
2313 hg
= drbd_asb_recover_0p(mdev
);
2315 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2316 * we might be here in C_WF_REPORT_PARAMS which is transient.
2317 * we do not need to wait for the after state change work either. */
2318 self
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2319 if (self
!= SS_SUCCESS
) {
2320 drbd_khelper(mdev
, "pri-lost-after-sb");
2322 dev_warn(DEV
, "Successfully gave up primary role.\n");
2332 static void drbd_uuid_dump(struct drbd_conf
*mdev
, char *text
, u64
*uuid
,
2333 u64 bits
, u64 flags
)
2336 dev_info(DEV
, "%s uuid info vanished while I was looking!\n", text
);
2339 dev_info(DEV
, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2341 (unsigned long long)uuid
[UI_CURRENT
],
2342 (unsigned long long)uuid
[UI_BITMAP
],
2343 (unsigned long long)uuid
[UI_HISTORY_START
],
2344 (unsigned long long)uuid
[UI_HISTORY_END
],
2345 (unsigned long long)bits
,
2346 (unsigned long long)flags
);
2350 100 after split brain try auto recover
2351 2 C_SYNC_SOURCE set BitMap
2352 1 C_SYNC_SOURCE use BitMap
2354 -1 C_SYNC_TARGET use BitMap
2355 -2 C_SYNC_TARGET set BitMap
2356 -100 after split brain, disconnect
2357 -1000 unrelated data
2359 static int drbd_uuid_compare(struct drbd_conf
*mdev
, int *rule_nr
) __must_hold(local
)
2364 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2365 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2368 if (self
== UUID_JUST_CREATED
&& peer
== UUID_JUST_CREATED
)
2372 if ((self
== UUID_JUST_CREATED
|| self
== (u64
)0) &&
2373 peer
!= UUID_JUST_CREATED
)
2377 if (self
!= UUID_JUST_CREATED
&&
2378 (peer
== UUID_JUST_CREATED
|| peer
== (u64
)0))
2382 int rct
, dc
; /* roles at crash time */
2384 if (mdev
->p_uuid
[UI_BITMAP
] == (u64
)0 && mdev
->ldev
->md
.uuid
[UI_BITMAP
] != (u64
)0) {
2386 if (mdev
->agreed_pro_version
< 91)
2389 if ((mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) &&
2390 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1))) {
2391 dev_info(DEV
, "was SyncSource, missed the resync finished event, corrected myself:\n");
2392 drbd_uuid_set_bm(mdev
, 0UL);
2394 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2395 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2398 dev_info(DEV
, "was SyncSource (peer failed to write sync_uuid)\n");
2405 if (mdev
->ldev
->md
.uuid
[UI_BITMAP
] == (u64
)0 && mdev
->p_uuid
[UI_BITMAP
] != (u64
)0) {
2407 if (mdev
->agreed_pro_version
< 91)
2410 if ((mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1)) &&
2411 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1))) {
2412 dev_info(DEV
, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2414 mdev
->p_uuid
[UI_HISTORY_START
+ 1] = mdev
->p_uuid
[UI_HISTORY_START
];
2415 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_BITMAP
];
2416 mdev
->p_uuid
[UI_BITMAP
] = 0UL;
2418 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2421 dev_info(DEV
, "was SyncTarget (failed to write sync_uuid)\n");
2428 /* Common power [off|failure] */
2429 rct
= (test_bit(CRASHED_PRIMARY
, &mdev
->flags
) ? 1 : 0) +
2430 (mdev
->p_uuid
[UI_FLAGS
] & 2);
2431 /* lowest bit is set when we were primary,
2432 * next bit (weight 2) is set when peer was primary */
2436 case 0: /* !self_pri && !peer_pri */ return 0;
2437 case 1: /* self_pri && !peer_pri */ return 1;
2438 case 2: /* !self_pri && peer_pri */ return -1;
2439 case 3: /* self_pri && peer_pri */
2440 dc
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
2446 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2451 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2453 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2454 peer
= mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1);
2456 /* The last P_SYNC_UUID did not get though. Undo the last start of
2457 resync as sync source modifications of the peer's UUIDs. */
2459 if (mdev
->agreed_pro_version
< 91)
2462 mdev
->p_uuid
[UI_BITMAP
] = mdev
->p_uuid
[UI_HISTORY_START
];
2463 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_HISTORY_START
+ 1];
2469 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2470 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2471 peer
= mdev
->p_uuid
[i
] & ~((u64
)1);
2477 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2478 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2483 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2485 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1);
2486 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2488 /* The last P_SYNC_UUID did not get though. Undo the last start of
2489 resync as sync source modifications of our UUIDs. */
2491 if (mdev
->agreed_pro_version
< 91)
2494 _drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
]);
2495 _drbd_uuid_set(mdev
, UI_HISTORY_START
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1]);
2497 dev_info(DEV
, "Undid last start of resync:\n");
2499 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2500 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2508 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2509 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2510 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2516 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2517 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2518 if (self
== peer
&& self
!= ((u64
)0))
2522 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2523 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2524 for (j
= UI_HISTORY_START
; j
<= UI_HISTORY_END
; j
++) {
2525 peer
= mdev
->p_uuid
[j
] & ~((u64
)1);
2534 /* drbd_sync_handshake() returns the new conn state on success, or
2535 CONN_MASK (-1) on failure.
2537 static enum drbd_conns
drbd_sync_handshake(struct drbd_conf
*mdev
, enum drbd_role peer_role
,
2538 enum drbd_disk_state peer_disk
) __must_hold(local
)
2541 enum drbd_conns rv
= C_MASK
;
2542 enum drbd_disk_state mydisk
;
2544 mydisk
= mdev
->state
.disk
;
2545 if (mydisk
== D_NEGOTIATING
)
2546 mydisk
= mdev
->new_state_tmp
.disk
;
2548 dev_info(DEV
, "drbd_sync_handshake:\n");
2549 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
, mdev
->comm_bm_set
, 0);
2550 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
,
2551 mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2553 hg
= drbd_uuid_compare(mdev
, &rule_nr
);
2555 dev_info(DEV
, "uuid_compare()=%d by rule %d\n", hg
, rule_nr
);
2558 dev_alert(DEV
, "Unrelated data, aborting!\n");
2562 dev_alert(DEV
, "To resolve this both sides have to support at least protocol\n");
2566 if ((mydisk
== D_INCONSISTENT
&& peer_disk
> D_INCONSISTENT
) ||
2567 (peer_disk
== D_INCONSISTENT
&& mydisk
> D_INCONSISTENT
)) {
2568 int f
= (hg
== -100) || abs(hg
) == 2;
2569 hg
= mydisk
> D_INCONSISTENT
? 1 : -1;
2572 dev_info(DEV
, "Becoming sync %s due to disk states.\n",
2573 hg
> 0 ? "source" : "target");
2577 drbd_khelper(mdev
, "initial-split-brain");
2579 if (hg
== 100 || (hg
== -100 && mdev
->net_conf
->always_asbp
)) {
2580 int pcount
= (mdev
->state
.role
== R_PRIMARY
)
2581 + (peer_role
== R_PRIMARY
);
2582 int forced
= (hg
== -100);
2586 hg
= drbd_asb_recover_0p(mdev
);
2589 hg
= drbd_asb_recover_1p(mdev
);
2592 hg
= drbd_asb_recover_2p(mdev
);
2595 if (abs(hg
) < 100) {
2596 dev_warn(DEV
, "Split-Brain detected, %d primaries, "
2597 "automatically solved. Sync from %s node\n",
2598 pcount
, (hg
< 0) ? "peer" : "this");
2600 dev_warn(DEV
, "Doing a full sync, since"
2601 " UUIDs where ambiguous.\n");
2608 if (mdev
->net_conf
->want_lose
&& !(mdev
->p_uuid
[UI_FLAGS
]&1))
2610 if (!mdev
->net_conf
->want_lose
&& (mdev
->p_uuid
[UI_FLAGS
]&1))
2614 dev_warn(DEV
, "Split-Brain detected, manually solved. "
2615 "Sync from %s node\n",
2616 (hg
< 0) ? "peer" : "this");
2620 /* FIXME this log message is not correct if we end up here
2621 * after an attempted attach on a diskless node.
2622 * We just refuse to attach -- well, we drop the "connection"
2623 * to that disk, in a way... */
2624 dev_alert(DEV
, "Split-Brain detected but unresolved, dropping connection!\n");
2625 drbd_khelper(mdev
, "split-brain");
2629 if (hg
> 0 && mydisk
<= D_INCONSISTENT
) {
2630 dev_err(DEV
, "I shall become SyncSource, but I am inconsistent!\n");
2634 if (hg
< 0 && /* by intention we do not use mydisk here. */
2635 mdev
->state
.role
== R_PRIMARY
&& mdev
->state
.disk
>= D_CONSISTENT
) {
2636 switch (mdev
->net_conf
->rr_conflict
) {
2637 case ASB_CALL_HELPER
:
2638 drbd_khelper(mdev
, "pri-lost");
2640 case ASB_DISCONNECT
:
2641 dev_err(DEV
, "I shall become SyncTarget, but I am primary!\n");
2644 dev_warn(DEV
, "Becoming SyncTarget, violating the stable-data"
2649 if (mdev
->net_conf
->dry_run
|| test_bit(CONN_DRY_RUN
, &mdev
->flags
)) {
2651 dev_info(DEV
, "dry-run connect: No resync, would become Connected immediately.\n");
2653 dev_info(DEV
, "dry-run connect: Would become %s, doing a %s resync.",
2654 drbd_conn_str(hg
> 0 ? C_SYNC_SOURCE
: C_SYNC_TARGET
),
2655 abs(hg
) >= 2 ? "full" : "bit-map based");
2660 dev_info(DEV
, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2661 if (drbd_bitmap_io(mdev
, &drbd_bmio_set_n_write
, "set_n_write from sync_handshake"))
2665 if (hg
> 0) { /* become sync source. */
2667 } else if (hg
< 0) { /* become sync target */
2671 if (drbd_bm_total_weight(mdev
)) {
2672 dev_info(DEV
, "No resync, but %lu bits in bitmap!\n",
2673 drbd_bm_total_weight(mdev
));
2680 /* returns 1 if invalid */
2681 static int cmp_after_sb(enum drbd_after_sb_p peer
, enum drbd_after_sb_p self
)
2683 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2684 if ((peer
== ASB_DISCARD_REMOTE
&& self
== ASB_DISCARD_LOCAL
) ||
2685 (self
== ASB_DISCARD_REMOTE
&& peer
== ASB_DISCARD_LOCAL
))
2688 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2689 if (peer
== ASB_DISCARD_REMOTE
|| peer
== ASB_DISCARD_LOCAL
||
2690 self
== ASB_DISCARD_REMOTE
|| self
== ASB_DISCARD_LOCAL
)
2693 /* everything else is valid if they are equal on both sides. */
2697 /* everything es is invalid. */
2701 static int receive_protocol(struct drbd_conf
*mdev
, struct p_header
*h
)
2703 struct p_protocol
*p
= (struct p_protocol
*)h
;
2704 int header_size
, data_size
;
2705 int p_proto
, p_after_sb_0p
, p_after_sb_1p
, p_after_sb_2p
;
2706 int p_want_lose
, p_two_primaries
, cf
;
2707 char p_integrity_alg
[SHARED_SECRET_MAX
] = "";
2709 header_size
= sizeof(*p
) - sizeof(*h
);
2710 data_size
= h
->length
- header_size
;
2712 if (drbd_recv(mdev
, h
->payload
, header_size
) != header_size
)
2715 p_proto
= be32_to_cpu(p
->protocol
);
2716 p_after_sb_0p
= be32_to_cpu(p
->after_sb_0p
);
2717 p_after_sb_1p
= be32_to_cpu(p
->after_sb_1p
);
2718 p_after_sb_2p
= be32_to_cpu(p
->after_sb_2p
);
2719 p_two_primaries
= be32_to_cpu(p
->two_primaries
);
2720 cf
= be32_to_cpu(p
->conn_flags
);
2721 p_want_lose
= cf
& CF_WANT_LOSE
;
2723 clear_bit(CONN_DRY_RUN
, &mdev
->flags
);
2725 if (cf
& CF_DRY_RUN
)
2726 set_bit(CONN_DRY_RUN
, &mdev
->flags
);
2728 if (p_proto
!= mdev
->net_conf
->wire_protocol
) {
2729 dev_err(DEV
, "incompatible communication protocols\n");
2733 if (cmp_after_sb(p_after_sb_0p
, mdev
->net_conf
->after_sb_0p
)) {
2734 dev_err(DEV
, "incompatible after-sb-0pri settings\n");
2738 if (cmp_after_sb(p_after_sb_1p
, mdev
->net_conf
->after_sb_1p
)) {
2739 dev_err(DEV
, "incompatible after-sb-1pri settings\n");
2743 if (cmp_after_sb(p_after_sb_2p
, mdev
->net_conf
->after_sb_2p
)) {
2744 dev_err(DEV
, "incompatible after-sb-2pri settings\n");
2748 if (p_want_lose
&& mdev
->net_conf
->want_lose
) {
2749 dev_err(DEV
, "both sides have the 'want_lose' flag set\n");
2753 if (p_two_primaries
!= mdev
->net_conf
->two_primaries
) {
2754 dev_err(DEV
, "incompatible setting of the two-primaries options\n");
2758 if (mdev
->agreed_pro_version
>= 87) {
2759 unsigned char *my_alg
= mdev
->net_conf
->integrity_alg
;
2761 if (drbd_recv(mdev
, p_integrity_alg
, data_size
) != data_size
)
2764 p_integrity_alg
[SHARED_SECRET_MAX
-1] = 0;
2765 if (strcmp(p_integrity_alg
, my_alg
)) {
2766 dev_err(DEV
, "incompatible setting of the data-integrity-alg\n");
2769 dev_info(DEV
, "data-integrity-alg: %s\n",
2770 my_alg
[0] ? my_alg
: (unsigned char *)"<not-used>");
2776 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2781 * input: alg name, feature name
2782 * return: NULL (alg name was "")
2783 * ERR_PTR(error) if something goes wrong
2784 * or the crypto hash ptr, if it worked out ok. */
2785 struct crypto_hash
*drbd_crypto_alloc_digest_safe(const struct drbd_conf
*mdev
,
2786 const char *alg
, const char *name
)
2788 struct crypto_hash
*tfm
;
2793 tfm
= crypto_alloc_hash(alg
, 0, CRYPTO_ALG_ASYNC
);
2795 dev_err(DEV
, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2796 alg
, name
, PTR_ERR(tfm
));
2799 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm
))) {
2800 crypto_free_hash(tfm
);
2801 dev_err(DEV
, "\"%s\" is not a digest (%s)\n", alg
, name
);
2802 return ERR_PTR(-EINVAL
);
2807 static int receive_SyncParam(struct drbd_conf
*mdev
, struct p_header
*h
)
2810 struct p_rs_param_95
*p
= (struct p_rs_param_95
*)h
;
2811 unsigned int header_size
, data_size
, exp_max_sz
;
2812 struct crypto_hash
*verify_tfm
= NULL
;
2813 struct crypto_hash
*csums_tfm
= NULL
;
2814 const int apv
= mdev
->agreed_pro_version
;
2815 int *rs_plan_s
= NULL
;
2818 exp_max_sz
= apv
<= 87 ? sizeof(struct p_rs_param
)
2819 : apv
== 88 ? sizeof(struct p_rs_param
)
2821 : apv
<= 94 ? sizeof(struct p_rs_param_89
)
2822 : /* apv >= 95 */ sizeof(struct p_rs_param_95
);
2824 if (h
->length
> exp_max_sz
) {
2825 dev_err(DEV
, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2826 h
->length
, exp_max_sz
);
2831 header_size
= sizeof(struct p_rs_param
) - sizeof(*h
);
2832 data_size
= h
->length
- header_size
;
2833 } else if (apv
<= 94) {
2834 header_size
= sizeof(struct p_rs_param_89
) - sizeof(*h
);
2835 data_size
= h
->length
- header_size
;
2836 D_ASSERT(data_size
== 0);
2838 header_size
= sizeof(struct p_rs_param_95
) - sizeof(*h
);
2839 data_size
= h
->length
- header_size
;
2840 D_ASSERT(data_size
== 0);
2843 /* initialize verify_alg and csums_alg */
2844 memset(p
->verify_alg
, 0, 2 * SHARED_SECRET_MAX
);
2846 if (drbd_recv(mdev
, h
->payload
, header_size
) != header_size
)
2849 mdev
->sync_conf
.rate
= be32_to_cpu(p
->rate
);
2853 if (data_size
> SHARED_SECRET_MAX
) {
2854 dev_err(DEV
, "verify-alg too long, "
2855 "peer wants %u, accepting only %u byte\n",
2856 data_size
, SHARED_SECRET_MAX
);
2860 if (drbd_recv(mdev
, p
->verify_alg
, data_size
) != data_size
)
2863 /* we expect NUL terminated string */
2864 /* but just in case someone tries to be evil */
2865 D_ASSERT(p
->verify_alg
[data_size
-1] == 0);
2866 p
->verify_alg
[data_size
-1] = 0;
2868 } else /* apv >= 89 */ {
2869 /* we still expect NUL terminated strings */
2870 /* but just in case someone tries to be evil */
2871 D_ASSERT(p
->verify_alg
[SHARED_SECRET_MAX
-1] == 0);
2872 D_ASSERT(p
->csums_alg
[SHARED_SECRET_MAX
-1] == 0);
2873 p
->verify_alg
[SHARED_SECRET_MAX
-1] = 0;
2874 p
->csums_alg
[SHARED_SECRET_MAX
-1] = 0;
2877 if (strcmp(mdev
->sync_conf
.verify_alg
, p
->verify_alg
)) {
2878 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
2879 dev_err(DEV
, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2880 mdev
->sync_conf
.verify_alg
, p
->verify_alg
);
2883 verify_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
2884 p
->verify_alg
, "verify-alg");
2885 if (IS_ERR(verify_tfm
)) {
2891 if (apv
>= 89 && strcmp(mdev
->sync_conf
.csums_alg
, p
->csums_alg
)) {
2892 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
2893 dev_err(DEV
, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2894 mdev
->sync_conf
.csums_alg
, p
->csums_alg
);
2897 csums_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
2898 p
->csums_alg
, "csums-alg");
2899 if (IS_ERR(csums_tfm
)) {
2906 mdev
->sync_conf
.rate
= be32_to_cpu(p
->rate
);
2907 mdev
->sync_conf
.c_plan_ahead
= be32_to_cpu(p
->c_plan_ahead
);
2908 mdev
->sync_conf
.c_delay_target
= be32_to_cpu(p
->c_delay_target
);
2909 mdev
->sync_conf
.c_fill_target
= be32_to_cpu(p
->c_fill_target
);
2910 mdev
->sync_conf
.c_max_rate
= be32_to_cpu(p
->c_max_rate
);
2912 fifo_size
= (mdev
->sync_conf
.c_plan_ahead
* 10 * SLEEP_TIME
) / HZ
;
2913 if (fifo_size
!= mdev
->rs_plan_s
.size
&& fifo_size
> 0) {
2914 rs_plan_s
= kzalloc(sizeof(int) * fifo_size
, GFP_KERNEL
);
2916 dev_err(DEV
, "kmalloc of fifo_buffer failed");
2922 spin_lock(&mdev
->peer_seq_lock
);
2923 /* lock against drbd_nl_syncer_conf() */
2925 strcpy(mdev
->sync_conf
.verify_alg
, p
->verify_alg
);
2926 mdev
->sync_conf
.verify_alg_len
= strlen(p
->verify_alg
) + 1;
2927 crypto_free_hash(mdev
->verify_tfm
);
2928 mdev
->verify_tfm
= verify_tfm
;
2929 dev_info(DEV
, "using verify-alg: \"%s\"\n", p
->verify_alg
);
2932 strcpy(mdev
->sync_conf
.csums_alg
, p
->csums_alg
);
2933 mdev
->sync_conf
.csums_alg_len
= strlen(p
->csums_alg
) + 1;
2934 crypto_free_hash(mdev
->csums_tfm
);
2935 mdev
->csums_tfm
= csums_tfm
;
2936 dev_info(DEV
, "using csums-alg: \"%s\"\n", p
->csums_alg
);
2938 if (fifo_size
!= mdev
->rs_plan_s
.size
) {
2939 kfree(mdev
->rs_plan_s
.values
);
2940 mdev
->rs_plan_s
.values
= rs_plan_s
;
2941 mdev
->rs_plan_s
.size
= fifo_size
;
2942 mdev
->rs_planed
= 0;
2944 spin_unlock(&mdev
->peer_seq_lock
);
2949 /* just for completeness: actually not needed,
2950 * as this is not reached if csums_tfm was ok. */
2951 crypto_free_hash(csums_tfm
);
2952 /* but free the verify_tfm again, if csums_tfm did not work out */
2953 crypto_free_hash(verify_tfm
);
2954 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2958 static void drbd_setup_order_type(struct drbd_conf
*mdev
, int peer
)
2960 /* sorry, we currently have no working implementation
2961 * of distributed TCQ */
2964 /* warn if the arguments differ by more than 12.5% */
2965 static void warn_if_differ_considerably(struct drbd_conf
*mdev
,
2966 const char *s
, sector_t a
, sector_t b
)
2969 if (a
== 0 || b
== 0)
2971 d
= (a
> b
) ? (a
- b
) : (b
- a
);
2972 if (d
> (a
>>3) || d
> (b
>>3))
2973 dev_warn(DEV
, "Considerable difference in %s: %llus vs. %llus\n", s
,
2974 (unsigned long long)a
, (unsigned long long)b
);
2977 static int receive_sizes(struct drbd_conf
*mdev
, struct p_header
*h
)
2979 struct p_sizes
*p
= (struct p_sizes
*)h
;
2980 enum determine_dev_size dd
= unchanged
;
2981 unsigned int max_seg_s
;
2982 sector_t p_size
, p_usize
, my_usize
;
2983 int ldsc
= 0; /* local disk size changed */
2984 enum dds_flags ddsf
;
2986 ERR_IF(h
->length
!= (sizeof(*p
)-sizeof(*h
))) return FALSE
;
2987 if (drbd_recv(mdev
, h
->payload
, h
->length
) != h
->length
)
2990 p_size
= be64_to_cpu(p
->d_size
);
2991 p_usize
= be64_to_cpu(p
->u_size
);
2993 if (p_size
== 0 && mdev
->state
.disk
== D_DISKLESS
) {
2994 dev_err(DEV
, "some backing storage is needed\n");
2995 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2999 /* just store the peer's disk size for now.
3000 * we still need to figure out whether we accept that. */
3001 mdev
->p_size
= p_size
;
3003 #define min_not_zero(l, r) (l == 0) ? r : ((r == 0) ? l : min(l, r))
3004 if (get_ldev(mdev
)) {
3005 warn_if_differ_considerably(mdev
, "lower level device sizes",
3006 p_size
, drbd_get_max_capacity(mdev
->ldev
));
3007 warn_if_differ_considerably(mdev
, "user requested size",
3008 p_usize
, mdev
->ldev
->dc
.disk_size
);
3010 /* if this is the first connect, or an otherwise expected
3011 * param exchange, choose the minimum */
3012 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
)
3013 p_usize
= min_not_zero((sector_t
)mdev
->ldev
->dc
.disk_size
,
3016 my_usize
= mdev
->ldev
->dc
.disk_size
;
3018 if (mdev
->ldev
->dc
.disk_size
!= p_usize
) {
3019 mdev
->ldev
->dc
.disk_size
= p_usize
;
3020 dev_info(DEV
, "Peer sets u_size to %lu sectors\n",
3021 (unsigned long)mdev
->ldev
->dc
.disk_size
);
3024 /* Never shrink a device with usable data during connect.
3025 But allow online shrinking if we are connected. */
3026 if (drbd_new_dev_size(mdev
, mdev
->ldev
, 0) <
3027 drbd_get_capacity(mdev
->this_bdev
) &&
3028 mdev
->state
.disk
>= D_OUTDATED
&&
3029 mdev
->state
.conn
< C_CONNECTED
) {
3030 dev_err(DEV
, "The peer's disk size is too small!\n");
3031 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3032 mdev
->ldev
->dc
.disk_size
= my_usize
;
3040 ddsf
= be16_to_cpu(p
->dds_flags
);
3041 if (get_ldev(mdev
)) {
3042 dd
= drbd_determin_dev_size(mdev
, ddsf
);
3044 if (dd
== dev_size_error
)
3048 /* I am diskless, need to accept the peer's size. */
3049 drbd_set_my_capacity(mdev
, p_size
);
3052 if (get_ldev(mdev
)) {
3053 if (mdev
->ldev
->known_size
!= drbd_get_capacity(mdev
->ldev
->backing_bdev
)) {
3054 mdev
->ldev
->known_size
= drbd_get_capacity(mdev
->ldev
->backing_bdev
);
3058 if (mdev
->agreed_pro_version
< 94)
3059 max_seg_s
= be32_to_cpu(p
->max_segment_size
);
3060 else /* drbd 8.3.8 onwards */
3061 max_seg_s
= DRBD_MAX_SEGMENT_SIZE
;
3063 if (max_seg_s
!= queue_max_segment_size(mdev
->rq_queue
))
3064 drbd_setup_queue_param(mdev
, max_seg_s
);
3066 drbd_setup_order_type(mdev
, be16_to_cpu(p
->queue_order_type
));
3070 if (mdev
->state
.conn
> C_WF_REPORT_PARAMS
) {
3071 if (be64_to_cpu(p
->c_size
) !=
3072 drbd_get_capacity(mdev
->this_bdev
) || ldsc
) {
3073 /* we have different sizes, probably peer
3074 * needs to know my new size... */
3075 drbd_send_sizes(mdev
, 0, ddsf
);
3077 if (test_and_clear_bit(RESIZE_PENDING
, &mdev
->flags
) ||
3078 (dd
== grew
&& mdev
->state
.conn
== C_CONNECTED
)) {
3079 if (mdev
->state
.pdsk
>= D_INCONSISTENT
&&
3080 mdev
->state
.disk
>= D_INCONSISTENT
) {
3081 if (ddsf
& DDSF_NO_RESYNC
)
3082 dev_info(DEV
, "Resync of new storage suppressed with --assume-clean\n");
3084 resync_after_online_grow(mdev
);
3086 set_bit(RESYNC_AFTER_NEG
, &mdev
->flags
);
3093 static int receive_uuids(struct drbd_conf
*mdev
, struct p_header
*h
)
3095 struct p_uuids
*p
= (struct p_uuids
*)h
;
3099 ERR_IF(h
->length
!= (sizeof(*p
)-sizeof(*h
))) return FALSE
;
3100 if (drbd_recv(mdev
, h
->payload
, h
->length
) != h
->length
)
3103 p_uuid
= kmalloc(sizeof(u64
)*UI_EXTENDED_SIZE
, GFP_NOIO
);
3105 for (i
= UI_CURRENT
; i
< UI_EXTENDED_SIZE
; i
++)
3106 p_uuid
[i
] = be64_to_cpu(p
->uuid
[i
]);
3108 kfree(mdev
->p_uuid
);
3109 mdev
->p_uuid
= p_uuid
;
3111 if (mdev
->state
.conn
< C_CONNECTED
&&
3112 mdev
->state
.disk
< D_INCONSISTENT
&&
3113 mdev
->state
.role
== R_PRIMARY
&&
3114 (mdev
->ed_uuid
& ~((u64
)1)) != (p_uuid
[UI_CURRENT
] & ~((u64
)1))) {
3115 dev_err(DEV
, "Can only connect to data with current UUID=%016llX\n",
3116 (unsigned long long)mdev
->ed_uuid
);
3117 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3121 if (get_ldev(mdev
)) {
3122 int skip_initial_sync
=
3123 mdev
->state
.conn
== C_CONNECTED
&&
3124 mdev
->agreed_pro_version
>= 90 &&
3125 mdev
->ldev
->md
.uuid
[UI_CURRENT
] == UUID_JUST_CREATED
&&
3126 (p_uuid
[UI_FLAGS
] & 8);
3127 if (skip_initial_sync
) {
3128 dev_info(DEV
, "Accepted new current UUID, preparing to skip initial sync\n");
3129 drbd_bitmap_io(mdev
, &drbd_bmio_clear_n_write
,
3130 "clear_n_write from receive_uuids");
3131 _drbd_uuid_set(mdev
, UI_CURRENT
, p_uuid
[UI_CURRENT
]);
3132 _drbd_uuid_set(mdev
, UI_BITMAP
, 0);
3133 _drbd_set_state(_NS2(mdev
, disk
, D_UP_TO_DATE
, pdsk
, D_UP_TO_DATE
),
3138 } else if (mdev
->state
.disk
< D_INCONSISTENT
&&
3139 mdev
->state
.role
== R_PRIMARY
) {
3140 /* I am a diskless primary, the peer just created a new current UUID
3142 drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3145 /* Before we test for the disk state, we should wait until an eventually
3146 ongoing cluster wide state change is finished. That is important if
3147 we are primary and are detaching from our disk. We need to see the
3148 new disk state... */
3149 wait_event(mdev
->misc_wait
, !test_bit(CLUSTER_ST_CHANGE
, &mdev
->flags
));
3150 if (mdev
->state
.conn
>= C_CONNECTED
&& mdev
->state
.disk
< D_INCONSISTENT
)
3151 drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3157 * convert_state() - Converts the peer's view of the cluster state to our point of view
3158 * @ps: The state as seen by the peer.
3160 static union drbd_state
convert_state(union drbd_state ps
)
3162 union drbd_state ms
;
3164 static enum drbd_conns c_tab
[] = {
3165 [C_CONNECTED
] = C_CONNECTED
,
3167 [C_STARTING_SYNC_S
] = C_STARTING_SYNC_T
,
3168 [C_STARTING_SYNC_T
] = C_STARTING_SYNC_S
,
3169 [C_DISCONNECTING
] = C_TEAR_DOWN
, /* C_NETWORK_FAILURE, */
3170 [C_VERIFY_S
] = C_VERIFY_T
,
3176 ms
.conn
= c_tab
[ps
.conn
];
3181 ms
.peer_isp
= (ps
.aftr_isp
| ps
.user_isp
);
3186 static int receive_req_state(struct drbd_conf
*mdev
, struct p_header
*h
)
3188 struct p_req_state
*p
= (struct p_req_state
*)h
;
3189 union drbd_state mask
, val
;
3192 ERR_IF(h
->length
!= (sizeof(*p
)-sizeof(*h
))) return FALSE
;
3193 if (drbd_recv(mdev
, h
->payload
, h
->length
) != h
->length
)
3196 mask
.i
= be32_to_cpu(p
->mask
);
3197 val
.i
= be32_to_cpu(p
->val
);
3199 if (test_bit(DISCARD_CONCURRENT
, &mdev
->flags
) &&
3200 test_bit(CLUSTER_ST_CHANGE
, &mdev
->flags
)) {
3201 drbd_send_sr_reply(mdev
, SS_CONCURRENT_ST_CHG
);
3205 mask
= convert_state(mask
);
3206 val
= convert_state(val
);
3208 rv
= drbd_change_state(mdev
, CS_VERBOSE
, mask
, val
);
3210 drbd_send_sr_reply(mdev
, rv
);
3216 static int receive_state(struct drbd_conf
*mdev
, struct p_header
*h
)
3218 struct p_state
*p
= (struct p_state
*)h
;
3219 enum drbd_conns nconn
, oconn
;
3220 union drbd_state ns
, peer_state
;
3221 enum drbd_disk_state real_peer_disk
;
3222 enum chg_state_flags cs_flags
;
3225 ERR_IF(h
->length
!= (sizeof(*p
)-sizeof(*h
)))
3228 if (drbd_recv(mdev
, h
->payload
, h
->length
) != h
->length
)
3231 peer_state
.i
= be32_to_cpu(p
->state
);
3233 real_peer_disk
= peer_state
.disk
;
3234 if (peer_state
.disk
== D_NEGOTIATING
) {
3235 real_peer_disk
= mdev
->p_uuid
[UI_FLAGS
] & 4 ? D_INCONSISTENT
: D_CONSISTENT
;
3236 dev_info(DEV
, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk
));
3239 spin_lock_irq(&mdev
->req_lock
);
3241 oconn
= nconn
= mdev
->state
.conn
;
3242 spin_unlock_irq(&mdev
->req_lock
);
3244 if (nconn
== C_WF_REPORT_PARAMS
)
3245 nconn
= C_CONNECTED
;
3247 if (mdev
->p_uuid
&& peer_state
.disk
>= D_NEGOTIATING
&&
3248 get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3249 int cr
; /* consider resync */
3251 /* if we established a new connection */
3252 cr
= (oconn
< C_CONNECTED
);
3253 /* if we had an established connection
3254 * and one of the nodes newly attaches a disk */
3255 cr
|= (oconn
== C_CONNECTED
&&
3256 (peer_state
.disk
== D_NEGOTIATING
||
3257 mdev
->state
.disk
== D_NEGOTIATING
));
3258 /* if we have both been inconsistent, and the peer has been
3259 * forced to be UpToDate with --overwrite-data */
3260 cr
|= test_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3261 /* if we had been plain connected, and the admin requested to
3262 * start a sync by "invalidate" or "invalidate-remote" */
3263 cr
|= (oconn
== C_CONNECTED
&&
3264 (peer_state
.conn
>= C_STARTING_SYNC_S
&&
3265 peer_state
.conn
<= C_WF_BITMAP_T
));
3268 nconn
= drbd_sync_handshake(mdev
, peer_state
.role
, real_peer_disk
);
3271 if (nconn
== C_MASK
) {
3272 nconn
= C_CONNECTED
;
3273 if (mdev
->state
.disk
== D_NEGOTIATING
) {
3274 drbd_force_state(mdev
, NS(disk
, D_DISKLESS
));
3275 } else if (peer_state
.disk
== D_NEGOTIATING
) {
3276 dev_err(DEV
, "Disk attach process on the peer node was aborted.\n");
3277 peer_state
.disk
= D_DISKLESS
;
3278 real_peer_disk
= D_DISKLESS
;
3280 if (test_and_clear_bit(CONN_DRY_RUN
, &mdev
->flags
))
3282 D_ASSERT(oconn
== C_WF_REPORT_PARAMS
);
3283 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3289 spin_lock_irq(&mdev
->req_lock
);
3290 if (mdev
->state
.conn
!= oconn
)
3292 clear_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3293 ns
.i
= mdev
->state
.i
;
3295 ns
.peer
= peer_state
.role
;
3296 ns
.pdsk
= real_peer_disk
;
3297 ns
.peer_isp
= (peer_state
.aftr_isp
| peer_state
.user_isp
);
3298 if ((nconn
== C_CONNECTED
|| nconn
== C_WF_BITMAP_S
) && ns
.disk
== D_NEGOTIATING
)
3299 ns
.disk
= mdev
->new_state_tmp
.disk
;
3300 cs_flags
= CS_VERBOSE
+ (oconn
< C_CONNECTED
&& nconn
>= C_CONNECTED
? 0 : CS_HARD
);
3301 if (ns
.pdsk
== D_CONSISTENT
&& ns
.susp
&& nconn
== C_CONNECTED
&& oconn
< C_CONNECTED
&&
3302 test_bit(NEW_CUR_UUID
, &mdev
->flags
)) {
3303 /* Do not allow tl_restart(resend) for a rebooted peer. We can only allow this
3304 for temporal network outages! */
3305 spin_unlock_irq(&mdev
->req_lock
);
3306 dev_err(DEV
, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3308 drbd_uuid_new_current(mdev
);
3309 clear_bit(NEW_CUR_UUID
, &mdev
->flags
);
3310 drbd_force_state(mdev
, NS2(conn
, C_PROTOCOL_ERROR
, susp
, 0));
3313 rv
= _drbd_set_state(mdev
, ns
, cs_flags
, NULL
);
3315 spin_unlock_irq(&mdev
->req_lock
);
3317 if (rv
< SS_SUCCESS
) {
3318 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3322 if (oconn
> C_WF_REPORT_PARAMS
) {
3323 if (nconn
> C_CONNECTED
&& peer_state
.conn
<= C_CONNECTED
&&
3324 peer_state
.disk
!= D_NEGOTIATING
) {
3325 /* we want resync, peer has not yet decided to sync... */
3326 /* Nowadays only used when forcing a node into primary role and
3327 setting its disk to UpToDate with that */
3328 drbd_send_uuids(mdev
);
3329 drbd_send_state(mdev
);
3333 mdev
->net_conf
->want_lose
= 0;
3335 drbd_md_sync(mdev
); /* update connected indicator, la_size, ... */
3340 static int receive_sync_uuid(struct drbd_conf
*mdev
, struct p_header
*h
)
3342 struct p_rs_uuid
*p
= (struct p_rs_uuid
*)h
;
3344 wait_event(mdev
->misc_wait
,
3345 mdev
->state
.conn
== C_WF_SYNC_UUID
||
3346 mdev
->state
.conn
< C_CONNECTED
||
3347 mdev
->state
.disk
< D_NEGOTIATING
);
3349 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3351 ERR_IF(h
->length
!= (sizeof(*p
)-sizeof(*h
))) return FALSE
;
3352 if (drbd_recv(mdev
, h
->payload
, h
->length
) != h
->length
)
3355 /* Here the _drbd_uuid_ functions are right, current should
3356 _not_ be rotated into the history */
3357 if (get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3358 _drbd_uuid_set(mdev
, UI_CURRENT
, be64_to_cpu(p
->uuid
));
3359 _drbd_uuid_set(mdev
, UI_BITMAP
, 0UL);
3361 drbd_start_resync(mdev
, C_SYNC_TARGET
);
3365 dev_err(DEV
, "Ignoring SyncUUID packet!\n");
3370 enum receive_bitmap_ret
{ OK
, DONE
, FAILED
};
3372 static enum receive_bitmap_ret
3373 receive_bitmap_plain(struct drbd_conf
*mdev
, struct p_header
*h
,
3374 unsigned long *buffer
, struct bm_xfer_ctx
*c
)
3376 unsigned num_words
= min_t(size_t, BM_PACKET_WORDS
, c
->bm_words
- c
->word_offset
);
3377 unsigned want
= num_words
* sizeof(long);
3379 if (want
!= h
->length
) {
3380 dev_err(DEV
, "%s:want (%u) != h->length (%u)\n", __func__
, want
, h
->length
);
3385 if (drbd_recv(mdev
, buffer
, want
) != want
)
3388 drbd_bm_merge_lel(mdev
, c
->word_offset
, num_words
, buffer
);
3390 c
->word_offset
+= num_words
;
3391 c
->bit_offset
= c
->word_offset
* BITS_PER_LONG
;
3392 if (c
->bit_offset
> c
->bm_bits
)
3393 c
->bit_offset
= c
->bm_bits
;
3398 static enum receive_bitmap_ret
3399 recv_bm_rle_bits(struct drbd_conf
*mdev
,
3400 struct p_compressed_bm
*p
,
3401 struct bm_xfer_ctx
*c
)
3403 struct bitstream bs
;
3407 unsigned long s
= c
->bit_offset
;
3409 int len
= p
->head
.length
- (sizeof(*p
) - sizeof(p
->head
));
3410 int toggle
= DCBP_get_start(p
);
3414 bitstream_init(&bs
, p
->code
, len
, DCBP_get_pad_bits(p
));
3416 bits
= bitstream_get_bits(&bs
, &look_ahead
, 64);
3420 for (have
= bits
; have
> 0; s
+= rl
, toggle
= !toggle
) {
3421 bits
= vli_decode_bits(&rl
, look_ahead
);
3427 if (e
>= c
->bm_bits
) {
3428 dev_err(DEV
, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e
);
3431 _drbd_bm_set_bits(mdev
, s
, e
);
3435 dev_err(DEV
, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3436 have
, bits
, look_ahead
,
3437 (unsigned int)(bs
.cur
.b
- p
->code
),
3438 (unsigned int)bs
.buf_len
);
3441 look_ahead
>>= bits
;
3444 bits
= bitstream_get_bits(&bs
, &tmp
, 64 - have
);
3447 look_ahead
|= tmp
<< have
;
3452 bm_xfer_ctx_bit_to_word_offset(c
);
3454 return (s
== c
->bm_bits
) ? DONE
: OK
;
3457 static enum receive_bitmap_ret
3458 decode_bitmap_c(struct drbd_conf
*mdev
,
3459 struct p_compressed_bm
*p
,
3460 struct bm_xfer_ctx
*c
)
3462 if (DCBP_get_code(p
) == RLE_VLI_Bits
)
3463 return recv_bm_rle_bits(mdev
, p
, c
);
3465 /* other variants had been implemented for evaluation,
3466 * but have been dropped as this one turned out to be "best"
3467 * during all our tests. */
3469 dev_err(DEV
, "receive_bitmap_c: unknown encoding %u\n", p
->encoding
);
3470 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3474 void INFO_bm_xfer_stats(struct drbd_conf
*mdev
,
3475 const char *direction
, struct bm_xfer_ctx
*c
)
3477 /* what would it take to transfer it "plaintext" */
3478 unsigned plain
= sizeof(struct p_header
) *
3479 ((c
->bm_words
+BM_PACKET_WORDS
-1)/BM_PACKET_WORDS
+1)
3480 + c
->bm_words
* sizeof(long);
3481 unsigned total
= c
->bytes
[0] + c
->bytes
[1];
3484 /* total can not be zero. but just in case: */
3488 /* don't report if not compressed */
3492 /* total < plain. check for overflow, still */
3493 r
= (total
> UINT_MAX
/1000) ? (total
/ (plain
/1000))
3494 : (1000 * total
/ plain
);
3500 dev_info(DEV
, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3501 "total %u; compression: %u.%u%%\n",
3503 c
->bytes
[1], c
->packets
[1],
3504 c
->bytes
[0], c
->packets
[0],
3505 total
, r
/10, r
% 10);
3508 /* Since we are processing the bitfield from lower addresses to higher,
3509 it does not matter if the process it in 32 bit chunks or 64 bit
3510 chunks as long as it is little endian. (Understand it as byte stream,
3511 beginning with the lowest byte...) If we would use big endian
3512 we would need to process it from the highest address to the lowest,
3513 in order to be agnostic to the 32 vs 64 bits issue.
3515 returns 0 on failure, 1 if we successfully received it. */
3516 static int receive_bitmap(struct drbd_conf
*mdev
, struct p_header
*h
)
3518 struct bm_xfer_ctx c
;
3520 enum receive_bitmap_ret ret
;
3523 wait_event(mdev
->misc_wait
, !atomic_read(&mdev
->ap_bio_cnt
));
3525 drbd_bm_lock(mdev
, "receive bitmap");
3527 /* maybe we should use some per thread scratch page,
3528 * and allocate that during initial device creation? */
3529 buffer
= (unsigned long *) __get_free_page(GFP_NOIO
);
3531 dev_err(DEV
, "failed to allocate one page buffer in %s\n", __func__
);
3535 c
= (struct bm_xfer_ctx
) {
3536 .bm_bits
= drbd_bm_bits(mdev
),
3537 .bm_words
= drbd_bm_words(mdev
),
3541 if (h
->command
== P_BITMAP
) {
3542 ret
= receive_bitmap_plain(mdev
, h
, buffer
, &c
);
3543 } else if (h
->command
== P_COMPRESSED_BITMAP
) {
3544 /* MAYBE: sanity check that we speak proto >= 90,
3545 * and the feature is enabled! */
3546 struct p_compressed_bm
*p
;
3548 if (h
->length
> BM_PACKET_PAYLOAD_BYTES
) {
3549 dev_err(DEV
, "ReportCBitmap packet too large\n");
3552 /* use the page buff */
3554 memcpy(p
, h
, sizeof(*h
));
3555 if (drbd_recv(mdev
, p
->head
.payload
, h
->length
) != h
->length
)
3557 if (p
->head
.length
<= (sizeof(*p
) - sizeof(p
->head
))) {
3558 dev_err(DEV
, "ReportCBitmap packet too small (l:%u)\n", p
->head
.length
);
3561 ret
= decode_bitmap_c(mdev
, p
, &c
);
3563 dev_warn(DEV
, "receive_bitmap: h->command neither ReportBitMap nor ReportCBitMap (is 0x%x)", h
->command
);
3567 c
.packets
[h
->command
== P_BITMAP
]++;
3568 c
.bytes
[h
->command
== P_BITMAP
] += sizeof(struct p_header
) + h
->length
;
3573 if (!drbd_recv_header(mdev
, h
))
3575 } while (ret
== OK
);
3579 INFO_bm_xfer_stats(mdev
, "receive", &c
);
3581 if (mdev
->state
.conn
== C_WF_BITMAP_T
) {
3582 ok
= !drbd_send_bitmap(mdev
);
3585 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
3586 ok
= _drbd_request_state(mdev
, NS(conn
, C_WF_SYNC_UUID
), CS_VERBOSE
);
3587 D_ASSERT(ok
== SS_SUCCESS
);
3588 } else if (mdev
->state
.conn
!= C_WF_BITMAP_S
) {
3589 /* admin may have requested C_DISCONNECTING,
3590 * other threads may have noticed network errors */
3591 dev_info(DEV
, "unexpected cstate (%s) in receive_bitmap\n",
3592 drbd_conn_str(mdev
->state
.conn
));
3597 drbd_bm_unlock(mdev
);
3598 if (ok
&& mdev
->state
.conn
== C_WF_BITMAP_S
)
3599 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
3600 free_page((unsigned long) buffer
);
3604 static int receive_skip_(struct drbd_conf
*mdev
, struct p_header
*h
, int silent
)
3606 /* TODO zero copy sink :) */
3607 static char sink
[128];
3611 dev_warn(DEV
, "skipping unknown optional packet type %d, l: %d!\n",
3612 h
->command
, h
->length
);
3616 want
= min_t(int, size
, sizeof(sink
));
3617 r
= drbd_recv(mdev
, sink
, want
);
3618 ERR_IF(r
<= 0) break;
3624 static int receive_skip(struct drbd_conf
*mdev
, struct p_header
*h
)
3626 return receive_skip_(mdev
, h
, 0);
3629 static int receive_skip_silent(struct drbd_conf
*mdev
, struct p_header
*h
)
3631 return receive_skip_(mdev
, h
, 1);
3634 static int receive_UnplugRemote(struct drbd_conf
*mdev
, struct p_header
*h
)
3636 if (mdev
->state
.disk
>= D_INCONSISTENT
)
3639 /* Make sure we've acked all the TCP data associated
3640 * with the data requests being unplugged */
3641 drbd_tcp_quickack(mdev
->data
.socket
);
3646 typedef int (*drbd_cmd_handler_f
)(struct drbd_conf
*, struct p_header
*);
3648 static drbd_cmd_handler_f drbd_default_handler
[] = {
3649 [P_DATA
] = receive_Data
,
3650 [P_DATA_REPLY
] = receive_DataReply
,
3651 [P_RS_DATA_REPLY
] = receive_RSDataReply
,
3652 [P_BARRIER
] = receive_Barrier
,
3653 [P_BITMAP
] = receive_bitmap
,
3654 [P_COMPRESSED_BITMAP
] = receive_bitmap
,
3655 [P_UNPLUG_REMOTE
] = receive_UnplugRemote
,
3656 [P_DATA_REQUEST
] = receive_DataRequest
,
3657 [P_RS_DATA_REQUEST
] = receive_DataRequest
,
3658 [P_SYNC_PARAM
] = receive_SyncParam
,
3659 [P_SYNC_PARAM89
] = receive_SyncParam
,
3660 [P_PROTOCOL
] = receive_protocol
,
3661 [P_UUIDS
] = receive_uuids
,
3662 [P_SIZES
] = receive_sizes
,
3663 [P_STATE
] = receive_state
,
3664 [P_STATE_CHG_REQ
] = receive_req_state
,
3665 [P_SYNC_UUID
] = receive_sync_uuid
,
3666 [P_OV_REQUEST
] = receive_DataRequest
,
3667 [P_OV_REPLY
] = receive_DataRequest
,
3668 [P_CSUM_RS_REQUEST
] = receive_DataRequest
,
3669 [P_DELAY_PROBE
] = receive_skip_silent
,
3670 /* anything missing from this table is in
3671 * the asender_tbl, see get_asender_cmd */
3675 static drbd_cmd_handler_f
*drbd_cmd_handler
= drbd_default_handler
;
3676 static drbd_cmd_handler_f
*drbd_opt_cmd_handler
;
3678 static void drbdd(struct drbd_conf
*mdev
)
3680 drbd_cmd_handler_f handler
;
3681 struct p_header
*header
= &mdev
->data
.rbuf
.header
;
3683 while (get_t_state(&mdev
->receiver
) == Running
) {
3684 drbd_thread_current_set_cpu(mdev
);
3685 if (!drbd_recv_header(mdev
, header
)) {
3686 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3690 if (header
->command
< P_MAX_CMD
)
3691 handler
= drbd_cmd_handler
[header
->command
];
3692 else if (P_MAY_IGNORE
< header
->command
3693 && header
->command
< P_MAX_OPT_CMD
)
3694 handler
= drbd_opt_cmd_handler
[header
->command
-P_MAY_IGNORE
];
3695 else if (header
->command
> P_MAX_OPT_CMD
)
3696 handler
= receive_skip
;
3700 if (unlikely(!handler
)) {
3701 dev_err(DEV
, "unknown packet type %d, l: %d!\n",
3702 header
->command
, header
->length
);
3703 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3706 if (unlikely(!handler(mdev
, header
))) {
3707 dev_err(DEV
, "error receiving %s, l: %d!\n",
3708 cmdname(header
->command
), header
->length
);
3709 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3715 void drbd_flush_workqueue(struct drbd_conf
*mdev
)
3717 struct drbd_wq_barrier barr
;
3719 barr
.w
.cb
= w_prev_work_done
;
3720 init_completion(&barr
.done
);
3721 drbd_queue_work(&mdev
->data
.work
, &barr
.w
);
3722 wait_for_completion(&barr
.done
);
3725 void drbd_free_tl_hash(struct drbd_conf
*mdev
)
3727 struct hlist_head
*h
;
3729 spin_lock_irq(&mdev
->req_lock
);
3731 if (!mdev
->tl_hash
|| mdev
->state
.conn
!= C_STANDALONE
) {
3732 spin_unlock_irq(&mdev
->req_lock
);
3736 for (h
= mdev
->ee_hash
; h
< mdev
->ee_hash
+ mdev
->ee_hash_s
; h
++)
3738 dev_err(DEV
, "ASSERT FAILED ee_hash[%u].first == %p, expected NULL\n",
3739 (int)(h
- mdev
->ee_hash
), h
->first
);
3740 kfree(mdev
->ee_hash
);
3741 mdev
->ee_hash
= NULL
;
3742 mdev
->ee_hash_s
= 0;
3745 for (h
= mdev
->tl_hash
; h
< mdev
->tl_hash
+ mdev
->tl_hash_s
; h
++)
3747 dev_err(DEV
, "ASSERT FAILED tl_hash[%u] == %p, expected NULL\n",
3748 (int)(h
- mdev
->tl_hash
), h
->first
);
3749 kfree(mdev
->tl_hash
);
3750 mdev
->tl_hash
= NULL
;
3751 mdev
->tl_hash_s
= 0;
3752 spin_unlock_irq(&mdev
->req_lock
);
3755 static void drbd_disconnect(struct drbd_conf
*mdev
)
3757 enum drbd_fencing_p fp
;
3758 union drbd_state os
, ns
;
3759 int rv
= SS_UNKNOWN_ERROR
;
3762 if (mdev
->state
.conn
== C_STANDALONE
)
3764 if (mdev
->state
.conn
>= C_WF_CONNECTION
)
3765 dev_err(DEV
, "ASSERT FAILED cstate = %s, expected < WFConnection\n",
3766 drbd_conn_str(mdev
->state
.conn
));
3768 /* asender does not clean up anything. it must not interfere, either */
3769 drbd_thread_stop(&mdev
->asender
);
3770 drbd_free_sock(mdev
);
3772 /* wait for current activity to cease. */
3773 spin_lock_irq(&mdev
->req_lock
);
3774 _drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
3775 _drbd_wait_ee_list_empty(mdev
, &mdev
->sync_ee
);
3776 _drbd_wait_ee_list_empty(mdev
, &mdev
->read_ee
);
3777 spin_unlock_irq(&mdev
->req_lock
);
3779 /* We do not have data structures that would allow us to
3780 * get the rs_pending_cnt down to 0 again.
3781 * * On C_SYNC_TARGET we do not have any data structures describing
3782 * the pending RSDataRequest's we have sent.
3783 * * On C_SYNC_SOURCE there is no data structure that tracks
3784 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
3785 * And no, it is not the sum of the reference counts in the
3786 * resync_LRU. The resync_LRU tracks the whole operation including
3787 * the disk-IO, while the rs_pending_cnt only tracks the blocks
3789 drbd_rs_cancel_all(mdev
);
3791 mdev
->rs_failed
= 0;
3792 atomic_set(&mdev
->rs_pending_cnt
, 0);
3793 wake_up(&mdev
->misc_wait
);
3795 /* make sure syncer is stopped and w_resume_next_sg queued */
3796 del_timer_sync(&mdev
->resync_timer
);
3797 set_bit(STOP_SYNC_TIMER
, &mdev
->flags
);
3798 resync_timer_fn((unsigned long)mdev
);
3800 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
3801 * w_make_resync_request etc. which may still be on the worker queue
3802 * to be "canceled" */
3803 drbd_flush_workqueue(mdev
);
3805 /* This also does reclaim_net_ee(). If we do this too early, we might
3806 * miss some resync ee and pages.*/
3807 drbd_process_done_ee(mdev
);
3809 kfree(mdev
->p_uuid
);
3810 mdev
->p_uuid
= NULL
;
3812 if (!mdev
->state
.susp
)
3815 dev_info(DEV
, "Connection closed\n");
3820 if (get_ldev(mdev
)) {
3821 fp
= mdev
->ldev
->dc
.fencing
;
3825 if (mdev
->state
.role
== R_PRIMARY
&& fp
>= FP_RESOURCE
&& mdev
->state
.pdsk
>= D_UNKNOWN
)
3826 drbd_try_outdate_peer_async(mdev
);
3828 spin_lock_irq(&mdev
->req_lock
);
3830 if (os
.conn
>= C_UNCONNECTED
) {
3831 /* Do not restart in case we are C_DISCONNECTING */
3833 ns
.conn
= C_UNCONNECTED
;
3834 rv
= _drbd_set_state(mdev
, ns
, CS_VERBOSE
, NULL
);
3836 spin_unlock_irq(&mdev
->req_lock
);
3838 if (os
.conn
== C_DISCONNECTING
) {
3839 wait_event(mdev
->net_cnt_wait
, atomic_read(&mdev
->net_cnt
) == 0);
3841 if (!mdev
->state
.susp
) {
3842 /* we must not free the tl_hash
3843 * while application io is still on the fly */
3844 wait_event(mdev
->misc_wait
, !atomic_read(&mdev
->ap_bio_cnt
));
3845 drbd_free_tl_hash(mdev
);
3848 crypto_free_hash(mdev
->cram_hmac_tfm
);
3849 mdev
->cram_hmac_tfm
= NULL
;
3851 kfree(mdev
->net_conf
);
3852 mdev
->net_conf
= NULL
;
3853 drbd_request_state(mdev
, NS(conn
, C_STANDALONE
));
3856 /* tcp_close and release of sendpage pages can be deferred. I don't
3857 * want to use SO_LINGER, because apparently it can be deferred for
3858 * more than 20 seconds (longest time I checked).
3860 * Actually we don't care for exactly when the network stack does its
3861 * put_page(), but release our reference on these pages right here.
3863 i
= drbd_release_ee(mdev
, &mdev
->net_ee
);
3865 dev_info(DEV
, "net_ee not empty, killed %u entries\n", i
);
3866 i
= atomic_read(&mdev
->pp_in_use
);
3868 dev_info(DEV
, "pp_in_use = %d, expected 0\n", i
);
3870 D_ASSERT(list_empty(&mdev
->read_ee
));
3871 D_ASSERT(list_empty(&mdev
->active_ee
));
3872 D_ASSERT(list_empty(&mdev
->sync_ee
));
3873 D_ASSERT(list_empty(&mdev
->done_ee
));
3875 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
3876 atomic_set(&mdev
->current_epoch
->epoch_size
, 0);
3877 D_ASSERT(list_empty(&mdev
->current_epoch
->list
));
3881 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
3882 * we can agree on is stored in agreed_pro_version.
3884 * feature flags and the reserved array should be enough room for future
3885 * enhancements of the handshake protocol, and possible plugins...
3887 * for now, they are expected to be zero, but ignored.
3889 static int drbd_send_handshake(struct drbd_conf
*mdev
)
3891 /* ASSERT current == mdev->receiver ... */
3892 struct p_handshake
*p
= &mdev
->data
.sbuf
.handshake
;
3895 if (mutex_lock_interruptible(&mdev
->data
.mutex
)) {
3896 dev_err(DEV
, "interrupted during initial handshake\n");
3897 return 0; /* interrupted. not ok. */
3900 if (mdev
->data
.socket
== NULL
) {
3901 mutex_unlock(&mdev
->data
.mutex
);
3905 memset(p
, 0, sizeof(*p
));
3906 p
->protocol_min
= cpu_to_be32(PRO_VERSION_MIN
);
3907 p
->protocol_max
= cpu_to_be32(PRO_VERSION_MAX
);
3908 ok
= _drbd_send_cmd( mdev
, mdev
->data
.socket
, P_HAND_SHAKE
,
3909 (struct p_header
*)p
, sizeof(*p
), 0 );
3910 mutex_unlock(&mdev
->data
.mutex
);
3916 * 1 yes, we have a valid connection
3917 * 0 oops, did not work out, please try again
3918 * -1 peer talks different language,
3919 * no point in trying again, please go standalone.
3921 static int drbd_do_handshake(struct drbd_conf
*mdev
)
3923 /* ASSERT current == mdev->receiver ... */
3924 struct p_handshake
*p
= &mdev
->data
.rbuf
.handshake
;
3925 const int expect
= sizeof(struct p_handshake
)
3926 -sizeof(struct p_header
);
3929 rv
= drbd_send_handshake(mdev
);
3933 rv
= drbd_recv_header(mdev
, &p
->head
);
3937 if (p
->head
.command
!= P_HAND_SHAKE
) {
3938 dev_err(DEV
, "expected HandShake packet, received: %s (0x%04x)\n",
3939 cmdname(p
->head
.command
), p
->head
.command
);
3943 if (p
->head
.length
!= expect
) {
3944 dev_err(DEV
, "expected HandShake length: %u, received: %u\n",
3945 expect
, p
->head
.length
);
3949 rv
= drbd_recv(mdev
, &p
->head
.payload
, expect
);
3952 dev_err(DEV
, "short read receiving handshake packet: l=%u\n", rv
);
3956 p
->protocol_min
= be32_to_cpu(p
->protocol_min
);
3957 p
->protocol_max
= be32_to_cpu(p
->protocol_max
);
3958 if (p
->protocol_max
== 0)
3959 p
->protocol_max
= p
->protocol_min
;
3961 if (PRO_VERSION_MAX
< p
->protocol_min
||
3962 PRO_VERSION_MIN
> p
->protocol_max
)
3965 mdev
->agreed_pro_version
= min_t(int, PRO_VERSION_MAX
, p
->protocol_max
);
3967 dev_info(DEV
, "Handshake successful: "
3968 "Agreed network protocol version %d\n", mdev
->agreed_pro_version
);
3973 dev_err(DEV
, "incompatible DRBD dialects: "
3974 "I support %d-%d, peer supports %d-%d\n",
3975 PRO_VERSION_MIN
, PRO_VERSION_MAX
,
3976 p
->protocol_min
, p
->protocol_max
);
3980 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
3981 static int drbd_do_auth(struct drbd_conf
*mdev
)
3983 dev_err(DEV
, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
3984 dev_err(DEV
, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
3988 #define CHALLENGE_LEN 64
3992 0 - failed, try again (network error),
3993 -1 - auth failed, don't try again.
3996 static int drbd_do_auth(struct drbd_conf
*mdev
)
3998 char my_challenge
[CHALLENGE_LEN
]; /* 64 Bytes... */
3999 struct scatterlist sg
;
4000 char *response
= NULL
;
4001 char *right_response
= NULL
;
4002 char *peers_ch
= NULL
;
4004 unsigned int key_len
= strlen(mdev
->net_conf
->shared_secret
);
4005 unsigned int resp_size
;
4006 struct hash_desc desc
;
4009 desc
.tfm
= mdev
->cram_hmac_tfm
;
4012 rv
= crypto_hash_setkey(mdev
->cram_hmac_tfm
,
4013 (u8
*)mdev
->net_conf
->shared_secret
, key_len
);
4015 dev_err(DEV
, "crypto_hash_setkey() failed with %d\n", rv
);
4020 get_random_bytes(my_challenge
, CHALLENGE_LEN
);
4022 rv
= drbd_send_cmd2(mdev
, P_AUTH_CHALLENGE
, my_challenge
, CHALLENGE_LEN
);
4026 rv
= drbd_recv_header(mdev
, &p
);
4030 if (p
.command
!= P_AUTH_CHALLENGE
) {
4031 dev_err(DEV
, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4032 cmdname(p
.command
), p
.command
);
4037 if (p
.length
> CHALLENGE_LEN
*2) {
4038 dev_err(DEV
, "expected AuthChallenge payload too big.\n");
4043 peers_ch
= kmalloc(p
.length
, GFP_NOIO
);
4044 if (peers_ch
== NULL
) {
4045 dev_err(DEV
, "kmalloc of peers_ch failed\n");
4050 rv
= drbd_recv(mdev
, peers_ch
, p
.length
);
4052 if (rv
!= p
.length
) {
4053 dev_err(DEV
, "short read AuthChallenge: l=%u\n", rv
);
4058 resp_size
= crypto_hash_digestsize(mdev
->cram_hmac_tfm
);
4059 response
= kmalloc(resp_size
, GFP_NOIO
);
4060 if (response
== NULL
) {
4061 dev_err(DEV
, "kmalloc of response failed\n");
4066 sg_init_table(&sg
, 1);
4067 sg_set_buf(&sg
, peers_ch
, p
.length
);
4069 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, response
);
4071 dev_err(DEV
, "crypto_hash_digest() failed with %d\n", rv
);
4076 rv
= drbd_send_cmd2(mdev
, P_AUTH_RESPONSE
, response
, resp_size
);
4080 rv
= drbd_recv_header(mdev
, &p
);
4084 if (p
.command
!= P_AUTH_RESPONSE
) {
4085 dev_err(DEV
, "expected AuthResponse packet, received: %s (0x%04x)\n",
4086 cmdname(p
.command
), p
.command
);
4091 if (p
.length
!= resp_size
) {
4092 dev_err(DEV
, "expected AuthResponse payload of wrong size\n");
4097 rv
= drbd_recv(mdev
, response
, resp_size
);
4099 if (rv
!= resp_size
) {
4100 dev_err(DEV
, "short read receiving AuthResponse: l=%u\n", rv
);
4105 right_response
= kmalloc(resp_size
, GFP_NOIO
);
4106 if (right_response
== NULL
) {
4107 dev_err(DEV
, "kmalloc of right_response failed\n");
4112 sg_set_buf(&sg
, my_challenge
, CHALLENGE_LEN
);
4114 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, right_response
);
4116 dev_err(DEV
, "crypto_hash_digest() failed with %d\n", rv
);
4121 rv
= !memcmp(response
, right_response
, resp_size
);
4124 dev_info(DEV
, "Peer authenticated using %d bytes of '%s' HMAC\n",
4125 resp_size
, mdev
->net_conf
->cram_hmac_alg
);
4132 kfree(right_response
);
4138 int drbdd_init(struct drbd_thread
*thi
)
4140 struct drbd_conf
*mdev
= thi
->mdev
;
4141 unsigned int minor
= mdev_to_minor(mdev
);
4144 sprintf(current
->comm
, "drbd%d_receiver", minor
);
4146 dev_info(DEV
, "receiver (re)started\n");
4149 h
= drbd_connect(mdev
);
4151 drbd_disconnect(mdev
);
4152 __set_current_state(TASK_INTERRUPTIBLE
);
4153 schedule_timeout(HZ
);
4156 dev_warn(DEV
, "Discarding network configuration.\n");
4157 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
4162 if (get_net_conf(mdev
)) {
4168 drbd_disconnect(mdev
);
4170 dev_info(DEV
, "receiver terminated\n");
4174 /* ********* acknowledge sender ******** */
4176 static int got_RqSReply(struct drbd_conf
*mdev
, struct p_header
*h
)
4178 struct p_req_state_reply
*p
= (struct p_req_state_reply
*)h
;
4180 int retcode
= be32_to_cpu(p
->retcode
);
4182 if (retcode
>= SS_SUCCESS
) {
4183 set_bit(CL_ST_CHG_SUCCESS
, &mdev
->flags
);
4185 set_bit(CL_ST_CHG_FAIL
, &mdev
->flags
);
4186 dev_err(DEV
, "Requested state change failed by peer: %s (%d)\n",
4187 drbd_set_st_err_str(retcode
), retcode
);
4189 wake_up(&mdev
->state_wait
);
4194 static int got_Ping(struct drbd_conf
*mdev
, struct p_header
*h
)
4196 return drbd_send_ping_ack(mdev
);
4200 static int got_PingAck(struct drbd_conf
*mdev
, struct p_header
*h
)
4202 /* restore idle timeout */
4203 mdev
->meta
.socket
->sk
->sk_rcvtimeo
= mdev
->net_conf
->ping_int
*HZ
;
4204 if (!test_and_set_bit(GOT_PING_ACK
, &mdev
->flags
))
4205 wake_up(&mdev
->misc_wait
);
4210 static int got_IsInSync(struct drbd_conf
*mdev
, struct p_header
*h
)
4212 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4213 sector_t sector
= be64_to_cpu(p
->sector
);
4214 int blksize
= be32_to_cpu(p
->blksize
);
4216 D_ASSERT(mdev
->agreed_pro_version
>= 89);
4218 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4220 drbd_rs_complete_io(mdev
, sector
);
4221 drbd_set_in_sync(mdev
, sector
, blksize
);
4222 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4223 mdev
->rs_same_csum
+= (blksize
>> BM_BLOCK_SHIFT
);
4224 dec_rs_pending(mdev
);
4225 atomic_add(blksize
>> 9, &mdev
->rs_sect_in
);
4230 /* when we receive the ACK for a write request,
4231 * verify that we actually know about it */
4232 static struct drbd_request
*_ack_id_to_req(struct drbd_conf
*mdev
,
4233 u64 id
, sector_t sector
)
4235 struct hlist_head
*slot
= tl_hash_slot(mdev
, sector
);
4236 struct hlist_node
*n
;
4237 struct drbd_request
*req
;
4239 hlist_for_each_entry(req
, n
, slot
, colision
) {
4240 if ((unsigned long)req
== (unsigned long)id
) {
4241 if (req
->sector
!= sector
) {
4242 dev_err(DEV
, "_ack_id_to_req: found req %p but it has "
4243 "wrong sector (%llus versus %llus)\n", req
,
4244 (unsigned long long)req
->sector
,
4245 (unsigned long long)sector
);
4251 dev_err(DEV
, "_ack_id_to_req: failed to find req %p, sector %llus in list\n",
4252 (void *)(unsigned long)id
, (unsigned long long)sector
);
4256 typedef struct drbd_request
*(req_validator_fn
)
4257 (struct drbd_conf
*mdev
, u64 id
, sector_t sector
);
4259 static int validate_req_change_req_state(struct drbd_conf
*mdev
,
4260 u64 id
, sector_t sector
, req_validator_fn validator
,
4261 const char *func
, enum drbd_req_event what
)
4263 struct drbd_request
*req
;
4264 struct bio_and_error m
;
4266 spin_lock_irq(&mdev
->req_lock
);
4267 req
= validator(mdev
, id
, sector
);
4268 if (unlikely(!req
)) {
4269 spin_unlock_irq(&mdev
->req_lock
);
4270 dev_err(DEV
, "%s: got a corrupt block_id/sector pair\n", func
);
4273 __req_mod(req
, what
, &m
);
4274 spin_unlock_irq(&mdev
->req_lock
);
4277 complete_master_bio(mdev
, &m
);
4281 static int got_BlockAck(struct drbd_conf
*mdev
, struct p_header
*h
)
4283 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4284 sector_t sector
= be64_to_cpu(p
->sector
);
4285 int blksize
= be32_to_cpu(p
->blksize
);
4286 enum drbd_req_event what
;
4288 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4290 if (is_syncer_block_id(p
->block_id
)) {
4291 drbd_set_in_sync(mdev
, sector
, blksize
);
4292 dec_rs_pending(mdev
);
4295 switch (be16_to_cpu(h
->command
)) {
4296 case P_RS_WRITE_ACK
:
4297 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4298 what
= write_acked_by_peer_and_sis
;
4301 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4302 what
= write_acked_by_peer
;
4305 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_B
);
4306 what
= recv_acked_by_peer
;
4309 D_ASSERT(mdev
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4310 what
= conflict_discarded_by_peer
;
4317 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4318 _ack_id_to_req
, __func__
, what
);
4321 static int got_NegAck(struct drbd_conf
*mdev
, struct p_header
*h
)
4323 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4324 sector_t sector
= be64_to_cpu(p
->sector
);
4326 if (__ratelimit(&drbd_ratelimit_state
))
4327 dev_warn(DEV
, "Got NegAck packet. Peer is in troubles?\n");
4329 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4331 if (is_syncer_block_id(p
->block_id
)) {
4332 int size
= be32_to_cpu(p
->blksize
);
4333 dec_rs_pending(mdev
);
4334 drbd_rs_failed_io(mdev
, sector
, size
);
4337 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4338 _ack_id_to_req
, __func__
, neg_acked
);
4341 static int got_NegDReply(struct drbd_conf
*mdev
, struct p_header
*h
)
4343 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4344 sector_t sector
= be64_to_cpu(p
->sector
);
4346 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4347 dev_err(DEV
, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4348 (unsigned long long)sector
, be32_to_cpu(p
->blksize
));
4350 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4351 _ar_id_to_req
, __func__
, neg_acked
);
4354 static int got_NegRSDReply(struct drbd_conf
*mdev
, struct p_header
*h
)
4358 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4360 sector
= be64_to_cpu(p
->sector
);
4361 size
= be32_to_cpu(p
->blksize
);
4363 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4365 dec_rs_pending(mdev
);
4367 if (get_ldev_if_state(mdev
, D_FAILED
)) {
4368 drbd_rs_complete_io(mdev
, sector
);
4369 drbd_rs_failed_io(mdev
, sector
, size
);
4376 static int got_BarrierAck(struct drbd_conf
*mdev
, struct p_header
*h
)
4378 struct p_barrier_ack
*p
= (struct p_barrier_ack
*)h
;
4380 tl_release(mdev
, p
->barrier
, be32_to_cpu(p
->set_size
));
4385 static int got_OVResult(struct drbd_conf
*mdev
, struct p_header
*h
)
4387 struct p_block_ack
*p
= (struct p_block_ack
*)h
;
4388 struct drbd_work
*w
;
4392 sector
= be64_to_cpu(p
->sector
);
4393 size
= be32_to_cpu(p
->blksize
);
4395 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4397 if (be64_to_cpu(p
->block_id
) == ID_OUT_OF_SYNC
)
4398 drbd_ov_oos_found(mdev
, sector
, size
);
4402 drbd_rs_complete_io(mdev
, sector
);
4403 dec_rs_pending(mdev
);
4405 if (--mdev
->ov_left
== 0) {
4406 w
= kmalloc(sizeof(*w
), GFP_NOIO
);
4408 w
->cb
= w_ov_finished
;
4409 drbd_queue_work_front(&mdev
->data
.work
, w
);
4411 dev_err(DEV
, "kmalloc(w) failed.");
4413 drbd_resync_finished(mdev
);
4419 static int got_something_to_ignore_m(struct drbd_conf
*mdev
, struct p_header
*h
)
4425 struct asender_cmd
{
4427 int (*process
)(struct drbd_conf
*mdev
, struct p_header
*h
);
4430 static struct asender_cmd
*get_asender_cmd(int cmd
)
4432 static struct asender_cmd asender_tbl
[] = {
4433 /* anything missing from this table is in
4434 * the drbd_cmd_handler (drbd_default_handler) table,
4435 * see the beginning of drbdd() */
4436 [P_PING
] = { sizeof(struct p_header
), got_Ping
},
4437 [P_PING_ACK
] = { sizeof(struct p_header
), got_PingAck
},
4438 [P_RECV_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4439 [P_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4440 [P_RS_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4441 [P_DISCARD_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4442 [P_NEG_ACK
] = { sizeof(struct p_block_ack
), got_NegAck
},
4443 [P_NEG_DREPLY
] = { sizeof(struct p_block_ack
), got_NegDReply
},
4444 [P_NEG_RS_DREPLY
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
4445 [P_OV_RESULT
] = { sizeof(struct p_block_ack
), got_OVResult
},
4446 [P_BARRIER_ACK
] = { sizeof(struct p_barrier_ack
), got_BarrierAck
},
4447 [P_STATE_CHG_REPLY
] = { sizeof(struct p_req_state_reply
), got_RqSReply
},
4448 [P_RS_IS_IN_SYNC
] = { sizeof(struct p_block_ack
), got_IsInSync
},
4449 [P_DELAY_PROBE
] = { sizeof(struct p_delay_probe
), got_something_to_ignore_m
},
4450 [P_MAX_CMD
] = { 0, NULL
},
4452 if (cmd
> P_MAX_CMD
|| asender_tbl
[cmd
].process
== NULL
)
4454 return &asender_tbl
[cmd
];
4457 int drbd_asender(struct drbd_thread
*thi
)
4459 struct drbd_conf
*mdev
= thi
->mdev
;
4460 struct p_header
*h
= &mdev
->meta
.rbuf
.header
;
4461 struct asender_cmd
*cmd
= NULL
;
4466 int expect
= sizeof(struct p_header
);
4469 sprintf(current
->comm
, "drbd%d_asender", mdev_to_minor(mdev
));
4471 current
->policy
= SCHED_RR
; /* Make this a realtime task! */
4472 current
->rt_priority
= 2; /* more important than all other tasks */
4474 while (get_t_state(thi
) == Running
) {
4475 drbd_thread_current_set_cpu(mdev
);
4476 if (test_and_clear_bit(SEND_PING
, &mdev
->flags
)) {
4477 ERR_IF(!drbd_send_ping(mdev
)) goto reconnect
;
4478 mdev
->meta
.socket
->sk
->sk_rcvtimeo
=
4479 mdev
->net_conf
->ping_timeo
*HZ
/10;
4482 /* conditionally cork;
4483 * it may hurt latency if we cork without much to send */
4484 if (!mdev
->net_conf
->no_cork
&&
4485 3 < atomic_read(&mdev
->unacked_cnt
))
4486 drbd_tcp_cork(mdev
->meta
.socket
);
4488 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4489 flush_signals(current
);
4490 if (!drbd_process_done_ee(mdev
)) {
4491 dev_err(DEV
, "process_done_ee() = NOT_OK\n");
4494 /* to avoid race with newly queued ACKs */
4495 set_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4496 spin_lock_irq(&mdev
->req_lock
);
4497 empty
= list_empty(&mdev
->done_ee
);
4498 spin_unlock_irq(&mdev
->req_lock
);
4499 /* new ack may have been queued right here,
4500 * but then there is also a signal pending,
4501 * and we start over... */
4505 /* but unconditionally uncork unless disabled */
4506 if (!mdev
->net_conf
->no_cork
)
4507 drbd_tcp_uncork(mdev
->meta
.socket
);
4509 /* short circuit, recv_msg would return EINTR anyways. */
4510 if (signal_pending(current
))
4513 rv
= drbd_recv_short(mdev
, mdev
->meta
.socket
,
4514 buf
, expect
-received
, 0);
4515 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4517 flush_signals(current
);
4520 * -EINTR (on meta) we got a signal
4521 * -EAGAIN (on meta) rcvtimeo expired
4522 * -ECONNRESET other side closed the connection
4523 * -ERESTARTSYS (on data) we got a signal
4524 * rv < 0 other than above: unexpected error!
4525 * rv == expected: full header or command
4526 * rv < expected: "woken" by signal during receive
4527 * rv == 0 : "connection shut down by peer"
4529 if (likely(rv
> 0)) {
4532 } else if (rv
== 0) {
4533 dev_err(DEV
, "meta connection shut down by peer.\n");
4535 } else if (rv
== -EAGAIN
) {
4536 if (mdev
->meta
.socket
->sk
->sk_rcvtimeo
==
4537 mdev
->net_conf
->ping_timeo
*HZ
/10) {
4538 dev_err(DEV
, "PingAck did not arrive in time.\n");
4541 set_bit(SEND_PING
, &mdev
->flags
);
4543 } else if (rv
== -EINTR
) {
4546 dev_err(DEV
, "sock_recvmsg returned %d\n", rv
);
4550 if (received
== expect
&& cmd
== NULL
) {
4551 if (unlikely(h
->magic
!= BE_DRBD_MAGIC
)) {
4552 dev_err(DEV
, "magic?? on meta m: 0x%lx c: %d l: %d\n",
4553 (long)be32_to_cpu(h
->magic
),
4554 h
->command
, h
->length
);
4557 cmd
= get_asender_cmd(be16_to_cpu(h
->command
));
4558 len
= be16_to_cpu(h
->length
);
4559 if (unlikely(cmd
== NULL
)) {
4560 dev_err(DEV
, "unknown command?? on meta m: 0x%lx c: %d l: %d\n",
4561 (long)be32_to_cpu(h
->magic
),
4562 h
->command
, h
->length
);
4565 expect
= cmd
->pkt_size
;
4566 ERR_IF(len
!= expect
-sizeof(struct p_header
))
4569 if (received
== expect
) {
4570 D_ASSERT(cmd
!= NULL
);
4571 if (!cmd
->process(mdev
, h
))
4576 expect
= sizeof(struct p_header
);
4583 drbd_force_state(mdev
, NS(conn
, C_NETWORK_FAILURE
));
4587 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
4589 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4591 D_ASSERT(mdev
->state
.conn
< C_CONNECTED
);
4592 dev_info(DEV
, "asender terminated\n");