Commit 9c373b26 authored by antirez's avatar antirez
Browse files

Threaded core commands: API simplification.

parent 34dcff0a
...@@ -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, void *options); typedef void (*coreThreadedCommandCallback)(client *c, 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
...@@ -7218,9 +7218,6 @@ typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc, vo ...@@ -7218,9 +7218,6 @@ typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc, vo
typedef struct { typedef struct {
RedisModuleBlockedClient *bc; /* Blocked client handle. */ RedisModuleBlockedClient *bc; /* Blocked client handle. */
coreThreadedCommandCallback callback; /* Threaded command callback. */ coreThreadedCommandCallback callback; /* Threaded command callback. */
robj **objv; /* Vector of Redis objects. */
int objc; /* Number of objects. */
int freecount; /* How many we need to free. */
void *options; /* A pointer with details about void *options; /* A pointer with details about
the command to execute. Passed the command to execute. Passed
to the callback. */ to the callback. */
...@@ -7230,12 +7227,7 @@ typedef struct { ...@@ -7230,12 +7227,7 @@ typedef struct {
* threaded Redis core command execution. */ * threaded Redis core command execution. */
void threadedCoreCommandFreePrivdata(RedisModuleCtx *ctx, void *privdata) { void threadedCoreCommandFreePrivdata(RedisModuleCtx *ctx, void *privdata) {
UNUSED(ctx); UNUSED(ctx);
tcprivdata *tcpd = privdata; zfree(privdata);
for (int j = 0; j < tcpd->freecount; j++)
decrRefCount(tcpd->objv[j]);
zfree(tcpd->objv);
zfree(tcpd->options);
zfree(tcpd);
CoreModuleBlockedClients--; CoreModuleBlockedClients--;
} }
...@@ -7249,7 +7241,7 @@ void *threadedCoreCommandEnty(void *argptr) { ...@@ -7249,7 +7241,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->options); tcpd->callback(tcpd->bc->reply_client,tcpd->options);
moduleUnblockClientByHandle(tcpd->bc,tcpd); moduleUnblockClientByHandle(tcpd->bc,tcpd);
return NULL; return NULL;
} }
...@@ -7265,21 +7257,16 @@ void *threadedCoreCommandEnty(void *argptr) { ...@@ -7265,21 +7257,16 @@ void *threadedCoreCommandEnty(void *argptr) {
* the threaded part of a command can also operate on key values without * the threaded part of a command can also operate on key values without
* any race condition. * any race condition.
* *
* The objv and objc parameters are used in order to pass objects to the
* thread that will execute the command. They must be either copies of
* 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
* conditions. At the end of the execution of the command, the first
* 'freecount' objects of the objv array will be freed.
*
* 'options' is some private information passed to the callback that * 'options' is some private information passed to the callback that
* may contain parameters that the "main thread half" if the command * may contain parameters that the "main thread half" if the command
* parsed from the command line of the command. Normally it will pass * parsed from the command line of the command. Normally it will pass
* things like the options in order to execute the command, like modifiers * things like the options in order to execute the command, the objects
* of the return value or alike. The pointer pointed by 'options' must * the command should work with (that may be copies or made inaccessible
* be allocated in the stack with zmalloc() and will be free automatically * by the main thread via other means), and so forth.
* at the end of the execution. It can be NULL. */ *
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount, void *options) { * The info pointed by 'options' must be allocated by the callback implementing
* the threaded half of the command once it's done. */
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options) {
RedisModuleCtx ctx = REDISMODULE_CTX_INIT; RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
ctx.module = coremodule; ctx.module = coremodule;
ctx.client = c; ctx.client = c;
...@@ -7293,10 +7280,6 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob ...@@ -7293,10 +7280,6 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob
RedisModuleBlockedClient *bc = RM_BlockClient(&ctx,NULL,NULL,threadedCoreCommandFreePrivdata,0); RedisModuleBlockedClient *bc = RM_BlockClient(&ctx,NULL,NULL,threadedCoreCommandFreePrivdata,0);
tcpd->bc = bc; tcpd->bc = bc;
tcpd->callback = callback; tcpd->callback = callback;
tcpd->objv = zmalloc(sizeof(robj*)*objc);
memcpy(tcpd->objv,objv,sizeof(robj*)*objc);
tcpd->objc = objc;
tcpd->freecount = freecount;
tcpd->options = options; tcpd->options = options;
bc->privdata = tcpd; bc->privdata = tcpd;
...@@ -7316,7 +7299,7 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob ...@@ -7316,7 +7299,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,tcpd->options); callback(c,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, void *options); typedef void (*coreThreadedCommandCallback)(client *c, void *options);
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount, void *options); void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options);
unsigned long runningThreadedCommandsCount(void); unsigned long runningThreadedCommandsCount(void);
/* Utils */ /* Utils */
......
...@@ -497,6 +497,7 @@ void stralgoCommand(client *c) { ...@@ -497,6 +497,7 @@ void stralgoCommand(client *c) {
/* Options passed by the main-thread half of STRALGO LCS command, to the /* Options passed by the main-thread half of STRALGO LCS command, to the
* threaded half. */ * threaded half. */
struct LCSOptions { struct LCSOptions {
robj *obja, *objb;
long long minmatchlen; long long minmatchlen;
int getlen; int getlen;
int getidx; int getidx;
...@@ -505,16 +506,15 @@ struct LCSOptions { ...@@ -505,16 +506,15 @@ struct LCSOptions {
/* 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 *options) { void stralgoLCSThreadedPart(client *c, void *options) {
uint32_t i, j; uint32_t i, j;
UNUSED(objc);
/* Compute the LCS using the vanilla dynamic programming technique of /* Compute the LCS using the vanilla dynamic programming technique of
* building a table of LCS(x,y) substrings. */ * building a table of LCS(x,y) substrings. */
sds a = objv[0]->ptr, b = objv[1]->ptr; struct LCSOptions *opt = options;
sds a = opt->obja->ptr, b = opt->objb->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;
/* 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
...@@ -643,6 +643,9 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc, void *options) { ...@@ -643,6 +643,9 @@ void stralgoLCSThreadedPart(client *c, robj **objv, int objc, void *options) {
/* Cleanup. */ /* Cleanup. */
sdsfree(result); sdsfree(result);
zfree(lcs); zfree(lcs);
decrRefCount(opt->obja);
decrRefCount(opt->objb);
zfree(opt);
return; return;
} }
...@@ -720,8 +723,8 @@ void stralgoLCS(client *c) { ...@@ -720,8 +723,8 @@ void stralgoLCS(client *c) {
robj *b = dupStringObject(objb); robj *b = dupStringObject(objb);
decrRefCount(obja); decrRefCount(obja);
decrRefCount(objb); decrRefCount(objb);
robj *objv[2] = {a,b}; opt->obja = a;
executeThreadedCommand(c, stralgoLCSThreadedPart,objv,2, opt->objb = b;
data_from_key?2:0, opt); executeThreadedCommand(c, stralgoLCSThreadedPart, 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