diff options
| -rw-r--r-- | include/block_processor.h | 111 | ||||
| -rw-r--r-- | lib/Makemodule.am | 2 | ||||
| -rw-r--r-- | lib/comp/block_processor.c | 72 | ||||
| -rw-r--r-- | lib/comp/create_block.c | 37 | ||||
| -rw-r--r-- | lib/comp/process_block.c | 37 |
5 files changed, 259 insertions, 0 deletions
diff --git a/include/block_processor.h b/include/block_processor.h new file mode 100644 index 0000000..4944969 --- /dev/null +++ b/include/block_processor.h @@ -0,0 +1,111 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +/* + * block_processor.h + * + * Copyright (C) 2019 David Oberhollenzer <goliath@infraroot.at> + */ +#ifndef BLOCK_PROCESSOR_H +#define BLOCK_PROCESSOR_H + +#include "config.h" +#include "compress.h" + +enum { + /* only calculate checksum, do NOT compress the data */ + BLK_DONT_COMPRESS = 0x0001, + + /* set by compressor worker if the block was actually compressed */ + BLK_IS_COMPRESSED = 0x0002, + + /* do not calculate block checksum */ + BLK_DONT_CHECKSUM = 0x0004, + + /* set by compressor worker if compression failed */ + BLK_COMPRESS_ERROR = 0x0008, + + /* first user setable block flag */ + BLK_USER = 0x0080 +}; + +typedef struct block_t { + /* used internally, ignored and overwritten when enqueueing blocks */ + struct block_t *next; + uint32_t sequence_number; + + /* Size of the data area */ + uint32_t size; + + /* checksum of the input data */ + uint32_t checksum; + + /* user settable file block index */ + uint32_t index; + + /* user pointer associated with the block */ + void *user; + + /* user settable flag field */ + uint32_t flags; + + /* raw data to be processed */ + uint8_t data[]; +} block_t; + +typedef struct block_processor_t block_processor_t; + +/* + Gets called for each processed block. May be called from a different thread + than the one that calls enqueue, but only from one thread at a time. + Guaranteed to be called on blocks in the order that they are submitted + to enqueue. + + A non-zero return value is interpreted as fatal error. + */ +typedef int (*block_cb)(void *user, block_t *blk); + +block_processor_t *block_processor_create(size_t max_block_size, + compressor_t *cmp, + unsigned int num_workers, + void *user, + block_cb callback); + +void block_processor_destroy(block_processor_t *proc); + +/* + Add a block to be processed. Returns non-zero on error and prints a message + to stderr. + + The function takes over ownership of the submitted block. It is freed with + a after processing and calling the block callback. + + Even on failure, the workers may still be running and + block_processor_finish must be called before cleaning up. +*/ +int block_processor_enqueue(block_processor_t *proc, block_t *block); + +/* + Wait for the compressor workers to finish. Returns zero on success, non-zero + if an internal error occoured or one of the block callbacks returned a + non-zero value. + */ +int block_processor_finish(block_processor_t *proc); + +/* + Convenience function to create a block structure and optionally fill it with + content. + + filename is used for printing error messages. If fd is a valid file + descriptor (>= 0), the function attempts to populate the payload data + from the input file. + */ +block_t *create_block(const char *filename, int fd, size_t size, + void *user, uint32_t flags); + +/* + Convenience function to process a data block. Returns 0 on success, + prints to stderr on failure. + */ +int process_block(block_t *block, compressor_t *cmp, + uint8_t *scratch, size_t scratch_size); + +#endif /* BLOCK_PROCESSOR_H */ diff --git a/lib/Makemodule.am b/lib/Makemodule.am index 82c910c..7c2c99d 100644 --- a/lib/Makemodule.am +++ b/lib/Makemodule.am @@ -18,6 +18,8 @@ libtar_a_CFLAGS = $(AM_CFLAGS) libtar_a_CPPFLAGS = $(AM_CPPFLAGS) libcompress_a_SOURCES = lib/comp/compressor.c lib/comp/internal.h +libcompress_a_SOURCES += lib/comp/block_processor.c include/block_processor.h +libcompress_a_SOURCES += lib/comp/create_block.c lib/comp/process_block.c libcompress_a_SOURCES += include/compress.h libcompress_a_CFLAGS = $(AM_CFLAGS) libcompress_a_CPPFLAGS = $(AM_CPPFLAGS) diff --git a/lib/comp/block_processor.c b/lib/comp/block_processor.c new file mode 100644 index 0000000..5938b3a --- /dev/null +++ b/lib/comp/block_processor.c @@ -0,0 +1,72 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +/* + * block_processor.c + * + * Copyright (C) 2019 David Oberhollenzer <goliath@infraroot.at> + */ +#include "config.h" + +#include "block_processor.h" +#include "util.h" + +#include <string.h> +#include <stdlib.h> +#include <stdio.h> + +struct block_processor_t { + size_t max_block_size; + compressor_t *cmp; + block_cb cb; + void *user; + int status; + + uint8_t scratch[]; +}; + +block_processor_t *block_processor_create(size_t max_block_size, + compressor_t *cmp, + unsigned int num_workers, + void *user, + block_cb callback) +{ + block_processor_t *proc = calloc(1, sizeof(*proc) + max_block_size); + (void)num_workers; + + if (proc == NULL) { + perror("Creating block processor"); + return NULL; + } + + proc->max_block_size = max_block_size; + proc->cmp = cmp; + proc->cb = callback; + proc->user = user; + return proc; +} + +void block_processor_destroy(block_processor_t *proc) +{ + free(proc); +} + +int block_processor_enqueue(block_processor_t *proc, block_t *block) +{ + if (process_block(block, proc->cmp, + proc->scratch, proc->max_block_size)) + goto fail; + + if (proc->cb(proc->user, block)) + goto fail; + + free(block); + return 0; +fail: + free(block); + proc->status = -1; + return -1; +} + +int block_processor_finish(block_processor_t *proc) +{ + return proc->status; +} diff --git a/lib/comp/create_block.c b/lib/comp/create_block.c new file mode 100644 index 0000000..e410091 --- /dev/null +++ b/lib/comp/create_block.c @@ -0,0 +1,37 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +/* + * create_block.c + * + * Copyright (C) 2019 David Oberhollenzer <goliath@infraroot.at> + */ +#include "config.h" + +#include "block_processor.h" +#include "util.h" + +#include <string.h> +#include <stdlib.h> +#include <stdio.h> + +block_t *create_block(const char *filename, int fd, size_t size, + void *user, uint32_t flags) +{ + block_t *blk = calloc(1, sizeof(*blk) + size); + + if (blk == NULL) { + perror(filename); + return NULL; + } + + if (fd >= 0) { + if (read_data(filename, fd, blk->data, size)) { + free(blk); + return NULL; + } + } + + blk->size = size; + blk->user = user; + blk->flags = flags; + return blk; +} diff --git a/lib/comp/process_block.c b/lib/comp/process_block.c new file mode 100644 index 0000000..76cd07d --- /dev/null +++ b/lib/comp/process_block.c @@ -0,0 +1,37 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +/* + * process_block.c + * + * Copyright (C) 2019 David Oberhollenzer <goliath@infraroot.at> + */ +#include "config.h" + +#include "block_processor.h" +#include "util.h" + +#include <string.h> + +int process_block(block_t *block, compressor_t *cmp, + uint8_t *scratch, size_t scratch_size) +{ + ssize_t ret; + + if (!(block->flags & BLK_DONT_CHECKSUM)) + block->checksum = update_crc32(0, block->data, block->size); + + if (!(block->flags & BLK_DONT_COMPRESS)) { + ret = cmp->do_block(cmp, block->data, block->size, + scratch, scratch_size); + + if (ret < 0) + return -1; + + if (ret > 0) { + memcpy(block->data, scratch, ret); + block->size = ret; + block->flags |= BLK_IS_COMPRESSED; + } + } + + return 0; +} |