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