pcompress.c

/*
 * This file is a part of Pcompress, a chunked parallel multi-
 * algorithm lossless compression and decompression program.
 *
 * Copyright (C) 2012-2013 Moinak Ghosh. All rights reserved.
 * Use is subject to license terms.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this program.
 * If not, see <http://www.gnu.org/licenses/>.
 *
 * moinakg@belenix.org, http://moinakg.wordpress.com/
 *
 */

/*
 * pcompress - Do a chunked parallel compression/decompression and archiving
 * of one or more files.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/param.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <strings.h>
#include <limits.h>
#include <unistd.h>
#include <libgen.h>
#include <utils.h>
#include <pcompress.h>
#include <allocator.h>
#include <rabin_dedup.h>

#ifndef _MPLV2_LICENSE_
#include <lzp.h>
#endif

#include <transpose.h>
#include <delta2/delta2.h>
#include <crypto/crypto_utils.h>
#include <crypto_xsalsa20.h>
#include <ctype.h>
#include <errno.h>
#include <pc_archive.h>
#include <filters/dispack/dis.hpp>
#include "filters/dict/DictFilter.h"

/*
 * We use 8MB chunks by default.
 */
#define	DEFAULT_CHUNKSIZE	(8 * 1024 * 1024)
#define	EIGHTY_PCT(x) ((x) - ((x)/5))

struct wdata {
	struct cmp_data **dary;
	int wfd;
	int nprocs;
	int64_t chunksize;
	pc_ctx_t *pctx;
};

pthread_mutex_t opt_parse = PTHREAD_MUTEX_INITIALIZER;

static void * writer_thread(void *dat);
static int init_algo(pc_ctx_t *pctx, const char *algo, int bail);
extern uint32_t lzma_crc32(const uint8_t *buf, uint64_t size, uint32_t crc);

void DLL_EXPORT
usage(pc_ctx_t *pctx)
{

	fprintf(stderr,
"\nPcompress Version %s\n"
"License: %s\n\n"
"See README.md for detailed usage.\n\n"
"Standard Usage\n"
"==============\n"
"    Standard usage consists of a few common options to control basic behavior with auto-\n"
"    setting of various other parameters.\n\n"
"    Archiving\n"
"    ---------\n"
"       %s -a [-v] [-l <compress level>] [-s <chunk size>] [-c <algorithm>]\n"
"                    [<file1> <directory1> <file2> ...] [-t <number>] [-S <chunk checksum>]\n"
"                    <archive filename or '-'>\n\n"
"       Archives a given set of files and/or directories into a compressed PAX archive which\n"
"       is then compressed.\n\n"
"       -a       Enables the archive mode.\n"
"       -l <compress level>\n"
"                Select a compression level from 1 (fast) to 14 (slow). Default: 6\n\n"
"       -s <chunk size>\n"
"                Specifies the maximum chunk size to split the data for parallelism. Values\n"
"                can be in bytes or with suffix(k - KB, m - MB, g - GB). Default: 8m\n"
"                Larger chunks can produce better compression at the cost of memory.\n\n"
"       -c <algorithm>\n"
"                The compression algorithm. Default algorithm when archiving is adapt2.\n"
"       -v       Enables verbose mode.\n\n"
"       -t <number>\n"
"                Sets the number of compression threads. Default: core count.\n"
"       -T       Disable separate metadata stream.\n"
"       -S <chunk checksum>\n"
"                The chunk verification checksum. Default: BLAKE256. Others are: CRC64, SHA256,\n"
"                SHA512, KECCAK256, KECCAK512, BLAKE256, BLAKE512.\n"
"       <archive filename>\n"
"                Pathname of the resulting archive. A '.pz' extension is automatically added\n"
"                if not already present. This can be '-' to output to stdout.\n\n"
"    Single File Compression\n"
"    -----------------------\n"
"       %s -c <algorithm> [-l <compress level>] [-s <chunk size>] [-p] [<file>]\n"
"                 [-t <number>] [-S <chunk checksum>] [<target file or '-'>]\n\n"
"       Takes a single file as input and produces a compressed file. Archiving is not performed.\n"
"       This can also work in streaming mode.\n\n"
"       -c <algorithm>\n"
"                See above. Also see section 'Compression Algorithms' in README.md for details.\n"
"       -l <compress level>\n"
"       -s <chunk size>\n"
"       -t <number>\n"
"       -S <chunk checksum>\n"
"                See above.\n"
"                Note: In singe file compression mode with adapt2 or adapt algorithm, larger\n"
"                      chunks may not necessarily produce better compression.\n"
"       -p       Make Pcompress work in streaming mode. Input is stdin, output is stdout.\n\n"
"       <target file>\n"
"                Pathname of the compressed file to be created or '-' for stdout.\n\n"
"    Decompression, Listing and Archive extraction\n"
"    ---------------------------------------------\n"
"       %s <-d|-i>  [-m] [-K] <compressed file or '-'> [<target file or directory>]\n\n"
"       -d        Extract archive to target dir or current dir.\n"
"       -i        Only list contents of the archive, do not extract.\n\n"
"       -m        Enable restoring *all* permissions, ACLs, Extended Attributes etc.\n"
"                 Equivalent to the '-p' option in tar.\n"
"       -K        Do not overwrite newer files.\n"
"       -m and -K are only meaningful if the compressed file is an archive. For single file\n"
"       compressed mode these options are ignored.\n\n"
"       <compressed file>\n"
"                 Specifies the compressed file or archive. This can be '-' to indicate reading\n"
"                 from stdin while write goes to <target file>\n\n"
"       <target file or directory>\n"
"                 If single file compression was used then this is the output file.\n"
"                 Default output name if omitted: <input filename>.out\n\n"
"                 If Archiving was done then this should be the name of a directory into which\n"
"                 extracted files are restored. Default if omitted: Current directory.\n\n",
	    UTILITY_VERSION, LICENSE_STRING, pctx->exec_name, pctx->exec_name, pctx->exec_name);
	fprintf(stderr,
"    Encryption\n"
"    ----------\n"
"       -e <ALGO> Encrypt chunks with the given encrption algorithm. The ALGO parameter\n"
"                 can be one of AES or SALSA20. Both are used in CTR stream encryption\n"
"                 mode. The password can be prompted from the user or read from a file.\n"
"                 Unique keys are generated every time pcompress is run even when giving\n"
"                 the same password. Default key length is 256-bits (see -k below).\n"
"       -w <pathname>\n"
"                 Provide a file which contains the encryption password. This file must\n"
"                 be readable and writable since it is zeroed out after the password is\n"
"                 read.\n"
"       -k <key length>\n"
"                 Specify key length. Can be 16 for 128 bit or 32 for 256 bit. Default\n"
"                 is 32 for 256 bit keys.\n\n");
}

static void
show_compression_stats(pc_ctx_t *pctx)
{
	log_msg(LOG_INFO, 0, "\nCompression Statistics");
	log_msg(LOG_INFO, 0, "======================");
	log_msg(LOG_INFO, 0, "Total chunks           : %u", pctx->chunk_num);
	if (pctx->chunk_num == 0) {
		log_msg(LOG_INFO, 0, "No statistics to display.");
	} else {
		log_msg(LOG_INFO, 0, "Best compressed chunk  : %s(%.2f%%)",
		    bytes_to_size(pctx->smallest_chunk),
		    (double)pctx->smallest_chunk/(double)pctx->chunksize*100);
		log_msg(LOG_INFO, 0, "Worst compressed chunk : %s(%.2f%%)",
		    bytes_to_size(pctx->largest_chunk),
		    (double)pctx->largest_chunk/(double)pctx->chunksize*100);
		pctx->avg_chunk /= pctx->chunk_num;
		log_msg(LOG_INFO, 0, "Avg compressed chunk   : %s(%.2f%%)\n",
		    bytes_to_size(pctx->avg_chunk),
		    (double)pctx->avg_chunk/(double)pctx->chunksize*100);
	}
}

/*
 * Wrapper functions to pre-process the buffer and then call the main compression routine.
 *
 * Byte 0: A flag to indicate which pre-processor was used.
 * Byte 1 - Byte 8: Size of buffer after pre-processing
 *
 * It is possible for a buffer to be only pre-processed and not compressed by the final
 * algorithm if the final one fails to compress for some reason. However the vice versa
 * is not allowed.
 */
static int
preproc_compress(pc_ctx_t *pctx, compress_func_ptr cmp_func, void *src, uint64_t srclen,
    void *dst, uint64_t *dstlen, int level, uchar_t chdr, int btype, void *data,
    algo_props_t *props, int interesting)
{
	uchar_t *dest = (uchar_t *)dst, type = 0;
	int result;
	uint64_t _dstlen, fromlen;
	uchar_t *from, *to;
	int stype, analyzed;
	analyzer_ctx_t actx;
	DEBUG_STAT_EN(double strt, en);

	_dstlen = *dstlen;
	from = src;
	to = dst;
	fromlen = srclen;
	result = 0;
	stype = PC_SUBTYPE(btype);
	analyzed = 0;

	if (btype == TYPE_UNKNOWN || stype == TYPE_ARCHIVE_TAR || stype == TYPE_PDF ||
	    PC_TYPE(btype) & TYPE_TEXT || interesting) {
		analyze_buffer(src, srclen, &actx);
		analyzed = 1;
		if (pctx->adapt_mode)
			adapt_set_analyzer_ctx(data, &actx);
	}

	/*
	 * Dispack is used for 32-bit EXE files via a libarchive filter routine.
	 * For 64-bit exes or AR archives we apply an E8E9 CALL/JMP transform filter.
	 */
	if (pctx->exe_preprocess) {
		int processed = 0;

		if (stype == TYPE_EXE32 ||  stype == TYPE_EXE32_PE ||
		    stype == TYPE_EXE64 || stype == TYPE_ARCHIVE_AR) {
			/*
			 * If file-level Dispack did not happen for 32-bit EXEs it was
			 * most likely that the file was large. So, as a workaround,
			 * we do raw-block Dispack here. However if even this fails to
			 * get any worthwhile reduction we do E8E9 as the final
			 * fallback.
			 */
			_dstlen = fromlen;
			result = dispack_encode((uchar_t *)from, fromlen, to, &_dstlen);
			if (result != -1) {
				uchar_t *tmp;
				tmp = from;
				from = to;
				to = tmp;
				fromlen = _dstlen;
				type |= PREPROC_TYPE_DISPACK;
				processed = 1;
			}
		}

		if (!processed) {
			_dstlen = fromlen;
			memcpy(to, from, fromlen);
			if (Forward_E89(to, fromlen) == 0) {
				uchar_t *tmp;
				tmp = from;
				from = to;
				to = tmp;
				fromlen = _dstlen;
				type |= PREPROC_TYPE_E8E9;
			}
		}
	}

	/*
	 * Enabling LZP also enables the DICT filter since we are dealing with text
	 * in any case.
	 */
	if (pctx->lzp_preprocess) {
		int b_type;

		b_type = btype;
		if (analyzed) {
			b_type = actx.ten_pct.btype;
		} else {
			b_type = analyze_buffer_simple(from, fromlen);
		}

		if (PC_TYPE(b_type) & TYPE_TEXT) {
			_dstlen = fromlen;
			result = dict_encode(from, fromlen, to, &_dstlen, (stype == TYPE_DNA_SEQ));
			if (result != -1) {
				uchar_t *tmp;
				tmp = from;
				from = to;
				to = tmp;
				fromlen = _dstlen;
				type |= PREPROC_TYPE_DICT;
			}
		}
	}

#ifndef _MPLV2_LICENSE_
	if (pctx->lzp_preprocess && stype != TYPE_BMP && stype != TYPE_TIFF) {
		int hashsize, b_type;
		int64_t result;

		b_type = btype;
		if (analyzed)
			b_type = actx.thirty_pct.btype;

		if (!(PC_TYPE(b_type) & TYPE_BINARY)) {
			hashsize = lzp_hash_size(level);
			result = lzp_compress((const uchar_t *)from, to, fromlen,
					      hashsize, LZP_DEFAULT_LZPMINLEN, 0);
			if (result >= 0 && result < srclen) {
				uchar_t *tmp;
				tmp = from;
				from = to;
				to = tmp;
				fromlen = result;
				type |= PREPROC_TYPE_LZP;
			}
		}
	}
#endif

	if (pctx->enable_delta2_encode && props->delta2_span > 0 &&
	    stype != TYPE_DNA_SEQ && stype != TYPE_BMP &&
	    stype != TYPE_TIFF && stype != TYPE_MP4) {
		int b_type;

		b_type = btype;
		if (analyzed)
			b_type = actx.ten_pct.btype;

		if (!(PC_TYPE(b_type) & TYPE_TEXT)) {
			_dstlen = fromlen;
			result = delta2_encode((uchar_t *)from, fromlen, to,
					       &_dstlen, props->delta2_span,
					       pctx->delta2_nstrides);
			if (result != -1) {
				uchar_t *tmp;
				tmp = from;
				from = to;
				to = tmp;
				fromlen = _dstlen;
				type |= PREPROC_TYPE_DELTA2;
			}
		}
	}

	/*
	 * Check which is the resulting buffer. If Encoded data is already sitting
	 * in src buffer then a memcpy() is not needed.
	 * Note that from,to ptrs are swapped after every encoding stage. So if
	 * from == dst, it means that encoded data is in dst.
	 */
	if (from == dst) {
		memcpy(src, dst, fromlen);
	}
	srclen = fromlen;

	*dest = type;
	U64_P(dest + 1) = htonll(srclen);
	_dstlen = srclen;
	DEBUG_STAT_EN(strt = get_wtime_millis());
	result = cmp_func(src, srclen, dest+9, &_dstlen, level, chdr,
	    btype, data);
	DEBUG_STAT_EN(en = get_wtime_millis());

	if (result > -1 && _dstlen < srclen) {
		*dest |= PREPROC_COMPRESSED;
		*dstlen = _dstlen + 9;
		DEBUG_STAT_EN(fprintf(stderr, "Chunk compression speed %.3f MB/s\n",
		    get_mb_s(srclen, strt, en)));
	} else {
		DEBUG_STAT_EN(fprintf(stderr, "Chunk did not compress.\n"));
		/*
		 * If compression failed but one of the pre-processing succeeded then
		 * type flags will be non-zero. In that case we still indicate a success
		 * result so that decompression will reverse the pre-processing. The
		 * type flags will indicate that compression was not done and the
		 * decompress routine will not be called.
		 */
		if (type > 0) {
			memcpy(dest+1, src, srclen);
			*dstlen = srclen + 1;
			result = 0;
		} else {
			/*
			 * Not Compressed and not preprocessed. Ensure that we signal
			 * error upstream, so this is handled correctly.
			 */
			result = -1;
		}
	}
	return (result);
}

static int
preproc_decompress(pc_ctx_t *pctx, compress_func_ptr dec_func, void *src, uint64_t srclen,
    void *dst, uint64_t *dstlen, int level, uchar_t chdr, int btype, void *data,
    algo_props_t *props)
{
	uchar_t *sorc = (uchar_t *)src, type;
	int result;
	uint64_t _dstlen = *dstlen, _dstlen1 = *dstlen;
	DEBUG_STAT_EN(double strt, en);

	type = *sorc;
	++sorc;
	--srclen;
	if (type & PREPROC_COMPRESSED) {
		*dstlen = ntohll(U64_P(sorc));
		sorc += 8;
		srclen -= 8;
		DEBUG_STAT_EN(strt = get_wtime_millis());
		result = dec_func(sorc, srclen, dst, dstlen, level, chdr, btype, data);
		DEBUG_STAT_EN(en = get_wtime_millis());

		if (result < 0) return (result);
		DEBUG_STAT_EN(fprintf(stderr, "Chunk decompression speed %.3f MB/s\n",
		    get_mb_s(srclen, strt, en)));
		memcpy(src, dst, *dstlen);
		srclen = *dstlen;
	} else {
		src = sorc;
	}

	if (type & PREPROC_TYPE_DELTA2) {
		result = delta2_decode((uchar_t *)src, srclen, (uchar_t *)dst, &_dstlen);
		if (result != -1) {
			memcpy(src, dst, _dstlen);
			srclen = _dstlen;
			*dstlen = _dstlen;
			_dstlen = _dstlen1;
		} else {
			log_msg(LOG_ERR, 0, "Delta2 decoding failed.");
			return (result);
		}
	}

	if (type & PREPROC_TYPE_LZP) {
#ifndef _MPLV2_LICENSE_
		int hashsize;
		int64_t result;
		hashsize = lzp_hash_size(level);
		result = lzp_decompress((const uchar_t *)src, (uchar_t *)dst, srclen,
					hashsize, LZP_DEFAULT_LZPMINLEN, 0);
		if (result > 0) {
			memcpy(src, dst, result);
			srclen = result;
			*dstlen = result;
		} else {
			log_msg(LOG_ERR, 0, "LZP decompression failed.");
			return ((int)result);
		}
#else
		log_msg(LOG_ERR, 0, "LZP feature not available in this build"
		    " (MPLv2). Aborting.");
		return (-1);
#endif
	}

	if (type & PREPROC_TYPE_DICT) {
		result = dict_decode(src, srclen, dst, &_dstlen);
		if (result != -1) {
			memcpy(src, dst, _dstlen);
			srclen = _dstlen;
			*dstlen = _dstlen;
			_dstlen = _dstlen1;
		} else {
			log_msg(LOG_ERR, 0, "DICT decoding failed.");
			return (result);
		}
	}

	if (type & PREPROC_TYPE_E8E9) {
		_dstlen1 = srclen;
		memcpy(dst, src, srclen);
		result = Inverse_E89(dst, srclen);
		if (result != -1) {
			*dstlen = _dstlen1;
		} else {
			log_msg(LOG_ERR, 0, "E8E9 decoding failed.");
			return (result);
		}

	} else if (type & PREPROC_TYPE_DISPACK) { // Backward compatibility
		result = dispack_decode((uchar_t *)src, srclen, (uchar_t *)dst, &_dstlen1);
		if (result != -1) {
			*dstlen = _dstlen1;
		} else {
			log_msg(LOG_ERR, 0, "Dispack decoding failed.");
			return (result);
		}
	}

	if (!(type & (PREPROC_COMPRESSED|PREPROC_TYPE_DELTA2|PREPROC_TYPE_LZP|
		      PREPROC_TYPE_DISPACK|PREPROC_TYPE_DICT|PREPROC_TYPE_E8E9))
	    && type > 0) {
		log_msg(LOG_ERR, 0, "Invalid preprocessing flags: %d", type);
		return (-1);
	}
	return (0);
}

