From 0a1a75fc04f29f4418273baa7dcb33c7751e0e4c Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sat, 7 Nov 2015 16:27:21 +0100 Subject: Adding upstream version 0.7. Signed-off-by: Daniel Baumann --- decompress.cc | 90 +++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 50 insertions(+), 40 deletions(-) (limited to 'decompress.cc') diff --git a/decompress.cc b/decompress.cc index a5d4994..df8d88e 100644 --- a/decompress.cc +++ b/decompress.cc @@ -38,6 +38,7 @@ namespace { +enum { max_packet_size = 1 << 20 }; long long in_size = 0; long long out_size = 0; @@ -59,17 +60,18 @@ public: private: int receive_worker_id; // worker queue currently receiving packets int deliver_worker_id; // worker queue currently delivering packets - Slot_tally slot_tally; + Slot_tally slot_tally; // limits the number of input packets std::vector< std::queue< Packet * > > ipacket_queues; std::vector< std::queue< Packet * > > opacket_queues; - int num_working; // Number of workers still running - const int num_workers; // Number of workers - const int num_slots; // max packets in circulation + int num_working; // number of workers still running + const int num_workers; // number of workers + int num_free; // remaining free output slots pthread_mutex_t imutex; pthread_cond_t iav_or_eof; // input packet available or splitter done pthread_mutex_t omutex; pthread_cond_t oav_or_exit; // output packet available or all workers exited - bool eof; // splitter done + pthread_cond_t slot_av; // free output slot available + bool eof; // splitter done public: Packet_courier( const int workers, const int slots ) @@ -78,11 +80,19 @@ public: receive_worker_id( 0 ), deliver_worker_id( 0 ), slot_tally( slots ), ipacket_queues( workers ), opacket_queues( workers ), num_working( workers ), - num_workers( workers ), num_slots( slots ), eof( false ) - { xinit( &iav_or_eof, &imutex ); xinit( &oav_or_exit, &omutex ); } + num_workers( workers ), num_free( 8 * slots ), eof( false ) + { + xinit( &iav_or_eof, &imutex ); + xinit( &oav_or_exit, &omutex ); xinit( &slot_av, 0 ); + } ~Packet_courier() - { xdestroy( &iav_or_eof, &imutex ); xdestroy( &oav_or_exit, &omutex ); } + { + xdestroy( &iav_or_eof, &imutex ); + xdestroy( &oav_or_exit, &omutex ); xdestroy( &slot_av, 0 ); + } + + const Slot_tally & tally() const { return slot_tally; } // make a packet with data received from splitter // if data == 0, move to next queue @@ -123,10 +133,9 @@ public: { if( ipacket->data != 0 ) slot_tally.leave_slot(); } else { - // Notify muxer when last worker exits + // notify muxer when last worker exits xlock( &omutex ); - if( --num_working == 0 ) - xsignal( &oav_or_exit ); + if( --num_working == 0 ) xsignal( &oav_or_exit ); xunlock( &omutex ); } return ipacket; @@ -136,6 +145,12 @@ public: void collect_packet( Packet * const opacket, const int worker_id ) { xlock( &omutex ); + if( opacket->data != 0 ) + { + while( worker_id != deliver_worker_id && num_free <= 0 ) + xwait( &slot_av, &omutex ); + --num_free; + } opacket_queues[worker_id].push( opacket ); if( worker_id == deliver_worker_id ) xsignal( &oav_or_exit ); xunlock( &omutex ); @@ -159,12 +174,14 @@ public: if( opacket_queues[deliver_worker_id].empty() ) break; opacket = opacket_queues[deliver_worker_id].front(); opacket_queues[deliver_worker_id].pop(); - if( opacket->data != 0 ) break; - else + if( opacket->data != 0 ) { - if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0; - delete opacket; opacket = 0; + if( ++num_free == 1 ) xsignal( &slot_av ); + break; } + if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0; + xbroadcast( &slot_av ); // restart deliver_worker_id thread + delete opacket; opacket = 0; } xunlock( &omutex ); return opacket; @@ -187,8 +204,6 @@ public: if( !opacket_queues[i].empty() ) return false; return true; } - - const Slot_tally & tally() const { return slot_tally; } }; @@ -197,7 +212,6 @@ struct Splitter_arg Packet_courier * courier; const Pretty_print * pp; int infd; - int packet_size; }; @@ -211,19 +225,19 @@ extern "C" void * dsplitter( void * arg ) const int infd = tmp.infd; const int hsize = 6; // header size const int tsize = 20; // trailer size - const int buffer_size = tmp.packet_size; + const int buffer_size = max_packet_size; const int base_buffer_size = tsize + buffer_size + hsize; uint8_t * const base_buffer = new( std::nothrow ) uint8_t[base_buffer_size]; - if( base_buffer == 0 ) { pp( "not enough memory" ); fatal(); } + if( base_buffer == 0 ) { pp( "Not enough memory" ); fatal(); } uint8_t * const buffer = base_buffer + tsize; int size = readblock( infd, buffer, buffer_size + hsize ) - hsize; bool at_stream_end = ( size < buffer_size ); if( size != buffer_size && errno ) - { pp(); show_error( "read error", errno ); fatal(); } + { pp(); show_error( "Read error", errno ); fatal(); } if( size <= tsize || buffer[0] != 'L' || buffer[1] != 'Z' || buffer[2] != 'I' || buffer[3] != 'P' ) - { pp( "bad magic number (file not in lzip format)" ); fatal(); } + { pp( "Bad magic number (file not in lzip format)" ); fatal(); } long long partial_member_size = 0; while( true ) @@ -239,7 +253,7 @@ extern "C" void * dsplitter( void * arg ) if( partial_member_size + newpos - pos == member_size ) { // header found uint8_t * data = new( std::nothrow ) uint8_t[newpos - pos]; - if( data == 0 ) { pp( "not enough memory" ); fatal(); } + if( data == 0 ) { pp( "Not enough memory" ); fatal(); } std::memcpy( data, buffer + pos, newpos - pos ); courier.receive_packet( data, newpos - pos ); courier.receive_packet( 0, 0 ); // end of member token @@ -251,7 +265,7 @@ extern "C" void * dsplitter( void * arg ) if( at_stream_end ) { uint8_t * data = new( std::nothrow ) uint8_t[size + hsize - pos]; - if( data == 0 ) { pp( "not enough memory" ); fatal(); } + if( data == 0 ) { pp( "Not enough memory" ); fatal(); } std::memcpy( data, buffer + pos, size + hsize - pos ); courier.receive_packet( data, size + hsize - pos ); courier.receive_packet( 0, 0 ); // end of member token @@ -261,7 +275,7 @@ extern "C" void * dsplitter( void * arg ) { partial_member_size += buffer_size - pos; uint8_t * data = new( std::nothrow ) uint8_t[buffer_size - pos]; - if( data == 0 ) { pp( "not enough memory" ); fatal(); } + if( data == 0 ) { pp( "Not enough memory" ); fatal(); } std::memcpy( data, buffer + pos, buffer_size - pos ); courier.receive_packet( data, buffer_size - pos ); } @@ -269,7 +283,7 @@ extern "C" void * dsplitter( void * arg ) size = readblock( infd, buffer + hsize, buffer_size ); at_stream_end = ( size < buffer_size ); if( size != buffer_size && errno ) - { pp(); show_error( "read error", errno ); fatal(); } + { pp(); show_error( "Read error", errno ); fatal(); } } delete[] base_buffer; courier.finish(); // no more packets to send @@ -282,7 +296,6 @@ struct Worker_arg Packet_courier * courier; const Pretty_print * pp; int worker_id; - int packet_size; }; @@ -294,12 +307,12 @@ extern "C" void * dworker( void * arg ) Packet_courier & courier = *tmp.courier; const Pretty_print & pp = *tmp.pp; const int worker_id = tmp.worker_id; - const int new_data_size = tmp.packet_size; + const int new_data_size = max_packet_size; uint8_t * new_data = new( std::nothrow ) uint8_t[new_data_size]; LZ_Decoder * const decoder = LZ_decompress_open(); if( !new_data || !decoder || LZ_decompress_errno( decoder ) != LZ_ok ) - { pp( "not enough memory" ); fatal(); } + { pp( "Not enough memory" ); fatal(); } int new_pos = 0; while( true ) @@ -345,7 +358,7 @@ extern "C" void * dworker( void * arg ) courier.collect_packet( opacket, worker_id ); new_pos = 0; new_data = new( std::nothrow ) uint8_t[new_data_size]; - if( new_data == 0 ) { pp( "not enough memory" ); fatal(); } + if( new_data == 0 ) { pp( "Not enough memory" ); fatal(); } } if( LZ_decompress_finished( decoder ) == 1 ) { @@ -367,7 +380,7 @@ extern "C" void * dworker( void * arg ) delete[] new_data; if( LZ_decompress_total_in_size( decoder ) != 0 ) - { pp( "error, remaining data in decoder" ); fatal(); } + { pp( "Error, some data remains in decoder" ); fatal(); } LZ_decompress_close( decoder ); return 0; } @@ -388,7 +401,7 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd ) { const int wr = writeblock( outfd, opacket->data, opacket->size ); if( wr != opacket->size ) - { pp(); show_error( "write error", errno ); fatal(); } + { pp(); show_error( "Write error", errno ); fatal(); } } delete[] opacket->data; delete opacket; @@ -406,33 +419,30 @@ int decompress( const int num_workers, const int num_slots, { in_size = 0; out_size = 0; - const int packet_size = 1 << 20; Packet_courier courier( num_workers, num_slots ); Splitter_arg splitter_arg; splitter_arg.courier = &courier; splitter_arg.pp = &pp; splitter_arg.infd = infd; - splitter_arg.packet_size = packet_size; pthread_t splitter_thread; int errcode = pthread_create( &splitter_thread, 0, dsplitter, &splitter_arg ); if( errcode ) - { show_error( "can't create splitter thread", errcode ); fatal(); } + { show_error( "Can't create splitter thread", errcode ); fatal(); } 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 == 0 || worker_threads == 0 ) - { pp( "not enough memory" ); fatal(); } + { pp( "Not enough memory" ); fatal(); } for( int i = 0; i < num_workers; ++i ) { worker_args[i].courier = &courier; worker_args[i].pp = &pp; worker_args[i].worker_id = i; - worker_args[i].packet_size = packet_size; errcode = pthread_create( &worker_threads[i], 0, dworker, &worker_args[i] ); if( errcode ) - { show_error( "can't create worker threads", errcode ); fatal(); } + { show_error( "Can't create worker threads", errcode ); fatal(); } } muxer( courier, pp, outfd ); @@ -441,14 +451,14 @@ int decompress( const int num_workers, const int num_slots, { errcode = pthread_join( worker_threads[i], 0 ); if( errcode ) - { show_error( "can't join worker threads", errcode ); fatal(); } + { show_error( "Can't join worker threads", errcode ); fatal(); } } delete[] worker_threads; worker_threads = 0; delete[] worker_args; worker_args = 0; errcode = pthread_join( splitter_thread, 0 ); if( errcode ) - { show_error( "can't join splitter thread", errcode ); fatal(); } + { show_error( "Can't join splitter thread", errcode ); fatal(); } if( verbosity >= 2 ) std::fprintf( stderr, "decompressed size %9lld, size %9lld. ", -- cgit v1.2.3