From 97e7f8aec3f464967bbe1b38ccfb357141134d09 Mon Sep 17 00:00:00 2001 From: antirez Date: Mon, 8 Nov 2010 11:52:03 +0100 Subject: [PATCH] non blocking loading of the DB / AOF with informations and ETA in INFO output --- src/aof.c | 10 ++++++++++ src/rdb.c | 37 +++++++++++++++++++++++++++++++++++++ src/redis.c | 42 ++++++++++++++++++++++++++++++++++++++++++ src/redis.h | 12 +++++++++++- 4 files changed, 100 insertions(+), 1 deletion(-) diff --git a/src/aof.c b/src/aof.c index 2396ba2c..959a5f52 100644 --- a/src/aof.c +++ b/src/aof.c @@ -218,6 +218,7 @@ int loadAppendOnlyFile(char *filename) { FILE *fp = fopen(filename,"r"); struct redis_stat sb; int appendonly = server.appendonly; + long loops = 0; if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) return REDIS_ERR; @@ -232,6 +233,8 @@ int loadAppendOnlyFile(char *filename) { server.appendonly = 0; fakeClient = createFakeClient(); + startLoading(fp); + while(1) { int argc, j; unsigned long len; @@ -241,6 +244,12 @@ int loadAppendOnlyFile(char *filename) { struct redisCommand *cmd; int force_swapout; + /* Serve the clients from time to time */ + if (!(loops++ % 1000)) { + loadingProgress(ftello(fp)); + aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); + } + if (fgets(buf,sizeof(buf),fp) == NULL) { if (feof(fp)) break; @@ -297,6 +306,7 @@ int loadAppendOnlyFile(char *filename) { fclose(fp); freeFakeClient(fakeClient); server.appendonly = appendonly; + stopLoading(); return REDIS_OK; readerr: diff --git a/src/rdb.c b/src/rdb.c index 589b536a..ce4b3566 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -7,6 +7,7 @@ #include #include #include +#include int rdbSaveType(FILE *fp, unsigned char type) { if (fwrite(&type,1,1,fp) == 0) return -1; @@ -793,6 +794,31 @@ robj *rdbLoadObject(int type, FILE *fp) { return o; } +/* Mark that we are loading in the global state and setup the fields + * needed to provide loading stats. */ +void startLoading(FILE *fp) { + struct stat sb; + + /* Load the DB */ + server.loading = 1; + server.loading_start_time = time(NULL); + if (fstat(fileno(fp), &sb) == -1) { + server.loading_total_bytes = 1; /* just to avoid division by zero */ + } else { + server.loading_total_bytes = sb.st_size; + } +} + +/* Refresh the loading progress info */ +void loadingProgress(off_t pos) { + server.loading_loaded_bytes = pos; +} + +/* Loading finished */ +void stopLoading(void) { + server.loading = 0; +} + int rdbLoad(char *filename) { FILE *fp; uint32_t dbid; @@ -801,6 +827,7 @@ int rdbLoad(char *filename) { redisDb *db = server.db+0; char buf[1024]; time_t expiretime, now = time(NULL); + long loops = 0; fp = fopen(filename,"r"); if (!fp) return REDIS_ERR; @@ -817,11 +844,20 @@ int rdbLoad(char *filename) { redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver); return REDIS_ERR; } + + startLoading(fp); while(1) { robj *key, *val; int force_swapout; expiretime = -1; + + /* Serve the clients from time to time */ + if (!(loops++ % 1000)) { + loadingProgress(ftello(fp)); + aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); + } + /* Read type. */ if ((type = rdbLoadType(fp)) == -1) goto eoferr; if (type == REDIS_EXPIRETIME) { @@ -900,6 +936,7 @@ int rdbLoad(char *filename) { } } fclose(fp); + stopLoading(); return REDIS_OK; eoferr: /* unexpected end of file is handled here with a fatal exit */ diff --git a/src/redis.c b/src/redis.c index 8519f46f..67bdc3ad 100644 --- a/src/redis.c +++ b/src/redis.c @@ -702,6 +702,8 @@ void createSharedObjects(void) { "-ERR source and destination objects are the same\r\n")); shared.outofrangeerr = createObject(REDIS_STRING,sdsnew( "-ERR index out of range\r\n")); + shared.loadingerr = createObject(REDIS_STRING,sdsnew( + "-LOADING Redis is loading the dataset in memory\r\n")); shared.space = createObject(REDIS_STRING,sdsnew(" ")); shared.colon = createObject(REDIS_STRING,sdsnew(":")); shared.plus = createObject(REDIS_STRING,sdsnew("+")); @@ -739,6 +741,7 @@ void initServerConfig() { server.verbosity = REDIS_VERBOSE; server.maxidletime = REDIS_MAXIDLETIME; server.saveparams = NULL; + server.loading = 0; server.logfile = NULL; /* NULL = log on standard output */ server.glueoutputbuf = 1; server.daemonize = 0; @@ -1006,6 +1009,12 @@ int processCommand(redisClient *c) { return REDIS_OK; } + /* Loading DB? Return an error if the command is not INFO */ + if (server.loading && cmd->proc != infoCommand) { + addReply(c, shared.loadingerr); + return REDIS_OK; + } + /* Exec the command */ if (c->flags & REDIS_MULTI && cmd->proc != execCommand && cmd->proc != discardCommand && @@ -1133,6 +1142,8 @@ sds genRedisInfoString(void) { "used_memory_rss:%zu\r\n" "mem_fragmentation_ratio:%.2f\r\n" "use_tcmalloc:%d\r\n" + "loading:%d\r\n" + "aof_enabled:%d\r\n" "changes_since_last_save:%lld\r\n" "bgsave_in_progress:%d\r\n" "last_save_time:%ld\r\n" @@ -1173,6 +1184,8 @@ sds genRedisInfoString(void) { #else 0, #endif + server.loading, + server.appendonly, server.dirty, server.bgsavechildpid != -1, server.lastsave, @@ -1243,6 +1256,35 @@ sds genRedisInfoString(void) { ); unlockThreadedIO(); } + if (server.loading) { + double perc; + time_t eta, elapsed; + off_t remaining_bytes = server.loading_total_bytes- + server.loading_loaded_bytes; + + perc = ((double)server.loading_loaded_bytes / + server.loading_total_bytes) * 100; + + elapsed = time(NULL)-server.loading_start_time; + if (elapsed == 0) { + eta = 1; /* A fake 1 second figure if we don't have enough info */ + } else { + eta = (elapsed*remaining_bytes)/server.loading_loaded_bytes; + } + + info = sdscatprintf(info, + "loading_start_time:%ld\r\n" + "loading_total_bytes:%llu\r\n" + "loading_loaded_bytes:%llu\r\n" + "loading_loaded_perc:%.2f\r\n" + "loading_eta_seconds:%ld\r\n" + ,(unsigned long) server.loading_start_time, + (unsigned long long) server.loading_total_bytes, + (unsigned long long) server.loading_loaded_bytes, + perc, + eta + ); + } for (j = 0; j < server.dbnum; j++) { long long keys, vkeys; diff --git a/src/redis.h b/src/redis.h index 043825af..30c88f40 100644 --- a/src/redis.h +++ b/src/redis.h @@ -340,7 +340,7 @@ struct sharedObjectsStruct { robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *cnegone, *pong, *space, *colon, *nullbulk, *nullmultibulk, *queued, *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr, - *outofrangeerr, *plus, + *outofrangeerr, *loadingerr, *plus, *select0, *select1, *select2, *select3, *select4, *select5, *select6, *select7, *select8, *select9, *messagebulk, *pmessagebulk, *subscribebulk, *unsubscribebulk, *mbulk3, @@ -361,6 +361,11 @@ struct redisServer { long long dirty_before_bgsave; /* used to restore dirty on failed BGSAVE */ list *clients; dict *commands; /* Command table hahs table */ + /* RDB / AOF loading information */ + int loading; + off_t loading_total_bytes; + off_t loading_loaded_bytes; + time_t loading_start_time; /* Fast pointers to often looked up command */ struct redisCommand *delCommand, *multiCommand; list *slaves, *monitors; @@ -726,6 +731,11 @@ int syncWithMaster(void); void updateSlavesWaitingBgsave(int bgsaveerr); void replicationCron(void); +/* Generic persistence functions */ +void startLoading(FILE *fp); +void loadingProgress(off_t pos); +void stopLoading(void); + /* RDB persistence */ int rdbLoad(char *filename); int rdbSaveBackground(char *filename); -- 2.45.2