replace parking_lot with tokio::sync::Mutex in some places

sivukhin · sivukhin · commit 045bef9941dc · 2024-08-15T15:13:40.000+04:00
diff --git a/libsql-server/src/connection/program.rs b/libsql-server/src/connection/program.rs
@@ -363,11 +363,15 @@ pub fn check_program_auth(
             }
             StmtKind::Attach(ref ns) => {
                 ctx.auth.has_right(ns, Permission::AttachRead)?;
-                if !ctx.meta_store.handle(ns.clone()).get().allow_attach {
-                    return Err(Error::NotAuthorized(format!(
-                        "Namespace `{ns}` doesn't allow attach"
-                    )));
-                }
+                return tokio::runtime::Handle::current().block_on(async {
+                    if !ctx.meta_store.handle(ns.clone()).await.get().allow_attach {
+                        return Err(Error::NotAuthorized(format!(
+                            "Namespace `{ns}` doesn't allow attach"
+                        )));
+                    } else {
+                        Ok(())
+                    }
+                });
             }
             StmtKind::Detach => (),
         }
diff --git a/libsql-server/src/http/admin/mod.rs b/libsql-server/src/http/admin/mod.rs
@@ -331,7 +331,7 @@ async fn handle_create_namespace<C: Connector>(
             ));
         }
         // TODO: move this check into meta store
