summaryrefslogtreecommitdiffstats
path: root/compress.cc
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2021-01-27 15:59:02 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2021-01-27 15:59:02 +0000
commit79b2a35a210a7565b877f0c023d6c9fceab9219f (patch)
tree0cd483f5b925da79bb13653ed8d0dea4663626f2 /compress.cc
parentAdding upstream version 1.8. (diff)
downloadplzip-79b2a35a210a7565b877f0c023d6c9fceab9219f.tar.xz
plzip-79b2a35a210a7565b877f0c023d6c9fceab9219f.zip
Adding upstream version 1.9.upstream/1.9
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'compress.cc')
-rw-r--r--compress.cc237
1 files changed, 122 insertions, 115 deletions
diff --git a/compress.cc b/compress.cc
index af36f95..d8e2536 100644
--- a/compress.cc
+++ b/compress.cc
@@ -1,19 +1,19 @@
-/* Plzip - Massively parallel implementation of lzip
- Copyright (C) 2009 Laszlo Ersek.
- Copyright (C) 2009-2019 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/>.
+/* Plzip - Massively parallel implementation of lzip
+ Copyright (C) 2009 Laszlo Ersek.
+ Copyright (C) 2009-2021 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
@@ -27,7 +27,6 @@
#include <cstring>
#include <string>
#include <vector>
-#include <pthread.h>
#include <stdint.h>
#include <unistd.h>
#include <lzlib.h>
@@ -39,9 +38,9 @@
#endif
-// Returns the number of bytes really read.
-// If (returned value < size) and (errno == 0), means EOF was reached.
-//
+/* Returns the number of bytes really read.
+ If (returned value < size) and (errno == 0), means EOF was reached.
+*/
int readblock( const int fd, uint8_t * const buf, const int size )
{
int sz = 0;
@@ -58,9 +57,9 @@ int readblock( const int fd, uint8_t * const buf, const int size )
}
-// Returns the number of bytes really written.
-// If (returned value < size), it is always an error.
-//
+/* Returns the number of bytes really written.
+ If (returned value < size), it is always an error.
+*/
int writeblock( const int fd, const uint8_t * const buf, const int size )
{
int sz = 0;
@@ -150,7 +149,7 @@ namespace {
unsigned long long in_size = 0;
unsigned long long out_size = 0;
-const char * const mem_msg = "Not enough memory. Try a smaller dictionary size";
+const char * const mem_msg2 = "Not enough memory. Try a smaller dictionary size.";
struct Packet // data block with a serial number
@@ -235,8 +234,7 @@ public:
xunlock( &imutex );
if( !ipacket ) // EOF
{
- // notify muxer when last worker exits
- xlock( &omutex );
+ xlock( &omutex ); // notify muxer when last worker exits
if( --num_working == 0 ) xsignal( &oav_or_exit );
xunlock( &omutex );
}
@@ -284,12 +282,16 @@ public:
void return_empty_packet() // return a slot to the tally
{ slot_tally.leave_slot(); }
- void finish() // splitter has no more packets to send
+ void finish( const int workers_spared )
{
- xlock( &imutex );
+ 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
@@ -303,52 +305,6 @@ public:
};
-struct Splitter_arg
- {
- Packet_courier * courier;
- const Pretty_print * pp;
- int infd;
- int data_size;
- int offset;
- };
-
-
- // split data from input file into chunks and pass them to
- // courier for packaging and distribution to workers.
-extern "C" void * csplitter( void * arg )
- {
- const Splitter_arg & tmp = *(const Splitter_arg *)arg;
- Packet_courier & courier = *tmp.courier;
- const Pretty_print & pp = *tmp.pp;
- const int infd = tmp.infd;
- const int data_size = tmp.data_size;
- const int offset = tmp.offset;
-
- for( bool first_post = true; ; first_post = false )
- {
- uint8_t * const data = new( std::nothrow ) uint8_t[offset+data_size];
- if( !data ) { pp( mem_msg ); 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( size < data_size ) break; // EOF
- }
- else
- {
- delete[] data;
- break;
- }
- }
- courier.finish(); // no more packets to send
- return 0;
- }
-
-
struct Worker_arg
{
Packet_courier * courier;
@@ -358,9 +314,18 @@ struct Worker_arg
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.
+/* 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;
@@ -386,7 +351,7 @@ extern "C" void * cworker( void * arg )
if( !encoder || LZ_compress_errno( encoder ) != LZ_ok )
{
if( !encoder || LZ_compress_errno( encoder ) == LZ_mem_error )
- pp( mem_msg );
+ pp( mem_msg2 );
else
internal_error( "invalid argument to encoder." );
cleanup_and_fail();
@@ -435,8 +400,57 @@ extern "C" void * cworker( void * arg )
}
- // get from courier the processed and sorted packets, and write
- // their contents to the output file.
+/* 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;
@@ -450,8 +464,7 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd )
const Packet * const opacket = packet_vector[i];
out_size += opacket->size;
- const int wr = writeblock( outfd, opacket->data, opacket->size );
- if( wr != 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();
@@ -462,8 +475,8 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd )
} // end namespace
- // init the courier, then start the splitter and the workers and
- // call the muxer.
+/* 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,
@@ -478,50 +491,44 @@ int compress( const unsigned long long cfile_size,
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.courier = &courier;
- splitter_arg.pp = &pp;
+ 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.offset = offset;
+ 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 ); cleanup_and_fail(); }
+ { show_error( "Can't create splitter thread", errcode );
+ delete[] worker_threads; return 1; }
if( verbosity >= 1 ) pp();
show_progress( 0, cfile_size, &pp ); // init
- Worker_arg worker_arg;
- worker_arg.courier = &courier;
- worker_arg.pp = &pp;
- worker_arg.dictionary_size = dictionary_size;
- worker_arg.match_len_limit = match_len_limit;
- worker_arg.offset = offset;
-
- pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers];
- if( !worker_threads ) { pp( mem_msg ); cleanup_and_fail(); }
- for( int i = 0; i < num_workers; ++i )
- {
- errcode = pthread_create( worker_threads + i, 0, cworker, &worker_arg );
- if( errcode )
- { show_error( "Can't create worker threads", errcode ); cleanup_and_fail(); }
- }
-
muxer( courier, pp, outfd );
- for( int i = num_workers - 1; i >= 0; --i )
- {
+ 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(); }
+ if( errcode ) { show_error( "Can't join worker threads", errcode );
+ cleanup_and_fail(); }
}
delete[] worker_threads;
- errcode = pthread_join( splitter_thread, 0 );
- if( errcode )
- { show_error( "Can't join splitter thread", errcode ); cleanup_and_fail(); }
-
if( verbosity >= 1 )
{
if( in_size == 0 || out_size == 0 )
@@ -537,14 +544,14 @@ int compress( const unsigned long long cfile_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",
- courier.icheck_counter,
- courier.iwait_counter,
- courier.ocheck_counter,
- courier.owait_counter );
+ 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;