// SPDX-License-Identifier: GPL-2.0-or-later
/*
   drbd_receiver.c

   This file is part of DRBD by Philipp Reisner and Lars Ellenberg.

   Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
   Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
   Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.

 */


#include <linux/module.h>

#include <linux/uaccess.h>
#include <net/sock.h>

#include <linux/drbd.h>
#include <linux/fs.h>
#include <linux/file.h>
#include <linux/in.h>
#include <linux/mm.h>
#include <linux/memcontrol.h>
#include <linux/mm_inline.h>
#include <linux/slab.h>
#include <uapi/linux/sched/types.h>
#include <linux/sched/signal.h>
#include <linux/pkt_sched.h>
#define __KERNEL_SYSCALLS__
#include <linux/unistd.h>
#include <linux/vmalloc.h>
#include <linux/random.h>
#include <linux/string.h>
#include <linux/scatterlist.h>
#include <linux/part_stat.h>
#include "drbd_int.h"
#include "drbd_protocol.h"
#include "drbd_req.h"
#include "drbd_vli.h"

#define PRO_FEATURES (DRBD_FF_TRIM|DRBD_FF_THIN_RESYNC|DRBD_FF_WSAME|DRBD_FF_WZEROES)

struct packet_info {
        enum drbd_packet cmd;
        unsigned int size;
        unsigned int vnr;
        void *data;
};

enum finish_epoch {
        FE_STILL_LIVE,
        FE_DESTROYED,
        FE_RECYCLED,
};

static int drbd_do_features(struct drbd_connection *connection);
static int drbd_do_auth(struct drbd_connection *connection);
static int drbd_disconnected(struct drbd_peer_device *);
static void conn_wait_active_ee_empty(struct drbd_connection *connection);
static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *, struct drbd_epoch *, enum epoch_event);
static int e_end_block(struct drbd_work *, int);


#define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)

/*
 * some helper functions to deal with single linked page lists,
 * page->private being our "next" pointer.
 */

/* If at least n pages are linked at head, get n pages off.
 * Otherwise, don't modify head, and return NULL.
 * Locking is the responsibility of the caller.
 */
static struct page *page_chain_del(struct page **head, int n)
{
        struct page *page;
        struct page *tmp;

        BUG_ON(!n);
        BUG_ON(!head);

        page = *head;

        if (!page)
                return NULL;

        while (page) {
                tmp = page_chain_next(page);
                if (--n == 0)
                        break; /* found sufficient pages */
                if (tmp == NULL)
                        /* insufficient pages, don't use any of them. */
                        return NULL;
                page = tmp;
        }

        /* add end of list marker for the returned list */
        set_page_private(page, 0);
        /* actual return value, and adjustment of head */
        page = *head;
        *head = tmp;
        return page;
}

/* may be used outside of locks to find the tail of a (usually short)
 * "private" page chain, before adding it back to a global chain head
 * with page_chain_add() under a spinlock. */
static struct page *page_chain_tail(struct page *page, int *len)
{
        struct page *tmp;
        int i = 1;
        while ((tmp = page_chain_next(page)))
                ++i, page = tmp;
        if (len)
                *len = i;
        return page;
}

static int page_chain_free(struct page *page)
{
        struct page *tmp;
        int i = 0;
        page_chain_for_each_safe(page, tmp) {
                put_page(page);
                ++i;
        }
        return i;
}

static void page_chain_add(struct page **head,
                struct page *chain_first, struct page *chain_last)
{
#if 1
        struct page *tmp;
        tmp = page_chain_tail(chain_first, NULL);
        BUG_ON(tmp != chain_last);
#endif

        /* add chain to head */
        set_page_private(chain_last, (unsigned long)*head);
        *head = chain_first;
}

static struct page *__drbd_alloc_pages(struct drbd_device *device,
                                       unsigned int number)
{
        struct page *page = NULL;
        struct page *tmp = NULL;
        unsigned int i = 0;

        /* Yes, testing drbd_pp_vacant outside the lock is racy.
         * So what. It saves a spin_lock. */
        if (drbd_pp_vacant >= number) {
                spin_lock(&drbd_pp_lock);
                page = page_chain_del(&drbd_pp_pool, number);
                if (page)
                        drbd_pp_vacant -= number;
                spin_unlock(&drbd_pp_lock);
                if (page)
                        return page;
        }

        /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD
         * "criss-cross" setup, that might cause write-out on some other DRBD,
         * which in turn might block on the other node at this very place.  */
        for (i = 0; i < number; i++) {
                tmp = alloc_page(GFP_TRY);
                if (!tmp)
                        break;
                set_page_private(tmp, (unsigned long)page);
                page = tmp;
        }

        if (i == number)
                return page;

        /* Not enough pages immediately available this time.
         * No need to jump around here, drbd_alloc_pages will retry this
         * function "soon". */
        if (page) {
                tmp = page_chain_tail(page, NULL);
                spin_lock(&drbd_pp_lock);
                page_chain_add(&drbd_pp_pool, page, tmp);
                drbd_pp_vacant += i;
                spin_unlock(&drbd_pp_lock);
        }
        return NULL;
}

static void reclaim_finished_net_peer_reqs(struct drbd_device *device,
                                           struct list_head *to_be_freed)
{
        struct drbd_peer_request *peer_req, *tmp;

        /* The EEs are always appended to the end of the list. Since
           they are sent in order over the wire, they have to finish
           in order. As soon as we see the first not finished we can
           stop to examine the list... */

        list_for_each_entry_safe(peer_req, tmp, &device->net_ee, w.list) {
                if (drbd_peer_req_has_active_page(peer_req))
                        break;
                list_move(&peer_req->w.list, to_be_freed);
        }
}

static void drbd_reclaim_net_peer_reqs(struct drbd_device *device)
{
        LIST_HEAD(reclaimed);
        struct drbd_peer_request *peer_req, *t;

        spin_lock_irq(&device->resource->req_lock);
        reclaim_finished_net_peer_reqs(device, &reclaimed);
        spin_unlock_irq(&device->resource->req_lock);
        list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
                drbd_free_net_peer_req(device, peer_req);
}

static void conn_reclaim_net_peer_reqs(struct drbd_connection *connection)
{
        struct drbd_peer_device *peer_device;
        int vnr;

        rcu_read_lock();
        idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
                struct drbd_device *device = peer_device->device;
                if (!atomic_read(&device->pp_in_use_by_net))
                        continue;

                kref_get(&device->kref);
                rcu_read_unlock();
                drbd_reclaim_net_peer_reqs(device);
                kref_put(&device->kref, drbd_destroy_device);
                rcu_read_lock();
        }
        rcu_read_unlock();
}

/**
 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
 * @device:     DRBD device.
 * @number:     number of pages requested
 * @retry:      whether to retry, if not enough pages are available right now
 *
 * Tries to allocate number pages, first from our own page pool, then from
 * the kernel.
 * Possibly retry until DRBD frees sufficient pages somewhere else.
 *
 * If this allocation would exceed the max_buffers setting, we throttle
 * allocation (schedule_timeout) to give the system some room to breathe.
 *
 * We do not use max-buffers as hard limit, because it could lead to
 * congestion and further to a distributed deadlock during online-verify or
 * (checksum based) resync, if the max-buffers, socket buffer sizes and
 * resync-rate settings are mis-configured.
 *
 * Returns a page chain linked via page->private.
 */
struct page *drbd_alloc_pages(struct drbd_peer_device *peer_device, unsigned int number,
                              bool retry)
{
        struct drbd_device *device = peer_device->device;
        struct page *page = NULL;
        struct net_conf *nc;
        DEFINE_WAIT(wait);
        unsigned int mxb;

        rcu_read_lock();
        nc = rcu_dereference(peer_device->connection->net_conf);
        mxb = nc ? nc->max_buffers : 1000000;
        rcu_read_unlock();

        if (atomic_read(&device->pp_in_use) < mxb)
                page = __drbd_alloc_pages(device, number);

        /* Try to keep the fast path fast, but occasionally we need
         * to reclaim the pages we lended to the network stack. */
        if (page && atomic_read(&device->pp_in_use_by_net) > 512)
                drbd_reclaim_net_peer_reqs(device);

        while (page == NULL) {
                prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE);

                drbd_reclaim_net_peer_reqs(device);

                if (atomic_read(&device->pp_in_use) < mxb) {
                        page = __drbd_alloc_pages(device, number);
                        if (page)
                                break;
                }

                if (!retry)
                        break;

                if (signal_pending(current)) {
                        drbd_warn(device, "drbd_alloc_pages interrupted!\n");
                        break;
                }

                if (schedule_timeout(HZ/10) == 0)
                        mxb = UINT_MAX;
        }
        finish_wait(&drbd_pp_wait, &wait);

        if (page)
                atomic_add(number, &device->pp_in_use);
        return page;
}

/* Must not be used from irq, as that may deadlock: see drbd_alloc_pages.
 * Is also used from inside an other spin_lock_irq(&resource->req_lock);
 * Either links the page chain back to the global pool,
 * or returns all pages to the system. */
static void drbd_free_pages(struct drbd_device *device, struct page *page, int is_net)
{
        atomic_t *a = is_net ? &device->pp_in_use_by_net : &device->pp_in_use;
        int i;

        if (page == NULL)
                return;

        if (drbd_pp_vacant > (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * drbd_minor_count)
                i = page_chain_free(page);
        else {
                struct page *tmp;
                tmp = page_chain_tail(page, &i);
                spin_lock(&drbd_pp_lock);
                page_chain_add(&drbd_pp_pool, page, tmp);
                drbd_pp_vacant += i;
                spin_unlock(&drbd_pp_lock);
        }
        i = atomic_sub_return(i, a);
        if (i < 0)
                drbd_warn(device, "ASSERTION FAILED: %s: %d < 0\n",
                        is_net ? "pp_in_use_by_net" : "pp_in_use", i);
        wake_up(&drbd_pp_wait);
}

/*
You need to hold the req_lock:
 _drbd_wait_ee_list_empty()

You must not have the req_lock:
 drbd_free_peer_req()
 drbd_alloc_peer_req()
 drbd_free_peer_reqs()
 drbd_ee_fix_bhs()
 drbd_finish_peer_reqs()
 drbd_clear_done_ee()
 drbd_wait_ee_list_empty()
*/

/* normal: payload_size == request size (bi_size)
 * w_same: payload_size == logical_block_size
 * trim: payload_size == 0 */
struct drbd_peer_request *
drbd_alloc_peer_req(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
                    unsigned int request_size, unsigned int payload_size, gfp_t gfp_mask) __must_hold(local)
{
        struct drbd_device *device = peer_device->device;
        struct drbd_peer_request *peer_req;
        struct page *page = NULL;
        unsigned nr_pages = (payload_size + PAGE_SIZE -1) >> PAGE_SHIFT;

        if (drbd_insert_fault(device, DRBD_FAULT_AL_EE))
                return NULL;

        peer_req = mempool_alloc(&drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM);
        if (!peer_req) {
                if (!(gfp_mask & __GFP_NOWARN))
                        drbd_err(device, "%s: allocation failed\n", __func__);
                return NULL;
        }

        if (nr_pages) {
                page = drbd_alloc_pages(peer_device, nr_pages,
                                        gfpflags_allow_blocking(gfp_mask));
                if (!page)
                        goto fail;
        }

        memset(peer_req, 0, sizeof(*peer_req));
        INIT_LIST_HEAD(&peer_req->w.list);
        drbd_clear_interval(&peer_req->i);
        peer_req->i.size = request_size;
        peer_req->i.sector = sector;
        peer_req->submit_jif = jiffies;
        peer_req->peer_device = peer_device;
        peer_req->pages = page;
        /*
         * The block_id is opaque to the receiver.  It is not endianness
         * converted, and sent back to the sender unchanged.
         */
        peer_req->block_id = id;

        return peer_req;

 fail:
        mempool_free(peer_req, &drbd_ee_mempool);
        return NULL;
}

void __drbd_free_peer_req(struct drbd_device *device, struct drbd_peer_request *peer_req,
                       int is_net)
{
        might_sleep();
        if (peer_req->flags & EE_HAS_DIGEST)
                kfree(peer_req->digest);
        drbd_free_pages(device, peer_req->pages, is_net);
        D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
        D_ASSERT(device, drbd_interval_empty(&peer_req->i));
        if (!expect(!(peer_req->flags & EE_CALL_AL_COMPLETE_IO))) {
                peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
                drbd_al_complete_io(device, &peer_req->i);
        }
        mempool_free(peer_req, &drbd_ee_mempool);
}

int drbd_free_peer_reqs(struct drbd_device *device, struct list_head *list)
{
        LIST_HEAD(work_list);
        struct drbd_peer_request *peer_req, *t;
        int count = 0;
        int is_net = list == &device->net_ee;

        spin_lock_irq(&device->resource->req_lock);
        list_splice_init(list, &work_list);
        spin_unlock_irq(&device->resource->req_lock);

        list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
                __drbd_free_peer_req(device, peer_req, is_net);
                count++;
        }
        return count;
}

/*
 * See also comments in _req_mod(,BARRIER_ACKED) and receive_Barrier.
 */
static int drbd_finish_peer_reqs(struct drbd_device *device)
{
        LIST_HEAD(work_list);
        LIST_HEAD(reclaimed);
        struct drbd_peer_request *peer_req, *t;
        int err = 0;

        spin_lock_irq(&device->resource->req_lock);
        reclaim_finished_net_peer_reqs(device, &reclaimed);
        list_splice_init(&device->done_ee, &work_list);
        spin_unlock_irq(&device->resource->req_lock);

        list_for_each_entry_safe(peer_req, t, &reclaimed, w.list)
                drbd_free_net_peer_req(device, peer_req);

        /* possible callbacks here:
         * e_end_block, and e_end_resync_block, e_send_superseded.
         * all ignore the last argument.
         */
        list_for_each_entry_safe(peer_req, t, &work_list, w.list) {
                int err2;

                /* list_del not necessary, next/prev members not touched */
                err2 = peer_req->w.cb(&peer_req->w, !!err);
                if (!err)
                        err = err2;
                drbd_free_peer_req(device, peer_req);
        }
        wake_up(&device->ee_wait);

        return err;
}

static void _drbd_wait_ee_list_empty(struct drbd_device *device,
                                     struct list_head *head)
{
        DEFINE_WAIT(wait);

        /* avoids spin_lock/unlock
         * and calling prepare_to_wait in the fast path */
        while (!list_empty(head)) {
                prepare_to_wait(&device->ee_wait, &wait, TASK_UNINTERRUPTIBLE);
                spin_unlock_irq(&device->resource->req_lock);
                io_schedule();
                finish_wait(&device->ee_wait, &wait);
                spin_lock_irq(&device->resource->req_lock);
        }
}

static void drbd_wait_ee_list_empty(struct drbd_device *device,
                                    struct list_head *head)
{
        spin_lock_irq(&device->resource->req_lock);
        _drbd_wait_ee_list_empty(device, head);
        spin_unlock_irq(&device->resource->req_lock);
}

static int drbd_recv_short(struct socket *sock, void *buf, size_t size, int flags)
{
        struct kvec iov = {
                .iov_base = buf,
                .iov_len = size,
        };
        struct msghdr msg = {
                .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL)
        };
        iov_iter_kvec(&msg.msg_iter, READ, &iov, 1, size);
        return sock_recvmsg(sock, &msg, msg.msg_flags);
}

static int drbd_recv(struct drbd_connection *connection, void *buf, size_t size)
{
        int rv;

        rv = drbd_recv_short(connection->data.socket, buf, size, 0);

        if (rv < 0) {
                if (rv == -ECONNRESET)
                        drbd_info(connection, "sock was reset by peer\n");
                else if (rv != -ERESTARTSYS)
                        drbd_err(connection, "sock_recvmsg returned %d\n", rv);
        } else if (rv == 0) {
                if (test_bit(DISCONNECT_SENT, &connection->flags)) {
                        long t;
                        rcu_read_lock();
                        t = rcu_dereference(connection->net_conf)->ping_timeo * HZ/10;
                        rcu_read_unlock();

                        t = wait_event_timeout(connection->ping_wait, connection->cstate < C_WF_REPORT_PARAMS, t);

                        if (t)
                                goto out;
                }
                drbd_info(connection, "sock was shut down by peer\n");
        }

        if (rv != size)
                conn_request_state(connection, NS(conn, C_BROKEN_PIPE), CS_HARD);

out:
        return rv;
}

static int drbd_recv_all(struct drbd_connection *connection, void *buf, size_t size)
{
        int err;

        err = drbd_recv(connection, buf, size);
        if (err != size) {
                if (err >= 0)
                        err = -EIO;
        } else
                err = 0;
        return err;
}

static int drbd_recv_all_warn(struct drbd_connection *connection, void *buf, size_t size)
{
        int err;

        err = drbd_recv_all(connection, buf, size);
        if (err && !signal_pending(current))
                drbd_warn(connection, "short read (expected size %d)\n", (int)size);
        return err;
}

/* quoting tcp(7):
 *   On individual connections, the socket buffer size must be set prior to the
 *   listen(2) or connect(2) calls in order to have it take effect.
 * This is our wrapper to do so.
 */
static void drbd_setbufsize(struct socket *sock, unsigned int snd,
                unsigned int rcv)
{
        /* open coded SO_SNDBUF, SO_RCVBUF */
        if (snd) {
                sock->sk->sk_sndbuf = snd;
                sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
        }
        if (rcv) {
                sock->sk->sk_rcvbuf = rcv;
                sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
        }
}

static struct socket *drbd_try_connect(struct drbd_connection *connection)
{
        const char *what;
        struct socket *sock;
        struct sockaddr_in6 src_in6;
        struct sockaddr_in6 peer_in6;
        struct net_conf *nc;
        int err, peer_addr_len, my_addr_len;
        int sndbuf_size, rcvbuf_size, connect_int;
        int disconnect_on_error = 1;

        rcu_read_lock();
        nc = rcu_dereference(connection->net_conf);
        if (!nc) {
                rcu_read_unlock();
                return NULL;
        }
        sndbuf_size = nc->sndbuf_size;
        rcvbuf_size = nc->rcvbuf_size;
        connect_int = nc->connect_int;
        rcu_read_unlock();

        my_addr_len = min_t(int, connection->my_addr_len, sizeof(src_in6));
        memcpy(&src_in6, &connection->my_addr, my_addr_len);

        if (((struct sockaddr *)&connection->my_addr)->sa_family == AF_INET6)
                src_in6.sin6_port = 0;
        else
                ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */

        peer_addr_len = min_t(int, connection->peer_addr_len, sizeof(src_in6));
        memcpy(&peer_in6, &connection->peer_addr, peer_addr_len);

        what = "sock_create_kern";
        err = sock_create_kern(&init_net, ((struct sockaddr *)&src_in6)->sa_family,
                               SOCK_STREAM, IPPROTO_TCP, &sock);
        if (err < 0) {
                sock = NULL;
                goto out;
        }

        sock->sk->sk_rcvtimeo =
        sock->sk->sk_sndtimeo = connect_int * HZ;
        drbd_setbufsize(sock, sndbuf_size, rcvbuf_size);

       /* explicitly bind to the configured IP as source IP
        *  for the outgoing connections.
        *  This is needed for multihomed hosts and to be
        *  able to use lo: interfaces for drbd.
        * Make sure to use 0 as port number, so linux selects
        *  a free one dynamically.
        */
        what = "bind before connect";
        err = sock->ops->bind(sock, (struct sockaddr *) &src_in6, my_addr_len);
        if (err < 0)
                goto out;

        /* connect may fail, peer not yet available.
         * stay C_WF_CONNECTION, don't go Disconnecting! */
        disconnect_on_error = 0;
        what = "connect";
        err = sock->ops->connect(sock, (struct sockaddr *) &peer_in6, peer_addr_len, 0);

out:
        if (err < 0) {
                if (sock) {
                        sock_release(sock);
                        sock = NULL;
                }
                switch (-err) {
                        /* timeout, busy, signal pending */
                case ETIMEDOUT: case EAGAIN: case EINPROGRESS:
                case EINTR: case ERESTARTSYS:
                        /* peer not (yet) available, network problem */
                case ECONNREFUSED: case ENETUNREACH:
                case EHOSTDOWN:    case EHOSTUNREACH:
                        disconnect_on_error = 0;
                        break;
                default:
                        drbd_err(connection, "%s failed, err = %d\n", what, err);
                }
                if (disconnect_on_error)
                        conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
        }

        return sock;
}

struct accept_wait_data {
        struct drbd_connection *connection;
        struct socket *s_listen;
        struct completion door_bell;
        void (*original_sk_state_change)(struct sock *sk);

};

static void drbd_incoming_connection(struct sock *sk)
{
        struct accept_wait_data *ad = sk->sk_user_data;
        void (*state_change)(struct sock *sk);

        state_change = ad->original_sk_state_change;
        if (sk->sk_state == TCP_ESTABLISHED)
                complete(&ad->door_bell);
        state_change(sk);
}

static int prepare_listen_socket(struct drbd_connection *connection, struct accept_wait_data *ad)
{
        int err, sndbuf_size, rcvbuf_size, my_addr_len;
        struct sockaddr_in6 my_addr;
        struct socket *s_listen;
        struct net_conf *nc;
        const char *what;

        rcu_read_lock();
        nc = rcu_dereference(connection->net_conf);
        if (!nc) {
                rcu_read_unlock();
                return -EIO;
        }
        sndbuf_size = nc->sndbuf_size;
        rcvbuf_size = nc->rcvbuf_size;
        rcu_read_unlock();

        my_addr_len = min_t(int, connection->my_addr_len, sizeof(struct sockaddr_in6));
        memcpy(&my_addr, &connection->my_addr, my_addr_len);

        what = "sock_create_kern";
        err = sock_create_kern(&init_net, ((struct sockaddr *)&my_addr)->sa_family,
                               SOCK_STREAM, IPPROTO_TCP, &s_listen);
        if (err) {
                s_listen = NULL;
                goto out;
        }

        s_listen->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
        drbd_setbufsize(s_listen, sndbuf_size, rcvbuf_size);

        what = "bind before listen";
        err = s_listen->ops->bind(s_listen, (struct sockaddr *)&my_addr, my_addr_len);
        if (err < 0)
                goto out;

        ad->s_listen = s_listen;
        write_lock_bh(&s_listen->sk->sk_callback_lock);
        ad->original_sk_state_change = s_listen->sk->sk_state_change;
        s_listen->sk->sk_state_change = drbd_incoming_connection;
        s_listen->sk->sk_user_data = ad;
        write_unlock_bh(&s_listen->sk->sk_callback_lock);

        what = "listen";
        err = s_listen->ops->listen(s_listen, 5);
        if (err < 0)
                goto out;

        return 0;
out:
        if (s_listen)
                sock_release(s_listen);
        if (err < 0) {
                if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
                        drbd_err(connection, "%s failed, err = %d\n", what, err);
                        conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
                }
        }

        return -EIO;
}

static void unregister_state_change(struct sock *sk, struct accept_wait_data *ad)
{
        write_lock_bh(&sk->sk_callback_lock);
        sk->sk_state_change = ad->original_sk_state_change;
        sk->sk_user_data = NULL;
        write_unlock_bh(&sk->sk_callback_lock);
}

static struct socket *drbd_wait_for_connect(struct drbd_connection *connection, struct accept_wait_data *ad)
{
        int timeo, connect_int, err = 0;
        struct socket *s_estab = NULL;
        struct net_conf *nc;

        rcu_read_lock();
        nc = rcu_dereference(connection->net_conf);
        if (!nc) {
                rcu_read_unlock();
                return NULL;
        }
        connect_int = nc->connect_int;
        rcu_read_unlock();

        timeo = connect_int * HZ;
        /* 28.5% random jitter */
        timeo += (prandom_u32() & 1) ? timeo / 7 : -timeo / 7;

        err = wait_for_completion_interruptible_timeout(&ad->door_bell, timeo);
        if (err <= 0)
                return NULL;

        err = kernel_accept(ad->s_listen, &s_estab, 0);
        if (err < 0) {
                if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) {
                        drbd_err(connection, "accept failed, err = %d\n", err);
                        conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
                }
        }

        if (s_estab)
                unregister_state_change(s_estab->sk, ad);

        return s_estab;
}

static int decode_header(struct drbd_connection *, void *, struct packet_info *);

static int send_first_packet(struct drbd_connection *connection, struct drbd_socket *sock,
                             enum drbd_packet cmd)
{
        if (!conn_prepare_command(connection, sock))
                return -EIO;
        return conn_send_command(connection, sock, cmd, 0, NULL, 0);
}

