From ea29183adab4ebfc15cae1adfb8d12bf807dd909 Mon Sep 17 00:00:00 2001
From: Lars Ellenberg <lars.ellenberg@linbit.com>
Date: Thu, 19 Jul 2018 13:19:58 +0200
Subject: [PATCH] drbd: introduce active_ee_cnt for number of active peer
 requests

convert waiting for list_empty(active_ee) waiting for active_ee_cnt == 0

expose both done_ee_cnt and active_ee_cnt in debugfs connection/debug.

This is again a preparation commit for deadlock avoidance when huge epochs
of peer requests consume too many activity log extents at once.
---
 drbd/drbd_debugfs.c  |  4 ++++
 drbd/drbd_int.h      |  1 +
 drbd/drbd_receiver.c | 45 ++++++++++++++++++++------------------------
 drbd/drbd_sender.c   |  4 +++-
 4 files changed, 28 insertions(+), 26 deletions(-)

diff --git a/drbd/drbd_debugfs.c b/drbd/drbd_debugfs.c
index fb213ec5..6000a602 100644
--- a/drbd/drbd_debugfs.c
+++ b/drbd/drbd_debugfs.c
@@ -859,6 +859,10 @@ static int connection_debug_show(struct seq_file *m, void *ignored)
 	in_flight = atomic_read(&connection->rs_in_flight);
 	seq_printf(m, "            rs_in_flight: %d KiB (%d sectors)\n", in_flight / 2, in_flight);
 
+	seq_printf(m, "             done_ee_cnt: %d\n"
+	              "           active_ee_cnt: %d\n",
+		atomic_read(&connection->done_ee_cnt),
+		atomic_read(&connection->active_ee_cnt));
 	return 0;
 }
 
