[Devel] [PATCH RH9] dm-ploop: port the standby mode feature.

[Alexander Atanasov]

Hi,

On 18.10.22 9:13, Kui Liu wrote:
> Hi, all
> 
> How about we introduce another block queue flag, say QUEUE_FLAG_STANDBY_EN, to indicate whether
> the device support the standby mode. the flag bit shall be set by SCST when a blockio device is created, if
> the blockio device is ploop device, then it will turn on standby mode check in ploop,  if the blockio
> device is not ploop device, it should have no impact at all.
> 
> As for ploop not being used by scst, the flag is always 0, so we can skip standby mode check in ploop by checking
> the flag at beginning of 'check_standby_mode()'.
> 
> How do you think of above implementation?
Something like this ?


-- 
Regards,
Alexander Atanasov
-------------- next part --------------
diff --git a/drivers/md/dm-ploop-map.c b/drivers/md/dm-ploop-map.c
index e8288e522e28..0c47754f3af2 100644
--- a/drivers/md/dm-ploop-map.c
+++ b/drivers/md/dm-ploop-map.c
@@ -19,6 +19,7 @@
 #include <uapi/linux/falloc.h>
 #include "dm-ploop.h"
 #include "dm-rq.h"
+#include "dm-core.h"
 
 #define PREALLOC_SIZE (128ULL * 1024 * 1024)
 
@@ -28,6 +29,8 @@ static void ploop_prq_endio(struct pio *pio, void *prq_ptr,
 
 #define DM_MSG_PREFIX "ploop"
 
+DEFINE_STATIC_KEY_FALSE(ploop_standby_check);
+
 static unsigned int ploop_pio_nr_segs(struct pio *pio)
 {
 	struct bvec_iter bi = {
@@ -1163,6 +1166,23 @@ static void ploop_queue_resubmit(struct pio *pio)
 	queue_work(ploop->wq, &ploop->worker);
 }
 
+static void ploop_check_standby_mode(struct ploop *ploop, long res) {
+	struct request_queue *q = ploop_blk_queue(ploop);
+	int prev;
+
+	/* move to standby if delta lease was stolen or mount is gone */
+	if (res != -EBUSY && res != -ENOTCONN && res != -EIO) {
+		return;
+	}
+
+	spin_lock_irq(&q->queue_lock);
+	prev = blk_queue_flag_test_and_set(QUEUE_FLAG_STANDBY, q);
+	spin_unlock_irq(&q->queue_lock);
+
+	if (!prev)
+		pr_info("ploop: switch into standby mode\n");
+}
+
 static void ploop_data_rw_complete(struct pio *pio)
 {
 	bool completed;
@@ -1175,6 +1195,8 @@ static void ploop_data_rw_complete(struct pio *pio)
 			ploop_queue_resubmit(pio);
 			return;
 		}
+		if (!static_branch_unlikely(&ploop_standby_check))
+			ploop_check_standby_mode(pio->ploop, pio->ret);
 		pio->bi_status = errno_to_blk_status(pio->ret);
 	}
 
@@ -1815,8 +1837,11 @@ void do_ploop_fsync_work(struct work_struct *ws)
 	ret = vfs_fsync(file, 0);
 
 	while ((pio = ploop_pio_list_pop(&flush_pios)) != NULL) {
-		if (unlikely(ret))
+		if (unlikely(ret)) {
 			pio->bi_status = errno_to_blk_status(ret);
+			if (!static_branch_unlikely(&ploop_standby_check))
+				ploop_check_standby_mode(ploop, ret);
+		}
 		ploop_pio_endio(pio);
 	}
 }