static int receive_first_packet(struct drbd_connection *connection, struct socket *sock)
{
        unsigned int header_size = drbd_header_size(connection);
        struct packet_info pi;
        struct net_conf *nc;
        int err;

        rcu_read_lock();
        nc = rcu_dereference(connection->net_conf);
        if (!nc) {
                rcu_read_unlock();
                return -EIO;
        }
        sock->sk->sk_rcvtimeo = nc->ping_timeo * 4 * HZ / 10;
        rcu_read_unlock();

        err = drbd_recv_short(sock, connection->data.rbuf, header_size, 0);
        if (err != header_size) {
                if (err >= 0)
                        err = -EIO;
                return err;
        }
        err = decode_header(connection, connection->data.rbuf, &pi);
        if (err)
                return err;
        return pi.cmd;
}

/**
 * drbd_socket_okay() - Free the socket if its connection is not okay
 * @sock:       pointer to the pointer to the socket.
 */
static bool drbd_socket_okay(struct socket **sock)
{
        int rr;
        char tb[4];

        if (!*sock)
                return false;

        rr = drbd_recv_short(*sock, tb, 4, MSG_DONTWAIT | MSG_PEEK);

        if (rr > 0 || rr == -EAGAIN) {
                return true;
        } else {
                sock_release(*sock);
                *sock = NULL;
                return false;
        }
}

static bool connection_established(struct drbd_connection *connection,
                                   struct socket **sock1,
                                   struct socket **sock2)
{
        struct net_conf *nc;
        int timeout;
        bool ok;

        if (!*sock1 || !*sock2)
                return false;

        rcu_read_lock();
        nc = rcu_dereference(connection->net_conf);
        timeout = (nc->sock_check_timeo ?: nc->ping_timeo) * HZ / 10;
        rcu_read_unlock();
        schedule_timeout_interruptible(timeout);

        ok = drbd_socket_okay(sock1);
        ok = drbd_socket_okay(sock2) && ok;

        return ok;
}

/* Gets called if a connection is established, or if a new minor gets created
   in a connection */
int drbd_connected(struct drbd_peer_device *peer_device)
{
        struct drbd_device *device = peer_device->device;
        int err;

        atomic_set(&device->packet_seq, 0);
        device->peer_seq = 0;

        device->state_mutex = peer_device->connection->agreed_pro_version < 100 ?
                &peer_device->connection->cstate_mutex :
                &device->own_state_mutex;

        err = drbd_send_sync_param(peer_device);
        if (!err)
                err = drbd_send_sizes(peer_device, 0, 0);
        if (!err)
                err = drbd_send_uuids(peer_device);
        if (!err)
                err = drbd_send_current_state(peer_device);
        clear_bit(USE_DEGR_WFC_T, &device->flags);
        clear_bit(RESIZE_PENDING, &device->flags);
        atomic_set(&device->ap_in_flight, 0);
        mod_timer(&device->request_timer, jiffies + HZ); /* just start it here. */
        return err;
}

/*
 * return values:
 *   1 yes, we have a valid connection
 *   0 oops, did not work out, please try again
 *  -1 peer talks different language,
 *     no point in trying again, please go standalone.
 *  -2 We do not have a network config...
 */
static int conn_connect(struct drbd_connection *connection)
{
        struct drbd_socket sock, msock;
        struct drbd_peer_device *peer_device;
        struct net_conf *nc;
        int vnr, timeout, h;
        bool discard_my_data, ok;
        enum drbd_state_rv rv;
        struct accept_wait_data ad = {
                .connection = connection,
                .door_bell = COMPLETION_INITIALIZER_ONSTACK(ad.door_bell),
        };

        clear_bit(DISCONNECT_SENT, &connection->flags);
        if (conn_request_state(connection, NS(conn, C_WF_CONNECTION), CS_VERBOSE) < SS_SUCCESS)
                return -2;

        mutex_init(&sock.mutex);
        sock.sbuf = connection->data.sbuf;
        sock.rbuf = connection->data.rbuf;
        sock.socket = NULL;
        mutex_init(&msock.mutex);
        msock.sbuf = connection->meta.sbuf;
        msock.rbuf = connection->meta.rbuf;
        msock.socket = NULL;

        /* Assume that the peer only understands protocol 80 until we know better.  */
        connection->agreed_pro_version = 80;

        if (prepare_listen_socket(connection, &ad))
                return 0;

        do {
                struct socket *s;

                s = drbd_try_connect(connection);
                if (s) {
                        if (!sock.socket) {
                                sock.socket = s;
                                send_first_packet(connection, &sock, P_INITIAL_DATA);
                        } else if (!msock.socket) {
                                clear_bit(RESOLVE_CONFLICTS, &connection->flags);
                                msock.socket = s;
                                send_first_packet(connection, &msock, P_INITIAL_META);
                        } else {
                                drbd_err(connection, "Logic error in conn_connect()\n");
                                goto out_release_sockets;
                        }
                }

                if (connection_established(connection, &sock.socket, &msock.socket))
                        break;

retry:
                s = drbd_wait_for_connect(connection, &ad);
                if (s) {
                        int fp = receive_first_packet(connection, s);
                        drbd_socket_okay(&sock.socket);
                        drbd_socket_okay(&msock.socket);
                        switch (fp) {
                        case P_INITIAL_DATA:
                                if (sock.socket) {
                                        drbd_warn(connection, "initial packet S crossed\n");
                                        sock_release(sock.socket);
                                        sock.socket = s;
                                        goto randomize;
                                }
                                sock.socket = s;
                                break;
                        case P_INITIAL_META:
                                set_bit(RESOLVE_CONFLICTS, &connection->flags);
                                if (msock.socket) {
                                        drbd_warn(connection, "initial packet M crossed\n");
                                        sock_release(msock.socket);
                                        msock.socket = s;
                                        goto randomize;
                                }
                                msock.socket = s;
                                break;

                        default:
                                drbd_warn(connection, "Error receiving initial packet\n");
                                sock_release(s);
randomize:
                                if (prandom_u32() & 1)
                                        goto retry;
                        }
                }

                if (connection->cstate <= C_DISCONNECTING)
                        goto out_release_sockets;
                if (signal_pending(current)) {
                        flush_signals(current);
                        smp_rmb();
                        if (get_t_state(&connection->receiver) == EXITING)
                                goto out_release_sockets;
                }

                ok = connection_established(connection, &sock.socket, &msock.socket);
        } while (!ok);

        if (ad.s_listen)
                sock_release(ad.s_listen);

        sock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */
        msock.socket->sk->sk_reuse = SK_CAN_REUSE; /* SO_REUSEADDR */

        sock.socket->sk->sk_allocation = GFP_NOIO;
        msock.socket->sk->sk_allocation = GFP_NOIO;

        sock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK;
        msock.socket->sk->sk_priority = TC_PRIO_INTERACTIVE;

        /* NOT YET ...
         * sock.socket->sk->sk_sndtimeo = connection->net_conf->timeout*HZ/10;
         * sock.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;
         * first set it to the P_CONNECTION_FEATURES timeout,
         * which we set to 4x the configured ping_timeout. */
        rcu_read_lock();
        nc = rcu_dereference(connection->net_conf);

        sock.socket->sk->sk_sndtimeo =
        sock.socket->sk->sk_rcvtimeo = nc->ping_timeo*4*HZ/10;

        msock.socket->sk->sk_rcvtimeo = nc->ping_int*HZ;
        timeout = nc->timeout * HZ / 10;
        discard_my_data = nc->discard_my_data;
        rcu_read_unlock();

        msock.socket->sk->sk_sndtimeo = timeout;

        /* we don't want delays.
         * we use TCP_CORK where appropriate, though */
        tcp_sock_set_nodelay(sock.socket->sk);
        tcp_sock_set_nodelay(msock.socket->sk);

        connection->data.socket = sock.socket;
        connection->meta.socket = msock.socket;
        connection->last_received = jiffies;

        h = drbd_do_features(connection);
        if (h <= 0)
                return h;

        if (connection->cram_hmac_tfm) {
                /* drbd_request_state(device, NS(conn, WFAuth)); */
                switch (drbd_do_auth(connection)) {
                case -1:
                        drbd_err(connection, "Authentication of peer failed\n");
                        return -1;
                case 0:
                        drbd_err(connection, "Authentication of peer failed, trying again.\n");
                        return 0;
                }
        }

        connection->data.socket->sk->sk_sndtimeo = timeout;
        connection->data.socket->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT;

        if (drbd_send_protocol(connection) == -EOPNOTSUPP)
                return -1;

        /* Prevent a race between resync-handshake and
         * being promoted to Primary.
         *
         * Grab and release the state mutex, so we know that any current
         * drbd_set_role() is finished, and any incoming drbd_set_role
         * will see the STATE_SENT flag, and wait for it to be cleared.
         */
        idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
                mutex_lock(peer_device->device->state_mutex);

        /* avoid a race with conn_request_state( C_DISCONNECTING ) */
        spin_lock_irq(&connection->resource->req_lock);
        set_bit(STATE_SENT, &connection->flags);
        spin_unlock_irq(&connection->resource->req_lock);

        idr_for_each_entry(&connection->peer_devices, peer_device, vnr)
                mutex_unlock(peer_device->device->state_mutex);

        rcu_read_lock();
        idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
                struct drbd_device *device = peer_device->device;
                kref_get(&device->kref);
                rcu_read_unlock();

                if (discard_my_data)
                        set_bit(DISCARD_MY_DATA, &device->flags);
                else
                        clear_bit(DISCARD_MY_DATA, &device->flags);

                drbd_connected(peer_device);
                kref_put(&device->kref, drbd_destroy_device);
                rcu_read_lock();
        }
        rcu_read_unlock();

        rv = conn_request_state(connection, NS(conn, C_WF_REPORT_PARAMS), CS_VERBOSE);
        if (rv < SS_SUCCESS || connection->cstate != C_WF_REPORT_PARAMS) {
                clear_bit(STATE_SENT, &connection->flags);
                return 0;
        }

        drbd_thread_start(&connection->ack_receiver);
        /* opencoded create_singlethread_workqueue(),
         * to be able to use format string arguments */
        connection->ack_sender =
                alloc_ordered_workqueue("drbd_as_%s", WQ_MEM_RECLAIM, connection->resource->name);
        if (!connection->ack_sender) {
                drbd_err(connection, "Failed to create workqueue ack_sender\n");
                return 0;
        }

        mutex_lock(&connection->resource->conf_update);
        /* The discard_my_data flag is a single-shot modifier to the next
         * connection attempt, the handshake of which is now well underway.
         * No need for rcu style copying of the whole struct
         * just to clear a single value. */
        connection->net_conf->discard_my_data = 0;
        mutex_unlock(&connection->resource->conf_update);

        return h;

out_release_sockets:
        if (ad.s_listen)
                sock_release(ad.s_listen);
        if (sock.socket)
                sock_release(sock.socket);
        if (msock.socket)
                sock_release(msock.socket);
        return -1;
}

static int decode_header(struct drbd_connection *connection, void *header, struct packet_info *pi)
{
        unsigned int header_size = drbd_header_size(connection);

        if (header_size == sizeof(struct p_header100) &&
            *(__be32 *)header == cpu_to_be32(DRBD_MAGIC_100)) {
                struct p_header100 *h = header;
                if (h->pad != 0) {
                        drbd_err(connection, "Header padding is not zero\n");
                        return -EINVAL;
                }
                pi->vnr = be16_to_cpu(h->volume);
                pi->cmd = be16_to_cpu(h->command);
                pi->size = be32_to_cpu(h->length);
        } else if (header_size == sizeof(struct p_header95) &&
                   *(__be16 *)header == cpu_to_be16(DRBD_MAGIC_BIG)) {
                struct p_header95 *h = header;
                pi->cmd = be16_to_cpu(h->command);
                pi->size = be32_to_cpu(h->length);
                pi->vnr = 0;
        } else if (header_size == sizeof(struct p_header80) &&
                   *(__be32 *)header == cpu_to_be32(DRBD_MAGIC)) {
                struct p_header80 *h = header;
                pi->cmd = be16_to_cpu(h->command);
                pi->size = be16_to_cpu(h->length);
                pi->vnr = 0;
        } else {
                drbd_err(connection, "Wrong magic value 0x%08x in protocol version %d\n",
                         be32_to_cpu(*(__be32 *)header),
                         connection->agreed_pro_version);
                return -EINVAL;
        }
        pi->data = header + header_size;
        return 0;
}

