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