@@ -1869,6 +1894,12 @@ int ploop_clone_and_map(struct dm_target *ti, struct request *rq,
 	struct ploop_rq *prq;
 	struct pio *pio;
 
+
+	if (static_branch_unlikely(&ploop_standby_check)) {
+		if (blk_queue_standby(ploop_blk_queue(ploop)))
+			return DM_MAPIO_KILL;
+	}
+
 	if (blk_rq_bytes(rq) && ploop_rq_valid(ploop, rq) < 0)
 		return DM_MAPIO_KILL;
 
diff --git a/drivers/md/dm-ploop-target.c b/drivers/md/dm-ploop-target.c
index 4f7dc36eee0c..dda0fe957c1f 100644
--- a/drivers/md/dm-ploop-target.c
+++ b/drivers/md/dm-ploop-target.c
@@ -21,6 +21,7 @@
 #include <linux/uio.h>
 #include <linux/error-injection.h>
 #include "dm-ploop.h"
+#include "dm-core.h"
 
 #define DM_MSG_PREFIX "ploop"
 
@@ -32,6 +33,7 @@ MODULE_PARM_DESC(ignore_signature_disk_in_use,
 static struct kmem_cache *prq_cache;
 static struct kmem_cache *pio_cache;
 struct kmem_cache *cow_cache;
+extern struct static_key ploop_standby_check;
 
 static void ploop_aio_do_completion(struct pio *pio)
 {
@@ -406,6 +408,14 @@ static int ploop_ctr(struct dm_target *ti, unsigned int argc, char **argv)
 	ti->private = ploop;
 	ploop->ti = ti;
 
+	if (blk_queue_standby_en(ploop_blk_queue(ploop))) {
+		blk_queue_flag_clear(QUEUE_FLAG_STANDBY, ploop_blk_queue(ploop));
+		if (!static_key_enabled(&ploop_standby_check)) {
+			static_key_enable(&ploop_standby_check);
+			pr_info("ploop: standby mode check enabled\n");
+		}
+	}
+
 	if (kstrtou32(argv[0], 10, &ploop->cluster_log) < 0) {
 		ret = -EINVAL;
 		ti->error = "could not parse cluster_log";
diff --git a/drivers/md/dm-ploop.h b/drivers/md/dm-ploop.h
index 5d953278a976..f7c21b0b1597 100644
--- a/drivers/md/dm-ploop.h
+++ b/drivers/md/dm-ploop.h
@@ -230,6 +230,7 @@ struct ploop {
 	struct timer_list enospc_timer;
 	bool event_enospc;
 };
+#define ploop_blk_queue(p) ((p)->ti->table->md->queue)
 
 struct ploop_rq {
 	struct request *rq;
diff --git a/include/linux/blkdev.h b/include/linux/blkdev.h
index 8fcd753c70b0..8928161e5b1d 100644
--- a/include/linux/blkdev.h
+++ b/include/linux/blkdev.h
@@ -434,6 +434,8 @@ struct request_queue {
 #define QUEUE_FLAG_RQ_ALLOC_TIME 27	/* record rq->alloc_time_ns */
 #define QUEUE_FLAG_HCTX_ACTIVE	28	/* at least one blk-mq hctx is active */
 #define QUEUE_FLAG_NOWAIT       29	/* device supports NOWAIT */
+#define QUEUE_FLAG_STANDBY      30      /* unable to handle read/write requests */
+#define QUEUE_FLAG_STANDBY_EN   30      /* enable standby queue flag */
 
 #define QUEUE_FLAG_MQ_DEFAULT	((1 << QUEUE_FLAG_IO_STAT) |		\
 				 (1 << QUEUE_FLAG_SAME_COMP) |		\
@@ -480,6 +482,8 @@ bool blk_queue_flag_test_and_set(unsigned int flag, struct request_queue *q);
 #define blk_queue_fua(q)	test_bit(QUEUE_FLAG_FUA, &(q)->queue_flags)
 #define blk_queue_registered(q)	test_bit(QUEUE_FLAG_REGISTERED, &(q)->queue_flags)
 #define blk_queue_nowait(q)	test_bit(QUEUE_FLAG_NOWAIT, &(q)->queue_flags)
+#define blk_queue_standby(q)    test_bit(QUEUE_FLAG_STANDBY, &(q)->queue_flags)
+#define blk_queue_standby_en(q)    test_bit(QUEUE_FLAG_STANDBY_EN, &(q)->queue_flags)
 
 extern void blk_set_pm_only(struct request_queue *q);
 extern void blk_clear_pm_only(struct request_queue *q);