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

fix(RestoreStreamer): Fix a few glitches (#2452)

Browse source
This commit is contained in:
Shahar Mike 2024-01-22 10:37:16 +02:00 • committed by GitHub
parent 07a6dc0712
commit 7debe3c685
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 19 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -76,6 +76,7 @@ void RestoreStreamer::Start(io::Sink* dest) {
return; return;
cursor = pt->Traverse(cursor, absl::bind_front(&RestoreStreamer::WriteBucket, this)); cursor = pt->Traverse(cursor, absl::bind_front(&RestoreStreamer::WriteBucket, this));
++last_yield;
if (last_yield >= 100) { if (last_yield >= 100) {
ThisFiber::Yield(); ThisFiber::Yield();
@ -111,22 +112,28 @@ void RestoreStreamer::WriteBucket(PrimeTable::bucket_iterator it) {
DCHECK_LT(it.GetVersion(), snapshot_version_); DCHECK_LT(it.GetVersion(), snapshot_version_);
it.SetVersion(snapshot_version_); it.SetVersion(snapshot_version_);
while (!it.is_done()) { {
const auto& pv = it->second; FiberAtomicGuard fg; // Can't switch fibers because that could invalidate iterator
string key_buffer; while (!it.is_done()) {
string_view key = it->first.GetSlice(&key_buffer); const auto& pv = it->second;
uint64_t expire = 0; string key_buffer;
if (pv.HasExpire()) { string_view key = it->first.GetSlice(&key_buffer);
auto eit = db_slice_->databases()[0]->expire.Find(it->first);
expire = db_slice_->ExpireTime(eit); uint64_t expire = 0;
if (pv.HasExpire()) {
auto eit = db_slice_->databases()[0]->expire.Find(it->first);
expire = db_slice_->ExpireTime(eit);
}
WriteEntry(key, pv, expire);
++it;
} }
WriteEntry(key, pv, expire);
++it;
} }
NotifyWritten(true);
} }
void RestoreStreamer::OnDbChange(DbIndex db_index, const DbSlice::ChangeReq& req) { void RestoreStreamer::OnDbChange(DbIndex db_index, const DbSlice::ChangeReq& req) {