Commit 840932fb authored by antirez's avatar antirez
Browse files

TCC: Threaded SUNION / SDIFF.

parent 1dbe1eae
...@@ -928,30 +928,28 @@ void sinterstoreCommand(client *c) { ...@@ -928,30 +928,28 @@ void sinterstoreCommand(client *c) {
#define SET_OP_DIFF 1 #define SET_OP_DIFF 1
#define SET_OP_INTER 2 #define SET_OP_INTER 2
void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, struct sunionDiffThreadOptions {
robj *dstkey, int op) { int op;
robj **sets = zmalloc(sizeof(robj*)*setnum); int setnum;
int threaded;
robj **sets;
robj *dstkey; /* Only used if threaded is false. */
};
void sunionDiffThreadedHalf(client *c, void *options) {
struct sunionDiffThreadOptions *opt = options;
robj **sets = opt->sets;
int setnum = opt->setnum;
int op = opt->op;
int threaded = opt->threaded;
robj *dstkey = opt->dstkey;
setTypeIterator *si; setTypeIterator *si;
robj *dstset = NULL; robj *dstset = NULL;
sds ele; sds ele;
int j, cardinality = 0; int j, cardinality = 0;
int diff_algo = 1; int diff_algo = 1;
for (j = 0; j < setnum; j++) {
robj *setobj = dstkey ?
lookupKeyWrite(c->db,setkeys[j]) :
lookupKeyRead(c->db,setkeys[j]);
if (!setobj) {
sets[j] = NULL;
continue;
}
if (checkType(c,setobj,OBJ_SET)) {
zfree(sets);
return;
}
sets[j] = setobj;
}
/* Select what DIFF algorithm to use. /* Select what DIFF algorithm to use.
* *
* Algorithm 1 is O(N*M) where N is the size of the element first set * Algorithm 1 is O(N*M) where N is the size of the element first set
...@@ -1064,7 +1062,9 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1064,7 +1062,9 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
sdsfree(ele); sdsfree(ele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
server.lazyfree_lazy_server_del ? freeObjAsync(dstset) : if (server.lazyfree_lazy_server_del && !threaded)
freeObjAsync(dstset);
else
decrRefCount(dstset); decrRefCount(dstset);
} else { } else {
/* If we have a target key where to store the resulting set /* If we have a target key where to store the resulting set
...@@ -1087,6 +1087,52 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1087,6 +1087,52 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
server.dirty++; server.dirty++;
} }
zfree(sets); zfree(sets);
zfree(opt);
}
void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
robj *dstkey, int op) {
robj **sets = zmalloc(sizeof(robj*)*setnum);
int threaded = (dstkey == NULL);
for (int j = 0; j < setnum; j++) {
robj *setobj = dstkey ?
lookupKeyWrite(c->db,setkeys[j]) :
lookupKeyRead(c->db,setkeys[j]);
if (!setobj) {
sets[j] = NULL;
continue;
}
if (checkType(c,setobj,OBJ_SET)) {
zfree(sets);
return;
}
/* Try to lock the key if this is a threaded execution. If we
* fail we just revert to non threaded execution. */
if (threaded) {
if (lockKey(c,setkeys[j],LOCKEDKEY_READ,&setobj) == C_ERR) {
unlockAllKeys(c);
threaded = 0;
}
}
sets[j] = setobj;
}
struct sunionDiffThreadOptions *opt = zmalloc(sizeof(*opt));
opt->op = op;
opt->setnum = setnum;
opt->sets = sets;
opt->dstkey = dstkey;
opt->threaded = threaded;
/* TODO: Execute in threaded mode only if we have at least a
* large set. */
if (threaded) {
executeThreadedCommand(c,sunionDiffThreadedHalf,opt);
} else {
sunionDiffThreadedHalf(c,opt);
}
} }
void sunionCommand(client *c) { void sunionCommand(client *c) {
......
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