Tests: Fix memory leaks in test_block_header.

[xz/debian.git] / src / liblzma / common / stream_decoder_mt.c

///////////////////////////////////////////////////////////////////////////////
//
/// \file       stream_decoder_mt.c
/// \brief      Multithreaded .xz Stream decoder
//
//  Authors:    Sebastian Andrzej Siewior
//              Lasse Collin
//
//  This file has been put into the public domain.
//  You can do whatever you want with this file.
//
///////////////////////////////////////////////////////////////////////////////

#include "common.h"
#include "block_decoder.h"
#include "stream_decoder.h"
#include "index.h"
#include "outqueue.h"


typedef enum {
        /// Waiting for work.
        /// Main thread may change this to THR_RUN or THR_EXIT.
        THR_IDLE,

        /// Decoding is in progress.
        /// Main thread may change this to THR_STOP or THR_EXIT.
        /// The worker thread may change this to THR_IDLE.
        THR_RUN,

        /// The main thread wants the thread to stop whatever it was doing
        /// but not exit. Main thread may change this to THR_EXIT.
        /// The worker thread may change this to THR_IDLE.
        THR_STOP,

        /// The main thread wants the thread to exit.
        THR_EXIT,

} worker_state;


typedef enum {
        /// Partial updates (storing of worker thread progress
        /// to lzma_outbuf) are disabled.
        PARTIAL_DISABLED,

        /// Main thread requests partial updates to be enabled but
        /// no partial update has been done by the worker thread yet.
        ///
        /// Changing from PARTIAL_DISABLED to PARTIAL_START requires
        /// use of the worker-thread mutex. Other transitions don't
        /// need a mutex.
        PARTIAL_START,

        /// Partial updates are enabled and the worker thread has done
        /// at least one partial update.
        PARTIAL_ENABLED,

} partial_update_mode;


struct worker_thread {
        /// Worker state is protected with our mutex.
        worker_state state;

        /// Input buffer that will contain the whole Block except Block Header.
        uint8_t *in;

        /// Amount of memory allocated for "in"
        size_t in_size;

        /// Number of bytes written to "in" by the main thread
        size_t in_filled;

        /// Number of bytes consumed from "in" by the worker thread.
        size_t in_pos;

        /// Amount of uncompressed data that has been decoded. This local
        /// copy is needed because updating outbuf->pos requires locking
        /// the main mutex (coder->mutex).
        size_t out_pos;

        /// Pointer to the main structure is needed to (1) lock the main
        /// mutex (coder->mutex) when updating outbuf->pos and (2) when
        /// putting this thread back to the stack of free threads.
        struct lzma_stream_coder *coder;

        /// The allocator is set by the main thread. Since a copy of the
        /// pointer is kept here, the application must not change the
        /// allocator before calling lzma_end().
        const lzma_allocator *allocator;

        /// Output queue buffer to which the uncompressed data is written.
        lzma_outbuf *outbuf;

        /// Amount of compressed data that has already been decompressed.
        /// This is updated from in_pos when our mutex is locked.
        /// This is size_t, not uint64_t, because per-thread progress
        /// is limited to sizes of allocated buffers.
        size_t progress_in;

        /// Like progress_in but for uncompressed data.
        size_t progress_out;

        /// Updating outbuf->pos requires locking the main mutex
        /// (coder->mutex). Since the main thread will only read output
        /// from the oldest outbuf in the queue, only the worker thread
        /// that is associated with the oldest outbuf needs to update its
        /// outbuf->pos. This avoids useless mutex contention that would
        /// happen if all worker threads were frequently locking the main
        /// mutex to update their outbuf->pos.
        ///
        /// Only when partial_update is something else than PARTIAL_DISABLED,
        /// this worker thread will update outbuf->pos after each call to
        /// the Block decoder.
        partial_update_mode partial_update;

        /// Block decoder
        lzma_next_coder block_decoder;

        /// Thread-specific Block options are needed because the Block
        /// decoder modifies the struct given to it at initialization.
        lzma_block block_options;

        /// Filter chain memory usage
        uint64_t mem_filters;

        /// Next structure in the stack of free worker threads.
        struct worker_thread *next;

        mythread_mutex mutex;
        mythread_cond cond;

        /// The ID of this thread is used to join the thread
        /// when it's not needed anymore.
        mythread thread_id;
};


struct lzma_stream_coder {
        enum {
                SEQ_STREAM_HEADER,
                SEQ_BLOCK_HEADER,
                SEQ_BLOCK_INIT,
                SEQ_BLOCK_THR_INIT,
                SEQ_BLOCK_THR_RUN,
                SEQ_BLOCK_DIRECT_INIT,
                SEQ_BLOCK_DIRECT_RUN,
                SEQ_INDEX_WAIT_OUTPUT,
                SEQ_INDEX_DECODE,
                SEQ_STREAM_FOOTER,
                SEQ_STREAM_PADDING,
                SEQ_ERROR,
        } sequence;

        /// Block decoder
        lzma_next_coder block_decoder;

        /// Every Block Header will be decoded into this structure.
        /// This is also used to initialize a Block decoder when in
        /// direct mode. In threaded mode, a thread-specific copy will
        /// be made for decoder initialization because the Block decoder
        /// will modify the structure given to it.
        lzma_block block_options;

        /// Buffer to hold a filter chain for Block Header decoding and
        /// initialization. These are freed after successful Block decoder
        /// initialization or at stream_decoder_mt_end(). The thread-specific
        /// copy of block_options won't hold a pointer to filters[] after
        /// initialization.
        lzma_filter filters[LZMA_FILTERS_MAX + 1];

        /// Stream Flags from Stream Header
        lzma_stream_flags stream_flags;

        /// Index is hashed so that it can be compared to the sizes of Blocks
        /// with O(1) memory usage.
        lzma_index_hash *index_hash;


        /// Maximum wait time if cannot use all the input and cannot
        /// fill the output buffer. This is in milliseconds.
        uint32_t timeout;


        /// Error code from a worker thread.
        ///
        /// \note       Use mutex.
        lzma_ret thread_error;

        /// Error code to return after pending output has been copied out. If
        /// set in read_output_and_wait(), this is a mirror of thread_error.
        /// If set in stream_decode_mt() then it's, for example, error that
        /// occurred when decoding Block Header.
        lzma_ret pending_error;

        /// Number of threads that will be created at maximum.
        uint32_t threads_max;

        /// Number of thread structures that have been initialized from
        /// "threads", and thus the number of worker threads actually
        /// created so far.
        uint32_t threads_initialized;

        /// Array of allocated thread-specific structures. When no threads
        /// are in use (direct mode) this is NULL. In threaded mode this
        /// points to an array of threads_max number of worker_thread structs.
        struct worker_thread *threads;

        /// Stack of free threads. When a thread finishes, it puts itself
        /// back into this stack. This starts as empty because threads
        /// are created only when actually needed.
        ///
        /// \note       Use mutex.
        struct worker_thread *threads_free;

        /// The most recent worker thread to which the main thread writes
        /// the new input from the application.
        struct worker_thread *thr;

        /// Output buffer queue for decompressed data from the worker threads
        ///
        /// \note       Use mutex with operations that need it.
        lzma_outq outq;

        mythread_mutex mutex;
        mythread_cond cond;


        /// Memory usage that will not be exceeded in multi-threaded mode.
        /// Single-threaded mode can exceed this even by a large amount.
        uint64_t memlimit_threading;

        /// Memory usage limit that should never be exceeded.
        /// LZMA_MEMLIMIT_ERROR will be returned if decoding isn't possible
        /// even in single-threaded mode without exceeding this limit.
        uint64_t memlimit_stop;

        /// Amount of memory in use by the direct mode decoder
        /// (coder->block_decoder). In threaded mode this is 0.
        uint64_t mem_direct_mode;

        /// Amount of memory needed by the running worker threads.
        /// This doesn't include the memory needed by the output buffer.
        ///
        /// \note       Use mutex.
        uint64_t mem_in_use;

        /// Amount of memory used by the idle (cached) threads.
        ///
        /// \note       Use mutex.
        uint64_t mem_cached;


        /// Amount of memory needed for the filter chain of the next Block.
        uint64_t mem_next_filters;

        /// Amount of memory needed for the thread-specific input buffer
        /// for the next Block.
        uint64_t mem_next_in;

        /// Amount of memory actually needed to decode the next Block
        /// in threaded mode. This is
        /// mem_next_filters + mem_next_in + memory needed for lzma_outbuf.
        uint64_t mem_next_block;


