4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
28 #include <asm/uaccess.h>
31 #include <linux/drbd.h>
33 #include <linux/file.h>
36 #include <linux/memcontrol.h>
37 #include <linux/mm_inline.h>
38 #include <linux/slab.h>
39 #include <linux/pkt_sched.h>
40 #define __KERNEL_SYSCALLS__
41 #include <linux/unistd.h>
42 #include <linux/vmalloc.h>
43 #include <linux/random.h>
44 #include <linux/string.h>
45 #include <linux/scatterlist.h>
64 static int drbd_do_features(struct drbd_tconn
*tconn
);
65 static int drbd_do_auth(struct drbd_tconn
*tconn
);
66 static int drbd_disconnected(struct drbd_conf
*mdev
);
68 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_tconn
*, struct drbd_epoch
*, enum epoch_event
);
69 static int e_end_block(struct drbd_work
*, int);
72 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
75 * some helper functions to deal with single linked page lists,
76 * page->private being our "next" pointer.
79 /* If at least n pages are linked at head, get n pages off.
80 * Otherwise, don't modify head, and return NULL.
81 * Locking is the responsibility of the caller.
83 static struct page
*page_chain_del(struct page
**head
, int n
)
97 tmp
= page_chain_next(page
);
99 break; /* found sufficient pages */
101 /* insufficient pages, don't use any of them. */
106 /* add end of list marker for the returned list */
107 set_page_private(page
, 0);
108 /* actual return value, and adjustment of head */
114 /* may be used outside of locks to find the tail of a (usually short)
115 * "private" page chain, before adding it back to a global chain head
116 * with page_chain_add() under a spinlock. */
117 static struct page
*page_chain_tail(struct page
*page
, int *len
)
121 while ((tmp
= page_chain_next(page
)))
128 static int page_chain_free(struct page
*page
)
132 page_chain_for_each_safe(page
, tmp
) {
139 static void page_chain_add(struct page
**head
,
140 struct page
*chain_first
, struct page
*chain_last
)
144 tmp
= page_chain_tail(chain_first
, NULL
);
145 BUG_ON(tmp
!= chain_last
);
148 /* add chain to head */
149 set_page_private(chain_last
, (unsigned long)*head
);
153 static struct page
*__drbd_alloc_pages(struct drbd_conf
*mdev
,
156 struct page
*page
= NULL
;
157 struct page
*tmp
= NULL
;
160 /* Yes, testing drbd_pp_vacant outside the lock is racy.
161 * So what. It saves a spin_lock. */
162 if (drbd_pp_vacant
>= number
) {
163 spin_lock(&drbd_pp_lock
);
164 page
= page_chain_del(&drbd_pp_pool
, number
);
166 drbd_pp_vacant
-= number
;
167 spin_unlock(&drbd_pp_lock
);
172 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
173 * "criss-cross" setup, that might cause write-out on some other DRBD,
174 * which in turn might block on the other node at this very place. */
175 for (i
= 0; i
< number
; i
++) {
176 tmp
= alloc_page(GFP_TRY
);
179 set_page_private(tmp
, (unsigned long)page
);
186 /* Not enough pages immediately available this time.
187 * No need to jump around here, drbd_alloc_pages will retry this
188 * function "soon". */
190 tmp
= page_chain_tail(page
, NULL
);
191 spin_lock(&drbd_pp_lock
);
192 page_chain_add(&drbd_pp_pool
, page
, tmp
);
194 spin_unlock(&drbd_pp_lock
);
199 static void reclaim_finished_net_peer_reqs(struct drbd_conf
*mdev
,
200 struct list_head
*to_be_freed
)
202 struct drbd_peer_request
*peer_req
;
203 struct list_head
*le
, *tle
;
205 /* The EEs are always appended to the end of the list. Since
206 they are sent in order over the wire, they have to finish
207 in order. As soon as we see the first not finished we can
208 stop to examine the list... */
210 list_for_each_safe(le
, tle
, &mdev
->net_ee
) {
211 peer_req
= list_entry(le
, struct drbd_peer_request
, w
.list
);
212 if (drbd_peer_req_has_active_page(peer_req
))
214 list_move(le
, to_be_freed
);
218 static void drbd_kick_lo_and_reclaim_net(struct drbd_conf
*mdev
)
220 LIST_HEAD(reclaimed
);
221 struct drbd_peer_request
*peer_req
, *t
;
223 spin_lock_irq(&mdev
->tconn
->req_lock
);
224 reclaim_finished_net_peer_reqs(mdev
, &reclaimed
);
225 spin_unlock_irq(&mdev
->tconn
->req_lock
);
227 list_for_each_entry_safe(peer_req
, t
, &reclaimed
, w
.list
)
228 drbd_free_net_peer_req(mdev
, peer_req
);
232 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
233 * @mdev: DRBD device.
234 * @number: number of pages requested
235 * @retry: whether to retry, if not enough pages are available right now
237 * Tries to allocate number pages, first from our own page pool, then from
238 * the kernel, unless this allocation would exceed the max_buffers setting.
239 * Possibly retry until DRBD frees sufficient pages somewhere else.
241 * Returns a page chain linked via page->private.
243 struct page
*drbd_alloc_pages(struct drbd_conf
*mdev
, unsigned int number
,
246 struct page
*page
= NULL
;
251 /* Yes, we may run up to @number over max_buffers. If we
252 * follow it strictly, the admin will get it wrong anyways. */
254 nc
= rcu_dereference(mdev
->tconn
->net_conf
);
255 mxb
= nc
? nc
->max_buffers
: 1000000;
258 if (atomic_read(&mdev
->pp_in_use
) < mxb
)
259 page
= __drbd_alloc_pages(mdev
, number
);
261 while (page
== NULL
) {
262 prepare_to_wait(&drbd_pp_wait
, &wait
, TASK_INTERRUPTIBLE
);
264 drbd_kick_lo_and_reclaim_net(mdev
);
266 if (atomic_read(&mdev
->pp_in_use
) < mxb
) {
267 page
= __drbd_alloc_pages(mdev
, number
);
275 if (signal_pending(current
)) {
276 dev_warn(DEV
, "drbd_alloc_pages interrupted!\n");
282 finish_wait(&drbd_pp_wait
, &wait
);
285 atomic_add(number
, &mdev
->pp_in_use
);
289 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
290 * Is also used from inside an other spin_lock_irq(&mdev->tconn->req_lock);
291 * Either links the page chain back to the global pool,
292 * or returns all pages to the system. */
293 static void drbd_free_pages(struct drbd_conf
*mdev
, struct page
*page
, int is_net
)
295 atomic_t
*a
= is_net
? &mdev
->pp_in_use_by_net
: &mdev
->pp_in_use
;
298 if (drbd_pp_vacant
> (DRBD_MAX_BIO_SIZE
/PAGE_SIZE
) * minor_count
)
299 i
= page_chain_free(page
);
302 tmp
= page_chain_tail(page
, &i
);
303 spin_lock(&drbd_pp_lock
);
304 page_chain_add(&drbd_pp_pool
, page
, tmp
);
306 spin_unlock(&drbd_pp_lock
);
308 i
= atomic_sub_return(i
, a
);
310 dev_warn(DEV
, "ASSERTION FAILED: %s: %d < 0\n",
311 is_net
? "pp_in_use_by_net" : "pp_in_use", i
);
312 wake_up(&drbd_pp_wait
);
316 You need to hold the req_lock:
317 _drbd_wait_ee_list_empty()
319 You must not have the req_lock:
321 drbd_alloc_peer_req()
322 drbd_free_peer_reqs()
324 drbd_finish_peer_reqs()
326 drbd_wait_ee_list_empty()
329 struct drbd_peer_request
*
330 drbd_alloc_peer_req(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
331 unsigned int data_size
, gfp_t gfp_mask
) __must_hold(local
)
333 struct drbd_peer_request
*peer_req
;
335 unsigned nr_pages
= (data_size
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
337 if (drbd_insert_fault(mdev
, DRBD_FAULT_AL_EE
))
340 peer_req
= mempool_alloc(drbd_ee_mempool
, gfp_mask
& ~__GFP_HIGHMEM
);
342 if (!(gfp_mask
& __GFP_NOWARN
))
343 dev_err(DEV
, "%s: allocation failed\n", __func__
);
347 page
= drbd_alloc_pages(mdev
, nr_pages
, (gfp_mask
& __GFP_WAIT
));
351 drbd_clear_interval(&peer_req
->i
);
352 peer_req
->i
.size
= data_size
;
353 peer_req
->i
.sector
= sector
;
354 peer_req
->i
.local
= false;
355 peer_req
->i
.waiting
= false;
357 peer_req
->epoch
= NULL
;
358 peer_req
->w
.mdev
= mdev
;
359 peer_req
->pages
= page
;
360 atomic_set(&peer_req
->pending_bios
, 0);
363 * The block_id is opaque to the receiver. It is not endianness
364 * converted, and sent back to the sender unchanged.
366 peer_req
->block_id
= id
;
371 mempool_free(peer_req
, drbd_ee_mempool
);
375 void __drbd_free_peer_req(struct drbd_conf
*mdev
, struct drbd_peer_request
*peer_req
,
378 if (peer_req
->flags
& EE_HAS_DIGEST
)
379 kfree(peer_req
->digest
);
380 drbd_free_pages(mdev
, peer_req
->pages
, is_net
);
381 D_ASSERT(atomic_read(&peer_req
->pending_bios
) == 0);
382 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
383 mempool_free(peer_req
, drbd_ee_mempool
);
386 int drbd_free_peer_reqs(struct drbd_conf
*mdev
, struct list_head
*list
)
388 LIST_HEAD(work_list
);
389 struct drbd_peer_request
*peer_req
, *t
;
391 int is_net
= list
== &mdev
->net_ee
;
393 spin_lock_irq(&mdev
->tconn
->req_lock
);
394 list_splice_init(list
, &work_list
);
395 spin_unlock_irq(&mdev
->tconn
->req_lock
);
397 list_for_each_entry_safe(peer_req
, t
, &work_list
, w
.list
) {
398 __drbd_free_peer_req(mdev
, peer_req
, is_net
);
405 * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
407 static int drbd_finish_peer_reqs(struct drbd_conf
*mdev
)
409 LIST_HEAD(work_list
);
410 LIST_HEAD(reclaimed
);
411 struct drbd_peer_request
*peer_req
, *t
;
414 spin_lock_irq(&mdev
->tconn
->req_lock
);
415 reclaim_finished_net_peer_reqs(mdev
, &reclaimed
);
416 list_splice_init(&mdev
->done_ee
, &work_list
);
417 spin_unlock_irq(&mdev
->tconn
->req_lock
);
419 list_for_each_entry_safe(peer_req
, t
, &reclaimed
, w
.list
)
420 drbd_free_net_peer_req(mdev
, peer_req
);
422 /* possible callbacks here:
423 * e_end_block, and e_end_resync_block, e_send_discard_write.
424 * all ignore the last argument.
426 list_for_each_entry_safe(peer_req
, t
, &work_list
, w
.list
) {
429 /* list_del not necessary, next/prev members not touched */
430 err2
= peer_req
->w
.cb(&peer_req
->w
, !!err
);
433 drbd_free_peer_req(mdev
, peer_req
);
435 wake_up(&mdev
->ee_wait
);
440 static void _drbd_wait_ee_list_empty(struct drbd_conf
*mdev
,
441 struct list_head
*head
)
445 /* avoids spin_lock/unlock
446 * and calling prepare_to_wait in the fast path */
447 while (!list_empty(head
)) {
448 prepare_to_wait(&mdev
->ee_wait
, &wait
, TASK_UNINTERRUPTIBLE
);
449 spin_unlock_irq(&mdev
->tconn
->req_lock
);
451 finish_wait(&mdev
->ee_wait
, &wait
);
452 spin_lock_irq(&mdev
->tconn
->req_lock
);
456 static void drbd_wait_ee_list_empty(struct drbd_conf
*mdev
,
457 struct list_head
*head
)
459 spin_lock_irq(&mdev
->tconn
->req_lock
);
460 _drbd_wait_ee_list_empty(mdev
, head
);
461 spin_unlock_irq(&mdev
->tconn
->req_lock
);
464 /* see also kernel_accept; which is only present since 2.6.18.
465 * also we want to log which part of it failed, exactly */
466 static int drbd_accept(const char **what
, struct socket
*sock
, struct socket
**newsock
)
468 struct sock
*sk
= sock
->sk
;
472 err
= sock
->ops
->listen(sock
, 5);
476 *what
= "sock_create_lite";
477 err
= sock_create_lite(sk
->sk_family
, sk
->sk_type
, sk
->sk_protocol
,
483 err
= sock
->ops
->accept(sock
, *newsock
, 0);
485 sock_release(*newsock
);
489 (*newsock
)->ops
= sock
->ops
;
495 static int drbd_recv_short(struct socket
*sock
, void *buf
, size_t size
, int flags
)
502 struct msghdr msg
= {
504 .msg_iov
= (struct iovec
*)&iov
,
505 .msg_flags
= (flags
? flags
: MSG_WAITALL
| MSG_NOSIGNAL
)
511 rv
= sock_recvmsg(sock
, &msg
, size
, msg
.msg_flags
);
517 static int drbd_recv(struct drbd_tconn
*tconn
, void *buf
, size_t size
)
524 struct msghdr msg
= {
526 .msg_iov
= (struct iovec
*)&iov
,
527 .msg_flags
= MSG_WAITALL
| MSG_NOSIGNAL
535 rv
= sock_recvmsg(tconn
->data
.socket
, &msg
, size
, msg
.msg_flags
);
540 * ECONNRESET other side closed the connection
541 * ERESTARTSYS (on sock) we got a signal
545 if (rv
== -ECONNRESET
)
546 conn_info(tconn
, "sock was reset by peer\n");
547 else if (rv
!= -ERESTARTSYS
)
548 conn_err(tconn
, "sock_recvmsg returned %d\n", rv
);
550 } else if (rv
== 0) {
551 conn_info(tconn
, "sock was shut down by peer\n");
554 /* signal came in, or peer/link went down,
555 * after we read a partial message
557 /* D_ASSERT(signal_pending(current)); */
565 conn_request_state(tconn
, NS(conn
, C_BROKEN_PIPE
), CS_HARD
);
570 static int drbd_recv_all(struct drbd_tconn
*tconn
, void *buf
, size_t size
)
574 err
= drbd_recv(tconn
, buf
, size
);
583 static int drbd_recv_all_warn(struct drbd_tconn
*tconn
, void *buf
, size_t size
)
587 err
= drbd_recv_all(tconn
, buf
, size
);
588 if (err
&& !signal_pending(current
))
589 conn_warn(tconn
, "short read (expected size %d)\n", (int)size
);
594 * On individual connections, the socket buffer size must be set prior to the
595 * listen(2) or connect(2) calls in order to have it take effect.
596 * This is our wrapper to do so.
598 static void drbd_setbufsize(struct socket
*sock
, unsigned int snd
,
601 /* open coded SO_SNDBUF, SO_RCVBUF */
603 sock
->sk
->sk_sndbuf
= snd
;
604 sock
->sk
->sk_userlocks
|= SOCK_SNDBUF_LOCK
;
607 sock
->sk
->sk_rcvbuf
= rcv
;
608 sock
->sk
->sk_userlocks
|= SOCK_RCVBUF_LOCK
;
612 static struct socket
*drbd_try_connect(struct drbd_tconn
*tconn
)
616 struct sockaddr_in6 src_in6
;
617 struct sockaddr_in6 peer_in6
;
619 int err
, peer_addr_len
, my_addr_len
;
620 int sndbuf_size
, rcvbuf_size
, connect_int
;
621 int disconnect_on_error
= 1;
624 nc
= rcu_dereference(tconn
->net_conf
);
629 sndbuf_size
= nc
->sndbuf_size
;
630 rcvbuf_size
= nc
->rcvbuf_size
;
631 connect_int
= nc
->connect_int
;
634 my_addr_len
= min_t(int, tconn
->my_addr_len
, sizeof(src_in6
));
635 memcpy(&src_in6
, &tconn
->my_addr
, my_addr_len
);
637 if (((struct sockaddr
*)&tconn
->my_addr
)->sa_family
== AF_INET6
)
638 src_in6
.sin6_port
= 0;
640 ((struct sockaddr_in
*)&src_in6
)->sin_port
= 0; /* AF_INET & AF_SCI */
642 peer_addr_len
= min_t(int, tconn
->peer_addr_len
, sizeof(src_in6
));
643 memcpy(&peer_in6
, &tconn
->peer_addr
, peer_addr_len
);
645 what
= "sock_create_kern";
646 err
= sock_create_kern(((struct sockaddr
*)&src_in6
)->sa_family
,
647 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
653 sock
->sk
->sk_rcvtimeo
=
654 sock
->sk
->sk_sndtimeo
= connect_int
* HZ
;
655 drbd_setbufsize(sock
, sndbuf_size
, rcvbuf_size
);
657 /* explicitly bind to the configured IP as source IP
658 * for the outgoing connections.
659 * This is needed for multihomed hosts and to be
660 * able to use lo: interfaces for drbd.
661 * Make sure to use 0 as port number, so linux selects
662 * a free one dynamically.
664 what
= "bind before connect";
665 err
= sock
->ops
->bind(sock
, (struct sockaddr
*) &src_in6
, my_addr_len
);
669 /* connect may fail, peer not yet available.
670 * stay C_WF_CONNECTION, don't go Disconnecting! */
671 disconnect_on_error
= 0;
673 err
= sock
->ops
->connect(sock
, (struct sockaddr
*) &peer_in6
, peer_addr_len
, 0);
682 /* timeout, busy, signal pending */
683 case ETIMEDOUT
: case EAGAIN
: case EINPROGRESS
:
684 case EINTR
: case ERESTARTSYS
:
685 /* peer not (yet) available, network problem */
686 case ECONNREFUSED
: case ENETUNREACH
:
687 case EHOSTDOWN
: case EHOSTUNREACH
:
688 disconnect_on_error
= 0;
691 conn_err(tconn
, "%s failed, err = %d\n", what
, err
);
693 if (disconnect_on_error
)
694 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
700 static struct socket
*drbd_wait_for_connect(struct drbd_tconn
*tconn
)
702 int timeo
, err
, my_addr_len
;
703 int sndbuf_size
, rcvbuf_size
, connect_int
;
704 struct socket
*s_estab
= NULL
, *s_listen
;
705 struct sockaddr_in6 my_addr
;
710 nc
= rcu_dereference(tconn
->net_conf
);
715 sndbuf_size
= nc
->sndbuf_size
;
716 rcvbuf_size
= nc
->rcvbuf_size
;
717 connect_int
= nc
->connect_int
;
720 my_addr_len
= min_t(int, tconn
->my_addr_len
, sizeof(struct sockaddr_in6
));
721 memcpy(&my_addr
, &tconn
->my_addr
, my_addr_len
);
723 what
= "sock_create_kern";
724 err
= sock_create_kern(((struct sockaddr
*)&my_addr
)->sa_family
,
725 SOCK_STREAM
, IPPROTO_TCP
, &s_listen
);
731 timeo
= connect_int
* HZ
;
732 timeo
+= (random32() & 1) ? timeo
/ 7 : -timeo
/ 7; /* 28.5% random jitter */
734 s_listen
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
735 s_listen
->sk
->sk_rcvtimeo
= timeo
;
736 s_listen
->sk
->sk_sndtimeo
= timeo
;
737 drbd_setbufsize(s_listen
, sndbuf_size
, rcvbuf_size
);
739 what
= "bind before listen";
740 err
= s_listen
->ops
->bind(s_listen
, (struct sockaddr
*)&my_addr
, my_addr_len
);
744 err
= drbd_accept(&what
, s_listen
, &s_estab
);
748 sock_release(s_listen
);
750 if (err
!= -EAGAIN
&& err
!= -EINTR
&& err
!= -ERESTARTSYS
) {
751 conn_err(tconn
, "%s failed, err = %d\n", what
, err
);
752 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
759 static int decode_header(struct drbd_tconn
*, void *, struct packet_info
*);
761 static int send_first_packet(struct drbd_tconn
*tconn
, struct drbd_socket
*sock
,
762 enum drbd_packet cmd
)
764 if (!conn_prepare_command(tconn
, sock
))
766 return conn_send_command(tconn
, sock
, cmd
, 0, NULL
, 0);
769 static int receive_first_packet(struct drbd_tconn
*tconn
, struct socket
*sock
)
771 unsigned int header_size
= drbd_header_size(tconn
);
772 struct packet_info pi
;
775 err
= drbd_recv_short(sock
, tconn
->data
.rbuf
, header_size
, 0);
776 if (err
!= header_size
) {
781 err
= decode_header(tconn
, tconn
->data
.rbuf
, &pi
);
788 * drbd_socket_okay() - Free the socket if its connection is not okay
789 * @sock: pointer to the pointer to the socket.
791 static int drbd_socket_okay(struct socket
**sock
)
799 rr
= drbd_recv_short(*sock
, tb
, 4, MSG_DONTWAIT
| MSG_PEEK
);
801 if (rr
> 0 || rr
== -EAGAIN
) {
809 /* Gets called if a connection is established, or if a new minor gets created
811 int drbd_connected(struct drbd_conf
*mdev
)
815 atomic_set(&mdev
->packet_seq
, 0);
818 mdev
->state_mutex
= mdev
->tconn
->agreed_pro_version
< 100 ?
819 &mdev
->tconn
->cstate_mutex
:
820 &mdev
->own_state_mutex
;
822 err
= drbd_send_sync_param(mdev
);
824 err
= drbd_send_sizes(mdev
, 0, 0);
826 err
= drbd_send_uuids(mdev
);
828 err
= drbd_send_current_state(mdev
);
829 clear_bit(USE_DEGR_WFC_T
, &mdev
->flags
);
830 clear_bit(RESIZE_PENDING
, &mdev
->flags
);
831 mod_timer(&mdev
->request_timer
, jiffies
+ HZ
); /* just start it here. */
837 * 1 yes, we have a valid connection
838 * 0 oops, did not work out, please try again
839 * -1 peer talks different language,
840 * no point in trying again, please go standalone.
841 * -2 We do not have a network config...
843 static int conn_connect(struct drbd_tconn
*tconn
)
845 struct socket
*sock
, *msock
;
846 struct drbd_conf
*mdev
;
848 int vnr
, timeout
, try, h
, ok
;
849 bool discard_my_data
;
851 if (conn_request_state(tconn
, NS(conn
, C_WF_CONNECTION
), CS_VERBOSE
) < SS_SUCCESS
)
854 clear_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
856 /* Assume that the peer only understands protocol 80 until we know better. */
857 tconn
->agreed_pro_version
= 80;
863 /* 3 tries, this should take less than a second! */
864 s
= drbd_try_connect(tconn
);
867 /* give the other side time to call bind() & listen() */
868 schedule_timeout_interruptible(HZ
/ 10);
872 if (!tconn
->data
.socket
) {
873 tconn
->data
.socket
= s
;
874 send_first_packet(tconn
, &tconn
->data
, P_INITIAL_DATA
);
875 } else if (!tconn
->meta
.socket
) {
876 tconn
->meta
.socket
= s
;
877 send_first_packet(tconn
, &tconn
->meta
, P_INITIAL_META
);
879 conn_err(tconn
, "Logic error in conn_connect()\n");
880 goto out_release_sockets
;
884 if (tconn
->data
.socket
&& tconn
->meta
.socket
) {
885 schedule_timeout_interruptible(tconn
->net_conf
->ping_timeo
*HZ
/10);
886 ok
= drbd_socket_okay(&tconn
->data
.socket
);
887 ok
= drbd_socket_okay(&tconn
->meta
.socket
) && ok
;
893 s
= drbd_wait_for_connect(tconn
);
895 try = receive_first_packet(tconn
, s
);
896 drbd_socket_okay(&tconn
->data
.socket
);
897 drbd_socket_okay(&tconn
->meta
.socket
);
900 if (tconn
->data
.socket
) {
901 conn_warn(tconn
, "initial packet S crossed\n");
902 sock_release(tconn
->data
.socket
);
904 tconn
->data
.socket
= s
;
907 if (tconn
->meta
.socket
) {
908 conn_warn(tconn
, "initial packet M crossed\n");
909 sock_release(tconn
->meta
.socket
);
911 tconn
->meta
.socket
= s
;
912 set_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
915 conn_warn(tconn
, "Error receiving initial packet\n");
922 if (tconn
->cstate
<= C_DISCONNECTING
)
923 goto out_release_sockets
;
924 if (signal_pending(current
)) {
925 flush_signals(current
);
927 if (get_t_state(&tconn
->receiver
) == EXITING
)
928 goto out_release_sockets
;
931 if (tconn
->data
.socket
&& &tconn
->meta
.socket
) {
932 ok
= drbd_socket_okay(&tconn
->data
.socket
);
933 ok
= drbd_socket_okay(&tconn
->meta
.socket
) && ok
;
939 sock
= tconn
->data
.socket
;
940 msock
= tconn
->meta
.socket
;
942 msock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
943 sock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
945 sock
->sk
->sk_allocation
= GFP_NOIO
;
946 msock
->sk
->sk_allocation
= GFP_NOIO
;
948 sock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE_BULK
;
949 msock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE
;
952 * sock->sk->sk_sndtimeo = tconn->net_conf->timeout*HZ/10;
953 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
954 * first set it to the P_CONNECTION_FEATURES timeout,
955 * which we set to 4x the configured ping_timeout. */
957 nc
= rcu_dereference(tconn
->net_conf
);
959 sock
->sk
->sk_sndtimeo
=
960 sock
->sk
->sk_rcvtimeo
= nc
->ping_timeo
*4*HZ
/10;
962 msock
->sk
->sk_rcvtimeo
= nc
->ping_int
*HZ
;
963 timeout
= nc
->timeout
* HZ
/ 10;
964 discard_my_data
= nc
->discard_my_data
;
967 msock
->sk
->sk_sndtimeo
= timeout
;
969 /* we don't want delays.
970 * we use TCP_CORK where appropriate, though */
971 drbd_tcp_nodelay(sock
);
972 drbd_tcp_nodelay(msock
);
974 tconn
->last_received
= jiffies
;
976 h
= drbd_do_features(tconn
);
980 if (tconn
->cram_hmac_tfm
) {
981 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
982 switch (drbd_do_auth(tconn
)) {
984 conn_err(tconn
, "Authentication of peer failed\n");
987 conn_err(tconn
, "Authentication of peer failed, trying again.\n");
992 sock
->sk
->sk_sndtimeo
= timeout
;
993 sock
->sk
->sk_rcvtimeo
= MAX_SCHEDULE_TIMEOUT
;
995 if (drbd_send_protocol(tconn
) == -EOPNOTSUPP
)
999 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
1000 kref_get(&mdev
->kref
);
1003 if (discard_my_data
)
1004 set_bit(DISCARD_MY_DATA
, &mdev
->flags
);
1006 clear_bit(DISCARD_MY_DATA
, &mdev
->flags
);
1008 drbd_connected(mdev
);
1009 kref_put(&mdev
->kref
, &drbd_minor_destroy
);
1014 if (conn_request_state(tconn
, NS(conn
, C_WF_REPORT_PARAMS
), CS_VERBOSE
) < SS_SUCCESS
)
1017 drbd_thread_start(&tconn
->asender
);
1019 mutex_lock(&tconn
->conf_update
);
1020 /* The discard_my_data flag is a single-shot modifier to the next
1021 * connection attempt, the handshake of which is now well underway.
1022 * No need for rcu style copying of the whole struct
1023 * just to clear a single value. */
1024 tconn
->net_conf
->discard_my_data
= 0;
1025 mutex_unlock(&tconn
->conf_update
);
1029 out_release_sockets
:
1030 if (tconn
->data
.socket
) {
1031 sock_release(tconn
->data
.socket
);
1032 tconn
->data
.socket
= NULL
;
1034 if (tconn
->meta
.socket
) {
1035 sock_release(tconn
->meta
.socket
);
1036 tconn
->meta
.socket
= NULL
;
1041 static int decode_header(struct drbd_tconn
*tconn
, void *header
, struct packet_info
*pi
)
1043 unsigned int header_size
= drbd_header_size(tconn
);
1045 if (header_size
== sizeof(struct p_header100
) &&
1046 *(__be32
*)header
== cpu_to_be32(DRBD_MAGIC_100
)) {
1047 struct p_header100
*h
= header
;
1049 conn_err(tconn
, "Header padding is not zero\n");
1052 pi
->vnr
= be16_to_cpu(h
->volume
);
1053 pi
->cmd
= be16_to_cpu(h
->command
);
1054 pi
->size
= be32_to_cpu(h
->length
);
1055 } else if (header_size
== sizeof(struct p_header95
) &&
1056 *(__be16
*)header
== cpu_to_be16(DRBD_MAGIC_BIG
)) {
1057 struct p_header95
*h
= header
;
1058 pi
->cmd
= be16_to_cpu(h
->command
);
1059 pi
->size
= be32_to_cpu(h
->length
);
1061 } else if (header_size
== sizeof(struct p_header80
) &&
1062 *(__be32
*)header
== cpu_to_be32(DRBD_MAGIC
)) {
1063 struct p_header80
*h
= header
;
1064 pi
->cmd
= be16_to_cpu(h
->command
);
1065 pi
->size
= be16_to_cpu(h
->length
);
1068 conn_err(tconn
, "Wrong magic value 0x%08x in protocol version %d\n",
1069 be32_to_cpu(*(__be32
*)header
),
1070 tconn
->agreed_pro_version
);
1073 pi
->data
= header
+ header_size
;
1077 static int drbd_recv_header(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1079 void *buffer
= tconn
->data
.rbuf
;
1082 err
= drbd_recv_all_warn(tconn
, buffer
, drbd_header_size(tconn
));
1086 err
= decode_header(tconn
, buffer
, pi
);
1087 tconn
->last_received
= jiffies
;
1092 static void drbd_flush(struct drbd_tconn
*tconn
)
1095 struct drbd_conf
*mdev
;
1098 if (tconn
->write_ordering
>= WO_bdev_flush
) {
1100 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
1101 if (!get_ldev(mdev
))
1103 kref_get(&mdev
->kref
);
1106 rv
= blkdev_issue_flush(mdev
->ldev
->backing_bdev
,
1109 dev_info(DEV
, "local disk flush failed with status %d\n", rv
);
1110 /* would rather check on EOPNOTSUPP, but that is not reliable.
1111 * don't try again for ANY return value != 0
1112 * if (rv == -EOPNOTSUPP) */
1113 drbd_bump_write_ordering(tconn
, WO_drain_io
);
1116 kref_put(&mdev
->kref
, &drbd_minor_destroy
);
1127 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1128 * @mdev: DRBD device.
1129 * @epoch: Epoch object.
1132 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_tconn
*tconn
,
1133 struct drbd_epoch
*epoch
,
1134 enum epoch_event ev
)
1137 struct drbd_epoch
*next_epoch
;
1138 enum finish_epoch rv
= FE_STILL_LIVE
;
1140 spin_lock(&tconn
->epoch_lock
);
1144 epoch_size
= atomic_read(&epoch
->epoch_size
);
1146 switch (ev
& ~EV_CLEANUP
) {
1148 atomic_dec(&epoch
->active
);
1150 case EV_GOT_BARRIER_NR
:
1151 set_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
);
1153 case EV_BECAME_LAST
:
1158 if (epoch_size
!= 0 &&
1159 atomic_read(&epoch
->active
) == 0 &&
1160 (test_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
) || ev
& EV_CLEANUP
)) {
1161 if (!(ev
& EV_CLEANUP
)) {
1162 spin_unlock(&tconn
->epoch_lock
);
1163 drbd_send_b_ack(epoch
->mdev
, epoch
->barrier_nr
, epoch_size
);
1164 spin_lock(&tconn
->epoch_lock
);
1166 if (test_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
))
1167 dec_unacked(epoch
->mdev
);
1169 if (tconn
->current_epoch
!= epoch
) {
1170 next_epoch
= list_entry(epoch
->list
.next
, struct drbd_epoch
, list
);
1171 list_del(&epoch
->list
);
1172 ev
= EV_BECAME_LAST
| (ev
& EV_CLEANUP
);
1176 if (rv
== FE_STILL_LIVE
)
1180 atomic_set(&epoch
->epoch_size
, 0);
1181 /* atomic_set(&epoch->active, 0); is already zero */
1182 if (rv
== FE_STILL_LIVE
)
1193 spin_unlock(&tconn
->epoch_lock
);
1199 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1200 * @tconn: DRBD connection.
1201 * @wo: Write ordering method to try.
1203 void drbd_bump_write_ordering(struct drbd_tconn
*tconn
, enum write_ordering_e wo
)
1205 struct disk_conf
*dc
;
1206 struct drbd_conf
*mdev
;
1207 enum write_ordering_e pwo
;
1209 static char *write_ordering_str
[] = {
1211 [WO_drain_io
] = "drain",
1212 [WO_bdev_flush
] = "flush",
1215 pwo
= tconn
->write_ordering
;
1218 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
1219 if (!get_ldev(mdev
))
1221 dc
= rcu_dereference(mdev
->ldev
->disk_conf
);
1223 if (wo
== WO_bdev_flush
&& !dc
->disk_flushes
)
1225 if (wo
== WO_drain_io
&& !dc
->disk_drain
)
1230 tconn
->write_ordering
= wo
;
1231 if (pwo
!= tconn
->write_ordering
|| wo
== WO_bdev_flush
)
1232 conn_info(tconn
, "Method to ensure write ordering: %s\n", write_ordering_str
[tconn
->write_ordering
]);
1236 * drbd_submit_peer_request()
1237 * @mdev: DRBD device.
1238 * @peer_req: peer request
1239 * @rw: flag field, see bio->bi_rw
1241 * May spread the pages to multiple bios,
1242 * depending on bio_add_page restrictions.
1244 * Returns 0 if all bios have been submitted,
1245 * -ENOMEM if we could not allocate enough bios,
1246 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1247 * single page to an empty bio (which should never happen and likely indicates
1248 * that the lower level IO stack is in some way broken). This has been observed
1249 * on certain Xen deployments.
1251 /* TODO allocate from our own bio_set. */
1252 int drbd_submit_peer_request(struct drbd_conf
*mdev
,
1253 struct drbd_peer_request
*peer_req
,
1254 const unsigned rw
, const int fault_type
)
1256 struct bio
*bios
= NULL
;
1258 struct page
*page
= peer_req
->pages
;
1259 sector_t sector
= peer_req
->i
.sector
;
1260 unsigned ds
= peer_req
->i
.size
;
1261 unsigned n_bios
= 0;
1262 unsigned nr_pages
= (ds
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
1265 /* In most cases, we will only need one bio. But in case the lower
1266 * level restrictions happen to be different at this offset on this
1267 * side than those of the sending peer, we may need to submit the
1268 * request in more than one bio.
1270 * Plain bio_alloc is good enough here, this is no DRBD internally
1271 * generated bio, but a bio allocated on behalf of the peer.
1274 bio
= bio_alloc(GFP_NOIO
, nr_pages
);
1276 dev_err(DEV
, "submit_ee: Allocation of a bio failed\n");
1279 /* > peer_req->i.sector, unless this is the first bio */
1280 bio
->bi_sector
= sector
;
1281 bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1283 bio
->bi_private
= peer_req
;
1284 bio
->bi_end_io
= drbd_peer_request_endio
;
1286 bio
->bi_next
= bios
;
1290 page_chain_for_each(page
) {
1291 unsigned len
= min_t(unsigned, ds
, PAGE_SIZE
);
1292 if (!bio_add_page(bio
, page
, len
, 0)) {
1293 /* A single page must always be possible!
1294 * But in case it fails anyways,
1295 * we deal with it, and complain (below). */
1296 if (bio
->bi_vcnt
== 0) {
1298 "bio_add_page failed for len=%u, "
1299 "bi_vcnt=0 (bi_sector=%llu)\n",
1300 len
, (unsigned long long)bio
->bi_sector
);
1310 D_ASSERT(page
== NULL
);
1313 atomic_set(&peer_req
->pending_bios
, n_bios
);
1316 bios
= bios
->bi_next
;
1317 bio
->bi_next
= NULL
;
1319 drbd_generic_make_request(mdev
, fault_type
, bio
);
1326 bios
= bios
->bi_next
;
1332 static void drbd_remove_epoch_entry_interval(struct drbd_conf
*mdev
,
1333 struct drbd_peer_request
*peer_req
)
1335 struct drbd_interval
*i
= &peer_req
->i
;
1337 drbd_remove_interval(&mdev
->write_requests
, i
);
1338 drbd_clear_interval(i
);
1340 /* Wake up any processes waiting for this peer request to complete. */
1342 wake_up(&mdev
->misc_wait
);
1345 void conn_wait_active_ee_empty(struct drbd_tconn
*tconn
)
1347 struct drbd_conf
*mdev
;
1351 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
1352 kref_get(&mdev
->kref
);
1354 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1355 kref_put(&mdev
->kref
, &drbd_minor_destroy
);
1361 static int receive_Barrier(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1363 struct drbd_conf
*mdev
;
1365 struct p_barrier
*p
= pi
->data
;
1366 struct drbd_epoch
*epoch
;
1368 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1374 tconn
->current_epoch
->barrier_nr
= p
->barrier
;
1375 tconn
->current_epoch
->mdev
= mdev
;
1376 rv
= drbd_may_finish_epoch(tconn
, tconn
->current_epoch
, EV_GOT_BARRIER_NR
);
1378 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1379 * the activity log, which means it would not be resynced in case the
1380 * R_PRIMARY crashes now.
1381 * Therefore we must send the barrier_ack after the barrier request was
1383 switch (tconn
->write_ordering
) {
1385 if (rv
== FE_RECYCLED
)
1388 /* receiver context, in the writeout path of the other node.
1389 * avoid potential distributed deadlock */
1390 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1394 dev_warn(DEV
, "Allocation of an epoch failed, slowing down\n");
1399 conn_wait_active_ee_empty(tconn
);
1402 if (atomic_read(&tconn
->current_epoch
->epoch_size
)) {
1403 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1408 epoch
= tconn
->current_epoch
;
1409 wait_event(mdev
->ee_wait
, atomic_read(&epoch
->epoch_size
) == 0);
1411 D_ASSERT(atomic_read(&epoch
->active
) == 0);
1412 D_ASSERT(epoch
->flags
== 0);
1416 dev_err(DEV
, "Strangeness in tconn->write_ordering %d\n", tconn
->write_ordering
);
1421 atomic_set(&epoch
->epoch_size
, 0);
1422 atomic_set(&epoch
->active
, 0);
1424 spin_lock(&tconn
->epoch_lock
);
1425 if (atomic_read(&tconn
->current_epoch
->epoch_size
)) {
1426 list_add(&epoch
->list
, &tconn
->current_epoch
->list
);
1427 tconn
->current_epoch
= epoch
;
1430 /* The current_epoch got recycled while we allocated this one... */
1433 spin_unlock(&tconn
->epoch_lock
);
1438 /* used from receive_RSDataReply (recv_resync_read)
1439 * and from receive_Data */
1440 static struct drbd_peer_request
*
1441 read_in_block(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
1442 int data_size
) __must_hold(local
)
1444 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1445 struct drbd_peer_request
*peer_req
;
1448 void *dig_in
= mdev
->tconn
->int_dig_in
;
1449 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1450 unsigned long *data
;
1453 if (mdev
->tconn
->peer_integrity_tfm
) {
1454 dgs
= crypto_hash_digestsize(mdev
->tconn
->peer_integrity_tfm
);
1456 * FIXME: Receive the incoming digest into the receive buffer
1457 * here, together with its struct p_data?
1459 err
= drbd_recv_all_warn(mdev
->tconn
, dig_in
, dgs
);
1465 if (!expect(data_size
!= 0))
1467 if (!expect(IS_ALIGNED(data_size
, 512)))
1469 if (!expect(data_size
<= DRBD_MAX_BIO_SIZE
))
1472 /* even though we trust out peer,
1473 * we sometimes have to double check. */
1474 if (sector
+ (data_size
>>9) > capacity
) {
1475 dev_err(DEV
, "request from peer beyond end of local disk: "
1476 "capacity: %llus < sector: %llus + size: %u\n",
1477 (unsigned long long)capacity
,
1478 (unsigned long long)sector
, data_size
);
1482 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1483 * "criss-cross" setup, that might cause write-out on some other DRBD,
1484 * which in turn might block on the other node at this very place. */
1485 peer_req
= drbd_alloc_peer_req(mdev
, id
, sector
, data_size
, GFP_NOIO
);
1490 page
= peer_req
->pages
;
1491 page_chain_for_each(page
) {
1492 unsigned len
= min_t(int, ds
, PAGE_SIZE
);
1494 err
= drbd_recv_all_warn(mdev
->tconn
, data
, len
);
1495 if (drbd_insert_fault(mdev
, DRBD_FAULT_RECEIVE
)) {
1496 dev_err(DEV
, "Fault injection: Corrupting data on receive\n");
1497 data
[0] = data
[0] ^ (unsigned long)-1;
1501 drbd_free_peer_req(mdev
, peer_req
);
1508 drbd_csum_ee(mdev
, mdev
->tconn
->peer_integrity_tfm
, peer_req
, dig_vv
);
1509 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1510 dev_err(DEV
, "Digest integrity check FAILED: %llus +%u\n",
1511 (unsigned long long)sector
, data_size
);
1512 drbd_free_peer_req(mdev
, peer_req
);
1516 mdev
->recv_cnt
+= data_size
>>9;
1520 /* drbd_drain_block() just takes a data block
1521 * out of the socket input buffer, and discards it.
1523 static int drbd_drain_block(struct drbd_conf
*mdev
, int data_size
)
1532 page
= drbd_alloc_pages(mdev
, 1, 1);
1536 unsigned int len
= min_t(int, data_size
, PAGE_SIZE
);
1538 err
= drbd_recv_all_warn(mdev
->tconn
, data
, len
);
1544 drbd_free_pages(mdev
, page
, 0);
1548 static int recv_dless_read(struct drbd_conf
*mdev
, struct drbd_request
*req
,
1549 sector_t sector
, int data_size
)
1551 struct bio_vec
*bvec
;
1553 int dgs
, err
, i
, expect
;
1554 void *dig_in
= mdev
->tconn
->int_dig_in
;
1555 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1558 if (mdev
->tconn
->peer_integrity_tfm
) {
1559 dgs
= crypto_hash_digestsize(mdev
->tconn
->peer_integrity_tfm
);
1560 err
= drbd_recv_all_warn(mdev
->tconn
, dig_in
, dgs
);
1566 /* optimistically update recv_cnt. if receiving fails below,
1567 * we disconnect anyways, and counters will be reset. */
1568 mdev
->recv_cnt
+= data_size
>>9;
1570 bio
= req
->master_bio
;
1571 D_ASSERT(sector
== bio
->bi_sector
);
1573 bio_for_each_segment(bvec
, bio
, i
) {
1574 void *mapped
= kmap(bvec
->bv_page
) + bvec
->bv_offset
;
1575 expect
= min_t(int, data_size
, bvec
->bv_len
);
1576 err
= drbd_recv_all_warn(mdev
->tconn
, mapped
, expect
);
1577 kunmap(bvec
->bv_page
);
1580 data_size
-= expect
;
1584 drbd_csum_bio(mdev
, mdev
->tconn
->peer_integrity_tfm
, bio
, dig_vv
);
1585 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1586 dev_err(DEV
, "Digest integrity check FAILED. Broken NICs?\n");
1591 D_ASSERT(data_size
== 0);
1596 * e_end_resync_block() is called in asender context via
1597 * drbd_finish_peer_reqs().
1599 static int e_end_resync_block(struct drbd_work
*w
, int unused
)
1601 struct drbd_peer_request
*peer_req
=
1602 container_of(w
, struct drbd_peer_request
, w
);
1603 struct drbd_conf
*mdev
= w
->mdev
;
1604 sector_t sector
= peer_req
->i
.sector
;
1607 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1609 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1610 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1611 err
= drbd_send_ack(mdev
, P_RS_WRITE_ACK
, peer_req
);
1613 /* Record failure to sync */
1614 drbd_rs_failed_io(mdev
, sector
, peer_req
->i
.size
);
1616 err
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1623 static int recv_resync_read(struct drbd_conf
*mdev
, sector_t sector
, int data_size
) __releases(local
)
1625 struct drbd_peer_request
*peer_req
;
1627 peer_req
= read_in_block(mdev
, ID_SYNCER
, sector
, data_size
);
1631 dec_rs_pending(mdev
);
1634 /* corresponding dec_unacked() in e_end_resync_block()
1635 * respective _drbd_clear_done_ee */
1637 peer_req
->w
.cb
= e_end_resync_block
;
1639 spin_lock_irq(&mdev
->tconn
->req_lock
);
1640 list_add(&peer_req
->w
.list
, &mdev
->sync_ee
);
1641 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1643 atomic_add(data_size
>> 9, &mdev
->rs_sect_ev
);
1644 if (drbd_submit_peer_request(mdev
, peer_req
, WRITE
, DRBD_FAULT_RS_WR
) == 0)
1647 /* don't care for the reason here */
1648 dev_err(DEV
, "submit failed, triggering re-connect\n");
1649 spin_lock_irq(&mdev
->tconn
->req_lock
);
1650 list_del(&peer_req
->w
.list
);
1651 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1653 drbd_free_peer_req(mdev
, peer_req
);
1659 static struct drbd_request
*
1660 find_request(struct drbd_conf
*mdev
, struct rb_root
*root
, u64 id
,
1661 sector_t sector
, bool missing_ok
, const char *func
)
1663 struct drbd_request
*req
;
1665 /* Request object according to our peer */
1666 req
= (struct drbd_request
*)(unsigned long)id
;
1667 if (drbd_contains_interval(root
, sector
, &req
->i
) && req
->i
.local
)
1670 dev_err(DEV
, "%s: failed to find request 0x%lx, sector %llus\n", func
,
1671 (unsigned long)id
, (unsigned long long)sector
);
1676 static int receive_DataReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1678 struct drbd_conf
*mdev
;
1679 struct drbd_request
*req
;
1682 struct p_data
*p
= pi
->data
;
1684 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1688 sector
= be64_to_cpu(p
->sector
);
1690 spin_lock_irq(&mdev
->tconn
->req_lock
);
1691 req
= find_request(mdev
, &mdev
->read_requests
, p
->block_id
, sector
, false, __func__
);
1692 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1696 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1697 * special casing it there for the various failure cases.
1698 * still no race with drbd_fail_pending_reads */
1699 err
= recv_dless_read(mdev
, req
, sector
, pi
->size
);
1701 req_mod(req
, DATA_RECEIVED
);
1702 /* else: nothing. handled from drbd_disconnect...
1703 * I don't think we may complete this just yet
1704 * in case we are "on-disconnect: freeze" */
1709 static int receive_RSDataReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1711 struct drbd_conf
*mdev
;
1714 struct p_data
*p
= pi
->data
;
1716 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1720 sector
= be64_to_cpu(p
->sector
);
1721 D_ASSERT(p
->block_id
== ID_SYNCER
);
1723 if (get_ldev(mdev
)) {
1724 /* data is submitted to disk within recv_resync_read.
1725 * corresponding put_ldev done below on error,
1726 * or in drbd_peer_request_endio. */
1727 err
= recv_resync_read(mdev
, sector
, pi
->size
);
1729 if (__ratelimit(&drbd_ratelimit_state
))
1730 dev_err(DEV
, "Can not write resync data to local disk.\n");
1732 err
= drbd_drain_block(mdev
, pi
->size
);
1734 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, pi
->size
);
1737 atomic_add(pi
->size
>> 9, &mdev
->rs_sect_in
);
1742 static int w_restart_write(struct drbd_work
*w
, int cancel
)
1744 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1745 struct drbd_conf
*mdev
= w
->mdev
;
1747 unsigned long start_time
;
1748 unsigned long flags
;
1750 spin_lock_irqsave(&mdev
->tconn
->req_lock
, flags
);
1751 if (!expect(req
->rq_state
& RQ_POSTPONED
)) {
1752 spin_unlock_irqrestore(&mdev
->tconn
->req_lock
, flags
);
1755 bio
= req
->master_bio
;
1756 start_time
= req
->start_time
;
1757 /* Postponed requests will not have their master_bio completed! */
1758 __req_mod(req
, DISCARD_WRITE
, NULL
);
1759 spin_unlock_irqrestore(&mdev
->tconn
->req_lock
, flags
);
1761 while (__drbd_make_request(mdev
, bio
, start_time
))
1766 static void restart_conflicting_writes(struct drbd_conf
*mdev
,
1767 sector_t sector
, int size
)
1769 struct drbd_interval
*i
;
1770 struct drbd_request
*req
;
1772 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1775 req
= container_of(i
, struct drbd_request
, i
);
1776 if (req
->rq_state
& RQ_LOCAL_PENDING
||
1777 !(req
->rq_state
& RQ_POSTPONED
))
1779 if (expect(list_empty(&req
->w
.list
))) {
1781 req
->w
.cb
= w_restart_write
;
1782 drbd_queue_work(&mdev
->tconn
->data
.work
, &req
->w
);
1788 * e_end_block() is called in asender context via drbd_finish_peer_reqs().
1790 static int e_end_block(struct drbd_work
*w
, int cancel
)
1792 struct drbd_peer_request
*peer_req
=
1793 container_of(w
, struct drbd_peer_request
, w
);
1794 struct drbd_conf
*mdev
= w
->mdev
;
1795 sector_t sector
= peer_req
->i
.sector
;
1798 if (peer_req
->flags
& EE_SEND_WRITE_ACK
) {
1799 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1800 pcmd
= (mdev
->state
.conn
>= C_SYNC_SOURCE
&&
1801 mdev
->state
.conn
<= C_PAUSED_SYNC_T
&&
1802 peer_req
->flags
& EE_MAY_SET_IN_SYNC
) ?
1803 P_RS_WRITE_ACK
: P_WRITE_ACK
;
1804 err
= drbd_send_ack(mdev
, pcmd
, peer_req
);
1805 if (pcmd
== P_RS_WRITE_ACK
)
1806 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1808 err
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1809 /* we expect it to be marked out of sync anyways...
1810 * maybe assert this? */
1814 /* we delete from the conflict detection hash _after_ we sent out the
1815 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1816 if (peer_req
->flags
& EE_IN_INTERVAL_TREE
) {
1817 spin_lock_irq(&mdev
->tconn
->req_lock
);
1818 D_ASSERT(!drbd_interval_empty(&peer_req
->i
));
1819 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
1820 if (peer_req
->flags
& EE_RESTART_REQUESTS
)
1821 restart_conflicting_writes(mdev
, sector
, peer_req
->i
.size
);
1822 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1824 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1826 drbd_may_finish_epoch(mdev
->tconn
, peer_req
->epoch
, EV_PUT
+ (cancel
? EV_CLEANUP
: 0));
1831 static int e_send_ack(struct drbd_work
*w
, enum drbd_packet ack
)
1833 struct drbd_conf
*mdev
= w
->mdev
;
1834 struct drbd_peer_request
*peer_req
=
1835 container_of(w
, struct drbd_peer_request
, w
);
1838 err
= drbd_send_ack(mdev
, ack
, peer_req
);
1844 static int e_send_discard_write(struct drbd_work
*w
, int unused
)
1846 return e_send_ack(w
, P_DISCARD_WRITE
);
1849 static int e_send_retry_write(struct drbd_work
*w
, int unused
)
1851 struct drbd_tconn
*tconn
= w
->mdev
->tconn
;
1853 return e_send_ack(w
, tconn
->agreed_pro_version
>= 100 ?
1854 P_RETRY_WRITE
: P_DISCARD_WRITE
);
1857 static bool seq_greater(u32 a
, u32 b
)
1860 * We assume 32-bit wrap-around here.
1861 * For 24-bit wrap-around, we would have to shift:
1864 return (s32
)a
- (s32
)b
> 0;
1867 static u32
seq_max(u32 a
, u32 b
)
1869 return seq_greater(a
, b
) ? a
: b
;
1872 static bool need_peer_seq(struct drbd_conf
*mdev
)
1874 struct drbd_tconn
*tconn
= mdev
->tconn
;
1878 * We only need to keep track of the last packet_seq number of our peer
1879 * if we are in dual-primary mode and we have the discard flag set; see
1880 * handle_write_conflicts().
1884 tp
= rcu_dereference(mdev
->tconn
->net_conf
)->two_primaries
;
1887 return tp
&& test_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
1890 static void update_peer_seq(struct drbd_conf
*mdev
, unsigned int peer_seq
)
1892 unsigned int newest_peer_seq
;
1894 if (need_peer_seq(mdev
)) {
1895 spin_lock(&mdev
->peer_seq_lock
);
1896 newest_peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1897 mdev
->peer_seq
= newest_peer_seq
;
1898 spin_unlock(&mdev
->peer_seq_lock
);
1899 /* wake up only if we actually changed mdev->peer_seq */
1900 if (peer_seq
== newest_peer_seq
)
1901 wake_up(&mdev
->seq_wait
);
1905 /* Called from receive_Data.
1906 * Synchronize packets on sock with packets on msock.
1908 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1909 * packet traveling on msock, they are still processed in the order they have
1912 * Note: we don't care for Ack packets overtaking P_DATA packets.
1914 * In case packet_seq is larger than mdev->peer_seq number, there are
1915 * outstanding packets on the msock. We wait for them to arrive.
1916 * In case we are the logically next packet, we update mdev->peer_seq
1917 * ourselves. Correctly handles 32bit wrap around.
1919 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1920 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1921 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1922 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1924 * returns 0 if we may process the packet,
1925 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1926 static int wait_for_and_update_peer_seq(struct drbd_conf
*mdev
, const u32 peer_seq
)
1932 if (!need_peer_seq(mdev
))
1935 spin_lock(&mdev
->peer_seq_lock
);
1937 if (!seq_greater(peer_seq
- 1, mdev
->peer_seq
)) {
1938 mdev
->peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1942 if (signal_pending(current
)) {
1946 prepare_to_wait(&mdev
->seq_wait
, &wait
, TASK_INTERRUPTIBLE
);
1947 spin_unlock(&mdev
->peer_seq_lock
);
1949 timeout
= rcu_dereference(mdev
->tconn
->net_conf
)->ping_timeo
*HZ
/10;
1951 timeout
= schedule_timeout(timeout
);
1952 spin_lock(&mdev
->peer_seq_lock
);
1955 dev_err(DEV
, "Timed out waiting for missing ack packets; disconnecting\n");
1959 spin_unlock(&mdev
->peer_seq_lock
);
1960 finish_wait(&mdev
->seq_wait
, &wait
);
1964 /* see also bio_flags_to_wire()
1965 * DRBD_REQ_*, because we need to semantically map the flags to data packet
1966 * flags and back. We may replicate to other kernel versions. */
1967 static unsigned long wire_flags_to_bio(struct drbd_conf
*mdev
, u32 dpf
)
1969 return (dpf
& DP_RW_SYNC
? REQ_SYNC
: 0) |
1970 (dpf
& DP_FUA
? REQ_FUA
: 0) |
1971 (dpf
& DP_FLUSH
? REQ_FLUSH
: 0) |
1972 (dpf
& DP_DISCARD
? REQ_DISCARD
: 0);
1975 static void fail_postponed_requests(struct drbd_conf
*mdev
, sector_t sector
,
1978 struct drbd_interval
*i
;
1981 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1982 struct drbd_request
*req
;
1983 struct bio_and_error m
;
1987 req
= container_of(i
, struct drbd_request
, i
);
1988 if (!(req
->rq_state
& RQ_POSTPONED
))
1990 req
->rq_state
&= ~RQ_POSTPONED
;
1991 __req_mod(req
, NEG_ACKED
, &m
);
1992 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1994 complete_master_bio(mdev
, &m
);
1995 spin_lock_irq(&mdev
->tconn
->req_lock
);
2000 static int handle_write_conflicts(struct drbd_conf
*mdev
,
2001 struct drbd_peer_request
*peer_req
)
2003 struct drbd_tconn
*tconn
= mdev
->tconn
;
2004 bool resolve_conflicts
= test_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
2005 sector_t sector
= peer_req
->i
.sector
;
2006 const unsigned int size
= peer_req
->i
.size
;
2007 struct drbd_interval
*i
;
2012 * Inserting the peer request into the write_requests tree will prevent
2013 * new conflicting local requests from being added.
2015 drbd_insert_interval(&mdev
->write_requests
, &peer_req
->i
);
2018 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
2019 if (i
== &peer_req
->i
)
2024 * Our peer has sent a conflicting remote request; this
2025 * should not happen in a two-node setup. Wait for the
2026 * earlier peer request to complete.
2028 err
= drbd_wait_misc(mdev
, i
);
2034 equal
= i
->sector
== sector
&& i
->size
== size
;
2035 if (resolve_conflicts
) {
2037 * If the peer request is fully contained within the
2038 * overlapping request, it can be discarded; otherwise,
2039 * it will be retried once all overlapping requests
2042 bool discard
= i
->sector
<= sector
&& i
->sector
+
2043 (i
->size
>> 9) >= sector
+ (size
>> 9);
2046 dev_alert(DEV
, "Concurrent writes detected: "
2047 "local=%llus +%u, remote=%llus +%u, "
2048 "assuming %s came first\n",
2049 (unsigned long long)i
->sector
, i
->size
,
2050 (unsigned long long)sector
, size
,
2051 discard
? "local" : "remote");
2054 peer_req
->w
.cb
= discard
? e_send_discard_write
:
2056 list_add_tail(&peer_req
->w
.list
, &mdev
->done_ee
);
2057 wake_asender(mdev
->tconn
);
2062 struct drbd_request
*req
=
2063 container_of(i
, struct drbd_request
, i
);
2066 dev_alert(DEV
, "Concurrent writes detected: "
2067 "local=%llus +%u, remote=%llus +%u\n",
2068 (unsigned long long)i
->sector
, i
->size
,
2069 (unsigned long long)sector
, size
);
2071 if (req
->rq_state
& RQ_LOCAL_PENDING
||
2072 !(req
->rq_state
& RQ_POSTPONED
)) {
2074 * Wait for the node with the discard flag to
2075 * decide if this request will be discarded or
2076 * retried. Requests that are discarded will
2077 * disappear from the write_requests tree.
2079 * In addition, wait for the conflicting
2080 * request to finish locally before submitting
2081 * the conflicting peer request.
2083 err
= drbd_wait_misc(mdev
, &req
->i
);
2085 _conn_request_state(mdev
->tconn
,
2086 NS(conn
, C_TIMEOUT
),
2088 fail_postponed_requests(mdev
, sector
, size
);
2094 * Remember to restart the conflicting requests after
2095 * the new peer request has completed.
2097 peer_req
->flags
|= EE_RESTART_REQUESTS
;
2104 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
2108 /* mirrored write */
2109 static int receive_Data(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2111 struct drbd_conf
*mdev
;
2113 struct drbd_peer_request
*peer_req
;
2114 struct p_data
*p
= pi
->data
;
2115 u32 peer_seq
= be32_to_cpu(p
->seq_num
);
2120 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
2124 if (!get_ldev(mdev
)) {
2127 err
= wait_for_and_update_peer_seq(mdev
, peer_seq
);
2128 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, pi
->size
);
2129 atomic_inc(&tconn
->current_epoch
->epoch_size
);
2130 err2
= drbd_drain_block(mdev
, pi
->size
);
2137 * Corresponding put_ldev done either below (on various errors), or in
2138 * drbd_peer_request_endio, if we successfully submit the data at the
2139 * end of this function.
2142 sector
= be64_to_cpu(p
->sector
);
2143 peer_req
= read_in_block(mdev
, p
->block_id
, sector
, pi
->size
);
2149 peer_req
->w
.cb
= e_end_block
;
2151 dp_flags
= be32_to_cpu(p
->dp_flags
);
2152 rw
|= wire_flags_to_bio(mdev
, dp_flags
);
2154 if (dp_flags
& DP_MAY_SET_IN_SYNC
)
2155 peer_req
->flags
|= EE_MAY_SET_IN_SYNC
;
2157 spin_lock(&tconn
->epoch_lock
);
2158 peer_req
->epoch
= tconn
->current_epoch
;
2159 atomic_inc(&peer_req
->epoch
->epoch_size
);
2160 atomic_inc(&peer_req
->epoch
->active
);
2161 spin_unlock(&tconn
->epoch_lock
);
2164 tp
= rcu_dereference(mdev
->tconn
->net_conf
)->two_primaries
;
2167 peer_req
->flags
|= EE_IN_INTERVAL_TREE
;
2168 err
= wait_for_and_update_peer_seq(mdev
, peer_seq
);
2170 goto out_interrupted
;
2171 spin_lock_irq(&mdev
->tconn
->req_lock
);
2172 err
= handle_write_conflicts(mdev
, peer_req
);
2174 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2175 if (err
== -ENOENT
) {
2179 goto out_interrupted
;
2182 spin_lock_irq(&mdev
->tconn
->req_lock
);
2183 list_add(&peer_req
->w
.list
, &mdev
->active_ee
);
2184 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2186 if (mdev
->tconn
->agreed_pro_version
< 100) {
2188 switch (rcu_dereference(mdev
->tconn
->net_conf
)->wire_protocol
) {
2190 dp_flags
|= DP_SEND_WRITE_ACK
;
2193 dp_flags
|= DP_SEND_RECEIVE_ACK
;
2199 if (dp_flags
& DP_SEND_WRITE_ACK
) {
2200 peer_req
->flags
|= EE_SEND_WRITE_ACK
;
2202 /* corresponding dec_unacked() in e_end_block()
2203 * respective _drbd_clear_done_ee */
2206 if (dp_flags
& DP_SEND_RECEIVE_ACK
) {
2207 /* I really don't like it that the receiver thread
2208 * sends on the msock, but anyways */
2209 drbd_send_ack(mdev
, P_RECV_ACK
, peer_req
);
2212 if (mdev
->state
.pdsk
< D_INCONSISTENT
) {
2213 /* In case we have the only disk of the cluster, */
2214 drbd_set_out_of_sync(mdev
, peer_req
->i
.sector
, peer_req
->i
.size
);
2215 peer_req
->flags
|= EE_CALL_AL_COMPLETE_IO
;
2216 peer_req
->flags
&= ~EE_MAY_SET_IN_SYNC
;
2217 drbd_al_begin_io(mdev
, &peer_req
->i
);
2220 err
= drbd_submit_peer_request(mdev
, peer_req
, rw
, DRBD_FAULT_DT_WR
);
2224 /* don't care for the reason here */
2225 dev_err(DEV
, "submit failed, triggering re-connect\n");
2226 spin_lock_irq(&mdev
->tconn
->req_lock
);
2227 list_del(&peer_req
->w
.list
);
2228 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
2229 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2230 if (peer_req
->flags
& EE_CALL_AL_COMPLETE_IO
)
2231 drbd_al_complete_io(mdev
, &peer_req
->i
);
2234 drbd_may_finish_epoch(tconn
, peer_req
->epoch
, EV_PUT
+ EV_CLEANUP
);
2236 drbd_free_peer_req(mdev
, peer_req
);
2240 /* We may throttle resync, if the lower device seems to be busy,
2241 * and current sync rate is above c_min_rate.
2243 * To decide whether or not the lower device is busy, we use a scheme similar
2244 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2245 * (more than 64 sectors) of activity we cannot account for with our own resync
2246 * activity, it obviously is "busy".
2248 * The current sync rate used here uses only the most recent two step marks,
2249 * to have a short time average so we can react faster.
2251 int drbd_rs_should_slow_down(struct drbd_conf
*mdev
, sector_t sector
)
2253 struct gendisk
*disk
= mdev
->ldev
->backing_bdev
->bd_contains
->bd_disk
;
2254 unsigned long db
, dt
, dbdt
;
2255 struct lc_element
*tmp
;
2258 unsigned int c_min_rate
;
2261 c_min_rate
= rcu_dereference(mdev
->ldev
->disk_conf
)->c_min_rate
;
2264 /* feature disabled? */
2265 if (c_min_rate
== 0)
2268 spin_lock_irq(&mdev
->al_lock
);
2269 tmp
= lc_find(mdev
->resync
, BM_SECT_TO_EXT(sector
));
2271 struct bm_extent
*bm_ext
= lc_entry(tmp
, struct bm_extent
, lce
);
2272 if (test_bit(BME_PRIORITY
, &bm_ext
->flags
)) {
2273 spin_unlock_irq(&mdev
->al_lock
);
2276 /* Do not slow down if app IO is already waiting for this extent */
2278 spin_unlock_irq(&mdev
->al_lock
);
2280 curr_events
= (int)part_stat_read(&disk
->part0
, sectors
[0]) +
2281 (int)part_stat_read(&disk
->part0
, sectors
[1]) -
2282 atomic_read(&mdev
->rs_sect_ev
);
2284 if (!mdev
->rs_last_events
|| curr_events
- mdev
->rs_last_events
> 64) {
2285 unsigned long rs_left
;
2288 mdev
->rs_last_events
= curr_events
;
2290 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2292 i
= (mdev
->rs_last_mark
+ DRBD_SYNC_MARKS
-1) % DRBD_SYNC_MARKS
;
2294 if (mdev
->state
.conn
== C_VERIFY_S
|| mdev
->state
.conn
== C_VERIFY_T
)
2295 rs_left
= mdev
->ov_left
;
2297 rs_left
= drbd_bm_total_weight(mdev
) - mdev
->rs_failed
;
2299 dt
= ((long)jiffies
- (long)mdev
->rs_mark_time
[i
]) / HZ
;
2302 db
= mdev
->rs_mark_left
[i
] - rs_left
;
2303 dbdt
= Bit2KB(db
/dt
);
2305 if (dbdt
> c_min_rate
)
2312 static int receive_DataRequest(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2314 struct drbd_conf
*mdev
;
2317 struct drbd_peer_request
*peer_req
;
2318 struct digest_info
*di
= NULL
;
2320 unsigned int fault_type
;
2321 struct p_block_req
*p
= pi
->data
;
2323 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
2326 capacity
= drbd_get_capacity(mdev
->this_bdev
);
2328 sector
= be64_to_cpu(p
->sector
);
2329 size
= be32_to_cpu(p
->blksize
);
2331 if (size
<= 0 || !IS_ALIGNED(size
, 512) || size
> DRBD_MAX_BIO_SIZE
) {
2332 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2333 (unsigned long long)sector
, size
);
2336 if (sector
+ (size
>>9) > capacity
) {
2337 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2338 (unsigned long long)sector
, size
);
2342 if (!get_ldev_if_state(mdev
, D_UP_TO_DATE
)) {
2345 case P_DATA_REQUEST
:
2346 drbd_send_ack_rp(mdev
, P_NEG_DREPLY
, p
);
2348 case P_RS_DATA_REQUEST
:
2349 case P_CSUM_RS_REQUEST
:
2351 drbd_send_ack_rp(mdev
, P_NEG_RS_DREPLY
, p
);
2355 dec_rs_pending(mdev
);
2356 drbd_send_ack_ex(mdev
, P_OV_RESULT
, sector
, size
, ID_IN_SYNC
);
2361 if (verb
&& __ratelimit(&drbd_ratelimit_state
))
2362 dev_err(DEV
, "Can not satisfy peer's read request, "
2363 "no local data.\n");
2365 /* drain possibly payload */
2366 return drbd_drain_block(mdev
, pi
->size
);
2369 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2370 * "criss-cross" setup, that might cause write-out on some other DRBD,
2371 * which in turn might block on the other node at this very place. */
2372 peer_req
= drbd_alloc_peer_req(mdev
, p
->block_id
, sector
, size
, GFP_NOIO
);
2379 case P_DATA_REQUEST
:
2380 peer_req
->w
.cb
= w_e_end_data_req
;
2381 fault_type
= DRBD_FAULT_DT_RD
;
2382 /* application IO, don't drbd_rs_begin_io */
2385 case P_RS_DATA_REQUEST
:
2386 peer_req
->w
.cb
= w_e_end_rsdata_req
;
2387 fault_type
= DRBD_FAULT_RS_RD
;
2388 /* used in the sector offset progress display */
2389 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2393 case P_CSUM_RS_REQUEST
:
2394 fault_type
= DRBD_FAULT_RS_RD
;
2395 di
= kmalloc(sizeof(*di
) + pi
->size
, GFP_NOIO
);
2399 di
->digest_size
= pi
->size
;
2400 di
->digest
= (((char *)di
)+sizeof(struct digest_info
));
2402 peer_req
->digest
= di
;
2403 peer_req
->flags
|= EE_HAS_DIGEST
;
2405 if (drbd_recv_all(mdev
->tconn
, di
->digest
, pi
->size
))
2408 if (pi
->cmd
== P_CSUM_RS_REQUEST
) {
2409 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
2410 peer_req
->w
.cb
= w_e_end_csum_rs_req
;
2411 /* used in the sector offset progress display */
2412 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2413 } else if (pi
->cmd
== P_OV_REPLY
) {
2414 /* track progress, we may need to throttle */
2415 atomic_add(size
>> 9, &mdev
->rs_sect_in
);
2416 peer_req
->w
.cb
= w_e_end_ov_reply
;
2417 dec_rs_pending(mdev
);
2418 /* drbd_rs_begin_io done when we sent this request,
2419 * but accounting still needs to be done. */
2420 goto submit_for_resync
;
2425 if (mdev
->ov_start_sector
== ~(sector_t
)0 &&
2426 mdev
->tconn
->agreed_pro_version
>= 90) {
2427 unsigned long now
= jiffies
;
2429 mdev
->ov_start_sector
= sector
;
2430 mdev
->ov_position
= sector
;
2431 mdev
->ov_left
= drbd_bm_bits(mdev
) - BM_SECT_TO_BIT(sector
);
2432 mdev
->rs_total
= mdev
->ov_left
;
2433 for (i
= 0; i
< DRBD_SYNC_MARKS
; i
++) {
2434 mdev
->rs_mark_left
[i
] = mdev
->ov_left
;
2435 mdev
->rs_mark_time
[i
] = now
;
2437 dev_info(DEV
, "Online Verify start sector: %llu\n",
2438 (unsigned long long)sector
);
2440 peer_req
->w
.cb
= w_e_end_ov_req
;
2441 fault_type
= DRBD_FAULT_RS_RD
;
2448 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2449 * wrt the receiver, but it is not as straightforward as it may seem.
2450 * Various places in the resync start and stop logic assume resync
2451 * requests are processed in order, requeuing this on the worker thread
2452 * introduces a bunch of new code for synchronization between threads.
2454 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2455 * "forever", throttling after drbd_rs_begin_io will lock that extent
2456 * for application writes for the same time. For now, just throttle
2457 * here, where the rest of the code expects the receiver to sleep for
2461 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2462 * this defers syncer requests for some time, before letting at least
2463 * on request through. The resync controller on the receiving side
2464 * will adapt to the incoming rate accordingly.
2466 * We cannot throttle here if remote is Primary/SyncTarget:
2467 * we would also throttle its application reads.
2468 * In that case, throttling is done on the SyncTarget only.
2470 if (mdev
->state
.peer
!= R_PRIMARY
&& drbd_rs_should_slow_down(mdev
, sector
))
2471 schedule_timeout_uninterruptible(HZ
/10);
2472 if (drbd_rs_begin_io(mdev
, sector
))
2476 atomic_add(size
>> 9, &mdev
->rs_sect_ev
);
2480 spin_lock_irq(&mdev
->tconn
->req_lock
);
2481 list_add_tail(&peer_req
->w
.list
, &mdev
->read_ee
);
2482 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2484 if (drbd_submit_peer_request(mdev
, peer_req
, READ
, fault_type
) == 0)
2487 /* don't care for the reason here */
2488 dev_err(DEV
, "submit failed, triggering re-connect\n");
2489 spin_lock_irq(&mdev
->tconn
->req_lock
);
2490 list_del(&peer_req
->w
.list
);
2491 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2492 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2496 drbd_free_peer_req(mdev
, peer_req
);
2500 static int drbd_asb_recover_0p(struct drbd_conf
*mdev
) __must_hold(local
)
2502 int self
, peer
, rv
= -100;
2503 unsigned long ch_self
, ch_peer
;
2504 enum drbd_after_sb_p after_sb_0p
;
2506 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2507 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2509 ch_peer
= mdev
->p_uuid
[UI_SIZE
];
2510 ch_self
= mdev
->comm_bm_set
;
2513 after_sb_0p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_0p
;
2515 switch (after_sb_0p
) {
2517 case ASB_DISCARD_SECONDARY
:
2518 case ASB_CALL_HELPER
:
2520 dev_err(DEV
, "Configuration error.\n");
2522 case ASB_DISCONNECT
:
2524 case ASB_DISCARD_YOUNGER_PRI
:
2525 if (self
== 0 && peer
== 1) {
2529 if (self
== 1 && peer
== 0) {
2533 /* Else fall through to one of the other strategies... */
2534 case ASB_DISCARD_OLDER_PRI
:
2535 if (self
== 0 && peer
== 1) {
2539 if (self
== 1 && peer
== 0) {
2543 /* Else fall through to one of the other strategies... */
2544 dev_warn(DEV
, "Discard younger/older primary did not find a decision\n"
2545 "Using discard-least-changes instead\n");
2546 case ASB_DISCARD_ZERO_CHG
:
2547 if (ch_peer
== 0 && ch_self
== 0) {
2548 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
)
2552 if (ch_peer
== 0) { rv
= 1; break; }
2553 if (ch_self
== 0) { rv
= -1; break; }
2555 if (after_sb_0p
== ASB_DISCARD_ZERO_CHG
)
2557 case ASB_DISCARD_LEAST_CHG
:
2558 if (ch_self
< ch_peer
)
2560 else if (ch_self
> ch_peer
)
2562 else /* ( ch_self == ch_peer ) */
2563 /* Well, then use something else. */
2564 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
)
2567 case ASB_DISCARD_LOCAL
:
2570 case ASB_DISCARD_REMOTE
:
2577 static int drbd_asb_recover_1p(struct drbd_conf
*mdev
) __must_hold(local
)
2580 enum drbd_after_sb_p after_sb_1p
;
2583 after_sb_1p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_1p
;
2585 switch (after_sb_1p
) {
2586 case ASB_DISCARD_YOUNGER_PRI
:
2587 case ASB_DISCARD_OLDER_PRI
:
2588 case ASB_DISCARD_LEAST_CHG
:
2589 case ASB_DISCARD_LOCAL
:
2590 case ASB_DISCARD_REMOTE
:
2591 case ASB_DISCARD_ZERO_CHG
:
2592 dev_err(DEV
, "Configuration error.\n");
2594 case ASB_DISCONNECT
:
2597 hg
= drbd_asb_recover_0p(mdev
);
2598 if (hg
== -1 && mdev
->state
.role
== R_SECONDARY
)
2600 if (hg
== 1 && mdev
->state
.role
== R_PRIMARY
)
2604 rv
= drbd_asb_recover_0p(mdev
);
2606 case ASB_DISCARD_SECONDARY
:
2607 return mdev
->state
.role
== R_PRIMARY
? 1 : -1;
2608 case ASB_CALL_HELPER
:
2609 hg
= drbd_asb_recover_0p(mdev
);
2610 if (hg
== -1 && mdev
->state
.role
== R_PRIMARY
) {
2611 enum drbd_state_rv rv2
;
2613 drbd_set_role(mdev
, R_SECONDARY
, 0);
2614 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2615 * we might be here in C_WF_REPORT_PARAMS which is transient.
2616 * we do not need to wait for the after state change work either. */
2617 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2618 if (rv2
!= SS_SUCCESS
) {
2619 drbd_khelper(mdev
, "pri-lost-after-sb");
2621 dev_warn(DEV
, "Successfully gave up primary role.\n");
2631 static int drbd_asb_recover_2p(struct drbd_conf
*mdev
) __must_hold(local
)
2634 enum drbd_after_sb_p after_sb_2p
;
2637 after_sb_2p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_2p
;
2639 switch (after_sb_2p
) {
2640 case ASB_DISCARD_YOUNGER_PRI
:
2641 case ASB_DISCARD_OLDER_PRI
:
2642 case ASB_DISCARD_LEAST_CHG
:
2643 case ASB_DISCARD_LOCAL
:
2644 case ASB_DISCARD_REMOTE
:
2646 case ASB_DISCARD_SECONDARY
:
2647 case ASB_DISCARD_ZERO_CHG
:
2648 dev_err(DEV
, "Configuration error.\n");
2651 rv
= drbd_asb_recover_0p(mdev
);
2653 case ASB_DISCONNECT
:
2655 case ASB_CALL_HELPER
:
2656 hg
= drbd_asb_recover_0p(mdev
);
2658 enum drbd_state_rv rv2
;
2660 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2661 * we might be here in C_WF_REPORT_PARAMS which is transient.
2662 * we do not need to wait for the after state change work either. */
2663 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2664 if (rv2
!= SS_SUCCESS
) {
2665 drbd_khelper(mdev
, "pri-lost-after-sb");
2667 dev_warn(DEV
, "Successfully gave up primary role.\n");
2677 static void drbd_uuid_dump(struct drbd_conf
*mdev
, char *text
, u64
*uuid
,
2678 u64 bits
, u64 flags
)
2681 dev_info(DEV
, "%s uuid info vanished while I was looking!\n", text
);
2684 dev_info(DEV
, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2686 (unsigned long long)uuid
[UI_CURRENT
],
2687 (unsigned long long)uuid
[UI_BITMAP
],
2688 (unsigned long long)uuid
[UI_HISTORY_START
],
2689 (unsigned long long)uuid
[UI_HISTORY_END
],
2690 (unsigned long long)bits
,
2691 (unsigned long long)flags
);
2695 100 after split brain try auto recover
2696 2 C_SYNC_SOURCE set BitMap
2697 1 C_SYNC_SOURCE use BitMap
2699 -1 C_SYNC_TARGET use BitMap
2700 -2 C_SYNC_TARGET set BitMap
2701 -100 after split brain, disconnect
2702 -1000 unrelated data
2703 -1091 requires proto 91
2704 -1096 requires proto 96
2706 static int drbd_uuid_compare(struct drbd_conf
*mdev
, int *rule_nr
) __must_hold(local
)
2711 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2712 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2715 if (self
== UUID_JUST_CREATED
&& peer
== UUID_JUST_CREATED
)
2719 if ((self
== UUID_JUST_CREATED
|| self
== (u64
)0) &&
2720 peer
!= UUID_JUST_CREATED
)
2724 if (self
!= UUID_JUST_CREATED
&&
2725 (peer
== UUID_JUST_CREATED
|| peer
== (u64
)0))
2729 int rct
, dc
; /* roles at crash time */
2731 if (mdev
->p_uuid
[UI_BITMAP
] == (u64
)0 && mdev
->ldev
->md
.uuid
[UI_BITMAP
] != (u64
)0) {
2733 if (mdev
->tconn
->agreed_pro_version
< 91)
2736 if ((mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) &&
2737 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1))) {
2738 dev_info(DEV
, "was SyncSource, missed the resync finished event, corrected myself:\n");
2739 drbd_uuid_set_bm(mdev
, 0UL);
2741 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2742 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2745 dev_info(DEV
, "was SyncSource (peer failed to write sync_uuid)\n");
2752 if (mdev
->ldev
->md
.uuid
[UI_BITMAP
] == (u64
)0 && mdev
->p_uuid
[UI_BITMAP
] != (u64
)0) {
2754 if (mdev
->tconn
->agreed_pro_version
< 91)
2757 if ((mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1)) &&
2758 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1))) {
2759 dev_info(DEV
, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2761 mdev
->p_uuid
[UI_HISTORY_START
+ 1] = mdev
->p_uuid
[UI_HISTORY_START
];
2762 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_BITMAP
];
2763 mdev
->p_uuid
[UI_BITMAP
] = 0UL;
2765 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2768 dev_info(DEV
, "was SyncTarget (failed to write sync_uuid)\n");
2775 /* Common power [off|failure] */
2776 rct
= (test_bit(CRASHED_PRIMARY
, &mdev
->flags
) ? 1 : 0) +
2777 (mdev
->p_uuid
[UI_FLAGS
] & 2);
2778 /* lowest bit is set when we were primary,
2779 * next bit (weight 2) is set when peer was primary */
2783 case 0: /* !self_pri && !peer_pri */ return 0;
2784 case 1: /* self_pri && !peer_pri */ return 1;
2785 case 2: /* !self_pri && peer_pri */ return -1;
2786 case 3: /* self_pri && peer_pri */
2787 dc
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
);
2793 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2798 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2800 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2801 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) ==
2802 (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) :
2803 peer
+ UUID_NEW_BM_OFFSET
== (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1))) {
2804 /* The last P_SYNC_UUID did not get though. Undo the last start of
2805 resync as sync source modifications of the peer's UUIDs. */
2807 if (mdev
->tconn
->agreed_pro_version
< 91)
2810 mdev
->p_uuid
[UI_BITMAP
] = mdev
->p_uuid
[UI_HISTORY_START
];
2811 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_HISTORY_START
+ 1];
2813 dev_info(DEV
, "Did not got last syncUUID packet, corrected:\n");
2814 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2821 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2822 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2823 peer
= mdev
->p_uuid
[i
] & ~((u64
)1);
2829 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2830 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2835 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2837 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2838 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) ==
2839 (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) :
2840 self
+ UUID_NEW_BM_OFFSET
== (mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1))) {
2841 /* The last P_SYNC_UUID did not get though. Undo the last start of
2842 resync as sync source modifications of our UUIDs. */
2844 if (mdev
->tconn
->agreed_pro_version
< 91)
2847 _drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
]);
2848 _drbd_uuid_set(mdev
, UI_HISTORY_START
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1]);
2850 dev_info(DEV
, "Last syncUUID did not get through, corrected:\n");
2851 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2852 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2860 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2861 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2862 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2868 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2869 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2870 if (self
== peer
&& self
!= ((u64
)0))
2874 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2875 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2876 for (j
= UI_HISTORY_START
; j
<= UI_HISTORY_END
; j
++) {
2877 peer
= mdev
->p_uuid
[j
] & ~((u64
)1);
2886 /* drbd_sync_handshake() returns the new conn state on success, or
2887 CONN_MASK (-1) on failure.
2889 static enum drbd_conns
drbd_sync_handshake(struct drbd_conf
*mdev
, enum drbd_role peer_role
,
2890 enum drbd_disk_state peer_disk
) __must_hold(local
)
2892 enum drbd_conns rv
= C_MASK
;
2893 enum drbd_disk_state mydisk
;
2894 struct net_conf
*nc
;
2895 int hg
, rule_nr
, rr_conflict
, tentative
;
2897 mydisk
= mdev
->state
.disk
;
2898 if (mydisk
== D_NEGOTIATING
)
2899 mydisk
= mdev
->new_state_tmp
.disk
;
2901 dev_info(DEV
, "drbd_sync_handshake:\n");
2902 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
, mdev
->comm_bm_set
, 0);
2903 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
,
2904 mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2906 hg
= drbd_uuid_compare(mdev
, &rule_nr
);
2908 dev_info(DEV
, "uuid_compare()=%d by rule %d\n", hg
, rule_nr
);
2911 dev_alert(DEV
, "Unrelated data, aborting!\n");
2915 dev_alert(DEV
, "To resolve this both sides have to support at least protocol %d\n", -hg
- 1000);
2919 if ((mydisk
== D_INCONSISTENT
&& peer_disk
> D_INCONSISTENT
) ||
2920 (peer_disk
== D_INCONSISTENT
&& mydisk
> D_INCONSISTENT
)) {
2921 int f
= (hg
== -100) || abs(hg
) == 2;
2922 hg
= mydisk
> D_INCONSISTENT
? 1 : -1;
2925 dev_info(DEV
, "Becoming sync %s due to disk states.\n",
2926 hg
> 0 ? "source" : "target");
2930 drbd_khelper(mdev
, "initial-split-brain");
2933 nc
= rcu_dereference(mdev
->tconn
->net_conf
);
2935 if (hg
== 100 || (hg
== -100 && nc
->always_asbp
)) {
2936 int pcount
= (mdev
->state
.role
== R_PRIMARY
)
2937 + (peer_role
== R_PRIMARY
);
2938 int forced
= (hg
== -100);
2942 hg
= drbd_asb_recover_0p(mdev
);
2945 hg
= drbd_asb_recover_1p(mdev
);
2948 hg
= drbd_asb_recover_2p(mdev
);
2951 if (abs(hg
) < 100) {
2952 dev_warn(DEV
, "Split-Brain detected, %d primaries, "
2953 "automatically solved. Sync from %s node\n",
2954 pcount
, (hg
< 0) ? "peer" : "this");
2956 dev_warn(DEV
, "Doing a full sync, since"
2957 " UUIDs where ambiguous.\n");
2964 if (test_bit(DISCARD_MY_DATA
, &mdev
->flags
) && !(mdev
->p_uuid
[UI_FLAGS
]&1))
2966 if (!test_bit(DISCARD_MY_DATA
, &mdev
->flags
) && (mdev
->p_uuid
[UI_FLAGS
]&1))
2970 dev_warn(DEV
, "Split-Brain detected, manually solved. "
2971 "Sync from %s node\n",
2972 (hg
< 0) ? "peer" : "this");
2974 rr_conflict
= nc
->rr_conflict
;
2975 tentative
= nc
->tentative
;
2979 /* FIXME this log message is not correct if we end up here
2980 * after an attempted attach on a diskless node.
2981 * We just refuse to attach -- well, we drop the "connection"
2982 * to that disk, in a way... */
2983 dev_alert(DEV
, "Split-Brain detected but unresolved, dropping connection!\n");
2984 drbd_khelper(mdev
, "split-brain");
2988 if (hg
> 0 && mydisk
<= D_INCONSISTENT
) {
2989 dev_err(DEV
, "I shall become SyncSource, but I am inconsistent!\n");
2993 if (hg
< 0 && /* by intention we do not use mydisk here. */
2994 mdev
->state
.role
== R_PRIMARY
&& mdev
->state
.disk
>= D_CONSISTENT
) {
2995 switch (rr_conflict
) {
2996 case ASB_CALL_HELPER
:
2997 drbd_khelper(mdev
, "pri-lost");
2999 case ASB_DISCONNECT
:
3000 dev_err(DEV
, "I shall become SyncTarget, but I am primary!\n");
3003 dev_warn(DEV
, "Becoming SyncTarget, violating the stable-data"
3008 if (tentative
|| test_bit(CONN_DRY_RUN
, &mdev
->tconn
->flags
)) {
3010 dev_info(DEV
, "dry-run connect: No resync, would become Connected immediately.\n");
3012 dev_info(DEV
, "dry-run connect: Would become %s, doing a %s resync.",
3013 drbd_conn_str(hg
> 0 ? C_SYNC_SOURCE
: C_SYNC_TARGET
),
3014 abs(hg
) >= 2 ? "full" : "bit-map based");
3019 dev_info(DEV
, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3020 if (drbd_bitmap_io(mdev
, &drbd_bmio_set_n_write
, "set_n_write from sync_handshake",
3021 BM_LOCKED_SET_ALLOWED
))
3025 if (hg
> 0) { /* become sync source. */
3027 } else if (hg
< 0) { /* become sync target */
3031 if (drbd_bm_total_weight(mdev
)) {
3032 dev_info(DEV
, "No resync, but %lu bits in bitmap!\n",
3033 drbd_bm_total_weight(mdev
));
3040 static enum drbd_after_sb_p
convert_after_sb(enum drbd_after_sb_p peer
)
3042 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3043 if (peer
== ASB_DISCARD_REMOTE
)
3044 return ASB_DISCARD_LOCAL
;
3046 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3047 if (peer
== ASB_DISCARD_LOCAL
)
3048 return ASB_DISCARD_REMOTE
;
3050 /* everything else is valid if they are equal on both sides. */
3054 static int receive_protocol(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3056 struct p_protocol
*p
= pi
->data
;
3057 enum drbd_after_sb_p p_after_sb_0p
, p_after_sb_1p
, p_after_sb_2p
;
3058 int p_proto
, p_discard_my_data
, p_two_primaries
, cf
;
3059 struct net_conf
*nc
, *old_net_conf
, *new_net_conf
= NULL
;
3060 char integrity_alg
[SHARED_SECRET_MAX
] = "";
3061 struct crypto_hash
*peer_integrity_tfm
= NULL
;
3062 void *int_dig_in
= NULL
, *int_dig_vv
= NULL
;
3064 p_proto
= be32_to_cpu(p
->protocol
);
3065 p_after_sb_0p
= be32_to_cpu(p
->after_sb_0p
);
3066 p_after_sb_1p
= be32_to_cpu(p
->after_sb_1p
);
3067 p_after_sb_2p
= be32_to_cpu(p
->after_sb_2p
);
3068 p_two_primaries
= be32_to_cpu(p
->two_primaries
);
3069 cf
= be32_to_cpu(p
->conn_flags
);
3070 p_discard_my_data
= cf
& CF_DISCARD_MY_DATA
;
3072 if (tconn
->agreed_pro_version
>= 87) {
3075 if (pi
->size
> sizeof(integrity_alg
))
3077 err
= drbd_recv_all(tconn
, integrity_alg
, pi
->size
);
3080 integrity_alg
[SHARED_SECRET_MAX
- 1] = 0;
3083 if (pi
->cmd
!= P_PROTOCOL_UPDATE
) {
3084 clear_bit(CONN_DRY_RUN
, &tconn
->flags
);
3086 if (cf
& CF_DRY_RUN
)
3087 set_bit(CONN_DRY_RUN
, &tconn
->flags
);
3090 nc
= rcu_dereference(tconn
->net_conf
);
3092 if (p_proto
!= nc
->wire_protocol
) {
3093 conn_err(tconn
, "incompatible %s settings\n", "protocol");
3094 goto disconnect_rcu_unlock
;
3097 if (convert_after_sb(p_after_sb_0p
) != nc
->after_sb_0p
) {
3098 conn_err(tconn
, "incompatible %s settings\n", "after-sb-0pri");
3099 goto disconnect_rcu_unlock
;
3102 if (convert_after_sb(p_after_sb_1p
) != nc
->after_sb_1p
) {
3103 conn_err(tconn
, "incompatible %s settings\n", "after-sb-1pri");
3104 goto disconnect_rcu_unlock
;
3107 if (convert_after_sb(p_after_sb_2p
) != nc
->after_sb_2p
) {
3108 conn_err(tconn
, "incompatible %s settings\n", "after-sb-2pri");
3109 goto disconnect_rcu_unlock
;
3112 if (p_discard_my_data
&& nc
->discard_my_data
) {
3113 conn_err(tconn
, "incompatible %s settings\n", "discard-my-data");
3114 goto disconnect_rcu_unlock
;
3117 if (p_two_primaries
!= nc
->two_primaries
) {
3118 conn_err(tconn
, "incompatible %s settings\n", "allow-two-primaries");
3119 goto disconnect_rcu_unlock
;
3122 if (strcmp(integrity_alg
, nc
->integrity_alg
)) {
3123 conn_err(tconn
, "incompatible %s settings\n", "data-integrity-alg");
3124 goto disconnect_rcu_unlock
;
3130 if (integrity_alg
[0]) {
3134 * We can only change the peer data integrity algorithm
3135 * here. Changing our own data integrity algorithm
3136 * requires that we send a P_PROTOCOL_UPDATE packet at
3137 * the same time; otherwise, the peer has no way to
3138 * tell between which packets the algorithm should
3142 peer_integrity_tfm
= crypto_alloc_hash(integrity_alg
, 0, CRYPTO_ALG_ASYNC
);
3143 if (!peer_integrity_tfm
) {
3144 conn_err(tconn
, "peer data-integrity-alg %s not supported\n",
3149 hash_size
= crypto_hash_digestsize(peer_integrity_tfm
);
3150 int_dig_in
= kmalloc(hash_size
, GFP_KERNEL
);
3151 int_dig_vv
= kmalloc(hash_size
, GFP_KERNEL
);
3152 if (!(int_dig_in
&& int_dig_vv
)) {
3153 conn_err(tconn
, "Allocation of buffers for data integrity checking failed\n");
3158 new_net_conf
= kmalloc(sizeof(struct net_conf
), GFP_KERNEL
);
3159 if (!new_net_conf
) {
3160 conn_err(tconn
, "Allocation of new net_conf failed\n");
3164 mutex_lock(&tconn
->data
.mutex
);
3165 mutex_lock(&tconn
->conf_update
);
3166 old_net_conf
= tconn
->net_conf
;
3167 *new_net_conf
= *old_net_conf
;
3169 new_net_conf
->wire_protocol
= p_proto
;
3170 new_net_conf
->after_sb_0p
= convert_after_sb(p_after_sb_0p
);
3171 new_net_conf
->after_sb_1p
= convert_after_sb(p_after_sb_1p
);
3172 new_net_conf
->after_sb_2p
= convert_after_sb(p_after_sb_2p
);
3173 new_net_conf
->two_primaries
= p_two_primaries
;
3175 rcu_assign_pointer(tconn
->net_conf
, new_net_conf
);
3176 mutex_unlock(&tconn
->conf_update
);
3177 mutex_unlock(&tconn
->data
.mutex
);
3179 crypto_free_hash(tconn
->peer_integrity_tfm
);
3180 kfree(tconn
->int_dig_in
);
3181 kfree(tconn
->int_dig_vv
);
3182 tconn
->peer_integrity_tfm
= peer_integrity_tfm
;
3183 tconn
->int_dig_in
= int_dig_in
;
3184 tconn
->int_dig_vv
= int_dig_vv
;
3186 if (strcmp(old_net_conf
->integrity_alg
, integrity_alg
))
3187 conn_info(tconn
, "peer data-integrity-alg: %s\n",
3188 integrity_alg
[0] ? integrity_alg
: "(none)");
3191 kfree(old_net_conf
);
3194 disconnect_rcu_unlock
:
3197 crypto_free_hash(peer_integrity_tfm
);
3200 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3205 * input: alg name, feature name
3206 * return: NULL (alg name was "")
3207 * ERR_PTR(error) if something goes wrong
3208 * or the crypto hash ptr, if it worked out ok. */
3209 struct crypto_hash
*drbd_crypto_alloc_digest_safe(const struct drbd_conf
*mdev
,
3210 const char *alg
, const char *name
)
3212 struct crypto_hash
*tfm
;
3217 tfm
= crypto_alloc_hash(alg
, 0, CRYPTO_ALG_ASYNC
);
3219 dev_err(DEV
, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3220 alg
, name
, PTR_ERR(tfm
));
3226 static int ignore_remaining_packet(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3228 void *buffer
= tconn
->data
.rbuf
;
3229 int size
= pi
->size
;
3232 int s
= min_t(int, size
, DRBD_SOCKET_BUFFER_SIZE
);
3233 s
= drbd_recv(tconn
, buffer
, s
);
3247 * config_unknown_volume - device configuration command for unknown volume
3249 * When a device is added to an existing connection, the node on which the
3250 * device is added first will send configuration commands to its peer but the
3251 * peer will not know about the device yet. It will warn and ignore these
3252 * commands. Once the device is added on the second node, the second node will
3253 * send the same device configuration commands, but in the other direction.
3255 * (We can also end up here if drbd is misconfigured.)
3257 static int config_unknown_volume(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3259 conn_warn(tconn
, "%s packet received for volume %u, which is not configured locally\n",
3260 cmdname(pi
->cmd
), pi
->vnr
);
3261 return ignore_remaining_packet(tconn
, pi
);
3264 static int receive_SyncParam(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3266 struct drbd_conf
*mdev
;
3267 struct p_rs_param_95
*p
;
3268 unsigned int header_size
, data_size
, exp_max_sz
;
3269 struct crypto_hash
*verify_tfm
= NULL
;
3270 struct crypto_hash
*csums_tfm
= NULL
;
3271 struct net_conf
*old_net_conf
, *new_net_conf
= NULL
;
3272 struct disk_conf
*old_disk_conf
= NULL
, *new_disk_conf
= NULL
;
3273 const int apv
= tconn
->agreed_pro_version
;
3274 struct fifo_buffer
*old_plan
= NULL
, *new_plan
= NULL
;
3278 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3280 return config_unknown_volume(tconn
, pi
);
3282 exp_max_sz
= apv
<= 87 ? sizeof(struct p_rs_param
)
3283 : apv
== 88 ? sizeof(struct p_rs_param
)
3285 : apv
<= 94 ? sizeof(struct p_rs_param_89
)
3286 : /* apv >= 95 */ sizeof(struct p_rs_param_95
);
3288 if (pi
->size
> exp_max_sz
) {
3289 dev_err(DEV
, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3290 pi
->size
, exp_max_sz
);
3295 header_size
= sizeof(struct p_rs_param
);
3296 data_size
= pi
->size
- header_size
;
3297 } else if (apv
<= 94) {
3298 header_size
= sizeof(struct p_rs_param_89
);
3299 data_size
= pi
->size
- header_size
;
3300 D_ASSERT(data_size
== 0);
3302 header_size
= sizeof(struct p_rs_param_95
);
3303 data_size
= pi
->size
- header_size
;
3304 D_ASSERT(data_size
== 0);
3307 /* initialize verify_alg and csums_alg */
3309 memset(p
->verify_alg
, 0, 2 * SHARED_SECRET_MAX
);
3311 err
= drbd_recv_all(mdev
->tconn
, p
, header_size
);
3315 mutex_lock(&mdev
->tconn
->conf_update
);
3316 old_net_conf
= mdev
->tconn
->net_conf
;
3317 if (get_ldev(mdev
)) {
3318 new_disk_conf
= kzalloc(sizeof(struct disk_conf
), GFP_KERNEL
);
3319 if (!new_disk_conf
) {
3321 mutex_unlock(&mdev
->tconn
->conf_update
);
3322 dev_err(DEV
, "Allocation of new disk_conf failed\n");
3326 old_disk_conf
= mdev
->ldev
->disk_conf
;
3327 *new_disk_conf
= *old_disk_conf
;
3329 new_disk_conf
->resync_rate
= be32_to_cpu(p
->resync_rate
);
3334 if (data_size
> SHARED_SECRET_MAX
) {
3335 dev_err(DEV
, "verify-alg too long, "
3336 "peer wants %u, accepting only %u byte\n",
3337 data_size
, SHARED_SECRET_MAX
);
3342 err
= drbd_recv_all(mdev
->tconn
, p
->verify_alg
, data_size
);
3345 /* we expect NUL terminated string */
3346 /* but just in case someone tries to be evil */
3347 D_ASSERT(p
->verify_alg
[data_size
-1] == 0);
3348 p
->verify_alg
[data_size
-1] = 0;
3350 } else /* apv >= 89 */ {
3351 /* we still expect NUL terminated strings */
3352 /* but just in case someone tries to be evil */
3353 D_ASSERT(p
->verify_alg
[SHARED_SECRET_MAX
-1] == 0);
3354 D_ASSERT(p
->csums_alg
[SHARED_SECRET_MAX
-1] == 0);
3355 p
->verify_alg
[SHARED_SECRET_MAX
-1] = 0;
3356 p
->csums_alg
[SHARED_SECRET_MAX
-1] = 0;
3359 if (strcmp(old_net_conf
->verify_alg
, p
->verify_alg
)) {
3360 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
3361 dev_err(DEV
, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3362 old_net_conf
->verify_alg
, p
->verify_alg
);
3365 verify_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
3366 p
->verify_alg
, "verify-alg");
3367 if (IS_ERR(verify_tfm
)) {
3373 if (apv
>= 89 && strcmp(old_net_conf
->csums_alg
, p
->csums_alg
)) {
3374 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
3375 dev_err(DEV
, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3376 old_net_conf
->csums_alg
, p
->csums_alg
);
3379 csums_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
3380 p
->csums_alg
, "csums-alg");
3381 if (IS_ERR(csums_tfm
)) {
3387 if (apv
> 94 && new_disk_conf
) {
3388 new_disk_conf
->c_plan_ahead
= be32_to_cpu(p
->c_plan_ahead
);
3389 new_disk_conf
->c_delay_target
= be32_to_cpu(p
->c_delay_target
);
3390 new_disk_conf
->c_fill_target
= be32_to_cpu(p
->c_fill_target
);
3391 new_disk_conf
->c_max_rate
= be32_to_cpu(p
->c_max_rate
);
3393 fifo_size
= (new_disk_conf
->c_plan_ahead
* 10 * SLEEP_TIME
) / HZ
;
3394 if (fifo_size
!= mdev
->rs_plan_s
->size
) {
3395 new_plan
= fifo_alloc(fifo_size
);
3397 dev_err(DEV
, "kmalloc of fifo_buffer failed");
3404 if (verify_tfm
|| csums_tfm
) {
3405 new_net_conf
= kzalloc(sizeof(struct net_conf
), GFP_KERNEL
);
3406 if (!new_net_conf
) {
3407 dev_err(DEV
, "Allocation of new net_conf failed\n");
3411 *new_net_conf
= *old_net_conf
;
3414 strcpy(new_net_conf
->verify_alg
, p
->verify_alg
);
3415 new_net_conf
->verify_alg_len
= strlen(p
->verify_alg
) + 1;
3416 crypto_free_hash(mdev
->tconn
->verify_tfm
);
3417 mdev
->tconn
->verify_tfm
= verify_tfm
;
3418 dev_info(DEV
, "using verify-alg: \"%s\"\n", p
->verify_alg
);
3421 strcpy(new_net_conf
->csums_alg
, p
->csums_alg
);
3422 new_net_conf
->csums_alg_len
= strlen(p
->csums_alg
) + 1;
3423 crypto_free_hash(mdev
->tconn
->csums_tfm
);
3424 mdev
->tconn
->csums_tfm
= csums_tfm
;
3425 dev_info(DEV
, "using csums-alg: \"%s\"\n", p
->csums_alg
);
3427 rcu_assign_pointer(tconn
->net_conf
, new_net_conf
);
3431 if (new_disk_conf
) {
3432 rcu_assign_pointer(mdev
->ldev
->disk_conf
, new_disk_conf
);
3437 old_plan
= mdev
->rs_plan_s
;
3438 rcu_assign_pointer(mdev
->rs_plan_s
, new_plan
);
3441 mutex_unlock(&mdev
->tconn
->conf_update
);
3444 kfree(old_net_conf
);
3445 kfree(old_disk_conf
);
3451 if (new_disk_conf
) {
3453 kfree(new_disk_conf
);
3455 mutex_unlock(&mdev
->tconn
->conf_update
);
3460 if (new_disk_conf
) {
3462 kfree(new_disk_conf
);
3464 mutex_unlock(&mdev
->tconn
->conf_update
);
3465 /* just for completeness: actually not needed,
3466 * as this is not reached if csums_tfm was ok. */
3467 crypto_free_hash(csums_tfm
);
3468 /* but free the verify_tfm again, if csums_tfm did not work out */
3469 crypto_free_hash(verify_tfm
);
3470 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3474 /* warn if the arguments differ by more than 12.5% */
3475 static void warn_if_differ_considerably(struct drbd_conf
*mdev
,
3476 const char *s
, sector_t a
, sector_t b
)
3479 if (a
== 0 || b
== 0)
3481 d
= (a
> b
) ? (a
- b
) : (b
- a
);
3482 if (d
> (a
>>3) || d
> (b
>>3))
3483 dev_warn(DEV
, "Considerable difference in %s: %llus vs. %llus\n", s
,
3484 (unsigned long long)a
, (unsigned long long)b
);
3487 static int receive_sizes(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3489 struct drbd_conf
*mdev
;
3490 struct p_sizes
*p
= pi
->data
;
3491 enum determine_dev_size dd
= unchanged
;
3492 sector_t p_size
, p_usize
, my_usize
;
3493 int ldsc
= 0; /* local disk size changed */
3494 enum dds_flags ddsf
;
3496 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3498 return config_unknown_volume(tconn
, pi
);
3500 p_size
= be64_to_cpu(p
->d_size
);
3501 p_usize
= be64_to_cpu(p
->u_size
);
3503 /* just store the peer's disk size for now.
3504 * we still need to figure out whether we accept that. */
3505 mdev
->p_size
= p_size
;
3507 if (get_ldev(mdev
)) {
3509 my_usize
= rcu_dereference(mdev
->ldev
->disk_conf
)->disk_size
;
3512 warn_if_differ_considerably(mdev
, "lower level device sizes",
3513 p_size
, drbd_get_max_capacity(mdev
->ldev
));
3514 warn_if_differ_considerably(mdev
, "user requested size",
3517 /* if this is the first connect, or an otherwise expected
3518 * param exchange, choose the minimum */
3519 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
)
3520 p_usize
= min_not_zero(my_usize
, p_usize
);
3522 /* Never shrink a device with usable data during connect.
3523 But allow online shrinking if we are connected. */
3524 if (drbd_new_dev_size(mdev
, mdev
->ldev
, p_usize
, 0) <
3525 drbd_get_capacity(mdev
->this_bdev
) &&
3526 mdev
->state
.disk
>= D_OUTDATED
&&
3527 mdev
->state
.conn
< C_CONNECTED
) {
3528 dev_err(DEV
, "The peer's disk size is too small!\n");
3529 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3534 if (my_usize
!= p_usize
) {
3535 struct disk_conf
*old_disk_conf
, *new_disk_conf
= NULL
;
3537 new_disk_conf
= kzalloc(sizeof(struct disk_conf
), GFP_KERNEL
);
3538 if (!new_disk_conf
) {
3539 dev_err(DEV
, "Allocation of new disk_conf failed\n");
3544 mutex_lock(&mdev
->tconn
->conf_update
);
3545 old_disk_conf
= mdev
->ldev
->disk_conf
;
3546 *new_disk_conf
= *old_disk_conf
;
3547 new_disk_conf
->disk_size
= p_usize
;
3549 rcu_assign_pointer(mdev
->ldev
->disk_conf
, new_disk_conf
);
3550 mutex_unlock(&mdev
->tconn
->conf_update
);
3552 kfree(old_disk_conf
);
3554 dev_info(DEV
, "Peer sets u_size to %lu sectors\n",
3555 (unsigned long)my_usize
);
3561 ddsf
= be16_to_cpu(p
->dds_flags
);
3562 if (get_ldev(mdev
)) {
3563 dd
= drbd_determine_dev_size(mdev
, ddsf
);
3565 if (dd
== dev_size_error
)
3569 /* I am diskless, need to accept the peer's size. */
3570 drbd_set_my_capacity(mdev
, p_size
);
3573 mdev
->peer_max_bio_size
= be32_to_cpu(p
->max_bio_size
);
3574 drbd_reconsider_max_bio_size(mdev
);
3576 if (get_ldev(mdev
)) {
3577 if (mdev
->ldev
->known_size
!= drbd_get_capacity(mdev
->ldev
->backing_bdev
)) {
3578 mdev
->ldev
->known_size
= drbd_get_capacity(mdev
->ldev
->backing_bdev
);
3585 if (mdev
->state
.conn
> C_WF_REPORT_PARAMS
) {
3586 if (be64_to_cpu(p
->c_size
) !=
3587 drbd_get_capacity(mdev
->this_bdev
) || ldsc
) {
3588 /* we have different sizes, probably peer
3589 * needs to know my new size... */
3590 drbd_send_sizes(mdev
, 0, ddsf
);
3592 if (test_and_clear_bit(RESIZE_PENDING
, &mdev
->flags
) ||
3593 (dd
== grew
&& mdev
->state
.conn
== C_CONNECTED
)) {
3594 if (mdev
->state
.pdsk
>= D_INCONSISTENT
&&
3595 mdev
->state
.disk
>= D_INCONSISTENT
) {
3596 if (ddsf
& DDSF_NO_RESYNC
)
3597 dev_info(DEV
, "Resync of new storage suppressed with --assume-clean\n");
3599 resync_after_online_grow(mdev
);
3601 set_bit(RESYNC_AFTER_NEG
, &mdev
->flags
);
3608 static int receive_uuids(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3610 struct drbd_conf
*mdev
;
3611 struct p_uuids
*p
= pi
->data
;
3613 int i
, updated_uuids
= 0;
3615 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3617 return config_unknown_volume(tconn
, pi
);
3619 p_uuid
= kmalloc(sizeof(u64
)*UI_EXTENDED_SIZE
, GFP_NOIO
);
3621 for (i
= UI_CURRENT
; i
< UI_EXTENDED_SIZE
; i
++)
3622 p_uuid
[i
] = be64_to_cpu(p
->uuid
[i
]);
3624 kfree(mdev
->p_uuid
);
3625 mdev
->p_uuid
= p_uuid
;
3627 if (mdev
->state
.conn
< C_CONNECTED
&&
3628 mdev
->state
.disk
< D_INCONSISTENT
&&
3629 mdev
->state
.role
== R_PRIMARY
&&
3630 (mdev
->ed_uuid
& ~((u64
)1)) != (p_uuid
[UI_CURRENT
] & ~((u64
)1))) {
3631 dev_err(DEV
, "Can only connect to data with current UUID=%016llX\n",
3632 (unsigned long long)mdev
->ed_uuid
);
3633 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3637 if (get_ldev(mdev
)) {
3638 int skip_initial_sync
=
3639 mdev
->state
.conn
== C_CONNECTED
&&
3640 mdev
->tconn
->agreed_pro_version
>= 90 &&
3641 mdev
->ldev
->md
.uuid
[UI_CURRENT
] == UUID_JUST_CREATED
&&
3642 (p_uuid
[UI_FLAGS
] & 8);
3643 if (skip_initial_sync
) {
3644 dev_info(DEV
, "Accepted new current UUID, preparing to skip initial sync\n");
3645 drbd_bitmap_io(mdev
, &drbd_bmio_clear_n_write
,
3646 "clear_n_write from receive_uuids",
3647 BM_LOCKED_TEST_ALLOWED
);
3648 _drbd_uuid_set(mdev
, UI_CURRENT
, p_uuid
[UI_CURRENT
]);
3649 _drbd_uuid_set(mdev
, UI_BITMAP
, 0);
3650 _drbd_set_state(_NS2(mdev
, disk
, D_UP_TO_DATE
, pdsk
, D_UP_TO_DATE
),
3656 } else if (mdev
->state
.disk
< D_INCONSISTENT
&&
3657 mdev
->state
.role
== R_PRIMARY
) {
3658 /* I am a diskless primary, the peer just created a new current UUID
3660 updated_uuids
= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3663 /* Before we test for the disk state, we should wait until an eventually
3664 ongoing cluster wide state change is finished. That is important if
3665 we are primary and are detaching from our disk. We need to see the
3666 new disk state... */
3667 mutex_lock(mdev
->state_mutex
);
3668 mutex_unlock(mdev
->state_mutex
);
3669 if (mdev
->state
.conn
>= C_CONNECTED
&& mdev
->state
.disk
< D_INCONSISTENT
)
3670 updated_uuids
|= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3673 drbd_print_uuids(mdev
, "receiver updated UUIDs to");
3679 * convert_state() - Converts the peer's view of the cluster state to our point of view
3680 * @ps: The state as seen by the peer.
3682 static union drbd_state
convert_state(union drbd_state ps
)
3684 union drbd_state ms
;
3686 static enum drbd_conns c_tab
[] = {
3687 [C_WF_REPORT_PARAMS
] = C_WF_REPORT_PARAMS
,
3688 [C_CONNECTED
] = C_CONNECTED
,
3690 [C_STARTING_SYNC_S
] = C_STARTING_SYNC_T
,
3691 [C_STARTING_SYNC_T
] = C_STARTING_SYNC_S
,
3692 [C_DISCONNECTING
] = C_TEAR_DOWN
, /* C_NETWORK_FAILURE, */
3693 [C_VERIFY_S
] = C_VERIFY_T
,
3699 ms
.conn
= c_tab
[ps
.conn
];
3704 ms
.peer_isp
= (ps
.aftr_isp
| ps
.user_isp
);
3709 static int receive_req_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3711 struct drbd_conf
*mdev
;
3712 struct p_req_state
*p
= pi
->data
;
3713 union drbd_state mask
, val
;
3714 enum drbd_state_rv rv
;
3716 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3720 mask
.i
= be32_to_cpu(p
->mask
);
3721 val
.i
= be32_to_cpu(p
->val
);
3723 if (test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
) &&
3724 mutex_is_locked(mdev
->state_mutex
)) {
3725 drbd_send_sr_reply(mdev
, SS_CONCURRENT_ST_CHG
);
3729 mask
= convert_state(mask
);
3730 val
= convert_state(val
);
3732 rv
= drbd_change_state(mdev
, CS_VERBOSE
, mask
, val
);
3733 drbd_send_sr_reply(mdev
, rv
);
3740 static int receive_req_conn_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3742 struct p_req_state
*p
= pi
->data
;
3743 union drbd_state mask
, val
;
3744 enum drbd_state_rv rv
;
3746 mask
.i
= be32_to_cpu(p
->mask
);
3747 val
.i
= be32_to_cpu(p
->val
);
3749 if (test_bit(DISCARD_CONCURRENT
, &tconn
->flags
) &&
3750 mutex_is_locked(&tconn
->cstate_mutex
)) {
3751 conn_send_sr_reply(tconn
, SS_CONCURRENT_ST_CHG
);
3755 mask
= convert_state(mask
);
3756 val
= convert_state(val
);
3758 rv
= conn_request_state(tconn
, mask
, val
, CS_VERBOSE
| CS_LOCAL_ONLY
| CS_IGN_OUTD_FAIL
);
3759 conn_send_sr_reply(tconn
, rv
);
3764 static int receive_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3766 struct drbd_conf
*mdev
;
3767 struct p_state
*p
= pi
->data
;
3768 union drbd_state os
, ns
, peer_state
;
3769 enum drbd_disk_state real_peer_disk
;
3770 enum chg_state_flags cs_flags
;
3773 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3775 return config_unknown_volume(tconn
, pi
);
3777 peer_state
.i
= be32_to_cpu(p
->state
);
3779 real_peer_disk
= peer_state
.disk
;
3780 if (peer_state
.disk
== D_NEGOTIATING
) {
3781 real_peer_disk
= mdev
->p_uuid
[UI_FLAGS
] & 4 ? D_INCONSISTENT
: D_CONSISTENT
;
3782 dev_info(DEV
, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk
));
3785 spin_lock_irq(&mdev
->tconn
->req_lock
);
3787 os
= ns
= drbd_read_state(mdev
);
3788 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3790 /* If this is the "end of sync" confirmation, usually the peer disk
3791 * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
3792 * set) resync started in PausedSyncT, or if the timing of pause-/
3793 * unpause-sync events has been "just right", the peer disk may
3794 * transition from D_CONSISTENT to D_UP_TO_DATE as well.
3796 if ((os
.pdsk
== D_INCONSISTENT
|| os
.pdsk
== D_CONSISTENT
) &&
3797 real_peer_disk
== D_UP_TO_DATE
&&
3798 os
.conn
> C_CONNECTED
&& os
.disk
== D_UP_TO_DATE
) {
3799 /* If we are (becoming) SyncSource, but peer is still in sync
3800 * preparation, ignore its uptodate-ness to avoid flapping, it
3801 * will change to inconsistent once the peer reaches active
3803 * It may have changed syncer-paused flags, however, so we
3804 * cannot ignore this completely. */
3805 if (peer_state
.conn
> C_CONNECTED
&&
3806 peer_state
.conn
< C_SYNC_SOURCE
)
3807 real_peer_disk
= D_INCONSISTENT
;
3809 /* if peer_state changes to connected at the same time,
3810 * it explicitly notifies us that it finished resync.
3811 * Maybe we should finish it up, too? */
3812 else if (os
.conn
>= C_SYNC_SOURCE
&&
3813 peer_state
.conn
== C_CONNECTED
) {
3814 if (drbd_bm_total_weight(mdev
) <= mdev
->rs_failed
)
3815 drbd_resync_finished(mdev
);
3820 /* peer says his disk is inconsistent, while we think it is uptodate,
3821 * and this happens while the peer still thinks we have a sync going on,
3822 * but we think we are already done with the sync.
3823 * We ignore this to avoid flapping pdsk.
3824 * This should not happen, if the peer is a recent version of drbd. */
3825 if (os
.pdsk
== D_UP_TO_DATE
&& real_peer_disk
== D_INCONSISTENT
&&
3826 os
.conn
== C_CONNECTED
&& peer_state
.conn
> C_SYNC_SOURCE
)
3827 real_peer_disk
= D_UP_TO_DATE
;
3829 if (ns
.conn
== C_WF_REPORT_PARAMS
)
3830 ns
.conn
= C_CONNECTED
;
3832 if (peer_state
.conn
== C_AHEAD
)
3835 if (mdev
->p_uuid
&& peer_state
.disk
>= D_NEGOTIATING
&&
3836 get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3837 int cr
; /* consider resync */
3839 /* if we established a new connection */
3840 cr
= (os
.conn
< C_CONNECTED
);
3841 /* if we had an established connection
3842 * and one of the nodes newly attaches a disk */
3843 cr
|= (os
.conn
== C_CONNECTED
&&
3844 (peer_state
.disk
== D_NEGOTIATING
||
3845 os
.disk
== D_NEGOTIATING
));
3846 /* if we have both been inconsistent, and the peer has been
3847 * forced to be UpToDate with --overwrite-data */
3848 cr
|= test_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3849 /* if we had been plain connected, and the admin requested to
3850 * start a sync by "invalidate" or "invalidate-remote" */
3851 cr
|= (os
.conn
== C_CONNECTED
&&
3852 (peer_state
.conn
>= C_STARTING_SYNC_S
&&
3853 peer_state
.conn
<= C_WF_BITMAP_T
));
3856 ns
.conn
= drbd_sync_handshake(mdev
, peer_state
.role
, real_peer_disk
);
3859 if (ns
.conn
== C_MASK
) {
3860 ns
.conn
= C_CONNECTED
;
3861 if (mdev
->state
.disk
== D_NEGOTIATING
) {
3862 drbd_force_state(mdev
, NS(disk
, D_FAILED
));
3863 } else if (peer_state
.disk
== D_NEGOTIATING
) {
3864 dev_err(DEV
, "Disk attach process on the peer node was aborted.\n");
3865 peer_state
.disk
= D_DISKLESS
;
3866 real_peer_disk
= D_DISKLESS
;
3868 if (test_and_clear_bit(CONN_DRY_RUN
, &mdev
->tconn
->flags
))
3870 D_ASSERT(os
.conn
== C_WF_REPORT_PARAMS
);
3871 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3877 spin_lock_irq(&mdev
->tconn
->req_lock
);
3878 if (os
.i
!= drbd_read_state(mdev
).i
)
3880 clear_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3881 ns
.peer
= peer_state
.role
;
3882 ns
.pdsk
= real_peer_disk
;
3883 ns
.peer_isp
= (peer_state
.aftr_isp
| peer_state
.user_isp
);
3884 if ((ns
.conn
== C_CONNECTED
|| ns
.conn
== C_WF_BITMAP_S
) && ns
.disk
== D_NEGOTIATING
)
3885 ns
.disk
= mdev
->new_state_tmp
.disk
;
3886 cs_flags
= CS_VERBOSE
+ (os
.conn
< C_CONNECTED
&& ns
.conn
>= C_CONNECTED
? 0 : CS_HARD
);
3887 if (ns
.pdsk
== D_CONSISTENT
&& drbd_suspended(mdev
) && ns
.conn
== C_CONNECTED
&& os
.conn
< C_CONNECTED
&&
3888 test_bit(NEW_CUR_UUID
, &mdev
->flags
)) {
3889 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
3890 for temporal network outages! */
3891 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3892 dev_err(DEV
, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3893 tl_clear(mdev
->tconn
);
3894 drbd_uuid_new_current(mdev
);
3895 clear_bit(NEW_CUR_UUID
, &mdev
->flags
);
3896 conn_request_state(mdev
->tconn
, NS2(conn
, C_PROTOCOL_ERROR
, susp
, 0), CS_HARD
);
3899 rv
= _drbd_set_state(mdev
, ns
, cs_flags
, NULL
);
3900 ns
= drbd_read_state(mdev
);
3901 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3903 if (rv
< SS_SUCCESS
) {
3904 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3908 if (os
.conn
> C_WF_REPORT_PARAMS
) {
3909 if (ns
.conn
> C_CONNECTED
&& peer_state
.conn
<= C_CONNECTED
&&
3910 peer_state
.disk
!= D_NEGOTIATING
) {
3911 /* we want resync, peer has not yet decided to sync... */
3912 /* Nowadays only used when forcing a node into primary role and
3913 setting its disk to UpToDate with that */
3914 drbd_send_uuids(mdev
);
3915 drbd_send_current_state(mdev
);
3919 clear_bit(DISCARD_MY_DATA
, &mdev
->flags
);
3921 drbd_md_sync(mdev
); /* update connected indicator, la_size, ... */
3926 static int receive_sync_uuid(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3928 struct drbd_conf
*mdev
;
3929 struct p_rs_uuid
*p
= pi
->data
;
3931 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3935 wait_event(mdev
->misc_wait
,
3936 mdev
->state
.conn
== C_WF_SYNC_UUID
||
3937 mdev
->state
.conn
== C_BEHIND
||
3938 mdev
->state
.conn
< C_CONNECTED
||
3939 mdev
->state
.disk
< D_NEGOTIATING
);
3941 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3943 /* Here the _drbd_uuid_ functions are right, current should
3944 _not_ be rotated into the history */
3945 if (get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3946 _drbd_uuid_set(mdev
, UI_CURRENT
, be64_to_cpu(p
->uuid
));
3947 _drbd_uuid_set(mdev
, UI_BITMAP
, 0UL);
3949 drbd_print_uuids(mdev
, "updated sync uuid");
3950 drbd_start_resync(mdev
, C_SYNC_TARGET
);
3954 dev_err(DEV
, "Ignoring SyncUUID packet!\n");
3960 * receive_bitmap_plain
3962 * Return 0 when done, 1 when another iteration is needed, and a negative error
3963 * code upon failure.
3966 receive_bitmap_plain(struct drbd_conf
*mdev
, unsigned int size
,
3967 unsigned long *p
, struct bm_xfer_ctx
*c
)
3969 unsigned int data_size
= DRBD_SOCKET_BUFFER_SIZE
-
3970 drbd_header_size(mdev
->tconn
);
3971 unsigned int num_words
= min_t(size_t, data_size
/ sizeof(*p
),
3972 c
->bm_words
- c
->word_offset
);
3973 unsigned int want
= num_words
* sizeof(*p
);
3977 dev_err(DEV
, "%s:want (%u) != size (%u)\n", __func__
, want
, size
);
3982 err
= drbd_recv_all(mdev
->tconn
, p
, want
);
3986 drbd_bm_merge_lel(mdev
, c
->word_offset
, num_words
, p
);
3988 c
->word_offset
+= num_words
;
3989 c
->bit_offset
= c
->word_offset
* BITS_PER_LONG
;
3990 if (c
->bit_offset
> c
->bm_bits
)
3991 c
->bit_offset
= c
->bm_bits
;
3996 static enum drbd_bitmap_code
dcbp_get_code(struct p_compressed_bm
*p
)
3998 return (enum drbd_bitmap_code
)(p
->encoding
& 0x0f);
4001 static int dcbp_get_start(struct p_compressed_bm
*p
)
4003 return (p
->encoding
& 0x80) != 0;
4006 static int dcbp_get_pad_bits(struct p_compressed_bm
*p
)
4008 return (p
->encoding
>> 4) & 0x7;
4014 * Return 0 when done, 1 when another iteration is needed, and a negative error
4015 * code upon failure.
4018 recv_bm_rle_bits(struct drbd_conf
*mdev
,
4019 struct p_compressed_bm
*p
,
4020 struct bm_xfer_ctx
*c
,
4023 struct bitstream bs
;
4027 unsigned long s
= c
->bit_offset
;
4029 int toggle
= dcbp_get_start(p
);
4033 bitstream_init(&bs
, p
->code
, len
, dcbp_get_pad_bits(p
));
4035 bits
= bitstream_get_bits(&bs
, &look_ahead
, 64);
4039 for (have
= bits
; have
> 0; s
+= rl
, toggle
= !toggle
) {
4040 bits
= vli_decode_bits(&rl
, look_ahead
);
4046 if (e
>= c
->bm_bits
) {
4047 dev_err(DEV
, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e
);
4050 _drbd_bm_set_bits(mdev
, s
, e
);
4054 dev_err(DEV
, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4055 have
, bits
, look_ahead
,
4056 (unsigned int)(bs
.cur
.b
- p
->code
),
4057 (unsigned int)bs
.buf_len
);
4060 look_ahead
>>= bits
;
4063 bits
= bitstream_get_bits(&bs
, &tmp
, 64 - have
);
4066 look_ahead
|= tmp
<< have
;
4071 bm_xfer_ctx_bit_to_word_offset(c
);
4073 return (s
!= c
->bm_bits
);
4079 * Return 0 when done, 1 when another iteration is needed, and a negative error
4080 * code upon failure.
4083 decode_bitmap_c(struct drbd_conf
*mdev
,
4084 struct p_compressed_bm
*p
,
4085 struct bm_xfer_ctx
*c
,
4088 if (dcbp_get_code(p
) == RLE_VLI_Bits
)
4089 return recv_bm_rle_bits(mdev
, p
, c
, len
- sizeof(*p
));
4091 /* other variants had been implemented for evaluation,
4092 * but have been dropped as this one turned out to be "best"
4093 * during all our tests. */
4095 dev_err(DEV
, "receive_bitmap_c: unknown encoding %u\n", p
->encoding
);
4096 conn_request_state(mdev
->tconn
, NS(conn
, C_PROTOCOL_ERROR
), CS_HARD
);
4100 void INFO_bm_xfer_stats(struct drbd_conf
*mdev
,
4101 const char *direction
, struct bm_xfer_ctx
*c
)
4103 /* what would it take to transfer it "plaintext" */
4104 unsigned int header_size
= drbd_header_size(mdev
->tconn
);
4105 unsigned int data_size
= DRBD_SOCKET_BUFFER_SIZE
- header_size
;
4106 unsigned int plain
=
4107 header_size
* (DIV_ROUND_UP(c
->bm_words
, data_size
) + 1) +
4108 c
->bm_words
* sizeof(unsigned long);
4109 unsigned int total
= c
->bytes
[0] + c
->bytes
[1];
4112 /* total can not be zero. but just in case: */
4116 /* don't report if not compressed */
4120 /* total < plain. check for overflow, still */
4121 r
= (total
> UINT_MAX
/1000) ? (total
/ (plain
/1000))
4122 : (1000 * total
/ plain
);
4128 dev_info(DEV
, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4129 "total %u; compression: %u.%u%%\n",
4131 c
->bytes
[1], c
->packets
[1],
4132 c
->bytes
[0], c
->packets
[0],
4133 total
, r
/10, r
% 10);
4136 /* Since we are processing the bitfield from lower addresses to higher,
4137 it does not matter if the process it in 32 bit chunks or 64 bit
4138 chunks as long as it is little endian. (Understand it as byte stream,
4139 beginning with the lowest byte...) If we would use big endian
4140 we would need to process it from the highest address to the lowest,
4141 in order to be agnostic to the 32 vs 64 bits issue.
4143 returns 0 on failure, 1 if we successfully received it. */
4144 static int receive_bitmap(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4146 struct drbd_conf
*mdev
;
4147 struct bm_xfer_ctx c
;
4150 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4154 drbd_bm_lock(mdev
, "receive bitmap", BM_LOCKED_SET_ALLOWED
);
4155 /* you are supposed to send additional out-of-sync information
4156 * if you actually set bits during this phase */
4158 c
= (struct bm_xfer_ctx
) {
4159 .bm_bits
= drbd_bm_bits(mdev
),
4160 .bm_words
= drbd_bm_words(mdev
),
4164 if (pi
->cmd
== P_BITMAP
)
4165 err
= receive_bitmap_plain(mdev
, pi
->size
, pi
->data
, &c
);
4166 else if (pi
->cmd
== P_COMPRESSED_BITMAP
) {
4167 /* MAYBE: sanity check that we speak proto >= 90,
4168 * and the feature is enabled! */
4169 struct p_compressed_bm
*p
= pi
->data
;
4171 if (pi
->size
> DRBD_SOCKET_BUFFER_SIZE
- drbd_header_size(tconn
)) {
4172 dev_err(DEV
, "ReportCBitmap packet too large\n");
4176 if (pi
->size
<= sizeof(*p
)) {
4177 dev_err(DEV
, "ReportCBitmap packet too small (l:%u)\n", pi
->size
);
4181 err
= drbd_recv_all(mdev
->tconn
, p
, pi
->size
);
4184 err
= decode_bitmap_c(mdev
, p
, &c
, pi
->size
);
4186 dev_warn(DEV
, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi
->cmd
);
4191 c
.packets
[pi
->cmd
== P_BITMAP
]++;
4192 c
.bytes
[pi
->cmd
== P_BITMAP
] += drbd_header_size(tconn
) + pi
->size
;
4199 err
= drbd_recv_header(mdev
->tconn
, pi
);
4204 INFO_bm_xfer_stats(mdev
, "receive", &c
);
4206 if (mdev
->state
.conn
== C_WF_BITMAP_T
) {
4207 enum drbd_state_rv rv
;
4209 err
= drbd_send_bitmap(mdev
);
4212 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4213 rv
= _drbd_request_state(mdev
, NS(conn
, C_WF_SYNC_UUID
), CS_VERBOSE
);
4214 D_ASSERT(rv
== SS_SUCCESS
);
4215 } else if (mdev
->state
.conn
!= C_WF_BITMAP_S
) {
4216 /* admin may have requested C_DISCONNECTING,
4217 * other threads may have noticed network errors */
4218 dev_info(DEV
, "unexpected cstate (%s) in receive_bitmap\n",
4219 drbd_conn_str(mdev
->state
.conn
));
4224 drbd_bm_unlock(mdev
);
4225 if (!err
&& mdev
->state
.conn
== C_WF_BITMAP_S
)
4226 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
4230 static int receive_skip(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4232 conn_warn(tconn
, "skipping unknown optional packet type %d, l: %d!\n",
4235 return ignore_remaining_packet(tconn
, pi
);
4238 static int receive_UnplugRemote(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4240 /* Make sure we've acked all the TCP data associated
4241 * with the data requests being unplugged */
4242 drbd_tcp_quickack(tconn
->data
.socket
);
4247 static int receive_out_of_sync(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4249 struct drbd_conf
*mdev
;
4250 struct p_block_desc
*p
= pi
->data
;
4252 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4256 switch (mdev
->state
.conn
) {
4257 case C_WF_SYNC_UUID
:
4262 dev_err(DEV
, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4263 drbd_conn_str(mdev
->state
.conn
));
4266 drbd_set_out_of_sync(mdev
, be64_to_cpu(p
->sector
), be32_to_cpu(p
->blksize
));
4274 int (*fn
)(struct drbd_tconn
*, struct packet_info
*);
4277 static struct data_cmd drbd_cmd_handler
[] = {
4278 [P_DATA
] = { 1, sizeof(struct p_data
), receive_Data
},
4279 [P_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_DataReply
},
4280 [P_RS_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_RSDataReply
} ,
4281 [P_BARRIER
] = { 0, sizeof(struct p_barrier
), receive_Barrier
} ,
4282 [P_BITMAP
] = { 1, 0, receive_bitmap
} ,
4283 [P_COMPRESSED_BITMAP
] = { 1, 0, receive_bitmap
} ,
4284 [P_UNPLUG_REMOTE
] = { 0, 0, receive_UnplugRemote
},
4285 [P_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4286 [P_RS_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4287 [P_SYNC_PARAM
] = { 1, 0, receive_SyncParam
},
4288 [P_SYNC_PARAM89
] = { 1, 0, receive_SyncParam
},
4289 [P_PROTOCOL
] = { 1, sizeof(struct p_protocol
), receive_protocol
},
4290 [P_UUIDS
] = { 0, sizeof(struct p_uuids
), receive_uuids
},
4291 [P_SIZES
] = { 0, sizeof(struct p_sizes
), receive_sizes
},
4292 [P_STATE
] = { 0, sizeof(struct p_state
), receive_state
},
4293 [P_STATE_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_state
},
4294 [P_SYNC_UUID
] = { 0, sizeof(struct p_rs_uuid
), receive_sync_uuid
},
4295 [P_OV_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4296 [P_OV_REPLY
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
4297 [P_CSUM_RS_REQUEST
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
4298 [P_DELAY_PROBE
] = { 0, sizeof(struct p_delay_probe93
), receive_skip
},
4299 [P_OUT_OF_SYNC
] = { 0, sizeof(struct p_block_desc
), receive_out_of_sync
},
4300 [P_CONN_ST_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_conn_state
},
4301 [P_PROTOCOL_UPDATE
] = { 1, sizeof(struct p_protocol
), receive_protocol
},
4304 static void drbdd(struct drbd_tconn
*tconn
)
4306 struct packet_info pi
;
4307 size_t shs
; /* sub header size */
4310 while (get_t_state(&tconn
->receiver
) == RUNNING
) {
4311 struct data_cmd
*cmd
;
4313 drbd_thread_current_set_cpu(&tconn
->receiver
);
4314 if (drbd_recv_header(tconn
, &pi
))
4317 cmd
= &drbd_cmd_handler
[pi
.cmd
];
4318 if (unlikely(pi
.cmd
>= ARRAY_SIZE(drbd_cmd_handler
) || !cmd
->fn
)) {
4319 conn_err(tconn
, "Unexpected data packet %s (0x%04x)",
4320 cmdname(pi
.cmd
), pi
.cmd
);
4324 shs
= cmd
->pkt_size
;
4325 if (pi
.size
> shs
&& !cmd
->expect_payload
) {
4326 conn_err(tconn
, "No payload expected %s l:%d\n",
4327 cmdname(pi
.cmd
), pi
.size
);
4332 err
= drbd_recv_all_warn(tconn
, pi
.data
, shs
);
4338 err
= cmd
->fn(tconn
, &pi
);
4340 conn_err(tconn
, "error receiving %s, e: %d l: %d!\n",
4341 cmdname(pi
.cmd
), err
, pi
.size
);
4348 conn_request_state(tconn
, NS(conn
, C_PROTOCOL_ERROR
), CS_HARD
);
4351 void conn_flush_workqueue(struct drbd_tconn
*tconn
)
4353 struct drbd_wq_barrier barr
;
4355 barr
.w
.cb
= w_prev_work_done
;
4356 barr
.w
.tconn
= tconn
;
4357 init_completion(&barr
.done
);
4358 drbd_queue_work(&tconn
->data
.work
, &barr
.w
);
4359 wait_for_completion(&barr
.done
);
4362 static void conn_disconnect(struct drbd_tconn
*tconn
)
4364 struct drbd_conf
*mdev
;
4368 if (tconn
->cstate
== C_STANDALONE
)
4371 /* asender does not clean up anything. it must not interfere, either */
4372 drbd_thread_stop(&tconn
->asender
);
4373 drbd_free_sock(tconn
);
4376 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
4377 kref_get(&mdev
->kref
);
4379 drbd_disconnected(mdev
);
4380 kref_put(&mdev
->kref
, &drbd_minor_destroy
);
4385 if (!list_empty(&tconn
->current_epoch
->list
))
4386 conn_err(tconn
, "ASSERTION FAILED: tconn->current_epoch->list not empty\n");
4387 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
4388 atomic_set(&tconn
->current_epoch
->epoch_size
, 0);
4390 conn_info(tconn
, "Connection closed\n");
4392 if (conn_highest_role(tconn
) == R_PRIMARY
&& conn_highest_pdsk(tconn
) >= D_UNKNOWN
)
4393 conn_try_outdate_peer_async(tconn
);
4395 spin_lock_irq(&tconn
->req_lock
);
4397 if (oc
>= C_UNCONNECTED
)
4398 _conn_request_state(tconn
, NS(conn
, C_UNCONNECTED
), CS_VERBOSE
);
4400 spin_unlock_irq(&tconn
->req_lock
);
4402 if (oc
== C_DISCONNECTING
)
4403 conn_request_state(tconn
, NS(conn
, C_STANDALONE
), CS_VERBOSE
| CS_HARD
);
4406 static int drbd_disconnected(struct drbd_conf
*mdev
)
4410 /* wait for current activity to cease. */
4411 spin_lock_irq(&mdev
->tconn
->req_lock
);
4412 _drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
4413 _drbd_wait_ee_list_empty(mdev
, &mdev
->sync_ee
);
4414 _drbd_wait_ee_list_empty(mdev
, &mdev
->read_ee
);
4415 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4417 /* We do not have data structures that would allow us to
4418 * get the rs_pending_cnt down to 0 again.
4419 * * On C_SYNC_TARGET we do not have any data structures describing
4420 * the pending RSDataRequest's we have sent.
4421 * * On C_SYNC_SOURCE there is no data structure that tracks
4422 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
4423 * And no, it is not the sum of the reference counts in the
4424 * resync_LRU. The resync_LRU tracks the whole operation including
4425 * the disk-IO, while the rs_pending_cnt only tracks the blocks
4427 drbd_rs_cancel_all(mdev
);
4429 mdev
->rs_failed
= 0;
4430 atomic_set(&mdev
->rs_pending_cnt
, 0);
4431 wake_up(&mdev
->misc_wait
);
4433 del_timer_sync(&mdev
->resync_timer
);
4434 resync_timer_fn((unsigned long)mdev
);
4436 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
4437 * w_make_resync_request etc. which may still be on the worker queue
4438 * to be "canceled" */
4439 drbd_flush_workqueue(mdev
);
4441 drbd_finish_peer_reqs(mdev
);
4443 /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
4444 might have issued a work again. The one before drbd_finish_peer_reqs() is
4445 necessary to reclain net_ee in drbd_finish_peer_reqs(). */
4446 drbd_flush_workqueue(mdev
);
4448 kfree(mdev
->p_uuid
);
4449 mdev
->p_uuid
= NULL
;
4451 if (!drbd_suspended(mdev
))
4452 tl_clear(mdev
->tconn
);
4456 /* serialize with bitmap writeout triggered by the state change,
4458 wait_event(mdev
->misc_wait
, !test_bit(BITMAP_IO
, &mdev
->flags
));
4460 /* tcp_close and release of sendpage pages can be deferred. I don't
4461 * want to use SO_LINGER, because apparently it can be deferred for
4462 * more than 20 seconds (longest time I checked).
4464 * Actually we don't care for exactly when the network stack does its
4465 * put_page(), but release our reference on these pages right here.
4467 i
= drbd_free_peer_reqs(mdev
, &mdev
->net_ee
);
4469 dev_info(DEV
, "net_ee not empty, killed %u entries\n", i
);
4470 i
= atomic_read(&mdev
->pp_in_use_by_net
);
4472 dev_info(DEV
, "pp_in_use_by_net = %d, expected 0\n", i
);
4473 i
= atomic_read(&mdev
->pp_in_use
);
4475 dev_info(DEV
, "pp_in_use = %d, expected 0\n", i
);
4477 D_ASSERT(list_empty(&mdev
->read_ee
));
4478 D_ASSERT(list_empty(&mdev
->active_ee
));
4479 D_ASSERT(list_empty(&mdev
->sync_ee
));
4480 D_ASSERT(list_empty(&mdev
->done_ee
));
4486 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
4487 * we can agree on is stored in agreed_pro_version.
4489 * feature flags and the reserved array should be enough room for future
4490 * enhancements of the handshake protocol, and possible plugins...
4492 * for now, they are expected to be zero, but ignored.
4494 static int drbd_send_features(struct drbd_tconn
*tconn
)
4496 struct drbd_socket
*sock
;
4497 struct p_connection_features
*p
;
4499 sock
= &tconn
->data
;
4500 p
= conn_prepare_command(tconn
, sock
);
4503 memset(p
, 0, sizeof(*p
));
4504 p
->protocol_min
= cpu_to_be32(PRO_VERSION_MIN
);
4505 p
->protocol_max
= cpu_to_be32(PRO_VERSION_MAX
);
4506 return conn_send_command(tconn
, sock
, P_CONNECTION_FEATURES
, sizeof(*p
), NULL
, 0);
4511 * 1 yes, we have a valid connection
4512 * 0 oops, did not work out, please try again
4513 * -1 peer talks different language,
4514 * no point in trying again, please go standalone.
4516 static int drbd_do_features(struct drbd_tconn
*tconn
)
4518 /* ASSERT current == tconn->receiver ... */
4519 struct p_connection_features
*p
;
4520 const int expect
= sizeof(struct p_connection_features
);
4521 struct packet_info pi
;
4524 err
= drbd_send_features(tconn
);
4528 err
= drbd_recv_header(tconn
, &pi
);
4532 if (pi
.cmd
!= P_CONNECTION_FEATURES
) {
4533 conn_err(tconn
, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
4534 cmdname(pi
.cmd
), pi
.cmd
);
4538 if (pi
.size
!= expect
) {
4539 conn_err(tconn
, "expected ConnectionFeatures length: %u, received: %u\n",
4545 err
= drbd_recv_all_warn(tconn
, p
, expect
);
4549 p
->protocol_min
= be32_to_cpu(p
->protocol_min
);
4550 p
->protocol_max
= be32_to_cpu(p
->protocol_max
);
4551 if (p
->protocol_max
== 0)
4552 p
->protocol_max
= p
->protocol_min
;
4554 if (PRO_VERSION_MAX
< p
->protocol_min
||
4555 PRO_VERSION_MIN
> p
->protocol_max
)
4558 tconn
->agreed_pro_version
= min_t(int, PRO_VERSION_MAX
, p
->protocol_max
);
4560 conn_info(tconn
, "Handshake successful: "
4561 "Agreed network protocol version %d\n", tconn
->agreed_pro_version
);
4566 conn_err(tconn
, "incompatible DRBD dialects: "
4567 "I support %d-%d, peer supports %d-%d\n",
4568 PRO_VERSION_MIN
, PRO_VERSION_MAX
,
4569 p
->protocol_min
, p
->protocol_max
);
4573 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4574 static int drbd_do_auth(struct drbd_tconn
*tconn
)
4576 dev_err(DEV
, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4577 dev_err(DEV
, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4581 #define CHALLENGE_LEN 64
4585 0 - failed, try again (network error),
4586 -1 - auth failed, don't try again.
4589 static int drbd_do_auth(struct drbd_tconn
*tconn
)
4591 struct drbd_socket
*sock
;
4592 char my_challenge
[CHALLENGE_LEN
]; /* 64 Bytes... */
4593 struct scatterlist sg
;
4594 char *response
= NULL
;
4595 char *right_response
= NULL
;
4596 char *peers_ch
= NULL
;
4597 unsigned int key_len
;
4598 char secret
[SHARED_SECRET_MAX
]; /* 64 byte */
4599 unsigned int resp_size
;
4600 struct hash_desc desc
;
4601 struct packet_info pi
;
4602 struct net_conf
*nc
;
4605 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
4608 nc
= rcu_dereference(tconn
->net_conf
);
4609 key_len
= strlen(nc
->shared_secret
);
4610 memcpy(secret
, nc
->shared_secret
, key_len
);
4613 desc
.tfm
= tconn
->cram_hmac_tfm
;
4616 rv
= crypto_hash_setkey(tconn
->cram_hmac_tfm
, (u8
*)secret
, key_len
);
4618 conn_err(tconn
, "crypto_hash_setkey() failed with %d\n", rv
);
4623 get_random_bytes(my_challenge
, CHALLENGE_LEN
);
4625 sock
= &tconn
->data
;
4626 if (!conn_prepare_command(tconn
, sock
)) {
4630 rv
= !conn_send_command(tconn
, sock
, P_AUTH_CHALLENGE
, 0,
4631 my_challenge
, CHALLENGE_LEN
);
4635 err
= drbd_recv_header(tconn
, &pi
);
4641 if (pi
.cmd
!= P_AUTH_CHALLENGE
) {
4642 conn_err(tconn
, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4643 cmdname(pi
.cmd
), pi
.cmd
);
4648 if (pi
.size
> CHALLENGE_LEN
* 2) {
4649 conn_err(tconn
, "expected AuthChallenge payload too big.\n");
4654 peers_ch
= kmalloc(pi
.size
, GFP_NOIO
);
4655 if (peers_ch
== NULL
) {
4656 conn_err(tconn
, "kmalloc of peers_ch failed\n");
4661 err
= drbd_recv_all_warn(tconn
, peers_ch
, pi
.size
);
4667 resp_size
= crypto_hash_digestsize(tconn
->cram_hmac_tfm
);
4668 response
= kmalloc(resp_size
, GFP_NOIO
);
4669 if (response
== NULL
) {
4670 conn_err(tconn
, "kmalloc of response failed\n");
4675 sg_init_table(&sg
, 1);
4676 sg_set_buf(&sg
, peers_ch
, pi
.size
);
4678 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, response
);
4680 conn_err(tconn
, "crypto_hash_digest() failed with %d\n", rv
);
4685 if (!conn_prepare_command(tconn
, sock
)) {
4689 rv
= !conn_send_command(tconn
, sock
, P_AUTH_RESPONSE
, 0,
4690 response
, resp_size
);
4694 err
= drbd_recv_header(tconn
, &pi
);
4700 if (pi
.cmd
!= P_AUTH_RESPONSE
) {
4701 conn_err(tconn
, "expected AuthResponse packet, received: %s (0x%04x)\n",
4702 cmdname(pi
.cmd
), pi
.cmd
);
4707 if (pi
.size
!= resp_size
) {
4708 conn_err(tconn
, "expected AuthResponse payload of wrong size\n");
4713 err
= drbd_recv_all_warn(tconn
, response
, resp_size
);
4719 right_response
= kmalloc(resp_size
, GFP_NOIO
);
4720 if (right_response
== NULL
) {
4721 conn_err(tconn
, "kmalloc of right_response failed\n");
4726 sg_set_buf(&sg
, my_challenge
, CHALLENGE_LEN
);
4728 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, right_response
);
4730 conn_err(tconn
, "crypto_hash_digest() failed with %d\n", rv
);
4735 rv
= !memcmp(response
, right_response
, resp_size
);
4738 conn_info(tconn
, "Peer authenticated using %d bytes HMAC\n",
4746 kfree(right_response
);
4752 int drbdd_init(struct drbd_thread
*thi
)
4754 struct drbd_tconn
*tconn
= thi
->tconn
;
4757 conn_info(tconn
, "receiver (re)started\n");
4760 h
= conn_connect(tconn
);
4762 conn_disconnect(tconn
);
4763 schedule_timeout_interruptible(HZ
);
4766 conn_warn(tconn
, "Discarding network configuration.\n");
4767 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
4774 conn_disconnect(tconn
);
4776 conn_info(tconn
, "receiver terminated\n");
4780 /* ********* acknowledge sender ******** */
4782 static int got_conn_RqSReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4784 struct p_req_state_reply
*p
= pi
->data
;
4785 int retcode
= be32_to_cpu(p
->retcode
);
4787 if (retcode
>= SS_SUCCESS
) {
4788 set_bit(CONN_WD_ST_CHG_OKAY
, &tconn
->flags
);
4790 set_bit(CONN_WD_ST_CHG_FAIL
, &tconn
->flags
);
4791 conn_err(tconn
, "Requested state change failed by peer: %s (%d)\n",
4792 drbd_set_st_err_str(retcode
), retcode
);
4794 wake_up(&tconn
->ping_wait
);
4799 static int got_RqSReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4801 struct drbd_conf
*mdev
;
4802 struct p_req_state_reply
*p
= pi
->data
;
4803 int retcode
= be32_to_cpu(p
->retcode
);
4805 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4809 if (retcode
>= SS_SUCCESS
) {
4810 set_bit(CL_ST_CHG_SUCCESS
, &mdev
->flags
);
4812 set_bit(CL_ST_CHG_FAIL
, &mdev
->flags
);
4813 dev_err(DEV
, "Requested state change failed by peer: %s (%d)\n",
4814 drbd_set_st_err_str(retcode
), retcode
);
4816 wake_up(&mdev
->state_wait
);
4821 static int got_Ping(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4823 return drbd_send_ping_ack(tconn
);
4827 static int got_PingAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4829 /* restore idle timeout */
4830 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= tconn
->net_conf
->ping_int
*HZ
;
4831 if (!test_and_set_bit(GOT_PING_ACK
, &tconn
->flags
))
4832 wake_up(&tconn
->ping_wait
);
4837 static int got_IsInSync(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4839 struct drbd_conf
*mdev
;
4840 struct p_block_ack
*p
= pi
->data
;
4841 sector_t sector
= be64_to_cpu(p
->sector
);
4842 int blksize
= be32_to_cpu(p
->blksize
);
4844 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4848 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
4850 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4852 if (get_ldev(mdev
)) {
4853 drbd_rs_complete_io(mdev
, sector
);
4854 drbd_set_in_sync(mdev
, sector
, blksize
);
4855 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4856 mdev
->rs_same_csum
+= (blksize
>> BM_BLOCK_SHIFT
);
4859 dec_rs_pending(mdev
);
4860 atomic_add(blksize
>> 9, &mdev
->rs_sect_in
);
4866 validate_req_change_req_state(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
4867 struct rb_root
*root
, const char *func
,
4868 enum drbd_req_event what
, bool missing_ok
)
4870 struct drbd_request
*req
;
4871 struct bio_and_error m
;
4873 spin_lock_irq(&mdev
->tconn
->req_lock
);
4874 req
= find_request(mdev
, root
, id
, sector
, missing_ok
, func
);
4875 if (unlikely(!req
)) {
4876 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4879 __req_mod(req
, what
, &m
);
4880 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4883 complete_master_bio(mdev
, &m
);
4887 static int got_BlockAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4889 struct drbd_conf
*mdev
;
4890 struct p_block_ack
*p
= pi
->data
;
4891 sector_t sector
= be64_to_cpu(p
->sector
);
4892 int blksize
= be32_to_cpu(p
->blksize
);
4893 enum drbd_req_event what
;
4895 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4899 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4901 if (p
->block_id
== ID_SYNCER
) {
4902 drbd_set_in_sync(mdev
, sector
, blksize
);
4903 dec_rs_pending(mdev
);
4907 case P_RS_WRITE_ACK
:
4908 what
= WRITE_ACKED_BY_PEER_AND_SIS
;
4911 what
= WRITE_ACKED_BY_PEER
;
4914 what
= RECV_ACKED_BY_PEER
;
4916 case P_DISCARD_WRITE
:
4917 what
= DISCARD_WRITE
;
4920 what
= POSTPONE_WRITE
;
4926 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4927 &mdev
->write_requests
, __func__
,
4931 static int got_NegAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4933 struct drbd_conf
*mdev
;
4934 struct p_block_ack
*p
= pi
->data
;
4935 sector_t sector
= be64_to_cpu(p
->sector
);
4936 int size
= be32_to_cpu(p
->blksize
);
4939 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4943 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4945 if (p
->block_id
== ID_SYNCER
) {
4946 dec_rs_pending(mdev
);
4947 drbd_rs_failed_io(mdev
, sector
, size
);
4951 err
= validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4952 &mdev
->write_requests
, __func__
,
4955 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
4956 The master bio might already be completed, therefore the
4957 request is no longer in the collision hash. */
4958 /* In Protocol B we might already have got a P_RECV_ACK
4959 but then get a P_NEG_ACK afterwards. */
4960 drbd_set_out_of_sync(mdev
, sector
, size
);
4965 static int got_NegDReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4967 struct drbd_conf
*mdev
;
4968 struct p_block_ack
*p
= pi
->data
;
4969 sector_t sector
= be64_to_cpu(p
->sector
);
4971 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4975 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4977 dev_err(DEV
, "Got NegDReply; Sector %llus, len %u.\n",
4978 (unsigned long long)sector
, be32_to_cpu(p
->blksize
));
4980 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4981 &mdev
->read_requests
, __func__
,
4985 static int got_NegRSDReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4987 struct drbd_conf
*mdev
;
4990 struct p_block_ack
*p
= pi
->data
;
4992 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4996 sector
= be64_to_cpu(p
->sector
);
4997 size
= be32_to_cpu(p
->blksize
);
4999 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
5001 dec_rs_pending(mdev
);
5003 if (get_ldev_if_state(mdev
, D_FAILED
)) {
5004 drbd_rs_complete_io(mdev
, sector
);
5006 case P_NEG_RS_DREPLY
:
5007 drbd_rs_failed_io(mdev
, sector
, size
);
5019 static int got_BarrierAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
5021 struct drbd_conf
*mdev
;
5022 struct p_barrier_ack
*p
= pi
->data
;
5024 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
5028 tl_release(mdev
->tconn
, p
->barrier
, be32_to_cpu(p
->set_size
));
5030 if (mdev
->state
.conn
== C_AHEAD
&&
5031 atomic_read(&mdev
->ap_in_flight
) == 0 &&
5032 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE
, &mdev
->flags
)) {
5033 mdev
->start_resync_timer
.expires
= jiffies
+ HZ
;
5034 add_timer(&mdev
->start_resync_timer
);
5040 static int got_OVResult(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
5042 struct drbd_conf
*mdev
;
5043 struct p_block_ack
*p
= pi
->data
;
5044 struct drbd_work
*w
;
5048 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
5052 sector
= be64_to_cpu(p
->sector
);
5053 size
= be32_to_cpu(p
->blksize
);
5055 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
5057 if (be64_to_cpu(p
->block_id
) == ID_OUT_OF_SYNC
)
5058 drbd_ov_out_of_sync_found(mdev
, sector
, size
);
5060 ov_out_of_sync_print(mdev
);
5062 if (!get_ldev(mdev
))
5065 drbd_rs_complete_io(mdev
, sector
);
5066 dec_rs_pending(mdev
);
5070 /* let's advance progress step marks only for every other megabyte */
5071 if ((mdev
->ov_left
& 0x200) == 0x200)
5072 drbd_advance_rs_marks(mdev
, mdev
->ov_left
);
5074 if (mdev
->ov_left
== 0) {
5075 w
= kmalloc(sizeof(*w
), GFP_NOIO
);
5077 w
->cb
= w_ov_finished
;
5079 drbd_queue_work_front(&mdev
->tconn
->data
.work
, w
);
5081 dev_err(DEV
, "kmalloc(w) failed.");
5082 ov_out_of_sync_print(mdev
);
5083 drbd_resync_finished(mdev
);
5090 static int got_skip(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
5095 static int tconn_finish_peer_reqs(struct drbd_tconn
*tconn
)
5097 struct drbd_conf
*mdev
;
5098 int vnr
, not_empty
= 0;
5101 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5102 flush_signals(current
);
5105 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
5106 kref_get(&mdev
->kref
);
5108 if (drbd_finish_peer_reqs(mdev
)) {
5109 kref_put(&mdev
->kref
, &drbd_minor_destroy
);
5112 kref_put(&mdev
->kref
, &drbd_minor_destroy
);
5115 set_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5117 spin_lock_irq(&tconn
->req_lock
);
5118 idr_for_each_entry(&tconn
->volumes
, mdev
, vnr
) {
5119 not_empty
= !list_empty(&mdev
->done_ee
);
5123 spin_unlock_irq(&tconn
->req_lock
);
5125 } while (not_empty
);
5130 struct asender_cmd
{
5132 int (*fn
)(struct drbd_tconn
*tconn
, struct packet_info
*);
5135 static struct asender_cmd asender_tbl
[] = {
5136 [P_PING
] = { 0, got_Ping
},
5137 [P_PING_ACK
] = { 0, got_PingAck
},
5138 [P_RECV_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5139 [P_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5140 [P_RS_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5141 [P_DISCARD_WRITE
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5142 [P_NEG_ACK
] = { sizeof(struct p_block_ack
), got_NegAck
},
5143 [P_NEG_DREPLY
] = { sizeof(struct p_block_ack
), got_NegDReply
},
5144 [P_NEG_RS_DREPLY
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
5145 [P_OV_RESULT
] = { sizeof(struct p_block_ack
), got_OVResult
},
5146 [P_BARRIER_ACK
] = { sizeof(struct p_barrier_ack
), got_BarrierAck
},
5147 [P_STATE_CHG_REPLY
] = { sizeof(struct p_req_state_reply
), got_RqSReply
},
5148 [P_RS_IS_IN_SYNC
] = { sizeof(struct p_block_ack
), got_IsInSync
},
5149 [P_DELAY_PROBE
] = { sizeof(struct p_delay_probe93
), got_skip
},
5150 [P_RS_CANCEL
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
5151 [P_CONN_ST_CHG_REPLY
]={ sizeof(struct p_req_state_reply
), got_conn_RqSReply
},
5152 [P_RETRY_WRITE
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5155 int drbd_asender(struct drbd_thread
*thi
)
5157 struct drbd_tconn
*tconn
= thi
->tconn
;
5158 struct asender_cmd
*cmd
= NULL
;
5159 struct packet_info pi
;
5161 void *buf
= tconn
->meta
.rbuf
;
5163 unsigned int header_size
= drbd_header_size(tconn
);
5164 int expect
= header_size
;
5165 bool ping_timeout_active
= false;
5166 struct net_conf
*nc
;
5167 int ping_timeo
, tcp_cork
, ping_int
;
5169 current
->policy
= SCHED_RR
; /* Make this a realtime task! */
5170 current
->rt_priority
= 2; /* more important than all other tasks */
5172 while (get_t_state(thi
) == RUNNING
) {
5173 drbd_thread_current_set_cpu(thi
);
5176 nc
= rcu_dereference(tconn
->net_conf
);
5177 ping_timeo
= nc
->ping_timeo
;
5178 tcp_cork
= nc
->tcp_cork
;
5179 ping_int
= nc
->ping_int
;
5182 if (test_and_clear_bit(SEND_PING
, &tconn
->flags
)) {
5183 if (drbd_send_ping(tconn
)) {
5184 conn_err(tconn
, "drbd_send_ping has failed\n");
5187 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= ping_timeo
* HZ
/ 10;
5188 ping_timeout_active
= true;
5191 /* TODO: conditionally cork; it may hurt latency if we cork without
5194 drbd_tcp_cork(tconn
->meta
.socket
);
5195 if (tconn_finish_peer_reqs(tconn
)) {
5196 conn_err(tconn
, "tconn_finish_peer_reqs() failed\n");
5199 /* but unconditionally uncork unless disabled */
5201 drbd_tcp_uncork(tconn
->meta
.socket
);
5203 /* short circuit, recv_msg would return EINTR anyways. */
5204 if (signal_pending(current
))
5207 rv
= drbd_recv_short(tconn
->meta
.socket
, buf
, expect
-received
, 0);
5208 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5210 flush_signals(current
);
5213 * -EINTR (on meta) we got a signal
5214 * -EAGAIN (on meta) rcvtimeo expired
5215 * -ECONNRESET other side closed the connection
5216 * -ERESTARTSYS (on data) we got a signal
5217 * rv < 0 other than above: unexpected error!
5218 * rv == expected: full header or command
5219 * rv < expected: "woken" by signal during receive
5220 * rv == 0 : "connection shut down by peer"
5222 if (likely(rv
> 0)) {
5225 } else if (rv
== 0) {
5226 conn_err(tconn
, "meta connection shut down by peer.\n");
5228 } else if (rv
== -EAGAIN
) {
5229 /* If the data socket received something meanwhile,
5230 * that is good enough: peer is still alive. */
5231 if (time_after(tconn
->last_received
,
5232 jiffies
- tconn
->meta
.socket
->sk
->sk_rcvtimeo
))
5234 if (ping_timeout_active
) {
5235 conn_err(tconn
, "PingAck did not arrive in time.\n");
5238 set_bit(SEND_PING
, &tconn
->flags
);
5240 } else if (rv
== -EINTR
) {
5243 conn_err(tconn
, "sock_recvmsg returned %d\n", rv
);
5247 if (received
== expect
&& cmd
== NULL
) {
5248 if (decode_header(tconn
, tconn
->meta
.rbuf
, &pi
))
5250 cmd
= &asender_tbl
[pi
.cmd
];
5251 if (pi
.cmd
>= ARRAY_SIZE(asender_tbl
) || !cmd
->fn
) {
5252 conn_err(tconn
, "Unexpected meta packet %s (0x%04x)\n",
5253 cmdname(pi
.cmd
), pi
.cmd
);
5256 expect
= header_size
+ cmd
->pkt_size
;
5257 if (pi
.size
!= expect
- header_size
) {
5258 conn_err(tconn
, "Wrong packet size on meta (c: %d, l: %d)\n",
5263 if (received
== expect
) {
5266 err
= cmd
->fn(tconn
, &pi
);
5268 conn_err(tconn
, "%pf failed\n", cmd
->fn
);
5272 tconn
->last_received
= jiffies
;
5274 if (cmd
== &asender_tbl
[P_PING_ACK
]) {
5275 /* restore idle timeout */
5276 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= ping_int
* HZ
;
5277 ping_timeout_active
= false;
5280 buf
= tconn
->meta
.rbuf
;
5282 expect
= header_size
;
5289 conn_request_state(tconn
, NS(conn
, C_NETWORK_FAILURE
), CS_HARD
);
5293 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
5295 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5297 conn_info(tconn
, "asender terminated\n");