Unverified Commit 53d1acd5 authored by ny0312's avatar ny0312 Committed by GitHub
Browse files

Always replicate time-to-live(TTL) as absolute timestamps in milliseconds (#8474)

Till now, on replica full-sync we used to transfer absolute time for TTL,
however when a command arrived (EXPIRE or EXPIREAT),
we used to propagate it as is to replicas (possibly with relative time),
but always translate it to EXPIREAT (absolute time) to AOF.

This commit changes that and will always use absolute time for propagation.
see discussion in #8433

Furthermore, we Introduce new commands: `EXPIRETIME/PEXPIRETIME`
that allow extracting the absolute TTL time from a key.
parent 501d7755
...@@ -575,43 +575,7 @@ sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) { ...@@ -575,43 +575,7 @@ sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
return dst; return dst;
} }
/* Create the sds representation of a PEXPIREAT command, using void feedAppendOnlyFile(int dictid, robj **argv, int argc) {
* 'seconds' as time to live and 'cmd' to understand what command
* we are translating into a PEXPIREAT.
*
* This command is used in order to translate EXPIRE and PEXPIRE commands
* into PEXPIREAT command so that we retain precision in the append only
* file, and the time is always absolute and not relative. */
sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) {
long long when;
robj *argv[3];
/* Make sure we can use strtoll */
seconds = getDecodedObject(seconds);
when = strtoll(seconds->ptr,NULL,10);
/* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */
if (cmd->proc == expireCommand || cmd->proc == setexCommand ||
cmd->proc == expireatCommand)
{
when *= 1000;
}
/* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */
if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
cmd->proc == setexCommand || cmd->proc == psetexCommand)
{
when += mstime();
}
decrRefCount(seconds);
argv[0] = shared.pexpireat;
argv[1] = key;
argv[2] = createStringObjectFromLongLong(when);
buf = catAppendOnlyGenericCommand(buf, 3, argv);
decrRefCount(argv[2]);
return buf;
}
void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
sds buf = sdsempty(); sds buf = sdsempty();
/* The DB this command was targeting is not the same as the last command /* The DB this command was targeting is not the same as the last command
* we appended. To issue a SELECT command is needed. */ * we appended. To issue a SELECT command is needed. */
...@@ -624,44 +588,9 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a ...@@ -624,44 +588,9 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
server.aof_selected_db = dictid; server.aof_selected_db = dictid;
} }
if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || /* All commands should be propagated the same way in AOF as in replication.
cmd->proc == expireatCommand) { * No need for AOF-specific translation. */
/* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */ buf = catAppendOnlyGenericCommand(buf,argc,argv);
buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
} else if (cmd->proc == setCommand && argc > 3) {
robj *pxarg = NULL;
/* When SET is used with EX/PX argument setGenericCommand propagates them with PX millisecond argument.
* So since the command arguments are re-written there, we can rely here on the index of PX being 3. */
if (!strcasecmp(argv[3]->ptr, "px")) {
pxarg = argv[4];
}
/* For AOF we convert SET key value relative time in milliseconds to SET key value absolute time in
* millisecond. Whenever the condition is true it implies that original SET has been transformed
* to SET PX with millisecond time argument so we do not need to worry about unit here.*/
if (pxarg) {
robj *millisecond = getDecodedObject(pxarg);
long long when = strtoll(millisecond->ptr,NULL,10);
when += mstime();
decrRefCount(millisecond);
robj *newargs[5];
newargs[0] = argv[0];
newargs[1] = argv[1];
newargs[2] = argv[2];
newargs[3] = shared.pxat;
newargs[4] = createStringObjectFromLongLong(when);
buf = catAppendOnlyGenericCommand(buf,5,newargs);
decrRefCount(newargs[4]);
} else {
buf = catAppendOnlyGenericCommand(buf,argc,argv);
}
} else {
/* All the other commands don't need translation or need the
* same translation already operated in the command vector
* for the replication itself. */
buf = catAppendOnlyGenericCommand(buf,argc,argv);
}
/* Append to the AOF buffer. This will be flushed on disk just before /* Append to the AOF buffer. This will be flushed on disk just before
* of re-entering the event loop, so before the client will get a * of re-entering the event loop, so before the client will get a
......
...@@ -5206,6 +5206,13 @@ void restoreCommand(client *c) { ...@@ -5206,6 +5206,13 @@ void restoreCommand(client *c) {
dbAdd(c->db,key,obj); dbAdd(c->db,key,obj);
if (ttl) { if (ttl) {
setExpire(c,c->db,key,ttl); setExpire(c,c->db,key,ttl);
if (!absttl) {
/* Propagate TTL as absolute timestamp */
robj *ttl_obj = createStringObjectFromLongLong(ttl);
rewriteClientCommandArgument(c,2,ttl_obj);
decrRefCount(ttl_obj);
rewriteClientCommandArgument(c,c->argc,shared.absttl);
}
} }
objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock,1000); objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock,1000);
signalModifiedKey(c,c->db,key); signalModifiedKey(c,c->db,key);
......
...@@ -539,6 +539,10 @@ void expireGenericCommand(client *c, long long basetime, int unit) { ...@@ -539,6 +539,10 @@ void expireGenericCommand(client *c, long long basetime, int unit) {
} else { } else {
setExpire(c,c->db,key,when); setExpire(c,c->db,key,when);
addReply(c,shared.cone); addReply(c,shared.cone);
/* Propagate as PEXPIREAT millisecond-timestamp */
robj *when_obj = createStringObjectFromLongLong(when);
rewriteClientCommandVector(c, 3, shared.pexpireat, key, when_obj);
decrRefCount(when_obj);
signalModifiedKey(c,c->db,key); signalModifiedKey(c,c->db,key);
notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
server.dirty++; server.dirty++;
...@@ -566,8 +570,8 @@ void pexpireatCommand(client *c) { ...@@ -566,8 +570,8 @@ void pexpireatCommand(client *c) {
expireGenericCommand(c,0,UNIT_MILLISECONDS); expireGenericCommand(c,0,UNIT_MILLISECONDS);
} }
/* Implements TTL and PTTL */ /* Implements TTL, PTTL and EXPIRETIME */
void ttlGenericCommand(client *c, int output_ms) { void ttlGenericCommand(client *c, int output_ms, int output_abs) {
long long expire, ttl = -1; long long expire, ttl = -1;
/* If the key does not exist at all, return -2 */ /* If the key does not exist at all, return -2 */
...@@ -575,11 +579,12 @@ void ttlGenericCommand(client *c, int output_ms) { ...@@ -575,11 +579,12 @@ void ttlGenericCommand(client *c, int output_ms) {
addReplyLongLong(c,-2); addReplyLongLong(c,-2);
return; return;
} }
/* The key exists. Return -1 if it has no expire, or the actual /* The key exists. Return -1 if it has no expire, or the actual
* TTL value otherwise. */ * TTL value otherwise. */
expire = getExpire(c->db,c->argv[1]); expire = getExpire(c->db,c->argv[1]);
if (expire != -1) { if (expire != -1) {
ttl = expire-mstime(); ttl = output_abs ? expire : expire-mstime();
if (ttl < 0) ttl = 0; if (ttl < 0) ttl = 0;
} }
if (ttl == -1) { if (ttl == -1) {
...@@ -591,12 +596,22 @@ void ttlGenericCommand(client *c, int output_ms) { ...@@ -591,12 +596,22 @@ void ttlGenericCommand(client *c, int output_ms) {
/* TTL key */ /* TTL key */
void ttlCommand(client *c) { void ttlCommand(client *c) {
ttlGenericCommand(c, 0); ttlGenericCommand(c, 0, 0);
} }
/* PTTL key */ /* PTTL key */
void pttlCommand(client *c) { void pttlCommand(client *c) {
ttlGenericCommand(c, 1); ttlGenericCommand(c, 1, 0);
}
/* EXPIRETIME key */
void expiretimeCommand(client *c) {
ttlGenericCommand(c, 0, 1);
}
/* PEXPIRETIME key */
void pexpiretimeCommand(client *c) {
ttlGenericCommand(c, 1, 1);
} }
/* PERSIST key */ /* PERSIST key */
......
...@@ -800,6 +800,14 @@ struct redisCommand redisCommandTable[] = { ...@@ -800,6 +800,14 @@ struct redisCommand redisCommandTable[] = {
"read-only fast random @keyspace", "read-only fast random @keyspace",
0,NULL,1,1,1,0,0,0}, 0,NULL,1,1,1,0,0,0},
{"expiretime",expiretimeCommand,2,
"read-only fast random @keyspace",
0,NULL,1,1,1,0,0,0},
{"pexpiretime",pexpiretimeCommand,2,
"read-only fast random @keyspace",
0,NULL,1,1,1,0,0,0},
{"persist",persistCommand,2, {"persist",persistCommand,2,
"write fast @keyspace", "write fast @keyspace",
0,NULL,1,1,1,0,0,0}, 0,NULL,1,1,1,0,0,0},
...@@ -2601,7 +2609,6 @@ void createSharedObjects(void) { ...@@ -2601,7 +2609,6 @@ void createSharedObjects(void) {
shared.left = createStringObject("left",4); shared.left = createStringObject("left",4);
shared.right = createStringObject("right",5); shared.right = createStringObject("right",5);
shared.pxat = createStringObject("PXAT", 4); shared.pxat = createStringObject("PXAT", 4);
shared.px = createStringObject("PX",2);
shared.time = createStringObject("TIME",4); shared.time = createStringObject("TIME",4);
shared.retrycount = createStringObject("RETRYCOUNT",10); shared.retrycount = createStringObject("RETRYCOUNT",10);
shared.force = createStringObject("FORCE",5); shared.force = createStringObject("FORCE",5);
...@@ -2611,6 +2618,7 @@ void createSharedObjects(void) { ...@@ -2611,6 +2618,7 @@ void createSharedObjects(void) {
shared.ping = createStringObject("ping",4); shared.ping = createStringObject("ping",4);
shared.setid = createStringObject("SETID",5); shared.setid = createStringObject("SETID",5);
shared.keepttl = createStringObject("KEEPTTL",7); shared.keepttl = createStringObject("KEEPTTL",7);
shared.absttl = createStringObject("ABSTTL",6);
shared.load = createStringObject("LOAD",4); shared.load = createStringObject("LOAD",4);
shared.createconsumer = createStringObject("CREATECONSUMER",14); shared.createconsumer = createStringObject("CREATECONSUMER",14);
shared.getack = createStringObject("GETACK",6); shared.getack = createStringObject("GETACK",6);
...@@ -3574,6 +3582,8 @@ struct redisCommand *lookupCommandOrOriginal(sds name) { ...@@ -3574,6 +3582,8 @@ struct redisCommand *lookupCommandOrOriginal(sds name) {
void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
int flags) int flags)
{ {
UNUSED(cmd);
if (!server.replication_allowed) if (!server.replication_allowed)
return; return;
...@@ -3590,7 +3600,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, ...@@ -3590,7 +3600,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
serverAssert(!(areClientsPaused() && !server.client_pause_in_transaction)); serverAssert(!(areClientsPaused() && !server.client_pause_in_transaction));
if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF) if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF)
feedAppendOnlyFile(cmd,dbid,argv,argc); feedAppendOnlyFile(dbid,argv,argc);
if (flags & PROPAGATE_REPL) if (flags & PROPAGATE_REPL)
replicationFeedSlaves(server.slaves,dbid,argv,argc); replicationFeedSlaves(server.slaves,dbid,argv,argc);
} }
......
...@@ -983,7 +983,7 @@ struct sharedObjectsStruct { ...@@ -983,7 +983,7 @@ struct sharedObjectsStruct {
*rpop, *lpop, *lpush, *rpoplpush, *lmove, *blmove, *zpopmin, *zpopmax, *rpop, *lpop, *lpush, *rpoplpush, *lmove, *blmove, *zpopmin, *zpopmax,
*emptyscan, *multi, *exec, *left, *right, *hset, *srem, *xgroup, *xclaim, *emptyscan, *multi, *exec, *left, *right, *hset, *srem, *xgroup, *xclaim,
*script, *replconf, *eval, *persist, *set, *pexpireat, *pexpire, *script, *replconf, *eval, *persist, *set, *pexpireat, *pexpire,
*time, *pxat, *px, *retrycount, *force, *justid, *time, *pxat, *absttl, *retrycount, *force, *justid,
*lastid, *ping, *setid, *keepttl, *load, *createconsumer, *lastid, *ping, *setid, *keepttl, *load, *createconsumer,
*getack, *special_asterick, *special_equals, *default_username, *redacted, *getack, *special_asterick, *special_equals, *default_username, *redacted,
*select[PROTO_SHARED_SELECT_CMDS], *select[PROTO_SHARED_SELECT_CMDS],
...@@ -2068,7 +2068,7 @@ int bg_unlink(const char *filename); ...@@ -2068,7 +2068,7 @@ int bg_unlink(const char *filename);
/* AOF persistence */ /* AOF persistence */
void flushAppendOnlyFile(int force); void flushAppendOnlyFile(int force);
void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc); void feedAppendOnlyFile(int dictid, robj **argv, int argc);
void aofRemoveTempFile(pid_t childpid); void aofRemoveTempFile(pid_t childpid);
int rewriteAppendOnlyFileBackground(void); int rewriteAppendOnlyFileBackground(void);
int loadAppendOnlyFile(char *filename); int loadAppendOnlyFile(char *filename);
...@@ -2561,6 +2561,8 @@ void getsetCommand(client *c); ...@@ -2561,6 +2561,8 @@ void getsetCommand(client *c);
void ttlCommand(client *c); void ttlCommand(client *c);
void touchCommand(client *c); void touchCommand(client *c);
void pttlCommand(client *c); void pttlCommand(client *c);
void expiretimeCommand(client *c);
void pexpiretimeCommand(client *c);
void persistCommand(client *c); void persistCommand(client *c);
void replicaofCommand(client *c); void replicaofCommand(client *c);
void roleCommand(client *c); void roleCommand(client *c);
......
...@@ -72,26 +72,13 @@ static int checkStringLength(client *c, long long size) { ...@@ -72,26 +72,13 @@ static int checkStringLength(client *c, long long size) {
#define OBJ_PXAT (1<<7) /* Set if timestamp in ms is given */ #define OBJ_PXAT (1<<7) /* Set if timestamp in ms is given */
#define OBJ_PERSIST (1<<8) /* Set if we need to remove the ttl */ #define OBJ_PERSIST (1<<8) /* Set if we need to remove the ttl */
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) { /* Forward declaration */
long long milliseconds = 0, when = 0; /* initialized to avoid any harmness warning */ static int getExpireMillisecondsOrReply(client *c, robj *expire, int flags, int unit, long long *milliseconds);
if (expire) { void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK) long long milliseconds = 0; /* initialized to avoid any harmness warning */
return; if (expire && getExpireMillisecondsOrReply(c, expire, flags, unit, &milliseconds) != C_OK) {
if (milliseconds <= 0 || (unit == UNIT_SECONDS && milliseconds > LLONG_MAX / 1000)) { return;
/* Negative value provided or multiplication is gonna overflow. */
addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
return;
}
if (unit == UNIT_SECONDS) milliseconds *= 1000;
when = milliseconds;
if ((flags & OBJ_PX) || (flags & OBJ_EX))
when += mstime();
if (when <= 0) {
/* Overflow detected. */
addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
return;
}
} }
if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) || if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
...@@ -108,24 +95,17 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, ...@@ -108,24 +95,17 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
genericSetKey(c,c->db,key, val,flags & OBJ_KEEPTTL,1); genericSetKey(c,c->db,key, val,flags & OBJ_KEEPTTL,1);
server.dirty++; server.dirty++;
notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
if (expire) { if (expire) {
setExpire(c,c->db,key,when); setExpire(c,c->db,key,milliseconds);
/* Propagate as SET Key Value PXAT millisecond-timestamp if there is
* EX/PX/EXAT/PXAT flag. */
robj *milliseconds_obj = createStringObjectFromLongLong(milliseconds);
rewriteClientCommandVector(c, 5, shared.set, key, val, shared.pxat, milliseconds_obj);
decrRefCount(milliseconds_obj);
notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
/* Propagate as SET Key Value PXAT millisecond-timestamp if there is EXAT/PXAT or
* propagate as SET Key Value PX millisecond if there is EX/PX flag.
*
* Additionally when we propagate the SET with PX (relative millisecond) we translate
* it again to SET with PXAT for the AOF.
*
* Additional care is required while modifying the argument order. AOF relies on the
* exp argument being at index 3. (see feedAppendOnlyFile)
* */
robj *exp = (flags & OBJ_PXAT) || (flags & OBJ_EXAT) ? shared.pxat : shared.px;
robj *millisecondObj = createStringObjectFromLongLong(milliseconds);
rewriteClientCommandVector(c,5,shared.set,key,val,exp,millisecondObj);
decrRefCount(millisecondObj);
} }
if (!(flags & OBJ_SET_GET)) { if (!(flags & OBJ_SET_GET)) {
addReply(c, ok_reply ? ok_reply : shared.ok); addReply(c, ok_reply ? ok_reply : shared.ok);
} }
...@@ -150,6 +130,45 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, ...@@ -150,6 +130,45 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
} }
} }
/*
* Extract the `expire` argument of a given GET/SET command as an absolute timestamp in milliseconds.
*
* "client" is the client that sent the `expire` argument.
* "expire" is the `expire` argument to be extracted.
* "flags" represents the behavior of the command (e.g. PX or EX).
* "unit" is the original unit of the given `expire` argument (e.g. UNIT_SECONDS).
* "milliseconds" is output argument.
*
* If return C_OK, "milliseconds" output argument will be set to the resulting absolute timestamp.
* If return C_ERR, an error reply has been added to the given client.
*/
static int getExpireMillisecondsOrReply(client *c, robj *expire, int flags, int unit, long long *milliseconds) {
int ret = getLongLongFromObjectOrReply(c, expire, milliseconds, NULL);
if (ret != C_OK) {
return ret;
}
if (*milliseconds <= 0 || (unit == UNIT_SECONDS && *milliseconds > LLONG_MAX / 1000)) {
/* Negative value provided or multiplication is gonna overflow. */
addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
return C_ERR;
}
if (unit == UNIT_SECONDS) *milliseconds *= 1000;
if ((flags & OBJ_PX) || (flags & OBJ_EX)) {
*milliseconds += mstime();
}
if (*milliseconds <= 0) {
/* Overflow detected. */
addReplyErrorFormat(c,"invalid expire time in %s",c->cmd->name);
return C_ERR;
}
return C_OK;
}
#define COMMAND_GET 0 #define COMMAND_GET 0
#define COMMAND_SET 1 #define COMMAND_SET 1
/* /*
...@@ -338,26 +357,10 @@ void getexCommand(client *c) { ...@@ -338,26 +357,10 @@ void getexCommand(client *c) {
return; return;
} }
long long milliseconds = 0, when = 0;
/* Validate the expiration time value first */ /* Validate the expiration time value first */
if (expire) { long long milliseconds = 0;
if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK) if (expire && getExpireMillisecondsOrReply(c, expire, flags, unit, &milliseconds) != C_OK) {
return; return;
if (milliseconds <= 0 || (unit == UNIT_SECONDS && milliseconds > LLONG_MAX / 1000)) {
/* Negative value provided or multiplication is gonna overflow. */
addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
return;
}
if (unit == UNIT_SECONDS) milliseconds *= 1000;
when = milliseconds;
if ((flags & OBJ_PX) || (flags & OBJ_EX))
when += mstime();
if (when <= 0) {
/* Overflow detected. */
addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
return;
}
} }
/* We need to do this before we expire the key or delete it */ /* We need to do this before we expire the key or delete it */
...@@ -377,12 +380,12 @@ void getexCommand(client *c) { ...@@ -377,12 +380,12 @@ void getexCommand(client *c) {
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id);
server.dirty++; server.dirty++;
} else if (expire) { } else if (expire) {
setExpire(c,c->db,c->argv[1],when); setExpire(c,c->db,c->argv[1],milliseconds);
/* Propagate */ /* Propagate as PXEXPIREAT millisecond-timestamp if there is
robj *exp = (flags & OBJ_PXAT) || (flags & OBJ_EXAT) ? shared.pexpireat : shared.pexpire; * EX/PX/EXAT/PXAT flag and the key has not expired. */
robj* millisecondObj = createStringObjectFromLongLong(milliseconds); robj *milliseconds_obj = createStringObjectFromLongLong(milliseconds);
rewriteClientCommandVector(c,3,exp,c->argv[1],millisecondObj); rewriteClientCommandVector(c,3,shared.pexpireat,c->argv[1],milliseconds_obj);
decrRefCount(millisecondObj); decrRefCount(milliseconds_obj);
signalModifiedKey(c, c->db, c->argv[1]); signalModifiedKey(c, c->db, c->argv[1]);
notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
......
...@@ -53,7 +53,7 @@ proc test_slave_load_expired_keys {aof} { ...@@ -53,7 +53,7 @@ proc test_slave_load_expired_keys {aof} {
# config the replica persistency and rewrite the config file to survive restart # config the replica persistency and rewrite the config file to survive restart
# note that this needs to be done before populating the volatile keys since # note that this needs to be done before populating the volatile keys since
# that triggers and AOFRW, and we rather the AOF file to have SETEX commands # that triggers and AOFRW, and we rather the AOF file to have 'SET PXAT' commands
# rather than an RDB with volatile keys # rather than an RDB with volatile keys
R $replica_id config set appendonly $aof R $replica_id config set appendonly $aof
R $replica_id config rewrite R $replica_id config rewrite
......
...@@ -228,10 +228,10 @@ tags {"aof"} { ...@@ -228,10 +228,10 @@ tags {"aof"} {
} }
} }
## Test that EXPIREAT is loaded correctly ## Test that PEXPIREAT is loaded correctly
create_aof { create_aof {
append_to_aof [formatCommand rpush list foo] append_to_aof [formatCommand rpush list foo]
append_to_aof [formatCommand expireat list 1000] append_to_aof [formatCommand pexpireat list 1000]
append_to_aof [formatCommand rpush list bar] append_to_aof [formatCommand rpush list bar]
} }
......
...@@ -776,3 +776,29 @@ proc punsubscribe {client {channels {}}} { ...@@ -776,3 +776,29 @@ proc punsubscribe {client {channels {}}} {
$client punsubscribe {*}$channels $client punsubscribe {*}$channels
consume_subscribe_messages $client punsubscribe $channels consume_subscribe_messages $client punsubscribe $channels
} }
proc read_from_aof {fp} {
# Input fp is a blocking binary file descriptor of an opened AOF file.
if {[gets $fp count] == -1} return ""
set count [string range $count 1 end]
# Return a list of arguments for the command.
set res {}
for {set j 0} {$j < $count} {incr j} {
read $fp 1
set arg [::redis::redis_bulk_read $fp]
if {$j == 0} {set arg [string tolower $arg]}
lappend res $arg
}
return $res
}
proc assert_aof_content {aof_path patterns} {
set fp [open $aof_path r]
fconfigure $fp -translation binary
fconfigure $fp -blocking 1
for {set j 0} {$j < [llength $patterns]} {incr j} {
assert_match [lindex $patterns $j] [read_from_aof $fp]
}
}
...@@ -142,16 +142,30 @@ start_server {tags {"expire"}} { ...@@ -142,16 +142,30 @@ start_server {tags {"expire"}} {
assert {$ttl > 900 && $ttl <= 1000} assert {$ttl > 900 && $ttl <= 1000}
} }
test {TTL / PTTL return -1 if key has no expire} { test {TTL / PTTL / EXPIRETIME / PEXPIRETIME return -1 if key has no expire} {
r del x r del x
r set x hello r set x hello
list [r ttl x] [r pttl x] list [r ttl x] [r pttl x] [r expiretime x] [r pexpiretime x]
} {-1 -1} } {-1 -1 -1 -1}
test {TTL / PTTL return -2 if key does not exit} { test {TTL / PTTL / EXPIRETIME / PEXPIRETIME return -2 if key does not exit} {
r del x r del x
list [r ttl x] [r pttl x] list [r ttl x] [r pttl x] [r expiretime x] [r pexpiretime x]
} {-2 -2} } {-2 -2 -2 -2}
test {EXPIRETIME returns absolute expiration time in seconds} {
r del x
set abs_expire [expr [clock seconds] + 100]
r set x somevalue exat $abs_expire
assert_equal [r expiretime x] $abs_expire
}
test {PEXPIRETIME returns absolute expiration time in milliseconds} {
r del x
set abs_expire [expr [clock milliseconds] + 100000]
r set x somevalue pxat $abs_expire
assert_equal [r pexpiretime x] $abs_expire
}
test {Redis should actively expire keys incrementally} { test {Redis should actively expire keys incrementally} {
r flushdb r flushdb
...@@ -264,62 +278,159 @@ start_server {tags {"expire"}} { ...@@ -264,62 +278,159 @@ start_server {tags {"expire"}} {
r ttl foo r ttl foo
} {-2} } {-2}
test {EXPIRE and SET/GETEX EX/PX/EXAT/PXAT option, TTL should not be reset after loadaof} { # Start a new server with empty data and AOF file.
# This test makes sure that expire times are propagated as absolute start_server {overrides {appendonly {yes} appendfilename {appendonly.aof} appendfsync always}} {
# times to the AOF file and not as relative time, so that when the AOF test {All time-to-live(TTL) in commands are propagated as absolute timestamp in milliseconds in AOF} {
# is reloaded the TTLs are not being shifted forward to the future. # This test makes sure that expire times are propagated as absolute
# We want the time to logically pass when the server is restarted! # times to the AOF file and not as relative time, so that when the AOF
# is reloaded the TTLs are not being shifted forward to the future.
r config set appendonly yes # We want the time to logically pass when the server is restarted!
r set foo1 bar EX 100
r set foo2 bar PX 100000 set aof [file join [lindex [r config get dir] 1] [lindex [r config get appendfilename] 1]]
r set foo3 bar
r set foo4 bar # Apply each TTL-related command to a unique key
r expire foo3 100 # SET commands
r pexpire foo4 100000 r set foo1 bar ex 100
r setex foo5 100 bar r set foo2 bar px 100000
r psetex foo6 100000 bar r set foo3 bar exat [expr [clock seconds]+100]
r set foo7 bar EXAT [expr [clock seconds] + 100] r set foo4 bar pxat [expr [clock milliseconds]+100000]
r set foo8 bar PXAT [expr [clock milliseconds] + 100000] r setex foo5 100 bar
r set foo9 bar r psetex foo6 100000 bar
r getex foo9 EX 100 # EXPIRE-family commands
r set foo10 bar r set foo7 bar
r getex foo10 PX 100000 r expire foo7 100
r set foo11 bar r set foo8 bar
r getex foo11 EXAT [expr [clock seconds] + 100] r pexpire foo8 100000
r set foo12 bar r set foo9 bar
r getex foo12 PXAT [expr [clock milliseconds] + 100000] r expireat foo9 [expr [clock seconds]+100]
r set foo10 bar
after 2000 r pexpireat foo10 [expr [clock seconds]*1000+100000]
r debug loadaof r set foo11 bar
assert_range [r ttl foo1] 90 98 r expireat foo11 [expr [clock seconds]-100]
assert_range [r ttl foo2] 90 98 # GETEX commands
assert_range [r ttl foo3] 90 98 r set foo12 bar
assert_range [r ttl foo4] 90 98 r getex foo12 ex 100
assert_range [r ttl foo5] 90 98 r set foo13 bar
assert_range [r ttl foo6] 90 98 r getex foo13 px 100000
assert_range [r ttl foo7] 90 98 r set foo14 bar
assert_range [r ttl foo8] 90 98 r getex foo14 exat [expr [clock seconds]+100]
assert_range [r ttl foo9] 90 98 r set foo15 bar
assert_range [r ttl foo10] 90 98 r getex foo15 pxat [expr [clock milliseconds]+100000]
assert_range [r ttl foo11] 90 98 # RESTORE commands
assert_range [r ttl foo12] 90 98 r set foo16 bar
set encoded [r dump foo16]
r restore foo17 100000 $encoded
r restore foo18 [expr [clock milliseconds]+100000] $encoded absttl
# Assert that each TTL-relatd command are persisted with absolute timestamps in AOF
assert_aof_content $aof {
{select *}
{set foo1 bar PXAT *}
{set foo2 bar PXAT *}
{set foo3 bar PXAT *}
{set foo4 bar PXAT *}
{set foo5 bar PXAT *}
{set foo6 bar PXAT *}
{set foo7 bar}
{pexpireat foo7 *}
{set foo8 bar}
{pexpireat foo8 *}
{set foo9 bar}
{pexpireat foo9 *}
{set foo10 bar}
{pexpireat foo10 *}
{set foo11 bar}
{del foo11}
{set foo12 bar}
{pexpireat foo12 *}
{set foo13 bar}
{pexpireat foo13 *}
{set foo14 bar}
{pexpireat foo14 *}
{set foo15 bar}
{pexpireat foo15 *}
{set foo16 bar}
{restore foo17 * {*} ABSTTL}
{restore foo18 * {*} absttl}
}
# Remember the absolute TTLs of all the keys
set ttl1 [r pexpiretime foo1]
set ttl2 [r pexpiretime foo2]
set ttl3 [r pexpiretime foo3]
set ttl4 [r pexpiretime foo4]
set ttl5 [r pexpiretime foo5]
set ttl6 [r pexpiretime foo6]
set ttl7 [r pexpiretime foo7]
set ttl8 [r pexpiretime foo8]
set ttl9 [r pexpiretime foo9]
set ttl10 [r pexpiretime foo10]
assert_equal "-2" [r pexpiretime foo11] ; # foo11 is gone
set ttl12 [r pexpiretime foo12]
set ttl13 [r pexpiretime foo13]
set ttl14 [r pexpiretime foo14]
set ttl15 [r pexpiretime foo15]
assert_equal "-1" [r pexpiretime foo16] ; # foo16 has no TTL
set ttl17 [r pexpiretime foo17]
set ttl18 [r pexpiretime foo18]
# Let some time pass and reload data from AOF
after 2000
r debug loadaof
# Assert that relative TTLs are roughly the same
assert_range [r ttl foo1] 90 98
assert_range [r ttl foo2] 90 98
assert_range [r ttl foo3] 90 98
assert_range [r ttl foo4] 90 98
assert_range [r ttl foo5] 90 98
assert_range [r ttl foo6] 90 98
assert_range [r ttl foo7] 90 98
assert_range [r ttl foo8] 90 98
assert_range [r ttl foo9] 90 98
assert_range [r ttl foo10] 90 98
assert_equal [r ttl foo11] "-2" ; # foo11 is gone
assert_range [r ttl foo12] 90 98
assert_range [r ttl foo13] 90 98
assert_range [r ttl foo14] 90 98
assert_range [r ttl foo15] 90 98
assert_equal [r ttl foo16] "-1" ; # foo16 has no TTL
assert_range [r ttl foo17] 90 98
assert_range [r ttl foo18] 90 98
# Assert that all keys have restored the same absolute TTLs from AOF
assert_equal [r pexpiretime foo1] $ttl1
assert_equal [r pexpiretime foo2] $ttl2
assert_equal [r pexpiretime foo3] $ttl3
assert_equal [r pexpiretime foo4] $ttl4
assert_equal [r pexpiretime foo5] $ttl5
assert_equal [r pexpiretime foo6] $ttl6
assert_equal [r pexpiretime foo7] $ttl7
assert_equal [r pexpiretime foo8] $ttl8
assert_equal [r pexpiretime foo9] $ttl9
assert_equal [r pexpiretime foo10] $ttl10
assert_equal [r pexpiretime foo11] "-2" ; # foo11 is gone
assert_equal [r pexpiretime foo12] $ttl12
assert_equal [r pexpiretime foo13] $ttl13
assert_equal [r pexpiretime foo14] $ttl14
assert_equal [r pexpiretime foo15] $ttl15
assert_equal [r pexpiretime foo16] "-1" ; # foo16 has no TTL
assert_equal [r pexpiretime foo17] $ttl17
assert_equal [r pexpiretime foo18] $ttl18
}
} }
test {EXPIRE relative and absolute propagation to replicas} { test {All TTL in commands are propagated as absolute timestamp in replication stream} {
# Make sure that relative and absolute expire commands are propagated # Make sure that both relative and absolute expire commands are propagated
# "as is" to replicas. # as absolute to replicas for two reasons:
# We want replicas to honor the same high level contract of expires that # 1) We want to avoid replicas retaining data much longer than primary due
# the master has, that is, we want the time to be counted logically # to replication lag.
# starting from the moment the write was received. This usually provides # 2) We want to unify the way TTLs are replicated in both RDB and replication
# the most coherent behavior from the point of view of the external # stream, which is as absolute timestamps.
# users, with TTLs that are similar from the POV of the external observer. # See: https://github.com/redis/redis/issues/8433
#
# This test is here to stop some innocent / eager optimization or cleanup
# from doing the wrong thing without proper discussion, see:
# https://github.com/redis/redis/pull/5171#issuecomment-409553266
set repl [attach_to_replication_stream] set repl [attach_to_replication_stream]
# SET commands
r set foo1 bar ex 200 r set foo1 bar ex 200
r set foo1 bar px 100000 r set foo1 bar px 100000
r set foo1 bar exat [expr [clock seconds]+100] r set foo1 bar exat [expr [clock seconds]+100]
...@@ -327,37 +438,110 @@ start_server {tags {"expire"}} { ...@@ -327,37 +438,110 @@ start_server {tags {"expire"}} {
r setex foo1 100 bar r setex foo1 100 bar
r psetex foo1 100000 bar r psetex foo1 100000 bar
r set foo2 bar r set foo2 bar
# EXPIRE-family commands
r expire foo2 100 r expire foo2 100
r pexpire foo2 100000 r pexpire foo2 100000
r set foo3 bar r set foo3 bar
r expireat foo3 [expr [clock seconds]+100] r expireat foo3 [expr [clock seconds]+100]
r pexpireat foo3 [expr [clock seconds]*1000+100000] r pexpireat foo3 [expr [clock seconds]*1000+100000]
r expireat foo3 [expr [clock seconds]-100] r expireat foo3 [expr [clock seconds]-100]
# GETEX-family commands
r set foo4 bar r set foo4 bar
r getex foo4 ex 200 r getex foo4 ex 200
r getex foo4 px 200000 r getex foo4 px 200000
r getex foo4 exat [expr [clock seconds]+100] r getex foo4 exat [expr [clock seconds]+100]
r getex foo4 pxat [expr [clock milliseconds]+10000] r getex foo4 pxat [expr [clock milliseconds]+10000]
# RESTORE commands
r set foo5 bar
set encoded [r dump foo5]
r restore foo6 100000 $encoded
r restore foo7 [expr [clock milliseconds]+100000] $encoded absttl
assert_replication_stream $repl { assert_replication_stream $repl {
{select *} {select *}
{set foo1 bar PX 200000}
{set foo1 bar PX 100000}
{set foo1 bar PXAT *} {set foo1 bar PXAT *}
{set foo1 bar PXAT *} {set foo1 bar PXAT *}
{set foo1 bar PX 100000} {set foo1 bar PXAT *}
{set foo1 bar PX 100000} {set foo1 bar PXAT *}
{set foo1 bar PXAT *}
{set foo1 bar PXAT *}
{set foo2 bar} {set foo2 bar}
{expire foo2 100} {pexpireat foo2 *}
{pexpire foo2 100000} {pexpireat foo2 *}
{set foo3 bar} {set foo3 bar}
{expireat foo3 *} {pexpireat foo3 *}
{pexpireat foo3 *} {pexpireat foo3 *}
{del foo3} {del foo3}
{set foo4 bar} {set foo4 bar}
{pexpire foo4 200000}
{pexpire foo4 200000}
{pexpireat foo4 *} {pexpireat foo4 *}
{pexpireat foo4 *} {pexpireat foo4 *}
{pexpireat foo4 *}
{pexpireat foo4 *}
{set foo5 bar}
{restore foo6 * {*} ABSTTL}
{restore foo7 * {*} absttl}
}
}
# Start another server to test replication of TTLs
start_server {} {
# Set the outer layer server as primary
set primary [srv -1 client]
set primary_host [srv -1 host]
set primary_port [srv -1 port]
# Set this inner layer server as replica
set replica [srv 0 client]
test {First server should have role slave after REPLICAOF} {
$replica replicaof $primary_host $primary_port
wait_for_condition 50 100 {
[s 0 role] eq {slave}
} else {
fail "Replication not started."
}
}
test {For all replicated TTL-related commands, absolute expire times are identical on primary and replica} {
# Apply each TTL-related command to a unique key on primary
# SET commands
$primary set foo1 bar ex 100
$primary set foo2 bar px 100000
$primary set foo3 bar exat [expr [clock seconds]+100]
$primary set foo4 bar pxat [expr [clock milliseconds]+100000]
$primary setex foo5 100 bar
$primary psetex foo6 100000 bar
# EXPIRE-family commands
$primary set foo7 bar
$primary expire foo7 100
$primary set foo8 bar
$primary pexpire foo8 100000
$primary set foo9 bar
$primary expireat foo9 [expr [clock seconds]+100]
$primary set foo10 bar
$primary pexpireat foo10 [expr [clock milliseconds]+100000]
# GETEX commands
$primary set foo11 bar
$primary getex foo11 ex 100
$primary set foo12 bar
$primary getex foo12 px 100000
$primary set foo13 bar
$primary getex foo13 exat [expr [clock seconds]+100]
$primary set foo14 bar
$primary getex foo14 pxat [expr [clock milliseconds]+100000]
# RESTORE commands
$primary set foo15 bar
set encoded [$primary dump foo15]
$primary restore foo16 100000 $encoded
$primary restore foo17 [expr [clock milliseconds]+100000] $encoded absttl
# Wait for replica to get the keys and TTLs
assert {[$primary wait 1 0] == 1}
# Verify absolute TTLs are identical on primary and replica for all keys
# This is because TTLs are always replicated as absolute values
foreach key [$primary keys *] {
assert_equal [$primary pexpiretime $key] [$replica pexpiretime $key]
}
} }
} }
...@@ -406,7 +590,7 @@ start_server {tags {"expire"}} { ...@@ -406,7 +590,7 @@ start_server {tags {"expire"}} {
r getex foo exat [expr [clock seconds]-100] r getex foo exat [expr [clock seconds]-100]
assert_replication_stream $repl { assert_replication_stream $repl {
{select *} {select *}
{set foo bar PX 100000} {set foo bar PXAT *}
{persist foo} {persist foo}
{del foo} {del foo}
} }
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment