Method flushQueuedKeys

src/SnapshotPayloadParseState.cpp:130–182

}

void SnapshotPayloadParseState::flushQueuedKeys() {
    if (vecqueuedKeys.empty())
        return;
    serverAssert(current_database >= 0);

    // TODO: We can't finish parse until all the work functions finish
    int idb = current_database;
    serverAssert(vecqueuedKeys.size() == vecqueuedVals.size());
    auto sizePrev = vecqueuedKeys.size();
    if (current_database < cserver.dbnum) {
        if (g_pserver->m_pstorageFactory) {
            (*insertsInFlight)++;
            std::weak_ptr<std::atomic<int>> insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
            g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
                g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
                (*(insertsInFlightTmp.lock()))--;
                delete pallocator;
            });
        } else {
            for (size_t ival = 0; ival < vecqueuedKeys.size(); ++ival) {
                size_t offset = 0;
                auto spexpire = deserializeExpire(vecqueuedVals[ival], vecqueuedValsCb[ival], &offset);    
                auto o = deserializeStoredObject(vecqueuedVals[ival] + offset, vecqueuedValsCb[ival] - offset);
                sds sdsKey = sdsnewlen(vecqueuedKeys[ival], -static_cast<ssize_t>(vecqueuedKeysCb[ival]));
                if (dbMerge(g_pserver->db[idb], sdsKey, o, false)) {
                    if (spexpire != nullptr)
                        g_pserver->db[idb]->setExpire(sdsKey, std::move(*spexpire));
                } else {
                    sdsfree(sdsKey);
                }
            }
            vecqueuedKeys.clear();
            vecqueuedKeysCb.clear();
            vecqueuedVals.clear();
            vecqueuedValsCb.clear();
        }
    } else {
        // else drop the data
        vecqueuedKeys.clear();
        vecqueuedKeysCb.clear();
        vecqueuedVals.clear();
        vecqueuedValsCb.clear();
        // Note: m_spallocator will get free'd when overwritten below
    }
    m_spallocator = std::make_unique<SlabAllocator>();
    cbQueued = 0;
    vecqueuedKeys.reserve(sizePrev);
    vecqueuedKeysCb.reserve(sizePrev);
    vecqueuedVals.reserve(sizePrev);
    vecqueuedValsCb.reserve(sizePrev);
    serverAssert(vecqueuedKeys.empty());
    serverAssert(vecqueuedVals.empty());
}


SnapshotPayloadParseState::SnapshotPayloadParseState() {