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_protocol.h"
  48 #include "drbd_req.h"
  49 #include "drbd_vli.h"
  50
  51 #define PRO_FEATURES (FF_TRIM)
  52
  53 struct packet_info {
  54         enum drbd_packet cmd;
  55         unsigned int size;
  56         unsigned int vnr;
  57         void *data;
  58 };
  59
  60 enum finish_epoch {
  61         FE_STILL_LIVE,
  62         FE_DESTROYED,
  63         FE_RECYCLED,
  64 };
  65
  66 static int drbd_do_features(struct drbd_connection *connection);
  67 static int drbd_do_auth(struct drbd_connection *connection);
  68 static int drbd_disconnected(struct drbd_peer_device *);
  69 static void conn_wait_active_ee_empty(struct drbd_connection *connection);
  70 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
  71 static int e_end_block(struct drbd_work *, int);
  72
  73
  74 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
  75
  76 /*
  77  * some helper functions to deal with single linked page lists,
  78  * page->private being our "next" pointer.
  79  */
  80
  81 /* If at least n pages are linked at head, get n pages off.
  82  * Otherwise, don't modify head, and return NULL.
  83  * Locking is the responsibility of the caller.
  84  */
  85 static struct page *page_chain_del(struct page **head, int n)
  86 {
  87         struct page *page;
  88         struct page *tmp;
  89
  90         BUG_ON(!n);
  91         BUG_ON(!head);
  92
  93         page = *head;
  94
  95         if (!page)
  96                 return NULL;
  97
  98         while (page) {
  99                 tmp = page_chain_next(page);
 100                 if (--n == 0)
 101                         break; /* found sufficient pages */
 102                 if (tmp == NULL)
 103                         /* insufficient pages, don't use any of them. */
 104                         return NULL;
 105                 page = tmp;
 106         }
 107
 108         /* add end of list marker for the returned list */
 109         set_page_private(page, 0);
 110         /* actual return value, and adjustment of head */
 111         page = *head;
 112         *head = tmp;
 113         return page;
 114 }
 115
 116 /* may be used outside of locks to find the tail of a (usually short)
 117  * "private" page chain, before adding it back to a global chain head
 118  * with page_chain_add() under a spinlock. */
 119 static struct page *page_chain_tail(struct page *page, int *len)
 120 {
 121         struct page *tmp;
 122         int i = 1;
 123         while ((tmp = page_chain_next(page)))
 124                 ++i, page = tmp;
 125         if (len)
 126                 *len = i;
 127         return page;
 128 }
 129
 130 static int page_chain_free(struct page *page)
 131 {
 132         struct page *tmp;
 133         int i = 0;
 134         page_chain_for_each_safe(page, tmp) {
 135                 put_page(page);
 136                 ++i;
 137         }
 138         return i;
 139 }
 140
 141 static void page_chain_add(struct page **head,
 142                 struct page *chain_first, struct page *chain_last)
 143 {
 144 #if 1
 145         struct page *tmp;
 146         tmp = page_chain_tail(chain_first, NULL);
 147         BUG_ON(tmp != chain_last);
 148 #endif
 149
 150         /* add chain to head */
 151         set_page_private(chain_last, (unsigned long)*head);
 152         *head = chain_first;
 153 }
 154
 155 static struct page *__drbd_alloc_pages(struct drbd_device *device,
 156                                        unsigned int number)
 157 {
 158         struct page *page = NULL;
 159         struct page *tmp = NULL;
 160         unsigned int i = 0;
 161
 162         /* Yes, testing drbd_pp_vacant outside the lock is racy.
 163          * So what. It saves a spin_lock. */
 164         if (drbd_pp_vacant >= number) {
 165                 spin_lock(&drbd_pp_lock);
 166                 page = page_chain_del(&drbd_pp_pool, number);
 167                 if (page)
 168                         drbd_pp_vacant -= number;
 169                 spin_unlock(&drbd_pp_lock);
 170                 if (page)
 171                         return page;
 172         }
 173
 174         /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
 175          * "criss-cross" setup, that might cause write-out on some other DRBD,
 176          * which in turn might block on the other node at this very place.  */
 177         for (i = 0; i < number; i++) {
 178                 tmp = alloc_page(GFP_TRY);
 179                 if (!tmp)
 180                         break;
 181                 set_page_private(tmp, (unsigned long)page);
 182                 page = tmp;
 183         }
 184
 185         if (i == number)
 186                 return page;
 187
 188         /* Not enough pages immediately available this time.
 189          * No need to jump around here, drbd_alloc_pages will retry this
 190          * function "soon". */
 191         if (page) {
 192                 tmp = page_chain_tail(page, NULL);
 193                 spin_lock(&drbd_pp_lock);
 194                 page_chain_add(&drbd_pp_pool, page, tmp);
 195                 drbd_pp_vacant += i;
 196                 spin_unlock(&drbd_pp_lock);
 197         }
 198         return NULL;
 199 }
 200
 201 static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
 202                                            struct list_head *to_be_freed)
 203 {
 204         struct drbd_peer_request *peer_req, *tmp;
 205
 206         /* The EEs are always appended to the end of the list. Since
 207            they are sent in order over the wire, they have to finish
 208            in order. As soon as we see the first not finished we can
 209            stop to examine the list... */
 210
 211         list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
 212                 if (drbd_peer_req_has_active_page(peer_req))
 213                         break;
 214                 list_move(&peer_req->w.list, to_be_freed);
 215         }
 216 }
 217
 218 static void drbd_reclaim_net_peer_reqs(struct drbd_device *device)
 219 {
 220         LIST_HEAD(reclaimed);
 221         struct drbd_peer_request *peer_req, *t;
 222
 223         spin_lock_irq(&device->resource->req_lock);
 224         reclaim_finished_net_peer_reqs(device, &reclaimed);
 225         spin_unlock_irq(&device->resource->req_lock);
 226         list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
 227                 drbd_free_net_peer_req(device, peer_req);
 228 }
 229
 230 static void conn_reclaim_net_peer_reqs(struct drbd_connection *connection)
 231 {
 232         struct drbd_peer_device *peer_device;
 233         int vnr;
 234
 235         rcu_read_lock();
 236         idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
 237                 struct drbd_device *device = peer_device->device;
 238                 if (!atomic_read(&device->pp_in_use_by_net))
 239                         continue;
 240
 241                 kref_get(&device->kref);
 242                 rcu_read_unlock();
 243                 drbd_reclaim_net_peer_reqs(device);
 244                 kref_put(&device->kref, drbd_destroy_device);
 245                 rcu_read_lock();
 246         }
 247         rcu_read_unlock();
 248 }
 249
 250 /**
 251  * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
 252  * @device:     DRBD device.
 253  * @number:     number of pages requested
 254  * @retry:      whether to retry, if not enough pages are available right now
 255  *
 256  * Tries to allocate number pages, first from our own page pool, then from
 257  * the kernel.
 258  * Possibly retry until DRBD frees sufficient pages somewhere else.
 259  *
 260  * If this allocation would exceed the max_buffers setting, we throttle
 261  * allocation (schedule_timeout) to give the system some room to breathe.
 262  *
 263  * We do not use max-buffers as hard limit, because it could lead to
 264  * congestion and further to a distributed deadlock during online-verify or
 265  * (checksum based) resync, if the max-buffers, socket buffer sizes and
 266  * resync-rate settings are mis-configured.
 267  *
 268  * Returns a page chain linked via page->private.
 269  */
 270 struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
 271                               bool retry)
 272 {
 273         struct drbd_device *device = peer_device->device;
 274         struct page *page = NULL;
 275         struct net_conf *nc;
 276         DEFINE_WAIT(wait);
 277         unsigned int mxb;
 278
 279         rcu_read_lock();
 280         nc = rcu_dereference(peer_device->connection->net_conf);
 281         mxb = nc ? nc->max_buffers : 1000000;
 282         rcu_read_unlock();
 283
 284         if (atomic_read(&device->pp_in_use) < mxb)
 285                 page = __drbd_alloc_pages(device, number);
 286
 287         /* Try to keep the fast path fast, but occasionally we need
 288          * to reclaim the pages we lended to the network stack. */
 289         if (page && atomic_read(&device->pp_in_use_by_net) > 512)
 290                 drbd_reclaim_net_peer_reqs(device);
 291
 292         while (page == NULL) {
 293                 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);
 294
 295                 drbd_reclaim_net_peer_reqs(device);
 296
 297                 if (atomic_read(&device->pp_in_use) < mxb) {
 298                         page = __drbd_alloc_pages(device, number);
 299                         if (page)
 300                                 break;
 301                 }
 302
 303                 if (!retry)
 304                         break;
 305
 306                 if (signal_pending(current)) {
 307                         drbd_warn(device, "drbd_alloc_pages interrupted!\n");
 308                         break;
 309                 }
 310
 311                 if (schedule_timeout(HZ/10) == 0)
 312                         mxb = UINT_MAX;
 313         }
 314         finish_wait(&drbd_pp_wait, &wait);
 315
 316         if (page)
 317                 atomic_add(number, &device->pp_in_use);
 318         return page;
 319 }
 320
 321 /* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
 322  * Is also used from inside an other spin_lock_irq(&resource->req_lock);
 323  * Either links the page chain back to the global pool,
 324  * or returns all pages to the system. */
 325 static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
 326 {
 327         atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
 328         int i;
 329
 330         if (page == NULL)
 331                 return;
 332
 333         if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * minor_count)
 334                 i = page_chain_free(page);
 335         else {
 336                 struct page *tmp;
 337                 tmp = page_chain_tail(page, &i);
 338                 spin_lock(&drbd_pp_lock);
 339                 page_chain_add(&drbd_pp_pool, page, tmp);
 340                 drbd_pp_vacant += i;
 341                 spin_unlock(&drbd_pp_lock);
 342         }
 343         i = atomic_sub_return(i, a);
 344         if (i < 0)
 345                 drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
 346                         is_net ? "pp_in_use_by_net" : "pp_in_use", i);
 347         wake_up(&drbd_pp_wait);
 348 }
 349
 350 /*
 351 You need to hold the req_lock:
 352  _drbd_wait_ee_list_empty()
 353
 354 You must not have the req_lock:
 355  drbd_free_peer_req()
 356  drbd_alloc_peer_req()
 357  drbd_free_peer_reqs()
 358  drbd_ee_fix_bhs()
 359  drbd_finish_peer_reqs()
 360  drbd_clear_done_ee()
 361  drbd_wait_ee_list_empty()
 362 */
 363
 364 struct drbd_peer_request *
 365 drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
 366                     unsigned int data_size, bool has_payload, gfp_t gfp_mask) __must_hold(local)
 367 {
 368         struct drbd_device *device = peer_device->device;
 369         struct drbd_peer_request *peer_req;
 370         struct page *page = NULL;
 371         unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
 372
 373         if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
 374                 return NULL;
 375
 376         peer_req = mempool_alloc(drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
 377         if (!peer_req) {
 378                 if (!(gfp_mask & __GFP_NOWARN))
 379                         drbd_err(device, "%s: allocation failed\n", __func__);
 380                 return NULL;
 381         }
 382
 383         if (has_payload && data_size) {
 384                 page = drbd_alloc_pages(peer_device, nr_pages,
 385                                         gfpflags_allow_blocking(gfp_mask));
 386                 if (!page)
 387                         goto fail;
 388         }
 389
 390         memset(peer_req, 0, sizeof(*peer_req));
 391         INIT_LIST_HEAD(&peer_req->w.list);
 392         drbd_clear_interval(&peer_req->i);
 393         peer_req->i.size = data_size;
 394         peer_req->i.sector = sector;
 395         peer_req->submit_jif = jiffies;
 396         peer_req->peer_device = peer_device;
 397         peer_req->pages = page;
 398         /*
 399          * The block_id is opaque to the receiver.  It is not endianness
 400          * converted, and sent back to the sender unchanged.
 401          */
 402         peer_req->block_id = id;
 403
 404         return peer_req;
 405
 406  fail:
 407         mempool_free(peer_req, drbd_ee_mempool);
 408         return NULL;
 409 }
 410
 411 void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
 412                        int is_net)
 413 {
 414         might_sleep();
 415         if (peer_req->flags & EE_HAS_DIGEST)
 416                 kfree(peer_req->digest);
 417         drbd_free_pages(device, peer_req->pages, is_net);
 418         D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
 419         D_ASSERT(device, drbd_interval_empty(&peer_req->i));
 420         if (!expect(!(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
 421                 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
 422                 drbd_al_complete_io(device, &peer_req->i);
 423         }
 424         mempool_free(peer_req, drbd_ee_mempool);
 425 }
 426
 427 int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
 428 {
 429         LIST_HEAD(work_list);
 430         struct drbd_peer_request *peer_req, *t;
 431         int count = 0;
 432         int is_net = list == &device->net_ee;
 433
 434         spin_lock_irq(&device->resource->req_lock);
 435         list_splice_init(list, &work_list);
 436         spin_unlock_irq(&device->resource->req_lock);
 437
 438         list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
 439                 __drbd_free_peer_req(device, peer_req, is_net);
 440                 count++;
 441         }
 442         return count;
 443 }
 444
 445 /*
 446  * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
 447  */
 448 static int drbd_finish_peer_reqs(struct drbd_device *device)
 449 {
 450         LIST_HEAD(work_list);
 451         LIST_HEAD(reclaimed);
 452         struct drbd_peer_request *peer_req, *t;
 453         int err = 0;
 454
 455         spin_lock_irq(&device->resource->req_lock);
 456         reclaim_finished_net_peer_reqs(device, &reclaimed);
 457         list_splice_init(&device->done_ee, &work_list);
 458         spin_unlock_irq(&device->resource->req_lock);
 459
 460         list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
 461                 drbd_free_net_peer_req(device, peer_req);
 462
 463         /* possible callbacks here:
 464          * e_end_block, and e_end_resync_block, e_send_superseded.
 465          * all ignore the last argument.
 466          */
 467         list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
 468                 int err2;
 469
 470                 /* list_del not necessary, next/prev members not touched */
 471                 err2 = peer_req->w.cb(&peer_req->w, !!err);
 472                 if (!err)
 473                         err = err2;
 474                 drbd_free_peer_req(device, peer_req);
 475         }
 476         wake_up(&device->ee_wait);
 477
 478         return err;
 479 }
 480
 481 static void _drbd_wait_ee_list_empty(struct drbd_device *device,
 482                                      struct list_head *head)
 483 {
 484         DEFINE_WAIT(wait);
 485
 486         /* avoids spin_lock/unlock
 487          * and calling prepare_to_wait in the fast path */
 488         while (!list_empty(head)) {
 489                 prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
 490                 spin_unlock_irq(&device->resource->req_lock);
 491                 io_schedule();
 492                 finish_wait(&device->ee_wait, &wait);
 493                 spin_lock_irq(&device->resource->req_lock);
 494         }
 495 }
 496
 497 static void drbd_wait_ee_list_empty(struct drbd_device *device,
 498                                     struct list_head *head)
 499 {
 500         spin_lock_irq(&device->resource->req_lock);
 501         _drbd_wait_ee_list_empty(device, head);
 502         spin_unlock_irq(&device->resource->req_lock);
 503 }
 504
 505 static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
 506 {
 507         struct kvec iov = {
 508                 .iov_base = buf,
 509                 .iov_len = size,
 510         };
 511         struct msghdr msg = {
 512                 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
 513         };
 514         return kernel_recvmsg(sock, &msg, &iov, 1, size, msg.msg_flags);
 515 }
 516
 517 static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
 518 {
 519         int rv;
 520
 521         rv = drbd_recv_short(connection->data.socket, buf, size, 0);
 522
 523         if (rv < 0) {
 524                 if (rv == -ECONNRESET)
 525                         drbd_info(connection, "sock was reset by peer\n");
 526                 else if (rv != -ERESTARTSYS)
 527                         drbd_err(connection, "sock_recvmsg returned %d\n", rv);
 528         } else if (rv == 0) {
 529                 if (test_bit(DISCONNECT_SENT, &connection->flags)) {
 530                         long t;
 531                         rcu_read_lock();
 532                         t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
 533                         rcu_read_unlock();
 534
 535                         t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);
 536
 537                         if (t)
 538                                 goto out;
 539                 }
 540                 drbd_info(connection, "sock was shut down by peer\n");
 541         }
 542
 543         if (rv != size)
 544                 conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);
 545
 546 out:
 547         return rv;
 548 }
 549
 550 static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
 551 {
 552         int err;
 553
 554         err = drbd_recv(connection, buf, size);
 555         if (err != size) {
 556                 if (err >= 0)
 557                         err = -EIO;
 558         } else
 559                 err = 0;
 560         return err;
 561 }
 562
 563 static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
 564 {
 565         int err;
 566
 567         err = drbd_recv_all(connection, buf, size);
 568         if (err && !signal_pending(current))
 569                 drbd_warn(connection, "short read (expected size %d)\n", (int)size);
 570         return err;
 571 }
 572
 573 /* quoting tcp(7):
 574  *   On individual connections, the socket buffer size must be set prior to the
 575  *   listen(2) or connect(2) calls in order to have it take effect.
 576  * This is our wrapper to do so.
 577  */
 578 static void drbd_setbufsize(struct socket *sock, unsigned int snd,
 579                 unsigned int rcv)
 580 {
 581         /* open coded SO_SNDBUF, SO_RCVBUF */
 582         if (snd) {
 583                 sock->sk->sk_sndbuf = snd;
 584                 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
 585         }
 586         if (rcv) {
 587                 sock->sk->sk_rcvbuf = rcv;
 588                 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
 589         }
 590 }
 591
 592 static struct socket *drbd_try_connect(struct drbd_connection *connection)
 593 {
 594         const char *what;
 595         struct socket *sock;
 596         struct sockaddr_in6 src_in6;
 597         struct sockaddr_in6 peer_in6;
 598         struct net_conf *nc;
 599         int err, peer_addr_len, my_addr_len;
 600         int sndbuf_size, rcvbuf_size, connect_int;
 601         int disconnect_on_error = 1;
 602
 603         rcu_read_lock();
 604         nc = rcu_dereference(connection->net_conf);
 605         if (!nc) {
 606                 rcu_read_unlock();
 607                 return NULL;
 608         }
 609         sndbuf_size = nc->sndbuf_size;
 610         rcvbuf_size = nc->rcvbuf_size;
 611         connect_int = nc->connect_int;
 612         rcu_read_unlock();
 613
 614         my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
 615         memcpy(&src_in6, &connection->my_addr, my_addr_len);
 616
 617         if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
 618                 src_in6.sin6_port = 0;
 619         else
 620                 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */
 621
 622         peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
 623         memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);
 624
 625         what = "sock_create_kern";
 626         err = sock_create_kern(&init_net, ((struct sockaddr *)&src_in6)->sa_family,
 627                                SOCK_STREAM, IPPROTO_TCP, &sock);
 628         if (err < 0) {
 629                 sock = NULL;
 630                 goto out;
 631         }
 632
 633         sock->sk->sk_rcvtimeo =
 634         sock->sk->sk_sndtimeo = connect_int * HZ;
 635         drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);
 636
 637        /* explicitly bind to the configured IP as source IP
 638         *  for the outgoing connections.
 639         *  This is needed for multihomed hosts and to be
 640         *  able to use lo: interfaces for drbd.
 641         * Make sure to use 0 as port number, so linux selects
 642         *  a free one dynamically.
 643         */
 644         what = "bind before connect";
 645         err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
 646         if (err < 0)
 647                 goto out;
 648
 649         /* connect may fail, peer not yet available.
 650          * stay C_WF_CONNECTION, don't go Disconnecting! */
 651         disconnect_on_error = 0;
 652         what = "connect";
 653         err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);
 654
 655 out:
 656         if (err < 0) {
 657                 if (sock) {
 658                         sock_release(sock);
 659                         sock = NULL;
 660                 }
 661                 switch (-err) {
 662                         /* timeout, busy, signal pending */
 663                 case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
 664                 case EINTR: case ERESTARTSYS:
 665                         /* peer not (yet) available, network problem */
 666                 case ECONNREFUSED: case ENETUNREACH:
 667                 case EHOSTDOWN:    case EHOSTUNREACH:
 668                         disconnect_on_error = 0;
 669                         break;
 670                 default:
 671                         drbd_err(connection, "%s failed, err = %d\n", what, err);
 672                 }
 673                 if (disconnect_on_error)
 674                         conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
 675         }
 676
 677         return sock;
 678 }
 679
 680 struct accept_wait_data {
 681         struct drbd_connection *connection;
 682         struct socket *s_listen;
 683         struct completion door_bell;
 684         void (*original_sk_state_change)(struct sock *sk);
 685
 686 };
 687
 688 static void drbd_incoming_connection(struct sock *sk)
 689 {
 690         struct accept_wait_data *ad = sk->sk_user_data;
 691         void (*state_change)(struct sock *sk);
 692
 693         state_change = ad->original_sk_state_change;
 694         if (sk->sk_state == TCP_ESTABLISHED)
 695                 complete(&ad->door_bell);
 696         state_change(sk);
 697 }
 698
 699 static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
 700 {
 701         int err, sndbuf_size, rcvbuf_size, my_addr_len;
 702         struct sockaddr_in6 my_addr;
 703         struct socket *s_listen;
 704         struct net_conf *nc;
 705         const char *what;
 706
 707         rcu_read_lock();
 708         nc = rcu_dereference(connection->net_conf);
 709         if (!nc) {
 710                 rcu_read_unlock();
 711                 return -EIO;
 712         }
 713         sndbuf_size = nc->sndbuf_size;
 714         rcvbuf_size = nc->rcvbuf_size;
 715         rcu_read_unlock();
 716
 717         my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
 718         memcpy(&my_addr, &connection->my_addr, my_addr_len);
 719
 720         what = "sock_create_kern";
 721         err = sock_create_kern(&init_net, ((struct sockaddr *)&my_addr)->sa_family,
 722                                SOCK_STREAM, IPPROTO_TCP, &s_listen);
 723         if (err) {
 724                 s_listen = NULL;
 725                 goto out;
 726         }
 727
 728         s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
 729         drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);
 730
 731         what = "bind before listen";
 732         err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
 733         if (err < 0)
 734                 goto out;
 735
 736         ad->s_listen = s_listen;
 737         write_lock_bh(&s_listen->sk->sk_callback_lock);
 738         ad->original_sk_state_change = s_listen->sk->sk_state_change;
 739         s_listen->sk->sk_state_change = drbd_incoming_connection;
 740         s_listen->sk->sk_user_data = ad;
 741         write_unlock_bh(&s_listen->sk->sk_callback_lock);
 742
 743         what = "listen";
 744         err = s_listen->ops->listen(s_listen, 5);
 745         if (err < 0)
 746                 goto out;
 747
 748         return 0;
 749 out:
 750         if (s_listen)
 751                 sock_release(s_listen);
 752         if (err < 0) {
 753                 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
 754                         drbd_err(connection, "%s failed, err = %d\n", what, err);
 755                         conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
 756                 }
 757         }
 758
 759         return -EIO;
 760 }
 761
 762 static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
 763 {
 764         write_lock_bh(&sk->sk_callback_lock);
 765         sk->sk_state_change = ad->original_sk_state_change;
 766         sk->sk_user_data = NULL;
 767         write_unlock_bh(&sk->sk_callback_lock);
 768 }
 769
 770 static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
 771 {
 772         int timeo, connect_int, err = 0;
 773         struct socket *s_estab = NULL;
 774         struct net_conf *nc;
 775
 776         rcu_read_lock();
 777         nc = rcu_dereference(connection->net_conf);
 778         if (!nc) {
 779                 rcu_read_unlock();
 780                 return NULL;
 781         }
 782         connect_int = nc->connect_int;
 783         rcu_read_unlock();
 784
 785         timeo = connect_int * HZ;
 786         /* 28.5% random jitter */
 787         timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;
 788
 789         err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
 790         if (err <= 0)
 791                 return NULL;
 792
 793         err = kernel_accept(ad->s_listen, &s_estab, 0);
 794         if (err < 0) {
 795                 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
 796                         drbd_err(connection, "accept failed, err = %d\n", err);
 797                         conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
 798                 }
 799         }
 800
 801         if (s_estab)
 802                 unregister_state_change(s_estab->sk, ad);
 803
 804         return s_estab;
 805 }
 806
 807 static int decode_header(struct drbd_connection *, void *, struct packet_info *);
 808
 809 static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
 810                              enum drbd_packet cmd)
 811 {
 812         if (!conn_prepare_command(connection, sock))
 813                 return -EIO;
 814         return conn_send_command(connection, sock, cmd, 0, NULL, 0);
 815 }
 816
 817 static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
 818 {
 819         unsigned int header_size = drbd_header_size(connection);
 820         struct packet_info pi;
 821         struct net_conf *nc;
 822         int err;
 823
 824         rcu_read_lock();
 825         nc = rcu_dereference(connection->net_conf);
 826         if (!nc) {
 827                 rcu_read_unlock();
 828                 return -EIO;
 829         }
 830         sock->sk->sk_rcvtimeo = nc->ping_timeo * 4 * HZ / 10;
 831         rcu_read_unlock();
 832
 833         err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
 834         if (err != header_size) {
 835                 if (err >= 0)
 836                         err = -EIO;
 837                 return err;
 838         }
 839         err = decode_header(connection, connection->data.rbuf, &pi);
 840         if (err)
 841                 return err;
 842         return pi.cmd;
 843 }
 844
 845 /**
 846  * drbd_socket_okay() - Free the socket if its connection is not okay
 847  * @sock:       pointer to the pointer to the socket.
 848  */
 849 static bool drbd_socket_okay(struct socket **sock)
 850 {
 851         int rr;
 852         char tb[4];
 853
 854         if (!*sock)
 855                 return false;
 856
 857         rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);
 858
 859         if (rr > 0 || rr == -EAGAIN) {
 860                 return true;
 861         } else {
 862                 sock_release(*sock);
 863                 *sock = NULL;
 864                 return false;
 865         }
 866 }
 867
 868 static bool connection_established(struct drbd_connection *connection,
 869                                    struct socket **sock1,
 870                                    struct socket **sock2)
 871 {
 872         struct net_conf *nc;
 873         int timeout;
 874         bool ok;
 875
 876         if (!*sock1 || !*sock2)
 877                 return false;
 878
 879         rcu_read_lock();
 880         nc = rcu_dereference(connection->net_conf);
 881         timeout = (nc->sock_check_timeo ?: nc->ping_timeo) * HZ / 10;
 882         rcu_read_unlock();
 883         schedule_timeout_interruptible(timeout);
 884
 885         ok = drbd_socket_okay(sock1);
 886         ok = drbd_socket_okay(sock2) && ok;
 887
 888         return ok;
 889 }
 890
 891 /* Gets called if a connection is established, or if a new minor gets created
 892    in a connection */
 893 int drbd_connected(struct drbd_peer_device *peer_device)
 894 {
 895         struct drbd_device *device = peer_device->device;
 896         int err;
 897
 898         atomic_set(&device->packet_seq, 0);
 899         device->peer_seq = 0;
 900
 901         device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
 902                 &peer_device->connection->cstate_mutex :
 903                 &device->own_state_mutex;
 904
 905         err = drbd_send_sync_param(peer_device);
 906         if (!err)
 907                 err = drbd_send_sizes(peer_device, 0, 0);
 908         if (!err)
 909                 err = drbd_send_uuids(peer_device);
 910         if (!err)
 911                 err = drbd_send_current_state(peer_device);
 912         clear_bit(USE_DEGR_WFC_T, &device->flags);
 913         clear_bit(RESIZE_PENDING, &device->flags);
 914         atomic_set(&device->ap_in_flight, 0);
 915         mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
 916         return err;
 917 }
 918
 919 /*
 920  * return values:
 921  *   1 yes, we have a valid connection
 922  *   0 oops, did not work out, please try again
 923  *  -1 peer talks different language,
 924  *     no point in trying again, please go standalone.
 925  *  -2 We do not have a network config...
 926  */
 927 static int conn_connect(struct drbd_connection *connection)
 928 {
 929         struct drbd_socket sock, msock;
 930         struct drbd_peer_device *peer_device;
 931         struct net_conf *nc;
 932         int vnr, timeout, h;
 933         bool discard_my_data, ok;
 934         enum drbd_state_rv rv;
 935         struct accept_wait_data ad = {
 936                 .connection = connection,
 937                 .door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
 938         };
 939
 940         clear_bit(DISCONNECT_SENT, &connection->flags);
 941         if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
 942                 return -2;
 943
 944         mutex_init(&sock.mutex);
 945         sock.sbuf = connection->data.sbuf;
 946         sock.rbuf = connection->data.rbuf;
 947         sock.socket = NULL;
 948         mutex_init(&msock.mutex);
 949         msock.sbuf = connection->meta.sbuf;
 950         msock.rbuf = connection->meta.rbuf;
 951         msock.socket = NULL;
 952
 953         /* Assume that the peer only understands protocol 80 until we know better.  */
 954         connection->agreed_pro_version = 80;
 955
 956         if (prepare_listen_socket(connection, &ad))
 957                 return 0;
 958
 959         do {
 960                 struct socket *s;
 961
 962                 s = drbd_try_connect(connection);
 963                 if (s) {
 964                         if (!sock.socket) {
 965                                 sock.socket = s;
 966                                 send_first_packet(connection, &sock, P_INITIAL_DATA);
 967                         } else if (!msock.socket) {
 968                                 clear_bit(RESOLVE_CONFLICTS, &connection->flags);
 969                                 msock.socket = s;
 970                                 send_first_packet(connection, &msock, P_INITIAL_META);
 971                         } else {
 972                                 drbd_err(connection, "Logic error in conn_connect()\n");
 973                                 goto out_release_sockets;
 974                         }
 975                 }
 976
 977                 if (connection_established(connection, &sock.socket, &msock.socket))
 978                         break;
 979
 980 retry:
 981                 s = drbd_wait_for_connect(connection, &ad);
 982                 if (s) {
 983                         int fp = receive_first_packet(connection, s);
 984                         drbd_socket_okay(&sock.socket);
 985                         drbd_socket_okay(&msock.socket);
 986                         switch (fp) {
 987                         case P_INITIAL_DATA:
 988                                 if (sock.socket) {
 989                                         drbd_warn(connection, "initial packet S crossed\n");
 990                                         sock_release(sock.socket);
 991                                         sock.socket = s;
 992                                         goto randomize;
 993                                 }
 994                                 sock.socket = s;
 995                                 break;
 996                         case P_INITIAL_META:
 997                                 set_bit(RESOLVE_CONFLICTS, &connection->flags);
 998                                 if (msock.socket) {
 999                                         drbd_warn(connection, "initial packet M crossed\n");
1000                                         sock_release(msock.socket);
1001                                         msock.socket = s;
1002                                         goto randomize;
1003                                 }
1004                                 msock.socket = s;
1005                                 break;
1006                         default:
1007                                 drbd_warn(connection, "Error receiving initial packet\n");
1008                                 sock_release(s);
1009 randomize:
1010                                 if (prandom_u32() & 1)
1011                                         goto retry;
1012                         }
1013                 }
1014
1015                 if (connection->cstate <= C_DISCONNECTING)
1016                         goto out_release_sockets;
1017                 if (signal_pending(current)) {
1018                         flush_signals(current);
1019                         smp_rmb();
1020                         if (get_t_state(&connection->receiver) == EXITING)
1021                                 goto out_release_sockets;
1022                 }
1023
1024                 ok = connection_established(connection, &sock.socket, &msock.socket);
1025         } while (!ok);
1026
1027         if (ad.s_listen)
1028                 sock_release(ad.s_listen);
1029
1030         sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1031         msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1032
1033         sock.socket->sk->sk_allocation = GFP_NOIO;
1034         msock.socket->sk->sk_allocation = GFP_NOIO;
1035
1036         sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
1037         msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
1038
1039         /* NOT YET ...
1040          * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
1041          * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1042          * first set it to the P_CONNECTION_FEATURES timeout,
1043          * which we set to 4x the configured ping_timeout. */
1044         rcu_read_lock();
1045         nc = rcu_dereference(connection->net_conf);
1046
1047         sock.socket->sk->sk_sndtimeo =
1048         sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
1049
1050         msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
1051         timeout = nc->timeout * HZ / 10;
1052         discard_my_data = nc->discard_my_data;
1053         rcu_read_unlock();
1054
1055         msock.socket->sk->sk_sndtimeo = timeout;
1056
1057         /* we don't want delays.
1058          * we use TCP_CORK where appropriate, though */
1059         drbd_tcp_nodelay(sock.socket);
1060         drbd_tcp_nodelay(msock.socket);
1061
1062         connection->data.socket = sock.socket;
1063         connection->meta.socket = msock.socket;
1064         connection->last_received = jiffies;
1065
1066         h = drbd_do_features(connection);
1067         if (h <= 0)
1068                 return h;
1069
1070         if (connection->cram_hmac_tfm) {
1071                 /* drbd_request_state(device, NS(conn, WFAuth)); */
1072                 switch (drbd_do_auth(connection)) {
1073                 case -1:
1074                         drbd_err(connection, "Authentication of peer failed\n");
1075                         return -1;
1076                 case 0:
1077                         drbd_err(connection, "Authentication of peer failed, trying again.\n");
1078                         return 0;
1079                 }
1080         }
1081
1082         connection->data.socket->sk->sk_sndtimeo = timeout;
1083         connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1084
1085         if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1086                 return -1;
1087
1088         /* Prevent a race between resync-handshake and
1089          * being promoted to Primary.
1090          *
1091          * Grab and release the state mutex, so we know that any current
1092          * drbd_set_role() is finished, and any incoming drbd_set_role
1093          * will see the STATE_SENT flag, and wait for it to be cleared.
1094          */
1095         idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1096                 mutex_lock(peer_device->device->state_mutex);
1097
1098         set_bit(STATE_SENT, &connection->flags);
1099
1100         idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1101                 mutex_unlock(peer_device->device->state_mutex);
1102
1103         rcu_read_lock();
1104         idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1105                 struct drbd_device *device = peer_device->device;
1106                 kref_get(&device->kref);
1107                 rcu_read_unlock();
1108
1109                 if (discard_my_data)
1110                         set_bit(DISCARD_MY_DATA, &device->flags);
1111                 else
1112                         clear_bit(DISCARD_MY_DATA, &device->flags);
1113
1114                 drbd_connected(peer_device);
1115                 kref_put(&device->kref, drbd_destroy_device);
1116                 rcu_read_lock();
1117         }
1118         rcu_read_unlock();
1119
1120         rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1121         if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1122                 clear_bit(STATE_SENT, &connection->flags);
1123                 return 0;
1124         }
1125
1126         drbd_thread_start(&connection->ack_receiver);
1127         /* opencoded create_singlethread_workqueue(),
1128          * to be able to use format string arguments */
1129         connection->ack_sender =
1130                 alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
1131         if (!connection->ack_sender) {
1132                 drbd_err(connection, "Failed to create workqueue ack_sender\n");
1133                 return 0;
1134         }
1135
1136         mutex_lock(&connection->resource->conf_update);
1137         /* The discard_my_data flag is a single-shot modifier to the next
1138          * connection attempt, the handshake of which is now well underway.
1139          * No need for rcu style copying of the whole struct
1140          * just to clear a single value. */
1141         connection->net_conf->discard_my_data = 0;
1142         mutex_unlock(&connection->resource->conf_update);
1143
1144         return h;
1145
1146 out_release_sockets:
1147         if (ad.s_listen)
1148                 sock_release(ad.s_listen);
1149         if (sock.socket)
1150                 sock_release(sock.socket);
1151         if (msock.socket)
1152                 sock_release(msock.socket);
1153         return -1;
1154 }
1155
1156 static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1157 {
1158         unsigned int header_size = drbd_header_size(connection);
1159
1160         if (header_size == sizeof(struct p_header100) &&
1161             *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1162                 struct p_header100 *h = header;
1163                 if (h->pad != 0) {
1164                         drbd_err(connection, "Header padding is not zero\n");
1165                         return -EINVAL;
1166                 }
1167                 pi->vnr = be16_to_cpu(h->volume);
1168                 pi->cmd = be16_to_cpu(h->command);
1169                 pi->size = be32_to_cpu(h->length);
1170         } else if (header_size == sizeof(struct p_header95) &&
1171                    *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1172                 struct p_header95 *h = header;
1173                 pi->cmd = be16_to_cpu(h->command);
1174                 pi->size = be32_to_cpu(h->length);
1175                 pi->vnr = 0;
1176         } else if (header_size == sizeof(struct p_header80) &&
1177                    *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1178                 struct p_header80 *h = header;
1179                 pi->cmd = be16_to_cpu(h->command);
1180                 pi->size = be16_to_cpu(h->length);
1181                 pi->vnr = 0;
1182         } else {
1183                 drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1184                          be32_to_cpu(*(__be32 *)header),
1185                          connection->agreed_pro_version);
1186                 return -EINVAL;
1187         }
1188         pi->data = header + header_size;
1189         return 0;
1190 }
1191
1192 static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1193 {
1194         void *buffer = connection->data.rbuf;
1195         int err;
1196
1197         err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1198         if (err)
1199                 return err;
1200
1201         err = decode_header(connection, buffer, pi);
1202         connection->last_received = jiffies;
1203
1204         return err;
1205 }
1206
1207 static void drbd_flush(struct drbd_connection *connection)
1208 {
1209         int rv;
1210         struct drbd_peer_device *peer_device;
1211         int vnr;
1212
1213         if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1214                 rcu_read_lock();
1215                 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1216                         struct drbd_device *device = peer_device->device;
1217
1218                         if (!get_ldev(device))
1219                                 continue;
1220                         kref_get(&device->kref);
1221                         rcu_read_unlock();
1222
1223                         /* Right now, we have only this one synchronous code path
1224                          * for flushes between request epochs.
1225                          * We may want to make those asynchronous,
1226                          * or at least parallelize the flushes to the volume devices.
1227                          */
1228                         device->flush_jif = jiffies;
1229                         set_bit(FLUSH_PENDING, &device->flags);
1230                         rv = blkdev_issue_flush(device->ldev->backing_bdev,
1231                                         GFP_NOIO, NULL);
1232                         clear_bit(FLUSH_PENDING, &device->flags);
1233                         if (rv) {
1234                                 drbd_info(device, "local disk flush failed with status %d\n", rv);
1235                                 /* would rather check on EOPNOTSUPP, but that is not reliable.
1236                                  * don't try again for ANY return value != 0
1237                                  * if (rv == -EOPNOTSUPP) */
1238                                 drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1239                         }
1240                         put_ldev(device);
1241                         kref_put(&device->kref, drbd_destroy_device);
1242
1243                         rcu_read_lock();
1244                         if (rv)
1245                                 break;
1246                 }
1247                 rcu_read_unlock();
1248         }
1249 }
1250
1251 /**
1252  * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1253  * @device:     DRBD device.
1254  * @epoch:      Epoch object.
1255  * @ev:         Epoch event.
1256  */
1257 static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1258                                                struct drbd_epoch *epoch,
1259                                                enum epoch_event ev)
1260 {
1261         int epoch_size;
1262         struct drbd_epoch *next_epoch;
1263         enum finish_epoch rv = FE_STILL_LIVE;
1264
1265         spin_lock(&connection->epoch_lock);
1266         do {
1267                 next_epoch = NULL;
1268
1269                 epoch_size = atomic_read(&epoch->epoch_size);
1270
1271                 switch (ev & ~EV_CLEANUP) {
1272                 case EV_PUT:
1273                         atomic_dec(&epoch->active);
1274                         break;
1275                 case EV_GOT_BARRIER_NR:
1276                         set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1277                         break;
1278                 case EV_BECAME_LAST:
1279                         /* nothing to do*/
1280                         break;
1281                 }
1282
1283                 if (epoch_size != 0 &&
1284                     atomic_read(&epoch->active) == 0 &&
1285                     (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1286                         if (!(ev & EV_CLEANUP)) {
1287                                 spin_unlock(&connection->epoch_lock);
1288                                 drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1289                                 spin_lock(&connection->epoch_lock);
1290                         }
1291 #if 0
1292                         /* FIXME: dec unacked on connection, once we have
1293                          * something to count pending connection packets in. */
1294                         if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1295                                 dec_unacked(epoch->connection);
1296 #endif
1297
1298                         if (connection->current_epoch != epoch) {
1299                                 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1300                                 list_del(&epoch->list);
1301                                 ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1302                                 connection->epochs--;
1303                                 kfree(epoch);
1304
1305                                 if (rv == FE_STILL_LIVE)
1306                                         rv = FE_DESTROYED;
1307                         } else {
1308                                 epoch->flags = 0;
1309                                 atomic_set(&epoch->epoch_size, 0);
1310                                 /* atomic_set(&epoch->active, 0); is already zero */
1311                                 if (rv == FE_STILL_LIVE)
1312                                         rv = FE_RECYCLED;
1313                         }
1314                 }
1315
1316                 if (!next_epoch)
1317                         break;
1318
1319                 epoch = next_epoch;
1320         } while (1);
1321
1322         spin_unlock(&connection->epoch_lock);
1323
1324         return rv;
1325 }
1326
1327 static enum write_ordering_e
1328 max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1329 {
1330         struct disk_conf *dc;
1331
1332         dc = rcu_dereference(bdev->disk_conf);
1333
1334         if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1335                 wo = WO_DRAIN_IO;
1336         if (wo == WO_DRAIN_IO && !dc->disk_drain)
1337                 wo = WO_NONE;
1338
1339         return wo;
1340 }
1341
1342 /**
1343  * drbd_bump_write_ordering() - Fall back to an other write ordering method
1344  * @connection: DRBD connection.
1345  * @wo:         Write ordering method to try.
1346  */
1347 void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1348                               enum write_ordering_e wo)
1349 {
1350         struct drbd_device *device;
1351         enum write_ordering_e pwo;
1352         int vnr;
1353         static char *write_ordering_str[] = {
1354                 [WO_NONE] = "none",
1355                 [WO_DRAIN_IO] = "drain",
1356                 [WO_BDEV_FLUSH] = "flush",
1357         };
1358
1359         pwo = resource->write_ordering;
1360         if (wo != WO_BDEV_FLUSH)
1361                 wo = min(pwo, wo);
1362         rcu_read_lock();
1363         idr_for_each_entry(&resource->devices, device, vnr) {
1364                 if (get_ldev(device)) {
1365                         wo = max_allowed_wo(device->ldev, wo);
1366                         if (device->ldev == bdev)
1367                                 bdev = NULL;
1368                         put_ldev(device);
1369                 }
1370         }
1371
1372         if (bdev)
1373                 wo = max_allowed_wo(bdev, wo);
1374
1375         rcu_read_unlock();
1376
1377         resource->write_ordering = wo;
1378         if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1379                 drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1380 }
1381
1382 /**
1383  * drbd_submit_peer_request()
1384  * @device:     DRBD device.
1385  * @peer_req:   peer request
1386  * @rw:         flag field, see bio->bi_rw
1387  *
1388  * May spread the pages to multiple bios,
1389  * depending on bio_add_page restrictions.
1390  *
1391  * Returns 0 if all bios have been submitted,
1392  * -ENOMEM if we could not allocate enough bios,
1393  * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1394  *  single page to an empty bio (which should never happen and likely indicates
1395  *  that the lower level IO stack is in some way broken). This has been observed
1396  *  on certain Xen deployments.
1397  */
1398 /* TODO allocate from our own bio_set. */
1399 int drbd_submit_peer_request(struct drbd_device *device,
1400                              struct drbd_peer_request *peer_req,
1401                              const unsigned rw, const int fault_type)
1402 {
1403         struct bio *bios = NULL;
1404         struct bio *bio;
1405         struct page *page = peer_req->pages;
1406         sector_t sector = peer_req->i.sector;
1407         unsigned data_size = peer_req->i.size;
1408         unsigned n_bios = 0;
1409         unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
1410         int err = -ENOMEM;
1411
1412         if (peer_req->flags & EE_IS_TRIM_USE_ZEROOUT) {
1413                 /* wait for all pending IO completions, before we start
1414                  * zeroing things out. */
1415                 conn_wait_active_ee_empty(peer_req->peer_device->connection);
1416                 /* add it to the active list now,
1417                  * so we can find it to present it in debugfs */
1418                 peer_req->submit_jif = jiffies;
1419                 peer_req->flags |= EE_SUBMITTED;
1420                 spin_lock_irq(&device->resource->req_lock);
1421                 list_add_tail(&peer_req->w.list, &device->active_ee);
1422                 spin_unlock_irq(&device->resource->req_lock);
1423                 if (blkdev_issue_zeroout(device->ldev->backing_bdev,
1424                         sector, data_size >> 9, GFP_NOIO, false))
1425                         peer_req->flags |= EE_WAS_ERROR;
1426                 drbd_endio_write_sec_final(peer_req);
1427                 return 0;
1428         }
1429
1430         /* Discards don't have any payload.
1431          * But the scsi layer still expects a bio_vec it can use internally,
1432          * see sd_setup_discard_cmnd() and blk_add_request_payload(). */
1433         if (peer_req->flags & EE_IS_TRIM)
1434                 nr_pages = 1;
1435
1436         /* In most cases, we will only need one bio.  But in case the lower
1437          * level restrictions happen to be different at this offset on this
1438          * side than those of the sending peer, we may need to submit the
1439          * request in more than one bio.
1440          *
1441          * Plain bio_alloc is good enough here, this is no DRBD internally
1442          * generated bio, but a bio allocated on behalf of the peer.
1443          */
1444 next_bio:
1445         bio = bio_alloc(GFP_NOIO, nr_pages);
1446         if (!bio) {
1447                 drbd_err(device, "submit_ee: Allocation of a bio failed (nr_pages=%u)\n", nr_pages);
1448                 goto fail;
1449         }
1450         /* > peer_req->i.sector, unless this is the first bio */
1451         bio->bi_iter.bi_sector = sector;
1452         bio->bi_bdev = device->ldev->backing_bdev;
1453         bio->bi_rw = rw;
1454         bio->bi_private = peer_req;
1455         bio->bi_end_io = drbd_peer_request_endio;
1456
1457         bio->bi_next = bios;
1458         bios = bio;
1459         ++n_bios;
1460
1461         if (rw & REQ_DISCARD) {
1462                 bio->bi_iter.bi_size = data_size;
1463                 goto submit;
1464         }
1465
1466         page_chain_for_each(page) {
1467                 unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1468                 if (!bio_add_page(bio, page, len, 0)) {
1469                         /* A single page must always be possible!
1470                          * But in case it fails anyways,
1471                          * we deal with it, and complain (below). */
1472                         if (bio->bi_vcnt == 0) {
1473                                 drbd_err(device,
1474                                         "bio_add_page failed for len=%u, "
1475                                         "bi_vcnt=0 (bi_sector=%llu)\n",
1476                                         len, (uint64_t)bio->bi_iter.bi_sector);
1477                                 err = -ENOSPC;
1478                                 goto fail;
1479                         }
1480                         goto next_bio;
1481                 }
1482                 data_size -= len;
1483                 sector += len >> 9;
1484                 --nr_pages;
1485         }
1486         D_ASSERT(device, data_size == 0);
1487 submit:
1488         D_ASSERT(device, page == NULL);
1489
1490         atomic_set(&peer_req->pending_bios, n_bios);
1491         /* for debugfs: update timestamp, mark as submitted */
1492         peer_req->submit_jif = jiffies;
1493         peer_req->flags |= EE_SUBMITTED;
1494         do {
1495                 bio = bios;
1496                 bios = bios->bi_next;
1497                 bio->bi_next = NULL;
1498
1499                 drbd_generic_make_request(device, fault_type, bio);
1500         } while (bios);
1501         return 0;
1502
1503 fail:
1504         while (bios) {
1505                 bio = bios;
1506                 bios = bios->bi_next;
1507                 bio_put(bio);
1508         }
1509         return err;
1510 }
1511
1512 static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1513                                              struct drbd_peer_request *peer_req)
1514 {
1515         struct drbd_interval *i = &peer_req->i;
1516
1517         drbd_remove_interval(&device->write_requests, i);
1518         drbd_clear_interval(i);
1519
1520         /* Wake up any processes waiting for this peer request to complete.  */
1521         if (i->waiting)
1522                 wake_up(&device->misc_wait);
1523 }
1524
1525 static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1526 {
1527         struct drbd_peer_device *peer_device;
1528         int vnr;
1529
1530         rcu_read_lock();
1531         idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1532                 struct drbd_device *device = peer_device->device;
1533
1534                 kref_get(&device->kref);
1535                 rcu_read_unlock();
1536                 drbd_wait_ee_list_empty(device, &device->active_ee);
1537                 kref_put(&device->kref, drbd_destroy_device);
1538                 rcu_read_lock();
1539         }
1540         rcu_read_unlock();
1541 }
1542
1543 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1544 {
1545         int rv;
1546         struct p_barrier *p = pi->data;
1547         struct drbd_epoch *epoch;
1548
1549         /* FIXME these are unacked on connection,
1550          * not a specific (peer)device.
1551          */
1552         connection->current_epoch->barrier_nr = p->barrier;
1553         connection->current_epoch->connection = connection;
1554         rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1555
1556         /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1557          * the activity log, which means it would not be resynced in case the
1558          * R_PRIMARY crashes now.
1559          * Therefore we must send the barrier_ack after the barrier request was
1560          * completed. */
1561         switch (connection->resource->write_ordering) {
1562         case WO_NONE:
1563                 if (rv == FE_RECYCLED)
1564                         return 0;
1565
1566                 /* receiver context, in the writeout path of the other node.
1567                  * avoid potential distributed deadlock */
1568                 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1569                 if (epoch)
1570                         break;
1571                 else
1572                         drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1573                         /* Fall through */
1574
1575         case WO_BDEV_FLUSH:
1576         case WO_DRAIN_IO:
1577                 conn_wait_active_ee_empty(connection);
1578                 drbd_flush(connection);
1579
1580                 if (atomic_read(&connection->current_epoch->epoch_size)) {
1581                         epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1582                         if (epoch)
1583                                 break;
1584                 }
1585
1586                 return 0;
1587         default:
1588                 drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1589                          connection->resource->write_ordering);
1590                 return -EIO;
1591         }
1592
1593         epoch->flags = 0;
1594         atomic_set(&epoch->epoch_size, 0);
1595         atomic_set(&epoch->active, 0);
1596
1597         spin_lock(&connection->epoch_lock);
1598         if (atomic_read(&connection->current_epoch->epoch_size)) {
1599                 list_add(&epoch->list, &connection->current_epoch->list);
1600                 connection->current_epoch = epoch;
1601                 connection->epochs++;
1602         } else {
1603                 /* The current_epoch got recycled while we allocated this one... */
1604                 kfree(epoch);
1605         }
1606         spin_unlock(&connection->epoch_lock);
1607
1608         return 0;
1609 }
1610
1611 /* used from receive_RSDataReply (recv_resync_read)
1612  * and from receive_Data */
1613 static struct drbd_peer_request *
1614 read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1615               struct packet_info *pi) __must_hold(local)
1616 {
1617         struct drbd_device *device = peer_device->device;
1618         const sector_t capacity = drbd_get_capacity(device->this_bdev);
1619         struct drbd_peer_request *peer_req;
1620         struct page *page;
1621         int digest_size, err;
1622         unsigned int data_size = pi->size, ds;
1623         void *dig_in = peer_device->connection->int_dig_in;
1624         void *dig_vv = peer_device->connection->int_dig_vv;
1625         unsigned long *data;
1626         struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1627
1628         digest_size = 0;
1629         if (!trim && peer_device->connection->peer_integrity_tfm) {
1630                 digest_size = crypto_ahash_digestsize(peer_device->connection->peer_integrity_tfm);
1631                 /*
1632                  * FIXME: Receive the incoming digest into the receive buffer
1633                  *        here, together with its struct p_data?
1634                  */
1635                 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1636                 if (err)
1637                         return NULL;
1638                 data_size -= digest_size;
1639         }
1640
1641         if (trim) {
1642                 D_ASSERT(peer_device, data_size == 0);
1643                 data_size = be32_to_cpu(trim->size);
1644         }
1645
1646         if (!expect(IS_ALIGNED(data_size, 512)))
1647                 return NULL;
1648         /* prepare for larger trim requests. */
1649         if (!trim && !expect(data_size <= DRBD_MAX_BIO_SIZE))
1650                 return NULL;
1651
1652         /* even though we trust out peer,
1653          * we sometimes have to double check. */
1654         if (sector + (data_size>>9) > capacity) {
1655                 drbd_err(device, "request from peer beyond end of local disk: "
1656                         "capacity: %llus < sector: %llus + size: %u\n",
1657                         (unsigned long long)capacity,
1658                         (unsigned long long)sector, data_size);
1659                 return NULL;
1660         }
1661
1662         /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1663          * "criss-cross" setup, that might cause write-out on some other DRBD,
1664          * which in turn might block on the other node at this very place.  */
1665         peer_req = drbd_alloc_peer_req(peer_device, id, sector, data_size, trim == NULL, GFP_NOIO);
1666         if (!peer_req)
1667                 return NULL;
1668
1669         peer_req->flags |= EE_WRITE;
1670         if (trim)
1671                 return peer_req;
1672
1673         ds = data_size;
1674         page = peer_req->pages;
1675         page_chain_for_each(page) {
1676                 unsigned len = min_t(int, ds, PAGE_SIZE);
1677                 data = kmap(page);
1678                 err = drbd_recv_all_warn(peer_device->connection, data, len);
1679                 if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1680                         drbd_err(device, "Fault injection: Corrupting data on receive\n");
1681                         data[0] = data[0] ^ (unsigned long)-1;
1682                 }
1683                 kunmap(page);
1684                 if (err) {
1685                         drbd_free_peer_req(device, peer_req);
1686                         return NULL;
1687                 }
1688                 ds -= len;
1689         }
1690
1691         if (digest_size) {
1692                 drbd_csum_ee(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv);
1693                 if (memcmp(dig_in, dig_vv, digest_size)) {
1694                         drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1695                                 (unsigned long long)sector, data_size);
1696                         drbd_free_peer_req(device, peer_req);
1697                         return NULL;
1698                 }
1699         }
1700         device->recv_cnt += data_size >> 9;
1701         return peer_req;
1702 }
1703
1704 /* drbd_drain_block() just takes a data block
1705  * out of the socket input buffer, and discards it.
1706  */
1707 static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1708 {
1709         struct page *page;
1710         int err = 0;
1711         void *data;
1712
1713         if (!data_size)
1714                 return 0;
1715
1716         page = drbd_alloc_pages(peer_device, 1, 1);
1717
1718         data = kmap(page);
1719         while (data_size) {
1720                 unsigned int len = min_t(int, data_size, PAGE_SIZE);
1721
1722                 err = drbd_recv_all_warn(peer_device->connection, data, len);
1723                 if (err)
1724                         break;
1725                 data_size -= len;
1726         }
1727         kunmap(page);
1728         drbd_free_pages(peer_device->device, page, 0);
1729         return err;
1730 }
1731
1732 static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
1733                            sector_t sector, int data_size)
1734 {
1735         struct bio_vec bvec;
1736         struct bvec_iter iter;
1737         struct bio *bio;
1738         int digest_size, err, expect;
1739         void *dig_in = peer_device->connection->int_dig_in;
1740         void *dig_vv = peer_device->connection->int_dig_vv;
1741
1742         digest_size = 0;
1743         if (peer_device->connection->peer_integrity_tfm) {
1744                 digest_size = crypto_ahash_digestsize(peer_device->connection->peer_integrity_tfm);
1745                 err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1746                 if (err)
1747                         return err;
1748                 data_size -= digest_size;
1749         }
1750
1751         /* optimistically update recv_cnt.  if receiving fails below,
1752          * we disconnect anyways, and counters will be reset. */
1753         peer_device->device->recv_cnt += data_size>>9;
1754
1755         bio = req->master_bio;
1756         D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
1757
1758         bio_for_each_segment(bvec, bio, iter) {
1759                 void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
1760                 expect = min_t(int, data_size, bvec.bv_len);
1761                 err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
1762                 kunmap(bvec.bv_page);
1763                 if (err)
1764                         return err;
1765                 data_size -= expect;
1766         }
1767
1768         if (digest_size) {
1769                 drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
1770                 if (memcmp(dig_in, dig_vv, digest_size)) {
1771                         drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
1772                         return -EINVAL;
1773                 }
1774         }
1775
1776         D_ASSERT(peer_device->device, data_size == 0);
1777         return 0;
1778 }
1779
1780 /*
1781  * e_end_resync_block() is called in ack_sender context via
1782  * drbd_finish_peer_reqs().
1783  */
1784 static int e_end_resync_block(struct drbd_work *w, int unused)
1785 {
1786         struct drbd_peer_request *peer_req =
1787                 container_of(w, struct drbd_peer_request, w);
1788         struct drbd_peer_device *peer_device = peer_req->peer_device;
1789         struct drbd_device *device = peer_device->device;
1790         sector_t sector = peer_req->i.sector;
1791         int err;
1792
1793         D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1794
1795         if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1796                 drbd_set_in_sync(device, sector, peer_req->i.size);
1797                 err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
1798         } else {
1799                 /* Record failure to sync */
1800                 drbd_rs_failed_io(device, sector, peer_req->i.size);
1801
1802                 err  = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1803         }
1804         dec_unacked(device);
1805
1806         return err;
1807 }
1808
1809 static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
1810                             struct packet_info *pi) __releases(local)
1811 {
1812         struct drbd_device *device = peer_device->device;
1813         struct drbd_peer_request *peer_req;
1814
1815         peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
1816         if (!peer_req)
1817                 goto fail;
1818
1819         dec_rs_pending(device);
1820
1821         inc_unacked(device);
1822         /* corresponding dec_unacked() in e_end_resync_block()
1823          * respective _drbd_clear_done_ee */
1824
1825         peer_req->w.cb = e_end_resync_block;
1826         peer_req->submit_jif = jiffies;
1827
1828         spin_lock_irq(&device->resource->req_lock);
1829         list_add_tail(&peer_req->w.list, &device->sync_ee);
1830         spin_unlock_irq(&device->resource->req_lock);
1831
1832         atomic_add(pi->size >> 9, &device->rs_sect_ev);
1833         if (drbd_submit_peer_request(device, peer_req, WRITE, DRBD_FAULT_RS_WR) == 0)
1834                 return 0;
1835
1836         /* don't care for the reason here */
1837         drbd_err(device, "submit failed, triggering re-connect\n");
1838         spin_lock_irq(&device->resource->req_lock);
1839         list_del(&peer_req->w.list);
1840         spin_unlock_irq(&device->resource->req_lock);
1841
1842         drbd_free_peer_req(device, peer_req);
1843 fail:
1844         put_ldev(device);
1845         return -EIO;
1846 }
1847
1848 static struct drbd_request *
1849 find_request(struct drbd_device *device, struct rb_root *root, u64 id,
1850              sector_t sector, bool missing_ok, const char *func)
1851 {
1852         struct drbd_request *req;
1853
1854         /* Request object according to our peer */
1855         req = (struct drbd_request *)(unsigned long)id;
1856         if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
1857                 return req;
1858         if (!missing_ok) {
1859                 drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
1860                         (unsigned long)id, (unsigned long long)sector);
1861         }
1862         return NULL;
1863 }
1864
1865 static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
1866 {
1867         struct drbd_peer_device *peer_device;
1868         struct drbd_device *device;
1869         struct drbd_request *req;
1870         sector_t sector;
1871         int err;
1872         struct p_data *p = pi->data;
1873
1874         peer_device = conn_peer_device(connection, pi->vnr);
1875         if (!peer_device)
1876                 return -EIO;
1877         device = peer_device->device;
1878
1879         sector = be64_to_cpu(p->sector);
1880
1881         spin_lock_irq(&device->resource->req_lock);
1882         req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
1883         spin_unlock_irq(&device->resource->req_lock);
1884         if (unlikely(!req))
1885                 return -EIO;
1886
1887         /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
1888          * special casing it there for the various failure cases.
1889          * still no race with drbd_fail_pending_reads */
1890         err = recv_dless_read(peer_device, req, sector, pi->size);
1891         if (!err)
1892                 req_mod(req, DATA_RECEIVED);
1893         /* else: nothing. handled from drbd_disconnect...
1894          * I don't think we may complete this just yet
1895          * in case we are "on-disconnect: freeze" */
1896
1897         return err;
1898 }
1899
1900 static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
1901 {
1902         struct drbd_peer_device *peer_device;
1903         struct drbd_device *device;
1904         sector_t sector;
1905         int err;
1906         struct p_data *p = pi->data;
1907
1908         peer_device = conn_peer_device(connection, pi->vnr);
1909         if (!peer_device)
1910                 return -EIO;
1911         device = peer_device->device;
1912
1913         sector = be64_to_cpu(p->sector);
1914         D_ASSERT(device, p->block_id == ID_SYNCER);
1915
1916         if (get_ldev(device)) {
1917                 /* data is submitted to disk within recv_resync_read.
1918                  * corresponding put_ldev done below on error,
1919                  * or in drbd_peer_request_endio. */
1920                 err = recv_resync_read(peer_device, sector, pi);
1921         } else {
1922                 if (__ratelimit(&drbd_ratelimit_state))
1923                         drbd_err(device, "Can not write resync data to local disk.\n");
1924
1925                 err = drbd_drain_block(peer_device, pi->size);
1926
1927                 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
1928         }
1929
1930         atomic_add(pi->size >> 9, &device->rs_sect_in);
1931
1932         return err;
1933 }
1934
1935 static void restart_conflicting_writes(struct drbd_device *device,
1936                                        sector_t sector, int size)
1937 {
1938         struct drbd_interval *i;
1939         struct drbd_request *req;
1940
1941         drbd_for_each_overlap(i, &device->write_requests, sector, size) {
1942                 if (!i->local)
1943                         continue;
1944                 req = container_of(i, struct drbd_request, i);
1945                 if (req->rq_state & RQ_LOCAL_PENDING ||
1946                     !(req->rq_state & RQ_POSTPONED))
1947                         continue;
1948                 /* as it is RQ_POSTPONED, this will cause it to
1949                  * be queued on the retry workqueue. */
1950                 __req_mod(req, CONFLICT_RESOLVED, NULL);
1951         }
1952 }
1953
1954 /*
1955  * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
1956  */
1957 static int e_end_block(struct drbd_work *w, int cancel)
1958 {
1959         struct drbd_peer_request *peer_req =
1960                 container_of(w, struct drbd_peer_request, w);
1961         struct drbd_peer_device *peer_device = peer_req->peer_device;
1962         struct drbd_device *device = peer_device->device;
1963         sector_t sector = peer_req->i.sector;
1964         int err = 0, pcmd;
1965
1966         if (peer_req->flags & EE_SEND_WRITE_ACK) {
1967                 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1968                         pcmd = (device->state.conn >= C_SYNC_SOURCE &&
1969                                 device->state.conn <= C_PAUSED_SYNC_T &&
1970                                 peer_req->flags & EE_MAY_SET_IN_SYNC) ?
1971                                 P_RS_WRITE_ACK : P_WRITE_ACK;
1972                         err = drbd_send_ack(peer_device, pcmd, peer_req);
1973                         if (pcmd == P_RS_WRITE_ACK)
1974                                 drbd_set_in_sync(device, sector, peer_req->i.size);
1975                 } else {
1976                         err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
1977                         /* we expect it to be marked out of sync anyways...
1978                          * maybe assert this?  */
1979                 }
1980                 dec_unacked(device);
1981         }
1982
1983         /* we delete from the conflict detection hash _after_ we sent out the
1984          * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right.  */
1985         if (peer_req->flags & EE_IN_INTERVAL_TREE) {
1986                 spin_lock_irq(&device->resource->req_lock);
1987                 D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
1988                 drbd_remove_epoch_entry_interval(device, peer_req);
1989                 if (peer_req->flags & EE_RESTART_REQUESTS)
1990                         restart_conflicting_writes(device, sector, peer_req->i.size);
1991                 spin_unlock_irq(&device->resource->req_lock);
1992         } else
1993                 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
1994
1995         drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
1996
1997         return err;
1998 }
1999
2000 static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2001 {
2002         struct drbd_peer_request *peer_req =
2003                 container_of(w, struct drbd_peer_request, w);
2004         struct drbd_peer_device *peer_device = peer_req->peer_device;
2005         int err;
2006
2007         err = drbd_send_ack(peer_device, ack, peer_req);
2008         dec_unacked(peer_device->device);
2009
2010         return err;
2011 }
2012
2013 static int e_send_superseded(struct drbd_work *w, int unused)
2014 {
2015         return e_send_ack(w, P_SUPERSEDED);
2016 }
2017
2018 static int e_send_retry_write(struct drbd_work *w, int unused)
2019 {
2020         struct drbd_peer_request *peer_req =
2021                 container_of(w, struct drbd_peer_request, w);
2022         struct drbd_connection *connection = peer_req->peer_device->connection;
2023
2024         return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2025                              P_RETRY_WRITE : P_SUPERSEDED);
2026 }
2027
2028 static bool seq_greater(u32 a, u32 b)
2029 {
2030         /*
2031          * We assume 32-bit wrap-around here.
2032          * For 24-bit wrap-around, we would have to shift:
2033          *  a <<= 8; b <<= 8;
2034          */
2035         return (s32)a - (s32)b > 0;
2036 }
2037
2038 static u32 seq_max(u32 a, u32 b)
2039 {
2040         return seq_greater(a, b) ? a : b;
2041 }
2042
2043 static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2044 {
2045         struct drbd_device *device = peer_device->device;
2046         unsigned int newest_peer_seq;
2047
2048         if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2049                 spin_lock(&device->peer_seq_lock);
2050                 newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2051                 device->peer_seq = newest_peer_seq;
2052                 spin_unlock(&device->peer_seq_lock);
2053                 /* wake up only if we actually changed device->peer_seq */
2054                 if (peer_seq == newest_peer_seq)
2055                         wake_up(&device->seq_wait);
2056         }
2057 }
2058
2059 static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2060 {
2061         return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2062 }
2063
2064 /* maybe change sync_ee into interval trees as well? */
2065 static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2066 {
2067         struct drbd_peer_request *rs_req;
2068         bool rv = 0;
2069
2070         spin_lock_irq(&device->resource->req_lock);
2071         list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2072                 if (overlaps(peer_req->i.sector, peer_req->i.size,
2073                              rs_req->i.sector, rs_req->i.size)) {
2074                         rv = 1;
2075                         break;
2076                 }
2077         }
2078         spin_unlock_irq(&device->resource->req_lock);
2079
2080         return rv;
2081 }
2082
2083 /* Called from receive_Data.
2084  * Synchronize packets on sock with packets on msock.
2085  *
2086  * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2087  * packet traveling on msock, they are still processed in the order they have
2088  * been sent.
2089  *
2090  * Note: we don't care for Ack packets overtaking P_DATA packets.
2091  *
2092  * In case packet_seq is larger than device->peer_seq number, there are
2093  * outstanding packets on the msock. We wait for them to arrive.
2094  * In case we are the logically next packet, we update device->peer_seq
2095  * ourselves. Correctly handles 32bit wrap around.
2096  *
2097  * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2098  * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2099  * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2100  * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2101  *
2102  * returns 0 if we may process the packet,
2103  * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2104 static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2105 {
2106         struct drbd_device *device = peer_device->device;
2107         DEFINE_WAIT(wait);
2108         long timeout;
2109         int ret = 0, tp;
2110
2111         if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2112                 return 0;
2113
2114         spin_lock(&device->peer_seq_lock);
2115         for (;;) {
2116                 if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2117                         device->peer_seq = seq_max(device->peer_seq, peer_seq);
2118                         break;
2119                 }
2120
2121                 if (signal_pending(current)) {
2122                         ret = -ERESTARTSYS;
2123                         break;
2124                 }
2125
2126                 rcu_read_lock();
2127                 tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2128                 rcu_read_unlock();
2129
2130                 if (!tp)
2131                         break;
2132
2133                 /* Only need to wait if two_primaries is enabled */
2134                 prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2135                 spin_unlock(&device->peer_seq_lock);
2136                 rcu_read_lock();
2137                 timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2138                 rcu_read_unlock();
2139                 timeout = schedule_timeout(timeout);
2140                 spin_lock(&device->peer_seq_lock);
2141                 if (!timeout) {
2142                         ret = -ETIMEDOUT;
2143                         drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2144                         break;
2145                 }
2146         }
2147         spin_unlock(&device->peer_seq_lock);
2148         finish_wait(&device->seq_wait, &wait);
2149         return ret;
2150 }
2151
2152 /* see also bio_flags_to_wire()
2153  * DRBD_REQ_*, because we need to semantically map the flags to data packet
2154  * flags and back. We may replicate to other kernel versions. */
2155 static unsigned long wire_flags_to_bio(u32 dpf)
2156 {
2157         return  (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2158                 (dpf & DP_FUA ? REQ_FUA : 0) |
2159                 (dpf & DP_FLUSH ? REQ_FLUSH : 0) |
2160                 (dpf & DP_DISCARD ? REQ_DISCARD : 0);
2161 }
2162
2163 static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2164                                     unsigned int size)
2165 {
2166         struct drbd_interval *i;
2167
2168     repeat:
2169         drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2170                 struct drbd_request *req;
2171                 struct bio_and_error m;
2172
2173                 if (!i->local)
2174                         continue;
2175                 req = container_of(i, struct drbd_request, i);
2176                 if (!(req->rq_state & RQ_POSTPONED))
2177                         continue;
2178                 req->rq_state &= ~RQ_POSTPONED;
2179                 __req_mod(req, NEG_ACKED, &m);
2180                 spin_unlock_irq(&device->resource->req_lock);
2181                 if (m.bio)
2182                         complete_master_bio(device, &m);
2183                 spin_lock_irq(&device->resource->req_lock);
2184                 goto repeat;
2185         }
2186 }
2187
2188 static int handle_write_conflicts(struct drbd_device *device,
2189                                   struct drbd_peer_request *peer_req)
2190 {
2191         struct drbd_connection *connection = peer_req->peer_device->connection;
2192         bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2193         sector_t sector = peer_req->i.sector;
2194         const unsigned int size = peer_req->i.size;
2195         struct drbd_interval *i;
2196         bool equal;
2197         int err;
2198
2199         /*
2200          * Inserting the peer request into the write_requests tree will prevent
2201          * new conflicting local requests from being added.
2202          */
2203         drbd_insert_interval(&device->write_requests, &peer_req->i);
2204
2205     repeat:
2206         drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2207                 if (i == &peer_req->i)
2208                         continue;
2209                 if (i->completed)
2210                         continue;
2211
2212                 if (!i->local) {
2213                         /*
2214                          * Our peer has sent a conflicting remote request; this
2215                          * should not happen in a two-node setup.  Wait for the
2216                          * earlier peer request to complete.
2217                          */
2218                         err = drbd_wait_misc(device, i);
2219                         if (err)
2220                                 goto out;
2221                         goto repeat;
2222                 }
2223
2224                 equal = i->sector == sector && i->size == size;
2225                 if (resolve_conflicts) {
2226                         /*
2227                          * If the peer request is fully contained within the
2228                          * overlapping request, it can be considered overwritten
2229                          * and thus superseded; otherwise, it will be retried
2230                          * once all overlapping requests have completed.
2231                          */
2232                         bool superseded = i->sector <= sector && i->sector +
2233                                        (i->size >> 9) >= sector + (size >> 9);
2234
2235                         if (!equal)
2236                                 drbd_alert(device, "Concurrent writes detected: "
2237                                                "local=%llus +%u, remote=%llus +%u, "
2238                                                "assuming %s came first\n",
2239                                           (unsigned long long)i->sector, i->size,
2240                                           (unsigned long long)sector, size,
2241                                           superseded ? "local" : "remote");
2242
2243                         peer_req->w.cb = superseded ? e_send_superseded :
2244                                                    e_send_retry_write;
2245                         list_add_tail(&peer_req->w.list, &device->done_ee);
2246                         queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2247
2248                         err = -ENOENT;
2249                         goto out;
2250                 } else {
2251                         struct drbd_request *req =
2252                                 container_of(i, struct drbd_request, i);
2253
2254                         if (!equal)
2255                                 drbd_alert(device, "Concurrent writes detected: "
2256                                                "local=%llus +%u, remote=%llus +%u\n",
2257                                           (unsigned long long)i->sector, i->size,
2258                                           (unsigned long long)sector, size);
2259
2260                         if (req->rq_state & RQ_LOCAL_PENDING ||
2261                             !(req->rq_state & RQ_POSTPONED)) {
2262                                 /*
2263                                  * Wait for the node with the discard flag to
2264                                  * decide if this request has been superseded
2265                                  * or needs to be retried.
2266                                  * Requests that have been superseded will
2267                                  * disappear from the write_requests tree.
2268                                  *
2269                                  * In addition, wait for the conflicting
2270                                  * request to finish locally before submitting
2271                                  * the conflicting peer request.
2272                                  */
2273                                 err = drbd_wait_misc(device, &req->i);
2274                                 if (err) {
2275                                         _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2276                                         fail_postponed_requests(device, sector, size);
2277                                         goto out;
2278                                 }
2279                                 goto repeat;
2280                         }
2281                         /*
2282                          * Remember to restart the conflicting requests after
2283                          * the new peer request has completed.
2284                          */
2285                         peer_req->flags |= EE_RESTART_REQUESTS;
2286                 }
2287         }
2288         err = 0;
2289
2290     out:
2291         if (err)
2292                 drbd_remove_epoch_entry_interval(device, peer_req);
2293         return err;
2294 }
2295
2296 /* mirrored write */
2297 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2298 {
2299         struct drbd_peer_device *peer_device;
2300         struct drbd_device *device;
2301         struct net_conf *nc;
2302         sector_t sector;
2303         struct drbd_peer_request *peer_req;
2304         struct p_data *p = pi->data;
2305         u32 peer_seq = be32_to_cpu(p->seq_num);
2306         int rw = WRITE;
2307         u32 dp_flags;
2308         int err, tp;
2309
2310         peer_device = conn_peer_device(connection, pi->vnr);
2311         if (!peer_device)
2312                 return -EIO;
2313         device = peer_device->device;
2314
2315         if (!get_ldev(device)) {
2316                 int err2;
2317
2318                 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2319                 drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2320                 atomic_inc(&connection->current_epoch->epoch_size);
2321                 err2 = drbd_drain_block(peer_device, pi->size);
2322                 if (!err)
2323                         err = err2;
2324                 return err;
2325         }
2326
2327         /*
2328          * Corresponding put_ldev done either below (on various errors), or in
2329          * drbd_peer_request_endio, if we successfully submit the data at the
2330          * end of this function.
2331          */
2332
2333         sector = be64_to_cpu(p->sector);
2334         peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2335         if (!peer_req) {
2336                 put_ldev(device);
2337                 return -EIO;
2338         }
2339
2340         peer_req->w.cb = e_end_block;
2341         peer_req->submit_jif = jiffies;
2342         peer_req->flags |= EE_APPLICATION;
2343
2344         dp_flags = be32_to_cpu(p->dp_flags);
2345         rw |= wire_flags_to_bio(dp_flags);
2346         if (pi->cmd == P_TRIM) {
2347                 struct request_queue *q = bdev_get_queue(device->ldev->backing_bdev);
2348                 peer_req->flags |= EE_IS_TRIM;
2349                 if (!blk_queue_discard(q))
2350                         peer_req->flags |= EE_IS_TRIM_USE_ZEROOUT;
2351                 D_ASSERT(peer_device, peer_req->i.size > 0);
2352                 D_ASSERT(peer_device, rw & REQ_DISCARD);
2353                 D_ASSERT(peer_device, peer_req->pages == NULL);
2354         } else if (peer_req->pages == NULL) {
2355                 D_ASSERT(device, peer_req->i.size == 0);
2356                 D_ASSERT(device, dp_flags & DP_FLUSH);
2357         }
2358
2359         if (dp_flags & DP_MAY_SET_IN_SYNC)
2360                 peer_req->flags |= EE_MAY_SET_IN_SYNC;
2361
2362         spin_lock(&connection->epoch_lock);
2363         peer_req->epoch = connection->current_epoch;
2364         atomic_inc(&peer_req->epoch->epoch_size);
2365         atomic_inc(&peer_req->epoch->active);
2366         spin_unlock(&connection->epoch_lock);
2367
2368         rcu_read_lock();
2369         nc = rcu_dereference(peer_device->connection->net_conf);
2370         tp = nc->two_primaries;
2371         if (peer_device->connection->agreed_pro_version < 100) {
2372                 switch (nc->wire_protocol) {
2373                 case DRBD_PROT_C:
2374                         dp_flags |= DP_SEND_WRITE_ACK;
2375                         break;
2376                 case DRBD_PROT_B:
2377                         dp_flags |= DP_SEND_RECEIVE_ACK;
2378                         break;
2379                 }
2380         }
2381         rcu_read_unlock();
2382
2383         if (dp_flags & DP_SEND_WRITE_ACK) {
2384                 peer_req->flags |= EE_SEND_WRITE_ACK;
2385                 inc_unacked(device);
2386                 /* corresponding dec_unacked() in e_end_block()
2387                  * respective _drbd_clear_done_ee */
2388         }
2389
2390         if (dp_flags & DP_SEND_RECEIVE_ACK) {
2391                 /* I really don't like it that the receiver thread
2392                  * sends on the msock, but anyways */
2393                 drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2394         }
2395
2396         if (tp) {
2397                 /* two primaries implies protocol C */
2398                 D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2399                 peer_req->flags |= EE_IN_INTERVAL_TREE;
2400                 err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2401                 if (err)
2402                         goto out_interrupted;
2403                 spin_lock_irq(&device->resource->req_lock);
2404                 err = handle_write_conflicts(device, peer_req);
2405                 if (err) {
2406                         spin_unlock_irq(&device->resource->req_lock);
2407                         if (err == -ENOENT) {
2408                                 put_ldev(device);
2409                                 return 0;
2410                         }
2411                         goto out_interrupted;
2412                 }
2413         } else {
2414                 update_peer_seq(peer_device, peer_seq);
2415                 spin_lock_irq(&device->resource->req_lock);
2416         }
2417         /* if we use the zeroout fallback code, we process synchronously
2418          * and we wait for all pending requests, respectively wait for
2419          * active_ee to become empty in drbd_submit_peer_request();
2420          * better not add ourselves here. */
2421         if ((peer_req->flags & EE_IS_TRIM_USE_ZEROOUT) == 0)
2422                 list_add_tail(&peer_req->w.list, &device->active_ee);
2423         spin_unlock_irq(&device->resource->req_lock);
2424
2425         if (device->state.conn == C_SYNC_TARGET)
2426                 wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2427
2428         if (device->state.pdsk < D_INCONSISTENT) {
2429                 /* In case we have the only disk of the cluster, */
2430                 drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2431                 peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2432                 drbd_al_begin_io(device, &peer_req->i);
2433                 peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2434         }
2435
2436         err = drbd_submit_peer_request(device, peer_req, rw, DRBD_FAULT_DT_WR);
2437         if (!err)
2438                 return 0;
2439
2440         /* don't care for the reason here */
2441         drbd_err(device, "submit failed, triggering re-connect\n");
2442         spin_lock_irq(&device->resource->req_lock);
2443         list_del(&peer_req->w.list);
2444         drbd_remove_epoch_entry_interval(device, peer_req);
2445         spin_unlock_irq(&device->resource->req_lock);
2446         if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2447                 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2448                 drbd_al_complete_io(device, &peer_req->i);
2449         }
2450
2451 out_interrupted:
2452         drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT + EV_CLEANUP);
2453         put_ldev(device);
2454         drbd_free_peer_req(device, peer_req);
2455         return err;
2456 }
2457
2458 /* We may throttle resync, if the lower device seems to be busy,
2459  * and current sync rate is above c_min_rate.
2460  *
2461  * To decide whether or not the lower device is busy, we use a scheme similar
2462  * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2463  * (more than 64 sectors) of activity we cannot account for with our own resync
2464  * activity, it obviously is "busy".
2465  *
2466  * The current sync rate used here uses only the most recent two step marks,
2467  * to have a short time average so we can react faster.
2468  */
2469 bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2470                 bool throttle_if_app_is_waiting)
2471 {
2472         struct lc_element *tmp;
2473         bool throttle = drbd_rs_c_min_rate_throttle(device);
2474
2475         if (!throttle || throttle_if_app_is_waiting)
2476                 return throttle;
2477
2478         spin_lock_irq(&device->al_lock);
2479         tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2480         if (tmp) {
2481                 struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2482                 if (test_bit(BME_PRIORITY, &bm_ext->flags))
2483                         throttle = false;
2484                 /* Do not slow down if app IO is already waiting for this extent,
2485                  * and our progress is necessary for application IO to complete. */
2486         }
2487         spin_unlock_irq(&device->al_lock);
2488
2489         return throttle;
2490 }
2491
2492 bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2493 {
2494         struct gendisk *disk = device->ldev->backing_bdev->bd_contains->bd_disk;
2495         unsigned long db, dt, dbdt;
2496         unsigned int c_min_rate;
2497         int curr_events;
2498
2499         rcu_read_lock();
2500         c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2501         rcu_read_unlock();
2502
2503         /* feature disabled? */
2504         if (c_min_rate == 0)
2505                 return false;
2506
2507         curr_events = (int)part_stat_read(&disk->part0, sectors[0]) +
2508                       (int)part_stat_read(&disk->part0, sectors[1]) -
2509                         atomic_read(&device->rs_sect_ev);
2510
2511         if (atomic_read(&device->ap_actlog_cnt)
2512             || curr_events - device->rs_last_events > 64) {
2513                 unsigned long rs_left;
2514                 int i;
2515
2516                 device->rs_last_events = curr_events;
2517
2518                 /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2519                  * approx. */
2520                 i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2521
2522                 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2523                         rs_left = device->ov_left;
2524                 else
2525                         rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2526
2527                 dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2528                 if (!dt)
2529                         dt++;
2530                 db = device->rs_mark_left[i] - rs_left;
2531                 dbdt = Bit2KB(db/dt);
2532
2533                 if (dbdt > c_min_rate)
2534                         return true;
2535         }
2536         return false;
2537 }
2538
2539 static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2540 {
2541         struct drbd_peer_device *peer_device;
2542         struct drbd_device *device;
2543         sector_t sector;
2544         sector_t capacity;
2545         struct drbd_peer_request *peer_req;
2546         struct digest_info *di = NULL;
2547         int size, verb;
2548         unsigned int fault_type;
2549         struct p_block_req *p = pi->data;
2550
2551         peer_device = conn_peer_device(connection, pi->vnr);
2552         if (!peer_device)
2553                 return -EIO;
2554         device = peer_device->device;
2555         capacity = drbd_get_capacity(device->this_bdev);
2556
2557         sector = be64_to_cpu(p->sector);
2558         size   = be32_to_cpu(p->blksize);
2559
2560         if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2561                 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2562                                 (unsigned long long)sector, size);
2563                 return -EINVAL;
2564         }
2565         if (sector + (size>>9) > capacity) {
2566                 drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2567                                 (unsigned long long)sector, size);
2568                 return -EINVAL;
2569         }
2570
2571         if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2572                 verb = 1;
2573                 switch (pi->cmd) {
2574                 case P_DATA_REQUEST:
2575                         drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2576                         break;
2577                 case P_RS_DATA_REQUEST:
2578                 case P_CSUM_RS_REQUEST:
2579                 case P_OV_REQUEST:
2580                         drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2581                         break;
2582                 case P_OV_REPLY:
2583                         verb = 0;
2584                         dec_rs_pending(device);
2585                         drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2586                         break;
2587                 default:
2588                         BUG();
2589                 }
2590                 if (verb && __ratelimit(&drbd_ratelimit_state))
2591                         drbd_err(device, "Can not satisfy peer's read request, "
2592                             "no local data.\n");
2593
2594                 /* drain possibly payload */
2595                 return drbd_drain_block(peer_device, pi->size);
2596         }
2597
2598         /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2599          * "criss-cross" setup, that might cause write-out on some other DRBD,
2600          * which in turn might block on the other node at this very place.  */
2601         peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2602                         true /* has real payload */, GFP_NOIO);
2603         if (!peer_req) {
2604                 put_ldev(device);
2605                 return -ENOMEM;
2606         }
2607
2608         switch (pi->cmd) {
2609         case P_DATA_REQUEST:
2610                 peer_req->w.cb = w_e_end_data_req;
2611                 fault_type = DRBD_FAULT_DT_RD;
2612                 /* application IO, don't drbd_rs_begin_io */
2613                 peer_req->flags |= EE_APPLICATION;
2614                 goto submit;
2615
2616         case P_RS_DATA_REQUEST:
2617                 peer_req->w.cb = w_e_end_rsdata_req;
2618                 fault_type = DRBD_FAULT_RS_RD;
2619                 /* used in the sector offset progress display */
2620                 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2621                 break;
2622
2623         case P_OV_REPLY:
2624         case P_CSUM_RS_REQUEST:
2625                 fault_type = DRBD_FAULT_RS_RD;
2626                 di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2627                 if (!di)
2628                         goto out_free_e;
2629
2630                 di->digest_size = pi->size;
2631                 di->digest = (((char *)di)+sizeof(struct digest_info));
2632
2633                 peer_req->digest = di;
2634                 peer_req->flags |= EE_HAS_DIGEST;
2635
2636                 if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2637                         goto out_free_e;
2638
2639                 if (pi->cmd == P_CSUM_RS_REQUEST) {
2640                         D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2641                         peer_req->w.cb = w_e_end_csum_rs_req;
2642                         /* used in the sector offset progress display */
2643                         device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2644                         /* remember to report stats in drbd_resync_finished */
2645                         device->use_csums = true;
2646                 } else if (pi->cmd == P_OV_REPLY) {
2647                         /* track progress, we may need to throttle */
2648                         atomic_add(size >> 9, &device->rs_sect_in);
2649                         peer_req->w.cb = w_e_end_ov_reply;
2650                         dec_rs_pending(device);
2651                         /* drbd_rs_begin_io done when we sent this request,
2652                          * but accounting still needs to be done. */
2653                         goto submit_for_resync;
2654                 }
2655                 break;
2656
2657         case P_OV_REQUEST:
2658                 if (device->ov_start_sector == ~(sector_t)0 &&
2659                     peer_device->connection->agreed_pro_version >= 90) {
2660                         unsigned long now = jiffies;
2661                         int i;
2662                         device->ov_start_sector = sector;
2663                         device->ov_position = sector;
2664                         device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2665                         device->rs_total = device->ov_left;
2666                         for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2667                                 device->rs_mark_left[i] = device->ov_left;
2668                                 device->rs_mark_time[i] = now;
2669                         }
2670                         drbd_info(device, "Online Verify start sector: %llu\n",
2671                                         (unsigned long long)sector);
2672                 }
2673                 peer_req->w.cb = w_e_end_ov_req;
2674                 fault_type = DRBD_FAULT_RS_RD;
2675                 break;
2676
2677         default:
2678                 BUG();
2679         }
2680
2681         /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2682          * wrt the receiver, but it is not as straightforward as it may seem.
2683          * Various places in the resync start and stop logic assume resync
2684          * requests are processed in order, requeuing this on the worker thread
2685          * introduces a bunch of new code for synchronization between threads.
2686          *
2687          * Unlimited throttling before drbd_rs_begin_io may stall the resync
2688          * "forever", throttling after drbd_rs_begin_io will lock that extent
2689          * for application writes for the same time.  For now, just throttle
2690          * here, where the rest of the code expects the receiver to sleep for
2691          * a while, anyways.
2692          */
2693
2694         /* Throttle before drbd_rs_begin_io, as that locks out application IO;
2695          * this defers syncer requests for some time, before letting at least
2696          * on request through.  The resync controller on the receiving side
2697          * will adapt to the incoming rate accordingly.
2698          *
2699          * We cannot throttle here if remote is Primary/SyncTarget:
2700          * we would also throttle its application reads.
2701          * In that case, throttling is done on the SyncTarget only.
2702          */
2703
2704         /* Even though this may be a resync request, we do add to "read_ee";
2705          * "sync_ee" is only used for resync WRITEs.
2706          * Add to list early, so debugfs can find this request
2707          * even if we have to sleep below. */
2708         spin_lock_irq(&device->resource->req_lock);
2709         list_add_tail(&peer_req->w.list, &device->read_ee);
2710         spin_unlock_irq(&device->resource->req_lock);
2711
2712         update_receiver_timing_details(connection, drbd_rs_should_slow_down);
2713         if (device->state.peer != R_PRIMARY
2714         && drbd_rs_should_slow_down(device, sector, false))
2715                 schedule_timeout_uninterruptible(HZ/10);
2716         update_receiver_timing_details(connection, drbd_rs_begin_io);
2717         if (drbd_rs_begin_io(device, sector))
2718                 goto out_free_e;
2719
2720 submit_for_resync:
2721         atomic_add(size >> 9, &device->rs_sect_ev);
2722
2723 submit:
2724         update_receiver_timing_details(connection, drbd_submit_peer_request);
2725         inc_unacked(device);
2726         if (drbd_submit_peer_request(device, peer_req, READ, fault_type) == 0)
2727                 return 0;
2728
2729         /* don't care for the reason here */
2730         drbd_err(device, "submit failed, triggering re-connect\n");
2731
2732 out_free_e:
2733         spin_lock_irq(&device->resource->req_lock);
2734         list_del(&peer_req->w.list);
2735         spin_unlock_irq(&device->resource->req_lock);
2736         /* no drbd_rs_complete_io(), we are dropping the connection anyways */
2737
2738         put_ldev(device);
2739         drbd_free_peer_req(device, peer_req);
2740         return -EIO;
2741 }
2742
2743 /**
2744  * drbd_asb_recover_0p  -  Recover after split-brain with no remaining primaries
2745  */
2746 static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
2747 {
2748         struct drbd_device *device = peer_device->device;
2749         int self, peer, rv = -100;
2750         unsigned long ch_self, ch_peer;
2751         enum drbd_after_sb_p after_sb_0p;
2752
2753         self = device->ldev->md.uuid[UI_BITMAP] & 1;
2754         peer = device->p_uuid[UI_BITMAP] & 1;
2755
2756         ch_peer = device->p_uuid[UI_SIZE];
2757         ch_self = device->comm_bm_set;
2758
2759         rcu_read_lock();
2760         after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
2761         rcu_read_unlock();
2762         switch (after_sb_0p) {
2763         case ASB_CONSENSUS:
2764         case ASB_DISCARD_SECONDARY:
2765         case ASB_CALL_HELPER:
2766         case ASB_VIOLENTLY:
2767                 drbd_err(device, "Configuration error.\n");
2768                 break;
2769         case ASB_DISCONNECT:
2770                 break;
2771         case ASB_DISCARD_YOUNGER_PRI:
2772                 if (self == 0 && peer == 1) {
2773                         rv = -1;
2774                         break;
2775                 }
2776                 if (self == 1 && peer == 0) {
2777                         rv =  1;
2778                         break;
2779                 }
2780                 /* Else fall through to one of the other strategies... */
2781         case ASB_DISCARD_OLDER_PRI:
2782                 if (self == 0 && peer == 1) {
2783                         rv = 1;
2784                         break;
2785                 }
2786                 if (self == 1 && peer == 0) {
2787                         rv = -1;
2788                         break;
2789                 }
2790                 /* Else fall through to one of the other strategies... */
2791                 drbd_warn(device, "Discard younger/older primary did not find a decision\n"
2792                      "Using discard-least-changes instead\n");
2793         case ASB_DISCARD_ZERO_CHG:
2794                 if (ch_peer == 0 && ch_self == 0) {
2795                         rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2796                                 ? -1 : 1;
2797                         break;
2798                 } else {
2799                         if (ch_peer == 0) { rv =  1; break; }
2800                         if (ch_self == 0) { rv = -1; break; }
2801                 }
2802                 if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
2803                         break;
2804         case ASB_DISCARD_LEAST_CHG:
2805                 if      (ch_self < ch_peer)
2806                         rv = -1;
2807                 else if (ch_self > ch_peer)
2808                         rv =  1;
2809                 else /* ( ch_self == ch_peer ) */
2810                      /* Well, then use something else. */
2811                         rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
2812                                 ? -1 : 1;
2813                 break;
2814         case ASB_DISCARD_LOCAL:
2815                 rv = -1;
2816                 break;
2817         case ASB_DISCARD_REMOTE:
2818                 rv =  1;
2819         }
2820
2821         return rv;
2822 }
2823
2824 /**
2825  * drbd_asb_recover_1p  -  Recover after split-brain with one remaining primary
2826  */
2827 static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
2828 {
2829         struct drbd_device *device = peer_device->device;
2830         int hg, rv = -100;
2831         enum drbd_after_sb_p after_sb_1p;
2832
2833         rcu_read_lock();
2834         after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
2835         rcu_read_unlock();
2836         switch (after_sb_1p) {
2837         case ASB_DISCARD_YOUNGER_PRI:
2838         case ASB_DISCARD_OLDER_PRI:
2839         case ASB_DISCARD_LEAST_CHG:
2840         case ASB_DISCARD_LOCAL:
2841         case ASB_DISCARD_REMOTE:
2842         case ASB_DISCARD_ZERO_CHG:
2843                 drbd_err(device, "Configuration error.\n");
2844                 break;
2845         case ASB_DISCONNECT:
2846                 break;
2847         case ASB_CONSENSUS:
2848                 hg = drbd_asb_recover_0p(peer_device);
2849                 if (hg == -1 && device->state.role == R_SECONDARY)
2850                         rv = hg;
2851                 if (hg == 1  && device->state.role == R_PRIMARY)
2852                         rv = hg;
2853                 break;
2854         case ASB_VIOLENTLY:
2855                 rv = drbd_asb_recover_0p(peer_device);
2856                 break;
2857         case ASB_DISCARD_SECONDARY:
2858                 return device->state.role == R_PRIMARY ? 1 : -1;
2859         case ASB_CALL_HELPER:
2860                 hg = drbd_asb_recover_0p(peer_device);
2861                 if (hg == -1 && device->state.role == R_PRIMARY) {
2862                         enum drbd_state_rv rv2;
2863
2864                          /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2865                           * we might be here in C_WF_REPORT_PARAMS which is transient.
2866                           * we do not need to wait for the after state change work either. */
2867                         rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
2868                         if (rv2 != SS_SUCCESS) {
2869                                 drbd_khelper(device, "pri-lost-after-sb");
2870                         } else {
2871                                 drbd_warn(device, "Successfully gave up primary role.\n");
2872                                 rv = hg;
2873                         }
2874                 } else
2875                         rv = hg;
2876         }
2877
2878         return rv;
2879 }
2880
2881 /**
2882  * drbd_asb_recover_2p  -  Recover after split-brain with two remaining primaries
2883  */
2884 static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
2885 {
2886         struct drbd_device *device = peer_device->device;
2887         int hg, rv = -100;
2888         enum drbd_after_sb_p after_sb_2p;
2889
2890         rcu_read_lock();
2891         after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
2892         rcu_read_unlock();
2893         switch (after_sb_2p) {
2894         case ASB_DISCARD_YOUNGER_PRI:
2895         case ASB_DISCARD_OLDER_PRI:
2896         case ASB_DISCARD_LEAST_CHG:
2897         case ASB_DISCARD_LOCAL:
2898         case ASB_DISCARD_REMOTE:
2899         case ASB_CONSENSUS:
2900         case ASB_DISCARD_SECONDARY:
2901         case ASB_DISCARD_ZERO_CHG:
2902                 drbd_err(device, "Configuration error.\n");
2903                 break;
2904         case ASB_VIOLENTLY:
2905                 rv = drbd_asb_recover_0p(peer_device);
2906                 break;
2907         case ASB_DISCONNECT:
2908                 break;
2909         case ASB_CALL_HELPER:
2910                 hg = drbd_asb_recover_0p(peer_device);
2911                 if (hg == -1) {
2912                         enum drbd_state_rv rv2;
2913
2914                          /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
2915                           * we might be here in C_WF_REPORT_PARAMS which is transient.
2916                           * we do not need to wait for the after state change work either. */
2917                         rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
2918                         if (rv2 != SS_SUCCESS) {
2919                                 drbd_khelper(device, "pri-lost-after-sb");
2920                         } else {
2921                                 drbd_warn(device, "Successfully gave up primary role.\n");
2922                                 rv = hg;
2923                         }
2924                 } else
2925                         rv = hg;
2926         }
2927
2928         return rv;
2929 }
2930
2931 static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
2932                            u64 bits, u64 flags)
2933 {
2934         if (!uuid) {
2935                 drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
2936                 return;
2937         }
2938         drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
2939              text,
2940              (unsigned long long)uuid[UI_CURRENT],
2941              (unsigned long long)uuid[UI_BITMAP],
2942              (unsigned long long)uuid[UI_HISTORY_START],
2943              (unsigned long long)uuid[UI_HISTORY_END],
2944              (unsigned long long)bits,
2945              (unsigned long long)flags);
2946 }
2947
2948 /*
2949   100   after split brain try auto recover
2950     2   C_SYNC_SOURCE set BitMap
2951     1   C_SYNC_SOURCE use BitMap
2952     0   no Sync
2953    -1   C_SYNC_TARGET use BitMap
2954    -2   C_SYNC_TARGET set BitMap
2955  -100   after split brain, disconnect
2956 -1000   unrelated data
2957 -1091   requires proto 91
2958 -1096   requires proto 96
2959  */
2960 static int drbd_uuid_compare(struct drbd_device *const device, int *rule_nr) __must_hold(local)
2961 {
2962         struct drbd_peer_device *const peer_device = first_peer_device(device);
2963         struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
2964         u64 self, peer;
2965         int i, j;
2966
2967         self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
2968         peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
2969
2970         *rule_nr = 10;
2971         if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
2972                 return 0;
2973
2974         *rule_nr = 20;
2975         if ((self == UUID_JUST_CREATED || self == (u64)0) &&
2976              peer != UUID_JUST_CREATED)
2977                 return -2;
2978
2979         *rule_nr = 30;
2980         if (self != UUID_JUST_CREATED &&
2981             (peer == UUID_JUST_CREATED || peer == (u64)0))
2982                 return 2;
2983
2984         if (self == peer) {
2985                 int rct, dc; /* roles at crash time */
2986
2987                 if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
2988
2989                         if (connection->agreed_pro_version < 91)
2990                                 return -1091;
2991
2992                         if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
2993                             (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
2994                                 drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
2995                                 drbd_uuid_move_history(device);
2996                                 device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
2997                                 device->ldev->md.uuid[UI_BITMAP] = 0;
2998
2999                                 drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3000                                                device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3001                                 *rule_nr = 34;
3002                         } else {
3003                                 drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3004                                 *rule_nr = 36;
3005                         }
3006
3007                         return 1;
3008                 }
3009
3010                 if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3011
3012                         if (connection->agreed_pro_version < 91)
3013                                 return -1091;
3014
3015                         if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3016                             (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3017                                 drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3018
3019                                 device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3020                                 device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3021                                 device->p_uuid[UI_BITMAP] = 0UL;
3022
3023                                 drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3024                                 *rule_nr = 35;
3025                         } else {
3026                                 drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3027                                 *rule_nr = 37;
3028                         }
3029
3030                         return -1;
3031                 }
3032
3033                 /* Common power [off|failure] */
3034                 rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3035                         (device->p_uuid[UI_FLAGS] & 2);
3036                 /* lowest bit is set when we were primary,
3037                  * next bit (weight 2) is set when peer was primary */
3038                 *rule_nr = 40;
3039
3040                 switch (rct) {
3041                 case 0: /* !self_pri && !peer_pri */ return 0;
3042                 case 1: /*  self_pri && !peer_pri */ return 1;
3043                 case 2: /* !self_pri &&  peer_pri */ return -1;
3044                 case 3: /*  self_pri &&  peer_pri */
3045                         dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3046                         return dc ? -1 : 1;
3047                 }
3048         }
3049
3050         *rule_nr = 50;
3051         peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3052         if (self == peer)
3053                 return -1;
3054
3055         *rule_nr = 51;
3056         peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3057         if (self == peer) {
3058                 if (connection->agreed_pro_version < 96 ?
3059                     (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3060                     (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3061                     peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3062                         /* The last P_SYNC_UUID did not get though. Undo the last start of
3063                            resync as sync source modifications of the peer's UUIDs. */
3064
3065                         if (connection->agreed_pro_version < 91)
3066                                 return -1091;
3067
3068                         device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3069                         device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3070
3071                         drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3072                         drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3073
3074                         return -1;
3075                 }
3076         }
3077
3078         *rule_nr = 60;
3079         self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3080         for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3081                 peer = device->p_uuid[i] & ~((u64)1);
3082                 if (self == peer)
3083                         return -2;
3084         }
3085
3086         *rule_nr = 70;
3087         self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3088         peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3089         if (self == peer)
3090                 return 1;
3091
3092         *rule_nr = 71;
3093         self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3094         if (self == peer) {
3095                 if (connection->agreed_pro_version < 96 ?
3096                     (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3097                     (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3098                     self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3099                         /* The last P_SYNC_UUID did not get though. Undo the last start of
3100                            resync as sync source modifications of our UUIDs. */
3101
3102                         if (connection->agreed_pro_version < 91)
3103                                 return -1091;
3104
3105                         __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3106                         __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3107
3108                         drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3109                         drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3110                                        device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3111
3112                         return 1;
3113                 }
3114         }
3115
3116
3117         *rule_nr = 80;
3118         peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3119         for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3120                 self = device->ldev->md.uuid[i] & ~((u64)1);
3121                 if (self == peer)
3122                         return 2;
3123         }
3124
3125         *rule_nr = 90;
3126         self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3127         peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3128         if (self == peer && self != ((u64)0))
3129                 return 100;
3130
3131         *rule_nr = 100;
3132         for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3133                 self = device->ldev->md.uuid[i] & ~((u64)1);
3134                 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3135                         peer = device->p_uuid[j] & ~((u64)1);
3136                         if (self == peer)
3137                                 return -100;
3138                 }
3139         }
3140
3141         return -1000;
3142 }
3143
3144 /* drbd_sync_handshake() returns the new conn state on success, or
3145    CONN_MASK (-1) on failure.
3146  */
3147 static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3148                                            enum drbd_role peer_role,
3149                                            enum drbd_disk_state peer_disk) __must_hold(local)
3150 {
3151         struct drbd_device *device = peer_device->device;
3152         enum drbd_conns rv = C_MASK;
3153         enum drbd_disk_state mydisk;
3154         struct net_conf *nc;
3155         int hg, rule_nr, rr_conflict, tentative;
3156
3157         mydisk = device->state.disk;
3158         if (mydisk == D_NEGOTIATING)
3159                 mydisk = device->new_state_tmp.disk;
3160
3161         drbd_info(device, "drbd_sync_handshake:\n");
3162
3163         spin_lock_irq(&device->ldev->md.uuid_lock);
3164         drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3165         drbd_uuid_dump(device, "peer", device->p_uuid,
3166                        device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3167
3168         hg = drbd_uuid_compare(device, &rule_nr);
3169         spin_unlock_irq(&device->ldev->md.uuid_lock);
3170
3171         drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3172
3173         if (hg == -1000) {
3174                 drbd_alert(device, "Unrelated data, aborting!\n");
3175                 return C_MASK;
3176         }
3177         if (hg < -1000) {
3178                 drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3179                 return C_MASK;
3180         }
3181
3182         if    ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3183             (peer_disk == D_INCONSISTENT && mydisk    > D_INCONSISTENT)) {
3184                 int f = (hg == -100) || abs(hg) == 2;
3185                 hg = mydisk > D_INCONSISTENT ? 1 : -1;
3186                 if (f)
3187                         hg = hg*2;
3188                 drbd_info(device, "Becoming sync %s due to disk states.\n",
3189                      hg > 0 ? "source" : "target");
3190         }
3191
3192         if (abs(hg) == 100)
3193                 drbd_khelper(device, "initial-split-brain");
3194
3195         rcu_read_lock();
3196         nc = rcu_dereference(peer_device->connection->net_conf);
3197
3198         if (hg == 100 || (hg == -100 && nc->always_asbp)) {
3199                 int pcount = (device->state.role == R_PRIMARY)
3200                            + (peer_role == R_PRIMARY);
3201                 int forced = (hg == -100);
3202
3203                 switch (pcount) {
3204                 case 0:
3205                         hg = drbd_asb_recover_0p(peer_device);
3206                         break;
3207                 case 1:
3208                         hg = drbd_asb_recover_1p(peer_device);
3209                         break;
3210                 case 2:
3211                         hg = drbd_asb_recover_2p(peer_device);
3212                         break;
3213                 }
3214                 if (abs(hg) < 100) {
3215                         drbd_warn(device, "Split-Brain detected, %d primaries, "
3216                              "automatically solved. Sync from %s node\n",
3217                              pcount, (hg < 0) ? "peer" : "this");
3218                         if (forced) {
3219                                 drbd_warn(device, "Doing a full sync, since"
3220                                      " UUIDs where ambiguous.\n");
3221                                 hg = hg*2;
3222                         }
3223                 }
3224         }
3225
3226         if (hg == -100) {
3227                 if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3228                         hg = -1;
3229                 if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3230                         hg = 1;
3231
3232                 if (abs(hg) < 100)
3233                         drbd_warn(device, "Split-Brain detected, manually solved. "
3234                              "Sync from %s node\n",
3235                              (hg < 0) ? "peer" : "this");
3236         }
3237         rr_conflict = nc->rr_conflict;
3238         tentative = nc->tentative;
3239         rcu_read_unlock();
3240
3241         if (hg == -100) {
3242                 /* FIXME this log message is not correct if we end up here
3243                  * after an attempted attach on a diskless node.
3244                  * We just refuse to attach -- well, we drop the "connection"
3245                  * to that disk, in a way... */
3246                 drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3247                 drbd_khelper(device, "split-brain");
3248                 return C_MASK;
3249         }
3250
3251         if (hg > 0 && mydisk <= D_INCONSISTENT) {
3252                 drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3253                 return C_MASK;
3254         }
3255
3256         if (hg < 0 && /* by intention we do not use mydisk here. */
3257             device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3258                 switch (rr_conflict) {
3259                 case ASB_CALL_HELPER:
3260                         drbd_khelper(device, "pri-lost");
3261                         /* fall through */
3262                 case ASB_DISCONNECT:
3263                         drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3264                         return C_MASK;
3265                 case ASB_VIOLENTLY:
3266                         drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3267                              "assumption\n");
3268                 }
3269         }
3270
3271         if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3272                 if (hg == 0)
3273                         drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3274                 else
3275                         drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3276                                  drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3277                                  abs(hg) >= 2 ? "full" : "bit-map based");
3278                 return C_MASK;
3279         }
3280
3281         if (abs(hg) >= 2) {
3282                 drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3283                 if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3284                                         BM_LOCKED_SET_ALLOWED))
3285                         return C_MASK;
3286         }
3287
3288         if (hg > 0) { /* become sync source. */
3289                 rv = C_WF_BITMAP_S;
3290         } else if (hg < 0) { /* become sync target */
3291                 rv = C_WF_BITMAP_T;
3292         } else {
3293                 rv = C_CONNECTED;
3294                 if (drbd_bm_total_weight(device)) {
3295                         drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3296                              drbd_bm_total_weight(device));
3297                 }
3298         }
3299
3300         return rv;
3301 }
3302
3303 static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3304 {
3305         /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3306         if (peer == ASB_DISCARD_REMOTE)
3307                 return ASB_DISCARD_LOCAL;
3308
3309         /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3310         if (peer == ASB_DISCARD_LOCAL)
3311                 return ASB_DISCARD_REMOTE;
3312
3313         /* everything else is valid if they are equal on both sides. */
3314         return peer;
3315 }
3316
3317 static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3318 {
3319         struct p_protocol *p = pi->data;
3320         enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3321         int p_proto, p_discard_my_data, p_two_primaries, cf;
3322         struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3323         char integrity_alg[SHARED_SECRET_MAX] = "";
3324         struct crypto_ahash *peer_integrity_tfm = NULL;
3325         void *int_dig_in = NULL, *int_dig_vv = NULL;
3326
3327         p_proto         = be32_to_cpu(p->protocol);
3328         p_after_sb_0p   = be32_to_cpu(p->after_sb_0p);
3329         p_after_sb_1p   = be32_to_cpu(p->after_sb_1p);
3330         p_after_sb_2p   = be32_to_cpu(p->after_sb_2p);
3331         p_two_primaries = be32_to_cpu(p->two_primaries);
3332         cf              = be32_to_cpu(p->conn_flags);
3333         p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3334
3335         if (connection->agreed_pro_version >= 87) {
3336                 int err;
3337
3338                 if (pi->size > sizeof(integrity_alg))
3339                         return -EIO;
3340                 err = drbd_recv_all(connection, integrity_alg, pi->size);
3341                 if (err)
3342                         return err;
3343                 integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3344         }
3345
3346         if (pi->cmd != P_PROTOCOL_UPDATE) {
3347                 clear_bit(CONN_DRY_RUN, &connection->flags);
3348
3349                 if (cf & CF_DRY_RUN)
3350                         set_bit(CONN_DRY_RUN, &connection->flags);
3351
3352                 rcu_read_lock();
3353                 nc = rcu_dereference(connection->net_conf);
3354
3355                 if (p_proto != nc->wire_protocol) {
3356                         drbd_err(connection, "incompatible %s settings\n", "protocol");
3357                         goto disconnect_rcu_unlock;
3358                 }
3359
3360                 if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3361                         drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3362                         goto disconnect_rcu_unlock;
3363                 }
3364
3365                 if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3366                         drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3367                         goto disconnect_rcu_unlock;
3368                 }
3369
3370                 if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3371                         drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3372                         goto disconnect_rcu_unlock;
3373                 }
3374
3375                 if (p_discard_my_data && nc->discard_my_data) {
3376                         drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3377                         goto disconnect_rcu_unlock;
3378                 }
3379
3380                 if (p_two_primaries != nc->two_primaries) {
3381                         drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3382                         goto disconnect_rcu_unlock;
3383                 }
3384
3385                 if (strcmp(integrity_alg, nc->integrity_alg)) {
3386                         drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3387                         goto disconnect_rcu_unlock;
3388                 }
3389
3390                 rcu_read_unlock();
3391         }
3392
3393         if (integrity_alg[0]) {
3394                 int hash_size;
3395
3396                 /*
3397                  * We can only change the peer data integrity algorithm
3398                  * here.  Changing our own data integrity algorithm
3399                  * requires that we send a P_PROTOCOL_UPDATE packet at
3400                  * the same time; otherwise, the peer has no way to
3401                  * tell between which packets the algorithm should
3402                  * change.
3403                  */
3404
3405                 peer_integrity_tfm = crypto_alloc_ahash(integrity_alg, 0, CRYPTO_ALG_ASYNC);
3406                 if (!peer_integrity_tfm) {
3407                         drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3408                                  integrity_alg);
3409                         goto disconnect;
3410                 }
3411
3412                 hash_size = crypto_ahash_digestsize(peer_integrity_tfm);
3413                 int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3414                 int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3415                 if (!(int_dig_in && int_dig_vv)) {
3416                         drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3417                         goto disconnect;
3418                 }
3419         }
3420
3421         new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3422         if (!new_net_conf) {
3423                 drbd_err(connection, "Allocation of new net_conf failed\n");
3424                 goto disconnect;
3425         }
3426
3427         mutex_lock(&connection->data.mutex);
3428         mutex_lock(&connection->resource->conf_update);
3429         old_net_conf = connection->net_conf;
3430         *new_net_conf = *old_net_conf;
3431
3432         new_net_conf->wire_protocol = p_proto;
3433         new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3434         new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3435         new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3436         new_net_conf->two_primaries = p_two_primaries;
3437
3438         rcu_assign_pointer(connection->net_conf, new_net_conf);
3439         mutex_unlock(&connection->resource->conf_update);
3440         mutex_unlock(&connection->data.mutex);
3441
3442         crypto_free_ahash(connection->peer_integrity_tfm);
3443         kfree(connection->int_dig_in);
3444         kfree(connection->int_dig_vv);
3445         connection->peer_integrity_tfm = peer_integrity_tfm;
3446         connection->int_dig_in = int_dig_in;
3447         connection->int_dig_vv = int_dig_vv;
3448
3449         if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3450                 drbd_info(connection, "peer data-integrity-alg: %s\n",
3451                           integrity_alg[0] ? integrity_alg : "(none)");
3452
3453         synchronize_rcu();
3454         kfree(old_net_conf);
3455         return 0;
3456
3457 disconnect_rcu_unlock:
3458         rcu_read_unlock();
3459 disconnect:
3460         crypto_free_ahash(peer_integrity_tfm);
3461         kfree(int_dig_in);
3462         kfree(int_dig_vv);
3463         conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3464         return -EIO;
3465 }
3466
3467 /* helper function
3468  * input: alg name, feature name
3469  * return: NULL (alg name was "")
3470  *         ERR_PTR(error) if something goes wrong
3471  *         or the crypto hash ptr, if it worked out ok. */
3472 static struct crypto_ahash *drbd_crypto_alloc_digest_safe(const struct drbd_device *device,
3473                 const char *alg, const char *name)
3474 {
3475         struct crypto_ahash *tfm;
3476
3477         if (!alg[0])
3478                 return NULL;
3479
3480         tfm = crypto_alloc_ahash(alg, 0, CRYPTO_ALG_ASYNC);
3481         if (IS_ERR(tfm)) {
3482                 drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3483                         alg, name, PTR_ERR(tfm));
3484                 return tfm;
3485         }
3486         return tfm;
3487 }
3488
3489 static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3490 {
3491         void *buffer = connection->data.rbuf;
3492         int size = pi->size;
3493
3494         while (size) {
3495                 int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3496                 s = drbd_recv(connection, buffer, s);
3497                 if (s <= 0) {
3498                         if (s < 0)
3499                                 return s;
3500                         break;
3501                 }
3502                 size -= s;
3503         }
3504         if (size)
3505                 return -EIO;
3506         return 0;
3507 }
3508
3509 /*
3510  * config_unknown_volume  -  device configuration command for unknown volume
3511  *
3512  * When a device is added to an existing connection, the node on which the
3513  * device is added first will send configuration commands to its peer but the
3514  * peer will not know about the device yet.  It will warn and ignore these
3515  * commands.  Once the device is added on the second node, the second node will
3516  * send the same device configuration commands, but in the other direction.
3517  *
3518  * (We can also end up here if drbd is misconfigured.)
3519  */
3520 static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3521 {
3522         drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3523                   cmdname(pi->cmd), pi->vnr);
3524         return ignore_remaining_packet(connection, pi);
3525 }
3526
3527 static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3528 {
3529         struct drbd_peer_device *peer_device;
3530         struct drbd_device *device;
3531         struct p_rs_param_95 *p;
3532         unsigned int header_size, data_size, exp_max_sz;
3533         struct crypto_ahash *verify_tfm = NULL;
3534         struct crypto_ahash *csums_tfm = NULL;
3535         struct net_conf *old_net_conf, *new_net_conf = NULL;
3536         struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3537         const int apv = connection->agreed_pro_version;
3538         struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3539         int fifo_size = 0;
3540         int err;
3541
3542         peer_device = conn_peer_device(connection, pi->vnr);
3543         if (!peer_device)
3544                 return config_unknown_volume(connection, pi);
3545         device = peer_device->device;
3546
3547         exp_max_sz  = apv <= 87 ? sizeof(struct p_rs_param)
3548                     : apv == 88 ? sizeof(struct p_rs_param)
3549                                         + SHARED_SECRET_MAX
3550                     : apv <= 94 ? sizeof(struct p_rs_param_89)
3551                     : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3552
3553         if (pi->size > exp_max_sz) {
3554                 drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3555                     pi->size, exp_max_sz);
3556                 return -EIO;
3557         }
3558
3559         if (apv <= 88) {
3560                 header_size = sizeof(struct p_rs_param);
3561                 data_size = pi->size - header_size;
3562         } else if (apv <= 94) {
3563                 header_size = sizeof(struct p_rs_param_89);
3564                 data_size = pi->size - header_size;
3565                 D_ASSERT(device, data_size == 0);
3566         } else {
3567                 header_size = sizeof(struct p_rs_param_95);
3568                 data_size = pi->size - header_size;
3569                 D_ASSERT(device, data_size == 0);
3570         }
3571
3572         /* initialize verify_alg and csums_alg */
3573         p = pi->data;
3574         memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
3575
3576         err = drbd_recv_all(peer_device->connection, p, header_size);
3577         if (err)
3578                 return err;
3579
3580         mutex_lock(&connection->resource->conf_update);
3581         old_net_conf = peer_device->connection->net_conf;
3582         if (get_ldev(device)) {
3583                 new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3584                 if (!new_disk_conf) {
3585                         put_ldev(device);
3586                         mutex_unlock(&connection->resource->conf_update);
3587                         drbd_err(device, "Allocation of new disk_conf failed\n");
3588                         return -ENOMEM;
3589                 }
3590
3591                 old_disk_conf = device->ldev->disk_conf;
3592                 *new_disk_conf = *old_disk_conf;
3593
3594                 new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3595         }
3596
3597         if (apv >= 88) {
3598                 if (apv == 88) {
3599                         if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3600                                 drbd_err(device, "verify-alg of wrong size, "
3601                                         "peer wants %u, accepting only up to %u byte\n",
3602                                         data_size, SHARED_SECRET_MAX);
3603                                 err = -EIO;
3604                                 goto reconnect;
3605                         }
3606
3607                         err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3608                         if (err)
3609                                 goto reconnect;
3610                         /* we expect NUL terminated string */
3611                         /* but just in case someone tries to be evil */
3612                         D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3613                         p->verify_alg[data_size-1] = 0;
3614
3615                 } else /* apv >= 89 */ {
3616                         /* we still expect NUL terminated strings */
3617                         /* but just in case someone tries to be evil */
3618                         D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3619                         D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3620                         p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3621                         p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3622                 }
3623
3624                 if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3625                         if (device->state.conn == C_WF_REPORT_PARAMS) {
3626                                 drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3627                                     old_net_conf->verify_alg, p->verify_alg);
3628                                 goto disconnect;
3629                         }
3630                         verify_tfm = drbd_crypto_alloc_digest_safe(device,
3631                                         p->verify_alg, "verify-alg");
3632                         if (IS_ERR(verify_tfm)) {
3633                                 verify_tfm = NULL;
3634                                 goto disconnect;
3635                         }
3636                 }
3637
3638                 if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3639                         if (device->state.conn == C_WF_REPORT_PARAMS) {
3640                                 drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3641                                     old_net_conf->csums_alg, p->csums_alg);
3642                                 goto disconnect;
3643                         }
3644                         csums_tfm = drbd_crypto_alloc_digest_safe(device,
3645                                         p->csums_alg, "csums-alg");
3646                         if (IS_ERR(csums_tfm)) {
3647                                 csums_tfm = NULL;
3648                                 goto disconnect;
3649                         }
3650                 }
3651
3652                 if (apv > 94 && new_disk_conf) {
3653                         new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
3654                         new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
3655                         new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
3656                         new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
3657
3658                         fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
3659                         if (fifo_size != device->rs_plan_s->size) {
3660                                 new_plan = fifo_alloc(fifo_size);
3661                                 if (!new_plan) {
3662                                         drbd_err(device, "kmalloc of fifo_buffer failed");
3663                                         put_ldev(device);
3664                                         goto disconnect;
3665                                 }
3666                         }
3667                 }
3668
3669                 if (verify_tfm || csums_tfm) {
3670                         new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
3671                         if (!new_net_conf) {
3672                                 drbd_err(device, "Allocation of new net_conf failed\n");
3673                                 goto disconnect;
3674                         }
3675
3676                         *new_net_conf = *old_net_conf;
3677
3678                         if (verify_tfm) {
3679                                 strcpy(new_net_conf->verify_alg, p->verify_alg);
3680                                 new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
3681                                 crypto_free_ahash(peer_device->connection->verify_tfm);
3682                                 peer_device->connection->verify_tfm = verify_tfm;
3683                                 drbd_info(device, "using verify-alg: \"%s\"\n", p->verify_alg);
3684                         }
3685                         if (csums_tfm) {
3686                                 strcpy(new_net_conf->csums_alg, p->csums_alg);
3687                                 new_net_conf->csums_alg_len = strlen(p->csums_alg) + 1;
3688                                 crypto_free_ahash(peer_device->connection->csums_tfm);
3689                                 peer_device->connection->csums_tfm = csums_tfm;
3690                                 drbd_info(device, "using csums-alg: \"%s\"\n", p->csums_alg);
3691                         }
3692                         rcu_assign_pointer(connection->net_conf, new_net_conf);
3693                 }
3694         }
3695
3696         if (new_disk_conf) {
3697                 rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3698                 put_ldev(device);
3699         }
3700
3701         if (new_plan) {
3702                 old_plan = device->rs_plan_s;
3703                 rcu_assign_pointer(device->rs_plan_s, new_plan);
3704         }
3705
3706         mutex_unlock(&connection->resource->conf_update);
3707         synchronize_rcu();
3708         if (new_net_conf)
3709                 kfree(old_net_conf);
3710         kfree(old_disk_conf);
3711         kfree(old_plan);
3712
3713         return 0;
3714
3715 reconnect:
3716         if (new_disk_conf) {
3717                 put_ldev(device);
3718                 kfree(new_disk_conf);
3719         }
3720         mutex_unlock(&connection->resource->conf_update);
3721         return -EIO;
3722
3723 disconnect:
3724         kfree(new_plan);
3725         if (new_disk_conf) {
3726                 put_ldev(device);
3727                 kfree(new_disk_conf);
3728         }
3729         mutex_unlock(&connection->resource->conf_update);
3730         /* just for completeness: actually not needed,
3731          * as this is not reached if csums_tfm was ok. */
3732         crypto_free_ahash(csums_tfm);
3733         /* but free the verify_tfm again, if csums_tfm did not work out */
3734         crypto_free_ahash(verify_tfm);
3735         conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3736         return -EIO;
3737 }
3738
3739 /* warn if the arguments differ by more than 12.5% */
3740 static void warn_if_differ_considerably(struct drbd_device *device,
3741         const char *s, sector_t a, sector_t b)
3742 {
3743         sector_t d;
3744         if (a == 0 || b == 0)
3745                 return;
3746         d = (a > b) ? (a - b) : (b - a);
3747         if (d > (a>>3) || d > (b>>3))
3748                 drbd_warn(device, "Considerable difference in %s: %llus vs. %llus\n", s,
3749                      (unsigned long long)a, (unsigned long long)b);
3750 }
3751
3752 static int receive_sizes(struct drbd_connection *connection, struct packet_info *pi)
3753 {
3754         struct drbd_peer_device *peer_device;
3755         struct drbd_device *device;
3756         struct p_sizes *p = pi->data;
3757         enum determine_dev_size dd = DS_UNCHANGED;
3758         sector_t p_size, p_usize, p_csize, my_usize;
3759         int ldsc = 0; /* local disk size changed */
3760         enum dds_flags ddsf;
3761
3762         peer_device = conn_peer_device(connection, pi->vnr);
3763         if (!peer_device)
3764                 return config_unknown_volume(connection, pi);
3765         device = peer_device->device;
3766
3767         p_size = be64_to_cpu(p->d_size);
3768         p_usize = be64_to_cpu(p->u_size);
3769         p_csize = be64_to_cpu(p->c_size);
3770
3771         /* just store the peer's disk size for now.
3772          * we still need to figure out whether we accept that. */
3773         device->p_size = p_size;
3774
3775         if (get_ldev(device)) {
3776                 rcu_read_lock();
3777                 my_usize = rcu_dereference(device->ldev->disk_conf)->disk_size;
3778                 rcu_read_unlock();
3779
3780                 warn_if_differ_considerably(device, "lower level device sizes",
3781                            p_size, drbd_get_max_capacity(device->ldev));
3782                 warn_if_differ_considerably(device, "user requested size",
3783                                             p_usize, my_usize);
3784
3785                 /* if this is the first connect, or an otherwise expected
3786                  * param exchange, choose the minimum */
3787                 if (device->state.conn == C_WF_REPORT_PARAMS)
3788                         p_usize = min_not_zero(my_usize, p_usize);
3789
3790                 /* Never shrink a device with usable data during connect.
3791                    But allow online shrinking if we are connected. */
3792                 if (drbd_new_dev_size(device, device->ldev, p_usize, 0) <
3793                     drbd_get_capacity(device->this_bdev) &&
3794                     device->state.disk >= D_OUTDATED &&
3795                     device->state.conn < C_CONNECTED) {
3796                         drbd_err(device, "The peer's disk size is too small!\n");
3797                         conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3798                         put_ldev(device);
3799                         return -EIO;
3800                 }
3801
3802                 if (my_usize != p_usize) {
3803                         struct disk_conf *old_disk_conf, *new_disk_conf = NULL;
3804
3805                         new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3806                         if (!new_disk_conf) {
3807                                 drbd_err(device, "Allocation of new disk_conf failed\n");
3808                                 put_ldev(device);
3809                                 return -ENOMEM;
3810                         }
3811
3812                         mutex_lock(&connection->resource->conf_update);
3813                         old_disk_conf = device->ldev->disk_conf;
3814                         *new_disk_conf = *old_disk_conf;
3815                         new_disk_conf->disk_size = p_usize;
3816
3817                         rcu_assign_pointer(device->ldev->disk_conf, new_disk_conf);
3818                         mutex_unlock(&connection->resource->conf_update);
3819                         synchronize_rcu();
3820                         kfree(old_disk_conf);
3821
3822                         drbd_info(device, "Peer sets u_size to %lu sectors\n",
3823                                  (unsigned long)my_usize);
3824                 }
3825
3826                 put_ldev(device);
3827         }
3828
3829         device->peer_max_bio_size = be32_to_cpu(p->max_bio_size);
3830         /* Leave drbd_reconsider_max_bio_size() before drbd_determine_dev_size().
3831            In case we cleared the QUEUE_FLAG_DISCARD from our queue in
3832            drbd_reconsider_max_bio_size(), we can be sure that after
3833            drbd_determine_dev_size() no REQ_DISCARDs are in the queue. */
3834
3835         ddsf = be16_to_cpu(p->dds_flags);
3836         if (get_ldev(device)) {
3837                 drbd_reconsider_max_bio_size(device, device->ldev);
3838                 dd = drbd_determine_dev_size(device, ddsf, NULL);
3839                 put_ldev(device);
3840                 if (dd == DS_ERROR)
3841                         return -EIO;
3842                 drbd_md_sync(device);
3843         } else {
3844                 /*
3845                  * I am diskless, need to accept the peer's *current* size.
3846                  * I must NOT accept the peers backing disk size,
3847                  * it may have been larger than mine all along...
3848                  *
3849                  * At this point, the peer knows more about my disk, or at
3850                  * least about what we last agreed upon, than myself.
3851                  * So if his c_size is less than his d_size, the most likely
3852                  * reason is that *my* d_size was smaller last time we checked.
3853                  *
3854                  * However, if he sends a zero current size,
3855                  * take his (user-capped or) backing disk size anyways.
3856                  */
3857                 drbd_reconsider_max_bio_size(device, NULL);
3858                 drbd_set_my_capacity(device, p_csize ?: p_usize ?: p_size);
3859         }
3860
3861         if (get_ldev(device)) {
3862                 if (device->ldev->known_size != drbd_get_capacity(device->ldev->backing_bdev)) {
3863                         device->ldev->known_size = drbd_get_capacity(device->ldev->backing_bdev);
3864                         ldsc = 1;
3865                 }
3866
3867                 put_ldev(device);
3868         }
3869
3870         if (device->state.conn > C_WF_REPORT_PARAMS) {
3871                 if (be64_to_cpu(p->c_size) !=
3872                     drbd_get_capacity(device->this_bdev) || ldsc) {
3873                         /* we have different sizes, probably peer
3874                          * needs to know my new size... */
3875                         drbd_send_sizes(peer_device, 0, ddsf);
3876                 }
3877                 if (test_and_clear_bit(RESIZE_PENDING, &device->flags) ||
3878                     (dd == DS_GREW && device->state.conn == C_CONNECTED)) {
3879                         if (device->state.pdsk >= D_INCONSISTENT &&
3880                             device->state.disk >= D_INCONSISTENT) {
3881                                 if (ddsf & DDSF_NO_RESYNC)
3882                                         drbd_info(device, "Resync of new storage suppressed with --assume-clean\n");
3883                                 else
3884                                         resync_after_online_grow(device);
3885                         } else
3886                                 set_bit(RESYNC_AFTER_NEG, &device->flags);
3887                 }
3888         }
3889
3890         return 0;
3891 }
3892
3893 static int receive_uuids(struct drbd_connection *connection, struct packet_info *pi)
3894 {
3895         struct drbd_peer_device *peer_device;
3896         struct drbd_device *device;
3897         struct p_uuids *p = pi->data;
3898         u64 *p_uuid;
3899         int i, updated_uuids = 0;
3900
3901         peer_device = conn_peer_device(connection, pi->vnr);
3902         if (!peer_device)
3903                 return config_unknown_volume(connection, pi);
3904         device = peer_device->device;
3905
3906         p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO);
3907         if (!p_uuid) {
3908                 drbd_err(device, "kmalloc of p_uuid failed\n");
3909                 return false;
3910         }
3911
3912         for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++)
3913                 p_uuid[i] = be64_to_cpu(p->uuid[i]);
3914
3915         kfree(device->p_uuid);
3916         device->p_uuid = p_uuid;
3917
3918         if (device->state.conn < C_CONNECTED &&
3919             device->state.disk < D_INCONSISTENT &&
3920             device->state.role == R_PRIMARY &&
3921             (device->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) {
3922                 drbd_err(device, "Can only connect to data with current UUID=%016llX\n",
3923                     (unsigned long long)device->ed_uuid);
3924                 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
3925                 return -EIO;
3926         }
3927
3928         if (get_ldev(device)) {
3929                 int skip_initial_sync =
3930                         device->state.conn == C_CONNECTED &&
3931                         peer_device->connection->agreed_pro_version >= 90 &&
3932                         device->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED &&
3933                         (p_uuid[UI_FLAGS] & 8);
3934                 if (skip_initial_sync) {
3935                         drbd_info(device, "Accepted new current UUID, preparing to skip initial sync\n");
3936                         drbd_bitmap_io(device, &drbd_bmio_clear_n_write,
3937                                         "clear_n_write from receive_uuids",
3938                                         BM_LOCKED_TEST_ALLOWED);
3939                         _drbd_uuid_set(device, UI_CURRENT, p_uuid[UI_CURRENT]);
3940                         _drbd_uuid_set(device, UI_BITMAP, 0);
3941                         _drbd_set_state(_NS2(device, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE),
3942                                         CS_VERBOSE, NULL);
3943                         drbd_md_sync(device);
3944                         updated_uuids = 1;
3945                 }
3946                 put_ldev(device);
3947         } else if (device->state.disk < D_INCONSISTENT &&
3948                    device->state.role == R_PRIMARY) {
3949                 /* I am a diskless primary, the peer just created a new current UUID
3950                    for me. */
3951                 updated_uuids = drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
3952         }
3953
3954         /* Before we test for the disk state, we should wait until an eventually
3955            ongoing cluster wide state change is finished. That is important if
3956            we are primary and are detaching from our disk. We need to see the
3957            new disk state... */
3958         mutex_lock(device->state_mutex);
3959         mutex_unlock(device->state_mutex);
3960         if (device->state.conn >= C_CONNECTED && device->state.disk < D_INCONSISTENT)
3961                 updated_uuids |= drbd_set_ed_uuid(device, p_uuid[UI_CURRENT]);
3962
3963         if (updated_uuids)
3964                 drbd_print_uuids(device, "receiver updated UUIDs to");
3965
3966         return 0;
3967 }
3968
3969 /**
3970  * convert_state() - Converts the peer's view of the cluster state to our point of view
3971  * @ps:         The state as seen by the peer.
3972  */
3973 static union drbd_state convert_state(union drbd_state ps)
3974 {
3975         union drbd_state ms;
3976
3977         static enum drbd_conns c_tab[] = {
3978                 [C_WF_REPORT_PARAMS] = C_WF_REPORT_PARAMS,
3979                 [C_CONNECTED] = C_CONNECTED,
3980
3981                 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T,
3982                 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S,
3983                 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */
3984                 [C_VERIFY_S]       = C_VERIFY_T,
3985                 [C_MASK]   = C_MASK,
3986         };
3987
3988         ms.i = ps.i;
3989
3990         ms.conn = c_tab[ps.conn];
3991         ms.peer = ps.role;
3992         ms.role = ps.peer;
3993         ms.pdsk = ps.disk;
3994         ms.disk = ps.pdsk;
3995         ms.peer_isp = (ps.aftr_isp | ps.user_isp);
3996
3997         return ms;
3998 }
3999
4000 static int receive_req_state(struct drbd_connection *connection, struct packet_info *pi)
4001 {
4002         struct drbd_peer_device *peer_device;
4003         struct drbd_device *device;
4004         struct p_req_state *p = pi->data;
4005         union drbd_state mask, val;
4006         enum drbd_state_rv rv;
4007
4008         peer_device = conn_peer_device(connection, pi->vnr);
4009         if (!peer_device)
4010                 return -EIO;
4011         device = peer_device->device;
4012
4013         mask.i = be32_to_cpu(p->mask);
4014         val.i = be32_to_cpu(p->val);
4015
4016         if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags) &&
4017             mutex_is_locked(device->state_mutex)) {
4018                 drbd_send_sr_reply(peer_device, SS_CONCURRENT_ST_CHG);
4019                 return 0;
4020         }
4021
4022         mask = convert_state(mask);
4023         val = convert_state(val);
4024
4025         rv = drbd_change_state(device, CS_VERBOSE, mask, val);
4026         drbd_send_sr_reply(peer_device, rv);
4027
4028         drbd_md_sync(device);
4029
4030         return 0;
4031 }
4032
4033 static int receive_req_conn_state(struct drbd_connection *connection, struct packet_info *pi)
4034 {
4035         struct p_req_state *p = pi->data;
4036         union drbd_state mask, val;
4037         enum drbd_state_rv rv;
4038
4039         mask.i = be32_to_cpu(p->mask);
4040         val.i = be32_to_cpu(p->val);
4041
4042         if (test_bit(RESOLVE_CONFLICTS, &connection->flags) &&
4043             mutex_is_locked(&connection->cstate_mutex)) {
4044                 conn_send_sr_reply(connection, SS_CONCURRENT_ST_CHG);
4045                 return 0;
4046         }
4047
4048         mask = convert_state(mask);
4049         val = convert_state(val);
4050
4051         rv = conn_request_state(connection, mask, val, CS_VERBOSE | CS_LOCAL_ONLY | CS_IGN_OUTD_FAIL);
4052         conn_send_sr_reply(connection, rv);
4053
4054         return 0;
4055 }
4056
4057 static int receive_state(struct drbd_connection *connection, struct packet_info *pi)
4058 {
4059         struct drbd_peer_device *peer_device;
4060         struct drbd_device *device;
4061         struct p_state *p = pi->data;
4062         union drbd_state os, ns, peer_state;
4063         enum drbd_disk_state real_peer_disk;
4064         enum chg_state_flags cs_flags;
4065         int rv;
4066
4067         peer_device = conn_peer_device(connection, pi->vnr);
4068         if (!peer_device)
4069                 return config_unknown_volume(connection, pi);
4070         device = peer_device->device;
4071
4072         peer_state.i = be32_to_cpu(p->state);
4073
4074         real_peer_disk = peer_state.disk;
4075         if (peer_state.disk == D_NEGOTIATING) {
4076                 real_peer_disk = device->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT;
4077                 drbd_info(device, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk));
4078         }
4079
4080         spin_lock_irq(&device->resource->req_lock);
4081  retry:
4082         os = ns = drbd_read_state(device);
4083         spin_unlock_irq(&device->resource->req_lock);
4084
4085         /* If some other part of the code (ack_receiver thread, timeout)
4086          * already decided to close the connection again,
4087          * we must not "re-establish" it here. */
4088         if (os.conn <= C_TEAR_DOWN)
4089                 return -ECONNRESET;
4090
4091         /* If this is the "end of sync" confirmation, usually the peer disk
4092          * transitions from D_INCONSISTENT to D_UP_TO_DATE. For empty (0 bits
4093          * set) resync started in PausedSyncT, or if the timing of pause-/
4094          * unpause-sync events has been "just right", the peer disk may
4095          * transition from D_CONSISTENT to D_UP_TO_DATE as well.
4096          */
4097         if ((os.pdsk == D_INCONSISTENT || os.pdsk == D_CONSISTENT) &&
4098             real_peer_disk == D_UP_TO_DATE &&
4099             os.conn > C_CONNECTED && os.disk == D_UP_TO_DATE) {
4100                 /* If we are (becoming) SyncSource, but peer is still in sync
4101                  * preparation, ignore its uptodate-ness to avoid flapping, it
4102                  * will change to inconsistent once the peer reaches active
4103                  * syncing states.
4104                  * It may have changed syncer-paused flags, however, so we
4105                  * cannot ignore this completely. */
4106                 if (peer_state.conn > C_CONNECTED &&
4107                     peer_state.conn < C_SYNC_SOURCE)
4108                         real_peer_disk = D_INCONSISTENT;
4109
4110                 /* if peer_state changes to connected at the same time,
4111                  * it explicitly notifies us that it finished resync.
4112                  * Maybe we should finish it up, too? */
4113                 else if (os.conn >= C_SYNC_SOURCE &&
4114                          peer_state.conn == C_CONNECTED) {
4115                         if (drbd_bm_total_weight(device) <= device->rs_failed)
4116                                 drbd_resync_finished(device);
4117                         return 0;
4118                 }
4119         }
4120
4121         /* explicit verify finished notification, stop sector reached. */
4122         if (os.conn == C_VERIFY_T && os.disk == D_UP_TO_DATE &&
4123             peer_state.conn == C_CONNECTED && real_peer_disk == D_UP_TO_DATE) {
4124                 ov_out_of_sync_print(device);
4125                 drbd_resync_finished(device);
4126                 return 0;
4127         }
4128
4129         /* peer says his disk is inconsistent, while we think it is uptodate,
4130          * and this happens while the peer still thinks we have a sync going on,
4131          * but we think we are already done with the sync.
4132          * We ignore this to avoid flapping pdsk.
4133          * This should not happen, if the peer is a recent version of drbd. */
4134         if (os.pdsk == D_UP_TO_DATE && real_peer_disk == D_INCONSISTENT &&
4135             os.conn == C_CONNECTED && peer_state.conn > C_SYNC_SOURCE)
4136                 real_peer_disk = D_UP_TO_DATE;
4137
4138         if (ns.conn == C_WF_REPORT_PARAMS)
4139                 ns.conn = C_CONNECTED;
4140
4141         if (peer_state.conn == C_AHEAD)
4142                 ns.conn = C_BEHIND;
4143
4144         if (device->p_uuid && peer_state.disk >= D_NEGOTIATING &&
4145             get_ldev_if_state(device, D_NEGOTIATING)) {
4146                 int cr; /* consider resync */
4147
4148                 /* if we established a new connection */
4149                 cr  = (os.conn < C_CONNECTED);
4150                 /* if we had an established connection
4151                  * and one of the nodes newly attaches a disk */
4152                 cr |= (os.conn == C_CONNECTED &&
4153                        (peer_state.disk == D_NEGOTIATING ||
4154                         os.disk == D_NEGOTIATING));
4155                 /* if we have both been inconsistent, and the peer has been
4156                  * forced to be UpToDate with --overwrite-data */
4157                 cr |= test_bit(CONSIDER_RESYNC, &device->flags);
4158                 /* if we had been plain connected, and the admin requested to
4159                  * start a sync by "invalidate" or "invalidate-remote" */
4160                 cr |= (os.conn == C_CONNECTED &&
4161                                 (peer_state.conn >= C_STARTING_SYNC_S &&
4162                                  peer_state.conn <= C_WF_BITMAP_T));
4163
4164                 if (cr)
4165                         ns.conn = drbd_sync_handshake(peer_device, peer_state.role, real_peer_disk);
4166
4167                 put_ldev(device);
4168                 if (ns.conn == C_MASK) {
4169                         ns.conn = C_CONNECTED;
4170                         if (device->state.disk == D_NEGOTIATING) {
4171                                 drbd_force_state(device, NS(disk, D_FAILED));
4172                         } else if (peer_state.disk == D_NEGOTIATING) {
4173                                 drbd_err(device, "Disk attach process on the peer node was aborted.\n");
4174                                 peer_state.disk = D_DISKLESS;
4175                                 real_peer_disk = D_DISKLESS;
4176                         } else {
4177                                 if (test_and_clear_bit(CONN_DRY_RUN, &peer_device->connection->flags))
4178                                         return -EIO;
4179                                 D_ASSERT(device, os.conn == C_WF_REPORT_PARAMS);
4180                                 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4181                                 return -EIO;
4182                         }
4183                 }
4184         }
4185
4186         spin_lock_irq(&device->resource->req_lock);
4187         if (os.i != drbd_read_state(device).i)
4188                 goto retry;
4189         clear_bit(CONSIDER_RESYNC, &device->flags);
4190         ns.peer = peer_state.role;
4191         ns.pdsk = real_peer_disk;
4192         ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp);
4193         if ((ns.conn == C_CONNECTED || ns.conn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING)
4194                 ns.disk = device->new_state_tmp.disk;
4195         cs_flags = CS_VERBOSE + (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED ? 0 : CS_HARD);
4196         if (ns.pdsk == D_CONSISTENT && drbd_suspended(device) && ns.conn == C_CONNECTED && os.conn < C_CONNECTED &&
4197             test_bit(NEW_CUR_UUID, &device->flags)) {
4198                 /* Do not allow tl_restart(RESEND) for a rebooted peer. We can only allow this
4199                    for temporal network outages! */
4200                 spin_unlock_irq(&device->resource->req_lock);
4201                 drbd_err(device, "Aborting Connect, can not thaw IO with an only Consistent peer\n");
4202                 tl_clear(peer_device->connection);
4203                 drbd_uuid_new_current(device);
4204                 clear_bit(NEW_CUR_UUID, &device->flags);
4205                 conn_request_state(peer_device->connection, NS2(conn, C_PROTOCOL_ERROR, susp, 0), CS_HARD);
4206                 return -EIO;
4207         }
4208         rv = _drbd_set_state(device, ns, cs_flags, NULL);
4209         ns = drbd_read_state(device);
4210         spin_unlock_irq(&device->resource->req_lock);
4211
4212         if (rv < SS_SUCCESS) {
4213                 conn_request_state(peer_device->connection, NS(conn, C_DISCONNECTING), CS_HARD);
4214                 return -EIO;
4215         }
4216
4217         if (os.conn > C_WF_REPORT_PARAMS) {
4218                 if (ns.conn > C_CONNECTED && peer_state.conn <= C_CONNECTED &&
4219                     peer_state.disk != D_NEGOTIATING ) {
4220                         /* we want resync, peer has not yet decided to sync... */
4221                         /* Nowadays only used when forcing a node into primary role and
4222                            setting its disk to UpToDate with that */
4223                         drbd_send_uuids(peer_device);
4224                         drbd_send_current_state(peer_device);
4225                 }
4226         }
4227
4228         clear_bit(DISCARD_MY_DATA, &device->flags);
4229
4230         drbd_md_sync(device); /* update connected indicator, la_size_sect, ... */
4231
4232         return 0;
4233 }
4234
4235 static int receive_sync_uuid(struct drbd_connection *connection, struct packet_info *pi)
4236 {
4237         struct drbd_peer_device *peer_device;
4238         struct drbd_device *device;
4239         struct p_rs_uuid *p = pi->data;
4240
4241         peer_device = conn_peer_device(connection, pi->vnr);
4242         if (!peer_device)
4243                 return -EIO;
4244         device = peer_device->device;
4245
4246         wait_event(device->misc_wait,
4247                    device->state.conn == C_WF_SYNC_UUID ||
4248                    device->state.conn == C_BEHIND ||
4249                    device->state.conn < C_CONNECTED ||
4250                    device->state.disk < D_NEGOTIATING);
4251
4252         /* D_ASSERT(device,  device->state.conn == C_WF_SYNC_UUID ); */
4253
4254         /* Here the _drbd_uuid_ functions are right, current should
4255            _not_ be rotated into the history */
4256         if (get_ldev_if_state(device, D_NEGOTIATING)) {
4257                 _drbd_uuid_set(device, UI_CURRENT, be64_to_cpu(p->uuid));
4258                 _drbd_uuid_set(device, UI_BITMAP, 0UL);
4259
4260                 drbd_print_uuids(device, "updated sync uuid");
4261                 drbd_start_resync(device, C_SYNC_TARGET);
4262
4263                 put_ldev(device);
4264         } else
4265                 drbd_err(device, "Ignoring SyncUUID packet!\n");
4266
4267         return 0;
4268 }
4269
4270 /**
4271  * receive_bitmap_plain
4272  *
4273  * Return 0 when done, 1 when another iteration is needed, and a negative error
4274  * code upon failure.
4275  */
4276 static int
4277 receive_bitmap_plain(struct drbd_peer_device *peer_device, unsigned int size,
4278                      unsigned long *p, struct bm_xfer_ctx *c)
4279 {
4280         unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE -
4281                                  drbd_header_size(peer_device->connection);
4282         unsigned int num_words = min_t(size_t, data_size / sizeof(*p),
4283                                        c->bm_words - c->word_offset);
4284         unsigned int want = num_words * sizeof(*p);
4285         int err;
4286
4287         if (want != size) {
4288                 drbd_err(peer_device, "%s:want (%u) != size (%u)\n", __func__, want, size);
4289                 return -EIO;
4290         }
4291         if (want == 0)
4292                 return 0;
4293         err = drbd_recv_all(peer_device->connection, p, want);
4294         if (err)
4295                 return err;
4296
4297         drbd_bm_merge_lel(peer_device->device, c->word_offset, num_words, p);
4298
4299         c->word_offset += num_words;
4300         c->bit_offset = c->word_offset * BITS_PER_LONG;
4301         if (c->bit_offset > c->bm_bits)
4302                 c->bit_offset = c->bm_bits;
4303
4304         return 1;
4305 }
4306
4307 static enum drbd_bitmap_code dcbp_get_code(struct p_compressed_bm *p)
4308 {
4309         return (enum drbd_bitmap_code)(p->encoding & 0x0f);
4310 }
4311
4312 static int dcbp_get_start(struct p_compressed_bm *p)
4313 {
4314         return (p->encoding & 0x80) != 0;
4315 }
4316
4317 static int dcbp_get_pad_bits(struct p_compressed_bm *p)
4318 {
4319         return (p->encoding >> 4) & 0x7;
4320 }
4321
4322 /**
4323  * recv_bm_rle_bits
4324  *
4325  * Return 0 when done, 1 when another iteration is needed, and a negative error
4326  * code upon failure.
4327  */
4328 static int
4329 recv_bm_rle_bits(struct drbd_peer_device *peer_device,
4330                 struct p_compressed_bm *p,
4331                  struct bm_xfer_ctx *c,
4332                  unsigned int len)
4333 {
4334         struct bitstream bs;
4335         u64 look_ahead;
4336         u64 rl;
4337         u64 tmp;
4338         unsigned long s = c->bit_offset;
4339         unsigned long e;
4340         int toggle = dcbp_get_start(p);
4341         int have;
4342         int bits;
4343
4344         bitstream_init(&bs, p->code, len, dcbp_get_pad_bits(p));
4345
4346         bits = bitstream_get_bits(&bs, &look_ahead, 64);
4347         if (bits < 0)
4348                 return -EIO;
4349
4350         for (have = bits; have > 0; s += rl, toggle = !toggle) {
4351                 bits = vli_decode_bits(&rl, look_ahead);
4352                 if (bits <= 0)
4353                         return -EIO;
4354
4355                 if (toggle) {
4356                         e = s + rl -1;
4357                         if (e >= c->bm_bits) {
4358                                 drbd_err(peer_device, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e);
4359                                 return -EIO;
4360                         }
4361                         _drbd_bm_set_bits(peer_device->device, s, e);
4362                 }
4363
4364                 if (have < bits) {
4365                         drbd_err(peer_device, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n",
4366                                 have, bits, look_ahead,
4367                                 (unsigned int)(bs.cur.b - p->code),
4368                                 (unsigned int)bs.buf_len);
4369                         return -EIO;
4370                 }
4371                 /* if we consumed all 64 bits, assign 0; >> 64 is "undefined"; */
4372                 if (likely(bits < 64))
4373                         look_ahead >>= bits;
4374                 else
4375                         look_ahead = 0;
4376                 have -= bits;
4377
4378                 bits = bitstream_get_bits(&bs, &tmp, 64 - have);
4379                 if (bits < 0)
4380                         return -EIO;
4381                 look_ahead |= tmp << have;
4382                 have += bits;
4383         }
4384
4385         c->bit_offset = s;
4386         bm_xfer_ctx_bit_to_word_offset(c);
4387
4388         return (s != c->bm_bits);
4389 }
4390
4391 /**
4392  * decode_bitmap_c
4393  *
4394  * Return 0 when done, 1 when another iteration is needed, and a negative error
4395  * code upon failure.
4396  */
4397 static int
4398 decode_bitmap_c(struct drbd_peer_device *peer_device,
4399                 struct p_compressed_bm *p,
4400                 struct bm_xfer_ctx *c,
4401                 unsigned int len)
4402 {
4403         if (dcbp_get_code(p) == RLE_VLI_Bits)
4404                 return recv_bm_rle_bits(peer_device, p, c, len - sizeof(*p));
4405
4406         /* other variants had been implemented for evaluation,
4407          * but have been dropped as this one turned out to be "best"
4408          * during all our tests. */
4409
4410         drbd_err(peer_device, "receive_bitmap_c: unknown encoding %u\n", p->encoding);
4411         conn_request_state(peer_device->connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4412         return -EIO;
4413 }
4414
4415 void INFO_bm_xfer_stats(struct drbd_device *device,
4416                 const char *direction, struct bm_xfer_ctx *c)
4417 {
4418         /* what would it take to transfer it "plaintext" */
4419         unsigned int header_size = drbd_header_size(first_peer_device(device)->connection);
4420         unsigned int data_size = DRBD_SOCKET_BUFFER_SIZE - header_size;
4421         unsigned int plain =
4422                 header_size * (DIV_ROUND_UP(c->bm_words, data_size) + 1) +
4423                 c->bm_words * sizeof(unsigned long);
4424         unsigned int total = c->bytes[0] + c->bytes[1];
4425         unsigned int r;
4426
4427         /* total can not be zero. but just in case: */
4428         if (total == 0)
4429                 return;
4430
4431         /* don't report if not compressed */
4432         if (total >= plain)
4433                 return;
4434
4435         /* total < plain. check for overflow, still */
4436         r = (total > UINT_MAX/1000) ? (total / (plain/1000))
4437                                     : (1000 * total / plain);
4438
4439         if (r > 1000)
4440                 r = 1000;
4441
4442         r = 1000 - r;
4443         drbd_info(device, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), "
4444              "total %u; compression: %u.%u%%\n",
4445                         direction,
4446                         c->bytes[1], c->packets[1],
4447                         c->bytes[0], c->packets[0],
4448                         total, r/10, r % 10);
4449 }
4450
4451 /* Since we are processing the bitfield from lower addresses to higher,
4452    it does not matter if the process it in 32 bit chunks or 64 bit
4453    chunks as long as it is little endian. (Understand it as byte stream,
4454    beginning with the lowest byte...) If we would use big endian
4455    we would need to process it from the highest address to the lowest,
4456    in order to be agnostic to the 32 vs 64 bits issue.
4457
4458    returns 0 on failure, 1 if we successfully received it. */
4459 static int receive_bitmap(struct drbd_connection *connection, struct packet_info *pi)
4460 {
4461         struct drbd_peer_device *peer_device;
4462         struct drbd_device *device;
4463         struct bm_xfer_ctx c;
4464         int err;
4465
4466         peer_device = conn_peer_device(connection, pi->vnr);
4467         if (!peer_device)
4468                 return -EIO;
4469         device = peer_device->device;
4470
4471         drbd_bm_lock(device, "receive bitmap", BM_LOCKED_SET_ALLOWED);
4472         /* you are supposed to send additional out-of-sync information
4473          * if you actually set bits during this phase */
4474
4475         c = (struct bm_xfer_ctx) {
4476                 .bm_bits = drbd_bm_bits(device),
4477                 .bm_words = drbd_bm_words(device),
4478         };
4479
4480         for(;;) {
4481                 if (pi->cmd == P_BITMAP)
4482                         err = receive_bitmap_plain(peer_device, pi->size, pi->data, &c);
4483                 else if (pi->cmd == P_COMPRESSED_BITMAP) {
4484                         /* MAYBE: sanity check that we speak proto >= 90,
4485                          * and the feature is enabled! */
4486                         struct p_compressed_bm *p = pi->data;
4487
4488                         if (pi->size > DRBD_SOCKET_BUFFER_SIZE - drbd_header_size(connection)) {
4489                                 drbd_err(device, "ReportCBitmap packet too large\n");
4490                                 err = -EIO;
4491                                 goto out;
4492                         }
4493                         if (pi->size <= sizeof(*p)) {
4494                                 drbd_err(device, "ReportCBitmap packet too small (l:%u)\n", pi->size);
4495                                 err = -EIO;
4496                                 goto out;
4497                         }
4498                         err = drbd_recv_all(peer_device->connection, p, pi->size);
4499                         if (err)
4500                                goto out;
4501                         err = decode_bitmap_c(peer_device, p, &c, pi->size);
4502                 } else {
4503                         drbd_warn(device, "receive_bitmap: cmd neither ReportBitMap nor ReportCBitMap (is 0x%x)", pi->cmd);
4504                         err = -EIO;
4505                         goto out;
4506                 }
4507
4508                 c.packets[pi->cmd == P_BITMAP]++;
4509                 c.bytes[pi->cmd == P_BITMAP] += drbd_header_size(connection) + pi->size;
4510
4511                 if (err <= 0) {
4512                         if (err < 0)
4513                                 goto out;
4514                         break;
4515                 }
4516                 err = drbd_recv_header(peer_device->connection, pi);
4517                 if (err)
4518                         goto out;
4519         }
4520
4521         INFO_bm_xfer_stats(device, "receive", &c);
4522
4523         if (device->state.conn == C_WF_BITMAP_T) {
4524                 enum drbd_state_rv rv;
4525
4526                 err = drbd_send_bitmap(device);
4527                 if (err)
4528                         goto out;
4529                 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */
4530                 rv = _drbd_request_state(device, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
4531                 D_ASSERT(device, rv == SS_SUCCESS);
4532         } else if (device->state.conn != C_WF_BITMAP_S) {
4533                 /* admin may have requested C_DISCONNECTING,
4534                  * other threads may have noticed network errors */
4535                 drbd_info(device, "unexpected cstate (%s) in receive_bitmap\n",
4536                     drbd_conn_str(device->state.conn));
4537         }
4538         err = 0;
4539
4540  out:
4541         drbd_bm_unlock(device);
4542         if (!err && device->state.conn == C_WF_BITMAP_S)
4543                 drbd_start_resync(device, C_SYNC_SOURCE);
4544         return err;
4545 }
4546
4547 static int receive_skip(struct drbd_connection *connection, struct packet_info *pi)
4548 {
4549         drbd_warn(connection, "skipping unknown optional packet type %d, l: %d!\n",
4550                  pi->cmd, pi->size);
4551
4552         return ignore_remaining_packet(connection, pi);
4553 }
4554
4555 static int receive_UnplugRemote(struct drbd_connection *connection, struct packet_info *pi)
4556 {
4557         /* Make sure we've acked all the TCP data associated
4558          * with the data requests being unplugged */
4559         drbd_tcp_quickack(connection->data.socket);
4560
4561         return 0;
4562 }
4563
4564 static int receive_out_of_sync(struct drbd_connection *connection, struct packet_info *pi)
4565 {
4566         struct drbd_peer_device *peer_device;
4567         struct drbd_device *device;
4568         struct p_block_desc *p = pi->data;
4569
4570         peer_device = conn_peer_device(connection, pi->vnr);
4571         if (!peer_device)
4572                 return -EIO;
4573         device = peer_device->device;
4574
4575         switch (device->state.conn) {
4576         case C_WF_SYNC_UUID:
4577         case C_WF_BITMAP_T:
4578         case C_BEHIND:
4579                         break;
4580         default:
4581                 drbd_err(device, "ASSERT FAILED cstate = %s, expected: WFSyncUUID|WFBitMapT|Behind\n",
4582                                 drbd_conn_str(device->state.conn));
4583         }
4584
4585         drbd_set_out_of_sync(device, be64_to_cpu(p->sector), be32_to_cpu(p->blksize));
4586
4587         return 0;
4588 }
4589
4590 struct data_cmd {
4591         int expect_payload;
4592         size_t pkt_size;
4593         int (*fn)(struct drbd_connection *, struct packet_info *);
4594 };
4595
4596 static struct data_cmd drbd_cmd_handler[] = {
4597         [P_DATA]            = { 1, sizeof(struct p_data), receive_Data },
4598         [P_DATA_REPLY]      = { 1, sizeof(struct p_data), receive_DataReply },
4599         [P_RS_DATA_REPLY]   = { 1, sizeof(struct p_data), receive_RSDataReply } ,
4600         [P_BARRIER]         = { 0, sizeof(struct p_barrier), receive_Barrier } ,
4601         [P_BITMAP]          = { 1, 0, receive_bitmap } ,
4602         [P_COMPRESSED_BITMAP] = { 1, 0, receive_bitmap } ,
4603         [P_UNPLUG_REMOTE]   = { 0, 0, receive_UnplugRemote },
4604         [P_DATA_REQUEST]    = { 0, sizeof(struct p_block_req), receive_DataRequest },
4605         [P_RS_DATA_REQUEST] = { 0, sizeof(struct p_block_req), receive_DataRequest },
4606         [P_SYNC_PARAM]      = { 1, 0, receive_SyncParam },
4607         [P_SYNC_PARAM89]    = { 1, 0, receive_SyncParam },
4608         [P_PROTOCOL]        = { 1, sizeof(struct p_protocol), receive_protocol },
4609         [P_UUIDS]           = { 0, sizeof(struct p_uuids), receive_uuids },
4610         [P_SIZES]           = { 0, sizeof(struct p_sizes), receive_sizes },
4611         [P_STATE]           = { 0, sizeof(struct p_state), receive_state },
4612         [P_STATE_CHG_REQ]   = { 0, sizeof(struct p_req_state), receive_req_state },
4613         [P_SYNC_UUID]       = { 0, sizeof(struct p_rs_uuid), receive_sync_uuid },
4614         [P_OV_REQUEST]      = { 0, sizeof(struct p_block_req), receive_DataRequest },
4615         [P_OV_REPLY]        = { 1, sizeof(struct p_block_req), receive_DataRequest },
4616         [P_CSUM_RS_REQUEST] = { 1, sizeof(struct p_block_req), receive_DataRequest },
4617         [P_DELAY_PROBE]     = { 0, sizeof(struct p_delay_probe93), receive_skip },
4618         [P_OUT_OF_SYNC]     = { 0, sizeof(struct p_block_desc), receive_out_of_sync },
4619         [P_CONN_ST_CHG_REQ] = { 0, sizeof(struct p_req_state), receive_req_conn_state },
4620         [P_PROTOCOL_UPDATE] = { 1, sizeof(struct p_protocol), receive_protocol },
4621         [P_TRIM]            = { 0, sizeof(struct p_trim), receive_Data },
4622 };
4623
4624 static void drbdd(struct drbd_connection *connection)
4625 {
4626         struct packet_info pi;
4627         size_t shs; /* sub header size */
4628         int err;
4629
4630         while (get_t_state(&connection->receiver) == RUNNING) {
4631                 struct data_cmd *cmd;
4632
4633                 drbd_thread_current_set_cpu(&connection->receiver);
4634                 update_receiver_timing_details(connection, drbd_recv_header);
4635                 if (drbd_recv_header(connection, &pi))
4636                         goto err_out;
4637
4638                 cmd = &drbd_cmd_handler[pi.cmd];
4639                 if (unlikely(pi.cmd >= ARRAY_SIZE(drbd_cmd_handler) || !cmd->fn)) {
4640                         drbd_err(connection, "Unexpected data packet %s (0x%04x)",
4641                                  cmdname(pi.cmd), pi.cmd);
4642                         goto err_out;
4643                 }
4644
4645                 shs = cmd->pkt_size;
4646                 if (pi.size > shs && !cmd->expect_payload) {
4647                         drbd_err(connection, "No payload expected %s l:%d\n",
4648                                  cmdname(pi.cmd), pi.size);
4649                         goto err_out;
4650                 }
4651
4652                 if (shs) {
4653                         update_receiver_timing_details(connection, drbd_recv_all_warn);
4654                         err = drbd_recv_all_warn(connection, pi.data, shs);
4655                         if (err)
4656                                 goto err_out;
4657                         pi.size -= shs;
4658                 }
4659
4660                 update_receiver_timing_details(connection, cmd->fn);
4661                 err = cmd->fn(connection, &pi);
4662                 if (err) {
4663                         drbd_err(connection, "error receiving %s, e: %d l: %d!\n",
4664                                  cmdname(pi.cmd), err, pi.size);
4665                         goto err_out;
4666                 }
4667         }
4668         return;
4669
4670     err_out:
4671         conn_request_state(connection, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
4672 }
4673
4674 static void conn_disconnect(struct drbd_connection *connection)
4675 {
4676         struct drbd_peer_device *peer_device;
4677         enum drbd_conns oc;
4678         int vnr;
4679
4680         if (connection->cstate == C_STANDALONE)
4681                 return;
4682
4683         /* We are about to start the cleanup after connection loss.
4684          * Make sure drbd_make_request knows about that.
4685          * Usually we should be in some network failure state already,
4686          * but just in case we are not, we fix it up here.
4687          */
4688         conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
4689
4690         /* ack_receiver does not clean up anything. it must not interfere, either */
4691         drbd_thread_stop(&connection->ack_receiver);
4692         if (connection->ack_sender) {
4693                 destroy_workqueue(connection->ack_sender);
4694                 connection->ack_sender = NULL;
4695         }
4696         drbd_free_sock(connection);
4697
4698         rcu_read_lock();
4699         idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
4700                 struct drbd_device *device = peer_device->device;
4701                 kref_get(&device->kref);
4702                 rcu_read_unlock();
4703                 drbd_disconnected(peer_device);
4704                 kref_put(&device->kref, drbd_destroy_device);
4705                 rcu_read_lock();
4706         }
4707         rcu_read_unlock();
4708
4709         if (!list_empty(&connection->current_epoch->list))
4710                 drbd_err(connection, "ASSERTION FAILED: connection->current_epoch->list not empty\n");
4711         /* ok, no more ee's on the fly, it is safe to reset the epoch_size */
4712         atomic_set(&connection->current_epoch->epoch_size, 0);
4713         connection->send.seen_any_write_yet = false;
4714
4715         drbd_info(connection, "Connection closed\n");
4716
4717         if (conn_highest_role(connection) == R_PRIMARY && conn_highest_pdsk(connection) >= D_UNKNOWN)
4718                 conn_try_outdate_peer_async(connection);
4719
4720         spin_lock_irq(&connection->resource->req_lock);
4721         oc = connection->cstate;
4722         if (oc >= C_UNCONNECTED)
4723                 _conn_request_state(connection, NS(conn, C_UNCONNECTED), CS_VERBOSE);
4724
4725         spin_unlock_irq(&connection->resource->req_lock);
4726
4727         if (oc == C_DISCONNECTING)
4728                 conn_request_state(connection, NS(conn, C_STANDALONE), CS_VERBOSE | CS_HARD);
4729 }
4730
4731 static int drbd_disconnected(struct drbd_peer_device *peer_device)
4732 {
4733         struct drbd_device *device = peer_device->device;
4734         unsigned int i;
4735
4736         /* wait for current activity to cease. */
4737         spin_lock_irq(&device->resource->req_lock);
4738         _drbd_wait_ee_list_empty(device, &device->active_ee);
4739         _drbd_wait_ee_list_empty(device, &device->sync_ee);
4740         _drbd_wait_ee_list_empty(device, &device->read_ee);
4741         spin_unlock_irq(&device->resource->req_lock);
4742
4743         /* We do not have data structures that would allow us to
4744          * get the rs_pending_cnt down to 0 again.
4745          *  * On C_SYNC_TARGET we do not have any data structures describing
4746          *    the pending RSDataRequest's we have sent.
4747          *  * On C_SYNC_SOURCE there is no data structure that tracks
4748          *    the P_RS_DATA_REPLY blocks that we sent to the SyncTarget.
4749          *  And no, it is not the sum of the reference counts in the
4750          *  resync_LRU. The resync_LRU tracks the whole operation including
4751          *  the disk-IO, while the rs_pending_cnt only tracks the blocks
4752          *  on the fly. */
4753         drbd_rs_cancel_all(device);
4754         device->rs_total = 0;
4755         device->rs_failed = 0;
4756         atomic_set(&device->rs_pending_cnt, 0);
4757         wake_up(&device->misc_wait);
4758
4759         del_timer_sync(&device->resync_timer);
4760         resync_timer_fn((unsigned long)device);
4761
4762         /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier,
4763          * w_make_resync_request etc. which may still be on the worker queue
4764          * to be "canceled" */
4765         drbd_flush_workqueue(&peer_device->connection->sender_work);
4766
4767         drbd_finish_peer_reqs(device);
4768
4769         /* This second workqueue flush is necessary, since drbd_finish_peer_reqs()
4770            might have issued a work again. The one before drbd_finish_peer_reqs() is
4771            necessary to reclain net_ee in drbd_finish_peer_reqs(). */
4772         drbd_flush_workqueue(&peer_device->connection->sender_work);
4773
4774         /* need to do it again, drbd_finish_peer_reqs() may have populated it
4775          * again via drbd_try_clear_on_disk_bm(). */
4776         drbd_rs_cancel_all(device);
4777
4778         kfree(device->p_uuid);
4779         device->p_uuid = NULL;
4780
4781         if (!drbd_suspended(device))
4782                 tl_clear(peer_device->connection);
4783
4784         drbd_md_sync(device);
4785
4786         /* serialize with bitmap writeout triggered by the state change,
4787          * if any. */
4788         wait_event(device->misc_wait, !test_bit(BITMAP_IO, &device->flags));
4789
4790         /* tcp_close and release of sendpage pages can be deferred.  I don't
4791          * want to use SO_LINGER, because apparently it can be deferred for
4792          * more than 20 seconds (longest time I checked).
4793          *
4794          * Actually we don't care for exactly when the network stack does its
4795          * put_page(), but release our reference on these pages right here.
4796          */
4797         i = drbd_free_peer_reqs(device, &device->net_ee);
4798         if (i)
4799                 drbd_info(device, "net_ee not empty, killed %u entries\n", i);
4800         i = atomic_read(&device->pp_in_use_by_net);
4801         if (i)
4802                 drbd_info(device, "pp_in_use_by_net = %d, expected 0\n", i);
4803         i = atomic_read(&device->pp_in_use);
4804         if (i)
4805                 drbd_info(device, "pp_in_use = %d, expected 0\n", i);
4806
4807         D_ASSERT(device, list_empty(&device->read_ee));
4808         D_ASSERT(device, list_empty(&device->active_ee));
4809         D_ASSERT(device, list_empty(&device->sync_ee));
4810         D_ASSERT(device, list_empty(&device->done_ee));
4811
4812         return 0;
4813 }
4814
4815 /*
4816  * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version
4817  * we can agree on is stored in agreed_pro_version.
4818  *
4819  * feature flags and the reserved array should be enough room for future
4820  * enhancements of the handshake protocol, and possible plugins...
4821  *
4822  * for now, they are expected to be zero, but ignored.
4823  */
4824 static int drbd_send_features(struct drbd_connection *connection)
4825 {
4826         struct drbd_socket *sock;
4827         struct p_connection_features *p;
4828
4829         sock = &connection->data;
4830         p = conn_prepare_command(connection, sock);
4831         if (!p)
4832                 return -EIO;
4833         memset(p, 0, sizeof(*p));
4834         p->protocol_min = cpu_to_be32(PRO_VERSION_MIN);
4835         p->protocol_max = cpu_to_be32(PRO_VERSION_MAX);
4836         p->feature_flags = cpu_to_be32(PRO_FEATURES);
4837         return conn_send_command(connection, sock, P_CONNECTION_FEATURES, sizeof(*p), NULL, 0);
4838 }
4839
4840 /*
4841  * return values:
4842  *   1 yes, we have a valid connection
4843  *   0 oops, did not work out, please try again
4844  *  -1 peer talks different language,
4845  *     no point in trying again, please go standalone.
4846  */
4847 static int drbd_do_features(struct drbd_connection *connection)
4848 {
4849         /* ASSERT current == connection->receiver ... */
4850         struct p_connection_features *p;
4851         const int expect = sizeof(struct p_connection_features);
4852         struct packet_info pi;
4853         int err;
4854
4855         err = drbd_send_features(connection);
4856         if (err)
4857                 return 0;
4858
4859         err = drbd_recv_header(connection, &pi);
4860         if (err)
4861                 return 0;
4862
4863         if (pi.cmd != P_CONNECTION_FEATURES) {
4864                 drbd_err(connection, "expected ConnectionFeatures packet, received: %s (0x%04x)\n",
4865                          cmdname(pi.cmd), pi.cmd);
4866                 return -1;
4867         }
4868
4869         if (pi.size != expect) {
4870                 drbd_err(connection, "expected ConnectionFeatures length: %u, received: %u\n",
4871                      expect, pi.size);
4872                 return -1;
4873         }
4874
4875         p = pi.data;
4876         err = drbd_recv_all_warn(connection, p, expect);
4877         if (err)
4878                 return 0;
4879
4880         p->protocol_min = be32_to_cpu(p->protocol_min);
4881         p->protocol_max = be32_to_cpu(p->protocol_max);
4882         if (p->protocol_max == 0)
4883                 p->protocol_max = p->protocol_min;
4884
4885         if (PRO_VERSION_MAX < p->protocol_min ||
4886             PRO_VERSION_MIN > p->protocol_max)
4887                 goto incompat;
4888
4889         connection->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max);
4890         connection->agreed_features = PRO_FEATURES & be32_to_cpu(p->feature_flags);
4891
4892         drbd_info(connection, "Handshake successful: "
4893              "Agreed network protocol version %d\n", connection->agreed_pro_version);
4894
4895         drbd_info(connection, "Agreed to%ssupport TRIM on protocol level\n",
4896                   connection->agreed_features & FF_TRIM ? " " : " not ");
4897
4898         return 1;
4899
4900  incompat:
4901         drbd_err(connection, "incompatible DRBD dialects: "
4902             "I support %d-%d, peer supports %d-%d\n",
4903             PRO_VERSION_MIN, PRO_VERSION_MAX,
4904             p->protocol_min, p->protocol_max);
4905         return -1;
4906 }
4907
4908 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE)
4909 static int drbd_do_auth(struct drbd_connection *connection)
4910 {
4911         drbd_err(connection, "This kernel was build without CONFIG_CRYPTO_HMAC.\n");
4912         drbd_err(connection, "You need to disable 'cram-hmac-alg' in drbd.conf.\n");
4913         return -1;
4914 }
4915 #else
4916 #define CHALLENGE_LEN 64
4917
4918 /* Return value:
4919         1 - auth succeeded,
4920         0 - failed, try again (network error),
4921         -1 - auth failed, don't try again.
4922 */
4923
4924 static int drbd_do_auth(struct drbd_connection *connection)
4925 {
4926         struct drbd_socket *sock;
4927         char my_challenge[CHALLENGE_LEN];  /* 64 Bytes... */
4928         char *response = NULL;
4929         char *right_response = NULL;
4930         char *peers_ch = NULL;
4931         unsigned int key_len;
4932         char secret[SHARED_SECRET_MAX]; /* 64 byte */
4933         unsigned int resp_size;
4934         SHASH_DESC_ON_STACK(desc, connection->cram_hmac_tfm);
4935         struct packet_info pi;
4936         struct net_conf *nc;
4937         int err, rv;
4938
4939         /* FIXME: Put the challenge/response into the preallocated socket buffer.  */
4940
4941         rcu_read_lock();
4942         nc = rcu_dereference(connection->net_conf);
4943         key_len = strlen(nc->shared_secret);
4944         memcpy(secret, nc->shared_secret, key_len);
4945         rcu_read_unlock();
4946
4947         desc->tfm = connection->cram_hmac_tfm;
4948         desc->flags = 0;
4949
4950         rv = crypto_shash_setkey(connection->cram_hmac_tfm, (u8 *)secret, key_len);
4951         if (rv) {
4952                 drbd_err(connection, "crypto_shash_setkey() failed with %d\n", rv);
4953                 rv = -1;
4954                 goto fail;
4955         }
4956
4957         get_random_bytes(my_challenge, CHALLENGE_LEN);
4958
4959         sock = &connection->data;
4960         if (!conn_prepare_command(connection, sock)) {
4961                 rv = 0;
4962                 goto fail;
4963         }
4964         rv = !conn_send_command(connection, sock, P_AUTH_CHALLENGE, 0,
4965                                 my_challenge, CHALLENGE_LEN);
4966         if (!rv)
4967                 goto fail;
4968
4969         err = drbd_recv_header(connection, &pi);
4970         if (err) {
4971                 rv = 0;
4972                 goto fail;
4973         }
4974
4975         if (pi.cmd != P_AUTH_CHALLENGE) {
4976                 drbd_err(connection, "expected AuthChallenge packet, received: %s (0x%04x)\n",
4977                          cmdname(pi.cmd), pi.cmd);
4978                 rv = 0;
4979                 goto fail;
4980         }
4981
4982         if (pi.size > CHALLENGE_LEN * 2) {
4983                 drbd_err(connection, "expected AuthChallenge payload too big.\n");
4984                 rv = -1;
4985                 goto fail;
4986         }
4987
4988         if (pi.size < CHALLENGE_LEN) {
4989                 drbd_err(connection, "AuthChallenge payload too small.\n");
4990                 rv = -1;
4991                 goto fail;
4992         }
4993
4994         peers_ch = kmalloc(pi.size, GFP_NOIO);
4995         if (peers_ch == NULL) {
4996                 drbd_err(connection, "kmalloc of peers_ch failed\n");
4997                 rv = -1;
4998                 goto fail;
4999         }
5000
5001         err = drbd_recv_all_warn(connection, peers_ch, pi.size);
5002         if (err) {
5003                 rv = 0;
5004                 goto fail;
5005         }
5006
5007         if (!memcmp(my_challenge, peers_ch, CHALLENGE_LEN)) {
5008                 drbd_err(connection, "Peer presented the same challenge!\n");
5009                 rv = -1;
5010                 goto fail;
5011         }
5012
5013         resp_size = crypto_shash_digestsize(connection->cram_hmac_tfm);
5014         response = kmalloc(resp_size, GFP_NOIO);
5015         if (response == NULL) {
5016                 drbd_err(connection, "kmalloc of response failed\n");
5017                 rv = -1;
5018                 goto fail;
5019         }
5020
5021         rv = crypto_shash_digest(desc, peers_ch, pi.size, response);
5022         if (rv) {
5023                 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5024                 rv = -1;
5025                 goto fail;
5026         }
5027
5028         if (!conn_prepare_command(connection, sock)) {
5029                 rv = 0;
5030                 goto fail;
5031         }
5032         rv = !conn_send_command(connection, sock, P_AUTH_RESPONSE, 0,
5033                                 response, resp_size);
5034         if (!rv)
5035                 goto fail;
5036
5037         err = drbd_recv_header(connection, &pi);
5038         if (err) {
5039                 rv = 0;
5040                 goto fail;
5041         }
5042
5043         if (pi.cmd != P_AUTH_RESPONSE) {
5044                 drbd_err(connection, "expected AuthResponse packet, received: %s (0x%04x)\n",
5045                          cmdname(pi.cmd), pi.cmd);
5046                 rv = 0;
5047                 goto fail;
5048         }
5049
5050         if (pi.size != resp_size) {
5051                 drbd_err(connection, "expected AuthResponse payload of wrong size\n");
5052                 rv = 0;
5053                 goto fail;
5054         }
5055
5056         err = drbd_recv_all_warn(connection, response , resp_size);
5057         if (err) {
5058                 rv = 0;
5059                 goto fail;
5060         }
5061
5062         right_response = kmalloc(resp_size, GFP_NOIO);
5063         if (right_response == NULL) {
5064                 drbd_err(connection, "kmalloc of right_response failed\n");
5065                 rv = -1;
5066                 goto fail;
5067         }
5068
5069         rv = crypto_shash_digest(desc, my_challenge, CHALLENGE_LEN,
5070                                  right_response);
5071         if (rv) {
5072                 drbd_err(connection, "crypto_hash_digest() failed with %d\n", rv);
5073                 rv = -1;
5074                 goto fail;
5075         }
5076
5077         rv = !memcmp(response, right_response, resp_size);
5078
5079         if (rv)
5080                 drbd_info(connection, "Peer authenticated using %d bytes HMAC\n",
5081                      resp_size);
5082         else
5083                 rv = -1;
5084
5085  fail:
5086         kfree(peers_ch);
5087         kfree(response);
5088         kfree(right_response);
5089         shash_desc_zero(desc);
5090
5091         return rv;
5092 }
5093 #endif
5094
5095 int drbd_receiver(struct drbd_thread *thi)
5096 {
5097         struct drbd_connection *connection = thi->connection;
5098         int h;
5099
5100         drbd_info(connection, "receiver (re)started\n");
5101
5102         do {
5103                 h = conn_connect(connection);
5104                 if (h == 0) {
5105                         conn_disconnect(connection);
5106                         schedule_timeout_interruptible(HZ);
5107                 }
5108                 if (h == -1) {
5109                         drbd_warn(connection, "Discarding network configuration.\n");
5110                         conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5111                 }
5112         } while (h == 0);
5113
5114         if (h > 0)
5115                 drbdd(connection);
5116
5117         conn_disconnect(connection);
5118
5119         drbd_info(connection, "receiver terminated\n");
5120         return 0;
5121 }
5122
5123 /* ********* acknowledge sender ******** */
5124
5125 static int got_conn_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5126 {
5127         struct p_req_state_reply *p = pi->data;
5128         int retcode = be32_to_cpu(p->retcode);
5129
5130         if (retcode >= SS_SUCCESS) {
5131                 set_bit(CONN_WD_ST_CHG_OKAY, &connection->flags);
5132         } else {
5133                 set_bit(CONN_WD_ST_CHG_FAIL, &connection->flags);
5134                 drbd_err(connection, "Requested state change failed by peer: %s (%d)\n",
5135                          drbd_set_st_err_str(retcode), retcode);
5136         }
5137         wake_up(&connection->ping_wait);
5138
5139         return 0;
5140 }
5141
5142 static int got_RqSReply(struct drbd_connection *connection, struct packet_info *pi)
5143 {
5144         struct drbd_peer_device *peer_device;
5145         struct drbd_device *device;
5146         struct p_req_state_reply *p = pi->data;
5147         int retcode = be32_to_cpu(p->retcode);
5148
5149         peer_device = conn_peer_device(connection, pi->vnr);
5150         if (!peer_device)
5151                 return -EIO;
5152         device = peer_device->device;
5153
5154         if (test_bit(CONN_WD_ST_CHG_REQ, &connection->flags)) {
5155                 D_ASSERT(device, connection->agreed_pro_version < 100);
5156                 return got_conn_RqSReply(connection, pi);
5157         }
5158
5159         if (retcode >= SS_SUCCESS) {
5160                 set_bit(CL_ST_CHG_SUCCESS, &device->flags);
5161         } else {
5162                 set_bit(CL_ST_CHG_FAIL, &device->flags);
5163                 drbd_err(device, "Requested state change failed by peer: %s (%d)\n",
5164                         drbd_set_st_err_str(retcode), retcode);
5165         }
5166         wake_up(&device->state_wait);
5167
5168         return 0;
5169 }
5170
5171 static int got_Ping(struct drbd_connection *connection, struct packet_info *pi)
5172 {
5173         return drbd_send_ping_ack(connection);
5174
5175 }
5176
5177 static int got_PingAck(struct drbd_connection *connection, struct packet_info *pi)
5178 {
5179         /* restore idle timeout */
5180         connection->meta.socket->sk->sk_rcvtimeo = connection->net_conf->ping_int*HZ;
5181         if (!test_and_set_bit(GOT_PING_ACK, &connection->flags))
5182                 wake_up(&connection->ping_wait);
5183
5184         return 0;
5185 }
5186
5187 static int got_IsInSync(struct drbd_connection *connection, struct packet_info *pi)
5188 {
5189         struct drbd_peer_device *peer_device;
5190         struct drbd_device *device;
5191         struct p_block_ack *p = pi->data;
5192         sector_t sector = be64_to_cpu(p->sector);
5193         int blksize = be32_to_cpu(p->blksize);
5194
5195         peer_device = conn_peer_device(connection, pi->vnr);
5196         if (!peer_device)
5197                 return -EIO;
5198         device = peer_device->device;
5199
5200         D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
5201
5202         update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5203
5204         if (get_ldev(device)) {
5205                 drbd_rs_complete_io(device, sector);
5206                 drbd_set_in_sync(device, sector, blksize);
5207                 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */
5208                 device->rs_same_csum += (blksize >> BM_BLOCK_SHIFT);
5209                 put_ldev(device);
5210         }
5211         dec_rs_pending(device);
5212         atomic_add(blksize >> 9, &device->rs_sect_in);
5213
5214         return 0;
5215 }
5216
5217 static int
5218 validate_req_change_req_state(struct drbd_device *device, u64 id, sector_t sector,
5219                               struct rb_root *root, const char *func,
5220                               enum drbd_req_event what, bool missing_ok)
5221 {
5222         struct drbd_request *req;
5223         struct bio_and_error m;
5224
5225         spin_lock_irq(&device->resource->req_lock);
5226         req = find_request(device, root, id, sector, missing_ok, func);
5227         if (unlikely(!req)) {
5228                 spin_unlock_irq(&device->resource->req_lock);
5229                 return -EIO;
5230         }
5231         __req_mod(req, what, &m);
5232         spin_unlock_irq(&device->resource->req_lock);
5233
5234         if (m.bio)
5235                 complete_master_bio(device, &m);
5236         return 0;
5237 }
5238
5239 static int got_BlockAck(struct drbd_connection *connection, struct packet_info *pi)
5240 {
5241         struct drbd_peer_device *peer_device;
5242         struct drbd_device *device;
5243         struct p_block_ack *p = pi->data;
5244         sector_t sector = be64_to_cpu(p->sector);
5245         int blksize = be32_to_cpu(p->blksize);
5246         enum drbd_req_event what;
5247
5248         peer_device = conn_peer_device(connection, pi->vnr);
5249         if (!peer_device)
5250                 return -EIO;
5251         device = peer_device->device;
5252
5253         update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5254
5255         if (p->block_id == ID_SYNCER) {
5256                 drbd_set_in_sync(device, sector, blksize);
5257                 dec_rs_pending(device);
5258                 return 0;
5259         }
5260         switch (pi->cmd) {
5261         case P_RS_WRITE_ACK:
5262                 what = WRITE_ACKED_BY_PEER_AND_SIS;
5263                 break;
5264         case P_WRITE_ACK:
5265                 what = WRITE_ACKED_BY_PEER;
5266                 break;
5267         case P_RECV_ACK:
5268                 what = RECV_ACKED_BY_PEER;
5269                 break;
5270         case P_SUPERSEDED:
5271                 what = CONFLICT_RESOLVED;
5272                 break;
5273         case P_RETRY_WRITE:
5274                 what = POSTPONE_WRITE;
5275                 break;
5276         default:
5277                 BUG();
5278         }
5279
5280         return validate_req_change_req_state(device, p->block_id, sector,
5281                                              &device->write_requests, __func__,
5282                                              what, false);
5283 }
5284
5285 static int got_NegAck(struct drbd_connection *connection, struct packet_info *pi)
5286 {
5287         struct drbd_peer_device *peer_device;
5288         struct drbd_device *device;
5289         struct p_block_ack *p = pi->data;
5290         sector_t sector = be64_to_cpu(p->sector);
5291         int size = be32_to_cpu(p->blksize);
5292         int err;
5293
5294         peer_device = conn_peer_device(connection, pi->vnr);
5295         if (!peer_device)
5296                 return -EIO;
5297         device = peer_device->device;
5298
5299         update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5300
5301         if (p->block_id == ID_SYNCER) {
5302                 dec_rs_pending(device);
5303                 drbd_rs_failed_io(device, sector, size);
5304                 return 0;
5305         }
5306
5307         err = validate_req_change_req_state(device, p->block_id, sector,
5308                                             &device->write_requests, __func__,
5309                                             NEG_ACKED, true);
5310         if (err) {
5311                 /* Protocol A has no P_WRITE_ACKs, but has P_NEG_ACKs.
5312                    The master bio might already be completed, therefore the
5313                    request is no longer in the collision hash. */
5314                 /* In Protocol B we might already have got a P_RECV_ACK
5315                    but then get a P_NEG_ACK afterwards. */
5316                 drbd_set_out_of_sync(device, sector, size);
5317         }
5318         return 0;
5319 }
5320
5321 static int got_NegDReply(struct drbd_connection *connection, struct packet_info *pi)
5322 {
5323         struct drbd_peer_device *peer_device;
5324         struct drbd_device *device;
5325         struct p_block_ack *p = pi->data;
5326         sector_t sector = be64_to_cpu(p->sector);
5327
5328         peer_device = conn_peer_device(connection, pi->vnr);
5329         if (!peer_device)
5330                 return -EIO;
5331         device = peer_device->device;
5332
5333         update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5334
5335         drbd_err(device, "Got NegDReply; Sector %llus, len %u.\n",
5336             (unsigned long long)sector, be32_to_cpu(p->blksize));
5337
5338         return validate_req_change_req_state(device, p->block_id, sector,
5339                                              &device->read_requests, __func__,
5340                                              NEG_ACKED, false);
5341 }
5342
5343 static int got_NegRSDReply(struct drbd_connection *connection, struct packet_info *pi)
5344 {
5345         struct drbd_peer_device *peer_device;
5346         struct drbd_device *device;
5347         sector_t sector;
5348         int size;
5349         struct p_block_ack *p = pi->data;
5350
5351         peer_device = conn_peer_device(connection, pi->vnr);
5352         if (!peer_device)
5353                 return -EIO;
5354         device = peer_device->device;
5355
5356         sector = be64_to_cpu(p->sector);
5357         size = be32_to_cpu(p->blksize);
5358
5359         update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5360
5361         dec_rs_pending(device);
5362
5363         if (get_ldev_if_state(device, D_FAILED)) {
5364                 drbd_rs_complete_io(device, sector);
5365                 switch (pi->cmd) {
5366                 case P_NEG_RS_DREPLY:
5367                         drbd_rs_failed_io(device, sector, size);
5368                 case P_RS_CANCEL:
5369                         break;
5370                 default:
5371                         BUG();
5372                 }
5373                 put_ldev(device);
5374         }
5375
5376         return 0;
5377 }
5378
5379 static int got_BarrierAck(struct drbd_connection *connection, struct packet_info *pi)
5380 {
5381         struct p_barrier_ack *p = pi->data;
5382         struct drbd_peer_device *peer_device;
5383         int vnr;
5384
5385         tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
5386
5387         rcu_read_lock();
5388         idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
5389                 struct drbd_device *device = peer_device->device;
5390
5391                 if (device->state.conn == C_AHEAD &&
5392                     atomic_read(&device->ap_in_flight) == 0 &&
5393                     !test_and_set_bit(AHEAD_TO_SYNC_SOURCE, &device->flags)) {
5394                         device->start_resync_timer.expires = jiffies + HZ;
5395                         add_timer(&device->start_resync_timer);
5396                 }
5397         }
5398         rcu_read_unlock();
5399
5400         return 0;
5401 }
5402
5403 static int got_OVResult(struct drbd_connection *connection, struct packet_info *pi)
5404 {
5405         struct drbd_peer_device *peer_device;
5406         struct drbd_device *device;
5407         struct p_block_ack *p = pi->data;
5408         struct drbd_device_work *dw;
5409         sector_t sector;
5410         int size;
5411
5412         peer_device = conn_peer_device(connection, pi->vnr);
5413         if (!peer_device)
5414                 return -EIO;
5415         device = peer_device->device;
5416
5417         sector = be64_to_cpu(p->sector);
5418         size = be32_to_cpu(p->blksize);
5419
5420         update_peer_seq(peer_device, be32_to_cpu(p->seq_num));
5421
5422         if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC)
5423                 drbd_ov_out_of_sync_found(device, sector, size);
5424         else
5425                 ov_out_of_sync_print(device);
5426
5427         if (!get_ldev(device))
5428                 return 0;
5429
5430         drbd_rs_complete_io(device, sector);
5431         dec_rs_pending(device);
5432
5433         --device->ov_left;
5434
5435         /* let's advance progress step marks only for every other megabyte */
5436         if ((device->ov_left & 0x200) == 0x200)
5437                 drbd_advance_rs_marks(device, device->ov_left);
5438
5439         if (device->ov_left == 0) {
5440                 dw = kmalloc(sizeof(*dw), GFP_NOIO);
5441                 if (dw) {
5442                         dw->w.cb = w_ov_finished;
5443                         dw->device = device;
5444                         drbd_queue_work(&peer_device->connection->sender_work, &dw->w);
5445                 } else {
5446                         drbd_err(device, "kmalloc(dw) failed.");
5447                         ov_out_of_sync_print(device);
5448                         drbd_resync_finished(device);
5449                 }
5450         }
5451         put_ldev(device);
5452         return 0;
5453 }
5454
5455 static int got_skip(struct drbd_connection *connection, struct packet_info *pi)
5456 {
5457         return 0;
5458 }
5459
5460 struct meta_sock_cmd {
5461         size_t pkt_size;
5462         int (*fn)(struct drbd_connection *connection, struct packet_info *);
5463 };
5464
5465 static void set_rcvtimeo(struct drbd_connection *connection, bool ping_timeout)
5466 {
5467         long t;
5468         struct net_conf *nc;
5469
5470         rcu_read_lock();
5471         nc = rcu_dereference(connection->net_conf);
5472         t = ping_timeout ? nc->ping_timeo : nc->ping_int;
5473         rcu_read_unlock();
5474
5475         t *= HZ;
5476         if (ping_timeout)
5477                 t /= 10;
5478
5479         connection->meta.socket->sk->sk_rcvtimeo = t;
5480 }
5481
5482 static void set_ping_timeout(struct drbd_connection *connection)
5483 {
5484         set_rcvtimeo(connection, 1);
5485 }
5486
5487 static void set_idle_timeout(struct drbd_connection *connection)
5488 {
5489         set_rcvtimeo(connection, 0);
5490 }
5491
5492 static struct meta_sock_cmd ack_receiver_tbl[] = {
5493         [P_PING]            = { 0, got_Ping },
5494         [P_PING_ACK]        = { 0, got_PingAck },
5495         [P_RECV_ACK]        = { sizeof(struct p_block_ack), got_BlockAck },
5496         [P_WRITE_ACK]       = { sizeof(struct p_block_ack), got_BlockAck },
5497         [P_RS_WRITE_ACK]    = { sizeof(struct p_block_ack), got_BlockAck },
5498         [P_SUPERSEDED]   = { sizeof(struct p_block_ack), got_BlockAck },
5499         [P_NEG_ACK]         = { sizeof(struct p_block_ack), got_NegAck },
5500         [P_NEG_DREPLY]      = { sizeof(struct p_block_ack), got_NegDReply },
5501         [P_NEG_RS_DREPLY]   = { sizeof(struct p_block_ack), got_NegRSDReply },
5502         [P_OV_RESULT]       = { sizeof(struct p_block_ack), got_OVResult },
5503         [P_BARRIER_ACK]     = { sizeof(struct p_barrier_ack), got_BarrierAck },
5504         [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
5505         [P_RS_IS_IN_SYNC]   = { sizeof(struct p_block_ack), got_IsInSync },
5506         [P_DELAY_PROBE]     = { sizeof(struct p_delay_probe93), got_skip },
5507         [P_RS_CANCEL]       = { sizeof(struct p_block_ack), got_NegRSDReply },
5508         [P_CONN_ST_CHG_REPLY]={ sizeof(struct p_req_state_reply), got_conn_RqSReply },
5509         [P_RETRY_WRITE]     = { sizeof(struct p_block_ack), got_BlockAck },
5510 };
5511
5512 int drbd_ack_receiver(struct drbd_thread *thi)
5513 {
5514         struct drbd_connection *connection = thi->connection;
5515         struct meta_sock_cmd *cmd = NULL;
5516         struct packet_info pi;
5517         unsigned long pre_recv_jif;
5518         int rv;
5519         void *buf    = connection->meta.rbuf;
5520         int received = 0;
5521         unsigned int header_size = drbd_header_size(connection);
5522         int expect   = header_size;
5523         bool ping_timeout_active = false;
5524         struct sched_param param = { .sched_priority = 2 };
5525
5526         rv = sched_setscheduler(current, SCHED_RR, &param);
5527         if (rv < 0)
5528                 drbd_err(connection, "drbd_ack_receiver: ERROR set priority, ret=%d\n", rv);
5529
5530         while (get_t_state(thi) == RUNNING) {
5531                 drbd_thread_current_set_cpu(thi);
5532
5533                 conn_reclaim_net_peer_reqs(connection);
5534
5535                 if (test_and_clear_bit(SEND_PING, &connection->flags)) {
5536                         if (drbd_send_ping(connection)) {
5537                                 drbd_err(connection, "drbd_send_ping has failed\n");
5538                                 goto reconnect;
5539                         }
5540                         set_ping_timeout(connection);
5541                         ping_timeout_active = true;
5542                 }
5543
5544                 pre_recv_jif = jiffies;
5545                 rv = drbd_recv_short(connection->meta.socket, buf, expect-received, 0);
5546
5547                 /* Note:
5548                  * -EINTR        (on meta) we got a signal
5549                  * -EAGAIN       (on meta) rcvtimeo expired
5550                  * -ECONNRESET   other side closed the connection
5551                  * -ERESTARTSYS  (on data) we got a signal
5552                  * rv <  0       other than above: unexpected error!
5553                  * rv == expected: full header or command
5554                  * rv <  expected: "woken" by signal during receive
5555                  * rv == 0       : "connection shut down by peer"
5556                  */
5557                 if (likely(rv > 0)) {
5558                         received += rv;
5559                         buf      += rv;
5560                 } else if (rv == 0) {
5561                         if (test_bit(DISCONNECT_SENT, &connection->flags)) {
5562                                 long t;
5563                                 rcu_read_lock();
5564                                 t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
5565                                 rcu_read_unlock();
5566
5567                                 t = wait_event_timeout(connection->ping_wait,
5568                                                        connection->cstate < C_WF_REPORT_PARAMS,
5569                                                        t);
5570                                 if (t)
5571                                         break;
5572                         }
5573                         drbd_err(connection, "meta connection shut down by peer.\n");
5574                         goto reconnect;
5575                 } else if (rv == -EAGAIN) {
5576                         /* If the data socket received something meanwhile,
5577                          * that is good enough: peer is still alive. */
5578                         if (time_after(connection->last_received, pre_recv_jif))
5579                                 continue;
5580                         if (ping_timeout_active) {
5581                                 drbd_err(connection, "PingAck did not arrive in time.\n");
5582                                 goto reconnect;
5583                         }
5584                         set_bit(SEND_PING, &connection->flags);
5585                         continue;
5586                 } else if (rv == -EINTR) {
5587                         /* maybe drbd_thread_stop(): the while condition will notice.
5588                          * maybe woken for send_ping: we'll send a ping above,
5589                          * and change the rcvtimeo */
5590                         flush_signals(current);
5591                         continue;
5592                 } else {
5593                         drbd_err(connection, "sock_recvmsg returned %d\n", rv);
5594                         goto reconnect;
5595                 }
5596
5597                 if (received == expect && cmd == NULL) {
5598                         if (decode_header(connection, connection->meta.rbuf, &pi))
5599                                 goto reconnect;
5600                         cmd = &ack_receiver_tbl[pi.cmd];
5601                         if (pi.cmd >= ARRAY_SIZE(ack_receiver_tbl) || !cmd->fn) {
5602                                 drbd_err(connection, "Unexpected meta packet %s (0x%04x)\n",
5603                                          cmdname(pi.cmd), pi.cmd);
5604                                 goto disconnect;
5605                         }
5606                         expect = header_size + cmd->pkt_size;
5607                         if (pi.size != expect - header_size) {
5608                                 drbd_err(connection, "Wrong packet size on meta (c: %d, l: %d)\n",
5609                                         pi.cmd, pi.size);
5610                                 goto reconnect;
5611                         }
5612                 }
5613                 if (received == expect) {
5614                         bool err;
5615
5616                         err = cmd->fn(connection, &pi);
5617                         if (err) {
5618                                 drbd_err(connection, "%pf failed\n", cmd->fn);
5619                                 goto reconnect;
5620                         }
5621
5622                         connection->last_received = jiffies;
5623
5624                         if (cmd == &ack_receiver_tbl[P_PING_ACK]) {
5625                                 set_idle_timeout(connection);
5626                                 ping_timeout_active = false;
5627                         }
5628
5629                         buf      = connection->meta.rbuf;
5630                         received = 0;
5631                         expect   = header_size;
5632                         cmd      = NULL;
5633                 }
5634         }
5635
5636         if (0) {
5637 reconnect:
5638                 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5639                 conn_md_sync(connection);
5640         }
5641         if (0) {
5642 disconnect:
5643                 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
5644         }
5645
5646         drbd_info(connection, "ack_receiver terminated\n");
5647
5648         return 0;
5649 }
5650
5651 void drbd_send_acks_wf(struct work_struct *ws)
5652 {
5653         struct drbd_peer_device *peer_device =
5654                 container_of(ws, struct drbd_peer_device, send_acks_work);
5655         struct drbd_connection *connection = peer_device->connection;
5656         struct drbd_device *device = peer_device->device;
5657         struct net_conf *nc;
5658         int tcp_cork, err;
5659
5660         rcu_read_lock();
5661         nc = rcu_dereference(connection->net_conf);
5662         tcp_cork = nc->tcp_cork;
5663         rcu_read_unlock();
5664
5665         if (tcp_cork)
5666                 drbd_tcp_cork(connection->meta.socket);
5667
5668         err = drbd_finish_peer_reqs(device);
5669         kref_put(&device->kref, drbd_destroy_device);
5670         /* get is in drbd_endio_write_sec_final(). That is necessary to keep the
5671            struct work_struct send_acks_work alive, which is in the peer_device object */
5672
5673         if (err) {
5674                 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
5675                 return;
5676         }
5677
5678         if (tcp_cork)
5679                 drbd_tcp_uncork(connection->meta.socket);
5680
5681         return;
5682 }