Merging upstream version 0.5.

Signed-off-by: Daniel Baumann <mail@daniel-baumann.ch>

1 files changed, 66 insertions, 45 deletions

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 <algorithm>
-#include <cassert>
 #include <cerrno>
 #include <climits>
 #include <csignal>
@@ -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;
   }