Userspace FILE*-flavored driver: multi-threaded streaming and the seekable FILE* open helper. More...

#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include "../../include/zxc_buffer.h"
#include "../../include/zxc_dict.h"
#include "../../include/zxc_error.h"
#include "../../include/zxc_seekable.h"
#include "../../include/zxc_stream.h"
#include "zxc_internal.h"
#include "zxc_threads.h"
#include <errno.h>
#include <sys/stat.h>

Macros
#define	ZXC_STREAM_PREAD_MIN_BLOCK ((size_t)64 * 1024)

#define	ZXC_STREAM_BATCH_BYTES ((size_t)256 * 1024)

#define	ZXC_STREAM_RING_BATCHED 2U

Enumerations
enum	job_status_t { JOB_STATUS_FREE , JOB_STATUS_FILLED , JOB_STATUS_PROCESSED }
	Represents the lifecycle states of a processing job within the ring buffer. More...

Functions
static int	zxc_ring_next (const zxc_stream_ctx_t *ctx, const int idx)
	The ring slot after `idx`.

static int64_t	zxc_pread_full (const int fd, void *dst, const size_t len, const uint64_t offset)
	Thread-safe `pread` of `len` bytes, retrying `EINTR` and partial transfers.

static int	zxc_stream_regular_fd (FILE *f)
	Descriptor of a regular-file stream, -1 for anything else (may be live).

static void	zxc_stream_open_positioned (zxc_stream_ctx_t ctx, FILE f_in, const int fd)
	Switches a regular file to positioned reads from its current position.

static int64_t	zxc_stream_read_at (const zxc_stream_ctx_t ctx, void dst, const size_t len, const uint64_t offset)
	Positioned read of the input, safe from any thread.

static int64_t	zxc_stream_in_read (zxc_stream_ctx_t ctx, FILE f, void *dst, const size_t len)
	Reads up to `len` input bytes on the reader thread: by position while the engine reads positioned, through stdio otherwise.

static void	zxc_stream_reader_fail (zxc_stream_ctx_t *ctx, const int code)
	Records a reader failure; workers write `fail_code` under the same lock.

static int	zxc_stream_process (const zxc_stream_ctx_t ctx, zxc_cctx_t cctx, const uint8_t in, const size_t in_sz, uint8_t out, const size_t out_cap, const uint64_t block_index)
	Runs one chunk through the codec the stream's mode selects.

static int	zxc_stream_decode_job (const zxc_stream_ctx_t ctx, zxc_cctx_t cctx, const zxc_stream_job_t job, const uint8_t in, uint8_t *dict_work)
	Decodes a job's blocks into its output.

static void *	zxc_stream_worker (void *arg)
	Worker thread: pull a job, process it, hand it to the writer.

static void *	zxc_async_writer (void *arg)
	Asynchronous writer thread function.

static void	zxc_stream_engine_stop (zxc_stream_engine_t *e, const int writer_started)
	Stops the threads, then frees the ring and its synchronisation objects.

static int	zxc_stream_engine_start (zxc_stream_engine_t e, FILE f_in, const int n_threads)
	Allocates the job ring and starts the workers and the writer.

static int	zxc_stream_engine_close (zxc_stream_engine_t e, FILE f_in)
	Stops a decompression engine; moves the `FILE*` to where positioned reads ended.

static void	zxc_stream_engine_drain (zxc_stream_engine_t *e)
	Waits until the writer has written every job queued so far.

static size_t	zxc_stream_read_blocks (zxc_stream_ctx_t ctx, FILE f_in, zxc_stream_job_t job, int n_blocks, int *eof)
	Fills a decompression job with up to `ctx->batch` consecutive blocks.

static size_t	zxc_stream_read_chunk (zxc_stream_ctx_t ctx, FILE f_in, zxc_stream_job_t job, int eof)
	Fills a compression job with the next chunk of the source.

static int	zxc_stream_read_loop (zxc_stream_ctx_t ctx, FILE f_in, int read_idx)
	Reads the input and feeds the ring until the stream ends.

static void	zxc_stream_finish_compress (zxc_stream_ctx_t ctx, writer_args_t w, FILE *f_out)
	Closes a compressed stream: EOF block, optional seek table, footer.

