From 745033b223d0c73153911d65dae915d36866a04d Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sat, 7 Nov 2015 16:18:02 +0100 Subject: Adding upstream version 0.5. Signed-off-by: Daniel Baumann --- compress.cc | 111 ++++++++++++++++++++++++++++++++++++------------------------ 1 file changed, 66 insertions(+), 45 deletions(-) (limited to 'compress.cc') diff --git a/compress.cc b/compress.cc index e66b537..3f5e143 100644 --- a/compress.cc +++ b/compress.cc @@ -19,7 +19,6 @@ #define _FILE_OFFSET_BITS 64 #include -#include #include #include #include @@ -49,70 +48,70 @@ 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(); } + int errcode = pthread_mutex_init( mutex, 0 ); + if( errcode ) { show_error( "pthread_mutex_init", errcode ); fatal(); } - ret = pthread_cond_init( cond, 0 ); - if( ret != 0 ) { show_error( "pthread_cond_init", ret ); fatal(); } + errcode = pthread_cond_init( cond, 0 ); + if( errcode ) { show_error( "pthread_cond_init", errcode ); 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(); } + int errcode = pthread_cond_destroy( cond ); + if( errcode ) { show_error( "pthread_cond_destroy", errcode ); fatal(); } - ret = pthread_mutex_destroy( mutex ); - if( ret != 0 ) { show_error( "pthread_mutex_destroy", ret ); fatal(); } + errcode = pthread_mutex_destroy( mutex ); + if( errcode ) { show_error( "pthread_mutex_destroy", errcode ); fatal(); } } void xlock( pthread_mutex_t * mutex ) { - int ret = pthread_mutex_lock( mutex ); - if( ret != 0 ) { show_error( "pthread_mutex_lock", ret ); fatal(); } + int errcode = pthread_mutex_lock( mutex ); + if( errcode ) { show_error( "pthread_mutex_lock", errcode ); fatal(); } } void xunlock( pthread_mutex_t * mutex ) { - int ret = pthread_mutex_unlock( mutex ); - if( ret != 0 ) { show_error( "pthread_mutex_unlock", ret ); fatal(); } + int errcode = pthread_mutex_unlock( mutex ); + if( errcode ) { show_error( "pthread_mutex_unlock", errcode ); 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(); } + int errcode = pthread_cond_wait( cond, mutex ); + if( errcode ) { show_error( "pthread_cond_wait", errcode ); fatal(); } } void xsignal( pthread_cond_t * cond ) { - int ret = pthread_cond_signal( cond ); - if( ret != 0 ) { show_error( "pthread_cond_signal", ret ); fatal(); } + int errcode = pthread_cond_signal( cond ); + if( errcode ) { show_error( "pthread_cond_signal", errcode ); fatal(); } } void xbroadcast( pthread_cond_t * cond ) { - int ret = pthread_cond_broadcast( cond ); - if( ret != 0 ) { show_error( "pthread_cond_broadcast", ret ); fatal(); } + int errcode = pthread_cond_broadcast( cond ); + if( errcode ) { show_error( "pthread_cond_broadcast", errcode ); 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(); } + int errcode = pthread_create( thread, 0, routine, arg ); + if( errcode ) { show_error( "pthread_create", errcode ); fatal(); } } void xjoin( pthread_t thread ) { - int ret = pthread_join( thread, 0 ); - if( ret != 0 ) { show_error( "pthread_join", ret ); fatal(); } + int errcode = pthread_join( thread, 0 ); + if( errcode ) { show_error( "pthread_join", errcode ); fatal(); } } @@ -125,8 +124,8 @@ long long out_size = 0; struct Packet // data block with a serial number { unsigned long long id; // serial number assigned as received - int size; // # of bytes in data uint8_t * data; + int size; // number of bytes in data (if any) }; @@ -168,8 +167,8 @@ public: { Packet * ipacket = new Packet; ipacket->id = receive_id++; - ipacket->size = size; ipacket->data = data; + ipacket->size = size; slot_tally.get_slot(); // wait for a free slot xlock( &imutex ); packet_queue.push( ipacket ); @@ -210,7 +209,8 @@ public: { xlock( &omutex ); // id collision shouldn't happen - assert( circular_buffer[opacket->id%num_slots] == 0 ); + if( circular_buffer[opacket->id%num_slots] != 0 ) + internal_error( "id collision in collect_packet" ); // Merge packet into circular buffer circular_buffer[opacket->id%num_slots] = opacket; if( opacket->id == deliver_id ) xsignal( &oav_or_exit ); @@ -260,6 +260,7 @@ public: struct Splitter_arg { Packet_courier * courier; + const Pretty_print * pp; int infd; int data_size; }; @@ -271,15 +272,17 @@ void * splitter( void * arg ) { const Splitter_arg & tmp = *(Splitter_arg *)arg; Packet_courier & courier = *tmp.courier; + const Pretty_print & pp = *tmp.pp; const int infd = tmp.infd; const int data_size = tmp.data_size; for( bool first_post = true; ; first_post = false ) { uint8_t * data = new( std::nothrow ) uint8_t[data_size]; - if( data == 0 ) { show_error( "not enough memory" ); fatal(); } + if( data == 0 ) { pp( "not enough memory" ); fatal(); } const int size = readblock( infd, data, data_size ); - if( size != data_size && errno ) { show_error( "read", errno ); fatal(); } + if( size != data_size && errno ) + { pp(); show_error( "read error", errno ); fatal(); } if( size > 0 || first_post ) // first packet can be empty { @@ -299,9 +302,10 @@ void * splitter( void * arg ) struct Worker_arg { + Packet_courier * courier; + const Pretty_print * pp; int dictionary_size; int match_len_limit; - Packet_courier * courier; }; @@ -310,9 +314,10 @@ struct Worker_arg void * worker( void * arg ) { const Worker_arg & tmp = *(Worker_arg *)arg; + Packet_courier & courier = *tmp.courier; + const Pretty_print & pp = *tmp.pp; const int dictionary_size = tmp.dictionary_size; const int match_len_limit = tmp.match_len_limit; - Packet_courier & courier = *tmp.courier; while( true ) { @@ -321,13 +326,19 @@ void * worker( void * arg ) const int compr_size = 42 + packet->size + ( ( packet->size + 7 ) / 8 ); uint8_t * const new_data = new( std::nothrow ) uint8_t[compr_size]; - if( new_data == 0 ) { show_error( "not enough memory" ); fatal(); } + if( new_data == 0 ) { pp( "not enough memory" ); fatal(); } const int dict_size = std::max( LZ_min_dictionary_size(), std::min( dictionary_size, packet->size ) ); 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(); } + { + if( !encoder || LZ_compress_errno( encoder ) == LZ_mem_error ) + pp( "not enough memory. Try a smaller dictionary size" ); + else + internal_error( "invalid argument to encoder" ); + fatal(); + } int written = 0; int new_size = 0; @@ -339,25 +350,33 @@ void * worker( void * arg ) { const int wr = LZ_compress_write( encoder, packet->data + written, packet->size - written ); - if( wr < 0 ) { show_error( "LZ_compress_write failed." ); fatal(); } + if( wr < 0 ) internal_error( "library error (LZ_compress_write)" ); written += wr; } if( written >= packet->size ) LZ_compress_finish( encoder ); } const int rd = LZ_compress_read( encoder, new_data + new_size, compr_size - new_size ); - if( rd < 0 ) { show_error( "LZ_compress_read failed." ); fatal(); } + if( rd < 0 ) + { + pp(); + if( verbosity >= 0 ) + std::fprintf( stderr, "LZ_compress_read error: %s.\n", + LZ_strerror( LZ_compress_errno( encoder ) ) ); + fatal(); + } new_size += rd; - assert( new_size <= compr_size ); + if( new_size > compr_size ) + internal_error( "packet size exceeded in worker" ); if( LZ_compress_finished( encoder ) == 1 ) break; } if( LZ_compress_close( encoder ) < 0 ) - { show_error( "LZ_compress_close failed." ); fatal(); } + { pp( "LZ_compress_close failed" ); fatal(); } delete[] packet->data; - packet->size = new_size; packet->data = new_data; + packet->size = new_size; courier.collect_packet( packet ); } return 0; @@ -366,7 +385,7 @@ void * worker( void * arg ) // get from courier the processed and sorted packets, and write // their contents to the output file. -void muxer( Packet_courier & courier, const int outfd ) +void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd ) { while( true ) { @@ -379,7 +398,7 @@ void muxer( Packet_courier & courier, const int outfd ) { const int wr = writeblock( outfd, opacket->data, opacket->size ); if( wr != opacket->size ) - { show_error( "write", errno ); fatal(); } + { pp(); show_error( "write error", errno ); fatal(); } } delete[] opacket->data; delete opacket; @@ -394,7 +413,7 @@ void muxer( Packet_courier & courier, const int outfd ) 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 ) + const Pretty_print & pp, const int debug_level ) { in_size = 0; out_size = 0; @@ -402,6 +421,7 @@ int compress( const int data_size, const int dictionary_size, Splitter_arg splitter_arg; splitter_arg.courier = &courier; + splitter_arg.pp = &pp; splitter_arg.infd = infd; splitter_arg.data_size = data_size; @@ -409,20 +429,21 @@ int compress( const int data_size, const int dictionary_size, xcreate( &splitter_thread, splitter, &splitter_arg ); Worker_arg worker_arg; + worker_arg.courier = &courier; + worker_arg.pp = &pp; worker_arg.dictionary_size = dictionary_size; worker_arg.match_len_limit = match_len_limit; - worker_arg.courier = &courier; pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers]; if( worker_threads == 0 ) - { show_error( "not enough memory" ); fatal(); } + { pp( "not enough memory" ); fatal(); } for( int i = 0; i < num_workers; ++i ) xcreate( &worker_threads[i], worker, &worker_arg ); - muxer( courier, outfd ); + muxer( courier, pp, outfd ); for( int i = num_workers - 1; i >= 0; --i ) - xjoin(worker_threads[i]); + xjoin( worker_threads[i] ); delete[] worker_threads; worker_threads = 0; xjoin( splitter_thread ); @@ -455,6 +476,6 @@ int compress( const int data_size, const int dictionary_size, courier.ocheck_counter, courier.owait_counter ); - assert( courier.finished() ); + if( !courier.finished() ) internal_error( "courier not finished" ); return 0; } -- cgit v1.2.3