path: root/main.cc

/*  Plzip - A parallel version of the lzip data compressor
    Copyright (C) 2009 Laszlo Ersek.
    Copyright (C) 2009, 2010 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 3 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/>.
*/
/*
    Return values: 0 for a normal exit, 1 for environmental problems
    (file not found, invalid flags, I/O errors, etc), 2 to indicate a
    corrupt or invalid input file, 3 for an internal consistency error
    (eg, bug) which caused plzip to panic.
*/

#define _FILE_OFFSET_BITS 64

#include <algorithm>
#include <cassert>
#include <cerrno>
#include <climits>
#include <csignal>
#include <cstdarg>
#include <cstddef>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <string>
#include <vector>
#include <fcntl.h>
#include <stdint.h>
#include <pthread.h>
#include <unistd.h>
#include <utime.h>
#include <sys/stat.h>
#include <lzlib.h>

#if CHAR_BIT != 8
#  error "Environments where CHAR_BIT != 8 are not supported."
#endif

#include "arg_parser.h"
#include "main.h"
#include "plzip.h"

#ifndef LLONG_MAX
#define LLONG_MAX  0x7FFFFFFFFFFFFFFFLL
#endif
#ifndef LLONG_MIN
#define LLONG_MIN  (-LLONG_MAX - 1LL)
#endif
#ifndef ULLONG_MAX
#define ULLONG_MAX 0xFFFFFFFFFFFFFFFFULL
#endif

void internal_error( const char * msg );


namespace {

const char * invocation_name = 0;
const char * const Program_name    = "Plzip";
const char * const program_name    = "plzip";
const char * const program_year    = "2010";

struct { const char * from; const char * to; } const known_extensions[] = {
  { ".lz",  ""     },
  { ".tlz", ".tar" },
  { 0,      0      } };

struct lzma_options
  {
  int dictionary_size;		// 4KiB..512MiB
  int match_len_limit;		// 5..273
  };

enum Mode { m_compress = 0, m_decompress, m_test };

std::string output_filename;
int outhandle = -1;
bool delete_output_on_interrupt = false;

class Pretty_print
  {
  const char * const stdin_name;
  const unsigned int stdin_name_len;
  unsigned int longest_name;
  std::string name_;
  mutable bool first_post;

public:
  Pretty_print( const std::vector< std::string > & filenames )
    : stdin_name( "(stdin)" ), stdin_name_len( std::strlen( stdin_name ) ),
      longest_name( 0 ), first_post( false )
    {
    for( unsigned int i = 0; i < filenames.size(); ++i )
      {
      const std::string & s = filenames[i];
      const unsigned int len = ( ( s == "-" ) ? stdin_name_len : s.size() );
      if( len > longest_name ) longest_name = len;
      }
    if( longest_name == 0 ) longest_name = stdin_name_len;
    }

  void set_name( const std::string & filename )
    {
    if( filename.size() && filename != "-" ) name_ = filename;
    else name_ = stdin_name;
    first_post = true;
    }

