X-Git-Url: https://wimlib.net/git/?p=wimlib;a=blobdiff_plain;f=src%2Fcompress_parallel.c;h=511ef1b84ee73451048c21020cf9af7845f4a8b9;hp=acf4ea58afc24b4299f7848d22301486c6be5bbe;hb=80b48ed7aeb8e6c9460ab254920818f9e36d656e;hpb=f18b7fc3361c4daac0ddd104af65a8eff8466fec

diff --git a/src/compress_parallel.c b/src/compress_parallel.c
index acf4ea58..511ef1b8 100644
--- a/src/compress_parallel.c
+++ b/src/compress_parallel.c
@@ -28,23 +28,15 @@
 #ifdef ENABLE_MULTITHREADED_COMPRESSION
 
 #include <errno.h>
-#include <limits.h>
 #include <pthread.h>
 #include <stdlib.h>
 #include <string.h>
-#include <unistd.h>
-#ifdef HAVE_SYS_SYSCTL_H
-#  include <sys/sysctl.h>
-#endif
 
 #include "wimlib/assert.h"
 #include "wimlib/chunk_compressor.h"
 #include "wimlib/error.h"
 #include "wimlib/list.h"
 #include "wimlib/util.h"
-#ifdef __WIN32__
-#  include "wimlib/win32.h" /* win32_get_number_of_processors() */
-#endif
 
 struct message_queue {
 	struct list_head list;
@@ -94,49 +86,7 @@ struct parallel_chunk_compressor {
 	size_t next_chunk_idx;
 };
 
-static unsigned
-get_default_num_threads(void)
-{
-	long n;
-#ifdef __WIN32__
-	n = win32_get_number_of_processors();
-#else
-	n = sysconf(_SC_NPROCESSORS_ONLN);
-#endif
-	if (n < 1 || n >= UINT_MAX) {
-		WARNING("Failed to determine number of processors; assuming 1.");
-		return 1;
-	}
-	return n;
-}
 
-static u64
-get_avail_memory(void)
-{
-#ifdef __WIN32__
-	u64 phys_bytes = win32_get_avail_memory();
-	if (phys_bytes == 0)
-		goto default_size;
-	return phys_bytes;
-#elif defined(_SC_PAGESIZE) && defined(_SC_PHYS_PAGES)
-	long page_size = sysconf(_SC_PAGESIZE);
-	long num_pages = sysconf(_SC_PHYS_PAGES);
-	if (page_size <= 0 || num_pages <= 0)
-		goto default_size;
-	return ((u64)page_size * (u64)num_pages);
-#else
-	int mib[2] = {CTL_HW, HW_MEMSIZE};
-	u64 memsize;
-	size_t len = sizeof(memsize);
-	if (sysctl(mib, ARRAY_LEN(mib), &memsize, &len, NULL, 0) < 0 || len != 8)
-		goto default_size;
-	return memsize;
-#endif
-
-default_size:
-	WARNING("Failed to determine available memory; assuming 1 GiB");
-	return 1ULL << 30;
-}
 
 static int
 message_queue_init(struct message_queue *q)
@@ -329,21 +279,17 @@ submit_compression_msg(struct parallel_chunk_compressor *ctx)
 	ctx->next_submit_msg = NULL;
 }
 
