summaryrefslogtreecommitdiffstats
path: root/compress.cc
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--compress.cc55
1 files changed, 30 insertions, 25 deletions
diff --git a/compress.cc b/compress.cc
index f055b03..7945cf0 100644
--- a/compress.cc
+++ b/compress.cc
@@ -48,11 +48,14 @@
void xinit( pthread_cond_t * cond, pthread_mutex_t * mutex )
{
- int errcode = pthread_mutex_init( mutex, 0 );
- if( errcode ) { show_error( "pthread_mutex_init", errcode ); fatal(); }
-
- errcode = pthread_cond_init( cond, 0 );
+ int errcode = pthread_cond_init( cond, 0 );
if( errcode ) { show_error( "pthread_cond_init", errcode ); fatal(); }
+
+ if( mutex )
+ {
+ errcode = pthread_mutex_init( mutex, 0 );
+ if( errcode ) { show_error( "pthread_mutex_init", errcode ); fatal(); }
+ }
}
@@ -61,8 +64,11 @@ void xdestroy( pthread_cond_t * cond, pthread_mutex_t * mutex )
int errcode = pthread_cond_destroy( cond );
if( errcode ) { show_error( "pthread_cond_destroy", errcode ); fatal(); }
- errcode = pthread_mutex_destroy( mutex );
- if( errcode ) { show_error( "pthread_mutex_destroy", errcode ); fatal(); }
+ if( mutex )
+ {
+ errcode = pthread_mutex_destroy( mutex );
+ if( errcode ) { show_error( "pthread_mutex_destroy", errcode ); fatal(); }
+ }
}
@@ -125,16 +131,16 @@ public:
private:
unsigned long long receive_id; // id assigned to next packet received
unsigned long long deliver_id; // id of next packet to be delivered
- Slot_tally slot_tally;
+ Slot_tally slot_tally; // limits the number of input packets
std::queue< Packet * > packet_queue;
std::vector< Packet * > circular_buffer;
- int num_working; // Number of workers still running
+ int num_working; // number of workers still running
const int num_slots; // max packets in circulation
pthread_mutex_t imutex;
pthread_cond_t iav_or_eof; // input packet available or splitter done
pthread_mutex_t omutex;
pthread_cond_t oav_or_exit; // output packet available or all workers exited
- bool eof; // splitter done
+ bool eof; // splitter done
public:
Packet_courier( const int num_workers, const int slots )
@@ -148,6 +154,8 @@ public:
~Packet_courier()
{ xdestroy( &iav_or_eof, &imutex ); xdestroy( &oav_or_exit, &omutex ); }
+ const Slot_tally & tally() const { return slot_tally; }
+
// make a packet with data received from splitter
void receive_packet( uint8_t * const data, const int size )
{
@@ -182,10 +190,9 @@ public:
xunlock( &imutex );
if( ipacket == 0 )
{
- // Notify muxer when last worker exits
+ // notify muxer when last worker exits
xlock( &omutex );
- if( --num_working == 0 )
- xsignal( &oav_or_exit );
+ if( --num_working == 0 ) xsignal( &oav_or_exit );
xunlock( &omutex );
}
return ipacket;
@@ -198,7 +205,7 @@ public:
// id collision shouldn't happen
if( circular_buffer[opacket->id%num_slots] != 0 )
internal_error( "id collision in collect_packet" );
- // Merge packet into circular buffer
+ // merge packet into circular buffer
circular_buffer[opacket->id%num_slots] = opacket;
if( opacket->id == deliver_id ) xsignal( &oav_or_exit );
xunlock( &omutex );
@@ -240,8 +247,6 @@ public:
if( circular_buffer[i] != 0 ) return false;
return true;
}
-
- const Slot_tally & tally() const { return slot_tally; }
};
@@ -267,10 +272,10 @@ extern "C" void * csplitter( void * arg )
for( bool first_post = true; ; first_post = false )
{
uint8_t * data = new( std::nothrow ) uint8_t[data_size];
- if( data == 0 ) { pp( "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 )
- { pp(); show_error( "read error", errno ); fatal(); }
+ { pp(); show_error( "Read error", errno ); fatal(); }
if( size > 0 || first_post ) // first packet can be empty
{
@@ -314,7 +319,7 @@ extern "C" void * cworker( 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 ) { pp( "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 =
@@ -322,7 +327,7 @@ extern "C" void * cworker( void * arg )
if( !encoder || LZ_compress_errno( encoder ) != LZ_ok )
{
if( !encoder || LZ_compress_errno( encoder ) == LZ_mem_error )
- pp( "not enough memory. Try a smaller dictionary size" );
+ pp( "Not enough memory. Try a smaller dictionary size" );
else
internal_error( "invalid argument to encoder" );
fatal();
@@ -386,7 +391,7 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd )
{
const int wr = writeblock( outfd, opacket->data, opacket->size );
if( wr != opacket->size )
- { pp(); show_error( "write error", errno ); fatal(); }
+ { pp(); show_error( "Write error", errno ); fatal(); }
}
delete[] opacket->data;
delete opacket;
@@ -416,7 +421,7 @@ int compress( const int data_size, const int dictionary_size,
pthread_t splitter_thread;
int errcode = pthread_create( &splitter_thread, 0, csplitter, &splitter_arg );
if( errcode )
- { show_error( "can't create splitter thread", errcode ); fatal(); }
+ { show_error( "Can't create splitter thread", errcode ); fatal(); }
Worker_arg worker_arg;
worker_arg.courier = &courier;
@@ -426,12 +431,12 @@ int compress( const int data_size, const int dictionary_size,
pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers];
if( worker_threads == 0 )
- { pp( "not enough memory" ); fatal(); }
+ { pp( "Not enough memory" ); fatal(); }
for( int i = 0; i < num_workers; ++i )
{
errcode = pthread_create( &worker_threads[i], 0, cworker, &worker_arg );
if( errcode )
- { show_error( "can't create worker threads", errcode ); fatal(); }
+ { show_error( "Can't create worker threads", errcode ); fatal(); }
}
muxer( courier, pp, outfd );
@@ -440,13 +445,13 @@ int compress( const int data_size, const int dictionary_size,
{
errcode = pthread_join( worker_threads[i], 0 );
if( errcode )
- { show_error( "can't join worker threads", errcode ); fatal(); }
+ { show_error( "Can't join worker threads", errcode ); fatal(); }
}
delete[] worker_threads; worker_threads = 0;
errcode = pthread_join( splitter_thread, 0 );
if( errcode )
- { show_error( "can't join splitter thread", errcode ); fatal(); }
+ { show_error( "Can't join splitter thread", errcode ); fatal(); }
if( verbosity >= 1 )
{