        /// Amount of compressed data in Stream Header + Blocks that have
        /// already been finished.
        ///
        /// \note       Use mutex.
        uint64_t progress_in;

        /// Amount of uncompressed data in Blocks that have already
        /// been finished.
        ///
        /// \note       Use mutex.
        uint64_t progress_out;


        /// If true, LZMA_NO_CHECK is returned if the Stream has
        /// no integrity check.
        bool tell_no_check;

        /// If true, LZMA_UNSUPPORTED_CHECK is returned if the Stream has
        /// an integrity check that isn't supported by this liblzma build.
        bool tell_unsupported_check;

        /// If true, LZMA_GET_CHECK is returned after decoding Stream Header.
        bool tell_any_check;

        /// If true, we will tell the Block decoder to skip calculating
        /// and verifying the integrity check.
        bool ignore_check;

        /// If true, we will decode concatenated Streams that possibly have
        /// Stream Padding between or after them. LZMA_STREAM_END is returned
        /// once the application isn't giving us any new input (LZMA_FINISH),
        /// and we aren't in the middle of a Stream, and possible
        /// Stream Padding is a multiple of four bytes.
        bool concatenated;

        /// If true, we will return any errors immediately instead of first
        /// producing all output before the location of the error.
        bool fail_fast;


        /// When decoding concatenated Streams, this is true as long as we
        /// are decoding the first Stream. This is needed to avoid misleading
        /// LZMA_FORMAT_ERROR in case the later Streams don't have valid magic
        /// bytes.
        bool first_stream;

        /// This is used to track if the previous call to stream_decode_mt()
        /// had output space (*out_pos < out_size) and managed to fill the
        /// output buffer (*out_pos == out_size). This may be set to true
        /// in read_output_and_wait(). This is read and then reset to false
        /// at the beginning of stream_decode_mt().
        ///
        /// This is needed to support applications that call lzma_code() in
        /// such a way that more input is provided only when lzma_code()
        /// didn't fill the output buffer completely. Basically, this makes
        /// it easier to convert such applications from single-threaded
        /// decoder to multi-threaded decoder.
        bool out_was_filled;

        /// Write position in buffer[] and position in Stream Padding
        size_t pos;

        /// Buffer to hold Stream Header, Block Header, and Stream Footer.
        /// Block Header has biggest maximum size.
        uint8_t buffer[LZMA_BLOCK_HEADER_SIZE_MAX];
};


/// Enables updating of outbuf->pos. This is a callback function that is
/// used with lzma_outq_enable_partial_output().
static void
worker_enable_partial_update(void *thr_ptr)
{
        struct worker_thread *thr = thr_ptr;

        mythread_sync(thr->mutex) {
                thr->partial_update = PARTIAL_START;
                mythread_cond_signal(&thr->cond);
        }
}


/// Things do to at THR_STOP or when finishing a Block.
/// This is called with thr->mutex locked.
static void
worker_stop(struct worker_thread *thr)
{
        // Update memory usage counters.
        thr->coder->mem_in_use -= thr->in_size;
        thr->in_size = 0; // thr->in was freed above.

        thr->coder->mem_in_use -= thr->mem_filters;
        thr->coder->mem_cached += thr->mem_filters;

        // Put this thread to the stack of free threads.
        thr->next = thr->coder->threads_free;
        thr->coder->threads_free = thr;

        mythread_cond_signal(&thr->coder->cond);
        return;
}


static MYTHREAD_RET_TYPE
worker_decoder(void *thr_ptr)
{
        struct worker_thread *thr = thr_ptr;
        size_t in_filled;
        partial_update_mode partial_update;
        lzma_ret ret;

next_loop_lock:

        mythread_mutex_lock(&thr->mutex);
next_loop_unlocked:

        if (thr->state == THR_IDLE) {
                mythread_cond_wait(&thr->cond, &thr->mutex);
                goto next_loop_unlocked;
        }

        if (thr->state == THR_EXIT) {
                mythread_mutex_unlock(&thr->mutex);

                lzma_free(thr->in, thr->allocator);
                lzma_next_end(&thr->block_decoder, thr->allocator);

                mythread_mutex_destroy(&thr->mutex);
                mythread_cond_destroy(&thr->cond);

                return MYTHREAD_RET_VALUE;
        }

        if (thr->state == THR_STOP) {
                thr->state = THR_IDLE;
                mythread_mutex_unlock(&thr->mutex);

                mythread_sync(thr->coder->mutex) {
                        worker_stop(thr);
                }

                goto next_loop_lock;
        }

        assert(thr->state == THR_RUN);

        // Update progress info for get_progress().
        thr->progress_in = thr->in_pos;
        thr->progress_out = thr->out_pos;

        // If we don't have any new input, wait for a signal from the main
        // thread except if partial output has just been enabled. In that
        // case we will do one normal run so that the partial output info
        // gets passed to the main thread. The call to block_decoder.code()
        // is useless but harmless as it can occur only once per Block.
        in_filled = thr->in_filled;
        partial_update = thr->partial_update;

        if (in_filled == thr->in_pos && partial_update != PARTIAL_START) {
                mythread_cond_wait(&thr->cond, &thr->mutex);
                goto next_loop_unlocked;
        }

        mythread_mutex_unlock(&thr->mutex);

        // Pass the input in small chunks to the Block decoder.
        // This way we react reasonably fast if we are told to stop/exit,
        // and (when partial update is enabled) we tell about our progress
        // to the main thread frequently enough.
        const size_t chunk_size = 16384;
        if ((in_filled - thr->in_pos) > chunk_size)
                in_filled = thr->in_pos + chunk_size;

        ret = thr->block_decoder.code(
                        thr->block_decoder.coder, thr->allocator,
                        thr->in, &thr->in_pos, in_filled,
                        thr->outbuf->buf, &thr->out_pos,
                        thr->outbuf->allocated, LZMA_RUN);

        if (ret == LZMA_OK) {
                if (partial_update != PARTIAL_DISABLED) {
                        // The main thread uses thr->mutex to change from
                        // PARTIAL_DISABLED to PARTIAL_START. The main thread
                        // doesn't care about this variable after that so we
                        // can safely change it here to PARTIAL_ENABLED
                        // without a mutex.
                        thr->partial_update = PARTIAL_ENABLED;

                        // The main thread is reading decompressed data
                        // from thr->outbuf. Tell the main thread about
                        // our progress.
                        //
                        // NOTE: It's possible that we consumed input without
                        // producing any new output so it's possible that
                        // only in_pos has changed. In case of PARTIAL_START
                        // it is possible that neither in_pos nor out_pos has
                        // changed.
                        mythread_sync(thr->coder->mutex) {
                                thr->outbuf->pos = thr->out_pos;
                                thr->outbuf->decoder_in_pos = thr->in_pos;
                                mythread_cond_signal(&thr->coder->cond);
                        }
                }

                goto next_loop_lock;
        }

        // Either we finished successfully (LZMA_STREAM_END) or an error
        // occurred. Both cases are handled almost identically. The error
        // case requires updating thr->coder->thread_error.
        //
        // The sizes are in the Block Header and the Block decoder
        // checks that they match, thus we know these:
        assert(ret != LZMA_STREAM_END || thr->in_pos == thr->in_size);
        assert(ret != LZMA_STREAM_END
                || thr->out_pos == thr->block_options.uncompressed_size);

        // Free the input buffer. Don't update in_size as we need
        // it later to update thr->coder->mem_in_use.
        lzma_free(thr->in, thr->allocator);
        thr->in = NULL;

        mythread_sync(thr->mutex) {
                if (thr->state != THR_EXIT)
                        thr->state = THR_IDLE;
        }

        mythread_sync(thr->coder->mutex) {
                // Move our progress info to the main thread.
                thr->coder->progress_in += thr->in_pos;
                thr->coder->progress_out += thr->out_pos;
                thr->progress_in = 0;
                thr->progress_out = 0;

                // Mark the outbuf as finished.
                thr->outbuf->pos = thr->out_pos;
                thr->outbuf->decoder_in_pos = thr->in_pos;
                thr->outbuf->finished = true;
                thr->outbuf->finish_ret = ret;
                thr->outbuf = NULL;

                // If an error occurred, tell it to the main thread.
                if (ret != LZMA_STREAM_END
                                && thr->coder->thread_error == LZMA_OK)
                        thr->coder->thread_error = ret;

                worker_stop(thr);
        }

        goto next_loop_lock;
}


