linux/drivers/block/drbd/drbd_receiver.c
<<
>>
Prefs
   1// SPDX-License-Identifier: GPL-2.0-or-later
   2/*
   3   drbd_receiver.c
   4
   5   This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
   6
   7   Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
   8   Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
   9   Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
  10
  11 */
  12
  13
  14#include <linux/module.h>
  15
  16#include <linux/uaccess.h>
  17#include <net/sock.h>
  18
  19#include <linux/drbd.h>
  20#include <linux/fs.h>
  21#include <linux/file.h>
  22#include <linux/in.h>
  23#include <linux/mm.h>
  24#include <linux/memcontrol.h>
  25#include <linux/mm_inline.h>
  26#include <linux/slab.h>
  27#include <uapi/linux/sched/types.h>
  28#include <linux/sched/signal.h>
  29#include <linux/pkt_sched.h>
  30#define __KERNEL_SYSCALLS__
  31#include <linux/unistd.h>
  32#include <linux/vmalloc.h>
  33#include <linux/random.h>
  34#include <linux/string.h>
  35#include <linux/scatterlist.h>
  36#include <linux/part_stat.h>
  37#include "drbd_int.h"
  38#include "drbd_protocol.h"
  39#include "drbd_req.h"
  40#include "drbd_vli.h"
  41
  42#define PRO_FEATURES (DRBD_FF_TRIM|DRBD_FF_THIN_RESYNC|DRBD_FF_WSAME|DRBD_FF_WZEROES)
  43
  44struct packet_info {
  45        enum drbd_packet cmd;
  46        unsigned int size;
  47        unsigned int vnr;
  48        void *data;
  49};
  50
  51enum finish_epoch {
  52        FE_STILL_LIVE,
  53        FE_DESTROYED,
  54        FE_RECYCLED,
  55};
  56
  57static int drbd_do_features(struct drbd_connection *connection);
  58static int drbd_do_auth(struct drbd_connection *connection);
  59static int drbd_disconnected(struct drbd_peer_device *);
  60static void conn_wait_active_ee_empty(struct drbd_connection *connection);
  61static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
  62static 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 */
  76static 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. */
 110static 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
 123static 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
 134static 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
 148static 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
 194static 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
 211static 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
 223static 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 */
 263struct 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. */
 318static 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/*
 344You need to hold the req_lock:
 345 _drbd_wait_ee_list_empty()
 346
 347You 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 */
 360struct drbd_peer_request *
 361drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
 362                    unsigned int request_size, unsigned int payload_size, gfp_t gfp_mask) __must_hold(local)
 363{
 364        struct drbd_device *device = peer_device->device;
 365        struct drbd_peer_request *peer_req;
 366        struct page *page = NULL;
 367        unsigned nr_pages = (payload_size + PAGE_SIZE -1) >> PAGE_SHIFT;
 368
 369        if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
 370                return NULL;
 371
 372        peer_req = mempool_alloc(&drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
 373        if (!peer_req) {
 374                if (!(gfp_mask & __GFP_NOWARN))
 375                        drbd_err(device, "%s: allocation failed\n", __func__);
 376                return NULL;
 377        }
 378
 379        if (nr_pages) {
 380                page = drbd_alloc_pages(peer_device, nr_pages,
 381                                        gfpflags_allow_blocking(gfp_mask));
 382                if (!page)
 383                        goto fail;
 384        }
 385
 386        memset(peer_req, 0, sizeof(*peer_req));
 387        INIT_LIST_HEAD(&peer_req->w.list);
 388        drbd_clear_interval(&peer_req->i);
 389        peer_req->i.size = request_size;
 390        peer_req->i.sector = sector;
 391        peer_req->submit_jif = jiffies;
 392        peer_req->peer_device = peer_device;
 393        peer_req->pages = page;
 394        /*
 395         * The block_id is opaque to the receiver.  It is not endianness
 396         * converted, and sent back to the sender unchanged.
 397         */
 398        peer_req->block_id = id;
 399
 400        return peer_req;
 401
 402 fail:
 403        mempool_free(peer_req, &drbd_ee_mempool);
 404        return NULL;
 405}
 406
 407void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
 408                       int is_net)
 409{
 410        might_sleep();
 411        if (peer_req->flags & EE_HAS_DIGEST)
 412                kfree(peer_req->digest);
 413        drbd_free_pages(device, peer_req->pages, is_net);
 414        D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
 415        D_ASSERT(device, drbd_interval_empty(&peer_req->i));
 416        if (!expect(!(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
 417                peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
 418                drbd_al_complete_io(device, &peer_req->i);
 419        }
 420        mempool_free(peer_req, &drbd_ee_mempool);
 421}
 422
 423int 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 */
 444static 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
 477static 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
 493static 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
 501static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
 502{
 503        struct kvec iov = {
 504                .iov_base = buf,
 505                .iov_len = size,
 506        };
 507        struct msghdr msg = {
 508                .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
 509        };
 510        iov_iter_kvec(&msg.msg_iter, READ, &iov, 1, size);
 511        return sock_recvmsg(sock, &msg, msg.msg_flags);
 512}
 513
 514static 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
 543out:
 544        return rv;
 545}
 546
 547static 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
 560static 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 */
 575static 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
 589static 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
 652out:
 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
 677struct 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
 685static 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
 696static 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;
 746out:
 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
 759static 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
 767static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
 768{
 769        int timeo, connect_int, err = 0;
 770        struct socket *s_estab = NULL;
 771        struct net_conf *nc;
 772
 773        rcu_read_lock();
 774        nc = rcu_dereference(connection->net_conf);
 775        if (!nc) {
 776                rcu_read_unlock();
 777                return NULL;
 778        }
 779        connect_int = nc->connect_int;
 780        rcu_read_unlock();
 781
 782        timeo = connect_int * HZ;
 783        /* 28.5% random jitter */
 784        timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;
 785
 786        err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
 787        if (err <= 0)
 788                return NULL;
 789
 790        err = kernel_accept(ad->s_listen, &s_estab, 0);
 791        if (err < 0) {
 792                if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
 793                        drbd_err(connection, "accept failed, err = %d\n", err);
 794                        conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
 795                }
 796        }
 797
 798        if (s_estab)
 799                unregister_state_change(s_estab->sk, ad);
 800
 801        return s_estab;
 802}
 803
 804static int decode_header(struct drbd_connection *, void *, struct packet_info *);
 805
 806static 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
 814static 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 */
 846static 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
 865static 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 */
 890int 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 */
 924static 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
 977retry:
 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);
