aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorAlex Auvolat <alex@adnab.me>2022-06-24 11:04:55 +0200
committerAlex Auvolat <alex@adnab.me>2022-06-24 11:04:55 +0200
commite7810e9cb3cdbe6aaecddddd1146bf15e5b50c7c (patch)
tree6a7ae3528371e32a2eea722e0ff214b46a3a9bc8
parent0837b3dacd994b35dbccbfefdd68be044eb4b5ac (diff)
downloadgarage-e7810e9cb3cdbe6aaecddddd1146bf15e5b50c7c.tar.gz
garage-e7810e9cb3cdbe6aaecddddd1146bf15e5b50c7c.zip
Smaller batches for index counter propagation
-rw-r--r--src/block/repair.rs1
-rw-r--r--src/model/index_counter.rs7
2 files changed, 6 insertions, 2 deletions
diff --git a/src/block/repair.rs b/src/block/repair.rs
index a5c01629..97989780 100644
--- a/src/block/repair.rs
+++ b/src/block/repair.rs
@@ -53,6 +53,7 @@ impl Worker for RepairWorker {
// This is mostly because the Rust bindings for SQLite assume a worst-case scenario
// where SQLite is not compiled in thread-safe mode, so we have to wrap everything
// in a mutex (see db/sqlite_adapter.rs and discussion in PR #322).
+ // TODO: maybe do this with tokio::task::spawn_blocking ?
let mut batch_of_hashes = vec![];
let start_bound = match self.next_start.as_ref() {
None => Bound::Unbounded,
diff --git a/src/model/index_counter.rs b/src/model/index_counter.rs
index 474ec12c..9d5aa955 100644
--- a/src/model/index_counter.rs
+++ b/src/model/index_counter.rs
@@ -429,7 +429,8 @@ impl<T: CountedItem> Worker for IndexPropagatorWorker<T> {
};
if !self.buf.is_empty() {
- let entries = self.buf.iter().map(|(_k, v)| v);
+ let entries_k = self.buf.keys().take(100).cloned().collect::<Vec<_>>();
+ let entries = entries_k.iter().map(|k| self.buf.get(k).unwrap());
if let Err(e) = self.index_counter.table.insert_many(entries).await {
self.errors += 1;
if self.errors >= 2 && *must_exit.borrow() {
@@ -441,7 +442,9 @@ impl<T: CountedItem> Worker for IndexPropagatorWorker<T> {
// things to go back to normal
return Err(e);
} else {
- self.buf.clear();
+ for k in entries_k {
+ self.buf.remove(&k);
+ }
self.errors = 0;
}