diff options
| -rw-r--r-- | src/expire.c | 19 | ||||
| -rw-r--r-- | src/module.c | 70 | ||||
| -rw-r--r-- | src/object.c | 4 | ||||
| -rw-r--r-- | src/rdb.c | 40 | ||||
| -rw-r--r-- | src/redis-cli.c | 89 | ||||
| -rw-r--r-- | src/redismodule.h | 5 | ||||
| -rw-r--r-- | src/replication.c | 17 | ||||
| -rw-r--r-- | src/scripting.c | 8 | ||||
| -rw-r--r-- | src/server.c | 8 | ||||
| -rw-r--r-- | tests/unit/latency-monitor.tcl | 14 | ||||
| -rw-r--r-- | tests/unit/type/incr.tcl | 7 |
11 files changed, 231 insertions, 50 deletions
diff --git a/src/expire.c b/src/expire.c index a02fe566a..81c9e23f5 100644 --- a/src/expire.c +++ b/src/expire.c @@ -103,7 +103,7 @@ void activeExpireCycle(int type) { int j, iteration = 0; int dbs_per_call = CRON_DBS_PER_CALL; - long long start = ustime(), timelimit; + long long start = ustime(), timelimit, elapsed; /* When clients are paused the dataset should be static not just from the * POV of clients not being able to write, but also from the POV of @@ -140,7 +140,7 @@ void activeExpireCycle(int type) { if (type == ACTIVE_EXPIRE_CYCLE_FAST) timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */ - for (j = 0; j < dbs_per_call; j++) { + for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) { int expired; redisDb *db = server.db+(current_db % server.dbnum); @@ -155,6 +155,7 @@ void activeExpireCycle(int type) { unsigned long num, slots; long long now, ttl_sum; int ttl_samples; + iteration++; /* If there is nothing to expire try next DB ASAP. */ if ((num = dictSize(db->expires)) == 0) { @@ -207,18 +208,20 @@ void activeExpireCycle(int type) { /* We can't block forever here even if there are many keys to * expire. So after a given amount of milliseconds return to the * caller waiting for the other active expire cycle. */ - iteration++; if ((iteration & 0xf) == 0) { /* check once every 16 iterations. */ - long long elapsed = ustime()-start; - - latencyAddSampleIfNeeded("expire-cycle",elapsed/1000); - if (elapsed > timelimit) timelimit_exit = 1; + elapsed = ustime()-start; + if (elapsed > timelimit) { + timelimit_exit = 1; + break; + } } - if (timelimit_exit) return; /* We don't repeat the cycle if there are less than 25% of keys * found expired in the current DB. */ } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4); } + + elapsed = ustime()-start; + latencyAddSampleIfNeeded("expire-cycle",elapsed/1000); } /*----------------------------------------------------------------------------- diff --git a/src/module.c b/src/module.c index afeb6e2cf..8a4c40f12 100644 --- a/src/module.c +++ b/src/module.c @@ -442,9 +442,7 @@ void moduleFreeContext(RedisModuleCtx *ctx) { void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) { client *c = ctx->client; - /* We don't want any automatic propagation here since in modules we handle - * replication / AOF propagation in explicit ways. */ - preventCommandPropagation(c); + if (c->flags & CLIENT_LUA) return; /* Handle the replication of the final EXEC, since whatever a command * emits is always wrappered around MULTI/EXEC. */ @@ -615,7 +613,7 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c sds cmdname = sdsnew(name); /* Check if the command name is busy. */ - if (lookupCommand((char*)name) != NULL) { + if (lookupCommand(cmdname) != NULL) { sdsfree(cmdname); return REDISMODULE_ERR; } @@ -650,7 +648,7 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c * * This is an internal function, Redis modules developers don't need * to use it. */ -void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int apiver){ +void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int apiver) { RedisModule *module; if (ctx->module != NULL) return; @@ -662,6 +660,15 @@ void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int api ctx->module = module; } +/* Return non-zero if the module name is busy. + * Otherwise zero is returned. */ +int RM_IsModuleNameBusy(const char *name) { + sds modulename = sdsnew(name); + dictEntry *de = dictFind(modules,modulename); + sdsfree(modulename); + return de != NULL; +} + /* Return the current UNIX time in milliseconds. */ long long RM_Milliseconds(void) { return mstime(); @@ -1164,6 +1171,9 @@ int RM_ReplyWithDouble(RedisModuleCtx *ctx, double d) { * in the context of a command execution. EXEC will be handled by the * RedisModuleCommandDispatcher() function. */ void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx) { + /* Skip this if client explicitly wrap the command with MULTI, or if + * the module command was called by a script. */ + if (ctx->client->flags & (CLIENT_MULTI|CLIENT_LUA)) return; /* If we already emitted MULTI return ASAP. */ if (ctx->flags & REDISMODULE_CTX_MULTI_EMITTED) return; /* If this is a thread safe context, we do not want to wrap commands @@ -1216,6 +1226,7 @@ int RM_Replicate(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...) /* Release the argv. */ for (j = 0; j < argc; j++) decrRefCount(argv[j]); zfree(argv); + server.dirty++; return REDISMODULE_OK; } @@ -1234,6 +1245,7 @@ int RM_ReplicateVerbatim(RedisModuleCtx *ctx) { alsoPropagate(ctx->client->cmd,ctx->client->db->id, ctx->client->argv,ctx->client->argc, PROPAGATE_AOF|PROPAGATE_REPL); + server.dirty++; return REDISMODULE_OK; } @@ -3733,6 +3745,28 @@ void moduleFreeModuleStructure(struct RedisModule *module) { zfree(module); } +void moduleUnregisterCommands(struct RedisModule *module) { + /* Unregister all the commands registered by this module. */ + dictIterator *di = dictGetSafeIterator(server.commands); + dictEntry *de; + while ((de = dictNext(di)) != NULL) { + struct redisCommand *cmd = dictGetVal(de); + if (cmd->proc == RedisModuleCommandDispatcher) { + RedisModuleCommandProxy *cp = + (void*)(unsigned long)cmd->getkeys_proc; + sds cmdname = cp->rediscmd->name; + if (cp->module == module) { + dictDelete(server.commands,cmdname); + dictDelete(server.orig_commands,cmdname); + sdsfree(cmdname); + zfree(cp->rediscmd); + zfree(cp); + } + } + } + dictReleaseIterator(di); +} + /* Load a module and initialize it. On success C_OK is returned, otherwise * C_ERR is returned. */ int moduleLoad(const char *path, void **module_argv, int module_argc) { @@ -3753,7 +3787,10 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) { return C_ERR; } if (onload((void*)&ctx,module_argv,module_argc) == REDISMODULE_ERR) { - if (ctx.module) moduleFreeModuleStructure(ctx.module); + if (ctx.module) { + moduleUnregisterCommands(ctx.module); + moduleFreeModuleStructure(ctx.module); + } dlclose(handle); serverLog(LL_WARNING, "Module %s initialization failed. Module not loaded",path); @@ -3787,25 +3824,7 @@ int moduleUnload(sds name) { return REDISMODULE_ERR; } - /* Unregister all the commands registered by this module. */ - dictIterator *di = dictGetSafeIterator(server.commands); - dictEntry *de; - while ((de = dictNext(di)) != NULL) { - struct redisCommand *cmd = dictGetVal(de); - if (cmd->proc == RedisModuleCommandDispatcher) { - RedisModuleCommandProxy *cp = - (void*)(unsigned long)cmd->getkeys_proc; - sds cmdname = cp->rediscmd->name; - if (cp->module == module) { - dictDelete(server.commands,cmdname); - dictDelete(server.orig_commands,cmdname); - sdsfree(cmdname); - zfree(cp->rediscmd); - zfree(cp); - } - } - } - dictReleaseIterator(di); + moduleUnregisterCommands(module); /* Unregister all the hooks. TODO: Yet no hooks support here. */ @@ -3900,6 +3919,7 @@ void moduleRegisterCoreAPI(void) { REGISTER_API(Strdup); REGISTER_API(CreateCommand); REGISTER_API(SetModuleAttribs); + REGISTER_API(IsModuleNameBusy); REGISTER_API(WrongArity); REGISTER_API(ReplyWithLongLong); REGISTER_API(ReplyWithError); diff --git a/src/object.c b/src/object.c index 4028380d4..e62b5b3d9 100644 --- a/src/object.c +++ b/src/object.c @@ -560,7 +560,7 @@ int getDoubleFromObject(const robj *o, double *target) { value = strtod(o->ptr, &eptr); if (sdslen(o->ptr) == 0 || isspace(((const char*)o->ptr)[0]) || - eptr[0] != '\0' || + (size_t)(eptr-(char*)o->ptr) != sdslen(o->ptr) || (errno == ERANGE && (value == HUGE_VAL || value == -HUGE_VAL || value == 0)) || isnan(value)) @@ -602,7 +602,7 @@ int getLongDoubleFromObject(robj *o, long double *target) { value = strtold(o->ptr, &eptr); if (sdslen(o->ptr) == 0 || isspace(((const char*)o->ptr)[0]) || - eptr[0] != '\0' || + (size_t)(eptr-(char*)o->ptr) != sdslen(o->ptr) || (errno == ERANGE && (value == HUGE_VAL || value == -HUGE_VAL || value == 0)) || isnan(value)) @@ -2000,6 +2000,9 @@ void bgsaveCommand(client *c) { } } + rdbSaveInfo rsi, *rsiptr; + rsiptr = rdbPopulateSaveInfo(&rsi); + if (server.rdb_child_pid != -1) { addReplyError(c,"Background save already in progress"); } else if (server.aof_child_pid != -1) { @@ -2012,7 +2015,7 @@ void bgsaveCommand(client *c) { "Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenever " "possible."); } - } else if (rdbSaveBackground(server.rdb_filename,NULL) == C_OK) { + } else if (rdbSaveBackground(server.rdb_filename,rsiptr) == C_OK) { addReplyStatus(c,"Background saving started"); } else { addReply(c,shared.err); @@ -2033,22 +2036,37 @@ rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi) { *rsi = rsi_init; /* If the instance is a master, we can populate the replication info - * in all the cases, even if sometimes in incomplete (but safe) form. */ - if (!server.masterhost) { - if (server.repl_backlog) rsi->repl_stream_db = server.slaveseldb; - /* Note that if repl_backlog is NULL, it means that histories - * following from this point will trigger a full synchronization - * generating a SELECT statement, so we can leave the currently - * selected DB set to -1. This allows a restarted master to reload - * its replication ID/offset when there are no connected slaves. */ + * only when repl_backlog is not NULL. If the repl_backlog is NULL, + * it means that the instance isn't in any replication chains. In this + * scenario the replication info is useless, because when a slave + * connects to us, the NULL repl_backlog will trigger a full + * synchronization, at the same time we will use a new replid and clear + * replid2. */ + if (!server.masterhost && server.repl_backlog) { + /* Note that when server.slaveseldb is -1, it means that this master + * didn't apply any write commands after a full synchronization. + * So we can let repl_stream_db be 0, this allows a restarted slave + * to reload replication ID/offset, it's safe because the next write + * command must generate a SELECT statement. */ + rsi->repl_stream_db = server.slaveseldb == -1 ? 0 : server.slaveseldb; return rsi; } - /* If the instance is a slave we need a connected master in order to - * fetch the currently selected DB. */ + /* If the instance is a slave we need a connected master + * in order to fetch the currently selected DB. */ if (server.master) { rsi->repl_stream_db = server.master->db->id; return rsi; } + + /* If we have a cached master we can use it in order to populate the + * replication selected DB info inside the RDB file: the slave can + * increment the master_repl_offset only from data arriving from the + * master, so if we are disconnected the offset in the cached master + * is valid. */ + if (server.cached_master) { + rsi->repl_stream_db = server.cached_master->db->id; + return rsi; + } return NULL; } diff --git a/src/redis-cli.c b/src/redis-cli.c index ca9fe6ad7..84eabf391 100644 --- a/src/redis-cli.c +++ b/src/redis-cli.c @@ -198,6 +198,92 @@ static sds getDotfilePath(char *envoverride, char *dotfilename) { return dotPath; } +/* URL-style percent decoding. */ +#define isHexChar(c) (isdigit(c) || (c >= 'a' && c <= 'f')) +#define decodeHexChar(c) (isdigit(c) ? c - '0' : c - 'a' + 10) +#define decodeHex(h, l) ((decodeHexChar(h) << 4) + decodeHexChar(l)) + +static sds percentDecode(const char *pe, size_t len) { + const char *end = pe + len; + sds ret = sdsempty(); + const char *curr = pe; + + while (curr < end) { + if (*curr == '%') { + if ((end - curr) < 2) { + fprintf(stderr, "Incomplete URI encoding\n"); + exit(1); + } + + char h = tolower(*(++curr)); + char l = tolower(*(++curr)); + if (!isHexChar(h) || !isHexChar(l)) { + fprintf(stderr, "Illegal character in URI encoding\n"); + exit(1); + } + char c = decodeHex(h, l); + ret = sdscatlen(ret, &c, 1); + curr++; + } else { + ret = sdscatlen(ret, curr++, 1); + } + } + + return ret; +} + +/* Parse a URI and extract the server connection information. + * URI scheme is based on the the provisional specification[1] excluding support + * for query parameters. Valid URIs are: + * scheme: "redis://" + * authority: [<username> ":"] <password> "@"] [<hostname> [":" <port>]] + * path: ["/" [<db>]] + * + * [1]: https://www.iana.org/assignments/uri-schemes/prov/redis */ +static void parseRedisUri(const char *uri) { + + const char *scheme = "redis://"; + const char *curr = uri; + const char *end = uri + strlen(uri); + const char *userinfo, *username, *port, *host, *path; + + /* URI must start with a valid scheme. */ + if (strncasecmp(scheme, curr, strlen(scheme))) { + fprintf(stderr,"Invalid URI scheme\n"); + exit(1); + } + curr += strlen(scheme); + if (curr == end) return; + + /* Extract user info. */ + if ((userinfo = strchr(curr,'@'))) { + if ((username = strchr(curr, ':')) && username < userinfo) { + /* If provided, username is ignored. */ + curr = username + 1; + } + + config.auth = percentDecode(curr, userinfo - curr); + curr = userinfo + 1; + } + if (curr == end) return; + + /* Extract host and port. */ + path = strchr(curr, '/'); + if (*curr != '/') { + host = path ? path - 1 : end; + if ((port = strchr(curr, ':'))) { + config.hostport = atoi(port + 1); + host = port - 1; + } + config.hostip = sdsnewlen(curr, host - curr + 1); + } + curr = path ? path + 1 : end; + if (curr == end) return; + + /* Extract database number. */ + config.dbnum = atoi(curr); +} + /*------------------------------------------------------------------------------ * Help functions *--------------------------------------------------------------------------- */ @@ -1002,6 +1088,8 @@ static int parseOptions(int argc, char **argv) { config.dbnum = atoi(argv[++i]); } else if (!strcmp(argv[i],"-a") && !lastarg) { config.auth = argv[++i]; + } else if (!strcmp(argv[i],"-u") && !lastarg) { + parseRedisUri(argv[++i]); } else if (!strcmp(argv[i],"--raw")) { config.output = OUTPUT_RAW; } else if (!strcmp(argv[i],"--no-raw")) { @@ -1109,6 +1197,7 @@ static void usage(void) { " -p <port> Server port (default: 6379).\n" " -s <socket> Server socket (overrides hostname and port).\n" " -a <password> Password to use when connecting to the server.\n" +" -u <uri> Server URI.\n" " -r <repeat> Execute specified command N times.\n" " -i <interval> When -r is used, waits <interval> seconds per command.\n" " It is possible to specify sub-second times like -i 0.1.\n" diff --git a/src/redismodule.h b/src/redismodule.h index 8df203aba..672951f78 100644 --- a/src/redismodule.h +++ b/src/redismodule.h @@ -143,7 +143,8 @@ void *REDISMODULE_API_FUNC(RedisModule_Calloc)(size_t nmemb, size_t size); char *REDISMODULE_API_FUNC(RedisModule_Strdup)(const char *str); int REDISMODULE_API_FUNC(RedisModule_GetApi)(const char *, void *); int REDISMODULE_API_FUNC(RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep); -int REDISMODULE_API_FUNC(RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver); +void REDISMODULE_API_FUNC(RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver); +int REDISMODULE_API_FUNC(RedisModule_IsModuleNameBusy)(const char *name); int REDISMODULE_API_FUNC(RedisModule_WrongArity)(RedisModuleCtx *ctx); int REDISMODULE_API_FUNC(RedisModule_ReplyWithLongLong)(RedisModuleCtx *ctx, long long ll); int REDISMODULE_API_FUNC(RedisModule_GetSelectedDb)(RedisModuleCtx *ctx); @@ -263,6 +264,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int REDISMODULE_GET_API(Strdup); REDISMODULE_GET_API(CreateCommand); REDISMODULE_GET_API(SetModuleAttribs); + REDISMODULE_GET_API(IsModuleNameBusy); REDISMODULE_GET_API(WrongArity); REDISMODULE_GET_API(ReplyWithLongLong); REDISMODULE_GET_API(ReplyWithError); @@ -370,6 +372,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int REDISMODULE_GET_API(AbortBlock); #endif + if (RedisModule_IsModuleNameBusy(name)) return REDISMODULE_ERR; RedisModule_SetModuleAttribs(ctx,name,ver,apiver); return REDISMODULE_OK; } diff --git a/src/replication.c b/src/replication.c index e0b3d910e..cf4db3e3a 100644 --- a/src/replication.c +++ b/src/replication.c @@ -2613,6 +2613,23 @@ void replicationCron(void) { time_t idle = server.unixtime - server.repl_no_slaves_since; if (idle > server.repl_backlog_time_limit) { + /* When we free the backlog, we always use a new + * replication ID and clear the ID2. This is needed + * because when there is no backlog, the master_repl_offset + * is not updated, but we would still retain our replication + * ID, leading to the following problem: + * + * 1. We are a master instance. + * 2. Our slave is promoted to master. It's repl-id-2 will + * be the same as our repl-id. + * 3. We, yet as master, receive some updates, that will not + * increment the master_repl_offset. + * 4. Later we are turned into a slave, connecto to the new + * master that will accept our PSYNC request by second + * replication ID, but there will be data inconsistency + * because we received writes. */ + changeReplicationId(); + clearReplicationId2(); freeReplicationBacklog(); serverLog(LL_NOTICE, "Replication backlog freed after %d seconds " diff --git a/src/scripting.c b/src/scripting.c index 8f8145b2c..d9f954068 100644 --- a/src/scripting.c +++ b/src/scripting.c @@ -358,6 +358,13 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { static size_t cached_objects_len[LUA_CMD_OBJCACHE_SIZE]; static int inuse = 0; /* Recursive calls detection. */ + /* Reflect MULTI state */ + if (server.lua_multi_emitted || (server.lua_caller->flags & CLIENT_MULTI)) { + c->flags |= CLIENT_MULTI; + } else { + c->flags &= ~CLIENT_MULTI; + } + /* By using Lua debug hooks it is possible to trigger a recursive call * to luaRedisGenericCommand(), which normally should never happen. * To make this function reentrant is futile and makes it slower, but @@ -535,6 +542,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { * a Lua script in the context of AOF and slaves. */ if (server.lua_replicate_commands && !server.lua_multi_emitted && + !(server.lua_caller->flags & CLIENT_MULTI) && server.lua_write_dirty && server.lua_repl != PROPAGATE_NONE) { diff --git a/src/server.c b/src/server.c index c9972115a..888926267 100644 --- a/src/server.c +++ b/src/server.c @@ -2265,8 +2265,9 @@ void call(client *c, int flags) { propagate_flags &= ~PROPAGATE_AOF; /* Call propagate() only if at least one of AOF / replication - * propagation is needed. */ - if (propagate_flags != PROPAGATE_NONE) + * propagation is needed. Note that modules commands handle replication + * in an explicit way, so we never replicate them automatically. */ + if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE)) propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags); } @@ -3539,7 +3540,8 @@ void loadDataFromDisk(void) { rsi.repl_id_is_set && rsi.repl_offset != -1 && /* Note that older implementations may save a repl_stream_db - * of -1 inside the RDB file. */ + * of -1 inside the RDB file in a wrong way, see more information + * in function rdbPopulateSaveInfo. */ rsi.repl_stream_db != -1) { memcpy(server.replid,rsi.repl_id,sizeof(server.replid)); diff --git a/tests/unit/latency-monitor.tcl b/tests/unit/latency-monitor.tcl index b736cad98..69da13f06 100644 --- a/tests/unit/latency-monitor.tcl +++ b/tests/unit/latency-monitor.tcl @@ -47,4 +47,18 @@ start_server {tags {"latency-monitor"}} { assert {[r latency reset] > 0} assert {[r latency latest] eq {}} } + + test {LATENCY of expire events are correctly collected} { + r config set latency-monitor-threshold 20 + r eval { + local i = 0 + while (i < 1000000) do + redis.call('sadd','mybigkey',i) + i = i+1 + end + } 0 + r pexpire mybigkey 1 + after 500 + assert_match {*expire-cycle*} [r latency latest] + } } diff --git a/tests/unit/type/incr.tcl b/tests/unit/type/incr.tcl index 2287aaae2..a58710d39 100644 --- a/tests/unit/type/incr.tcl +++ b/tests/unit/type/incr.tcl @@ -144,4 +144,11 @@ start_server {tags {"incr"}} { r set foo 1 roundFloat [r incrbyfloat foo -1.1] } {-0.1} + + test {string to double with null terminator} { + r set foo 1 + r setrange foo 2 2 + catch {r incrbyfloat foo 1} err + format $err + } {ERR*valid*} } |