Cleanup block processor error handling

If an error happpens, immediately stop everything and report only the
first error that happened.

This also fixes a memory leak in the error paths of the serial and
pthread based block processors.

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

4 files changed, 81 insertions, 25 deletions

diff --git a/lib/sqfs/blk_proc/internal.h b/lib/sqfs/blk_proc/internal.h
index b5af751..8cb1a23 100644
--- a/lib/sqfs/blk_proc/internal.h
+++ b/lib/sqfs/blk_proc/internal.h
@@ -59,13 +59,13 @@ struct sqfs_block_processor_t {
 	sqfs_block_t *done;
 	bool terminate;
 	size_t backlog;
+	int status;
 
 	/* used by main thread only */
 	uint32_t enqueue_id;
 	uint32_t dequeue_id;
 
 	unsigned int num_workers;
-	int status;
 	size_t max_backlog;
 
 	size_t devblksz;
diff --git a/lib/sqfs/blk_proc/process_block.c b/lib/sqfs/blk_proc/process_block.c
index 5d1fa58..643d964 100644
--- a/lib/sqfs/blk_proc/process_block.c
+++ b/lib/sqfs/blk_proc/process_block.c
@@ -20,11 +20,8 @@ int sqfs_block_process(sqfs_block_t *block, sqfs_compressor_t *cmp,
 	if (!(block->flags & SQFS_BLK_DONT_COMPRESS)) {
 		ret = cmp->do_block(cmp, block->data, block->size,
 				    scratch, scratch_size);
-
-		if (ret < 0) {
-			block->flags |= SQFS_BLK_COMPRESS_ERROR;
+		if (ret < 0)
 			return ret;
-		}
 
 		if (ret > 0) {
 			memcpy(block->data, scratch, ret);
@@ -188,19 +185,24 @@ static int handle_block(sqfs_block_processor_t *proc, sqfs_block_t *blk)
 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;
 
-		if (it->flags & SQFS_BLK_COMPRESS_ERROR) {
-			proc->status = SQFS_ERROR_COMRPESSOR;
-		} else if (proc->status == 0) {
-			proc->status = handle_block(proc, it);
-		}
-
+		err = handle_block(proc, it);
 		free(it);
+
+		if (err) {
+			while (queue != NULL) {
+				it = queue;
+				queue = it->next;
+				free(it);
+			}
+			return err;
+		}
 	}
 
-	return proc->status;
+	return 0;
 }
diff --git a/lib/sqfs/blk_proc/pthread.c b/lib/sqfs/blk_proc/pthread.c
index f1874c0..fb83bbe 100644
--- a/lib/sqfs/blk_proc/pthread.c
+++ b/lib/sqfs/blk_proc/pthread.c
@@ -33,21 +33,26 @@ static void *worker_proc(void *arg)
 	compress_worker_t *worker = arg;
 	sqfs_block_processor_t *shared = worker->shared;
 	sqfs_block_t *blk = NULL;
+	int status = 0;
 
 	for (;;) {
 		pthread_mutex_lock(&shared->mtx);
 		if (blk != NULL) {
 			store_completed_block(shared, blk);
 			shared->backlog -= 1;
+
+			if (status != 0 && shared->status == 0)
+				shared->status = status;
 			pthread_cond_broadcast(&shared->done_cond);
 		}
 
-		while (shared->queue == NULL && !shared->terminate) {
+		while (shared->queue == NULL && !shared->terminate &&
+		       shared->status == 0) {
 			pthread_cond_wait(&shared->queue_cond,
 					  &shared->mtx);
 		}
 
-		if (shared->terminate) {
+		if (shared->terminate || shared->status != 0) {
 			pthread_mutex_unlock(&shared->mtx);
 			break;
 		}
@@ -60,8 +65,8 @@ static void *worker_proc(void *arg)
 			shared->queue_last = NULL;
 		pthread_mutex_unlock(&shared->mtx);
 
-		sqfs_block_process(blk, worker->cmp, worker->scratch,
-				   shared->max_block_size);
+		status = sqfs_block_process(blk, worker->cmp, worker->scratch,
+					    shared->max_block_size);
 	}
 	return NULL;
 }
@@ -148,6 +153,8 @@ fail_init:
 	pthread_cond_destroy(&proc->done_cond);
 	pthread_cond_destroy(&proc->queue_cond);
 	pthread_mutex_destroy(&proc->mtx);
+	free(proc->fragments);
+	free(proc->blocks);
 	free(proc);
 	return NULL;
 }
@@ -185,6 +192,8 @@ void sqfs_block_processor_destroy(sqfs_block_processor_t *proc)
 		free(blk);
 	}
 
+	free(proc->fragments);
+	free(proc->blocks);
 	free(proc);
 }
 