/*
 * This routine is called in multiple threads. Calls the decompression handler
 * as encoded in the file header. For adaptive mode the handler adapt_decompress()
 * in turns looks at the chunk header and calls the actual decompression
 * routine.
 */
static void *
perform_decompress(void *dat)
{
	struct cmp_data *tdat = (struct cmp_data *)dat;
	uint64_t _chunksize;
	uint64_t dedupe_index_sz, dedupe_data_sz, dedupe_index_sz_cmp, dedupe_data_sz_cmp;
	int rv = 0;
	unsigned int blknum;
	uchar_t checksum[CKSUM_MAX_BYTES];
	uchar_t HDR;
	uchar_t *cseg;
	pc_ctx_t *pctx;

	pctx = tdat->pctx;
redo:
	Sem_Wait(&tdat->start_sem);
	if (pctx->main_cancel)
		return (NULL);

	if (unlikely(tdat->cancel)) {
		tdat->len_cmp = 0;
		Sem_Post(&tdat->cmp_done_sem);
		return (0);
	}

	/*
	 * If the last read returned a 0 quit.
	 */
	if (tdat->rbytes == 0) {
		tdat->len_cmp = 0;
		goto cont;
	}

	cseg = tdat->compressed_chunk + pctx->cksum_bytes + pctx->mac_bytes;
	HDR = *cseg;
	cseg += CHUNK_FLAG_SZ;
	_chunksize = tdat->chunksize;
	if (HDR & CHSIZE_MASK) {
		uchar_t *rseg;

		tdat->rbytes -= ORIGINAL_CHUNKSZ;
		tdat->len_cmp -= ORIGINAL_CHUNKSZ;
		rseg = tdat->compressed_chunk + tdat->rbytes;
		_chunksize = ntohll(*((int64_t *)rseg));
	}

	/*
	 * If this was encrypted:
	 * Verify HMAC first before anything else and then decrypt compressed data.
	 */
	if (pctx->encrypt_type) {
		unsigned int len;
		DEBUG_STAT_EN(double strt, en);

		DEBUG_STAT_EN(strt = get_wtime_millis());
		len = pctx->mac_bytes;
		deserialize_checksum(checksum, tdat->compressed_chunk + pctx->cksum_bytes,
		    pctx->mac_bytes);
		memset(tdat->compressed_chunk + pctx->cksum_bytes, 0, pctx->mac_bytes);
		hmac_reinit(&tdat->chunk_hmac);
		hmac_update(&tdat->chunk_hmac, (uchar_t *)&tdat->len_cmp_be, sizeof (tdat->len_cmp_be));
		hmac_update(&tdat->chunk_hmac, tdat->compressed_chunk, tdat->rbytes);
		if (HDR & CHSIZE_MASK) {
			uchar_t *rseg;
			rseg = tdat->compressed_chunk + tdat->rbytes;
			hmac_update(&tdat->chunk_hmac, rseg, ORIGINAL_CHUNKSZ);
		}
		hmac_final(&tdat->chunk_hmac, tdat->checksum, &len);
		if (memcmp(checksum, tdat->checksum, len) != 0) {
			/*
			 * HMAC verification failure is fatal.
			 */
			log_msg(LOG_ERR, 0, "Chunk %d, HMAC verification failed", tdat->id);
			pctx->main_cancel = 1;
			tdat->len_cmp = 0;
			pctx->t_errored = 1;
			Sem_Post(&tdat->cmp_done_sem);
			return (NULL);
		}
		DEBUG_STAT_EN(en = get_wtime_millis());
		DEBUG_STAT_EN(fprintf(stderr, "HMAC Verification speed %.3f MB/s",
			      get_mb_s(tdat->rbytes + sizeof (tdat->len_cmp_be), strt, en)));

		/*
		 * Encryption algorithm should not change the size and
		 * encryption is in-place.
		 */
		DEBUG_STAT_EN(strt = get_wtime_millis());
		rv = crypto_buf(&(pctx->crypto_ctx), cseg, cseg, tdat->len_cmp, tdat->id);
		if (rv == -1) {
			/*
			 * Decryption failure is fatal.
			 */
			log_msg(LOG_ERR, 0, "Chunk %d, Decryption failed", tdat->id);
			pctx->main_cancel = 1;
			tdat->len_cmp = 0;
			Sem_Post(&tdat->cmp_done_sem);
			return (NULL);
		}
		DEBUG_STAT_EN(en = get_wtime_millis());
		DEBUG_STAT_EN(fprintf(stderr, "Decryption speed %.3f MB/s\n",
			      get_mb_s(tdat->len_cmp, strt, en)));
	} else if (pctx->mac_bytes > 0) {
		/*
		 * Verify header CRC32 in non-crypto mode.
		 */
		uint32_t crc1, crc2;

		crc1 = htonl(U32_P(tdat->compressed_chunk + pctx->cksum_bytes));
		memset(tdat->compressed_chunk + pctx->cksum_bytes, 0, pctx->mac_bytes);
		crc2 = lzma_crc32((uchar_t *)&tdat->len_cmp_be, sizeof (tdat->len_cmp_be), 0);
		crc2 = lzma_crc32(tdat->compressed_chunk,
		    pctx->cksum_bytes + pctx->mac_bytes + CHUNK_FLAG_SZ, crc2);
		if (HDR & CHSIZE_MASK) {
			uchar_t *rseg;
			rseg = tdat->compressed_chunk + tdat->rbytes;
			crc2 = lzma_crc32(rseg, ORIGINAL_CHUNKSZ, crc2);
		}

		if (crc1 != crc2) {
			/*
			 * Header CRC32 verification failure is fatal.
			 */
			log_msg(LOG_ERR, 0, "Chunk %d, Header CRC verification failed", tdat->id);
			pctx->main_cancel = 1;
			tdat->len_cmp = 0;
			pctx->t_errored = 1;
			Sem_Post(&tdat->cmp_done_sem);
			return (NULL);
		}

		/*
		 * Now that header CRC32 was verified, recover the stored message
		 * digest.
		 */
		deserialize_checksum(tdat->checksum, tdat->compressed_chunk, pctx->cksum_bytes);
	}

	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan || pctx->enable_rabin_global) &&
	    (HDR & CHUNK_FLAG_DEDUP)) {
		uchar_t *cmpbuf, *ubuf;

		/* Extract various sizes from dedupe header. */
		parse_dedupe_hdr(cseg, &blknum, &dedupe_index_sz, &dedupe_data_sz,
				&dedupe_index_sz_cmp, &dedupe_data_sz_cmp, &_chunksize);
		memcpy(tdat->uncompressed_chunk, cseg, RABIN_HDR_SIZE);

		/*
		 * Uncompress the data chunk first and then uncompress the index.
		 * The uncompress routines can use extra bytes at the end for temporary
		 * state/dictionary info. Since data chunk directly follows index
		 * uncompressing index first corrupts the data.
		 */
		cmpbuf = cseg + RABIN_HDR_SIZE + dedupe_index_sz_cmp;
		ubuf = tdat->uncompressed_chunk + RABIN_HDR_SIZE + dedupe_index_sz;
		if (HDR & COMPRESSED) {
			if (HDR & CHUNK_FLAG_PREPROC) {
				rv = preproc_decompress(pctx, tdat->decompress, cmpbuf,
				    dedupe_data_sz_cmp,	ubuf, &_chunksize, tdat->level,
				    HDR, pctx->btype, tdat->data, tdat->props);
			} else {
				DEBUG_STAT_EN(double strt, en);

				DEBUG_STAT_EN(strt = get_wtime_millis());
				rv = tdat->decompress(cmpbuf, dedupe_data_sz_cmp, ubuf, &_chunksize,
				    tdat->level, HDR, pctx->btype, tdat->data);
				DEBUG_STAT_EN(en = get_wtime_millis());
				DEBUG_STAT_EN(fprintf(stderr, "Chunk %d decompression speed %.3f MB/s\n",
						      tdat->id, get_mb_s(_chunksize, strt, en)));
			}
			if (rv == -1) {
				tdat->len_cmp = 0;
				log_msg(LOG_ERR, 0, "ERROR: Chunk %d, decompression failed.", tdat->id);
				pctx->t_errored = 1;
				goto cont;
			}
		} else {
			memcpy(ubuf, cmpbuf, _chunksize);
		}

		rv = 0;
		cmpbuf = cseg + RABIN_HDR_SIZE;
		ubuf = tdat->uncompressed_chunk + RABIN_HDR_SIZE;

		if (dedupe_index_sz >= 90 && dedupe_index_sz > dedupe_index_sz_cmp) {
			/* Index should be at least 90 bytes to have been compressed. */
			rv = lzma_decompress(cmpbuf, dedupe_index_sz_cmp, ubuf,
			    &dedupe_index_sz, tdat->rctx->level, 0, TYPE_BINARY, tdat->rctx->lzma_data);
		} else {
			memcpy(ubuf, cmpbuf, dedupe_index_sz);
		}

		/*
		 * Recover from transposed index.
		 */
		transpose(ubuf, cmpbuf, dedupe_index_sz, sizeof (uint32_t), COL);
		memcpy(ubuf, cmpbuf, dedupe_index_sz);

	} else {
		if (HDR & COMPRESSED) {
			if (HDR & CHUNK_FLAG_PREPROC) {
				rv = preproc_decompress(pctx, tdat->decompress, cseg, tdat->len_cmp,
				    tdat->uncompressed_chunk, &_chunksize, tdat->level, HDR, pctx->btype,
				    tdat->data, tdat->props);
			} else {
				DEBUG_STAT_EN(double strt, en);

				DEBUG_STAT_EN(strt = get_wtime_millis());
				rv = tdat->decompress(cseg, tdat->len_cmp, tdat->uncompressed_chunk,
				    &_chunksize, tdat->level, HDR, pctx->btype, tdat->data);
				DEBUG_STAT_EN(en = get_wtime_millis());
				DEBUG_STAT_EN(fprintf(stderr, "Chunk decompression speed %.3f MB/s\n",
						get_mb_s(_chunksize, strt, en)));
			}
		} else {
			memcpy(tdat->uncompressed_chunk, cseg, _chunksize);
		}
	}
	tdat->len_cmp = _chunksize;

	if (rv == -1) {
		tdat->len_cmp = 0;
		log_msg(LOG_ERR, 0, "ERROR: Chunk %d, decompression failed.", tdat->id);
		pctx->t_errored = 1;
		goto cont;
	}
	/* Rebuild chunk from dedup blocks. */
	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan) && (HDR & CHUNK_FLAG_DEDUP)) {
		dedupe_context_t *rctx;
		uchar_t *tmp;

		rctx = tdat->rctx;
		reset_dedupe_context(tdat->rctx);
		rctx->cbuf = tdat->compressed_chunk;
		dedupe_decompress(rctx, tdat->uncompressed_chunk, &(tdat->len_cmp));
		if (!rctx->valid) {
			log_msg(LOG_ERR, 0, "ERROR: Chunk %d, dedup recovery failed.", tdat->id);
			rv = -1;
			tdat->len_cmp = 0;
			pctx->t_errored = 1;
			goto cont;
		}
		_chunksize = tdat->len_cmp;
		tmp = tdat->uncompressed_chunk;
		tdat->uncompressed_chunk = tdat->compressed_chunk;
		tdat->compressed_chunk = tmp;
		tdat->cmp_seg = tdat->uncompressed_chunk;
	} else {
		/*
		 * This chunk was not deduplicated, however we still need to down the
		 * semaphore in order to maintain proper thread coordination. We do this after
		 * decompression to achieve better concurrency. Decompression does not need
		 * to wait for the previous thread's dedupe recovery to complete.
		 */
		if (pctx->enable_rabin_global) {
			Sem_Wait(tdat->rctx->index_sem);
		}
	}

	if (!pctx->encrypt_type) {
		/*
		 * Re-compute checksum of original uncompressed chunk.
		 * If it does not match we set length of chunk to 0 to indicate
		 * exit to the writer thread.
		 */
		compute_checksum(checksum, pctx->cksum, tdat->uncompressed_chunk,
		    _chunksize, tdat->cksum_mt, 1);
		if (memcmp(checksum, tdat->checksum, pctx->cksum_bytes) != 0) {
			tdat->len_cmp = 0;
			log_msg(LOG_ERR, 0, "ERROR: Chunk %d, checksums do not match.", tdat->id);
			pctx->t_errored = 1;
			pctx->main_cancel = 1;
		}
	}

cont:
	Sem_Post(&tdat->cmp_done_sem);
	if (!pctx->t_errored)
		goto redo;
	return (NULL);
}

/*
 * File decompression routine.
 *
 * Compressed file Format
 * ----------------------
 * File Header:
 * Algorithm string:  8 bytes.
 * Version number:    2 bytes.
 * Global Flags:      2 bytes.
 * Chunk size:        8 bytes.
 * Compression Level: 4 bytes.
 *
 * Chunk Header:
 * Compressed length: 8 bytes.
 * Checksum:          Upto 64 bytes.
 * Chunk flags:       1 byte.
 *
 * Chunk Flags, 8 bits:
 * I  I  I  I  I  I  I  I
 * |  |     |     |  |  |
 * |  '-----'     |  |  `- 0 - Uncompressed
 * |     |        |  |     1 - Compressed
 * |     |        |  |
 * |     |        |  `---- 1 - Chunk was Deduped
 * |     |        `------- 1 - Chunk was pre-compressed
 * |     |
 * |     |                 1 - Bzip2 (Adaptive Mode)
 * |     `---------------- 2 - Lzma (Adaptive Mode)
 * |                       3 - PPMD (Adaptive Mode)
 * |
 * `---------------------- 1 - Chunk size flag (if original chunk is of variable length)
 *
 * A file trailer to indicate end.
 * Zero Compressed length: 8 zero bytes.
 */
#define UNCOMP_BAIL err = 1; goto uncomp_done

