Commit 34dcff0a authored by antirez's avatar antirez
Browse files

Threaded core commands: ability to pass options.

parent 2aef66d3
...@@ -7210,7 +7210,7 @@ void processModuleLoadingProgressEvent(int is_aof) { ...@@ -7210,7 +7210,7 @@ void processModuleLoadingProgressEvent(int is_aof) {
* using abstractions implemented by the modules interface, like the * using abstractions implemented by the modules interface, like the
* ability to run threaded operations. * ability to run threaded operations.
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc); typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc, void *options);
/* This sturcture is used in order to pass state from the main thread to /* This sturcture is used in order to pass state from the main thread to
* the thread that will execute the command, and then to the function * the thread that will execute the command, and then to the function
...@@ -7221,6 +7221,9 @@ typedef struct { ...@@ -7221,6 +7221,9 @@ typedef struct {
robj **objv; /* Vector of Redis objects. */ robj **objv; /* Vector of Redis objects. */
int objc; /* Number of objects. */ int objc; /* Number of objects. */
int freecount; /* How many we need to free. */ int freecount; /* How many we need to free. */
void *options; /* A pointer with details about
the command to execute. Passed
to the callback. */
} tcprivdata; } tcprivdata;
/* Free the private data allocated for a client blocked because of a /* Free the private data allocated for a client blocked because of a
...@@ -7231,6 +7234,7 @@ void threadedCoreCommandFreePrivdata(RedisModuleCtx *ctx, void *privdata) { ...@@ -7231,6 +7234,7 @@ void threadedCoreCommandFreePrivdata(RedisModuleCtx *ctx, void *privdata) {
for (int j = 0; j < tcpd->freecount; j++) for (int j = 0; j < tcpd->freecount; j++)
decrRefCount(tcpd->objv[j]); decrRefCount(tcpd->objv[j]);
zfree(tcpd->objv); zfree(tcpd->objv);
zfree(tcpd->options);
zfree(tcpd); zfree(tcpd);
CoreModuleBlockedClients--; CoreModuleBlockedClients--;
} }
...@@ -7245,7 +7249,7 @@ void *threadedCoreCommandEnty(void *argptr) { ...@@ -7245,7 +7249,7 @@ void *threadedCoreCommandEnty(void *argptr) {
* use Redis low level API to accumulate the reply there, and later when * use Redis low level API to accumulate the reply there, and later when
* the client is unblocked, the reply will be concatenated to the * the client is unblocked, the reply will be concatenated to the
* real client. */ * real client. */
tcpd->callback(tcpd->bc->reply_client,tcpd->objv,tcpd->objc); tcpd->callback(tcpd->bc->reply_client,tcpd->objv,tcpd->objc,tcpd->options);
moduleUnblockClientByHandle(tcpd->bc,tcpd); moduleUnblockClientByHandle(tcpd->bc,tcpd);
return NULL; return NULL;
} }
...@@ -7266,8 +7270,16 @@ void *threadedCoreCommandEnty(void *argptr) { ...@@ -7266,8 +7270,16 @@ void *threadedCoreCommandEnty(void *argptr) {
* objects so that the thread is the only owner, or Redis should have * objects so that the thread is the only owner, or Redis should have
* some other machanism in order to make sure that there are no race * some other machanism in order to make sure that there are no race
* conditions. At the end of the execution of the command, the first * conditions. At the end of the execution of the command, the first
* 'freecount' objects of the objv array will be freed. */ * 'freecount' objects of the objv array will be freed.
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount) { *
* 'options' is some private information passed to the callback that
* may contain parameters that the "main thread half" if the command
* parsed from the command line of the command. Normally it will pass
* things like the options in order to execute the command, like modifiers
* of the return value or alike. The pointer pointed by 'options' must
* be allocated in the stack with zmalloc() and will be free automatically
* at the end of the execution. It can be NULL. */
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount, void *options) {
RedisModuleCtx ctx = REDISMODULE_CTX_INIT; RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
ctx.module = coremodule; ctx.module = coremodule;
ctx.client = c; ctx.client = c;
...@@ -7285,6 +7297,7 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob ...@@ -7285,6 +7297,7 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob
memcpy(tcpd->objv,objv,sizeof(robj*)*objc); memcpy(tcpd->objv,objv,sizeof(robj*)*objc);
tcpd->objc = objc; tcpd->objc = objc;
tcpd->freecount = freecount; tcpd->freecount = freecount;
tcpd->options = options;
bc->privdata = tcpd; bc->privdata = tcpd;
/* We need the thread to work on a copy of the client argument /* We need the thread to work on a copy of the client argument
...@@ -7303,7 +7316,7 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob ...@@ -7303,7 +7316,7 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob
{ {
RM_AbortBlock(bc); RM_AbortBlock(bc);
/* Execute the command synchronously if we can't spawn a thread.. */ /* Execute the command synchronously if we can't spawn a thread.. */
callback(c,tcpd->objv,tcpd->objc); callback(c,tcpd->objv,tcpd->objc,tcpd->options);
threadedCoreCommandFreePrivdata(&ctx,tcpd); threadedCoreCommandFreePrivdata(&ctx,tcpd);
} }
moduleFreeContext(&ctx); moduleFreeContext(&ctx);
......
...@@ -1582,8 +1582,8 @@ int moduleClientIsBlockedOnKeys(client *c); ...@@ -1582,8 +1582,8 @@ int moduleClientIsBlockedOnKeys(client *c);
void moduleNotifyUserChanged(client *c); void moduleNotifyUserChanged(client *c);
/* Modules functionalities exported to core commands. */ /* Modules functionalities exported to core commands. */
typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc); typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc, void *options);
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount); void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount, void *options);
unsigned long runningThreadedCommandsCount(void); unsigned long runningThreadedCommandsCount(void);
/* Utils */ /* Utils */
......
...@@ -494,9 +494,18 @@ void stralgoCommand(client *c) { ...@@ -494,9 +494,18 @@ void stralgoCommand(client *c) {
} }
} }
/* Options passed by the main-thread half of STRALGO LCS command, to the
* threaded half. */
struct LCSOptions {
long long minmatchlen;
int getlen;
int getidx;
int withmatchlen;
};
/* This implements the threaded part of STRALGO LCS. It's the callback /* This implements the threaded part of STRALGO LCS. It's the callback
* we provide to the background execution engine. */ * we provide to the background execution engine. */
void stralgoLCSThreadedPart(client *c, robj **objv, int objc) { void stralgoLCSThreadedPart(client *c, robj **objv, int objc, void *options) {
uint32_t i, j; uint32_t i, j;
UNUSED(objc); UNUSED(objc);
...@@ -505,11 +514,7 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) { ...@@ -505,11 +514,7 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) {
sds a = objv[0]->ptr, b = objv[1]->ptr; sds a = objv[0]->ptr, b = objv[1]->ptr;
uint32_t alen = sdslen(a); uint32_t alen = sdslen(a);
uint32_t blen = sdslen(b); uint32_t blen = sdslen(b);
struct LCSOptions *opt = options;
/* FIXME: really populate these values with data in
* objv. */
long long minmatchlen = 0;
int getlen = 0, getidx = 0, withmatchlen = 0;
/* Setup an uint32_t array to store at LCS[i,j] the length of the /* Setup an uint32_t array to store at LCS[i,j] the length of the
* LCS A0..i-1, B0..j-1. Note that we have a linear array here, so * LCS A0..i-1, B0..j-1. Note that we have a linear array here, so
...@@ -552,12 +557,12 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) { ...@@ -552,12 +557,12 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) {
brange_end = 0; brange_end = 0;
/* Do we need to compute the actual LCS string? Allocate it in that case. */ /* Do we need to compute the actual LCS string? Allocate it in that case. */
int computelcs = getidx || !getlen; int computelcs = opt->getidx || !opt->getlen;
if (computelcs) result = sdsnewlen(SDS_NOINIT,idx); if (computelcs) result = sdsnewlen(SDS_NOINIT,idx);
/* Start with a deferred array if we have to emit the ranges. */ /* Start with a deferred array if we have to emit the ranges. */
uint32_t arraylen = 0; /* Number of ranges emitted in the array. */ uint32_t arraylen = 0; /* Number of ranges emitted in the array. */
if (getidx) { if (opt->getidx) {
addReplyMapLen(c,2); addReplyMapLen(c,2);
addReplyBulkCString(c,"matches"); addReplyBulkCString(c,"matches");
arraylenptr = addReplyDeferredLen(c); arraylenptr = addReplyDeferredLen(c);
...@@ -606,16 +611,16 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) { ...@@ -606,16 +611,16 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) {
/* Emit the current range if needed. */ /* Emit the current range if needed. */
uint32_t match_len = arange_end - arange_start + 1; uint32_t match_len = arange_end - arange_start + 1;
if (emit_range) { if (emit_range) {
if (minmatchlen == 0 || match_len >= minmatchlen) { if (opt->minmatchlen == 0 || match_len >= opt->minmatchlen) {
if (arraylenptr) { if (arraylenptr) {
addReplyArrayLen(c,2+withmatchlen); addReplyArrayLen(c,2+opt->withmatchlen);
addReplyArrayLen(c,2); addReplyArrayLen(c,2);
addReplyLongLong(c,arange_start); addReplyLongLong(c,arange_start);
addReplyLongLong(c,arange_end); addReplyLongLong(c,arange_end);
addReplyArrayLen(c,2); addReplyArrayLen(c,2);
addReplyLongLong(c,brange_start); addReplyLongLong(c,brange_start);
addReplyLongLong(c,brange_end); addReplyLongLong(c,brange_end);
if (withmatchlen) addReplyLongLong(c,match_len); if (opt->withmatchlen) addReplyLongLong(c,match_len);
arraylen++; arraylen++;
} }
} }
...@@ -628,7 +633,7 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) { ...@@ -628,7 +633,7 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) {
addReplyBulkCString(c,"len"); addReplyBulkCString(c,"len");
addReplyLongLong(c,LCS(alen,blen)); addReplyLongLong(c,LCS(alen,blen));
setDeferredArrayLen(c,arraylenptr,arraylen); setDeferredArrayLen(c,arraylenptr,arraylen);
} else if (getlen) { } else if (opt->getlen) {
addReplyLongLong(c,LCS(alen,blen)); addReplyLongLong(c,LCS(alen,blen));
} else { } else {
addReplyBulkSds(c,result); addReplyBulkSds(c,result);
...@@ -644,25 +649,29 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) { ...@@ -644,25 +649,29 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc) {
/* STRALGO LCS [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN] /* STRALGO LCS [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN]
* STRINGS <string> <string> | KEYS <keya> <keyb> */ * STRINGS <string> <string> | KEYS <keya> <keyb> */
void stralgoLCS(client *c) { void stralgoLCS(client *c) {
long long minmatchlen = 0; struct LCSOptions *opt = zmalloc(sizeof(*opt));
int getlen = 0, getidx = 0, withmatchlen = 0; opt->minmatchlen = 0;
opt->getlen = 0;
opt->getidx = 0;
opt->withmatchlen = 0;
int data_from_key = 0; int data_from_key = 0;
robj *obja = NULL, *objb = NULL; robj *obja = NULL, *objb = NULL;
for (int j = 2; j < c->argc; j++) { for (int j = 2; j < c->argc; j++) {
char *opt = c->argv[j]->ptr; char *optname = c->argv[j]->ptr;
int moreargs = (c->argc-1) - j; int moreargs = (c->argc-1) - j;
if (!strcasecmp(opt,"IDX")) { if (!strcasecmp(optname,"IDX")) {
getidx = 1; opt->getidx = 1;
} else if (!strcasecmp(opt,"LEN")) { } else if (!strcasecmp(optname,"LEN")) {
getlen = 1; opt->getlen = 1;
} else if (!strcasecmp(opt,"WITHMATCHLEN")) { } else if (!strcasecmp(optname,"WITHMATCHLEN")) {
withmatchlen = 1; opt->withmatchlen = 1;
} else if (!strcasecmp(opt,"MINMATCHLEN") && moreargs) { } else if (!strcasecmp(optname,"MINMATCHLEN") && moreargs) {
if (getLongLongFromObjectOrReply(c,c->argv[j+1],&minmatchlen,NULL) if (getLongLongFromObjectOrReply(c,c->argv[j+1],
!= C_OK) return; &opt->minmatchlen,NULL) != C_OK) return;
if (minmatchlen < 0) minmatchlen = 0; if (opt->minmatchlen < 0) opt->minmatchlen = 0;
j++; j++;
} else if (!strcasecmp(opt,"STRINGS") && moreargs > 1) { } else if (!strcasecmp(opt,"STRINGS") && moreargs > 1) {
if (obja != NULL) { if (obja != NULL) {
...@@ -698,7 +707,7 @@ void stralgoLCS(client *c) { ...@@ -698,7 +707,7 @@ void stralgoLCS(client *c) {
addReplyError(c,"Please specify two strings: " addReplyError(c,"Please specify two strings: "
"STRINGS or KEYS options are mandatory"); "STRINGS or KEYS options are mandatory");
return; return;
} else if (getlen && getidx) { } else if (opt->getlen && opt->getidx) {
addReplyError(c, addReplyError(c,
"If you want both the length and indexes, please " "If you want both the length and indexes, please "
"just use IDX."); "just use IDX.");
...@@ -713,6 +722,6 @@ void stralgoLCS(client *c) { ...@@ -713,6 +722,6 @@ void stralgoLCS(client *c) {
decrRefCount(objb); decrRefCount(objb);
robj *objv[2] = {a,b}; robj *objv[2] = {a,b};
executeThreadedCommand(c, stralgoLCSThreadedPart,objv,2, executeThreadedCommand(c, stralgoLCSThreadedPart,objv,2,
data_from_key?2:0); data_from_key?2:0, opt);
} }
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