#ifdef ENABLE_MULTITHREADED_COMPRESSION
+#include <errno.h>
+#include <pthread.h>
+#include <stdlib.h>
+#include <string.h>
+
#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
-
-#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
struct message_queue {
struct list_head list;
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)
return;
if (ctx->num_started_threads != 0) {
- DEBUG("Terminating %u compressor threads", ctx->num_started_threads);
message_queue_terminate(&ctx->chunks_to_compress_queue);
for (i = 0; i < ctx->num_started_threads; i++)
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);
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);
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 "
- "parallel chunk compressor.");
+ if (num_threads == 1)
return -1;
- }
if (max_memory == 0)
- max_memory = get_avail_memory();
+ max_memory = get_available_memory();
desired_num_threads = num_threads;
desired_num_threads, num_threads);
}
- if (num_threads == 1) {
- DEBUG("Only 1 thread; Not bothering with "
- "parallel chunk compressor.");
+ if (num_threads == 1)
return -2;
- }
ret = WIMLIB_ERR_NOMEM;
ctx = CALLOC(1, sizeof(*ctx));
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;
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;
ctx->num_started_threads < num_threads;
ctx->num_started_threads++)
{
- DEBUG("pthread_create thread %u of %u",
- ctx->num_started_threads + 1, num_threads);
ret = pthread_create(&ctx->thread_data[ctx->num_started_threads].thread,
NULL,
compressor_thread_proc,
git://wimlib.net / wimlib / blobdiff