static void	zxc_stream_finish_decompress (zxc_stream_ctx_t ctx, const uint64_t frame_out, FILE f_in)
	Consumes and validates the trailer of a decompressed stream.

static int	zxc_stream_read_exact (zxc_stream_ctx_t ctx, FILE f, void *dst, const size_t len, const int short_code)
	Reads exactly `len` bytes, telling a clean end from a short read.

static int	zxc_stream_read_frame_header (zxc_stream_ctx_t ctx, FILE f_in, const uint8_t magic, const uint32_t dict_id, size_t chunk, int has_checksum, int has_seek)
	Reads and checks a frame header whose magic word is already read.

static int64_t	zxc_stream_decode_frame (zxc_stream_engine_t e, FILE f_in, const int has_seek)
	Decodes one frame's blocks, SEK block and footer through a started engine.

static int	zxc_file_span (FILE f, long long saved, uint64_t *size)
	The size of `f`; its position goes to `saved`, for the caller to restore.

int64_t	zxc_stream_compress (FILE f_in, FILE f_out, const zxc_compress_opts_t *opts)
	Compresses a `FILE` stream to another `FILE` stream.

int64_t	zxc_stream_decompress (FILE f_in, FILE f_out, const zxc_decompress_opts_t *opts)
	Decompresses a `FILE` stream to another `FILE` stream.

static int	zxc_file_scan_read (const zxc_scan_src_t src, const uint64_t off, void dst, const size_t len)
	zxc_scan_src_t over a `FILE*`.

static int	zxc_stream_read_frame_info (FILE f_in, zxc_frame_info_t info)
	Reads the header and the end of the file in `f_in`, from offset 0, and validates the frame they describe; restores the stream position.

int64_t	zxc_stream_get_decompressed_size (FILE *f_in)
	Reads the total decompressed size from the footers of a file.

int	zxc_stream_get_frame_info (FILE f_in, zxc_frame_info_t info, const size_t info_size)
	Reads a file's frame header and footer, without decoding.

int	zxc_stream_get_last_frame_info (FILE f_in, const uint64_t end, zxc_frame_info_t info, const size_t info_size)
	Reads the frame that ends at offset `end` of a file, without decoding.

static int64_t	zxc_stdio_read_at (void vctx, void dst, size_t len, uint64_t offset)
	Thread-safe positioned read backing the seekable `FILE*` reader.

zxc_seekable *	zxc_seekable_open_file (FILE *f)
	Opens a seekable archive backed by an open `FILE*`.

Detailed Description

Userspace FILE*-flavored driver: multi-threaded streaming and the seekable FILE* open helper.

Two distinct subsystems live in this translation unit because they share the same userspace-only host requirements (<stdio.h>, threading, and platform file-descriptor extraction): keeping them together means a single TU to exclude when building for kernel / freestanding targets.

Streaming engine: a ring-buffer producer / worker / consumer pipeline that parallelises block processing over FILE* streams. Public API: zxc_stream_compress, zxc_stream_decompress, zxc_stream_get_decompressed_size.
Seekable FILE* wrapper: builds a zxc_reader_t whose read_at uses pread / ReadFile on the file descriptor extracted from a FILE*, then delegates to zxc_seekable_open_reader. Public API: zxc_seekable_open_file.

Definition in file zxc_driver.c.

Macro Definition Documentation

◆ ZXC_STREAM_BATCH_BYTES

#define ZXC_STREAM_BATCH_BYTES ((size_t)256 * 1024)

Definition at line 66 of file zxc_driver.c.

◆ ZXC_STREAM_PREAD_MIN_BLOCK

#define ZXC_STREAM_PREAD_MIN_BLOCK ((size_t)64 * 1024)

Definition at line 62 of file zxc_driver.c.

◆ ZXC_STREAM_RING_BATCHED

#define ZXC_STREAM_RING_BATCHED 2U

Definition at line 69 of file zxc_driver.c.

Enumeration Type Documentation

◆ job_status_t

enum job_status_t

Represents the lifecycle states of a processing job within the ring buffer.

Enumerator
JOB_STATUS_FREE	The job slot is empty and available to be filled with new data by the writer.
JOB_STATUS_FILLED	The job slot has been populated with input data and is ready for processing by a worker.
JOB_STATUS_PROCESSED	The worker has finished processing the data; the result is ready to be consumed/written out.

