Skip to content

Commit 12e89a6

Browse files
dangleptrheng
dangleptr
and
heng
authored
Fix the wrong usage for LRUCache (#2267)
* Fix the wrong usage for LRUCache * Overwrite the old value when calling insert * Address comments * Address liuyu's comments * Address critical27's comments Co-authored-by: heng <heng.chen@vesoft.com>
1 parent 0f8ce2e commit 12e89a6

File tree

4 files changed

+93
-77
lines changed

4 files changed

+93
-77
lines changed

‎src/common/base/ConcurrentLRUCache.h

+10-6
Original file line numberDiff line numberDiff line change
@@ -206,19 +206,23 @@ class LRU {
206206
}
207207

208208
void insert(key_type&& key, value_type&& value) {
209-
typename map_type::iterator i = map_.find(key);
210-
if (i == map_.end()) {
209+
typename map_type::iterator it = map_.find(key);
210+
if (it == map_.end()) {
211211
// insert item into the cache, but first check if it is full
212212
if (size() >= capacity_) {
213213
VLOG(3) << "Size:" << size() << ", capacity " << capacity_;
214214
// cache is full, evict the least recently used item
215215
evict();
216216
}
217217
VLOG(3) << "Insert key " << key << ", val " << value;
218-
// insert the new item
219218
list_.push_front(key);
220219
map_.emplace(std::forward<key_type>(key),
221220
std::make_tuple(std::forward<value_type>(value), list_.begin()));
221+
} else {
222+
// Overwrite the value
223+
std::get<0>(it->second) = std::move(value);
224+
typename list_type::iterator j = std::get<1>(it->second);
225+
list_.splice(list_.begin(), list_, j);
222226
}
223227
}
224228

@@ -296,9 +300,9 @@ class LRU {
296300
map_type map_;
297301
list_type list_;
298302
size_t capacity_;
299-
std::atomic_uint64_t total_{0};
300-
std::atomic_uint64_t hits_{0};
301-
std::atomic_uint64_t evicts_{0};
303+
uint64_t total_{0};
304+
uint64_t hits_{0};
305+
uint64_t evicts_{0};
302306
};
303307

304308
} // namespace nebula

‎src/common/base/test/ConcurrentLRUCacheTest.cpp

+15
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,21 @@ TEST(ConcurrentLRUCacheTest, MultiThreadsTest) {
139139
EXPECT_EQ(10000, cache.total());
140140
}
141141

142+
TEST(ConcurrentLRUCacheTest, OverwriteTest) {
143+
ConcurrentLRUCache<int32_t, std::string> cache(1024);
144+
cache.insert(10, "ten");
145+
{
146+
auto v = cache.get(10);
147+
EXPECT_TRUE(v.ok());
148+
EXPECT_EQ("ten", v.value());
149+
}
150+
cache.insert(10, "ten_v1");
151+
{
152+
auto v = cache.get(10);
153+
EXPECT_TRUE(v.ok());
154+
EXPECT_EQ("ten_v1", v.value());
155+
}
156+
}
142157

143158
} // namespace nebula
144159

‎src/storage/mutate/AddVerticesProcessor.cpp

+66-69
Original file line numberDiff line numberDiff line change
@@ -30,78 +30,77 @@ void AddVerticesProcessor::process(const cpp2::AddVerticesRequest& req) {
3030
}
3131

