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 enum write_ordering_e pwo
;
1170 static char *write_ordering_str
[] = {
1172 [WO_drain_io
] = "drain",
1173 [WO_bdev_flush
] = "flush",
1176 pwo
= mdev
->write_ordering
;
1178 if (wo
== WO_bdev_flush
&& mdev
->ldev
->dc
.no_disk_flush
)
1180 if (wo
== WO_drain_io
&& mdev
->ldev
->dc
.no_disk_drain
)
1182 mdev
->write_ordering
= wo
;
1183 if (pwo
!= mdev
->write_ordering
|| wo
== WO_bdev_flush
)
1184 dev_info(DEV
, "Method to ensure write ordering: %s\n", write_ordering_str
[mdev
->write_ordering
]);
1188 * drbd_submit_peer_request()
1189 * @mdev: DRBD device.
1190 * @peer_req: peer request
1191 * @rw: flag field, see bio->bi_rw
1193 * May spread the pages to multiple bios,
1194 * depending on bio_add_page restrictions.
1196 * Returns 0 if all bios have been submitted,
1197 * -ENOMEM if we could not allocate enough bios,
1198 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1199 * single page to an empty bio (which should never happen and likely indicates
1200 * that the lower level IO stack is in some way broken). This has been observed
1201 * on certain Xen deployments.
1203 /* TODO allocate from our own bio_set. */
1204 int drbd_submit_peer_request(struct drbd_conf
*mdev
,
1205 struct drbd_peer_request
*peer_req
,
1206 const unsigned rw
, const int fault_type
)
1208 struct bio
*bios
= NULL
;
1210 struct page
*page
= peer_req
->pages
;
1211 sector_t sector
= peer_req
->i
.sector
;
1212 unsigned ds
= peer_req
->i
.size
;
1213 unsigned n_bios
= 0;
1214 unsigned nr_pages
= (ds
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
1217 /* In most cases, we will only need one bio. But in case the lower
1218 * level restrictions happen to be different at this offset on this
1219 * side than those of the sending peer, we may need to submit the
1220 * request in more than one bio.
1222 * Plain bio_alloc is good enough here, this is no DRBD internally
1223 * generated bio, but a bio allocated on behalf of the peer.
1226 bio
= bio_alloc(GFP_NOIO
, nr_pages
);
1228 dev_err(DEV
, "submit_ee: Allocation of a bio failed\n");
1231 /* > peer_req->i.sector, unless this is the first bio */
1232 bio
->bi_sector
= sector
;
1233 bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1235 bio
->bi_private
= peer_req
;
1236 bio
->bi_end_io
= drbd_peer_request_endio
;
1238 bio
->bi_next
= bios
;
1242 page_chain_for_each(page
) {
1243 unsigned len
= min_t(unsigned, ds
, PAGE_SIZE
);
1244 if (!bio_add_page(bio
, page
, len
, 0)) {
1245 /* A single page must always be possible!
1246 * But in case it fails anyways,
1247 * we deal with it, and complain (below). */
1248 if (bio
->bi_vcnt
== 0) {
1250 "bio_add_page failed for len=%u, "
1251 "bi_vcnt=0 (bi_sector=%llu)\n",
1252 len
, (unsigned long long)bio
->bi_sector
);
1262 D_ASSERT(page
== NULL
);
1265 atomic_set(&peer_req
->pending_bios
, n_bios
);
1268 bios
= bios
->bi_next
;
1269 bio
->bi_next
= NULL
;
1271 drbd_generic_make_request(mdev
, fault_type
, bio
);
1278 bios
= bios
->bi_next
;
1284 static void drbd_remove_epoch_entry_interval(struct drbd_conf
*mdev
,
1285 struct drbd_peer_request
*peer_req
)
1287 struct drbd_interval
*i
= &peer_req
->i
;
1289 drbd_remove_interval(&mdev
->write_requests
, i
);
1290 drbd_clear_interval(i
);
1292 /* Wake up any processes waiting for this peer request to complete. */
1294 wake_up(&mdev
->misc_wait
);
1297 static int receive_Barrier(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1299 struct drbd_conf
*mdev
;
1301 struct p_barrier
*p
= pi
->data
;
1302 struct drbd_epoch
*epoch
;
1304 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1310 mdev
->current_epoch
->barrier_nr
= p
->barrier
;
1311 rv
= drbd_may_finish_epoch(mdev
, mdev
->current_epoch
, EV_GOT_BARRIER_NR
);
1313 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1314 * the activity log, which means it would not be resynced in case the
1315 * R_PRIMARY crashes now.
1316 * Therefore we must send the barrier_ack after the barrier request was
1318 switch (mdev
->write_ordering
) {
1320 if (rv
== FE_RECYCLED
)
1323 /* receiver context, in the writeout path of the other node.
1324 * avoid potential distributed deadlock */
1325 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1329 dev_warn(DEV
, "Allocation of an epoch failed, slowing down\n");
1334 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1337 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1338 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1343 epoch
= mdev
->current_epoch
;
1344 wait_event(mdev
->ee_wait
, atomic_read(&epoch
->epoch_size
) == 0);
1346 D_ASSERT(atomic_read(&epoch
->active
) == 0);
1347 D_ASSERT(epoch
->flags
== 0);
1351 dev_err(DEV
, "Strangeness in mdev->write_ordering %d\n", mdev
->write_ordering
);
1356 atomic_set(&epoch
->epoch_size
, 0);
1357 atomic_set(&epoch
->active
, 0);
1359 spin_lock(&mdev
->epoch_lock
);
1360 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1361 list_add(&epoch
->list
, &mdev
->current_epoch
->list
);
1362 mdev
->current_epoch
= epoch
;
1365 /* The current_epoch got recycled while we allocated this one... */
1368 spin_unlock(&mdev
->epoch_lock
);
1373 /* used from receive_RSDataReply (recv_resync_read)
1374 * and from receive_Data */
1375 static struct drbd_peer_request
*
1376 read_in_block(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
1377 int data_size
) __must_hold(local
)
1379 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1380 struct drbd_peer_request
*peer_req
;
1383 void *dig_in
= mdev
->tconn
->int_dig_in
;
1384 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1385 unsigned long *data
;
1388 if (mdev
->tconn
->peer_integrity_tfm
) {
1389 dgs
= crypto_hash_digestsize(mdev
->tconn
->peer_integrity_tfm
);
1391 * FIXME: Receive the incoming digest into the receive buffer
1392 * here, together with its struct p_data?
1394 err
= drbd_recv_all_warn(mdev
->tconn
, dig_in
, dgs
);
1400 if (!expect(data_size
!= 0))
1402 if (!expect(IS_ALIGNED(data_size
, 512)))
1404 if (!expect(data_size
<= DRBD_MAX_BIO_SIZE
))
1407 /* even though we trust out peer,
1408 * we sometimes have to double check. */
1409 if (sector
+ (data_size
>>9) > capacity
) {
1410 dev_err(DEV
, "request from peer beyond end of local disk: "
1411 "capacity: %llus < sector: %llus + size: %u\n",
1412 (unsigned long long)capacity
,
1413 (unsigned long long)sector
, data_size
);
1417 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1418 * "criss-cross" setup, that might cause write-out on some other DRBD,
1419 * which in turn might block on the other node at this very place. */
1420 peer_req
= drbd_alloc_peer_req(mdev
, id
, sector
, data_size
, GFP_NOIO
);
1425 page
= peer_req
->pages
;
1426 page_chain_for_each(page
) {
1427 unsigned len
= min_t(int, ds
, PAGE_SIZE
);
1429 err
= drbd_recv_all_warn(mdev
->tconn
, data
, len
);
1430 if (drbd_insert_fault(mdev
, DRBD_FAULT_RECEIVE
)) {
1431 dev_err(DEV
, "Fault injection: Corrupting data on receive\n");
1432 data
[0] = data
[0] ^ (unsigned long)-1;
1436 drbd_free_peer_req(mdev
, peer_req
);
1443 drbd_csum_ee(mdev
, mdev
->tconn
->peer_integrity_tfm
, peer_req
, dig_vv
);
1444 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1445 dev_err(DEV
, "Digest integrity check FAILED: %llus +%u\n",
1446 (unsigned long long)sector
, data_size
);
1447 drbd_free_peer_req(mdev
, peer_req
);
1451 mdev
->recv_cnt
+= data_size
>>9;
1455 /* drbd_drain_block() just takes a data block
1456 * out of the socket input buffer, and discards it.
1458 static int drbd_drain_block(struct drbd_conf
*mdev
, int data_size
)
1467 page
= drbd_alloc_pages(mdev
, 1, 1);
1471 unsigned int len
= min_t(int, data_size
, PAGE_SIZE
);
1473 err
= drbd_recv_all_warn(mdev
->tconn
, data
, len
);
1479 drbd_free_pages(mdev
, page
, 0);
1483 static int recv_dless_read(struct drbd_conf
*mdev
, struct drbd_request
*req
,
1484 sector_t sector
, int data_size
)
1486 struct bio_vec
*bvec
;
1488 int dgs
, err
, i
, expect
;
1489 void *dig_in
= mdev
->tconn
->int_dig_in
;
1490 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1493 if (mdev
->tconn
->peer_integrity_tfm
) {
1494 dgs
= crypto_hash_digestsize(mdev
->tconn
->peer_integrity_tfm
);
1495 err
= drbd_recv_all_warn(mdev
->tconn
, dig_in
, dgs
);
1501 /* optimistically update recv_cnt. if receiving fails below,
1502 * we disconnect anyways, and counters will be reset. */
1503 mdev
->recv_cnt
+= data_size
>>9;
1505 bio
= req
->master_bio
;
1506 D_ASSERT(sector
== bio
->bi_sector
);
1508 bio_for_each_segment(bvec
, bio
, i
) {
1509 void *mapped
= kmap(bvec
->bv_page
) + bvec
->bv_offset
;
1510 expect
= min_t(int, data_size
, bvec
->bv_len
);
1511 err
= drbd_recv_all_warn(mdev
->tconn
, mapped
, expect
);
1512 kunmap(bvec
->bv_page
);
1515 data_size
-= expect
;
1519 drbd_csum_bio(mdev
, mdev
->tconn
->peer_integrity_tfm
, bio
, dig_vv
);
1520 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1521 dev_err(DEV
, "Digest integrity check FAILED. Broken NICs?\n");
1526 D_ASSERT(data_size
== 0);
1531 * e_end_resync_block() is called in asender context via
1532 * drbd_finish_peer_reqs().
1534 static int e_end_resync_block(struct drbd_work
*w
, int unused
)
1536 struct drbd_peer_request
*peer_req
=
1537 container_of(w
, struct drbd_peer_request
, w
);
1538 struct drbd_conf
*mdev
= w
->mdev
;
1539 sector_t sector
= peer_req
->i
.sector
;
1542 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1544 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1545 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1546 err
= drbd_send_ack(mdev
, P_RS_WRITE_ACK
, peer_req
);
1548 /* Record failure to sync */
1549 drbd_rs_failed_io(mdev
, sector
, peer_req
->i
.size
);
1551 err
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1558 static int recv_resync_read(struct drbd_conf
*mdev
, sector_t sector
, int data_size
) __releases(local
)
1560 struct drbd_peer_request
*peer_req
;
1562 peer_req
= read_in_block(mdev
, ID_SYNCER
, sector
, data_size
);
1566 dec_rs_pending(mdev
);
1569 /* corresponding dec_unacked() in e_end_resync_block()
1570 * respective _drbd_clear_done_ee */
1572 peer_req
->w
.cb
= e_end_resync_block
;
1574 spin_lock_irq(&mdev
->tconn
->req_lock
);
1575 list_add(&peer_req
->w
.list
, &mdev
->sync_ee
);
1576 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1578 atomic_add(data_size
>> 9, &mdev
->rs_sect_ev
);
1579 if (drbd_submit_peer_request(mdev
, peer_req
, WRITE
, DRBD_FAULT_RS_WR
) == 0)
1582 /* don't care for the reason here */
1583 dev_err(DEV
, "submit failed, triggering re-connect\n");
1584 spin_lock_irq(&mdev
->tconn
->req_lock
);
1585 list_del(&peer_req
->w
.list
);
1586 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1588 drbd_free_peer_req(mdev
, peer_req
);
1594 static struct drbd_request
*
1595 find_request(struct drbd_conf
*mdev
, struct rb_root
*root
, u64 id
,
1596 sector_t sector
, bool missing_ok
, const char *func
)
1598 struct drbd_request
*req
;
1600 /* Request object according to our peer */
1601 req
= (struct drbd_request
*)(unsigned long)id
;
1602 if (drbd_contains_interval(root
, sector
, &req
->i
) && req
->i
.local
)
1605 dev_err(DEV
, "%s: failed to find request %lu, sector %llus\n", func
,
1606 (unsigned long)id
, (unsigned long long)sector
);
1611 static int receive_DataReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1613 struct drbd_conf
*mdev
;
1614 struct drbd_request
*req
;
1617 struct p_data
*p
= pi
->data
;
1619 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1623 sector
= be64_to_cpu(p
->sector
);
1625 spin_lock_irq(&mdev
->tconn
->req_lock
);
1626 req
= find_request(mdev
, &mdev
->read_requests
, p
->block_id
, sector
, false, __func__
);
1627 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1631 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1632 * special casing it there for the various failure cases.
1633 * still no race with drbd_fail_pending_reads */
1634 err
= recv_dless_read(mdev
, req
, sector
, pi
->size
);
1636 req_mod(req
, DATA_RECEIVED
);
1637 /* else: nothing. handled from drbd_disconnect...
1638 * I don't think we may complete this just yet
1639 * in case we are "on-disconnect: freeze" */
1644 static int receive_RSDataReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
1646 struct drbd_conf
*mdev
;
1649 struct p_data
*p
= pi
->data
;
1651 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
1655 sector
= be64_to_cpu(p
->sector
);
1656 D_ASSERT(p
->block_id
== ID_SYNCER
);
1658 if (get_ldev(mdev
)) {
1659 /* data is submitted to disk within recv_resync_read.
1660 * corresponding put_ldev done below on error,
1661 * or in drbd_peer_request_endio. */
1662 err
= recv_resync_read(mdev
, sector
, pi
->size
);
1664 if (__ratelimit(&drbd_ratelimit_state
))
1665 dev_err(DEV
, "Can not write resync data to local disk.\n");
1667 err
= drbd_drain_block(mdev
, pi
->size
);
1669 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, pi
->size
);
1672 atomic_add(pi
->size
>> 9, &mdev
->rs_sect_in
);
1677 static int w_restart_write(struct drbd_work
*w
, int cancel
)
1679 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1680 struct drbd_conf
*mdev
= w
->mdev
;
1682 unsigned long start_time
;
1683 unsigned long flags
;
1685 spin_lock_irqsave(&mdev
->tconn
->req_lock
, flags
);
1686 if (!expect(req
->rq_state
& RQ_POSTPONED
)) {
1687 spin_unlock_irqrestore(&mdev
->tconn
->req_lock
, flags
);
1690 bio
= req
->master_bio
;
1691 start_time
= req
->start_time
;
1692 /* Postponed requests will not have their master_bio completed! */
1693 __req_mod(req
, DISCARD_WRITE
, NULL
);
1694 spin_unlock_irqrestore(&mdev
->tconn
->req_lock
, flags
);
1696 while (__drbd_make_request(mdev
, bio
, start_time
))
1701 static void restart_conflicting_writes(struct drbd_conf
*mdev
,
1702 sector_t sector
, int size
)
1704 struct drbd_interval
*i
;
1705 struct drbd_request
*req
;
1707 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1710 req
= container_of(i
, struct drbd_request
, i
);
1711 if (req
->rq_state
& RQ_LOCAL_PENDING
||
1712 !(req
->rq_state
& RQ_POSTPONED
))
1714 if (expect(list_empty(&req
->w
.list
))) {
1716 req
->w
.cb
= w_restart_write
;
1717 drbd_queue_work(&mdev
->tconn
->data
.work
, &req
->w
);
1723 * e_end_block() is called in asender context via drbd_finish_peer_reqs().
1725 static int e_end_block(struct drbd_work
*w
, int cancel
)
1727 struct drbd_peer_request
*peer_req
=
1728 container_of(w
, struct drbd_peer_request
, w
);
1729 struct drbd_conf
*mdev
= w
->mdev
;
1730 sector_t sector
= peer_req
->i
.sector
;
1733 if (peer_req
->flags
& EE_SEND_WRITE_ACK
) {
1734 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1735 pcmd
= (mdev
->state
.conn
>= C_SYNC_SOURCE
&&
1736 mdev
->state
.conn
<= C_PAUSED_SYNC_T
&&
1737 peer_req
->flags
& EE_MAY_SET_IN_SYNC
) ?
1738 P_RS_WRITE_ACK
: P_WRITE_ACK
;
1739 err
= drbd_send_ack(mdev
, pcmd
, peer_req
);
1740 if (pcmd
== P_RS_WRITE_ACK
)
1741 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1743 err
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1744 /* we expect it to be marked out of sync anyways...
1745 * maybe assert this? */
1749 /* we delete from the conflict detection hash _after_ we sent out the
1750 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1751 if (peer_req
->flags
& EE_IN_INTERVAL_TREE
) {
1752 spin_lock_irq(&mdev
->tconn
->req_lock
);
1753 D_ASSERT(!drbd_interval_empty(&peer_req
->i
));
1754 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
1755 if (peer_req
->flags
& EE_RESTART_REQUESTS
)
1756 restart_conflicting_writes(mdev
, sector
, peer_req
->i
.size
);
1757 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1759 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1761 drbd_may_finish_epoch(mdev
, peer_req
->epoch
, EV_PUT
+ (cancel
? EV_CLEANUP
: 0));
1766 static int e_send_ack(struct drbd_work
*w
, enum drbd_packet ack
)
1768 struct drbd_conf
*mdev
= w
->mdev
;
1769 struct drbd_peer_request
*peer_req
=
1770 container_of(w
, struct drbd_peer_request
, w
);
1773 err
= drbd_send_ack(mdev
, ack
, peer_req
);
1779 static int e_send_discard_write(struct drbd_work
*w
, int unused
)
1781 return e_send_ack(w
, P_DISCARD_WRITE
);
1784 static int e_send_retry_write(struct drbd_work
*w
, int unused
)
1786 struct drbd_tconn
*tconn
= w
->mdev
->tconn
;
1788 return e_send_ack(w
, tconn
->agreed_pro_version
>= 100 ?
1789 P_RETRY_WRITE
: P_DISCARD_WRITE
);
1792 static bool seq_greater(u32 a
, u32 b
)
1795 * We assume 32-bit wrap-around here.
1796 * For 24-bit wrap-around, we would have to shift:
1799 return (s32
)a
- (s32
)b
> 0;
1802 static u32
seq_max(u32 a
, u32 b
)
1804 return seq_greater(a
, b
) ? a
: b
;
1807 static bool need_peer_seq(struct drbd_conf
*mdev
)
1809 struct drbd_tconn
*tconn
= mdev
->tconn
;
1813 * We only need to keep track of the last packet_seq number of our peer
1814 * if we are in dual-primary mode and we have the discard flag set; see
1815 * handle_write_conflicts().
1819 tp
= rcu_dereference(mdev
->tconn
->net_conf
)->two_primaries
;
1822 return tp
&& test_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
1825 static void update_peer_seq(struct drbd_conf
*mdev
, unsigned int peer_seq
)
1827 unsigned int newest_peer_seq
;
1829 if (need_peer_seq(mdev
)) {
1830 spin_lock(&mdev
->peer_seq_lock
);
1831 newest_peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1832 mdev
->peer_seq
= newest_peer_seq
;
1833 spin_unlock(&mdev
->peer_seq_lock
);
1834 /* wake up only if we actually changed mdev->peer_seq */
1835 if (peer_seq
== newest_peer_seq
)
1836 wake_up(&mdev
->seq_wait
);
1840 /* Called from receive_Data.
1841 * Synchronize packets on sock with packets on msock.
1843 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1844 * packet traveling on msock, they are still processed in the order they have
1847 * Note: we don't care for Ack packets overtaking P_DATA packets.
1849 * In case packet_seq is larger than mdev->peer_seq number, there are
1850 * outstanding packets on the msock. We wait for them to arrive.
1851 * In case we are the logically next packet, we update mdev->peer_seq
1852 * ourselves. Correctly handles 32bit wrap around.
1854 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1855 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1856 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1857 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1859 * returns 0 if we may process the packet,
1860 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1861 static int wait_for_and_update_peer_seq(struct drbd_conf
*mdev
, const u32 peer_seq
)
1867 if (!need_peer_seq(mdev
))
1870 spin_lock(&mdev
->peer_seq_lock
);
1872 if (!seq_greater(peer_seq
- 1, mdev
->peer_seq
)) {
1873 mdev
->peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1877 if (signal_pending(current
)) {
1881 prepare_to_wait(&mdev
->seq_wait
, &wait
, TASK_INTERRUPTIBLE
);
1882 spin_unlock(&mdev
->peer_seq_lock
);
1884 timeout
= rcu_dereference(mdev
->tconn
->net_conf
)->ping_timeo
*HZ
/10;
1886 timeout
= schedule_timeout(timeout
);
1887 spin_lock(&mdev
->peer_seq_lock
);
1890 dev_err(DEV
, "Timed out waiting for missing ack packets; disconnecting\n");
1894 spin_unlock(&mdev
->peer_seq_lock
);
1895 finish_wait(&mdev
->seq_wait
, &wait
);
1899 /* see also bio_flags_to_wire()
1900 * DRBD_REQ_*, because we need to semantically map the flags to data packet
1901 * flags and back. We may replicate to other kernel versions. */
1902 static unsigned long wire_flags_to_bio(struct drbd_conf
*mdev
, u32 dpf
)
1904 return (dpf
& DP_RW_SYNC
? REQ_SYNC
: 0) |
1905 (dpf
& DP_FUA
? REQ_FUA
: 0) |
1906 (dpf
& DP_FLUSH
? REQ_FLUSH
: 0) |
1907 (dpf
& DP_DISCARD
? REQ_DISCARD
: 0);
1910 static void fail_postponed_requests(struct drbd_conf
*mdev
, sector_t sector
,
1913 struct drbd_interval
*i
;
1916 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1917 struct drbd_request
*req
;
1918 struct bio_and_error m
;
1922 req
= container_of(i
, struct drbd_request
, i
);
1923 if (!(req
->rq_state
& RQ_POSTPONED
))
1925 req
->rq_state
&= ~RQ_POSTPONED
;
1926 __req_mod(req
, NEG_ACKED
, &m
);
1927 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1929 complete_master_bio(mdev
, &m
);
1930 spin_lock_irq(&mdev
->tconn
->req_lock
);
1935 static int handle_write_conflicts(struct drbd_conf
*mdev
,
1936 struct drbd_peer_request
*peer_req
)
1938 struct drbd_tconn
*tconn
= mdev
->tconn
;
1939 bool resolve_conflicts
= test_bit(DISCARD_CONCURRENT
, &tconn
->flags
);
1940 sector_t sector
= peer_req
->i
.sector
;
1941 const unsigned int size
= peer_req
->i
.size
;
1942 struct drbd_interval
*i
;
1947 * Inserting the peer request into the write_requests tree will prevent
1948 * new conflicting local requests from being added.
1950 drbd_insert_interval(&mdev
->write_requests
, &peer_req
->i
);
1953 drbd_for_each_overlap(i
, &mdev
->write_requests
, sector
, size
) {
1954 if (i
== &peer_req
->i
)
1959 * Our peer has sent a conflicting remote request; this
1960 * should not happen in a two-node setup. Wait for the
1961 * earlier peer request to complete.
1963 err
= drbd_wait_misc(mdev
, i
);
1969 equal
= i
->sector
== sector
&& i
->size
== size
;
1970 if (resolve_conflicts
) {
1972 * If the peer request is fully contained within the
1973 * overlapping request, it can be discarded; otherwise,
1974 * it will be retried once all overlapping requests
1977 bool discard
= i
->sector
<= sector
&& i
->sector
+
1978 (i
->size
>> 9) >= sector
+ (size
>> 9);
1981 dev_alert(DEV
, "Concurrent writes detected: "
1982 "local=%llus +%u, remote=%llus +%u, "
1983 "assuming %s came first\n",
1984 (unsigned long long)i
->sector
, i
->size
,
1985 (unsigned long long)sector
, size
,
1986 discard
? "local" : "remote");
1989 peer_req
->w
.cb
= discard
? e_send_discard_write
:
1991 list_add_tail(&peer_req
->w
.list
, &mdev
->done_ee
);
1992 wake_asender(mdev
->tconn
);
1997 struct drbd_request
*req
=
1998 container_of(i
, struct drbd_request
, i
);
2001 dev_alert(DEV
, "Concurrent writes detected: "
2002 "local=%llus +%u, remote=%llus +%u\n",
2003 (unsigned long long)i
->sector
, i
->size
,
2004 (unsigned long long)sector
, size
);
2006 if (req
->rq_state
& RQ_LOCAL_PENDING
||
2007 !(req
->rq_state
& RQ_POSTPONED
)) {
2009 * Wait for the node with the discard flag to
2010 * decide if this request will be discarded or
2011 * retried. Requests that are discarded will
2012 * disappear from the write_requests tree.
2014 * In addition, wait for the conflicting
2015 * request to finish locally before submitting
2016 * the conflicting peer request.
2018 err
= drbd_wait_misc(mdev
, &req
->i
);
2020 _conn_request_state(mdev
->tconn
,
2021 NS(conn
, C_TIMEOUT
),
2023 fail_postponed_requests(mdev
, sector
, size
);
2029 * Remember to restart the conflicting requests after
2030 * the new peer request has completed.
2032 peer_req
->flags
|= EE_RESTART_REQUESTS
;
2039 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
2043 /* mirrored write */
2044 static int receive_Data(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2046 struct drbd_conf
*mdev
;
2048 struct drbd_peer_request
*peer_req
;
2049 struct p_data
*p
= pi
->data
;
2050 u32 peer_seq
= be32_to_cpu(p
->seq_num
);
2055 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
2059 if (!get_ldev(mdev
)) {
2062 err
= wait_for_and_update_peer_seq(mdev
, peer_seq
);
2063 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, pi
->size
);
2064 atomic_inc(&mdev
->current_epoch
->epoch_size
);
2065 err2
= drbd_drain_block(mdev
, pi
->size
);
2072 * Corresponding put_ldev done either below (on various errors), or in
2073 * drbd_peer_request_endio, if we successfully submit the data at the
2074 * end of this function.
2077 sector
= be64_to_cpu(p
->sector
);
2078 peer_req
= read_in_block(mdev
, p
->block_id
, sector
, pi
->size
);
2084 peer_req
->w
.cb
= e_end_block
;
2086 dp_flags
= be32_to_cpu(p
->dp_flags
);
2087 rw
|= wire_flags_to_bio(mdev
, dp_flags
);
2089 if (dp_flags
& DP_MAY_SET_IN_SYNC
)
2090 peer_req
->flags
|= EE_MAY_SET_IN_SYNC
;
2092 spin_lock(&mdev
->epoch_lock
);
2093 peer_req
->epoch
= mdev
->current_epoch
;
2094 atomic_inc(&peer_req
->epoch
->epoch_size
);
2095 atomic_inc(&peer_req
->epoch
->active
);
2096 spin_unlock(&mdev
->epoch_lock
);
2099 tp
= rcu_dereference(mdev
->tconn
->net_conf
)->two_primaries
;
2102 peer_req
->flags
|= EE_IN_INTERVAL_TREE
;
2103 err
= wait_for_and_update_peer_seq(mdev
, peer_seq
);
2105 goto out_interrupted
;
2106 spin_lock_irq(&mdev
->tconn
->req_lock
);
2107 err
= handle_write_conflicts(mdev
, peer_req
);
2109 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2110 if (err
== -ENOENT
) {
2114 goto out_interrupted
;
2117 spin_lock_irq(&mdev
->tconn
->req_lock
);
2118 list_add(&peer_req
->w
.list
, &mdev
->active_ee
);
2119 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2121 if (mdev
->tconn
->agreed_pro_version
< 100) {
2123 switch (rcu_dereference(mdev
->tconn
->net_conf
)->wire_protocol
) {
2125 dp_flags
|= DP_SEND_WRITE_ACK
;
2128 dp_flags
|= DP_SEND_RECEIVE_ACK
;
2134 if (dp_flags
& DP_SEND_WRITE_ACK
) {
2135 peer_req
->flags
|= EE_SEND_WRITE_ACK
;
2137 /* corresponding dec_unacked() in e_end_block()
2138 * respective _drbd_clear_done_ee */
2141 if (dp_flags
& DP_SEND_RECEIVE_ACK
) {
2142 /* I really don't like it that the receiver thread
2143 * sends on the msock, but anyways */
2144 drbd_send_ack(mdev
, P_RECV_ACK
, peer_req
);
2147 if (mdev
->state
.pdsk
< D_INCONSISTENT
) {
2148 /* In case we have the only disk of the cluster, */
2149 drbd_set_out_of_sync(mdev
, peer_req
->i
.sector
, peer_req
->i
.size
);
2150 peer_req
->flags
|= EE_CALL_AL_COMPLETE_IO
;
2151 peer_req
->flags
&= ~EE_MAY_SET_IN_SYNC
;
2152 drbd_al_begin_io(mdev
, &peer_req
->i
);
2155 err
= drbd_submit_peer_request(mdev
, peer_req
, rw
, DRBD_FAULT_DT_WR
);
2159 /* don't care for the reason here */
2160 dev_err(DEV
, "submit failed, triggering re-connect\n");
2161 spin_lock_irq(&mdev
->tconn
->req_lock
);
2162 list_del(&peer_req
->w
.list
);
2163 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
2164 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2165 if (peer_req
->flags
& EE_CALL_AL_COMPLETE_IO
)
2166 drbd_al_complete_io(mdev
, &peer_req
->i
);
2169 drbd_may_finish_epoch(mdev
, peer_req
->epoch
, EV_PUT
+ EV_CLEANUP
);
2171 drbd_free_peer_req(mdev
, peer_req
);
2175 /* We may throttle resync, if the lower device seems to be busy,
2176 * and current sync rate is above c_min_rate.
2178 * To decide whether or not the lower device is busy, we use a scheme similar
2179 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2180 * (more than 64 sectors) of activity we cannot account for with our own resync
2181 * activity, it obviously is "busy".
2183 * The current sync rate used here uses only the most recent two step marks,
2184 * to have a short time average so we can react faster.
2186 int drbd_rs_should_slow_down(struct drbd_conf
*mdev
, sector_t sector
)
2188 struct gendisk
*disk
= mdev
->ldev
->backing_bdev
->bd_contains
->bd_disk
;
2189 unsigned long db
, dt
, dbdt
;
2190 struct lc_element
*tmp
;
2194 /* feature disabled? */
2195 if (mdev
->ldev
->dc
.c_min_rate
== 0)
2198 spin_lock_irq(&mdev
->al_lock
);
2199 tmp
= lc_find(mdev
->resync
, BM_SECT_TO_EXT(sector
));
2201 struct bm_extent
*bm_ext
= lc_entry(tmp
, struct bm_extent
, lce
);
2202 if (test_bit(BME_PRIORITY
, &bm_ext
->flags
)) {
2203 spin_unlock_irq(&mdev
->al_lock
);
2206 /* Do not slow down if app IO is already waiting for this extent */
2208 spin_unlock_irq(&mdev
->al_lock
);
2210 curr_events
= (int)part_stat_read(&disk
->part0
, sectors
[0]) +
2211 (int)part_stat_read(&disk
->part0
, sectors
[1]) -
2212 atomic_read(&mdev
->rs_sect_ev
);
2214 if (!mdev
->rs_last_events
|| curr_events
- mdev
->rs_last_events
> 64) {
2215 unsigned long rs_left
;
2218 mdev
->rs_last_events
= curr_events
;
2220 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2222 i
= (mdev
->rs_last_mark
+ DRBD_SYNC_MARKS
-1) % DRBD_SYNC_MARKS
;
2224 if (mdev
->state
.conn
== C_VERIFY_S
|| mdev
->state
.conn
== C_VERIFY_T
)
2225 rs_left
= mdev
->ov_left
;
2227 rs_left
= drbd_bm_total_weight(mdev
) - mdev
->rs_failed
;
2229 dt
= ((long)jiffies
- (long)mdev
->rs_mark_time
[i
]) / HZ
;
2232 db
= mdev
->rs_mark_left
[i
] - rs_left
;
2233 dbdt
= Bit2KB(db
/dt
);
2235 if (dbdt
> mdev
->ldev
->dc
.c_min_rate
)
2242 static int receive_DataRequest(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2244 struct drbd_conf
*mdev
;
2247 struct drbd_peer_request
*peer_req
;
2248 struct digest_info
*di
= NULL
;
2250 unsigned int fault_type
;
2251 struct p_block_req
*p
= pi
->data
;
2253 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
2256 capacity
= drbd_get_capacity(mdev
->this_bdev
);
2258 sector
= be64_to_cpu(p
->sector
);
2259 size
= be32_to_cpu(p
->blksize
);
2261 if (size
<= 0 || !IS_ALIGNED(size
, 512) || size
> DRBD_MAX_BIO_SIZE
) {
2262 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2263 (unsigned long long)sector
, size
);
2266 if (sector
+ (size
>>9) > capacity
) {
2267 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2268 (unsigned long long)sector
, size
);
2272 if (!get_ldev_if_state(mdev
, D_UP_TO_DATE
)) {
2275 case P_DATA_REQUEST
:
2276 drbd_send_ack_rp(mdev
, P_NEG_DREPLY
, p
);
2278 case P_RS_DATA_REQUEST
:
2279 case P_CSUM_RS_REQUEST
:
2281 drbd_send_ack_rp(mdev
, P_NEG_RS_DREPLY
, p
);
2285 dec_rs_pending(mdev
);
2286 drbd_send_ack_ex(mdev
, P_OV_RESULT
, sector
, size
, ID_IN_SYNC
);
2291 if (verb
&& __ratelimit(&drbd_ratelimit_state
))
2292 dev_err(DEV
, "Can not satisfy peer's read request, "
2293 "no local data.\n");
2295 /* drain possibly payload */
2296 return drbd_drain_block(mdev
, pi
->size
);
2299 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2300 * "criss-cross" setup, that might cause write-out on some other DRBD,
2301 * which in turn might block on the other node at this very place. */
2302 peer_req
= drbd_alloc_peer_req(mdev
, p
->block_id
, sector
, size
, GFP_NOIO
);
2309 case P_DATA_REQUEST
:
2310 peer_req
->w
.cb
= w_e_end_data_req
;
2311 fault_type
= DRBD_FAULT_DT_RD
;
2312 /* application IO, don't drbd_rs_begin_io */
2315 case P_RS_DATA_REQUEST
:
2316 peer_req
->w
.cb
= w_e_end_rsdata_req
;
2317 fault_type
= DRBD_FAULT_RS_RD
;
2318 /* used in the sector offset progress display */
2319 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2323 case P_CSUM_RS_REQUEST
:
2324 fault_type
= DRBD_FAULT_RS_RD
;
2325 di
= kmalloc(sizeof(*di
) + pi
->size
, GFP_NOIO
);
2329 di
->digest_size
= pi
->size
;
2330 di
->digest
= (((char *)di
)+sizeof(struct digest_info
));
2332 peer_req
->digest
= di
;
2333 peer_req
->flags
|= EE_HAS_DIGEST
;
2335 if (drbd_recv_all(mdev
->tconn
, di
->digest
, pi
->size
))
2338 if (pi
->cmd
== P_CSUM_RS_REQUEST
) {
2339 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
2340 peer_req
->w
.cb
= w_e_end_csum_rs_req
;
2341 /* used in the sector offset progress display */
2342 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2343 } else if (pi
->cmd
== P_OV_REPLY
) {
2344 /* track progress, we may need to throttle */
2345 atomic_add(size
>> 9, &mdev
->rs_sect_in
);
2346 peer_req
->w
.cb
= w_e_end_ov_reply
;
2347 dec_rs_pending(mdev
);
2348 /* drbd_rs_begin_io done when we sent this request,
2349 * but accounting still needs to be done. */
2350 goto submit_for_resync
;
2355 if (mdev
->ov_start_sector
== ~(sector_t
)0 &&
2356 mdev
->tconn
->agreed_pro_version
>= 90) {
2357 unsigned long now
= jiffies
;
2359 mdev
->ov_start_sector
= sector
;
2360 mdev
->ov_position
= sector
;
2361 mdev
->ov_left
= drbd_bm_bits(mdev
) - BM_SECT_TO_BIT(sector
);
2362 mdev
->rs_total
= mdev
->ov_left
;
2363 for (i
= 0; i
< DRBD_SYNC_MARKS
; i
++) {
2364 mdev
->rs_mark_left
[i
] = mdev
->ov_left
;
2365 mdev
->rs_mark_time
[i
] = now
;
2367 dev_info(DEV
, "Online Verify start sector: %llu\n",
2368 (unsigned long long)sector
);
2370 peer_req
->w
.cb
= w_e_end_ov_req
;
2371 fault_type
= DRBD_FAULT_RS_RD
;
2378 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2379 * wrt the receiver, but it is not as straightforward as it may seem.
2380 * Various places in the resync start and stop logic assume resync
2381 * requests are processed in order, requeuing this on the worker thread
2382 * introduces a bunch of new code for synchronization between threads.
2384 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2385 * "forever", throttling after drbd_rs_begin_io will lock that extent
2386 * for application writes for the same time. For now, just throttle
2387 * here, where the rest of the code expects the receiver to sleep for
2391 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2392 * this defers syncer requests for some time, before letting at least
2393 * on request through. The resync controller on the receiving side
2394 * will adapt to the incoming rate accordingly.
2396 * We cannot throttle here if remote is Primary/SyncTarget:
2397 * we would also throttle its application reads.
2398 * In that case, throttling is done on the SyncTarget only.
2400 if (mdev
->state
.peer
!= R_PRIMARY
&& drbd_rs_should_slow_down(mdev
, sector
))
2401 schedule_timeout_uninterruptible(HZ
/10);
2402 if (drbd_rs_begin_io(mdev
, sector
))
2406 atomic_add(size
>> 9, &mdev
->rs_sect_ev
);
2410 spin_lock_irq(&mdev
->tconn
->req_lock
);
2411 list_add_tail(&peer_req
->w
.list
, &mdev
->read_ee
);
2412 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2414 if (drbd_submit_peer_request(mdev
, peer_req
, READ
, fault_type
) == 0)
2417 /* don't care for the reason here */
2418 dev_err(DEV
, "submit failed, triggering re-connect\n");
2419 spin_lock_irq(&mdev
->tconn
->req_lock
);
2420 list_del(&peer_req
->w
.list
);
2421 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2422 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2426 drbd_free_peer_req(mdev
, peer_req
);
2430 static int drbd_asb_recover_0p(struct drbd_conf
*mdev
) __must_hold(local
)
2432 int self
, peer
, rv
= -100;
2433 unsigned long ch_self
, ch_peer
;
2434 enum drbd_after_sb_p after_sb_0p
;
2436 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2437 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2439 ch_peer
= mdev
->p_uuid
[UI_SIZE
];
2440 ch_self
= mdev
->comm_bm_set
;
2443 after_sb_0p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_0p
;
2445 switch (after_sb_0p
) {
2447 case ASB_DISCARD_SECONDARY
:
2448 case ASB_CALL_HELPER
:
2450 dev_err(DEV
, "Configuration error.\n");
2452 case ASB_DISCONNECT
:
2454 case ASB_DISCARD_YOUNGER_PRI
:
2455 if (self
== 0 && peer
== 1) {
2459 if (self
== 1 && peer
== 0) {
2463 /* Else fall through to one of the other strategies... */
2464 case ASB_DISCARD_OLDER_PRI
:
2465 if (self
== 0 && peer
== 1) {
2469 if (self
== 1 && peer
== 0) {
2473 /* Else fall through to one of the other strategies... */
2474 dev_warn(DEV
, "Discard younger/older primary did not find a decision\n"
2475 "Using discard-least-changes instead\n");
2476 case ASB_DISCARD_ZERO_CHG
:
2477 if (ch_peer
== 0 && ch_self
== 0) {
2478 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
)
2482 if (ch_peer
== 0) { rv
= 1; break; }
2483 if (ch_self
== 0) { rv
= -1; break; }
2485 if (after_sb_0p
== ASB_DISCARD_ZERO_CHG
)
2487 case ASB_DISCARD_LEAST_CHG
:
2488 if (ch_self
< ch_peer
)
2490 else if (ch_self
> ch_peer
)
2492 else /* ( ch_self == ch_peer ) */
2493 /* Well, then use something else. */
2494 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
)
2497 case ASB_DISCARD_LOCAL
:
2500 case ASB_DISCARD_REMOTE
:
2507 static int drbd_asb_recover_1p(struct drbd_conf
*mdev
) __must_hold(local
)
2510 enum drbd_after_sb_p after_sb_1p
;
2513 after_sb_1p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_1p
;
2515 switch (after_sb_1p
) {
2516 case ASB_DISCARD_YOUNGER_PRI
:
2517 case ASB_DISCARD_OLDER_PRI
:
2518 case ASB_DISCARD_LEAST_CHG
:
2519 case ASB_DISCARD_LOCAL
:
2520 case ASB_DISCARD_REMOTE
:
2521 case ASB_DISCARD_ZERO_CHG
:
2522 dev_err(DEV
, "Configuration error.\n");
2524 case ASB_DISCONNECT
:
2527 hg
= drbd_asb_recover_0p(mdev
);
2528 if (hg
== -1 && mdev
->state
.role
== R_SECONDARY
)
2530 if (hg
== 1 && mdev
->state
.role
== R_PRIMARY
)
2534 rv
= drbd_asb_recover_0p(mdev
);
2536 case ASB_DISCARD_SECONDARY
:
2537 return mdev
->state
.role
== R_PRIMARY
? 1 : -1;
2538 case ASB_CALL_HELPER
:
2539 hg
= drbd_asb_recover_0p(mdev
);
2540 if (hg
== -1 && mdev
->state
.role
== R_PRIMARY
) {
2541 enum drbd_state_rv rv2
;
2543 drbd_set_role(mdev
, R_SECONDARY
, 0);
2544 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2545 * we might be here in C_WF_REPORT_PARAMS which is transient.
2546 * we do not need to wait for the after state change work either. */
2547 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2548 if (rv2
!= SS_SUCCESS
) {
2549 drbd_khelper(mdev
, "pri-lost-after-sb");
2551 dev_warn(DEV
, "Successfully gave up primary role.\n");
2561 static int drbd_asb_recover_2p(struct drbd_conf
*mdev
) __must_hold(local
)
2564 enum drbd_after_sb_p after_sb_2p
;
2567 after_sb_2p
= rcu_dereference(mdev
->tconn
->net_conf
)->after_sb_2p
;
2569 switch (after_sb_2p
) {
2570 case ASB_DISCARD_YOUNGER_PRI
:
2571 case ASB_DISCARD_OLDER_PRI
:
2572 case ASB_DISCARD_LEAST_CHG
:
2573 case ASB_DISCARD_LOCAL
:
2574 case ASB_DISCARD_REMOTE
:
2576 case ASB_DISCARD_SECONDARY
:
2577 case ASB_DISCARD_ZERO_CHG
:
2578 dev_err(DEV
, "Configuration error.\n");
2581 rv
= drbd_asb_recover_0p(mdev
);
2583 case ASB_DISCONNECT
:
2585 case ASB_CALL_HELPER
:
2586 hg
= drbd_asb_recover_0p(mdev
);
2588 enum drbd_state_rv rv2
;
2590 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2591 * we might be here in C_WF_REPORT_PARAMS which is transient.
2592 * we do not need to wait for the after state change work either. */
2593 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2594 if (rv2
!= SS_SUCCESS
) {
2595 drbd_khelper(mdev
, "pri-lost-after-sb");
2597 dev_warn(DEV
, "Successfully gave up primary role.\n");
2607 static void drbd_uuid_dump(struct drbd_conf
*mdev
, char *text
, u64
*uuid
,
2608 u64 bits
, u64 flags
)
2611 dev_info(DEV
, "%s uuid info vanished while I was looking!\n", text
);
2614 dev_info(DEV
, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2616 (unsigned long long)uuid
[UI_CURRENT
],
2617 (unsigned long long)uuid
[UI_BITMAP
],
2618 (unsigned long long)uuid
[UI_HISTORY_START
],
2619 (unsigned long long)uuid
[UI_HISTORY_END
],
2620 (unsigned long long)bits
,
2621 (unsigned long long)flags
);
2625 100 after split brain try auto recover
2626 2 C_SYNC_SOURCE set BitMap
2627 1 C_SYNC_SOURCE use BitMap
2629 -1 C_SYNC_TARGET use BitMap
2630 -2 C_SYNC_TARGET set BitMap
2631 -100 after split brain, disconnect
2632 -1000 unrelated data
2633 -1091 requires proto 91
2634 -1096 requires proto 96
2636 static int drbd_uuid_compare(struct drbd_conf
*mdev
, int *rule_nr
) __must_hold(local
)
2641 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2642 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2645 if (self
== UUID_JUST_CREATED
&& peer
== UUID_JUST_CREATED
)
2649 if ((self
== UUID_JUST_CREATED
|| self
== (u64
)0) &&
2650 peer
!= UUID_JUST_CREATED
)
2654 if (self
!= UUID_JUST_CREATED
&&
2655 (peer
== UUID_JUST_CREATED
|| peer
== (u64
)0))
2659 int rct
, dc
; /* roles at crash time */
2661 if (mdev
->p_uuid
[UI_BITMAP
] == (u64
)0 && mdev
->ldev
->md
.uuid
[UI_BITMAP
] != (u64
)0) {
2663 if (mdev
->tconn
->agreed_pro_version
< 91)
2666 if ((mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) &&
2667 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1))) {
2668 dev_info(DEV
, "was SyncSource, missed the resync finished event, corrected myself:\n");
2669 drbd_uuid_set_bm(mdev
, 0UL);
2671 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2672 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2675 dev_info(DEV
, "was SyncSource (peer failed to write sync_uuid)\n");
2682 if (mdev
->ldev
->md
.uuid
[UI_BITMAP
] == (u64
)0 && mdev
->p_uuid
[UI_BITMAP
] != (u64
)0) {
2684 if (mdev
->tconn
->agreed_pro_version
< 91)
2687 if ((mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1)) &&
2688 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1))) {
2689 dev_info(DEV
, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2691 mdev
->p_uuid
[UI_HISTORY_START
+ 1] = mdev
->p_uuid
[UI_HISTORY_START
];
2692 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_BITMAP
];
2693 mdev
->p_uuid
[UI_BITMAP
] = 0UL;
2695 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2698 dev_info(DEV
, "was SyncTarget (failed to write sync_uuid)\n");
2705 /* Common power [off|failure] */
2706 rct
= (test_bit(CRASHED_PRIMARY
, &mdev
->flags
) ? 1 : 0) +
2707 (mdev
->p_uuid
[UI_FLAGS
] & 2);
2708 /* lowest bit is set when we were primary,
2709 * next bit (weight 2) is set when peer was primary */
2713 case 0: /* !self_pri && !peer_pri */ return 0;
2714 case 1: /* self_pri && !peer_pri */ return 1;
2715 case 2: /* !self_pri && peer_pri */ return -1;
2716 case 3: /* self_pri && peer_pri */
2717 dc
= test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
);
2723 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2728 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2730 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2731 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) ==
2732 (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) :
2733 peer
+ UUID_NEW_BM_OFFSET
== (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1))) {
2734 /* The last P_SYNC_UUID did not get though. Undo the last start of
2735 resync as sync source modifications of the peer's UUIDs. */
2737 if (mdev
->tconn
->agreed_pro_version
< 91)
2740 mdev
->p_uuid
[UI_BITMAP
] = mdev
->p_uuid
[UI_HISTORY_START
];
2741 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_HISTORY_START
+ 1];
2743 dev_info(DEV
, "Did not got last syncUUID packet, corrected:\n");
2744 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2751 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2752 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2753 peer
= mdev
->p_uuid
[i
] & ~((u64
)1);
2759 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2760 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2765 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2767 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2768 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) ==
2769 (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) :
2770 self
+ UUID_NEW_BM_OFFSET
== (mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1))) {
2771 /* The last P_SYNC_UUID did not get though. Undo the last start of
2772 resync as sync source modifications of our UUIDs. */
2774 if (mdev
->tconn
->agreed_pro_version
< 91)
2777 _drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
]);
2778 _drbd_uuid_set(mdev
, UI_HISTORY_START
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1]);
2780 dev_info(DEV
, "Last syncUUID did not get through, corrected:\n");
2781 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2782 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2790 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2791 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2792 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2798 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2799 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2800 if (self
== peer
&& self
!= ((u64
)0))
2804 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2805 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2806 for (j
= UI_HISTORY_START
; j
<= UI_HISTORY_END
; j
++) {
2807 peer
= mdev
->p_uuid
[j
] & ~((u64
)1);
2816 /* drbd_sync_handshake() returns the new conn state on success, or
2817 CONN_MASK (-1) on failure.
2819 static enum drbd_conns
drbd_sync_handshake(struct drbd_conf
*mdev
, enum drbd_role peer_role
,
2820 enum drbd_disk_state peer_disk
) __must_hold(local
)
2822 enum drbd_conns rv
= C_MASK
;
2823 enum drbd_disk_state mydisk
;
2824 struct net_conf
*nc
;
2825 int hg
, rule_nr
, rr_conflict
, dry_run
;
2827 mydisk
= mdev
->state
.disk
;
2828 if (mydisk
== D_NEGOTIATING
)
2829 mydisk
= mdev
->new_state_tmp
.disk
;
2831 dev_info(DEV
, "drbd_sync_handshake:\n");
2832 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
, mdev
->comm_bm_set
, 0);
2833 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
,
2834 mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2836 hg
= drbd_uuid_compare(mdev
, &rule_nr
);
2838 dev_info(DEV
, "uuid_compare()=%d by rule %d\n", hg
, rule_nr
);
2841 dev_alert(DEV
, "Unrelated data, aborting!\n");
2845 dev_alert(DEV
, "To resolve this both sides have to support at least protocol %d\n", -hg
- 1000);
2849 if ((mydisk
== D_INCONSISTENT
&& peer_disk
> D_INCONSISTENT
) ||
2850 (peer_disk
== D_INCONSISTENT
&& mydisk
> D_INCONSISTENT
)) {
2851 int f
= (hg
== -100) || abs(hg
) == 2;
2852 hg
= mydisk
> D_INCONSISTENT
? 1 : -1;
2855 dev_info(DEV
, "Becoming sync %s due to disk states.\n",
2856 hg
> 0 ? "source" : "target");
2860 drbd_khelper(mdev
, "initial-split-brain");
2863 nc
= rcu_dereference(mdev
->tconn
->net_conf
);
2865 if (hg
== 100 || (hg
== -100 && nc
->always_asbp
)) {
2866 int pcount
= (mdev
->state
.role
== R_PRIMARY
)
2867 + (peer_role
== R_PRIMARY
);
2868 int forced
= (hg
== -100);
2872 hg
= drbd_asb_recover_0p(mdev
);
2875 hg
= drbd_asb_recover_1p(mdev
);
2878 hg
= drbd_asb_recover_2p(mdev
);
2881 if (abs(hg
) < 100) {
2882 dev_warn(DEV
, "Split-Brain detected, %d primaries, "
2883 "automatically solved. Sync from %s node\n",
2884 pcount
, (hg
< 0) ? "peer" : "this");
2886 dev_warn(DEV
, "Doing a full sync, since"
2887 " UUIDs where ambiguous.\n");
2894 if (nc
->want_lose
&& !(mdev
->p_uuid
[UI_FLAGS
]&1))
2896 if (!nc
->want_lose
&& (mdev
->p_uuid
[UI_FLAGS
]&1))
2900 dev_warn(DEV
, "Split-Brain detected, manually solved. "
2901 "Sync from %s node\n",
2902 (hg
< 0) ? "peer" : "this");
2904 rr_conflict
= nc
->rr_conflict
;
2905 dry_run
= nc
->dry_run
;
2909 /* FIXME this log message is not correct if we end up here
2910 * after an attempted attach on a diskless node.
2911 * We just refuse to attach -- well, we drop the "connection"
2912 * to that disk, in a way... */
2913 dev_alert(DEV
, "Split-Brain detected but unresolved, dropping connection!\n");
2914 drbd_khelper(mdev
, "split-brain");
2918 if (hg
> 0 && mydisk
<= D_INCONSISTENT
) {
2919 dev_err(DEV
, "I shall become SyncSource, but I am inconsistent!\n");
2923 if (hg
< 0 && /* by intention we do not use mydisk here. */
2924 mdev
->state
.role
== R_PRIMARY
&& mdev
->state
.disk
>= D_CONSISTENT
) {
2925 switch (rr_conflict
) {
2926 case ASB_CALL_HELPER
:
2927 drbd_khelper(mdev
, "pri-lost");
2929 case ASB_DISCONNECT
:
2930 dev_err(DEV
, "I shall become SyncTarget, but I am primary!\n");
2933 dev_warn(DEV
, "Becoming SyncTarget, violating the stable-data"
2938 if (dry_run
|| test_bit(CONN_DRY_RUN
, &mdev
->tconn
->flags
)) {
2940 dev_info(DEV
, "dry-run connect: No resync, would become Connected immediately.\n");
2942 dev_info(DEV
, "dry-run connect: Would become %s, doing a %s resync.",
2943 drbd_conn_str(hg
> 0 ? C_SYNC_SOURCE
: C_SYNC_TARGET
),
2944 abs(hg
) >= 2 ? "full" : "bit-map based");
2949 dev_info(DEV
, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2950 if (drbd_bitmap_io(mdev
, &drbd_bmio_set_n_write
, "set_n_write from sync_handshake",
2951 BM_LOCKED_SET_ALLOWED
))
2955 if (hg
> 0) { /* become sync source. */
2957 } else if (hg
< 0) { /* become sync target */
2961 if (drbd_bm_total_weight(mdev
)) {
2962 dev_info(DEV
, "No resync, but %lu bits in bitmap!\n",
2963 drbd_bm_total_weight(mdev
));
2970 /* returns 1 if invalid */
2971 static int cmp_after_sb(enum drbd_after_sb_p peer
, enum drbd_after_sb_p self
)
2973 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2974 if ((peer
== ASB_DISCARD_REMOTE
&& self
== ASB_DISCARD_LOCAL
) ||
2975 (self
== ASB_DISCARD_REMOTE
&& peer
== ASB_DISCARD_LOCAL
))
2978 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2979 if (peer
== ASB_DISCARD_REMOTE
|| peer
== ASB_DISCARD_LOCAL
||
2980 self
== ASB_DISCARD_REMOTE
|| self
== ASB_DISCARD_LOCAL
)
2983 /* everything else is valid if they are equal on both sides. */
2987 /* everything es is invalid. */
2991 static int receive_protocol(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
2993 struct p_protocol
*p
= pi
->data
;
2994 int p_proto
, p_after_sb_0p
, p_after_sb_1p
, p_after_sb_2p
;
2995 int p_want_lose
, p_two_primaries
, cf
;
2996 struct net_conf
*nc
;
2998 p_proto
= be32_to_cpu(p
->protocol
);
2999 p_after_sb_0p
= be32_to_cpu(p
->after_sb_0p
);
3000 p_after_sb_1p
= be32_to_cpu(p
->after_sb_1p
);
3001 p_after_sb_2p
= be32_to_cpu(p
->after_sb_2p
);
3002 p_two_primaries
= be32_to_cpu(p
->two_primaries
);
3003 cf
= be32_to_cpu(p
->conn_flags
);
3004 p_want_lose
= cf
& CF_WANT_LOSE
;
3006 if (tconn
->agreed_pro_version
>= 87) {
3007 char integrity_alg
[SHARED_SECRET_MAX
];
3008 struct crypto_hash
*tfm
= NULL
;
3011 if (pi
->size
> sizeof(integrity_alg
))
3013 err
= drbd_recv_all(tconn
, integrity_alg
, pi
->size
);
3016 integrity_alg
[SHARED_SECRET_MAX
-1] = 0;
3018 if (integrity_alg
[0]) {
3019 tfm
= crypto_alloc_hash(integrity_alg
, 0, CRYPTO_ALG_ASYNC
);
3021 conn_err(tconn
, "peer data-integrity-alg %s not supported\n",
3025 conn_info(tconn
, "peer data-integrity-alg: %s\n", integrity_alg
);
3028 if (tconn
->peer_integrity_tfm
)
3029 crypto_free_hash(tconn
->peer_integrity_tfm
);
3030 tconn
->peer_integrity_tfm
= tfm
;
3033 clear_bit(CONN_DRY_RUN
, &tconn
->flags
);
3035 if (cf
& CF_DRY_RUN
)
3036 set_bit(CONN_DRY_RUN
, &tconn
->flags
);
3039 nc
= rcu_dereference(tconn
->net_conf
);
3041 if (p_proto
!= nc
->wire_protocol
&& tconn
->agreed_pro_version
< 100) {
3042 conn_err(tconn
, "incompatible communication protocols\n");
3043 goto disconnect_rcu_unlock
;
3046 if (cmp_after_sb(p_after_sb_0p
, nc
->after_sb_0p
)) {
3047 conn_err(tconn
, "incompatible after-sb-0pri settings\n");
3048 goto disconnect_rcu_unlock
;
3051 if (cmp_after_sb(p_after_sb_1p
, nc
->after_sb_1p
)) {
3052 conn_err(tconn
, "incompatible after-sb-1pri settings\n");
3053 goto disconnect_rcu_unlock
;
3056 if (cmp_after_sb(p_after_sb_2p
, nc
->after_sb_2p
)) {
3057 conn_err(tconn
, "incompatible after-sb-2pri settings\n");
3058 goto disconnect_rcu_unlock
;
3061 if (p_want_lose
&& nc
->want_lose
) {
3062 conn_err(tconn
, "both sides have the 'want_lose' flag set\n");
3063 goto disconnect_rcu_unlock
;
3066 if (p_two_primaries
!= nc
->two_primaries
) {
3067 conn_err(tconn
, "incompatible setting of the two-primaries options\n");
3068 goto disconnect_rcu_unlock
;
3075 disconnect_rcu_unlock
:
3078 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3083 * input: alg name, feature name
3084 * return: NULL (alg name was "")
3085 * ERR_PTR(error) if something goes wrong
3086 * or the crypto hash ptr, if it worked out ok. */
3087 struct crypto_hash
*drbd_crypto_alloc_digest_safe(const struct drbd_conf
*mdev
,
3088 const char *alg
, const char *name
)
3090 struct crypto_hash
*tfm
;
3095 tfm
= crypto_alloc_hash(alg
, 0, CRYPTO_ALG_ASYNC
);
3097 dev_err(DEV
, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3098 alg
, name
, PTR_ERR(tfm
));
3104 static int ignore_remaining_packet(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3106 void *buffer
= tconn
->data
.rbuf
;
3107 int size
= pi
->size
;
3110 int s
= min_t(int, size
, DRBD_SOCKET_BUFFER_SIZE
);
3111 s
= drbd_recv(tconn
, buffer
, s
);
3125 * config_unknown_volume - device configuration command for unknown volume
3127 * When a device is added to an existing connection, the node on which the
3128 * device is added first will send configuration commands to its peer but the
3129 * peer will not know about the device yet. It will warn and ignore these
3130 * commands. Once the device is added on the second node, the second node will
3131 * send the same device configuration commands, but in the other direction.
3133 * (We can also end up here if drbd is misconfigured.)
3135 static int config_unknown_volume(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3137 conn_warn(tconn
, "Volume %u unknown; ignoring %s packet\n",
3138 pi
->vnr
, cmdname(pi
->cmd
));
3139 return ignore_remaining_packet(tconn
, pi
);
3142 static int receive_SyncParam(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3144 struct drbd_conf
*mdev
;
3145 struct p_rs_param_95
*p
;
3146 unsigned int header_size
, data_size
, exp_max_sz
;
3147 struct crypto_hash
*verify_tfm
= NULL
;
3148 struct crypto_hash
*csums_tfm
= NULL
;
3149 struct net_conf
*old_conf
, *new_conf
= NULL
;
3150 const int apv
= tconn
->agreed_pro_version
;
3151 int *rs_plan_s
= NULL
;
3155 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3157 return config_unknown_volume(tconn
, pi
);
3159 exp_max_sz
= apv
<= 87 ? sizeof(struct p_rs_param
)
3160 : apv
== 88 ? sizeof(struct p_rs_param
)
3162 : apv
<= 94 ? sizeof(struct p_rs_param_89
)
3163 : /* apv >= 95 */ sizeof(struct p_rs_param_95
);
3165 if (pi
->size
> exp_max_sz
) {
3166 dev_err(DEV
, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3167 pi
->size
, exp_max_sz
);
3172 header_size
= sizeof(struct p_rs_param
);
3173 data_size
= pi
->size
- header_size
;
3174 } else if (apv
<= 94) {
3175 header_size
= sizeof(struct p_rs_param_89
);
3176 data_size
= pi
->size
- header_size
;
3177 D_ASSERT(data_size
== 0);
3179 header_size
= sizeof(struct p_rs_param_95
);
3180 data_size
= pi
->size
- header_size
;
3181 D_ASSERT(data_size
== 0);
3184 /* initialize verify_alg and csums_alg */
3186 memset(p
->verify_alg
, 0, 2 * SHARED_SECRET_MAX
);
3188 err
= drbd_recv_all(mdev
->tconn
, p
, header_size
);
3192 if (get_ldev(mdev
)) {
3193 mdev
->ldev
->dc
.resync_rate
= be32_to_cpu(p
->rate
);
3199 if (data_size
> SHARED_SECRET_MAX
) {
3200 dev_err(DEV
, "verify-alg too long, "
3201 "peer wants %u, accepting only %u byte\n",
3202 data_size
, SHARED_SECRET_MAX
);
3206 err
= drbd_recv_all(mdev
->tconn
, p
->verify_alg
, data_size
);
3210 /* we expect NUL terminated string */
3211 /* but just in case someone tries to be evil */
3212 D_ASSERT(p
->verify_alg
[data_size
-1] == 0);
3213 p
->verify_alg
[data_size
-1] = 0;
3215 } else /* apv >= 89 */ {
3216 /* we still expect NUL terminated strings */
3217 /* but just in case someone tries to be evil */
3218 D_ASSERT(p
->verify_alg
[SHARED_SECRET_MAX
-1] == 0);
3219 D_ASSERT(p
->csums_alg
[SHARED_SECRET_MAX
-1] == 0);
3220 p
->verify_alg
[SHARED_SECRET_MAX
-1] = 0;
3221 p
->csums_alg
[SHARED_SECRET_MAX
-1] = 0;
3224 mutex_lock(&mdev
->tconn
->conf_update
);
3225 old_conf
= mdev
->tconn
->net_conf
;
3227 if (strcmp(old_conf
->verify_alg
, p
->verify_alg
)) {
3228 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
3229 dev_err(DEV
, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3230 old_conf
->verify_alg
, p
->verify_alg
);
3233 verify_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
3234 p
->verify_alg
, "verify-alg");
3235 if (IS_ERR(verify_tfm
)) {
3241 if (apv
>= 89 && strcmp(old_conf
->csums_alg
, p
->csums_alg
)) {
3242 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
3243 dev_err(DEV
, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3244 old_conf
->csums_alg
, p
->csums_alg
);
3247 csums_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
3248 p
->csums_alg
, "csums-alg");
3249 if (IS_ERR(csums_tfm
)) {
3255 if (apv
> 94 && get_ldev(mdev
)) {
3256 mdev
->ldev
->dc
.resync_rate
= be32_to_cpu(p
->rate
);
3257 mdev
->ldev
->dc
.c_plan_ahead
= be32_to_cpu(p
->c_plan_ahead
);
3258 mdev
->ldev
->dc
.c_delay_target
= be32_to_cpu(p
->c_delay_target
);
3259 mdev
->ldev
->dc
.c_fill_target
= be32_to_cpu(p
->c_fill_target
);
3260 mdev
->ldev
->dc
.c_max_rate
= be32_to_cpu(p
->c_max_rate
);
3262 fifo_size
= (mdev
->ldev
->dc
.c_plan_ahead
* 10 * SLEEP_TIME
) / HZ
;
3263 if (fifo_size
!= mdev
->rs_plan_s
.size
&& fifo_size
> 0) {
3264 rs_plan_s
= kzalloc(sizeof(int) * fifo_size
, GFP_KERNEL
);
3266 dev_err(DEV
, "kmalloc of fifo_buffer failed");
3274 if (verify_tfm
|| csums_tfm
) {
3275 new_conf
= kzalloc(sizeof(struct net_conf
), GFP_KERNEL
);
3277 dev_err(DEV
, "Allocation of new net_conf failed\n");
3281 *new_conf
= *old_conf
;
3284 strcpy(new_conf
->verify_alg
, p
->verify_alg
);
3285 new_conf
->verify_alg_len
= strlen(p
->verify_alg
) + 1;
3286 crypto_free_hash(mdev
->tconn
->verify_tfm
);
3287 mdev
->tconn
->verify_tfm
= verify_tfm
;
3288 dev_info(DEV
, "using verify-alg: \"%s\"\n", p
->verify_alg
);
3291 strcpy(new_conf
->csums_alg
, p
->csums_alg
);
3292 new_conf
->csums_alg_len
= strlen(p
->csums_alg
) + 1;
3293 crypto_free_hash(mdev
->tconn
->csums_tfm
);
3294 mdev
->tconn
->csums_tfm
= csums_tfm
;
3295 dev_info(DEV
, "using csums-alg: \"%s\"\n", p
->csums_alg
);
3297 rcu_assign_pointer(tconn
->net_conf
, new_conf
);
3299 mutex_unlock(&mdev
->tconn
->conf_update
);
3305 spin_lock(&mdev
->peer_seq_lock
);
3306 if (fifo_size
!= mdev
->rs_plan_s
.size
) {
3307 kfree(mdev
->rs_plan_s
.values
);
3308 mdev
->rs_plan_s
.values
= rs_plan_s
;
3309 mdev
->rs_plan_s
.size
= fifo_size
;
3310 mdev
->rs_planed
= 0;
3312 spin_unlock(&mdev
->peer_seq_lock
);
3317 mutex_unlock(&mdev
->tconn
->conf_update
);
3318 /* just for completeness: actually not needed,
3319 * as this is not reached if csums_tfm was ok. */
3320 crypto_free_hash(csums_tfm
);
3321 /* but free the verify_tfm again, if csums_tfm did not work out */
3322 crypto_free_hash(verify_tfm
);
3323 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3327 /* warn if the arguments differ by more than 12.5% */
3328 static void warn_if_differ_considerably(struct drbd_conf
*mdev
,
3329 const char *s
, sector_t a
, sector_t b
)
3332 if (a
== 0 || b
== 0)
3334 d
= (a
> b
) ? (a
- b
) : (b
- a
);
3335 if (d
> (a
>>3) || d
> (b
>>3))
3336 dev_warn(DEV
, "Considerable difference in %s: %llus vs. %llus\n", s
,
3337 (unsigned long long)a
, (unsigned long long)b
);
3340 static int receive_sizes(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3342 struct drbd_conf
*mdev
;
3343 struct p_sizes
*p
= pi
->data
;
3344 enum determine_dev_size dd
= unchanged
;
3345 sector_t p_size
, p_usize
, my_usize
;
3346 int ldsc
= 0; /* local disk size changed */
3347 enum dds_flags ddsf
;
3349 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3351 return config_unknown_volume(tconn
, pi
);
3353 p_size
= be64_to_cpu(p
->d_size
);
3354 p_usize
= be64_to_cpu(p
->u_size
);
3356 /* just store the peer's disk size for now.
3357 * we still need to figure out whether we accept that. */
3358 mdev
->p_size
= p_size
;
3360 if (get_ldev(mdev
)) {
3361 warn_if_differ_considerably(mdev
, "lower level device sizes",
3362 p_size
, drbd_get_max_capacity(mdev
->ldev
));
3363 warn_if_differ_considerably(mdev
, "user requested size",
3364 p_usize
, mdev
->ldev
->dc
.disk_size
);
3366 /* if this is the first connect, or an otherwise expected
3367 * param exchange, choose the minimum */
3368 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
)
3369 p_usize
= min_not_zero((sector_t
)mdev
->ldev
->dc
.disk_size
,
3372 my_usize
= mdev
->ldev
->dc
.disk_size
;
3374 if (mdev
->ldev
->dc
.disk_size
!= p_usize
) {
3375 mdev
->ldev
->dc
.disk_size
= p_usize
;
3376 dev_info(DEV
, "Peer sets u_size to %lu sectors\n",
3377 (unsigned long)mdev
->ldev
->dc
.disk_size
);
3380 /* Never shrink a device with usable data during connect.
3381 But allow online shrinking if we are connected. */
3382 if (drbd_new_dev_size(mdev
, mdev
->ldev
, p_usize
, 0) <
3383 drbd_get_capacity(mdev
->this_bdev
) &&
3384 mdev
->state
.disk
>= D_OUTDATED
&&
3385 mdev
->state
.conn
< C_CONNECTED
) {
3386 dev_err(DEV
, "The peer's disk size is too small!\n");
3387 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3388 mdev
->ldev
->dc
.disk_size
= my_usize
;
3395 ddsf
= be16_to_cpu(p
->dds_flags
);
3396 if (get_ldev(mdev
)) {
3397 dd
= drbd_determine_dev_size(mdev
, ddsf
);
3399 if (dd
== dev_size_error
)
3403 /* I am diskless, need to accept the peer's size. */
3404 drbd_set_my_capacity(mdev
, p_size
);
3407 mdev
->peer_max_bio_size
= be32_to_cpu(p
->max_bio_size
);
3408 drbd_reconsider_max_bio_size(mdev
);
3410 if (get_ldev(mdev
)) {
3411 if (mdev
->ldev
->known_size
!= drbd_get_capacity(mdev
->ldev
->backing_bdev
)) {
3412 mdev
->ldev
->known_size
= drbd_get_capacity(mdev
->ldev
->backing_bdev
);
3419 if (mdev
->state
.conn
> C_WF_REPORT_PARAMS
) {
3420 if (be64_to_cpu(p
->c_size
) !=
3421 drbd_get_capacity(mdev
->this_bdev
) || ldsc
) {
3422 /* we have different sizes, probably peer
3423 * needs to know my new size... */
3424 drbd_send_sizes(mdev
, 0, ddsf
);
3426 if (test_and_clear_bit(RESIZE_PENDING
, &mdev
->flags
) ||
3427 (dd
== grew
&& mdev
->state
.conn
== C_CONNECTED
)) {
3428 if (mdev
->state
.pdsk
>= D_INCONSISTENT
&&
3429 mdev
->state
.disk
>= D_INCONSISTENT
) {
3430 if (ddsf
& DDSF_NO_RESYNC
)
3431 dev_info(DEV
, "Resync of new storage suppressed with --assume-clean\n");
3433 resync_after_online_grow(mdev
);
3435 set_bit(RESYNC_AFTER_NEG
, &mdev
->flags
);
3442 static int receive_uuids(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3444 struct drbd_conf
*mdev
;
3445 struct p_uuids
*p
= pi
->data
;
3447 int i
, updated_uuids
= 0;
3449 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3451 return config_unknown_volume(tconn
, pi
);
3453 p_uuid
= kmalloc(sizeof(u64
)*UI_EXTENDED_SIZE
, GFP_NOIO
);
3455 for (i
= UI_CURRENT
; i
< UI_EXTENDED_SIZE
; i
++)
3456 p_uuid
[i
] = be64_to_cpu(p
->uuid
[i
]);
3458 kfree(mdev
->p_uuid
);
3459 mdev
->p_uuid
= p_uuid
;
3461 if (mdev
->state
.conn
< C_CONNECTED
&&
3462 mdev
->state
.disk
< D_INCONSISTENT
&&
3463 mdev
->state
.role
== R_PRIMARY
&&
3464 (mdev
->ed_uuid
& ~((u64
)1)) != (p_uuid
[UI_CURRENT
] & ~((u64
)1))) {
3465 dev_err(DEV
, "Can only connect to data with current UUID=%016llX\n",
3466 (unsigned long long)mdev
->ed_uuid
);
3467 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3471 if (get_ldev(mdev
)) {
3472 int skip_initial_sync
=
3473 mdev
->state
.conn
== C_CONNECTED
&&
3474 mdev
->tconn
->agreed_pro_version
>= 90 &&
3475 mdev
->ldev
->md
.uuid
[UI_CURRENT
] == UUID_JUST_CREATED
&&
3476 (p_uuid
[UI_FLAGS
] & 8);
3477 if (skip_initial_sync
) {
3478 dev_info(DEV
, "Accepted new current UUID, preparing to skip initial sync\n");
3479 drbd_bitmap_io(mdev
, &drbd_bmio_clear_n_write
,
3480 "clear_n_write from receive_uuids",
3481 BM_LOCKED_TEST_ALLOWED
);
3482 _drbd_uuid_set(mdev
, UI_CURRENT
, p_uuid
[UI_CURRENT
]);
3483 _drbd_uuid_set(mdev
, UI_BITMAP
, 0);
3484 _drbd_set_state(_NS2(mdev
, disk
, D_UP_TO_DATE
, pdsk
, D_UP_TO_DATE
),
3490 } else if (mdev
->state
.disk
< D_INCONSISTENT
&&
3491 mdev
->state
.role
== R_PRIMARY
) {
3492 /* I am a diskless primary, the peer just created a new current UUID
3494 updated_uuids
= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3497 /* Before we test for the disk state, we should wait until an eventually
3498 ongoing cluster wide state change is finished. That is important if
3499 we are primary and are detaching from our disk. We need to see the
3500 new disk state... */
3501 mutex_lock(mdev
->state_mutex
);
3502 mutex_unlock(mdev
->state_mutex
);
3503 if (mdev
->state
.conn
>= C_CONNECTED
&& mdev
->state
.disk
< D_INCONSISTENT
)
3504 updated_uuids
|= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3507 drbd_print_uuids(mdev
, "receiver updated UUIDs to");
3513 * convert_state() - Converts the peer's view of the cluster state to our point of view
3514 * @ps: The state as seen by the peer.
3516 static union drbd_state
convert_state(union drbd_state ps
)
3518 union drbd_state ms
;
3520 static enum drbd_conns c_tab
[] = {
3521 [C_CONNECTED
] = C_CONNECTED
,
3523 [C_STARTING_SYNC_S
] = C_STARTING_SYNC_T
,
3524 [C_STARTING_SYNC_T
] = C_STARTING_SYNC_S
,
3525 [C_DISCONNECTING
] = C_TEAR_DOWN
, /* C_NETWORK_FAILURE, */
3526 [C_VERIFY_S
] = C_VERIFY_T
,
3532 ms
.conn
= c_tab
[ps
.conn
];
3537 ms
.peer_isp
= (ps
.aftr_isp
| ps
.user_isp
);
3542 static int receive_req_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3544 struct drbd_conf
*mdev
;
3545 struct p_req_state
*p
= pi
->data
;
3546 union drbd_state mask
, val
;
3547 enum drbd_state_rv rv
;
3549 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3553 mask
.i
= be32_to_cpu(p
->mask
);
3554 val
.i
= be32_to_cpu(p
->val
);
3556 if (test_bit(DISCARD_CONCURRENT
, &mdev
->tconn
->flags
) &&
3557 mutex_is_locked(mdev
->state_mutex
)) {
3558 drbd_send_sr_reply(mdev
, SS_CONCURRENT_ST_CHG
);
3562 mask
= convert_state(mask
);
3563 val
= convert_state(val
);
3565 rv
= drbd_change_state(mdev
, CS_VERBOSE
, mask
, val
);
3566 drbd_send_sr_reply(mdev
, rv
);
3573 static int receive_req_conn_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3575 struct p_req_state
*p
= pi
->data
;
3576 union drbd_state mask
, val
;
3577 enum drbd_state_rv rv
;
3579 mask
.i
= be32_to_cpu(p
->mask
);
3580 val
.i
= be32_to_cpu(p
->val
);
3582 if (test_bit(DISCARD_CONCURRENT
, &tconn
->flags
) &&
3583 mutex_is_locked(&tconn
->cstate_mutex
)) {
3584 conn_send_sr_reply(tconn
, SS_CONCURRENT_ST_CHG
);
3588 mask
= convert_state(mask
);
3589 val
= convert_state(val
);
3591 rv
= conn_request_state(tconn
, mask
, val
, CS_VERBOSE
| CS_LOCAL_ONLY
| CS_IGN_OUTD_FAIL
);
3592 conn_send_sr_reply(tconn
, rv
);
3597 static int receive_state(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3599 struct drbd_conf
*mdev
;
3600 struct p_state
*p
= pi
->data
;
3601 union drbd_state os
, ns
, peer_state
;
3602 enum drbd_disk_state real_peer_disk
;
3603 enum chg_state_flags cs_flags
;
3606 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3608 return config_unknown_volume(tconn
, pi
);
3610 peer_state
.i
= be32_to_cpu(p
->state
);
3612 real_peer_disk
= peer_state
.disk
;
3613 if (peer_state
.disk
== D_NEGOTIATING
) {
3614 real_peer_disk
= mdev
->p_uuid
[UI_FLAGS
] & 4 ? D_INCONSISTENT
: D_CONSISTENT
;
3615 dev_info(DEV
, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk
));
3618 spin_lock_irq(&mdev
->tconn
->req_lock
);
3620 os
= ns
= drbd_read_state(mdev
);
3621 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3623 /* peer says his disk is uptodate, while we think it is inconsistent,
3624 * and this happens while we think we have a sync going on. */
3625 if (os
.pdsk
== D_INCONSISTENT
&& real_peer_disk
== D_UP_TO_DATE
&&
3626 os
.conn
> C_CONNECTED
&& os
.disk
== D_UP_TO_DATE
) {
3627 /* If we are (becoming) SyncSource, but peer is still in sync
3628 * preparation, ignore its uptodate-ness to avoid flapping, it
3629 * will change to inconsistent once the peer reaches active
3631 * It may have changed syncer-paused flags, however, so we
3632 * cannot ignore this completely. */
3633 if (peer_state
.conn
> C_CONNECTED
&&
3634 peer_state
.conn
< C_SYNC_SOURCE
)
3635 real_peer_disk
= D_INCONSISTENT
;
3637 /* if peer_state changes to connected at the same time,
3638 * it explicitly notifies us that it finished resync.
3639 * Maybe we should finish it up, too? */
3640 else if (os
.conn
>= C_SYNC_SOURCE
&&
3641 peer_state
.conn
== C_CONNECTED
) {
3642 if (drbd_bm_total_weight(mdev
) <= mdev
->rs_failed
)
3643 drbd_resync_finished(mdev
);
3648 /* peer says his disk is inconsistent, while we think it is uptodate,
3649 * and this happens while the peer still thinks we have a sync going on,
3650 * but we think we are already done with the sync.
3651 * We ignore this to avoid flapping pdsk.
3652 * This should not happen, if the peer is a recent version of drbd. */
3653 if (os
.pdsk
== D_UP_TO_DATE
&& real_peer_disk
== D_INCONSISTENT
&&
3654 os
.conn
== C_CONNECTED
&& peer_state
.conn
> C_SYNC_SOURCE
)
3655 real_peer_disk
= D_UP_TO_DATE
;
3657 if (ns
.conn
== C_WF_REPORT_PARAMS
)
3658 ns
.conn
= C_CONNECTED
;
3660 if (peer_state
.conn
== C_AHEAD
)
3663 if (mdev
->p_uuid
&& peer_state
.disk
>= D_NEGOTIATING
&&
3664 get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3665 int cr
; /* consider resync */
3667 /* if we established a new connection */
3668 cr
= (os
.conn
< C_CONNECTED
);
3669 /* if we had an established connection
3670 * and one of the nodes newly attaches a disk */
3671 cr
|= (os
.conn
== C_CONNECTED
&&
3672 (peer_state
.disk
== D_NEGOTIATING
||
3673 os
.disk
== D_NEGOTIATING
));
3674 /* if we have both been inconsistent, and the peer has been
3675 * forced to be UpToDate with --overwrite-data */
3676 cr
|= test_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3677 /* if we had been plain connected, and the admin requested to
3678 * start a sync by "invalidate" or "invalidate-remote" */
3679 cr
|= (os
.conn
== C_CONNECTED
&&
3680 (peer_state
.conn
>= C_STARTING_SYNC_S
&&
3681 peer_state
.conn
<= C_WF_BITMAP_T
));
3684 ns
.conn
= drbd_sync_handshake(mdev
, peer_state
.role
, real_peer_disk
);
3687 if (ns
.conn
== C_MASK
) {
3688 ns
.conn
= C_CONNECTED
;
3689 if (mdev
->state
.disk
== D_NEGOTIATING
) {
3690 drbd_force_state(mdev
, NS(disk
, D_FAILED
));
3691 } else if (peer_state
.disk
== D_NEGOTIATING
) {
3692 dev_err(DEV
, "Disk attach process on the peer node was aborted.\n");
3693 peer_state
.disk
= D_DISKLESS
;
3694 real_peer_disk
= D_DISKLESS
;
3696 if (test_and_clear_bit(CONN_DRY_RUN
, &mdev
->tconn
->flags
))
3698 D_ASSERT(os
.conn
== C_WF_REPORT_PARAMS
);
3699 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3705 spin_lock_irq(&mdev
->tconn
->req_lock
);
3706 if (os
.i
!= drbd_read_state(mdev
).i
)
3708 clear_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3709 ns
.peer
= peer_state
.role
;
3710 ns
.pdsk
= real_peer_disk
;
3711 ns
.peer_isp
= (peer_state
.aftr_isp
| peer_state
.user_isp
);
3712 if ((ns
.conn
== C_CONNECTED
|| ns
.conn
== C_WF_BITMAP_S
) && ns
.disk
== D_NEGOTIATING
)
3713 ns
.disk
= mdev
->new_state_tmp
.disk
;
3714 cs_flags
= CS_VERBOSE
+ (os
.conn
< C_CONNECTED
&& ns
.conn
>= C_CONNECTED
? 0 : CS_HARD
);
3715 if (ns
.pdsk
== D_CONSISTENT
&& drbd_suspended(mdev
) && ns
.conn
== C_CONNECTED
&& os
.conn
< C_CONNECTED
&&
3716 test_bit(NEW_CUR_UUID
, &mdev
->flags
)) {
3717 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
3718 for temporal network outages! */
3719 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3720 dev_err(DEV
, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3721 tl_clear(mdev
->tconn
);
3722 drbd_uuid_new_current(mdev
);
3723 clear_bit(NEW_CUR_UUID
, &mdev
->flags
);
3724 conn_request_state(mdev
->tconn
, NS2(conn
, C_PROTOCOL_ERROR
, susp
, 0), CS_HARD
);
3727 rv
= _drbd_set_state(mdev
, ns
, cs_flags
, NULL
);
3728 ns
= drbd_read_state(mdev
);
3729 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3731 if (rv
< SS_SUCCESS
) {
3732 conn_request_state(mdev
->tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
3736 if (os
.conn
> C_WF_REPORT_PARAMS
) {
3737 if (ns
.conn
> C_CONNECTED
&& peer_state
.conn
<= C_CONNECTED
&&
3738 peer_state
.disk
!= D_NEGOTIATING
) {
3739 /* we want resync, peer has not yet decided to sync... */
3740 /* Nowadays only used when forcing a node into primary role and
3741 setting its disk to UpToDate with that */
3742 drbd_send_uuids(mdev
);
3743 drbd_send_state(mdev
);
3747 mutex_lock(&mdev
->tconn
->conf_update
);
3748 mdev
->tconn
->net_conf
->want_lose
= 0; /* without copy; single bit op is atomic */
3749 mutex_unlock(&mdev
->tconn
->conf_update
);
3751 drbd_md_sync(mdev
); /* update connected indicator, la_size, ... */
3756 static int receive_sync_uuid(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3758 struct drbd_conf
*mdev
;
3759 struct p_rs_uuid
*p
= pi
->data
;
3761 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3765 wait_event(mdev
->misc_wait
,
3766 mdev
->state
.conn
== C_WF_SYNC_UUID
||
3767 mdev
->state
.conn
== C_BEHIND
||
3768 mdev
->state
.conn
< C_CONNECTED
||
3769 mdev
->state
.disk
< D_NEGOTIATING
);
3771 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3773 /* Here the _drbd_uuid_ functions are right, current should
3774 _not_ be rotated into the history */
3775 if (get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3776 _drbd_uuid_set(mdev
, UI_CURRENT
, be64_to_cpu(p
->uuid
));
3777 _drbd_uuid_set(mdev
, UI_BITMAP
, 0UL);
3779 drbd_print_uuids(mdev
, "updated sync uuid");
3780 drbd_start_resync(mdev
, C_SYNC_TARGET
);
3784 dev_err(DEV
, "Ignoring SyncUUID packet!\n");
3790 * receive_bitmap_plain
3792 * Return 0 when done, 1 when another iteration is needed, and a negative error
3793 * code upon failure.
3796 receive_bitmap_plain(struct drbd_conf
*mdev
, unsigned int size
,
3797 unsigned long *p
, struct bm_xfer_ctx
*c
)
3799 unsigned int data_size
= DRBD_SOCKET_BUFFER_SIZE
-
3800 drbd_header_size(mdev
->tconn
);
3801 unsigned int num_words
= min_t(size_t, data_size
/ sizeof(*p
),
3802 c
->bm_words
- c
->word_offset
);
3803 unsigned int want
= num_words
* sizeof(*p
);
3807 dev_err(DEV
, "%s:want (%u) != size (%u)\n", __func__
, want
, size
);
3812 err
= drbd_recv_all(mdev
->tconn
, p
, want
);
3816 drbd_bm_merge_lel(mdev
, c
->word_offset
, num_words
, p
);
3818 c
->word_offset
+= num_words
;
3819 c
->bit_offset
= c
->word_offset
* BITS_PER_LONG
;
3820 if (c
->bit_offset
> c
->bm_bits
)
3821 c
->bit_offset
= c
->bm_bits
;
3826 static enum drbd_bitmap_code
dcbp_get_code(struct p_compressed_bm
*p
)
3828 return (enum drbd_bitmap_code
)(p
->encoding
& 0x0f);
3831 static int dcbp_get_start(struct p_compressed_bm
*p
)
3833 return (p
->encoding
& 0x80) != 0;
3836 static int dcbp_get_pad_bits(struct p_compressed_bm
*p
)
3838 return (p
->encoding
>> 4) & 0x7;
3844 * Return 0 when done, 1 when another iteration is needed, and a negative error
3845 * code upon failure.
3848 recv_bm_rle_bits(struct drbd_conf
*mdev
,
3849 struct p_compressed_bm
*p
,
3850 struct bm_xfer_ctx
*c
,
3853 struct bitstream bs
;
3857 unsigned long s
= c
->bit_offset
;
3859 int toggle
= dcbp_get_start(p
);
3863 bitstream_init(&bs
, p
->code
, len
, dcbp_get_pad_bits(p
));
3865 bits
= bitstream_get_bits(&bs
, &look_ahead
, 64);
3869 for (have
= bits
; have
> 0; s
+= rl
, toggle
= !toggle
) {
3870 bits
= vli_decode_bits(&rl
, look_ahead
);
3876 if (e
>= c
->bm_bits
) {
3877 dev_err(DEV
, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e
);
3880 _drbd_bm_set_bits(mdev
, s
, e
);
3884 dev_err(DEV
, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3885 have
, bits
, look_ahead
,
3886 (unsigned int)(bs
.cur
.b
- p
->code
),
3887 (unsigned int)bs
.buf_len
);
3890 look_ahead
>>= bits
;
3893 bits
= bitstream_get_bits(&bs
, &tmp
, 64 - have
);
3896 look_ahead
|= tmp
<< have
;
3901 bm_xfer_ctx_bit_to_word_offset(c
);
3903 return (s
!= c
->bm_bits
);
3909 * Return 0 when done, 1 when another iteration is needed, and a negative error
3910 * code upon failure.
3913 decode_bitmap_c(struct drbd_conf
*mdev
,
3914 struct p_compressed_bm
*p
,
3915 struct bm_xfer_ctx
*c
,
3918 if (dcbp_get_code(p
) == RLE_VLI_Bits
)
3919 return recv_bm_rle_bits(mdev
, p
, c
, len
- sizeof(*p
));
3921 /* other variants had been implemented for evaluation,
3922 * but have been dropped as this one turned out to be "best"
3923 * during all our tests. */
3925 dev_err(DEV
, "receive_bitmap_c: unknown encoding %u\n", p
->encoding
);
3926 conn_request_state(mdev
->tconn
, NS(conn
, C_PROTOCOL_ERROR
), CS_HARD
);
3930 void INFO_bm_xfer_stats(struct drbd_conf
*mdev
,
3931 const char *direction
, struct bm_xfer_ctx
*c
)
3933 /* what would it take to transfer it "plaintext" */
3934 unsigned int header_size
= drbd_header_size(mdev
->tconn
);
3935 unsigned int data_size
= DRBD_SOCKET_BUFFER_SIZE
- header_size
;
3936 unsigned int plain
=
3937 header_size
* (DIV_ROUND_UP(c
->bm_words
, data_size
) + 1) +
3938 c
->bm_words
* sizeof(unsigned long);
3939 unsigned int total
= c
->bytes
[0] + c
->bytes
[1];
3942 /* total can not be zero. but just in case: */
3946 /* don't report if not compressed */
3950 /* total < plain. check for overflow, still */
3951 r
= (total
> UINT_MAX
/1000) ? (total
/ (plain
/1000))
3952 : (1000 * total
/ plain
);
3958 dev_info(DEV
, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3959 "total %u; compression: %u.%u%%\n",
3961 c
->bytes
[1], c
->packets
[1],
3962 c
->bytes
[0], c
->packets
[0],
3963 total
, r
/10, r
% 10);
3966 /* Since we are processing the bitfield from lower addresses to higher,
3967 it does not matter if the process it in 32 bit chunks or 64 bit
3968 chunks as long as it is little endian. (Understand it as byte stream,
3969 beginning with the lowest byte...) If we would use big endian
3970 we would need to process it from the highest address to the lowest,
3971 in order to be agnostic to the 32 vs 64 bits issue.
3973 returns 0 on failure, 1 if we successfully received it. */
3974 static int receive_bitmap(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
3976 struct drbd_conf
*mdev
;
3977 struct bm_xfer_ctx c
;
3980 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
3984 drbd_bm_lock(mdev
, "receive bitmap", BM_LOCKED_SET_ALLOWED
);
3985 /* you are supposed to send additional out-of-sync information
3986 * if you actually set bits during this phase */
3988 c
= (struct bm_xfer_ctx
) {
3989 .bm_bits
= drbd_bm_bits(mdev
),
3990 .bm_words
= drbd_bm_words(mdev
),
3994 if (pi
->cmd
== P_BITMAP
)
3995 err
= receive_bitmap_plain(mdev
, pi
->size
, pi
->data
, &c
);
3996 else if (pi
->cmd
== P_COMPRESSED_BITMAP
) {
3997 /* MAYBE: sanity check that we speak proto >= 90,
3998 * and the feature is enabled! */
3999 struct p_compressed_bm
*p
= pi
->data
;
4001 if (pi
->size
> DRBD_SOCKET_BUFFER_SIZE
- drbd_header_size(tconn
)) {
4002 dev_err(DEV
, "ReportCBitmap packet too large\n");
4006 if (pi
->size
<= sizeof(*p
)) {
4007 dev_err(DEV
, "ReportCBitmap packet too small (l:%u)\n", pi
->size
);
4011 err
= drbd_recv_all(mdev
->tconn
, p
, pi
->size
);
4014 err
= decode_bitmap_c(mdev
, p
, &c
, pi
->size
);
4016 dev_warn(DEV
, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi
->cmd
);
4021 c
.packets
[pi
->cmd
== P_BITMAP
]++;
4022 c
.bytes
[pi
->cmd
== P_BITMAP
] += drbd_header_size(tconn
) + pi
->size
;
4029 err
= drbd_recv_header(mdev
->tconn
, pi
);
4034 INFO_bm_xfer_stats(mdev
, "receive", &c
);
4036 if (mdev
->state
.conn
== C_WF_BITMAP_T
) {
4037 enum drbd_state_rv rv
;
4039 err
= drbd_send_bitmap(mdev
);
4042 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4043 rv
= _drbd_request_state(mdev
, NS(conn
, C_WF_SYNC_UUID
), CS_VERBOSE
);
4044 D_ASSERT(rv
== SS_SUCCESS
);
4045 } else if (mdev
->state
.conn
!= C_WF_BITMAP_S
) {
4046 /* admin may have requested C_DISCONNECTING,
4047 * other threads may have noticed network errors */
4048 dev_info(DEV
, "unexpected cstate (%s) in receive_bitmap\n",
4049 drbd_conn_str(mdev
->state
.conn
));
4054 drbd_bm_unlock(mdev
);
4055 if (!err
&& mdev
->state
.conn
== C_WF_BITMAP_S
)
4056 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
4060 static int receive_skip(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4062 conn_warn(tconn
, "skipping unknown optional packet type %d, l: %d!\n",
4065 return ignore_remaining_packet(tconn
, pi
);
4068 static int receive_UnplugRemote(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4070 /* Make sure we've acked all the TCP data associated
4071 * with the data requests being unplugged */
4072 drbd_tcp_quickack(tconn
->data
.socket
);
4077 static int receive_out_of_sync(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4079 struct drbd_conf
*mdev
;
4080 struct p_block_desc
*p
= pi
->data
;
4082 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4086 switch (mdev
->state
.conn
) {
4087 case C_WF_SYNC_UUID
:
4092 dev_err(DEV
, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4093 drbd_conn_str(mdev
->state
.conn
));
4096 drbd_set_out_of_sync(mdev
, be64_to_cpu(p
->sector
), be32_to_cpu(p
->blksize
));
4104 int (*fn
)(struct drbd_tconn
*, struct packet_info
*);
4107 static struct data_cmd drbd_cmd_handler
[] = {
4108 [P_DATA
] = { 1, sizeof(struct p_data
), receive_Data
},
4109 [P_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_DataReply
},
4110 [P_RS_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_RSDataReply
} ,
4111 [P_BARRIER
] = { 0, sizeof(struct p_barrier
), receive_Barrier
} ,
4112 [P_BITMAP
] = { 1, 0, receive_bitmap
} ,
4113 [P_COMPRESSED_BITMAP
] = { 1, 0, receive_bitmap
} ,
4114 [P_UNPLUG_REMOTE
] = { 0, 0, receive_UnplugRemote
},
4115 [P_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4116 [P_RS_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4117 [P_SYNC_PARAM
] = { 1, 0, receive_SyncParam
},
4118 [P_SYNC_PARAM89
] = { 1, 0, receive_SyncParam
},
4119 [P_PROTOCOL
] = { 1, sizeof(struct p_protocol
), receive_protocol
},
4120 [P_UUIDS
] = { 0, sizeof(struct p_uuids
), receive_uuids
},
4121 [P_SIZES
] = { 0, sizeof(struct p_sizes
), receive_sizes
},
4122 [P_STATE
] = { 0, sizeof(struct p_state
), receive_state
},
4123 [P_STATE_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_state
},
4124 [P_SYNC_UUID
] = { 0, sizeof(struct p_rs_uuid
), receive_sync_uuid
},
4125 [P_OV_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
4126 [P_OV_REPLY
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
4127 [P_CSUM_RS_REQUEST
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
4128 [P_DELAY_PROBE
] = { 0, sizeof(struct p_delay_probe93
), receive_skip
},
4129 [P_OUT_OF_SYNC
] = { 0, sizeof(struct p_block_desc
), receive_out_of_sync
},
4130 [P_CONN_ST_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_conn_state
},
4133 static void drbdd(struct drbd_tconn
*tconn
)
4135 struct packet_info pi
;
4136 size_t shs
; /* sub header size */
4139 while (get_t_state(&tconn
->receiver
) == RUNNING
) {
4140 struct data_cmd
*cmd
;
4142 drbd_thread_current_set_cpu(&tconn
->receiver
);
4143 if (drbd_recv_header(tconn
, &pi
))
4146 cmd
= &drbd_cmd_handler
[pi
.cmd
];
4147 if (unlikely(pi
.cmd
>= ARRAY_SIZE(drbd_cmd_handler
) || !cmd
->fn
)) {
4148 conn_err(tconn
, "unknown packet type %d, l: %d!\n", pi
.cmd
, pi
.size
);
4152 shs
= cmd
->pkt_size
;
4153 if (pi
.size
> shs
&& !cmd
->expect_payload
) {
4154 conn_err(tconn
, "No payload expected %s l:%d\n", cmdname(pi
.cmd
), pi
.size
);
4159 err
= drbd_recv_all_warn(tconn
, pi
.data
, shs
);
4165 err
= cmd
->fn(tconn
, &pi
);
4167 conn_err(tconn
, "error receiving %s, e: %d l: %d!\n",
4168 cmdname(pi
.cmd
), err
, pi
.size
);
4175 conn_request_state(tconn
, NS(conn
, C_PROTOCOL_ERROR
), CS_HARD
);
4178 void conn_flush_workqueue(struct drbd_tconn
*tconn
)
4180 struct drbd_wq_barrier barr
;
4182 barr
.w
.cb
= w_prev_work_done
;
4183 barr
.w
.tconn
= tconn
;
4184 init_completion(&barr
.done
);
4185 drbd_queue_work(&tconn
->data
.work
, &barr
.w
);
4186 wait_for_completion(&barr
.done
);
4189 static void drbd_disconnect(struct drbd_tconn
*tconn
)
4192 int rv
= SS_UNKNOWN_ERROR
;
4194 if (tconn
->cstate
== C_STANDALONE
)
4197 /* asender does not clean up anything. it must not interfere, either */
4198 drbd_thread_stop(&tconn
->asender
);
4199 drbd_free_sock(tconn
);
4201 down_read(&drbd_cfg_rwsem
);
4202 idr_for_each(&tconn
->volumes
, drbd_disconnected
, tconn
);
4203 up_read(&drbd_cfg_rwsem
);
4204 conn_info(tconn
, "Connection closed\n");
4206 if (conn_highest_role(tconn
) == R_PRIMARY
&& conn_highest_pdsk(tconn
) >= D_UNKNOWN
)
4207 conn_try_outdate_peer_async(tconn
);
4209 spin_lock_irq(&tconn
->req_lock
);
4211 if (oc
>= C_UNCONNECTED
)
4212 rv
= _conn_request_state(tconn
, NS(conn
, C_UNCONNECTED
), CS_VERBOSE
);
4214 spin_unlock_irq(&tconn
->req_lock
);
4216 if (oc
== C_DISCONNECTING
)
4217 conn_request_state(tconn
, NS(conn
, C_STANDALONE
), CS_VERBOSE
| CS_HARD
);
4220 static int drbd_disconnected(int vnr
, void *p
, void *data
)
4222 struct drbd_conf
*mdev
= (struct drbd_conf
*)p
;
4223 enum drbd_fencing_p fp
;
4226 /* wait for current activity to cease. */
4227 spin_lock_irq(&mdev
->tconn
->req_lock
);
4228 _drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
4229 _drbd_wait_ee_list_empty(mdev
, &mdev
->sync_ee
);
4230 _drbd_wait_ee_list_empty(mdev
, &mdev
->read_ee
);
4231 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4233 /* We do not have data structures that would allow us to
4234 * get the rs_pending_cnt down to 0 again.
4235 * * On C_SYNC_TARGET we do not have any data structures describing
4236 * the pending RSDataRequest's we have sent.
4237 * * On C_SYNC_SOURCE there is no data structure that tracks
4238 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
4239 * And no, it is not the sum of the reference counts in the
4240 * resync_LRU. The resync_LRU tracks the whole operation including
4241 * the disk-IO, while the rs_pending_cnt only tracks the blocks
4243 drbd_rs_cancel_all(mdev
);
4245 mdev
->rs_failed
= 0;
4246 atomic_set(&mdev
->rs_pending_cnt
, 0);
4247 wake_up(&mdev
->misc_wait
);
4249 del_timer(&mdev
->request_timer
);
4251 del_timer_sync(&mdev
->resync_timer
);
4252 resync_timer_fn((unsigned long)mdev
);
4254 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
4255 * w_make_resync_request etc. which may still be on the worker queue
4256 * to be "canceled" */
4257 drbd_flush_workqueue(mdev
);
4259 drbd_finish_peer_reqs(mdev
);
4261 kfree(mdev
->p_uuid
);
4262 mdev
->p_uuid
= NULL
;
4264 if (!drbd_suspended(mdev
))
4265 tl_clear(mdev
->tconn
);
4270 if (get_ldev(mdev
)) {
4271 fp
= mdev
->ldev
->dc
.fencing
;
4275 /* serialize with bitmap writeout triggered by the state change,
4277 wait_event(mdev
->misc_wait
, !test_bit(BITMAP_IO
, &mdev
->flags
));
4279 /* tcp_close and release of sendpage pages can be deferred. I don't
4280 * want to use SO_LINGER, because apparently it can be deferred for
4281 * more than 20 seconds (longest time I checked).
4283 * Actually we don't care for exactly when the network stack does its
4284 * put_page(), but release our reference on these pages right here.
4286 i
= drbd_free_peer_reqs(mdev
, &mdev
->net_ee
);
4288 dev_info(DEV
, "net_ee not empty, killed %u entries\n", i
);
4289 i
= atomic_read(&mdev
->pp_in_use_by_net
);
4291 dev_info(DEV
, "pp_in_use_by_net = %d, expected 0\n", i
);
4292 i
= atomic_read(&mdev
->pp_in_use
);
4294 dev_info(DEV
, "pp_in_use = %d, expected 0\n", i
);
4296 D_ASSERT(list_empty(&mdev
->read_ee
));
4297 D_ASSERT(list_empty(&mdev
->active_ee
));
4298 D_ASSERT(list_empty(&mdev
->sync_ee
));
4299 D_ASSERT(list_empty(&mdev
->done_ee
));
4301 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
4302 atomic_set(&mdev
->current_epoch
->epoch_size
, 0);
4303 D_ASSERT(list_empty(&mdev
->current_epoch
->list
));
4309 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
4310 * we can agree on is stored in agreed_pro_version.
4312 * feature flags and the reserved array should be enough room for future
4313 * enhancements of the handshake protocol, and possible plugins...
4315 * for now, they are expected to be zero, but ignored.
4317 static int drbd_send_features(struct drbd_tconn
*tconn
)
4319 struct drbd_socket
*sock
;
4320 struct p_connection_features
*p
;
4322 sock
= &tconn
->data
;
4323 p
= conn_prepare_command(tconn
, sock
);
4326 memset(p
, 0, sizeof(*p
));
4327 p
->protocol_min
= cpu_to_be32(PRO_VERSION_MIN
);
4328 p
->protocol_max
= cpu_to_be32(PRO_VERSION_MAX
);
4329 return conn_send_command(tconn
, sock
, P_CONNECTION_FEATURES
, sizeof(*p
), NULL
, 0);
4334 * 1 yes, we have a valid connection
4335 * 0 oops, did not work out, please try again
4336 * -1 peer talks different language,
4337 * no point in trying again, please go standalone.
4339 static int drbd_do_features(struct drbd_tconn
*tconn
)
4341 /* ASSERT current == tconn->receiver ... */
4342 struct p_connection_features
*p
;
4343 const int expect
= sizeof(struct p_connection_features
);
4344 struct packet_info pi
;
4347 err
= drbd_send_features(tconn
);
4351 err
= drbd_recv_header(tconn
, &pi
);
4355 if (pi
.cmd
!= P_CONNECTION_FEATURES
) {
4356 conn_err(tconn
, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
4357 cmdname(pi
.cmd
), pi
.cmd
);
4361 if (pi
.size
!= expect
) {
4362 conn_err(tconn
, "expected ConnectionFeatures length: %u, received: %u\n",
4368 err
= drbd_recv_all_warn(tconn
, p
, expect
);
4372 p
->protocol_min
= be32_to_cpu(p
->protocol_min
);
4373 p
->protocol_max
= be32_to_cpu(p
->protocol_max
);
4374 if (p
->protocol_max
== 0)
4375 p
->protocol_max
= p
->protocol_min
;
4377 if (PRO_VERSION_MAX
< p
->protocol_min
||
4378 PRO_VERSION_MIN
> p
->protocol_max
)
4381 tconn
->agreed_pro_version
= min_t(int, PRO_VERSION_MAX
, p
->protocol_max
);
4383 conn_info(tconn
, "Handshake successful: "
4384 "Agreed network protocol version %d\n", tconn
->agreed_pro_version
);
4389 conn_err(tconn
, "incompatible DRBD dialects: "
4390 "I support %d-%d, peer supports %d-%d\n",
4391 PRO_VERSION_MIN
, PRO_VERSION_MAX
,
4392 p
->protocol_min
, p
->protocol_max
);
4396 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4397 static int drbd_do_auth(struct drbd_tconn
*tconn
)
4399 dev_err(DEV
, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4400 dev_err(DEV
, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4404 #define CHALLENGE_LEN 64
4408 0 - failed, try again (network error),
4409 -1 - auth failed, don't try again.
4412 static int drbd_do_auth(struct drbd_tconn
*tconn
)
4414 struct drbd_socket
*sock
;
4415 char my_challenge
[CHALLENGE_LEN
]; /* 64 Bytes... */
4416 struct scatterlist sg
;
4417 char *response
= NULL
;
4418 char *right_response
= NULL
;
4419 char *peers_ch
= NULL
;
4420 unsigned int key_len
;
4421 char secret
[SHARED_SECRET_MAX
]; /* 64 byte */
4422 unsigned int resp_size
;
4423 struct hash_desc desc
;
4424 struct packet_info pi
;
4425 struct net_conf
*nc
;
4428 /* FIXME: Put the challenge/response into the preallocated socket buffer. */
4431 nc
= rcu_dereference(tconn
->net_conf
);
4432 key_len
= strlen(nc
->shared_secret
);
4433 memcpy(secret
, nc
->shared_secret
, key_len
);
4436 desc
.tfm
= tconn
->cram_hmac_tfm
;
4439 rv
= crypto_hash_setkey(tconn
->cram_hmac_tfm
, (u8
*)secret
, key_len
);
4441 conn_err(tconn
, "crypto_hash_setkey() failed with %d\n", rv
);
4446 get_random_bytes(my_challenge
, CHALLENGE_LEN
);
4448 sock
= &tconn
->data
;
4449 if (!conn_prepare_command(tconn
, sock
)) {
4453 rv
= !conn_send_command(tconn
, sock
, P_AUTH_CHALLENGE
, 0,
4454 my_challenge
, CHALLENGE_LEN
);
4458 err
= drbd_recv_header(tconn
, &pi
);
4464 if (pi
.cmd
!= P_AUTH_CHALLENGE
) {
4465 conn_err(tconn
, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4466 cmdname(pi
.cmd
), pi
.cmd
);
4471 if (pi
.size
> CHALLENGE_LEN
* 2) {
4472 conn_err(tconn
, "expected AuthChallenge payload too big.\n");
4477 peers_ch
= kmalloc(pi
.size
, GFP_NOIO
);
4478 if (peers_ch
== NULL
) {
4479 conn_err(tconn
, "kmalloc of peers_ch failed\n");
4484 err
= drbd_recv_all_warn(tconn
, peers_ch
, pi
.size
);
4490 resp_size
= crypto_hash_digestsize(tconn
->cram_hmac_tfm
);
4491 response
= kmalloc(resp_size
, GFP_NOIO
);
4492 if (response
== NULL
) {
4493 conn_err(tconn
, "kmalloc of response failed\n");
4498 sg_init_table(&sg
, 1);
4499 sg_set_buf(&sg
, peers_ch
, pi
.size
);
4501 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, response
);
4503 conn_err(tconn
, "crypto_hash_digest() failed with %d\n", rv
);
4508 if (!conn_prepare_command(tconn
, sock
)) {
4512 rv
= !conn_send_command(tconn
, sock
, P_AUTH_RESPONSE
, 0,
4513 response
, resp_size
);
4517 err
= drbd_recv_header(tconn
, &pi
);
4523 if (pi
.cmd
!= P_AUTH_RESPONSE
) {
4524 conn_err(tconn
, "expected AuthResponse packet, received: %s (0x%04x)\n",
4525 cmdname(pi
.cmd
), pi
.cmd
);
4530 if (pi
.size
!= resp_size
) {
4531 conn_err(tconn
, "expected AuthResponse payload of wrong size\n");
4536 err
= drbd_recv_all_warn(tconn
, response
, resp_size
);
4542 right_response
= kmalloc(resp_size
, GFP_NOIO
);
4543 if (right_response
== NULL
) {
4544 conn_err(tconn
, "kmalloc of right_response failed\n");
4549 sg_set_buf(&sg
, my_challenge
, CHALLENGE_LEN
);
4551 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, right_response
);
4553 conn_err(tconn
, "crypto_hash_digest() failed with %d\n", rv
);
4558 rv
= !memcmp(response
, right_response
, resp_size
);
4561 conn_info(tconn
, "Peer authenticated using %d bytes HMAC\n",
4569 kfree(right_response
);
4575 int drbdd_init(struct drbd_thread
*thi
)
4577 struct drbd_tconn
*tconn
= thi
->tconn
;
4580 conn_info(tconn
, "receiver (re)started\n");
4583 h
= drbd_connect(tconn
);
4585 drbd_disconnect(tconn
);
4586 schedule_timeout_interruptible(HZ
);
4589 conn_warn(tconn
, "Discarding network configuration.\n");
4590 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
4597 drbd_disconnect(tconn
);
4599 conn_info(tconn
, "receiver terminated\n");
4603 /* ********* acknowledge sender ******** */
4605 static int got_conn_RqSReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4607 struct p_req_state_reply
*p
= pi
->data
;
4608 int retcode
= be32_to_cpu(p
->retcode
);
4610 if (retcode
>= SS_SUCCESS
) {
4611 set_bit(CONN_WD_ST_CHG_OKAY
, &tconn
->flags
);
4613 set_bit(CONN_WD_ST_CHG_FAIL
, &tconn
->flags
);
4614 conn_err(tconn
, "Requested state change failed by peer: %s (%d)\n",
4615 drbd_set_st_err_str(retcode
), retcode
);
4617 wake_up(&tconn
->ping_wait
);
4622 static int got_RqSReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4624 struct drbd_conf
*mdev
;
4625 struct p_req_state_reply
*p
= pi
->data
;
4626 int retcode
= be32_to_cpu(p
->retcode
);
4628 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4632 if (retcode
>= SS_SUCCESS
) {
4633 set_bit(CL_ST_CHG_SUCCESS
, &mdev
->flags
);
4635 set_bit(CL_ST_CHG_FAIL
, &mdev
->flags
);
4636 dev_err(DEV
, "Requested state change failed by peer: %s (%d)\n",
4637 drbd_set_st_err_str(retcode
), retcode
);
4639 wake_up(&mdev
->state_wait
);
4644 static int got_Ping(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4646 return drbd_send_ping_ack(tconn
);
4650 static int got_PingAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4652 /* restore idle timeout */
4653 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= tconn
->net_conf
->ping_int
*HZ
;
4654 if (!test_and_set_bit(GOT_PING_ACK
, &tconn
->flags
))
4655 wake_up(&tconn
->ping_wait
);
4660 static int got_IsInSync(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4662 struct drbd_conf
*mdev
;
4663 struct p_block_ack
*p
= pi
->data
;
4664 sector_t sector
= be64_to_cpu(p
->sector
);
4665 int blksize
= be32_to_cpu(p
->blksize
);
4667 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4671 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
4673 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4675 if (get_ldev(mdev
)) {
4676 drbd_rs_complete_io(mdev
, sector
);
4677 drbd_set_in_sync(mdev
, sector
, blksize
);
4678 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4679 mdev
->rs_same_csum
+= (blksize
>> BM_BLOCK_SHIFT
);
4682 dec_rs_pending(mdev
);
4683 atomic_add(blksize
>> 9, &mdev
->rs_sect_in
);
4689 validate_req_change_req_state(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
4690 struct rb_root
*root
, const char *func
,
4691 enum drbd_req_event what
, bool missing_ok
)
4693 struct drbd_request
*req
;
4694 struct bio_and_error m
;
4696 spin_lock_irq(&mdev
->tconn
->req_lock
);
4697 req
= find_request(mdev
, root
, id
, sector
, missing_ok
, func
);
4698 if (unlikely(!req
)) {
4699 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4702 __req_mod(req
, what
, &m
);
4703 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4706 complete_master_bio(mdev
, &m
);
4710 static int got_BlockAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4712 struct drbd_conf
*mdev
;
4713 struct p_block_ack
*p
= pi
->data
;
4714 sector_t sector
= be64_to_cpu(p
->sector
);
4715 int blksize
= be32_to_cpu(p
->blksize
);
4716 enum drbd_req_event what
;
4718 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4722 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4724 if (p
->block_id
== ID_SYNCER
) {
4725 drbd_set_in_sync(mdev
, sector
, blksize
);
4726 dec_rs_pending(mdev
);
4730 case P_RS_WRITE_ACK
:
4731 what
= WRITE_ACKED_BY_PEER_AND_SIS
;
4734 what
= WRITE_ACKED_BY_PEER
;
4737 what
= RECV_ACKED_BY_PEER
;
4739 case P_DISCARD_WRITE
:
4740 what
= DISCARD_WRITE
;
4743 what
= POSTPONE_WRITE
;
4749 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4750 &mdev
->write_requests
, __func__
,
4754 static int got_NegAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4756 struct drbd_conf
*mdev
;
4757 struct p_block_ack
*p
= pi
->data
;
4758 sector_t sector
= be64_to_cpu(p
->sector
);
4759 int size
= be32_to_cpu(p
->blksize
);
4762 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4766 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4768 if (p
->block_id
== ID_SYNCER
) {
4769 dec_rs_pending(mdev
);
4770 drbd_rs_failed_io(mdev
, sector
, size
);
4774 err
= validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4775 &mdev
->write_requests
, __func__
,
4778 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
4779 The master bio might already be completed, therefore the
4780 request is no longer in the collision hash. */
4781 /* In Protocol B we might already have got a P_RECV_ACK
4782 but then get a P_NEG_ACK afterwards. */
4783 drbd_set_out_of_sync(mdev
, sector
, size
);
4788 static int got_NegDReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4790 struct drbd_conf
*mdev
;
4791 struct p_block_ack
*p
= pi
->data
;
4792 sector_t sector
= be64_to_cpu(p
->sector
);
4794 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4798 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4800 dev_err(DEV
, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4801 (unsigned long long)sector
, be32_to_cpu(p
->blksize
));
4803 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4804 &mdev
->read_requests
, __func__
,
4808 static int got_NegRSDReply(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4810 struct drbd_conf
*mdev
;
4813 struct p_block_ack
*p
= pi
->data
;
4815 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4819 sector
= be64_to_cpu(p
->sector
);
4820 size
= be32_to_cpu(p
->blksize
);
4822 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4824 dec_rs_pending(mdev
);
4826 if (get_ldev_if_state(mdev
, D_FAILED
)) {
4827 drbd_rs_complete_io(mdev
, sector
);
4829 case P_NEG_RS_DREPLY
:
4830 drbd_rs_failed_io(mdev
, sector
, size
);
4842 static int got_BarrierAck(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4844 struct drbd_conf
*mdev
;
4845 struct p_barrier_ack
*p
= pi
->data
;
4847 mdev
= vnr_to_mdev(tconn
, pi
->vnr
);
4851 tl_release(mdev
->tconn
, p
->barrier
, be32_to_cpu(p
->set_size
));
4853 if (mdev
->state
.conn
== C_AHEAD
&&
4854 atomic_read(&mdev
->ap_in_flight
) == 0 &&
4855 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE
, &mdev
->current_epoch
->flags
)) {
4856 mdev
->start_resync_timer
.expires
= jiffies
+ HZ
;
4857 add_timer(&mdev
->start_resync_timer
);
4863 static int got_OVResult(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4865 struct drbd_conf
*mdev
;
4866 struct p_block_ack
*p
= pi
->data
;
4867 struct drbd_work
*w
;
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 if (be64_to_cpu(p
->block_id
) == ID_OUT_OF_SYNC
)
4881 drbd_ov_out_of_sync_found(mdev
, sector
, size
);
4883 ov_out_of_sync_print(mdev
);
4885 if (!get_ldev(mdev
))
4888 drbd_rs_complete_io(mdev
, sector
);
4889 dec_rs_pending(mdev
);
4893 /* let's advance progress step marks only for every other megabyte */
4894 if ((mdev
->ov_left
& 0x200) == 0x200)
4895 drbd_advance_rs_marks(mdev
, mdev
->ov_left
);
4897 if (mdev
->ov_left
== 0) {
4898 w
= kmalloc(sizeof(*w
), GFP_NOIO
);
4900 w
->cb
= w_ov_finished
;
4902 drbd_queue_work_front(&mdev
->tconn
->data
.work
, w
);
4904 dev_err(DEV
, "kmalloc(w) failed.");
4905 ov_out_of_sync_print(mdev
);
4906 drbd_resync_finished(mdev
);
4913 static int got_skip(struct drbd_tconn
*tconn
, struct packet_info
*pi
)
4918 static int tconn_finish_peer_reqs(struct drbd_tconn
*tconn
)
4920 struct drbd_conf
*mdev
;
4921 int i
, not_empty
= 0;
4924 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
4925 flush_signals(current
);
4926 down_read(&drbd_cfg_rwsem
);
4927 idr_for_each_entry(&tconn
->volumes
, mdev
, i
) {
4928 if (drbd_finish_peer_reqs(mdev
)) {
4929 up_read(&drbd_cfg_rwsem
);
4930 return 1; /* error */
4933 up_read(&drbd_cfg_rwsem
);
4934 set_bit(SIGNAL_ASENDER
, &tconn
->flags
);
4936 spin_lock_irq(&tconn
->req_lock
);
4938 idr_for_each_entry(&tconn
->volumes
, mdev
, i
) {
4939 not_empty
= !list_empty(&mdev
->done_ee
);
4944 spin_unlock_irq(&tconn
->req_lock
);
4945 } while (not_empty
);
4950 struct asender_cmd
{
4952 int (*fn
)(struct drbd_tconn
*tconn
, struct packet_info
*);
4955 static struct asender_cmd asender_tbl
[] = {
4956 [P_PING
] = { 0, got_Ping
},
4957 [P_PING_ACK
] = { 0, got_PingAck
},
4958 [P_RECV_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4959 [P_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4960 [P_RS_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4961 [P_DISCARD_WRITE
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4962 [P_NEG_ACK
] = { sizeof(struct p_block_ack
), got_NegAck
},
4963 [P_NEG_DREPLY
] = { sizeof(struct p_block_ack
), got_NegDReply
},
4964 [P_NEG_RS_DREPLY
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
4965 [P_OV_RESULT
] = { sizeof(struct p_block_ack
), got_OVResult
},
4966 [P_BARRIER_ACK
] = { sizeof(struct p_barrier_ack
), got_BarrierAck
},
4967 [P_STATE_CHG_REPLY
] = { sizeof(struct p_req_state_reply
), got_RqSReply
},
4968 [P_RS_IS_IN_SYNC
] = { sizeof(struct p_block_ack
), got_IsInSync
},
4969 [P_DELAY_PROBE
] = { sizeof(struct p_delay_probe93
), got_skip
},
4970 [P_RS_CANCEL
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
4971 [P_CONN_ST_CHG_REPLY
]={ sizeof(struct p_req_state_reply
), got_conn_RqSReply
},
4972 [P_RETRY_WRITE
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4975 int drbd_asender(struct drbd_thread
*thi
)
4977 struct drbd_tconn
*tconn
= thi
->tconn
;
4978 struct asender_cmd
*cmd
= NULL
;
4979 struct packet_info pi
;
4981 void *buf
= tconn
->meta
.rbuf
;
4983 unsigned int header_size
= drbd_header_size(tconn
);
4984 int expect
= header_size
;
4985 bool ping_timeout_active
= false;
4986 struct net_conf
*nc
;
4987 int ping_timeo
, no_cork
, ping_int
;
4989 current
->policy
= SCHED_RR
; /* Make this a realtime task! */
4990 current
->rt_priority
= 2; /* more important than all other tasks */
4992 while (get_t_state(thi
) == RUNNING
) {
4993 drbd_thread_current_set_cpu(thi
);
4996 nc
= rcu_dereference(tconn
->net_conf
);
4997 ping_timeo
= nc
->ping_timeo
;
4998 no_cork
= nc
->no_cork
;
4999 ping_int
= nc
->ping_int
;
5002 if (test_and_clear_bit(SEND_PING
, &tconn
->flags
)) {
5003 if (drbd_send_ping(tconn
)) {
5004 conn_err(tconn
, "drbd_send_ping has failed\n");
5007 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= ping_timeo
* HZ
/ 10;
5008 ping_timeout_active
= true;
5011 /* TODO: conditionally cork; it may hurt latency if we cork without
5014 drbd_tcp_cork(tconn
->meta
.socket
);
5015 if (tconn_finish_peer_reqs(tconn
)) {
5016 conn_err(tconn
, "tconn_finish_peer_reqs() failed\n");
5019 /* but unconditionally uncork unless disabled */
5021 drbd_tcp_uncork(tconn
->meta
.socket
);
5023 /* short circuit, recv_msg would return EINTR anyways. */
5024 if (signal_pending(current
))
5027 rv
= drbd_recv_short(tconn
->meta
.socket
, buf
, expect
-received
, 0);
5028 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5030 flush_signals(current
);
5033 * -EINTR (on meta) we got a signal
5034 * -EAGAIN (on meta) rcvtimeo expired
5035 * -ECONNRESET other side closed the connection
5036 * -ERESTARTSYS (on data) we got a signal
5037 * rv < 0 other than above: unexpected error!
5038 * rv == expected: full header or command
5039 * rv < expected: "woken" by signal during receive
5040 * rv == 0 : "connection shut down by peer"
5042 if (likely(rv
> 0)) {
5045 } else if (rv
== 0) {
5046 conn_err(tconn
, "meta connection shut down by peer.\n");
5048 } else if (rv
== -EAGAIN
) {
5049 /* If the data socket received something meanwhile,
5050 * that is good enough: peer is still alive. */
5051 if (time_after(tconn
->last_received
,
5052 jiffies
- tconn
->meta
.socket
->sk
->sk_rcvtimeo
))
5054 if (ping_timeout_active
) {
5055 conn_err(tconn
, "PingAck did not arrive in time.\n");
5058 set_bit(SEND_PING
, &tconn
->flags
);
5060 } else if (rv
== -EINTR
) {
5063 conn_err(tconn
, "sock_recvmsg returned %d\n", rv
);
5067 if (received
== expect
&& cmd
== NULL
) {
5068 if (decode_header(tconn
, tconn
->meta
.rbuf
, &pi
))
5070 cmd
= &asender_tbl
[pi
.cmd
];
5071 if (pi
.cmd
>= ARRAY_SIZE(asender_tbl
) || !cmd
->fn
) {
5072 conn_err(tconn
, "unknown command %d on meta (l: %d)\n",
5076 expect
= header_size
+ cmd
->pkt_size
;
5077 if (pi
.size
!= expect
- header_size
) {
5078 conn_err(tconn
, "Wrong packet size on meta (c: %d, l: %d)\n",
5083 if (received
== expect
) {
5086 err
= cmd
->fn(tconn
, &pi
);
5088 conn_err(tconn
, "%pf failed\n", cmd
->fn
);
5092 tconn
->last_received
= jiffies
;
5094 if (cmd
== &asender_tbl
[P_PING_ACK
]) {
5095 /* restore idle timeout */
5096 tconn
->meta
.socket
->sk
->sk_rcvtimeo
= ping_int
* HZ
;
5097 ping_timeout_active
= false;
5100 buf
= tconn
->meta
.rbuf
;
5102 expect
= header_size
;
5109 conn_request_state(tconn
, NS(conn
, C_NETWORK_FAILURE
), CS_HARD
);
5113 conn_request_state(tconn
, NS(conn
, C_DISCONNECTING
), CS_HARD
);
5115 clear_bit(SIGNAL_ASENDER
, &tconn
->flags
);
5117 conn_info(tconn
, "asender terminated\n");