/// Tells the worker threads to exit and waits for them to terminate.
static void
threads_end(struct lzma_stream_coder *coder, const lzma_allocator *allocator)
{
        for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
                mythread_sync(coder->threads[i].mutex) {
                        coder->threads[i].state = THR_EXIT;
                        mythread_cond_signal(&coder->threads[i].cond);
                }
        }

        for (uint32_t i = 0; i < coder->threads_initialized; ++i)
                mythread_join(coder->threads[i].thread_id);

        lzma_free(coder->threads, allocator);
        coder->threads_initialized = 0;
        coder->threads = NULL;
        coder->threads_free = NULL;

        // The threads don't update these when they exit. Do it here.
        coder->mem_in_use = 0;
        coder->mem_cached = 0;

        return;
}


static void
threads_stop(struct lzma_stream_coder *coder)
{
        for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
                mythread_sync(coder->threads[i].mutex) {
                        // The state must be changed conditionally because
                        // THR_IDLE -> THR_STOP is not a valid state change.
                        if (coder->threads[i].state != THR_IDLE) {
                                coder->threads[i].state = THR_STOP;
                                mythread_cond_signal(&coder->threads[i].cond);
                        }
                }
        }

        return;
}


/// Initialize a new worker_thread structure and create a new thread.
static lzma_ret
initialize_new_thread(struct lzma_stream_coder *coder,
                const lzma_allocator *allocator)
{
        // Allocate the coder->threads array if needed. It's done here instead
        // of when initializing the decoder because we don't need this if we
        // use the direct mode (we may even free coder->threads in the middle
        // of the file if we switch from threaded to direct mode).
        if (coder->threads == NULL) {
                coder->threads = lzma_alloc(
                        coder->threads_max * sizeof(struct worker_thread),
                        allocator);

                if (coder->threads == NULL)
                        return LZMA_MEM_ERROR;
        }

        // Pick a free structure.
        assert(coder->threads_initialized < coder->threads_max);
        struct worker_thread *thr
                        = &coder->threads[coder->threads_initialized];

        if (mythread_mutex_init(&thr->mutex))
                goto error_mutex;

        if (mythread_cond_init(&thr->cond))
                goto error_cond;

        thr->state = THR_IDLE;
        thr->in = NULL;
        thr->in_size = 0;
        thr->allocator = allocator;
        thr->coder = coder;
        thr->outbuf = NULL;
        thr->block_decoder = LZMA_NEXT_CODER_INIT;
        thr->mem_filters = 0;

        if (mythread_create(&thr->thread_id, worker_decoder, thr))
                goto error_thread;

        ++coder->threads_initialized;
        coder->thr = thr;

        return LZMA_OK;

error_thread:
        mythread_cond_destroy(&thr->cond);

error_cond:
        mythread_mutex_destroy(&thr->mutex);

error_mutex:
        return LZMA_MEM_ERROR;
}


static lzma_ret
get_thread(struct lzma_stream_coder *coder, const lzma_allocator *allocator)
{
        // If there is a free structure on the stack, use it.
        mythread_sync(coder->mutex) {
                if (coder->threads_free != NULL) {
                        coder->thr = coder->threads_free;
                        coder->threads_free = coder->threads_free->next;

                        // The thread is no longer in the cache so substract
                        // it from the cached memory usage. Don't add it
                        // to mem_in_use though; the caller will handle it
                        // since it knows how much memory it will actually
                        // use (the filter chain might change).
                        coder->mem_cached -= coder->thr->mem_filters;
                }
        }

        if (coder->thr == NULL) {
                assert(coder->threads_initialized < coder->threads_max);

                // Initialize a new thread.
                return_if_error(initialize_new_thread(coder, allocator));
        }

        coder->thr->in_filled = 0;
        coder->thr->in_pos = 0;
        coder->thr->out_pos = 0;

        coder->thr->progress_in = 0;
        coder->thr->progress_out = 0;

        coder->thr->partial_update = PARTIAL_DISABLED;

        return LZMA_OK;
}


static lzma_ret
read_output_and_wait(struct lzma_stream_coder *coder,
                const lzma_allocator *allocator,
                uint8_t *restrict out, size_t *restrict out_pos,
                size_t out_size,
                bool *input_is_possible,
                bool waiting_allowed,
                mythread_condtime *wait_abs, bool *has_blocked)
{
        lzma_ret ret = LZMA_OK;

        mythread_sync(coder->mutex) {
                do {
                        // Get as much output from the queue as is possible
                        // without blocking.
                        const size_t out_start = *out_pos;
                        do {
                                ret = lzma_outq_read(&coder->outq, allocator,
                                                out, out_pos, out_size,
                                                NULL, NULL);

                                // If a Block was finished, tell the worker
                                // thread of the next Block (if it is still
                                // running) to start telling the main thread
                                // when new output is available.
                                if (ret == LZMA_STREAM_END)
                                        lzma_outq_enable_partial_output(
                                                &coder->outq,
                                                &worker_enable_partial_update);

                                // Loop until a Block wasn't finished.
                                // It's important to loop around even if
                                // *out_pos == out_size because there could
                                // be an empty Block that will return
                                // LZMA_STREAM_END without needing any
                                // output space.
                        } while (ret == LZMA_STREAM_END);

                        // Check if lzma_outq_read reported an error from
                        // the Block decoder.
                        if (ret != LZMA_OK)
                                break;

                        // If the output buffer is now full but it wasn't full
                        // when this function was called, set out_was_filled.
                        // This way the next call to stream_decode_mt() knows
                        // that some output was produced and no output space
                        // remained in the previous call to stream_decode_mt().
                        if (*out_pos == out_size && *out_pos != out_start)
                                coder->out_was_filled = true;

                        // Check if any thread has indicated an error.
                        if (coder->thread_error != LZMA_OK) {
                                // If LZMA_FAIL_FAST was used, report errors
                                // from worker threads immediately.
                                if (coder->fail_fast) {
                                        ret = coder->thread_error;
                                        break;
                                }

                                // Otherwise set pending_error. The value we
                                // set here will not actually get used other
                                // than working as a flag that an error has
                                // occurred. This is because in SEQ_ERROR
                                // all output before the error will be read
                                // first by calling this function, and once we
                                // reach the location of the (first) error the
                                // error code from the above lzma_outq_read()
                                // will be returned to the application.
                                //
                                // Use LZMA_PROG_ERROR since the value should
                                // never leak to the application. It's
                                // possible that pending_error has already
                                // been set but that doesn't matter: if we get
                                // here, pending_error only works as a flag.
                                coder->pending_error = LZMA_PROG_ERROR;
                        }

                        // Check if decoding of the next Block can be started.
                        // The memusage of the active threads must be low
                        // enough, there must be a free buffer slot in the
                        // output queue, and there must be a free thread
                        // (that can be either created or an existing one
                        // reused).
                        //
                        // NOTE: This is checked after reading the output
                        // above because reading the output can free a slot in
                        // the output queue and also reduce active memusage.
                        //
                        // NOTE: If output queue is empty, then input will
                        // always be possible.
                        if (input_is_possible != NULL
                                        && coder->memlimit_threading
                                                - coder->mem_in_use
                                                - coder->outq.mem_in_use
                                                >= coder->mem_next_block
                                        && lzma_outq_has_buf(&coder->outq)
                                        && (coder->threads_initialized
                                                        < coder->threads_max
                                                || coder->threads_free
                                                        != NULL)) {
                                *input_is_possible = true;
                                break;
                        }

                        // If the caller doesn't want us to block, return now.
                        if (!waiting_allowed)
                                break;

                        // This check is needed only when input_is_possible
                        // is NULL. We must return if we aren't waiting for
                        // input to become possible and there is no more
                        // output coming from the queue.
                        if (lzma_outq_is_empty(&coder->outq)) {
                                assert(input_is_possible == NULL);
                                break;
                        }

                        // If there is more data available from the queue,
                        // our out buffer must be full and we need to return
                        // so that the application can provide more output
                        // space.
                        //
                        // NOTE: In general lzma_outq_is_readable() can return
                        // true also when there are no more bytes available.
                        // This can happen when a Block has finished without
                        // providing any new output. We know that this is not
                        // the case because in the beginning of this loop we
                        // tried to read as much as possible even when we had
                        // no output space left and the mutex has been locked
                        // all the time (so worker threads cannot have changed
                        // anything). Thus there must be actual pending output
                        // in the queue.
                        if (lzma_outq_is_readable(&coder->outq)) {
                                assert(*out_pos == out_size);
                                break;
                        }

                        // If the application stops providing more input
                        // in the middle of a Block, there will eventually
                        // be one worker thread left that is stuck waiting for
                        // more input (that might never arrive) and a matching
                        // outbuf which the worker thread cannot finish due
                        // to lack of input. We must detect this situation,
                        // otherwise we would end up waiting indefinitely
                        // (if no timeout is in use) or keep returning
                        // LZMA_TIMED_OUT while making no progress. Thus, the
                        // application would never get LZMA_BUF_ERROR from
                        // lzma_code() which would tell the application that
                        // no more progress is possible. No LZMA_BUF_ERROR
                        // means that, for example, truncated .xz files could
                        // cause an infinite loop.
                        //
                        // A worker thread doing partial updates will
                        // store not only the output position in outbuf->pos
                        // but also the matching input position in
                        // outbuf->decoder_in_pos. Here we check if that
                        // input position matches the amount of input that
                        // the worker thread has been given (in_filled).
                        // If so, we must return and not wait as no more
                        // output will be coming without first getting more
                        // input to the worker thread. If the application
                        // keeps calling lzma_code() without providing more
                        // input, it will eventually get LZMA_BUF_ERROR.
                        //
                        // NOTE: We can read partial_update and in_filled
                        // without thr->mutex as only the main thread
                        // modifies these variables. decoder_in_pos requires
                        // coder->mutex which we are already holding.
                        if (coder->thr != NULL && coder->thr->partial_update
                                        != PARTIAL_DISABLED) {
                                // There is exactly one outbuf in the queue.
                                assert(coder->thr->outbuf == coder->outq.head);
                                assert(coder->thr->outbuf == coder->outq.tail);

                                if (coder->thr->outbuf->decoder_in_pos
                                                == coder->thr->in_filled)
                                        break;
                        }

                        // Wait for input or output to become possible.
                        if (coder->timeout != 0) {
                                // See the comment in stream_encoder_mt.c
                                // about why mythread_condtime_set() is used
                                // like this.
                                //
                                // FIXME?
                                // In contrast to the encoder, this calls
                                // _condtime_set while the mutex is locked.
                                if (!*has_blocked) {
                                        *has_blocked = true;
                                        mythread_condtime_set(wait_abs,
                                                        &coder->cond,
                                                        coder->timeout);
                                }

                                if (mythread_cond_timedwait(&coder->cond,
                                                &coder->mutex,
                                                wait_abs) != 0) {
                                        ret = LZMA_TIMED_OUT;
                                        break;
                                }
                        } else {
                                mythread_cond_wait(&coder->cond,
                                                &coder->mutex);
                        }
                } while (ret == LZMA_OK);
        }

        // If we are returning an error, then the application cannot get
        // more output from us and thus keeping the threads running is
        // useless and waste of CPU time.
        if (ret != LZMA_OK && ret != LZMA_TIMED_OUT)
                threads_stop(coder);

        return ret;
}


static lzma_ret
decode_block_header(struct lzma_stream_coder *coder,
                const lzma_allocator *allocator, const uint8_t *restrict in,
                size_t *restrict in_pos, size_t in_size)
{
        if (*in_pos >= in_size)
                return LZMA_OK;

        if (coder->pos == 0) {
                // Detect if it's Index.
                if (in[*in_pos] == INDEX_INDICATOR)
                        return LZMA_INDEX_DETECTED;

                // Calculate the size of the Block Header. Note that
                // Block Header decoder wants to see this byte too
                // so don't advance *in_pos.
                coder->block_options.header_size
                                = lzma_block_header_size_decode(
                                        in[*in_pos]);
        }

        // Copy the Block Header to the internal buffer.
        lzma_bufcpy(in, in_pos, in_size, coder->buffer, &coder->pos,
                        coder->block_options.header_size);

        // Return if we didn't get the whole Block Header yet.
        if (coder->pos < coder->block_options.header_size)
                return LZMA_OK;

        coder->pos = 0;

        // Version 1 is needed to support the .ignore_check option.
        coder->block_options.version = 1;

        // Block Header decoder will initialize all members of this array
        // so we don't need to do it here.
        coder->block_options.filters = coder->filters;

        // Decode the Block Header.
        return_if_error(lzma_block_header_decode(&coder->block_options,
                        allocator, coder->buffer));

        // If LZMA_IGNORE_CHECK was used, this flag needs to be set.
        // It has to be set after lzma_block_header_decode() because
        // it always resets this to false.
        coder->block_options.ignore_check = coder->ignore_check;

        // coder->block_options is ready now.
        return LZMA_STREAM_END;
}


/// Get the size of the Compressed Data + Block Padding + Check.
static size_t
comp_blk_size(const struct lzma_stream_coder *coder)
{
        return vli_ceil4(coder->block_options.compressed_size)
                        + lzma_check_size(coder->stream_flags.check);
}


/// Returns true if the size (compressed or uncompressed) is such that
/// threaded decompression cannot be used. Sizes that are too big compared
/// to SIZE_MAX must be rejected to avoid integer overflows and truncations
/// when lzma_vli is assigned to a size_t.
static bool
is_direct_mode_needed(lzma_vli size)
{
        return size == LZMA_VLI_UNKNOWN || size > SIZE_MAX / 3;
}


static lzma_ret
stream_decoder_reset(struct lzma_stream_coder *coder,
                const lzma_allocator *allocator)
{
        // Initialize the Index hash used to verify the Index.
        coder->index_hash = lzma_index_hash_init(coder->index_hash, allocator);
        if (coder->index_hash == NULL)
                return LZMA_MEM_ERROR;

        // Reset the rest of the variables.
        coder->sequence = SEQ_STREAM_HEADER;
        coder->pos = 0;

        return LZMA_OK;
}


