networkException/garage
1
0
Fork 0
forked from Deuxfleurs/garage
Code Pull requests Activity

Lock once for insert_many

Browse source
This commit is contained in:
Alex 2023-01-11 11:35:36 +01:00
parent 32aab06929
commit 09a3dad0f2
Signed by untrusted user: lx
GPG key ID: 0E496D15096376BE
1 changed files with 22 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
src/model/k2v/rpc.rs
View file

@ -7,7 +7,7 @@
use std::collections::{BTreeMap, HashMap}; use std::collections::{BTreeMap, HashMap};
use std::convert::TryInto; use std::convert::TryInto;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration; use std::time::Duration;
use async_trait::async_trait; use async_trait::async_trait;
@ -72,7 +72,12 @@ impl Rpc for K2VRpc {
pub struct K2VRpcHandler { pub struct K2VRpcHandler {
system: Arc<System>, system: Arc<System>,
item_table: Arc<Table<K2VItemTable, TableShardedReplication>>, item_table: Arc<Table<K2VItemTable, TableShardedReplication>>,
// Using a mutex on the local_timestamp_tree is not strictly necessary,
// but it helps to not try to do several inserts at the same time,
// which would create transaction conflicts and force many useless retries.
local_timestamp_tree: Mutex<db::Tree>, local_timestamp_tree: Mutex<db::Tree>,
endpoint: Arc<Endpoint<K2VRpc, Self>>, endpoint: Arc<Endpoint<K2VRpc, Self>>,
subscriptions: Arc<SubscriptionManager>, subscriptions: Arc<SubscriptionManager>,
} }
@ -323,7 +328,10 @@ impl K2VRpcHandler {
// ---- internal handlers ---- // ---- internal handlers ----
async fn handle_insert(&self, item: &InsertedItem) -> Result<K2VRpc, Error> { async fn handle_insert(&self, item: &InsertedItem) -> Result<K2VRpc, Error> {
let new = self.local_insert(item)?; let new = {
let local_timestamp_tree = self.local_timestamp_tree.lock().unwrap();
self.local_insert(&local_timestamp_tree, item)?
};
// Propagate to rest of network // Propagate to rest of network
if let Some(updated) = new { if let Some(updated) = new {
@ -336,13 +344,16 @@ impl K2VRpcHandler {
async fn handle_insert_many(&self, items: &[InsertedItem]) -> Result<K2VRpc, Error> { async fn handle_insert_many(&self, items: &[InsertedItem]) -> Result<K2VRpc, Error> {
let mut updated_vec = vec![]; let mut updated_vec = vec![];
{
let local_timestamp_tree = self.local_timestamp_tree.lock().unwrap();
for item in items { for item in items {
let new = self.local_insert(item)?; let new = self.local_insert(&local_timestamp_tree, item)?;
if let Some(updated) = new { if let Some(updated) = new {
updated_vec.push(updated); updated_vec.push(updated);
} }
} }
}
// Propagate to rest of network // Propagate to rest of network
if !updated_vec.is_empty() { if !updated_vec.is_empty() {
@ -352,12 +363,11 @@ impl K2VRpcHandler {
Ok(K2VRpc::Ok) Ok(K2VRpc::Ok)
} }
fn local_insert(&self, item: &InsertedItem) -> Result<Option<K2VItem>, Error> { fn local_insert(
// Using a mutex on the local_timestamp_tree is not strictly necessary, &self,
// but it helps to not try to do several inserts at the same time, local_timestamp_tree: &MutexGuard<'_, db::Tree>,
// which would create transaction conflicts and force many useless retries. item: &InsertedItem,
let local_timestamp_tree = self.local_timestamp_tree.lock().unwrap(); ) -> Result<Option<K2VItem>, Error> {
let now = now_msec(); let now = now_msec();
self.item_table self.item_table