[dm-devel] [dm-6.4 PATCH 0/8] dm bufio, thin: improve concurrent IO performance

[dm-devel] [dm-6.4 PATCH 0/8] dm bufio, thin: improve concurrent IO performance

[Mike Snitzer]

Hi,

This patchset has been staged in linux-next for the 6.4 merge
window. As detailed in the "Testing" section of the header for patch
5: these changes have been extensively tested in terms of dm-thin (and
dm-cache as much as device-mapper-test-suite provides).

But further testing from other more exotic DM bufio consumers is
needed (e.g. dm-verity and dm-integrity developers/consumers who are
Cc'd).

Please use whatever tests you have to put pressure on bufio.
In particular: focused testing of bufio's shrinker, verify
DM_BUFIO_CLIENT_NO_SLEEP (via dm-verity's tasklets), etc.

There is still time to add fixes for 6.4 should any issues turn
up.

All review is welcome (I'll rebase to include Reviewed-by or Acked-by
tags). Also, positive reports in the form of Tested-by replies (to
patch 5) are also welcome!

Thanks,
Mike

Joe Thornber (4):
  dm bio prison v1: improve concurrent IO performance
  dm bufio: remove unused dm_bufio_release_move interface
  dm bufio: improve concurrent IO performance
  dm thin: speed up cell_defer_no_holder()

Mike Snitzer (4):
  dm: split discards further if target sets max_discard_granularity
  dm bufio: move dm_buffer struct
  dm bufio: move dm_bufio_client members to avoid spanning cachelines
  dm bufio: rename buffer_cache struct to dm_buffer_cache

 drivers/md/dm-bio-prison-v1.c |   87 +-
 drivers/md/dm-bio-prison-v1.h |   10 +
 drivers/md/dm-bufio.c         | 1939 +++++++++++++++++++++++----------
 drivers/md/dm-thin.c          |   23 +-
 drivers/md/dm.c               |   54 +-
 include/linux/device-mapper.h |    6 +
 include/linux/dm-bufio.h      |    6 -
 include/uapi/linux/dm-ioctl.h |    4 +-
 8 files changed, 1458 insertions(+), 671 deletions(-)

-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 1/8] dm: split discards further if target sets max_discard_granularity

[Mike Snitzer]

The block core (bio_split_discard) will already split discards based
on the 'discard_granularity' and 'max_discard_sectors' queue_limits.
But the DM thin target also needs to ensure that it doesn't receive a
discard that spans a 'max_discard_sectors' boundary.

Introduce a dm_target 'max_discard_granularity' flag that if set will
cause DM core to split discard bios relative to 'max_discard_sectors'.
This treats 'discard_granularity' as a "min_discard_granularity" and
'max_discard_sectors' as a "max_discard_granularity".

Requested-by: Joe Thornber <ejt@redhat.com>
Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm.c               | 54 +++++++++++++++++++----------------
 include/linux/device-mapper.h |  6 ++++
 include/uapi/linux/dm-ioctl.h |  4 +--
 3 files changed, 37 insertions(+), 27 deletions(-)

diff --git a/drivers/md/dm.c b/drivers/md/dm.c
index b6ace995b9ca..eeb58f89369e 100644
--- a/drivers/md/dm.c
+++ b/drivers/md/dm.c
@@ -1162,7 +1162,8 @@ static inline sector_t max_io_len_target_boundary(struct dm_target *ti,
 	return ti->len - target_offset;
 }
 
-static sector_t max_io_len(struct dm_target *ti, sector_t sector)
+static sector_t __max_io_len(struct dm_target *ti, sector_t sector,
+			     unsigned int max_granularity)
 {
 	sector_t target_offset = dm_target_offset(ti, sector);
 	sector_t len = max_io_len_target_boundary(ti, target_offset);
@@ -1173,11 +1174,16 @@ static sector_t max_io_len(struct dm_target *ti, sector_t sector)
 	 *   explains why stacked chunk_sectors based splitting via
 	 *   bio_split_to_limits() isn't possible here.
 	 */
-	if (!ti->max_io_len)
+	if (!max_granularity)
 		return len;
 	return min_t(sector_t, len,
 		min(queue_max_sectors(ti->table->md->queue),
-		    blk_chunk_sectors_left(target_offset, ti->max_io_len)));
+		    blk_chunk_sectors_left(target_offset, max_granularity)));
+}
+
+static inline sector_t max_io_len(struct dm_target *ti, sector_t sector)
+{
+	return __max_io_len(ti, sector, ti->max_io_len);
 }
 
 int dm_set_target_max_io_len(struct dm_target *ti, sector_t len)
@@ -1561,26 +1567,6 @@ static void __send_empty_flush(struct clone_info *ci)
 	bio_uninit(ci->bio);
 }
 
