block processor: simplify backlog accounting

Simply count the number of blocks we hand out (malloc'ed or recycled)
and decrease the counter when we put blocks back for recycling.

The sync() part becomes a little more complicated, because we can get
stuck with a backlog of 1 or 2 because we have a fragment or current
block buffer in use. We also need to accout for this when creating the
processor, because we need to be able to request at least 2 blocks
without stalling.

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

3 files changed, 26 insertions, 8 deletions

diff --git a/lib/sqfs/block_processor/backend.c b/lib/sqfs/block_processor/backend.c
index ff142c9..68e299e 100644
--- a/lib/sqfs/block_processor/backend.c
+++ b/lib/sqfs/block_processor/backend.c
@@ -48,6 +48,8 @@ static void release_old_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 {
 	blk->next = proc->free_list;
 	proc->free_list = blk;
+
+	proc->backlog -= 1;
 }
 
 static int process_completed_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
@@ -216,7 +218,8 @@ static int process_completed_fragment(sqfs_block_processor_t *proc,
 	return 0;
 fail:
 	free(chunk);
-	free(frag);
+	if (frag != NULL)
+		release_old_block(proc, frag);
 	return err;
 }
 
@@ -236,7 +239,6 @@ static void store_io_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 	}
 
 	blk->next = it;
-	proc->backlog += 1;
 }
 
 int dequeue_block(sqfs_block_processor_t *proc)
@@ -253,7 +255,6 @@ int dequeue_block(sqfs_block_processor_t *proc)
 			blk = proc->io_queue;
 			proc->io_queue = blk->next;
 			proc->io_deq_seq_num += 1;
-			proc->backlog -= 1;
 
 			status = process_completed_block(proc, blk);
 			if (status != 0)
@@ -270,8 +271,6 @@ int dequeue_block(sqfs_block_processor_t *proc)
 			return status ? status : SQFS_ERROR_INTERNAL;
 		}
 
-		proc->backlog -= 1;
-
 		if (blk->flags & SQFS_BLK_IS_FRAGMENT) {
 			status = process_completed_fragment(proc, blk);
 			if (status != 0)
diff --git a/lib/sqfs/block_processor/block_processor.c b/lib/sqfs/block_processor/block_processor.c
index f9313db..29496f9 100644
--- a/lib/sqfs/block_processor/block_processor.c
+++ b/lib/sqfs/block_processor/block_processor.c
@@ -98,7 +98,20 @@ int sqfs_block_processor_sync(sqfs_block_processor_t *proc)
 {
 	int ret;
 
-	while (proc->backlog > 0) {
+	for (;;) {
+		if (proc->backlog == 0)
+			break;
+
+		if ((proc->backlog == 1) &&
+		    (proc->frag_block != NULL || proc->blk_current != NULL)) {
+			break;
+		}
+
+		if ((proc->backlog == 2) &&
+		    proc->frag_block != NULL && proc->blk_current != NULL) {
+			break;
+		}
+
 		ret = dequeue_block(proc);
 		if (ret != 0)
 			return ret;
@@ -162,6 +175,10 @@ int sqfs_block_processor_create_ex(const sqfs_block_processor_desc_t *desc,
 	proc->stats.size = sizeof(proc->stats);
 	((sqfs_object_t *)proc)->destroy = block_processor_destroy;
 
+	/* we need at least one current data block + one fragment block */
+	if (proc->max_backlog < 2)
+		proc->max_backlog = 2;
+
 	/* create the thread pool */
 	proc->pool = thread_pool_create(desc->num_workers, process_block);
 	if (proc->pool == NULL) {
diff --git a/lib/sqfs/block_processor/frontend.c b/lib/sqfs/block_processor/frontend.c
index ccf54c0..a4a38e9 100644
--- a/lib/sqfs/block_processor/frontend.c
+++ b/lib/sqfs/block_processor/frontend.c
@@ -28,6 +28,8 @@ static int get_new_block(sqfs_block_processor_t *proc, sqfs_block_t **out)
 
 	memset(blk, 0, sizeof(*blk));
 	*out = blk;
+
+	proc->backlog += 1;
 	return 0;
 }
 
@@ -56,11 +58,11 @@ int enqueue_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 		if (status == 0)
 			status = SQFS_ERROR_ALLOC;
 
-		free(blk);
+		blk->next = proc->free_list;
+		proc->free_list = blk;
 		return status;
 	}
 
-	proc->backlog += 1;
 	return 0;
 }