1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
|
//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2024
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#include "td/db/binlog/ConcurrentBinlog.h"
#include "td/utils/logging.h"
#include "td/utils/misc.h"
#include "td/utils/OrderedEventsProcessor.h"
#include "td/utils/SliceBuilder.h"
#include "td/utils/Time.h"
#include <map>
namespace td {
namespace detail {
class BinlogActor final : public Actor {
public:
BinlogActor(unique_ptr<Binlog> binlog, uint64 seq_no) : binlog_(std::move(binlog)), processor_(seq_no) {
}
void close(Promise<> promise) {
binlog_->close().ensure();
LOG(INFO) << "Finished to close binlog";
stop();
promise.set_value(Unit()); // setting promise can complete closing and destroy the current actor context
}
void close_and_destroy(Promise<> promise) {
binlog_->close_and_destroy().ensure();
LOG(INFO) << "Finished to destroy binlog";
stop();
promise.set_value(Unit()); // setting promise can complete closing and destroy the current actor context
}
struct Event {
BufferSlice raw_event;
Promise<> sync_promise;
BinlogDebugInfo debug_info;
};
void erase_batch(uint64 seq_no, std::vector<uint64> event_ids) {
for (auto event_id : event_ids) {
auto event = BinlogEvent::create_raw(event_id, BinlogEvent::ServiceTypes::Empty, BinlogEvent::Flags::Rewrite,
EmptyStorer());
add_raw_event(seq_no, std::move(event), Promise<Unit>(), BinlogDebugInfo{__FILE__, __LINE__});
seq_no++;
}
}
void add_raw_event(uint64 seq_no, BufferSlice &&raw_event, Promise<> &&promise, BinlogDebugInfo info) {
processor_.add(seq_no, Event{std::move(raw_event), std::move(promise), info}, [&](uint64 event_id, Event &&event) {
if (!event.raw_event.empty()) {
do_add_raw_event(std::move(event.raw_event), event.debug_info);
}
do_lazy_sync(std::move(event.sync_promise));
});
flush_immediate_sync();
try_flush();
}
void force_sync(Promise<> &&promise) {
auto seq_no = processor_.max_unfinished_seq_no();
if (processor_.max_finished_seq_no() == seq_no) {
do_immediate_sync(std::move(promise));
} else {
immediate_sync_promises_.emplace(seq_no, std::move(promise));
}
}
void force_flush() {
// TODO: use same logic as in force_sync
binlog_->flush();
flush_flag_ = false;
}
void change_key(DbKey db_key, Promise<> promise) {
binlog_->change_key(std::move(db_key));
promise.set_value(Unit());
}
private:
unique_ptr<Binlog> binlog_;
OrderedEventsProcessor<Event> processor_;
std::multimap<uint64, Promise<>> immediate_sync_promises_;
std::vector<Promise<>> sync_promises_;
bool force_sync_flag_ = false;
bool lazy_sync_flag_ = false;
bool flush_flag_ = false;
double wakeup_at_ = 0;
static constexpr double FLUSH_TIMEOUT = 0.001; // 1ms
void wakeup_after(double after) {
auto now = Time::now_cached();
wakeup_at(now + after);
}
void wakeup_at(double at) {
if (wakeup_at_ == 0 || wakeup_at_ > at) {
wakeup_at_ = at;
set_timeout_at(wakeup_at_);
}
}
void do_add_raw_event(BufferSlice &&raw_event, BinlogDebugInfo info) {
binlog_->add_raw_event(std::move(raw_event), info);
}
void try_flush() {
auto need_flush_since = binlog_->need_flush_since();
auto now = Time::now_cached();
if (now > need_flush_since + FLUSH_TIMEOUT - 1e-9) {
binlog_->flush();
} else {
if (!force_sync_flag_) {
flush_flag_ = true;
wakeup_at(need_flush_since + FLUSH_TIMEOUT);
}
}
}
void flush_immediate_sync() {
auto seq_no = processor_.max_finished_seq_no();
for (auto it = immediate_sync_promises_.begin(), end = immediate_sync_promises_.end();
it != end && it->first <= seq_no; it = immediate_sync_promises_.erase(it)) {
do_immediate_sync(std::move(it->second));
}
}
void do_immediate_sync(Promise<> &&promise) {
if (promise) {
sync_promises_.emplace_back(std::move(promise));
}
if (!force_sync_flag_) {
force_sync_flag_ = true;
wakeup_after(0.003);
}
}
void do_lazy_sync(Promise<> &&promise) {
if (!promise) {
return;
}
sync_promises_.emplace_back(std::move(promise));
if (!lazy_sync_flag_ && !force_sync_flag_) {
wakeup_after(30);
lazy_sync_flag_ = true;
}
}
void timeout_expired() final {
bool need_sync = lazy_sync_flag_ || force_sync_flag_;
lazy_sync_flag_ = false;
force_sync_flag_ = false;
bool need_flush = flush_flag_;
flush_flag_ = false;
wakeup_at_ = 0;
if (need_sync) {
binlog_->sync();
// LOG(ERROR) << "BINLOG SYNC";
set_promises(sync_promises_);
} else if (need_flush) {
try_flush();
// LOG(ERROR) << "BINLOG FLUSH";
}
}
};
} // namespace detail
ConcurrentBinlog::ConcurrentBinlog() = default;
ConcurrentBinlog::~ConcurrentBinlog() = default;
ConcurrentBinlog::ConcurrentBinlog(unique_ptr<Binlog> binlog, int scheduler_id) {
init_impl(std::move(binlog), scheduler_id);
}
Result<BinlogInfo> ConcurrentBinlog::init(string path, const Callback &callback, DbKey db_key, DbKey old_db_key,
int scheduler_id) {
auto binlog = make_unique<Binlog>();
TRY_STATUS(binlog->init(std::move(path), callback, std::move(db_key), std::move(old_db_key)));
auto info = binlog->get_info();
init_impl(std::move(binlog), scheduler_id);
return info;
}
void ConcurrentBinlog::init_impl(unique_ptr<Binlog> binlog, int32 scheduler_id) {
path_ = binlog->get_path().str();
last_event_id_ = binlog->peek_next_event_id();
binlog_actor_ = create_actor_on_scheduler<detail::BinlogActor>(PSLICE() << "Binlog " << path_, scheduler_id,
std::move(binlog), last_event_id_);
}
void ConcurrentBinlog::close_impl(Promise<> promise) {
send_closure(std::move(binlog_actor_), &detail::BinlogActor::close, std::move(promise));
}
void ConcurrentBinlog::close_and_destroy_impl(Promise<> promise) {
send_closure(std::move(binlog_actor_), &detail::BinlogActor::close_and_destroy, std::move(promise));
}
void ConcurrentBinlog::add_raw_event_impl(uint64 event_id, BufferSlice &&raw_event, Promise<> promise,
BinlogDebugInfo info) {
send_closure(binlog_actor_, &detail::BinlogActor::add_raw_event, event_id, std::move(raw_event), std::move(promise),
info);
}
void ConcurrentBinlog::force_sync(Promise<> promise) {
send_closure(binlog_actor_, &detail::BinlogActor::force_sync, std::move(promise));
}
void ConcurrentBinlog::force_flush() {
send_closure(binlog_actor_, &detail::BinlogActor::force_flush);
}
void ConcurrentBinlog::change_key(DbKey db_key, Promise<> promise) {
send_closure(binlog_actor_, &detail::BinlogActor::change_key, std::move(db_key), std::move(promise));
}
uint64 ConcurrentBinlog::erase_batch(vector<uint64> event_ids) {
auto shift = narrow_cast<int32>(event_ids.size());
if (shift == 0) {
return 0;
}
auto seq_no = next_event_id(shift);
send_closure(binlog_actor_, &detail::BinlogActor::erase_batch, seq_no, std::move(event_ids));
return seq_no;
}
} // namespace td
|
