Skip to content

Commit 54b1f00

Browse files
levlamlevlam
committed
Use WaitFreeHashMap in BinlogKeyValue.
1 parent 6614c14 commit 54b1f00

2 files changed

Lines changed: 33 additions & 65 deletions

File tree

‎tddb/td/db/BinlogKeyValue.h‎

Lines changed: 33 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
#include "td/utils/StorerBase.h"
2626
#include "td/utils/tl_parsers.h"
2727
#include "td/utils/tl_storers.h"
28+
#include "td/utils/WaitFreeHashMap.h"
2829

2930
#include <functional>
3031
#include <memory>
@@ -92,7 +93,7 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
9293
LOG(ERROR) << "Have event with empty key";
9394
return;
9495
}
95-
map_.emplace(event.key.str(), std::make_pair(event.value.str(), binlog_event.id_));
96+
map_.set(event.key.str(), std::make_pair(event.value.str(), binlog_event.id_));
9697
},
9798
std::move(db_key), DbKey::empty(), scheduler_id));
9899
return Status::OK();
@@ -120,7 +121,7 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
120121
LOG(ERROR) << "Have external event with empty key";
121122
return;
122123
}
123-
map_.emplace(event.key.str(), std::make_pair(event.value.str(), binlog_event.id_));
124+
map_.set(event.key.str(), std::make_pair(event.value.str(), binlog_event.id_));
124125
}
125126

