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