Merge git://git.kernel.org/pub/scm/linux/kernel/git/gregkh/staging-2.6
[sfrench/cifs-2.6.git] / drivers / block / drbd / drbd_receiver.c
1 /*
2    drbd_receiver.c
3
4    This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
5
6    Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7    Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8    Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
9
10    drbd is free software; you can redistribute it and/or modify
11    it under the terms of the GNU General Public License as published by
12    the Free Software Foundation; either version 2, or (at your option)
13    any later version.
14
15    drbd is distributed in the hope that it will be useful,
16    but WITHOUT ANY WARRANTY; without even the implied warranty of
17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
18    GNU General Public License for more details.
19
20    You should have received a copy of the GNU General Public License
21    along with drbd; see the file COPYING.  If not, write to
22    the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
23  */
24
25
26 #include <linux/module.h>
27
28 #include <asm/uaccess.h>
29 #include <net/sock.h>
30
31 #include <linux/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"
49 #include "drbd_req.h"
50
51 #include "drbd_vli.h"
52
53 struct flush_work {
54         struct drbd_work w;
55         struct drbd_epoch *epoch;
56 };
57
58 enum finish_epoch {
59         FE_STILL_LIVE,
60         FE_DESTROYED,
61         FE_RECYCLED,
62 };
63
64 static int drbd_do_handshake(struct drbd_conf *mdev);
65 static int drbd_do_auth(struct drbd_conf *mdev);
66
67 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *, struct drbd_epoch *, enum epoch_event);
68 static int e_end_block(struct drbd_conf *, struct drbd_work *, int);
69
70 static 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
83 static 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. */
112 static 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
118 static 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
136 static 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  */
159 static 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) */
198 static 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
221 static 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 /*
253 You need to hold the req_lock:
254  _drbd_wait_ee_list_empty()
255
256 You 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
267 struct 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
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
362 void drbd_free_ee(struct drbd_conf *mdev, struct drbd_epoch_entry *e)
363 {
364         struct bio *bio = e->private_bio;
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
371 int 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  */
398 static 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) {
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
427 void _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
443 void 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 */
452 static 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
478 out:
479         return err;
480 }
481
482 static 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
505 static 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
558 static 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
609 out:
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
634 static 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
667 out:
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
681 static 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
689 static 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  */
707 static 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  */
734 static 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
787 retry:
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)); */
881                 switch (drbd_do_auth(mdev)) {
882                 case -1:
883                         dev_err(DEV, "Authentication of peer failed\n");
884                         return -1;
885                 case 0:
886                         dev_err(DEV, "Authentication of peer failed, trying again.\n");
887                         return 0;
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
902         drbd_send_protocol(mdev);
903         drbd_send_sync_param(mdev, &mdev->sync_conf);
904         drbd_send_sizes(mdev, 0);
905         drbd_send_uuids(mdev);
906         drbd_send_state(mdev);
907         clear_bit(USE_DEGR_WFC_T, &mdev->flags);
908         clear_bit(RESIZE_PENDING, &mdev->flags);
909
910         return 1;
911
912 out_release_sockets:
913         if (sock)
914                 sock_release(sock);
915         if (msock)
916                 sock_release(msock);
917         return -1;
918 }
919
920 static int drbd_recv_header(struct drbd_conf *mdev, struct p_header *h)
921 {
922         int r;
923
924         r = drbd_recv(mdev, h, sizeof(*h));
925
926         if (unlikely(r != sizeof(*h))) {
927                 dev_err(DEV, "short read expecting header on sock: r=%d\n", r);
928                 return FALSE;
929         };
930         h->command = be16_to_cpu(h->command);
931         h->length  = be16_to_cpu(h->length);
932         if (unlikely(h->magic != BE_DRBD_MAGIC)) {
933                 dev_err(DEV, "magic?? on data m: 0x%lx c: %d l: %d\n",
934                     (long)be32_to_cpu(h->magic),
935                     h->command, h->length);
936                 return FALSE;
937         }
938         mdev->last_received = jiffies;
939
940         return TRUE;
941 }
942
943 static enum finish_epoch drbd_flush_after_epoch(struct drbd_conf *mdev, struct drbd_epoch *epoch)
944 {
945         int rv;
946
947         if (mdev->write_ordering >= WO_bdev_flush && get_ldev(mdev)) {
948                 rv = blkdev_issue_flush(mdev->ldev->backing_bdev, NULL);
949                 if (rv) {
950                         dev_err(DEV, "local disk flush failed with status %d\n", rv);
951                         /* would rather check on EOPNOTSUPP, but that is not reliable.
952                          * don't try again for ANY return value != 0
953                          * if (rv == -EOPNOTSUPP) */
954                         drbd_bump_write_ordering(mdev, WO_drain_io);
955                 }
956                 put_ldev(mdev);
957         }
958
959         return drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE);
960 }
961
962 static int w_flush(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
963 {
964         struct flush_work *fw = (struct flush_work *)w;
965         struct drbd_epoch *epoch = fw->epoch;
966
967         kfree(w);
968
969         if (!test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags))
970                 drbd_flush_after_epoch(mdev, epoch);
971
972         drbd_may_finish_epoch(mdev, epoch, EV_PUT |
973                               (mdev->state.conn < C_CONNECTED ? EV_CLEANUP : 0));
974
975         return 1;
976 }
977
978 /**
979  * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
980  * @mdev:       DRBD device.
981  * @epoch:      Epoch object.
982  * @ev:         Epoch event.
983  */
984 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *mdev,
985                                                struct drbd_epoch *epoch,
986                                                enum epoch_event ev)
987 {
988         int finish, epoch_size;
989         struct drbd_epoch *next_epoch;
990         int schedule_flush = 0;
991         enum finish_epoch rv = FE_STILL_LIVE;
992
993         spin_lock(&mdev->epoch_lock);
994         do {
995                 next_epoch = NULL;
996                 finish = 0;
997
998                 epoch_size = atomic_read(&epoch->epoch_size);
999
1000                 switch (ev & ~EV_CLEANUP) {
1001                 case EV_PUT:
1002                         atomic_dec(&epoch->active);
1003                         break;
1004                 case EV_GOT_BARRIER_NR:
1005                         set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1006
1007                         /* Special case: If we just switched from WO_bio_barrier to
1008                            WO_bdev_flush we should not finish the current epoch */
1009                         if (test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags) && epoch_size == 1 &&
1010                             mdev->write_ordering != WO_bio_barrier &&
1011                             epoch == mdev->current_epoch)
1012                                 clear_bit(DE_CONTAINS_A_BARRIER, &epoch->flags);
1013                         break;
1014                 case EV_BARRIER_DONE:
1015                         set_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE, &epoch->flags);
1016                         break;
1017                 case EV_BECAME_LAST:
1018                         /* nothing to do*/
1019                         break;
1020                 }
1021
1022                 if (epoch_size != 0 &&
1023                     atomic_read(&epoch->active) == 0 &&
1024                     test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) &&
1025                     epoch->list.prev == &mdev->current_epoch->list &&
1026                     !test_bit(DE_IS_FINISHING, &epoch->flags)) {
1027                         /* Nearly all conditions are met to finish that epoch... */
1028                         if (test_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE, &epoch->flags) ||
1029                             mdev->write_ordering == WO_none ||
1030                             (epoch_size == 1 && test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags)) ||
1031                             ev & EV_CLEANUP) {
1032                                 finish = 1;
1033                                 set_bit(DE_IS_FINISHING, &epoch->flags);
1034                         } else if (!test_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags) &&
1035                                  mdev->write_ordering == WO_bio_barrier) {
1036                                 atomic_inc(&epoch->active);
1037                                 schedule_flush = 1;
1038                         }
1039                 }
1040                 if (finish) {
1041                         if (!(ev & EV_CLEANUP)) {
1042                                 spin_unlock(&mdev->epoch_lock);
1043                                 drbd_send_b_ack(mdev, epoch->barrier_nr, epoch_size);
1044                                 spin_lock(&mdev->epoch_lock);
1045                         }
1046                         dec_unacked(mdev);
1047
1048                         if (mdev->current_epoch != epoch) {
1049                                 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1050                                 list_del(&epoch->list);
1051                                 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1052                                 mdev->epochs--;
1053                                 kfree(epoch);
1054
1055                                 if (rv == FE_STILL_LIVE)
1056                                         rv = FE_DESTROYED;
1057                         } else {
1058                                 epoch->flags = 0;
1059                                 atomic_set(&epoch->epoch_size, 0);
1060                                 /* atomic_set(&epoch->active, 0); is alrady zero */
1061                                 if (rv == FE_STILL_LIVE)
1062                                         rv = FE_RECYCLED;
1063                         }
1064                 }
1065
1066                 if (!next_epoch)
1067                         break;
1068
1069                 epoch = next_epoch;
1070         } while (1);
1071
1072         spin_unlock(&mdev->epoch_lock);
1073
1074         if (schedule_flush) {
1075                 struct flush_work *fw;
1076                 fw = kmalloc(sizeof(*fw), GFP_ATOMIC);
1077                 if (fw) {
1078                         fw->w.cb = w_flush;
1079                         fw->epoch = epoch;
1080                         drbd_queue_work(&mdev->data.work, &fw->w);
1081                 } else {
1082                         dev_warn(DEV, "Could not kmalloc a flush_work obj\n");
1083                         set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags);
1084                         /* That is not a recursion, only one level */
1085                         drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE);
1086                         drbd_may_finish_epoch(mdev, epoch, EV_PUT);
1087                 }
1088         }
1089
1090         return rv;
1091 }
1092
1093 /**
1094  * drbd_bump_write_ordering() - Fall back to an other write ordering method
1095  * @mdev:       DRBD device.
1096  * @wo:         Write ordering method to try.
1097  */
1098 void drbd_bump_write_ordering(struct drbd_conf *mdev, enum write_ordering_e wo) __must_hold(local)
1099 {
1100         enum write_ordering_e pwo;
1101         static char *write_ordering_str[] = {
1102                 [WO_none] = "none",
1103                 [WO_drain_io] = "drain",
1104                 [WO_bdev_flush] = "flush",
1105                 [WO_bio_barrier] = "barrier",
1106         };
1107
1108         pwo = mdev->write_ordering;
1109         wo = min(pwo, wo);
1110         if (wo == WO_bio_barrier && mdev->ldev->dc.no_disk_barrier)
1111                 wo = WO_bdev_flush;
1112         if (wo == WO_bdev_flush && mdev->ldev->dc.no_disk_flush)
1113                 wo = WO_drain_io;
1114         if (wo == WO_drain_io && mdev->ldev->dc.no_disk_drain)
1115                 wo = WO_none;
1116         mdev->write_ordering = wo;
1117         if (pwo != mdev->write_ordering || wo == WO_bio_barrier)
1118                 dev_info(DEV, "Method to ensure write ordering: %s\n", write_ordering_str[mdev->write_ordering]);
1119 }
1120
1121 /**
1122  * w_e_reissue() - Worker callback; Resubmit a bio, without BIO_RW_BARRIER set
1123  * @mdev:       DRBD device.
1124  * @w:          work object.
1125  * @cancel:     The connection will be closed anyways (unused in this callback)
1126  */
1127 int w_e_reissue(struct drbd_conf *mdev, struct drbd_work *w, int cancel) __releases(local)
1128 {
1129         struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1130         struct bio *bio = e->private_bio;
1131
1132         /* We leave DE_CONTAINS_A_BARRIER and EE_IS_BARRIER in place,
1133            (and DE_BARRIER_IN_NEXT_EPOCH_ISSUED in the previous Epoch)
1134            so that we can finish that epoch in drbd_may_finish_epoch().
1135            That is necessary if we already have a long chain of Epochs, before
1136            we realize that BIO_RW_BARRIER is actually not supported */
1137
1138         /* As long as the -ENOTSUPP on the barrier is reported immediately
1139            that will never trigger. If it is reported late, we will just
1140            print that warning and continue correctly for all future requests
1141            with WO_bdev_flush */
1142         if (previous_epoch(mdev, e->epoch))
1143                 dev_warn(DEV, "Write ordering was not enforced (one time event)\n");
1144
1145         /* prepare bio for re-submit,
1146          * re-init volatile members */
1147         /* we still have a local reference,
1148          * get_ldev was done in receive_Data. */
1149         bio->bi_bdev = mdev->ldev->backing_bdev;
1150         bio->bi_sector = e->sector;
1151         bio->bi_size = e->size;
1152         bio->bi_idx = 0;
1153
1154         bio->bi_flags &= ~(BIO_POOL_MASK - 1);
1155         bio->bi_flags |= 1 << BIO_UPTODATE;
1156
1157         /* don't know whether this is necessary: */
1158         bio->bi_phys_segments = 0;
1159         bio->bi_next = NULL;
1160
1161         /* these should be unchanged: */
1162         /* bio->bi_end_io = drbd_endio_write_sec; */
1163         /* bio->bi_vcnt = whatever; */
1164
1165         e->w.cb = e_end_block;
1166
1167         /* This is no longer a barrier request. */
1168         bio->bi_rw &= ~(1UL << BIO_RW_BARRIER);
1169
1170         drbd_generic_make_request(mdev, DRBD_FAULT_DT_WR, bio);
1171
1172         return 1;
1173 }
1174
1175 static int receive_Barrier(struct drbd_conf *mdev, struct p_header *h)
1176 {
1177         int rv, issue_flush;
1178         struct p_barrier *p = (struct p_barrier *)h;
1179         struct drbd_epoch *epoch;
1180
1181         ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
1182
1183         rv = drbd_recv(mdev, h->payload, h->length);
1184         ERR_IF(rv != h->length) return FALSE;
1185
1186         inc_unacked(mdev);
1187
1188         if (mdev->net_conf->wire_protocol != DRBD_PROT_C)
1189                 drbd_kick_lo(mdev);
1190
1191         mdev->current_epoch->barrier_nr = p->barrier;
1192         rv = drbd_may_finish_epoch(mdev, mdev->current_epoch, EV_GOT_BARRIER_NR);
1193
1194         /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1195          * the activity log, which means it would not be resynced in case the
1196          * R_PRIMARY crashes now.
1197          * Therefore we must send the barrier_ack after the barrier request was
1198          * completed. */
1199         switch (mdev->write_ordering) {
1200         case WO_bio_barrier:
1201         case WO_none:
1202                 if (rv == FE_RECYCLED)
1203                         return TRUE;
1204                 break;
1205
1206         case WO_bdev_flush:
1207         case WO_drain_io:
1208                 if (rv == FE_STILL_LIVE) {
1209                         set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &mdev->current_epoch->flags);
1210                         drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
1211                         rv = drbd_flush_after_epoch(mdev, mdev->current_epoch);
1212                 }
1213                 if (rv == FE_RECYCLED)
1214                         return TRUE;
1215
1216                 /* The asender will send all the ACKs and barrier ACKs out, since
1217                    all EEs moved from the active_ee to the done_ee. We need to
1218                    provide a new epoch object for the EEs that come in soon */
1219                 break;
1220         }
1221
1222         /* receiver context, in the writeout path of the other node.
1223          * avoid potential distributed deadlock */
1224         epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1225         if (!epoch) {
1226                 dev_warn(DEV, "Allocation of an epoch failed, slowing down\n");
1227                 issue_flush = !test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags);
1228                 drbd_wait_ee_list_empty(mdev, &mdev->active_ee);
1229                 if (issue_flush) {
1230                         rv = drbd_flush_after_epoch(mdev, mdev->current_epoch);
1231                         if (rv == FE_RECYCLED)
1232                                 return TRUE;
1233                 }
1234
1235                 drbd_wait_ee_list_empty(mdev, &mdev->done_ee);
1236
1237                 return TRUE;
1238         }
1239
1240         epoch->flags = 0;
1241         atomic_set(&epoch->epoch_size, 0);
1242         atomic_set(&epoch->active, 0);
1243
1244         spin_lock(&mdev->epoch_lock);
1245         if (atomic_read(&mdev->current_epoch->epoch_size)) {
1246                 list_add(&epoch->list, &mdev->current_epoch->list);
1247                 mdev->current_epoch = epoch;
1248                 mdev->epochs++;
1249         } else {
1250                 /* The current_epoch got recycled while we allocated this one... */
1251                 kfree(epoch);
1252         }
1253         spin_unlock(&mdev->epoch_lock);
1254
1255         return TRUE;
1256 }
1257
1258 /* used from receive_RSDataReply (recv_resync_read)
1259  * and from receive_Data */
1260 static struct drbd_epoch_entry *
1261 read_in_block(struct drbd_conf *mdev, u64 id, sector_t sector, int data_size) __must_hold(local)
1262 {
1263         struct drbd_epoch_entry *e;
1264         struct bio_vec *bvec;
1265         struct page *page;
1266         struct bio *bio;
1267         int dgs, ds, i, rr;
1268         void *dig_in = mdev->int_dig_in;
1269         void *dig_vv = mdev->int_dig_vv;
1270
1271         dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
1272                 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
1273
1274         if (dgs) {
1275                 rr = drbd_recv(mdev, dig_in, dgs);
1276                 if (rr != dgs) {
1277                         dev_warn(DEV, "short read receiving data digest: read %d expected %d\n",
1278                              rr, dgs);
1279                         return NULL;
1280                 }
1281         }
1282
1283         data_size -= dgs;
1284
1285         ERR_IF(data_size &  0x1ff) return NULL;
1286         ERR_IF(data_size >  DRBD_MAX_SEGMENT_SIZE) return NULL;
1287
1288         /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1289          * "criss-cross" setup, that might cause write-out on some other DRBD,
1290          * which in turn might block on the other node at this very place.  */
1291         e = drbd_alloc_ee(mdev, id, sector, data_size, GFP_NOIO);
1292         if (!e)
1293                 return NULL;
1294         bio = e->private_bio;
1295         ds = data_size;
1296         bio_for_each_segment(bvec, bio, i) {
1297                 page = bvec->bv_page;
1298                 rr = drbd_recv(mdev, kmap(page), min_t(int, ds, PAGE_SIZE));
1299                 kunmap(page);
1300                 if (rr != min_t(int, ds, PAGE_SIZE)) {
1301                         drbd_free_ee(mdev, e);
1302                         dev_warn(DEV, "short read receiving data: read %d expected %d\n",
1303                              rr, min_t(int, ds, PAGE_SIZE));
1304                         return NULL;
1305                 }
1306                 ds -= rr;
1307         }
1308
1309         if (dgs) {
1310                 drbd_csum(mdev, mdev->integrity_r_tfm, bio, dig_vv);
1311                 if (memcmp(dig_in, dig_vv, dgs)) {
1312                         dev_err(DEV, "Digest integrity check FAILED.\n");
1313                         drbd_bcast_ee(mdev, "digest failed",
1314                                         dgs, dig_in, dig_vv, e);
1315                         drbd_free_ee(mdev, e);
1316                         return NULL;
1317                 }
1318         }
1319         mdev->recv_cnt += data_size>>9;
1320         return e;
1321 }
1322
1323 /* drbd_drain_block() just takes a data block
1324  * out of the socket input buffer, and discards it.
1325  */
1326 static int drbd_drain_block(struct drbd_conf *mdev, int data_size)
1327 {
1328         struct page *page;
1329         int rr, rv = 1;
1330         void *data;
1331
1332         page = drbd_pp_alloc(mdev, 1);
1333
1334         data = kmap(page);
1335         while (data_size) {
1336                 rr = drbd_recv(mdev, data, min_t(int, data_size, PAGE_SIZE));
1337                 if (rr != min_t(int, data_size, PAGE_SIZE)) {
1338                         rv = 0;
1339                         dev_warn(DEV, "short read receiving data: read %d expected %d\n",
1340                              rr, min_t(int, data_size, PAGE_SIZE));
1341                         break;
1342                 }
1343                 data_size -= rr;
1344         }
1345         kunmap(page);
1346         drbd_pp_free(mdev, page);
1347         return rv;
1348 }
1349
1350 static int recv_dless_read(struct drbd_conf *mdev, struct drbd_request *req,
1351                            sector_t sector, int data_size)
1352 {
1353         struct bio_vec *bvec;
1354         struct bio *bio;
1355         int dgs, rr, i, expect;
1356         void *dig_in = mdev->int_dig_in;
1357         void *dig_vv = mdev->int_dig_vv;
1358
1359         dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
1360                 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
1361
1362         if (dgs) {
1363                 rr = drbd_recv(mdev, dig_in, dgs);
1364                 if (rr != dgs) {
1365                         dev_warn(DEV, "short read receiving data reply digest: read %d expected %d\n",
1366                              rr, dgs);
1367                         return 0;
1368                 }
1369         }
1370
1371         data_size -= dgs;
1372
1373         /* optimistically update recv_cnt.  if receiving fails below,
1374          * we disconnect anyways, and counters will be reset. */
1375         mdev->recv_cnt += data_size>>9;
1376
1377         bio = req->master_bio;
1378         D_ASSERT(sector == bio->bi_sector);
1379
1380         bio_for_each_segment(bvec, bio, i) {
1381                 expect = min_t(int, data_size, bvec->bv_len);
1382                 rr = drbd_recv(mdev,
1383                              kmap(bvec->bv_page)+bvec->bv_offset,
1384                              expect);
1385                 kunmap(bvec->bv_page);
1386                 if (rr != expect) {
1387                         dev_warn(DEV, "short read receiving data reply: "
1388                              "read %d expected %d\n",
1389                              rr, expect);
1390                         return 0;
1391                 }
1392                 data_size -= rr;
1393         }
1394
1395         if (dgs) {
1396                 drbd_csum(mdev, mdev->integrity_r_tfm, bio, dig_vv);
1397                 if (memcmp(dig_in, dig_vv, dgs)) {
1398                         dev_err(DEV, "Digest integrity check FAILED. Broken NICs?\n");
1399                         return 0;
1400                 }
1401         }
1402
1403         D_ASSERT(data_size == 0);
1404         return 1;
1405 }
1406
1407 /* e_end_resync_block() is called via
1408  * drbd_process_done_ee() by asender only */
1409 static int e_end_resync_block(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1410 {
1411         struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1412         sector_t sector = e->sector;
1413         int ok;
1414
1415         D_ASSERT(hlist_unhashed(&e->colision));
1416
1417         if (likely(drbd_bio_uptodate(e->private_bio))) {
1418                 drbd_set_in_sync(mdev, sector, e->size);
1419                 ok = drbd_send_ack(mdev, P_RS_WRITE_ACK, e);
1420         } else {
1421                 /* Record failure to sync */
1422                 drbd_rs_failed_io(mdev, sector, e->size);
1423
1424                 ok  = drbd_send_ack(mdev, P_NEG_ACK, e);
1425         }
1426         dec_unacked(mdev);
1427
1428         return ok;
1429 }
1430
1431 static int recv_resync_read(struct drbd_conf *mdev, sector_t sector, int data_size) __releases(local)
1432 {
1433         struct drbd_epoch_entry *e;
1434
1435         e = read_in_block(mdev, ID_SYNCER, sector, data_size);
1436         if (!e) {
1437                 put_ldev(mdev);
1438                 return FALSE;
1439         }
1440
1441         dec_rs_pending(mdev);
1442
1443         e->private_bio->bi_end_io = drbd_endio_write_sec;
1444         e->private_bio->bi_rw = WRITE;
1445         e->w.cb = e_end_resync_block;
1446
1447         inc_unacked(mdev);
1448         /* corresponding dec_unacked() in e_end_resync_block()
1449          * respective _drbd_clear_done_ee */
1450
1451         spin_lock_irq(&mdev->req_lock);
1452         list_add(&e->w.list, &mdev->sync_ee);
1453         spin_unlock_irq(&mdev->req_lock);
1454
1455         drbd_generic_make_request(mdev, DRBD_FAULT_RS_WR, e->private_bio);
1456         /* accounting done in endio */
1457
1458         maybe_kick_lo(mdev);
1459         return TRUE;
1460 }
1461
1462 static int receive_DataReply(struct drbd_conf *mdev, struct p_header *h)
1463 {
1464         struct drbd_request *req;
1465         sector_t sector;
1466         unsigned int header_size, data_size;
1467         int ok;
1468         struct p_data *p = (struct p_data *)h;
1469
1470         header_size = sizeof(*p) - sizeof(*h);
1471         data_size   = h->length  - header_size;
1472
1473         ERR_IF(data_size == 0) return FALSE;
1474
1475         if (drbd_recv(mdev, h->payload, header_size) != header_size)
1476                 return FALSE;
1477
1478         sector = be64_to_cpu(p->sector);
1479
1480         spin_lock_irq(&mdev->req_lock);
1481         req = _ar_id_to_req(mdev, p->block_id, sector);
1482         spin_unlock_irq(&mdev->req_lock);
1483         if (unlikely(!req)) {
1484                 dev_err(DEV, "Got a corrupt block_id/sector pair(1).\n");
1485                 return FALSE;
1486         }
1487
1488         /* hlist_del(&req->colision) is done in _req_may_be_done, to avoid
1489          * special casing it there for the various failure cases.
1490          * still no race with drbd_fail_pending_reads */
1491         ok = recv_dless_read(mdev, req, sector, data_size);
1492
1493         if (ok)
1494                 req_mod(req, data_received);
1495         /* else: nothing. handled from drbd_disconnect...
1496          * I don't think we may complete this just yet
1497          * in case we are "on-disconnect: freeze" */
1498
1499         return ok;
1500 }
1501
1502 static int receive_RSDataReply(struct drbd_conf *mdev, struct p_header *h)
1503 {
1504         sector_t sector;
1505         unsigned int header_size, data_size;
1506         int ok;
1507         struct p_data *p = (struct p_data *)h;
1508
1509         header_size = sizeof(*p) - sizeof(*h);
1510         data_size   = h->length  - header_size;
1511
1512         ERR_IF(data_size == 0) return FALSE;
1513
1514         if (drbd_recv(mdev, h->payload, header_size) != header_size)
1515                 return FALSE;
1516
1517         sector = be64_to_cpu(p->sector);
1518         D_ASSERT(p->block_id == ID_SYNCER);
1519
1520         if (get_ldev(mdev)) {
1521                 /* data is submitted to disk within recv_resync_read.
1522                  * corresponding put_ldev done below on error,
1523                  * or in drbd_endio_write_sec. */
1524                 ok = recv_resync_read(mdev, sector, data_size);
1525         } else {
1526                 if (__ratelimit(&drbd_ratelimit_state))
1527                         dev_err(DEV, "Can not write resync data to local disk.\n");
1528
1529                 ok = drbd_drain_block(mdev, data_size);
1530
1531                 drbd_send_ack_dp(mdev, P_NEG_ACK, p);
1532         }
1533
1534         return ok;
1535 }
1536
1537 /* e_end_block() is called via drbd_process_done_ee().
1538  * this means this function only runs in the asender thread
1539  */
1540 static int e_end_block(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1541 {
1542         struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1543         sector_t sector = e->sector;
1544         struct drbd_epoch *epoch;
1545         int ok = 1, pcmd;
1546
1547         if (e->flags & EE_IS_BARRIER) {
1548                 epoch = previous_epoch(mdev, e->epoch);
1549                 if (epoch)
1550                         drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE + (cancel ? EV_CLEANUP : 0));
1551         }
1552
1553         if (mdev->net_conf->wire_protocol == DRBD_PROT_C) {
1554                 if (likely(drbd_bio_uptodate(e->private_bio))) {
1555                         pcmd = (mdev->state.conn >= C_SYNC_SOURCE &&
1556                                 mdev->state.conn <= C_PAUSED_SYNC_T &&
1557                                 e->flags & EE_MAY_SET_IN_SYNC) ?
1558                                 P_RS_WRITE_ACK : P_WRITE_ACK;
1559                         ok &= drbd_send_ack(mdev, pcmd, e);
1560                         if (pcmd == P_RS_WRITE_ACK)
1561                                 drbd_set_in_sync(mdev, sector, e->size);
1562                 } else {
1563                         ok  = drbd_send_ack(mdev, P_NEG_ACK, e);
1564                         /* we expect it to be marked out of sync anyways...
1565                          * maybe assert this?  */
1566                 }
1567                 dec_unacked(mdev);
1568         }
1569         /* we delete from the conflict detection hash _after_ we sent out the
1570          * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right.  */
1571         if (mdev->net_conf->two_primaries) {
1572                 spin_lock_irq(&mdev->req_lock);
1573                 D_ASSERT(!hlist_unhashed(&e->colision));
1574                 hlist_del_init(&e->colision);
1575                 spin_unlock_irq(&mdev->req_lock);
1576         } else {
1577                 D_ASSERT(hlist_unhashed(&e->colision));
1578         }
1579
1580         drbd_may_finish_epoch(mdev, e->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
1581
1582         return ok;
1583 }
1584
1585 static int e_send_discard_ack(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1586 {
1587         struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w;
1588         int ok = 1;
1589
1590         D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
1591         ok = drbd_send_ack(mdev, P_DISCARD_ACK, e);
1592
1593         spin_lock_irq(&mdev->req_lock);
1594         D_ASSERT(!hlist_unhashed(&e->colision));
1595         hlist_del_init(&e->colision);
1596         spin_unlock_irq(&mdev->req_lock);
1597
1598         dec_unacked(mdev);
1599
1600         return ok;
1601 }
1602
1603 /* Called from receive_Data.
1604  * Synchronize packets on sock with packets on msock.
1605  *
1606  * This is here so even when a P_DATA packet traveling via sock overtook an Ack
1607  * packet traveling on msock, they are still processed in the order they have
1608  * been sent.
1609  *
1610  * Note: we don't care for Ack packets overtaking P_DATA packets.
1611  *
1612  * In case packet_seq is larger than mdev->peer_seq number, there are
1613  * outstanding packets on the msock. We wait for them to arrive.
1614  * In case we are the logically next packet, we update mdev->peer_seq
1615  * ourselves. Correctly handles 32bit wrap around.
1616  *
1617  * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
1618  * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
1619  * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
1620  * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
1621  *
1622  * returns 0 if we may process the packet,
1623  * -ERESTARTSYS if we were interrupted (by disconnect signal). */
1624 static int drbd_wait_peer_seq(struct drbd_conf *mdev, const u32 packet_seq)
1625 {
1626         DEFINE_WAIT(wait);
1627         unsigned int p_seq;
1628         long timeout;
1629         int ret = 0;
1630         spin_lock(&mdev->peer_seq_lock);
1631         for (;;) {
1632                 prepare_to_wait(&mdev->seq_wait, &wait, TASK_INTERRUPTIBLE);
1633                 if (seq_le(packet_seq, mdev->peer_seq+1))
1634                         break;
1635                 if (signal_pending(current)) {
1636                         ret = -ERESTARTSYS;
1637                         break;
1638                 }
1639                 p_seq = mdev->peer_seq;
1640                 spin_unlock(&mdev->peer_seq_lock);
1641                 timeout = schedule_timeout(30*HZ);
1642                 spin_lock(&mdev->peer_seq_lock);
1643                 if (timeout == 0 && p_seq == mdev->peer_seq) {
1644                         ret = -ETIMEDOUT;
1645                         dev_err(DEV, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n");
1646                         break;
1647                 }
1648         }
1649         finish_wait(&mdev->seq_wait, &wait);
1650         if (mdev->peer_seq+1 == packet_seq)
1651                 mdev->peer_seq++;
1652         spin_unlock(&mdev->peer_seq_lock);
1653         return ret;
1654 }
1655
1656 /* mirrored write */
1657 static int receive_Data(struct drbd_conf *mdev, struct p_header *h)
1658 {
1659         sector_t sector;
1660         struct drbd_epoch_entry *e;
1661         struct p_data *p = (struct p_data *)h;
1662         int header_size, data_size;
1663         int rw = WRITE;
1664         u32 dp_flags;
1665
1666         header_size = sizeof(*p) - sizeof(*h);
1667         data_size   = h->length  - header_size;
1668
1669         ERR_IF(data_size == 0) return FALSE;
1670
1671         if (drbd_recv(mdev, h->payload, header_size) != header_size)
1672                 return FALSE;
1673
1674         if (!get_ldev(mdev)) {
1675                 if (__ratelimit(&drbd_ratelimit_state))
1676                         dev_err(DEV, "Can not write mirrored data block "
1677                             "to local disk.\n");
1678                 spin_lock(&mdev->peer_seq_lock);
1679                 if (mdev->peer_seq+1 == be32_to_cpu(p->seq_num))
1680                         mdev->peer_seq++;
1681                 spin_unlock(&mdev->peer_seq_lock);
1682
1683                 drbd_send_ack_dp(mdev, P_NEG_ACK, p);
1684                 atomic_inc(&mdev->current_epoch->epoch_size);
1685                 return drbd_drain_block(mdev, data_size);
1686         }
1687
1688         /* get_ldev(mdev) successful.
1689          * Corresponding put_ldev done either below (on various errors),
1690          * or in drbd_endio_write_sec, if we successfully submit the data at
1691          * the end of this function. */
1692
1693         sector = be64_to_cpu(p->sector);
1694         e = read_in_block(mdev, p->block_id, sector, data_size);
1695         if (!e) {
1696                 put_ldev(mdev);
1697                 return FALSE;
1698         }
1699
1700         e->private_bio->bi_end_io = drbd_endio_write_sec;
1701         e->w.cb = e_end_block;
1702
1703         spin_lock(&mdev->epoch_lock);
1704         e->epoch = mdev->current_epoch;
1705         atomic_inc(&e->epoch->epoch_size);
1706         atomic_inc(&e->epoch->active);
1707
1708         if (mdev->write_ordering == WO_bio_barrier && atomic_read(&e->epoch->epoch_size) == 1) {
1709                 struct drbd_epoch *epoch;
1710                 /* Issue a barrier if we start a new epoch, and the previous epoch
1711                    was not a epoch containing a single request which already was
1712                    a Barrier. */
1713                 epoch = list_entry(e->epoch->list.prev, struct drbd_epoch, list);
1714                 if (epoch == e->epoch) {
1715                         set_bit(DE_CONTAINS_A_BARRIER, &e->epoch->flags);
1716                         rw |= (1<<BIO_RW_BARRIER);
1717                         e->flags |= EE_IS_BARRIER;
1718                 } else {
1719                         if (atomic_read(&epoch->epoch_size) > 1 ||
1720                             !test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags)) {
1721                                 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags);
1722                                 set_bit(DE_CONTAINS_A_BARRIER, &e->epoch->flags);
1723                                 rw |= (1<<BIO_RW_BARRIER);
1724                                 e->flags |= EE_IS_BARRIER;
1725                         }
1726                 }
1727         }
1728         spin_unlock(&mdev->epoch_lock);
1729
1730         dp_flags = be32_to_cpu(p->dp_flags);
1731         if (dp_flags & DP_HARDBARRIER) {
1732                 dev_err(DEV, "ASSERT FAILED would have submitted barrier request\n");
1733                 /* rw |= (1<<BIO_RW_BARRIER); */
1734         }
1735         if (dp_flags & DP_RW_SYNC)
1736                 rw |= (1<<BIO_RW_SYNCIO) | (1<<BIO_RW_UNPLUG);
1737         if (dp_flags & DP_MAY_SET_IN_SYNC)
1738                 e->flags |= EE_MAY_SET_IN_SYNC;
1739
1740         /* I'm the receiver, I do hold a net_cnt reference. */
1741         if (!mdev->net_conf->two_primaries) {
1742                 spin_lock_irq(&mdev->req_lock);
1743         } else {
1744                 /* don't get the req_lock yet,
1745                  * we may sleep in drbd_wait_peer_seq */
1746                 const int size = e->size;
1747                 const int discard = test_bit(DISCARD_CONCURRENT, &mdev->flags);
1748                 DEFINE_WAIT(wait);
1749                 struct drbd_request *i;
1750                 struct hlist_node *n;
1751                 struct hlist_head *slot;
1752                 int first;
1753
1754                 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C);
1755                 BUG_ON(mdev->ee_hash == NULL);
1756                 BUG_ON(mdev->tl_hash == NULL);
1757
1758                 /* conflict detection and handling:
1759                  * 1. wait on the sequence number,
1760                  *    in case this data packet overtook ACK packets.
1761                  * 2. check our hash tables for conflicting requests.
1762                  *    we only need to walk the tl_hash, since an ee can not
1763                  *    have a conflict with an other ee: on the submitting
1764                  *    node, the corresponding req had already been conflicting,
1765                  *    and a conflicting req is never sent.
1766                  *
1767                  * Note: for two_primaries, we are protocol C,
1768                  * so there cannot be any request that is DONE
1769                  * but still on the transfer log.
1770                  *
1771                  * unconditionally add to the ee_hash.
1772                  *
1773                  * if no conflicting request is found:
1774                  *    submit.
1775                  *
1776                  * if any conflicting request is found
1777                  * that has not yet been acked,
1778                  * AND I have the "discard concurrent writes" flag:
1779                  *       queue (via done_ee) the P_DISCARD_ACK; OUT.
1780                  *
1781                  * if any conflicting request is found:
1782                  *       block the receiver, waiting on misc_wait
1783                  *       until no more conflicting requests are there,
1784                  *       or we get interrupted (disconnect).
1785                  *
1786                  *       we do not just write after local io completion of those
1787                  *       requests, but only after req is done completely, i.e.
1788                  *       we wait for the P_DISCARD_ACK to arrive!
1789                  *
1790                  *       then proceed normally, i.e. submit.
1791                  */
1792                 if (drbd_wait_peer_seq(mdev, be32_to_cpu(p->seq_num)))
1793                         goto out_interrupted;
1794
1795                 spin_lock_irq(&mdev->req_lock);
1796
1797                 hlist_add_head(&e->colision, ee_hash_slot(mdev, sector));
1798
1799 #define OVERLAPS overlaps(i->sector, i->size, sector, size)
1800                 slot = tl_hash_slot(mdev, sector);
1801                 first = 1;
1802                 for (;;) {
1803                         int have_unacked = 0;
1804                         int have_conflict = 0;
1805                         prepare_to_wait(&mdev->misc_wait, &wait,
1806                                 TASK_INTERRUPTIBLE);
1807                         hlist_for_each_entry(i, n, slot, colision) {
1808                                 if (OVERLAPS) {
1809                                         /* only ALERT on first iteration,
1810                                          * we may be woken up early... */
1811                                         if (first)
1812                                                 dev_alert(DEV, "%s[%u] Concurrent local write detected!"
1813                                                       " new: %llus +%u; pending: %llus +%u\n",
1814                                                       current->comm, current->pid,
1815                                                       (unsigned long long)sector, size,
1816                                                       (unsigned long long)i->sector, i->size);
1817                                         if (i->rq_state & RQ_NET_PENDING)
1818                                                 ++have_unacked;
1819                                         ++have_conflict;
1820                                 }
1821                         }
1822 #undef OVERLAPS
1823                         if (!have_conflict)
1824                                 break;
1825
1826                         /* Discard Ack only for the _first_ iteration */
1827                         if (first && discard && have_unacked) {
1828                                 dev_alert(DEV, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n",
1829                                      (unsigned long long)sector);
1830                                 inc_unacked(mdev);
1831                                 e->w.cb = e_send_discard_ack;
1832                                 list_add_tail(&e->w.list, &mdev->done_ee);
1833
1834                                 spin_unlock_irq(&mdev->req_lock);
1835
1836                                 /* we could probably send that P_DISCARD_ACK ourselves,
1837                                  * but I don't like the receiver using the msock */
1838
1839                                 put_ldev(mdev);
1840                                 wake_asender(mdev);
1841                                 finish_wait(&mdev->misc_wait, &wait);
1842                                 return TRUE;
1843                         }
1844
1845                         if (signal_pending(current)) {
1846                                 hlist_del_init(&e->colision);
1847
1848                                 spin_unlock_irq(&mdev->req_lock);
1849
1850                                 finish_wait(&mdev->misc_wait, &wait);
1851                                 goto out_interrupted;
1852                         }
1853
1854                         spin_unlock_irq(&mdev->req_lock);
1855                         if (first) {
1856                                 first = 0;
1857                                 dev_alert(DEV, "Concurrent write! [W AFTERWARDS] "
1858                                      "sec=%llus\n", (unsigned long long)sector);
1859                         } else if (discard) {
1860                                 /* we had none on the first iteration.
1861                                  * there must be none now. */
1862                                 D_ASSERT(have_unacked == 0);
1863                         }
1864                         schedule();
1865                         spin_lock_irq(&mdev->req_lock);
1866                 }
1867                 finish_wait(&mdev->misc_wait, &wait);
1868         }
1869
1870         list_add(&e->w.list, &mdev->active_ee);
1871         spin_unlock_irq(&mdev->req_lock);
1872
1873         switch (mdev->net_conf->wire_protocol) {
1874         case DRBD_PROT_C:
1875                 inc_unacked(mdev);
1876                 /* corresponding dec_unacked() in e_end_block()
1877                  * respective _drbd_clear_done_ee */
1878                 break;
1879         case DRBD_PROT_B:
1880                 /* I really don't like it that the receiver thread
1881                  * sends on the msock, but anyways */
1882                 drbd_send_ack(mdev, P_RECV_ACK, e);
1883                 break;
1884         case DRBD_PROT_A:
1885                 /* nothing to do */
1886                 break;
1887         }
1888
1889         if (mdev->state.pdsk == D_DISKLESS) {
1890                 /* In case we have the only disk of the cluster, */
1891                 drbd_set_out_of_sync(mdev, e->sector, e->size);
1892                 e->flags |= EE_CALL_AL_COMPLETE_IO;
1893                 drbd_al_begin_io(mdev, e->sector);
1894         }
1895
1896         e->private_bio->bi_rw = rw;
1897         drbd_generic_make_request(mdev, DRBD_FAULT_DT_WR, e->private_bio);
1898         /* accounting done in endio */
1899
1900         maybe_kick_lo(mdev);
1901         return TRUE;
1902
1903 out_interrupted:
1904         /* yes, the epoch_size now is imbalanced.
1905          * but we drop the connection anyways, so we don't have a chance to
1906          * receive a barrier... atomic_inc(&mdev->epoch_size); */
1907         put_ldev(mdev);
1908         drbd_free_ee(mdev, e);
1909         return FALSE;
1910 }
1911
1912 static int receive_DataRequest(struct drbd_conf *mdev, struct p_header *h)
1913 {
1914         sector_t sector;
1915         const sector_t capacity = drbd_get_capacity(mdev->this_bdev);
1916         struct drbd_epoch_entry *e;
1917         struct digest_info *di = NULL;
1918         int size, digest_size;
1919         unsigned int fault_type;
1920         struct p_block_req *p =
1921                 (struct p_block_req *)h;
1922         const int brps = sizeof(*p)-sizeof(*h);
1923
1924         if (drbd_recv(mdev, h->payload, brps) != brps)
1925                 return FALSE;
1926
1927         sector = be64_to_cpu(p->sector);
1928         size   = be32_to_cpu(p->blksize);
1929
1930         if (size <= 0 || (size & 0x1ff) != 0 || size > DRBD_MAX_SEGMENT_SIZE) {
1931                 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
1932                                 (unsigned long long)sector, size);
1933                 return FALSE;
1934         }
1935         if (sector + (size>>9) > capacity) {
1936                 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
1937                                 (unsigned long long)sector, size);
1938                 return FALSE;
1939         }
1940
1941         if (!get_ldev_if_state(mdev, D_UP_TO_DATE)) {
1942                 if (__ratelimit(&drbd_ratelimit_state))
1943                         dev_err(DEV, "Can not satisfy peer's read request, "
1944                             "no local data.\n");
1945                 drbd_send_ack_rp(mdev, h->command == P_DATA_REQUEST ? P_NEG_DREPLY :
1946                                  P_NEG_RS_DREPLY , p);
1947                 return TRUE;
1948         }
1949
1950         /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1951          * "criss-cross" setup, that might cause write-out on some other DRBD,
1952          * which in turn might block on the other node at this very place.  */
1953         e = drbd_alloc_ee(mdev, p->block_id, sector, size, GFP_NOIO);
1954         if (!e) {
1955                 put_ldev(mdev);
1956                 return FALSE;
1957         }
1958
1959         e->private_bio->bi_rw = READ;
1960         e->private_bio->bi_end_io = drbd_endio_read_sec;
1961
1962         switch (h->command) {
1963         case P_DATA_REQUEST:
1964                 e->w.cb = w_e_end_data_req;
1965                 fault_type = DRBD_FAULT_DT_RD;
1966                 break;
1967         case P_RS_DATA_REQUEST:
1968                 e->w.cb = w_e_end_rsdata_req;
1969                 fault_type = DRBD_FAULT_RS_RD;
1970                 /* Eventually this should become asynchronously. Currently it
1971                  * blocks the whole receiver just to delay the reading of a
1972                  * resync data block.
1973                  * the drbd_work_queue mechanism is made for this...
1974                  */
1975                 if (!drbd_rs_begin_io(mdev, sector)) {
1976                         /* we have been interrupted,
1977                          * probably connection lost! */
1978                         D_ASSERT(signal_pending(current));
1979                         goto out_free_e;
1980                 }
1981                 break;
1982
1983         case P_OV_REPLY:
1984         case P_CSUM_RS_REQUEST:
1985                 fault_type = DRBD_FAULT_RS_RD;
1986                 digest_size = h->length - brps ;
1987                 di = kmalloc(sizeof(*di) + digest_size, GFP_NOIO);
1988                 if (!di)
1989                         goto out_free_e;
1990
1991                 di->digest_size = digest_size;
1992                 di->digest = (((char *)di)+sizeof(struct digest_info));
1993
1994                 if (drbd_recv(mdev, di->digest, digest_size) != digest_size)
1995                         goto out_free_e;
1996
1997                 e->block_id = (u64)(unsigned long)di;
1998                 if (h->command == P_CSUM_RS_REQUEST) {
1999                         D_ASSERT(mdev->agreed_pro_version >= 89);
2000                         e->w.cb = w_e_end_csum_rs_req;
2001                 } else if (h->command == P_OV_REPLY) {
2002                         e->w.cb = w_e_end_ov_reply;
2003                         dec_rs_pending(mdev);
2004                         break;
2005                 }
2006
2007                 if (!drbd_rs_begin_io(mdev, sector)) {
2008                         /* we have been interrupted, probably connection lost! */
2009                         D_ASSERT(signal_pending(current));
2010                         goto out_free_e;
2011                 }
2012                 break;
2013
2014         case P_OV_REQUEST:
2015                 if (mdev->state.conn >= C_CONNECTED &&
2016                     mdev->state.conn != C_VERIFY_T)
2017                         dev_warn(DEV, "ASSERT FAILED: got P_OV_REQUEST while being %s\n",
2018                                 drbd_conn_str(mdev->state.conn));
2019                 if (mdev->ov_start_sector == ~(sector_t)0 &&
2020                     mdev->agreed_pro_version >= 90) {
2021                         mdev->ov_start_sector = sector;
2022                         mdev->ov_position = sector;
2023                         mdev->ov_left = mdev->rs_total - BM_SECT_TO_BIT(sector);
2024                         dev_info(DEV, "Online Verify start sector: %llu\n",
2025                                         (unsigned long long)sector);
2026                 }
2027                 e->w.cb = w_e_end_ov_req;
2028                 fault_type = DRBD_FAULT_RS_RD;
2029                 /* Eventually this should become asynchronous. Currently it
2030                  * blocks the whole receiver just to delay the reading of a
2031                  * resync data block.
2032                  * the drbd_work_queue mechanism is made for this...
2033                  */
2034                 if (!drbd_rs_begin_io(mdev, sector)) {
2035                         /* we have been interrupted,
2036                          * probably connection lost! */
2037                         D_ASSERT(signal_pending(current));
2038                         goto out_free_e;
2039                 }
2040                 break;
2041
2042
2043         default:
2044                 dev_err(DEV, "unexpected command (%s) in receive_DataRequest\n",
2045                     cmdname(h->command));
2046                 fault_type = DRBD_FAULT_MAX;
2047         }
2048
2049         spin_lock_irq(&mdev->req_lock);
2050         list_add(&e->w.list, &mdev->read_ee);
2051         spin_unlock_irq(&mdev->req_lock);
2052
2053         inc_unacked(mdev);
2054
2055         drbd_generic_make_request(mdev, fault_type, e->private_bio);
2056         maybe_kick_lo(mdev);
2057
2058         return TRUE;
2059
2060 out_free_e:
2061         kfree(di);
2062         put_ldev(mdev);
2063         drbd_free_ee(mdev, e);
2064         return FALSE;
2065 }
2066
2067 static int drbd_asb_recover_0p(struct drbd_conf *mdev) __must_hold(local)
2068 {
2069         int self, peer, rv = -100;
2070         unsigned long ch_self, ch_peer;
2071
2072         self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2073         peer = mdev->p_uuid[UI_BITMAP] & 1;
2074
2075         ch_peer = mdev->p_uuid[UI_SIZE];
2076         ch_self = mdev->comm_bm_set;
2077
2078         switch (mdev->net_conf->after_sb_0p) {
2079         case ASB_CONSENSUS:
2080         case ASB_DISCARD_SECONDARY:
2081         case ASB_CALL_HELPER:
2082                 dev_err(DEV, "Configuration error.\n");
2083                 break;
2084         case ASB_DISCONNECT:
2085                 break;
2086         case ASB_DISCARD_YOUNGER_PRI:
2087                 if (self == 0 && peer == 1) {
2088                         rv = -1;
2089                         break;
2090                 }
2091                 if (self == 1 && peer == 0) {
2092                         rv =  1;
2093                         break;
2094                 }
2095                 /* Else fall through to one of the other strategies... */
2096         case ASB_DISCARD_OLDER_PRI:
2097                 if (self == 0 && peer == 1) {
2098                         rv = 1;
2099                         break;
2100                 }
2101                 if (self == 1 && peer == 0) {
2102                         rv = -1;
2103                         break;
2104                 }
2105                 /* Else fall through to one of the other strategies... */
2106                 dev_warn(DEV, "Discard younger/older primary did not find a decision\n"
2107                      "Using discard-least-changes instead\n");
2108         case ASB_DISCARD_ZERO_CHG:
2109                 if (ch_peer == 0 && ch_self == 0) {
2110                         rv = test_bit(DISCARD_CONCURRENT, &mdev->flags)
2111                                 ? -1 : 1;
2112                         break;
2113                 } else {
2114                         if (ch_peer == 0) { rv =  1; break; }
2115                         if (ch_self == 0) { rv = -1; break; }
2116                 }
2117                 if (mdev->net_conf->after_sb_0p == ASB_DISCARD_ZERO_CHG)
2118                         break;
2119         case ASB_DISCARD_LEAST_CHG:
2120                 if      (ch_self < ch_peer)
2121                         rv = -1;
2122                 else if (ch_self > ch_peer)
2123                         rv =  1;
2124                 else /* ( ch_self == ch_peer ) */
2125                      /* Well, then use something else. */
2126                         rv = test_bit(DISCARD_CONCURRENT, &mdev->flags)
2127                                 ? -1 : 1;
2128                 break;
2129         case ASB_DISCARD_LOCAL:
2130                 rv = -1;
2131                 break;
2132         case ASB_DISCARD_REMOTE:
2133                 rv =  1;
2134         }
2135
2136         return rv;
2137 }
2138
2139 static int drbd_asb_recover_1p(struct drbd_conf *mdev) __must_hold(local)
2140 {
2141         int self, peer, hg, rv = -100;
2142
2143         self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2144         peer = mdev->p_uuid[UI_BITMAP] & 1;
2145
2146         switch (mdev->net_conf->after_sb_1p) {
2147         case ASB_DISCARD_YOUNGER_PRI:
2148         case ASB_DISCARD_OLDER_PRI:
2149         case ASB_DISCARD_LEAST_CHG:
2150         case ASB_DISCARD_LOCAL:
2151         case ASB_DISCARD_REMOTE:
2152                 dev_err(DEV, "Configuration error.\n");
2153                 break;
2154         case ASB_DISCONNECT:
2155                 break;
2156         case ASB_CONSENSUS:
2157                 hg = drbd_asb_recover_0p(mdev);
2158                 if (hg == -1 && mdev->state.role == R_SECONDARY)
2159                         rv = hg;
2160                 if (hg == 1  && mdev->state.role == R_PRIMARY)
2161                         rv = hg;
2162                 break;
2163         case ASB_VIOLENTLY:
2164                 rv = drbd_asb_recover_0p(mdev);
2165                 break;
2166         case ASB_DISCARD_SECONDARY:
2167                 return mdev->state.role == R_PRIMARY ? 1 : -1;
2168         case ASB_CALL_HELPER:
2169                 hg = drbd_asb_recover_0p(mdev);
2170                 if (hg == -1 && mdev->state.role == R_PRIMARY) {
2171                         self = drbd_set_role(mdev, R_SECONDARY, 0);
2172                          /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2173                           * we might be here in C_WF_REPORT_PARAMS which is transient.
2174                           * we do not need to wait for the after state change work either. */
2175                         self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY));
2176                         if (self != SS_SUCCESS) {
2177                                 drbd_khelper(mdev, "pri-lost-after-sb");
2178                         } else {
2179                                 dev_warn(DEV, "Successfully gave up primary role.\n");
2180                                 rv = hg;
2181                         }
2182                 } else
2183                         rv = hg;
2184         }
2185
2186         return rv;
2187 }
2188
2189 static int drbd_asb_recover_2p(struct drbd_conf *mdev) __must_hold(local)
2190 {
2191         int self, peer, hg, rv = -100;
2192
2193         self = mdev->ldev->md.uuid[UI_BITMAP] & 1;
2194         peer = mdev->p_uuid[UI_BITMAP] & 1;
2195
2196         switch (mdev->net_conf->after_sb_2p) {
2197         case ASB_DISCARD_YOUNGER_PRI:
2198         case ASB_DISCARD_OLDER_PRI:
2199         case ASB_DISCARD_LEAST_CHG:
2200         case ASB_DISCARD_LOCAL:
2201         case ASB_DISCARD_REMOTE:
2202         case ASB_CONSENSUS:
2203         case ASB_DISCARD_SECONDARY:
2204                 dev_err(DEV, "Configuration error.\n");
2205                 break;
2206         case ASB_VIOLENTLY:
2207                 rv = drbd_asb_recover_0p(mdev);
2208                 break;
2209         case ASB_DISCONNECT:
2210                 break;
2211         case ASB_CALL_HELPER:
2212                 hg = drbd_asb_recover_0p(mdev);
2213                 if (hg == -1) {
2214                          /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2215                           * we might be here in C_WF_REPORT_PARAMS which is transient.
2216                           * we do not need to wait for the after state change work either. */
2217                         self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY));
2218                         if (self != SS_SUCCESS) {
2219                                 drbd_khelper(mdev, "pri-lost-after-sb");
2220                         } else {
2221                                 dev_warn(DEV, "Successfully gave up primary role.\n");
2222                                 rv = hg;
2223                         }
2224                 } else
2225                         rv = hg;
2226         }
2227
2228         return rv;
2229 }
2230
2231 static void drbd_uuid_dump(struct drbd_conf *mdev, char *text, u64 *uuid,
2232                            u64 bits, u64 flags)
2233 {
2234         if (!uuid) {
2235                 dev_info(DEV, "%s uuid info vanished while I was looking!\n", text);
2236                 return;
2237         }
2238         dev_info(DEV, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2239              text,
2240              (unsigned long long)uuid[UI_CURRENT],
2241              (unsigned long long)uuid[UI_BITMAP],
2242              (unsigned long long)uuid[UI_HISTORY_START],
2243              (unsigned long long)uuid[UI_HISTORY_END],
2244              (unsigned long long)bits,
2245              (unsigned long long)flags);
2246 }
2247
2248 /*
2249   100   after split brain try auto recover
2250     2   C_SYNC_SOURCE set BitMap
2251     1   C_SYNC_SOURCE use BitMap
2252     0   no Sync
2253    -1   C_SYNC_TARGET use BitMap
2254    -2   C_SYNC_TARGET set BitMap
2255  -100   after split brain, disconnect
2256 -1000   unrelated data
2257  */
2258 static int drbd_uuid_compare(struct drbd_conf *mdev, int *rule_nr) __must_hold(local)
2259 {
2260         u64 self, peer;
2261         int i, j;
2262
2263         self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2264         peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2265
2266         *rule_nr = 10;
2267         if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
2268                 return 0;
2269
2270         *rule_nr = 20;
2271         if ((self == UUID_JUST_CREATED || self == (u64)0) &&
2272              peer != UUID_JUST_CREATED)
2273                 return -2;
2274
2275         *rule_nr = 30;
2276         if (self != UUID_JUST_CREATED &&
2277             (peer == UUID_JUST_CREATED || peer == (u64)0))
2278                 return 2;
2279
2280         if (self == peer) {
2281                 int rct, dc; /* roles at crash time */
2282
2283                 if (mdev->p_uuid[UI_BITMAP] == (u64)0 && mdev->ldev->md.uuid[UI_BITMAP] != (u64)0) {
2284
2285                         if (mdev->agreed_pro_version < 91)
2286                                 return -1001;
2287
2288                         if ((mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
2289                             (mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
2290                                 dev_info(DEV, "was SyncSource, missed the resync finished event, corrected myself:\n");
2291                                 drbd_uuid_set_bm(mdev, 0UL);
2292
2293                                 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid,
2294                                                mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0);
2295                                 *rule_nr = 34;
2296                         } else {
2297                                 dev_info(DEV, "was SyncSource (peer failed to write sync_uuid)\n");
2298                                 *rule_nr = 36;
2299                         }
2300
2301                         return 1;
2302                 }
2303
2304                 if (mdev->ldev->md.uuid[UI_BITMAP] == (u64)0 && mdev->p_uuid[UI_BITMAP] != (u64)0) {
2305
2306                         if (mdev->agreed_pro_version < 91)
2307                                 return -1001;
2308
2309                         if ((mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_BITMAP] & ~((u64)1)) &&
2310                             (mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
2311                                 dev_info(DEV, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
2312
2313                                 mdev->p_uuid[UI_HISTORY_START + 1] = mdev->p_uuid[UI_HISTORY_START];
2314                                 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_BITMAP];
2315                                 mdev->p_uuid[UI_BITMAP] = 0UL;
2316
2317                                 drbd_uuid_dump(mdev, "peer", mdev->p_uuid, mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2318                                 *rule_nr = 35;
2319                         } else {
2320                                 dev_info(DEV, "was SyncTarget (failed to write sync_uuid)\n");
2321                                 *rule_nr = 37;
2322                         }
2323
2324                         return -1;
2325                 }
2326
2327                 /* Common power [off|failure] */
2328                 rct = (test_bit(CRASHED_PRIMARY, &mdev->flags) ? 1 : 0) +
2329                         (mdev->p_uuid[UI_FLAGS] & 2);
2330                 /* lowest bit is set when we were primary,
2331                  * next bit (weight 2) is set when peer was primary */
2332                 *rule_nr = 40;
2333
2334                 switch (rct) {
2335                 case 0: /* !self_pri && !peer_pri */ return 0;
2336                 case 1: /*  self_pri && !peer_pri */ return 1;
2337                 case 2: /* !self_pri &&  peer_pri */ return -1;
2338                 case 3: /*  self_pri &&  peer_pri */
2339                         dc = test_bit(DISCARD_CONCURRENT, &mdev->flags);
2340                         return dc ? -1 : 1;
2341                 }
2342         }
2343
2344         *rule_nr = 50;
2345         peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2346         if (self == peer)
2347                 return -1;
2348
2349         *rule_nr = 51;
2350         peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1);
2351         if (self == peer) {
2352                 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2353                 peer = mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1);
2354                 if (self == peer) {
2355                         /* The last P_SYNC_UUID did not get though. Undo the last start of
2356                            resync as sync source modifications of the peer's UUIDs. */
2357
2358                         if (mdev->agreed_pro_version < 91)
2359                                 return -1001;
2360
2361                         mdev->p_uuid[UI_BITMAP] = mdev->p_uuid[UI_HISTORY_START];
2362                         mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_HISTORY_START + 1];
2363                         return -1;
2364                 }
2365         }
2366
2367         *rule_nr = 60;
2368         self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2369         for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2370                 peer = mdev->p_uuid[i] & ~((u64)1);
2371                 if (self == peer)
2372                         return -2;
2373         }
2374
2375         *rule_nr = 70;
2376         self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2377         peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2378         if (self == peer)
2379                 return 1;
2380
2381         *rule_nr = 71;
2382         self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
2383         if (self == peer) {
2384                 self = mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1);
2385                 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1);
2386                 if (self == peer) {
2387                         /* The last P_SYNC_UUID did not get though. Undo the last start of
2388                            resync as sync source modifications of our UUIDs. */
2389
2390                         if (mdev->agreed_pro_version < 91)
2391                                 return -1001;
2392
2393                         _drbd_uuid_set(mdev, UI_BITMAP, mdev->ldev->md.uuid[UI_HISTORY_START]);
2394                         _drbd_uuid_set(mdev, UI_HISTORY_START, mdev->ldev->md.uuid[UI_HISTORY_START + 1]);
2395
2396                         dev_info(DEV, "Undid last start of resync:\n");
2397
2398                         drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid,
2399                                        mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0);
2400
2401                         return 1;
2402                 }
2403         }
2404
2405
2406         *rule_nr = 80;
2407         peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1);
2408         for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2409                 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2410                 if (self == peer)
2411                         return 2;
2412         }
2413
2414         *rule_nr = 90;
2415         self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
2416         peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1);
2417         if (self == peer && self != ((u64)0))
2418                 return 100;
2419
2420         *rule_nr = 100;
2421         for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
2422                 self = mdev->ldev->md.uuid[i] & ~((u64)1);
2423                 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
2424                         peer = mdev->p_uuid[j] & ~((u64)1);
2425                         if (self == peer)
2426                                 return -100;
2427                 }
2428         }
2429
2430         return -1000;
2431 }
2432
2433 /* drbd_sync_handshake() returns the new conn state on success, or
2434    CONN_MASK (-1) on failure.
2435  */
2436 static enum drbd_conns drbd_sync_handshake(struct drbd_conf *mdev, enum drbd_role peer_role,
2437                                            enum drbd_disk_state peer_disk) __must_hold(local)
2438 {
2439         int hg, rule_nr;
2440         enum drbd_conns rv = C_MASK;
2441         enum drbd_disk_state mydisk;
2442
2443         mydisk = mdev->state.disk;
2444         if (mydisk == D_NEGOTIATING)
2445                 mydisk = mdev->new_state_tmp.disk;
2446
2447         dev_info(DEV, "drbd_sync_handshake:\n");
2448         drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, mdev->comm_bm_set, 0);
2449         drbd_uuid_dump(mdev, "peer", mdev->p_uuid,
2450                        mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]);
2451
2452         hg = drbd_uuid_compare(mdev, &rule_nr);
2453
2454         dev_info(DEV, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
2455
2456         if (hg == -1000) {
2457                 dev_alert(DEV, "Unrelated data, aborting!\n");
2458                 return C_MASK;
2459         }
2460         if (hg == -1001) {
2461                 dev_alert(DEV, "To resolve this both sides have to support at least protocol\n");
2462                 return C_MASK;
2463         }
2464
2465         if    ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
2466             (peer_disk == D_INCONSISTENT && mydisk    > D_INCONSISTENT)) {
2467                 int f = (hg == -100) || abs(hg) == 2;
2468                 hg = mydisk > D_INCONSISTENT ? 1 : -1;
2469                 if (f)
2470                         hg = hg*2;
2471                 dev_info(DEV, "Becoming sync %s due to disk states.\n",
2472                      hg > 0 ? "source" : "target");
2473         }
2474
2475         if (hg == 100 || (hg == -100 && mdev->net_conf->always_asbp)) {
2476                 int pcount = (mdev->state.role == R_PRIMARY)
2477                            + (peer_role == R_PRIMARY);
2478                 int forced = (hg == -100);
2479
2480                 switch (pcount) {
2481                 case 0:
2482                         hg = drbd_asb_recover_0p(mdev);
2483                         break;
2484                 case 1:
2485                         hg = drbd_asb_recover_1p(mdev);
2486                         break;
2487                 case 2:
2488                         hg = drbd_asb_recover_2p(mdev);
2489                         break;
2490                 }
2491                 if (abs(hg) < 100) {
2492                         dev_warn(DEV, "Split-Brain detected, %d primaries, "
2493                              "automatically solved. Sync from %s node\n",
2494                              pcount, (hg < 0) ? "peer" : "this");
2495                         if (forced) {
2496                                 dev_warn(DEV, "Doing a full sync, since"
2497                                      " UUIDs where ambiguous.\n");
2498                                 hg = hg*2;
2499                         }
2500                 }
2501         }
2502
2503         if (hg == -100) {
2504                 if (mdev->net_conf->want_lose && !(mdev->p_uuid[UI_FLAGS]&1))
2505                         hg = -1;
2506                 if (!mdev->net_conf->want_lose && (mdev->p_uuid[UI_FLAGS]&1))
2507                         hg = 1;
2508
2509                 if (abs(hg) < 100)
2510                         dev_warn(DEV, "Split-Brain detected, manually solved. "
2511                              "Sync from %s node\n",
2512                              (hg < 0) ? "peer" : "this");
2513         }
2514
2515         if (hg == -100) {
2516                 dev_alert(DEV, "Split-Brain detected, dropping connection!\n");
2517                 drbd_khelper(mdev, "split-brain");
2518                 return C_MASK;
2519         }
2520
2521         if (hg > 0 && mydisk <= D_INCONSISTENT) {
2522                 dev_err(DEV, "I shall become SyncSource, but I am inconsistent!\n");
2523                 return C_MASK;
2524         }
2525
2526         if (hg < 0 && /* by intention we do not use mydisk here. */
2527             mdev->state.role == R_PRIMARY && mdev->state.disk >= D_CONSISTENT) {
2528                 switch (mdev->net_conf->rr_conflict) {
2529                 case ASB_CALL_HELPER:
2530                         drbd_khelper(mdev, "pri-lost");
2531                         /* fall through */
2532                 case ASB_DISCONNECT:
2533                         dev_err(DEV, "I shall become SyncTarget, but I am primary!\n");
2534                         return C_MASK;
2535                 case ASB_VIOLENTLY:
2536                         dev_warn(DEV, "Becoming SyncTarget, violating the stable-data"
2537                              "assumption\n");
2538                 }
2539         }
2540
2541         if (abs(hg) >= 2) {
2542                 dev_info(DEV, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
2543                 if (drbd_bitmap_io(mdev, &drbd_bmio_set_n_write, "set_n_write from sync_handshake"))
2544                         return C_MASK;
2545         }
2546
2547         if (hg > 0) { /* become sync source. */
2548                 rv = C_WF_BITMAP_S;
2549         } else if (hg < 0) { /* become sync target */
2550                 rv = C_WF_BITMAP_T;
2551         } else {
2552                 rv = C_CONNECTED;
2553                 if (drbd_bm_total_weight(mdev)) {
2554                         dev_info(DEV, "No resync, but %lu bits in bitmap!\n",
2555                              drbd_bm_total_weight(mdev));
2556                 }
2557         }
2558
2559         return rv;
2560 }
2561
2562 /* returns 1 if invalid */
2563 static int cmp_after_sb(enum drbd_after_sb_p peer, enum drbd_after_sb_p self)
2564 {
2565         /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
2566         if ((peer == ASB_DISCARD_REMOTE && self == ASB_DISCARD_LOCAL) ||
2567             (self == ASB_DISCARD_REMOTE && peer == ASB_DISCARD_LOCAL))
2568                 return 0;
2569
2570         /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
2571         if (peer == ASB_DISCARD_REMOTE || peer == ASB_DISCARD_LOCAL ||
2572             self == ASB_DISCARD_REMOTE || self == ASB_DISCARD_LOCAL)
2573                 return 1;
2574
2575         /* everything else is valid if they are equal on both sides. */
2576         if (peer == self)
2577                 return 0;
2578
2579         /* everything es is invalid. */
2580         return 1;
2581 }
2582
2583 static int receive_protocol(struct drbd_conf *mdev, struct p_header *h)
2584 {
2585         struct p_protocol *p = (struct p_protocol *)h;
2586         int header_size, data_size;
2587         int p_proto, p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
2588         int p_want_lose, p_two_primaries;
2589         char p_integrity_alg[SHARED_SECRET_MAX] = "";
2590
2591         header_size = sizeof(*p) - sizeof(*h);
2592         data_size   = h->length  - header_size;
2593
2594         if (drbd_recv(mdev, h->payload, header_size) != header_size)
2595                 return FALSE;
2596
2597         p_proto         = be32_to_cpu(p->protocol);
2598         p_after_sb_0p   = be32_to_cpu(p->after_sb_0p);
2599         p_after_sb_1p   = be32_to_cpu(p->after_sb_1p);
2600         p_after_sb_2p   = be32_to_cpu(p->after_sb_2p);
2601         p_want_lose     = be32_to_cpu(p->want_lose);
2602         p_two_primaries = be32_to_cpu(p->two_primaries);
2603
2604         if (p_proto != mdev->net_conf->wire_protocol) {
2605                 dev_err(DEV, "incompatible communication protocols\n");
2606                 goto disconnect;
2607         }
2608
2609         if (cmp_after_sb(p_after_sb_0p, mdev->net_conf->after_sb_0p)) {
2610                 dev_err(DEV, "incompatible after-sb-0pri settings\n");
2611                 goto disconnect;
2612         }
2613
2614         if (cmp_after_sb(p_after_sb_1p, mdev->net_conf->after_sb_1p)) {
2615                 dev_err(DEV, "incompatible after-sb-1pri settings\n");
2616                 goto disconnect;
2617         }
2618
2619         if (cmp_after_sb(p_after_sb_2p, mdev->net_conf->after_sb_2p)) {
2620                 dev_err(DEV, "incompatible after-sb-2pri settings\n");
2621                 goto disconnect;
2622         }
2623
2624         if (p_want_lose && mdev->net_conf->want_lose) {
2625                 dev_err(DEV, "both sides have the 'want_lose' flag set\n");
2626                 goto disconnect;
2627         }
2628
2629         if (p_two_primaries != mdev->net_conf->two_primaries) {
2630                 dev_err(DEV, "incompatible setting of the two-primaries options\n");
2631                 goto disconnect;
2632         }
2633
2634         if (mdev->agreed_pro_version >= 87) {
2635                 unsigned char *my_alg = mdev->net_conf->integrity_alg;
2636
2637                 if (drbd_recv(mdev, p_integrity_alg, data_size) != data_size)
2638                         return FALSE;
2639
2640                 p_integrity_alg[SHARED_SECRET_MAX-1] = 0;
2641                 if (strcmp(p_integrity_alg, my_alg)) {
2642                         dev_err(DEV, "incompatible setting of the data-integrity-alg\n");
2643                         goto disconnect;
2644                 }
2645                 dev_info(DEV, "data-integrity-alg: %s\n",
2646                      my_alg[0] ? my_alg : (unsigned char *)"<not-used>");
2647         }
2648
2649         return TRUE;
2650
2651 disconnect:
2652         drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2653         return FALSE;
2654 }
2655
2656 /* helper function
2657  * input: alg name, feature name
2658  * return: NULL (alg name was "")
2659  *         ERR_PTR(error) if something goes wrong
2660  *         or the crypto hash ptr, if it worked out ok. */
2661 struct crypto_hash *drbd_crypto_alloc_digest_safe(const struct drbd_conf *mdev,
2662                 const char *alg, const char *name)
2663 {
2664         struct crypto_hash *tfm;
2665
2666         if (!alg[0])
2667                 return NULL;
2668
2669         tfm = crypto_alloc_hash(alg, 0, CRYPTO_ALG_ASYNC);
2670         if (IS_ERR(tfm)) {
2671                 dev_err(DEV, "Can not allocate \"%s\" as %s (reason: %ld)\n",
2672                         alg, name, PTR_ERR(tfm));
2673                 return tfm;
2674         }
2675         if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm))) {
2676                 crypto_free_hash(tfm);
2677                 dev_err(DEV, "\"%s\" is not a digest (%s)\n", alg, name);
2678                 return ERR_PTR(-EINVAL);
2679         }
2680         return tfm;
2681 }
2682
2683 static int receive_SyncParam(struct drbd_conf *mdev, struct p_header *h)
2684 {
2685         int ok = TRUE;
2686         struct p_rs_param_89 *p = (struct p_rs_param_89 *)h;
2687         unsigned int header_size, data_size, exp_max_sz;
2688         struct crypto_hash *verify_tfm = NULL;
2689         struct crypto_hash *csums_tfm = NULL;
2690         const int apv = mdev->agreed_pro_version;
2691
2692         exp_max_sz  = apv <= 87 ? sizeof(struct p_rs_param)
2693                     : apv == 88 ? sizeof(struct p_rs_param)
2694                                         + SHARED_SECRET_MAX
2695                     : /* 89 */    sizeof(struct p_rs_param_89);
2696
2697         if (h->length > exp_max_sz) {
2698                 dev_err(DEV, "SyncParam packet too long: received %u, expected <= %u bytes\n",
2699                     h->length, exp_max_sz);
2700                 return FALSE;
2701         }
2702
2703         if (apv <= 88) {
2704                 header_size = sizeof(struct p_rs_param) - sizeof(*h);
2705                 data_size   = h->length  - header_size;
2706         } else /* apv >= 89 */ {
2707                 header_size = sizeof(struct p_rs_param_89) - sizeof(*h);
2708                 data_size   = h->length  - header_size;
2709                 D_ASSERT(data_size == 0);
2710         }
2711
2712         /* initialize verify_alg and csums_alg */
2713         memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
2714
2715         if (drbd_recv(mdev, h->payload, header_size) != header_size)
2716                 return FALSE;
2717
2718         mdev->sync_conf.rate      = be32_to_cpu(p->rate);
2719
2720         if (apv >= 88) {
2721                 if (apv == 88) {
2722                         if (data_size > SHARED_SECRET_MAX) {
2723                                 dev_err(DEV, "verify-alg too long, "
2724                                     "peer wants %u, accepting only %u byte\n",
2725                                                 data_size, SHARED_SECRET_MAX);
2726                                 return FALSE;
2727                         }
2728
2729                         if (drbd_recv(mdev, p->verify_alg, data_size) != data_size)
2730                                 return FALSE;
2731
2732                         /* we expect NUL terminated string */
2733                         /* but just in case someone tries to be evil */
2734                         D_ASSERT(p->verify_alg[data_size-1] == 0);
2735                         p->verify_alg[data_size-1] = 0;
2736
2737                 } else /* apv >= 89 */ {
2738                         /* we still expect NUL terminated strings */
2739                         /* but just in case someone tries to be evil */
2740                         D_ASSERT(p->verify_alg[SHARED_SECRET_MAX-1] == 0);
2741                         D_ASSERT(p->csums_alg[SHARED_SECRET_MAX-1] == 0);
2742                         p->verify_alg[SHARED_SECRET_MAX-1] = 0;
2743                         p->csums_alg[SHARED_SECRET_MAX-1] = 0;
2744                 }
2745
2746                 if (strcmp(mdev->sync_conf.verify_alg, p->verify_alg)) {
2747                         if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2748                                 dev_err(DEV, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
2749                                     mdev->sync_conf.verify_alg, p->verify_alg);
2750                                 goto disconnect;
2751                         }
2752                         verify_tfm = drbd_crypto_alloc_digest_safe(mdev,
2753                                         p->verify_alg, "verify-alg");
2754                         if (IS_ERR(verify_tfm)) {
2755                                 verify_tfm = NULL;
2756                                 goto disconnect;
2757                         }
2758                 }
2759
2760                 if (apv >= 89 && strcmp(mdev->sync_conf.csums_alg, p->csums_alg)) {
2761                         if (mdev->state.conn == C_WF_REPORT_PARAMS) {
2762                                 dev_err(DEV, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
2763                                     mdev->sync_conf.csums_alg, p->csums_alg);
2764                                 goto disconnect;
2765                         }
2766                         csums_tfm = drbd_crypto_alloc_digest_safe(mdev,
2767                                         p->csums_alg, "csums-alg");
2768                         if (IS_ERR(csums_tfm)) {
2769                                 csums_tfm = NULL;
2770                                 goto disconnect;
2771                         }
2772                 }
2773
2774
2775                 spin_lock(&mdev->peer_seq_lock);
2776                 /* lock against drbd_nl_syncer_conf() */
2777                 if (verify_tfm) {
2778                         strcpy(mdev->sync_conf.verify_alg, p->verify_alg);
2779                         mdev->sync_conf.verify_alg_len = strlen(p->verify_alg) + 1;
2780                         crypto_free_hash(mdev->verify_tfm);
2781                         mdev->verify_tfm = verify_tfm;
2782                         dev_info(DEV, "using verify-alg: \"%s\"\n", p->verify_alg);
2783                 }
2784                 if (csums_tfm) {
2785                         strcpy(mdev->sync_conf.csums_alg, p->csums_alg);
2786                         mdev->sync_conf.csums_alg_len = strlen(p->csums_alg) + 1;
2787                         crypto_free_hash(mdev->csums_tfm);
2788                         mdev->csums_tfm = csums_tfm;
2789                         dev_info(DEV, "using csums-alg: \"%s\"\n", p->csums_alg);
2790                 }
2791                 spin_unlock(&mdev->peer_seq_lock);
2792         }
2793
2794         return ok;
2795 disconnect:
2796         /* just for completeness: actually not needed,
2797          * as this is not reached if csums_tfm was ok. */
2798         crypto_free_hash(csums_tfm);
2799         /* but free the verify_tfm again, if csums_tfm did not work out */
2800         crypto_free_hash(verify_tfm);
2801         drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2802         return FALSE;
2803 }
2804
2805 static void drbd_setup_order_type(struct drbd_conf *mdev, int peer)
2806 {
2807         /* sorry, we currently have no working implementation
2808          * of distributed TCQ */
2809 }
2810
2811 /* warn if the arguments differ by more than 12.5% */
2812 static void warn_if_differ_considerably(struct drbd_conf *mdev,
2813         const char *s, sector_t a, sector_t b)
2814 {
2815         sector_t d;
2816         if (a == 0 || b == 0)
2817                 return;
2818         d = (a > b) ? (a - b) : (b - a);
2819         if (d > (a>>3) || d > (b>>3))
2820                 dev_warn(DEV, "Considerable difference in %s: %llus vs. %llus\n", s,
2821                      (unsigned long long)a, (unsigned long long)b);
2822 }
2823
2824 static int receive_sizes(struct drbd_conf *mdev, struct p_header *h)
2825 {
2826         struct p_sizes *p = (struct p_sizes *)h;
2827         enum determine_dev_size dd = unchanged;
2828         unsigned int max_seg_s;
2829         sector_t p_size, p_usize, my_usize;
2830         int ldsc = 0; /* local disk size changed */
2831         enum drbd_conns nconn;
2832
2833         ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
2834         if (drbd_recv(mdev, h->payload, h->length) != h->length)
2835                 return FALSE;
2836
2837         p_size = be64_to_cpu(p->d_size);
2838         p_usize = be64_to_cpu(p->u_size);
2839
2840         if (p_size == 0 && mdev->state.disk == D_DISKLESS) {
2841                 dev_err(DEV, "some backing storage is needed\n");
2842                 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2843                 return FALSE;
2844         }
2845
2846         /* just store the peer's disk size for now.
2847          * we still need to figure out whether we accept that. */
2848         mdev->p_size = p_size;
2849
2850 #define min_not_zero(l, r) (l == 0) ? r : ((r == 0) ? l : min(l, r))
2851         if (get_ldev(mdev)) {
2852                 warn_if_differ_considerably(mdev, "lower level device sizes",
2853                            p_size, drbd_get_max_capacity(mdev->ldev));
2854                 warn_if_differ_considerably(mdev, "user requested size",
2855                                             p_usize, mdev->ldev->dc.disk_size);
2856
2857                 /* if this is the first connect, or an otherwise expected
2858                  * param exchange, choose the minimum */
2859                 if (mdev->state.conn == C_WF_REPORT_PARAMS)
2860                         p_usize = min_not_zero((sector_t)mdev->ldev->dc.disk_size,
2861                                              p_usize);
2862
2863                 my_usize = mdev->ldev->dc.disk_size;
2864
2865                 if (mdev->ldev->dc.disk_size != p_usize) {
2866                         mdev->ldev->dc.disk_size = p_usize;
2867                         dev_info(DEV, "Peer sets u_size to %lu sectors\n",
2868                              (unsigned long)mdev->ldev->dc.disk_size);
2869                 }
2870
2871                 /* Never shrink a device with usable data during connect.
2872                    But allow online shrinking if we are connected. */
2873                 if (drbd_new_dev_size(mdev, mdev->ldev, 0) <
2874                    drbd_get_capacity(mdev->this_bdev) &&
2875                    mdev->state.disk >= D_OUTDATED &&
2876                    mdev->state.conn < C_CONNECTED) {
2877                         dev_err(DEV, "The peer's disk size is too small!\n");
2878                         drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2879                         mdev->ldev->dc.disk_size = my_usize;
2880                         put_ldev(mdev);
2881                         return FALSE;
2882                 }
2883                 put_ldev(mdev);
2884         }
2885 #undef min_not_zero
2886
2887         if (get_ldev(mdev)) {
2888           dd = drbd_determin_dev_size(mdev, 0);
2889                 put_ldev(mdev);
2890                 if (dd == dev_size_error)
2891                         return FALSE;
2892                 drbd_md_sync(mdev);
2893         } else {
2894                 /* I am diskless, need to accept the peer's size. */
2895                 drbd_set_my_capacity(mdev, p_size);
2896         }
2897
2898         if (mdev->p_uuid && mdev->state.conn <= C_CONNECTED && get_ldev(mdev)) {
2899                 nconn = drbd_sync_handshake(mdev,
2900                                 mdev->state.peer, mdev->state.pdsk);
2901                 put_ldev(mdev);
2902
2903                 if (nconn == C_MASK) {
2904                         drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2905                         return FALSE;
2906                 }
2907
2908                 if (drbd_request_state(mdev, NS(conn, nconn)) < SS_SUCCESS) {
2909                         drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2910                         return FALSE;
2911                 }
2912         }
2913
2914         if (get_ldev(mdev)) {
2915                 if (mdev->ldev->known_size != drbd_get_capacity(mdev->ldev->backing_bdev)) {
2916                         mdev->ldev->known_size = drbd_get_capacity(mdev->ldev->backing_bdev);
2917                         ldsc = 1;
2918                 }
2919
2920                 max_seg_s = be32_to_cpu(p->max_segment_size);
2921                 if (max_seg_s != queue_max_segment_size(mdev->rq_queue))
2922                         drbd_setup_queue_param(mdev, max_seg_s);
2923
2924                 drbd_setup_order_type(mdev, be32_to_cpu(p->queue_order_type));
2925                 put_ldev(mdev);
2926         }
2927
2928         if (mdev->state.conn > C_WF_REPORT_PARAMS) {
2929                 if (be64_to_cpu(p->c_size) !=
2930                     drbd_get_capacity(mdev->this_bdev) || ldsc) {
2931                         /* we have different sizes, probably peer
2932                          * needs to know my new size... */
2933                         drbd_send_sizes(mdev, 0);
2934                 }
2935                 if (test_and_clear_bit(RESIZE_PENDING, &mdev->flags) ||
2936                     (dd == grew && mdev->state.conn == C_CONNECTED)) {
2937                         if (mdev->state.pdsk >= D_INCONSISTENT &&
2938                             mdev->state.disk >= D_INCONSISTENT)
2939                                 resync_after_online_grow(mdev);
2940                         else
2941                                 set_bit(RESYNC_AFTER_NEG, &mdev->flags);
2942                 }
2943         }
2944
2945         return TRUE;
2946 }
2947
2948 static int receive_uuids(struct drbd_conf *mdev, struct p_header *h)
2949 {
2950         struct p_uuids *p = (struct p_uuids *)h;
2951         u64 *p_uuid;
2952         int i;
2953
2954         ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE;
2955         if (drbd_recv(mdev, h->payload, h->length) != h->length)
2956                 return FALSE;
2957
2958         p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
2959
2960         for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
2961                 p_uuid[i] = be64_to_cpu(p->uuid[i]);
2962
2963         kfree(mdev->p_uuid);
2964         mdev->p_uuid = p_uuid;
2965
2966         if (mdev->state.conn < C_CONNECTED &&
2967             mdev->state.disk < D_INCONSISTENT &&
2968             mdev->state.role == R_PRIMARY &&
2969             (mdev->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
2970                 dev_err(DEV, "Can only connect to data with current UUID=%016llX\n",
2971                     (unsigned long long)mdev->ed_uuid);
2972                 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
2973                 return FALSE;
2974         }
2975
2976         if (get_ldev(mdev)) {
2977                 int skip_initial_sync =
2978                         mdev->state.conn == C_CONNECTED &&
2979                         mdev->agreed_pro_version >= 90 &&
2980                         mdev->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
2981                         (p_uuid[UI_FLAGS] & 8);
2982                 if (skip_initial_sync) {
2983                         dev_info(DEV, "Accepted new current UUID, preparing to skip initial sync\n");
2984                         drbd_bitmap_io(mdev, &drbd_bmio_clear_n_write,
2985                                         "clea