dm bio prison: pass cell memory in

Change the dm_bio_prison interface so that instead of allocating memory internally, dm_bio_detain is supplied with a pre-allocated cell each time it is called. This enables a subsequent patch to move the allocation of the struct dm_bio_prison_cell outside the thin target's mapping function so it can no longer block there. Signed-off-by: Joe Thornber <ejt@redhat.com> Signed-off-by: Alasdair G Kergon <agk@redhat.com>

dm bio prison: pass cell memory in
Change the dm_bio_prison interface so that instead of allocating memory internally, dm_bio_detain is supplied with a pre-allocated cell each time it is called. This enables a subsequent patch to move the allocation of the struct dm_bio_prison_cell outside the thin target's mapping function so it can no longer block there. Signed-off-by: Joe Thornber <ejt@redhat.com> Signed-off-by: Alasdair G Kergon <agk@redhat.com>
Joe Thornber · Alasdair G Kergon
1 parent 4e7f1f9089
Showing 3 changed files with 176 additions and 101 deletions Side-by-side Diff
drivers/md/dm-bio-prison.c
drivers/md/dm-bio-prison.h
drivers/md/dm-thin.c
@@ -16,7 +16,6 @@
  
 struct dm_bio_prison_cell {
 	struct hlist_node list;
-	struct dm_bio_prison *prison;
 	struct dm_cell_key key;
 	struct bio *holder;
 	struct bio_list bios;
@@ -87,6 +86,19 @@
 }
 EXPORT_SYMBOL_GPL(dm_bio_prison_destroy);
  
+struct dm_bio_prison_cell *dm_bio_prison_alloc_cell(struct dm_bio_prison *prison, gfp_t gfp)
+{
+	return mempool_alloc(prison->cell_pool, gfp);
+}
+EXPORT_SYMBOL_GPL(dm_bio_prison_alloc_cell);
+
+void dm_bio_prison_free_cell(struct dm_bio_prison *prison,
+			     struct dm_bio_prison_cell *cell)
+{
+	mempool_free(cell, prison->cell_pool);
+}
+EXPORT_SYMBOL_GPL(dm_bio_prison_free_cell);
+
 static uint32_t hash_key(struct dm_bio_prison *prison, struct dm_cell_key *key)
 {
 	const unsigned long BIG_PRIME = 4294967291UL;
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
@@ -114,91 +126,86 @@
 	return NULL;
 }
  
-/*
- * This may block if a new cell needs allocating.  You must ensure that
- * cells will be unlocked even if the calling thread is blocked.
- *
- * Returns 1 if the cell was already held, 0 if @inmate is the new holder.
- */
-int dm_bio_detain(struct dm_bio_prison *prison, struct dm_cell_key *key,
-		  struct bio *inmate, struct dm_bio_prison_cell **ref)
+static void __setup_new_cell(struct dm_bio_prison *prison,
+			     struct dm_cell_key *key,
+			     struct bio *holder,
+			     uint32_t hash,
+			     struct dm_bio_prison_cell *cell)
 {
-	int r = 1;
-	unsigned long flags;
+	memcpy(&cell->key, key, sizeof(cell->key));
+	cell->holder = holder;
+	bio_list_init(&cell->bios);
+	hlist_add_head(&cell->list, prison->cells + hash);
+}
+
+static int __bio_detain(struct dm_bio_prison *prison,
+			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, *cell2;
+	struct dm_bio_prison_cell *cell;
  
-	BUG_ON(hash > prison->nr_buckets);
-
-	spin_lock_irqsave(&prison->lock, flags);
-
 	cell = __search_bucket(prison->cells + hash, key);
 	if (cell) {
-		bio_list_add(&cell->bios, inmate);
-		goto out;
+		if (inmate)
+			bio_list_add(&cell->bios, inmate);
+		*cell_result = cell;
+		return 1;
 	}
  
-	/*
-	 * Allocate a new cell
-	 */
-	spin_unlock_irqrestore(&prison->lock, flags);
-	cell2 = mempool_alloc(prison->cell_pool, GFP_NOIO);
-	spin_lock_irqsave(&prison->lock, flags);
+	__setup_new_cell(prison, key, inmate, hash, cell_prealloc);
+	*cell_result = cell_prealloc;
+	return 0;
+}
  
-	/*
-	 * We've been unlocked, so we have to double check that
-	 * nobody else has inserted this cell in the meantime.
-	 */
-	cell = __search_bucket(prison->cells + hash, key);
-	if (cell) {
-		mempool_free(cell2, prison->cell_pool);
-		bio_list_add(&cell->bios, inmate);
-		goto out;
-	}
+static int bio_detain(struct dm_bio_prison *prison,
+		      struct dm_cell_key *key,
+		      struct bio *inmate,
+		      struct dm_bio_prison_cell *cell_prealloc,
+		      struct dm_bio_prison_cell **cell_result)
+{
+	int r;
+	unsigned long flags;
  
-	/*
-	 * Use new cell.
-	 */
-	cell = cell2;
-
-	cell->prison = prison;
-	memcpy(&cell->key, key, sizeof(cell->key));
-	cell->holder = inmate;
-	bio_list_init(&cell->bios);
-	hlist_add_head(&cell->list, prison->cells + hash);
-
-	r = 0;
-
-out:
+	spin_lock_irqsave(&prison->lock, flags);
+	r = __bio_detain(prison, key, inmate, cell_prealloc, cell_result);
 	spin_unlock_irqrestore(&prison->lock, flags);
  
-	*ref = cell;
-
 	return r;
 }