Definition at line 91 of file zxc_driver.c.

Function Documentation

◆ zxc_async_writer()

static void * zxc_async_writer ( void * arg )

static

Asynchronous writer thread function.

This function runs as a separate thread responsible for writing processed data chunks to the output file. It operates on a ring buffer of jobs shared with the reader and worker threads.

Ordering Enforcement: The writer MUST write blocks in the exact order they were read. Even if worker threads finish jobs out of order (e.g., job 2 finishes before job 1), the writer waits for ctx->write_idx (job 1) to be JOB_STATUS_PROCESSED.

Workflow:

Wait: Sleeps on cond_writer until the job at ctx->write_idx is ready.
Write: Writes the out_buf to the file.
Release: Sets the job status to JOB_STATUS_FREE and signals cond_reader, allowing the main thread to reuse this slot for new input.
Advance: Increments ctx->write_idx to wait for the next sequential block.

Parameters

[in] arg Pointer to a writer_args_t structure containing the stream context, the output file handle, and a counter for total bytes written.

Returns: Always returns NULL.

Definition at line 617 of file zxc_driver.c.

References JOB_STATUS_FREE, JOB_STATUS_PROCESSED, ZXC_BLOCK_CHECKSUM_SIZE, zxc_digest_combine(), ZXC_ERROR_MEMORY, zxc_le32(), ZXC_REALLOC, and zxc_ring_next().

Referenced by zxc_stream_engine_start().

◆ zxc_file_scan_read()

static int zxc_file_scan_read	(	const zxc_scan_src_t *	src,
		const uint64_t	off,
		void *	dst,
		const size_t	len
	)

static

zxc_scan_src_t over a FILE*.

Definition at line 1387 of file zxc_driver.c.

References ZXC_ERROR_IO, and ZXC_OK.

Referenced by zxc_stream_get_decompressed_size(), zxc_stream_get_last_frame_info(), and zxc_stream_read_frame_info().

◆ zxc_file_span()

static int zxc_file_span	(	FILE *	f,
		long long *	saved,
		uint64_t *	size
	)

static

The size of f; its position goes to saved, for the caller to restore.

Definition at line 1205 of file zxc_driver.c.

References ZXC_ERROR_IO, and ZXC_OK.

Referenced by zxc_stream_compress(), zxc_stream_get_decompressed_size(), zxc_stream_get_last_frame_info(), and zxc_stream_read_frame_info().

◆ zxc_pread_full()

static int64_t zxc_pread_full	(	const int	fd,
		void *	dst,
		const size_t	len,
		const uint64_t	offset
	)

static

Thread-safe pread of len bytes, retrying EINTR and partial transfers.

A 32-bit off_t rejects offsets past 2 GiB.

Returns: Bytes read, short only at the end of the file, or -1 on error.

Definition at line 336 of file zxc_driver.c.

Referenced by zxc_stdio_read_at(), and zxc_stream_read_at().

◆ zxc_ring_next()

static int zxc_ring_next	(	const zxc_stream_ctx_t *	ctx,
		const int	idx
	)

inlinestatic

The ring slot after idx.

Definition at line 270 of file zxc_driver.c.

Referenced by zxc_async_writer(), zxc_stream_read_loop(), and zxc_stream_worker().

◆ zxc_stdio_read_at()

static int64_t zxc_stdio_read_at	(	void *	vctx,
		void *	dst,
		size_t	len,
		uint64_t	offset
	)

static

Thread-safe positioned read backing the seekable FILE* reader.

POSIX implementation: zxc_pread_full, which carries its own offset and so is safe to call concurrently from multiple worker threads on one descriptor.

Parameters

[in]	vctx	`zxc_stdio_ctx_t` carrying the file descriptor.
[out]	dst	Destination buffer (at least `len` bytes).
[in]	len	Number of bytes to read.
[in]	offset	Absolute byte offset to read from.

Returns: len on a full read, otherwise ZXC_ERROR_IO.

Definition at line 1530 of file zxc_driver.c.

References ZXC_ERROR_IO, and zxc_pread_full().

Referenced by zxc_seekable_open_file().

◆ zxc_stream_decode_frame()

static int64_t zxc_stream_decode_frame	(	zxc_stream_engine_t *	e,
		FILE *	f_in,
		const int	has_seek
	)

