4 files changed, 41 insertions, 67 deletions

diff --git a/lib/sqfs/block_processor/frontend.c b/lib/sqfs/block_processor/frontend.c
index 9e9ecc7..daf12e7 100644
--- a/lib/sqfs/block_processor/frontend.c
+++ b/lib/sqfs/block_processor/frontend.c
@@ -197,6 +197,33 @@ int sqfs_block_processor_submit_block(sqfs_block_processor_t *proc, void *user,
 	return proc->append_to_work_queue(proc, blk);
 }
 
+int sqfs_block_processor_sync(sqfs_block_processor_t *proc)
+{
+	return proc->sync(proc);
+}
+
+int sqfs_block_processor_finish(sqfs_block_processor_t *proc)
+{
+	sqfs_block_t *blk;
+	int status;
+
+	status = proc->sync(proc);
+
+	if (status == 0 && proc->frag_block != NULL) {
+		blk = proc->frag_block;
+		blk->next = NULL;
+		blk->flags |= BLK_FLAG_MANUAL_SUBMISSION;
+		proc->frag_block = NULL;
+
+		status = proc->append_to_work_queue(proc, blk);
+
+		if (status == 0)
+			status = proc->sync(proc);
+	}
+
+	return status;
+}
+
 const sqfs_block_processor_stats_t
 *sqfs_block_processor_get_stats(const sqfs_block_processor_t *proc)
 {
diff --git a/lib/sqfs/block_processor/internal.h b/lib/sqfs/block_processor/internal.h
index 0c7e4e6..496d415 100644
--- a/lib/sqfs/block_processor/internal.h
+++ b/lib/sqfs/block_processor/internal.h
@@ -87,6 +87,8 @@ struct sqfs_block_processor_t {
 
 	int (*append_to_work_queue)(sqfs_block_processor_t *proc,
 				    sqfs_block_t *block);
+
+	int (*sync)(sqfs_block_processor_t *proc);
 };
 
 SQFS_INTERNAL int block_processor_init(sqfs_block_processor_t *base,
diff --git a/lib/sqfs/block_processor/serial.c b/lib/sqfs/block_processor/serial.c
index 9768a7f..7b3825d 100644
--- a/lib/sqfs/block_processor/serial.c
+++ b/lib/sqfs/block_processor/serial.c
@@ -75,6 +75,11 @@ fail:
 	return sproc->status;
 }
 
+static int block_processor_sync(sqfs_block_processor_t *proc)
+{
+	return ((serial_block_processor_t *)proc)->status;
+}
+
 sqfs_block_processor_t *sqfs_block_processor_create(size_t max_block_size,
 						    sqfs_compressor_t *cmp,
 						    unsigned int num_workers,
@@ -94,37 +99,8 @@ sqfs_block_processor_t *sqfs_block_processor_create(size_t max_block_size,
 		return NULL;
 	}
 
+	proc->base.sync = block_processor_sync;
 	proc->base.append_to_work_queue = append_to_work_queue;
 	((sqfs_object_t *)proc)->destroy = block_processor_destroy;
 	return (sqfs_block_processor_t *)proc;
 }
-
-int sqfs_block_processor_sync(sqfs_block_processor_t *proc)
-{
-	return ((serial_block_processor_t *)proc)->status;
-}
-
-int sqfs_block_processor_finish(sqfs_block_processor_t *proc)
-{
-	serial_block_processor_t *sproc = (serial_block_processor_t *)proc;
-
-	if (proc->frag_block == NULL || sproc->status != 0)
-		goto fail;
-
-	sproc->status = proc->process_block(proc->frag_block, proc->cmp,
-					    sproc->scratch,
-					    proc->max_block_size);
-	if (sproc->status != 0)
-		goto fail;
-
-	sproc->status = proc->process_completed_block(proc, proc->frag_block);
-	proc->frag_block = NULL;
-	return sproc->status;
-fail:
-	if (proc->frag_block != NULL) {
-		free_block_list(proc->frag_block->frag_list);
-		free(proc->frag_block);
-		proc->frag_block = NULL;
-	}
-	return sproc->status;
-}
diff --git a/lib/sqfs/block_processor/winpthread.c b/lib/sqfs/block_processor/winpthread.c
index afd66c1..e39f795 100644
--- a/lib/sqfs/block_processor/winpthread.c
+++ b/lib/sqfs/block_processor/winpthread.c
@@ -363,6 +363,11 @@ static int append_to_work_queue(sqfs_block_processor_t *proc,
 	return status;
 }
 
+static int block_processor_sync(sqfs_block_processor_t *proc)
+{
+	return append_to_work_queue(proc, NULL);
+}
+
 static thread_pool_processor_t *block_processor_create(size_t max_block_size,
 						       sqfs_compressor_t *cmp,
 						       unsigned int num_workers,
@@ -386,6 +391,7 @@ static thread_pool_processor_t *block_processor_create(size_t max_block_size,
 		return NULL;
 	}
 
+	proc->base.sync = block_processor_sync;
 	proc->base.append_to_work_queue = append_to_work_queue;
 	proc->num_workers = num_workers;
 	proc->max_backlog = max_backlog;
@@ -411,43 +417,6 @@ fail:
 	return NULL;
 }
 
-int sqfs_block_processor_sync(sqfs_block_processor_t *proc)
-{
-	return append_to_work_queue(proc, NULL);
-}
-
-int sqfs_block_processor_finish(sqfs_block_processor_t *proc)
-{
-	thread_pool_processor_t *thproc = (thread_pool_processor_t *)proc;
-	sqfs_block_t *blk;
-	int status;
-
-	status = sqfs_block_processor_sync(proc);
-
-	if (status == 0 && proc->frag_block != NULL) {
-		blk = proc->frag_block;
-		blk->next = NULL;
-		proc->frag_block = NULL;
-
-		status = proc->process_block(blk, proc->cmp,
-					     thproc->workers[0]->scratch,
-					     proc->max_block_size);
-
-		if (status == 0)
-			status = handle_io_queue(thproc, blk);
-
-		if (status != 0) {
-			LOCK(&thproc->mtx);
-			if (thproc->status == 0)
-				thproc->status = status;
-			SIGNAL_ALL(&thproc->queue_cond);
-			UNLOCK(&thproc->mtx);
-		}
-	}
-
-	return status;
-}
-
 #if defined(_WIN32) || defined(__WINDOWS__)
 sqfs_block_processor_t *sqfs_block_processor_create(size_t max_block_size,
 						    sqfs_compressor_t *cmp,