mirrors/dragonflydb-dragonfly
1
0
Fork 0
mirror of https://github.com/dragonflydb/dragonfly.git synced 2024-12-14 11:58:02 +00:00
Code Activity

fix(cluster): fix incorrect version checking and resource double free (#2499)

Browse source 
* fix(cluster): fix incorrect version checking and resource double free
This commit is contained in:
Borys 2024-01-29 14:14:21 +02:00 committed by GitHub
parent e4862f227d
commit 1dee082f86
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 11 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/server/cluster/outgoing_slot_migration.cc
View file

@ -12,15 +12,16 @@ namespace dfly {
class OutgoingMigration::SliceSlotMigration {
public:
SliceSlotMigration(DbSlice* slice, SlotSet slots, uint32_t sync_id, journal::Journal* journal,
Context* cntx)
Context* cntx, io::Sink* dest)
: streamer_(slice, std::move(slots), sync_id, journal, cntx) {
}
void Start(io::Sink* dest) {
streamer_.Start(dest);
state_ = MigrationState::C_FULL_SYNC;
}
~SliceSlotMigration() {
streamer_.Cancel();
}
MigrationState GetState() const {
return state_ == MigrationState::C_FULL_SYNC && streamer_.IsSnapshotFinished()
? MigrationState::C_STABLE_SYNC
@ -52,8 +53,7 @@ void OutgoingMigration::StartFlow(DbSlice* slice, uint32_t sync_id, journal::Jou
std::lock_guard lck(flows_mu_);
slot_migrations_[shard_id] =
std::make_unique<SliceSlotMigration>(slice, std::move(sset), sync_id, journal, &cntx_);
slot_migrations_[shard_id]->Start(dest);
std::make_unique<SliceSlotMigration>(slice, std::move(sset), sync_id, journal, &cntx_, dest);
}
MigrationState OutgoingMigration::GetState() const {

17
src/server/journal/streamer.cc
View file

@ -99,12 +99,6 @@ void RestoreStreamer::Cancel() {
JournalStreamer::Cancel();
}
RestoreStreamer::~RestoreStreamer() {
fiber_cancellation_.Cancel();
snapshot_fb_.JoinIfNeeded();
db_slice_->UnregisterOnChange(snapshot_version_);
}
bool RestoreStreamer::ShouldWrite(const journal::JournalItem& item) const {
if (!item.slot.has_value()) {
return false;
@ -122,11 +116,10 @@ bool RestoreStreamer::ShouldWrite(SlotId slot_id) const {
}
void RestoreStreamer::WriteBucket(PrimeTable::bucket_iterator it) {
DCHECK_LT(it.GetVersion(), snapshot_version_);
it.SetVersion(snapshot_version_);
bool is_data_present = false;
{
if (it.GetVersion() < snapshot_version_) {
it.SetVersion(snapshot_version_);
FiberAtomicGuard fg; // Can't switch fibers because that could invalidate iterator
string key_buffer; // we can reuse it
for (; !it.is_done(); ++it) {
@ -157,9 +150,7 @@ void RestoreStreamer::OnDbChange(DbIndex db_index, const DbSlice::ChangeReq& req
PrimeTable* table = db_slice_->GetTables(0).first;
if (const PrimeTable::bucket_iterator* bit = req.update()) {
if (bit->GetVersion() < snapshot_version_) {
WriteBucket(*bit);
}
WriteBucket(*bit);
} else {
string_view key = get<string_view>(req.change);
table->CVCUponInsert(snapshot_version_, key, [this](PrimeTable::bucket_iterator it) {

3
src/server/journal/streamer.h
View file

@ -57,14 +57,13 @@ class RestoreStreamer : public JournalStreamer {
Context* cntx);
void Start(io::Sink* dest) override;
// Cancel() must be called if Start() is called
void Cancel() override;
bool IsSnapshotFinished() const {
return snapshot_finished_;
}
~RestoreStreamer();
private:
void OnDbChange(DbIndex db_index, const DbSlice::ChangeReq& req);
bool ShouldWrite(const journal::JournalItem& item) const override;