static void drbd_unplug_all_devices(struct drbd_connection *connection)
{
        if (current->plug == &connection->receiver_plug) {
                blk_finish_plug(&connection->receiver_plug);
                blk_start_plug(&connection->receiver_plug);
        } /* else: maybe just schedule() ?? */
}

static int drbd_recv_header(struct drbd_connection *connection, struct packet_info *pi)
{
        void *buffer = connection->data.rbuf;
        int err;

        err = drbd_recv_all_warn(connection, buffer, drbd_header_size(connection));
        if (err)
                return err;

        err = decode_header(connection, buffer, pi);
        connection->last_received = jiffies;

        return err;
}

static int drbd_recv_header_maybe_unplug(struct drbd_connection *connection, struct packet_info *pi)
{
        void *buffer = connection->data.rbuf;
        unsigned int size = drbd_header_size(connection);
        int err;

        err = drbd_recv_short(connection->data.socket, buffer, size, MSG_NOSIGNAL|MSG_DONTWAIT);
        if (err != size) {
                /* If we have nothing in the receive buffer now, to reduce
                 * application latency, try to drain the backend queues as
                 * quickly as possible, and let remote TCP know what we have
                 * received so far. */
                if (err == -EAGAIN) {
                        tcp_sock_set_quickack(connection->data.socket->sk, 2);
                        drbd_unplug_all_devices(connection);
                }
                if (err > 0) {
                        buffer += err;
                        size -= err;
                }
                err = drbd_recv_all_warn(connection, buffer, size);
                if (err)
                        return err;
        }

        err = decode_header(connection, connection->data.rbuf, pi);
        connection->last_received = jiffies;

        return err;
}
/* This is blkdev_issue_flush, but asynchronous.
 * We want to submit to all component volumes in parallel,
 * then wait for all completions.
 */
struct issue_flush_context {
        atomic_t pending;
        int error;
        struct completion done;
};
struct one_flush_context {
        struct drbd_device *device;
        struct issue_flush_context *ctx;
};

static void one_flush_endio(struct bio *bio)
{
        struct one_flush_context *octx = bio->bi_private;
        struct drbd_device *device = octx->device;
        struct issue_flush_context *ctx = octx->ctx;

        if (bio->bi_status) {
                ctx->error = blk_status_to_errno(bio->bi_status);
                drbd_info(device, "local disk FLUSH FAILED with status %d\n", bio->bi_status);
        }
        kfree(octx);
        bio_put(bio);

        clear_bit(FLUSH_PENDING, &device->flags);
        put_ldev(device);
        kref_put(&device->kref, drbd_destroy_device);

        if (atomic_dec_and_test(&ctx->pending))
                complete(&ctx->done);
}

static void submit_one_flush(struct drbd_device *device, struct issue_flush_context *ctx)
{
        struct bio *bio = bio_alloc(GFP_NOIO, 0);
        struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);
        if (!bio || !octx) {
                drbd_warn(device, "Could not allocate a bio, CANNOT ISSUE FLUSH\n");
                /* FIXME: what else can I do now?  disconnecting or detaching
                 * really does not help to improve the state of the world, either.
                 */
                kfree(octx);
                if (bio)
                        bio_put(bio);

                ctx->error = -ENOMEM;
                put_ldev(device);
                kref_put(&device->kref, drbd_destroy_device);
                return;
        }

        octx->device = device;
        octx->ctx = ctx;
        bio_set_dev(bio, device->ldev->backing_bdev);
        bio->bi_private = octx;
        bio->bi_end_io = one_flush_endio;
        bio->bi_opf = REQ_OP_FLUSH | REQ_PREFLUSH;

        device->flush_jif = jiffies;
        set_bit(FLUSH_PENDING, &device->flags);
        atomic_inc(&ctx->pending);
        submit_bio(bio);
}

static void drbd_flush(struct drbd_connection *connection)
{
        if (connection->resource->write_ordering >= WO_BDEV_FLUSH) {
                struct drbd_peer_device *peer_device;
                struct issue_flush_context ctx;
                int vnr;

                atomic_set(&ctx.pending, 1);
                ctx.error = 0;
                init_completion(&ctx.done);

                rcu_read_lock();
                idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
                        struct drbd_device *device = peer_device->device;

                        if (!get_ldev(device))
                                continue;
                        kref_get(&device->kref);
                        rcu_read_unlock();

                        submit_one_flush(device, &ctx);

                        rcu_read_lock();
                }
                rcu_read_unlock();

                /* Do we want to add a timeout,
                 * if disk-timeout is set? */
                if (!atomic_dec_and_test(&ctx.pending))
                        wait_for_completion(&ctx.done);

                if (ctx.error) {
                        /* would rather check on EOPNOTSUPP, but that is not reliable.
                         * don't try again for ANY return value != 0
                         * if (rv == -EOPNOTSUPP) */
                        /* Any error is already reported by bio_endio callback. */
                        drbd_bump_write_ordering(connection->resource, NULL, WO_DRAIN_IO);
                }
        }
}

/**
 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
 * @device:     DRBD device.
 * @epoch:      Epoch object.
 * @ev:         Epoch event.
 */
static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connection,
                                               struct drbd_epoch *epoch,
                                               enum epoch_event ev)
{
        int epoch_size;
        struct drbd_epoch *next_epoch;
        enum finish_epoch rv = FE_STILL_LIVE;

        spin_lock(&connection->epoch_lock);
        do {
                next_epoch = NULL;

                epoch_size = atomic_read(&epoch->epoch_size);

                switch (ev & ~EV_CLEANUP) {
                case EV_PUT:
                        atomic_dec(&epoch->active);
                        break;
                case EV_GOT_BARRIER_NR:
                        set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags);
                        break;
                case EV_BECAME_LAST:
                        /* nothing to do*/
                        break;
                }

                if (epoch_size != 0 &&
                    atomic_read(&epoch->active) == 0 &&
                    (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) || ev & EV_CLEANUP)) {
                        if (!(ev & EV_CLEANUP)) {
                                spin_unlock(&connection->epoch_lock);
                                drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
                                spin_lock(&connection->epoch_lock);
                        }
#if 0
                        /* FIXME: dec unacked on connection, once we have
                         * something to count pending connection packets in. */
                        if (test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags))
                                dec_unacked(epoch->connection);
#endif

                        if (connection->current_epoch != epoch) {
                                next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list);
                                list_del(&epoch->list);
                                ev = EV_BECAME_LAST | (ev & EV_CLEANUP);
                                connection->epochs--;
                                kfree(epoch);

                                if (rv == FE_STILL_LIVE)
                                        rv = FE_DESTROYED;
                        } else {
                                epoch->flags = 0;
                                atomic_set(&epoch->epoch_size, 0);
                                /* atomic_set(&epoch->active, 0); is already zero */
                                if (rv == FE_STILL_LIVE)
                                        rv = FE_RECYCLED;
                        }
                }

                if (!next_epoch)
                        break;

                epoch = next_epoch;
        } while (1);

        spin_unlock(&connection->epoch_lock);

        return rv;
}

static enum write_ordering_e
max_allowed_wo(struct drbd_backing_dev *bdev, enum write_ordering_e wo)
{
        struct disk_conf *dc;

        dc = rcu_dereference(bdev->disk_conf);

        if (wo == WO_BDEV_FLUSH && !dc->disk_flushes)
                wo = WO_DRAIN_IO;
        if (wo == WO_DRAIN_IO && !dc->disk_drain)
                wo = WO_NONE;

        return wo;
}

/**
 * drbd_bump_write_ordering() - Fall back to an other write ordering method
 * @connection: DRBD connection.
 * @wo:         Write ordering method to try.
 */
void drbd_bump_write_ordering(struct drbd_resource *resource, struct drbd_backing_dev *bdev,
                              enum write_ordering_e wo)
{
        struct drbd_device *device;
        enum write_ordering_e pwo;
        int vnr;
        static char *write_ordering_str[] = {
                [WO_NONE] = "none",
                [WO_DRAIN_IO] = "drain",
                [WO_BDEV_FLUSH] = "flush",
        };

        pwo = resource->write_ordering;
        if (wo != WO_BDEV_FLUSH)
                wo = min(pwo, wo);
        rcu_read_lock();
        idr_for_each_entry(&resource->devices, device, vnr) {
                if (get_ldev(device)) {
                        wo = max_allowed_wo(device->ldev, wo);
                        if (device->ldev == bdev)
                                bdev = NULL;
                        put_ldev(device);
                }
        }

        if (bdev)
                wo = max_allowed_wo(bdev, wo);

        rcu_read_unlock();

        resource->write_ordering = wo;
        if (pwo != resource->write_ordering || wo == WO_BDEV_FLUSH)
                drbd_info(resource, "Method to ensure write ordering: %s\n", write_ordering_str[resource->write_ordering]);
}

/*
 * Mapping "discard" to ZEROOUT with UNMAP does not work for us:
 * Drivers have to "announce" q->limits.max_write_zeroes_sectors, or it
 * will directly go to fallback mode, submitting normal writes, and
 * never even try to UNMAP.
 *
 * And dm-thin does not do this (yet), mostly because in general it has
 * to assume that "skip_block_zeroing" is set.  See also:
 * https://www.mail-archive.com/dm-devel%40redhat.com/msg07965.html
 * https://www.redhat.com/archives/dm-devel/2018-January/msg00271.html
 *
 * We *may* ignore the discard-zeroes-data setting, if so configured.
 *
 * Assumption is that this "discard_zeroes_data=0" is only because the backend
 * may ignore partial unaligned discards.
 *
 * LVM/DM thin as of at least
 *   LVM version:     2.02.115(2)-RHEL7 (2015-01-28)
 *   Library version: 1.02.93-RHEL7 (2015-01-28)
 *   Driver version:  4.29.0
 * still behaves this way.
 *
 * For unaligned (wrt. alignment and granularity) or too small discards,
 * we zero-out the initial (and/or) trailing unaligned partial chunks,
 * but discard all the aligned full chunks.
 *
 * At least for LVM/DM thin, with skip_block_zeroing=false,
 * the result is effectively "discard_zeroes_data=1".
 */
/* flags: EE_TRIM|EE_ZEROOUT */
int drbd_issue_discard_or_zero_out(struct drbd_device *device, sector_t start, unsigned int nr_sectors, int flags)
{
        struct block_device *bdev = device->ldev->backing_bdev;
        struct request_queue *q = bdev_get_queue(bdev);
        sector_t tmp, nr;
        unsigned int max_discard_sectors, granularity;
        int alignment;
        int err = 0;

        if ((flags & EE_ZEROOUT) || !(flags & EE_TRIM))
                goto zero_out;

        /* Zero-sector (unknown) and one-sector granularities are the same.  */
        granularity = max(q->limits.discard_granularity >> 9, 1U);
        alignment = (bdev_discard_alignment(bdev) >> 9) % granularity;

        max_discard_sectors = min(q->limits.max_discard_sectors, (1U << 22));
        max_discard_sectors -= max_discard_sectors % granularity;
        if (unlikely(!max_discard_sectors))
                goto zero_out;

        if (nr_sectors < granularity)
                goto zero_out;

        tmp = start;
        if (sector_div(tmp, granularity) != alignment) {
                if (nr_sectors < 2*granularity)
                        goto zero_out;
                /* start + gran - (start + gran - align) % gran */
                tmp = start + granularity - alignment;
                tmp = start + granularity - sector_div(tmp, granularity);

                nr = tmp - start;
                /* don't flag BLKDEV_ZERO_NOUNMAP, we don't know how many
                 * layers are below us, some may have smaller granularity */
                err |= blkdev_issue_zeroout(bdev, start, nr, GFP_NOIO, 0);
                nr_sectors -= nr;
                start = tmp;
        }
        while (nr_sectors >= max_discard_sectors) {
                err |= blkdev_issue_discard(bdev, start, max_discard_sectors, GFP_NOIO, 0);
                nr_sectors -= max_discard_sectors;
                start += max_discard_sectors;
        }
        if (nr_sectors) {
                /* max_discard_sectors is unsigned int (and a multiple of
                 * granularity, we made sure of that above already);
                 * nr is < max_discard_sectors;
                 * I don't need sector_div here, even though nr is sector_t */
                nr = nr_sectors;
                nr -= (unsigned int)nr % granularity;
                if (nr) {
                        err |= blkdev_issue_discard(bdev, start, nr, GFP_NOIO, 0);
                        nr_sectors -= nr;
                        start += nr;
                }
        }
 zero_out:
        if (nr_sectors) {
                err |= blkdev_issue_zeroout(bdev, start, nr_sectors, GFP_NOIO,
                                (flags & EE_TRIM) ? 0 : BLKDEV_ZERO_NOUNMAP);
        }
        return err != 0;
}

