diff options
author | Alex Auvolat <alex@adnab.me> | 2023-09-21 15:32:25 +0200 |
---|---|---|
committer | Alex Auvolat <alex@adnab.me> | 2023-09-21 15:35:31 +0200 |
commit | f97168f80567f43e15cf236092703e6ae5d8dc2e (patch) | |
tree | ca9b06fa1d9bf3af9b33c5db0a8c09a53b403fa9 /src/table | |
parent | fd7d8fec59c617b40e480ff855894cf35fdcfb40 (diff) | |
download | garage-f97168f80567f43e15cf236092703e6ae5d8dc2e.tar.gz garage-f97168f80567f43e15cf236092703e6ae5d8dc2e.zip |
garage_db: refactor transactions and add on_commit mechanism
Diffstat (limited to 'src/table')
-rw-r--r-- | src/table/data.rs | 16 | ||||
-rw-r--r-- | src/table/merkle.rs | 4 | ||||
-rw-r--r-- | src/table/queue.rs | 2 |
3 files changed, 11 insertions, 11 deletions
diff --git a/src/table/data.rs b/src/table/data.rs index 26101da4..665f4142 100644 --- a/src/table/data.rs +++ b/src/table/data.rs @@ -203,14 +203,14 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> { ) -> Result<Option<F::E>, Error> { let tree_key = self.tree_key(partition_key, sort_key); - let changed = self.store.db().transaction(|mut tx| { + let changed = self.store.db().transaction(|tx| { let (old_entry, old_bytes, new_entry) = match tx.get(&self.store, &tree_key)? { Some(old_bytes) => { let old_entry = self.decode_entry(&old_bytes).map_err(db::TxError::Abort)?; - let new_entry = update_fn(&mut tx, Some(old_entry.clone()))?; + let new_entry = update_fn(tx, Some(old_entry.clone()))?; (Some(old_entry), Some(old_bytes), new_entry) } - None => (None, None, update_fn(&mut tx, None)?), + None => (None, None, update_fn(tx, None)?), }; // Changed can be true in two scenarios @@ -233,7 +233,7 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> { tx.insert(&self.store, &tree_key, new_bytes)?; self.instance - .updated(&mut tx, old_entry.as_ref(), Some(&new_entry))?; + .updated(tx, old_entry.as_ref(), Some(&new_entry))?; Ok(Some((new_entry, new_bytes_hash))) } else { @@ -270,14 +270,14 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> { let removed = self .store .db() - .transaction(|mut tx| match tx.get(&self.store, k)? { + .transaction(|tx| match tx.get(&self.store, k)? { Some(cur_v) if cur_v == v => { let old_entry = self.decode_entry(v).map_err(db::TxError::Abort)?; tx.remove(&self.store, k)?; tx.insert(&self.merkle_todo, k, vec![])?; - self.instance.updated(&mut tx, Some(&old_entry), None)?; + self.instance.updated(tx, Some(&old_entry), None)?; Ok(true) } _ => Ok(false), @@ -298,14 +298,14 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> { let removed = self .store .db() - .transaction(|mut tx| match tx.get(&self.store, k)? { + .transaction(|tx| match tx.get(&self.store, k)? { Some(cur_v) if blake2sum(&cur_v[..]) == vhash => { let old_entry = self.decode_entry(&cur_v[..]).map_err(db::TxError::Abort)?; tx.remove(&self.store, k)?; tx.insert(&self.merkle_todo, k, vec![])?; - self.instance.updated(&mut tx, Some(&old_entry), None)?; + self.instance.updated(tx, Some(&old_entry), None)?; Ok(true) } _ => Ok(false), diff --git a/src/table/merkle.rs b/src/table/merkle.rs index e86d0251..4577f872 100644 --- a/src/table/merkle.rs +++ b/src/table/merkle.rs @@ -108,9 +108,9 @@ impl<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> { self.data .merkle_tree .db() - .transaction(|mut tx| self.update_item_rec(&mut tx, k, &khash, &key, new_vhash))?; + .transaction(|tx| self.update_item_rec(tx, k, &khash, &key, new_vhash))?; - let deleted = self.data.merkle_todo.db().transaction(|mut tx| { + let deleted = self.data.merkle_todo.db().transaction(|tx| { let remove = matches!(tx.get(&self.data.merkle_todo, k)?, Some(ov) if ov == vhash_by); if remove { tx.remove(&self.data.merkle_todo, k)?; diff --git a/src/table/queue.rs b/src/table/queue.rs index 096ac8b4..ffe0a4a7 100644 --- a/src/table/queue.rs +++ b/src/table/queue.rs @@ -53,7 +53,7 @@ impl<F: TableSchema, R: TableReplication> Worker for InsertQueueWorker<F, R> { self.0.insert_many(values).await?; - self.0.data.insert_queue.db().transaction(|mut tx| { + self.0.data.insert_queue.db().transaction(|tx| { for (k, v) in kv_pairs.iter() { if let Some(v2) = tx.get(&self.0.data.insert_queue, k)? { if &v2 == v { |