+
+int dm_bio_detain(struct dm_bio_prison *prison,
+		  struct dm_cell_key *key,
+		  struct bio *inmate,
+		  struct dm_bio_prison_cell *cell_prealloc,
+		  struct dm_bio_prison_cell **cell_result)
+{
+	return bio_detain(prison, key, inmate, cell_prealloc, cell_result);
+}
 EXPORT_SYMBOL_GPL(dm_bio_detain);
  
 /*
  * @inmates must have been initialised prior to this call
  */
-static void __cell_release(struct dm_bio_prison_cell *cell, struct bio_list *inmates)
+static void __cell_release(struct dm_bio_prison_cell *cell,
+			   struct bio_list *inmates)
 {
-	struct dm_bio_prison *prison = cell->prison;
-
 	hlist_del(&cell->list);
  
 	if (inmates) {
-		bio_list_add(inmates, cell->holder);
+		if (cell->holder)
+			bio_list_add(inmates, cell->holder);
 		bio_list_merge(inmates, &cell->bios);
 	}
-
-	mempool_free(cell, prison->cell_pool);
 }
  
-void dm_cell_release(struct dm_bio_prison_cell *cell, struct bio_list *bios)
+void dm_cell_release(struct dm_bio_prison *prison,
+		     struct dm_bio_prison_cell *cell,
+		     struct bio_list *bios)
 {
 	unsigned long flags;
-	struct dm_bio_prison *prison = cell->prison;
  
 	spin_lock_irqsave(&prison->lock, flags);
 	__cell_release(cell, bios);
  
  
  
  
@@ -209,20 +216,18 @@
 /*
  * Sometimes we don't want the holder, just the additional bios.
  */
-static void __cell_release_no_holder(struct dm_bio_prison_cell *cell, struct bio_list *inmates)
+static void __cell_release_no_holder(struct dm_bio_prison_cell *cell,
+				     struct bio_list *inmates)
 {
-	struct dm_bio_prison *prison = cell->prison;
-
 	hlist_del(&cell->list);
 	bio_list_merge(inmates, &cell->bios);
-
-	mempool_free(cell, prison->cell_pool);
 }
  
-void dm_cell_release_no_holder(struct dm_bio_prison_cell *cell, struct bio_list *inmates)
+void dm_cell_release_no_holder(struct dm_bio_prison *prison,
+			       struct dm_bio_prison_cell *cell,
+			       struct bio_list *inmates)
 {
 	unsigned long flags;
-	struct dm_bio_prison *prison = cell->prison;
  
 	spin_lock_irqsave(&prison->lock, flags);
 	__cell_release_no_holder(cell, inmates);
  
@@ -230,9 +235,9 @@
 }
 EXPORT_SYMBOL_GPL(dm_cell_release_no_holder);
  
-void dm_cell_error(struct dm_bio_prison_cell *cell)
+void dm_cell_error(struct dm_bio_prison *prison,
+		   struct dm_bio_prison_cell *cell)
 {
-	struct dm_bio_prison *prison = cell->prison;
 	struct bio_list bios;
 	struct bio *bio;
 	unsigned long flags;
@@ -35,17 +35,36 @@
 void dm_bio_prison_destroy(struct dm_bio_prison *prison);
  
 /*
- * This may block if a new cell needs allocating.  You must ensure that
- * cells will be unlocked even if the calling thread is blocked.
+ * These two functions just wrap a mempool.  This is a transitory step:
+ * Eventually all bio prison clients should manage their own cell memory.
  *
+ * Like mempool_alloc(), dm_bio_prison_alloc_cell() can only fail if called
+ * in interrupt context or passed GFP_NOWAIT.
+ */
+struct dm_bio_prison_cell *dm_bio_prison_alloc_cell(struct dm_bio_prison *prison,
+						    gfp_t gfp);
+void dm_bio_prison_free_cell(struct dm_bio_prison *prison,
+			     struct dm_bio_prison_cell *cell);
+
+/*
+ * An atomic op that combines retrieving a cell, and adding a bio to it.
+ *
  * Returns 1 if the cell was already held, 0 if @inmate is the new holder.
  */
-int dm_bio_detain(struct dm_bio_prison *prison, struct dm_cell_key *key,
-		  struct bio *inmate, struct dm_bio_prison_cell **ref);
+int dm_bio_detain(struct dm_bio_prison *prison,
+		  struct dm_cell_key *key,
+		  struct bio *inmate,
+		  struct dm_bio_prison_cell *cell_prealloc,
+		  struct dm_bio_prison_cell **cell_result);
  
-void dm_cell_release(struct dm_bio_prison_cell *cell, struct bio_list *bios);
-void dm_cell_release_no_holder(struct dm_bio_prison_cell *cell, struct bio_list *inmates);
-void dm_cell_error(struct dm_bio_prison_cell *cell);
+void dm_cell_release(struct dm_bio_prison *prison,
+		     struct dm_bio_prison_cell *cell,
+		     struct bio_list *bios);
+void dm_cell_release_no_holder(struct dm_bio_prison *prison,
+			       struct dm_bio_prison_cell *cell,
+			       struct bio_list *inmates);
+void dm_cell_error(struct dm_bio_prison *prison,
+		   struct dm_bio_prison_cell *cell);
  
 /*----------------------------------------------------------------*/
  
@@ -229,6 +229,54 @@
  
 /*----------------------------------------------------------------*/
  
+static int bio_detain(struct pool *pool, struct dm_cell_key *key, struct bio *bio,
+		      struct dm_bio_prison_cell **cell_result)
+{
+	int r;
+	struct dm_bio_prison_cell *cell_prealloc;
+
+	/*
+	 * Allocate a cell from the prison's mempool.
+	 * This might block but it can't fail.
+	 */
+	cell_prealloc = dm_bio_prison_alloc_cell(pool->prison, GFP_NOIO);
+
+	r = dm_bio_detain(pool->prison, key, bio, cell_prealloc, cell_result);
+	if (r)
+		/*
+		 * We reused an old cell; we can get rid of
+		 * the new one.
+		 */
+		dm_bio_prison_free_cell(pool->prison, cell_prealloc);
+
+	return r;
+}
+
+static void cell_release(struct pool *pool,
+			 struct dm_bio_prison_cell *cell,
+			 struct bio_list *bios)
+{
+	dm_cell_release(pool->prison, cell, bios);
+	dm_bio_prison_free_cell(pool->prison, cell);
+}
+
+static void cell_release_no_holder(struct pool *pool,
+				   struct dm_bio_prison_cell *cell,
+				   struct bio_list *bios)
+{
+	dm_cell_release_no_holder(pool->prison, cell, bios);
+	dm_bio_prison_free_cell(pool->prison, cell);
+}
+
+static void cell_error(struct pool *pool,
+		       struct dm_bio_prison_cell *cell)
+{
+	dm_cell_error(pool->prison, cell);
+	dm_bio_prison_free_cell(pool->prison, cell);
+}
+
+/*----------------------------------------------------------------*/
+
 /*
  * A global list of pools that uses a struct mapped_device as a key.
  */
  
@@ -524,14 +572,14 @@
 	unsigned long flags;
  
 	spin_lock_irqsave(&pool->lock, flags);
-	dm_cell_release(cell, &pool->deferred_bios);
+	cell_release(pool, cell, &pool->deferred_bios);
 	spin_unlock_irqrestore(&tc->pool->lock, flags);
  
 	wake_worker(pool);
 }
  
 /*
- * Same as cell_defer except it omits the original holder of the cell.
+ * Same as cell_defer above, except it omits the original holder of the cell.
  */
 static void cell_defer_no_holder(struct thin_c *tc, struct dm_bio_prison_cell *cell)
 {
@@ -539,7 +587,7 @@
 	unsigned long flags;
  
 	spin_lock_irqsave(&pool->lock, flags);
-	dm_cell_release_no_holder(cell, &pool->deferred_bios);
+	cell_release_no_holder(pool, cell, &pool->deferred_bios);
 	spin_unlock_irqrestore(&pool->lock, flags);
  
 	wake_worker(pool);
  
@@ -549,13 +597,14 @@
 {
 	if (m->bio)
 		m->bio->bi_end_io = m->saved_bi_end_io;
-	dm_cell_error(m->cell);
+	cell_error(m->tc->pool, m->cell);
 	list_del(&m->list);
 	mempool_free(m, m->tc->pool->mapping_pool);
 }
 static void process_prepared_mapping(struct dm_thin_new_mapping *m)
 {
 	struct thin_c *tc = m->tc;
+	struct pool *pool = tc->pool;
 	struct bio *bio;
 	int r;
  
@@ -564,7 +613,7 @@
 		bio->bi_end_io = m->saved_bi_end_io;
  
 	if (m->err) {
-		dm_cell_error(m->cell);
+		cell_error(pool, m->cell);
 		goto out;
 	}
  
@@ -576,7 +625,7 @@
 	r = dm_thin_insert_block(tc->td, m->virt_block, m->data_block);
 	if (r) {
 		DMERR_LIMIT("dm_thin_insert_block() failed");
-		dm_cell_error(m->cell);
+		cell_error(pool, m->cell);
 		goto out;
 	}
  
@@ -594,7 +643,7 @@
  
 out:
 	list_del(&m->list);
-	mempool_free(m, tc->pool->mapping_pool);
+	mempool_free(m, pool->mapping_pool);
 }
  
 static void process_prepared_discard_fail(struct dm_thin_new_mapping *m)
@@ -745,7 +794,7 @@
 		if (r < 0) {
 			mempool_free(m, pool->mapping_pool);
 			DMERR_LIMIT("dm_kcopyd_copy() failed");
-			dm_cell_error(cell);
+			cell_error(pool, cell);
 		}
 	}
 }
