From bacc76a057bcd90d61bfe3584bd3cdbadc748364 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Wed, 8 Apr 2020 22:00:41 +0200 Subject: [PATCH] Some work in actually storing things --- Cargo.lock | 46 +++++------------ Cargo.toml | 3 +- src/api_server.rs | 70 +++++++++++--------------- src/data.rs | 24 +-------- src/main.rs | 5 ++ src/membership.rs | 7 ++- src/proto.rs | 3 +- src/rpc_server.rs | 24 ++++++--- src/server.rs | 55 ++++++++++++++++++-- src/table.rs | 116 +++++++++++++++++++++++++++++++++++++++++++ src/version_table.rs | 59 ++++++++++++++++++++++ 11 files changed, 302 insertions(+), 110 deletions(-) create mode 100644 src/table.rs create mode 100644 src/version_table.rs diff --git a/Cargo.lock b/Cargo.lock index bed369c7..b5dd999f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5,6 +5,16 @@ name = "arc-swap" version = "0.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "async-trait" +version = "0.1.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.10 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.17 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "autocfg" version = "1.0.0" @@ -270,6 +280,7 @@ dependencies = [ name = "garage" version = "0.1.0" dependencies = [ + "async-trait 0.1.30 (registry+https://github.com/rust-lang/crates.io-index)", "bincode 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)", "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", "err-derive 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)", @@ -280,12 +291,12 @@ dependencies = [ "hex 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)", "http 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)", "hyper 0.13.4 (registry+https://github.com/rust-lang/crates.io-index)", - "kv 0.20.2 (registry+https://github.com/rust-lang/crates.io-index)", "rand 0.7.3 (registry+https://github.com/rust-lang/crates.io-index)", "rmp-serde 0.14.3 (registry+https://github.com/rust-lang/crates.io-index)", "serde 1.0.106 (registry+https://github.com/rust-lang/crates.io-index)", "serde_bytes 0.11.3 (registry+https://github.com/rust-lang/crates.io-index)", "sha2 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)", + "sled 0.31.0 (registry+https://github.com/rust-lang/crates.io-index)", "structopt 0.3.12 (registry+https://github.com/rust-lang/crates.io-index)", "tokio 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)", "toml 0.5.6 (registry+https://github.com/rust-lang/crates.io-index)", @@ -425,17 +436,6 @@ dependencies = [ "winapi-build 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)", ] -[[package]] -name = "kv" -version = "0.20.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -dependencies = [ - "serde 1.0.106 (registry+https://github.com/rust-lang/crates.io-index)", - "sled 0.31.0 (registry+https://github.com/rust-lang/crates.io-index)", - "thiserror 1.0.14 (registry+https://github.com/rust-lang/crates.io-index)", - "toml 0.5.6 (registry+https://github.com/rust-lang/crates.io-index)", -] - [[package]] name = "lazy_static" version = "1.4.0" @@ -895,24 +895,6 @@ dependencies = [ "unicode-width 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)", ] -[[package]] -name = "thiserror" -version = "1.0.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -dependencies = [ - "thiserror-impl 1.0.14 (registry+https://github.com/rust-lang/crates.io-index)", -] - -[[package]] -name = "thiserror-impl" -version = "1.0.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -dependencies = [ - "proc-macro2 1.0.10 (registry+https://github.com/rust-lang/crates.io-index)", - "quote 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", - "syn 1.0.17 (registry+https://github.com/rust-lang/crates.io-index)", -] - [[package]] name = "time" version = "0.1.42" @@ -1066,6 +1048,7 @@ dependencies = [ [metadata] "checksum arc-swap 0.4.5 (registry+https://github.com/rust-lang/crates.io-index)" = "d663a8e9a99154b5fb793032533f6328da35e23aac63d5c152279aa8ba356825" +"checksum async-trait 0.1.30 (registry+https://github.com/rust-lang/crates.io-index)" = "da71fef07bc806586090247e971229289f64c210a278ee5ae419314eb386b31d" "checksum autocfg 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "f8aac770f1885fd7e387acedd76065302551364496e46b3dd00860b2f8359b9d" "checksum bincode 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5753e2a71534719bf3f4e57006c3a4f0d2c672a4b676eec84161f763eca87dbf" "checksum bitflags 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693" @@ -1112,7 +1095,6 @@ dependencies = [ "checksum iovec 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "b2b3ea6ff95e175473f8ffe6a7eb7c00d054240321b84c57051175fe3c1e075e" "checksum itoa 0.4.5 (registry+https://github.com/rust-lang/crates.io-index)" = "b8b7a7c0c47db5545ed3fef7468ee7bb5b74691498139e4b3f6a20685dc6dd8e" "checksum kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d" -"checksum kv 0.20.2 (registry+https://github.com/rust-lang/crates.io-index)" = "cde1d1ce05b9169af359041cf834075ca4478f5917e6198d9454773e98aae143" "checksum lazy_static 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" "checksum libc 0.2.68 (registry+https://github.com/rust-lang/crates.io-index)" = "dea0c0405123bba743ee3f91f49b1c7cfb684eef0da0a50110f758ccf24cdff0" "checksum lock_api 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "79b2de95ecb4691949fea4716ca53cdbcfccb2c612e19644a8bad05edcf9f47b" @@ -1166,8 +1148,6 @@ dependencies = [ "checksum syn-mid 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7be3539f6c128a931cf19dcee741c1af532c7fd387baa739c03dd2e96479338a" "checksum synstructure 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "67656ea1dc1b41b1451851562ea232ec2e5a80242139f7e679ceccfb5d61f545" "checksum textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060" -"checksum thiserror 1.0.14 (registry+https://github.com/rust-lang/crates.io-index)" = "f0570dc61221295909abdb95c739f2e74325e14293b2026b0a7e195091ec54ae" -"checksum thiserror-impl 1.0.14 (registry+https://github.com/rust-lang/crates.io-index)" = "227362df41d566be41a28f64401e07a043157c21c14b9785a0d8e256f940a8fd" "checksum time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)" = "db8dcfca086c1143c9270ac42a2bbd8a7ee477b78ac8e45b19abfb0cbede4b6f" "checksum tokio 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)" = "ee5a0dd887e37d37390c13ff8ac830f992307fe30a1fff0ab8427af67211ba28" "checksum tokio-macros 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)" = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389" diff --git a/Cargo.toml b/Cargo.toml index d8225737..51bdd070 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ edition = "2018" bytes = "0.4" http = "0.2" hyper = "0.13" -kv = "0.20" +sled = "0.31" futures = "0.3" futures-core = "0.3" futures-channel = "0.3" @@ -26,3 +26,4 @@ structopt = { version = "0.3", default-features = false } rand = "0.7" hex = "0.3" sha2 = "0.8" +async-trait = "0.1.30" diff --git a/src/api_server.rs b/src/api_server.rs index c1b4d81d..8acd15d8 100644 --- a/src/api_server.rs +++ b/src/api_server.rs @@ -9,21 +9,21 @@ use hyper::{Body, Method, Request, Response, Server, StatusCode}; use futures::future::Future; use crate::error::Error; -use crate::membership::System; use crate::data::*; use crate::proto::*; use crate::rpc_client::*; +use crate::server::Garage; -pub async fn run_api_server(sys: Arc, shutdown_signal: impl Future) -> Result<(), hyper::Error> { - let addr = ([0, 0, 0, 0], sys.config.api_port).into(); +pub async fn run_api_server(garage: Arc, shutdown_signal: impl Future) -> Result<(), hyper::Error> { + let addr = ([0, 0, 0, 0], garage.system.config.api_port).into(); let service = make_service_fn(|conn: &AddrStream| { - let sys = sys.clone(); + let garage = garage.clone(); let client_addr = conn.remote_addr(); async move { Ok::<_, Error>(service_fn(move |req: Request| { - let sys = sys.clone(); - handler(sys, req, client_addr) + let garage = garage.clone(); + handler(garage, req, client_addr) })) } }); @@ -36,8 +36,8 @@ pub async fn run_api_server(sys: Arc, shutdown_signal: impl Future, req: Request, addr: SocketAddr) -> Result, Error> { - match handler_inner(sys, req, addr).await { +async fn handler(garage: Arc, req: Request, addr: SocketAddr) -> Result, Error> { + match handler_inner(garage, req, addr).await { Ok(x) => Ok(x), Err(Error::BadRequest(e)) => { let mut bad_request = Response::new(Body::from(format!("{}\n", e))); @@ -53,7 +53,7 @@ async fn handler(sys: Arc, req: Request, addr: SocketAddr) -> Resu } } -async fn handler_inner(sys: Arc, req: Request, addr: SocketAddr) -> Result, Error> { +async fn handler_inner(garage: Arc, req: Request, addr: SocketAddr) -> Result, Error> { eprintln!("{} {} {}", addr, req.method(), req.uri()); let bucket = req.headers() @@ -75,7 +75,7 @@ async fn handler_inner(sys: Arc, req: Request, addr: SocketAddr) - .map(|x| x.to_str()) .unwrap_or(Ok("blob"))? .to_string(); - let version_uuid = handle_put(sys, &mime_type, &bucket, &key, req.into_body()).await?; + let version_uuid = handle_put(garage, &mime_type, &bucket, &key, req.into_body()).await?; Ok(Response::new(Body::from( format!("Version UUID: {:?}", version_uuid), ))) @@ -84,22 +84,24 @@ async fn handler_inner(sys: Arc, req: Request, addr: SocketAddr) - } } -async fn handle_put(sys: Arc, +async fn handle_put(garage: Arc, mime_type: &str, bucket: &str, key: &str, body: Body) -> Result { let version_uuid = gen_uuid(); - let mut chunker = BodyChunker::new(body, sys.config.block_size); + let mut chunker = BodyChunker::new(body, garage.system.config.block_size); let first_block = match chunker.next().await? { Some(x) => x, None => return Err(Error::BadRequest(format!("Empty body"))), }; - let mut version = VersionMeta{ + let version_key = VersionMetaKey{ bucket: bucket.to_string(), key: key.to_string(), + }; + let mut version_value = VersionMetaValue { timestamp: now_msec(), uuid: version_uuid.clone(), mime_type: mime_type.to_string(), @@ -107,27 +109,17 @@ async fn handle_put(sys: Arc, is_complete: false, data: VersionData::DeleteMarker, }; - let version_who = sys.members.read().await - .walk_ring(&version_uuid, sys.config.meta_replication_factor); if first_block.len() < INLINE_THRESHOLD { - version.data = VersionData::Inline(first_block); - version.is_complete = true; - rpc_try_call_many(sys.clone(), - &version_who[..], - &Message::AdvertiseVersion(version), - (sys.config.meta_replication_factor+1)/2, - DEFAULT_TIMEOUT).await?; + version_value.data = VersionData::Inline(first_block); + version_value.is_complete = true; + garage.version_table.insert(&version_key, &version_value).await?; return Ok(version_uuid) } let first_block_hash = hash(&first_block[..]); - version.data = VersionData::FirstBlock(first_block_hash); - rpc_try_call_many(sys.clone(), - &version_who[..], - &Message::AdvertiseVersion(version.clone()), - (sys.config.meta_replication_factor+1)/2, - DEFAULT_TIMEOUT).await?; + version_value.data = VersionData::FirstBlock(first_block_hash); + garage.version_table.insert(&version_key, &version_value).await?; let block_meta = BlockMeta{ version_uuid: version_uuid.clone(), @@ -135,7 +127,7 @@ async fn handle_put(sys: Arc, hash: hash(&first_block[..]), }; let mut next_offset = first_block.len(); - let mut put_curr_block = put_block(sys.clone(), block_meta, first_block); + let mut put_curr_block = put_block(garage.clone(), block_meta, first_block); loop { let (_, next_block) = futures::try_join!(put_curr_block, chunker.next())?; if let Some(block) = next_block { @@ -145,7 +137,7 @@ async fn handle_put(sys: Arc, hash: hash(&block[..]), }; next_offset += block.len(); - put_curr_block = put_block(sys.clone(), block_meta, block); + put_curr_block = put_block(garage.clone(), block_meta, block); } else { break; } @@ -153,25 +145,21 @@ async fn handle_put(sys: Arc, // TODO: if at any step we have an error, we should undo everything we did - version.is_complete = true; - rpc_try_call_many(sys.clone(), - &version_who[..], - &Message::AdvertiseVersion(version), - (sys.config.meta_replication_factor+1)/2, - DEFAULT_TIMEOUT).await?; + version_value.is_complete = true; + garage.version_table.insert(&version_key, &version_value).await?; Ok(version_uuid) } -async fn put_block(sys: Arc, meta: BlockMeta, data: Vec) -> Result<(), Error> { - let who = sys.members.read().await - .walk_ring(&meta.hash, sys.config.meta_replication_factor); - rpc_try_call_many(sys.clone(), +async fn put_block(garage: Arc, meta: BlockMeta, data: Vec) -> Result<(), Error> { + let who = garage.system.members.read().await + .walk_ring(&meta.hash, garage.system.config.meta_replication_factor); + rpc_try_call_many(garage.system.clone(), &who[..], &Message::PutBlock(PutBlockMessage{ meta, data, }), - (sys.config.meta_replication_factor+1)/2, + (garage.system.config.meta_replication_factor+1)/2, DEFAULT_TIMEOUT).await?; Ok(()) } diff --git a/src/data.rs b/src/data.rs index 3c71b782..bbe9aa1d 100644 --- a/src/data.rs +++ b/src/data.rs @@ -106,7 +106,7 @@ pub struct NetworkConfigEntry { // Data management -pub const INLINE_THRESHOLD: usize = 2048; +pub const INLINE_THRESHOLD: usize = 3072; #[derive(Debug, Serialize, Deserialize)] pub struct SplitpointMeta { @@ -118,27 +118,7 @@ pub struct SplitpointMeta { pub deleted: bool, } -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct VersionMeta { - pub bucket: String, - pub key: String, - - pub timestamp: u64, - pub uuid: UUID, - - pub mime_type: String, - pub size: u64, - pub is_complete: bool, - - pub data: VersionData, -} - -#[derive(Clone, Debug, Serialize, Deserialize)] -pub enum VersionData { - DeleteMarker, - Inline(#[serde(with="serde_bytes")] Vec), - FirstBlock(Hash), -} +pub use crate::version_table::*; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct BlockMeta { diff --git a/src/main.rs b/src/main.rs index 1e4107c2..aa0f23dc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,12 @@ mod error; mod data; mod proto; + mod membership; +mod table; + +mod version_table; + mod server; mod rpc_server; mod rpc_client; diff --git a/src/membership.rs b/src/membership.rs index 69805f2a..e1eeae41 100644 --- a/src/membership.rs +++ b/src/membership.rs @@ -124,6 +124,11 @@ impl Members { i - 1 } }; + + self.walk_ring_from_pos(start, n) + } + + fn walk_ring_from_pos(&self, start: usize, n: usize) -> Vec { let mut ret = vec![]; let mut datacenters = vec![]; @@ -143,7 +148,7 @@ impl Members { } ret - } + } } fn read_network_config(metadata_dir: &PathBuf) -> Result { diff --git a/src/proto.rs b/src/proto.rs index 04b8e2b2..99ab8fbe 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -18,7 +18,8 @@ pub enum Message { AdvertiseConfig(NetworkConfig), PutBlock(PutBlockMessage), - AdvertiseVersion(VersionMeta), + + TableRPC(String, #[serde(with = "serde_bytes")] Vec), } #[derive(Debug, Serialize, Deserialize)] diff --git a/src/rpc_server.rs b/src/rpc_server.rs index 55c7482b..eda300c4 100644 --- a/src/rpc_server.rs +++ b/src/rpc_server.rs @@ -1,6 +1,7 @@ use std::net::SocketAddr; use std::sync::Arc; +use futures_util::future::FutureExt; use bytes::IntoBuf; use hyper::service::{make_service_fn, service_fn}; use hyper::server::conn::AddrStream; @@ -9,7 +10,7 @@ use futures::future::Future; use crate::error::Error; use crate::proto::Message; -use crate::membership::System; +use crate::server::Garage; fn err_to_msg(x: Result) -> Message { match x { @@ -18,7 +19,7 @@ fn err_to_msg(x: Result) -> Message { } } -async fn handler(sys: Arc, req: Request, addr: SocketAddr) -> Result, Error> { +async fn handler(garage: Arc, req: Request, addr: SocketAddr) -> Result, Error> { if req.method() != &Method::POST { let mut bad_request = Response::default(); *bad_request.status_mut() = StatusCode::BAD_REQUEST; @@ -30,12 +31,21 @@ async fn handler(sys: Arc, req: Request, addr: SocketAddr) -> Resu eprintln!("RPC from {}: {:?}", addr, msg); + let sys = garage.system.clone(); let resp = err_to_msg(match &msg { Message::Ping(ping) => sys.handle_ping(&addr, ping).await, Message::PullStatus => sys.handle_pull_status().await, Message::PullConfig => sys.handle_pull_config().await, Message::AdvertiseNodesUp(adv) => sys.handle_advertise_nodes_up(adv).await, Message::AdvertiseConfig(adv) => sys.handle_advertise_config(adv).await, + Message::TableRPC(table, msg) => { + if let Some(rpc_handler) = garage.table_rpc_handlers.get(table) { + rpc_handler.handle(&msg[..]).await + .map(|rep| Message::TableRPC(table.to_string(), rep)) + } else { + Ok(Message::Error(format!("Unknown table: {}", table))) + } + } _ => Ok(Message::Error(format!("Unexpected message: {:?}", msg))), }); @@ -46,16 +56,16 @@ async fn handler(sys: Arc, req: Request, addr: SocketAddr) -> Resu } -pub async fn run_rpc_server(sys: Arc, shutdown_signal: impl Future) -> Result<(), hyper::Error> { - let bind_addr = ([0, 0, 0, 0], sys.config.rpc_port).into(); +pub async fn run_rpc_server(garage: Arc, shutdown_signal: impl Future) -> Result<(), hyper::Error> { + let bind_addr = ([0, 0, 0, 0], garage.system.config.rpc_port).into(); let service = make_service_fn(|conn: &AddrStream| { let client_addr = conn.remote_addr(); - let sys = sys.clone(); + let garage = garage.clone(); async move { Ok::<_, Error>(service_fn(move |req: Request| { - let sys = sys.clone(); - handler(sys, req, client_addr) + let garage = garage.clone(); + handler(garage, req, client_addr) })) } }); diff --git a/src/server.rs b/src/server.rs index d5da8c17..31f1cc28 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::io::{Read, Write}; use std::sync::Arc; use std::net::SocketAddr; @@ -6,10 +7,51 @@ use futures::channel::oneshot; use serde::Deserialize; use crate::data::*; +use crate::proto::*; use crate::error::Error; use crate::membership::System; use crate::api_server; use crate::rpc_server; +use crate::table::*; + +pub struct Garage { + pub db: sled::Db, + pub system: Arc, + + pub table_rpc_handlers: HashMap>, + + pub version_table: Arc>, +} + +impl Garage { + pub fn new(config: Config, id: UUID, db: sled::Db) -> Self { + let system = Arc::new(System::new(config, id)); + + let meta_rep_param = TableReplicationParams{ + replication_factor: system.config.meta_replication_factor, + write_quorum: (system.config.meta_replication_factor+1)/2, + read_quorum: (system.config.meta_replication_factor+1)/2, + timeout: DEFAULT_TIMEOUT, + }; + + let version_table = Arc::new(Table::new( + system.clone(), + &db, + "version".to_string(), + meta_rep_param.clone())); + + let mut garage = Self{ + db, + system: system.clone(), + table_rpc_handlers: HashMap::new(), + version_table, + }; + garage.table_rpc_handlers.insert( + garage.version_table.name.clone(), + garage.version_table.clone().rpc_handler()); + garage + } +} fn default_block_size() -> usize { 1048576 @@ -88,20 +130,25 @@ pub async fn run_server(config_file: PathBuf) -> Result<(), Error> { let config = read_config(config_file) .expect("Unable to read config file"); + let mut db_path = config.metadata_dir.clone(); + db_path.push("garage_metadata"); + let db = sled::open(db_path) + .expect("Unable to open DB"); + let id = gen_node_id(&config.metadata_dir) .expect("Unable to read or generate node ID"); println!("Node ID: {}", hex::encode(&id)); - let sys = Arc::new(System::new(config, id)); + let garage = Arc::new(Garage::new(config, id, db)); let (tx1, rx1) = oneshot::channel(); let (tx2, rx2) = oneshot::channel(); - let rpc_server = rpc_server::run_rpc_server(sys.clone(), wait_from(rx1)); - let api_server = api_server::run_api_server(sys.clone(), wait_from(rx2)); + let rpc_server = rpc_server::run_rpc_server(garage.clone(), wait_from(rx1)); + let api_server = api_server::run_api_server(garage.clone(), wait_from(rx2)); tokio::spawn(shutdown_signal(vec![tx1, tx2])); - tokio::spawn(sys.bootstrap()); + tokio::spawn(garage.system.clone().bootstrap()); futures::try_join!(rpc_server, api_server)?; Ok(()) diff --git a/src/table.rs b/src/table.rs new file mode 100644 index 00000000..5c8e93a5 --- /dev/null +++ b/src/table.rs @@ -0,0 +1,116 @@ +use std::marker::PhantomData; +use std::time::Duration; +use std::sync::Arc; +use serde::{Serialize, Deserialize}; +use async_trait::async_trait; + +use crate::error::Error; +use crate::proto::*; +use crate::data::*; +use crate::membership::System; +use crate::rpc_client::*; + + +pub struct Table { + phantom: PhantomData, + + pub name: String, + + pub system: Arc, + pub store: sled::Tree, + pub partitions: Vec, + + pub param: TableReplicationParams, +} + +#[derive(Clone)] +pub struct TableReplicationParams { + pub replication_factor: usize, + pub read_quorum: usize, + pub write_quorum: usize, + pub timeout: Duration, +} + +#[async_trait] +pub trait TableRpcHandler { + async fn handle(&self, rpc: &[u8]) -> Result, Error>; +} + +struct TableRpcHandlerAdapter { + table: Arc>, +} + +#[async_trait] +impl TableRpcHandler for TableRpcHandlerAdapter { + async fn handle(&self, rpc: &[u8]) -> Result, Error> { + let msg = rmp_serde::decode::from_read_ref::<_, TableRPC>(rpc)?; + let rep = self.table.handle(msg).await?; + Ok(rmp_serde::encode::to_vec_named(&rep)?) + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub enum TableRPC { + Update(F::K, F::V), +} + +pub struct Partition { + pub begin: Hash, + pub end: Hash, + pub other_nodes: Vec, +} + +pub trait KeyHash { + fn hash(&self) -> Hash; +} + +pub trait ValueMerge { + fn merge(&mut self, other: &Self); +} + +#[async_trait] +pub trait TableFormat: Send + Sync { + type K: Clone + Serialize + for<'de> Deserialize<'de> + KeyHash + Send + Sync; + type V: Clone + Serialize + for<'de> Deserialize<'de> + ValueMerge + Send + Sync; + + async fn updated(&self, key: &Self::K, old: Option<&Self::V>, new: &Self::V); +} + +impl Table { + pub fn new(system: Arc, db: &sled::Db, name: String, param: TableReplicationParams) -> Self { + let store = db.open_tree(&name) + .expect("Unable to open DB tree"); + Self{ + phantom: PhantomData::default(), + name, + system, + store, + partitions: Vec::new(), + param, + } + } + + pub fn rpc_handler(self: Arc) -> Box { + Box::new(TableRpcHandlerAdapter::{ table: self }) + } + + pub async fn insert(&self, k: &F::K, v: &F::V) -> Result<(), Error> { + unimplemented!(); + + let hash = k.hash(); + let who = self.system.members.read().await + .walk_ring(&hash, self.param.replication_factor); + + let msg = rmp_serde::encode::to_vec_named(&TableRPC::::Update(k.clone(), v.clone()))?; + rpc_try_call_many(self.system.clone(), + &who[..], + &Message::TableRPC(self.name.to_string(), msg), + self.param.write_quorum, + self.param.timeout).await?; + Ok(()) + } + + async fn handle(&self, msg: TableRPC) -> Result, Error> { + unimplemented!() + } +} diff --git a/src/version_table.rs b/src/version_table.rs new file mode 100644 index 00000000..d857ac12 --- /dev/null +++ b/src/version_table.rs @@ -0,0 +1,59 @@ +use std::sync::Arc; +use serde::{Serialize, Deserialize}; +use async_trait::async_trait; + +use crate::data::*; +use crate::table::*; +use crate::membership::System; + + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct VersionMetaKey { + pub bucket: String, + pub key: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct VersionMetaValue { + pub timestamp: u64, + pub uuid: UUID, + + pub mime_type: String, + pub size: u64, + pub is_complete: bool, + + pub data: VersionData, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum VersionData { + DeleteMarker, + Inline(#[serde(with="serde_bytes")] Vec), + FirstBlock(Hash), +} + +pub struct VersionTable { + system: Arc, +} + +impl KeyHash for VersionMetaKey { + fn hash(&self) -> Hash { + hash(self.bucket.as_bytes()) + } +} + +impl ValueMerge for VersionMetaValue { + fn merge(&mut self, other: &Self) { + unimplemented!() + } +} + +#[async_trait] +impl TableFormat for VersionTable { + type K = VersionMetaKey; + type V = VersionMetaValue; + + async fn updated(&self, key: &Self::K, old: Option<&Self::V>, new: &Self::V) { + unimplemented!() + } +}