From d8ea998eb2cf2aec97a9590c6cfab17eb77b265c Mon Sep 17 00:00:00 2001 From: Andrea Mazzoleni Date: Wed, 9 Sep 2026 20:39:33 +0200 Subject: [PATCH] Support native advise modes in buffered streams --- cmdline/mkstream.c | 16 +++++--- cmdline/state.c | 20 ++++++++-- cmdline/stream.c | 91 ++++++++++++++++++++++++---------------------- cmdline/stream.h | 7 +++- 4 files changed, 80 insertions(+), 54 deletions(-) diff --git a/cmdline/mkstream.c b/cmdline/mkstream.c index 71068cbe..fdcf6074 100644 --- a/cmdline/mkstream.c +++ b/cmdline/mkstream.c @@ -10,7 +10,7 @@ #define BUFFER_MAX 64 #define STR_MAX 128 -void test(void) +void test(int advise_mode) { struct stream* s; char file[32]; @@ -24,7 +24,7 @@ void test(void) crc32c_init(); - s = sopen_multi_write(STREAM_MAX, STREAM_FLAGS_CRC); + s = sopen_multi_write(STREAM_MAX, advise_mode | STREAM_FLAGS_CRC); for (i = 0; i < STREAM_MAX; ++i) { snprintf(file, sizeof(file), "stream%u.bin", i); remove(file); @@ -104,7 +104,7 @@ void test(void) uint32_t get_crc_computed; snprintf(file, sizeof(file), "stream%u.bin", i); - s = sopen_read(file, STREAM_FLAGS_CRC); + s = sopen_read(file, advise_mode | STREAM_FLAGS_CRC); if (s == 0) { /* LCOV_EXCL_START */ exit(EXIT_FAILURE); @@ -193,7 +193,7 @@ void test(void) unsigned char buf[4]; snprintf(file, sizeof(file), "stream%u.bin", i); - s = sopen_read(file, STREAM_FLAGS_CRC); + s = sopen_read(file, advise_mode | STREAM_FLAGS_CRC); if (s == 0) { /* LCOV_EXCL_START */ exit(EXIT_FAILURE); @@ -248,7 +248,13 @@ int main(void) printf("Test stream buffer size %u\n", i); - test(); + test(ADVISE_DEFAULT); + } + + STREAM_SIZE = 64 * 1024; + for (i = ADVISE_SEQUENTIAL; i <= ADVISE_DISCARD_WINDOW; ++i) { + printf("Test stream advise mode %u\n", i); + test(i); } return 0; diff --git a/cmdline/state.c b/cmdline/state.c index 56c01fd7..6197a0a5 100644 --- a/cmdline/state.c +++ b/cmdline/state.c @@ -32,6 +32,18 @@ #define HAVE_MT_VERIFY 1 #endif +/* + * Start content-file writeback in bounded windows to avoid dirty-page + * throttling; the subsequent fsync() remains responsible for durability. + * + * Best test-rewrite serialization/write times with four 4 GB content files: + * ADVISE_SEQUENTIAL: 128s + * ADVISE_FLUSH: 86s + * ADVISE_DISCARD_WINDOW: 73s + * ADVISE_FLUSH_WINDOW: 61s + */ +#define CONTENT_WRITE_ADVISE ADVISE_FLUSH_WINDOW + const char* lev_name(unsigned l) { switch (l) { @@ -4653,7 +4665,7 @@ static void state_write_content(struct snapraid_state* state, uint32_t* out_crc) } } - f = sopen_write(tmp, STREAM_FLAGS_SEQUENTIAL | STREAM_FLAGS_CRC); + f = sopen_write(tmp, CONTENT_WRITE_ADVISE | STREAM_FLAGS_CRC); if (f == 0) { /* LCOV_EXCL_START */ log_fatal(errno, "Error opening the temporary content file '%s'. %s.\n", tmp, strerror(errno)); @@ -4805,7 +4817,7 @@ static void state_write_content(struct snapraid_state* state, uint32_t* out_crc) } /* open all the content files */ - f = sopen_multi_write(count_content, STREAM_FLAGS_SEQUENTIAL | STREAM_FLAGS_CRC); + f = sopen_multi_write(count_content, CONTENT_WRITE_ADVISE | STREAM_FLAGS_CRC); if (!f) { /* LCOV_EXCL_START */ log_fatal(errno, "Error opening the content files.\n"); @@ -5001,7 +5013,7 @@ void state_read(struct snapraid_state* state) msg_progress("Loading state from %s...\n", path); - f = sopen_read(path, STREAM_FLAGS_SEQUENTIAL | STREAM_FLAGS_CRC); + f = sopen_read(path, ADVISE_SEQUENTIAL | STREAM_FLAGS_CRC); if (f != 0) { /* if found stop the search */ break; @@ -5214,7 +5226,7 @@ static void state_verify_content(struct snapraid_state* state, uint32_t crc) STREAM* f; pathprint(tmp, sizeof(tmp), "%s.tmp", content->content); - f = sopen_read(tmp, STREAM_FLAGS_SEQUENTIAL | STREAM_FLAGS_CRC); + f = sopen_read(tmp, ADVISE_SEQUENTIAL | STREAM_FLAGS_CRC); if (f == 0) { /* LCOV_EXCL_START */ log_fatal(errno, "Error reopening the temporary content file '%s'. %s.\n", tmp, strerror(errno)); diff --git a/cmdline/stream.c b/cmdline/stream.c index aadf917a..8e5d6458 100644 --- a/cmdline/stream.c +++ b/cmdline/stream.c @@ -14,23 +14,22 @@ unsigned STREAM_SIZE = 64 * 1024; STREAM* sopen_read(const char* file, int flags) { -#if HAVE_POSIX_FADVISE - int ret; -#endif STREAM* s = malloc_nofail(sizeof(STREAM)); struct stat st; + int ret; s->handle_size = 1; s->handle = malloc_nofail(sizeof(struct stream_handle)); + s->advise = malloc_nofail(sizeof(struct advise_struct)); s->flags = flags; + advise_init(&s->advise[0], s->flags & STREAM_FLAGS_ADVISE_MASK); - int open_flags = O_RDONLY | O_BINARY; - if (s->flags & STREAM_FLAGS_SEQUENTIAL) - open_flags |= O_SEQUENTIAL; + int open_flags = O_RDONLY | O_BINARY | advise_flags(&s->advise[0]); pathcpy(s->handle[0].path, sizeof(s->handle[0].path), file); s->handle[0].f = open(file, open_flags); if (s->handle[0].f == -1) { + free(s->advise); free(s->handle); free(s); return 0; @@ -39,33 +38,23 @@ STREAM* sopen_read(const char* file, int flags) /* get file info */ if (fstat(s->handle[0].f, &st) != 0) { close(s->handle[0].f); + free(s->advise); free(s->handle); free(s); return 0; } s->size = st.st_size; - -#if HAVE_POSIX_FADVISE - if (s->flags & STREAM_FLAGS_SEQUENTIAL) { - /* advise sequential access */ - ret = posix_fadvise_wrapper(s->handle[0].f, 0, 0, POSIX_FADV_SEQUENTIAL); - if (ret == ENOSYS) { - log_error(errno, "WARNING! fadvise() is not supported in this platform. Performance may not be optimal!\n"); - /* call is not supported, like in armhf, see posix_fadvise manpage */ - ret = 0; - } - if (ret != 0) { - /* LCOV_EXCL_START */ - close(s->handle[0].f); - free(s->handle); - free(s); - errno = ret; /* posix_fadvise return the error code */ - return 0; - /* LCOV_EXCL_STOP */ - } + ret = advise_open(&s->advise[0], s->handle[0].f); + if (ret != 0) { + /* LCOV_EXCL_START */ + close(s->handle[0].f); + free(s->advise); + free(s->handle); + free(s); + return 0; + /* LCOV_EXCL_STOP */ } -#endif s->buffer_size = STREAM_SIZE; if (s->buffer_size > s->size) @@ -92,10 +81,13 @@ STREAM* sopen_multi_write(unsigned count, int flags) s->handle_size = count; s->handle = nalloc_nofail(count, sizeof(struct stream_handle)); + s->advise = nalloc_nofail(count, sizeof(struct advise_struct)); s->flags = flags; - for (i = 0; i < count; ++i) + for (i = 0; i < count; ++i) { s->handle[i].f = -1; + advise_init(&s->advise[i], s->flags & STREAM_FLAGS_ADVISE_MASK); + } s->buffer_size = STREAM_SIZE; s->buffer = malloc_nofail_test(s->buffer_size); @@ -115,16 +107,11 @@ STREAM* sopen_multi_write(unsigned count, int flags) int sopen_multi_file(STREAM* s, unsigned i, const char* file) { -#if HAVE_POSIX_FADVISE - int ret; -#endif int f; pathcpy(s->handle[i].path, sizeof(s->handle[i].path), file); - int open_flags = O_WRONLY | O_CREAT | O_EXCL | O_BINARY; - if (s->flags & STREAM_FLAGS_SEQUENTIAL) - open_flags |= O_SEQUENTIAL; ; + int open_flags = O_WRONLY | O_CREAT | O_EXCL | O_BINARY | advise_flags(&s->advise[i]); /* O_EXCL to be resilient ensure to always create a new file and not use a stale link to the original file */ f = open(file, open_flags, 0600); @@ -134,23 +121,17 @@ int sopen_multi_file(STREAM* s, unsigned i, const char* file) /* LCOV_EXCL_STOP */ } -#if HAVE_POSIX_FADVISE - if (s->flags & STREAM_FLAGS_SEQUENTIAL) { - /* advise sequential access */ - ret = posix_fadvise_wrapper(f, 0, 0, POSIX_FADV_SEQUENTIAL); - if (ret == ENOSYS) { - /* call is not supported, like in armhf, see posix_fadvise manpage */ - ret = 0; - } + { + int ret; + + ret = advise_open(&s->advise[i], f); if (ret != 0) { /* LCOV_EXCL_START */ close(f); - errno = ret; /* posix_fadvise return the error code */ return -1; /* LCOV_EXCL_STOP */ } } -#endif s->handle[i].f = f; @@ -185,6 +166,11 @@ int sclose(STREAM* s) for (i = 0; i < s->handle_size; ++i) { /* if open failed, there is nothing to close */ if (s->handle[i].f != -1) { + if (advise_close(&s->advise[i], s->handle[i].f) != 0) { + /* LCOV_EXCL_START */ + fail = 1; + /* LCOV_EXCL_STOP */ + } if (close(s->handle[i].f) != 0) { /* LCOV_EXCL_START */ fail = 1; @@ -193,6 +179,7 @@ int sclose(STREAM* s) } } + free(s->advise); free(s->handle); free(s->buffer); free(s); @@ -252,6 +239,13 @@ static int sfill(STREAM* s) } /* assume that it can be a shorter read, so ret < s->buffer_size doesn't mean that we reached EOF */ + if (advise_read(&s->advise[0], s->handle[0].f, s->offset, ret) != 0) { + /* LCOV_EXCL_START */ + s->state = STREAM_STATE_ERROR; + return EOF; + /* LCOV_EXCL_STOP */ + } + /* update the crc */ if (s->flags & STREAM_FLAGS_CRC) { s->crc_uncached = s->crc; @@ -352,6 +346,17 @@ int sflush(STREAM* s) } while (count < size); } + for (i = 0; i < s->handle_size; ++i) { + /* ADVISE_DISCARD modes only discard ranges after waiting for writeback. */ + if (advise_write(&s->advise[i], s->handle[i].f, s->offset, size) != 0) { + /* LCOV_EXCL_START */ + s->state = STREAM_STATE_ERROR; + s->state_index = i; + return EOF; + /* LCOV_EXCL_STOP */ + } + } + /* * Update the crc *after* writing the data. * diff --git a/cmdline/stream.h b/cmdline/stream.h index 9a963612..b444ca47 100644 --- a/cmdline/stream.h +++ b/cmdline/stream.h @@ -4,13 +4,14 @@ #ifndef __STREAM_H #define __STREAM_H +#include "support.h" #include "util.h" /****************************************************************************/ /* stream */ -#define STREAM_FLAGS_SEQUENTIAL 1 /**< Advise a squential read. */ -#define STREAM_FLAGS_CRC 2 /**< Enable the CRC computation. */ +#define STREAM_FLAGS_ADVISE_MASK 0xFF /**< Low bits containing one of ADVISE_*. */ +#define STREAM_FLAGS_CRC 0x100 /**< Enable the CRC computation. */ /** * Size of the buffer of the stream. @@ -83,6 +84,8 @@ struct stream { * In writing, it's all the data wrote calling sput() functions. */ uint32_t crc_stream; + + struct advise_struct* advise; /**< Cache advice state for each handle. */ }; /**