drivers/block/drbd/drbd_receiver.c

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