gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge branch 'unstable' of github.com:/antirez/redis into unstable

Browse Source
This commit is contained in:
antirez 2019-11-19 11:10:35 +01:00
commit 77ad9aaafe
6 changed files with 431 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
runtest-moduleapi
View File

@ -22,5 +22,6 @@ $TCLSH tests/test_helper.tcl \
--single unit/moduleapi/hooks \
--single unit/moduleapi/misc \
--single unit/moduleapi/blockonkeys \
--single unit/moduleapi/datatype
--single unit/moduleapi/scan \
--single unit/moduleapi/datatype \
"${@}"

270
src/module.c
View File

@ -1849,6 +1849,18 @@ int RM_SelectDb(RedisModuleCtx *ctx, int newid) {
return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR;
}
/* Initialize a RedisModuleKey struct */
static void moduleInitKey(RedisModuleKey *kp, RedisModuleCtx *ctx, robj *keyname, robj *value, int mode){
kp->ctx = ctx;
kp->db = ctx->client->db;
kp->key = keyname;
incrRefCount(keyname);
kp->value = value;
kp->iter = NULL;
kp->mode = mode;
zsetKeyReset(kp);
}
/* Return an handle representing a Redis key, so that it is possible
* to call other APIs with the key handle as argument to perform
* operations on the key.
@ -1879,27 +1891,25 @@ void *RM_OpenKey(RedisModuleCtx *ctx, robj *keyname, int mode) {
/* Setup the key handle. */
kp = zmalloc(sizeof(*kp));
kp->ctx = ctx;
kp->db = ctx->client->db;
kp->key = keyname;
incrRefCount(keyname);
kp->value = value;
kp->iter = NULL;
kp->mode = mode;
zsetKeyReset(kp);
moduleInitKey(kp, ctx, keyname, value, mode);
autoMemoryAdd(ctx,REDISMODULE_AM_KEY,kp);
return (void*)kp;
}
/* Close a key handle. */
void RM_CloseKey(RedisModuleKey *key) {
if (key == NULL) return;
/* Destroy a RedisModuleKey struct (freeing is the responsibility of the caller). */
static void moduleCloseKey(RedisModuleKey *key) {
int signal = SHOULD_SIGNAL_MODIFIED_KEYS(key->ctx);
if ((key->mode & REDISMODULE_WRITE) && signal)
signalModifiedKey(key->db,key->key);
/* TODO: if (key->iter) RM_KeyIteratorStop(kp); */
RM_ZsetRangeStop(key);
decrRefCount(key->key);
}
/* Close a key handle. */
void RM_CloseKey(RedisModuleKey *key) {
if (key == NULL) return;
moduleCloseKey(key);
autoMemoryFreed(key->ctx,REDISMODULE_AM_KEY,key);
zfree(key);
}
@ -5945,6 +5955,239 @@ int RM_CommandFilterArgDelete(RedisModuleCommandFilterCtx *fctx, int pos)
return REDISMODULE_OK;
}
/* --------------------------------------------------------------------------
* Scanning keyspace and hashes
* -------------------------------------------------------------------------- */
typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
typedef struct {
RedisModuleCtx *ctx;
void* user_data;
RedisModuleScanCB fn;
} ScanCBData;
typedef struct RedisModuleScanCursor{
int cursor;
int done;
}RedisModuleScanCursor;
static void moduleScanCallback(void *privdata, const dictEntry *de) {
ScanCBData *data = privdata;
sds key = dictGetKey(de);
robj* val = dictGetVal(de);
RedisModuleString *keyname = createObject(OBJ_STRING,sdsdup(key));
/* Setup the key handle. */
RedisModuleKey kp = {0};
moduleInitKey(&kp, data->ctx, keyname, val, REDISMODULE_READ);
data->fn(data->ctx, keyname, &kp, data->user_data);
moduleCloseKey(&kp);
decrRefCount(keyname);
}
/* Create a new cursor to be used with RedisModule_Scan */
RedisModuleScanCursor *RM_ScanCursorCreate() {
RedisModuleScanCursor* cursor = zmalloc(sizeof(*cursor));
cursor->cursor = 0;
cursor->done = 0;
return cursor;
}
/* Restart an existing cursor. The keys will be rescanned. */
void RM_ScanCursorRestart(RedisModuleScanCursor *cursor) {
cursor->cursor = 0;
cursor->done = 0;
}
/* Destroy the cursor struct. */
void RM_ScanCursorDestroy(RedisModuleScanCursor *cursor) {
zfree(cursor);
}
/* Scan api that allows a module to scan all the keys and value in the selected db.
*
* Callback for scan implementation.
* void scan_callback(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
* - ctx - the redis module context provided to for the scan.
* - keyname - owned by the caller and need to be retained if used after this function.
* - key - holds info on the key and value, it is provided as best effort, in some cases it might
* be NULL, in which case the user should (can) use RedisModule_OpenKey (and CloseKey too).
* when it is provided, it is owned by the caller and will be free when the callback returns.
* - privdata - the user data provided to RedisModule_Scan.
*
* The way it should be used:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* while(RedisModule_Scan(ctx, c, callback, privateData));
* RedisModule_ScanCursorDestroy(c);
*
* It is also possible to use this API from another thread while the lock is acquired durring
* the actuall call to RM_Scan:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* RedisModule_ThreadSafeContextLock(ctx);
* while(RedisModule_Scan(ctx, c, callback, privateData)){
* RedisModule_ThreadSafeContextUnlock(ctx);
* // do some background job
* RedisModule_ThreadSafeContextLock(ctx);
* }
* RedisModule_ScanCursorDestroy(c);
*
* The function will return 1 if there are more elements to scan and 0 otherwise,
* possibly setting errno if the call failed.
* It is also possible to restart and existing cursor using RM_CursorRestart. */
int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanCB fn, void *privdata) {
if (cursor->done) {
errno = ENOENT;
return 0;
}
int ret = 1;
ScanCBData data = { ctx, privdata, fn };
cursor->cursor = dictScan(ctx->client->db->dict, cursor->cursor, moduleScanCallback, NULL, &data);
if (cursor->cursor == 0) {
cursor->done = 1;
ret = 0;
}
errno = 0;
return ret;
}
typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata);
typedef struct {
RedisModuleKey *key;
void* user_data;
RedisModuleScanKeyCB fn;
} ScanKeyCBData;
static void moduleScanKeyCallback(void *privdata, const dictEntry *de) {
ScanKeyCBData *data = privdata;
sds key = dictGetKey(de);
robj *o = data->key->value;
robj *field = createStringObject(key, sdslen(key));
robj *value = NULL;
if (o->type == OBJ_SET) {
value = NULL;
} else if (o->type == OBJ_HASH) {
sds val = dictGetVal(de);
value = createStringObject(val, sdslen(val));
} else if (o->type == OBJ_ZSET) {
double *val = (double*)dictGetVal(de);
value = createStringObjectFromLongDouble(*val, 0);
}
data->fn(data->key, field, value, data->user_data);
decrRefCount(field);
if (value) decrRefCount(value);
}
/* Scan api that allows a module to scan the elements in a hash, set or sorted set key
*
* Callback for scan implementation.
* void scan_callback(RedisModuleKey *key, RedisModuleString* field, RedisModuleString* value, void *privdata);
* - key - the redis key context provided to for the scan.
* - field - field name, owned by the caller and need to be retained if used
* after this function.
* - value - value string or NULL for set type, owned by the caller and need to
* be retained if used after this function.
* - privdata - the user data provided to RedisModule_ScanKey.
*
* The way it should be used:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* RedisModuleKey *key = RedisModule_OpenKey(...)
* while(RedisModule_ScanKey(key, c, callback, privateData));
* RedisModule_CloseKey(key);
* RedisModule_ScanCursorDestroy(c);
*
* It is also possible to use this API from another thread while the lock is acquired durring
* the actuall call to RM_Scan, and re-opening the key each time:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* RedisModule_ThreadSafeContextLock(ctx);
* RedisModuleKey *key = RedisModule_OpenKey(...)
* while(RedisModule_ScanKey(ctx, c, callback, privateData)){
* RedisModule_CloseKey(key);
* RedisModule_ThreadSafeContextUnlock(ctx);
* // do some background job
* RedisModule_ThreadSafeContextLock(ctx);
* RedisModuleKey *key = RedisModule_OpenKey(...)
* }
* RedisModule_CloseKey(key);
* RedisModule_ScanCursorDestroy(c);
*
* The function will return 1 if there are more elements to scan and 0 otherwise,
* possibly setting errno if the call failed.
* It is also possible to restart and existing cursor using RM_CursorRestart. */
int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleScanKeyCB fn, void *privdata) {
if (key == NULL || key->value == NULL) {
errno = EINVAL;
return 0;
}
dict *ht = NULL;
robj *o = key->value;
if (o->type == OBJ_SET) {
if (o->encoding == OBJ_ENCODING_HT)
ht = o->ptr;
} else if (o->type == OBJ_HASH) {
if (o->encoding == OBJ_ENCODING_HT)
ht = o->ptr;
} else if (o->type == OBJ_ZSET) {
if (o->encoding == OBJ_ENCODING_SKIPLIST)
ht = ((zset *)o->ptr)->dict;
} else {
errno = EINVAL;
return 0;
}
if (cursor->done) {
errno = ENOENT;
return 0;
}
int ret = 1;
if (ht) {
ScanKeyCBData data = { key, privdata, fn };
cursor->cursor = dictScan(ht, cursor->cursor, moduleScanKeyCallback, NULL, &data);
if (cursor->cursor == 0) {
cursor->done = 1;
ret = 0;
}
} else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_INTSET) {
int pos = 0;
int64_t ll;
while(intsetGet(o->ptr,pos++,&ll)) {
robj *field = createStringObjectFromLongLong(ll);
fn(key, field, NULL, privdata);
decrRefCount(field);
}
cursor->cursor = 1;
cursor->done = 1;
ret = 0;
} else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) {
unsigned char *p = ziplistIndex(o->ptr,0);
unsigned char *vstr;
unsigned int vlen;
long long vll;
while(p) {
ziplistGet(p,&vstr,&vlen,&vll);
robj *field = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createStringObjectFromLongLong(vll);
p = ziplistNext(o->ptr,p);
ziplistGet(p,&vstr,&vlen,&vll);
robj *value = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createStringObjectFromLongLong(vll);
fn(key, field, value, privdata);
p = ziplistNext(o->ptr,p);
decrRefCount(field);
decrRefCount(value);
}
cursor->cursor = 1;
cursor->done = 1;
ret = 0;
}
errno = 0;
return ret;
}
/* --------------------------------------------------------------------------
* Module fork API
* -------------------------------------------------------------------------- */
@ -7054,4 +7297,9 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(BlockClientOnKeys);
REGISTER_API(SignalKeyAsReady);
REGISTER_API(GetBlockedClientReadyKey);
REGISTER_API(ScanCursorCreate);
REGISTER_API(ScanCursorDestroy);
REGISTER_API(ScanCursorRestart);
REGISTER_API(Scan);
REGISTER_API(ScanKey);
}

