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