static

Decodes one frame's blocks, SEK block and footer through a started engine.

Returns: Bytes the frame produced, or a negative zxc_error_t.

Definition at line 1181 of file zxc_driver.c.

References ZXC_ERROR_IO, ZXC_FILE_HEADER_SIZE, zxc_stream_engine_drain(), zxc_stream_finish_decompress(), and zxc_stream_read_loop().

Referenced by zxc_stream_decompress().

◆ zxc_stream_decode_job()

static int zxc_stream_decode_job	(	const zxc_stream_ctx_t *	ctx,
		zxc_cctx_t *	cctx,
		const zxc_stream_job_t *	job,
		const uint8_t *	in,
		uint8_t *	dict_work
	)

static

Decodes a job's blocks into its output.

Headers are re-validated: with positioned reads the file may have changed since the reader measured the job.

Returns: Bytes decoded, or a negative zxc_error_t.

Definition at line 457 of file zxc_driver.c.

References zxc_block_header_t::comp_size, ZXC_BLOCK_CHECKSUM_SIZE, ZXC_BLOCK_HEADER_SIZE, ZXC_DECOMPRESS_TAIL_PAD, ZXC_ERROR_CORRUPT_DATA, ZXC_OK, zxc_read_block_header(), and zxc_stream_process().

Referenced by zxc_stream_worker().

◆ zxc_stream_engine_close()

static int zxc_stream_engine_close	(	zxc_stream_engine_t *	e,
		FILE *	f_in
	)

static

Stops a decompression engine; moves the FILE* to where positioned reads ended.

Definition at line 847 of file zxc_driver.c.

References ZXC_ERROR_IO, ZXC_OK, and zxc_stream_engine_stop().

Referenced by zxc_stream_decompress().

◆ zxc_stream_engine_drain()

static void zxc_stream_engine_drain ( zxc_stream_engine_t * e )

static

Waits until the writer has written every job queued so far.

Definition at line 855 of file zxc_driver.c.

References JOB_STATUS_FREE.

Referenced by zxc_stream_decode_frame().

◆ zxc_stream_engine_start()

static int zxc_stream_engine_start	(	zxc_stream_engine_t *	e,
		FILE *	f_in,
		const int	n_threads
	)

static

Allocates the job ring and starts the workers and the writer.

Architecture: Producer-Consumer with Ring Buffer

Ring Buffer: A fixed-size array of zxc_stream_job_t structures.
Producer (Main Thread): Reads chunks from f_in and fills "Free" slots in the ring buffer. It blocks if no slots are free (backpressure).
Workers: Pick up "Filled" jobs from a queue, process them, and mark them as "Processed".
Consumer (Writer Thread): Waits for the next sequential job to be "Processed", writes it to f_out, and marks the slot as "Free".

Double-Buffering & Zero-Copy: We allocate alloc_in and alloc_out buffers for each job. The reader reads directly into in_buf, and the writer writes directly from out_buf, minimizing memory copies.

Regular-file Input (decompression): Small blocks travel several to a job (ZXC_STREAM_BATCH_BYTES); with two threads or more, larger ones are read by the workers and all n_threads decode. Pipes and memory streams keep one block per job through fread.

The caller fills the run's fields of e->ctx (mode, level, block size, checksums, dictionary, progress) and of e->w (output, seek table, counters).

Parameters

[in,out]	e	The engine, its run fields filled.
[in]	f_in	The input stream.
[in]	n_threads	Thread count; 0 or less auto-detects the online processors.

Returns: ZXC_OK, or ZXC_ERROR_MEMORY with nothing left to free.

Definition at line 764 of file zxc_driver.c.

References JOB_STATUS_FREE, ZXC_ALIGNED_MALLOC, ZXC_ALIGNMENT_MASK, zxc_async_writer(), ZXC_CACHE_LINE_SIZE, zxc_compress_bound(), ZXC_DECOMPRESS_TAIL_PAD, ZXC_ERROR_MEMORY, ZXC_MALLOC, ZXC_MAX_THREADS, zxc_num_procs(), ZXC_OK, ZXC_PAD_SIZE, zxc_stream_engine_stop(), zxc_stream_open_positioned(), zxc_stream_regular_fd(), and zxc_stream_worker().