13
src/redismodule.h
View File

@ -392,6 +392,7 @@ typedef struct RedisModuleDictIter RedisModuleDictIter;
typedef struct RedisModuleCommandFilterCtx RedisModuleCommandFilterCtx;
typedef struct RedisModuleCommandFilter RedisModuleCommandFilter;
typedef struct RedisModuleInfoCtx RedisModuleInfoCtx;
typedef struct RedisModuleScanCursor RedisModuleScanCursor;
typedef int (*RedisModuleCmdFunc)(RedisModuleCtx *ctx, RedisModuleString **argv, int argc);
typedef void (*RedisModuleDisconnectFunc)(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc);
@ -409,6 +410,8 @@ typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data);
typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter);
typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
typedef void (*RedisModuleInfoFunc)(RedisModuleInfoCtx *ctx, int for_crash_report);
typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata);
#define REDISMODULE_TYPE_METHOD_VERSION 2
typedef struct RedisModuleTypeMethods {
@ -591,6 +594,11 @@ int REDISMODULE_API_FUNC(RedisModule_GetLRUOrLFU)(RedisModuleKey *key, long long
RedisModuleBlockedClient *REDISMODULE_API_FUNC(RedisModule_BlockClientOnKeys)(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms, RedisModuleString **keys, int numkeys, void *privdata);
void REDISMODULE_API_FUNC(RedisModule_SignalKeyAsReady)(RedisModuleCtx *ctx, RedisModuleString *key);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_GetBlockedClientReadyKey)(RedisModuleCtx *ctx);
RedisModuleScanCursor *REDISMODULE_API_FUNC(RedisModule_ScanCursorCreate)();
void REDISMODULE_API_FUNC(RedisModule_ScanCursorRestart)(RedisModuleScanCursor *cursor);
void REDISMODULE_API_FUNC(RedisModule_ScanCursorDestroy)(RedisModuleScanCursor *cursor);
int REDISMODULE_API_FUNC(RedisModule_Scan)(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanCB fn, void *privdata);
int REDISMODULE_API_FUNC(RedisModule_ScanKey)(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleScanKeyCB fn, void *privdata);
/* Experimental APIs */
#ifdef REDISMODULE_EXPERIMENTAL_API
@ -805,6 +813,11 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(BlockClientOnKeys);
REDISMODULE_GET_API(SignalKeyAsReady);
REDISMODULE_GET_API(GetBlockedClientReadyKey);
REDISMODULE_GET_API(ScanCursorCreate);
REDISMODULE_GET_API(ScanCursorRestart);
REDISMODULE_GET_API(ScanCursorDestroy);
REDISMODULE_GET_API(Scan);
REDISMODULE_GET_API(ScanKey);
#ifdef REDISMODULE_EXPERIMENTAL_API
REDISMODULE_GET_API(GetThreadSafeContext);

1
tests/modules/Makefile
View File

@ -20,6 +20,7 @@ TEST_MODULES = \
misc.so \
hooks.so \
blockonkeys.so \
scan.so \
datatype.so
.PHONY: all

109
tests/modules/scan.c Normal file
View File

@ -0,0 +1,109 @@
#include "redismodule.h"
#include <string.h>
#include <assert.h>
#include <unistd.h>
typedef struct {
size_t nkeys;
} scan_strings_pd;
void scan_strings_callback(RedisModuleCtx *ctx, RedisModuleString* keyname, RedisModuleKey* key, void *privdata) {
scan_strings_pd* pd = privdata;
int was_opened = 0;
if (!key) {
key = RedisModule_OpenKey(ctx, keyname, REDISMODULE_READ);
was_opened = 1;
}
if (RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_STRING) {
size_t len;
char * data = RedisModule_StringDMA(key, &len, REDISMODULE_READ);
RedisModule_ReplyWithArray(ctx, 2);
RedisModule_ReplyWithString(ctx, keyname);
RedisModule_ReplyWithStringBuffer(ctx, data, len);
pd->nkeys++;
}
if (was_opened)
RedisModule_CloseKey(key);
}
int scan_strings(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
scan_strings_pd pd = {
.nkeys = 0,
};
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
RedisModuleScanCursor* cursor = RedisModule_ScanCursorCreate();
while(RedisModule_Scan(ctx, cursor, scan_strings_callback, &pd));
RedisModule_ScanCursorDestroy(cursor);
RedisModule_ReplySetArrayLength(ctx, pd.nkeys);
return REDISMODULE_OK;
}
typedef struct {
RedisModuleCtx *ctx;
size_t nreplies;
} scan_key_pd;
void scan_key_callback(RedisModuleKey *key, RedisModuleString* field, RedisModuleString* value, void *privdata) {
REDISMODULE_NOT_USED(key);
scan_key_pd* pd = privdata;
RedisModule_ReplyWithArray(pd->ctx, 2);
RedisModule_ReplyWithString(pd->ctx, field);
if (value)
RedisModule_ReplyWithString(pd->ctx, value);
else
RedisModule_ReplyWithNull(pd->ctx);
pd->nreplies++;
}
int scan_key(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc != 2) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
scan_key_pd pd = {
.ctx = ctx,
.nreplies = 0,
};
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
if (!key) {
RedisModule_ReplyWithError(ctx, "not found");
return REDISMODULE_OK;
}
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
RedisModuleScanCursor* cursor = RedisModule_ScanCursorCreate();
while(RedisModule_ScanKey(key, cursor, scan_key_callback, &pd));
RedisModule_ScanCursorDestroy(cursor);
RedisModule_ReplySetArrayLength(ctx, pd.nreplies);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx, "scan", 1, REDISMODULE_APIVER_1)== REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "scan.scan_strings", scan_strings, "", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "scan.scan_key", scan_key, "", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}