  void reset() const throw() { if( name_.size() ) first_post = true; }
  const char * name() const throw() { return name_.c_str(); }
  void operator()( const char * const msg = 0 ) const throw();
  };


void show_help() throw()
  {
  std::printf( "%s - A parallel version of the lzip data compressor.\n", Program_name );
  std::printf( "\nUsage: %s [options] [files]\n", invocation_name );
  std::printf( "\nOptions:\n" );
  std::printf( "  -h, --help                 display this help and exit\n" );
  std::printf( "  -V, --version              output version information and exit\n" );
//  std::printf( "  -b, --member-size=<n>      set member size limit in bytes\n" );
  std::printf( "  -c, --stdout               send output to standard output\n" );
  std::printf( "  -d, --decompress           decompress\n" );
  std::printf( "  -f, --force                overwrite existing output files\n" );
  std::printf( "  -k, --keep                 keep (don't delete) input files\n" );
  std::printf( "  -m, --match-length=<n>     set match length limit in bytes [80]\n" );
  std::printf( "  -n, --threads=<n>          set the number of (de)compression threads\n" );
  std::printf( "  -o, --output=<file>        if reading stdin, place the output into <file>\n" );
  std::printf( "  -q, --quiet                suppress all messages\n" );
  std::printf( "  -s, --dictionary-size=<n>  set dictionary size limit in bytes [8MiB]\n" );
//  std::printf( "  -S, --volume-size=<n>      set volume size limit in bytes\n" );
  std::printf( "  -t, --test                 test compressed file integrity\n" );
  std::printf( "  -v, --verbose              be verbose (a 2nd -v gives more)\n" );
  std::printf( "  -1 .. -9                   set compression level [default 6]\n" );
  std::printf( "      --fast                 alias for -1\n" );
  std::printf( "      --best                 alias for -9\n" );
  if( verbosity > 0 )
    {
    std::printf( "  -D, --debug=<level>        (0-3) print debug statistics to stderr\n" );
    }
  std::printf( "If no file names are given, %s compresses or decompresses\n", program_name );
  std::printf( "from standard input to standard output.\n" );
  std::printf( "Numbers may be followed by a multiplier: k = kB = 10^3 = 1000,\n" );
  std::printf( "Ki = KiB = 2^10 = 1024, M = 10^6, Mi = 2^20, G = 10^9, Gi = 2^30, etc...\n" );
  std::printf( "\nReport bugs to lzip-bug@nongnu.org\n" );
  std::printf( "Plzip home page: http://www.nongnu.org/lzip/plzip.html\n" );
  }


void show_version() throw()
  {
  std::printf( "%s %s\n", Program_name, PROGVERSION );
  std::printf( "Copyright (C) 2009 Laszlo Ersek.\n" );
  std::printf( "Copyright (C) %s Antonio Diaz Diaz.\n", program_year );
  std::printf( "License GPLv3+: GNU GPL version 3 or later <http://gnu.org/licenses/gpl.html>\n" );
  std::printf( "This is free software: you are free to change and redistribute it.\n" );
  std::printf( "There is NO WARRANTY, to the extent permitted by law.\n" );
  }


long long getnum( const char * ptr, const int bs = 0,
                  const long long llimit = LLONG_MIN + 1,
                  const long long ulimit = LLONG_MAX ) throw()
  {
  errno = 0;
  char *tail;
  long long result = strtoll( ptr, &tail, 0 );
  if( tail == ptr )
    {
    show_error( "bad or missing numerical argument", 0, true );
    std::exit( 1 );
    }

  if( !errno && tail[0] )
    {
    int factor = ( tail[1] == 'i' ) ? 1024 : 1000;
    int exponent = 0;
    bool bad_multiplier = false;
    switch( tail[0] )
      {
      case ' ': break;
      case 'b': if( bs > 0 ) { factor = bs; exponent = 1; }
                else bad_multiplier = true;
                break;
      case 'Y': exponent = 8; break;
      case 'Z': exponent = 7; break;
      case 'E': exponent = 6; break;
      case 'P': exponent = 5; break;
      case 'T': exponent = 4; break;
      case 'G': exponent = 3; break;
      case 'M': exponent = 2; break;
      case 'K': if( factor == 1024 ) exponent = 1; else bad_multiplier = true;
                break;
      case 'k': if( factor == 1000 ) exponent = 1; else bad_multiplier = true;
                break;
      default : bad_multiplier = true;
      }
    if( bad_multiplier )
      {
      show_error( "bad multiplier in numerical argument", 0, true );
      std::exit( 1 );
      }
    for( int i = 0; i < exponent; ++i )
      {
      if( LLONG_MAX / factor >= llabs( result ) ) result *= factor;
      else { errno = ERANGE; break; }
      }
    }
  if( !errno && ( result < llimit || result > ulimit ) ) errno = ERANGE;
  if( errno )
    {
    show_error( "numerical argument out of limits" );
    std::exit( 1 );
    }
  return result;
  }


int get_dict_size( const char * arg ) throw()
  {
  char *tail;
  int bits = std::strtol( arg, &tail, 0 );
  if( bits >= LZ_min_dictionary_bits() &&
      bits <= LZ_max_dictionary_bits() && *tail == 0 )
    return ( 1 << bits );
  return getnum( arg, 0, LZ_min_dictionary_size(), LZ_max_dictionary_size() );
  }


int extension_index( const std::string & name ) throw()
  {
  for( int i = 0; known_extensions[i].from; ++i )
    {
    const std::string ext( known_extensions[i].from );
    if( name.size() > ext.size() &&
        name.compare( name.size() - ext.size(), ext.size(), ext ) == 0 )
      return i;
    }
  return -1;
  }


int open_instream( const std::string & name, struct stat * in_statsp,
                   const Mode program_mode, const int eindex,
                   const bool force, const bool to_stdout ) throw()
  {
  int inhandle = -1;
  if( program_mode == m_compress && !force && eindex >= 0 )
    {
    if( verbosity >= 0 )
      std::fprintf( stderr, "%s: input file `%s' already has `%s' suffix.\n",
                    program_name, name.c_str(),
                    known_extensions[eindex].from );
    }
  else
    {
    inhandle = open( name.c_str(), O_RDONLY );
    if( inhandle < 0 )
      {
      if( verbosity >= 0 )
        std::fprintf( stderr, "%s: Can't open input file `%s': %s.\n",
                      program_name, name.c_str(), std::strerror( errno ) );
      }
    else
      {
      const int i = fstat( inhandle, in_statsp );
      const mode_t & mode = in_statsp->st_mode;
      if( i < 0 || !( S_ISREG( mode ) || ( to_stdout &&
                      ( S_ISFIFO( mode ) || S_ISSOCK( mode ) ||
                        S_ISBLK( mode ) || S_ISCHR( mode ) ) ) ) )
        {
        if( verbosity >= 0 )
          std::fprintf( stderr, "%s: input file `%s' is not a regular file%s.\n",
                        program_name, name.c_str(),
                        to_stdout ? "" : " and `--stdout' was not specified" );
        close( inhandle );
        inhandle = -1;
        }
      }
    }
  return inhandle;
  }


void set_c_outname( const std::string & name ) throw()
  {
  output_filename = name;
  output_filename += known_extensions[0].from;
  }


void set_d_outname( const std::string & name, const int i ) throw()
  {
  if( i >= 0 )
    {
    const std::string from( known_extensions[i].from );
    if( name.size() > from.size() )
      {
      output_filename.assign( name, 0, name.size() - from.size() );
      output_filename += known_extensions[i].to;
      return;
      }
    }
  output_filename = name; output_filename += ".out";
  if( verbosity >= 0 )
    std::fprintf( stderr, "%s: can't guess original name for `%s' -- using `%s'.\n",
                  program_name, name.c_str(), output_filename.c_str() );
  }


bool open_outstream( const bool force ) throw()
  {
  if( force )
    outhandle = open( output_filename.c_str(), O_CREAT | O_TRUNC | O_WRONLY,
                      S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH );
  else outhandle = open( output_filename.c_str(), O_CREAT | O_EXCL | O_WRONLY,
                         S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH );
  if( outhandle < 0 )
    {
    if( errno == EEXIST ) outhandle = -2; else outhandle = -1;
    if( verbosity >= 0 )
      {
      if( outhandle == -2 )
        std::fprintf( stderr, "%s: Output file %s already exists, skipping.\n",
                      program_name, output_filename.c_str() );
      else
        std::fprintf( stderr, "%s: Can't create output file `%s': %s.\n",
                      program_name, output_filename.c_str(), std::strerror( errno ) );
      }
    }
  return ( outhandle >= 0 );
  }


bool check_tty( const int inhandle, const Mode program_mode ) throw()
  {
  if( program_mode == m_compress && isatty( outhandle ) )
    {
    show_error( "I won't write compressed data to a terminal.", 0, true );
    return false;
    }
  if( ( program_mode == m_decompress || program_mode == m_test ) &&
      isatty( inhandle ) )
    {
    show_error( "I won't read compressed data from a terminal.", 0, true );
    return false;
    }
  return true;
  }


void cleanup_and_fail( const int retval ) throw()
  {
  if( delete_output_on_interrupt )
    {
    if( verbosity >= 0 )
      std::fprintf( stderr, "%s: Deleting output file `%s', if it exists.\n",
               program_name, output_filename.c_str() );
    if( outhandle >= 0 ) { close( outhandle ); outhandle = -1; }
    if( std::remove( output_filename.c_str() ) != 0 )
      show_error( "WARNING: deletion of output file (apparently) failed." );
    }
  std::exit( retval );
  }


