meta: add compress module for (de)compressing data

CMakeLists.txt

@@ -5,9 +5,11 @@ include (TestBigEndian)
 
 set(CMAKE_C_STANDARD 99)
 set(CMAKE_C_EXTENSIONS OFF)
+
+set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} ${CMAKE_SOURCE_DIR}/cmake)
 set_property(GLOBAL PROPERTY USE_FOLDERS ON)
 
-set(b_modules core object serial term cmd io)
+set(b_modules core object serial term cmd io compress)
 
 set(b_system_name ${CMAKE_SYSTEM_NAME})
 string(TOLOWER ${b_system_name} b_system_name)

cmake/FindZSTD.cmake

@@ -0,0 +1,21 @@
+# taken from: https://github.com/facebook/folly/blob/master/CMake/FindZstd.cmake
+# SPDX-FileCopyrightText: Facebook, Inc. and its affiliates.
+# SPDX-License-Identifier: Apache-2.0
+#
+# - Try to find Facebook zstd library
+# This will define
+# ZSTD_FOUND
+# ZSTD_INCLUDE_DIR
+# ZSTD_LIBRARY
+#
+
+find_path(ZSTD_INCLUDE_DIR NAMES zstd.h)
+find_library(ZSTD_LIBRARY NAMES zstd)
+
+include(FindPackageHandleStandardArgs)
+FIND_PACKAGE_HANDLE_STANDARD_ARGS(
+    ZSTD DEFAULT_MSG
+    ZSTD_LIBRARY ZSTD_INCLUDE_DIR
+)
+
+mark_as_advanced(ZSTD_INCLUDE_DIR ZSTD_LIBRARY)

compress/CMakeLists.txt

@@ -0,0 +1,19 @@
+include(../cmake/Templates.cmake)
+
+find_package(ZSTD)
+
+if (ZSTD_FOUND)
+	set(libs ${libs} ${ZSTD_LIBRARY})
+	set(include_dirs ${include_dirs} ${ZSTD_INCLUDE_DIR})
+	set(function_sources ${function_sources} ${CMAKE_CURRENT_SOURCE_DIR}/function/zstd.c)
+	set(defines ${defines} B_COMPRESSOR_SUPPORTED_ZSTD)
+	message(STATUS "Enabling ZSTD support in blue-compress")
+endif ()
+
+add_bluelib_module(
+	NAME compress
+	DEPENDENCIES core
+	EXTRA_SOURCES ${function_sources}
+	DEFINES ${defines}
+	LIBS ${libs}
+	INCLUDE_DIRS ${include_dirs})

compress/compressor.c

