diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2018-02-13 07:06:07 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2018-02-13 07:06:07 +0000 |
commit | ac32e8eabf1b97208c4ccdfe908aea863d09d1f3 (patch) | |
tree | 97843c59827d6028ef778206eb2d194f772126dc /compress.cc | |
parent | Adding upstream version 1.6. (diff) | |
download | plzip-ac32e8eabf1b97208c4ccdfe908aea863d09d1f3.tar.xz plzip-ac32e8eabf1b97208c4ccdfe908aea863d09d1f3.zip |
Adding upstream version 1.7.upstream/1.7
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'compress.cc')
-rw-r--r-- | compress.cc | 103 |
1 files changed, 56 insertions, 47 deletions
diff --git a/compress.cc b/compress.cc index 5bcd999..beae59e 100644 --- a/compress.cc +++ b/compress.cc @@ -1,6 +1,6 @@ /* Plzip - Parallel compressor compatible with lzip Copyright (C) 2009 Laszlo Ersek. - Copyright (C) 2009-2017 Antonio Diaz Diaz. + Copyright (C) 2009-2018 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 @@ -159,8 +159,9 @@ struct Packet // data block with a serial number uint8_t * data; int size; // number of bytes in data (if any) unsigned id; // serial number assigned as received - Packet( uint8_t * const d, const int s, const unsigned i ) - : data( d ), size( s ), id( i ) {} + Packet() : data( 0 ), size( 0 ), id( 0 ) {} + void init( uint8_t * const d, const int s, const unsigned i ) + { data = d; size = s; id = i; } }; @@ -173,10 +174,11 @@ public: unsigned owait_counter; private: unsigned receive_id; // id assigned to next packet received + unsigned distrib_id; // id of next packet to be distributed unsigned deliver_id; // id of next packet to be delivered Slot_tally slot_tally; // limits the number of input packets - std::queue< Packet * > packet_queue; - std::vector< const Packet * > circular_buffer; + std::vector< Packet > circular_ibuffer; + std::vector< const Packet * > circular_obuffer; int num_working; // number of workers still running const int num_slots; // max packets in circulation pthread_mutex_t imutex; @@ -192,8 +194,9 @@ public: Packet_courier( const int workers, const int slots ) : icheck_counter( 0 ), iwait_counter( 0 ), ocheck_counter( 0 ), owait_counter( 0 ), - receive_id( 0 ), deliver_id( 0 ), - slot_tally( slots ), circular_buffer( slots, (Packet *) 0 ), + receive_id( 0 ), distrib_id( 0 ), deliver_id( 0 ), + slot_tally( slots ), circular_ibuffer( slots ), + circular_obuffer( slots, (Packet *) 0 ), num_working( workers ), num_slots( slots ), eof( false ) { xinit_mutex( &imutex ); xinit_cond( &iav_or_eof ); @@ -206,13 +209,13 @@ public: xdestroy_cond( &iav_or_eof ); xdestroy_mutex( &imutex ); } - // make a packet with data received from splitter + // fill a packet with data received from splitter void receive_packet( uint8_t * const data, const int size ) { - Packet * const ipacket = new Packet( data, size, receive_id++ ); slot_tally.get_slot(); // wait for a free slot xlock( &imutex ); - packet_queue.push( ipacket ); + circular_ibuffer[receive_id % num_slots].init( data, size, receive_id ); + ++receive_id; xsignal( &iav_or_eof ); xunlock( &imutex ); } @@ -223,18 +226,15 @@ public: Packet * ipacket = 0; xlock( &imutex ); ++icheck_counter; - while( packet_queue.empty() && !eof ) + while( receive_id == distrib_id && !eof ) // no packets to distribute { ++iwait_counter; xwait( &iav_or_eof, &imutex ); } - if( !packet_queue.empty() ) - { - ipacket = packet_queue.front(); - packet_queue.pop(); - } + if( receive_id != distrib_id ) + { ipacket = &circular_ibuffer[distrib_id % num_slots]; ++distrib_id; } xunlock( &imutex ); - if( !ipacket ) + if( !ipacket ) // EOF { // notify muxer when last worker exits xlock( &omutex ); @@ -250,10 +250,10 @@ public: const int i = opacket->id % num_slots; xlock( &omutex ); // id collision shouldn't happen - if( circular_buffer[i] != 0 ) + if( circular_obuffer[i] != 0 ) internal_error( "id collision in collect_packet." ); // merge packet into circular buffer - circular_buffer[i] = opacket; + circular_obuffer[i] = opacket; if( opacket->id == deliver_id ) xsignal( &oav_or_exit ); xunlock( &omutex ); } @@ -264,7 +264,7 @@ public: xlock( &omutex ); ++ocheck_counter; int i = deliver_id % num_slots; - while( circular_buffer[i] == 0 && num_working > 0 ) + while( circular_obuffer[i] == 0 && num_working > 0 ) { ++owait_counter; xwait( &oav_or_exit, &omutex ); @@ -272,18 +272,19 @@ public: packet_vector.clear(); while( true ) { - const Packet * const opacket = circular_buffer[i]; + const Packet * const opacket = circular_obuffer[i]; if( !opacket ) break; packet_vector.push_back( opacket ); - circular_buffer[i] = 0; + circular_obuffer[i] = 0; ++deliver_id; i = deliver_id % num_slots; } xunlock( &omutex ); - if( packet_vector.size() ) // return slots to the tally - slot_tally.leave_slots( packet_vector.size() ); } + void return_empty_packet() // return a slot to the tally + { slot_tally.leave_slot(); } + void finish() // splitter has no more packets to send { xlock( &imutex ); @@ -294,10 +295,10 @@ public: bool finished() // all packets delivered to muxer { - if( !slot_tally.all_free() || !eof || !packet_queue.empty() || + if( !slot_tally.all_free() || !eof || receive_id != distrib_id || num_working != 0 ) return false; for( int i = 0; i < num_slots; ++i ) - if( circular_buffer[i] != 0 ) return false; + if( circular_obuffer[i] != 0 ) return false; return true; } }; @@ -369,26 +370,32 @@ extern "C" void * cworker( void * arg ) const int dictionary_size = tmp.dictionary_size; const int match_len_limit = tmp.match_len_limit; const int offset = tmp.offset; + LZ_Encoder * encoder = 0; while( true ) { Packet * const packet = courier.distribute_packet(); if( !packet ) break; // no more packets to process - const bool fast = dictionary_size == 65535 && match_len_limit == 16; - const int dict_size = fast ? dictionary_size : - std::max( std::min( dictionary_size, packet->size ), - LZ_min_dictionary_size() ); - LZ_Encoder * const encoder = - LZ_compress_open( dict_size, match_len_limit, LLONG_MAX ); - if( !encoder || LZ_compress_errno( encoder ) != LZ_ok ) + if( !encoder ) { - if( !encoder || LZ_compress_errno( encoder ) == LZ_mem_error ) - pp( mem_msg ); - else - internal_error( "invalid argument to encoder." ); - cleanup_and_fail(); + const bool fast = dictionary_size == 65535 && match_len_limit == 16; + const int dict_size = fast ? dictionary_size : + std::max( std::min( dictionary_size, packet->size ), + LZ_min_dictionary_size() ); + encoder = LZ_compress_open( dict_size, match_len_limit, LLONG_MAX ); + if( !encoder || LZ_compress_errno( encoder ) != LZ_ok ) + { + if( !encoder || LZ_compress_errno( encoder ) == LZ_mem_error ) + pp( mem_msg ); + else + internal_error( "invalid argument to encoder." ); + cleanup_and_fail(); + } } + else + if( LZ_compress_restart_member( encoder, LLONG_MAX ) < 0 ) + { pp( "LZ_compress_restart_member failed." ); cleanup_and_fail(); } int written = 0; int new_pos = 0; @@ -422,13 +429,12 @@ extern "C" void * cworker( void * arg ) if( LZ_compress_finished( encoder ) == 1 ) break; } - if( LZ_compress_close( encoder ) < 0 ) - { pp( "LZ_compress_close failed." ); cleanup_and_fail(); } - if( packet->size > 0 ) show_progress( packet->size ); packet->size = new_pos; courier.collect_packet( packet ); } + if( encoder && LZ_compress_close( encoder ) < 0 ) + { pp( "LZ_compress_close failed." ); cleanup_and_fail(); } return 0; } @@ -452,7 +458,7 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd ) if( wr != opacket->size ) { pp(); show_error( "Write error", errno ); cleanup_and_fail(); } delete[] opacket->data; - delete opacket; + courier.return_empty_packet(); } } } @@ -462,7 +468,8 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd ) // init the courier, then start the splitter and the workers and // call the muxer. -int compress( const int data_size, const int dictionary_size, +int compress( const unsigned long long cfile_size, + const int data_size, const int dictionary_size, const int match_len_limit, const int num_workers, const int infd, const int outfd, const Pretty_print & pp, const int debug_level ) @@ -486,6 +493,8 @@ int compress( const int data_size, const int dictionary_size, int errcode = pthread_create( &splitter_thread, 0, csplitter, &splitter_arg ); if( errcode ) { show_error( "Can't create splitter thread", errcode ); cleanup_and_fail(); } + if( verbosity >= 1 ) pp(); + show_progress( 0, cfile_size, &pp ); // init Worker_arg worker_arg; worker_arg.courier = &courier; @@ -522,11 +531,11 @@ int compress( const int data_size, const int dictionary_size, if( in_size == 0 || out_size == 0 ) std::fputs( " no data compressed.\n", stderr ); else - std::fprintf( stderr, "%6.3f:1, %6.3f bits/byte, " - "%5.2f%% saved, %llu in, %llu out.\n", + std::fprintf( stderr, "%6.3f:1, %5.2f%% ratio, %5.2f%% saved, " + "%llu in, %llu out.\n", (double)in_size / out_size, - ( 8.0 * out_size ) / in_size, - 100.0 * ( 1.0 - ( (double)out_size / in_size ) ), + ( 100.0 * out_size ) / in_size, + 100.0 - ( ( 100.0 * out_size ) / in_size ), in_size, out_size ); } |