Make data writer use block processor

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

8 files changed, 53 insertions, 35 deletions

diff --git a/doc/gensquashfs.1 b/doc/gensquashfs.1
index fca45cd..85c13d5 100644
--- a/doc/gensquashfs.1
+++ b/doc/gensquashfs.1
@@ -29,6 +29,11 @@ and the default selection.
 A comma seperated list of extra options for the selected compressor. Specify
 \fBhelp\fR to get a list of available options.
 .TP
+\fB\-\-num\-jobs\fR, \fB\-j\fR <count>
+If gensquashfs was compiled with a built in pthread based parallel data
+compressor, this option can be used to set the number of compressor
+threads. If not set, the default is 1.
+.TP
 \fB\-\-block\-size\fR, \fB\-b\fR <size>
 Block size to use for Squashfs image.
 Defaults to 131072.
diff --git a/doc/tar2sqfs.1 b/doc/tar2sqfs.1
index 7965a26..463125b 100644
--- a/doc/tar2sqfs.1
+++ b/doc/tar2sqfs.1
@@ -23,6 +23,11 @@ and the default selection.
 A comma seperated list of extra options for the selected compressor. Specify
 \fBhelp\fR to get a list of available options.
 .TP
+\fB\-\-num\-jobs\fR, \fB\-j\fR <count>
+If tar2sqfs was compiled with a built in pthread based parallel data
+compressor, this option can be used to set the number of compressor
+threads. If not set, the default is 1.
+.TP
 \fB\-\-block\-size\fR, \fB\-b\fR <size>
 Block size to use for Squashfs image.
 Defaults to 131072.
diff --git a/include/data_writer.h b/include/data_writer.h
index 70ae879..f3b296d 100644
--- a/include/data_writer.h
+++ b/include/data_writer.h
@@ -36,7 +36,8 @@ enum {
   Returns NULL on failure and prints errors to stderr.
  */
 data_writer_t *data_writer_create(sqfs_super_t *super, compressor_t *cmp,
-				  int outfd, size_t devblksize);
+				  int outfd, size_t devblksize,
+				  unsigned int num_jobs);
 
 void data_writer_destroy(data_writer_t *data);
 
diff --git a/lib/sqfs/data_writer.c b/lib/sqfs/data_writer.c
index 193232d..55c92ad 100644
--- a/lib/sqfs/data_writer.c
+++ b/lib/sqfs/data_writer.c
@@ -27,12 +27,11 @@ struct data_writer_t {
 	uint64_t bytes_written;
 	off_t start;
 
+	block_processor_t *proc;
+	compressor_t *cmp;
 	file_info_t *list;
 	sqfs_super_t *super;
-	compressor_t *cmp;
 	int outfd;
-
-	uint8_t scratch[];
 };
 
 enum {
@@ -87,18 +86,19 @@ static int allign_file(data_writer_t *data)
 	return 0;
 }
 
-static int handle_data_block(data_writer_t *data, block_t *blk)
+static int block_callback(void *user, block_t *blk)
 {
 	file_info_t *fi = blk->user;
+	data_writer_t *data = user;
 	uint64_t ref, offset;
 	uint32_t out;
 
 	if (blk->flags & BLK_FIRST_BLOCK) {
 		if (save_position(data))
-			goto fail;
+			return -1;
 
 		if ((blk->flags & BLK_ALLIGN) && allign_file(data) != 0)
-			goto fail;
+			return -1;
 
 		fi->startblock = data->super->bytes_used;
 	}
@@ -106,11 +106,6 @@ static int handle_data_block(data_writer_t *data, block_t *blk)
 	if (blk->size == 0)
 		goto skip_sentinel;
 
-	if (process_block(blk, data->cmp, data->scratch,
-			  data->super->block_size)) {
-		goto fail;
-	}
-
 	out = blk->size;
 	if (!(blk->flags & BLK_IS_COMPRESSED))
 		out |= 1 << 24;
@@ -130,7 +125,7 @@ static int handle_data_block(data_writer_t *data, block_t *blk)
 
 	if (write_data("writing data block", data->outfd,
 		       blk->data, blk->size)) {
-		goto fail;
+		return -1;
 	}
 
 	data->super->bytes_used += blk->size;
@@ -138,7 +133,7 @@ static int handle_data_block(data_writer_t *data, block_t *blk)
 skip_sentinel:
 	if (blk->flags & BLK_LAST_BLOCK) {
 		if ((blk->flags & BLK_ALLIGN) && allign_file(data) != 0)
-			goto fail;
+			return -1;
 
 		ref = find_equal_blocks(fi, data->list,
 					data->super->block_size);
@@ -147,15 +142,11 @@ skip_sentinel:
 			fi->flags |= FILE_FLAG_BLOCKS_ARE_DUPLICATE;
 
 			if (restore_position(data))
-				goto fail;
+				return -1;
 		}
 	}
 