int DLL_EXPORT
start_decompress(pc_ctx_t *pctx, const char *filename, char *to_filename)
{
	char algorithm[ALGO_SZ];
	struct stat sbuf;
	struct wdata w;
	int compfd = -1, compfd2 = -1, p, dedupe_flag;
	int uncompfd = -1, err, np, bail;
	int thread = 0, level;
	uint32_t nprocs = 1, i;
	unsigned short version, flags;
	int64_t chunksize, compressed_chunksize;
	struct cmp_data **dary, *tdat;
	pthread_t writer_thr;
	algo_props_t props;

	err = 0;
	flags = 0;
	thread = 0;
	dary = NULL;
	init_algo_props(&props);

	/*
	 * Open files and do sanity checks.
	 */
	if (!pctx->pipe_mode) {
		if (filename == NULL) {
			pctx->pipe_mode = 1;
			compfd = fileno(stdin);
			if (compfd == -1) {
				log_msg(LOG_ERR, 1, "fileno ");
				UNCOMP_BAIL;
			}
			sbuf.st_size = 0;
		} else {
			if ((compfd = open(filename, O_RDONLY, 0)) == -1) {
				log_msg(LOG_ERR, 1, "Cannot open: %s", filename);
				return (1);
			}

			if (fstat(compfd, &sbuf) == -1) {
				log_msg(LOG_ERR, 1, "Cannot stat: %s", filename);
				return (1);
			}
			if (sbuf.st_size == 0)
				return (1);
		}
	} else {
		compfd = fileno(stdin);
		if (compfd == -1) {
			log_msg(LOG_ERR, 1, "fileno ");
			UNCOMP_BAIL;
		}
	}

	/*
	 * Read file header pieces and verify.
	 */
	if (Read(compfd, algorithm, ALGO_SZ) < ALGO_SZ) {
		log_msg(LOG_ERR, 1, "Read: ");
		UNCOMP_BAIL;
	}
	if (init_algo(pctx, algorithm, 0) != 0) {
		if (pctx->pipe_mode || filename == NULL)
			log_msg(LOG_ERR, 0, "Input stream is not pcompressed.");
		else
			log_msg(LOG_ERR, 0, "%s is not a pcompressed file.", filename);
		UNCOMP_BAIL;
	}
	pctx->algo = algorithm;

	if (Read(compfd, &version, sizeof (version)) < sizeof (version) ||
	    Read(compfd, &flags, sizeof (flags)) < sizeof (flags) ||
	    Read(compfd, &chunksize, sizeof (chunksize)) < sizeof (chunksize) ||
	    Read(compfd, &level, sizeof (level)) < sizeof (level)) {
		log_msg(LOG_ERR, 1, "Read: ");
		UNCOMP_BAIL;
	}

	version = ntohs(version);
	flags = ntohs(flags);
	chunksize = ntohll(chunksize);
	level = ntohl(level);

	/*
	 * Check for ridiculous values (malicious tampering or otherwise).
	 */
	if (version > VERSION) {
		log_msg(LOG_ERR, 0, "Cannot handle newer archive version %d, capability %d",
			version, VERSION);
		err = 1;
		goto uncomp_done;
	}
	if (chunksize > EIGHTY_PCT(get_total_ram())) {
		log_msg(LOG_ERR, 0, "Chunk size must not exceed 80%% of total RAM.");
		err = 1;
		goto uncomp_done;
	}
	if (level > MAX_LEVEL || level < 0) {
		log_msg(LOG_ERR, 0, "Invalid compression level in header: %d", level);
		err = 1;
		goto uncomp_done;
	}
	if (version < VERSION-4) {
		log_msg(LOG_ERR, 0, "Unsupported version: %d", version);
		err = 1;
		goto uncomp_done;
	}

	/*
	 * First check for archive mode. In that case the to_filename must be a directory.
	 */
	if (flags & FLAG_ARCHIVE) {
		if (flags & FLAG_META_STREAM && version > 9)
			pctx->meta_stream = 1;

		/*
		 * Archives with metadata streams cannot be decoded in pipe mode.
		 */
		if (pctx->pipe_mode && pctx->meta_stream) {
			log_msg(LOG_ERR, 0,
			    "Cannot extract archive with metadata stream in pipe mode.");
		}

		/*
		 * If to_filename is not set, we just use the current directory.
		 */
		if (to_filename == NULL) {
			to_filename = ".";
			pctx->to_filename = ".";
		}
		pctx->archive_mode = 1;
		if (stat(to_filename, &sbuf) == -1) {
			if (errno != ENOENT) {
				log_msg(LOG_ERR, 1, "Target path is not a directory.");
				err = 1;
				goto uncomp_done;
			}
			if (mkdir(to_filename,
			    S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH) == -1) {
				log_msg(LOG_ERR, 1, "Unable to create target directory %s.",
				    to_filename);
				err = 1;
				goto uncomp_done;
			}
			if (stat(to_filename, &sbuf) == -1) {
				log_msg(LOG_ERR, 1, "Unable to correctly create target directory %s.",
				    to_filename);
				err = 1;
				goto uncomp_done;
			}
		}
		if (!S_ISDIR(sbuf.st_mode)) {
			log_msg(LOG_ERR, 0, "Target path is not a directory.", to_filename);
			err = 1;
			goto uncomp_done;
		}

		/*
		 * Open another fd to the compressed archive. This is used by the metadata
		 * thread.
		 */
		if (pctx->meta_stream) {
			if ((compfd2 = open(filename, O_RDONLY, 0)) == -1) {
				log_msg(LOG_ERR, 1, "Cannot open: %s", filename);
				err = 1;
				goto uncomp_done;
			}
		}
	} else {
		const char *origf;

		if (pctx->list_mode) {
			log_msg(LOG_ERR, 0, "Nothing to list. The compressed file "
			    "is not an archive.");
			err = 1;
			goto uncomp_done;
		}
		if (to_filename == NULL && !pctx->pipe_mode) {
			char *pos;

			/*
			 * Use unused space in archive_members_file buffer to hold generated
			 * filename so that it need not be explicitly freed at the end.
			 */
			to_filename = pctx->archive_members_file;
			pctx->to_filename = pctx->archive_members_file;
			pos = strrchr(filename, '.');
			if (pos != NULL) {
				if ((pos[0] == 'p' || pos[0] == 'P') &&
				    (pos[1] == 'z' || pos[1] == 'Z')) {
					memcpy(to_filename, filename, pos - filename);
				} else {
					pos = NULL;
				}
			}

			/*
			 * If no .pz extension is found then use <filename>.out as the
			 * decompressed file name.
			 */
			if (pos == NULL) {
				strcpy(to_filename, filename);
				strcat(to_filename, ".out");
				log_msg(LOG_WARN, 0, "Using %s for output file name.", to_filename);
			}
		}
		if (!pctx->pipe_mode) {
			origf = to_filename;
			if ((to_filename = realpath(origf, NULL)) != NULL) {
				free((void *)(to_filename));
				log_msg(LOG_ERR, 0, "File %s exists", origf);
				err = 1;
				goto uncomp_done;
			}
			to_filename = (char *)origf;
		}
	}

	compressed_chunksize = chunksize + CHUNK_HDR_SZ + zlib_buf_extra(chunksize);

	if (pctx->_props_func) {
		pctx->_props_func(&props, level, chunksize);
		if (chunksize + props.buf_extra > compressed_chunksize) {
			compressed_chunksize += (chunksize + props.buf_extra - 
			    compressed_chunksize);
		}
	}

	dedupe_flag = RABIN_DEDUPE_SEGMENTED; // Silence the compiler
	if (flags & FLAG_DEDUP) {
		pctx->enable_rabin_scan = 1;
		dedupe_flag = RABIN_DEDUPE_SEGMENTED;

		if (flags & FLAG_DEDUP_FIXED) {
			if (version > 7) {
				if (pctx->pipe_mode) {
					log_msg(LOG_ERR, 0, "Global Deduplication is not "
					    "supported with pipe mode.");
					err = 1;
					goto uncomp_done;
				}
				pctx->enable_rabin_global = 1;
				dedupe_flag = RABIN_DEDUPE_FILE_GLOBAL;
			} else {
				log_msg(LOG_ERR, 0, "Invalid file deduplication flags.");
				err = 1;
				goto uncomp_done;
			}
		}
	} else if (flags & FLAG_DEDUP_FIXED) {
		pctx->enable_fixed_scan = 1;
		dedupe_flag = RABIN_DEDUPE_FIXED;
	}

	if (flags & FLAG_SINGLE_CHUNK) {
		props.is_single_chunk = 1;
	}

	pctx->cksum = flags & CKSUM_MASK;

	/*
	 * Backward compatibility check for SKEIN in archives version 5 or below.
	 * In newer versions BLAKE uses same IDs as SKEIN.
	 */
	if (version <= 5) {
		if (pctx->cksum == CKSUM_BLAKE256) pctx->cksum = CKSUM_SKEIN256;
		if (pctx->cksum == CKSUM_BLAKE512) pctx->cksum = CKSUM_SKEIN512;
	}
	if (get_checksum_props(NULL, &(pctx->cksum), &(pctx->cksum_bytes),
	    &(pctx->mac_bytes), 1) == -1) {
		log_msg(LOG_ERR, 0, "Invalid checksum algorithm code: %d. "
		    "File corrupt ?", pctx->cksum);
		UNCOMP_BAIL;
	}

	/*
	 * Archives older than version 5 did not support MACs.
	 */
	if (version < 5)
		pctx->mac_bytes = 0;

	/*
	 * If encryption is enabled initialize crypto.
	 */
	if (flags & MASK_CRYPTO_ALG) {
		int saltlen, noncelen;
		uchar_t *salt1, *salt2;
		uchar_t nonce[MAX_NONCE], n1[MAX_NONCE];
		uchar_t pw[MAX_PW_LEN];
		int pw_len;
		mac_ctx_t hdr_mac;
		uchar_t hdr_hash1[pctx->mac_bytes], hdr_hash2[pctx->mac_bytes];
		unsigned int hlen;
		unsigned short d1;
		unsigned int d2;
		uint64_t d3;

		/*
		 * In encrypted files we do not have a normal digest. The HMAC
		 * is computed over header and encrypted data.
		 */
		pctx->cksum_bytes = 0;
		pw_len = -1;
		compressed_chunksize += pctx->mac_bytes;
		pctx->encrypt_type = flags & MASK_CRYPTO_ALG;
		if (version < 7)
			pctx->keylen = OLD_KEYLEN;

		if (pctx->encrypt_type == CRYPTO_ALG_AES) {
			noncelen = 8;
		} else if (pctx->encrypt_type == CRYPTO_ALG_SALSA20) {
			noncelen = XSALSA20_CRYPTO_NONCEBYTES;
		} else {
			log_msg(LOG_ERR, 0, "Invalid Encryption algorithm code: %d. File corrupt ?",
				pctx->encrypt_type);
			UNCOMP_BAIL;
		}
		if (Read(compfd, &saltlen, sizeof (saltlen)) < sizeof (saltlen)) {
			log_msg(LOG_ERR, 1, "Read: ");
			UNCOMP_BAIL;
		}
		saltlen = ntohl(saltlen);
		salt1 = (uchar_t *)malloc(saltlen);
		salt2 = (uchar_t *)malloc(saltlen);
		if (Read(compfd, salt1, saltlen) < saltlen) {
			free(salt1);  free(salt2);
			log_msg(LOG_ERR, 1, "Read: ");
			UNCOMP_BAIL;
		}
		deserialize_checksum(salt2, salt1, saltlen);

		if (Read(compfd, n1, noncelen) < noncelen) {
			memset(salt2, 0, saltlen);
			free(salt2);
			memset(salt1, 0, saltlen);
			free(salt1);
			log_msg(LOG_ERR, 1, "Read: ");
			UNCOMP_BAIL;
		}

		if (pctx->encrypt_type == CRYPTO_ALG_AES) {
			U64_P(nonce) = ntohll(U64_P(n1));

		} else if (pctx->encrypt_type == CRYPTO_ALG_SALSA20) {
			deserialize_checksum(nonce, n1, noncelen);
		}

		if (version > 6) {
			if (Read(compfd, &(pctx->keylen), sizeof (pctx->keylen)) < sizeof (pctx->keylen)) {
				memset(salt2, 0, saltlen);
				free(salt2);
				memset(salt1, 0, saltlen);
				free(salt1);
				log_msg(LOG_ERR, 1, "Read: ");
				UNCOMP_BAIL;
			}
			pctx->keylen = ntohl(pctx->keylen);
		}

		if (Read(compfd, hdr_hash1, pctx->mac_bytes) < pctx->mac_bytes) {
			memset(salt2, 0, saltlen);
			free(salt2);
			memset(salt1, 0, saltlen);
			free(salt1);
			log_msg(LOG_ERR, 1, "Read: ");
			UNCOMP_BAIL;
		}
		deserialize_checksum(hdr_hash2, hdr_hash1, pctx->mac_bytes);

		if (!pctx->pwd_file && !pctx->user_pw) {
			pw_len = get_pw_string(pw,
				"Please enter decryption password", 0);
			if (pw_len == -1) {
				memset(salt2, 0, saltlen);
				free(salt2);
				memset(salt1, 0, saltlen);
				free(salt1);
				log_msg(LOG_ERR, 0, "Failed to get password.");
				UNCOMP_BAIL;
			}
		} else if (!pctx->user_pw) {
			int fd, len;
			uchar_t zero[MAX_PW_LEN];

			/*
			 * Read password from a file and zero out the file after reading.
			 */
			memset(zero, 0, MAX_PW_LEN);
			fd = open(pctx->pwd_file, O_RDWR);
			if (fd != -1) {
				pw_len = (int)lseek(fd, 0, SEEK_END);
				if (pw_len != -1) {
					if (pw_len > MAX_PW_LEN) pw_len = MAX_PW_LEN-1;
					lseek(fd, 0, SEEK_SET);
					len = (int)Read(fd, pw, pw_len);
					if (len != -1 && len == pw_len) {
						pw[pw_len] = '\0';
						if (isspace(pw[pw_len - 1]))
							pw[pw_len-1] = '\0';
						lseek(fd, 0, SEEK_SET);
						Write(fd, zero, pw_len);
						len = ftruncate(fd, 0);
						/*^^^ Make compiler happy. */
					} else {
						pw_len = -1;
					}
				}
			}
			if (pw_len == -1) {
				log_msg(LOG_ERR, 1, " ");
				memset(salt2, 0, saltlen);
				free(salt2);
				memset(salt1, 0, saltlen);
				free(salt1);
				log_msg(LOG_ERR, 0, "Failed to get password.");
				UNCOMP_BAIL;
			}
			close(fd);
		}

		if (pctx->user_pw) {
			if (init_crypto(&(pctx->crypto_ctx), pctx->user_pw, pctx->user_pw_len,
			    pctx->encrypt_type, salt2, saltlen, pctx->keylen, nonce,
			    DECRYPT_FLAG) == -1) {
				memset(salt2, 0, saltlen);
				free(salt2);
				memset(salt1, 0, saltlen);
				free(salt1);
				memset(pctx->user_pw, 0, pctx->user_pw_len);
				log_msg(LOG_ERR, 0, "Failed to initialize crypto");
				UNCOMP_BAIL;
			}
			memset(pctx->user_pw, 0, pctx->user_pw_len);
			pctx->user_pw = NULL;
			pctx->user_pw_len = 0;
		} else {
			if (init_crypto(&(pctx->crypto_ctx), pw, pw_len, pctx->encrypt_type, salt2,
			    saltlen, pctx->keylen, nonce, DECRYPT_FLAG) == -1) {
				memset(salt2, 0, saltlen);
				free(salt2);
				memset(salt1, 0, saltlen);
				free(salt1);
				memset(pw, 0, MAX_PW_LEN);
				log_msg(LOG_ERR, 0, "Failed to initialize crypto");
				UNCOMP_BAIL;
			}
			memset(pw, 0, MAX_PW_LEN);
		}
		memset(salt2, 0, saltlen);
		free(salt2);
		memset(nonce, 0, noncelen);

		/*
		 * Verify file header HMAC.
		 */
		if (hmac_init(&hdr_mac, pctx->cksum, &(pctx->crypto_ctx)) == -1) {
			log_msg(LOG_ERR, 0, "Cannot initialize header hmac.");
			UNCOMP_BAIL;
		}
		hmac_update(&hdr_mac, (uchar_t *)pctx->algo, ALGO_SZ);
		d1 = htons(version);
		hmac_update(&hdr_mac, (uchar_t *)&d1, sizeof (version));
		d1 = htons(flags);
		hmac_update(&hdr_mac, (uchar_t *)&d1, sizeof (flags));
		d3 = htonll(chunksize);
		hmac_update(&hdr_mac, (uchar_t *)&d3, sizeof (chunksize));
		d2 = htonl(level);
		hmac_update(&hdr_mac, (uchar_t *)&d2, sizeof (level));
		if (version > 6) {
			d2 = htonl(saltlen);
			hmac_update(&hdr_mac, (uchar_t *)&d2, sizeof (saltlen));
			hmac_update(&hdr_mac, salt1, saltlen);
			hmac_update(&hdr_mac, n1, noncelen);
			d2 = htonl(pctx->keylen);
			hmac_update(&hdr_mac, (uchar_t *)&d2, sizeof (pctx->keylen));
		}
		hmac_final(&hdr_mac, hdr_hash1, &hlen);
		hmac_cleanup(&hdr_mac);
		memset(salt1, 0, saltlen);
		free(salt1);
		memset(n1, 0, noncelen);
		if (memcmp(hdr_hash2, hdr_hash1, pctx->mac_bytes) != 0) {
			log_msg(LOG_ERR, 0, "Header verification failed! File "
			    "tampered or wrong password.");
			UNCOMP_BAIL;
		}
	} else if (version >= 5) {
		uint32_t crc1, crc2;
		unsigned short d1;
		unsigned int d2;
		uint64_t ch;

		/*
		 * Verify file header CRC32 in non-crypto mode.
		 */
		if (Read(compfd, &crc1, sizeof (crc1)) < sizeof (crc1)) {
			log_msg(LOG_ERR, 1, "Read: ");
			UNCOMP_BAIL;
		}
		crc1 = htonl(crc1);
		pctx->mac_bytes = sizeof (uint32_t);

		crc2 = lzma_crc32((uchar_t *)pctx->algo, ALGO_SZ, 0);
		d1 = htons(version);
		crc2 = lzma_crc32((uchar_t *)&d1, sizeof (version), crc2);
		d1 = htons(flags);
		crc2 = lzma_crc32((uchar_t *)&d1, sizeof (version), crc2);
		ch = htonll(chunksize);
		crc2 = lzma_crc32((uchar_t *)&ch, sizeof (ch), crc2);
		d2 = htonl(level);
		crc2 = lzma_crc32((uchar_t *)&d2, sizeof (level), crc2);
		if (crc1 != crc2) {
			log_msg(LOG_ERR, 0, "Header verification failed! File tampered "
			    "or wrong password.");
			UNCOMP_BAIL;
		}
	}

	if (flags & FLAG_ARCHIVE) {
		if (pctx->enable_rabin_global) {
			char cwd[MAXPATHLEN];

			if (to_filename[0] != PATHSEP_CHAR) {
				if (getcwd(cwd, MAXPATHLEN) == NULL) {
					log_msg(LOG_ERR, 1, "Cannot get current dir");
					UNCOMP_BAIL;
				}

				snprintf(pctx->archive_temp_file, sizeof (pctx->archive_temp_file),
				    "%s" PATHSEP_STR "%s" PATHSEP_STR ".data", cwd, to_filename);
			} else {
				snprintf(pctx->archive_temp_file, sizeof (pctx->archive_temp_file),
					 "%s" PATHSEP_STR ".data", to_filename);
			}
			if ((pctx->archive_temp_fd = open(pctx->archive_temp_file,
			    O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR)) == -1) {
				log_msg(LOG_ERR, 1, "Cannot open temporary data file in "
				    "target directory.");
				UNCOMP_BAIL;
			}
			add_fname(pctx->archive_temp_file);
		}

		/*
		 * If we are having a metadata stream, get the current position of the main
		 * fd. The secondary fd must be set to the same position so that metadata
		 * thread can start scanning for chunks after the header and any info chunks.
		 *
		 * NOTE: This is done here to allow setup_extractor() call later to work.
		 */
		if (pctx->meta_stream) {
			off_t cpos = lseek(compfd, 0, SEEK_CUR);
			cpos = lseek(compfd2, cpos, SEEK_SET);
			if (cpos == -1) {
				log_msg(LOG_ERR, 1, "Can't seek in metadata fd: ");
				UNCOMP_BAIL;
			}

			/*
			 * Finally create the metadata context.
			 */
			pctx->meta_ctx = meta_ctx_create(pctx, VERSION, compfd2);
			if (pctx->meta_ctx == NULL) {
				close(compfd2);
				UNCOMP_BAIL;
			}
		}

		uncompfd = -1;
		if (setup_extractor(pctx) == -1) {
			log_msg(LOG_ERR, 0, "Setup of extraction context failed.");
			UNCOMP_BAIL;
		}

		if (start_extractor(pctx) == -1) {
			log_msg(LOG_ERR, 0, "Unable to start extraction thread.");
			UNCOMP_BAIL;
		}
	} else {
		if (!pctx->pipe_mode) {
			if ((uncompfd = open(to_filename, O_WRONLY|O_CREAT|O_TRUNC,
			    S_IRUSR|S_IWUSR)) == -1) {
				log_msg(LOG_ERR, 1, "Cannot open: %s", to_filename);
				UNCOMP_BAIL;
			}
		} else {
			uncompfd = fileno(stdout);
			if (uncompfd == -1) {
				log_msg(LOG_ERR, 1, "fileno ");
				UNCOMP_BAIL;
			}
		}
	}

	/*
	 * WARNING: NO Further file header/info chunk processing beyond this point.
	 *          Doing so will BREAK Separate Metadata stream processing.
	 */

	nprocs = (uint32_t)sysconf(_SC_NPROCESSORS_ONLN);
	if (pctx->archive_mode) {
		nprocs = nprocs > 1 ? nprocs-1:nprocs;
	}

	if (pctx->nthreads > 0 && pctx->nthreads < nprocs)
		nprocs = pctx->nthreads;
	else
		pctx->nthreads = nprocs;

	set_threadcounts(&props, &(pctx->nthreads), nprocs, DECOMPRESS_THREADS);
	if (props.is_single_chunk)
		pctx->nthreads = 1;
	/*
	 * If we are trying to list the archive contents, and the archive has a
	 * metadata stream, then we do not do any data decompression. Only
	 * metadata is decompressed.
	 */
	if (pctx->list_mode && pctx->meta_stream) {
		pctx->nthreads = 0;
		pctx->temp_mmap_buf = (uchar_t *)slab_alloc(NULL, chunksize);
		pctx->temp_mmap_len = chunksize;
	}

	if (pctx->nthreads * props.nthreads > 1)
		log_msg(LOG_INFO, 0, "Scaling to %d threads", pctx->nthreads * props.nthreads);
	else
		log_msg(LOG_INFO, 0, "Scaling to 1 thread");
	nprocs = pctx->nthreads;
	slab_cache_add(compressed_chunksize);
	slab_cache_add(chunksize);
	slab_cache_add(sizeof (struct cmp_data));

	dary = (struct cmp_data **)slab_calloc(NULL, nprocs, sizeof (struct cmp_data *));
	for (i = 0; i < nprocs; i++) {
		dary[i] = (struct cmp_data *)slab_alloc(NULL, sizeof (struct cmp_data));
		if (!dary[i]) {
			log_msg(LOG_ERR, 0, "1: Out of memory");
			UNCOMP_BAIL;
		}
		tdat = dary[i];
		tdat->pctx = pctx;
		tdat->compressed_chunk = NULL;
		tdat->uncompressed_chunk = NULL;
		tdat->chunksize = chunksize;
		tdat->compress = pctx->_compress_func;
		tdat->decompress = pctx->_decompress_func;
		tdat->cancel = 0;
		tdat->decompressing = 1;
		if (props.is_single_chunk) {
			tdat->cksum_mt = 1;
			if (version == 6) {
				tdat->cksum_mt = 2; // Indicate old format parallel hash
			}
		} else {
			tdat->cksum_mt = 0;
		}
		tdat->level = level;
		tdat->data = NULL;
		tdat->props = &props;
		Sem_Init(&(tdat->start_sem), 0, 0);
		Sem_Init(&(tdat->cmp_done_sem), 0, 0);
		Sem_Init(&(tdat->write_done_sem), 0, 1);
		Sem_Init(&(tdat->index_sem), 0, 0);

		if (pctx->_init_func) {
			if (pctx->_init_func(&(tdat->data), &(tdat->level), props.nthreads, chunksize,
			    version, DECOMPRESS) != 0) {
				UNCOMP_BAIL;
			}
		}

		/*
		 * The last parameter is freeram. It is not needed during decompression.
		 */
		if (pctx->enable_rabin_scan || pctx->enable_fixed_scan || pctx->enable_rabin_global) {
			tdat->rctx = create_dedupe_context(chunksize, compressed_chunksize,
			    pctx->rab_blk_size, pctx->algo, &props, pctx->enable_delta_encode,
			    dedupe_flag, version, DECOMPRESS, 0, NULL, pctx->pipe_mode, nprocs, 0);
			if (tdat->rctx == NULL) {
				UNCOMP_BAIL;
			}
			if (pctx->enable_rabin_global) {
				if (pctx->archive_mode) {
					if ((tdat->rctx->out_fd = open(pctx->archive_temp_file,
					    O_RDONLY, 0)) == -1) {
						log_msg(LOG_ERR, 1, "Unable to get new read handle"
						    " to output file");
						UNCOMP_BAIL;
					}
				} else {
					if ((tdat->rctx->out_fd = open(to_filename, O_RDONLY, 0))
					    == -1) {
						log_msg(LOG_ERR, 1, "Unable to get new read handle"
						    " to output file");
						UNCOMP_BAIL;
					}
				}
			}
			tdat->rctx->index_sem = &(tdat->index_sem);
		} else {
			tdat->rctx = NULL;
		}

		if (pctx->encrypt_type) {
			if (hmac_init(&tdat->chunk_hmac, pctx->cksum, &(pctx->crypto_ctx)) == -1) {
				log_msg(LOG_ERR, 0, "Cannot initialize chunk hmac.");
				UNCOMP_BAIL;
			}
		}
		if (pthread_create(&(tdat->thr), NULL, perform_decompress,
		    (void *)tdat) != 0) {
			log_msg(LOG_ERR, 1, "Error in thread creation: ");
			UNCOMP_BAIL;
		}
	}
	thread = 1;

	if (pctx->enable_rabin_global) {
		for (i = 0; i < nprocs; i++) {
			tdat = dary[i];
			tdat->rctx->index_sem_next = &(dary[(i + 1) % nprocs]->index_sem);
		}
	}
	// When doing global dedupe first thread does not wait to start dedupe recovery.
	if (nprocs > 0)
		Sem_Post(&(dary[0]->index_sem));

	if (pctx->encrypt_type) {
		/* Erase encryption key bytes stored as a plain array. No longer reqd. */
		crypto_clean_pkey(&(pctx->crypto_ctx));
	}

	if (!(pctx->list_mode && pctx->meta_stream)) {
		w.dary = dary;
		w.wfd = uncompfd;
		w.nprocs = nprocs;
		w.chunksize = chunksize;
		w.pctx = pctx;
		if (pthread_create(&writer_thr, NULL, writer_thread, (void *)(&w)) != 0) {
			log_msg(LOG_ERR, 1, "Error in thread creation: ");
			UNCOMP_BAIL;
		}
		thread = 2;
	}

	/*
	 * Now read from the compressed file in variable compressed chunk size.
	 * First the size is read from the chunk header and then as many bytes +
	 * checksum size are read and passed to decompression thread.
	 * Chunk sequencing is ensured.
	 */
	pctx->chunk_num = 0;
	np = 0;
	bail = 0;
	if (nprocs == 0)
		bail = 1;
	while (!bail) {
		int64_t rb;

		if (pctx->main_cancel) break;
		for (p = 0; p < nprocs; p++) {
			np = p;
			tdat = dary[p];
			Sem_Wait(&tdat->write_done_sem);
			if (pctx->main_cancel) break;
			tdat->id = pctx->chunk_num;
			if (tdat->rctx) tdat->rctx->id = tdat->id;

redo:
			/*
			 * First read length of compressed chunk.
			 */
			rb = Read(compfd, &tdat->len_cmp, sizeof (tdat->len_cmp));
			if (rb != sizeof (tdat->len_cmp)) {
				if (rb < 0) log_msg(LOG_ERR, 1, "Read: ");
				else
					log_msg(LOG_ERR, 0, "Incomplete chunk %d header,"
					    "file corrupt", pctx->chunk_num);
				UNCOMP_BAIL;
			}
			tdat->len_cmp_be = tdat->len_cmp; // Needed for HMAC
			tdat->len_cmp = htonll(tdat->len_cmp);

			/*
			 * Check for ridiculous length.
			 */
			if (tdat->len_cmp > chunksize + 256) {
				log_msg(LOG_ERR, 0, "Compressed length too big for chunk: %d",
				    pctx->chunk_num);
				UNCOMP_BAIL;
			}

			/*
			 * Zero compressed len means end of file.
			 */
			if (tdat->len_cmp == 0) {
				bail = 1;
				break;

			} else if (tdat->len_cmp == METADATA_INDICATOR) {
				if (!pctx->meta_stream) {
					log_msg(LOG_ERR, 0, "Invalid chunk %d length: %" PRIu64 "\n",
						pctx->chunk_num, tdat->len_cmp);
					UNCOMP_BAIL;
				}
				/*
				 * If compressed length indicates a metadata chunk. Read it's length
				 * and skip the chunk.
				 */
				rb = Read(compfd, &tdat->len_cmp_be, sizeof (tdat->len_cmp_be));
				if (rb != sizeof (tdat->len_cmp_be)) {
					if (rb < 0) log_msg(LOG_ERR, 1, "Read: ");
					else
						log_msg(LOG_ERR, 0, "Incomplete chunk %d header,"
							"file corrupt", pctx->chunk_num);
					UNCOMP_BAIL;
				}

				/*
				 * We will be reading and skipping this chunk next.
				 */
				tdat->len_cmp_be = LE64(tdat->len_cmp_be);
			}

			/*
			 * Delayed allocation. Allocate chunks if not already done. The compressed
			 * file format does not provide any info on how many chunks are there in
			 * order to allow pipe mode operation. So delayed allocation during
			 * decompression allows to avoid allocating per-thread chunks which will
			 * never be used. This can happen if chunk count < thread count.
			 */
			if (!tdat->compressed_chunk && tdat->len_cmp != METADATA_INDICATOR) {
				tdat->compressed_chunk = (uchar_t *)slab_alloc(NULL,
				    compressed_chunksize);
				tdat->uncompressed_chunk = (uchar_t *)slab_alloc(NULL,
				    compressed_chunksize);
				if (!tdat->compressed_chunk || !tdat->uncompressed_chunk) {
					log_msg(LOG_ERR, 0, "2: Out of memory");
					UNCOMP_BAIL;
				}
				tdat->cmp_seg = tdat->uncompressed_chunk;
			}

			if (tdat->len_cmp != METADATA_INDICATOR) {
				if (tdat->len_cmp > pctx->largest_chunk)
					pctx->largest_chunk = tdat->len_cmp;
				if (tdat->len_cmp < pctx->smallest_chunk)
					pctx->smallest_chunk = tdat->len_cmp;
				pctx->avg_chunk += tdat->len_cmp;

				/*
				 * Now read compressed chunk including the checksum. This is
				 * a seek if it is just a listing operation. No data is
				 * processed in that case.
				 */
				rb = tdat->len_cmp + pctx->cksum_bytes + pctx->mac_bytes +
				    CHUNK_FLAG_SZ;
				tdat->rbytes = Read(compfd, tdat->compressed_chunk, rb);
			} else {
				off_t cpos = lseek(compfd, 0, SEEK_CUR);

				 /* Two values already read */
				rb = tdat->len_cmp_be + METADATA_HDR_SZ - 16;
				tdat->rbytes = lseek(compfd, rb, SEEK_CUR);
				if (tdat->rbytes > 0)
					tdat->rbytes = tdat->rbytes - cpos;
			}
			if (pctx->main_cancel) break;
			if (tdat->rbytes < rb) {
				if (tdat->rbytes < 0) {
					log_msg(LOG_ERR, 1, "Read: ");
					UNCOMP_BAIL;
				} else {
					log_msg(LOG_ERR, 0, "Incomplete chunk %d, file corrupt.",
					    pctx->chunk_num);
					UNCOMP_BAIL;
				}
			}

			/*
			 * If we just skipped a metadata chunk, redo the read to go to the next one.
			 */
			if (tdat->len_cmp == METADATA_INDICATOR) {
				goto redo;
			}
			Sem_Post(&tdat->start_sem);
			++(pctx->chunk_num);
		}
	}

	if (!pctx->main_cancel) {
		for (p = 0; p < nprocs; p++) {
			if (p == np) continue;
			tdat = dary[p];
			Sem_Wait(&tdat->write_done_sem);
// VS begin
			if (pctx->main_cancel) break;
// VS end
		}
	}
uncomp_done:
	if (pctx->t_errored) err = pctx->t_errored;
	if (thread) {
		for (i = 0; i < nprocs; i++) {
			tdat = dary[i];
			tdat->cancel = 1;
			tdat->len_cmp = 0;
			Sem_Post(&tdat->start_sem);
			Sem_Post(&tdat->cmp_done_sem);
			pthread_join(tdat->thr, NULL);
		}
		if (thread == 2)
			pthread_join(writer_thr, NULL);
	}

	/*
	 * Ownership and mode of target should be same as original.
	 */
	if (filename != NULL && uncompfd != -1) {
		fchmod(uncompfd, sbuf.st_mode);
		if (fchown(uncompfd, sbuf.st_uid, sbuf.st_gid) == -1)
			log_msg(LOG_ERR, 1, "Chown ");
	}
	if (dary != NULL) {
		for (i = 0; i < nprocs; i++) {
			if (!dary[i]) continue;
			if (dary[i]->uncompressed_chunk)
				slab_release(NULL, dary[i]->uncompressed_chunk);
			if (dary[i]->compressed_chunk)
				slab_release(NULL, dary[i]->compressed_chunk);
			if (pctx->_deinit_func)
				pctx->_deinit_func(&(dary[i]->data));
			if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan)) {
				destroy_dedupe_context(dary[i]->rctx);
			}
			Sem_Destroy(&(dary[i]->start_sem));
			Sem_Destroy(&(dary[i]->cmp_done_sem));
			Sem_Destroy(&(dary[i]->write_done_sem));
			Sem_Destroy(&(dary[i]->index_sem));

			slab_release(NULL, dary[i]);
		}
		slab_release(NULL, dary);
	}
	if (!pctx->pipe_mode) {
		if (filename && compfd != -1) close(compfd);
		if (uncompfd != -1) close(uncompfd);
	}
	if (pctx->archive_mode) {
		pthread_join(pctx->archive_thread, NULL);
		if (pctx->meta_stream) {
			meta_ctx_done(pctx->meta_ctx);
			if (pctx->list_mode) {
				slab_release(NULL, pctx->temp_mmap_buf);
			}
		}
		if (pctx->enable_rabin_global) {
			close(pctx->archive_temp_fd);
			unlink(pctx->archive_temp_file);
		}
		Sem_Destroy(&(pctx->read_sem));
		Sem_Destroy(&(pctx->write_sem));
	}

	if (!pctx->hide_cmp_stats) show_compression_stats(pctx);

	return (err);
}

