Commit f302c75f authored by antirez's avatar antirez
Browse files

THreaded core commands: key unlocking WIP.

parent 426b8643
...@@ -155,6 +155,8 @@ void unblockClient(client *c) { ...@@ -155,6 +155,8 @@ void unblockClient(client *c) {
} else if (btype == BLOCKED_MODULE) { } else if (btype == BLOCKED_MODULE) {
if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c); if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c);
unblockClientFromModule(c); unblockClientFromModule(c);
} else if (btype == BLOCKED_LOCK) {
/* We'll handle it at the end of the function. */
} else { } else {
serverPanic("Unknown btype in unblockClient()."); serverPanic("Unknown btype in unblockClient().");
} }
......
...@@ -1777,10 +1777,10 @@ int lockedClientListMatch(void *a, void *b) { ...@@ -1777,10 +1777,10 @@ int lockedClientListMatch(void *a, void *b) {
* Signaling of the need to lock keys is performed using the * Signaling of the need to lock keys is performed using the
* wouldLockKey() API (XXX: yet to be implemented). Check the function * wouldLockKey() API (XXX: yet to be implemented). Check the function
* documentation for more information. */ * documentation for more information. */
int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) { int lockKey(client *c, robj *key, int locktype, robj **optr) {
dictEntry *de; dictEntry *de;
de = dictFind(db->locked_keys,key); de = dictFind(c->db->locked_keys,key);
lockedKey *lk = de ? dictGetVal(de) : NULL; lockedKey *lk = de ? dictGetVal(de) : NULL;
if (lk == NULL) { if (lk == NULL) {
...@@ -1792,7 +1792,9 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) { ...@@ -1792,7 +1792,9 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) {
listSetMatchMethod(lk->owners,lockedClientListMatch); listSetMatchMethod(lk->owners,lockedClientListMatch);
listSetFreeMethod(lk->waiting,zfree); listSetFreeMethod(lk->waiting,zfree);
listSetFreeMethod(lk->owners,zfree); listSetFreeMethod(lk->owners,zfree);
lk->obj = lookupKeyReadWithFlags(db,key,LOOKUP_NOTOUCH); lk->obj = lookupKeyReadWithFlags(c->db,key,LOOKUP_NOTOUCH);
dictAdd(c->db->locked_keys,key,lk);
incrRefCount(key);
} else { } else {
/* If there is already a lock, it is incompatible with a new lock /* If there is already a lock, it is incompatible with a new lock
* both in the case the lock is of write type, or we want to lock * both in the case the lock is of write type, or we want to lock
...@@ -1801,12 +1803,7 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) { ...@@ -1801,12 +1803,7 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) {
return C_ERR; return C_ERR;
} }
/* Lock allowed: put the client in the list of owners. */ /* We need to remember that this client locked this key: for
lockedKeyClient *lkc = zmalloc(sizeof(*lkc));
lkc->id = c->id;
listAddNodeTail(lk->owners,lkc);
/* We also need to remember that this client locked this key: for
* now we set this information in the real client locking the key, * now we set this information in the real client locking the key,
* however later this information is moved into the blocked client * however later this information is moved into the blocked client
* handle that we use for threaded execution. The reason is that we * handle that we use for threaded execution. The reason is that we
...@@ -1816,11 +1813,25 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) { ...@@ -1816,11 +1813,25 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) {
c->locked = dictCreate(&objectKeyPointerValueDictType,NULL); c->locked = dictCreate(&objectKeyPointerValueDictType,NULL);
de = dictAddRaw(c->locked,key,NULL); de = dictAddRaw(c->locked,key,NULL);
/* Remember the client ID of the lock, so if we move this in the if (de == NULL) {
/* The key is already locked for this client? Odd but let's allow
* that without crashing. So that command implementations don't
* have to do complex checks in the command line in case of repeating
* keys. */
*optr = lk->obj;
return C_OK;
}
incrRefCount(key);
/* Lock allowed: put the client in the list of owners. */
lockedKeyClient *lkc = zmalloc(sizeof(*lkc));
lkc->id = c->id;
listAddNodeTail(lk->owners,lkc);
/* Remember the client ID of the lock, so if we move this in the
* blocked handle, we can still track the original ID even if the * blocked handle, we can still track the original ID even if the
* client is gone. */ * client is gone. */
dictSetUnsignedIntegerVal(de,c->id); dictSetUnsignedIntegerVal(de,c->id);
incrRefCount(key);
*optr = lk->obj; *optr = lk->obj;
return C_OK; return C_OK;
} }
...@@ -1837,3 +1848,66 @@ int queueClientIfKeyIsLocked(client *c, robj *key) { ...@@ -1837,3 +1848,66 @@ int queueClientIfKeyIsLocked(client *c, robj *key) {
return C_OK; return C_OK;
} }
/* Unlock the specified key in the context of the specified client.
* This API should not be called by the command implementation itself,
* instead unlockAllKeys() should be used. */
void unlockKey(client *c, robj *key, uint64_t owner_id) {
lockedKey *lk = dictFetchValue(c->db->locked_keys,key);
if (lk == NULL) return;
struct lockedKeyClient lkc = {owner_id};
listNode *ln = listSearchKey(lk->owners,&lkc);
if (ln == NULL) return;
/* Unlock the key, and if this key dropped to a locked count of
* zero, we need to resume the clients in the waiting list. */
listDelNode(lk->owners,ln);
if (listLength(lk->owners) != 0) return;
listIter li;
listRewind(lk->waiting,&li);
while((ln = listNext(&li))) {
lockedKeyClient *lkc = listNodeValue(ln);
client *blocked = lookupClientByID(lkc->id);
if (blocked != NULL &&
blocked->flags & CLIENT_BLOCKED &&
blocked->btype == BLOCKED_LOCK)
{
printf("UNBLOCK %llu\n", blocked->id);
/* XXX: Check if the client is yet locked in some other key
* before unblocking it? Right now we just unblock it, and
* it will get blocked again if there are yet keys that are
* locked for it. */
unblockClient(blocked);
}
listDelNode(lk->waiting,ln);
}
/* At this point the key is no longer locked by any owner, nor
* it has any client in the waiting list. Remove it from the
* dictionary. */
dictDelete(c->db->locked_keys,key);
listRelease(lk->owners);
listRelease(lk->waiting);
zfree(lk);
}
/* Unlock all the keys locked in the specified client. We need to pass
* the client ID of the original client that locked the key, since the
* list of locked keys may be transfered to the blocked client handle.
* As a side effect the function also will release the dictionary of
* locked keys. */
void unlockAllKeys(client *c) {
if (c->locked == NULL) return;
dictIterator *di;
dictEntry *de;
di = dictGetSafeIterator(c->locked);
while((de = dictNext(di)) != NULL) {
uint64_t owner_id = dictGetUnsignedIntegerVal(de);
robj *key = dictGetKey(de);
unlockKey(c,key,owner_id);
}
dictReleaseIterator(di);
dictRelease(c->locked);
c->locked = NULL;
}
...@@ -7227,9 +7227,9 @@ typedef struct { ...@@ -7227,9 +7227,9 @@ 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);
/* TODO: unlock the key here. This can be as simple as putting the tcprivdata *tcpd = privdata;
* locked key in an unlock queue or alike if we don't want to do
* it synchronously here. */ unlockAllKeys(tcpd->bc->reply_client);
zfree(privdata); zfree(privdata);
CoreModuleBlockedClients--; CoreModuleBlockedClients--;
} }
...@@ -7286,6 +7286,13 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, voi ...@@ -7286,6 +7286,13 @@ void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, voi
tcpd->options = options; tcpd->options = options;
bc->privdata = tcpd; bc->privdata = tcpd;
/* Move the list of locked keys in the reply client of the
* blocked handle: this way if the real client is freed because of
* disconnection, we still unlock the keys in the right moment, that
* is once the thread returns. */
bc->reply_client->locked = c->locked;
c->locked = NULL;
/* 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
* vector. */ * vector. */
client *copy = bc->reply_client; client *copy = bc->reply_client;
......
...@@ -1156,6 +1156,17 @@ void freeClient(client *c) { ...@@ -1156,6 +1156,17 @@ void freeClient(client *c) {
unwatchAllKeys(c); unwatchAllKeys(c);
listRelease(c->watched_keys); listRelease(c->watched_keys);
/* Unlock the keys and free the dictionary of locked keys.
* Note that if the client disconnects while there is a thread executing
* a threaded command for it, actually the locked keys were passed to
* the blocked client handle, so c->locked will be NULL, and the keys
* will really be unlocked when the thread returns instead.
*
* However we may get here with locked keys in case for some reason
* the client didn't call executeThreadedCommand() after locking the
* key. */
unlockAllKeys(c);
/* Unsubscribe from all the pubsub channels */ /* Unsubscribe from all the pubsub channels */
pubsubUnsubscribeAllChannels(c,0); pubsubUnsubscribeAllChannels(c,0);
pubsubUnsubscribeAllPatterns(c,0); pubsubUnsubscribeAllPatterns(c,0);
......
...@@ -1628,8 +1628,9 @@ void moduleNotifyUserChanged(client *c); ...@@ -1628,8 +1628,9 @@ void moduleNotifyUserChanged(client *c);
typedef void (*coreThreadedCommandCallback)(client *c, void *options); typedef void (*coreThreadedCommandCallback)(client *c, void *options);
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options); void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options);
unsigned long runningThreadedCommandsCount(void); unsigned long runningThreadedCommandsCount(void);
int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr); int lockKey(client *c, robj *key, int locktype, robj **optr);
int queueClientIfKeyIsLocked(client *c, robj *key); int queueClientIfKeyIsLocked(client *c, robj *key);
void unlockAllKeys(client *c);
/* Utils */ /* Utils */
long long ustime(void); long long ustime(void);
......
...@@ -395,6 +395,37 @@ void rpopCommand(client *c) { ...@@ -395,6 +395,37 @@ void rpopCommand(client *c) {
popGenericCommand(c,LIST_TAIL); popGenericCommand(c,LIST_TAIL);
} }
struct lrangeThreadOptions {
long start, rangelen;
robj *o;
};
void lrangeThreadedHalf(client *c, void *options) {
struct lrangeThreadOptions *opt = options;
/* Return the result in form of a multi-bulk reply */
addReplyArrayLen(c,opt->rangelen);
if (opt->o->encoding == OBJ_ENCODING_QUICKLIST) {
listTypeIterator *iter = listTypeInitIterator(opt->o, opt->start,
LIST_TAIL);
while(opt->rangelen--) {
listTypeEntry entry;
listTypeNext(iter, &entry);
quicklistEntry *qe = &entry.entry;
if (qe->value) {
addReplyBulkCBuffer(c,qe->value,qe->sz);
} else {
addReplyBulkLongLong(c,qe->longval);
}
}
listTypeReleaseIterator(iter);
} else {
serverPanic("List encoding is not QUICKLIST!");
}
zfree(options);
}
void lrangeCommand(client *c) { void lrangeCommand(client *c) {
robj *o; robj *o;
long start, end, llen, rangelen; long start, end, llen, rangelen;
...@@ -420,24 +451,15 @@ void lrangeCommand(client *c) { ...@@ -420,24 +451,15 @@ void lrangeCommand(client *c) {
if (end >= llen) end = llen-1; if (end >= llen) end = llen-1;
rangelen = (end-start)+1; rangelen = (end-start)+1;
/* Return the result in form of a multi-bulk reply */ struct lrangeThreadOptions *opt = zmalloc(sizeof(*opt));
addReplyArrayLen(c,rangelen); opt->start = start;
if (o->encoding == OBJ_ENCODING_QUICKLIST) { opt->rangelen = rangelen;
listTypeIterator *iter = listTypeInitIterator(o, start, LIST_TAIL); if (lockKey(c,c->argv[1],LOCKEDKEY_READ,&opt->o) == C_ERR) {
/* In the case of LRANGE, we execute the command synchronously
while(rangelen--) { * if we are unable to get a lock. */
listTypeEntry entry; lrangeThreadedHalf(c,opt);
listTypeNext(iter, &entry);
quicklistEntry *qe = &entry.entry;
if (qe->value) {
addReplyBulkCBuffer(c,qe->value,qe->sz);
} else {
addReplyBulkLongLong(c,qe->longval);
}
}
listTypeReleaseIterator(iter);
} else { } else {
serverPanic("List encoding is not QUICKLIST!"); executeThreadedCommand(c, lrangeThreadedHalf, opt);
} }
} }
......
...@@ -496,7 +496,7 @@ void stralgoCommand(client *c) { ...@@ -496,7 +496,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 LCSThreadOptions {
robj *obja, *objb; robj *obja, *objb;
long long minmatchlen; long long minmatchlen;
int getlen; int getlen;
...@@ -506,12 +506,12 @@ struct LCSOptions { ...@@ -506,12 +506,12 @@ 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, void *options) { void stralgoLCSThreadedHalf(client *c, void *options) {
uint32_t i, j; uint32_t i, j;
/* 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. */
struct LCSOptions *opt = options; struct LCSThreadOptions *opt = options;
sds a = opt->obja->ptr, b = opt->objb->ptr; 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);
...@@ -652,7 +652,7 @@ void stralgoLCSThreadedPart(client *c, void *options) { ...@@ -652,7 +652,7 @@ void stralgoLCSThreadedPart(client *c, void *options) {
/* 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) {
struct LCSOptions *opt = zmalloc(sizeof(*opt)); struct LCSThreadOptions *opt = zmalloc(sizeof(*opt));
opt->minmatchlen = 0; opt->minmatchlen = 0;
opt->getlen = 0; opt->getlen = 0;
opt->getidx = 0; opt->getidx = 0;
...@@ -676,7 +676,7 @@ void stralgoLCS(client *c) { ...@@ -676,7 +676,7 @@ void stralgoLCS(client *c) {
&opt->minmatchlen,NULL) != C_OK) return; &opt->minmatchlen,NULL) != C_OK) return;
if (opt->minmatchlen < 0) opt->minmatchlen = 0; if (opt->minmatchlen < 0) opt->minmatchlen = 0;
j++; j++;
} else if (!strcasecmp(opt,"STRINGS") && moreargs > 1) { } else if (!strcasecmp(optname,"STRINGS") && moreargs > 1) {
if (obja != NULL) { if (obja != NULL) {
addReplyError(c,"Either use STRINGS or KEYS"); addReplyError(c,"Either use STRINGS or KEYS");
goto syntaxerr; goto syntaxerr;
...@@ -688,7 +688,7 @@ void stralgoLCS(client *c) { ...@@ -688,7 +688,7 @@ void stralgoLCS(client *c) {
incrRefCount(obja); incrRefCount(obja);
incrRefCount(objb); incrRefCount(objb);
j += 2; j += 2;
} else if (!strcasecmp(opt,"KEYS") && moreargs > 1) { } else if (!strcasecmp(optname,"KEYS") && moreargs > 1) {
if (obja != NULL) { if (obja != NULL) {
addReplyError(c,"Either use STRINGS or KEYS"); addReplyError(c,"Either use STRINGS or KEYS");
goto syntaxerr; goto syntaxerr;
...@@ -725,7 +725,7 @@ void stralgoLCS(client *c) { ...@@ -725,7 +725,7 @@ void stralgoLCS(client *c) {
decrRefCount(objb); decrRefCount(objb);
opt->obja = a; opt->obja = a;
opt->objb = b; opt->objb = b;
executeThreadedCommand(c, stralgoLCSThreadedPart, opt); executeThreadedCommand(c, stralgoLCSThreadedHalf, opt);
return; return;
syntaxerr: syntaxerr:
......
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