gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor scan to use client::asyncCommand

Browse Source 
Former-commit-id: 4de596631f48626b770d0217c7ff21001ea46bcf
This commit is contained in:
malavan 2021-08-26 13:55:32 +00:00
parent d658891bf0
commit 7e10c9e9c3
1 changed files with 18 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
src/db.cpp
View File

@ -1183,17 +1183,10 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
if (o == nullptr && count >= 100)
{
// Do an async version
const redisDbPersistentDataSnapshot *snapshot = nullptr;
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
snapshot = c->db->createSnapshot(c->mvccCheckpoint, false /* fOptional */);
if (snapshot != nullptr)
{
aeEventLoop *el = serverTL->el;
blockClient(c, BLOCKED_ASYNC);
redisDb *db = c->db;
sds patCopy = pat ? sdsdup(pat) : nullptr;
sds typeCopy = type ? sdsdup(type) : nullptr;
g_pserver->asyncworkqueue->AddWorkFunction([c, snapshot, cursor, count, keys, el, db, patCopy, typeCopy, use_pattern]{
if (c->asyncCommand(
[c, cursor, keys, pat, type, use_pattern, count] (const redisDbPersistentDataSnapshot * snapshot) {
sds patCopy = pat ? sdsdup(pat) : nullptr;
sds typeCopy = type ? sdsdup(type) : nullptr;
auto cursorResult = snapshot->scan_threadsafe(cursor, count, typeCopy, keys);
if (use_pattern) {
listNode *ln = listFirst(keys);
@ -1214,30 +1207,20 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
sdsfree(patCopy);
if (typeCopy != nullptr)
sdsfree(typeCopy);
aePostFunction(el, [c, snapshot, keys, db, cursorResult, use_pattern]{
aeReleaseLock(); // we need to lock with coordination of the client
std::unique_lock<decltype(c->lock)> lock(c->lock);
AeLocker locker;
locker.arm(c);
unblockClient(c);
mstime_t timeScanFilter;
latencyStartMonitor(timeScanFilter);
scanFilterAndReply(c, keys, nullptr, nullptr, false, nullptr, cursorResult);
latencyEndMonitor(timeScanFilter);
latencyAddSampleIfNeeded("scan-async-filter", timeScanFilter);
locker.disarm();
lock.unlock();
db->endSnapshotAsync(snapshot);
listSetFreeMethod(keys,decrRefCountVoid);
listRelease(keys);
aeAcquireLock();
});
});
return (void *)cursorResult;
},
[c, keys] (const redisDbPersistentDataSnapshot * snapshot, void *data) {
mstime_t timeScanFilter;
latencyStartMonitor(timeScanFilter);
scanFilterAndReply(c, keys, nullptr, nullptr, false, nullptr, (unsigned long)data);
latencyEndMonitor(timeScanFilter);
latencyAddSampleIfNeeded("scan-async-filter", timeScanFilter);
},
[keys] (const redisDbPersistentDataSnapshot * snapshot, void *data) {
listSetFreeMethod(keys,decrRefCountVoid);
listRelease(keys);
}
)) {
return;
}
}