Merging upstream version 0.7.

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

1 files changed, 50 insertions, 40 deletions

diff --git a/decompress.cc b/decompress.cc
index a5d4994..df8d88e 100644
--- a/decompress.cc
+++ b/decompress.cc
@@ -38,6 +38,7 @@
 
 namespace {
 
+enum { max_packet_size = 1 << 20 };
 long long in_size = 0;
 long long out_size = 0;
 
@@ -59,17 +60,18 @@ public:
 private:
   int receive_worker_id;	// worker queue currently receiving packets
   int deliver_worker_id;	// worker queue currently delivering packets
-  Slot_tally slot_tally;
+  Slot_tally slot_tally;		// limits the number of input packets
   std::vector< std::queue< Packet * > > ipacket_queues;
   std::vector< std::queue< Packet * > > opacket_queues;
-  int num_working;			// Number of workers still running
-  const int num_workers;		// Number of workers
-  const int num_slots;			// max packets in circulation
+  int num_working;			// number of workers still running
+  const int num_workers;		// number of workers
+  int num_free;				// remaining free output slots
   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
+  pthread_cond_t slot_av;		// free output slot available
+  bool eof;				// splitter done
 
 public:
   Packet_courier( const int workers, const int slots )
@@ -78,11 +80,19 @@ public:
       receive_worker_id( 0 ), deliver_worker_id( 0 ),
       slot_tally( slots ), ipacket_queues( workers ),
       opacket_queues( workers ), num_working( workers ),
-      num_workers( workers ), num_slots( slots ), eof( false )
-    { xinit( &iav_or_eof, &imutex ); xinit( &oav_or_exit, &omutex ); }
+      num_workers( workers ), num_free( 8 * slots ), eof( false )
+    {
+    xinit( &iav_or_eof, &imutex );
+    xinit( &oav_or_exit, &omutex ); xinit( &slot_av, 0 );
+    }
 
   ~Packet_courier()
-    { xdestroy( &iav_or_eof, &imutex ); xdestroy( &oav_or_exit, &omutex ); }
+    {
+    xdestroy( &iav_or_eof, &imutex );
+    xdestroy( &oav_or_exit, &omutex ); xdestroy( &slot_av, 0 );
+    }
+
+  const Slot_tally & tally() const { return slot_tally; }
 
   // make a packet with data received from splitter
   // if data == 0, move to next queue
@@ -123,10 +133,9 @@ public:
       { if( ipacket->data != 0 ) slot_tally.leave_slot(); }
     else
       {
-      // 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;
@@ -136,6 +145,12 @@ public:
   void collect_packet( Packet * const opacket, const int worker_id )
     {
     xlock( &omutex );
+    if( opacket->data != 0 )
+      {
+      while( worker_id != deliver_worker_id && num_free <= 0 )
+        xwait( &slot_av, &omutex );
+      --num_free;
+      }
     opacket_queues[worker_id].push( opacket );
     if( worker_id == deliver_worker_id ) xsignal( &oav_or_exit );
     xunlock( &omutex );
@@ -159,12 +174,14 @@ public:
       if( opacket_queues[deliver_worker_id].empty() ) break;
       opacket = opacket_queues[deliver_worker_id].front();
       opacket_queues[deliver_worker_id].pop();
-      if( opacket->data != 0 ) break;
-      else
+      if( opacket->data != 0 )
         {
-        if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0;
-        delete opacket; opacket = 0;
+        if( ++num_free == 1 ) xsignal( &slot_av );
+        break;
         }
+      if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0;
+      xbroadcast( &slot_av );		// restart deliver_worker_id thread
+      delete opacket; opacket = 0;
       }
     xunlock( &omutex );
     return opacket;
@@ -187,8 +204,6 @@ public:
       if( !opacket_queues[i].empty() ) return false;
     return true;
     }
-
-  const Slot_tally & tally() const { return slot_tally; }
   };
 
 
@@ -197,7 +212,6 @@ struct Splitter_arg
   Packet_courier * courier;
   const Pretty_print * pp;
   int infd;
-  int packet_size;
   };
 
 
@@ -211,19 +225,19 @@ extern "C" void * dsplitter( void * arg )
   const int infd = tmp.infd;
   const int hsize = 6;				// header size
   const int tsize = 20;				// trailer size
-  const int buffer_size = tmp.packet_size;
+  const int buffer_size = max_packet_size;
   const int base_buffer_size = tsize + buffer_size + hsize;
   uint8_t * const base_buffer = new( std::nothrow ) uint8_t[base_buffer_size];
-  if( base_buffer == 0 ) { pp( "not enough memory" ); fatal(); }
+  if( base_buffer == 0 ) { pp( "Not enough memory" ); fatal(); }
   uint8_t * const buffer = base_buffer + tsize;
 
   int size = readblock( infd, buffer, buffer_size + hsize ) - hsize;
   bool at_stream_end = ( size < buffer_size );
   if( size != buffer_size && errno )
-    { pp(); show_error( "read error", errno ); fatal(); }
+    { pp(); show_error( "Read error", errno ); fatal(); }
   if( size <= tsize || buffer[0] != 'L' || buffer[1] != 'Z' ||
       buffer[2] != 'I' || buffer[3] != 'P' )
-    { pp( "bad magic number (file not in lzip format)" ); fatal(); }
+    { pp( "Bad magic number (file not in lzip format)" ); fatal(); }
 
   long long partial_member_size = 0;
   while( true )