static lzma_ret
stream_decode_mt(void *coder_ptr, const lzma_allocator *allocator,
                 const uint8_t *restrict in, size_t *restrict in_pos,
                 size_t in_size,
                 uint8_t *restrict out, size_t *restrict out_pos,
                 size_t out_size, lzma_action action)
{
        struct lzma_stream_coder *coder = coder_ptr;

        mythread_condtime wait_abs;
        bool has_blocked = false;

        // Determine if in SEQ_BLOCK_HEADER and SEQ_BLOCK_THR_RUN we should
        // tell read_output_and_wait() to wait until it can fill the output
        // buffer (or a timeout occurs). Two conditions must be met:
        //
        // (1) If the caller provided no new input. The reason for this
        //     can be, for example, the end of the file or that there is
        //     a pause in the input stream and more input is available
        //     a little later. In this situation we should wait for output
        //     because otherwise we would end up in a busy-waiting loop where
        //     we make no progress and the application just calls us again
        //     without providing any new input. This would then result in
        //     LZMA_BUF_ERROR even though more output would be available
        //     once the worker threads decode more data.
        //
        // (2) Even if (1) is true, we will not wait if the previous call to
        //     this function managed to produce some output and the output
        //     buffer became full. This is for compatibility with applications
        //     that call lzma_code() in such a way that new input is provided
        //     only when the output buffer didn't become full. Without this
        //     trick such applications would have bad performance (bad
        //     parallelization due to decoder not getting input fast enough).
        //
        //     NOTE: Such loops might require that timeout is disabled (0)
        //     if they assume that output-not-full implies that all input has
        //     been consumed. If and only if timeout is enabled, we may return
        //     when output isn't full *and* not all input has been consumed.
        //
        // However, if LZMA_FINISH is used, the above is ignored and we always
        // wait (timeout can still cause us to return) because we know that
        // we won't get any more input. This matters if the input file is
        // truncated and we are doing single-shot decoding, that is,
        // timeout = 0 and LZMA_FINISH is used on the first call to
        // lzma_code() and the output buffer is known to be big enough
        // to hold all uncompressed data:
        //
        //   - If LZMA_FINISH wasn't handled specially, we could return
        //     LZMA_OK before providing all output that is possible with the
        //     truncated input. The rest would be available if lzma_code() was
        //     called again but then it's not single-shot decoding anymore.
        //
        //   - By handling LZMA_FINISH specially here, the first call will
        //     produce all the output, matching the behavior of the
        //     single-threaded decoder.
        //
        // So it's a very specific corner case but also easy to avoid. Note
        // that this special handling of LZMA_FINISH has no effect for
        // single-shot decoding when the input file is valid (not truncated);
        // premature LZMA_OK wouldn't be possible as long as timeout = 0.
        const bool waiting_allowed = action == LZMA_FINISH
                        || (*in_pos == in_size && !coder->out_was_filled);
        coder->out_was_filled = false;

        while (true)
        switch (coder->sequence) {
        case SEQ_STREAM_HEADER: {
                // Copy the Stream Header to the internal buffer.
                const size_t in_old = *in_pos;
                lzma_bufcpy(in, in_pos, in_size, coder->buffer, &coder->pos,
                                LZMA_STREAM_HEADER_SIZE);
                coder->progress_in += *in_pos - in_old;

                // Return if we didn't get the whole Stream Header yet.
                if (coder->pos < LZMA_STREAM_HEADER_SIZE)
                        return LZMA_OK;

                coder->pos = 0;

                // Decode the Stream Header.
                const lzma_ret ret = lzma_stream_header_decode(
                                &coder->stream_flags, coder->buffer);
                if (ret != LZMA_OK)
                        return ret == LZMA_FORMAT_ERROR && !coder->first_stream
                                        ? LZMA_DATA_ERROR : ret;

                // If we are decoding concatenated Streams, and the later
                // Streams have invalid Header Magic Bytes, we give
                // LZMA_DATA_ERROR instead of LZMA_FORMAT_ERROR.
                coder->first_stream = false;

                // Copy the type of the Check so that Block Header and Block
                // decoders see it.
                coder->block_options.check = coder->stream_flags.check;

                // Even if we return LZMA_*_CHECK below, we want
                // to continue from Block Header decoding.
                coder->sequence = SEQ_BLOCK_HEADER;

                // Detect if there's no integrity check or if it is
                // unsupported if those were requested by the application.
                if (coder->tell_no_check && coder->stream_flags.check
                                == LZMA_CHECK_NONE)
                        return LZMA_NO_CHECK;

                if (coder->tell_unsupported_check
                                && !lzma_check_is_supported(
                                        coder->stream_flags.check))
                        return LZMA_UNSUPPORTED_CHECK;

                if (coder->tell_any_check)
                        return LZMA_GET_CHECK;
        }

        // Fall through

        case SEQ_BLOCK_HEADER: {
                const size_t in_old = *in_pos;
                const lzma_ret ret = decode_block_header(coder, allocator,
                                in, in_pos, in_size);
                coder->progress_in += *in_pos - in_old;

                if (ret == LZMA_OK) {
                        // We didn't decode the whole Block Header yet.
                        //
                        // Read output from the queue before returning. This
                        // is important because it is possible that the
                        // application doesn't have any new input available
                        // immediately. If we didn't try to copy output from
                        // the output queue here, lzma_code() could end up
                        // returning LZMA_BUF_ERROR even though queued output
                        // is available.
                        //
                        // If the lzma_code() call provided at least one input
                        // byte, only copy as much data from the output queue
                        // as is available immediately. This way the
                        // application will be able to provide more input
                        // without a delay.
                        //
                        // On the other hand, if lzma_code() was called with
                        // an empty input buffer(*), treat it specially: try
                        // to fill the output buffer even if it requires
                        // waiting for the worker threads to provide output
                        // (timeout, if specified, can still cause us to
                        // return).
                        //
                        //   - This way the application will be able to get all
                        //     data that can be decoded from the input provided
                        //     so far.
                        //
                        //   - We avoid both premature LZMA_BUF_ERROR and
                        //     busy-waiting where the application repeatedly
                        //     calls lzma_code() which immediately returns
                        //     LZMA_OK without providing new data.
                        //
                        //   - If the queue becomes empty, we won't wait
                        //     anything and will return LZMA_OK immediately
                        //     (coder->timeout is completely ignored).
                        //
                        // (*) See the comment at the beginning of this
                        //     function how waiting_allowed is determined
                        //     and why there is an exception to the rule
                        //     of "called with an empty input buffer".
                        assert(*in_pos == in_size);

                        // If LZMA_FINISH was used we know that we won't get
                        // more input, so the file must be truncated if we
                        // get here. If worker threads don't detect any
                        // errors, eventually there will be no more output
                        // while we keep returning LZMA_OK which gets
                        // converted to LZMA_BUF_ERROR in lzma_code().
                        //
                        // If fail-fast is enabled then we will return
                        // immediately using LZMA_DATA_ERROR instead of
                        // LZMA_OK or LZMA_BUF_ERROR. Rationale for the
                        // error code:
                        //
                        //   - Worker threads may have a large amount of
                        //     not-yet-decoded input data and we don't
                        //     know for sure if all data is valid. Bad
                        //     data there would result in LZMA_DATA_ERROR
                        //     when fail-fast isn't used.
                        //
                        //   - Immediate LZMA_BUF_ERROR would be a bit weird
                        //     considering the older liblzma code. lzma_code()
                        //     even has an assertion to prevent coders from
                        //     returning LZMA_BUF_ERROR directly.
                        //
                        // The downside of this is that with fail-fast apps
                        // cannot always distinguish between corrupt and
                        // truncated files.
                        if (action == LZMA_FINISH && coder->fail_fast) {
                                // We won't produce any more output. Stop
                                // the unfinished worker threads so they
                                // won't waste CPU time.
                                threads_stop(coder);
                                return LZMA_DATA_ERROR;
                        }

                        // read_output_and_wait() will call threads_stop()
                        // if needed so with that we can use return_if_error.
                        return_if_error(read_output_and_wait(coder, allocator,
                                out, out_pos, out_size,
                                NULL, waiting_allowed,
                                &wait_abs, &has_blocked));

                        if (coder->pending_error != LZMA_OK) {
                                coder->sequence = SEQ_ERROR;
                                break;
                        }

                        return LZMA_OK;
                }

                if (ret == LZMA_INDEX_DETECTED) {
                        coder->sequence = SEQ_INDEX_WAIT_OUTPUT;
                        break;
                }

                // See if an error occurred.
                if (ret != LZMA_STREAM_END) {
                        // NOTE: Here and in all other places where
                        // pending_error is set, it may overwrite the value
                        // (LZMA_PROG_ERROR) set by read_output_and_wait().
                        // That function might overwrite value set here too.
                        // These are fine because when read_output_and_wait()
                        // sets pending_error, it actually works as a flag
                        // variable only ("some error has occurred") and the
                        // actual value of pending_error is not used in
                        // SEQ_ERROR. In such cases SEQ_ERROR will eventually
                        // get the correct error code from the return value of
                        // a later read_output_and_wait() call.
                        coder->pending_error = ret;
                        coder->sequence = SEQ_ERROR;
                        break;
                }

                // Calculate the memory usage of the filters / Block decoder.
                coder->mem_next_filters = lzma_raw_decoder_memusage(
                                coder->filters);

                if (coder->mem_next_filters == UINT64_MAX) {
                        // One or more unknown Filter IDs.
                        coder->pending_error = LZMA_OPTIONS_ERROR;
                        coder->sequence = SEQ_ERROR;
                        break;
                }

                coder->sequence = SEQ_BLOCK_INIT;
        }

        // Fall through

        case SEQ_BLOCK_INIT: {
                // Check if decoding is possible at all with the current
                // memlimit_stop which we must never exceed.
                //
                // This needs to be the first thing in SEQ_BLOCK_INIT
                // to make it possible to restart decoding after increasing
                // memlimit_stop with lzma_memlimit_set().
                if (coder->mem_next_filters > coder->memlimit_stop) {
                        // Flush pending output before returning
                        // LZMA_MEMLIMIT_ERROR. If the application doesn't
                        // want to increase the limit, at least it will get
                        // all the output possible so far.
                        return_if_error(read_output_and_wait(coder, allocator,
                                        out, out_pos, out_size,
                                        NULL, true, &wait_abs, &has_blocked));

                        if (!lzma_outq_is_empty(&coder->outq))
                                return LZMA_OK;

                        return LZMA_MEMLIMIT_ERROR;
                }

                // Check if the size information is available in Block Header.
                // If it is, check if the sizes are small enough that we don't
                // need to worry *too* much about integer overflows later in
                // the code. If these conditions are not met, we must use the
                // single-threaded direct mode.
                if (is_direct_mode_needed(coder->block_options.compressed_size)
                                || is_direct_mode_needed(
                                coder->block_options.uncompressed_size)) {
                        coder->sequence = SEQ_BLOCK_DIRECT_INIT;
                        break;
                }

                // Calculate the amount of memory needed for the input and
                // output buffers in threaded mode.
                //
                // These cannot overflow because we already checked that
                // the sizes are small enough using is_direct_mode_needed().
                coder->mem_next_in = comp_blk_size(coder);
                const uint64_t mem_buffers = coder->mem_next_in
                                + lzma_outq_outbuf_memusage(
                                coder->block_options.uncompressed_size);

                // Add the amount needed by the filters.
                // Avoid integer overflows.
                if (UINT64_MAX - mem_buffers < coder->mem_next_filters) {
                        // Use direct mode if the memusage would overflow.
                        // This is a theoretical case that shouldn't happen
                        // in practice unless the input file is weird (broken
                        // or malicious).
                        coder->sequence = SEQ_BLOCK_DIRECT_INIT;
                        break;
                }

                // Amount of memory needed to decode this Block in
                // threaded mode:
                coder->mem_next_block = coder->mem_next_filters + mem_buffers;

                // If this alone would exceed memlimit_threading, then we must
                // use the single-threaded direct mode.
                if (coder->mem_next_block > coder->memlimit_threading) {
                        coder->sequence = SEQ_BLOCK_DIRECT_INIT;
                        break;
                }

                // Use the threaded mode. Free the direct mode decoder in
                // case it has been initialized.
                lzma_next_end(&coder->block_decoder, allocator);
                coder->mem_direct_mode = 0;

                // Since we already know what the sizes are supposed to be,
                // we can already add them to the Index hash. The Block
                // decoder will verify the values while decoding.
                const lzma_ret ret = lzma_index_hash_append(coder->index_hash,
                                lzma_block_unpadded_size(
                                        &coder->block_options),
                                coder->block_options.uncompressed_size);
                if (ret != LZMA_OK) {
                        coder->pending_error = ret;
                        coder->sequence = SEQ_ERROR;
                        break;
                }

                coder->sequence = SEQ_BLOCK_THR_INIT;
        }

        // Fall through

        case SEQ_BLOCK_THR_INIT: {
                // We need to wait for a multiple conditions to become true
                // until we can initialize the Block decoder and let a worker
                // thread decode it:
                //
                //   - Wait for the memory usage of the active threads to drop
                //     so that starting the decoding of this Block won't make
                //     us go over memlimit_threading.
                //
                //   - Wait for at least one free output queue slot.
                //
                //   - Wait for a free worker thread.
                //
                // While we wait, we must copy decompressed data to the out
                // buffer and catch possible decoder errors.
                //
                // read_output_and_wait() does all the above.
                bool block_can_start = false;

                return_if_error(read_output_and_wait(coder, allocator,
                                out, out_pos, out_size,
                                &block_can_start, true,
                                &wait_abs, &has_blocked));

                if (coder->pending_error != LZMA_OK) {
                        coder->sequence = SEQ_ERROR;
                        break;
                }

                if (!block_can_start) {
                        // It's not a timeout because return_if_error handles
                        // it already. Output queue cannot be empty either
                        // because in that case block_can_start would have
                        // been true. Thus the output buffer must be full and
                        // the queue isn't empty.
                        assert(*out_pos == out_size);
                        assert(!lzma_outq_is_empty(&coder->outq));
                        return LZMA_OK;
                }

                // We know that we can start decoding this Block without
                // exceeding memlimit_threading. However, to stay below
                // memlimit_threading may require freeing some of the
                // cached memory.
                //
                // Get a local copy of variables that require locking the
                // mutex. It is fine if the worker threads modify the real
                // values after we read these as those changes can only be
                // towards more favorable conditions (less memory in use,
                // more in cache).
                //
                // These are initalized to silence warnings.
                uint64_t mem_in_use = 0;
                uint64_t mem_cached = 0;
                struct worker_thread *thr = NULL;

                mythread_sync(coder->mutex) {
                        mem_in_use = coder->mem_in_use;
                        mem_cached = coder->mem_cached;
                        thr = coder->threads_free;
                }

                // The maximum amount of memory that can be held by other
                // threads and cached buffers while allowing us to start
                // decoding the next Block.
                const uint64_t mem_max = coder->memlimit_threading
                                - coder->mem_next_block;

                // If the existing allocations are so large that starting
                // to decode this Block might exceed memlimit_threads,
                // try to free memory from the output queue cache first.
                //
                // NOTE: This math assumes the worst case. It's possible
                // that the limit wouldn't be exceeded if the existing cached
                // allocations are reused.
                if (mem_in_use + mem_cached + coder->outq.mem_allocated
                                > mem_max) {
                        // Clear the outq cache except leave one buffer in
                        // the cache if its size is correct. That way we
                        // don't free and almost immediately reallocate
                        // an identical buffer.
                        lzma_outq_clear_cache2(&coder->outq, allocator,
                                coder->block_options.uncompressed_size);
                }

                // If there is at least one worker_thread in the cache and
                // the existing allocations are so large that starting to
                // decode this Block might exceed memlimit_threads, free
                // memory by freeing cached Block decoders.
                //
                // NOTE: The comparison is different here than above.
                // Here we don't care about cached buffers in outq anymore
                // and only look at memory actually in use. This is because
                // if there is something in outq cache, it's a single buffer
                // that can be used as is. We ensured this in the above
                // if-block.
                uint64_t mem_freed = 0;
                if (thr != NULL && mem_in_use + mem_cached
                                + coder->outq.mem_in_use > mem_max) {
                        // Don't free the first Block decoder if its memory
                        // usage isn't greater than what this Block will need.
                        // Typically the same filter chain is used for all
                        // Blocks so this way the allocations can be reused
                        // when get_thread() picks the first worker_thread
                        // from the cache.
                        if (thr->mem_filters <= coder->mem_next_filters)
                                thr = thr->next;

                        while (thr != NULL) {
                                lzma_next_end(&thr->block_decoder, allocator);
                                mem_freed += thr->mem_filters;
                                thr->mem_filters = 0;
                                thr = thr->next;
                        }
                }

                // Update the memory usage counters. Note that coder->mem_*
                // may have changed since we read them so we must substract
                // or add the changes.
                mythread_sync(coder->mutex) {
                        coder->mem_cached -= mem_freed;

                        // Memory needed for the filters and the input buffer.
                        // The output queue takes care of its own counter so
                        // we don't touch it here.
                        //
                        // NOTE: After this, coder->mem_in_use +
                        // coder->mem_cached might count the same thing twice.
                        // If so, this will get corrected in get_thread() when
                        // a worker_thread is picked from coder->free_threads
                        // and its memory usage is substracted from mem_cached.
                        coder->mem_in_use += coder->mem_next_in
                                        + coder->mem_next_filters;
                }

                // Allocate memory for the output buffer in the output queue.
                lzma_ret ret = lzma_outq_prealloc_buf(
                                &coder->outq, allocator,
                                coder->block_options.uncompressed_size);
                if (ret != LZMA_OK) {
                        threads_stop(coder);
                        return ret;
                }

                // Set up coder->thr.
                ret = get_thread(coder, allocator);
                if (ret != LZMA_OK) {
                        threads_stop(coder);
                        return ret;
                }

                // The new Block decoder memory usage is already counted in
                // coder->mem_in_use. Store it in the thread too.
                coder->thr->mem_filters = coder->mem_next_filters;

                // Initialize the Block decoder.
                coder->thr->block_options = coder->block_options;
                ret = lzma_block_decoder_init(
                                        &coder->thr->block_decoder, allocator,
                                        &coder->thr->block_options);

                // Free the allocated filter options since they are needed
                // only to initialize the Block decoder.
                lzma_filters_free(coder->filters, allocator);
                coder->thr->block_options.filters = NULL;

                // Check if memory usage calculation and Block encoder
                // initialization succeeded.
                if (ret != LZMA_OK) {
                        coder->pending_error = ret;
                        coder->sequence = SEQ_ERROR;
                        break;
                }

                // Allocate the input buffer.
                coder->thr->in_size = coder->mem_next_in;
                coder->thr->in = lzma_alloc(coder->thr->in_size, allocator);
                if (coder->thr->in == NULL) {
                        threads_stop(coder);
                        return LZMA_MEM_ERROR;
                }

                // Get the preallocated output buffer.
                coder->thr->outbuf = lzma_outq_get_buf(
                                &coder->outq, coder->thr);

                // Start the decoder.
                mythread_sync(coder->thr->mutex) {
                        assert(coder->thr->state == THR_IDLE);
                        coder->thr->state = THR_RUN;
                        mythread_cond_signal(&coder->thr->cond);
                }

                // Enable output from the thread that holds the oldest output
                // buffer in the output queue (if such a thread exists).
                mythread_sync(coder->mutex) {
                        lzma_outq_enable_partial_output(&coder->outq,
                                        &worker_enable_partial_update);
                }

                coder->sequence = SEQ_BLOCK_THR_RUN;
        }

        // Fall through

        case SEQ_BLOCK_THR_RUN: {
                if (action == LZMA_FINISH && coder->fail_fast) {
                        // We know that we won't get more input and that
                        // the caller wants fail-fast behavior. If we see
                        // that we don't have enough input to finish this
                        // Block, return LZMA_DATA_ERROR immediately.
                        // See SEQ_BLOCK_HEADER for the error code rationale.
                        const size_t in_avail = in_size - *in_pos;
                        const size_t in_needed = coder->thr->in_size
                                        - coder->thr->in_filled;
                        if (in_avail < in_needed) {
                                threads_stop(coder);
                                return LZMA_DATA_ERROR;
                        }
                }

                // Copy input to the worker thread.
                size_t cur_in_filled = coder->thr->in_filled;
                lzma_bufcpy(in, in_pos, in_size, coder->thr->in,
                                &cur_in_filled, coder->thr->in_size);

                // Tell the thread how much we copied.
                mythread_sync(coder->thr->mutex) {
                        coder->thr->in_filled = cur_in_filled;

                        // NOTE: Most of the time we are copying input faster
                        // than the thread can decode so most of the time
                        // calling mythread_cond_signal() is useless but
                        // we cannot make it conditional because thr->in_pos
                        // is updated without a mutex. And the overhead should
                        // be very much negligible anyway.
                        mythread_cond_signal(&coder->thr->cond);
                }

                // Read output from the output queue. Just like in
                // SEQ_BLOCK_HEADER, we wait to fill the output buffer
                // only if waiting_allowed was set to true in the beginning
                // of this function (see the comment there).
                return_if_error(read_output_and_wait(coder, allocator,
                                out, out_pos, out_size,
                                NULL, waiting_allowed,
                                &wait_abs, &has_blocked));

                if (coder->pending_error != LZMA_OK) {
                        coder->sequence = SEQ_ERROR;
                        break;
                }

                // Return if the input didn't contain the whole Block.
                if (coder->thr->in_filled < coder->thr->in_size) {
                        assert(*in_pos == in_size);
                        return LZMA_OK;
                }

                // The whole Block has been copied to the thread-specific
                // buffer. Continue from the next Block Header or Index.
                coder->thr = NULL;
                coder->sequence = SEQ_BLOCK_HEADER;
                break;
        }

        case SEQ_BLOCK_DIRECT_INIT: {
                // Wait for the threads to finish and that all decoded data
                // has been copied to the output. That is, wait until the
                // output queue becomes empty.
                //
                // NOTE: No need to check for coder->pending_error as
                // we aren't consuming any input until the queue is empty
                // and if there is a pending error, read_output_and_wait()
                // will eventually return it before the queue is empty.
                return_if_error(read_output_and_wait(coder, allocator,
                                out, out_pos, out_size,
                                NULL, true, &wait_abs, &has_blocked));
                if (!lzma_outq_is_empty(&coder->outq))
                        return LZMA_OK;

                // Free the cached output buffers.
                lzma_outq_clear_cache(&coder->outq, allocator);

                // Get rid of the worker threads, including the coder->threads
                // array.
                threads_end(coder, allocator);

                // Initialize the Block decoder.
                const lzma_ret ret = lzma_block_decoder_init(
                                &coder->block_decoder, allocator,
                                &coder->block_options);

                // Free the allocated filter options since they are needed
                // only to initialize the Block decoder.
                lzma_filters_free(coder->filters, allocator);
                coder->block_options.filters = NULL;

                // Check if Block decoder initialization succeeded.
                if (ret != LZMA_OK)
                        return ret;

                // Make the memory usage visible to _memconfig().
                coder->mem_direct_mode = coder->mem_next_filters;

                coder->sequence = SEQ_BLOCK_DIRECT_RUN;
        }

        // Fall through

        case SEQ_BLOCK_DIRECT_RUN: {
                const size_t in_old = *in_pos;
                const size_t out_old = *out_pos;
                const lzma_ret ret = coder->block_decoder.code(
                                coder->block_decoder.coder, allocator,
                                in, in_pos, in_size, out, out_pos, out_size,
                                action);
                coder->progress_in += *in_pos - in_old;
                coder->progress_out += *out_pos - out_old;

                if (ret != LZMA_STREAM_END)
                        return ret;

                // Block decoded successfully. Add the new size pair to
                // the Index hash.
                return_if_error(lzma_index_hash_append(coder->index_hash,
                                lzma_block_unpadded_size(
                                        &coder->block_options),
                                coder->block_options.uncompressed_size));

                coder->sequence = SEQ_BLOCK_HEADER;
                break;
        }

        case SEQ_INDEX_WAIT_OUTPUT:
                // Flush the output from all worker threads so that we can
                // decode the Index without thinking about threading.
                return_if_error(read_output_and_wait(coder, allocator,
                                out, out_pos, out_size,
                                NULL, true, &wait_abs, &has_blocked));

                if (!lzma_outq_is_empty(&coder->outq))
                        return LZMA_OK;

                coder->sequence = SEQ_INDEX_DECODE;

        // Fall through

        case SEQ_INDEX_DECODE: {
                // If we don't have any input, don't call
                // lzma_index_hash_decode() since it would return
                // LZMA_BUF_ERROR, which we must not do here.
                if (*in_pos >= in_size)
                        return LZMA_OK;

                // Decode the Index and compare it to the hash calculated
                // from the sizes of the Blocks (if any).
                const size_t in_old = *in_pos;
                const lzma_ret ret = lzma_index_hash_decode(coder->index_hash,
                                in, in_pos, in_size);
                coder->progress_in += *in_pos - in_old;
                if (ret != LZMA_STREAM_END)
                        return ret;

                coder->sequence = SEQ_STREAM_FOOTER;
        }

        // Fall through

        case SEQ_STREAM_FOOTER: {
                // Copy the Stream Footer to the internal buffer.
                const size_t in_old = *in_pos;
                lzma_bufcpy(in, in_pos, in_size, coder->buffer, &coder->pos,
                                LZMA_STREAM_HEADER_SIZE);
                coder->progress_in += *in_pos - in_old;

                // Return if we didn't get the whole Stream Footer yet.
                if (coder->pos < LZMA_STREAM_HEADER_SIZE)
                        return LZMA_OK;

                coder->pos = 0;

                // Decode the Stream Footer. The decoder gives
                // LZMA_FORMAT_ERROR if the magic bytes don't match,
                // so convert that return code to LZMA_DATA_ERROR.
                lzma_stream_flags footer_flags;
                const lzma_ret ret = lzma_stream_footer_decode(
                                &footer_flags, coder->buffer);
                if (ret != LZMA_OK)
                        return ret == LZMA_FORMAT_ERROR
                                        ? LZMA_DATA_ERROR : ret;

                // Check that Index Size stored in the Stream Footer matches
                // the real size of the Index field.
                if (lzma_index_hash_size(coder->index_hash)
                                != footer_flags.backward_size)
                        return LZMA_DATA_ERROR;

                // Compare that the Stream Flags fields are identical in
                // both Stream Header and Stream Footer.
                return_if_error(lzma_stream_flags_compare(
                                &coder->stream_flags, &footer_flags));

                if (!coder->concatenated)
                        return LZMA_STREAM_END;

                coder->sequence = SEQ_STREAM_PADDING;
        }

        // Fall through

        case SEQ_STREAM_PADDING:
                assert(coder->concatenated);

                // Skip over possible Stream Padding.
                while (true) {
                        if (*in_pos >= in_size) {
                                // Unless LZMA_FINISH was used, we cannot
                                // know if there's more input coming later.
                                if (action != LZMA_FINISH)
                                        return LZMA_OK;

                                // Stream Padding must be a multiple of
                                // four bytes.
                                return coder->pos == 0
                                                ? LZMA_STREAM_END
                                                : LZMA_DATA_ERROR;
                        }

                        // If the byte is not zero, it probably indicates
                        // beginning of a new Stream (or the file is corrupt).
                        if (in[*in_pos] != 0x00)
                                break;

                        ++*in_pos;
                        ++coder->progress_in;
                        coder->pos = (coder->pos + 1) & 3;
                }

                // Stream Padding must be a multiple of four bytes (empty
                // Stream Padding is OK).
                if (coder->pos != 0) {
                        ++*in_pos;
                        ++coder->progress_in;
                        return LZMA_DATA_ERROR;
                }

                // Prepare to decode the next Stream.
                return_if_error(stream_decoder_reset(coder, allocator));
                break;

        case SEQ_ERROR:
                if (!coder->fail_fast) {
                        // Let the application get all data before the point
                        // where the error was detected. This matches the
                        // behavior of single-threaded use.
                        //
                        // FIXME? Some errors (LZMA_MEM_ERROR) don't get here,
                        // they are returned immediately. Thus in rare cases
                        // the output will be less than in the single-threaded
                        // mode. Maybe this doesn't matter much in practice.
                        return_if_error(read_output_and_wait(coder, allocator,
                                        out, out_pos, out_size,
                                        NULL, true, &wait_abs, &has_blocked));

                        // We get here only if the error happened in the main
                        // thread, for example, unsupported Block Header.
                        if (!lzma_outq_is_empty(&coder->outq))
                                return LZMA_OK;
                }

                // We only get here if no errors were detected by the worker
                // threads. Errors from worker threads would have already been
                // returned by the call to read_output_and_wait() above.
                return coder->pending_error;

        default:
                assert(0);
                return LZMA_PROG_ERROR;
        }

        // Never reached
}