     // Set permissions, owner and times.
void close_and_set_permissions( const struct stat * const in_statsp )
  {
  bool error = false;
  if( in_statsp )
    {
    if( fchmod( outhandle, in_statsp->st_mode ) != 0 ) error = true;
    else (void)fchown( outhandle, in_statsp->st_uid, in_statsp->st_gid );
    // fchown will in many cases return with EPERM, which can be safely ignored.
    }
  if( close( outhandle ) == 0 ) outhandle = -1;
  else cleanup_and_fail( 1 );
  delete_output_on_interrupt = false;
  if( !in_statsp ) return;
  if( !error )
    {
    struct utimbuf t;
    t.actime = in_statsp->st_atime;
    t.modtime = in_statsp->st_mtime;
    if( utime( output_filename.c_str(), &t ) != 0 ) error = true;
    }
  if( error )
    {
    show_error( "I can't change output file attributes." );
    cleanup_and_fail( 1 );
    }
  }


int do_decompress( LZ_Decoder * const decoder, const int inhandle,
                   const Pretty_print & pp, const bool testing )
  {
  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];

  if( verbosity >= 1 ) pp();
  while( true )
    {
    int in_size = std::min( LZ_decompress_write_size( decoder ), in_buffer_size );
    if( in_size > 0 )
      {
      const int max_in_size = in_size;
      in_size = readblock( inhandle, (char *)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)" );
      }
    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 )
      {
      const LZ_Errno lz_errno = LZ_decompress_errno( decoder );
      if( lz_errno == LZ_header_error )
        {
        if( LZ_decompress_total_out_size( decoder ) > 0 )
          break;				// trailing garbage
        pp( "error reading member header" );
        return 1;
        }
      if( lz_errno == LZ_mem_error )
        {
        pp( "not enough memory. Find a machine with more memory" );
        return 1;
        }
      pp();
      if( lz_errno == LZ_unexpected_eof )
        {
        if( verbosity >= 0 )
          std::fprintf( stderr, "file ends unexpectedly at pos %lld\n",
                        LZ_decompress_total_in_size( decoder ) );
        return 2;
        }
      if( verbosity >= 0 )
        std::fprintf( stderr, "LZ_decompress_read error: %s.\n",
                      LZ_strerror( LZ_decompress_errno( decoder ) ) );
      return 1;
      }
    else if( out_size > 0 && outhandle >= 0 )
      {
      const int wr = writeblock( outhandle, (char *)out_buffer, out_size );
      if( wr != out_size )
        { pp(); show_error( "write error", errno ); return 1; }
      }
    if( LZ_decompress_finished( decoder ) == 1 ) break;
    if( in_size == 0 && out_size == 0 )
      internal_error( "library error (LZ_decompress_read)" );
    }
  if( verbosity >= 1 )
    { if( testing ) std::fprintf( stderr, "ok\n" );
      else std::fprintf( stderr, "done\n" ); }
  return 0;
  }


