Quelle  drbd_receiver.c   Sprache: C

// SPDX-License-Identifier: GPL-2.0-only
/*
   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>
#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 <linux/mempool.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)

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

 /* 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 = mempool_alloc(&drbd_buffer_page_pool, GFP_TRY);
  if (!tmp)
   goto fail;
  set_page_private(tmp, (unsigned long)page);
  page = tmp;
 }
 return page;
fail:
 page_chain_for_each_safe(page, tmp) {
  set_page_private(page, 0);
  mempool_free(page, &drbd_buffer_page_pool);
 }
 return NULL;
}

/**
 * drbd_alloc_pages() - Returns @number pages, retries forever (or until signalled)
 * @peer_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;
 struct net_conf *nc;
 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)
  schedule_timeout_interruptible(HZ / 10);
 page = __drbd_alloc_pages(number);

 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)
{
 struct page *tmp;
 int i = 0;

 if (page == NULL)
  return;

 page_chain_for_each_safe(page, tmp) {
  set_page_private(page, 0);
  if (page_count(page) == 1)
   mempool_free(page, &drbd_buffer_page_pool);
  else
   put_page(page);
  i++;
 }
 i = atomic_sub_return(i, &device->pp_in_use);
 if (i < 0)
  drbd_warn(device, "ASSERTION FAILED: pp_in_use: %d < 0\n", i);
}

/*
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 int nr_pages = PFN_UP(payload_size);

 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;
  if (!mempool_is_saturated(&drbd_buffer_page_pool))
   peer_req->flags |= EE_RELEASE_TO_MEMPOOL;
 }

 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)
{
 might_sleep();
 if (peer_req->flags & EE_HAS_DIGEST)
  kfree(peer_req->digest);
 drbd_free_pages(device, peer_req->pages);
 D_ASSERT(device, atomic_read(&peer_req->pending_bios) == 0);
 D_ASSERT(device, drbd_interval_empty(&peer_req->i));
 if (!expect(device, !(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;

 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);
  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);
 struct drbd_peer_request *peer_req, *t;
 int err = 0;

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

 /* 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, ITER_DEST, &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 += get_random_u32_below(2) ? 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 (get_random_u32_below(2))
     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_use_task_frag = false;
 msock.socket->sk->sk_use_task_frag = false;

 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(device->ldev->backing_bdev, 0,
        REQ_OP_WRITE | REQ_PREFLUSH, GFP_NOIO);
 struct one_flush_context *octx = kmalloc(sizeof(*octx), GFP_NOIO);

 if (!octx) {
  drbd_warn(device, "Could not allocate a octx, 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.
 */
  bio_put(bio);

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

 octx->device = device;
 octx->ctx = ctx;
 bio->bi_private = octx;
 bio->bi_end_io = one_flush_endio;

 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.
 * @connection: DRBD connection.
 * @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
 * @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;
 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(bdev_discard_granularity(bdev) >> 9, 1U);
 alignment = (bdev_discard_alignment(bdev) >> 9) % granularity;

 max_discard_sectors = min(bdev_max_discard_sectors(bdev), (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);
  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);
   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 disk_conf *dc;
 bool can_do;

 if (!bdev_max_discard_sectors(device->ldev->backing_bdev))
  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 int peer_request_fault_type(struct drbd_peer_request *peer_req)
{
 if (peer_req_op(peer_req) == REQ_OP_READ) {
  return peer_req->flags & EE_APPLICATION ?
   DRBD_FAULT_DT_RD : DRBD_FAULT_RS_RD;
 } else {
  return peer_req->flags & EE_APPLICATION ?
   DRBD_FAULT_DT_WR : DRBD_FAULT_RS_WR;
 }
}

/**
 * drbd_submit_peer_request()
 * @peer_req: peer request
 *
 * 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_peer_request *peer_req)
{
 struct drbd_device *device = peer_req->peer_device->device;
 struct bio *bios = NULL;
 struct bio *bio;
 struct page *page = peer_req->pages;
 sector_t sector = peer_req->i.sector;
 unsigned int data_size = peer_req->i.size;
 unsigned int n_bios = 0;
 unsigned int nr_pages = PFN_UP(data_size);

 /* 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_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);
  }

  drbd_issue_peer_discard_or_zero_out(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:
 /* _DISCARD, _WRITE_ZEROES handled above.
 * REQ_OP_FLUSH (empty flush) not expected,
 * should have been mapped to a "drbd protocol barrier".
 * REQ_OP_SECURE_ERASE: I don't see how we could ever support that.
 */
 if (!(peer_req_op(peer_req) == REQ_OP_WRITE ||
    peer_req_op(peer_req) == REQ_OP_READ)) {
  drbd_err(device, "Invalid bio op received: 0x%x\n", peer_req->opf);
  return -EINVAL;
 }

 bio = bio_alloc(device->ldev->backing_bdev, nr_pages, peer_req->opf, GFP_NOIO);
 /* > peer_req->i.sector, unless this is the first bio */
 bio->bi_iter.bi_sector = sector;
 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, peer_request_fault_type(peer_req), bio);
 } while (bios);
 return 0;
}

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 = get_capacity(device->vdisk);
 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;

 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. */
 ds = data_size;
 if (trim) {
  if (!expect(peer_device, data_size == 0))
   return NULL;
  ds = be32_to_cpu(trim->size);
 } else if (zeroes) {
  if (!expect(peer_device, data_size == 0))
   return NULL;
  ds = be32_to_cpu(zeroes->size);
 }

 if (!expect(peer_device, IS_ALIGNED(ds, 512)))
  return NULL;
 if (trim || zeroes) {
  if (!expect(peer_device, ds <= (DRBD_MAX_BBIO_SECTORS << 9)))
   return NULL;
 } else if (!expect(peer_device, 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;
 }

 /* 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);
 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 = bvec_kmap_local(&bvec);
  expect = min_t(int, data_size, bvec.bv_len);
  err = drbd_recv_all_warn(peer_device->connection, mapped, expect);
  kunmap_local(mapped);
  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(peer_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(peer_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(peer_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->opf = REQ_OP_WRITE;
 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(peer_req) == 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;

 err = recv_dless_read(peer_device, req, sector, pi->size);
 if (!err)
  req_mod(req, DATA_RECEIVED, peer_device);
 /* 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 (drbd_ratelimit())
   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, 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(peer_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;
}

static enum req_op wire_flags_to_bio_op(u32 dpf)
{
 if (dpf & DP_ZEROES)
  return REQ_OP_WRITE_ZEROES;
 if (dpf & DP_DISCARD)
  return REQ_OP_DISCARD;
 else
  return REQ_OP_WRITE;
}

/* see also bio_flags_to_wire() */
static blk_opf_t wire_flags_to_bio(struct drbd_connection *connection, u32 dpf)
{
 return wire_flags_to_bio_op(dpf) |
  (dpf & DP_RW_SYNC ? REQ_SYNC : 0) |
  (dpf & DP_FUA ? REQ_FUA : 0) |
  (dpf & DP_FLUSH ? REQ_PREFLUSH : 0);
}

static void fail_postponed_requests(struct drbd_device *device, sector_t sector,
        unsigned int size)
{
 struct drbd_peer_device *peer_device = first_peer_device(device);
 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, peer_device, &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;
  }

--> --------------------

--> maximum size reached

--> --------------------