src/lzma/process.c - jrn/xz - Git at Google

 ///////////////////////////////////////////////////////////////////////////////
 //
 /// \file       process.c
 /// \brief      Compresses or uncompresses a file
 //
 //  Copyright (C) 2007 Lasse Collin
 //
 //  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 2.1 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.
 //
 ///////////////////////////////////////////////////////////////////////////////

 #include "private.h"


 typedef struct {
 	lzma_stream strm;
 	void *options;

 	file_pair *pair;

 	/// We don't need this for *anything* but seems that at least with
 	/// glibc pthread_create() doesn't allow NULL.
 	pthread_t thread;

 	bool in_use;

 } thread_data;


 /// Number of available threads
 static size_t free_threads;

 /// Thread-specific data
 static thread_data *threads;

 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

 /// Attributes of new coder threads. They are created in detached state.
 /// Coder threads signal to the service thread themselves when they are done.
 static pthread_attr_t thread_attr;


 //////////
 // Init //
 //////////

 extern void
 process_init(void)
 {
 	threads = malloc(sizeof(thread_data) * opt_threads);
 	if (threads == NULL) {
 		out_of_memory();
 		my_exit(ERROR);
 	}

 	for (size_t i = 0; i < opt_threads; ++i)
 		threads[i] = (thread_data){
 			.strm = LZMA_STREAM_INIT_VAR,
 			.options = NULL,
 			.pair = NULL,
 			.in_use = false,
 		};

 	if (pthread_attr_init(&thread_attr)
 			|| pthread_attr_setdetachstate(
 				&thread_attr, PTHREAD_CREATE_DETACHED)) {
 		out_of_memory();
 		my_exit(ERROR);
 	}

 	free_threads = opt_threads;

 	return;
 }


 //////////////////////////
 // Thread-specific data //
 //////////////////////////

 static thread_data *
 get_thread_data(void)
 {
 	pthread_mutex_lock(&mutex);

 	while (free_threads == 0) {
 		pthread_cond_wait(&cond, &mutex);

 		if (user_abort) {
 			pthread_cond_signal(&cond);
 			pthread_mutex_unlock(&mutex);
 			return NULL;
 		}
 	}

 	thread_data *t = threads;
 	while (t->in_use)
 		++t;

 	t->in_use = true;
 	--free_threads;

 	pthread_mutex_unlock(&mutex);

 	return t;
 }


 static void
 release_thread_data(thread_data *t)
 {
 	pthread_mutex_lock(&mutex);

 	t->in_use = false;
 	++free_threads;

 	pthread_cond_signal(&cond);
 	pthread_mutex_unlock(&mutex);

 	return;
 }


 static int
 create_thread(void *(*func)(thread_data *t), thread_data *t)
 {
 	if (opt_threads == 1) {
 		func(t);
 	} else {
 		const int err = pthread_create(&t->thread, &thread_attr,
 				(void *(*)(void *))(func), t);
 		if (err) {
 			errmsg(V_ERROR, _("Cannot create a thread: %s"),
 					strerror(err));
 			user_abort = 1;
 			return -1;
 		}
 	}

 	return 0;
 }


 /////////////////////////
 // One thread per file //
 /////////////////////////

 static int
 single_init(thread_data *t)
 {
 	lzma_ret ret;

 	if (opt_mode == MODE_COMPRESS) {
 		const lzma_vli uncompressed_size
 				= t->pair->src_fd != STDIN_FILENO
 				? (lzma_vli)(t->pair->src_st.st_size)
 				: LZMA_VLI_VALUE_UNKNOWN;

 		// TODO Support Multi-Block Streams to store Extra.
 		if (opt_header == HEADER_ALONE) {
 			lzma_options_alone alone;
 			alone.uncompressed_size = uncompressed_size;
 			memcpy(&alone.lzma, opt_filters[0].options,
 					sizeof(alone.lzma));
 			ret = lzma_alone_encoder(&t->strm, &alone);
 		} else {
 			lzma_options_stream stream = {
 				.check = opt_check,
 				.has_crc32 = true,
 				.uncompressed_size = uncompressed_size,
 				.alignment = 0,
 			};
 			memcpy(stream.filters, opt_filters,
 					sizeof(stream.filters));
 			ret = lzma_stream_encoder_single(&t->strm, &stream);
 		}
 	} else {
 		// TODO Restrict file format if requested on the command line.
 		ret = lzma_auto_decoder(&t->strm, NULL, NULL);
 	}

 	if (ret != LZMA_OK) {
 		if (ret == LZMA_MEM_ERROR)
 			out_of_memory();
 		else
 			internal_error();

 		return -1;
 	}

 	return 0;
 }


 static lzma_ret
 single_skip_padding(thread_data *t, uint8_t *in_buf)
 {
 	// Handle decoding of concatenated Streams. There can be arbitrary
 	// number of nul-byte padding between the Streams, which must be
 	// ignored.
 	//
 	// NOTE: Concatenating LZMA_Alone files works only if at least
 	// one of lc, lp, and pb is non-zero. Using the concatenation
 	// on LZMA_Alone files is strongly discouraged.
 	while (true) {
 		while (t->strm.avail_in > 0) {
 			if (*t->strm.next_in != '\0')
 				return LZMA_OK;

 			++t->strm.next_in;
 			--t->strm.avail_in;
 		}

 		if (t->pair->src_eof)
 			return LZMA_STREAM_END;

 		t->strm.next_in = in_buf;
 		t->strm.avail_in = io_read(t->pair, in_buf, BUFSIZ);
 		if (t->strm.avail_in == SIZE_MAX)
 			return LZMA_DATA_ERROR;
 	}
 }


 static void *
 single(thread_data *t)
 {
 	if (single_init(t)) {
 		io_close(t->pair, false);
 		release_thread_data(t);
 		return NULL;
 	}

 	uint8_t in_buf[BUFSIZ];
 	uint8_t out_buf[BUFSIZ];
 	lzma_action action = LZMA_RUN;
 	lzma_ret ret;
 	bool success = false;

 	t->strm.avail_in = 0;

 	while (!user_abort) {
 		if (t->strm.avail_in == 0 && !t->pair->src_eof) {
 			t->strm.next_in = in_buf;
 			t->strm.avail_in = io_read(t->pair, in_buf, BUFSIZ);

 			if (t->strm.avail_in == SIZE_MAX)
 				break;
 			else if (t->pair->src_eof
 					&& opt_mode == MODE_COMPRESS)
 				action = LZMA_FINISH;
 		}

 		t->strm.next_out = out_buf;
 		t->strm.avail_out = BUFSIZ;

 		ret = lzma_code(&t->strm, action);

 		if (opt_mode != MODE_TEST)
 			if (io_write(t->pair, out_buf,
 					BUFSIZ - t->strm.avail_out))
 				break;

 		if (ret != LZMA_OK) {
 			if (ret == LZMA_STREAM_END) {
 				if (opt_mode == MODE_COMPRESS) {
 					success = true;
 					break;
 				}

 				// Support decoding concatenated .lzma files.
 				ret = single_skip_padding(t, in_buf);

 				if (ret == LZMA_STREAM_END) {
 					assert(t->pair->src_eof);
 					success = true;
 					break;
 				}

 				if (ret == LZMA_OK && !single_init(t))
 					continue;

 				break;

 			} else {
 				errmsg(V_ERROR, "%s: %s", t->pair->src_name,
 						str_strm_error(ret));
 				break;
 			}
 		}
 	}

 	io_close(t->pair, success);
 	release_thread_data(t);

 	return NULL;
 }


 ///////////////////////////////
 // Multiple threads per file //
 ///////////////////////////////

 // TODO

 // I'm not sure what would the best way to implement this. Here's one
 // possible way:
 //  - Reader thread would read the input data and control the coders threads.
 //  - Every coder thread is associated with input and output buffer pools.
 //    The input buffer pool is filled by reader thread, and the output buffer
 //    pool is emptied by the writer thread.
 //  - Writer thread writes the output data of the oldest living coder thread.
 //
 // The per-file thread started by the application's main thread is used as
 // the reader thread. In the beginning, it starts the writer thread and the
 // first coder thread. The coder thread would be left waiting for input from
 // the reader thread, and the writer thread would be waiting for input from
 // the coder thread.
 //
 // The reader thread reads the input data into a ring buffer, whose size
 // depends on the value returned by lzma_chunk_size(). If the ring buffer
 // gets full, the buffer is marked "to be finished", which indicates to
 // the coder thread that no more input is coming. Then a new coder thread
 // would be started.
 //
 // TODO

 /*
 typedef struct {
 	/// Buffers
 	uint8_t (*buffers)[BUFSIZ];

 	/// Number of buffers
 	size_t buffer_count;

 	/// buffers[read_pos] is the buffer currently being read. Once finish
 	/// is true and read_pos == write_pos, end of input has been reached.
 	size_t read_pos;

 	/// buffers[write_pos] is the buffer into which data is currently
 	/// being written.
 	size_t write_pos;

 	/// This variable matters only when read_pos == write_pos && finish.
 	/// In that case, this variable will contain the size of the
 	/// buffers[read_pos].
 	size_t last_size;

 	/// True once no more data is being written to the buffer. When this
 	/// is set, the last_size variable must have been set too.
 	bool finish;

 	/// Mutex to protect access to the variables in this structure
 	pthread_mutex_t mutex;

 	/// Condition to indicate when another thread can continue
 	pthread_cond_t cond;
 } mem_pool;


 static foo
 multi_reader(thread_data *t)
 {
 	bool done = false;

 	do {
 		const size_t size = io_read(t->pair,
 				m->buffers + m->write_pos, BUFSIZ);
 		if (size == SIZE_MAX) {
 			// TODO
 		} else if (t->pair->src_eof) {
 			m->last_size = size;
 		}

 		pthread_mutex_lock(&m->mutex);

 		if (++m->write_pos == m->buffer_count)
 			m->write_pos = 0;

 		if (m->write_pos == m->read_pos || t->pair->src_eof)
 			m->finish = true;

 		pthread_cond_signal(&m->cond);
 		pthread_mutex_unlock(&m->mutex);

 	} while (!m->finish);

 	return done ? 0 : -1;
 }


 static foo
 multi_code()
 {
 	lzma_action = LZMA_RUN;

 	while (true) {
 		pthread_mutex_lock(&m->mutex);

 		while (m->read_pos == m->write_pos && !m->finish)
 			pthread_cond_wait(&m->cond, &m->mutex);

 		pthread_mutex_unlock(&m->mutex);

 		if (m->finish) {
 			t->strm.avail_in = m->last_size;
 			if (opt_mode == MODE_COMPRESS)
 				action = LZMA_FINISH;
 		} else {
 			t->strm.avail_in = BUFSIZ;
 		}

 		t->strm.next_in = m->buffers + m->read_pos;

 		const lzma_ret ret = lzma_code(&t->strm, action);

 	}
 }

 */


 ///////////////////////
 // Starting new file //
 ///////////////////////

 extern void
 process_file(const char *filename)
 {
 	thread_data *t = get_thread_data();
 	if (t == NULL)
 		return; // User abort

 	// If this fails, it shows appropriate error messages too.
 	t->pair = io_open(filename);
 	if (t->pair == NULL) {
 		release_thread_data(t);
 		return;
 	}

 	// TODO Currently only one-thread-per-file mode is implemented.

 	if (create_thread(&single, t)) {
 		io_close(t->pair, false);
 		release_thread_data(t);
 	}

 	return;
 }
	///////////////////////////////////////////////////////////////////////////////
	//
	/// \file process.c
	/// \brief Compresses or uncompresses a file
	//
	// Copyright (C) 2007 Lasse Collin
	//
	// 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 2.1 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.
	//
	///////////////////////////////////////////////////////////////////////////////

	#include "private.h"


	typedef struct {
	lzma_stream strm;
	void *options;

	file_pair *pair;

	/// We don't need this for anything but seems that at least with
	/// glibc pthread_create() doesn't allow NULL.
	pthread_t thread;

	bool in_use;

	} thread_data;


	/// Number of available threads
	static size_t free_threads;

	/// Thread-specific data
	static thread_data *threads;

	static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
	static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

	/// Attributes of new coder threads. They are created in detached state.
	/// Coder threads signal to the service thread themselves when they are done.
	static pthread_attr_t thread_attr;


	//////////
	// Init //
	//////////

	extern void
	process_init(void)
	{
	threads = malloc(sizeof(thread_data) * opt_threads);
	if (threads == NULL) {
	out_of_memory();
	my_exit(ERROR);
	}

	for (size_t i = 0; i < opt_threads; ++i)
	threads[i] = (thread_data){
	.strm = LZMA_STREAM_INIT_VAR,
	.options = NULL,
	.pair = NULL,
	.in_use = false,
	};

	if (pthread_attr_init(&thread_attr)
	\|\| pthread_attr_setdetachstate(
	&thread_attr, PTHREAD_CREATE_DETACHED)) {
	out_of_memory();
	my_exit(ERROR);
	}

	free_threads = opt_threads;

	return;
	}


	//////////////////////////
	// Thread-specific data //
	//////////////////////////

	static thread_data *
	get_thread_data(void)
	{
	pthread_mutex_lock(&mutex);

	while (free_threads == 0) {
	pthread_cond_wait(&cond, &mutex);

	if (user_abort) {
	pthread_cond_signal(&cond);
	pthread_mutex_unlock(&mutex);
	return NULL;
	}
	}

	thread_data *t = threads;
	while (t->in_use)
	++t;

	t->in_use = true;
	--free_threads;

	pthread_mutex_unlock(&mutex);

	return t;
	}


	static void
	release_thread_data(thread_data *t)
	{
	pthread_mutex_lock(&mutex);

	t->in_use = false;
	++free_threads;

	pthread_cond_signal(&cond);
	pthread_mutex_unlock(&mutex);

	return;
	}


	static int
	create_thread(void (func)(thread_data t), thread_data t)
	{
	if (opt_threads == 1) {
	func(t);
	} else {
	const int err = pthread_create(&t->thread, &thread_attr,
	(void ()(void *))(func), t);
	if (err) {
	errmsg(V_ERROR, _("Cannot create a thread: %s"),
	strerror(err));
	user_abort = 1;
	return -1;
	}
	}

	return 0;
	}


	/////////////////////////
	// One thread per file //
	/////////////////////////

	static int
	single_init(thread_data *t)
	{
	lzma_ret ret;

	if (opt_mode == MODE_COMPRESS) {
	const lzma_vli uncompressed_size
	= t->pair->src_fd != STDIN_FILENO
	? (lzma_vli)(t->pair->src_st.st_size)
	: LZMA_VLI_VALUE_UNKNOWN;

	// TODO Support Multi-Block Streams to store Extra.
	if (opt_header == HEADER_ALONE) {
	lzma_options_alone alone;
	alone.uncompressed_size = uncompressed_size;
	memcpy(&alone.lzma, opt_filters[0].options,
	sizeof(alone.lzma));
	ret = lzma_alone_encoder(&t->strm, &alone);
	} else {
	lzma_options_stream stream = {
	.check = opt_check,
	.has_crc32 = true,
	.uncompressed_size = uncompressed_size,
	.alignment = 0,
	};
	memcpy(stream.filters, opt_filters,
	sizeof(stream.filters));
	ret = lzma_stream_encoder_single(&t->strm, &stream);
	}
	} else {
	// TODO Restrict file format if requested on the command line.
	ret = lzma_auto_decoder(&t->strm, NULL, NULL);
	}

	if (ret != LZMA_OK) {
	if (ret == LZMA_MEM_ERROR)
	out_of_memory();
	else
	internal_error();

	return -1;
	}

	return 0;
	}


	static lzma_ret
	single_skip_padding(thread_data t, uint8_t in_buf)
	{
	// Handle decoding of concatenated Streams. There can be arbitrary
	// number of nul-byte padding between the Streams, which must be
	// ignored.
	//
	// NOTE: Concatenating LZMA_Alone files works only if at least
	// one of lc, lp, and pb is non-zero. Using the concatenation
	// on LZMA_Alone files is strongly discouraged.
	while (true) {
	while (t->strm.avail_in > 0) {
	if (*t->strm.next_in != '\0')
	return LZMA_OK;

	++t->strm.next_in;
	--t->strm.avail_in;
	}

	if (t->pair->src_eof)
	return LZMA_STREAM_END;

	t->strm.next_in = in_buf;
	t->strm.avail_in = io_read(t->pair, in_buf, BUFSIZ);
	if (t->strm.avail_in == SIZE_MAX)
	return LZMA_DATA_ERROR;
	}
	}


	static void *
	single(thread_data *t)
	{
	if (single_init(t)) {
	io_close(t->pair, false);
	release_thread_data(t);
	return NULL;
	}

	uint8_t in_buf[BUFSIZ];
	uint8_t out_buf[BUFSIZ];
	lzma_action action = LZMA_RUN;
	lzma_ret ret;
	bool success = false;

	t->strm.avail_in = 0;

	while (!user_abort) {
	if (t->strm.avail_in == 0 && !t->pair->src_eof) {
	t->strm.next_in = in_buf;
	t->strm.avail_in = io_read(t->pair, in_buf, BUFSIZ);

	if (t->strm.avail_in == SIZE_MAX)
	break;
	else if (t->pair->src_eof
	&& opt_mode == MODE_COMPRESS)
	action = LZMA_FINISH;
	}

	t->strm.next_out = out_buf;
	t->strm.avail_out = BUFSIZ;

	ret = lzma_code(&t->strm, action);

	if (opt_mode != MODE_TEST)
	if (io_write(t->pair, out_buf,
	BUFSIZ - t->strm.avail_out))
	break;

	if (ret != LZMA_OK) {
	if (ret == LZMA_STREAM_END) {
	if (opt_mode == MODE_COMPRESS) {
	success = true;
	break;
	}

	// Support decoding concatenated .lzma files.
	ret = single_skip_padding(t, in_buf);

	if (ret == LZMA_STREAM_END) {
	assert(t->pair->src_eof);
	success = true;
	break;
	}

	if (ret == LZMA_OK && !single_init(t))
	continue;

	break;

	} else {
	errmsg(V_ERROR, "%s: %s", t->pair->src_name,
	str_strm_error(ret));
	break;
	}
	}
	}

	io_close(t->pair, success);
	release_thread_data(t);

	return NULL;
	}


	///////////////////////////////
	// Multiple threads per file //
	///////////////////////////////

	// TODO

	// I'm not sure what would the best way to implement this. Here's one
	// possible way:
	// - Reader thread would read the input data and control the coders threads.
	// - Every coder thread is associated with input and output buffer pools.
	// The input buffer pool is filled by reader thread, and the output buffer
	// pool is emptied by the writer thread.
	// - Writer thread writes the output data of the oldest living coder thread.
	//
	// The per-file thread started by the application's main thread is used as
	// the reader thread. In the beginning, it starts the writer thread and the
	// first coder thread. The coder thread would be left waiting for input from
	// the reader thread, and the writer thread would be waiting for input from
	// the coder thread.
	//
	// The reader thread reads the input data into a ring buffer, whose size
	// depends on the value returned by lzma_chunk_size(). If the ring buffer
	// gets full, the buffer is marked "to be finished", which indicates to
	// the coder thread that no more input is coming. Then a new coder thread
	// would be started.
	//
	// TODO

	/*
	typedef struct {
	/// Buffers
	uint8_t (*buffers)[BUFSIZ];

	/// Number of buffers
	size_t buffer_count;

	/// buffers[read_pos] is the buffer currently being read. Once finish
	/// is true and read_pos == write_pos, end of input has been reached.
	size_t read_pos;

	/// buffers[write_pos] is the buffer into which data is currently
	/// being written.
	size_t write_pos;

	/// This variable matters only when read_pos == write_pos && finish.
	/// In that case, this variable will contain the size of the
	/// buffers[read_pos].
	size_t last_size;

	/// True once no more data is being written to the buffer. When this
	/// is set, the last_size variable must have been set too.
	bool finish;

	/// Mutex to protect access to the variables in this structure
	pthread_mutex_t mutex;

	/// Condition to indicate when another thread can continue
	pthread_cond_t cond;
	} mem_pool;


	static foo
	multi_reader(thread_data *t)
	{
	bool done = false;

	do {
	const size_t size = io_read(t->pair,
	m->buffers + m->write_pos, BUFSIZ);
	if (size == SIZE_MAX) {
	// TODO
	} else if (t->pair->src_eof) {
	m->last_size = size;
	}

	pthread_mutex_lock(&m->mutex);

	if (++m->write_pos == m->buffer_count)
	m->write_pos = 0;

	if (m->write_pos == m->read_pos \|\| t->pair->src_eof)
	m->finish = true;

	pthread_cond_signal(&m->cond);
	pthread_mutex_unlock(&m->mutex);

	} while (!m->finish);

	return done ? 0 : -1;
	}


	static foo
	multi_code()
	{
	lzma_action = LZMA_RUN;

	while (true) {
	pthread_mutex_lock(&m->mutex);

	while (m->read_pos == m->write_pos && !m->finish)
	pthread_cond_wait(&m->cond, &m->mutex);

	pthread_mutex_unlock(&m->mutex);

	if (m->finish) {
	t->strm.avail_in = m->last_size;
	if (opt_mode == MODE_COMPRESS)
	action = LZMA_FINISH;
	} else {
	t->strm.avail_in = BUFSIZ;
	}

	t->strm.next_in = m->buffers + m->read_pos;

	const lzma_ret ret = lzma_code(&t->strm, action);

	}
	}

	*/


	///////////////////////
	// Starting new file //
	///////////////////////

	extern void
	process_file(const char *filename)
	{
	thread_data *t = get_thread_data();
	if (t == NULL)
	return; // User abort

	// If this fails, it shows appropriate error messages too.
	t->pair = io_open(filename);
	if (t->pair == NULL) {
	release_thread_data(t);
	return;
	}

	// TODO Currently only one-thread-per-file mode is implemented.

	if (create_thread(&single, t)) {
	io_close(t->pair, false);
	release_thread_data(t);
	}

	return;
	}