Merging upstream version 0.13.
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
parent
dea1c6852b
commit
ab79495362
21 changed files with 280 additions and 225 deletions
150
list_lz.cc
150
list_lz.cc
|
@ -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;
|
||||
}
|
||||
}
|
||||
extended.fill_from_ustar( header ); // copy metadata from header
|
||||
|
||||
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();
|
||||
const bool skip = check_skip_filename( parser, name_pending,
|
||||
extended.path().c_str(), 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; }
|
||||
}
|
||||
|
||||
if( extended.file_size() == 0 &&
|
||||
( typeflag == tf_regular || typeflag == tf_hiperf ) )
|
||||
extended.file_size( parse_octal( header + size_o, size_l ) );
|
||||
|
||||
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
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue