Skip to content

Commit aae81c3

Browse files
Shylock-HgShylock-Hg
authored
Merge branch 'master' into fix/wrong-properites-prune-length
2 parents a773d69 + 150e418 commit aae81c3

16 files changed

+129
-120
lines changed

‎.github/workflows/nightly.yml

+1-1
Original file line numberDiff line numberDiff line change
@@ -189,7 +189,7 @@ jobs:
189189
timeout-minutes: 2
190190
- name: coverage
191191
run: |
192-
~/.local/bin/fastcov -d build -l -o fastcov.info -p --exclude /usr/include --exclude=/opt/vesoft --exclude scanner.lex
192+
~/.local/bin/fastcov -d build -l -o fastcov.info -p --exclude /usr/include /usr/lib /opt/vesoft build/ tests/ /test .lex .yy
193193
- uses: codecov/codecov-action@v2
194194
with:
195195
files: fastcov.info

‎.github/workflows/pull_request.yml

+1-1
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ jobs:
185185
- name: coverage
186186
if: ${{ matrix.compiler == 'gcc-9.3' && matrix.os == 'ubuntu2004' }}
187187
run: |
188-
~/.local/bin/fastcov -d build -l -o fastcov.info -p --exclude /usr/include --exclude=/opt/vesoft --exclude scanner.lex
188+
~/.local/bin/fastcov -d build -l -o fastcov.info -p --exclude /usr/include /usr/lib /opt/vesoft build/ tests/ /test .lex .yy
189189
- uses: codecov/codecov-action@v2
190190
if: ${{ matrix.compiler == 'gcc-9.3' && matrix.os == 'ubuntu2004' }}
191191
with:

‎src/graph/executor/admin/SubmitJobExecutor.cpp

