diff options
Diffstat (limited to 'plzip.cc')
| -rw-r--r-- | plzip.cc | 548 |
1 files changed, 0 insertions, 548 deletions
diff --git a/plzip.cc b/plzip.cc deleted file mode 100644 index dcae860..0000000 --- a/plzip.cc +++ /dev/null @@ -1,548 +0,0 @@ -/* Plzip - A parallel version of the lzip data compressor - Copyright (C) 2009 Laszlo Ersek. - 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 - the Free Software Foundation, either version 3 of the License, or - (at your option) any later version. - - This program is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with this program. If not, see <http://www.gnu.org/licenses/>. -*/ - -#define _FILE_OFFSET_BITS 64 - -#include <algorithm> -#include <cassert> -#include <cerrno> -#include <climits> -#include <csignal> -#include <cstdio> -#include <cstdlib> -#include <vector> -#include <pthread.h> -#include <stdint.h> -#include <unistd.h> -#include <lzlib.h> - -#include "plzip.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; - std::fprintf( stderr, "malloc(%lu) == %p\n", (unsigned long)size, ret ); - if( ret == 0 ) errno = save_errno; - return ret; - } - - -void trace_free( void *ptr ) - { - std::fprintf( stderr, "free(%p)\n", ptr ); - free( ptr ); - } - - -void * xalloc( size_t size ) - { - void *ret = (*mallocf)( size ); - if( ret == 0 ) { show_error( "not enough memory", errno ); fatal(); } - return ret; - } - - -void xinit( pthread_cond_t * cond, pthread_mutex_t * mutex ) - { - int ret = pthread_mutex_init( mutex, 0 ); - if( ret != 0 ) { show_error( "pthread_mutex_init", ret ); fatal(); } - - ret = pthread_cond_init( cond, 0 ); - if( ret != 0 ) { show_error( "pthread_cond_init", ret ); fatal(); } - } - - -void xdestroy( pthread_cond_t * cond, pthread_mutex_t * mutex ) - { - int ret = pthread_cond_destroy( cond ); - if( ret != 0 ) { show_error( "pthread_cond_destroy", ret ); fatal(); } - - ret = pthread_mutex_destroy( mutex ); - if( ret != 0 ) { show_error( "pthread_mutex_destroy", ret ); fatal(); } - } - - -void xlock( pthread_mutex_t * mutex ) - { - int ret = pthread_mutex_lock( mutex ); - if( ret != 0 ) { show_error( "pthread_mutex_lock", ret ); fatal(); } - } - - -void xunlock( pthread_mutex_t * mutex ) - { - int ret = pthread_mutex_unlock( mutex ); - if( ret != 0 ) { show_error( "pthread_mutex_unlock", ret ); fatal(); } - } - - -void xwait( pthread_cond_t * cond, pthread_mutex_t * mutex ) - { - int ret = pthread_cond_wait( cond, mutex ); - if( ret != 0 ) { show_error( "pthread_cond_wait", ret ); fatal(); } - } - - -void xsignal( pthread_cond_t * cond ) - { - int ret = pthread_cond_signal( cond ); - if( ret != 0 ) { show_error( "pthread_cond_signal", ret ); fatal(); } - } - - -void xbroadcast( pthread_cond_t * cond ) - { - int ret = pthread_cond_broadcast( cond ); - if( ret != 0 ) { show_error( "pthread_cond_broadcast", ret ); fatal(); } - } - - -void xcreate( pthread_t *thread, void *(*routine)(void *), void *arg ) - { - int ret = pthread_create( thread, 0, routine, arg ); - if( ret != 0 ) { show_error( "pthread_create", ret ); fatal(); } - } - - -void xjoin( pthread_t thread ) - { - int ret = pthread_join( thread, 0 ); - if( ret != 0 ) { show_error( "pthread_join", ret ); fatal(); } - } - - -struct Slot_tally // Synchronizes splitter to muxer - { - unsigned long check_counter; - unsigned long wait_counter; - int num_free; // Number of free slots - pthread_mutex_t mutex; - pthread_cond_t slot_av; // Free slot available - - Slot_tally( const int slots ) - : check_counter( 0 ), wait_counter( 0 ), num_free( slots ) - { xinit( &slot_av, &mutex ); } - - ~Slot_tally() { xdestroy( &slot_av, &mutex ); } - }; - - -struct S2w_blk // Splitter to worker data block - { - unsigned long long id; // Block serial number as read from infd - S2w_blk *next; // Next in queue - int loaded; // # of bytes in plain, may be 0 for 1st - uint8_t plain[1]; // Data read from infd, allocated: data_size - }; - - -struct S2w_queue - { - S2w_blk * head; // Next ready worker shall compress this - S2w_blk * tail; // Splitter will append here - unsigned long check_counter; - unsigned long wait_counter; - pthread_mutex_t mutex; - pthread_cond_t av_or_eof; // New block available or splitter done - bool eof; // Splitter done - - S2w_queue() - : head( 0 ), tail( 0 ), check_counter( 0 ), wait_counter( 0 ), eof( false ) - { xinit( &av_or_eof, &mutex ); } - - ~S2w_queue() { xdestroy( &av_or_eof, &mutex ); } - }; - - -struct W2m_blk // Worker to muxer data block - { - unsigned long long id; // Block index as read from infd - W2m_blk *next; // Next block in list (unordered) - int produced; // Number of bytes in compr - uint8_t compr[1]; // Data to write to outfd, alloc.: compr_size - }; - - -struct W2m_queue - { - unsigned long long needed_id; // Block needed for resuming writing - W2m_blk *head; // Block list (unordered) - unsigned long check_counter; - unsigned long wait_counter; - int num_working; // Number of workers still running - pthread_mutex_t mutex; - pthread_cond_t av_or_exit; // New block available or all workers exited - - W2m_queue( const int num_workers ) - : needed_id( 0 ), head( 0 ), check_counter( 0 ), wait_counter( 0 ), - num_working( num_workers ) - { xinit( &av_or_exit, &mutex ); } - - ~W2m_queue() { xdestroy( &av_or_exit, &mutex ); } - }; - - -struct Splitter_arg - { - Slot_tally * slot_tally; - S2w_queue * s2w_queue; - int infd; - int data_size; - int s2w_blk_size; - }; - - -void * splitter( void * arg ) - { - const Splitter_arg & tmp = *(Splitter_arg *)arg; - Slot_tally & slot_tally = *tmp.slot_tally; - S2w_queue & s2w_queue = *tmp.s2w_queue; - const int infd = tmp.infd; - const int data_size = tmp.data_size; - const int s2w_blk_size = tmp.s2w_blk_size; - - for( unsigned long long id = 0; ; ++id ) - { - S2w_blk * s2w_blk = (S2w_blk *)xalloc( s2w_blk_size ); - - // Fill block - const int rd = readblock( infd, s2w_blk->plain, data_size ); - if( rd != data_size && errno ) { show_error( "read", errno ); fatal(); } - - if( rd > 0 || id == 0 ) // first block can be empty - { - s2w_blk->id = id; - s2w_blk->next = 0; - s2w_blk->loaded = rd; - in_size += rd; - xlock( &slot_tally.mutex ); // Grab a free slot - ++slot_tally.check_counter; - while( slot_tally.num_free == 0 ) - { - ++slot_tally.wait_counter; - xwait( &slot_tally.slot_av, &slot_tally.mutex ); - } - --slot_tally.num_free; - xunlock( &slot_tally.mutex ); - } - else - { (*freef)( s2w_blk ); s2w_blk = 0; } - - xlock( &s2w_queue.mutex ); - if( s2w_blk != 0 ) - { - if( s2w_queue.tail == 0 ) s2w_queue.head = s2w_blk; - else s2w_queue.tail->next = s2w_blk; - s2w_queue.tail = s2w_blk; - xsignal( &s2w_queue.av_or_eof ); - } - else - { - s2w_queue.eof = true; - xbroadcast( &s2w_queue.av_or_eof ); - } - xunlock( &s2w_queue.mutex ); - - if( s2w_blk == 0 ) break; - } - return 0; - } - - -void work_compr( const int dictionary_size, const int match_len_limit, - const S2w_blk & s2w_blk, W2m_queue & w2m_queue, - const int compr_size, const int w2m_blk_size ) - { - assert( s2w_blk.loaded > 0 || s2w_blk.id == 0 ); - - W2m_blk * w2m_blk = (W2m_blk *)xalloc( w2m_blk_size ); - - const int dict_size = std::max( LZ_min_dictionary_size(), - std::min( dictionary_size, s2w_blk.loaded ) ); - LZ_Encoder * const encoder = - LZ_compress_open( dict_size, match_len_limit, LLONG_MAX ); - if( !encoder || LZ_compress_errno( encoder ) != LZ_ok ) - { show_error( "LZ_compress_open failed." ); fatal(); } - - int written = 0; - w2m_blk->produced = 0; - while( true ) - { - if( LZ_compress_write_size( encoder ) > 0 ) - { - if( written < s2w_blk.loaded ) - { - const int wr = LZ_compress_write( encoder, s2w_blk.plain + written, - s2w_blk.loaded - written ); - if( wr < 0 ) { show_error( "LZ_compress_write failed." ); fatal(); } - written += wr; - } - if( written >= s2w_blk.loaded ) LZ_compress_finish( encoder ); - } - assert( w2m_blk->produced < compr_size ); - const int rd = LZ_compress_read( encoder, w2m_blk->compr + w2m_blk->produced, - compr_size - w2m_blk->produced ); - if( rd < 0 ) { show_error( "LZ_compress_read failed." ); fatal(); } - w2m_blk->produced += rd; - if( LZ_compress_finished( encoder ) == 1 ) break; - } - - if( LZ_compress_close( encoder ) < 0 ) - { show_error( "LZ_compress_close failed." ); fatal(); } - - w2m_blk->id = s2w_blk.id; - - // Push block to muxer queue - xlock( &w2m_queue.mutex ); - w2m_blk->next = w2m_queue.head; - w2m_queue.head = w2m_blk; - if( w2m_blk->id == w2m_queue.needed_id ) xsignal( &w2m_queue.av_or_exit ); - xunlock( &w2m_queue.mutex ); - } - - -struct Worker_arg - { - int dictionary_size; - int match_len_limit; - S2w_queue * s2w_queue; - W2m_queue * w2m_queue; - int compr_size; - int w2m_blk_size; - }; - - -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_queue & s2w_queue = *tmp.s2w_queue; - W2m_queue & w2m_queue = *tmp.w2m_queue; - const int compr_size = tmp.compr_size; - const int w2m_blk_size = tmp.w2m_blk_size; - - while( true ) - { - S2w_blk *s2w_blk; - - // Grab a block to work on - xlock( &s2w_queue.mutex ); - ++s2w_queue.check_counter; - while( s2w_queue.head == 0 && !s2w_queue.eof ) - { - ++s2w_queue.wait_counter; - xwait( &s2w_queue.av_or_eof, &s2w_queue.mutex ); - } - if( s2w_queue.head == 0 ) // No blocks available and splitter exited - { - xunlock( &s2w_queue.mutex ); - break; - } - s2w_blk = s2w_queue.head; - s2w_queue.head = s2w_blk->next; - if( s2w_queue.head == 0 ) s2w_queue.tail = 0; - xunlock( &s2w_queue.mutex ); - - work_compr( dictionary_size, match_len_limit, *s2w_blk, w2m_queue, - compr_size, w2m_blk_size ); - (*freef)( s2w_blk ); - } - - // Notify muxer when last worker exits - xlock( &w2m_queue.mutex ); - if( --w2m_queue.num_working == 0 && w2m_queue.head == 0 ) - xsignal( &w2m_queue.av_or_exit ); - xunlock( &w2m_queue.mutex ); - return 0; - } - - -void muxer( Slot_tally & slot_tally, W2m_queue & w2m_queue, - 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( &w2m_queue.mutex ); - while( true ) - { - // Grab all available compressed blocks in one step - ++w2m_queue.check_counter; - while( w2m_queue.head == 0 && w2m_queue.num_working > 0 ) - { - ++w2m_queue.wait_counter; - xwait( &w2m_queue.av_or_exit, &w2m_queue.mutex ); - } - if( w2m_queue.head == 0 ) break; // queue is empty. all workers exited - - W2m_blk * w2m_blk = w2m_queue.head; - w2m_queue.head = 0; - xunlock( &w2m_queue.mutex ); - - // Merge blocks fetched this time into circular buffer - do { - // 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( w2m_blk != 0 ); - - // Write out initial continuous sequence of reordered blocks - while( true ) - { - w2m_blk = circular_buffer[needed_id%num_slots]; - if( w2m_blk == 0 ) break; - - out_size += w2m_blk->produced; - - if( outfd >= 0 ) - { - const int wr = writeblock( outfd, w2m_blk->compr, w2m_blk->produced ); - if( wr != w2m_blk->produced ) - { show_error( "write", errno ); fatal(); } - } - circular_buffer[needed_id%num_slots] = 0; - ++needed_id; - - xlock( &slot_tally.mutex ); - if( slot_tally.num_free++ == 0 ) xsignal( &slot_tally.slot_av ); - xunlock( &slot_tally.mutex ); - - (*freef)( w2m_blk ); - } - - xlock( &w2m_queue.mutex ); - w2m_queue.needed_id = needed_id; - } - xunlock( &w2m_queue.mutex ); - - for( int i = 0; i < num_slots; ++i ) - if( circular_buffer[i] != 0 ) - { show_error( "circular buffer not empty" ); fatal(); } - } - -} // end namespace - - -int compress( const int data_size, const int dictionary_size, - const int match_len_limit, const int num_workers, - const int num_slots, const int infd, const int outfd, - const int debug_level ) - { - if( debug_level & 2 ) { mallocf = trace_malloc; freef = trace_free; } - else { mallocf = malloc; freef = free; } - - Slot_tally slot_tally( num_slots ); - S2w_queue s2w_queue; - W2m_queue w2m_queue( num_workers ); - - Splitter_arg splitter_arg; - splitter_arg.slot_tally = &slot_tally; - splitter_arg.s2w_queue = &s2w_queue; - splitter_arg.infd = infd; - splitter_arg.data_size = data_size; - splitter_arg.s2w_blk_size = sizeof (S2w_blk) + data_size - 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_queue = &s2w_queue; - worker_arg.w2m_queue = &w2m_queue; - worker_arg.compr_size = 6 + 20 + ( ( data_size / 8 ) * 9 ); - worker_arg.w2m_blk_size = sizeof (W2m_blk) + worker_arg.compr_size - 1; - - pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers]; - if( worker_threads == 0 ) - { show_error( "not enough memory.", errno ); fatal(); } - for( int i = 0; i < num_workers; ++i ) - xcreate( &worker_threads[i], worker, &worker_arg ); - - muxer( slot_tally, w2m_queue, 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 (unsigned long) * 8 ) / 3 + 1; - if( debug_level & 1 ) - std::fprintf( stderr, - "any worker tried to consume from splitter: %*lu\n" - "any worker stalled : %*lu\n" - "muxer tried to consume from workers : %*lu\n" - "muxer stalled : %*lu\n" - "splitter tried to fill a block : %*lu\n" - "splitter stalled : %*lu\n", - FW, s2w_queue.check_counter, - FW, s2w_queue.wait_counter, - FW, w2m_queue.check_counter, - FW, w2m_queue.wait_counter, - FW, slot_tally.check_counter, - FW, slot_tally.wait_counter ); - - assert( slot_tally.num_free == num_slots ); - assert( s2w_queue.eof ); - assert( s2w_queue.head == 0 ); - assert( s2w_queue.tail == 0 ); - assert( w2m_queue.num_working == 0 ); - assert( w2m_queue.head == 0 ); - return 0; - } |