static void
stream_decoder_mt_end(void *coder_ptr, const lzma_allocator *allocator)
{
        struct lzma_stream_coder *coder = coder_ptr;

        threads_end(coder, allocator);
        lzma_outq_end(&coder->outq, allocator);

        lzma_next_end(&coder->block_decoder, allocator);
        lzma_filters_free(coder->filters, allocator);
        lzma_index_hash_end(coder->index_hash, allocator);

        lzma_free(coder, allocator);
        return;
}


static lzma_check
stream_decoder_mt_get_check(const void *coder_ptr)
{
        const struct lzma_stream_coder *coder = coder_ptr;
        return coder->stream_flags.check;
}


static lzma_ret
stream_decoder_mt_memconfig(void *coder_ptr, uint64_t *memusage,
                uint64_t *old_memlimit, uint64_t new_memlimit)
{
        // NOTE: This function gets/sets memlimit_stop. For now,
        // memlimit_threading cannot be modified after initialization.
        //
        // *memusage will include cached memory too. Excluding cached memory
        // would be misleading and it wouldn't help the applications to
        // know how much memory is actually needed to decompress the file
        // because the higher the number of threads and the memlimits are
        // the more memory the decoder may use.
        //
        // Setting a new limit includes the cached memory too and too low
        // limits will be rejected. Alternative could be to free the cached
        // memory immediately if that helps to bring the limit down but
        // the current way is the simplest. It's unlikely that limit needs
        // to be lowered in the middle of a file anyway; the typical reason
        // to want a new limit is to increase after LZMA_MEMLIMIT_ERROR
        // and even such use isn't common.
        struct lzma_stream_coder *coder = coder_ptr;

        mythread_sync(coder->mutex) {
                *memusage = coder->mem_direct_mode
                                + coder->mem_in_use
                                + coder->mem_cached
                                + coder->outq.mem_allocated;
        }

        // If no filter chains are allocated, *memusage may be zero.
        // Always return at least LZMA_MEMUSAGE_BASE.
        if (*memusage < LZMA_MEMUSAGE_BASE)
                *memusage = LZMA_MEMUSAGE_BASE;

        *old_memlimit = coder->memlimit_stop;

        if (new_memlimit != 0) {
                if (new_memlimit < *memusage)
                        return LZMA_MEMLIMIT_ERROR;

                coder->memlimit_stop = new_memlimit;
        }

        return LZMA_OK;
}


