mirror of
https://github.com/amadvance/snapraid.git
synced 2026-10-26 10:51:54 -04:00
Support native advise modes in buffered streams
This commit is contained in:
Andrea Mazzoleni
parent
b762ab9e73
commit
d8ea998eb2
4 changed files with 80 additions and 54 deletions
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue