diff options
Diffstat (limited to 'create_lz.cc')
| -rw-r--r-- | create_lz.cc | 560 |
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 ); + } |