static bool can_do_reliable_discards(struct drbd_device *device)
{
        struct request_queue *q = bdev_get_queue(device->ldev->backing_bdev);
        struct disk_conf *dc;
        bool can_do;

        if (!blk_queue_discard(q))
                return false;

        rcu_read_lock();
        dc = rcu_dereference(device->ldev->disk_conf);
        can_do = dc->discard_zeroes_if_aligned;
        rcu_read_unlock();
        return can_do;
}

static void drbd_issue_peer_discard_or_zero_out(struct drbd_device *device, struct drbd_peer_request *peer_req)
{
        /* If the backend cannot discard, or does not guarantee
         * read-back zeroes in discarded ranges, we fall back to
         * zero-out.  Unless configuration specifically requested
         * otherwise. */
        if (!can_do_reliable_discards(device))
                peer_req->flags |= EE_ZEROOUT;

        if (drbd_issue_discard_or_zero_out(device, peer_req->i.sector,
            peer_req->i.size >> 9, peer_req->flags & (EE_ZEROOUT|EE_TRIM)))
                peer_req->flags |= EE_WAS_ERROR;
        drbd_endio_write_sec_final(peer_req);
}

static void drbd_issue_peer_wsame(struct drbd_device *device,
                                  struct drbd_peer_request *peer_req)
{
        struct block_device *bdev = device->ldev->backing_bdev;
        sector_t s = peer_req->i.sector;
        sector_t nr = peer_req->i.size >> 9;
        if (blkdev_issue_write_same(bdev, s, nr, GFP_NOIO, peer_req->pages))
                peer_req->flags |= EE_WAS_ERROR;
        drbd_endio_write_sec_final(peer_req);
}


/**
 * drbd_submit_peer_request()
 * @device:     DRBD device.
 * @peer_req:   peer request
 * @rw:         flag field, see bio->bi_opf
 *
 * May spread the pages to multiple bios,
 * depending on bio_add_page restrictions.
 *
 * Returns 0 if all bios have been submitted,
 * -ENOMEM if we could not allocate enough bios,
 * -ENOSPC (any better suggestion?) if we have not been able to bio_add_page a
 *  single page to an empty bio (which should never happen and likely indicates
 *  that the lower level IO stack is in some way broken). This has been observed
 *  on certain Xen deployments.
 */
/* TODO allocate from our own bio_set. */
int drbd_submit_peer_request(struct drbd_device *device,
                             struct drbd_peer_request *peer_req,
                             const unsigned op, const unsigned op_flags,
                             const int fault_type)
{
        struct bio *bios = NULL;
        struct bio *bio;
        struct page *page = peer_req->pages;
        sector_t sector = peer_req->i.sector;
        unsigned data_size = peer_req->i.size;
        unsigned n_bios = 0;
        unsigned nr_pages = (data_size + PAGE_SIZE -1) >> PAGE_SHIFT;
        int err = -ENOMEM;

        /* TRIM/DISCARD: for now, always use the helper function
         * blkdev_issue_zeroout(..., discard=true).
         * It's synchronous, but it does the right thing wrt. bio splitting.
         * Correctness first, performance later.  Next step is to code an
         * asynchronous variant of the same.
         */
        if (peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) {
                /* wait for all pending IO completions, before we start
                 * zeroing things out. */
                conn_wait_active_ee_empty(peer_req->peer_device->connection);
                /* add it to the active list now,
                 * so we can find it to present it in debugfs */
                peer_req->submit_jif = jiffies;
                peer_req->flags |= EE_SUBMITTED;

                /* If this was a resync request from receive_rs_deallocated(),
                 * it is already on the sync_ee list */
                if (list_empty(&peer_req->w.list)) {
                        spin_lock_irq(&device->resource->req_lock);
                        list_add_tail(&peer_req->w.list, &device->active_ee);
                        spin_unlock_irq(&device->resource->req_lock);
                }

                if (peer_req->flags & (EE_TRIM|EE_ZEROOUT))
                        drbd_issue_peer_discard_or_zero_out(device, peer_req);
                else /* EE_WRITE_SAME */
                        drbd_issue_peer_wsame(device, peer_req);
                return 0;
        }

        /* In most cases, we will only need one bio.  But in case the lower
         * level restrictions happen to be different at this offset on this
         * side than those of the sending peer, we may need to submit the
         * request in more than one bio.
         *
         * Plain bio_alloc is good enough here, this is no DRBD internally
         * generated bio, but a bio allocated on behalf of the peer.
         */
next_bio:
        bio = bio_alloc(GFP_NOIO, nr_pages);
        if (!bio) {
                drbd_err(device, "submit_ee: Allocation of a bio failed (nr_pages=%u)\n", nr_pages);
                goto fail;
        }
        /* > peer_req->i.sector, unless this is the first bio */
        bio->bi_iter.bi_sector = sector;
        bio_set_dev(bio, device->ldev->backing_bdev);
        bio_set_op_attrs(bio, op, op_flags);
        bio->bi_private = peer_req;
        bio->bi_end_io = drbd_peer_request_endio;

        bio->bi_next = bios;
        bios = bio;
        ++n_bios;

        page_chain_for_each(page) {
                unsigned len = min_t(unsigned, data_size, PAGE_SIZE);
                if (!bio_add_page(bio, page, len, 0))
                        goto next_bio;
                data_size -= len;
                sector += len >> 9;
                --nr_pages;
        }
        D_ASSERT(device, data_size == 0);
        D_ASSERT(device, page == NULL);

        atomic_set(&peer_req->pending_bios, n_bios);
        /* for debugfs: update timestamp, mark as submitted */
        peer_req->submit_jif = jiffies;
        peer_req->flags |= EE_SUBMITTED;
        do {
                bio = bios;
                bios = bios->bi_next;
                bio->bi_next = NULL;

                drbd_submit_bio_noacct(device, fault_type, bio);
        } while (bios);
        return 0;

fail:
        while (bios) {
                bio = bios;
                bios = bios->bi_next;
                bio_put(bio);
        }
        return err;
}

static void drbd_remove_epoch_entry_interval(struct drbd_device *device,
                                             struct drbd_peer_request *peer_req)
{
        struct drbd_interval *i = &peer_req->i;

        drbd_remove_interval(&device->write_requests, i);
        drbd_clear_interval(i);

        /* Wake up any processes waiting for this peer request to complete.  */
        if (i->waiting)
                wake_up(&device->misc_wait);
}

static void conn_wait_active_ee_empty(struct drbd_connection *connection)
{
        struct drbd_peer_device *peer_device;
        int vnr;

        rcu_read_lock();
        idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
                struct drbd_device *device = peer_device->device;

                kref_get(&device->kref);
                rcu_read_unlock();
                drbd_wait_ee_list_empty(device, &device->active_ee);
                kref_put(&device->kref, drbd_destroy_device);
                rcu_read_lock();
        }
        rcu_read_unlock();
}

static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
{
        int rv;
        struct p_barrier *p = pi->data;
        struct drbd_epoch *epoch;

        /* FIXME these are unacked on connection,
         * not a specific (peer)device.
         */
        connection->current_epoch->barrier_nr = p->barrier;
        connection->current_epoch->connection = connection;
        rv = drbd_may_finish_epoch(connection, connection->current_epoch, EV_GOT_BARRIER_NR);

        /* P_BARRIER_ACK may imply that the corresponding extent is dropped from
         * the activity log, which means it would not be resynced in case the
         * R_PRIMARY crashes now.
         * Therefore we must send the barrier_ack after the barrier request was
         * completed. */
        switch (connection->resource->write_ordering) {
        case WO_NONE:
                if (rv == FE_RECYCLED)
                        return 0;

                /* receiver context, in the writeout path of the other node.
                 * avoid potential distributed deadlock */
                epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
                if (epoch)
                        break;
                else
                        drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
                fallthrough;

        case WO_BDEV_FLUSH:
        case WO_DRAIN_IO:
                conn_wait_active_ee_empty(connection);
                drbd_flush(connection);

                if (atomic_read(&connection->current_epoch->epoch_size)) {
                        epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
                        if (epoch)
                                break;
                }

                return 0;
        default:
                drbd_err(connection, "Strangeness in connection->write_ordering %d\n",
                         connection->resource->write_ordering);
                return -EIO;
        }

        epoch->flags = 0;
        atomic_set(&epoch->epoch_size, 0);
        atomic_set(&epoch->active, 0);

        spin_lock(&connection->epoch_lock);
        if (atomic_read(&connection->current_epoch->epoch_size)) {
                list_add(&epoch->list, &connection->current_epoch->list);
                connection->current_epoch = epoch;
                connection->epochs++;
        } else {
                /* The current_epoch got recycled while we allocated this one... */
                kfree(epoch);
        }
        spin_unlock(&connection->epoch_lock);

        return 0;
}

/* quick wrapper in case payload size != request_size (write same) */
static void drbd_csum_ee_size(struct crypto_shash *h,
                              struct drbd_peer_request *r, void *d,
                              unsigned int payload_size)
{
        unsigned int tmp = r->i.size;
        r->i.size = payload_size;
        drbd_csum_ee(h, r, d);
        r->i.size = tmp;
}

/* used from receive_RSDataReply (recv_resync_read)
 * and from receive_Data.
 * data_size: actual payload ("data in")
 *      for normal writes that is bi_size.
 *      for discards, that is zero.
 *      for write same, it is logical_block_size.
 * both trim and write same have the bi_size ("data len to be affected")
 * as extra argument in the packet header.
 */
static struct drbd_peer_request *
read_in_block(struct drbd_peer_device *peer_device, u64 id, sector_t sector,
              struct packet_info *pi) __must_hold(local)
{
        struct drbd_device *device = peer_device->device;
        const sector_t capacity = drbd_get_capacity(device->this_bdev);
        struct drbd_peer_request *peer_req;
        struct page *page;
        int digest_size, err;
        unsigned int data_size = pi->size, ds;
        void *dig_in = peer_device->connection->int_dig_in;
        void *dig_vv = peer_device->connection->int_dig_vv;
        unsigned long *data;
        struct p_trim *trim = (pi->cmd == P_TRIM) ? pi->data : NULL;
        struct p_trim *zeroes = (pi->cmd == P_ZEROES) ? pi->data : NULL;
        struct p_trim *wsame = (pi->cmd == P_WSAME) ? pi->data : NULL;

        digest_size = 0;
        if (!trim && peer_device->connection->peer_integrity_tfm) {
                digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
                /*
                 * FIXME: Receive the incoming digest into the receive buffer
                 *        here, together with its struct p_data?
                 */
                err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
                if (err)
                        return NULL;
                data_size -= digest_size;
        }

        /* assume request_size == data_size, but special case trim and wsame. */
        ds = data_size;
        if (trim) {
                if (!expect(data_size == 0))
                        return NULL;
                ds = be32_to_cpu(trim->size);
        } else if (zeroes) {
                if (!expect(data_size == 0))
                        return NULL;
                ds = be32_to_cpu(zeroes->size);
        } else if (wsame) {
                if (data_size != queue_logical_block_size(device->rq_queue)) {
                        drbd_err(peer_device, "data size (%u) != drbd logical block size (%u)\n",
                                data_size, queue_logical_block_size(device->rq_queue));
                        return NULL;
                }
                if (data_size != bdev_logical_block_size(device->ldev->backing_bdev)) {
                        drbd_err(peer_device, "data size (%u) != backend logical block size (%u)\n",
                                data_size, bdev_logical_block_size(device->ldev->backing_bdev));
                        return NULL;
                }
                ds = be32_to_cpu(wsame->size);
        }