-static bool
-parallel_chunk_compressor_submit_chunk(struct chunk_compressor *_ctx,
-				       const void *chunk, u32 size)
+static void *
+parallel_chunk_compressor_get_chunk_buffer(struct chunk_compressor *_ctx)
 {
 	struct parallel_chunk_compressor *ctx = (struct parallel_chunk_compressor *)_ctx;
 	struct message *msg;
 
-	wimlib_assert(size > 0);
-	wimlib_assert(size <= ctx->base.out_chunk_size);
-
 	if (ctx->next_submit_msg) {
 		msg = ctx->next_submit_msg;
 	} else {
 		if (list_empty(&ctx->available_msgs))
-			return false;
+			return NULL;
 
 		msg = list_entry(ctx->available_msgs.next, struct message, list);
 		list_del(&msg->list);
@@ -351,22 +297,33 @@ parallel_chunk_compressor_submit_chunk(struct chunk_compressor *_ctx,
 		msg->num_filled_chunks = 0;
 	}
 
-	memcpy(msg->uncompressed_chunks[msg->num_filled_chunks], chunk, size);
-	msg->uncompressed_chunk_sizes[msg->num_filled_chunks] = size;
+	return msg->uncompressed_chunks[msg->num_filled_chunks];
+}
+
+static void
+parallel_chunk_compressor_signal_chunk_filled(struct chunk_compressor *_ctx, u32 usize)
+{
+	struct parallel_chunk_compressor *ctx = (struct parallel_chunk_compressor *)_ctx;
+	struct message *msg;
+
+	wimlib_assert(usize > 0);
+	wimlib_assert(usize <= ctx->base.out_chunk_size);
+	wimlib_assert(ctx->next_submit_msg);
+
+	msg = ctx->next_submit_msg;
+	msg->uncompressed_chunk_sizes[msg->num_filled_chunks] = usize;
 	if (++msg->num_filled_chunks == msg->num_alloc_chunks)
 		submit_compression_msg(ctx);
-	return true;
 }
 
 static bool
-parallel_chunk_compressor_get_chunk(struct chunk_compressor *_ctx,
-				    const void **cdata_ret, u32 *csize_ret,
-				    u32 *usize_ret)
+parallel_chunk_compressor_get_compression_result(struct chunk_compressor *_ctx,
+						 const void **cdata_ret, u32 *csize_ret,
+						 u32 *usize_ret)
 {
 	struct parallel_chunk_compressor *ctx = (struct parallel_chunk_compressor *)_ctx;
 	struct message *msg;
 
-
 	if (ctx->next_submit_msg)
 		submit_compression_msg(ctx);
 
@@ -418,7 +375,7 @@ new_parallel_chunk_compressor(int out_ctype, u32 out_chunk_size,
 	wimlib_assert(out_chunk_size > 0);
 
 	if (num_threads == 0)
-		num_threads = get_default_num_threads();
+		num_threads = get_available_cpus();
 
 	if (num_threads == 1) {
 		DEBUG("Only 1 thread; Not bothering with "
@@ -427,7 +384,7 @@ new_parallel_chunk_compressor(int out_ctype, u32 out_chunk_size,
 	}
 
 	if (max_memory == 0)
-		max_memory = get_avail_memory();
+		max_memory = get_available_memory();
 
 	desired_num_threads = num_threads;
 
@@ -490,8 +447,9 @@ new_parallel_chunk_compressor(int out_ctype, u32 out_chunk_size,
 	ctx->base.out_ctype = out_ctype;
 	ctx->base.out_chunk_size = out_chunk_size;
 	ctx->base.destroy = parallel_chunk_compressor_destroy;
-	ctx->base.submit_chunk = parallel_chunk_compressor_submit_chunk;
-	ctx->base.get_chunk = parallel_chunk_compressor_get_chunk;
+	ctx->base.get_chunk_buffer = parallel_chunk_compressor_get_chunk_buffer;
+	ctx->base.signal_chunk_filled = parallel_chunk_compressor_signal_chunk_filled;
+	ctx->base.get_compression_result = parallel_chunk_compressor_get_compression_result;
 
 	ctx->num_thread_data = num_threads;
 
@@ -515,7 +473,8 @@ new_parallel_chunk_compressor(int out_ctype, u32 out_chunk_size,
 
 		dat->chunks_to_compress_queue = &ctx->chunks_to_compress_queue;
 		dat->compressed_chunks_queue = &ctx->compressed_chunks_queue;
-		ret = wimlib_create_compressor(out_ctype, out_chunk_size, 0,
+		ret = wimlib_create_compressor(out_ctype, out_chunk_size,
+					       WIMLIB_COMPRESSOR_FLAG_DESTRUCTIVE,
 					       &dat->compressor);
 		if (ret)
 			goto err;