static void *
perform_compress(void *dat) {
	struct cmp_data *tdat = (struct cmp_data *)dat;
	typeof (tdat->chunksize) _chunksize, len_cmp, dedupe_index_sz, index_size_cmp;
	int type, rv;
	uchar_t *compressed_chunk;
	int64_t rbytes;
	pc_ctx_t *pctx;

	pctx = tdat->pctx;
redo:
	Sem_Wait(&tdat->start_sem);
	if (unlikely(tdat->cancel)) {
		tdat->len_cmp = 0;
		Sem_Post(&tdat->cmp_done_sem);
		return (0);
	}

	compressed_chunk = tdat->compressed_chunk + CHUNK_FLAG_SZ;
	rbytes = tdat->rbytes;
	dedupe_index_sz = 0;
	type = COMPRESSED;

	/* Perform Dedup if enabled. */
	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan)) {
		dedupe_context_t *rctx;
		uint64_t rb = tdat->rbytes;

		/*
		 * Compute checksum of original uncompressed chunk. When doing dedup
		 * cmp_seg hold original data instead of uncompressed_chunk. We dedup
		 * into uncompressed_chunk so that compress transforms uncompressed_chunk
		 * back into cmp_seg. Avoids an extra memcpy().
		 */
		if (!pctx->encrypt_type)
			compute_checksum(tdat->checksum, pctx->cksum, tdat->cmp_seg, tdat->rbytes,
					 tdat->cksum_mt, 1);

		rctx = tdat->rctx;
		reset_dedupe_context(tdat->rctx);
		rctx->cbuf = tdat->uncompressed_chunk;
		dedupe_index_sz = dedupe_compress(tdat->rctx, tdat->cmp_seg, &rb, 0,
						  NULL, tdat->cksum_mt);
		tdat->rbytes = rb;
		if (!rctx->valid) {
			memcpy(tdat->uncompressed_chunk, tdat->cmp_seg, rbytes);
			tdat->rbytes = rbytes;
		}
	} else {
		/*
		 * Compute checksum of original uncompressed chunk.
		 */
		if (!pctx->encrypt_type)
			compute_checksum(tdat->checksum, pctx->cksum, tdat->uncompressed_chunk,
					 tdat->rbytes, tdat->cksum_mt, 1);
	}

	/*
	 * If doing dedup we compress rabin index and deduped data separately.
	 * The rabin index array values can pollute the compressor's dictionary thereby
	 * reducing compression effectiveness of the data chunk. So we separate them.
	 */
	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan) && tdat->rctx->valid) {
		uint64_t o_chunksize;
		_chunksize = tdat->rbytes - dedupe_index_sz - RABIN_HDR_SIZE;
		index_size_cmp = dedupe_index_sz;
		rv = 0;

		/*
		 * Do a matrix transpose of the index table with the hope of improving
		 * compression ratio subsequently.
		 */
		transpose(tdat->uncompressed_chunk + RABIN_HDR_SIZE,
		    compressed_chunk + RABIN_HDR_SIZE, dedupe_index_sz,
		    sizeof (uint32_t), ROW);
		memcpy(tdat->uncompressed_chunk + RABIN_HDR_SIZE,
		    compressed_chunk + RABIN_HDR_SIZE, dedupe_index_sz);

		if (dedupe_index_sz >= 90) {
			/* Compress index if it is at least 90 bytes. */
			rv = lzma_compress(tdat->uncompressed_chunk + RABIN_HDR_SIZE,
			    dedupe_index_sz, compressed_chunk + RABIN_HDR_SIZE,
			    &index_size_cmp, tdat->rctx->level, 255, TYPE_BINARY,
			    tdat->rctx->lzma_data);

			/* 
			 * If index compression fails or does not produce a smaller result
			 * retain it as is. In that case compressed size == original size
			 * and it will be handled correctly during decompression.
			 */
			if (rv != 0 || index_size_cmp >= dedupe_index_sz) {
				index_size_cmp = dedupe_index_sz;
				goto plain_index;
			}
		} else {
plain_index:
			memcpy(compressed_chunk + RABIN_HDR_SIZE,
			    tdat->uncompressed_chunk + RABIN_HDR_SIZE, dedupe_index_sz);
		}

		index_size_cmp += RABIN_HDR_SIZE;
		dedupe_index_sz += RABIN_HDR_SIZE;
		memcpy(compressed_chunk, tdat->uncompressed_chunk, RABIN_HDR_SIZE);
		o_chunksize = _chunksize;

		/* Compress data chunk. */
		if (_chunksize == 0) {
			rv = -1;
		} else if (pctx->preprocess_mode) {
			rv = preproc_compress(pctx, tdat->compress,
			    tdat->uncompressed_chunk + dedupe_index_sz, _chunksize,
			    compressed_chunk + index_size_cmp, &_chunksize, tdat->level, 0,
			    tdat->btype, tdat->data, tdat->props, tdat->interesting);
		} else {
			DEBUG_STAT_EN(double strt, en);

			DEBUG_STAT_EN(strt = get_wtime_millis());
			rv = tdat->compress(tdat->uncompressed_chunk + dedupe_index_sz,
			    _chunksize, compressed_chunk + index_size_cmp, &_chunksize,
			    tdat->level, 0, tdat->btype, tdat->data);
			DEBUG_STAT_EN(en = get_wtime_millis());
			DEBUG_STAT_EN(fprintf(stderr, "Chunk compression speed %.3f MB/s\n",
					      get_mb_s(_chunksize, strt, en)));
		}

		/* Can't compress data just retain as-is. */
		if (rv < 0 || _chunksize >= o_chunksize) {
			_chunksize = o_chunksize;
			type = UNCOMPRESSED;
			memcpy(compressed_chunk + index_size_cmp,
			    tdat->uncompressed_chunk + dedupe_index_sz, _chunksize);
		}
		/* Now update rabin header with the compressed sizes. */
		update_dedupe_hdr(compressed_chunk, index_size_cmp - RABIN_HDR_SIZE, _chunksize);
		_chunksize += index_size_cmp;
	} else {
		_chunksize = tdat->rbytes;
		if (pctx->preprocess_mode) {
			rv = preproc_compress(pctx, tdat->compress, tdat->uncompressed_chunk,
			    tdat->rbytes, compressed_chunk, &_chunksize, tdat->level, 0,
			    tdat->btype, tdat->data, tdat->props, tdat->interesting);
		} else {
			DEBUG_STAT_EN(double strt, en);

			DEBUG_STAT_EN(strt = get_wtime_millis());
			rv = tdat->compress(tdat->uncompressed_chunk, tdat->rbytes,
			    compressed_chunk, &_chunksize, tdat->level, 0, tdat->btype,
			    tdat->data);
			DEBUG_STAT_EN(en = get_wtime_millis());
			DEBUG_STAT_EN(fprintf(stderr, "Chunk compression speed %.3f MB/s\n",
					      get_mb_s(_chunksize, strt, en)));
		}
	}

	/*
	 * Sanity check to ensure compressed data is lesser than original.
	 * If at all compression expands/does not shrink data then the chunk
	 * will be left uncompressed. Also if the compression errored the
	 * chunk will be left uncompressed.
	 *
	 * HOWEVER, increased chunk size is allowed in preprocessing mode since
	 * there are unavoidable cases where an E8E9 filter is applied and then
	 * later on compression does not happen. So we have to retain information
	 * that E8E9 hapened, to recover the data correctly. In this corner case
	 * the chunk size is increased by 1 byte for the preproc header.
	 */
	tdat->len_cmp = _chunksize;
	if ((_chunksize >= tdat->rbytes && !pctx->preprocess_mode) || rv < 0) {
		if (!(pctx->enable_rabin_scan || pctx->enable_fixed_scan) || !tdat->rctx->valid)
			memcpy(compressed_chunk, tdat->uncompressed_chunk, tdat->rbytes);
		type = UNCOMPRESSED;
		tdat->len_cmp = tdat->rbytes;
		if (rv < 0) rv = COMPRESS_NONE;
	}

	/*
	 * Now perform encryption on the compressed data, if requested.
	 */
	if (pctx->encrypt_type) {
		int ret;
		DEBUG_STAT_EN(double strt, en);

		/*
		 * Encryption algorithm must not change the size and
		 * encryption is in-place.
		 */
		DEBUG_STAT_EN(strt = get_wtime_millis());
		ret = crypto_buf(&(pctx->crypto_ctx), compressed_chunk, compressed_chunk,
			tdat->len_cmp, tdat->id);
		if (ret == -1) {
			/*
			 * Encryption failure is fatal.
			 */
			pctx->main_cancel = 1;
			tdat->len_cmp = 0;
			pctx->t_errored = 1;
			Sem_Post(&tdat->cmp_done_sem);
			return (0);
		}
		DEBUG_STAT_EN(en = get_wtime_millis());
		DEBUG_STAT_EN(fprintf(stderr, "Encryption speed %.3f MB/s\n",
			      get_mb_s(tdat->len_cmp, strt, en)));
	}

	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan) && tdat->rctx->valid) {
		type |= CHUNK_FLAG_DEDUP;
	}
	if (pctx->preprocess_mode) {
		type |= CHUNK_FLAG_PREPROC;
	}

	/*
	 * Insert compressed chunk length and checksum into chunk header.
	 */
	len_cmp = tdat->len_cmp;
	*((typeof (len_cmp) *)(tdat->cmp_seg)) = htonll(tdat->len_cmp);
	if (!pctx->encrypt_type)
		serialize_checksum(tdat->checksum, tdat->cmp_seg + sizeof (tdat->len_cmp),
		    pctx->cksum_bytes);
	tdat->len_cmp += CHUNK_FLAG_SZ;
	tdat->len_cmp += sizeof (len_cmp);
	tdat->len_cmp += (pctx->cksum_bytes + pctx->mac_bytes);
	rbytes = tdat->len_cmp - len_cmp; // HDR size for HMAC

	/*
	 * In adaptive mode return value from compression function function indicates
	 * which algorithm was used on the chunk. We have to store that.
	 */
	if (pctx->adapt_mode)
		type |= (rv << 4);

	/*
	 * If chunk is less than max chunksize, store this length as well.
	 */
	if (tdat->rbytes < tdat->chunksize) {
		type |= CHSIZE_MASK;
		*((typeof (tdat->rbytes) *)(tdat->cmp_seg + tdat->len_cmp)) = htonll(tdat->rbytes);
		tdat->len_cmp += ORIGINAL_CHUNKSZ;
		len_cmp += ORIGINAL_CHUNKSZ;
		*((typeof (len_cmp) *)(tdat->cmp_seg)) = htonll(len_cmp);
	}
	/*
	 * Set the chunk header flags.
	 */
	*(tdat->compressed_chunk) = type;

	/*
	 * If encrypting, compute HMAC for full chunk including header.
	 */
	if (pctx->encrypt_type) {
		uchar_t *mac_ptr;
		unsigned int hlen;
		uchar_t chash[pctx->mac_bytes];
		DEBUG_STAT_EN(double strt, en);

		/* Clean out mac_bytes to 0 for stable HMAC. */
		DEBUG_STAT_EN(strt = get_wtime_millis());
		mac_ptr = tdat->cmp_seg + sizeof (tdat->len_cmp) + pctx->cksum_bytes;
		memset(mac_ptr, 0, pctx->mac_bytes);
		hmac_reinit(&tdat->chunk_hmac);
		hmac_update(&tdat->chunk_hmac, tdat->cmp_seg, tdat->len_cmp);
		hmac_final(&tdat->chunk_hmac, chash, &hlen);
		serialize_checksum(chash, mac_ptr, hlen);
		DEBUG_STAT_EN(en = get_wtime_millis());
		DEBUG_STAT_EN(fprintf(stderr, "HMAC Computation speed %.3f MB/s\n",
			      get_mb_s(tdat->len_cmp, strt, en)));
	} else {
		/*
		 * Compute header CRC32 in non-crypto mode.
		 */
		uchar_t *mac_ptr;
		uint32_t crc;

		/* Clean out mac_bytes to 0 for stable CRC32. */
		mac_ptr = tdat->cmp_seg + sizeof (tdat->len_cmp) + pctx->cksum_bytes;
		memset(mac_ptr, 0, pctx->mac_bytes);
		crc = lzma_crc32(tdat->cmp_seg, rbytes, 0);
		if (type & CHSIZE_MASK)
			crc = lzma_crc32(tdat->cmp_seg + tdat->len_cmp - ORIGINAL_CHUNKSZ,
			    ORIGINAL_CHUNKSZ, crc);
		U32_P(mac_ptr) = htonl(crc);
	}

	Sem_Post(&tdat->cmp_done_sem);
	goto redo;
}