Referenced by zxc_stream_compress(), and zxc_stream_decompress().

◆ zxc_stream_engine_stop()

static void zxc_stream_engine_stop	(	zxc_stream_engine_t *	e,
		const int	writer_started
	)

static

Stops the threads, then frees the ring and its synchronisation objects.

The writer is told to stop by a terminator job at read_idx, once the jobs before it are written. writer_started is 0 when setup failed before it.

Definition at line 707 of file zxc_driver.c.

References JOB_STATUS_FREE, JOB_STATUS_PROCESSED, ZXC_ALIGNED_FREE, and ZXC_FREE.

Referenced by zxc_stream_compress(), zxc_stream_engine_close(), and zxc_stream_engine_start().

◆ zxc_stream_finish_compress()

static void zxc_stream_finish_compress	(	zxc_stream_ctx_t *	ctx,
		writer_args_t *	w,
		FILE *	f_out
	)

static

Closes a compressed stream: EOF block, optional seek table, footer.

Runs once the writer thread has drained, so it appends straight to the file.

Definition at line 1021 of file zxc_driver.c.

References zxc_block_header_t::block_type, ZXC_BLOCK_HEADER_SIZE, ZXC_FILE_HEADER_SIZE, ZXC_FOOTER_MAX_SIZE_WITH_DIGEST, ZXC_SEEK_GROUP, ZXC_SEEK_GROUP_BYTES, zxc_seek_group_len(), zxc_seek_table_header(), zxc_seek_write_group(), zxc_write_block_header(), and zxc_write_file_footer().

Referenced by zxc_stream_compress().

◆ zxc_stream_finish_decompress()

static void zxc_stream_finish_decompress	(	zxc_stream_ctx_t *	ctx,
		const uint64_t	frame_out,
		FILE *	f_in
	)

static

Consumes and validates the trailer of a decompressed stream.

Definition at line 1069 of file zxc_driver.c.

References zxc_footer_layout_t::len, ZXC_BLOCK_HEADER_SIZE, zxc_check_file_footer(), zxc_check_seek_header(), ZXC_ERROR_BAD_CHECKSUM, ZXC_ERROR_CORRUPT_DATA, ZXC_ERROR_IO, zxc_footer_layout(), ZXC_FOOTER_MAX_SIZE_WITH_DIGEST, ZXC_OK, and zxc_stream_in_read().

Referenced by zxc_stream_decode_frame().

◆ zxc_stream_in_read()

static int64_t zxc_stream_in_read	(	zxc_stream_ctx_t *	ctx,
		FILE *	f,
		void *	dst,
		const size_t	len
	)

static

Reads up to len input bytes on the reader thread: by position while the engine reads positioned, through stdio otherwise.

Parameters

[in]	ctx	Started engine, or NULL to read through stdio.
[in]	f	The input stream.
[out]	dst	Destination.
[in]	len	Bytes wanted.

Returns: Bytes read, short only at the end of the input, or ZXC_ERROR_IO.

Definition at line 414 of file zxc_driver.c.

References ZXC_ERROR_IO, and zxc_stream_read_at().

Referenced by zxc_stream_finish_decompress(), zxc_stream_read_blocks(), and zxc_stream_read_exact().

◆ zxc_stream_open_positioned()

static void zxc_stream_open_positioned	(	zxc_stream_ctx_t *	ctx,
		FILE *	f_in,
		const int	fd
	)

static

Switches a regular file to positioned reads from its current position.

Definition at line 370 of file zxc_driver.c.

Referenced by zxc_stream_engine_start().

◆ zxc_stream_process()

static int zxc_stream_process	(	const zxc_stream_ctx_t *	ctx,
		zxc_cctx_t *	cctx,
		const uint8_t *	in,
		const size_t	in_sz,
		uint8_t *	out,
		const size_t	out_cap,
		const uint64_t	block_index
	)

inlinestatic

Runs one chunk through the codec the stream's mode selects.

Both codecs are called by name: the compressor takes a mutable context, the decoder a const one, so no single function pointer type fits both.

Definition at line 441 of file zxc_driver.c.

References zxc_compress_chunk_wrapper(), and zxc_decompress_chunk_wrapper().

Referenced by zxc_stream_decode_job(), and zxc_stream_worker().

◆ zxc_stream_read_at()