int decompress( const int inhandle, const Pretty_print & pp,
                const bool testing )
  {
  LZ_Decoder * const decoder = LZ_decompress_open();
  int retval;

  if( !decoder || LZ_decompress_errno( decoder ) != LZ_ok )
    {
    pp( "not enough memory" );
    retval = 1;
    }
  else retval = do_decompress( decoder, inhandle, pp, testing );
  LZ_decompress_close( decoder );
  return retval;
  }


extern "C" void signal_handler( int ) throw()
  {
  show_error( "Control-C or similar caught, quitting." );
  cleanup_and_fail( 1 );
  }


void set_signals() throw()
  {
  signal( SIGHUP, signal_handler );
  signal( SIGINT, signal_handler );
  signal( SIGTERM, signal_handler );
  }

} // end namespace


int verbosity = 0;


void Pretty_print::operator()( const char * const msg ) const throw()
  {
  if( verbosity >= 0 )
    {
    if( first_post )
      {
      first_post = false;
      std::fprintf( stderr, "  %s: ", name_.c_str() );
      for( unsigned int i = 0; i < longest_name - name_.size(); ++i )
        std::fprintf( stderr, " " );
      if( !msg ) std::fflush( stderr );
      }
    if( msg ) std::fprintf( stderr, "%s.\n", msg );
    }
  }


void show_error( const char * msg, const int errcode, const bool help ) throw()
  {
  if( verbosity >= 0 )
    {
    if( msg && msg[0] != 0 )
      {
      std::fprintf( stderr, "%s: %s", program_name, msg );
      if( errcode > 0 ) std::fprintf( stderr, ": %s", std::strerror( errcode ) );
      std::fprintf( stderr, "\n" );
      }
    if( help && invocation_name && invocation_name[0] != 0 )
      std::fprintf( stderr, "Try `%s --help' for more information.\n", invocation_name );
    }
  }


void internal_error( const char * msg )
  {
  std::string s( "internal error: " ); s += msg;
  show_error( s.c_str() );
  std::exit( 3 );
  }


/* Private stuff needed by fatal(). */
static pthread_t main_thread;

static pid_t pid;


/* Public utility variables and functions. */

