mirrors/dragonfly
1
0
Fork 0
mirror of https://github.com/dragonflydb/dragonfly.git synced 2025-05-10 18:05:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(replication): fullsync phase write to sync on noop (#3084)

Browse source 
* fix replication: fullsync phase write to sync on noop

Signed-off-by: adi_holden <adi@dragonflydb.io>
This commit is contained in:
adiholden 2024-05-27 17:52:07 +03:00 • committed by GitHub
parent c45f7bfea5
commit b2213b05d1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 14 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

11
src/server/snapshot.cc
View file

@ -330,12 +330,17 @@ void SliceSnapshot::OnDbChange(DbIndex db_index, const DbSlice::ChangeReq& req)
// no database switch can be performed between those two calls, because they are part of one // no database switch can be performed between those two calls, because they are part of one
// transaction. // transaction.
void SliceSnapshot::OnJournalEntry(const journal::JournalItem& item, bool await) { void SliceSnapshot::OnJournalEntry(const journal::JournalItem& item, bool await) {
// We ignore EXEC and NOOP entries because we they have no meaning during // We ignore EXEC entries because we they have no meaning during
// the LOAD phase on replica. // the LOAD phase on replica.
if (item.opcode == journal::Op::NOOP || item.opcode == journal::Op::EXEC) if (item.opcode == journal::Op::EXEC)
return; return;
serializer_->WriteJournalEntry(item.data); // To enable journal flushing to sync after non auto journal command is executed we call
// TriggerJournalWriteToSink. This call uses the NOOP opcode with await=true. Since there is no
// additional journal change to serialize, it simply invokes PushSerializedToChannel.
if (item.opcode != journal::Op::NOOP) {
serializer_->WriteJournalEntry(item.data);
}
if (await) { if (await) {
// This is the only place that flushes in streaming mode // This is the only place that flushes in streaming mode

9
src/server/transaction.cc
View file

@ -1413,21 +1413,24 @@ void Transaction::LogAutoJournalOnShard(EngineShard* shard, RunnableResult resul
return; return;
} }
// If autojournaling was disabled and not re-enabled, skip it // If autojournaling was disabled and not re-enabled the callback is writing to journal.
// We do not allow preemption in callbacks and therefor the call to RecordJournal from
// from callbacks does not allow await.
// To make sure we flush the changes to sync we call TriggerJournalWriteToSink here.
if ((cid_->opt_mask() & CO::NO_AUTOJOURNAL) && !re_enabled_auto_journal_) { if ((cid_->opt_mask() & CO::NO_AUTOJOURNAL) && !re_enabled_auto_journal_) {
TriggerJournalWriteToSink(); TriggerJournalWriteToSink();
return; return;
} }
// TODO: Handle complex commands like LMPOP correctly once they are implemented.
journal::Entry::Payload entry_payload; journal::Entry::Payload entry_payload;
string_view cmd{cid_->name()}; string_view cmd{cid_->name()};
if (unique_shard_cnt_ == 1 || kv_args_.empty()) { if (unique_shard_cnt_ == 1 || kv_args_.empty()) {
entry_payload = journal::Entry::Payload(cmd, full_args_); entry_payload = journal::Entry::Payload(cmd, full_args_);
} else { } else {
entry_payload = journal::Entry::Payload(cmd, GetShardArgs(shard->shard_id()).AsSlice()); entry_payload = journal::Entry::Payload(cmd, GetShardArgs(shard->shard_id()).AsSlice());
} }
// Record to journal autojournal commands, here we allow await which anables writing to sync
// the journal change.
LogJournalOnShard(shard, std::move(entry_payload), unique_shard_cnt_, false, true); LogJournalOnShard(shard, std::move(entry_payload), unique_shard_cnt_, false, true);
} }