From aed0e3024cff2b2802dad891d2c52fb55b523f50 Mon Sep 17 00:00:00 2001 From: Dmitri Pavlov Date: Tue, 13 Oct 2026 16:36:15 +0300 Subject: [PATCH] chunkio: add optional file-slot pool and make chunk counters thread-safe Signed-off-by: Dmitri Pavlov --- CMakeLists.txt | 4 + include/chunkio/chunkio.h | 6 + include/chunkio/cio_chunk.h | 2 + include/chunkio/cio_file.h | 1 + include/chunkio/cio_file_native.h | 1 + include/chunkio/cio_stream.h | 22 + src/CMakeLists.txt | 6 + src/chunkio.c | 2 + src/cio_chunk.c | 152 ++- src/cio_file.c | 171 +++- src/cio_file_unix.c | 34 + src/cio_file_win32.c | 39 + src/cio_scan.c | 36 +- src/cio_stream.c | 284 ++++++ tests/CMakeLists.txt | 1 + tests/fs_pool.c | 1432 +++++++++++++++++++++++++++++ 16 files changed, 2171 insertions(+), 22 deletions(-) create mode 100644 tests/fs_pool.c diff --git a/CMakeLists.txt b/CMakeLists.txt index 5a7f1c6..f2b44b5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,6 +1,10 @@ cmake_minimum_required(VERSION 3.20) project(chunk-io C) +if(NOT WIN32) + find_package(Threads REQUIRED) +endif() + set(CIO_VERSION_MAJOR 1) set(CIO_VERSION_MINOR 5) set(CIO_VERSION_PATCH 4) diff --git a/include/chunkio/chunkio.h b/include/chunkio/chunkio.h index c60ddac..fadd676 100644 --- a/include/chunkio/chunkio.h +++ b/include/chunkio/chunkio.h @@ -84,6 +84,12 @@ struct cio_options { /* chunk handlings */ int realloc_size_hint; + + /* + * File-slot pool: number of reusable chunk file slots per stream + * (0 = disabled, stock behaviour). + */ + int file_pool_size; }; struct cio_ctx { diff --git a/include/chunkio/cio_chunk.h b/include/chunkio/cio_chunk.h index 369401b..8473b67 100644 --- a/include/chunkio/cio_chunk.h +++ b/include/chunkio/cio_chunk.h @@ -89,6 +89,8 @@ char *cio_version(); size_t cio_chunk_counter_total_add(struct cio_ctx *ctx); size_t cio_chunk_counter_total_sub(struct cio_ctx *ctx); size_t cio_chunk_counter_total_up_add(struct cio_ctx *ctx); +size_t cio_chunk_counter_total_up_get(struct cio_ctx *ctx); +int cio_chunk_counter_total_up_try_inc(struct cio_ctx *ctx); size_t cio_chunk_counter_total_up_sub(struct cio_ctx *ctx); #endif diff --git a/include/chunkio/cio_file.h b/include/chunkio/cio_file.h index f05d386..77b94b3 100644 --- a/include/chunkio/cio_file.h +++ b/include/chunkio/cio_file.h @@ -40,6 +40,7 @@ struct cio_file { size_t realloc_size; /* chunk size to increase alloc */ char *path; /* root path + stream */ char *map; /* map of data */ + int pool_slot; /* file-slot pool index (-1 = not pooled) */ struct cio_ctx *ctx; /* owning context */ #ifdef _WIN32 HANDLE backing_file; diff --git a/include/chunkio/cio_file_native.h b/include/chunkio/cio_file_native.h index 1658495..44fd92d 100644 --- a/include/chunkio/cio_file_native.h +++ b/include/chunkio/cio_file_native.h @@ -50,6 +50,7 @@ int cio_file_native_open(struct cio_file *cf); int cio_file_native_close(struct cio_file *cf); int cio_file_native_delete(struct cio_file *cf); int cio_file_native_delete_by_path(const char *path); +int cio_file_native_truncate_by_path(const char *path); int cio_file_native_sync(struct cio_file *cf, int sync_mode); int cio_file_native_resize(struct cio_file *cf, size_t new_size); diff --git a/include/chunkio/cio_stream.h b/include/chunkio/cio_stream.h index 22e34c2..bf6e5a0 100644 --- a/include/chunkio/cio_stream.h +++ b/include/chunkio/cio_stream.h @@ -30,6 +30,19 @@ struct cio_stream { struct mk_list chunks_up; /* list of chunks who are 'up' */ struct mk_list chunks_down; /* list of chunks who are 'down' */ void *parent; /* ref to parent ctx */ + + /* + * File-slot pool: when enabled (storage.file_pool_size option), chunkio + * reuses a bounded set of chunk file names per stream. Chunks are + * truncated on delete instead of unlinked, so kernel dentry/inode + * objects are reused and the slab footprint stays bounded instead of + * growing per chunk. + */ + int pool_size; /* number of slots (0 = disabled) */ + unsigned char *pool_busy; /* busy bitmap, one bit per slot */ + int pool_warned; /* fallback warning already logged */ + int scanning; /* set while cio_load() scans existing files */ + void *pool_lock; /* opaque lock protecting pool_busy */ }; struct cio_stream *cio_stream_create(struct cio_ctx *ctx, const char *name, @@ -40,4 +53,13 @@ void cio_stream_destroy(struct cio_stream *st); void cio_stream_destroy_all(struct cio_ctx *ctx); size_t cio_stream_size_chunks_up(struct cio_stream *st); +/* file-slot pool API (used by the filesystem backend) */ +int cio_stream_pool_acquire(struct cio_stream *st); +void cio_stream_pool_release(struct cio_stream *st, int slot); +void cio_stream_pool_mark_busy(struct cio_stream *st, int slot); +int cio_stream_pool_slot_from_name(const char *name); +void cio_stream_pool_slot_name(int slot, char *buf, size_t len); +int cio_stream_pool_slot_valid(struct cio_stream *st, int slot); +int cio_stream_pool_exhausted_warned(struct cio_stream *st); + #endif diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 3590143..70f6050 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -15,6 +15,12 @@ set(src set(libs cio-crc32) +if(NOT WIN32) + set(libs + ${libs} + Threads::Threads) +endif() + if(${CMAKE_SYSTEM_NAME} MATCHES "Windows") set(src ${src} diff --git a/src/chunkio.c b/src/chunkio.c index 87cb131..6e3e5ab 100644 --- a/src/chunkio.c +++ b/src/chunkio.c @@ -79,6 +79,7 @@ void cio_options_init(struct cio_options *options) options->log_level = CIO_LOG_INFO; options->flags = CIO_OPEN_RW; options->realloc_size_hint = CIO_DISABLE_REALLOC_HINT; + options->file_pool_size = 0; } struct cio_ctx *cio_create(struct cio_options *options) @@ -126,6 +127,7 @@ struct cio_ctx *cio_create(struct cio_options *options) ctx->page_size = cio_getpagesize(); ctx->max_chunks_up = CIO_MAX_CHUNKS_UP; ctx->options.flags = options->flags; + ctx->options.file_pool_size = options->file_pool_size; ctx->realloc_size_hint = CIO_DISABLE_REALLOC_HINT; if (options->user != NULL) { diff --git a/src/cio_chunk.c b/src/cio_chunk.c index 7917cd2..90fe2bb 100644 --- a/src/cio_chunk.c +++ b/src/cio_chunk.c @@ -168,6 +168,40 @@ int cio_chunk_delete(struct cio_ctx *ctx, struct cio_stream *st, const char *nam #endif if (st->type == CIO_STORE_FS) { + struct mk_list *head; + struct cio_chunk *ch; + struct cio_chunk *loaded; + + /* + * A pooled chunk's logical name is not its on-disk name, so a delete + * by logical name must go through the close path to resolve the real + * slot file and release the slot. Only pooled chunks are closed: for + * every other loaded chunk the historical behaviour is kept (the file + * is removed, the chunk object is left for its owner), so callers + * holding a chunk pointer are unaffected when the pool is disabled. + * + * The per-stream chunk lists are not locked: chunkio does not support + * concurrent operations on one stream (Fluent Bit gives each input + * thread its own stream). The scan path calls this single-threaded. + */ + loaded = NULL; + mk_list_foreach(head, &st->chunks) { + ch = mk_list_entry(head, struct cio_chunk, _head); + if (strcmp(ch->name, name) == 0) { + loaded = ch; + break; + } + } + + if (loaded != NULL) { + struct cio_file *cf = (struct cio_file *) loaded->backend; + + if (cf != NULL && cf->pool_slot >= 0) { + cio_chunk_close(loaded, CIO_TRUE); + return CIO_OK; + } + } + result = cio_file_delete(ctx, st, name); } else { @@ -613,30 +647,132 @@ char *cio_version() * Counters API */ +/* + * Chunk counters are shared by every stream of a context, and Fluent Bit can + * update chunks from multiple input threads when inputs run in threaded mode; + * use atomic operations so concurrent updates cannot be lost. A lost + * decrement underflows 'total_chunks_up' (0 - 1) and then every new chunk is + * created 'down' (writes fail with "file is not mmap()ed"). + */ +static inline size_t cio_counter_add(size_t *counter) +{ +#if defined(_WIN32) +#if defined(_WIN64) + return (size_t) _InterlockedIncrement64((volatile LONG64 *) counter); +#else + return (size_t) _InterlockedIncrement((volatile LONG *) counter); +#endif +#else + return __atomic_add_fetch(counter, 1, __ATOMIC_SEQ_CST); +#endif +} + +static inline size_t cio_counter_sub(size_t *counter) +{ +#if defined(_WIN32) +#if defined(_WIN64) + return (size_t) _InterlockedDecrement64((volatile LONG64 *) counter); +#else + return (size_t) _InterlockedDecrement((volatile LONG *) counter); +#endif +#else + return __atomic_sub_fetch(counter, 1, __ATOMIC_SEQ_CST); +#endif +} + +static inline size_t cio_counter_get(size_t *counter) +{ +#if defined(_WIN32) +#if defined(_WIN64) + return (size_t) _InterlockedCompareExchange64((volatile LONG64 *) counter, + 0, 0); +#else + return (size_t) _InterlockedCompareExchange((volatile LONG *) counter, + 0, 0); +#endif +#else + return __atomic_load_n(counter, __ATOMIC_RELAXED); +#endif +} + +static inline int cio_counter_cas(size_t *counter, size_t *expected, + size_t desired) +{ +#if defined(_WIN32) +#if defined(_WIN64) + LONG64 prev; + + prev = _InterlockedCompareExchange64((volatile LONG64 *) counter, + (LONG64) desired, + (LONG64) *expected); + if ((size_t) prev == *expected) { + return 1; + } + *expected = (size_t) prev; + return 0; +#else + LONG prev; + + prev = _InterlockedCompareExchange((volatile LONG *) counter, + (LONG) desired, (LONG) *expected); + if ((size_t) prev == *expected) { + return 1; + } + *expected = (size_t) prev; + return 0; +#endif +#else + return __atomic_compare_exchange_n(counter, expected, desired, 0, + __ATOMIC_SEQ_CST, __ATOMIC_RELAXED); +#endif +} + /* Increase the number of total chunks registered (+1) */ size_t cio_chunk_counter_total_add(struct cio_ctx *ctx) { - ctx->total_chunks++; - return ctx->total_chunks; + return cio_counter_add(&ctx->total_chunks); } /* Decrease the total number of chunks (-1) */ size_t cio_chunk_counter_total_sub(struct cio_ctx *ctx) { - ctx->total_chunks--; - return ctx->total_chunks; + return cio_counter_sub(&ctx->total_chunks); } /* Increase the number of total chunks up in memory (+1) */ size_t cio_chunk_counter_total_up_add(struct cio_ctx *ctx) { - ctx->total_chunks_up++; - return ctx->total_chunks_up; + return cio_counter_add(&ctx->total_chunks_up); } /* Decrease the total number of chunks up in memory (-1) */ size_t cio_chunk_counter_total_up_sub(struct cio_ctx *ctx) { - ctx->total_chunks_up--; - return ctx->total_chunks_up; + return cio_counter_sub(&ctx->total_chunks_up); +} + +/* Get the number of chunks up in memory */ +size_t cio_chunk_counter_total_up_get(struct cio_ctx *ctx) +{ + return cio_counter_get(&ctx->total_chunks_up); +} + +/* + * Try to reserve one 'up' chunk. Returns CIO_TRUE with the counter already + * incremented, or CIO_FALSE when max_chunks_up is reached. The check and the + * increment are a single atomic operation so concurrent threads cannot + * overshoot the configured limit. + */ +int cio_chunk_counter_total_up_try_inc(struct cio_ctx *ctx) +{ + size_t cur; + + do { + cur = cio_counter_get(&ctx->total_chunks_up); + if (cur >= ctx->max_chunks_up) { + return CIO_FALSE; + } + } while (cio_counter_cas(&ctx->total_chunks_up, &cur, cur + 1) == 0); + + return CIO_TRUE; } diff --git a/src/cio_file.c b/src/cio_file.c index 26dc992..c5a6e45 100644 --- a/src/cio_file.c +++ b/src/cio_file.c @@ -486,8 +486,8 @@ static int mmap_file(struct cio_ctx *ctx, struct cio_chunk *ch, size_t size) cf->st_content = cio_file_st_get_content(cf->map); cio_log_debug(ctx, "%s:%s mapped OK", ch->st->name, ch->name); - /* The mmap succeeded, adjust the counters */ - cio_chunk_counter_total_up_add(ctx); + /* Counter accounting is done by the caller, which reserves the slot + * atomically before mapping (see cio_chunk_counter_total_up_try_inc) */ return CIO_OK; } @@ -504,6 +504,13 @@ int cio_file_lookup_group(char *group, void **result) int cio_file_read_prepare(struct cio_ctx *ctx, struct cio_chunk *ch) { + /* + * Map a file for a metadata/content read without enforcing max_chunks_up. + * No counter accounting happens here: a 'down' chunk has no file + * descriptor and cio_file_native_map() refuses to map it, so this either + * fails or returns early when the file is already mapped (which the stock + * mmap_file() did not count either). + */ return mmap_file(ctx, ch, 0); } @@ -563,11 +570,8 @@ int cio_file_content_copy(struct cio_chunk *ch, */ static inline int open_and_up(struct cio_ctx *ctx) { - if (ctx->total_chunks_up >= ctx->max_chunks_up) { - return CIO_FALSE; - } - - return CIO_TRUE; + /* atomically reserves an 'up' slot (counter incremented on success) */ + return cio_chunk_counter_total_up_try_inc(ctx); } /* @@ -641,12 +645,18 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, int *err) { char error_message[256]; + char slot_name[64]; char *path; int ret; + int slot; + int pool_slot; + int reserved; struct cio_file *cf; (void) size; + reserved = CIO_FALSE; + ret = cio_file_native_filename_check(ch->name); if (ret != CIO_OK) { cio_log_error(ctx, "[cio file] invalid file name"); @@ -659,10 +669,53 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, return NULL; } + /* + * File-slot pool: a chunk name that is not a slot name is a new chunk, so + * take a free slot and use its file name. Never look up the caller's + * (unique) logical path: a failed lookup would leave a negative dentry in + * the cache on every flush, which is exactly the growth we are avoiding. + * Recovery opens existing files by their slot names; during a scan, a + * non-slot name is a legacy or fallback chunk file that must be recovered + * in place (no slot is allocated and no path is redirected). + */ + pool_slot = -1; + if (st->pool_size > 0) { + slot = cio_stream_pool_slot_from_name(ch->name); + if (slot >= 0) { + cio_stream_pool_mark_busy(st, slot); + pool_slot = slot; + } + else if (st->scanning == CIO_TRUE) { + /* legacy file: keep its own path, it is unlinked on delete */ + } + else { + slot = cio_stream_pool_acquire(st); + if (slot >= 0) { + cio_stream_pool_slot_name(slot, slot_name, sizeof(slot_name)); + free(path); + path = cio_file_native_compose_path(ctx->options.root_path, + st->name, slot_name); + if (path == NULL) { + cio_stream_pool_release(st, slot); + + return NULL; + } + pool_slot = slot; + } + else if (!cio_stream_pool_exhausted_warned(st)) { + cio_log_warn(ctx, "[cio file] file pool exhausted (%d slots), " + "falling back to unique chunk names", st->pool_size); + } + } + } + /* Create file context */ cf = calloc(1, sizeof(struct cio_file)); if (!cf) { cio_errno(); + if (pool_slot >= 0) { + cio_stream_pool_release(st, pool_slot); + } free(path); return NULL; @@ -687,6 +740,7 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, cf->ctx = ctx; cf->auto_remap_warned = CIO_FALSE; cf->map_truncated_warned = CIO_FALSE; + cf->pool_slot = pool_slot; ch->backend = cf; #ifdef _WIN32 @@ -714,10 +768,19 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, return cf; } + /* up-slot reserved: counter already incremented */ + reserved = CIO_TRUE; + /* Open the file */ ret = cio_file_native_open(cf); if (ret != CIO_OK) { + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ctx); + } + if (cf->pool_slot >= 0) { + cio_stream_pool_release(st, cf->pool_slot); + } free(path); free(cf); @@ -732,6 +795,12 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, if (ret != CIO_OK) { cio_file_native_close(cf); + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ctx); + } + if (cf->pool_slot >= 0) { + cio_stream_pool_release(st, cf->pool_slot); + } free(path); free(cf); @@ -757,6 +826,12 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, cio_file_native_close(cf); + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ctx); + } + if (cf->pool_slot >= 0) { + cio_stream_pool_release(st, cf->pool_slot); + } free(path); free(cf); @@ -768,6 +843,12 @@ struct cio_file *cio_file_open(struct cio_ctx *ctx, if (ret == CIO_ERROR || ret == CIO_CORRUPTED || ret == CIO_RETRY) { cio_file_native_close(cf); + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ctx); + } + if (cf->pool_slot >= 0) { + cio_stream_pool_release(st, cf->pool_slot); + } free(path); free(cf); @@ -789,6 +870,7 @@ int cio_file_delete(struct cio_ctx *ctx, struct cio_stream *st, const char *name { char *path; int ret; + int slot; ret = cio_file_native_filename_check((char *) name); if (ret != CIO_OK) { @@ -802,7 +884,28 @@ int cio_file_delete(struct cio_ctx *ctx, struct cio_stream *st, const char *name return CIO_ERROR; } - ret = cio_file_native_delete_by_path(path); + /* + * Pooled slot files are truncated (not unlinked) and returned to the pool + * so their kernel dentry/inode objects stay cached and reusable. Only + * truncate when the slot belongs to the current pool; otherwise unlink + * (pool disabled, or slot left over from a larger pool). + */ + slot = cio_stream_pool_slot_from_name(name); + if (cio_stream_pool_slot_valid(st, slot) == CIO_TRUE) { + ret = cio_file_native_truncate_by_path(path); + if (ret == CIO_OK) { + cio_stream_pool_release(st, slot); + } + else { + cio_log_error(ctx, + "[cio file] cannot recycle pooled chunk file %s/%s; " + "slot kept busy", + st->name, name); + } + } + else { + ret = cio_file_native_delete_by_path(path); + } free(path); @@ -816,6 +919,7 @@ int cio_file_delete(struct cio_ctx *ctx, struct cio_stream *st, const char *name static int _cio_file_up(struct cio_chunk *ch, int enforced) { int ret; + int reserved = CIO_FALSE; struct cio_file *cf = (struct cio_file *) ch->backend; if (cf->map) { @@ -839,12 +943,17 @@ static int _cio_file_up(struct cio_chunk *ch, int enforced) if (ret == CIO_FALSE) { return CIO_ERROR; } + /* up-slot reserved: counter already incremented */ + reserved = CIO_TRUE; } /* Open file */ ret = cio_file_native_open(cf); if (ret != CIO_OK) { + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ch->ctx); + } cio_log_error(ch->ctx, "[cio file] cannot open chunk: %s/%s", ch->st->name, ch->name); return CIO_ERROR; @@ -852,6 +961,11 @@ static int _cio_file_up(struct cio_chunk *ch, int enforced) ret = cio_file_update_size(cf); if (ret != CIO_OK) { + cio_file_native_close(cf); + + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ch->ctx); + } return CIO_ERROR; } @@ -879,6 +993,19 @@ static int _cio_file_up(struct cio_chunk *ch, int enforced) cio_file_native_close(cf); } + if (ret != CIO_OK) { + if (reserved == CIO_TRUE) { + cio_chunk_counter_total_up_sub(ch->ctx); + } + return ret; + } + + /* non-enforced bring-up: account the new up chunk here, the enforced + * path already reserved the slot atomically */ + if (reserved == CIO_FALSE) { + cio_chunk_counter_total_up_add(ch->ctx); + } + return ret; } @@ -977,12 +1104,30 @@ void cio_file_close(struct cio_chunk *ch, int delete) /* Should we delete the content from the file system ? */ if (delete == CIO_TRUE) { - ret = cio_file_native_delete(cf); + if (cio_stream_pool_slot_valid(ch->st, cf->pool_slot) == CIO_TRUE) { + /* pooled files are recycled: truncate and free the slot only if + * the truncate succeeded, otherwise the slot stays busy so stale + * content is never handed to a new chunk */ + ret = cio_file_native_truncate_by_path(cf->path); + if (ret == CIO_OK) { + cio_stream_pool_release(ch->st, cf->pool_slot); + } + else { + cio_log_error(ch->ctx, + "[cio file] cannot recycle pooled chunk %s:%s; " + "slot kept busy", + ch->st->name, ch->name); + } + } + else { + /* not a slot of the current pool (disabled or shrunk): unlink */ + ret = cio_file_native_delete(cf); - if (ret != CIO_OK) { - cio_log_error(ch->ctx, - "[cio file] error deleting file at close %s:%s", - ch->st->name, ch->name); + if (ret != CIO_OK) { + cio_log_error(ch->ctx, + "[cio file] error deleting file at close %s:%s", + ch->st->name, ch->name); + } } } diff --git a/src/cio_file_unix.c b/src/cio_file_unix.c index 9dfb6de..5f5c4eb 100644 --- a/src/cio_file_unix.c +++ b/src/cio_file_unix.c @@ -569,3 +569,37 @@ int cio_file_native_resize(struct cio_file *cf, size_t new_size) return result; } + +/* + * Pool support: truncate a file to zero bytes (used by the file-slot pool to + * recycle a slot without unlinking it, keeping the kernel dentry/inode objects + * in place). + */ +int cio_file_native_truncate_by_path(const char *path) +{ + int fd; + + if (!path) { + return CIO_ERROR; + } + + /* + * O_TRUNC is enough: the content was already delivered, so at worst a + * lost truncate after a crash means the chunk is re-read and re-sent + * (at-least-once), which is cheaper than an fsync per recycled chunk. + */ + fd = open(path, O_WRONLY | O_TRUNC); + if (fd == -1) { + if (errno == ENOENT) { + /* file never created (e.g. a 'down' chunk): nothing to do */ + return CIO_OK; + } + cio_file_native_report_os_error(); + + return CIO_ERROR; + } + + close(fd); + + return CIO_OK; +} diff --git a/src/cio_file_win32.c b/src/cio_file_win32.c index 43b7a24..66d2f4b 100644 --- a/src/cio_file_win32.c +++ b/src/cio_file_win32.c @@ -624,3 +624,42 @@ int cio_file_native_resize(struct cio_file *cf, size_t new_size) return CIO_OK; } + +/* + * Pool support: truncate a file to zero bytes (used by the file-slot pool to + * recycle a slot without unlinking it, keeping the kernel dentry/inode objects + * in place). + */ +int cio_file_native_truncate_by_path(const char *path) +{ + HANDLE h; + LARGE_INTEGER pos; + + if (!path) { + return CIO_ERROR; + } + + h = CreateFileA(path, GENERIC_WRITE, FILE_SHARE_READ | FILE_SHARE_WRITE, + NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL); + if (h == INVALID_HANDLE_VALUE) { + if (GetLastError() == ERROR_FILE_NOT_FOUND) { + /* file never created (e.g. a 'down' chunk): nothing to do */ + return CIO_OK; + } + cio_file_native_report_os_error(); + + return CIO_ERROR; + } + + pos.QuadPart = 0; + if (!SetFilePointerEx(h, pos, NULL, FILE_BEGIN) || !SetEndOfFile(h)) { + cio_file_native_report_os_error(); + CloseHandle(h); + + return CIO_ERROR; + } + + CloseHandle(h); + + return CIO_OK; +} diff --git a/src/cio_scan.c b/src/cio_scan.c index ca2f20c..94a7e87 100644 --- a/src/cio_scan.c +++ b/src/cio_scan.c @@ -21,11 +21,13 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -45,6 +47,7 @@ static int cio_scan_stream_files(struct cio_ctx *ctx, struct cio_stream *st, int err; int ext_off; int ext_len = 0; + int slot; char *path; DIR *dir; struct dirent *ent; @@ -102,8 +105,39 @@ static int cio_scan_stream_files(struct cio_ctx *ctx, struct cio_stream *st, ctx->last_chunk_error = 0; - /* register every directory as a stream */ + /* + * File-slot pool: an empty slot file is a free slot, not a chunk. + * Leave it in place so new chunks reuse it; if the slot number is + * outside the current pool size (left over from a larger pool) it is + * not reusable, so unlink it instead of leaving an orphan behind. + */ + slot = cio_stream_pool_slot_from_name(ent->d_name); + if (slot >= 0) { + struct stat sb; + char *spath; + + spath = cio_file_native_compose_path(ctx->options.root_path, + st->name, ent->d_name); + if (spath) { + if (stat(spath, &sb) == 0 && sb.st_size == 0) { + if (cio_stream_pool_slot_valid(st, slot) == CIO_FALSE) { + cio_file_native_delete_by_path(spath); + } + free(spath); + continue; + } + free(spath); + } + } + + /* + * Mark the stream as scanning while the existing file is opened: the + * filesystem backend must recover non-slot (legacy or fallback) names + * in place instead of redirecting them to a fresh slot. + */ + st->scanning = CIO_TRUE; cio_chunk_open(ctx, st, ent->d_name, ctx->options.flags, 0, &err); + st->scanning = CIO_FALSE; if (ctx->options.flags & CIO_DELETE_IRRECOVERABLE) { if (err == CIO_CORRUPTED) { diff --git a/src/cio_stream.c b/src/cio_stream.c index 547d7ee..b9cc897 100644 --- a/src/cio_stream.c +++ b/src/cio_stream.c @@ -31,6 +31,272 @@ #include +#ifdef _WIN32 +#include +typedef CRITICAL_SECTION cio_pool_lock_t; +#else +#include +typedef pthread_mutex_t cio_pool_lock_t; +#endif + +#define CIO_FILE_POOL_MAX_SLOTS 4096 +#define CIO_FILE_POOL_PREFIX "slot-" + +static int pool_lock_create(cio_pool_lock_t **out) +{ + *out = calloc(1, sizeof(cio_pool_lock_t)); + if (!*out) { + return -1; + } +#ifdef _WIN32 + InitializeCriticalSection(*out); +#else + if (pthread_mutex_init(*out, NULL) != 0) { + free(*out); + *out = NULL; + return -1; + } +#endif + return 0; +} + +static void pool_lock_destroy(cio_pool_lock_t *lock) +{ + if (!lock) { + return; + } +#ifdef _WIN32 + DeleteCriticalSection(lock); +#else + pthread_mutex_destroy(lock); +#endif + free(lock); +} + +static void pool_lock(cio_pool_lock_t *lock) +{ + if (!lock) { + return; + } +#ifdef _WIN32 + EnterCriticalSection(lock); +#else + pthread_mutex_lock(lock); +#endif +} + +static void pool_unlock(cio_pool_lock_t *lock) +{ + if (!lock) { + return; + } +#ifdef _WIN32 + LeaveCriticalSection(lock); +#else + pthread_mutex_unlock(lock); +#endif +} + +static int pool_option_size(struct cio_ctx *ctx) +{ + long val; + + if (!ctx) { + return 0; + } + + val = ctx->options.file_pool_size; + if (val <= 0) { + return 0; + } + if (val > CIO_FILE_POOL_MAX_SLOTS) { + val = CIO_FILE_POOL_MAX_SLOTS; + } + + return (int) val; +} + +static void pool_init(struct cio_stream *st) +{ + cio_pool_lock_t *lock; + int size; + + st->pool_size = 0; + st->pool_busy = NULL; + st->pool_warned = 0; + st->pool_lock = NULL; + + size = pool_option_size((struct cio_ctx *) st->parent); + if (size <= 0) { + return; + } + + st->pool_busy = calloc(1, (size + 7) / 8); + if (!st->pool_busy) { + cio_log_error((struct cio_ctx *) st->parent, + "[cio stream] could not allocate file pool bitmap"); + return; + } + + if (pool_lock_create(&lock) != 0) { + free(st->pool_busy); + st->pool_busy = NULL; + cio_log_error((struct cio_ctx *) st->parent, + "[cio stream] could not allocate file pool lock"); + return; + } + + st->pool_lock = lock; + st->pool_size = size; +} + +static void pool_destroy(struct cio_stream *st) +{ + if (!st) { + return; + } + if (st->pool_busy) { + free(st->pool_busy); + st->pool_busy = NULL; + } + if (st->pool_lock) { + pool_lock_destroy((cio_pool_lock_t *) st->pool_lock); + st->pool_lock = NULL; + } + st->pool_size = 0; +} + +int cio_stream_pool_acquire(struct cio_stream *st) +{ + int i; + int byte; + int bit; + + if (!st || st->pool_size <= 0 || !st->pool_busy) { + return -1; + } + + pool_lock((cio_pool_lock_t *) st->pool_lock); + for (i = 0; i < st->pool_size; i++) { + byte = i / 8; + bit = i % 8; + if (!(st->pool_busy[byte] & (1 << bit))) { + st->pool_busy[byte] |= (1 << bit); + pool_unlock((cio_pool_lock_t *) st->pool_lock); + return i; + } + } + pool_unlock((cio_pool_lock_t *) st->pool_lock); + + return -1; +} + +void cio_stream_pool_release(struct cio_stream *st, int slot) +{ + int byte; + int bit; + + if (!st || slot < 0 || slot >= st->pool_size || !st->pool_busy) { + return; + } + + byte = slot / 8; + bit = slot % 8; + pool_lock((cio_pool_lock_t *) st->pool_lock); + st->pool_busy[byte] &= (unsigned char) ~(1 << bit); + pool_unlock((cio_pool_lock_t *) st->pool_lock); +} + +void cio_stream_pool_mark_busy(struct cio_stream *st, int slot) +{ + int byte; + int bit; + + if (!st || slot < 0 || slot >= st->pool_size || !st->pool_busy) { + return; + } + + byte = slot / 8; + bit = slot % 8; + pool_lock((cio_pool_lock_t *) st->pool_lock); + st->pool_busy[byte] |= (1 << bit); + pool_unlock((cio_pool_lock_t *) st->pool_lock); +} + +int cio_stream_pool_slot_from_name(const char *name) +{ + int i; + int slot; + const char *p; + size_t prefix_len; + + if (!name) { + return -1; + } + + prefix_len = strlen(CIO_FILE_POOL_PREFIX); + if (strncmp(name, CIO_FILE_POOL_PREFIX, prefix_len) != 0) { + return -1; + } + + /* + * Only the canonical form is a slot: slot-<6 digits>.flb + * (reject slot-5.flb, slot-000005.flb.orig, etc.). + */ + p = name + prefix_len; + for (i = 0; i < 6; i++) { + if (p[i] < '0' || p[i] > '9') { + return -1; + } + } + if (strcmp(p + 6, ".flb") != 0) { + return -1; + } + + slot = 0; + for (i = 0; i < 6; i++) { + slot = slot * 10 + (p[i] - '0'); + } + + return slot; +} + +void cio_stream_pool_slot_name(int slot, char *buf, size_t len) +{ + if (!buf || len == 0) { + return; + } + if (slot < 0 || slot > 999999) { + buf[0] = '\0'; + return; + } + snprintf(buf, len, CIO_FILE_POOL_PREFIX "%06d.flb", slot); +} + +int cio_stream_pool_slot_valid(struct cio_stream *st, int slot) +{ + if (!st || slot < 0 || slot >= st->pool_size) { + return CIO_FALSE; + } + + return CIO_TRUE; +} + +int cio_stream_pool_exhausted_warned(struct cio_stream *st) +{ + int warned; + + if (!st) { + return 1; + } + pool_lock((cio_pool_lock_t *) st->pool_lock); + warned = st->pool_warned; + st->pool_warned = 1; + pool_unlock((cio_pool_lock_t *) st->pool_lock); + + return warned; +} + static char *get_stream_path(struct cio_ctx *ctx, struct cio_stream *st) { int ret; @@ -173,6 +439,17 @@ struct cio_stream *cio_stream_create(struct cio_ctx *ctx, const char *name, mk_list_init(&st->chunks_down); mk_list_add(&st->_head, &ctx->streams); + /* pool state must be valid for every stream type (destroy frees it) */ + st->pool_size = 0; + st->pool_busy = NULL; + st->pool_warned = 0; + st->pool_lock = NULL; + st->scanning = CIO_FALSE; + + if (type == CIO_STORE_FS) { + pool_init(st); + } + cio_log_debug(ctx, "[cio stream] new stream registered: %s", name); return st; } @@ -185,6 +462,8 @@ void cio_stream_destroy(struct cio_stream *st) /* close all files */ cio_chunk_close_stream(st); + pool_destroy(st); + /* destroy stream */ mk_list_del(&st->_head); free(st->name); @@ -209,6 +488,11 @@ int cio_stream_delete(struct cio_stream *st) cio_chunk_close(ch, CIO_TRUE); } + /* the stream directory is wiped below: all slots become free */ + if (st->pool_busy) { + memset(st->pool_busy, 0, (st->pool_size + 7) / 8); + } + #ifdef CIO_HAVE_BACKEND_FILESYSTEM /* If the stream is filesystem based, destroy the real directory */ if (st->type == CIO_STORE_FS) { diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 5bbd37e..51da784 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -9,6 +9,7 @@ if(CIO_BACKEND_FILESYSTEM) ${UNIT_TESTS_FILES} stream.c fs.c + fs_pool.c metadata_update.c ) endif() diff --git a/tests/fs_pool.c b/tests/fs_pool.c new file mode 100644 index 0000000..a835693 --- /dev/null +++ b/tests/fs_pool.c @@ -0,0 +1,1432 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Chunk I/O + * ========= + * + * Copyright (C) 2015-2026 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/* File-slot pool tests: bounded chunk file names, recycle-on-delete, + * recovery, backlog fallback and concurrency. + */ + +#ifndef _WIN32 +#include +#include +#include +#include +#endif +#include +#include +#ifdef _WIN32 +#include +#else +#include +#endif +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "cio_tests_internal.h" + +#ifdef _WIN32 +/* + * chunkio_compat.h (via cio_chunk.h) already includes winsock2.h and + * windows.h in the right order; include it explicitly after the chunkio + * headers for the CreateThread-based concurrency test. + */ +#include +#endif + +#define CIO_POOL_ENV "/tmp/cio-fs-pool-test/" +#define CIO_POOL_NAME "pool-test" + +#define POOL_SIZE 4 + +/* Logging callback, once called it just turn on the log_check flag */ +static int log_cb(struct cio_ctx *ctx, int level, const char *file, int line, + char *str) +{ + (void) ctx; + (void) level; + + printf("[cio-test-pool] %-60s => %s:%i\n", str, file, line); + return 0; +} + +/* Count regular files inside a stream directory */ +#ifdef _WIN32 +static int count_stream_files(const char *stream) +{ + int n = 0; + char pattern[512]; + struct _finddata_t fd; + intptr_t handle; + + snprintf(pattern, sizeof(pattern), "%s%s\\*", CIO_POOL_ENV, stream); + handle = _findfirst(pattern, &fd); + if (handle == -1) { + return -1; + } + + do { + if ((fd.attrib & _A_SUBDIR) == 0) { + n++; + } + } while (_findnext(handle, &fd) == 0); + + _findclose(handle); + return n; +} +#else +static int count_stream_files(const char *stream) +{ + int n = 0; + char path[512]; + DIR *dir; + struct dirent *ent; + + snprintf(path, sizeof(path), "%s%s", CIO_POOL_ENV, stream); + dir = opendir(path); + if (!dir) { + return -1; + } + + while ((ent = readdir(dir)) != NULL) { + if (ent->d_name[0] == '.') { + continue; + } + if (ent->d_type != DT_REG) { + continue; + } + n++; + } + closedir(dir); + + return n; +} +#endif + +static struct cio_ctx *pool_ctx_create(int pool_size, int extra_flags) +{ + struct cio_options opts; + struct cio_ctx *ctx; + + cio_options_init(&opts); + opts.root_path = CIO_POOL_ENV; + opts.log_cb = log_cb; + opts.flags = CIO_CHECKSUM | extra_flags; + opts.file_pool_size = pool_size; + + ctx = cio_create(&opts); + return ctx; +} + +static int write_and_close(struct cio_ctx *ctx, struct cio_stream *st, + const char *name, const char *payload, + int delete_chunk) +{ + int err; + int ret = CIO_OK; + void *out_buf = NULL; + size_t out_size = 0; + struct cio_chunk *chunk; + + chunk = cio_chunk_open(ctx, st, name, CIO_OPEN, 4096, &err); + if (!chunk) { + return -1; + } + + if (payload) { + ret = cio_chunk_write(chunk, payload, strlen(payload)); + if (ret != CIO_OK) { + cio_chunk_close(chunk, CIO_TRUE); + return -1; + } + + /* read back while the chunk is up: catches stale slot content */ + ret = cio_chunk_get_content_copy(chunk, &out_buf, &out_size); + if (ret != CIO_OK || out_size != strlen(payload) || + memcmp(out_buf, payload, out_size) != 0) { + if (out_buf) { + free(out_buf); + } + cio_chunk_close(chunk, CIO_TRUE); + return -1; + } + free(out_buf); + } + + cio_chunk_close(chunk, delete_chunk ? CIO_TRUE : CIO_FALSE); + return 0; +} + +/* ------------------------------------------------------------------------- */ +/* 1. name parsing helpers */ +/* ------------------------------------------------------------------------- */ +static void test_pool_slot_name_parse() +{ + char name[64]; + + printf("\n"); + + TEST_CHECK(cio_stream_pool_slot_from_name("slot-000000.flb") == 0); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-000123.flb") == 123); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-999999.flb") == 999999); + TEST_CHECK(cio_stream_pool_slot_from_name("other-000001.flb") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-.flb") == -1); + + /* only the canonical slot-<6 digits>.flb form is a slot */ + TEST_CHECK(cio_stream_pool_slot_from_name("slot-5.flb") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-5") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-000005") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-000005.flb.orig") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-0000005.flb") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-1000000.flb") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name("slot-00000a.flb") == -1); + TEST_CHECK(cio_stream_pool_slot_from_name(NULL) == -1); + + cio_stream_pool_slot_name(7, name, sizeof(name)); + TEST_CHECK(strcmp(name, "slot-000007.flb") == 0); + TEST_CHECK(cio_stream_pool_slot_from_name(name) == 7); +} + +/* ------------------------------------------------------------------------- */ +/* 2. pool disabled: legacy behaviour (unlink on delete) */ +/* ------------------------------------------------------------------------- */ +static void test_pool_disabled_legacy() +{ + int i; + int n; + int err; + char name[64]; + char payload[64]; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(0, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + + for (i = 0; i < 10; i++) { + snprintf(name, sizeof(name), "chunk-%04d.flb", i); + snprintf(payload, sizeof(payload), "payload-%d", i); + err = write_and_close(ctx, st, name, payload, CIO_TRUE); + TEST_CHECK(err == 0); + } + + /* legacy: every deleted chunk is unlinked */ + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 0); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 5b. legacy (non-slot) chunk files are recovered in place, not redirected */ +/* ------------------------------------------------------------------------- */ +static void test_pool_recovery_legacy_name() +{ + int err; + int n; + int found; + size_t size; + char *buf = NULL; + struct mk_list *head; + struct cio_chunk *chunk; + struct cio_chunk *legacy = NULL; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + /* phase 1: write a chunk with the pool disabled (legacy unique name) */ + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(0, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 0); + + err = write_and_close(ctx, st, "legacy-0001.flb", "legacy-payload", + CIO_FALSE); + TEST_CHECK(err == 0); + cio_destroy(ctx); + + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 1); + + /* + * phase 2: enable the pool and rescan. The legacy file must be recovered + * in place under its own name; no slot file may be created for it. + */ + ctx = pool_ctx_create(4, 0); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_get(ctx, CIO_POOL_NAME); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 4); + + TEST_CHECK(mk_list_size(&st->chunks) == 1); + TEST_CHECK(count_stream_files(CIO_POOL_NAME) == 1); + + found = 0; + mk_list_foreach(head, &st->chunks) { + chunk = mk_list_entry(head, struct cio_chunk, _head); + TEST_CHECK(strcmp(chunk->name, "legacy-0001.flb") == 0); + + err = cio_chunk_get_content_copy(chunk, (void **) &buf, &size); + TEST_CHECK(err == CIO_OK); + if (err == CIO_OK) { + TEST_CHECK(size == strlen("legacy-payload")); + TEST_CHECK(memcmp(buf, "legacy-payload", size) == 0); + found = 1; + free(buf); + } + } + TEST_CHECK(found == 1); + + /* + * phase 3: new chunks use slots; the recovered legacy chunk is still + * unlinked (not truncated) when it is deleted. + */ + err = write_and_close(ctx, st, "new-0001.flb", "new-payload", CIO_TRUE); + TEST_CHECK(err == 0); + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 2); + + mk_list_foreach(head, &st->chunks) { + chunk = mk_list_entry(head, struct cio_chunk, _head); + if (strcmp(chunk->name, "legacy-0001.flb") == 0) { + legacy = chunk; + break; + } + } + TEST_CHECK(legacy != NULL); + if (legacy) { + cio_chunk_close(legacy, CIO_TRUE); + } + + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 1); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 3. pool enabled: file count stays bounded, content always correct */ +/* ------------------------------------------------------------------------- */ +static void test_pool_reuse_bounds_files() +{ + int i; + int n; + char name[64]; + char payload[64]; + struct cio_ctx *ctx; + struct cio_stream *st; + struct cio_options opts; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + cio_options_init(&opts); + opts.root_path = CIO_POOL_ENV; + opts.log_cb = log_cb; + opts.file_pool_size = 4; + ctx = cio_create(&opts); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 4); + TEST_CHECK(st->pool_busy != NULL); + + for (i = 0; i < 50; i++) { + snprintf(name, sizeof(name), "logical-%04d.flb", i); + snprintf(payload, sizeof(payload), "iteration-%d-payload", i); + + n = write_and_close(ctx, st, name, payload, CIO_TRUE); + TEST_CHECK(n == 0); + + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n >= 0 && n <= POOL_SIZE); + } + + /* all slots ever used remain (truncated) after the loop; the exact count + * depends on the allocation policy (lowest free slot first) */ + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n >= 1 && n <= POOL_SIZE); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 4. backlog: more live chunks than slots -> unique fallback names */ +/* ------------------------------------------------------------------------- */ +static void test_pool_backlog_fallback() +{ + int i; + int n; + int err; + char name[64]; + char payload[64]; + struct cio_ctx *ctx; + struct cio_stream *st; + struct cio_chunk *chunks[5]; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(2, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 2); + + /* hold 5 live chunks: 2 pooled + 3 fallback */ + for (i = 0; i < 5; i++) { + snprintf(name, sizeof(name), "backlog-%d.flb", i); + snprintf(payload, sizeof(payload), "backlog-%d", i); + chunks[i] = cio_chunk_open(ctx, st, name, CIO_OPEN, 4096, &err); + TEST_CHECK(chunks[i] != NULL); + err = cio_chunk_write(chunks[i], payload, strlen(payload)); + TEST_CHECK(err == CIO_OK); + } + + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 5); + + for (i = 0; i < 5; i++) { + cio_chunk_close(chunks[i], CIO_TRUE); + } + + /* pooled files remain (truncated), fallback files are unlinked */ + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 2); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 5. recovery: scan keeps live chunks, frees empty slots, no collision */ +/* ------------------------------------------------------------------------- */ +static void test_pool_recovery() +{ + int i; + int n; + int err; + int found; + size_t size; + char *buf = NULL; + char name[64]; + struct mk_list *head; + struct cio_chunk *chunk; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + /* phase 1: one live chunk (no delete) + one deleted */ + ctx = pool_ctx_create(4, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + + err = write_and_close(ctx, st, "keep.flb", "keep-payload", CIO_FALSE); + TEST_CHECK(err == 0); + err = write_and_close(ctx, st, "drop.flb", "drop-payload", CIO_TRUE); + TEST_CHECK(err == 0); + cio_destroy(ctx); + + /* phase 2: rescan the same root */ + ctx = pool_ctx_create(4, 0); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_get(ctx, CIO_POOL_NAME); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 4); + + TEST_CHECK(mk_list_size(&st->chunks) == 1); + + found = 0; + mk_list_foreach(head, &st->chunks) { + chunk = mk_list_entry(head, struct cio_chunk, _head); + err = cio_chunk_get_content_copy(chunk, (void **) &buf, &size); + TEST_CHECK(err == CIO_OK); + if (err == CIO_OK) { + TEST_CHECK(size == strlen("keep-payload")); + TEST_CHECK(memcmp(buf, "keep-payload", size) == 0); + found = 1; + free(buf); + } + } + TEST_CHECK(found == 1); + + /* phase 3: new chunks must not clobber the recovered one */ + for (i = 0; i < 10; i++) { + snprintf(name, sizeof(name), "after-%d.flb", i); + err = write_and_close(ctx, st, name, "new-payload", CIO_TRUE); + TEST_CHECK(err == 0); + } + + found = 0; + mk_list_foreach(head, &st->chunks) { + chunk = mk_list_entry(head, struct cio_chunk, _head); + err = cio_chunk_get_content_copy(chunk, (void **) &buf, &size); + TEST_CHECK(err == CIO_OK); + if (err == CIO_OK) { + TEST_CHECK(memcmp(buf, "keep-payload", size) == 0); + found = 1; + free(buf); + } + } + TEST_CHECK(found == 1); + + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n >= 1 && n <= POOL_SIZE); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 6. corrupt slot file is truncated during scan and becomes reusable */ +/* ------------------------------------------------------------------------- */ +static void test_pool_corrupt_slot() +{ + int fd; + int err; + int n; + char path[512]; + char garbage[256]; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(2, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + err = write_and_close(ctx, st, "corrupt-me.flb", "x", CIO_FALSE); + TEST_CHECK(err == 0); + cio_destroy(ctx); + + /* find the slot file and overwrite it with garbage */ + snprintf(path, sizeof(path), "%s%s/slot-000000.flb", + CIO_POOL_ENV, CIO_POOL_NAME); + fd = open(path, O_WRONLY | O_TRUNC); + TEST_CHECK(fd != -1); + if (fd != -1) { + memset(garbage, 0x5a, sizeof(garbage)); + TEST_CHECK(write(fd, garbage, sizeof(garbage)) == + (ssize_t) sizeof(garbage)); + close(fd); + } + + /* rescan with delete-irrecoverable: the corrupt slot is truncated */ + ctx = pool_ctx_create(2, CIO_DELETE_IRRECOVERABLE); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_get(ctx, CIO_POOL_NAME); + TEST_CHECK(st != NULL); + + /* the slot file remains but must be empty (reusable) */ + { + struct stat sb; + TEST_CHECK(stat(path, &sb) == 0); + TEST_CHECK(sb.st_size == 0); + } + + /* and a new chunk can use it again */ + err = write_and_close(ctx, st, "reuse.flb", "reuse-payload", CIO_TRUE); + TEST_CHECK(err == 0); + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n >= 1 && n <= 2); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 7. pool allocator lock: no double allocation under threads */ +/* ------------------------------------------------------------------------- */ +#ifndef _WIN32 +struct lock_thread { + struct cio_stream *st; + int iterations; + int errors; +}; + +static volatile int held[POOL_SIZE]; + +static void *lock_worker(void *arg) +{ + int i; + int slot; + struct lock_thread *t = arg; + + for (i = 0; i < t->iterations; i++) { + slot = cio_stream_pool_acquire(t->st); + if (slot < 0 || slot >= POOL_SIZE) { + t->errors++; + continue; + } + + /* if the slot was already held, the lock is broken */ + if (__sync_fetch_and_add(&held[slot], 1) != 0) { + t->errors++; + } + sched_yield(); + __sync_fetch_and_sub(&held[slot], 1); + + cio_stream_pool_release(t->st, slot); + } + + return NULL; +} + +static void test_pool_lock_threads() +{ + int i; + pthread_t threads[4]; + struct lock_thread args[4]; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(4, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + + memset((void *) held, 0, sizeof(held)); + + for (i = 0; i < 4; i++) { + args[i].st = st; + args[i].iterations = 5000; + args[i].errors = 0; + TEST_CHECK(pthread_create(&threads[i], NULL, lock_worker, + &args[i]) == 0); + } + for (i = 0; i < 4; i++) { + pthread_join(threads[i], NULL); + TEST_CHECK(args[i].errors == 0); + } + + /* pool must be fully free again */ + for (i = 0; i < POOL_SIZE; i++) { + TEST_CHECK(held[i] == 0); + } + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 8. threaded chunk create/write/delete over the pool */ +/* ------------------------------------------------------------------------- */ +struct chunk_thread { + struct cio_ctx *ctx; + struct cio_stream *st; + int tid; + int iterations; + int errors; +}; + +static void *chunk_worker(void *arg) +{ + int i; + int err; + char name[64]; + char payload[96]; + char *buf = NULL; + size_t size = 0; + struct cio_chunk *chunk; + struct chunk_thread *t = arg; + + for (i = 0; i < t->iterations; i++) { + snprintf(name, sizeof(name), "t%d-chunk-%d.flb", t->tid, i); + snprintf(payload, sizeof(payload), "thread-%d-iteration-%d", t->tid, i); + + chunk = cio_chunk_open(t->ctx, t->st, name, CIO_OPEN, 4096, &err); + if (!chunk) { + t->errors++; + continue; + } + if (cio_chunk_write(chunk, payload, strlen(payload)) != CIO_OK) { + t->errors++; + cio_chunk_close(chunk, CIO_TRUE); + continue; + } + if (cio_chunk_get_content_copy(chunk, (void **) &buf, &size) != CIO_OK || + size != strlen(payload) || memcmp(buf, payload, size) != 0) { + t->errors++; + } + if (buf) { + free(buf); + buf = NULL; + } + cio_chunk_close(chunk, CIO_TRUE); + } + + return NULL; +} + +static void test_pool_threaded_chunks() +{ + int i; + int n; + pthread_t threads[2]; + struct chunk_thread args[2]; + struct cio_ctx *ctx; + struct cio_stream *streams[2]; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(8, 0); + TEST_CHECK(ctx != NULL); + /* avoid the global up-chunk counter becoming the bottleneck in this test */ + cio_set_max_chunks_up(ctx, 1024); + + /* + * Fluent Bit gives each input its own stream, so concurrent input threads + * never share a stream object; model that here. + */ + for (i = 0; i < 2; i++) { + char sname[64]; + snprintf(sname, sizeof(sname), "pool-t%d", i); + streams[i] = cio_stream_create(ctx, sname, CIO_STORE_FS); + TEST_CHECK(streams[i] != NULL); + TEST_CHECK(streams[i]->pool_size == 8); + } + + for (i = 0; i < 2; i++) { + args[i].ctx = ctx; + args[i].st = streams[i]; + args[i].tid = i; + args[i].iterations = 30; + args[i].errors = 0; + TEST_CHECK(pthread_create(&threads[i], NULL, chunk_worker, + &args[i]) == 0); + } + for (i = 0; i < 2; i++) { + pthread_join(threads[i], NULL); + TEST_CHECK(args[i].errors == 0); + } + + for (i = 0; i < 2; i++) { + char sname[64]; + snprintf(sname, sizeof(sname), "pool-t%d", i); + n = count_stream_files(sname); + TEST_CHECK(n >= 0 && n <= 8); + } + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 10b. the max_chunks_up reservation under concurrent opens */ +/* ------------------------------------------------------------------------- */ +struct limit_thread { + struct cio_ctx *ctx; + struct cio_stream *st; + pthread_barrier_t *barrier; + struct cio_chunk *chunk; + int tid; + int errors; +}; + +static void *limit_worker(void *arg) +{ + struct limit_thread *t = arg; + char name[64]; + int err = 0; + + snprintf(name, sizeof(name), "limit-%d.flb", t->tid); + + pthread_barrier_wait(t->barrier); + + t->chunk = cio_chunk_open(t->ctx, t->st, name, CIO_OPEN_RW, 4096, &err); + if (!t->chunk) { + t->errors++; + } + + return NULL; +} + +static void test_pool_threaded_max_chunks() +{ + int i; + int round; + int ups; + pthread_t threads[8]; + pthread_barrier_t barrier; + struct limit_thread args[8]; + struct cio_ctx *ctx; + struct cio_stream *streams[8]; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(16, 0); + TEST_CHECK(ctx != NULL); + cio_set_max_chunks_up(ctx, 4); + TEST_CHECK(pthread_barrier_init(&barrier, NULL, 8) == 0); + + /* + * Fluent Bit gives each input its own stream; the up-chunk counter and the + * limit are context-wide, so that is what the threads race on. + */ + for (i = 0; i < 8; i++) { + char sname[64]; + snprintf(sname, sizeof(sname), "limit-t%d", i); + streams[i] = cio_stream_create(ctx, sname, CIO_STORE_FS); + TEST_CHECK(streams[i] != NULL); + + args[i].ctx = ctx; + args[i].st = streams[i]; + args[i].barrier = &barrier; + args[i].chunk = NULL; + args[i].tid = i; + args[i].errors = 0; + } + + for (round = 0; round < 20; round++) { + ups = 0; + + for (i = 0; i < 8; i++) { + args[i].chunk = NULL; + args[i].errors = 0; + TEST_CHECK(pthread_create(&threads[i], NULL, limit_worker, + &args[i]) == 0); + } + for (i = 0; i < 8; i++) { + pthread_join(threads[i], NULL); + TEST_CHECK(args[i].errors == 0); + if (args[i].chunk != NULL && + cio_chunk_is_up(args[i].chunk) == CIO_TRUE) { + ups++; + } + } + + /* exactly max_chunks_up opens must have succeeded, never more */ + TEST_CHECK(ups == 4); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 4); + + for (i = 0; i < 8; i++) { + if (args[i].chunk) { + cio_chunk_close(args[i].chunk, CIO_TRUE); + } + } + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 0); + } + + pthread_barrier_destroy(&barrier); + cio_destroy(ctx); +} +#endif + +#ifdef _WIN32 +/* ------------------------------------------------------------------------- */ +/* 10c. the same max_chunks_up race on the win32 thread/atomic path */ +/* ------------------------------------------------------------------------- */ +struct limit_thread_win { + struct cio_ctx *ctx; + struct cio_stream *st; + struct cio_chunk *chunk; + int tid; + int errors; +}; + +static DWORD WINAPI limit_worker_win(LPVOID arg) +{ + struct limit_thread_win *t = arg; + char name[64]; + int err = 0; + + snprintf(name, sizeof(name), "limit-%d.flb", t->tid); + + t->chunk = cio_chunk_open(t->ctx, t->st, name, CIO_OPEN_RW, 4096, &err); + if (!t->chunk) { + t->errors++; + } + + return 0; +} + +static void test_pool_threaded_max_chunks_win() +{ + int i; + int round; + int ups; + HANDLE threads[8]; + struct limit_thread_win args[8]; + struct cio_ctx *ctx; + struct cio_stream *streams[8]; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(16, 0); + TEST_CHECK(ctx != NULL); + cio_set_max_chunks_up(ctx, 4); + + for (i = 0; i < 8; i++) { + char sname[64]; + snprintf(sname, sizeof(sname), "limit-t%d", i); + streams[i] = cio_stream_create(ctx, sname, CIO_STORE_FS); + TEST_CHECK(streams[i] != NULL); + + args[i].ctx = ctx; + args[i].st = streams[i]; + args[i].chunk = NULL; + args[i].tid = i; + args[i].errors = 0; + } + + for (round = 0; round < 20; round++) { + ups = 0; + + /* create every thread suspended so they all race from ResumeThread */ + for (i = 0; i < 8; i++) { + args[i].chunk = NULL; + args[i].errors = 0; + threads[i] = CreateThread(NULL, 0, limit_worker_win, &args[i], + CREATE_SUSPENDED, NULL); + TEST_CHECK(threads[i] != NULL); + } + for (i = 0; i < 8; i++) { + ResumeThread(threads[i]); + } + WaitForMultipleObjects(8, threads, TRUE, INFINITE); + + for (i = 0; i < 8; i++) { + CloseHandle(threads[i]); + TEST_CHECK(args[i].errors == 0); + if (args[i].chunk != NULL && + cio_chunk_is_up(args[i].chunk) == CIO_TRUE) { + ups++; + } + } + + TEST_CHECK(ups == 4); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 4); + + for (i = 0; i < 8; i++) { + if (args[i].chunk) { + cio_chunk_close(args[i].chunk, CIO_TRUE); + } + } + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 0); + } + + cio_destroy(ctx); +} +#endif + +/* ------------------------------------------------------------------------- */ +/* 10. pool size option is clamped to the supported maximum */ +/* ------------------------------------------------------------------------- */ +static void test_pool_option_clamp() +{ + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(100000, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 4096); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 11. invalid pool sizes fall back to disabled */ +/* ------------------------------------------------------------------------- */ +static void test_pool_option_invalid() +{ + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(-5, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 0); + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 12. a single slot still recycles correctly */ +/* ------------------------------------------------------------------------- */ +static void test_pool_single_slot() +{ + int i; + int n; + char name[64]; + char payload[64]; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(1, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 1); + + for (i = 0; i < 10; i++) { + snprintf(name, sizeof(name), "one-%04d.flb", i); + snprintf(payload, sizeof(payload), "one-%d", i); + n = write_and_close(ctx, st, name, payload, CIO_TRUE); + TEST_CHECK(n == 0); + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 1); + } + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 13. memory streams ignore the option and keep valid pool state */ +/* ------------------------------------------------------------------------- */ +static void test_pool_mem_stream_ignores_option() +{ + int err; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(8, 0); + TEST_CHECK(ctx != NULL); + + st = cio_stream_create(ctx, "mem-pool-test", CIO_STORE_MEM); + TEST_CHECK(st != NULL); + TEST_CHECK(st->pool_size == 0); + TEST_CHECK(st->pool_busy == NULL); + + err = write_and_close(ctx, st, "mem-chunk.flb", "mem-payload", CIO_TRUE); + TEST_CHECK(err == 0); + + /* destroy exercises pool_destroy() for the memory stream */ + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 14. delete by logical name resolves the real slot file and frees the slot */ +/* ------------------------------------------------------------------------- */ +static void test_pool_delete_by_name() +{ + int err; + int n; + struct cio_chunk *chunk; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(4, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + + /* keep the chunk loaded, then delete it by its logical name */ + chunk = cio_chunk_open(ctx, st, "by-name.flb", CIO_OPEN, 4096, &err); + TEST_CHECK(chunk != NULL); + TEST_CHECK(cio_chunk_write(chunk, "by-name-payload", 15) == CIO_OK); + TEST_CHECK(cio_chunk_delete(ctx, st, "by-name.flb") == CIO_OK); + + /* exactly one (empty) slot file remains, no stale content */ + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 1); + + /* the slot is reusable and the new chunk has no stale data */ + err = write_and_close(ctx, st, "next.flb", "next-payload", CIO_TRUE); + TEST_CHECK(err == 0); + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 1); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 14b. up-chunk counter balance: limit, failed maps and failed opens */ +/* ------------------------------------------------------------------------- */ +static void test_pool_counter_balance() +{ + int err; + int fd; + size_t size; + char *buf = NULL; + char path[512]; + struct cio_chunk *a; + struct cio_chunk *b; + struct cio_chunk *loaded; + struct cio_ctx *ctx; + struct cio_stream *st; + struct cio_options opts; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(4, 0); + TEST_CHECK(ctx != NULL); + cio_set_max_chunks_up(ctx, 1); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + + /* A is up, B hits the limit and stays down */ + a = cio_chunk_open(ctx, st, "counter-a.flb", CIO_OPEN_RW, 4096, &err); + TEST_CHECK(a != NULL && cio_chunk_is_up(a) == CIO_TRUE); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 1); + + b = cio_chunk_open(ctx, st, "counter-b.flb", CIO_OPEN_RW, 4096, &err); + TEST_CHECK(b != NULL && cio_chunk_is_up(b) == CIO_FALSE); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 1); + + /* a read on a down chunk fails and must not touch the counter */ + TEST_CHECK(cio_meta_size(b) == -1); + TEST_CHECK(cio_chunk_get_content(b, &buf, &size) != CIO_OK); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 1); + + /* the non-enforced up/down pair stays balanced */ + TEST_CHECK(cio_chunk_up_force(b) == CIO_OK); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 2); + TEST_CHECK(cio_chunk_down(b) == CIO_OK); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 1); + + cio_chunk_close(b, CIO_TRUE); + cio_chunk_close(a, CIO_TRUE); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 0); + + /* a failed open of a corrupt slot must not consume an up-slot */ + snprintf(path, sizeof(path), "%s%s/slot-000000.flb", + CIO_POOL_ENV, CIO_POOL_NAME); + fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600); + TEST_CHECK(fd != -1); + if (fd != -1) { + TEST_CHECK(write(fd, "garbage-payload", 15) == 15); + close(fd); + } + a = cio_chunk_open(ctx, st, "slot-000000.flb", CIO_OPEN_RD, 0, &err); + TEST_CHECK(a == NULL); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 0); + + cio_destroy(ctx); + cio_utils_recursive_delete(CIO_POOL_ENV); + + /* pool disabled: delete-by-name keeps stock semantics (chunk untouched) */ + cio_options_init(&opts); + opts.root_path = CIO_POOL_ENV; + opts.log_cb = log_cb; + opts.flags = CIO_CHECKSUM; + ctx = cio_create(&opts); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + cio_set_max_chunks_up(ctx, 8); + + loaded = cio_chunk_open(ctx, st, "loaded.flb", CIO_OPEN_RW, 4096, &err); + TEST_CHECK(loaded != NULL && cio_chunk_is_up(loaded) == CIO_TRUE); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 1); + + /* + * Delete-by-name must not close/free a loaded chunk when the pool is + * disabled (stock semantics); Windows cannot unlink an open file, so only + * the chunk state is asserted here, not the delete return value. + */ + cio_chunk_delete(ctx, st, "loaded.flb"); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 1); + TEST_CHECK(mk_list_size(&st->chunks) == 1); + TEST_CHECK(cio_chunk_is_up(loaded) == CIO_TRUE); + + cio_chunk_close(loaded, CIO_FALSE); + TEST_CHECK(cio_chunk_counter_total_up_get(ctx) == 0); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 15. a failed truncate keeps the slot busy (no stale data reuse) */ +/* ------------------------------------------------------------------------- */ +#ifndef _WIN32 +static void test_pool_truncate_failure_keeps_slot() +{ + int err; + char path[512]; + struct cio_chunk *chunk; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + ctx = pool_ctx_create(1, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + + chunk = cio_chunk_open(ctx, st, "keep-slot.flb", CIO_OPEN, 4096, &err); + TEST_CHECK(chunk != NULL); + TEST_CHECK(cio_chunk_write(chunk, "data", 4) == CIO_OK); + + /* replace the slot file with a directory so truncation fails */ + snprintf(path, sizeof(path), "%s%s/slot-000000.flb", + CIO_POOL_ENV, CIO_POOL_NAME); + TEST_CHECK(unlink(path) == 0); + TEST_CHECK(mkdir(path, 0755) == 0); + + /* delete path: truncate fails, so the slot must stay busy */ + cio_chunk_close(chunk, CIO_TRUE); + TEST_CHECK((st->pool_busy[0] & 0x01) != 0); + + TEST_CHECK(rmdir(path) == 0); + cio_destroy(ctx); +} +#endif + +/* ------------------------------------------------------------------------- */ +/* 16. shrinking the pool: stale slots are cleaned, live ones stay usable */ +/* ------------------------------------------------------------------------- */ +static void test_pool_shrunk_pool_cleanup() +{ + int i; + int n; + int err; + int fd; + char name[64]; + char path[512]; + struct stat sb; + struct mk_list *head; + struct mk_list *tmp; + struct cio_chunk *ch; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + /* phase 1: pool of 8 leaves 8 live slot files behind */ + ctx = pool_ctx_create(8, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + for (i = 0; i < 8; i++) { + snprintf(name, sizeof(name), "live-%d.flb", i); + err = write_and_close(ctx, st, name, "shrunk-payload", CIO_FALSE); + TEST_CHECK(err == 0); + } + /* plus one empty slot from an even larger pool */ + snprintf(path, sizeof(path), "%s%s/slot-000099.flb", + CIO_POOL_ENV, CIO_POOL_NAME); + fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600); + TEST_CHECK(fd != -1); + if (fd != -1) { + close(fd); + } + cio_destroy(ctx); + TEST_CHECK(count_stream_files(CIO_POOL_NAME) == 9); + + /* phase 2: pool shrunk to 2 */ + ctx = pool_ctx_create(2, 0); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_get(ctx, CIO_POOL_NAME); + TEST_CHECK(st != NULL); + + /* the empty out-of-range slot was unlinked during the scan */ + TEST_CHECK(stat(path, &sb) != 0); + + /* delete all recovered chunks: in-range slots stay, others are unlinked */ + mk_list_foreach_safe(head, tmp, &st->chunks) { + ch = mk_list_entry(head, struct cio_chunk, _head); + cio_chunk_close(ch, CIO_TRUE); + } + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 2); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 17. empty slot left by a crash after truncate is reused */ +/* ------------------------------------------------------------------------- */ +static void test_pool_crash_after_truncate() +{ + int err; + int n; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + /* phase 1: a delivered chunk leaves an empty slot file behind */ + ctx = pool_ctx_create(2, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + err = write_and_close(ctx, st, "delivered.flb", "delivered", CIO_TRUE); + TEST_CHECK(err == 0); + cio_destroy(ctx); + TEST_CHECK(count_stream_files(CIO_POOL_NAME) == 1); + + /* phase 2: scan must treat it as a free slot and reuse it */ + ctx = pool_ctx_create(2, 0); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_get(ctx, CIO_POOL_NAME); + TEST_CHECK(st != NULL); + TEST_CHECK(mk_list_size(&st->chunks) == 0); + + err = write_and_close(ctx, st, "reuse.flb", "reuse-payload", CIO_TRUE); + TEST_CHECK(err == 0); + n = count_stream_files(CIO_POOL_NAME); + TEST_CHECK(n == 1); + + cio_destroy(ctx); +} + +/* ------------------------------------------------------------------------- */ +/* 18. partial chunk from a crash mid-write is discarded, pool stays usable */ +/* ------------------------------------------------------------------------- */ +static void test_pool_crash_mid_write() +{ + int err; + int fd; + char path[512]; + struct stat sb; + struct cio_ctx *ctx; + struct cio_stream *st; + + printf("\n"); + + cio_utils_recursive_delete(CIO_POOL_ENV); + + /* phase 1: write a valid chunk with the pool disabled */ + ctx = pool_ctx_create(0, 0); + TEST_CHECK(ctx != NULL); + st = cio_stream_create(ctx, CIO_POOL_NAME, CIO_STORE_FS); + TEST_CHECK(st != NULL); + err = write_and_close(ctx, st, "partial.flb", "partial-payload", CIO_FALSE); + TEST_CHECK(err == 0); + cio_destroy(ctx); + + snprintf(path, sizeof(path), "%s%s/partial.flb", + CIO_POOL_ENV, CIO_POOL_NAME); + TEST_CHECK(stat(path, &sb) == 0); + fd = open(path, O_WRONLY); + TEST_CHECK(fd != -1); + if (fd != -1) { +#ifdef _WIN32 + TEST_CHECK(_chsize(fd, 5) == 0); +#else + TEST_CHECK(ftruncate(fd, 5) == 0); +#endif + close(fd); + } + + /* phase 2: the partial chunk is dropped, the stream keeps working */ + ctx = pool_ctx_create(2, CIO_DELETE_IRRECOVERABLE); + TEST_CHECK(ctx != NULL); + TEST_CHECK(cio_load(ctx, ".flb") == 0); + st = cio_stream_get(ctx, CIO_POOL_NAME); + TEST_CHECK(st != NULL); + TEST_CHECK(mk_list_size(&st->chunks) == 0); + TEST_CHECK(stat(path, &sb) != 0); /* irrecoverable file removed */ + + err = write_and_close(ctx, st, "after.flb", "after-payload", CIO_TRUE); + TEST_CHECK(err == 0); + + cio_destroy(ctx); +} + +TEST_LIST = { + {"pool_slot_name_parse", test_pool_slot_name_parse}, + {"pool_disabled_legacy", test_pool_disabled_legacy}, + {"pool_reuse_bounds_files", test_pool_reuse_bounds_files}, + {"pool_backlog_fallback", test_pool_backlog_fallback}, + {"pool_recovery", test_pool_recovery}, + {"pool_recovery_legacy_name", test_pool_recovery_legacy_name}, + {"pool_corrupt_slot", test_pool_corrupt_slot}, + {"pool_option_clamp", test_pool_option_clamp}, + {"pool_option_invalid", test_pool_option_invalid}, + {"pool_single_slot", test_pool_single_slot}, + {"pool_mem_stream_ignores_option", test_pool_mem_stream_ignores_option}, + {"pool_delete_by_name", test_pool_delete_by_name}, + {"pool_counter_balance", test_pool_counter_balance}, +#ifndef _WIN32 + {"pool_truncate_failure_keeps_slot", test_pool_truncate_failure_keeps_slot}, +#endif + {"pool_shrunk_pool_cleanup", test_pool_shrunk_pool_cleanup}, + {"pool_crash_after_truncate", test_pool_crash_after_truncate}, + {"pool_crash_mid_write", test_pool_crash_mid_write}, +#ifndef _WIN32 + {"pool_lock_threads", test_pool_lock_threads}, + {"pool_threaded_chunks", test_pool_threaded_chunks}, + {"pool_threaded_max_chunks", test_pool_threaded_max_chunks}, +#endif +#ifdef _WIN32 + {"pool_threaded_max_chunks_win", test_pool_threaded_max_chunks_win}, +#endif + { 0 } +};