+2-2
Original file line numberDiff line numberDiff line change
@@ -131,8 +131,8 @@ nebula::DataSet SubmitJobExecutor::buildShowResultData(
131131
v.emplace_back(Row({jd.get_job_id(),
132132
apache::thrift::util::enumNameSafe(meta::cpp2::JobType::DATA_BALANCE),
133133
apache::thrift::util::enumNameSafe(jd.get_status()),
134-
convertJobTimestampToDateTime(jd.get_start_time()).toString(),
135-
convertJobTimestampToDateTime(jd.get_stop_time()).toString(),
134+
convertJobTimestampToDateTime(jd.get_start_time()),
135+
convertJobTimestampToDateTime(jd.get_stop_time()),
136136
apache::thrift::util::enumNameSafe(jd.get_code())}));
137137
for (size_t i = index; i < paras.size() - 1; i++) {
138138
meta::cpp2::BalanceTask tsk;

‎src/kvstore/Listener.cpp

+73-69
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ void Listener::stop() {
9999
bool Listener::preProcessLog(LogID logId,
100100
TermID termId,
101101
ClusterID clusterId,
102-
const std::string& log) {
102+
folly::StringPiece log) {
103103
UNUSED(logId);
104104
UNUSED(termId);
105105
UNUSED(clusterId);
@@ -147,6 +147,7 @@ void Listener::doApply() {
147147
if (isStopped()) {
148148
return;
149149
}
150+
150151
if (needToCleanupSnapshot()) {
151152
cleanupSnapshot();
152153
}
@@ -156,87 +157,89 @@ void Listener::doApply() {
156157
bgWorkers_->addDelayTask(
157158
FLAGS_listener_commit_interval_secs * 1000, &Listener::doApply, this);
158159
};
160+
processLogs();
161+
});
162+
}
159163

160-
std::unique_ptr<LogIterator> iter;
161-
{
162-
std::lock_guard<std::mutex> guard(raftLock_);
163-
if (lastApplyLogId_ >= committedLogId_) {
164-
return;
165-
}
166-
iter = wal_->iterator(lastApplyLogId_ + 1, committedLogId_);
164+
void Listener::processLogs() {
165+
std::unique_ptr<LogIterator> iter;
166+
{
167+
std::lock_guard<std::mutex> guard(raftLock_);
168+
if (lastApplyLogId_ >= committedLogId_) {
169+
return;
167170
}
171+
iter = wal_->iterator(lastApplyLogId_ + 1, committedLogId_);
172+
}
168173

169-
LogID lastApplyId = -1;
170-
// the kv pair which can sync to remote safely
171-
std::vector<KV> data;
172-
while (iter->valid()) {
173-
lastApplyId = iter->logId();
174-
175-
auto log = iter->logMsg();
176-
if (log.empty()) {
177-
// skip the heartbeat
178-
++(*iter);
179-
continue;
180-
}
174+
LogID lastApplyId = -1;
175+
// the kv pair which can sync to remote safely
176+
std::vector<KV> data;
177+
while (iter->valid()) {
178+
lastApplyId = iter->logId();
181179

182-
DCHECK_GE(log.size(), sizeof(int64_t) + 1 + sizeof(uint32_t));
183-
switch (log[sizeof(int64_t)]) {
184-
case OP_PUT: {
185-
auto pieces = decodeMultiValues(log);
186-
DCHECK_EQ(2, pieces.size());
187-
data.emplace_back(pieces[0], pieces[1]);
188-
break;
189-
}
190-
case OP_MULTI_PUT: {
191-
auto kvs = decodeMultiValues(log);
192-
DCHECK_EQ((kvs.size() + 1) / 2, kvs.size() / 2);
193-
for (size_t i = 0; i < kvs.size(); i += 2) {
194-
data.emplace_back(kvs[i], kvs[i + 1]);
195-
}
196-
break;
197-
}
198-
case OP_REMOVE:
199-
case OP_REMOVE_RANGE:
200-
case OP_MULTI_REMOVE: {
201-
break;
180+
auto log = iter->logMsg();
181+
if (log.empty()) {
182+
// skip the heartbeat
183+
++(*iter);
184+
continue;
185+
}
186+
187+
DCHECK_GE(log.size(), sizeof(int64_t) + 1 + sizeof(uint32_t));
188+
switch (log[sizeof(int64_t)]) {
189+
case OP_PUT: {
190+
auto pieces = decodeMultiValues(log);
191+
DCHECK_EQ(2, pieces.size());
192+
data.emplace_back(pieces[0], pieces[1]);
193+
break;
194+
}
195+
case OP_MULTI_PUT: {
196+
auto kvs = decodeMultiValues(log);
197+
DCHECK_EQ(0, kvs.size() % 2);
198+
for (size_t i = 0; i < kvs.size(); i += 2) {
199+
data.emplace_back(kvs[i], kvs[i + 1]);
202200
}
203-
case OP_BATCH_WRITE: {
204-
auto batch = decodeBatchValue(log);
205-
for (auto& op : batch) {
206-
// OP_BATCH_PUT and OP_BATCH_REMOVE_RANGE is ignored
207-
if (op.first == BatchLogType::OP_BATCH_PUT) {
208-
data.emplace_back(op.second.first, op.second.second);
209-
}
201+
break;
202+
}
203+
case OP_REMOVE:
204+
case OP_REMOVE_RANGE:
205+
case OP_MULTI_REMOVE: {
206+
break;
207+
}
208+
case OP_BATCH_WRITE: {
209+
auto batch = decodeBatchValue(log);
210+
for (auto& op : batch) {
211+
// OP_BATCH_REMOVE and OP_BATCH_REMOVE_RANGE is igored
212+
if (op.first == BatchLogType::OP_BATCH_PUT) {
213+
data.emplace_back(op.second.first, op.second.second);
210214
}
211-
break;
212-
}
213-
case OP_TRANS_LEADER:
214-
case OP_ADD_LEARNER:
215-
case OP_ADD_PEER:
216-
case OP_REMOVE_PEER: {
217-
break;
218-
}
219-
default: {
220-
VLOG(2) << idStr_
221-
<< "Should not reach here. Unknown operation: " << static_cast<int32_t>(log[0]);
222215
}
216+
break;
223217
}
224-
225-
if (static_cast<int32_t>(data.size()) > FLAGS_listener_commit_batch_size) {
218+
case OP_TRANS_LEADER:
219+
case OP_ADD_LEARNER:
220+
case OP_ADD_PEER:
221+
case OP_REMOVE_PEER: {
226222
break;
227223
}
228-
++(*iter);
224+
default: {
225+
VLOG(2) << idStr_ << "Unknown operation: " << static_cast<int32_t>(log[0]);
226+
}
229227
}
230228

231-
// apply to state machine
232-
if (lastApplyId != -1 && apply(data)) {
233-
std::lock_guard<std::mutex> guard(raftLock_);
234-
lastApplyLogId_ = lastApplyId;
235-
persist(committedLogId_, term_, lastApplyLogId_);
236-
VLOG(2) << idStr_ << "Listener succeeded apply log to " << lastApplyLogId_;
237-
lastApplyTime_ = time::WallClock::fastNowInMilliSec();
229+
if (static_cast<int32_t>(data.size()) > FLAGS_listener_commit_batch_size) {
230+
break;
238231
}
239-
});
232+
++(*iter);
233+
}
234+
235+
// apply to state machine
236+
if (lastApplyId != -1 && apply(data)) {
237+
std::lock_guard<std::mutex> guard(raftLock_);
238+
lastApplyLogId_ = lastApplyId;
239+
persist(committedLogId_, term_, lastApplyLogId_);
240+
VLOG(2) << idStr_ << "Listener succeeded apply log to " << lastApplyLogId_;
241+
lastApplyTime_ = time::WallClock::fastNowInMilliSec();
242+
}
240243
}
241244

242245
std::tuple<nebula::cpp2::ErrorCode, int64_t, int64_t> Listener::commitSnapshot(
@@ -303,5 +306,6 @@ bool Listener::pursueLeaderDone() {
303306
"pursue leader : leaderCommitId={}, lastApplyLogId_={}", leaderCommitId_, lastApplyLogId_);
304307
return (leaderCommitId_ - lastApplyLogId_) <= FLAGS_listener_pursue_leader_threshold;
305308
}
309+
306310
} // namespace kvstore
307311
} // namespace nebula

‎src/kvstore/Listener.h

+8-5
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,10 @@ using RaftClient = thrift::ThriftClientManager<raftex::cpp2::RaftexServiceAsyncC
6868
* TermID committedLogTerm,
6969
* bool finished) override;
7070
*
71+
* // extra cleanup work, will be invoked when listener is about to be removed,
72+
* // or raft is reset
73+
* void cleanup()；
74+
*
7175
* * Must implement in derived class
7276
* // extra initialize work could do here
7377
* void init()
@@ -83,10 +87,6 @@ using RaftClient = thrift::ThriftClientManager<raftex::cpp2::RaftexServiceAsyncC
8387
*
8488
* // persist last commit log id/term and lastApplyId
8589
* bool persist(LogID, TermID, LogID)
86-
*
87-
* // extra cleanup work, will be invoked when listener is about to be removed,
88-
* // or raft is reset
89-
* nebula::cpp2::ErrorCode cleanup() = 0
9090
*/
9191
class Listener : public raftex::RaftPart {
9292
public:
@@ -270,7 +270,7 @@ class Listener : public raftex::RaftPart {
270270
bool preProcessLog(LogID logId,
271271
TermID termId,
272272
ClusterID clusterId,
273-
const std::string& log) override;
273+
folly::StringPiece log) override;
274274

275275
/**
276276
* @brief If the listener falls behind way to much than leader, the leader will send all its data
@@ -295,6 +295,9 @@ class Listener : public raftex::RaftPart {
295295
*/
296296
void doApply();
297297

298+
// Process logs and then call apply to execute
299+
virtual void processLogs();
300+
298301
protected:
299302
LogID leaderCommitId_ = 0;
300303
LogID lastApplyLogId_ = 0;

‎src/kvstore/Part.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -409,7 +409,7 @@ nebula::cpp2::ErrorCode Part::putCommitMsg(WriteBatch* batch,
409409
return batch->put(NebulaKeyUtils::systemCommitKey(partId_), commitMsg);
410410
}
411411

412-
bool Part::preProcessLog(LogID logId, TermID termId, ClusterID clusterId, const std::string& log) {
412+
bool Part::preProcessLog(LogID logId, TermID termId, ClusterID clusterId, folly::StringPiece log) {
413413
// We should apply any membership change which happens before start time. Because when we start
414414
// up, the peers comes from meta, has already contains all previous changes.
415415
VLOG(4) << idStr_ << "logId " << logId << ", termId " << termId << ", clusterId " << clusterId;

‎src/kvstore/Part.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,7 @@ class Part : public raftex::RaftPart {
274274
bool preProcessLog(LogID logId,
275275
TermID termId,
276276
ClusterID clusterId,
277-
const std::string& log) override;
277+
folly::StringPiece log) override;
278278

279279
/**
280280
* @brief If a raft peer falls behind way to much than leader, the leader will send all its data

‎src/kvstore/raftex/RaftPart.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -368,7 +368,7 @@ RaftPart::RaftPart(
368368
walRoot,
369369
std::move(info),
370370
std::move(policy),
371-
[this](LogID logId, TermID logTermId, ClusterID logClusterId, const std::string& log) {
371+
[this](LogID logId, TermID logTermId, ClusterID logClusterId, folly::StringPiece log) {
372372
return this->preProcessLog(logId, logTermId, logClusterId, log);
373373
},
374374
diskMan);

‎src/kvstore/raftex/RaftPart.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -516,7 +516,7 @@ class RaftPart : public std::enable_shared_from_this<RaftPart> {
516516
virtual bool preProcessLog(LogID logId,
517517
TermID termId,
518518
ClusterID clusterId,
519-
const std::string& log) = 0;
519+
folly::StringPiece log) = 0;
520520

521521
/**
522522
* @brief If raft node falls behind way to much than leader, the leader will send all its data in

‎src/kvstore/raftex/test/TestShard.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ class TestShard : public RaftPart {
8282
bool wait,
8383
bool needLock) override;
8484

85-
bool preProcessLog(LogID, TermID, ClusterID, const std::string& log) override {
85+
bool preProcessLog(LogID, TermID, ClusterID, folly::StringPiece log) override {
8686
if (!log.empty()) {
8787
switch (static_cast<CommandType>(log[0])) {
8888
case CommandType::ADD_LEARNER: {

‎src/kvstore/test/DiskManagerTest.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ TEST(DiskManagerTest, WalNoSpaceTest) {
126126
walPath,
127127
info,
128128
policy,
129-
[](LogID, TermID, ClusterID, const std::string&) { return true; },
129+
[](LogID, TermID, ClusterID, folly::StringPiece) { return true; },
130130
diskMan);
131131

132132
diskMan->freeBytes_[0] = FLAGS_minimum_reserved_bytes + 10000;

‎src/kvstore/wal/AtomicLogBuffer.h

+6-6
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,8 @@ struct Record {
2626
Record(Record&& record) noexcept = default;
2727
Record& operator=(Record&& record) noexcept = default;
2828

29-
Record(ClusterID clusterId, TermID termId, std::string msg)
30-
: clusterId_(clusterId), termId_(termId), msg_(std::move(msg)) {}
29+
Record(ClusterID clusterId, TermID termId, folly::StringPiece msg)
30+
: clusterId_(clusterId), termId_(termId), msg_(msg.toString()) {}
3131

3232
int32_t size() const {
3333
return sizeof(ClusterID) + sizeof(TermID) + msg_.size();
@@ -52,7 +52,7 @@ struct Node {
5252
}
5353

5454
/**
55-
* @brief Add a record to current not
55+
* @brief Add a record to current node
5656
*
5757
* @param rec Record to add
5858
* @return Whether operation succeed or not
@@ -267,8 +267,8 @@ class AtomicLogBuffer : public std::enable_shared_from_this<AtomicLogBuffer> {
267267
* @param clusterId Cluster id of log
268268
* @param msg Log message
269269
*/
270-
void push(LogID logId, TermID termId, ClusterID clusterId, std::string&& msg) {
271-
push(logId, Record(clusterId, termId, std::move(msg)));
270+
void push(LogID logId, TermID termId, ClusterID clusterId, folly::StringPiece msg) {
271+
push(logId, Record(clusterId, termId, msg));
272272
}
273273

274274
/**
@@ -327,7 +327,7 @@ class AtomicLogBuffer : public std::enable_shared_from_this<AtomicLogBuffer> {
327327
explicit AtomicLogBuffer(int32_t capacity) : capacity_(capacity) {}
328328

329329
/**
330-
* @brief Find the noe which contains the log with given id
330+
* @brief Find the node which contains the log with given id
331331
*
332332
* @param logId Log it to seek
333333
* @return Node* Return the node contains the log, return nullptr if not found

‎src/kvstore/wal/FileBasedWal.cpp

+7-5
Original file line numberDiff line numberDiff line change
@@ -439,7 +439,10 @@ void FileBasedWal::scanLastWal(WalFileInfoPtr info, LogID firstId) {
439439
close(fd);
440440
}
441441

442-
bool FileBasedWal::appendLogInternal(LogID id, TermID term, ClusterID cluster, std::string msg) {
442+
bool FileBasedWal::appendLogInternal(LogID id,
443+
TermID term,
444+
ClusterID cluster,
445+
folly::StringPiece msg) {
443446
if (lastLogId_ != 0 && firstLogId_ != 0 && id != lastLogId_ + 1) {
444447
VLOG(3) << idStr_ << "There is a gap in the log id. The last log id is " << lastLogId_
445448
<< ", and the id being appended is " << id;
@@ -493,7 +496,7 @@ bool FileBasedWal::appendLogInternal(LogID id, TermID term, ClusterID cluster, s
493496
firstLogId_ = id;
494497
}
495498

496-
logBuffer_->push(id, term, cluster, std::move(msg));
499+
logBuffer_->push(id, term, cluster, msg);
497500
return true;
498501
}
499502

@@ -502,7 +505,7 @@ bool FileBasedWal::appendLog(LogID id, TermID term, ClusterID cluster, std::stri
502505
VLOG_EVERY_N(2, 1000) << idStr_ << "Failed to appendLogs because of no more space";
503506
return false;
504507
}
505-
if (!appendLogInternal(id, term, cluster, std::move(msg))) {
508+
if (!appendLogInternal(id, term, cluster, folly::StringPiece(msg))) {
506509
VLOG(3) << "Failed to append log for logId " << id;
507510
return false;
508511
}
@@ -515,8 +518,7 @@ bool FileBasedWal::appendLogs(LogIterator& iter) {
515518
return false;
516519
}
517520
for (; iter.valid(); ++iter) {
518-
if (!appendLogInternal(
519-
iter.logId(), iter.logTerm(), iter.logSource(), iter.logMsg().toString())) {
521+
if (!appendLogInternal(iter.logId(), iter.logTerm(), iter.logSource(), iter.logMsg())) {
520522
VLOG(3) << idStr_ << "Failed to append log for logId " << iter.logId();
521523
return false;
522524
}

0 commit comments

Comments
 (0)