Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 2 commits into from
Jan 29, 2024
Merged

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7da3233
fix(cluster): fix incorrect version checking and resource double free
BorysTheDev Jan 29, 2024
c299edc
fix: add versiong check
BorysTheDev Jan 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions src/server/cluster/outgoing_slot_migration.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down
8 changes: 1 addition & 7 deletions src/server/journal/streamer.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -122,7 +116,7 @@ bool RestoreStreamer::ShouldWrite(SlotId slot_id) const {
}

void RestoreStreamer::WriteBucket(PrimeTable::bucket_iterator it) {
DCHECK_LT(it.GetVersion(), snapshot_version_);
DCHECK_LE(it.GetVersion(), snapshot_version_);
BorysTheDev marked this conversation as resolved.
Show resolved Hide resolved
it.SetVersion(snapshot_version_);

bool is_data_present = false;
Expand Down
3 changes: 1 addition & 2 deletions src/server/journal/streamer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading