4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
28 #include <asm/uaccess.h>
31 #include <linux/drbd.h>
33 #include <linux/file.h>
36 #include <linux/memcontrol.h>
37 #include <linux/mm_inline.h>
38 #include <linux/slab.h>
39 #include <linux/pkt_sched.h>
40 #define __KERNEL_SYSCALLS__
41 #include <linux/unistd.h>
42 #include <linux/vmalloc.h>
43 #include <linux/random.h>
44 #include <linux/string.h>
45 #include <linux/scatterlist.h>
57 static int drbd_do_handshake(struct drbd_conf
*mdev
);
58 static int drbd_do_auth(struct drbd_conf
*mdev
);
60 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*, struct drbd_epoch
*, enum epoch_event
);
61 static int e_end_block(struct drbd_conf
*, struct drbd_work
*, int);
64 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
67 * some helper functions to deal with single linked page lists,
68 * page->private being our "next" pointer.
71 /* If at least n pages are linked at head, get n pages off.
72 * Otherwise, don't modify head, and return NULL.
73 * Locking is the responsibility of the caller.
75 static struct page
*page_chain_del(struct page
**head
, int n
)
89 tmp
= page_chain_next(page
);
91 break; /* found sufficient pages */
93 /* insufficient pages, don't use any of them. */
98 /* add end of list marker for the returned list */
99 set_page_private(page
, 0);
100 /* actual return value, and adjustment of head */
106 /* may be used outside of locks to find the tail of a (usually short)
107 * "private" page chain, before adding it back to a global chain head
108 * with page_chain_add() under a spinlock. */
109 static struct page
*page_chain_tail(struct page
*page
, int *len
)
113 while ((tmp
= page_chain_next(page
)))
120 static int page_chain_free(struct page
*page
)
124 page_chain_for_each_safe(page
, tmp
) {
131 static void page_chain_add(struct page
**head
,
132 struct page
*chain_first
, struct page
*chain_last
)
136 tmp
= page_chain_tail(chain_first
, NULL
);
137 BUG_ON(tmp
!= chain_last
);
140 /* add chain to head */
141 set_page_private(chain_last
, (unsigned long)*head
);
145 static struct page
*drbd_pp_first_pages_or_try_alloc(struct drbd_conf
*mdev
, int number
)
147 struct page
*page
= NULL
;
148 struct page
*tmp
= NULL
;
151 /* Yes, testing drbd_pp_vacant outside the lock is racy.
152 * So what. It saves a spin_lock. */
153 if (drbd_pp_vacant
>= number
) {
154 spin_lock(&drbd_pp_lock
);
155 page
= page_chain_del(&drbd_pp_pool
, number
);
157 drbd_pp_vacant
-= number
;
158 spin_unlock(&drbd_pp_lock
);
163 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
164 * "criss-cross" setup, that might cause write-out on some other DRBD,
165 * which in turn might block on the other node at this very place. */
166 for (i
= 0; i
< number
; i
++) {
167 tmp
= alloc_page(GFP_TRY
);
170 set_page_private(tmp
, (unsigned long)page
);
177 /* Not enough pages immediately available this time.
178 * No need to jump around here, drbd_pp_alloc will retry this
179 * function "soon". */
181 tmp
= page_chain_tail(page
, NULL
);
182 spin_lock(&drbd_pp_lock
);
183 page_chain_add(&drbd_pp_pool
, page
, tmp
);
185 spin_unlock(&drbd_pp_lock
);
190 static void reclaim_net_ee(struct drbd_conf
*mdev
, struct list_head
*to_be_freed
)
192 struct drbd_peer_request
*peer_req
;
193 struct list_head
*le
, *tle
;
195 /* The EEs are always appended to the end of the list. Since
196 they are sent in order over the wire, they have to finish
197 in order. As soon as we see the first not finished we can
198 stop to examine the list... */
200 list_for_each_safe(le
, tle
, &mdev
->net_ee
) {
201 peer_req
= list_entry(le
, struct drbd_peer_request
, w
.list
);
202 if (drbd_ee_has_active_page(peer_req
))
204 list_move(le
, to_be_freed
);
208 static void drbd_kick_lo_and_reclaim_net(struct drbd_conf
*mdev
)
210 LIST_HEAD(reclaimed
);
211 struct drbd_peer_request
*peer_req
, *t
;
213 spin_lock_irq(&mdev
->tconn
->req_lock
);
214 reclaim_net_ee(mdev
, &reclaimed
);
215 spin_unlock_irq(&mdev
->tconn
->req_lock
);
217 list_for_each_entry_safe(peer_req
, t
, &reclaimed
, w
.list
)
218 drbd_free_net_ee(mdev
, peer_req
);
222 * drbd_pp_alloc() - Returns @number pages, retries forever (or until signalled)
223 * @mdev: DRBD device.
224 * @number: number of pages requested
225 * @retry: whether to retry, if not enough pages are available right now
227 * Tries to allocate number pages, first from our own page pool, then from
228 * the kernel, unless this allocation would exceed the max_buffers setting.
229 * Possibly retry until DRBD frees sufficient pages somewhere else.
231 * Returns a page chain linked via page->private.
233 static struct page
*drbd_pp_alloc(struct drbd_conf
*mdev
, unsigned number
, bool retry
)
235 struct page
*page
= NULL
;
238 /* Yes, we may run up to @number over max_buffers. If we
239 * follow it strictly, the admin will get it wrong anyways. */
240 if (atomic_read(&mdev
->pp_in_use
) < mdev
->tconn
->net_conf
->max_buffers
)
241 page
= drbd_pp_first_pages_or_try_alloc(mdev
, number
);
243 while (page
== NULL
) {
244 prepare_to_wait(&drbd_pp_wait
, &wait
, TASK_INTERRUPTIBLE
);
246 drbd_kick_lo_and_reclaim_net(mdev
);
248 if (atomic_read(&mdev
->pp_in_use
) < mdev
->tconn
->net_conf
->max_buffers
) {
249 page
= drbd_pp_first_pages_or_try_alloc(mdev
, number
);
257 if (signal_pending(current
)) {
258 dev_warn(DEV
, "drbd_pp_alloc interrupted!\n");
264 finish_wait(&drbd_pp_wait
, &wait
);
267 atomic_add(number
, &mdev
->pp_in_use
);
271 /* Must not be used from irq, as that may deadlock: see drbd_pp_alloc.
272 * Is also used from inside an other spin_lock_irq(&mdev->tconn->req_lock);
273 * Either links the page chain back to the global pool,
274 * or returns all pages to the system. */
275 static void drbd_pp_free(struct drbd_conf
*mdev
, struct page
*page
, int is_net
)
277 atomic_t
*a
= is_net
? &mdev
->pp_in_use_by_net
: &mdev
->pp_in_use
;
280 if (drbd_pp_vacant
> (DRBD_MAX_BIO_SIZE
/PAGE_SIZE
)*minor_count
)
281 i
= page_chain_free(page
);
284 tmp
= page_chain_tail(page
, &i
);
285 spin_lock(&drbd_pp_lock
);
286 page_chain_add(&drbd_pp_pool
, page
, tmp
);
288 spin_unlock(&drbd_pp_lock
);
290 i
= atomic_sub_return(i
, a
);
292 dev_warn(DEV
, "ASSERTION FAILED: %s: %d < 0\n",
293 is_net
? "pp_in_use_by_net" : "pp_in_use", i
);
294 wake_up(&drbd_pp_wait
);
298 You need to hold the req_lock:
299 _drbd_wait_ee_list_empty()
301 You must not have the req_lock:
307 drbd_process_done_ee()
309 drbd_wait_ee_list_empty()
312 struct drbd_peer_request
*
313 drbd_alloc_ee(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
314 unsigned int data_size
, gfp_t gfp_mask
) __must_hold(local
)
316 struct drbd_peer_request
*peer_req
;
318 unsigned nr_pages
= (data_size
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
320 if (drbd_insert_fault(mdev
, DRBD_FAULT_AL_EE
))
323 peer_req
= mempool_alloc(drbd_ee_mempool
, gfp_mask
& ~__GFP_HIGHMEM
);
325 if (!(gfp_mask
& __GFP_NOWARN
))
326 dev_err(DEV
, "alloc_ee: Allocation of an EE failed\n");
330 page
= drbd_pp_alloc(mdev
, nr_pages
, (gfp_mask
& __GFP_WAIT
));
334 drbd_clear_interval(&peer_req
->i
);
335 peer_req
->i
.size
= data_size
;
336 peer_req
->i
.sector
= sector
;
337 peer_req
->i
.local
= false;
338 peer_req
->i
.waiting
= false;
340 peer_req
->epoch
= NULL
;
341 peer_req
->mdev
= mdev
;
342 peer_req
->pages
= page
;
343 atomic_set(&peer_req
->pending_bios
, 0);
346 * The block_id is opaque to the receiver. It is not endianness
347 * converted, and sent back to the sender unchanged.
349 peer_req
->block_id
= id
;
354 mempool_free(peer_req
, drbd_ee_mempool
);
358 void drbd_free_some_ee(struct drbd_conf
*mdev
, struct drbd_peer_request
*peer_req
,
361 if (peer_req
->flags
& EE_HAS_DIGEST
)
362 kfree(peer_req
->digest
);
363 drbd_pp_free(mdev
, peer_req
->pages
, is_net
);
364 D_ASSERT(atomic_read(&peer_req
->pending_bios
) == 0);
365 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
366 mempool_free(peer_req
, drbd_ee_mempool
);
369 int drbd_release_ee(struct drbd_conf
*mdev
, struct list_head
*list
)
371 LIST_HEAD(work_list
);
372 struct drbd_peer_request
*peer_req
, *t
;
374 int is_net
= list
== &mdev
->net_ee
;
376 spin_lock_irq(&mdev
->tconn
->req_lock
);
377 list_splice_init(list
, &work_list
);
378 spin_unlock_irq(&mdev
->tconn
->req_lock
);
380 list_for_each_entry_safe(peer_req
, t
, &work_list
, w
.list
) {
381 drbd_free_some_ee(mdev
, peer_req
, is_net
);
389 * This function is called from _asender only_
390 * but see also comments in _req_mod(,BARRIER_ACKED)
391 * and receive_Barrier.
393 * Move entries from net_ee to done_ee, if ready.
394 * Grab done_ee, call all callbacks, free the entries.
395 * The callbacks typically send out ACKs.
397 static int drbd_process_done_ee(struct drbd_conf
*mdev
)
399 LIST_HEAD(work_list
);
400 LIST_HEAD(reclaimed
);
401 struct drbd_peer_request
*peer_req
, *t
;
402 int ok
= (mdev
->state
.conn
>= C_WF_REPORT_PARAMS
);
404 spin_lock_irq(&mdev
->tconn
->req_lock
);
405 reclaim_net_ee(mdev
, &reclaimed
);
406 list_splice_init(&mdev
->done_ee
, &work_list
);
407 spin_unlock_irq(&mdev
->tconn
->req_lock
);
409 list_for_each_entry_safe(peer_req
, t
, &reclaimed
, w
.list
)
410 drbd_free_net_ee(mdev
, peer_req
);
412 /* possible callbacks here:
413 * e_end_block, and e_end_resync_block, e_send_discard_ack.
414 * all ignore the last argument.
416 list_for_each_entry_safe(peer_req
, t
, &work_list
, w
.list
) {
417 /* list_del not necessary, next/prev members not touched */
418 ok
= peer_req
->w
.cb(mdev
, &peer_req
->w
, !ok
) && ok
;
419 drbd_free_ee(mdev
, peer_req
);
421 wake_up(&mdev
->ee_wait
);
426 void _drbd_wait_ee_list_empty(struct drbd_conf
*mdev
, struct list_head
*head
)
430 /* avoids spin_lock/unlock
431 * and calling prepare_to_wait in the fast path */
432 while (!list_empty(head
)) {
433 prepare_to_wait(&mdev
->ee_wait
, &wait
, TASK_UNINTERRUPTIBLE
);
434 spin_unlock_irq(&mdev
->tconn
->req_lock
);
436 finish_wait(&mdev
->ee_wait
, &wait
);
437 spin_lock_irq(&mdev
->tconn
->req_lock
);
441 void drbd_wait_ee_list_empty(struct drbd_conf
*mdev
, struct list_head
*head
)
443 spin_lock_irq(&mdev
->tconn
->req_lock
);
444 _drbd_wait_ee_list_empty(mdev
, head
);
445 spin_unlock_irq(&mdev
->tconn
->req_lock
);
448 /* see also kernel_accept; which is only present since 2.6.18.
449 * also we want to log which part of it failed, exactly */
450 static int drbd_accept(struct drbd_conf
*mdev
, const char **what
,
451 struct socket
*sock
, struct socket
**newsock
)
453 struct sock
*sk
= sock
->sk
;
457 err
= sock
->ops
->listen(sock
, 5);
461 *what
= "sock_create_lite";
462 err
= sock_create_lite(sk
->sk_family
, sk
->sk_type
, sk
->sk_protocol
,
468 err
= sock
->ops
->accept(sock
, *newsock
, 0);
470 sock_release(*newsock
);
474 (*newsock
)->ops
= sock
->ops
;
480 static int drbd_recv_short(struct drbd_conf
*mdev
, struct socket
*sock
,
481 void *buf
, size_t size
, int flags
)
488 struct msghdr msg
= {
490 .msg_iov
= (struct iovec
*)&iov
,
491 .msg_flags
= (flags
? flags
: MSG_WAITALL
| MSG_NOSIGNAL
)
497 rv
= sock_recvmsg(sock
, &msg
, size
, msg
.msg_flags
);
503 static int drbd_recv(struct drbd_conf
*mdev
, void *buf
, size_t size
)
510 struct msghdr msg
= {
512 .msg_iov
= (struct iovec
*)&iov
,
513 .msg_flags
= MSG_WAITALL
| MSG_NOSIGNAL
521 rv
= sock_recvmsg(mdev
->tconn
->data
.socket
, &msg
, size
, msg
.msg_flags
);
526 * ECONNRESET other side closed the connection
527 * ERESTARTSYS (on sock) we got a signal
531 if (rv
== -ECONNRESET
)
532 dev_info(DEV
, "sock was reset by peer\n");
533 else if (rv
!= -ERESTARTSYS
)
534 dev_err(DEV
, "sock_recvmsg returned %d\n", rv
);
536 } else if (rv
== 0) {
537 dev_info(DEV
, "sock was shut down by peer\n");
540 /* signal came in, or peer/link went down,
541 * after we read a partial message
543 /* D_ASSERT(signal_pending(current)); */
551 drbd_force_state(mdev
, NS(conn
, C_BROKEN_PIPE
));
557 * On individual connections, the socket buffer size must be set prior to the
558 * listen(2) or connect(2) calls in order to have it take effect.
559 * This is our wrapper to do so.
561 static void drbd_setbufsize(struct socket
*sock
, unsigned int snd
,
564 /* open coded SO_SNDBUF, SO_RCVBUF */
566 sock
->sk
->sk_sndbuf
= snd
;
567 sock
->sk
->sk_userlocks
|= SOCK_SNDBUF_LOCK
;
570 sock
->sk
->sk_rcvbuf
= rcv
;
571 sock
->sk
->sk_userlocks
|= SOCK_RCVBUF_LOCK
;
575 static struct socket
*drbd_try_connect(struct drbd_conf
*mdev
)
579 struct sockaddr_in6 src_in6
;
581 int disconnect_on_error
= 1;
583 if (!get_net_conf(mdev
->tconn
))
586 what
= "sock_create_kern";
587 err
= sock_create_kern(((struct sockaddr
*)mdev
->tconn
->net_conf
->my_addr
)->sa_family
,
588 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
594 sock
->sk
->sk_rcvtimeo
=
595 sock
->sk
->sk_sndtimeo
= mdev
->tconn
->net_conf
->try_connect_int
*HZ
;
596 drbd_setbufsize(sock
, mdev
->tconn
->net_conf
->sndbuf_size
,
597 mdev
->tconn
->net_conf
->rcvbuf_size
);
599 /* explicitly bind to the configured IP as source IP
600 * for the outgoing connections.
601 * This is needed for multihomed hosts and to be
602 * able to use lo: interfaces for drbd.
603 * Make sure to use 0 as port number, so linux selects
604 * a free one dynamically.
606 memcpy(&src_in6
, mdev
->tconn
->net_conf
->my_addr
,
607 min_t(int, mdev
->tconn
->net_conf
->my_addr_len
, sizeof(src_in6
)));
608 if (((struct sockaddr
*)mdev
->tconn
->net_conf
->my_addr
)->sa_family
== AF_INET6
)
609 src_in6
.sin6_port
= 0;
611 ((struct sockaddr_in
*)&src_in6
)->sin_port
= 0; /* AF_INET & AF_SCI */
613 what
= "bind before connect";
614 err
= sock
->ops
->bind(sock
,
615 (struct sockaddr
*) &src_in6
,
616 mdev
->tconn
->net_conf
->my_addr_len
);
620 /* connect may fail, peer not yet available.
621 * stay C_WF_CONNECTION, don't go Disconnecting! */
622 disconnect_on_error
= 0;
624 err
= sock
->ops
->connect(sock
,
625 (struct sockaddr
*)mdev
->tconn
->net_conf
->peer_addr
,
626 mdev
->tconn
->net_conf
->peer_addr_len
, 0);
635 /* timeout, busy, signal pending */
636 case ETIMEDOUT
: case EAGAIN
: case EINPROGRESS
:
637 case EINTR
: case ERESTARTSYS
:
638 /* peer not (yet) available, network problem */
639 case ECONNREFUSED
: case ENETUNREACH
:
640 case EHOSTDOWN
: case EHOSTUNREACH
:
641 disconnect_on_error
= 0;
644 dev_err(DEV
, "%s failed, err = %d\n", what
, err
);
646 if (disconnect_on_error
)
647 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
649 put_net_conf(mdev
->tconn
);
653 static struct socket
*drbd_wait_for_connect(struct drbd_conf
*mdev
)
656 struct socket
*s_estab
= NULL
, *s_listen
;
659 if (!get_net_conf(mdev
->tconn
))
662 what
= "sock_create_kern";
663 err
= sock_create_kern(((struct sockaddr
*)mdev
->tconn
->net_conf
->my_addr
)->sa_family
,
664 SOCK_STREAM
, IPPROTO_TCP
, &s_listen
);
670 timeo
= mdev
->tconn
->net_conf
->try_connect_int
* HZ
;
671 timeo
+= (random32() & 1) ? timeo
/ 7 : -timeo
/ 7; /* 28.5% random jitter */
673 s_listen
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
674 s_listen
->sk
->sk_rcvtimeo
= timeo
;
675 s_listen
->sk
->sk_sndtimeo
= timeo
;
676 drbd_setbufsize(s_listen
, mdev
->tconn
->net_conf
->sndbuf_size
,
677 mdev
->tconn
->net_conf
->rcvbuf_size
);
679 what
= "bind before listen";
680 err
= s_listen
->ops
->bind(s_listen
,
681 (struct sockaddr
*) mdev
->tconn
->net_conf
->my_addr
,
682 mdev
->tconn
->net_conf
->my_addr_len
);
686 err
= drbd_accept(mdev
, &what
, s_listen
, &s_estab
);
690 sock_release(s_listen
);
692 if (err
!= -EAGAIN
&& err
!= -EINTR
&& err
!= -ERESTARTSYS
) {
693 dev_err(DEV
, "%s failed, err = %d\n", what
, err
);
694 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
697 put_net_conf(mdev
->tconn
);
702 static int drbd_send_fp(struct drbd_conf
*mdev
, struct socket
*sock
,
703 enum drbd_packet cmd
)
705 struct p_header
*h
= &mdev
->tconn
->data
.sbuf
.header
;
707 return _drbd_send_cmd(mdev
, sock
, cmd
, h
, sizeof(*h
), 0);
710 static enum drbd_packet
drbd_recv_fp(struct drbd_conf
*mdev
,
713 struct p_header80
*h
= &mdev
->tconn
->data
.rbuf
.header
.h80
;
716 rr
= drbd_recv_short(mdev
, sock
, h
, sizeof(*h
), 0);
718 if (rr
== sizeof(*h
) && h
->magic
== cpu_to_be32(DRBD_MAGIC
))
719 return be16_to_cpu(h
->command
);
725 * drbd_socket_okay() - Free the socket if its connection is not okay
726 * @mdev: DRBD device.
727 * @sock: pointer to the pointer to the socket.
729 static int drbd_socket_okay(struct drbd_conf
*mdev
, struct socket
**sock
)
737 rr
= drbd_recv_short(mdev
, *sock
, tb
, 4, MSG_DONTWAIT
| MSG_PEEK
);
739 if (rr
> 0 || rr
== -EAGAIN
) {
750 * 1 yes, we have a valid connection
751 * 0 oops, did not work out, please try again
752 * -1 peer talks different language,
753 * no point in trying again, please go standalone.
754 * -2 We do not have a network config...
756 static int drbd_connect(struct drbd_conf
*mdev
)
758 struct socket
*s
, *sock
, *msock
;
761 D_ASSERT(!mdev
->tconn
->data
.socket
);
763 if (drbd_request_state(mdev
, NS(conn
, C_WF_CONNECTION
)) < SS_SUCCESS
)
766 clear_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
767 mdev
->tconn
->agreed_pro_version
= 99;
768 /* agreed_pro_version must be smaller than 100 so we send the old
769 header (h80) in the first packet and in the handshake packet. */
776 /* 3 tries, this should take less than a second! */
777 s
= drbd_try_connect(mdev
);
780 /* give the other side time to call bind() & listen() */
781 schedule_timeout_interruptible(HZ
/ 10);
786 drbd_send_fp(mdev
, s
, P_HAND_SHAKE_S
);
790 drbd_send_fp(mdev
, s
, P_HAND_SHAKE_M
);
794 dev_err(DEV
, "Logic error in drbd_connect()\n");
795 goto out_release_sockets
;
800 schedule_timeout_interruptible(mdev
->tconn
->net_conf
->ping_timeo
*HZ
/10);
801 ok
= drbd_socket_okay(mdev
, &sock
);
802 ok
= drbd_socket_okay(mdev
, &msock
) && ok
;
808 s
= drbd_wait_for_connect(mdev
);
810 try = drbd_recv_fp(mdev
, s
);
811 drbd_socket_okay(mdev
, &sock
);
812 drbd_socket_okay(mdev
, &msock
);
816 dev_warn(DEV
, "initial packet S crossed\n");
823 dev_warn(DEV
, "initial packet M crossed\n");
827 set_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
830 dev_warn(DEV
, "Error receiving initial packet\n");
837 if (mdev
->state
.conn
<= C_DISCONNECTING
)
838 goto out_release_sockets
;
839 if (signal_pending(current
)) {
840 flush_signals(current
);
842 if (get_t_state(&mdev
->tconn
->receiver
) == EXITING
)
843 goto out_release_sockets
;
847 ok
= drbd_socket_okay(mdev
, &sock
);
848 ok
= drbd_socket_okay(mdev
, &msock
) && ok
;
854 msock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
855 sock
->sk
->sk_reuse
= 1; /* SO_REUSEADDR */
857 sock
->sk
->sk_allocation
= GFP_NOIO
;
858 msock
->sk
->sk_allocation
= GFP_NOIO
;
860 sock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE_BULK
;
861 msock
->sk
->sk_priority
= TC_PRIO_INTERACTIVE
;
864 * sock->sk->sk_sndtimeo = mdev->tconn->net_conf->timeout*HZ/10;
865 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
866 * first set it to the P_HAND_SHAKE timeout,
867 * which we set to 4x the configured ping_timeout. */
868 sock
->sk
->sk_sndtimeo
=
869 sock
->sk
->sk_rcvtimeo
= mdev
->tconn
->net_conf
->ping_timeo
*4*HZ
/10;
871 msock
->sk
->sk_sndtimeo
= mdev
->tconn
->net_conf
->timeout
*HZ
/10;
872 msock
->sk
->sk_rcvtimeo
= mdev
->tconn
->net_conf
->ping_int
*HZ
;
874 /* we don't want delays.
875 * we use TCP_CORK where appropriate, though */
876 drbd_tcp_nodelay(sock
);
877 drbd_tcp_nodelay(msock
);
879 mdev
->tconn
->data
.socket
= sock
;
880 mdev
->tconn
->meta
.socket
= msock
;
881 mdev
->tconn
->last_received
= jiffies
;
883 D_ASSERT(mdev
->tconn
->asender
.task
== NULL
);
885 h
= drbd_do_handshake(mdev
);
889 if (mdev
->tconn
->cram_hmac_tfm
) {
890 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
891 switch (drbd_do_auth(mdev
)) {
893 dev_err(DEV
, "Authentication of peer failed\n");
896 dev_err(DEV
, "Authentication of peer failed, trying again.\n");
901 if (drbd_request_state(mdev
, NS(conn
, C_WF_REPORT_PARAMS
)) < SS_SUCCESS
)
904 sock
->sk
->sk_sndtimeo
= mdev
->tconn
->net_conf
->timeout
*HZ
/10;
905 sock
->sk
->sk_rcvtimeo
= MAX_SCHEDULE_TIMEOUT
;
907 atomic_set(&mdev
->packet_seq
, 0);
910 drbd_thread_start(&mdev
->tconn
->asender
);
912 if (drbd_send_protocol(mdev
) == -1)
914 drbd_send_sync_param(mdev
, &mdev
->sync_conf
);
915 drbd_send_sizes(mdev
, 0, 0);
916 drbd_send_uuids(mdev
);
917 drbd_send_state(mdev
);
918 clear_bit(USE_DEGR_WFC_T
, &mdev
->flags
);
919 clear_bit(RESIZE_PENDING
, &mdev
->flags
);
920 mod_timer(&mdev
->request_timer
, jiffies
+ HZ
); /* just start it here. */
932 static bool decode_header(struct drbd_conf
*mdev
, struct p_header
*h
,
933 enum drbd_packet
*cmd
, unsigned int *packet_size
)
935 if (h
->h80
.magic
== cpu_to_be32(DRBD_MAGIC
)) {
936 *cmd
= be16_to_cpu(h
->h80
.command
);
937 *packet_size
= be16_to_cpu(h
->h80
.length
);
938 } else if (h
->h95
.magic
== cpu_to_be16(DRBD_MAGIC_BIG
)) {
939 *cmd
= be16_to_cpu(h
->h95
.command
);
940 *packet_size
= be32_to_cpu(h
->h95
.length
) & 0x00ffffff;
942 dev_err(DEV
, "magic?? on data m: 0x%08x c: %d l: %d\n",
943 be32_to_cpu(h
->h80
.magic
),
944 be16_to_cpu(h
->h80
.command
),
945 be16_to_cpu(h
->h80
.length
));
951 static int drbd_recv_header(struct drbd_conf
*mdev
, enum drbd_packet
*cmd
,
952 unsigned int *packet_size
)
954 struct p_header
*h
= &mdev
->tconn
->data
.rbuf
.header
;
957 r
= drbd_recv(mdev
, h
, sizeof(*h
));
958 if (unlikely(r
!= sizeof(*h
))) {
959 if (!signal_pending(current
))
960 dev_warn(DEV
, "short read expecting header on sock: r=%d\n", r
);
964 r
= decode_header(mdev
, h
, cmd
, packet_size
);
965 mdev
->tconn
->last_received
= jiffies
;
970 static void drbd_flush(struct drbd_conf
*mdev
)
974 if (mdev
->write_ordering
>= WO_bdev_flush
&& get_ldev(mdev
)) {
975 rv
= blkdev_issue_flush(mdev
->ldev
->backing_bdev
, GFP_KERNEL
,
978 dev_err(DEV
, "local disk flush failed with status %d\n", rv
);
979 /* would rather check on EOPNOTSUPP, but that is not reliable.
980 * don't try again for ANY return value != 0
981 * if (rv == -EOPNOTSUPP) */
982 drbd_bump_write_ordering(mdev
, WO_drain_io
);
989 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
990 * @mdev: DRBD device.
991 * @epoch: Epoch object.
994 static enum finish_epoch
drbd_may_finish_epoch(struct drbd_conf
*mdev
,
995 struct drbd_epoch
*epoch
,
999 struct drbd_epoch
*next_epoch
;
1000 enum finish_epoch rv
= FE_STILL_LIVE
;
1002 spin_lock(&mdev
->epoch_lock
);
1006 epoch_size
= atomic_read(&epoch
->epoch_size
);
1008 switch (ev
& ~EV_CLEANUP
) {
1010 atomic_dec(&epoch
->active
);
1012 case EV_GOT_BARRIER_NR
:
1013 set_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
);
1015 case EV_BECAME_LAST
:
1020 if (epoch_size
!= 0 &&
1021 atomic_read(&epoch
->active
) == 0 &&
1022 test_bit(DE_HAVE_BARRIER_NUMBER
, &epoch
->flags
)) {
1023 if (!(ev
& EV_CLEANUP
)) {
1024 spin_unlock(&mdev
->epoch_lock
);
1025 drbd_send_b_ack(mdev
, epoch
->barrier_nr
, epoch_size
);
1026 spin_lock(&mdev
->epoch_lock
);
1030 if (mdev
->current_epoch
!= epoch
) {
1031 next_epoch
= list_entry(epoch
->list
.next
, struct drbd_epoch
, list
);
1032 list_del(&epoch
->list
);
1033 ev
= EV_BECAME_LAST
| (ev
& EV_CLEANUP
);
1037 if (rv
== FE_STILL_LIVE
)
1041 atomic_set(&epoch
->epoch_size
, 0);
1042 /* atomic_set(&epoch->active, 0); is already zero */
1043 if (rv
== FE_STILL_LIVE
)
1045 wake_up(&mdev
->ee_wait
);
1055 spin_unlock(&mdev
->epoch_lock
);
1061 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1062 * @mdev: DRBD device.
1063 * @wo: Write ordering method to try.
1065 void drbd_bump_write_ordering(struct drbd_conf
*mdev
, enum write_ordering_e wo
) __must_hold(local
)
1067 enum write_ordering_e pwo
;
1068 static char *write_ordering_str
[] = {
1070 [WO_drain_io
] = "drain",
1071 [WO_bdev_flush
] = "flush",
1074 pwo
= mdev
->write_ordering
;
1076 if (wo
== WO_bdev_flush
&& mdev
->ldev
->dc
.no_disk_flush
)
1078 if (wo
== WO_drain_io
&& mdev
->ldev
->dc
.no_disk_drain
)
1080 mdev
->write_ordering
= wo
;
1081 if (pwo
!= mdev
->write_ordering
|| wo
== WO_bdev_flush
)
1082 dev_info(DEV
, "Method to ensure write ordering: %s\n", write_ordering_str
[mdev
->write_ordering
]);
1087 * @mdev: DRBD device.
1088 * @peer_req: peer request
1089 * @rw: flag field, see bio->bi_rw
1091 * May spread the pages to multiple bios,
1092 * depending on bio_add_page restrictions.
1094 * Returns 0 if all bios have been submitted,
1095 * -ENOMEM if we could not allocate enough bios,
1096 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1097 * single page to an empty bio (which should never happen and likely indicates
1098 * that the lower level IO stack is in some way broken). This has been observed
1099 * on certain Xen deployments.
1101 /* TODO allocate from our own bio_set. */
1102 int drbd_submit_ee(struct drbd_conf
*mdev
, struct drbd_peer_request
*peer_req
,
1103 const unsigned rw
, const int fault_type
)
1105 struct bio
*bios
= NULL
;
1107 struct page
*page
= peer_req
->pages
;
1108 sector_t sector
= peer_req
->i
.sector
;
1109 unsigned ds
= peer_req
->i
.size
;
1110 unsigned n_bios
= 0;
1111 unsigned nr_pages
= (ds
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
1114 /* In most cases, we will only need one bio. But in case the lower
1115 * level restrictions happen to be different at this offset on this
1116 * side than those of the sending peer, we may need to submit the
1117 * request in more than one bio. */
1119 bio
= bio_alloc(GFP_NOIO
, nr_pages
);
1121 dev_err(DEV
, "submit_ee: Allocation of a bio failed\n");
1124 /* > peer_req->i.sector, unless this is the first bio */
1125 bio
->bi_sector
= sector
;
1126 bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1128 bio
->bi_private
= peer_req
;
1129 bio
->bi_end_io
= drbd_endio_sec
;
1131 bio
->bi_next
= bios
;
1135 page_chain_for_each(page
) {
1136 unsigned len
= min_t(unsigned, ds
, PAGE_SIZE
);
1137 if (!bio_add_page(bio
, page
, len
, 0)) {
1138 /* A single page must always be possible!
1139 * But in case it fails anyways,
1140 * we deal with it, and complain (below). */
1141 if (bio
->bi_vcnt
== 0) {
1143 "bio_add_page failed for len=%u, "
1144 "bi_vcnt=0 (bi_sector=%llu)\n",
1145 len
, (unsigned long long)bio
->bi_sector
);
1155 D_ASSERT(page
== NULL
);
1158 atomic_set(&peer_req
->pending_bios
, n_bios
);
1161 bios
= bios
->bi_next
;
1162 bio
->bi_next
= NULL
;
1164 drbd_generic_make_request(mdev
, fault_type
, bio
);
1171 bios
= bios
->bi_next
;
1177 static void drbd_remove_epoch_entry_interval(struct drbd_conf
*mdev
,
1178 struct drbd_peer_request
*peer_req
)
1180 struct drbd_interval
*i
= &peer_req
->i
;
1182 drbd_remove_interval(&mdev
->write_requests
, i
);
1183 drbd_clear_interval(i
);
1185 /* Wake up any processes waiting for this peer request to complete. */
1187 wake_up(&mdev
->misc_wait
);
1190 static int receive_Barrier(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
1191 unsigned int data_size
)
1194 struct p_barrier
*p
= &mdev
->tconn
->data
.rbuf
.barrier
;
1195 struct drbd_epoch
*epoch
;
1199 mdev
->current_epoch
->barrier_nr
= p
->barrier
;
1200 rv
= drbd_may_finish_epoch(mdev
, mdev
->current_epoch
, EV_GOT_BARRIER_NR
);
1202 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1203 * the activity log, which means it would not be resynced in case the
1204 * R_PRIMARY crashes now.
1205 * Therefore we must send the barrier_ack after the barrier request was
1207 switch (mdev
->write_ordering
) {
1209 if (rv
== FE_RECYCLED
)
1212 /* receiver context, in the writeout path of the other node.
1213 * avoid potential distributed deadlock */
1214 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1218 dev_warn(DEV
, "Allocation of an epoch failed, slowing down\n");
1223 drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
1226 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1227 epoch
= kmalloc(sizeof(struct drbd_epoch
), GFP_NOIO
);
1232 epoch
= mdev
->current_epoch
;
1233 wait_event(mdev
->ee_wait
, atomic_read(&epoch
->epoch_size
) == 0);
1235 D_ASSERT(atomic_read(&epoch
->active
) == 0);
1236 D_ASSERT(epoch
->flags
== 0);
1240 dev_err(DEV
, "Strangeness in mdev->write_ordering %d\n", mdev
->write_ordering
);
1245 atomic_set(&epoch
->epoch_size
, 0);
1246 atomic_set(&epoch
->active
, 0);
1248 spin_lock(&mdev
->epoch_lock
);
1249 if (atomic_read(&mdev
->current_epoch
->epoch_size
)) {
1250 list_add(&epoch
->list
, &mdev
->current_epoch
->list
);
1251 mdev
->current_epoch
= epoch
;
1254 /* The current_epoch got recycled while we allocated this one... */
1257 spin_unlock(&mdev
->epoch_lock
);
1262 /* used from receive_RSDataReply (recv_resync_read)
1263 * and from receive_Data */
1264 static struct drbd_peer_request
*
1265 read_in_block(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
1266 int data_size
) __must_hold(local
)
1268 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
1269 struct drbd_peer_request
*peer_req
;
1272 void *dig_in
= mdev
->tconn
->int_dig_in
;
1273 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1274 unsigned long *data
;
1276 dgs
= (mdev
->tconn
->agreed_pro_version
>= 87 && mdev
->tconn
->integrity_r_tfm
) ?
1277 crypto_hash_digestsize(mdev
->tconn
->integrity_r_tfm
) : 0;
1280 rr
= drbd_recv(mdev
, dig_in
, dgs
);
1282 if (!signal_pending(current
))
1284 "short read receiving data digest: read %d expected %d\n",
1292 if (!expect(data_size
!= 0))
1294 if (!expect(IS_ALIGNED(data_size
, 512)))
1296 if (!expect(data_size
<= DRBD_MAX_BIO_SIZE
))
1299 /* even though we trust out peer,
1300 * we sometimes have to double check. */
1301 if (sector
+ (data_size
>>9) > capacity
) {
1302 dev_err(DEV
, "request from peer beyond end of local disk: "
1303 "capacity: %llus < sector: %llus + size: %u\n",
1304 (unsigned long long)capacity
,
1305 (unsigned long long)sector
, data_size
);
1309 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1310 * "criss-cross" setup, that might cause write-out on some other DRBD,
1311 * which in turn might block on the other node at this very place. */
1312 peer_req
= drbd_alloc_ee(mdev
, id
, sector
, data_size
, GFP_NOIO
);
1317 page
= peer_req
->pages
;
1318 page_chain_for_each(page
) {
1319 unsigned len
= min_t(int, ds
, PAGE_SIZE
);
1321 rr
= drbd_recv(mdev
, data
, len
);
1322 if (drbd_insert_fault(mdev
, DRBD_FAULT_RECEIVE
)) {
1323 dev_err(DEV
, "Fault injection: Corrupting data on receive\n");
1324 data
[0] = data
[0] ^ (unsigned long)-1;
1328 drbd_free_ee(mdev
, peer_req
);
1329 if (!signal_pending(current
))
1330 dev_warn(DEV
, "short read receiving data: read %d expected %d\n",
1338 drbd_csum_ee(mdev
, mdev
->tconn
->integrity_r_tfm
, peer_req
, dig_vv
);
1339 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1340 dev_err(DEV
, "Digest integrity check FAILED: %llus +%u\n",
1341 (unsigned long long)sector
, data_size
);
1342 drbd_bcast_ee(mdev
, "digest failed",
1343 dgs
, dig_in
, dig_vv
, peer_req
);
1344 drbd_free_ee(mdev
, peer_req
);
1348 mdev
->recv_cnt
+= data_size
>>9;
1352 /* drbd_drain_block() just takes a data block
1353 * out of the socket input buffer, and discards it.
1355 static int drbd_drain_block(struct drbd_conf
*mdev
, int data_size
)
1364 page
= drbd_pp_alloc(mdev
, 1, 1);
1368 rr
= drbd_recv(mdev
, data
, min_t(int, data_size
, PAGE_SIZE
));
1369 if (rr
!= min_t(int, data_size
, PAGE_SIZE
)) {
1371 if (!signal_pending(current
))
1373 "short read receiving data: read %d expected %d\n",
1374 rr
, min_t(int, data_size
, PAGE_SIZE
));
1380 drbd_pp_free(mdev
, page
, 0);
1384 static int recv_dless_read(struct drbd_conf
*mdev
, struct drbd_request
*req
,
1385 sector_t sector
, int data_size
)
1387 struct bio_vec
*bvec
;
1389 int dgs
, rr
, i
, expect
;
1390 void *dig_in
= mdev
->tconn
->int_dig_in
;
1391 void *dig_vv
= mdev
->tconn
->int_dig_vv
;
1393 dgs
= (mdev
->tconn
->agreed_pro_version
>= 87 && mdev
->tconn
->integrity_r_tfm
) ?
1394 crypto_hash_digestsize(mdev
->tconn
->integrity_r_tfm
) : 0;
1397 rr
= drbd_recv(mdev
, dig_in
, dgs
);
1399 if (!signal_pending(current
))
1401 "short read receiving data reply digest: read %d expected %d\n",
1409 /* optimistically update recv_cnt. if receiving fails below,
1410 * we disconnect anyways, and counters will be reset. */
1411 mdev
->recv_cnt
+= data_size
>>9;
1413 bio
= req
->master_bio
;
1414 D_ASSERT(sector
== bio
->bi_sector
);
1416 bio_for_each_segment(bvec
, bio
, i
) {
1417 expect
= min_t(int, data_size
, bvec
->bv_len
);
1418 rr
= drbd_recv(mdev
,
1419 kmap(bvec
->bv_page
)+bvec
->bv_offset
,
1421 kunmap(bvec
->bv_page
);
1423 if (!signal_pending(current
))
1424 dev_warn(DEV
, "short read receiving data reply: "
1425 "read %d expected %d\n",
1433 drbd_csum_bio(mdev
, mdev
->tconn
->integrity_r_tfm
, bio
, dig_vv
);
1434 if (memcmp(dig_in
, dig_vv
, dgs
)) {
1435 dev_err(DEV
, "Digest integrity check FAILED. Broken NICs?\n");
1440 D_ASSERT(data_size
== 0);
1444 /* e_end_resync_block() is called via
1445 * drbd_process_done_ee() by asender only */
1446 static int e_end_resync_block(struct drbd_conf
*mdev
, struct drbd_work
*w
, int unused
)
1448 struct drbd_peer_request
*peer_req
= (struct drbd_peer_request
*)w
;
1449 sector_t sector
= peer_req
->i
.sector
;
1452 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1454 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1455 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1456 ok
= drbd_send_ack(mdev
, P_RS_WRITE_ACK
, peer_req
);
1458 /* Record failure to sync */
1459 drbd_rs_failed_io(mdev
, sector
, peer_req
->i
.size
);
1461 ok
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1468 static int recv_resync_read(struct drbd_conf
*mdev
, sector_t sector
, int data_size
) __releases(local
)
1470 struct drbd_peer_request
*peer_req
;
1472 peer_req
= read_in_block(mdev
, ID_SYNCER
, sector
, data_size
);
1476 dec_rs_pending(mdev
);
1479 /* corresponding dec_unacked() in e_end_resync_block()
1480 * respective _drbd_clear_done_ee */
1482 peer_req
->w
.cb
= e_end_resync_block
;
1484 spin_lock_irq(&mdev
->tconn
->req_lock
);
1485 list_add(&peer_req
->w
.list
, &mdev
->sync_ee
);
1486 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1488 atomic_add(data_size
>> 9, &mdev
->rs_sect_ev
);
1489 if (drbd_submit_ee(mdev
, peer_req
, WRITE
, DRBD_FAULT_RS_WR
) == 0)
1492 /* don't care for the reason here */
1493 dev_err(DEV
, "submit failed, triggering re-connect\n");
1494 spin_lock_irq(&mdev
->tconn
->req_lock
);
1495 list_del(&peer_req
->w
.list
);
1496 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1498 drbd_free_ee(mdev
, peer_req
);
1504 static struct drbd_request
*
1505 find_request(struct drbd_conf
*mdev
, struct rb_root
*root
, u64 id
,
1506 sector_t sector
, bool missing_ok
, const char *func
)
1508 struct drbd_request
*req
;
1510 /* Request object according to our peer */
1511 req
= (struct drbd_request
*)(unsigned long)id
;
1512 if (drbd_contains_interval(root
, sector
, &req
->i
) && req
->i
.local
)
1515 dev_err(DEV
, "%s: failed to find request %lu, sector %llus\n", func
,
1516 (unsigned long)id
, (unsigned long long)sector
);
1521 static int receive_DataReply(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
1522 unsigned int data_size
)
1524 struct drbd_request
*req
;
1527 struct p_data
*p
= &mdev
->tconn
->data
.rbuf
.data
;
1529 sector
= be64_to_cpu(p
->sector
);
1531 spin_lock_irq(&mdev
->tconn
->req_lock
);
1532 req
= find_request(mdev
, &mdev
->read_requests
, p
->block_id
, sector
, false, __func__
);
1533 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1537 /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1538 * special casing it there for the various failure cases.
1539 * still no race with drbd_fail_pending_reads */
1540 ok
= recv_dless_read(mdev
, req
, sector
, data_size
);
1543 req_mod(req
, DATA_RECEIVED
);
1544 /* else: nothing. handled from drbd_disconnect...
1545 * I don't think we may complete this just yet
1546 * in case we are "on-disconnect: freeze" */
1551 static int receive_RSDataReply(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
1552 unsigned int data_size
)
1556 struct p_data
*p
= &mdev
->tconn
->data
.rbuf
.data
;
1558 sector
= be64_to_cpu(p
->sector
);
1559 D_ASSERT(p
->block_id
== ID_SYNCER
);
1561 if (get_ldev(mdev
)) {
1562 /* data is submitted to disk within recv_resync_read.
1563 * corresponding put_ldev done below on error,
1564 * or in drbd_endio_sec. */
1565 ok
= recv_resync_read(mdev
, sector
, data_size
);
1567 if (__ratelimit(&drbd_ratelimit_state
))
1568 dev_err(DEV
, "Can not write resync data to local disk.\n");
1570 ok
= drbd_drain_block(mdev
, data_size
);
1572 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, data_size
);
1575 atomic_add(data_size
>> 9, &mdev
->rs_sect_in
);
1580 /* e_end_block() is called via drbd_process_done_ee().
1581 * this means this function only runs in the asender thread
1583 static int e_end_block(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1585 struct drbd_peer_request
*peer_req
= (struct drbd_peer_request
*)w
;
1586 sector_t sector
= peer_req
->i
.sector
;
1589 if (mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_C
) {
1590 if (likely((peer_req
->flags
& EE_WAS_ERROR
) == 0)) {
1591 pcmd
= (mdev
->state
.conn
>= C_SYNC_SOURCE
&&
1592 mdev
->state
.conn
<= C_PAUSED_SYNC_T
&&
1593 peer_req
->flags
& EE_MAY_SET_IN_SYNC
) ?
1594 P_RS_WRITE_ACK
: P_WRITE_ACK
;
1595 ok
&= drbd_send_ack(mdev
, pcmd
, peer_req
);
1596 if (pcmd
== P_RS_WRITE_ACK
)
1597 drbd_set_in_sync(mdev
, sector
, peer_req
->i
.size
);
1599 ok
= drbd_send_ack(mdev
, P_NEG_ACK
, peer_req
);
1600 /* we expect it to be marked out of sync anyways...
1601 * maybe assert this? */
1605 /* we delete from the conflict detection hash _after_ we sent out the
1606 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1607 if (mdev
->tconn
->net_conf
->two_primaries
) {
1608 spin_lock_irq(&mdev
->tconn
->req_lock
);
1609 D_ASSERT(!drbd_interval_empty(&peer_req
->i
));
1610 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
1611 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1613 D_ASSERT(drbd_interval_empty(&peer_req
->i
));
1615 drbd_may_finish_epoch(mdev
, peer_req
->epoch
, EV_PUT
+ (cancel
? EV_CLEANUP
: 0));
1620 static int e_send_discard_ack(struct drbd_conf
*mdev
, struct drbd_work
*w
, int unused
)
1622 struct drbd_peer_request
*peer_req
= (struct drbd_peer_request
*)w
;
1625 D_ASSERT(mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_C
);
1626 ok
= drbd_send_ack(mdev
, P_DISCARD_ACK
, peer_req
);
1628 spin_lock_irq(&mdev
->tconn
->req_lock
);
1629 D_ASSERT(!drbd_interval_empty(&peer_req
->i
));
1630 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
1631 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1638 static bool seq_greater(u32 a
, u32 b
)
1641 * We assume 32-bit wrap-around here.
1642 * For 24-bit wrap-around, we would have to shift:
1645 return (s32
)a
- (s32
)b
> 0;
1648 static u32
seq_max(u32 a
, u32 b
)
1650 return seq_greater(a
, b
) ? a
: b
;
1653 static void update_peer_seq(struct drbd_conf
*mdev
, unsigned int peer_seq
)
1655 unsigned int old_peer_seq
;
1657 spin_lock(&mdev
->peer_seq_lock
);
1658 old_peer_seq
= mdev
->peer_seq
;
1659 mdev
->peer_seq
= seq_max(mdev
->peer_seq
, peer_seq
);
1660 spin_unlock(&mdev
->peer_seq_lock
);
1661 if (old_peer_seq
!= peer_seq
)
1662 wake_up(&mdev
->seq_wait
);
1665 /* Called from receive_Data.
1666 * Synchronize packets on sock with packets on msock.
1668 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1669 * packet traveling on msock, they are still processed in the order they have
1672 * Note: we don't care for Ack packets overtaking P_DATA packets.
1674 * In case packet_seq is larger than mdev->peer_seq number, there are
1675 * outstanding packets on the msock. We wait for them to arrive.
1676 * In case we are the logically next packet, we update mdev->peer_seq
1677 * ourselves. Correctly handles 32bit wrap around.
1679 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1680 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1681 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1682 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1684 * returns 0 if we may process the packet,
1685 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1686 static int drbd_wait_peer_seq(struct drbd_conf
*mdev
, const u32 packet_seq
)
1692 spin_lock(&mdev
->peer_seq_lock
);
1694 prepare_to_wait(&mdev
->seq_wait
, &wait
, TASK_INTERRUPTIBLE
);
1695 if (!seq_greater(packet_seq
, mdev
->peer_seq
+ 1))
1697 if (signal_pending(current
)) {
1701 p_seq
= mdev
->peer_seq
;
1702 spin_unlock(&mdev
->peer_seq_lock
);
1703 timeout
= schedule_timeout(30*HZ
);
1704 spin_lock(&mdev
->peer_seq_lock
);
1705 if (timeout
== 0 && p_seq
== mdev
->peer_seq
) {
1707 dev_err(DEV
, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n");
1711 finish_wait(&mdev
->seq_wait
, &wait
);
1712 if (mdev
->peer_seq
+1 == packet_seq
)
1714 spin_unlock(&mdev
->peer_seq_lock
);
1718 /* see also bio_flags_to_wire()
1719 * DRBD_REQ_*, because we need to semantically map the flags to data packet
1720 * flags and back. We may replicate to other kernel versions. */
1721 static unsigned long wire_flags_to_bio(struct drbd_conf
*mdev
, u32 dpf
)
1723 return (dpf
& DP_RW_SYNC
? REQ_SYNC
: 0) |
1724 (dpf
& DP_FUA
? REQ_FUA
: 0) |
1725 (dpf
& DP_FLUSH
? REQ_FLUSH
: 0) |
1726 (dpf
& DP_DISCARD
? REQ_DISCARD
: 0);
1729 /* mirrored write */
1730 static int receive_Data(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
1731 unsigned int data_size
)
1734 struct drbd_peer_request
*peer_req
;
1735 struct p_data
*p
= &mdev
->tconn
->data
.rbuf
.data
;
1739 if (!get_ldev(mdev
)) {
1740 spin_lock(&mdev
->peer_seq_lock
);
1741 if (mdev
->peer_seq
+1 == be32_to_cpu(p
->seq_num
))
1743 spin_unlock(&mdev
->peer_seq_lock
);
1745 drbd_send_ack_dp(mdev
, P_NEG_ACK
, p
, data_size
);
1746 atomic_inc(&mdev
->current_epoch
->epoch_size
);
1747 return drbd_drain_block(mdev
, data_size
);
1750 /* get_ldev(mdev) successful.
1751 * Corresponding put_ldev done either below (on various errors),
1752 * or in drbd_endio_sec, if we successfully submit the data at
1753 * the end of this function. */
1755 sector
= be64_to_cpu(p
->sector
);
1756 peer_req
= read_in_block(mdev
, p
->block_id
, sector
, data_size
);
1762 peer_req
->w
.cb
= e_end_block
;
1764 dp_flags
= be32_to_cpu(p
->dp_flags
);
1765 rw
|= wire_flags_to_bio(mdev
, dp_flags
);
1767 if (dp_flags
& DP_MAY_SET_IN_SYNC
)
1768 peer_req
->flags
|= EE_MAY_SET_IN_SYNC
;
1770 spin_lock(&mdev
->epoch_lock
);
1771 peer_req
->epoch
= mdev
->current_epoch
;
1772 atomic_inc(&peer_req
->epoch
->epoch_size
);
1773 atomic_inc(&peer_req
->epoch
->active
);
1774 spin_unlock(&mdev
->epoch_lock
);
1776 /* I'm the receiver, I do hold a net_cnt reference. */
1777 if (!mdev
->tconn
->net_conf
->two_primaries
) {
1778 spin_lock_irq(&mdev
->tconn
->req_lock
);
1780 /* don't get the req_lock yet,
1781 * we may sleep in drbd_wait_peer_seq */
1782 const int size
= peer_req
->i
.size
;
1783 const int discard
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
1787 D_ASSERT(mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_C
);
1789 /* conflict detection and handling:
1790 * 1. wait on the sequence number,
1791 * in case this data packet overtook ACK packets.
1792 * 2. check for conflicting write requests.
1794 * Note: for two_primaries, we are protocol C,
1795 * so there cannot be any request that is DONE
1796 * but still on the transfer log.
1798 * if no conflicting request is found:
1801 * if any conflicting request is found
1802 * that has not yet been acked,
1803 * AND I have the "discard concurrent writes" flag:
1804 * queue (via done_ee) the P_DISCARD_ACK; OUT.
1806 * if any conflicting request is found:
1807 * block the receiver, waiting on misc_wait
1808 * until no more conflicting requests are there,
1809 * or we get interrupted (disconnect).
1811 * we do not just write after local io completion of those
1812 * requests, but only after req is done completely, i.e.
1813 * we wait for the P_DISCARD_ACK to arrive!
1815 * then proceed normally, i.e. submit.
1817 if (drbd_wait_peer_seq(mdev
, be32_to_cpu(p
->seq_num
)))
1818 goto out_interrupted
;
1820 spin_lock_irq(&mdev
->tconn
->req_lock
);
1824 struct drbd_interval
*i
;
1825 int have_unacked
= 0;
1826 int have_conflict
= 0;
1827 prepare_to_wait(&mdev
->misc_wait
, &wait
,
1828 TASK_INTERRUPTIBLE
);
1830 i
= drbd_find_overlap(&mdev
->write_requests
, sector
, size
);
1832 /* only ALERT on first iteration,
1833 * we may be woken up early... */
1835 dev_alert(DEV
, "%s[%u] Concurrent %s write detected!"
1836 " new: %llus +%u; pending: %llus +%u\n",
1837 current
->comm
, current
->pid
,
1838 i
->local
? "local" : "remote",
1839 (unsigned long long)sector
, size
,
1840 (unsigned long long)i
->sector
, i
->size
);
1843 struct drbd_request
*req2
;
1845 req2
= container_of(i
, struct drbd_request
, i
);
1846 if (req2
->rq_state
& RQ_NET_PENDING
)
1854 /* Discard Ack only for the _first_ iteration */
1855 if (first
&& discard
&& have_unacked
) {
1856 dev_alert(DEV
, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n",
1857 (unsigned long long)sector
);
1859 peer_req
->w
.cb
= e_send_discard_ack
;
1860 list_add_tail(&peer_req
->w
.list
, &mdev
->done_ee
);
1862 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1864 /* we could probably send that P_DISCARD_ACK ourselves,
1865 * but I don't like the receiver using the msock */
1869 finish_wait(&mdev
->misc_wait
, &wait
);
1873 if (signal_pending(current
)) {
1874 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1875 finish_wait(&mdev
->misc_wait
, &wait
);
1876 goto out_interrupted
;
1879 /* Indicate to wake up mdev->misc_wait upon completion. */
1882 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1885 dev_alert(DEV
, "Concurrent write! [W AFTERWARDS] "
1886 "sec=%llus\n", (unsigned long long)sector
);
1887 } else if (discard
) {
1888 /* we had none on the first iteration.
1889 * there must be none now. */
1890 D_ASSERT(have_unacked
== 0);
1893 spin_lock_irq(&mdev
->tconn
->req_lock
);
1895 finish_wait(&mdev
->misc_wait
, &wait
);
1897 drbd_insert_interval(&mdev
->write_requests
, &peer_req
->i
);
1900 list_add(&peer_req
->w
.list
, &mdev
->active_ee
);
1901 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1903 switch (mdev
->tconn
->net_conf
->wire_protocol
) {
1906 /* corresponding dec_unacked() in e_end_block()
1907 * respective _drbd_clear_done_ee */
1910 /* I really don't like it that the receiver thread
1911 * sends on the msock, but anyways */
1912 drbd_send_ack(mdev
, P_RECV_ACK
, peer_req
);
1919 if (mdev
->state
.pdsk
< D_INCONSISTENT
) {
1920 /* In case we have the only disk of the cluster, */
1921 drbd_set_out_of_sync(mdev
, peer_req
->i
.sector
, peer_req
->i
.size
);
1922 peer_req
->flags
|= EE_CALL_AL_COMPLETE_IO
;
1923 peer_req
->flags
&= ~EE_MAY_SET_IN_SYNC
;
1924 drbd_al_begin_io(mdev
, peer_req
->i
.sector
);
1927 if (drbd_submit_ee(mdev
, peer_req
, rw
, DRBD_FAULT_DT_WR
) == 0)
1930 /* don't care for the reason here */
1931 dev_err(DEV
, "submit failed, triggering re-connect\n");
1932 spin_lock_irq(&mdev
->tconn
->req_lock
);
1933 list_del(&peer_req
->w
.list
);
1934 drbd_remove_epoch_entry_interval(mdev
, peer_req
);
1935 spin_unlock_irq(&mdev
->tconn
->req_lock
);
1936 if (peer_req
->flags
& EE_CALL_AL_COMPLETE_IO
)
1937 drbd_al_complete_io(mdev
, peer_req
->i
.sector
);
1940 drbd_may_finish_epoch(mdev
, peer_req
->epoch
, EV_PUT
+ EV_CLEANUP
);
1942 drbd_free_ee(mdev
, peer_req
);
1946 /* We may throttle resync, if the lower device seems to be busy,
1947 * and current sync rate is above c_min_rate.
1949 * To decide whether or not the lower device is busy, we use a scheme similar
1950 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
1951 * (more than 64 sectors) of activity we cannot account for with our own resync
1952 * activity, it obviously is "busy".
1954 * The current sync rate used here uses only the most recent two step marks,
1955 * to have a short time average so we can react faster.
1957 int drbd_rs_should_slow_down(struct drbd_conf
*mdev
, sector_t sector
)
1959 struct gendisk
*disk
= mdev
->ldev
->backing_bdev
->bd_contains
->bd_disk
;
1960 unsigned long db
, dt
, dbdt
;
1961 struct lc_element
*tmp
;
1965 /* feature disabled? */
1966 if (mdev
->sync_conf
.c_min_rate
== 0)
1969 spin_lock_irq(&mdev
->al_lock
);
1970 tmp
= lc_find(mdev
->resync
, BM_SECT_TO_EXT(sector
));
1972 struct bm_extent
*bm_ext
= lc_entry(tmp
, struct bm_extent
, lce
);
1973 if (test_bit(BME_PRIORITY
, &bm_ext
->flags
)) {
1974 spin_unlock_irq(&mdev
->al_lock
);
1977 /* Do not slow down if app IO is already waiting for this extent */
1979 spin_unlock_irq(&mdev
->al_lock
);
1981 curr_events
= (int)part_stat_read(&disk
->part0
, sectors
[0]) +
1982 (int)part_stat_read(&disk
->part0
, sectors
[1]) -
1983 atomic_read(&mdev
->rs_sect_ev
);
1985 if (!mdev
->rs_last_events
|| curr_events
- mdev
->rs_last_events
> 64) {
1986 unsigned long rs_left
;
1989 mdev
->rs_last_events
= curr_events
;
1991 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
1993 i
= (mdev
->rs_last_mark
+ DRBD_SYNC_MARKS
-1) % DRBD_SYNC_MARKS
;
1995 if (mdev
->state
.conn
== C_VERIFY_S
|| mdev
->state
.conn
== C_VERIFY_T
)
1996 rs_left
= mdev
->ov_left
;
1998 rs_left
= drbd_bm_total_weight(mdev
) - mdev
->rs_failed
;
2000 dt
= ((long)jiffies
- (long)mdev
->rs_mark_time
[i
]) / HZ
;
2003 db
= mdev
->rs_mark_left
[i
] - rs_left
;
2004 dbdt
= Bit2KB(db
/dt
);
2006 if (dbdt
> mdev
->sync_conf
.c_min_rate
)
2013 static int receive_DataRequest(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
2014 unsigned int digest_size
)
2017 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
2018 struct drbd_peer_request
*peer_req
;
2019 struct digest_info
*di
= NULL
;
2021 unsigned int fault_type
;
2022 struct p_block_req
*p
= &mdev
->tconn
->data
.rbuf
.block_req
;
2024 sector
= be64_to_cpu(p
->sector
);
2025 size
= be32_to_cpu(p
->blksize
);
2027 if (size
<= 0 || (size
& 0x1ff) != 0 || size
> DRBD_MAX_BIO_SIZE
) {
2028 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2029 (unsigned long long)sector
, size
);
2032 if (sector
+ (size
>>9) > capacity
) {
2033 dev_err(DEV
, "%s:%d: sector: %llus, size: %u\n", __FILE__
, __LINE__
,
2034 (unsigned long long)sector
, size
);
2038 if (!get_ldev_if_state(mdev
, D_UP_TO_DATE
)) {
2041 case P_DATA_REQUEST
:
2042 drbd_send_ack_rp(mdev
, P_NEG_DREPLY
, p
);
2044 case P_RS_DATA_REQUEST
:
2045 case P_CSUM_RS_REQUEST
:
2047 drbd_send_ack_rp(mdev
, P_NEG_RS_DREPLY
, p
);
2051 dec_rs_pending(mdev
);
2052 drbd_send_ack_ex(mdev
, P_OV_RESULT
, sector
, size
, ID_IN_SYNC
);
2055 dev_err(DEV
, "unexpected command (%s) in receive_DataRequest\n",
2058 if (verb
&& __ratelimit(&drbd_ratelimit_state
))
2059 dev_err(DEV
, "Can not satisfy peer's read request, "
2060 "no local data.\n");
2062 /* drain possibly payload */
2063 return drbd_drain_block(mdev
, digest_size
);
2066 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2067 * "criss-cross" setup, that might cause write-out on some other DRBD,
2068 * which in turn might block on the other node at this very place. */
2069 peer_req
= drbd_alloc_ee(mdev
, p
->block_id
, sector
, size
, GFP_NOIO
);
2076 case P_DATA_REQUEST
:
2077 peer_req
->w
.cb
= w_e_end_data_req
;
2078 fault_type
= DRBD_FAULT_DT_RD
;
2079 /* application IO, don't drbd_rs_begin_io */
2082 case P_RS_DATA_REQUEST
:
2083 peer_req
->w
.cb
= w_e_end_rsdata_req
;
2084 fault_type
= DRBD_FAULT_RS_RD
;
2085 /* used in the sector offset progress display */
2086 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2090 case P_CSUM_RS_REQUEST
:
2091 fault_type
= DRBD_FAULT_RS_RD
;
2092 di
= kmalloc(sizeof(*di
) + digest_size
, GFP_NOIO
);
2096 di
->digest_size
= digest_size
;
2097 di
->digest
= (((char *)di
)+sizeof(struct digest_info
));
2099 peer_req
->digest
= di
;
2100 peer_req
->flags
|= EE_HAS_DIGEST
;
2102 if (drbd_recv(mdev
, di
->digest
, digest_size
) != digest_size
)
2105 if (cmd
== P_CSUM_RS_REQUEST
) {
2106 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
2107 peer_req
->w
.cb
= w_e_end_csum_rs_req
;
2108 /* used in the sector offset progress display */
2109 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
2110 } else if (cmd
== P_OV_REPLY
) {
2111 /* track progress, we may need to throttle */
2112 atomic_add(size
>> 9, &mdev
->rs_sect_in
);
2113 peer_req
->w
.cb
= w_e_end_ov_reply
;
2114 dec_rs_pending(mdev
);
2115 /* drbd_rs_begin_io done when we sent this request,
2116 * but accounting still needs to be done. */
2117 goto submit_for_resync
;
2122 if (mdev
->ov_start_sector
== ~(sector_t
)0 &&
2123 mdev
->tconn
->agreed_pro_version
>= 90) {
2124 unsigned long now
= jiffies
;
2126 mdev
->ov_start_sector
= sector
;
2127 mdev
->ov_position
= sector
;
2128 mdev
->ov_left
= drbd_bm_bits(mdev
) - BM_SECT_TO_BIT(sector
);
2129 mdev
->rs_total
= mdev
->ov_left
;
2130 for (i
= 0; i
< DRBD_SYNC_MARKS
; i
++) {
2131 mdev
->rs_mark_left
[i
] = mdev
->ov_left
;
2132 mdev
->rs_mark_time
[i
] = now
;
2134 dev_info(DEV
, "Online Verify start sector: %llu\n",
2135 (unsigned long long)sector
);
2137 peer_req
->w
.cb
= w_e_end_ov_req
;
2138 fault_type
= DRBD_FAULT_RS_RD
;
2142 dev_err(DEV
, "unexpected command (%s) in receive_DataRequest\n",
2144 fault_type
= DRBD_FAULT_MAX
;
2148 /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2149 * wrt the receiver, but it is not as straightforward as it may seem.
2150 * Various places in the resync start and stop logic assume resync
2151 * requests are processed in order, requeuing this on the worker thread
2152 * introduces a bunch of new code for synchronization between threads.
2154 * Unlimited throttling before drbd_rs_begin_io may stall the resync
2155 * "forever", throttling after drbd_rs_begin_io will lock that extent
2156 * for application writes for the same time. For now, just throttle
2157 * here, where the rest of the code expects the receiver to sleep for
2161 /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2162 * this defers syncer requests for some time, before letting at least
2163 * on request through. The resync controller on the receiving side
2164 * will adapt to the incoming rate accordingly.
2166 * We cannot throttle here if remote is Primary/SyncTarget:
2167 * we would also throttle its application reads.
2168 * In that case, throttling is done on the SyncTarget only.
2170 if (mdev
->state
.peer
!= R_PRIMARY
&& drbd_rs_should_slow_down(mdev
, sector
))
2171 schedule_timeout_uninterruptible(HZ
/10);
2172 if (drbd_rs_begin_io(mdev
, sector
))
2176 atomic_add(size
>> 9, &mdev
->rs_sect_ev
);
2180 spin_lock_irq(&mdev
->tconn
->req_lock
);
2181 list_add_tail(&peer_req
->w
.list
, &mdev
->read_ee
);
2182 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2184 if (drbd_submit_ee(mdev
, peer_req
, READ
, fault_type
) == 0)
2187 /* don't care for the reason here */
2188 dev_err(DEV
, "submit failed, triggering re-connect\n");
2189 spin_lock_irq(&mdev
->tconn
->req_lock
);
2190 list_del(&peer_req
->w
.list
);
2191 spin_unlock_irq(&mdev
->tconn
->req_lock
);
2192 /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2196 drbd_free_ee(mdev
, peer_req
);
2200 static int drbd_asb_recover_0p(struct drbd_conf
*mdev
) __must_hold(local
)
2202 int self
, peer
, rv
= -100;
2203 unsigned long ch_self
, ch_peer
;
2205 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & 1;
2206 peer
= mdev
->p_uuid
[UI_BITMAP
] & 1;
2208 ch_peer
= mdev
->p_uuid
[UI_SIZE
];
2209 ch_self
= mdev
->comm_bm_set
;
2211 switch (mdev
->tconn
->net_conf
->after_sb_0p
) {
2213 case ASB_DISCARD_SECONDARY
:
2214 case ASB_CALL_HELPER
:
2215 dev_err(DEV
, "Configuration error.\n");
2217 case ASB_DISCONNECT
:
2219 case ASB_DISCARD_YOUNGER_PRI
:
2220 if (self
== 0 && peer
== 1) {
2224 if (self
== 1 && peer
== 0) {
2228 /* Else fall through to one of the other strategies... */
2229 case ASB_DISCARD_OLDER_PRI
:
2230 if (self
== 0 && peer
== 1) {
2234 if (self
== 1 && peer
== 0) {
2238 /* Else fall through to one of the other strategies... */
2239 dev_warn(DEV
, "Discard younger/older primary did not find a decision\n"
2240 "Using discard-least-changes instead\n");
2241 case ASB_DISCARD_ZERO_CHG
:
2242 if (ch_peer
== 0 && ch_self
== 0) {
2243 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
)
2247 if (ch_peer
== 0) { rv
= 1; break; }
2248 if (ch_self
== 0) { rv
= -1; break; }
2250 if (mdev
->tconn
->net_conf
->after_sb_0p
== ASB_DISCARD_ZERO_CHG
)
2252 case ASB_DISCARD_LEAST_CHG
:
2253 if (ch_self
< ch_peer
)
2255 else if (ch_self
> ch_peer
)
2257 else /* ( ch_self == ch_peer ) */
2258 /* Well, then use something else. */
2259 rv
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
)
2262 case ASB_DISCARD_LOCAL
:
2265 case ASB_DISCARD_REMOTE
:
2272 static int drbd_asb_recover_1p(struct drbd_conf
*mdev
) __must_hold(local
)
2276 switch (mdev
->tconn
->net_conf
->after_sb_1p
) {
2277 case ASB_DISCARD_YOUNGER_PRI
:
2278 case ASB_DISCARD_OLDER_PRI
:
2279 case ASB_DISCARD_LEAST_CHG
:
2280 case ASB_DISCARD_LOCAL
:
2281 case ASB_DISCARD_REMOTE
:
2282 dev_err(DEV
, "Configuration error.\n");
2284 case ASB_DISCONNECT
:
2287 hg
= drbd_asb_recover_0p(mdev
);
2288 if (hg
== -1 && mdev
->state
.role
== R_SECONDARY
)
2290 if (hg
== 1 && mdev
->state
.role
== R_PRIMARY
)
2294 rv
= drbd_asb_recover_0p(mdev
);
2296 case ASB_DISCARD_SECONDARY
:
2297 return mdev
->state
.role
== R_PRIMARY
? 1 : -1;
2298 case ASB_CALL_HELPER
:
2299 hg
= drbd_asb_recover_0p(mdev
);
2300 if (hg
== -1 && mdev
->state
.role
== R_PRIMARY
) {
2301 enum drbd_state_rv rv2
;
2303 drbd_set_role(mdev
, R_SECONDARY
, 0);
2304 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2305 * we might be here in C_WF_REPORT_PARAMS which is transient.
2306 * we do not need to wait for the after state change work either. */
2307 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2308 if (rv2
!= SS_SUCCESS
) {
2309 drbd_khelper(mdev
, "pri-lost-after-sb");
2311 dev_warn(DEV
, "Successfully gave up primary role.\n");
2321 static int drbd_asb_recover_2p(struct drbd_conf
*mdev
) __must_hold(local
)
2325 switch (mdev
->tconn
->net_conf
->after_sb_2p
) {
2326 case ASB_DISCARD_YOUNGER_PRI
:
2327 case ASB_DISCARD_OLDER_PRI
:
2328 case ASB_DISCARD_LEAST_CHG
:
2329 case ASB_DISCARD_LOCAL
:
2330 case ASB_DISCARD_REMOTE
:
2332 case ASB_DISCARD_SECONDARY
:
2333 dev_err(DEV
, "Configuration error.\n");
2336 rv
= drbd_asb_recover_0p(mdev
);
2338 case ASB_DISCONNECT
:
2340 case ASB_CALL_HELPER
:
2341 hg
= drbd_asb_recover_0p(mdev
);
2343 enum drbd_state_rv rv2
;
2345 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2346 * we might be here in C_WF_REPORT_PARAMS which is transient.
2347 * we do not need to wait for the after state change work either. */
2348 rv2
= drbd_change_state(mdev
, CS_VERBOSE
, NS(role
, R_SECONDARY
));
2349 if (rv2
!= SS_SUCCESS
) {
2350 drbd_khelper(mdev
, "pri-lost-after-sb");
2352 dev_warn(DEV
, "Successfully gave up primary role.\n");
2362 static void drbd_uuid_dump(struct drbd_conf
*mdev
, char *text
, u64
*uuid
,
2363 u64 bits
, u64 flags
)
2366 dev_info(DEV
, "%s uuid info vanished while I was looking!\n", text
);
2369 dev_info(DEV
, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2371 (unsigned long long)uuid
[UI_CURRENT
],
2372 (unsigned long long)uuid
[UI_BITMAP
],
2373 (unsigned long long)uuid
[UI_HISTORY_START
],
2374 (unsigned long long)uuid
[UI_HISTORY_END
],
2375 (unsigned long long)bits
,
2376 (unsigned long long)flags
);
2380 100 after split brain try auto recover
2381 2 C_SYNC_SOURCE set BitMap
2382 1 C_SYNC_SOURCE use BitMap
2384 -1 C_SYNC_TARGET use BitMap
2385 -2 C_SYNC_TARGET set BitMap
2386 -100 after split brain, disconnect
2387 -1000 unrelated data
2388 -1091 requires proto 91
2389 -1096 requires proto 96
2391 static int drbd_uuid_compare(struct drbd_conf
*mdev
, int *rule_nr
) __must_hold(local
)
2396 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2397 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2400 if (self
== UUID_JUST_CREATED
&& peer
== UUID_JUST_CREATED
)
2404 if ((self
== UUID_JUST_CREATED
|| self
== (u64
)0) &&
2405 peer
!= UUID_JUST_CREATED
)
2409 if (self
!= UUID_JUST_CREATED
&&
2410 (peer
== UUID_JUST_CREATED
|| peer
== (u64
)0))
2414 int rct
, dc
; /* roles at crash time */
2416 if (mdev
->p_uuid
[UI_BITMAP
] == (u64
)0 && mdev
->ldev
->md
.uuid
[UI_BITMAP
] != (u64
)0) {
2418 if (mdev
->tconn
->agreed_pro_version
< 91)
2421 if ((mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) &&
2422 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1))) {
2423 dev_info(DEV
, "was SyncSource, missed the resync finished event, corrected myself:\n");
2424 drbd_uuid_set_bm(mdev
, 0UL);
2426 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2427 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2430 dev_info(DEV
, "was SyncSource (peer failed to write sync_uuid)\n");
2437 if (mdev
->ldev
->md
.uuid
[UI_BITMAP
] == (u64
)0 && mdev
->p_uuid
[UI_BITMAP
] != (u64
)0) {
2439 if (mdev
->tconn
->agreed_pro_version
< 91)
2442 if ((mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) == (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1)) &&
2443 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) == (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1))) {
2444 dev_info(DEV
, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2446 mdev
->p_uuid
[UI_HISTORY_START
+ 1] = mdev
->p_uuid
[UI_HISTORY_START
];
2447 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_BITMAP
];
2448 mdev
->p_uuid
[UI_BITMAP
] = 0UL;
2450 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2453 dev_info(DEV
, "was SyncTarget (failed to write sync_uuid)\n");
2460 /* Common power [off|failure] */
2461 rct
= (test_bit(CRASHED_PRIMARY
, &mdev
->flags
) ? 1 : 0) +
2462 (mdev
->p_uuid
[UI_FLAGS
] & 2);
2463 /* lowest bit is set when we were primary,
2464 * next bit (weight 2) is set when peer was primary */
2468 case 0: /* !self_pri && !peer_pri */ return 0;
2469 case 1: /* self_pri && !peer_pri */ return 1;
2470 case 2: /* !self_pri && peer_pri */ return -1;
2471 case 3: /* self_pri && peer_pri */
2472 dc
= test_bit(DISCARD_CONCURRENT
, &mdev
->flags
);
2478 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2483 peer
= mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1);
2485 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2486 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1)) ==
2487 (mdev
->p_uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) :
2488 peer
+ UUID_NEW_BM_OFFSET
== (mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1))) {
2489 /* The last P_SYNC_UUID did not get though. Undo the last start of
2490 resync as sync source modifications of the peer's UUIDs. */
2492 if (mdev
->tconn
->agreed_pro_version
< 91)
2495 mdev
->p_uuid
[UI_BITMAP
] = mdev
->p_uuid
[UI_HISTORY_START
];
2496 mdev
->p_uuid
[UI_HISTORY_START
] = mdev
->p_uuid
[UI_HISTORY_START
+ 1];
2498 dev_info(DEV
, "Did not got last syncUUID packet, corrected:\n");
2499 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
, mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2506 self
= mdev
->ldev
->md
.uuid
[UI_CURRENT
] & ~((u64
)1);
2507 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2508 peer
= mdev
->p_uuid
[i
] & ~((u64
)1);
2514 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2515 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2520 self
= mdev
->ldev
->md
.uuid
[UI_HISTORY_START
] & ~((u64
)1);
2522 if (mdev
->tconn
->agreed_pro_version
< 96 ?
2523 (mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1] & ~((u64
)1)) ==
2524 (mdev
->p_uuid
[UI_HISTORY_START
] & ~((u64
)1)) :
2525 self
+ UUID_NEW_BM_OFFSET
== (mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1))) {
2526 /* The last P_SYNC_UUID did not get though. Undo the last start of
2527 resync as sync source modifications of our UUIDs. */
2529 if (mdev
->tconn
->agreed_pro_version
< 91)
2532 _drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
]);
2533 _drbd_uuid_set(mdev
, UI_HISTORY_START
, mdev
->ldev
->md
.uuid
[UI_HISTORY_START
+ 1]);
2535 dev_info(DEV
, "Last syncUUID did not get through, corrected:\n");
2536 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
,
2537 mdev
->state
.disk
>= D_NEGOTIATING
? drbd_bm_total_weight(mdev
) : 0, 0);
2545 peer
= mdev
->p_uuid
[UI_CURRENT
] & ~((u64
)1);
2546 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2547 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2553 self
= mdev
->ldev
->md
.uuid
[UI_BITMAP
] & ~((u64
)1);
2554 peer
= mdev
->p_uuid
[UI_BITMAP
] & ~((u64
)1);
2555 if (self
== peer
&& self
!= ((u64
)0))
2559 for (i
= UI_HISTORY_START
; i
<= UI_HISTORY_END
; i
++) {
2560 self
= mdev
->ldev
->md
.uuid
[i
] & ~((u64
)1);
2561 for (j
= UI_HISTORY_START
; j
<= UI_HISTORY_END
; j
++) {
2562 peer
= mdev
->p_uuid
[j
] & ~((u64
)1);
2571 /* drbd_sync_handshake() returns the new conn state on success, or
2572 CONN_MASK (-1) on failure.
2574 static enum drbd_conns
drbd_sync_handshake(struct drbd_conf
*mdev
, enum drbd_role peer_role
,
2575 enum drbd_disk_state peer_disk
) __must_hold(local
)
2578 enum drbd_conns rv
= C_MASK
;
2579 enum drbd_disk_state mydisk
;
2581 mydisk
= mdev
->state
.disk
;
2582 if (mydisk
== D_NEGOTIATING
)
2583 mydisk
= mdev
->new_state_tmp
.disk
;
2585 dev_info(DEV
, "drbd_sync_handshake:\n");
2586 drbd_uuid_dump(mdev
, "self", mdev
->ldev
->md
.uuid
, mdev
->comm_bm_set
, 0);
2587 drbd_uuid_dump(mdev
, "peer", mdev
->p_uuid
,
2588 mdev
->p_uuid
[UI_SIZE
], mdev
->p_uuid
[UI_FLAGS
]);
2590 hg
= drbd_uuid_compare(mdev
, &rule_nr
);
2592 dev_info(DEV
, "uuid_compare()=%d by rule %d\n", hg
, rule_nr
);
2595 dev_alert(DEV
, "Unrelated data, aborting!\n");
2599 dev_alert(DEV
, "To resolve this both sides have to support at least protocol %d\n", -hg
- 1000);
2603 if ((mydisk
== D_INCONSISTENT
&& peer_disk
> D_INCONSISTENT
) ||
2604 (peer_disk
== D_INCONSISTENT
&& mydisk
> D_INCONSISTENT
)) {
2605 int f
= (hg
== -100) || abs(hg
) == 2;
2606 hg
= mydisk
> D_INCONSISTENT
? 1 : -1;
2609 dev_info(DEV
, "Becoming sync %s due to disk states.\n",
2610 hg
> 0 ? "source" : "target");
2614 drbd_khelper(mdev
, "initial-split-brain");
2616 if (hg
== 100 || (hg
== -100 && mdev
->tconn
->net_conf
->always_asbp
)) {
2617 int pcount
= (mdev
->state
.role
== R_PRIMARY
)
2618 + (peer_role
== R_PRIMARY
);
2619 int forced
= (hg
== -100);
2623 hg
= drbd_asb_recover_0p(mdev
);
2626 hg
= drbd_asb_recover_1p(mdev
);
2629 hg
= drbd_asb_recover_2p(mdev
);
2632 if (abs(hg
) < 100) {
2633 dev_warn(DEV
, "Split-Brain detected, %d primaries, "
2634 "automatically solved. Sync from %s node\n",
2635 pcount
, (hg
< 0) ? "peer" : "this");
2637 dev_warn(DEV
, "Doing a full sync, since"
2638 " UUIDs where ambiguous.\n");
2645 if (mdev
->tconn
->net_conf
->want_lose
&& !(mdev
->p_uuid
[UI_FLAGS
]&1))
2647 if (!mdev
->tconn
->net_conf
->want_lose
&& (mdev
->p_uuid
[UI_FLAGS
]&1))
2651 dev_warn(DEV
, "Split-Brain detected, manually solved. "
2652 "Sync from %s node\n",
2653 (hg
< 0) ? "peer" : "this");
2657 /* FIXME this log message is not correct if we end up here
2658 * after an attempted attach on a diskless node.
2659 * We just refuse to attach -- well, we drop the "connection"
2660 * to that disk, in a way... */
2661 dev_alert(DEV
, "Split-Brain detected but unresolved, dropping connection!\n");
2662 drbd_khelper(mdev
, "split-brain");
2666 if (hg
> 0 && mydisk
<= D_INCONSISTENT
) {
2667 dev_err(DEV
, "I shall become SyncSource, but I am inconsistent!\n");
2671 if (hg
< 0 && /* by intention we do not use mydisk here. */
2672 mdev
->state
.role
== R_PRIMARY
&& mdev
->state
.disk
>= D_CONSISTENT
) {
2673 switch (mdev
->tconn
->net_conf
->rr_conflict
) {
2674 case ASB_CALL_HELPER
:
2675 drbd_khelper(mdev
, "pri-lost");
2677 case ASB_DISCONNECT
:
2678 dev_err(DEV
, "I shall become SyncTarget, but I am primary!\n");
2681 dev_warn(DEV
, "Becoming SyncTarget, violating the stable-data"
2686 if (mdev
->tconn
->net_conf
->dry_run
|| test_bit(CONN_DRY_RUN
, &mdev
->flags
)) {
2688 dev_info(DEV
, "dry-run connect: No resync, would become Connected immediately.\n");
2690 dev_info(DEV
, "dry-run connect: Would become %s, doing a %s resync.",
2691 drbd_conn_str(hg
> 0 ? C_SYNC_SOURCE
: C_SYNC_TARGET
),
2692 abs(hg
) >= 2 ? "full" : "bit-map based");
2697 dev_info(DEV
, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2698 if (drbd_bitmap_io(mdev
, &drbd_bmio_set_n_write
, "set_n_write from sync_handshake",
2699 BM_LOCKED_SET_ALLOWED
))
2703 if (hg
> 0) { /* become sync source. */
2705 } else if (hg
< 0) { /* become sync target */
2709 if (drbd_bm_total_weight(mdev
)) {
2710 dev_info(DEV
, "No resync, but %lu bits in bitmap!\n",
2711 drbd_bm_total_weight(mdev
));
2718 /* returns 1 if invalid */
2719 static int cmp_after_sb(enum drbd_after_sb_p peer
, enum drbd_after_sb_p self
)
2721 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2722 if ((peer
== ASB_DISCARD_REMOTE
&& self
== ASB_DISCARD_LOCAL
) ||
2723 (self
== ASB_DISCARD_REMOTE
&& peer
== ASB_DISCARD_LOCAL
))
2726 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2727 if (peer
== ASB_DISCARD_REMOTE
|| peer
== ASB_DISCARD_LOCAL
||
2728 self
== ASB_DISCARD_REMOTE
|| self
== ASB_DISCARD_LOCAL
)
2731 /* everything else is valid if they are equal on both sides. */
2735 /* everything es is invalid. */
2739 static int receive_protocol(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
2740 unsigned int data_size
)
2742 struct p_protocol
*p
= &mdev
->tconn
->data
.rbuf
.protocol
;
2743 int p_proto
, p_after_sb_0p
, p_after_sb_1p
, p_after_sb_2p
;
2744 int p_want_lose
, p_two_primaries
, cf
;
2745 char p_integrity_alg
[SHARED_SECRET_MAX
] = "";
2747 p_proto
= be32_to_cpu(p
->protocol
);
2748 p_after_sb_0p
= be32_to_cpu(p
->after_sb_0p
);
2749 p_after_sb_1p
= be32_to_cpu(p
->after_sb_1p
);
2750 p_after_sb_2p
= be32_to_cpu(p
->after_sb_2p
);
2751 p_two_primaries
= be32_to_cpu(p
->two_primaries
);
2752 cf
= be32_to_cpu(p
->conn_flags
);
2753 p_want_lose
= cf
& CF_WANT_LOSE
;
2755 clear_bit(CONN_DRY_RUN
, &mdev
->flags
);
2757 if (cf
& CF_DRY_RUN
)
2758 set_bit(CONN_DRY_RUN
, &mdev
->flags
);
2760 if (p_proto
!= mdev
->tconn
->net_conf
->wire_protocol
) {
2761 dev_err(DEV
, "incompatible communication protocols\n");
2765 if (cmp_after_sb(p_after_sb_0p
, mdev
->tconn
->net_conf
->after_sb_0p
)) {
2766 dev_err(DEV
, "incompatible after-sb-0pri settings\n");
2770 if (cmp_after_sb(p_after_sb_1p
, mdev
->tconn
->net_conf
->after_sb_1p
)) {
2771 dev_err(DEV
, "incompatible after-sb-1pri settings\n");
2775 if (cmp_after_sb(p_after_sb_2p
, mdev
->tconn
->net_conf
->after_sb_2p
)) {
2776 dev_err(DEV
, "incompatible after-sb-2pri settings\n");
2780 if (p_want_lose
&& mdev
->tconn
->net_conf
->want_lose
) {
2781 dev_err(DEV
, "both sides have the 'want_lose' flag set\n");
2785 if (p_two_primaries
!= mdev
->tconn
->net_conf
->two_primaries
) {
2786 dev_err(DEV
, "incompatible setting of the two-primaries options\n");
2790 if (mdev
->tconn
->agreed_pro_version
>= 87) {
2791 unsigned char *my_alg
= mdev
->tconn
->net_conf
->integrity_alg
;
2793 if (drbd_recv(mdev
, p_integrity_alg
, data_size
) != data_size
)
2796 p_integrity_alg
[SHARED_SECRET_MAX
-1] = 0;
2797 if (strcmp(p_integrity_alg
, my_alg
)) {
2798 dev_err(DEV
, "incompatible setting of the data-integrity-alg\n");
2801 dev_info(DEV
, "data-integrity-alg: %s\n",
2802 my_alg
[0] ? my_alg
: (unsigned char *)"<not-used>");
2808 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2813 * input: alg name, feature name
2814 * return: NULL (alg name was "")
2815 * ERR_PTR(error) if something goes wrong
2816 * or the crypto hash ptr, if it worked out ok. */
2817 struct crypto_hash
*drbd_crypto_alloc_digest_safe(const struct drbd_conf
*mdev
,
2818 const char *alg
, const char *name
)
2820 struct crypto_hash
*tfm
;
2825 tfm
= crypto_alloc_hash(alg
, 0, CRYPTO_ALG_ASYNC
);
2827 dev_err(DEV
, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2828 alg
, name
, PTR_ERR(tfm
));
2831 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm
))) {
2832 crypto_free_hash(tfm
);
2833 dev_err(DEV
, "\"%s\" is not a digest (%s)\n", alg
, name
);
2834 return ERR_PTR(-EINVAL
);
2839 static int receive_SyncParam(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
2840 unsigned int packet_size
)
2843 struct p_rs_param_95
*p
= &mdev
->tconn
->data
.rbuf
.rs_param_95
;
2844 unsigned int header_size
, data_size
, exp_max_sz
;
2845 struct crypto_hash
*verify_tfm
= NULL
;
2846 struct crypto_hash
*csums_tfm
= NULL
;
2847 const int apv
= mdev
->tconn
->agreed_pro_version
;
2848 int *rs_plan_s
= NULL
;
2851 exp_max_sz
= apv
<= 87 ? sizeof(struct p_rs_param
)
2852 : apv
== 88 ? sizeof(struct p_rs_param
)
2854 : apv
<= 94 ? sizeof(struct p_rs_param_89
)
2855 : /* apv >= 95 */ sizeof(struct p_rs_param_95
);
2857 if (packet_size
> exp_max_sz
) {
2858 dev_err(DEV
, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2859 packet_size
, exp_max_sz
);
2864 header_size
= sizeof(struct p_rs_param
) - sizeof(struct p_header
);
2865 data_size
= packet_size
- header_size
;
2866 } else if (apv
<= 94) {
2867 header_size
= sizeof(struct p_rs_param_89
) - sizeof(struct p_header
);
2868 data_size
= packet_size
- header_size
;
2869 D_ASSERT(data_size
== 0);
2871 header_size
= sizeof(struct p_rs_param_95
) - sizeof(struct p_header
);
2872 data_size
= packet_size
- header_size
;
2873 D_ASSERT(data_size
== 0);
2876 /* initialize verify_alg and csums_alg */
2877 memset(p
->verify_alg
, 0, 2 * SHARED_SECRET_MAX
);
2879 if (drbd_recv(mdev
, &p
->head
.payload
, header_size
) != header_size
)
2882 mdev
->sync_conf
.rate
= be32_to_cpu(p
->rate
);
2886 if (data_size
> SHARED_SECRET_MAX
) {
2887 dev_err(DEV
, "verify-alg too long, "
2888 "peer wants %u, accepting only %u byte\n",
2889 data_size
, SHARED_SECRET_MAX
);
2893 if (drbd_recv(mdev
, p
->verify_alg
, data_size
) != data_size
)
2896 /* we expect NUL terminated string */
2897 /* but just in case someone tries to be evil */
2898 D_ASSERT(p
->verify_alg
[data_size
-1] == 0);
2899 p
->verify_alg
[data_size
-1] = 0;
2901 } else /* apv >= 89 */ {
2902 /* we still expect NUL terminated strings */
2903 /* but just in case someone tries to be evil */
2904 D_ASSERT(p
->verify_alg
[SHARED_SECRET_MAX
-1] == 0);
2905 D_ASSERT(p
->csums_alg
[SHARED_SECRET_MAX
-1] == 0);
2906 p
->verify_alg
[SHARED_SECRET_MAX
-1] = 0;
2907 p
->csums_alg
[SHARED_SECRET_MAX
-1] = 0;
2910 if (strcmp(mdev
->sync_conf
.verify_alg
, p
->verify_alg
)) {
2911 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
2912 dev_err(DEV
, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2913 mdev
->sync_conf
.verify_alg
, p
->verify_alg
);
2916 verify_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
2917 p
->verify_alg
, "verify-alg");
2918 if (IS_ERR(verify_tfm
)) {
2924 if (apv
>= 89 && strcmp(mdev
->sync_conf
.csums_alg
, p
->csums_alg
)) {
2925 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
) {
2926 dev_err(DEV
, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2927 mdev
->sync_conf
.csums_alg
, p
->csums_alg
);
2930 csums_tfm
= drbd_crypto_alloc_digest_safe(mdev
,
2931 p
->csums_alg
, "csums-alg");
2932 if (IS_ERR(csums_tfm
)) {
2939 mdev
->sync_conf
.rate
= be32_to_cpu(p
->rate
);
2940 mdev
->sync_conf
.c_plan_ahead
= be32_to_cpu(p
->c_plan_ahead
);
2941 mdev
->sync_conf
.c_delay_target
= be32_to_cpu(p
->c_delay_target
);
2942 mdev
->sync_conf
.c_fill_target
= be32_to_cpu(p
->c_fill_target
);
2943 mdev
->sync_conf
.c_max_rate
= be32_to_cpu(p
->c_max_rate
);
2945 fifo_size
= (mdev
->sync_conf
.c_plan_ahead
* 10 * SLEEP_TIME
) / HZ
;
2946 if (fifo_size
!= mdev
->rs_plan_s
.size
&& fifo_size
> 0) {
2947 rs_plan_s
= kzalloc(sizeof(int) * fifo_size
, GFP_KERNEL
);
2949 dev_err(DEV
, "kmalloc of fifo_buffer failed");
2955 spin_lock(&mdev
->peer_seq_lock
);
2956 /* lock against drbd_nl_syncer_conf() */
2958 strcpy(mdev
->sync_conf
.verify_alg
, p
->verify_alg
);
2959 mdev
->sync_conf
.verify_alg_len
= strlen(p
->verify_alg
) + 1;
2960 crypto_free_hash(mdev
->verify_tfm
);
2961 mdev
->verify_tfm
= verify_tfm
;
2962 dev_info(DEV
, "using verify-alg: \"%s\"\n", p
->verify_alg
);
2965 strcpy(mdev
->sync_conf
.csums_alg
, p
->csums_alg
);
2966 mdev
->sync_conf
.csums_alg_len
= strlen(p
->csums_alg
) + 1;
2967 crypto_free_hash(mdev
->csums_tfm
);
2968 mdev
->csums_tfm
= csums_tfm
;
2969 dev_info(DEV
, "using csums-alg: \"%s\"\n", p
->csums_alg
);
2971 if (fifo_size
!= mdev
->rs_plan_s
.size
) {
2972 kfree(mdev
->rs_plan_s
.values
);
2973 mdev
->rs_plan_s
.values
= rs_plan_s
;
2974 mdev
->rs_plan_s
.size
= fifo_size
;
2975 mdev
->rs_planed
= 0;
2977 spin_unlock(&mdev
->peer_seq_lock
);
2982 /* just for completeness: actually not needed,
2983 * as this is not reached if csums_tfm was ok. */
2984 crypto_free_hash(csums_tfm
);
2985 /* but free the verify_tfm again, if csums_tfm did not work out */
2986 crypto_free_hash(verify_tfm
);
2987 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
2991 /* warn if the arguments differ by more than 12.5% */
2992 static void warn_if_differ_considerably(struct drbd_conf
*mdev
,
2993 const char *s
, sector_t a
, sector_t b
)
2996 if (a
== 0 || b
== 0)
2998 d
= (a
> b
) ? (a
- b
) : (b
- a
);
2999 if (d
> (a
>>3) || d
> (b
>>3))
3000 dev_warn(DEV
, "Considerable difference in %s: %llus vs. %llus\n", s
,
3001 (unsigned long long)a
, (unsigned long long)b
);
3004 static int receive_sizes(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3005 unsigned int data_size
)
3007 struct p_sizes
*p
= &mdev
->tconn
->data
.rbuf
.sizes
;
3008 enum determine_dev_size dd
= unchanged
;
3009 sector_t p_size
, p_usize
, my_usize
;
3010 int ldsc
= 0; /* local disk size changed */
3011 enum dds_flags ddsf
;
3013 p_size
= be64_to_cpu(p
->d_size
);
3014 p_usize
= be64_to_cpu(p
->u_size
);
3016 if (p_size
== 0 && mdev
->state
.disk
== D_DISKLESS
) {
3017 dev_err(DEV
, "some backing storage is needed\n");
3018 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3022 /* just store the peer's disk size for now.
3023 * we still need to figure out whether we accept that. */
3024 mdev
->p_size
= p_size
;
3026 if (get_ldev(mdev
)) {
3027 warn_if_differ_considerably(mdev
, "lower level device sizes",
3028 p_size
, drbd_get_max_capacity(mdev
->ldev
));
3029 warn_if_differ_considerably(mdev
, "user requested size",
3030 p_usize
, mdev
->ldev
->dc
.disk_size
);
3032 /* if this is the first connect, or an otherwise expected
3033 * param exchange, choose the minimum */
3034 if (mdev
->state
.conn
== C_WF_REPORT_PARAMS
)
3035 p_usize
= min_not_zero((sector_t
)mdev
->ldev
->dc
.disk_size
,
3038 my_usize
= mdev
->ldev
->dc
.disk_size
;
3040 if (mdev
->ldev
->dc
.disk_size
!= p_usize
) {
3041 mdev
->ldev
->dc
.disk_size
= p_usize
;
3042 dev_info(DEV
, "Peer sets u_size to %lu sectors\n",
3043 (unsigned long)mdev
->ldev
->dc
.disk_size
);
3046 /* Never shrink a device with usable data during connect.
3047 But allow online shrinking if we are connected. */
3048 if (drbd_new_dev_size(mdev
, mdev
->ldev
, 0) <
3049 drbd_get_capacity(mdev
->this_bdev
) &&
3050 mdev
->state
.disk
>= D_OUTDATED
&&
3051 mdev
->state
.conn
< C_CONNECTED
) {
3052 dev_err(DEV
, "The peer's disk size is too small!\n");
3053 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3054 mdev
->ldev
->dc
.disk_size
= my_usize
;
3061 ddsf
= be16_to_cpu(p
->dds_flags
);
3062 if (get_ldev(mdev
)) {
3063 dd
= drbd_determine_dev_size(mdev
, ddsf
);
3065 if (dd
== dev_size_error
)
3069 /* I am diskless, need to accept the peer's size. */
3070 drbd_set_my_capacity(mdev
, p_size
);
3073 mdev
->peer_max_bio_size
= be32_to_cpu(p
->max_bio_size
);
3074 drbd_reconsider_max_bio_size(mdev
);
3076 if (get_ldev(mdev
)) {
3077 if (mdev
->ldev
->known_size
!= drbd_get_capacity(mdev
->ldev
->backing_bdev
)) {
3078 mdev
->ldev
->known_size
= drbd_get_capacity(mdev
->ldev
->backing_bdev
);
3085 if (mdev
->state
.conn
> C_WF_REPORT_PARAMS
) {
3086 if (be64_to_cpu(p
->c_size
) !=
3087 drbd_get_capacity(mdev
->this_bdev
) || ldsc
) {
3088 /* we have different sizes, probably peer
3089 * needs to know my new size... */
3090 drbd_send_sizes(mdev
, 0, ddsf
);
3092 if (test_and_clear_bit(RESIZE_PENDING
, &mdev
->flags
) ||
3093 (dd
== grew
&& mdev
->state
.conn
== C_CONNECTED
)) {
3094 if (mdev
->state
.pdsk
>= D_INCONSISTENT
&&
3095 mdev
->state
.disk
>= D_INCONSISTENT
) {
3096 if (ddsf
& DDSF_NO_RESYNC
)
3097 dev_info(DEV
, "Resync of new storage suppressed with --assume-clean\n");
3099 resync_after_online_grow(mdev
);
3101 set_bit(RESYNC_AFTER_NEG
, &mdev
->flags
);
3108 static int receive_uuids(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3109 unsigned int data_size
)
3111 struct p_uuids
*p
= &mdev
->tconn
->data
.rbuf
.uuids
;
3113 int i
, updated_uuids
= 0;
3115 p_uuid
= kmalloc(sizeof(u64
)*UI_EXTENDED_SIZE
, GFP_NOIO
);
3117 for (i
= UI_CURRENT
; i
< UI_EXTENDED_SIZE
; i
++)
3118 p_uuid
[i
] = be64_to_cpu(p
->uuid
[i
]);
3120 kfree(mdev
->p_uuid
);
3121 mdev
->p_uuid
= p_uuid
;
3123 if (mdev
->state
.conn
< C_CONNECTED
&&
3124 mdev
->state
.disk
< D_INCONSISTENT
&&
3125 mdev
->state
.role
== R_PRIMARY
&&
3126 (mdev
->ed_uuid
& ~((u64
)1)) != (p_uuid
[UI_CURRENT
] & ~((u64
)1))) {
3127 dev_err(DEV
, "Can only connect to data with current UUID=%016llX\n",
3128 (unsigned long long)mdev
->ed_uuid
);
3129 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3133 if (get_ldev(mdev
)) {
3134 int skip_initial_sync
=
3135 mdev
->state
.conn
== C_CONNECTED
&&
3136 mdev
->tconn
->agreed_pro_version
>= 90 &&
3137 mdev
->ldev
->md
.uuid
[UI_CURRENT
] == UUID_JUST_CREATED
&&
3138 (p_uuid
[UI_FLAGS
] & 8);
3139 if (skip_initial_sync
) {
3140 dev_info(DEV
, "Accepted new current UUID, preparing to skip initial sync\n");
3141 drbd_bitmap_io(mdev
, &drbd_bmio_clear_n_write
,
3142 "clear_n_write from receive_uuids",
3143 BM_LOCKED_TEST_ALLOWED
);
3144 _drbd_uuid_set(mdev
, UI_CURRENT
, p_uuid
[UI_CURRENT
]);
3145 _drbd_uuid_set(mdev
, UI_BITMAP
, 0);
3146 _drbd_set_state(_NS2(mdev
, disk
, D_UP_TO_DATE
, pdsk
, D_UP_TO_DATE
),
3152 } else if (mdev
->state
.disk
< D_INCONSISTENT
&&
3153 mdev
->state
.role
== R_PRIMARY
) {
3154 /* I am a diskless primary, the peer just created a new current UUID
3156 updated_uuids
= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3159 /* Before we test for the disk state, we should wait until an eventually
3160 ongoing cluster wide state change is finished. That is important if
3161 we are primary and are detaching from our disk. We need to see the
3162 new disk state... */
3163 wait_event(mdev
->misc_wait
, !test_bit(CLUSTER_ST_CHANGE
, &mdev
->flags
));
3164 if (mdev
->state
.conn
>= C_CONNECTED
&& mdev
->state
.disk
< D_INCONSISTENT
)
3165 updated_uuids
|= drbd_set_ed_uuid(mdev
, p_uuid
[UI_CURRENT
]);
3168 drbd_print_uuids(mdev
, "receiver updated UUIDs to");
3174 * convert_state() - Converts the peer's view of the cluster state to our point of view
3175 * @ps: The state as seen by the peer.
3177 static union drbd_state
convert_state(union drbd_state ps
)
3179 union drbd_state ms
;
3181 static enum drbd_conns c_tab
[] = {
3182 [C_CONNECTED
] = C_CONNECTED
,
3184 [C_STARTING_SYNC_S
] = C_STARTING_SYNC_T
,
3185 [C_STARTING_SYNC_T
] = C_STARTING_SYNC_S
,
3186 [C_DISCONNECTING
] = C_TEAR_DOWN
, /* C_NETWORK_FAILURE, */
3187 [C_VERIFY_S
] = C_VERIFY_T
,
3193 ms
.conn
= c_tab
[ps
.conn
];
3198 ms
.peer_isp
= (ps
.aftr_isp
| ps
.user_isp
);
3203 static int receive_req_state(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3204 unsigned int data_size
)
3206 struct p_req_state
*p
= &mdev
->tconn
->data
.rbuf
.req_state
;
3207 union drbd_state mask
, val
;
3208 enum drbd_state_rv rv
;
3210 mask
.i
= be32_to_cpu(p
->mask
);
3211 val
.i
= be32_to_cpu(p
->val
);
3213 if (test_bit(DISCARD_CONCURRENT
, &mdev
->flags
) &&
3214 test_bit(CLUSTER_ST_CHANGE
, &mdev
->flags
)) {
3215 drbd_send_sr_reply(mdev
, SS_CONCURRENT_ST_CHG
);
3219 mask
= convert_state(mask
);
3220 val
= convert_state(val
);
3222 rv
= drbd_change_state(mdev
, CS_VERBOSE
, mask
, val
);
3224 drbd_send_sr_reply(mdev
, rv
);
3230 static int receive_state(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3231 unsigned int data_size
)
3233 struct p_state
*p
= &mdev
->tconn
->data
.rbuf
.state
;
3234 union drbd_state os
, ns
, peer_state
;
3235 enum drbd_disk_state real_peer_disk
;
3236 enum chg_state_flags cs_flags
;
3239 peer_state
.i
= be32_to_cpu(p
->state
);
3241 real_peer_disk
= peer_state
.disk
;
3242 if (peer_state
.disk
== D_NEGOTIATING
) {
3243 real_peer_disk
= mdev
->p_uuid
[UI_FLAGS
] & 4 ? D_INCONSISTENT
: D_CONSISTENT
;
3244 dev_info(DEV
, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk
));
3247 spin_lock_irq(&mdev
->tconn
->req_lock
);
3249 os
= ns
= mdev
->state
;
3250 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3252 /* peer says his disk is uptodate, while we think it is inconsistent,
3253 * and this happens while we think we have a sync going on. */
3254 if (os
.pdsk
== D_INCONSISTENT
&& real_peer_disk
== D_UP_TO_DATE
&&
3255 os
.conn
> C_CONNECTED
&& os
.disk
== D_UP_TO_DATE
) {
3256 /* If we are (becoming) SyncSource, but peer is still in sync
3257 * preparation, ignore its uptodate-ness to avoid flapping, it
3258 * will change to inconsistent once the peer reaches active
3260 * It may have changed syncer-paused flags, however, so we
3261 * cannot ignore this completely. */
3262 if (peer_state
.conn
> C_CONNECTED
&&
3263 peer_state
.conn
< C_SYNC_SOURCE
)
3264 real_peer_disk
= D_INCONSISTENT
;
3266 /* if peer_state changes to connected at the same time,
3267 * it explicitly notifies us that it finished resync.
3268 * Maybe we should finish it up, too? */
3269 else if (os
.conn
>= C_SYNC_SOURCE
&&
3270 peer_state
.conn
== C_CONNECTED
) {
3271 if (drbd_bm_total_weight(mdev
) <= mdev
->rs_failed
)
3272 drbd_resync_finished(mdev
);
3277 /* peer says his disk is inconsistent, while we think it is uptodate,
3278 * and this happens while the peer still thinks we have a sync going on,
3279 * but we think we are already done with the sync.
3280 * We ignore this to avoid flapping pdsk.
3281 * This should not happen, if the peer is a recent version of drbd. */
3282 if (os
.pdsk
== D_UP_TO_DATE
&& real_peer_disk
== D_INCONSISTENT
&&
3283 os
.conn
== C_CONNECTED
&& peer_state
.conn
> C_SYNC_SOURCE
)
3284 real_peer_disk
= D_UP_TO_DATE
;
3286 if (ns
.conn
== C_WF_REPORT_PARAMS
)
3287 ns
.conn
= C_CONNECTED
;
3289 if (peer_state
.conn
== C_AHEAD
)
3292 if (mdev
->p_uuid
&& peer_state
.disk
>= D_NEGOTIATING
&&
3293 get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3294 int cr
; /* consider resync */
3296 /* if we established a new connection */
3297 cr
= (os
.conn
< C_CONNECTED
);
3298 /* if we had an established connection
3299 * and one of the nodes newly attaches a disk */
3300 cr
|= (os
.conn
== C_CONNECTED
&&
3301 (peer_state
.disk
== D_NEGOTIATING
||
3302 os
.disk
== D_NEGOTIATING
));
3303 /* if we have both been inconsistent, and the peer has been
3304 * forced to be UpToDate with --overwrite-data */
3305 cr
|= test_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3306 /* if we had been plain connected, and the admin requested to
3307 * start a sync by "invalidate" or "invalidate-remote" */
3308 cr
|= (os
.conn
== C_CONNECTED
&&
3309 (peer_state
.conn
>= C_STARTING_SYNC_S
&&
3310 peer_state
.conn
<= C_WF_BITMAP_T
));
3313 ns
.conn
= drbd_sync_handshake(mdev
, peer_state
.role
, real_peer_disk
);
3316 if (ns
.conn
== C_MASK
) {
3317 ns
.conn
= C_CONNECTED
;
3318 if (mdev
->state
.disk
== D_NEGOTIATING
) {
3319 drbd_force_state(mdev
, NS(disk
, D_FAILED
));
3320 } else if (peer_state
.disk
== D_NEGOTIATING
) {
3321 dev_err(DEV
, "Disk attach process on the peer node was aborted.\n");
3322 peer_state
.disk
= D_DISKLESS
;
3323 real_peer_disk
= D_DISKLESS
;
3325 if (test_and_clear_bit(CONN_DRY_RUN
, &mdev
->flags
))
3327 D_ASSERT(os
.conn
== C_WF_REPORT_PARAMS
);
3328 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3334 spin_lock_irq(&mdev
->tconn
->req_lock
);
3335 if (mdev
->state
.i
!= os
.i
)
3337 clear_bit(CONSIDER_RESYNC
, &mdev
->flags
);
3338 ns
.peer
= peer_state
.role
;
3339 ns
.pdsk
= real_peer_disk
;
3340 ns
.peer_isp
= (peer_state
.aftr_isp
| peer_state
.user_isp
);
3341 if ((ns
.conn
== C_CONNECTED
|| ns
.conn
== C_WF_BITMAP_S
) && ns
.disk
== D_NEGOTIATING
)
3342 ns
.disk
= mdev
->new_state_tmp
.disk
;
3343 cs_flags
= CS_VERBOSE
+ (os
.conn
< C_CONNECTED
&& ns
.conn
>= C_CONNECTED
? 0 : CS_HARD
);
3344 if (ns
.pdsk
== D_CONSISTENT
&& is_susp(ns
) && ns
.conn
== C_CONNECTED
&& os
.conn
< C_CONNECTED
&&
3345 test_bit(NEW_CUR_UUID
, &mdev
->flags
)) {
3346 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
3347 for temporal network outages! */
3348 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3349 dev_err(DEV
, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
3351 drbd_uuid_new_current(mdev
);
3352 clear_bit(NEW_CUR_UUID
, &mdev
->flags
);
3353 drbd_force_state(mdev
, NS2(conn
, C_PROTOCOL_ERROR
, susp
, 0));
3356 rv
= _drbd_set_state(mdev
, ns
, cs_flags
, NULL
);
3358 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3360 if (rv
< SS_SUCCESS
) {
3361 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
3365 if (os
.conn
> C_WF_REPORT_PARAMS
) {
3366 if (ns
.conn
> C_CONNECTED
&& peer_state
.conn
<= C_CONNECTED
&&
3367 peer_state
.disk
!= D_NEGOTIATING
) {
3368 /* we want resync, peer has not yet decided to sync... */
3369 /* Nowadays only used when forcing a node into primary role and
3370 setting its disk to UpToDate with that */
3371 drbd_send_uuids(mdev
);
3372 drbd_send_state(mdev
);
3376 mdev
->tconn
->net_conf
->want_lose
= 0;
3378 drbd_md_sync(mdev
); /* update connected indicator, la_size, ... */
3383 static int receive_sync_uuid(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3384 unsigned int data_size
)
3386 struct p_rs_uuid
*p
= &mdev
->tconn
->data
.rbuf
.rs_uuid
;
3388 wait_event(mdev
->misc_wait
,
3389 mdev
->state
.conn
== C_WF_SYNC_UUID
||
3390 mdev
->state
.conn
== C_BEHIND
||
3391 mdev
->state
.conn
< C_CONNECTED
||
3392 mdev
->state
.disk
< D_NEGOTIATING
);
3394 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3396 /* Here the _drbd_uuid_ functions are right, current should
3397 _not_ be rotated into the history */
3398 if (get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
3399 _drbd_uuid_set(mdev
, UI_CURRENT
, be64_to_cpu(p
->uuid
));
3400 _drbd_uuid_set(mdev
, UI_BITMAP
, 0UL);
3402 drbd_print_uuids(mdev
, "updated sync uuid");
3403 drbd_start_resync(mdev
, C_SYNC_TARGET
);
3407 dev_err(DEV
, "Ignoring SyncUUID packet!\n");
3413 * receive_bitmap_plain
3415 * Return 0 when done, 1 when another iteration is needed, and a negative error
3416 * code upon failure.
3419 receive_bitmap_plain(struct drbd_conf
*mdev
, unsigned int data_size
,
3420 unsigned long *buffer
, struct bm_xfer_ctx
*c
)
3422 unsigned num_words
= min_t(size_t, BM_PACKET_WORDS
, c
->bm_words
- c
->word_offset
);
3423 unsigned want
= num_words
* sizeof(long);
3426 if (want
!= data_size
) {
3427 dev_err(DEV
, "%s:want (%u) != data_size (%u)\n", __func__
, want
, data_size
);
3432 err
= drbd_recv(mdev
, buffer
, want
);
3439 drbd_bm_merge_lel(mdev
, c
->word_offset
, num_words
, buffer
);
3441 c
->word_offset
+= num_words
;
3442 c
->bit_offset
= c
->word_offset
* BITS_PER_LONG
;
3443 if (c
->bit_offset
> c
->bm_bits
)
3444 c
->bit_offset
= c
->bm_bits
;
3452 * Return 0 when done, 1 when another iteration is needed, and a negative error
3453 * code upon failure.
3456 recv_bm_rle_bits(struct drbd_conf
*mdev
,
3457 struct p_compressed_bm
*p
,
3458 struct bm_xfer_ctx
*c
,
3461 struct bitstream bs
;
3465 unsigned long s
= c
->bit_offset
;
3467 int toggle
= DCBP_get_start(p
);
3471 bitstream_init(&bs
, p
->code
, len
, DCBP_get_pad_bits(p
));
3473 bits
= bitstream_get_bits(&bs
, &look_ahead
, 64);
3477 for (have
= bits
; have
> 0; s
+= rl
, toggle
= !toggle
) {
3478 bits
= vli_decode_bits(&rl
, look_ahead
);
3484 if (e
>= c
->bm_bits
) {
3485 dev_err(DEV
, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e
);
3488 _drbd_bm_set_bits(mdev
, s
, e
);
3492 dev_err(DEV
, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3493 have
, bits
, look_ahead
,
3494 (unsigned int)(bs
.cur
.b
- p
->code
),
3495 (unsigned int)bs
.buf_len
);
3498 look_ahead
>>= bits
;
3501 bits
= bitstream_get_bits(&bs
, &tmp
, 64 - have
);
3504 look_ahead
|= tmp
<< have
;
3509 bm_xfer_ctx_bit_to_word_offset(c
);
3511 return (s
!= c
->bm_bits
);
3517 * Return 0 when done, 1 when another iteration is needed, and a negative error
3518 * code upon failure.
3521 decode_bitmap_c(struct drbd_conf
*mdev
,
3522 struct p_compressed_bm
*p
,
3523 struct bm_xfer_ctx
*c
,
3526 if (DCBP_get_code(p
) == RLE_VLI_Bits
)
3527 return recv_bm_rle_bits(mdev
, p
, c
, len
);
3529 /* other variants had been implemented for evaluation,
3530 * but have been dropped as this one turned out to be "best"
3531 * during all our tests. */
3533 dev_err(DEV
, "receive_bitmap_c: unknown encoding %u\n", p
->encoding
);
3534 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3538 void INFO_bm_xfer_stats(struct drbd_conf
*mdev
,
3539 const char *direction
, struct bm_xfer_ctx
*c
)
3541 /* what would it take to transfer it "plaintext" */
3542 unsigned plain
= sizeof(struct p_header
) *
3543 ((c
->bm_words
+BM_PACKET_WORDS
-1)/BM_PACKET_WORDS
+1)
3544 + c
->bm_words
* sizeof(long);
3545 unsigned total
= c
->bytes
[0] + c
->bytes
[1];
3548 /* total can not be zero. but just in case: */
3552 /* don't report if not compressed */
3556 /* total < plain. check for overflow, still */
3557 r
= (total
> UINT_MAX
/1000) ? (total
/ (plain
/1000))
3558 : (1000 * total
/ plain
);
3564 dev_info(DEV
, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3565 "total %u; compression: %u.%u%%\n",
3567 c
->bytes
[1], c
->packets
[1],
3568 c
->bytes
[0], c
->packets
[0],
3569 total
, r
/10, r
% 10);
3572 /* Since we are processing the bitfield from lower addresses to higher,
3573 it does not matter if the process it in 32 bit chunks or 64 bit
3574 chunks as long as it is little endian. (Understand it as byte stream,
3575 beginning with the lowest byte...) If we would use big endian
3576 we would need to process it from the highest address to the lowest,
3577 in order to be agnostic to the 32 vs 64 bits issue.
3579 returns 0 on failure, 1 if we successfully received it. */
3580 static int receive_bitmap(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3581 unsigned int data_size
)
3583 struct bm_xfer_ctx c
;
3587 struct p_header
*h
= &mdev
->tconn
->data
.rbuf
.header
;
3589 drbd_bm_lock(mdev
, "receive bitmap", BM_LOCKED_SET_ALLOWED
);
3590 /* you are supposed to send additional out-of-sync information
3591 * if you actually set bits during this phase */
3593 /* maybe we should use some per thread scratch page,
3594 * and allocate that during initial device creation? */
3595 buffer
= (unsigned long *) __get_free_page(GFP_NOIO
);
3597 dev_err(DEV
, "failed to allocate one page buffer in %s\n", __func__
);
3601 c
= (struct bm_xfer_ctx
) {
3602 .bm_bits
= drbd_bm_bits(mdev
),
3603 .bm_words
= drbd_bm_words(mdev
),
3607 if (cmd
== P_BITMAP
) {
3608 err
= receive_bitmap_plain(mdev
, data_size
, buffer
, &c
);
3609 } else if (cmd
== P_COMPRESSED_BITMAP
) {
3610 /* MAYBE: sanity check that we speak proto >= 90,
3611 * and the feature is enabled! */
3612 struct p_compressed_bm
*p
;
3614 if (data_size
> BM_PACKET_PAYLOAD_BYTES
) {
3615 dev_err(DEV
, "ReportCBitmap packet too large\n");
3618 /* use the page buff */
3620 memcpy(p
, h
, sizeof(*h
));
3621 if (drbd_recv(mdev
, p
->head
.payload
, data_size
) != data_size
)
3623 if (data_size
<= (sizeof(*p
) - sizeof(p
->head
))) {
3624 dev_err(DEV
, "ReportCBitmap packet too small (l:%u)\n", data_size
);
3627 err
= decode_bitmap_c(mdev
, p
, &c
, data_size
);
3629 dev_warn(DEV
, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", cmd
);
3633 c
.packets
[cmd
== P_BITMAP
]++;
3634 c
.bytes
[cmd
== P_BITMAP
] += sizeof(struct p_header
) + data_size
;
3641 if (!drbd_recv_header(mdev
, &cmd
, &data_size
))
3645 INFO_bm_xfer_stats(mdev
, "receive", &c
);
3647 if (mdev
->state
.conn
== C_WF_BITMAP_T
) {
3648 enum drbd_state_rv rv
;
3650 ok
= !drbd_send_bitmap(mdev
);
3653 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
3654 rv
= _drbd_request_state(mdev
, NS(conn
, C_WF_SYNC_UUID
), CS_VERBOSE
);
3655 D_ASSERT(rv
== SS_SUCCESS
);
3656 } else if (mdev
->state
.conn
!= C_WF_BITMAP_S
) {
3657 /* admin may have requested C_DISCONNECTING,
3658 * other threads may have noticed network errors */
3659 dev_info(DEV
, "unexpected cstate (%s) in receive_bitmap\n",
3660 drbd_conn_str(mdev
->state
.conn
));
3665 drbd_bm_unlock(mdev
);
3666 if (ok
&& mdev
->state
.conn
== C_WF_BITMAP_S
)
3667 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
3668 free_page((unsigned long) buffer
);
3672 static int receive_skip(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3673 unsigned int data_size
)
3675 /* TODO zero copy sink :) */
3676 static char sink
[128];
3679 dev_warn(DEV
, "skipping unknown optional packet type %d, l: %d!\n",
3684 want
= min_t(int, size
, sizeof(sink
));
3685 r
= drbd_recv(mdev
, sink
, want
);
3693 static int receive_UnplugRemote(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3694 unsigned int data_size
)
3696 /* Make sure we've acked all the TCP data associated
3697 * with the data requests being unplugged */
3698 drbd_tcp_quickack(mdev
->tconn
->data
.socket
);
3703 static int receive_out_of_sync(struct drbd_conf
*mdev
, enum drbd_packet cmd
,
3704 unsigned int data_size
)
3706 struct p_block_desc
*p
= &mdev
->tconn
->data
.rbuf
.block_desc
;
3708 switch (mdev
->state
.conn
) {
3709 case C_WF_SYNC_UUID
:
3714 dev_err(DEV
, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
3715 drbd_conn_str(mdev
->state
.conn
));
3718 drbd_set_out_of_sync(mdev
, be64_to_cpu(p
->sector
), be32_to_cpu(p
->blksize
));
3723 typedef int (*drbd_cmd_handler_f
)(struct drbd_conf
*, enum drbd_packet cmd
,
3724 unsigned int to_receive
);
3729 drbd_cmd_handler_f function
;
3732 static struct data_cmd drbd_cmd_handler
[] = {
3733 [P_DATA
] = { 1, sizeof(struct p_data
), receive_Data
},
3734 [P_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_DataReply
},
3735 [P_RS_DATA_REPLY
] = { 1, sizeof(struct p_data
), receive_RSDataReply
} ,
3736 [P_BARRIER
] = { 0, sizeof(struct p_barrier
), receive_Barrier
} ,
3737 [P_BITMAP
] = { 1, sizeof(struct p_header
), receive_bitmap
} ,
3738 [P_COMPRESSED_BITMAP
] = { 1, sizeof(struct p_header
), receive_bitmap
} ,
3739 [P_UNPLUG_REMOTE
] = { 0, sizeof(struct p_header
), receive_UnplugRemote
},
3740 [P_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
3741 [P_RS_DATA_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
3742 [P_SYNC_PARAM
] = { 1, sizeof(struct p_header
), receive_SyncParam
},
3743 [P_SYNC_PARAM89
] = { 1, sizeof(struct p_header
), receive_SyncParam
},
3744 [P_PROTOCOL
] = { 1, sizeof(struct p_protocol
), receive_protocol
},
3745 [P_UUIDS
] = { 0, sizeof(struct p_uuids
), receive_uuids
},
3746 [P_SIZES
] = { 0, sizeof(struct p_sizes
), receive_sizes
},
3747 [P_STATE
] = { 0, sizeof(struct p_state
), receive_state
},
3748 [P_STATE_CHG_REQ
] = { 0, sizeof(struct p_req_state
), receive_req_state
},
3749 [P_SYNC_UUID
] = { 0, sizeof(struct p_rs_uuid
), receive_sync_uuid
},
3750 [P_OV_REQUEST
] = { 0, sizeof(struct p_block_req
), receive_DataRequest
},
3751 [P_OV_REPLY
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
3752 [P_CSUM_RS_REQUEST
] = { 1, sizeof(struct p_block_req
), receive_DataRequest
},
3753 [P_DELAY_PROBE
] = { 0, sizeof(struct p_delay_probe93
), receive_skip
},
3754 [P_OUT_OF_SYNC
] = { 0, sizeof(struct p_block_desc
), receive_out_of_sync
},
3755 /* anything missing from this table is in
3756 * the asender_tbl, see get_asender_cmd */
3757 [P_MAX_CMD
] = { 0, 0, NULL
},
3760 /* All handler functions that expect a sub-header get that sub-heder in
3761 mdev->tconn->data.rbuf.header.head.payload.
3763 Usually in mdev->tconn->data.rbuf.header.head the callback can find the usual
3764 p_header, but they may not rely on that. Since there is also p_header95 !
3767 static void drbdd(struct drbd_conf
*mdev
)
3769 struct p_header
*header
= &mdev
->tconn
->data
.rbuf
.header
;
3770 unsigned int packet_size
;
3771 enum drbd_packet cmd
;
3772 size_t shs
; /* sub header size */
3775 while (get_t_state(&mdev
->tconn
->receiver
) == RUNNING
) {
3776 drbd_thread_current_set_cpu(mdev
);
3777 if (!drbd_recv_header(mdev
, &cmd
, &packet_size
))
3780 if (unlikely(cmd
>= P_MAX_CMD
|| !drbd_cmd_handler
[cmd
].function
)) {
3781 dev_err(DEV
, "unknown packet type %d, l: %d!\n", cmd
, packet_size
);
3785 shs
= drbd_cmd_handler
[cmd
].pkt_size
- sizeof(struct p_header
);
3786 if (packet_size
- shs
> 0 && !drbd_cmd_handler
[cmd
].expect_payload
) {
3787 dev_err(DEV
, "No payload expected %s l:%d\n", cmdname(cmd
), packet_size
);
3792 rv
= drbd_recv(mdev
, &header
->payload
, shs
);
3793 if (unlikely(rv
!= shs
)) {
3794 if (!signal_pending(current
))
3795 dev_warn(DEV
, "short read while reading sub header: rv=%d\n", rv
);
3800 rv
= drbd_cmd_handler
[cmd
].function(mdev
, cmd
, packet_size
- shs
);
3802 if (unlikely(!rv
)) {
3803 dev_err(DEV
, "error receiving %s, l: %d!\n",
3804 cmdname(cmd
), packet_size
);
3811 drbd_force_state(mdev
, NS(conn
, C_PROTOCOL_ERROR
));
3813 /* If we leave here, we probably want to update at least the
3814 * "Connected" indicator on stable storage. Do so explicitly here. */
3818 void drbd_flush_workqueue(struct drbd_tconn
*tconn
)
3820 struct drbd_wq_barrier barr
;
3822 barr
.w
.cb
= w_prev_work_done
;
3823 init_completion(&barr
.done
);
3824 drbd_queue_work(&tconn
->data
.work
, &barr
.w
);
3825 wait_for_completion(&barr
.done
);
3828 static void drbd_disconnect(struct drbd_conf
*mdev
)
3830 enum drbd_fencing_p fp
;
3831 union drbd_state os
, ns
;
3832 int rv
= SS_UNKNOWN_ERROR
;
3835 if (mdev
->state
.conn
== C_STANDALONE
)
3838 /* asender does not clean up anything. it must not interfere, either */
3839 drbd_thread_stop(&mdev
->tconn
->asender
);
3840 drbd_free_sock(mdev
);
3842 /* wait for current activity to cease. */
3843 spin_lock_irq(&mdev
->tconn
->req_lock
);
3844 _drbd_wait_ee_list_empty(mdev
, &mdev
->active_ee
);
3845 _drbd_wait_ee_list_empty(mdev
, &mdev
->sync_ee
);
3846 _drbd_wait_ee_list_empty(mdev
, &mdev
->read_ee
);
3847 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3849 /* We do not have data structures that would allow us to
3850 * get the rs_pending_cnt down to 0 again.
3851 * * On C_SYNC_TARGET we do not have any data structures describing
3852 * the pending RSDataRequest's we have sent.
3853 * * On C_SYNC_SOURCE there is no data structure that tracks
3854 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
3855 * And no, it is not the sum of the reference counts in the
3856 * resync_LRU. The resync_LRU tracks the whole operation including
3857 * the disk-IO, while the rs_pending_cnt only tracks the blocks
3859 drbd_rs_cancel_all(mdev
);
3861 mdev
->rs_failed
= 0;
3862 atomic_set(&mdev
->rs_pending_cnt
, 0);
3863 wake_up(&mdev
->misc_wait
);
3865 del_timer(&mdev
->request_timer
);
3867 /* make sure syncer is stopped and w_resume_next_sg queued */
3868 del_timer_sync(&mdev
->resync_timer
);
3869 resync_timer_fn((unsigned long)mdev
);
3871 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
3872 * w_make_resync_request etc. which may still be on the worker queue
3873 * to be "canceled" */
3874 drbd_flush_workqueue(mdev
->tconn
);
3876 /* This also does reclaim_net_ee(). If we do this too early, we might
3877 * miss some resync ee and pages.*/
3878 drbd_process_done_ee(mdev
);
3880 kfree(mdev
->p_uuid
);
3881 mdev
->p_uuid
= NULL
;
3883 if (!is_susp(mdev
->state
))
3886 dev_info(DEV
, "Connection closed\n");
3891 if (get_ldev(mdev
)) {
3892 fp
= mdev
->ldev
->dc
.fencing
;
3896 if (mdev
->state
.role
== R_PRIMARY
&& fp
>= FP_RESOURCE
&& mdev
->state
.pdsk
>= D_UNKNOWN
)
3897 drbd_try_outdate_peer_async(mdev
);
3899 spin_lock_irq(&mdev
->tconn
->req_lock
);
3901 if (os
.conn
>= C_UNCONNECTED
) {
3902 /* Do not restart in case we are C_DISCONNECTING */
3904 ns
.conn
= C_UNCONNECTED
;
3905 rv
= _drbd_set_state(mdev
, ns
, CS_VERBOSE
, NULL
);
3907 spin_unlock_irq(&mdev
->tconn
->req_lock
);
3909 if (os
.conn
== C_DISCONNECTING
) {
3910 wait_event(mdev
->tconn
->net_cnt_wait
, atomic_read(&mdev
->tconn
->net_cnt
) == 0);
3912 crypto_free_hash(mdev
->tconn
->cram_hmac_tfm
);
3913 mdev
->tconn
->cram_hmac_tfm
= NULL
;
3915 kfree(mdev
->tconn
->net_conf
);
3916 mdev
->tconn
->net_conf
= NULL
;
3917 drbd_request_state(mdev
, NS(conn
, C_STANDALONE
));
3920 /* serialize with bitmap writeout triggered by the state change,
3922 wait_event(mdev
->misc_wait
, !test_bit(BITMAP_IO
, &mdev
->flags
));
3924 /* tcp_close and release of sendpage pages can be deferred. I don't
3925 * want to use SO_LINGER, because apparently it can be deferred for
3926 * more than 20 seconds (longest time I checked).
3928 * Actually we don't care for exactly when the network stack does its
3929 * put_page(), but release our reference on these pages right here.
3931 i
= drbd_release_ee(mdev
, &mdev
->net_ee
);
3933 dev_info(DEV
, "net_ee not empty, killed %u entries\n", i
);
3934 i
= atomic_read(&mdev
->pp_in_use_by_net
);
3936 dev_info(DEV
, "pp_in_use_by_net = %d, expected 0\n", i
);
3937 i
= atomic_read(&mdev
->pp_in_use
);
3939 dev_info(DEV
, "pp_in_use = %d, expected 0\n", i
);
3941 D_ASSERT(list_empty(&mdev
->read_ee
));
3942 D_ASSERT(list_empty(&mdev
->active_ee
));
3943 D_ASSERT(list_empty(&mdev
->sync_ee
));
3944 D_ASSERT(list_empty(&mdev
->done_ee
));
3946 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
3947 atomic_set(&mdev
->current_epoch
->epoch_size
, 0);
3948 D_ASSERT(list_empty(&mdev
->current_epoch
->list
));
3952 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
3953 * we can agree on is stored in agreed_pro_version.
3955 * feature flags and the reserved array should be enough room for future
3956 * enhancements of the handshake protocol, and possible plugins...
3958 * for now, they are expected to be zero, but ignored.
3960 static int drbd_send_handshake(struct drbd_conf
*mdev
)
3962 /* ASSERT current == mdev->tconn->receiver ... */
3963 struct p_handshake
*p
= &mdev
->tconn
->data
.sbuf
.handshake
;
3966 if (mutex_lock_interruptible(&mdev
->tconn
->data
.mutex
)) {
3967 dev_err(DEV
, "interrupted during initial handshake\n");
3968 return 0; /* interrupted. not ok. */
3971 if (mdev
->tconn
->data
.socket
== NULL
) {
3972 mutex_unlock(&mdev
->tconn
->data
.mutex
);
3976 memset(p
, 0, sizeof(*p
));
3977 p
->protocol_min
= cpu_to_be32(PRO_VERSION_MIN
);
3978 p
->protocol_max
= cpu_to_be32(PRO_VERSION_MAX
);
3979 ok
= _drbd_send_cmd(mdev
, mdev
->tconn
->data
.socket
, P_HAND_SHAKE
,
3980 &p
->head
, sizeof(*p
), 0 );
3981 mutex_unlock(&mdev
->tconn
->data
.mutex
);
3987 * 1 yes, we have a valid connection
3988 * 0 oops, did not work out, please try again
3989 * -1 peer talks different language,
3990 * no point in trying again, please go standalone.
3992 static int drbd_do_handshake(struct drbd_conf
*mdev
)
3994 /* ASSERT current == mdev->tconn->receiver ... */
3995 struct p_handshake
*p
= &mdev
->tconn
->data
.rbuf
.handshake
;
3996 const int expect
= sizeof(struct p_handshake
) - sizeof(struct p_header80
);
3997 unsigned int length
;
3998 enum drbd_packet cmd
;
4001 rv
= drbd_send_handshake(mdev
);
4005 rv
= drbd_recv_header(mdev
, &cmd
, &length
);
4009 if (cmd
!= P_HAND_SHAKE
) {
4010 dev_err(DEV
, "expected HandShake packet, received: %s (0x%04x)\n",
4015 if (length
!= expect
) {
4016 dev_err(DEV
, "expected HandShake length: %u, received: %u\n",
4021 rv
= drbd_recv(mdev
, &p
->head
.payload
, expect
);
4024 if (!signal_pending(current
))
4025 dev_warn(DEV
, "short read receiving handshake packet: l=%u\n", rv
);
4029 p
->protocol_min
= be32_to_cpu(p
->protocol_min
);
4030 p
->protocol_max
= be32_to_cpu(p
->protocol_max
);
4031 if (p
->protocol_max
== 0)
4032 p
->protocol_max
= p
->protocol_min
;
4034 if (PRO_VERSION_MAX
< p
->protocol_min
||
4035 PRO_VERSION_MIN
> p
->protocol_max
)
4038 mdev
->tconn
->agreed_pro_version
= min_t(int, PRO_VERSION_MAX
, p
->protocol_max
);
4040 dev_info(DEV
, "Handshake successful: "
4041 "Agreed network protocol version %d\n", mdev
->tconn
->agreed_pro_version
);
4046 dev_err(DEV
, "incompatible DRBD dialects: "
4047 "I support %d-%d, peer supports %d-%d\n",
4048 PRO_VERSION_MIN
, PRO_VERSION_MAX
,
4049 p
->protocol_min
, p
->protocol_max
);
4053 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4054 static int drbd_do_auth(struct drbd_conf
*mdev
)
4056 dev_err(DEV
, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4057 dev_err(DEV
, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4061 #define CHALLENGE_LEN 64
4065 0 - failed, try again (network error),
4066 -1 - auth failed, don't try again.
4069 static int drbd_do_auth(struct drbd_conf
*mdev
)
4071 char my_challenge
[CHALLENGE_LEN
]; /* 64 Bytes... */
4072 struct scatterlist sg
;
4073 char *response
= NULL
;
4074 char *right_response
= NULL
;
4075 char *peers_ch
= NULL
;
4076 unsigned int key_len
= strlen(mdev
->tconn
->net_conf
->shared_secret
);
4077 unsigned int resp_size
;
4078 struct hash_desc desc
;
4079 enum drbd_packet cmd
;
4080 unsigned int length
;
4083 desc
.tfm
= mdev
->tconn
->cram_hmac_tfm
;
4086 rv
= crypto_hash_setkey(mdev
->tconn
->cram_hmac_tfm
,
4087 (u8
*)mdev
->tconn
->net_conf
->shared_secret
, key_len
);
4089 dev_err(DEV
, "crypto_hash_setkey() failed with %d\n", rv
);
4094 get_random_bytes(my_challenge
, CHALLENGE_LEN
);
4096 rv
= drbd_send_cmd2(mdev
, P_AUTH_CHALLENGE
, my_challenge
, CHALLENGE_LEN
);
4100 rv
= drbd_recv_header(mdev
, &cmd
, &length
);
4104 if (cmd
!= P_AUTH_CHALLENGE
) {
4105 dev_err(DEV
, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4111 if (length
> CHALLENGE_LEN
* 2) {
4112 dev_err(DEV
, "expected AuthChallenge payload too big.\n");
4117 peers_ch
= kmalloc(length
, GFP_NOIO
);
4118 if (peers_ch
== NULL
) {
4119 dev_err(DEV
, "kmalloc of peers_ch failed\n");
4124 rv
= drbd_recv(mdev
, peers_ch
, length
);
4127 if (!signal_pending(current
))
4128 dev_warn(DEV
, "short read AuthChallenge: l=%u\n", rv
);
4133 resp_size
= crypto_hash_digestsize(mdev
->tconn
->cram_hmac_tfm
);
4134 response
= kmalloc(resp_size
, GFP_NOIO
);
4135 if (response
== NULL
) {
4136 dev_err(DEV
, "kmalloc of response failed\n");
4141 sg_init_table(&sg
, 1);
4142 sg_set_buf(&sg
, peers_ch
, length
);
4144 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, response
);
4146 dev_err(DEV
, "crypto_hash_digest() failed with %d\n", rv
);
4151 rv
= drbd_send_cmd2(mdev
, P_AUTH_RESPONSE
, response
, resp_size
);
4155 rv
= drbd_recv_header(mdev
, &cmd
, &length
);
4159 if (cmd
!= P_AUTH_RESPONSE
) {
4160 dev_err(DEV
, "expected AuthResponse packet, received: %s (0x%04x)\n",
4166 if (length
!= resp_size
) {
4167 dev_err(DEV
, "expected AuthResponse payload of wrong size\n");
4172 rv
= drbd_recv(mdev
, response
, resp_size
);
4174 if (rv
!= resp_size
) {
4175 if (!signal_pending(current
))
4176 dev_warn(DEV
, "short read receiving AuthResponse: l=%u\n", rv
);
4181 right_response
= kmalloc(resp_size
, GFP_NOIO
);
4182 if (right_response
== NULL
) {
4183 dev_err(DEV
, "kmalloc of right_response failed\n");
4188 sg_set_buf(&sg
, my_challenge
, CHALLENGE_LEN
);
4190 rv
= crypto_hash_digest(&desc
, &sg
, sg
.length
, right_response
);
4192 dev_err(DEV
, "crypto_hash_digest() failed with %d\n", rv
);
4197 rv
= !memcmp(response
, right_response
, resp_size
);
4200 dev_info(DEV
, "Peer authenticated using %d bytes of '%s' HMAC\n",
4201 resp_size
, mdev
->tconn
->net_conf
->cram_hmac_alg
);
4208 kfree(right_response
);
4214 int drbdd_init(struct drbd_thread
*thi
)
4216 struct drbd_conf
*mdev
= thi
->mdev
;
4217 unsigned int minor
= mdev_to_minor(mdev
);
4220 sprintf(current
->comm
, "drbd%d_receiver", minor
);
4222 dev_info(DEV
, "receiver (re)started\n");
4225 h
= drbd_connect(mdev
);
4227 drbd_disconnect(mdev
);
4228 schedule_timeout_interruptible(HZ
);
4231 dev_warn(DEV
, "Discarding network configuration.\n");
4232 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
4237 if (get_net_conf(mdev
->tconn
)) {
4239 put_net_conf(mdev
->tconn
);
4243 drbd_disconnect(mdev
);
4245 dev_info(DEV
, "receiver terminated\n");
4249 /* ********* acknowledge sender ******** */
4251 static int got_RqSReply(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4253 struct p_req_state_reply
*p
= &mdev
->tconn
->meta
.rbuf
.req_state_reply
;
4255 int retcode
= be32_to_cpu(p
->retcode
);
4257 if (retcode
>= SS_SUCCESS
) {
4258 set_bit(CL_ST_CHG_SUCCESS
, &mdev
->flags
);
4260 set_bit(CL_ST_CHG_FAIL
, &mdev
->flags
);
4261 dev_err(DEV
, "Requested state change failed by peer: %s (%d)\n",
4262 drbd_set_st_err_str(retcode
), retcode
);
4264 wake_up(&mdev
->state_wait
);
4269 static int got_Ping(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4271 return drbd_send_ping_ack(mdev
);
4275 static int got_PingAck(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4277 /* restore idle timeout */
4278 mdev
->tconn
->meta
.socket
->sk
->sk_rcvtimeo
= mdev
->tconn
->net_conf
->ping_int
*HZ
;
4279 if (!test_and_set_bit(GOT_PING_ACK
, &mdev
->flags
))
4280 wake_up(&mdev
->misc_wait
);
4285 static int got_IsInSync(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4287 struct p_block_ack
*p
= &mdev
->tconn
->meta
.rbuf
.block_ack
;
4288 sector_t sector
= be64_to_cpu(p
->sector
);
4289 int blksize
= be32_to_cpu(p
->blksize
);
4291 D_ASSERT(mdev
->tconn
->agreed_pro_version
>= 89);
4293 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4295 if (get_ldev(mdev
)) {
4296 drbd_rs_complete_io(mdev
, sector
);
4297 drbd_set_in_sync(mdev
, sector
, blksize
);
4298 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4299 mdev
->rs_same_csum
+= (blksize
>> BM_BLOCK_SHIFT
);
4302 dec_rs_pending(mdev
);
4303 atomic_add(blksize
>> 9, &mdev
->rs_sect_in
);
4309 validate_req_change_req_state(struct drbd_conf
*mdev
, u64 id
, sector_t sector
,
4310 struct rb_root
*root
, const char *func
,
4311 enum drbd_req_event what
, bool missing_ok
)
4313 struct drbd_request
*req
;
4314 struct bio_and_error m
;
4316 spin_lock_irq(&mdev
->tconn
->req_lock
);
4317 req
= find_request(mdev
, root
, id
, sector
, missing_ok
, func
);
4318 if (unlikely(!req
)) {
4319 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4322 __req_mod(req
, what
, &m
);
4323 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4326 complete_master_bio(mdev
, &m
);
4330 static int got_BlockAck(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4332 struct p_block_ack
*p
= &mdev
->tconn
->meta
.rbuf
.block_ack
;
4333 sector_t sector
= be64_to_cpu(p
->sector
);
4334 int blksize
= be32_to_cpu(p
->blksize
);
4335 enum drbd_req_event what
;
4337 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4339 if (p
->block_id
== ID_SYNCER
) {
4340 drbd_set_in_sync(mdev
, sector
, blksize
);
4341 dec_rs_pending(mdev
);
4345 case P_RS_WRITE_ACK
:
4346 D_ASSERT(mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4347 what
= WRITE_ACKED_BY_PEER_AND_SIS
;
4350 D_ASSERT(mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4351 what
= WRITE_ACKED_BY_PEER
;
4354 D_ASSERT(mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_B
);
4355 what
= RECV_ACKED_BY_PEER
;
4358 D_ASSERT(mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_C
);
4359 what
= CONFLICT_DISCARDED_BY_PEER
;
4366 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4367 &mdev
->write_requests
, __func__
,
4371 static int got_NegAck(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4373 struct p_block_ack
*p
= &mdev
->tconn
->meta
.rbuf
.block_ack
;
4374 sector_t sector
= be64_to_cpu(p
->sector
);
4375 int size
= be32_to_cpu(p
->blksize
);
4376 bool missing_ok
= mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_A
||
4377 mdev
->tconn
->net_conf
->wire_protocol
== DRBD_PROT_B
;
4380 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4382 if (p
->block_id
== ID_SYNCER
) {
4383 dec_rs_pending(mdev
);
4384 drbd_rs_failed_io(mdev
, sector
, size
);
4388 found
= validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4389 &mdev
->write_requests
, __func__
,
4390 NEG_ACKED
, missing_ok
);
4392 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
4393 The master bio might already be completed, therefore the
4394 request is no longer in the collision hash. */
4395 /* In Protocol B we might already have got a P_RECV_ACK
4396 but then get a P_NEG_ACK afterwards. */
4399 drbd_set_out_of_sync(mdev
, sector
, size
);
4404 static int got_NegDReply(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4406 struct p_block_ack
*p
= &mdev
->tconn
->meta
.rbuf
.block_ack
;
4407 sector_t sector
= be64_to_cpu(p
->sector
);
4409 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4410 dev_err(DEV
, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4411 (unsigned long long)sector
, be32_to_cpu(p
->blksize
));
4413 return validate_req_change_req_state(mdev
, p
->block_id
, sector
,
4414 &mdev
->read_requests
, __func__
,
4418 static int got_NegRSDReply(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4422 struct p_block_ack
*p
= &mdev
->tconn
->meta
.rbuf
.block_ack
;
4424 sector
= be64_to_cpu(p
->sector
);
4425 size
= be32_to_cpu(p
->blksize
);
4427 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4429 dec_rs_pending(mdev
);
4431 if (get_ldev_if_state(mdev
, D_FAILED
)) {
4432 drbd_rs_complete_io(mdev
, sector
);
4434 case P_NEG_RS_DREPLY
:
4435 drbd_rs_failed_io(mdev
, sector
, size
);
4449 static int got_BarrierAck(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4451 struct p_barrier_ack
*p
= &mdev
->tconn
->meta
.rbuf
.barrier_ack
;
4453 tl_release(mdev
, p
->barrier
, be32_to_cpu(p
->set_size
));
4455 if (mdev
->state
.conn
== C_AHEAD
&&
4456 atomic_read(&mdev
->ap_in_flight
) == 0 &&
4457 !test_and_set_bit(AHEAD_TO_SYNC_SOURCE
, &mdev
->current_epoch
->flags
)) {
4458 mdev
->start_resync_timer
.expires
= jiffies
+ HZ
;
4459 add_timer(&mdev
->start_resync_timer
);
4465 static int got_OVResult(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4467 struct p_block_ack
*p
= &mdev
->tconn
->meta
.rbuf
.block_ack
;
4468 struct drbd_work
*w
;
4472 sector
= be64_to_cpu(p
->sector
);
4473 size
= be32_to_cpu(p
->blksize
);
4475 update_peer_seq(mdev
, be32_to_cpu(p
->seq_num
));
4477 if (be64_to_cpu(p
->block_id
) == ID_OUT_OF_SYNC
)
4478 drbd_ov_oos_found(mdev
, sector
, size
);
4482 if (!get_ldev(mdev
))
4485 drbd_rs_complete_io(mdev
, sector
);
4486 dec_rs_pending(mdev
);
4490 /* let's advance progress step marks only for every other megabyte */
4491 if ((mdev
->ov_left
& 0x200) == 0x200)
4492 drbd_advance_rs_marks(mdev
, mdev
->ov_left
);
4494 if (mdev
->ov_left
== 0) {
4495 w
= kmalloc(sizeof(*w
), GFP_NOIO
);
4497 w
->cb
= w_ov_finished
;
4498 drbd_queue_work_front(&mdev
->tconn
->data
.work
, w
);
4500 dev_err(DEV
, "kmalloc(w) failed.");
4502 drbd_resync_finished(mdev
);
4509 static int got_skip(struct drbd_conf
*mdev
, enum drbd_packet cmd
)
4514 struct asender_cmd
{
4516 int (*process
)(struct drbd_conf
*mdev
, enum drbd_packet cmd
);
4519 static struct asender_cmd
*get_asender_cmd(int cmd
)
4521 static struct asender_cmd asender_tbl
[] = {
4522 /* anything missing from this table is in
4523 * the drbd_cmd_handler (drbd_default_handler) table,
4524 * see the beginning of drbdd() */
4525 [P_PING
] = { sizeof(struct p_header
), got_Ping
},
4526 [P_PING_ACK
] = { sizeof(struct p_header
), got_PingAck
},
4527 [P_RECV_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4528 [P_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4529 [P_RS_WRITE_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4530 [P_DISCARD_ACK
] = { sizeof(struct p_block_ack
), got_BlockAck
},
4531 [P_NEG_ACK
] = { sizeof(struct p_block_ack
), got_NegAck
},
4532 [P_NEG_DREPLY
] = { sizeof(struct p_block_ack
), got_NegDReply
},
4533 [P_NEG_RS_DREPLY
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
4534 [P_OV_RESULT
] = { sizeof(struct p_block_ack
), got_OVResult
},
4535 [P_BARRIER_ACK
] = { sizeof(struct p_barrier_ack
), got_BarrierAck
},
4536 [P_STATE_CHG_REPLY
] = { sizeof(struct p_req_state_reply
), got_RqSReply
},
4537 [P_RS_IS_IN_SYNC
] = { sizeof(struct p_block_ack
), got_IsInSync
},
4538 [P_DELAY_PROBE
] = { sizeof(struct p_delay_probe93
), got_skip
},
4539 [P_RS_CANCEL
] = { sizeof(struct p_block_ack
), got_NegRSDReply
},
4540 [P_MAX_CMD
] = { 0, NULL
},
4542 if (cmd
> P_MAX_CMD
|| asender_tbl
[cmd
].process
== NULL
)
4544 return &asender_tbl
[cmd
];
4547 int drbd_asender(struct drbd_thread
*thi
)
4549 struct drbd_conf
*mdev
= thi
->mdev
;
4550 struct p_header
*h
= &mdev
->tconn
->meta
.rbuf
.header
;
4551 struct asender_cmd
*cmd
= NULL
;
4556 int expect
= sizeof(struct p_header
);
4557 int ping_timeout_active
= 0;
4558 int empty
, pkt_size
;
4559 enum drbd_packet cmd_nr
;
4561 sprintf(current
->comm
, "drbd%d_asender", mdev_to_minor(mdev
));
4563 current
->policy
= SCHED_RR
; /* Make this a realtime task! */
4564 current
->rt_priority
= 2; /* more important than all other tasks */
4566 while (get_t_state(thi
) == RUNNING
) {
4567 drbd_thread_current_set_cpu(mdev
);
4568 if (test_and_clear_bit(SEND_PING
, &mdev
->flags
)) {
4569 if (!drbd_send_ping(mdev
)) {
4570 dev_err(DEV
, "drbd_send_ping has failed\n");
4573 mdev
->tconn
->meta
.socket
->sk
->sk_rcvtimeo
=
4574 mdev
->tconn
->net_conf
->ping_timeo
*HZ
/10;
4575 ping_timeout_active
= 1;
4578 /* conditionally cork;
4579 * it may hurt latency if we cork without much to send */
4580 if (!mdev
->tconn
->net_conf
->no_cork
&&
4581 3 < atomic_read(&mdev
->unacked_cnt
))
4582 drbd_tcp_cork(mdev
->tconn
->meta
.socket
);
4584 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4585 flush_signals(current
);
4586 if (!drbd_process_done_ee(mdev
))
4588 /* to avoid race with newly queued ACKs */
4589 set_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4590 spin_lock_irq(&mdev
->tconn
->req_lock
);
4591 empty
= list_empty(&mdev
->done_ee
);
4592 spin_unlock_irq(&mdev
->tconn
->req_lock
);
4593 /* new ack may have been queued right here,
4594 * but then there is also a signal pending,
4595 * and we start over... */
4599 /* but unconditionally uncork unless disabled */
4600 if (!mdev
->tconn
->net_conf
->no_cork
)
4601 drbd_tcp_uncork(mdev
->tconn
->meta
.socket
);
4603 /* short circuit, recv_msg would return EINTR anyways. */
4604 if (signal_pending(current
))
4607 rv
= drbd_recv_short(mdev
, mdev
->tconn
->meta
.socket
,
4608 buf
, expect
-received
, 0);
4609 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4611 flush_signals(current
);
4614 * -EINTR (on meta) we got a signal
4615 * -EAGAIN (on meta) rcvtimeo expired
4616 * -ECONNRESET other side closed the connection
4617 * -ERESTARTSYS (on data) we got a signal
4618 * rv < 0 other than above: unexpected error!
4619 * rv == expected: full header or command
4620 * rv < expected: "woken" by signal during receive
4621 * rv == 0 : "connection shut down by peer"
4623 if (likely(rv
> 0)) {
4626 } else if (rv
== 0) {
4627 dev_err(DEV
, "meta connection shut down by peer.\n");
4629 } else if (rv
== -EAGAIN
) {
4630 /* If the data socket received something meanwhile,
4631 * that is good enough: peer is still alive. */
4632 if (time_after(mdev
->tconn
->last_received
,
4633 jiffies
- mdev
->tconn
->meta
.socket
->sk
->sk_rcvtimeo
))
4635 if (ping_timeout_active
) {
4636 dev_err(DEV
, "PingAck did not arrive in time.\n");
4639 set_bit(SEND_PING
, &mdev
->flags
);
4641 } else if (rv
== -EINTR
) {
4644 dev_err(DEV
, "sock_recvmsg returned %d\n", rv
);
4648 if (received
== expect
&& cmd
== NULL
) {
4649 if (!decode_header(mdev
, h
, &cmd_nr
, &pkt_size
))
4651 cmd
= get_asender_cmd(cmd_nr
);
4652 if (unlikely(cmd
== NULL
)) {
4653 dev_err(DEV
, "unknown command %d on meta (l: %d)\n",
4657 expect
= cmd
->pkt_size
;
4658 if (pkt_size
!= expect
- sizeof(struct p_header
)) {
4659 dev_err(DEV
, "Wrong packet size on meta (c: %d, l: %d)\n",
4664 if (received
== expect
) {
4665 mdev
->tconn
->last_received
= jiffies
;
4666 D_ASSERT(cmd
!= NULL
);
4667 if (!cmd
->process(mdev
, cmd_nr
))
4670 /* the idle_timeout (ping-int)
4671 * has been restored in got_PingAck() */
4672 if (cmd
== get_asender_cmd(P_PING_ACK
))
4673 ping_timeout_active
= 0;
4677 expect
= sizeof(struct p_header
);
4684 drbd_force_state(mdev
, NS(conn
, C_NETWORK_FAILURE
));
4689 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
4692 clear_bit(SIGNAL_ASENDER
, &mdev
->flags
);
4694 D_ASSERT(mdev
->state
.conn
< C_CONNECTED
);
4695 dev_info(DEV
, "asender terminated\n");