static void *
writer_thread(void *dat) {
	int p;
	struct wdata *w = (struct wdata *)dat;
	struct cmp_data *tdat;
	int64_t wbytes;
	pc_ctx_t *pctx;

	pctx = w->pctx;
repeat:
	for (p = 0; p < w->nprocs; p++) {
		tdat = w->dary[p];
		Sem_Wait(&tdat->cmp_done_sem);
		if (tdat->len_cmp == 0) {
			goto do_cancel;
		}

		if (pctx->do_compress) {
			if (tdat->len_cmp > pctx->largest_chunk)
				pctx->largest_chunk = tdat->len_cmp;
			if (tdat->len_cmp < pctx->smallest_chunk)
				pctx->smallest_chunk = tdat->len_cmp;
			pctx->avg_chunk += tdat->len_cmp;
		}

		if (pctx->archive_mode && tdat->decompressing) {
			wbytes = archiver_write(pctx, tdat->cmp_seg, tdat->len_cmp);
		} else {
			pthread_mutex_lock(&pctx->write_mutex);
			wbytes = Write(w->wfd, tdat->cmp_seg, tdat->len_cmp);
			pthread_mutex_unlock(&pctx->write_mutex);
		}
		if (pctx->archive_temp_fd != -1 && wbytes == tdat->len_cmp) {
			wbytes = Write(pctx->archive_temp_fd, tdat->cmp_seg, tdat->len_cmp);
		}
		if (unlikely(wbytes != tdat->len_cmp)) {
			log_msg(LOG_ERR, 1, "Chunk Write (expected: %" PRIu64
			    ", written: %" PRId64 ") : ", tdat->len_cmp, wbytes);
do_cancel:
			pctx->main_cancel = 1;
			tdat->cancel = 1;
			Sem_Post(&tdat->start_sem);
			if (tdat->rctx && pctx->enable_rabin_global)
				Sem_Post(tdat->rctx->index_sem_next);
			Sem_Post(&tdat->write_done_sem);
			return (0);
		}
		if (tdat->decompressing && tdat->rctx && pctx->enable_rabin_global) {
			Sem_Post(tdat->rctx->index_sem_next);
		}
		Sem_Post(&tdat->write_done_sem);
	}
	goto repeat;
}

/*
 * File compression routine. Can use as many threads as there are
 * logical cores unless user specified something different. There is
 * not much to gain from nthreads > n logical cores however.
 */
#define COMP_BAIL err = 1; goto comp_done