@@ -811,7 +860,7 @@
 		if (r < 0) {
 			mempool_free(m, pool->mapping_pool);
 			DMERR_LIMIT("dm_kcopyd_zero() failed");
-			dm_cell_error(cell);
+			cell_error(pool, cell);
 		}
 	}
 }
  
@@ -917,13 +966,13 @@
 	spin_unlock_irqrestore(&pool->lock, flags);
 }
  
-static void no_space(struct dm_bio_prison_cell *cell)
+static void no_space(struct pool *pool, struct dm_bio_prison_cell *cell)
 {
 	struct bio *bio;
 	struct bio_list bios;
  
 	bio_list_init(&bios);
-	dm_cell_release(cell, &bios);
+	cell_release(pool, cell, &bios);
  
 	while ((bio = bio_list_pop(&bios)))
 		retry_on_resume(bio);
@@ -941,7 +990,7 @@
 	struct dm_thin_new_mapping *m;
  
 	build_virtual_key(tc->td, block, &key);
-	if (dm_bio_detain(tc->pool->prison, &key, bio, &cell))
+	if (bio_detain(tc->pool, &key, bio, &cell))
 		return;
  
 	r = dm_thin_find_block(tc->td, block, 1, &lookup_result);
@@ -953,7 +1002,7 @@
 		 * on this block.
 		 */
 		build_data_key(tc->td, lookup_result.block, &key2);
-		if (dm_bio_detain(tc->pool->prison, &key2, bio, &cell2)) {
+		if (bio_detain(tc->pool, &key2, bio, &cell2)) {
 			cell_defer_no_holder(tc, cell);
 			break;
 		}
  
@@ -1029,13 +1078,13 @@
 		break;
  
 	case -ENOSPC:
-		no_space(cell);
+		no_space(tc->pool, cell);
 		break;
  
 	default:
 		DMERR_LIMIT("%s: alloc_data_block() failed: error = %d",
 			    __func__, r);
-		dm_cell_error(cell);
+		cell_error(tc->pool, cell);
 		break;
 	}
 }
