pxzcat.c

   1 /* pxzcat derived from nbdkit
   2  * Copyright (C) 2013 Red Hat Inc.
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are
   7  * met:
   8  *
   9  * * Redistributions of source code must retain the above copyright
  10  * notice, this list of conditions and the following disclaimer.
  11  *
  12  * * Redistributions in binary form must reproduce the above copyright
  13  * notice, this list of conditions and the following disclaimer in the
  14  * documentation and/or other materials provided with the distribution.
  15  *
  16  * * Neither the name of Red Hat nor the names of its contributors may be
  17  * used to endorse or promote products derived from this software without
  18  * specific prior written permission.
  19  *
  20  * THIS SOFTWARE IS PROVIDED BY RED HAT AND CONTRIBUTORS ''AS IS'' AND
  21  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  22  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  23  * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL RED HAT OR
  24  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  27  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  28  * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  29  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  30  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  31  * SUCH DAMAGE.
  32  */
  33
  34 #include <config.h>
  35
  36 #include <stdio.h>
  37 #include <stdlib.h>
  38 #include <string.h>
  39 #include <stdint.h>
  40 #include <inttypes.h>
  41 #include <unistd.h>
  42 #include <fcntl.h>
  43 #include <sys/types.h>
  44 #include <error.h>
  45 #include <errno.h>
  46 #include <getopt.h>
  47 #include <pthread.h>
  48
  49 #include <lzma.h>
  50
  51 #define DEBUG 0
  52
  53 #if DEBUG
  54 #define debug(fs,...) fprintf (stderr, "pxzcat: debug: " fs "\n", ## __VA_ARGS__)
  55 #else
  56 #define debug(fs,...) /* nothing */
  57 #endif
  58
  59 /* Size of buffers used in decompression loop. */
  60 #define BUFFER_SIZE (64*1024)
  61
  62 #define XZ_HEADER_MAGIC     "\xfd" "7zXZ\0"
  63 #define XZ_HEADER_MAGIC_LEN 6
  64 #define XZ_FOOTER_MAGIC     "YZ"
  65 #define XZ_FOOTER_MAGIC_LEN 2
  66
  67 static void usage (int exitcode);
  68 static void xzfile_uncompress (const char *filename, const char *outputfile, unsigned nr_threads);
  69 static int check_header_magic (int fd);
  70 static lzma_index *parse_indexes (const char *filename, int fd);
  71 static void iter_blocks (lzma_index *idx, unsigned nr_threads, const char *filename, int fd, const char *outputfile, int ofd);
  72
  73 static struct option long_options[] = {
  74   { "output",   required_argument,  0, 'o' },
  75   { "threads",  required_argument,  0, 'T' },
  76   { "help",     0,                  0, '?' },
  77   { NULL,       0,                  0, 0   }
  78 };
  79
  80 static const char *options = "o:T:";
  81
  82 int
  83 main (int argc, char *argv[])
  84 {
  85   int c;
  86   int longopt_index;
  87   unsigned nr_threads = 0;
  88   const char *outputfile = NULL;
  89
  90   for (;;) {
  91     c = getopt_long (argc, argv, options, long_options, &longopt_index);
  92     if (c == -1)
  93       break;
  94
  95     switch (c) {
  96       /* Long option with no short opt equivalent. */
  97     case 0:
  98       abort ();
  99
 100     case 'o':
 101       outputfile = optarg;
 102       break;
 103
 104     case 'T':
 105       if (sscanf (optarg, "%u", &nr_threads) != 1)
 106         error (EXIT_FAILURE, 0, "cannot parse -T option");
 107       break;
 108
 109     case '?':
 110       usage (EXIT_SUCCESS);
 111
 112     default:
 113       usage (EXIT_FAILURE);
 114     }
 115   }
 116
 117   if (optind != argc - 1)
 118     usage (EXIT_FAILURE);
 119
 120   if (outputfile == NULL)
 121     error (EXIT_FAILURE, 0, "you must give the -o (output file) option");
 122
 123   /* -T 0 (default) means use all cores. */
 124   if (nr_threads == 0) {
 125     long i = sysconf (_SC_NPROCESSORS_ONLN);
 126     if (i <= 0)
 127       error (EXIT_FAILURE, errno, "could not get number of cores");
 128     nr_threads = (unsigned) i;
 129   }
 130   debug ("nr_threads = %u", nr_threads);
 131
 132   xzfile_uncompress (argv[optind], outputfile, nr_threads);
 133
 134   exit (EXIT_SUCCESS);
 135 }
 136
 137 static void
 138 usage (int exitcode)
 139 {
 140   printf ("usage: pxzcat -o output [-T #threads] input.xz\n");
 141   exit (exitcode);
 142 }
 143
 144 static void
 145 xzfile_uncompress (const char *filename, const char *outputfile,
 146                    unsigned nr_threads)
 147 {
 148   int fd, ofd;
 149   uint64_t size;
 150   lzma_index *idx;
 151
 152   /* Open the file. */
 153   fd = open (filename, O_RDONLY);
 154   if (fd == -1)
 155     error (EXIT_FAILURE, errno, "open: %s", filename);
 156
 157   /* Check file magic. */
 158   if (!check_header_magic (fd))
 159     error (EXIT_FAILURE, 0, "%s: not an xz file", filename);
 160
 161   /* Read and parse the indexes. */
 162   idx = parse_indexes (filename, fd);
 163
 164   /* Get the file uncompressed size, create the output file. */
 165   size = lzma_index_uncompressed_size (idx);
 166   debug ("uncompressed size = %" PRIu64 " bytes", size);
 167
 168   /* Avoid annoying ext4 auto_da_alloc which causes a flush on close
 169    * unless we are very careful about not truncating the file when it
 170    * has zero size.  (Thanks Eric Sandeen)
 171    */
 172   unlink (outputfile);
 173
 174   ofd = open (outputfile, O_WRONLY|O_CREAT|O_TRUNC|O_NOCTTY, 0644);
 175   if (ofd == -1)
 176     error (EXIT_FAILURE, errno, "open: %s", outputfile);
 177   /* See above about auto_da_alloc. */
 178   write (ofd, " ", 1);
 179
 180   if (ftruncate (ofd, size) == -1)
 181     error (EXIT_FAILURE, errno, "ftruncate: %s", outputfile);
 182
 183   /* Tell the kernel we won't read the output file. */
 184   posix_fadvise (fd, 0, 0, POSIX_FADV_RANDOM|POSIX_FADV_DONTNEED);
 185
 186   /* Iterate over blocks. */
 187   iter_blocks (idx, nr_threads, filename, fd, outputfile, ofd);
 188
 189   close (fd);
 190 }
 191
 192 static int
 193 check_header_magic (int fd)
 194 {
 195   char buf[XZ_HEADER_MAGIC_LEN];
 196
 197   if (lseek (fd, 0, SEEK_SET) == -1)
 198     return 0;
 199   if (read (fd, buf, XZ_HEADER_MAGIC_LEN) != XZ_HEADER_MAGIC_LEN)
 200     return 0;
 201   if (memcmp (buf, XZ_HEADER_MAGIC, XZ_HEADER_MAGIC_LEN) != 0)
 202     return 0;
 203   return 1;
 204 }
 205
 206 /* For explanation of this function, see src/xz/list.c:parse_indexes
 207  * in the xz sources.
 208  */
 209 static lzma_index *
 210 parse_indexes (const char *filename, int fd)
 211 {
 212   lzma_ret r;
 213   off_t pos, index_size;
 214   uint8_t footer[LZMA_STREAM_HEADER_SIZE];
 215   uint8_t header[LZMA_STREAM_HEADER_SIZE];
 216   lzma_stream_flags footer_flags;
 217   lzma_stream_flags header_flags;
 218   lzma_stream strm = LZMA_STREAM_INIT;
 219   ssize_t n;
 220   lzma_index *combined_index = NULL;
 221   lzma_index *this_index = NULL;
 222   lzma_vli stream_padding = 0;
 223   size_t nr_streams = 0;
 224
 225   /* Check file size is a multiple of 4 bytes. */
 226   pos = lseek (fd, 0, SEEK_END);
 227   if (pos == (off_t) -1)
 228     error (EXIT_FAILURE, errno, "%s: lseek", filename);
 229
 230   if ((pos & 3) != 0)
 231     error (EXIT_FAILURE, 0,
 232            "%s: not an xz file: size is not a multiple of 4 bytes",
 233            filename);
 234
 235   /* Jump backwards through the file identifying each stream. */
 236   while (pos > 0) {
 237     debug ("looping through streams: pos = %" PRIu64, (uint64_t) pos);
 238
 239     if (pos < LZMA_STREAM_HEADER_SIZE)
 240       error (EXIT_FAILURE, 0,
 241              "%s: corrupted file at %" PRIu64, filename, (uint64_t) pos);
 242
 243     if (lseek (fd, -LZMA_STREAM_HEADER_SIZE, SEEK_CUR) == -1)
 244       error (EXIT_FAILURE, errno, "%s: lseek", filename);
 245
 246     if (read (fd, footer, LZMA_STREAM_HEADER_SIZE) != LZMA_STREAM_HEADER_SIZE)
 247       error (EXIT_FAILURE, errno, "%s: read stream footer", filename);
 248
 249     /* Skip stream padding. */
 250     if (footer[8] == 0 && footer[9] == 0 &&
 251         footer[10] == 0 && footer[11] == 0) {
 252       stream_padding += 4;
 253       pos -= 4;
 254       continue;
 255     }
 256
 257     pos -= LZMA_STREAM_HEADER_SIZE;
 258     nr_streams++;
 259
 260     debug ("decode stream footer at pos = %" PRIu64, (uint64_t) pos);
 261
 262     /* Does the stream footer look reasonable? */
 263     r = lzma_stream_footer_decode (&footer_flags, footer);
 264     if (r != LZMA_OK)
 265       error (EXIT_FAILURE, 0,
 266              "%s: invalid stream footer (error %d)", filename, r);
 267
 268     debug ("backward_size = %" PRIu64, (uint64_t) footer_flags.backward_size);
 269     index_size = footer_flags.backward_size;
 270     if (pos < index_size + LZMA_STREAM_HEADER_SIZE)
 271       error (EXIT_FAILURE, 0, "%s: invalid stream footer", filename);
 272
 273     pos -= index_size;
 274     debug ("decode index at pos = %" PRIu64, (uint64_t) pos);
 275
 276     /* Seek backwards to the index of this stream. */
 277     if (lseek (fd, pos, SEEK_SET) == -1)
 278       error (EXIT_FAILURE, errno, "%s: lseek", filename);
 279
 280     /* Decode the index. */
 281     r = lzma_index_decoder (&strm, &this_index, UINT64_MAX);
 282     if (r != LZMA_OK)
 283       error (EXIT_FAILURE, 0,
 284              "%s: invalid stream index (error %d)", filename, r);
 285
 286     do {
 287       uint8_t buf[BUFSIZ];
 288
 289       strm.avail_in = index_size;
 290       if (strm.avail_in > BUFSIZ)
 291         strm.avail_in = BUFSIZ;
 292
 293       n = read (fd, &buf, strm.avail_in);
 294       if (n == -1)
 295         error (EXIT_FAILURE, errno, "%s: read", filename);
 296
 297       index_size -= strm.avail_in;
 298
 299       strm.next_in = buf;
 300       r = lzma_code (&strm, LZMA_RUN);
 301     } while (r == LZMA_OK);
 302
 303     if (r != LZMA_STREAM_END)
 304       error (EXIT_FAILURE, 0, "%s: could not parse index (error %d)",
 305              filename, r);
 306
 307     pos -= lzma_index_total_size (this_index) + LZMA_STREAM_HEADER_SIZE;
 308
 309     debug ("decode stream header at pos = %" PRIu64, (uint64_t) pos);
 310
 311     /* Read and decode the stream header. */
 312     if (lseek (fd, pos, SEEK_SET) == -1)
 313       error (EXIT_FAILURE, errno, "%s: lseek", filename);
 314
 315     if (read (fd, header, LZMA_STREAM_HEADER_SIZE) != LZMA_STREAM_HEADER_SIZE)
 316       error (EXIT_FAILURE, errno, "%s: read stream header", filename);
 317
 318     r = lzma_stream_header_decode (&header_flags, header);
 319     if (r != LZMA_OK)
 320       error (EXIT_FAILURE, 0,
 321              "%s: invalid stream header (error %d)", filename, r);
 322
 323     /* Header and footer of the stream should be equal. */
 324     r = lzma_stream_flags_compare (&header_flags, &footer_flags);
 325     if (r != LZMA_OK)
 326       error (EXIT_FAILURE, 0,
 327              "%s: header and footer of stream are not equal (error %d)",
 328              filename, r);
 329
 330     /* Store the decoded stream flags in this_index. */
 331     r = lzma_index_stream_flags (this_index, &footer_flags);
 332     if (r != LZMA_OK)
 333       error (EXIT_FAILURE, 0,
 334              "%s: cannot read stream_flags from index (error %d)",
 335              filename, r);
 336
 337     /* Store the amount of stream padding so far.  Needed to calculate
 338      * compressed offsets correctly in multi-stream files.
 339      */
 340     r = lzma_index_stream_padding (this_index, stream_padding);
 341     if (r != LZMA_OK)
 342       error (EXIT_FAILURE, 0,
 343              "%s: cannot set stream_padding in index (error %d)",
 344              filename, r);
 345
 346     if (combined_index != NULL) {
 347       r = lzma_index_cat (this_index, combined_index, NULL);
 348       if (r != LZMA_OK)
 349         error (EXIT_FAILURE, 0, "%s: cannot combine indexes", filename);
 350     }
 351
 352     combined_index = this_index;
 353     this_index = NULL;
 354   }
 355
 356   lzma_end (&strm);
 357
 358   return combined_index;
 359 }
 360
 361 /* Return true iff the buffer is all zero bytes.
 362  *
 363  * Note that gcc is smart enough to optimize this properly:
 364  * http://stackoverflow.com/questions/1493936/faster-means-of-checking-for-an-empty-buffer-in-c/1493989#1493989
 365  */
 366 static inline int
 367 is_zero (const char *buffer, size_t size)
 368 {
 369   size_t i;
 370
 371   for (i = 0; i < size; ++i) {
 372     if (buffer[i] != 0)
 373       return 0;
 374   }
 375
 376   return 1;
 377 }
 378
 379 struct global_state {
 380   /* Current iterator.  Threads update this, but it is protected by a
 381    * mutex, and each thread takes a copy of it when working on it.
 382    */
 383   lzma_index_iter iter;
 384   lzma_bool iter_finished;
 385   pthread_mutex_t iter_mutex;
 386
 387   /* Note that all threads are accessing these fds, so you have
 388    * to use pread/pwrite instead of lseek!
 389    */
 390
 391   /* Input file. */
 392   const char *filename;
 393   int fd;
 394
 395   /* Output file. */
 396   const char *outputfile;
 397   int ofd;
 398 };
 399
 400 struct per_thread_state {
 401   unsigned thread_num;
 402   struct global_state *global;
 403   int status;
 404 };
 405
 406 /* Create threads to iterate over the blocks and uncompress. */
 407 static void *worker_thread (void *vp);
 408
 409 static void
 410 iter_blocks (lzma_index *idx, unsigned nr_threads,
 411              const char *filename, int fd, const char *outputfile, int ofd)
 412 {
 413   struct global_state global;
 414   struct per_thread_state per_thread[nr_threads];
 415   pthread_t thread[nr_threads];
 416   unsigned u, nr_errors;
 417   int err;
 418   void *status;
 419
 420   lzma_index_iter_init (&global.iter, idx);
 421   global.iter_finished = 0;
 422   err = pthread_mutex_init (&global.iter_mutex, NULL);
 423   if (err != 0)
 424     error (EXIT_FAILURE, err, "pthread_mutex_init");
 425
 426   global.filename = filename;
 427   global.fd = fd;
 428   global.outputfile = outputfile;
 429   global.ofd = ofd;
 430
 431   for (u = 0; u < nr_threads; ++u) {
 432     per_thread[u].thread_num = u;
 433     per_thread[u].global = &global;
 434   }
 435
 436   /* Start the threads. */
 437   for (u = 0; u < nr_threads; ++u) {
 438     err = pthread_create (&thread[u], NULL, worker_thread, &per_thread[u]);
 439     if (err != 0)
 440       error (EXIT_FAILURE, err, "pthread_create (%u)", u);
 441   }
 442
 443   /* Wait for the threads to exit. */
 444   nr_errors = 0;
 445   for (u = 0; u < nr_threads; ++u) {
 446     err = pthread_join (thread[u], &status);
 447     if (err != 0) {
 448       error (0, err, "pthread_join (%u)", u);
 449       nr_errors++;
 450     }
 451     if (*(int *)status == -1)
 452       nr_errors++;
 453   }
 454
 455   if (nr_errors > 0)
 456     exit (EXIT_FAILURE);
 457 }
 458
 459 /* Iterate over the blocks and uncompress. */
 460 static void *
 461 worker_thread (void *vp)
 462 {
 463   struct per_thread_state *state = vp;
 464   struct global_state *global = state->global;
 465   lzma_index_iter iter;
 466   int err;
 467   off_t position, oposition;
 468   uint8_t header[LZMA_BLOCK_HEADER_SIZE_MAX];
 469   ssize_t n;
 470   lzma_block block;
 471   lzma_filter filters[LZMA_FILTERS_MAX + 1];
 472   lzma_ret r;
 473   lzma_stream strm = LZMA_STREAM_INIT;
 474   uint8_t buf[BUFFER_SIZE];
 475   char outbuf[BUFFER_SIZE];
 476   size_t i;
 477   lzma_bool iter_finished;
 478
 479   state->status = -1;
 480
 481   for (;;) {
 482     /* Get the next block. */
 483     err = pthread_mutex_lock (&global->iter_mutex);
 484     if (err != 0) abort ();
 485     iter_finished = global->iter_finished;
 486     if (!iter_finished) {
 487       iter_finished = global->iter_finished =
 488         lzma_index_iter_next (&global->iter, LZMA_INDEX_ITER_NONEMPTY_BLOCK);
 489       if (!iter_finished)
 490         /* Take a local copy of this iterator since another thread will
 491          * update the global version.
 492          */
 493         iter = global->iter;
 494     }
 495     err = pthread_mutex_unlock (&global->iter_mutex);
 496     if (err != 0) abort ();
 497     if (iter_finished)
 498       break;
 499
 500     /* Read the block header.  Start by reading a single byte which
 501      * tell us how big the block header is.
 502      */
 503     position = iter.block.compressed_file_offset;
 504     n = pread (global->fd, header, 1, position);
 505     if (n == 0) {
 506       error (0, 0,
 507              "%s: read: unexpected end of file reading block header byte",
 508              global->filename);
 509       return &state->status;
 510     }
 511     if (n == -1) {
 512       error (0, errno, "%s: read", global->filename);
 513       return &state->status;
 514     }
 515     position++;
 516
 517     if (header[0] == '\0') {
 518       error (0, errno,
 519              "%s: read: unexpected invalid block in file, header[0] = 0",
 520              global->filename);
 521       return &state->status;
 522     }
 523
 524     block.version = 0;
 525     block.check = iter.stream.flags->check;
 526     block.filters = filters;
 527     block.header_size = lzma_block_header_size_decode (header[0]);
 528
 529     /* Now read and decode the block header. */
 530     n = pread (global->fd, &header[1], block.header_size-1, position);
 531     if (n >= 0 && n != block.header_size-1) {
 532       error (0, 0,
 533              "%s: read: unexpected end of file reading block header",
 534              global->filename);
 535       return &state->status;
 536     }
 537     if (n == -1) {
 538       error (0, errno, "%s: read", global->filename);
 539       return &state->status;
 540     }
 541     position += n;
 542
 543     r = lzma_block_header_decode (&block, NULL, header);
 544     if (r != LZMA_OK) {
 545       error (0, errno, "%s: invalid block header (error %d)",
 546              global->filename, r);
 547       return &state->status;
 548     }
 549
 550     /* What this actually does is it checks that the block header
 551      * matches the index.
 552      */
 553     r = lzma_block_compressed_size (&block, iter.block.unpadded_size);
 554     if (r != LZMA_OK) {
 555       error (0, errno,
 556              "%s: cannot calculate compressed size (error %d)",
 557              global->filename, r);
 558       return &state->status;
 559     }
 560
 561     /* Where we will start writing to. */
 562     oposition = iter.block.uncompressed_file_offset;
 563
 564     /* Read the block data and uncompress it. */
 565     r = lzma_block_decoder (&strm, &block);
 566     if (r != LZMA_OK) {
 567       error (0, 0, "%s: invalid block (error %d)", global->filename, r);
 568       return &state->status;
 569     }
 570
 571     strm.next_in = NULL;
 572     strm.avail_in = 0;
 573     strm.next_out = outbuf;
 574     strm.avail_out = sizeof outbuf;
 575
 576     for (;;) {
 577       lzma_action action = LZMA_RUN;
 578
 579       if (strm.avail_in == 0) {
 580         strm.next_in = buf;
 581         n = pread (global->fd, buf, sizeof buf, position);
 582         if (n == -1) {
 583           error (0, errno, "%s: read", global->filename);
 584           return &state->status;
 585         }
 586         position += n;
 587         strm.avail_in = n;
 588         if (n == 0)
 589           action = LZMA_FINISH;
 590       }
 591
 592       r = lzma_code (&strm, action);
 593
 594       if (strm.avail_out == 0 || r == LZMA_STREAM_END) {
 595         size_t wsz = sizeof outbuf - strm.avail_out;
 596
 597         /* Don't write if the block is all zero, to preserve output file
 598          * sparseness.  However we have to update oposition.
 599          */
 600         if (!is_zero (outbuf, wsz)) {
 601           if (pwrite (global->ofd, outbuf, wsz, oposition) != wsz) {
 602             /* XXX Handle short writes. */
 603             error (0, errno, "%s: write", global->filename);
 604             return &state->status;
 605           }
 606         }
 607         oposition += wsz;
 608
 609         strm.next_out = outbuf;
 610         strm.avail_out = sizeof outbuf;
 611       }
 612
 613       if (r == LZMA_STREAM_END)
 614         break;
 615       if (r != LZMA_OK) {
 616         error (0, 0,
 617                "%s: could not parse block data (error %d)",
 618                global->filename, r);
 619         return &state->status;
 620       }
 621     }
 622
 623     lzma_end (&strm);
 624
 625     for (i = 0; filters[i].id != LZMA_VLI_UNKNOWN; ++i)
 626       free (filters[i].options);
 627   }
 628
 629   state->status = 0;
 630   return &state->status;
 631 }