126127
void external_init_finish(std::shared_ptr<BinlogT> binlog) {
@@ -136,30 +137,22 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
136137

137138
SeqNo set(string key, string value) final {
138139
auto lock = rw_mutex_.lock_write();
139-
uint64 old_event_id = 0;
140140
CHECK(!key.empty());
141-
auto it_ok = map_.emplace(key, std::make_pair(value, 0));
142-
if (!it_ok.second) {
143-
if (it_ok.first->second.first == value) {
144-
return 0;
145-
}
146-
VLOG(binlog) << "Change value of key " << key << " from " << hex_encode(it_ok.first->second.first) << " to "
147-
<< hex_encode(value);
148-
old_event_id = it_ok.first->second.second;
149-
it_ok.first->second.first = value;
150-
} else {
151-
VLOG(binlog) << "Set value of key " << key << " to " << hex_encode(value);
141+
auto &it = map_[key];
142+
if (it.second != 0 && it.first == value) {
143+
return 0;
152144
}
153-
bool rewrite = false;
154-
uint64 event_id;
155145
auto seq_no = binlog_->next_event_id();
156-
if (old_event_id != 0) {
146+
bool rewrite = false;
147+
if (it.second != 0) {
148+
VLOG(binlog) << "Change value of key " << key << " from " << hex_encode(it.first) << " to " << hex_encode(value);
157149
rewrite = true;
158-
event_id = old_event_id;
159150
} else {
160-
event_id = seq_no;
161-
it_ok.first->second.second = event_id;
151+
VLOG(binlog) << "Set value of key " << key << " to " << hex_encode(value);
152+
it.second = seq_no;
162153
}
154+
it.first = value;
155+
auto event_id = it.second;
163156

164157
lock.reset();
165158
add_event(seq_no,
@@ -169,13 +162,13 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
169162

170163
SeqNo erase(const string &key) final {
171164
auto lock = rw_mutex_.lock_write();
172-
auto it = map_.find(key);
173-
if (it == map_.end()) {
165+
if (map_.count(key) == 0) {
174166
return 0;
175167
}
176-
VLOG(binlog) << "Remove value of key " << key << ", which is " << hex_encode(it->second.first);
177-
uint64 event_id = it->second.second;
178-
map_.erase(it);
168+
auto &value = map_[key];
169+
VLOG(binlog) << "Remove value of key " << key << ", which is " << hex_encode(value.first);
170+
uint64 event_id = value.second;
171+
map_.erase(key);
179172
auto seq_no = binlog_->next_event_id();
180173
lock.reset();
181174
add_event(seq_no, BinlogEvent::create_raw(event_id, BinlogEvent::ServiceTypes::Empty, BinlogEvent::Flags::Rewrite,
@@ -186,11 +179,11 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
186179
SeqNo erase_batch(vector<string> keys) final {
187180
auto lock = rw_mutex_.lock_write();
188181
vector<uint64> log_event_ids;
189-
for (auto &key : keys) {
190-
auto it = map_.find(key);
191-
if (it != map_.end()) {
192-
log_event_ids.push_back(it->second.second);
193-
map_.erase(it);
182+
for (const auto &key : keys) {
183+
if (map_.count(key) != 0) {
184+
auto &value = map_[key];
185+
log_event_ids.push_back(value.second);
186+
map_.erase(key);
194187
}
195188
}
196189
if (log_event_ids.empty()) {
@@ -211,12 +204,7 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
211204

212205
string get(const string &key) final {
213206
auto lock = rw_mutex_.lock_read();
214-
auto it = map_.find(key);
215-
if (it == map_.end()) {
216-
return string();
217-
}
218-
VLOG(binlog) << "Get value of key " << key << ", which is " << hex_encode(it->second.first);
219-
return it->second.first;
207+
return map_.get(key).first;
220208
}
221209

222210
void force_sync(Promise<> &&promise, const char *source) final {
@@ -229,36 +217,32 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
229217

230218
void for_each(std::function<void(Slice, Slice)> func) final {
231219
auto lock = rw_mutex_.lock_write();
232-
for (const auto &kv : map_) {
233-
func(kv.first, kv.second.first);
234-
}
220+
map_.foreach([&func](const string &key, const std::pair<string, uint64> &value) { func(key, value.first); });
235221
}
236222

237223
std::unordered_map<string, string, Hash<string>> prefix_get(Slice prefix) final {
238224
auto lock = rw_mutex_.lock_write();
239225
std::unordered_map<string, string, Hash<string>> res;
240-
for (const auto &kv : map_) {
241-
if (begins_with(kv.first, prefix)) {
242-
res.emplace(kv.first.substr(prefix.size()), kv.second.first);
226+
map_.foreach([&res, prefix](const string &key, const std::pair<string, uint64> &value) {
227+
if (begins_with(key, prefix)) {
228+
res.emplace(key.substr(prefix.size()), value.first);
243229
}
244-
}
230+
});
245231
return res;
246232
}
247233

248234
FlatHashMap<string, string> get_all() final {
249235
auto lock = rw_mutex_.lock_write();
250236
FlatHashMap<string, string> res;
251-
res.reserve(map_.size());
252-
for (const auto &kv : map_) {
253-
res.emplace(kv.first, kv.second.first);
254-
}
237+
res.reserve(map_.calc_size());
238+
map_.foreach([&res](const string &key, const std::pair<string, uint64> &value) { res.emplace(key, value.first); });
255239
return res;
256240
}
257241

258242
void erase_by_prefix(Slice prefix) final {
259243
auto lock = rw_mutex_.lock_write();
260244
vector<uint64> event_ids;
261-
table_remove_if(map_, [&](const auto &it) {
245+
map_.remove_if([&event_ids, prefix](const auto &it) {
262246
if (begins_with(it.first, prefix)) {
263247
event_ids.push_back(it.second.second);
264248
return true;
@@ -282,7 +266,7 @@ class BinlogKeyValue final : public KeyValueSyncInterface {
282266
}
283267

284268
private:
285-
FlatHashMap<string, std::pair<string, uint64>> map_;
269+
WaitFreeHashMap<string, std::pair<string, uint64>> map_;
286270
std::shared_ptr<BinlogT> binlog_;
287271
RwMutex rw_mutex_;
288272
int32 magic_ = MAGIC;

‎tdutils/td/utils/algorithm.h‎

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -301,20 +301,4 @@ bool table_remove_if(FlatHashTable<NodeT, HashT, EqT> &table, FuncT &&func) {
301301
return table.remove_if(func);
302302
}
303303

304-
template <class KeyT, class ValueT, class HashT, class EqT>
305-
class WaitFreeHashMap;
306-
307-
template <class KeyT, class ValueT, class HashT, class EqT, class FuncT>
308-
bool table_remove_if(WaitFreeHashMap<KeyT, ValueT, HashT, EqT> &table, const FuncT &func) {
309-
return table.remove_if(func);
310-
}
311-
312-
template <class KeyT, class HashT, class EqT>
313-
class WaitFreeHashSet;
314-
315-
template <class KeyT, class HashT, class EqT, class FuncT>
316-
bool table_remove_if(WaitFreeHashSet<KeyT, HashT, EqT> &table, const FuncT &func) {
317-
return table.remove_if(func);
318-
}
319-
320304
} // namespace td

0 commit comments

Comments
 (0)