Skip to content

fts: serialize concurrent mutations of QFTSEdgeCompute scores map #69

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adsharma merged 1 commit into from
Aug 28, 2026
Merged

fts: serialize concurrent mutations of QFTSEdgeCompute scores map #69

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions fts/src/function/query_fts_index.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
#include "function/query_fts_index.h"

#include <mutex>
#include <queue>

#include "binder/binder.h"
Expand Down Expand Up @@ -131,10 +132,18 @@ struct ScoreInfo {
struct QFTSEdgeCompute final : EdgeCompute {
node_id_map_t<ScoreInfo>& scores;
const std::unordered_map<offset_t, uint64_t>& dfs;
// The scores map is shared by reference between copies of this edge compute (one per
// worker thread in a parallel frontier task), so all mutations must be serialized.
std::shared_ptr<std::mutex> scoresMutex;

QFTSEdgeCompute(node_id_map_t<ScoreInfo>& scores,
const std::unordered_map<offset_t, uint64_t>& dfs)
: scores{scores}, dfs{dfs} {}
: scores{scores}, dfs{dfs}, scoresMutex{std::make_shared<std::mutex>()} {}

QFTSEdgeCompute(node_id_map_t<ScoreInfo>& scores,
const std::unordered_map<offset_t, uint64_t>& dfs,
std::shared_ptr<std::mutex> scoresMutex)
: scores{scores}, dfs{dfs}, scoresMutex{std::move(scoresMutex)} {}

std::vector<nodeID_t> edgeCompute(nodeID_t boundNodeID, graph::NbrScanState::Chunk& resultChunk,
bool) override {
Expand All @@ -143,6 +152,7 @@ struct QFTSEdgeCompute final : EdgeCompute {
std::vector<nodeID_t> activeNodes;
resultChunk.forEach([&](auto neighbors, auto propertyVectors, auto i) {
auto docNodeID = neighbors[i];
std::lock_guard guard{*scoresMutex};
if (!scores.contains(docNodeID)) {
scores.emplace(docNodeID, ScoreInfo{});
}
Expand All @@ -154,7 +164,7 @@ struct QFTSEdgeCompute final : EdgeCompute {
}

std::unique_ptr<EdgeCompute> copy() override {
return std::make_unique<QFTSEdgeCompute>(scores, dfs);
return std::make_unique<QFTSEdgeCompute>(scores, dfs, scoresMutex);
}
};

Expand Down
Loading