/*
  This can be called from any thread, main thread or sub-threads alike, since
  they all call common helper functions that call fatal() in case of an error.
*/
void fatal()
{
  if( pthread_equal(pthread_self(), main_thread) )
    cleanup_and_fail( 1 );
  else
    {
    if( 0 == kill(pid, SIGUSR1) )
      pthread_exit(0);
    }
  _exit( 1 );
}


void
fail(const char *fmt, int err, ...)
{
  va_list args;

  /* Locking stderr should also protect strerror(). */
  flockfile(stderr);
  (void)fprintf(stderr, "%s: ", program_name);

  va_start(args, err);
  (void)vfprintf(stderr, fmt, args);
  va_end(args);

  (void)fprintf(stderr, ": %s\n", strerror(err));
  funlockfile(stderr);
  /* Stream stderr is never fully buffered originally. */
  fatal();
}


void
xinit(Cond *cond)
{
  pthread_mutexattr_t attr;

  int ret = pthread_mutexattr_init(&attr);
  if( ret != 0 ) {
    fail("pthread_mutexattr_init()", ret);
  }

  ret = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
  if( ret != 0 ) {
    fail("pthread_mutexattr_settype()", ret);
  }

  ret = pthread_mutex_init(&cond->lock, &attr);
  if( ret != 0 ) {
    fail("pthread_mutex_init()", ret);
  }

  ret = pthread_mutexattr_destroy(&attr);
  if( ret != 0 ) {
    fail("pthread_mutexattr_destroy()", ret);
  }

  ret = pthread_cond_init(&cond->cond, 0);
  if( ret != 0 ) {
    fail("pthread_cond_init()", ret);
  }

  cond->ccount = 0;
  cond->wcount = 0;
}


void
xdestroy(Cond *cond)
{
  int ret = pthread_cond_destroy(&cond->cond);
  if( ret != 0 ) {
    fail("pthread_cond_destroy()", ret);
  }

  ret = pthread_mutex_destroy(&cond->lock);
  if( ret != 0 ) {
    fail("pthread_mutex_destroy()", ret);
  }
}


void
xlock(Cond *cond)
{
  int ret = pthread_mutex_lock(&cond->lock);
  if( ret != 0 ) {
    fail("pthread_mutex_lock()", ret);
  }
}


void
xlock_pred(Cond *cond)
{
  xlock(cond);
  ++cond->ccount;
}


void
xunlock(Cond *cond)
{
  int ret = pthread_mutex_unlock(&cond->lock);
  if( ret != 0 ) {
    fail("pthread_mutex_unlock()", ret);
  }
}


void
xwait(Cond *cond)
{
  ++cond->wcount;
  int ret = pthread_cond_wait(&cond->cond, &cond->lock);
  if( ret != 0 ) {
    fail("pthread_cond_wait()", ret);
  }
  ++cond->ccount;
}


void
xsignal(Cond *cond)
{
  int ret = pthread_cond_signal(&cond->cond);
  if( ret != 0 ) {
    fail("pthread_cond_signal()", ret);
  }
}


void
xbroadcast(Cond *cond)
{
  int ret = pthread_cond_broadcast(&cond->cond);
  if( ret != 0 ) {
    fail("pthread_cond_broadcast()", ret);
  }
}


void
xcreate(pthread_t *thread, void *(*routine)(void *), void *arg)
{
  int ret = pthread_create(thread, 0, routine, arg);
  if( ret != 0 ) {
    fail("pthread_create()", ret);
  }
}


void
xjoin(pthread_t thread)
{
  int ret = pthread_join(thread, 0);
  if( ret != 0 ) {
    fail("pthread_join()", ret);
  }
}


void
xraise(int sig)
{
  if( -1 == kill(pid, sig) ) {
    fail("kill()", errno);
  }
}


/* Private stuff part 2. */


static void
xsigemptyset(sigset_t *set)
{
  if( -1 == sigemptyset(set) ) {
    fail("sigemptyset()", errno);
  }
}


static void
xsigaddset(sigset_t *set, int signo)
{
  if( -1 == sigaddset(set, signo) ) {
    fail("sigaddset()", errno);
  }
}


static void
xsigmask(int how, const sigset_t *set, sigset_t *oset)
{
  int ret = pthread_sigmask(how, set, oset);
  if( ret != 0 ) {
    fail("pthread_sigmask()", ret);
  }
}


static void
xsigaction(int sig, void (*handler)(int))
{
  struct sigaction act;

  act.sa_handler = handler;
  xsigemptyset(&act.sa_mask);
  act.sa_flags = 0;

  if( -1 == sigaction(sig, &act, 0) ) {
    fail("sigaction()", errno);
  }
}


