diff --git a/CMakeLists.txt b/CMakeLists.txt index 6cb119c..e5a402f 100644 --- a/CMakeLists.txt +++ b/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) diff --git a/cmake/FindZSTD.cmake b/cmake/FindZSTD.cmake new file mode 100644 index 0000000..d821e46 --- /dev/null +++ b/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) diff --git a/compress/CMakeLists.txt b/compress/CMakeLists.txt new file mode 100644 index 0000000..3707dd1 --- /dev/null +++ b/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}) diff --git a/compress/compressor.c b/compress/compressor.c new file mode 100644 index 0000000..5b483f0 --- /dev/null +++ b/compress/compressor.c @@ -0,0 +1,131 @@ +#include "compressor.h" + +#include "function.h" + +#include +#include +#include +#include + +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; +} diff --git a/compress/compressor.h b/compress/compressor.h new file mode 100644 index 0000000..ce47de4 --- /dev/null +++ b/compress/compressor.h @@ -0,0 +1,22 @@ +#ifndef _COMPRESSOR_H_ +#define _COMPRESSOR_H_ + +#include + +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 diff --git a/compress/cstream.c b/compress/cstream.c new file mode 100644 index 0000000..e69de29 diff --git a/compress/cstream.h b/compress/cstream.h new file mode 100644 index 0000000..e69de29 diff --git a/compress/function.c b/compress/function.c new file mode 100644 index 0000000..5a2cd82 --- /dev/null +++ b/compress/function.c @@ -0,0 +1,48 @@ +#include "function.h" + +#include +#include + +#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; +} diff --git a/compress/function.h b/compress/function.h new file mode 100644 index 0000000..7a1133e --- /dev/null +++ b/compress/function.h @@ -0,0 +1,24 @@ +#ifndef _FUNCTION_H_ +#define _FUNCTION_H_ + +#include +#include +#include + +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 diff --git a/compress/function/zstd.c b/compress/function/zstd.c new file mode 100644 index 0000000..e369fe5 --- /dev/null +++ b/compress/function/zstd.c @@ -0,0 +1,287 @@ +#include "../compressor.h" +#include "../function.h" + +#include +#include + +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, +}; diff --git a/compress/include/blue/compress.h b/compress/include/blue/compress.h new file mode 100644 index 0000000..e69de29 diff --git a/compress/include/blue/compress/compressor.h b/compress/include/blue/compress/compressor.h new file mode 100644 index 0000000..3f732ad --- /dev/null +++ b/compress/include/blue/compress/compressor.h @@ -0,0 +1,33 @@ +#ifndef BLUELIB_COMPRESS_COMPRESSOR_H_ +#define BLUELIB_COMPRESS_COMPRESSOR_H_ + +#include +#include +#include + +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 diff --git a/compress/include/blue/compress/cstream.h b/compress/include/blue/compress/cstream.h new file mode 100644 index 0000000..e69de29 diff --git a/compress/include/blue/compress/function.h b/compress/include/blue/compress/function.h new file mode 100644 index 0000000..b4d248d --- /dev/null +++ b/compress/include/blue/compress/function.h @@ -0,0 +1,22 @@ +#ifndef BLUELIB_COMPRESS_FUNCTION_H_ +#define BLUELIB_COMPRESS_FUNCTION_H_ + +#include +#include + +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