drivers/block/drbd/drbd_receiver.c

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