1006randomize:
1007                                if (prandom_u32() & 1)
1008                                        goto retry;
1009                        }
1010                }
1011
1012                if (connection->cstate <= C_DISCONNECTING)
1013                        goto out_release_sockets;
1014                if (signal_pending(current)) {
1015                        flush_signals(current);
1016                        smp_rmb();
1017                        if (get_t_state(&connection->receiver) == EXITING)
1018                                goto out_release_sockets;
1019                }
1020
1021                ok = connection_established(connection, &sock.socket, &msock.socket);
1022        } while (!ok);
1023
1024        if (ad.s_listen)
1025                sock_release(ad.s_listen);
1026
1027        sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1028        msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
1029
1030        sock.socket->sk->sk_allocation = GFP_NOIO;
1031        msock.socket->sk->sk_allocation = GFP_NOIO;
1032
1033        sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
1034        msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;
1035
1036        /* NOT YET ...
1037         * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
1038         * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1039         * first set it to the P_CONNECTION_FEATURES timeout,
1040         * which we set to 4x the configured ping_timeout. */
1041        rcu_read_lock();
1042        nc = rcu_dereference(connection->net_conf);
1043
1044        sock.socket->sk->sk_sndtimeo =
1045        sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;
1046
1047        msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
1048        timeout = nc->timeout * HZ / 10;
1049        discard_my_data = nc->discard_my_data;
1050        rcu_read_unlock();
1051
1052        msock.socket->sk->sk_sndtimeo = timeout;
1053
1054        /* we don't want delays.
1055         * we use TCP_CORK where appropriate, though */
1056        tcp_sock_set_nodelay(sock.socket->sk);
1057        tcp_sock_set_nodelay(msock.socket->sk);
1058
1059        connection->data.socket = sock.socket;
1060        connection->meta.socket = msock.socket;
1061        connection->last_received = jiffies;
1062
1063        h = drbd_do_features(connection);
1064        if (h <= 0)
1065                return h;
1066
1067        if (connection->cram_hmac_tfm) {
1068                /* drbd_request_state(device, NS(conn, WFAuth)); */
1069                switch (drbd_do_auth(connection)) {
1070                case -1:
1071                        drbd_err(connection, "Authentication of peer failed\n");
1072                        return -1;
1073                case 0:
1074                        drbd_err(connection, "Authentication of peer failed, trying again.\n");
1075                        return 0;
1076                }
1077        }
1078
1079        connection->data.socket->sk->sk_sndtimeo = timeout;
1080        connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
1081
1082        if (drbd_send_protocol(connection) == -EOPNOTSUPP)
1083                return -1;
1084
1085        /* Prevent a race between resync-handshake and
1086         * being promoted to Primary.
1087         *
1088         * Grab and release the state mutex, so we know that any current
1089         * drbd_set_role() is finished, and any incoming drbd_set_role
1090         * will see the STATE_SENT flag, and wait for it to be cleared.
1091         */
1092        idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1093                mutex_lock(peer_device->device->state_mutex);
1094
1095        /* avoid a race with conn_request_state( C_DISCONNECTING ) */
1096        spin_lock_irq(&connection->resource->req_lock);
1097        set_bit(STATE_SENT, &connection->flags);
1098        spin_unlock_irq(&connection->resource->req_lock);
1099
1100        idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
1101                mutex_unlock(peer_device->device->state_mutex);
1102
1103        rcu_read_lock();
1104        idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1105                struct drbd_device *device = peer_device->device;
1106                kref_get(&device->kref);
1107                rcu_read_unlock();
1108
1109                if (discard_my_data)
1110                        set_bit(DISCARD_MY_DATA, &device->flags);
1111                else
1112                        clear_bit(DISCARD_MY_DATA, &device->flags);
1113
1114                drbd_connected(peer_device);
1115                kref_put(&device->kref, drbd_destroy_device);
1116                rcu_read_lock();
1117        }
1118        rcu_read_unlock();
1119
1120        rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
1121        if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
1122                clear_bit(STATE_SENT, &connection->flags);
1123                return 0;
1124        }
1125
1126        drbd_thread_start(&connection->ack_receiver);
1127        /* opencoded create_singlethread_workqueue(),
1128         * to be able to use format string arguments */
1129        connection->ack_sender =
1130                alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
1131        if (!connection->ack_sender) {
1132                drbd_err(connection, "Failed to create workqueue ack_sender\n");
1133                return 0;
1134        }
1135
1136        mutex_lock(&connection->resource->conf_update);
1137        /* The discard_my_data flag is a single-shot modifier to the next
1138         * connection attempt, the handshake of which is now well underway.
1139         * No need for rcu style copying of the whole struct
1140         * just to clear a single value. */
1141        connection->net_conf->discard_my_data = 0;
1142        mutex_unlock(&connection->resource->conf_update);
1143
1144        return h;
1145
1146out_release_sockets:
1147        if (ad.s_listen)
1148                sock_release(ad.s_listen);
1149        if (sock.socket)
1150                sock_release(sock.socket);
1151        if (msock.socket)
1152                sock_release(msock.socket);
1153        return -1;
1154}
1155
1156static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
1157{
1158        unsigned int header_size = drbd_header_size(connection);
1159
1160        if (header_size == sizeof(struct p_header100) &&
1161            *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
1162                struct p_header100 *h = header;
1163                if (h->pad != 0) {
1164                        drbd_err(connection, "Header padding is not zero\n");
1165                        return -EINVAL;
1166                }
1167                pi->vnr = be16_to_cpu(h->volume);
1168                pi->cmd = be16_to_cpu(h->command);
1169                pi->size = be32_to_cpu(h->length);
1170        } else if (header_size == sizeof(struct p_header95) &&
1171                   *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
1172                struct p_header95 *h = header;
1173                pi->cmd = be16_to_cpu(h->command);
1174                pi->size = be32_to_cpu(h->length);
1175                pi->vnr = 0;
1176        } else if (header_size == sizeof(struct p_header80) &&
1177                   *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
1178                struct p_header80 *h = header;
1179                pi->cmd = be16_to_cpu(h->command);
1180                pi->size = be16_to_cpu(h->length);
1181                pi->vnr = 0;
1182        } else {
1183                drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
1184                         be32_to_cpu(*(__be32 *)header),
1185                         connection->agreed_pro_version);
1186                return -EINVAL;
1187        }
1188        pi->data = header + header_size;
1189        return 0;
1190}
1191
1192static void drbd_unplug_all_devices(struct drbd_connection *connection)
1193{
1194        if (current->plug == &connection->receiver_plug) {
1195                blk_finish_plug(&connection->receiver_plug);
1196                blk_start_plug(&connection->receiver_plug);
1197        } /* else: maybe just schedule() ?? */
1198}
1199
1200static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
1201{
1202        void *buffer = connection->data.rbuf;
1203        int err;
1204
1205        err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
1206        if (err)
1207                return err;
1208
1209        err = decode_header(connection, buffer, pi);
1210        connection->last_received = jiffies;
1211
1212        return err;
1213}
1214
1215static int drbd_recv_header_maybe_unplug(struct drbd_connection *connection, struct packet_info *pi)
1216{
1217        void *buffer = connection->data.rbuf;
1218        unsigned int size = drbd_header_size(connection);
1219        int err;
1220
1221        err = drbd_recv_short(connection->data.socket, buffer, size, MSG_NOSIGNAL|MSG_DONTWAIT);
1222        if (err != size) {
1223                /* If we have nothing in the receive buffer now, to reduce
1224                 * application latency, try to drain the backend queues as
1225                 * quickly as possible, and let remote TCP know what we have
1226                 * received so far. */
1227                if (err == -EAGAIN) {
1228                        tcp_sock_set_quickack(connection->data.socket->sk, 2);
1229                        drbd_unplug_all_devices(connection);
1230                }
1231                if (err > 0) {
1232                        buffer += err;
1233                        size -= err;
1234                }
1235                err = drbd_recv_all_warn(connection, buffer, size);
1236                if (err)
1237                        return err;
1238        }
1239
1240        err = decode_header(connection, connection->data.rbuf, pi);
1241        connection->last_received = jiffies;
1242
1243        return err;
1244}
1245/* This is blkdev_issue_flush, but asynchronous.
1246 * We want to submit to all component volumes in parallel,
1247 * then wait for all completions.
1248 */
1249struct issue_flush_context {
1250        atomic_t pending;
1251        int error;
1252        struct completion done;
1253};
1254struct one_flush_context {
1255        struct drbd_device *device;
1256        struct issue_flush_context *ctx;
1257};
1258
1259static void one_flush_endio(struct bio *bio)
1260{
1261        struct one_flush_context *octx = bio->bi_private;
1262        struct drbd_device *device = octx->device;
1263        struct issue_flush_context *ctx = octx->ctx;
1264
1265        if (bio->bi_status) {
1266                ctx->error = blk_status_to_errno(bio->bi_status);
1267                drbd_info(device, "local disk FLUSH FAILED with status %d\n", bio->bi_status);
1268        }
1269        kfree(octx);
1270        bio_put(bio);
1271
1272        clear_bit(FLUSH_PENDING, &device->flags);
1273        put_ldev(device);
1274        kref_put(&device->kref, drbd_destroy_device);
1275
1276        if (atomic_dec_and_test(&ctx->pending))
1277                complete(&ctx->done);
1278}
1279
1280static void submit_one_flush(struct drbd_device *device, struct issue_flush_context *ctx)
1281{
1282        struct bio *bio = bio_alloc(GFP_NOIO, 0);
1283        struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);
1284        if (!bio || !octx) {
1285                drbd_warn(device, "Could not allocate a bio, CANNOT ISSUE FLUSH\n");
1286                /* FIXME: what else can I do now?  disconnecting or detaching
1287                 * really does not help to improve the state of the world, either.
1288                 */
1289                kfree(octx);
1290                if (bio)
1291                        bio_put(bio);
1292
1293                ctx->error = -ENOMEM;
1294                put_ldev(device);
1295                kref_put(&device->kref, drbd_destroy_device);
1296                return;
1297        }
1298
1299        octx->device = device;
1300        octx->ctx = ctx;
1301        bio_set_dev(bio, device->ldev->backing_bdev);
1302        bio->bi_private = octx;
1303        bio->bi_end_io = one_flush_endio;
1304        bio->bi_opf = REQ_OP_FLUSH | REQ_PREFLUSH;
1305
1306        device->flush_jif = jiffies;
1307        set_bit(FLUSH_PENDING, &device->flags);
1308        atomic_inc(&ctx->pending);
1309        submit_bio(bio);
1310}
1311
1312static void drbd_flush(struct drbd_connection *connection)
1313{
1314        if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
1315                struct drbd_peer_device *peer_device;
1316                struct issue_flush_context ctx;
1317                int vnr;
1318
1319                atomic_set(&ctx.pending, 1);
1320                ctx.error = 0;
1321                init_completion(&ctx.done);
1322
1323                rcu_read_lock();
1324                idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1325                        struct drbd_device *device = peer_device->device;
1326
1327                        if (!get_ldev(device))
1328                                continue;
1329                        kref_get(&device->kref);
1330                        rcu_read_unlock();
1331
1332                        submit_one_flush(device, &ctx);
1333
1334                        rcu_read_lock();
1335                }
1336                rcu_read_unlock();
1337
1338                /* Do we want to add a timeout,
1339                 * if disk-timeout is set? */
1340                if (!atomic_dec_and_test(&ctx.pending))
1341                        wait_for_completion(&ctx.done);
1342
1343                if (ctx.error) {
1344                        /* would rather check on EOPNOTSUPP, but that is not reliable.
1345                         * don't try again for ANY return value != 0
1346                         * if (rv == -EOPNOTSUPP) */
1347                        /* Any error is already reported by bio_endio callback. */
1348                        drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
1349                }
1350        }
1351}
1352
1353/**
1354 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
1355 * @connection: DRBD connection.
1356 * @epoch:      Epoch object.
1357 * @ev:         Epoch event.
1358 */
1359static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
1360                                               struct drbd_epoch *epoch,
1361                                               enum epoch_event ev)
1362{
1363        int epoch_size;
1364        struct drbd_epoch *next_epoch;
1365        enum finish_epoch rv = FE_STILL_LIVE;
1366
1367        spin_lock(&connection->epoch_lock);
1368        do {
1369                next_epoch = NULL;
1370
1371                epoch_size = atomic_read(&epoch->epoch_size);
1372
1373                switch (ev & ~EV_CLEANUP) {
1374                case EV_PUT:
1375                        atomic_dec(&epoch->active);
1376                        break;
1377                case EV_GOT_BARRIER_NR:
1378                        set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
1379                        break;
1380                case EV_BECAME_LAST:
1381                        /* nothing to do*/
1382                        break;
1383                }
1384
1385                if (epoch_size != 0 &&
1386                    atomic_read(&epoch->active) == 0 &&
1387                    (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
1388                        if (!(ev & EV_CLEANUP)) {
1389                                spin_unlock(&connection->epoch_lock);
1390                                drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
1391                                spin_lock(&connection->epoch_lock);
1392                        }
1393#if 0
1394                        /* FIXME: dec unacked on connection, once we have
1395                         * something to count pending connection packets in. */
1396                        if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
1397                                dec_unacked(epoch->connection);
1398#endif
1399
1400                        if (connection->current_epoch != epoch) {
1401                                next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
1402                                list_del(&epoch->list);
1403                                ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
1404                                connection->epochs--;
1405                                kfree(epoch);
1406
1407                                if (rv == FE_STILL_LIVE)
1408                                        rv = FE_DESTROYED;
1409                        } else {
1410                                epoch->flags = 0;
1411                                atomic_set(&epoch->epoch_size, 0);
1412                                /* atomic_set(&epoch->active, 0); is already zero */
1413                                if (rv == FE_STILL_LIVE)
1414                                        rv = FE_RECYCLED;
1415                        }
1416                }
1417
1418                if (!next_epoch)
1419                        break;
1420
1421                epoch = next_epoch;
1422        } while (1);
1423
1424        spin_unlock(&connection->epoch_lock);
1425
1426        return rv;
1427}
1428
1429static enum write_ordering_e
1430max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
1431{
1432        struct disk_conf *dc;
1433
1434        dc = rcu_dereference(bdev->disk_conf);
1435
1436        if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
1437                wo = WO_DRAIN_IO;
1438        if (wo == WO_DRAIN_IO && !dc->disk_drain)
1439                wo = WO_NONE;
1440
1441        return wo;
1442}
1443
1444/*
1445 * drbd_bump_write_ordering() - Fall back to an other write ordering method
1446 * @wo:         Write ordering method to try.
1447 */
1448void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
1449                              enum write_ordering_e wo)
1450{
1451        struct drbd_device *device;
1452        enum write_ordering_e pwo;
1453        int vnr;
1454        static char *write_ordering_str[] = {
1455                [WO_NONE] = "none",
1456                [WO_DRAIN_IO] = "drain",
1457                [WO_BDEV_FLUSH] = "flush",
1458        };
1459
1460        pwo = resource->write_ordering;
1461        if (wo != WO_BDEV_FLUSH)
1462                wo = min(pwo, wo);
1463        rcu_read_lock();
1464        idr_for_each_entry(&resource->devices, device, vnr) {
1465                if (get_ldev(device)) {
1466                        wo = max_allowed_wo(device->ldev, wo);
1467                        if (device->ldev == bdev)
1468                                bdev = NULL;
1469                        put_ldev(device);
1470                }
1471        }
1472
1473        if (bdev)
1474                wo = max_allowed_wo(bdev, wo);
1475
1476        rcu_read_unlock();
1477
1478        resource->write_ordering = wo;
1479        if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
1480                drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
1481}
1482
1483/*
1484 * Mapping "discard" to ZEROOUT with UNMAP does not work for us:
1485 * Drivers have to "announce" q->limits.max_write_zeroes_sectors, or it
1486 * will directly go to fallback mode, submitting normal writes, and
1487 * never even try to UNMAP.
1488 *
1489 * And dm-thin does not do this (yet), mostly because in general it has
1490 * to assume that "skip_block_zeroing" is set.  See also:
1491 * https://www.mail-archive.com/dm-devel%40redhat.com/msg07965.html
1492 * https://www.redhat.com/archives/dm-devel/2018-January/msg00271.html
1493 *
1494 * We *may* ignore the discard-zeroes-data setting, if so configured.
1495 *
1496 * Assumption is that this "discard_zeroes_data=0" is only because the backend
1497 * may ignore partial unaligned discards.
1498 *
1499 * LVM/DM thin as of at least
1500 *   LVM version:     2.02.115(2)-RHEL7 (2015-01-28)
1501 *   Library version: 1.02.93-RHEL7 (2015-01-28)
1502 *   Driver version:  4.29.0
1503 * still behaves this way.
1504 *
1505 * For unaligned (wrt. alignment and granularity) or too small discards,
1506 * we zero-out the initial (and/or) trailing unaligned partial chunks,
1507 * but discard all the aligned full chunks.
1508 *
1509 * At least for LVM/DM thin, with skip_block_zeroing=false,
1510 * the result is effectively "discard_zeroes_data=1".
1511 */
1512/* flags: EE_TRIM|EE_ZEROOUT */
1513int drbd_issue_discard_or_zero_out(struct drbd_device *device, sector_t start, unsigned int nr_sectors, int flags)
1514{
1515        struct block_device *bdev = device->ldev->backing_bdev;
1516        struct request_queue *q = bdev_get_queue(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(q->limits.discard_granularity >> 9, 1U);
1527        alignment = (bdev_discard_alignment(bdev) >> 9) % granularity;
1528
1529        max_discard_sectors = min(q->limits.max_discard_sectors, (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, GFP_NOIO, 0);
1554                nr_sectors -= max_discard_sectors;
1555                start += max_discard_sectors;
1556        }
1557        if (nr_sectors) {
1558                /* max_discard_sectors is unsigned int (and a multiple of
1559                 * granularity, we made sure of that above already);
1560                 * nr is < max_discard_sectors;
1561                 * I don't need sector_div here, even though nr is sector_t */
1562                nr = nr_sectors;
1563                nr -= (unsigned int)nr % granularity;
1564                if (nr) {
1565                        err |= blkdev_issue_discard(bdev, start, nr, GFP_NOIO, 0);
1566                        nr_sectors -= nr;
1567                        start += nr;
1568                }
1569        }
1570 zero_out:
1571        if (nr_sectors) {
1572                err |= blkdev_issue_zeroout(bdev, start, nr_sectors, GFP_NOIO,
1573                                (flags & EE_TRIM) ? 0 : BLKDEV_ZERO_NOUNMAP);
1574        }
1575        return err != 0;
1576}
1577
1578static bool can_do_reliable_discards(struct drbd_device *device)
1579{
1580        struct request_queue *q = bdev_get_queue(device->ldev->backing_bdev);
1581        struct disk_conf *dc;
1582        bool can_do;
1583
1584        if (!blk_queue_discard(q))
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
1594static 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
1609static void drbd_issue_peer_wsame(struct drbd_device *device,
1610                                  struct drbd_peer_request *peer_req)
1611{
1612        struct block_device *bdev = device->ldev->backing_bdev;
1613        sector_t s = peer_req->i.sector;
1614        sector_t nr = peer_req->i.size >> 9;
1615        if (blkdev_issue_write_same(bdev, s, nr, GFP_NOIO, peer_req->pages))
1616                peer_req->flags |= EE_WAS_ERROR;
1617        drbd_endio_write_sec_final(peer_req);
1618}
1619
1620
1621/*
1622 * drbd_submit_peer_request()
1623 * @device:     DRBD device.
1624 * @peer_req:   peer request
1625 *
1626 * May spread the pages to multiple bios,
1627 * depending on bio_add_page restrictions.
1628 *
1629 * Returns 0 if all bios have been submitted,
1630 * -ENOMEM if we could not allocate enough bios,
1631 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
1632 *  single page to an empty bio (which should never happen and likely indicates
1633 *  that the lower level IO stack is in some way broken). This has been observed
1634 *  on certain Xen deployments.
1635 */
1636/* TODO allocate from our own bio_set. */
1637int drbd_submit_peer_request(struct drbd_device *device,
1638                             struct drbd_peer_request *peer_req,
1639                             const unsigned op, const unsigned op_flags,
1640                             const int fault_type)
1641{
1642        struct bio *bios = NULL;
1643        struct bio *bio;
1644        struct page *page = peer_req->pages;
1645        sector_t sector = peer_req->i.sector;
1646        unsigned data_size = peer_req->i.size;
1647        unsigned n_bios = 0;
1648        unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
1649        int err = -ENOMEM;
1650
1651        /* TRIM/DISCARD: for now, always use the helper function
1652         * blkdev_issue_zeroout(..., discard=true).
1653         * It's synchronous, but it does the right thing wrt. bio splitting.
1654         * Correctness first, performance later.  Next step is to code an
1655         * asynchronous variant of the same.
1656         */
1657        if (peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) {
1658                /* wait for all pending IO completions, before we start
1659                 * zeroing things out. */
1660                conn_wait_active_ee_empty(peer_req->peer_device->connection);
1661                /* add it to the active list now,
1662                 * so we can find it to present it in debugfs */
1663                peer_req->submit_jif = jiffies;
1664                peer_req->flags |= EE_SUBMITTED;
1665
1666                /* If this was a resync request from receive_rs_deallocated(),
1667                 * it is already on the sync_ee list */
1668                if (list_empty(&peer_req->w.list)) {
1669                        spin_lock_irq(&device->resource->req_lock);
1670                        list_add_tail(&peer_req->w.list, &device->active_ee);
1671                        spin_unlock_irq(&device->resource->req_lock);
1672                }
1673
1674                if (peer_req->flags & (EE_TRIM|EE_ZEROOUT))
1675                        drbd_issue_peer_discard_or_zero_out(device, peer_req);
1676                else /* EE_WRITE_SAME */
1677                        drbd_issue_peer_wsame(device, peer_req);
1678                return 0;
1679        }
1680
1681        /* In most cases, we will only need one bio.  But in case the lower
1682         * level restrictions happen to be different at this offset on this
1683         * side than those of the sending peer, we may need to submit the
1684         * request in more than one bio.
1685         *
1686         * Plain bio_alloc is good enough here, this is no DRBD internally
1687         * generated bio, but a bio allocated on behalf of the peer.
1688         */
1689next_bio:
1690        bio = bio_alloc(GFP_NOIO, nr_pages);
1691        if (!bio) {
1692                drbd_err(device, "submit_ee: Allocation of a bio failed (nr_pages=%u)\n", nr_pages);
1693                goto fail;
1694        }
1695        /* > peer_req->i.sector, unless this is the first bio */
1696        bio->bi_iter.bi_sector = sector;
1697        bio_set_dev(bio, device->ldev->backing_bdev);
1698        bio_set_op_attrs(bio, op, op_flags);
1699        bio->bi_private = peer_req;
1700        bio->bi_end_io = drbd_peer_request_endio;
1701
1702        bio->bi_next = bios;
1703        bios = bio;
1704        ++n_bios;
1705
1706        page_chain_for_each(page) {
1707                unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
1708                if (!bio_add_page(bio, page, len, 0))
1709                        goto next_bio;
1710                data_size -= len;
1711                sector += len >> 9;
1712                --nr_pages;
1713        }
1714        D_ASSERT(device, data_size == 0);
1715        D_ASSERT(device, page == NULL);
1716
1717        atomic_set(&peer_req->pending_bios, n_bios);
1718        /* for debugfs: update timestamp, mark as submitted */
1719        peer_req->submit_jif = jiffies;
1720        peer_req->flags |= EE_SUBMITTED;
1721        do {
1722                bio = bios;
1723                bios = bios->bi_next;
1724                bio->bi_next = NULL;
1725
1726                drbd_submit_bio_noacct(device, fault_type, bio);
1727        } while (bios);
1728        return 0;
1729
1730fail:
1731        while (bios) {
1732                bio = bios;
1733                bios = bios->bi_next;
1734                bio_put(bio);
1735        }
1736        return err;
1737}
1738
1739static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
1740                                             struct drbd_peer_request *peer_req)
1741{
1742        struct drbd_interval *i = &peer_req->i;
1743
1744        drbd_remove_interval(&device->write_requests, i);
1745        drbd_clear_interval(i);
1746
1747        /* Wake up any processes waiting for this peer request to complete.  */
1748        if (i->waiting)
1749                wake_up(&device->misc_wait);
1750}
1751
1752static void conn_wait_active_ee_empty(struct drbd_connection *connection)
1753{
1754        struct drbd_peer_device *peer_device;
1755        int vnr;
1756
1757        rcu_read_lock();
1758        idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1759                struct drbd_device *device = peer_device->device;
1760
1761                kref_get(&device->kref);
1762                rcu_read_unlock();
1763                drbd_wait_ee_list_empty(device, &device->active_ee);
1764                kref_put(&device->kref, drbd_destroy_device);
1765                rcu_read_lock();
1766        }
1767        rcu_read_unlock();
1768}
1769
1770static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
1771{
1772        int rv;
1773        struct p_barrier *p = pi->data;
1774        struct drbd_epoch *epoch;
1775
1776        /* FIXME these are unacked on connection,
1777         * not a specific (peer)device.
1778         */
1779        connection->current_epoch->barrier_nr = p->barrier;
1780        connection->current_epoch->connection = connection;
1781        rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);
1782
1783        /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
1784         * the activity log, which means it would not be resynced in case the
1785         * R_PRIMARY crashes now.
1786         * Therefore we must send the barrier_ack after the barrier request was
1787         * completed. */
1788        switch (connection->resource->write_ordering) {
1789        case WO_NONE:
1790                if (rv == FE_RECYCLED)
1791                        return 0;
1792
1793                /* receiver context, in the writeout path of the other node.
1794                 * avoid potential distributed deadlock */
1795                epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1796                if (epoch)
1797                        break;
1798                else
1799                        drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
1800                fallthrough;
1801
1802        case WO_BDEV_FLUSH:
1803        case WO_DRAIN_IO:
1804                conn_wait_active_ee_empty(connection);
1805                drbd_flush(connection);
1806
1807                if (atomic_read(&connection->current_epoch->epoch_size)) {
1808                        epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
1809                        if (epoch)
1810                                break;
1811                }
1812
1813                return 0;
1814        default:
1815                drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
1816                         connection->resource->write_ordering);
1817                return -EIO;
1818        }
1819
1820        epoch->flags = 0;
1821        atomic_set(&epoch->epoch_size, 0);
1822        atomic_set(&epoch->active, 0);
1823
1824        spin_lock(&connection->epoch_lock);
1825        if (atomic_read(&connection->current_epoch->epoch_size)) {
1826                list_add(&epoch->list, &connection->current_epoch->list);
1827                connection->current_epoch = epoch;
1828                connection->epochs++;
1829        } else {
1830                /* The current_epoch got recycled while we allocated this one... */
1831                kfree(epoch);
1832        }
1833        spin_unlock(&connection->epoch_lock);
1834
1835        return 0;
1836}
1837
1838/* quick wrapper in case payload size != request_size (write same) */
1839static void drbd_csum_ee_size(struct crypto_shash *h,
1840                              struct drbd_peer_request *r, void *d,
1841                              unsigned int payload_size)
1842{
1843        unsigned int tmp = r->i.size;
1844        r->i.size = payload_size;
1845        drbd_csum_ee(h, r, d);
1846        r->i.size = tmp;
1847}
1848
1849/* used from receive_RSDataReply (recv_resync_read)
1850 * and from receive_Data.
1851 * data_size: actual payload ("data in")
1852 *      for normal writes that is bi_size.
1853 *      for discards, that is zero.
1854 *      for write same, it is logical_block_size.
1855 * both trim and write same have the bi_size ("data len to be affected")
1856 * as extra argument in the packet header.
1857 */
1858static struct drbd_peer_request *
1859read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
1860              struct packet_info *pi) __must_hold(local)
1861{
1862        struct drbd_device *device = peer_device->device;
1863        const sector_t capacity = get_capacity(device->vdisk);
1864        struct drbd_peer_request *peer_req;
1865        struct page *page;
1866        int digest_size, err;
1867        unsigned int data_size = pi->size, ds;
1868        void *dig_in = peer_device->connection->int_dig_in;
1869        void *dig_vv = peer_device->connection->int_dig_vv;
1870        unsigned long *data;
1871        struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
1872        struct p_trim *zeroes = (pi->cmd == P_ZEROES) ? pi->data : NULL;
1873        struct p_trim *wsame = (pi->cmd == P_WSAME) ? pi->data : NULL;
1874
1875        digest_size = 0;
1876        if (!trim && peer_device->connection->peer_integrity_tfm) {
1877                digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
1878                /*
1879                 * FIXME: Receive the incoming digest into the receive buffer
1880                 *        here, together with its struct p_data?
1881                 */
1882                err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
1883                if (err)
1884                        return NULL;
1885                data_size -= digest_size;
1886        }
1887
1888        /* assume request_size == data_size, but special case trim and wsame. */
1889        ds = data_size;
1890        if (trim) {
1891                if (!expect(data_size == 0))
1892                        return NULL;
1893                ds = be32_to_cpu(trim->size);
1894        } else if (zeroes) {
1895                if (!expect(data_size == 0))
1896                        return NULL;
1897                ds = be32_to_cpu(zeroes->size);
1898        } else if (wsame) {
1899                if (data_size != queue_logical_block_size(device->rq_queue)) {
1900                        drbd_err(peer_device, "data size (%u) != drbd logical block size (%u)\n",
1901                                data_size, queue_logical_block_size(device->rq_queue));
1902                        return NULL;
1903                }
1904                if (data_size != bdev_logical_block_size(device->ldev->backing_bdev)) {
1905                        drbd_err(peer_device, "data size (%u) != backend logical block size (%u)\n",
1906                                data_size, bdev_logical_block_size(device->ldev->backing_bdev));
1907                        return NULL;
1908                }
1909                ds = be32_to_cpu(wsame->size);
1910        }
1911
1912        if (!expect(IS_ALIGNED(ds, 512)))
1913                return NULL;
1914        if (trim || wsame || zeroes) {
1915                if (!expect(ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
1916                        return NULL;
1917        } else if (!expect(ds <= DRBD_MAX_BIO_SIZE))
1918                return NULL;
1919
1920        /* even though we trust out peer,
1921         * we sometimes have to double check. */
1922        if (sector + (ds>>9) > capacity) {
1923                drbd_err(device, "request from peer beyond end of local disk: "
1924                        "capacity: %llus < sector: %llus + size: %u\n",
1925                        (unsigned long long)capacity,
1926                        (unsigned long long)sector, ds);
1927                return NULL;
1928        }
1929
1930        /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
1931         * "criss-cross" setup, that might cause write-out on some other DRBD,
1932         * which in turn might block on the other node at this very place.  */
1933        peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
1934        if (!peer_req)
1935                return NULL;
1936
1937        peer_req->flags |= EE_WRITE;
1938        if (trim) {
1939                peer_req->flags |= EE_TRIM;
1940                return peer_req;
1941        }
1942        if (zeroes) {
1943                peer_req->flags |= EE_ZEROOUT;
1944                return peer_req;
1945        }
1946        if (wsame)
1947                peer_req->flags |= EE_WRITE_SAME;
1948
1949        /* receive payload size bytes into page chain */
1950        ds = data_size;
1951        page = peer_req->pages;
1952        page_chain_for_each(page) {
1953                unsigned len = min_t(int, ds, PAGE_SIZE);
1954                data = kmap(page);
1955                err = drbd_recv_all_warn(peer_device->connection, data, len);
1956                if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
1957                        drbd_err(device, "Fault injection: Corrupting data on receive\n");
1958                        data[0] = data[0] ^ (unsigned long)-1;
1959                }
1960                kunmap(page);
1961                if (err) {
1962                        drbd_free_peer_req(device, peer_req);
1963                        return NULL;
1964                }
1965                ds -= len;
1966        }
1967
1968        if (digest_size) {
1969                drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
1970                if (memcmp(dig_in, dig_vv, digest_size)) {
1971                        drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
1972                                (unsigned long long)sector, data_size);
1973                        drbd_free_peer_req(device, peer_req);
1974                        return NULL;
1975                }
1976        }
1977        device->recv_cnt += data_size >> 9;
1978        return peer_req;
1979}
1980
1981/* drbd_drain_block() just takes a data block
1982 * out of the socket input buffer, and discards it.
1983 */
1984static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
1985{
1986        struct page *page;
1987        int err = 0;
1988        void *data;
1989
1990        if (!data_size)
1991                return 0;
1992
1993        page = drbd_alloc_pages(peer_device, 1, 1);
1994
1995        data = kmap(page);
1996        while (data_size) {
1997                unsigned int len = min_t(int, data_size, PAGE_SIZE);
1998
1999                err = drbd_recv_all_warn(peer_device->connection, data, len);
2000                if (err)
2001                        break;
2002                data_size -= len;
2003        }
2004        kunmap(page);
2005        drbd_free_pages(peer_device->device, page, 0);
2006        return err;
2007}
2008
2009static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
2010                           sector_t sector, int data_size)
2011{
2012        struct bio_vec bvec;
2013        struct bvec_iter iter;
2014        struct bio *bio;
2015        int digest_size, err, expect;
2016        void *dig_in = peer_device->connection->int_dig_in;
2017        void *dig_vv = peer_device->connection->int_dig_vv;
2018
2019        digest_size = 0;
2020        if (peer_device->connection->peer_integrity_tfm) {
2021                digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
2022                err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
2023                if (err)
2024                        return err;
2025                data_size -= digest_size;
2026        }
2027
2028        /* optimistically update recv_cnt.  if receiving fails below,
2029         * we disconnect anyways, and counters will be reset. */
2030        peer_device->device->recv_cnt += data_size>>9;
2031
2032        bio = req->master_bio;
2033        D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);
2034
2035        bio_for_each_segment(bvec, bio, iter) {
2036                void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
2037                expect = min_t(int, data_size, bvec.bv_len);
2038                err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
2039                kunmap(bvec.bv_page);
2040                if (err)
2041                        return err;
2042                data_size -= expect;
2043        }
2044
2045        if (digest_size) {
2046                drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
2047                if (memcmp(dig_in, dig_vv, digest_size)) {
2048                        drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
2049                        return -EINVAL;
2050                }
2051        }
2052
2053        D_ASSERT(peer_device->device, data_size == 0);
2054        return 0;
2055}
2056
2057/*
2058 * e_end_resync_block() is called in ack_sender context via
2059 * drbd_finish_peer_reqs().
2060 */
2061static int e_end_resync_block(struct drbd_work *w, int unused)
2062{
2063        struct drbd_peer_request *peer_req =
2064                container_of(w, struct drbd_peer_request, w);
2065        struct drbd_peer_device *peer_device = peer_req->peer_device;
2066        struct drbd_device *device = peer_device->device;
2067        sector_t sector = peer_req->i.sector;
2068        int err;
2069
2070        D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2071
2072        if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2073                drbd_set_in_sync(device, sector, peer_req->i.size);
2074                err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
2075        } else {
2076                /* Record failure to sync */
2077                drbd_rs_failed_io(device, sector, peer_req->i.size);
2078
2079                err  = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2080        }
2081        dec_unacked(device);
2082
2083        return err;
2084}
2085
2086static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
2087                            struct packet_info *pi) __releases(local)
2088{
2089        struct drbd_device *device = peer_device->device;
2090        struct drbd_peer_request *peer_req;
2091
2092        peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
2093        if (!peer_req)
2094                goto fail;
2095
2096        dec_rs_pending(device);
2097
2098        inc_unacked(device);
2099        /* corresponding dec_unacked() in e_end_resync_block()
2100         * respective _drbd_clear_done_ee */
2101
2102        peer_req->w.cb = e_end_resync_block;
2103        peer_req->submit_jif = jiffies;
2104
2105        spin_lock_irq(&device->resource->req_lock);
2106        list_add_tail(&peer_req->w.list, &device->sync_ee);
2107        spin_unlock_irq(&device->resource->req_lock);
2108
2109        atomic_add(pi->size >> 9, &device->rs_sect_ev);
2110        if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE, 0,
2111                                     DRBD_FAULT_RS_WR) == 0)
2112                return 0;
2113
2114        /* don't care for the reason here */
2115        drbd_err(device, "submit failed, triggering re-connect\n");
2116        spin_lock_irq(&device->resource->req_lock);
2117        list_del(&peer_req->w.list);
2118        spin_unlock_irq(&device->resource->req_lock);
2119
2120        drbd_free_peer_req(device, peer_req);
2121fail:
2122        put_ldev(device);
2123        return -EIO;
2124}
2125
2126static struct drbd_request *
2127find_request(struct drbd_device *device, struct rb_root *root, u64 id,
2128             sector_t sector, bool missing_ok, const char *func)
2129{
2130        struct drbd_request *req;
2131
2132        /* Request object according to our peer */
2133        req = (struct drbd_request *)(unsigned long)id;
2134        if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
2135                return req;
2136        if (!missing_ok) {
2137                drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
2138                        (unsigned long)id, (unsigned long long)sector);
2139        }
2140        return NULL;
2141}
2142
2143static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
2144{
2145        struct drbd_peer_device *peer_device;
2146        struct drbd_device *device;
2147        struct drbd_request *req;
2148        sector_t sector;
2149        int err;
2150        struct p_data *p = pi->data;
2151
2152        peer_device = conn_peer_device(connection, pi->vnr);
2153        if (!peer_device)
2154                return -EIO;
2155        device = peer_device->device;
2156
2157        sector = be64_to_cpu(p->sector);
2158
2159        spin_lock_irq(&device->resource->req_lock);
2160        req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
2161        spin_unlock_irq(&device->resource->req_lock);
2162        if (unlikely(!req))
2163                return -EIO;
2164
2165        /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
2166         * special casing it there for the various failure cases.
2167         * still no race with drbd_fail_pending_reads */
2168        err = recv_dless_read(peer_device, req, sector, pi->size);
2169        if (!err)
2170                req_mod(req, DATA_RECEIVED);
2171        /* else: nothing. handled from drbd_disconnect...
2172         * I don't think we may complete this just yet
2173         * in case we are "on-disconnect: freeze" */
2174
2175        return err;
2176}
2177
2178static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
2179{
2180        struct drbd_peer_device *peer_device;
2181        struct drbd_device *device;
2182        sector_t sector;
2183        int err;
2184        struct p_data *p = pi->data;
2185
2186        peer_device = conn_peer_device(connection, pi->vnr);
2187        if (!peer_device)
2188                return -EIO;
2189        device = peer_device->device;
2190
2191        sector = be64_to_cpu(p->sector);
2192        D_ASSERT(device, p->block_id == ID_SYNCER);
2193
2194        if (get_ldev(device)) {
2195                /* data is submitted to disk within recv_resync_read.
2196                 * corresponding put_ldev done below on error,
2197                 * or in drbd_peer_request_endio. */
2198                err = recv_resync_read(peer_device, sector, pi);
2199        } else {
2200                if (__ratelimit(&drbd_ratelimit_state))
2201                        drbd_err(device, "Can not write resync data to local disk.\n");
2202
2203                err = drbd_drain_block(peer_device, pi->size);
2204
2205                drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2206        }
2207
2208        atomic_add(pi->size >> 9, &device->rs_sect_in);
2209
2210        return err;
2211}
2212
2213static void restart_conflicting_writes(struct drbd_device *device,
2214                                       sector_t sector, int size)
2215{
2216        struct drbd_interval *i;
2217        struct drbd_request *req;
2218
2219        drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2220                if (!i->local)
2221                        continue;
2222                req = container_of(i, struct drbd_request, i);
2223                if (req->rq_state & RQ_LOCAL_PENDING ||
2224                    !(req->rq_state & RQ_POSTPONED))
2225                        continue;
2226                /* as it is RQ_POSTPONED, this will cause it to
2227                 * be queued on the retry workqueue. */
2228                __req_mod(req, CONFLICT_RESOLVED, NULL);
2229        }
2230}
2231
2232/*
2233 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
2234 */
2235static int e_end_block(struct drbd_work *w, int cancel)
2236{
2237        struct drbd_peer_request *peer_req =
2238                container_of(w, struct drbd_peer_request, w);
2239        struct drbd_peer_device *peer_device = peer_req->peer_device;
2240        struct drbd_device *device = peer_device->device;
2241        sector_t sector = peer_req->i.sector;
2242        int err = 0, pcmd;
2243
2244        if (peer_req->flags & EE_SEND_WRITE_ACK) {
2245                if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
2246                        pcmd = (device->state.conn >= C_SYNC_SOURCE &&
2247                                device->state.conn <= C_PAUSED_SYNC_T &&
2248                                peer_req->flags & EE_MAY_SET_IN_SYNC) ?
2249                                P_RS_WRITE_ACK : P_WRITE_ACK;
2250                        err = drbd_send_ack(peer_device, pcmd, peer_req);
2251                        if (pcmd == P_RS_WRITE_ACK)
2252                                drbd_set_in_sync(device, sector, peer_req->i.size);
2253                } else {
2254                        err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
2255                        /* we expect it to be marked out of sync anyways...
2256                         * maybe assert this?  */
2257                }
2258                dec_unacked(device);
2259        }
2260
2261        /* we delete from the conflict detection hash _after_ we sent out the
2262         * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right.  */
2263        if (peer_req->flags & EE_IN_INTERVAL_TREE) {
2264                spin_lock_irq(&device->resource->req_lock);
2265                D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
2266                drbd_remove_epoch_entry_interval(device, peer_req);
2267                if (peer_req->flags & EE_RESTART_REQUESTS)
2268                        restart_conflicting_writes(device, sector, peer_req->i.size);
2269                spin_unlock_irq(&device->resource->req_lock);
2270        } else
2271                D_ASSERT(device, drbd_interval_empty(&peer_req->i));
2272
2273        drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));
2274
2275        return err;
2276}
2277
2278static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
2279{
2280        struct drbd_peer_request *peer_req =
2281                container_of(w, struct drbd_peer_request, w);
2282        struct drbd_peer_device *peer_device = peer_req->peer_device;
2283        int err;
2284
2285        err = drbd_send_ack(peer_device, ack, peer_req);
2286        dec_unacked(peer_device->device);
2287
2288        return err;
2289}
2290
2291static int e_send_superseded(struct drbd_work *w, int unused)
2292{
2293        return e_send_ack(w, P_SUPERSEDED);
2294}
2295
2296static int e_send_retry_write(struct drbd_work *w, int unused)
2297{
2298        struct drbd_peer_request *peer_req =
2299                container_of(w, struct drbd_peer_request, w);
2300        struct drbd_connection *connection = peer_req->peer_device->connection;
2301
2302        return e_send_ack(w, connection->agreed_pro_version >= 100 ?
2303                             P_RETRY_WRITE : P_SUPERSEDED);
2304}
2305
2306static bool seq_greater(u32 a, u32 b)
2307{
2308        /*
2309         * We assume 32-bit wrap-around here.
2310         * For 24-bit wrap-around, we would have to shift:
2311         *  a <<= 8; b <<= 8;
2312         */
2313        return (s32)a - (s32)b > 0;
2314}
2315
2316static u32 seq_max(u32 a, u32 b)
2317{
2318        return seq_greater(a, b) ? a : b;
2319}
2320
2321static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
2322{
2323        struct drbd_device *device = peer_device->device;
2324        unsigned int newest_peer_seq;
2325
2326        if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
2327                spin_lock(&device->peer_seq_lock);
2328                newest_peer_seq = seq_max(device->peer_seq, peer_seq);
2329                device->peer_seq = newest_peer_seq;
2330                spin_unlock(&device->peer_seq_lock);
2331                /* wake up only if we actually changed device->peer_seq */
2332                if (peer_seq == newest_peer_seq)
2333                        wake_up(&device->seq_wait);
2334        }
2335}
2336
2337static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
2338{
2339        return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
2340}
2341
2342/* maybe change sync_ee into interval trees as well? */
2343static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
2344{
2345        struct drbd_peer_request *rs_req;
2346        bool rv = false;
2347
2348        spin_lock_irq(&device->resource->req_lock);
2349        list_for_each_entry(rs_req, &device->sync_ee, w.list) {
2350                if (overlaps(peer_req->i.sector, peer_req->i.size,
2351                             rs_req->i.sector, rs_req->i.size)) {
2352                        rv = true;
2353                        break;
2354                }
2355        }
2356        spin_unlock_irq(&device->resource->req_lock);
2357
2358        return rv;
2359}
2360
2361/* Called from receive_Data.
2362 * Synchronize packets on sock with packets on msock.
2363 *
2364 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
2365 * packet traveling on msock, they are still processed in the order they have
2366 * been sent.
2367 *
2368 * Note: we don't care for Ack packets overtaking P_DATA packets.
2369 *
2370 * In case packet_seq is larger than device->peer_seq number, there are
2371 * outstanding packets on the msock. We wait for them to arrive.
2372 * In case we are the logically next packet, we update device->peer_seq
2373 * ourselves. Correctly handles 32bit wrap around.
2374 *
2375 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
2376 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
2377 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
2378 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
2379 *
2380 * returns 0 if we may process the packet,
2381 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
2382static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
2383{
2384        struct drbd_device *device = peer_device->device;
2385        DEFINE_WAIT(wait);
2386        long timeout;
2387        int ret = 0, tp;
2388
2389        if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
2390                return 0;
2391
2392        spin_lock(&device->peer_seq_lock);
2393        for (;;) {
2394                if (!seq_greater(peer_seq - 1, device->peer_seq)) {
2395                        device->peer_seq = seq_max(device->peer_seq, peer_seq);
2396                        break;
2397                }
2398
2399                if (signal_pending(current)) {
2400                        ret = -ERESTARTSYS;
2401                        break;
2402                }
2403
2404                rcu_read_lock();
2405                tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
2406                rcu_read_unlock();
2407
2408                if (!tp)
2409                        break;
2410
2411                /* Only need to wait if two_primaries is enabled */
2412                prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
2413                spin_unlock(&device->peer_seq_lock);
2414                rcu_read_lock();
2415                timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
2416                rcu_read_unlock();
2417                timeout = schedule_timeout(timeout);
2418                spin_lock(&device->peer_seq_lock);
2419                if (!timeout) {
2420                        ret = -ETIMEDOUT;
2421                        drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
2422                        break;
2423                }
2424        }
2425        spin_unlock(&device->peer_seq_lock);
2426        finish_wait(&device->seq_wait, &wait);
2427        return ret;
2428}
2429
2430/* see also bio_flags_to_wire()
2431 * DRBD_REQ_*, because we need to semantically map the flags to data packet
2432 * flags and back. We may replicate to other kernel versions. */
2433static unsigned long wire_flags_to_bio_flags(u32 dpf)
2434{
2435        return  (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
2436                (dpf & DP_FUA ? REQ_FUA : 0) |
2437                (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
2438}
2439
2440static unsigned long wire_flags_to_bio_op(u32 dpf)
2441{
2442        if (dpf & DP_ZEROES)
2443                return REQ_OP_WRITE_ZEROES;
2444        if (dpf & DP_DISCARD)
2445                return REQ_OP_DISCARD;
2446        if (dpf & DP_WSAME)
2447                return REQ_OP_WRITE_SAME;
2448        else
2449                return REQ_OP_WRITE;
2450}
2451
2452static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
2453                                    unsigned int size)
2454{
2455        struct drbd_interval *i;
2456
2457    repeat:
2458        drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2459                struct drbd_request *req;
2460                struct bio_and_error m;
2461
2462                if (!i->local)
2463                        continue;
2464                req = container_of(i, struct drbd_request, i);
2465                if (!(req->rq_state & RQ_POSTPONED))
2466                        continue;
2467                req->rq_state &= ~RQ_POSTPONED;
2468                __req_mod(req, NEG_ACKED, &m);
2469                spin_unlock_irq(&device->resource->req_lock);
2470                if (m.bio)
2471                        complete_master_bio(device, &m);
2472                spin_lock_irq(&device->resource->req_lock);
2473                goto repeat;
2474        }
2475}
2476
2477static int handle_write_conflicts(struct drbd_device *device,
2478                                  struct drbd_peer_request *peer_req)
2479{
2480        struct drbd_connection *connection = peer_req->peer_device->connection;
2481        bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
2482        sector_t sector = peer_req->i.sector;
2483        const unsigned int size = peer_req->i.size;
2484        struct drbd_interval *i;
2485        bool equal;
2486        int err;
2487
2488        /*
2489         * Inserting the peer request into the write_requests tree will prevent
2490         * new conflicting local requests from being added.
2491         */
2492        drbd_insert_interval(&device->write_requests, &peer_req->i);
2493
2494    repeat:
2495        drbd_for_each_overlap(i, &device->write_requests, sector, size) {
2496                if (i == &peer_req->i)
2497                        continue;
2498                if (i->completed)
2499                        continue;
2500
2501                if (!i->local) {
2502                        /*
2503                         * Our peer has sent a conflicting remote request; this
2504                         * should not happen in a two-node setup.  Wait for the
2505                         * earlier peer request to complete.
2506                         */
2507                        err = drbd_wait_misc(device, i);
2508                        if (err)
2509                                goto out;
2510                        goto repeat;
2511                }
2512
2513                equal = i->sector == sector && i->size == size;
2514                if (resolve_conflicts) {
2515                        /*
2516                         * If the peer request is fully contained within the
2517                         * overlapping request, it can be considered overwritten
2518                         * and thus superseded; otherwise, it will be retried
2519                         * once all overlapping requests have completed.
2520                         */
2521                        bool superseded = i->sector <= sector && i->sector +
2522                                       (i->size >> 9) >= sector + (size >> 9);
2523
2524                        if (!equal)
2525                                drbd_alert(device, "Concurrent writes detected: "
2526                                               "local=%llus +%u, remote=%llus +%u, "
2527                                               "assuming %s came first\n",
2528                                          (unsigned long long)i->sector, i->size,
2529                                          (unsigned long long)sector, size,
2530                                          superseded ? "local" : "remote");
2531
2532                        peer_req->w.cb = superseded ? e_send_superseded :
2533                                                   e_send_retry_write;
2534                        list_add_tail(&peer_req->w.list, &device->done_ee);
2535                        queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);
2536
2537                        err = -ENOENT;
2538                        goto out;
2539                } else {
2540                        struct drbd_request *req =
2541                                container_of(i, struct drbd_request, i);
2542
2543                        if (!equal)
2544                                drbd_alert(device, "Concurrent writes detected: "
2545                                               "local=%llus +%u, remote=%llus +%u\n",
2546                                          (unsigned long long)i->sector, i->size,
2547                                          (unsigned long long)sector, size);
2548
2549                        if (req->rq_state & RQ_LOCAL_PENDING ||
2550                            !(req->rq_state & RQ_POSTPONED)) {
2551                                /*
2552                                 * Wait for the node with the discard flag to
2553                                 * decide if this request has been superseded
2554                                 * or needs to be retried.
2555                                 * Requests that have been superseded will
2556                                 * disappear from the write_requests tree.
2557                                 *
2558                                 * In addition, wait for the conflicting
2559                                 * request to finish locally before submitting
2560                                 * the conflicting peer request.
2561                                 */
2562                                err = drbd_wait_misc(device, &req->i);
2563                                if (err) {
2564                                        _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
2565                                        fail_postponed_requests(device, sector, size);
2566                                        goto out;
2567                                }
2568                                goto repeat;
2569                        }
2570                        /*
2571                         * Remember to restart the conflicting requests after
2572                         * the new peer request has completed.
2573                         */
2574                        peer_req->flags |= EE_RESTART_REQUESTS;
2575                }
2576        }
2577        err = 0;
2578
2579    out:
2580        if (err)
2581                drbd_remove_epoch_entry_interval(device, peer_req);
2582        return err;
2583}
2584
2585/* mirrored write */
2586static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
2587{
2588        struct drbd_peer_device *peer_device;
2589        struct drbd_device *device;
2590        struct net_conf *nc;
2591        sector_t sector;
2592        struct drbd_peer_request *peer_req;
2593        struct p_data *p = pi->data;
2594        u32 peer_seq = be32_to_cpu(p->seq_num);
2595        int op, op_flags;
2596        u32 dp_flags;
2597        int err, tp;
2598
2599        peer_device = conn_peer_device(connection, pi->vnr);
2600        if (!peer_device)
2601                return -EIO;
2602        device = peer_device->device;
2603
2604        if (!get_ldev(device)) {
2605                int err2;
2606
2607                err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2608                drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
2609                atomic_inc(&connection->current_epoch->epoch_size);
2610                err2 = drbd_drain_block(peer_device, pi->size);
2611                if (!err)
2612                        err = err2;
2613                return err;
2614        }
2615
2616        /*
2617         * Corresponding put_ldev done either below (on various errors), or in
2618         * drbd_peer_request_endio, if we successfully submit the data at the
2619         * end of this function.
2620         */
2621
2622        sector = be64_to_cpu(p->sector);
2623        peer_req = read_in_block(peer_device, p->block_id, sector, pi);
2624        if (!peer_req) {
2625                put_ldev(device);
2626                return -EIO;
2627        }
2628
2629        peer_req->w.cb = e_end_block;
2630        peer_req->submit_jif = jiffies;
2631        peer_req->flags |= EE_APPLICATION;
2632
2633        dp_flags = be32_to_cpu(p->dp_flags);
2634        op = wire_flags_to_bio_op(dp_flags);
2635        op_flags = wire_flags_to_bio_flags(dp_flags);
2636        if (pi->cmd == P_TRIM) {
2637                D_ASSERT(peer_device, peer_req->i.size > 0);
2638                D_ASSERT(peer_device, op == REQ_OP_DISCARD);
2639                D_ASSERT(peer_device, peer_req->pages == NULL);
2640                /* need to play safe: an older DRBD sender
2641                 * may mean zero-out while sending P_TRIM. */
2642                if (0 == (connection->agreed_features & DRBD_FF_WZEROES))
2643                        peer_req->flags |= EE_ZEROOUT;
2644        } else if (pi->cmd == P_ZEROES) {
2645                D_ASSERT(peer_device, peer_req->i.size > 0);
2646                D_ASSERT(peer_device, op == REQ_OP_WRITE_ZEROES);
2647                D_ASSERT(peer_device, peer_req->pages == NULL);
2648                /* Do (not) pass down BLKDEV_ZERO_NOUNMAP? */
2649                if (dp_flags & DP_DISCARD)
2650                        peer_req->flags |= EE_TRIM;
2651        } else if (peer_req->pages == NULL) {
2652                D_ASSERT(device, peer_req->i.size == 0);
2653                D_ASSERT(device, dp_flags & DP_FLUSH);
2654        }
2655
2656        if (dp_flags & DP_MAY_SET_IN_SYNC)
2657                peer_req->flags |= EE_MAY_SET_IN_SYNC;
2658
2659        spin_lock(&connection->epoch_lock);
2660        peer_req->epoch = connection->current_epoch;
2661        atomic_inc(&peer_req->epoch->epoch_size);
2662        atomic_inc(&peer_req->epoch->active);
2663        spin_unlock(&connection->epoch_lock);
2664
2665        rcu_read_lock();
2666        nc = rcu_dereference(peer_device->connection->net_conf);
2667        tp = nc->two_primaries;
2668        if (peer_device->connection->agreed_pro_version < 100) {
2669                switch (nc->wire_protocol) {
2670                case DRBD_PROT_C:
2671                        dp_flags |= DP_SEND_WRITE_ACK;
2672                        break;
2673                case DRBD_PROT_B:
2674                        dp_flags |= DP_SEND_RECEIVE_ACK;
2675                        break;
2676                }
2677        }
2678        rcu_read_unlock();
2679
2680        if (dp_flags & DP_SEND_WRITE_ACK) {
2681                peer_req->flags |= EE_SEND_WRITE_ACK;
2682                inc_unacked(device);
2683                /* corresponding dec_unacked() in e_end_block()
2684                 * respective _drbd_clear_done_ee */
2685        }
2686
2687        if (dp_flags & DP_SEND_RECEIVE_ACK) {
2688                /* I really don't like it that the receiver thread
2689                 * sends on the msock, but anyways */
2690                drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
2691        }
2692
2693        if (tp) {
2694                /* two primaries implies protocol C */
2695                D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
2696                peer_req->flags |= EE_IN_INTERVAL_TREE;
2697                err = wait_for_and_update_peer_seq(peer_device, peer_seq);
2698                if (err)
2699                        goto out_interrupted;
2700                spin_lock_irq(&device->resource->req_lock);
2701                err = handle_write_conflicts(device, peer_req);
2702                if (err) {
2703                        spin_unlock_irq(&device->resource->req_lock);
2704                        if (err == -ENOENT) {
2705                                put_ldev(device);
2706                                return 0;
2707                        }
2708                        goto out_interrupted;
2709                }
2710        } else {
2711                update_peer_seq(peer_device, peer_seq);
2712                spin_lock_irq(&device->resource->req_lock);
2713        }
2714        /* TRIM and WRITE_SAME are processed synchronously,
2715         * we wait for all pending requests, respectively wait for
2716         * active_ee to become empty in drbd_submit_peer_request();
2717         * better not add ourselves here. */
2718        if ((peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) == 0)
2719                list_add_tail(&peer_req->w.list, &device->active_ee);
2720        spin_unlock_irq(&device->resource->req_lock);
2721
2722        if (device->state.conn == C_SYNC_TARGET)
2723                wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));
2724
2725        if (device->state.pdsk < D_INCONSISTENT) {
2726                /* In case we have the only disk of the cluster, */
2727                drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
2728                peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
2729                drbd_al_begin_io(device, &peer_req->i);
2730                peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
2731        }
2732
2733        err = drbd_submit_peer_request(device, peer_req, op, op_flags,
2734                                       DRBD_FAULT_DT_WR);
2735        if (!err)
2736                return 0;
2737
2738        /* don't care for the reason here */
2739        drbd_err(device, "submit failed, triggering re-connect\n");
2740        spin_lock_irq(&device->resource->req_lock);
2741        list_del(&peer_req->w.list);
2742        drbd_remove_epoch_entry_interval(device, peer_req);
2743        spin_unlock_irq(&device->resource->req_lock);
2744        if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
2745                peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
2746                drbd_al_complete_io(device, &peer_req->i);
2747        }
2748
2749out_interrupted:
2750        drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
2751        put_ldev(device);
2752        drbd_free_peer_req(device, peer_req);
2753        return err;
2754}
2755
2756/* We may throttle resync, if the lower device seems to be busy,
2757 * and current sync rate is above c_min_rate.
2758 *
2759 * To decide whether or not the lower device is busy, we use a scheme similar
2760 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
2761 * (more than 64 sectors) of activity we cannot account for with our own resync
2762 * activity, it obviously is "busy".
2763 *
2764 * The current sync rate used here uses only the most recent two step marks,
2765 * to have a short time average so we can react faster.
2766 */
2767bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
2768                bool throttle_if_app_is_waiting)
2769{
2770        struct lc_element *tmp;
2771        bool throttle = drbd_rs_c_min_rate_throttle(device);
2772
2773        if (!throttle || throttle_if_app_is_waiting)
2774                return throttle;
2775
2776        spin_lock_irq(&device->al_lock);
2777        tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
2778        if (tmp) {
2779                struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
2780                if (test_bit(BME_PRIORITY, &bm_ext->flags))
2781                        throttle = false;
2782                /* Do not slow down if app IO is already waiting for this extent,
2783                 * and our progress is necessary for application IO to complete. */
2784        }
2785        spin_unlock_irq(&device->al_lock);
2786
2787        return throttle;
2788}
2789
2790bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
2791{
2792        struct gendisk *disk = device->ldev->backing_bdev->bd_disk;
2793        unsigned long db, dt, dbdt;
2794        unsigned int c_min_rate;
2795        int curr_events;
2796
2797        rcu_read_lock();
2798        c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
2799        rcu_read_unlock();
2800
2801        /* feature disabled? */
2802        if (c_min_rate == 0)
2803                return false;
2804
2805        curr_events = (int)part_stat_read_accum(disk->part0, sectors) -
2806                        atomic_read(&device->rs_sect_ev);
2807
2808        if (atomic_read(&device->ap_actlog_cnt)
2809            || curr_events - device->rs_last_events > 64) {
2810                unsigned long rs_left;
2811                int i;
2812
2813                device->rs_last_events = curr_events;
2814
2815                /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
2816                 * approx. */
2817                i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;
2818
2819                if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
2820                        rs_left = device->ov_left;
2821                else
2822                        rs_left = drbd_bm_total_weight(device) - device->rs_failed;
2823
2824                dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
2825                if (!dt)
2826                        dt++;
2827                db = device->rs_mark_left[i] - rs_left;
2828                dbdt = Bit2KB(db/dt);
2829
2830                if (dbdt > c_min_rate)
2831                        return true;
2832        }
2833        return false;
2834}
2835
2836static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
2837{
2838        struct drbd_peer_device *peer_device;
2839        struct drbd_device *device;
2840        sector_t sector;
2841        sector_t capacity;
2842        struct drbd_peer_request *peer_req;
2843        struct digest_info *di = NULL;
2844        int size, verb;
2845        unsigned int fault_type;
2846        struct p_block_req *p = pi->data;
2847
2848        peer_device = conn_peer_device(connection, pi->vnr);
2849        if (!peer_device)
2850                return -EIO;
2851        device = peer_device->device;
2852        capacity = get_capacity(device->vdisk);
2853
2854        sector = be64_to_cpu(p->sector);
2855        size   = be32_to_cpu(p->blksize);
2856
2857        if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
2858                drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2859                                (unsigned long long)sector, size);
2860                return -EINVAL;
2861        }
2862        if (sector + (size>>9) > capacity) {
2863                drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
2864                                (unsigned long long)sector, size);
2865                return -EINVAL;
2866        }
2867
2868        if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
2869                verb = 1;
2870                switch (pi->cmd) {
2871                case P_DATA_REQUEST:
2872                        drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
2873                        break;
2874                case P_RS_THIN_REQ:
2875                case P_RS_DATA_REQUEST:
2876                case P_CSUM_RS_REQUEST:
2877                case P_OV_REQUEST:
2878                        drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
2879                        break;
2880                case P_OV_REPLY:
2881                        verb = 0;
2882                        dec_rs_pending(device);
2883                        drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
2884                        break;
2885                default:
2886                        BUG();
2887                }
2888                if (verb && __ratelimit(&drbd_ratelimit_state))
2889                        drbd_err(device, "Can not satisfy peer's read request, "
2890                            "no local data.\n");
2891
2892                /* drain possibly payload */
2893                return drbd_drain_block(peer_device, pi->size);
2894        }
2895
2896        /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
2897         * "criss-cross" setup, that might cause write-out on some other DRBD,
2898         * which in turn might block on the other node at this very place.  */
2899        peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
2900                        size, GFP_NOIO);
2901        if (!peer_req) {
2902                put_ldev(device);
2903                return -ENOMEM;
2904        }
2905
2906        switch (pi->cmd) {
2907        case P_DATA_REQUEST:
2908                peer_req->w.cb = w_e_end_data_req;
2909                fault_type = DRBD_FAULT_DT_RD;
2910                /* application IO, don't drbd_rs_begin_io */
2911                peer_req->flags |= EE_APPLICATION;
2912                goto submit;
2913
2914        case P_RS_THIN_REQ:
2915                /* If at some point in the future we have a smart way to
2916                   find out if this data block is completely deallocated,
2917                   then we would do something smarter here than reading
2918                   the block... */
2919                peer_req->flags |= EE_RS_THIN_REQ;
2920                fallthrough;
2921        case P_RS_DATA_REQUEST:
2922                peer_req->w.cb = w_e_end_rsdata_req;
2923                fault_type = DRBD_FAULT_RS_RD;
2924                /* used in the sector offset progress display */
2925                device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2926                break;
2927
2928        case P_OV_REPLY:
2929        case P_CSUM_RS_REQUEST:
2930                fault_type = DRBD_FAULT_RS_RD;
2931                di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
2932                if (!di)
2933                        goto out_free_e;
2934
2935                di->digest_size = pi->size;
2936                di->digest = (((char *)di)+sizeof(struct digest_info));
2937
2938                peer_req->digest = di;
2939                peer_req->flags |= EE_HAS_DIGEST;
2940
2941                if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
2942                        goto out_free_e;
2943
2944                if (pi->cmd == P_CSUM_RS_REQUEST) {
2945                        D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
2946                        peer_req->w.cb = w_e_end_csum_rs_req;
2947                        /* used in the sector offset progress display */
2948                        device->bm_resync_fo = BM_SECT_TO_BIT(sector);
2949                        /* remember to report stats in drbd_resync_finished */
2950                        device->use_csums = true;
2951                } else if (pi->cmd == P_OV_REPLY) {
2952                        /* track progress, we may need to throttle */
2953                        atomic_add(size >> 9, &device->rs_sect_in);
2954                        peer_req->w.cb = w_e_end_ov_reply;
2955                        dec_rs_pending(device);
2956                        /* drbd_rs_begin_io done when we sent this request,
2957                         * but accounting still needs to be done. */
2958                        goto submit_for_resync;
2959                }
2960                break;
2961
2962        case P_OV_REQUEST:
2963                if (device->ov_start_sector == ~(sector_t)0 &&
2964                    peer_device->connection->agreed_pro_version >= 90) {
2965                        unsigned long now = jiffies;
2966                        int i;
2967                        device->ov_start_sector = sector;
2968                        device->ov_position = sector;
2969                        device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
2970                        device->rs_total = device->ov_left;
2971                        for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2972                                device->rs_mark_left[i] = device->ov_left;
2973                                device->rs_mark_time[i] = now;
2974                        }
2975                        drbd_info(device, "Online Verify start sector: %llu\n",
2976                                        (unsigned long long)sector);
2977                }
2978                peer_req->w.cb = w_e_end_ov_req;
2979                fault_type = DRBD_FAULT_RS_RD;
2980                break;
2981
2982        default:
2983                BUG();
2984        }
2985
2986        /* Throttle, drbd_rs_begin_io and submit should become asynchronous
2987         * wrt the receiver, but it is not as straightforward as it may seem.
2988         * Various places in the resync start and stop logic assume resync
2989         * requests are processed in order, requeuing this on the worker thread
2990         * introduces a bunch of new code for synchronization between threads.
2991         *
2992         * Unlimited throttling before drbd_rs_begin_io may stall the resync
2993         * "forever", throttling after drbd_rs_begin_io will lock that extent
2994         * for application writes for the same time.  For now, just throttle
2995         * here, where the rest of the code expects the receiver to sleep for
2996         * a while, anyways.
2997         */
2998
2999        /* Throttle before drbd_rs_begin_io, as that locks out application IO;
3000         * this defers syncer requests for some time, before letting at least
3001         * on request through.  The resync controller on the receiving side
3002         * will adapt to the incoming rate accordingly.
3003         *
3004         * We cannot throttle here if remote is Primary/SyncTarget:
3005         * we would also throttle its application reads.
3006         * In that case, throttling is done on the SyncTarget only.
3007         */
3008
3009        /* Even though this may be a resync request, we do add to "read_ee";
3010         * "sync_ee" is only used for resync WRITEs.
3011         * Add to list early, so debugfs can find this request
3012         * even if we have to sleep below. */
3013        spin_lock_irq(&device->resource->req_lock);
3014        list_add_tail(&peer_req->w.list, &device->read_ee);
3015        spin_unlock_irq(&device->resource->req_lock);
3016
3017        update_receiver_timing_details(connection, drbd_rs_should_slow_down);
3018        if (device->state.peer != R_PRIMARY
3019        && drbd_rs_should_slow_down(device, sector, false))
3020                schedule_timeout_uninterruptible(HZ/10);
3021        update_receiver_timing_details(connection, drbd_rs_begin_io);
3022        if (drbd_rs_begin_io(device, sector))
3023                goto out_free_e;
3024
3025submit_for_resync:
3026        atomic_add(size >> 9, &device->rs_sect_ev);
3027
3028submit:
3029        update_receiver_timing_details(connection, drbd_submit_peer_request);
3030        inc_unacked(device);
3031        if (drbd_submit_peer_request(device, peer_req, REQ_OP_READ, 0,
3032                                     fault_type) == 0)
3033                return 0;
3034
3035        /* don't care for the reason here */
3036        drbd_err(device, "submit failed, triggering re-connect\n");
3037
3038out_free_e:
3039        spin_lock_irq(&device->resource->req_lock);
3040        list_del(&peer_req->w.list);
3041        spin_unlock_irq(&device->resource->req_lock);
3042        /* no drbd_rs_complete_io(), we are dropping the connection anyways */
3043
3044        put_ldev(device);
3045        drbd_free_peer_req(device, peer_req);
3046        return -EIO;
3047}
3048
3049/*
3050 * drbd_asb_recover_0p  -  Recover after split-brain with no remaining primaries
3051 */
3052static int drbd_asb_recover_0p(struct drbd_peer_device *peer_device) __must_hold(local)
3053{
3054        struct drbd_device *device = peer_device->device;
3055        int self, peer, rv = -100;
3056        unsigned long ch_self, ch_peer;
3057        enum drbd_after_sb_p after_sb_0p;
3058
3059        self = device->ldev->md.uuid[UI_BITMAP] & 1;
3060        peer = device->p_uuid[UI_BITMAP] & 1;
3061
3062        ch_peer = device->p_uuid[UI_SIZE];
3063        ch_self = device->comm_bm_set;
3064
3065        rcu_read_lock();
3066        after_sb_0p = rcu_dereference(peer_device->connection->net_conf)->after_sb_0p;
3067        rcu_read_unlock();
3068        switch (after_sb_0p) {
3069        case ASB_CONSENSUS:
3070        case ASB_DISCARD_SECONDARY:
3071        case ASB_CALL_HELPER:
3072        case ASB_VIOLENTLY:
3073                drbd_err(device, "Configuration error.\n");
3074                break;
3075        case ASB_DISCONNECT:
3076                break;
3077        case ASB_DISCARD_YOUNGER_PRI:
3078                if (self == 0 && peer == 1) {
3079                        rv = -1;
3080                        break;
3081                }
3082                if (self == 1 && peer == 0) {
3083                        rv =  1;
3084                        break;
3085                }
3086                fallthrough;    /* to one of the other strategies */
3087        case ASB_DISCARD_OLDER_PRI:
3088                if (self == 0 && peer == 1) {
3089                        rv = 1;
3090                        break;
3091                }
3092                if (self == 1 && peer == 0) {
3093                        rv = -1;
3094                        break;
3095                }
3096                /* Else fall through to one of the other strategies... */
3097                drbd_warn(device, "Discard younger/older primary did not find a decision\n"
3098                     "Using discard-least-changes instead\n");
3099                fallthrough;
3100        case ASB_DISCARD_ZERO_CHG:
3101                if (ch_peer == 0 && ch_self == 0) {
3102                        rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3103                                ? -1 : 1;
3104                        break;
3105                } else {
3106                        if (ch_peer == 0) { rv =  1; break; }
3107                        if (ch_self == 0) { rv = -1; break; }
3108                }
3109                if (after_sb_0p == ASB_DISCARD_ZERO_CHG)
3110                        break;
3111                fallthrough;
3112        case ASB_DISCARD_LEAST_CHG:
3113                if      (ch_self < ch_peer)
3114                        rv = -1;
3115                else if (ch_self > ch_peer)
3116                        rv =  1;
3117                else /* ( ch_self == ch_peer ) */
3118                     /* Well, then use something else. */
3119                        rv = test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)
3120                                ? -1 : 1;
3121                break;
3122        case ASB_DISCARD_LOCAL:
3123                rv = -1;
3124                break;
3125        case ASB_DISCARD_REMOTE:
3126                rv =  1;
3127        }
3128
3129        return rv;
3130}
3131
3132/*
3133 * drbd_asb_recover_1p  -  Recover after split-brain with one remaining primary
3134 */
3135static int drbd_asb_recover_1p(struct drbd_peer_device *peer_device) __must_hold(local)
3136{
3137        struct drbd_device *device = peer_device->device;
3138        int hg, rv = -100;
3139        enum drbd_after_sb_p after_sb_1p;
3140
3141        rcu_read_lock();
3142        after_sb_1p = rcu_dereference(peer_device->connection->net_conf)->after_sb_1p;
3143        rcu_read_unlock();
3144        switch (after_sb_1p) {
3145        case ASB_DISCARD_YOUNGER_PRI:
3146        case ASB_DISCARD_OLDER_PRI:
3147        case ASB_DISCARD_LEAST_CHG:
3148        case ASB_DISCARD_LOCAL:
3149        case ASB_DISCARD_REMOTE:
3150        case ASB_DISCARD_ZERO_CHG:
3151                drbd_err(device, "Configuration error.\n");
3152                break;
3153        case ASB_DISCONNECT:
3154                break;
3155        case ASB_CONSENSUS:
3156                hg = drbd_asb_recover_0p(peer_device);
3157                if (hg == -1 && device->state.role == R_SECONDARY)
3158                        rv = hg;
3159                if (hg == 1  && device->state.role == R_PRIMARY)
3160                        rv = hg;
3161                break;
3162        case ASB_VIOLENTLY:
3163                rv = drbd_asb_recover_0p(peer_device);
3164                break;
3165        case ASB_DISCARD_SECONDARY:
3166                return device->state.role == R_PRIMARY ? 1 : -1;
3167        case ASB_CALL_HELPER:
3168                hg = drbd_asb_recover_0p(peer_device);
3169                if (hg == -1 && device->state.role == R_PRIMARY) {
3170                        enum drbd_state_rv rv2;
3171
3172                         /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3173                          * we might be here in C_WF_REPORT_PARAMS which is transient.
3174                          * we do not need to wait for the after state change work either. */
3175                        rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3176                        if (rv2 != SS_SUCCESS) {
3177                                drbd_khelper(device, "pri-lost-after-sb");
3178                        } else {
3179                                drbd_warn(device, "Successfully gave up primary role.\n");
3180                                rv = hg;
3181                        }
3182                } else
3183                        rv = hg;
3184        }
3185
3186        return rv;
3187}
3188
3189/*
3190 * drbd_asb_recover_2p  -  Recover after split-brain with two remaining primaries
3191 */
3192static int drbd_asb_recover_2p(struct drbd_peer_device *peer_device) __must_hold(local)
3193{
3194        struct drbd_device *device = peer_device->device;
3195        int hg, rv = -100;
3196        enum drbd_after_sb_p after_sb_2p;
3197
3198        rcu_read_lock();
3199        after_sb_2p = rcu_dereference(peer_device->connection->net_conf)->after_sb_2p;
3200        rcu_read_unlock();
3201        switch (after_sb_2p) {
3202        case ASB_DISCARD_YOUNGER_PRI:
3203        case ASB_DISCARD_OLDER_PRI:
3204        case ASB_DISCARD_LEAST_CHG:
3205        case ASB_DISCARD_LOCAL:
3206        case ASB_DISCARD_REMOTE:
3207        case ASB_CONSENSUS:
3208        case ASB_DISCARD_SECONDARY:
3209        case ASB_DISCARD_ZERO_CHG:
3210                drbd_err(device, "Configuration error.\n");
3211                break;
3212        case ASB_VIOLENTLY:
3213                rv = drbd_asb_recover_0p(peer_device);
3214                break;
3215        case ASB_DISCONNECT:
3216                break;
3217        case ASB_CALL_HELPER:
3218                hg = drbd_asb_recover_0p(peer_device);
3219                if (hg == -1) {
3220                        enum drbd_state_rv rv2;
3221
3222                         /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE,
3223                          * we might be here in C_WF_REPORT_PARAMS which is transient.
3224                          * we do not need to wait for the after state change work either. */
3225                        rv2 = drbd_change_state(device, CS_VERBOSE, NS(role, R_SECONDARY));
3226                        if (rv2 != SS_SUCCESS) {
3227                                drbd_khelper(device, "pri-lost-after-sb");
3228                        } else {
3229                                drbd_warn(device, "Successfully gave up primary role.\n");
3230                                rv = hg;
3231                        }
3232                } else
3233                        rv = hg;
3234        }
3235
3236        return rv;
3237}
3238
3239static void drbd_uuid_dump(struct drbd_device *device, char *text, u64 *uuid,
3240                           u64 bits, u64 flags)
3241{
3242        if (!uuid) {
3243                drbd_info(device, "%s uuid info vanished while I was looking!\n", text);
3244                return;
3245        }
3246        drbd_info(device, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n",
3247             text,
3248             (unsigned long long)uuid[UI_CURRENT],
3249             (unsigned long long)uuid[UI_BITMAP],
3250             (unsigned long long)uuid[UI_HISTORY_START],
3251             (unsigned long long)uuid[UI_HISTORY_END],
3252             (unsigned long long)bits,
3253             (unsigned long long)flags);
3254}
3255
3256/*
3257  100   after split brain try auto recover
3258    2   C_SYNC_SOURCE set BitMap
3259    1   C_SYNC_SOURCE use BitMap
3260    0   no Sync
3261   -1   C_SYNC_TARGET use BitMap
3262   -2   C_SYNC_TARGET set BitMap
3263 -100   after split brain, disconnect
3264-1000   unrelated data
3265-1091   requires proto 91
3266-1096   requires proto 96
3267 */
3268
3269static int drbd_uuid_compare(struct drbd_device *const device, enum drbd_role const peer_role, int *rule_nr) __must_hold(local)
3270{
3271        struct drbd_peer_device *const peer_device = first_peer_device(device);
3272        struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
3273        u64 self, peer;
3274        int i, j;
3275
3276        self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3277        peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3278
3279        *rule_nr = 10;
3280        if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED)
3281                return 0;
3282
3283        *rule_nr = 20;
3284        if ((self == UUID_JUST_CREATED || self == (u64)0) &&
3285             peer != UUID_JUST_CREATED)
3286                return -2;
3287
3288        *rule_nr = 30;
3289        if (self != UUID_JUST_CREATED &&
3290            (peer == UUID_JUST_CREATED || peer == (u64)0))
3291                return 2;
3292
3293        if (self == peer) {
3294                int rct, dc; /* roles at crash time */
3295
3296                if (device->p_uuid[UI_BITMAP] == (u64)0 && device->ldev->md.uuid[UI_BITMAP] != (u64)0) {
3297
3298                        if (connection->agreed_pro_version < 91)
3299                                return -1091;
3300
3301                        if ((device->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) &&
3302                            (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) {
3303                                drbd_info(device, "was SyncSource, missed the resync finished event, corrected myself:\n");
3304                                drbd_uuid_move_history(device);
3305                                device->ldev->md.uuid[UI_HISTORY_START] = device->ldev->md.uuid[UI_BITMAP];
3306                                device->ldev->md.uuid[UI_BITMAP] = 0;
3307
3308                                drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3309                                               device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3310                                *rule_nr = 34;
3311                        } else {
3312                                drbd_info(device, "was SyncSource (peer failed to write sync_uuid)\n");
3313                                *rule_nr = 36;
3314                        }
3315
3316                        return 1;
3317                }
3318
3319                if (device->ldev->md.uuid[UI_BITMAP] == (u64)0 && device->p_uuid[UI_BITMAP] != (u64)0) {
3320
3321                        if (connection->agreed_pro_version < 91)
3322                                return -1091;
3323
3324                        if ((device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (device->p_uuid[UI_BITMAP] & ~((u64)1)) &&
3325                            (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (device->p_uuid[UI_HISTORY_START] & ~((u64)1))) {
3326                                drbd_info(device, "was SyncTarget, peer missed the resync finished event, corrected peer:\n");
3327
3328                                device->p_uuid[UI_HISTORY_START + 1] = device->p_uuid[UI_HISTORY_START];
3329                                device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_BITMAP];
3330                                device->p_uuid[UI_BITMAP] = 0UL;
3331
3332                                drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3333                                *rule_nr = 35;
3334                        } else {
3335                                drbd_info(device, "was SyncTarget (failed to write sync_uuid)\n");
3336                                *rule_nr = 37;
3337                        }
3338
3339                        return -1;
3340                }
3341
3342                /* Common power [off|failure] */
3343                rct = (test_bit(CRASHED_PRIMARY, &device->flags) ? 1 : 0) +
3344                        (device->p_uuid[UI_FLAGS] & 2);
3345                /* lowest bit is set when we were primary,
3346                 * next bit (weight 2) is set when peer was primary */
3347                *rule_nr = 40;
3348
3349                /* Neither has the "crashed primary" flag set,
3350                 * only a replication link hickup. */
3351                if (rct == 0)
3352                        return 0;
3353
3354                /* Current UUID equal and no bitmap uuid; does not necessarily
3355                 * mean this was a "simultaneous hard crash", maybe IO was
3356                 * frozen, so no UUID-bump happened.
3357                 * This is a protocol change, overload DRBD_FF_WSAME as flag
3358                 * for "new-enough" peer DRBD version. */
3359                if (device->state.role == R_PRIMARY || peer_role == R_PRIMARY) {
3360                        *rule_nr = 41;
3361                        if (!(connection->agreed_features & DRBD_FF_WSAME)) {
3362                                drbd_warn(peer_device, "Equivalent unrotated UUIDs, but current primary present.\n");
3363                                return -(0x10000 | PRO_VERSION_MAX | (DRBD_FF_WSAME << 8));
3364                        }
3365                        if (device->state.role == R_PRIMARY && peer_role == R_PRIMARY) {
3366                                /* At least one has the "crashed primary" bit set,
3367                                 * both are primary now, but neither has rotated its UUIDs?
3368                                 * "Can not happen." */
3369                                drbd_err(peer_device, "Equivalent unrotated UUIDs, but both are primary. Can not resolve this.\n");
3370                                return -100;
3371                        }
3372                        if (device->state.role == R_PRIMARY)
3373                                return 1;
3374                        return -1;
3375                }
3376
3377                /* Both are secondary.
3378                 * Really looks like recovery from simultaneous hard crash.
3379                 * Check which had been primary before, and arbitrate. */
3380                switch (rct) {
3381                case 0: /* !self_pri && !peer_pri */ return 0; /* already handled */
3382                case 1: /*  self_pri && !peer_pri */ return 1;
3383                case 2: /* !self_pri &&  peer_pri */ return -1;
3384                case 3: /*  self_pri &&  peer_pri */
3385                        dc = test_bit(RESOLVE_CONFLICTS, &connection->flags);
3386                        return dc ? -1 : 1;
3387                }
3388        }
3389
3390        *rule_nr = 50;
3391        peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3392        if (self == peer)
3393                return -1;
3394
3395        *rule_nr = 51;
3396        peer = device->p_uuid[UI_HISTORY_START] & ~((u64)1);
3397        if (self == peer) {
3398                if (connection->agreed_pro_version < 96 ?
3399                    (device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) ==
3400                    (device->p_uuid[UI_HISTORY_START + 1] & ~((u64)1)) :
3401                    peer + UUID_NEW_BM_OFFSET == (device->p_uuid[UI_BITMAP] & ~((u64)1))) {
3402                        /* The last P_SYNC_UUID did not get though. Undo the last start of
3403                           resync as sync source modifications of the peer's UUIDs. */
3404
3405                        if (connection->agreed_pro_version < 91)
3406                                return -1091;
3407
3408                        device->p_uuid[UI_BITMAP] = device->p_uuid[UI_HISTORY_START];
3409                        device->p_uuid[UI_HISTORY_START] = device->p_uuid[UI_HISTORY_START + 1];
3410
3411                        drbd_info(device, "Lost last syncUUID packet, corrected:\n");
3412                        drbd_uuid_dump(device, "peer", device->p_uuid, device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3413
3414                        return -1;
3415                }
3416        }
3417
3418        *rule_nr = 60;
3419        self = device->ldev->md.uuid[UI_CURRENT] & ~((u64)1);
3420        for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3421                peer = device->p_uuid[i] & ~((u64)1);
3422                if (self == peer)
3423                        return -2;
3424        }
3425
3426        *rule_nr = 70;
3427        self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3428        peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3429        if (self == peer)
3430                return 1;
3431
3432        *rule_nr = 71;
3433        self = device->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1);
3434        if (self == peer) {
3435                if (connection->agreed_pro_version < 96 ?
3436                    (device->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) ==
3437                    (device->p_uuid[UI_HISTORY_START] & ~((u64)1)) :
3438                    self + UUID_NEW_BM_OFFSET == (device->ldev->md.uuid[UI_BITMAP] & ~((u64)1))) {
3439                        /* The last P_SYNC_UUID did not get though. Undo the last start of
3440                           resync as sync source modifications of our UUIDs. */
3441
3442                        if (connection->agreed_pro_version < 91)
3443                                return -1091;
3444
3445                        __drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_HISTORY_START]);
3446                        __drbd_uuid_set(device, UI_HISTORY_START, device->ldev->md.uuid[UI_HISTORY_START + 1]);
3447
3448                        drbd_info(device, "Last syncUUID did not get through, corrected:\n");
3449                        drbd_uuid_dump(device, "self", device->ldev->md.uuid,
3450                                       device->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(device) : 0, 0);
3451
3452                        return 1;
3453                }
3454        }
3455
3456
3457        *rule_nr = 80;
3458        peer = device->p_uuid[UI_CURRENT] & ~((u64)1);
3459        for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3460                self = device->ldev->md.uuid[i] & ~((u64)1);
3461                if (self == peer)
3462                        return 2;
3463        }
3464
3465        *rule_nr = 90;
3466        self = device->ldev->md.uuid[UI_BITMAP] & ~((u64)1);
3467        peer = device->p_uuid[UI_BITMAP] & ~((u64)1);
3468        if (self == peer && self != ((u64)0))
3469                return 100;
3470
3471        *rule_nr = 100;
3472        for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) {
3473                self = device->ldev->md.uuid[i] & ~((u64)1);
3474                for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) {
3475                        peer = device->p_uuid[j] & ~((u64)1);
3476                        if (self == peer)
3477                                return -100;
3478                }
3479        }
3480
3481        return -1000;
3482}
3483
3484/* drbd_sync_handshake() returns the new conn state on success, or
3485   CONN_MASK (-1) on failure.
3486 */
3487static enum drbd_conns drbd_sync_handshake(struct drbd_peer_device *peer_device,
3488                                           enum drbd_role peer_role,
3489                                           enum drbd_disk_state peer_disk) __must_hold(local)
3490{
3491        struct drbd_device *device = peer_device->device;
3492        enum drbd_conns rv = C_MASK;
3493        enum drbd_disk_state mydisk;
3494        struct net_conf *nc;
3495        int hg, rule_nr, rr_conflict, tentative, always_asbp;
3496
3497        mydisk = device->state.disk;
3498        if (mydisk == D_NEGOTIATING)
3499                mydisk = device->new_state_tmp.disk;
3500
3501        drbd_info(device, "drbd_sync_handshake:\n");
3502
3503        spin_lock_irq(&device->ldev->md.uuid_lock);
3504        drbd_uuid_dump(device, "self", device->ldev->md.uuid, device->comm_bm_set, 0);
3505        drbd_uuid_dump(device, "peer", device->p_uuid,
3506                       device->p_uuid[UI_SIZE], device->p_uuid[UI_FLAGS]);
3507
3508        hg = drbd_uuid_compare(device, peer_role, &rule_nr);
3509        spin_unlock_irq(&device->ldev->md.uuid_lock);
3510
3511        drbd_info(device, "uuid_compare()=%d by rule %d\n", hg, rule_nr);
3512
3513        if (hg == -1000) {
3514                drbd_alert(device, "Unrelated data, aborting!\n");
3515                return C_MASK;
3516        }
3517        if (hg < -0x10000) {
3518                int proto, fflags;
3519                hg = -hg;
3520                proto = hg & 0xff;
3521                fflags = (hg >> 8) & 0xff;
3522                drbd_alert(device, "To resolve this both sides have to support at least protocol %d and feature flags 0x%x\n",
3523                                        proto, fflags);
3524                return C_MASK;
3525        }
3526        if (hg < -1000) {
3527                drbd_alert(device, "To resolve this both sides have to support at least protocol %d\n", -hg - 1000);
3528                return C_MASK;
3529        }
3530
3531        if    ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) ||
3532            (peer_disk == D_INCONSISTENT && mydisk    > D_INCONSISTENT)) {
3533                int f = (hg == -100) || abs(hg) == 2;
3534                hg = mydisk > D_INCONSISTENT ? 1 : -1;
3535                if (f)
3536                        hg = hg*2;
3537                drbd_info(device, "Becoming sync %s due to disk states.\n",
3538                     hg > 0 ? "source" : "target");
3539        }
3540
3541        if (abs(hg) == 100)
3542                drbd_khelper(device, "initial-split-brain");
3543
3544        rcu_read_lock();
3545        nc = rcu_dereference(peer_device->connection->net_conf);
3546        always_asbp = nc->always_asbp;
3547        rr_conflict = nc->rr_conflict;
3548        tentative = nc->tentative;
3549        rcu_read_unlock();
3550
3551        if (hg == 100 || (hg == -100 && always_asbp)) {
3552                int pcount = (device->state.role == R_PRIMARY)
3553                           + (peer_role == R_PRIMARY);
3554                int forced = (hg == -100);
3555
3556                switch (pcount) {
3557                case 0:
3558                        hg = drbd_asb_recover_0p(peer_device);
3559                        break;
3560                case 1:
3561                        hg = drbd_asb_recover_1p(peer_device);
3562                        break;
3563                case 2:
3564                        hg = drbd_asb_recover_2p(peer_device);
3565                        break;
3566                }
3567                if (abs(hg) < 100) {
3568                        drbd_warn(device, "Split-Brain detected, %d primaries, "
3569                             "automatically solved. Sync from %s node\n",
3570                             pcount, (hg < 0) ? "peer" : "this");
3571                        if (forced) {
3572                                drbd_warn(device, "Doing a full sync, since"
3573                                     " UUIDs where ambiguous.\n");
3574                                hg = hg*2;
3575                        }
3576                }
3577        }
3578
3579        if (hg == -100) {
3580                if (test_bit(DISCARD_MY_DATA, &device->flags) && !(device->p_uuid[UI_FLAGS]&1))
3581                        hg = -1;
3582                if (!test_bit(DISCARD_MY_DATA, &device->flags) && (device->p_uuid[UI_FLAGS]&1))
3583                        hg = 1;
3584
3585                if (abs(hg) < 100)
3586                        drbd_warn(device, "Split-Brain detected, manually solved. "
3587                             "Sync from %s node\n",
3588                             (hg < 0) ? "peer" : "this");
3589        }
3590
3591        if (hg == -100) {
3592                /* FIXME this log message is not correct if we end up here
3593                 * after an attempted attach on a diskless node.
3594                 * We just refuse to attach -- well, we drop the "connection"
3595                 * to that disk, in a way... */
3596                drbd_alert(device, "Split-Brain detected but unresolved, dropping connection!\n");
3597                drbd_khelper(device, "split-brain");
3598                return C_MASK;
3599        }
3600
3601        if (hg > 0 && mydisk <= D_INCONSISTENT) {
3602                drbd_err(device, "I shall become SyncSource, but I am inconsistent!\n");
3603                return C_MASK;
3604        }
3605
3606        if (hg < 0 && /* by intention we do not use mydisk here. */
3607            device->state.role == R_PRIMARY && device->state.disk >= D_CONSISTENT) {
3608                switch (rr_conflict) {
3609                case ASB_CALL_HELPER:
3610                        drbd_khelper(device, "pri-lost");
3611                        fallthrough;
3612                case ASB_DISCONNECT:
3613                        drbd_err(device, "I shall become SyncTarget, but I am primary!\n");
3614                        return C_MASK;
3615                case ASB_VIOLENTLY:
3616                        drbd_warn(device, "Becoming SyncTarget, violating the stable-data"
3617                             "assumption\n");
3618                }
3619        }
3620
3621        if (tentative || test_bit(CONN_DRY_RUN, &peer_device->connection->flags)) {
3622                if (hg == 0)
3623                        drbd_info(device, "dry-run connect: No resync, would become Connected immediately.\n");
3624                else
3625                        drbd_info(device, "dry-run connect: Would become %s, doing a %s resync.",
3626                                 drbd_conn_str(hg > 0 ? C_SYNC_SOURCE : C_SYNC_TARGET),
3627                                 abs(hg) >= 2 ? "full" : "bit-map based");
3628                return C_MASK;
3629        }
3630
3631        if (abs(hg) >= 2) {
3632                drbd_info(device, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n");
3633                if (drbd_bitmap_io(device, &drbd_bmio_set_n_write, "set_n_write from sync_handshake",
3634                                        BM_LOCKED_SET_ALLOWED))
3635                        return C_MASK;
3636        }
3637
3638        if (hg > 0) { /* become sync source. */
3639                rv = C_WF_BITMAP_S;
3640        } else if (hg < 0) { /* become sync target */
3641                rv = C_WF_BITMAP_T;
3642        } else {
3643                rv = C_CONNECTED;
3644                if (drbd_bm_total_weight(device)) {
3645                        drbd_info(device, "No resync, but %lu bits in bitmap!\n",
3646                             drbd_bm_total_weight(device));
3647                }
3648        }
3649
3650        return rv;
3651}
3652
3653static enum drbd_after_sb_p convert_after_sb(enum drbd_after_sb_p peer)
3654{
3655        /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */
3656        if (peer == ASB_DISCARD_REMOTE)
3657                return ASB_DISCARD_LOCAL;
3658
3659        /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */
3660        if (peer == ASB_DISCARD_LOCAL)
3661                return ASB_DISCARD_REMOTE;
3662
3663        /* everything else is valid if they are equal on both sides. */
3664        return peer;
3665}
3666
3667static int receive_protocol(struct drbd_connection *connection, struct packet_info *pi)
3668{
3669        struct p_protocol *p = pi->data;
3670        enum drbd_after_sb_p p_after_sb_0p, p_after_sb_1p, p_after_sb_2p;
3671        int p_proto, p_discard_my_data, p_two_primaries, cf;
3672        struct net_conf *nc, *old_net_conf, *new_net_conf = NULL;
3673        char integrity_alg[SHARED_SECRET_MAX] = "";
3674        struct crypto_shash *peer_integrity_tfm = NULL;
3675        void *int_dig_in = NULL, *int_dig_vv = NULL;
3676
3677        p_proto         = be32_to_cpu(p->protocol);
3678        p_after_sb_0p   = be32_to_cpu(p->after_sb_0p);
3679        p_after_sb_1p   = be32_to_cpu(p->after_sb_1p);
3680        p_after_sb_2p   = be32_to_cpu(p->after_sb_2p);
3681        p_two_primaries = be32_to_cpu(p->two_primaries);
3682        cf              = be32_to_cpu(p->conn_flags);
3683        p_discard_my_data = cf & CF_DISCARD_MY_DATA;
3684
3685        if (connection->agreed_pro_version >= 87) {
3686                int err;
3687
3688                if (pi->size > sizeof(integrity_alg))
3689                        return -EIO;
3690                err = drbd_recv_all(connection, integrity_alg, pi->size);
3691                if (err)
3692                        return err;
3693                integrity_alg[SHARED_SECRET_MAX - 1] = 0;
3694        }
3695
3696        if (pi->cmd != P_PROTOCOL_UPDATE) {
3697                clear_bit(CONN_DRY_RUN, &connection->flags);
3698
3699                if (cf & CF_DRY_RUN)
3700                        set_bit(CONN_DRY_RUN, &connection->flags);
3701
3702                rcu_read_lock();
3703                nc = rcu_dereference(connection->net_conf);
3704
3705                if (p_proto != nc->wire_protocol) {
3706                        drbd_err(connection, "incompatible %s settings\n", "protocol");
3707                        goto disconnect_rcu_unlock;
3708                }
3709
3710                if (convert_after_sb(p_after_sb_0p) != nc->after_sb_0p) {
3711                        drbd_err(connection, "incompatible %s settings\n", "after-sb-0pri");
3712                        goto disconnect_rcu_unlock;
3713                }
3714
3715                if (convert_after_sb(p_after_sb_1p) != nc->after_sb_1p) {
3716                        drbd_err(connection, "incompatible %s settings\n", "after-sb-1pri");
3717                        goto disconnect_rcu_unlock;
3718                }
3719
3720                if (convert_after_sb(p_after_sb_2p) != nc->after_sb_2p) {
3721                        drbd_err(connection, "incompatible %s settings\n", "after-sb-2pri");
3722                        goto disconnect_rcu_unlock;
3723                }
3724
3725                if (p_discard_my_data && nc->discard_my_data) {
3726                        drbd_err(connection, "incompatible %s settings\n", "discard-my-data");
3727                        goto disconnect_rcu_unlock;
3728                }
3729
3730                if (p_two_primaries != nc->two_primaries) {
3731                        drbd_err(connection, "incompatible %s settings\n", "allow-two-primaries");
3732                        goto disconnect_rcu_unlock;
3733                }
3734
3735                if (strcmp(integrity_alg, nc->integrity_alg)) {
3736                        drbd_err(connection, "incompatible %s settings\n", "data-integrity-alg");
3737                        goto disconnect_rcu_unlock;
3738                }
3739
3740                rcu_read_unlock();
3741        }
3742
3743        if (integrity_alg[0]) {
3744                int hash_size;
3745
3746                /*
3747                 * We can only change the peer data integrity algorithm
3748                 * here.  Changing our own data integrity algorithm
3749                 * requires that we send a P_PROTOCOL_UPDATE packet at
3750                 * the same time; otherwise, the peer has no way to
3751                 * tell between which packets the algorithm should
3752                 * change.
3753                 */
3754
3755                peer_integrity_tfm = crypto_alloc_shash(integrity_alg, 0, 0);
3756                if (IS_ERR(peer_integrity_tfm)) {
3757                        peer_integrity_tfm = NULL;
3758                        drbd_err(connection, "peer data-integrity-alg %s not supported\n",
3759                                 integrity_alg);
3760                        goto disconnect;
3761                }
3762
3763                hash_size = crypto_shash_digestsize(peer_integrity_tfm);
3764                int_dig_in = kmalloc(hash_size, GFP_KERNEL);
3765                int_dig_vv = kmalloc(hash_size, GFP_KERNEL);
3766                if (!(int_dig_in && int_dig_vv)) {
3767                        drbd_err(connection, "Allocation of buffers for data integrity checking failed\n");
3768                        goto disconnect;
3769                }
3770        }
3771
3772        new_net_conf = kmalloc(sizeof(struct net_conf), GFP_KERNEL);
3773        if (!new_net_conf)
3774                goto disconnect;
3775
3776        mutex_lock(&connection->data.mutex);
3777        mutex_lock(&connection->resource->conf_update);
3778        old_net_conf = connection->net_conf;
3779        *new_net_conf = *old_net_conf;
3780
3781        new_net_conf->wire_protocol = p_proto;
3782        new_net_conf->after_sb_0p = convert_after_sb(p_after_sb_0p);
3783        new_net_conf->after_sb_1p = convert_after_sb(p_after_sb_1p);
3784        new_net_conf->after_sb_2p = convert_after_sb(p_after_sb_2p);
3785        new_net_conf->two_primaries = p_two_primaries;
3786
3787        rcu_assign_pointer(connection->net_conf, new_net_conf);
3788        mutex_unlock(&connection->resource->conf_update);
3789        mutex_unlock(&connection->data.mutex);
3790
3791        crypto_free_shash(connection->peer_integrity_tfm);
3792        kfree(connection->int_dig_in);
3793        kfree(connection->int_dig_vv);
3794        connection->peer_integrity_tfm = peer_integrity_tfm;
3795        connection->int_dig_in = int_dig_in;
3796        connection->int_dig_vv = int_dig_vv;
3797
3798        if (strcmp(old_net_conf->integrity_alg, integrity_alg))
3799                drbd_info(connection, "peer data-integrity-alg: %s\n",
3800                          integrity_alg[0] ? integrity_alg : "(none)");
3801
3802        synchronize_rcu();
3803        kfree(old_net_conf);
3804        return 0;
3805
3806disconnect_rcu_unlock:
3807        rcu_read_unlock();
3808disconnect:
3809        crypto_free_shash(peer_integrity_tfm);
3810        kfree(int_dig_in);
3811        kfree(int_dig_vv);
3812        conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
3813        return -EIO;
3814}
3815
3816/* helper function
3817 * input: alg name, feature name
3818 * return: NULL (alg name was "")
3819 *         ERR_PTR(error) if something goes wrong
3820 *         or the crypto hash ptr, if it worked out ok. */
3821static struct crypto_shash *drbd_crypto_alloc_digest_safe(
3822                const struct drbd_device *device,
3823                const char *alg, const char *name)
3824{
3825        struct crypto_shash *tfm;
3826
3827        if (!alg[0])
3828                return NULL;
3829
3830        tfm = crypto_alloc_shash(alg, 0, 0);
3831        if (IS_ERR(tfm)) {
3832                drbd_err(device, "Can not allocate \"%s\" as %s (reason: %ld)\n",
3833                        alg, name, PTR_ERR(tfm));
3834                return tfm;
3835        }
3836        return tfm;
3837}
3838
3839static int ignore_remaining_packet(struct drbd_connection *connection, struct packet_info *pi)
3840{
3841        void *buffer = connection->data.rbuf;
3842        int size = pi->size;
3843
3844        while (size) {
3845                int s = min_t(int, size, DRBD_SOCKET_BUFFER_SIZE);
3846                s = drbd_recv(connection, buffer, s);
3847                if (s <= 0) {
3848                        if (s < 0)
3849                                return s;
3850                        break;
3851                }
3852                size -= s;
3853        }
3854        if (size)
3855                return -EIO;
3856        return 0;
3857}
3858
3859/*
3860 * config_unknown_volume  -  device configuration command for unknown volume
3861 *
3862 * When a device is added to an existing connection, the node on which the
3863 * device is added first will send configuration commands to its peer but the
3864 * peer will not know about the device yet.  It will warn and ignore these
3865 * commands.  Once the device is added on the second node, the second node will
3866 * send the same device configuration commands, but in the other direction.
3867 *
3868 * (We can also end up here if drbd is misconfigured.)
3869 */
3870static int config_unknown_volume(struct drbd_connection *connection, struct packet_info *pi)
3871{
3872        drbd_warn(connection, "%s packet received for volume %u, which is not configured locally\n",
3873                  cmdname(pi->cmd), pi->vnr);
3874        return ignore_remaining_packet(connection, pi);
3875}
3876
3877static int receive_SyncParam(struct drbd_connection *connection, struct packet_info *pi)
3878{
3879        struct drbd_peer_device *peer_device;
3880        struct drbd_device *device;
3881        struct p_rs_param_95 *p;
3882        unsigned int header_size, data_size, exp_max_sz;
3883        struct crypto_shash *verify_tfm = NULL;
3884        struct crypto_shash *csums_tfm = NULL;
3885        struct net_conf *old_net_conf, *new_net_conf = NULL;
3886        struct disk_conf *old_disk_conf = NULL, *new_disk_conf = NULL;
3887        const int apv = connection->agreed_pro_version;
3888        struct fifo_buffer *old_plan = NULL, *new_plan = NULL;
3889        unsigned int fifo_size = 0;
3890        int err;
3891
3892        peer_device = conn_peer_device(connection, pi->vnr);
3893        if (!peer_device)
3894                return config_unknown_volume(connection, pi);
3895        device = peer_device->device;
3896
3897        exp_max_sz  = apv <= 87 ? sizeof(struct p_rs_param)
3898                    : apv == 88 ? sizeof(struct p_rs_param)
3899                                        + SHARED_SECRET_MAX
3900                    : apv <= 94 ? sizeof(struct p_rs_param_89)
3901                    : /* apv >= 95 */ sizeof(struct p_rs_param_95);
3902
3903        if (pi->size > exp_max_sz) {
3904                drbd_err(device, "SyncParam packet too long: received %u, expected <= %u bytes\n",
3905                    pi->size, exp_max_sz);
3906                return -EIO;
3907        }
3908
3909        if (apv <= 88) {
3910                header_size = sizeof(struct p_rs_param);
3911                data_size = pi->size - header_size;
3912        } else if (apv <= 94) {
3913                header_size = sizeof(struct p_rs_param_89);
3914                data_size = pi->size - header_size;
3915                D_ASSERT(device, data_size == 0);
3916        } else {
3917                header_size = sizeof(struct p_rs_param_95);
3918                data_size = pi->size - header_size;
3919                D_ASSERT(device, data_size == 0);
3920        }
3921
3922        /* initialize verify_alg and csums_alg */
3923        p = pi->data;
3924        memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
3925
3926        err = drbd_recv_all(peer_device->connection, p, header_size);
3927        if (err)
3928                return err;
3929
3930        mutex_lock(&connection->resource->conf_update);
3931        old_net_conf = peer_device->connection->net_conf;
3932        if (get_ldev(device)) {
3933                new_disk_conf = kzalloc(sizeof(struct disk_conf), GFP_KERNEL);
3934                if (!new_disk_conf) {
3935                        put_ldev(device);
3936                        mutex_unlock(&connection->resource->conf_update);
3937                        drbd_err(device, "Allocation of new disk_conf failed\n");
3938                        return -ENOMEM;
3939                }
3940
3941                old_disk_conf = device->ldev->disk_conf;
3942                *new_disk_conf = *old_disk_conf;
3943
3944                new_disk_conf->resync_rate = be32_to_cpu(p->resync_rate);
3945        }
3946
3947        if (apv >= 88) {
3948                if (apv == 88) {
3949                        if (data_size > SHARED_SECRET_MAX || data_size == 0) {
3950                                drbd_err(device, "verify-alg of wrong size, "
3951                                        "peer wants %u, accepting only up to %u byte\n",
3952                                        data_size, SHARED_SECRET_MAX);
3953                                err = -EIO;
3954                                goto reconnect;
3955                        }
3956
3957                        err = drbd_recv_all(peer_device->connection, p->verify_alg, data_size);
3958                        if (err)
3959                                goto reconnect;
3960                        /* we expect NUL terminated string */
3961                        /* but just in case someone tries to be evil */
3962                        D_ASSERT(device, p->verify_alg[data_size-1] == 0);
3963                        p->verify_alg[data_size-1] = 0;
3964
3965                } else /* apv >= 89 */ {
3966                        /* we still expect NUL terminated strings */
3967                        /* but just in case someone tries to be evil */
3968                        D_ASSERT(device, p->verify_alg[SHARED_SECRET_MAX-1] == 0);
3969                        D_ASSERT(device, p->csums_alg[SHARED_SECRET_MAX-1] == 0);
3970                        p->verify_alg[SHARED_SECRET_MAX-1] = 0;
3971                        p->csums_alg[SHARED_SECRET_MAX-1] = 0;
3972                }
3973
3974                if (strcmp(old_net_conf->verify_alg, p->verify_alg)) {
3975                        if (device->state.conn == C_WF_REPORT_PARAMS) {
3976                                drbd_err(device, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n",
3977                                    old_net_conf->verify_alg, p->verify_alg);
3978                                goto disconnect;
3979                        }
3980                        verify_tfm = drbd_crypto_alloc_digest_safe(device,
3981                                        p->verify_alg, "verify-alg");
3982                        if (IS_ERR(verify_tfm)) {
3983                                verify_tfm = NULL;
3984                                goto disconnect;
3985                        }
3986                }
3987
3988                if (apv >= 89 && strcmp(old_net_conf->csums_alg, p->csums_alg)) {
3989                        if (device->state.conn == C_WF_REPORT_PARAMS) {
3990                                drbd_err(device, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n",
3991                                    old_net_conf->csums_alg, p->csums_alg);
3992                                goto disconnect;
3993                        }
3994                        csums_tfm = drbd_crypto_alloc_digest_safe(device,
3995                                        p->csums_alg, "csums-alg");
3996                        if (IS_ERR(csums_tfm)) {
3997                                csums_tfm = NULL;
3998                                goto disconnect;
3999                        }
4000                }
4001
4002                if (apv > 94 && new_disk_conf) {
4003                        new_disk_conf->c_plan_ahead = be32_to_cpu(p->c_plan_ahead);
4004                        new_disk_conf->c_delay_target = be32_to_cpu(p->c_delay_target);
4005                        new_disk_conf->c_fill_target = be32_to_cpu(p->c_fill_target);
4006                        new_disk_conf->c_max_rate = be32_to_cpu(p->c_max_rate);
4007
4008                        fifo_size = (new_disk_conf->c_plan_ahead * 10 * SLEEP_TIME) / HZ;
4009                        if (fifo_size != device->rs_plan_s->size) {
4010                                new_plan = fifo_alloc(fifo_size);
4011                                if (!new_plan) {
4012                                        drbd_err(device, "kmalloc of fifo_buffer failed");
4013                                        put_ldev(device);
4014                                        goto disconnect;
4015                                }
4016                        }
4017                }
4018
4019                if (verify_tfm || csums_tfm) {
4020                        new_net_conf = kzalloc(sizeof(struct net_conf), GFP_KERNEL);
4021                        if (!new_net_conf)
4022                                goto disconnect;
4023
4024                        *new_net_conf = *old_net_conf;
4025
4026                        if (verify_tfm) {
4027                                strcpy(new_net_conf->verify_alg, p->verify_alg);
4028                                new_net_conf->verify_alg_len = strlen(p->verify_alg) + 1;
4029                                crypto_free_shash(peer_device->connection->verify_tfm);
4030                                peer_device->connection->verify_tfm = verify_tfm;
4031