outqueue.c

// SPDX-License-Identifier: 0BSD

///////////////////////////////////////////////////////////////////////////////
//
/// \file       outqueue.c
/// \brief      Output queue handling in multithreaded coding
//
//  Author:     Lasse Collin
//
///////////////////////////////////////////////////////////////////////////////

#include "outqueue.h"


/// Get the maximum number of buffers that may be allocated based
/// on the number of threads. For now this is twice the number of threads.
/// It's a compromise between RAM usage and keeping the worker threads busy
/// when buffers finish out of order.
#define GET_BUFS_LIMIT(threads) (2 * (threads))


extern uint64_t
lzma_outq_memusage(uint64_t buf_size_max, uint32_t threads)
{
	// This is to ease integer overflow checking: We may allocate up to
	// GET_BUFS_LIMIT(LZMA_THREADS_MAX) buffers and we need some extra
	// memory for other data structures too (that's the /2).
	//
	// lzma_outq_prealloc_buf() will still accept bigger buffers than this.
	const uint64_t limit
			= UINT64_MAX / GET_BUFS_LIMIT(LZMA_THREADS_MAX) / 2;

	if (threads > LZMA_THREADS_MAX || buf_size_max > limit)
		return UINT64_MAX;

	return GET_BUFS_LIMIT(threads)
			* lzma_outq_outbuf_memusage(buf_size_max);
}


static void
move_head_to_cache(lzma_outq *outq, const lzma_allocator *allocator)
{
	assert(outq->head != NULL);
	assert(outq->tail != NULL);
	assert(outq->bufs_in_use > 0);

	lzma_outbuf *buf = outq->head;
	outq->head = buf->next;
	if (outq->head == NULL)
		outq->tail = NULL;

	if (outq->cache != NULL && outq->cache->allocated != buf->allocated)
		lzma_outq_clear_cache(outq, allocator);

	buf->next = outq->cache;
	outq->cache = buf;

	--outq->bufs_in_use;
	outq->mem_in_use -= lzma_outq_outbuf_memusage(buf->allocated);

	return;
}


static void
free_one_cached_buffer(lzma_outq *outq, const lzma_allocator *allocator)
{
	assert(outq->cache != NULL);

	lzma_outbuf *buf = outq->cache;
	outq->cache = buf->next;

	--outq->bufs_allocated;
	outq->mem_allocated -= lzma_outq_outbuf_memusage(buf->allocated);

	lzma_free(buf, allocator);
	return;
}


extern void
lzma_outq_clear_cache(lzma_outq *outq, const lzma_allocator *allocator)
{
	while (outq->cache != NULL)
		free_one_cached_buffer(outq, allocator);

	return;
}


extern void
lzma_outq_clear_cache2(lzma_outq *outq, const lzma_allocator *allocator,
		size_t keep_size)
{
	if (outq->cache == NULL)
		return;

	// Free all but one.
	while (outq->cache->next != NULL)
		free_one_cached_buffer(outq, allocator);

	// Free the last one only if its size doesn't equal to keep_size.
	if (outq->cache->allocated != keep_size)
		free_one_cached_buffer(outq, allocator);

	return;
}


extern lzma_ret
lzma_outq_init(lzma_outq *outq, const lzma_allocator *allocator,
		uint32_t threads)
{
	if (threads > LZMA_THREADS_MAX)
		return LZMA_OPTIONS_ERROR;

	const uint32_t bufs_limit = GET_BUFS_LIMIT(threads);

	// Clear head/tail.
	while (outq->head != NULL)
		move_head_to_cache(outq, allocator);

	// If new buf_limit is lower than the old one, we may need to free
	// a few cached buffers.
	while (bufs_limit < outq->bufs_allocated)
		free_one_cached_buffer(outq, allocator);

	outq->bufs_limit = bufs_limit;
	outq->read_pos = 0;

	return LZMA_OK;
}


extern void
lzma_outq_end(lzma_outq *outq, const lzma_allocator *allocator)
{
	while (outq->head != NULL)
		move_head_to_cache(outq, allocator);

	lzma_outq_clear_cache(outq, allocator);
	return;
}