        if (!expect(IS_ALIGNED(ds, 512)))
                return NULL;
        if (trim || wsame || zeroes) {
                if (!expect(ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
                        return NULL;
        } else if (!expect(ds <= DRBD_MAX_BIO_SIZE))
                return NULL;

        /* even though we trust out peer,
         * we sometimes have to double check. */
        if (sector + (ds>>9) > capacity) {
                drbd_err(device, "request from peer beyond end of local disk: "
                        "capacity: %llus < sector: %llus + size: %u\n",
                        (unsigned long long)capacity,
                        (unsigned long long)sector, ds);
                return NULL;
        }

        /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
         * "criss-cross" setup, that might cause write-out on some other DRBD,
         * which in turn might block on the other node at this very place.  */
        peer_req = drbd_alloc_peer_req(peer_device, id, sector, ds, data_size, GFP_NOIO);
        if (!peer_req)
                return NULL;

        peer_req->flags |= EE_WRITE;
        if (trim) {
                peer_req->flags |= EE_TRIM;
                return peer_req;
        }
        if (zeroes) {
                peer_req->flags |= EE_ZEROOUT;
                return peer_req;
        }
        if (wsame)
                peer_req->flags |= EE_WRITE_SAME;

        /* receive payload size bytes into page chain */
        ds = data_size;
        page = peer_req->pages;
        page_chain_for_each(page) {
                unsigned len = min_t(int, ds, PAGE_SIZE);
                data = kmap(page);
                err = drbd_recv_all_warn(peer_device->connection, data, len);
                if (drbd_insert_fault(device, DRBD_FAULT_RECEIVE)) {
                        drbd_err(device, "Fault injection: Corrupting data on receive\n");
                        data[0] = data[0] ^ (unsigned long)-1;
                }
                kunmap(page);
                if (err) {
                        drbd_free_peer_req(device, peer_req);
                        return NULL;
                }
                ds -= len;
        }

        if (digest_size) {
                drbd_csum_ee_size(peer_device->connection->peer_integrity_tfm, peer_req, dig_vv, data_size);
                if (memcmp(dig_in, dig_vv, digest_size)) {
                        drbd_err(device, "Digest integrity check FAILED: %llus +%u\n",
                                (unsigned long long)sector, data_size);
                        drbd_free_peer_req(device, peer_req);
                        return NULL;
                }
        }
        device->recv_cnt += data_size >> 9;
        return peer_req;
}

/* drbd_drain_block() just takes a data block
 * out of the socket input buffer, and discards it.
 */
static int drbd_drain_block(struct drbd_peer_device *peer_device, int data_size)
{
        struct page *page;
        int err = 0;
        void *data;

        if (!data_size)
                return 0;

        page = drbd_alloc_pages(peer_device, 1, 1);

        data = kmap(page);
        while (data_size) {
                unsigned int len = min_t(int, data_size, PAGE_SIZE);

                err = drbd_recv_all_warn(peer_device->connection, data, len);
                if (err)

                        break;
                data_size -= len;
        }
        kunmap(page);
        drbd_free_pages(peer_device->device, page, 0);
        return err;
}

static int recv_dless_read(struct drbd_peer_device *peer_device, struct drbd_request *req,
                           sector_t sector, int data_size)
{
        struct bio_vec bvec;
        struct bvec_iter iter;
        struct bio *bio;
        int digest_size, err, expect;
        void *dig_in = peer_device->connection->int_dig_in;
        void *dig_vv = peer_device->connection->int_dig_vv;

        digest_size = 0;
        if (peer_device->connection->peer_integrity_tfm) {
                digest_size = crypto_shash_digestsize(peer_device->connection->peer_integrity_tfm);
                err = drbd_recv_all_warn(peer_device->connection, dig_in, digest_size);
                if (err)
                        return err;
                data_size -= digest_size;
        }

        /* optimistically update recv_cnt.  if receiving fails below,
         * we disconnect anyways, and counters will be reset. */
        peer_device->device->recv_cnt += data_size>>9;

        bio = req->master_bio;
        D_ASSERT(peer_device->device, sector == bio->bi_iter.bi_sector);

        bio_for_each_segment(bvec, bio, iter) {
                void *mapped = kmap(bvec.bv_page) + bvec.bv_offset;
                expect = min_t(int, data_size, bvec.bv_len);
                err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
                kunmap(bvec.bv_page);
                if (err)
                        return err;
                data_size -= expect;
        }

        if (digest_size) {
                drbd_csum_bio(peer_device->connection->peer_integrity_tfm, bio, dig_vv);
                if (memcmp(dig_in, dig_vv, digest_size)) {
                        drbd_err(peer_device, "Digest integrity check FAILED. Broken NICs?\n");
                        return -EINVAL;
                }
        }

        D_ASSERT(peer_device->device, data_size == 0);
        return 0;
}

/*
 * e_end_resync_block() is called in ack_sender context via
 * drbd_finish_peer_reqs().
 */
static int e_end_resync_block(struct drbd_work *w, int unused)
{
        struct drbd_peer_request *peer_req =
                container_of(w, struct drbd_peer_request, w);
        struct drbd_peer_device *peer_device = peer_req->peer_device;
        struct drbd_device *device = peer_device->device;
        sector_t sector = peer_req->i.sector;
        int err;

        D_ASSERT(device, drbd_interval_empty(&peer_req->i));

        if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
                drbd_set_in_sync(device, sector, peer_req->i.size);
                err = drbd_send_ack(peer_device, P_RS_WRITE_ACK, peer_req);
        } else {
                /* Record failure to sync */
                drbd_rs_failed_io(device, sector, peer_req->i.size);

                err  = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
        }
        dec_unacked(device);

        return err;
}

static int recv_resync_read(struct drbd_peer_device *peer_device, sector_t sector,
                            struct packet_info *pi) __releases(local)
{
        struct drbd_device *device = peer_device->device;
        struct drbd_peer_request *peer_req;

        peer_req = read_in_block(peer_device, ID_SYNCER, sector, pi);
        if (!peer_req)
                goto fail;

        dec_rs_pending(device);

        inc_unacked(device);
        /* corresponding dec_unacked() in e_end_resync_block()
         * respective _drbd_clear_done_ee */

        peer_req->w.cb = e_end_resync_block;
        peer_req->submit_jif = jiffies;

        spin_lock_irq(&device->resource->req_lock);
        list_add_tail(&peer_req->w.list, &device->sync_ee);
        spin_unlock_irq(&device->resource->req_lock);

        atomic_add(pi->size >> 9, &device->rs_sect_ev);
        if (drbd_submit_peer_request(device, peer_req, REQ_OP_WRITE, 0,
                                     DRBD_FAULT_RS_WR) == 0)
                return 0;

        /* don't care for the reason here */
        drbd_err(device, "submit failed, triggering re-connect\n");
        spin_lock_irq(&device->resource->req_lock);
        list_del(&peer_req->w.list);
        spin_unlock_irq(&device->resource->req_lock);

        drbd_free_peer_req(device, peer_req);
fail:
        put_ldev(device);
        return -EIO;
}

static struct drbd_request *
find_request(struct drbd_device *device, struct rb_root *root, u64 id,
             sector_t sector, bool missing_ok, const char *func)
{
        struct drbd_request *req;

        /* Request object according to our peer */
        req = (struct drbd_request *)(unsigned long)id;
        if (drbd_contains_interval(root, sector, &req->i) && req->i.local)
                return req;
        if (!missing_ok) {
                drbd_err(device, "%s: failed to find request 0x%lx, sector %llus\n", func,
                        (unsigned long)id, (unsigned long long)sector);
        }
        return NULL;
}

static int receive_DataReply(struct drbd_connection *connection, struct packet_info *pi)
{
        struct drbd_peer_device *peer_device;
        struct drbd_device *device;
        struct drbd_request *req;
        sector_t sector;
        int err;
        struct p_data *p = pi->data;

        peer_device = conn_peer_device(connection, pi->vnr);
        if (!peer_device)
                return -EIO;
        device = peer_device->device;

        sector = be64_to_cpu(p->sector);

        spin_lock_irq(&device->resource->req_lock);
        req = find_request(device, &device->read_requests, p->block_id, sector, false, __func__);
        spin_unlock_irq(&device->resource->req_lock);
        if (unlikely(!req))
                return -EIO;

        /* hlist_del(&req->collision) is done in _req_may_be_done, to avoid
         * special casing it there for the various failure cases.
         * still no race with drbd_fail_pending_reads */
        err = recv_dless_read(peer_device, req, sector, pi->size);
        if (!err)
                req_mod(req, DATA_RECEIVED);
        /* else: nothing. handled from drbd_disconnect...
         * I don't think we may complete this just yet
         * in case we are "on-disconnect: freeze" */

        return err;
}

static int receive_RSDataReply(struct drbd_connection *connection, struct packet_info *pi)
{
        struct drbd_peer_device *peer_device;
        struct drbd_device *device;
        sector_t sector;
        int err;
        struct p_data *p = pi->data;

        peer_device = conn_peer_device(connection, pi->vnr);
        if (!peer_device)
                return -EIO;
        device = peer_device->device;

        sector = be64_to_cpu(p->sector);
        D_ASSERT(device, p->block_id == ID_SYNCER);

        if (get_ldev(device)) {
                /* data is submitted to disk within recv_resync_read.
                 * corresponding put_ldev done below on error,
                 * or in drbd_peer_request_endio. */
                err = recv_resync_read(peer_device, sector, pi);
        } else {
                if (__ratelimit(&drbd_ratelimit_state))
                        drbd_err(device, "Can not write resync data to local disk.\n");

                err = drbd_drain_block(peer_device, pi->size);

                drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
        }

        atomic_add(pi->size >> 9, &device->rs_sect_in);

        return err;
}

static void restart_conflicting_writes(struct drbd_device *device,
                                       sector_t sector, int size)
{
        struct drbd_interval *i;
        struct drbd_request *req;

        drbd_for_each_overlap(i, &device->write_requests, sector, size) {
                if (!i->local)
                        continue;
                req = container_of(i, struct drbd_request, i);
                if (req->rq_state & RQ_LOCAL_PENDING ||
                    !(req->rq_state & RQ_POSTPONED))
                        continue;
                /* as it is RQ_POSTPONED, this will cause it to
                 * be queued on the retry workqueue. */
                __req_mod(req, CONFLICT_RESOLVED, NULL);
        }
}

/*
 * e_end_block() is called in ack_sender context via drbd_finish_peer_reqs().
 */
static int e_end_block(struct drbd_work *w, int cancel)
{
        struct drbd_peer_request *peer_req =
                container_of(w, struct drbd_peer_request, w);
        struct drbd_peer_device *peer_device = peer_req->peer_device;
        struct drbd_device *device = peer_device->device;
        sector_t sector = peer_req->i.sector;
        int err = 0, pcmd;

        if (peer_req->flags & EE_SEND_WRITE_ACK) {
                if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
                        pcmd = (device->state.conn >= C_SYNC_SOURCE &&
                                device->state.conn <= C_PAUSED_SYNC_T &&
                                peer_req->flags & EE_MAY_SET_IN_SYNC) ?
                                P_RS_WRITE_ACK : P_WRITE_ACK;
                        err = drbd_send_ack(peer_device, pcmd, peer_req);
                        if (pcmd == P_RS_WRITE_ACK)
                                drbd_set_in_sync(device, sector, peer_req->i.size);
                } else {
                        err = drbd_send_ack(peer_device, P_NEG_ACK, peer_req);
                        /* we expect it to be marked out of sync anyways...
                         * maybe assert this?  */
                }
                dec_unacked(device);
        }

        /* we delete from the conflict detection hash _after_ we sent out the
         * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right.  */
        if (peer_req->flags & EE_IN_INTERVAL_TREE) {
                spin_lock_irq(&device->resource->req_lock);
                D_ASSERT(device, !drbd_interval_empty(&peer_req->i));
                drbd_remove_epoch_entry_interval(device, peer_req);
                if (peer_req->flags & EE_RESTART_REQUESTS)
                        restart_conflicting_writes(device, sector, peer_req->i.size);
                spin_unlock_irq(&device->resource->req_lock);
        } else
                D_ASSERT(device, drbd_interval_empty(&peer_req->i));

        drbd_may_finish_epoch(peer_device->connection, peer_req->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0));

        return err;
}

static int e_send_ack(struct drbd_work *w, enum drbd_packet ack)
{
        struct drbd_peer_request *peer_req =
                container_of(w, struct drbd_peer_request, w);
        struct drbd_peer_device *peer_device = peer_req->peer_device;
        int err;

        err = drbd_send_ack(peer_device, ack, peer_req);
        dec_unacked(peer_device->device);

        return err;
}

static int e_send_superseded(struct drbd_work *w, int unused)
{
        return e_send_ack(w, P_SUPERSEDED);
}

