clean Uuid resolver actor

This commit is contained in:
mpostma 2021-03-08 16:27:29 +01:00
parent 06403a5708
commit 4e1597bd1d
No known key found for this signature in database
GPG Key ID: CBC8A7C1D7A28C3A
2 changed files with 74 additions and 57 deletions

View File

@ -10,7 +10,6 @@ use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use actix_web::web::{Bytes, Payload}; use actix_web::web::{Bytes, Payload};
use anyhow::Context;
use futures::stream::StreamExt; use futures::stream::StreamExt;
use milli::update::{IndexDocumentsMethod, UpdateFormat}; use milli::update::{IndexDocumentsMethod, UpdateFormat};
use serde::{Serialize, Deserialize}; use serde::{Serialize, Deserialize};
@ -108,7 +107,7 @@ impl IndexController {
} }
pub async fn clear_documents(&self, index: String) -> anyhow::Result<UpdateStatus> { pub async fn clear_documents(&self, index: String) -> anyhow::Result<UpdateStatus> {
let uuid = self.uuid_resolver.resolve(index).await?.unwrap(); let uuid = self.uuid_resolver.resolve(index).await?;
let meta = UpdateMeta::ClearDocuments; let meta = UpdateMeta::ClearDocuments;
let (_, receiver) = mpsc::channel(1); let (_, receiver) = mpsc::channel(1);
let status = self.update_handle.update(meta, receiver, uuid).await?; let status = self.update_handle.update(meta, receiver, uuid).await?;
@ -116,7 +115,7 @@ impl IndexController {
} }
pub async fn delete_documents(&self, index: String, document_ids: Vec<String>) -> anyhow::Result<UpdateStatus> { pub async fn delete_documents(&self, index: String, document_ids: Vec<String>) -> anyhow::Result<UpdateStatus> {
let uuid = self.uuid_resolver.resolve(index).await.unwrap().unwrap(); let uuid = self.uuid_resolver.resolve(index).await?;
let meta = UpdateMeta::DeleteDocuments; let meta = UpdateMeta::DeleteDocuments;
let (sender, receiver) = mpsc::channel(10); let (sender, receiver) = mpsc::channel(10);
@ -153,8 +152,7 @@ impl IndexController {
pub async fn delete_index(&self, index_uid: String) -> anyhow::Result<()> { pub async fn delete_index(&self, index_uid: String) -> anyhow::Result<()> {
let uuid = self.uuid_resolver let uuid = self.uuid_resolver
.delete(index_uid) .delete(index_uid)
.await? .await?;
.context("index not found")?;
self.update_handle.delete(uuid.clone()).await?; self.update_handle.delete(uuid.clone()).await?;
self.index_handle.delete(uuid).await?; self.index_handle.delete(uuid).await?;
Ok(()) Ok(())
@ -163,16 +161,14 @@ impl IndexController {
pub async fn update_status(&self, index: String, id: u64) -> anyhow::Result<Option<UpdateStatus>> { pub async fn update_status(&self, index: String, id: u64) -> anyhow::Result<Option<UpdateStatus>> {
let uuid = self.uuid_resolver let uuid = self.uuid_resolver
.resolve(index) .resolve(index)
.await? .await?;
.context("index not found")?;
let result = self.update_handle.update_status(uuid, id).await?; let result = self.update_handle.update_status(uuid, id).await?;
Ok(result) Ok(result)
} }
pub async fn all_update_status(&self, index: String) -> anyhow::Result<Vec<UpdateStatus>> { pub async fn all_update_status(&self, index: String) -> anyhow::Result<Vec<UpdateStatus>> {
let uuid = self.uuid_resolver let uuid = self.uuid_resolver
.resolve(index).await? .resolve(index).await?;
.context("index not found")?;
let result = self.update_handle.get_all_updates_status(uuid).await?; let result = self.update_handle.get_all_updates_status(uuid).await?;
Ok(result) Ok(result)
} }
@ -195,8 +191,7 @@ impl IndexController {
pub async fn settings(&self, index: String) -> anyhow::Result<Settings> { pub async fn settings(&self, index: String) -> anyhow::Result<Settings> {
let uuid = self.uuid_resolver let uuid = self.uuid_resolver
.resolve(index.clone()) .resolve(index.clone())
.await? .await?;
.with_context(|| format!("Index {:?} doesn't exist", index))?;
let settings = self.index_handle.settings(uuid).await?; let settings = self.index_handle.settings(uuid).await?;
Ok(settings) Ok(settings)
} }
@ -210,8 +205,7 @@ impl IndexController {
) -> anyhow::Result<Vec<Document>> { ) -> anyhow::Result<Vec<Document>> {
let uuid = self.uuid_resolver let uuid = self.uuid_resolver
.resolve(index.clone()) .resolve(index.clone())
.await? .await?;
.with_context(|| format!("Index {:?} doesn't exist", index))?;
let documents = self.index_handle.documents(uuid, offset, limit, attributes_to_retrieve).await?; let documents = self.index_handle.documents(uuid, offset, limit, attributes_to_retrieve).await?;
Ok(documents) Ok(documents)
} }
@ -224,8 +218,7 @@ impl IndexController {
) -> anyhow::Result<Document> { ) -> anyhow::Result<Document> {
let uuid = self.uuid_resolver let uuid = self.uuid_resolver
.resolve(index.clone()) .resolve(index.clone())
.await? .await?;
.with_context(|| format!("Index {:?} doesn't exist", index))?;
let document = self.index_handle.document(uuid, doc_id, attributes_to_retrieve).await?; let document = self.index_handle.document(uuid, doc_id, attributes_to_retrieve).await?;
Ok(document) Ok(document)
} }
@ -235,21 +228,18 @@ impl IndexController {
} }
pub async fn search(&self, name: String, query: SearchQuery) -> anyhow::Result<SearchResult> { pub async fn search(&self, name: String, query: SearchQuery) -> anyhow::Result<SearchResult> {
let uuid = self.uuid_resolver.resolve(name).await.unwrap().unwrap(); let uuid = self.uuid_resolver.resolve(name).await?;
let result = self.index_handle.search(uuid, query).await?; let result = self.index_handle.search(uuid, query).await?;
Ok(result) Ok(result)
} }
pub async fn get_index(&self, name: String) -> anyhow::Result<Option<IndexMetadata>> { pub async fn get_index(&self, name: String) -> anyhow::Result<Option<IndexMetadata>> {
let uuid = self.uuid_resolver.resolve(name.clone()).await?; let uuid = self.uuid_resolver.resolve(name.clone()).await?;
if let Some(uuid) = uuid {
let result = self.index_handle let result = self.index_handle
.get_index_meta(uuid) .get_index_meta(uuid)
.await? .await?
.map(|meta| IndexMetadata { name, meta }); .map(|meta| IndexMetadata { name, meta });
return Ok(result) Ok(result)
}
Ok(None)
} }
} }

View File

@ -1,10 +1,10 @@
use thiserror::Error; use log::{info, warn};
use tokio::sync::{RwLock, mpsc, oneshot}; use std::collections::hash_map::Entry;
use uuid::Uuid;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use std::collections::hash_map::Entry; use thiserror::Error;
use log::{info, warn}; use tokio::sync::{mpsc, oneshot, RwLock};
use uuid::Uuid;
pub type Result<T> = std::result::Result<T, UuidError>; pub type Result<T> = std::result::Result<T, UuidError>;
@ -12,7 +12,7 @@ pub type Result<T> = std::result::Result<T, UuidError>;
enum UuidResolveMsg { enum UuidResolveMsg {
Resolve { Resolve {
name: String, name: String,
ret: oneshot::Sender<Result<Option<Uuid>>>, ret: oneshot::Sender<Result<Uuid>>,
}, },
GetOrCreate { GetOrCreate {
name: String, name: String,
@ -24,7 +24,7 @@ enum UuidResolveMsg {
}, },
Delete { Delete {
name: String, name: String,
ret: oneshot::Sender<Result<Option<Uuid>>>, ret: oneshot::Sender<Result<Uuid>>,
}, },
List { List {
ret: oneshot::Sender<Result<Vec<(String, Uuid)>>>, ret: oneshot::Sender<Result<Vec<(String, Uuid)>>>,
@ -46,13 +46,20 @@ impl<S: UuidStore> UuidResolverActor<S> {
info!("uuid resolver started"); info!("uuid resolver started");
// TODO: benchmark and use buffered streams to improve throughput.
loop { loop {
match self.inbox.recv().await { match self.inbox.recv().await {
Some(Create { name, ret }) => self.handle_create(name, ret).await, Some(Create { name, ret }) => {
Some(GetOrCreate { name, ret }) => self.handle_get_or_create(name, ret).await, let _ = ret.send(self.handle_create(name).await);
Some(Resolve { name, ret }) => self.handle_resolve(name, ret).await, }
Some(Delete { name, ret }) => self.handle_delete(name, ret).await, Some(GetOrCreate { name, ret }) => {
let _ = ret.send(self.handle_get_or_create(name).await);
}
Some(Resolve { name, ret }) => {
let _ = ret.send(self.handle_resolve(name).await);
}
Some(Delete { name, ret }) => {
let _ = ret.send(self.handle_delete(name).await);
}
Some(List { ret }) => { Some(List { ret }) => {
let _ = ret.send(self.handle_list().await); let _ = ret.send(self.handle_list().await);
} }
@ -64,24 +71,26 @@ impl<S: UuidStore> UuidResolverActor<S> {
warn!("exiting uuid resolver loop"); warn!("exiting uuid resolver loop");
} }
async fn handle_create(&self, name: String, ret: oneshot::Sender<Result<Uuid>>) { async fn handle_create(&self, name: String) -> Result<Uuid> {
let result = self.store.create_uuid(name, true).await; self.store.create_uuid(name, true).await
let _ = ret.send(result);
} }
async fn handle_get_or_create(&self, name: String, ret: oneshot::Sender<Result<Uuid>>) { async fn handle_get_or_create(&self, name: String) -> Result<Uuid> {
let result = self.store.create_uuid(name, false).await; self.store.create_uuid(name, false).await
let _ = ret.send(result);
} }
async fn handle_resolve(&self, name: String, ret: oneshot::Sender<Result<Option<Uuid>>>) { async fn handle_resolve(&self, name: String) -> Result<Uuid> {
let result = self.store.get_uuid(&name).await; self.store
let _ = ret.send(result); .get_uuid(&name)
.await?
.ok_or(UuidError::UnexistingIndex(name))
} }
async fn handle_delete(&self, name: String, ret: oneshot::Sender<Result<Option<Uuid>>>) { async fn handle_delete(&self, name: String) -> Result<Uuid> {
let result = self.store.delete(&name).await; self.store
let _ = ret.send(result); .delete(&name)
.await?
.ok_or(UuidError::UnexistingIndex(name))
} }
async fn handle_list(&self) -> Result<Vec<(String, Uuid)>> { async fn handle_list(&self) -> Result<Vec<(String, Uuid)>> {
@ -104,39 +113,49 @@ impl UuidResolverHandle {
Self { sender } Self { sender }
} }
pub async fn resolve(&self, name: String) -> anyhow::Result<Option<Uuid>> { pub async fn resolve(&self, name: String) -> anyhow::Result<Uuid> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = UuidResolveMsg::Resolve { name, ret }; let msg = UuidResolveMsg::Resolve { name, ret };
let _ = self.sender.send(msg).await; let _ = self.sender.send(msg).await;
Ok(receiver.await.expect("Uuid resolver actor has been killed")?) Ok(receiver
.await
.expect("Uuid resolver actor has been killed")?)
} }
pub async fn get_or_create(&self, name: String) -> Result<Uuid> { pub async fn get_or_create(&self, name: String) -> Result<Uuid> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = UuidResolveMsg::GetOrCreate { name, ret }; let msg = UuidResolveMsg::GetOrCreate { name, ret };
let _ = self.sender.send(msg).await; let _ = self.sender.send(msg).await;
Ok(receiver.await.expect("Uuid resolver actor has been killed")?) Ok(receiver
.await
.expect("Uuid resolver actor has been killed")?)
} }
pub async fn create(&self, name: String) -> anyhow::Result<Uuid> { pub async fn create(&self, name: String) -> anyhow::Result<Uuid> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = UuidResolveMsg::Create { name, ret }; let msg = UuidResolveMsg::Create { name, ret };
let _ = self.sender.send(msg).await; let _ = self.sender.send(msg).await;
Ok(receiver.await.expect("Uuid resolver actor has been killed")?) Ok(receiver
.await
.expect("Uuid resolver actor has been killed")?)
} }
pub async fn delete(&self, name: String) -> anyhow::Result<Option<Uuid>> { pub async fn delete(&self, name: String) -> anyhow::Result<Uuid> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = UuidResolveMsg::Delete { name, ret }; let msg = UuidResolveMsg::Delete { name, ret };
let _ = self.sender.send(msg).await; let _ = self.sender.send(msg).await;
Ok(receiver.await.expect("Uuid resolver actor has been killed")?) Ok(receiver
.await
.expect("Uuid resolver actor has been killed")?)
} }
pub async fn list(&self) -> anyhow::Result<Vec<(String, Uuid)>> { pub async fn list(&self) -> anyhow::Result<Vec<(String, Uuid)>> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = UuidResolveMsg::List { ret }; let msg = UuidResolveMsg::List { ret };
let _ = self.sender.send(msg).await; let _ = self.sender.send(msg).await;
Ok(receiver.await.expect("Uuid resolver actor has been killed")?) Ok(receiver
.await
.expect("Uuid resolver actor has been killed")?)
} }
} }
@ -144,6 +163,8 @@ impl UuidResolverHandle {
pub enum UuidError { pub enum UuidError {
#[error("Name already exist.")] #[error("Name already exist.")]
NameAlreadyExist, NameAlreadyExist,
#[error("Index \"{0}\" doesn't exist.")]
UnexistingIndex(String),
} }
#[async_trait::async_trait] #[async_trait::async_trait]
@ -168,7 +189,7 @@ impl UuidStore for MapUuidStore {
} else { } else {
Ok(entry.get().clone()) Ok(entry.get().clone())
} }
}, }
Entry::Vacant(entry) => { Entry::Vacant(entry) => {
let uuid = Uuid::new_v4(); let uuid = Uuid::new_v4();
let uuid = entry.insert(uuid); let uuid = entry.insert(uuid);
@ -186,7 +207,13 @@ impl UuidStore for MapUuidStore {
} }
async fn list(&self) -> Result<Vec<(String, Uuid)>> { async fn list(&self) -> Result<Vec<(String, Uuid)>> {
let list = self.0.read().await.iter().map(|(name, uuid)| (name.to_owned(), uuid.clone())).collect(); let list = self
.0
.read()
.await
.iter()
.map(|(name, uuid)| (name.to_owned(), uuid.clone()))
.collect();
Ok(list) Ok(list)
} }
} }