enum Caught_sig { CS_INT = 1, CS_TERM, CS_USR1, CS_USR2 };

static volatile sig_atomic_t caught_sig;


extern "C" void sighandler( int sig )
  {
  /* sig_atomic_t is nowhere required to be able to hold signal values. */
  switch( sig )
    {
    case SIGINT : caught_sig = CS_INT;  break;
    case SIGTERM: caught_sig = CS_TERM; break;
    case SIGUSR1: caught_sig = CS_USR1; break;
    case SIGUSR2: caught_sig = CS_USR2; break;
    default: internal_error( "caught signal not in set" );
    }
  }


static void compress( const lzma_options & encoder_options, const int num_workers,
                      int debug_level, int num_slots, int infd, int outfd,
                      const Pretty_print & pp, const sigset_t *unblocked )
  {
  /*
    We could wait for signals with either sigwait() or sigsuspend(). SUSv2
    states about sigwait() that its effect on signal actions is unspecified.
    SUSv3 still claims the same.

    The SUSv2 description of sigsuspend() talks about both the thread and the
    whole process being suspended until a signal arrives, although thread
    suspension seems much more likely from the wording. They note that they
    filed a clarification request for this. SUSv3 cleans this up and chooses
    thread suspension which was more logical anyway.

    I favor sigsuspend() because I need to re-raise SIGTERM and SIGINT, and
    unspecified action behavior with sigwait() seems messy.

    13-OCT-2009 lacos
  */

  if( verbosity >= 1 ) pp();

  Muxer_arg muxer_arg;
  muxer_arg.dictionary_size = encoder_options.dictionary_size;
  muxer_arg.match_len_limit = encoder_options.match_len_limit;
  muxer_arg.num_workers = num_workers;
  muxer_arg.num_slots = num_slots;
  muxer_arg.debug_level = debug_level;
  muxer_arg.infd = infd;
  muxer_arg.outfd = outfd;

  pthread_t muxer_thread;
  xcreate(&muxer_thread, muxer, &muxer_arg);

  /* Unblock signals, wait for them, then block them again. */
  {
    int ret = sigsuspend(unblocked);
    assert(-1 == ret && EINTR == errno);
  }

  switch( caught_sig ) {
    case CS_INT:
    case CS_TERM:	// FIXME remove output file
      {
        int sig;
        sigset_t mask;

        sig = (CS_INT == caught_sig) ? SIGINT : SIGTERM;
        /*
          We might have inherited a SIG_IGN from the parent, but that would
          make no sense here. 24-OCT-2009 lacos
        */
        xsigaction(sig, SIG_DFL);
        xraise(sig);

        xsigemptyset(&mask);
        xsigaddset(&mask, sig);
        xsigmask(SIG_UNBLOCK, &mask, 0);
      }
      /*
        We shouldn't reach this point, but if we do for some reason, fall
        through.
      */

    case CS_USR1:
      /* Error from a non-main thread via fatal(). */
      fatal();

    case CS_USR2:
      /* Muxer thread joined other sub-threads and finished successfully. */
      break;

    default:
      assert(0);
  }

  xjoin(muxer_thread);
  }


static void
sigs_mod(int block_n_catch, sigset_t *oset)
  {
  void (*handler)(int);

  if( block_n_catch ) {
    sigset_t mask;

    xsigemptyset(&mask);
    xsigaddset(&mask, SIGINT);
    xsigaddset(&mask, SIGTERM);
    xsigaddset(&mask, SIGUSR1);
    xsigaddset(&mask, SIGUSR2);
    xsigmask(SIG_BLOCK, &mask, oset);

    handler = sighandler;
  }
  else {
    handler = SIG_DFL;
  }

  xsigaction(SIGINT,  handler);
  xsigaction(SIGTERM, handler);
  xsigaction(SIGUSR1, handler);
  xsigaction(SIGUSR2, handler);

  if( !block_n_catch ) {
    xsigmask(SIG_SETMASK, oset, 0);
  }
  }


// Returns the number of bytes really read.
// If (returned value < size) and (errno == 0), means EOF was reached.
//
int readblock( const int fd, char * buf, const int size ) throw()
  {
  int rest = size;
  errno = 0;
  while( rest > 0 )
    {
    errno = 0;
    const int n = read( fd, buf + size - rest, rest );
    if( n > 0 ) rest -= n;
    else if( n == 0 ) break;
    else if( errno != EINTR && errno != EAGAIN ) break;
    }
  return ( rest > 0 ) ? size - rest : size;
  }


