diff options
Diffstat (limited to 'decompress.cc')
| -rw-r--r-- | decompress.cc | 453 |
1 files changed, 398 insertions, 55 deletions
diff --git a/decompress.cc b/decompress.cc index 0a125d0..85baaf6 100644 --- a/decompress.cc +++ b/decompress.cc @@ -38,86 +38,429 @@ namespace { -int do_decompress( LZ_Decoder * const decoder, const int infd, const int outfd, - const Pretty_print & pp, const bool testing ) +long long in_size = 0; +long long out_size = 0; + + +struct Packet // data block { - const int in_buffer_size = 65536, out_buffer_size = 8 * in_buffer_size; - uint8_t in_buffer[in_buffer_size], out_buffer[out_buffer_size]; + uint8_t * data; // data == 0 means end of member + int size; // number of bytes in data (if any) + }; - while( true ) + +class Packet_courier // moves packets around + { +public: + unsigned long icheck_counter; + unsigned long iwait_counter; + unsigned long ocheck_counter; + unsigned long 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; + 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 + 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 + +public: + Packet_courier( const int workers, const int slots ) + : icheck_counter( 0 ), iwait_counter( 0 ), + ocheck_counter( 0 ), owait_counter( 0 ), + 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 ); } + + ~Packet_courier() + { xdestroy( &iav_or_eof, &imutex ); xdestroy( &oav_or_exit, &omutex ); } + + // make a packet with data received from splitter + // if data == 0, move to next queue + void receive_packet( uint8_t * const data, const int size ) + { + Packet * ipacket = new Packet; + ipacket->data = data; + ipacket->size = size; + if( data != 0 ) + { in_size += size; slot_tally.get_slot(); } // wait for a free slot + xlock( &imutex ); + ipacket_queues[receive_worker_id].push( ipacket ); + xbroadcast( &iav_or_eof ); + xunlock( &imutex ); + if( data == 0 && ++receive_worker_id >= num_workers ) + receive_worker_id = 0; + } + + // distribute a packet to a worker + Packet * distribute_packet( const int worker_id ) { - int in_size = std::min( LZ_decompress_write_size( decoder ), in_buffer_size ); - if( in_size > 0 ) + Packet * ipacket = 0; + xlock( &imutex ); + ++icheck_counter; + while( ipacket_queues[worker_id].empty() && !eof ) { - const int max_in_size = in_size; - in_size = readblock( infd, in_buffer, max_in_size ); - if( in_size != max_in_size && errno ) - { pp(); show_error( "read error", errno ); return 1; } - if( in_size == 0 ) LZ_decompress_finish( decoder ); - else if( in_size != LZ_decompress_write( decoder, in_buffer, in_size ) ) - internal_error( "library error (LZ_decompress_write)" ); + ++iwait_counter; + xwait( &iav_or_eof, &imutex ); } - int out_size = LZ_decompress_read( decoder, out_buffer, out_buffer_size ); -// std::fprintf( stderr, "%5d in_size, %6d out_size.\n", in_size, out_size ); - if( out_size < 0 ) + if( !ipacket_queues[worker_id].empty() ) { - const LZ_Errno lz_errno = LZ_decompress_errno( decoder ); - if( lz_errno == LZ_header_error ) + ipacket = ipacket_queues[worker_id].front(); + ipacket_queues[worker_id].pop(); + } + xunlock( &imutex ); + if( ipacket != 0 ) + { if( ipacket->data != 0 ) 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 a packet from a worker + void collect_packet( Packet * const opacket, const int worker_id ) + { + xlock( &omutex ); + opacket_queues[worker_id].push( opacket ); + if( worker_id == deliver_worker_id ) xsignal( &oav_or_exit ); + xunlock( &omutex ); + } + + // deliver a packet to muxer + // if packet data == 0, move to next queue and wait again + Packet * deliver_packet() + { + Packet * opacket = 0; + xlock( &omutex ); + ++ocheck_counter; + while( true ) + { + while( opacket_queues[deliver_worker_id].empty() && num_working > 0 ) { - if( LZ_decompress_total_out_size( decoder ) > 0 ) - break; // trailing garbage - pp( "error reading member header" ); - return 1; + ++owait_counter; + xwait( &oav_or_exit, &omutex ); } - if( lz_errno == LZ_mem_error ) + 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 { - pp( "not enough memory. Find a machine with more memory" ); - return 1; + if( ++deliver_worker_id >= num_workers ) deliver_worker_id = 0; + delete opacket; opacket = 0; } - pp(); - if( lz_errno == LZ_unexpected_eof ) + } + xunlock( &omutex ); + return opacket; + } + + void finish() // splitter 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; + } + + const Slot_tally & tally() const { return slot_tally; } + }; + + +struct Splitter_arg + { + Packet_courier * courier; + const Pretty_print * pp; + int infd; + int packet_size; + }; + + + // split data from input file into chunks and pass them to + // courier for packaging and distribution to workers. +void * splitter( void * arg ) + { + const Splitter_arg & tmp = *(Splitter_arg *)arg; + Packet_courier & courier = *tmp.courier; + const Pretty_print & pp = *tmp.pp; + 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 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(); } + 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(); } + 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(); } + + long long partial_member_size = 0; + while( true ) + { + int pos = 0; + for( int newpos = 1; newpos <= size; ++newpos ) + if( buffer[newpos] == 'L' && buffer[newpos+1] == 'Z' && + buffer[newpos+2] == 'I' && buffer[newpos+3] == 'P' ) { - if( verbosity >= 0 ) - std::fprintf( stderr, "file ends unexpectedly at pos %lld\n", - LZ_decompress_total_in_size( decoder ) ); - return 2; + long long member_size = 0; + for( int i = 1; i <= 8; ++i ) + { member_size <<= 8; member_size += base_buffer[tsize+newpos-i]; } + 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(); } + std::memcpy( data, buffer + pos, newpos - pos ); + courier.receive_packet( data, newpos - pos ); + courier.receive_packet( 0, 0 ); // end of member token + partial_member_size = 0; + pos = newpos; + } } - if( verbosity >= 0 ) - std::fprintf( stderr, "LZ_decompress_read error: %s.\n", - LZ_strerror( LZ_decompress_errno( decoder ) ) ); - return 1; + + if( at_stream_end ) + { + uint8_t * data = new( std::nothrow ) uint8_t[size + hsize - pos]; + 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 + break; } - else if( out_size > 0 && outfd >= 0 ) + if( pos < buffer_size ) { - const int wr = writeblock( outfd, out_buffer, out_size ); - if( wr != out_size ) - { pp(); show_error( "write error", errno ); return 1; } + 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(); } + std::memcpy( data, buffer + pos, buffer_size - pos ); + courier.receive_packet( data, buffer_size - pos ); } - if( LZ_decompress_finished( decoder ) == 1 ) break; - if( in_size == 0 && out_size == 0 ) - internal_error( "library error (LZ_decompress_read)" ); + std::memcpy( base_buffer, base_buffer + buffer_size, tsize + hsize ); + 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(); } } - if( verbosity >= 1 ) - { if( testing ) std::fprintf( stderr, "ok\n" ); - else std::fprintf( stderr, "done\n" ); } + delete[] base_buffer; + courier.finish(); // no more packets to send return 0; } -} // end namespace + +struct Worker_arg + { + Packet_courier * courier; + const Pretty_print * pp; + int worker_id; + int packet_size; + }; -int decompress( const int infd, const int outfd, const Pretty_print & pp, - const bool testing ) + // consume packets from courier, decompress their contents, and + // give the produced packets to courier. +void * worker( void * arg ) { + const Worker_arg & tmp = *(Worker_arg *)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; + + uint8_t * new_data = new( std::nothrow ) uint8_t[new_data_size]; LZ_Decoder * const decoder = LZ_decompress_open(); - int retval; + if( !new_data || !decoder || LZ_decompress_errno( decoder ) != LZ_ok ) + { pp( "not enough memory" ); fatal(); } + int new_pos = 0; - if( !decoder || LZ_decompress_errno( decoder ) != LZ_ok ) + while( true ) { - pp( "not enough memory" ); - retval = 1; + Packet * ipacket = courier.distribute_packet( worker_id ); + if( ipacket == 0 ) break; // no more packets to process + if( ipacket->data == 0 ) LZ_decompress_finish( decoder ); + + int written = 0; + while( true ) + { + if( LZ_decompress_write_size( decoder ) > 0 && written < ipacket->size ) + { + const int wr = LZ_decompress_write( decoder, ipacket->data + written, + ipacket->size - written ); + if( wr < 0 ) internal_error( "library error (LZ_decompress_write)" ); + written += wr; + if( written > ipacket->size ) + internal_error( "ipacket size exceeded in worker" ); + } + while( true ) // read and pack decompressed data + { + const int rd = LZ_decompress_read( decoder, new_data + new_pos, + new_data_size - new_pos ); + if( rd < 0 ) + { + pp(); + if( verbosity >= 0 ) + std::fprintf( stderr, "LZ_decompress_read error in worker %d: %s.\n", + worker_id, LZ_strerror( LZ_decompress_errno( decoder ) ) ); + fatal(); + } + new_pos += rd; + if( new_pos > new_data_size ) + internal_error( "opacket size exceeded in worker" ); + if( new_pos == new_data_size || LZ_decompress_finished( decoder ) == 1 ) + { + if( new_pos > 0 ) // make data packet + { + Packet * opacket = new Packet; + opacket->data = new_data; + opacket->size = new_pos; + 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( LZ_decompress_finished( decoder ) == 1 ) + { + LZ_decompress_reset( decoder ); + Packet * opacket = new Packet; // end of member token + opacket->data = 0; + opacket->size = 0; + courier.collect_packet( opacket, worker_id ); + break; + } + } + if( rd == 0 ) break; + } + if( ipacket->data == 0 ) { delete ipacket; break; } + if( written == ipacket->size ) + { delete[] ipacket->data; delete ipacket; break; } + } } - else retval = do_decompress( decoder, infd, outfd, pp, testing ); + + delete[] new_data; + if( LZ_decompress_total_in_size( decoder ) != 0 ) + { pp( "error, remaining data in decoder" ); fatal(); } LZ_decompress_close( decoder ); - return retval; + return 0; + } + + + // get from courier the processed and sorted packets, and write + // their contents to the output file. +void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd ) + { + while( true ) + { + Packet * opacket = courier.deliver_packet(); + if( opacket == 0 ) break; // queue is empty. all workers exited + + out_size += opacket->size; + + if( outfd >= 0 ) + { + const int wr = writeblock( outfd, opacket->data, opacket->size ); + if( wr != opacket->size ) + { pp(); show_error( "write error", errno ); fatal(); } + } + delete[] opacket->data; + delete opacket; + } + } + +} // end namespace + + + // init the courier, then start the splitter and the workers and + // call the muxer. +int decompress( const int num_workers, const int num_slots, + const int infd, const int outfd, const Pretty_print & pp, + const int debug_level, const bool testing ) + { + 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; + xcreate( &splitter_thread, splitter, &splitter_arg ); + + 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(); } + 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; + xcreate( &worker_threads[i], worker, &worker_args[i] ); + } + + muxer( courier, pp, outfd ); + + for( int i = num_workers - 1; i >= 0; --i ) + xjoin( worker_threads[i] ); + delete[] worker_threads; worker_threads = 0; + delete[] worker_args; worker_args = 0; + + xjoin( splitter_thread ); + + if( verbosity >= 2 ) + std::fprintf( stderr, "decompressed size %9lld, size %9lld. ", + out_size, in_size ); + + if( verbosity >= 1 ) + { if( testing ) std::fprintf( stderr, "ok\n" ); + else std::fprintf( stderr, "done\n" ); } + + if( debug_level & 1 ) + std::fprintf( stderr, + "splitter tried to send a packet %8lu times\n" + "splitter had to wait %8lu times\n" + "any worker tried to consume from splitter %8lu times\n" + "any worker had to wait %8lu times\n" + "muxer tried to consume from workers %8lu times\n" + "muxer had to wait %8lu times\n", + courier.tally().check_counter, + courier.tally().wait_counter, + courier.icheck_counter, + courier.iwait_counter, + courier.ocheck_counter, + courier.owait_counter ); + + if( !courier.finished() ) internal_error( "courier not finished" ); + return 0; } |