block processor: keep duplicate copies of in-flight fragment blocks

If we want full, byte-for byte, verification of fragments during
de-duplication we need to check back with the blocks already written
to disk, or with the ones that are in flight.

The previous, extremely hacky approach simply locked up the thread
pool and investigated the queues. For the new approach, we treat the
thread pool as completely opaque and don't try to touch it.

This commit modifies the block processor to keep duplicate copies of
each submitted fragment block around, that are cleaned up once the
block is dequeued and written to disk. So instead of touching the
thread pool, we can simply investigate the in-fligth-block list and
the current block, before resorting to reading back fragment blocks
from the file.

Signed-off-by: David Oberhollenzer <david.oberhollenzer@sigma-star.at>

4 files changed, 47 insertions, 12 deletions

diff --git a/lib/sqfs/block_processor/backend.c b/lib/sqfs/block_processor/backend.c
index 68e299e..d71c632 100644
--- a/lib/sqfs/block_processor/backend.c
+++ b/lib/sqfs/block_processor/backend.c
@@ -58,6 +58,24 @@ static int process_completed_block(sqfs_block_processor_t *proc, sqfs_block_t *b
 	sqfs_u32 size;
 	int err;
 
+	if (blk->flags & SQFS_BLK_FRAGMENT_BLOCK) {
+		sqfs_block_t *it = proc->fblk_in_flight, *prev = NULL;
+
+		while (it != NULL && it->index != blk->index) {
+			prev = it;
+			it = it->next;
+		}
+
+		if (it != NULL) {
+			if (prev == NULL) {
+				proc->fblk_in_flight = it->next;
+			} else {
+				prev->next = it->next;
+			}
+			free(it);
+		}
+	}
+
 	err = proc->wr->write_data_block(proc->wr, blk->user, blk->size,
 					 blk->checksum,
 					 blk->flags & ~BLK_FLAG_INTERNAL,
diff --git a/lib/sqfs/block_processor/block_processor.c b/lib/sqfs/block_processor/block_processor.c
index 29496f9..19a538a 100644
--- a/lib/sqfs/block_processor/block_processor.c
+++ b/lib/sqfs/block_processor/block_processor.c
@@ -56,25 +56,25 @@ static void ht_delete_function(struct hash_entry *entry)
 	free(entry->data);
 }
 
+static void free_block_list(sqfs_block_t *list)
+{
+	while (list != NULL) {
+		sqfs_block_t *it = list;
+		list = it->next;
+		free(it);
+	}
+}
+
 static void block_processor_destroy(sqfs_object_t *base)
 {
 	sqfs_block_processor_t *proc = (sqfs_block_processor_t *)base;
-	sqfs_block_t *it;
 
 	free(proc->frag_block);
 	free(proc->blk_current);
 
-	while (proc->free_list != NULL) {
-		it = proc->free_list;
-		proc->free_list = it->next;
-		free(it);
-	}
-
-	while (proc->io_queue != NULL) {
-		it = proc->io_queue;
-		proc->io_queue = it->next;
-		free(it);
-	}
+	free_block_list(proc->free_list);
+	free_block_list(proc->io_queue);
+	free_block_list(proc->fblk_in_flight);
 
 	if (proc->frag_ht != NULL)
 		hash_table_destroy(proc->frag_ht, ht_delete_function);
diff --git a/lib/sqfs/block_processor/frontend.c b/lib/sqfs/block_processor/frontend.c
index a4a38e9..e8a4207 100644
--- a/lib/sqfs/block_processor/frontend.c
+++ b/lib/sqfs/block_processor/frontend.c
@@ -52,6 +52,21 @@ int enqueue_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 {
 	int status;
 
+	if ((blk->flags & SQFS_BLK_FRAGMENT_BLOCK) &&
+	    proc->file != NULL && proc->uncmp != NULL) {
+		sqfs_block_t *copy = alloc_flex(sizeof(*copy), 1, blk->size);
+
+		if (copy == NULL)
+			return SQFS_ERROR_ALLOC;
+
+		copy->size = blk->size;
+		copy->index = blk->index;
+		memcpy(copy->data, blk->data, blk->size);
+
+		copy->next = proc->fblk_in_flight;
+		proc->fblk_in_flight = copy;
+	}
+
 	if (proc->pool->submit(proc->pool, blk) != 0) {
 		status = proc->pool->get_status(proc->pool);
 
diff --git a/lib/sqfs/block_processor/internal.h b/lib/sqfs/block_processor/internal.h
index ef8ff0f..72e862a 100644
--- a/lib/sqfs/block_processor/internal.h
+++ b/lib/sqfs/block_processor/internal.h
@@ -98,6 +98,8 @@ struct sqfs_block_processor_t {
 	sqfs_block_t *io_queue;
 	sqfs_u32 io_seq_num;
 	sqfs_u32 io_deq_seq_num;
+
+	sqfs_block_t *fblk_in_flight;
 };
 
 SQFS_INTERNAL int enqueue_block(sqfs_block_processor_t *proc,