aboutsummaryrefslogtreecommitdiffstats
path: root/drivers/block/drbd/drbd_receiver.c
diff options
context:
space:
mode:
Diffstat (limited to 'drivers/block/drbd/drbd_receiver.c')
-rw-r--r--drivers/block/drbd/drbd_receiver.c4426
1 files changed, 4426 insertions, 0 deletions
diff --git a/drivers/block/drbd/drbd_receiver.c b/drivers/block/drbd/drbd_receiver.c
new file mode 100644
index 000000000000..c548f24f54a1
--- /dev/null
+++ b/drivers/block/drbd/drbd_receiver.c
@@ -0,0 +1,4426 @@
1/*
2 drbd_receiver.c
3
4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
5
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>.
9
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)
13 any later version.
14
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.
19
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.
23 */
24
25
26#include <linux/module.h>
27
28#include <asm/uaccess.h>
29#include <net/sock.h>
30
31#include <linux/version.h>
32#include <linux/drbd.h>
33#include <linux/fs.h>
34#include <linux/file.h>
35#include <linux/in.h>
36#include <linux/mm.h>
37#include <linux/memcontrol.h>
38#include <linux/mm_inline.h>
39#include <linux/slab.h>
40#include <linux/smp_lock.h>
41#include <linux/pkt_sched.h>
42#define __KERNEL_SYSCALLS__
43#include <linux/unistd.h>
44#include <linux/vmalloc.h>
45#include <linux/random.h>
46#include <linux/mm.h>
47#include <linux/string.h>
48#include <linux/scatterlist.h>
49#include "drbd_int.h"
50#include "drbd_req.h"
51
52#include "drbd_vli.h"
53
54struct flush_work {
55 struct drbd_work w;
56 struct drbd_epoch *epoch;
57};
58
59enum finish_epoch {
60 FE_STILL_LIVE,
61 FE_DESTROYED,
62 FE_RECYCLED,
63};
64
65static int drbd_do_handshake(struct drbd_conf *mdev);
66static int drbd_do_auth(struct drbd_conf *mdev);
67
68static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *, struct drbd_epoch *, enum epoch_event);
69static int e_end_block(struct drbd_conf *, struct drbd_work *, int);
70
71static struct drbd_epoch *previous_epoch(struct drbd_conf *mdev, struct drbd_epoch *epoch)
72{
73 struct drbd_epoch *prev;
74 spin_lock(&mdev->epoch_lock);
75 prev = list_entry(epoch->list.prev, struct drbd_epoch, list);
76 if (prev == epoch || prev == mdev->current_epoch)
77 prev = NULL;
78 spin_unlock(&mdev->epoch_lock);
79 return prev;
80}
81
82#define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
83
84static struct page *drbd_pp_first_page_or_try_alloc(struct drbd_conf *mdev)
85{
86 struct page *page = NULL;
87
88 /* Yes, testing drbd_pp_vacant outside the lock is racy.
89 * So what. It saves a spin_lock. */
90 if (drbd_pp_vacant > 0) {
91 spin_lock(&drbd_pp_lock);
92 page = drbd_pp_pool;
93 if (page) {
94 drbd_pp_pool = (struct page *)page_private(page);
95 set_page_private(page, 0); /* just to be polite */
96 drbd_pp_vacant--;
97 }
98 spin_unlock(&drbd_pp_lock);
99 }
100 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
101 * "criss-cross" setup, that might cause write-out on some other DRBD,
102 * which in turn might block on the other node at this very place. */
103 if (!page)
104 page = alloc_page(GFP_TRY);
105 if (page)
106 atomic_inc(&mdev->pp_in_use);
107 return page;
108}
109
110/* kick lower level device, if we have more than (arbitrary number)
111 * reference counts on it, which typically are locally submitted io
112 * requests. don't use unacked_cnt, so we speed up proto A and B, too. */
113static void maybe_kick_lo(struct drbd_conf *mdev)
114{
115 if (atomic_read(&mdev->local_cnt) >= mdev->net_conf->unplug_watermark)
116 drbd_kick_lo(mdev);
117}
118
119static void reclaim_net_ee(struct drbd_conf *mdev, struct list_head *to_be_freed)
120{
121 struct drbd_epoch_entry *e;
122 struct list_head *le, *tle;
123
124 /* The EEs are always appended to the end of the list. Since
125 they are sent in order over the wire, they have to finish
126 in order. As soon as we see the first not finished we can
127 stop to examine the list... */
128
129 list_for_each_safe(le, tle, &mdev->net_ee) {
130 e = list_entry(le, struct drbd_epoch_entry, w.list);
131 if (drbd_bio_has_active_page(e->private_bio))
132 break;
133 list_move(le, to_be_freed);
134 }
135}
136
137static void drbd_kick_lo_and_reclaim_net(struct drbd_conf *mdev)
138{
139 LIST_HEAD(reclaimed);
140 struct drbd_epoch_entry *e, *t;
141
142 maybe_kick_lo(mdev);
143 spin_lock_irq(&mdev->req_lock);
144 reclaim_net_ee(mdev, &reclaimed);
145 spin_unlock_irq(&mdev->req_lock);
146
147 list_for_each_entry_safe(e, t, &reclaimed, w.list)
148 drbd_free_ee(mdev, e);
149}
150
151/**
152 * drbd_pp_alloc() - Returns a page, fails only if a signal comes in
153 * @mdev: DRBD device.
154 * @retry: whether or not to retry allocation forever (or until signalled)
155 *
156 * Tries to allocate a page, first from our own page pool, then from the
157 * kernel, unless this allocation would exceed the max_buffers setting.
158 * If @retry is non-zero, retry until DRBD frees a page somewhere else.
159 */
160static struct page *drbd_pp_alloc(struct drbd_conf *mdev, int retry)
161{
162 struct page *page = NULL;
163 DEFINE_WAIT(wait);
164
165 if (atomic_read(&mdev->pp_in_use) < mdev->net_conf->max_buffers) {
166 page = drbd_pp_first_page_or_try_alloc(mdev);
167 if (page)
168 return page;
169 }
170
171 for (;;) {
172 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
173
174 drbd_kick_lo_and_reclaim_net(mdev);
175
176 if (atomic_read(&mdev->pp_in_use) < mdev->net_conf->max_buffers) {
177 page = drbd_pp_first_page_or_try_alloc(mdev);
178 if (page)
179 break;
180 }
181
182 if (!retry)
183 break;
184
185 if (signal_pending(current)) {
186 dev_warn(DEV, "drbd_pp_alloc interrupted!\n");
187 break;
188 }
189
190 schedule();
191 }
192 finish_wait(&drbd_pp_wait, &wait);
193
194 return page;
195}
196
197/* Must not be used from irq, as that may deadlock: see drbd_pp_alloc.
198 * Is also used from inside an other spin_lock_irq(&mdev->req_lock) */
199static void drbd_pp_free(struct drbd_conf *mdev, struct page *page)
200{
201 int free_it;
202
203 spin_lock(&drbd_pp_lock);
204 if (drbd_pp_vacant > (DRBD_MAX_SEGMENT_SIZE/PAGE_SIZE)*minor_count) {
205 free_it = 1;
206 } else {
207 set_page_private(page, (unsigned long)drbd_pp_pool);
208 drbd_pp_pool = page;
209 drbd_pp_vacant++;
210 free_it = 0;
211 }
212 spin_unlock(&drbd_pp_lock);
213
214 atomic_dec(&mdev->pp_in_use);
215
216 if (free_it)
217 __free_page(page);
218
219 wake_up(&drbd_pp_wait);
220}
221
222static void drbd_pp_free_bio_pages(struct drbd_conf *mdev, struct bio *bio)
223{
224 struct page *p_to_be_freed = NULL;
225 struct page *page;
226 struct bio_vec *bvec;
227 int i;
228
229 spin_lock(&drbd_pp_lock);
230 __bio_for_each_segment(bvec, bio, i, 0) {
231 if (drbd_pp_vacant > (DRBD_MAX_SEGMENT_SIZE/PAGE_SIZE)*minor_count) {
232 set_page_private(bvec->bv_page, (unsigned long)p_to_be_freed);
233 p_to_be_freed = bvec->bv_page;
234 } else {
235 set_page_private(bvec->bv_page, (unsigned long)drbd_pp_pool);
236 drbd_pp_pool = bvec->bv_page;
237 drbd_pp_vacant++;
238 }
239 }
240 spin_unlock(&drbd_pp_lock);
241 atomic_sub(bio->bi_vcnt, &mdev->pp_in_use);
242
243 while (p_to_be_freed) {
244 page = p_to_be_freed;
245 p_to_be_freed = (struct page *)page_private(page);
246 set_page_private(page, 0); /* just to be polite */
247 put_page(page);
248 }
249
250 wake_up(&drbd_pp_wait);
251}
252
253/*
254You need to hold the req_lock:
255 _drbd_wait_ee_list_empty()
256
257You must not have the req_lock:
258 drbd_free_ee()
259 drbd_alloc_ee()
260 drbd_init_ee()
261 drbd_release_ee()
262 drbd_ee_fix_bhs()
263 drbd_process_done_ee()
264 drbd_clear_done_ee()
265 drbd_wait_ee_list_empty()
266*/
267
268struct drbd_epoch_entry *drbd_alloc_ee(struct drbd_conf *mdev,
269 u64 id,
270 sector_t sector,
271 unsigned int data_size,
272 gfp_t gfp_mask) __must_hold(local)
273{
274 struct request_queue *q;
275 struct drbd_epoch_entry *e;
276 struct page *page;
277 struct bio *bio;
278 unsigned int ds;
279
280 if (FAULT_ACTIVE(mdev, DRBD_FAULT_AL_EE))
281 return NULL;
282
283 e = mempool_alloc(drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
284 if (!e) {
285 if (!(gfp_mask & __GFP_NOWARN))
286 dev_err(DEV, "alloc_ee: Allocation of an EE failed\n");
287 return NULL;
288 }
289
290 bio = bio_alloc(gfp_mask & ~__GFP_HIGHMEM, div_ceil(data_size, PAGE_SIZE));
291 if (!bio) {
292 if (!(gfp_mask & __GFP_NOWARN))
293 dev_err(DEV, "alloc_ee: Allocation of a bio failed\n");
294 goto fail1;
295 }
296
297 bio->bi_bdev = mdev->ldev->backing_bdev;
298 bio->bi_sector = sector;
299
300 ds = data_size;
301 while (ds) {
302 page = drbd_pp_alloc(mdev, (gfp_mask & __GFP_WAIT));
303 if (!page) {
304 if (!(gfp_mask & __GFP_NOWARN))
305 dev_err(DEV, "alloc_ee: Allocation of a page failed\n");
306 goto fail2;
307 }
308 if (!bio_add_page(bio, page, min_t(int, ds, PAGE_SIZE), 0)) {
309 drbd_pp_free(mdev, page);
310 dev_err(DEV, "alloc_ee: bio_add_page(s=%llu,"
311 "data_size=%u,ds=%u) failed\n",
312 (unsigned long long)sector, data_size, ds);
313
314 q = bdev_get_queue(bio->bi_bdev);
315 if (q->merge_bvec_fn) {
316 struct bvec_merge_data bvm = {
317 .bi_bdev = bio->bi_bdev,
318 .bi_sector = bio->bi_sector,
319 .bi_size = bio->bi_size,
320 .bi_rw = bio->bi_rw,
321 };
322 int l = q->merge_bvec_fn(q, &bvm,
323 &bio->bi_io_vec[bio->bi_vcnt]);
324 dev_err(DEV, "merge_bvec_fn() = %d\n", l);
325 }
326
327 /* dump more of the bio. */
328 dev_err(DEV, "bio->bi_max_vecs = %d\n", bio->bi_max_vecs);
329 dev_err(DEV, "bio->bi_vcnt = %d\n", bio->bi_vcnt);
330 dev_err(DEV, "bio->bi_size = %d\n", bio->bi_size);
331 dev_err(DEV, "bio->bi_phys_segments = %d\n", bio->bi_phys_segments);
332
333 goto fail2;
334 break;
335 }
336 ds -= min_t(int, ds, PAGE_SIZE);
337 }
338
339 D_ASSERT(data_size == bio->bi_size);
340
341 bio->bi_private = e;
342 e->mdev = mdev;
343 e->sector = sector;
344 e->size = bio->bi_size;
345
346 e->private_bio = bio;
347 e->block_id = id;
348 INIT_HLIST_NODE(&e->colision);
349 e->epoch = NULL;
350 e->flags = 0;
351
352 return e;
353
354 fail2:
355 drbd_pp_free_bio_pages(mdev, bio);
356 bio_put(bio);
357 fail1:
358 mempool_free(e, drbd_ee_mempool);
359
360 return NULL;
361}
362
363void drbd_free_ee(struct drbd_conf *mdev, struct drbd_epoch_entry *e)
364{
365 struct bio *bio = e->private_bio;
366 drbd_pp_free_bio_pages(mdev, bio);
367 bio_put(bio);
368 D_ASSERT(hlist_unhashed(&e->colision));
369 mempool_free(e, drbd_ee_mempool);
370}
371
372int drbd_release_ee(struct drbd_conf *mdev, struct list_head *list)
373{
374 LIST_HEAD(work_list);
375 struct drbd_epoch_entry *e, *t;
376 int count = 0;
377
378 spin_lock_irq(&mdev->req_lock);
379 list_splice_init(list, &work_list);
380 spin_unlock_irq(&mdev->req_lock);
381
382 list_for_each_entry_safe(e, t, &work_list, w.list) {
383 drbd_free_ee(mdev, e);
384 count++;
385 }
386 return count;
387}
388
389
390/*
391 * This function is called from _asender only_
392 * but see also comments in _req_mod(,barrier_acked)
393 * and receive_Barrier.
394 *
395 * Move entries from net_ee to done_ee, if ready.
396 * Grab done_ee, call all callbacks, free the entries.
397 * The callbacks typically send out ACKs.
398 */
399static int drbd_process_done_ee(struct drbd_conf *mdev)
400{
401 LIST_HEAD(work_list);
402 LIST_HEAD(reclaimed);
403 struct drbd_epoch_entry *e, *t;
404 int ok = (mdev->state.conn >= C_WF_REPORT_PARAMS);
405
406 spin_lock_irq(&mdev->req_lock);
407 reclaim_net_ee(mdev, &reclaimed);
408 list_splice_init(&mdev->done_ee, &work_list);
409 spin_unlock_irq(&mdev->req_lock);
410
411 list_for_each_entry_safe(e, t, &reclaimed, w.list)
412 drbd_free_ee(mdev, e);
413
414 /* possible callbacks here:
415 * e_end_block, and e_end_resync_block, e_send_discard_ack.
416 * all ignore the last argument.
417 */
418 list_for_each_entry_safe(e, t, &work_list, w.list) {
419 /* list_del not necessary, next/prev members not touched */
420 ok = e->w.cb(mdev, &e->w, !ok) && ok;
421 drbd_free_ee(mdev, e);
422 }
423 wake_up(&mdev->ee_wait);
424
425 return ok;
426}
427
428void _drbd_wait_ee_list_empty(struct drbd_conf *mdev, struct list_head *head)
429{
430 DEFINE_WAIT(wait);
431
432 /* avoids spin_lock/unlock
433 * and calling prepare_to_wait in the fast path */
434 while (!list_empty(head)) {
435 prepare_to_wait(&mdev->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
436 spin_unlock_irq(&mdev->req_lock);
437 drbd_kick_lo(mdev);
438 schedule();
439 finish_wait(&mdev->ee_wait, &wait);
440 spin_lock_irq(&mdev->req_lock);
441 }
442}
443
444void drbd_wait_ee_list_empty(struct drbd_conf *mdev, struct list_head *head)
445{
446 spin_lock_irq(&mdev->req_lock);
447 _drbd_wait_ee_list_empty(mdev, head);
448 spin_unlock_irq(&mdev->req_lock);
449}
450
451/* see also kernel_accept; which is only present since 2.6.18.
452 * also we want to log which part of it failed, exactly */
453static int drbd_accept(struct drbd_conf *mdev, const char **what,
454 struct socket *sock, struct socket **newsock)
455{
456 struct sock *sk = sock->sk;
457 int err = 0;
458
459 *what = "listen";
460 err = sock->ops->listen(sock, 5);
461 if (err < 0)
462 goto out;
463
464 *what = "sock_create_lite";
465 err = sock_create_lite(sk->sk_family, sk->sk_type, sk->sk_protocol,
466 newsock);
467 if (err < 0)
468 goto out;
469
470 *what = "accept";
471 err = sock->ops->accept(sock, *newsock, 0);
472 if (err < 0) {
473 sock_release(*newsock);
474 *newsock = NULL;
475 goto out;
476 }
477 (*newsock)->ops = sock->ops;
478
479out:
480 return err;
481}
482
483static int drbd_recv_short(struct drbd_conf *mdev, struct socket *sock,
484 void *buf, size_t size, int flags)
485{
486 mm_segment_t oldfs;
487 struct kvec iov = {
488 .iov_base = buf,
489 .iov_len = size,
490 };
491 struct msghdr msg = {
492 .msg_iovlen = 1,
493 .msg_iov = (struct iovec *)&iov,
494 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
495 };
496 int rv;
497
498 oldfs = get_fs();
499 set_fs(KERNEL_DS);
500 rv = sock_recvmsg(sock, &msg, size, msg.msg_flags);
501 set_fs(oldfs);
502
503 return rv;
504}
505
506static int drbd_recv(struct drbd_conf *mdev, void *buf, size_t size)
507{
508 mm_segment_t oldfs;
509 struct kvec iov = {
510 .iov_base = buf,
511 .iov_len = size,
512 };
513 struct msghdr msg = {
514 .msg_iovlen = 1,
515 .msg_iov = (struct iovec *)&iov,
516 .msg_flags = MSG_WAITALL | MSG_NOSIGNAL
517 };
518 int rv;
519
520 oldfs = get_fs();
521 set_fs(KERNEL_DS);
522
523 for (;;) {
524 rv = sock_recvmsg(mdev->data.socket, &msg, size, msg.msg_flags);
525 if (rv == size)
526 break;
527
528 /* Note:
529 * ECONNRESET other side closed the connection
530 * ERESTARTSYS (on sock) we got a signal
531 */
532
533 if (rv < 0) {
534 if (rv == -ECONNRESET)
535 dev_info(DEV, "sock was reset by peer\n");
536 else if (rv != -ERESTARTSYS)
537 dev_err(DEV, "sock_recvmsg returned %d\n", rv);
538 break;
539 } else if (rv == 0) {
540 dev_info(DEV, "sock was shut down by peer\n");
541 break;
542 } else {
543 /* signal came in, or peer/link went down,
544 * after we read a partial message
545 */
546 /* D_ASSERT(signal_pending(current)); */
547 break;
548 }
549 };
550
551 set_fs(oldfs);
552
553 if (rv != size)
554 drbd_force_state(mdev, NS(conn, C_BROKEN_PIPE));
555
556 return rv;
557}
558
559static struct socket *drbd_try_connect(struct drbd_conf *mdev)
560{
561 const char *what;
562 struct socket *sock;
563 struct sockaddr_in6 src_in6;
564 int err;
565 int disconnect_on_error = 1;
566
567 if (!get_net_conf(mdev))
568 return NULL;
569
570 what = "sock_create_kern";
571 err = sock_create_kern(((struct sockaddr *)mdev->net_conf->my_addr)->sa_family,
572 SOCK_STREAM, IPPROTO_TCP, &sock);
573 if (err < 0) {
574 sock = NULL;
575 goto out;
576 }
577
578 sock->sk->sk_rcvtimeo =
579 sock->sk->sk_sndtimeo = mdev->net_conf->try_connect_int*HZ;
580
581 /* explicitly bind to the configured IP as source IP
582 * for the outgoing connections.
583 * This is needed for multihomed hosts and to be
584 * able to use lo: interfaces for drbd.
585 * Make sure to use 0 as port number, so linux selects
586 * a free one dynamically.
587 */
588 memcpy(&src_in6, mdev->net_conf->my_addr,
589 min_t(int, mdev->net_conf->my_addr_len, sizeof(src_in6)));
590 if (((struct sockaddr *)mdev->net_conf->my_addr)->sa_family == AF_INET6)
591 src_in6.sin6_port = 0;
592 else
593 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
594
595 what = "bind before connect";
596 err = sock->ops->bind(sock,
597 (struct sockaddr *) &src_in6,
598 mdev->net_conf->my_addr_len);
599 if (err < 0)
600 goto out;
601
602 /* connect may fail, peer not yet available.
603 * stay C_WF_CONNECTION, don't go Disconnecting! */
604 disconnect_on_error = 0;
605 what = "connect";
606 err = sock->ops->connect(sock,
607 (struct sockaddr *)mdev->net_conf->peer_addr,
608 mdev->net_conf->peer_addr_len, 0);
609
610out:
611 if (err < 0) {
612 if (sock) {
613 sock_release(sock);
614 sock = NULL;
615 }
616 switch (-err) {
617 /* timeout, busy, signal pending */
618 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
619 case EINTR: case ERESTARTSYS:
620 /* peer not (yet) available, network problem */
621 case ECONNREFUSED: case ENETUNREACH:
622 case EHOSTDOWN: case EHOSTUNREACH:
623 disconnect_on_error = 0;
624 break;
625 default:
626 dev_err(DEV, "%s failed, err = %d\n", what, err);
627 }
628 if (disconnect_on_error)
629 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
630 }
631 put_net_conf(mdev);
632 return sock;
633}
634
635static struct socket *drbd_wait_for_connect(struct drbd_conf *mdev)
636{
637 int timeo, err;
638 struct socket *s_estab = NULL, *s_listen;
639 const char *what;
640
641 if (!get_net_conf(mdev))
642 return NULL;
643
644 what = "sock_create_kern";
645 err = sock_create_kern(((struct sockaddr *)mdev->net_conf->my_addr)->sa_family,
646 SOCK_STREAM, IPPROTO_TCP, &s_listen);
647 if (err) {
648 s_listen = NULL;
649 goto out;
650 }
651
652 timeo = mdev->net_conf->try_connect_int * HZ;
653 timeo += (random32() & 1) ? timeo / 7 : -timeo / 7; /* 28.5% random jitter */
654
655 s_listen->sk->sk_reuse = 1; /* SO_REUSEADDR */
656 s_listen->sk->sk_rcvtimeo = timeo;
657 s_listen->sk->sk_sndtimeo = timeo;
658
659 what = "bind before listen";
660 err = s_listen->ops->bind(s_listen,
661 (struct sockaddr *) mdev->net_conf->my_addr,
662 mdev->net_conf->my_addr_len);
663 if (err < 0)
664 goto out;
665
666 err = drbd_accept(mdev, &what, s_listen, &s_estab);
667
668out:
669 if (s_listen)
670 sock_release(s_listen);
671 if (err < 0) {
672 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
673 dev_err(DEV, "%s failed, err = %d\n", what, err);
674 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
675 }
676 }
677 put_net_conf(mdev);
678
679 return s_estab;
680}
681
682static int drbd_send_fp(struct drbd_conf *mdev,
683 struct socket *sock, enum drbd_packets cmd)
684{
685 struct p_header *h = (struct p_header *) &mdev->data.sbuf.header;
686
687 return _drbd_send_cmd(mdev, sock, cmd, h, sizeof(*h), 0);
688}
689
690static enum drbd_packets drbd_recv_fp(struct drbd_conf *mdev, struct socket *sock)
691{
692 struct p_header *h = (struct p_header *) &mdev->data.sbuf.header;
693 int rr;
694
695 rr = drbd_recv_short(mdev, sock, h, sizeof(*h), 0);
696
697 if (rr == sizeof(*h) && h->magic == BE_DRBD_MAGIC)
698 return be16_to_cpu(h->command);
699
700 return 0xffff;
701}
702
703/**
704 * drbd_socket_okay() - Free the socket if its connection is not okay
705 * @mdev: DRBD device.
706 * @sock: pointer to the pointer to the socket.
707 */
708static int drbd_socket_okay(struct drbd_conf *mdev, struct socket **sock)
709{
710 int rr;
711 char tb[4];
712
713 if (!*sock)
714 return FALSE;
715
716 rr = drbd_recv_short(mdev, *sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
717
718 if (rr > 0 || rr == -EAGAIN) {
719 return TRUE;
720 } else {
721 sock_release(*sock);
722 *sock = NULL;
723 return FALSE;
724 }
725}
726
727/*
728 * return values:
729 * 1 yes, we have a valid connection
730 * 0 oops, did not work out, please try again
731 * -1 peer talks different language,
732 * no point in trying again, please go standalone.
733 * -2 We do not have a network config...
734 */
735static int drbd_connect(struct drbd_conf *mdev)
736{
737 struct socket *s, *sock, *msock;
738 int try, h, ok;
739
740 D_ASSERT(!mdev->data.socket);
741
742 if (test_and_clear_bit(CREATE_BARRIER, &mdev->flags))
743 dev_err(DEV, "CREATE_BARRIER flag was set in drbd_connect - now cleared!\n");
744
745 if (drbd_request_state(mdev, NS(conn, C_WF_CONNECTION)) < SS_SUCCESS)
746 return -2;
747
748 clear_bit(DISCARD_CONCURRENT, &mdev->flags);
749
750 sock = NULL;
751 msock = NULL;
752
753 do {
754 for (try = 0;;) {
755 /* 3 tries, this should take less than a second! */
756 s = drbd_try_connect(mdev);
757 if (s || ++try >= 3)
758 break;
759 /* give the other side time to call bind() & listen() */
760 __set_current_state(TASK_INTERRUPTIBLE);
761 schedule_timeout(HZ / 10);
762 }
763
764 if (s) {
765 if (!sock) {
766 drbd_send_fp(mdev, s, P_HAND_SHAKE_S);
767 sock = s;
768 s = NULL;
769 } else if (!msock) {
770 drbd_send_fp(mdev, s, P_HAND_SHAKE_M);
771 msock = s;
772 s = NULL;
773 } else {
774 dev_err(DEV, "Logic error in drbd_connect()\n");
775 goto out_release_sockets;
776 }
777 }
778
779 if (sock && msock) {
780 __set_current_state(TASK_INTERRUPTIBLE);
781 schedule_timeout(HZ / 10);
782 ok = drbd_socket_okay(mdev, &sock);
783 ok = drbd_socket_okay(mdev, &msock) && ok;
784 if (ok)
785 break;
786 }
787
788retry:
789 s = drbd_wait_for_connect(mdev);
790 if (s) {
791 try = drbd_recv_fp(mdev, s);
792 drbd_socket_okay(mdev, &sock);
793 drbd_socket_okay(mdev, &msock);
794 switch (try) {
795 case P_HAND_SHAKE_S:
796 if (sock) {
797 dev_warn(DEV, "initial packet S crossed\n");
798 sock_release(sock);
799 }
800 sock = s;
801 break;
802 case P_HAND_SHAKE_M:
803 if (msock) {
804 dev_warn(DEV, "initial packet M crossed\n");
805 sock_release(msock);
806 }
807 msock = s;
808 set_bit(DISCARD_CONCURRENT, &mdev->flags);
809 break;
810 default:
811 dev_warn(DEV, "Error receiving initial packet\n");
812 sock_release(s);
813 if (random32() & 1)
814 goto retry;
815 }
816 }
817
818 if (mdev->state.conn <= C_DISCONNECTING)
819 goto out_release_sockets;
820 if (signal_pending(current)) {
821 flush_signals(current);
822 smp_rmb();
823 if (get_t_state(&mdev->receiver) == Exiting)
824 goto out_release_sockets;
825 }
826
827 if (sock && msock) {
828 ok = drbd_socket_okay(mdev, &sock);
829 ok = drbd_socket_okay(mdev, &msock) && ok;
830 if (ok)
831 break;
832 }
833 } while (1);
834
835 msock->sk->sk_reuse = 1; /* SO_REUSEADDR */
836 sock->sk->sk_reuse = 1; /* SO_REUSEADDR */
837
838 sock->sk->sk_allocation = GFP_NOIO;
839 msock->sk->sk_allocation = GFP_NOIO;
840
841 sock->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
842 msock->sk->sk_priority = TC_PRIO_INTERACTIVE;
843
844 if (mdev->net_conf->sndbuf_size) {
845 sock->sk->sk_sndbuf = mdev->net_conf->sndbuf_size;
846 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
847 }
848
849 if (mdev->net_conf->rcvbuf_size) {
850 sock->sk->sk_rcvbuf = mdev->net_conf->rcvbuf_size;
851 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
852 }
853
854 /* NOT YET ...
855 * sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
856 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
857 * first set it to the P_HAND_SHAKE timeout,
858 * which we set to 4x the configured ping_timeout. */
859 sock->sk->sk_sndtimeo =
860 sock->sk->sk_rcvtimeo = mdev->net_conf->ping_timeo*4*HZ/10;
861
862 msock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
863 msock->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ;
864
865 /* we don't want delays.
866 * we use TCP_CORK where apropriate, though */
867 drbd_tcp_nodelay(sock);
868 drbd_tcp_nodelay(msock);
869
870 mdev->data.socket = sock;
871 mdev->meta.socket = msock;
872 mdev->last_received = jiffies;
873
874 D_ASSERT(mdev->asender.task == NULL);
875
876 h = drbd_do_handshake(mdev);
877 if (h <= 0)
878 return h;
879
880 if (mdev->cram_hmac_tfm) {
881 /* drbd_request_state(mdev, NS(conn, WFAuth)); */
882 if (!drbd_do_auth(mdev)) {
883 dev_err(DEV, "Authentication of peer failed\n");
884 return -1;
885 }
886 }
887
888 if (drbd_request_state(mdev, NS(conn, C_WF_REPORT_PARAMS)) < SS_SUCCESS)
889 return 0;
890
891 sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10;
892 sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
893
894 atomic_set(&mdev->packet_seq, 0);
895 mdev->peer_seq = 0;
896
897 drbd_thread_start(&mdev->asender);
898
899 drbd_send_protocol(mdev);
900 drbd_send_sync_param(mdev, &mdev->sync_conf);
901 drbd_send_sizes(mdev, 0);
902 drbd_send_uuids(mdev);
903 drbd_send_state(mdev);
904 clear_bit(USE_DEGR_WFC_T, &mdev->flags);
905 clear_bit(RESIZE_PENDING, &mdev->flags);
906
907 return 1;
908
909out_release_sockets:
910 if (sock)
911 sock_release(sock);
912 if (msock)
913 sock_release(msock);
914 return -1;
915}
916
917static int drbd_recv_header(struct drbd_conf *mdev, struct p_header *h)
918{
919 int r;
920
921 r = drbd_recv(mdev, h, sizeof(*h));
922
923 if (unlikely(r != sizeof(*h))) {
924 dev_err(DEV, "short read expecting header on sock: r=%d\n", r);
925 return FALSE;
926 };
927 h->command = be16_to_cpu(h->command);
928 h->length = be16_to_cpu(h->length);
929 if (unlikely(h->magic != BE_DRBD_MAGIC)) {
930 dev_err(DEV, "magic?? on data m: 0x%lx c: %d l: %d\n",
931 (long)be32_to_cpu(h->magic),
932 h->command, h->length);
933 return FALSE;
934 }
935 mdev->last_received = jiffies;
936
937 return TRUE;
938}
939
940static enum finish_epoch drbd_flush_after_epoch(struct drbd_conf *mdev, struct drbd_epoch *epoch)
941{
942 int rv;
943
944 if (mdev->write_ordering >= WO_bdev_flush && get_ldev(mdev)) {
945 rv = blkdev_issue_flush(mdev->ldev->backing_bdev, NULL);
946 if (rv) {
947 dev_err(DEV, "local disk flush failed with status %d\n", rv);
948 /* would rather check on EOPNOTSUPP, but that is not reliable.
949 * don't try again for ANY return value != 0
950 * if (rv == -EOPNOTSUPP) */
951 drbd_bump_write_ordering(mdev, WO_drain_io);
952 }
953 put_ldev(mdev);
954 }
955
956 return drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE);
957}
958
959static int w_flush(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
960{
961 struct flush_work *fw = (struct flush_work *)w;
962 struct drbd_epoch *epoch = fw->epoch;
963
964 kfree(w);
965
966 if (!test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags))
967 drbd_flush_after_epoch(mdev, epoch);
968
969 drbd_may_finish_epoch(mdev, epoch, EV_PUT |
970 (mdev->state.conn < C_CONNECTED ? EV_CLEANUP : 0));
971
972 return 1;
973}
974
975/**
976 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
977 * @mdev: DRBD device.
978 * @epoch: Epoch object.
979 * @ev: Epoch event.
980 */
981static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *mdev,
982 struct drbd_epoch *epoch,
983 enum epoch_event ev)
984{
985 int finish, epoch_size;
986 struct drbd_epoch *next_epoch;
987 int schedule_flush = 0;
988 enum finish_epoch rv = FE_STILL_LIVE;
989
990 spin_lock(&mdev->epoch_lock);
991 do {
992 next_epoch = NULL;
993 finish = 0;
994
995 epoch_size = atomic_read(&epoch->epoch_size);
996
997 switch (ev & ~EV_CLEANUP) {
998 case EV_PUT:
999 atomic_dec(&epoch->active);
1000 break;
1001 case EV_GOT_BARRIER_NR:
1002 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1003
1004 /* Special case: If we just switched from WO_bio_barrier to
1005 WO_bdev_flush we should not finish the current epoch */
1006 if (test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags) && epoch_size == 1 &&
1007 mdev->write_ordering != WO_bio_barrier &&
1008 epoch == mdev->current_epoch)
1009 clear_bit(DE_CONTAINS_A_BARRIER, &epoch->flags);
1010 break;
1011 case EV_BARRIER_DONE:
1012 set_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE, &epoch->flags);
1013 break;
1014 case EV_BECAME_LAST:
1015 /* nothing to do*/
1016 break;
1017 }
1018
1019 if (epoch_size != 0 &&
1020 atomic_read(&epoch->active) == 0 &&
1021 test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) &&
1022 epoch->list.prev == &mdev->current_epoch->list &&
1023 !test_bit(DE_IS_FINISHING, &epoch->flags)) {
1024 /* Nearly all conditions are met to finish that epoch... */
1025 if (test_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE, &epoch->flags) ||
1026 mdev->write_ordering == WO_none ||
1027 (epoch_size == 1 && test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags)) ||
1028 ev & EV_CLEANUP) {
1029 finish = 1;
1030 set_bit(DE_IS_FINISHING, &epoch->flags);
1031 } else if (!test_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags) &&
1032 mdev->write_ordering == WO_bio_barrier) {
1033 atomic_inc(&epoch->active);
1034 schedule_flush = 1;
1035 }
1036 }
1037 if (finish) {
1038 if (!(ev & EV_CLEANUP)) {
1039 spin_unlock(&mdev->epoch_lock);
1040 drbd_send_b_ack(mdev, epoch->barrier_nr, epoch_size);
1041 spin_lock(&mdev->epoch_lock);
1042 }
1043 dec_unacked(mdev);
1044
1045 if (mdev->current_epoch != epoch) {
1046 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1047 list_del(&epoch->list);
1048 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1049 mdev->epochs--;
1050 kfree(epoch);
1051
1052 if (rv == FE_STILL_LIVE)
1053 rv = FE_DESTROYED;
1054 } else {
1055 epoch->flags = 0;
1056 atomic_set(&epoch->epoch_size, 0);
1057 /* atomic_set(&epoch->active, 0); is alrady zero */
1058 if (rv == FE_STILL_LIVE)
1059 rv = FE_RECYCLED;
1060 }
1061 }
1062
1063 if (!next_epoch)
1064 break;
1065
1066 epoch = next_epoch;
1067 } while (1);
1068
1069 spin_unlock(&mdev->epoch_lock);
1070
1071 if (schedule_flush) {
1072 struct flush_work *fw;
1073 fw = kmalloc(sizeof(*fw), GFP_ATOMIC);
1074 if (fw) {
1075 fw->w.cb = w_flush;
1076 fw->epoch = epoch;
1077 drbd_queue_work(&mdev->data.work, &fw->w);
1078 } else {
1079 dev_warn(DEV, "Could not kmalloc a flush_work obj\n");
1080 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags);
1081 /* That is not a recursion, only one level */
1082 drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE);
1083 drbd_may_finish_epoch(mdev, epoch, EV_PUT);
1084 }
1085 }
1086
1087 return rv;
1088}
1089
1090/**
1091 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1092 * @mdev: DRBD device.
1093 * @wo: Write ordering method to try.
1094 */
1095void drbd_bump_write_ordering(struct drbd_conf *mdev, enum write_ordering_e wo) __must_hold(local)
1096{
1097 enum write_ordering_e pwo;
1098 static char *write_ordering_str[] = {
1099 [WO_none] = "none",
1100 [WO_drain_io] = "drain",
1101 [WO_bdev_flush] = "flush",
1102 [WO_bio_barrier] = "barrier",
1103 };
1104
1105 pwo = mdev->write_ordering;
1106 wo = min(pwo, wo);
1107 if (wo == WO_bio_barrier && mdev->ldev->dc.no_disk_barrier)
1108 wo = WO_bdev_flush;
1109 if (wo == WO_bdev_flush && mdev->ldev->dc.no_disk_flush)
1110 wo = WO_drain_io;
1111 if (wo == WO_drain_io && mdev->ldev->dc.no_disk_drain)
1112 wo = WO_none;
1113 mdev->write_ordering = wo;
1114 if (pwo != mdev->write_ordering || wo == WO_bio_barrier)
1115 dev_info(DEV, "Method to ensure write ordering: %s\n", write_ordering_str[mdev->write_ordering]);
1116}
1117
1118/**
1119 * w_e_reissue() - Worker callback; Resubmit a bio, without BIO_RW_BARRIER set
1120 * @mdev: DRBD device.
1121 * @w: work object.
1122 * @cancel: The connection will be closed anyways (unused in this callback)
1123 */
1124int w_e_reissue(struct drbd_conf *mdev, struct drbd_work *w, int cancel) __releases(local)
1125{
1126 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1127 struct bio *bio = e->private_bio;
1128
1129 /* We leave DE_CONTAINS_A_BARRIER and EE_IS_BARRIER in place,
1130 (and DE_BARRIER_IN_NEXT_EPOCH_ISSUED in the previous Epoch)
1131 so that we can finish that epoch in drbd_may_finish_epoch().
1132 That is necessary if we already have a long chain of Epochs, before
1133 we realize that BIO_RW_BARRIER is actually not supported */
1134
1135 /* As long as the -ENOTSUPP on the barrier is reported immediately
1136 that will never trigger. If it is reported late, we will just
1137 print that warning and continue correctly for all future requests
1138 with WO_bdev_flush */
1139 if (previous_epoch(mdev, e->epoch))
1140 dev_warn(DEV, "Write ordering was not enforced (one time event)\n");
1141
1142 /* prepare bio for re-submit,
1143 * re-init volatile members */
1144 /* we still have a local reference,
1145 * get_ldev was done in receive_Data. */
1146 bio->bi_bdev = mdev->ldev->backing_bdev;
1147 bio->bi_sector = e->sector;
1148 bio->bi_size = e->size;
1149 bio->bi_idx = 0;
1150
1151 bio->bi_flags &= ~(BIO_POOL_MASK - 1);
1152 bio->bi_flags |= 1 << BIO_UPTODATE;
1153
1154 /* don't know whether this is necessary: */
1155 bio->bi_phys_segments = 0;
1156 bio->bi_next = NULL;
1157
1158 /* these should be unchanged: */
1159 /* bio->bi_end_io = drbd_endio_write_sec; */
1160 /* bio->bi_vcnt = whatever; */
1161
1162 e->w.cb = e_end_block;
1163
1164 /* This is no longer a barrier request. */
1165 bio->bi_rw &= ~(1UL << BIO_RW_BARRIER);
1166
1167 drbd_generic_make_request(mdev, DRBD_FAULT_DT_WR, bio);
1168
1169 return 1;
1170}
1171
1172static int receive_Barrier(struct drbd_conf *mdev, struct p_header *h)
1173{
1174 int rv, issue_flush;
1175 struct p_barrier *p = (struct p_barrier *)h;
1176 struct drbd_epoch *epoch;
1177
1178 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
1179
1180 rv = drbd_recv(mdev, h->payload, h->length);
1181 ERR_IF(rv != h->length) return FALSE;
1182
1183 inc_unacked(mdev);
1184
1185 if (mdev->net_conf->wire_protocol != DRBD_PROT_C)
1186 drbd_kick_lo(mdev);
1187
1188 mdev->current_epoch->barrier_nr = p->barrier;
1189 rv = drbd_may_finish_epoch(mdev, mdev->current_epoch, EV_GOT_BARRIER_NR);
1190
1191 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1192 * the activity log, which means it would not be resynced in case the
1193 * R_PRIMARY crashes now.
1194 * Therefore we must send the barrier_ack after the barrier request was
1195 * completed. */
1196 switch (mdev->write_ordering) {
1197 case WO_bio_barrier:
1198 case WO_none:
1199 if (rv == FE_RECYCLED)
1200 return TRUE;
1201 break;
1202
1203 case WO_bdev_flush:
1204 case WO_drain_io:
1205 D_ASSERT(rv == FE_STILL_LIVE);
1206 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &mdev->current_epoch->flags);
1207 drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
1208 rv = drbd_flush_after_epoch(mdev, mdev->current_epoch);
1209 if (rv == FE_RECYCLED)
1210 return TRUE;
1211
1212 /* The asender will send all the ACKs and barrier ACKs out, since
1213 all EEs moved from the active_ee to the done_ee. We need to
1214 provide a new epoch object for the EEs that come in soon */
1215 break;
1216 }
1217
1218 /* receiver context, in the writeout path of the other node.
1219 * avoid potential distributed deadlock */
1220 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1221 if (!epoch) {
1222 dev_warn(DEV, "Allocation of an epoch failed, slowing down\n");
1223 issue_flush = !test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags);
1224 drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
1225 if (issue_flush) {
1226 rv = drbd_flush_after_epoch(mdev, mdev->current_epoch);
1227 if (rv == FE_RECYCLED)
1228 return TRUE;
1229 }
1230
1231 drbd_wait_ee_list_empty(mdev, &mdev->done_ee);
1232
1233 return TRUE;
1234 }
1235
1236 epoch->flags = 0;
1237 atomic_set(&epoch->epoch_size, 0);
1238 atomic_set(&epoch->active, 0);
1239
1240 spin_lock(&mdev->epoch_lock);
1241 if (atomic_read(&mdev->current_epoch->epoch_size)) {
1242 list_add(&epoch->list, &mdev->current_epoch->list);
1243 mdev->current_epoch = epoch;
1244 mdev->epochs++;
1245 } else {
1246 /* The current_epoch got recycled while we allocated this one... */
1247 kfree(epoch);
1248 }
1249 spin_unlock(&mdev->epoch_lock);
1250
1251 return TRUE;
1252}
1253
1254/* used from receive_RSDataReply (recv_resync_read)
1255 * and from receive_Data */
1256static struct drbd_epoch_entry *
1257read_in_block(struct drbd_conf *mdev, u64 id, sector_t sector, int data_size) __must_hold(local)
1258{
1259 struct drbd_epoch_entry *e;
1260 struct bio_vec *bvec;
1261 struct page *page;
1262 struct bio *bio;
1263 int dgs, ds, i, rr;
1264 void *dig_in = mdev->int_dig_in;
1265 void *dig_vv = mdev->int_dig_vv;
1266
1267 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
1268 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
1269
1270 if (dgs) {
1271 rr = drbd_recv(mdev, dig_in, dgs);
1272 if (rr != dgs) {
1273 dev_warn(DEV, "short read receiving data digest: read %d expected %d\n",
1274 rr, dgs);
1275 return NULL;
1276 }
1277 }
1278
1279 data_size -= dgs;
1280
1281 ERR_IF(data_size & 0x1ff) return NULL;
1282 ERR_IF(data_size > DRBD_MAX_SEGMENT_SIZE) return NULL;
1283
1284 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1285 * "criss-cross" setup, that might cause write-out on some other DRBD,
1286 * which in turn might block on the other node at this very place. */
1287 e = drbd_alloc_ee(mdev, id, sector, data_size, GFP_NOIO);
1288 if (!e)
1289 return NULL;
1290 bio = e->private_bio;
1291 ds = data_size;
1292 bio_for_each_segment(bvec, bio, i) {
1293 page = bvec->bv_page;
1294 rr = drbd_recv(mdev, kmap(page), min_t(int, ds, PAGE_SIZE));
1295 kunmap(page);
1296 if (rr != min_t(int, ds, PAGE_SIZE)) {
1297 drbd_free_ee(mdev, e);
1298 dev_warn(DEV, "short read receiving data: read %d expected %d\n",
1299 rr, min_t(int, ds, PAGE_SIZE));
1300 return NULL;
1301 }
1302 ds -= rr;
1303 }
1304
1305 if (dgs) {
1306 drbd_csum(mdev, mdev->integrity_r_tfm, bio, dig_vv);
1307 if (memcmp(dig_in, dig_vv, dgs)) {
1308 dev_err(DEV, "Digest integrity check FAILED.\n");
1309 drbd_bcast_ee(mdev, "digest failed",
1310 dgs, dig_in, dig_vv, e);
1311 drbd_free_ee(mdev, e);
1312 return NULL;
1313 }
1314 }
1315 mdev->recv_cnt += data_size>>9;
1316 return e;
1317}
1318
1319/* drbd_drain_block() just takes a data block
1320 * out of the socket input buffer, and discards it.
1321 */
1322static int drbd_drain_block(struct drbd_conf *mdev, int data_size)
1323{
1324 struct page *page;
1325 int rr, rv = 1;
1326 void *data;
1327
1328 page = drbd_pp_alloc(mdev, 1);
1329
1330 data = kmap(page);
1331 while (data_size) {
1332 rr = drbd_recv(mdev, data, min_t(int, data_size, PAGE_SIZE));
1333 if (rr != min_t(int, data_size, PAGE_SIZE)) {
1334 rv = 0;
1335 dev_warn(DEV, "short read receiving data: read %d expected %d\n",
1336 rr, min_t(int, data_size, PAGE_SIZE));
1337 break;
1338 }
1339 data_size -= rr;
1340 }
1341 kunmap(page);
1342 drbd_pp_free(mdev, page);
1343 return rv;
1344}
1345
1346static int recv_dless_read(struct drbd_conf *mdev, struct drbd_request *req,
1347 sector_t sector, int data_size)
1348{
1349 struct bio_vec *bvec;
1350 struct bio *bio;
1351 int dgs, rr, i, expect;
1352 void *dig_in = mdev->int_dig_in;
1353 void *dig_vv = mdev->int_dig_vv;
1354
1355 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
1356 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
1357
1358 if (dgs) {
1359 rr = drbd_recv(mdev, dig_in, dgs);
1360 if (rr != dgs) {
1361 dev_warn(DEV, "short read receiving data reply digest: read %d expected %d\n",
1362 rr, dgs);
1363 return 0;
1364 }
1365 }
1366
1367 data_size -= dgs;
1368
1369 /* optimistically update recv_cnt. if receiving fails below,
1370 * we disconnect anyways, and counters will be reset. */
1371 mdev->recv_cnt += data_size>>9;
1372
1373 bio = req->master_bio;
1374 D_ASSERT(sector == bio->bi_sector);
1375
1376 bio_for_each_segment(bvec, bio, i) {
1377 expect = min_t(int, data_size, bvec->bv_len);
1378 rr = drbd_recv(mdev,
1379 kmap(bvec->bv_page)+bvec->bv_offset,
1380 expect);
1381 kunmap(bvec->bv_page);
1382 if (rr != expect) {
1383 dev_warn(DEV, "short read receiving data reply: "
1384 "read %d expected %d\n",
1385 rr, expect);
1386 return 0;
1387 }
1388 data_size -= rr;
1389 }
1390
1391 if (dgs) {
1392 drbd_csum(mdev, mdev->integrity_r_tfm, bio, dig_vv);
1393 if (memcmp(dig_in, dig_vv, dgs)) {
1394 dev_err(DEV, "Digest integrity check FAILED. Broken NICs?\n");
1395 return 0;
1396 }
1397 }
1398
1399 D_ASSERT(data_size == 0);
1400 return 1;
1401}
1402
1403/* e_end_resync_block() is called via
1404 * drbd_process_done_ee() by asender only */
1405static int e_end_resync_block(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1406{
1407 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1408 sector_t sector = e->sector;
1409 int ok;
1410
1411 D_ASSERT(hlist_unhashed(&e->colision));
1412
1413 if (likely(drbd_bio_uptodate(e->private_bio))) {
1414 drbd_set_in_sync(mdev, sector, e->size);
1415 ok = drbd_send_ack(mdev, P_RS_WRITE_ACK, e);
1416 } else {
1417 /* Record failure to sync */
1418 drbd_rs_failed_io(mdev, sector, e->size);
1419
1420 ok = drbd_send_ack(mdev, P_NEG_ACK, e);
1421 }
1422 dec_unacked(mdev);
1423
1424 return ok;
1425}
1426
1427static int recv_resync_read(struct drbd_conf *mdev, sector_t sector, int data_size) __releases(local)
1428{
1429 struct drbd_epoch_entry *e;
1430
1431 e = read_in_block(mdev, ID_SYNCER, sector, data_size);
1432 if (!e) {
1433 put_ldev(mdev);
1434 return FALSE;
1435 }
1436
1437 dec_rs_pending(mdev);
1438
1439 e->private_bio->bi_end_io = drbd_endio_write_sec;
1440 e->private_bio->bi_rw = WRITE;
1441 e->w.cb = e_end_resync_block;
1442
1443 inc_unacked(mdev);
1444 /* corresponding dec_unacked() in e_end_resync_block()
1445 * respective _drbd_clear_done_ee */
1446
1447 spin_lock_irq(&mdev->req_lock);
1448 list_add(&e->w.list, &mdev->sync_ee);
1449 spin_unlock_irq(&mdev->req_lock);
1450
1451 drbd_generic_make_request(mdev, DRBD_FAULT_RS_WR, e->private_bio);
1452 /* accounting done in endio */
1453
1454 maybe_kick_lo(mdev);
1455 return TRUE;
1456}
1457
1458static int receive_DataReply(struct drbd_conf *mdev, struct p_header *h)
1459{
1460 struct drbd_request *req;
1461 sector_t sector;
1462 unsigned int header_size, data_size;
1463 int ok;
1464 struct p_data *p = (struct p_data *)h;
1465
1466 header_size = sizeof(*p) - sizeof(*h);
1467 data_size = h->length - header_size;
1468
1469 ERR_IF(data_size == 0) return FALSE;
1470
1471 if (drbd_recv(mdev, h->payload, header_size) != header_size)
1472 return FALSE;
1473
1474 sector = be64_to_cpu(p->sector);
1475
1476 spin_lock_irq(&mdev->req_lock);
1477 req = _ar_id_to_req(mdev, p->block_id, sector);
1478 spin_unlock_irq(&mdev->req_lock);
1479 if (unlikely(!req)) {
1480 dev_err(DEV, "Got a corrupt block_id/sector pair(1).\n");
1481 return FALSE;
1482 }
1483
1484 /* hlist_del(&req->colision) is done in _req_may_be_done, to avoid
1485 * special casing it there for the various failure cases.
1486 * still no race with drbd_fail_pending_reads */
1487 ok = recv_dless_read(mdev, req, sector, data_size);
1488
1489 if (ok)
1490 req_mod(req, data_received);
1491 /* else: nothing. handled from drbd_disconnect...
1492 * I don't think we may complete this just yet
1493 * in case we are "on-disconnect: freeze" */
1494
1495 return ok;
1496}
1497
1498static int receive_RSDataReply(struct drbd_conf *mdev, struct p_header *h)
1499{
1500 sector_t sector;
1501 unsigned int header_size, data_size;
1502 int ok;
1503 struct p_data *p = (struct p_data *)h;
1504
1505 header_size = sizeof(*p) - sizeof(*h);
1506 data_size = h->length - header_size;
1507
1508 ERR_IF(data_size == 0) return FALSE;
1509
1510 if (drbd_recv(mdev, h->payload, header_size) != header_size)
1511 return FALSE;
1512
1513 sector = be64_to_cpu(p->sector);
1514 D_ASSERT(p->block_id == ID_SYNCER);
1515
1516 if (get_ldev(mdev)) {
1517 /* data is submitted to disk within recv_resync_read.
1518 * corresponding put_ldev done below on error,
1519 * or in drbd_endio_write_sec. */
1520 ok = recv_resync_read(mdev, sector, data_size);
1521 } else {
1522 if (__ratelimit(&drbd_ratelimit_state))
1523 dev_err(DEV, "Can not write resync data to local disk.\n");
1524
1525 ok = drbd_drain_block(mdev, data_size);
1526
1527 drbd_send_ack_dp(mdev, P_NEG_ACK, p);
1528 }
1529
1530 return ok;
1531}
1532
1533/* e_end_block() is called via drbd_process_done_ee().
1534 * this means this function only runs in the asender thread
1535 */
1536static int e_end_block(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1537{
1538 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1539 sector_t sector = e->sector;
1540 struct drbd_epoch *epoch;
1541 int ok = 1, pcmd;
1542
1543 if (e->flags & EE_IS_BARRIER) {
1544 epoch = previous_epoch(mdev, e->epoch);
1545 if (epoch)
1546 drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE + (cancel ? EV_CLEANUP : 0));
1547 }
1548
1549 if (mdev->net_conf->wire_protocol == DRBD_PROT_C) {
1550 if (likely(drbd_bio_uptodate(e->private_bio))) {
1551 pcmd = (mdev->state.conn >= C_SYNC_SOURCE &&
1552 mdev->state.conn <= C_PAUSED_SYNC_T &&
1553 e->flags & EE_MAY_SET_IN_SYNC) ?
1554 P_RS_WRITE_ACK : P_WRITE_ACK;
1555 ok &= drbd_send_ack(mdev, pcmd, e);
1556 if (pcmd == P_RS_WRITE_ACK)
1557 drbd_set_in_sync(mdev, sector, e->size);
1558 } else {
1559 ok = drbd_send_ack(mdev, P_NEG_ACK, e);
1560 /* we expect it to be marked out of sync anyways...
1561 * maybe assert this? */
1562 }
1563 dec_unacked(mdev);
1564 }
1565 /* we delete from the conflict detection hash _after_ we sent out the
1566 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */
1567 if (mdev->net_conf->two_primaries) {
1568 spin_lock_irq(&mdev->req_lock);
1569 D_ASSERT(!hlist_unhashed(&e->colision));
1570 hlist_del_init(&e->colision);
1571 spin_unlock_irq(&mdev->req_lock);
1572 } else {
1573 D_ASSERT(hlist_unhashed(&e->colision));
1574 }
1575
1576 drbd_may_finish_epoch(mdev, e->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
1577
1578 return ok;
1579}
1580
1581static int e_send_discard_ack(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1582{
1583 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1584 int ok = 1;
1585
1586 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
1587 ok = drbd_send_ack(mdev, P_DISCARD_ACK, e);
1588
1589 spin_lock_irq(&mdev->req_lock);
1590 D_ASSERT(!hlist_unhashed(&e->colision));
1591 hlist_del_init(&e->colision);
1592 spin_unlock_irq(&mdev->req_lock);
1593
1594 dec_unacked(mdev);
1595
1596 return ok;
1597}
1598
1599/* Called from receive_Data.
1600 * Synchronize packets on sock with packets on msock.
1601 *
1602 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1603 * packet traveling on msock, they are still processed in the order they have
1604 * been sent.
1605 *
1606 * Note: we don't care for Ack packets overtaking P_DATA packets.
1607 *
1608 * In case packet_seq is larger than mdev->peer_seq number, there are
1609 * outstanding packets on the msock. We wait for them to arrive.
1610 * In case we are the logically next packet, we update mdev->peer_seq
1611 * ourselves. Correctly handles 32bit wrap around.
1612 *
1613 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1614 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1615 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1616 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1617 *
1618 * returns 0 if we may process the packet,
1619 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1620static int drbd_wait_peer_seq(struct drbd_conf *mdev, const u32 packet_seq)
1621{
1622 DEFINE_WAIT(wait);
1623 unsigned int p_seq;
1624 long timeout;
1625 int ret = 0;
1626 spin_lock(&mdev->peer_seq_lock);
1627 for (;;) {
1628 prepare_to_wait(&mdev->seq_wait, &wait, TASK_INTERRUPTIBLE);
1629 if (seq_le(packet_seq, mdev->peer_seq+1))
1630 break;
1631 if (signal_pending(current)) {
1632 ret = -ERESTARTSYS;
1633 break;
1634 }
1635 p_seq = mdev->peer_seq;
1636 spin_unlock(&mdev->peer_seq_lock);
1637 timeout = schedule_timeout(30*HZ);
1638 spin_lock(&mdev->peer_seq_lock);
1639 if (timeout == 0 && p_seq == mdev->peer_seq) {
1640 ret = -ETIMEDOUT;
1641 dev_err(DEV, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n");
1642 break;
1643 }
1644 }
1645 finish_wait(&mdev->seq_wait, &wait);
1646 if (mdev->peer_seq+1 == packet_seq)
1647 mdev->peer_seq++;
1648 spin_unlock(&mdev->peer_seq_lock);
1649 return ret;
1650}
1651
1652/* mirrored write */
1653static int receive_Data(struct drbd_conf *mdev, struct p_header *h)
1654{
1655 sector_t sector;
1656 struct drbd_epoch_entry *e;
1657 struct p_data *p = (struct p_data *)h;
1658 int header_size, data_size;
1659 int rw = WRITE;
1660 u32 dp_flags;
1661
1662 header_size = sizeof(*p) - sizeof(*h);
1663 data_size = h->length - header_size;
1664
1665 ERR_IF(data_size == 0) return FALSE;
1666
1667 if (drbd_recv(mdev, h->payload, header_size) != header_size)
1668 return FALSE;
1669
1670 if (!get_ldev(mdev)) {
1671 if (__ratelimit(&drbd_ratelimit_state))
1672 dev_err(DEV, "Can not write mirrored data block "
1673 "to local disk.\n");
1674 spin_lock(&mdev->peer_seq_lock);
1675 if (mdev->peer_seq+1 == be32_to_cpu(p->seq_num))
1676 mdev->peer_seq++;
1677 spin_unlock(&mdev->peer_seq_lock);
1678
1679 drbd_send_ack_dp(mdev, P_NEG_ACK, p);
1680 atomic_inc(&mdev->current_epoch->epoch_size);
1681 return drbd_drain_block(mdev, data_size);
1682 }
1683
1684 /* get_ldev(mdev) successful.
1685 * Corresponding put_ldev done either below (on various errors),
1686 * or in drbd_endio_write_sec, if we successfully submit the data at
1687 * the end of this function. */
1688
1689 sector = be64_to_cpu(p->sector);
1690 e = read_in_block(mdev, p->block_id, sector, data_size);
1691 if (!e) {
1692 put_ldev(mdev);
1693 return FALSE;
1694 }
1695
1696 e->private_bio->bi_end_io = drbd_endio_write_sec;
1697 e->w.cb = e_end_block;
1698
1699 spin_lock(&mdev->epoch_lock);
1700 e->epoch = mdev->current_epoch;
1701 atomic_inc(&e->epoch->epoch_size);
1702 atomic_inc(&e->epoch->active);
1703
1704 if (mdev->write_ordering == WO_bio_barrier && atomic_read(&e->epoch->epoch_size) == 1) {
1705 struct drbd_epoch *epoch;
1706 /* Issue a barrier if we start a new epoch, and the previous epoch
1707 was not a epoch containing a single request which already was
1708 a Barrier. */
1709 epoch = list_entry(e->epoch->list.prev, struct drbd_epoch, list);
1710 if (epoch == e->epoch) {
1711 set_bit(DE_CONTAINS_A_BARRIER, &e->epoch->flags);
1712 rw |= (1<<BIO_RW_BARRIER);
1713 e->flags |= EE_IS_BARRIER;
1714 } else {
1715 if (atomic_read(&epoch->epoch_size) > 1 ||
1716 !test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags)) {
1717 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags);
1718 set_bit(DE_CONTAINS_A_BARRIER, &e->epoch->flags);
1719 rw |= (1<<BIO_RW_BARRIER);
1720 e->flags |= EE_IS_BARRIER;
1721 }
1722 }
1723 }
1724 spin_unlock(&mdev->epoch_lock);
1725
1726 dp_flags = be32_to_cpu(p->dp_flags);
1727 if (dp_flags & DP_HARDBARRIER) {
1728 dev_err(DEV, "ASSERT FAILED would have submitted barrier request\n");
1729 /* rw |= (1<<BIO_RW_BARRIER); */
1730 }
1731 if (dp_flags & DP_RW_SYNC)
1732 rw |= (1<<BIO_RW_SYNCIO) | (1<<BIO_RW_UNPLUG);
1733 if (dp_flags & DP_MAY_SET_IN_SYNC)
1734 e->flags |= EE_MAY_SET_IN_SYNC;
1735
1736 /* I'm the receiver, I do hold a net_cnt reference. */
1737 if (!mdev->net_conf->two_primaries) {
1738 spin_lock_irq(&mdev->req_lock);
1739 } else {
1740 /* don't get the req_lock yet,
1741 * we may sleep in drbd_wait_peer_seq */
1742 const int size = e->size;
1743 const int discard = test_bit(DISCARD_CONCURRENT, &mdev->flags);
1744 DEFINE_WAIT(wait);
1745 struct drbd_request *i;
1746 struct hlist_node *n;
1747 struct hlist_head *slot;
1748 int first;
1749
1750 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
1751 BUG_ON(mdev->ee_hash == NULL);
1752 BUG_ON(mdev->tl_hash == NULL);
1753
1754 /* conflict detection and handling:
1755 * 1. wait on the sequence number,
1756 * in case this data packet overtook ACK packets.
1757 * 2. check our hash tables for conflicting requests.
1758 * we only need to walk the tl_hash, since an ee can not
1759 * have a conflict with an other ee: on the submitting
1760 * node, the corresponding req had already been conflicting,
1761 * and a conflicting req is never sent.
1762 *
1763 * Note: for two_primaries, we are protocol C,
1764 * so there cannot be any request that is DONE
1765 * but still on the transfer log.
1766 *
1767 * unconditionally add to the ee_hash.
1768 *
1769 * if no conflicting request is found:
1770 * submit.
1771 *
1772 * if any conflicting request is found
1773 * that has not yet been acked,
1774 * AND I have the "discard concurrent writes" flag:
1775 * queue (via done_ee) the P_DISCARD_ACK; OUT.
1776 *
1777 * if any conflicting request is found:
1778 * block the receiver, waiting on misc_wait
1779 * until no more conflicting requests are there,
1780 * or we get interrupted (disconnect).
1781 *
1782 * we do not just write after local io completion of those
1783 * requests, but only after req is done completely, i.e.
1784 * we wait for the P_DISCARD_ACK to arrive!
1785 *
1786 * then proceed normally, i.e. submit.
1787 */
1788 if (drbd_wait_peer_seq(mdev, be32_to_cpu(p->seq_num)))
1789 goto out_interrupted;
1790
1791 spin_lock_irq(&mdev->req_lock);
1792
1793 hlist_add_head(&e->colision, ee_hash_slot(mdev, sector));
1794
1795#define OVERLAPS overlaps(i->sector, i->size, sector, size)
1796 slot = tl_hash_slot(mdev, sector);
1797 first = 1;
1798 for (;;) {
1799 int have_unacked = 0;
1800 int have_conflict = 0;
1801 prepare_to_wait(&mdev->misc_wait, &wait,
1802 TASK_INTERRUPTIBLE);
1803 hlist_for_each_entry(i, n, slot, colision) {
1804 if (OVERLAPS) {
1805 /* only ALERT on first iteration,
1806 * we may be woken up early... */
1807 if (first)
1808 dev_alert(DEV, "%s[%u] Concurrent local write detected!"
1809 " new: %llus +%u; pending: %llus +%u\n",
1810 current->comm, current->pid,
1811 (unsigned long long)sector, size,
1812 (unsigned long long)i->sector, i->size);
1813 if (i->rq_state & RQ_NET_PENDING)
1814 ++have_unacked;
1815 ++have_conflict;
1816 }
1817 }
1818#undef OVERLAPS
1819 if (!have_conflict)
1820 break;
1821
1822 /* Discard Ack only for the _first_ iteration */
1823 if (first && discard && have_unacked) {
1824 dev_alert(DEV, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n",
1825 (unsigned long long)sector);
1826 inc_unacked(mdev);
1827 e->w.cb = e_send_discard_ack;
1828 list_add_tail(&e->w.list, &mdev->done_ee);
1829
1830 spin_unlock_irq(&mdev->req_lock);
1831
1832 /* we could probably send that P_DISCARD_ACK ourselves,
1833 * but I don't like the receiver using the msock */
1834
1835 put_ldev(mdev);
1836 wake_asender(mdev);
1837 finish_wait(&mdev->misc_wait, &wait);
1838 return TRUE;
1839 }
1840
1841 if (signal_pending(current)) {
1842 hlist_del_init(&e->colision);
1843
1844 spin_unlock_irq(&mdev->req_lock);
1845
1846 finish_wait(&mdev->misc_wait, &wait);
1847 goto out_interrupted;
1848 }
1849
1850 spin_unlock_irq(&mdev->req_lock);
1851 if (first) {
1852 first = 0;
1853 dev_alert(DEV, "Concurrent write! [W AFTERWARDS] "
1854 "sec=%llus\n", (unsigned long long)sector);
1855 } else if (discard) {
1856 /* we had none on the first iteration.
1857 * there must be none now. */
1858 D_ASSERT(have_unacked == 0);
1859 }
1860 schedule();
1861 spin_lock_irq(&mdev->req_lock);
1862 }
1863 finish_wait(&mdev->misc_wait, &wait);
1864 }
1865
1866 list_add(&e->w.list, &mdev->active_ee);
1867 spin_unlock_irq(&mdev->req_lock);
1868
1869 switch (mdev->net_conf->wire_protocol) {
1870 case DRBD_PROT_C:
1871 inc_unacked(mdev);
1872 /* corresponding dec_unacked() in e_end_block()
1873 * respective _drbd_clear_done_ee */
1874 break;
1875 case DRBD_PROT_B:
1876 /* I really don't like it that the receiver thread
1877 * sends on the msock, but anyways */
1878 drbd_send_ack(mdev, P_RECV_ACK, e);
1879 break;
1880 case DRBD_PROT_A:
1881 /* nothing to do */
1882 break;
1883 }
1884
1885 if (mdev->state.pdsk == D_DISKLESS) {
1886 /* In case we have the only disk of the cluster, */
1887 drbd_set_out_of_sync(mdev, e->sector, e->size);
1888 e->flags |= EE_CALL_AL_COMPLETE_IO;
1889 drbd_al_begin_io(mdev, e->sector);
1890 }
1891
1892 e->private_bio->bi_rw = rw;
1893 drbd_generic_make_request(mdev, DRBD_FAULT_DT_WR, e->private_bio);
1894 /* accounting done in endio */
1895
1896 maybe_kick_lo(mdev);
1897 return TRUE;
1898
1899out_interrupted:
1900 /* yes, the epoch_size now is imbalanced.
1901 * but we drop the connection anyways, so we don't have a chance to
1902 * receive a barrier... atomic_inc(&mdev->epoch_size); */
1903 put_ldev(mdev);
1904 drbd_free_ee(mdev, e);
1905 return FALSE;
1906}
1907
1908static int receive_DataRequest(struct drbd_conf *mdev, struct p_header *h)
1909{
1910 sector_t sector;
1911 const sector_t capacity = drbd_get_capacity(mdev->this_bdev);
1912 struct drbd_epoch_entry *e;
1913 struct digest_info *di = NULL;
1914 int size, digest_size;
1915 unsigned int fault_type;
1916 struct p_block_req *p =
1917 (struct p_block_req *)h;
1918 const int brps = sizeof(*p)-sizeof(*h);
1919
1920 if (drbd_recv(mdev, h->payload, brps) != brps)
1921 return FALSE;
1922
1923 sector = be64_to_cpu(p->sector);
1924 size = be32_to_cpu(p->blksize);
1925
1926 if (size <= 0 || (size & 0x1ff) != 0 || size > DRBD_MAX_SEGMENT_SIZE) {
1927 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
1928 (unsigned long long)sector, size);
1929 return FALSE;
1930 }
1931 if (sector + (size>>9) > capacity) {
1932 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
1933 (unsigned long long)sector, size);
1934 return FALSE;
1935 }
1936
1937 if (!get_ldev_if_state(mdev, D_UP_TO_DATE)) {
1938 if (__ratelimit(&drbd_ratelimit_state))
1939 dev_err(DEV, "Can not satisfy peer's read request, "
1940 "no local data.\n");
1941 drbd_send_ack_rp(mdev, h->command == P_DATA_REQUEST ? P_NEG_DREPLY :
1942 P_NEG_RS_DREPLY , p);
1943 return TRUE;
1944 }
1945
1946 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1947 * "criss-cross" setup, that might cause write-out on some other DRBD,
1948 * which in turn might block on the other node at this very place. */
1949 e = drbd_alloc_ee(mdev, p->block_id, sector, size, GFP_NOIO);
1950 if (!e) {
1951 put_ldev(mdev);
1952 return FALSE;
1953 }
1954
1955 e->private_bio->bi_rw = READ;
1956 e->private_bio->bi_end_io = drbd_endio_read_sec;
1957
1958 switch (h->command) {
1959 case P_DATA_REQUEST:
1960 e->w.cb = w_e_end_data_req;
1961 fault_type = DRBD_FAULT_DT_RD;
1962 break;
1963 case P_RS_DATA_REQUEST:
1964 e->w.cb = w_e_end_rsdata_req;
1965 fault_type = DRBD_FAULT_RS_RD;
1966 /* Eventually this should become asynchronously. Currently it
1967 * blocks the whole receiver just to delay the reading of a
1968 * resync data block.
1969 * the drbd_work_queue mechanism is made for this...
1970 */
1971 if (!drbd_rs_begin_io(mdev, sector)) {
1972 /* we have been interrupted,
1973 * probably connection lost! */
1974 D_ASSERT(signal_pending(current));
1975 goto out_free_e;
1976 }
1977 break;
1978
1979 case P_OV_REPLY:
1980 case P_CSUM_RS_REQUEST:
1981 fault_type = DRBD_FAULT_RS_RD;
1982 digest_size = h->length - brps ;
1983 di = kmalloc(sizeof(*di) + digest_size, GFP_NOIO);
1984 if (!di)
1985 goto out_free_e;
1986
1987 di->digest_size = digest_size;
1988 di->digest = (((char *)di)+sizeof(struct digest_info));
1989
1990 if (drbd_recv(mdev, di->digest, digest_size) != digest_size)
1991 goto out_free_e;
1992
1993 e->block_id = (u64)(unsigned long)di;
1994 if (h->command == P_CSUM_RS_REQUEST) {
1995 D_ASSERT(mdev->agreed_pro_version >= 89);
1996 e->w.cb = w_e_end_csum_rs_req;
1997 } else if (h->command == P_OV_REPLY) {
1998 e->w.cb = w_e_end_ov_reply;
1999 dec_rs_pending(mdev);
2000 break;
2001 }
2002
2003 if (!drbd_rs_begin_io(mdev, sector)) {
2004 /* we have been interrupted, probably connection lost! */
2005 D_ASSERT(signal_pending(current));
2006 goto out_free_e;
2007 }
2008 break;
2009
2010 case P_OV_REQUEST:
2011 if (mdev->state.conn >= C_CONNECTED &&
2012 mdev->state.conn != C_VERIFY_T)
2013 dev_warn(DEV, "ASSERT FAILED: got P_OV_REQUEST while being %s\n",
2014 drbd_conn_str(mdev->state.conn));
2015 if (mdev->ov_start_sector == ~(sector_t)0 &&
2016 mdev->agreed_pro_version >= 90) {
2017 mdev->ov_start_sector = sector;
2018 mdev->ov_position = sector;
2019 mdev->ov_left = mdev->rs_total - BM_SECT_TO_BIT(sector);
2020 dev_info(DEV, "Online Verify start sector: %llu\n",
2021 (unsigned long long)sector);
2022 }
2023 e->w.cb = w_e_end_ov_req;
2024 fault_type = DRBD_FAULT_RS_RD;
2025 /* Eventually this should become asynchronous. Currently it
2026 * blocks the whole receiver just to delay the reading of a
2027 * resync data block.
2028 * the drbd_work_queue mechanism is made for this...
2029 */
2030 if (!drbd_rs_begin_io(mdev, sector)) {
2031 /* we have been interrupted,
2032 * probably connection lost! */
2033 D_ASSERT(signal_pending(current));
2034 goto out_free_e;
2035 }
2036 break;
2037
2038
2039 default:
2040 dev_err(DEV, "unexpected command (%s) in receive_DataRequest\n",
2041 cmdname(h->command));
2042 fault_type = DRBD_FAULT_MAX;
2043 }
2044
2045 spin_lock_irq(&mdev->req_lock);
2046 list_add(&e->w.list, &mdev->read_ee);
2047 spin_unlock_irq(&mdev->req_lock);
2048
2049 inc_unacked(mdev);
2050
2051 drbd_generic_make_request(mdev, fault_type, e->private_bio);
2052 maybe_kick_lo(mdev);
2053
2054 return TRUE;
2055
2056out_free_e:
2057 kfree(di);
2058 put_ldev(mdev);
2059 drbd_free_ee(mdev, e);
2060 return FALSE;
2061}
2062
2063static int drbd_asb_recover_0p(struct drbd_conf *mdev) __must_hold(local)
2064{
2065 int self, peer, rv = -100;
2066 unsigned long ch_self, ch_peer;
2067
2068 self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2069 peer = mdev->p_uuid[UI_BITMAP] & 1;
2070
2071 ch_peer = mdev->p_uuid[UI_SIZE];
2072 ch_self = mdev->comm_bm_set;
2073
2074 switch (mdev->net_conf->after_sb_0p) {
2075 case ASB_CONSENSUS:
2076 case ASB_DISCARD_SECONDARY:
2077 case ASB_CALL_HELPER:
2078 dev_err(DEV, "Configuration error.\n");
2079 break;
2080 case ASB_DISCONNECT:
2081 break;
2082 case ASB_DISCARD_YOUNGER_PRI:
2083 if (self == 0 && peer == 1) {
2084 rv = -1;
2085 break;
2086 }
2087 if (self == 1 && peer == 0) {
2088 rv = 1;
2089 break;
2090 }
2091 /* Else fall through to one of the other strategies... */
2092 case ASB_DISCARD_OLDER_PRI:
2093 if (self == 0 && peer == 1) {
2094 rv = 1;
2095 break;
2096 }
2097 if (self == 1 && peer == 0) {
2098 rv = -1;
2099 break;
2100 }
2101 /* Else fall through to one of the other strategies... */
2102 dev_warn(DEV, "Discard younger/older primary did not find a decision\n"
2103 "Using discard-least-changes instead\n");
2104 case ASB_DISCARD_ZERO_CHG:
2105 if (ch_peer == 0 && ch_self == 0) {
2106 rv = test_bit(DISCARD_CONCURRENT, &mdev->flags)
2107 ? -1 : 1;
2108 break;
2109 } else {
2110 if (ch_peer == 0) { rv = 1; break; }
2111 if (ch_self == 0) { rv = -1; break; }
2112 }
2113 if (mdev->net_conf->after_sb_0p == ASB_DISCARD_ZERO_CHG)
2114 break;
2115 case ASB_DISCARD_LEAST_CHG:
2116 if (ch_self < ch_peer)
2117 rv = -1;
2118 else if (ch_self > ch_peer)
2119 rv = 1;
2120 else /* ( ch_self == ch_peer ) */
2121 /* Well, then use something else. */
2122 rv = test_bit(DISCARD_CONCURRENT, &mdev->flags)
2123 ? -1 : 1;
2124 break;
2125 case ASB_DISCARD_LOCAL:
2126 rv = -1;
2127 break;
2128 case ASB_DISCARD_REMOTE:
2129 rv = 1;
2130 }
2131
2132 return rv;
2133}
2134
2135static int drbd_asb_recover_1p(struct drbd_conf *mdev) __must_hold(local)
2136{
2137 int self, peer, hg, rv = -100;
2138
2139 self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2140 peer = mdev->p_uuid[UI_BITMAP] & 1;
2141
2142 switch (mdev->net_conf->after_sb_1p) {
2143 case ASB_DISCARD_YOUNGER_PRI:
2144 case ASB_DISCARD_OLDER_PRI:
2145 case ASB_DISCARD_LEAST_CHG:
2146 case ASB_DISCARD_LOCAL:
2147 case ASB_DISCARD_REMOTE:
2148 dev_err(DEV, "Configuration error.\n");
2149 break;
2150 case ASB_DISCONNECT:
2151 break;
2152 case ASB_CONSENSUS:
2153 hg = drbd_asb_recover_0p(mdev);
2154 if (hg == -1 && mdev->state.role == R_SECONDARY)
2155 rv = hg;
2156 if (hg == 1 && mdev->state.role == R_PRIMARY)
2157 rv = hg;
2158 break;
2159 case ASB_VIOLENTLY:
2160 rv = drbd_asb_recover_0p(mdev);
2161 break;
2162 case ASB_DISCARD_SECONDARY:
2163 return mdev->state.role == R_PRIMARY ? 1 : -1;
2164 case ASB_CALL_HELPER:
2165 hg = drbd_asb_recover_0p(mdev);
2166 if (hg == -1 && mdev->state.role == R_PRIMARY) {
2167 self = drbd_set_role(mdev, R_SECONDARY, 0);
2168 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2169 * we might be here in C_WF_REPORT_PARAMS which is transient.
2170 * we do not need to wait for the after state change work either. */
2171 self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY));
2172 if (self != SS_SUCCESS) {
2173 drbd_khelper(mdev, "pri-lost-after-sb");
2174 } else {
2175 dev_warn(DEV, "Successfully gave up primary role.\n");
2176 rv = hg;
2177 }
2178 } else
2179 rv = hg;
2180 }
2181
2182 return rv;
2183}
2184
2185static int drbd_asb_recover_2p(struct drbd_conf *mdev) __must_hold(local)
2186{
2187 int self, peer, hg, rv = -100;
2188
2189 self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2190 peer = mdev->p_uuid[UI_BITMAP] & 1;
2191
2192 switch (mdev->net_conf->after_sb_2p) {
2193 case ASB_DISCARD_YOUNGER_PRI:
2194 case ASB_DISCARD_OLDER_PRI:
2195 case ASB_DISCARD_LEAST_CHG:
2196 case ASB_DISCARD_LOCAL:
2197 case ASB_DISCARD_REMOTE:
2198 case ASB_CONSENSUS:
2199 case ASB_DISCARD_SECONDARY:
2200 dev_err(DEV, "Configuration error.\n");
2201 break;
2202 case ASB_VIOLENTLY:
2203 rv = drbd_asb_recover_0p(mdev);
2204 break;
2205 case ASB_DISCONNECT:
2206 break;
2207 case ASB_CALL_HELPER:
2208 hg = drbd_asb_recover_0p(mdev);
2209 if (hg == -1) {
2210 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2211 * we might be here in C_WF_REPORT_PARAMS which is transient.
2212 * we do not need to wait for the after state change work either. */
2213 self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY));
2214 if (self != SS_SUCCESS) {
2215 drbd_khelper(mdev, "pri-lost-after-sb");
2216 } else {
2217 dev_warn(DEV, "Successfully gave up primary role.\n");
2218 rv = hg;
2219 }
2220 } else
2221 rv = hg;
2222 }
2223
2224 return rv;
2225}
2226
2227static void drbd_uuid_dump(struct drbd_conf *mdev, char *text, u64 *uuid,
2228 u64 bits, u64 flags)
2229{
2230 if (!uuid) {
2231 dev_info(DEV, "%s uuid info vanished while I was looking!\n", text);
2232 return;
2233 }
2234 dev_info(DEV, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2235 text,
2236 (unsigned long long)uuid[UI_CURRENT],
2237 (unsigned long long)uuid[UI_BITMAP],
2238 (unsigned long long)uuid[UI_HISTORY_START],
2239 (unsigned long long)uuid[UI_HISTORY_END],
2240 (unsigned long long)bits,
2241 (unsigned long long)flags);
2242}
2243
2244/*
2245 100 after split brain try auto recover
2246 2 C_SYNC_SOURCE set BitMap
2247 1 C_SYNC_SOURCE use BitMap
2248 0 no Sync
2249 -1 C_SYNC_TARGET use BitMap
2250 -2 C_SYNC_TARGET set BitMap
2251 -100 after split brain, disconnect
2252-1000 unrelated data
2253 */
2254static int drbd_uuid_compare(struct drbd_conf *mdev, int *rule_nr) __must_hold(local)
2255{
2256 u64 self, peer;
2257 int i, j;
2258
2259 self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2260 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2261
2262 *rule_nr = 10;
2263 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
2264 return 0;
2265
2266 *rule_nr = 20;
2267 if ((self == UUID_JUST_CREATED || self == (u64)0) &&
2268 peer != UUID_JUST_CREATED)
2269 return -2;
2270
2271 *rule_nr = 30;
2272 if (self != UUID_JUST_CREATED &&
2273 (peer == UUID_JUST_CREATED || peer == (u64)0))
2274 return 2;
2275
2276 if (self == peer) {
2277 int rct, dc; /* roles at crash time */
2278
2279 if (mdev->p_uuid[UI_BITMAP] == (u64)0 && mdev->ldev->md.uuid[UI_BITMAP] != (u64)0) {
2280
2281 if (mdev->agreed_pro_version < 91)
2282 return -1001;
2283
2284 if ((mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
2285 (mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
2286 dev_info(DEV, "was SyncSource, missed the resync finished event, corrected myself:\n");
2287 drbd_uuid_set_bm(mdev, 0UL);
2288
2289 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid,
2290 mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0);
2291 *rule_nr = 34;
2292 } else {
2293 dev_info(DEV, "was SyncSource (peer failed to write sync_uuid)\n");
2294 *rule_nr = 36;
2295 }
2296
2297 return 1;
2298 }
2299
2300 if (mdev->ldev->md.uuid[UI_BITMAP] == (u64)0 && mdev->p_uuid[UI_BITMAP] != (u64)0) {
2301
2302 if (mdev->agreed_pro_version < 91)
2303 return -1001;
2304
2305 if ((mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_BITMAP] & ~((u64)1)) &&
2306 (mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
2307 dev_info(DEV, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2308
2309 mdev->p_uuid[UI_HISTORY_START + 1] = mdev->p_uuid[UI_HISTORY_START];
2310 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_BITMAP];
2311 mdev->p_uuid[UI_BITMAP] = 0UL;
2312
2313 drbd_uuid_dump(mdev, "peer", mdev->p_uuid, mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2314 *rule_nr = 35;
2315 } else {
2316 dev_info(DEV, "was SyncTarget (failed to write sync_uuid)\n");
2317 *rule_nr = 37;
2318 }
2319
2320 return -1;
2321 }
2322
2323 /* Common power [off|failure] */
2324 rct = (test_bit(CRASHED_PRIMARY, &mdev->flags) ? 1 : 0) +
2325 (mdev->p_uuid[UI_FLAGS] & 2);
2326 /* lowest bit is set when we were primary,
2327 * next bit (weight 2) is set when peer was primary */
2328 *rule_nr = 40;
2329
2330 switch (rct) {
2331 case 0: /* !self_pri && !peer_pri */ return 0;
2332 case 1: /* self_pri && !peer_pri */ return 1;
2333 case 2: /* !self_pri && peer_pri */ return -1;
2334 case 3: /* self_pri && peer_pri */
2335 dc = test_bit(DISCARD_CONCURRENT, &mdev->flags);
2336 return dc ? -1 : 1;
2337 }
2338 }
2339
2340 *rule_nr = 50;
2341 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2342 if (self == peer)
2343 return -1;
2344
2345 *rule_nr = 51;
2346 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1);
2347 if (self == peer) {
2348 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2349 peer = mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1);
2350 if (self == peer) {
2351 /* The last P_SYNC_UUID did not get though. Undo the last start of
2352 resync as sync source modifications of the peer's UUIDs. */
2353
2354 if (mdev->agreed_pro_version < 91)
2355 return -1001;
2356
2357 mdev->p_uuid[UI_BITMAP] = mdev->p_uuid[UI_HISTORY_START];
2358 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_HISTORY_START + 1];
2359 return -1;
2360 }
2361 }
2362
2363 *rule_nr = 60;
2364 self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2365 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2366 peer = mdev->p_uuid[i] & ~((u64)1);
2367 if (self == peer)
2368 return -2;
2369 }
2370
2371 *rule_nr = 70;
2372 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2373 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2374 if (self == peer)
2375 return 1;
2376
2377 *rule_nr = 71;
2378 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2379 if (self == peer) {
2380 self = mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1);
2381 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1);
2382 if (self == peer) {
2383 /* The last P_SYNC_UUID did not get though. Undo the last start of
2384 resync as sync source modifications of our UUIDs. */
2385
2386 if (mdev->agreed_pro_version < 91)
2387 return -1001;
2388
2389 _drbd_uuid_set(mdev, UI_BITMAP, mdev->ldev->md.uuid[UI_HISTORY_START]);
2390 _drbd_uuid_set(mdev, UI_HISTORY_START, mdev->ldev->md.uuid[UI_HISTORY_START + 1]);
2391
2392 dev_info(DEV, "Undid last start of resync:\n");
2393
2394 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid,
2395 mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0);
2396
2397 return 1;
2398 }
2399 }
2400
2401
2402 *rule_nr = 80;
2403 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2404 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2405 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2406 if (self == peer)
2407 return 2;
2408 }
2409
2410 *rule_nr = 90;
2411 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2412 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2413 if (self == peer && self != ((u64)0))
2414 return 100;
2415
2416 *rule_nr = 100;
2417 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2418 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2419 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
2420 peer = mdev->p_uuid[j] & ~((u64)1);
2421 if (self == peer)
2422 return -100;
2423 }
2424 }
2425
2426 return -1000;
2427}
2428
2429/* drbd_sync_handshake() returns the new conn state on success, or
2430 CONN_MASK (-1) on failure.
2431 */
2432static enum drbd_conns drbd_sync_handshake(struct drbd_conf *mdev, enum drbd_role peer_role,
2433 enum drbd_disk_state peer_disk) __must_hold(local)
2434{
2435 int hg, rule_nr;
2436 enum drbd_conns rv = C_MASK;
2437 enum drbd_disk_state mydisk;
2438
2439 mydisk = mdev->state.disk;
2440 if (mydisk == D_NEGOTIATING)
2441 mydisk = mdev->new_state_tmp.disk;
2442
2443 dev_info(DEV, "drbd_sync_handshake:\n");
2444 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, mdev->comm_bm_set, 0);
2445 drbd_uuid_dump(mdev, "peer", mdev->p_uuid,
2446 mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2447
2448 hg = drbd_uuid_compare(mdev, &rule_nr);
2449
2450 dev_info(DEV, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
2451
2452 if (hg == -1000) {
2453 dev_alert(DEV, "Unrelated data, aborting!\n");
2454 return C_MASK;
2455 }
2456 if (hg == -1001) {
2457 dev_alert(DEV, "To resolve this both sides have to support at least protocol\n");
2458 return C_MASK;
2459 }
2460
2461 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
2462 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
2463 int f = (hg == -100) || abs(hg) == 2;
2464 hg = mydisk > D_INCONSISTENT ? 1 : -1;
2465 if (f)
2466 hg = hg*2;
2467 dev_info(DEV, "Becoming sync %s due to disk states.\n",
2468 hg > 0 ? "source" : "target");
2469 }
2470
2471 if (hg == 100 || (hg == -100 && mdev->net_conf->always_asbp)) {
2472 int pcount = (mdev->state.role == R_PRIMARY)
2473 + (peer_role == R_PRIMARY);
2474 int forced = (hg == -100);
2475
2476 switch (pcount) {
2477 case 0:
2478 hg = drbd_asb_recover_0p(mdev);
2479 break;
2480 case 1:
2481 hg = drbd_asb_recover_1p(mdev);
2482 break;
2483 case 2:
2484 hg = drbd_asb_recover_2p(mdev);
2485 break;
2486 }
2487 if (abs(hg) < 100) {
2488 dev_warn(DEV, "Split-Brain detected, %d primaries, "
2489 "automatically solved. Sync from %s node\n",
2490 pcount, (hg < 0) ? "peer" : "this");
2491 if (forced) {
2492 dev_warn(DEV, "Doing a full sync, since"
2493 " UUIDs where ambiguous.\n");
2494 hg = hg*2;
2495 }
2496 }
2497 }
2498
2499 if (hg == -100) {
2500 if (mdev->net_conf->want_lose && !(mdev->p_uuid[UI_FLAGS]&1))
2501 hg = -1;
2502 if (!mdev->net_conf->want_lose && (mdev->p_uuid[UI_FLAGS]&1))
2503 hg = 1;
2504
2505 if (abs(hg) < 100)
2506 dev_warn(DEV, "Split-Brain detected, manually solved. "
2507 "Sync from %s node\n",
2508 (hg < 0) ? "peer" : "this");
2509 }
2510
2511 if (hg == -100) {
2512 dev_alert(DEV, "Split-Brain detected, dropping connection!\n");
2513 drbd_khelper(mdev, "split-brain");
2514 return C_MASK;
2515 }
2516
2517 if (hg > 0 && mydisk <= D_INCONSISTENT) {
2518 dev_err(DEV, "I shall become SyncSource, but I am inconsistent!\n");
2519 return C_MASK;
2520 }
2521
2522 if (hg < 0 && /* by intention we do not use mydisk here. */
2523 mdev->state.role == R_PRIMARY && mdev->state.disk >= D_CONSISTENT) {
2524 switch (mdev->net_conf->rr_conflict) {
2525 case ASB_CALL_HELPER:
2526 drbd_khelper(mdev, "pri-lost");
2527 /* fall through */
2528 case ASB_DISCONNECT:
2529 dev_err(DEV, "I shall become SyncTarget, but I am primary!\n");
2530 return C_MASK;
2531 case ASB_VIOLENTLY:
2532 dev_warn(DEV, "Becoming SyncTarget, violating the stable-data"
2533 "assumption\n");
2534 }
2535 }
2536
2537 if (abs(hg) >= 2) {
2538 dev_info(DEV, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2539 if (drbd_bitmap_io(mdev, &drbd_bmio_set_n_write, "set_n_write from sync_handshake"))
2540 return C_MASK;
2541 }
2542
2543 if (hg > 0) { /* become sync source. */
2544 rv = C_WF_BITMAP_S;
2545 } else if (hg < 0) { /* become sync target */
2546 rv = C_WF_BITMAP_T;
2547 } else {
2548 rv = C_CONNECTED;
2549 if (drbd_bm_total_weight(mdev)) {
2550 dev_info(DEV, "No resync, but %lu bits in bitmap!\n",
2551 drbd_bm_total_weight(mdev));
2552 }
2553 }
2554
2555 return rv;
2556}
2557
2558/* returns 1 if invalid */
2559static int cmp_after_sb(enum drbd_after_sb_p peer, enum drbd_after_sb_p self)
2560{
2561 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2562 if ((peer == ASB_DISCARD_REMOTE && self == ASB_DISCARD_LOCAL) ||
2563 (self == ASB_DISCARD_REMOTE && peer == ASB_DISCARD_LOCAL))
2564 return 0;
2565
2566 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2567 if (peer == ASB_DISCARD_REMOTE || peer == ASB_DISCARD_LOCAL ||
2568 self == ASB_DISCARD_REMOTE || self == ASB_DISCARD_LOCAL)
2569 return 1;
2570
2571 /* everything else is valid if they are equal on both sides. */
2572 if (peer == self)
2573 return 0;
2574
2575 /* everything es is invalid. */
2576 return 1;
2577}
2578
2579static int receive_protocol(struct drbd_conf *mdev, struct p_header *h)
2580{
2581 struct p_protocol *p = (struct p_protocol *)h;
2582 int header_size, data_size;
2583 int p_proto, p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
2584 int p_want_lose, p_two_primaries;
2585 char p_integrity_alg[SHARED_SECRET_MAX] = "";
2586
2587 header_size = sizeof(*p) - sizeof(*h);
2588 data_size = h->length - header_size;
2589
2590 if (drbd_recv(mdev, h->payload, header_size) != header_size)
2591 return FALSE;
2592
2593 p_proto = be32_to_cpu(p->protocol);
2594 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
2595 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
2596 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
2597 p_want_lose = be32_to_cpu(p->want_lose);
2598 p_two_primaries = be32_to_cpu(p->two_primaries);
2599
2600 if (p_proto != mdev->net_conf->wire_protocol) {
2601 dev_err(DEV, "incompatible communication protocols\n");
2602 goto disconnect;
2603 }
2604
2605 if (cmp_after_sb(p_after_sb_0p, mdev->net_conf->after_sb_0p)) {
2606 dev_err(DEV, "incompatible after-sb-0pri settings\n");
2607 goto disconnect;
2608 }
2609
2610 if (cmp_after_sb(p_after_sb_1p, mdev->net_conf->after_sb_1p)) {
2611 dev_err(DEV, "incompatible after-sb-1pri settings\n");
2612 goto disconnect;
2613 }
2614
2615 if (cmp_after_sb(p_after_sb_2p, mdev->net_conf->after_sb_2p)) {
2616 dev_err(DEV, "incompatible after-sb-2pri settings\n");
2617 goto disconnect;
2618 }
2619
2620 if (p_want_lose && mdev->net_conf->want_lose) {
2621 dev_err(DEV, "both sides have the 'want_lose' flag set\n");
2622 goto disconnect;
2623 }
2624
2625 if (p_two_primaries != mdev->net_conf->two_primaries) {
2626 dev_err(DEV, "incompatible setting of the two-primaries options\n");
2627 goto disconnect;
2628 }
2629
2630 if (mdev->agreed_pro_version >= 87) {
2631 unsigned char *my_alg = mdev->net_conf->integrity_alg;
2632
2633 if (drbd_recv(mdev, p_integrity_alg, data_size) != data_size)
2634 return FALSE;
2635
2636 p_integrity_alg[SHARED_SECRET_MAX-1] = 0;
2637 if (strcmp(p_integrity_alg, my_alg)) {
2638 dev_err(DEV, "incompatible setting of the data-integrity-alg\n");
2639 goto disconnect;
2640 }
2641 dev_info(DEV, "data-integrity-alg: %s\n",
2642 my_alg[0] ? my_alg : (unsigned char *)"<not-used>");
2643 }
2644
2645 return TRUE;
2646
2647disconnect:
2648 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2649 return FALSE;
2650}
2651
2652/* helper function
2653 * input: alg name, feature name
2654 * return: NULL (alg name was "")
2655 * ERR_PTR(error) if something goes wrong
2656 * or the crypto hash ptr, if it worked out ok. */
2657struct crypto_hash *drbd_crypto_alloc_digest_safe(const struct drbd_conf *mdev,
2658 const char *alg, const char *name)
2659{
2660 struct crypto_hash *tfm;
2661
2662 if (!alg[0])
2663 return NULL;
2664
2665 tfm = crypto_alloc_hash(alg, 0, CRYPTO_ALG_ASYNC);
2666 if (IS_ERR(tfm)) {
2667 dev_err(DEV, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2668 alg, name, PTR_ERR(tfm));
2669 return tfm;
2670 }
2671 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm))) {
2672 crypto_free_hash(tfm);
2673 dev_err(DEV, "\"%s\" is not a digest (%s)\n", alg, name);
2674 return ERR_PTR(-EINVAL);
2675 }
2676 return tfm;
2677}
2678
2679static int receive_SyncParam(struct drbd_conf *mdev, struct p_header *h)
2680{
2681 int ok = TRUE;
2682 struct p_rs_param_89 *p = (struct p_rs_param_89 *)h;
2683 unsigned int header_size, data_size, exp_max_sz;
2684 struct crypto_hash *verify_tfm = NULL;
2685 struct crypto_hash *csums_tfm = NULL;
2686 const int apv = mdev->agreed_pro_version;
2687
2688 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
2689 : apv == 88 ? sizeof(struct p_rs_param)
2690 + SHARED_SECRET_MAX
2691 : /* 89 */ sizeof(struct p_rs_param_89);
2692
2693 if (h->length > exp_max_sz) {
2694 dev_err(DEV, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2695 h->length, exp_max_sz);
2696 return FALSE;
2697 }
2698
2699 if (apv <= 88) {
2700 header_size = sizeof(struct p_rs_param) - sizeof(*h);
2701 data_size = h->length - header_size;
2702 } else /* apv >= 89 */ {
2703 header_size = sizeof(struct p_rs_param_89) - sizeof(*h);
2704 data_size = h->length - header_size;
2705 D_ASSERT(data_size == 0);
2706 }
2707
2708 /* initialize verify_alg and csums_alg */
2709 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
2710
2711 if (drbd_recv(mdev, h->payload, header_size) != header_size)
2712 return FALSE;
2713
2714 mdev->sync_conf.rate = be32_to_cpu(p->rate);
2715
2716 if (apv >= 88) {
2717 if (apv == 88) {
2718 if (data_size > SHARED_SECRET_MAX) {
2719 dev_err(DEV, "verify-alg too long, "
2720 "peer wants %u, accepting only %u byte\n",
2721 data_size, SHARED_SECRET_MAX);
2722 return FALSE;
2723 }
2724
2725 if (drbd_recv(mdev, p->verify_alg, data_size) != data_size)
2726 return FALSE;
2727
2728 /* we expect NUL terminated string */
2729 /* but just in case someone tries to be evil */
2730 D_ASSERT(p->verify_alg[data_size-1] == 0);
2731 p->verify_alg[data_size-1] = 0;
2732
2733 } else /* apv >= 89 */ {
2734 /* we still expect NUL terminated strings */
2735 /* but just in case someone tries to be evil */
2736 D_ASSERT(p->verify_alg[SHARED_SECRET_MAX-1] == 0);
2737 D_ASSERT(p->csums_alg[SHARED_SECRET_MAX-1] == 0);
2738 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
2739 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
2740 }
2741
2742 if (strcmp(mdev->sync_conf.verify_alg, p->verify_alg)) {
2743 if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2744 dev_err(DEV, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2745 mdev->sync_conf.verify_alg, p->verify_alg);
2746 goto disconnect;
2747 }
2748 verify_tfm = drbd_crypto_alloc_digest_safe(mdev,
2749 p->verify_alg, "verify-alg");
2750 if (IS_ERR(verify_tfm)) {
2751 verify_tfm = NULL;
2752 goto disconnect;
2753 }
2754 }
2755
2756 if (apv >= 89 && strcmp(mdev->sync_conf.csums_alg, p->csums_alg)) {
2757 if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2758 dev_err(DEV, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2759 mdev->sync_conf.csums_alg, p->csums_alg);
2760 goto disconnect;
2761 }
2762 csums_tfm = drbd_crypto_alloc_digest_safe(mdev,
2763 p->csums_alg, "csums-alg");
2764 if (IS_ERR(csums_tfm)) {
2765 csums_tfm = NULL;
2766 goto disconnect;
2767 }
2768 }
2769
2770
2771 spin_lock(&mdev->peer_seq_lock);
2772 /* lock against drbd_nl_syncer_conf() */
2773 if (verify_tfm) {
2774 strcpy(mdev->sync_conf.verify_alg, p->verify_alg);
2775 mdev->sync_conf.verify_alg_len = strlen(p->verify_alg) + 1;
2776 crypto_free_hash(mdev->verify_tfm);
2777 mdev->verify_tfm = verify_tfm;
2778 dev_info(DEV, "using verify-alg: \"%s\"\n", p->verify_alg);
2779 }
2780 if (csums_tfm) {
2781 strcpy(mdev->sync_conf.csums_alg, p->csums_alg);
2782 mdev->sync_conf.csums_alg_len = strlen(p->csums_alg) + 1;
2783 crypto_free_hash(mdev->csums_tfm);
2784 mdev->csums_tfm = csums_tfm;
2785 dev_info(DEV, "using csums-alg: \"%s\"\n", p->csums_alg);
2786 }
2787 spin_unlock(&mdev->peer_seq_lock);
2788 }
2789
2790 return ok;
2791disconnect:
2792 /* just for completeness: actually not needed,
2793 * as this is not reached if csums_tfm was ok. */
2794 crypto_free_hash(csums_tfm);
2795 /* but free the verify_tfm again, if csums_tfm did not work out */
2796 crypto_free_hash(verify_tfm);
2797 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2798 return FALSE;
2799}
2800
2801static void drbd_setup_order_type(struct drbd_conf *mdev, int peer)
2802{
2803 /* sorry, we currently have no working implementation
2804 * of distributed TCQ */
2805}
2806
2807/* warn if the arguments differ by more than 12.5% */
2808static void warn_if_differ_considerably(struct drbd_conf *mdev,
2809 const char *s, sector_t a, sector_t b)
2810{
2811 sector_t d;
2812 if (a == 0 || b == 0)
2813 return;
2814 d = (a > b) ? (a - b) : (b - a);
2815 if (d > (a>>3) || d > (b>>3))
2816 dev_warn(DEV, "Considerable difference in %s: %llus vs. %llus\n", s,
2817 (unsigned long long)a, (unsigned long long)b);
2818}
2819
2820static int receive_sizes(struct drbd_conf *mdev, struct p_header *h)
2821{
2822 struct p_sizes *p = (struct p_sizes *)h;
2823 enum determine_dev_size dd = unchanged;
2824 unsigned int max_seg_s;
2825 sector_t p_size, p_usize, my_usize;
2826 int ldsc = 0; /* local disk size changed */
2827 enum drbd_conns nconn;
2828
2829 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
2830 if (drbd_recv(mdev, h->payload, h->length) != h->length)
2831 return FALSE;
2832
2833 p_size = be64_to_cpu(p->d_size);
2834 p_usize = be64_to_cpu(p->u_size);
2835
2836 if (p_size == 0 && mdev->state.disk == D_DISKLESS) {
2837 dev_err(DEV, "some backing storage is needed\n");
2838 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2839 return FALSE;
2840 }
2841
2842 /* just store the peer's disk size for now.
2843 * we still need to figure out whether we accept that. */
2844 mdev->p_size = p_size;
2845
2846#define min_not_zero(l, r) (l == 0) ? r : ((r == 0) ? l : min(l, r))
2847 if (get_ldev(mdev)) {
2848 warn_if_differ_considerably(mdev, "lower level device sizes",
2849 p_size, drbd_get_max_capacity(mdev->ldev));
2850 warn_if_differ_considerably(mdev, "user requested size",
2851 p_usize, mdev->ldev->dc.disk_size);
2852
2853 /* if this is the first connect, or an otherwise expected
2854 * param exchange, choose the minimum */
2855 if (mdev->state.conn == C_WF_REPORT_PARAMS)
2856 p_usize = min_not_zero((sector_t)mdev->ldev->dc.disk_size,
2857 p_usize);
2858
2859 my_usize = mdev->ldev->dc.disk_size;
2860
2861 if (mdev->ldev->dc.disk_size != p_usize) {
2862 mdev->ldev->dc.disk_size = p_usize;
2863 dev_info(DEV, "Peer sets u_size to %lu sectors\n",
2864 (unsigned long)mdev->ldev->dc.disk_size);
2865 }
2866
2867 /* Never shrink a device with usable data during connect.
2868 But allow online shrinking if we are connected. */
2869 if (drbd_new_dev_size(mdev, mdev->ldev) <
2870 drbd_get_capacity(mdev->this_bdev) &&
2871 mdev->state.disk >= D_OUTDATED &&
2872 mdev->state.conn < C_CONNECTED) {
2873 dev_err(DEV, "The peer's disk size is too small!\n");
2874 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2875 mdev->ldev->dc.disk_size = my_usize;
2876 put_ldev(mdev);
2877 return FALSE;
2878 }
2879 put_ldev(mdev);
2880 }
2881#undef min_not_zero
2882
2883 if (get_ldev(mdev)) {
2884 dd = drbd_determin_dev_size(mdev);
2885 put_ldev(mdev);
2886 if (dd == dev_size_error)
2887 return FALSE;
2888 drbd_md_sync(mdev);
2889 } else {
2890 /* I am diskless, need to accept the peer's size. */
2891 drbd_set_my_capacity(mdev, p_size);
2892 }
2893
2894 if (mdev->p_uuid && mdev->state.conn <= C_CONNECTED && get_ldev(mdev)) {
2895 nconn = drbd_sync_handshake(mdev,
2896 mdev->state.peer, mdev->state.pdsk);
2897 put_ldev(mdev);
2898
2899 if (nconn == C_MASK) {
2900 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2901 return FALSE;
2902 }
2903
2904 if (drbd_request_state(mdev, NS(conn, nconn)) < SS_SUCCESS) {
2905 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2906 return FALSE;
2907 }
2908 }
2909
2910 if (get_ldev(mdev)) {
2911 if (mdev->ldev->known_size != drbd_get_capacity(mdev->ldev->backing_bdev)) {
2912 mdev->ldev->known_size = drbd_get_capacity(mdev->ldev->backing_bdev);
2913 ldsc = 1;
2914 }
2915
2916 max_seg_s = be32_to_cpu(p->max_segment_size);
2917 if (max_seg_s != queue_max_segment_size(mdev->rq_queue))
2918 drbd_setup_queue_param(mdev, max_seg_s);
2919
2920 drbd_setup_order_type(mdev, be32_to_cpu(p->queue_order_type));
2921 put_ldev(mdev);
2922 }
2923
2924 if (mdev->state.conn > C_WF_REPORT_PARAMS) {
2925 if (be64_to_cpu(p->c_size) !=
2926 drbd_get_capacity(mdev->this_bdev) || ldsc) {
2927 /* we have different sizes, probably peer
2928 * needs to know my new size... */
2929 drbd_send_sizes(mdev, 0);
2930 }
2931 if (test_and_clear_bit(RESIZE_PENDING, &mdev->flags) ||
2932 (dd == grew && mdev->state.conn == C_CONNECTED)) {
2933 if (mdev->state.pdsk >= D_INCONSISTENT &&
2934 mdev->state.disk >= D_INCONSISTENT)
2935 resync_after_online_grow(mdev);
2936 else
2937 set_bit(RESYNC_AFTER_NEG, &mdev->flags);
2938 }
2939 }
2940
2941 return TRUE;
2942}
2943
2944static int receive_uuids(struct drbd_conf *mdev, struct p_header *h)
2945{
2946 struct p_uuids *p = (struct p_uuids *)h;
2947 u64 *p_uuid;
2948 int i;
2949
2950 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
2951 if (drbd_recv(mdev, h->payload, h->length) != h->length)
2952 return FALSE;
2953
2954 p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
2955
2956 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
2957 p_uuid[i] = be64_to_cpu(p->uuid[i]);
2958
2959 kfree(mdev->p_uuid);
2960 mdev->p_uuid = p_uuid;
2961
2962 if (mdev->state.conn < C_CONNECTED &&
2963 mdev->state.disk < D_INCONSISTENT &&
2964 mdev->state.role == R_PRIMARY &&
2965 (mdev->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
2966 dev_err(DEV, "Can only connect to data with current UUID=%016llX\n",
2967 (unsigned long long)mdev->ed_uuid);
2968 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2969 return FALSE;
2970 }
2971
2972 if (get_ldev(mdev)) {
2973 int skip_initial_sync =
2974 mdev->state.conn == C_CONNECTED &&
2975 mdev->agreed_pro_version >= 90 &&
2976 mdev->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
2977 (p_uuid[UI_FLAGS] & 8);
2978 if (skip_initial_sync) {
2979 dev_info(DEV, "Accepted new current UUID, preparing to skip initial sync\n");
2980 drbd_bitmap_io(mdev, &drbd_bmio_clear_n_write,
2981 "clear_n_write from receive_uuids");
2982 _drbd_uuid_set(mdev, UI_CURRENT, p_uuid[UI_CURRENT]);
2983 _drbd_uuid_set(mdev, UI_BITMAP, 0);
2984 _drbd_set_state(_NS2(mdev, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
2985 CS_VERBOSE, NULL);
2986 drbd_md_sync(mdev);
2987 }
2988 put_ldev(mdev);
2989 }
2990
2991 /* Before we test for the disk state, we should wait until an eventually
2992 ongoing cluster wide state change is finished. That is important if
2993 we are primary and are detaching from our disk. We need to see the
2994 new disk state... */
2995 wait_event(mdev->misc_wait, !test_bit(CLUSTER_ST_CHANGE, &mdev->flags));
2996 if (mdev->state.conn >= C_CONNECTED && mdev->state.disk < D_INCONSISTENT)
2997 drbd_set_ed_uuid(mdev, p_uuid[UI_CURRENT]);
2998
2999 return TRUE;
3000}
3001
3002/**
3003 * convert_state() - Converts the peer's view of the cluster state to our point of view
3004 * @ps: The state as seen by the peer.
3005 */
3006static union drbd_state convert_state(union drbd_state ps)
3007{
3008 union drbd_state ms;
3009
3010 static enum drbd_conns c_tab[] = {
3011 [C_CONNECTED] = C_CONNECTED,
3012
3013 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
3014 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
3015 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
3016 [C_VERIFY_S] = C_VERIFY_T,
3017 [C_MASK] = C_MASK,
3018 };
3019
3020 ms.i = ps.i;
3021
3022 ms.conn = c_tab[ps.conn];
3023 ms.peer = ps.role;
3024 ms.role = ps.peer;
3025 ms.pdsk = ps.disk;
3026 ms.disk = ps.pdsk;
3027 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
3028
3029 return ms;
3030}
3031
3032static int receive_req_state(struct drbd_conf *mdev, struct p_header *h)
3033{
3034 struct p_req_state *p = (struct p_req_state *)h;
3035 union drbd_state mask, val;
3036 int rv;
3037
3038 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
3039 if (drbd_recv(mdev, h->payload, h->length) != h->length)
3040 return FALSE;
3041
3042 mask.i = be32_to_cpu(p->mask);
3043 val.i = be32_to_cpu(p->val);
3044
3045 if (test_bit(DISCARD_CONCURRENT, &mdev->flags) &&
3046 test_bit(CLUSTER_ST_CHANGE, &mdev->flags)) {
3047 drbd_send_sr_reply(mdev, SS_CONCURRENT_ST_CHG);
3048 return TRUE;
3049 }
3050
3051 mask = convert_state(mask);
3052 val = convert_state(val);
3053
3054 rv = drbd_change_state(mdev, CS_VERBOSE, mask, val);
3055
3056 drbd_send_sr_reply(mdev, rv);
3057 drbd_md_sync(mdev);
3058
3059 return TRUE;
3060}
3061
3062static int receive_state(struct drbd_conf *mdev, struct p_header *h)
3063{
3064 struct p_state *p = (struct p_state *)h;
3065 enum drbd_conns nconn, oconn;
3066 union drbd_state ns, peer_state;
3067 enum drbd_disk_state real_peer_disk;
3068 int rv;
3069
3070 ERR_IF(h->length != (sizeof(*p)-sizeof(*h)))
3071 return FALSE;
3072
3073 if (drbd_recv(mdev, h->payload, h->length) != h->length)
3074 return FALSE;
3075
3076 peer_state.i = be32_to_cpu(p->state);
3077
3078 real_peer_disk = peer_state.disk;
3079 if (peer_state.disk == D_NEGOTIATING) {
3080 real_peer_disk = mdev->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
3081 dev_info(DEV, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
3082 }
3083
3084 spin_lock_irq(&mdev->req_lock);
3085 retry:
3086 oconn = nconn = mdev->state.conn;
3087 spin_unlock_irq(&mdev->req_lock);
3088
3089 if (nconn == C_WF_REPORT_PARAMS)
3090 nconn = C_CONNECTED;
3091
3092 if (mdev->p_uuid && peer_state.disk >= D_NEGOTIATING &&
3093 get_ldev_if_state(mdev, D_NEGOTIATING)) {
3094 int cr; /* consider resync */
3095
3096 /* if we established a new connection */
3097 cr = (oconn < C_CONNECTED);
3098 /* if we had an established connection
3099 * and one of the nodes newly attaches a disk */
3100 cr |= (oconn == C_CONNECTED &&
3101 (peer_state.disk == D_NEGOTIATING ||
3102 mdev->state.disk == D_NEGOTIATING));
3103 /* if we have both been inconsistent, and the peer has been
3104 * forced to be UpToDate with --overwrite-data */
3105 cr |= test_bit(CONSIDER_RESYNC, &mdev->flags);
3106 /* if we had been plain connected, and the admin requested to
3107 * start a sync by "invalidate" or "invalidate-remote" */
3108 cr |= (oconn == C_CONNECTED &&
3109 (peer_state.conn >= C_STARTING_SYNC_S &&
3110 peer_state.conn <= C_WF_BITMAP_T));
3111
3112 if (cr)
3113 nconn = drbd_sync_handshake(mdev, peer_state.role, real_peer_disk);
3114
3115 put_ldev(mdev);
3116 if (nconn == C_MASK) {
3117 if (mdev->state.disk == D_NEGOTIATING) {
3118 drbd_force_state(mdev, NS(disk, D_DISKLESS));
3119 nconn = C_CONNECTED;
3120 } else if (peer_state.disk == D_NEGOTIATING) {
3121 dev_err(DEV, "Disk attach process on the peer node was aborted.\n");
3122 peer_state.disk = D_DISKLESS;
3123 } else {
3124 D_ASSERT(oconn == C_WF_REPORT_PARAMS);
3125 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3126 return FALSE;
3127 }
3128 }
3129 }
3130
3131 spin_lock_irq(&mdev->req_lock);
3132 if (mdev->state.conn != oconn)
3133 goto retry;
3134 clear_bit(CONSIDER_RESYNC, &mdev->flags);
3135 ns.i = mdev->state.i;
3136 ns.conn = nconn;
3137 ns.peer = peer_state.role;
3138 ns.pdsk = real_peer_disk;
3139 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
3140 if ((nconn == C_CONNECTED || nconn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
3141 ns.disk = mdev->new_state_tmp.disk;
3142
3143 rv = _drbd_set_state(mdev, ns, CS_VERBOSE | CS_HARD, NULL);
3144 ns = mdev->state;
3145 spin_unlock_irq(&mdev->req_lock);
3146
3147 if (rv < SS_SUCCESS) {
3148 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3149 return FALSE;
3150 }
3151
3152 if (oconn > C_WF_REPORT_PARAMS) {
3153 if (nconn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
3154 peer_state.disk != D_NEGOTIATING ) {
3155 /* we want resync, peer has not yet decided to sync... */
3156 /* Nowadays only used when forcing a node into primary role and
3157 setting its disk to UpToDate with that */
3158 drbd_send_uuids(mdev);
3159 drbd_send_state(mdev);
3160 }
3161 }
3162
3163 mdev->net_conf->want_lose = 0;
3164
3165 drbd_md_sync(mdev); /* update connected indicator, la_size, ... */
3166
3167 return TRUE;
3168}
3169
3170static int receive_sync_uuid(struct drbd_conf *mdev, struct p_header *h)
3171{
3172 struct p_rs_uuid *p = (struct p_rs_uuid *)h;
3173
3174 wait_event(mdev->misc_wait,
3175 mdev->state.conn == C_WF_SYNC_UUID ||
3176 mdev->state.conn < C_CONNECTED ||
3177 mdev->state.disk < D_NEGOTIATING);
3178
3179 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3180
3181 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
3182 if (drbd_recv(mdev, h->payload, h->length) != h->length)
3183 return FALSE;
3184
3185 /* Here the _drbd_uuid_ functions are right, current should
3186 _not_ be rotated into the history */
3187 if (get_ldev_if_state(mdev, D_NEGOTIATING)) {
3188 _drbd_uuid_set(mdev, UI_CURRENT, be64_to_cpu(p->uuid));
3189 _drbd_uuid_set(mdev, UI_BITMAP, 0UL);
3190
3191 drbd_start_resync(mdev, C_SYNC_TARGET);
3192
3193 put_ldev(mdev);
3194 } else
3195 dev_err(DEV, "Ignoring SyncUUID packet!\n");
3196
3197 return TRUE;
3198}
3199
3200enum receive_bitmap_ret { OK, DONE, FAILED };
3201
3202static enum receive_bitmap_ret
3203receive_bitmap_plain(struct drbd_conf *mdev, struct p_header *h,
3204 unsigned long *buffer, struct bm_xfer_ctx *c)
3205{
3206 unsigned num_words = min_t(size_t, BM_PACKET_WORDS, c->bm_words - c->word_offset);
3207 unsigned want = num_words * sizeof(long);
3208
3209 if (want != h->length) {
3210 dev_err(DEV, "%s:want (%u) != h->length (%u)\n", __func__, want, h->length);
3211 return FAILED;
3212 }
3213 if (want == 0)
3214 return DONE;
3215 if (drbd_recv(mdev, buffer, want) != want)
3216 return FAILED;
3217
3218 drbd_bm_merge_lel(mdev, c->word_offset, num_words, buffer);
3219
3220 c->word_offset += num_words;
3221 c->bit_offset = c->word_offset * BITS_PER_LONG;
3222 if (c->bit_offset > c->bm_bits)
3223 c->bit_offset = c->bm_bits;
3224
3225 return OK;
3226}
3227
3228static enum receive_bitmap_ret
3229recv_bm_rle_bits(struct drbd_conf *mdev,
3230 struct p_compressed_bm *p,
3231 struct bm_xfer_ctx *c)
3232{
3233 struct bitstream bs;
3234 u64 look_ahead;
3235 u64 rl;
3236 u64 tmp;
3237 unsigned long s = c->bit_offset;
3238 unsigned long e;
3239 int len = p->head.length - (sizeof(*p) - sizeof(p->head));
3240 int toggle = DCBP_get_start(p);
3241 int have;
3242 int bits;
3243
3244 bitstream_init(&bs, p->code, len, DCBP_get_pad_bits(p));
3245
3246 bits = bitstream_get_bits(&bs, &look_ahead, 64);
3247 if (bits < 0)
3248 return FAILED;
3249
3250 for (have = bits; have > 0; s += rl, toggle = !toggle) {
3251 bits = vli_decode_bits(&rl, look_ahead);
3252 if (bits <= 0)
3253 return FAILED;
3254
3255 if (toggle) {
3256 e = s + rl -1;
3257 if (e >= c->bm_bits) {
3258 dev_err(DEV, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
3259 return FAILED;
3260 }
3261 _drbd_bm_set_bits(mdev, s, e);
3262 }
3263
3264 if (have < bits) {
3265 dev_err(DEV, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3266 have, bits, look_ahead,
3267 (unsigned int)(bs.cur.b - p->code),
3268 (unsigned int)bs.buf_len);
3269 return FAILED;
3270 }
3271 look_ahead >>= bits;
3272 have -= bits;
3273
3274 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
3275 if (bits < 0)
3276 return FAILED;
3277 look_ahead |= tmp << have;
3278 have += bits;
3279 }
3280
3281 c->bit_offset = s;
3282 bm_xfer_ctx_bit_to_word_offset(c);
3283
3284 return (s == c->bm_bits) ? DONE : OK;
3285}
3286
3287static enum receive_bitmap_ret
3288decode_bitmap_c(struct drbd_conf *mdev,
3289 struct p_compressed_bm *p,
3290 struct bm_xfer_ctx *c)
3291{
3292 if (DCBP_get_code(p) == RLE_VLI_Bits)
3293 return recv_bm_rle_bits(mdev, p, c);
3294
3295 /* other variants had been implemented for evaluation,
3296 * but have been dropped as this one turned out to be "best"
3297 * during all our tests. */
3298
3299 dev_err(DEV, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
3300 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3301 return FAILED;
3302}
3303
3304void INFO_bm_xfer_stats(struct drbd_conf *mdev,
3305 const char *direction, struct bm_xfer_ctx *c)
3306{
3307 /* what would it take to transfer it "plaintext" */
3308 unsigned plain = sizeof(struct p_header) *
3309 ((c->bm_words+BM_PACKET_WORDS-1)/BM_PACKET_WORDS+1)
3310 + c->bm_words * sizeof(long);
3311 unsigned total = c->bytes[0] + c->bytes[1];
3312 unsigned r;
3313
3314 /* total can not be zero. but just in case: */
3315 if (total == 0)
3316 return;
3317
3318 /* don't report if not compressed */
3319 if (total >= plain)
3320 return;
3321
3322 /* total < plain. check for overflow, still */
3323 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
3324 : (1000 * total / plain);
3325
3326 if (r > 1000)
3327 r = 1000;
3328
3329 r = 1000 - r;
3330 dev_info(DEV, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3331 "total %u; compression: %u.%u%%\n",
3332 direction,
3333 c->bytes[1], c->packets[1],
3334 c->bytes[0], c->packets[0],
3335 total, r/10, r % 10);
3336}
3337
3338/* Since we are processing the bitfield from lower addresses to higher,
3339 it does not matter if the process it in 32 bit chunks or 64 bit
3340 chunks as long as it is little endian. (Understand it as byte stream,
3341 beginning with the lowest byte...) If we would use big endian
3342 we would need to process it from the highest address to the lowest,
3343 in order to be agnostic to the 32 vs 64 bits issue.
3344
3345 returns 0 on failure, 1 if we successfully received it. */
3346static int receive_bitmap(struct drbd_conf *mdev, struct p_header *h)
3347{
3348 struct bm_xfer_ctx c;
3349 void *buffer;
3350 enum receive_bitmap_ret ret;
3351 int ok = FALSE;
3352
3353 wait_event(mdev->misc_wait, !atomic_read(&mdev->ap_bio_cnt));
3354
3355 drbd_bm_lock(mdev, "receive bitmap");
3356
3357 /* maybe we should use some per thread scratch page,
3358 * and allocate that during initial device creation? */
3359 buffer = (unsigned long *) __get_free_page(GFP_NOIO);
3360 if (!buffer) {
3361 dev_err(DEV, "failed to allocate one page buffer in %s\n", __func__);
3362 goto out;
3363 }
3364
3365 c = (struct bm_xfer_ctx) {
3366 .bm_bits = drbd_bm_bits(mdev),
3367 .bm_words = drbd_bm_words(mdev),
3368 };
3369
3370 do {
3371 if (h->command == P_BITMAP) {
3372 ret = receive_bitmap_plain(mdev, h, buffer, &c);
3373 } else if (h->command == P_COMPRESSED_BITMAP) {
3374 /* MAYBE: sanity check that we speak proto >= 90,
3375 * and the feature is enabled! */
3376 struct p_compressed_bm *p;
3377
3378 if (h->length > BM_PACKET_PAYLOAD_BYTES) {
3379 dev_err(DEV, "ReportCBitmap packet too large\n");
3380 goto out;
3381 }
3382 /* use the page buff */
3383 p = buffer;
3384 memcpy(p, h, sizeof(*h));
3385 if (drbd_recv(mdev, p->head.payload, h->length) != h->length)
3386 goto out;
3387 if (p->head.length <= (sizeof(*p) - sizeof(p->head))) {
3388 dev_err(DEV, "ReportCBitmap packet too small (l:%u)\n", p->head.length);
3389 return FAILED;
3390 }
3391 ret = decode_bitmap_c(mdev, p, &c);
3392 } else {
3393 dev_warn(DEV, "receive_bitmap: h->command neither ReportBitMap nor ReportCBitMap (is 0x%x)", h->command);
3394 goto out;
3395 }
3396
3397 c.packets[h->command == P_BITMAP]++;
3398 c.bytes[h->command == P_BITMAP] += sizeof(struct p_header) + h->length;
3399
3400 if (ret != OK)
3401 break;
3402
3403 if (!drbd_recv_header(mdev, h))
3404 goto out;
3405 } while (ret == OK);
3406 if (ret == FAILED)
3407 goto out;
3408
3409 INFO_bm_xfer_stats(mdev, "receive", &c);
3410
3411 if (mdev->state.conn == C_WF_BITMAP_T) {
3412 ok = !drbd_send_bitmap(mdev);
3413 if (!ok)
3414 goto out;
3415 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
3416 ok = _drbd_request_state(mdev, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
3417 D_ASSERT(ok == SS_SUCCESS);
3418 } else if (mdev->state.conn != C_WF_BITMAP_S) {
3419 /* admin may have requested C_DISCONNECTING,
3420 * other threads may have noticed network errors */
3421 dev_info(DEV, "unexpected cstate (%s) in receive_bitmap\n",
3422 drbd_conn_str(mdev->state.conn));
3423 }
3424
3425 ok = TRUE;
3426 out:
3427 drbd_bm_unlock(mdev);
3428 if (ok && mdev->state.conn == C_WF_BITMAP_S)
3429 drbd_start_resync(mdev, C_SYNC_SOURCE);
3430 free_page((unsigned long) buffer);
3431 return ok;
3432}
3433
3434static int receive_skip(struct drbd_conf *mdev, struct p_header *h)
3435{
3436 /* TODO zero copy sink :) */
3437 static char sink[128];
3438 int size, want, r;
3439
3440 dev_warn(DEV, "skipping unknown optional packet type %d, l: %d!\n",
3441 h->command, h->length);
3442
3443 size = h->length;
3444 while (size > 0) {
3445 want = min_t(int, size, sizeof(sink));
3446 r = drbd_recv(mdev, sink, want);
3447 ERR_IF(r <= 0) break;
3448 size -= r;
3449 }
3450 return size == 0;
3451}
3452
3453static int receive_UnplugRemote(struct drbd_conf *mdev, struct p_header *h)
3454{
3455 if (mdev->state.disk >= D_INCONSISTENT)
3456 drbd_kick_lo(mdev);
3457
3458 /* Make sure we've acked all the TCP data associated
3459 * with the data requests being unplugged */
3460 drbd_tcp_quickack(mdev->data.socket);
3461
3462 return TRUE;
3463}
3464
3465typedef int (*drbd_cmd_handler_f)(struct drbd_conf *, struct p_header *);
3466
3467static drbd_cmd_handler_f drbd_default_handler[] = {
3468 [P_DATA] = receive_Data,
3469 [P_DATA_REPLY] = receive_DataReply,
3470 [P_RS_DATA_REPLY] = receive_RSDataReply,
3471 [P_BARRIER] = receive_Barrier,
3472 [P_BITMAP] = receive_bitmap,
3473 [P_COMPRESSED_BITMAP] = receive_bitmap,
3474 [P_UNPLUG_REMOTE] = receive_UnplugRemote,
3475 [P_DATA_REQUEST] = receive_DataRequest,
3476 [P_RS_DATA_REQUEST] = receive_DataRequest,
3477 [P_SYNC_PARAM] = receive_SyncParam,
3478 [P_SYNC_PARAM89] = receive_SyncParam,
3479 [P_PROTOCOL] = receive_protocol,
3480 [P_UUIDS] = receive_uuids,
3481 [P_SIZES] = receive_sizes,
3482 [P_STATE] = receive_state,
3483 [P_STATE_CHG_REQ] = receive_req_state,
3484 [P_SYNC_UUID] = receive_sync_uuid,
3485 [P_OV_REQUEST] = receive_DataRequest,
3486 [P_OV_REPLY] = receive_DataRequest,
3487 [P_CSUM_RS_REQUEST] = receive_DataRequest,
3488 /* anything missing from this table is in
3489 * the asender_tbl, see get_asender_cmd */
3490 [P_MAX_CMD] = NULL,
3491};
3492
3493static drbd_cmd_handler_f *drbd_cmd_handler = drbd_default_handler;
3494static drbd_cmd_handler_f *drbd_opt_cmd_handler;
3495
3496static void drbdd(struct drbd_conf *mdev)
3497{
3498 drbd_cmd_handler_f handler;
3499 struct p_header *header = &mdev->data.rbuf.header;
3500
3501 while (get_t_state(&mdev->receiver) == Running) {
3502 drbd_thread_current_set_cpu(mdev);
3503 if (!drbd_recv_header(mdev, header)) {
3504 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3505 break;
3506 }
3507
3508 if (header->command < P_MAX_CMD)
3509 handler = drbd_cmd_handler[header->command];
3510 else if (P_MAY_IGNORE < header->command
3511 && header->command < P_MAX_OPT_CMD)
3512 handler = drbd_opt_cmd_handler[header->command-P_MAY_IGNORE];
3513 else if (header->command > P_MAX_OPT_CMD)
3514 handler = receive_skip;
3515 else
3516 handler = NULL;
3517
3518 if (unlikely(!handler)) {
3519 dev_err(DEV, "unknown packet type %d, l: %d!\n",
3520 header->command, header->length);
3521 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3522 break;
3523 }
3524 if (unlikely(!handler(mdev, header))) {
3525 dev_err(DEV, "error receiving %s, l: %d!\n",
3526 cmdname(header->command), header->length);
3527 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3528 break;
3529 }
3530 }
3531}
3532
3533static void drbd_fail_pending_reads(struct drbd_conf *mdev)
3534{
3535 struct hlist_head *slot;
3536 struct hlist_node *pos;
3537 struct hlist_node *tmp;
3538 struct drbd_request *req;
3539 int i;
3540
3541 /*
3542 * Application READ requests
3543 */
3544 spin_lock_irq(&mdev->req_lock);
3545 for (i = 0; i < APP_R_HSIZE; i++) {
3546 slot = mdev->app_reads_hash+i;
3547 hlist_for_each_entry_safe(req, pos, tmp, slot, colision) {
3548 /* it may (but should not any longer!)
3549 * be on the work queue; if that assert triggers,
3550 * we need to also grab the
3551 * spin_lock_irq(&mdev->data.work.q_lock);
3552 * and list_del_init here. */
3553 D_ASSERT(list_empty(&req->w.list));
3554 /* It would be nice to complete outside of spinlock.
3555 * But this is easier for now. */
3556 _req_mod(req, connection_lost_while_pending);
3557 }
3558 }
3559 for (i = 0; i < APP_R_HSIZE; i++)
3560 if (!hlist_empty(mdev->app_reads_hash+i))
3561 dev_warn(DEV, "ASSERT FAILED: app_reads_hash[%d].first: "
3562 "%p, should be NULL\n", i, mdev->app_reads_hash[i].first);
3563
3564 memset(mdev->app_reads_hash, 0, APP_R_HSIZE*sizeof(void *));
3565 spin_unlock_irq(&mdev->req_lock);
3566}
3567
3568void drbd_flush_workqueue(struct drbd_conf *mdev)
3569{
3570 struct drbd_wq_barrier barr;
3571
3572 barr.w.cb = w_prev_work_done;
3573 init_completion(&barr.done);
3574 drbd_queue_work(&mdev->data.work, &barr.w);
3575 wait_for_completion(&barr.done);
3576}
3577
3578static void drbd_disconnect(struct drbd_conf *mdev)
3579{
3580 enum drbd_fencing_p fp;
3581 union drbd_state os, ns;
3582 int rv = SS_UNKNOWN_ERROR;
3583 unsigned int i;
3584
3585 if (mdev->state.conn == C_STANDALONE)
3586 return;
3587 if (mdev->state.conn >= C_WF_CONNECTION)
3588 dev_err(DEV, "ASSERT FAILED cstate = %s, expected < WFConnection\n",
3589 drbd_conn_str(mdev->state.conn));
3590
3591 /* asender does not clean up anything. it must not interfere, either */
3592 drbd_thread_stop(&mdev->asender);
3593
3594 mutex_lock(&mdev->data.mutex);
3595 drbd_free_sock(mdev);
3596 mutex_unlock(&mdev->data.mutex);
3597
3598 spin_lock_irq(&mdev->req_lock);
3599 _drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
3600 _drbd_wait_ee_list_empty(mdev, &mdev->sync_ee);
3601 _drbd_wait_ee_list_empty(mdev, &mdev->read_ee);
3602 spin_unlock_irq(&mdev->req_lock);
3603
3604 /* We do not have data structures that would allow us to
3605 * get the rs_pending_cnt down to 0 again.
3606 * * On C_SYNC_TARGET we do not have any data structures describing
3607 * the pending RSDataRequest's we have sent.
3608 * * On C_SYNC_SOURCE there is no data structure that tracks
3609 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
3610 * And no, it is not the sum of the reference counts in the
3611 * resync_LRU. The resync_LRU tracks the whole operation including
3612 * the disk-IO, while the rs_pending_cnt only tracks the blocks
3613 * on the fly. */
3614 drbd_rs_cancel_all(mdev);
3615 mdev->rs_total = 0;
3616 mdev->rs_failed = 0;
3617 atomic_set(&mdev->rs_pending_cnt, 0);
3618 wake_up(&mdev->misc_wait);
3619
3620 /* make sure syncer is stopped and w_resume_next_sg queued */
3621 del_timer_sync(&mdev->resync_timer);
3622 set_bit(STOP_SYNC_TIMER, &mdev->flags);
3623 resync_timer_fn((unsigned long)mdev);
3624
3625 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
3626 * w_make_resync_request etc. which may still be on the worker queue
3627 * to be "canceled" */
3628 drbd_flush_workqueue(mdev);
3629
3630 /* This also does reclaim_net_ee(). If we do this too early, we might
3631 * miss some resync ee and pages.*/
3632 drbd_process_done_ee(mdev);
3633
3634 kfree(mdev->p_uuid);
3635 mdev->p_uuid = NULL;
3636
3637 if (!mdev->state.susp)
3638 tl_clear(mdev);
3639
3640 drbd_fail_pending_reads(mdev);
3641
3642 dev_info(DEV, "Connection closed\n");
3643
3644 drbd_md_sync(mdev);
3645
3646 fp = FP_DONT_CARE;
3647 if (get_ldev(mdev)) {
3648 fp = mdev->ldev->dc.fencing;
3649 put_ldev(mdev);
3650 }
3651
3652 if (mdev->state.role == R_PRIMARY) {
3653 if (fp >= FP_RESOURCE && mdev->state.pdsk >= D_UNKNOWN) {
3654 enum drbd_disk_state nps = drbd_try_outdate_peer(mdev);
3655 drbd_request_state(mdev, NS(pdsk, nps));
3656 }
3657 }
3658
3659 spin_lock_irq(&mdev->req_lock);
3660 os = mdev->state;
3661 if (os.conn >= C_UNCONNECTED) {
3662 /* Do not restart in case we are C_DISCONNECTING */
3663 ns = os;
3664 ns.conn = C_UNCONNECTED;
3665 rv = _drbd_set_state(mdev, ns, CS_VERBOSE, NULL);
3666 }
3667 spin_unlock_irq(&mdev->req_lock);
3668
3669 if (os.conn == C_DISCONNECTING) {
3670 struct hlist_head *h;
3671 wait_event(mdev->misc_wait, atomic_read(&mdev->net_cnt) == 0);
3672
3673 /* we must not free the tl_hash
3674 * while application io is still on the fly */
3675 wait_event(mdev->misc_wait, atomic_read(&mdev->ap_bio_cnt) == 0);
3676
3677 spin_lock_irq(&mdev->req_lock);
3678 /* paranoia code */
3679 for (h = mdev->ee_hash; h < mdev->ee_hash + mdev->ee_hash_s; h++)
3680 if (h->first)
3681 dev_err(DEV, "ASSERT FAILED ee_hash[%u].first == %p, expected NULL\n",
3682 (int)(h - mdev->ee_hash), h->first);
3683 kfree(mdev->ee_hash);
3684 mdev->ee_hash = NULL;
3685 mdev->ee_hash_s = 0;
3686
3687 /* paranoia code */
3688 for (h = mdev->tl_hash; h < mdev->tl_hash + mdev->tl_hash_s; h++)
3689 if (h->first)
3690 dev_err(DEV, "ASSERT FAILED tl_hash[%u] == %p, expected NULL\n",
3691 (int)(h - mdev->tl_hash), h->first);
3692 kfree(mdev->tl_hash);
3693 mdev->tl_hash = NULL;
3694 mdev->tl_hash_s = 0;
3695 spin_unlock_irq(&mdev->req_lock);
3696
3697 crypto_free_hash(mdev->cram_hmac_tfm);
3698 mdev->cram_hmac_tfm = NULL;
3699
3700 kfree(mdev->net_conf);
3701 mdev->net_conf = NULL;
3702 drbd_request_state(mdev, NS(conn, C_STANDALONE));
3703 }
3704
3705 /* tcp_close and release of sendpage pages can be deferred. I don't
3706 * want to use SO_LINGER, because apparently it can be deferred for
3707 * more than 20 seconds (longest time I checked).
3708 *
3709 * Actually we don't care for exactly when the network stack does its
3710 * put_page(), but release our reference on these pages right here.
3711 */
3712 i = drbd_release_ee(mdev, &mdev->net_ee);
3713 if (i)
3714 dev_info(DEV, "net_ee not empty, killed %u entries\n", i);
3715 i = atomic_read(&mdev->pp_in_use);
3716 if (i)
3717 dev_info(DEV, "pp_in_use = %u, expected 0\n", i);
3718
3719 D_ASSERT(list_empty(&mdev->read_ee));
3720 D_ASSERT(list_empty(&mdev->active_ee));
3721 D_ASSERT(list_empty(&mdev->sync_ee));
3722 D_ASSERT(list_empty(&mdev->done_ee));
3723
3724 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
3725 atomic_set(&mdev->current_epoch->epoch_size, 0);
3726 D_ASSERT(list_empty(&mdev->current_epoch->list));
3727}
3728
3729/*
3730 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
3731 * we can agree on is stored in agreed_pro_version.
3732 *
3733 * feature flags and the reserved array should be enough room for future
3734 * enhancements of the handshake protocol, and possible plugins...
3735 *
3736 * for now, they are expected to be zero, but ignored.
3737 */
3738static int drbd_send_handshake(struct drbd_conf *mdev)
3739{
3740 /* ASSERT current == mdev->receiver ... */
3741 struct p_handshake *p = &mdev->data.sbuf.handshake;
3742 int ok;
3743
3744 if (mutex_lock_interruptible(&mdev->data.mutex)) {
3745 dev_err(DEV, "interrupted during initial handshake\n");
3746 return 0; /* interrupted. not ok. */
3747 }
3748
3749 if (mdev->data.socket == NULL) {
3750 mutex_unlock(&mdev->data.mutex);
3751 return 0;
3752 }
3753
3754 memset(p, 0, sizeof(*p));
3755 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
3756 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
3757 ok = _drbd_send_cmd( mdev, mdev->data.socket, P_HAND_SHAKE,
3758 (struct p_header *)p, sizeof(*p), 0 );
3759 mutex_unlock(&mdev->data.mutex);
3760 return ok;
3761}
3762
3763/*
3764 * return values:
3765 * 1 yes, we have a valid connection
3766 * 0 oops, did not work out, please try again
3767 * -1 peer talks different language,
3768 * no point in trying again, please go standalone.
3769 */
3770static int drbd_do_handshake(struct drbd_conf *mdev)
3771{
3772 /* ASSERT current == mdev->receiver ... */
3773 struct p_handshake *p = &mdev->data.rbuf.handshake;
3774 const int expect = sizeof(struct p_handshake)
3775 -sizeof(struct p_header);
3776 int rv;
3777
3778 rv = drbd_send_handshake(mdev);
3779 if (!rv)
3780 return 0;
3781
3782 rv = drbd_recv_header(mdev, &p->head);
3783 if (!rv)
3784 return 0;
3785
3786 if (p->head.command != P_HAND_SHAKE) {
3787 dev_err(DEV, "expected HandShake packet, received: %s (0x%04x)\n",
3788 cmdname(p->head.command), p->head.command);
3789 return -1;
3790 }
3791
3792 if (p->head.length != expect) {
3793 dev_err(DEV, "expected HandShake length: %u, received: %u\n",
3794 expect, p->head.length);
3795 return -1;
3796 }
3797
3798 rv = drbd_recv(mdev, &p->head.payload, expect);
3799
3800 if (rv != expect) {
3801 dev_err(DEV, "short read receiving handshake packet: l=%u\n", rv);
3802 return 0;
3803 }
3804
3805 p->protocol_min = be32_to_cpu(p->protocol_min);
3806 p->protocol_max = be32_to_cpu(p->protocol_max);
3807 if (p->protocol_max == 0)
3808 p->protocol_max = p->protocol_min;
3809
3810 if (PRO_VERSION_MAX < p->protocol_min ||
3811 PRO_VERSION_MIN > p->protocol_max)
3812 goto incompat;
3813
3814 mdev->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
3815
3816 dev_info(DEV, "Handshake successful: "
3817 "Agreed network protocol version %d\n", mdev->agreed_pro_version);
3818
3819 return 1;
3820
3821 incompat:
3822 dev_err(DEV, "incompatible DRBD dialects: "
3823 "I support %d-%d, peer supports %d-%d\n",
3824 PRO_VERSION_MIN, PRO_VERSION_MAX,
3825 p->protocol_min, p->protocol_max);
3826 return -1;
3827}
3828
3829#if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
3830static int drbd_do_auth(struct drbd_conf *mdev)
3831{
3832 dev_err(DEV, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
3833 dev_err(DEV, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
3834 return 0;
3835}
3836#else
3837#define CHALLENGE_LEN 64
3838static int drbd_do_auth(struct drbd_conf *mdev)
3839{
3840 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
3841 struct scatterlist sg;
3842 char *response = NULL;
3843 char *right_response = NULL;
3844 char *peers_ch = NULL;
3845 struct p_header p;
3846 unsigned int key_len = strlen(mdev->net_conf->shared_secret);
3847 unsigned int resp_size;
3848 struct hash_desc desc;
3849 int rv;
3850
3851 desc.tfm = mdev->cram_hmac_tfm;
3852 desc.flags = 0;
3853
3854 rv = crypto_hash_setkey(mdev->cram_hmac_tfm,
3855 (u8 *)mdev->net_conf->shared_secret, key_len);
3856 if (rv) {
3857 dev_err(DEV, "crypto_hash_setkey() failed with %d\n", rv);
3858 rv = 0;
3859 goto fail;
3860 }
3861
3862 get_random_bytes(my_challenge, CHALLENGE_LEN);
3863
3864 rv = drbd_send_cmd2(mdev, P_AUTH_CHALLENGE, my_challenge, CHALLENGE_LEN);
3865 if (!rv)
3866 goto fail;
3867
3868 rv = drbd_recv_header(mdev, &p);
3869 if (!rv)
3870 goto fail;
3871
3872 if (p.command != P_AUTH_CHALLENGE) {
3873 dev_err(DEV, "expected AuthChallenge packet, received: %s (0x%04x)\n",
3874 cmdname(p.command), p.command);
3875 rv = 0;
3876 goto fail;
3877 }
3878
3879 if (p.length > CHALLENGE_LEN*2) {
3880 dev_err(DEV, "expected AuthChallenge payload too big.\n");
3881 rv = 0;
3882 goto fail;
3883 }
3884
3885 peers_ch = kmalloc(p.length, GFP_NOIO);
3886 if (peers_ch == NULL) {
3887 dev_err(DEV, "kmalloc of peers_ch failed\n");
3888 rv = 0;
3889 goto fail;
3890 }
3891
3892 rv = drbd_recv(mdev, peers_ch, p.length);
3893
3894 if (rv != p.length) {
3895 dev_err(DEV, "short read AuthChallenge: l=%u\n", rv);
3896 rv = 0;
3897 goto fail;
3898 }
3899
3900 resp_size = crypto_hash_digestsize(mdev->cram_hmac_tfm);
3901 response = kmalloc(resp_size, GFP_NOIO);
3902 if (response == NULL) {
3903 dev_err(DEV, "kmalloc of response failed\n");
3904 rv = 0;
3905 goto fail;
3906 }
3907
3908 sg_init_table(&sg, 1);
3909 sg_set_buf(&sg, peers_ch, p.length);
3910
3911 rv = crypto_hash_digest(&desc, &sg, sg.length, response);
3912 if (rv) {
3913 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv);
3914 rv = 0;
3915 goto fail;
3916 }
3917
3918 rv = drbd_send_cmd2(mdev, P_AUTH_RESPONSE, response, resp_size);
3919 if (!rv)
3920 goto fail;
3921
3922 rv = drbd_recv_header(mdev, &p);
3923 if (!rv)
3924 goto fail;
3925
3926 if (p.command != P_AUTH_RESPONSE) {
3927 dev_err(DEV, "expected AuthResponse packet, received: %s (0x%04x)\n",
3928 cmdname(p.command), p.command);
3929 rv = 0;
3930 goto fail;
3931 }
3932
3933 if (p.length != resp_size) {
3934 dev_err(DEV, "expected AuthResponse payload of wrong size\n");
3935 rv = 0;
3936 goto fail;
3937 }
3938
3939 rv = drbd_recv(mdev, response , resp_size);
3940
3941 if (rv != resp_size) {
3942 dev_err(DEV, "short read receiving AuthResponse: l=%u\n", rv);
3943 rv = 0;
3944 goto fail;
3945 }
3946
3947 right_response = kmalloc(resp_size, GFP_NOIO);
3948 if (response == NULL) {
3949 dev_err(DEV, "kmalloc of right_response failed\n");
3950 rv = 0;
3951 goto fail;
3952 }
3953
3954 sg_set_buf(&sg, my_challenge, CHALLENGE_LEN);
3955
3956 rv = crypto_hash_digest(&desc, &sg, sg.length, right_response);
3957 if (rv) {
3958 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv);
3959 rv = 0;
3960 goto fail;
3961 }
3962
3963 rv = !memcmp(response, right_response, resp_size);
3964
3965 if (rv)
3966 dev_info(DEV, "Peer authenticated using %d bytes of '%s' HMAC\n",
3967 resp_size, mdev->net_conf->cram_hmac_alg);
3968
3969 fail:
3970 kfree(peers_ch);
3971 kfree(response);
3972 kfree(right_response);
3973
3974 return rv;
3975}
3976#endif
3977
3978int drbdd_init(struct drbd_thread *thi)
3979{
3980 struct drbd_conf *mdev = thi->mdev;
3981 unsigned int minor = mdev_to_minor(mdev);
3982 int h;
3983
3984 sprintf(current->comm, "drbd%d_receiver", minor);
3985
3986 dev_info(DEV, "receiver (re)started\n");
3987
3988 do {
3989 h = drbd_connect(mdev);
3990 if (h == 0) {
3991 drbd_disconnect(mdev);
3992 __set_current_state(TASK_INTERRUPTIBLE);
3993 schedule_timeout(HZ);
3994 }
3995 if (h == -1) {
3996 dev_warn(DEV, "Discarding network configuration.\n");
3997 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3998 }
3999 } while (h == 0);
4000
4001 if (h > 0) {
4002 if (get_net_conf(mdev)) {
4003 drbdd(mdev);
4004 put_net_conf(mdev);
4005 }
4006 }
4007
4008 drbd_disconnect(mdev);
4009
4010 dev_info(DEV, "receiver terminated\n");
4011 return 0;
4012}
4013
4014/* ********* acknowledge sender ******** */
4015
4016static int got_RqSReply(struct drbd_conf *mdev, struct p_header *h)
4017{
4018 struct p_req_state_reply *p = (struct p_req_state_reply *)h;
4019
4020 int retcode = be32_to_cpu(p->retcode);
4021
4022 if (retcode >= SS_SUCCESS) {
4023 set_bit(CL_ST_CHG_SUCCESS, &mdev->flags);
4024 } else {
4025 set_bit(CL_ST_CHG_FAIL, &mdev->flags);
4026 dev_err(DEV, "Requested state change failed by peer: %s (%d)\n",
4027 drbd_set_st_err_str(retcode), retcode);
4028 }
4029 wake_up(&mdev->state_wait);
4030
4031 return TRUE;
4032}
4033
4034static int got_Ping(struct drbd_conf *mdev, struct p_header *h)
4035{
4036 return drbd_send_ping_ack(mdev);
4037
4038}
4039
4040static int got_PingAck(struct drbd_conf *mdev, struct p_header *h)
4041{
4042 /* restore idle timeout */
4043 mdev->meta.socket->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ;
4044
4045 return TRUE;
4046}
4047
4048static int got_IsInSync(struct drbd_conf *mdev, struct p_header *h)
4049{
4050 struct p_block_ack *p = (struct p_block_ack *)h;
4051 sector_t sector = be64_to_cpu(p->sector);
4052 int blksize = be32_to_cpu(p->blksize);
4053
4054 D_ASSERT(mdev->agreed_pro_version >= 89);
4055
4056 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4057
4058 drbd_rs_complete_io(mdev, sector);
4059 drbd_set_in_sync(mdev, sector, blksize);
4060 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4061 mdev->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
4062 dec_rs_pending(mdev);
4063
4064 return TRUE;
4065}
4066
4067/* when we receive the ACK for a write request,
4068 * verify that we actually know about it */
4069static struct drbd_request *_ack_id_to_req(struct drbd_conf *mdev,
4070 u64 id, sector_t sector)
4071{
4072 struct hlist_head *slot = tl_hash_slot(mdev, sector);
4073 struct hlist_node *n;
4074 struct drbd_request *req;
4075
4076 hlist_for_each_entry(req, n, slot, colision) {
4077 if ((unsigned long)req == (unsigned long)id) {
4078 if (req->sector != sector) {
4079 dev_err(DEV, "_ack_id_to_req: found req %p but it has "
4080 "wrong sector (%llus versus %llus)\n", req,
4081 (unsigned long long)req->sector,
4082 (unsigned long long)sector);
4083 break;
4084 }
4085 return req;
4086 }
4087 }
4088 dev_err(DEV, "_ack_id_to_req: failed to find req %p, sector %llus in list\n",
4089 (void *)(unsigned long)id, (unsigned long long)sector);
4090 return NULL;
4091}
4092
4093typedef struct drbd_request *(req_validator_fn)
4094 (struct drbd_conf *mdev, u64 id, sector_t sector);
4095
4096static int validate_req_change_req_state(struct drbd_conf *mdev,
4097 u64 id, sector_t sector, req_validator_fn validator,
4098 const char *func, enum drbd_req_event what)
4099{
4100 struct drbd_request *req;
4101 struct bio_and_error m;
4102
4103 spin_lock_irq(&mdev->req_lock);
4104 req = validator(mdev, id, sector);
4105 if (unlikely(!req)) {
4106 spin_unlock_irq(&mdev->req_lock);
4107 dev_err(DEV, "%s: got a corrupt block_id/sector pair\n", func);
4108 return FALSE;
4109 }
4110 __req_mod(req, what, &m);
4111 spin_unlock_irq(&mdev->req_lock);
4112
4113 if (m.bio)
4114 complete_master_bio(mdev, &m);
4115 return TRUE;
4116}
4117
4118static int got_BlockAck(struct drbd_conf *mdev, struct p_header *h)
4119{
4120 struct p_block_ack *p = (struct p_block_ack *)h;
4121 sector_t sector = be64_to_cpu(p->sector);
4122 int blksize = be32_to_cpu(p->blksize);
4123 enum drbd_req_event what;
4124
4125 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4126
4127 if (is_syncer_block_id(p->block_id)) {
4128 drbd_set_in_sync(mdev, sector, blksize);
4129 dec_rs_pending(mdev);
4130 return TRUE;
4131 }
4132 switch (be16_to_cpu(h->command)) {
4133 case P_RS_WRITE_ACK:
4134 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4135 what = write_acked_by_peer_and_sis;
4136 break;
4137 case P_WRITE_ACK:
4138 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4139 what = write_acked_by_peer;
4140 break;
4141 case P_RECV_ACK:
4142 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_B);
4143 what = recv_acked_by_peer;
4144 break;
4145 case P_DISCARD_ACK:
4146 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4147 what = conflict_discarded_by_peer;
4148 break;
4149 default:
4150 D_ASSERT(0);
4151 return FALSE;
4152 }
4153
4154 return validate_req_change_req_state(mdev, p->block_id, sector,
4155 _ack_id_to_req, __func__ , what);
4156}
4157
4158static int got_NegAck(struct drbd_conf *mdev, struct p_header *h)
4159{
4160 struct p_block_ack *p = (struct p_block_ack *)h;
4161 sector_t sector = be64_to_cpu(p->sector);
4162
4163 if (__ratelimit(&drbd_ratelimit_state))
4164 dev_warn(DEV, "Got NegAck packet. Peer is in troubles?\n");
4165
4166 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4167
4168 if (is_syncer_block_id(p->block_id)) {
4169 int size = be32_to_cpu(p->blksize);
4170 dec_rs_pending(mdev);
4171 drbd_rs_failed_io(mdev, sector, size);
4172 return TRUE;
4173 }
4174 return validate_req_change_req_state(mdev, p->block_id, sector,
4175 _ack_id_to_req, __func__ , neg_acked);
4176}
4177
4178static int got_NegDReply(struct drbd_conf *mdev, struct p_header *h)
4179{
4180 struct p_block_ack *p = (struct p_block_ack *)h;
4181 sector_t sector = be64_to_cpu(p->sector);
4182
4183 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4184 dev_err(DEV, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4185 (unsigned long long)sector, be32_to_cpu(p->blksize));
4186
4187 return validate_req_change_req_state(mdev, p->block_id, sector,
4188 _ar_id_to_req, __func__ , neg_acked);
4189}
4190
4191static int got_NegRSDReply(struct drbd_conf *mdev, struct p_header *h)
4192{
4193 sector_t sector;
4194 int size;
4195 struct p_block_ack *p = (struct p_block_ack *)h;
4196
4197 sector = be64_to_cpu(p->sector);
4198 size = be32_to_cpu(p->blksize);
4199 D_ASSERT(p->block_id == ID_SYNCER);
4200
4201 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4202
4203 dec_rs_pending(mdev);
4204
4205 if (get_ldev_if_state(mdev, D_FAILED)) {
4206 drbd_rs_complete_io(mdev, sector);
4207 drbd_rs_failed_io(mdev, sector, size);
4208 put_ldev(mdev);
4209 }
4210
4211 return TRUE;
4212}
4213
4214static int got_BarrierAck(struct drbd_conf *mdev, struct p_header *h)
4215{
4216 struct p_barrier_ack *p = (struct p_barrier_ack *)h;
4217
4218 tl_release(mdev, p->barrier, be32_to_cpu(p->set_size));
4219
4220 return TRUE;
4221}
4222
4223static int got_OVResult(struct drbd_conf *mdev, struct p_header *h)
4224{
4225 struct p_block_ack *p = (struct p_block_ack *)h;
4226 struct drbd_work *w;
4227 sector_t sector;
4228 int size;
4229
4230 sector = be64_to_cpu(p->sector);
4231 size = be32_to_cpu(p->blksize);
4232
4233 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4234
4235 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
4236 drbd_ov_oos_found(mdev, sector, size);
4237 else
4238 ov_oos_print(mdev);
4239
4240 drbd_rs_complete_io(mdev, sector);
4241 dec_rs_pending(mdev);
4242
4243 if (--mdev->ov_left == 0) {
4244 w = kmalloc(sizeof(*w), GFP_NOIO);
4245 if (w) {
4246 w->cb = w_ov_finished;
4247 drbd_queue_work_front(&mdev->data.work, w);
4248 } else {
4249 dev_err(DEV, "kmalloc(w) failed.");
4250 ov_oos_print(mdev);
4251 drbd_resync_finished(mdev);
4252 }
4253 }
4254 return TRUE;
4255}
4256
4257struct asender_cmd {
4258 size_t pkt_size;
4259 int (*process)(struct drbd_conf *mdev, struct p_header *h);
4260};
4261
4262static struct asender_cmd *get_asender_cmd(int cmd)
4263{
4264 static struct asender_cmd asender_tbl[] = {
4265 /* anything missing from this table is in
4266 * the drbd_cmd_handler (drbd_default_handler) table,
4267 * see the beginning of drbdd() */
4268 [P_PING] = { sizeof(struct p_header), got_Ping },
4269 [P_PING_ACK] = { sizeof(struct p_header), got_PingAck },
4270 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4271 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4272 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4273 [P_DISCARD_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4274 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
4275 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
4276 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply},
4277 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
4278 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
4279 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
4280 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
4281 [P_MAX_CMD] = { 0, NULL },
4282 };
4283 if (cmd > P_MAX_CMD || asender_tbl[cmd].process == NULL)
4284 return NULL;
4285 return &asender_tbl[cmd];
4286}
4287
4288int drbd_asender(struct drbd_thread *thi)
4289{
4290 struct drbd_conf *mdev = thi->mdev;
4291 struct p_header *h = &mdev->meta.rbuf.header;
4292 struct asender_cmd *cmd = NULL;
4293
4294 int rv, len;
4295 void *buf = h;
4296 int received = 0;
4297 int expect = sizeof(struct p_header);
4298 int empty;
4299
4300 sprintf(current->comm, "drbd%d_asender", mdev_to_minor(mdev));
4301
4302 current->policy = SCHED_RR; /* Make this a realtime task! */
4303 current->rt_priority = 2; /* more important than all other tasks */
4304
4305 while (get_t_state(thi) == Running) {
4306 drbd_thread_current_set_cpu(mdev);
4307 if (test_and_clear_bit(SEND_PING, &mdev->flags)) {
4308 ERR_IF(!drbd_send_ping(mdev)) goto reconnect;
4309 mdev->meta.socket->sk->sk_rcvtimeo =
4310 mdev->net_conf->ping_timeo*HZ/10;
4311 }
4312
4313 /* conditionally cork;
4314 * it may hurt latency if we cork without much to send */
4315 if (!mdev->net_conf->no_cork &&
4316 3 < atomic_read(&mdev->unacked_cnt))
4317 drbd_tcp_cork(mdev->meta.socket);
4318 while (1) {
4319 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4320 flush_signals(current);
4321 if (!drbd_process_done_ee(mdev)) {
4322 dev_err(DEV, "process_done_ee() = NOT_OK\n");
4323 goto reconnect;
4324 }
4325 /* to avoid race with newly queued ACKs */
4326 set_bit(SIGNAL_ASENDER, &mdev->flags);
4327 spin_lock_irq(&mdev->req_lock);
4328 empty = list_empty(&mdev->done_ee);
4329 spin_unlock_irq(&mdev->req_lock);
4330 /* new ack may have been queued right here,
4331 * but then there is also a signal pending,
4332 * and we start over... */
4333 if (empty)
4334 break;
4335 }
4336 /* but unconditionally uncork unless disabled */
4337 if (!mdev->net_conf->no_cork)
4338 drbd_tcp_uncork(mdev->meta.socket);
4339
4340 /* short circuit, recv_msg would return EINTR anyways. */
4341 if (signal_pending(current))
4342 continue;
4343
4344 rv = drbd_recv_short(mdev, mdev->meta.socket,
4345 buf, expect-received, 0);
4346 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4347
4348 flush_signals(current);
4349
4350 /* Note:
4351 * -EINTR (on meta) we got a signal
4352 * -EAGAIN (on meta) rcvtimeo expired
4353 * -ECONNRESET other side closed the connection
4354 * -ERESTARTSYS (on data) we got a signal
4355 * rv < 0 other than above: unexpected error!
4356 * rv == expected: full header or command
4357 * rv < expected: "woken" by signal during receive
4358 * rv == 0 : "connection shut down by peer"
4359 */
4360 if (likely(rv > 0)) {
4361 received += rv;
4362 buf += rv;
4363 } else if (rv == 0) {
4364 dev_err(DEV, "meta connection shut down by peer.\n");
4365 goto reconnect;
4366 } else if (rv == -EAGAIN) {
4367 if (mdev->meta.socket->sk->sk_rcvtimeo ==
4368 mdev->net_conf->ping_timeo*HZ/10) {
4369 dev_err(DEV, "PingAck did not arrive in time.\n");
4370 goto reconnect;
4371 }
4372 set_bit(SEND_PING, &mdev->flags);
4373 continue;
4374 } else if (rv == -EINTR) {
4375 continue;
4376 } else {
4377 dev_err(DEV, "sock_recvmsg returned %d\n", rv);
4378 goto reconnect;
4379 }
4380
4381 if (received == expect && cmd == NULL) {
4382 if (unlikely(h->magic != BE_DRBD_MAGIC)) {
4383 dev_err(DEV, "magic?? on meta m: 0x%lx c: %d l: %d\n",
4384 (long)be32_to_cpu(h->magic),
4385 h->command, h->length);
4386 goto reconnect;
4387 }
4388 cmd = get_asender_cmd(be16_to_cpu(h->command));
4389 len = be16_to_cpu(h->length);
4390 if (unlikely(cmd == NULL)) {
4391 dev_err(DEV, "unknown command?? on meta m: 0x%lx c: %d l: %d\n",
4392 (long)be32_to_cpu(h->magic),
4393 h->command, h->length);
4394 goto disconnect;
4395 }
4396 expect = cmd->pkt_size;
4397 ERR_IF(len != expect-sizeof(struct p_header))
4398 goto reconnect;
4399 }
4400 if (received == expect) {
4401 D_ASSERT(cmd != NULL);
4402 if (!cmd->process(mdev, h))
4403 goto reconnect;
4404
4405 buf = h;
4406 received = 0;
4407 expect = sizeof(struct p_header);
4408 cmd = NULL;
4409 }
4410 }
4411
4412 if (0) {
4413reconnect:
4414 drbd_force_state(mdev, NS(conn, C_NETWORK_FAILURE));
4415 }
4416 if (0) {
4417disconnect:
4418 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
4419 }
4420 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4421
4422 D_ASSERT(mdev->state.conn < C_CONNECTED);
4423 dev_info(DEV, "asender terminated\n");
4424
4425 return 0;
4426}
generated by cgit v1.2.2 (git 2.25.0) at 2024-12-18 03:34:54 -0500