int DLL_EXPORT
start_compress(pc_ctx_t *pctx, const char *filename, uint64_t chunksize, int level)
{
	struct wdata w;
	char tmpfile1[MAXPATHLEN], tmpdir[MAXPATHLEN];
	char to_filename[MAXPATHLEN];
	uint64_t compressed_chunksize, n_chunksize, file_offset;
	int64_t rbytes, rabin_count;
	unsigned short version, flags;
	struct stat sbuf;
	int compfd = -1, uncompfd = -1, err;
	int thread, bail, single_chunk;
	uint32_t i, nprocs, np, p, dedupe_flag;
	struct cmp_data **dary = NULL, *tdat;
	pthread_t writer_thr;
	uchar_t *cread_buf, *pos;
	dedupe_context_t *rctx;
	algo_props_t props;
	my_sysinfo msys_info;

	init_algo_props(&props);
	props.cksum = pctx->cksum;
	props.buf_extra = 0;
	cread_buf = NULL;
	pctx->btype = TYPE_UNKNOWN;
	flags = 0;
	sbuf.st_size = 0;
	err = 0;
	thread = 0;
	dedupe_flag = RABIN_DEDUPE_SEGMENTED; // Silence the compiler
	compressed_chunksize = 0;

	if (pctx->encrypt_type) {
		uchar_t pw[MAX_PW_LEN];
		int pw_len = -1;

		compressed_chunksize += pctx->mac_bytes;
		if (!pctx->pwd_file && !pctx->user_pw) {
			pw_len = get_pw_string(pw,
				"Please enter encryption password", 1);
			if (pw_len == -1) {
				log_msg(LOG_ERR, 0, "Failed to get password.");
				return (1);
			}
		} else if (!pctx->user_pw) {
			int fd, len;
			uchar_t zero[MAX_PW_LEN];

			/*
			 * Read password from a file and zero out the file after reading.
			 */
			memset(zero, 0, MAX_PW_LEN);
			fd = open(pctx->pwd_file, O_RDWR);
			if (fd != -1) {
				pw_len = (int)lseek(fd, 0, SEEK_END);
				if (pw_len != -1) {
					if (pw_len > MAX_PW_LEN) pw_len = MAX_PW_LEN-1;
					lseek(fd, 0, SEEK_SET);
					len = (int)Read(fd, pw, pw_len);
					if (len != -1 && len == pw_len) {
						pw[pw_len] = '\0';
						if (isspace(pw[pw_len - 1]))
							pw[pw_len-1] = '\0';
						lseek(fd, 0, SEEK_SET);
						Write(fd, zero, pw_len);
					} else {
						pw_len = -1;
					}
				}
			}
			if (pw_len == -1) {
				log_msg(LOG_ERR, 1, "Failed to get password.");
				return (1);
			}
			close(fd);
		}
		if (pctx->user_pw) {
			if (init_crypto(&(pctx->crypto_ctx), pctx->user_pw, pctx->user_pw_len,
			    pctx->encrypt_type, NULL, 0, pctx->keylen, 0, ENCRYPT_FLAG) == -1) {
				memset(pctx->user_pw, 0, pctx->user_pw_len);
				log_msg(LOG_ERR, 0, "Failed to initialize crypto.");
				return (1);
			}
			memset(pctx->user_pw, 0, pctx->user_pw_len);
			pctx->user_pw = NULL;
			pctx->user_pw_len = 0;
		} else {
			if (init_crypto(&(pctx->crypto_ctx), pw, pw_len, pctx->encrypt_type, NULL,
			    0, pctx->keylen, 0, ENCRYPT_FLAG) == -1) {
				memset(pw, 0, MAX_PW_LEN);
				log_msg(LOG_ERR, 0, "Failed to initialize crypto.");
				return (1);
			}
			memset(pw, 0, MAX_PW_LEN);
		}
	}

	single_chunk = 0;
	rctx = NULL;

	/*
	 * Get number of lCPUs. When archiving with advanced filters, we use one less
	 * lCPU to reduce threads due to increased memory requirements.
	 */
	nprocs = (uint32_t)sysconf(_SC_NPROCESSORS_ONLN);
	if (pctx->archive_mode && (pctx->enable_packjpg || pctx->enable_wavpack)) {
		nprocs = nprocs > 1 ? nprocs-1:nprocs;
	}

	if (pctx->nthreads > 0 && pctx->nthreads < nprocs)
		nprocs = pctx->nthreads;
	else
		pctx->nthreads = nprocs;

	/* A host of sanity checks. */
	if (!pctx->pipe_mode) {
		char *tmp;
		if (!(pctx->archive_mode)) {
			if ((uncompfd = open(filename, O_RDONLY, 0)) == -1) {
				log_msg(LOG_ERR, 1, "Cannot open: %s", filename);
				return (1);
			}

			if (fstat(uncompfd, &sbuf) == -1) {
				close(uncompfd);
				log_msg(LOG_ERR, 1, "Cannot stat: %s", filename);
				return (1);
			}

			if (!S_ISREG(sbuf.st_mode)) {
				close(uncompfd);
				log_msg(LOG_ERR, 0, "File %s is not a regular file.", filename);
				return (1);
			}

			if (sbuf.st_size == 0) {
				close(uncompfd);
				return (1);
			}
		} else {
			if (setup_archiver(pctx, &sbuf) == -1) {
				log_msg(LOG_ERR, 0, "Setup archiver failed.");
				return (1);
			}
		}

		/*
		 * Adjust chunk size for small files. We then get an archive with
		 * a single chunk for the entire file.
		 * This is not valid for archive mode since we cannot accurately estimate
		 * final archive size.
		 */
		if (sbuf.st_size <= chunksize && !(pctx->archive_mode)) {
			chunksize = sbuf.st_size;
			pctx->enable_rabin_split = 0; // Do not split for whole files.
			pctx->nthreads = 1;
			single_chunk = 1;
			props.is_single_chunk = 1;
			flags |= FLAG_SINGLE_CHUNK;

			/*
			 * Disable deduplication if file is too small.
			 */
			if (chunksize < RAB_MIN_CHUNK_SIZE) {
				pctx->enable_rabin_scan = 0;
				pctx->enable_rabin_global = 0;
			}

			/*
			 * Switch to simple Deduplication if global is enabled.
			 */
			if (pctx->enable_rabin_global) {
				unsigned short flg;
				pctx->enable_rabin_scan = 1;
				pctx->enable_rabin_global = 0;
				dedupe_flag = RABIN_DEDUPE_SEGMENTED;
				flg = FLAG_DEDUP_FIXED;
				flags &= ~flg;
			}
		} else {
			if (pctx->nthreads == 0 || pctx->nthreads > sbuf.st_size / chunksize) {
				pctx->nthreads = (int)(sbuf.st_size / chunksize);
				if (sbuf.st_size % chunksize)
					pctx->nthreads++;
			}
		}

		/*
		 * Create a temporary file to hold compressed data which is renamed at
		 * the end. The target file name is same as original file with the '.pz'
		 * extension appended unless '-' was specified to output to stdout.
		 */
		if (filename) {
			strcpy(tmpfile1, filename);
			strcpy(tmpfile1, dirname(tmpfile1));
		} else {
			char *tmp1;
			if (!(pctx->archive_mode)) {
				log_msg(LOG_ERR, 0, "Inconsistent NULL Filename when Not archiving.");
				COMP_BAIL;
			}
			tmp1 = get_temp_dir();
			strcpy(tmpfile1, tmp1);
			free(tmp1);
		}

		tmp = getenv("PCOMPRESS_CACHE_DIR");
		if (tmp == NULL || !chk_dir(tmp)) {
			strcpy(tmpdir, tmpfile1);
		} else {
			strcpy(tmpdir, tmp);
		}

		if (pctx->pipe_out) {
			compfd = fileno(stdout);
			if (compfd == -1) {
				log_msg(LOG_ERR, 1, "fileno ");
				COMP_BAIL;
			}
		} else {
			if (pctx->to_filename == NULL) {
				strcat(tmpfile1, "/.pcompXXXXXX");
				snprintf(to_filename, sizeof (to_filename), "%s" COMP_EXTN, filename);
				if ((compfd = mkstemp(tmpfile1)) == -1) {
					log_msg(LOG_ERR, 1, "mkstemp ");
					COMP_BAIL;
				}
				add_fname(tmpfile1);
			} else {
				if (!endswith(pctx->to_filename, COMP_EXTN))
					snprintf(to_filename, sizeof (to_filename),
					    "%s" COMP_EXTN, pctx->to_filename);
				else
					snprintf(to_filename, sizeof (to_filename),
					    "%s", pctx->to_filename);
				if ((compfd = open(to_filename, O_CREAT|O_RDWR, S_IRUSR|S_IWUSR)) == -1) {
					log_msg(LOG_ERR, 1, "open ");
					COMP_BAIL;
				}
				add_fname(to_filename);
			}
		}
	} else {
		char *tmp;

		/*
		 * Use stdin/stdout for pipe mode.
		 */
		compfd = fileno(stdout);
		if (compfd == -1) {
			log_msg(LOG_ERR, 1, "fileno ");
			COMP_BAIL;
		}
		uncompfd = fileno(stdin);
		if (uncompfd == -1) {
			log_msg(LOG_ERR, 1, "fileno ");
			COMP_BAIL;
		}

		/*
		 * Get a workable temporary dir. Required if global dedupe is enabled.
		 */
		tmp = get_temp_dir();
		strcpy(tmpdir, tmp);
		free(tmp);
	}

	if (pctx->enable_rabin_global) {
		my_sysinfo msys_info;

		get_sys_limits(&msys_info);
		global_dedupe_bufadjust(pctx->rab_blk_size, &chunksize, 0, pctx->algo,
		    pctx->cksum, CKSUM_BLAKE256, sbuf.st_size, msys_info.freeram,
		    pctx->nthreads, pctx->pipe_mode);
	}

	/*
	 * Compressed buffer size must include zlib/dedup scratch space and
	 * chunk header space.
	 * See http://www.zlib.net/manual.html#compress2
	 *
	 * We do this unconditionally whether user mentioned zlib or not
	 * to keep it simple. While zlib scratch space is only needed at
	 * runtime, chunk header is stored in the file.
	 *
	 * See start_decompress() routine for details of chunk header.
	 * We also keep extra 8-byte space for the last chunk's size.
	 * compressed_chunksize might also already be set to accomodate a HMAC
	 * when encrypting, so we have to add to than here. Otherwise it is set
	 * to 0.
	 */
	compressed_chunksize += chunksize + CHUNK_HDR_SZ + zlib_buf_extra(chunksize);
	if (pctx->_props_func) {
		pctx->_props_func(&props, level, chunksize);
		if (chunksize + props.buf_extra > compressed_chunksize) {
			compressed_chunksize += (chunksize + props.buf_extra - 
			    compressed_chunksize);
		}
	}

	if (pctx->enable_rabin_scan || pctx->enable_fixed_scan || pctx->enable_rabin_global) {
		if (pctx->enable_rabin_global) {
			flags |= (FLAG_DEDUP | FLAG_DEDUP_FIXED);
			dedupe_flag = RABIN_DEDUPE_FILE_GLOBAL;
		} else if (pctx->enable_rabin_scan) {
			flags |= FLAG_DEDUP;
			dedupe_flag = RABIN_DEDUPE_SEGMENTED;
		} else {
			flags |= FLAG_DEDUP_FIXED;
			dedupe_flag = RABIN_DEDUPE_FIXED;
		}
		/* Additional scratch space for dedup arrays. */
		if (chunksize + dedupe_buf_extra(chunksize, 0, pctx->algo, pctx->enable_delta_encode)
		    > compressed_chunksize) {
			compressed_chunksize += (chunksize +
			    dedupe_buf_extra(chunksize, 0, pctx->algo, pctx->enable_delta_encode)) -
			    compressed_chunksize;
		}
	}

	slab_cache_add(chunksize);
	slab_cache_add(compressed_chunksize);
	slab_cache_add(sizeof (struct cmp_data));

	if (pctx->encrypt_type)
		flags |= pctx->encrypt_type;

	set_threadcounts(&props, &(pctx->nthreads), nprocs, COMPRESS_THREADS);
	if (pctx->nthreads * props.nthreads > 1)
		log_msg(LOG_INFO, 0, "Scaling to %d threads", pctx->nthreads * props.nthreads);
	else
		log_msg(LOG_INFO, 0, "Scaling to 1 thread");
	nprocs = pctx->nthreads;
	dary = (struct cmp_data **)slab_calloc(NULL, nprocs, sizeof (struct cmp_data *));
	cread_buf = (uchar_t *)slab_alloc(NULL, compressed_chunksize);
	if (!cread_buf) {
		log_msg(LOG_ERR, 0, "3: Out of memory");
		COMP_BAIL;
	}

	for (i = 0; i < nprocs; i++) {
		dary[i] = (struct cmp_data *)slab_alloc(NULL, sizeof (struct cmp_data));
		if (!dary[i]) {
			log_msg(LOG_ERR, 0, "4: Out of memory");
			COMP_BAIL;
		}
		tdat = dary[i];
		tdat->pctx = pctx;
		tdat->cmp_seg = NULL;
		tdat->chunksize = chunksize;
		tdat->compress = pctx->_compress_func;
		tdat->decompress = pctx->_decompress_func;
		tdat->uncompressed_chunk = (uchar_t *)1;
		if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan)) {
			if (single_chunk)
				tdat->cmp_seg = (uchar_t *)1;
			else
				tdat->cmp_seg = (uchar_t *)slab_alloc(NULL, compressed_chunksize);
			tdat->uncompressed_chunk = (uchar_t *)slab_alloc(NULL,
				compressed_chunksize);
		} else {
			if (single_chunk)
				tdat->uncompressed_chunk = (uchar_t *)1;
			else
				tdat->uncompressed_chunk = (uchar_t *)slab_alloc(NULL,
					compressed_chunksize);
			tdat->cmp_seg = (uchar_t *)slab_alloc(NULL, compressed_chunksize);
		}
		tdat->compressed_chunk = tdat->cmp_seg + COMPRESSED_CHUNKSZ +
		    pctx->cksum_bytes + pctx->mac_bytes;
		if (!tdat->cmp_seg || !tdat->uncompressed_chunk) {
			log_msg(LOG_ERR, 0, "5: Out of memory");
			COMP_BAIL;
		}
		tdat->cancel = 0;
		tdat->decompressing = 0;
		if (single_chunk)
			tdat->cksum_mt = 1;
		else
			tdat->cksum_mt = 0;
		tdat->level = level;
		tdat->data = NULL;
		tdat->rctx = NULL;
		tdat->props = &props;
		Sem_Init(&(tdat->start_sem), 0, 0);
		Sem_Init(&(tdat->cmp_done_sem), 0, 0);
		Sem_Init(&(tdat->write_done_sem), 0, 1);
		Sem_Init(&(tdat->index_sem), 0, 0);

		if (pctx->_init_func) {
			if (pctx->_init_func(&(tdat->data), &(tdat->level), props.nthreads,
			    chunksize, VERSION, COMPRESS) != 0) {
				COMP_BAIL;
			}
		}

		if (pctx->encrypt_type) {
			if (hmac_init(&tdat->chunk_hmac, pctx->cksum, &(pctx->crypto_ctx)) == -1) {
				log_msg(LOG_ERR, 0, "Cannot initialize chunk hmac.");
				COMP_BAIL;
			}
		}
		if (pthread_create(&(tdat->thr), NULL, perform_compress,
		    (void *)tdat) != 0) {
			log_msg(LOG_ERR, 1, "Error in thread creation: ");
			COMP_BAIL;
		}
	}

	/*
	 * Now create the metadata handler context. This is relevant in archive mode where
	 * the underlying libarchive metadata is compressed into a separate stream of
	 * metadata chunks.
	 */
	if (pctx->meta_stream) {
		pctx->meta_ctx = meta_ctx_create(pctx, VERSION, compfd);
		if (pctx->meta_ctx == NULL) {
			COMP_BAIL;
		}
	}
	thread = 1;

	/*
	 * initialize Dedupe Context here after all other allocations so that index size can be
	 * correctly computed based on free memory. The freeram got here is adjusted amount.
	 * When archiving, filter scratch buffer is taken into account.
	 */
	get_sys_limits(&msys_info);

	if (pctx->enable_packjpg || pctx->enable_wavpack) {
		if (FILTER_SCRATCH_SIZE_MAX >= msys_info.freeram ||
		    msys_info.freeram - FILTER_SCRATCH_SIZE_MAX < FILTER_SCRATCH_SIZE_MAX) {
			log_msg(LOG_WARN, 0, "Not enough memory. Disabling advanced filters.");
			disable_all_filters();
		} else {
			msys_info.freeram -= FILTER_SCRATCH_SIZE_MAX;
		}
	}

	if (pctx->enable_rabin_scan || pctx->enable_fixed_scan || pctx->enable_rabin_global) {
		for (i = 0; i < nprocs; i++) {
			tdat = dary[i];
			tdat->rctx = create_dedupe_context(chunksize, compressed_chunksize,
			    pctx->rab_blk_size, pctx->algo, &props, pctx->enable_delta_encode,
			    dedupe_flag, VERSION, COMPRESS, sbuf.st_size, tmpdir,
			    pctx->pipe_mode, nprocs, msys_info.freeram);
			if (tdat->rctx == NULL) {
				COMP_BAIL;
			}

			tdat->rctx->show_chunks = pctx->show_chunks;
			tdat->rctx->index_sem = &(tdat->index_sem);
			tdat->rctx->id = i;
		}
	}
	if (pctx->enable_rabin_global) {
		for (i = 0; i < nprocs; i++) {
			tdat = dary[i];
			tdat->rctx->index_sem_next = &(dary[(i + 1) % nprocs]->index_sem);
		}
		// When doing global dedupe first thread does not wait to access the index.
		Sem_Post(&(dary[0]->index_sem));
	}

	w.dary = dary;
	w.wfd = compfd;
	w.nprocs = nprocs;
	w.pctx = pctx;
	if (pthread_create(&writer_thr, NULL, writer_thread, (void *)(&w)) != 0) {
		log_msg(LOG_ERR, 1, "Error in thread creation: ");
		COMP_BAIL;
	}
	thread = 2;

	/*
	 * Start the archiver thread if needed.
	 */
	if (pctx->archive_mode) {
		if (start_archiver(pctx) != 0) {
			COMP_BAIL;
		}
		flags |= FLAG_ARCHIVE;
		if (pctx->meta_stream)
			flags |= FLAG_META_STREAM;
	}

	/*
	 * Write out file header. First insert hdr elements into mem buffer
	 * then write out the full hdr in one shot.
	 */
	flags |= pctx->cksum;
	memset(cread_buf, 0, ALGO_SZ);
	strncpy((char *)cread_buf, pctx->algo, ALGO_SZ);
	version = htons(VERSION);
	flags = htons(flags);
	n_chunksize = htonll(chunksize);
	level = htonl(level);
	pos = cread_buf + ALGO_SZ;
	memcpy(pos, &version, sizeof (version));
	pos += sizeof (version);
	memcpy(pos, &flags, sizeof (flags));
	pos += sizeof (flags);
	memcpy(pos, &n_chunksize, sizeof (n_chunksize));
	pos += sizeof (n_chunksize);
	memcpy(pos, &level, sizeof (level));
	pos += sizeof (level);

	/*
	 * If encryption is enabled, include salt, nonce and keylen in the header
	 * to be HMAC-ed (archive version 7 and greater).
	 */
	if (pctx->encrypt_type) {
		*((int *)pos) = htonl(pctx->crypto_ctx.saltlen);
		pos += sizeof (int);
		serialize_checksum(pctx->crypto_ctx.salt, pos, pctx->crypto_ctx.saltlen);
		pos += pctx->crypto_ctx.saltlen;
		if (pctx->encrypt_type == CRYPTO_ALG_AES) {
			U64_P(pos) = htonll(U64_P(crypto_nonce(&(pctx->crypto_ctx))));
			pos += 8;

		} else if (pctx->encrypt_type == CRYPTO_ALG_SALSA20) {
			serialize_checksum(crypto_nonce(&(pctx->crypto_ctx)), pos,
			    XSALSA20_CRYPTO_NONCEBYTES);
			pos += XSALSA20_CRYPTO_NONCEBYTES;
		}
		*((int *)pos) = htonl(pctx->keylen);
		pos += sizeof (int);
	}
	if (Write(compfd, cread_buf, pos - cread_buf) != pos - cread_buf) {
		log_msg(LOG_ERR, 1, "Write ");
		COMP_BAIL;
	}

	/*
	 * If encryption is enabled, compute header HMAC and write it.
	 */
	if (pctx->encrypt_type) {
		mac_ctx_t hdr_mac;
		uchar_t hdr_hash[pctx->mac_bytes];
		unsigned int hlen;

		if (hmac_init(&hdr_mac, pctx->cksum, &(pctx->crypto_ctx)) == -1) {
			log_msg(LOG_ERR, 0, "Cannot initialize header hmac.");
			COMP_BAIL;
		}
		hmac_update(&hdr_mac, cread_buf, pos - cread_buf);
		hmac_final(&hdr_mac, hdr_hash, &hlen);
		hmac_cleanup(&hdr_mac);

		/* Erase encryption key bytes stored as a plain array. No longer reqd. */
		crypto_clean_pkey(&(pctx->crypto_ctx));

		pos = cread_buf;
		serialize_checksum(hdr_hash, pos, hlen);
		pos += hlen;
		if (Write(compfd, cread_buf, pos - cread_buf) != pos - cread_buf) {
			log_msg(LOG_ERR, 1, "Write ");
			COMP_BAIL;
		}
	} else {
		/*
		 * Compute header CRC32 and store that. Only archive version 5 and above.
		 */
		uint32_t crc = lzma_crc32(cread_buf, pos - cread_buf, 0);
		U32_P(cread_buf) = htonl(crc);
		if (Write(compfd, cread_buf, sizeof (uint32_t)) != sizeof (uint32_t)) {
			log_msg(LOG_ERR, 1, "Write ");
			COMP_BAIL;
		}
	}

	/*
	 * Now read from the uncompressed file in 'chunksize' sized chunks, independently
	 * compress each chunk and write it out. Chunk sequencing is ensured.
	 */
	pctx->chunk_num = 0;
	np = 0;
	bail = 0;
	pctx->largest_chunk = 0;
	pctx->smallest_chunk = chunksize;
	pctx->avg_chunk = 0;
	rabin_count = 0;

	/*
	 * Read the first chunk into a spare buffer (a simple double-buffering).
	 */
	file_offset = 0;
	pctx->interesting = 0;
	if (pctx->enable_rabin_split) {
		rctx = create_dedupe_context(chunksize, 0, pctx->rab_blk_size, pctx->algo, &props,
		    pctx->enable_delta_encode, pctx->enable_fixed_scan, VERSION, COMPRESS, 0, NULL,
		    pctx->pipe_mode, nprocs, msys_info.freeram);
		if (pctx->archive_mode)
			rbytes = Read_Adjusted(uncompfd, cread_buf, chunksize, &rabin_count, rctx, pctx);
		else
			rbytes = Read_Adjusted(uncompfd, cread_buf, chunksize, &rabin_count, rctx, NULL);
	} else {
		if (pctx->archive_mode)
			rbytes = archiver_read(pctx, cread_buf, chunksize);
		else
			rbytes = Read(uncompfd, cread_buf, chunksize);
	}

	while (!bail) {
		uchar_t *tmp;

		if (pctx->main_cancel) break;
		for (p = 0; p < nprocs; p++) {
			np = p;
			tdat = dary[p];
			if (pctx->main_cancel) break;
			/* Wait for previous chunk compression to complete. */
			Sem_Wait(&tdat->write_done_sem);
			if (pctx->main_cancel) break;

			if (rbytes == 0) { /* EOF */
				bail = 1;
				break;
			}

			/*
			 * Once previous chunk is done swap already read buffer and
			 * it's size into the thread data.
			 * Normally it goes into uncompressed_chunk, because that's what it is.
			 * With dedup enabled however, we do some jugglery to save additional
			 * memory usage and avoid a memcpy, so it goes into the compressed_chunk
			 * area:
			 * cmp_seg -> dedup -> uncompressed_chunk -> compression -> cmp_seg
			 */
			tdat->id = pctx->chunk_num;
			tdat->rbytes = rbytes;
			tdat->interesting = pctx->interesting;
			tdat->btype = pctx->btype; // Have to copy btype for this buffer as pctx->btype will change
			if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan || pctx->enable_rabin_global)) {
				tmp = tdat->cmp_seg;
				tdat->cmp_seg = cread_buf;
				cread_buf = tmp;
				tdat->compressed_chunk = tdat->cmp_seg + COMPRESSED_CHUNKSZ +
				    pctx->cksum_bytes + pctx->mac_bytes;
				if (tdat->rctx) tdat->rctx->file_offset = file_offset;

				/*
				 * If there is data after the last rabin boundary in the chunk, then
				 * rabin_count will be non-zero. We carry over the data to the beginning
				 * of the next chunk.
				 */
				if (rabin_count) {
					memcpy(cread_buf,
					    tdat->cmp_seg + rabin_count, rbytes - rabin_count);
					tdat->rbytes = rabin_count;
					rabin_count = rbytes - rabin_count;
				}
			} else {
				tmp = tdat->uncompressed_chunk;
				tdat->uncompressed_chunk = cread_buf;
				cread_buf = tmp;
			}
			file_offset += tdat->rbytes;

			if (rbytes < chunksize) {
				if (rbytes < 0) {
					bail = 1;
					log_msg(LOG_ERR, 1, "Read: ");
					COMP_BAIL;
				}
			}

			/* Signal the compression thread to start */
			Sem_Post(&tdat->start_sem);
			++(pctx->chunk_num);

			if (single_chunk) {
				rbytes = 0;
				continue;
			}

			/*
			 * Read the next buffer we want to process while previous
			 * buffer is in progress.
			 */
			pctx->interesting = 0;
			if (pctx->enable_rabin_split) {
				if (pctx->archive_mode)
					rbytes = Read_Adjusted(uncompfd, cread_buf, chunksize,
					    &rabin_count, rctx, pctx);
				else
					rbytes = Read_Adjusted(uncompfd, cread_buf, chunksize,
					    &rabin_count, rctx, NULL);
			} else {
				if (pctx->archive_mode)
					rbytes = archiver_read(pctx, cread_buf, chunksize);
				else
					rbytes = Read(uncompfd, cread_buf, chunksize);
			}
		}
	}

	if (!pctx->main_cancel) {
		/* Wait for all remaining chunks to finish. */
		for (p = 0; p < nprocs; p++) {
			if (p == np) continue;
			tdat = dary[p];
			Sem_Wait(&tdat->write_done_sem);
		}
	} else {
		err = 1;
	}