@@ -1053,7 +1102,7 @@
 	 * of being broken so we have nothing further to do here.
 	 */
 	build_data_key(tc->td, lookup_result->block, &key);
-	if (dm_bio_detain(pool->prison, &key, bio, &cell))
+	if (bio_detain(pool, &key, bio, &cell))
 		return;
  
 	if (bio_data_dir(bio) == WRITE && bio->bi_size)
  
@@ -1074,12 +1123,13 @@
 {
 	int r;
 	dm_block_t data_block;
+	struct pool *pool = tc->pool;
  
 	/*
 	 * Remap empty bios (flushes) immediately, without provisioning.
 	 */
 	if (!bio->bi_size) {
-		inc_all_io_entry(tc->pool, bio);
+		inc_all_io_entry(pool, bio);
 		cell_defer_no_holder(tc, cell);
  
 		remap_and_issue(tc, bio, 0);
  
@@ -1106,14 +1156,14 @@
 		break;
  
 	case -ENOSPC:
-		no_space(cell);
+		no_space(pool, cell);
 		break;
  
 	default:
 		DMERR_LIMIT("%s: alloc_data_block() failed: error = %d",
 			    __func__, r);
-		set_pool_mode(tc->pool, PM_READ_ONLY);
-		dm_cell_error(cell);
+		set_pool_mode(pool, PM_READ_ONLY);
+		cell_error(pool, cell);
 		break;
 	}
 }