@@ -239,7 +253,7 @@ extern "C" void * dsplitter( void * arg )
         if( partial_member_size + newpos - pos == member_size )
           {						// header found
           uint8_t * data = new( std::nothrow ) uint8_t[newpos - pos];
-          if( data == 0 ) { pp( "not enough memory" ); fatal(); }
+          if( data == 0 ) { pp( "Not enough memory" ); fatal(); }
           std::memcpy( data, buffer + pos, newpos - pos );
           courier.receive_packet( data, newpos - pos );
           courier.receive_packet( 0, 0 );	// end of member token
@@ -251,7 +265,7 @@ extern "C" void * dsplitter( void * arg )
     if( at_stream_end )
       {
       uint8_t * data = new( std::nothrow ) uint8_t[size + hsize - pos];
-      if( data == 0 ) { pp( "not enough memory" ); fatal(); }
+      if( data == 0 ) { pp( "Not enough memory" ); fatal(); }
       std::memcpy( data, buffer + pos, size + hsize - pos );
       courier.receive_packet( data, size + hsize - pos );
       courier.receive_packet( 0, 0 );		// end of member token
@@ -261,7 +275,7 @@ extern "C" void * dsplitter( void * arg )
       {
       partial_member_size += buffer_size - pos;
       uint8_t * data = new( std::nothrow ) uint8_t[buffer_size - pos];
-      if( data == 0 ) { pp( "not enough memory" ); fatal(); }
+      if( data == 0 ) { pp( "Not enough memory" ); fatal(); }
       std::memcpy( data, buffer + pos, buffer_size - pos );
       courier.receive_packet( data, buffer_size - pos );
       }
@@ -269,7 +283,7 @@ extern "C" void * dsplitter( void * arg )
     size = readblock( infd, buffer + hsize, buffer_size );
     at_stream_end = ( size < buffer_size );
     if( size != buffer_size && errno )
-      { pp(); show_error( "read error", errno ); fatal(); }
+      { pp(); show_error( "Read error", errno ); fatal(); }
     }
   delete[] base_buffer;
   courier.finish();			// no more packets to send
@@ -282,7 +296,6 @@ struct Worker_arg
   Packet_courier * courier;
   const Pretty_print * pp;
   int worker_id;
-  int packet_size;
   };
 
 
@@ -294,12 +307,12 @@ extern "C" void * dworker( void * arg )
   Packet_courier & courier = *tmp.courier;
   const Pretty_print & pp = *tmp.pp;
   const int worker_id = tmp.worker_id;
-  const int new_data_size = tmp.packet_size;
+  const int new_data_size = max_packet_size;
 
   uint8_t * new_data = new( std::nothrow ) uint8_t[new_data_size];
   LZ_Decoder * const decoder = LZ_decompress_open();
   if( !new_data || !decoder || LZ_decompress_errno( decoder ) != LZ_ok )
-    { pp( "not enough memory" ); fatal(); }
+    { pp( "Not enough memory" ); fatal(); }
   int new_pos = 0;
 
   while( true )
@@ -345,7 +358,7 @@ extern "C" void * dworker( void * arg )
             courier.collect_packet( opacket, worker_id );
             new_pos = 0;
             new_data = new( std::nothrow ) uint8_t[new_data_size];
-            if( new_data == 0 ) { pp( "not enough memory" ); fatal(); }
+            if( new_data == 0 ) { pp( "Not enough memory" ); fatal(); }
             }
           if( LZ_decompress_finished( decoder ) == 1 )
             {
@@ -367,7 +380,7 @@ extern "C" void * dworker( void * arg )
 
   delete[] new_data;
   if( LZ_decompress_total_in_size( decoder ) != 0 )
-    { pp( "error, remaining data in decoder" ); fatal(); }
+    { pp( "Error, some data remains in decoder" ); fatal(); }
   LZ_decompress_close( decoder );
   return 0;
   }
@@ -388,7 +401,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;
@@ -406,33 +419,30 @@ int decompress( const int num_workers, const int num_slots,
   {
   in_size = 0;
   out_size = 0;
-  const int packet_size = 1 << 20;
   Packet_courier courier( num_workers, num_slots );
 
   Splitter_arg splitter_arg;
   splitter_arg.courier = &courier;
   splitter_arg.pp = &pp;
   splitter_arg.infd = infd;
-  splitter_arg.packet_size = packet_size;
 
   pthread_t splitter_thread;
   int errcode = pthread_create( &splitter_thread, 0, dsplitter, &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_args = new( std::nothrow ) Worker_arg[num_workers];
   pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers];
   if( worker_args == 0 || worker_threads == 0 )
-    { pp( "not enough memory" ); fatal(); }
+    { pp( "Not enough memory" ); fatal(); }
   for( int i = 0; i < num_workers; ++i )
     {
     worker_args[i].courier = &courier;
     worker_args[i].pp = &pp;
     worker_args[i].worker_id = i;
-    worker_args[i].packet_size = packet_size;
     errcode = pthread_create( &worker_threads[i], 0, dworker, &worker_args[i] );
     if( errcode )
-      { show_error( "can't create worker threads", errcode ); fatal(); }
+      { show_error( "Can't create worker threads", errcode ); fatal(); }
     }
 
   muxer( courier, pp, outfd );
@@ -441,14 +451,14 @@ int decompress( const int num_workers, const int num_slots,
     {
     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;
   delete[] worker_args; worker_args = 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 >= 2 )
     std::fprintf( stderr, "decompressed size %9lld, size %9lld.  ",