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

[Alexander Atanasov]

On 18.10.22 12:34, Alexander Atanasov wrote:
> 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 ?

Sorry for the attachment . Added check for blk_queue_standby_en in
ploop_check_standby_mode.

-- 
Regards,
Alexander Atanasov

diff --git a/drivers/md/dm-ploop-map.c b/drivers/md/dm-ploop-map.c
index e8288e522e28..a247eb169b32 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,26 @@ 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;
+
+	if (!blk_queue_standby_en(q))
+		return;
+
+	/* 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 +1198,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 +1840,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 +1897,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);