-static void __send_changing_extent_only(struct clone_info *ci, struct dm_target *ti,
-					unsigned int num_bios)
-{
-	unsigned int len, bios;
-
-	len = min_t(sector_t, ci->sector_count,
-		    max_io_len_target_boundary(ti, dm_target_offset(ti, ci->sector)));
-
-	atomic_add(num_bios, &ci->io->io_count);
-	bios = __send_duplicate_bios(ci, ti, num_bios, &len);
-	/*
-	 * alloc_io() takes one extra reference for submission, so the
-	 * reference won't reach 0 without the following (+1) subtraction
-	 */
-	atomic_sub(num_bios - bios + 1, &ci->io->io_count);
-
-	ci->sector += len;
-	ci->sector_count -= len;
-}
-
 static bool is_abnormal_io(struct bio *bio)
 {
 	enum req_op op = bio_op(bio);
@@ -1602,11 +1588,16 @@ static bool is_abnormal_io(struct bio *bio)
 static blk_status_t __process_abnormal_io(struct clone_info *ci,
 					  struct dm_target *ti)
 {
-	unsigned int num_bios = 0;
+	unsigned int bios, num_bios = 0;
+	unsigned int len, max_granularity = 0;
 
 	switch (bio_op(ci->bio)) {
 	case REQ_OP_DISCARD:
 		num_bios = ti->num_discard_bios;
+		if (ti->max_discard_granularity) {
+			struct queue_limits *limits = dm_get_queue_limits(ti->table->md);
+			max_granularity = limits->max_discard_sectors;
+		}
 		break;
 	case REQ_OP_SECURE_ERASE:
 		num_bios = ti->num_secure_erase_bios;
@@ -1627,7 +1618,20 @@ static blk_status_t __process_abnormal_io(struct clone_info *ci,
 	if (unlikely(!num_bios))
 		return BLK_STS_NOTSUPP;
 
-	__send_changing_extent_only(ci, ti, num_bios);
+	len = min_t(sector_t, ci->sector_count,
+		    __max_io_len(ti, ci->sector, max_granularity));
+
+	atomic_add(num_bios, &ci->io->io_count);
+	bios = __send_duplicate_bios(ci, ti, num_bios, &len);
+	/*
+	 * alloc_io() takes one extra reference for submission, so the
+	 * reference won't reach 0 without the following (+1) subtraction
+	 */
+	atomic_sub(num_bios - bios + 1, &ci->io->io_count);
+
+	ci->sector += len;
+	ci->sector_count -= len;
+
 	return BLK_STS_OK;
 }
 
diff --git a/include/linux/device-mapper.h b/include/linux/device-mapper.h
index 7975483816e4..8aa6b3ea91fa 100644
--- a/include/linux/device-mapper.h
+++ b/include/linux/device-mapper.h
@@ -358,6 +358,12 @@ struct dm_target {
 	 */
 	bool discards_supported:1;
 
+	/*
+	 * Set if this target requires that discards be split on both
+	 * 'discard_granularity' and 'max_discard_sectors' boundaries.
+	 */
+	bool max_discard_granularity:1;
+
 	/*
 	 * Set if we need to limit the number of in-flight bios when swapping.
 	 */
diff --git a/include/uapi/linux/dm-ioctl.h b/include/uapi/linux/dm-ioctl.h
index 7edf335778ba..1990b5700f69 100644
--- a/include/uapi/linux/dm-ioctl.h
+++ b/include/uapi/linux/dm-ioctl.h
@@ -286,9 +286,9 @@ enum {
 #define DM_DEV_SET_GEOMETRY	_IOWR(DM_IOCTL, DM_DEV_SET_GEOMETRY_CMD, struct dm_ioctl)
 
 #define DM_VERSION_MAJOR	4
-#define DM_VERSION_MINOR	47
+#define DM_VERSION_MINOR	48
 #define DM_VERSION_PATCHLEVEL	0
-#define DM_VERSION_EXTRA	"-ioctl (2022-07-28)"
+#define DM_VERSION_EXTRA	"-ioctl (2023-03-01)"
 
 /* Status bits */
 #define DM_READONLY_FLAG	(1 << 0) /* In/Out */
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

Re: [dm-devel] [dm-6.4 PATCH 1/8] dm: split discards further if target sets max_discard_granularity

[Joe Thornber]

[-- Attachment #1.1: Type: text/plain, Size: 6585 bytes --]

Also nacking this patch since you refused to test it.  I said I'd test
since you haven't, but you couldn't wait.

- Joe

On Wed, Mar 22, 2023 at 6:19 PM Mike Snitzer <snitzer@kernel.org> wrote:

> The block core (bio_split_discard) will already split discards based
> on the 'discard_granularity' and 'max_discard_sectors' queue_limits.
> But the DM thin target also needs to ensure that it doesn't receive a
> discard that spans a 'max_discard_sectors' boundary.
>
> Introduce a dm_target 'max_discard_granularity' flag that if set will
> cause DM core to split discard bios relative to 'max_discard_sectors'.
> This treats 'discard_granularity' as a "min_discard_granularity" and
> 'max_discard_sectors' as a "max_discard_granularity".
>
> Requested-by: Joe Thornber <ejt@redhat.com>
> Signed-off-by: Mike Snitzer <snitzer@kernel.org>
> ---
>  drivers/md/dm.c               | 54 +++++++++++++++++++----------------
>  include/linux/device-mapper.h |  6 ++++
>  include/uapi/linux/dm-ioctl.h |  4 +--
>  3 files changed, 37 insertions(+), 27 deletions(-)
>
> diff --git a/drivers/md/dm.c b/drivers/md/dm.c
> index b6ace995b9ca..eeb58f89369e 100644
> --- a/drivers/md/dm.c
> +++ b/drivers/md/dm.c
> @@ -1162,7 +1162,8 @@ static inline sector_t
> max_io_len_target_boundary(struct dm_target *ti,
>         return ti->len - target_offset;
>  }
>
> -static sector_t max_io_len(struct dm_target *ti, sector_t sector)
> +static sector_t __max_io_len(struct dm_target *ti, sector_t sector,
> +                            unsigned int max_granularity)
>  {
>         sector_t target_offset = dm_target_offset(ti, sector);
>         sector_t len = max_io_len_target_boundary(ti, target_offset);
> @@ -1173,11 +1174,16 @@ static sector_t max_io_len(struct dm_target *ti,
> sector_t sector)
>          *   explains why stacked chunk_sectors based splitting via
>          *   bio_split_to_limits() isn't possible here.
>          */
> -       if (!ti->max_io_len)
> +       if (!max_granularity)
>                 return len;
>         return min_t(sector_t, len,
>                 min(queue_max_sectors(ti->table->md->queue),
> -                   blk_chunk_sectors_left(target_offset,
> ti->max_io_len)));
> +                   blk_chunk_sectors_left(target_offset,
> max_granularity)));
> +}
> +
> +static inline sector_t max_io_len(struct dm_target *ti, sector_t sector)
> +{
> +       return __max_io_len(ti, sector, ti->max_io_len);
>  }
>
>  int dm_set_target_max_io_len(struct dm_target *ti, sector_t len)
> @@ -1561,26 +1567,6 @@ static void __send_empty_flush(struct clone_info
> *ci)
>         bio_uninit(ci->bio);
>  }
>
> -static void __send_changing_extent_only(struct clone_info *ci, struct
> dm_target *ti,
> -                                       unsigned int num_bios)
> -{
> -       unsigned int len, bios;
> -
> -       len = min_t(sector_t, ci->sector_count,
> -                   max_io_len_target_boundary(ti, dm_target_offset(ti,
> ci->sector)));
> -
> -       atomic_add(num_bios, &ci->io->io_count);
> -       bios = __send_duplicate_bios(ci, ti, num_bios, &len);
> -       /*
> -        * alloc_io() takes one extra reference for submission, so the
> -        * reference won't reach 0 without the following (+1) subtraction
> -        */
> -       atomic_sub(num_bios - bios + 1, &ci->io->io_count);
> -
> -       ci->sector += len;
> -       ci->sector_count -= len;
> -}
> -
>  static bool is_abnormal_io(struct bio *bio)
>  {
>         enum req_op op = bio_op(bio);
> @@ -1602,11 +1588,16 @@ static bool is_abnormal_io(struct bio *bio)
>  static blk_status_t __process_abnormal_io(struct clone_info *ci,
>                                           struct dm_target *ti)
>  {
> -       unsigned int num_bios = 0;
> +       unsigned int bios, num_bios = 0;
> +       unsigned int len, max_granularity = 0;
>
>         switch (bio_op(ci->bio)) {
>         case REQ_OP_DISCARD:
>                 num_bios = ti->num_discard_bios;
> +               if (ti->max_discard_granularity) {
> +                       struct queue_limits *limits =
> dm_get_queue_limits(ti->table->md);
> +                       max_granularity = limits->max_discard_sectors;
> +               }
>                 break;
>         case REQ_OP_SECURE_ERASE:
>                 num_bios = ti->num_secure_erase_bios;
> @@ -1627,7 +1618,20 @@ static blk_status_t __process_abnormal_io(struct
> clone_info *ci,
>         if (unlikely(!num_bios))
>                 return BLK_STS_NOTSUPP;
>
> -       __send_changing_extent_only(ci, ti, num_bios);
> +       len = min_t(sector_t, ci->sector_count,
> +                   __max_io_len(ti, ci->sector, max_granularity));
> +
> +       atomic_add(num_bios, &ci->io->io_count);
> +       bios = __send_duplicate_bios(ci, ti, num_bios, &len);
> +       /*
> +        * alloc_io() takes one extra reference for submission, so the
> +        * reference won't reach 0 without the following (+1) subtraction
> +        */
> +       atomic_sub(num_bios - bios + 1, &ci->io->io_count);
> +
> +       ci->sector += len;
> +       ci->sector_count -= len;
> +
>         return BLK_STS_OK;
>  }
>
> diff --git a/include/linux/device-mapper.h b/include/linux/device-mapper.h
> index 7975483816e4..8aa6b3ea91fa 100644
> --- a/include/linux/device-mapper.h
> +++ b/include/linux/device-mapper.h
> @@ -358,6 +358,12 @@ struct dm_target {
>          */
>         bool discards_supported:1;
>
> +       /*
> +        * Set if this target requires that discards be split on both
> +        * 'discard_granularity' and 'max_discard_sectors' boundaries.
> +        */
> +       bool max_discard_granularity:1;
> +
>         /*
>          * Set if we need to limit the number of in-flight bios when
> swapping.
>          */
> diff --git a/include/uapi/linux/dm-ioctl.h b/include/uapi/linux/dm-ioctl.h
> index 7edf335778ba..1990b5700f69 100644
> --- a/include/uapi/linux/dm-ioctl.h
> +++ b/include/uapi/linux/dm-ioctl.h
> @@ -286,9 +286,9 @@ enum {
>  #define DM_DEV_SET_GEOMETRY    _IOWR(DM_IOCTL, DM_DEV_SET_GEOMETRY_CMD,
> struct dm_ioctl)
>
>  #define DM_VERSION_MAJOR       4
> -#define DM_VERSION_MINOR       47
> +#define DM_VERSION_MINOR       48
>  #define DM_VERSION_PATCHLEVEL  0
> -#define DM_VERSION_EXTRA       "-ioctl (2022-07-28)"
> +#define DM_VERSION_EXTRA       "-ioctl (2023-03-01)"
>
>  /* Status bits */
>  #define DM_READONLY_FLAG       (1 << 0) /* In/Out */
> --
> 2.40.0
>
>

[-- Attachment #1.2: Type: text/html, Size: 8008 bytes --]

[-- Attachment #2: Type: text/plain, Size: 98 bytes --]

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

Re: [dm-devel] [dm-6.4 PATCH 1/8] dm: split discards further if target sets max_discard_granularity

[Mike Snitzer]

On Thu, Mar 23 2023 at  4:00P -0400,
Joe Thornber <thornber@redhat.com> wrote:

> Also nacking this patch since you refused to test it.  I said I'd test
> since you haven't, but you couldn't wait.

[Pleaase don't top-post.]

That's fine. I had already dropped this patch from linux-next due to
my testing showing it doesn't work.  I didn't think to email the list
last night.  My bad.

But I didn't refuse to test it. That implies malice or something.

Mike

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 2/8] dm bio prison v1: improve concurrent IO performance

[Mike Snitzer]

From: Joe Thornber <ejt@redhat.com>

Split the bio prison into multiple regions, with a separate rbtree and
associated lock for each region.

The range of a key (block_end - block_begin) must not exceed
BIO_PRISON_MAX_RANGE.

To get fast bio prison locking and not damage the performance of
discards too much the bio-prison must stipulate that discards should
not cross a BIO_PRISON_MAX_RANGE boundary. This is achieved by
updating dm-thin.c to set limits->max_discard_sectors in terms of
BIO_PRISON_MAX_RANGE _and_ setting the thin and thin-pool targets'
max_discard_granularity to true.

Signed-off-by: Joe Thornber <ejt@redhat.com>
Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-bio-prison-v1.c | 87 +++++++++++++++++++++++------------
 drivers/md/dm-bio-prison-v1.h | 10 ++++
 drivers/md/dm-thin.c          |  9 ++--
 3 files changed, 72 insertions(+), 34 deletions(-)

diff --git a/drivers/md/dm-bio-prison-v1.c b/drivers/md/dm-bio-prison-v1.c
index c4c05d5d8909..e35ed236e0dc 100644
--- a/drivers/md/dm-bio-prison-v1.c
+++ b/drivers/md/dm-bio-prison-v1.c
@@ -16,11 +16,17 @@
 
 /*----------------------------------------------------------------*/
 
+#define NR_LOCKS 64
+#define LOCK_MASK (NR_LOCKS - 1)
 #define MIN_CELLS 1024
 
+struct lock {
+	spinlock_t lock;
+} ____cacheline_aligned_in_smp;
+
 struct dm_bio_prison {
-	spinlock_t lock;
-	struct rb_root cells;
+	struct lock lock[NR_LOCKS];
+	struct rb_root cells[NR_LOCKS];
 	mempool_t cell_pool;
 };
 
@@ -34,13 +40,17 @@ static struct kmem_cache *_cell_cache;
  */
 struct dm_bio_prison *dm_bio_prison_create(void)
 {
-	struct dm_bio_prison *prison = kzalloc(sizeof(*prison), GFP_KERNEL);
 	int ret;
+	unsigned i;
+	struct dm_bio_prison *prison = kzalloc(sizeof(*prison), GFP_KERNEL);
 
 	if (!prison)
 		return NULL;
 
-	spin_lock_init(&prison->lock);
+	for (i = 0; i < NR_LOCKS; i++) {
+		spin_lock_init(&prison->lock[i].lock);
+		prison->cells[i] = RB_ROOT;
+	}
 
 	ret = mempool_init_slab_pool(&prison->cell_pool, MIN_CELLS, _cell_cache);
 	if (ret) {
@@ -48,8 +58,6 @@ struct dm_bio_prison *dm_bio_prison_create(void)
 		return NULL;
 	}
 
-	prison->cells = RB_ROOT;
-
 	return prison;
 }
 EXPORT_SYMBOL_GPL(dm_bio_prison_create);
@@ -107,14 +115,26 @@ static int cmp_keys(struct dm_cell_key *lhs,
 	return 0;
 }
 
-static int __bio_detain(struct dm_bio_prison *prison,
+static unsigned lock_nr(struct dm_cell_key *key)
+{
+	return (key->block_begin >> BIO_PRISON_MAX_RANGE_SHIFT) & LOCK_MASK;
+}
+
+static void check_range(struct dm_cell_key *key)
+{
+	BUG_ON(key->block_end - key->block_begin > BIO_PRISON_MAX_RANGE);
+	BUG_ON((key->block_begin >> BIO_PRISON_MAX_RANGE_SHIFT) !=
+	       ((key->block_end - 1) >> BIO_PRISON_MAX_RANGE_SHIFT));
+}
+
+static int __bio_detain(struct rb_root *root,
 			struct dm_cell_key *key,
 			struct bio *inmate,
 			struct dm_bio_prison_cell *cell_prealloc,
 			struct dm_bio_prison_cell **cell_result)
 {
 	int r;
-	struct rb_node **new = &prison->cells.rb_node, *parent = NULL;
+	struct rb_node **new = &root->rb_node, *parent = NULL;
 
 	while (*new) {
 		struct dm_bio_prison_cell *cell =
@@ -139,7 +159,7 @@ static int __bio_detain(struct dm_bio_prison *prison,
 	*cell_result = cell_prealloc;
 
 	rb_link_node(&cell_prealloc->node, parent, new);
-	rb_insert_color(&cell_prealloc->node, &prison->cells);
+	rb_insert_color(&cell_prealloc->node, root);
 
 	return 0;
 }
@@ -151,10 +171,12 @@ static int bio_detain(struct dm_bio_prison *prison,
 		      struct dm_bio_prison_cell **cell_result)
 {
 	int r;
+	unsigned l = lock_nr(key);
+	check_range(key);
 
-	spin_lock_irq(&prison->lock);
-	r = __bio_detain(prison, key, inmate, cell_prealloc, cell_result);
-	spin_unlock_irq(&prison->lock);
+	spin_lock_irq(&prison->lock[l].lock);
+	r = __bio_detain(&prison->cells[l], key, inmate, cell_prealloc, cell_result);
+	spin_unlock_irq(&prison->lock[l].lock);
 
 	return r;
 }
@@ -181,11 +203,11 @@ EXPORT_SYMBOL_GPL(dm_get_cell);
 /*
  * @inmates must have been initialised prior to this call
  */
-static void __cell_release(struct dm_bio_prison *prison,
+static void __cell_release(struct rb_root *root,
 			   struct dm_bio_prison_cell *cell,
 			   struct bio_list *inmates)
 {
-	rb_erase(&cell->node, &prison->cells);
+	rb_erase(&cell->node, root);
 
 	if (inmates) {
 		if (cell->holder)
@@ -198,20 +220,22 @@ void dm_cell_release(struct dm_bio_prison *prison,
 		     struct dm_bio_prison_cell *cell,
 		     struct bio_list *bios)
 {
-	spin_lock_irq(&prison->lock);
-	__cell_release(prison, cell, bios);
-	spin_unlock_irq(&prison->lock);
+	unsigned l = lock_nr(&cell->key);
+
+	spin_lock_irq(&prison->lock[l].lock);
+	__cell_release(&prison->cells[l], cell, bios);
+	spin_unlock_irq(&prison->lock[l].lock);
 }
 EXPORT_SYMBOL_GPL(dm_cell_release);
 
 /*
  * Sometimes we don't want the holder, just the additional bios.
  */
-static void __cell_release_no_holder(struct dm_bio_prison *prison,
+static void __cell_release_no_holder(struct rb_root *root,
 				     struct dm_bio_prison_cell *cell,
 				     struct bio_list *inmates)
 {
-	rb_erase(&cell->node, &prison->cells);
+	rb_erase(&cell->node, root);
 	bio_list_merge(inmates, &cell->bios);
 }
 
@@ -219,11 +243,12 @@ void dm_cell_release_no_holder(struct dm_bio_prison *prison,
 			       struct dm_bio_prison_cell *cell,
 			       struct bio_list *inmates)
 {
+	unsigned l = lock_nr(&cell->key);
 	unsigned long flags;
 
-	spin_lock_irqsave(&prison->lock, flags);
-	__cell_release_no_holder(prison, cell, inmates);
-	spin_unlock_irqrestore(&prison->lock, flags);
+	spin_lock_irqsave(&prison->lock[l].lock, flags);
+	__cell_release_no_holder(&prison->cells[l], cell, inmates);
+	spin_unlock_irqrestore(&prison->lock[l].lock, flags);
 }
 EXPORT_SYMBOL_GPL(dm_cell_release_no_holder);
 
@@ -248,18 +273,19 @@ void dm_cell_visit_release(struct dm_bio_prison *prison,
 			   void *context,
 			   struct dm_bio_prison_cell *cell)
 {
-	spin_lock_irq(&prison->lock);
+	unsigned l = lock_nr(&cell->key);
+	spin_lock_irq(&prison->lock[l].lock);
 	visit_fn(context, cell);
-	rb_erase(&cell->node, &prison->cells);
-	spin_unlock_irq(&prison->lock);
+	rb_erase(&cell->node, &prison->cells[l]);
+	spin_unlock_irq(&prison->lock[l].lock);
 }
 EXPORT_SYMBOL_GPL(dm_cell_visit_release);
 
-static int __promote_or_release(struct dm_bio_prison *prison,
+static int __promote_or_release(struct rb_root *root,
 				struct dm_bio_prison_cell *cell)
 {
 	if (bio_list_empty(&cell->bios)) {
-		rb_erase(&cell->node, &prison->cells);
+		rb_erase(&cell->node, root);
 		return 1;
 	}
 
@@ -271,10 +297,11 @@ int dm_cell_promote_or_release(struct dm_bio_prison *prison,
 			       struct dm_bio_prison_cell *cell)
 {
 	int r;
+	unsigned l = lock_nr(&cell->key);
 
-	spin_lock_irq(&prison->lock);
-	r = __promote_or_release(prison, cell);
-	spin_unlock_irq(&prison->lock);
+	spin_lock_irq(&prison->lock[l].lock);
+	r = __promote_or_release(&prison->cells[l], cell);
+	spin_unlock_irq(&prison->lock[l].lock);
 
 	return r;
 }
diff --git a/drivers/md/dm-bio-prison-v1.h b/drivers/md/dm-bio-prison-v1.h
index dfbf1e94cb75..0b8acd6708fb 100644
--- a/drivers/md/dm-bio-prison-v1.h
+++ b/drivers/md/dm-bio-prison-v1.h
@@ -34,6 +34,16 @@ struct dm_cell_key {
 	dm_block_t block_begin, block_end;
 };
 
+/*
+ * The range of a key (block_end - block_begin) must not
+ * exceed BIO_PRISON_MAX_RANGE.  Also the range must not
+ * cross a similarly sized boundary.
+ *
+ * Must be a power of 2.
+ */
+#define BIO_PRISON_MAX_RANGE 1024
+#define BIO_PRISON_MAX_RANGE_SHIFT 10
+
 /*
  * Treat this as opaque, only in header so callers can manage allocation
  * themselves.
diff --git a/drivers/md/dm-thin.c b/drivers/md/dm-thin.c
index 13d4677baafd..286e4d9e45c8 100644
--- a/drivers/md/dm-thin.c
+++ b/drivers/md/dm-thin.c
@@ -3378,13 +3378,13 @@ static int pool_ctr(struct dm_target *ti, unsigned int argc, char **argv)
 	 */
 	if (pf.discard_enabled && pf.discard_passdown) {
 		ti->num_discard_bios = 1;
-
 		/*
 		 * Setting 'discards_supported' circumvents the normal
 		 * stacking of discard limits (this keeps the pool and
 		 * thin devices' discard limits consistent).
 		 */
 		ti->discards_supported = true;
+		ti->max_discard_granularity = true;
 	}
 	ti->private = pt;
 
@@ -4094,7 +4094,7 @@ static struct target_type pool_target = {
 	.name = "thin-pool",
 	.features = DM_TARGET_SINGLETON | DM_TARGET_ALWAYS_WRITEABLE |
 		    DM_TARGET_IMMUTABLE,
-	.version = {1, 22, 0},
+	.version = {1, 23, 0},
 	.module = THIS_MODULE,
 	.ctr = pool_ctr,
 	.dtr = pool_dtr,
@@ -4259,6 +4259,7 @@ static int thin_ctr(struct dm_target *ti, unsigned int argc, char **argv)
 	if (tc->pool->pf.discard_enabled) {
 		ti->discards_supported = true;
 		ti->num_discard_bios = 1;
+		ti->max_discard_granularity = true;
 	}
 
 	mutex_unlock(&dm_thin_pool_table.mutex);
@@ -4474,12 +4475,12 @@ static void thin_io_hints(struct dm_target *ti, struct queue_limits *limits)
 		return;
 
 	limits->discard_granularity = pool->sectors_per_block << SECTOR_SHIFT;
-	limits->max_discard_sectors = 2048 * 1024 * 16; /* 16G */
+	limits->max_discard_sectors = pool->sectors_per_block * BIO_PRISON_MAX_RANGE;
 }
 
 static struct target_type thin_target = {
 	.name = "thin",
-	.version = {1, 22, 0},
+	.version = {1, 23, 0},
 	.module	= THIS_MODULE,
 	.ctr = thin_ctr,
 	.dtr = thin_dtr,
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 3/8] dm bufio: remove unused dm_bufio_release_move interface

[Mike Snitzer]

From: Joe Thornber <ejt@redhat.com>

Was used by multi-snapshot DM target that never went upstream.

Signed-off-by: Joe Thornber <ejt@redhat.com>
Acked-by: Mikulas Patocka <mpatocka@redhat.com>
Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-bufio.c    | 77 ----------------------------------------
 include/linux/dm-bufio.h |  6 ----
 2 files changed, 83 deletions(-)

diff --git a/drivers/md/dm-bufio.c b/drivers/md/dm-bufio.c
index cf077f9b30c3..79434b38f368 100644
--- a/drivers/md/dm-bufio.c
+++ b/drivers/md/dm-bufio.c
@@ -1415,83 +1415,6 @@ int dm_bufio_issue_discard(struct dm_bufio_client *c, sector_t block, sector_t c
 }
 EXPORT_SYMBOL_GPL(dm_bufio_issue_discard);
 
-/*
- * We first delete any other buffer that may be at that new location.
- *
- * Then, we write the buffer to the original location if it was dirty.
- *
- * Then, if we are the only one who is holding the buffer, relink the buffer
- * in the buffer tree for the new location.
- *
- * If there was someone else holding the buffer, we write it to the new
- * location but not relink it, because that other user needs to have the buffer
- * at the same place.
- */
-void dm_bufio_release_move(struct dm_buffer *b, sector_t new_block)
-{
-	struct dm_bufio_client *c = b->c;
-	struct dm_buffer *new;
-
-	BUG_ON(dm_bufio_in_request());
-
-	dm_bufio_lock(c);
-
-retry:
-	new = __find(c, new_block);
-	if (new) {
-		if (new->hold_count) {
-			__wait_for_free_buffer(c);
-			goto retry;
-		}
-
-		/*
-		 * FIXME: Is there any point waiting for a write that's going
-		 * to be overwritten in a bit?
-		 */
-		__make_buffer_clean(new);
-		__unlink_buffer(new);
-		__free_buffer_wake(new);
-	}
-
-	BUG_ON(!b->hold_count);
-	BUG_ON(test_bit(B_READING, &b->state));
-
-	__write_dirty_buffer(b, NULL);
-	if (b->hold_count == 1) {
-		wait_on_bit_io(&b->state, B_WRITING,
-			       TASK_UNINTERRUPTIBLE);
-		set_bit(B_DIRTY, &b->state);
-		b->dirty_start = 0;
-		b->dirty_end = c->block_size;
-		__unlink_buffer(b);
-		__link_buffer(b, new_block, LIST_DIRTY);
-	} else {
-		sector_t old_block;
-
-		wait_on_bit_lock_io(&b->state, B_WRITING,
-				    TASK_UNINTERRUPTIBLE);
-		/*
-		 * Relink buffer to "new_block" so that write_callback
-		 * sees "new_block" as a block number.
-		 * After the write, link the buffer back to old_block.
-		 * All this must be done in bufio lock, so that block number
-		 * change isn't visible to other threads.
-		 */
-		old_block = b->block;
-		__unlink_buffer(b);
-		__link_buffer(b, new_block, b->list_mode);
-		submit_io(b, REQ_OP_WRITE, write_endio);
-		wait_on_bit_io(&b->state, B_WRITING,
-			       TASK_UNINTERRUPTIBLE);
-		__unlink_buffer(b);
-		__link_buffer(b, old_block, b->list_mode);
-	}
-
-	dm_bufio_unlock(c);
-	dm_bufio_release(b);
-}
-EXPORT_SYMBOL_GPL(dm_bufio_release_move);
-
 static void forget_buffer_locked(struct dm_buffer *b)
 {
 	if (likely(!b->hold_count) && likely(!smp_load_acquire(&b->state))) {
diff --git a/include/linux/dm-bufio.h b/include/linux/dm-bufio.h
index 2056743aaaaa..681656a1c03d 100644
--- a/include/linux/dm-bufio.h
+++ b/include/linux/dm-bufio.h
@@ -130,12 +130,6 @@ int dm_bufio_issue_flush(struct dm_bufio_client *c);
  */
 int dm_bufio_issue_discard(struct dm_bufio_client *c, sector_t block, sector_t count);
 
-/*
- * Like dm_bufio_release but also move the buffer to the new
- * block. dm_bufio_write_dirty_buffers is needed to commit the new block.
- */
-void dm_bufio_release_move(struct dm_buffer *b, sector_t new_block);
-
 /*
  * Free the given buffer.
  * This is just a hint, if the buffer is in use or dirty, this function
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 4/8] dm bufio: move dm_buffer struct

[Mike Snitzer]

Movement prepares for finer grained dm_buffer changes, in the next
commit, to be more easily seen.

Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-bufio.c | 97 ++++++++++++++++++++++---------------------
 1 file changed, 50 insertions(+), 47 deletions(-)

diff --git a/drivers/md/dm-bufio.c b/drivers/md/dm-bufio.c
index 79434b38f368..1de1bdcda1ce 100644
--- a/drivers/md/dm-bufio.c
+++ b/drivers/md/dm-bufio.c
@@ -66,6 +66,56 @@
 #define LIST_DIRTY	1
 #define LIST_SIZE	2
 
+/*
+ * Buffer state bits.
+ */
+#define B_READING	0
+#define B_WRITING	1
+#define B_DIRTY		2
+
+/*
+ * Describes how the block was allocated:
+ * kmem_cache_alloc(), __get_free_pages() or vmalloc().
+ * See the comment at alloc_buffer_data.
+ */
+enum data_mode {
+	DATA_MODE_SLAB = 0,
+	DATA_MODE_GET_FREE_PAGES = 1,
+	DATA_MODE_VMALLOC = 2,
+	DATA_MODE_LIMIT = 3
+};
+
+struct dm_buffer {
+	struct rb_node node;
+	struct list_head lru_list;
+	struct list_head global_list;
+
+	sector_t block;
+	void *data;
+	unsigned char data_mode;		/* DATA_MODE_* */
+
+	unsigned int accessed;
+	unsigned int hold_count;
+	unsigned long last_accessed;
+
+	unsigned char list_mode;		/* LIST_* */
+	blk_status_t read_error;
+	blk_status_t write_error;
+	unsigned long state;
+	unsigned int dirty_start;
+	unsigned int dirty_end;
+	unsigned int write_start;
+	unsigned int write_end;
+	struct dm_bufio_client *c;
+	struct list_head write_list;
+	void (*end_io)(struct dm_buffer *b, blk_status_t bs);
+#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
+#define MAX_STACK 10
+	unsigned int stack_len;
+	unsigned long stack_entries[MAX_STACK];
+#endif
+};
+
 /*
  * Linking of buffers:
  *	All buffers are linked to buffer_tree with their node field.
@@ -117,53 +167,6 @@ struct dm_bufio_client {
 	atomic_long_t need_shrink;
 };
 
-/*
- * Buffer state bits.
- */
-#define B_READING	0
-#define B_WRITING	1
-#define B_DIRTY		2
-
-/*
- * Describes how the block was allocated:
- * kmem_cache_alloc(), __get_free_pages() or vmalloc().
- * See the comment at alloc_buffer_data.
- */
-enum data_mode {
-	DATA_MODE_SLAB = 0,
-	DATA_MODE_GET_FREE_PAGES = 1,
-	DATA_MODE_VMALLOC = 2,
-	DATA_MODE_LIMIT = 3
-};
-
-struct dm_buffer {
-	struct rb_node node;
-	struct list_head lru_list;
-	struct list_head global_list;
-	sector_t block;
-	void *data;
-	unsigned char data_mode;		/* DATA_MODE_* */
-	unsigned char list_mode;		/* LIST_* */
-	blk_status_t read_error;
-	blk_status_t write_error;
-	unsigned int accessed;
-	unsigned int hold_count;
-	unsigned long state;
-	unsigned long last_accessed;
-	unsigned int dirty_start;
-	unsigned int dirty_end;
-	unsigned int write_start;
-	unsigned int write_end;
-	struct dm_bufio_client *c;
-	struct list_head write_list;
-	void (*end_io)(struct dm_buffer *buf, blk_status_t stat);
-#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
-#define MAX_STACK 10
-	unsigned int stack_len;
-	unsigned long stack_entries[MAX_STACK];
-#endif
-};
-
 static DEFINE_STATIC_KEY_FALSE(no_sleep_enabled);
 
 /*----------------------------------------------------------------*/
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 5/8] dm bufio: improve concurrent IO performance

[Mike Snitzer]

From: Joe Thornber <ejt@redhat.com>

When multiple threads perform IO to a thin device, the underlying
dm_bufio object can become a bottleneck; slowing down access to btree
nodes that store the thin metadata. Prior to this commit, each bufio
instance had a single mutex that was taken for every bufio operation.

This commit concentrates on improving the common case: in which a user
of dm_bufio wishes to access, but not modify, a buffer which is already
within the dm_bufio cache.

Implementation::

  The code has been refactored; pulling out an 'lru' abstraction and a
  'buffer cache' abstraction.

 LRU:
  A clock algorithm is used in the LRU abstraction.  This avoids
  relinking list nodes, which would require a write lock protecting
  it.

  None of the lru methods are threadsafe; locking must be done at a
  higher level.

 Buffer cache:
  The buffer cache is responsible for managing the holder count,
  tracking clean/dirty state, and choosing buffers via predicates.
  Higher level code is responsible for allocation of buffers, IO and
  eviction/cache sizing.

  The buffer cache has thread safe methods for acquiring a reference
  to an existing buffer. All other methods in buffer cache are _not_
  threadsafe, and only contain enough locking to guarantee the safe
  methods.

  Rather than a single mutex, sharded rw_semaphores are used to allow
  concurrent threads to 'get' buffers. Each rw_semaphore protects its
  own rbtree of buffer entries.

Testing::

  Some of the low level functions have been unit tested using dm-unit:
    https://github.com/jthornber/dm-unit/blob/main/src/tests/bufio.rs

  Higher level concurrency and IO is tested via a test only target
  found here:
    https://github.com/jthornber/linux/blob/2023-02-28-thin-concurrency-7/drivers/md/dm-bufio-test.c

  The associated userland side of these tests is here:
    https://github.com/jthornber/dmtest-python/blob/main/src/dmtest/bufio/bufio_tests.py

  In addition the full dmtest suite of tests (dm-thin, dm-cache, etc)
  has been run (~450 tests).

Performance::

  Most bufio operations have unchanged performance. But if multiple
  threads are attempting to get buffers concurrently, and these
  buffers are already in the cache then there's a big speed up. Eg,
  one test has 16 'hotspot' threads simulating btree lookups while
  another thread dirties the whole device. In this case the hotspot
  threads acquire the buffers about 25 times faster.

Signed-off-by: Joe Thornber <ejt@redhat.com>
Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-bufio.c | 1769 ++++++++++++++++++++++++++++++-----------
 1 file changed, 1292 insertions(+), 477 deletions(-)

diff --git a/drivers/md/dm-bufio.c b/drivers/md/dm-bufio.c
index 1de1bdcda1ce..56ef160331ce 100644
--- a/drivers/md/dm-bufio.c
+++ b/drivers/md/dm-bufio.c
@@ -66,6 +66,247 @@
 #define LIST_DIRTY	1
 #define LIST_SIZE	2
 
+/*--------------------------------------------------------------*/
+
+/*
+ * Rather than use an LRU list, we use a clock algorithm where entries
+ * are held in a circular list.  When an entry is 'hit' a reference bit
+ * is set.  The least recently used entry is approximated by running a
+ * cursor around the list selecting unreferenced entries. Referenced
+ * entries have their reference bit cleared as the cursor passes them.
+ */
+struct lru_entry {
+	struct list_head list;
+	atomic_t referenced;
+};
+
+struct lru_iter {
+	struct lru *lru;
+	struct list_head list;
+	struct lru_entry *stop;
+	struct lru_entry *e;
+};
+
+struct lru {
+	struct list_head *cursor;
+	unsigned long count;
+
+	struct list_head iterators;
+};
+
+/*--------------*/
+
+static void lru_init(struct lru *lru)
+{
+	lru->cursor = NULL;
+	lru->count = 0;
+	INIT_LIST_HEAD(&lru->iterators);
+}
+
+static void lru_destroy(struct lru *lru)
+{
+	BUG_ON(lru->cursor);
+	BUG_ON(!list_empty(&lru->iterators));
+}
+
+static inline unsigned int lru_count(struct lru *lru)
+{
+	return lru->count;
+}
+
+/*
+ * Insert a new entry into the lru.
+ */
+static void lru_insert(struct lru *lru, struct lru_entry *le)
+{
+	/*
+	 * Don't be tempted to set to 1, makes the lru aspect
+	 * perform poorly.
+	 */
+	atomic_set(&le->referenced, 0);
+
+	if (lru->cursor)
+		list_add_tail(&le->list, lru->cursor);
+
+	else {
+		INIT_LIST_HEAD(&le->list);
+		lru->cursor = &le->list;
+	}
+	lru->count++;
+}
+
+/*--------------*/
+
+/*
+ * Convert a list_head pointer to an lru_entry pointer.
+ */
+static inline struct lru_entry *to_le(struct list_head *l)
+{
+	return container_of(l, struct lru_entry, list);
+}
+
+/*
+ * Initialize an lru_iter and add it to the list of cursors in the lru.
+ */
+static void lru_iter_begin(struct lru *lru, struct lru_iter *it)
+{
+	it->lru = lru;
+	it->stop = lru->cursor ? to_le(lru->cursor->prev) : NULL;
+	it->e = lru->cursor ? to_le(lru->cursor) : NULL;
+	list_add(&it->list, &lru->iterators);
+}
+
+/*
+ * Remove an lru_iter from the list of cursors in the lru.
+ */
+static void lru_iter_end(struct lru_iter *it)
+{
+	list_del(&it->list);
+}
+
+/* Predicate function type to be used with lru_iter_next */
+typedef bool (*iter_predicate)(struct lru_entry *le, void *context);
+
+/*
+ * Advance the cursor to the next entry that passes the
+ * predicate, and return that entry.  Returns NULL if the
+ * iteration is complete.
+ */
+static struct lru_entry *lru_iter_next(struct lru_iter *it,
+				       iter_predicate pred, void *context)
+{
+	struct lru_entry *e;
+
+	while (it->e) {
+		e = it->e;
+
+		/* advance the cursor */
+		if (it->e == it->stop)
+			it->e = NULL;
+		else
+			it->e = to_le(it->e->list.next);
+
+		if (pred(e, context))
+			return e;
+	}
+
+	return NULL;
+}
+
+/*
+ * Invalidate a specific lru_entry and update all cursors in
+ * the lru accordingly.
+ */
+static void lru_iter_invalidate(struct lru *lru, struct lru_entry *e)
+{
+	struct lru_iter *it;
+
+	list_for_each_entry(it, &lru->iterators, list) {
+		/* Move c->e forwards if necc. */
+		if (it->e == e) {
+			it->e = to_le(it->e->list.next);
+			if (it->e == e)
+				it->e = NULL;
+		}
+
+		/* Move it->stop backwards if necc. */
+		if (it->stop == e) {
+			it->stop = to_le(it->stop->list.prev);
+			if (it->stop == e)
+				it->stop = NULL;
+		}
+	}
+}
+
+/*--------------*/
+
+/*
+ * Remove a specific entry from the lru.
+ */
+static void lru_remove(struct lru *lru, struct lru_entry *le)
+{
+	lru_iter_invalidate(lru, le);
+	if (lru->count == 1)
+		lru->cursor = NULL;
+	else {
+		if (lru->cursor == &le->list)
+			lru->cursor = lru->cursor->next;
+		list_del(&le->list);
+	}
+	lru->count--;
+}
+
+/*
+ * Mark as referenced.
+ */
+static inline void lru_reference(struct lru_entry *le)
+{
+	atomic_set(&le->referenced, 1);
+}
+
+/*--------------*/
+
+/*
+ * Remove the least recently used entry (approx), that passes the predicate.
+ * Returns NULL on failure.
+ */
+enum evict_result {
+	ER_EVICT,
+	ER_DONT_EVICT,
+	ER_STOP, /* stop looking for something to evict */
+};
+
+typedef enum evict_result (*le_predicate)(struct lru_entry *le, void *context);
+
+static struct lru_entry *lru_evict(struct lru *lru, le_predicate pred, void *context)
+{
+	unsigned int tested = 0;
+	struct list_head *h = lru->cursor;
+	struct lru_entry *le;
+
+	if (!h)
+		return NULL;
+	/*
+	 * In the worst case we have to loop around twice. Once to clear
+	 * the reference flags, and then again to discover the predicate
+	 * fails for all entries.
+	 */
+	while (tested < lru->count) {
+		le = container_of(h, struct lru_entry, list);
+
+		if (atomic_read(&le->referenced))
+			atomic_set(&le->referenced, 0);
+		else {
+			tested++;
+			switch (pred(le, context)) {
+			case ER_EVICT:
+				/*
+				 * Adjust the cursor, so we start the next
+				 * search from here.
+				 */
+				lru->cursor = le->list.next;
+				lru_remove(lru, le);
+				return le;
+
+			case ER_DONT_EVICT:
+				break;
+
+			case ER_STOP:
+				lru->cursor = le->list.next;
+				return NULL;
+			}
+		}
+
+		h = h->next;
+
+		cond_resched();
+	}
+
+	return NULL;
+}
+
+/*--------------------------------------------------------------*/
+
 /*
  * Buffer state bits.
  */
@@ -86,28 +327,37 @@ enum data_mode {
 };
 
 struct dm_buffer {
+	/* protected by the locks in buffer_cache */
 	struct rb_node node;
-	struct list_head lru_list;
-	struct list_head global_list;
 
+	/* immutable, so don't need protecting */
 	sector_t block;
 	void *data;
 	unsigned char data_mode;		/* DATA_MODE_* */
 
-	unsigned int accessed;
-	unsigned int hold_count;
+	/*
+	 * These two fields are used in isolation, so do not need
+	 * a surrounding lock. NOTE: attempting to use refcount_t
+	 * instead of atomic_t caused crashes!
+	 */
+	atomic_t hold_count;
 	unsigned long last_accessed;
 
+	/*
+	 * Everything else is protected by the mutex in
+	 * dm_bufio_client
+	 */
+	unsigned long state;
+	struct lru_entry lru;
 	unsigned char list_mode;		/* LIST_* */
 	blk_status_t read_error;
 	blk_status_t write_error;
-	unsigned long state;
 	unsigned int dirty_start;
 	unsigned int dirty_end;
 	unsigned int write_start;
 	unsigned int write_end;
-	struct dm_bufio_client *c;
 	struct list_head write_list;
+	struct dm_bufio_client *c;
 	void (*end_io)(struct dm_buffer *b, blk_status_t bs);
 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
 #define MAX_STACK 10
@@ -116,9 +366,579 @@ struct dm_buffer {
 #endif
 };
 
+/*--------------------------------------------------------------*/
+
+/*
+ * The buffer cache manages buffers, particularly:
+ *  - inc/dec of holder count
+ *  - setting the last_accessed field
+ *  - maintains clean/dirty state along with lru
+ *  - selecting buffers that match predicates
+ *
+ * It does *not* handle:
+ *  - allocation/freeing of buffers.
+ *  - IO
+ *  - Eviction or cache sizing.
+ *
+ * cache_get() and cache_put() are threadsafe, you do not need to
+ * protect these calls with a surrounding mutex.  All the other
+ * methods are not threadsafe; they do use locking primitives, but
+ * only enough to ensure get/put are threadsafe.
+ */
+
+#define NR_LOCKS 64
+#define LOCKS_MASK (NR_LOCKS - 1)
+
+struct tree_lock {
+	struct rw_semaphore lock;
+} ____cacheline_aligned_in_smp;
+
+struct buffer_cache {
+	/*
+	 * We spread entries across multiple trees to reduce contention
+	 * on the locks.
+	 */
+	struct tree_lock locks[NR_LOCKS];
+	struct rb_root roots[NR_LOCKS];
+	struct lru lru[LIST_SIZE];
+};
+
+/*
+ * We want to scatter the buffers across the different rbtrees
+ * to improve concurrency, but we also want to encourage the
+ * lock_history optimisation (below). Scattering runs of 16
+ * buffers seems to be a good compromise.
+ */
+static inline unsigned int cache_index(sector_t block)
+{
+	return block & LOCKS_MASK;
+}
+
+static inline void cache_read_lock(struct buffer_cache *bc, sector_t block)
+{
+	down_read(&bc->locks[cache_index(block)].lock);
+}
+
+static inline void cache_read_unlock(struct buffer_cache *bc, sector_t block)
+{
+	up_read(&bc->locks[cache_index(block)].lock);
+}
+
+static inline void cache_write_lock(struct buffer_cache *bc, sector_t block)
+{
+	down_write(&bc->locks[cache_index(block)].lock);
+}
+
+static inline void cache_write_unlock(struct buffer_cache *bc, sector_t block)
+{
+	up_write(&bc->locks[cache_index(block)].lock);
+}
+
+/*
+ * Sometimes we want to repeatedly get and drop locks as part of an iteration.
+ * This struct helps avoid redundant drop and gets of the same lock.
+ */
+struct lock_history {
+	struct buffer_cache *cache;
+	bool write;
+	unsigned int previous;
+};
+
+static void lh_init(struct lock_history *lh, struct buffer_cache *cache, bool write)
+{
+	lh->cache = cache;
+	lh->write = write;
+	lh->previous = NR_LOCKS;  /* indicates no previous */
+}
+
+static void __lh_lock(struct lock_history *lh, unsigned int index)
+{
+	if (lh->write)
+		down_write(&lh->cache->locks[index].lock);
+	else
+		down_read(&lh->cache->locks[index].lock);
+}
+
+static void __lh_unlock(struct lock_history *lh, unsigned int index)
+{
+	if (lh->write)
+		up_write(&lh->cache->locks[index].lock);
+	else
+		up_read(&lh->cache->locks[index].lock);
+}
+
+/*
+ * Make sure you call this since it will unlock the final lock.
+ */
+static void lh_exit(struct lock_history *lh)
+{
+	if (lh->previous != NR_LOCKS) {
+		__lh_unlock(lh, lh->previous);
+		lh->previous = NR_LOCKS;
+	}
+}
+
+/*
+ * Named 'next' because there is no corresponding
+ * 'up/unlock' call since it's done automatically.
+ */
+static void lh_next(struct lock_history *lh, sector_t b)
+{
+	unsigned int index = cache_index(b);
+
+	if (lh->previous != NR_LOCKS) {
+		if (lh->previous != index) {
+			__lh_unlock(lh, lh->previous);
+			__lh_lock(lh, index);
+			lh->previous = index;
+		}
+	} else {
+		__lh_lock(lh, index);
+		lh->previous = index;
+	}
+}
+
+static struct dm_buffer *le_to_buffer(struct lru_entry *le)
+{
+	return container_of(le, struct dm_buffer, lru);
+}
+
+static struct dm_buffer *list_to_buffer(struct list_head *l)
+{
+	struct lru_entry *le = list_entry(l, struct lru_entry, list);
+
+	if (!le)
+		return NULL;
+
+	return le_to_buffer(le);
+}
+
+static void cache_init(struct buffer_cache *bc)
+{
+	unsigned int i;
+
+	for (i = 0; i < NR_LOCKS; i++) {
+		init_rwsem(&bc->locks[i].lock);
+		bc->roots[i] = RB_ROOT;
+	}
+
+	lru_init(&bc->lru[LIST_CLEAN]);
+	lru_init(&bc->lru[LIST_DIRTY]);
+}
+
+static void cache_destroy(struct buffer_cache *bc)
+{
+	unsigned int i;
+
+	for (i = 0; i < NR_LOCKS; i++)
+		BUG_ON(!RB_EMPTY_ROOT(&bc->roots[i]));
+
+	lru_destroy(&bc->lru[LIST_CLEAN]);
+	lru_destroy(&bc->lru[LIST_DIRTY]);
+}
+
+/*
+ * not threadsafe, or racey depending how you look at it
+ */
+static unsigned long cache_count(struct buffer_cache *bc, int list_mode)
+{
+	return lru_count(&bc->lru[list_mode]);
+}
+
+/*--------------*/
+
+/*
+ * not threadsafe
+ */
+static unsigned long cache_total(struct buffer_cache *bc)
+{
+	return lru_count(&bc->lru[LIST_CLEAN]) +
+		lru_count(&bc->lru[LIST_DIRTY]);
+}
+
+/*--------------*/
+
+/*
+ * Gets a specific buffer, indexed by block.
+ * If the buffer is found then its holder count will be incremented and
+ * lru_reference will be called.
+ *
+ * threadsafe
+ */
+static struct dm_buffer *__cache_get(const struct rb_root *root, sector_t block)
+{
+	struct rb_node *n = root->rb_node;
+	struct dm_buffer *b;
+
+	while (n) {
+		b = container_of(n, struct dm_buffer, node);
+
+		if (b->block == block)
+			return b;
+
+		n = block < b->block ? n->rb_left : n->rb_right;
+	}
+
+	return NULL;
+}
+
+static void __cache_inc_buffer(struct dm_buffer *b)
+{
+	atomic_inc(&b->hold_count);
+	WRITE_ONCE(b->last_accessed, jiffies);
+}
+
+static struct dm_buffer *cache_get(struct buffer_cache *bc, sector_t block)
+{
+	struct dm_buffer *b;
+
+	cache_read_lock(bc, block);
+	b = __cache_get(&bc->roots[cache_index(block)], block);
+	if (b) {
+		lru_reference(&b->lru);
+		__cache_inc_buffer(b);
+	}
+	cache_read_unlock(bc, block);
+
+	return b;
+}
+
+/*--------------*/
+
+/*
+ * Returns true if the hold count hits zero.
+ * threadsafe
+ */
+static bool cache_put(struct buffer_cache *bc, struct dm_buffer *b)
+{
+	bool r;
+
+	cache_read_lock(bc, b->block);
+	BUG_ON(!atomic_read(&b->hold_count));
+	r = atomic_dec_and_test(&b->hold_count);
+	cache_read_unlock(bc, b->block);
+
+	return r;
+}
+
+/*--------------*/
+
+typedef enum evict_result (*b_predicate)(struct dm_buffer *, void *);
+
+/*
+ * Evicts a buffer based on a predicate.  The oldest buffer that
+ * matches the predicate will be selected.  In addition to the
+ * predicate the hold_count of the selected buffer will be zero.
+ */
+struct evict_wrapper {
+	struct lock_history *lh;
+	b_predicate pred;
+	void *context;
+};
+
+/*
+ * Wraps the buffer predicate turning it into an lru predicate.  Adds
+ * extra test for hold_count.
+ */
+static enum evict_result __evict_pred(struct lru_entry *le, void *context)
+{
+	struct evict_wrapper *w = context;
+	struct dm_buffer *b = le_to_buffer(le);
+
+	lh_next(w->lh, b->block);
+
+	if (atomic_read(&b->hold_count))
+		return ER_DONT_EVICT;
+
+	return w->pred(b, w->context);
+}
+
+static struct dm_buffer *__cache_evict(struct buffer_cache *bc, int list_mode,
+				       b_predicate pred, void *context,
+				       struct lock_history *lh)
+{
+	struct evict_wrapper w = {.lh = lh, .pred = pred, .context = context};
+	struct lru_entry *le;
+	struct dm_buffer *b;
+
+	le = lru_evict(&bc->lru[list_mode], __evict_pred, &w);
+	if (!le)
+		return NULL;
+
+	b = le_to_buffer(le);
+	/* __evict_pred will have locked the appropriate tree. */
+	rb_erase(&b->node, &bc->roots[cache_index(b->block)]);
+
+	return b;
+}
+
+static struct dm_buffer *cache_evict(struct buffer_cache *bc, int list_mode,
+				     b_predicate pred, void *context)
+{
+	struct dm_buffer *b;
+	struct lock_history lh;
+
+	lh_init(&lh, bc, true);
+	b = __cache_evict(bc, list_mode, pred, context, &lh);
+	lh_exit(&lh);
+
+	return b;
+}
+
+/*--------------*/
+
+/*
+ * Mark a buffer as clean or dirty. Not threadsafe.
+ */
+static void cache_mark(struct buffer_cache *bc, struct dm_buffer *b, int list_mode)
+{
+	cache_write_lock(bc, b->block);
+	if (list_mode != b->list_mode) {
+		lru_remove(&bc->lru[b->list_mode], &b->lru);
+		b->list_mode = list_mode;
+		lru_insert(&bc->lru[b->list_mode], &b->lru);
+	}
+	cache_write_unlock(bc, b->block);
+}
+
+/*--------------*/
+
+/*
+ * Runs through the lru associated with 'old_mode', if the predicate matches then
+ * it moves them to 'new_mode'.  Not threadsafe.
+ */
+static void __cache_mark_many(struct buffer_cache *bc, int old_mode, int new_mode,
+			      b_predicate pred, void *context, struct lock_history *lh)
+{
+	struct lru_entry *le;
+	struct dm_buffer *b;
+	struct evict_wrapper w = {.lh = lh, .pred = pred, .context = context};
+
+	while (true) {
+		le = lru_evict(&bc->lru[old_mode], __evict_pred, &w);
+		if (!le)
+			break;
+
+		b = le_to_buffer(le);
+		b->list_mode = new_mode;
+		lru_insert(&bc->lru[b->list_mode], &b->lru);
+	}
+}
+
+static void cache_mark_many(struct buffer_cache *bc, int old_mode, int new_mode,
+			    b_predicate pred, void *context)
+{
+	struct lock_history lh;
+
+	lh_init(&lh, bc, true);
+	__cache_mark_many(bc, old_mode, new_mode, pred, context, &lh);
+	lh_exit(&lh);
+}
+
+/*--------------*/
+
+/*
+ * Iterates through all clean or dirty entries calling a function for each
+ * entry.  The callback may terminate the iteration early.  Not threadsafe.
+ */
+
+/*
+ * Iterator functions should return one of these actions to indicate
+ * how the iteration should proceed.
+ */
+enum it_action {
+	IT_NEXT,
+	IT_COMPLETE,
+};
+
+typedef enum it_action (*iter_fn)(struct dm_buffer *b, void *context);
+
+static void __cache_iterate(struct buffer_cache *bc, int list_mode,
+			    iter_fn fn, void *context, struct lock_history *lh)
+{
+	struct lru *lru = &bc->lru[list_mode];
+	struct lru_entry *le, *first;
+
+	if (!lru->cursor)
+		return;
+
+	first = le = to_le(lru->cursor);
+	do {
+		struct dm_buffer *b = le_to_buffer(le);
+
+		lh_next(lh, b->block);
+
+		switch (fn(b, context)) {
+		case IT_NEXT:
+			break;
+
+		case IT_COMPLETE:
+			return;
+		}
+		cond_resched();
+
+		le = to_le(le->list.next);
+	} while (le != first);
+}
+
+static void cache_iterate(struct buffer_cache *bc, int list_mode,
+			  iter_fn fn, void *context)
+{
+	struct lock_history lh;
+
+	lh_init(&lh, bc, false);
+	__cache_iterate(bc, list_mode, fn, context, &lh);
+	lh_exit(&lh);
+}
+
+/*--------------*/
+
+/*
+ * Passes ownership of the buffer to the cache. Returns false if the
+ * buffer was already present (in which case ownership does not pass).
+ * eg, a race with another thread.
+ *
+ * Holder count should be 1 on insertion.
+ *
+ * Not threadsafe.
+ */
+static bool __cache_insert(struct rb_root *root, struct dm_buffer *b)
+{
+	struct rb_node **new = &root->rb_node, *parent = NULL;
+	struct dm_buffer *found;
+
+	while (*new) {
+		found = container_of(*new, struct dm_buffer, node);
+
+		if (found->block == b->block)
+			return false;
+
+		parent = *new;
+		new = b->block < found->block ?
+			&found->node.rb_left : &found->node.rb_right;
+	}
+
+	rb_link_node(&b->node, parent, new);
+	rb_insert_color(&b->node, root);
+
+	return true;
+}
+
+static bool cache_insert(struct buffer_cache *bc, struct dm_buffer *b)
+{
+	bool r;
+
+	BUG_ON(b->list_mode >= LIST_SIZE);
+
+	cache_write_lock(bc, b->block);
+	BUG_ON(atomic_read(&b->hold_count) != 1);
+	r = __cache_insert(&bc->roots[cache_index(b->block)], b);
+	if (r)
+		lru_insert(&bc->lru[b->list_mode], &b->lru);
+	cache_write_unlock(bc, b->block);
+
+	return r;
+}
+
+/*--------------*/
+
+/*
+ * Removes buffer from cache, ownership of the buffer passes back to the caller.
+ * Fails if the hold_count is not one (ie. the caller holds the only reference).
+ *
+ * Not threadsafe.
+ */
+static bool cache_remove(struct buffer_cache *bc, struct dm_buffer *b)
+{
+	bool r;
+
+	cache_write_lock(bc, b->block);
+
+	if (atomic_read(&b->hold_count) != 1)
+		r = false;
+
+	else {
+		r = true;
+		rb_erase(&b->node, &bc->roots[cache_index(b->block)]);
+		lru_remove(&bc->lru[b->list_mode], &b->lru);
+	}
+
+	cache_write_unlock(bc, b->block);
+
+	return r;
+}
+
+/*--------------*/
+
+typedef void (*b_release)(struct dm_buffer *);
+
+static struct dm_buffer *__find_next(struct rb_root *root, sector_t block)
+{
+	struct rb_node *n = root->rb_node;
+	struct dm_buffer *b;
+	struct dm_buffer *best = NULL;
+
+	while (n) {
+		b = container_of(n, struct dm_buffer, node);
+
+		if (b->block == block)
+			return b;
+
+		if (block <= b->block) {
+			n = n->rb_left;
+			best = b;
+		} else
+			n = n->rb_right;
+	}
+
+	return best;
+}
+
+static void __remove_range(struct buffer_cache *bc,
+			   struct rb_root *root,
+			   sector_t begin, sector_t end,
+			   b_predicate pred, b_release release)
+{
+	struct dm_buffer *b;
+
+	while (true) {
+		cond_resched();
+
+		b = __find_next(root, begin);
+		if (!b || (b->block >= end))
+			break;
+
+		begin = b->block + 1;
+
+		if (atomic_read(&b->hold_count))
+			continue;
+
+		if (pred(b, NULL) == ER_EVICT) {
+			rb_erase(&b->node, root);
+			lru_remove(&bc->lru[b->list_mode], &b->lru);
+			release(b);
+		}
+	}
+}
+
+static void cache_remove_range(struct buffer_cache *bc,
+			       sector_t begin, sector_t end,
+			       b_predicate pred, b_release release)
+{
+	unsigned int i;
+
+	for (i = 0; i < NR_LOCKS; i++) {
+		down_write(&bc->locks[i].lock);
+		__remove_range(bc, &bc->roots[i], begin, end, pred, release);
+		up_write(&bc->locks[i].lock);
+	}
+}
+
+/*----------------------------------------------------------------*/
+
 /*
  * Linking of buffers:
- *	All buffers are linked to buffer_tree with their node field.
+ *	All buffers are linked to buffer_cache with their node field.
  *
  *	Clean buffers that are not being written (B_WRITING not set)
  *	are linked to lru[LIST_CLEAN] with their lru_list field.
@@ -136,9 +956,6 @@ struct dm_bufio_client {
 	spinlock_t spinlock;
 	bool no_sleep;
 
-	struct list_head lru[LIST_SIZE];
-	unsigned long n_buffers[LIST_SIZE];
-
 	struct block_device *bdev;
 	unsigned int block_size;
 	s8 sectors_per_block_bits;
@@ -153,7 +970,7 @@ struct dm_bufio_client {
 
 	unsigned int minimum_buffers;
 
-	struct rb_root buffer_tree;
+	struct buffer_cache cache;
 	wait_queue_head_t free_buffer_wait;
 
 	sector_t start;
@@ -165,6 +982,11 @@ struct dm_bufio_client {
 	struct shrinker shrinker;
 	struct work_struct shrink_work;
 	atomic_long_t need_shrink;
+
+	/*
+	 * Used by global_cleanup to sort the clients list.
+	 */
+	unsigned long oldest_buffer;
 };
 
 static DEFINE_STATIC_KEY_FALSE(no_sleep_enabled);
@@ -181,14 +1003,6 @@ static void dm_bufio_lock(struct dm_bufio_client *c)
 		mutex_lock_nested(&c->lock, dm_bufio_in_request());
 }
 
-static int dm_bufio_trylock(struct dm_bufio_client *c)
-{
-	if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep)
-		return spin_trylock_bh(&c->spinlock);
-	else
-		return mutex_trylock(&c->lock);
-}
-
 static void dm_bufio_unlock(struct dm_bufio_client *c)
 {
 	if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep)
@@ -217,10 +1031,6 @@ static unsigned long dm_bufio_cache_size_latch;
 
 static DEFINE_SPINLOCK(global_spinlock);
 
-static LIST_HEAD(global_queue);
-
-static unsigned long global_num;
-
 /*
  * Buffers are freed after this timeout
  */
@@ -254,7 +1064,6 @@ static struct workqueue_struct *dm_bufio_wq;
 static struct delayed_work dm_bufio_cleanup_old_work;
 static struct work_struct dm_bufio_replacement_work;
 
-
 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
 static void buffer_record_stack(struct dm_buffer *b)
 {
@@ -262,78 +1071,6 @@ static void buffer_record_stack(struct dm_buffer *b)
 }
 #endif
 
-/*
- *----------------------------------------------------------------
- * A red/black tree acts as an index for all the buffers.
- *----------------------------------------------------------------
- */
-static struct dm_buffer *__find(struct dm_bufio_client *c, sector_t block)
-{
-	struct rb_node *n = c->buffer_tree.rb_node;
-	struct dm_buffer *b;
-
-	while (n) {
-		b = container_of(n, struct dm_buffer, node);
-
-		if (b->block == block)
-			return b;
-
-		n = block < b->block ? n->rb_left : n->rb_right;
-	}
-
-	return NULL;
-}
-
-static struct dm_buffer *__find_next(struct dm_bufio_client *c, sector_t block)
-{
-	struct rb_node *n = c->buffer_tree.rb_node;
-	struct dm_buffer *b;
-	struct dm_buffer *best = NULL;
-
-	while (n) {
-		b = container_of(n, struct dm_buffer, node);
-
-		if (b->block == block)
-			return b;
-
-		if (block <= b->block) {
-			n = n->rb_left;
-			best = b;
-		} else {
-			n = n->rb_right;
-		}
-	}
-
-	return best;
-}
-
-static void __insert(struct dm_bufio_client *c, struct dm_buffer *b)
-{
-	struct rb_node **new = &c->buffer_tree.rb_node, *parent = NULL;
-	struct dm_buffer *found;
-
-	while (*new) {
-		found = container_of(*new, struct dm_buffer, node);
-
-		if (found->block == b->block) {
-			BUG_ON(found != b);
-			return;
-		}
-
-		parent = *new;
-		new = b->block < found->block ?
-			&found->node.rb_left : &found->node.rb_right;
-	}
-
-	rb_link_node(&b->node, parent, new);
-	rb_insert_color(&b->node, &c->buffer_tree);
-}
-
-static void __remove(struct dm_bufio_client *c, struct dm_buffer *b)
-{
-	rb_erase(&b->node, &c->buffer_tree);
-}
-
 /*----------------------------------------------------------------*/
 
 static void adjust_total_allocated(struct dm_buffer *b, bool unlink)
@@ -361,16 +1098,9 @@ static void adjust_total_allocated(struct dm_buffer *b, bool unlink)
 	if (dm_bufio_current_allocated > dm_bufio_peak_allocated)
 		dm_bufio_peak_allocated = dm_bufio_current_allocated;
 
-	b->accessed = 1;
-
 	if (!unlink) {
-		list_add(&b->global_list, &global_queue);
-		global_num++;
 		if (dm_bufio_current_allocated > dm_bufio_cache_size)
 			queue_work(dm_bufio_wq, &dm_bufio_replacement_work);
-	} else {
-		list_del(&b->global_list);
-		global_num--;
 	}
 
 	spin_unlock(&global_spinlock);
@@ -486,8 +1216,9 @@ static void free_buffer_data(struct dm_bufio_client *c,
  */
 static struct dm_buffer *alloc_buffer(struct dm_bufio_client *c, gfp_t gfp_mask)
 {
-	struct dm_buffer *b = kmem_cache_alloc(c->slab_buffer, gfp_mask);
+	struct dm_buffer *b;
 
+	b = kmem_cache_alloc(c->slab_buffer, gfp_mask);
 	if (!b)
 		return NULL;
 
@@ -498,6 +1229,7 @@ static struct dm_buffer *alloc_buffer(struct dm_bufio_client *c, gfp_t gfp_mask)
 		kmem_cache_free(c->slab_buffer, b);
 		return NULL;
 	}
+	adjust_total_allocated(b, false);
 
 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
 	b->stack_len = 0;
@@ -512,61 +1244,11 @@ static void free_buffer(struct dm_buffer *b)
 {
 	struct dm_bufio_client *c = b->c;
 
+	adjust_total_allocated(b, true);
 	free_buffer_data(c, b->data, b->data_mode);
 	kmem_cache_free(c->slab_buffer, b);
 }
 
-/*
- * Link buffer to the buffer tree and clean or dirty queue.
- */
-static void __link_buffer(struct dm_buffer *b, sector_t block, int dirty)
-{
-	struct dm_bufio_client *c = b->c;
-
-	c->n_buffers[dirty]++;
-	b->block = block;
-	b->list_mode = dirty;
-	list_add(&b->lru_list, &c->lru[dirty]);
-	__insert(b->c, b);
-	b->last_accessed = jiffies;
-
-	adjust_total_allocated(b, false);
-}
-
-/*
- * Unlink buffer from the buffer tree and dirty or clean queue.
- */
-static void __unlink_buffer(struct dm_buffer *b)
-{
-	struct dm_bufio_client *c = b->c;
-
-	BUG_ON(!c->n_buffers[b->list_mode]);
-
-	c->n_buffers[b->list_mode]--;
-	__remove(b->c, b);
-	list_del(&b->lru_list);
-
-	adjust_total_allocated(b, true);
-}
-
-/*
- * Place the buffer to the head of dirty or clean LRU queue.
- */
-static void __relink_lru(struct dm_buffer *b, int dirty)
-{
-	struct dm_bufio_client *c = b->c;
-
-	b->accessed = 1;
-
-	BUG_ON(!c->n_buffers[b->list_mode]);
-
-	c->n_buffers[b->list_mode]--;
-	c->n_buffers[dirty]++;
-	b->list_mode = dirty;
-	list_move(&b->lru_list, &c->lru[dirty]);
-	b->last_accessed = jiffies;
-}
-
 /*
  *--------------------------------------------------------------------------
  * Submit I/O on the buffer.
@@ -806,7 +1488,7 @@ static void __flush_write_list(struct list_head *write_list)
  */
 static void __make_buffer_clean(struct dm_buffer *b)
 {
-	BUG_ON(b->hold_count);
+	BUG_ON(atomic_read(&b->hold_count));
 
 	/* smp_load_acquire() pairs with read_endio()'s smp_mb__before_atomic() */
 	if (!smp_load_acquire(&b->state))	/* fast case */
@@ -817,6 +1499,26 @@ static void __make_buffer_clean(struct dm_buffer *b)
 	wait_on_bit_io(&b->state, B_WRITING, TASK_UNINTERRUPTIBLE);
 }
 
+static enum evict_result is_clean(struct dm_buffer *b, void *context)
+{
+	struct dm_bufio_client *c = context;
+
+	BUG_ON(test_bit(B_WRITING, &b->state));
+	BUG_ON(test_bit(B_DIRTY, &b->state));
+
+	if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep &&
+	    unlikely(test_bit(B_READING, &b->state)))
+		return ER_DONT_EVICT;
+
+	return ER_EVICT;
+}
+
+static enum evict_result is_dirty(struct dm_buffer *b, void *context)
+{
+	BUG_ON(test_bit(B_READING, &b->state));
+	return ER_EVICT;
+}
+
 /*
  * Find some buffer that is not held by anybody, clean it, unlink it and
  * return it.
@@ -825,34 +1527,20 @@ static struct dm_buffer *__get_unclaimed_buffer(struct dm_bufio_client *c)
 {
 	struct dm_buffer *b;
 
-	list_for_each_entry_reverse(b, &c->lru[LIST_CLEAN], lru_list) {
-		BUG_ON(test_bit(B_WRITING, &b->state));
-		BUG_ON(test_bit(B_DIRTY, &b->state));
-
-		if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep &&
-		    unlikely(test_bit_acquire(B_READING, &b->state)))
-			continue;
-
-		if (!b->hold_count) {
-			__make_buffer_clean(b);
-			__unlink_buffer(b);
-			return b;
-		}
-		cond_resched();
+	b = cache_evict(&c->cache, LIST_CLEAN, is_clean, c);
+	if (b) {
+		/* this also waits for pending reads */
+		__make_buffer_clean(b);
+		return b;
 	}
 
 	if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep)
 		return NULL;
 
-	list_for_each_entry_reverse(b, &c->lru[LIST_DIRTY], lru_list) {
-		BUG_ON(test_bit(B_READING, &b->state));
-
-		if (!b->hold_count) {
-			__make_buffer_clean(b);
-			__unlink_buffer(b);
-			return b;
-		}
-		cond_resched();
+	b = cache_evict(&c->cache, LIST_DIRTY, is_dirty, NULL);
+	if (b) {
+		__make_buffer_clean(b);
+		return b;
 	}
 
 	return NULL;
@@ -873,7 +1561,12 @@ static void __wait_for_free_buffer(struct dm_bufio_client *c)
 	set_current_state(TASK_UNINTERRUPTIBLE);
 	dm_bufio_unlock(c);
 
-	io_schedule();
+	/*
+	 * It's possible to miss a wake up event since we don't always
+	 * hold c->lock when wake_up is called.  So we have a timeout here,
+	 * just in case.
+	 */
+	io_schedule_timeout(5 * HZ);
 
 	remove_wait_queue(&c->free_buffer_wait, &wait);
 
@@ -931,9 +1624,8 @@ static struct dm_buffer *__alloc_buffer_wait_no_callback(struct dm_bufio_client
 		}
 
 		if (!list_empty(&c->reserved_buffers)) {
-			b = list_entry(c->reserved_buffers.next,
-				       struct dm_buffer, lru_list);
-			list_del(&b->lru_list);
+			b = list_to_buffer(c->reserved_buffers.next);
+			list_del(&b->lru.list);
 			c->need_reserved_buffers++;
 
 			return b;
@@ -967,36 +1659,55 @@ static void __free_buffer_wake(struct dm_buffer *b)
 {
 	struct dm_bufio_client *c = b->c;
 
+	b->block = -1;
 	if (!c->need_reserved_buffers)
 		free_buffer(b);
 	else {
-		list_add(&b->lru_list, &c->reserved_buffers);
+		list_add(&b->lru.list, &c->reserved_buffers);
 		c->need_reserved_buffers--;
 	}
 
 	wake_up(&c->free_buffer_wait);
 }
 
+static enum evict_result cleaned(struct dm_buffer *b, void *context)
+{
+	BUG_ON(test_bit(B_READING, &b->state));
+
+	if (test_bit(B_DIRTY, &b->state) || test_bit(B_WRITING, &b->state))
+		return ER_DONT_EVICT;
+	else
+		return ER_EVICT;
+}
+
+static void __move_clean_buffers(struct dm_bufio_client *c)
+{
+	cache_mark_many(&c->cache, LIST_DIRTY, LIST_CLEAN, cleaned, NULL);
+}
+
+struct write_context {
+	int no_wait;
+	struct list_head *write_list;
+};
+
+static enum it_action write_one(struct dm_buffer *b, void *context)
+{
+	struct write_context *wc = context;
+
+	if (wc->no_wait && test_bit(B_WRITING, &b->state))
+		return IT_COMPLETE;
+
+	__write_dirty_buffer(b, wc->write_list);
+	return IT_NEXT;
+}
+
 static void __write_dirty_buffers_async(struct dm_bufio_client *c, int no_wait,
 					struct list_head *write_list)
 {
-	struct dm_buffer *b, *tmp;
+	struct write_context wc = {.no_wait = no_wait, .write_list = write_list};
 
-	list_for_each_entry_safe_reverse(b, tmp, &c->lru[LIST_DIRTY], lru_list) {
-		BUG_ON(test_bit(B_READING, &b->state));
-
-		if (!test_bit(B_DIRTY, &b->state) &&
-		    !test_bit(B_WRITING, &b->state)) {
-			__relink_lru(b, LIST_CLEAN);
-			continue;
-		}
-
-		if (no_wait && test_bit(B_WRITING, &b->state))
-			return;
-
-		__write_dirty_buffer(b, write_list);
-		cond_resched();
-	}
+	__move_clean_buffers(c);
+	cache_iterate(&c->cache, LIST_DIRTY, write_one, &wc);
 }
 
 /*
@@ -1007,7 +1718,8 @@ static void __write_dirty_buffers_async(struct dm_bufio_client *c, int no_wait,
 static void __check_watermark(struct dm_bufio_client *c,
 			      struct list_head *write_list)
 {
-	if (c->n_buffers[LIST_DIRTY] > c->n_buffers[LIST_CLEAN] * DM_BUFIO_WRITEBACK_RATIO)
+	if (cache_count(&c->cache, LIST_DIRTY) >
+		cache_count(&c->cache, LIST_CLEAN) * DM_BUFIO_WRITEBACK_RATIO)
 		__write_dirty_buffers_async(c, 1, write_list);
 }
 
@@ -1017,20 +1729,30 @@ static void __check_watermark(struct dm_bufio_client *c,
  *--------------------------------------------------------------
  */
 
+static void cache_put_and_wake(struct dm_bufio_client *c, struct dm_buffer *b)
+{
+	/*
+	 * Relying on waitqueue_active() is racey, but we sleep
+	 * with schedule_timeout anyway.
+	 */
+	if (cache_put(&c->cache, b) &&
+	    unlikely(waitqueue_active(&c->free_buffer_wait)))
+		wake_up(&c->free_buffer_wait);
+}
+
+/*
+ * This assumes you have already checked the cache to see if the buffer
+ * is already present (it will recheck after dropping the lock for allocation).
+ */
 static struct dm_buffer *__bufio_new(struct dm_bufio_client *c, sector_t block,
 				     enum new_flag nf, int *need_submit,
 				     struct list_head *write_list)
 {
 	struct dm_buffer *b, *new_b = NULL;
-
 	*need_submit = 0;
 
-	b = __find(c, block);
-	if (b)
-		goto found_buffer;
-
-	if (nf == NF_GET)
-		return NULL;
+	/* This can't be called with NF_GET */
+	BUG_ON(nf == NF_GET);
 
 	new_b = __alloc_buffer_wait(c, nf);
 	if (!new_b)
@@ -1040,7 +1762,7 @@ static struct dm_buffer *__bufio_new(struct dm_bufio_client *c, sector_t block,
 	 * We've had a period where the mutex was unlocked, so need to
 	 * recheck the buffer tree.
 	 */
-	b = __find(c, block);
+	b = cache_get(&c->cache, block);
 	if (b) {
 		__free_buffer_wake(new_b);
 		goto found_buffer;
@@ -1049,24 +1771,35 @@ static struct dm_buffer *__bufio_new(struct dm_bufio_client *c, sector_t block,
 	__check_watermark(c, write_list);
 
 	b = new_b;
-	b->hold_count = 1;
+	atomic_set(&b->hold_count, 1);
+	WRITE_ONCE(b->last_accessed, jiffies);
+	b->block = block;
 	b->read_error = 0;
 	b->write_error = 0;
-	__link_buffer(b, block, LIST_CLEAN);
+	b->list_mode = LIST_CLEAN;
 
-	if (nf == NF_FRESH) {
+	if (nf == NF_FRESH)
 		b->state = 0;
-		return b;
+	else {
+		b->state = 1 << B_READING;
+		*need_submit = 1;
 	}
 
-	b->state = 1 << B_READING;
-	*need_submit = 1;
+	/*
+	 * We mustn't insert into the cache until the B_READING state
+	 * is set.  Otherwise another thread could get it and use
+	 * it before it had been read.
+	 */
+	cache_insert(&c->cache, b);
 
 	return b;
 
 found_buffer:
-	if (nf == NF_PREFETCH)
+	if (nf == NF_PREFETCH) {
+		cache_put_and_wake(c, b);
 		return NULL;
+	}
+
 	/*
 	 * Note: it is essential that we don't wait for the buffer to be
 	 * read if dm_bufio_get function is used. Both dm_bufio_get and
@@ -1074,12 +1807,11 @@ static struct dm_buffer *__bufio_new(struct dm_bufio_client *c, sector_t block,
 	 * If the user called both dm_bufio_prefetch and dm_bufio_get on
 	 * the same buffer, it would deadlock if we waited.
 	 */
-	if (nf == NF_GET && unlikely(test_bit_acquire(B_READING, &b->state)))
+	if (nf == NF_GET && unlikely(test_bit_acquire(B_READING, &b->state))) {
+		cache_put_and_wake(c, b);
 		return NULL;
+	}
 
-	b->hold_count++;
-	__relink_lru(b, test_bit(B_DIRTY, &b->state) ||
-		     test_bit(B_WRITING, &b->state));
 	return b;
 }
 
@@ -1109,18 +1841,49 @@ static void read_endio(struct dm_buffer *b, blk_status_t status)
 static void *new_read(struct dm_bufio_client *c, sector_t block,
 		      enum new_flag nf, struct dm_buffer **bp)
 {
-	int need_submit;
+	int need_submit = 0;
 	struct dm_buffer *b;
 
 	LIST_HEAD(write_list);
+	*bp = NULL;
+
+	/*
+	 * Fast path, hopefully the block is already in the cache.  No need
+	 * to get the client lock for this.
+	 */
+	b = cache_get(&c->cache, block);
+	if (b) {
+		if (nf == NF_PREFETCH) {
+			cache_put_and_wake(c, b);
+			return NULL;
+		}
+
+		/*
+		 * Note: it is essential that we don't wait for the buffer to be
+		 * read if dm_bufio_get function is used. Both dm_bufio_get and
+		 * dm_bufio_prefetch can be used in the driver request routine.
+		 * If the user called both dm_bufio_prefetch and dm_bufio_get on
+		 * the same buffer, it would deadlock if we waited.
+		 */
+		if (nf == NF_GET && unlikely(test_bit_acquire(B_READING, &b->state))) {
+			cache_put_and_wake(c, b);
+			return NULL;
+		}
+	}
+
+	if (!b) {
+		if (nf == NF_GET)
+			return NULL;
+
+		dm_bufio_lock(c);
+		b = __bufio_new(c, block, nf, &need_submit, &write_list);
+		dm_bufio_unlock(c);
+	}
 
-	dm_bufio_lock(c);
-	b = __bufio_new(c, block, nf, &need_submit, &write_list);
 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
-	if (b && b->hold_count == 1)
+	if (b && (atomic_read(&b->hold_count) == 1))
 		buffer_record_stack(b);
 #endif
-	dm_bufio_unlock(c);
 
 	__flush_write_list(&write_list);
 
@@ -1141,7 +1904,6 @@ static void *new_read(struct dm_bufio_client *c, sector_t block,
 	}
 
 	*bp = b;
-
 	return b->data;
 }
 
@@ -1156,7 +1918,6 @@ void *dm_bufio_read(struct dm_bufio_client *c, sector_t block,
 		    struct dm_buffer **bp)
 {
 	BUG_ON(dm_bufio_in_request());
-
 	return new_read(c, block, NF_READ, bp);
 }
 EXPORT_SYMBOL_GPL(dm_bufio_read);
@@ -1165,7 +1926,6 @@ void *dm_bufio_new(struct dm_bufio_client *c, sector_t block,
 		   struct dm_buffer **bp)
 {
 	BUG_ON(dm_bufio_in_request());
-
 	return new_read(c, block, NF_FRESH, bp);
 }
 EXPORT_SYMBOL_GPL(dm_bufio_new);
@@ -1180,12 +1940,19 @@ void dm_bufio_prefetch(struct dm_bufio_client *c,
 	BUG_ON(dm_bufio_in_request());
 
 	blk_start_plug(&plug);
-	dm_bufio_lock(c);
 
 	for (; n_blocks--; block++) {
 		int need_submit;
 		struct dm_buffer *b;
 
+		b = cache_get(&c->cache, block);
+		if (b) {
+			/* already in cache */
+			cache_put_and_wake(c, b);
+			continue;
+		}
+
+		dm_bufio_lock(c);
 		b = __bufio_new(c, block, NF_PREFETCH, &need_submit,
 				&write_list);
 		if (unlikely(!list_empty(&write_list))) {
@@ -1195,6 +1962,7 @@ void dm_bufio_prefetch(struct dm_bufio_client *c,
 			blk_start_plug(&plug);
 			dm_bufio_lock(c);
 		}
+
 		if (unlikely(b != NULL)) {
 			dm_bufio_unlock(c);
 
@@ -1208,10 +1976,9 @@ void dm_bufio_prefetch(struct dm_bufio_client *c,
 				goto flush_plug;
 			dm_bufio_lock(c);
 		}
+		dm_bufio_unlock(c);
 	}
 
-	dm_bufio_unlock(c);
-
 flush_plug:
 	blk_finish_plug(&plug);
 }
@@ -1221,29 +1988,28 @@ void dm_bufio_release(struct dm_buffer *b)
 {
 	struct dm_bufio_client *c = b->c;
 
-	dm_bufio_lock(c);
+	/*
+	 * If there were errors on the buffer, and the buffer is not
+	 * to be written, free the buffer. There is no point in caching
+	 * invalid buffer.
+	 */
+	if ((b->read_error || b->write_error) &&
+	    !test_bit_acquire(B_READING, &b->state) &&
+	    !test_bit(B_WRITING, &b->state) &&
+	    !test_bit(B_DIRTY, &b->state)) {
+		dm_bufio_lock(c);
 
-	BUG_ON(!b->hold_count);
-
-	b->hold_count--;
-	if (!b->hold_count) {
-		wake_up(&c->free_buffer_wait);
-
-		/*
-		 * If there were errors on the buffer, and the buffer is not
-		 * to be written, free the buffer. There is no point in caching
-		 * invalid buffer.
-		 */
-		if ((b->read_error || b->write_error) &&
-		    !test_bit_acquire(B_READING, &b->state) &&
-		    !test_bit(B_WRITING, &b->state) &&
-		    !test_bit(B_DIRTY, &b->state)) {
-			__unlink_buffer(b);
+		/* cache remove can fail if there are other holders */
+		if (cache_remove(&c->cache, b)) {
 			__free_buffer_wake(b);
+			dm_bufio_unlock(c);
+			return;
 		}
+
+		dm_bufio_unlock(c);
 	}
 
-	dm_bufio_unlock(c);
+	cache_put_and_wake(c, b);
 }
 EXPORT_SYMBOL_GPL(dm_bufio_release);
 
@@ -1262,7 +2028,7 @@ void dm_bufio_mark_partial_buffer_dirty(struct dm_buffer *b,
 	if (!test_and_set_bit(B_DIRTY, &b->state)) {
 		b->dirty_start = start;
 		b->dirty_end = end;
-		__relink_lru(b, LIST_DIRTY);
+		cache_mark(&c->cache, b, LIST_DIRTY);
 	} else {
 		if (start < b->dirty_start)
 			b->dirty_start = start;
@@ -1300,11 +2066,19 @@ EXPORT_SYMBOL_GPL(dm_bufio_write_dirty_buffers_async);
  *
  * Finally, we flush hardware disk cache.
  */
+static bool is_writing(struct lru_entry *e, void *context)
+{
+	struct dm_buffer *b = le_to_buffer(e);
+
+	return test_bit(B_WRITING, &b->state);
+}
+
 int dm_bufio_write_dirty_buffers(struct dm_bufio_client *c)
 {
 	int a, f;
-	unsigned long buffers_processed = 0;
-	struct dm_buffer *b, *tmp;
+	unsigned long nr_buffers;
+	struct lru_entry *e;
+	struct lru_iter it;
 
 	LIST_HEAD(write_list);
 
@@ -1314,52 +2088,32 @@ int dm_bufio_write_dirty_buffers(struct dm_bufio_client *c)
 	__flush_write_list(&write_list);
 	dm_bufio_lock(c);
 
-again:
-	list_for_each_entry_safe_reverse(b, tmp, &c->lru[LIST_DIRTY], lru_list) {
-		int dropped_lock = 0;
-
-		if (buffers_processed < c->n_buffers[LIST_DIRTY])
-			buffers_processed++;
+	nr_buffers = cache_count(&c->cache, LIST_DIRTY);
+	lru_iter_begin(&c->cache.lru[LIST_DIRTY], &it);
+	while ((e = lru_iter_next(&it, is_writing, c))) {
+		struct dm_buffer *b = le_to_buffer(e);
+		__cache_inc_buffer(b);
 
 		BUG_ON(test_bit(B_READING, &b->state));
 
-		if (test_bit(B_WRITING, &b->state)) {
-			if (buffers_processed < c->n_buffers[LIST_DIRTY]) {
-				dropped_lock = 1;
-				b->hold_count++;
-				dm_bufio_unlock(c);
-				wait_on_bit_io(&b->state, B_WRITING,
-					       TASK_UNINTERRUPTIBLE);
-				dm_bufio_lock(c);
-				b->hold_count--;
-			} else
-				wait_on_bit_io(&b->state, B_WRITING,
-					       TASK_UNINTERRUPTIBLE);
-		}
+		if (nr_buffers) {
+			nr_buffers--;
+			dm_bufio_unlock(c);
+			wait_on_bit_io(&b->state, B_WRITING, TASK_UNINTERRUPTIBLE);
+			dm_bufio_lock(c);
+		} else
+			wait_on_bit_io(&b->state, B_WRITING, TASK_UNINTERRUPTIBLE);
 
-		if (!test_bit(B_DIRTY, &b->state) &&
-		    !test_bit(B_WRITING, &b->state))
-			__relink_lru(b, LIST_CLEAN);
+		if (!test_bit(B_DIRTY, &b->state) && !test_bit(B_WRITING, &b->state))
+			cache_mark(&c->cache, b, LIST_CLEAN);
+
+		cache_put_and_wake(c, b);
 
 		cond_resched();
-
-		/*
-		 * If we dropped the lock, the list is no longer consistent,
-		 * so we must restart the search.
-		 *
-		 * In the most common case, the buffer just processed is
-		 * relinked to the clean list, so we won't loop scanning the
-		 * same buffer again and again.
-		 *
-		 * This may livelock if there is another thread simultaneously
-		 * dirtying buffers, so we count the number of buffers walked
-		 * and if it exceeds the total number of buffers, it means that
-		 * someone is doing some writes simultaneously with us.  In
-		 * this case, stop, dropping the lock.
-		 */
-		if (dropped_lock)
-			goto again;
 	}
+
+	lru_iter_end(&it);
+
 	wake_up(&c->free_buffer_wait);
 	dm_bufio_unlock(c);
 
@@ -1418,12 +2172,22 @@ int dm_bufio_issue_discard(struct dm_bufio_client *c, sector_t block, sector_t c
 }
 EXPORT_SYMBOL_GPL(dm_bufio_issue_discard);
 
-static void forget_buffer_locked(struct dm_buffer *b)
+static bool forget_buffer(struct dm_bufio_client *c, sector_t block)
 {
-	if (likely(!b->hold_count) && likely(!smp_load_acquire(&b->state))) {
-		__unlink_buffer(b);
-		__free_buffer_wake(b);
+	struct dm_buffer *b;
+
+	b = cache_get(&c->cache, block);
+	if (b) {
+		if (likely(!smp_load_acquire(&b->state))) {
+			if (cache_remove(&c->cache, b))
+				__free_buffer_wake(b);
+			else
+				cache_put_and_wake(c, b);
+		} else
+			cache_put_and_wake(c, b);
 	}
+
+	return b ? true : false;
 }
 
 /*
@@ -1434,38 +2198,22 @@ static void forget_buffer_locked(struct dm_buffer *b)
  */
 void dm_bufio_forget(struct dm_bufio_client *c, sector_t block)
 {
-	struct dm_buffer *b;
-
 	dm_bufio_lock(c);
-
-	b = __find(c, block);
-	if (b)
-		forget_buffer_locked(b);
-
+	forget_buffer(c, block);
 	dm_bufio_unlock(c);
 }
 EXPORT_SYMBOL_GPL(dm_bufio_forget);
 
+static enum evict_result idle(struct dm_buffer *b, void *context)
+{
+	return b->state ? ER_DONT_EVICT : ER_EVICT;
+}
+
 void dm_bufio_forget_buffers(struct dm_bufio_client *c, sector_t block, sector_t n_blocks)
 {
-	struct dm_buffer *b;
-	sector_t end_block = block + n_blocks;
-
-	while (block < end_block) {
-		dm_bufio_lock(c);
-
-		b = __find_next(c, block);
-		if (b) {
-			block = b->block + 1;
-			forget_buffer_locked(b);
-		}
-
-		dm_bufio_unlock(c);
-
-		if (!b)
-			break;
-	}
-
+	dm_bufio_lock(c);
+	cache_remove_range(&c->cache, block, block + n_blocks, idle, __free_buffer_wake);
+	dm_bufio_unlock(c);
 }
 EXPORT_SYMBOL_GPL(dm_bufio_forget_buffers);
 
@@ -1527,11 +2275,26 @@ struct dm_bufio_client *dm_bufio_get_client(struct dm_buffer *b)
 }
 EXPORT_SYMBOL_GPL(dm_bufio_get_client);
 
+static enum it_action warn_leak(struct dm_buffer *b, void *context)
+{
+	bool *warned = context;
+
+	WARN_ON(!(*warned));
+	*warned = true;
+	DMERR("leaked buffer %llx, hold count %u, list %d",
+	      (unsigned long long)b->block, atomic_read(&b->hold_count), b->list_mode);
+#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
+	stack_trace_print(b->stack_entries, b->stack_len, 1);
+	/* mark unclaimed to avoid BUG_ON below */
+	atomic_set(&b->hold_count, 0);
+#endif
+	return IT_NEXT;
+}
+
 static void drop_buffers(struct dm_bufio_client *c)
 {
-	struct dm_buffer *b;
 	int i;
-	bool warned = false;
+	struct dm_buffer *b;
 
 	BUG_ON(dm_bufio_in_request());
 
@@ -1545,18 +2308,11 @@ static void drop_buffers(struct dm_bufio_client *c)
 	while ((b = __get_unclaimed_buffer(c)))
 		__free_buffer_wake(b);
 
-	for (i = 0; i < LIST_SIZE; i++)
-		list_for_each_entry(b, &c->lru[i], lru_list) {
-			WARN_ON(!warned);
-			warned = true;
-			DMERR("leaked buffer %llx, hold count %u, list %d",
-			      (unsigned long long)b->block, b->hold_count, i);
-#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
-			stack_trace_print(b->stack_entries, b->stack_len, 1);
-			/* mark unclaimed to avoid BUG_ON below */
-			b->hold_count = 0;
-#endif
-		}
+	for (i = 0; i < LIST_SIZE; i++) {
+		bool warned = false;
+
+		cache_iterate(&c->cache, i, warn_leak, &warned);
+	}
 
 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
 	while ((b = __get_unclaimed_buffer(c)))
@@ -1564,39 +2320,11 @@ static void drop_buffers(struct dm_bufio_client *c)
 #endif
 
 	for (i = 0; i < LIST_SIZE; i++)
-		BUG_ON(!list_empty(&c->lru[i]));
+		BUG_ON(cache_count(&c->cache, i));
 
 	dm_bufio_unlock(c);
 }
 
-/*
- * We may not be able to evict this buffer if IO pending or the client
- * is still using it.  Caller is expected to know buffer is too old.
- *
- * And if GFP_NOFS is used, we must not do any I/O because we hold
- * dm_bufio_clients_lock and we would risk deadlock if the I/O gets
- * rerouted to different bufio client.
- */
-static bool __try_evict_buffer(struct dm_buffer *b, gfp_t gfp)
-{
-	if (!(gfp & __GFP_FS) ||
-	    (static_branch_unlikely(&no_sleep_enabled) && b->c->no_sleep)) {
-		if (test_bit_acquire(B_READING, &b->state) ||
-		    test_bit(B_WRITING, &b->state) ||
-		    test_bit(B_DIRTY, &b->state))
-			return false;
-	}
-
-	if (b->hold_count)
-		return false;
-
-	__make_buffer_clean(b);
-	__unlink_buffer(b);
-	__free_buffer_wake(b);
-
-	return true;
-}
-
 static unsigned long get_retain_buffers(struct dm_bufio_client *c)
 {
 	unsigned long retain_bytes = READ_ONCE(dm_bufio_retain_bytes);
@@ -1612,22 +2340,30 @@ static unsigned long get_retain_buffers(struct dm_bufio_client *c)
 static void __scan(struct dm_bufio_client *c)
 {
 	int l;
-	struct dm_buffer *b, *tmp;
+	struct dm_buffer *b;
 	unsigned long freed = 0;
-	unsigned long count = c->n_buffers[LIST_CLEAN] +
-			      c->n_buffers[LIST_DIRTY];
 	unsigned long retain_target = get_retain_buffers(c);
+	unsigned long count = cache_total(&c->cache);
 
 	for (l = 0; l < LIST_SIZE; l++) {
-		list_for_each_entry_safe_reverse(b, tmp, &c->lru[l], lru_list) {
+		while (true) {
 			if (count - freed <= retain_target)
 				atomic_long_set(&c->need_shrink, 0);
+
 			if (!atomic_long_read(&c->need_shrink))
-				return;
-			if (__try_evict_buffer(b, GFP_KERNEL)) {
-				atomic_long_dec(&c->need_shrink);
-				freed++;
-			}
+				break;
+
+			b = cache_evict(&c->cache, l,
+					l == LIST_CLEAN ? is_clean : is_dirty, c);
+			if (!b)
+				break;
+
+			BUG_ON(b->list_mode != l);
+			__make_buffer_clean(b);
+			__free_buffer_wake(b);
+
+			atomic_long_dec(&c->need_shrink);
+			freed++;
 			cond_resched();
 		}
 	}
@@ -1656,8 +2392,7 @@ static unsigned long dm_bufio_shrink_scan(struct shrinker *shrink, struct shrink
 static unsigned long dm_bufio_shrink_count(struct shrinker *shrink, struct shrink_control *sc)
 {
 	struct dm_bufio_client *c = container_of(shrink, struct dm_bufio_client, shrinker);
-	unsigned long count = READ_ONCE(c->n_buffers[LIST_CLEAN]) +
-			      READ_ONCE(c->n_buffers[LIST_DIRTY]);
+	unsigned long count = cache_total(&c->cache);
 	unsigned long retain_target = get_retain_buffers(c);
 	unsigned long queued_for_cleanup = atomic_long_read(&c->need_shrink);
 
@@ -1685,7 +2420,6 @@ struct dm_bufio_client *dm_bufio_client_create(struct block_device *bdev, unsign
 {
 	int r;
 	struct dm_bufio_client *c;
-	unsigned int i;
 	char slab_name[27];
 
 	if (!block_size || block_size & ((1 << SECTOR_SHIFT) - 1)) {
@@ -1699,7 +2433,7 @@ struct dm_bufio_client *dm_bufio_client_create(struct block_device *bdev, unsign
 		r = -ENOMEM;
 		goto bad_client;
 	}
-	c->buffer_tree = RB_ROOT;
+	cache_init(&c->cache);
 
 	c->bdev = bdev;
 	c->block_size = block_size;
@@ -1716,11 +2450,6 @@ struct dm_bufio_client *dm_bufio_client_create(struct block_device *bdev, unsign
 		static_branch_inc(&no_sleep_enabled);
 	}
 
-	for (i = 0; i < LIST_SIZE; i++) {
-		INIT_LIST_HEAD(&c->lru[i]);
-		c->n_buffers[i] = 0;
-	}
-
 	mutex_init(&c->lock);
 	spin_lock_init(&c->spinlock);
 	INIT_LIST_HEAD(&c->reserved_buffers);
@@ -1792,9 +2521,9 @@ struct dm_bufio_client *dm_bufio_client_create(struct block_device *bdev, unsign
 
 bad:
 	while (!list_empty(&c->reserved_buffers)) {
-		struct dm_buffer *b = list_entry(c->reserved_buffers.next,
-						 struct dm_buffer, lru_list);
-		list_del(&b->lru_list);
+		struct dm_buffer *b = list_to_buffer(c->reserved_buffers.next);
+
+		list_del(&b->lru.list);
 		free_buffer(b);
 	}
 	kmem_cache_destroy(c->slab_cache);
@@ -1831,23 +2560,23 @@ void dm_bufio_client_destroy(struct dm_bufio_client *c)
 
 	mutex_unlock(&dm_bufio_clients_lock);
 
-	BUG_ON(!RB_EMPTY_ROOT(&c->buffer_tree));
 	BUG_ON(c->need_reserved_buffers);
 
 	while (!list_empty(&c->reserved_buffers)) {
-		struct dm_buffer *b = list_entry(c->reserved_buffers.next,
-						 struct dm_buffer, lru_list);
-		list_del(&b->lru_list);
+		struct dm_buffer *b = list_to_buffer(c->reserved_buffers.next);
+
+		list_del(&b->lru.list);
 		free_buffer(b);
 	}
 
 	for (i = 0; i < LIST_SIZE; i++)
-		if (c->n_buffers[i])
-			DMERR("leaked buffer count %d: %ld", i, c->n_buffers[i]);
+		if (cache_count(&c->cache, i))
+			DMERR("leaked buffer count %d: %lu", i, cache_count(&c->cache, i));
 
 	for (i = 0; i < LIST_SIZE; i++)
-		BUG_ON(c->n_buffers[i]);
+		BUG_ON(cache_count(&c->cache, i));
 
+	cache_destroy(&c->cache);
 	kmem_cache_destroy(c->slab_cache);
 	kmem_cache_destroy(c->slab_buffer);
 	dm_io_client_destroy(c->dm_io);
@@ -1864,6 +2593,8 @@ void dm_bufio_set_sector_offset(struct dm_bufio_client *c, sector_t start)
 }
 EXPORT_SYMBOL_GPL(dm_bufio_set_sector_offset);
 
+/*--------------------------------------------------------------*/
+
 static unsigned int get_max_age_hz(void)
 {
 	unsigned int max_age = READ_ONCE(dm_bufio_max_age);
@@ -1876,13 +2607,74 @@ static unsigned int get_max_age_hz(void)
 
 static bool older_than(struct dm_buffer *b, unsigned long age_hz)
 {
-	return time_after_eq(jiffies, b->last_accessed + age_hz);
+	return time_after_eq(jiffies, READ_ONCE(b->last_accessed) + age_hz);
 }
 
-static void __evict_old_buffers(struct dm_bufio_client *c, unsigned long age_hz)
+struct evict_params {
+	gfp_t gfp;
+	unsigned long age_hz;
+
+	/*
+	 * This gets updated with the largest last_accessed (ie. most
+	 * recently used) of the evicted buffers.  It will not be reinitialised
+	 * by __evict_many(), so you can use it across multiple invocations.
+	 */
+	unsigned long last_accessed;
+};
+
+/*
+ * We may not be able to evict this buffer if IO pending or the client
+ * is still using it.
+ *
+ * And if GFP_NOFS is used, we must not do any I/O because we hold
+ * dm_bufio_clients_lock and we would risk deadlock if the I/O gets
+ * rerouted to different bufio client.
+ */
+static enum evict_result select_for_evict(struct dm_buffer *b, void *context)
+{
+	struct evict_params *params = context;
+
+	if (!(params->gfp & __GFP_FS) ||
+	    (static_branch_unlikely(&no_sleep_enabled) && b->c->no_sleep)) {
+		if (test_bit_acquire(B_READING, &b->state) ||
+		    test_bit(B_WRITING, &b->state) ||
+		    test_bit(B_DIRTY, &b->state))
+			return ER_DONT_EVICT;
+	}
+
+	return older_than(b, params->age_hz) ? ER_EVICT : ER_STOP;
+}
+
+static unsigned int __evict_many(struct dm_bufio_client *c,
+				 struct evict_params *params,
+				 int list_mode, unsigned int max_count)
+{
+	unsigned int count;
+	unsigned long last_accessed;
+	struct dm_buffer *b;
+
+	for (count = 0; count < max_count; count++) {
+		b = cache_evict(&c->cache, list_mode, select_for_evict, params);
+		if (!b)
+			break;
+
+		last_accessed = READ_ONCE(b->last_accessed);
+		if (time_after_eq(params->last_accessed, last_accessed))
+			params->last_accessed = last_accessed;
+
+		__make_buffer_clean(b);
+		__free_buffer_wake(b);
+
+		cond_resched();
+	}
+
+	return count;
+}
+
+static void evict_old_buffers(struct dm_bufio_client *c, unsigned long age_hz)
 {
-	struct dm_buffer *b, *tmp;
-	unsigned long retain_target = get_retain_buffers(c);
+	struct evict_params params = {.gfp = 0, .age_hz = age_hz, .last_accessed = 0};
+	unsigned long retain = get_retain_buffers(c);
 	unsigned long count;
 	LIST_HEAD(write_list);
 
@@ -1895,91 +2687,13 @@ static void __evict_old_buffers(struct dm_bufio_client *c, unsigned long age_hz)
 		dm_bufio_lock(c);
 	}
 
-	count = c->n_buffers[LIST_CLEAN] + c->n_buffers[LIST_DIRTY];
-	list_for_each_entry_safe_reverse(b, tmp, &c->lru[LIST_CLEAN], lru_list) {
-		if (count <= retain_target)
-			break;
-
-		if (!older_than(b, age_hz))
-			break;
-
-		if (__try_evict_buffer(b, 0))
-			count--;
-
-		cond_resched();
-	}
+	count = cache_total(&c->cache);
+	if (count > retain)
+		__evict_many(c, &params, LIST_CLEAN, count - retain);
 
 	dm_bufio_unlock(c);
 }
 
-static void do_global_cleanup(struct work_struct *w)
-{
-	struct dm_bufio_client *locked_client = NULL;
-	struct dm_bufio_client *current_client;
-	struct dm_buffer *b;
-	unsigned int spinlock_hold_count;
-	unsigned long threshold = dm_bufio_cache_size -
-		dm_bufio_cache_size / DM_BUFIO_LOW_WATERMARK_RATIO;
-	unsigned long loops = global_num * 2;
-
-	mutex_lock(&dm_bufio_clients_lock);
-
-	while (1) {
-		cond_resched();
-
-		spin_lock(&global_spinlock);
-		if (unlikely(dm_bufio_current_allocated <= threshold))
-			break;
-
-		spinlock_hold_count = 0;
-get_next:
-		if (!loops--)
-			break;
-		if (unlikely(list_empty(&global_queue)))
-			break;
-		b = list_entry(global_queue.prev, struct dm_buffer, global_list);
-
-		if (b->accessed) {
-			b->accessed = 0;
-			list_move(&b->global_list, &global_queue);
-			if (likely(++spinlock_hold_count < 16))
-				goto get_next;
-			spin_unlock(&global_spinlock);
-			continue;
-		}
-
-		current_client = b->c;
-		if (unlikely(current_client != locked_client)) {
-			if (locked_client)
-				dm_bufio_unlock(locked_client);
-
-			if (!dm_bufio_trylock(current_client)) {
-				spin_unlock(&global_spinlock);
-				dm_bufio_lock(current_client);
-				locked_client = current_client;
-				continue;
-			}
-
-			locked_client = current_client;
-		}
-
-		spin_unlock(&global_spinlock);
-
-		if (unlikely(!__try_evict_buffer(b, GFP_KERNEL))) {
-			spin_lock(&global_spinlock);
-			list_move(&b->global_list, &global_queue);
-			spin_unlock(&global_spinlock);
-		}
-	}
-
-	spin_unlock(&global_spinlock);
-
-	if (locked_client)
-		dm_bufio_unlock(locked_client);
-
-	mutex_unlock(&dm_bufio_clients_lock);
-}
-
 static void cleanup_old_buffers(void)
 {
 	unsigned long max_age_hz = get_max_age_hz();
@@ -1990,7 +2704,7 @@ static void cleanup_old_buffers(void)
 	__cache_size_refresh();
 
 	list_for_each_entry(c, &dm_bufio_all_clients, client_list)
-		__evict_old_buffers(c, max_age_hz);
+		evict_old_buffers(c, max_age_hz);
 
 	mutex_unlock(&dm_bufio_clients_lock);
 }
@@ -2003,6 +2717,107 @@ static void work_fn(struct work_struct *w)
 			   DM_BUFIO_WORK_TIMER_SECS * HZ);
 }
 
+/*--------------------------------------------------------------*/
+
+/*
+ * Global cleanup tries to evict the oldest buffers from across _all_
+ * the clients.  It does this by repeatedly evicting a few buffers from
+ * the client that holds the oldest buffer.  It's approximate, but hopefully
+ * good enough.
+ */
+static struct dm_bufio_client *__pop_client(void)
+{
+	struct list_head *h;
+
+	if (list_empty(&dm_bufio_all_clients))
+		return NULL;
+
+	h = dm_bufio_all_clients.next;
+	list_del(h);
+	return container_of(h, struct dm_bufio_client, client_list);
+}
+
+/*
+ * Inserts the client in the global client list based on its
+ * 'oldest_buffer' field.
+ */
+static void __insert_client(struct dm_bufio_client *new_client)
+{
+	struct dm_bufio_client *c;
+	struct list_head *h = dm_bufio_all_clients.next;
+
+	while (h != &dm_bufio_all_clients) {
+		c = container_of(h, struct dm_bufio_client, client_list);
+		if (time_after_eq(c->oldest_buffer, new_client->oldest_buffer))
+			break;
+		h = h->next;
+	}
+
+	list_add_tail(&new_client->client_list, h);
+}
+
+static unsigned int __evict_a_few(unsigned int nr_buffers)
+{
+	unsigned int count;
+	struct dm_bufio_client *c;
+	struct evict_params params = {
+		.gfp = GFP_KERNEL,
+		.age_hz = 0,
+		/* set to jiffies in case there are no buffers in this client */
+		.last_accessed = jiffies
+	};
+
+	c = __pop_client();
+	if (!c)
+		return 0;
+
+	dm_bufio_lock(c);
+	count = __evict_many(c, &params, LIST_CLEAN, nr_buffers);
+	dm_bufio_unlock(c);
+
+	if (count)
+		c->oldest_buffer = params.last_accessed;
+	__insert_client(c);
+
+	return count;
+}
+
+static void check_watermarks(void)
+{
+	LIST_HEAD(write_list);
+	struct dm_bufio_client *c;
+
+	mutex_lock(&dm_bufio_clients_lock);
+	list_for_each_entry(c, &dm_bufio_all_clients, client_list) {
+		dm_bufio_lock(c);
+		__check_watermark(c, &write_list);
+		dm_bufio_unlock(c);
+	}
+	mutex_unlock(&dm_bufio_clients_lock);
+
+	__flush_write_list(&write_list);
+}
+
+static void evict_old(void)
+{
+	unsigned long threshold = dm_bufio_cache_size -
+		dm_bufio_cache_size / DM_BUFIO_LOW_WATERMARK_RATIO;
+
+	mutex_lock(&dm_bufio_clients_lock);
+	while (dm_bufio_current_allocated > threshold) {
+		if (!__evict_a_few(64))
+			break;
+		cond_resched();
+	}
+	mutex_unlock(&dm_bufio_clients_lock);
+}
+
+static void do_global_cleanup(struct work_struct *w)
+{
+	check_watermarks();
+	evict_old();
+}
+
 /*
  *--------------------------------------------------------------
  * Module setup
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

Re: [dm-devel] [dm-6.4 PATCH 5/8] dm bufio: improve concurrent IO performance

[Joe Thornber]

[-- Attachment #1.1: Type: text/plain, Size: 81840 bytes --]

Mike,

I'm Nacking this patch here.  You've taken the comment from my latest
patch, and attached it to something else.  I don't know if you've taken an
older patch and updated it, or taken the latest patch and downgraded it.

For instance the latest patch uses the name 'struct dm_buffer_cache', but
it looks like you've renamed to 'struct buffer_cache' and then posted
another patch in the series that renames back to dm_buffer_cache.

I also see the comment change:

original "A refcount_t for hold_count does not work." which has changed to
"NOTE: attempting to use refcount_t instead of atomic_t caused crashes!".
Talk about over dramatisation (If the hold_count is incorrect we detect and
BUG).

I can't stand by my code if I don't know what's being sent up in my name.

Please use the patchset from this branch:
https://github.com/jthornber/linux/tree/2023-03-21-thin-concurrency-8

- Joe



On Wed, Mar 22, 2023 at 6:19 PM Mike Snitzer <snitzer@kernel.org> wrote:

> From: Joe Thornber <ejt@redhat.com>
>
> When multiple threads perform IO to a thin device, the underlying
> dm_bufio object can become a bottleneck; slowing down access to btree
> nodes that store the thin metadata. Prior to this commit, each bufio
> instance had a single mutex that was taken for every bufio operation.
>
> This commit concentrates on improving the common case: in which a user
> of dm_bufio wishes to access, but not modify, a buffer which is already
> within the dm_bufio cache.
>
> Implementation::
>
>   The code has been refactored; pulling out an 'lru' abstraction and a
>   'buffer cache' abstraction.
>
>  LRU:
>   A clock algorithm is used in the LRU abstraction.  This avoids
>   relinking list nodes, which would require a write lock protecting
>   it.
>
>   None of the lru methods are threadsafe; locking must be done at a
>   higher level.
>
>  Buffer cache:
>   The buffer cache is responsible for managing the holder count,
>   tracking clean/dirty state, and choosing buffers via predicates.
>   Higher level code is responsible for allocation of buffers, IO and
>   eviction/cache sizing.
>
>   The buffer cache has thread safe methods for acquiring a reference
>   to an existing buffer. All other methods in buffer cache are _not_
>   threadsafe, and only contain enough locking to guarantee the safe
>   methods.
>
>   Rather than a single mutex, sharded rw_semaphores are used to allow
>   concurrent threads to 'get' buffers. Each rw_semaphore protects its
>   own rbtree of buffer entries.
>
> Testing::
>
>   Some of the low level functions have been unit tested using dm-unit:
>     https://github.com/jthornber/dm-unit/blob/main/src/tests/bufio.rs
>
>   Higher level concurrency and IO is tested via a test only target
>   found here:
>
> https://github.com/jthornber/linux/blob/2023-02-28-thin-concurrency-7/drivers/md/dm-bufio-test.c
>
>   The associated userland side of these tests is here:
>
> https://github.com/jthornber/dmtest-python/blob/main/src/dmtest/bufio/bufio_tests.py
>
>   In addition the full dmtest suite of tests (dm-thin, dm-cache, etc)
>   has been run (~450 tests).
>
> Performance::
>
>   Most bufio operations have unchanged performance. But if multiple
>   threads are attempting to get buffers concurrently, and these
>   buffers are already in the cache then there's a big speed up. Eg,
>   one test has 16 'hotspot' threads simulating btree lookups while
>   another thread dirties the whole device. In this case the hotspot
>   threads acquire the buffers about 25 times faster.
>
> Signed-off-by: Joe Thornber <ejt@redhat.com>
> Signed-off-by: Mike Snitzer <snitzer@kernel.org>
> ---
>  drivers/md/dm-bufio.c | 1769 ++++++++++++++++++++++++++++++-----------
>  1 file changed, 1292 insertions(+), 477 deletions(-)
>
> diff --git a/drivers/md/dm-bufio.c b/drivers/md/dm-bufio.c
> index 1de1bdcda1ce..56ef160331ce 100644
> --- a/drivers/md/dm-bufio.c
> +++ b/drivers/md/dm-bufio.c
> @@ -66,6 +66,247 @@
>  #define LIST_DIRTY     1
>  #define LIST_SIZE      2
>
> +/*--------------------------------------------------------------*/
> +
> +/*
> + * Rather than use an LRU list, we use a clock algorithm where entries
> + * are held in a circular list.  When an entry is 'hit' a reference bit
> + * is set.  The least recently used entry is approximated by running a
> + * cursor around the list selecting unreferenced entries. Referenced
> + * entries have their reference bit cleared as the cursor passes them.
> + */
> +struct lru_entry {
> +       struct list_head list;
> +       atomic_t referenced;
> +};
> +
> +struct lru_iter {
> +       struct lru *lru;
> +       struct list_head list;
> +       struct lru_entry *stop;
> +       struct lru_entry *e;
> +};
> +
> +struct lru {
> +       struct list_head *cursor;
> +       unsigned long count;
> +
> +       struct list_head iterators;
> +};
> +
> +/*--------------*/
> +
> +static void lru_init(struct lru *lru)
> +{
> +       lru->cursor = NULL;
> +       lru->count = 0;
> +       INIT_LIST_HEAD(&lru->iterators);
> +}
> +
> +static void lru_destroy(struct lru *lru)
> +{
> +       BUG_ON(lru->cursor);
> +       BUG_ON(!list_empty(&lru->iterators));
> +}
> +
> +static inline unsigned int lru_count(struct lru *lru)
> +{
> +       return lru->count;
> +}
> +
> +/*
> + * Insert a new entry into the lru.
> + */
> +static void lru_insert(struct lru *lru, struct lru_entry *le)
> +{
> +       /*
> +        * Don't be tempted to set to 1, makes the lru aspect
> +        * perform poorly.
> +        */
> +       atomic_set(&le->referenced, 0);
> +
> +       if (lru->cursor)
> +               list_add_tail(&le->list, lru->cursor);
> +
> +       else {
> +               INIT_LIST_HEAD(&le->list);
> +               lru->cursor = &le->list;
> +       }
> +       lru->count++;
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Convert a list_head pointer to an lru_entry pointer.
> + */
> +static inline struct lru_entry *to_le(struct list_head *l)
> +{
> +       return container_of(l, struct lru_entry, list);
> +}
> +
> +/*
> + * Initialize an lru_iter and add it to the list of cursors in the lru.
> + */
> +static void lru_iter_begin(struct lru *lru, struct lru_iter *it)
> +{
> +       it->lru = lru;
> +       it->stop = lru->cursor ? to_le(lru->cursor->prev) : NULL;
> +       it->e = lru->cursor ? to_le(lru->cursor) : NULL;
> +       list_add(&it->list, &lru->iterators);
> +}
> +
> +/*
> + * Remove an lru_iter from the list of cursors in the lru.
> + */
> +static void lru_iter_end(struct lru_iter *it)
> +{
> +       list_del(&it->list);
> +}
> +
> +/* Predicate function type to be used with lru_iter_next */
> +typedef bool (*iter_predicate)(struct lru_entry *le, void *context);
> +
> +/*
> + * Advance the cursor to the next entry that passes the
> + * predicate, and return that entry.  Returns NULL if the
> + * iteration is complete.
> + */
> +static struct lru_entry *lru_iter_next(struct lru_iter *it,
> +                                      iter_predicate pred, void *context)
> +{
> +       struct lru_entry *e;
> +
> +       while (it->e) {
> +               e = it->e;
> +
> +               /* advance the cursor */
> +               if (it->e == it->stop)
> +                       it->e = NULL;
> +               else
> +                       it->e = to_le(it->e->list.next);
> +
> +               if (pred(e, context))
> +                       return e;
> +       }
> +
> +       return NULL;
> +}
> +
> +/*
> + * Invalidate a specific lru_entry and update all cursors in
> + * the lru accordingly.
> + */
> +static void lru_iter_invalidate(struct lru *lru, struct lru_entry *e)
> +{
> +       struct lru_iter *it;
> +
> +       list_for_each_entry(it, &lru->iterators, list) {
> +               /* Move c->e forwards if necc. */
> +               if (it->e == e) {
> +                       it->e = to_le(it->e->list.next);
> +                       if (it->e == e)
> +                               it->e = NULL;
> +               }
> +
> +               /* Move it->stop backwards if necc. */
> +               if (it->stop == e) {
> +                       it->stop = to_le(it->stop->list.prev);
> +                       if (it->stop == e)
> +                               it->stop = NULL;
> +               }
> +       }
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Remove a specific entry from the lru.
> + */
> +static void lru_remove(struct lru *lru, struct lru_entry *le)
> +{
> +       lru_iter_invalidate(lru, le);
> +       if (lru->count == 1)
> +               lru->cursor = NULL;
> +       else {
> +               if (lru->cursor == &le->list)
> +                       lru->cursor = lru->cursor->next;
> +               list_del(&le->list);
> +       }
> +       lru->count--;
> +}
> +
> +/*
> + * Mark as referenced.
> + */
> +static inline void lru_reference(struct lru_entry *le)
> +{
> +       atomic_set(&le->referenced, 1);
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Remove the least recently used entry (approx), that passes the
> predicate.
> + * Returns NULL on failure.
> + */
> +enum evict_result {
> +       ER_EVICT,
> +       ER_DONT_EVICT,
> +       ER_STOP, /* stop looking for something to evict */
> +};
> +
> +typedef enum evict_result (*le_predicate)(struct lru_entry *le, void
> *context);
> +
> +static struct lru_entry *lru_evict(struct lru *lru, le_predicate pred,
> void *context)
> +{
> +       unsigned int tested = 0;
> +       struct list_head *h = lru->cursor;
> +       struct lru_entry *le;
> +
> +       if (!h)
> +               return NULL;
> +       /*
> +        * In the worst case we have to loop around twice. Once to clear
> +        * the reference flags, and then again to discover the predicate
> +        * fails for all entries.
> +        */
> +       while (tested < lru->count) {
> +               le = container_of(h, struct lru_entry, list);
> +
> +               if (atomic_read(&le->referenced))
> +                       atomic_set(&le->referenced, 0);
> +               else {
> +                       tested++;
> +                       switch (pred(le, context)) {
> +                       case ER_EVICT:
> +                               /*
> +                                * Adjust the cursor, so we start the next
> +                                * search from here.
> +                                */
> +                               lru->cursor = le->list.next;
> +                               lru_remove(lru, le);
> +                               return le;
> +
> +                       case ER_DONT_EVICT:
> +                               break;
> +
> +                       case ER_STOP:
> +                               lru->cursor = le->list.next;
> +                               return NULL;
> +                       }
> +               }
> +
> +               h = h->next;
> +
> +               cond_resched();
> +       }
> +
> +       return NULL;
> +}
> +
> +/*--------------------------------------------------------------*/
> +
>  /*
>   * Buffer state bits.
>   */
> @@ -86,28 +327,37 @@ enum data_mode {
>  };
>
>  struct dm_buffer {
> +       /* protected by the locks in buffer_cache */
>         struct rb_node node;
> -       struct list_head lru_list;
> -       struct list_head global_list;
>
> +       /* immutable, so don't need protecting */
>         sector_t block;
>         void *data;
>         unsigned char data_mode;                /* DATA_MODE_* */
>
> -       unsigned int accessed;
> -       unsigned int hold_count;
> +       /*
> +        * These two fields are used in isolation, so do not need
> +        * a surrounding lock. NOTE: attempting to use refcount_t
> +        * instead of atomic_t caused crashes!
> +        */
> +       atomic_t hold_count;
>         unsigned long last_accessed;
>
> +       /*
> +        * Everything else is protected by the mutex in
> +        * dm_bufio_client
> +        */
> +       unsigned long state;
> +       struct lru_entry lru;
>         unsigned char list_mode;                /* LIST_* */
>         blk_status_t read_error;
>         blk_status_t write_error;
> -       unsigned long state;
>         unsigned int dirty_start;
>         unsigned int dirty_end;
>         unsigned int write_start;
>         unsigned int write_end;
> -       struct dm_bufio_client *c;
>         struct list_head write_list;
> +       struct dm_bufio_client *c;
>         void (*end_io)(struct dm_buffer *b, blk_status_t bs);
>  #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
>  #define MAX_STACK 10
> @@ -116,9 +366,579 @@ struct dm_buffer {
>  #endif
>  };
>
> +/*--------------------------------------------------------------*/
> +
> +/*
> + * The buffer cache manages buffers, particularly:
> + *  - inc/dec of holder count
> + *  - setting the last_accessed field
> + *  - maintains clean/dirty state along with lru
> + *  - selecting buffers that match predicates
> + *
> + * It does *not* handle:
> + *  - allocation/freeing of buffers.
> + *  - IO
> + *  - Eviction or cache sizing.
> + *
> + * cache_get() and cache_put() are threadsafe, you do not need to
> + * protect these calls with a surrounding mutex.  All the other
> + * methods are not threadsafe; they do use locking primitives, but
> + * only enough to ensure get/put are threadsafe.
> + */
> +
> +#define NR_LOCKS 64
> +#define LOCKS_MASK (NR_LOCKS - 1)
> +
> +struct tree_lock {
> +       struct rw_semaphore lock;
> +} ____cacheline_aligned_in_smp;
> +
> +struct buffer_cache {
> +       /*
> +        * We spread entries across multiple trees to reduce contention
> +        * on the locks.
> +        */
> +       struct tree_lock locks[NR_LOCKS];
> +       struct rb_root roots[NR_LOCKS];
> +       struct lru lru[LIST_SIZE];
> +};
> +
> +/*
> + * We want to scatter the buffers across the different rbtrees
> + * to improve concurrency, but we also want to encourage the
> + * lock_history optimisation (below). Scattering runs of 16
> + * buffers seems to be a good compromise.
> + */
> +static inline unsigned int cache_index(sector_t block)
> +{
> +       return block & LOCKS_MASK;
> +}
> +
> +static inline void cache_read_lock(struct buffer_cache *bc, sector_t
> block)
> +{
> +       down_read(&bc->locks[cache_index(block)].lock);
> +}
> +
> +static inline void cache_read_unlock(struct buffer_cache *bc, sector_t
> block)
> +{
> +       up_read(&bc->locks[cache_index(block)].lock);
> +}
> +
> +static inline void cache_write_lock(struct buffer_cache *bc, sector_t
> block)
> +{
> +       down_write(&bc->locks[cache_index(block)].lock);
> +}
> +
> +static inline void cache_write_unlock(struct buffer_cache *bc, sector_t
> block)
> +{
> +       up_write(&bc->locks[cache_index(block)].lock);
> +}
> +
> +/*
> + * Sometimes we want to repeatedly get and drop locks as part of an
> iteration.
> + * This struct helps avoid redundant drop and gets of the same lock.
> + */
> +struct lock_history {
> +       struct buffer_cache *cache;
> +       bool write;
> +       unsigned int previous;
> +};
> +
> +static void lh_init(struct lock_history *lh, struct buffer_cache *cache,
> bool write)
> +{
> +       lh->cache = cache;
> +       lh->write = write;
> +       lh->previous = NR_LOCKS;  /* indicates no previous */
> +}
> +
> +static void __lh_lock(struct lock_history *lh, unsigned int index)
> +{
> +       if (lh->write)
> +               down_write(&lh->cache->locks[index].lock);
> +       else
> +               down_read(&lh->cache->locks[index].lock);
> +}
> +
> +static void __lh_unlock(struct lock_history *lh, unsigned int index)
> +{
> +       if (lh->write)
> +               up_write(&lh->cache->locks[index].lock);
> +       else
> +               up_read(&lh->cache->locks[index].lock);
> +}
> +
> +/*
> + * Make sure you call this since it will unlock the final lock.
> + */
> +static void lh_exit(struct lock_history *lh)
> +{
> +       if (lh->previous != NR_LOCKS) {
> +               __lh_unlock(lh, lh->previous);
> +               lh->previous = NR_LOCKS;
> +       }
> +}
> +
> +/*
> + * Named 'next' because there is no corresponding
> + * 'up/unlock' call since it's done automatically.
> + */
> +static void lh_next(struct lock_history *lh, sector_t b)
> +{
> +       unsigned int index = cache_index(b);
> +
> +       if (lh->previous != NR_LOCKS) {
> +               if (lh->previous != index) {
> +                       __lh_unlock(lh, lh->previous);
> +                       __lh_lock(lh, index);
> +                       lh->previous = index;
> +               }
> +       } else {
> +               __lh_lock(lh, index);
> +               lh->previous = index;
> +       }
> +}
> +
> +static struct dm_buffer *le_to_buffer(struct lru_entry *le)
> +{
> +       return container_of(le, struct dm_buffer, lru);
> +}
> +
> +static struct dm_buffer *list_to_buffer(struct list_head *l)
> +{
> +       struct lru_entry *le = list_entry(l, struct lru_entry, list);
> +
> +       if (!le)
> +               return NULL;
> +
> +       return le_to_buffer(le);
> +}
> +
> +static void cache_init(struct buffer_cache *bc)
> +{
> +       unsigned int i;
> +
> +       for (i = 0; i < NR_LOCKS; i++) {
> +               init_rwsem(&bc->locks[i].lock);
> +               bc->roots[i] = RB_ROOT;
> +       }
> +
> +       lru_init(&bc->lru[LIST_CLEAN]);
> +       lru_init(&bc->lru[LIST_DIRTY]);
> +}
> +
> +static void cache_destroy(struct buffer_cache *bc)
> +{
> +       unsigned int i;
> +
> +       for (i = 0; i < NR_LOCKS; i++)
> +               BUG_ON(!RB_EMPTY_ROOT(&bc->roots[i]));
> +
> +       lru_destroy(&bc->lru[LIST_CLEAN]);
> +       lru_destroy(&bc->lru[LIST_DIRTY]);
> +}
> +
> +/*
> + * not threadsafe, or racey depending how you look at it
> + */
> +static unsigned long cache_count(struct buffer_cache *bc, int list_mode)
> +{
> +       return lru_count(&bc->lru[list_mode]);
> +}
> +
> +/*--------------*/
> +
> +/*
> + * not threadsafe
> + */
> +static unsigned long cache_total(struct buffer_cache *bc)
> +{
> +       return lru_count(&bc->lru[LIST_CLEAN]) +
> +               lru_count(&bc->lru[LIST_DIRTY]);
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Gets a specific buffer, indexed by block.
> + * If the buffer is found then its holder count will be incremented and
> + * lru_reference will be called.
> + *
> + * threadsafe
> + */
> +static struct dm_buffer *__cache_get(const struct rb_root *root, sector_t
> block)
> +{
> +       struct rb_node *n = root->rb_node;
> +       struct dm_buffer *b;
> +
> +       while (n) {
> +               b = container_of(n, struct dm_buffer, node);
> +
> +               if (b->block == block)
> +                       return b;
> +
> +               n = block < b->block ? n->rb_left : n->rb_right;
> +       }
> +
> +       return NULL;
> +}
> +
> +static void __cache_inc_buffer(struct dm_buffer *b)
> +{
> +       atomic_inc(&b->hold_count);
> +       WRITE_ONCE(b->last_accessed, jiffies);
> +}
> +
> +static struct dm_buffer *cache_get(struct buffer_cache *bc, sector_t
> block)
> +{
> +       struct dm_buffer *b;
> +
> +       cache_read_lock(bc, block);
> +       b = __cache_get(&bc->roots[cache_index(block)], block);
> +       if (b) {
> +               lru_reference(&b->lru);
> +               __cache_inc_buffer(b);
> +       }
> +       cache_read_unlock(bc, block);
> +
> +       return b;
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Returns true if the hold count hits zero.
> + * threadsafe
> + */
> +static bool cache_put(struct buffer_cache *bc, struct dm_buffer *b)
> +{
> +       bool r;
> +
> +       cache_read_lock(bc, b->block);
> +       BUG_ON(!atomic_read(&b->hold_count));
> +       r = atomic_dec_and_test(&b->hold_count);
> +       cache_read_unlock(bc, b->block);
> +
> +       return r;
> +}
> +
> +/*--------------*/
> +
> +typedef enum evict_result (*b_predicate)(struct dm_buffer *, void *);
> +
> +/*
> + * Evicts a buffer based on a predicate.  The oldest buffer that
> + * matches the predicate will be selected.  In addition to the
> + * predicate the hold_count of the selected buffer will be zero.
> + */
> +struct evict_wrapper {
> +       struct lock_history *lh;
> +       b_predicate pred;
> +       void *context;
> +};
> +
> +/*
> + * Wraps the buffer predicate turning it into an lru predicate.  Adds
> + * extra test for hold_count.
> + */
> +static enum evict_result __evict_pred(struct lru_entry *le, void *context)
> +{
> +       struct evict_wrapper *w = context;
> +       struct dm_buffer *b = le_to_buffer(le);
> +
> +       lh_next(w->lh, b->block);
> +
> +       if (atomic_read(&b->hold_count))
> +               return ER_DONT_EVICT;
> +
> +       return w->pred(b, w->context);
> +}
> +
> +static struct dm_buffer *__cache_evict(struct buffer_cache *bc, int
> list_mode,
> +                                      b_predicate pred, void *context,
> +                                      struct lock_history *lh)
> +{
> +       struct evict_wrapper w = {.lh = lh, .pred = pred, .context =
> context};
> +       struct lru_entry *le;
> +       struct dm_buffer *b;
> +
> +       le = lru_evict(&bc->lru[list_mode], __evict_pred, &w);
> +       if (!le)
> +               return NULL;
> +
> +       b = le_to_buffer(le);
> +       /* __evict_pred will have locked the appropriate tree. */
> +       rb_erase(&b->node, &bc->roots[cache_index(b->block)]);
> +
> +       return b;
> +}
> +
> +static struct dm_buffer *cache_evict(struct buffer_cache *bc, int
> list_mode,
> +                                    b_predicate pred, void *context)
> +{
> +       struct dm_buffer *b;
> +       struct lock_history lh;
> +
> +       lh_init(&lh, bc, true);
> +       b = __cache_evict(bc, list_mode, pred, context, &lh);
> +       lh_exit(&lh);
> +
> +       return b;
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Mark a buffer as clean or dirty. Not threadsafe.
> + */
> +static void cache_mark(struct buffer_cache *bc, struct dm_buffer *b, int
> list_mode)
> +{
> +       cache_write_lock(bc, b->block);
> +       if (list_mode != b->list_mode) {
> +               lru_remove(&bc->lru[b->list_mode], &b->lru);
> +               b->list_mode = list_mode;
> +               lru_insert(&bc->lru[b->list_mode], &b->lru);
> +       }
> +       cache_write_unlock(bc, b->block);
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Runs through the lru associated with 'old_mode', if the predicate
> matches then
> + * it moves them to 'new_mode'.  Not threadsafe.
> + */
> +static void __cache_mark_many(struct buffer_cache *bc, int old_mode, int
> new_mode,
> +                             b_predicate pred, void *context, struct
> lock_history *lh)
> +{
> +       struct lru_entry *le;
> +       struct dm_buffer *b;
> +       struct evict_wrapper w = {.lh = lh, .pred = pred, .context =
> context};
> +
> +       while (true) {
> +               le = lru_evict(&bc->lru[old_mode], __evict_pred, &w);
> +               if (!le)
> +                       break;
> +
> +               b = le_to_buffer(le);
> +               b->list_mode = new_mode;
> +               lru_insert(&bc->lru[b->list_mode], &b->lru);
> +       }
> +}
> +
> +static void cache_mark_many(struct buffer_cache *bc, int old_mode, int
> new_mode,
> +                           b_predicate pred, void *context)
> +{
> +       struct lock_history lh;
> +
> +       lh_init(&lh, bc, true);
> +       __cache_mark_many(bc, old_mode, new_mode, pred, context, &lh);
> +       lh_exit(&lh);
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Iterates through all clean or dirty entries calling a function for each
> + * entry.  The callback may terminate the iteration early.  Not
> threadsafe.
> + */
> +
> +/*
> + * Iterator functions should return one of these actions to indicate
> + * how the iteration should proceed.
> + */
> +enum it_action {
> +       IT_NEXT,
> +       IT_COMPLETE,
> +};
> +
> +typedef enum it_action (*iter_fn)(struct dm_buffer *b, void *context);
> +
> +static void __cache_iterate(struct buffer_cache *bc, int list_mode,
> +                           iter_fn fn, void *context, struct lock_history
> *lh)
> +{
> +       struct lru *lru = &bc->lru[list_mode];
> +       struct lru_entry *le, *first;
> +
> +       if (!lru->cursor)
> +               return;
> +
> +       first = le = to_le(lru->cursor);
> +       do {
> +               struct dm_buffer *b = le_to_buffer(le);
> +
> +               lh_next(lh, b->block);
> +
> +               switch (fn(b, context)) {
> +               case IT_NEXT:
> +                       break;
> +
> +               case IT_COMPLETE:
> +                       return;
> +               }
> +               cond_resched();
> +
> +               le = to_le(le->list.next);
> +       } while (le != first);
> +}
> +
> +static void cache_iterate(struct buffer_cache *bc, int list_mode,
> +                         iter_fn fn, void *context)
> +{
> +       struct lock_history lh;
> +
> +       lh_init(&lh, bc, false);
> +       __cache_iterate(bc, list_mode, fn, context, &lh);
> +       lh_exit(&lh);
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Passes ownership of the buffer to the cache. Returns false if the
> + * buffer was already present (in which case ownership does not pass).
> + * eg, a race with another thread.
> + *
> + * Holder count should be 1 on insertion.
> + *
> + * Not threadsafe.
> + */
> +static bool __cache_insert(struct rb_root *root, struct dm_buffer *b)
> +{
> +       struct rb_node **new = &root->rb_node, *parent = NULL;
> +       struct dm_buffer *found;
> +
> +       while (*new) {
> +               found = container_of(*new, struct dm_buffer, node);
> +
> +               if (found->block == b->block)
> +                       return false;
> +
> +               parent = *new;
> +               new = b->block < found->block ?
> +                       &found->node.rb_left : &found->node.rb_right;
> +       }
> +
> +       rb_link_node(&b->node, parent, new);
> +       rb_insert_color(&b->node, root);
> +
> +       return true;
> +}
> +
> +static bool cache_insert(struct buffer_cache *bc, struct dm_buffer *b)
> +{
> +       bool r;
> +
> +       BUG_ON(b->list_mode >= LIST_SIZE);
> +
> +       cache_write_lock(bc, b->block);
> +       BUG_ON(atomic_read(&b->hold_count) != 1);
> +       r = __cache_insert(&bc->roots[cache_index(b->block)], b);
> +       if (r)
> +               lru_insert(&bc->lru[b->list_mode], &b->lru);
> +       cache_write_unlock(bc, b->block);
> +
> +       return r;
> +}
> +
> +/*--------------*/
> +
> +/*
> + * Removes buffer from cache, ownership of the buffer passes back to the
> caller.
> + * Fails if the hold_count is not one (ie. the caller holds the only
> reference).
> + *
> + * Not threadsafe.
> + */
> +static bool cache_remove(struct buffer_cache *bc, struct dm_buffer *b)
> +{
> +       bool r;
> +
> +       cache_write_lock(bc, b->block);
> +
> +       if (atomic_read(&b->hold_count) != 1)
> +               r = false;
> +
> +       else {
> +               r = true;
> +               rb_erase(&b->node, &bc->roots[cache_index(b->block)]);
> +               lru_remove(&bc->lru[b->list_mode], &b->lru);
> +       }
> +
> +       cache_write_unlock(bc, b->block);
> +
> +       return r;
> +}
> +
> +/*--------------*/
> +
> +typedef void (*b_release)(struct dm_buffer *);
> +
> +static struct dm_buffer *__find_next(struct rb_root *root, sector_t block)
> +{
> +       struct rb_node *n = root->rb_node;
> +       struct dm_buffer *b;
> +       struct dm_buffer *best = NULL;
> +
> +       while (n) {
> +               b = container_of(n, struct dm_buffer, node);
> +
> +               if (b->block == block)
> +                       return b;
> +
> +               if (block <= b->block) {
> +                       n = n->rb_left;
> +                       best = b;
> +               } else
> +                       n = n->rb_right;
> +       }
> +
> +       return best;
> +}
> +
> +static void __remove_range(struct buffer_cache *bc,
> +                          struct rb_root *root,
> +                          sector_t begin, sector_t end,
> +                          b_predicate pred, b_release release)
> +{
> +       struct dm_buffer *b;
> +
> +       while (true) {
> +               cond_resched();
> +
> +               b = __find_next(root, begin);
> +               if (!b || (b->block >= end))
> +                       break;
> +
> +               begin = b->block + 1;
> +
> +               if (atomic_read(&b->hold_count))
> +                       continue;
> +
> +               if (pred(b, NULL) == ER_EVICT) {
> +                       rb_erase(&b->node, root);
> +                       lru_remove(&bc->lru[b->list_mode], &b->lru);
> +                       release(b);
> +               }
> +       }
> +}
> +
> +static void cache_remove_range(struct buffer_cache *bc,
> +                              sector_t begin, sector_t end,
> +                              b_predicate pred, b_release release)
> +{
> +       unsigned int i;
> +
> +       for (i = 0; i < NR_LOCKS; i++) {
> +               down_write(&bc->locks[i].lock);
> +               __remove_range(bc, &bc->roots[i], begin, end, pred,
> release);
> +               up_write(&bc->locks[i].lock);
> +       }
> +}
> +
> +/*----------------------------------------------------------------*/
> +
>  /*
>   * Linking of buffers:
> - *     All buffers are linked to buffer_tree with their node field.
> + *     All buffers are linked to buffer_cache with their node field.
>   *
>   *     Clean buffers that are not being written (B_WRITING not set)
>   *     are linked to lru[LIST_CLEAN] with their lru_list field.
> @@ -136,9 +956,6 @@ struct dm_bufio_client {
>         spinlock_t spinlock;
>         bool no_sleep;
>
> -       struct list_head lru[LIST_SIZE];
> -       unsigned long n_buffers[LIST_SIZE];
> -
>         struct block_device *bdev;
>         unsigned int block_size;
>         s8 sectors_per_block_bits;
> @@ -153,7 +970,7 @@ struct dm_bufio_client {
>
>         unsigned int minimum_buffers;
>
> -       struct rb_root buffer_tree;
> +       struct buffer_cache cache;
>         wait_queue_head_t free_buffer_wait;
>
>         sector_t start;
> @@ -165,6 +982,11 @@ struct dm_bufio_client {
>         struct shrinker shrinker;
>         struct work_struct shrink_work;
>         atomic_long_t need_shrink;
> +
> +       /*
> +        * Used by global_cleanup to sort the clients list.
> +        */
> +       unsigned long oldest_buffer;
>  };
>
>  static DEFINE_STATIC_KEY_FALSE(no_sleep_enabled);
> @@ -181,14 +1003,6 @@ static void dm_bufio_lock(struct dm_bufio_client *c)
>                 mutex_lock_nested(&c->lock, dm_bufio_in_request());
>  }
>
> -static int dm_bufio_trylock(struct dm_bufio_client *c)
> -{
> -       if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep)
> -               return spin_trylock_bh(&c->spinlock);
> -       else
> -               return mutex_trylock(&c->lock);
> -}
> -
>  static void dm_bufio_unlock(struct dm_bufio_client *c)
>  {
>         if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep)
> @@ -217,10 +1031,6 @@ static unsigned long dm_bufio_cache_size_latch;
>
>  static DEFINE_SPINLOCK(global_spinlock);
>
> -static LIST_HEAD(global_queue);
> -
> -static unsigned long global_num;
> -
>  /*
>   * Buffers are freed after this timeout
>   */
> @@ -254,7 +1064,6 @@ static struct workqueue_struct *dm_bufio_wq;
>  static struct delayed_work dm_bufio_cleanup_old_work;
>  static struct work_struct dm_bufio_replacement_work;
>
> -
>  #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
>  static void buffer_record_stack(struct dm_buffer *b)
>  {
> @@ -262,78 +1071,6 @@ static void buffer_record_stack(struct dm_buffer *b)
>  }
>  #endif
>
> -/*
> - *----------------------------------------------------------------
> - * A red/black tree acts as an index for all the buffers.
> - *----------------------------------------------------------------
> - */
> -static struct dm_buffer *__find(struct dm_bufio_client *c, sector_t block)
> -{
> -       struct rb_node *n = c->buffer_tree.rb_node;
> -       struct dm_buffer *b;
> -
> -       while (n) {
> -               b = container_of(n, struct dm_buffer, node);
> -
> -               if (b->block == block)
> -                       return b;
> -
> -               n = block < b->block ? n->rb_left : n->rb_right;
> -       }
> -
> -       return NULL;
> -}
> -
> -static struct dm_buffer *__find_next(struct dm_bufio_client *c, sector_t
> block)
> -{
> -       struct rb_node *n = c->buffer_tree.rb_node;
> -       struct dm_buffer *b;
> -       struct dm_buffer *best = NULL;
> -
> -       while (n) {
> -               b = container_of(n, struct dm_buffer, node);
> -
> -               if (b->block == block)
> -                       return b;
> -
> -               if (block <= b->block) {
> -                       n = n->rb_left;
> -                       best = b;
> -               } else {
> -                       n = n->rb_right;
> -               }
> -       }
> -
> -       return best;
> -}
> -
> -static void __insert(struct dm_bufio_client *c, struct dm_buffer *b)
> -{
> -       struct rb_node **new = &c->buffer_tree.rb_node, *parent = NULL;
> -       struct dm_buffer *found;
> -
> -       while (*new) {
> -               found = container_of(*new, struct dm_buffer, node);
> -
> -               if (found->block == b->block) {
> -                       BUG_ON(found != b);
> -                       return;
> -               }
> -
> -               parent = *new;
> -               new = b->block < found->block ?
> -                       &found->node.rb_left : &found->node.rb_right;
> -       }
> -
> -       rb_link_node(&b->node, parent, new);
> -       rb_insert_color(&b->node, &c->buffer_tree);
> -}
> -
> -static void __remove(struct dm_bufio_client *c, struct dm_buffer *b)
> -{
> -       rb_erase(&b->node, &c->buffer_tree);
> -}
> -
>  /*----------------------------------------------------------------*/
>
>  static void adjust_total_allocated(struct dm_buffer *b, bool unlink)
> @@ -361,16 +1098,9 @@ static void adjust_total_allocated(struct dm_buffer
> *b, bool unlink)
>         if (dm_bufio_current_allocated > dm_bufio_peak_allocated)
>                 dm_bufio_peak_allocated = dm_bufio_current_allocated;
>
> -       b->accessed = 1;
> -
>         if (!unlink) {
> -               list_add(&b->global_list, &global_queue);
> -               global_num++;
>                 if (dm_bufio_current_allocated > dm_bufio_cache_size)
>                         queue_work(dm_bufio_wq,
> &dm_bufio_replacement_work);
> -       } else {
> -               list_del(&b->global_list);
> -               global_num--;
>         }
>
>         spin_unlock(&global_spinlock);
> @@ -486,8 +1216,9 @@ static void free_buffer_data(struct dm_bufio_client
> *c,
>   */
>  static struct dm_buffer *alloc_buffer(struct dm_bufio_client *c, gfp_t
> gfp_mask)
>  {
> -       struct dm_buffer *b = kmem_cache_alloc(c->slab_buffer, gfp_mask);
> +       struct dm_buffer *b;
>
> +       b = kmem_cache_alloc(c->slab_buffer, gfp_mask);
>         if (!b)
>                 return NULL;
>
> @@ -498,6 +1229,7 @@ static struct dm_buffer *alloc_buffer(struct
> dm_bufio_client *c, gfp_t gfp_mask)
>                 kmem_cache_free(c->slab_buffer, b);
>                 return NULL;
>         }
> +       adjust_total_allocated(b, false);
>
>  #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
>         b->stack_len = 0;
> @@ -512,61 +1244,11 @@ static void free_buffer(struct dm_buffer *b)
>  {
>         struct dm_bufio_client *c = b->c;
>
> +       adjust_total_allocated(b, true);
>         free_buffer_data(c, b->data, b->data_mode);
>         kmem_cache_free(c->slab_buffer, b);
>  }
>
> -/*
> - * Link buffer to the buffer tree and clean or dirty queue.
> - */
> -static void __link_buffer(struct dm_buffer *b, sector_t block, int dirty)
> -{
> -       struct dm_bufio_client *c = b->c;
> -
> -       c->n_buffers[dirty]++;
> -       b->block = block;
> -       b->list_mode = dirty;
> -       list_add(&b->lru_list, &c->lru[dirty]);
> -       __insert(b->c, b);
> -       b->last_accessed = jiffies;
> -
> -       adjust_total_allocated(b, false);
> -}
> -
> -/*
> - * Unlink buffer from the buffer tree and dirty or clean queue.
> - */
> -static void __unlink_buffer(struct dm_buffer *b)
> -{
> -       struct dm_bufio_client *c = b->c;
> -
> -       BUG_ON(!c->n_buffers[b->list_mode]);
> -
> -       c->n_buffers[b->list_mode]--;
> -       __remove(b->c, b);
> -       list_del(&b->lru_list);
> -
> -       adjust_total_allocated(b, true);
> -}
> -
> -/*
> - * Place the buffer to the head of dirty or clean LRU queue.
> - */
> -static void __relink_lru(struct dm_buffer *b, int dirty)
> -{
> -       struct dm_bufio_client *c = b->c;
> -
> -       b->accessed = 1;
> -
> -       BUG_ON(!c->n_buffers[b->list_mode]);
> -
> -       c->n_buffers[b->list_mode]--;
> -       c->n_buffers[dirty]++;
> -       b->list_mode = dirty;
> -       list_move(&b->lru_list, &c->lru[dirty]);
> -       b->last_accessed = jiffies;
> -}
> -
>  /*
>
> *--------------------------------------------------------------------------
>   * Submit I/O on the buffer.
> @@ -806,7 +1488,7 @@ static void __flush_write_list(struct list_head
> *write_list)
>   */
>  static void __make_buffer_clean(struct dm_buffer *b)
>  {
> -       BUG_ON(b->hold_count);
> +       BUG_ON(atomic_read(&b->hold_count));
>
>         /* smp_load_acquire() pairs with read_endio()'s
> smp_mb__before_atomic() */
>         if (!smp_load_acquire(&b->state))       /* fast case */
> @@ -817,6 +1499,26 @@ static void __make_buffer_clean(struct dm_buffer *b)
>         wait_on_bit_io(&b->state, B_WRITING, TASK_UNINTERRUPTIBLE);
>  }
>
> +static enum evict_result is_clean(struct dm_buffer *b, void *context)
> +{
> +       struct dm_bufio_client *c = context;
> +
> +       BUG_ON(test_bit(B_WRITING, &b->state));
> +       BUG_ON(test_bit(B_DIRTY, &b->state));
> +
> +       if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep &&
> +           unlikely(test_bit(B_READING, &b->state)))
> +               return ER_DONT_EVICT;
> +
> +       return ER_EVICT;
> +}
> +
> +static enum evict_result is_dirty(struct dm_buffer *b, void *context)
> +{
> +       BUG_ON(test_bit(B_READING, &b->state));
> +       return ER_EVICT;
> +}
> +
>  /*
>   * Find some buffer that is not held by anybody, clean it, unlink it and
>   * return it.
> @@ -825,34 +1527,20 @@ static struct dm_buffer
> *__get_unclaimed_buffer(struct dm_bufio_client *c)
>  {
>         struct dm_buffer *b;
>
> -       list_for_each_entry_reverse(b, &c->lru[LIST_CLEAN], lru_list) {
> -               BUG_ON(test_bit(B_WRITING, &b->state));
> -               BUG_ON(test_bit(B_DIRTY, &b->state));
> -
> -               if (static_branch_unlikely(&no_sleep_enabled) &&
> c->no_sleep &&
> -                   unlikely(test_bit_acquire(B_READING, &b->state)))
> -                       continue;
> -
> -               if (!b->hold_count) {
> -                       __make_buffer_clean(b);
> -                       __unlink_buffer(b);
> -                       return b;
> -               }
> -               cond_resched();
> +       b = cache_evict(&c->cache, LIST_CLEAN, is_clean, c);
> +       if (b) {
> +               /* this also waits for pending reads */
> +               __make_buffer_clean(b);
> +               return b;
>         }
>
>         if (static_branch_unlikely(&no_sleep_enabled) && c->no_sleep)
>                 return NULL;
>
> -       list_for_each_entry_reverse(b, &c->lru[LIST_DIRTY], lru_list) {
> -               BUG_ON(test_bit(B_READING, &b->state));
> -
> -               if (!b->hold_count) {
> -                       __make_buffer_clean(b);
> -                       __unlink_buffer(b);
> -                       return b;
> -               }
> -               cond_resched();
> +       b = cache_evict(&c->cache, LIST_DIRTY, is_dirty, NULL);
> +       if (b) {
> +               __make_buffer_clean(b);
> +               return b;
>         }
>
>         return NULL;
> @@ -873,7 +1561,12 @@ static void __wait_for_free_buffer(struct
> dm_bufio_client *c)
>         set_current_state(TASK_UNINTERRUPTIBLE);
>         dm_bufio_unlock(c);
>
> -       io_schedule();
> +       /*
> +        * It's possible to miss a wake up event since we don't always
> +        * hold c->lock when wake_up is called.  So we have a timeout here,
> +        * just in case.
> +        */
> +       io_schedule_timeout(5 * HZ);
>
>         remove_wait_queue(&c->free_buffer_wait, &wait);
>
> @@ -931,9 +1624,8 @@ static struct dm_buffer
> *__alloc_buffer_wait_no_callback(struct dm_bufio_client
>                 }
>
>                 if (!list_empty(&c->reserved_buffers)) {
> -                       b = list_entry(c->reserved_buffers.next,
> -                                      struct dm_buffer, lru_list);
> -                       list_del(&b->lru_list);
> +                       b = list_to_buffer(c->reserved_buffers.next);
> +                       list_del(&b->lru.list);
>                         c->need_reserved_buffers++;
>
>                         return b;
> @@ -967,36 +1659,55 @@ static void __free_buffer_wake(struct dm_buffer *b)
>  {
>         struct dm_bufio_client *c = b->c;
>
> +       b->block = -1;
>         if (!c->need_reserved_buffers)
>                 free_buffer(b);
>         else {
> -               list_add(&b->lru_list, &c->reserved_buffers);
> +               list_add(&b->lru.list, &c->reserved_buffers);
>                 c->need_reserved_buffers--;
>         }
>
>         wake_up(&c->free_buffer_wait);
>  }
>
> +static enum evict_result cleaned(struct dm_buffer *b, void *context)
> +{
> +       BUG_ON(test_bit(B_READING, &b->state));
> +
> +       if (test_bit(B_DIRTY, &b->state) || test_bit(B_WRITING, &b->state))
> +               return ER_DONT_EVICT;
> +       else
> +               return ER_EVICT;
> +}
> +
> +static void __move_clean_buffers(struct dm_bufio_client *c)
> +{
> +       cache_mark_many(&c->cache, LIST_DIRTY, LIST_CLEAN, cleaned, NULL);
> +}
> +
> +struct write_context {
> +       int no_wait;
> +       struct list_head *write_list;
> +};
> +
> +static enum it_action write_one(struct dm_buffer *b, void *context)
> +{
> +       struct write_context *wc = context;
> +
> +       if (wc->no_wait && test_bit(B_WRITING, &b->state))
> +               return IT_COMPLETE;
> +
> +       __write_dirty_buffer(b, wc->write_list);
> +       return IT_NEXT;
> +}
> +
>  static void __write_dirty_buffers_async(struct dm_bufio_client *c, int
> no_wait,
>                                         struct list_head *write_list)
>  {
> -       struct dm_buffer *b, *tmp;
> +       struct write_context wc = {.no_wait = no_wait, .write_list =
> write_list};
>
> -       list_for_each_entry_safe_reverse(b, tmp, &c->lru[LIST_DIRTY],
> lru_list) {
> -               BUG_ON(test_bit(B_READING, &b->state));
> -
> -               if (!test_bit(B_DIRTY, &b->state) &&
> -                   !test_bit(B_WRITING, &b->state)) {
> -                       __relink_lru(b, LIST_CLEAN);
> -                       continue;
> -               }
> -
> -               if (no_wait && test_bit(B_WRITING, &b->state))
> -                       return;
> -
> -               __write_dirty_buffer(b, write_list);
> -               cond_resched();
> -       }
> +       __move_clean_buffers(c);
> +       cache_iterate(&c->cache, LIST_DIRTY, write_one, &wc);
>  }
>
>  /*
> @@ -1007,7 +1718,8 @@ static void __write_dirty_buffers_async(struct
> dm_bufio_client *c, int no_wait,
>  static void __check_watermark(struct dm_bufio_client *c,
>                               struct list_head *write_list)
>  {
> -       if (c->n_buffers[LIST_DIRTY] > c->n_buffers[LIST_CLEAN] *
> DM_BUFIO_WRITEBACK_RATIO)
> +       if (cache_count(&c->cache, LIST_DIRTY) >
> +               cache_count(&c->cache, LIST_CLEAN) *
> DM_BUFIO_WRITEBACK_RATIO)
>                 __write_dirty_buffers_async(c, 1, write_list);
>  }
>
> @@ -1017,20 +1729,30 @@ static void __check_watermark(struct
> dm_bufio_client *c,
>   *--------------------------------------------------------------
>   */
>
> +static void cache_put_and_wake(struct dm_bufio_client *c, struct
> dm_buffer *b)
> +{
> +       /*
> +        * Relying on waitqueue_active() is racey, but we sleep
> +        * with schedule_timeout anyway.
> +        */
> +       if (cache_put(&c->cache, b) &&
> +           unlikely(waitqueue_active(&c->free_buffer_wait)))
> +               wake_up(&c->free_buffer_wait);
> +}
> +
> +/*
> + * This assumes you have already checked the cache to see if the buffer
> + * is already present (it will recheck after dropping the lock for
> allocation).
> + */
>  static struct dm_buffer *__bufio_new(struct dm_bufio_client *c, sector_t
> block,
>                                      enum new_flag nf, int *need_submit,
>                                      struct list_head *write_list)
>  {
>         struct dm_buffer *b, *new_b = NULL;
> -
>         *need_submit = 0;
>
> -       b = __find(c, block);
> -       if (b)
> -               goto found_buffer;
> -
> -       if (nf == NF_GET)
> -               return NULL;
> +       /* This can't be called with NF_GET */
> +       BUG_ON(nf == NF_GET);
>
>         new_b = __alloc_buffer_wait(c, nf);
>         if (!new_b)
> @@ -1040,7 +1762,7 @@ static struct dm_buffer *__bufio_new(struct
> dm_bufio_client *c, sector_t block,
>          * We've had a period where the mutex was unlocked, so need to
>          * recheck the buffer tree.
>          */
> -       b = __find(c, block);
> +       b = cache_get(&c->cache, block);
>         if (b) {
>                 __free_buffer_wake(new_b);
>                 goto found_buffer;
> @@ -1049,24 +1771,35 @@ static struct dm_buffer *__bufio_new(struct
> dm_bufio_client *c, sector_t block,
>         __check_watermark(c, write_list);
>
>         b = new_b;
> -       b->hold_count = 1;
> +       atomic_set(&b->hold_count, 1);
> +       WRITE_ONCE(b->last_accessed, jiffies);
> +       b->block = block;
>         b->read_error = 0;
>         b->write_error = 0;
> -       __link_buffer(b, block, LIST_CLEAN);
> +       b->list_mode = LIST_CLEAN;
>
> -       if (nf == NF_FRESH) {
> +       if (nf == NF_FRESH)
>                 b->state = 0;
> -               return b;
> +       else {
> +               b->state = 1 << B_READING;
> +               *need_submit = 1;
>         }
>
> -       b->state = 1 << B_READING;
> -       *need_submit = 1;
> +       /*
> +        * We mustn't insert into the cache until the B_READING state
> +        * is set.  Otherwise another thread could get it and use
> +        * it before it had been read.
> +        */
> +       cache_insert(&c->cache, b);
>
>         return b;
>
>  found_buffer:
> -       if (nf == NF_PREFETCH)
> +       if (nf == NF_PREFETCH) {
> +               cache_put_and_wake(c, b);
>                 return NULL;
> +       }
> +
>         /*
>          * Note: it is essential that we don't wait for the buffer to be
>          * read if dm_bufio_get function is used. Both dm_bufio_get and
> @@ -1074,12 +1807,11 @@ static struct dm_buffer *__bufio_new(struct
> dm_bufio_client *c, sector_t block,
>          * If the user called both dm_bufio_prefetch and dm_bufio_get on
>          * the same buffer, it would deadlock if we waited.
>          */
> -       if (nf == NF_GET && unlikely(test_bit_acquire(B_READING,
> &b->state)))
> +       if (nf == NF_GET && unlikely(test_bit_acquire(B_READING,
> &b->state))) {
> +               cache_put_and_wake(c, b);
>                 return NULL;
> +       }
>
> -       b->hold_count++;
> -       __relink_lru(b, test_bit(B_DIRTY, &b->state) ||
> -                    test_bit(B_WRITING, &b->state));
>         return b;
>  }
>
> @@ -1109,18 +1841,49 @@ static void read_endio(struct dm_buffer *b,
> blk_status_t status)
>  static void *new_read(struct dm_bufio_client *c, sector_t block,
>                       enum new_flag nf, struct dm_buffer **bp)
>  {
> -       int need_submit;
> +       int need_submit = 0;
>         struct dm_buffer *b;
>
>         LIST_HEAD(write_list);
> +       *bp = NULL;
> +
> +       /*
> +        * Fast path, hopefully the block is already in the cache.  No need
> +        * to get the client lock for this.
> +        */
> +       b = cache_get(&c->cache, block);
> +       if (b) {
> +               if (nf == NF_PREFETCH) {
> +                       cache_put_and_wake(c, b);
> +                       return NULL;
> +               }
> +
> +               /*
> +                * Note: it is essential that we don't wait for the buffer
> to be
> +                * read if dm_bufio_get function is used. Both
> dm_bufio_get and
> +                * dm_bufio_prefetch can be used in the driver request
> routine.
> +                * If the user called both dm_bufio_prefetch and
> dm_bufio_get on
> +                * the same buffer, it would deadlock if we waited.
> +                */
> +               if (nf == NF_GET && unlikely(test_bit_acquire(B_READING,
> &b->state))) {
> +                       cache_put_and_wake(c, b);
> +                       return NULL;
> +               }
> +       }
> +
> +       if (!b) {
> +               if (nf == NF_GET)
> +                       return NULL;
> +
> +               dm_bufio_lock(c);
> +               b = __bufio_new(c, block, nf, &need_submit, &write_list);
> +               dm_bufio_unlock(c);
> +       }
>
> -       dm_bufio_lock(c);
> -       b = __bufio_new(c, block, nf, &need_submit, &write_list);
>  #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
> -       if (b && b->hold_count == 1)
> +       if (b && (atomic_read(&b->hold_count) == 1))
>                 buffer_record_stack(b);
>  #endif
> -       dm_bufio_unlock(c);
>
>         __flush_write_list(&write_list);
>
> @@ -1141,7 +1904,6 @@ static void *new_read(struct dm_bufio_client *c,
> sector_t block,
>         }
>
>         *bp = b;
> -
>         return b->data;
>  }
>
> @@ -1156,7 +1918,6 @@ void *dm_bufio_read(struct dm_bufio_client *c,
> sector_t block,
>                     struct dm_buffer **bp)
>  {
>         BUG_ON(dm_bufio_in_request());
> -
>         return new_read(c, block, NF_READ, bp);
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_read);
> @@ -1165,7 +1926,6 @@ void *dm_bufio_new(struct dm_bufio_client *c,
> sector_t block,
>                    struct dm_buffer **bp)
>  {
>         BUG_ON(dm_bufio_in_request());
> -
>         return new_read(c, block, NF_FRESH, bp);
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_new);
> @@ -1180,12 +1940,19 @@ void dm_bufio_prefetch(struct dm_bufio_client *c,
>         BUG_ON(dm_bufio_in_request());
>
>         blk_start_plug(&plug);
> -       dm_bufio_lock(c);
>
>         for (; n_blocks--; block++) {
>                 int need_submit;
>                 struct dm_buffer *b;
>
> +               b = cache_get(&c->cache, block);
> +               if (b) {
> +                       /* already in cache */
> +                       cache_put_and_wake(c, b);
> +                       continue;
> +               }
> +
> +               dm_bufio_lock(c);
>                 b = __bufio_new(c, block, NF_PREFETCH, &need_submit,
>                                 &write_list);
>                 if (unlikely(!list_empty(&write_list))) {
> @@ -1195,6 +1962,7 @@ void dm_bufio_prefetch(struct dm_bufio_client *c,
>                         blk_start_plug(&plug);
>                         dm_bufio_lock(c);
>                 }
> +
>                 if (unlikely(b != NULL)) {
>                         dm_bufio_unlock(c);
>
> @@ -1208,10 +1976,9 @@ void dm_bufio_prefetch(struct dm_bufio_client *c,
>                                 goto flush_plug;
>                         dm_bufio_lock(c);
>                 }
> +               dm_bufio_unlock(c);
>         }
>
> -       dm_bufio_unlock(c);
> -
>  flush_plug:
>         blk_finish_plug(&plug);
>  }
> @@ -1221,29 +1988,28 @@ void dm_bufio_release(struct dm_buffer *b)
>  {
>         struct dm_bufio_client *c = b->c;
>
> -       dm_bufio_lock(c);
> +       /*
> +        * If there were errors on the buffer, and the buffer is not
> +        * to be written, free the buffer. There is no point in caching
> +        * invalid buffer.
> +        */
> +       if ((b->read_error || b->write_error) &&
> +           !test_bit_acquire(B_READING, &b->state) &&
> +           !test_bit(B_WRITING, &b->state) &&
> +           !test_bit(B_DIRTY, &b->state)) {
> +               dm_bufio_lock(c);
>
> -       BUG_ON(!b->hold_count);
> -
> -       b->hold_count--;
> -       if (!b->hold_count) {
> -               wake_up(&c->free_buffer_wait);
> -
> -               /*
> -                * If there were errors on the buffer, and the buffer is
> not
> -                * to be written, free the buffer. There is no point in
> caching
> -                * invalid buffer.
> -                */
> -               if ((b->read_error || b->write_error) &&
> -                   !test_bit_acquire(B_READING, &b->state) &&
> -                   !test_bit(B_WRITING, &b->state) &&
> -                   !test_bit(B_DIRTY, &b->state)) {
> -                       __unlink_buffer(b);
> +               /* cache remove can fail if there are other holders */
> +               if (cache_remove(&c->cache, b)) {
>                         __free_buffer_wake(b);
> +                       dm_bufio_unlock(c);
> +                       return;
>                 }
> +
> +               dm_bufio_unlock(c);
>         }
>
> -       dm_bufio_unlock(c);
> +       cache_put_and_wake(c, b);
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_release);
>
> @@ -1262,7 +2028,7 @@ void dm_bufio_mark_partial_buffer_dirty(struct
> dm_buffer *b,
>         if (!test_and_set_bit(B_DIRTY, &b->state)) {
>                 b->dirty_start = start;
>                 b->dirty_end = end;
> -               __relink_lru(b, LIST_DIRTY);
> +               cache_mark(&c->cache, b, LIST_DIRTY);
>         } else {
>                 if (start < b->dirty_start)
>                         b->dirty_start = start;
> @@ -1300,11 +2066,19 @@
> EXPORT_SYMBOL_GPL(dm_bufio_write_dirty_buffers_async);
>   *
>   * Finally, we flush hardware disk cache.
>   */
> +static bool is_writing(struct lru_entry *e, void *context)
> +{
> +       struct dm_buffer *b = le_to_buffer(e);
> +
> +       return test_bit(B_WRITING, &b->state);
> +}
> +
>  int dm_bufio_write_dirty_buffers(struct dm_bufio_client *c)
>  {
>         int a, f;
> -       unsigned long buffers_processed = 0;
> -       struct dm_buffer *b, *tmp;
> +       unsigned long nr_buffers;
> +       struct lru_entry *e;
> +       struct lru_iter it;
>
>         LIST_HEAD(write_list);
>
> @@ -1314,52 +2088,32 @@ int dm_bufio_write_dirty_buffers(struct
> dm_bufio_client *c)
>         __flush_write_list(&write_list);
>         dm_bufio_lock(c);
>
> -again:
> -       list_for_each_entry_safe_reverse(b, tmp, &c->lru[LIST_DIRTY],
> lru_list) {
> -               int dropped_lock = 0;
> -
> -               if (buffers_processed < c->n_buffers[LIST_DIRTY])
> -                       buffers_processed++;
> +       nr_buffers = cache_count(&c->cache, LIST_DIRTY);
> +       lru_iter_begin(&c->cache.lru[LIST_DIRTY], &it);
> +       while ((e = lru_iter_next(&it, is_writing, c))) {
> +               struct dm_buffer *b = le_to_buffer(e);
> +               __cache_inc_buffer(b);
>
>                 BUG_ON(test_bit(B_READING, &b->state));
>
> -               if (test_bit(B_WRITING, &b->state)) {
> -                       if (buffers_processed < c->n_buffers[LIST_DIRTY]) {
> -                               dropped_lock = 1;
> -                               b->hold_count++;
> -                               dm_bufio_unlock(c);
> -                               wait_on_bit_io(&b->state, B_WRITING,
> -                                              TASK_UNINTERRUPTIBLE);
> -                               dm_bufio_lock(c);
> -                               b->hold_count--;
> -                       } else
> -                               wait_on_bit_io(&b->state, B_WRITING,
> -                                              TASK_UNINTERRUPTIBLE);
> -               }
> +               if (nr_buffers) {
> +                       nr_buffers--;
> +                       dm_bufio_unlock(c);
> +                       wait_on_bit_io(&b->state, B_WRITING,
> TASK_UNINTERRUPTIBLE);
> +                       dm_bufio_lock(c);
> +               } else
> +                       wait_on_bit_io(&b->state, B_WRITING,
> TASK_UNINTERRUPTIBLE);
>
> -               if (!test_bit(B_DIRTY, &b->state) &&
> -                   !test_bit(B_WRITING, &b->state))
> -                       __relink_lru(b, LIST_CLEAN);
> +               if (!test_bit(B_DIRTY, &b->state) && !test_bit(B_WRITING,
> &b->state))
> +                       cache_mark(&c->cache, b, LIST_CLEAN);
> +
> +               cache_put_and_wake(c, b);
>
>                 cond_resched();
> -
> -               /*
> -                * If we dropped the lock, the list is no longer
> consistent,
> -                * so we must restart the search.
> -                *
> -                * In the most common case, the buffer just processed is
> -                * relinked to the clean list, so we won't loop scanning
> the
> -                * same buffer again and again.
> -                *
> -                * This may livelock if there is another thread
> simultaneously
> -                * dirtying buffers, so we count the number of buffers
> walked
> -                * and if it exceeds the total number of buffers, it means
> that
> -                * someone is doing some writes simultaneously with us.  In
> -                * this case, stop, dropping the lock.
> -                */
> -               if (dropped_lock)
> -                       goto again;
>         }
> +
> +       lru_iter_end(&it);
> +
>         wake_up(&c->free_buffer_wait);
>         dm_bufio_unlock(c);
>
> @@ -1418,12 +2172,22 @@ int dm_bufio_issue_discard(struct dm_bufio_client
> *c, sector_t block, sector_t c
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_issue_discard);
>
> -static void forget_buffer_locked(struct dm_buffer *b)
> +static bool forget_buffer(struct dm_bufio_client *c, sector_t block)
>  {
> -       if (likely(!b->hold_count) &&
> likely(!smp_load_acquire(&b->state))) {
> -               __unlink_buffer(b);
> -               __free_buffer_wake(b);
> +       struct dm_buffer *b;
> +
> +       b = cache_get(&c->cache, block);
> +       if (b) {
> +               if (likely(!smp_load_acquire(&b->state))) {
> +                       if (cache_remove(&c->cache, b))
> +                               __free_buffer_wake(b);
> +                       else
> +                               cache_put_and_wake(c, b);
> +               } else
> +                       cache_put_and_wake(c, b);
>         }
> +
> +       return b ? true : false;
>  }
>
>  /*
> @@ -1434,38 +2198,22 @@ static void forget_buffer_locked(struct dm_buffer
> *b)
>   */
>  void dm_bufio_forget(struct dm_bufio_client *c, sector_t block)
>  {
> -       struct dm_buffer *b;
> -
>         dm_bufio_lock(c);
> -
> -       b = __find(c, block);
> -       if (b)
> -               forget_buffer_locked(b);
> -
> +       forget_buffer(c, block);
>         dm_bufio_unlock(c);
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_forget);
>
> +static enum evict_result idle(struct dm_buffer *b, void *context)
> +{
> +       return b->state ? ER_DONT_EVICT : ER_EVICT;
> +}
> +
>  void dm_bufio_forget_buffers(struct dm_bufio_client *c, sector_t block,
> sector_t n_blocks)
>  {
> -       struct dm_buffer *b;
> -       sector_t end_block = block + n_blocks;
> -
> -       while (block < end_block) {
> -               dm_bufio_lock(c);
> -
> -               b = __find_next(c, block);
> -               if (b) {
> -                       block = b->block + 1;
> -                       forget_buffer_locked(b);
> -               }
> -
> -               dm_bufio_unlock(c);
> -
> -               if (!b)
> -                       break;
> -       }
> -
> +       dm_bufio_lock(c);
> +       cache_remove_range(&c->cache, block, block + n_blocks, idle,
> __free_buffer_wake);
> +       dm_bufio_unlock(c);
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_forget_buffers);
>
> @@ -1527,11 +2275,26 @@ struct dm_bufio_client *dm_bufio_get_client(struct
> dm_buffer *b)
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_get_client);
>
> +static enum it_action warn_leak(struct dm_buffer *b, void *context)
> +{
> +       bool *warned = context;
> +
> +       WARN_ON(!(*warned));
> +       *warned = true;
> +       DMERR("leaked buffer %llx, hold count %u, list %d",
> +             (unsigned long long)b->block, atomic_read(&b->hold_count),
> b->list_mode);
> +#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
> +       stack_trace_print(b->stack_entries, b->stack_len, 1);
> +       /* mark unclaimed to avoid BUG_ON below */
> +       atomic_set(&b->hold_count, 0);
> +#endif
> +       return IT_NEXT;
> +}
> +
>  static void drop_buffers(struct dm_bufio_client *c)
>  {
> -       struct dm_buffer *b;
>         int i;
> -       bool warned = false;
> +       struct dm_buffer *b;
>
>         BUG_ON(dm_bufio_in_request());
>
> @@ -1545,18 +2308,11 @@ static void drop_buffers(struct dm_bufio_client *c)
>         while ((b = __get_unclaimed_buffer(c)))
>                 __free_buffer_wake(b);
>
> -       for (i = 0; i < LIST_SIZE; i++)
> -               list_for_each_entry(b, &c->lru[i], lru_list) {
> -                       WARN_ON(!warned);
> -                       warned = true;
> -                       DMERR("leaked buffer %llx, hold count %u, list %d",
> -                             (unsigned long long)b->block, b->hold_count,
> i);
> -#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
> -                       stack_trace_print(b->stack_entries, b->stack_len,
> 1);
> -                       /* mark unclaimed to avoid BUG_ON below */
> -                       b->hold_count = 0;
> -#endif
> -               }
> +       for (i = 0; i < LIST_SIZE; i++) {
> +               bool warned = false;
> +
> +               cache_iterate(&c->cache, i, warn_leak, &warned);
> +       }
>
>  #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
>         while ((b = __get_unclaimed_buffer(c)))
> @@ -1564,39 +2320,11 @@ static void drop_buffers(struct dm_bufio_client *c)
>  #endif
>
>         for (i = 0; i < LIST_SIZE; i++)
> -               BUG_ON(!list_empty(&c->lru[i]));
> +               BUG_ON(cache_count(&c->cache, i));
>
>         dm_bufio_unlock(c);
>  }
>
> -/*
> - * We may not be able to evict this buffer if IO pending or the client
> - * is still using it.  Caller is expected to know buffer is too old.
> - *
> - * And if GFP_NOFS is used, we must not do any I/O because we hold
> - * dm_bufio_clients_lock and we would risk deadlock if the I/O gets
> - * rerouted to different bufio client.
> - */
> -static bool __try_evict_buffer(struct dm_buffer *b, gfp_t gfp)
> -{
> -       if (!(gfp & __GFP_FS) ||
> -           (static_branch_unlikely(&no_sleep_enabled) && b->c->no_sleep))
> {
> -               if (test_bit_acquire(B_READING, &b->state) ||
> -                   test_bit(B_WRITING, &b->state) ||
> -                   test_bit(B_DIRTY, &b->state))
> -                       return false;
> -       }
> -
> -       if (b->hold_count)
> -               return false;
> -
> -       __make_buffer_clean(b);
> -       __unlink_buffer(b);
> -       __free_buffer_wake(b);
> -
> -       return true;
> -}
> -
>  static unsigned long get_retain_buffers(struct dm_bufio_client *c)
>  {
>         unsigned long retain_bytes = READ_ONCE(dm_bufio_retain_bytes);
> @@ -1612,22 +2340,30 @@ static unsigned long get_retain_buffers(struct
> dm_bufio_client *c)
>  static void __scan(struct dm_bufio_client *c)
>  {
>         int l;
> -       struct dm_buffer *b, *tmp;
> +       struct dm_buffer *b;
>         unsigned long freed = 0;
> -       unsigned long count = c->n_buffers[LIST_CLEAN] +
> -                             c->n_buffers[LIST_DIRTY];
>         unsigned long retain_target = get_retain_buffers(c);
> +       unsigned long count = cache_total(&c->cache);
>
>         for (l = 0; l < LIST_SIZE; l++) {
> -               list_for_each_entry_safe_reverse(b, tmp, &c->lru[l],
> lru_list) {
> +               while (true) {
>                         if (count - freed <= retain_target)
>                                 atomic_long_set(&c->need_shrink, 0);
> +
>                         if (!atomic_long_read(&c->need_shrink))
> -                               return;
> -                       if (__try_evict_buffer(b, GFP_KERNEL)) {
> -                               atomic_long_dec(&c->need_shrink);
> -                               freed++;
> -                       }
> +                               break;
> +
> +                       b = cache_evict(&c->cache, l,
> +                                       l == LIST_CLEAN ? is_clean :
> is_dirty, c);
> +                       if (!b)
> +                               break;
> +
> +                       BUG_ON(b->list_mode != l);
> +                       __make_buffer_clean(b);
> +                       __free_buffer_wake(b);
> +
> +                       atomic_long_dec(&c->need_shrink);
> +                       freed++;
>                         cond_resched();
>                 }
>         }
> @@ -1656,8 +2392,7 @@ static unsigned long dm_bufio_shrink_scan(struct
> shrinker *shrink, struct shrink
>  static unsigned long dm_bufio_shrink_count(struct shrinker *shrink,
> struct shrink_control *sc)
>  {
>         struct dm_bufio_client *c = container_of(shrink, struct
> dm_bufio_client, shrinker);
> -       unsigned long count = READ_ONCE(c->n_buffers[LIST_CLEAN]) +
> -                             READ_ONCE(c->n_buffers[LIST_DIRTY]);
> +       unsigned long count = cache_total(&c->cache);
>         unsigned long retain_target = get_retain_buffers(c);
>         unsigned long queued_for_cleanup =
> atomic_long_read(&c->need_shrink);
>
> @@ -1685,7 +2420,6 @@ struct dm_bufio_client
> *dm_bufio_client_create(struct block_device *bdev, unsign
>  {
>         int r;
>         struct dm_bufio_client *c;
> -       unsigned int i;
>         char slab_name[27];
>
>         if (!block_size || block_size & ((1 << SECTOR_SHIFT) - 1)) {
> @@ -1699,7 +2433,7 @@ struct dm_bufio_client
> *dm_bufio_client_create(struct block_device *bdev, unsign
>                 r = -ENOMEM;
>                 goto bad_client;
>         }
> -       c->buffer_tree = RB_ROOT;
> +       cache_init(&c->cache);
>
>         c->bdev = bdev;
>         c->block_size = block_size;
> @@ -1716,11 +2450,6 @@ struct dm_bufio_client
> *dm_bufio_client_create(struct block_device *bdev, unsign
>                 static_branch_inc(&no_sleep_enabled);
>         }
>
> -       for (i = 0; i < LIST_SIZE; i++) {
> -               INIT_LIST_HEAD(&c->lru[i]);
> -               c->n_buffers[i] = 0;
> -       }
> -
>         mutex_init(&c->lock);
>         spin_lock_init(&c->spinlock);
>         INIT_LIST_HEAD(&c->reserved_buffers);
> @@ -1792,9 +2521,9 @@ struct dm_bufio_client
> *dm_bufio_client_create(struct block_device *bdev, unsign
>
>  bad:
>         while (!list_empty(&c->reserved_buffers)) {
> -               struct dm_buffer *b = list_entry(c->reserved_buffers.next,
> -                                                struct dm_buffer,
> lru_list);
> -               list_del(&b->lru_list);
> +               struct dm_buffer *b =
> list_to_buffer(c->reserved_buffers.next);
> +
> +               list_del(&b->lru.list);
>                 free_buffer(b);
>         }
>         kmem_cache_destroy(c->slab_cache);
> @@ -1831,23 +2560,23 @@ void dm_bufio_client_destroy(struct
> dm_bufio_client *c)
>
>         mutex_unlock(&dm_bufio_clients_lock);
>
> -       BUG_ON(!RB_EMPTY_ROOT(&c->buffer_tree));
>         BUG_ON(c->need_reserved_buffers);
>
>         while (!list_empty(&c->reserved_buffers)) {
> -               struct dm_buffer *b = list_entry(c->reserved_buffers.next,
> -                                                struct dm_buffer,
> lru_list);
> -               list_del(&b->lru_list);
> +               struct dm_buffer *b =
> list_to_buffer(c->reserved_buffers.next);
> +
> +               list_del(&b->lru.list);
>                 free_buffer(b);
>         }
>
>         for (i = 0; i < LIST_SIZE; i++)
> -               if (c->n_buffers[i])
> -                       DMERR("leaked buffer count %d: %ld", i,
> c->n_buffers[i]);
> +               if (cache_count(&c->cache, i))
> +                       DMERR("leaked buffer count %d: %lu", i,
> cache_count(&c->cache, i));
>
>         for (i = 0; i < LIST_SIZE; i++)
> -               BUG_ON(c->n_buffers[i]);
> +               BUG_ON(cache_count(&c->cache, i));
>
> +       cache_destroy(&c->cache);
>         kmem_cache_destroy(c->slab_cache);
>         kmem_cache_destroy(c->slab_buffer);
>         dm_io_client_destroy(c->dm_io);
> @@ -1864,6 +2593,8 @@ void dm_bufio_set_sector_offset(struct
> dm_bufio_client *c, sector_t start)
>  }
>  EXPORT_SYMBOL_GPL(dm_bufio_set_sector_offset);
>
> +/*--------------------------------------------------------------*/
> +
>  static unsigned int get_max_age_hz(void)
>  {
>         unsigned int max_age = READ_ONCE(dm_bufio_max_age);
> @@ -1876,13 +2607,74 @@ static unsigned int get_max_age_hz(void)
>
>  static bool older_than(struct dm_buffer *b, unsigned long age_hz)
>  {
> -       return time_after_eq(jiffies, b->last_accessed + age_hz);
> +       return time_after_eq(jiffies, READ_ONCE(b->last_accessed) +
> age_hz);
>  }
>
> -static void __evict_old_buffers(struct dm_bufio_client *c, unsigned long
> age_hz)
> +struct evict_params {
> +       gfp_t gfp;
> +       unsigned long age_hz;
> +
> +       /*
> +        * This gets updated with the largest last_accessed (ie. most
> +        * recently used) of the evicted buffers.  It will not be
> reinitialised
> +        * by __evict_many(), so you can use it across multiple
> invocations.
> +        */
> +       unsigned long last_accessed;
> +};
> +
> +/*
> + * We may not be able to evict this buffer if IO pending or the client
> + * is still using it.
> + *
> + * And if GFP_NOFS is used, we must not do any I/O because we hold
> + * dm_bufio_clients_lock and we would risk deadlock if the I/O gets
> + * rerouted to different bufio client.
> + */
> +static enum evict_result select_for_evict(struct dm_buffer *b, void
> *context)
> +{
> +       struct evict_params *params = context;
> +
> +       if (!(params->gfp & __GFP_FS) ||
> +           (static_branch_unlikely(&no_sleep_enabled) && b->c->no_sleep))
> {
> +               if (test_bit_acquire(B_READING, &b->state) ||
> +                   test_bit(B_WRITING, &b->state) ||
> +                   test_bit(B_DIRTY, &b->state))
> +                       return ER_DONT_EVICT;
> +       }
> +
> +       return older_than(b, params->age_hz) ? ER_EVICT : ER_STOP;
> +}
> +
> +static unsigned int __evict_many(struct dm_bufio_client *c,
> +                                struct evict_params *params,
> +                                int list_mode, unsigned int max_count)
> +{
> +       unsigned int count;
> +       unsigned long last_accessed;
> +       struct dm_buffer *b;
> +
> +       for (count = 0; count < max_count; count++) {
> +               b = cache_evict(&c->cache, list_mode, select_for_evict,
> params);
> +               if (!b)
> +                       break;
> +
> +               last_accessed = READ_ONCE(b->last_accessed);
> +               if (time_after_eq(params->last_accessed, last_accessed))
> +                       params->last_accessed = last_accessed;
> +
> +               __make_buffer_clean(b);
> +               __free_buffer_wake(b);
> +
> +               cond_resched();
> +       }
> +
> +       return count;
> +}
> +
> +static void evict_old_buffers(struct dm_bufio_client *c, unsigned long
> age_hz)
>  {
> -       struct dm_buffer *b, *tmp;
> -       unsigned long retain_target = get_retain_buffers(c);
> +       struct evict_params params = {.gfp = 0, .age_hz = age_hz,
> .last_accessed = 0};
> +       unsigned long retain = get_retain_buffers(c);
>         unsigned long count;
>         LIST_HEAD(write_list);
>
> @@ -1895,91 +2687,13 @@ static void __evict_old_buffers(struct
> dm_bufio_client *c, unsigned long age_hz)
>                 dm_bufio_lock(c);
>         }
>
> -       count = c->n_buffers[LIST_CLEAN] + c->n_buffers[LIST_DIRTY];
> -       list_for_each_entry_safe_reverse(b, tmp, &c->lru[LIST_CLEAN],
> lru_list) {
> -               if (count <= retain_target)
> -                       break;
> -
> -               if (!older_than(b, age_hz))
> -                       break;
> -
> -               if (__try_evict_buffer(b, 0))
> -                       count--;
> -
> -               cond_resched();
> -       }
> +       count = cache_total(&c->cache);
> +       if (count > retain)
> +               __evict_many(c, &params, LIST_CLEAN, count - retain);
>
>         dm_bufio_unlock(c);
>  }
>
> -static void do_global_cleanup(struct work_struct *w)
> -{
> -       struct dm_bufio_client *locked_client = NULL;
> -       struct dm_bufio_client *current_client;
> -       struct dm_buffer *b;
> -       unsigned int spinlock_hold_count;
> -       unsigned long threshold = dm_bufio_cache_size -
> -               dm_bufio_cache_size / DM_BUFIO_LOW_WATERMARK_RATIO;
> -       unsigned long loops = global_num * 2;
> -
> -       mutex_lock(&dm_bufio_clients_lock);
> -
> -       while (1) {
> -               cond_resched();
> -
> -               spin_lock(&global_spinlock);
> -               if (unlikely(dm_bufio_current_allocated <= threshold))
> -                       break;
> -
> -               spinlock_hold_count = 0;
> -get_next:
> -               if (!loops--)
> -                       break;
> -               if (unlikely(list_empty(&global_queue)))
> -                       break;
> -               b = list_entry(global_queue.prev, struct dm_buffer,
> global_list);
> -
> -               if (b->accessed) {
> -                       b->accessed = 0;
> -                       list_move(&b->global_list, &global_queue);
> -                       if (likely(++spinlock_hold_count < 16))
> -                               goto get_next;
> -                       spin_unlock(&global_spinlock);
> -                       continue;
> -               }
> -
> -               current_client = b->c;
> -               if (unlikely(current_client != locked_client)) {
> -                       if (locked_client)
> -                               dm_bufio_unlock(locked_client);
> -
> -                       if (!dm_bufio_trylock(current_client)) {
> -                               spin_unlock(&global_spinlock);
> -                               dm_bufio_lock(current_client);
> -                               locked_client = current_client;
> -                               continue;
> -                       }
> -
> -                       locked_client = current_client;
> -               }
> -
> -               spin_unlock(&global_spinlock);
> -
> -               if (unlikely(!__try_evict_buffer(b, GFP_KERNEL))) {
> -                       spin_lock(&global_spinlock);
> -                       list_move(&b->global_list, &global_queue);
> -                       spin_unlock(&global_spinlock);
> -               }
> -       }
> -
> -       spin_unlock(&global_spinlock);
> -
> -       if (locked_client)
> -               dm_bufio_unlock(locked_client);
> -
> -       mutex_unlock(&dm_bufio_clients_lock);
> -}
> -
>  static void cleanup_old_buffers(void)
>  {
>         unsigned long max_age_hz = get_max_age_hz();
> @@ -1990,7 +2704,7 @@ static void cleanup_old_buffers(void)
>         __cache_size_refresh();
>
>         list_for_each_entry(c, &dm_bufio_all_clients, client_list)
> -               __evict_old_buffers(c, max_age_hz);
> +               evict_old_buffers(c, max_age_hz);
>
>         mutex_unlock(&dm_bufio_clients_lock);
>  }
> @@ -2003,6 +2717,107 @@ static void work_fn(struct work_struct *w)
>                            DM_BUFIO_WORK_TIMER_SECS * HZ);
>  }
>
> +/*--------------------------------------------------------------*/
> +
> +/*
> + * Global cleanup tries to evict the oldest buffers from across _all_
> + * the clients.  It does this by repeatedly evicting a few buffers from
> + * the client that holds the oldest buffer.  It's approximate, but
> hopefully
> + * good enough.
> + */
> +static struct dm_bufio_client *__pop_client(void)
> +{
> +       struct list_head *h;
> +
> +       if (list_empty(&dm_bufio_all_clients))
> +               return NULL;
> +
> +       h = dm_bufio_all_clients.next;
> +       list_del(h);
> +       return container_of(h, struct dm_bufio_client, client_list);
> +}
> +
> +/*
> + * Inserts the client in the global client list based on its
> + * 'oldest_buffer' field.
> + */
> +static void __insert_client(struct dm_bufio_client *new_client)
> +{
> +       struct dm_bufio_client *c;
> +       struct list_head *h = dm_bufio_all_clients.next;
> +
> +       while (h != &dm_bufio_all_clients) {
> +               c = container_of(h, struct dm_bufio_client, client_list);
> +               if (time_after_eq(c->oldest_buffer,
> new_client->oldest_buffer))
> +                       break;
> +               h = h->next;
> +       }
> +
> +       list_add_tail(&new_client->client_list, h);
> +}
> +
> +static unsigned int __evict_a_few(unsigned int nr_buffers)
> +{
> +       unsigned int count;
> +       struct dm_bufio_client *c;
> +       struct evict_params params = {
> +               .gfp = GFP_KERNEL,
> +               .age_hz = 0,
> +               /* set to jiffies in case there are no buffers in this
> client */
> +               .last_accessed = jiffies
> +       };
> +
> +       c = __pop_client();
> +       if (!c)
> +               return 0;
> +
> +       dm_bufio_lock(c);
> +       count = __evict_many(c, &params, LIST_CLEAN, nr_buffers);
> +       dm_bufio_unlock(c);
> +
> +       if (count)
> +               c->oldest_buffer = params.last_accessed;
> +       __insert_client(c);
> +
> +       return count;
> +}
> +
> +static void check_watermarks(void)
> +{
> +       LIST_HEAD(write_list);
> +       struct dm_bufio_client *c;
> +
> +       mutex_lock(&dm_bufio_clients_lock);
> +       list_for_each_entry(c, &dm_bufio_all_clients, client_list) {
> +               dm_bufio_lock(c);
> +               __check_watermark(c, &write_list);
> +               dm_bufio_unlock(c);
> +       }
> +       mutex_unlock(&dm_bufio_clients_lock);
> +
> +       __flush_write_list(&write_list);
> +}
> +
> +static void evict_old(void)
> +{
> +       unsigned long threshold = dm_bufio_cache_size -
> +               dm_bufio_cache_size / DM_BUFIO_LOW_WATERMARK_RATIO;
> +
> +       mutex_lock(&dm_bufio_clients_lock);
> +       while (dm_bufio_current_allocated > threshold) {
> +               if (!__evict_a_few(64))
> +                       break;
> +               cond_resched();
> +       }
> +       mutex_unlock(&dm_bufio_clients_lock);
> +}
> +
> +static void do_global_cleanup(struct work_struct *w)
> +{
> +       check_watermarks();
> +       evict_old();
> +}
> +
>  /*
>   *--------------------------------------------------------------
>   * Module setup
> --
> 2.40.0
>
>

[-- Attachment #1.2: Type: text/html, Size: 98966 bytes --]

[-- Attachment #2: Type: text/plain, Size: 98 bytes --]

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

Re: [dm-devel] [dm-6.4 PATCH 5/8] dm bufio: improve concurrent IO performance

[Mike Snitzer]

On Thu, Mar 23 2023 at  3:51P -0400,
Joe Thornber <thornber@redhat.com> wrote:

> Mike,
> 
> I'm Nacking this patch here.  You've taken the comment from my latest
> patch, and attached it to something else.  I don't know if you've taken an
> older patch and updated it, or taken the latest patch and downgraded it.

I lean on git to tell me if code is identical.

Since we're now clearly keeping score here. I used git to compare what
I had here (with various small whitespace tweaks and other nits):
https://git.kernel.org/pub/scm/linux/kernel/git/snitzer/linux.git/log/?h=2023-03-01-thin-concurrency-8

with your -8 code:
https://github.com/jthornber/linux/tree/2023-03-21-thin-concurrency-8

I had to deal with the problem that you didn't use my tree you rebased
so I had to reconcile not losing my handful of fluff.

I leaned on the fact that your -7 and -8 were identical in code:
git diff ejt/2023-02-28-thin-concurrency-7 ejt/2023-03-21-thin-concurrency-8 -- drivers/md/dm-bufio.c

I could've just generated my fluff and folded it into your new clean
-8.  But I went the other way around, instead I picked up your other
incremental changes from -7 and folded them into what I had. No real
reason, that's just how I did it.

> For instance the latest patch uses the name 'struct dm_buffer_cache', but
> it looks like you've renamed to 'struct buffer_cache' and then posted
> another patch in the series that renames back to dm_buffer_cache.

Right, I was cherry-picking your incremantal changes your staged in -7
before your clean patch rollup in your -8. How the struct rename
occurred to the code didn't much matter to me, so long as the change
was made.  But yeah I should've just folded it.  Not a big deal.

> I also see the comment change:
> 
> original "A refcount_t for hold_count does not work." which has changed to
> "NOTE: attempting to use refcount_t instead of atomic_t caused crashes!".
> Talk about over dramatisation (If the hold_count is incorrect we detect and
> BUG).

Yes, but I was aiming to leave breadcrumbs if/when someone might think
to retry using refcount_t to see why it "does not work". I was trying
to save future me or you or someone else some time. By making it
clear: there will be dragons.

But in doing so I unwittingly already unleashed different dragons ;)

> I can't stand by my code if I don't know what's being sent up in my name.
>
> Please use the patchset from this branch:
> https://github.com/jthornber/linux/tree/2023-03-21-thin-concurrency-8

We've resolved this now interactively in chat, but I had to reply on
dm-devel to let others know.  Again, on a code level what I posted to
dm-devel reflects what was staged in 2023-03-21-thin-concurrency-8:

For me this yields my fluff I was referring to above:
git diff ejt/2023-03-21-thin-concurrency-8 dm/dm-6.4 -- drivers/md/dm-bufio.c

Biggest difference being dm_buffer struct member reordering (which I
had discussed with you).

For the benefit of others, I since dropped patch 1 and patch 2
(max_discard_granularity and bio-prison-v1 patch) but pulled them to
the HEAD of this new development branch (that has extra debugging):
https://git.kernel.org/pub/scm/linux/kernel/git/snitzer/linux.git/log/?h=dm-6.4

I'll go back to bed now, hopefully you'll have this discard issue
fixed in a couple hours when I return (as we discussed). ;)

Mike

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 6/8] dm bufio: move dm_bufio_client members to avoid spanning cachelines

[Mike Snitzer]

Movement also consolidates holes in dm_bufio_client struct. But the
overall size of the struct isn't changed.

Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-bufio.c | 24 +++++++++++++-----------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/drivers/md/dm-bufio.c b/drivers/md/dm-bufio.c
index 56ef160331ce..b418ee8e3853 100644
--- a/drivers/md/dm-bufio.c
+++ b/drivers/md/dm-bufio.c
@@ -952,13 +952,16 @@ static void cache_remove_range(struct buffer_cache *bc,
  *	context.
  */
 struct dm_bufio_client {
-	struct mutex lock;
-	spinlock_t spinlock;
-	bool no_sleep;
-
 	struct block_device *bdev;
 	unsigned int block_size;
 	s8 sectors_per_block_bits;
+
+	bool no_sleep;
+	struct mutex lock;
+	spinlock_t spinlock;
+
+	int async_write_error;
+
 	void (*alloc_callback)(struct dm_buffer *buf);
 	void (*write_callback)(struct dm_buffer *buf);
 	struct kmem_cache *slab_buffer;
@@ -970,23 +973,22 @@ struct dm_bufio_client {
 
 	unsigned int minimum_buffers;
 
-	struct buffer_cache cache;
-	wait_queue_head_t free_buffer_wait;
-
 	sector_t start;
 
-	int async_write_error;
-
-	struct list_head client_list;
-
 	struct shrinker shrinker;
 	struct work_struct shrink_work;
 	atomic_long_t need_shrink;
 
+	wait_queue_head_t free_buffer_wait;
+
+	struct list_head client_list;
+
 	/*
 	 * Used by global_cleanup to sort the clients list.
 	 */
 	unsigned long oldest_buffer;
+
+	struct buffer_cache cache;
 };
 
 static DEFINE_STATIC_KEY_FALSE(no_sleep_enabled);
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 7/8] dm bufio: rename buffer_cache struct to dm_buffer_cache

[Mike Snitzer]

Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-bufio.c | 52 +++++++++++++++++++++----------------------
 1 file changed, 26 insertions(+), 26 deletions(-)

diff --git a/drivers/md/dm-bufio.c b/drivers/md/dm-bufio.c
index b418ee8e3853..e7b901baa878 100644
--- a/drivers/md/dm-bufio.c
+++ b/drivers/md/dm-bufio.c
@@ -327,7 +327,7 @@ enum data_mode {
 };
 
 struct dm_buffer {
-	/* protected by the locks in buffer_cache */
+	/* protected by the locks in dm_buffer_cache */
 	struct rb_node node;
 
 	/* immutable, so don't need protecting */
@@ -393,7 +393,7 @@ struct tree_lock {
 	struct rw_semaphore lock;
 } ____cacheline_aligned_in_smp;
 
-struct buffer_cache {
+struct dm_buffer_cache {
 	/*
 	 * We spread entries across multiple trees to reduce contention
 	 * on the locks.
@@ -414,22 +414,22 @@ static inline unsigned int cache_index(sector_t block)
 	return block & LOCKS_MASK;
 }
 
-static inline void cache_read_lock(struct buffer_cache *bc, sector_t block)
+static inline void cache_read_lock(struct dm_buffer_cache *bc, sector_t block)
 {
 	down_read(&bc->locks[cache_index(block)].lock);
 }
 
-static inline void cache_read_unlock(struct buffer_cache *bc, sector_t block)
+static inline void cache_read_unlock(struct dm_buffer_cache *bc, sector_t block)
 {
 	up_read(&bc->locks[cache_index(block)].lock);
 }
 
-static inline void cache_write_lock(struct buffer_cache *bc, sector_t block)
+static inline void cache_write_lock(struct dm_buffer_cache *bc, sector_t block)
 {
 	down_write(&bc->locks[cache_index(block)].lock);
 }
 
-static inline void cache_write_unlock(struct buffer_cache *bc, sector_t block)
+static inline void cache_write_unlock(struct dm_buffer_cache *bc, sector_t block)
 {
 	up_write(&bc->locks[cache_index(block)].lock);
 }
@@ -439,12 +439,12 @@ static inline void cache_write_unlock(struct buffer_cache *bc, sector_t block)
  * This struct helps avoid redundant drop and gets of the same lock.
  */
 struct lock_history {
-	struct buffer_cache *cache;
+	struct dm_buffer_cache *cache;
 	bool write;
 	unsigned int previous;
 };
 
-static void lh_init(struct lock_history *lh, struct buffer_cache *cache, bool write)
+static void lh_init(struct lock_history *lh, struct dm_buffer_cache *cache, bool write)
 {
 	lh->cache = cache;
 	lh->write = write;
@@ -513,7 +513,7 @@ static struct dm_buffer *list_to_buffer(struct list_head *l)
 	return le_to_buffer(le);
 }
 
-static void cache_init(struct buffer_cache *bc)
+static void cache_init(struct dm_buffer_cache *bc)
 {
 	unsigned int i;
 
@@ -526,7 +526,7 @@ static void cache_init(struct buffer_cache *bc)
 	lru_init(&bc->lru[LIST_DIRTY]);
 }
 
-static void cache_destroy(struct buffer_cache *bc)
+static void cache_destroy(struct dm_buffer_cache *bc)
 {
 	unsigned int i;
 
@@ -540,7 +540,7 @@ static void cache_destroy(struct buffer_cache *bc)
 /*
  * not threadsafe, or racey depending how you look at it
  */
-static unsigned long cache_count(struct buffer_cache *bc, int list_mode)
+static unsigned long cache_count(struct dm_buffer_cache *bc, int list_mode)
 {
 	return lru_count(&bc->lru[list_mode]);
 }
@@ -550,7 +550,7 @@ static unsigned long cache_count(struct buffer_cache *bc, int list_mode)
 /*
  * not threadsafe
  */
-static unsigned long cache_total(struct buffer_cache *bc)
+static unsigned long cache_total(struct dm_buffer_cache *bc)
 {
 	return lru_count(&bc->lru[LIST_CLEAN]) +
 		lru_count(&bc->lru[LIST_DIRTY]);
@@ -588,7 +588,7 @@ static void __cache_inc_buffer(struct dm_buffer *b)
 	WRITE_ONCE(b->last_accessed, jiffies);
 }
 
-static struct dm_buffer *cache_get(struct buffer_cache *bc, sector_t block)
+static struct dm_buffer *cache_get(struct dm_buffer_cache *bc, sector_t block)
 {
 	struct dm_buffer *b;
 
@@ -609,7 +609,7 @@ static struct dm_buffer *cache_get(struct buffer_cache *bc, sector_t block)
  * Returns true if the hold count hits zero.
  * threadsafe
  */
-static bool cache_put(struct buffer_cache *bc, struct dm_buffer *b)
+static bool cache_put(struct dm_buffer_cache *bc, struct dm_buffer *b)
 {
 	bool r;
 
@@ -653,7 +653,7 @@ static enum evict_result __evict_pred(struct lru_entry *le, void *context)
 	return w->pred(b, w->context);
 }
 
-static struct dm_buffer *__cache_evict(struct buffer_cache *bc, int list_mode,
+static struct dm_buffer *__cache_evict(struct dm_buffer_cache *bc, int list_mode,
 				       b_predicate pred, void *context,
 				       struct lock_history *lh)
 {
@@ -672,7 +672,7 @@ static struct dm_buffer *__cache_evict(struct buffer_cache *bc, int list_mode,
 	return b;
 }
 
-static struct dm_buffer *cache_evict(struct buffer_cache *bc, int list_mode,
+static struct dm_buffer *cache_evict(struct dm_buffer_cache *bc, int list_mode,
 				     b_predicate pred, void *context)
 {
 	struct dm_buffer *b;
@@ -690,7 +690,7 @@ static struct dm_buffer *cache_evict(struct buffer_cache *bc, int list_mode,
 /*
  * Mark a buffer as clean or dirty. Not threadsafe.
  */
-static void cache_mark(struct buffer_cache *bc, struct dm_buffer *b, int list_mode)
+static void cache_mark(struct dm_buffer_cache *bc, struct dm_buffer *b, int list_mode)
 {
 	cache_write_lock(bc, b->block);
 	if (list_mode != b->list_mode) {
@@ -707,7 +707,7 @@ static void cache_mark(struct buffer_cache *bc, struct dm_buffer *b, int list_mo
  * Runs through the lru associated with 'old_mode', if the predicate matches then
  * it moves them to 'new_mode'.  Not threadsafe.
  */
-static void __cache_mark_many(struct buffer_cache *bc, int old_mode, int new_mode,
+static void __cache_mark_many(struct dm_buffer_cache *bc, int old_mode, int new_mode,
 			      b_predicate pred, void *context, struct lock_history *lh)
 {
 	struct lru_entry *le;
@@ -725,7 +725,7 @@ static void __cache_mark_many(struct buffer_cache *bc, int old_mode, int new_mod
 	}
 }
 
-static void cache_mark_many(struct buffer_cache *bc, int old_mode, int new_mode,
+static void cache_mark_many(struct dm_buffer_cache *bc, int old_mode, int new_mode,
 			    b_predicate pred, void *context)
 {
 	struct lock_history lh;
@@ -753,7 +753,7 @@ enum it_action {
 
 typedef enum it_action (*iter_fn)(struct dm_buffer *b, void *context);
 
-static void __cache_iterate(struct buffer_cache *bc, int list_mode,
+static void __cache_iterate(struct dm_buffer_cache *bc, int list_mode,
 			    iter_fn fn, void *context, struct lock_history *lh)
 {
 	struct lru *lru = &bc->lru[list_mode];
@@ -781,7 +781,7 @@ static void __cache_iterate(struct buffer_cache *bc, int list_mode,
 	} while (le != first);
 }
 
-static void cache_iterate(struct buffer_cache *bc, int list_mode,
+static void cache_iterate(struct dm_buffer_cache *bc, int list_mode,
 			  iter_fn fn, void *context)
 {
 	struct lock_history lh;
@@ -824,7 +824,7 @@ static bool __cache_insert(struct rb_root *root, struct dm_buffer *b)
 	return true;
 }
 
-static bool cache_insert(struct buffer_cache *bc, struct dm_buffer *b)
+static bool cache_insert(struct dm_buffer_cache *bc, struct dm_buffer *b)
 {
 	bool r;
 
@@ -848,7 +848,7 @@ static bool cache_insert(struct buffer_cache *bc, struct dm_buffer *b)
  *
  * Not threadsafe.
  */
-static bool cache_remove(struct buffer_cache *bc, struct dm_buffer *b)
+static bool cache_remove(struct dm_buffer_cache *bc, struct dm_buffer *b)
 {
 	bool r;
 
@@ -894,7 +894,7 @@ static struct dm_buffer *__find_next(struct rb_root *root, sector_t block)
 	return best;
 }
 
-static void __remove_range(struct buffer_cache *bc,
+static void __remove_range(struct dm_buffer_cache *bc,
 			   struct rb_root *root,
 			   sector_t begin, sector_t end,
 			   b_predicate pred, b_release release)
@@ -921,7 +921,7 @@ static void __remove_range(struct buffer_cache *bc,
 	}
 }
 
-static void cache_remove_range(struct buffer_cache *bc,
+static void cache_remove_range(struct dm_buffer_cache *bc,
 			       sector_t begin, sector_t end,
 			       b_predicate pred, b_release release)
 {
@@ -988,7 +988,7 @@ struct dm_bufio_client {
 	 */
 	unsigned long oldest_buffer;
 
-	struct buffer_cache cache;
+	struct dm_buffer_cache cache;
 };
 
 static DEFINE_STATIC_KEY_FALSE(no_sleep_enabled);
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel

[dm-devel] [dm-6.4 PATCH 8/8] dm thin: speed up cell_defer_no_holder()

[Mike Snitzer]

From: Joe Thornber <ejt@redhat.com>

Reduce the time that a spinlock is held in cell_defer_no_holder().

Signed-off-by: Joe Thornber <ejt@redhat.com>
Signed-off-by: Mike Snitzer <snitzer@kernel.org>
---
 drivers/md/dm-thin.c | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/drivers/md/dm-thin.c b/drivers/md/dm-thin.c
index 286e4d9e45c8..583159dc8084 100644
--- a/drivers/md/dm-thin.c
+++ b/drivers/md/dm-thin.c
@@ -883,15 +883,17 @@ static void cell_defer_no_holder(struct thin_c *tc, struct dm_bio_prison_cell *c
 {
 	struct pool *pool = tc->pool;
 	unsigned long flags;
-	int has_work;
+	struct bio_list bios;
 
-	spin_lock_irqsave(&tc->lock, flags);
-	cell_release_no_holder(pool, cell, &tc->deferred_bio_list);
-	has_work = !bio_list_empty(&tc->deferred_bio_list);
-	spin_unlock_irqrestore(&tc->lock, flags);
+	bio_list_init(&bios);
+	cell_release_no_holder(pool, cell, &bios);
 
-	if (has_work)
+	if (!bio_list_empty(&bios)) {
+		spin_lock_irqsave(&tc->lock, flags);
+		bio_list_merge(&tc->deferred_bio_list, &bios);
+		spin_unlock_irqrestore(&tc->lock, flags);
 		wake_worker(pool);
+	}
 }
 
 static void thin_defer_bio(struct thin_c *tc, struct bio *bio);
-- 
2.40.0

--
dm-devel mailing list
dm-devel@redhat.com
https://listman.redhat.com/mailman/listinfo/dm-devel