diff --git a/HISTORY b/HISTORY index 48cb5f62..8bdf8716 100644 --- a/HISTORY +++ b/HISTORY @@ -5,6 +5,7 @@ SnapRAID HISTORY =========== * Added a new "dup" command to find all the duplicate files. * Added a new option "--filter-nohidden" to exclude hidden files. + * Faster and parallel writing of content files. * The example configuration files now put the content files in the data disks instead than in the parity disks. diff --git a/TODO b/TODO index 18b2a4bb..1f9b1bf1 100644 --- a/TODO +++ b/TODO @@ -10,13 +10,6 @@ This is the list of TODO items for SnapRAID. * In fix an existing symlink with the same name of a file to be recovered may stop the process making the create() operation to fail. -* Check why saving the content files is so slow. fprintf() may be the bottleneck. - -* Save all the content files in parallel. We can use a single thread writing -the same stuffs to all the file descriptors. -This is still safe, as the written data is going to replace the old one -with an atomic rename operation. - * Recheck the 2nd issue reported at: https://sourceforge.net/projects/snapraid/forums/forum/1677233/topic/4747897 * If a directory exists with the same name of a parity/content file be more explicative diff --git a/state.c b/state.c index a50eefac..58235b9e 100644 --- a/state.c +++ b/state.c @@ -280,6 +280,11 @@ void state_config(struct snapraid_state* state, const char* path, int verbose, i exit(EXIT_FAILURE); } + if (strcmp(buffer, "/dev/null") == 0 || strcmp(buffer, "NUL") == 0 || strcmp(buffer, "nul") == 0) { + fprintf(stderr, "You cannot use the null device as 'content' specification in '%s' at line %u\n", path, line); + exit(EXIT_FAILURE); + } + /* get the device of the directory containing the content file */ pathimport(device, sizeof(device), buffer); slash = strrchr(device, '/'); @@ -961,51 +966,79 @@ void state_read(struct snapraid_state* state) } } -static void state_write_one(struct snapraid_state* state, const char* path, unsigned* out_count_file, unsigned* out_count_block, unsigned* out_count_link) +void state_write(struct snapraid_state* state) { - FILE* f; - char tmp[PATH_MAX]; + STREAM* f; unsigned count_file; unsigned count_block; unsigned count_link; + unsigned count_content; tommy_node* i; - int ret; + unsigned k; count_file = 0; count_block = 0; count_link = 0; - printf("Saving state to %s...\n", path); + /* count the content files */ + count_content = 0; + i = tommy_list_head(&state->contentlist); + while (i) { + struct snapraid_content* content = i->data; + printf("Saving state to %s...\n", content->content); + ++count_content; + i = i->next; + } - /* ignore some special files, this is an undocument feature */ - if (strcmp(path, "/dev/null") == 0 || strcmp(path, "NUL") == 0 || strcmp(path, "nul") == 0) - return; - - pathprint(tmp, sizeof(tmp), "%s.tmp", path); - f = fopen(tmp, "wt"); + /* open all the content files */ + f = sopen_multi_write(count_content); if (!f) { - fprintf(stderr, "Error opening for writing the content file '%s'\n", tmp); + fprintf(stderr, "Error opening the content files\n"); exit(EXIT_FAILURE); } - ret = fprintf(f, "blksize %u\n", state->block_size); - if (ret < 0) { - fprintf(stderr, "Error writing the content file '%s' in fprintf(). %s.\n", tmp, strerror(errno)); + k = 0; + i = tommy_list_head(&state->contentlist); + while (i) { + struct snapraid_content* content = i->data; + char tmp[PATH_MAX]; + pathprint(tmp, sizeof(tmp), "%s.tmp", content->content); + if (sopen_multi_file(f, k, tmp) != 0) { + fprintf(stderr, "Error operning the content file '%s'. %s.\n", tmp, strerror(errno)); + exit(EXIT_FAILURE); + } + ++k; + i = i->next; + } + + sputs("blksize ", f); + sputu32(state->block_size, f); + sputeol(f); + if (serror(f)) { + fprintf(stderr, "Error writing the content file '%s'. %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } - ret = fprintf(f, "checksum %s\n", state->hash == HASH_MD5 ? "md5" : "murmur3"); - if (ret < 0) { - fprintf(stderr, "Error writing the content file '%s' in fprintf(). %s.\n", tmp, strerror(errno)); + if (state->hash == HASH_MD5) + sputs("checksum md5", f); + else + sputs("checksum murmur3", f); + sputeol(f); + if (serror(f)) { + fprintf(stderr, "Error writing the content file '%s'. %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } /* for each map */ for(i=state->maplist;i!=0;i=i->next) { struct snapraid_map* map = i->data; - ret = fprintf(f, "map %s %u\n", map->name, map->position); - if (ret < 0) { - fprintf(stderr, "Error writing the content file '%s' in fprintf(). %s.\n", tmp, strerror(errno)); + sputs("map ", f); + sputs(map->name, f); + sputc(' ', f); + sputu32(map->position, f); + sputeol(f); + if (serror(f)) { + fprintf(stderr, "Error writing the content file '%s'. %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } } @@ -1027,33 +1060,42 @@ static void state_write_one(struct snapraid_state* state, const char* path, unsi mtime = file->mtime; inode = file->inode, - ret = fprintf(f,"file %s %"PRIu64" %"PRIu64" %"PRIu64" %s\n", disk->name, size, mtime, inode, file->sub); - if (ret < 0) { - fprintf(stderr, "Error writing the content file '%s' in fprintf(). %s.\n", tmp, strerror(errno)); + sputs("file ", f); + sputs(disk->name, f); + sputc(' ', f); + sputu64(size, f); + sputc(' ', f); + sputu64(mtime, f); + sputc(' ', f); + sputu64(inode, f); + sputc(' ', f); + sputs(file->sub, f); + sputeol(f); + if (serror(f)) { + fprintf(stderr, "Error writing the content file '%s'. %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } /* for each block */ for(k=0;kblockmax;++k) { struct snapraid_block* block = &file->blockvec[k]; - const char* tag; if (block_flag_has(block, BLOCK_HAS_PARITY)) { - tag = "blk"; + sputs("blk ", f); } else { - tag = "inv"; + sputs("inv ", f); } if (block_flag_has(block, BLOCK_HAS_HASH)) { - char s_hash[HASH_SIZE*2+1]; - strenchex(s_hash, block->hash, HASH_SIZE); - s_hash[HASH_SIZE*2] = 0; - ret = fprintf(f, "%s %u %s\n", tag, block->parity_pos, s_hash); + sputu32(block->parity_pos, f); + sputc(' ', f); + sputhex(block->hash, HASH_SIZE, f); } else { - ret = fprintf(f, "%s %u\n", tag, block->parity_pos); + sputu32(block->parity_pos, f); } - if (ret < 0) { - fprintf(stderr, "Error writing the content file '%s' in fprintf(). %s.\n", tmp, strerror(errno)); + sputeol(f); + if (serror(f)) { + fprintf(stderr, "Error writing the content file '%s'. %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } @@ -1067,9 +1109,16 @@ static void state_write_one(struct snapraid_state* state, const char* path, unsi for(j=disk->linklist;j!=0;j=j->next) { struct snapraid_link* link = j->data; - ret = fprintf(f,"symlink %s %s\nto %s\n", disk->name, link->sub, link->linkto); - if (ret < 0) { - fprintf(stderr, "Error writing the content file '%s' in fprintf(). %s.\n", tmp, strerror(errno)); + sputs("symlink ", f); + sputs(disk->name, f); + sputc(' ', f); + sputs(link->sub, f); + sputeol(f); + sputs("to ", f); + sputs(link->linkto, f); + sputeol(f); + if (serror(f)) { + fprintf(stderr, "Error writing the content file '%s'. %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } @@ -1080,52 +1129,33 @@ static void state_write_one(struct snapraid_state* state, const char* path, unsi /* Use the sequence fflush() -> fsync() -> fclose() -> rename() to ensure */ /* than even in a system crash event we have one valid copy of the file. */ - if (fflush(f) != 0) { - fprintf(stderr, "Error writing the content file '%s', in fflush(). %s.\n", tmp, strerror(errno)); + if (sflush(f) != 0) { + fprintf(stderr, "Error writing the content file '%s', in flush(). %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } #if HAVE_FSYNC - if (fsync(fileno(f)) != 0) { - fprintf(stderr, "Error writing the content file '%s' in fsync(). %s.\n", tmp, strerror(errno)); + if (ssync(f) != 0) { + fprintf(stderr, "Error writing the content file '%s' in sync(). %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } #endif - if (fclose(f) != 0) { - fprintf(stderr, "Error writing the content file '%s' in close(). %s.\n", tmp, strerror(errno)); + if (sclose(f) != 0) { + fprintf(stderr, "Error writing the content file '%s' in close(). %s.\n", serrorfile(f), strerror(errno)); exit(EXIT_FAILURE); } - if (rename(tmp, path) != 0) { - fprintf(stderr, "Error renaming the content file '%s' to '%s' in rename(). %s.\n", tmp, path, strerror(errno)); - exit(EXIT_FAILURE); - } - - if (out_count_file) - *out_count_file = count_file; - if (out_count_block) - *out_count_block = count_block; - if (out_count_link) - *out_count_link = count_link; -} - -void state_write(struct snapraid_state* state) -{ - unsigned count_file; - unsigned count_block; - unsigned count_link; - tommy_node* node; - - count_file = 0; - count_block = 0; - count_link = 0; - - node = tommy_list_head(&state->contentlist); - while (node) { - struct snapraid_content* content = node->data; - state_write_one(state, content->content, &count_file, &count_block, &count_link); - node = node->next; + i = tommy_list_head(&state->contentlist); + while (i) { + struct snapraid_content* content = i->data; + char tmp[PATH_MAX]; + pathprint(tmp, sizeof(tmp), "%s.tmp", content->content); + if (rename(tmp, content->content) != 0) { + fprintf(stderr, "Error renaming the content file '%s' to '%s' in rename(). %s.\n", tmp, content->content, strerror(errno)); + exit(EXIT_FAILURE); + } + i = i->next; } if (state->verbose) { diff --git a/state.h b/state.h index 514c23c3..cb7ea40e 100644 --- a/state.h +++ b/state.h @@ -65,12 +65,12 @@ void state_init(struct snapraid_state* state); void state_done(struct snapraid_state* state); /** - * Read the configuration file. + * Reads the configuration file. */ void state_config(struct snapraid_state* state, const char* path, int verbose, int gui, int force_zero, int force_empty, int filter_hidden, int find_by_name, int expect_unrecoverable, int expect_recoverable, int skip_device); /** - * Read the state. + * Reads the state. */ void state_read(struct snapraid_state* state); diff --git a/util.c b/util.c index 38f9a210..12ff7c57 100644 --- a/util.c +++ b/util.c @@ -90,8 +90,13 @@ STREAM* sopen_read(const char* file) { STREAM* s = malloc_nofail(sizeof(STREAM)); - s->f = open(file, O_RDONLY | O_BINARY | O_SEQUENTIAL); - if (s->f == -1) { + s->handle_size = 1; + s->handle = malloc_nofail(sizeof(struct stream_handle)); + + pathcpy(s->handle[0].path, sizeof(s->handle[0].path), file); + s->handle[0].f = open(file, O_RDONLY | O_BINARY | O_SEQUENTIAL); + if (s->handle[0].f == -1) { + free(s->handle); free(s); return 0; } @@ -99,33 +104,117 @@ STREAM* sopen_read(const char* file) s->buffer = malloc_nofail(STREAM_SIZE); s->pos = s->buffer; s->end = s->buffer; - s->state = STREAM_OK; + s->state = STREAM_STATE_READ; + s->state_index = 0; return s; } -void sclose(STREAM* s) +STREAM* sopen_multi_write(unsigned count) { - close(s->f); - free(s->buffer); - free(s); + unsigned i; + + STREAM* s = malloc_nofail(sizeof(STREAM)); + + s->handle_size = count; + s->handle = malloc_nofail(count * sizeof(struct stream_handle)); + + for(i=0;ihandle[i].f = -1; + + s->buffer = malloc_nofail(STREAM_SIZE); + s->pos = s->buffer; + s->end = s->buffer + STREAM_SIZE; + s->state = STREAM_STATE_WRITE; + s->state_index = 0; + + return s; } -int sflow(STREAM* s) +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); + + f = open(file, O_WRONLY | O_CREAT | O_TRUNC | O_BINARY | O_SEQUENTIAL, 0600); + if (f == -1) { + return -1; + } + +#if HAVE_POSIX_FADVISE + /* advise sequential access */ + ret = posix_fadvise(f, 0, 0, POSIX_FADV_SEQUENTIAL); + if (ret != 0) { + close(f); + return -1; + } +#endif + + s->handle[i].f = f; + + return 0; +} + +STREAM* sopen_write(const char* file) +{ + STREAM* s; + + s = sopen_multi_write(1); + if (!s) + return 0; + + if (sopen_multi_file(s, 0, file) != 0) { + sclose(s); + return 0; + } + + return s; +} + +int sclose(STREAM* s) +{ + int fail = 0; + unsigned i; + + if (s->state == STREAM_STATE_WRITE) { + if (sflush(s) != 0) + fail = 1; + } + + for(i=0;ihandle_size;++i) { + if (close(s->handle[i].f) != 0) + fail = 1; + } + + free(s->handle); + free(s->buffer); + free(s); + + if (fail) + return -1; + + return 0; +} + +int sfill(STREAM* s) { ssize_t ret; - if (s->state != STREAM_OK) + if (s->state != STREAM_STATE_READ) return EOF; - ret = read(s->f, s->buffer, STREAM_SIZE); + ret = read(s->handle[0].f, s->buffer, STREAM_SIZE); if (ret < 0) { - s->state = STREAM_ERROR; + s->state = STREAM_STATE_ERROR; return EOF; } if (ret == 0) { - s->state = STREAM_EOF; + s->state = STREAM_STATE_EOF; return EOF; } @@ -135,6 +224,34 @@ int sflow(STREAM* s) return *s->pos++; } +int sflush(STREAM* s) +{ + ssize_t ret; + ssize_t size; + unsigned i; + + if (s->state != STREAM_STATE_WRITE) + return EOF; + + size = s->pos - s->buffer; + if (!size) + return 0; + + for(i=0;ihandle_size;++i) { + ret = write(s->handle[i].f, s->buffer, size); + + if (ret != size) { + s->state = STREAM_STATE_ERROR; + s->state_index = i; + return EOF; + } + } + + s->pos = s->buffer; + + return 0; +} + int sgettok(STREAM* f, char* str, int size) { char* i = str; @@ -355,6 +472,93 @@ int sgethex(STREAM* f, void* void_data, int size) return 0; } +int sputs(const char* str, STREAM* f) +{ + while (*str) { + if (sputc(*str, f) != 0) + return -1; + ++str; + } + + return 0; +} + +int sputu32(uint32_t value, STREAM* s) +{ + char buf[16]; + int i; + + if (!value) + return sputc('0', s); + + i = sizeof(buf); + buf[--i] = 0; + + while (value) { + buf[--i] = (value % 10) + '0'; + value /= 10; + } + + return sputs(buf + i, s); +} + +int sputu64(uint64_t value, STREAM* s) +{ + char buf[32]; + int i; + + if (!value) + return sputc('0', s); + if (value <= 0xFFFFFFFF) + return sputu32((uint32_t)value, s); + + i = sizeof(buf); + buf[--i] = 0; + + while (value) { + buf[--i] = (value % 10) + '0'; + value /= 10; + } + + return sputs(buf + i, s); +} + +int sputhex(const void* void_data, int size, STREAM* f) +{ + const unsigned char* data = void_data; + + while (size) { + unsigned b = *data; + + if (sputc(strhexset[b >> 4], f) != 0) + return -1; + if (sputc(strhexset[b & 0xF], f) != 0) + return -1; + + ++data; + --size; + } + + return 0; +} + +#if HAVE_FSYNC +int ssync(STREAM* s) +{ + unsigned i; + + for(i=0;ihandle_size;++i) { + if (fsync(s->handle[i].f) != 0) { + s->state = STREAM_STATE_ERROR; + s->state_index = i; + return -1; + } + } + + return 0; +} +#endif + /****************************************************************************/ /* path */ diff --git a/util.h b/util.h index 16763ef7..2ab16c96 100644 --- a/util.h +++ b/util.h @@ -34,18 +34,26 @@ char* strdechex(void* void_data, unsigned data_len, char* str); /****************************************************************************/ /* stream */ -#define STREAM_SIZE (64*1024) +#define STREAM_SIZE (64*1024) /**< Size of the buffer of the stream. */ -#define STREAM_OK 0 -#define STREAM_ERROR -1 -#define STREAM_EOF 1 +#define STREAM_STATE_READ 0 /**< The stream is in a normal state of read. */ +#define STREAM_STATE_WRITE 1 /**< The stream is in a normal state of write. */ +#define STREAM_STATE_ERROR -1 /**< An error was encoutered. */ +#define STREAM_STATE_EOF 2 /**< The end of file was encoutered. */ + +struct stream_handle { + int f; /**< Handle of the file. */ + char path[PATH_MAX]; /**< Path of the file. */ +}; struct stream { - unsigned char* buffer; - unsigned char* pos; - unsigned char* end; - int state; - int f; + unsigned char* buffer; /**< Buffer of the stream. */ + unsigned char* pos; /**< Current position in the buffer. */ + unsigned char* end; /**< End position of the buffer. */ + int state; /**< State of the stream. One of STREAM_STATE. */ + int state_index; /**< Index of the handle causing a state change. */ + unsigned handle_size; /**< Number of handles. */ + struct stream_handle* handle; /**< Set of handles. */ }; /** @@ -54,19 +62,42 @@ struct stream { typedef struct stream STREAM; /** - * Open a stream for reading. Like fopen(). + * Opens a stream for reading. Like fopen("r"). */ STREAM* sopen_read(const char* file); /** - * Close a stream. Like fclose(). + * Opens a stream for writing. Like fopen("w"). */ -void sclose(STREAM* s); +STREAM* sopen_write(const char* file); /** - * Fill the stream buffer and read a char. + * Opens a set of streams for writing. Like fopen("w"). */ -int sflow(STREAM* s); +STREAM* sopen_multi_write(unsigned count); + +/** + * Specifies the file to open. + */ +int sopen_multi_file(STREAM* s, unsigned i, const char* file); + +/** + * Closes a stream. Like fclose(). + */ +int sclose(STREAM* s); + +/** + * Fills the read stream buffer and read a char. + * \note Don't call this directly, but use sgetc(). + * \return The char read, or EOF on error. + */ +int sfill(STREAM* s); + +/** + * Flushes the write stream buffer. + * \return 0 on success, or EOF on error. + */ +int sflush(STREAM* s); /** * Checks if the buffer has enough data loaded. @@ -85,7 +116,7 @@ static inline unsigned char* sptrget(STREAM* s) } /** - * Set the current stream ptr. + * Sets the current stream ptr. */ static inline void sptrset(STREAM* s, unsigned char* ptr) { @@ -93,17 +124,49 @@ static inline void sptrset(STREAM* s, unsigned char* ptr) } /** - * Read a char. Like fgetc(). + * Checks the error status. Like ferror(). + */ +static inline int serror(STREAM* s) +{ + return s->state == STREAM_STATE_ERROR; +} + +/** + * Gets the index of the handle that caused the error. + */ +static inline int serrorindex(STREAM* s) +{ + return s->state_index; +} + +/** + * Gets the path of the handle that caused the error. + */ +static inline const char* serrorfile(STREAM* s) +{ + return s->handle[s->state_index].path; +} + +/** + * Sync the stream. Like fsync(). + */ +int ssync(STREAM* s); + +/****************************************************************************/ +/* get */ + +/** + * Reads a char. Like fgetc(). */ static inline int sgetc(STREAM* s) { if (s->pos == s->end) - return sflow(s); + return sfill(s); return *s->pos++; } /** - * Unread a char. + * Unreads a char. * Like ungetc() but you have to unget the same char read. */ static inline void sungetc(int c, STREAM* s) @@ -113,15 +176,7 @@ static inline void sungetc(int c, STREAM* s) } /** - * Check the error status. Like ferror(). - */ -static inline int serror(STREAM* s) -{ - return s->state == STREAM_ERROR; -} - -/** - * Get a char from a stream, ignoring one '\r'. + * Gets a char from a stream, ignoring one '\r'. */ static inline int sgeteol(STREAM* f) { @@ -135,7 +190,7 @@ static inline int sgeteol(STREAM* f) } /** - * Read all the spaces and tabs. + * Reads all the spaces and tabs. * Returns the number of spaces and tabs read. */ static inline int sgetspace(STREAM* f) @@ -154,15 +209,15 @@ static inline int sgetspace(STREAM* f) } /** - * Read until the first space or tab. + * Reads until the first space or tab. * Stops at the first ' ', '\t', '\n' or EOF. * Returns <0 if the buffer is too small, or the number of chars read. */ int sgettok(STREAM* f, char* str, int size); /** - * Read until the end of line. - * Stops at the first '\n' or EOF. + * Reads until the end of line. + * Stops at the first '\n' or EOF. Note that '\n' is left in the stream. * Returns <0 if the buffer is too small, or the number of chars read. */ int sgetline(STREAM* f, char* str, int size); @@ -173,25 +228,79 @@ int sgetline(STREAM* f, char* str, int size); int sgetlasttok(STREAM* f, char* str, int size); /** - * Read a 32 bit number. + * Reads a 32 bit number. * Stops at the first not digit char or EOF. * Returns <0 if there isn't enough to read. */ int sgetu32(STREAM* f, uint32_t* value); /** - * Read a 64 bit number. + * Reads a 64 bit number. * Stops at the first not digit char. * Returns <0 if there isn't enough to read. */ int sgetu64(STREAM* f, uint64_t* value); /** - * Read an hexadecimal string of fixed length. + * Reads an hexadecimal string of fixed length. * Returns <0 if there isn't enough to read. */ int sgethex(STREAM* f, void* data, int size); +/****************************************************************************/ +/* put */ + +/** + * Writes a char. Like fputc(). + * Returns 0 on succes or -1 on error. + */ +static inline int sputc(int c, STREAM* s) +{ + if (s->pos == s->end) { + if (sflush(s) != 0) + return -1; + } + *s->pos++ = c; + return 0; +} + +/** + * Writes a end of line. + * Returns 0 on succes or -1 on error. + */ +static inline int sputeol(STREAM* s) +{ +#ifdef _WIN32 + if (sputc('\r', s) != 0) + return -1; +#endif + return sputc('\n', s); +} + +/** + * Writes a string. + * Returns 0 on succes or -1 on error. + */ +int sputs(const char* str, STREAM* f); + +/** + * Writes a 32 bit number. + * Returns 0 on succes or -1 on error. + */ +int sputu32(uint32_t value, STREAM* s); + +/** + * Writes a 64 bit number. + * Returns 0 on succes or -1 on error. + */ +int sputu64(uint64_t value, STREAM* s); + +/** + * Writes a hexadecimal string of fixed length. + * Returns 0 on succes or -1 on error. + */ +int sputhex(const void* void_data, int size, STREAM* s); + /****************************************************************************/ /* path */