@@ -1121,6 +1171,7 @@
 static void process_bio(struct thin_c *tc, struct bio *bio)
 {
 	int r;
+	struct pool *pool = tc->pool;
 	dm_block_t block = get_bio_block(tc, bio);
 	struct dm_bio_prison_cell *cell;
 	struct dm_cell_key key;
@@ -1131,7 +1182,7 @@
 	 * being provisioned so we have nothing further to do here.
 	 */
 	build_virtual_key(tc->td, block, &key);
-	if (dm_bio_detain(tc->pool->prison, &key, bio, &cell))
+	if (bio_detain(pool, &key, bio, &cell))
 		return;
  
 	r = dm_thin_find_block(tc->td, block, 1, &lookup_result);
  
@@ -1139,9 +1190,9 @@
 	case 0:
 		if (lookup_result.shared) {
 			process_shared_bio(tc, bio, block, &lookup_result);
-			cell_defer_no_holder(tc, cell);
+			cell_defer_no_holder(tc, cell); /* FIXME: pass this cell into process_shared? */
 		} else {
-			inc_all_io_entry(tc->pool, bio);
+			inc_all_io_entry(pool, bio);
 			cell_defer_no_holder(tc, cell);
  
 			remap_and_issue(tc, bio, lookup_result.block);
@@ -1150,7 +1201,7 @@
  
 	case -ENODATA:
 		if (bio_data_dir(bio) == READ && tc->origin_dev) {
-			inc_all_io_entry(tc->pool, bio);
+			inc_all_io_entry(pool, bio);
 			cell_defer_no_holder(tc, cell);
  
 			remap_to_origin_and_issue(tc, bio);
  
@@ -1429,11 +1480,11 @@
 		}
  
 		build_virtual_key(tc->td, block, &key);
-		if (dm_bio_detain(tc->pool->prison, &key, bio, &cell1))
+		if (bio_detain(tc->pool, &key, bio, &cell1))
 			return DM_MAPIO_SUBMITTED;
  
 		build_data_key(tc->td, result.block, &key);
-		if (dm_bio_detain(tc->pool->prison, &key, bio, &cell2)) {
+		if (bio_detain(tc->pool, &key, bio, &cell2)) {
 			cell_defer_no_holder(tc, cell1);
 			return DM_MAPIO_SUBMITTED;
 		}
...	...	@@ -16,7 +16,6 @@
16	16
17	17	struct dm_bio_prison_cell {
18	18	struct hlist_node list;
19		- struct dm_bio_prison *prison;
20	19	struct dm_cell_key key;
21	20	struct bio *holder;
22	21	struct bio_list bios;
...	...	@@ -87,6 +86,19 @@
87	86	}
88	87	EXPORT_SYMBOL_GPL(dm_bio_prison_destroy);
89	88
	89	+struct dm_bio_prison_cell dm_bio_prison_alloc_cell(struct dm_bio_prison prison, gfp_t gfp)
	90	+{
	91	+ return mempool_alloc(prison->cell_pool, gfp);
	92	+}
	93	+EXPORT_SYMBOL_GPL(dm_bio_prison_alloc_cell);
	94	+
	95	+void dm_bio_prison_free_cell(struct dm_bio_prison *prison,
	96	+ struct dm_bio_prison_cell *cell)
	97	+{
	98	+ mempool_free(cell, prison->cell_pool);
	99	+}
	100	+EXPORT_SYMBOL_GPL(dm_bio_prison_free_cell);
	101	+
90	102	static uint32_t hash_key(struct dm_bio_prison prison, struct dm_cell_key key)
91	103	{
92	104	const unsigned long BIG_PRIME = 4294967291UL;
93	105
94	106
95	107
96	108
97	109
98	110
99	111
100	112
101	113
102	114
103	115
104	116
105	117
106	118
107	119
...	...	@@ -114,91 +126,86 @@
114	126	return NULL;
115	127	}
116	128
117		-/*
118		- * This may block if a new cell needs allocating. You must ensure that
119		- * cells will be unlocked even if the calling thread is blocked.
120		- *
121		- * Returns 1 if the cell was already held, 0 if @inmate is the new holder.
122		- */
123		-int dm_bio_detain(struct dm_bio_prison prison, struct dm_cell_key key,
124		- struct bio inmate, struct dm_bio_prison_cell *ref)
	129	+static void __setup_new_cell(struct dm_bio_prison *prison,
	130	+ struct dm_cell_key *key,
	131	+ struct bio *holder,
	132	+ uint32_t hash,
	133	+ struct dm_bio_prison_cell *cell)
125	134	{
126		- int r = 1;
127		- unsigned long flags;
	135	+ memcpy(&cell->key, key, sizeof(cell->key));
	136	+ cell->holder = holder;
	137	+ bio_list_init(&cell->bios);
	138	+ hlist_add_head(&cell->list, prison->cells + hash);
	139	+}
	140	+
	141	+static int __bio_detain(struct dm_bio_prison *prison,
	142	+ struct dm_cell_key *key,
	143	+ struct bio *inmate,
	144	+ struct dm_bio_prison_cell *cell_prealloc,
	145	+ struct dm_bio_prison_cell **cell_result)
	146	+{
128	147	uint32_t hash = hash_key(prison, key);
129		- struct dm_bio_prison_cell cell, cell2;
	148	+ struct dm_bio_prison_cell *cell;
130	149
131		- BUG_ON(hash > prison->nr_buckets);
132		-
133		- spin_lock_irqsave(&prison->lock, flags);
134		-
135	150	cell = __search_bucket(prison->cells + hash, key);
136	151	if (cell) {
137		- bio_list_add(&cell->bios, inmate);
138		- goto out;
	152	+ if (inmate)
	153	+ bio_list_add(&cell->bios, inmate);
	154	+ *cell_result = cell;
	155	+ return 1;
139	156	}
140	157
141		- /*
142		- * Allocate a new cell
143		- */
144		- spin_unlock_irqrestore(&prison->lock, flags);
145		- cell2 = mempool_alloc(prison->cell_pool, GFP_NOIO);
146		- spin_lock_irqsave(&prison->lock, flags);
	158	+ __setup_new_cell(prison, key, inmate, hash, cell_prealloc);
	159	+ *cell_result = cell_prealloc;
	160	+ return 0;
	161	+}
147	162
148		- /*
149		- * We've been unlocked, so we have to double check that
150		- * nobody else has inserted this cell in the meantime.
151		- */
152		- cell = __search_bucket(prison->cells + hash, key);
153		- if (cell) {
154		- mempool_free(cell2, prison->cell_pool);
155		- bio_list_add(&cell->bios, inmate);
156		- goto out;
157		- }
	163	+static int bio_detain(struct dm_bio_prison *prison,
	164	+ struct dm_cell_key *key,
	165	+ struct bio *inmate,
	166	+ struct dm_bio_prison_cell *cell_prealloc,
	167	+ struct dm_bio_prison_cell **cell_result)
	168	+{
	169	+ int r;
	170	+ unsigned long flags;
158	171
159		- /*
160		- * Use new cell.
161		- */
162		- cell = cell2;
163		-
164		- cell->prison = prison;
165		- memcpy(&cell->key, key, sizeof(cell->key));
166		- cell->holder = inmate;
167		- bio_list_init(&cell->bios);
168		- hlist_add_head(&cell->list, prison->cells + hash);
169		-
170		- r = 0;
171		-
172		-out:
	172	+ spin_lock_irqsave(&prison->lock, flags);
	173	+ r = __bio_detain(prison, key, inmate, cell_prealloc, cell_result);
173	174	spin_unlock_irqrestore(&prison->lock, flags);
174	175
175		- *ref = cell;
176		-
177	176	return r;
178	177	}
	178	+
	179	+int dm_bio_detain(struct dm_bio_prison *prison,
	180	+ struct dm_cell_key *key,
	181	+ struct bio *inmate,
	182	+ struct dm_bio_prison_cell *cell_prealloc,
	183	+ struct dm_bio_prison_cell **cell_result)
	184	+{
	185	+ return bio_detain(prison, key, inmate, cell_prealloc, cell_result);
	186	+}
179	187	EXPORT_SYMBOL_GPL(dm_bio_detain);
180	188
181	189	/*
182	190	* @inmates must have been initialised prior to this call
183	191	*/
184		-static void __cell_release(struct dm_bio_prison_cell cell, struct bio_list inmates)
	192	+static void __cell_release(struct dm_bio_prison_cell *cell,
	193	+ struct bio_list *inmates)
185	194	{
186		- struct dm_bio_prison *prison = cell->prison;
187		-
188	195	hlist_del(&cell->list);
189	196
190	197	if (inmates) {
191		- bio_list_add(inmates, cell->holder);
	198	+ if (cell->holder)
	199	+ bio_list_add(inmates, cell->holder);
192	200	bio_list_merge(inmates, &cell->bios);
193	201	}
194		-
195		- mempool_free(cell, prison->cell_pool);
196	202	}
197	203
198		-void dm_cell_release(struct dm_bio_prison_cell cell, struct bio_list bios)
	204	+void dm_cell_release(struct dm_bio_prison *prison,
	205	+ struct dm_bio_prison_cell *cell,
	206	+ struct bio_list *bios)
199	207	{
200	208	unsigned long flags;
201		- struct dm_bio_prison *prison = cell->prison;
202	209
203	210	spin_lock_irqsave(&prison->lock, flags);
204	211	__cell_release(cell, bios);
205	212
206	213
207	214
208	215
...	...	@@ -209,20 +216,18 @@
209	216	/*
210	217	* Sometimes we don't want the holder, just the additional bios.
211	218	*/
212		-static void __cell_release_no_holder(struct dm_bio_prison_cell cell, struct bio_list inmates)
	219	+static void __cell_release_no_holder(struct dm_bio_prison_cell *cell,
	220	+ struct bio_list *inmates)
213	221	{
214		- struct dm_bio_prison *prison = cell->prison;
215		-
216	222	hlist_del(&cell->list);
217	223	bio_list_merge(inmates, &cell->bios);
218		-
219		- mempool_free(cell, prison->cell_pool);
220	224	}
221	225
222		-void dm_cell_release_no_holder(struct dm_bio_prison_cell cell, struct bio_list inmates)
	226	+void dm_cell_release_no_holder(struct dm_bio_prison *prison,
	227	+ struct dm_bio_prison_cell *cell,
	228	+ struct bio_list *inmates)
223	229	{
224	230	unsigned long flags;
225		- struct dm_bio_prison *prison = cell->prison;
226	231
227	232	spin_lock_irqsave(&prison->lock, flags);
228	233	__cell_release_no_holder(cell, inmates);
229	234
...	...	@@ -230,9 +235,9 @@
230	235	}
231	236	EXPORT_SYMBOL_GPL(dm_cell_release_no_holder);
232	237
233		-void dm_cell_error(struct dm_bio_prison_cell *cell)
	238	+void dm_cell_error(struct dm_bio_prison *prison,
	239	+ struct dm_bio_prison_cell *cell)
234	240	{
235		- struct dm_bio_prison *prison = cell->prison;
236	241	struct bio_list bios;
237	242	struct bio *bio;
238	243	unsigned long flags;
...	...	@@ -35,17 +35,36 @@
35	35	void dm_bio_prison_destroy(struct dm_bio_prison *prison);
36	36
37	37	/*
38		- * This may block if a new cell needs allocating. You must ensure that
39		- * cells will be unlocked even if the calling thread is blocked.
	38	+ * These two functions just wrap a mempool. This is a transitory step:
	39	+ * Eventually all bio prison clients should manage their own cell memory.
40	40	*
	41	+ * Like mempool_alloc(), dm_bio_prison_alloc_cell() can only fail if called
	42	+ * in interrupt context or passed GFP_NOWAIT.
	43	+ */
	44	+struct dm_bio_prison_cell dm_bio_prison_alloc_cell(struct dm_bio_prison prison,
	45	+ gfp_t gfp);
	46	+void dm_bio_prison_free_cell(struct dm_bio_prison *prison,
	47	+ struct dm_bio_prison_cell *cell);
	48	+
	49	+/*
	50	+ * An atomic op that combines retrieving a cell, and adding a bio to it.
	51	+ *
41	52	* Returns 1 if the cell was already held, 0 if @inmate is the new holder.
42	53	*/
43		-int dm_bio_detain(struct dm_bio_prison prison, struct dm_cell_key key,
44		- struct bio inmate, struct dm_bio_prison_cell *ref);
	54	+int dm_bio_detain(struct dm_bio_prison *prison,
	55	+ struct dm_cell_key *key,
	56	+ struct bio *inmate,
	57	+ struct dm_bio_prison_cell *cell_prealloc,
	58	+ struct dm_bio_prison_cell **cell_result);
45	59
46		-void dm_cell_release(struct dm_bio_prison_cell cell, struct bio_list bios);
47		-void dm_cell_release_no_holder(struct dm_bio_prison_cell cell, struct bio_list inmates);
48		-void dm_cell_error(struct dm_bio_prison_cell *cell);
	60	+void dm_cell_release(struct dm_bio_prison *prison,
	61	+ struct dm_bio_prison_cell *cell,
	62	+ struct bio_list *bios);
	63	+void dm_cell_release_no_holder(struct dm_bio_prison *prison,
	64	+ struct dm_bio_prison_cell *cell,
	65	+ struct bio_list *inmates);
	66	+void dm_cell_error(struct dm_bio_prison *prison,
	67	+ struct dm_bio_prison_cell *cell);
49	68
50	69	/----------------------------------------------------------------/
51	70
...	...	@@ -229,6 +229,54 @@
229	229
230	230	/----------------------------------------------------------------/
231	231
	232	+static int bio_detain(struct pool pool, struct dm_cell_key key, struct bio *bio,
	233	+ struct dm_bio_prison_cell **cell_result)
	234	+{
	235	+ int r;
	236	+ struct dm_bio_prison_cell *cell_prealloc;
	237	+
	238	+ /*
	239	+ * Allocate a cell from the prison's mempool.
	240	+ * This might block but it can't fail.
	241	+ */
	242	+ cell_prealloc = dm_bio_prison_alloc_cell(pool->prison, GFP_NOIO);
	243	+
	244	+ r = dm_bio_detain(pool->prison, key, bio, cell_prealloc, cell_result);
	245	+ if (r)
	246	+ /*
	247	+ * We reused an old cell; we can get rid of
	248	+ * the new one.
	249	+ */
	250	+ dm_bio_prison_free_cell(pool->prison, cell_prealloc);
	251	+
	252	+ return r;
	253	+}
	254	+
	255	+static void cell_release(struct pool *pool,
	256	+ struct dm_bio_prison_cell *cell,
	257	+ struct bio_list *bios)
	258	+{
	259	+ dm_cell_release(pool->prison, cell, bios);
	260	+ dm_bio_prison_free_cell(pool->prison, cell);
	261	+}
	262	+
	263	+static void cell_release_no_holder(struct pool *pool,
	264	+ struct dm_bio_prison_cell *cell,
	265	+ struct bio_list *bios)
	266	+{
	267	+ dm_cell_release_no_holder(pool->prison, cell, bios);
	268	+ dm_bio_prison_free_cell(pool->prison, cell);
	269	+}
	270	+
	271	+static void cell_error(struct pool *pool,
	272	+ struct dm_bio_prison_cell *cell)
	273	+{
	274	+ dm_cell_error(pool->prison, cell);
	275	+ dm_bio_prison_free_cell(pool->prison, cell);
	276	+}
	277	+
	278	+/----------------------------------------------------------------/
	279	+
232	280	/*
233	281	* A global list of pools that uses a struct mapped_device as a key.
234	282	*/
235	283
...	...	@@ -524,14 +572,14 @@
524	572	unsigned long flags;
525	573
526	574	spin_lock_irqsave(&pool->lock, flags);
527		- dm_cell_release(cell, &pool->deferred_bios);
	575	+ cell_release(pool, cell, &pool->deferred_bios);
528	576	spin_unlock_irqrestore(&tc->pool->lock, flags);
529	577
530	578	wake_worker(pool);
531	579	}
532	580
533	581	/*
534		- * Same as cell_defer except it omits the original holder of the cell.
	582	+ * Same as cell_defer above, except it omits the original holder of the cell.
535	583	*/
536	584	static void cell_defer_no_holder(struct thin_c tc, struct dm_bio_prison_cell cell)
537	585	{
...	...	@@ -539,7 +587,7 @@
539	587	unsigned long flags;
540	588
541	589	spin_lock_irqsave(&pool->lock, flags);
542		- dm_cell_release_no_holder(cell, &pool->deferred_bios);
	590	+ cell_release_no_holder(pool, cell, &pool->deferred_bios);
543	591	spin_unlock_irqrestore(&pool->lock, flags);
544	592
545	593	wake_worker(pool);
546	594
...	...	@@ -549,13 +597,14 @@
549	597	{
550	598	if (m->bio)
551	599	m->bio->bi_end_io = m->saved_bi_end_io;
552		- dm_cell_error(m->cell);
	600	+ cell_error(m->tc->pool, m->cell);
553	601	list_del(&m->list);
554	602	mempool_free(m, m->tc->pool->mapping_pool);
555	603	}
556	604	static void process_prepared_mapping(struct dm_thin_new_mapping *m)
557	605	{
558	606	struct thin_c *tc = m->tc;
	607	+ struct pool *pool = tc->pool;
559	608	struct bio *bio;
560	609	int r;
561	610
...	...	@@ -564,7 +613,7 @@
564	613	bio->bi_end_io = m->saved_bi_end_io;
565	614
566	615	if (m->err) {
567		- dm_cell_error(m->cell);
	616	+ cell_error(pool, m->cell);
568	617	goto out;
569	618	}
570	619
...	...	@@ -576,7 +625,7 @@
576	625	r = dm_thin_insert_block(tc->td, m->virt_block, m->data_block);
577	626	if (r) {
578	627	DMERR_LIMIT("dm_thin_insert_block() failed");
579		- dm_cell_error(m->cell);
	628	+ cell_error(pool, m->cell);
580	629	goto out;
581	630	}
582	631
...	...	@@ -594,7 +643,7 @@
594	643
595	644	out:
596	645	list_del(&m->list);
597		- mempool_free(m, tc->pool->mapping_pool);
	646	+ mempool_free(m, pool->mapping_pool);
598	647	}
599	648
600	649	static void process_prepared_discard_fail(struct dm_thin_new_mapping *m)
...	...	@@ -745,7 +794,7 @@
745	794	if (r < 0) {
746	795	mempool_free(m, pool->mapping_pool);
747	796	DMERR_LIMIT("dm_kcopyd_copy() failed");
748		- dm_cell_error(cell);
	797	+ cell_error(pool, cell);
749	798	}
750	799	}
751	800	}
...	...	@@ -811,7 +860,7 @@
811	860	if (r < 0) {
812	861	mempool_free(m, pool->mapping_pool);
813	862	DMERR_LIMIT("dm_kcopyd_zero() failed");
814		- dm_cell_error(cell);
	863	+ cell_error(pool, cell);
815	864	}
816	865	}
817	866	}
818	867
...	...	@@ -917,13 +966,13 @@
917	966	spin_unlock_irqrestore(&pool->lock, flags);
918	967	}
919	968
920		-static void no_space(struct dm_bio_prison_cell *cell)
	969	+static void no_space(struct pool pool, struct dm_bio_prison_cell cell)