static int64_t zxc_stream_read_at	(	const zxc_stream_ctx_t *	ctx,
		void *	dst,
		const size_t	len,
		const uint64_t	offset
	)

static

Positioned read of the input, safe from any thread.

Returns: Bytes read, short only at the end of the file, or ZXC_ERROR_IO.

Definition at line 390 of file zxc_driver.c.

References ZXC_ERROR_IO, and zxc_pread_full().

Referenced by zxc_stream_in_read(), zxc_stream_read_blocks(), and zxc_stream_worker().

◆ zxc_stream_read_blocks()

static size_t zxc_stream_read_blocks	(	zxc_stream_ctx_t *	ctx,
		FILE *	f_in,
		zxc_stream_job_t *	job,
		int *	n_blocks,
		int *	eof
	)

static

Fills a decompression job with up to ctx->batch consecutive blocks.

A failure sets io_error and drops the blocks gathered: nothing is queued.

Returns: Bytes the job holds, 0 when there is nothing to queue.

Definition at line 873 of file zxc_driver.c.

References zxc_block_header_t::block_type, zxc_block_header_t::comp_size, ZXC_BLOCK_CHECKSUM_SIZE, ZXC_BLOCK_HEADER_SIZE, zxc_check_eof_header(), zxc_digest_combine(), ZXC_ERROR_BAD_BLOCK_SIZE, ZXC_ERROR_BAD_HEADER, ZXC_ERROR_CORRUPT_DATA, zxc_le32(), ZXC_OK, zxc_read_block_header(), zxc_stream_in_read(), zxc_stream_read_at(), and zxc_stream_reader_fail().

Referenced by zxc_stream_read_loop().

◆ zxc_stream_read_chunk()

static size_t zxc_stream_read_chunk	(	zxc_stream_ctx_t *	ctx,
		FILE *	f_in,
		zxc_stream_job_t *	job,
		int *	eof
	)

static

Fills a compression job with the next chunk of the source.

Definition at line 969 of file zxc_driver.c.

Referenced by zxc_stream_read_loop().

◆ zxc_stream_read_exact()

static int zxc_stream_read_exact	(	zxc_stream_ctx_t *	ctx,
		FILE *	f,
		void *	dst,
		const size_t	len,
		const int	short_code
	)

static

Reads exactly len bytes, telling a clean end from a short read.

Parameters

[in]	ctx	Started engine, or NULL before the first frame.
[in]	f	The input stream.
[out]	dst	Destination.
[in]	len	Bytes to read.
[in]	short_code	Returned when the input ends inside the range.

Returns: ZXC_OK; 1 when the input ended before the first byte; short_code when it ended inside the range; ZXC_ERROR_IO on a read error.

Definition at line 1135 of file zxc_driver.c.

References ZXC_ERROR_IO, ZXC_OK, and zxc_stream_in_read().

Referenced by zxc_stream_decompress(), and zxc_stream_read_frame_header().

◆ zxc_stream_read_frame_header()

static int zxc_stream_read_frame_header	(	zxc_stream_ctx_t *	ctx,
		FILE *	f_in,
		const uint8_t *	magic,
		const uint32_t	dict_id,
		size_t *	chunk,
		int *	has_checksum,
		int *	has_seek
	)

static

Reads and checks a frame header whose magic word is already read.

Parameters

[in]	ctx	Started engine, or NULL before the first frame.
[in]	f_in	The input stream, just past the magic word.
[in]	magic	The magic word's 4 bytes.
[in]	dict_id	Id of the caller's dictionary, 0 without one.
[out]	chunk	The frame's block size.
[out]	has_checksum	Whether the frame carries checksums.
[out]	has_seek	Whether it ends with a seek table.

Returns: ZXC_OK, or a negative zxc_error_t.

Definition at line 1155 of file zxc_driver.c.

References ZXC_ERROR_DICT_MISMATCH, ZXC_ERROR_DICT_REQUIRED, ZXC_ERROR_SRC_TOO_SMALL, ZXC_FILE_HEADER_SIZE, ZXC_OK, zxc_read_file_header(), and zxc_stream_read_exact().

Referenced by zxc_stream_decompress().

◆ zxc_stream_read_frame_info()

static int zxc_stream_read_frame_info	(	FILE *	f_in,
		zxc_frame_info_t *	info
	)

