Commit 72935b9d authored by Vitaly Arbuzov's avatar Vitaly Arbuzov
Browse files

Merge branch 'unstable' into dict-split-by-slot

parents 6baf20af 6948daca
...@@ -34,6 +34,10 @@ typedef long long ustime_t; ...@@ -34,6 +34,10 @@ typedef long long ustime_t;
#define REDISMODULE_OK 0 #define REDISMODULE_OK 0
#define REDISMODULE_ERR 1 #define REDISMODULE_ERR 1
/* Module Based Authentication status return values. */
#define REDISMODULE_AUTH_HANDLED 0
#define REDISMODULE_AUTH_NOT_HANDLED 1
/* API versions. */ /* API versions. */
#define REDISMODULE_APIVER_1 1 #define REDISMODULE_APIVER_1 1
...@@ -89,6 +93,7 @@ typedef long long ustime_t; ...@@ -89,6 +93,7 @@ typedef long long ustime_t;
#define REDISMODULE_REPLY_BIG_NUMBER 9 #define REDISMODULE_REPLY_BIG_NUMBER 9
#define REDISMODULE_REPLY_VERBATIM_STRING 10 #define REDISMODULE_REPLY_VERBATIM_STRING 10
#define REDISMODULE_REPLY_ATTRIBUTE 11 #define REDISMODULE_REPLY_ATTRIBUTE 11
#define REDISMODULE_REPLY_PROMISE 12
/* Postponed array length. */ /* Postponed array length. */
#define REDISMODULE_POSTPONED_ARRAY_LEN -1 /* Deprecated, please use REDISMODULE_POSTPONED_LEN */ #define REDISMODULE_POSTPONED_ARRAY_LEN -1 /* Deprecated, please use REDISMODULE_POSTPONED_LEN */
...@@ -912,6 +917,8 @@ typedef int (*RedisModuleConfigSetNumericFunc)(const char *name, long long val, ...@@ -912,6 +917,8 @@ typedef int (*RedisModuleConfigSetNumericFunc)(const char *name, long long val,
typedef int (*RedisModuleConfigSetBoolFunc)(const char *name, int val, void *privdata, RedisModuleString **err); typedef int (*RedisModuleConfigSetBoolFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
typedef int (*RedisModuleConfigSetEnumFunc)(const char *name, int val, void *privdata, RedisModuleString **err); typedef int (*RedisModuleConfigSetEnumFunc)(const char *name, int val, void *privdata, RedisModuleString **err);
typedef int (*RedisModuleConfigApplyFunc)(RedisModuleCtx *ctx, void *privdata, RedisModuleString **err); typedef int (*RedisModuleConfigApplyFunc)(RedisModuleCtx *ctx, void *privdata, RedisModuleString **err);
typedef void (*RedisModuleOnUnblocked)(RedisModuleCtx *ctx, RedisModuleCallReply *reply, void *private_data);
typedef int (*RedisModuleAuthCallback)(RedisModuleCtx *ctx, RedisModuleString *username, RedisModuleString *password, RedisModuleString **err);
typedef struct RedisModuleTypeMethods { typedef struct RedisModuleTypeMethods {
uint64_t version; uint64_t version;
...@@ -959,6 +966,7 @@ REDISMODULE_API int (*RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char ...@@ -959,6 +966,7 @@ REDISMODULE_API int (*RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char
REDISMODULE_API RedisModuleCommand *(*RedisModule_GetCommand)(RedisModuleCtx *ctx, const char *name) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCommand *(*RedisModule_GetCommand)(RedisModuleCtx *ctx, const char *name) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_CreateSubcommand)(RedisModuleCommand *parent, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_CreateSubcommand)(RedisModuleCommand *parent, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_SetCommandInfo)(RedisModuleCommand *command, const RedisModuleCommandInfo *info) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_SetCommandInfo)(RedisModuleCommand *command, const RedisModuleCommandInfo *info) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_SetCommandACLCategories)(RedisModuleCommand *command, const char *ctgrsflags) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_IsModuleNameBusy)(const char *name) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_IsModuleNameBusy)(const char *name) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_WrongArity)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_WrongArity)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -989,6 +997,8 @@ REDISMODULE_API const char* (*RedisModule_CallReplyVerbatim)(RedisModuleCallRepl ...@@ -989,6 +997,8 @@ REDISMODULE_API const char* (*RedisModule_CallReplyVerbatim)(RedisModuleCallRepl
REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplySetElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplySetElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_CallReplyMapElement)(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_CallReplyMapElement)(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_CallReplyAttributeElement)(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_CallReplyAttributeElement)(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_CallReplyPromiseSetUnblockHandler)(RedisModuleCallReply *reply, RedisModuleOnUnblocked on_unblock, void *private_data) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_CallReplyPromiseAbort)(RedisModuleCallReply *reply, void **private_data) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyAttribute)(RedisModuleCallReply *reply) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyAttribute)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
REDISMODULE_API size_t (*RedisModule_CallReplyLength)(RedisModuleCallReply *reply) REDISMODULE_ATTR; REDISMODULE_API size_t (*RedisModule_CallReplyLength)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR;
...@@ -1164,6 +1174,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_DictPrev)(RedisModuleCtx *ctx, ...@@ -1164,6 +1174,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_DictPrev)(RedisModuleCtx *ctx,
REDISMODULE_API int (*RedisModule_DictCompareC)(RedisModuleDictIter *di, const char *op, void *key, size_t keylen) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_DictCompareC)(RedisModuleDictIter *di, const char *op, void *key, size_t keylen) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_DictCompare)(RedisModuleDictIter *di, const char *op, RedisModuleString *key) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_DictCompare)(RedisModuleDictIter *di, const char *op, RedisModuleString *key) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_RegisterInfoFunc)(RedisModuleCtx *ctx, RedisModuleInfoFunc cb) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_RegisterInfoFunc)(RedisModuleCtx *ctx, RedisModuleInfoFunc cb) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_RegisterAuthCallback)(RedisModuleCtx *ctx, RedisModuleAuthCallback cb) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_InfoAddSection)(RedisModuleInfoCtx *ctx, const char *name) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_InfoAddSection)(RedisModuleInfoCtx *ctx, const char *name) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_InfoBeginDictField)(RedisModuleInfoCtx *ctx, const char *name) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_InfoBeginDictField)(RedisModuleInfoCtx *ctx, const char *name) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_InfoEndDictField)(RedisModuleInfoCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_InfoEndDictField)(RedisModuleInfoCtx *ctx) REDISMODULE_ATTR;
...@@ -1201,6 +1212,9 @@ REDISMODULE_API int (*RedisModule_GetServerVersion)() REDISMODULE_ATTR; ...@@ -1201,6 +1212,9 @@ REDISMODULE_API int (*RedisModule_GetServerVersion)() REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_GetTypeMethodVersion)() REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_GetTypeMethodVersion)() REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_Yield)(RedisModuleCtx *ctx, int flags, const char *busy_reply) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_Yield)(RedisModuleCtx *ctx, int flags, const char *busy_reply) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleBlockedClient * (*RedisModule_BlockClient)(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms) REDISMODULE_ATTR; REDISMODULE_API RedisModuleBlockedClient * (*RedisModule_BlockClient)(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms) REDISMODULE_ATTR;
REDISMODULE_API void * (*RedisModule_BlockClientGetPrivateData)(RedisModuleBlockedClient *blocked_client) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_BlockClientSetPrivateData)(RedisModuleBlockedClient *blocked_client, void *private_data) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleBlockedClient * (*RedisModule_BlockClientOnAuth)(RedisModuleCtx *ctx, RedisModuleAuthCallback reply_callback, void (*free_privdata)(RedisModuleCtx*,void*)) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_UnblockClient)(RedisModuleBlockedClient *bc, void *privdata) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_UnblockClient)(RedisModuleBlockedClient *bc, void *privdata) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_IsBlockedReplyRequest)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_IsBlockedReplyRequest)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_IsBlockedTimeoutRequest)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_IsBlockedTimeoutRequest)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -1264,6 +1278,7 @@ REDISMODULE_API int (*RedisModule_ACLCheckCommandPermissions)(RedisModuleUser *u ...@@ -1264,6 +1278,7 @@ REDISMODULE_API int (*RedisModule_ACLCheckCommandPermissions)(RedisModuleUser *u
REDISMODULE_API int (*RedisModule_ACLCheckKeyPermissions)(RedisModuleUser *user, RedisModuleString *key, int flags) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_ACLCheckKeyPermissions)(RedisModuleUser *user, RedisModuleString *key, int flags) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_ACLCheckChannelPermissions)(RedisModuleUser *user, RedisModuleString *ch, int literal) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_ACLCheckChannelPermissions)(RedisModuleUser *user, RedisModuleString *ch, int literal) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_ACLAddLogEntry)(RedisModuleCtx *ctx, RedisModuleUser *user, RedisModuleString *object, RedisModuleACLLogEntryReason reason) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_ACLAddLogEntry)(RedisModuleCtx *ctx, RedisModuleUser *user, RedisModuleString *object, RedisModuleACLLogEntryReason reason) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_ACLAddLogEntryByUserName)(RedisModuleCtx *ctx, RedisModuleString *user, RedisModuleString *object, RedisModuleACLLogEntryReason reason) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_AuthenticateClientWithACLUser)(RedisModuleCtx *ctx, const char *name, size_t len, RedisModuleUserChangedFunc callback, void *privdata, uint64_t *client_id) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_AuthenticateClientWithACLUser)(RedisModuleCtx *ctx, const char *name, size_t len, RedisModuleUserChangedFunc callback, void *privdata, uint64_t *client_id) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_AuthenticateClientWithUser)(RedisModuleCtx *ctx, RedisModuleUser *user, RedisModuleUserChangedFunc callback, void *privdata, uint64_t *client_id) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_AuthenticateClientWithUser)(RedisModuleCtx *ctx, RedisModuleUser *user, RedisModuleUserChangedFunc callback, void *privdata, uint64_t *client_id) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_DeauthenticateAndCloseClient)(RedisModuleCtx *ctx, uint64_t client_id) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_DeauthenticateAndCloseClient)(RedisModuleCtx *ctx, uint64_t client_id) REDISMODULE_ATTR;
...@@ -1306,6 +1321,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1306,6 +1321,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(GetCommand); REDISMODULE_GET_API(GetCommand);
REDISMODULE_GET_API(CreateSubcommand); REDISMODULE_GET_API(CreateSubcommand);
REDISMODULE_GET_API(SetCommandInfo); REDISMODULE_GET_API(SetCommandInfo);
REDISMODULE_GET_API(SetCommandACLCategories);
REDISMODULE_GET_API(SetModuleAttribs); REDISMODULE_GET_API(SetModuleAttribs);
REDISMODULE_GET_API(IsModuleNameBusy); REDISMODULE_GET_API(IsModuleNameBusy);
REDISMODULE_GET_API(WrongArity); REDISMODULE_GET_API(WrongArity);
...@@ -1365,6 +1381,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1365,6 +1381,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(CallReplySetElement); REDISMODULE_GET_API(CallReplySetElement);
REDISMODULE_GET_API(CallReplyMapElement); REDISMODULE_GET_API(CallReplyMapElement);
REDISMODULE_GET_API(CallReplyAttributeElement); REDISMODULE_GET_API(CallReplyAttributeElement);
REDISMODULE_GET_API(CallReplyPromiseSetUnblockHandler);
REDISMODULE_GET_API(CallReplyPromiseAbort);
REDISMODULE_GET_API(CallReplyAttribute); REDISMODULE_GET_API(CallReplyAttribute);
REDISMODULE_GET_API(CallReplyType); REDISMODULE_GET_API(CallReplyType);
REDISMODULE_GET_API(CallReplyLength); REDISMODULE_GET_API(CallReplyLength);
...@@ -1506,6 +1524,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1506,6 +1524,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(DictCompare); REDISMODULE_GET_API(DictCompare);
REDISMODULE_GET_API(DictCompareC); REDISMODULE_GET_API(DictCompareC);
REDISMODULE_GET_API(RegisterInfoFunc); REDISMODULE_GET_API(RegisterInfoFunc);
REDISMODULE_GET_API(RegisterAuthCallback);
REDISMODULE_GET_API(InfoAddSection); REDISMODULE_GET_API(InfoAddSection);
REDISMODULE_GET_API(InfoBeginDictField); REDISMODULE_GET_API(InfoBeginDictField);
REDISMODULE_GET_API(InfoEndDictField); REDISMODULE_GET_API(InfoEndDictField);
...@@ -1554,6 +1573,9 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1554,6 +1573,9 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ThreadSafeContextTryLock); REDISMODULE_GET_API(ThreadSafeContextTryLock);
REDISMODULE_GET_API(ThreadSafeContextUnlock); REDISMODULE_GET_API(ThreadSafeContextUnlock);
REDISMODULE_GET_API(BlockClient); REDISMODULE_GET_API(BlockClient);
REDISMODULE_GET_API(BlockClientGetPrivateData);
REDISMODULE_GET_API(BlockClientSetPrivateData);
REDISMODULE_GET_API(BlockClientOnAuth);
REDISMODULE_GET_API(UnblockClient); REDISMODULE_GET_API(UnblockClient);
REDISMODULE_GET_API(IsBlockedReplyRequest); REDISMODULE_GET_API(IsBlockedReplyRequest);
REDISMODULE_GET_API(IsBlockedTimeoutRequest); REDISMODULE_GET_API(IsBlockedTimeoutRequest);
...@@ -1611,6 +1633,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1611,6 +1633,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ACLCheckKeyPermissions); REDISMODULE_GET_API(ACLCheckKeyPermissions);
REDISMODULE_GET_API(ACLCheckChannelPermissions); REDISMODULE_GET_API(ACLCheckChannelPermissions);
REDISMODULE_GET_API(ACLAddLogEntry); REDISMODULE_GET_API(ACLAddLogEntry);
REDISMODULE_GET_API(ACLAddLogEntryByUserName);
REDISMODULE_GET_API(DeauthenticateAndCloseClient); REDISMODULE_GET_API(DeauthenticateAndCloseClient);
REDISMODULE_GET_API(AuthenticateClientWithACLUser); REDISMODULE_GET_API(AuthenticateClientWithACLUser);
REDISMODULE_GET_API(AuthenticateClientWithUser); REDISMODULE_GET_API(AuthenticateClientWithUser);
......
...@@ -332,9 +332,8 @@ void feedReplicationBuffer(char *s, size_t len) { ...@@ -332,9 +332,8 @@ void feedReplicationBuffer(char *s, size_t len) {
static long long repl_block_id = 0; static long long repl_block_id = 0;
if (server.repl_backlog == NULL) return; if (server.repl_backlog == NULL) return;
server.master_repl_offset += len;
server.repl_backlog->histlen += len;
while(len > 0) {
size_t start_pos = 0; /* The position of referenced block to start sending. */ size_t start_pos = 0; /* The position of referenced block to start sending. */
listNode *start_node = NULL; /* Replica/backlog starts referenced node. */ listNode *start_node = NULL; /* Replica/backlog starts referenced node. */
int add_new_block = 0; /* Create new block if current block is total used. */ int add_new_block = 0; /* Create new block if current block is total used. */
...@@ -353,20 +352,27 @@ void feedReplicationBuffer(char *s, size_t len) { ...@@ -353,20 +352,27 @@ void feedReplicationBuffer(char *s, size_t len) {
tail->used += copy; tail->used += copy;
s += copy; s += copy;
len -= copy; len -= copy;
server.master_repl_offset += copy;
server.repl_backlog->histlen += copy;
} }
if (len) { if (len) {
/* Create a new node, make sure it is allocated to at /* Create a new node, make sure it is allocated to at
* least PROTO_REPLY_CHUNK_BYTES */ * least PROTO_REPLY_CHUNK_BYTES */
size_t usable_size; size_t usable_size;
size_t size = (len < PROTO_REPLY_CHUNK_BYTES) ? PROTO_REPLY_CHUNK_BYTES : len; /* Avoid creating nodes smaller than PROTO_REPLY_CHUNK_BYTES, so that we can append more data into them,
* and also avoid creating nodes bigger than repl_backlog_size / 16, so that we won't have huge nodes that can't
* trim when we only still need to hold a small portion from them. */
size_t limit = max((size_t)server.repl_backlog_size / 16, (size_t)PROTO_REPLY_CHUNK_BYTES);
size_t size = min(max(len, (size_t)PROTO_REPLY_CHUNK_BYTES), limit);
tail = zmalloc_usable(size + sizeof(replBufBlock), &usable_size); tail = zmalloc_usable(size + sizeof(replBufBlock), &usable_size);
/* Take over the allocation's internal fragmentation */ /* Take over the allocation's internal fragmentation */
tail->size = usable_size - sizeof(replBufBlock); tail->size = usable_size - sizeof(replBufBlock);
tail->used = len; size_t copy = (tail->size >= len) ? len : tail->size;
tail->used = copy;
tail->refcount = 0; tail->refcount = 0;
tail->repl_offset = server.master_repl_offset - tail->used + 1; tail->repl_offset = server.master_repl_offset + 1;
tail->id = repl_block_id++; tail->id = repl_block_id++;
memcpy(tail->buf, s, len); memcpy(tail->buf, s, copy);
listAddNodeTail(server.repl_buffer_blocks, tail); listAddNodeTail(server.repl_buffer_blocks, tail);
/* We also count the list node memory into replication buffer memory. */ /* We also count the list node memory into replication buffer memory. */
server.repl_buffer_mem += (usable_size + sizeof(listNode)); server.repl_buffer_mem += (usable_size + sizeof(listNode));
...@@ -375,6 +381,10 @@ void feedReplicationBuffer(char *s, size_t len) { ...@@ -375,6 +381,10 @@ void feedReplicationBuffer(char *s, size_t len) {
start_node = listLast(server.repl_buffer_blocks); start_node = listLast(server.repl_buffer_blocks);
start_pos = 0; start_pos = 0;
} }
s += copy;
len -= copy;
server.master_repl_offset += copy;
server.repl_backlog->histlen += copy;
} }
/* For output buffer of replicas. */ /* For output buffer of replicas. */
...@@ -415,6 +425,7 @@ void feedReplicationBuffer(char *s, size_t len) { ...@@ -415,6 +425,7 @@ void feedReplicationBuffer(char *s, size_t len) {
* new backlog node is added and one should be removed. See also comments * new backlog node is added and one should be removed. See also comments
* in freeMemoryGetNotCountedMemory for details. */ * in freeMemoryGetNotCountedMemory for details. */
incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
}
} }
/* Propagate write commands to replication stream. /* Propagate write commands to replication stream.
...@@ -440,7 +451,13 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) { ...@@ -440,7 +451,13 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
/* If there aren't slaves, and there is no backlog buffer to populate, /* If there aren't slaves, and there is no backlog buffer to populate,
* we can return ASAP. */ * we can return ASAP. */
if (server.repl_backlog == NULL && listLength(slaves) == 0) return; if (server.repl_backlog == NULL && listLength(slaves) == 0) {
/* We increment the repl_offset anyway, since we use that for tracking AOF fsyncs
* even when there's no replication active. This code will not be reached if AOF
* is also disabled. */
server.master_repl_offset += 1;
return;
}
/* We can't have slaves attached and no backlog. */ /* We can't have slaves attached and no backlog. */
serverAssert(!(listLength(slaves) != 0 && server.repl_backlog == NULL)); serverAssert(!(listLength(slaves) != 0 && server.repl_backlog == NULL));
...@@ -1129,11 +1146,12 @@ void syncCommand(client *c) { ...@@ -1129,11 +1146,12 @@ void syncCommand(client *c) {
* eof: supports EOF-style RDB transfer for diskless replication. * eof: supports EOF-style RDB transfer for diskless replication.
* psync2: supports PSYNC v2, so understands +CONTINUE <new repl ID>. * psync2: supports PSYNC v2, so understands +CONTINUE <new repl ID>.
* *
* - ack <offset> * - ack <offset> [fack <aofofs>]
* Replica informs the master the amount of replication stream that it * Replica informs the master the amount of replication stream that it
* processed so far. * processed so far, and optionally the replication offset fsynced to the AOF file.
* This special pattern doesn't reply to the caller.
* *
* - getack * - getack <dummy>
* Unlike other subcommands, this is used by master to get the replication * Unlike other subcommands, this is used by master to get the replication
* offset from a replica. * offset from a replica.
* *
...@@ -1190,6 +1208,12 @@ void replconfCommand(client *c) { ...@@ -1190,6 +1208,12 @@ void replconfCommand(client *c) {
return; return;
if (offset > c->repl_ack_off) if (offset > c->repl_ack_off)
c->repl_ack_off = offset; c->repl_ack_off = offset;
if (c->argc > j+3 && !strcasecmp(c->argv[j+2]->ptr,"fack")) {
if ((getLongLongFromObject(c->argv[j+3], &offset) != C_OK))
return;
if (offset > c->repl_aof_off)
c->repl_aof_off = offset;
}
c->repl_ack_time = server.unixtime; c->repl_ack_time = server.unixtime;
/* If this was a diskless replication, we need to really put /* If this was a diskless replication, we need to really put
* the slave online when the first ACK is received (which * the slave online when the first ACK is received (which
...@@ -1740,16 +1764,16 @@ void replicationCreateMasterClient(connection *conn, int dbid) { ...@@ -1740,16 +1764,16 @@ void replicationCreateMasterClient(connection *conn, int dbid) {
connSetReadHandler(server.master->conn, readQueryFromClient); connSetReadHandler(server.master->conn, readQueryFromClient);
/** /**
     * Important note: * Important note:
     * The CLIENT_DENY_BLOCKING flag is not, and should not, be set here. * The CLIENT_DENY_BLOCKING flag is not, and should not, be set here.
     * For commands like BLPOP, it makes no sense to block the master * For commands like BLPOP, it makes no sense to block the master
     * connection, and such blocking attempt will probably cause deadlock and * connection, and such blocking attempt will probably cause deadlock and
     * break the replication. We consider such a thing as a bug because * break the replication. We consider such a thing as a bug because
    * commands as BLPOP should never be sent on the replication link. * commands as BLPOP should never be sent on the replication link.
     * A possible use-case for blocking the replication link is if a module wants * A possible use-case for blocking the replication link is if a module wants
     * to pass the execution to a background thread and unblock after the * to pass the execution to a background thread and unblock after the
     * execution is done. This is the reason why we allow blocking the replication * execution is done. This is the reason why we allow blocking the replication
     * connection. */ * connection. */
server.master->flags |= CLIENT_MASTER; server.master->flags |= CLIENT_MASTER;
server.master->authenticated = 1; server.master->authenticated = 1;
...@@ -3237,11 +3261,16 @@ void replicationSendAck(void) { ...@@ -3237,11 +3261,16 @@ void replicationSendAck(void) {
client *c = server.master; client *c = server.master;
if (c != NULL) { if (c != NULL) {
int send_fack = server.fsynced_reploff != -1;
c->flags |= CLIENT_MASTER_FORCE_REPLY; c->flags |= CLIENT_MASTER_FORCE_REPLY;
addReplyArrayLen(c,3); addReplyArrayLen(c,send_fack ? 5 : 3);
addReplyBulkCString(c,"REPLCONF"); addReplyBulkCString(c,"REPLCONF");
addReplyBulkCString(c,"ACK"); addReplyBulkCString(c,"ACK");
addReplyBulkLongLong(c,c->reploff); addReplyBulkLongLong(c,c->reploff);
if (send_fack) {
addReplyBulkCString(c,"FACK");
addReplyBulkLongLong(c,server.fsynced_reploff);
}
c->flags &= ~CLIENT_MASTER_FORCE_REPLY; c->flags &= ~CLIENT_MASTER_FORCE_REPLY;
} }
} }
...@@ -3476,6 +3505,23 @@ int replicationCountAcksByOffset(long long offset) { ...@@ -3476,6 +3505,23 @@ int replicationCountAcksByOffset(long long offset) {
return count; return count;
} }
/* Return the number of replicas that already acknowledged the specified
* replication offset being AOF fsynced. */
int replicationCountAOFAcksByOffset(long long offset) {
listIter li;
listNode *ln;
int count = 0;
listRewind(server.slaves,&li);
while((ln = listNext(&li))) {
client *slave = ln->value;
if (slave->replstate != SLAVE_STATE_ONLINE) continue;
if (slave->repl_aof_off >= offset) count++;
}
return count;
}
/* WAIT for N replicas to acknowledge the processing of our latest /* WAIT for N replicas to acknowledge the processing of our latest
* write command (and all the previous commands). */ * write command (and all the previous commands). */
void waitCommand(client *c) { void waitCommand(client *c) {
...@@ -3496,7 +3542,7 @@ void waitCommand(client *c) { ...@@ -3496,7 +3542,7 @@ void waitCommand(client *c) {
/* First try without blocking at all. */ /* First try without blocking at all. */
ackreplicas = replicationCountAcksByOffset(c->woff); ackreplicas = replicationCountAcksByOffset(c->woff);
if (ackreplicas >= numreplicas || c->flags & CLIENT_MULTI) { if (ackreplicas >= numreplicas || c->flags & CLIENT_DENY_BLOCKING) {
addReplyLongLong(c,ackreplicas); addReplyLongLong(c,ackreplicas);
return; return;
} }
...@@ -3510,6 +3556,48 @@ void waitCommand(client *c) { ...@@ -3510,6 +3556,48 @@ void waitCommand(client *c) {
replicationRequestAckFromSlaves(); replicationRequestAckFromSlaves();
} }
/* WAIT for N replicas and / or local master to acknowledge our latest
* write command got synced to the disk. */
void waitaofCommand(client *c) {
mstime_t timeout;
long numreplicas, numlocal, ackreplicas, acklocal;
/* Argument parsing. */
if (getRangeLongFromObjectOrReply(c,c->argv[1],0,1,&numlocal,NULL) != C_OK)
return;
if (getPositiveLongFromObjectOrReply(c,c->argv[2],&numreplicas,NULL) != C_OK)
return;
if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_MILLISECONDS) != C_OK)
return;
if (server.masterhost) {
addReplyError(c,"WAITAOF cannot be used with replica instances. Please also note that writes to replicas are just local and are not propagated.");
return;
}
if (numlocal && !server.aof_enabled) {
addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled.");
return;
}
/* First try without blocking at all. */
ackreplicas = replicationCountAOFAcksByOffset(c->woff);
acklocal = server.fsynced_reploff >= c->woff;
if ((ackreplicas >= numreplicas && acklocal >= numlocal) || c->flags & CLIENT_DENY_BLOCKING) {
addReplyArrayLen(c,2);
addReplyLongLong(c,acklocal);
addReplyLongLong(c,ackreplicas);
return;
}
/* Otherwise block the client and put it into our list of clients
* waiting for ack from slaves. */
blockForAofFsync(c,timeout,c->woff,numlocal,numreplicas);
/* Make sure that the server will send an ACK request to all the slaves
* before returning to the event loop. */
replicationRequestAckFromSlaves();
}
/* This is called by unblockClient() to perform the blocking op type /* This is called by unblockClient() to perform the blocking op type
* specific cleanup. We just remove the client from the list of clients * specific cleanup. We just remove the client from the list of clients
* waiting for replica acks. Never call it directly, call unblockClient() * waiting for replica acks. Never call it directly, call unblockClient()
...@@ -3520,38 +3608,78 @@ void unblockClientWaitingReplicas(client *c) { ...@@ -3520,38 +3608,78 @@ void unblockClientWaitingReplicas(client *c) {
listDelNode(server.clients_waiting_acks,ln); listDelNode(server.clients_waiting_acks,ln);
} }
/* Check if there are clients blocked in WAIT that can be unblocked since /* Check if there are clients blocked in WAIT or WAITAOF that can be unblocked
* we received enough ACKs from slaves. */ * since we received enough ACKs from slaves. */
void processClientsWaitingReplicas(void) { void processClientsWaitingReplicas(void) {
long long last_offset = 0; long long last_offset = 0;
long long last_aof_offset = 0;
int last_numreplicas = 0; int last_numreplicas = 0;
int last_aof_numreplicas = 0;
listIter li; listIter li;
listNode *ln; listNode *ln;
listRewind(server.clients_waiting_acks,&li); listRewind(server.clients_waiting_acks,&li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
int numlocal = 0;
int numreplicas = 0;
client *c = ln->value; client *c = ln->value;
int is_wait_aof = c->bstate.btype == BLOCKED_WAITAOF;
if (is_wait_aof && c->bstate.numlocal && !server.aof_enabled) {
addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled.");
unblockClient(c, 1);
continue;
}
/* Every time we find a client that is satisfied for a given /* Every time we find a client that is satisfied for a given
* offset and number of replicas, we remember it so the next client * offset and number of replicas, we remember it so the next client
* may be unblocked without calling replicationCountAcksByOffset() * may be unblocked without calling replicationCountAcksByOffset()
* or calling replicationCountAOFAcksByOffset()
* if the requested offset / replicas were equal or less. */ * if the requested offset / replicas were equal or less. */
if (last_offset && last_offset >= c->bstate.reploffset && if (!is_wait_aof && last_offset && last_offset >= c->bstate.reploffset &&
last_numreplicas >= c->bstate.numreplicas) last_numreplicas >= c->bstate.numreplicas)
{ {
unblockClient(c); numreplicas = last_numreplicas;
addReplyLongLong(c,last_numreplicas); } else if (is_wait_aof && last_aof_offset && last_aof_offset >= c->bstate.reploffset &&
last_aof_numreplicas >= c->bstate.numreplicas)
{
numreplicas = last_aof_numreplicas;
} else { } else {
int numreplicas = replicationCountAcksByOffset(c->bstate.reploffset); numreplicas = is_wait_aof ?
replicationCountAOFAcksByOffset(c->bstate.reploffset) :
replicationCountAcksByOffset(c->bstate.reploffset);
/* Check if the number of replicas is satisfied. */
if (numreplicas < c->bstate.numreplicas) continue;
if (numreplicas >= c->bstate.numreplicas) { if (is_wait_aof) {
last_aof_offset = c->bstate.reploffset;
last_aof_numreplicas = numreplicas;
} else {
last_offset = c->bstate.reploffset; last_offset = c->bstate.reploffset;
last_numreplicas = numreplicas; last_numreplicas = numreplicas;
unblockClient(c);
addReplyLongLong(c,numreplicas);
} }
} }
/* Check if the local constraint of WAITAOF is served */
if (is_wait_aof) {
numlocal = server.fsynced_reploff >= c->bstate.reploffset;
if (numlocal < c->bstate.numlocal) continue;
}
/* Reply before unblocking, because unblock client calls reqresAppendResponse */
if (is_wait_aof) {
/* WAITAOF has an array reply */
addReplyArrayLen(c, 2);
addReplyLongLong(c, numlocal);
addReplyLongLong(c, numreplicas);
} else {
addReplyLongLong(c, numreplicas);
}
unblockClient(c, 1);
} }
} }
......
...@@ -35,7 +35,7 @@ ...@@ -35,7 +35,7 @@
* callback represents a different reply type. Each callback gets a p_ctx that * callback represents a different reply type. Each callback gets a p_ctx that
* was given to the parseReply function. The callbacks also give the protocol * was given to the parseReply function. The callbacks also give the protocol
* (underlying blob) of the current reply and the size. * (underlying blob) of the current reply and the size.
 * *
* Some callbacks also get the parser object itself: * Some callbacks also get the parser object itself:
* - array_callback * - array_callback
* - set_callback * - set_callback
......
...@@ -160,7 +160,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca ...@@ -160,7 +160,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
return C_ERR; return C_ERR;
} }
/* Deny writes if we're unale to persist. */ /* Deny writes if we're unable to persist. */
int deny_write_type = writeCommandsDeniedByDiskError(); int deny_write_type = writeCommandsDeniedByDiskError();
if (deny_write_type != DISK_ERROR_TYPE_NONE && !obey_client) { if (deny_write_type != DISK_ERROR_TYPE_NONE && !obey_client) {
if (deny_write_type == DISK_ERROR_TYPE_RDB) if (deny_write_type == DISK_ERROR_TYPE_RDB)
......
...@@ -1562,8 +1562,8 @@ static void luaMaskCountHook(lua_State *lua, lua_Debug *ar) { ...@@ -1562,8 +1562,8 @@ static void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
/* /*
* Set the hook to invoke all the time so the user * Set the hook to invoke all the time so the user
         * will not be able to catch the error with pcall and invoke * will not be able to catch the error with pcall and invoke
         * pcall again which will prevent the script from ever been killed * pcall again which will prevent the script from ever been killed
*/ */
lua_sethook(lua, luaMaskCountHook, LUA_MASKLINE, 0); lua_sethook(lua, luaMaskCountHook, LUA_MASKLINE, 0);
......
...@@ -77,6 +77,7 @@ typedef struct sentinelAddr { ...@@ -77,6 +77,7 @@ typedef struct sentinelAddr {
#define SRI_FORCE_FAILOVER (1<<11) /* Force failover with master up. */ #define SRI_FORCE_FAILOVER (1<<11) /* Force failover with master up. */
#define SRI_SCRIPT_KILL_SENT (1<<12) /* SCRIPT KILL already sent on -BUSY */ #define SRI_SCRIPT_KILL_SENT (1<<12) /* SCRIPT KILL already sent on -BUSY */
#define SRI_MASTER_REBOOT (1<<13) /* Master was detected as rebooting */ #define SRI_MASTER_REBOOT (1<<13) /* Master was detected as rebooting */
/* Note: when adding new flags, please check the flags section in addReplySentinelRedisInstance. */
/* Note: times are in milliseconds. */ /* Note: times are in milliseconds. */
#define SENTINEL_PING_PERIOD 1000 #define SENTINEL_PING_PERIOD 1000
...@@ -2272,12 +2273,8 @@ void rewriteConfigSentinelOption(struct rewriteConfigState *state) { ...@@ -2272,12 +2273,8 @@ void rewriteConfigSentinelOption(struct rewriteConfigState *state) {
/* This function uses the config rewriting Redis engine in order to persist /* This function uses the config rewriting Redis engine in order to persist
* the state of the Sentinel in the current configuration file. * the state of the Sentinel in the current configuration file.
* *
* Before returning the function calls fsync() against the generated
* configuration file to make sure changes are committed to disk.
*
* On failure the function logs a warning on the Redis log. */ * On failure the function logs a warning on the Redis log. */
int sentinelFlushConfig(void) { int sentinelFlushConfig(void) {
int fd = -1;
int saved_hz = server.hz; int saved_hz = server.hz;
int rewrite_status; int rewrite_status;
...@@ -2285,17 +2282,13 @@ int sentinelFlushConfig(void) { ...@@ -2285,17 +2282,13 @@ int sentinelFlushConfig(void) {
rewrite_status = rewriteConfig(server.configfile, 0); rewrite_status = rewriteConfig(server.configfile, 0);
server.hz = saved_hz; server.hz = saved_hz;
if (rewrite_status == -1) goto werr; if (rewrite_status == -1) {
if ((fd = open(server.configfile,O_RDONLY)) == -1) goto werr;
if (fsync(fd) == -1) goto werr;
if (close(fd) == EOF) goto werr;
serverLog(LL_NOTICE,"Sentinel new configuration saved on disk");
return C_OK;
werr:
serverLog(LL_WARNING,"WARNING: Sentinel was not able to save the new configuration on disk!!!: %s", strerror(errno)); serverLog(LL_WARNING,"WARNING: Sentinel was not able to save the new configuration on disk!!!: %s", strerror(errno));
if (fd != -1) close(fd);
return C_ERR; return C_ERR;
} else {
serverLog(LL_NOTICE,"Sentinel new configuration saved on disk");
return C_OK;
}
} }
/* Call sentinelFlushConfig() produce a success/error reply to the /* Call sentinelFlushConfig() produce a success/error reply to the
...@@ -2776,7 +2769,9 @@ void sentinelInfoReplyCallback(redisAsyncContext *c, void *reply, void *privdata ...@@ -2776,7 +2769,9 @@ void sentinelInfoReplyCallback(redisAsyncContext *c, void *reply, void *privdata
link->pending_commands--; link->pending_commands--;
r = reply; r = reply;
if (r->type == REDIS_REPLY_STRING) /* INFO reply type is verbatim in resp3. Normally, sentinel will not use
* resp3 but this is required for testing (see logreqres.c). */
if (r->type == REDIS_REPLY_STRING || r->type == REDIS_REPLY_VERB)
sentinelRefreshInstanceInfo(ri,r->str); sentinelRefreshInstanceInfo(ri,r->str);
} }
...@@ -2987,8 +2982,10 @@ void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privd ...@@ -2987,8 +2982,10 @@ void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privd
ri->link->pc_last_activity = mstime(); ri->link->pc_last_activity = mstime();
/* Sanity check in the reply we expect, so that the code that follows /* Sanity check in the reply we expect, so that the code that follows
* can avoid to check for details. */ * can avoid to check for details.
if (r->type != REDIS_REPLY_ARRAY || * Note: Reply type is PUSH in resp3. Normally, sentinel will not use
* resp3 but this is required for testing (see logreqres.c). */
if ((r->type != REDIS_REPLY_ARRAY && r->type != REDIS_REPLY_PUSH) ||
r->elements != 3 || r->elements != 3 ||
r->element[0]->type != REDIS_REPLY_STRING || r->element[0]->type != REDIS_REPLY_STRING ||
r->element[1]->type != REDIS_REPLY_STRING || r->element[1]->type != REDIS_REPLY_STRING ||
...@@ -3352,6 +3349,7 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) { ...@@ -3352,6 +3349,7 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) {
if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,"); if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,");
if (ri->flags & SRI_FORCE_FAILOVER) flags = sdscat(flags,"force_failover,"); if (ri->flags & SRI_FORCE_FAILOVER) flags = sdscat(flags,"force_failover,");
if (ri->flags & SRI_SCRIPT_KILL_SENT) flags = sdscat(flags,"script_kill_sent,"); if (ri->flags & SRI_SCRIPT_KILL_SENT) flags = sdscat(flags,"script_kill_sent,");
if (ri->flags & SRI_MASTER_REBOOT) flags = sdscat(flags,"master_reboot,");
if (sdslen(flags) != 0) sdsrange(flags,0,-2); /* remove last "," */ if (sdslen(flags) != 0) sdsrange(flags,0,-2); /* remove last "," */
addReplyBulkCString(c,flags); addReplyBulkCString(c,flags);
......
...@@ -216,7 +216,7 @@ mstime_t commandTimeSnapshot(void) { ...@@ -216,7 +216,7 @@ mstime_t commandTimeSnapshot(void) {
* may re-open the same key multiple times, can invalidate an already * may re-open the same key multiple times, can invalidate an already
* open object in a next call, if the next call will see the key expired, * open object in a next call, if the next call will see the key expired,
* while the first did not. * while the first did not.
* This is specificlally important in the context of scripts, where we * This is specifically important in the context of scripts, where we
* pretend that time freezes. This way a key can expire only the first time * pretend that time freezes. This way a key can expire only the first time
* it is accessed and not in the middle of the script execution, making * it is accessed and not in the middle of the script execution, making
* propagation to slaves / AOF consistent. See issue #1525 for more info. * propagation to slaves / AOF consistent. See issue #1525 for more info.
...@@ -1154,6 +1154,26 @@ void updateCachedTime(int update_daylight_info) { ...@@ -1154,6 +1154,26 @@ void updateCachedTime(int update_daylight_info) {
updateCachedTimeWithUs(update_daylight_info, us); updateCachedTimeWithUs(update_daylight_info, us);
} }
/* Performing required operations in order to enter an execution unit.
* In general, if we are already inside an execution unit then there is nothing to do,
* otherwise we need to update cache times so the same cached time will be used all over
* the execution unit.
* update_cached_time - if 0, will not update the cached time even if required.
* us - if not zero, use this time for cached time, otherwise get current time. */
void enterExecutionUnit(int update_cached_time, long long us) {
if (server.execution_nesting++ == 0 && update_cached_time) {
if (us == 0) {
us = ustime();
}
updateCachedTimeWithUs(0, us);
server.cmd_time_snapshot = server.mstime;
}
}
void exitExecutionUnit() {
--server.execution_nesting;
}
void checkChildrenDone(void) { void checkChildrenDone(void) {
int statloc = 0; int statloc = 0;
pid_t pid; pid_t pid;
...@@ -1663,7 +1683,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -1663,7 +1683,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST); activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
/* Unblock all the clients blocked for synchronous replication /* Unblock all the clients blocked for synchronous replication
* in WAIT. */ * in WAIT or WAITAOF. */
if (listLength(server.clients_waiting_acks)) if (listLength(server.clients_waiting_acks))
processClientsWaitingReplicas(); processClientsWaitingReplicas();
...@@ -1723,6 +1743,15 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -1723,6 +1743,15 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
if (server.aof_state == AOF_ON || server.aof_state == AOF_WAIT_REWRITE) if (server.aof_state == AOF_ON || server.aof_state == AOF_WAIT_REWRITE)
flushAppendOnlyFile(0); flushAppendOnlyFile(0);
/* Update the fsynced replica offset.
* If an initial rewrite is in progress then not all data is guaranteed to have actually been
* persisted to disk yet, so we cannot update the field. We will wait for the rewrite to complete. */
if (server.aof_state == AOF_ON && server.fsynced_reploff != -1) {
long long fsynced_reploff_pending;
atomicGet(server.fsynced_reploff_pending, fsynced_reploff_pending);
server.fsynced_reploff = fsynced_reploff_pending;
}
/* Handle writes with pending output buffers. */ /* Handle writes with pending output buffers. */
handleClientsWithPendingWritesUsingThreads(); handleClientsWithPendingWritesUsingThreads();
...@@ -2051,6 +2080,7 @@ void initServerConfig(void) { ...@@ -2051,6 +2080,7 @@ void initServerConfig(void) {
server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT; server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT;
server.repl_down_since = 0; /* Never connected, repl is down since EVER. */ server.repl_down_since = 0; /* Never connected, repl is down since EVER. */
server.master_repl_offset = 0; server.master_repl_offset = 0;
server.fsynced_reploff_pending = 0;
/* Replication partial resync backlog */ /* Replication partial resync backlog */
server.repl_backlog = NULL; server.repl_backlog = NULL;
...@@ -2529,6 +2559,7 @@ void initServer(void) { ...@@ -2529,6 +2559,7 @@ void initServer(void) {
/* Initialization after setting defaults from the config system. */ /* Initialization after setting defaults from the config system. */
server.aof_state = server.aof_enabled ? AOF_ON : AOF_OFF; server.aof_state = server.aof_enabled ? AOF_ON : AOF_OFF;
server.fsynced_reploff = server.aof_enabled ? 0 : -1;
server.hz = server.config_hz; server.hz = server.config_hz;
server.pid = getpid(); server.pid = getpid();
server.in_fork_child = CHILD_TYPE_NONE; server.in_fork_child = CHILD_TYPE_NONE;
...@@ -3479,7 +3510,7 @@ void call(client *c, int flags) { ...@@ -3479,7 +3510,7 @@ void call(client *c, int flags) {
* and a client which is reprocessing command again (after being unblocked). * and a client which is reprocessing command again (after being unblocked).
* Blocked clients can be blocked in different places and not always it means the call() function has been * Blocked clients can be blocked in different places and not always it means the call() function has been
* called. For example this is required for avoiding double logging to monitors.*/ * called. For example this is required for avoiding double logging to monitors.*/
int reprocessing_command = ((!server.execution_nesting) && (c->flags & CLIENT_EXECUTING_COMMAND)) ? 1 : 0; int reprocessing_command = flags & CMD_CALL_REPROCESSING;
/* Initialization: clear the flags that must be set by the command on /* Initialization: clear the flags that must be set by the command on
* demand, and initialize the array for additional commands propagation. */ * demand, and initialize the array for additional commands propagation. */
...@@ -3494,17 +3525,17 @@ void call(client *c, int flags) { ...@@ -3494,17 +3525,17 @@ void call(client *c, int flags) {
/* Call the command. */ /* Call the command. */
dirty = server.dirty; dirty = server.dirty;
long long old_master_repl_offset = server.master_repl_offset;
incrCommandStatsOnError(NULL, 0); incrCommandStatsOnError(NULL, 0);
const long long call_timer = ustime(); const long long call_timer = ustime();
enterExecutionUnit(1, call_timer);
/* Update cache time, and indicate we are starting command execution. /* setting the CLIENT_EXECUTING_COMMAND flag so we will avoid
* in case we have nested calls we want to update only on the first call */ * sending client side caching message in the middle of a command reply.
if (server.execution_nesting++ == 0) { * In case of blocking commands, the flag will be un-set only after successfully
updateCachedTimeWithUs(0,call_timer); * re-processing and unblock the client.*/
server.cmd_time_snapshot = server.mstime;
c->flags |= CLIENT_EXECUTING_COMMAND; c->flags |= CLIENT_EXECUTING_COMMAND;
}
monotime monotonic_start = 0; monotime monotonic_start = 0;
if (monotonicGetType() == MONOTONIC_CLOCK_HW) if (monotonicGetType() == MONOTONIC_CLOCK_HW)
...@@ -3512,12 +3543,11 @@ void call(client *c, int flags) { ...@@ -3512,12 +3543,11 @@ void call(client *c, int flags) {
c->cmd->proc(c); c->cmd->proc(c);
if (--server.execution_nesting == 0) { exitExecutionUnit();
/* In case client is blocked after trying to execute the command, /* In case client is blocked after trying to execute the command,
* it means the execution is not yet completed and we MIGHT reprocess the command in the future. */ * it means the execution is not yet completed and we MIGHT reprocess the command in the future. */
if (!(c->flags & CLIENT_BLOCKED)) if (!(c->flags & CLIENT_BLOCKED)) c->flags &= ~(CLIENT_EXECUTING_COMMAND);
c->flags &= ~(CLIENT_EXECUTING_COMMAND);
}
/* In order to avoid performance implication due to querying the clock using a system call 3 times, /* In order to avoid performance implication due to querying the clock using a system call 3 times,
* we use a monotonic clock, when we are sure its cost is very low, and fall back to non-monotonic call otherwise. */ * we use a monotonic clock, when we are sure its cost is very low, and fall back to non-monotonic call otherwise. */
...@@ -3615,9 +3645,11 @@ void call(client *c, int flags) { ...@@ -3615,9 +3645,11 @@ void call(client *c, int flags) {
* implementation called preventCommandPropagation() or similar, * implementation called preventCommandPropagation() or similar,
* or if we don't have the call() flags to do so. */ * or if we don't have the call() flags to do so. */
if (c->flags & CLIENT_PREVENT_REPL_PROP || if (c->flags & CLIENT_PREVENT_REPL_PROP ||
c->flags & CLIENT_MODULE_PREVENT_REPL_PROP ||
!(flags & CMD_CALL_PROPAGATE_REPL)) !(flags & CMD_CALL_PROPAGATE_REPL))
propagate_flags &= ~PROPAGATE_REPL; propagate_flags &= ~PROPAGATE_REPL;
if (c->flags & CLIENT_PREVENT_AOF_PROP || if (c->flags & CLIENT_PREVENT_AOF_PROP ||
c->flags & CLIENT_MODULE_PREVENT_AOF_PROP ||
!(flags & CMD_CALL_PROPAGATE_AOF)) !(flags & CMD_CALL_PROPAGATE_AOF))
propagate_flags &= ~PROPAGATE_AOF; propagate_flags &= ~PROPAGATE_AOF;
...@@ -3662,6 +3694,11 @@ void call(client *c, int flags) { ...@@ -3662,6 +3694,11 @@ void call(client *c, int flags) {
/* Do some maintenance job and cleanup */ /* Do some maintenance job and cleanup */
afterCommand(c); afterCommand(c);
/* Remember the replication offset of the client, right after its last
* command that resulted in propagation. */
if (old_master_repl_offset != server.master_repl_offset)
c->woff = server.master_repl_offset;
/* Client pause takes effect after a transaction has finished. This needs /* Client pause takes effect after a transaction has finished. This needs
* to be located after everything is propagated. */ * to be located after everything is propagated. */
if (!server.in_exec && server.client_pause_in_transaction) { if (!server.in_exec && server.client_pause_in_transaction) {
...@@ -3802,12 +3839,16 @@ int processCommand(client *c) { ...@@ -3802,12 +3839,16 @@ int processCommand(client *c) {
serverAssert(!scriptIsRunning()); serverAssert(!scriptIsRunning());
} }
moduleCallCommandFilters(c);
/* in case we are starting to ProcessCommand and we already have a command we assume /* in case we are starting to ProcessCommand and we already have a command we assume
* this is a reprocessing of this command, so we do not want to perform some of the actions again. */ * this is a reprocessing of this command, so we do not want to perform some of the actions again. */
int client_reprocessing_command = c->cmd ? 1 : 0; int client_reprocessing_command = c->cmd ? 1 : 0;
/* only run command filter if not reprocessing command */
if (!client_reprocessing_command) {
moduleCallCommandFilters(c);
reqresAppendRequest(c);
}
/* Handle possible security attacks. */ /* Handle possible security attacks. */
if (!strcasecmp(c->argv[0]->ptr,"host:") || !strcasecmp(c->argv[0]->ptr,"post")) { if (!strcasecmp(c->argv[0]->ptr,"host:") || !strcasecmp(c->argv[0]->ptr,"post")) {
securityWarningCommand(c); securityWarningCommand(c);
...@@ -4120,8 +4161,9 @@ int processCommand(client *c) { ...@@ -4120,8 +4161,9 @@ int processCommand(client *c) {
queueMultiCommand(c, cmd_flags); queueMultiCommand(c, cmd_flags);
addReply(c,shared.queued); addReply(c,shared.queued);
} else { } else {
call(c,CMD_CALL_FULL); int flags = CMD_CALL_FULL;
c->woff = server.master_repl_offset; if (client_reprocessing_command) flags |= CMD_CALL_REPROCESSING;
call(c,flags);
if (listLength(server.ready_keys)) if (listLength(server.ready_keys))
handleClientsBlockedOnKeys(); handleClientsBlockedOnKeys();
} }
...@@ -4665,30 +4707,41 @@ void addReplyCommandArgList(client *c, struct redisCommandArg *args, int num_arg ...@@ -4665,30 +4707,41 @@ void addReplyCommandArgList(client *c, struct redisCommandArg *args, int num_arg
} }
} }
/* Must match redisCommandRESP2Type */ #ifdef LOG_REQ_RES
const char *RESP2_TYPE_STR[] = {
"simple-string",
"error",
"integer",
"bulk-string",
"null-bulk-string",
"array",
"null-array",
};
/* Must match redisCommandRESP3Type */ void addReplyJson(client *c, struct jsonObject *rs) {
const char *RESP3_TYPE_STR[] = { addReplyMapLen(c, rs->length);
"simple-string",
"error", for (int i = 0; i < rs->length; i++) {
"integer", struct jsonObjectElement *curr = &rs->elements[i];
"double", addReplyBulkCString(c, curr->key);
"bulk-string", switch (curr->type) {
"array", case (JSON_TYPE_BOOLEAN):
"map", addReplyBool(c, curr->value.boolean);
"set", break;
"bool", case (JSON_TYPE_INTEGER):
"null", addReplyLongLong(c, curr->value.integer);
}; break;
case (JSON_TYPE_STRING):
addReplyBulkCString(c, curr->value.string);
break;
case (JSON_TYPE_OBJECT):
addReplyJson(c, curr->value.object);
break;
case (JSON_TYPE_ARRAY):
addReplyArrayLen(c, curr->value.array.length);
for (int k = 0; k < curr->value.array.length; k++) {
struct jsonObject *object = curr->value.array.objects[k];
addReplyJson(c, object);
}
break;
default:
serverPanic("Invalid JSON type %d", curr->type);
}
}
}
#endif
void addReplyCommandHistory(client *c, struct redisCommand *cmd) { void addReplyCommandHistory(client *c, struct redisCommand *cmd) {
addReplySetLen(c, cmd->num_history); addReplySetLen(c, cmd->num_history);
...@@ -4886,6 +4939,9 @@ void addReplyCommandDocs(client *c, struct redisCommand *cmd) { ...@@ -4886,6 +4939,9 @@ void addReplyCommandDocs(client *c, struct redisCommand *cmd) {
if (cmd->deprecated_since) maplen++; if (cmd->deprecated_since) maplen++;
if (cmd->replaced_by) maplen++; if (cmd->replaced_by) maplen++;
if (cmd->history) maplen++; if (cmd->history) maplen++;
#ifdef LOG_REQ_RES
if (cmd->reply_schema) maplen++;
#endif
if (cmd->args) maplen++; if (cmd->args) maplen++;
if (cmd->subcommands_dict) maplen++; if (cmd->subcommands_dict) maplen++;
addReplyMapLen(c, maplen); addReplyMapLen(c, maplen);
...@@ -4927,6 +4983,12 @@ void addReplyCommandDocs(client *c, struct redisCommand *cmd) { ...@@ -4927,6 +4983,12 @@ void addReplyCommandDocs(client *c, struct redisCommand *cmd) {
addReplyBulkCString(c, "history"); addReplyBulkCString(c, "history");
addReplyCommandHistory(c, cmd); addReplyCommandHistory(c, cmd);
} }
#ifdef LOG_REQ_RES
if (cmd->reply_schema) {
addReplyBulkCString(c, "reply_schema");
addReplyJson(c, cmd->reply_schema);
}
#endif
if (cmd->args) { if (cmd->args) {
addReplyBulkCString(c, "arguments"); addReplyBulkCString(c, "arguments");
addReplyCommandArgList(c, cmd->args, cmd->num_args); addReplyCommandArgList(c, cmd->args, cmd->num_args);
......
...@@ -395,6 +395,11 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT]; ...@@ -395,6 +395,11 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
#define CLIENT_ALLOW_OOM (1ULL<<44) /* Client used by RM_Call is allowed to fully execute #define CLIENT_ALLOW_OOM (1ULL<<44) /* Client used by RM_Call is allowed to fully execute
scripts even when in OOM */ scripts even when in OOM */
#define CLIENT_NO_TOUCH (1ULL<<45) /* This client will not touch LFU/LRU stats. */ #define CLIENT_NO_TOUCH (1ULL<<45) /* This client will not touch LFU/LRU stats. */
#define CLIENT_PUSHING (1ULL<<46) /* This client is pushing notifications. */
#define CLIENT_MODULE_AUTH_HAS_RESULT (1ULL<<47) /* Indicates a client in the middle of module based
auth had been authenticated from the Module. */
#define CLIENT_MODULE_PREVENT_AOF_PROP (1ULL<<48) /* Module client do not want to propagate to AOF */
#define CLIENT_MODULE_PREVENT_REPL_PROP (1ULL<<49) /* Module client do not want to propagate to replica */
/* Client block type (btype field in client structure) /* Client block type (btype field in client structure)
* if CLIENT_BLOCKED flag is set. */ * if CLIENT_BLOCKED flag is set. */
...@@ -402,6 +407,7 @@ typedef enum blocking_type { ...@@ -402,6 +407,7 @@ typedef enum blocking_type {
BLOCKED_NONE, /* Not blocked, no CLIENT_BLOCKED flag set. */ BLOCKED_NONE, /* Not blocked, no CLIENT_BLOCKED flag set. */
BLOCKED_LIST, /* BLPOP & co. */ BLOCKED_LIST, /* BLPOP & co. */
BLOCKED_WAIT, /* WAIT for synchronous replication. */ BLOCKED_WAIT, /* WAIT for synchronous replication. */
BLOCKED_WAITAOF, /* WAITAOF for AOF file fsync. */
BLOCKED_MODULE, /* Blocked by a loadable module. */ BLOCKED_MODULE, /* Blocked by a loadable module. */
BLOCKED_STREAM, /* XREAD. */ BLOCKED_STREAM, /* XREAD. */
BLOCKED_ZSET, /* BZPOP et al. */ BLOCKED_ZSET, /* BZPOP et al. */
...@@ -584,6 +590,7 @@ typedef enum { ...@@ -584,6 +590,7 @@ typedef enum {
#define CMD_CALL_NONE 0 #define CMD_CALL_NONE 0
#define CMD_CALL_PROPAGATE_AOF (1<<0) #define CMD_CALL_PROPAGATE_AOF (1<<0)
#define CMD_CALL_PROPAGATE_REPL (1<<1) #define CMD_CALL_PROPAGATE_REPL (1<<1)
#define CMD_CALL_REPROCESSING (1<<2)
#define CMD_CALL_PROPAGATE (CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL) #define CMD_CALL_PROPAGATE (CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL)
#define CMD_CALL_FULL (CMD_CALL_PROPAGATE) #define CMD_CALL_FULL (CMD_CALL_PROPAGATE)
#define CMD_CALL_FROM_MODULE (1<<2) /* From RM_Call */ #define CMD_CALL_FROM_MODULE (1<<2) /* From RM_Call */
...@@ -742,6 +749,7 @@ typedef void (*moduleTypeFreeFunc2)(struct RedisModuleKeyOptCtx *ctx, void *valu ...@@ -742,6 +749,7 @@ typedef void (*moduleTypeFreeFunc2)(struct RedisModuleKeyOptCtx *ctx, void *valu
typedef size_t (*moduleTypeFreeEffortFunc2)(struct RedisModuleKeyOptCtx *ctx, const void *value); typedef size_t (*moduleTypeFreeEffortFunc2)(struct RedisModuleKeyOptCtx *ctx, const void *value);
typedef void (*moduleTypeUnlinkFunc2)(struct RedisModuleKeyOptCtx *ctx, void *value); typedef void (*moduleTypeUnlinkFunc2)(struct RedisModuleKeyOptCtx *ctx, void *value);
typedef void *(*moduleTypeCopyFunc2)(struct RedisModuleKeyOptCtx *ctx, const void *value); typedef void *(*moduleTypeCopyFunc2)(struct RedisModuleKeyOptCtx *ctx, const void *value);
typedef int (*moduleTypeAuthCallback)(struct RedisModuleCtx *ctx, void *username, void *password, const char **err);
/* The module type, which is referenced in each value of a given type, defines /* The module type, which is referenced in each value of a given type, defines
...@@ -809,6 +817,8 @@ struct RedisModule { ...@@ -809,6 +817,8 @@ struct RedisModule {
RedisModuleInfoFunc info_cb; /* Callback for module to add INFO fields. */ RedisModuleInfoFunc info_cb; /* Callback for module to add INFO fields. */
RedisModuleDefragFunc defrag_cb; /* Callback for global data defrag. */ RedisModuleDefragFunc defrag_cb; /* Callback for global data defrag. */
struct moduleLoadQueueEntry *loadmod; /* Module load arguments for config rewrite. */ struct moduleLoadQueueEntry *loadmod; /* Module load arguments for config rewrite. */
int num_commands_with_acl_categories; /* Number of commands in this module included in acl categories */
int onload; /* Flag to identify if the call is being made from Onload (0 or 1) */
}; };
typedef struct RedisModule RedisModule; typedef struct RedisModule RedisModule;
...@@ -858,6 +868,9 @@ struct RedisModuleDigest { ...@@ -858,6 +868,9 @@ struct RedisModuleDigest {
memset(mdvar.x,0,sizeof(mdvar.x)); \ memset(mdvar.x,0,sizeof(mdvar.x)); \
} while(0) } while(0)
/* Macro to check if the client is in the middle of module based authentication. */
#define clientHasModuleAuthInProgress(c) ((c)->module_auth_ctx != NULL)
/* Objects encoding. Some kind of objects like Strings and Hashes can be /* Objects encoding. Some kind of objects like Strings and Hashes can be
* internally represented in multiple ways. The 'encoding' field of the object * internally represented in multiple ways. The 'encoding' field of the object
* is set to one of this fields for this object. */ * is set to one of this fields for this object. */
...@@ -891,7 +904,7 @@ struct redisObject { ...@@ -891,7 +904,7 @@ struct redisObject {
void *ptr; void *ptr;
}; };
/* The a string name for an object's type as listed above /* The string name for an object's type as listed above
* Native types are checked against the OBJ_STRING, OBJ_LIST, OBJ_* defines, * Native types are checked against the OBJ_STRING, OBJ_LIST, OBJ_* defines,
* and Module types have their registered name returned. */ * and Module types have their registered name returned. */
char *getObjectTypeName(robj*); char *getObjectTypeName(robj*);
...@@ -924,7 +937,7 @@ typedef struct clientReplyBlock { ...@@ -924,7 +937,7 @@ typedef struct clientReplyBlock {
* | / \ * | / \
* | / \ * | / \
* | / \ * | / \
* Repl Backlog Replia_A Replia_B * Repl Backlog Replica_A Replica_B
* *
* Each replica or replication backlog increments only the refcount of the * Each replica or replication backlog increments only the refcount of the
* 'ref_repl_buf_node' which it points to. So when replica walks to the next * 'ref_repl_buf_node' which it points to. So when replica walks to the next
...@@ -1012,14 +1025,19 @@ typedef struct blockingState { ...@@ -1012,14 +1025,19 @@ typedef struct blockingState {
/* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM or any other Keys related blocking */ /* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM or any other Keys related blocking */
dict *keys; /* The keys we are blocked on */ dict *keys; /* The keys we are blocked on */
/* BLOCKED_WAIT */ /* BLOCKED_WAIT and BLOCKED_WAITAOF */
int numreplicas; /* Number of replicas we are waiting for ACK. */ int numreplicas; /* Number of replicas we are waiting for ACK. */
int numlocal; /* Indication if WAITAOF is waiting for local fsync. */
long long reploffset; /* Replication offset to reach. */ long long reploffset; /* Replication offset to reach. */
/* BLOCKED_MODULE */ /* BLOCKED_MODULE */
void *module_blocked_handle; /* RedisModuleBlockedClient structure. void *module_blocked_handle; /* RedisModuleBlockedClient structure.
which is opaque for the Redis core, only which is opaque for the Redis core, only
handled in module.c. */ handled in module.c. */
void *async_rm_call_handle; /* RedisModuleAsyncRMCallPromise structure.
which is opaque for the Redis core, only
handled in module.c. */
} blockingState; } blockingState;
/* The following structure represents a node in the server.ready_keys list, /* The following structure represents a node in the server.ready_keys list,
...@@ -1083,7 +1101,7 @@ typedef struct { ...@@ -1083,7 +1101,7 @@ typedef struct {
need more reserved IDs use UINT64_MAX-1, need more reserved IDs use UINT64_MAX-1,
-2, ... and so forth. */ -2, ... and so forth. */
/* Replication backlog is not separate memory, it just is one consumer of /* Replication backlog is not a separate memory, it just is one consumer of
* the global replication buffer. This structure records the reference of * the global replication buffer. This structure records the reference of
* replication buffers. Since the replication buffer block list may be very long, * replication buffers. Since the replication buffer block list may be very long,
* it would cost much time to search replication offset on partial resync, so * it would cost much time to search replication offset on partial resync, so
...@@ -1106,6 +1124,31 @@ typedef struct { ...@@ -1106,6 +1124,31 @@ typedef struct {
size_t mem_usage_sum; size_t mem_usage_sum;
} clientMemUsageBucket; } clientMemUsageBucket;
#ifdef LOG_REQ_RES
/* Structure used to log client's requests and their
* responses (see logreqres.c) */
typedef struct {
/* General */
int argv_logged; /* 1 if the command was logged */
/* Vars for log buffer */
unsigned char *buf; /* Buffer holding the data (request and response) */
size_t used;
size_t capacity;
/* Vars for offsets within the client's reply */
struct {
/* General */
int saved; /* 1 if we already saved the offset (first time we call addReply*) */
/* Offset within the static reply buffer */
int bufpos;
/* Offset within the reply block list */
struct {
int index;
size_t used;
} last_node;
} offset;
} clientReqResInfo;
#endif
typedef struct client { typedef struct client {
uint64_t id; /* Client incremental unique ID. */ uint64_t id; /* Client incremental unique ID. */
uint64_t flags; /* Client flags: CLIENT_* macros. */ uint64_t flags; /* Client flags: CLIENT_* macros. */
...@@ -1154,6 +1197,7 @@ typedef struct client { ...@@ -1154,6 +1197,7 @@ typedef struct client {
long long reploff; /* Applied replication offset if this is a master. */ long long reploff; /* Applied replication offset if this is a master. */
long long repl_applied; /* Applied replication data count in querybuf, if this is a replica. */ long long repl_applied; /* Applied replication data count in querybuf, if this is a replica. */
long long repl_ack_off; /* Replication ack offset, if this is a slave. */ long long repl_ack_off; /* Replication ack offset, if this is a slave. */
long long repl_aof_off; /* Replication AOF fsync ack offset, if this is a slave. */
long long repl_ack_time;/* Replication ack time, if this is a slave. */ long long repl_ack_time;/* Replication ack time, if this is a slave. */
long long repl_last_partial_write; /* The last time the server did a partial write from the RDB child pipe to this replica */ long long repl_last_partial_write; /* The last time the server did a partial write from the RDB child pipe to this replica */
long long psync_initial_offset; /* FULLRESYNC reply offset other slaves long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
...@@ -1176,6 +1220,12 @@ typedef struct client { ...@@ -1176,6 +1220,12 @@ typedef struct client {
listNode *client_list_node; /* list node in client list */ listNode *client_list_node; /* list node in client list */
listNode *postponed_list_node; /* list node within the postponed list */ listNode *postponed_list_node; /* list node within the postponed list */
listNode *pending_read_list_node; /* list node in clients pending read list */ listNode *pending_read_list_node; /* list node in clients pending read list */
void *module_blocked_client; /* Pointer to the RedisModuleBlockedClient associated with this
* client. This is set in case of module authentication before the
* unblocked client is reprocessed to handle reply callbacks. */
void *module_auth_ctx; /* Ongoing / attempted module based auth callback's ctx.
* This is only tracked within the context of the command attempting
* authentication. If not NULL, it means module auth is in progress. */
RedisModuleUserChangedFunc auth_callback; /* Module callback to execute RedisModuleUserChangedFunc auth_callback; /* Module callback to execute
* when the authenticated user * when the authenticated user
* changes. */ * changes. */
...@@ -1187,7 +1237,7 @@ typedef struct client { ...@@ -1187,7 +1237,7 @@ typedef struct client {
* unloaded for cleanup. Opaque for Redis Core.*/ * unloaded for cleanup. Opaque for Redis Core.*/
/* If this client is in tracking mode and this field is non zero, /* If this client is in tracking mode and this field is non zero,
* invalidation messages for keys fetched by this client will be send to * invalidation messages for keys fetched by this client will be sent to
* the specified client ID. */ * the specified client ID. */
uint64_t client_tracking_redirection; uint64_t client_tracking_redirection;
rax *client_tracking_prefixes; /* A dictionary of prefixes we are already rax *client_tracking_prefixes; /* A dictionary of prefixes we are already
...@@ -1217,6 +1267,9 @@ typedef struct client { ...@@ -1217,6 +1267,9 @@ typedef struct client {
int bufpos; int bufpos;
size_t buf_usable_size; /* Usable size of buffer. */ size_t buf_usable_size; /* Usable size of buffer. */
char *buf; char *buf;
#ifdef LOG_REQ_RES
clientReqResInfo reqres;
#endif
} client; } client;
/* ACL information */ /* ACL information */
...@@ -1545,6 +1598,11 @@ struct redisServer { ...@@ -1545,6 +1598,11 @@ struct redisServer {
client *current_client; /* The client that triggered the command execution (External or AOF). */ client *current_client; /* The client that triggered the command execution (External or AOF). */
client *executing_client; /* The client executing the current command (possibly script or module). */ client *executing_client; /* The client executing the current command (possibly script or module). */
#ifdef LOG_REQ_RES
char *req_res_logfile; /* Path of log file for logging all requests and their replies. If NULL, no logging will be performed */
unsigned int client_default_resp;
#endif
/* Stuff for client mem eviction */ /* Stuff for client mem eviction */
clientMemUsageBucket* client_mem_usage_buckets; clientMemUsageBucket* client_mem_usage_buckets;
...@@ -1742,7 +1800,7 @@ struct redisServer { ...@@ -1742,7 +1800,7 @@ struct redisServer {
char *rdb_pipe_buff; /* In diskless replication, this buffer holds data */ char *rdb_pipe_buff; /* In diskless replication, this buffer holds data */
int rdb_pipe_bufflen; /* that was read from the rdb pipe. */ int rdb_pipe_bufflen; /* that was read from the rdb pipe. */
int rdb_key_save_delay; /* Delay in microseconds between keys while int rdb_key_save_delay; /* Delay in microseconds between keys while
* writing the RDB. (for testings). negative * writing aof or rdb. (for testings). negative
* value means fractions of microseconds (on average). */ * value means fractions of microseconds (on average). */
int key_load_delay; /* Delay in microseconds between keys while int key_load_delay; /* Delay in microseconds between keys while
* loading aof or rdb. (for testings). negative * loading aof or rdb. (for testings). negative
...@@ -1773,6 +1831,11 @@ struct redisServer { ...@@ -1773,6 +1831,11 @@ struct redisServer {
char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/ char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/
long long master_repl_offset; /* My current replication offset */ long long master_repl_offset; /* My current replication offset */
long long second_replid_offset; /* Accept offsets up to this for replid2. */ long long second_replid_offset; /* Accept offsets up to this for replid2. */
redisAtomic long long fsynced_reploff_pending;/* Largest replication offset to
* potentially have been fsynced, applied to
fsynced_reploff only when AOF state is AOF_ON
(not during the initial rewrite) */
long long fsynced_reploff; /* Largest replication offset that has been confirmed to be fsynced */
int slaveseldb; /* Last SELECTed DB in replication output */ int slaveseldb; /* Last SELECTed DB in replication output */
int repl_ping_slave_period; /* Master pings the slave every N seconds */ int repl_ping_slave_period; /* Master pings the slave every N seconds */
replBacklog *repl_backlog; /* Replication backlog for partial syncs */ replBacklog *repl_backlog; /* Replication backlog for partial syncs */
...@@ -1830,7 +1893,7 @@ struct redisServer { ...@@ -1830,7 +1893,7 @@ struct redisServer {
long long master_initial_offset; /* Master PSYNC offset. */ long long master_initial_offset; /* Master PSYNC offset. */
int repl_slave_lazy_flush; /* Lazy FLUSHALL before loading DB? */ int repl_slave_lazy_flush; /* Lazy FLUSHALL before loading DB? */
/* Synchronous replication. */ /* Synchronous replication. */
list *clients_waiting_acks; /* Clients waiting in WAIT command. */ list *clients_waiting_acks; /* Clients waiting in WAIT or WAITAOF. */
int get_ack_from_slaves; /* If true we send REPLCONF GETACK. */ int get_ack_from_slaves; /* If true we send REPLCONF GETACK. */
/* Limits */ /* Limits */
unsigned int maxclients; /* Max number of simultaneous clients */ unsigned int maxclients; /* Max number of simultaneous clients */
...@@ -1915,7 +1978,7 @@ struct redisServer { ...@@ -1915,7 +1978,7 @@ struct redisServer {
REDISMODULE_CLUSTER_FLAG_*. */ REDISMODULE_CLUSTER_FLAG_*. */
int cluster_allow_reads_when_down; /* Are reads allowed when the cluster int cluster_allow_reads_when_down; /* Are reads allowed when the cluster
is down? */ is down? */
int cluster_config_file_lock_fd; /* cluster config fd, will be flock */ int cluster_config_file_lock_fd; /* cluster config fd, will be flocked. */
unsigned long long cluster_link_msg_queue_limit_bytes; /* Memory usage limit on individual link msg queue */ unsigned long long cluster_link_msg_queue_limit_bytes; /* Memory usage limit on individual link msg queue */
int cluster_drop_packet_filter; /* Debug config that allows tactically int cluster_drop_packet_filter; /* Debug config that allows tactically
* dropping packets of a specific type */ * dropping packets of a specific type */
...@@ -2111,30 +2174,38 @@ typedef struct redisCommandArg { ...@@ -2111,30 +2174,38 @@ typedef struct redisCommandArg {
int num_args; int num_args;
} redisCommandArg; } redisCommandArg;
/* Must be synced with RESP2_TYPE_STR and generate-command-code.py */ #ifdef LOG_REQ_RES
typedef enum {
RESP2_SIMPLE_STRING, /* Must be synced with generate-command-code.py */
RESP2_ERROR,
RESP2_INTEGER,
RESP2_BULK_STRING,
RESP2_NULL_BULK_STRING,
RESP2_ARRAY,
RESP2_NULL_ARRAY,
} redisCommandRESP2Type;
/* Must be synced with RESP3_TYPE_STR and generate-command-code.py */
typedef enum { typedef enum {
RESP3_SIMPLE_STRING, JSON_TYPE_STRING,
RESP3_ERROR, JSON_TYPE_INTEGER,
RESP3_INTEGER, JSON_TYPE_BOOLEAN,
RESP3_DOUBLE, JSON_TYPE_OBJECT,
RESP3_BULK_STRING, JSON_TYPE_ARRAY,
RESP3_ARRAY, } jsonType;
RESP3_MAP,
RESP3_SET, typedef struct jsonObjectElement {
RESP3_BOOL, jsonType type;
RESP3_NULL, const char *key;
} redisCommandRESP3Type; union {
const char *string;
long long integer;
int boolean;
struct jsonObject *object;
struct {
struct jsonObject **objects;
int length;
} array;
} value;
} jsonObjectElement;
typedef struct jsonObject {
struct jsonObjectElement *elements;
int length;
} jsonObject;
#endif
/* WARNING! This struct must match RedisModuleCommandHistoryEntry */ /* WARNING! This struct must match RedisModuleCommandHistoryEntry */
typedef struct { typedef struct {
...@@ -2285,6 +2356,10 @@ struct redisCommand { ...@@ -2285,6 +2356,10 @@ struct redisCommand {
struct redisCommand *subcommands; struct redisCommand *subcommands;
/* Array of arguments (may be NULL) */ /* Array of arguments (may be NULL) */
struct redisCommandArg *args; struct redisCommandArg *args;
#ifdef LOG_REQ_RES
/* Reply schema */
struct jsonObject *reply_schema;
#endif
/* Runtime populated data */ /* Runtime populated data */
long long microseconds, calls, rejected_calls, failed_calls; long long microseconds, calls, rejected_calls, failed_calls;
...@@ -2435,7 +2510,7 @@ void moduleInitModulesSystem(void); ...@@ -2435,7 +2510,7 @@ void moduleInitModulesSystem(void);
void moduleInitModulesSystemLast(void); void moduleInitModulesSystemLast(void);
void modulesCron(void); void modulesCron(void);
int moduleLoad(const char *path, void **argv, int argc, int is_loadex); int moduleLoad(const char *path, void **argv, int argc, int is_loadex);
int moduleUnload(sds name); int moduleUnload(sds name, const char **errmsg);
void moduleLoadFromQueue(void); void moduleLoadFromQueue(void);
int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int moduleGetCommandChannelsViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int moduleGetCommandChannelsViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
...@@ -2444,6 +2519,7 @@ void moduleTypeNameByID(char *name, uint64_t moduleid); ...@@ -2444,6 +2519,7 @@ void moduleTypeNameByID(char *name, uint64_t moduleid);
const char *moduleTypeModuleName(moduleType *mt); const char *moduleTypeModuleName(moduleType *mt);
const char *moduleNameFromCommand(struct redisCommand *cmd); const char *moduleNameFromCommand(struct redisCommand *cmd);
void moduleFreeContext(struct RedisModuleCtx *ctx); void moduleFreeContext(struct RedisModuleCtx *ctx);
void moduleCallCommandUnblockedHandler(client *c);
void unblockClientFromModule(client *c); void unblockClientFromModule(client *c);
void moduleHandleBlockedClients(void); void moduleHandleBlockedClients(void);
void moduleBlockedClientTimedOut(client *c); void moduleBlockedClientTimedOut(client *c);
...@@ -2566,7 +2642,7 @@ char *getClientPeerId(client *client); ...@@ -2566,7 +2642,7 @@ char *getClientPeerId(client *client);
char *getClientSockName(client *client); char *getClientSockName(client *client);
sds catClientInfoString(sds s, client *client); sds catClientInfoString(sds s, client *client);
sds getAllClientsInfoString(int type); sds getAllClientsInfoString(int type);
int clientSetName(client *c, robj *name); int clientSetName(client *c, robj *name, const char **err);
void rewriteClientCommandVector(client *c, int argc, ...); void rewriteClientCommandVector(client *c, int argc, ...);
void rewriteClientCommandArgument(client *c, int i, robj *newval); void rewriteClientCommandArgument(client *c, int i, robj *newval);
void replaceClientCommandVector(client *c, int argc, robj **argv); void replaceClientCommandVector(client *c, int argc, robj **argv);
...@@ -2609,6 +2685,12 @@ client *lookupClientByID(uint64_t id); ...@@ -2609,6 +2685,12 @@ client *lookupClientByID(uint64_t id);
int authRequired(client *c); int authRequired(client *c);
void putClientInPendingWriteQueue(client *c); void putClientInPendingWriteQueue(client *c);
/* logreqres.c - logging of requests and responses */
void reqresReset(client *c, int free_buf);
void reqresSaveClientReplyOffset(client *c);
size_t reqresAppendRequest(client *c);
size_t reqresAppendResponse(client *c);
#ifdef __GNUC__ #ifdef __GNUC__
void addReplyErrorFormatEx(client *c, int flags, const char *fmt, ...) void addReplyErrorFormatEx(client *c, int flags, const char *fmt, ...)
__attribute__((format(printf, 3, 4))); __attribute__((format(printf, 3, 4)));
...@@ -2759,6 +2841,7 @@ int checkGoodReplicasStatus(void); ...@@ -2759,6 +2841,7 @@ int checkGoodReplicasStatus(void);
void processClientsWaitingReplicas(void); void processClientsWaitingReplicas(void);
void unblockClientWaitingReplicas(client *c); void unblockClientWaitingReplicas(client *c);
int replicationCountAcksByOffset(long long offset); int replicationCountAcksByOffset(long long offset);
int replicationCountAOFAcksByOffset(long long offset);
void replicationSendNewlineToMaster(void); void replicationSendNewlineToMaster(void);
long long replicationGetSlaveOffset(void); long long replicationGetSlaveOffset(void);
char *replicationGetSlaveName(client *c); char *replicationGetSlaveName(client *c);
...@@ -2856,8 +2939,18 @@ void ACLInit(void); ...@@ -2856,8 +2939,18 @@ void ACLInit(void);
#define ACL_WRITE_PERMISSION (1<<1) #define ACL_WRITE_PERMISSION (1<<1)
#define ACL_ALL_PERMISSION (ACL_READ_PERMISSION|ACL_WRITE_PERMISSION) #define ACL_ALL_PERMISSION (ACL_READ_PERMISSION|ACL_WRITE_PERMISSION)
/* Return codes for Authentication functions to indicate the result. */
typedef enum {
AUTH_OK = 0,
AUTH_ERR,
AUTH_NOT_HANDLED,
AUTH_BLOCKED
} AuthResult;
int ACLCheckUserCredentials(robj *username, robj *password); int ACLCheckUserCredentials(robj *username, robj *password);
int ACLAuthenticateUser(client *c, robj *username, robj *password); int ACLAuthenticateUser(client *c, robj *username, robj *password, robj **err);
int checkModuleAuthentication(client *c, robj *username, robj *password, robj **err);
void addAuthErrReply(client *c, robj *err);
unsigned long ACLGetCommandID(sds cmdname); unsigned long ACLGetCommandID(sds cmdname);
void ACLClearCommandID(void); void ACLClearCommandID(void);
user *ACLGetUserByName(const char *name, size_t namelen); user *ACLGetUserByName(const char *name, size_t namelen);
...@@ -2881,6 +2974,7 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username ...@@ -2881,6 +2974,7 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username
sds getAclErrorMessage(int acl_res, user *user, struct redisCommand *cmd, sds errored_val, int verbose); sds getAclErrorMessage(int acl_res, user *user, struct redisCommand *cmd, sds errored_val, int verbose);
void ACLUpdateDefaultUserPassword(sds password); void ACLUpdateDefaultUserPassword(sds password);
sds genRedisInfoStringACLStats(sds info); sds genRedisInfoStringACLStats(sds info);
void ACLRecomputeCommandBitsFromCommandRulesAllUsers();
/* Sorted sets data type */ /* Sorted sets data type */
...@@ -3008,6 +3102,8 @@ void adjustOpenFilesLimit(void); ...@@ -3008,6 +3102,8 @@ void adjustOpenFilesLimit(void);
void incrementErrorCount(const char *fullerr, size_t namelen); void incrementErrorCount(const char *fullerr, size_t namelen);
void closeListeningSockets(int unlink_unix_socket); void closeListeningSockets(int unlink_unix_socket);
void updateCachedTime(int update_daylight_info); void updateCachedTime(int update_daylight_info);
void enterExecutionUnit(int update_cached_time, long long us);
void exitExecutionUnit();
void resetServerStats(void); void resetServerStats(void);
void activeDefragCycle(void); void activeDefragCycle(void);
unsigned int getLRUClock(void); unsigned int getLRUClock(void);
...@@ -3314,7 +3410,7 @@ typedef struct luaScript { ...@@ -3314,7 +3410,7 @@ typedef struct luaScript {
void processUnblockedClients(void); void processUnblockedClients(void);
void initClientBlockingState(client *c); void initClientBlockingState(client *c);
void blockClient(client *c, int btype); void blockClient(client *c, int btype);
void unblockClient(client *c); void unblockClient(client *c, int queue_for_reprocessing);
void unblockClientOnTimeout(client *c); void unblockClientOnTimeout(client *c);
void unblockClientOnError(client *c, const char *err_str); void unblockClientOnError(client *c, const char *err_str);
void queueClientForReprocessing(client *c); void queueClientForReprocessing(client *c);
...@@ -3327,6 +3423,7 @@ void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeo ...@@ -3327,6 +3423,7 @@ void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeo
void blockClientShutdown(client *c); void blockClientShutdown(client *c);
void blockPostponeClient(client *c); void blockPostponeClient(client *c);
void blockForReplication(client *c, mstime_t timeout, long long offset, long numreplicas); void blockForReplication(client *c, mstime_t timeout, long long offset, long numreplicas);
void blockForAofFsync(client *c, mstime_t timeout, long long offset, int numlocal, long numreplicas);
void signalDeletedKeyAsReady(redisDb *db, robj *key, int type); void signalDeletedKeyAsReady(redisDb *db, robj *key, int type);
void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_errors); void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_errors);
void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with); void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with);
...@@ -3591,6 +3688,7 @@ void bitcountCommand(client *c); ...@@ -3591,6 +3688,7 @@ void bitcountCommand(client *c);
void bitposCommand(client *c); void bitposCommand(client *c);
void replconfCommand(client *c); void replconfCommand(client *c);
void waitCommand(client *c); void waitCommand(client *c);
void waitaofCommand(client *c);
void georadiusbymemberCommand(client *c); void georadiusbymemberCommand(client *c);
void georadiusbymemberroCommand(client *c); void georadiusbymemberroCommand(client *c);
void georadiusCommand(client *c); void georadiusCommand(client *c);
......
...@@ -57,7 +57,10 @@ static int label_margin_top = 1; ...@@ -57,7 +57,10 @@ static int label_margin_top = 1;
struct sequence *createSparklineSequence(void) { struct sequence *createSparklineSequence(void) {
struct sequence *seq = zmalloc(sizeof(*seq)); struct sequence *seq = zmalloc(sizeof(*seq));
seq->length = 0; seq->length = 0;
seq->labels = 0;
seq->samples = NULL; seq->samples = NULL;
seq->min = 0.0f;
seq->max = 0.0f;
return seq; return seq;
} }
......
...@@ -526,7 +526,7 @@ void lpushxCommand(client *c) { ...@@ -526,7 +526,7 @@ void lpushxCommand(client *c) {
pushGenericCommand(c,LIST_HEAD,1); pushGenericCommand(c,LIST_HEAD,1);
} }
/* RPUSH <key> <element> [<element> ...] */ /* RPUSHX <key> <element> [<element> ...] */
void rpushxCommand(client *c) { void rpushxCommand(client *c) {
pushGenericCommand(c,LIST_TAIL,1); pushGenericCommand(c,LIST_TAIL,1);
} }
......
...@@ -2208,9 +2208,9 @@ void xreadCommand(client *c) { ...@@ -2208,9 +2208,9 @@ void xreadCommand(client *c) {
streams_arg = i+1; streams_arg = i+1;
streams_count = (c->argc-streams_arg); streams_count = (c->argc-streams_arg);
if ((streams_count % 2) != 0) { if ((streams_count % 2) != 0) {
addReplyError(c,"Unbalanced XREAD list of streams: " addReplyErrorFormat(c,"Unbalanced '%s' list of streams: "
"for each stream key an ID or '$' must be " "for each stream key an ID or '>' must be "
"specified."); "specified.", c->cmd->fullname);
return; return;
} }
streams_count /= 2; /* We have two arguments for each stream. */ streams_count /= 2; /* We have two arguments for each stream. */
......
...@@ -559,7 +559,6 @@ void mgetCommand(client *c) { ...@@ -559,7 +559,6 @@ void mgetCommand(client *c) {
void msetGenericCommand(client *c, int nx) { void msetGenericCommand(client *c, int nx) {
int j; int j;
int setkey_flags = 0;
if ((c->argc % 2) == 0) { if ((c->argc % 2) == 0) {
addReplyErrorArity(c); addReplyErrorArity(c);
...@@ -575,12 +574,11 @@ void msetGenericCommand(client *c, int nx) { ...@@ -575,12 +574,11 @@ void msetGenericCommand(client *c, int nx) {
return; return;
} }
} }
setkey_flags |= SETKEY_DOESNT_EXIST;
} }
for (j = 1; j < c->argc; j += 2) { for (j = 1; j < c->argc; j += 2) {
c->argv[j+1] = tryObjectEncoding(c->argv[j+1]); c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
setKey(c, c->db, c->argv[j], c->argv[j + 1], setkey_flags); setKey(c, c->db, c->argv[j], c->argv[j + 1], 0);
notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[j],c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[j],c->db->id);
} }
server.dirty += (c->argc-1)/2; server.dirty += (c->argc-1)/2;
......
...@@ -266,6 +266,9 @@ void trackingRememberKeys(client *tracking, client *executing) { ...@@ -266,6 +266,9 @@ void trackingRememberKeys(client *tracking, client *executing) {
* - Following a flush command, to send a single RESP NULL to indicate * - Following a flush command, to send a single RESP NULL to indicate
* that all keys are now invalid. */ * that all keys are now invalid. */
void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
uint64_t old_flags = c->flags;
c->flags |= CLIENT_PUSHING;
int using_redirection = 0; int using_redirection = 0;
if (c->client_tracking_redirection) { if (c->client_tracking_redirection) {
client *redir = lookupClientByID(c->client_tracking_redirection); client *redir = lookupClientByID(c->client_tracking_redirection);
...@@ -279,10 +282,14 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { ...@@ -279,10 +282,14 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
addReplyBulkCBuffer(c,"tracking-redir-broken",21); addReplyBulkCBuffer(c,"tracking-redir-broken",21);
addReplyLongLong(c,c->client_tracking_redirection); addReplyLongLong(c,c->client_tracking_redirection);
} }
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
return; return;
} }
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
c = redir; c = redir;
using_redirection = 1; using_redirection = 1;
old_flags = c->flags;
c->flags |= CLIENT_PUSHING;
} }
/* Only send such info for clients in RESP version 3 or more. However /* Only send such info for clients in RESP version 3 or more. However
...@@ -301,6 +308,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { ...@@ -301,6 +308,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
* redirecting to another client. We can't send anything to * redirecting to another client. We can't send anything to
* it since RESP2 does not support push messages in the same * it since RESP2 does not support push messages in the same
* connection. */ * connection. */
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
return; return;
} }
...@@ -312,6 +320,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { ...@@ -312,6 +320,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
addReplyBulkCBuffer(c,keyname,keylen); addReplyBulkCBuffer(c,keyname,keylen);
} }
updateClientMemUsageAndBucket(c); updateClientMemUsageAndBucket(c);
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
} }
/* This function is called when a key is modified in Redis and in the case /* This function is called when a key is modified in Redis and in the case
......
...@@ -74,3 +74,11 @@ test "CLUSTER RESET SOFT test" { ...@@ -74,3 +74,11 @@ test "CLUSTER RESET SOFT test" {
R 1 CLUSTER RESET SOFT R 1 CLUSTER RESET SOFT
assert {[get_info_field [R 1 cluster info] cluster_current_epoch] eq $last_epoch_node1} assert {[get_info_field [R 1 cluster info] cluster_current_epoch] eq $last_epoch_node1}
} }
test "Coverage: CLUSTER HELP" {
assert_match "*CLUSTER <subcommand> *" [R 0 CLUSTER HELP]
}
test "Coverage: ASKING" {
assert_equal {OK} [R 0 ASKING]
}
...@@ -20,6 +20,12 @@ test "Can't read from replica without READONLY" { ...@@ -20,6 +20,12 @@ test "Can't read from replica without READONLY" {
assert {[string range $err 0 4] eq {MOVED}} assert {[string range $err 0 4] eq {MOVED}}
} }
test "Can't read from replica after READWRITE" {
$replica READWRITE
catch {$replica GET a} err
assert {[string range $err 0 4] eq {MOVED}}
}
test "Can read from replica after READONLY" { test "Can read from replica after READONLY" {
$replica READONLY $replica READONLY
assert {[$replica GET a] eq {1}} assert {[$replica GET a] eq {1}}
......
...@@ -105,6 +105,15 @@ proc spawn_instance {type base_port count {conf {}} {base_conf_file ""}} { ...@@ -105,6 +105,15 @@ proc spawn_instance {type base_port count {conf {}} {base_conf_file ""}} {
} else { } else {
puts $cfg "port $port" puts $cfg "port $port"
} }
if {$::log_req_res} {
puts $cfg "req-res-logfile stdout.reqres"
}
if {$::force_resp3} {
puts $cfg "client-default-resp 3"
}
puts $cfg "repl-diskless-sync-delay 0" puts $cfg "repl-diskless-sync-delay 0"
puts $cfg "dir ./$dirname" puts $cfg "dir ./$dirname"
puts $cfg "logfile log.txt" puts $cfg "logfile log.txt"
...@@ -293,6 +302,10 @@ proc parse_options {} { ...@@ -293,6 +302,10 @@ proc parse_options {} {
set ::stop_on_failure 1 set ::stop_on_failure 1
} elseif {$opt eq {--loop}} { } elseif {$opt eq {--loop}} {
set ::loop 1 set ::loop 1
} elseif {$opt eq {--log-req-res}} {
set ::log_req_res 1
} elseif {$opt eq {--force-resp3}} {
set ::force_resp3 1
} elseif {$opt eq "--help"} { } elseif {$opt eq "--help"} {
puts "--single <pattern> Only runs tests specified by pattern." puts "--single <pattern> Only runs tests specified by pattern."
puts "--dont-clean Keep log files on exit." puts "--dont-clean Keep log files on exit."
......
...@@ -827,7 +827,7 @@ test {corrupt payload: fuzzer findings - set with duplicate elements causes sdif ...@@ -827,7 +827,7 @@ test {corrupt payload: fuzzer findings - set with duplicate elements causes sdif
assert_equal {0 2 4 6 8 _1 _3 _3 _5 _9} [lsort [r smembers _key]] assert_equal {0 2 4 6 8 _1 _3 _3 _5 _9} [lsort [r smembers _key]]
assert_equal {0 2 4 6 8 _1 _3 _5 _9} [lsort [r sdiff _key]] assert_equal {0 2 4 6 8 _1 _3 _5 _9} [lsort [r sdiff _key]]
} }
} } {} {logreqres:skip} ;# This test violates {"uniqueItems": true}
} ;# tags } ;# tags
...@@ -218,6 +218,7 @@ start_server {} { ...@@ -218,6 +218,7 @@ start_server {} {
test {Test RDB load info} { test {Test RDB load info} {
r debug populate 1000 r debug populate 1000
r save r save
assert {[r lastsave] <= [lindex [r time] 0]}
restart_server 0 true false restart_server 0 true false
wait_done_loading r wait_done_loading r
assert {[s rdb_last_load_keys_expired] == 0} assert {[s rdb_last_load_keys_expired] == 0}
......
...@@ -25,7 +25,7 @@ proc default_set_get_checks {} { ...@@ -25,7 +25,7 @@ proc default_set_get_checks {} {
assert_match {} [cmdstat lrange] assert_match {} [cmdstat lrange]
} }
start_server {tags {"benchmark network external:skip"}} { start_server {tags {"benchmark network external:skip logreqres:skip"}} {
start_server {} { start_server {} {
set master_host [srv 0 host] set master_host [srv 0 host]
set master_port [srv 0 port] set master_port [srv 0 port]
......
...@@ -60,7 +60,7 @@ start_server {tags {"cli"}} { ...@@ -60,7 +60,7 @@ start_server {tags {"cli"}} {
# Helpers to run tests in interactive mode # Helpers to run tests in interactive mode
proc format_output {output} { proc format_output {output} {
set _ [string trimright [regsub -all "\r" $output ""] "\n"] set _ [string trimright $output "\n"]
} }
proc run_command {fd cmd} { proc run_command {fd cmd} {
...@@ -76,6 +76,12 @@ start_server {tags {"cli"}} { ...@@ -76,6 +76,12 @@ start_server {tags {"cli"}} {
unset ::env(FAKETTY) unset ::env(FAKETTY)
} }
proc test_interactive_nontty_cli {name code} {
set fd [open_cli]
test "Interactive non-TTY CLI: $name" $code
close_cli $fd
}
# Helpers to run tests where stdout is not a tty # Helpers to run tests where stdout is not a tty
proc write_tmpfile {contents} { proc write_tmpfile {contents} {
set tmp [tmpfile "cli"] set tmp [tmpfile "cli"]
...@@ -142,7 +148,8 @@ start_server {tags {"cli"}} { ...@@ -142,7 +148,8 @@ start_server {tags {"cli"}} {
test_interactive_cli "INFO response should be printed raw" { test_interactive_cli "INFO response should be printed raw" {
set lines [split [run_command $fd info] "\n"] set lines [split [run_command $fd info] "\n"]
foreach line $lines { foreach line $lines {
if {![regexp {^$|^#|^[^#:]+:} $line]} { # Info lines end in \r\n, so they now end in \r.
if {![regexp {^\r$|^#|^[^#:]+:} $line]} {
fail "Malformed info line: $line" fail "Malformed info line: $line"
} }
} }
...@@ -186,6 +193,83 @@ start_server {tags {"cli"}} { ...@@ -186,6 +193,83 @@ start_server {tags {"cli"}} {
assert_equal "bar" [r get key] assert_equal "bar" [r get key]
} }
test_interactive_cli "Subscribed mode" {
if {$::force_resp3} {
run_command $fd "hello 3"
}
set reading "Reading messages... (press Ctrl-C to quit or any key to type command)\r"
set erase "\033\[K"; # Erases the "Reading messages..." line.
# Subscribe to some channels.
set sub1 "1) \"subscribe\"\n2) \"ch1\"\n3) (integer) 1\n"
set sub2 "1) \"subscribe\"\n2) \"ch2\"\n3) (integer) 2\n"
set sub3 "1) \"subscribe\"\n2) \"ch3\"\n3) (integer) 3\n"
assert_equal $sub1$sub2$sub3$reading \
[run_command $fd "subscribe ch1 ch2 ch3"]
# Receive pubsub message.
r publish ch2 hello
set message "1) \"message\"\n2) \"ch2\"\n3) \"hello\"\n"
assert_equal $erase$message$reading [read_cli $fd]
# Unsubscribe some.
set unsub1 "1) \"unsubscribe\"\n2) \"ch1\"\n3) (integer) 2\n"
set unsub2 "1) \"unsubscribe\"\n2) \"ch2\"\n3) (integer) 1\n"
assert_equal $erase$unsub1$unsub2$reading \
[run_command $fd "unsubscribe ch1 ch2"]
run_command $fd "hello 2"
# Command forbidden in subscribed mode (RESP2).
set err "(error) ERR Can't execute 'get': only (P|S)SUBSCRIBE / (P|S)UNSUBSCRIBE / PING / QUIT / RESET are allowed in this context\n"
assert_equal $erase$err$reading [run_command $fd "get k"]
# Command allowed in subscribed mode.
set pong "1) \"pong\"\n2) \"\"\n"
assert_equal $erase$pong$reading [run_command $fd "ping"]
# Reset exits subscribed mode.
assert_equal ${erase}RESET [run_command $fd "reset"]
assert_equal PONG [run_command $fd "ping"]
# Check TTY output of push messages in RESP3 has ")" prefix (to be changed to ">" in the future).
assert_match "1#*" [run_command $fd "hello 3"]
set sub1 "1) \"subscribe\"\n2) \"ch1\"\n3) (integer) 1\n"
assert_equal $sub1$reading \
[run_command $fd "subscribe ch1"]
}
test_interactive_nontty_cli "Subscribed mode" {
# Raw output and no "Reading messages..." info message.
# Use RESP3 in this test case.
assert_match {*proto 3*} [run_command $fd "hello 3"]
# Subscribe to some channels.
set sub1 "subscribe\nch1\n1"
set sub2 "subscribe\nch2\n2"
assert_equal $sub1\n$sub2 \
[run_command $fd "subscribe ch1 ch2"]
assert_equal OK [run_command $fd "client tracking on"]
assert_equal OK [run_command $fd "set k 42"]
assert_equal 42 [run_command $fd "get k"]
# Interleaving invalidate and pubsub messages.
r publish ch1 hello
r del k
r publish ch2 world
set message1 "message\nch1\nhello"
set invalidate "invalidate\nk"
set message2 "message\nch2\nworld"
assert_equal $message1\n$invalidate\n$message2\n [read_cli $fd]
# Unsubscribe all.
set unsub1 "unsubscribe\nch1\n1"
set unsub2 "unsubscribe\nch2\n0"
assert_equal $unsub1\n$unsub2 [run_command $fd "unsubscribe ch1 ch2"]
}
test_tty_cli "Status reply" { test_tty_cli "Status reply" {
assert_equal "OK" [run_cli set key bar] assert_equal "OK" [run_cli set key bar]
assert_equal "bar" [r get key] assert_equal "bar" [r get key]
......
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