AcctIdx: when disk index insert needs to grow, drop locks, grow, then restart flush (#20333)
This commit is contained in:
parent
acfe76b622
commit
e31c065544
|
@ -664,6 +664,8 @@ impl<T: IndexValue> InMemAccountsIndex<T> {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// may have to loop if disk has to grow and we have to restart
|
||||||
|
loop {
|
||||||
let mut removes = Vec::default();
|
let mut removes = Vec::default();
|
||||||
let mut removes_random = Vec::default();
|
let mut removes_random = Vec::default();
|
||||||
let disk = self.storage.disk.as_ref().unwrap();
|
let disk = self.storage.disk.as_ref().unwrap();
|
||||||
|
@ -671,6 +673,7 @@ impl<T: IndexValue> InMemAccountsIndex<T> {
|
||||||
let mut updates = Vec::default();
|
let mut updates = Vec::default();
|
||||||
let m = Measure::start("flush_scan");
|
let m = Measure::start("flush_scan");
|
||||||
let mut flush_entries_updated_on_disk = 0;
|
let mut flush_entries_updated_on_disk = 0;
|
||||||
|
let mut disk_resize = Ok(());
|
||||||
// scan and update loop
|
// scan and update loop
|
||||||
{
|
{
|
||||||
let map = self.map().read().unwrap();
|
let map = self.map().read().unwrap();
|
||||||
|
@ -698,11 +701,21 @@ impl<T: IndexValue> InMemAccountsIndex<T> {
|
||||||
// it is possible that the item in the cache is marked as dirty while these updates are happening. That is ok.
|
// it is possible that the item in the cache is marked as dirty while these updates are happening. That is ok.
|
||||||
let m = Measure::start("flush_update");
|
let m = Measure::start("flush_update");
|
||||||
for (k, v) in updates.into_iter() {
|
for (k, v) in updates.into_iter() {
|
||||||
|
if disk_resize.is_ok() {
|
||||||
if v.dirty() {
|
if v.dirty() {
|
||||||
continue; // marked dirty after we grabbed it above, so handle this the next time this bucket is flushed
|
continue; // marked dirty after we grabbed it above, so handle this the next time this bucket is flushed
|
||||||
}
|
}
|
||||||
flush_entries_updated_on_disk += 1;
|
flush_entries_updated_on_disk += 1;
|
||||||
disk.insert(self.bin, &k, (&v.slot_list.read().unwrap(), v.ref_count()));
|
disk_resize = disk.try_insert(
|
||||||
|
self.bin,
|
||||||
|
&k,
|
||||||
|
(&v.slot_list.read().unwrap(), v.ref_count()),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if disk_resize.is_err() {
|
||||||
|
// disk needs to resize, so mark all unprocessed items as dirty again so we pick them up after the resize
|
||||||
|
v.set_dirty(true);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Self::update_time_stat(&self.stats().flush_update_us, m);
|
Self::update_time_stat(&self.stats().flush_update_us, m);
|
||||||
Self::update_stat(
|
Self::update_stat(
|
||||||
|
@ -712,6 +725,8 @@ impl<T: IndexValue> InMemAccountsIndex<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
let m = Measure::start("flush_remove");
|
let m = Measure::start("flush_remove");
|
||||||
|
match disk_resize {
|
||||||
|
Ok(_) => {
|
||||||
if !self.flush_remove_from_cache(removes, current_age, startup, false) {
|
if !self.flush_remove_from_cache(removes, current_age, startup, false) {
|
||||||
iterate_for_age = false; // did not make it all the way through this bucket, so didn't handle age completely
|
iterate_for_age = false; // did not make it all the way through this bucket, so didn't handle age completely
|
||||||
}
|
}
|
||||||
|
@ -725,6 +740,15 @@ impl<T: IndexValue> InMemAccountsIndex<T> {
|
||||||
assert_eq!(current_age, self.storage.current_age());
|
assert_eq!(current_age, self.storage.current_age());
|
||||||
self.set_has_aged(current_age);
|
self.set_has_aged(current_age);
|
||||||
}
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
// grow the bucket, outside of all in-mem locks.
|
||||||
|
// then, loop to try again
|
||||||
|
disk.grow(self.bin, err);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// remove keys in 'removes' from in-mem cache due to age
|
// remove keys in 'removes' from in-mem cache due to age
|
||||||
|
|
Loading…
Reference in New Issue