Merge branch 'unstable' of https://github.com/antirez/redis into unstable
Former-commit-id: 9322d604eea7b48df3feff47ce2c04f82291228f
This commit is contained in:
commit
8a9e9ff3f4
@ -1248,7 +1248,7 @@ int rewriteModuleObject(rio *r, robj *key, robj *o) {
|
||||
RedisModuleIO io;
|
||||
moduleValue *mv = ptrFromObj(o);
|
||||
moduleType *mt = mv->type;
|
||||
moduleInitIOContext(io,mt,r);
|
||||
moduleInitIOContext(io,mt,r,key);
|
||||
mt->aof_rewrite(&io,key,mv->value);
|
||||
if (io.ctx) {
|
||||
moduleFreeContext(io.ctx);
|
||||
|
@ -4776,7 +4776,7 @@ NULL
|
||||
|
||||
/* Generates a DUMP-format representation of the object 'o', adding it to the
|
||||
* io stream pointed by 'rio'. This function can't fail. */
|
||||
void createDumpPayload(rio *payload, robj *o) {
|
||||
void createDumpPayload(rio *payload, robj *o, robj *key) {
|
||||
unsigned char buf[2];
|
||||
uint64_t crc;
|
||||
|
||||
@ -4784,7 +4784,7 @@ void createDumpPayload(rio *payload, robj *o) {
|
||||
* byte followed by the serialized object. This is understood by RESTORE. */
|
||||
rioInitWithBuffer(payload,sdsempty());
|
||||
serverAssert(rdbSaveObjectType(payload,o));
|
||||
serverAssert(rdbSaveObject(payload,o));
|
||||
serverAssert(rdbSaveObject(payload,o,key));
|
||||
|
||||
/* Write the footer, this is how it looks like:
|
||||
* ----------------+---------------------+---------------+
|
||||
@ -4842,7 +4842,7 @@ void dumpCommand(client *c) {
|
||||
}
|
||||
|
||||
/* Create the DUMP encoded representation. */
|
||||
createDumpPayload(&payload,o);
|
||||
createDumpPayload(&payload,o,c->argv[1]);
|
||||
|
||||
/* Transfer to the client */
|
||||
dumpobj = createObject(OBJ_STRING,payload.io.buffer.ptr);
|
||||
@ -4915,7 +4915,7 @@ void restoreCommand(client *c) {
|
||||
|
||||
rioInitWithBuffer(&payload,ptrFromObj(c->argv[3]));
|
||||
if (((type = rdbLoadObjectType(&payload)) == -1) ||
|
||||
((obj = rdbLoadObject(type,&payload)) == NULL))
|
||||
((obj = rdbLoadObject(type,&payload,c->argv[1])) == NULL))
|
||||
{
|
||||
addReplyError(c,"Bad data format");
|
||||
return;
|
||||
@ -5203,7 +5203,7 @@ try_again:
|
||||
|
||||
/* Emit the payload argument, that is the serialized object using
|
||||
* the DUMP format. */
|
||||
createDumpPayload(&payload,ov[j]);
|
||||
createDumpPayload(&payload,ov[j],kv[j]);
|
||||
serverAssertWithInfo(c,NULL,
|
||||
rioWriteBulkString(&cmd,payload.io.buffer.ptr,
|
||||
sdslen(payload.io.buffer.ptr)));
|
||||
|
@ -3439,6 +3439,14 @@ RedisModuleCtx *RM_GetContextFromIO(RedisModuleIO *io) {
|
||||
return io->ctx;
|
||||
}
|
||||
|
||||
/* Returns a RedisModuleString with the name of the key currently saving or
|
||||
* loading, when an IO data type callback is called. There is no guarantee
|
||||
* that the key name is always available, so this may return NULL.
|
||||
*/
|
||||
const RedisModuleString *RM_GetKeyNameFromIO(RedisModuleIO *io) {
|
||||
return io->key;
|
||||
}
|
||||
|
||||
/* --------------------------------------------------------------------------
|
||||
* Logging
|
||||
* -------------------------------------------------------------------------- */
|
||||
@ -5195,6 +5203,7 @@ void moduleRegisterCoreAPI(void) {
|
||||
REGISTER_API(RetainString);
|
||||
REGISTER_API(StringCompare);
|
||||
REGISTER_API(GetContextFromIO);
|
||||
REGISTER_API(GetKeyNameFromIO);
|
||||
REGISTER_API(BlockClient);
|
||||
REGISTER_API(UnblockClient);
|
||||
REGISTER_API(IsBlockedReplyRequest);
|
||||
|
14
src/rdb.c
14
src/rdb.c
@ -752,7 +752,7 @@ size_t rdbSaveStreamConsumers(rio *rdb, streamCG *cg) {
|
||||
|
||||
/* Save a Redis object.
|
||||
* Returns -1 on error, number of bytes written on success. */
|
||||
ssize_t rdbSaveObject(rio *rdb, robj *o) {
|
||||
ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
|
||||
ssize_t n = 0, nwritten = 0;
|
||||
|
||||
if (o->type == OBJ_STRING) {
|
||||
@ -967,7 +967,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
|
||||
RedisModuleIO io;
|
||||
moduleValue *mv = ptrFromObj(o);
|
||||
moduleType *mt = mv->type;
|
||||
moduleInitIOContext(io,mt,rdb);
|
||||
moduleInitIOContext(io,mt,rdb,key);
|
||||
|
||||
/* Write the "module" identifier as prefix, so that we'll be able
|
||||
* to call the right module during loading. */
|
||||
@ -997,7 +997,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
|
||||
* this length with very little changes to the code. In the future
|
||||
* we could switch to a faster solution. */
|
||||
size_t rdbSavedObjectLen(robj *o) {
|
||||
ssize_t len = rdbSaveObject(NULL,o);
|
||||
ssize_t len = rdbSaveObject(NULL,o,NULL);
|
||||
serverAssertWithInfo(NULL,o,len != -1);
|
||||
return len;
|
||||
}
|
||||
@ -1039,7 +1039,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
|
||||
/* Save type, key, value */
|
||||
if (rdbSaveObjectType(rdb,val) == -1) return -1;
|
||||
if (rdbSaveStringObject(rdb,key) == -1) return -1;
|
||||
if (rdbSaveObject(rdb,val) == -1) return -1;
|
||||
if (rdbSaveObject(rdb,val,key) == -1) return -1;
|
||||
return 1;
|
||||
}
|
||||
|
||||
@ -1401,7 +1401,7 @@ robj *rdbLoadCheckModuleValue(rio *rdb, char *modulename) {
|
||||
|
||||
/* Load a Redis object of the specified type from the specified file.
|
||||
* On success a newly allocated object is returned, otherwise NULL. */
|
||||
robj *rdbLoadObject(int rdbtype, rio *rdb) {
|
||||
robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
|
||||
robj *o = NULL, *ele, *dec;
|
||||
uint64_t len;
|
||||
unsigned int i;
|
||||
@ -1788,7 +1788,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
|
||||
exit(1);
|
||||
}
|
||||
RedisModuleIO io;
|
||||
moduleInitIOContext(io,mt,rdb);
|
||||
moduleInitIOContext(io,mt,rdb,key);
|
||||
io.ver = (rdbtype == RDB_TYPE_MODULE) ? 1 : 2;
|
||||
/* Call the rdb_load method of the module providing the 10 bit
|
||||
* encoding version in the lower 10 bits of the module ID. */
|
||||
@ -2044,7 +2044,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
|
||||
/* Read key */
|
||||
if ((key = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
|
||||
/* Read value */
|
||||
if ((val = rdbLoadObject(type,rdb)) == NULL) goto eoferr;
|
||||
if ((val = rdbLoadObject(type,rdb,key)) == NULL) goto eoferr;
|
||||
/* Check if the key already expired. This function is used when loading
|
||||
* an RDB file from disk, either at startup, or when an RDB was
|
||||
* received from the master. In the latter case, the master is
|
||||
|
@ -144,9 +144,9 @@ int rdbSaveFile(char *filename, rdbSaveInfo *rsi);
|
||||
int rdbSaveFd(int fd, rdbSaveInfo *rsi);
|
||||
int rdbSaveS3(char *path, rdbSaveInfo *rsi);
|
||||
int rdbLoadS3(char *path, rdbSaveInfo *rsi);
|
||||
ssize_t rdbSaveObject(rio *rdb, robj *o);
|
||||
ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key);
|
||||
size_t rdbSavedObjectLen(robj *o);
|
||||
robj *rdbLoadObject(int type, rio *rdb);
|
||||
robj *rdbLoadObject(int type, rio *rdb, robj *key);
|
||||
void backgroundSaveDoneHandler(int exitcode, int bysignal);
|
||||
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime);
|
||||
robj *rdbLoadStringObject(rio *rdb);
|
||||
|
@ -285,7 +285,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
|
||||
rdbstate.keys++;
|
||||
/* Read value */
|
||||
rdbstate.doing = RDB_CHECK_DOING_READ_OBJECT_VALUE;
|
||||
if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
|
||||
if ((val = rdbLoadObject(type,&rdb,key)) == NULL) goto eoferr;
|
||||
/* Check if the key already expired. */
|
||||
if (expiretime != -1 && expiretime < now)
|
||||
rdbstate.already_expired++;
|
||||
|
@ -278,6 +278,7 @@ int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, Re
|
||||
void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, RedisModuleString *str);
|
||||
int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, RedisModuleString *b);
|
||||
RedisModuleCtx *REDISMODULE_API_FUNC(RedisModule_GetContextFromIO)(RedisModuleIO *io);
|
||||
const RedisModuleString *REDISMODULE_API_FUNC(RedisModule_GetKeyNameFromIO)(RedisModuleIO *io);
|
||||
long long REDISMODULE_API_FUNC(RedisModule_Milliseconds)(void);
|
||||
void REDISMODULE_API_FUNC(RedisModule_DigestAddStringBuffer)(RedisModuleDigest *md, unsigned char *ele, size_t len);
|
||||
void REDISMODULE_API_FUNC(RedisModule_DigestAddLongLong)(RedisModuleDigest *md, long long ele);
|
||||
@ -442,6 +443,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
|
||||
REDISMODULE_GET_API(RetainString);
|
||||
REDISMODULE_GET_API(StringCompare);
|
||||
REDISMODULE_GET_API(GetContextFromIO);
|
||||
REDISMODULE_GET_API(GetKeyNameFromIO);
|
||||
REDISMODULE_GET_API(Milliseconds);
|
||||
REDISMODULE_GET_API(DigestAddStringBuffer);
|
||||
REDISMODULE_GET_API(DigestAddLongLong);
|
||||
|
@ -1143,14 +1143,23 @@ void replicationCreateMasterClient(int fd, int dbid) {
|
||||
if (dbid != -1) selectDb(server.master,dbid);
|
||||
}
|
||||
|
||||
void restartAOF() {
|
||||
int retry = 10;
|
||||
while (retry-- && startAppendOnly() == C_ERR) {
|
||||
serverLog(LL_WARNING,"Failed enabling the AOF after successful master synchronization! Trying it again in one second.");
|
||||
/* This function will try to re-enable the AOF file after the
|
||||
* master-replica synchronization: if it fails after multiple attempts
|
||||
* the replica cannot be considered reliable and exists with an
|
||||
* error. */
|
||||
void restartAOFAfterSYNC() {
|
||||
unsigned int tries, max_tries = 10;
|
||||
for (tries = 0; tries < max_tries; ++tries) {
|
||||
if (startAppendOnly() == C_OK) break;
|
||||
serverLog(LL_WARNING,
|
||||
"Failed enabling the AOF after successful master synchronization! "
|
||||
"Trying it again in one second.");
|
||||
sleep(1);
|
||||
}
|
||||
if (!retry) {
|
||||
serverLog(LL_WARNING,"FATAL: this replica instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now.");
|
||||
if (tries == max_tries) {
|
||||
serverLog(LL_WARNING,
|
||||
"FATAL: this replica instance finished the synchronization with "
|
||||
"its master, but the AOF can't be turned on. Exiting now.");
|
||||
exit(1);
|
||||
}
|
||||
}
|
||||
@ -1339,7 +1348,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
cancelReplicationHandshake();
|
||||
/* Re-enable the AOF if we disabled it earlier, in order to restore
|
||||
* the original configuration. */
|
||||
if (aof_is_enabled) restartAOF();
|
||||
if (aof_is_enabled) restartAOFAfterSYNC();
|
||||
return;
|
||||
}
|
||||
/* Final setup of the connected slave <- master link */
|
||||
@ -1364,7 +1373,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
/* Restart the AOF subsystem now that we finished the sync. This
|
||||
* will trigger an AOF rewrite, and when done will start appending
|
||||
* to the new file. */
|
||||
if (aof_is_enabled) restartAOF();
|
||||
if (aof_is_enabled) restartAOFAfterSYNC();
|
||||
}
|
||||
return;
|
||||
|
||||
|
@ -719,7 +719,7 @@ struct redisCommand redisCommandTable[] = {
|
||||
|
||||
{"touch",touchCommand,-2,
|
||||
"read-only fast @keyspace",
|
||||
0,NULL,1,1,1,0,0,0},
|
||||
0,NULL,1,-1,1,0,0,0},
|
||||
|
||||
{"pttl",pttlCommand,2,
|
||||
"read-only fast random @keyspace",
|
||||
@ -867,7 +867,7 @@ struct redisCommand redisCommandTable[] = {
|
||||
"no-script @keyspace",
|
||||
0,NULL,0,0,0,0,0,0},
|
||||
|
||||
{"command",commandCommand,0,
|
||||
{"command",commandCommand,-1,
|
||||
"ok-loading ok-stale random @connection",
|
||||
0,NULL,0,0,0,0,0,0},
|
||||
|
||||
@ -4631,6 +4631,7 @@ static void sigShutdownHandler(int sig) {
|
||||
rdbRemoveTempFile(getpid());
|
||||
exit(1); /* Exit with an error since this was not a clean shutdown. */
|
||||
} else if (server.loading) {
|
||||
serverLogFromHandler(LL_WARNING, "Received shutdown signal during loading, exiting now.");
|
||||
exit(0);
|
||||
}
|
||||
|
||||
|
@ -595,16 +595,18 @@ typedef struct RedisModuleIO {
|
||||
int ver; /* Module serialization version: 1 (old),
|
||||
* 2 (current version with opcodes annotation). */
|
||||
struct RedisModuleCtx *ctx; /* Optional context, see RM_GetContextFromIO()*/
|
||||
struct redisObject *key; /* Optional name of key processed */
|
||||
} RedisModuleIO;
|
||||
|
||||
/* Macro to initialize an IO context. Note that the 'ver' field is populated
|
||||
* inside rdb.c according to the version of the value to load. */
|
||||
#define moduleInitIOContext(iovar,mtype,rioptr) do { \
|
||||
#define moduleInitIOContext(iovar,mtype,rioptr,keyptr) do { \
|
||||
iovar.prio = rioptr; \
|
||||
iovar.type = mtype; \
|
||||
iovar.bytes = 0; \
|
||||
iovar.error = 0; \
|
||||
iovar.ver = 0; \
|
||||
iovar.key = keyptr; \
|
||||
iovar.ctx = NULL; \
|
||||
} while(0);
|
||||
|
||||
|
55
src/sort.c
55
src/sort.c
@ -58,7 +58,7 @@ redisSortOperation *createSortOperation(int type, robj *pattern) {
|
||||
*
|
||||
* The returned object will always have its refcount increased by 1
|
||||
* when it is non-NULL. */
|
||||
robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
|
||||
robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst, int writeflag) {
|
||||
char *p, *f, *k;
|
||||
sds spat, ssub;
|
||||
robj *keyobj, *fieldobj = NULL, *o;
|
||||
@ -106,7 +106,10 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
|
||||
decrRefCount(subst); /* Incremented by decodeObject() */
|
||||
|
||||
/* Lookup substituted key */
|
||||
o = lookupKeyRead(db,keyobj);
|
||||
if (!writeflag)
|
||||
o = lookupKeyRead(db,keyobj);
|
||||
else
|
||||
o = lookupKeyWrite(db,keyobj);
|
||||
if (o == NULL) goto noobj;
|
||||
|
||||
if (fieldobj) {
|
||||
@ -198,30 +201,12 @@ void sortCommand(client *c) {
|
||||
robj *sortval, *sortby = NULL, *storekey = NULL;
|
||||
redisSortObject *vector; /* Resulting vector to sort */
|
||||
|
||||
/* Lookup the key to sort. It must be of the right types */
|
||||
sortval = lookupKeyRead(c->db,c->argv[1]);
|
||||
if (sortval && sortval->type != OBJ_SET &&
|
||||
sortval->type != OBJ_LIST &&
|
||||
sortval->type != OBJ_ZSET)
|
||||
{
|
||||
addReply(c,shared.wrongtypeerr);
|
||||
return;
|
||||
}
|
||||
|
||||
/* Create a list of operations to perform for every sorted element.
|
||||
* Operations can be GET */
|
||||
operations = listCreate();
|
||||
listSetFreeMethod(operations,zfree);
|
||||
j = 2; /* options start at argv[2] */
|
||||
|
||||
/* Now we need to protect sortval incrementing its count, in the future
|
||||
* SORT may have options able to overwrite/delete keys during the sorting
|
||||
* and the sorted key itself may get destroyed */
|
||||
if (sortval)
|
||||
incrRefCount(sortval);
|
||||
else
|
||||
sortval = createQuicklistObject();
|
||||
|
||||
/* The SORT command has an SQL-alike syntax, parse it */
|
||||
while(j < c->argc) {
|
||||
int leftargs = c->argc-j-1;
|
||||
@ -280,11 +265,33 @@ void sortCommand(client *c) {
|
||||
|
||||
/* Handle syntax errors set during options parsing. */
|
||||
if (syntax_error) {
|
||||
decrRefCount(sortval);
|
||||
listRelease(operations);
|
||||
return;
|
||||
}
|
||||
|
||||
/* Lookup the key to sort. It must be of the right types */
|
||||
if (storekey)
|
||||
sortval = lookupKeyRead(c->db,c->argv[1]);
|
||||
else
|
||||
sortval = lookupKeyWrite(c->db,c->argv[1]);
|
||||
if (sortval && sortval->type != OBJ_SET &&
|
||||
sortval->type != OBJ_LIST &&
|
||||
sortval->type != OBJ_ZSET)
|
||||
{
|
||||
listRelease(operations);
|
||||
addReply(c,shared.wrongtypeerr);
|
||||
return;
|
||||
}
|
||||
|
||||
/* Now we need to protect sortval incrementing its count, in the future
|
||||
* SORT may have options able to overwrite/delete keys during the sorting
|
||||
* and the sorted key itself may get destroyed */
|
||||
if (sortval)
|
||||
incrRefCount(sortval);
|
||||
else
|
||||
sortval = createQuicklistObject();
|
||||
|
||||
|
||||
/* When sorting a set with no sort specified, we must sort the output
|
||||
* so the result is consistent across scripting and replication.
|
||||
*
|
||||
@ -452,7 +459,7 @@ void sortCommand(client *c) {
|
||||
robj *byval;
|
||||
if (sortby) {
|
||||
/* lookup value to sort by */
|
||||
byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
|
||||
byval = lookupKeyByPattern(c->db,sortby,vector[j].obj,storekey!=NULL);
|
||||
if (!byval) continue;
|
||||
} else {
|
||||
/* use object itself to sort by */
|
||||
@ -515,7 +522,7 @@ void sortCommand(client *c) {
|
||||
while((ln = listNext(&li))) {
|
||||
redisSortOperation *sop = ln->value;
|
||||
robj *val = lookupKeyByPattern(c->db,sop->pattern,
|
||||
vector[j].obj);
|
||||
vector[j].obj,storekey!=NULL);
|
||||
|
||||
if (sop->type == SORT_OP_GET) {
|
||||
if (!val) {
|
||||
@ -545,7 +552,7 @@ void sortCommand(client *c) {
|
||||
while((ln = listNext(&li))) {
|
||||
redisSortOperation *sop = ln->value;
|
||||
robj *val = lookupKeyByPattern(c->db,sop->pattern,
|
||||
vector[j].obj);
|
||||
vector[j].obj,storekey!=NULL);
|
||||
|
||||
if (sop->type == SORT_OP_GET) {
|
||||
if (!val) val = createStringObject("",0);
|
||||
|
@ -415,7 +415,7 @@ void spopWithCountCommand(client *c) {
|
||||
|
||||
/* Make sure a key with the name inputted exists, and that it's type is
|
||||
* indeed a set. Otherwise, return nil */
|
||||
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
|
||||
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
|
||||
== NULL || checkType(c,set,OBJ_SET)) return;
|
||||
|
||||
/* If count is zero, serve an empty multibulk ASAP to avoid special
|
||||
|
@ -159,6 +159,10 @@ void *zrealloc(void *ptr, size_t size, enum MALLOC_CLASS class) {
|
||||
size_t oldsize;
|
||||
void *newptr;
|
||||
|
||||
if (size == 0 && ptr != NULL) {
|
||||
zfree(ptr);
|
||||
return NULL;
|
||||
}
|
||||
if (ptr == NULL) return zmalloc(size, class);
|
||||
#ifdef HAVE_MALLOC_SIZE
|
||||
oldsize = zmalloc_size(ptr);
|
||||
|
Loading…
x
Reference in New Issue
Block a user