mirror of https://github.com/chipsenkbeil/distant
Support sequential batch processing (#201)
parent
efad345a0d
commit
4fb9045152
@ -0,0 +1,347 @@
|
|||||||
|
use std::io;
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
use async_trait::async_trait;
|
||||||
|
use distant_core::{
|
||||||
|
DistantApi, DistantApiServerHandler, DistantChannelExt, DistantClient, DistantCtx,
|
||||||
|
};
|
||||||
|
use distant_net::auth::{DummyAuthHandler, Verifier};
|
||||||
|
use distant_net::client::Client;
|
||||||
|
use distant_net::common::{InmemoryTransport, OneshotListener};
|
||||||
|
use distant_net::server::{Server, ServerRef};
|
||||||
|
|
||||||
|
/// Stands up an inmemory client and server using the given api.
|
||||||
|
async fn setup(
|
||||||
|
api: impl DistantApi<LocalData = ()> + Send + Sync + 'static,
|
||||||
|
) -> (DistantClient, Box<dyn ServerRef>) {
|
||||||
|
let (t1, t2) = InmemoryTransport::pair(100);
|
||||||
|
|
||||||
|
let server = Server::new()
|
||||||
|
.handler(DistantApiServerHandler::new(api))
|
||||||
|
.verifier(Verifier::none())
|
||||||
|
.start(OneshotListener::from_value(t2))
|
||||||
|
.expect("Failed to start server");
|
||||||
|
|
||||||
|
let client: DistantClient = Client::build()
|
||||||
|
.auth_handler(DummyAuthHandler)
|
||||||
|
.connector(t1)
|
||||||
|
.connect()
|
||||||
|
.await
|
||||||
|
.expect("Failed to connect to server");
|
||||||
|
|
||||||
|
(client, server)
|
||||||
|
}
|
||||||
|
|
||||||
|
mod single {
|
||||||
|
use super::*;
|
||||||
|
use test_log::test;
|
||||||
|
|
||||||
|
#[test(tokio::test)]
|
||||||
|
async fn should_support_single_request_returning_error() {
|
||||||
|
struct TestDistantApi;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl DistantApi for TestDistantApi {
|
||||||
|
type LocalData = ();
|
||||||
|
|
||||||
|
async fn read_file(
|
||||||
|
&self,
|
||||||
|
_ctx: DistantCtx<Self::LocalData>,
|
||||||
|
_path: PathBuf,
|
||||||
|
) -> io::Result<Vec<u8>> {
|
||||||
|
Err(io::Error::new(io::ErrorKind::NotFound, "test error"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let (mut client, _server) = setup(TestDistantApi).await;
|
||||||
|
|
||||||
|
let error = client.read_file(PathBuf::from("file")).await.unwrap_err();
|
||||||
|
assert_eq!(error.kind(), io::ErrorKind::NotFound);
|
||||||
|
assert_eq!(error.to_string(), "test error");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test(tokio::test)]
|
||||||
|
async fn should_support_single_request_returning_success() {
|
||||||
|
struct TestDistantApi;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl DistantApi for TestDistantApi {
|
||||||
|
type LocalData = ();
|
||||||
|
|
||||||
|
async fn read_file(
|
||||||
|
&self,
|
||||||
|
_ctx: DistantCtx<Self::LocalData>,
|
||||||
|
_path: PathBuf,
|
||||||
|
) -> io::Result<Vec<u8>> {
|
||||||
|
Ok(b"hello world".to_vec())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let (mut client, _server) = setup(TestDistantApi).await;
|
||||||
|
|
||||||
|
let contents = client.read_file(PathBuf::from("file")).await.unwrap();
|
||||||
|
assert_eq!(contents, b"hello world");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
mod batch_parallel {
|
||||||
|
use super::*;
|
||||||
|
use distant_net::common::Request;
|
||||||
|
use distant_protocol::{Msg, Request as RequestPayload};
|
||||||
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||||
|
use test_log::test;
|
||||||
|
|
||||||
|
#[test(tokio::test)]
|
||||||
|
async fn should_support_multiple_requests_running_in_parallel() {
|
||||||
|
struct TestDistantApi;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl DistantApi for TestDistantApi {
|
||||||
|
type LocalData = ();
|
||||||
|
|
||||||
|
async fn read_file(
|
||||||
|
&self,
|
||||||
|
_ctx: DistantCtx<Self::LocalData>,
|
||||||
|
path: PathBuf,
|
||||||
|
) -> io::Result<Vec<u8>> {
|
||||||
|
if path.to_str().unwrap() == "slow" {
|
||||||
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
let time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
|
||||||
|
Ok((time.as_millis() as u64).to_be_bytes().to_vec())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let (mut client, _server) = setup(TestDistantApi).await;
|
||||||
|
|
||||||
|
let request = Request::new(Msg::batch([
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file1"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("slow"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file2"),
|
||||||
|
},
|
||||||
|
]));
|
||||||
|
|
||||||
|
let response = client.send(request).await.unwrap();
|
||||||
|
let payloads = response.payload.into_batch().unwrap();
|
||||||
|
|
||||||
|
// Collect our times from the reading
|
||||||
|
let mut times = Vec::new();
|
||||||
|
for payload in payloads {
|
||||||
|
match payload {
|
||||||
|
distant_protocol::Response::Blob { data } => {
|
||||||
|
let mut buf = [0u8; 8];
|
||||||
|
buf.copy_from_slice(&data[..8]);
|
||||||
|
times.push(u64::from_be_bytes(buf));
|
||||||
|
}
|
||||||
|
x => panic!("Unexpected payload: {x:?}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify that these ran in parallel as the first and third requests should not be
|
||||||
|
// over 500 milliseconds apart due to the sleep in the middle!
|
||||||
|
let diff = times[0].abs_diff(times[2]);
|
||||||
|
assert!(diff <= 500, "Sequential ordering detected");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test(tokio::test)]
|
||||||
|
async fn should_run_all_requests_even_if_some_fail() {
|
||||||
|
struct TestDistantApi;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl DistantApi for TestDistantApi {
|
||||||
|
type LocalData = ();
|
||||||
|
|
||||||
|
async fn read_file(
|
||||||
|
&self,
|
||||||
|
_ctx: DistantCtx<Self::LocalData>,
|
||||||
|
path: PathBuf,
|
||||||
|
) -> io::Result<Vec<u8>> {
|
||||||
|
if path.to_str().unwrap() == "fail" {
|
||||||
|
return Err(io::Error::new(io::ErrorKind::Other, "test error"));
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(Vec::new())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let (mut client, _server) = setup(TestDistantApi).await;
|
||||||
|
|
||||||
|
let request = Request::new(Msg::batch([
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file1"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("fail"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file2"),
|
||||||
|
},
|
||||||
|
]));
|
||||||
|
|
||||||
|
let response = client.send(request).await.unwrap();
|
||||||
|
let payloads = response.payload.into_batch().unwrap();
|
||||||
|
|
||||||
|
// Should be a success, error, and success
|
||||||
|
assert!(
|
||||||
|
matches!(payloads[0], distant_protocol::Response::Blob { .. }),
|
||||||
|
"Unexpected payloads[0]: {:?}",
|
||||||
|
payloads[0]
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
matches!(
|
||||||
|
&payloads[1],
|
||||||
|
distant_protocol::Response::Error(distant_protocol::Error { kind, description })
|
||||||
|
if matches!(kind, distant_protocol::ErrorKind::Other) && description == "test error"
|
||||||
|
),
|
||||||
|
"Unexpected payloads[1]: {:?}",
|
||||||
|
payloads[1]
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
matches!(payloads[2], distant_protocol::Response::Blob { .. }),
|
||||||
|
"Unexpected payloads[2]: {:?}",
|
||||||
|
payloads[2]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
mod batch_sequence {
|
||||||
|
use super::*;
|
||||||
|
use distant_net::common::Request;
|
||||||
|
use distant_protocol::{Msg, Request as RequestPayload};
|
||||||
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||||
|
use test_log::test;
|
||||||
|
|
||||||
|
#[test(tokio::test)]
|
||||||
|
async fn should_support_multiple_requests_running_in_sequence() {
|
||||||
|
struct TestDistantApi;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl DistantApi for TestDistantApi {
|
||||||
|
type LocalData = ();
|
||||||
|
|
||||||
|
async fn read_file(
|
||||||
|
&self,
|
||||||
|
_ctx: DistantCtx<Self::LocalData>,
|
||||||
|
path: PathBuf,
|
||||||
|
) -> io::Result<Vec<u8>> {
|
||||||
|
if path.to_str().unwrap() == "slow" {
|
||||||
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
let time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
|
||||||
|
Ok((time.as_millis() as u64).to_be_bytes().to_vec())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let (mut client, _server) = setup(TestDistantApi).await;
|
||||||
|
|
||||||
|
let mut request = Request::new(Msg::batch([
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file1"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("slow"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file2"),
|
||||||
|
},
|
||||||
|
]));
|
||||||
|
|
||||||
|
// Mark as running in sequence
|
||||||
|
request.header.insert("sequence", true);
|
||||||
|
|
||||||
|
let response = client.send(request).await.unwrap();
|
||||||
|
let payloads = response.payload.into_batch().unwrap();
|
||||||
|
|
||||||
|
// Collect our times from the reading
|
||||||
|
let mut times = Vec::new();
|
||||||
|
for payload in payloads {
|
||||||
|
match payload {
|
||||||
|
distant_protocol::Response::Blob { data } => {
|
||||||
|
let mut buf = [0u8; 8];
|
||||||
|
buf.copy_from_slice(&data[..8]);
|
||||||
|
times.push(u64::from_be_bytes(buf));
|
||||||
|
}
|
||||||
|
x => panic!("Unexpected payload: {x:?}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify that these ran in sequence as the first and third requests should be
|
||||||
|
// over 500 milliseconds apart due to the sleep in the middle!
|
||||||
|
let diff = times[0].abs_diff(times[2]);
|
||||||
|
assert!(diff > 500, "Parallel ordering detected");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test(tokio::test)]
|
||||||
|
async fn should_interrupt_any_requests_following_a_failure() {
|
||||||
|
struct TestDistantApi;
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl DistantApi for TestDistantApi {
|
||||||
|
type LocalData = ();
|
||||||
|
|
||||||
|
async fn read_file(
|
||||||
|
&self,
|
||||||
|
_ctx: DistantCtx<Self::LocalData>,
|
||||||
|
path: PathBuf,
|
||||||
|
) -> io::Result<Vec<u8>> {
|
||||||
|
if path.to_str().unwrap() == "fail" {
|
||||||
|
return Err(io::Error::new(io::ErrorKind::Other, "test error"));
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(Vec::new())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let (mut client, _server) = setup(TestDistantApi).await;
|
||||||
|
|
||||||
|
let mut request = Request::new(Msg::batch([
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file1"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("fail"),
|
||||||
|
},
|
||||||
|
RequestPayload::FileRead {
|
||||||
|
path: PathBuf::from("file2"),
|
||||||
|
},
|
||||||
|
]));
|
||||||
|
|
||||||
|
// Mark as running in sequence
|
||||||
|
request.header.insert("sequence", true);
|
||||||
|
|
||||||
|
let response = client.send(request).await.unwrap();
|
||||||
|
let payloads = response.payload.into_batch().unwrap();
|
||||||
|
|
||||||
|
// Should be a success, error, and interrupt
|
||||||
|
assert!(
|
||||||
|
matches!(payloads[0], distant_protocol::Response::Blob { .. }),
|
||||||
|
"Unexpected payloads[0]: {:?}",
|
||||||
|
payloads[0]
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
matches!(
|
||||||
|
&payloads[1],
|
||||||
|
distant_protocol::Response::Error(distant_protocol::Error { kind, description })
|
||||||
|
if matches!(kind, distant_protocol::ErrorKind::Other) && description == "test error"
|
||||||
|
),
|
||||||
|
"Unexpected payloads[1]: {:?}",
|
||||||
|
payloads[1]
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
matches!(
|
||||||
|
&payloads[2],
|
||||||
|
distant_protocol::Response::Error(distant_protocol::Error { kind, .. })
|
||||||
|
if matches!(kind, distant_protocol::ErrorKind::Interrupted)
|
||||||
|
),
|
||||||
|
"Unexpected payloads[2]: {:?}",
|
||||||
|
payloads[2]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,110 @@
|
|||||||
|
use crate::common::utils;
|
||||||
|
use serde::de::DeserializeOwned;
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
use std::borrow::Cow;
|
||||||
|
use std::io;
|
||||||
|
use std::ops::{Deref, DerefMut};
|
||||||
|
|
||||||
|
/// Generic value type for data passed through header.
|
||||||
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
#[serde(transparent)]
|
||||||
|
pub struct Value(serde_json::Value);
|
||||||
|
|
||||||
|
impl Value {
|
||||||
|
/// Creates a new [`Value`] by converting `value` to the underlying type.
|
||||||
|
pub fn new(value: impl Into<serde_json::Value>) -> Self {
|
||||||
|
Self(value.into())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Serializes the value into bytes.
|
||||||
|
pub fn to_vec(&self) -> io::Result<Vec<u8>> {
|
||||||
|
utils::serialize_to_vec(self)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Deserializes the value from bytes.
|
||||||
|
pub fn from_slice(slice: &[u8]) -> io::Result<Self> {
|
||||||
|
utils::deserialize_from_slice(slice)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Attempts to convert this generic value to a specific type.
|
||||||
|
pub fn cast_as<T>(self) -> io::Result<T>
|
||||||
|
where
|
||||||
|
T: DeserializeOwned,
|
||||||
|
{
|
||||||
|
serde_json::from_value(self.0).map_err(|x| io::Error::new(io::ErrorKind::InvalidData, x))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Deref for Value {
|
||||||
|
type Target = serde_json::Value;
|
||||||
|
|
||||||
|
fn deref(&self) -> &Self::Target {
|
||||||
|
&self.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl DerefMut for Value {
|
||||||
|
fn deref_mut(&mut self) -> &mut Self::Target {
|
||||||
|
&mut self.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
macro_rules! impl_from {
|
||||||
|
($($type:ty),+) => {
|
||||||
|
$(
|
||||||
|
impl From<$type> for Value {
|
||||||
|
fn from(x: $type) -> Self {
|
||||||
|
Self(From::from(x))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
)+
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
impl_from!(
|
||||||
|
(),
|
||||||
|
i8, i16, i32, i64, isize,
|
||||||
|
u8, u16, u32, u64, usize,
|
||||||
|
f32, f64,
|
||||||
|
bool, String, serde_json::Number,
|
||||||
|
serde_json::Map<String, serde_json::Value>
|
||||||
|
);
|
||||||
|
|
||||||
|
impl<'a, T> From<&'a [T]> for Value
|
||||||
|
where
|
||||||
|
T: Clone + Into<serde_json::Value>,
|
||||||
|
{
|
||||||
|
fn from(x: &'a [T]) -> Self {
|
||||||
|
Self(From::from(x))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'a> From<&'a str> for Value {
|
||||||
|
fn from(x: &'a str) -> Self {
|
||||||
|
Self(From::from(x))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'a> From<Cow<'a, str>> for Value {
|
||||||
|
fn from(x: Cow<'a, str>) -> Self {
|
||||||
|
Self(From::from(x))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> From<Option<T>> for Value
|
||||||
|
where
|
||||||
|
T: Into<serde_json::Value>,
|
||||||
|
{
|
||||||
|
fn from(x: Option<T>) -> Self {
|
||||||
|
Self(From::from(x))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> From<Vec<T>> for Value
|
||||||
|
where
|
||||||
|
T: Into<serde_json::Value>,
|
||||||
|
{
|
||||||
|
fn from(x: Vec<T>) -> Self {
|
||||||
|
Self(From::from(x))
|
||||||
|
}
|
||||||
|
}
|
Loading…
Reference in New Issue