?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
On Tue, Nov 06, 2007 at 02:33:53AM -0800, akpm@linux-foundation.org wrote:
[mmotm.git] / drivers / block / drbd / drbd_receiver.c
blob360baf60f574a0b67154bfc5c1eb3afe28316719
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
54 struct flush_work {
55 struct drbd_work w;
56 struct drbd_epoch *epoch;
57 };
58
59 enum finish_epoch {
60 FE_STILL_LIVE,
61 FE_DESTROYED,
62 FE_RECYCLED,
63 };
64
65 static int drbd_do_handshake(struct drbd_conf *mdev);
66 static int drbd_do_auth(struct drbd_conf *mdev);
67
68 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *, struct drbd_epoch *, enum epoch_event);
69 static int e_end_block(struct drbd_conf *, struct drbd_work *, int);
70
71 static 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
84 static 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. */
113 static 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
119 static 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
137 static 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 */
160 static 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) */
199 static 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
222 static 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 /*
254 You need to hold the req_lock:
255 _drbd_wait_ee_list_empty()
256
257 You 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
268 struct 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
363 void 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
372 int 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 */
399 static 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
428 void _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
444 void 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 */
453 static 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
479 out:
480 return err;
481 }
482
483 static 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
506 static 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
559 static 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
610 out:
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
635 static 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
668 out:
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
682 static 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
690 static 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 */
708 static 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 */
735 static 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
788 retry:
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
909 out_release_sockets:
910 if (sock)
911 sock_release(sock);
912 if (msock)
913 sock_release(msock);
914 return -1;
915 }
916
917 static 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
940 static 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
959 static 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 */
981 static 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 */
1095 void 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 */
1124 int 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
1172 static 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 */
1256 static struct drbd_epoch_entry *
1257 read_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 */
1322 static 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
1346 static 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 */
1405 static 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
1427 static 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
1458 static 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
1498 static 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 */
1536 static 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
1581 static 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). */
1620 static 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 */
1653 static 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
1899 out_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
1908 static 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
2056 out_free_e:
2057 kfree(di);
2058 put_ldev(mdev);
2059 drbd_free_ee(mdev, e);
2060 return FALSE;
2061 }
2062
2063 static 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 found 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
2135 static 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
2185 static 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
2227 static 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 */
2254 static 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 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2404 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2405 if (self == peer)
2406 return 2;
2407 }
2408
2409 *rule_nr = 90;
2410 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2411 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2412 if (self == peer && self != ((u64)0))
2413 return 100;
2414
2415 *rule_nr = 100;
2416 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2417 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2418 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
2419 peer = mdev->p_uuid[j] & ~((u64)1);
2420 if (self == peer)
2421 return -100;
2422 }
2423 }
2424
2425 return -1000;
2426 }
2427
2428 /* drbd_sync_handshake() returns the new conn state on success, or
2429 CONN_MASK (-1) on failure.
2430 */
2431 static enum drbd_conns drbd_sync_handshake(struct drbd_conf *mdev, enum drbd_role peer_role,
2432 enum drbd_disk_state peer_disk) __must_hold(local)
2433 {
2434 int hg, rule_nr;
2435 enum drbd_conns rv = C_MASK;
2436 enum drbd_disk_state mydisk;
2437
2438 mydisk = mdev->state.disk;
2439 if (mydisk == D_NEGOTIATING)
2440 mydisk = mdev->new_state_tmp.disk;
2441
2442 dev_info(DEV, "drbd_sync_handshake:\n");
2443 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, mdev->comm_bm_set, 0);
2444 drbd_uuid_dump(mdev, "peer", mdev->p_uuid,
2445 mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2446
2447 hg = drbd_uuid_compare(mdev, &rule_nr);
2448
2449 dev_info(DEV, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
2450
2451 if (hg == -1000) {
2452 dev_alert(DEV, "Unrelated data, aborting!\n");
2453 return C_MASK;
2454 }
2455 if (hg == -1001) {
2456 dev_alert(DEV, "To resolve this both sides have to support at least protocol\n");
2457 return C_MASK;
2458 }
2459
2460 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
2461 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) {
2462 int f = (hg == -100) || abs(hg) == 2;
2463 hg = mydisk > D_INCONSISTENT ? 1 : -1;
2464 if (f)
2465 hg = hg*2;
2466 dev_info(DEV, "Becoming sync %s due to disk states.\n",
2467 hg > 0 ? "source" : "target");
2468 }
2469
2470 if (hg == 100 || (hg == -100 && mdev->net_conf->always_asbp)) {
2471 int pcount = (mdev->state.role == R_PRIMARY)
2472 + (peer_role == R_PRIMARY);
2473 int forced = (hg == -100);
2474
2475 switch (pcount) {
2476 case 0:
2477 hg = drbd_asb_recover_0p(mdev);
2478 break;
2479 case 1:
2480 hg = drbd_asb_recover_1p(mdev);
2481 break;
2482 case 2:
2483 hg = drbd_asb_recover_2p(mdev);
2484 break;
2485 }
2486 if (abs(hg) < 100) {
2487 dev_warn(DEV, "Split-Brain detected, %d primaries, "
2488 "automatically solved. Sync from %s node\n",
2489 pcount, (hg < 0) ? "peer" : "this");
2490 if (forced) {
2491 dev_warn(DEV, "Doing a full sync, since"
2492 " UUIDs where ambiguous.\n");
2493 hg = hg*2;
2494 }
2495 }
2496 }
2497
2498 if (hg == -100) {
2499 if (mdev->net_conf->want_lose && !(mdev->p_uuid[UI_FLAGS]&1))
2500 hg = -1;
2501 if (!mdev->net_conf->want_lose && (mdev->p_uuid[UI_FLAGS]&1))
2502 hg = 1;
2503
2504 if (abs(hg) < 100)
2505 dev_warn(DEV, "Split-Brain detected, manually solved. "
2506 "Sync from %s node\n",
2507 (hg < 0) ? "peer" : "this");
2508 }
2509
2510 if (hg == -100) {
2511 dev_alert(DEV, "Split-Brain detected, dropping connection!\n");
2512 drbd_khelper(mdev, "split-brain");
2513 return C_MASK;
2514 }
2515
2516 if (hg > 0 && mydisk <= D_INCONSISTENT) {
2517 dev_err(DEV, "I shall become SyncSource, but I am inconsistent!\n");
2518 return C_MASK;
2519 }
2520
2521 if (hg < 0 && /* by intention we do not use mydisk here. */
2522 mdev->state.role == R_PRIMARY && mdev->state.disk >= D_CONSISTENT) {
2523 switch (mdev->net_conf->rr_conflict) {
2524 case ASB_CALL_HELPER:
2525 drbd_khelper(mdev, "pri-lost");
2526 /* fall through */
2527 case ASB_DISCONNECT:
2528 dev_err(DEV, "I shall become SyncTarget, but I am primary!\n");
2529 return C_MASK;
2530 case ASB_VIOLENTLY:
2531 dev_warn(DEV, "Becoming SyncTarget, violating the stable-data"
2532 "assumption\n");
2533 }
2534 }
2535
2536 if (abs(hg) >= 2) {
2537 dev_info(DEV, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2538 if (drbd_bitmap_io(mdev, &drbd_bmio_set_n_write, "set_n_write from sync_handshake"))
2539 return C_MASK;
2540 }
2541
2542 if (hg > 0) { /* become sync source. */
2543 rv = C_WF_BITMAP_S;
2544 } else if (hg < 0) { /* become sync target */
2545 rv = C_WF_BITMAP_T;
2546 } else {
2547 rv = C_CONNECTED;
2548 if (drbd_bm_total_weight(mdev)) {
2549 dev_info(DEV, "No resync, but %lu bits in bitmap!\n",
2550 drbd_bm_total_weight(mdev));
2551 }
2552 }
2553
2554 return rv;
2555 }
2556
2557 /* returns 1 if invalid */
2558 static int cmp_after_sb(enum drbd_after_sb_p peer, enum drbd_after_sb_p self)
2559 {
2560 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2561 if ((peer == ASB_DISCARD_REMOTE && self == ASB_DISCARD_LOCAL) ||
2562 (self == ASB_DISCARD_REMOTE && peer == ASB_DISCARD_LOCAL))
2563 return 0;
2564
2565 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2566 if (peer == ASB_DISCARD_REMOTE || peer == ASB_DISCARD_LOCAL ||
2567 self == ASB_DISCARD_REMOTE || self == ASB_DISCARD_LOCAL)
2568 return 1;
2569
2570 /* everything else is valid if they are equal on both sides. */
2571 if (peer == self)
2572 return 0;
2573
2574 /* everything es is invalid. */
2575 return 1;
2576 }
2577
2578 static int receive_protocol(struct drbd_conf *mdev, struct p_header *h)
2579 {
2580 struct p_protocol *p = (struct p_protocol *)h;
2581 int header_size, data_size;
2582 int p_proto, p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
2583 int p_want_lose, p_two_primaries;
2584 char p_integrity_alg[SHARED_SECRET_MAX] = "";
2585
2586 header_size = sizeof(*p) - sizeof(*h);
2587 data_size = h->length - header_size;
2588
2589 if (drbd_recv(mdev, h->payload, header_size) != header_size)
2590 return FALSE;
2591
2592 p_proto = be32_to_cpu(p->protocol);
2593 p_after_sb_0p = be32_to_cpu(p->after_sb_0p);
2594 p_after_sb_1p = be32_to_cpu(p->after_sb_1p);
2595 p_after_sb_2p = be32_to_cpu(p->after_sb_2p);
2596 p_want_lose = be32_to_cpu(p->want_lose);
2597 p_two_primaries = be32_to_cpu(p->two_primaries);
2598
2599 if (p_proto != mdev->net_conf->wire_protocol) {
2600 dev_err(DEV, "incompatible communication protocols\n");
2601 goto disconnect;
2602 }
2603
2604 if (cmp_after_sb(p_after_sb_0p, mdev->net_conf->after_sb_0p)) {
2605 dev_err(DEV, "incompatible after-sb-0pri settings\n");
2606 goto disconnect;
2607 }
2608
2609 if (cmp_after_sb(p_after_sb_1p, mdev->net_conf->after_sb_1p)) {
2610 dev_err(DEV, "incompatible after-sb-1pri settings\n");
2611 goto disconnect;
2612 }
2613
2614 if (cmp_after_sb(p_after_sb_2p, mdev->net_conf->after_sb_2p)) {
2615 dev_err(DEV, "incompatible after-sb-2pri settings\n");
2616 goto disconnect;
2617 }
2618
2619 if (p_want_lose && mdev->net_conf->want_lose) {
2620 dev_err(DEV, "both sides have the 'want_lose' flag set\n");
2621 goto disconnect;
2622 }
2623
2624 if (p_two_primaries != mdev->net_conf->two_primaries) {
2625 dev_err(DEV, "incompatible setting of the two-primaries options\n");
2626 goto disconnect;
2627 }
2628
2629 if (mdev->agreed_pro_version >= 87) {
2630 unsigned char *my_alg = mdev->net_conf->integrity_alg;
2631
2632 if (drbd_recv(mdev, p_integrity_alg, data_size) != data_size)
2633 return FALSE;
2634
2635 p_integrity_alg[SHARED_SECRET_MAX-1] = 0;
2636 if (strcmp(p_integrity_alg, my_alg)) {
2637 dev_err(DEV, "incompatible setting of the data-integrity-alg\n");
2638 goto disconnect;
2639 }
2640 dev_info(DEV, "data-integrity-alg: %s\n",
2641 my_alg[0] ? my_alg : (unsigned char *)"<not-used>");
2642 }
2643
2644 return TRUE;
2645
2646 disconnect:
2647 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2648 return FALSE;
2649 }
2650
2651 /* helper function
2652 * input: alg name, feature name
2653 * return: NULL (alg name was "")
2654 * ERR_PTR(error) if something goes wrong
2655 * or the crypto hash ptr, if it worked out ok. */
2656 struct crypto_hash *drbd_crypto_alloc_digest_safe(const struct drbd_conf *mdev,
2657 const char *alg, const char *name)
2658 {
2659 struct crypto_hash *tfm;
2660
2661 if (!alg[0])
2662 return NULL;
2663
2664 tfm = crypto_alloc_hash(alg, 0, CRYPTO_ALG_ASYNC);
2665 if (IS_ERR(tfm)) {
2666 dev_err(DEV, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2667 alg, name, PTR_ERR(tfm));
2668 return tfm;
2669 }
2670 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm))) {
2671 crypto_free_hash(tfm);
2672 dev_err(DEV, "\"%s\" is not a digest (%s)\n", alg, name);
2673 return ERR_PTR(-EINVAL);
2674 }
2675 return tfm;
2676 }
2677
2678 static int receive_SyncParam(struct drbd_conf *mdev, struct p_header *h)
2679 {
2680 int ok = TRUE;
2681 struct p_rs_param_89 *p = (struct p_rs_param_89 *)h;
2682 unsigned int header_size, data_size, exp_max_sz;
2683 struct crypto_hash *verify_tfm = NULL;
2684 struct crypto_hash *csums_tfm = NULL;
2685 const int apv = mdev->agreed_pro_version;
2686
2687 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param)
2688 : apv == 88 ? sizeof(struct p_rs_param)
2689 + SHARED_SECRET_MAX
2690 : /* 89 */ sizeof(struct p_rs_param_89);
2691
2692 if (h->length > exp_max_sz) {
2693 dev_err(DEV, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2694 h->length, exp_max_sz);
2695 return FALSE;
2696 }
2697
2698 if (apv <= 88) {
2699 header_size = sizeof(struct p_rs_param) - sizeof(*h);
2700 data_size = h->length - header_size;
2701 } else /* apv >= 89 */ {
2702 header_size = sizeof(struct p_rs_param_89) - sizeof(*h);
2703 data_size = h->length - header_size;
2704 D_ASSERT(data_size == 0);
2705 }
2706
2707 /* initialize verify_alg and csums_alg */
2708 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
2709
2710 if (drbd_recv(mdev, h->payload, header_size) != header_size)
2711 return FALSE;
2712
2713 mdev->sync_conf.rate = be32_to_cpu(p->rate);
2714
2715 if (apv >= 88) {
2716 if (apv == 88) {
2717 if (data_size > SHARED_SECRET_MAX) {
2718 dev_err(DEV, "verify-alg too long, "
2719 "peer wants %u, accepting only %u byte\n",
2720 data_size, SHARED_SECRET_MAX);
2721 return FALSE;
2722 }
2723
2724 if (drbd_recv(mdev, p->verify_alg, data_size) != data_size)
2725 return FALSE;
2726
2727 /* we expect NUL terminated string */
2728 /* but just in case someone tries to be evil */
2729 D_ASSERT(p->verify_alg[data_size-1] == 0);
2730 p->verify_alg[data_size-1] = 0;
2731
2732 } else /* apv >= 89 */ {
2733 /* we still expect NUL terminated strings */
2734 /* but just in case someone tries to be evil */
2735 D_ASSERT(p->verify_alg[SHARED_SECRET_MAX-1] == 0);
2736 D_ASSERT(p->csums_alg[SHARED_SECRET_MAX-1] == 0);
2737 p->verify_alg[SHARED_SECRET_MAX-1] = 0;
2738 p->csums_alg[SHARED_SECRET_MAX-1] = 0;
2739 }
2740
2741 if (strcmp(mdev->sync_conf.verify_alg, p->verify_alg)) {
2742 if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2743 dev_err(DEV, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2744 mdev->sync_conf.verify_alg, p->verify_alg);
2745 goto disconnect;
2746 }
2747 verify_tfm = drbd_crypto_alloc_digest_safe(mdev,
2748 p->verify_alg, "verify-alg");
2749 if (IS_ERR(verify_tfm)) {
2750 verify_tfm = NULL;
2751 goto disconnect;
2752 }
2753 }
2754
2755 if (apv >= 89 && strcmp(mdev->sync_conf.csums_alg, p->csums_alg)) {
2756 if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2757 dev_err(DEV, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2758 mdev->sync_conf.csums_alg, p->csums_alg);
2759 goto disconnect;
2760 }
2761 csums_tfm = drbd_crypto_alloc_digest_safe(mdev,
2762 p->csums_alg, "csums-alg");
2763 if (IS_ERR(csums_tfm)) {
2764 csums_tfm = NULL;
2765 goto disconnect;
2766 }
2767 }
2768
2769
2770 spin_lock(&mdev->peer_seq_lock);
2771 /* lock against drbd_nl_syncer_conf() */
2772 if (verify_tfm) {
2773 strcpy(mdev->sync_conf.verify_alg, p->verify_alg);
2774 mdev->sync_conf.verify_alg_len = strlen(p->verify_alg) + 1;
2775 crypto_free_hash(mdev->verify_tfm);
2776 mdev->verify_tfm = verify_tfm;
2777 dev_info(DEV, "using verify-alg: \"%s\"\n", p->verify_alg);
2778 }
2779 if (csums_tfm) {
2780 strcpy(mdev->sync_conf.csums_alg, p->csums_alg);
2781 mdev->sync_conf.csums_alg_len = strlen(p->csums_alg) + 1;
2782 crypto_free_hash(mdev->csums_tfm);
2783 mdev->csums_tfm = csums_tfm;
2784 dev_info(DEV, "using csums-alg: \"%s\"\n", p->csums_alg);
2785 }
2786 spin_unlock(&mdev->peer_seq_lock);
2787 }
2788
2789 return ok;
2790 disconnect:
2791 /* just for completeness: actually not needed,
2792 * as this is not reached if csums_tfm was ok. */
2793 crypto_free_hash(csums_tfm);
2794 /* but free the verify_tfm again, if csums_tfm did not work out */
2795 crypto_free_hash(verify_tfm);
2796 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2797 return FALSE;
2798 }
2799
2800 static void drbd_setup_order_type(struct drbd_conf *mdev, int peer)
2801 {
2802 /* sorry, we currently have no working implementation
2803 * of distributed TCQ */
2804 }
2805
2806 /* warn if the arguments differ by more than 12.5% */
2807 static void warn_if_differ_considerably(struct drbd_conf *mdev,
2808 const char *s, sector_t a, sector_t b)
2809 {
2810 sector_t d;
2811 if (a == 0 || b == 0)
2812 return;
2813 d = (a > b) ? (a - b) : (b - a);
2814 if (d > (a>>3) || d > (b>>3))
2815 dev_warn(DEV, "Considerable difference in %s: %llus vs. %llus\n", s,
2816 (unsigned long long)a, (unsigned long long)b);
2817 }
2818
2819 static int receive_sizes(struct drbd_conf *mdev, struct p_header *h)
2820 {
2821 struct p_sizes *p = (struct p_sizes *)h;
2822 enum determine_dev_size dd = unchanged;
2823 unsigned int max_seg_s;
2824 sector_t p_size, p_usize, my_usize;
2825 int ldsc = 0; /* local disk size changed */
2826 enum drbd_conns nconn;
2827
2828 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
2829 if (drbd_recv(mdev, h->payload, h->length) != h->length)
2830 return FALSE;
2831
2832 p_size = be64_to_cpu(p->d_size);
2833 p_usize = be64_to_cpu(p->u_size);
2834
2835 if (p_size == 0 && mdev->state.disk == D_DISKLESS) {
2836 dev_err(DEV, "some backing storage is needed\n");
2837 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2838 return FALSE;
2839 }
2840
2841 /* just store the peer's disk size for now.
2842 * we still need to figure out whether we accept that. */
2843 mdev->p_size = p_size;
2844
2845 #define min_not_zero(l, r) (l == 0) ? r : ((r == 0) ? l : min(l, r))
2846 if (get_ldev(mdev)) {
2847 warn_if_differ_considerably(mdev, "lower level device sizes",
2848 p_size, drbd_get_max_capacity(mdev->ldev));
2849 warn_if_differ_considerably(mdev, "user requested size",
2850 p_usize, mdev->ldev->dc.disk_size);
2851
2852 /* if this is the first connect, or an otherwise expected
2853 * param exchange, choose the minimum */
2854 if (mdev->state.conn == C_WF_REPORT_PARAMS)
2855 p_usize = min_not_zero((sector_t)mdev->ldev->dc.disk_size,
2856 p_usize);
2857
2858 my_usize = mdev->ldev->dc.disk_size;
2859
2860 if (mdev->ldev->dc.disk_size != p_usize) {
2861 mdev->ldev->dc.disk_size = p_usize;
2862 dev_info(DEV, "Peer sets u_size to %lu sectors\n",
2863 (unsigned long)mdev->ldev->dc.disk_size);
2864 }
2865
2866 /* Never shrink a device with usable data during connect.
2867 But allow online shrinking if we are connected. */
2868 if (drbd_new_dev_size(mdev, mdev->ldev) <
2869 drbd_get_capacity(mdev->this_bdev) &&
2870 mdev->state.disk >= D_OUTDATED &&
2871 mdev->state.conn < C_CONNECTED) {
2872 dev_err(DEV, "The peer's disk size is too small!\n");
2873 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2874 mdev->ldev->dc.disk_size = my_usize;
2875 put_ldev(mdev);
2876 return FALSE;
2877 }
2878 put_ldev(mdev);
2879 }
2880 #undef min_not_zero
2881
2882 if (get_ldev(mdev)) {
2883 dd = drbd_determin_dev_size(mdev);
2884 put_ldev(mdev);
2885 if (dd == dev_size_error)
2886 return FALSE;
2887 drbd_md_sync(mdev);
2888 } else {
2889 /* I am diskless, need to accept the peer's size. */
2890 drbd_set_my_capacity(mdev, p_size);
2891 }
2892
2893 if (mdev->p_uuid && mdev->state.conn <= C_CONNECTED && get_ldev(mdev)) {
2894 nconn = drbd_sync_handshake(mdev,
2895 mdev->state.peer, mdev->state.pdsk);
2896 put_ldev(mdev);
2897
2898 if (nconn == C_MASK) {
2899 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2900 return FALSE;
2901 }
2902
2903 if (drbd_request_state(mdev, NS(conn, nconn)) < SS_SUCCESS) {
2904 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2905 return FALSE;
2906 }
2907 }
2908
2909 if (get_ldev(mdev)) {
2910 if (mdev->ldev->known_size != drbd_get_capacity(mdev->ldev->backing_bdev)) {
2911 mdev->ldev->known_size = drbd_get_capacity(mdev->ldev->backing_bdev);
2912 ldsc = 1;
2913 }
2914
2915 max_seg_s = be32_to_cpu(p->max_segment_size);
2916 if (max_seg_s != queue_max_segment_size(mdev->rq_queue))
2917 drbd_setup_queue_param(mdev, max_seg_s);
2918
2919 drbd_setup_order_type(mdev, be32_to_cpu(p->queue_order_type));
2920 put_ldev(mdev);
2921 }
2922
2923 if (mdev->state.conn > C_WF_REPORT_PARAMS) {
2924 if (be64_to_cpu(p->c_size) !=
2925 drbd_get_capacity(mdev->this_bdev) || ldsc) {
2926 /* we have different sizes, probably peer
2927 * needs to know my new size... */
2928 drbd_send_sizes(mdev, 0);
2929 }
2930 if (test_and_clear_bit(RESIZE_PENDING, &mdev->flags) ||
2931 (dd == grew && mdev->state.conn == C_CONNECTED)) {
2932 if (mdev->state.pdsk >= D_INCONSISTENT &&
2933 mdev->state.disk >= D_INCONSISTENT)
2934 resync_after_online_grow(mdev);
2935 else
2936 set_bit(RESYNC_AFTER_NEG, &mdev->flags);
2937 }
2938 }
2939
2940 return TRUE;
2941 }
2942
2943 static int receive_uuids(struct drbd_conf *mdev, struct p_header *h)
2944 {
2945 struct p_uuids *p = (struct p_uuids *)h;
2946 u64 *p_uuid;
2947 int i;
2948
2949 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
2950 if (drbd_recv(mdev, h->payload, h->length) != h->length)
2951 return FALSE;
2952
2953 p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
2954
2955 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
2956 p_uuid[i] = be64_to_cpu(p->uuid[i]);
2957
2958 kfree(mdev->p_uuid);
2959 mdev->p_uuid = p_uuid;
2960
2961 if (mdev->state.conn < C_CONNECTED &&
2962 mdev->state.disk < D_INCONSISTENT &&
2963 mdev->state.role == R_PRIMARY &&
2964 (mdev->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
2965 dev_err(DEV, "Can only connect to data with current UUID=%016llX\n",
2966 (unsigned long long)mdev->ed_uuid);
2967 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2968 return FALSE;
2969 }
2970
2971 if (get_ldev(mdev)) {
2972 int skip_initial_sync =
2973 mdev->state.conn == C_CONNECTED &&
2974 mdev->agreed_pro_version >= 90 &&
2975 mdev->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
2976 (p_uuid[UI_FLAGS] & 8);
2977 if (skip_initial_sync) {
2978 dev_info(DEV, "Accepted new current UUID, preparing to skip initial sync\n");
2979 drbd_bitmap_io(mdev, &drbd_bmio_clear_n_write,
2980 "clear_n_write from receive_uuids");
2981 _drbd_uuid_set(mdev, UI_CURRENT, p_uuid[UI_CURRENT]);
2982 _drbd_uuid_set(mdev, UI_BITMAP, 0);
2983 _drbd_set_state(_NS2(mdev, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
2984 CS_VERBOSE, NULL);
2985 drbd_md_sync(mdev);
2986 }
2987 put_ldev(mdev);
2988 }
2989
2990 /* Before we test for the disk state, we should wait until an eventually
2991 ongoing cluster wide state change is finished. That is important if
2992 we are primary and are detaching from our disk. We need to see the
2993 new disk state... */
2994 wait_event(mdev->misc_wait, !test_bit(CLUSTER_ST_CHANGE, &mdev->flags));
2995 if (mdev->state.conn >= C_CONNECTED && mdev->state.disk < D_INCONSISTENT)
2996 drbd_set_ed_uuid(mdev, p_uuid[UI_CURRENT]);
2997
2998 return TRUE;
2999 }
3000
3001 /**
3002 * convert_state() - Converts the peer's view of the cluster state to our point of view
3003 * @ps: The state as seen by the peer.
3004 */
3005 static union drbd_state convert_state(union drbd_state ps)
3006 {
3007 union drbd_state ms;
3008
3009 static enum drbd_conns c_tab[] = {
3010 [C_CONNECTED] = C_CONNECTED,
3011
3012 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
3013 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
3014 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
3015 [C_VERIFY_S] = C_VERIFY_T,
3016 [C_MASK] = C_MASK,
3017 };
3018
3019 ms.i = ps.i;
3020
3021 ms.conn = c_tab[ps.conn];
3022 ms.peer = ps.role;
3023 ms.role = ps.peer;
3024 ms.pdsk = ps.disk;
3025 ms.disk = ps.pdsk;
3026 ms.peer_isp = (ps.aftr_isp | ps.user_isp);
3027
3028 return ms;
3029 }
3030
3031 static int receive_req_state(struct drbd_conf *mdev, struct p_header *h)
3032 {
3033 struct p_req_state *p = (struct p_req_state *)h;
3034 union drbd_state mask, val;
3035 int rv;
3036
3037 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
3038 if (drbd_recv(mdev, h->payload, h->length) != h->length)
3039 return FALSE;
3040
3041 mask.i = be32_to_cpu(p->mask);
3042 val.i = be32_to_cpu(p->val);
3043
3044 if (test_bit(DISCARD_CONCURRENT, &mdev->flags) &&
3045 test_bit(CLUSTER_ST_CHANGE, &mdev->flags)) {
3046 drbd_send_sr_reply(mdev, SS_CONCURRENT_ST_CHG);
3047 return TRUE;
3048 }
3049
3050 mask = convert_state(mask);
3051 val = convert_state(val);
3052
3053 rv = drbd_change_state(mdev, CS_VERBOSE, mask, val);
3054
3055 drbd_send_sr_reply(mdev, rv);
3056 drbd_md_sync(mdev);
3057
3058 return TRUE;
3059 }
3060
3061 static int receive_state(struct drbd_conf *mdev, struct p_header *h)
3062 {
3063 struct p_state *p = (struct p_state *)h;
3064 enum drbd_conns nconn, oconn;
3065 union drbd_state ns, peer_state;
3066 enum drbd_disk_state real_peer_disk;
3067 int rv;
3068
3069 ERR_IF(h->length != (sizeof(*p)-sizeof(*h)))
3070 return FALSE;
3071
3072 if (drbd_recv(mdev, h->payload, h->length) != h->length)
3073 return FALSE;
3074
3075 peer_state.i = be32_to_cpu(p->state);
3076
3077 real_peer_disk = peer_state.disk;
3078 if (peer_state.disk == D_NEGOTIATING) {
3079 real_peer_disk = mdev->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
3080 dev_info(DEV, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
3081 }
3082
3083 spin_lock_irq(&mdev->req_lock);
3084 retry:
3085 oconn = nconn = mdev->state.conn;
3086 spin_unlock_irq(&mdev->req_lock);
3087
3088 if (nconn == C_WF_REPORT_PARAMS)
3089 nconn = C_CONNECTED;
3090
3091 if (mdev->p_uuid && peer_state.disk >= D_NEGOTIATING &&
3092 get_ldev_if_state(mdev, D_NEGOTIATING)) {
3093 int cr; /* consider resync */
3094
3095 /* if we established a new connection */
3096 cr = (oconn < C_CONNECTED);
3097 /* if we had an established connection
3098 * and one of the nodes newly attaches a disk */
3099 cr |= (oconn == C_CONNECTED &&
3100 (peer_state.disk == D_NEGOTIATING ||
3101 mdev->state.disk == D_NEGOTIATING));
3102 /* if we have both been inconsistent, and the peer has been
3103 * forced to be UpToDate with --overwrite-data */
3104 cr |= test_bit(CONSIDER_RESYNC, &mdev->flags);
3105 /* if we had been plain connected, and the admin requested to
3106 * start a sync by "invalidate" or "invalidate-remote" */
3107 cr |= (oconn == C_CONNECTED &&
3108 (peer_state.conn >= C_STARTING_SYNC_S &&
3109 peer_state.conn <= C_WF_BITMAP_T));
3110
3111 if (cr)
3112 nconn = drbd_sync_handshake(mdev, peer_state.role, real_peer_disk);
3113
3114 put_ldev(mdev);
3115 if (nconn == C_MASK) {
3116 if (mdev->state.disk == D_NEGOTIATING) {
3117 drbd_force_state(mdev, NS(disk, D_DISKLESS));
3118 nconn = C_CONNECTED;
3119 } else if (peer_state.disk == D_NEGOTIATING) {
3120 dev_err(DEV, "Disk attach process on the peer node was aborted.\n");
3121 peer_state.disk = D_DISKLESS;
3122 } else {
3123 D_ASSERT(oconn == C_WF_REPORT_PARAMS);
3124 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3125 return FALSE;
3126 }
3127 }
3128 }
3129
3130 spin_lock_irq(&mdev->req_lock);
3131 if (mdev->state.conn != oconn)
3132 goto retry;
3133 clear_bit(CONSIDER_RESYNC, &mdev->flags);
3134 ns.i = mdev->state.i;
3135 ns.conn = nconn;
3136 ns.peer = peer_state.role;
3137 ns.pdsk = real_peer_disk;
3138 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
3139 if ((nconn == C_CONNECTED || nconn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
3140 ns.disk = mdev->new_state_tmp.disk;
3141
3142 rv = _drbd_set_state(mdev, ns, CS_VERBOSE | CS_HARD, NULL);
3143 ns = mdev->state;
3144 spin_unlock_irq(&mdev->req_lock);
3145
3146 if (rv < SS_SUCCESS) {
3147 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3148 return FALSE;
3149 }
3150
3151 if (oconn > C_WF_REPORT_PARAMS) {
3152 if (nconn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
3153 peer_state.disk != D_NEGOTIATING ) {
3154 /* we want resync, peer has not yet decided to sync... */
3155 /* Nowadays only used when forcing a node into primary role and
3156 setting its disk to UpToDate with that */
3157 drbd_send_uuids(mdev);
3158 drbd_send_state(mdev);
3159 }
3160 }
3161
3162 mdev->net_conf->want_lose = 0;
3163
3164 drbd_md_sync(mdev); /* update connected indicator, la_size, ... */
3165
3166 return TRUE;
3167 }
3168
3169 static int receive_sync_uuid(struct drbd_conf *mdev, struct p_header *h)
3170 {
3171 struct p_rs_uuid *p = (struct p_rs_uuid *)h;
3172
3173 wait_event(mdev->misc_wait,
3174 mdev->state.conn == C_WF_SYNC_UUID ||
3175 mdev->state.conn < C_CONNECTED ||
3176 mdev->state.disk < D_NEGOTIATING);
3177
3178 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */
3179
3180 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
3181 if (drbd_recv(mdev, h->payload, h->length) != h->length)
3182 return FALSE;
3183
3184 /* Here the _drbd_uuid_ functions are right, current should
3185 _not_ be rotated into the history */
3186 if (get_ldev_if_state(mdev, D_NEGOTIATING)) {
3187 _drbd_uuid_set(mdev, UI_CURRENT, be64_to_cpu(p->uuid));
3188 _drbd_uuid_set(mdev, UI_BITMAP, 0UL);
3189
3190 drbd_start_resync(mdev, C_SYNC_TARGET);
3191
3192 put_ldev(mdev);
3193 } else
3194 dev_err(DEV, "Ignoring SyncUUID packet!\n");
3195
3196 return TRUE;
3197 }
3198
3199 enum receive_bitmap_ret { OK, DONE, FAILED };
3200
3201 static enum receive_bitmap_ret
3202 receive_bitmap_plain(struct drbd_conf *mdev, struct p_header *h,
3203 unsigned long *buffer, struct bm_xfer_ctx *c)
3204 {
3205 unsigned num_words = min_t(size_t, BM_PACKET_WORDS, c->bm_words - c->word_offset);
3206 unsigned want = num_words * sizeof(long);
3207
3208 if (want != h->length) {
3209 dev_err(DEV, "%s:want (%u) != h->length (%u)\n", __func__, want, h->length);
3210 return FAILED;
3211 }
3212 if (want == 0)
3213 return DONE;
3214 if (drbd_recv(mdev, buffer, want) != want)
3215 return FAILED;
3216
3217 drbd_bm_merge_lel(mdev, c->word_offset, num_words, buffer);
3218
3219 c->word_offset += num_words;
3220 c->bit_offset = c->word_offset * BITS_PER_LONG;
3221 if (c->bit_offset > c->bm_bits)
3222 c->bit_offset = c->bm_bits;
3223
3224 return OK;
3225 }
3226
3227 static enum receive_bitmap_ret
3228 recv_bm_rle_bits(struct drbd_conf *mdev,
3229 struct p_compressed_bm *p,
3230 struct bm_xfer_ctx *c)
3231 {
3232 struct bitstream bs;
3233 u64 look_ahead;
3234 u64 rl;
3235 u64 tmp;
3236 unsigned long s = c->bit_offset;
3237 unsigned long e;
3238 int len = p->head.length - (sizeof(*p) - sizeof(p->head));
3239 int toggle = DCBP_get_start(p);
3240 int have;
3241 int bits;
3242
3243 bitstream_init(&bs, p->code, len, DCBP_get_pad_bits(p));
3244
3245 bits = bitstream_get_bits(&bs, &look_ahead, 64);
3246 if (bits < 0)
3247 return FAILED;
3248
3249 for (have = bits; have > 0; s += rl, toggle = !toggle) {
3250 bits = vli_decode_bits(&rl, look_ahead);
3251 if (bits <= 0)
3252 return FAILED;
3253
3254 if (toggle) {
3255 e = s + rl -1;
3256 if (e >= c->bm_bits) {
3257 dev_err(DEV, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
3258 return FAILED;
3259 }
3260 _drbd_bm_set_bits(mdev, s, e);
3261 }
3262
3263 if (have < bits) {
3264 dev_err(DEV, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
3265 have, bits, look_ahead,
3266 (unsigned int)(bs.cur.b - p->code),
3267 (unsigned int)bs.buf_len);
3268 return FAILED;
3269 }
3270 look_ahead >>= bits;
3271 have -= bits;
3272
3273 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
3274 if (bits < 0)
3275 return FAILED;
3276 look_ahead |= tmp << have;
3277 have += bits;
3278 }
3279
3280 c->bit_offset = s;
3281 bm_xfer_ctx_bit_to_word_offset(c);
3282
3283 return (s == c->bm_bits) ? DONE : OK;
3284 }
3285
3286 static enum receive_bitmap_ret
3287 decode_bitmap_c(struct drbd_conf *mdev,
3288 struct p_compressed_bm *p,
3289 struct bm_xfer_ctx *c)
3290 {
3291 if (DCBP_get_code(p) == RLE_VLI_Bits)
3292 return recv_bm_rle_bits(mdev, p, c);
3293
3294 /* other variants had been implemented for evaluation,
3295 * but have been dropped as this one turned out to be "best"
3296 * during all our tests. */
3297
3298 dev_err(DEV, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
3299 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3300 return FAILED;
3301 }
3302
3303 void INFO_bm_xfer_stats(struct drbd_conf *mdev,
3304 const char *direction, struct bm_xfer_ctx *c)
3305 {
3306 /* what would it take to transfer it "plaintext" */
3307 unsigned plain = sizeof(struct p_header) *
3308 ((c->bm_words+BM_PACKET_WORDS-1)/BM_PACKET_WORDS+1)
3309 + c->bm_words * sizeof(long);
3310 unsigned total = c->bytes[0] + c->bytes[1];
3311 unsigned r;
3312
3313 /* total can not be zero. but just in case: */
3314 if (total == 0)
3315 return;
3316
3317 /* don't report if not compressed */
3318 if (total >= plain)
3319 return;
3320
3321 /* total < plain. check for overflow, still */
3322 r = (total > UINT_MAX/1000) ? (total / (plain/1000))
3323 : (1000 * total / plain);
3324
3325 if (r > 1000)
3326 r = 1000;
3327
3328 r = 1000 - r;
3329 dev_info(DEV, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
3330 "total %u; compression: %u.%u%%\n",
3331 direction,
3332 c->bytes[1], c->packets[1],
3333 c->bytes[0], c->packets[0],
3334 total, r/10, r % 10);
3335 }
3336
3337 /* Since we are processing the bitfield from lower addresses to higher,
3338 it does not matter if the process it in 32 bit chunks or 64 bit
3339 chunks as long as it is little endian. (Understand it as byte stream,
3340 beginning with the lowest byte...) If we would use big endian
3341 we would need to process it from the highest address to the lowest,
3342 in order to be agnostic to the 32 vs 64 bits issue.
3343
3344 returns 0 on failure, 1 if we successfully received it. */
3345 static int receive_bitmap(struct drbd_conf *mdev, struct p_header *h)
3346 {
3347 struct bm_xfer_ctx c;
3348 void *buffer;
3349 enum receive_bitmap_ret ret;
3350 int ok = FALSE;
3351
3352 wait_event(mdev->misc_wait, !atomic_read(&mdev->ap_bio_cnt));
3353
3354 drbd_bm_lock(mdev, "receive bitmap");
3355
3356 /* maybe we should use some per thread scratch page,
3357 * and allocate that during initial device creation? */
3358 buffer = (unsigned long *) __get_free_page(GFP_NOIO);
3359 if (!buffer) {
3360 dev_err(DEV, "failed to allocate one page buffer in %s\n", __func__);
3361 goto out;
3362 }
3363
3364 c = (struct bm_xfer_ctx) {
3365 .bm_bits = drbd_bm_bits(mdev),
3366 .bm_words = drbd_bm_words(mdev),
3367 };
3368
3369 do {
3370 if (h->command == P_BITMAP) {
3371 ret = receive_bitmap_plain(mdev, h, buffer, &c);
3372 } else if (h->command == P_COMPRESSED_BITMAP) {
3373 /* MAYBE: sanity check that we speak proto >= 90,
3374 * and the feature is enabled! */
3375 struct p_compressed_bm *p;
3376
3377 if (h->length > BM_PACKET_PAYLOAD_BYTES) {
3378 dev_err(DEV, "ReportCBitmap packet too large\n");
3379 goto out;
3380 }
3381 /* use the page buff */
3382 p = buffer;
3383 memcpy(p, h, sizeof(*h));
3384 if (drbd_recv(mdev, p->head.payload, h->length) != h->length)
3385 goto out;
3386 if (p->head.length <= (sizeof(*p) - sizeof(p->head))) {
3387 dev_err(DEV, "ReportCBitmap packet too small (l:%u)\n", p->head.length);
3388 return FAILED;
3389 }
3390 ret = decode_bitmap_c(mdev, p, &c);
3391 } else {
3392 dev_warn(DEV, "receive_bitmap: h->command neither ReportBitMap nor ReportCBitMap (is 0x%x)", h->command);
3393 goto out;
3394 }
3395
3396 c.packets[h->command == P_BITMAP]++;
3397 c.bytes[h->command == P_BITMAP] += sizeof(struct p_header) + h->length;
3398
3399 if (ret != OK)
3400 break;
3401
3402 if (!drbd_recv_header(mdev, h))
3403 goto out;
3404 } while (ret == OK);
3405 if (ret == FAILED)
3406 goto out;
3407
3408 INFO_bm_xfer_stats(mdev, "receive", &c);
3409
3410 if (mdev->state.conn == C_WF_BITMAP_T) {
3411 ok = !drbd_send_bitmap(mdev);
3412 if (!ok)
3413 goto out;
3414 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
3415 ok = _drbd_request_state(mdev, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
3416 D_ASSERT(ok == SS_SUCCESS);
3417 } else if (mdev->state.conn != C_WF_BITMAP_S) {
3418 /* admin may have requested C_DISCONNECTING,
3419 * other threads may have noticed network errors */
3420 dev_info(DEV, "unexpected cstate (%s) in receive_bitmap\n",
3421 drbd_conn_str(mdev->state.conn));
3422 }
3423
3424 ok = TRUE;
3425 out:
3426 drbd_bm_unlock(mdev);
3427 if (ok && mdev->state.conn == C_WF_BITMAP_S)
3428 drbd_start_resync(mdev, C_SYNC_SOURCE);
3429 free_page((unsigned long) buffer);
3430 return ok;
3431 }
3432
3433 static int receive_skip(struct drbd_conf *mdev, struct p_header *h)
3434 {
3435 /* TODO zero copy sink :) */
3436 static char sink[128];
3437 int size, want, r;
3438
3439 dev_warn(DEV, "skipping unknown optional packet type %d, l: %d!\n",
3440 h->command, h->length);
3441
3442 size = h->length;
3443 while (size > 0) {
3444 want = min_t(int, size, sizeof(sink));
3445 r = drbd_recv(mdev, sink, want);
3446 ERR_IF(r <= 0) break;
3447 size -= r;
3448 }
3449 return size == 0;
3450 }
3451
3452 static int receive_UnplugRemote(struct drbd_conf *mdev, struct p_header *h)
3453 {
3454 if (mdev->state.disk >= D_INCONSISTENT)
3455 drbd_kick_lo(mdev);
3456
3457 /* Make sure we've acked all the TCP data associated
3458 * with the data requests being unplugged */
3459 drbd_tcp_quickack(mdev->data.socket);
3460
3461 return TRUE;
3462 }
3463
3464 typedef int (*drbd_cmd_handler_f)(struct drbd_conf *, struct p_header *);
3465
3466 static drbd_cmd_handler_f drbd_default_handler[] = {
3467 [P_DATA] = receive_Data,
3468 [P_DATA_REPLY] = receive_DataReply,
3469 [P_RS_DATA_REPLY] = receive_RSDataReply,
3470 [P_BARRIER] = receive_Barrier,
3471 [P_BITMAP] = receive_bitmap,
3472 [P_COMPRESSED_BITMAP] = receive_bitmap,
3473 [P_UNPLUG_REMOTE] = receive_UnplugRemote,
3474 [P_DATA_REQUEST] = receive_DataRequest,
3475 [P_RS_DATA_REQUEST] = receive_DataRequest,
3476 [P_SYNC_PARAM] = receive_SyncParam,
3477 [P_SYNC_PARAM89] = receive_SyncParam,
3478 [P_PROTOCOL] = receive_protocol,
3479 [P_UUIDS] = receive_uuids,
3480 [P_SIZES] = receive_sizes,
3481 [P_STATE] = receive_state,
3482 [P_STATE_CHG_REQ] = receive_req_state,
3483 [P_SYNC_UUID] = receive_sync_uuid,
3484 [P_OV_REQUEST] = receive_DataRequest,
3485 [P_OV_REPLY] = receive_DataRequest,
3486 [P_CSUM_RS_REQUEST] = receive_DataRequest,
3487 /* anything missing from this table is in
3488 * the asender_tbl, see get_asender_cmd */
3489 [P_MAX_CMD] = NULL,
3490 };
3491
3492 static drbd_cmd_handler_f *drbd_cmd_handler = drbd_default_handler;
3493 static drbd_cmd_handler_f *drbd_opt_cmd_handler;
3494
3495 static void drbdd(struct drbd_conf *mdev)
3496 {
3497 drbd_cmd_handler_f handler;
3498 struct p_header *header = &mdev->data.rbuf.header;
3499
3500 while (get_t_state(&mdev->receiver) == Running) {
3501 drbd_thread_current_set_cpu(mdev);
3502 if (!drbd_recv_header(mdev, header))
3503 break;
3504
3505 if (header->command < P_MAX_CMD)
3506 handler = drbd_cmd_handler[header->command];
3507 else if (P_MAY_IGNORE < header->command
3508 && header->command < P_MAX_OPT_CMD)
3509 handler = drbd_opt_cmd_handler[header->command-P_MAY_IGNORE];
3510 else if (header->command > P_MAX_OPT_CMD)
3511 handler = receive_skip;
3512 else
3513 handler = NULL;
3514
3515 if (unlikely(!handler)) {
3516 dev_err(DEV, "unknown packet type %d, l: %d!\n",
3517 header->command, header->length);
3518 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3519 break;
3520 }
3521 if (unlikely(!handler(mdev, header))) {
3522 dev_err(DEV, "error receiving %s, l: %d!\n",
3523 cmdname(header->command), header->length);
3524 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
3525 break;
3526 }
3527 }
3528 }
3529
3530 static void drbd_fail_pending_reads(struct drbd_conf *mdev)
3531 {
3532 struct hlist_head *slot;
3533 struct hlist_node *pos;
3534 struct hlist_node *tmp;
3535 struct drbd_request *req;
3536 int i;
3537
3538 /*
3539 * Application READ requests
3540 */
3541 spin_lock_irq(&mdev->req_lock);
3542 for (i = 0; i < APP_R_HSIZE; i++) {
3543 slot = mdev->app_reads_hash+i;
3544 hlist_for_each_entry_safe(req, pos, tmp, slot, colision) {
3545 /* it may (but should not any longer!)
3546 * be on the work queue; if that assert triggers,
3547 * we need to also grab the
3548 * spin_lock_irq(&mdev->data.work.q_lock);
3549 * and list_del_init here. */
3550 D_ASSERT(list_empty(&req->w.list));
3551 /* It would be nice to complete outside of spinlock.
3552 * But this is easier for now. */
3553 _req_mod(req, connection_lost_while_pending);
3554 }
3555 }
3556 for (i = 0; i < APP_R_HSIZE; i++)
3557 if (!hlist_empty(mdev->app_reads_hash+i))
3558 dev_warn(DEV, "ASSERT FAILED: app_reads_hash[%d].first: "
3559 "%p, should be NULL\n", i, mdev->app_reads_hash[i].first);
3560
3561 memset(mdev->app_reads_hash, 0, APP_R_HSIZE*sizeof(void *));
3562 spin_unlock_irq(&mdev->req_lock);
3563 }
3564
3565 void drbd_flush_workqueue(struct drbd_conf *mdev)
3566 {
3567 struct drbd_wq_barrier barr;
3568
3569 barr.w.cb = w_prev_work_done;
3570 init_completion(&barr.done);
3571 drbd_queue_work(&mdev->data.work, &barr.w);
3572 wait_for_completion(&barr.done);
3573 }
3574
3575 static void drbd_disconnect(struct drbd_conf *mdev)
3576 {
3577 enum drbd_fencing_p fp;
3578 union drbd_state os, ns;
3579 int rv = SS_UNKNOWN_ERROR;
3580 unsigned int i;
3581
3582 if (mdev->state.conn == C_STANDALONE)
3583 return;
3584 if (mdev->state.conn >= C_WF_CONNECTION)
3585 dev_err(DEV, "ASSERT FAILED cstate = %s, expected < WFConnection\n",
3586 drbd_conn_str(mdev->state.conn));
3587
3588 /* asender does not clean up anything. it must not interfere, either */
3589 drbd_thread_stop(&mdev->asender);
3590
3591 mutex_lock(&mdev->data.mutex);
3592 drbd_free_sock(mdev);
3593 mutex_unlock(&mdev->data.mutex);
3594
3595 spin_lock_irq(&mdev->req_lock);
3596 _drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
3597 _drbd_wait_ee_list_empty(mdev, &mdev->sync_ee);
3598 _drbd_wait_ee_list_empty(mdev, &mdev->read_ee);
3599 spin_unlock_irq(&mdev->req_lock);
3600
3601 /* We do not have data structures that would allow us to
3602 * get the rs_pending_cnt down to 0 again.
3603 * * On C_SYNC_TARGET we do not have any data structures describing
3604 * the pending RSDataRequest's we have sent.
3605 * * On C_SYNC_SOURCE there is no data structure that tracks
3606 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
3607 * And no, it is not the sum of the reference counts in the
3608 * resync_LRU. The resync_LRU tracks the whole operation including
3609 * the disk-IO, while the rs_pending_cnt only tracks the blocks
3610 * on the fly. */
3611 drbd_rs_cancel_all(mdev);
3612 mdev->rs_total = 0;
3613 mdev->rs_failed = 0;
3614 atomic_set(&mdev->rs_pending_cnt, 0);
3615 wake_up(&mdev->misc_wait);
3616
3617 /* make sure syncer is stopped and w_resume_next_sg queued */
3618 del_timer_sync(&mdev->resync_timer);
3619 set_bit(STOP_SYNC_TIMER, &mdev->flags);
3620 resync_timer_fn((unsigned long)mdev);
3621
3622 /* so we can be sure that all remote or resync reads
3623 * made it at least to net_ee */
3624 wait_event(mdev->misc_wait, !atomic_read(&mdev->local_cnt));
3625
3626 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
3627 * w_make_resync_request etc. which may still be on the worker queue
3628 * to be "canceled" */
3629 drbd_flush_workqueue(mdev);
3630
3631 /* This also does reclaim_net_ee(). If we do this too early, we might
3632 * miss some resync ee and pages.*/
3633 drbd_process_done_ee(mdev);
3634
3635 kfree(mdev->p_uuid);
3636 mdev->p_uuid = NULL;
3637
3638 if (!mdev->state.susp)
3639 tl_clear(mdev);
3640
3641 drbd_fail_pending_reads(mdev);
3642
3643 dev_info(DEV, "Connection closed\n");
3644
3645 drbd_md_sync(mdev);
3646
3647 fp = FP_DONT_CARE;
3648 if (get_ldev(mdev)) {
3649 fp = mdev->ldev->dc.fencing;
3650 put_ldev(mdev);
3651 }
3652
3653 if (mdev->state.role == R_PRIMARY) {
3654 if (fp >= FP_RESOURCE && mdev->state.pdsk >= D_UNKNOWN) {
3655 enum drbd_disk_state nps = drbd_try_outdate_peer(mdev);
3656 drbd_request_state(mdev, NS(pdsk, nps));
3657 }
3658 }
3659
3660 spin_lock_irq(&mdev->req_lock);
3661 os = mdev->state;
3662 if (os.conn >= C_UNCONNECTED) {
3663 /* Do not restart in case we are C_DISCONNECTING */
3664 ns = os;
3665 ns.conn = C_UNCONNECTED;
3666 rv = _drbd_set_state(mdev, ns, CS_VERBOSE, NULL);
3667 }
3668 spin_unlock_irq(&mdev->req_lock);
3669
3670 if (os.conn == C_DISCONNECTING) {
3671 struct hlist_head *h;
3672 wait_event(mdev->misc_wait, atomic_read(&mdev->net_cnt) == 0);
3673
3674 /* we must not free the tl_hash
3675 * while application io is still on the fly */
3676 wait_event(mdev->misc_wait, atomic_read(&mdev->ap_bio_cnt) == 0);
3677
3678 spin_lock_irq(&mdev->req_lock);
3679 /* paranoia code */
3680 for (h = mdev->ee_hash; h < mdev->ee_hash + mdev->ee_hash_s; h++)
3681 if (h->first)
3682 dev_err(DEV, "ASSERT FAILED ee_hash[%u].first == %p, expected NULL\n",
3683 (int)(h - mdev->ee_hash), h->first);
3684 kfree(mdev->ee_hash);
3685 mdev->ee_hash = NULL;
3686 mdev->ee_hash_s = 0;
3687
3688 /* paranoia code */
3689 for (h = mdev->tl_hash; h < mdev->tl_hash + mdev->tl_hash_s; h++)
3690 if (h->first)
3691 dev_err(DEV, "ASSERT FAILED tl_hash[%u] == %p, expected NULL\n",
3692 (int)(h - mdev->tl_hash), h->first);
3693 kfree(mdev->tl_hash);
3694 mdev->tl_hash = NULL;
3695 mdev->tl_hash_s = 0;
3696 spin_unlock_irq(&mdev->req_lock);
3697
3698 crypto_free_hash(mdev->cram_hmac_tfm);
3699 mdev->cram_hmac_tfm = NULL;
3700
3701 kfree(mdev->net_conf);
3702 mdev->net_conf = NULL;
3703 drbd_request_state(mdev, NS(conn, C_STANDALONE));
3704 }
3705
3706 /* tcp_close and release of sendpage pages can be deferred. I don't
3707 * want to use SO_LINGER, because apparently it can be deferred for
3708 * more than 20 seconds (longest time I checked).
3709 *
3710 * Actually we don't care for exactly when the network stack does its
3711 * put_page(), but release our reference on these pages right here.
3712 */
3713 i = drbd_release_ee(mdev, &mdev->net_ee);
3714 if (i)
3715 dev_info(DEV, "net_ee not empty, killed %u entries\n", i);
3716 i = atomic_read(&mdev->pp_in_use);
3717 if (i)
3718 dev_info(DEV, "pp_in_use = %u, expected 0\n", i);
3719
3720 D_ASSERT(list_empty(&mdev->read_ee));
3721 D_ASSERT(list_empty(&mdev->active_ee));
3722 D_ASSERT(list_empty(&mdev->sync_ee));
3723 D_ASSERT(list_empty(&mdev->done_ee));
3724
3725 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
3726 atomic_set(&mdev->current_epoch->epoch_size, 0);
3727 D_ASSERT(list_empty(&mdev->current_epoch->list));
3728 }
3729
3730 /*
3731 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
3732 * we can agree on is stored in agreed_pro_version.
3733 *
3734 * feature flags and the reserved array should be enough room for future
3735 * enhancements of the handshake protocol, and possible plugins...
3736 *
3737 * for now, they are expected to be zero, but ignored.
3738 */
3739 static int drbd_send_handshake(struct drbd_conf *mdev)
3740 {
3741 /* ASSERT current == mdev->receiver ... */
3742 struct p_handshake *p = &mdev->data.sbuf.handshake;
3743 int ok;
3744
3745 if (mutex_lock_interruptible(&mdev->data.mutex)) {
3746 dev_err(DEV, "interrupted during initial handshake\n");
3747 return 0; /* interrupted. not ok. */
3748 }
3749
3750 if (mdev->data.socket == NULL) {
3751 mutex_unlock(&mdev->data.mutex);
3752 return 0;
3753 }
3754
3755 memset(p, 0, sizeof(*p));
3756 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
3757 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
3758 ok = _drbd_send_cmd( mdev, mdev->data.socket, P_HAND_SHAKE,
3759 (struct p_header *)p, sizeof(*p), 0 );
3760 mutex_unlock(&mdev->data.mutex);
3761 return ok;
3762 }
3763
3764 /*
3765 * return values:
3766 * 1 yes, we have a valid connection
3767 * 0 oops, did not work out, please try again
3768 * -1 peer talks different language,
3769 * no point in trying again, please go standalone.
3770 */
3771 static int drbd_do_handshake(struct drbd_conf *mdev)
3772 {
3773 /* ASSERT current == mdev->receiver ... */
3774 struct p_handshake *p = &mdev->data.rbuf.handshake;
3775 const int expect = sizeof(struct p_handshake)
3776 -sizeof(struct p_header);
3777 int rv;
3778
3779 rv = drbd_send_handshake(mdev);
3780 if (!rv)
3781 return 0;
3782
3783 rv = drbd_recv_header(mdev, &p->head);
3784 if (!rv)
3785 return 0;
3786
3787 if (p->head.command != P_HAND_SHAKE) {
3788 dev_err(DEV, "expected HandShake packet, received: %s (0x%04x)\n",
3789 cmdname(p->head.command), p->head.command);
3790 return -1;
3791 }
3792
3793 if (p->head.length != expect) {
3794 dev_err(DEV, "expected HandShake length: %u, received: %u\n",
3795 expect, p->head.length);
3796 return -1;
3797 }
3798
3799 rv = drbd_recv(mdev, &p->head.payload, expect);
3800
3801 if (rv != expect) {
3802 dev_err(DEV, "short read receiving handshake packet: l=%u\n", rv);
3803 return 0;
3804 }
3805
3806 p->protocol_min = be32_to_cpu(p->protocol_min);
3807 p->protocol_max = be32_to_cpu(p->protocol_max);
3808 if (p->protocol_max == 0)
3809 p->protocol_max = p->protocol_min;
3810
3811 if (PRO_VERSION_MAX < p->protocol_min ||
3812 PRO_VERSION_MIN > p->protocol_max)
3813 goto incompat;
3814
3815 mdev->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
3816
3817 dev_info(DEV, "Handshake successful: "
3818 "Agreed network protocol version %d\n", mdev->agreed_pro_version);
3819
3820 return 1;
3821
3822 incompat:
3823 dev_err(DEV, "incompatible DRBD dialects: "
3824 "I support %d-%d, peer supports %d-%d\n",
3825 PRO_VERSION_MIN, PRO_VERSION_MAX,
3826 p->protocol_min, p->protocol_max);
3827 return -1;
3828 }
3829
3830 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
3831 static int drbd_do_auth(struct drbd_conf *mdev)
3832 {
3833 dev_err(DEV, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
3834 dev_err(DEV, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
3835 return 0;
3836 }
3837 #else
3838 #define CHALLENGE_LEN 64
3839 static int drbd_do_auth(struct drbd_conf *mdev)
3840 {
3841 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */
3842 struct scatterlist sg;
3843 char *response = NULL;
3844 char *right_response = NULL;
3845 char *peers_ch = NULL;
3846 struct p_header p;
3847 unsigned int key_len = strlen(mdev->net_conf->shared_secret);
3848 unsigned int resp_size;
3849 struct hash_desc desc;
3850 int rv;
3851
3852 desc.tfm = mdev->cram_hmac_tfm;
3853 desc.flags = 0;
3854
3855 rv = crypto_hash_setkey(mdev->cram_hmac_tfm,
3856 (u8 *)mdev->net_conf->shared_secret, key_len);
3857 if (rv) {
3858 dev_err(DEV, "crypto_hash_setkey() failed with %d\n", rv);
3859 rv = 0;
3860 goto fail;
3861 }
3862
3863 get_random_bytes(my_challenge, CHALLENGE_LEN);
3864
3865 rv = drbd_send_cmd2(mdev, P_AUTH_CHALLENGE, my_challenge, CHALLENGE_LEN);
3866 if (!rv)
3867 goto fail;
3868
3869 rv = drbd_recv_header(mdev, &p);
3870 if (!rv)
3871 goto fail;
3872
3873 if (p.command != P_AUTH_CHALLENGE) {
3874 dev_err(DEV, "expected AuthChallenge packet, received: %s (0x%04x)\n",
3875 cmdname(p.command), p.command);
3876 rv = 0;
3877 goto fail;
3878 }
3879
3880 if (p.length > CHALLENGE_LEN*2) {
3881 dev_err(DEV, "expected AuthChallenge payload too big.\n");
3882 rv = 0;
3883 goto fail;
3884 }
3885
3886 peers_ch = kmalloc(p.length, GFP_NOIO);
3887 if (peers_ch == NULL) {
3888 dev_err(DEV, "kmalloc of peers_ch failed\n");
3889 rv = 0;
3890 goto fail;
3891 }
3892
3893 rv = drbd_recv(mdev, peers_ch, p.length);
3894
3895 if (rv != p.length) {
3896 dev_err(DEV, "short read AuthChallenge: l=%u\n", rv);
3897 rv = 0;
3898 goto fail;
3899 }
3900
3901 resp_size = crypto_hash_digestsize(mdev->cram_hmac_tfm);
3902 response = kmalloc(resp_size, GFP_NOIO);
3903 if (response == NULL) {
3904 dev_err(DEV, "kmalloc of response failed\n");
3905 rv = 0;
3906 goto fail;
3907 }
3908
3909 sg_init_table(&sg, 1);
3910 sg_set_buf(&sg, peers_ch, p.length);
3911
3912 rv = crypto_hash_digest(&desc, &sg, sg.length, response);
3913 if (rv) {
3914 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv);
3915 rv = 0;
3916 goto fail;
3917 }
3918
3919 rv = drbd_send_cmd2(mdev, P_AUTH_RESPONSE, response, resp_size);
3920 if (!rv)
3921 goto fail;
3922
3923 rv = drbd_recv_header(mdev, &p);
3924 if (!rv)
3925 goto fail;
3926
3927 if (p.command != P_AUTH_RESPONSE) {
3928 dev_err(DEV, "expected AuthResponse packet, received: %s (0x%04x)\n",
3929 cmdname(p.command), p.command);
3930 rv = 0;
3931 goto fail;
3932 }
3933
3934 if (p.length != resp_size) {
3935 dev_err(DEV, "expected AuthResponse payload of wrong size\n");
3936 rv = 0;
3937 goto fail;
3938 }
3939
3940 rv = drbd_recv(mdev, response , resp_size);
3941
3942 if (rv != resp_size) {
3943 dev_err(DEV, "short read receiving AuthResponse: l=%u\n", rv);
3944 rv = 0;
3945 goto fail;
3946 }
3947
3948 right_response = kmalloc(resp_size, GFP_NOIO);
3949 if (response == NULL) {
3950 dev_err(DEV, "kmalloc of right_response failed\n");
3951 rv = 0;
3952 goto fail;
3953 }
3954
3955 sg_set_buf(&sg, my_challenge, CHALLENGE_LEN);
3956
3957 rv = crypto_hash_digest(&desc, &sg, sg.length, right_response);
3958 if (rv) {
3959 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv);
3960 rv = 0;
3961 goto fail;
3962 }
3963
3964 rv = !memcmp(response, right_response, resp_size);
3965
3966 if (rv)
3967 dev_info(DEV, "Peer authenticated using %d bytes of '%s' HMAC\n",
3968 resp_size, mdev->net_conf->cram_hmac_alg);
3969
3970 fail:
3971 kfree(peers_ch);
3972 kfree(response);
3973 kfree(right_response);
3974
3975 return rv;
3976 }
3977 #endif
3978
3979 int drbdd_init(struct drbd_thread *thi)
3980 {
3981 struct drbd_conf *mdev = thi->mdev;
3982 unsigned int minor = mdev_to_minor(mdev);
3983 int h;
3984
3985 sprintf(current->comm, "drbd%d_receiver", minor);
3986
3987 dev_info(DEV, "receiver (re)started\n");
3988
3989 do {
3990 h = drbd_connect(mdev);
3991 if (h == 0) {
3992 drbd_disconnect(mdev);
3993 __set_current_state(TASK_INTERRUPTIBLE);
3994 schedule_timeout(HZ);
3995 }
3996 if (h == -1) {
3997 dev_warn(DEV, "Discarding network configuration.\n");
3998 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
3999 }
4000 } while (h == 0);
4001
4002 if (h > 0) {
4003 if (get_net_conf(mdev)) {
4004 drbdd(mdev);
4005 put_net_conf(mdev);
4006 }
4007 }
4008
4009 drbd_disconnect(mdev);
4010
4011 dev_info(DEV, "receiver terminated\n");
4012 return 0;
4013 }
4014
4015 /* ********* acknowledge sender ******** */
4016
4017 static int got_RqSReply(struct drbd_conf *mdev, struct p_header *h)
4018 {
4019 struct p_req_state_reply *p = (struct p_req_state_reply *)h;
4020
4021 int retcode = be32_to_cpu(p->retcode);
4022
4023 if (retcode >= SS_SUCCESS) {
4024 set_bit(CL_ST_CHG_SUCCESS, &mdev->flags);
4025 } else {
4026 set_bit(CL_ST_CHG_FAIL, &mdev->flags);
4027 dev_err(DEV, "Requested state change failed by peer: %s (%d)\n",
4028 drbd_set_st_err_str(retcode), retcode);
4029 }
4030 wake_up(&mdev->state_wait);
4031
4032 return TRUE;
4033 }
4034
4035 static int got_Ping(struct drbd_conf *mdev, struct p_header *h)
4036 {
4037 return drbd_send_ping_ack(mdev);
4038
4039 }
4040
4041 static int got_PingAck(struct drbd_conf *mdev, struct p_header *h)
4042 {
4043 /* restore idle timeout */
4044 mdev->meta.socket->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ;
4045
4046 return TRUE;
4047 }
4048
4049 static int got_IsInSync(struct drbd_conf *mdev, struct p_header *h)
4050 {
4051 struct p_block_ack *p = (struct p_block_ack *)h;
4052 sector_t sector = be64_to_cpu(p->sector);
4053 int blksize = be32_to_cpu(p->blksize);
4054
4055 D_ASSERT(mdev->agreed_pro_version >= 89);
4056
4057 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4058
4059 drbd_rs_complete_io(mdev, sector);
4060 drbd_set_in_sync(mdev, sector, blksize);
4061 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
4062 mdev->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
4063 dec_rs_pending(mdev);
4064
4065 return TRUE;
4066 }
4067
4068 /* when we receive the ACK for a write request,
4069 * verify that we actually know about it */
4070 static struct drbd_request *_ack_id_to_req(struct drbd_conf *mdev,
4071 u64 id, sector_t sector)
4072 {
4073 struct hlist_head *slot = tl_hash_slot(mdev, sector);
4074 struct hlist_node *n;
4075 struct drbd_request *req;
4076
4077 hlist_for_each_entry(req, n, slot, colision) {
4078 if ((unsigned long)req == (unsigned long)id) {
4079 if (req->sector != sector) {
4080 dev_err(DEV, "_ack_id_to_req: found req %p but it has "
4081 "wrong sector (%llus versus %llus)\n", req,
4082 (unsigned long long)req->sector,
4083 (unsigned long long)sector);
4084 break;
4085 }
4086 return req;
4087 }
4088 }
4089 dev_err(DEV, "_ack_id_to_req: failed to find req %p, sector %llus in list\n",
4090 (void *)(unsigned long)id, (unsigned long long)sector);
4091 return NULL;
4092 }
4093
4094 typedef struct drbd_request *(req_validator_fn)
4095 (struct drbd_conf *mdev, u64 id, sector_t sector);
4096
4097 static int validate_req_change_req_state(struct drbd_conf *mdev,
4098 u64 id, sector_t sector, req_validator_fn validator,
4099 const char *func, enum drbd_req_event what)
4100 {
4101 struct drbd_request *req;
4102 struct bio_and_error m;
4103
4104 spin_lock_irq(&mdev->req_lock);
4105 req = validator(mdev, id, sector);
4106 if (unlikely(!req)) {
4107 spin_unlock_irq(&mdev->req_lock);
4108 dev_err(DEV, "%s: got a corrupt block_id/sector pair\n", func);
4109 return FALSE;
4110 }
4111 __req_mod(req, what, &m);
4112 spin_unlock_irq(&mdev->req_lock);
4113
4114 if (m.bio)
4115 complete_master_bio(mdev, &m);
4116 return TRUE;
4117 }
4118
4119 static int got_BlockAck(struct drbd_conf *mdev, struct p_header *h)
4120 {
4121 struct p_block_ack *p = (struct p_block_ack *)h;
4122 sector_t sector = be64_to_cpu(p->sector);
4123 int blksize = be32_to_cpu(p->blksize);
4124 enum drbd_req_event what;
4125
4126 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4127
4128 if (is_syncer_block_id(p->block_id)) {
4129 drbd_set_in_sync(mdev, sector, blksize);
4130 dec_rs_pending(mdev);
4131 return TRUE;
4132 }
4133 switch (be16_to_cpu(h->command)) {
4134 case P_RS_WRITE_ACK:
4135 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4136 what = write_acked_by_peer_and_sis;
4137 break;
4138 case P_WRITE_ACK:
4139 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4140 what = write_acked_by_peer;
4141 break;
4142 case P_RECV_ACK:
4143 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_B);
4144 what = recv_acked_by_peer;
4145 break;
4146 case P_DISCARD_ACK:
4147 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
4148 what = conflict_discarded_by_peer;
4149 break;
4150 default:
4151 D_ASSERT(0);
4152 return FALSE;
4153 }
4154
4155 return validate_req_change_req_state(mdev, p->block_id, sector,
4156 _ack_id_to_req, __func__ , what);
4157 }
4158
4159 static int got_NegAck(struct drbd_conf *mdev, struct p_header *h)
4160 {
4161 struct p_block_ack *p = (struct p_block_ack *)h;
4162 sector_t sector = be64_to_cpu(p->sector);
4163
4164 if (__ratelimit(&drbd_ratelimit_state))
4165 dev_warn(DEV, "Got NegAck packet. Peer is in troubles?\n");
4166
4167 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4168
4169 if (is_syncer_block_id(p->block_id)) {
4170 int size = be32_to_cpu(p->blksize);
4171 dec_rs_pending(mdev);
4172 drbd_rs_failed_io(mdev, sector, size);
4173 return TRUE;
4174 }
4175 return validate_req_change_req_state(mdev, p->block_id, sector,
4176 _ack_id_to_req, __func__ , neg_acked);
4177 }
4178
4179 static int got_NegDReply(struct drbd_conf *mdev, struct p_header *h)
4180 {
4181 struct p_block_ack *p = (struct p_block_ack *)h;
4182 sector_t sector = be64_to_cpu(p->sector);
4183
4184 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4185 dev_err(DEV, "Got NegDReply; Sector %llus, len %u; Fail original request.\n",
4186 (unsigned long long)sector, be32_to_cpu(p->blksize));
4187
4188 return validate_req_change_req_state(mdev, p->block_id, sector,
4189 _ar_id_to_req, __func__ , neg_acked);
4190 }
4191
4192 static int got_NegRSDReply(struct drbd_conf *mdev, struct p_header *h)
4193 {
4194 sector_t sector;
4195 int size;
4196 struct p_block_ack *p = (struct p_block_ack *)h;
4197
4198 sector = be64_to_cpu(p->sector);
4199 size = be32_to_cpu(p->blksize);
4200 D_ASSERT(p->block_id == ID_SYNCER);
4201
4202 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4203
4204 dec_rs_pending(mdev);
4205
4206 if (get_ldev_if_state(mdev, D_FAILED)) {
4207 drbd_rs_complete_io(mdev, sector);
4208 drbd_rs_failed_io(mdev, sector, size);
4209 put_ldev(mdev);
4210 }
4211
4212 return TRUE;
4213 }
4214
4215 static int got_BarrierAck(struct drbd_conf *mdev, struct p_header *h)
4216 {
4217 struct p_barrier_ack *p = (struct p_barrier_ack *)h;
4218
4219 tl_release(mdev, p->barrier, be32_to_cpu(p->set_size));
4220
4221 return TRUE;
4222 }
4223
4224 static int got_OVResult(struct drbd_conf *mdev, struct p_header *h)
4225 {
4226 struct p_block_ack *p = (struct p_block_ack *)h;
4227 struct drbd_work *w;
4228 sector_t sector;
4229 int size;
4230
4231 sector = be64_to_cpu(p->sector);
4232 size = be32_to_cpu(p->blksize);
4233
4234 update_peer_seq(mdev, be32_to_cpu(p->seq_num));
4235
4236 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
4237 drbd_ov_oos_found(mdev, sector, size);
4238 else
4239 ov_oos_print(mdev);
4240
4241 drbd_rs_complete_io(mdev, sector);
4242 dec_rs_pending(mdev);
4243
4244 if (--mdev->ov_left == 0) {
4245 w = kmalloc(sizeof(*w), GFP_NOIO);
4246 if (w) {
4247 w->cb = w_ov_finished;
4248 drbd_queue_work_front(&mdev->data.work, w);
4249 } else {
4250 dev_err(DEV, "kmalloc(w) failed.");
4251 ov_oos_print(mdev);
4252 drbd_resync_finished(mdev);
4253 }
4254 }
4255 return TRUE;
4256 }
4257
4258 struct asender_cmd {
4259 size_t pkt_size;
4260 int (*process)(struct drbd_conf *mdev, struct p_header *h);
4261 };
4262
4263 static struct asender_cmd *get_asender_cmd(int cmd)
4264 {
4265 static struct asender_cmd asender_tbl[] = {
4266 /* anything missing from this table is in
4267 * the drbd_cmd_handler (drbd_default_handler) table,
4268 * see the beginning of drbdd() */
4269 [P_PING] = { sizeof(struct p_header), got_Ping },
4270 [P_PING_ACK] = { sizeof(struct p_header), got_PingAck },
4271 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4272 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4273 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4274 [P_DISCARD_ACK] = { sizeof(struct p_block_ack), got_BlockAck },
4275 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck },
4276 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply },
4277 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply},
4278 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult },
4279 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck },
4280 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
4281 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync },
4282 [P_MAX_CMD] = { 0, NULL },
4283 };
4284 if (cmd > P_MAX_CMD || asender_tbl[cmd].process == NULL)
4285 return NULL;
4286 return &asender_tbl[cmd];
4287 }
4288
4289 int drbd_asender(struct drbd_thread *thi)
4290 {
4291 struct drbd_conf *mdev = thi->mdev;
4292 struct p_header *h = &mdev->meta.rbuf.header;
4293 struct asender_cmd *cmd = NULL;
4294
4295 int rv, len;
4296 void *buf = h;
4297 int received = 0;
4298 int expect = sizeof(struct p_header);
4299 int empty;
4300
4301 sprintf(current->comm, "drbd%d_asender", mdev_to_minor(mdev));
4302
4303 current->policy = SCHED_RR; /* Make this a realtime task! */
4304 current->rt_priority = 2; /* more important than all other tasks */
4305
4306 while (get_t_state(thi) == Running) {
4307 drbd_thread_current_set_cpu(mdev);
4308 if (test_and_clear_bit(SEND_PING, &mdev->flags)) {
4309 ERR_IF(!drbd_send_ping(mdev)) goto reconnect;
4310 mdev->meta.socket->sk->sk_rcvtimeo =
4311 mdev->net_conf->ping_timeo*HZ/10;
4312 }
4313
4314 /* conditionally cork;
4315 * it may hurt latency if we cork without much to send */
4316 if (!mdev->net_conf->no_cork &&
4317 3 < atomic_read(&mdev->unacked_cnt))
4318 drbd_tcp_cork(mdev->meta.socket);
4319 while (1) {
4320 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4321 flush_signals(current);
4322 if (!drbd_process_done_ee(mdev)) {
4323 dev_err(DEV, "process_done_ee() = NOT_OK\n");
4324 goto reconnect;
4325 }
4326 /* to avoid race with newly queued ACKs */
4327 set_bit(SIGNAL_ASENDER, &mdev->flags);
4328 spin_lock_irq(&mdev->req_lock);
4329 empty = list_empty(&mdev->done_ee);
4330 spin_unlock_irq(&mdev->req_lock);
4331 /* new ack may have been queued right here,
4332 * but then there is also a signal pending,
4333 * and we start over... */
4334 if (empty)
4335 break;
4336 }
4337 /* but unconditionally uncork unless disabled */
4338 if (!mdev->net_conf->no_cork)
4339 drbd_tcp_uncork(mdev->meta.socket);
4340
4341 /* short circuit, recv_msg would return EINTR anyways. */
4342 if (signal_pending(current))
4343 continue;
4344
4345 rv = drbd_recv_short(mdev, mdev->meta.socket,
4346 buf, expect-received, 0);
4347 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4348
4349 flush_signals(current);
4350
4351 /* Note:
4352 * -EINTR (on meta) we got a signal
4353 * -EAGAIN (on meta) rcvtimeo expired
4354 * -ECONNRESET other side closed the connection
4355 * -ERESTARTSYS (on data) we got a signal
4356 * rv < 0 other than above: unexpected error!
4357 * rv == expected: full header or command
4358 * rv < expected: "woken" by signal during receive
4359 * rv == 0 : "connection shut down by peer"
4360 */
4361 if (likely(rv > 0)) {
4362 received += rv;
4363 buf += rv;
4364 } else if (rv == 0) {
4365 dev_err(DEV, "meta connection shut down by peer.\n");
4366 goto reconnect;
4367 } else if (rv == -EAGAIN) {
4368 if (mdev->meta.socket->sk->sk_rcvtimeo ==
4369 mdev->net_conf->ping_timeo*HZ/10) {
4370 dev_err(DEV, "PingAck did not arrive in time.\n");
4371 goto reconnect;
4372 }
4373 set_bit(SEND_PING, &mdev->flags);
4374 continue;
4375 } else if (rv == -EINTR) {
4376 continue;
4377 } else {
4378 dev_err(DEV, "sock_recvmsg returned %d\n", rv);
4379 goto reconnect;
4380 }
4381
4382 if (received == expect && cmd == NULL) {
4383 if (unlikely(h->magic != BE_DRBD_MAGIC)) {
4384 dev_err(DEV, "magic?? on meta m: 0x%lx c: %d l: %d\n",
4385 (long)be32_to_cpu(h->magic),
4386 h->command, h->length);
4387 goto reconnect;
4388 }
4389 cmd = get_asender_cmd(be16_to_cpu(h->command));
4390 len = be16_to_cpu(h->length);
4391 if (unlikely(cmd == NULL)) {
4392 dev_err(DEV, "unknown command?? on meta m: 0x%lx c: %d l: %d\n",
4393 (long)be32_to_cpu(h->magic),
4394 h->command, h->length);
4395 goto disconnect;
4396 }
4397 expect = cmd->pkt_size;
4398 ERR_IF(len != expect-sizeof(struct p_header))
4399 goto reconnect;
4400 }
4401 if (received == expect) {
4402 D_ASSERT(cmd != NULL);
4403 if (!cmd->process(mdev, h))
4404 goto reconnect;
4405
4406 buf = h;
4407 received = 0;
4408 expect = sizeof(struct p_header);
4409 cmd = NULL;
4410 }
4411 }
4412
4413 if (0) {
4414 reconnect:
4415 drbd_force_state(mdev, NS(conn, C_NETWORK_FAILURE));
4416 }
4417 if (0) {
4418 disconnect:
4419 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
4420 }
4421 clear_bit(SIGNAL_ASENDER, &mdev->flags);
4422
4423 D_ASSERT(mdev->state.conn < C_CONNECTED);
4424 dev_info(DEV, "asender terminated\n");
4425
4426 return 0;
4427 }
mm of the moment repository