static int e_send_retry_write(struct drbd_work *w, int unused)
{
        struct drbd_peer_request *peer_req =
                container_of(w, struct drbd_peer_request, w);
        struct drbd_connection *connection = peer_req->peer_device->connection;

        return e_send_ack(w, connection->agreed_pro_version >= 100 ?
                             P_RETRY_WRITE : P_SUPERSEDED);
}

static bool seq_greater(u32 a, u32 b)
{
        /*
         * We assume 32-bit wrap-around here.
         * For 24-bit wrap-around, we would have to shift:
         *  a <<= 8; b <<= 8;
         */
        return (s32)a - (s32)b > 0;
}

static u32 seq_max(u32 a, u32 b)
{
        return seq_greater(a, b) ? a : b;
}

static void update_peer_seq(struct drbd_peer_device *peer_device, unsigned int peer_seq)
{
        struct drbd_device *device = peer_device->device;
        unsigned int newest_peer_seq;

        if (test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags)) {
                spin_lock(&device->peer_seq_lock);
                newest_peer_seq = seq_max(device->peer_seq, peer_seq);
                device->peer_seq = newest_peer_seq;
                spin_unlock(&device->peer_seq_lock);
                /* wake up only if we actually changed device->peer_seq */
                if (peer_seq == newest_peer_seq)
                        wake_up(&device->seq_wait);
        }
}

static inline int overlaps(sector_t s1, int l1, sector_t s2, int l2)
{
        return !((s1 + (l1>>9) <= s2) || (s1 >= s2 + (l2>>9)));
}

/* maybe change sync_ee into interval trees as well? */
static bool overlapping_resync_write(struct drbd_device *device, struct drbd_peer_request *peer_req)
{
        struct drbd_peer_request *rs_req;
        bool rv = false;

        spin_lock_irq(&device->resource->req_lock);
        list_for_each_entry(rs_req, &device->sync_ee, w.list) {
                if (overlaps(peer_req->i.sector, peer_req->i.size,
                             rs_req->i.sector, rs_req->i.size)) {
                        rv = true;
                        break;
                }
        }
        spin_unlock_irq(&device->resource->req_lock);

        return rv;
}

/* Called from receive_Data.
 * Synchronize packets on sock with packets on msock.
 *
 * This is here so even when a P_DATA packet traveling via sock overtook an Ack
 * packet traveling on msock, they are still processed in the order they have
 * been sent.
 *
 * Note: we don't care for Ack packets overtaking P_DATA packets.
 *
 * In case packet_seq is larger than device->peer_seq number, there are
 * outstanding packets on the msock. We wait for them to arrive.
 * In case we are the logically next packet, we update device->peer_seq
 * ourselves. Correctly handles 32bit wrap around.
 *
 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second,
 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds
 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have
 * 1<<9 == 512 seconds aka ages for the 32bit wrap around...
 *
 * returns 0 if we may process the packet,
 * -ERESTARTSYS if we were interrupted (by disconnect signal). */
static int wait_for_and_update_peer_seq(struct drbd_peer_device *peer_device, const u32 peer_seq)
{
        struct drbd_device *device = peer_device->device;
        DEFINE_WAIT(wait);
        long timeout;
        int ret = 0, tp;

        if (!test_bit(RESOLVE_CONFLICTS, &peer_device->connection->flags))
                return 0;

        spin_lock(&device->peer_seq_lock);
        for (;;) {
                if (!seq_greater(peer_seq - 1, device->peer_seq)) {
                        device->peer_seq = seq_max(device->peer_seq, peer_seq);
                        break;
                }

                if (signal_pending(current)) {
                        ret = -ERESTARTSYS;
                        break;
                }

                rcu_read_lock();
                tp = rcu_dereference(peer_device->connection->net_conf)->two_primaries;
                rcu_read_unlock();

                if (!tp)
                        break;

                /* Only need to wait if two_primaries is enabled */
                prepare_to_wait(&device->seq_wait, &wait, TASK_INTERRUPTIBLE);
                spin_unlock(&device->peer_seq_lock);
                rcu_read_lock();
                timeout = rcu_dereference(peer_device->connection->net_conf)->ping_timeo*HZ/10;
                rcu_read_unlock();
                timeout = schedule_timeout(timeout);
                spin_lock(&device->peer_seq_lock);
                if (!timeout) {
                        ret = -ETIMEDOUT;
                        drbd_err(device, "Timed out waiting for missing ack packets; disconnecting\n");
                        break;
                }
        }
        spin_unlock(&device->peer_seq_lock);
        finish_wait(&device->seq_wait, &wait);
        return ret;
}

/* see also bio_flags_to_wire()
 * DRBD_REQ_*, because we need to semantically map the flags to data packet
 * flags and back. We may replicate to other kernel versions. */
