Allocate separate crypt io, if the requesest must be split.

There are 2 problems with the current implementation

1) After split, it waits for finishing first clone,
which can cause some performance degradation.

2) There is only one waitqueue, if there are still running
previous cyrpt io, and the following io is going to be split,
code wait for wake_up call.
Unfortunatelly this call could be from previos io, and fires
fragment processing too early -> possible data corruption.

Solve it by allocating new crypt io for the all fragments
and maintain base_io counter for them.

In low memory situation, this only add one more allocated
io from pool (base_io + first fragment).
If the pool is empty, next allocation simple waits for previous 
fragments to complete.

The split operation is very uncommon, but happens sometimes.

Signed-off-by: Milan Broz <mbroz@redhat.com>
---
 drivers/md/dm-crypt.c |   43 ++++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 42 insertions(+), 1 deletion(-)

Index: linux/drivers/md/dm-crypt.c
===================================================================
--- linux.orig/drivers/md/dm-crypt.c	2008-08-04 14:40:15.000000000 +0100
+++ linux/drivers/md/dm-crypt.c	2008-08-04 14:40:17.000000000 +0100
@@ -56,6 +56,7 @@ struct dm_crypt_io {
 	atomic_t pending;
 	int error;
 	sector_t sector;
+	struct dm_crypt_io *base_io;
 };
 
 struct dm_crypt_request {
@@ -533,6 +534,7 @@ static struct dm_crypt_io *crypt_io_allo
 	io->base_bio = bio;
 	io->sector = sector;
 	io->error = 0;
+	io->base_io = NULL;
 	atomic_set(&io->pending, 0);
 
 	return io;
@@ -554,7 +556,14 @@ static void crypt_dec_pending(struct dm_
 	if (!atomic_dec_and_test(&io->pending))
 		return;
 
-	bio_endio(io->base_bio, io->error);
+	/* Split crypt io must not free base_bio */
+	if (unlikely(io->base_io)) {
+		if (io->error && !io->base_io->error)
+			io->base_io->error = io->error;
+		crypt_dec_pending(io->base_io);
+	} else
+		bio_endio(io->base_bio, io->error);
+
 	mempool_free(io, cc->io_pool);
 }
 
@@ -697,6 +706,7 @@ static void kcryptd_crypt_write_convert(
 {
 	struct crypt_config *cc = io->target->private;
 	struct bio *clone;
+	struct dm_crypt_io *new_io;
 	unsigned crypt_finished, out_of_pages = 0;
 	unsigned remaining = io->base_bio->bi_size;
 	sector_t sector = io->sector;
@@ -751,6 +761,37 @@ static void kcryptd_crypt_write_convert(
 		if (unlikely(out_of_pages))
 			congestion_wait(WRITE, HZ/100);
 
+		/*
+		 * Running async crypto is gonna to be split because
+		 * crypto context cannot be shared -> alloc new io
+		 */
+		if (unlikely(!crypt_finished && remaining)) {
+			new_io = crypt_io_alloc(io->target, io->base_bio,
+						sector);
+
+			/* Initialize context to point after submitted crypt */
+			crypt_inc_pending(new_io);
+			crypt_convert_init(cc, &new_io->ctx, NULL,
+					   io->base_bio, sector);
+			new_io->ctx.idx_in = io->ctx.idx_in;
+			new_io->ctx.offset_in = io->ctx.offset_in;
+
+			/*
+			 * For first fragment, keep pending count.
+			 * Next fragments increase base_io instead
+			 * of previous fragment pending count.
+			 */
+			if (io->base_io) {
+				new_io->base_io = io->base_io;
+				crypt_inc_pending(io->base_io);
+				crypt_dec_pending(io);
+			} else
+				new_io->base_io = io;
+
+			/* Switch new io and restart the machinery */
+			io = new_io;
+		}
+
 		if (unlikely(remaining))
 			wait_event(cc->writeq, !atomic_read(&io->ctx.pending));
 	}