NOTE: modified to patch drbd_headers directly, instead of submodule

From 3d98754edb385f5a45b5df42f436b99b785b705d Mon Sep 17 00:00:00 2001
From: Lars Ellenberg <lars.ellenberg@linbit.com>
Date: Thu, 19 Jul 2018 15:07:52 +0200
Subject: [PATCH] drbd: protocol 114: fix distributed deadlock on secondary
 activity log

If a Primary node, sends more writes to scattered LBA offsets than can
possibly fit into the activity log on the Secondary node, either because
it has a much larger al-extents setting than its peer, or because it is
a diskless client node and simply has no activity log itself,
the receiver may block indefinetely when receiving the next epoch
closing P_BARRIER.

The receiver has to wait for all "pending IO" before sending the
P_BARRIER_ACK, pending IO is waiting for the activity log to make
progress, activity log extents are blocked by references of older
peer requests which wait for confirmation (via P_PEER_ACK) that
they can be put, P_PEER_ACK can only be sent for requests which
had been P_BARRIER_ACKed by all peers first. Distributed deadlock.

The workaround to avoid the situation was
to set max-epoch-size <= al-extents.

This deadlock is now fixed by detecting a possible in-epoch activity log
congestion on the receiving side, and in that case do the equivalent of
what a receive_Barrier() whould have done, which is wait for all pending
IO to drain and submit cache flushes, unless disabled, then send a
new P_CONFIRM_STABLE command via the control socket to the Primary
can then send out the P_PEER_ACK(s) necessary to make progress on the
activity log.
---
 drbd-headers             |   2 +-
 drbd/drbd_int.h          |   9 +-
 drbd/drbd_main.c         |  79 +++++++++++++---
 drbd/drbd_receiver.c     | 188 ++++++++++++++++++++++++++++++++-------
 drbd/drbd_req.c          |  14 +--
 drbd/linux/drbd_config.h |   2 +-
 6 files changed, 237 insertions(+), 57 deletions(-)