diff --git a/drbd/drbd_int.h b/drbd/drbd_int.h
index 3e9eb110..9b7724f1 100644
--- a/drbd/drbd_int.h
+++ b/drbd/drbd_int.h
@@ -968,6 +968,7 @@ struct drbd_connection {
 	struct list_head peer_requests; /* All peer requests in the order we received them.. */
 	u64 last_dagtag_sector;
 
+	atomic_t active_ee_cnt;
 	struct list_head active_ee; /* IO in progress (P_DATA gets written to disk) */
 	struct list_head sync_ee;   /* IO in progress (P_RS_DATA_REPLY gets written to disk) */
 	struct list_head read_ee;   /* [RS]P_DATA_REQUEST being read */
diff --git a/drbd/drbd_receiver.c b/drbd/drbd_receiver.c
index 87825f61..e45165da 100644
--- a/drbd/drbd_receiver.c
+++ b/drbd/drbd_receiver.c
@@ -1438,12 +1438,6 @@ static void conn_wait_ee_empty(struct drbd_connection *connection, struct list_h
 	wait_event(connection->ee_wait, conn_wait_ee_cond(connection, head));
 }
 
-static void conn_wait_ee_empty_or_disconnect(struct drbd_connection *connection, struct list_head *head)
-{
-	wait_event(connection->ee_wait,
-		   conn_wait_ee_cond(connection, head) || connection->cstate[NOW] < C_CONNECTED);
-}
-
 /**
  * drbd_submit_peer_request()
  * @device:	DRBD device.
@@ -1486,20 +1480,9 @@ int drbd_submit_peer_request(struct drbd_device *device,
 	 * asynchronous variant of the same.
 	 */
 	if (peer_req->flags & (EE_TRIM|EE_WRITE_SAME|EE_ZEROOUT)) {
-		struct drbd_connection *connection = 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, &connection->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 */
@@ -1693,6 +1676,18 @@ static void conn_wait_done_ee_empty(struct drbd_connection *connection)
 	wait_event(connection->ee_wait, atomic_read(&connection->done_ee_cnt) == 0);
 }
 
+static void conn_wait_active_ee_empty_or_disconnect(struct drbd_connection *connection)
+{
+	if (atomic_read(&connection->active_ee_cnt) == 0)
+		return;
+
+	drbd_unplug_all_devices(connection);
+
+	wait_event(connection->ee_wait,
+		atomic_read(&connection->active_ee_cnt) == 0
+		|| connection->cstate[NOW] < C_CONNECTED);
+}
+
 static int receive_Barrier(struct drbd_connection *connection, struct packet_info *pi)
 {
 	struct drbd_transport_ops *tr_ops = connection->transport.ops;
@@ -1726,7 +1721,7 @@ static int receive_Barrier(struct drbd_connection *connection, struct packet_inf
 	case WO_DRAIN_IO:
 		if (rv == FE_STILL_LIVE) {
 			set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &connection->current_epoch->flags);
-			conn_wait_ee_empty_or_disconnect(connection, &connection->active_ee);
+			conn_wait_active_ee_empty_or_disconnect(connection);
 			rv = drbd_flush_after_epoch(connection, connection->current_epoch);
 		}
 		if (rv == FE_RECYCLED)
@@ -1744,7 +1739,7 @@ static int receive_Barrier(struct drbd_connection *connection, struct packet_inf
 	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);
-		conn_wait_ee_empty_or_disconnect(connection, &connection->active_ee);
+		conn_wait_active_ee_empty_or_disconnect(connection);
 		if (issue_flush) {
 			rv = drbd_flush_after_epoch(connection, connection->current_epoch);
 			if (rv == FE_RECYCLED)
@@ -2774,12 +2769,9 @@ static int receive_Data(struct drbd_connection *connection, struct packet_info *
 		update_peer_seq(peer_device, d.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, &connection->active_ee);
+	/* Added to list here already, so debugfs can find it.
+	 */
+	list_add_tail(&peer_req->w.list, &connection->active_ee);
 	if (connection->agreed_pro_version >= 110)
 		list_add_tail(&peer_req->recv_order, &connection->peer_requests);
 	spin_unlock_irq(&device->resource->req_lock);
@@ -2798,6 +2790,8 @@ static int receive_Data(struct drbd_connection *connection, struct packet_info *
 	 * Remember the op_flags. */
 	peer_req->op_flags = op_flags;
 
+	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
@@ -2885,6 +2879,7 @@ void drbd_cleanup_peer_requests_wfa(struct drbd_device *device, struct list_head
 	spin_lock_irq(&device->resource->req_lock);
 	list_for_each_entry(peer_req, cleanup, wait_for_actlog) {
 		list_del(&peer_req->w.list); /* should be on the "->active_ee" list */
+		atomic_dec(&peer_req->peer_device->connection->active_ee_cnt);
 		list_del_init(&peer_req->recv_order);
 		drbd_remove_peer_req_interval(device, peer_req);
 	}
diff --git a/drbd/drbd_sender.c b/drbd/drbd_sender.c
index ff49655d..053cff9a 100644
--- a/drbd/drbd_sender.c
+++ b/drbd/drbd_sender.c
@@ -147,6 +147,8 @@ void drbd_endio_write_sec_final(struct drbd_peer_request *peer_req) __releases(l
 		__release(local);
 		spin_unlock_irqrestore(&device->resource->req_lock, flags);
 		drbd_queue_work(&connection->sender_work, &peer_req->w);
+		if (atomic_dec_and_test(&connection->active_ee_cnt))
+			wake_up(&connection->ee_wait);
 		return;
 	}
 
@@ -181,7 +183,7 @@ void drbd_endio_write_sec_final(struct drbd_peer_request *peer_req) __releases(l
 	if (block_id == ID_SYNCER)
 		do_wake = list_empty(&connection->sync_ee);
 	else
-		do_wake = list_empty(&connection->active_ee);
+		do_wake = atomic_dec_and_test(&connection->active_ee_cnt);
 
 	/* FIXME do we want to detach for failed REQ_DISCARD?
 	 * ((peer_req->flags & (EE_WAS_ERROR|EE_TRIM)) == EE_WAS_ERROR) */
-- 
2.34.1

