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