47
tests/unit/moduleapi/scan.tcl Normal file
View File

@ -0,0 +1,47 @@
set testmodule [file normalize tests/modules/scan.so]
start_server {tags {"modules"}} {
r module load $testmodule
test {Module scan keyspace} {
# the module create a scan command with filtering which also return values
r set x 1
r set y 2
r set z 3
r hset h f v
lsort [r scan.scan_strings]
} {{x 1} {y 2} {z 3}}
test {Module scan hash ziplist} {
r hmset hh f1 v1 f2 v2
lsort [r scan.scan_key hh]
} {{f1 v1} {f2 v2}}
test {Module scan hash dict} {
r config set hash-max-ziplist-entries 2
r hmset hh f3 v3
lsort [r scan.scan_key hh]
} {{f1 v1} {f2 v2} {f3 v3}}
test {Module scan zset ziplist} {
r zadd zz 1 f1 2 f2
lsort [r scan.scan_key zz]
} {{f1 1} {f2 2}}
test {Module scan zset dict} {
r config set zset-max-ziplist-entries 2
r zadd zz 3 f3
lsort [r scan.scan_key zz]
} {{f1 1} {f2 2} {f3 3}}
test {Module scan set intset} {
r sadd ss 1 2
lsort [r scan.scan_key ss]
} {{1 {}} {2 {}}}
test {Module scan set dict} {
r config set set-max-intset-entries 2
r sadd ss 3
lsort [r scan.scan_key ss]
} {{1 {}} {2 {}} {3 {}}}
}