static unsigned long wire_flags_to_bio_flags(u32 dpf)
{
        return  (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
                (dpf & DP_FUA ? REQ_FUA : 0) |
                (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
}

static unsigned long wire_flags_to_bio_op(u32 dpf)
{
        if (dpf & DP_ZEROES)
                return REQ_OP_WRITE_ZEROES;
        if (dpf & DP_DISCARD)
                return REQ_OP_DISCARD;
        if (dpf & DP_WSAME)
                return REQ_OP_WRITE_SAME;
        else
                return REQ_OP_WRITE;
}

static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
                                    unsigned int size)
{
        struct drbd_interval *i;

    repeat:
        drbd_for_each_overlap(i, &device->write_requests, sector, size) {
                struct drbd_request *req;
                struct bio_and_error m;

                if (!i->local)
                        continue;
                req = container_of(i, struct drbd_request, i);
                if (!(req->rq_state & RQ_POSTPONED))
                        continue;
                req->rq_state &= ~RQ_POSTPONED;
                __req_mod(req, NEG_ACKED, &m);
                spin_unlock_irq(&device->resource->req_lock);
                if (m.bio)
                        complete_master_bio(device, &m);
                spin_lock_irq(&device->resource->req_lock);
                goto repeat;
        }
}

static int handle_write_conflicts(struct drbd_device *device,
                                  struct drbd_peer_request *peer_req)
{
        struct drbd_connection *connection = peer_req->peer_device->connection;
        bool resolve_conflicts = test_bit(RESOLVE_CONFLICTS, &connection->flags);
        sector_t sector = peer_req->i.sector;
        const unsigned int size = peer_req->i.size;
        struct drbd_interval *i;
        bool equal;
        int err;

        /*
         * Inserting the peer request into the write_requests tree will prevent
         * new conflicting local requests from being added.
         */
        drbd_insert_interval(&device->write_requests, &peer_req->i);

    repeat:
        drbd_for_each_overlap(i, &device->write_requests, sector, size) {
                if (i == &peer_req->i)
                        continue;
                if (i->completed)
                        continue;

                if (!i->local) {
                        /*
                         * Our peer has sent a conflicting remote request; this
                         * should not happen in a two-node setup.  Wait for the
                         * earlier peer request to complete.
                         */
                        err = drbd_wait_misc(device, i);
                        if (err)
                                goto out;
                        goto repeat;
                }

                equal = i->sector == sector && i->size == size;
                if (resolve_conflicts) {
                        /*
                         * If the peer request is fully contained within the
                         * overlapping request, it can be considered overwritten
                         * and thus superseded; otherwise, it will be retried
                         * once all overlapping requests have completed.
                         */
                        bool superseded = i->sector <= sector && i->sector +
                                       (i->size >> 9) >= sector + (size >> 9);

                        if (!equal)
                                drbd_alert(device, "Concurrent writes detected: "
                                               "local=%llus +%u, remote=%llus +%u, "
                                               "assuming %s came first\n",
                                          (unsigned long long)i->sector, i->size,
                                          (unsigned long long)sector, size,
                                          superseded ? "local" : "remote");

                        peer_req->w.cb = superseded ? e_send_superseded :
                                                   e_send_retry_write;
                        list_add_tail(&peer_req->w.list, &device->done_ee);
                        queue_work(connection->ack_sender, &peer_req->peer_device->send_acks_work);

                        err = -ENOENT;
                        goto out;
                } else {
                        struct drbd_request *req =
                                container_of(i, struct drbd_request, i);

                        if (!equal)
                                drbd_alert(device, "Concurrent writes detected: "
                                               "local=%llus +%u, remote=%llus +%u\n",
                                          (unsigned long long)i->sector, i->size,
                                          (unsigned long long)sector, size);

                        if (req->rq_state & RQ_LOCAL_PENDING ||
                            !(req->rq_state & RQ_POSTPONED)) {
                                /*
                                 * Wait for the node with the discard flag to
                                 * decide if this request has been superseded
                                 * or needs to be retried.
                                 * Requests that have been superseded will
                                 * disappear from the write_requests tree.
                                 *
                                 * In addition, wait for the conflicting
                                 * request to finish locally before submitting
                                 * the conflicting peer request.
                                 */
                                err = drbd_wait_misc(device, &req->i);
                                if (err) {
                                        _conn_request_state(connection, NS(conn, C_TIMEOUT), CS_HARD);
                                        fail_postponed_requests(device, sector, size);
                                        goto out;
                                }
                                goto repeat;
                        }
                        /*
                         * Remember to restart the conflicting requests after
                         * the new peer request has completed.
                         */
                        peer_req->flags |= EE_RESTART_REQUESTS;
                }
        }
        err = 0;

    out:
        if (err)
                drbd_remove_epoch_entry_interval(device, peer_req);
        return err;
}

/* mirrored write */
static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
{
        struct drbd_peer_device *peer_device;
        struct drbd_device *device;
        struct net_conf *nc;
        sector_t sector;
        struct drbd_peer_request *peer_req;
        struct p_data *p = pi->data;
        u32 peer_seq = be32_to_cpu(p->seq_num);
        int op, op_flags;
        u32 dp_flags;
        int err, tp;

        peer_device = conn_peer_device(connection, pi->vnr);
        if (!peer_device)
                return -EIO;
        device = peer_device->device;

        if (!get_ldev(device)) {
                int err2;

                err = wait_for_and_update_peer_seq(peer_device, peer_seq);
                drbd_send_ack_dp(peer_device, P_NEG_ACK, p, pi->size);
                atomic_inc(&connection->current_epoch->epoch_size);
                err2 = drbd_drain_block(peer_device, pi->size);
                if (!err)
                        err = err2;
                return err;
        }

        /*
         * Corresponding put_ldev done either below (on various errors), or in
         * drbd_peer_request_endio, if we successfully submit the data at the
         * end of this function.
         */

        sector = be64_to_cpu(p->sector);
        peer_req = read_in_block(peer_device, p->block_id, sector, pi);
        if (!peer_req) {
                put_ldev(device);
                return -EIO;
        }

        peer_req->w.cb = e_end_block;
        peer_req->submit_jif = jiffies;
        peer_req->flags |= EE_APPLICATION;

        dp_flags = be32_to_cpu(p->dp_flags);
        op = wire_flags_to_bio_op(dp_flags);
        op_flags = wire_flags_to_bio_flags(dp_flags);
        if (pi->cmd == P_TRIM) {
                D_ASSERT(peer_device, peer_req->i.size > 0);
                D_ASSERT(peer_device, op == REQ_OP_DISCARD);
                D_ASSERT(peer_device, peer_req->pages == NULL);
                /* need to play safe: an older DRBD sender
                 * may mean zero-out while sending P_TRIM. */
                if (0 == (connection->agreed_features & DRBD_FF_WZEROES))
                        peer_req->flags |= EE_ZEROOUT;
        } else if (pi->cmd == P_ZEROES) {
                D_ASSERT(peer_device, peer_req->i.size > 0);
                D_ASSERT(peer_device, op == REQ_OP_WRITE_ZEROES);
                D_ASSERT(peer_device, peer_req->pages == NULL);
                /* Do (not) pass down BLKDEV_ZERO_NOUNMAP? */
                if (dp_flags & DP_DISCARD)
                        peer_req->flags |= EE_TRIM;
        } else if (peer_req->pages == NULL) {
                D_ASSERT(device, peer_req->i.size == 0);
                D_ASSERT(device, dp_flags & DP_FLUSH);
        }

        if (dp_flags & DP_MAY_SET_IN_SYNC)
                peer_req->flags |= EE_MAY_SET_IN_SYNC;

        spin_lock(&connection->epoch_lock);
        peer_req->epoch = connection->current_epoch;
        atomic_inc(&peer_req->epoch->epoch_size);
        atomic_inc(&peer_req->epoch->active);
        spin_unlock(&connection->epoch_lock);

        rcu_read_lock();
        nc = rcu_dereference(peer_device->connection->net_conf);
        tp = nc->two_primaries;
        if (peer_device->connection->agreed_pro_version < 100) {
                switch (nc->wire_protocol) {
                case DRBD_PROT_C:
                        dp_flags |= DP_SEND_WRITE_ACK;
                        break;
                case DRBD_PROT_B:
                        dp_flags |= DP_SEND_RECEIVE_ACK;
                        break;
                }
        }
        rcu_read_unlock();

        if (dp_flags & DP_SEND_WRITE_ACK) {
                peer_req->flags |= EE_SEND_WRITE_ACK;
                inc_unacked(device);
                /* corresponding dec_unacked() in e_end_block()
                 * respective _drbd_clear_done_ee */
        }

        if (dp_flags & DP_SEND_RECEIVE_ACK) {
                /* I really don't like it that the receiver thread
                 * sends on the msock, but anyways */
                drbd_send_ack(peer_device, P_RECV_ACK, peer_req);
        }

        if (tp) {
                /* two primaries implies protocol C */
                D_ASSERT(device, dp_flags & DP_SEND_WRITE_ACK);
                peer_req->flags |= EE_IN_INTERVAL_TREE;
                err = wait_for_and_update_peer_seq(peer_device, peer_seq);
                if (err)
                        goto out_interrupted;
                spin_lock_irq(&device->resource->req_lock);
                err = handle_write_conflicts(device, peer_req);
                if (err) {
                        spin_unlock_irq(&device->resource->req_lock);
                        if (err == -ENOENT) {
                                put_ldev(device);
                                return 0;
                        }
                        goto out_interrupted;
                }
        } else {
                update_peer_seq(peer_device, peer_seq);
                spin_lock_irq(&device->resource->req_lock);
        }
        /* TRIM and WRITE_SAME are processed synchronously,
         * we wait for all pending requests, respectively wait for
         * active_ee to become empty in drbd_submit_peer_request();
         * better not add ourselves here. */
        if ((peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) == 0)
                list_add_tail(&peer_req->w.list, &device->active_ee);
        spin_unlock_irq(&device->resource->req_lock);

        if (device->state.conn == C_SYNC_TARGET)
                wait_event(device->ee_wait, !overlapping_resync_write(device, peer_req));

        if (device->state.pdsk < D_INCONSISTENT) {
                /* In case we have the only disk of the cluster, */
                drbd_set_out_of_sync(device, peer_req->i.sector, peer_req->i.size);
                peer_req->flags &= ~EE_MAY_SET_IN_SYNC;
                drbd_al_begin_io(device, &peer_req->i);
                peer_req->flags |= EE_CALL_AL_COMPLETE_IO;
        }

        err = drbd_submit_peer_request(device, peer_req, op, op_flags,
                                       DRBD_FAULT_DT_WR);
        if (!err)
                return 0;

        /* don't care for the reason here */
        drbd_err(device, "submit failed, triggering re-connect\n");
        spin_lock_irq(&device->resource->req_lock);
        list_del(&peer_req->w.list);
        drbd_remove_epoch_entry_interval(device, peer_req);
        spin_unlock_irq(&device->resource->req_lock);
        if (peer_req->flags & EE_CALL_AL_COMPLETE_IO) {
                peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
                drbd_al_complete_io(device, &peer_req->i);
        }

out_interrupted:
        drbd_may_finish_epoch(connection, peer_req->epoch, EV_PUT | EV_CLEANUP);
        put_ldev(device);
        drbd_free_peer_req(device, peer_req);
        return err;
}

/* We may throttle resync, if the lower device seems to be busy,
 * and current sync rate is above c_min_rate.
 *
 * To decide whether or not the lower device is busy, we use a scheme similar
 * to MD RAID is_mddev_idle(): if the partition stats reveal "significant"
 * (more than 64 sectors) of activity we cannot account for with our own resync
 * activity, it obviously is "busy".
 *
 * The current sync rate used here uses only the most recent two step marks,
 * to have a short time average so we can react faster.
 */
bool drbd_rs_should_slow_down(struct drbd_device *device, sector_t sector,
                bool throttle_if_app_is_waiting)
{
        struct lc_element *tmp;
        bool throttle = drbd_rs_c_min_rate_throttle(device);

        if (!throttle || throttle_if_app_is_waiting)
                return throttle;

        spin_lock_irq(&device->al_lock);
        tmp = lc_find(device->resync, BM_SECT_TO_EXT(sector));
        if (tmp) {
                struct bm_extent *bm_ext = lc_entry(tmp, struct bm_extent, lce);
                if (test_bit(BME_PRIORITY, &bm_ext->flags))
                        throttle = false;
                /* Do not slow down if app IO is already waiting for this extent,
                 * and our progress is necessary for application IO to complete. */
        }
        spin_unlock_irq(&device->al_lock);

        return throttle;
}

bool drbd_rs_c_min_rate_throttle(struct drbd_device *device)
{
        struct gendisk *disk = device->ldev->backing_bdev->bd_contains->bd_disk;
        unsigned long db, dt, dbdt;
        unsigned int c_min_rate;
        int curr_events;

        rcu_read_lock();
        c_min_rate = rcu_dereference(device->ldev->disk_conf)->c_min_rate;
        rcu_read_unlock();

        /* feature disabled? */
        if (c_min_rate == 0)
                return false;

        curr_events = (int)part_stat_read_accum(&disk->part0, sectors) -
                        atomic_read(&device->rs_sect_ev);

        if (atomic_read(&device->ap_actlog_cnt)
            || curr_events - device->rs_last_events > 64) {
                unsigned long rs_left;
                int i;

                device->rs_last_events = curr_events;

                /* sync speed average over the last 2*DRBD_SYNC_MARK_STEP,
                 * approx. */
                i = (device->rs_last_mark + DRBD_SYNC_MARKS-1) % DRBD_SYNC_MARKS;

                if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
                        rs_left = device->ov_left;
                else
                        rs_left = drbd_bm_total_weight(device) - device->rs_failed;

                dt = ((long)jiffies - (long)device->rs_mark_time[i]) / HZ;
                if (!dt)
                        dt++;
                db = device->rs_mark_left[i] - rs_left;
                dbdt = Bit2KB(db/dt);

                if (dbdt > c_min_rate)
                        return true;
        }
        return false;
}

static int receive_DataRequest(struct drbd_connection *connection, struct packet_info *pi)
{
        struct drbd_peer_device *peer_device;
        struct drbd_device *device;
        sector_t sector;
        sector_t capacity;
        struct drbd_peer_request *peer_req;
        struct digest_info *di = NULL;
        int size, verb;
        unsigned int fault_type;
        struct p_block_req *p = pi->data;

        peer_device = conn_peer_device(connection, pi->vnr);
        if (!peer_device)
                return -EIO;
        device = peer_device->device;
        capacity = drbd_get_capacity(device->this_bdev);

        sector = be64_to_cpu(p->sector);
        size   = be32_to_cpu(p->blksize);

        if (size <= 0 || !IS_ALIGNED(size, 512) || size > DRBD_MAX_BIO_SIZE) {
                drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
                                (unsigned long long)sector, size);
                return -EINVAL;
        }
        if (sector + (size>>9) > capacity) {
                drbd_err(device, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__,
                                (unsigned long long)sector, size);
                return -EINVAL;
        }

        if (!get_ldev_if_state(device, D_UP_TO_DATE)) {
                verb = 1;
                switch (pi->cmd) {
                case P_DATA_REQUEST:
                        drbd_send_ack_rp(peer_device, P_NEG_DREPLY, p);
                        break;
                case P_RS_THIN_REQ:
                case P_RS_DATA_REQUEST:
                case P_CSUM_RS_REQUEST:
                case P_OV_REQUEST:
                        drbd_send_ack_rp(peer_device, P_NEG_RS_DREPLY , p);
                        break;
                case P_OV_REPLY:
                        verb = 0;
                        dec_rs_pending(device);
                        drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size, ID_IN_SYNC);
                        break;
                default:
                        BUG();
                }
                if (verb && __ratelimit(&drbd_ratelimit_state))
                        drbd_err(device, "Can not satisfy peer's read request, "
                            "no local data.\n");

                /* drain possibly payload */
                return drbd_drain_block(peer_device, pi->size);
        }

        /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD
         * "criss-cross" setup, that might cause write-out on some other DRBD,
         * which in turn might block on the other node at this very place.  */
        peer_req = drbd_alloc_peer_req(peer_device, p->block_id, sector, size,
                        size, GFP_NOIO);
        if (!peer_req) {
                put_ldev(device);
                return -ENOMEM;
        }

        switch (pi->cmd) {
        case P_DATA_REQUEST:
                peer_req->w.cb = w_e_end_data_req;
                fault_type = DRBD_FAULT_DT_RD;
                /* application IO, don't drbd_rs_begin_io */
                peer_req->flags |= EE_APPLICATION;
                goto submit;

        case P_RS_THIN_REQ:
                /* If at some point in the future we have a smart way to
                   find out if this data block is completely deallocated,
                   then we would do something smarter here than reading
                   the block... */
                peer_req->flags |= EE_RS_THIN_REQ;
                fallthrough;
        case P_RS_DATA_REQUEST:
                peer_req->w.cb = w_e_end_rsdata_req;
                fault_type = DRBD_FAULT_RS_RD;
                /* used in the sector offset progress display */
                device->bm_resync_fo = BM_SECT_TO_BIT(sector);
                break;

        case P_OV_REPLY:
        case P_CSUM_RS_REQUEST:
                fault_type = DRBD_FAULT_RS_RD;
                di = kmalloc(sizeof(*di) + pi->size, GFP_NOIO);
                if (!di)
                        goto out_free_e;

                di->digest_size = pi->size;
                di->digest = (((char *)di)+sizeof(struct digest_info));

                peer_req->digest = di;
                peer_req->flags |= EE_HAS_DIGEST;

                if (drbd_recv_all(peer_device->connection, di->digest, pi->size))
                        goto out_free_e;

                if (pi->cmd == P_CSUM_RS_REQUEST) {
                        D_ASSERT(device, peer_device->connection->agreed_pro_version >= 89);
                        peer_req->w.cb = w_e_end_csum_rs_req;
                        /* used in the sector offset progress display */
                        device->bm_resync_fo = BM_SECT_TO_BIT(sector);
                        /* remember to report stats in drbd_resync_finished */
                        device->use_csums = true;
                } else if (pi->cmd == P_OV_REPLY) {
                        /* track progress, we may need to throttle */
                        atomic_add(size >> 9, &device->rs_sect_in);
                        peer_req->w.cb = w_e_end_ov_reply;
                        dec_rs_pending(device);
                        /* drbd_rs_begin_io done when we sent this request,
                         * but accounting still needs to be done. */
                        goto submit_for_resync;
                }
                break;

        case P_OV_REQUEST:
                if (device->ov_start_sector == ~(sector_t)0 &&
                    peer_device->connection->agreed_pro_version >= 90) {
                        unsigned long now = jiffies;
                        int i;
                        device->ov_start_sector = sector;
                        device->ov_position = sector;
                        device->ov_left = drbd_bm_bits(device) - BM_SECT_TO_BIT(sector);
                        device->rs_total = device->ov_left;
                        for (i = 0; i < DRBD_SYNC_MARKS; i++) {
                                device->rs_mark_left[i] = device->ov_left;
                                device->rs_mark_time[i] = now;
                        }
                        drbd_info(device, "Online Verify start sector: %llu\n",
                                        (unsigned long long)sector);
                }
                peer_req->w.cb = w_e_end_ov_req;
                fault_type = DRBD_FAULT_RS_RD;
                break;

        default:
                BUG();
        }

        /* Throttle, drbd_rs_begin_io and submit should become asynchronous
         * wrt the receiver, but it is not as straightforward as it may seem.
         * Various places in the resync start and stop logic assume resync
         * requests are processed in order, requeuing this on the worker thread
         * introduces a bunch of new code for synchronization between threads.
         *
         * Unlimited throttling before drbd_rs_begin_io may stall the resync
         * "forever", throttling after drbd_rs_begin_io will lock that extent
         * for application writes for the same time.  For now, just throttle
         * here, where the rest of the code expects the receiver to sleep for
         * a while, anyways.
         */

        /* Throttle before drbd_rs_begin_io, as that locks out application IO;
         * this defers syncer requests for some time, before letting at least