summaryrefslogtreecommitdiffstats
path: root/create_lz.cc
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2019-02-14 05:12:04 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2019-02-14 05:12:04 +0000
commite9e3fad677df4b5329912c4dd611a8de620f15cb (patch)
tree3bc1ab40775cf56a94d8b9cd4ce14a71111b8545 /create_lz.cc
parentAdding upstream version 0.10a. (diff)
downloadtarlz-e9e3fad677df4b5329912c4dd611a8de620f15cb.tar.xz
tarlz-e9e3fad677df4b5329912c4dd611a8de620f15cb.zip
Adding upstream version 0.11.upstream/0.11
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'create_lz.cc')
-rw-r--r--create_lz.cc560
1 files changed, 560 insertions, 0 deletions
diff --git a/create_lz.cc b/create_lz.cc
new file mode 100644
index 0000000..48c6a3a
--- /dev/null
+++ b/create_lz.cc
@@ -0,0 +1,560 @@
+/* Tarlz - Archiver with multimember lzip compression
+ Copyright (C) 2013-2019 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 <climits>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <queue>
+#include <string>
+#include <vector>
+#include <pthread.h>
+#include <stdint.h>
+#include <unistd.h>
+#include <sys/stat.h>
+#include <ftw.h>
+#include <lzlib.h>
+
+#include "arg_parser.h"
+#include "tarlz.h"
+
+
+namespace {
+
+enum { max_packet_size = 1 << 20 };
+class Packet_courier;
+Packet_courier * courierp = 0; // local vars needed by add_member
+unsigned long long partial_data_size = 0; // size of current block
+
+
+struct Ipacket // filename, file size and headers
+ {
+ const unsigned 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 unsigned long long s,
+ const Extended * const ext, const uint8_t * const head )
+ : file_size( s ), 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.size() )
+ 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.size() ) 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( const char * const filename, const struct stat *,
+ const int flag, struct FTW * )
+ {
+ unsigned long long file_size = 0;
+ // 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; }
+
+ if( solidity == bsolid &&
+ block_is_full( *extended, file_size, partial_data_size ) )
+ courierp->receive_packet( new Ipacket ); // end of group
+
+ courierp->receive_packet( new Ipacket( filename, file_size, extended, header ) );
+
+ if( 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
+ {
+ Packet_courier * courier;
+ const Arg_parser * parser;
+ };
+
+
+/* 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;
+ Packet_courier & courier = *tmp.courier;
+ const Arg_parser & parser = *tmp.parser;
+
+ for( int i = 0; i < parser.arguments(); ++i ) // parse command line
+ {
+ const int code = parser.code( i );
+ const std::string & arg = parser.argument( i );
+ const char * filename = arg.c_str();
+ if( code == 'C' && chdir( filename ) != 0 )
+ { show_file_error( filename, "Error changing working directory", errno );
+ cleanup_and_fail(); }
+ if( code ) continue; // skip options
+ if( 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(); }
+ struct stat st;
+ if( lstat( filename, &st ) != 0 ) // filename from command line
+ { show_file_error( filename, "Can't stat input file", errno );
+ set_error_status( 1 ); }
+ else if( nftw( filename, add_member, 16, FTW_PHYS ) != 0 )
+ cleanup_and_fail(); // write error or oom
+ else if( solidity == dsolid ) // end of group
+ courier.receive_packet( new Ipacket );
+ }
+
+ if( 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;
+ }
+
+
+/* Writes ibuf to encoder. To minimize dictionary size, it does not read
+ from encoder until encoder's input buffer is full or finish is true.
+ Sends opacket to courier and allocates 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 ) ) );
+ cleanup_and_fail();
+ }
+ 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 ); cleanup_and_fail(); }
+ if( LZ_compress_finished( encoder ) == 1 ) break;
+ }
+ }
+ 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( 2 * header_size ); // extended header + data
+ if( !rbuf.size() ) { show_error( mem_msg2 ); cleanup_and_fail(); }
+
+ int opos = 0;
+ 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, flush encoder
+ {
+ if( !encoder ) { delete ipacket; continue; } // nothing to flush
+ loop_encode( 0, 0, data, opos, courier, encoder, worker_id, true );
+ courier.collect_packet( new Opacket, worker_id ); // end of member token
+ if( LZ_compress_restart_member( encoder, LLONG_MAX ) < 0 )
+ { show_error( "LZ_compress_restart_member failed." ); cleanup_and_fail(); }
+ delete ipacket; continue;
+ }
+
+ const int infd =
+ ipacket->file_size ? open_instream( ipacket->filename.c_str() ) : -1;
+ if( ipacket->file_size && infd < 0 )
+ { delete[] ipacket->header; delete ipacket->extended; delete ipacket;
+ set_error_status( 1 ); continue; }
+
+ if( !encoder )
+ {
+ 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." );
+ cleanup_and_fail();
+ }
+ }
+
+ if( !ipacket->extended->empty() ) // compress extended block
+ {
+ const long long ebsize = ipacket->extended->format_block( rbuf );
+ if( ebsize < 0 )
+ { show_error( "Error formatting extended records." ); cleanup_and_fail(); }
+ /* Limit the size of the extended block to INT_MAX - 1 so that it can
+ be fed to lzlib as one buffer. */
+ if( ebsize >= INT_MAX )
+ { show_error( "Extended records size >= INT_MAX." ); cleanup_and_fail(); }
+ loop_encode( (const uint8_t *)rbuf(), 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 )
+ {
+ enum { bufsize = 32 * header_size };
+ uint8_t buf[bufsize];
+ unsigned long long rest = ipacket->file_size;
+ while( rest > 0 )
+ {
+ int size = std::min( rest, (unsigned long long)bufsize );
+ const int rd = readblock( infd, buf, size );
+ rest -= rd;
+ if( rd != size )
+ {
+ if( verbosity >= 0 )
+ std::fprintf( stderr, "File '%s' ends unexpectedly at pos %llu\n",
+ ipacket->filename.c_str(), ipacket->file_size - rest );
+ close( infd ); cleanup_and_fail();
+ }
+ 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( ipacket->filename.c_str(), "Error closing file", errno );
+ cleanup_and_fail(); }
+ }
+ delete ipacket;
+ }
+ if( data ) delete[] data;
+ if( encoder && LZ_compress_close( encoder ) < 0 )
+ { show_error( "LZ_compress_close failed." ); cleanup_and_fail(); }
+ return 0;
+ }
+
+
+/* Get from courier the processed and sorted packets, and write
+ their contents to the output archive. */
+bool muxer( Packet_courier & courier, const char * const archive_name,
+ const int outfd )
+ {
+ while( true )
+ {
+ const Opacket * const opacket = courier.deliver_packet();
+ if( !opacket ) break; // queue is empty. all workers exited
+
+ const int wr = writeblock( outfd, opacket->data, opacket->size );
+ if( wr != opacket->size )
+ { show_file_error( archive_name, "Write error", errno ); return false; }
+ delete[] opacket->data;
+ delete opacket;
+ }
+ return true;
+ }
+
+} // end namespace
+
+
+// init the courier, then start the grouper and the workers and call the muxer
+int encode_lz( const char * const archive_name, const Arg_parser & parser,
+ const int dictionary_size, const int match_len_limit,
+ const int num_workers, const int outfd, const int debug_level )
+ {
+ const int in_slots = 65536; // max small files (<=512B) in 64 MiB
+ const int total_in_slots = ( INT_MAX / num_workers >= in_slots ) ?
+ num_workers * in_slots : INT_MAX;
+ const int out_slots = 64;
+
+ Packet_courier courier( num_workers, total_in_slots, out_slots );
+ courierp = &courier; // needed by add_member
+
+ Grouper_arg grouper_arg;
+ grouper_arg.courier = &courier;
+ grouper_arg.parser = &parser;
+
+ 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 ); return 1; }
+ 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 ); return 1; }
+ }
+
+ if( !muxer( courier, archive_name, outfd ) ) return 1;
+
+ 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 ); return 1; }
+ }
+ delete[] worker_threads;
+ delete[] worker_args;
+
+ errcode = pthread_join( grouper_thread, 0 );
+ if( errcode )
+ { show_error( "Can't join grouper thread", errcode ); return 1; }
+
+ // write End-Of-Archive records
+ int retval = 0;
+ enum { eof_member_size = 44 };
+ const uint8_t eof_member[eof_member_size] = {
+ 0x4C, 0x5A, 0x49, 0x50, 0x01, 0x0C, 0x00, 0x00, 0x6F, 0xFD, 0xFF, 0xFF,
+ 0xA3, 0xB7, 0x80, 0x0C, 0x82, 0xDB, 0xFF, 0xFF, 0x9F, 0xF0, 0x00, 0x00,
+ 0x2E, 0xAF, 0xB5, 0xEF, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+ 0x2C, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
+ if( writeblock( outfd, eof_member, eof_member_size ) != eof_member_size )
+ { show_error( "Error writing end-of-archive blocks", errno );
+ retval = 1; }
+
+ if( close( outfd ) != 0 && !retval )
+ { show_error( "Error closing archive", errno ); retval = 1; }
+
+ if( 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( "courier not finished." );
+ return final_exit_status( retval );
+ }