Hello community, here is the log from the commit of package pbzip2 for openSUSE:Factory checked in at Mon Feb 21 00:54:05 CET 2011.
-------- --- pbzip2/pbzip2.changes 2010-12-20 19:51:46.000000000 +0100 +++ /mounts/work_src_done/STABLE/pbzip2/pbzip2.changes 2011-02-20 00:14:22.000000000 +0100 @@ -1,0 +2,15 @@ +Sat Feb 19 23:02:36 UTC 2011 - [email protected] + +- update to 1.1.2: + * modifies Consumer_decompress throttling to prevent potential + deadlock/infinite loops in certain situations + * fixes a deadlock bug and performance issue when consumer is working with + long bzip2 sequences + * fixes a bug which caused a hang while decompressing a prematurely truncated + bzip2 stream + * fixes a hang on decompression of some truncated archives + * adds --ignore-trailing-garbage for non-standard archives such as gentoo + packages +- reformatted changes file + +------------------------------------------------------------------- @@ -4 +19,2 @@ -- reverting to upstream release, changed version to 1.1.0+0 to be newer than 1.1.0_git... +- reverting to upstream release, changed version to 1.1.0+0 to be newer than + 1.1.0_git... @@ -6,2 +22,4 @@ - * pbzip2-makefile.patch: adds support for BINDIR, MANDIR, DESTDIR and OPTFLAGS - * pbzip2-fix_printf_format.patch: fix the many printf format errors in debug and error reporting statements + * pbzip2-makefile.patch: adds support for BINDIR, MANDIR, DESTDIR and + OPTFLAGS + * pbzip2-fix_printf_format.patch: fix the many printf format errors in debug + and error reporting statements @@ -9,3 +27,6 @@ -- replace macros.pbzip2 with macros.pbzip2.in and a placeholder to inject the build time value of %{_bindir} -- add symbolic links to the manpage for pbunzip2 and pbzcat too, thanks to darix for pointing this out -- changed compression of upstream tarball from xz to bzip2 in order to build on older openSUSE and SLE releases +- replace macros.pbzip2 with macros.pbzip2.in and a placeholder to inject the + build time value of %{_bindir} +- add symbolic links to the manpage for pbunzip2 and pbzcat too, thanks to + darix for pointing this out +- changed compression of upstream tarball from xz to bzip2 in order to build on + older openSUSE and SLE releases @@ -28,3 +49,6 @@ - * improves the decompression performance of long bzip2 streams of large single-stream bzip2 blocks - * pbzip2 should now decompress files created with bzip2 at least as quickly as bzip2, or slightly faster - * handles decompression of long bzip2 streams incrementally instead of loading whole streams in memory at once + * improves the decompression performance of long bzip2 streams of large + single-stream bzip2 blocks + * pbzip2 should now decompress files created with bzip2 at least as quickly + as bzip2, or slightly faster + * handles decompression of long bzip2 streams incrementally instead of + loading whole streams in memory at once @@ -36,3 +60,6 @@ - * adds support for multi-threaded decompression using STDIN/pipes and throttling compression to prevent memory exhaustion with slow output pipe - * fixes a bug that did not allow command line parameters to be used when compressing data from STDIN - * contains major improvements to protection of shared variables, error and signal handling, and program termination + * adds support for multi-threaded decompression using STDIN/pipes and + throttling compression to prevent memory exhaustion with slow output pipe + * fixes a bug that did not allow command line parameters to be used when + compressing data from STDIN + * contains major improvements to protection of shared variables, error and + signal handling, and program termination calling whatdependson for head-i586 Old: ---- pbzip2-1.1.1.tar.bz2 New: ---- pbzip2-1.1.2.tar.bz2 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Other differences: ------------------ ++++++ pbzip2.spec ++++++ --- /var/tmp/diff_new_pack.QZ5hVR/_old 2011-02-21 00:53:50.000000000 +0100 +++ /var/tmp/diff_new_pack.QZ5hVR/_new 2011-02-21 00:53:50.000000000 +0100 @@ -1,7 +1,7 @@ # -# spec file for package pbzip2 (Version 1.1.1+0) +# spec file for package pbzip2 # -# Copyright (c) 2010 SUSE LINUX Products GmbH, Nuernberg, Germany. +# Copyright (c) 2011 SUSE LINUX Products GmbH, Nuernberg, Germany. # # All modifications and additions to the file contributed by third parties # remain the property of their copyright owners, unless otherwise agreed @@ -18,9 +18,9 @@ Name: pbzip2 -Version: 1.1.1+0 +Version: 1.1.2 Release: 1 -%define pkg_version 1.1.1 +%define pkg_version %{version} License: BSD4c Summary: Parallelized Implementation of bzip2 Url: http://compression.ca/pbzip2/ @@ -87,4 +87,5 @@ %doc %{_mandir}/man1/pbunzip2.1%{ext_man} %doc %{_mandir}/man1/pbzcat.1%{ext_man} %config %{_sysconfdir}/rpm/macros.%{name} + %changelog ++++++ pbzip2-1.1.1.tar.bz2 -> pbzip2-1.1.2.tar.bz2 ++++++ diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/AUTHORS new/pbzip2-1.1.2/AUTHORS --- old/pbzip2-1.1.1/AUTHORS 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/AUTHORS 2011-02-19 11:10:22.000000000 +0100 @@ -181,8 +181,22 @@ disabled and enabled monitoring of segmented long bzip2 streams - Fixed issue with Sun Studio compiler - required explicit declaration of static const members in .cpp. + - consumer_decompress throttling loosed a bit to prevent potential + deadlock/infinite loop in certain situations. (Addition to all-empty-block + tails in OutputBuffer is non-blocking now). + - fixed error message for block size range (max size was wrong) + - consumer_decompress: fixed bug which caused hang while decompressing + prematurely truncated bzip2 stream. + - modified fileWriter to prevent from throttling when output buffers are full + (condition signalling added when block is ready to wake up sleeping writer early) + - Debug print bug fixed in queue::remove. + - Debuging and error handling improvements and refactoring. + - Fixed hang on decompress of some truncated archives (bug #590225). + - Implemented --ignore-trailing-garbage feature (bug #594868) + - Fixed hang on decompress of some truncated archives (bug #590225) -Specials thanks for suggestions and testing + +Special thanks for suggestions and testing ------------------------------------------- -Phillippe Welsh, James Terhune, Dru Lemley, Bryan Stillwell, George Chalissery, Kir Kolyshkin, Madhu Kangara, Mike Furr, Joergen Ramskov, Kurt Fitzner, Peter Cordes, Oliver Falk, Jindrich Novy, Benjamin Reed, Chris Dearman, Richard Russon, An�bal Monsalve Salazar, Jim Leonard, Paul Pluzhniko, Robert Archard, Coran Fisher, Ken Takusagawa, David Pyke, Matt Turner, Damien Ancelin, �lvaro Reguly, Ivan Voras, John Dalton, Sami Liedes, Rene Georgi, Ren� Rh�aume, Jeroen Roovers, Reinhard Schiedermeier, Kari Pahula, Elbert Pol, Nico Vrouwe, Eduardo Terol, Samuel Thibault, Michael Fuereder, Jari Aalto, Scott Emery, Steven Chamberlain, Yavor Nikolov, Nikita Zhuk, Joao Seabra, Conn Clark, Mark A. Haun, Tim Bielawa, Michal Gorny, Mikolaj Habdank, Christian Kujau, Marc-Christian Petersen, Piero Ottuzzi, Ephraim Ofir. +Phillippe Welsh, James Terhune, Dru Lemley, Bryan Stillwell, George Chalissery, Kir Kolyshkin, Madhu Kangara, Mike Furr, Joergen Ramskov, Kurt Fitzner, Peter Cordes, Oliver Falk, Jindrich Novy, Benjamin Reed, Chris Dearman, Richard Russon, An�bal Monsalve Salazar, Jim Leonard, Paul Pluzhniko, Robert Archard, Coran Fisher, Ken Takusagawa, David Pyke, Matt Turner, Damien Ancelin, �lvaro Reguly, Ivan Voras, John Dalton, Sami Liedes, Rene Georgi, Ren� Rh�aume, Jeroen Roovers, Reinhard Schiedermeier, Kari Pahula, Elbert Pol, Nico Vrouwe, Eduardo Terol, Samuel Thibault, Michael Fuereder, Jari Aalto, Scott Emery, Steven Chamberlain, Yavor Nikolov, Nikita Zhuk, Joao Seabra, Conn Clark, Mark A. Haun, Tim Bielawa, Michal Gorny, Mikolaj Habdank, Christian Kujau, Marc-Christian Petersen, Piero Ottuzzi, Ephraim Ofir, Laszlo Ersek, Dima Tisnek. diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/COPYING new/pbzip2-1.1.2/COPYING --- old/pbzip2-1.1.1/COPYING 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/COPYING 2011-02-19 11:10:22.000000000 +0100 @@ -1,4 +1,4 @@ -This program, "pbzip2" is copyright (C) 2003-2010 Jeff Gilchrist. +This program, "pbzip2" is copyright (C) 2003-2011 Jeff Gilchrist. All rights reserved. The library "libbzip2" which pbzip2 uses, is copyright @@ -37,4 +37,4 @@ Jeff Gilchrist, Ottawa, Canada. [email protected] -pbzip2 version 1.1.1 of April 17, 2010 +pbzip2 version 1.1.2 of Feb 19, 2011 diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/ChangeLog new/pbzip2-1.1.2/ChangeLog --- old/pbzip2-1.1.1/ChangeLog 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/ChangeLog 2011-02-19 11:10:22.000000000 +0100 @@ -1,3 +1,27 @@ +Changes in 1.1.2 (Feb 19, 2011) +- Fix directdecompress segfault when destination file can't be + opened (e.g. read-only) (bug #717852) +- Implemented --ignore-trailing-garbage feature (bug #594868) +- Fixed hang on decompress of some truncated archives (bug #590225) +- Pulled an error check out of normal logic block for clarity +- Debug print added after BZ2_bzDecompress to track it's return code. +- A debug print fixed in queue::remove +- Increased max memory usage limit from 1GB to 2GB +- If no -m switch given on command line, default max memory limit + will now automatically increase from 100 MB to minimum amount + of memory required to support the number of CPUs requested +- Improved performance when output buffer is full +- Fixed bug which caused hang while decompressing prematurely + truncated bzip2 stream +- Consumer_decompress throttling modified to prevent potential + deadlock/infinite loop in certain situations (Thanks to Laszlo + Ersek for finding and helping track down the cause of this bug) +- Fixed deadlock bug and performance issue when consumer working + with long bzip2 sequences (Thanks to Tanguy Fautre for finding) +- Fixed error message for block size range (max size was wrong) +- Moved #include <pthread.h> from pbzip2.cpp to pbzip2.h to fix + OS/2 compiler issue + Changes in 1.1.1 (Apr 17, 2010) - Modified decompression to use low-level libbz2 API to improve performance of long bzip2 streams of large single-stream bzip2 diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/Makefile new/pbzip2-1.1.2/Makefile --- old/pbzip2-1.1.1/Makefile 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/Makefile 2011-02-19 11:10:22.000000000 +0100 @@ -27,6 +27,12 @@ # Comment out CFLAGS line below to disable Thread stack size customization CFLAGS += -DUSE_STACKSIZE_CUSTOMIZATION +# Comment out CFLAGS line below to explicity set ignore trailing garbage +# default behavior: 0 - disabled; 1 - enabled (ignore garbage by default) +# If IGNORE_TRAILING_GARBAGE is not defined: behavior is automatically determined +# by program name: bzip2, bunzip2, bzcat - ignore garbage; otherwise - not. +#CFLAGS += -DIGNORE_TRAILING_GARBAGE=1 + # On some compilers -pthreads CFLAGS += -pthread diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/README new/pbzip2-1.1.2/README --- old/pbzip2-1.1.1/README 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/README 2011-02-19 11:10:22.000000000 +0100 @@ -1,6 +1,6 @@ -April 17, 2010 +Feb 19, 2011 -Parallel BZIP2 v1.1.1 - by: Jeff Gilchrist <[email protected]> +Parallel BZIP2 v1.1.2 - by: Jeff Gilchrist <[email protected]> Available at: http://compression.ca/ This is the README for pbzip2, a parallel implementation of the @@ -135,6 +135,7 @@ -V Display version info for pbzip2 then exit -z,--compress Compress file (default) -1,--fast ... -9,--best Set BWT block size to 100k .. 900k (default 900k). + --ignore-trailing-garbage=# Ignore trailing garbage flag (1 - ignored; 0 - forbidden) Example: pbzip2 myfile.tar diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/pbzip2.1 new/pbzip2-1.1.2/pbzip2.1 --- old/pbzip2-1.1.1/pbzip2.1 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/pbzip2.1 2011-02-19 11:10:22.000000000 +0100 @@ -1,6 +1,6 @@ .TH pbzip2 1 .SH NAME -pbzip2 \- parallel bzip2 file compressor, v1.1.1 +pbzip2 \- parallel bzip2 file compressor, v1.1.2 .SH SYNOPSIS .B pbzip2 .RB [ " \-123456789 " ] @@ -83,6 +83,9 @@ .TP .B \-1,\-\-fast ... \-9,\-\-best Set BWT block size to 100k .. 900k (default 900k). +.TP +.B \-\-ignore-trailing-garbage=# +Ignore trailing garbage flag (1 - ignored; 0 - forbidden) .SH FILE SIZES You should be able to compress files larger than 4GB with .I pbzip2. diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/pbzip2.cpp new/pbzip2-1.1.2/pbzip2.cpp --- old/pbzip2-1.1.1/pbzip2.cpp 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/pbzip2.cpp 2011-02-19 11:10:22.000000000 +0100 @@ -9,7 +9,7 @@ * - uses libbzip2 by Julian Seward (http://sources.redhat.com/bzip2/) * - Major contributions by Yavor Nikolov <[email protected]> * - * Date : April 17, 2010 + * Date : Feb 19, 2011 * * TODO * Known Issues @@ -169,6 +169,25 @@ * disabled and enabled monitoring of segmented long bzip2 streams * - Fixed issue with Sun Studio compiler - required explicit declaration * of static const members in .cpp. + * - consumer_decompress throttling loosed a bit to prevent potential + * deadlock/infinite loop in certain situations. (Addition to all-empty-block + * tails in OutputBuffer is non-blocking now). + * - fixed error message for block size range (max size was wrong) + * - consumer_decompress: fixed bug which caused hang while decompressing + * prematurely truncated bzip2 stream. + * - modified fileWriter to prevent from throttling when output buffers are full + * (condition signalling added when block is ready to wake up sleeping writer early) + * - Fixed deadlock bug possible with stuck consumers waiting for other one + * on long multi-segment sequence. + * - Resolved performance issue: all have been waiting for any consumer + * working on long-sequence until it's finished even when there were enough + * free slots in the input queue. + * - Debug print bug fixed in queue::remove. + * - Debuging and error handling improvements and refactoring. + * - Fixed hang on decompress of some truncated archives (bug #590225). + * - Implemented --ignore-trailing-garbage feature (bug #594868) + * - Fixed hang on decompress of some truncated archives (bug #590225) + * * * Specials thanks for suggestions and testing: Phillippe Welsh, * James Terhune, Dru Lemley, Bryan Stillwell, George Chalissery, @@ -182,10 +201,10 @@ * Jari Aalto, Scott Emery, Steven Chamberlain, Yavor Nikolov, Nikita Zhuk, * Joao Seabra, Conn Clark, Mark A. Haun, Tim Bielawa, Michal Gorny, * Mikolaj Habdank, Christian Kujau, Marc-Christian Petersen, Piero Ottuzzi, - * Ephraim Ofir. + * Ephraim Ofir, Laszlo Ersek, Dima Tisnek, Tanguy Fautre. * * - * This program, "pbzip2" is copyright (C) 2003-2010 Jeff Gilchrist. + * This program, "pbzip2" is copyright (C) 2003-2011 Jeff Gilchrist. * All rights reserved. * * The library "libbzip2" which pbzip2 uses, is copyright @@ -224,7 +243,7 @@ * * Jeff Gilchrist, Ottawa, Canada. * [email protected] - * pbzip2 version 1.1.1 of April 17, 2010 + * pbzip2 version 1.1.2 of Feb 19, 2011 * */ #include "pbzip2.h" @@ -240,7 +259,6 @@ #include <sys/stat.h> #include <errno.h> #include <fcntl.h> -#include <pthread.h> #include <signal.h> #include <stdio.h> #include <stdarg.h> @@ -260,6 +278,7 @@ static int finishedFlag = 0; // Main thread work finished (about to exit) static int unfinishedWorkCleaned = 0; static int numCPU = 2; +static int IgnoreTrailingGarbageFlag = 0; // ingnore trailing garbage on decompress flag static int QUEUESIZE = 2; static int SIG_HANDLER_QUIT_SIGNAL = SIGUSR1; // signal used to stop SignalHandlerThread #ifdef USE_STACKSIZE_CUSTOMIZATION @@ -278,6 +297,7 @@ static size_t NumBufferedTailBlocks = 0; static size_t NumBufferedBlocksMax = 0; static int NextBlockToWrite; +static int LastGoodBlock; // set only to terminate write prematurely (ignoring garbage) static size_t OutBufferPosToWrite; // = 0; // position in output buffer static int Verbosity = 0; static int QuietMode = 1; @@ -294,6 +314,8 @@ static pthread_mutex_t ProgressIndicatorsMutex = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t *notTooMuchNumBuffered; static pthread_cond_t TerminateCond = PTHREAD_COND_INITIALIZER; +static pthread_cond_t OutBufferHeadNotEmpty = PTHREAD_COND_INITIALIZER; +static pthread_cond_t ErrStateChangeCond = PTHREAD_COND_INITIALIZER; static pthread_attr_t ChildThreadAttributes; static struct stat fileMetaData; static const char *sigInFilename = NULL; @@ -308,6 +330,9 @@ inline int syncGetTerminateFlag(); inline void syncSetTerminateFlag(int newValue); inline void syncSetFinishedFlag(int newValue); +inline void setLastGoodBlock(int newValue); +inline void syncSetLastGoodBlock(int newValue); +inline int syncGetLastGoodBlock(); void cleanupUnfinishedWork(); void cleanupAndQuit(int exitCode); int initSignalMask(); @@ -333,6 +358,7 @@ void outputBufferInit(size_t size); outBuff * outputBufferAdd(const outBuff & element, const char *caller); outBuff * outputBufferSeqAddNext(outBuff * preveElement, outBuff * newElement); +inline size_t getOutputBufferPos(int blockNum); int getFileMetaData(const char *); int writeFileMetaData(const char *); int testBZ2ErrorHandling(int, BZFILE *, int); @@ -340,6 +366,21 @@ ssize_t bufread(int hf, char *buf, size_t bsize); int detectCPUs(void); +inline bool isIgnoredTrailingGarbage(); +int waitForPreviousBlock(int blockNum); +inline int getLastGoodBlockBeforeErr(int errBlockNumber, int outSequenceNumber); +inline int issueDecompressError(int bzret, const outBuff * fileData, + int outSequenceNumber, const bz_stream & strm, const char * errmsg, + int exitCode); +int decompressErrCheckSingle(int bzret, const outBuff * fileData, + int outSequenceNumber, const bz_stream & strm, const char * errmsg, + bool isTrailingGarbageErr); +int decompressErrCheck(int bzret, const outBuff * fileData, + int outSequenceNumber, const bz_stream & strm); +int producerDecompressCheckInterrupt(int hInfile, outBuff *& fileData, int lastBlock); + + + /* * Pointers to functions used by plain C pthreads API require C calling * conventions. @@ -599,6 +640,52 @@ safe_mutex_unlock(&TerminateFlagMutex); } +inline void setLastGoodBlock(int newValue) +{ + #ifdef PBZIP_DEBUG + unsigned long long thid = (unsigned long long) pthread_self(); + fprintf(stderr, "(%"PRIu64") setLastGoodBlock: %d -> %d\n", thid, LastGoodBlock, newValue ); + #endif + + if ( (LastGoodBlock == -1) || (newValue < LastGoodBlock) ) + { + LastGoodBlock = newValue; + + safe_cond_signal(&ErrStateChangeCond); + safe_cond_signal(&OutBufferHeadNotEmpty); + + // wake up all other possibly blocked on cond threads + pthread_cond_broadcast(notTooMuchNumBuffered); + if (FifoQueue != NULL) + { + pthread_cond_broadcast(FifoQueue->notFull); + pthread_cond_broadcast(FifoQueue->notEmpty); + } + } +} + +inline void syncSetLastGoodBlock(int newValue) +{ + safe_mutex_lock(OutMutex); + setLastGoodBlock(newValue); + safe_mutex_unlock(OutMutex); +} + +inline int syncGetLastGoodBlock() +{ + int ret; + safe_mutex_lock(OutMutex); + ret = LastGoodBlock; + safe_mutex_unlock(OutMutex); + + return ret; +} + +inline bool isIgnoredTrailingGarbage() +{ + return (IgnoreTrailingGarbageFlag != 0); +} + /* ********************************************************* Print error message and optionally exit or abort @@ -630,6 +717,226 @@ return exitCode; } +/** + * + * @return -1 - terminate flag set (error) + * 0 - prev block is OK + * 2 - lower block number already in error state + */ +int waitForPreviousBlock(int blockNum) +{ + #ifdef PBZIP_DEBUG + unsigned long long thid = (unsigned long long) pthread_self(); + fprintf(stderr, "(%"PRIu64") waitForPreviousBlock before check: LastGoodBlock=%d; blockNum=%d; NextBlockToWrite=%d\n", + thid, + LastGoodBlock, blockNum, NextBlockToWrite ); + #endif + + for (;;) + { + if (syncGetTerminateFlag() != 0) + { + #ifdef PBZIP_DEBUG + fprintf(stderr, "(%"PRIu64") waitForPreviousBlock terminated [%d]: blockNum=%d\n", + thid, -1, blockNum ); + #endif + return -1; + } + + safe_mutex_lock(OutMutex); + + #ifdef PBZIP_DEBUG + fprintf(stderr, "(%"PRIu64") waitForPreviousBlock before check: LastGoodBlock=%d; blockNum=%d; NextBlockToWrite=%d\n", + thid, LastGoodBlock, blockNum, NextBlockToWrite ); + #endif + + if (blockNum <= NextBlockToWrite) + { + #ifdef PBZIP_DEBUG + fprintf(stderr, "(%"PRIu64") waitForPreviousBlock exit [%d]: LastGoodBlock=%d; blockNum=%d; NextBlockToWrite=%d\n", + thid, 0, LastGoodBlock, blockNum, NextBlockToWrite ); + #endif + safe_mutex_unlock(OutMutex); + return 0; + } + + if ( (LastGoodBlock != -1) && (LastGoodBlock < blockNum) ) + { + #ifdef PBZIP_DEBUG + fprintf(stderr, "(%"PRIu64") waitForPreviousBlock exit [%d]: LastGoodBlock=%d; blockNum=%d; NextBlockToWrite=%d\n", + thid, 2, LastGoodBlock, blockNum, NextBlockToWrite ); + #endif + safe_mutex_unlock(OutMutex); + return 2; + } + + #ifdef PBZIP_DEBUG + fprintf(stderr, "(%"PRIu64") waitForPreviousBlock to sleep: LastGoodBlock=%d; blockNum=%d; NextBlockToWrite=%d\n", + thid, LastGoodBlock, blockNum, NextBlockToWrite ); + #endif + + safe_cond_timed_wait(&ErrStateChangeCond, OutMutex, 1, "waitForPreviousBlock"); + + safe_mutex_unlock(OutMutex); + } +} + +/** + * + * @param errBlockNumber + * @param outSequenceNumber + * @return Last input block not after the given which resulted in good out blocks. + * -1 if such don't exist. + */ +inline int getLastGoodBlockBeforeErr(int errBlockNumber, int outSequenceNumber) +{ + // if we got the error just in the beginning of a bzip2 stream + if ( outSequenceNumber != -1 ) + { + return errBlockNumber; + } + else + { + return errBlockNumber - 1; + } +} + +/** + * Helper function delegating to handle_error with the relevant error + * message + * + * @param bzret + * @param fileData + * @param outSequenceNumber + * @param strm + * @param errmsg + * @param exitCode + * @return exitCode is returned + */ +inline int issueDecompressError(int bzret, const outBuff * fileData, + int outSequenceNumber, const bz_stream & strm, const char * errmsg, + int exitCode) +{ + handle_error(EF_EXIT, exitCode, + "pbzip2: %s: ret=%d; block=%d; seq=%d; isLastInSeq=%d; avail_in=%d\n", + errmsg, bzret, fileData->blockNumber, + outSequenceNumber, (int)fileData->isLastInSequence, strm.avail_in); + return exitCode; +} + +/** + * Handle an error condition which is either trailing garbage-like one or not. + * + * + * @param bzret + * @param fileData + * @param outSequenceNumber + * @param strm + * @param errmsg + * @param isTrailingGarbageErr + * @return ret < 0 - fatal error; + * 0 - OK (no error at all); + * 1 - first block of ignored trailing garbage; + * 2 - error already signalled for earlier block + */ +int decompressErrCheckSingle(int bzret, const outBuff * fileData, + int outSequenceNumber, const bz_stream & strm, const char * errmsg, + bool isTrailingGarbageErr) +{ + int lastGoodBlock = getLastGoodBlockBeforeErr(fileData->blockNumber, outSequenceNumber); + + #ifdef PBZIP_DEBUG + fprintf(stderr, "enter decompressErrCheckSingle: msg=%s; ret=%d; block=%d; seq=%d; isLastInSeq=%d; avail_in=%d; lastGoodBlock=%d\n", + errmsg, bzret, fileData->blockNumber, + outSequenceNumber, (int)fileData->isLastInSequence, strm.avail_in, lastGoodBlock); + #endif + + if ( (lastGoodBlock == -1) || !isIgnoredTrailingGarbage() ) + { + issueDecompressError(bzret, fileData, outSequenceNumber, strm, errmsg, -1); + return -1; + } + else + { + // Cut off larger block numbers + syncSetLastGoodBlock(lastGoodBlock); + // wait until the state of previous block is known + int prevState = waitForPreviousBlock(lastGoodBlock); + + if (prevState == 0) + { + // we're the first error block + + if (isTrailingGarbageErr) + { + // Trailing garbage detected and ignored - not a fatal warning + if (QuietMode != 1) + { + fprintf(stderr, "pbzip2: *WARNING: Trailing garbage after EOF ignored!\n"); + } + return 1; + } + else + { + // the first error is not kind of trailing garbage -> fatal one + issueDecompressError(bzret, fileData, outSequenceNumber, strm, errmsg, -1); + return -1; + } + } + else if (prevState == 2) + { + // we're not the first error + return 2; + } + else // (prevState == -1) + { + // fatal state encountered + return -1; + } + } +} + +/** + * + * @param bzret + * @param fileData + * @param outSequenceNumber + * @param strm + * @return ret < 0 - fatal error; + * 0 - OK (no error at all); + * 1 - first block of ignored trailing garbage; + * 2 - error already signalled for earlier block + */ +int decompressErrCheck(int bzret, const outBuff * fileData, + int outSequenceNumber, const bz_stream & strm) +{ + if ( (bzret == BZ_STREAM_END) && + ((strm.avail_in != 0) || !fileData->isLastInSequence) ) + { + // Potential trailing garbage + return decompressErrCheckSingle(bzret, fileData, outSequenceNumber, strm, + "*ERROR during BZ2_bzDecompress - trailing garbage", true); + } + else if ( (bzret != BZ_STREAM_END) && (bzret != BZ_OK) ) + { + return decompressErrCheckSingle(bzret, fileData, outSequenceNumber, strm, + "*ERROR during BZ2_bzDecompress - failure exit code", false); + } + else if ( strm.avail_in != 0 ) + { + return decompressErrCheckSingle(bzret, fileData, outSequenceNumber, strm, + "*ERROR unconsumed in after BZ2_bzDecompress loop", false); + } + else if ( (bzret != BZ_STREAM_END) && fileData->isLastInSequence ) + { + return decompressErrCheckSingle(bzret, fileData, outSequenceNumber, strm, + "*ERROR on decompress - last in segment reached before BZ_STREAM_END", + false); + } + + return 0; +} + /* * Initialize and set thread signal mask */ @@ -871,6 +1178,49 @@ return (s != searchBufEnd) ? s : NULL; } +/** + * Check for interrupt conditions - report if any and perform the relevant + * cleanup + * + * @param hInfile + * @param fileData + * @param lastBlock + * @return 0 - not interrupted; 1 - interrupted (terminate flag or other error encountered) + */ +int producerDecompressCheckInterrupt(int hInfile, outBuff *& fileData, int lastBlock) +{ + bool isInterrupted = false; + + if (syncGetLastGoodBlock() != -1) + { + isInterrupted = true; + + #ifdef PBZIP_DEBUG + fprintf (stderr, "producer_decompress: interrupt1 - LastGoodBlock set. " + "Last produced=%d\n", lastBlock); + #endif + } + if (syncGetTerminateFlag() != 0) + { + isInterrupted = true; + + #ifdef PBZIP_DEBUG + fprintf (stderr, "producer_decompress: interrupt2 - TerminateFlag set. " + "Last produced=%d\n", lastBlock); + #endif + } + + if (isInterrupted) + { + close(hInfile); + disposeMemorySingle(fileData); + + return 1; + } + + return 0; +} + /* ********************************************************* Function works in single pass. It's Splitting long @@ -893,6 +1243,11 @@ fprintf(stderr, " -> Bytes Read: %u bytes...\n", fileData->bufSize); #endif + if (producerDecompressCheckInterrupt(hInfile, fileData, NumBlocks) != 0) + { + return 0; + } + if (QuietMode != 1) { // give warning to user if block is larger than 250 million bytes @@ -912,6 +1267,12 @@ fprintf (stderr, "producer: queue FULL.\n"); #endif safe_cond_wait (fifo->notFull, fifo->mut); + + if (producerDecompressCheckInterrupt(hInfile, fileData, NumBlocks) != 0) + { + safe_mutex_unlock(fifo->mut); + return 0; + } } #ifdef PBZIP_DEBUG fprintf(stderr, "producer: Buffer: %x Size: %"PRIu64" Block: %d\n", fileData->buf, @@ -936,7 +1297,8 @@ // last stream is always dummy one (either error or eof) delete fileData; - if (bz2StreamScanner.failed() || !bz2StreamScanner.eof()) + + if (bz2StreamScanner.failed()) { handle_error(EF_EXIT, 1, "pbzip2: producer_decompress: *ERROR: when reading bzip2 input stream\n"); return -1; @@ -957,6 +1319,45 @@ return 0; } +/** + * Check for interrupt conditions - report if any and perform the relevant + * cleanup + * + * @return 0 - not interrupted; 1 - interrupted (terminate flag or other error encountered) + */ +int consumerDecompressCheckInterrupt(const outBuff * lastElement) +{ + bool isInterrupted = false; + + #ifdef PBZIP_DEBUG + unsigned long long thid = (unsigned long long) pthread_self(); + #endif + + if (syncGetTerminateFlag() != 0) + { + isInterrupted = true; + + #ifdef PBZIP_DEBUG + fprintf (stderr, "(%"PRIu64") producer_decompress: interrupt1 - TerminateFlag set.\n", thid); + #endif + } + if ( (syncGetLastGoodBlock() != -1) && + ( (lastElement == NULL) || (lastElement->blockNumber > syncGetLastGoodBlock()) ) ) + { + isInterrupted = true; + + #ifdef PBZIP_DEBUG + fprintf (stderr, "(%"PRIu64") consumer_decompress: terminating1 - LastGoodBlock set [%d].\n", thid, syncGetLastGoodBlock()); + #endif + } + + if (isInterrupted) + { + return 1; + } + + return 0; +} /* ********************************************************* @@ -970,8 +1371,9 @@ char *DecompressedData = NULL; unsigned int outSize = 0; outBuff * prevOutBlockInSequence = NULL; - int outSequenceNumber = 0; // sequence number in multi-part output blocks + int outSequenceNumber = -1; // sequence number in multi-part output blocks unsigned int processedIn = 0; + int errState = 0; bz_stream strm; strm.bzalloc = NULL; @@ -980,17 +1382,14 @@ for (;;) { - if (syncGetTerminateFlag() != 0) - { - #ifdef PBZIP_DEBUG - fprintf (stderr, "consumer: terminating1 - terminateFlag set\n"); - #endif - return (NULL); - } - safe_mutex_lock(fifo->mut); for (;;) { + if (consumerDecompressCheckInterrupt(fileData) != 0) + { + return (NULL); + } + if (!fifo->empty && (fifo->remove(fileData) == 1)) { // block retreived - break the loop and continue further @@ -1006,13 +1405,24 @@ // finished - either OK or terminated forcibly pthread_mutex_unlock(fifo->mut); // BZ2_bzDecompressEnd( &strm ); - if (lastFileData != NULL) + + if ((syncGetTerminateFlag() == 0) && (outSequenceNumber != -1)) { - delete lastFileData; + handle_error(EF_EXIT, -1, "pbzip2: *ERROR on decompress - " + "premature end of archive stream (block=%d; seq=%d; outseq=%d)!\n", + lastFileData->blockNumber, + lastFileData->sequenceNumber, + outSequenceNumber); } #ifdef PBZIP_DEBUG - fprintf (stderr, "consumer: exiting2\n"); + else + { + fprintf (stderr, "consumer: exiting2\n"); + } #endif + + disposeMemorySingle( lastFileData ); + return (NULL); } @@ -1024,6 +1434,7 @@ } #ifdef PBZIP_DEBUG + fprintf(stderr, "consumer: FileData: %x\n", fileData); fprintf(stderr, "consumer: Buffer: %x Size: %u Block: %d\n", fileData->buf, (unsigned)fileData->bufSize, fileData->blockNumber); #endif @@ -1090,12 +1501,27 @@ bzret = BZ2_bzDecompress(&strm); processedIn += (availIn - strm.avail_in); + #ifdef PBZIP_DEBUG + fprintf(stderr, "decompress: BZ2_bzDecompress=%d; block=%d; seq=%d; prev=%llx; avail_in=%u; avail_out=%u\n", + bzret, + fileData->blockNumber, outSequenceNumber, + (unsigned long long) prevOutBlockInSequence, + strm.avail_in, strm.avail_out); + #endif + // issue out block if out buffer is full or stream end is detected if ( ((bzret == BZ_OK) && strm.avail_out == 0) || (bzret == BZ_STREAM_END) ) { outBuff * addret = NULL; unsigned int len = outSize - strm.avail_out; bool isLast = (bzret == BZ_STREAM_END); + + if ( isLast && ( (strm.avail_in != 0) || !fileData->isLastInSequence ) ) + { + // trailng garbage detected + syncSetLastGoodBlock(fileData->blockNumber); + } + if (outSequenceNumber>0) { ++outSequenceNumber; @@ -1122,10 +1548,7 @@ } else // sequenceNumber = 0 { - if (bzret == BZ_OK) - { - ++outSequenceNumber; - } + outSequenceNumber = (bzret == BZ_OK) ? 1 : 0; addret = outputBufferAdd(outBuff( DecompressedData, len, fileData->blockNumber, @@ -1150,40 +1573,25 @@ DecompressedData = NULL; } } - - if ((bzret != BZ_STREAM_END) && (bzret != BZ_OK)) - { - handle_error(EF_EXIT, -1, "pbzip2: *ERROR during BZ2_bzDecompress: ret=%d; block=%d; seq=%d; avail_in=%d\n", - bzret, fileData->blockNumber, outSequenceNumber, strm.avail_in); - return (NULL); - } - - if (strm.avail_in != 0) - { - handle_error(EF_EXIT, -1, "pbzip2: *ERROR unconsumed in after BZ2_bzDecompress loop:" - "ret=%d; block=%d; seq=%d; avail_in=%d\n", - bzret, fileData->blockNumber, outSequenceNumber, strm.avail_in); - return (NULL); - } + /* + * < 0 - fatal error; + * 0 - OK (no error at all); + * 1 - first block of ignored trailing garbage; + * 2 - error already signalled for earlier block + */ + errState = decompressErrCheck(bzret, fileData, outSequenceNumber, strm); + if (bzret == BZ_STREAM_END) { - if (!(fileData->isLastInSequence)) - { - handle_error(EF_EXIT, -1, "pbzip2: *ERROR on decompress - """ - "in segments for stream ended but BZ_STREAM_END not reached: ret=%d; block=%d; seq=%d\n", - bzret, fileData->blockNumber, outSequenceNumber); - return (NULL); - } - bzret = BZ2_bzDecompressEnd(&strm); - if (bzret != BZ_OK) + if ( (bzret != BZ_OK) && ((errState == 0) || (errState == 1)) ) { handle_error(EF_EXIT, -1, "pbzip2: *ERROR during BZ2_bzDecompressEnd: %d\n", bzret); return (NULL); } - outSequenceNumber = 0; + outSequenceNumber = -1; prevOutBlockInSequence = NULL; } @@ -1200,6 +1608,14 @@ fflush(stderr); #endif + if (errState != 0) + { + #ifdef PBZIP_DEBUG + fprintf (stderr, "consumer: exiting prematurely: errState=%d\n", errState); + #endif + + return (NULL); + } } // for #ifdef PBZIP_DEBUG @@ -1283,12 +1699,24 @@ (unsigned long long)lastnext); #endif + if ( (LastGoodBlock != -1) && (NextBlockToWrite > LastGoodBlock) ) + { + #ifdef PBZIP_DEBUG + fprintf (stderr, "fileWriter [b:%d]: quit - LastGoodBlock=%d\n", + currBlock, LastGoodBlock); + #endif + safe_mutex_unlock(OutMutex); + + break; + } + if ((OutputBuffer[outBufferPos].buf == NULL) && ((prevBlockInSequence == NULL) || (prevBlockInSequence->next == NULL))) { + safe_cond_timed_wait(&OutBufferHeadNotEmpty, OutMutex, 1, "fileWriter"); safe_mutex_unlock(OutMutex); // sleep a little so we don't go into a tight loop using up all the CPU - usleep(50000); + // usleep(50000); continue; } else @@ -1358,6 +1786,7 @@ } // --NumBufferedBlocks; // to be removed safe_cond_broadcast(notTooMuchNumBuffered); + safe_cond_broadcast(&ErrStateChangeCond); safe_mutex_unlock(OutMutex); if (outBlock->sequenceNumber > 2) @@ -1434,20 +1863,20 @@ } safe_cond_broadcast(notTooMuchNumBuffered); // not really needed - if (QuietMode != 1) - { - // print current completion status - percentComplete = 100; - - #ifdef PBZIP_DEBUG - fprintf(stderr, "Completed: %d%% NextBlockToWrite: %d/%u \r", percentComplete, NextBlockToWrite, NumBufferedBlocksMax); + if (QuietMode != 1) + { + // print current completion status + percentComplete = 100; + + #ifdef PBZIP_DEBUG + fprintf(stderr, "Completed: %d%% NextBlockToWrite: %d/%u \r", percentComplete, NextBlockToWrite, NumBufferedBlocksMax); + fflush(stderr); + #else + + fprintf(stderr, "Completed: %d%% \r", percentComplete); fflush(stderr); - #else - - fprintf(stderr, "Completed: %d%% \r", percentComplete); - fflush(stderr); - #endif - } + #endif + } return (NULL); } @@ -1717,7 +2146,13 @@ // see if we are outputting to stdout if (OutputStdOut == 0) { - stream = fopen(OutFilename, "wb"); + stream = fopen(OutFilename, "wb"); + if (stream == NULL) + { + handle_error(EF_NOQUIT, -1, + "pbzip2: *ERROR: Could not open output file [%s]! Aborting...\n", OutFilename); + return -1; + } } else { @@ -2248,10 +2683,7 @@ q->size = queueSize; - q->empty = 1; - q->full = 0; - q->head = 0; - q->tail = 0; + q->clear(); q->mut = NULL; q->mut = new(std::nothrow) pthread_mutex_t; @@ -2365,11 +2797,36 @@ safe_mutex_unlock(OutMutex); } +/** + * Get output buffer index corresponding to the given absolute blockNumber + * (buffer is used in circular mode) + * + * @param blockNum - absolute block number to translate + * @return 0-based Output Buffer index where blockNum data should go + */ +inline size_t getOutputBufferPos(int blockNum) +{ + // calculate output buffer position (used in circular mode) + size_t outBuffPos = OutBufferPosToWrite + blockNum - NextBlockToWrite; + + if (outBuffPos >= NumBufferedBlocksMax) + { + outBuffPos -= NumBufferedBlocksMax; + } + + return outBuffPos; +} + +/** + * Add next element to the given out buffer tail. + * + */ outBuff * outputBufferSeqAddNext(outBuff * preveElement, outBuff * newElement) { safe_mutex_lock(OutMutex); - while (NumBufferedTailBlocks >= NumBufferedBlocksMax) + while ((NumBufferedTailBlocks >= NumBufferedBlocksMax) && + (preveElement->buf != NULL)) { if (syncGetTerminateFlag() != 0) { @@ -2380,6 +2837,15 @@ return NULL; } + if ( (LastGoodBlock != -1) && (LastGoodBlock < newElement->blockNumber) ) + { + #ifdef PBZIP_DEBUG + fprintf (stderr, "%s: terminating3 - LastGoodBlock set\n", "consumer"); + #endif + pthread_mutex_unlock(OutMutex); + return NULL; + } + #ifdef PBZIP_DEBUG fprintf (stderr, "%s/outputBufferSeqAddNext: Throttling from FileWriter backlog: %d\n", "consumer", NumBufferedBlocks); #endif @@ -2390,6 +2856,14 @@ ++NumBufferedTailBlocks; + // size_t outBufPos = getOutputBufferPos(newElement->blockNumber); + if (preveElement->buf == NULL) + { + // fileWriter has already consumed the previous block. Let it know + // for that one early + safe_cond_signal(&OutBufferHeadNotEmpty); + } + safe_mutex_unlock(OutMutex); return newElement; @@ -2399,10 +2873,8 @@ * Store an item in OutputBuffer out bin. Synchronization is embedded to protect * from simultaneous access. * - * @param in - item buffer - * @param bufSize - item buffer size - * @param blockNum - block number in the whole stream (not the position in buffer) - * @param caller - calling function (used for logging and debug purposes) + * @param elem - output buffer element to add + * @param caller - used for debug purposes (caller function name) * * @return pointer to added element on success; NULL - on error */ @@ -2424,6 +2896,15 @@ return NULL; } + if ( (LastGoodBlock != -1) && (LastGoodBlock < element.blockNumber) ) + { + #ifdef PBZIP_DEBUG + fprintf (stderr, "%s: terminating3 - LastGoodBlock set\n", "consumer"); + #endif + pthread_mutex_unlock(OutMutex); + return NULL; + } + #ifdef PBZIP_DEBUG fprintf (stderr, "%s: Throttling from FileWriter backlog: %d\n", caller, NumBufferedBlocks); #endif @@ -2431,15 +2912,16 @@ } // calculate output buffer position (used in circular mode) - size_t outBuffPos = OutBufferPosToWrite + element.blockNumber - NextBlockToWrite; - if (outBuffPos >= NumBufferedBlocksMax) - { - outBuffPos -= NumBufferedBlocksMax; - } + size_t outBuffPos = getOutputBufferPos(element.blockNumber); OutputBuffer[outBuffPos] = element; ++NumBufferedBlocks; + if (NextBlockToWrite == element.blockNumber) + { + safe_cond_signal(&OutBufferHeadNotEmpty); + } + safe_mutex_unlock(OutMutex); return &(OutputBuffer[outBuffPos]); @@ -2706,8 +3188,8 @@ */ void banner() { - fprintf(stderr, "Parallel BZIP2 v1.1.1 - by: Jeff Gilchrist [http://compression.ca]\n";); - fprintf(stderr, "[Apr. 17, 2010] (uses libbzip2 by Julian Seward)\n"); + fprintf(stderr, "Parallel BZIP2 v1.1.2 - by: Jeff Gilchrist [http://compression.ca]\n";); + fprintf(stderr, "[Feb. 19, 2011] (uses libbzip2 by Julian Seward)\n"); fprintf(stderr, "Major contributions: Yavor Nikolov <[email protected]>\n"); return; @@ -2755,11 +3237,14 @@ fprintf(stderr, " -t,--test Test compressed file integrity\n"); fprintf(stderr, " -v,--verbose Verbose mode\n"); fprintf(stderr, " -V,--version Display version info for pbzip2 then exit\n"); - fprintf(stderr, " -z,--compress Compress file (default)\n\n"); + fprintf(stderr, " -z,--compress Compress file (default)\n"); + fprintf(stderr, " --ignore-trailing-garbage=# Ignore trailing garbage flag (1 - ignored; 0 - forbidden)\n"); + fprintf(stderr, "\n"); fprintf(stderr, "Example: pbzip2 -b15vk myfile.tar\n"); fprintf(stderr, "Example: pbzip2 -p4 -r -5 myfile.tar second*.txt\n"); fprintf(stderr, "Example: tar cf myfile.tar.bz2 --use-compress-prog=pbzip2 dir_to_compress/\n"); - fprintf(stderr, "Example: pbzip2 -d -m500 myfile.tar.bz2\n\n"); + fprintf(stderr, "Example: pbzip2 -d -m500 myfile.tar.bz2\n"); + fprintf(stderr, "\n"); exit(-1); } @@ -2805,6 +3290,7 @@ int numBlocks = 0; int blockSize = 9*100000; int maxMemory = 100000000; + int maxMemorySwitch = 0; int decompress = 0; int compress = 0; int testFile = 0; @@ -2849,6 +3335,20 @@ { decompress = OutputStdOut = keep = 1; } + + #ifdef IGNORE_TRAILING_GARBAGE + // default behavior is hard-coded (still dynamically changeable) + IgnoreTrailingGarbageFlag = IGNORE_TRAILING_GARBAGE; + #else + // default depends on program name + if ((strcmp(progName, "bzip2") == 0) || (strcmp(progName, "BZIP2") == 0) || + (strcmp(progName, "bunzip2") == 0) || (strcmp(progName, "BUNZIP2") == 0) || + (strcmp(progName, "bzcat") == 0) || (strcmp(progName, "BZCAT") == 0)) + { + // Favour traditional non-parallel bzip2 behavior + IgnoreTrailingGarbageFlag = 1; + } + #endif // IGNORE_TRAILING_GARBAGE FileListCount = 0; FileList = new(std::nothrow) char *[argc]; @@ -2944,6 +3444,18 @@ { banner(); exit(0); } + else if (strcmp(argv[i], "--ignore-trailing-garbage") == 0 ) + { + IgnoreTrailingGarbageFlag = 1; + } + else if (strcmp(argv[i], "--ignore-trailing-garbage=1") == 0 ) + { + IgnoreTrailingGarbageFlag = 1; + } + else if (strcmp(argv[i], "--ignore-trailing-garbage=0") == 0 ) + { + IgnoreTrailingGarbageFlag = 0; + } continue; } @@ -3001,7 +3513,7 @@ blockSize = atoi(cmdLineTemp)*100000; if ((blockSize < 100000) || (blockSize > 1000000000)) { - fprintf(stderr,"pbzip2: *ERROR: File block size Min: 100k and Max: 10000k! Aborting...\n"); + fprintf(stderr,"pbzip2: *ERROR: File block size Min: 100k and Max: 1000000k! Aborting...\n"); return 1; } j += cmdLineTempCount; @@ -3023,11 +3535,12 @@ strncpy(cmdLineTemp, argv[i]+j+1, cmdLineTempCount); cmdLineTemp[cmdLineTempCount] = '\0'; maxMemory = atoi(cmdLineTemp)*1000000; - if ((maxMemory < 1000000) || (maxMemory > 1000000000)) + if ((maxMemory < 1000000) || (maxMemory > 2000000000)) { - fprintf(stderr,"pbzip2: *ERROR: Memory usage size Min: 1MB and Max: 1000MB! Aborting...\n"); + fprintf(stderr,"pbzip2: *ERROR: Memory usage size Min: 1MB and Max: 2000MB! Aborting...\n"); return 1; } + maxMemorySwitch = 1; j += cmdLineTempCount; #ifdef PBZIP_DEBUG fprintf(stderr, "-m%d\n", maxMemory); @@ -3285,6 +3798,12 @@ if (ChildThreadStackSize > 0) fprintf(stderr, " Stack Size: %d KB\n", ChildThreadStackSize/1024); #endif + + if (decompress == 1) + { + fprintf(stderr, " Ignore Trailng Garbage: %s\n", + (IgnoreTrailingGarbageFlag == 1) ? "on" : "off" ); + } } fprintf(stderr, "-------------------------------------------\n"); } @@ -3647,6 +4166,24 @@ fprintf(stderr, "pbzip2: maxMemory: %d blockSize: %d\n", maxMemory, blockSize); fprintf(stderr, "pbzip2: NumBufferedBlocksMax: %u\n", NumBufferedBlocksMax); #endif + // check to see if our max buffered blocks is less than numCPU, if yes increase maxMemory + // to support numCPU requested unless -m switch given by user + if (NumBufferedBlocksMax < (size_t)numCPU) + { + if (maxMemorySwitch == 0) + { + NumBufferedBlocksMax = numCPU; + if (QuietMode != 1) + fprintf(stderr, "*Warning* Max memory limit increased to %d MB to support %d CPUs\n", ((NumBufferedBlocksMax + (numCPU * 2)) * blockSize)/1000000, numCPU); + } + else + { + if (QuietMode != 1) + fprintf(stderr, "*Warning* CPU usage and performance may be suboptimal due to max memory limit.\n"); + } + } + + LastGoodBlock = -1; // create output buffer outputBufferInit(NumBufferedBlocksMax); @@ -3777,6 +4314,11 @@ } } + if (syncGetTerminateFlag() != 0) + { + errLevel = 1; + } + if (OutputStdOut == 0) { // write store file meta data to output file @@ -3806,10 +4348,7 @@ // reclaim memory OutputBuffer.clear(); - fifo->empty = 1; - fifo->full = 0; - fifo->head = 0; - fifo->tail = 0; + fifo->clear(); if (QuietMode != 1) fprintf(stderr, "-------------------------------------------\n"); diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/pbzip2.h new/pbzip2-1.1.2/pbzip2.h --- old/pbzip2-1.1.1/pbzip2.h 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/pbzip2.h 2011-02-19 11:10:22.000000000 +0100 @@ -3,6 +3,11 @@ * Author: Yavor Nikolov * * Created on March 6, 2010, 10:18 PM + * + * Change History: + * 2010-05-20 - by Yavor Nikolov + * - Transformed input queue as queue of queues to avoid deadlock and conten + * tion issues. */ #ifndef _PBZIP2_H @@ -47,6 +52,7 @@ #endif #include <stdio.h> +#include <pthread.h> } // uncomment for debug output @@ -89,6 +95,7 @@ unsigned int inSize; // original size before compression/decompressoin bool isLastInSequence; outBuff * next; // next in longer sequence of buffers for this block + //outBuff * last; // last in sequence (here as quick hack since global one would be enough) outBuff( char * aBuf = NULL, @@ -104,7 +111,8 @@ sequenceNumber(aSequenceNumber), inSize(aInSize), isLastInSequence(isLast), - next(aNext) + next(aNext)//, + //last(NULL) {} } outBuff; @@ -122,20 +130,72 @@ ElementTypePtr *qData; long size; + long count; // actual element count, including tail-queue ones long head, tail; int full, empty; + int topLevelFull, topLevelEmpty; pthread_mutex_t *mut; pthread_cond_t *notFull, *notEmpty; pthread_t *consumers; + ElementTypePtr lastElement; // most recently added element + + queue(): count(0), lastElement(NULL) + {} + + /** + * Reset the queue putting it to initial empty state. + */ + void clear() + { + empty = 1; + full = 0; + head = 0; + tail = 0; + count = 0; + lastElement = NULL; + topLevelFull = 0; + topLevelEmpty = 1; + } void add(ElementTypePtr element) { - qData[tail] = element; - ++tail; + #ifdef PBZIP_DEBUG + fprintf (stderr, "queue::add: elem=%llx\n", + (unsigned long long)element); + + if (element != NULL) + { + fprintf (stderr, " queue::add: seq=%d; blk=%d; islast=%d\n", + element->sequenceNumber, + element->blockNumber, + (int)element->isLastInSequence); + } + #endif + + if ( element->sequenceNumber > 1 ) + { + // multi-part sequence: append to same one + lastElement->next = element; + } + else + { + // primary element (either first in sequence; or a standalone one) + qData[tail] = element; + ++tail; + + if (tail == size) + tail = 0; + + if (tail == head) + topLevelFull = 1; - if (tail == size) - tail = 0; - if (tail == head) + topLevelEmpty = 0; + } + + lastElement = element; + ++count; + + if (count == size) full = 1; empty = 0; @@ -143,6 +203,7 @@ /** * Remove the head returning it into element. If the given element is + * tail of multi-segment sequence - just moves to next segment. * * @param element - removed element is copied here * @return 1 on success; 0 - on denied request; -1 - on error @@ -152,33 +213,62 @@ ElementTypePtr & headElem = qData[head]; #ifdef PBZIP_DEBUG - fprintf (stderr, "queue::remove: head=%llx; elem=%llx\n", + fprintf (stderr, "queue::remove: head=%llx; elem=%llx; count=%ld\n", (unsigned long long)headElem, - (unsigned long long)element); + (unsigned long long)element, + count); if (headElem != NULL) { - fprintf (stderr, " queue::remove: head: seq=%d; blk=%d; islast=%d\n", + fprintf (stderr, " queue::remove: head: seq=%d; blk=%d; islast=%d\n", headElem->sequenceNumber, headElem->blockNumber, (int)headElem->isLastInSequence); } + + if (element != NULL) + { + fprintf (stderr, " queue::remove: element: seq=%d; blk=%d; islast=%d\n", + element->sequenceNumber, + element->blockNumber, + (int)element->isLastInSequence); + } #endif - if ((headElem->sequenceNumber > 1) && - ((element == NULL) || (headElem->blockNumber != element->blockNumber))) + if ( (element != NULL) && !element->isLastInSequence ) + { + if (element->next != NULL) + { + element = element->next; + } + else + { + // 2+ part of long-sequence BZ2 stream. Next + // segment is not ready yet. + return 0; + } + } + else if (topLevelEmpty) { - // 2+ part of long-sequence BZ2 stream. Deny if requestor is not - // aware of that return 0; } + else + { + element = headElem; + ++head; + + if (head == size) + head = 0; - element = headElem; - ++head; + if (head == tail) + topLevelEmpty = 1; + + topLevelFull = 0; + } - if (head == size) - head = 0; - if (head == tail) + --count; + + if (count == 0) empty = 1; full = 0; @@ -226,6 +316,19 @@ } } +/** + * Dispose the given buffer memory if not NULL and make it NULL. Provided + * buffer should be allocated with new. + */ +template <typename C> +inline void disposeMemorySingle(C *& pBuff) +{ + if (pBuff != NULL) + { + delete pBuff; + pBuff = NULL; + } +} /** * Check if a given string ends with a given suffix ignoring case difference. diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/pbzip2-1.1.1/pbzip2.spec new/pbzip2-1.1.2/pbzip2.spec --- old/pbzip2-1.1.1/pbzip2.spec 2010-04-17 20:37:39.000000000 +0200 +++ new/pbzip2-1.1.2/pbzip2.spec 2011-02-19 11:10:22.000000000 +0100 @@ -1,5 +1,5 @@ Name: pbzip2 -Version: 1.1.1 +Version: 1.1.2 Release: 1%{?dist} Summary: Parallel implementation of bzip2 URL: http://www.compression.ca/pbzip2/ @@ -52,6 +52,9 @@ %changelog +* Sat Feb 19 2011 Jeff Gilchrist <[email protected]> - 1.1.2-1 +- Release 1.1.2 + * Sat Apr 17 2010 Jeff Gilchrist <[email protected]> - 1.1.1-1 - Release 1.1.1 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Remember to have fun... -- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
