Adding upstream version 0.25.upstream/0.25upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 594 insertions, 0 deletions

diff --git a/create_lz.cc b/create_lz.cc
new file mode 100644
index 0000000..5436bf5
--- /dev/null
+++ b/create_lz.cc
@@ -0,0 +1,594 @@
+/* Tarlz - Archiver with multimember lzip compression
+   Copyright (C) 2013-2024 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 2 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 <cerrno>
+#include <cstdio>
+#include <queue>
+#include <pthread.h>
+#include <stdint.h>	// for lzlib.h
+#include <unistd.h>
+#include <sys/stat.h>
+#include <ftw.h>
+#include <lzlib.h>
+
+#include "tarlz.h"
+#include "arg_parser.h"
+#include "common_mutex.h"
+#include "create.h"
+
+
+namespace {
+
+const Cl_options * gcl_opts = 0;	// local vars needed by add_member_lz
+enum { max_packet_size = 1 << 20 };
+class Packet_courier;
+Packet_courier * courierp = 0;
+unsigned long long partial_data_size = 0;	// size of current block
+
+
+class Slot_tally
+  {
+  const int num_slots;				// total slots
+  int num_free;					// remaining free slots
+  pthread_mutex_t mutex;
+  pthread_cond_t slot_av;			// slot available
+
+  Slot_tally( const Slot_tally & );		// declared as private
+  void operator=( const Slot_tally & );		// declared as private
+
+public:
+  explicit Slot_tally( const int slots )
+    : num_slots( slots ), num_free( slots )
+    { xinit_mutex( &mutex ); xinit_cond( &slot_av ); }
+
+  ~Slot_tally() { xdestroy_cond( &slot_av ); xdestroy_mutex( &mutex ); }
+
+  bool all_free() { return ( num_free == num_slots ); }
+
+  void get_slot()				// wait for a free slot
+    {
+    xlock( &mutex );
+    while( num_free <= 0 ) xwait( &slot_av, &mutex );
+    --num_free;
+    xunlock( &mutex );
+    }
+
+  void leave_slot()				// return a slot to the tally
+    {
+    xlock( &mutex );
+    if( ++num_free == 1 ) xsignal( &slot_av );	// num_free was 0
+    xunlock( &mutex );
+    }
+  };
+
+
+struct Ipacket			// filename, file size and headers
+  {
+  const long long file_size;
+  const std::string filename;	// filename.empty() means end of lzip member
+  const Extended * const extended;
+  const uint8_t * const header;
+
+  Ipacket() : file_size( 0 ), extended( 0 ), header( 0 ) {}
+  Ipacket( const char * const name, const long long fs,
+           const Extended * const ext, const uint8_t * const head )
+    : file_size( fs ), filename( name ), extended( ext ), header( head ) {}
+  };
+
+struct Opacket		// compressed data to be written to the archive
+  {
+  const uint8_t * const data;	// data == 0 means end of lzip member
+  const int size;		// number of bytes in data (if any)
+
+  Opacket() : data( 0 ), size( 0 ) {}
+  Opacket( uint8_t * const d, const int s ) : data( d ), size( s ) {}
+  };
+
+
+class Packet_courier			// moves packets around
+  {
+public:
+  unsigned icheck_counter;
+  unsigned iwait_counter;
+  unsigned ocheck_counter;
+  unsigned owait_counter;
+private:
+  int receive_worker_id;	// worker queue currently receiving packets
+  int deliver_worker_id;	// worker queue currently delivering packets
+  Slot_tally slot_tally;		// limits the number of input packets
+  std::vector< std::queue< const Ipacket * > > ipacket_queues;
+  std::vector< std::queue< const Opacket * > > opacket_queues;
+  int num_working;			// number of workers still running
+  const int num_workers;		// number of workers
+  const unsigned out_slots;		// max output packets per queue
+  pthread_mutex_t imutex;
+  pthread_cond_t iav_or_eof;	// input packet available or grouper done
+  pthread_mutex_t omutex;
+  pthread_cond_t oav_or_exit;	// output packet available or all workers exited
+  std::vector< pthread_cond_t > slot_av;	// output slot available
+  bool eof;					// grouper done
+
+  Packet_courier( const Packet_courier & );	// declared as private
+  void operator=( const Packet_courier & );	// declared as private
+
+public:
+  Packet_courier( const int workers, const int in_slots, const int oslots )
+    : icheck_counter( 0 ), iwait_counter( 0 ),
+      ocheck_counter( 0 ), owait_counter( 0 ),
+      receive_worker_id( 0 ), deliver_worker_id( 0 ),
+      slot_tally( in_slots ), ipacket_queues( workers ),
+      opacket_queues( workers ), num_working( workers ),
+      num_workers( workers ), out_slots( oslots ), slot_av( workers ),
+      eof( false )
+    {
+    xinit_mutex( &imutex ); xinit_cond( &iav_or_eof );
+    xinit_mutex( &omutex ); xinit_cond( &oav_or_exit );
+    for( unsigned i = 0; i < slot_av.size(); ++i ) xinit_cond( &slot_av[i] );
+    }
+
+  ~Packet_courier()
+    {
+    for( unsigned i = 0; i < slot_av.size(); ++i ) xdestroy_cond( &slot_av[i] );
+    xdestroy_cond( &oav_or_exit ); xdestroy_mutex( &omutex );
+    xdestroy_cond( &iav_or_eof ); xdestroy_mutex( &imutex );
+    }
+
+  /* Receive an ipacket from grouper.
+     If filename.empty() (end of lzip member token), move to next queue. */
+  void receive_packet( const Ipacket * const ipacket )
+    {
+    if( !ipacket->filename.empty() )
+      slot_tally.get_slot();		// wait for a free slot
+    xlock( &imutex );
+    ipacket_queues[receive_worker_id].push( ipacket );
+    if( ipacket->filename.empty() && ++receive_worker_id >= num_workers )
+      receive_worker_id = 0;
+    xbroadcast( &iav_or_eof );
+    xunlock( &imutex );
+    }
+
+  // distribute an ipacket to a worker
+  const Ipacket * distribute_packet( const int worker_id )
+    {
+    const Ipacket * ipacket = 0;
+    xlock( &imutex );
+    ++icheck_counter;
+    while( ipacket_queues[worker_id].empty() && !eof )
+      {
+      ++iwait_counter;
+      xwait( &iav_or_eof, &imutex );
+      }
+    if( !ipacket_queues[worker_id].empty() )
+      {
+      ipacket = ipacket_queues[worker_id].front();
+      ipacket_queues[worker_id].pop();
+      }
+    xunlock( &imutex );
+    if( ipacket )
+      { if( !ipacket->filename.empty() ) slot_tally.leave_slot(); }
+    else
+      {
+      // notify muxer when last worker exits
+      xlock( &omutex );
+      if( --num_working == 0 ) xsignal( &oav_or_exit );
+      xunlock( &omutex );
+      }
+    return ipacket;
+    }
+
+  // collect an opacket from a worker
+  void collect_packet( const Opacket * const opacket, const int worker_id )
+    {
+    xlock( &omutex );
+    if( opacket->data )
+      {
+      while( opacket_queues[worker_id].size() >= out_slots )
+        xwait( &slot_av[worker_id], &omutex );
+      }
+    opacket_queues[worker_id].push( opacket );
+    if( worker_id == deliver_worker_id ) xsignal( &oav_or_exit );
+    xunlock( &omutex );
+    }
+
+  /* Deliver an opacket to muxer.
+     If opacket data == 0, move to next queue and wait again. */
+  const Opacket * deliver_packet()
+    {
+    const Opacket * opacket = 0;
+    xlock( &omutex );
+    ++ocheck_counter;
+    while( true )
+      {
+      while( opacket_queues[deliver_worker_id].empty() && num_working > 0 )
+        {
+        ++owait_counter;
+        xwait( &oav_or_exit, &omutex );
+        }
+      if( opacket_queues[deliver_worker_id].empty() ) break;
+      opacket = opacket_queues[deliver_worker_id].front();
+      opacket_queues[deliver_worker_id].pop();
+      if( opacket_queues[deliver_worker_id].size() + 1 == out_slots )
+        xsignal( &slot_av[deliver_worker_id] );
+      if( opacket->data ) break;
+      if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0;
+      delete opacket; opacket = 0;
+      }
+    xunlock( &omutex );
+    return opacket;
+    }
+
+  void finish()			// grouper has no more packets to send
+    {
+    xlock( &imutex );
+    eof = true;
+    xbroadcast( &iav_or_eof );
+    xunlock( &imutex );
+    }
+
+  bool finished()		// all packets delivered to muxer
+    {
+    if( !slot_tally.all_free() || !eof || num_working != 0 ) return false;
+    for( int i = 0; i < num_workers; ++i )
+      if( !ipacket_queues[i].empty() ) return false;
+    for( int i = 0; i < num_workers; ++i )
+      if( !opacket_queues[i].empty() ) return false;
+    return true;
+    }
+  };
+
+
+// send one ipacket with tar member metadata to courier
+int add_member_lz( const char * const filename, const struct stat *,
+                   const int flag, struct FTW * )
+  {
+  if( Exclude::excluded( filename ) ) return 0;		// skip excluded files
+  long long file_size;
+  // metadata for extended records
+  Extended * const extended = new( std::nothrow ) Extended;
+  uint8_t * const header = extended ? new( std::nothrow ) Tar_header : 0;
+  if( !header )
+    { show_error( mem_msg ); if( extended ) delete extended; return 1; }
+  if( !fill_headers( filename, *extended, header, file_size, flag ) )
+    { delete[] header; delete extended; return 0; }
+  print_removed_prefix( extended->removed_prefix );
+
+  if( gcl_opts->solidity == bsolid )
+    {
+    const int ebsize = extended->full_size();
+    if( ebsize < 0 ) { show_error( extended->full_size_error() ); return 1; }
+    if( block_is_full( ebsize, file_size, gcl_opts->data_size,
+                       partial_data_size ) )
+      courierp->receive_packet( new Ipacket );		// end of group
+    }
+  courierp->receive_packet( new Ipacket( filename, file_size, extended, header ) );
+
+  if( gcl_opts->solidity == no_solid )		// one tar member per group
+    courierp->receive_packet( new Ipacket );
+  if( verbosity >= 1 ) std::fprintf( stderr, "%s\n", filename );
+  return 0;
+  }
+
+
+struct Grouper_arg
+  {
+  const Cl_options * cl_opts;
+  Packet_courier * courier;
+  };
+
+
+/* Package metadata of the files to be archived and pass them to the
+   courier for distribution to workers.
+*/
+extern "C" void * grouper( void * arg )
+  {
+  const Grouper_arg & tmp = *(const Grouper_arg *)arg;
+  const Cl_options & cl_opts = *tmp.cl_opts;
+  Packet_courier & courier = *tmp.courier;
+
+  for( int i = 0; i < cl_opts.parser.arguments(); ++i )	// parse command line
+    {
+    const int code = cl_opts.parser.code( i );
+    const std::string & arg = cl_opts.parser.argument( i );
+    const char * filename = arg.c_str();
+    if( code == 'C' && chdir( filename ) != 0 )
+      { show_file_error( filename, chdir_msg, errno ); exit_fail_mt(); }
+    if( code ) continue;				// skip options
+    if( cl_opts.parser.argument( i ).empty() ) continue; // skip empty names
+    std::string deslashed;		// arg without trailing slashes
+    unsigned len = arg.size();
+    while( len > 1 && arg[len-1] == '/' ) --len;
+    if( len < arg.size() )
+      { deslashed.assign( arg, 0, len ); filename = deslashed.c_str(); }
+    if( Exclude::excluded( filename ) ) continue;	// skip excluded files
+    struct stat st;
+    if( lstat( filename, &st ) != 0 )	// filename from command line
+      { show_file_error( filename, cant_stat, errno ); set_error_status( 1 ); }
+    else if( nftw( filename, add_member_lz, 16,
+                   cl_opts.dereference ? 0 : FTW_PHYS ) != 0 )
+      exit_fail_mt();			// write error or OOM
+    else if( cl_opts.solidity == dsolid )	// end of group
+      courier.receive_packet( new Ipacket );
+    }
+
+  if( cl_opts.solidity == bsolid && partial_data_size )	// finish last block
+    { partial_data_size = 0; courierp->receive_packet( new Ipacket ); }
+  courier.finish();			// no more packets to send
+  return 0;
+  }
+
+
+/* Write ibuf to encoder. To minimize dictionary size, do not read from
+   encoder until encoder's input buffer is full or finish is true.
+   Send opacket to courier and allocate new obuf each time obuf is full.
+*/
+void loop_encode( const uint8_t * const ibuf, const int isize,
+                  uint8_t * & obuf, int & opos, Packet_courier & courier,
+                  LZ_Encoder * const encoder, const int worker_id,
+                  const bool finish = false )
+  {
+  int ipos = 0;
+  if( opos < 0 || opos > max_packet_size )
+    internal_error( "bad buffer index in loop_encode." );
+  while( true )
+    {
+    if( ipos < isize )
+      {
+      const int wr = LZ_compress_write( encoder, ibuf + ipos, isize - ipos );
+      if( wr < 0 ) internal_error( "library error (LZ_compress_write)." );
+      ipos += wr;
+      }
+    if( ipos >= isize )					// ibuf is empty
+      { if( finish ) LZ_compress_finish( encoder ); else break; }
+    const int rd =
+      LZ_compress_read( encoder, obuf + opos, max_packet_size - opos );
+    if( rd < 0 )
+      {
+      if( verbosity >= 0 )
+        std::fprintf( stderr, "LZ_compress_read error: %s\n",
+                      LZ_strerror( LZ_compress_errno( encoder ) ) );
+      exit_fail_mt();
+      }
+    opos += rd;
+    // obuf is full or last opacket in lzip member
+    if( opos >= max_packet_size || LZ_compress_finished( encoder ) == 1 )
+      {
+      if( opos > max_packet_size )
+        internal_error( "opacket size exceeded in worker." );
+      courier.collect_packet( new Opacket( obuf, opos ), worker_id );
+      opos = 0; obuf = new( std::nothrow ) uint8_t[max_packet_size];
+      if( !obuf ) { show_error( mem_msg2 ); exit_fail_mt(); }
+      if( LZ_compress_finished( encoder ) == 1 )
+        {
+        if( LZ_compress_restart_member( encoder, LLONG_MAX ) >= 0 ) break;
+        show_error( "LZ_compress_restart_member failed." ); exit_fail_mt();
+        }
+      }
+    }
+  if( ipos > isize ) internal_error( "ipacket size exceeded in worker." );
+  if( ipos < isize ) internal_error( "input not fully consumed in worker." );
+  }
+
+
+struct Worker_arg
+  {
+  Packet_courier * courier;
+  int dictionary_size;
+  int match_len_limit;
+  int worker_id;
+  };
+
+
+/* Get ipackets from courier, compress headers and file data, and give the
+   opackets produced to courier.
+*/
+extern "C" void * cworker( void * arg )
+  {
+  const Worker_arg & tmp = *(const Worker_arg *)arg;
+  Packet_courier & courier = *tmp.courier;
+  const int dictionary_size = tmp.dictionary_size;
+  const int match_len_limit = tmp.match_len_limit;
+  const int worker_id = tmp.worker_id;
+
+  LZ_Encoder * encoder = 0;
+  uint8_t * data = 0;
+  Resizable_buffer rbuf;			// extended header + data
+  if( !rbuf.size() ) { show_error( mem_msg2 ); exit_fail_mt(); }
+
+  int opos = 0;
+  bool flushed = true;		// avoid producing empty lzip members
+  while( true )
+    {
+    const Ipacket * const ipacket = courier.distribute_packet( worker_id );
+    if( !ipacket ) break;		// no more packets to process
+    if( ipacket->filename.empty() )	// end of group
+      {
+      if( !flushed )			// this lzip member is not empty
+        loop_encode( 0, 0, data, opos, courier, encoder, worker_id, true );
+      courier.collect_packet( new Opacket, worker_id );	// end of member token
+      flushed = true; delete ipacket; continue;
+      }
+
+    const char * const filename = ipacket->filename.c_str();
+    const int infd = ipacket->file_size ? open_instream( filename ) : -1;
+    if( ipacket->file_size && infd < 0 )	// can't read file data
+      { delete[] ipacket->header; delete ipacket->extended; delete ipacket;
+        set_error_status( 1 ); continue; }	// skip file
+
+    flushed = false;
+    if( !encoder )		// init encoder just before using it
+      {
+      data = new( std::nothrow ) uint8_t[max_packet_size];
+      encoder = LZ_compress_open( dictionary_size, match_len_limit, LLONG_MAX );
+      if( !data || !encoder || LZ_compress_errno( encoder ) != LZ_ok )
+        {
+        if( !data || !encoder || LZ_compress_errno( encoder ) == LZ_mem_error )
+          show_error( mem_msg2 );
+        else
+          internal_error( "invalid argument to encoder." );
+        exit_fail_mt();
+        }
+      }
+
+    const int ebsize = ipacket->extended->format_block( rbuf );	// may be 0
+    if( ebsize < 0 )
+      { show_error( ipacket->extended->full_size_error() ); exit_fail_mt(); }
+    if( ebsize > 0 )				// compress extended block
+      loop_encode( rbuf.u8(), ebsize, data, opos, courier, encoder, worker_id );
+    // compress ustar header
+    loop_encode( ipacket->header, header_size, data, opos, courier,
+                 encoder, worker_id );
+    delete[] ipacket->header; delete ipacket->extended;
+
+    if( ipacket->file_size )
+      {
+      const long long bufsize = 32 * header_size;
+      uint8_t buf[bufsize];
+      long long rest = ipacket->file_size;
+      while( rest > 0 )
+        {
+        int size = std::min( rest, bufsize );
+        const int rd = readblock( infd, buf, size );
+        rest -= rd;
+        if( rd != size )
+          {
+          show_atpos_error( filename, ipacket->file_size - rest, false );
+          close( infd ); exit_fail_mt();
+          }
+        if( rest == 0 )				// last read
+          {
+          const int rem = ipacket->file_size % header_size;
+          if( rem > 0 )
+            { const int padding = header_size - rem;
+              std::memset( buf + size, 0, padding ); size += padding; }
+          }
+        // compress size bytes of file
+        loop_encode( buf, size, data, opos, courier, encoder, worker_id );
+        }
+      if( close( infd ) != 0 )
+        { show_file_error( filename, eclosf_msg, errno ); exit_fail_mt(); }
+      }
+    if( gcl_opts->warn_newer && archive_attrs.is_newer( filename ) )
+      { show_file_error( filename, "File is newer than the archive." );
+        set_error_status( 1 ); }
+    delete ipacket;
+    }
+  if( data ) delete[] data;
+  if( encoder && LZ_compress_close( encoder ) < 0 )
+    { show_error( "LZ_compress_close failed." ); exit_fail_mt(); }
+  return 0;
+  }
+
+
+/* Get from courier the processed and sorted packets, and write
+   their contents to the output archive.
+*/
+void muxer( Packet_courier & courier, const int outfd )
+  {
+  while( true )
+    {
+    const Opacket * const opacket = courier.deliver_packet();
+    if( !opacket ) break;	// queue is empty. all workers exited
+
+    if( !writeblock_wrapper( outfd, opacket->data, opacket->size ) )
+      exit_fail_mt();
+    delete[] opacket->data;
+    delete opacket;
+    }
+  }
+
+} // end namespace
+
+
+// init the courier, then start the grouper and the workers and call the muxer
+int encode_lz( const Cl_options & cl_opts, const char * const archive_namep,
+               const int outfd )
+  {
+  const int in_slots = 65536;		// max small files (<=512B) in 64 MiB
+  const int num_workers = cl_opts.num_workers;
+  const int total_in_slots = ( INT_MAX / num_workers >= in_slots ) ?
+                             num_workers * in_slots : INT_MAX;
+  const int dictionary_size = option_mapping[cl_opts.level].dictionary_size;
+  const int match_len_limit = option_mapping[cl_opts.level].match_len_limit;
+  gcl_opts = &cl_opts;
+
+  /* If an error happens after any threads have been started, exit must be
+     called before courier goes out of scope. */
+  Packet_courier courier( num_workers, total_in_slots, cl_opts.out_slots );
+  courierp = &courier;			// needed by add_member_lz
+
+  Grouper_arg grouper_arg;
+  grouper_arg.cl_opts = &cl_opts;
+  grouper_arg.courier = &courier;
+
+  pthread_t grouper_thread;
+  int errcode = pthread_create( &grouper_thread, 0, grouper, &grouper_arg );
+  if( errcode )
+    { show_error( "Can't create grouper thread", errcode ); return 1; }
+
+  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 || !worker_threads )
+    { show_error( mem_msg ); exit_fail_mt(); }
+  for( int i = 0; i < num_workers; ++i )
+    {
+    worker_args[i].courier = &courier;
+    worker_args[i].dictionary_size = dictionary_size;
+    worker_args[i].match_len_limit = match_len_limit;
+    worker_args[i].worker_id = i;
+    errcode = pthread_create( &worker_threads[i], 0, cworker, &worker_args[i] );
+    if( errcode )
+      { show_error( "Can't create worker threads", errcode ); exit_fail_mt(); }
+    }
+
+  muxer( courier, outfd );
+
+  for( int i = num_workers - 1; i >= 0; --i )
+    {
+    errcode = pthread_join( worker_threads[i], 0 );
+    if( errcode )
+      { show_error( "Can't join worker threads", errcode ); exit_fail_mt(); }
+    }
+  delete[] worker_threads;
+  delete[] worker_args;
+
+  errcode = pthread_join( grouper_thread, 0 );
+  if( errcode )
+    { show_error( "Can't join grouper thread", errcode ); exit_fail_mt(); }
+
+  // write End-Of-Archive records
+  int retval = !write_eoa_records( outfd, true );
+
+  if( close( outfd ) != 0 && retval == 0 )
+    { show_file_error( archive_namep, eclosa_msg, errno ); retval = 1; }
+
+  if( cl_opts.debug_level & 1 )
+    std::fprintf( stderr,
+      "any worker tried to consume from grouper %8u times\n"
+      "any worker had to wait                   %8u times\n"
+      "muxer tried to consume from workers      %8u times\n"
+      "muxer had to wait                        %8u times\n",
+      courier.icheck_counter,
+      courier.iwait_counter,
+      courier.ocheck_counter,
+      courier.owait_counter );
+
+  if( !courier.finished() ) internal_error( conofin_msg );
+  return final_exit_status( retval );
+  }