static

Reads the header and the end of the file in f_in, from offset 0, and validates the frame they describe; restores the stream position.

Definition at line 1399 of file zxc_driver.c.

References zxc_file_scan_read(), zxc_file_span(), ZXC_OK, and zxc_scan_whole_frame().

Referenced by zxc_stream_get_frame_info().

◆ zxc_stream_read_loop()

static int zxc_stream_read_loop	(	zxc_stream_ctx_t *	ctx,
		FILE *	f_in,
		int	read_idx
	)

static

Reads the input and feeds the ring until the stream ends.

Fills the ring from slot read_idx on; returns the slot after the last job.

Definition at line 982 of file zxc_driver.c.

References JOB_STATUS_FILLED, JOB_STATUS_FREE, zxc_ring_next(), zxc_stream_read_blocks(), and zxc_stream_read_chunk().

Referenced by zxc_stream_compress(), and zxc_stream_decode_frame().

◆ zxc_stream_reader_fail()

static void zxc_stream_reader_fail	(	zxc_stream_ctx_t *	ctx,
		const int	code
	)

static

Records a reader failure; workers write fail_code under the same lock.

Definition at line 428 of file zxc_driver.c.

Referenced by zxc_stream_read_blocks().

◆ zxc_stream_regular_fd()

static int zxc_stream_regular_fd ( FILE * f )

static

Descriptor of a regular-file stream, -1 for anything else (may be live).

Definition at line 354 of file zxc_driver.c.

Referenced by zxc_stream_engine_start().

◆ zxc_stream_worker()

static void * zxc_stream_worker ( void * arg )

static

Worker thread: pull a job, process it, hand it to the writer.

Sleeps on cond_worker until worker_queue has a job, then runs zxc_stream_process over it and marks it JOB_STATUS_PROCESSED. Each worker owns a thread-local zxc_cctx_t so the parallel part never touches a shared context. With positioned reads it also reads each job's blocks itself.

The writer is signalled only when the finished job is the one it is waiting for (jid == ctx->write_idx). Jobs completing out of order stay silent, which keeps the writer from waking up on work it cannot yet emit.

Parameters

[in] arg The shared zxc_stream_ctx_t.

Returns: Always NULL.

Definition at line 504 of file zxc_driver.c.

References zxc_cctx_t::checksum_enabled, zxc_cctx_t::compression_level, zxc_cctx_t::dict_buffer, zxc_cctx_t::dict_huf, JOB_STATUS_PROCESSED, zxc_block_size_ceil(), zxc_cctx_attach_dict_huf(), zxc_cctx_free(), zxc_cctx_init(), ZXC_ERROR_IO, ZXC_FREE, ZXC_MALLOC, ZXC_OK, zxc_ring_next(), zxc_stream_decode_job(), zxc_stream_process(), and zxc_stream_read_at().

Referenced by zxc_stream_engine_start().

Macros

Enumerations

Functions

Detailed Description

Macro Definition Documentation

◆ ZXC_STREAM_BATCH_BYTES

◆ ZXC_STREAM_PREAD_MIN_BLOCK

◆ ZXC_STREAM_RING_BATCHED

Enumeration Type Documentation

◆ job_status_t

Function Documentation

◆ zxc_async_writer()

◆ zxc_file_scan_read()

◆ zxc_file_span()

◆ zxc_pread_full()

◆ zxc_ring_next()

◆ zxc_stdio_read_at()

◆ zxc_stream_decode_frame()

◆ zxc_stream_decode_job()

◆ zxc_stream_engine_close()

◆ zxc_stream_engine_drain()

◆ zxc_stream_engine_start()

◆ zxc_stream_engine_stop()

◆ zxc_stream_finish_compress()

◆ zxc_stream_finish_decompress()

◆ zxc_stream_in_read()

◆ zxc_stream_open_positioned()

◆ zxc_stream_process()

◆ zxc_stream_read_at()

◆ zxc_stream_read_blocks()

◆ zxc_stream_read_chunk()

◆ zxc_stream_read_exact()

◆ zxc_stream_read_frame_header()

◆ zxc_stream_read_frame_info()

◆ zxc_stream_read_loop()

◆ zxc_stream_reader_fail()

◆ zxc_stream_regular_fd()

◆ zxc_stream_worker()