wip manager

Author: MarinPostma

Cargo.lock

@@ -12,6 +12,7 @@ color-eyre = "0.6.2"
 futures = "0.3.28"
 hyper = { version = "0.14.27", features = ["h2"] }
 libsqlx = { version = "0.1.0", path = "../libsqlx" }
+moka = { version = "0.11.2", features = ["future"] }
 serde = { version = "1.0.166", features = ["derive"] }
 tokio = { version = "1.29.1", features = ["full"] }
 tracing = "0.1.37"

libsqlx-server/Cargo.toml

@@ -1,9 +1,22 @@
-use serde::{Serialize, Deserialize};
+use serde::{Deserialize, Serialize};
 
+/// Structural supertype of AllocConfig, used for checking the meta version. Subsequent version of
+/// AllocConfig need to conform to this prototype.
 #[derive(Debug, Serialize, Deserialize)]
-pub enum AllocConfig {
+struct ConfigVersion {
+    config_version: u32,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AllocConfig {
+    pub config_version: u32,
+    pub max_conccurent_connection: u32,
+    pub id: String,
+    pub db_config: DbConfig,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum DbConfig {
     Primary { },
-    Replica {
-        primary_node_id: String,
-    }
+    Replica { primary_node_id: String },
 }

libsqlx-server/src/allocation/config.rs

@@ -1,18 +1,24 @@
 use std::collections::HashMap;
+use std::path::PathBuf;
 
-use tokio::{sync::{mpsc, oneshot}, task::{JoinSet, block_in_place}};
+use libsqlx::Database as _;
+use libsqlx::libsql::{LibsqlDatabase, LogCompactor, LogFile, PrimaryType};
+use tokio::task::{block_in_place, JoinSet};
+use tokio::sync::{mpsc, oneshot};
+
+use self::config::{AllocConfig, DbConfig};
 
 pub mod config;
 
 type ExecFn = Box<dyn FnOnce(&mut dyn libsqlx::Connection) + Send>;
 
 #[derive(Clone)]
-struct ConnectionId {
+pub struct ConnectionId {
     id: u32,
     close_sender: mpsc::Sender<()>,
 }
 
-enum AllocationMessage {
+pub enum AllocationMessage {
     /// Execute callback against connection
     Exec {
         connection_id: ConnectionId,
@@ -22,30 +28,61 @@ enum AllocationMessage {
     NewConnExec {
         exec: ExecFn,
         ret: oneshot::Sender<ConnectionId>,
-    }
+    },
+}
+
+pub enum Database {
+    Primary(libsqlx::libsql::LibsqlDatabase<PrimaryType>),
 }
 
-enum Database {}
+struct Compactor;
+
+impl LogCompactor for Compactor {
+    fn should_compact(&self, _log: &LogFile) -> bool {
+        false
+    }
+
+    fn compact(
+        &self,
+        _log: LogFile,
+        _path: std::path::PathBuf,
+        _size_after: u32,
+    ) -> Result<(), Box<dyn std::error::Error + Sync + Send + 'static>> {
+        todo!()
+    }
+}
 
 impl Database {
+    pub fn from_config(config: &AllocConfig, path: PathBuf) -> Self {
+        match config.db_config {
+            DbConfig::Primary {} => {
+                let db = LibsqlDatabase::new_primary(path, Compactor, false).unwrap();
+                Self::Primary(db)
+            }
+            DbConfig::Replica { .. } => todo!(),
+        }
+    }
+
     fn connect(&self) -> Box<dyn libsqlx::Connection + Send> {
-        todo!();
+        match self {
+            Database::Primary(db) => Box::new(db.connect().unwrap()),
+        }
     }
 }
 
 pub struct Allocation {
-    inbox: mpsc::Receiver<AllocationMessage>,
-    database: Database,
+    pub inbox: mpsc::Receiver<AllocationMessage>,
+    pub database: Database,
     /// senders to the spawned connections
-    connections: HashMap<u32, mpsc::Sender<ExecFn>>,
+    pub connections: HashMap<u32, mpsc::Sender<ExecFn>>,
     /// spawned connection futures, returning their connection id on completion.
-    connections_futs: JoinSet<u32>,
-    next_conn_id: u32,
-    max_concurrent_connections: u32,
+    pub connections_futs: JoinSet<u32>,
+    pub next_conn_id: u32,
+    pub max_concurrent_connections: u32,
 }
 
 impl Allocation {
-    async fn run(mut self) {
+    pub async fn run(mut self) {
         loop {
             tokio::select! {
                 Some(msg) = self.inbox.recv() => {
@@ -86,23 +123,19 @@ impl Allocation {
             exec: exec_receiver,
         };
 
-
         self.connections_futs.spawn(conn.run());
         // This should never block!
         assert!(exec_sender.try_send(exec).is_ok());
         assert!(self.connections.insert(id, exec_sender).is_none());
 
-        ConnectionId {
-            id,
-            close_sender,
-        }
+        ConnectionId { id, close_sender }
     }
 
     fn next_conn_id(&mut self) -> u32 {
         loop {
             self.next_conn_id = self.next_conn_id.wrapping_add(1);
             if !self.connections.contains_key(&self.next_conn_id) {
-                return self.next_conn_id
+                return self.next_conn_id;
             }
         }
     }

libsqlx-server/src/allocation/mod.rs

@@ -0,0 +1,5 @@
+use uuid::Uuid;
+
+mod store;
+
+pub type DatabaseId = Uuid;

libsqlx-server/src/databases/mod.rs

@@ -0,0 +1,12 @@
+use std::collections::HashMap;
+
+use super::DatabaseId;
+
+pub enum Database {
+    Replica,
+    Primary,
+}
+
+pub struct DatabaseManager {
+    databases: HashMap<DatabaseId, Database>,
+}

libsqlx-server/src/databases/store.rs

@@ -6,7 +6,7 @@ use hyper::server::accept::Accept;
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncRead, AsyncWrite};
 
-use crate::{meta::MetaStore, allocation::config::AllocConfig};
+use crate::{allocation::config::AllocConfig, meta::MetaStore};
 
 pub struct AdminServerConfig {}
 
@@ -22,6 +22,7 @@ where
     let state = AdminServerState {
         meta_store: todo!(),
     };
+
     let app = Router::new()
         .route("/manage/allocation/create", post(allocate))
         .with_state(Arc::new(state));
@@ -36,7 +37,7 @@ where
 struct ErrorResponse {}
 
 #[derive(Serialize, Debug)]
-struct AllocateResp { }
+struct AllocateResp {}
 
 #[derive(Deserialize, Debug)]
 struct AllocateReq {

libsqlx-server/src/http/admin.rs

@@ -5,6 +5,7 @@ use tracing_subscriber::prelude::*;
 mod allocation;
 mod databases;
 mod http;
+mod manager;
 mod meta;
 
 #[tokio::main]

libsqlx-server/src/main.rs

@@ -0,0 +1,48 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use moka::future::Cache;
+use tokio::sync::mpsc;
+use tokio::task::JoinSet;
+
+use crate::allocation::{Allocation, AllocationMessage, Database};
+use crate::meta::MetaStore;
+
+pub struct Manager {
+    cache: Cache<String, mpsc::Sender<AllocationMessage>>,
+    meta_store: Arc<MetaStore>,
+    db_path: PathBuf,
+}
+
+const MAX_ALLOC_MESSAGE_QUEUE_LEN: usize = 32;
+
+impl Manager {
+    pub async fn alloc(&self, alloc_id: &str) -> mpsc::Sender<AllocationMessage> {
+        if let Some(sender) = self.cache.get(alloc_id) {
+            return sender.clone();
+        }
+
+        if let Some(config) = self.meta_store.meta(alloc_id).await {
+            let path = self.db_path.join("dbs").join(alloc_id);
+            tokio::fs::create_dir_all(&path).await.unwrap();
+            let (alloc_sender, inbox) = mpsc::channel(MAX_ALLOC_MESSAGE_QUEUE_LEN);
+            let alloc = Allocation {
+                inbox,
+                database: Database::from_config(&config, path),
+                connections: HashMap::new(),
+                connections_futs: JoinSet::new(),
+                next_conn_id: 0,
+                max_concurrent_connections: config.max_conccurent_connection,
+            };
+
+            tokio::spawn(alloc.run());
+
+            self.cache.insert(alloc_id.to_string(), alloc_sender.clone()).await;
+
+            return alloc_sender;
+        }
+
+        todo!("alloc doesn't exist")
+    }
+}

libsqlx-server/src/manager.rs

@@ -1,10 +1,38 @@
+use std::path::Path;
+
+use libsqlx::{libsql::LibsqlDatabase, Database as _, Connection, Program};
 use uuid::Uuid;
+use tokio::sync::mpsc;
 
 use crate::allocation::config::AllocConfig;
 
-pub struct MetaStore {}
+type ExecFn = Box<dyn FnOnce(&mut libsqlx::libsql::LibsqlConnection<()>)>;
+
+pub struct MetaStore {
+    sender: mpsc::Sender<ExecFn>,
+}
 
 impl MetaStore {
-    pub async fn allocate(&self, alloc_id: &str, meta: &AllocConfig) {}
+    pub fn new(path: &Path) -> Self {
+        let path = path.join("meta");
+        let (sender, receiver) = mpsc::channel(64);
+        tokio::task::spawn_blocking(move || {
+            let db = LibsqlDatabase::new_plain(path).unwrap();
+            let conn = db.connect().unwrap();
+            conn.execute("create table if not exist allocations (id TEXT, meta BLOB)");
+        });
+
+        Self { sender }
+
+    }
+    pub async fn allocate(&self, alloc_id: &str, meta: &AllocConfig) {
+        self.sender.send(Box::new(|conn| {
+            conn.execute_program(Program::seq(&["INSERT INTO tttt   u        "]), result_builder)
+
+        }));
+    }
     pub async fn deallocate(&self, alloc_id: Uuid) {}
+    pub async fn meta(&self, alloc_id: &str) -> Option<AllocConfig> {
+        todo!()
+    }
 }

libsqlx-server/src/meta.rs

@@ -1,4 +1,8 @@
-use crate::program::Program;
+use rusqlite::types::Value;
+
+use crate::QueryBuilderConfig;
+use crate::program::{Program, Step};
+use crate::query::Query;
 use crate::result_builder::ResultBuilder;
 
 #[derive(Debug, Clone)]
@@ -30,6 +34,52 @@ pub trait Connection {
 
     /// Parse the SQL statement and return information about it.
     fn describe(&self, sql: String) -> crate::Result<DescribeResponse>;
+
+    /// execute a single query
+    fn execute(&mut self, query: Query) -> crate::Result<Vec<Vec<Value>>> {
+        #[derive(Default)]
+        struct RowsBuilder {
+            error: Option<crate::error::Error>,
+            rows: Vec<Vec<Value>>,
+            current_row: Vec<Value>,
+        }
+
+        impl ResultBuilder for RowsBuilder {
+            fn init(&mut self, _config: &QueryBuilderConfig) -> std::result::Result<(), crate::QueryResultBuilderError> {
+                self.error = None;
+                self.rows.clear();
+                self.current_row.clear();
+
+                Ok(())
+            }
+
+            fn add_row_value(&mut self, v: rusqlite::types::ValueRef) -> Result<(), crate::QueryResultBuilderError> {
+                self.current_row.push(v.into());
+                Ok(())
+            }
+
+            fn finish_row(&mut self) -> Result<(), crate::QueryResultBuilderError> {
+                let row = std::mem::take(&mut self.current_row);
+                self.rows.push(row);
+
+                Ok(())
+            }
+
+            fn step_error(&mut self, error: crate::error::Error) -> Result<(), crate::QueryResultBuilderError> {
+                self.error.replace(error);
+                Ok(())
+            }
+        }
+
+        let pgm = Program::new(vec![Step { cond: None, query }]);
+        let mut builder = RowsBuilder::default();
+        self.execute_program(pgm, &mut builder)?;
+        if let Some(err) = builder.error.take() {
+            Err(err)
+        } else {
+            Ok(builder.rows)
+        } 
+    }
 }
 
 impl Connection for Box<dyn Connection> {

libsqlx/src/connection.rs

@@ -237,7 +237,11 @@ fn eval_cond(cond: &Cond, results: &[bool]) -> Result<bool> {
 }
 
 impl<C> Connection for LibsqlConnection<C> {
-    fn execute_program(&mut self, pgm: Program, builder: &mut dyn ResultBuilder) -> crate::Result<()> {
+    fn execute_program(
+        &mut self,
+        pgm: Program,
+        builder: &mut dyn ResultBuilder,
+    ) -> crate::Result<()> {
         self.run(pgm, builder)
     }