@@ -0,0 +1,131 @@
+#include "compressor.h"
+
+#include "function.h"
+
+#include <blue/compress/compressor.h>
+#include <blue/core/ringbuffer.h>
+#include <stdlib.h>
+#include <string.h>
+
+enum b_status b_compressor_create(
+	const struct b_compression_function *func, enum b_compressor_mode mode,
+	struct b_ringbuffer *inbuf, struct b_ringbuffer *outbuf,
+	struct b_compressor **out)
+{
+	size_t ctx_size = sizeof(struct b_compressor) + func->f_ctx_size;
+	struct b_compressor *compressor = malloc(ctx_size);
+	if (!compressor) {
+		return B_ERR_NO_MEMORY;
+	}
+
+	memset(compressor, 0x0, ctx_size);
+
+	compressor->c_func = func;
+	compressor->c_in = inbuf;
+	compressor->c_out = outbuf;
+	compressor->c_mode = mode;
+
+	enum b_status status = B_SUCCESS;
+
+	if (func->f_init) {
+		status = func->f_init(compressor);
+	}
+
+	if (!B_OK(status)) {
+		free(compressor);
+		compressor = NULL;
+	}
+
+	*out = compressor;
+	return status;
+}
+
+enum b_status b_compressor_destroy(struct b_compressor *compressor)
+{
+	enum b_status status = B_SUCCESS;
+
+	if (compressor->c_func->f_fini) {
+		status = compressor->c_func->f_fini(compressor);
+	}
+
+	if (!B_OK(status)) {
+		return status;
+	}
+
+	free(compressor);
+
+	return B_SUCCESS;
+}
+
+enum b_status b_compressor_compress(struct b_compressor *compressor)
+{
+	if (compressor->c_mode != B_COMPRESSOR_MODE_COMPRESS) {
+		return B_ERR_BAD_STATE;
+	}
+
+	if (!compressor->c_func->f_compress) {
+		return B_ERR_NOT_SUPPORTED;
+	}
+
+	return compressor->c_func->f_compress(compressor);
+}
+
+enum b_status b_compressor_compress_end(struct b_compressor *compressor)
+{
+	if (compressor->c_mode != B_COMPRESSOR_MODE_COMPRESS) {
+		return B_ERR_BAD_STATE;
+	}
+
+	if (!compressor->c_func->f_compress_end) {
+		return B_ERR_NOT_SUPPORTED;
+	}
+
+	while (b_ringbuffer_available_data_remaining(compressor->c_in)) {
+		if (!b_ringbuffer_write_capacity_remaining(compressor->c_out)) {
+			return B_ERR_NO_SPACE;
+		}
+
+		enum b_status status = b_compressor_compress(compressor);
+		if (!B_OK(status)) {
+			return status;
+		}
+	}
+
+	return compressor->c_func->f_compress_end(compressor);
+}
+
+enum b_status b_compressor_decompress(struct b_compressor *compressor)
+{
+	if (compressor->c_mode != B_COMPRESSOR_MODE_DECOMPRESS) {
+		return B_ERR_BAD_STATE;
+	}
+
+	if (!compressor->c_func->f_decompress) {
+		return B_ERR_NOT_SUPPORTED;
+	}
+
+	return compressor->c_func->f_decompress(compressor);
+}
+
+enum b_status b_compressor_reset(struct b_compressor *compressor)
+{
+	compressor->c_flags &= ~COMPRESSOR_EOF;
+
+	if (compressor->c_func->f_reset) {
+		return compressor->c_func->f_reset(compressor);
+	}
+
+	return B_SUCCESS;
+}
+
+bool b_compressor_eof(const struct b_compressor *compressor)
+{
+	return (compressor->c_flags & COMPRESSOR_EOF) != 0;
+}
+
+void *b_compressor_get_function_ctx(struct b_compressor *compressor)
+{
+	unsigned char *p = (unsigned char *)compressor;
+	p += sizeof *compressor;
+	return p;
+}

compress/compressor.h

@@ -0,0 +1,22 @@
+#ifndef _COMPRESSOR_H_
+#define _COMPRESSOR_H_
+
+#include <blue/compress/compressor.h>
+
+struct b_compression_function;
+struct b_ringbuffer;
+
+enum compressor_flags {
+	COMPRESSOR_EOF = 0x01u,
+};
+
+struct b_compressor {
+	enum compressor_flags c_flags;
+	enum b_compressor_mode c_mode;
+	const struct b_compression_function *c_func;
+	struct b_ringbuffer *c_in, *c_out;
+};
+
+extern void *b_compressor_get_function_ctx(struct b_compressor *compressor);
+
+#endif

compress/function.c

@@ -0,0 +1,48 @@
+#include "function.h"
+
+#include <blue/compress/compressor.h>
+#include <blue/compress/function.h>
+
+#ifdef B_COMPRESSOR_SUPPORTED_ZSTD
+extern const struct b_compression_function z__b_compression_function_zstd;
+#endif
+
+static const struct b_compression_function *compressor_functions[] = {
+#ifdef B_COMPRESSOR_SUPPORTED_ZSTD
+	[B_COMPRESSOR_FUNCTION_ZSTD] = &z__b_compression_function_zstd,
+#endif
+};
+static const size_t nr_compressor_functions
+	= sizeof compressor_functions / sizeof compressor_functions[0];
+
+const struct b_compression_function *b_compression_function_get_by_id(
+	enum b_compression_function_id id)
+{
+	if (id < 0 || id >= nr_compressor_functions) {
+		return NULL;
+	}
+
+	return compressor_functions[id];
+}
+
+enum b_status b_compression_function_get_buffer_size(
+	const struct b_compression_function *func, enum b_compressor_mode mode,
+	size_t *inbuf_size, size_t *outbuf_size)
+{
+	if (!func->f_buffer_size) {
+		return B_ERR_NOT_SUPPORTED;
+	}
+
+	size_t in = 0, out = 0;
+	enum b_status status = func->f_buffer_size(mode, &in, &out);
+
+	if (inbuf_size) {
+		*inbuf_size = in;
+	}
+
+	if (outbuf_size) {
+		*outbuf_size = out;
+	}
+
+	return status;
+}