921	970	{
922	971	struct bio *bio;
923	972	struct bio_list bios;
924	973
925	974	bio_list_init(&bios);
926		- dm_cell_release(cell, &bios);
	975	+ cell_release(pool, cell, &bios);
927	976
928	977	while ((bio = bio_list_pop(&bios)))
929	978	retry_on_resume(bio);
...	...	@@ -941,7 +990,7 @@
941	990	struct dm_thin_new_mapping *m;
942	991
943	992	build_virtual_key(tc->td, block, &key);
944		- if (dm_bio_detain(tc->pool->prison, &key, bio, &cell))
	993	+ if (bio_detain(tc->pool, &key, bio, &cell))
945	994	return;
946	995
947	996	r = dm_thin_find_block(tc->td, block, 1, &lookup_result);
...	...	@@ -953,7 +1002,7 @@
953	1002	* on this block.
954	1003	*/
955	1004	build_data_key(tc->td, lookup_result.block, &key2);
956		- if (dm_bio_detain(tc->pool->prison, &key2, bio, &cell2)) {
	1005	+ if (bio_detain(tc->pool, &key2, bio, &cell2)) {
957	1006	cell_defer_no_holder(tc, cell);
958	1007	break;
959	1008	}
960	1009
...	...	@@ -1029,13 +1078,13 @@
1029	1078	break;
1030	1079
1031	1080	case -ENOSPC:
1032		- no_space(cell);
	1081	+ no_space(tc->pool, cell);
1033	1082	break;
1034	1083
1035	1084	default:
1036	1085	DMERR_LIMIT("%s: alloc_data_block() failed: error = %d",
1037	1086	__func__, r);
1038		- dm_cell_error(cell);
	1087	+ cell_error(tc->pool, cell);
1039	1088	break;
1040	1089	}
1041	1090	}
...	...	@@ -1053,7 +1102,7 @@
1053	1102	* of being broken so we have nothing further to do here.
1054	1103	*/
1055	1104	build_data_key(tc->td, lookup_result->block, &key);
1056		- if (dm_bio_detain(pool->prison, &key, bio, &cell))
	1105	+ if (bio_detain(pool, &key, bio, &cell))