3232
CHECK_NOTNULL(kvstore_);
33-
if (indexes_.empty()) {
34-
std::for_each(partVertices.begin(), partVertices.end(), [&](auto& pv) {
35-
auto partId = pv.first;
36-
const auto& vertices = pv.second;
37-
std::vector<kvstore::KV> data;
38-
std::for_each(vertices.begin(), vertices.end(), [&](auto& v) {
39-
const auto& tags = v.get_tags();
40-
std::for_each(tags.begin(), tags.end(), [&](auto& tag) {
41-
VLOG(3) << "PartitionID: " << partId << ", VertexID: " << v.get_id()
42-
<< ", TagID: " << tag.get_tag_id() << ", TagVersion: " << version;
43-
auto key = NebulaKeyUtils::vertexKey(partId, v.get_id(),
44-
tag.get_tag_id(), version);
45-
data.emplace_back(std::move(key), std::move(tag.get_props()));
46-
if (FLAGS_enable_vertex_cache && vertexCache_ != nullptr) {
47-
vertexCache_->evict(std::make_pair(v.get_id(), tag.get_tag_id()));
48-
VLOG(3) << "Evict cache for vId " << v.get_id()
49-
<< ", tagId " << tag.get_tag_id();
50-
}
51-
});
33+
std::unordered_set<std::pair<VertexID, TagID>> uniqueIDs;
34+
uniqueIDs.reserve(128);
35+
36+
std::for_each(partVertices.begin(), partVertices.end(), [&](auto& pv) {
37+
std::vector<kvstore::KV> data;
38+
data.reserve(128);
39+
std::vector<std::tuple<VertexID, TagID, std::string>> cacheData;
40+
if (FLAGS_enable_vertex_cache && vertexCache_ != nullptr) {
41+
cacheData.reserve(128);
42+
}
43+
auto partId = pv.first;
44+
const auto& vertices = pv.second;
45+
46+
uniqueIDs.clear();
47+
std::for_each(vertices.rbegin(), vertices.rend(), [&](auto& v) {
48+
const auto& tags = v.get_tags();
49+
std::for_each(tags.begin(), tags.end(), [&](auto& tag) {
50+
auto uniqueKey = std::make_pair(v.get_id(), tag.get_tag_id());
51+
if (uniqueIDs.find(uniqueKey) != uniqueIDs.end()) {
52+
return;
53+
}
54+
55+
VLOG(3) << "PartitionID: " << partId << ", VertexID: " << v.get_id()
56+
<< ", TagID: " << tag.get_tag_id() << ", TagVersion: " << version;
57+
auto key = NebulaKeyUtils::vertexKey(partId, v.get_id(),
58+
tag.get_tag_id(), version);
59+
if (FLAGS_enable_vertex_cache && vertexCache_ != nullptr) {
60+
cacheData.emplace_back(v.get_id(), tag.get_tag_id(), tag.get_props());
61+
}
62+
data.emplace_back(std::move(key), std::move(tag.get_props()));
63+
uniqueIDs.emplace(uniqueKey);
5264
});
53-
doPut(spaceId_, partId, std::move(data));
54-
});
55-
} else {
56-
std::for_each(partVertices.begin(), partVertices.end(), [&](auto &pv) {
57-
auto partId = pv.first;
58-
auto atomic = [version, partId, vertices = std::move(pv.second), this]()
59-
-> folly::Optional<std::string> {
60-
return addVertices(version, partId, vertices);
61-
};
62-
auto callback = [partId, this](kvstore::ResultCode code) {
63-
handleAsync(spaceId_, partId, code);
64-
};
65-
this->kvstore_->asyncAtomicOp(spaceId_, partId, atomic, callback);
6665
});
67-
}
68-
}
6966

70-
std::string AddVerticesProcessor::addVertices(int64_t version, PartitionID partId,
71-
const std::vector<cpp2::Vertex>& vertices) {
72-
std::unique_ptr<kvstore::BatchHolder> batchHolder = std::make_unique<kvstore::BatchHolder>();
73-
/*
74-
* Define the map newIndexes to avoid inserting duplicate vertex.
75-
* This map means :
76-
* map<vertex_unique_key, prop_value> ,
77-
* -- vertex_unique_key is only used as the unique key , for example:
78-
* insert below vertices in the same request:
79-
* kv(part1_vid1_tag1 , v1)
80-
* kv(part1_vid1_tag1 , v2)
81-
* kv(part1_vid1_tag1 , v3)
82-
* kv(part1_vid1_tag1 , v4)
83-
*
84-
* Ultimately, kv(part1_vid1_tag1 , v4) . It's just what I need.
85-
*/
86-
std::map<std::string, std::string> newVertices;
87-
std::for_each(vertices.begin(), vertices.end(), [&](auto& v) {
88-
auto vId = v.get_id();
89-
const auto& tags = v.get_tags();
90-
std::for_each(tags.begin(), tags.end(), [&](auto& tag) {
91-
auto tagId = tag.get_tag_id();
92-
auto prop = tag.get_props();
93-
VLOG(3) << "PartitionID: " << partId << ", VertexID: " << vId
94-
<< ", TagID: " << tagId << ", TagVersion: " << version;
95-
auto key = NebulaKeyUtils::vertexKey(partId, vId, tagId, version);
96-
newVertices[key] = std::move(prop);
97-
if (FLAGS_enable_vertex_cache && this->vertexCache_ != nullptr) {
98-
this->vertexCache_->evict(std::make_pair(vId, tagId));
99-
VLOG(3) << "Evict cache for vId " << vId << ", tagId " << tagId;
67+
auto callback = [partId,
68+
this,
69+
cacheData = std::move(cacheData)] (kvstore::ResultCode code) mutable {
70+
if (FLAGS_enable_vertex_cache
71+
&& vertexCache_ != nullptr
72+
&& code == kvstore::ResultCode::SUCCEEDED) {
73+
for (auto&& tup : cacheData) {
74+
vertexCache_->insert(std::make_pair(std::get<0>(tup),
75+
std::get<1>(tup)),
76+
std::move(std::get<2>(tup)));
77+
}
10078
}
101-
});
79+
handleAsync(spaceId_, partId, code);
80+
};
81+
if (indexes_.empty()) {
82+
this->kvstore_->asyncMultiPut(spaceId_,
83+
partId,
84+
std::move(data),
85+
std::move(callback));
86+
} else {
87+
auto atomicOp = [partId, data = std::move(data), this]() mutable
88+
-> folly::Optional<std::string> {
89+
return addVerticesWithIndex(partId, std::move(data));
90+
};
91+
92+
this->kvstore_->asyncAtomicOp(spaceId_,
93+
partId,
94+
std::move(atomicOp),
95+
std::move(callback));
96+
}
10297
});
98+
}
10399

104-
for (auto& v : newVertices) {
100+
std::string AddVerticesProcessor::addVerticesWithIndex(PartitionID partId,
101+
std::vector<kvstore::KV>&& data) {
102+
std::unique_ptr<kvstore::BatchHolder> batchHolder = std::make_unique<kvstore::BatchHolder>();
103+
for (auto& v : data) {
105104
std::string val;
106105
RowReader nReader = RowReader::getEmptyRowReader();
107106
auto tagId = NebulaKeyUtils::getTagId(v.first);
@@ -150,9 +149,7 @@ std::string AddVerticesProcessor::addVertices(int64_t version, PartitionID partI
150149
/*
151150
* step 3 , Insert new vertex data
152151
*/
153-
auto key = v.first;
154-
auto prop = v.second;
155-
batchHolder->put(std::move(key), std::move(prop));
152+
batchHolder->put(std::move(v.first), std::move(v.second));
156153
}
157154
return encodeBatchValue(batchHolder->getBatch());
158155
}

‎src/storage/mutate/AddVerticesProcessor.h

+2-2
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,8 @@ class AddVerticesProcessor : public BaseProcessor<cpp2::ExecResponse> {
3939
, indexMan_(indexMan)
4040
, vertexCache_(cache) {}
4141

42-
std::string addVertices(int64_t version, PartitionID partId,
43-
const std::vector<cpp2::Vertex>& vertices);
42+
std::string addVerticesWithIndex(PartitionID partId,
43+
std::vector<kvstore::KV>&& data);
4444

4545
std::string findObsoleteIndex(PartitionID partId,
4646
VertexID vId,

0 commit comments

Comments
 (0)