comp_done:
	/*
	 * First close the input fd of uncompressed data. If archiving this will cause
	 * the archive thread to exit and cleanup.
	 */
	if (!pctx->pipe_mode) {
		if (uncompfd != -1) close(uncompfd);
	}
	if (pctx->meta_stream) {
		meta_ctx_done(pctx->meta_ctx);
		archiver_close(pctx);
	}

	if (pctx->t_errored) err = pctx->t_errored;
	if (thread) {
		for (i = 0; i < nprocs; i++) {
			tdat = dary[i];
			tdat->cancel = 1;
			tdat->len_cmp = 0;
			Sem_Post(&tdat->start_sem);
			Sem_Post(&tdat->cmp_done_sem);
			pthread_join(tdat->thr, NULL);
			if (pctx->encrypt_type)
				hmac_cleanup(&tdat->chunk_hmac);
		}
		if (thread == 2)
			pthread_join(writer_thr, NULL);
	}

	if (err) {
		if (compfd != -1 && !pctx->pipe_mode && !pctx->pipe_out) {
			unlink(tmpfile1);
			rm_fname(tmpfile1);
		}
		if (filename)
			log_msg(LOG_ERR, 0, "Error compressing file: %s", filename);
		else
			log_msg(LOG_ERR, 0, "Error compressing");
	} else {
		/*
		* Write a trailer of zero chunk length.
		*/
		compressed_chunksize = 0;
		if (Write(compfd, &compressed_chunksize,
		    sizeof (compressed_chunksize)) < 0) {
			log_msg(LOG_ERR, 1, "Write ");
			err = 1;
		}

		/*
		 * Rename the temporary file to the actual compressed file
		 * unless we are in a pipe.
		 */
		if (!pctx->pipe_mode && !pctx->pipe_out) {
			/*
			 * Ownership and mode of target should be same as original.
			 */
			fchmod(compfd, sbuf.st_mode);
			if (fchown(compfd, sbuf.st_uid, sbuf.st_gid) == -1)
				log_msg(LOG_ERR, 1, "chown ");
			close(compfd);

			if (pctx->to_filename == NULL) {
				if (rename(tmpfile1, to_filename) == -1) {
					log_msg(LOG_ERR, 1, "Cannot rename temporary file ");
					unlink(tmpfile1);
				}
				rm_fname(tmpfile1);
			} else {
				rm_fname(to_filename);
			}
		}
	}
	if (dary != NULL) {
		for (i = 0; i < nprocs; i++) {
			if (!dary[i]) continue;
			if (dary[i]->uncompressed_chunk != (uchar_t *)1)
				slab_release(NULL, dary[i]->uncompressed_chunk);
			if (dary[i]->cmp_seg != (uchar_t *)1)
				slab_release(NULL, dary[i]->cmp_seg);
			if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan)) {
				destroy_dedupe_context(dary[i]->rctx);
			}
			if (pctx->_deinit_func)
				pctx->_deinit_func(&(dary[i]->data));
			Sem_Destroy(&(dary[i]->start_sem));
			Sem_Destroy(&(dary[i]->cmp_done_sem));
			Sem_Destroy(&(dary[i]->write_done_sem));
			Sem_Destroy(&(dary[i]->index_sem));

			slab_release(NULL, dary[i]);
		}
		slab_release(NULL, dary);
	}
	if (pctx->enable_rabin_split) destroy_dedupe_context(rctx);
	if (cread_buf != (uchar_t *)1)
		slab_release(NULL, cread_buf);
	if (!pctx->pipe_mode) {
		if (compfd != -1) close(compfd);
	}

	if (pctx->archive_mode) {
		struct fn_list *fn, *fn1;

		pthread_join(pctx->archive_thread, NULL);
		fn = pctx->fn;
		while (fn) {
			fn1 = fn;
			fn = fn->next;
			slab_release(NULL, fn1);
		}
		Sem_Destroy(&(pctx->read_sem));
		Sem_Destroy(&(pctx->write_sem));
	}
	if (!pctx->hide_cmp_stats) show_compression_stats(pctx);
	pctx->_stats_func(!pctx->hide_cmp_stats);

	return (err);
}

/*
 * Check the algorithm requested and set the callback routine pointers.
 */
static int
init_algo(pc_ctx_t *pctx, const char *algo, int bail)
{
	int rv = 1;
	char algorithm[8];

	/* Copy given string into known length buffer to avoid memcmp() overruns. */
	strncpy(algorithm, algo, 8);
	pctx->_props_func = NULL;
	if (memcmp(algorithm, "zlib", 4) == 0) {
		pctx->_compress_func = zlib_compress;
		pctx->_decompress_func = zlib_decompress;
		pctx->_init_func = zlib_init;
		pctx->_deinit_func = zlib_deinit;
		pctx->_stats_func = zlib_stats;
		pctx->_props_func = zlib_props;
		rv = 0;

	} else if (memcmp(algorithm, "lzmaMt", 6) == 0) {
		pctx->_compress_func = lzma_compress;
		pctx->_decompress_func = lzma_decompress;
		pctx->_init_func = lzma_init;
		pctx->_deinit_func = lzma_deinit;
		pctx->_stats_func = lzma_stats;
		pctx->_props_func = lzma_mt_props;
		rv = 0;

	} else if (memcmp(algorithm, "lzma", 4) == 0) {
		pctx->_compress_func = lzma_compress;
		pctx->_decompress_func = lzma_decompress;
		pctx->_init_func = lzma_init;
		pctx->_deinit_func = lzma_deinit;
		pctx->_stats_func = lzma_stats;
		pctx->_props_func = lzma_props;
		rv = 0;

	} else if (memcmp(algorithm, "bzip2", 5) == 0) {
		pctx->_compress_func = bzip2_compress;
		pctx->_decompress_func = bzip2_decompress;
		pctx->_init_func = bzip2_init;
		pctx->_deinit_func = NULL;
		pctx->_stats_func = bzip2_stats;
		pctx->_props_func = bzip2_props;
		rv = 0;

	} else if (memcmp(algorithm, "ppmd", 4) == 0) {
		pctx->_compress_func = ppmd_compress;
		pctx->_decompress_func = ppmd_decompress;
		pctx->_init_func = ppmd_init;
		pctx->_deinit_func = ppmd_deinit;
		pctx->_stats_func = ppmd_stats;
		pctx->_props_func = ppmd_props;
		rv = 0;

	} else if (memcmp(algorithm, "lzfx", 4) == 0) {
		pctx->_compress_func = lz_fx_compress;
		pctx->_decompress_func = lz_fx_decompress;
		pctx->_init_func = lz_fx_init;
		pctx->_deinit_func = lz_fx_deinit;
		pctx->_stats_func = lz_fx_stats;
		pctx->_props_func = lz_fx_props;
		rv = 0;

	} else if (memcmp(algorithm, "lz4", 3) == 0) {
		pctx->_compress_func = lz4_compress;
		pctx->_decompress_func = lz4_decompress;
		pctx->_init_func = lz4_init;
		pctx->_deinit_func = lz4_deinit;
		pctx->_stats_func = lz4_stats;
		pctx->_props_func = lz4_props;
		rv = 0;

	} else if (memcmp(algorithm, "none", 4) == 0) {
		pctx->_compress_func = none_compress;
		pctx->_decompress_func = none_decompress;
		pctx->_init_func = none_init;
		pctx->_deinit_func = none_deinit;
		pctx->_stats_func = none_stats;
		pctx->_props_func = none_props;
		rv = 0;

	/* adapt2 and adapt ordering of the checks matter here. */
	} else if (memcmp(algorithm, "adapt2", 6) == 0) {
		pctx->_compress_func = adapt_compress;
		pctx->_decompress_func = adapt_decompress;
		pctx->_init_func = adapt2_init;
		pctx->_deinit_func = adapt_deinit;
		pctx->_stats_func = adapt_stats;
		pctx->_props_func = adapt_props;
		pctx->adapt_mode = 2;
		pctx->enable_analyzer = 1;
		rv = 0;

	} else if (memcmp(algorithm, "adapt", 5) == 0) {
		pctx->_compress_func = adapt_compress;
		pctx->_decompress_func = adapt_decompress;
		pctx->_init_func = adapt_init;
		pctx->_deinit_func = adapt_deinit;
		pctx->_stats_func = adapt_stats;
		pctx->_props_func = adapt_props;
		pctx->adapt_mode = 1;
		pctx->enable_analyzer = 1;
		rv = 0;
#ifdef ENABLE_PC_LIBBSC
	} else if (memcmp(algorithm, "libbsc", 6) == 0) {
		pctx->_compress_func = libbsc_compress;
		pctx->_decompress_func = libbsc_decompress;
		pctx->_init_func = libbsc_init;
		pctx->_deinit_func = libbsc_deinit;
		pctx->_stats_func = libbsc_stats;
		pctx->_props_func = libbsc_props;
		rv = 0;
#endif
	}

	return (rv);
}

/*
 * Pcompress context handling functions.
 */
pc_ctx_t DLL_EXPORT * 
create_pc_context(void)
{
	pc_ctx_t *ctx = (pc_ctx_t *)malloc(sizeof (pc_ctx_t));

	slab_init();
	init_pcompress();
	init_archive_mod();

	memset(ctx, 0, sizeof (pc_ctx_t));
	ctx->exec_name = (char *)malloc(NAME_MAX);
	ctx->hide_mem_stats = 1;
	ctx->hide_cmp_stats = 1;
	ctx->enable_rabin_split = 1;
	ctx->rab_blk_size = -1;
	ctx->archive_temp_fd = -1;
	ctx->pagesize = sysconf(_SC_PAGE_SIZE);
	ctx->btype = TYPE_UNKNOWN;
	ctx->delta2_nstrides = NSTRIDES_STANDARD;
	pthread_mutex_init(&ctx->write_mutex, NULL);

	return (ctx);
}

void DLL_EXPORT
destroy_pc_context(pc_ctx_t *pctx)
{
	if (pctx->do_compress)
		free((void *)(pctx->filename));
	if (pctx->pwd_file)
		free(pctx->pwd_file);
	free((void *)(pctx->exec_name));
	slab_cleanup(pctx->hide_mem_stats);
	free(pctx);
}

int DLL_EXPORT
init_pc_context_argstr(pc_ctx_t *pctx, char *args)
{
	int ac;
	char *av[128];
	char *sptr, *tok;

	ac = 0;
	tok = strtok_r(args, " ", &sptr);

	while (tok != NULL && ac < 128) {
		av[ac++] = tok;
		tok = strtok_r(NULL, " ", &sptr);
	}
	if (ac > 0)
		return (init_pc_context(pctx, ac, av));
	return (0);
}

