lzip/plzip
1
0
Fork 0
Code Issues Activity

Merging upstream version 0.6.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-24 03:27:29 +01:00
parent 237ee44a6a
commit b724aa0729
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
11 changed files with 235 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

42
compress.cc
View file

@ -28,8 +28,8 @@
#include <queue>
#include <string>
#include <vector>
#include <inttypes.h>
#include <pthread.h>
#include <stdint.h>
#include <unistd.h>
#include <lzlib.h>
@ -101,20 +101,6 @@ void xbroadcast( pthread_cond_t * cond )
}
void xcreate( pthread_t *thread, void *(*routine)(void *), void *arg )
{
int errcode = pthread_create( thread, 0, routine, arg );
if( errcode ) { show_error( "pthread_create", errcode ); fatal(); }
}
void xjoin( pthread_t thread )
{
int errcode = pthread_join( thread, 0 );
if( errcode ) { show_error( "pthread_join", errcode ); fatal(); }
}
namespace {
long long in_size = 0;
@ -186,6 +172,7 @@ public:
{
++iwait_counter;
xwait( &iav_or_eof, &imutex );
++icheck_counter;
}
if( !packet_queue.empty() )
{
@ -226,6 +213,7 @@ public:
{
++owait_counter;
xwait( &oav_or_exit, &omutex );
++ocheck_counter;
}
Packet * opacket = circular_buffer[deliver_id%num_slots];
circular_buffer[deliver_id%num_slots] = 0;
@ -268,7 +256,7 @@ struct Splitter_arg
// split data from input file into chunks and pass them to
// courier for packaging and distribution to workers.
void * splitter( void * arg )
extern "C" void * csplitter( void * arg )
{
const Splitter_arg & tmp = *(Splitter_arg *)arg;
Packet_courier & courier = *tmp.courier;
@ -311,7 +299,7 @@ struct Worker_arg
// get packets from courier, replace their contents, and return
// them to courier.
void * worker( void * arg )
extern "C" void * cworker( void * arg )
{
const Worker_arg & tmp = *(Worker_arg *)arg;
Packet_courier & courier = *tmp.courier;
@ -426,7 +414,9 @@ int compress( const int data_size, const int dictionary_size,
splitter_arg.data_size = data_size;
pthread_t splitter_thread;
xcreate( &splitter_thread, splitter, &splitter_arg );
int errcode = pthread_create( &splitter_thread, 0, csplitter, &splitter_arg );
if( errcode )
{ show_error( "can't create splitter thread", errcode ); fatal(); }
Worker_arg worker_arg;
worker_arg.courier = &courier;
@ -438,15 +428,25 @@ int compress( const int data_size, const int dictionary_size,
if( worker_threads == 0 )
{ pp( "not enough memory" ); fatal(); }
for( int i = 0; i < num_workers; ++i )
xcreate( &worker_threads[i], worker, &worker_arg );
{
errcode = pthread_create( &worker_threads[i], 0, cworker, &worker_arg );
if( errcode )
{ show_error( "can't create worker threads", errcode ); fatal(); }
}
muxer( courier, pp, outfd );
for( int i = num_workers - 1; i >= 0; --i )
xjoin( worker_threads[i] );
{
errcode = pthread_join( worker_threads[i], 0 );
if( errcode )
{ show_error( "can't join worker threads", errcode ); fatal(); }
}
delete[] worker_threads; worker_threads = 0;
xjoin( splitter_thread );
errcode = pthread_join( splitter_thread, 0 );
if( errcode )
{ show_error( "can't join splitter thread", errcode ); fatal(); }
if( verbosity >= 1 )
{