summaryrefslogtreecommitdiffstats
path: root/plzip.cc
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--plzip.cc684
1 files changed, 272 insertions, 412 deletions
diff --git a/plzip.cc b/plzip.cc
index f110437..7ac8144 100644
--- a/plzip.cc
+++ b/plzip.cc
@@ -1,6 +1,6 @@
/* Plzip - A parallel version of the lzip data compressor
Copyright (C) 2009 Laszlo Ersek.
- Copyright (C) 2009 Antonio Diaz Diaz.
+ Copyright (C) 2009, 2010 Antonio Diaz Diaz.
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
@@ -18,27 +18,74 @@
#define _FILE_OFFSET_BITS 64
+#include <algorithm>
#include <cassert>
#include <cerrno>
#include <climits>
#include <csignal>
#include <cstdio>
#include <cstdlib>
+#include <vector>
#include <stdint.h>
#include <unistd.h>
#include <lzlib.h>
#include "main.h"
#include "plzip.h"
-#include "lacos_rbtree.h"
+
+#ifndef LLONG_MAX
+#define LLONG_MAX 0x7FFFFFFFFFFFFFFFLL
+#endif
+#ifndef LLONG_MIN
+#define LLONG_MIN (-LLONG_MAX - 1LL)
+#endif
+#ifndef ULLONG_MAX
+#define ULLONG_MAX 0xFFFFFFFFFFFFFFFFULL
+#endif
+
+
+namespace {
+
+long long in_size = 0;
+long long out_size = 0;
+
+void *(*mallocf)(size_t size);
+void (*freef)(void *ptr);
+
+
+void * trace_malloc(size_t size)
+{
+ int save_errno = 0;
+
+ void * ret = malloc(size);
+ if( ret == 0 ) save_errno = errno;
+ fprintf(stderr, "malloc(%lu) == %p\n", (long unsigned)size, ret);
+ if( ret == 0 ) errno = save_errno;
+ return ret;
+}
+
+
+void trace_free(void *ptr)
+{
+ fprintf(stderr, "free(%p)\n", ptr);
+ free(ptr);
+}
+
+
+void * xalloc(size_t size)
+{
+ void *ret = (*mallocf)(size);
+ if( 0 == ret ) fail("(*mallocf)()", errno);
+ return ret;
+}
struct S2w_blk /* Splitter to workers. */
{
- uint64_t id; /* Block serial number as read from infd. */
+ unsigned long long id; /* Block serial number as read from infd. */
S2w_blk *next; /* Next in queue. */
size_t loaded; /* # of bytes in plain, may be 0 for 1st. */
- char unsigned plain[1]; /* Data read from infd, allocated: sizeof_plain. */
+ unsigned char plain[1]; /* Data read from infd, allocated: sizeof_plain. */
};
@@ -51,7 +98,7 @@ struct S2w_q
};
-static void
+void
s2w_q_init(S2w_q *s2w_q)
{
xinit(&s2w_q->av_or_eof);
@@ -61,7 +108,7 @@ s2w_q_init(S2w_q *s2w_q)
}
-static void
+void
s2w_q_uninit(S2w_q *s2w_q)
{
assert(0 != s2w_q->eof);
@@ -71,251 +118,169 @@ s2w_q_uninit(S2w_q *s2w_q)
}
-struct W2m_blk /* Workers to muxer. */
+struct W2m_blk /* Workers to muxer data block. */
{
- uint64_t id; /* Block index as read from infd. */
+ unsigned long long id; /* Block index as read from infd. */
W2m_blk *next; /* Next block in list (unordered). */
- size_t produced; /* Number of bytes in compr. */
- char unsigned compr[1]; /* Data to write to outfd, alloc.: sizeof_compr. */
+ int produced; /* Number of bytes in compr. */
+ unsigned char compr[1]; /* Data to write to outfd, alloc.: sizeof_compr. */
};
-static int
-w2m_blk_cmp(const void *v_a, const void *v_b)
-{
- uint64_t a,
- b;
-
- a = ((const W2m_blk *)v_a)->id;
- b = ((const W2m_blk *)v_b)->id;
-
- return
- a < b ? -1
- : a > b ? 1
- : 0;
-}
-
-
struct W2m_q
{
Cond av_or_exit; /* New block available or all workers exited. */
- uint64_t needed; /* Block needed for resuming writing. */
+ unsigned long long needed; /* Block needed for resuming writing. */
W2m_blk *head; /* Block list (unordered). */
unsigned working; /* Number of workers still running. */
};
-static void
-w2m_q_init(W2m_q *w2m_q, unsigned num_worker)
+void
+w2m_q_init(W2m_q *w2m_q, int num_workers)
{
- assert(0u < num_worker);
+ assert(0 < num_workers);
xinit(&w2m_q->av_or_exit);
- w2m_q->needed = 0u;
+ w2m_q->needed = 0;
w2m_q->head = 0;
- w2m_q->working = num_worker;
+ w2m_q->working = num_workers;
}
-static void
+void
w2m_q_uninit(W2m_q *w2m_q)
{
- assert(0u == w2m_q->working);
+ assert(0 == w2m_q->working);
assert(0 == w2m_q->head);
xdestroy(&w2m_q->av_or_exit);
}
-struct M2s_q /* Muxer to splitter. */
-{
- Cond av; /* Free slot available. */
- unsigned num_free; /* Number of free slots. */
-};
-
-
-static void
-m2s_q_init(M2s_q *m2s_q, unsigned num_free)
-{
- assert(0u < num_free);
- xinit(&m2s_q->av);
- m2s_q->num_free = num_free;
-}
+struct M2s_q // Muxer to splitter queue
+ {
+ Cond av; // Free slot available
+ int num_free; // Number of free slots
+ M2s_q( const int slots )
+ {
+ xinit(&av);
+ num_free = slots;
+ }
-static void
-m2s_q_uninit(M2s_q *m2s_q, unsigned num_free)
-{
- assert(m2s_q->num_free == num_free);
- xdestroy(&m2s_q->av);
-}
+ ~M2s_q() { xdestroy(&av); }
+ };
-static void
-split(M2s_q *m2s_q, S2w_q *s2w_q, int infd,
- const size_t sizeof_plain, const size_t sizeof_s2w_blk)
-{
- uint64_t id;
- ssize_t rd;
+struct Splitter_arg
+ {
+ M2s_q *m2s_q;
+ S2w_q *s2w_q;
+ int infd;
+ int sizeof_plain;
+ size_t sizeof_S2w_blk;
+ };
- id = 0u;
- do {
- S2w_blk *s2w_blk;
- size_t vacant;
+void * splitter( void * arg )
+ {
+ const Splitter_arg & tmp = *(Splitter_arg *)arg;
+ M2s_q *m2s_q = tmp.m2s_q;
+ S2w_q *s2w_q = tmp.s2w_q;
+ const int infd = tmp.infd;
+ const int sizeof_plain = tmp.sizeof_plain;
+ const size_t sizeof_s2w_blk = tmp.sizeof_S2w_blk;
+
+ for( unsigned long long id = 0; ; ++id )
+ {
/* Grab a free slot. */
xlock_pred(&m2s_q->av);
- while (0u == m2s_q->num_free) {
- xwait(&m2s_q->av);
- }
+ while( m2s_q->num_free == 0 ) xwait(&m2s_q->av);
--m2s_q->num_free;
xunlock(&m2s_q->av);
- s2w_blk = (S2w_blk *)xalloc(sizeof_s2w_blk);
+ S2w_blk * s2w_blk = (S2w_blk *)xalloc(sizeof_s2w_blk);
/* Fill block. */
- vacant = sizeof_plain;
- do {
- rd = read(infd, s2w_blk->plain + (sizeof_plain - vacant),
- vacant > (size_t)SSIZE_MAX ? (size_t)SSIZE_MAX : vacant);
- } while (0 < rd && 0u < (vacant -= (size_t)rd));
+ const int rd = readblock( infd, (char *)s2w_blk->plain, sizeof_plain );
+ if( rd != sizeof_plain && errno ) fail("read()", errno);
- /* Read error. */
- if (-1 == rd) {
- fail("read()", errno);
- }
-
- if (sizeof_plain == vacant && 0u < id) {
+ if( rd == 0 && id != 0 )
+ {
/* EOF on first read, but not for first input block. */
(*freef)(s2w_blk);
xlock(&m2s_q->av);
++m2s_q->num_free;
xunlock(&m2s_q->av);
- }
- else {
+ }
+ else
+ {
s2w_blk->id = id;
s2w_blk->next = 0;
- s2w_blk->loaded = sizeof_plain - vacant;
- }
-
- /* We either push a block, or set EOF, or both. */
- assert(sizeof_plain > vacant || 0 == rd);
+ s2w_blk->loaded = rd;
+ in_size += rd;
+ }
xlock(&s2w_q->av_or_eof);
- if (0 == s2w_q->head) {
- xbroadcast(&s2w_q->av_or_eof);
- }
+ if( s2w_q->head == 0 ) xbroadcast(&s2w_q->av_or_eof);
- if (sizeof_plain > vacant || 0u == id) {
- if (0 == s2w_q->tail) {
- s2w_q->head = s2w_blk;
- }
- else {
- s2w_q->tail->next = s2w_blk;
- }
+ if( rd > 0 || id == 0 )
+ {
+ if( s2w_q->tail == 0 ) s2w_q->head = s2w_blk;
+ else s2w_q->tail->next = s2w_blk;
s2w_q->tail = s2w_blk;
- }
- s2w_q->eof = (0 == rd);
+ }
+ s2w_q->eof = ( rd == 0 );
xunlock(&s2w_q->av_or_eof);
- /*
- If we didn't push a block, then this is bogus, but then we did set EOF,
- so it doesn't matter, because we'll leave immediately.
- */
- ++id;
- } while (0 < rd);
-}
-
-
-struct Split_arg
-{
- M2s_q *m2s_q;
- S2w_q *s2w_q;
- int infd;
- size_t sizeof_plain,
- sizeof_S2w_blk;
-};
-
-
-static void *
-split_wrap(void *v_split_arg)
-{
- Split_arg *split_arg = (Split_arg *)v_split_arg;
-
- split(
- split_arg->m2s_q,
- split_arg->s2w_q,
- split_arg->infd,
- split_arg->sizeof_plain,
- split_arg->sizeof_S2w_blk
- );
+ if( rd <= 0 ) break;
+ }
return 0;
-}
+ }
-static void
-work_lz_rd(W2m_blk *w2m_blk, const size_t sizeof_compr, void *lz)
+void work_lz_rd(W2m_blk *w2m_blk, const int sizeof_compr, LZ_Encoder * lz)
{
int rd;
assert(w2m_blk->produced < sizeof_compr);
rd = LZ_compress_read(lz, w2m_blk->compr + w2m_blk->produced,
sizeof_compr - w2m_blk->produced);
- if (-1 == rd) {
+ if( -1 == rd ) {
show_error( "LZ_compress_read() failed." );
fatal();
}
- w2m_blk->produced += (size_t)rd;
+ w2m_blk->produced += rd;
}
-struct Compr_lev
-{
- unsigned dict_size,
- mx_match;
-};
-
-
-static const Compr_lev compr_lev[] = {
- { 1u * 1024u * 1024u, 10u },
- { 1u * 1024u * 1024u, 12u },
- { 1u * 1024u * 1024u, 17u },
- { 2u * 1024u * 1024u, 26u },
- { 4u * 1024u * 1024u, 44u },
- { 8u * 1024u * 1024u, 80u },
- { 16u * 1024u * 1024u, 108u },
- { 16u * 1024u * 1024u, 163u },
- { 32u * 1024u * 1024u, 273u }
-};
-
-
-static void
-work_compr(S2w_blk *s2w_blk, W2m_q *w2m_q, unsigned clidx,
- const size_t sizeof_compr, const size_t sizeof_w2m_blk)
+void work_compr( const int dictionary_size, const int match_len_limit,
+ S2w_blk *s2w_blk, W2m_q *w2m_q,
+ const int sizeof_compr, const size_t sizeof_w2m_blk )
{
W2m_blk *w2m_blk;
- assert(0u < s2w_blk->loaded || 0u == s2w_blk->id);
+ assert(0 < s2w_blk->loaded || 0 == s2w_blk->id);
w2m_blk = (W2m_blk *)xalloc(sizeof_w2m_blk);
/* Single member compression. Settings like with lzip -6. */
{
- void *lz;
+ LZ_Encoder * lz;
size_t written;
- lz = LZ_compress_open(compr_lev[clidx].dict_size,
- compr_lev[clidx].mx_match, (uint64_t)-1 / 2u);
- if (LZ_ok != LZ_compress_errno(lz)) {
+ lz = LZ_compress_open( dictionary_size, match_len_limit, LLONG_MAX );
+ if( LZ_ok != LZ_compress_errno(lz) ) {
show_error( "LZ_compress_open() failed." );
fatal();
}
- written = 0u;
- w2m_blk->produced = 0u;
- while (written < s2w_blk->loaded) {
+ written = 0;
+ w2m_blk->produced = 0;
+ while( written < s2w_blk->loaded ) {
int wr;
wr = LZ_compress_write(lz, s2w_blk->plain + written,
s2w_blk->loaded - written);
- if (-1 == wr) {
+ if( -1 == wr ) {
show_error( "LZ_compress_write() failed." );
fatal();
}
@@ -324,16 +289,16 @@ work_compr(S2w_blk *s2w_blk, W2m_q *w2m_q, unsigned clidx,
work_lz_rd(w2m_blk, sizeof_compr, lz);
}
- if (-1 == LZ_compress_finish(lz)) {
+ if( -1 == LZ_compress_finish(lz) ) {
show_error( "LZ_compress_finish() failed." );
fatal();
}
- while (!LZ_compress_finished(lz)) {
+ while( !LZ_compress_finished(lz) ) {
work_lz_rd(w2m_blk, sizeof_compr, lz);
}
- if (-1 == LZ_compress_close(lz)) {
+ if( -1 == LZ_compress_close(lz) ) {
show_error( "LZ_compress_close() failed." );
fatal();
}
@@ -345,290 +310,201 @@ work_compr(S2w_blk *s2w_blk, W2m_q *w2m_q, unsigned clidx,
xlock(&w2m_q->av_or_exit);
w2m_blk->next = w2m_q->head;
w2m_q->head = w2m_blk;
- if (w2m_blk->id == w2m_q->needed) {
+ if( w2m_blk->id == w2m_q->needed ) {
xsignal(&w2m_q->av_or_exit);
}
xunlock(&w2m_q->av_or_exit);
}
-static void
-work(S2w_q *s2w_q, W2m_q *w2m_q, unsigned clidx,
- const size_t sizeof_compr, const size_t sizeof_w2m_blk)
-{
- for (;;) {
+struct Worker_arg
+ {
+ int dictionary_size;
+ int match_len_limit;
+ S2w_q *s2w_q;
+ W2m_q *w2m_q;
+ int sizeof_compr;
+ size_t sizeof_W2m_blk;
+ };
+
+
+void * worker( void * arg )
+ {
+ const Worker_arg & tmp = *(Worker_arg *)arg;
+ const int dictionary_size = tmp.dictionary_size;
+ const int match_len_limit = tmp.match_len_limit;
+ S2w_q *s2w_q = tmp.s2w_q;
+ W2m_q *w2m_q = tmp.w2m_q;
+ const int sizeof_compr = tmp.sizeof_compr;
+ const size_t sizeof_w2m_blk = tmp.sizeof_W2m_blk;
+
+ while( true )
+ {
S2w_blk *s2w_blk;
/* Grab a block to work on. */
xlock_pred(&s2w_q->av_or_eof);
- while (0 == s2w_q->head && !s2w_q->eof) {
+ while( 0 == s2w_q->head && !s2w_q->eof ) {
xwait(&s2w_q->av_or_eof);
}
- if (0 == s2w_q->head) {
+ if( 0 == s2w_q->head ) {
/* No blocks available and splitter exited. */
xunlock(&s2w_q->av_or_eof);
break;
}
s2w_blk = s2w_q->head;
s2w_q->head = s2w_blk->next;
- if (0 == s2w_q->head) {
+ if( 0 == s2w_q->head ) {
s2w_q->tail = 0;
}
xunlock(&s2w_q->av_or_eof);
- work_compr(s2w_blk, w2m_q, clidx, sizeof_compr, sizeof_w2m_blk);
+ work_compr( dictionary_size, match_len_limit, s2w_blk, w2m_q,
+ sizeof_compr, sizeof_w2m_blk );
(*freef)(s2w_blk);
}
/* Notify muxer when last worker exits. */
xlock(&w2m_q->av_or_exit);
- if (0u == --w2m_q->working && 0 == w2m_q->head) {
+ if( 0 == --w2m_q->working && 0 == w2m_q->head ) {
xsignal(&w2m_q->av_or_exit);
}
xunlock(&w2m_q->av_or_exit);
-}
-
-
-struct Work_arg
-{
- S2w_q *s2w_q;
- W2m_q *w2m_q;
- unsigned clidx;
- size_t sizeof_compr,
- sizeof_W2m_blk;
-};
-
-
-static void *
-work_wrap(void *v_work_arg)
-{
- Work_arg *work_arg;
-
- work_arg = (Work_arg *)v_work_arg;
- work(
- work_arg->s2w_q,
- work_arg->w2m_q,
- work_arg->clidx,
- work_arg->sizeof_compr,
- work_arg->sizeof_W2m_blk
- );
return 0;
-}
-
-
-static void *
-reord_alloc(size_t size, void *)
-{
- return (*mallocf)(size);
-}
-
-
-static void
-reord_dealloc(void *ptr, void *)
-{
- (*freef)(ptr);
-}
-
-
-static void
-mux_write(M2s_q *m2s_q, lacos_rbtree_node **reord,
- uint64_t *reord_needed, int outfd)
-{
- assert(0 != *reord);
-
- /*
- Go on until the tree becomes empty or the next block is found to be
- missing.
- */
- do {
- lacos_rbtree_node *reord_head;
- W2m_blk *reord_w2m_blk;
-
- reord_head = lacos_rbtree_min(*reord);
- assert(0 != reord_head);
-
- reord_w2m_blk = (W2m_blk *)(*(void **)reord_head);
- if (reord_w2m_blk->id != *reord_needed) {
- break;
- }
-
- /* Write out "reord_w2m_blk". */
- if (-1 != outfd) {
- char unsigned *cp;
-
- cp = reord_w2m_blk->compr;
- while (reord_w2m_blk->produced > 0u) {
- ssize_t written;
-
- written = write(outfd, cp, reord_w2m_blk->produced > (size_t)SSIZE_MAX
- ? (size_t)SSIZE_MAX : reord_w2m_blk->produced);
- if (-1 == written) {
- fail("write()", errno);
- }
-
- reord_w2m_blk->produced -= (size_t)written;
- cp += written;
- }
- }
-
- ++*reord_needed;
-
- xlock(&m2s_q->av);
- if (0u == m2s_q->num_free++) {
- xsignal(&m2s_q->av);
- }
- xunlock(&m2s_q->av);
-
- lacos_rbtree_delete(
- reord, /* new_root */
- reord_head, /* old_node */
- 0, /* old_data */
- reord_dealloc, /* dealloc() */
- 0 /* alloc_ctl */
- );
-
- /* Release "reord_w2m_blk". */
- (*freef)(reord_w2m_blk);
- } while (0 != *reord);
-}
-
+ }
-static void
-mux(W2m_q *w2m_q, M2s_q *m2s_q, int outfd)
-{
- lacos_rbtree_node *reord;
- uint64_t reord_needed;
- reord = 0;
- reord_needed = 0u;
+void muxer_loop( W2m_q *w2m_q, M2s_q *m2s_q, const int num_slots, const int outfd )
+ {
+ unsigned long long needed_id = 0;
+ std::vector< W2m_blk * > circular_buffer( num_slots, (W2m_blk *)0 );
xlock_pred(&w2m_q->av_or_exit);
- for (;;) {
- W2m_blk *w2m_blk;
-
+ while( true )
+ {
/* Grab all available compressed blocks in one step. */
- while (0 == w2m_q->head && 0u < w2m_q->working) {
+ while( w2m_q->head == 0 && w2m_q->working > 0 )
xwait(&w2m_q->av_or_exit);
- }
- if (0 == w2m_q->head) {
- /* w2m_q is empty and all workers exited */
- break;
- }
+ if( w2m_q->head == 0 ) break; // queue is empty. all workers exited
- w2m_blk = w2m_q->head;
+ W2m_blk * w2m_blk = w2m_q->head;
w2m_q->head = 0;
xunlock(&w2m_q->av_or_exit);
- /* Merge blocks fetched this time into tree. */
+ // Merge blocks fetched this time into circular buffer
do {
- lacos_rbtree_node *new_node;
- W2m_blk *next;
-
- if (-1 == lacos_rbtree_insert(
- &reord, /* new_root */
- &new_node, /* new_node */
- w2m_blk, /* new_data */
- w2m_blk_cmp, /* cmp() */
- reord_alloc, /* alloc() */
- 0 /* alloc_ctl */
- )) {
- /* id collision shouldn't happen */
- assert(0 == new_node);
- show_error( "lacos_rbtree_insert(): out of memory." );
- fatal();
- }
-
- next = w2m_blk->next;
+ // id collision shouldn't happen
+ assert( circular_buffer[w2m_blk->id%num_slots] == 0 );
+ circular_buffer[w2m_blk->id%num_slots] = w2m_blk;
+ W2m_blk * next = w2m_blk->next;
w2m_blk->next = 0;
w2m_blk = next;
- } while (0 != w2m_blk);
+ } while( w2m_blk != 0 );
+
+ // Write out initial continuous sequence of reordered blocks
+ while( true )
+ {
+ W2m_blk * needed_w2m_blk = circular_buffer[needed_id%num_slots];
+ if( needed_w2m_blk == 0 ) break;
+
+ out_size += needed_w2m_blk->produced;
+
+ if( outfd >= 0 )
+ {
+ const int wr = writeblock( outfd, (char *)needed_w2m_blk->compr, needed_w2m_blk->produced );
+ if( wr != needed_w2m_blk->produced ) fail("write()", errno);
+ }
+ circular_buffer[needed_id%num_slots] = 0;
+ ++needed_id;
+
+ xlock(&m2s_q->av);
+ if( 0 == m2s_q->num_free++ ) xsignal(&m2s_q->av);
+ xunlock(&m2s_q->av);
- /* Write out initial continuous sequence of reordered blocks. */
- mux_write(m2s_q, &reord, &reord_needed, outfd);
+ (*freef)(needed_w2m_blk);
+ }
xlock_pred(&w2m_q->av_or_exit);
- w2m_q->needed = reord_needed;
- }
+ w2m_q->needed = needed_id;
+ }
xunlock(&w2m_q->av_or_exit);
- assert(0 == reord);
-}
+ for( int i = 0; i < num_slots; ++i )
+ if( circular_buffer[i] != 0 )
+ { show_error( "circular buffer not empty" ); fatal(); }
+ }
+} // end namespace
-static void
-plzip(unsigned num_worker, unsigned num_slot, unsigned clidx, int print_cctrs,
- int infd, int outfd)
-{
+
+void * muxer( void * arg )
+ {
+ const Muxer_arg & tmp = *(Muxer_arg *)arg;
+ const int dictionary_size = tmp.dictionary_size;
+ const int match_len_limit = tmp.match_len_limit;
+ const int num_workers = tmp.num_workers;
+ const int num_slots = tmp.num_slots;
+ const int debug_level = tmp.debug_level;
+ const int infd = tmp.infd;
+ const int outfd = tmp.outfd;
S2w_q s2w_q;
W2m_q w2m_q;
- M2s_q m2s_q;
- Split_arg split_arg;
- pthread_t splitter;
- Work_arg work_arg;
- pthread_t *worker;
- unsigned i;
- assert(clidx < sizeof compr_lev / sizeof compr_lev[0]);
+ if( debug_level & 2 ) { mallocf = trace_malloc; freef = trace_free; }
+ else { mallocf = malloc; freef = free; }
s2w_q_init(&s2w_q);
- w2m_q_init(&w2m_q, num_worker);
- m2s_q_init(&m2s_q, num_slot);
-
-#define SIZES(struc, arr, arsz_unsigned, arg) \
- do { \
- unsigned tmp; \
-\
- tmp = arsz_unsigned; \
- if ((size_t)-1 < tmp) { \
- show_error( "size_t overflow in sizeof_" #arr "." ); \
- fatal(); \
- } \
- arg ## _arg . sizeof_ ## arr = tmp; \
-\
- if ((size_t)-1 - sizeof(struc) \
- < arg ## _arg . sizeof_ ## arr - (size_t)1) { \
- show_error( "size_t overflow in sizeof_" #struc "." ); \
- fatal(); \
- } \
- arg ## _arg . sizeof_ ## struc = sizeof(struc) \
- + (arg ## _arg . sizeof_ ## arr - (size_t)1); \
- } while (0)
-
- split_arg.m2s_q = &m2s_q;
- split_arg.s2w_q = &s2w_q;
- split_arg.infd = infd;
- SIZES(S2w_blk, plain, 2u * compr_lev[clidx].dict_size, split);
- xcreate(&splitter, split_wrap, &split_arg);
-
- work_arg.s2w_q = &s2w_q;
- work_arg.w2m_q = &w2m_q;
- work_arg.clidx = clidx;
- SIZES(W2m_blk, compr, (4u + 1u + 1u)
- + ((unsigned)split_arg.sizeof_plain * 9u + 7u) / 8u + (4u + 8u + 8u),
- work);
-
-#undef SIZES
-
- assert(0u < num_worker);
- assert((size_t)-1 / sizeof *worker >= num_worker);
- worker = (pthread_t *)xalloc(num_worker * sizeof *worker);
- for (i = 0u; i < num_worker; ++i) {
- xcreate(&worker[i], work_wrap, &work_arg);
- }
-
- mux(&w2m_q, &m2s_q, outfd);
-
- i = num_worker;
- do {
- xjoin(worker[--i]);
- } while (0u < i);
- (*freef)(worker);
-
- xjoin(splitter);
+ w2m_q_init(&w2m_q, num_workers);
+ M2s_q m2s_q( num_slots );
+
+
+ Splitter_arg splitter_arg;
+ splitter_arg.m2s_q = &m2s_q;
+ splitter_arg.s2w_q = &s2w_q;
+ splitter_arg.infd = infd;
+ splitter_arg.sizeof_plain = 2 * std::max( 65536, dictionary_size );
+ splitter_arg.sizeof_S2w_blk = sizeof(S2w_blk) + splitter_arg.sizeof_plain - 1;
+
+ pthread_t splitter_thread;
+ xcreate(&splitter_thread, splitter, &splitter_arg);
+
+ Worker_arg worker_arg;
+ worker_arg.dictionary_size = dictionary_size;
+ worker_arg.match_len_limit = match_len_limit;
+ worker_arg.s2w_q = &s2w_q;
+ worker_arg.w2m_q = &w2m_q;
+ worker_arg.sizeof_compr = 6 + 20 + ( ( splitter_arg.sizeof_plain / 8 ) * 9 );
+ worker_arg.sizeof_W2m_blk = sizeof(W2m_blk) + worker_arg.sizeof_compr - 1;
+
+ pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers];
+ if( worker_threads == 0 ) fail("not enough memory.", errno);
+ for( int i = 0; i < num_workers; ++i )
+ xcreate(&worker_threads[i], worker, &worker_arg);
+
+ muxer_loop( &w2m_q, &m2s_q, num_slots, outfd );
+
+ for( int i = num_workers - 1; i >= 0; --i )
+ xjoin(worker_threads[i]);
+ delete[] worker_threads; worker_threads = 0;
+
+ xjoin(splitter_thread);
+
+ if( verbosity >= 1 )
+ {
+ if( in_size <= 0 || out_size <= 0 )
+ std::fprintf( stderr, "no data compressed.\n" );
+ else
+ std::fprintf( stderr, "%6.3f:1, %6.3f bits/byte, "
+ "%5.2f%% saved, %lld in, %lld out.\n",
+ (double)in_size / out_size,
+ ( 8.0 * out_size ) / in_size,
+ 100.0 * ( 1.0 - ( (double)out_size / in_size ) ),
+ in_size, out_size );
+ }
const int FW = ( sizeof(long unsigned) * 8 ) / 3 + 1;
- if (print_cctrs && 0 > fprintf(stderr,
+ if( ( debug_level & 1 ) && 0 > fprintf(stderr,
"any worker tried to consume from splitter: %*lu\n"
"any worker stalled : %*lu\n"
"muxer tried to consume from workers : %*lu\n"
@@ -645,25 +521,9 @@ plzip(unsigned num_worker, unsigned num_slot, unsigned clidx, int print_cctrs,
fatal();
}
- m2s_q_uninit(&m2s_q, num_slot);
+ assert( m2s_q.num_free == num_slots );
w2m_q_uninit(&w2m_q);
s2w_q_uninit(&s2w_q);
-}
-
-
-void *
-plzip_wrap(void *v_arg)
-{
- Plzip_arg *arg = (Plzip_arg *)v_arg;
- plzip(
- arg->num_worker,
- arg->num_slot,
- arg->clidx,
- arg->print_cctrs,
- arg->infd,
- arg->outfd
- );
-
xraise(SIGUSR2);
return 0;
-}
+ }