int DLL_EXPORT
init_pc_context(pc_ctx_t *pctx, int argc, char *argv[])
{
	int opt, num_rem, err, my_optind;
	char *pos;
	struct filter_flags ff;

	pctx->level = -1;
	err = 0;
	pctx->keylen = DEFAULT_KEYLEN;
	pctx->chunksize = 0;
	pos = argv[0] + strlen(argv[0]);
	while (*pos != '/' && pos > argv[0]) pos--;
	if (*pos == '/') pos++;
	strcpy(pctx->exec_name, pos);
	pctx->advanced_opts = 0;
	ff.enable_packjpg = 0;
	ff.enable_wavpack = 0;
	ff.exe_preprocess = 0;

	pthread_mutex_lock(&opt_parse);
	while ((opt = getopt(argc, argv, "dc:s:l:pt:MCDGEe:w:LPS:B:Fk:avmKjxiTn")) != -1) {
		int ovr;
		int64_t chunksize;

		switch (opt) {
		    case 'i':
			pctx->list_mode = 1; // List mode also sets decompress flag
		    case 'd':
			pctx->do_uncompress = 1;
			break;

		    case 'c':
			pctx->do_compress = 1;
			pctx->algo = optarg;
			if (init_algo(pctx, pctx->algo, 1) != 0) {
				log_msg(LOG_ERR, 0, "Invalid algorithm %s", optarg);
				return (1);
			}
			break;

		    case 's':
			ovr = parse_numeric(&chunksize, optarg);
			if (ovr == 1) {
				log_msg(LOG_ERR, 0, "Chunk size too large %s", optarg);
				return (1);

			} else if (ovr == 2) {
				log_msg(LOG_ERR, 0, "Invalid number %s", optarg);
				return (1);
			}
			pctx->chunksize = chunksize;

			if (pctx->chunksize < MIN_CHUNK) {
				log_msg(LOG_ERR, 0, "Minimum chunk size is %ld", MIN_CHUNK);
				return (1);
			}
			if (pctx->chunksize > EIGHTY_PCT(get_total_ram())) {
				log_msg(LOG_ERR, 0, "Chunk size must not exceed 80%% of total RAM.");
				return (1);
			}
			break;

		    case 'l':
			pctx->level = atoi(optarg);
			if (pctx->level < 0 || pctx->level > MAX_LEVEL) {
				log_msg(LOG_ERR, 0, "Compression level should be in range 0 - 14");
				return (1);
			}
			break;

		    case 'B':
			pctx->rab_blk_size = atoi(optarg);
			if (pctx->rab_blk_size < 0 || pctx->rab_blk_size > 5) {
				log_msg(LOG_ERR, 0, "Average Dedupe block size must be in range 0 (2k), 1 (4k) .. 5 (64k)");
				return (1);
			}
			break;

		    case 'p':
			pctx->pipe_mode = 1;
			break;

		    case 't':
			pctx->nthreads = atoi(optarg);
			if (pctx->nthreads < 1 || pctx->nthreads > 256) {
				log_msg(LOG_ERR, 0, "Thread count should be in range 1 - 256");
				return (1);
			}
			break;

		    case 'M':
			pctx->hide_mem_stats = 0;
			break;

		    case 'C':
			if (pctx->hide_cmp_stats)
				pctx->hide_cmp_stats = 0;
			else
				pctx->show_chunks = 1;
			break;

		    case 'D':
			pctx->advanced_opts = 1;
			pctx->enable_rabin_scan = 1;
			break;

		    case 'G':
			pctx->advanced_opts = 1;
			pctx->enable_rabin_global = 1;
			break;

		    case 'E':
			pctx->advanced_opts = 1;
			pctx->enable_rabin_scan = 1;
			if (!pctx->enable_delta_encode)
				pctx->enable_delta_encode = DELTA_NORMAL;
			else
				pctx->enable_delta_encode = DELTA_EXTRA;
			break;

		    case 'e':
			pctx->encrypt_type = get_crypto_alg(optarg);
			if (pctx->encrypt_type == 0) {
				log_msg(LOG_ERR, 0, "Invalid encryption algorithm. "
				    "Should be AES or SALSA20.", optarg);
				return (1);
			}
			break;

		    case 'w':
			pctx->pwd_file = strdup(optarg);
			break;

		    case 'F':
			pctx->advanced_opts = 1;
			pctx->enable_fixed_scan = 1;
			pctx->enable_rabin_split = 0;
			break;

#ifndef _MPLV2_LICENSE_
		    case 'L':
			pctx->advanced_opts = 1;
			pctx->lzp_preprocess = 1;
			break;
#endif

		    case 'P':
			pctx->advanced_opts = 1;
			pctx->enable_delta2_encode = 1;
			break;

		    case 'k':
			pctx->keylen = atoi(optarg);
			if ((pctx->keylen != 16 && pctx->keylen != 32) || pctx->keylen > MAX_KEYLEN) {
				log_msg(LOG_ERR, 0, "Encryption KEY length should be 16 or 32.", optarg);
				return (1);
			}
			break;

		    case 'S':
			if (get_checksum_props(optarg, &(pctx->cksum), &(pctx->cksum_bytes),
			    &(pctx->mac_bytes), 0) == -1) {
				log_msg(LOG_ERR, 0, "Invalid checksum type %s", optarg);
				return (1);
			}
			break;

		    case 'a':
			pctx->archive_mode = 1;
			pctx->do_compress = 1;
			break;

		    case 'v':
			set_log_level(LOG_VERBOSE);
			break;

		    case 'm':
			pctx->force_archive_perms = 1;
			break;

		    case 'K':
			pctx->no_overwrite_newer = 1;
			break;

		    case 'j':
			pctx->advanced_opts = 1;
			ff.enable_packjpg = 1;
			ff.enable_wavpack = 1;
			break;

		    case 'x':
			pctx->advanced_opts = 1;
			pctx->exe_preprocess = 1;
			ff.exe_preprocess = 1;
			break;

		    case 'T':
			pctx->meta_stream = -1;
			break;

		   case 'n':
			pctx->enable_archive_sort = -1;
			break;

		    case '?':
		    default:
			return (2);
			break;
		}
	}
	my_optind = optind;
	optind = 0;
	pthread_mutex_unlock(&opt_parse);

	if ((pctx->do_compress && pctx->do_uncompress) || (!pctx->do_compress && !pctx->do_uncompress)) {
		return (2);
	}

	if (pctx->archive_mode && pctx->do_uncompress) {
		log_msg(LOG_ERR, 0, "'-a' flag is only for archive creation.");
		return (1);
	}

	if (pctx->archive_mode && pctx->pipe_mode) {
		log_msg(LOG_ERR, 0, "Full pipeline mode is meaningless with archiver.");
		return (1);
	}

	/*
	 * Default compression algorithm during archiving is Adaptive2.
	 */
	if (pctx->archive_mode && pctx->algo == NULL) {
		pctx->algo = "adapt2";
		init_algo(pctx, pctx->algo, 1);
	}

	if (pctx->level == -1 && pctx->do_compress) {
		if (memcmp(pctx->algo, "lz4", 3) == 0) {
			pctx->level = 1;
		} else {
			pctx->level = 6;
		}
	}

	/*
	 * Sorting of members when archiving is enabled for compression levels >6 (>2 for lz4),
	 * unless it is explicitly disabled via '-n'.
	 */
	if (pctx->enable_archive_sort != -1 && pctx->do_compress) {
		if ((memcmp(pctx->algo, "lz4", 3) == 0 && pctx->level > 1) || pctx->level > 4)
			pctx->enable_archive_sort = 1;
	} else {
		pctx->enable_archive_sort = 0;
	}

	if (pctx->rab_blk_size == -1) {
		if (!pctx->enable_rabin_global)
			pctx->rab_blk_size = 0;
		else
			pctx->rab_blk_size = RAB_BLK_DEFAULT;
	}

	pctx->min_chunk = MIN_CHUNK;
	if (pctx->enable_rabin_scan)
		pctx->min_chunk = RAB_MIN_CHUNK_SIZE;
	if (pctx->enable_rabin_global)
		pctx->min_chunk = RAB_MIN_CHUNK_SIZE_GLOBAL;

	/*
	 * Remaining mandatory arguments are the filenames.
	 */
	num_rem = argc - my_optind;
	if (pctx->pipe_mode && num_rem > 0 ) {
		log_msg(LOG_ERR, 0, "Filename(s) unexpected for pipe mode");
		return (1);
	}

	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan) && !pctx->do_compress) {
		log_msg(LOG_ERR, 0, "Deduplication is only used during compression.");
		return (1);
	}
	if (!pctx->enable_rabin_scan)
		pctx->enable_rabin_split = 0;

	if (pctx->enable_fixed_scan && (pctx->enable_rabin_scan ||
	    pctx->enable_delta_encode || pctx->enable_rabin_split)) {
		log_msg(LOG_ERR, 0, "Rabin Deduplication and Fixed block Deduplication"
		    "are mutually exclusive");
		return (1);
	}

	if (!pctx->do_compress && pctx->encrypt_type) {
		log_msg(LOG_ERR, 0, "Encryption only makes sense when compressing!");
		return (1);

	} else if (pctx->pipe_mode && pctx->encrypt_type && !pctx->pwd_file) {
		log_msg(LOG_ERR, 0, "Pipe mode requires password to be provided in a file.");
		return (1);
	}

	/*
	 * Global Deduplication can use Rabin or Fixed chunking. Default, if not specified,
	 * is to use Rabin.
	 */
	if (pctx->enable_rabin_global && !pctx->enable_rabin_scan && !pctx->enable_fixed_scan) {
		pctx->enable_rabin_scan = 1;
		pctx->enable_rabin_split = 1;
	}

	if (pctx->enable_rabin_global && pctx->enable_delta_encode) {
		log_msg(LOG_ERR, 0, "Global Deduplication does not support Delta Compression.");
		return (1);
	}

	/*
	 * EXE, PackJPG and WavPack are only valid when archiving files.
	 */
	if ((pctx->exe_preprocess || ff.enable_packjpg || ff.enable_wavpack)
	    && !pctx->archive_mode) {
		log_msg(LOG_ERR, 0, "Executable File Preprocessor and PackJPG are "
		    "only valid when archiving.");
		return (1);
	}

	if (num_rem == 0 && !pctx->pipe_mode) {
		log_msg(LOG_ERR, 0, "Expected at least one filename.");
		return (1);

	} else if (num_rem == 1 || num_rem == 2 || (num_rem > 0 && pctx->archive_mode)) {
		if (pctx->do_compress) {
			char apath[MAXPATHLEN];

			/*
			 * If archiving, resolve the list of pathnames on the cmdline.
			 */
			if (pctx->archive_mode) {
				struct fn_list **fn;
				int valid_paths;

				slab_cache_add(sizeof (struct fn_list));
				pctx->filename = NULL;
				fn = &(pctx->fn);
				valid_paths = 0;
				while (num_rem > 0) {
					char *filename;

					if ((filename = realpath(argv[my_optind], NULL)) != NULL) {
						free(filename);
						*fn = slab_alloc(NULL, sizeof (struct fn_list));
						(*fn)->filename = strdup(argv[my_optind]);
						(*fn)->next = NULL;
						fn = &((*fn)->next);
						valid_paths++;
					} else {
						log_msg(LOG_WARN, 1, "%s", argv[my_optind]);
					}
					num_rem--;
					my_optind++;

					/*
					 * If multiple pathnames are provided, last one must be the archive name.
					 * This check here handles that case. If only one pathname is provided
					 * then archive name can be derived and num_rem here will be 0 so it
					 * exits normally in the loop check above.
					 */
					if (num_rem == 1) break;
				}
				if (valid_paths == 0) {
					log_msg(LOG_ERR, 0, "No usable paths found to archive.");
					return (1);
				}
				if (valid_paths == 1)
					pctx->filename = pctx->fn->filename;
			} else {
				if ((pctx->filename = realpath(argv[my_optind], NULL)) == NULL) {
					log_msg(LOG_ERR, 1, "%s", argv[my_optind]);
					return (1);
				}
				num_rem--;
				my_optind++;
			}

			if (num_rem > 0) {
				if (*(argv[my_optind]) == '-') {
					pctx->to_filename = "-";
					pctx->pipe_out = 1;
					pctx->to_filename = NULL;
				} else {
					strcpy(apath, argv[my_optind]);
					if (!endswith(apath, COMP_EXTN))
						strcat(apath, COMP_EXTN);
					pctx->to_filename = realpath(apath, NULL);

					/* Check if compressed file exists */
					if (pctx->to_filename != NULL) {
						log_msg(LOG_ERR, 0, "Compressed file %s exists",
						    pctx->to_filename);
						free((void *)(pctx->to_filename));
						return (1);
					}
					pctx->to_filename = argv[my_optind];
				}
			} else {
				strcpy(apath, pctx->filename);
				strcat(apath, COMP_EXTN);
				pctx->to_filename = realpath(apath, NULL);

				/* Check if compressed file exists */
				if (pctx->to_filename != NULL) {
					log_msg(LOG_ERR, 0, "Compressed file %s exists",
					    pctx->to_filename);
					free((void *)(pctx->to_filename));
					return (1);
				}
			}
		} else if (pctx->do_uncompress) {
			/*
			 * While decompressing, input can be stdin and output a physical file.
			 */
			if (*(argv[my_optind]) == '-') {
				pctx->filename = NULL;
			} else {
				if ((pctx->filename = realpath(argv[my_optind], NULL)) == NULL) {
					log_msg(LOG_ERR, 1, "%s", argv[my_optind]);
					return (1);
				}
			}
			if (num_rem == 2) {
				my_optind++;
				pctx->to_filename = argv[my_optind];
			} else {
				pctx->to_filename = NULL;
			}
		} else {
			return (1);
		}
	} else if (num_rem > 2) {
		log_msg(LOG_ERR, 0, "Too many filenames.");
		return (1);
	}
	pctx->main_cancel = 0;

	if (pctx->cksum == 0)
		get_checksum_props(DEFAULT_CKSUM, &(pctx->cksum), &(pctx->cksum_bytes),
				   &(pctx->mac_bytes), 0);

	if ((pctx->enable_rabin_scan || pctx->enable_fixed_scan) && pctx->cksum == CKSUM_CRC64) {
		log_msg(LOG_ERR, 0, "CRC64 checksum is not suitable for Deduplication.");
		return (1);
	}

	if (!pctx->encrypt_type) {
		/*
		 * If not encrypting we compute a header CRC32.
		 */
		pctx->mac_bytes = sizeof (uint32_t); // CRC32 in non-crypto mode
	} else {
		/*
		 * When encrypting we do not compute a normal digest. The HMAC
		 * is computed over header and encrypted data.
		 */
		pctx->cksum_bytes = 0;
	}

	if (pctx->do_compress) {
		struct stat sbuf;

		if (pctx->filename && stat(pctx->filename, &sbuf) == -1) {
			log_msg(LOG_ERR, 1, "Cannot stat: %s", pctx->filename);
			return (1);
		}

		if (pctx->chunksize == 0) {
			if (pctx->level < 9) {
				pctx->chunksize = DEFAULT_CHUNKSIZE;
			} else {
				pctx->chunksize = DEFAULT_CHUNKSIZE + (pctx->level - 8) *
				    DEFAULT_CHUNKSIZE/4;
			}
		}

		if (pctx->archive_mode) {
			if (pctx->meta_stream != -1)
				pctx->meta_stream = 1;
			else
				pctx->meta_stream = 0;
		}

		if (pctx->pipe_mode)
			pctx->meta_stream = 0;

		/*
		 * Auto-select filters and preprocessing modes based on compresion level.
		 * This is not done if user explicitly specified advanced options.
		 */
		if (!pctx->advanced_opts) {
			/*
			 * Selectively enable filters while archiving, depending on compression level.
			 */
			if (pctx->archive_mode) {
				if (pctx->level > 10) {
					ff.enable_packjpg = 1;
					ff.enable_wavpack = 1;
				}
				if (pctx->level > 8) {
					pctx->exe_preprocess = 1;
					ff.exe_preprocess = 1;
				}
				init_filters(&ff);
				pctx->enable_packjpg = ff.enable_packjpg;
				pctx->enable_wavpack = ff.enable_wavpack;
			}

			/*
			 * Enable other preprocessors based on compresion level.
			 */
			if (pctx->level > 4) pctx->enable_delta2_encode = 1;
			if (pctx->level > 9) pctx->lzp_preprocess = 1;
			if (pctx->level > 3) {
				if (pctx->chunksize >= RAB_MIN_CHUNK_SIZE_GLOBAL)
					pctx->enable_rabin_global = 1;
				if (pctx->chunksize >= RAB_MIN_CHUNK_SIZE) {
					pctx->enable_rabin_scan = 1;
					pctx->enable_rabin_split = 1;
				}
				pctx->rab_blk_size = 2;
			}
			if (pctx->level > 9) pctx->delta2_nstrides = NSTRIDES_EXTRA;
		}
		if (pctx->lzp_preprocess || pctx->enable_delta2_encode || pctx->exe_preprocess) {
			pctx->preprocess_mode = 1;
			pctx->enable_analyzer = 1;
		}
	} else if (pctx->do_uncompress) {
		struct filter_flags ff;
		/*
		 * Enable all filters while decompressing. Obviously!
		 */
		ff.enable_packjpg = 1;
		ff.enable_wavpack = 1;
		ff.exe_preprocess = 1;
		pctx->enable_packjpg = 1;
		pctx->enable_wavpack = 1;
		pctx->exe_preprocess = 1;
		init_filters(&ff);
	}
	pctx->inited = 1;

	return (0);
}

int DLL_EXPORT
start_pcompress(pc_ctx_t *pctx)
{
	int err;

	if (!pctx->inited)
		return (1);

	handle_signals();
	err = 0;
	if (pctx->do_compress)
		err = start_compress(pctx, pctx->filename, pctx->chunksize, pctx->level);
	else if (pctx->do_uncompress)
		err = start_decompress(pctx, pctx->filename, pctx->to_filename);
	return (err);
}

/*
 * Setter functions for various parameters in the context.
 */
void DLL_EXPORT
pc_set_userpw(pc_ctx_t *pctx, unsigned char *pwdata, int pwlen)
{
	pctx->user_pw = pwdata;
	pctx->user_pw_len = pwlen;
}