1057	1106	return;
1058	1107
1059	1108	if (bio_data_dir(bio) == WRITE && bio->bi_size)
1060	1109
...	...	@@ -1074,12 +1123,13 @@
1074	1123	{
1075	1124	int r;
1076	1125	dm_block_t data_block;
	1126	+ struct pool *pool = tc->pool;
1077	1127
1078	1128	/*
1079	1129	* Remap empty bios (flushes) immediately, without provisioning.
1080	1130	*/
1081	1131	if (!bio->bi_size) {
1082		- inc_all_io_entry(tc->pool, bio);
	1132	+ inc_all_io_entry(pool, bio);
1083	1133	cell_defer_no_holder(tc, cell);
1084	1134
1085	1135	remap_and_issue(tc, bio, 0);
1086	1136
...	...	@@ -1106,14 +1156,14 @@
1106	1156	break;
1107	1157
1108	1158	case -ENOSPC:
1109		- no_space(cell);
	1159	+ no_space(pool, cell);
1110	1160	break;
1111	1161
1112	1162	default:
1113	1163	DMERR_LIMIT("%s: alloc_data_block() failed: error = %d",
1114	1164	__func__, r);
1115		- set_pool_mode(tc->pool, PM_READ_ONLY);
1116		- dm_cell_error(cell);
	1165	+ set_pool_mode(pool, PM_READ_ONLY);
	1166	+ cell_error(pool, cell);
1117	1167	break;
1118	1168	}
1119	1169	}
...	...	@@ -1121,6 +1171,7 @@
1121	1171	static void process_bio(struct thin_c tc, struct bio bio)
1122	1172	{
1123	1173	int r;
	1174	+ struct pool *pool = tc->pool;
1124	1175	dm_block_t block = get_bio_block(tc, bio);
1125	1176	struct dm_bio_prison_cell *cell;
1126	1177	struct dm_cell_key key;
...	...	@@ -1131,7 +1182,7 @@
1131	1182	* being provisioned so we have nothing further to do here.
1132	1183	*/
1133	1184	build_virtual_key(tc->td, block, &key);
1134		- if (dm_bio_detain(tc->pool->prison, &key, bio, &cell))
	1185	+ if (bio_detain(pool, &key, bio, &cell))
1135	1186	return;
1136	1187
1137	1188	r = dm_thin_find_block(tc->td, block, 1, &lookup_result);
1138	1189
...	...	@@ -1139,9 +1190,9 @@
1139	1190	case 0:
1140	1191	if (lookup_result.shared) {
1141	1192	process_shared_bio(tc, bio, block, &lookup_result);
1142		- cell_defer_no_holder(tc, cell);
	1193	+ cell_defer_no_holder(tc, cell); /* FIXME: pass this cell into process_shared? */
1143	1194	} else {
1144		- inc_all_io_entry(tc->pool, bio);
	1195	+ inc_all_io_entry(pool, bio);
1145	1196	cell_defer_no_holder(tc, cell);
1146	1197
1147	1198	remap_and_issue(tc, bio, lookup_result.block);
...	...	@@ -1150,7 +1201,7 @@
1150	1201
1151	1202	case -ENODATA:
1152	1203	if (bio_data_dir(bio) == READ && tc->origin_dev) {
1153		- inc_all_io_entry(tc->pool, bio);
	1204	+ inc_all_io_entry(pool, bio);
1154	1205	cell_defer_no_holder(tc, cell);
1155	1206
1156	1207	remap_to_origin_and_issue(tc, bio);
1157	1208
...	...	@@ -1429,11 +1480,11 @@
1429	1480	}
1430	1481
1431	1482	build_virtual_key(tc->td, block, &key);
1432		- if (dm_bio_detain(tc->pool->prison, &key, bio, &cell1))
	1483	+ if (bio_detain(tc->pool, &key, bio, &cell1))
1433	1484	return DM_MAPIO_SUBMITTED;
1434	1485
1435	1486	build_data_key(tc->td, result.block, &key);
1436		- if (dm_bio_detain(tc->pool->prison, &key, bio, &cell2)) {
	1487	+ if (bio_detain(tc->pool, &key, bio, &cell2)) {
1437	1488	cell_defer_no_holder(tc, cell1);
1438	1489	return DM_MAPIO_SUBMITTED;
1439	1490	}