@@ -192,14 +201,24 @@ int sqfs_block_processor_enqueue(sqfs_block_processor_t *proc,
 				 sqfs_block_t *block)
 {
 	sqfs_block_t *queue = NULL, *it, *prev;
+	int status;
 
-	if (block->flags & ~SQFS_BLK_USER_SETTABLE_FLAGS)
-		return SQFS_ERROR_UNSUPPORTED;
+	pthread_mutex_lock(&proc->mtx);
+	if (proc->status != 0) {
+		status = proc->status;
+		goto fail;
+	}
+
+	if (block->flags & ~SQFS_BLK_USER_SETTABLE_FLAGS) {
+		status = SQFS_ERROR_UNSUPPORTED;
+		proc->status = status;
+		pthread_cond_broadcast(&proc->queue_cond);
+		goto fail;
+	}
 
 	block->sequence_number = proc->enqueue_id++;
 	block->next = NULL;
 
-	pthread_mutex_lock(&proc->mtx);
 	if (block->size == 0) {
 		block->checksum = 0;
 		store_completed_block(proc, block);
@@ -236,22 +255,31 @@ int sqfs_block_processor_enqueue(sqfs_block_processor_t *proc,
 	pthread_mutex_unlock(&proc->mtx);
 
 	return process_completed_blocks(proc, queue);
+fail:
+	pthread_mutex_unlock(&proc->mtx);
+	free(block);
+	return status;
 }
 
 int sqfs_block_processor_finish(sqfs_block_processor_t *proc)
 {
 	sqfs_block_t *queue, *it;
+	int status;
 
 	pthread_mutex_lock(&proc->mtx);
-	while (proc->backlog > 0)
+	while (proc->backlog > 0 && proc->status == 0)
 		pthread_cond_wait(&proc->done_cond, &proc->mtx);
 
+	if (proc->status != 0) {
+		status = proc->status;
+		goto fail;
+	}
+
 	for (it = proc->done; it != NULL; it = it->next) {
 		if (it->sequence_number != proc->dequeue_id++) {
-			pthread_mutex_unlock(&proc->mtx);
-
-			/* XXX: this would actually be a BUG */
-			return SQFS_ERROR_INTERNAL;
+			status = SQFS_ERROR_INTERNAL;
+			proc->status = status;
+			goto fail;
 		}
 	}
 
@@ -259,5 +287,28 @@ int sqfs_block_processor_finish(sqfs_block_processor_t *proc)
 	proc->done = NULL;
 	pthread_mutex_unlock(&proc->mtx);
 
-	return process_completed_blocks(proc, queue);
+	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;
+	}
+
+	return 0;
+fail:
+	while (proc->queue != NULL) {
+		it = proc->queue;
+		proc->queue = it->next;
+		free(it);
+	}
+
+	while (proc->done != NULL) {
+		it = proc->done;
+		proc->done = it->next;
+		free(it);
+	}
+
+	pthread_mutex_unlock(&proc->mtx);
+	return status;
 }
diff --git a/lib/sqfs/blk_proc/serial.c b/lib/sqfs/blk_proc/serial.c
index b7bc545..8e842fb 100644
--- a/lib/sqfs/blk_proc/serial.c
+++ b/lib/sqfs/blk_proc/serial.c
@@ -40,6 +40,8 @@ sqfs_block_processor_t *sqfs_block_processor_create(size_t max_block_size,
 
 void sqfs_block_processor_destroy(sqfs_block_processor_t *proc)
 {
+	free(proc->fragments);
+	free(proc->blocks);
 	free(proc);
 }
 
@@ -53,6 +55,7 @@ int sqfs_block_processor_enqueue(sqfs_block_processor_t *proc,
 
 	if (block->flags & ~SQFS_BLK_USER_SETTABLE_FLAGS) {
 		proc->status = SQFS_ERROR_UNSUPPORTED;
+		free(block);
 		return proc->status;
 	}