dm bio prison: implement per bucket locking in the dm_bio_prison hash…

… table

Split the single per bio-prison lock by using per bucket locking.  Per
bucket locking benefits both dm-thin and dm-cache targets by reducing
bio-prison lock contention.

Signed-off-by: Heinz Mauelshagen <heinzm@redhat.com>
Signed-off-by: Joe Thornber <ejt@redhat.com>
Signed-off-by: Mike Snitzer <snitzer@redhat.com>

drivers/md/dm-bio-prison.c

@@ -14,13 +14,17 @@
 
 /*----------------------------------------------------------------*/
 
-struct dm_bio_prison {
+struct bucket {
 	spinlock_t lock;
+	struct hlist_head cells;
+};
+
+struct dm_bio_prison {
 	mempool_t *cell_pool;
 
 	unsigned nr_buckets;
 	unsigned hash_mask;
-	struct hlist_head *cells;
+	struct bucket *buckets;
 };
 
 /*----------------------------------------------------------------*/
@@ -40,6 +44,12 @@ static uint32_t calc_nr_buckets(unsigned nr_cells)
 
 static struct kmem_cache *_cell_cache;
 
+static void init_bucket(struct bucket *b)
+{
+	spin_lock_init(&b->lock);
+	INIT_HLIST_HEAD(&b->cells);
+}
+
 /*
  * @nr_cells should be the number of cells you want in use _concurrently_.
  * Don't confuse it with the number of distinct keys.
@@ -49,13 +59,12 @@ struct dm_bio_prison *dm_bio_prison_create(unsigned nr_cells)
 	unsigned i;
 	uint32_t nr_buckets = calc_nr_buckets(nr_cells);
 	size_t len = sizeof(struct dm_bio_prison) +
-		(sizeof(struct hlist_head) * nr_buckets);
+		(sizeof(struct bucket) * nr_buckets);
 	struct dm_bio_prison *prison = kmalloc(len, GFP_KERNEL);
 
 	if (!prison)
 		return NULL;
 
-	spin_lock_init(&prison->lock);
 	prison->cell_pool = mempool_create_slab_pool(nr_cells, _cell_cache);
 	if (!prison->cell_pool) {
 		kfree(prison);
@@ -64,9 +73,9 @@ struct dm_bio_prison *dm_bio_prison_create(unsigned nr_cells)
 
 	prison->nr_buckets = nr_buckets;
 	prison->hash_mask = nr_buckets - 1;
-	prison->cells = (struct hlist_head *) (prison + 1);
+	prison->buckets = (struct bucket *) (prison + 1);
 	for (i = 0; i < nr_buckets; i++)
-		INIT_HLIST_HEAD(prison->cells + i);
+		init_bucket(prison->buckets + i);
 
 	return prison;
 }
@@ -107,48 +116,52 @@ static int keys_equal(struct dm_cell_key *lhs, struct dm_cell_key *rhs)
 		       (lhs->block == rhs->block);
 }
 
-static struct dm_bio_prison_cell *__search_bucket(struct hlist_head *bucket,
+static struct bucket *get_bucket(struct dm_bio_prison *prison,
+				 struct dm_cell_key *key)
+{
+	return prison->buckets + hash_key(prison, key);
+}
+
+static struct dm_bio_prison_cell *__search_bucket(struct bucket *b,
 						  struct dm_cell_key *key)
 {
 	struct dm_bio_prison_cell *cell;
 
-	hlist_for_each_entry(cell, bucket, list)
+	hlist_for_each_entry(cell, &b->cells, list)
 		if (keys_equal(&cell->key, key))
 			return cell;
 
 	return NULL;
 }
 
-static void __setup_new_cell(struct dm_bio_prison *prison,
+static void __setup_new_cell(struct bucket *b,
 			     struct dm_cell_key *key,
 			     struct bio *holder,
-			     uint32_t hash,
 			     struct dm_bio_prison_cell *cell)
 {
 	memcpy(&cell->key, key, sizeof(cell->key));
 	cell->holder = holder;
 	bio_list_init(&cell->bios);
-	hlist_add_head(&cell->list, prison->cells + hash);
+	hlist_add_head(&cell->list, &b->cells);
 }
 
-static int __bio_detain(struct dm_bio_prison *prison,
+static int __bio_detain(struct bucket *b,
 			struct dm_cell_key *key,
 			struct bio *inmate,
 			struct dm_bio_prison_cell *cell_prealloc,
 			struct dm_bio_prison_cell **cell_result)
 {
-	uint32_t hash = hash_key(prison, key);
 	struct dm_bio_prison_cell *cell;
 
-	cell = __search_bucket(prison->cells + hash, key);
+	cell = __search_bucket(b, key);
 	if (cell) {
 		if (inmate)
 			bio_list_add(&cell->bios, inmate);
 		*cell_result = cell;
 		return 1;
 	}
 
-	__setup_new_cell(prison, key, inmate, hash, cell_prealloc);
+	__setup_new_cell(b, key, inmate, cell_prealloc);
 	*cell_result = cell_prealloc;
 	return 0;
 }
@@ -161,10 +174,11 @@ static int bio_detain(struct dm_bio_prison *prison,
 {
 	int r;
 	unsigned long flags;
+	struct bucket *b = get_bucket(prison, key);
 
-	spin_lock_irqsave(&prison->lock, flags);
-	r = __bio_detain(prison, key, inmate, cell_prealloc, cell_result);
-	spin_unlock_irqrestore(&prison->lock, flags);
+	spin_lock_irqsave(&b->lock, flags);
+	r = __bio_detain(b, key, inmate, cell_prealloc, cell_result);
+	spin_unlock_irqrestore(&b->lock, flags);
 
 	return r;
 }
@@ -208,10 +222,11 @@ void dm_cell_release(struct dm_bio_prison *prison,
 		     struct bio_list *bios)
 {
 	unsigned long flags;
+	struct bucket *b = get_bucket(prison, &cell->key);
 
-	spin_lock_irqsave(&prison->lock, flags);
+	spin_lock_irqsave(&b->lock, flags);
 	__cell_release(cell, bios);
-	spin_unlock_irqrestore(&prison->lock, flags);
+	spin_unlock_irqrestore(&b->lock, flags);
 }
 EXPORT_SYMBOL_GPL(dm_cell_release);
 
@@ -230,10 +245,11 @@ void dm_cell_release_no_holder(struct dm_bio_prison *prison,
 			       struct bio_list *inmates)
 {
 	unsigned long flags;
+	struct bucket *b = get_bucket(prison, &cell->key);
 
-	spin_lock_irqsave(&prison->lock, flags);
+	spin_lock_irqsave(&b->lock, flags);
 	__cell_release_no_holder(cell, inmates);
-	spin_unlock_irqrestore(&prison->lock, flags);
+	spin_unlock_irqrestore(&b->lock, flags);
 }
 EXPORT_SYMBOL_GPL(dm_cell_release_no_holder);
 
@@ -242,13 +258,9 @@ void dm_cell_error(struct dm_bio_prison *prison,
 {
 	struct bio_list bios;
 	struct bio *bio;
-	unsigned long flags;
 
 	bio_list_init(&bios);
-
-	spin_lock_irqsave(&prison->lock, flags);
-	__cell_release(cell, &bios);
-	spin_unlock_irqrestore(&prison->lock, flags);
+	dm_cell_release(prison, cell, &bios);
 
 	while ((bio = bio_list_pop(&bios)))
 		bio_endio(bio, error);