// Returns the number of bytes really written.
// If (returned value < size), it is always an error.
//
int writeblock( const int fd, const char * buf, const int size ) throw()
  {
  int rest = size;
  errno = 0;
  while( rest > 0 )
    {
    errno = 0;
    const int n = write( fd, buf + size - rest, rest );
    if( n > 0 ) rest -= n;
    else if( errno && errno != EINTR && errno != EAGAIN ) break;
    }
  return ( rest > 0 ) ? size - rest : size;
  }


int main( const int argc, const char * argv[] )
  {
  // Mapping from gzip/bzip2 style 1..9 compression modes
  // to the corresponding LZMA compression modes.
  const lzma_options option_mapping[] =
    {
    { 1 << 20,  10 },		// -1
    { 1 << 20,  12 },		// -2
    { 1 << 20,  17 },		// -3
    { 1 << 21,  26 },		// -4
    { 1 << 22,  44 },		// -5
    { 1 << 23,  80 },		// -6
    { 1 << 24, 108 },		// -7
    { 1 << 24, 163 },		// -8
    { 1 << 25, 273 } };		// -9
  lzma_options encoder_options = option_mapping[5];	// default = "-6"
  int debug_level = 0;
  int inhandle = -1;
  int num_workers = 0;		// Start this many worker threads
  Mode program_mode = m_compress;
  bool force = false;
  bool keep_input_files = false;
  bool to_stdout = false;
  std::string input_filename;
  std::string default_output_filename;
  std::vector< std::string > filenames;
  invocation_name = argv[0];

  if( LZ_version()[0] != LZ_version_string[0] )
    internal_error( "bad library version" );

  main_thread = pthread_self();
  pid = getpid();

  xsigaction(SIGPIPE, SIG_IGN);
  xsigaction(SIGXFSZ, SIG_IGN);

  const int slots_per_worker = 2;
  long max_workers = sysconf( _SC_THREAD_THREADS_MAX );
  if( max_workers < 1 || max_workers > INT_MAX / slots_per_worker )
    max_workers = INT_MAX / slots_per_worker;
  if( max_workers > INT_MAX / (int)sizeof( pthread_t ) )
    max_workers = INT_MAX / sizeof( pthread_t );

  const Arg_parser::Option options[] =
    {
    { '1', "fast",            Arg_parser::no  },
    { '2',  0,                Arg_parser::no  },
    { '3',  0,                Arg_parser::no  },
    { '4',  0,                Arg_parser::no  },
    { '5',  0,                Arg_parser::no  },
    { '6',  0,                Arg_parser::no  },
    { '7',  0,                Arg_parser::no  },
    { '8',  0,                Arg_parser::no  },
    { '9', "best",            Arg_parser::no  },
    { 'b', "member-size",     Arg_parser::yes },
    { 'c', "stdout",          Arg_parser::no  },
    { 'd', "decompress",      Arg_parser::no  },
    { 'D', "debug",           Arg_parser::yes },
    { 'f', "force",           Arg_parser::no  },
    { 'h', "help",            Arg_parser::no  },
    { 'k', "keep",            Arg_parser::no  },
    { 'm', "match-length",    Arg_parser::yes },
    { 'n', "threads",         Arg_parser::yes },
    { 'o', "output",          Arg_parser::yes },
    { 'q', "quiet",           Arg_parser::no  },
    { 's', "dictionary-size", Arg_parser::yes },
    { 'S', "volume-size",     Arg_parser::yes },
    { 't', "test",            Arg_parser::no  },
    { 'v', "verbose",         Arg_parser::no  },
    { 'V', "version",         Arg_parser::no  },
    {  0 ,  0,                Arg_parser::no  } };

  Arg_parser parser( argc, argv, options );
  if( parser.error().size() )				// bad option
    { show_error( parser.error().c_str(), 0, true ); return 1; }

  int argind = 0;
  for( ; argind < parser.arguments(); ++argind )
    {
    const int code = parser.code( argind );
    if( !code ) break;					// no more options
    const char * arg = parser.argument( argind ).c_str();
    switch( code )
      {
      case '1': case '2': case '3':
      case '4': case '5': case '6':
      case '7': case '8': case '9':
                encoder_options = option_mapping[code-'1']; break;
      case 'b': break;
      case 'c': to_stdout = true; break;
      case 'd': program_mode = m_decompress; break;
      case 'D': debug_level = getnum( arg, 0, 0, 3 );
                break;
      case 'f': force = true; break;
      case 'h': show_help(); return 0;
      case 'k': keep_input_files = true; break;
      case 'm': encoder_options.match_len_limit =
                getnum( arg, 0, LZ_min_match_len_limit(),
                                LZ_max_match_len_limit() ); break;
      case 'o': default_output_filename = arg; break;
      case 'n': num_workers = getnum( arg, 0, 1, max_workers ); break;
      case 'q': verbosity = -1; break;
      case 's': encoder_options.dictionary_size = get_dict_size( arg );
                break;
      case 'S': break;
      case 't': program_mode = m_test; break;
      case 'v': if( verbosity < 4 ) ++verbosity; break;
      case 'V': show_version(); return 0;
      default : internal_error( "uncaught option" );
      }
    }

  if( num_workers <= 0 )
    {
    long num_online = sysconf( _SC_NPROCESSORS_ONLN );
    if( num_online <= 0 ) num_online = 2;
    num_workers = std::min( num_online, max_workers );
    }
  const int num_slots = num_workers * slots_per_worker;

  bool filenames_given = false;
  for( ; argind < parser.arguments(); ++argind )
    {
    if( parser.argument( argind ) != "-" ) filenames_given = true;
    filenames.push_back( parser.argument( argind ) );
    }

  if( filenames.empty() ) filenames.push_back("-");
  if( filenames_given && program_mode != m_compress ) set_signals();

  Pretty_print pp( filenames );
  if( program_mode == m_test )
    outhandle = -1;

  int retval = 0;
  for( unsigned int i = 0; i < filenames.size(); ++i )
    {
    struct stat in_stats;
    output_filename.clear();

    if( !filenames[i].size() || filenames[i] == "-" )
      {
      input_filename.clear();
      inhandle = STDIN_FILENO;
      if( program_mode != m_test )
        {
        if( to_stdout || !default_output_filename.size() )
          outhandle = STDOUT_FILENO;
        else
          {
          if( program_mode == m_compress )
            set_c_outname( default_output_filename );
          else output_filename = default_output_filename;
          if( !open_outstream( force ) )
            {
            if( outhandle == -1 && retval < 1 ) retval = 1;
            close( inhandle ); inhandle = -1;
            continue;
            }
          }
        }
      }
    else
      {
      input_filename = filenames[i];
      const int eindex = extension_index( input_filename );
      inhandle = open_instream( input_filename, &in_stats, program_mode,
                                eindex, force, to_stdout );
      if( inhandle < 0 ) { if( retval < 1 ) retval = 1; continue; }
      if( program_mode != m_test )
        {
        if( to_stdout ) outhandle = STDOUT_FILENO;
        else
          {
          if( program_mode == m_compress )
            set_c_outname( input_filename );
          else set_d_outname( input_filename, eindex );
          if( !open_outstream( force ) )
            {
            if( outhandle == -1 && retval < 1 ) retval = 1;
            close( inhandle ); inhandle = -1;
            continue;
            }
          }
        }
      }

    if( !check_tty( inhandle, program_mode ) ) return 1;

    if( output_filename.size() && !to_stdout && program_mode != m_test )
      delete_output_on_interrupt = true;
    const struct stat * const in_statsp = input_filename.size() ? &in_stats : 0;
    pp.set_name( input_filename );
    int tmp = 0;
    if( program_mode == m_compress )
      {
      sigset_t unblocked;
      sigs_mod(1, &unblocked);
      compress( encoder_options, num_workers, debug_level, num_slots, inhandle, outhandle, pp, &unblocked );
      sigs_mod(0, &unblocked);
      }
    else
      tmp = decompress( inhandle, pp, program_mode == m_test );
    if( tmp > retval ) retval = tmp;
    if( tmp && program_mode != m_test ) cleanup_and_fail( retval );

    if( delete_output_on_interrupt )
      close_and_set_permissions( in_statsp );
    if( input_filename.size() )
      {
      close( inhandle ); inhandle = -1;
      if( !keep_input_files && !to_stdout && program_mode != m_test )
        std::remove( input_filename.c_str() );
      }
    }
  if( outhandle >= 0 && close( outhandle ) != 0 )
    {
    if( verbosity >= 0 )
      std::fprintf( stderr, "%s: Can't close stdout: %s.\n",
                    program_name, std::strerror( errno ) );
    if( retval < 1 ) retval = 1;
    }
  return retval;
  }