compress/function.h

@@ -0,0 +1,24 @@
+#ifndef _FUNCTION_H_
+#define _FUNCTION_H_
+
+#include <blue/compress/compressor.h>
+#include <blue/core/status.h>
+#include <stddef.h>
+
+struct b_compressor;
+
+struct b_compression_function {
+	const char *f_name;
+	size_t f_ctx_size;
+
+	enum b_status (*f_buffer_size)(enum b_compressor_mode, size_t *, size_t *);
+
+	enum b_status (*f_init)(struct b_compressor *);
+	enum b_status (*f_fini)(struct b_compressor *);
+	enum b_status (*f_reset)(struct b_compressor *);
+	enum b_status (*f_compress)(struct b_compressor *);
+	enum b_status (*f_compress_end)(struct b_compressor *);
+	enum b_status (*f_decompress)(struct b_compressor *);
+};
+
+#endif

compress/function/zstd.c

@@ -0,0 +1,287 @@
+#include "../compressor.h"
+#include "../function.h"
+
+#include <blue/core/ringbuffer.h>
+#include <zstd.h>
+
+struct zstd_ctx {
+	union {
+		ZSTD_CCtx *zstd_c;
+		ZSTD_DCtx *zstd_d;
+	};
+};
+
+static enum b_status buffer_size(
+	enum b_compressor_mode mode, size_t *inbuf_size, size_t *outbuf_size)
+{
+	switch (mode) {
+	case B_COMPRESSOR_MODE_COMPRESS:
+		*inbuf_size = ZSTD_CStreamInSize();
+		*outbuf_size = ZSTD_CStreamOutSize();
+		break;
+	case B_COMPRESSOR_MODE_DECOMPRESS:
+		*inbuf_size = ZSTD_DStreamInSize();
+		*outbuf_size = ZSTD_DStreamOutSize();
+		break;
+	default:
+		return B_ERR_INVALID_ARGUMENT;
+	}
+
+	return B_SUCCESS;
+}
+
+static enum b_status init(struct b_compressor *compressor)
+{
+	struct zstd_ctx *ctx = b_compressor_get_function_ctx(compressor);
+	switch (compressor->c_mode) {
+	case B_COMPRESSOR_MODE_COMPRESS:
+		ctx->zstd_c = ZSTD_createCCtx();
+		if (!ctx->zstd_c) {
+			return B_ERR_NO_MEMORY;
+		}
+		break;
+	case B_COMPRESSOR_MODE_DECOMPRESS:
+		ctx->zstd_d = ZSTD_createDCtx();
+		if (!ctx->zstd_d) {
+			return B_ERR_NO_MEMORY;
+		}
+		break;
+	default:
+		return B_ERR_BAD_STATE;
+	}
+
+	return B_SUCCESS;
+}
+
+static enum b_status fini(struct b_compressor *compressor)
+{
+	struct zstd_ctx *ctx = b_compressor_get_function_ctx(compressor);
+	switch (compressor->c_mode) {
+	case B_COMPRESSOR_MODE_COMPRESS:
+		ZSTD_freeCCtx(ctx->zstd_c);
+		break;
+	case B_COMPRESSOR_MODE_DECOMPRESS:
+		ZSTD_freeDCtx(ctx->zstd_d);
+		break;
+	default:
+		return B_ERR_BAD_STATE;
+	}
+
+	return B_SUCCESS;
+}
+
+static enum b_status reset(struct b_compressor *compressor)
+{
+	struct zstd_ctx *ctx = b_compressor_get_function_ctx(compressor);
+	switch (compressor->c_mode) {
+	case B_COMPRESSOR_MODE_COMPRESS:
+		ZSTD_CCtx_reset(ctx->zstd_c, ZSTD_reset_session_only);
+		break;
+	case B_COMPRESSOR_MODE_DECOMPRESS:
+		ZSTD_DCtx_reset(ctx->zstd_d, ZSTD_reset_session_only);
+		break;
+	default:
+		return B_ERR_BAD_STATE;
+	}
+
+	return B_SUCCESS;
+}
+
+static enum b_status compress(struct b_compressor *compressor)
+{
+	enum b_status status = B_SUCCESS;
+	struct zstd_ctx *ctx = b_compressor_get_function_ctx(compressor);
+	struct b_ringbuffer *in = compressor->c_in;
+	struct b_ringbuffer *out = compressor->c_out;
+
+	if (b_ringbuffer_available_data_remaining(in) == 0) {
+		return B_ERR_NO_DATA;
+	}
+
+	if (b_ringbuffer_write_capacity_remaining(out) == 0) {
+		return B_ERR_NO_SPACE;
+	}
+
+	size_t nr_consumed = 0;
+
+	while (1) {
+		size_t in_available = 0, out_capacity = 0;
+		void *in_buf = NULL, *out_buf = NULL;
+
+		status = b_ringbuffer_open_read_buffer(in, &in_buf, &in_available);
+		if (!B_OK(status)) {
+			break;
+		}
+
+		status = b_ringbuffer_open_write_buffer(
+			out, &out_buf, &out_capacity);
+		if (!B_OK(status)) {
+			b_ringbuffer_close_read_buffer(in, &in_buf, 0);
+			break;
+		}
+
+		ZSTD_inBuffer z_in = {
+			.src = in_buf,
+			.pos = 0,
+			.size = in_available,
+		};
+
+		ZSTD_outBuffer z_out = {
+			.dst = out_buf,
+			.pos = 0,
+			.size = out_capacity,
+		};
+
+		do {
+			size_t ret = ZSTD_compressStream2(
+				ctx->zstd_c, &z_out, &z_in, ZSTD_e_continue);
+			if (ZSTD_isError(ret)) {
+				status = B_ERR_COMPRESSION_FAILURE;
+				break;
+			}
+		} while (z_in.pos < z_in.size && z_out.pos < z_out.size);
+
+		nr_consumed += z_in.pos;
+
+		b_ringbuffer_close_read_buffer(in, &in_buf, z_in.pos);
+		b_ringbuffer_close_write_buffer(out, &out_buf, z_out.pos);
+	}
+
+	if ((status == B_ERR_NO_SPACE || status == B_ERR_NO_DATA)
+	    && nr_consumed > 0) {
+		status = B_SUCCESS;
+	}
+
+	return status;
+}
+
+static enum b_status compress_end(struct b_compressor *compressor)
+{
+	enum b_status status = B_SUCCESS;
+	struct zstd_ctx *ctx = b_compressor_get_function_ctx(compressor);
+
+	struct b_ringbuffer *out = compressor->c_out;
+	if (b_ringbuffer_write_capacity_remaining(out) == 0) {
+		return B_ERR_NO_SPACE;
+	}
+
+	bool finished = false;
+	do {
+		void *out_buf = NULL;
+		size_t out_capacity = 0;
+		status = b_ringbuffer_open_write_buffer(
+			out, &out_buf, &out_capacity);
+		if (!B_OK(status)) {
+			break;
+		}
+
+		ZSTD_inBuffer z_in = {0};
+		ZSTD_outBuffer z_out = {
+			.dst = out_buf,
+			.pos = 0,
+			.size = out_capacity,
+		};
+
+		do {
+			size_t ret = ZSTD_compressStream2(
+				ctx->zstd_c, &z_out, &z_in, ZSTD_e_end);
+			if (ZSTD_isError(ret)) {
+				status = B_ERR_COMPRESSION_FAILURE;
+				finished = true;
+			}
+
+			if (ret == 0) {
+				compressor->c_flags |= COMPRESSOR_EOF;
+				finished = true;
+			}
+		} while (!finished && z_out.pos < z_out.size);
+
+		b_ringbuffer_close_write_buffer(out, &out_buf, z_out.pos);
+	} while (!finished);
+
+	return status;
+}
+
+static enum b_status decompress(struct b_compressor *compressor)
+{
+	enum b_status status = B_SUCCESS;
+	struct zstd_ctx *ctx = b_compressor_get_function_ctx(compressor);
+	struct b_ringbuffer *in = compressor->c_in;
+	struct b_ringbuffer *out = compressor->c_out;
+
+	if (b_ringbuffer_available_data_remaining(in) == 0) {
+		return B_ERR_NO_DATA;
+	}
+
+	if (b_ringbuffer_write_capacity_remaining(out) == 0) {
+		return B_ERR_NO_SPACE;
+	}
+
+	size_t nr_consumed = 0;
+
+	while (!(compressor->c_flags & COMPRESSOR_EOF)) {
+		size_t in_available = 0, out_capacity = 0;
+		void *in_buf = NULL, *out_buf = NULL;
+
+		status = b_ringbuffer_open_read_buffer(in, &in_buf, &in_available);
+		if (!B_OK(status)) {
+			break;
+		}
+
+		status = b_ringbuffer_open_write_buffer(
+			out, &out_buf, &out_capacity);
+		if (!B_OK(status)) {
+			b_ringbuffer_close_read_buffer(in, &in_buf, 0);
+			break;
+		}
+
+		ZSTD_inBuffer z_in = {
+			.src = in_buf,
+			.pos = 0,
+			.size = in_available,
+		};
+
+		ZSTD_outBuffer z_out = {
+			.dst = out_buf,
+			.pos = 0,
+			.size = out_capacity,
+		};
+
+		do {
+			size_t ret = ZSTD_decompressStream(
+				ctx->zstd_d, &z_out, &z_in);
+			if (ZSTD_isError(ret)) {
+				status = B_ERR_COMPRESSION_FAILURE;
+				break;
+			}
+
+			if (ret == 0) {
+				compressor->c_flags |= COMPRESSOR_EOF;
+				break;
+			}
+		} while (z_in.pos < z_in.size && z_out.pos < z_out.size);
+
+		nr_consumed += z_in.pos;
+
+		b_ringbuffer_close_read_buffer(in, &in_buf, z_in.pos);
+		b_ringbuffer_close_write_buffer(out, &out_buf, z_out.pos);
+	}
+
+	if ((status == B_ERR_NO_SPACE || status == B_ERR_NO_SPACE)
+	    && nr_consumed > 0) {
+		status = B_SUCCESS;
+	}
+
+	return status;
+}
+
+const struct b_compression_function z__b_compression_function_zstd = {
+	.f_name = "zstd",
+	.f_buffer_size = buffer_size,
+	.f_init = init,
+	.f_fini = fini,
+	.f_reset = reset,
+	.f_compress = compress,
+	.f_compress_end = compress_end,
+	.f_decompress = decompress,
+};

