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(int vnr
, void *p
, void *data
);
68 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*, 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
, try_connect_int
;
621 int disconnect_on_error
= 1;
624 nc
= rcu_dereference(tconn
->net_conf
);
630 sndbuf_size
= nc
->sndbuf_size
;
631 rcvbuf_size
= nc
->rcvbuf_size
;
632 try_connect_int
= nc
->try_connect_int
;
634 my_addr_len
= min_t(int, nc
->my_addr_len
, sizeof(src_in6
));
635 memcpy(&src_in6
, nc
->my_addr
, my_addr_len
);
637 if (((struct sockaddr
*)nc
->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, nc
->peer_addr_len
, sizeof(src_in6
));
643 memcpy(&peer_in6
, nc
->peer_addr
, peer_addr_len
);
647 what
= "sock_create_kern";
648 err
= sock_create_kern(((struct sockaddr
*)&src_in6
)->sa_family
,
649 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
655 sock
->sk
->sk_rcvtimeo
=
656 sock
->sk
->sk_sndtimeo
= try_connect_int
* HZ
;
657 drbd_setbufsize(sock
, sndbuf_size
, rcvbuf_size
);
659 /* explicitly bind to the configured IP as source IP
660 * for the outgoing connections.
661 * This is needed for multihomed hosts and to be
662 * able to use lo: interfaces for drbd.
663 * Make sure to use 0 as port number, so linux selects
664 * a free one dynamically.
666 what
= "bind before connect";
667 err
= sock
->ops
->bind(sock
, (struct sockaddr
*) &src_in6
, my_addr_len
);
671 /* connect may fail, peer not yet available.
672 * stay C_WF_CONNECTION, don't go Disconnecting! */
673 disconnect_on_error
= 0;
675 err
= sock
->ops
->connect(sock
, (struct sockaddr
*) &peer_in6
, peer_addr_len
, 0);
684 /* timeout, busy, signal pending */
685 case ETIMEDOUT
: case EAGAIN
: case EINPROGRESS
:
686 case EINTR
: case ERESTARTSYS
:
687 /* peer not (yet) available, network problem */
688 case ECONNREFUSED
: case ENETUNREACH
:
689 case EHOSTDOWN
: case EHOSTUNREACH
:
690 disconnect_on_error
= 0;
693 conn_err(tconn
, "%s failed, err = %d\n", what
, err
);
695 if (disconnect_on_error
)
696 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
702 static struct socket
*drbd_wait_for_connect(struct drbd_tconn
*tconn
)
704 int timeo
, err
, my_addr_len
;
705 int sndbuf_size
, rcvbuf_size
, try_connect_int
;
706 struct socket
*s_estab
= NULL
, *s_listen
;
707 struct sockaddr_in6 my_addr
;
712 nc
= rcu_dereference(tconn
->net_conf
);
718 sndbuf_size
= nc
->sndbuf_size
;
719 rcvbuf_size
= nc
->rcvbuf_size
;
720 try_connect_int
= nc
->try_connect_int
;
722 my_addr_len
= min_t(int, nc
->my_addr_len
, sizeof(struct sockaddr_in6
));
723 memcpy(&my_addr
, nc
->my_addr
, my_addr_len
);
726 what
= "sock_create_kern";
727 err
= sock_create_kern(((struct sockaddr
*)&my_addr
)->sa_family
,
728 SOCK_STREAM
, IPPROTO_TCP
, &s_listen
);
734 timeo
= try_connect_int
* HZ
;
735 timeo
+= (random32() & 1) ? timeo
/ 7 : -timeo
/ 7; /* 28.5% random jitter */
737 s_listen
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
738 s_listen
->sk
->sk_rcvtimeo
= timeo
;
739 s_listen
->sk
->sk_sndtimeo
= timeo
;
740 drbd_setbufsize(s_listen
, sndbuf_size
, rcvbuf_size
);
742 what
= "bind before listen";
743 err
= s_listen
->ops
->bind(s_listen
, (struct sockaddr
*)&my_addr
, my_addr_len
);
747 err
= drbd_accept(&what
, s_listen
, &s_estab
);
751 sock_release(s_listen
);
753 if (err
!= -EAGAIN
&& err
!= -EINTR
&& err
!= -ERESTARTSYS
) {
754 conn_err(tconn
, "%s failed, err = %d\n", what
, err
);
755 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
762 static int decode_header(struct drbd_tconn
*, void *, struct packet_info
*);
764 static int send_first_packet(struct drbd_tconn
*tconn
, struct drbd_socket
*sock
,
765 enum drbd_packet cmd
)
767 if (!conn_prepare_command(tconn
, sock
))
769 return conn_send_command(tconn
, sock
, cmd
, 0, NULL
, 0);
772 static int receive_first_packet(struct drbd_tconn
*tconn
, struct socket
*sock
)
774 unsigned int header_size
= drbd_header_size(tconn
);
775 struct packet_info pi
;
778 err
= drbd_recv_short(sock
, tconn
->data
.rbuf
, header_size
, 0);
779 if (err
!= header_size
) {
784 err
= decode_header(tconn
, tconn
->data
.rbuf
, &pi
);
791 * drbd_socket_okay() - Free the socket if its connection is not okay
792 * @sock: pointer to the pointer to the socket.
794 static int drbd_socket_okay(struct socket
**sock
)
802 rr
= drbd_recv_short(*sock
, tb
, 4, MSG_DONTWAIT
| MSG_PEEK
);
804 if (rr
> 0 || rr
== -EAGAIN
) {
812 /* Gets called if a connection is established, or if a new minor gets created
814 int drbd_connected(int vnr
, void *p
, void *data
)
816 struct drbd_conf
*mdev
= (struct drbd_conf
*)p
;
819 atomic_set(&mdev
->packet_seq
, 0);
822 mdev
->state_mutex
= mdev
->tconn
->agreed_pro_version
< 100 ?
823 &mdev
->tconn
->cstate_mutex
:
824 &mdev
->own_state_mutex
;
826 err
= drbd_send_sync_param(mdev
);
828 err
= drbd_send_sizes(mdev
, 0, 0);
830 err
= drbd_send_uuids(mdev
);
832 err
= drbd_send_state(mdev
);
833 clear_bit(USE_DEGR_WFC_T
, &mdev
->flags
);
834 clear_bit(RESIZE_PENDING
, &mdev
->flags
);
835 mod_timer(&mdev
->request_timer
, jiffies
+ HZ
); /* just start it here. */
841 * 1 yes, we have a valid connection
842 * 0 oops, did not work out, please try again
843 * -1 peer talks different language,
844 * no point in trying again, please go standalone.
845 * -2 We do not have a network config...
847 static int drbd_connect(struct drbd_tconn
*tconn
)
849 struct socket
*sock
, *msock
;
851 int timeout
, try, h
, ok
;
853 if (conn_request_state(tconn
, NS(conn
, C_WF_CONNECTION
), CS_VERBOSE
) < SS_SUCCESS
)
856 clear_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
858 /* Assume that the peer only understands protocol 80 until we know better. */
859 tconn
->agreed_pro_version
= 80;
865 /* 3 tries, this should take less than a second! */
866 s
= drbd_try_connect(tconn
);
869 /* give the other side time to call bind() & listen() */
870 schedule_timeout_interruptible(HZ
/ 10);
874 if (!tconn
->data
.socket
) {
875 tconn
->data
.socket
= s
;
876 send_first_packet(tconn
, &tconn
->data
, P_INITIAL_DATA
);
877 } else if (!tconn
->meta
.socket
) {
878 tconn
->meta
.socket
= s
;
879 send_first_packet(tconn
, &tconn
->meta
, P_INITIAL_META
);
881 conn_err(tconn
, "Logic error in drbd_connect()\n");
882 goto out_release_sockets
;
886 if (tconn
->data
.socket
&& tconn
->meta
.socket
) {
887 schedule_timeout_interruptible(tconn
->net_conf
->ping_timeo
*HZ
/10);
888 ok
= drbd_socket_okay(&tconn
->data
.socket
);
889 ok
= drbd_socket_okay(&tconn
->meta
.socket
) && ok
;
895 s
= drbd_wait_for_connect(tconn
);
897 try = receive_first_packet(tconn
, s
);
898 drbd_socket_okay(&tconn
->data
.socket
);
899 drbd_socket_okay(&tconn
->meta
.socket
);
902 if (tconn
->data
.socket
) {
903 conn_warn(tconn
, "initial packet S crossed\n");
904 sock_release(tconn
->data
.socket
);
906 tconn
->data
.socket
= s
;
909 if (tconn
->meta
.socket
) {
910 conn_warn(tconn
, "initial packet M crossed\n");
911 sock_release(tconn
->meta
.socket
);
913 tconn
->meta
.socket
= s
;
914 set_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
917 conn_warn(tconn
, "Error receiving initial packet\n");
924 if (tconn
->cstate
<= C_DISCONNECTING
)
925 goto out_release_sockets
;
926 if (signal_pending(current
)) {
927 flush_signals(current
);
929 if (get_t_state(&tconn
->receiver
) == EXITING
)
930 goto out_release_sockets
;
933 if (tconn
->data
.socket
&& &tconn
->meta
.socket
) {
934 ok
= drbd_socket_okay(&tconn
->data
.socket
);
935 ok
= drbd_socket_okay(&tconn
->meta
.socket
) && ok
;
941 sock
= tconn
->data
.socket
;
942 msock
= tconn
->meta
.socket
;
944 msock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
945 sock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
947 sock
->sk
->sk_allocation
= GFP_NOIO
;
948 msock
->sk
->sk_allocation
= GFP_NOIO
;
950 sock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE_BULK
;
951 msock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE
;
954 * sock->sk->sk_sndtimeo = tconn->net_conf->timeout*HZ/10;
955 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
956 * first set it to the P_CONNECTION_FEATURES timeout,
957 * which we set to 4x the configured ping_timeout. */
959 nc
= rcu_dereference(tconn
->net_conf
);
961 sock
->sk
->sk_sndtimeo
=
962 sock
->sk
->sk_rcvtimeo
= nc
->ping_timeo
*4*HZ
/10;
964 msock
->sk
->sk_rcvtimeo
= nc
->ping_int
*HZ
;
965 timeout
= nc
->timeout
* HZ
/ 10;
968 msock
->sk
->sk_sndtimeo
= timeout
;
970 /* we don't want delays.
971 * we use TCP_CORK where appropriate, though */
972 drbd_tcp_nodelay(sock
);
973 drbd_tcp_nodelay(msock
);
975 tconn
->last_received
= jiffies
;
977 h
= drbd_do_features(tconn
);
981 if (tconn
->cram_hmac_tfm
) {
982 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
983 switch (drbd_do_auth(tconn
)) {
985 conn_err(tconn
, "Authentication of peer failed\n");
988 conn_err(tconn
, "Authentication of peer failed, trying again.\n");
993 if (conn_request_state(tconn
, NS(conn
, C_WF_REPORT_PARAMS
), CS_VERBOSE
) < SS_SUCCESS
)
996 sock
->sk
->sk_sndtimeo
= timeout
;
997 sock
->sk
->sk_rcvtimeo
= MAX_SCHEDULE_TIMEOUT
;
999 drbd_thread_start(&tconn
->asender
);
1001 if (drbd_send_protocol(tconn
) == -EOPNOTSUPP
)
1004 down_read(&drbd_cfg_rwsem
);
1005 h
= !idr_for_each(&tconn
->volumes
, drbd_connected
, tconn
);
1006 up_read(&drbd_cfg_rwsem
);
1009 out_release_sockets
:
1010 if (tconn
->data
.socket
) {
1011 sock_release(tconn
->data
.socket
);
1012 tconn
->data
.socket
= NULL
;
1014 if (tconn
->meta
.socket
) {
1015 sock_release(tconn
->meta
.socket
);
1016 tconn
->meta
.socket
= NULL
;
1021 static int decode_header(struct drbd_tconn
*tconn
, void *header
, struct packet_info
*pi
)
1023 unsigned int header_size
= drbd_header_size(tconn
);
1025 if (header_size
== sizeof(struct p_header100
) &&
1026 *(__be32
*)header
== cpu_to_be32(DRBD_MAGIC_100
)) {
1027 struct p_header100
*h
= header
;
1029 conn_err(tconn
, "Header padding is not zero\n");
1032 pi
->vnr
= be16_to_cpu(h
->volume
);
1033 pi
->cmd
= be16_to_cpu(h
->command
);
1034 pi
->size
= be32_to_cpu(h
->length
);
1035 } else if (header_size
== sizeof(struct p_header95
) &&
1036 *(__be16
*)header
== cpu_to_be16(DRBD_MAGIC_BIG
)) {
1037 struct p_header95
*h
= header
;
1038 pi
->cmd
= be16_to_cpu(h
->command
);
1039 pi
->size
= be32_to_cpu(h
->length
);
1041 } else if (header_size
== sizeof(struct p_header80
) &&
1042 *(__be32
*)header
== cpu_to_be32(DRBD_MAGIC
)) {
1043 struct p_header80
*h
= header
;
1044 pi
->cmd
= be16_to_cpu(h
->command
);
1045 pi
->size
= be16_to_cpu(h
->length
);
1048 conn_err(tconn
, "Wrong magic value 0x%08x in protocol version %d\n",
1049 be32_to_cpu(*(__be32
*)header
),
1050 tconn
->agreed_pro_version
);
1053 pi
->data
= header
+ header_size
;
1057 static int drbd_recv_header(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1059 void *buffer
= tconn
->data
.rbuf
;
1062 err
= drbd_recv_all_warn(tconn
, buffer
, drbd_header_size(tconn
));
1066 err
= decode_header(tconn
, buffer
, pi
);
1067 tconn
->last_received
= jiffies
;
1072 static void drbd_flush(struct drbd_conf
*mdev
)
1076 if (mdev
->write_ordering
>= WO_bdev_flush
&& get_ldev(mdev
)) {
1077 rv
= blkdev_issue_flush(mdev
->ldev
->backing_bdev
, GFP_KERNEL
,
1080 dev_err(DEV
, "local disk flush failed with status %d\n", rv
);
1081 /* would rather check on EOPNOTSUPP, but that is not reliable.
1082 * don't try again for ANY return value != 0
1083 * if (rv == -EOPNOTSUPP) */
1084 drbd_bump_write_ordering(mdev
, WO_drain_io
);
1091 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1092 * @mdev: DRBD device.
1093 * @epoch: Epoch object.
1096 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*mdev
,
1097 struct drbd_epoch
*epoch
,
1098 enum epoch_event ev
)
1101 struct drbd_epoch
*next_epoch
;
1102 enum finish_epoch rv
= FE_STILL_LIVE
;
1104 spin_lock(&mdev
->epoch_lock
);
1108 epoch_size
= atomic_read(&epoch
->epoch_size
);
1110 switch (ev
& ~EV_CLEANUP
) {
1112 atomic_dec(&epoch
->active
);
1114 case EV_GOT_BARRIER_NR
:
1115 set_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
);
1117 case EV_BECAME_LAST
:
1122 if (epoch_size
!= 0 &&
1123 atomic_read(&epoch
->active
) == 0 &&
1124 test_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
)) {
1125 if (!(ev
& EV_CLEANUP
)) {
1126 spin_unlock(&mdev
->epoch_lock
);
1127 drbd_send_b_ack(mdev
, epoch
->barrier_nr
, epoch_size
);
1128 spin_lock(&mdev
->epoch_lock
);
1132 if (mdev
->current_epoch
!= epoch
) {
1133 next_epoch
= list_entry(epoch
->list
.next
, struct drbd_epoch
, list
);
1134 list_del(&epoch
->list
);
1135 ev
= EV_BECAME_LAST
| (ev
& EV_CLEANUP
);
1139 if (rv
== FE_STILL_LIVE
)
1143 atomic_set(&epoch
->epoch_size
, 0);
1144 /* atomic_set(&epoch->active, 0); is already zero */
1145 if (rv
== FE_STILL_LIVE
)
1147 wake_up(&mdev
->ee_wait
);
1157 spin_unlock(&mdev
->epoch_lock
);
1163 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1164 * @mdev: DRBD device.
1165 * @wo: Write ordering method to try.
1167 void drbd_bump_write_ordering(struct drbd_conf
*mdev
, enum write_ordering_e wo
) __must_hold(local
)
1169 struct disk_conf
*dc
;
1170 enum write_ordering_e pwo
;
1171 static char *write_ordering_str
[] = {
1173 [WO_drain_io
] = "drain",
1174 [WO_bdev_flush
] = "flush",
1177 pwo
= mdev
->write_ordering
;
1180 dc
= rcu_dereference(mdev
->ldev
->disk_conf
);
1182 if (wo
== WO_bdev_flush
&& dc
->no_disk_flush
)
1184 if (wo
== WO_drain_io
&& dc
->no_disk_drain
)
1187 mdev
->write_ordering
= wo
;
1188 if (pwo
!= mdev
->write_ordering
|| wo
== WO_bdev_flush
)
1189 dev_info(DEV
, "Method to ensure write ordering: %s\n", write_ordering_str
[mdev
->write_ordering
]);
1193 * drbd_submit_peer_request()
1194 * @mdev: DRBD device.
1195 * @peer_req: peer request
1196 * @rw: flag field, see bio->bi_rw
1198 * May spread the pages to multiple bios,
1199 * depending on bio_add_page restrictions.
1201 * Returns 0 if all bios have been submitted,
1202 * -ENOMEM if we could not allocate enough bios,
1203 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1204 * single page to an empty bio (which should never happen and likely indicates
1205 * that the lower level IO stack is in some way broken). This has been observed
1206 * on certain Xen deployments.
1208 /* TODO allocate from our own bio_set. */
1209 int drbd_submit_peer_request(struct drbd_conf
*mdev
,
1210 struct drbd_peer_request
*peer_req
,
1211 const unsigned rw
, const int fault_type
)
1213 struct bio
*bios
= NULL
;
1215 struct page
*page
= peer_req
->pages
;
1216 sector_t sector
= peer_req
->i
.sector
;
1217 unsigned ds
= peer_req
->i
.size
;
1218 unsigned n_bios
= 0;
1219 unsigned nr_pages
= (ds
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
1222 /* In most cases, we will only need one bio. But in case the lower
1223 * level restrictions happen to be different at this offset on this
1224 * side than those of the sending peer, we may need to submit the
1225 * request in more than one bio.
1227 * Plain bio_alloc is good enough here, this is no DRBD internally
1228 * generated bio, but a bio allocated on behalf of the peer.
1231 bio
= bio_alloc(GFP_NOIO
, nr_pages
);
1233 dev_err(DEV
, "submit_ee: Allocation of a bio failed\n");
1236 /* > peer_req->i.sector, unless this is the first bio */
1237 bio
->bi_sector
= sector
;
1238 bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1240 bio
->bi_private
= peer_req
;
1241 bio
->bi_end_io
= drbd_peer_request_endio
;
1243 bio
->bi_next
= bios
;
1247 page_chain_for_each(page
) {
1248 unsigned len
= min_t(unsigned, ds
, PAGE_SIZE
);
1249 if (!bio_add_page(bio
, page
, len
, 0)) {
1250 /* A single page must always be possible!
1251 * But in case it fails anyways,
1252 * we deal with it, and complain (below). */
1253 if (bio
->bi_vcnt
== 0) {
1255 "bio_add_page failed for len=%u, "
1256 "bi_vcnt=0 (bi_sector=%llu)\n",
1257 len
, (unsigned long long)bio
->bi_sector
);
1267 D_ASSERT(page
== NULL
);
1270 atomic_set(&peer_req
->pending_bios
, n_bios
);
1273 bios
= bios
->bi_next
;
1274 bio
->bi_next
= NULL
;
1276 drbd_generic_make_request(mdev
, fault_type
, bio
);
1283 bios
= bios
->bi_next
;
1289 static void drbd_remove_epoch_entry_interval(struct drbd_conf
*mdev
,
1290 struct drbd_peer_request
*peer_req
)
1292 struct drbd_interval
*i
= &peer_req
->i
;
1294 drbd_remove_interval(&mdev
->write_requests
, i
);
1295 drbd_clear_interval(i
);
1297 /* Wake up any processes waiting for this peer request to complete. */
1299 wake_up(&mdev
->misc_wait
);
1302 static int receive_Barrier(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1304 struct drbd_conf
*mdev
;
1306 struct p_barrier
*p
= pi
->data
;
1307 struct drbd_epoch
*epoch
;
1309 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1315 mdev
->current_epoch
->barrier_nr
= p
->barrier
;
1316 rv
= drbd_may_finish_epoch(mdev
, mdev
->current_epoch
, EV_GOT_BARRIER_NR
);
1318 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1319 * the activity log, which means it would not be resynced in case the
1320 * R_PRIMARY crashes now.
1321 * Therefore we must send the barrier_ack after the barrier request was
1323 switch (mdev
->write_ordering
) {
1325 if (rv
== FE_RECYCLED
)
1328 /* receiver context, in the writeout path of the other node.
1329 * avoid potential distributed deadlock */
1330 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1334 dev_warn(DEV
, "Allocation of an epoch failed, slowing down\n");
1339 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1342 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1343 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1348 epoch
= mdev
->current_epoch
;
1349 wait_event(mdev
->ee_wait
, atomic_read(&epoch
->epoch_size
) == 0);
1351 D_ASSERT(atomic_read(&epoch
->active
) == 0);
1352 D_ASSERT(epoch
->flags
== 0);
1356 dev_err(DEV
, "Strangeness in mdev->write_ordering %d\n", mdev
->write_ordering
);
1361 atomic_set(&epoch
->epoch_size
, 0);
1362 atomic_set(&epoch
->active
, 0);
1364 spin_lock(&mdev
->epoch_lock
);
1365 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1366 list_add(&epoch
->list
, &mdev
->current_epoch
->list
);
1367 mdev
->current_epoch
= epoch
;
1370 /* The current_epoch got recycled while we allocated this one... */
1373 spin_unlock(&mdev
->epoch_lock
);
1378 /* used from receive_RSDataReply (recv_resync_read)
1379 * and from receive_Data */
1380 static struct drbd_peer_request
*
1381 read_in_block(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
1382 int data_size
) __must_hold(local
)
1384 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1385 struct drbd_peer_request
*peer_req
;
1388 void *dig_in
= mdev
->tconn
->int_dig_in
;
1389 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1390 unsigned long *data
;
1393 if (mdev
->tconn
->peer_integrity_tfm
) {
1394 dgs
= crypto_hash_digestsize(mdev
->tconn
->peer_integrity_tfm
);
1396 * FIXME: Receive the incoming digest into the receive buffer
1397 * here, together with its struct p_data?
1399 err
= drbd_recv_all_warn(mdev
->tconn
, dig_in
, dgs
);
1405 if (!expect(data_size
!= 0))
1407 if (!expect(IS_ALIGNED(data_size
, 512)))
1409 if (!expect(data_size
<= DRBD_MAX_BIO_SIZE
))
1412 /* even though we trust out peer,
1413 * we sometimes have to double check. */
1414 if (sector
+ (data_size
>>9) > capacity
) {
1415 dev_err(DEV
, "request from peer beyond end of local disk: "
1416 "capacity: %llus < sector: %llus + size: %u\n",
1417 (unsigned long long)capacity
,
1418 (unsigned long long)sector
, data_size
);
1422 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1423 * "criss-cross" setup, that might cause write-out on some other DRBD,
1424 * which in turn might block on the other node at this very place. */
1425 peer_req
= drbd_alloc_peer_req(mdev
, id
, sector
, data_size
, GFP_NOIO
);
1430 page
= peer_req
->pages
;
1431 page_chain_for_each(page
) {
1432 unsigned len
= min_t(int, ds
, PAGE_SIZE
);
1434 err
= drbd_recv_all_warn(mdev
->tconn
, data
, len
);
1435 if (drbd_insert_fault(mdev
, DRBD_FAULT_RECEIVE
)) {
1436 dev_err(DEV
, "Fault injection: Corrupting data on receive\n");
1437 data
[0] = data
[0] ^ (unsigned long)-1;
1441 drbd_free_peer_req(mdev
, peer_req
);
1448 drbd_csum_ee(mdev
, mdev
->tconn
->peer_integrity_tfm
, peer_req
, dig_vv
);
1449 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1450 dev_err(DEV
, "Digest integrity check FAILED: %llus +%u\n",
1451 (unsigned long long)sector
, data_size
);
1452 drbd_free_peer_req(mdev
, peer_req
);
1456 mdev
->recv_cnt
+= data_size
>>9;
1460 /* drbd_drain_block() just takes a data block
1461 * out of the socket input buffer, and discards it.
1463 static int drbd_drain_block(struct drbd_conf
*mdev
, int data_size
)
1472 page
= drbd_alloc_pages(mdev
, 1, 1);
1476 unsigned int len
= min_t(int, data_size
, PAGE_SIZE
);
1478 err
= drbd_recv_all_warn(mdev
->tconn
, data
, len
);
1484 drbd_free_pages(mdev
, page
, 0);
1488 static int recv_dless_read(struct drbd_conf
*mdev
, struct drbd_request
*req
,
1489 sector_t sector
, int data_size
)
1491 struct bio_vec
*bvec
;
1493 int dgs
, err
, i
, expect
;
1494 void *dig_in
= mdev
->tconn
->int_dig_in
;
1495 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1498 if (mdev
->tconn
->peer_integrity_tfm
) {
1499 dgs
= crypto_hash_digestsize(mdev
->tconn
->peer_integrity_tfm
);
1500 err
= drbd_recv_all_warn(mdev
->tconn
, dig_in
, dgs
);
1506 /* optimistically update recv_cnt. if receiving fails below,
1507 * we disconnect anyways, and counters will be reset. */
1508 mdev
->recv_cnt
+= data_size
>>9;
1510 bio
= req
->master_bio
;
1511 D_ASSERT(sector
== bio
->bi_sector
);
1513 bio_for_each_segment(bvec
, bio
, i
) {
1514 void *mapped
= kmap(bvec
->bv_page
) + bvec
->bv_offset
;
1515 expect
= min_t(int, data_size
, bvec
->bv_len
);
1516 err
= drbd_recv_all_warn(mdev
->tconn
, mapped
, expect
);
1517 kunmap(bvec
->bv_page
);
1520 data_size
-= expect
;
1524 drbd_csum_bio(mdev
, mdev
->tconn
->peer_integrity_tfm
, bio
, dig_vv
);
1525 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1526 dev_err(DEV
, "Digest integrity check FAILED. Broken NICs?\n");
1531 D_ASSERT(data_size
== 0);
1536 * e_end_resync_block() is called in asender context via
1537 * drbd_finish_peer_reqs().
1539 static int e_end_resync_block(struct drbd_work
*w
, int unused
)
1541 struct drbd_peer_request
*peer_req
=
1542 container_of(w
, struct drbd_peer_request
, w
);
1543 struct drbd_conf
*mdev
= w
->mdev
;
1544 sector_t sector
= peer_req
->i
.sector
;
1547 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1549 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1550 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1551 err
= drbd_send_ack(mdev
, P_RS_WRITE_ACK
, peer_req
);
1553 /* Record failure to sync */
1554 drbd_rs_failed_io(mdev
, sector
, peer_req
->i
.size
);
1556 err
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1563 static int recv_resync_read(struct drbd_conf
*mdev
, sector_t sector
, int data_size
) __releases(local
)
1565 struct drbd_peer_request
*peer_req
;
1567 peer_req
= read_in_block(mdev
, ID_SYNCER
, sector
, data_size
);
1571 dec_rs_pending(mdev
);
1574 /* corresponding dec_unacked() in e_end_resync_block()
1575 * respective _drbd_clear_done_ee */
1577 peer_req
->w
.cb
= e_end_resync_block
;
1579 spin_lock_irq(&mdev
->tconn
->req_lock
);
1580 list_add(&peer_req
->w
.list
, &mdev
->sync_ee
);
1581 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1583 atomic_add(data_size
>> 9, &mdev
->rs_sect_ev
);
1584 if (drbd_submit_peer_request(mdev
, peer_req
, WRITE
, DRBD_FAULT_RS_WR
) == 0)
1587 /* don't care for the reason here */
1588 dev_err(DEV
, "submit failed, triggering re-connect\n");
1589 spin_lock_irq(&mdev
->tconn
->req_lock
);
1590 list_del(&peer_req
->w
.list
);
1591 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1593 drbd_free_peer_req(mdev
, peer_req
);
1599 static struct drbd_request
*
1600 find_request(struct drbd_conf
*mdev
, struct rb_root
*root
, u64 id
,
1601 sector_t sector
, bool missing_ok
, const char *func
)
1603 struct drbd_request
*req
;
1605 /* Request object according to our peer */
1606 req
= (struct drbd_request
*)(unsigned long)id
;
1607 if (drbd_contains_interval(root
, sector
, &req
->i
) && req
->i
.local
)
1610 dev_err(DEV
, "%s: failed to find request %lu, sector %llus\n", func
,
1611 (unsigned long)id
, (unsigned long long)sector
);
1616 static int receive_DataReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1618 struct drbd_conf
*mdev
;
1619 struct drbd_request
*req
;
1622 struct p_data
*p
= pi
->data
;
1624 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1628 sector
= be64_to_cpu(p
->sector
);
1630 spin_lock_irq(&mdev
->tconn
->req_lock
);
1631 req
= find_request(mdev
, &mdev
->read_requests
, p
->block_id
, sector
, false, __func__
);
1632 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1636 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1637 * special casing it there for the various failure cases.
1638 * still no race with drbd_fail_pending_reads */
1639 err
= recv_dless_read(mdev
, req
, sector
, pi
->size
);
1641 req_mod(req
, DATA_RECEIVED
);
1642 /* else: nothing. handled from drbd_disconnect...
1643 * I don't think we may complete this just yet
1644 * in case we are "on-disconnect: freeze" */
1649 static int receive_RSDataReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1651 struct drbd_conf
*mdev
;
1654 struct p_data
*p
= pi
->data
;
1656 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1660 sector
= be64_to_cpu(p
->sector
);
1661 D_ASSERT(p
->block_id
== ID_SYNCER
);
1663 if (get_ldev(mdev
)) {
1664 /* data is submitted to disk within recv_resync_read.
1665 * corresponding put_ldev done below on error,
1666 * or in drbd_peer_request_endio. */
1667 err
= recv_resync_read(mdev
, sector
, pi
->size
);
1669 if (__ratelimit(&drbd_ratelimit_state
))
1670 dev_err(DEV
, "Can not write resync data to local disk.\n");
1672 err
= drbd_drain_block(mdev
, pi
->size
);
1674 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, pi
->size
);
1677 atomic_add(pi
->size
>> 9, &mdev
->rs_sect_in
);
1682 static int w_restart_write(struct drbd_work
*w
, int cancel
)
1684 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1685 struct drbd_conf
*mdev
= w
->mdev
;
1687 unsigned long start_time
;
1688 unsigned long flags
;
1690 spin_lock_irqsave(&mdev
->tconn
->req_lock
, flags
);
1691 if (!expect(req
->rq_state
& RQ_POSTPONED
)) {
1692 spin_unlock_irqrestore(&mdev
->tconn
->req_lock
, flags
);
1695 bio
= req
->master_bio
;
1696 start_time
= req
->start_time
;
1697 /* Postponed requests will not have their master_bio completed! */
1698 __req_mod(req
, DISCARD_WRITE
, NULL
);
1699 spin_unlock_irqrestore(&mdev
->tconn
->req_lock
, flags
);
1701 while (__drbd_make_request(mdev
, bio
, start_time
))
1706 static void restart_conflicting_writes(struct drbd_conf
*mdev
,
1707 sector_t sector
, int size
)
1709 struct drbd_interval
*i
;
1710 struct drbd_request
*req
;
1712 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1715 req
= container_of(i
, struct drbd_request
, i
);
1716 if (req
->rq_state
& RQ_LOCAL_PENDING
||
1717 !(req
->rq_state
& RQ_POSTPONED
))
1719 if (expect(list_empty(&req
->w
.list
))) {
1721 req
->w
.cb
= w_restart_write
;
1722 drbd_queue_work(&mdev
->tconn
->data
.work
, &req
->w
);
1728 * e_end_block() is called in asender context via drbd_finish_peer_reqs().
1730 static int e_end_block(struct drbd_work
*w
, int cancel
)
1732 struct drbd_peer_request
*peer_req
=
1733 container_of(w
, struct drbd_peer_request
, w
);
1734 struct drbd_conf
*mdev
= w
->mdev
;
1735 sector_t sector
= peer_req
->i
.sector
;
1738 if (peer_req
->flags
& EE_SEND_WRITE_ACK
) {
1739 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1740 pcmd
= (mdev
->state
.conn
>= C_SYNC_SOURCE
&&
1741 mdev
->state
.conn
<= C_PAUSED_SYNC_T
&&
1742 peer_req
->flags
& EE_MAY_SET_IN_SYNC
) ?
1743 P_RS_WRITE_ACK
: P_WRITE_ACK
;
1744 err
= drbd_send_ack(mdev
, pcmd
, peer_req
);
1745 if (pcmd
== P_RS_WRITE_ACK
)
1746 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1748 err
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1749 /* we expect it to be marked out of sync anyways...
1750 * maybe assert this? */
1754 /* we delete from the conflict detection hash _after_ we sent out the
1755 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1756 if (peer_req
->flags
& EE_IN_INTERVAL_TREE
) {
1757 spin_lock_irq(&mdev
->tconn
->req_lock
);
1758 D_ASSERT(!drbd_interval_empty(&peer_req
->i
));
1759 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
1760 if (peer_req
->flags
& EE_RESTART_REQUESTS
)
1761 restart_conflicting_writes(mdev
, sector
, peer_req
->i
.size
);
1762 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1764 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1766 drbd_may_finish_epoch(mdev
, peer_req
->epoch
, EV_PUT
+ (cancel
? EV_CLEANUP
: 0));
1771 static int e_send_ack(struct drbd_work
*w
, enum drbd_packet ack
)
1773 struct drbd_conf
*mdev
= w
->mdev
;
1774 struct drbd_peer_request
*peer_req
=
1775 container_of(w
, struct drbd_peer_request
, w
);
1778 err
= drbd_send_ack(mdev
, ack
, peer_req
);
1784 static int e_send_discard_write(struct drbd_work
*w
, int unused
)
1786 return e_send_ack(w
, P_DISCARD_WRITE
);
1789 static int e_send_retry_write(struct drbd_work
*w
, int unused
)
1791 struct drbd_tconn
*tconn
= w
->mdev
->tconn
;
1793 return e_send_ack(w
, tconn
->agreed_pro_version
>= 100 ?
1794 P_RETRY_WRITE
: P_DISCARD_WRITE
);
1797 static bool seq_greater(u32 a
, u32 b
)
1800 * We assume 32-bit wrap-around here.
1801 * For 24-bit wrap-around, we would have to shift:
1804 return (s32
)a
- (s32
)b
> 0;
1807 static u32
seq_max(u32 a
, u32 b
)
1809 return seq_greater(a
, b
) ? a
: b
;
1812 static bool need_peer_seq(struct drbd_conf
*mdev
)
1814 struct drbd_tconn
*tconn
= mdev
->tconn
;
1818 * We only need to keep track of the last packet_seq number of our peer
1819 * if we are in dual-primary mode and we have the discard flag set; see
1820 * handle_write_conflicts().
1824 tp
= rcu_dereference(mdev
->tconn
->net_conf
)->two_primaries
;
1827 return tp
&& test_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
1830 static void update_peer_seq(struct drbd_conf
*mdev
, unsigned int peer_seq
)
1832 unsigned int newest_peer_seq
;
1834 if (need_peer_seq(mdev
)) {
1835 spin_lock(&mdev
->peer_seq_lock
);
1836 newest_peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1837 mdev
->peer_seq
= newest_peer_seq
;
1838 spin_unlock(&mdev
->peer_seq_lock
);
1839 /* wake up only if we actually changed mdev->peer_seq */
1840 if (peer_seq
== newest_peer_seq
)
1841 wake_up(&mdev
->seq_wait
);
1845 /* Called from receive_Data.
1846 * Synchronize packets on sock with packets on msock.
1848 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1849 * packet traveling on msock, they are still processed in the order they have
1852 * Note: we don't care for Ack packets overtaking P_DATA packets.
1854 * In case packet_seq is larger than mdev->peer_seq number, there are
1855 * outstanding packets on the msock. We wait for them to arrive.
1856 * In case we are the logically next packet, we update mdev->peer_seq
1857 * ourselves. Correctly handles 32bit wrap around.
1859 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1860 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1861 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1862 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1864 * returns 0 if we may process the packet,
1865 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1866 static int wait_for_and_update_peer_seq(struct drbd_conf
*mdev
, const u32 peer_seq
)
1872 if (!need_peer_seq(mdev
))
1875 spin_lock(&mdev
->peer_seq_lock
);
1877 if (!seq_greater(peer_seq
- 1, mdev
->peer_seq
)) {
1878 mdev
->peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1882 if (signal_pending(current
)) {
1886 prepare_to_wait(&mdev
->seq_wait
, &wait
, TASK_INTERRUPTIBLE
);
1887 spin_unlock(&mdev
->peer_seq_lock
);
1889 timeout
= rcu_dereference(mdev
->tconn
->net_conf
)->ping_timeo
*HZ
/10;
1891 timeout
= schedule_timeout(timeout
);
1892 spin_lock(&mdev
->peer_seq_lock
);
1895 dev_err(DEV
, "Timed out waiting for missing ack packets; disconnecting\n");
1899 spin_unlock(&mdev
->peer_seq_lock
);
1900 finish_wait(&mdev
->seq_wait
, &wait
);
1904 /* see also bio_flags_to_wire()
1905 * DRBD_REQ_*, because we need to semantically map the flags to data packet
1906 * flags and back. We may replicate to other kernel versions. */
1907 static unsigned long wire_flags_to_bio(struct drbd_conf
*mdev
, u32 dpf
)
1909 return (dpf
& DP_RW_SYNC
? REQ_SYNC
: 0) |
1910 (dpf
& DP_FUA
? REQ_FUA
: 0) |
1911 (dpf
& DP_FLUSH
? REQ_FLUSH
: 0) |
1912 (dpf
& DP_DISCARD
? REQ_DISCARD
: 0);
1915 static void fail_postponed_requests(struct drbd_conf
*mdev
, sector_t sector
,
1918 struct drbd_interval
*i
;
1921 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1922 struct drbd_request
*req
;
1923 struct bio_and_error m
;
1927 req
= container_of(i
, struct drbd_request
, i
);
1928 if (!(req
->rq_state
& RQ_POSTPONED
))
1930 req
->rq_state
&= ~RQ_POSTPONED
;
1931 __req_mod(req
, NEG_ACKED
, &m
);
1932 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1934 complete_master_bio(mdev
, &m
);
1935 spin_lock_irq(&mdev
->tconn
->req_lock
);
1940 static int handle_write_conflicts(struct drbd_conf
*mdev
,
1941 struct drbd_peer_request
*peer_req
)
1943 struct drbd_tconn
*tconn
= mdev
->tconn
;
1944 bool resolve_conflicts
= test_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
1945 sector_t sector
= peer_req
->i
.sector
;
1946 const unsigned int size
= peer_req
->i
.size
;
1947 struct drbd_interval
*i
;
1952 * Inserting the peer request into the write_requests tree will prevent
1953 * new conflicting local requests from being added.
1955 drbd_insert_interval(&mdev
->write_requests
, &peer_req
->i
);
1958 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1959 if (i
== &peer_req
->i
)
1964 * Our peer has sent a conflicting remote request; this
1965 * should not happen in a two-node setup. Wait for the
1966 * earlier peer request to complete.
1968 err
= drbd_wait_misc(mdev
, i
);
1974 equal
= i
->sector
== sector
&& i
->size
== size
;
1975 if (resolve_conflicts
) {
1977 * If the peer request is fully contained within the
1978 * overlapping request, it can be discarded; otherwise,
1979 * it will be retried once all overlapping requests
1982 bool discard
= i
->sector
<= sector
&& i
->sector
+
1983 (i
->size
>> 9) >= sector
+ (size
>> 9);
1986 dev_alert(DEV
, "Concurrent writes detected: "
1987 "local=%llus +%u, remote=%llus +%u, "
1988 "assuming %s came first\n",
1989 (unsigned long long)i
->sector
, i
->size
,
1990 (unsigned long long)sector
, size
,
1991 discard
? "local" : "remote");
1994 peer_req
->w
.cb
= discard
? e_send_discard_write
:
1996 list_add_tail(&peer_req
->w
.list
, &mdev
->done_ee
);
1997 wake_asender(mdev
->tconn
);
2002 struct drbd_request
*req
=
2003 container_of(i
, struct drbd_request
, i
);
2006 dev_alert(DEV
, "Concurrent writes detected: "
2007 "local=%llus +%u, remote=%llus +%u\n",
2008 (unsigned long long)i
->sector
, i
->size
,
2009 (unsigned long long)sector
, size
);
2011 if (req
->rq_state
& RQ_LOCAL_PENDING
||
2012 !(req
->rq_state
& RQ_POSTPONED
)) {
2014 * Wait for the node with the discard flag to
2015 * decide if this request will be discarded or
2016 * retried. Requests that are discarded will
2017 * disappear from the write_requests tree.
2019 * In addition, wait for the conflicting
2020 * request to finish locally before submitting
2021 * the conflicting peer request.
2023 err
= drbd_wait_misc(mdev
, &req
->i
);
2025 _conn_request_state(mdev
->tconn
,
2026 NS(conn
, C_TIMEOUT
),
2028 fail_postponed_requests(mdev
, sector
, size
);
2034 * Remember to restart the conflicting requests after
2035 * the new peer request has completed.
2037 peer_req
->flags
|= EE_RESTART_REQUESTS
;
2044 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
2048 /* mirrored write */
2049 static int receive_Data(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2051 struct drbd_conf
*mdev
;
2053 struct drbd_peer_request
*peer_req
;
2054 struct p_data
*p
= pi
->data
;
2055 u32 peer_seq
= be32_to_cpu(p
->seq_num
);
2060 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
2064 if (!get_ldev(mdev
)) {
2067 err
= wait_for_and_update_peer_seq(mdev
, peer_seq
);
2068 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, pi
->size
);
2069 atomic_inc(&mdev
->current_epoch
->epoch_size
);
2070 err2
= drbd_drain_block(mdev
, pi
->size
);
2077 * Corresponding put_ldev done either below (on various errors), or in
2078 * drbd_peer_request_endio, if we successfully submit the data at the
2079 * end of this function.
2082 sector
= be64_to_cpu(p
->sector
);
2083 peer_req
= read_in_block(mdev
, p
->block_id
, sector
, pi
->size
);
2089 peer_req
->w
.cb
= e_end_block
;
2091 dp_flags
= be32_to_cpu(p
->dp_flags
);
2092 rw
|= wire_flags_to_bio(mdev
, dp_flags
);
2094 if (dp_flags
& DP_MAY_SET_IN_SYNC
)
2095 peer_req
->flags
|= EE_MAY_SET_IN_SYNC
;
2097 spin_lock(&mdev
->epoch_lock
);
2098 peer_req
->epoch
= mdev
->current_epoch
;
2099 atomic_inc(&peer_req
->epoch
->epoch_size
);
2100 atomic_inc(&peer_req
->epoch
->active
);
2101 spin_unlock(&mdev
->epoch_lock
);
2104 tp
= rcu_dereference(mdev
->tconn
->net_conf
)->two_primaries
;
2107 peer_req
->flags
|= EE_IN_INTERVAL_TREE
;
2108 err
= wait_for_and_update_peer_seq(mdev
, peer_seq
);
2110 goto out_interrupted
;
2111 spin_lock_irq(&mdev
->tconn
->req_lock
);
2112 err
= handle_write_conflicts(mdev
, peer_req
);
2114 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2115 if (err
== -ENOENT
) {
2119 goto out_interrupted
;
2122 spin_lock_irq(&mdev
->tconn
->req_lock
);
2123 list_add(&peer_req
->w
.list
, &mdev
->active_ee
);
2124 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2126 if (mdev
->tconn
->agreed_pro_version
< 100) {
2128 switch (rcu_dereference(mdev
->tconn
->net_conf
)->wire_protocol
) {
2130 dp_flags
|= DP_SEND_WRITE_ACK
;
2133 dp_flags
|= DP_SEND_RECEIVE_ACK
;
2139 if (dp_flags
& DP_SEND_WRITE_ACK
) {
2140 peer_req
->flags
|= EE_SEND_WRITE_ACK
;
2142 /* corresponding dec_unacked() in e_end_block()
2143 * respective _drbd_clear_done_ee */
2146 if (dp_flags
& DP_SEND_RECEIVE_ACK
) {
2147 /* I really don't like it that the receiver thread
2148 * sends on the msock, but anyways */
2149 drbd_send_ack(mdev
, P_RECV_ACK
, peer_req
);
2152 if (mdev
->state
.pdsk
< D_INCONSISTENT
) {
2153 /* In case we have the only disk of the cluster, */
2154 drbd_set_out_of_sync(mdev
, peer_req
->i
.sector
, peer_req
->i
.size
);
2155 peer_req
->flags
|= EE_CALL_AL_COMPLETE_IO
;
2156 peer_req
->flags
&= ~EE_MAY_SET_IN_SYNC
;
2157 drbd_al_begin_io(mdev
, &peer_req
->i
);
2160 err
= drbd_submit_peer_request(mdev
, peer_req
, rw
, DRBD_FAULT_DT_WR
);
2164 /* don't care for the reason here */
2165 dev_err(DEV
, "submit failed, triggering re-connect\n");
2166 spin_lock_irq(&mdev
->tconn
->req_lock
);
2167 list_del(&peer_req
->w
.list
);
2168 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
2169 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2170 if (peer_req
->flags
& EE_CALL_AL_COMPLETE_IO
)
2171 drbd_al_complete_io(mdev
, &peer_req
->i
);
2174 drbd_may_finish_epoch(mdev
, peer_req
->epoch
, EV_PUT
+ EV_CLEANUP
);
2176 drbd_free_peer_req(mdev
, peer_req
);
2180 /* We may throttle resync, if the lower device seems to be busy,
2181 * and current sync rate is above c_min_rate.
2183 * To decide whether or not the lower device is busy, we use a scheme similar
2184 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2185 * (more than 64 sectors) of activity we cannot account for with our own resync
2186 * activity, it obviously is "busy".
2188 * The current sync rate used here uses only the most recent two step marks,
2189 * to have a short time average so we can react faster.
2191 int drbd_rs_should_slow_down(struct drbd_conf
*mdev
, sector_t sector
)
2193 struct gendisk
*disk
= mdev
->ldev
->backing_bdev
->bd_contains
->bd_disk
;
2194 unsigned long db
, dt
, dbdt
;
2195 struct lc_element
*tmp
;
2198 unsigned int c_min_rate
;
2201 c_min_rate
= rcu_dereference(mdev
->ldev
->disk_conf
)->c_min_rate
;
2204 /* feature disabled? */
2205 if (c_min_rate
== 0)
2208 spin_lock_irq(&mdev
->al_lock
);
2209 tmp
= lc_find(mdev
->resync
, BM_SECT_TO_EXT(sector
));
2211 struct bm_extent
*bm_ext
= lc_entry(tmp
, struct bm_extent
, lce
);
2212 if (test_bit(BME_PRIORITY
, &bm_ext
->flags
)) {
2213 spin_unlock_irq(&mdev
->al_lock
);
2216 /* Do not slow down if app IO is already waiting for this extent */
2218 spin_unlock_irq(&mdev
->al_lock
);
2220 curr_events
= (int)part_stat_read(&disk
->part0
, sectors
[0]) +
2221 (int)part_stat_read(&disk
->part0
, sectors
[1]) -
2222 atomic_read(&mdev
->rs_sect_ev
);
2224 if (!mdev
->rs_last_events
|| curr_events
- mdev
->rs_last_events
> 64) {
2225 unsigned long rs_left
;
2228 mdev
->rs_last_events
= curr_events
;
2230 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2232 i
= (mdev
->rs_last_mark
+ DRBD_SYNC_MARKS
-1) % DRBD_SYNC_MARKS
;
2234 if (mdev
->state
.conn
== C_VERIFY_S
|| mdev
->state
.conn
== C_VERIFY_T
)
2235 rs_left
= mdev
->ov_left
;
2237 rs_left
= drbd_bm_total_weight(mdev
) - mdev
->rs_failed
;
2239 dt
= ((long)jiffies
- (long)mdev
->rs_mark_time
[i
]) / HZ
;
2242 db
= mdev
->rs_mark_left
[i
] - rs_left
;
2243 dbdt
= Bit2KB(db
/dt
);
2245 if (dbdt
> c_min_rate
)
2252 static int receive_DataRequest(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2254 struct drbd_conf
*mdev
;
2257 struct drbd_peer_request
*peer_req
;
2258 struct digest_info
*di
= NULL
;
2260 unsigned int fault_type
;
2261 struct p_block_req
*p
= pi
->data
;
2263 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
2266 capacity
= drbd_get_capacity(mdev
->this_bdev
);
2268 sector
= be64_to_cpu(p
->sector
);
2269 size
= be32_to_cpu(p
->blksize
);
2271 if (size
<= 0 || !IS_ALIGNED(size
, 512) || size
> DRBD_MAX_BIO_SIZE
) {
2272 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2273 (unsigned long long)sector
, size
);
2276 if (sector
+ (size
>>9) > capacity
) {
2277 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2278 (unsigned long long)sector
, size
);
2282 if (!get_ldev_if_state(mdev
, D_UP_TO_DATE
)) {
2285 case P_DATA_REQUEST
:
2286 drbd_send_ack_rp(mdev
, P_NEG_DREPLY
, p
);
2288 case P_RS_DATA_REQUEST
:
2289 case P_CSUM_RS_REQUEST
:
2291 drbd_send_ack_rp(mdev
, P_NEG_RS_DREPLY
, p
);
2295 dec_rs_pending(mdev
);
2296 drbd_send_ack_ex(mdev
, P_OV_RESULT
, sector
, size
, ID_IN_SYNC
);
2301 if (verb
&& __ratelimit(&drbd_ratelimit_state
))
2302 dev_err(DEV
, "Can not satisfy peer's read request, "
2303 "no local data.\n");
2305 /* drain possibly payload */
2306 return drbd_drain_block(mdev
, pi
->size
);
2309 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2310 * "criss-cross" setup, that might cause write-out on some other DRBD,
2311 * which in turn might block on the other node at this very place. */
2312 peer_req
= drbd_alloc_peer_req(mdev
, p
->block_id
, sector
, size
, GFP_NOIO
);
2319 case P_DATA_REQUEST
:
2320 peer_req
->w
.cb
= w_e_end_data_req
;
2321 fault_type
= DRBD_FAULT_DT_RD
;
2322 /* application IO, don't drbd_rs_begin_io */
2325 case P_RS_DATA_REQUEST
:
2326 peer_req
->w
.cb
= w_e_end_rsdata_req
;
2327 fault_type
= DRBD_FAULT_RS_RD
;
2328 /* used in the sector offset progress display */
2329 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2333 case P_CSUM_RS_REQUEST
:
2334 fault_type
= DRBD_FAULT_RS_RD
;
2335 di
= kmalloc(sizeof(*di
) + pi
->size
, GFP_NOIO
);
2339 di
->digest_size
= pi
->size
;
2340 di
->digest
= (((char *)di
)+sizeof(struct digest_info
));
2342 peer_req
->digest
= di
;
2343 peer_req
->flags
|= EE_HAS_DIGEST
;
2345 if (drbd_recv_all(mdev
->tconn
, di
->digest
, pi
->size
))
2348 if (pi
->cmd
== P_CSUM_RS_REQUEST
) {
2349 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
2350 peer_req
->w
.cb
= w_e_end_csum_rs_req
;
2351 /* used in the sector offset progress display */
2352 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2353 } else if (pi
->cmd
== P_OV_REPLY
) {
2354 /* track progress, we may need to throttle */
2355 atomic_add(size
>> 9, &mdev
->rs_sect_in
);
2356 peer_req
->w
.cb
= w_e_end_ov_reply
;
2357 dec_rs_pending(mdev
);
2358 /* drbd_rs_begin_io done when we sent this request,
2359 * but accounting still needs to be done. */
2360 goto submit_for_resync
;
2365 if (mdev
->ov_start_sector
== ~(sector_t
)0 &&
2366 mdev
->tconn
->agreed_pro_version
>= 90) {
2367 unsigned long now
= jiffies
;
2369 mdev
->ov_start_sector
= sector
;
2370 mdev
->ov_position
= sector
;
2371 mdev
->ov_left
= drbd_bm_bits(mdev
) - BM_SECT_TO_BIT(sector
);
2372 mdev
->rs_total
= mdev
->ov_left
;
2373 for (i
= 0; i
< DRBD_SYNC_MARKS
; i
++) {
2374 mdev
->rs_mark_left
[i
] = mdev
->ov_left
;
2375 mdev
->rs_mark_time
[i
] = now
;
2377 dev_info(DEV
, "Online Verify start sector: %llu\n",
2378 (unsigned long long)sector
);
2380 peer_req
->w
.cb
= w_e_end_ov_req
;
2381 fault_type
= DRBD_FAULT_RS_RD
;
2388 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2389 * wrt the receiver, but it is not as straightforward as it may seem.
2390 * Various places in the resync start and stop logic assume resync
2391 * requests are processed in order, requeuing this on the worker thread
2392 * introduces a bunch of new code for synchronization between threads.
2394 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2395 * "forever", throttling after drbd_rs_begin_io will lock that extent
2396 * for application writes for the same time. For now, just throttle
2397 * here, where the rest of the code expects the receiver to sleep for
2401 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2402 * this defers syncer requests for some time, before letting at least
2403 * on request through. The resync controller on the receiving side
2404 * will adapt to the incoming rate accordingly.
2406 * We cannot throttle here if remote is Primary/SyncTarget:
2407 * we would also throttle its application reads.
2408 * In that case, throttling is done on the SyncTarget only.
2410 if (mdev
->state
.peer
!= R_PRIMARY
&& drbd_rs_should_slow_down(mdev
, sector
))
2411 schedule_timeout_uninterruptible(HZ
/10);
2412 if (drbd_rs_begin_io(mdev
, sector
))
2416 atomic_add(size
>> 9, &mdev
->rs_sect_ev
);
2420 spin_lock_irq(&mdev
->tconn
->req_lock
);
2421 list_add_tail(&peer_req
->w
.list
, &mdev
->read_ee
);
2422 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2424 if (drbd_submit_peer_request(mdev
, peer_req
, READ
, fault_type
) == 0)
2427 /* don't care for the reason here */
2428 dev_err(DEV
, "submit failed, triggering re-connect\n");
2429 spin_lock_irq(&mdev
->tconn
->req_lock
);
2430 list_del(&peer_req
->w
.list
);
2431 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2432 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2436 drbd_free_peer_req(mdev
, peer_req
);
2440 static int drbd_asb_recover_0p(struct drbd_conf
*mdev
) __must_hold(local
)
2442 int self
, peer
, rv
= -100;
2443 unsigned long ch_self
, ch_peer
;
2444 enum drbd_after_sb_p after_sb_0p
;
2446 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2447 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2449 ch_peer
= mdev
->p_uuid
[UI_SIZE
];
2450 ch_self
= mdev
->comm_bm_set
;
2453 after_sb_0p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_0p
;
2455 switch (after_sb_0p
) {
2457 case ASB_DISCARD_SECONDARY
:
2458 case ASB_CALL_HELPER
:
2460 dev_err(DEV
, "Configuration error.\n");
2462 case ASB_DISCONNECT
:
2464 case ASB_DISCARD_YOUNGER_PRI
:
2465 if (self
== 0 && peer
== 1) {
2469 if (self
== 1 && peer
== 0) {
2473 /* Else fall through to one of the other strategies... */
2474 case ASB_DISCARD_OLDER_PRI
:
2475 if (self
== 0 && peer
== 1) {
2479 if (self
== 1 && peer
== 0) {
2483 /* Else fall through to one of the other strategies... */
2484 dev_warn(DEV
, "Discard younger/older primary did not find a decision\n"
2485 "Using discard-least-changes instead\n");
2486 case ASB_DISCARD_ZERO_CHG
:
2487 if (ch_peer
== 0 && ch_self
== 0) {
2488 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
)
2492 if (ch_peer
== 0) { rv
= 1; break; }
2493 if (ch_self
== 0) { rv
= -1; break; }
2495 if (after_sb_0p
== ASB_DISCARD_ZERO_CHG
)
2497 case ASB_DISCARD_LEAST_CHG
:
2498 if (ch_self
< ch_peer
)
2500 else if (ch_self
> ch_peer
)
2502 else /* ( ch_self == ch_peer ) */
2503 /* Well, then use something else. */
2504 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
)
2507 case ASB_DISCARD_LOCAL
:
2510 case ASB_DISCARD_REMOTE
:
2517 static int drbd_asb_recover_1p(struct drbd_conf
*mdev
) __must_hold(local
)
2520 enum drbd_after_sb_p after_sb_1p
;
2523 after_sb_1p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_1p
;
2525 switch (after_sb_1p
) {
2526 case ASB_DISCARD_YOUNGER_PRI
:
2527 case ASB_DISCARD_OLDER_PRI
:
2528 case ASB_DISCARD_LEAST_CHG
:
2529 case ASB_DISCARD_LOCAL
:
2530 case ASB_DISCARD_REMOTE
:
2531 case ASB_DISCARD_ZERO_CHG
:
2532 dev_err(DEV
, "Configuration error.\n");
2534 case ASB_DISCONNECT
:
2537 hg
= drbd_asb_recover_0p(mdev
);
2538 if (hg
== -1 && mdev
->state
.role
== R_SECONDARY
)
2540 if (hg
== 1 && mdev
->state
.role
== R_PRIMARY
)
2544 rv
= drbd_asb_recover_0p(mdev
);
2546 case ASB_DISCARD_SECONDARY
:
2547 return mdev
->state
.role
== R_PRIMARY
? 1 : -1;
2548 case ASB_CALL_HELPER
:
2549 hg
= drbd_asb_recover_0p(mdev
);
2550 if (hg
== -1 && mdev
->state
.role
== R_PRIMARY
) {
2551 enum drbd_state_rv rv2
;
2553 drbd_set_role(mdev
, R_SECONDARY
, 0);
2554 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2555 * we might be here in C_WF_REPORT_PARAMS which is transient.
2556 * we do not need to wait for the after state change work either. */
2557 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2558 if (rv2
!= SS_SUCCESS
) {
2559 drbd_khelper(mdev
, "pri-lost-after-sb");
2561 dev_warn(DEV
, "Successfully gave up primary role.\n");
2571 static int drbd_asb_recover_2p(struct drbd_conf
*mdev
) __must_hold(local
)
2574 enum drbd_after_sb_p after_sb_2p
;
2577 after_sb_2p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_2p
;
2579 switch (after_sb_2p
) {
2580 case ASB_DISCARD_YOUNGER_PRI
:
2581 case ASB_DISCARD_OLDER_PRI
:
2582 case ASB_DISCARD_LEAST_CHG
:
2583 case ASB_DISCARD_LOCAL
:
2584 case ASB_DISCARD_REMOTE
:
2586 case ASB_DISCARD_SECONDARY
:
2587 case ASB_DISCARD_ZERO_CHG
:
2588 dev_err(DEV
, "Configuration error.\n");
2591 rv
= drbd_asb_recover_0p(mdev
);
2593 case ASB_DISCONNECT
:
2595 case ASB_CALL_HELPER
:
2596 hg
= drbd_asb_recover_0p(mdev
);
2598 enum drbd_state_rv rv2
;
2600 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2601 * we might be here in C_WF_REPORT_PARAMS which is transient.
2602 * we do not need to wait for the after state change work either. */
2603 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2604 if (rv2
!= SS_SUCCESS
) {
2605 drbd_khelper(mdev
, "pri-lost-after-sb");
2607 dev_warn(DEV
, "Successfully gave up primary role.\n");
2617 static void drbd_uuid_dump(struct drbd_conf
*mdev
, char *text
, u64
*uuid
,
2618 u64 bits
, u64 flags
)
2621 dev_info(DEV
, "%s uuid info vanished while I was looking!\n", text
);
2624 dev_info(DEV
, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2626 (unsigned long long)uuid
[UI_CURRENT
],
2627 (unsigned long long)uuid
[UI_BITMAP
],
2628 (unsigned long long)uuid
[UI_HISTORY_START
],
2629 (unsigned long long)uuid
[UI_HISTORY_END
],
2630 (unsigned long long)bits
,
2631 (unsigned long long)flags
);
2635 100 after split brain try auto recover
2636 2 C_SYNC_SOURCE set BitMap
2637 1 C_SYNC_SOURCE use BitMap
2639 -1 C_SYNC_TARGET use BitMap
2640 -2 C_SYNC_TARGET set BitMap
2641 -100 after split brain, disconnect
2642 -1000 unrelated data
2643 -1091 requires proto 91
2644 -1096 requires proto 96
2646 static int drbd_uuid_compare(struct drbd_conf
*mdev
, int *rule_nr
) __must_hold(local
)
2651 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2652 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2655 if (self
== UUID_JUST_CREATED
&& peer
== UUID_JUST_CREATED
)
2659 if ((self
== UUID_JUST_CREATED
|| self
== (u64
)0) &&
2660 peer
!= UUID_JUST_CREATED
)
2664 if (self
!= UUID_JUST_CREATED
&&
2665 (peer
== UUID_JUST_CREATED
|| peer
== (u64
)0))
2669 int rct
, dc
; /* roles at crash time */
2671 if (mdev
->p_uuid
[UI_BITMAP
] == (u64
)0 && mdev
->ldev
->md
.uuid
[UI_BITMAP
] != (u64
)0) {
2673 if (mdev
->tconn
->agreed_pro_version
< 91)
2676 if ((mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) &&
2677 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1))) {
2678 dev_info(DEV
, "was SyncSource, missed the resync finished event, corrected myself:\n");
2679 drbd_uuid_set_bm(mdev
, 0UL);
2681 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2682 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2685 dev_info(DEV
, "was SyncSource (peer failed to write sync_uuid)\n");
2692 if (mdev
->ldev
->md
.uuid
[UI_BITMAP
] == (u64
)0 && mdev
->p_uuid
[UI_BITMAP
] != (u64
)0) {
2694 if (mdev
->tconn
->agreed_pro_version
< 91)
2697 if ((mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1)) &&
2698 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1))) {
2699 dev_info(DEV
, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2701 mdev
->p_uuid
[UI_HISTORY_START
+ 1] = mdev
->p_uuid
[UI_HISTORY_START
];
2702 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_BITMAP
];
2703 mdev
->p_uuid
[UI_BITMAP
] = 0UL;
2705 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2708 dev_info(DEV
, "was SyncTarget (failed to write sync_uuid)\n");
2715 /* Common power [off|failure] */
2716 rct
= (test_bit(CRASHED_PRIMARY
, &mdev
->flags
) ? 1 : 0) +
2717 (mdev
->p_uuid
[UI_FLAGS
] & 2);
2718 /* lowest bit is set when we were primary,
2719 * next bit (weight 2) is set when peer was primary */
2723 case 0: /* !self_pri && !peer_pri */ return 0;
2724 case 1: /* self_pri && !peer_pri */ return 1;
2725 case 2: /* !self_pri && peer_pri */ return -1;
2726 case 3: /* self_pri && peer_pri */
2727 dc
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
);
2733 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2738 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2740 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2741 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) ==
2742 (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) :
2743 peer
+ UUID_NEW_BM_OFFSET
== (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1))) {
2744 /* The last P_SYNC_UUID did not get though. Undo the last start of
2745 resync as sync source modifications of the peer's UUIDs. */
2747 if (mdev
->tconn
->agreed_pro_version
< 91)
2750 mdev
->p_uuid
[UI_BITMAP
] = mdev
->p_uuid
[UI_HISTORY_START
];
2751 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_HISTORY_START
+ 1];
2753 dev_info(DEV
, "Did not got last syncUUID packet, corrected:\n");
2754 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2761 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2762 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2763 peer
= mdev
->p_uuid
[i
] & ~((u64
)1);
2769 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2770 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2775 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2777 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2778 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) ==
2779 (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) :
2780 self
+ UUID_NEW_BM_OFFSET
== (mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1))) {
2781 /* The last P_SYNC_UUID did not get though. Undo the last start of
2782 resync as sync source modifications of our UUIDs. */
2784 if (mdev
->tconn
->agreed_pro_version
< 91)
2787 _drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
]);
2788 _drbd_uuid_set(mdev
, UI_HISTORY_START
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1]);
2790 dev_info(DEV
, "Last syncUUID did not get through, corrected:\n");
2791 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2792 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2800 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2801 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2802 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2808 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2809 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2810 if (self
== peer
&& self
!= ((u64
)0))
2814 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2815 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2816 for (j
= UI_HISTORY_START
; j
<= UI_HISTORY_END
; j
++) {
2817 peer
= mdev
->p_uuid
[j
] & ~((u64
)1);
2826 /* drbd_sync_handshake() returns the new conn state on success, or
2827 CONN_MASK (-1) on failure.
2829 static enum drbd_conns
drbd_sync_handshake(struct drbd_conf
*mdev
, enum drbd_role peer_role
,
2830 enum drbd_disk_state peer_disk
) __must_hold(local
)
2832 enum drbd_conns rv
= C_MASK
;
2833 enum drbd_disk_state mydisk
;
2834 struct net_conf
*nc
;
2835 int hg
, rule_nr
, rr_conflict
, dry_run
;
2837 mydisk
= mdev
->state
.disk
;
2838 if (mydisk
== D_NEGOTIATING
)
2839 mydisk
= mdev
->new_state_tmp
.disk
;
2841 dev_info(DEV
, "drbd_sync_handshake:\n");
2842 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
, mdev
->comm_bm_set
, 0);
2843 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
,
2844 mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2846 hg
= drbd_uuid_compare(mdev
, &rule_nr
);
2848 dev_info(DEV
, "uuid_compare()=%d by rule %d\n", hg
, rule_nr
);
2851 dev_alert(DEV
, "Unrelated data, aborting!\n");
2855 dev_alert(DEV
, "To resolve this both sides have to support at least protocol %d\n", -hg
- 1000);
2859 if ((mydisk
== D_INCONSISTENT
&& peer_disk
> D_INCONSISTENT
) ||
2860 (peer_disk
== D_INCONSISTENT
&& mydisk
> D_INCONSISTENT
)) {
2861 int f
= (hg
== -100) || abs(hg
) == 2;
2862 hg
= mydisk
> D_INCONSISTENT
? 1 : -1;
2865 dev_info(DEV
, "Becoming sync %s due to disk states.\n",
2866 hg
> 0 ? "source" : "target");
2870 drbd_khelper(mdev
, "initial-split-brain");
2873 nc
= rcu_dereference(mdev
->tconn
->net_conf
);
2875 if (hg
== 100 || (hg
== -100 && nc
->always_asbp
)) {
2876 int pcount
= (mdev
->state
.role
== R_PRIMARY
)
2877 + (peer_role
== R_PRIMARY
);
2878 int forced
= (hg
== -100);
2882 hg
= drbd_asb_recover_0p(mdev
);
2885 hg
= drbd_asb_recover_1p(mdev
);
2888 hg
= drbd_asb_recover_2p(mdev
);
2891 if (abs(hg
) < 100) {
2892 dev_warn(DEV
, "Split-Brain detected, %d primaries, "
2893 "automatically solved. Sync from %s node\n",
2894 pcount
, (hg
< 0) ? "peer" : "this");
2896 dev_warn(DEV
, "Doing a full sync, since"
2897 " UUIDs where ambiguous.\n");
2904 if (nc
->want_lose
&& !(mdev
->p_uuid
[UI_FLAGS
]&1))
2906 if (!nc
->want_lose
&& (mdev
->p_uuid
[UI_FLAGS
]&1))
2910 dev_warn(DEV
, "Split-Brain detected, manually solved. "
2911 "Sync from %s node\n",
2912 (hg
< 0) ? "peer" : "this");
2914 rr_conflict
= nc
->rr_conflict
;
2915 dry_run
= nc
->dry_run
;
2919 /* FIXME this log message is not correct if we end up here
2920 * after an attempted attach on a diskless node.
2921 * We just refuse to attach -- well, we drop the "connection"
2922 * to that disk, in a way... */
2923 dev_alert(DEV
, "Split-Brain detected but unresolved, dropping connection!\n");
2924 drbd_khelper(mdev
, "split-brain");
2928 if (hg
> 0 && mydisk
<= D_INCONSISTENT
) {
2929 dev_err(DEV
, "I shall become SyncSource, but I am inconsistent!\n");
2933 if (hg
< 0 && /* by intention we do not use mydisk here. */
2934 mdev
->state
.role
== R_PRIMARY
&& mdev
->state
.disk
>= D_CONSISTENT
) {
2935 switch (rr_conflict
) {
2936 case ASB_CALL_HELPER
:
2937 drbd_khelper(mdev
, "pri-lost");
2939 case ASB_DISCONNECT
:
2940 dev_err(DEV
, "I shall become SyncTarget, but I am primary!\n");
2943 dev_warn(DEV
, "Becoming SyncTarget, violating the stable-data"
2948 if (dry_run
|| test_bit(CONN_DRY_RUN
, &mdev
->tconn
->flags
)) {
2950 dev_info(DEV
, "dry-run connect: No resync, would become Connected immediately.\n");
2952 dev_info(DEV
, "dry-run connect: Would become %s, doing a %s resync.",
2953 drbd_conn_str(hg
> 0 ? C_SYNC_SOURCE
: C_SYNC_TARGET
),
2954 abs(hg
) >= 2 ? "full" : "bit-map based");
2959 dev_info(DEV
, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2960 if (drbd_bitmap_io(mdev
, &drbd_bmio_set_n_write
, "set_n_write from sync_handshake",
2961 BM_LOCKED_SET_ALLOWED
))
2965 if (hg
> 0) { /* become sync source. */
2967 } else if (hg
< 0) { /* become sync target */
2971 if (drbd_bm_total_weight(mdev
)) {
2972 dev_info(DEV
, "No resync, but %lu bits in bitmap!\n",
2973 drbd_bm_total_weight(mdev
));
2980 /* returns 1 if invalid */
2981 static int cmp_after_sb(enum drbd_after_sb_p peer
, enum drbd_after_sb_p self
)
2983 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2984 if ((peer
== ASB_DISCARD_REMOTE
&& self
== ASB_DISCARD_LOCAL
) ||
2985 (self
== ASB_DISCARD_REMOTE
&& peer
== ASB_DISCARD_LOCAL
))
2988 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2989 if (peer
== ASB_DISCARD_REMOTE
|| peer
== ASB_DISCARD_LOCAL
||
2990 self
== ASB_DISCARD_REMOTE
|| self
== ASB_DISCARD_LOCAL
)
2993 /* everything else is valid if they are equal on both sides. */
2997 /* everything es is invalid. */
3001 static int receive_protocol(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3003 struct p_protocol
*p
= pi
->data
;
3004 int p_proto
, p_after_sb_0p
, p_after_sb_1p
, p_after_sb_2p
;
3005 int p_want_lose
, p_two_primaries
, cf
;
3006 struct net_conf
*nc
;
3008 p_proto
= be32_to_cpu(p
->protocol
);
3009 p_after_sb_0p
= be32_to_cpu(p
->after_sb_0p
);
3010 p_after_sb_1p
= be32_to_cpu(p
->after_sb_1p
);
3011 p_after_sb_2p
= be32_to_cpu(p
->after_sb_2p
);
3012 p_two_primaries
= be32_to_cpu(p
->two_primaries
);
3013 cf
= be32_to_cpu(p
->conn_flags
);
3014 p_want_lose
= cf
& CF_WANT_LOSE
;
3016 if (tconn
->agreed_pro_version
>= 87) {
3017 char integrity_alg
[SHARED_SECRET_MAX
];
3018 struct crypto_hash
*tfm
= NULL
;
3021 if (pi
->size
> sizeof(integrity_alg
))
3023 err
= drbd_recv_all(tconn
, integrity_alg
, pi
->size
);
3026 integrity_alg
[SHARED_SECRET_MAX
-1] = 0;
3028 if (integrity_alg
[0]) {
3029 tfm
= crypto_alloc_hash(integrity_alg
, 0, CRYPTO_ALG_ASYNC
);
3031 conn_err(tconn
, "peer data-integrity-alg %s not supported\n",
3035 conn_info(tconn
, "peer data-integrity-alg: %s\n", integrity_alg
);
3038 if (tconn
->peer_integrity_tfm
)
3039 crypto_free_hash(tconn
->peer_integrity_tfm
);
3040 tconn
->peer_integrity_tfm
= tfm
;
3043 clear_bit(CONN_DRY_RUN
, &tconn
->flags
);
3045 if (cf
& CF_DRY_RUN
)
3046 set_bit(CONN_DRY_RUN
, &tconn
->flags
);
3049 nc
= rcu_dereference(tconn
->net_conf
);
3051 if (p_proto
!= nc
->wire_protocol
&& tconn
->agreed_pro_version
< 100) {
3052 conn_err(tconn
, "incompatible communication protocols\n");
3053 goto disconnect_rcu_unlock
;
3056 if (cmp_after_sb(p_after_sb_0p
, nc
->after_sb_0p
)) {
3057 conn_err(tconn
, "incompatible after-sb-0pri settings\n");
3058 goto disconnect_rcu_unlock
;
3061 if (cmp_after_sb(p_after_sb_1p
, nc
->after_sb_1p
)) {
3062 conn_err(tconn
, "incompatible after-sb-1pri settings\n");
3063 goto disconnect_rcu_unlock
;
3066 if (cmp_after_sb(p_after_sb_2p
, nc
->after_sb_2p
)) {
3067 conn_err(tconn
, "incompatible after-sb-2pri settings\n");
3068 goto disconnect_rcu_unlock
;
3071 if (p_want_lose
&& nc
->want_lose
) {
3072 conn_err(tconn
, "both sides have the 'want_lose' flag set\n");
3073 goto disconnect_rcu_unlock
;
3076 if (p_two_primaries
!= nc
->two_primaries
) {
3077 conn_err(tconn
, "incompatible setting of the two-primaries options\n");
3078 goto disconnect_rcu_unlock
;
3085 disconnect_rcu_unlock
:
3088 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3093 * input: alg name, feature name
3094 * return: NULL (alg name was "")
3095 * ERR_PTR(error) if something goes wrong
3096 * or the crypto hash ptr, if it worked out ok. */
3097 struct crypto_hash
*drbd_crypto_alloc_digest_safe(const struct drbd_conf
*mdev
,
3098 const char *alg
, const char *name
)
3100 struct crypto_hash
*tfm
;
3105 tfm
= crypto_alloc_hash(alg
, 0, CRYPTO_ALG_ASYNC
);
3107 dev_err(DEV
, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3108 alg
, name
, PTR_ERR(tfm
));
3114 static int ignore_remaining_packet(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3116 void *buffer
= tconn
->data
.rbuf
;
3117 int size
= pi
->size
;
3120 int s
= min_t(int, size
, DRBD_SOCKET_BUFFER_SIZE
);
3121 s
= drbd_recv(tconn
, buffer
, s
);
3135 * config_unknown_volume - device configuration command for unknown volume
3137 * When a device is added to an existing connection, the node on which the
3138 * device is added first will send configuration commands to its peer but the
3139 * peer will not know about the device yet. It will warn and ignore these
3140 * commands. Once the device is added on the second node, the second node will
3141 * send the same device configuration commands, but in the other direction.
3143 * (We can also end up here if drbd is misconfigured.)
3145 static int config_unknown_volume(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3147 conn_warn(tconn
, "Volume %u unknown; ignoring %s packet\n",
3148 pi
->vnr
, cmdname(pi
->cmd
));
3149 return ignore_remaining_packet(tconn
, pi
);
3152 static int receive_SyncParam(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3154 struct drbd_conf
*mdev
;
3155 struct p_rs_param_95
*p
;
3156 unsigned int header_size
, data_size
, exp_max_sz
;
3157 struct crypto_hash
*verify_tfm
= NULL
;
3158 struct crypto_hash
*csums_tfm
= NULL
;
3159 struct net_conf
*old_net_conf
, *new_net_conf
= NULL
;
3160 struct disk_conf
*old_disk_conf
= NULL
, *new_disk_conf
= NULL
;
3161 const int apv
= tconn
->agreed_pro_version
;
3162 struct fifo_buffer
*old_plan
= NULL
, *new_plan
= NULL
;
3166 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3168 return config_unknown_volume(tconn
, pi
);
3170 exp_max_sz
= apv
<= 87 ? sizeof(struct p_rs_param
)
3171 : apv
== 88 ? sizeof(struct p_rs_param
)
3173 : apv
<= 94 ? sizeof(struct p_rs_param_89
)
3174 : /* apv >= 95 */ sizeof(struct p_rs_param_95
);
3176 if (pi
->size
> exp_max_sz
) {
3177 dev_err(DEV
, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3178 pi
->size
, exp_max_sz
);
3183 header_size
= sizeof(struct p_rs_param
);
3184 data_size
= pi
->size
- header_size
;
3185 } else if (apv
<= 94) {
3186 header_size
= sizeof(struct p_rs_param_89
);
3187 data_size
= pi
->size
- header_size
;
3188 D_ASSERT(data_size
== 0);
3190 header_size
= sizeof(struct p_rs_param_95
);
3191 data_size
= pi
->size
- header_size
;
3192 D_ASSERT(data_size
== 0);
3195 /* initialize verify_alg and csums_alg */
3197 memset(p
->verify_alg
, 0, 2 * SHARED_SECRET_MAX
);
3199 err
= drbd_recv_all(mdev
->tconn
, p
, header_size
);
3203 mutex_lock(&mdev
->tconn
->conf_update
);
3204 old_net_conf
= mdev
->tconn
->net_conf
;
3205 if (get_ldev(mdev
)) {
3206 new_disk_conf
= kzalloc(sizeof(struct disk_conf
), GFP_KERNEL
);
3207 if (!new_disk_conf
) {
3209 mutex_unlock(&mdev
->tconn
->conf_update
);
3210 dev_err(DEV
, "Allocation of new disk_conf failed\n");
3214 old_disk_conf
= mdev
->ldev
->disk_conf
;
3215 *new_disk_conf
= *old_disk_conf
;
3217 new_disk_conf
->resync_rate
= be32_to_cpu(p
->rate
);
3222 if (data_size
> SHARED_SECRET_MAX
) {
3223 dev_err(DEV
, "verify-alg too long, "
3224 "peer wants %u, accepting only %u byte\n",
3225 data_size
, SHARED_SECRET_MAX
);
3230 err
= drbd_recv_all(mdev
->tconn
, p
->verify_alg
, data_size
);
3233 /* we expect NUL terminated string */
3234 /* but just in case someone tries to be evil */
3235 D_ASSERT(p
->verify_alg
[data_size
-1] == 0);
3236 p
->verify_alg
[data_size
-1] = 0;
3238 } else /* apv >= 89 */ {
3239 /* we still expect NUL terminated strings */
3240 /* but just in case someone tries to be evil */
3241 D_ASSERT(p
->verify_alg
[SHARED_SECRET_MAX
-1] == 0);
3242 D_ASSERT(p
->csums_alg
[SHARED_SECRET_MAX
-1] == 0);
3243 p
->verify_alg
[SHARED_SECRET_MAX
-1] = 0;
3244 p
->csums_alg
[SHARED_SECRET_MAX
-1] = 0;
3247 if (strcmp(old_net_conf
->verify_alg
, p
->verify_alg
)) {
3248 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
3249 dev_err(DEV
, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3250 old_net_conf
->verify_alg
, p
->verify_alg
);
3253 verify_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
3254 p
->verify_alg
, "verify-alg");
3255 if (IS_ERR(verify_tfm
)) {
3261 if (apv
>= 89 && strcmp(old_net_conf
->csums_alg
, p
->csums_alg
)) {
3262 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
3263 dev_err(DEV
, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3264 old_net_conf
->csums_alg
, p
->csums_alg
);
3267 csums_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
3268 p
->csums_alg
, "csums-alg");
3269 if (IS_ERR(csums_tfm
)) {
3275 if (apv
> 94 && new_disk_conf
) {
3276 new_disk_conf
->c_plan_ahead
= be32_to_cpu(p
->c_plan_ahead
);
3277 new_disk_conf
->c_delay_target
= be32_to_cpu(p
->c_delay_target
);
3278 new_disk_conf
->c_fill_target
= be32_to_cpu(p
->c_fill_target
);
3279 new_disk_conf
->c_max_rate
= be32_to_cpu(p
->c_max_rate
);
3281 fifo_size
= (new_disk_conf
->c_plan_ahead
* 10 * SLEEP_TIME
) / HZ
;
3282 if (fifo_size
!= mdev
->rs_plan_s
->size
) {
3283 new_plan
= fifo_alloc(fifo_size
);
3285 dev_err(DEV
, "kmalloc of fifo_buffer failed");
3292 if (verify_tfm
|| csums_tfm
) {
3293 new_net_conf
= kzalloc(sizeof(struct net_conf
), GFP_KERNEL
);
3294 if (!new_net_conf
) {
3295 dev_err(DEV
, "Allocation of new net_conf failed\n");
3299 *new_net_conf
= *old_net_conf
;
3302 strcpy(new_net_conf
->verify_alg
, p
->verify_alg
);
3303 new_net_conf
->verify_alg_len
= strlen(p
->verify_alg
) + 1;
3304 crypto_free_hash(mdev
->tconn
->verify_tfm
);
3305 mdev
->tconn
->verify_tfm
= verify_tfm
;
3306 dev_info(DEV
, "using verify-alg: \"%s\"\n", p
->verify_alg
);
3309 strcpy(new_net_conf
->csums_alg
, p
->csums_alg
);
3310 new_net_conf
->csums_alg_len
= strlen(p
->csums_alg
) + 1;
3311 crypto_free_hash(mdev
->tconn
->csums_tfm
);
3312 mdev
->tconn
->csums_tfm
= csums_tfm
;
3313 dev_info(DEV
, "using csums-alg: \"%s\"\n", p
->csums_alg
);
3315 rcu_assign_pointer(tconn
->net_conf
, new_net_conf
);
3319 if (new_disk_conf
) {
3320 rcu_assign_pointer(mdev
->ldev
->disk_conf
, new_disk_conf
);
3325 old_plan
= mdev
->rs_plan_s
;
3326 rcu_assign_pointer(mdev
->rs_plan_s
, new_plan
);
3329 mutex_unlock(&mdev
->tconn
->conf_update
);
3332 kfree(old_net_conf
);
3333 kfree(old_disk_conf
);
3339 if (new_disk_conf
) {
3341 kfree(new_disk_conf
);
3343 mutex_unlock(&mdev
->tconn
->conf_update
);
3348 if (new_disk_conf
) {
3350 kfree(new_disk_conf
);
3352 mutex_unlock(&mdev
->tconn
->conf_update
);
3353 /* just for completeness: actually not needed,
3354 * as this is not reached if csums_tfm was ok. */
3355 crypto_free_hash(csums_tfm
);
3356 /* but free the verify_tfm again, if csums_tfm did not work out */
3357 crypto_free_hash(verify_tfm
);
3358 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3362 /* warn if the arguments differ by more than 12.5% */
3363 static void warn_if_differ_considerably(struct drbd_conf
*mdev
,
3364 const char *s
, sector_t a
, sector_t b
)
3367 if (a
== 0 || b
== 0)
3369 d
= (a
> b
) ? (a
- b
) : (b
- a
);
3370 if (d
> (a
>>3) || d
> (b
>>3))
3371 dev_warn(DEV
, "Considerable difference in %s: %llus vs. %llus\n", s
,
3372 (unsigned long long)a
, (unsigned long long)b
);
3375 static int receive_sizes(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3377 struct drbd_conf
*mdev
;
3378 struct p_sizes
*p
= pi
->data
;
3379 enum determine_dev_size dd
= unchanged
;
3380 sector_t p_size
, p_usize
, my_usize
;
3381 int ldsc
= 0; /* local disk size changed */
3382 enum dds_flags ddsf
;
3384 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3386 return config_unknown_volume(tconn
, pi
);
3388 p_size
= be64_to_cpu(p
->d_size
);
3389 p_usize
= be64_to_cpu(p
->u_size
);
3391 /* just store the peer's disk size for now.
3392 * we still need to figure out whether we accept that. */
3393 mdev
->p_size
= p_size
;
3395 if (get_ldev(mdev
)) {
3397 my_usize
= rcu_dereference(mdev
->ldev
->disk_conf
)->disk_size
;
3400 warn_if_differ_considerably(mdev
, "lower level device sizes",
3401 p_size
, drbd_get_max_capacity(mdev
->ldev
));
3402 warn_if_differ_considerably(mdev
, "user requested size",
3405 /* if this is the first connect, or an otherwise expected
3406 * param exchange, choose the minimum */
3407 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
)
3408 p_usize
= min_not_zero(my_usize
, p_usize
);
3410 /* Never shrink a device with usable data during connect.
3411 But allow online shrinking if we are connected. */
3412 if (drbd_new_dev_size(mdev
, mdev
->ldev
, p_usize
, 0) <
3413 drbd_get_capacity(mdev
->this_bdev
) &&
3414 mdev
->state
.disk
>= D_OUTDATED
&&
3415 mdev
->state
.conn
< C_CONNECTED
) {
3416 dev_err(DEV
, "The peer's disk size is too small!\n");
3417 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3422 if (my_usize
!= p_usize
) {
3423 struct disk_conf
*old_disk_conf
, *new_disk_conf
= NULL
;
3425 new_disk_conf
= kzalloc(sizeof(struct disk_conf
), GFP_KERNEL
);
3426 if (!new_disk_conf
) {
3427 dev_err(DEV
, "Allocation of new disk_conf failed\n");
3432 mutex_lock(&mdev
->tconn
->conf_update
);
3433 old_disk_conf
= mdev
->ldev
->disk_conf
;
3434 *new_disk_conf
= *old_disk_conf
;
3435 new_disk_conf
->disk_size
= p_usize
;
3437 rcu_assign_pointer(mdev
->ldev
->disk_conf
, new_disk_conf
);
3438 mutex_unlock(&mdev
->tconn
->conf_update
);
3440 kfree(old_disk_conf
);
3442 dev_info(DEV
, "Peer sets u_size to %lu sectors\n",
3443 (unsigned long)my_usize
);
3449 ddsf
= be16_to_cpu(p
->dds_flags
);
3450 if (get_ldev(mdev
)) {
3451 dd
= drbd_determine_dev_size(mdev
, ddsf
);
3453 if (dd
== dev_size_error
)
3457 /* I am diskless, need to accept the peer's size. */
3458 drbd_set_my_capacity(mdev
, p_size
);
3461 mdev
->peer_max_bio_size
= be32_to_cpu(p
->max_bio_size
);
3462 drbd_reconsider_max_bio_size(mdev
);
3464 if (get_ldev(mdev
)) {
3465 if (mdev
->ldev
->known_size
!= drbd_get_capacity(mdev
->ldev
->backing_bdev
)) {
3466 mdev
->ldev
->known_size
= drbd_get_capacity(mdev
->ldev
->backing_bdev
);
3473 if (mdev
->state
.conn
> C_WF_REPORT_PARAMS
) {
3474 if (be64_to_cpu(p
->c_size
) !=
3475 drbd_get_capacity(mdev
->this_bdev
) || ldsc
) {
3476 /* we have different sizes, probably peer
3477 * needs to know my new size... */
3478 drbd_send_sizes(mdev
, 0, ddsf
);
3480 if (test_and_clear_bit(RESIZE_PENDING
, &mdev
->flags
) ||
3481 (dd
== grew
&& mdev
->state
.conn
== C_CONNECTED
)) {
3482 if (mdev
->state
.pdsk
>= D_INCONSISTENT
&&
3483 mdev
->state
.disk
>= D_INCONSISTENT
) {
3484 if (ddsf
& DDSF_NO_RESYNC
)
3485 dev_info(DEV
, "Resync of new storage suppressed with --assume-clean\n");
3487 resync_after_online_grow(mdev
);
3489 set_bit(RESYNC_AFTER_NEG
, &mdev
->flags
);
3496 static int receive_uuids(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3498 struct drbd_conf
*mdev
;
3499 struct p_uuids
*p
= pi
->data
;
3501 int i
, updated_uuids
= 0;
3503 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3505 return config_unknown_volume(tconn
, pi
);
3507 p_uuid
= kmalloc(sizeof(u64
)*UI_EXTENDED_SIZE
, GFP_NOIO
);
3509 for (i
= UI_CURRENT
; i
< UI_EXTENDED_SIZE
; i
++)
3510 p_uuid
[i
] = be64_to_cpu(p
->uuid
[i
]);
3512 kfree(mdev
->p_uuid
);
3513 mdev
->p_uuid
= p_uuid
;
3515 if (mdev
->state
.conn
< C_CONNECTED
&&
3516 mdev
->state
.disk
< D_INCONSISTENT
&&
3517 mdev
->state
.role
== R_PRIMARY
&&
3518 (mdev
->ed_uuid
& ~((u64
)1)) != (p_uuid
[UI_CURRENT
] & ~((u64
)1))) {
3519 dev_err(DEV
, "Can only connect to data with current UUID=%016llX\n",
3520 (unsigned long long)mdev
->ed_uuid
);
3521 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3525 if (get_ldev(mdev
)) {
3526 int skip_initial_sync
=
3527 mdev
->state
.conn
== C_CONNECTED
&&
3528 mdev
->tconn
->agreed_pro_version
>= 90 &&
3529 mdev
->ldev
->md
.uuid
[UI_CURRENT
] == UUID_JUST_CREATED
&&
3530 (p_uuid
[UI_FLAGS
] & 8);
3531 if (skip_initial_sync
) {
3532 dev_info(DEV
, "Accepted new current UUID, preparing to skip initial sync\n");
3533 drbd_bitmap_io(mdev
, &drbd_bmio_clear_n_write
,
3534 "clear_n_write from receive_uuids",
3535 BM_LOCKED_TEST_ALLOWED
);
3536 _drbd_uuid_set(mdev
, UI_CURRENT
, p_uuid
[UI_CURRENT
]);
3537 _drbd_uuid_set(mdev
, UI_BITMAP
, 0);
3538 _drbd_set_state(_NS2(mdev
, disk
, D_UP_TO_DATE
, pdsk
, D_UP_TO_DATE
),
3544 } else if (mdev
->state
.disk
< D_INCONSISTENT
&&
3545 mdev
->state
.role
== R_PRIMARY
) {
3546 /* I am a diskless primary, the peer just created a new current UUID
3548 updated_uuids
= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3551 /* Before we test for the disk state, we should wait until an eventually
3552 ongoing cluster wide state change is finished. That is important if
3553 we are primary and are detaching from our disk. We need to see the
3554 new disk state... */
3555 mutex_lock(mdev
->state_mutex
);
3556 mutex_unlock(mdev
->state_mutex
);
3557 if (mdev
->state
.conn
>= C_CONNECTED
&& mdev
->state
.disk
< D_INCONSISTENT
)
3558 updated_uuids
|= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3561 drbd_print_uuids(mdev
, "receiver updated UUIDs to");
3567 * convert_state() - Converts the peer's view of the cluster state to our point of view
3568 * @ps: The state as seen by the peer.
3570 static union drbd_state
convert_state(union drbd_state ps
)
3572 union drbd_state ms
;
3574 static enum drbd_conns c_tab
[] = {
3575 [C_CONNECTED
] = C_CONNECTED
,
3577 [C_STARTING_SYNC_S
] = C_STARTING_SYNC_T
,
3578 [C_STARTING_SYNC_T
] = C_STARTING_SYNC_S
,
3579 [C_DISCONNECTING
] = C_TEAR_DOWN
, /* C_NETWORK_FAILURE, */
3580 [C_VERIFY_S
] = C_VERIFY_T
,
3586 ms
.conn
= c_tab
[ps
.conn
];
3591 ms
.peer_isp
= (ps
.aftr_isp
| ps
.user_isp
);
3596 static int receive_req_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3598 struct drbd_conf
*mdev
;
3599 struct p_req_state
*p
= pi
->data
;
3600 union drbd_state mask
, val
;
3601 enum drbd_state_rv rv
;
3603 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3607 mask
.i
= be32_to_cpu(p
->mask
);
3608 val
.i
= be32_to_cpu(p
->val
);
3610 if (test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
) &&
3611 mutex_is_locked(mdev
->state_mutex
)) {
3612 drbd_send_sr_reply(mdev
, SS_CONCURRENT_ST_CHG
);
3616 mask
= convert_state(mask
);
3617 val
= convert_state(val
);
3619 rv
= drbd_change_state(mdev
, CS_VERBOSE
, mask
, val
);
3620 drbd_send_sr_reply(mdev
, rv
);
3627 static int receive_req_conn_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3629 struct p_req_state
*p
= pi
->data
;
3630 union drbd_state mask
, val
;
3631 enum drbd_state_rv rv
;
3633 mask
.i
= be32_to_cpu(p
->mask
);
3634 val
.i
= be32_to_cpu(p
->val
);
3636 if (test_bit(DISCARD_CONCURRENT
, &tconn
->flags
) &&
3637 mutex_is_locked(&tconn
->cstate_mutex
)) {
3638 conn_send_sr_reply(tconn
, SS_CONCURRENT_ST_CHG
);
3642 mask
= convert_state(mask
);
3643 val
= convert_state(val
);
3645 rv
= conn_request_state(tconn
, mask
, val
, CS_VERBOSE
| CS_LOCAL_ONLY
| CS_IGN_OUTD_FAIL
);
3646 conn_send_sr_reply(tconn
, rv
);
3651 static int receive_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3653 struct drbd_conf
*mdev
;
3654 struct p_state
*p
= pi
->data
;
3655 union drbd_state os
, ns
, peer_state
;
3656 enum drbd_disk_state real_peer_disk
;
3657 enum chg_state_flags cs_flags
;
3660 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3662 return config_unknown_volume(tconn
, pi
);
3664 peer_state
.i
= be32_to_cpu(p
->state
);
3666 real_peer_disk
= peer_state
.disk
;
3667 if (peer_state
.disk
== D_NEGOTIATING
) {
3668 real_peer_disk
= mdev
->p_uuid
[UI_FLAGS
] & 4 ? D_INCONSISTENT
: D_CONSISTENT
;
3669 dev_info(DEV
, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk
));
3672 spin_lock_irq(&mdev
->tconn
->req_lock
);
3674 os
= ns
= drbd_read_state(mdev
);
3675 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3677 /* peer says his disk is uptodate, while we think it is inconsistent,
3678 * and this happens while we think we have a sync going on. */
3679 if (os
.pdsk
== D_INCONSISTENT
&& real_peer_disk
== D_UP_TO_DATE
&&
3680 os
.conn
> C_CONNECTED
&& os
.disk
== D_UP_TO_DATE
) {
3681 /* If we are (becoming) SyncSource, but peer is still in sync
3682 * preparation, ignore its uptodate-ness to avoid flapping, it
3683 * will change to inconsistent once the peer reaches active
3685 * It may have changed syncer-paused flags, however, so we
3686 * cannot ignore this completely. */
3687 if (peer_state
.conn
> C_CONNECTED
&&
3688 peer_state
.conn
< C_SYNC_SOURCE
)
3689 real_peer_disk
= D_INCONSISTENT
;
3691 /* if peer_state changes to connected at the same time,
3692 * it explicitly notifies us that it finished resync.
3693 * Maybe we should finish it up, too? */
3694 else if (os
.conn
>= C_SYNC_SOURCE
&&
3695 peer_state
.conn
== C_CONNECTED
) {
3696 if (drbd_bm_total_weight(mdev
) <= mdev
->rs_failed
)
3697 drbd_resync_finished(mdev
);
3702 /* peer says his disk is inconsistent, while we think it is uptodate,
3703 * and this happens while the peer still thinks we have a sync going on,
3704 * but we think we are already done with the sync.
3705 * We ignore this to avoid flapping pdsk.
3706 * This should not happen, if the peer is a recent version of drbd. */
3707 if (os
.pdsk
== D_UP_TO_DATE
&& real_peer_disk
== D_INCONSISTENT
&&
3708 os
.conn
== C_CONNECTED
&& peer_state
.conn
> C_SYNC_SOURCE
)
3709 real_peer_disk
= D_UP_TO_DATE
;
3711 if (ns
.conn
== C_WF_REPORT_PARAMS
)
3712 ns
.conn
= C_CONNECTED
;
3714 if (peer_state
.conn
== C_AHEAD
)
3717 if (mdev
->p_uuid
&& peer_state
.disk
>= D_NEGOTIATING
&&
3718 get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3719 int cr
; /* consider resync */
3721 /* if we established a new connection */
3722 cr
= (os
.conn
< C_CONNECTED
);
3723 /* if we had an established connection
3724 * and one of the nodes newly attaches a disk */
3725 cr
|= (os
.conn
== C_CONNECTED
&&
3726 (peer_state
.disk
== D_NEGOTIATING
||
3727 os
.disk
== D_NEGOTIATING
));
3728 /* if we have both been inconsistent, and the peer has been
3729 * forced to be UpToDate with --overwrite-data */
3730 cr
|= test_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3731 /* if we had been plain connected, and the admin requested to
3732 * start a sync by "invalidate" or "invalidate-remote" */
3733 cr
|= (os
.conn
== C_CONNECTED
&&
3734 (peer_state
.conn
>= C_STARTING_SYNC_S
&&
3735 peer_state
.conn
<= C_WF_BITMAP_T
));
3738 ns
.conn
= drbd_sync_handshake(mdev
, peer_state
.role
, real_peer_disk
);
3741 if (ns
.conn
== C_MASK
) {
3742 ns
.conn
= C_CONNECTED
;
3743 if (mdev
->state
.disk
== D_NEGOTIATING
) {
3744 drbd_force_state(mdev
, NS(disk
, D_FAILED
));
3745 } else if (peer_state
.disk
== D_NEGOTIATING
) {
3746 dev_err(DEV
, "Disk attach process on the peer node was aborted.\n");
3747 peer_state
.disk
= D_DISKLESS
;
3748 real_peer_disk
= D_DISKLESS
;
3750 if (test_and_clear_bit(CONN_DRY_RUN
, &mdev
->tconn
->flags
))
3752 D_ASSERT(os
.conn
== C_WF_REPORT_PARAMS
);
3753 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3759 spin_lock_irq(&mdev
->tconn
->req_lock
);
3760 if (os
.i
!= drbd_read_state(mdev
).i
)
3762 clear_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3763 ns
.peer
= peer_state
.role
;
3764 ns
.pdsk
= real_peer_disk
;
3765 ns
.peer_isp
= (peer_state
.aftr_isp
| peer_state
.user_isp
);
3766 if ((ns
.conn
== C_CONNECTED
|| ns
.conn
== C_WF_BITMAP_S
) && ns
.disk
== D_NEGOTIATING
)
3767 ns
.disk
= mdev
->new_state_tmp
.disk
;
3768 cs_flags
= CS_VERBOSE
+ (os
.conn
< C_CONNECTED
&& ns
.conn
>= C_CONNECTED
? 0 : CS_HARD
);
3769 if (ns
.pdsk
== D_CONSISTENT
&& drbd_suspended(mdev
) && ns
.conn
== C_CONNECTED
&& os
.conn
< C_CONNECTED
&&
3770 test_bit(NEW_CUR_UUID
, &mdev
->flags
)) {
3771 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
3772 for temporal network outages! */
3773 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3774 dev_err(DEV
, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3775 tl_clear(mdev
->tconn
);
3776 drbd_uuid_new_current(mdev
);
3777 clear_bit(NEW_CUR_UUID
, &mdev
->flags
);
3778 conn_request_state(mdev
->tconn
, NS2(conn
, C_PROTOCOL_ERROR
, susp
, 0), CS_HARD
);
3781 rv
= _drbd_set_state(mdev
, ns
, cs_flags
, NULL
);
3782 ns
= drbd_read_state(mdev
);
3783 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3785 if (rv
< SS_SUCCESS
) {
3786 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3790 if (os
.conn
> C_WF_REPORT_PARAMS
) {
3791 if (ns
.conn
> C_CONNECTED
&& peer_state
.conn
<= C_CONNECTED
&&
3792 peer_state
.disk
!= D_NEGOTIATING
) {
3793 /* we want resync, peer has not yet decided to sync... */
3794 /* Nowadays only used when forcing a node into primary role and
3795 setting its disk to UpToDate with that */
3796 drbd_send_uuids(mdev
);
3797 drbd_send_state(mdev
);
3801 mutex_lock(&mdev
->tconn
->conf_update
);
3802 mdev
->tconn
->net_conf
->want_lose
= 0; /* without copy; single bit op is atomic */
3803 mutex_unlock(&mdev
->tconn
->conf_update
);
3805 drbd_md_sync(mdev
); /* update connected indicator, la_size, ... */
3810 static int receive_sync_uuid(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3812 struct drbd_conf
*mdev
;
3813 struct p_rs_uuid
*p
= pi
->data
;
3815 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3819 wait_event(mdev
->misc_wait
,
3820 mdev
->state
.conn
== C_WF_SYNC_UUID
||
3821 mdev
->state
.conn
== C_BEHIND
||
3822 mdev
->state
.conn
< C_CONNECTED
||
3823 mdev
->state
.disk
< D_NEGOTIATING
);
3825 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3827 /* Here the _drbd_uuid_ functions are right, current should
3828 _not_ be rotated into the history */
3829 if (get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3830 _drbd_uuid_set(mdev
, UI_CURRENT
, be64_to_cpu(p
->uuid
));
3831 _drbd_uuid_set(mdev
, UI_BITMAP
, 0UL);
3833 drbd_print_uuids(mdev
, "updated sync uuid");
3834 drbd_start_resync(mdev
, C_SYNC_TARGET
);
3838 dev_err(DEV
, "Ignoring SyncUUID packet!\n");
3844 * receive_bitmap_plain
3846 * Return 0 when done, 1 when another iteration is needed, and a negative error
3847 * code upon failure.
3850 receive_bitmap_plain(struct drbd_conf
*mdev
, unsigned int size
,
3851 unsigned long *p
, struct bm_xfer_ctx
*c
)
3853 unsigned int data_size
= DRBD_SOCKET_BUFFER_SIZE
-
3854 drbd_header_size(mdev
->tconn
);
3855 unsigned int num_words
= min_t(size_t, data_size
/ sizeof(*p
),
3856 c
->bm_words
- c
->word_offset
);
3857 unsigned int want
= num_words
* sizeof(*p
);
3861 dev_err(DEV
, "%s:want (%u) != size (%u)\n", __func__
, want
, size
);
3866 err
= drbd_recv_all(mdev
->tconn
, p
, want
);
3870 drbd_bm_merge_lel(mdev
, c
->word_offset
, num_words
, p
);
3872 c
->word_offset
+= num_words
;
3873 c
->bit_offset
= c
->word_offset
* BITS_PER_LONG
;
3874 if (c
->bit_offset
> c
->bm_bits
)
3875 c
->bit_offset
= c
->bm_bits
;
3880 static enum drbd_bitmap_code
dcbp_get_code(struct p_compressed_bm
*p
)
3882 return (enum drbd_bitmap_code
)(p
->encoding
& 0x0f);
3885 static int dcbp_get_start(struct p_compressed_bm
*p
)
3887 return (p
->encoding
& 0x80) != 0;
3890 static int dcbp_get_pad_bits(struct p_compressed_bm
*p
)
3892 return (p
->encoding
>> 4) & 0x7;
3898 * Return 0 when done, 1 when another iteration is needed, and a negative error
3899 * code upon failure.
3902 recv_bm_rle_bits(struct drbd_conf
*mdev
,
3903 struct p_compressed_bm
*p
,
3904 struct bm_xfer_ctx
*c
,
3907 struct bitstream bs
;
3911 unsigned long s
= c
->bit_offset
;
3913 int toggle
= dcbp_get_start(p
);
3917 bitstream_init(&bs
, p
->code
, len
, dcbp_get_pad_bits(p
));
3919 bits
= bitstream_get_bits(&bs
, &look_ahead
, 64);
3923 for (have
= bits
; have
> 0; s
+= rl
, toggle
= !toggle
) {
3924 bits
= vli_decode_bits(&rl
, look_ahead
);
3930 if (e
>= c
->bm_bits
) {
3931 dev_err(DEV
, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e
);
3934 _drbd_bm_set_bits(mdev
, s
, e
);
3938 dev_err(DEV
, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3939 have
, bits
, look_ahead
,
3940 (unsigned int)(bs
.cur
.b
- p
->code
),
3941 (unsigned int)bs
.buf_len
);
3944 look_ahead
>>= bits
;
3947 bits
= bitstream_get_bits(&bs
, &tmp
, 64 - have
);
3950 look_ahead
|= tmp
<< have
;
3955 bm_xfer_ctx_bit_to_word_offset(c
);
3957 return (s
!= c
->bm_bits
);
3963 * Return 0 when done, 1 when another iteration is needed, and a negative error
3964 * code upon failure.
3967 decode_bitmap_c(struct drbd_conf
*mdev
,
3968 struct p_compressed_bm
*p
,
3969 struct bm_xfer_ctx
*c
,
3972 if (dcbp_get_code(p
) == RLE_VLI_Bits
)
3973 return recv_bm_rle_bits(mdev
, p
, c
, len
- sizeof(*p
));
3975 /* other variants had been implemented for evaluation,
3976 * but have been dropped as this one turned out to be "best"
3977 * during all our tests. */
3979 dev_err(DEV
, "receive_bitmap_c: unknown encoding %u\n", p
->encoding
);
3980 conn_request_state(mdev
->tconn
, NS(conn
, C_PROTOCOL_ERROR
), CS_HARD
);
3984 void INFO_bm_xfer_stats(struct drbd_conf
*mdev
,
3985 const char *direction
, struct bm_xfer_ctx
*c
)
3987 /* what would it take to transfer it "plaintext" */
3988 unsigned int header_size
= drbd_header_size(mdev
->tconn
);
3989 unsigned int data_size
= DRBD_SOCKET_BUFFER_SIZE
- header_size
;
3990 unsigned int plain
=
3991 header_size
* (DIV_ROUND_UP(c
->bm_words
, data_size
) + 1) +
3992 c
->bm_words
* sizeof(unsigned long);
3993 unsigned int total
= c
->bytes
[0] + c
->bytes
[1];
3996 /* total can not be zero. but just in case: */
4000 /* don't report if not compressed */
4004 /* total < plain. check for overflow, still */
4005 r
= (total
> UINT_MAX
/1000) ? (total
/ (plain
/1000))
4006 : (1000 * total
/ plain
);
4012 dev_info(DEV
, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4013 "total %u; compression: %u.%u%%\n",
4015 c
->bytes
[1], c
->packets
[1],
4016 c
->bytes
[0], c
->packets
[0],
4017 total
, r
/10, r
% 10);
4020 /* Since we are processing the bitfield from lower addresses to higher,
4021 it does not matter if the process it in 32 bit chunks or 64 bit
4022 chunks as long as it is little endian. (Understand it as byte stream,
4023 beginning with the lowest byte...) If we would use big endian
4024 we would need to process it from the highest address to the lowest,
4025 in order to be agnostic to the 32 vs 64 bits issue.
4027 returns 0 on failure, 1 if we successfully received it. */
4028 static int receive_bitmap(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4030 struct drbd_conf
*mdev
;
4031 struct bm_xfer_ctx c
;
4034 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4038 drbd_bm_lock(mdev
, "receive bitmap", BM_LOCKED_SET_ALLOWED
);
4039 /* you are supposed to send additional out-of-sync information
4040 * if you actually set bits during this phase */
4042 c
= (struct bm_xfer_ctx
) {
4043 .bm_bits
= drbd_bm_bits(mdev
),
4044 .bm_words
= drbd_bm_words(mdev
),
4048 if (pi
->cmd
== P_BITMAP
)
4049 err
= receive_bitmap_plain(mdev
, pi
->size
, pi
->data
, &c
);
4050 else if (pi
->cmd
== P_COMPRESSED_BITMAP
) {
4051 /* MAYBE: sanity check that we speak proto >= 90,
4052 * and the feature is enabled! */
4053 struct p_compressed_bm
*p
= pi
->data
;
4055 if (pi
->size
> DRBD_SOCKET_BUFFER_SIZE
- drbd_header_size(tconn
)) {
4056 dev_err(DEV
, "ReportCBitmap packet too large\n");
4060 if (pi
->size
<= sizeof(*p
)) {
4061 dev_err(DEV
, "ReportCBitmap packet too small (l:%u)\n", pi
->size
);
4065 err
= drbd_recv_all(mdev
->tconn
, p
, pi
->size
);
4068 err
= decode_bitmap_c(mdev
, p
, &c
, pi
->size
);
4070 dev_warn(DEV
, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi
->cmd
);
4075 c
.packets
[pi
->cmd
== P_BITMAP
]++;
4076 c
.bytes
[pi
->cmd
== P_BITMAP
] += drbd_header_size(tconn
) + pi
->size
;
4083 err
= drbd_recv_header(mdev
->tconn
, pi
);
4088 INFO_bm_xfer_stats(mdev
, "receive", &c
);
4090 if (mdev
->state
.conn
== C_WF_BITMAP_T
) {
4091 enum drbd_state_rv rv
;
4093 err
= drbd_send_bitmap(mdev
);
4096 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4097 rv
= _drbd_request_state(mdev
, NS(conn
, C_WF_SYNC_UUID
), CS_VERBOSE
);
4098 D_ASSERT(rv
== SS_SUCCESS
);
4099 } else if (mdev
->state
.conn
!= C_WF_BITMAP_S
) {
4100 /* admin may have requested C_DISCONNECTING,
4101 * other threads may have noticed network errors */
4102 dev_info(DEV
, "unexpected cstate (%s) in receive_bitmap\n",
4103 drbd_conn_str(mdev
->state
.conn
));
4108 drbd_bm_unlock(mdev
);
4109 if (!err
&& mdev
->state
.conn
== C_WF_BITMAP_S
)
4110 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
4114 static int receive_skip(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4116 conn_warn(tconn
, "skipping unknown optional packet type %d, l: %d!\n",
4119 return ignore_remaining_packet(tconn
, pi
);
4122 static int receive_UnplugRemote(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4124 /* Make sure we've acked all the TCP data associated
4125 * with the data requests being unplugged */
4126 drbd_tcp_quickack(tconn
->data
.socket
);
4131 static int receive_out_of_sync(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4133 struct drbd_conf
*mdev
;
4134 struct p_block_desc
*p
= pi
->data
;
4136 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4140 switch (mdev
->state
.conn
) {
4141 case C_WF_SYNC_UUID
:
4146 dev_err(DEV
, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4147 drbd_conn_str(mdev
->state
.conn
));
4150 drbd_set_out_of_sync(mdev
, be64_to_cpu(p
->sector
), be32_to_cpu(p
->blksize
));
4158 int (*fn
)(struct drbd_tconn
*, struct packet_info
*);
4161 static struct data_cmd drbd_cmd_handler
[] = {
4162 [P_DATA
] = { 1, sizeof(struct p_data
), receive_Data
},
4163 [P_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_DataReply
},
4164 [P_RS_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_RSDataReply
} ,
4165 [P_BARRIER
] = { 0, sizeof(struct p_barrier
), receive_Barrier
} ,
4166 [P_BITMAP
] = { 1, 0, receive_bitmap
} ,
4167 [P_COMPRESSED_BITMAP
] = { 1, 0, receive_bitmap
} ,
4168 [P_UNPLUG_REMOTE
] = { 0, 0, receive_UnplugRemote
},
4169 [P_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4170 [P_RS_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4171 [P_SYNC_PARAM
] = { 1, 0, receive_SyncParam
},
4172 [P_SYNC_PARAM89
] = { 1, 0, receive_SyncParam
},
4173 [P_PROTOCOL
] = { 1, sizeof(struct p_protocol
), receive_protocol
},
4174 [P_UUIDS
] = { 0, sizeof(struct p_uuids
), receive_uuids
},
4175 [P_SIZES
] = { 0, sizeof(struct p_sizes
), receive_sizes
},
4176 [P_STATE
] = { 0, sizeof(struct p_state
), receive_state
},
4177 [P_STATE_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_state
},
4178 [P_SYNC_UUID
] = { 0, sizeof(struct p_rs_uuid
), receive_sync_uuid
},
4179 [P_OV_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4180 [P_OV_REPLY
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
4181 [P_CSUM_RS_REQUEST
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
4182 [P_DELAY_PROBE
] = { 0, sizeof(struct p_delay_probe93
), receive_skip
},
4183 [P_OUT_OF_SYNC
] = { 0, sizeof(struct p_block_desc
), receive_out_of_sync
},
4184 [P_CONN_ST_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_conn_state
},
4187 static void drbdd(struct drbd_tconn
*tconn
)
4189 struct packet_info pi
;
4190 size_t shs
; /* sub header size */
4193 while (get_t_state(&tconn
->receiver
) == RUNNING
) {
4194 struct data_cmd
*cmd
;
4196 drbd_thread_current_set_cpu(&tconn
->receiver
);
4197 if (drbd_recv_header(tconn
, &pi
))
4200 cmd
= &drbd_cmd_handler
[pi
.cmd
];
4201 if (unlikely(pi
.cmd
>= ARRAY_SIZE(drbd_cmd_handler
) || !cmd
->fn
)) {
4202 conn_err(tconn
, "unknown packet type %d, l: %d!\n", pi
.cmd
, pi
.size
);
4206 shs
= cmd
->pkt_size
;
4207 if (pi
.size
> shs
&& !cmd
->expect_payload
) {
4208 conn_err(tconn
, "No payload expected %s l:%d\n", cmdname(pi
.cmd
), pi
.size
);
4213 err
= drbd_recv_all_warn(tconn
, pi
.data
, shs
);
4219 err
= cmd
->fn(tconn
, &pi
);
4221 conn_err(tconn
, "error receiving %s, e: %d l: %d!\n",
4222 cmdname(pi
.cmd
), err
, pi
.size
);
4229 conn_request_state(tconn
, NS(conn
, C_PROTOCOL_ERROR
), CS_HARD
);
4232 void conn_flush_workqueue(struct drbd_tconn
*tconn
)
4234 struct drbd_wq_barrier barr
;
4236 barr
.w
.cb
= w_prev_work_done
;
4237 barr
.w
.tconn
= tconn
;
4238 init_completion(&barr
.done
);
4239 drbd_queue_work(&tconn
->data
.work
, &barr
.w
);
4240 wait_for_completion(&barr
.done
);
4243 static void drbd_disconnect(struct drbd_tconn
*tconn
)
4246 int rv
= SS_UNKNOWN_ERROR
;
4248 if (tconn
->cstate
== C_STANDALONE
)
4251 /* asender does not clean up anything. it must not interfere, either */
4252 drbd_thread_stop(&tconn
->asender
);
4253 drbd_free_sock(tconn
);
4255 down_read(&drbd_cfg_rwsem
);
4256 idr_for_each(&tconn
->volumes
, drbd_disconnected
, tconn
);
4257 up_read(&drbd_cfg_rwsem
);
4258 conn_info(tconn
, "Connection closed\n");
4260 if (conn_highest_role(tconn
) == R_PRIMARY
&& conn_highest_pdsk(tconn
) >= D_UNKNOWN
)
4261 conn_try_outdate_peer_async(tconn
);
4263 spin_lock_irq(&tconn
->req_lock
);
4265 if (oc
>= C_UNCONNECTED
)
4266 rv
= _conn_request_state(tconn
, NS(conn
, C_UNCONNECTED
), CS_VERBOSE
);
4268 spin_unlock_irq(&tconn
->req_lock
);
4270 if (oc
== C_DISCONNECTING
)
4271 conn_request_state(tconn
, NS(conn
, C_STANDALONE
), CS_VERBOSE
| CS_HARD
);
4274 static int drbd_disconnected(int vnr
, void *p
, void *data
)
4276 struct drbd_conf
*mdev
= (struct drbd_conf
*)p
;
4277 enum drbd_fencing_p fp
;
4280 /* wait for current activity to cease. */
4281 spin_lock_irq(&mdev
->tconn
->req_lock
);
4282 _drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
4283 _drbd_wait_ee_list_empty(mdev
, &mdev
->sync_ee
);
4284 _drbd_wait_ee_list_empty(mdev
, &mdev
->read_ee
);
4285 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4287 /* We do not have data structures that would allow us to
4288 * get the rs_pending_cnt down to 0 again.
4289 * * On C_SYNC_TARGET we do not have any data structures describing
4290 * the pending RSDataRequest's we have sent.
4291 * * On C_SYNC_SOURCE there is no data structure that tracks
4292 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
4293 * And no, it is not the sum of the reference counts in the
4294 * resync_LRU. The resync_LRU tracks the whole operation including
4295 * the disk-IO, while the rs_pending_cnt only tracks the blocks
4297 drbd_rs_cancel_all(mdev
);
4299 mdev
->rs_failed
= 0;
4300 atomic_set(&mdev
->rs_pending_cnt
, 0);
4301 wake_up(&mdev
->misc_wait
);
4303 del_timer(&mdev
->request_timer
);
4305 del_timer_sync(&mdev
->resync_timer
);
4306 resync_timer_fn((unsigned long)mdev
);
4308 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
4309 * w_make_resync_request etc. which may still be on the worker queue
4310 * to be "canceled" */
4311 drbd_flush_workqueue(mdev
);
4313 drbd_finish_peer_reqs(mdev
);
4315 kfree(mdev
->p_uuid
);
4316 mdev
->p_uuid
= NULL
;
4318 if (!drbd_suspended(mdev
))
4319 tl_clear(mdev
->tconn
);
4324 if (get_ldev(mdev
)) {
4326 fp
= rcu_dereference(mdev
->ldev
->disk_conf
)->fencing
;
4331 /* serialize with bitmap writeout triggered by the state change,
4333 wait_event(mdev
->misc_wait
, !test_bit(BITMAP_IO
, &mdev
->flags
));
4335 /* tcp_close and release of sendpage pages can be deferred. I don't
4336 * want to use SO_LINGER, because apparently it can be deferred for
4337 * more than 20 seconds (longest time I checked).
4339 * Actually we don't care for exactly when the network stack does its
4340 * put_page(), but release our reference on these pages right here.
4342 i
= drbd_free_peer_reqs(mdev
, &mdev
->net_ee
);
4344 dev_info(DEV
, "net_ee not empty, killed %u entries\n", i
);
4345 i
= atomic_read(&mdev
->pp_in_use_by_net
);
4347 dev_info(DEV
, "pp_in_use_by_net = %d, expected 0\n", i
);
4348 i
= atomic_read(&mdev
->pp_in_use
);
4350 dev_info(DEV
, "pp_in_use = %d, expected 0\n", i
);
4352 D_ASSERT(list_empty(&mdev
->read_ee
));
4353 D_ASSERT(list_empty(&mdev
->active_ee
));
4354 D_ASSERT(list_empty(&mdev
->sync_ee
));
4355 D_ASSERT(list_empty(&mdev
->done_ee
));
4357 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
4358 atomic_set(&mdev
->current_epoch
->epoch_size
, 0);
4359 D_ASSERT(list_empty(&mdev
->current_epoch
->list
));
4365 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
4366 * we can agree on is stored in agreed_pro_version.
4368 * feature flags and the reserved array should be enough room for future
4369 * enhancements of the handshake protocol, and possible plugins...
4371 * for now, they are expected to be zero, but ignored.
4373 static int drbd_send_features(struct drbd_tconn
*tconn
)
4375 struct drbd_socket
*sock
;
4376 struct p_connection_features
*p
;
4378 sock
= &tconn
->data
;
4379 p
= conn_prepare_command(tconn
, sock
);
4382 memset(p
, 0, sizeof(*p
));
4383 p
->protocol_min
= cpu_to_be32(PRO_VERSION_MIN
);
4384 p
->protocol_max
= cpu_to_be32(PRO_VERSION_MAX
);
4385 return conn_send_command(tconn
, sock
, P_CONNECTION_FEATURES
, sizeof(*p
), NULL
, 0);
4390 * 1 yes, we have a valid connection
4391 * 0 oops, did not work out, please try again
4392 * -1 peer talks different language,
4393 * no point in trying again, please go standalone.
4395 static int drbd_do_features(struct drbd_tconn
*tconn
)
4397 /* ASSERT current == tconn->receiver ... */
4398 struct p_connection_features
*p
;
4399 const int expect
= sizeof(struct p_connection_features
);
4400 struct packet_info pi
;
4403 err
= drbd_send_features(tconn
);
4407 err
= drbd_recv_header(tconn
, &pi
);
4411 if (pi
.cmd
!= P_CONNECTION_FEATURES
) {
4412 conn_err(tconn
, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
4413 cmdname(pi
.cmd
), pi
.cmd
);
4417 if (pi
.size
!= expect
) {
4418 conn_err(tconn
, "expected ConnectionFeatures length: %u, received: %u\n",
4424 err
= drbd_recv_all_warn(tconn
, p
, expect
);
4428 p
->protocol_min
= be32_to_cpu(p
->protocol_min
);
4429 p
->protocol_max
= be32_to_cpu(p
->protocol_max
);
4430 if (p
->protocol_max
== 0)
4431 p
->protocol_max
= p
->protocol_min
;
4433 if (PRO_VERSION_MAX
< p
->protocol_min
||
4434 PRO_VERSION_MIN
> p
->protocol_max
)
4437 tconn
->agreed_pro_version
= min_t(int, PRO_VERSION_MAX
, p
->protocol_max
);
4439 conn_info(tconn
, "Handshake successful: "
4440 "Agreed network protocol version %d\n", tconn
->agreed_pro_version
);
4445 conn_err(tconn
, "incompatible DRBD dialects: "
4446 "I support %d-%d, peer supports %d-%d\n",
4447 PRO_VERSION_MIN
, PRO_VERSION_MAX
,
4448 p
->protocol_min
, p
->protocol_max
);
4452 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4453 static int drbd_do_auth(struct drbd_tconn
*tconn
)
4455 dev_err(DEV
, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4456 dev_err(DEV
, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4460 #define CHALLENGE_LEN 64
4464 0 - failed, try again (network error),
4465 -1 - auth failed, don't try again.
4468 static int drbd_do_auth(struct drbd_tconn
*tconn
)
4470 struct drbd_socket
*sock
;
4471 char my_challenge
[CHALLENGE_LEN
]; /* 64 Bytes... */
4472 struct scatterlist sg
;
4473 char *response
= NULL
;
4474 char *right_response
= NULL
;
4475 char *peers_ch
= NULL
;
4476 unsigned int key_len
;
4477 char secret
[SHARED_SECRET_MAX
]; /* 64 byte */
4478 unsigned int resp_size
;
4479 struct hash_desc desc
;
4480 struct packet_info pi
;
4481 struct net_conf
*nc
;
4484 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
4487 nc
= rcu_dereference(tconn
->net_conf
);
4488 key_len
= strlen(nc
->shared_secret
);
4489 memcpy(secret
, nc
->shared_secret
, key_len
);
4492 desc
.tfm
= tconn
->cram_hmac_tfm
;
4495 rv
= crypto_hash_setkey(tconn
->cram_hmac_tfm
, (u8
*)secret
, key_len
);
4497 conn_err(tconn
, "crypto_hash_setkey() failed with %d\n", rv
);
4502 get_random_bytes(my_challenge
, CHALLENGE_LEN
);
4504 sock
= &tconn
->data
;
4505 if (!conn_prepare_command(tconn
, sock
)) {
4509 rv
= !conn_send_command(tconn
, sock
, P_AUTH_CHALLENGE
, 0,
4510 my_challenge
, CHALLENGE_LEN
);
4514 err
= drbd_recv_header(tconn
, &pi
);
4520 if (pi
.cmd
!= P_AUTH_CHALLENGE
) {
4521 conn_err(tconn
, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4522 cmdname(pi
.cmd
), pi
.cmd
);
4527 if (pi
.size
> CHALLENGE_LEN
* 2) {
4528 conn_err(tconn
, "expected AuthChallenge payload too big.\n");
4533 peers_ch
= kmalloc(pi
.size
, GFP_NOIO
);
4534 if (peers_ch
== NULL
) {
4535 conn_err(tconn
, "kmalloc of peers_ch failed\n");
4540 err
= drbd_recv_all_warn(tconn
, peers_ch
, pi
.size
);
4546 resp_size
= crypto_hash_digestsize(tconn
->cram_hmac_tfm
);
4547 response
= kmalloc(resp_size
, GFP_NOIO
);
4548 if (response
== NULL
) {
4549 conn_err(tconn
, "kmalloc of response failed\n");
4554 sg_init_table(&sg
, 1);
4555 sg_set_buf(&sg
, peers_ch
, pi
.size
);
4557 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, response
);
4559 conn_err(tconn
, "crypto_hash_digest() failed with %d\n", rv
);
4564 if (!conn_prepare_command(tconn
, sock
)) {
4568 rv
= !conn_send_command(tconn
, sock
, P_AUTH_RESPONSE
, 0,
4569 response
, resp_size
);
4573 err
= drbd_recv_header(tconn
, &pi
);
4579 if (pi
.cmd
!= P_AUTH_RESPONSE
) {
4580 conn_err(tconn
, "expected AuthResponse packet, received: %s (0x%04x)\n",
4581 cmdname(pi
.cmd
), pi
.cmd
);
4586 if (pi
.size
!= resp_size
) {
4587 conn_err(tconn
, "expected AuthResponse payload of wrong size\n");
4592 err
= drbd_recv_all_warn(tconn
, response
, resp_size
);
4598 right_response
= kmalloc(resp_size
, GFP_NOIO
);
4599 if (right_response
== NULL
) {
4600 conn_err(tconn
, "kmalloc of right_response failed\n");
4605 sg_set_buf(&sg
, my_challenge
, CHALLENGE_LEN
);
4607 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, right_response
);
4609 conn_err(tconn
, "crypto_hash_digest() failed with %d\n", rv
);
4614 rv
= !memcmp(response
, right_response
, resp_size
);
4617 conn_info(tconn
, "Peer authenticated using %d bytes HMAC\n",
4625 kfree(right_response
);
4631 int drbdd_init(struct drbd_thread
*thi
)
4633 struct drbd_tconn
*tconn
= thi
->tconn
;
4636 conn_info(tconn
, "receiver (re)started\n");
4639 h
= drbd_connect(tconn
);
4641 drbd_disconnect(tconn
);
4642 schedule_timeout_interruptible(HZ
);
4645 conn_warn(tconn
, "Discarding network configuration.\n");
4646 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
4653 drbd_disconnect(tconn
);
4655 conn_info(tconn
, "receiver terminated\n");
4659 /* ********* acknowledge sender ******** */
4661 static int got_conn_RqSReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4663 struct p_req_state_reply
*p
= pi
->data
;
4664 int retcode
= be32_to_cpu(p
->retcode
);
4666 if (retcode
>= SS_SUCCESS
) {
4667 set_bit(CONN_WD_ST_CHG_OKAY
, &tconn
->flags
);
4669 set_bit(CONN_WD_ST_CHG_FAIL
, &tconn
->flags
);
4670 conn_err(tconn
, "Requested state change failed by peer: %s (%d)\n",
4671 drbd_set_st_err_str(retcode
), retcode
);
4673 wake_up(&tconn
->ping_wait
);
4678 static int got_RqSReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4680 struct drbd_conf
*mdev
;
4681 struct p_req_state_reply
*p
= pi
->data
;
4682 int retcode
= be32_to_cpu(p
->retcode
);
4684 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4688 if (retcode
>= SS_SUCCESS
) {
4689 set_bit(CL_ST_CHG_SUCCESS
, &mdev
->flags
);
4691 set_bit(CL_ST_CHG_FAIL
, &mdev
->flags
);
4692 dev_err(DEV
, "Requested state change failed by peer: %s (%d)\n",
4693 drbd_set_st_err_str(retcode
), retcode
);
4695 wake_up(&mdev
->state_wait
);
4700 static int got_Ping(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4702 return drbd_send_ping_ack(tconn
);
4706 static int got_PingAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4708 /* restore idle timeout */
4709 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= tconn
->net_conf
->ping_int
*HZ
;
4710 if (!test_and_set_bit(GOT_PING_ACK
, &tconn
->flags
))
4711 wake_up(&tconn
->ping_wait
);
4716 static int got_IsInSync(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4718 struct drbd_conf
*mdev
;
4719 struct p_block_ack
*p
= pi
->data
;
4720 sector_t sector
= be64_to_cpu(p
->sector
);
4721 int blksize
= be32_to_cpu(p
->blksize
);
4723 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4727 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
4729 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4731 if (get_ldev(mdev
)) {
4732 drbd_rs_complete_io(mdev
, sector
);
4733 drbd_set_in_sync(mdev
, sector
, blksize
);
4734 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4735 mdev
->rs_same_csum
+= (blksize
>> BM_BLOCK_SHIFT
);
4738 dec_rs_pending(mdev
);
4739 atomic_add(blksize
>> 9, &mdev
->rs_sect_in
);
4745 validate_req_change_req_state(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
4746 struct rb_root
*root
, const char *func
,
4747 enum drbd_req_event what
, bool missing_ok
)
4749 struct drbd_request
*req
;
4750 struct bio_and_error m
;
4752 spin_lock_irq(&mdev
->tconn
->req_lock
);
4753 req
= find_request(mdev
, root
, id
, sector
, missing_ok
, func
);
4754 if (unlikely(!req
)) {
4755 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4758 __req_mod(req
, what
, &m
);
4759 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4762 complete_master_bio(mdev
, &m
);
4766 static int got_BlockAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4768 struct drbd_conf
*mdev
;
4769 struct p_block_ack
*p
= pi
->data
;
4770 sector_t sector
= be64_to_cpu(p
->sector
);
4771 int blksize
= be32_to_cpu(p
->blksize
);
4772 enum drbd_req_event what
;
4774 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4778 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4780 if (p
->block_id
== ID_SYNCER
) {
4781 drbd_set_in_sync(mdev
, sector
, blksize
);
4782 dec_rs_pending(mdev
);
4786 case P_RS_WRITE_ACK
:
4787 what
= WRITE_ACKED_BY_PEER_AND_SIS
;
4790 what
= WRITE_ACKED_BY_PEER
;
4793 what
= RECV_ACKED_BY_PEER
;
4795 case P_DISCARD_WRITE
:
4796 what
= DISCARD_WRITE
;
4799 what
= POSTPONE_WRITE
;
4805 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4806 &mdev
->write_requests
, __func__
,
4810 static int got_NegAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4812 struct drbd_conf
*mdev
;
4813 struct p_block_ack
*p
= pi
->data
;
4814 sector_t sector
= be64_to_cpu(p
->sector
);
4815 int size
= be32_to_cpu(p
->blksize
);
4818 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4822 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4824 if (p
->block_id
== ID_SYNCER
) {
4825 dec_rs_pending(mdev
);
4826 drbd_rs_failed_io(mdev
, sector
, size
);
4830 err
= validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4831 &mdev
->write_requests
, __func__
,
4834 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
4835 The master bio might already be completed, therefore the
4836 request is no longer in the collision hash. */
4837 /* In Protocol B we might already have got a P_RECV_ACK
4838 but then get a P_NEG_ACK afterwards. */
4839 drbd_set_out_of_sync(mdev
, sector
, size
);
4844 static int got_NegDReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4846 struct drbd_conf
*mdev
;
4847 struct p_block_ack
*p
= pi
->data
;
4848 sector_t sector
= be64_to_cpu(p
->sector
);
4850 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4854 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4856 dev_err(DEV
, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4857 (unsigned long long)sector
, be32_to_cpu(p
->blksize
));
4859 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4860 &mdev
->read_requests
, __func__
,
4864 static int got_NegRSDReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4866 struct drbd_conf
*mdev
;
4869 struct p_block_ack
*p
= pi
->data
;
4871 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4875 sector
= be64_to_cpu(p
->sector
);
4876 size
= be32_to_cpu(p
->blksize
);
4878 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4880 dec_rs_pending(mdev
);
4882 if (get_ldev_if_state(mdev
, D_FAILED
)) {
4883 drbd_rs_complete_io(mdev
, sector
);
4885 case P_NEG_RS_DREPLY
:
4886 drbd_rs_failed_io(mdev
, sector
, size
);
4898 static int got_BarrierAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4900 struct drbd_conf
*mdev
;
4901 struct p_barrier_ack
*p
= pi
->data
;
4903 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4907 tl_release(mdev
->tconn
, p
->barrier
, be32_to_cpu(p
->set_size
));
4909 if (mdev
->state
.conn
== C_AHEAD
&&
4910 atomic_read(&mdev
->ap_in_flight
) == 0 &&
4911 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE
, &mdev
->current_epoch
->flags
)) {
4912 mdev
->start_resync_timer
.expires
= jiffies
+ HZ
;
4913 add_timer(&mdev
->start_resync_timer
);
4919 static int got_OVResult(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4921 struct drbd_conf
*mdev
;
4922 struct p_block_ack
*p
= pi
->data
;
4923 struct drbd_work
*w
;
4927 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4931 sector
= be64_to_cpu(p
->sector
);
4932 size
= be32_to_cpu(p
->blksize
);
4934 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4936 if (be64_to_cpu(p
->block_id
) == ID_OUT_OF_SYNC
)
4937 drbd_ov_out_of_sync_found(mdev
, sector
, size
);
4939 ov_out_of_sync_print(mdev
);
4941 if (!get_ldev(mdev
))
4944 drbd_rs_complete_io(mdev
, sector
);
4945 dec_rs_pending(mdev
);
4949 /* let's advance progress step marks only for every other megabyte */
4950 if ((mdev
->ov_left
& 0x200) == 0x200)
4951 drbd_advance_rs_marks(mdev
, mdev
->ov_left
);
4953 if (mdev
->ov_left
== 0) {
4954 w
= kmalloc(sizeof(*w
), GFP_NOIO
);
4956 w
->cb
= w_ov_finished
;
4958 drbd_queue_work_front(&mdev
->tconn
->data
.work
, w
);
4960 dev_err(DEV
, "kmalloc(w) failed.");
4961 ov_out_of_sync_print(mdev
);
4962 drbd_resync_finished(mdev
);
4969 static int got_skip(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4974 static int tconn_finish_peer_reqs(struct drbd_tconn
*tconn
)
4976 struct drbd_conf
*mdev
;
4977 int i
, not_empty
= 0;
4980 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
4981 flush_signals(current
);
4982 down_read(&drbd_cfg_rwsem
);
4983 idr_for_each_entry(&tconn
->volumes
, mdev
, i
) {
4984 if (drbd_finish_peer_reqs(mdev
)) {
4985 up_read(&drbd_cfg_rwsem
);
4986 return 1; /* error */
4989 up_read(&drbd_cfg_rwsem
);
4990 set_bit(SIGNAL_ASENDER
, &tconn
->flags
);
4992 spin_lock_irq(&tconn
->req_lock
);
4994 idr_for_each_entry(&tconn
->volumes
, mdev
, i
) {
4995 not_empty
= !list_empty(&mdev
->done_ee
);
5000 spin_unlock_irq(&tconn
->req_lock
);
5001 } while (not_empty
);
5006 struct asender_cmd
{
5008 int (*fn
)(struct drbd_tconn
*tconn
, struct packet_info
*);
5011 static struct asender_cmd asender_tbl
[] = {
5012 [P_PING
] = { 0, got_Ping
},
5013 [P_PING_ACK
] = { 0, got_PingAck
},
5014 [P_RECV_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5015 [P_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5016 [P_RS_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5017 [P_DISCARD_WRITE
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5018 [P_NEG_ACK
] = { sizeof(struct p_block_ack
), got_NegAck
},
5019 [P_NEG_DREPLY
] = { sizeof(struct p_block_ack
), got_NegDReply
},
5020 [P_NEG_RS_DREPLY
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
5021 [P_OV_RESULT
] = { sizeof(struct p_block_ack
), got_OVResult
},
5022 [P_BARRIER_ACK
] = { sizeof(struct p_barrier_ack
), got_BarrierAck
},
5023 [P_STATE_CHG_REPLY
] = { sizeof(struct p_req_state_reply
), got_RqSReply
},
5024 [P_RS_IS_IN_SYNC
] = { sizeof(struct p_block_ack
), got_IsInSync
},
5025 [P_DELAY_PROBE
] = { sizeof(struct p_delay_probe93
), got_skip
},
5026 [P_RS_CANCEL
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
5027 [P_CONN_ST_CHG_REPLY
]={ sizeof(struct p_req_state_reply
), got_conn_RqSReply
},
5028 [P_RETRY_WRITE
] = { sizeof(struct p_block_ack
), got_BlockAck
},
5031 int drbd_asender(struct drbd_thread
*thi
)
5033 struct drbd_tconn
*tconn
= thi
->tconn
;
5034 struct asender_cmd
*cmd
= NULL
;
5035 struct packet_info pi
;
5037 void *buf
= tconn
->meta
.rbuf
;
5039 unsigned int header_size
= drbd_header_size(tconn
);
5040 int expect
= header_size
;
5041 bool ping_timeout_active
= false;
5042 struct net_conf
*nc
;
5043 int ping_timeo
, no_cork
, ping_int
;
5045 current
->policy
= SCHED_RR
; /* Make this a realtime task! */
5046 current
->rt_priority
= 2; /* more important than all other tasks */
5048 while (get_t_state(thi
) == RUNNING
) {
5049 drbd_thread_current_set_cpu(thi
);
5052 nc
= rcu_dereference(tconn
->net_conf
);
5053 ping_timeo
= nc
->ping_timeo
;
5054 no_cork
= nc
->no_cork
;
5055 ping_int
= nc
->ping_int
;
5058 if (test_and_clear_bit(SEND_PING
, &tconn
->flags
)) {
5059 if (drbd_send_ping(tconn
)) {
5060 conn_err(tconn
, "drbd_send_ping has failed\n");
5063 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= ping_timeo
* HZ
/ 10;
5064 ping_timeout_active
= true;
5067 /* TODO: conditionally cork; it may hurt latency if we cork without
5070 drbd_tcp_cork(tconn
->meta
.socket
);
5071 if (tconn_finish_peer_reqs(tconn
)) {
5072 conn_err(tconn
, "tconn_finish_peer_reqs() failed\n");
5075 /* but unconditionally uncork unless disabled */
5077 drbd_tcp_uncork(tconn
->meta
.socket
);
5079 /* short circuit, recv_msg would return EINTR anyways. */
5080 if (signal_pending(current
))
5083 rv
= drbd_recv_short(tconn
->meta
.socket
, buf
, expect
-received
, 0);
5084 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5086 flush_signals(current
);
5089 * -EINTR (on meta) we got a signal
5090 * -EAGAIN (on meta) rcvtimeo expired
5091 * -ECONNRESET other side closed the connection
5092 * -ERESTARTSYS (on data) we got a signal
5093 * rv < 0 other than above: unexpected error!
5094 * rv == expected: full header or command
5095 * rv < expected: "woken" by signal during receive
5096 * rv == 0 : "connection shut down by peer"
5098 if (likely(rv
> 0)) {
5101 } else if (rv
== 0) {
5102 conn_err(tconn
, "meta connection shut down by peer.\n");
5104 } else if (rv
== -EAGAIN
) {
5105 /* If the data socket received something meanwhile,
5106 * that is good enough: peer is still alive. */
5107 if (time_after(tconn
->last_received
,
5108 jiffies
- tconn
->meta
.socket
->sk
->sk_rcvtimeo
))
5110 if (ping_timeout_active
) {
5111 conn_err(tconn
, "PingAck did not arrive in time.\n");
5114 set_bit(SEND_PING
, &tconn
->flags
);
5116 } else if (rv
== -EINTR
) {
5119 conn_err(tconn
, "sock_recvmsg returned %d\n", rv
);
5123 if (received
== expect
&& cmd
== NULL
) {
5124 if (decode_header(tconn
, tconn
->meta
.rbuf
, &pi
))
5126 cmd
= &asender_tbl
[pi
.cmd
];
5127 if (pi
.cmd
>= ARRAY_SIZE(asender_tbl
) || !cmd
->fn
) {
5128 conn_err(tconn
, "unknown command %d on meta (l: %d)\n",
5132 expect
= header_size
+ cmd
->pkt_size
;
5133 if (pi
.size
!= expect
- header_size
) {
5134 conn_err(tconn
, "Wrong packet size on meta (c: %d, l: %d)\n",
5139 if (received
== expect
) {
5142 err
= cmd
->fn(tconn
, &pi
);
5144 conn_err(tconn
, "%pf failed\n", cmd
->fn
);
5148 tconn
->last_received
= jiffies
;
5150 if (cmd
== &asender_tbl
[P_PING_ACK
]) {
5151 /* restore idle timeout */
5152 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= ping_int
* HZ
;
5153 ping_timeout_active
= false;
5156 buf
= tconn
->meta
.rbuf
;
5158 expect
= header_size
;
5165 conn_request_state(tconn
, NS(conn
, C_NETWORK_FAILURE
), CS_HARD
);
5169 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
5171 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5173 conn_info(tconn
, "asender terminated\n");