-        if !app_state.namespaces.exists(&ns) {
+        if !app_state.namespaces.exists(&ns).await {
             return Err(Error::NamespaceDoesntExist(ns.to_string()));
         }
 
diff --git a/libsql-server/src/namespace/meta_store.rs b/libsql-server/src/namespace/meta_store.rs
@@ -70,8 +70,8 @@ struct MetaStoreInner {
     // TODO(lucio): Use a concurrent hashmap so we don't block connection creation
     // when we are updating the config. The config si already synced via the watch
     // channel.
-    configs: Mutex<HashMap<NamespaceName, Sender<InnerConfig>>>,
-    conn: Mutex<MetaStoreConnection>,
+    configs: tokio::sync::Mutex<HashMap<NamespaceName, Sender<InnerConfig>>>,
+    conn: tokio::sync::Mutex<MetaStoreConnection>,
     wal_manager: MetaStoreWalManager,
 }
 
@@ -313,7 +313,7 @@ fn process(msg: ChangeMsg, inner: Arc<MetaStoreInner>) {
         } else {
             Ok(())
         };
-        let mut configs = inner.configs.lock();
+        let mut configs = inner.configs.blocking_lock();
         if let Some(config_watch) = configs.get_mut(&namespace) {
             let new_version = config_watch.borrow().version.wrapping_add(1);
 
@@ -330,7 +330,7 @@ fn process(msg: ChangeMsg, inner: Arc<MetaStoreInner>) {
         let _ = ret_chan.send(ret);
     } else {
         let ret = if flush {
-            let mut configs = inner.configs.lock();
+            let mut configs = inner.configs.blocking_lock();
             if let Some(config_watch) = configs.get_mut(&namespace) {
                 let config = config_watch.subscribe().borrow().clone();
                 try_process(&inner, &namespace, &config.config)
@@ -351,7 +351,7 @@ fn try_process(
 ) -> Result<()> {
     let config_encoded = metadata::DatabaseConfig::from(&*config).encode_to_vec();
 
-    let mut conn = inner.conn.lock();
+    let mut conn = inner.conn.blocking_lock();
     if let Some(schema) = config.shared_schema_name.as_ref() {
         let tx = conn.transaction()?;
         if let Some(ref schema) = config.shared_schema_name {
@@ -470,11 +470,11 @@ impl MetaStore {
         Ok(Self { changes_tx, inner })
     }
 
-    pub fn handle(&self, namespace: NamespaceName) -> MetaStoreHandle {
+    pub async fn handle(&self, namespace: NamespaceName) -> MetaStoreHandle {
         tracing::debug!("getting meta store handle");
         let change_tx = self.changes_tx.clone();
 
-        let mut configs = self.inner.configs.lock();
+        let mut configs = self.inner.configs.lock().await;
         let sender = configs.entry(namespace.clone()).or_insert_with(|| {
             // TODO(lucio): if no entry exists we need to ensure we send the update to
             // the bg channel.
@@ -495,11 +495,11 @@ impl MetaStore {
     pub fn remove(&self, namespace: NamespaceName) -> Result<Option<Arc<DatabaseConfig>>> {
         tracing::debug!("removing namespace `{}` from meta store", namespace);
 
-        let mut configs = self.inner.configs.lock();
+        let mut configs = self.inner.configs.blocking_lock();
         let r = if let Some(sender) = configs.get(&namespace) {
             tracing::debug!("removed namespace `{}` from meta store", namespace);
             let config = sender.borrow().clone();
-            let mut conn = self.inner.conn.lock();
+            let mut conn = self.inner.conn.blocking_lock();
             let tx = conn.transaction()?;
             if config.config.is_shared_schema {
                 if crate::schema::db::schema_has_linked_dbs(&tx, &namespace)? {
@@ -535,8 +535,8 @@ impl MetaStore {
     // TODO: we need to either make sure that the metastore is restored
     // before we start accepting connections or we need to contact bottomless
     // here to check if a namespace exists. Preferably the former.
-    pub fn exists(&self, namespace: &NamespaceName) -> bool {
-        self.inner.configs.lock().contains_key(namespace)
+    pub async fn exists(&self, namespace: &NamespaceName) -> bool {
+        self.inner.configs.lock().await.contains_key(namespace)
     }
 
     pub(crate) async fn shutdown(&self) -> crate::Result<()> {
@@ -559,7 +559,7 @@ impl MetaStore {
     ) -> crate::Result<MigrationSummary> {
         let inner = self.inner.clone();
         let summary = tokio::task::spawn_blocking(move || {
-            let mut conn = inner.conn.lock();
+            let mut conn = inner.conn.blocking_lock();
             crate::schema::get_migrations_summary(&mut conn, schema)
         })
         .await
@@ -574,7 +574,7 @@ impl MetaStore {
     ) -> crate::Result<Option<MigrationDetails>> {
         let inner = self.inner.clone();
         let details = tokio::task::spawn_blocking(move || {
-            let mut conn = inner.conn.lock();
+            let mut conn = inner.conn.blocking_lock();
             crate::schema::get_migration_details(&mut conn, schema, job_id)
         })
         .await
diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs
@@ -99,8 +99,8 @@ impl NamespaceStore {
         })
     }
 
-    pub fn exists(&self, namespace: &NamespaceName) -> bool {
-        self.inner.metadata.exists(namespace)
+    pub async fn exists(&self, namespace: &NamespaceName) -> bool {
+        self.inner.metadata.exists(namespace).await
     }
 
     pub async fn destroy(&self, namespace: NamespaceName, prune_all: bool) -> crate::Result<()> {
@@ -173,7 +173,7 @@ impl NamespaceStore {
             ns.destroy().await?;
         }
 
-        let db_config = self.inner.metadata.handle(namespace.clone());
+        let db_config = self.inner.metadata.handle(namespace.clone()).await;
         // destroy on-disk database
         self.cleanup(
             &namespace,
@@ -226,7 +226,7 @@ impl NamespaceStore {
         }
 
         // check that the source namespace exists
-        if !self.inner.metadata.exists(&from) {
+        if !self.inner.metadata.exists(&from).await {
             return Err(crate::error::Error::NamespaceDoesntExist(from.to_string()));
         }
 
@@ -241,11 +241,11 @@ impl NamespaceStore {
         }
 
         // FIXME: we could potentially delete the namespace while trying to fork it
-        if !self.inner.metadata.exists(&from) {
+        if !self.inner.metadata.exists(&from).await {
             return Err(crate::Error::NamespaceDoesntExist(from.to_string()));
         }
 
-        let from_config = self.inner.metadata.handle(from.clone());
+        let from_config = self.inner.metadata.handle(from.clone()).await;
         let from_entry = self
             .load_namespace(&from, from_config.clone(), RestoreOption::Latest)
             .await?;
@@ -280,7 +280,7 @@ impl NamespaceStore {
             should_delete: true,
         };
 
-        let handle = self.inner.metadata.handle(to.clone());
+        let handle = self.inner.metadata.handle(to.clone()).await;
         handle
             .store_and_maybe_flush(Some(to_config.into()), false)
             .await?;
@@ -328,7 +328,7 @@ impl NamespaceStore {
         Fun: FnOnce(&Namespace) -> R,
     {
         if namespace != NamespaceName::default()
-            && !self.inner.metadata.exists(&namespace)
+            && !self.inner.metadata.exists(&namespace).await
             && !self.inner.allow_lazy_creation
         {
             return Err(Error::NamespaceDoesntExist(namespace.to_string()));
@@ -346,7 +346,7 @@ impl NamespaceStore {
             }
         };
 
-        let handle = self.inner.metadata.handle(namespace.to_owned());
+        let handle = self.inner.metadata.handle(namespace.to_owned()).await;
         f(self
             .load_namespace(&namespace, handle, RestoreOption::Latest)
             .await?)
@@ -440,12 +440,12 @@ impl NamespaceStore {
         // FIXME: move the default namespace check out of this function.
         if self.inner.allow_lazy_creation || namespace == NamespaceName::default() {
             tracing::trace!("auto-creating the namespace");
-        } else if self.inner.metadata.exists(&namespace) {
+        } else if self.inner.metadata.exists(&namespace).await {
             return Err(Error::NamespaceAlreadyExist(namespace.to_string()));
         }
 
         let db_config = Arc::new(db_config);
-        let handle = self.inner.metadata.handle(namespace.clone());
+        let handle = self.inner.metadata.handle(namespace.clone()).await;
         tracing::debug!("storing db config");
         handle.store(db_config).await?;
         tracing::debug!("completed storing db config, loading namespace");
diff --git a/libsql-server/src/schema/db.rs b/libsql-server/src/schema/db.rs
@@ -482,6 +482,7 @@ mod test {
     async fn register_schema(meta_store: &MetaStore, schema: &'static str) {
         meta_store
             .handle(schema.into())
+            .await
             .store(DatabaseConfig {
                 is_shared_schema: true,
                 ..Default::default()
@@ -497,6 +498,7 @@ mod test {
     ) -> crate::Result<()> {
         meta_store
             .handle(name.into())
+            .await
             .store(DatabaseConfig {
                 shared_schema_name: Some(schema.into()),
                 ..Default::default()
@@ -561,6 +563,7 @@ mod test {
         // necessary checks beforehand, and return a nice error message.
         assert!(meta_store
             .handle("ns1".into())
+            .await
             .store(DatabaseConfig {
                 shared_schema_name: Some("schema1".into()),
                 ..Default::default()

Original file line number	Diff line number	Diff line change
`@@ -331,7 +331,7 @@ async fn handle_create_namespace<C: Connector>(`
`331`	`331`	`));`
`332`	`332`	`}`
`333`	`333`	`// TODO: move this check into meta store`
`334`		`- if !app_state.namespaces.exists(&ns) {`
	`334`	`+ if !app_state.namespaces.exists(&ns).await {`
`335`	`335`	`return Err(Error::NamespaceDoesntExist(ns.to_string()));`
`336`	`336`	`}`
`337`	`337`