compress/include/blue/compress/compressor.h

@@ -0,0 +1,33 @@
+#ifndef BLUELIB_COMPRESS_COMPRESSOR_H_
+#define BLUELIB_COMPRESS_COMPRESSOR_H_
+
+#include <blue/core/misc.h>
+#include <blue/core/status.h>
+#include <stdbool.h>
+
+struct b_ringbuffer;
+struct b_compression_function;
+
+typedef struct b_compressor b_compressor;
+
+typedef enum b_compressor_mode {
+	B_COMPRESSOR_MODE_COMPRESS,
+	B_COMPRESSOR_MODE_DECOMPRESS,
+} b_compressor_mode;
+
+BLUE_API b_status b_compressor_create(
+	const struct b_compression_function *func, b_compressor_mode mode,
+	struct b_ringbuffer *inbuf, struct b_ringbuffer *outbuf,
+	b_compressor **out);
+BLUE_API b_status b_compressor_destroy(b_compressor *compressor);
+
+BLUE_API b_status b_compressor_get_buffer_size(
+	const b_compressor *compressor, size_t *inbuf_size, size_t *outbuf_size);
+
+BLUE_API b_status b_compressor_compress(b_compressor *compressor);
+BLUE_API b_status b_compressor_compress_end(b_compressor *compressor);
+BLUE_API b_status b_compressor_decompress(b_compressor *compressor);
+BLUE_API b_status b_compressor_reset(b_compressor *compressor);
+BLUE_API bool b_compressor_eof(const b_compressor *compressor);
+
+#endif

compress/include/blue/compress/function.h

@@ -0,0 +1,22 @@
+#ifndef BLUELIB_COMPRESS_FUNCTION_H_
+#define BLUELIB_COMPRESS_FUNCTION_H_
+
+#include <blue/core/misc.h>
+#include <blue/core/status.h>
+
+enum b_compressor_mode;
+
+typedef struct b_compression_function b_compression_function;
+
+typedef enum b_compression_function_id {
+	B_COMPRESSOR_FUNCTION_NONE = 0,
+	B_COMPRESSOR_FUNCTION_ZSTD,
+} b_compression_function_id;
+
+BLUE_API const b_compression_function *b_compression_function_get_by_id(
+	b_compression_function_id id);
+BLUE_API b_status b_compression_function_get_buffer_size(
+	const b_compression_function *func, enum b_compressor_mode mode,
+	size_t *inbuf_size, size_t *outbuf_size);
+
+#endif