extern lzma_ret
lzma_outq_prealloc_buf(lzma_outq *outq, const lzma_allocator *allocator,
		size_t size)
{
	// Caller must have checked it with lzma_outq_has_buf().
	assert(outq->bufs_in_use < outq->bufs_limit);

	// If there already is appropriately-sized buffer in the cache,
	// we need to do nothing.
	if (outq->cache != NULL && outq->cache->allocated == size)
		return LZMA_OK;

	if (size > SIZE_MAX - sizeof(lzma_outbuf))
		return LZMA_MEM_ERROR;

	const size_t alloc_size = lzma_outq_outbuf_memusage(size);

	// The cache may have buffers but their size is wrong.
	lzma_outq_clear_cache(outq, allocator);

	outq->cache = lzma_alloc(alloc_size, allocator);
	if (outq->cache == NULL)
		return LZMA_MEM_ERROR;

	outq->cache->next = NULL;
	outq->cache->allocated = size;

	++outq->bufs_allocated;
	outq->mem_allocated += alloc_size;

	return LZMA_OK;
}


extern lzma_outbuf *
lzma_outq_get_buf(lzma_outq *outq, void *worker)
{
	// Caller must have used lzma_outq_prealloc_buf() to ensure these.
	assert(outq->bufs_in_use < outq->bufs_limit);
	assert(outq->bufs_in_use < outq->bufs_allocated);
	assert(outq->cache != NULL);

	lzma_outbuf *buf = outq->cache;
	outq->cache = buf->next;
	buf->next = NULL;

	if (outq->tail != NULL) {
		assert(outq->head != NULL);
		outq->tail->next = buf;
	} else {
		assert(outq->head == NULL);
		outq->head = buf;
	}

	outq->tail = buf;

	buf->worker = worker;
	buf->finished = false;
	buf->finish_ret = LZMA_STREAM_END;
	buf->pos = 0;
	buf->decoder_in_pos = 0;

	buf->unpadded_size = 0;
	buf->uncompressed_size = 0;

	++outq->bufs_in_use;
	outq->mem_in_use += lzma_outq_outbuf_memusage(buf->allocated);

	return buf;
}


extern bool
lzma_outq_is_readable(const lzma_outq *outq)
{
	if (outq->head == NULL)
		return false;

	return outq->read_pos < outq->head->pos || outq->head->finished;
}


extern lzma_ret
lzma_outq_read(lzma_outq *restrict outq,
		const lzma_allocator *restrict allocator,
		uint8_t *restrict out, size_t *restrict out_pos,
		size_t out_size,
		lzma_vli *restrict unpadded_size,
		lzma_vli *restrict uncompressed_size)
{
	// There must be at least one buffer from which to read.
	if (outq->bufs_in_use == 0)
		return LZMA_OK;

	// Get the buffer.
	lzma_outbuf *buf = outq->head;

	// Copy from the buffer to output.
	//
	// FIXME? In threaded decoder it may be bad to do this copy while
	// the mutex is being held.
	lzma_bufcpy(buf->buf, &outq->read_pos, buf->pos,
			out, out_pos, out_size);

	// Return if we didn't get all the data from the buffer.
	if (!buf->finished || outq->read_pos < buf->pos)
		return LZMA_OK;

	// The buffer was finished. Tell the caller its size information.
	if (unpadded_size != NULL)
		*unpadded_size = buf->unpadded_size;

	if (uncompressed_size != NULL)
		*uncompressed_size = buf->uncompressed_size;

	// Remember the return value.
	const lzma_ret finish_ret = buf->finish_ret;

	// Free this buffer for further use.
	move_head_to_cache(outq, allocator);
	outq->read_pos = 0;

	return finish_ret;
}


extern void
lzma_outq_enable_partial_output(lzma_outq *outq,
		void (*enable_partial_output)(void *worker))
{
	if (outq->head != NULL && !outq->head->finished
			&& outq->head->worker != NULL) {
		enable_partial_output(outq->head->worker);

		// Set it to NULL since calling it twice is pointless.
		outq->head->worker = NULL;
	}

	return;
}