static void
stream_decoder_mt_get_progress(void *coder_ptr,
                uint64_t *progress_in, uint64_t *progress_out)
{
        struct lzma_stream_coder *coder = coder_ptr;

        // Lock coder->mutex to prevent finishing threads from moving their
        // progress info from the worker_thread structure to lzma_stream_coder.
        mythread_sync(coder->mutex) {
                *progress_in = coder->progress_in;
                *progress_out = coder->progress_out;

                for (size_t i = 0; i < coder->threads_initialized; ++i) {
                        mythread_sync(coder->threads[i].mutex) {
                                *progress_in += coder->threads[i].progress_in;
                                *progress_out += coder->threads[i]
                                                .progress_out;
                        }
                }
        }

        return;
}


static lzma_ret
stream_decoder_mt_init(lzma_next_coder *next, const lzma_allocator *allocator,
                       const lzma_mt *options)
{
        struct lzma_stream_coder *coder;

        if (options->threads == 0 || options->threads > LZMA_THREADS_MAX)
                return LZMA_OPTIONS_ERROR;

        if (options->flags & ~LZMA_SUPPORTED_FLAGS)
                return LZMA_OPTIONS_ERROR;

        lzma_next_coder_init(&stream_decoder_mt_init, next, allocator);

        coder = next->coder;
        if (!coder) {
                coder = lzma_alloc(sizeof(struct lzma_stream_coder), allocator);
                if (coder == NULL)
                        return LZMA_MEM_ERROR;

                next->coder = coder;

                if (mythread_mutex_init(&coder->mutex)) {
                        lzma_free(coder, allocator);
                        return LZMA_MEM_ERROR;
                }

                if (mythread_cond_init(&coder->cond)) {
                        mythread_mutex_destroy(&coder->mutex);
                        lzma_free(coder, allocator);
                        return LZMA_MEM_ERROR;
                }

                next->code = &stream_decode_mt;
                next->end = &stream_decoder_mt_end;
                next->get_check = &stream_decoder_mt_get_check;
                next->memconfig = &stream_decoder_mt_memconfig;
                next->get_progress = &stream_decoder_mt_get_progress;

                coder->filters[0].id = LZMA_VLI_UNKNOWN;
                memzero(&coder->outq, sizeof(coder->outq));

                coder->block_decoder = LZMA_NEXT_CODER_INIT;
                coder->mem_direct_mode = 0;

                coder->index_hash = NULL;
                coder->threads = NULL;
                coder->threads_free = NULL;
                coder->threads_initialized = 0;
        }

        // Cleanup old filter chain if one remains after unfinished decoding
        // of a previous Stream.
        lzma_filters_free(coder->filters, allocator);

        // By allocating threads from scratch we can start memory-usage
        // accounting from scratch, too. Changes in filter and block sizes may
        // affect number of threads.
        //
        // FIXME? Reusing should be easy but unlike the single-threaded
        // decoder, with some types of input file combinations reusing
        // could leave quite a lot of memory allocated but unused (first
        // file could allocate a lot, the next files could use fewer
        // threads and some of the allocations from the first file would not
        // get freed unless memlimit_threading forces us to clear caches).
        //
        // NOTE: The direct mode decoder isn't freed here if one exists.
        // It will be reused or freed as needed in the main loop.
        threads_end(coder, allocator);

        // All memusage counters start at 0 (including mem_direct_mode).
        // The little extra that is needed for the structs in this file
        // get accounted well enough by the filter chain memory usage
        // which adds LZMA_MEMUSAGE_BASE for each chain. However,
        // stream_decoder_mt_memconfig() has to handle this specially so that
        // it will never return less than LZMA_MEMUSAGE_BASE as memory usage.
        coder->mem_in_use = 0;
        coder->mem_cached = 0;
        coder->mem_next_block = 0;

        coder->progress_in = 0;
        coder->progress_out = 0;

        coder->sequence = SEQ_STREAM_HEADER;
        coder->thread_error = LZMA_OK;
        coder->pending_error = LZMA_OK;
        coder->thr = NULL;

        coder->timeout = options->timeout;

        coder->memlimit_threading = my_max(1, options->memlimit_threading);
        coder->memlimit_stop = my_max(1, options->memlimit_stop);
        if (coder->memlimit_threading > coder->memlimit_stop)
                coder->memlimit_threading = coder->memlimit_stop;

        coder->tell_no_check = (options->flags & LZMA_TELL_NO_CHECK) != 0;
        coder->tell_unsupported_check
                        = (options->flags & LZMA_TELL_UNSUPPORTED_CHECK) != 0;
        coder->tell_any_check = (options->flags & LZMA_TELL_ANY_CHECK) != 0;
        coder->ignore_check = (options->flags & LZMA_IGNORE_CHECK) != 0;
        coder->concatenated = (options->flags & LZMA_CONCATENATED) != 0;
        coder->fail_fast = (options->flags & LZMA_FAIL_FAST) != 0;

        coder->first_stream = true;
        coder->out_was_filled = false;
        coder->pos = 0;

        coder->threads_max = options->threads;

        return_if_error(lzma_outq_init(&coder->outq, allocator,
                                       coder->threads_max));

        return stream_decoder_reset(coder, allocator);
}


extern LZMA_API(lzma_ret)
lzma_stream_decoder_mt(lzma_stream *strm, const lzma_mt *options)
{
        lzma_next_strm_init(stream_decoder_mt_init, strm, options);

        strm->internal->supported_actions[LZMA_RUN] = true;
        strm->internal->supported_actions[LZMA_FINISH] = true;

        return LZMA_OK;
}