-	free(blk);
 	return 0;
-fail:
-	free(blk);
-	return -1;
 }
 
 /*****************************************************************************/
@@ -268,7 +259,7 @@ static int add_sentinel_block(data_writer_t *data, file_info_t *fi,
 	blk->user = fi;
 	blk->flags = BLK_DONT_COMPRESS | BLK_DONT_CHECKSUM | flags;
 
-	return handle_data_block(data, blk);
+	return block_processor_enqueue(data->proc, blk);
 }
 
 int write_data_from_fd(data_writer_t *data, file_info_t *fi,
@@ -285,6 +276,9 @@ int write_data_from_fd(data_writer_t *data, file_info_t *fi,
 	if (flags & DW_ALLIGN_DEVBLK)
 		blk_flags |= BLK_ALLIGN;
 
+	fi->next = data->list;
+	data->list = fi;
+
 	for (; file_size > 0; file_size -= diff) {
 		if (file_size > data->super->block_size) {
 			diff = data->super->block_size;
@@ -319,7 +313,7 @@ int write_data_from_fd(data_writer_t *data, file_info_t *fi,
 			if (handle_fragment(data, blk))
 				return -1;
 		} else {
-			if (handle_data_block(data, blk))
+			if (block_processor_enqueue(data->proc, blk))
 				return -1;
 
 			blk_flags &= ~BLK_FIRST_BLOCK;
@@ -333,8 +327,6 @@ int write_data_from_fd(data_writer_t *data, file_info_t *fi,
 			return -1;
 	}
 
-	fi->next = data->list;
-	data->list = fi;
 	return 0;
 }
 
@@ -454,7 +446,7 @@ int write_data_from_fd_condensed(data_writer_t *data, file_info_t *fi,
 			if (handle_fragment(data, blk))
 				return -1;
 		} else {
-			if (handle_data_block(data, blk))
+			if (block_processor_enqueue(data->proc, blk))
 				return -1;
 
 			blk_flags &= ~BLK_FIRST_BLOCK;
@@ -468,24 +460,24 @@ int write_data_from_fd_condensed(data_writer_t *data, file_info_t *fi,
 			return -1;
 	}
 
-	fi->next = data->list;
-	data->list = fi;
 	return 0;
 }
 
 data_writer_t *data_writer_create(sqfs_super_t *super, compressor_t *cmp,
-				  int outfd, size_t devblksize)
+				  int outfd, size_t devblksize,
+				  unsigned int num_jobs)
 {
-	data_writer_t *data;
+	data_writer_t *data = calloc(1, sizeof(*data));
 
-	data = calloc(1, sizeof(*data) + super->block_size * 3);
 	if (data == NULL) {
 		perror("creating data writer");
 		return NULL;
 	}
 
-	data->super = super;
+	data->proc = block_processor_create(super->block_size, cmp, num_jobs,
+					    data, block_callback);
 	data->cmp = cmp;
+	data->super = super;
 	data->outfd = outfd;
 	data->devblksz = devblksize;
 	return data;
@@ -493,6 +485,7 @@ data_writer_t *data_writer_create(sqfs_super_t *super, compressor_t *cmp,
 
 void data_writer_destroy(data_writer_t *data)
 {
+	block_processor_destroy(data->proc);
 	free(data->fragments);
 	free(data);
 }
@@ -527,5 +520,5 @@ int data_writer_sync(data_writer_t *data)
 			return -1;
 	}
 
-	return 0;
+	return block_processor_finish(data->proc);
 }
diff --git a/mkfs/mkfs.c b/mkfs/mkfs.c
index 449fa81..b2bdec2 100644
--- a/mkfs/mkfs.c
+++ b/mkfs/mkfs.c
@@ -160,7 +160,8 @@ int main(int argc, char **argv)
 		super.bytes_used += ret;
 	}
 
-	data = data_writer_create(&super, cmp, outfd, opt.devblksz);
+	data = data_writer_create(&super, cmp, outfd,
+				  opt.devblksz, opt.num_jobs);
 	if (data == NULL)
 		goto out_cmp;
 
diff --git a/mkfs/mkfs.h b/mkfs/mkfs.h
index 16ab887..c73a2c1 100644
--- a/mkfs/mkfs.h
+++ b/mkfs/mkfs.h
@@ -36,6 +36,7 @@ typedef struct {
 	int blksz;
 	int devblksz;
 	unsigned int dirscan_flags;
+	unsigned int num_jobs;
 	bool exportable;
 	bool quiet;
 	const char *infile;
diff --git a/mkfs/options.c b/mkfs/options.c
index 99a7cee..fb50d7b 100644
--- a/mkfs/options.c
+++ b/mkfs/options.c
@@ -14,6 +14,7 @@ static struct option long_opts[] = {
 	{ "comp-extra", required_argument, NULL, 'X' },
 	{ "pack-file", required_argument, NULL, 'F' },
 	{ "pack-dir", required_argument, NULL, 'D' },
+	{ "num-jobs", required_argument, NULL, 'j' },
 	{ "keep-time", no_argument, NULL, 'k' },
 #ifdef HAVE_SYS_XATTR_H
 	{ "keep-xattr", no_argument, NULL, 'x' },
@@ -29,7 +30,7 @@ static struct option long_opts[] = {
 	{ "help", no_argument, NULL, 'h' },
 };
 
-static const char *short_opts = "F:D:X:c:b:B:d:kxoefqhV"
+static const char *short_opts = "F:D:X:c:b:B:d:j:kxoefqhV"
 #ifdef WITH_SELINUX
 "s:"
 #endif
@@ -63,6 +64,7 @@ static const char *help_string =
 "  --comp-extra, -X <options>  A comma seperated list of extra options for\n"
 "                              the selected compressor. Specify 'help' to\n"
 "                              get a list of available options.\n"
+"  --num-jobs, -j <count>      Number of compressor jobs to create.\n"
 "  --block-size, -b <size>     Block size to use for Squashfs image.\n"
 "                              Defaults to %u.\n"
 "  --dev-block-size, -B <size> Device block size to padd the image to.\n"
@@ -148,6 +150,7 @@ void process_command_line(options_t *opt, int argc, char **argv)
 	opt->compressor = compressor_get_default();
 	opt->blksz = SQFS_DEFAULT_BLOCK_SIZE;
 	opt->devblksz = SQFS_DEVBLK_SIZE;
+	opt->num_jobs = 1;
 
 	for (;;) {
 		i = getopt_long(argc, argv, short_opts, long_opts, NULL);
@@ -173,6 +176,9 @@ void process_command_line(options_t *opt, int argc, char **argv)
 		case 'b':
 			opt->blksz = strtol(optarg, NULL, 0);
 			break;
+		case 'j':
+			opt->num_jobs = strtol(optarg, NULL, 0);
+			break;
 		case 'B':
 			opt->devblksz = strtol(optarg, NULL, 0);
 			if (opt->devblksz < 1024) {
diff --git a/tar/tar2sqfs.c b/tar/tar2sqfs.c
index f583fd4..ed3aea1 100644
--- a/tar/tar2sqfs.c
+++ b/tar/tar2sqfs.c
@@ -30,6 +30,7 @@ static struct option long_opts[] = {
 	{ "block-size", required_argument, NULL, 'b' },
 	{ "dev-block-size", required_argument, NULL, 'B' },
 	{ "defaults", required_argument, NULL, 'd' },
+	{ "num-jobs", required_argument, NULL, 'j' },
 	{ "comp-extra", required_argument, NULL, 'X' },
 	{ "no-skip", no_argument, NULL, 's' },
 	{ "no-xattr", no_argument, NULL, 'x' },
@@ -41,7 +42,7 @@ static struct option long_opts[] = {
 	{ "version", no_argument, NULL, 'V' },
 };
 
-static const char *short_opts = "c:b:B:d:X:sxekfqhV";
+static const char *short_opts = "c:b:B:d:X:j:sxekfqhV";
 
 static const char *usagestr =
 "Usage: tar2sqfs [OPTIONS...] <sqfsfile>\n"
@@ -56,6 +57,7 @@ static const char *usagestr =
 "  --comp-extra, -X <options>  A comma seperated list of extra options for\n"
 "                              the selected compressor. Specify 'help' to\n"
 "                              get a list of available options.\n"
+"  --num-jobs, -j <count>      Number of compressor jobs to create.\n"
 "  --block-size, -b <size>     Block size to use for Squashfs image.\n"
 "                              Defaults to %u.\n"
 "  --dev-block-size, -B <size> Device block size to padd the image to.\n"
@@ -92,6 +94,7 @@ static int block_size = SQFS_DEFAULT_BLOCK_SIZE;
 static size_t devblksize = SQFS_DEVBLK_SIZE;
 static bool quiet = false;
 static int outmode = O_WRONLY | O_CREAT | O_EXCL;
+static unsigned int num_jobs = 1;
 static E_SQFS_COMPRESSOR comp_id;
 static char *comp_extra = NULL;
 static char *fs_defaults = NULL;
@@ -139,6 +142,9 @@ static void process_args(int argc, char **argv)
 				exit(EXIT_FAILURE);
 			}
 			break;
+		case 'j':
+			num_jobs = strtol(optarg, NULL, 0);
+			break;
 		case 'X':
 			comp_extra = optarg;
 			break;
@@ -387,7 +393,7 @@ int main(int argc, char **argv)
 		super.bytes_used += ret;
 	}
 
-	data = data_writer_create(&super, cmp, outfd, devblksize);
+	data = data_writer_create(&super, cmp, outfd, devblksize, num_jobs);
 	if (data == NULL)
 		goto out_cmp;