diff options
Diffstat (limited to 'list_lz.cc')
| -rw-r--r-- | list_lz.cc | 150 |
1 files changed, 71 insertions, 79 deletions
@@ -145,13 +145,32 @@ void xbroadcast( pthread_cond_t * const cond ) } +bool check_skip_filename( const Arg_parser & parser, + std::vector< char > & name_pending, + const char * const filename, const int filenames ) + { + bool skip = filenames > 0; + if( skip ) + for( int i = 0; i < parser.arguments(); ++i ) + if( !parser.code( i ) && parser.argument( i ).size() ) + { + const char * const name = + remove_leading_dotslash( parser.argument( i ).c_str() ); + if( compare_prefix_dir( name, filename ) || + compare_tslash( name, filename ) ) + { skip = false; name_pending[i] = false; break; } + } + return skip; + } + + namespace { struct Packet // member name and metadata or error message { enum Status { ok, member_done, error }; long member_id; // lzip member containing the header of this tar member - std::string line; // member name and metadata ready to print + std::string line; // member name and metadata ready to print, if any Status status; Packet( const long i, const char * const msg, const Status s = ok ) : member_id( i ), line( msg ), status( s ) {} @@ -167,7 +186,7 @@ private: long error_member_id; // first lzip member with error/misalign/eof int deliver_worker_id; // worker queue currently delivering packets int master_worker_id; // worker in charge if error/misalignment/eof - std::vector< std::queue< Packet * > > opacket_queues; + std::vector< std::queue< const Packet * > > 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 @@ -233,8 +252,9 @@ public: xunlock( &omutex ); } - // collect a packet from a worker - bool collect_packet( Packet * const opacket, const int worker_id ) + /* Collect a packet from a worker. + If a packet is rejected, the worker must terminate. */ + bool collect_packet( const Packet * const opacket, const int worker_id ) { xlock( &omutex ); if( ( mastership_granted() && master_worker_id != worker_id ) || @@ -249,27 +269,31 @@ public: } /* Deliver a packet to muxer. - If packet.status == Packet::member_done, move to next queue. */ - Packet * deliver_packet() + If packet.status == Packet::member_done, move to next queue. + If packet.line.empty(), wait again (empty lzip member). */ + const Packet * deliver_packet() { - Packet * opacket = 0; + const Packet * opacket = 0; xlock( &omutex ); ++ocheck_counter; - while( opacket_queues[deliver_worker_id].empty() && num_working > 0 ) - { - ++owait_counter; - if( !mastership_granted() && error_member_id >= 0 ) - xbroadcast( &check_master ); // mastership requested not yet granted - xwait( &oav_or_exit, &omutex ); - } - if( !opacket_queues[deliver_worker_id].empty() ) + while( true ) { + while( opacket_queues[deliver_worker_id].empty() && num_working > 0 ) + { + ++owait_counter; + if( !mastership_granted() && error_member_id >= 0 ) + xbroadcast( &check_master ); // mastership requested not yet granted + 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->status == Packet::member_done && !mastership_granted() ) { if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0; } + if( !opacket->line.empty() ) break; + delete opacket; opacket = 0; } xunlock( &omutex ); return opacket; @@ -349,14 +373,15 @@ int list_member_lz( LZ_Decoder * const decoder, const int infd, if( data_rest < 0 ) // tar member exceeds lzip member end { if( courier.request_mastership( member_id, worker_id ) ) master = true; - else return 2; + else { *msg = "tar member exceeds lzip member end"; return 2; } } if( verbosity < 0 || skip ) rbuf()[0] = 0; else format_member_name( extended, header, rbuf, verbosity > 0 ); - Packet * const opacket = new Packet( member_id, rbuf(), - data_rest ? Packet::ok : Packet::member_done ); - courier.collect_packet( opacket, worker_id ); + const Packet * const opacket = new Packet( member_id, rbuf(), + data_rest ? Packet::ok : Packet::member_done ); + if( !courier.collect_packet( opacket, worker_id ) ) + { *msg = "other worker found an error"; return 1; } if( !data_rest ) { data_pos = mdata_end; return 0; } const unsigned bufsize = 32 * header_size; @@ -441,6 +466,12 @@ extern "C" void * tworker( void * arg ) long long data_end = mdata_end; long long file_pos = lzip_index.mblock( i ).pos(); long long member_end = lzip_index.mblock( i ).end(); + if( data_pos >= data_end ) // empty lzip member + { + const Packet * const opacket = new Packet( i, "", Packet::member_done ); + if( !courier.collect_packet( opacket, worker_id ) ) goto done; + continue; + } Extended extended; // metadata from extended records int retval = 0; @@ -459,7 +490,7 @@ extern "C" void * tworker( void * arg ) master = true; if( ret > 0 ) { - Packet * const opacket = new Packet( i, msg, Packet::error ); + const Packet * const opacket = new Packet( i, msg, Packet::error ); courier.collect_packet( opacket, worker_id ); goto done; } @@ -472,7 +503,7 @@ extern "C" void * tworker( void * arg ) if( !courier.request_mastership( i, worker_id ) ) goto done; master = true; if( block_is_zero( header, header_size ) ) break; // EOF - Packet * const opacket = new Packet( i, + const Packet * const opacket = new Packet( i, ( data_pos > header_size ) ? "Corrupt or invalid header." : "This does not look like a POSIX tar.lz archive.", Packet::error ); courier.collect_packet( opacket, worker_id ); @@ -495,9 +526,9 @@ extern "C" void * tworker( void * arg ) if( ret > 0 ) { if( !msg ) msg = "Error in global extended records."; - Packet * const opacket = new Packet( i, msg, Packet::error ); + const Packet * const opacket = new Packet( i, msg, Packet::error ); courier.collect_packet( opacket, worker_id ); - if( ret == 2 ) goto done; + goto done; } // member_end exceeded, process rest of file else { data_end = lzip_index.udata_size(); member_end = cdata_size; } @@ -521,10 +552,9 @@ extern "C" void * tworker( void * arg ) if( ret > 0 ) { if( !msg ) msg = "Error in extended records."; - Packet * const opacket = new Packet( i, msg, Packet::error ); + const Packet * const opacket = new Packet( i, msg, Packet::error ); courier.collect_packet( opacket, worker_id ); - extended.reset(); - if( ret == 2 ) goto done; + goto done; } // member_end exceeded, process rest of file else { data_end = lzip_index.udata_size(); member_end = cdata_size; } @@ -534,70 +564,33 @@ extern "C" void * tworker( void * arg ) } prev_extended = false; - if( extended.linkpath().empty() ) // copy linkpath from ustar header - { - int len = 0; - while( len < linkname_l && header[linkname_o+len] ) ++len; - while( len > 1 && header[linkname_o+len-1] == '/' ) --len; // trailing '/' - if( len > 0 ) - { - const uint8_t c = header[linkname_o+len]; header[linkname_o+len] = 0; - extended.linkpath( (const char *)header + linkname_o ); - header[linkname_o+len] = c; - } - } - - if( extended.path().empty() ) // copy path from ustar header - { - char stored_name[prefix_l+1+name_l+1]; - int len = 0; - while( len < prefix_l && header[prefix_o+len] ) - { stored_name[len] = header[prefix_o+len]; ++len; } - if( len && header[name_o] ) stored_name[len++] = '/'; - for( int i = 0; i < name_l && header[name_o+i]; ++i ) - { stored_name[len] = header[name_o+i]; ++len; } - while( len > 0 && stored_name[len-1] == '/' ) --len; // trailing '/' - stored_name[len] = 0; - extended.path( remove_leading_dotslash( stored_name ) ); - } - const char * const filename = extended.path().c_str(); - - bool skip = filenames > 0; - if( skip ) - for( int i = 0; i < parser.arguments(); ++i ) - if( !parser.code( i ) && parser.argument( i ).size() ) - { - const char * const name = - remove_leading_dotslash( parser.argument( i ).c_str() ); - if( compare_prefix_dir( name, filename ) || - compare_tslash( name, filename ) ) - { skip = false; name_pending[i] = false; break; } - } + extended.fill_from_ustar( header ); // copy metadata from header - if( extended.file_size() == 0 && - ( typeflag == tf_regular || typeflag == tf_hiperf ) ) - extended.file_size( parse_octal( header + size_o, size_l ) ); + const bool skip = check_skip_filename( parser, name_pending, + extended.path().c_str(), filenames ); - retval = list_member_lz( decoder, infd, file_pos, member_end, - cdata_size, data_pos, mdata_end, courier, - extended, header, rbuf, i, worker_id, &msg, skip ); + retval = list_member_lz( decoder, infd, file_pos, member_end, cdata_size, + data_pos, mdata_end, courier, extended, + header, rbuf, i, worker_id, &msg, skip ); extended.reset(); if( retval < 0 ) // member_end exceeded, process rest of file { master = true; data_end = lzip_index.udata_size(); member_end = cdata_size; } else if( retval > 0 ) - { show_error( msg ); - show_error( "Error is not recoverable: exiting now." ); - cleanup_and_fail( 2 ); } + { + const Packet * const opacket = new Packet( i, msg, Packet::error ); + courier.collect_packet( opacket, worker_id ); + goto done; + } } } +done: if( LZ_decompress_close( decoder ) < 0 ) { - Packet * const opacket = new Packet( lzip_index.members(), + const Packet * const opacket = new Packet( lzip_index.members(), "LZ_decompress_close failed.", Packet::error ); courier.collect_packet( opacket, worker_id ); } -done: courier.worker_finished(); return 0; } @@ -609,14 +602,13 @@ void muxer( Packet_courier & courier ) { while( true ) { - Packet * const opacket = courier.deliver_packet(); + const Packet * const opacket = courier.deliver_packet(); if( !opacket ) break; // queue is empty. all workers exited if( opacket->status == Packet::error ) { show_error( opacket->line.c_str() ); cleanup_and_fail( 2 ); } if( opacket->line.size() ) - { std::fputs( opacket->line.c_str(), stdout ); - std::fflush( stdout ); } + { std::fputs( opacket->line.c_str(), stdout ); std::fflush( stdout ); } delete opacket; } if( !courier.mastership_granted() ) // no worker found EOF blocks |