Process completed blocks one-by-one

This commit makes sure the block processor enqueue function dequeues
at most a single block which it dispatches.

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

4 files changed, 37 insertions, 49 deletions

diff --git a/lib/sqfs/blk_proc/internal.h b/lib/sqfs/blk_proc/internal.h
index 8cb1a23..ac628c8 100644
--- a/lib/sqfs/blk_proc/internal.h
+++ b/lib/sqfs/blk_proc/internal.h
@@ -97,7 +97,7 @@ SQFS_INTERNAL
 int sqfs_block_process(sqfs_block_t *block, sqfs_compressor_t *cmp,
 		       uint8_t *scratch, size_t scratch_size);
 
-SQFS_INTERNAL int process_completed_blocks(sqfs_block_processor_t *proc,
-					   sqfs_block_t *queue);
+SQFS_INTERNAL int process_completed_block(sqfs_block_processor_t *proc,
+					  sqfs_block_t *block);
 
 #endif /* INTERNAL_H */
diff --git a/lib/sqfs/blk_proc/process_block.c b/lib/sqfs/blk_proc/process_block.c
index 643d964..b303278 100644
--- a/lib/sqfs/blk_proc/process_block.c
+++ b/lib/sqfs/blk_proc/process_block.c
@@ -105,7 +105,7 @@ static size_t grow_fragment_table(sqfs_block_processor_t *proc, size_t index)
 	return 0;
 }
 
-static int handle_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
+int process_completed_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 {
 	size_t start, count;
 	uint64_t offset;
@@ -181,28 +181,3 @@ static int handle_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 
 	return 0;
 }
-
-int process_completed_blocks(sqfs_block_processor_t *proc, sqfs_block_t *queue)
-{
-	sqfs_block_t *it;
-	int err;
-
-	while (queue != NULL) {
-		it = queue;
-		queue = queue->next;
-
-		err = handle_block(proc, it);
-		free(it);
-
-		if (err) {
-			while (queue != NULL) {
-				it = queue;
-				queue = it->next;
-				free(it);
-			}
-			return err;
-		}
-	}
-
-	return 0;
-}
diff --git a/lib/sqfs/blk_proc/pthread.c b/lib/sqfs/blk_proc/pthread.c
index fb83bbe..ee73e52 100644
--- a/lib/sqfs/blk_proc/pthread.c
+++ b/lib/sqfs/blk_proc/pthread.c
@@ -200,7 +200,6 @@ void sqfs_block_processor_destroy(sqfs_block_processor_t *proc)
 int sqfs_block_processor_enqueue(sqfs_block_processor_t *proc,
 				 sqfs_block_t *block)
 {
-	sqfs_block_t *queue = NULL, *it, *prev;
 	int status;
 
 	pthread_mutex_lock(&proc->mtx);
@@ -236,25 +235,30 @@ int sqfs_block_processor_enqueue(sqfs_block_processor_t *proc,
 		proc->backlog += 1;
 	}
 
-	it = proc->done;
-	prev = NULL;
-
-	while (it != NULL && it->sequence_number == proc->dequeue_id) {
-		prev = it;
-		it = it->next;
+	if (proc->done != NULL &&
+	    proc->done->sequence_number == proc->dequeue_id) {
+		block = proc->done;
+		proc->done = proc->done->next;
 		proc->dequeue_id += 1;
-	}
-
-	if (prev != NULL) {
-		queue = proc->done;
-		prev->next = NULL;
-		proc->done = it;
+	} else {
+		block = NULL;
 	}
 
 	pthread_cond_broadcast(&proc->queue_cond);
 	pthread_mutex_unlock(&proc->mtx);
 
-	return process_completed_blocks(proc, queue);
+	if (block == NULL)
+		return 0;
+
+	status = process_completed_block(proc, block);
+	if (status != 0) {
+		pthread_mutex_lock(&proc->mtx);
+		proc->status = status;
+		goto fail;
+	}
+
+	free(block);
+	return 0;
 fail:
 	pthread_mutex_unlock(&proc->mtx);
 	free(block);
@@ -287,12 +291,19 @@ int sqfs_block_processor_finish(sqfs_block_processor_t *proc)
 	proc->done = NULL;
 	pthread_mutex_unlock(&proc->mtx);
 
-	status = process_completed_blocks(proc, queue);
-	if (status != 0) {
-		pthread_mutex_lock(&proc->mtx);
-		proc->status = status;
-		pthread_cond_broadcast(&proc->queue_cond);
-		goto fail;
+	while (queue != NULL) {
+		it = queue;
+		queue = queue->next;
+		it->next = NULL;
+
+		status = process_completed_block(proc, it);
+		free(it);
+		if (status != 0) {
+			pthread_mutex_lock(&proc->mtx);
+			proc->status = status;
+			pthread_cond_broadcast(&proc->queue_cond);
+			goto fail;
+		}
 	}
 
 	return 0;
diff --git a/lib/sqfs/blk_proc/serial.c b/lib/sqfs/blk_proc/serial.c
index 8e842fb..850c33d 100644
--- a/lib/sqfs/blk_proc/serial.c
+++ b/lib/sqfs/blk_proc/serial.c
@@ -68,7 +68,9 @@ int sqfs_block_processor_enqueue(sqfs_block_processor_t *proc,
 	}
 
 	block->next = NULL;
-	return process_completed_blocks(proc, block);
+	proc->status = process_completed_block(proc, block);
+	free(block);
+	return proc->status;
 }
 
 int sqfs_block_processor_finish(sqfs_block_processor_t *proc)