Commit 80efd2e8 authored by antirez's avatar antirez
Browse files

Threaded core commands: WIP #1.

parent 66ade863
......@@ -372,6 +372,7 @@ typedef struct RedisModuleUser {
/* The Redis core itself registeres a module in order to use module services
* for things like blocking clients for threaded execution of slow commands. */
RedisModule *coremodule;
_Atomic unsigned long CoreModuleBlockedClients = 0;
/* --------------------------------------------------------------------------
* Prototypes
......@@ -4610,6 +4611,7 @@ void RM_SetDisconnectCallback(RedisModuleBlockedClient *bc, RedisModuleDisconnec
* When this happens the RedisModuleBlockedClient structure in the queue
* will have the 'client' field set to NULL. */
void moduleHandleBlockedClients(void) {
if (moduleCount() == 0 && CoreModuleBlockedClients == 0) return;
listNode *ln;
RedisModuleBlockedClient *bc;
......@@ -7227,7 +7229,9 @@ void threadedCoreCommandFreePrivdata(RedisModuleCtx *ctx, void *privdata) {
tcprivdata *tcpd = privdata;
for (int j = 0; j < tcpd->freecount; j++)
decrRefCount(tcpd->objv[j]);
zfree(tcpd->objv);
zfree(tcpd);
CoreModuleBlockedClients--;
}
/* The entry point of the threaded execution. */
......@@ -7272,10 +7276,12 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob
tcprivdata *tcpd = zmalloc(sizeof(*tcpd));
/* Block the client. */
CoreModuleBlockedClients++;
RedisModuleBlockedClient *bc = RM_BlockClient(&ctx,NULL,NULL,threadedCoreCommandFreePrivdata,0);
tcpd->bc = bc;
tcpd->callback = callback;
tcpd->objv = objv;
tcpd->objv = zmalloc(sizeof(robj*)*objc);
memcpy(tcpd->objv,objv,sizeof(robj*)*objc);
tcpd->objc = objc;
tcpd->freecount = freecount;
bc->privdata = tcpd;
......@@ -7285,7 +7291,7 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, rob
client *copy = bc->reply_client;
copy->argc = c->argc;
copy->argv = zmalloc(sizeof(robj*) * c->argc);
for (int j = 0; j < c->argc; c++)
for (int j = 0; j < c->argc; j++)
copy->argv[j] = createStringObject(c->argv[j]->ptr,
sdslen(c->argv[j]->ptr));
......
......@@ -2151,7 +2151,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* Check if there are clients unblocked by modules that implement
* blocking commands. */
if (moduleCount()) moduleHandleBlockedClients();
moduleHandleBlockedClients();
/* Try to process pending commands for clients that were just unblocked. */
if (listLength(server.unblocked_clients))
......
......@@ -1581,6 +1581,10 @@ void moduleUnblockClient(client *c);
int moduleClientIsBlockedOnKeys(client *c);
void moduleNotifyUserChanged(client *c);
/* Modules functionalities exported to core commands. */
typedef void (*coreThreadedCommandCallback)(client *c, robj **objv, int objc);
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, robj **objv, int objc, int freecount);
/* Utils */
long long ustime(void);
long long mstime(void);
......
......@@ -494,73 +494,23 @@ void stralgoCommand(client *c) {
}
}
/* STRALGO LCS [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN]
* STRINGS <string> <string> | KEYS <keya> <keyb> */
void stralgoLCS(client *c) {
/* This implements the threaded part of STRALGO LCS. It's the callback
* we provide to the background execution engine. */
void stralgoLCSThreadedPart(client *c, robj **objv, int objc) {
uint32_t i, j;
long long minmatchlen = 0;
sds a = NULL, b = NULL;
int getlen = 0, getidx = 0, withmatchlen = 0;
robj *obja = NULL, *objb = NULL;
for (j = 2; j < (uint32_t)c->argc; j++) {
char *opt = c->argv[j]->ptr;
int moreargs = (c->argc-1) - j;
if (!strcasecmp(opt,"IDX")) {
getidx = 1;
} else if (!strcasecmp(opt,"LEN")) {
getlen = 1;
} else if (!strcasecmp(opt,"WITHMATCHLEN")) {
withmatchlen = 1;
} else if (!strcasecmp(opt,"MINMATCHLEN") && moreargs) {
if (getLongLongFromObjectOrReply(c,c->argv[j+1],&minmatchlen,NULL)
!= C_OK) return;
if (minmatchlen < 0) minmatchlen = 0;
j++;
} else if (!strcasecmp(opt,"STRINGS") && moreargs > 1) {
if (a != NULL) {
addReplyError(c,"Either use STRINGS or KEYS");
return;
}
a = c->argv[j+1]->ptr;
b = c->argv[j+2]->ptr;
j += 2;
} else if (!strcasecmp(opt,"KEYS") && moreargs > 1) {
if (a != NULL) {
addReplyError(c,"Either use STRINGS or KEYS");
return;
}
obja = lookupKeyRead(c->db,c->argv[j+1]);
objb = lookupKeyRead(c->db,c->argv[j+2]);
obja = obja ? getDecodedObject(obja) : createStringObject("",0);
objb = objb ? getDecodedObject(objb) : createStringObject("",0);
a = obja->ptr;
b = objb->ptr;
j += 2;
} else {
addReply(c,shared.syntaxerr);
return;
}
}
/* Complain if the user passed ambiguous parameters. */
if (a == NULL) {
addReplyError(c,"Please specify two strings: "
"STRINGS or KEYS options are mandatory");
return;
} else if (getlen && getidx) {
addReplyError(c,
"If you want both the length and indexes, please "
"just use IDX.");
return;
}
UNUSED(objc);
/* Compute the LCS using the vanilla dynamic programming technique of
* building a table of LCS(x,y) substrings. */
sds a = objv[0]->ptr, b = objv[1]->ptr;
uint32_t alen = sdslen(a);
uint32_t blen = sdslen(b);
/* 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
* LCS A0..i-1, B0..j-1. Note that we have a linear array here, so
* we index it as LCS[j+(blen+1)*j] */
......@@ -673,8 +623,6 @@ void stralgoLCS(client *c) {
}
}
/* Signal modified key, increment dirty, ... */
/* Reply depending on the given options. */
if (arraylenptr) {
addReplyBulkCString(c,"len");
......@@ -688,10 +636,83 @@ void stralgoLCS(client *c) {
}
/* Cleanup. */
if (obja) decrRefCount(obja);
if (objb) decrRefCount(objb);
sdsfree(result);
zfree(lcs);
return;
}
/* STRALGO LCS [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN]
* STRINGS <string> <string> | KEYS <keya> <keyb> */
void stralgoLCS(client *c) {
long long minmatchlen = 0;
int getlen = 0, getidx = 0, withmatchlen = 0;
int data_from_key = 0;
robj *obja = NULL, *objb = NULL;
for (int j = 2; j < c->argc; j++) {
char *opt = c->argv[j]->ptr;
int moreargs = (c->argc-1) - j;
if (!strcasecmp(opt,"IDX")) {
getidx = 1;
} else if (!strcasecmp(opt,"LEN")) {
getlen = 1;
} else if (!strcasecmp(opt,"WITHMATCHLEN")) {
withmatchlen = 1;
} else if (!strcasecmp(opt,"MINMATCHLEN") && moreargs) {
if (getLongLongFromObjectOrReply(c,c->argv[j+1],&minmatchlen,NULL)
!= C_OK) return;
if (minmatchlen < 0) minmatchlen = 0;
j++;
} else if (!strcasecmp(opt,"STRINGS") && moreargs > 1) {
if (obja != NULL) {
addReplyError(c,"Either use STRINGS or KEYS");
return;
}
obja = c->argv[j+1];
objb = c->argv[j+2];
/* We increment the objects refcount only to make the cleanup
* code path the same as the KEYS option. */
incrRefCount(obja);
incrRefCount(objb);
j += 2;
} else if (!strcasecmp(opt,"KEYS") && moreargs > 1) {
if (obja != NULL) {
addReplyError(c,"Either use STRINGS or KEYS");
return;
}
obja = lookupKeyRead(c->db,c->argv[j+1]);
objb = lookupKeyRead(c->db,c->argv[j+2]);
obja = obja ? getDecodedObject(obja) : createStringObject("",0);
objb = objb ? getDecodedObject(objb) : createStringObject("",0);
j += 2;
data_from_key = 1;
} else {
addReply(c,shared.syntaxerr);
return;
}
}
/* Complain if the user passed ambiguous parameters. */
if (obja == NULL) {
addReplyError(c,"Please specify two strings: "
"STRINGS or KEYS options are mandatory");
return;
} else if (getlen && getidx) {
addReplyError(c,
"If you want both the length and indexes, please "
"just use IDX.");
return;
}
/* Switch to the threaded execution of the command. We always work on
* a copy of the data. */
robj *a = dupStringObject(obja);
robj *b = dupStringObject(objb);
decrRefCount(obja);
decrRefCount(objb);
robj *objv[2] = {a,b};
executeThreadedCommand(c, stralgoLCSThreadedPart,objv,2,
data_from_key?2:0);
}
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