#diff --git a/drbd-headers b/drbd-headers
#index 165a5c71..6fc42ada 160000
#--- a/drbd-headers
#+++ b/drbd-headers
#@@ -1 +1 @@
#-Subproject commit 165a5c717bbe24ef1a050305f3b4b229cabdf398
#+Subproject commit 6fc42ada4b6ed4eb5c40c89b09bc8115d5e62476
diff --git a/drbd/drbd_int.h b/drbd/drbd_int.h
index a4244f96..0683be24 100644
--- a/drbd/drbd_int.h
+++ b/drbd/drbd_int.h
@@ -382,10 +382,12 @@ struct drbd_request {
 
 struct drbd_epoch {
 	struct drbd_connection *connection;
+	struct drbd_peer_request *oldest_unconfirmed_peer_req;
 	struct list_head list;
 	unsigned int barrier_nr;
 	atomic_t epoch_size; /* increased on every request added. */
 	atomic_t active;     /* increased on every req. added, and dec on every finished. */
+	atomic_t confirmed;  /* adjusted for every P_CONFIRM_STABLE */
 	unsigned long flags;
 };
 
@@ -1465,8 +1467,11 @@ extern void drbd_thread_current_set_cpu(struct drbd_thread *thi);
 #else
 #define drbd_thread_current_set_cpu(A) ({})
 #endif
-extern void tl_release(struct drbd_connection *, unsigned int barrier_nr,
-		       unsigned int set_size);
+extern void tl_release(struct drbd_connection *,
+			uint64_t o_block_id,
+			uint64_t y_block_id,
+			unsigned int barrier_nr,
+			unsigned int set_size);
 extern void drbd_free_sock(struct drbd_connection *connection);
 
 extern int __drbd_send_protocol(struct drbd_connection *connection, enum drbd_packet cmd);
diff --git a/drbd/drbd_main.c b/drbd/drbd_main.c
index 718b4dd9..18199daa 100644
--- a/drbd/drbd_main.c
+++ b/drbd/drbd_main.c
@@ -59,7 +59,7 @@
 #include <linux/drbd_limits.h>
 #include "drbd_int.h"
 #include "drbd_protocol.h"
-#include "drbd_req.h" /* only for _req_mod in tl_release */
+#include "drbd_req.h"
 #include "drbd_vli.h"
 #include "drbd_debugfs.h"
 #include "drbd_meta_data.h"
@@ -353,31 +353,47 @@ static void tl_abort_for_each_req_ref(struct drbd_request *next, struct list_hea
 /**
  * tl_release() - mark as BARRIER_ACKED all requests in the corresponding transfer log epoch
  * @device:	DRBD device.
+ * @o_block_id: "block id" aka expected pointer address of the oldest request
+ * @y_block_id: "block id" aka expected pointer address of the youngest request
+ *		confirmed to be on stable storage.
  * @barrier_nr:	Expected identifier of the DRBD write barrier packet.
- * @set_size:	Expected number of requests before that barrier.
+ * @set_size:	Expected number of requests before that barrier, respectively
+ *		number of requests in the interval [o_block_id;y_block_id]
+ *
+ * Called for both P_BARRIER_ACK and P_CONFIRM_STABLE,
+ * which is similar to an unsolicited partial barrier ack.
+ *
+ * Either barrier_nr (for barrier acks) or both o_block_id and y_blockid (for
+ * confirm stable) are given.  For barrier acks, all requests in the epoch
+ * designated by "barrier_nr" are confirmed to be on stable storage.
+ *
+ * For confirm stable, both o_block_id and y_block_id are given, barrier_nr is
+ * ignored, and all requests from "o_block_id" up to and including y_block_id
+ * are confirmed to be on stable storage on the reporting peer.
  *
  * In case the passed barrier_nr or set_size does not match the oldest
  * epoch of not yet barrier-acked requests, this function will cause a
  * termination of the connection.
  */
-void tl_release(struct drbd_connection *connection, unsigned int barrier_nr,
+void tl_release(struct drbd_connection *connection,
+		uint64_t o_block_id,
+		uint64_t y_block_id,
+		unsigned int barrier_nr,
 		unsigned int set_size)
 {
 	struct drbd_resource *resource = connection->resource;
 	struct drbd_request *r;
 	struct drbd_request *req = NULL;
+	struct drbd_request *req_y = NULL;
 	int expect_epoch = 0;
 	int expect_size = 0;
+	const int idx = connection->peer_node_id;
 
-	spin_lock_irq(&connection->resource->req_lock);
+	spin_lock_irq(&resource->req_lock);
 
 	/* find oldest not yet barrier-acked write request,
 	 * count writes in its epoch. */
 	list_for_each_entry(r, &resource->transfer_log, tl_requests) {
-		struct drbd_peer_device *peer_device =
-			conn_peer_device(connection, r->device->vnr);
-		const int idx = peer_device->node_id;
-
 		if (!req) {
 			if (!(r->local_rq_state & RQ_WRITE))
 				continue;
@@ -397,9 +413,35 @@ void tl_release(struct drbd_connection *connection, unsigned int barrier_nr,
 			/* if (!(s & RQ_NET_MASK)): not expected */
 			expect_size++;
 		}
+		if (y_block_id && (struct drbd_request*)(unsigned long)y_block_id == r) {
+			req_y = r;
+			break;
+		}
 	}
 
 	/* first some paranoia code */
+	if (o_block_id) {
+		if ((struct drbd_request*)(unsigned long)o_block_id != req) {
+			drbd_err(connection, "BAD! ConfirmedStable: expected %p, found %p\n",
+				(struct drbd_request*)(unsigned long)o_block_id, req);
+			goto bail;
+		}
+		if (!req_y) {
+			drbd_err(connection, "BAD! ConfirmedStable: expected youngest request %p NOT found\n",
+				(struct drbd_req*)(unsigned long)y_block_id);
+			goto bail;
+		}
+		/* A P_CONFIRM_STABLE cannot tell me the to-be-expected barrier nr,
+		 * it does not know it yet. But we just confirmed it knew the
+		 * expected request, so just use that one. */
+		barrier_nr = expect_epoch;
+		/* Both requests referenced must be in the same epoch. */
+		if (req_y->epoch != expect_epoch) {
+			drbd_err(connection, "BAD! ConfirmedStable: reported requests not in the same epoch (%u != %u)\n",
+				req->epoch, req_y->epoch);
+			goto bail;
+		}
+	}
 	if (req == NULL) {
 		drbd_err(connection, "BAD! BarrierAck #%u received, but no epoch in tl!?\n",
 			 barrier_nr);
@@ -412,8 +454,12 @@ void tl_release(struct drbd_connection *connection, unsigned int barrier_nr,
 	}
 
 	if (expect_size != set_size) {
-		drbd_err(connection, "BAD! BarrierAck #%u received with n_writes=%u, expected n_writes=%u!\n",
-			 barrier_nr, set_size, expect_size);
+		if (!o_block_id)
+			drbd_err(connection, "BAD! BarrierAck #%u received with n_writes=%u, expected n_writes=%u!\n",
+				 barrier_nr, set_size, expect_size);
+		else
+			drbd_err(connection, "BAD! ConfirmedStable [%p,%p] received with n_writes=%u, expected n_writes=%u!\n",
+				 req, req_y, set_size, expect_size);
 		goto bail;
 	}
 
@@ -432,10 +478,17 @@ void tl_release(struct drbd_connection *connection, unsigned int barrier_nr,
 		}
 		peer_device = conn_peer_device(connection, req->device->vnr);
 		_req_mod(req, BARRIER_ACKED, peer_device);
+		if (req == req_y) {
+			tl_abort_for_each_req_ref(r, &resource->transfer_log);
+			break;
+		}
 	}
-	spin_unlock_irq(&connection->resource->req_lock);
+	spin_unlock_irq(&resource->req_lock);
 
-	if (barrier_nr == connection->send.last_sent_epoch_nr) {
+	/* urgently flush out peer acks for P_CONFIRM_STABLE */
+	if (req_y) {
+		drbd_flush_peer_acks(resource);
+	} else if (barrier_nr == connection->send.last_sent_epoch_nr) {
 		clear_bit(BARRIER_ACK_PENDING, &connection->flags);
 		wake_up(&resource->barrier_wait);
 	}
@@ -443,7 +496,7 @@ void tl_release(struct drbd_connection *connection, unsigned int barrier_nr,
 	return;
 
 bail:
-	spin_unlock_irq(&connection->resource->req_lock);
+	spin_unlock_irq(&resource->req_lock);
 	change_cstate(connection, C_PROTOCOL_ERROR, CS_HARD);
 }
 
diff --git a/drbd/drbd_receiver.c b/drbd/drbd_receiver.c
index bae8cdb6..a9e22d5a 100644
--- a/drbd/drbd_receiver.c
+++ b/drbd/drbd_receiver.c
@@ -1050,6 +1050,10 @@ static enum finish_epoch drbd_flush_after_epoch(struct drbd_connection *connecti
 		}
 	}
 
+	/* If called before sending P_CONFIRM_STABLE, we don't have the epoch
+	 * (and must not finish it yet, anyways) */
+	if (epoch == NULL)
+		return FE_STILL_LIVE;
 	return drbd_may_finish_epoch(connection, epoch, EV_BARRIER_DONE);
 }
 
@@ -1085,6 +1089,52 @@ static void drbd_send_b_ack(struct drbd_connection *connection, u32 barrier_nr,
 	send_command(connection, -1, P_BARRIER_ACK, CONTROL_STREAM);
 }
 
+static void drbd_send_confirm_stable(struct drbd_peer_request *peer_req)
+{
+	struct drbd_connection *connection = peer_req->peer_device->connection;
+	struct drbd_resource *resource = connection->resource;
+	struct drbd_epoch *epoch = peer_req->epoch;
+	struct drbd_peer_request *oldest, *youngest;
+	struct p_confirm_stable *p;
+	int count;
+
+	if (connection->cstate[NOW] < C_CONNECTED)
+		return;
+
+	/* peer_req is not on stable storage yet, but the only one in this epoch.
+	 * Nothing to confirm, just wait for the normal barrier_ack and peer_ack
+	 * to do their work. */
+	oldest = epoch->oldest_unconfirmed_peer_req;
+	if (oldest == peer_req)
+		return;
+
+	p = conn_prepare_command(connection, sizeof(*p), CONTROL_STREAM);
+	if (!p)
+		return;
+
+	/* receive_Data() does a list_add_tail() for every requests, which
+	 * means the oldest is .next, the currently blocked one that triggered
+	 * this code path is .prev, and the youngest that now should be on
+	 * stable storage is .prev->prev */
+	spin_lock_irq(&resource->req_lock);
+	youngest = list_entry(peer_req->recv_order.prev, struct drbd_peer_request, recv_order);
+	spin_unlock_irq(&resource->req_lock);
+
+	count = atomic_read(&epoch->epoch_size) - atomic_read(&epoch->confirmed) - 1;
+	atomic_add(count, &epoch->confirmed);
+	epoch->oldest_unconfirmed_peer_req = peer_req;
+
+	D_ASSERT(connection, oldest->epoch == youngest->epoch);
+	D_ASSERT(connection, count > 0);
+
+	p->oldest_block_id = oldest->block_id;
+	p->youngest_block_id = youngest->block_id;
+	p->set_size = cpu_to_be32(count);
+	p->pad = 0;
+
+	send_command(connection, -1, P_CONFIRM_STABLE, CONTROL_STREAM);
+}
+
 /**
  * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it.
  * @connection:	DRBD connection.
@@ -1150,6 +1200,8 @@ static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connectio
 		}
 		if (finish) {
 			if (!(ev & EV_CLEANUP)) {
+				/* adjust for nr requests already confirmed via P_CONFIRM_STABLE, if any. */
+				epoch_size -= atomic_read(&epoch->confirmed);
 				spin_unlock(&connection->epoch_lock);
 				drbd_send_b_ack(epoch->connection, epoch->barrier_nr, epoch_size);
 				spin_lock(&connection->epoch_lock);
@@ -1171,8 +1223,10 @@ static enum finish_epoch drbd_may_finish_epoch(struct drbd_connection *connectio
 				if (rv == FE_STILL_LIVE)
 					rv = FE_DESTROYED;
 			} else {
+				epoch->oldest_unconfirmed_peer_req = NULL;
 				epoch->flags = 0;
 				atomic_set(&epoch->epoch_size, 0);
+				atomic_set(&epoch->confirmed, 0);
 				/* atomic_set(&epoch->active, 0); is alrady zero */
 				if (rv == FE_STILL_LIVE)
 					rv = FE_RECYCLED;
@@ -1686,9 +1740,11 @@ int w_e_reissue(struct drbd_work *w, int cancel) __releases(local)
 	}
 }
 
-static void conn_wait_done_ee_empty(struct drbd_connection *connection)
+static void conn_wait_done_ee_empty_or_disconnect(struct drbd_connection *connection)
 {
-	wait_event(connection->ee_wait, atomic_read(&connection->done_ee_cnt) == 0);
+	wait_event(connection->ee_wait,
+		atomic_read(&connection->done_ee_cnt) == 0
+		|| connection->cstate[NOW] < C_CONNECTED);
 }
 
 static void conn_wait_active_ee_empty_or_disconnect(struct drbd_connection *connection)
@@ -1750,7 +1806,7 @@ static int receive_Barrier(struct drbd_connection *connection, struct packet_inf
 
 	/* receiver context, in the writeout path of the other node.
 	 * avoid potential distributed deadlock */
-	epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO);
+	epoch = kzalloc(sizeof(struct drbd_epoch), GFP_NOIO);
 	if (!epoch) {
 		drbd_warn(connection, "Allocation of an epoch failed, slowing down\n");
 		issue_flush = !test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &connection->current_epoch->flags);
@@ -1761,15 +1817,11 @@ static int receive_Barrier(struct drbd_connection *connection, struct packet_inf
 				return 0;
 		}
 
-		conn_wait_done_ee_empty(connection);
+		conn_wait_done_ee_empty_or_disconnect(connection);
 
 		return 0;
 	}
 
-	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);
@@ -2589,7 +2641,6 @@ static int handle_write_conflicts(struct drbd_peer_request *peer_req)
 
 static void drbd_queue_peer_request(struct drbd_device *device, struct drbd_peer_request *peer_req)
 {
-	atomic_add(interval_to_al_extents(&peer_req->i), &device->wait_for_actlog_ecnt);
 	atomic_inc(&device->wait_for_actlog);
 	spin_lock_irq(&device->resource->req_lock);
 	list_add_tail(&peer_req->wait_for_actlog, &device->submit.peer_writes);
@@ -2599,6 +2650,77 @@ static void drbd_queue_peer_request(struct drbd_device *device, struct drbd_peer
 	wake_up(&device->al_wait);
 }
 
+/* FIXME
+ * TODO grab the device->al_lock *once*, and check:
+ *     if possible, non-blocking get the reference(s),
+ *     if transaction is required, queue them up,
+ *        AND account for the queued up worst-case slot consumption
+ *     if available slots, corrected by other accounting, suggest
+ *        that we might block on this now or later,
+ *        *FIRST* drain, then flush, then send P_CONFIRM_STABLE,
+ *        then wait for available slots to be sufficient.
+ */
+static enum { DRBD_PAL_QUEUE, DRBD_PAL_DISCONNECTED, DRBD_PAL_SUBMIT }
+prepare_activity_log(struct drbd_peer_request *peer_req)
+{
+	struct drbd_peer_device *peer_device = peer_req->peer_device;
+	struct drbd_connection *connection = peer_device->connection;
+	struct drbd_device *device = peer_device->device;
+
+	struct lru_cache *al;
+	int nr_al_extents = interval_to_al_extents(&peer_req->i);
+	int nr, used, ecnt;
+	int ret = DRBD_PAL_SUBMIT;
+
+	/* In protocol < 110 (which is compat mode 8.4 <-> 9.0),
+	 * we must not block in the activity log here, that would
+	 * deadlock during an ongoing resync with the drbd_rs_begin_io
+	 * we did when receiving the resync request.
+	 *
+	 * We still need to update the activity log, if ours is the
+	 * only remaining disk, in which case there cannot be a resync,
+	 * and the deadlock paths cannot be taken.
+	 */
+	if (connection->agreed_pro_version < 110 &&
+	    peer_device->disk_state[NOW] >= D_INCONSISTENT)
+		return DRBD_PAL_SUBMIT;
+
+	/* Let the activity log know we are about to use it.
+	 * See also drbd_request_prepare() for the "request" entry point. */
+	ecnt = atomic_add_return(nr_al_extents, &device->wait_for_actlog_ecnt);
+
+	spin_lock(&device->al_lock);
+	al = device->act_log;
+	nr = al->nr_elements;
+	used = al->used;
+	spin_unlock(&device->al_lock);
+
+	/* note: due to the slight delay between being accounted in "used" after
+	 * being committed to the activity log with drbd_al_begin_io_commit(),
+	 * and being subtracted from "wait_for_actlog_ecnt" in __drbd_submit_peer_request(),
+	 * this can err, but only on the conservative side (overestimating ecnt). */
+	if (ecnt > nr - used) {
+		conn_wait_active_ee_empty_or_disconnect(connection);
+		drbd_flush_after_epoch(connection, NULL);
+		conn_wait_done_ee_empty_or_disconnect(connection);
+
+		/* would this peer even understand me? */
+		if (connection->agreed_pro_version >= 114)
+			drbd_send_confirm_stable(peer_req);
+
+		if  (drbd_al_begin_io_for_peer(peer_device, &peer_req->i))
+			ret = DRBD_PAL_DISCONNECTED;
+	} else if (nr_al_extents != 1 || !drbd_al_begin_io_fastpath(device, &peer_req->i)) {
+		ret = DRBD_PAL_QUEUE;
+	}
+	if (ret == DRBD_PAL_SUBMIT)
+		peer_req->flags |= EE_IN_ACTLOG;
+	if (ret != DRBD_PAL_QUEUE)
+		atomic_sub(nr_al_extents, &device->wait_for_actlog_ecnt);
+
+	return ret;
+}
+
 /* mirrored write */
 static int receive_Data(struct drbd_connection *connection, struct packet_info *pi)
 {
@@ -2712,6 +2834,8 @@ static int receive_Data(struct drbd_connection *connection, struct packet_info *
 	peer_req->epoch = connection->current_epoch;
 	atomic_inc(&peer_req->epoch->epoch_size);
 	atomic_inc(&peer_req->epoch->active);
+	if (peer_req->epoch->oldest_unconfirmed_peer_req == NULL)
+		peer_req->epoch->oldest_unconfirmed_peer_req = peer_req;
 
 	if (connection->resource->write_ordering == WO_BIO_BARRIER &&
 	    atomic_read(&peer_req->epoch->epoch_size) == 1) {
@@ -2785,6 +2909,8 @@ static int receive_Data(struct drbd_connection *connection, struct packet_info *
 		spin_lock_irq(&device->resource->req_lock);
 	}
 	/* Added to list here already, so debugfs can find it.
+	 * NOTE: active_ee_cnt is only increased *after* we checked we won't
+	 * need to wait for current activity to drain in prepare_activity_log()
 	 */
 	list_add_tail(&peer_req->w.list, &connection->active_ee);
 	if (connection->agreed_pro_version >= 110)
@@ -2805,34 +2931,18 @@ static int receive_Data(struct drbd_connection *connection, struct packet_info *
 	 * Remember the op_flags. */
 	peer_req->op_flags = op_flags;
 
+	err = prepare_activity_log(peer_req);
+	if (err == DRBD_PAL_DISCONNECTED)
+		goto disconnect_during_al_begin_io;
+
 	atomic_inc(&connection->active_ee_cnt);
 
-	/* In protocol < 110 (which is compat mode 8.4 <-> 9.0),
-	 * we must not block in the activity log here, that would
-	 * deadlock during an ongoing resync with the drbd_rs_begin_io
-	 * we did when receiving the resync request.
-	 *
-	 * We still need to update the activity log, if ours is the
-	 * only remaining disk, in which case there cannot be a resync,
-	 * and the deadlock paths cannot be taken.
-	 */
-	if (connection->agreed_pro_version >= 110 ||
-	    peer_device->disk_state[NOW] < D_INCONSISTENT) {
-		/* For now, it is easier to still handle some "special" requests
-		 * "synchronously" from receiver context */
-		if (peer_req->flags & (EE_TRIM|EE_ZEROOUT|EE_WRITE_SAME|EE_IS_BARRIER)) {
-			err = drbd_al_begin_io_for_peer(peer_device, &peer_req->i);
-			if (err)
-				goto disconnect_during_al_begin_io;
-		} else if (!drbd_al_begin_io_fastpath(device, &peer_req->i)) {
-			drbd_queue_peer_request(device, peer_req);
-			return 0;
-		}
-		peer_req->flags |= EE_IN_ACTLOG;
+	if (err == DRBD_PAL_QUEUE) {
+		drbd_queue_peer_request(device, peer_req);
+		return 0;
 	}
 
-	err = drbd_submit_peer_request(device, peer_req, op, op_flags,
-				       DRBD_FAULT_DT_WR);
+	err = drbd_submit_peer_request(device, peer_req, op, op_flags, DRBD_FAULT_DT_WR);
 	if (!err)
 		return 0;
 
@@ -8403,7 +8513,16 @@ static int got_BarrierAck(struct drbd_connection *connection, struct packet_info
 {
 	struct p_barrier_ack *p = pi->data;
 
-	tl_release(connection, p->barrier, be32_to_cpu(p->set_size));
+	tl_release(connection, 0, 0, p->barrier, be32_to_cpu(p->set_size));
+
+	return 0;
+}
+
+static int got_confirm_stable(struct drbd_connection *connection, struct packet_info *pi)
+{
+	struct p_confirm_stable *p = pi->data;
+
+	tl_release(connection, p->oldest_block_id, p->youngest_block_id, 0, be32_to_cpu(p->set_size));
 
 	return 0;
 }
@@ -8694,6 +8813,7 @@ static struct meta_sock_cmd ack_receiver_tbl[] = {
 	[P_NEG_RS_DREPLY]   = { sizeof(struct p_block_ack), got_NegRSDReply },
 	[P_OV_RESULT]	    = { sizeof(struct p_block_ack), got_OVResult },
 	[P_BARRIER_ACK]	    = { sizeof(struct p_barrier_ack), got_BarrierAck },
+	[P_CONFIRM_STABLE]  = { sizeof(struct p_confirm_stable), got_confirm_stable },
 	[P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply },
 	[P_RS_IS_IN_SYNC]   = { sizeof(struct p_block_ack), got_IsInSync },
 	[P_DELAY_PROBE]     = { sizeof(struct p_delay_probe93), got_skip },
diff --git a/drbd/drbd_req.c b/drbd/drbd_req.c
index 36f9d930..fe6a956a 100644
--- a/drbd/drbd_req.c
+++ b/drbd/drbd_req.c
@@ -191,9 +191,10 @@ void drbd_req_destroy(struct kref *kref)
 	struct drbd_device *device;
 	struct drbd_peer_device *peer_device;
 	unsigned int s;
-	bool was_last_ref = false;
+	bool was_last_ref;
 
  tail_recursion:
+	was_last_ref = false;
 	device = req->device;
 	s = req->local_rq_state;
 	destroy_next = req->destroy_next;
@@ -954,7 +955,7 @@ int __req_mod(struct drbd_request *req, enum drbd_req_event what,
 
 		/* close the epoch, in case it outgrew the limit,
 		 * or this is a "batch bio" */
-		if (!(req->local_rq_state & (RQ_UNMAP|RQ_WSAME))) {
+		if (!(req->local_rq_state & (RQ_UNMAP|RQ_WSAME|RQ_ZEROES))) {
 			rcu_read_lock();
 			nc = rcu_dereference(peer_device->connection->transport.net_conf);
 			p = nc->max_epoch_size;
@@ -1589,6 +1590,11 @@ drbd_request_prepare(struct drbd_device *device, struct bio *bio, ktime_t start_
 		return req;
 
 	/* Let the activity log know we are about to use it...
+	 * FIXME
+	 * Needs to slow down to not congest on the activity log, in case we
+	 * have multiple primaries and the peer sends huge scattered epochs.
+	 * See also how peer_requests are handled
+	 * in receive_Data() { ... prepare_activity_log(); ... }
 	 */
 	atomic_add(interval_to_al_extents(&req->i), &device->wait_for_actlog_ecnt);
 
@@ -1905,10 +1911,6 @@ static void __drbd_submit_peer_request(struct drbd_peer_request *peer_req)
 	struct drbd_device *device = peer_device->device;
 	int err;
 
-	D_ASSERT(peer_device,
-		0 == (peer_req->flags & (EE_IS_BARRIER|EE_TRIM|
-					EE_ZEROOUT|EE_WRITE_SAME)));
-
 	peer_req->flags |= EE_IN_ACTLOG;
 	atomic_sub(interval_to_al_extents(&peer_req->i), &device->wait_for_actlog_ecnt);
 	atomic_dec(&device->wait_for_actlog);
diff --git a/drbd/linux/drbd_config.h b/drbd/linux/drbd_config.h
index 50cdb12b..c4873c52 100644
--- a/drbd/linux/drbd_config.h
+++ b/drbd/linux/drbd_config.h
@@ -34,7 +34,7 @@ extern const char *drbd_buildtag(void);
 
 #define REL_VERSION "9.0.14-1"
 #define PRO_VERSION_MIN 86
-#define PRO_VERSION_MAX 113
+#define PRO_VERSION_MAX 114
 
 #ifndef __CHECKER__   /* for a sparse run, we need all STATICs */
 #define DBG_ALL_SYMBOLS /* no static functs, improves quality of OOPS traces */
-- 
2.34.1

