summaryrefslogtreecommitdiffstats
path: root/compress.cc
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--compress.cc558
1 files changed, 558 insertions, 0 deletions
diff --git a/compress.cc b/compress.cc
new file mode 100644
index 0000000..39dbff1
--- /dev/null
+++ b/compress.cc
@@ -0,0 +1,558 @@
+/* Plzip - Massively parallel implementation of lzip
+ Copyright (C) 2009 Laszlo Ersek.
+ Copyright (C) 2009-2022 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
+ the Free Software Foundation, either version 2 of the License, or
+ (at your option) any later version.
+
+ This program is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#define _FILE_OFFSET_BITS 64
+
+#include <algorithm>
+#include <cerrno>
+#include <climits>
+#include <csignal>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <string>
+#include <vector>
+#include <stdint.h>
+#include <unistd.h>
+#include <lzlib.h>
+
+#include "lzip.h"
+
+#ifndef LLONG_MAX
+#define LLONG_MAX 0x7FFFFFFFFFFFFFFFLL
+#endif
+
+
+/* Return the number of bytes really read.
+ If (value returned < size) and (errno == 0), means EOF was reached.
+*/
+int readblock( const int fd, uint8_t * const buf, const int size )
+ {
+ int sz = 0;
+ errno = 0;
+ while( sz < size )
+ {
+ const int n = read( fd, buf + sz, size - sz );
+ if( n > 0 ) sz += n;
+ else if( n == 0 ) break; // EOF
+ else if( errno != EINTR ) break;
+ errno = 0;
+ }
+ return sz;
+ }
+
+
+/* Return the number of bytes really written.
+ If (value returned < size), it is always an error.
+*/
+int writeblock( const int fd, const uint8_t * const buf, const int size )
+ {
+ int sz = 0;
+ errno = 0;
+ while( sz < size )
+ {
+ const int n = write( fd, buf + sz, size - sz );
+ if( n > 0 ) sz += n;
+ else if( n < 0 && errno != EINTR ) break;
+ errno = 0;
+ }
+ return sz;
+ }
+
+
+void xinit_mutex( pthread_mutex_t * const mutex )
+ {
+ const int errcode = pthread_mutex_init( mutex, 0 );
+ if( errcode )
+ { show_error( "pthread_mutex_init", errcode ); cleanup_and_fail(); }
+ }
+
+void xinit_cond( pthread_cond_t * const cond )
+ {
+ const int errcode = pthread_cond_init( cond, 0 );
+ if( errcode )
+ { show_error( "pthread_cond_init", errcode ); cleanup_and_fail(); }
+ }
+
+
+void xdestroy_mutex( pthread_mutex_t * const mutex )
+ {
+ const int errcode = pthread_mutex_destroy( mutex );
+ if( errcode )
+ { show_error( "pthread_mutex_destroy", errcode ); cleanup_and_fail(); }
+ }
+
+void xdestroy_cond( pthread_cond_t * const cond )
+ {
+ const int errcode = pthread_cond_destroy( cond );
+ if( errcode )
+ { show_error( "pthread_cond_destroy", errcode ); cleanup_and_fail(); }
+ }
+
+
+void xlock( pthread_mutex_t * const mutex )
+ {
+ const int errcode = pthread_mutex_lock( mutex );
+ if( errcode )
+ { show_error( "pthread_mutex_lock", errcode ); cleanup_and_fail(); }
+ }
+
+
+void xunlock( pthread_mutex_t * const mutex )
+ {
+ const int errcode = pthread_mutex_unlock( mutex );
+ if( errcode )
+ { show_error( "pthread_mutex_unlock", errcode ); cleanup_and_fail(); }
+ }
+
+
+void xwait( pthread_cond_t * const cond, pthread_mutex_t * const mutex )
+ {
+ const int errcode = pthread_cond_wait( cond, mutex );
+ if( errcode )
+ { show_error( "pthread_cond_wait", errcode ); cleanup_and_fail(); }
+ }
+
+
+void xsignal( pthread_cond_t * const cond )
+ {
+ const int errcode = pthread_cond_signal( cond );
+ if( errcode )
+ { show_error( "pthread_cond_signal", errcode ); cleanup_and_fail(); }
+ }
+
+
+void xbroadcast( pthread_cond_t * const cond )
+ {
+ const int errcode = pthread_cond_broadcast( cond );
+ if( errcode )
+ { show_error( "pthread_cond_broadcast", errcode ); cleanup_and_fail(); }
+ }
+
+
+namespace {
+
+unsigned long long in_size = 0;
+unsigned long long out_size = 0;
+const char * const mem_msg2 = "Not enough memory. Try a smaller dictionary size.";
+
+
+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() : data( 0 ), size( 0 ), id( 0 ) {}
+ void init( uint8_t * const d, const int s, const unsigned i )
+ { data = d; size = s; id = i; }
+ };
+
+
+class Packet_courier // moves packets around
+ {
+public:
+ unsigned icheck_counter;
+ unsigned iwait_counter;
+ unsigned ocheck_counter;
+ 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::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;
+ 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
+
+ Packet_courier( const Packet_courier & ); // declared as private
+ void operator=( const Packet_courier & ); // declared as private
+
+public:
+ Packet_courier( const int workers, const int slots )
+ : icheck_counter( 0 ), iwait_counter( 0 ),
+ ocheck_counter( 0 ), owait_counter( 0 ),
+ receive_id( 0 ), distrib_id( 0 ), deliver_id( 0 ),
+ slot_tally( slots ), circular_ibuffer( slots ),
+ circular_obuffer( slots, (const Packet *) 0 ),
+ num_working( workers ), num_slots( slots ), eof( false )
+ {
+ xinit_mutex( &imutex ); xinit_cond( &iav_or_eof );
+ xinit_mutex( &omutex ); xinit_cond( &oav_or_exit );
+ }
+
+ ~Packet_courier()
+ {
+ xdestroy_cond( &oav_or_exit ); xdestroy_mutex( &omutex );
+ xdestroy_cond( &iav_or_eof ); xdestroy_mutex( &imutex );
+ }
+
+ // fill a packet with data received from splitter
+ void receive_packet( uint8_t * const data, const int size )
+ {
+ slot_tally.get_slot(); // wait for a free slot
+ xlock( &imutex );
+ circular_ibuffer[receive_id % num_slots].init( data, size, receive_id );
+ ++receive_id;
+ xsignal( &iav_or_eof );
+ xunlock( &imutex );
+ }
+
+ // distribute a packet to a worker
+ Packet * distribute_packet()
+ {
+ Packet * ipacket = 0;
+ xlock( &imutex );
+ ++icheck_counter;
+ while( receive_id == distrib_id && !eof ) // no packets to distribute
+ {
+ ++iwait_counter;
+ xwait( &iav_or_eof, &imutex );
+ }
+ if( receive_id != distrib_id )
+ { ipacket = &circular_ibuffer[distrib_id % num_slots]; ++distrib_id; }
+ xunlock( &imutex );
+ if( !ipacket ) // EOF
+ {
+ xlock( &omutex ); // notify muxer when last worker exits
+ if( --num_working == 0 ) xsignal( &oav_or_exit );
+ xunlock( &omutex );
+ }
+ return ipacket;
+ }
+
+ // collect a packet from a worker
+ void collect_packet( const Packet * const opacket )
+ {
+ const int i = opacket->id % num_slots;
+ xlock( &omutex );
+ // id collision shouldn't happen
+ if( circular_obuffer[i] != 0 )
+ internal_error( "id collision in collect_packet." );
+ // merge packet into circular buffer
+ circular_obuffer[i] = opacket;
+ if( opacket->id == deliver_id ) xsignal( &oav_or_exit );
+ xunlock( &omutex );
+ }
+
+ // deliver packets to muxer
+ void deliver_packets( std::vector< const Packet * > & packet_vector )
+ {
+ xlock( &omutex );
+ ++ocheck_counter;
+ int i = deliver_id % num_slots;
+ while( circular_obuffer[i] == 0 && num_working > 0 )
+ {
+ ++owait_counter;
+ xwait( &oav_or_exit, &omutex );
+ }
+ packet_vector.clear();
+ while( true )
+ {
+ const Packet * const opacket = circular_obuffer[i];
+ if( !opacket ) break;
+ packet_vector.push_back( opacket );
+ circular_obuffer[i] = 0;
+ ++deliver_id;
+ i = deliver_id % num_slots;
+ }
+ xunlock( &omutex );
+ }
+
+ void return_empty_packet() // return a slot to the tally
+ { slot_tally.leave_slot(); }
+
+ void finish( const int workers_spared )
+ {
+ xlock( &imutex ); // splitter has no more packets to send
+ eof = true;
+ xbroadcast( &iav_or_eof );
+ xunlock( &imutex );
+ xlock( &omutex ); // notify muxer if all workers have exited
+ num_working -= workers_spared;
+ if( num_working <= 0 ) xsignal( &oav_or_exit );
+ xunlock( &omutex );
+ }
+
+ bool finished() // all packets delivered to muxer
+ {
+ 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_obuffer[i] != 0 ) return false;
+ return true;
+ }
+ };
+
+
+struct Worker_arg
+ {
+ Packet_courier * courier;
+ const Pretty_print * pp;
+ int dictionary_size;
+ int match_len_limit;
+ int offset;
+ };
+
+struct Splitter_arg
+ {
+ struct Worker_arg worker_arg;
+ pthread_t * worker_threads;
+ int infd;
+ int data_size;
+ int num_workers; // returned by splitter to main thread
+ };
+
+
+/* Get packets from courier, replace their contents, and return them to
+ courier. */
+extern "C" void * cworker( void * arg )
+ {
+ const Worker_arg & tmp = *(const Worker_arg *)arg;
+ Packet_courier & courier = *tmp.courier;
+ const Pretty_print & pp = *tmp.pp;
+ 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
+
+ if( !encoder )
+ {
+ 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_msg2 );
+ 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;
+ while( true )
+ {
+ if( written < packet->size )
+ {
+ const int wr = LZ_compress_write( encoder,
+ packet->data + offset + written,
+ packet->size - written );
+ if( wr < 0 ) internal_error( "library error (LZ_compress_write)." );
+ written += wr;
+ }
+ if( written >= packet->size ) LZ_compress_finish( encoder );
+ const int rd = LZ_compress_read( encoder, packet->data + new_pos,
+ offset + written - new_pos );
+ if( rd < 0 )
+ {
+ pp();
+ if( verbosity >= 0 )
+ std::fprintf( stderr, "LZ_compress_read error: %s\n",
+ LZ_strerror( LZ_compress_errno( encoder ) ) );
+ cleanup_and_fail();
+ }
+ new_pos += rd;
+ if( new_pos >= offset + written )
+ internal_error( "packet size exceeded in worker." );
+ if( LZ_compress_finished( encoder ) == 1 ) break;
+ }
+
+ 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;
+ }
+
+
+/* Split data from input file into chunks and pass them to courier for
+ packaging and distribution to workers.
+ Start a worker per packet up to a maximum of num_workers.
+*/
+extern "C" void * csplitter( void * arg )
+ {
+ Splitter_arg & tmp = *(Splitter_arg *)arg;
+ Packet_courier & courier = *tmp.worker_arg.courier;
+ const Pretty_print & pp = *tmp.worker_arg.pp;
+ pthread_t * const worker_threads = tmp.worker_threads;
+ const int offset = tmp.worker_arg.offset;
+ const int infd = tmp.infd;
+ const int data_size = tmp.data_size;
+ int i = 0; // number of workers started
+
+ for( bool first_post = true; ; first_post = false )
+ {
+ uint8_t * const data = new( std::nothrow ) uint8_t[offset+data_size];
+ if( !data ) { pp( mem_msg2 ); cleanup_and_fail(); }
+ const int size = readblock( infd, data + offset, data_size );
+ if( size != data_size && errno )
+ { pp(); show_error( "Read error", errno ); cleanup_and_fail(); }
+
+ if( size > 0 || first_post ) // first packet may be empty
+ {
+ in_size += size;
+ courier.receive_packet( data, size );
+ if( i < tmp.num_workers ) // start a new worker
+ {
+ const int errcode =
+ pthread_create( &worker_threads[i++], 0, cworker, &tmp.worker_arg );
+ if( errcode ) { show_error( "Can't create worker threads", errcode );
+ cleanup_and_fail(); }
+ }
+ if( size < data_size ) break; // EOF
+ }
+ else
+ {
+ delete[] data;
+ break;
+ }
+ }
+ courier.finish( tmp.num_workers - i ); // no more packets to send
+ tmp.num_workers = i;
+ 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 )
+ {
+ std::vector< const Packet * > packet_vector;
+ while( true )
+ {
+ courier.deliver_packets( packet_vector );
+ if( packet_vector.empty() ) break; // all workers exited
+
+ for( unsigned i = 0; i < packet_vector.size(); ++i )
+ {
+ const Packet * const opacket = packet_vector[i];
+ out_size += opacket->size;
+
+ if( writeblock( outfd, opacket->data, opacket->size ) != opacket->size )
+ { pp(); show_error( "Write error", errno ); cleanup_and_fail(); }
+ delete[] opacket->data;
+ courier.return_empty_packet();
+ }
+ }
+ }
+
+} // end namespace
+
+
+/* Init the courier, then start the splitter and the workers and call the
+ muxer. */
+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 )
+ {
+ const int offset = data_size / 8; // offset for compression in-place
+ const int slots_per_worker = 2;
+ const int num_slots =
+ ( ( num_workers > 1 ) ? num_workers * slots_per_worker : 1 );
+ in_size = 0;
+ out_size = 0;
+ Packet_courier courier( num_workers, num_slots );
+
+ if( debug_level & 2 ) std::fputs( "compress.\n", stderr );
+
+ pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers];
+ if( !worker_threads ) { pp( mem_msg ); return 1; }
+
+ Splitter_arg splitter_arg;
+ splitter_arg.worker_arg.courier = &courier;
+ splitter_arg.worker_arg.pp = &pp;
+ splitter_arg.worker_arg.dictionary_size = dictionary_size;
+ splitter_arg.worker_arg.match_len_limit = match_len_limit;
+ splitter_arg.worker_arg.offset = offset;
+ splitter_arg.worker_threads = worker_threads;
+ splitter_arg.infd = infd;
+ splitter_arg.data_size = data_size;
+ splitter_arg.num_workers = num_workers;
+
+ pthread_t splitter_thread;
+ int errcode = pthread_create( &splitter_thread, 0, csplitter, &splitter_arg );
+ if( errcode )
+ { show_error( "Can't create splitter thread", errcode );
+ delete[] worker_threads; return 1; }
+ if( verbosity >= 1 ) pp();
+ show_progress( 0, cfile_size, &pp ); // init
+
+ muxer( courier, pp, outfd );
+
+ errcode = pthread_join( splitter_thread, 0 );
+ if( errcode ) { show_error( "Can't join splitter thread", errcode );
+ cleanup_and_fail(); }
+
+ for( int i = splitter_arg.num_workers; --i >= 0; )
+ { // join only the workers started
+ errcode = pthread_join( worker_threads[i], 0 );
+ if( errcode ) { show_error( "Can't join worker threads", errcode );
+ cleanup_and_fail(); }
+ }
+ delete[] worker_threads;
+
+ if( verbosity >= 1 )
+ {
+ if( in_size == 0 || out_size == 0 )
+ std::fputs( " no data compressed.\n", stderr );
+ else
+ std::fprintf( stderr, "%6.3f:1, %5.2f%% ratio, %5.2f%% saved, "
+ "%llu in, %llu out.\n",
+ (double)in_size / out_size,
+ ( 100.0 * out_size ) / in_size,
+ 100.0 - ( ( 100.0 * out_size ) / in_size ),
+ in_size, out_size );
+ }
+
+ if( debug_level & 1 )
+ std::fprintf( stderr,
+ "workers started %8u\n"
+ "any worker tried to consume from splitter %8u times\n"
+ "any worker had to wait %8u times\n"
+ "muxer tried to consume from workers %8u times\n"
+ "muxer had to wait %8u times\n",
+ splitter_arg.num_workers,
+ courier.icheck_counter, courier.iwait_counter,
+ courier.ocheck_counter, courier.owait_counter );
+
+ if( !courier.finished() ) internal_error( "courier not finished." );
+ return 0;
+ }