[WIP] Rust client

Cargo.toml

@@ -3,12 +3,14 @@ version = "0.3.0"
 [workspace]
 
 members = [
+    "rain_client",
     "rain_core",
     "rain_server",
     "rain_task",
     "rain_task_test",
 ]
 
 [patch.crates-io]
+rain_client = { path = "rain_client" }
 rain_core = { path = "rain_core" }
 rain_task = { path = "rain_task" }

rain_client/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "rain_client"
+version = "0.3.0"
+
+description = "Distributed computational framework for large-scale task-based pipelines. Client library in Rust."
+# documentation = "https://docs.rs/rain_task/" # default docs.rs
+homepage = "https://github.com/substantic/rain"
+repository = "https://github.com/substantic/rain/"
+readme = "README.md"
+authors = [
+    "Stanislav Bohm <[email protected]>",
+    "Tomas Gavenciak <[email protected]>",
+    "Vojtech Cima <[email protected]>",
+    ]
+license = "MIT"
+
+exclude = ["testing/**/*"]
+
+[badges]
+travis-ci = { repository = "substantic/rain", branch = "master" }
+maintenance = { status = "actively-developed" }
+
+[dependencies]
+capnp-rpc = "0.8"
+error-chain = "0.11"
+futures = "0.1"
+log = "0.4"
+rain_core = "0.3.0"
+tokio-core = "0.1"
+serde = "1"
+serde_json = "1"

rain_client/src/client/client.rs

@@ -0,0 +1,73 @@
+use std::error::Error;
+use std::net::SocketAddr;
+
+use super::communicator::Communicator;
+use super::session::Session;
+use rain_core::CLIENT_PROTOCOL_VERSION;
+use std::rc::Rc;
+
+pub struct Client {
+    comm: Rc<Communicator>,
+}
+
+impl Client {
+    pub fn new(scheduler: SocketAddr) -> Result<Self, Box<Error>> {
+        let comm = Rc::new(Communicator::new(scheduler, CLIENT_PROTOCOL_VERSION)?);
+
+        Ok(Client { comm })
+    }
+
+    pub fn new_session(&self) -> Result<Session, Box<Error>> {
+        let session_id = self.comm.new_session()?;
+        Ok(Session::new(session_id, self.comm.clone()))
+    }
+
+    pub fn terminate_server(&self) -> Result<(), Box<Error>> {
+        self.comm.terminate_server()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::localcluster::LocalCluster;
+    use super::super::tasks::CommonTasks;
+    use super::Client;
+    use super::Session;
+    use std::env;
+
+    #[allow(dead_code)]
+    struct TestContext {
+        cluster: LocalCluster,
+        client: Client,
+        session: Session,
+    }
+
+    fn ctx() -> TestContext {
+        let rain = env::var("RAIN_BINARY").unwrap();
+
+        let cluster = LocalCluster::new(&rain).unwrap();
+        let client = cluster.create_client().unwrap();
+        let session = client.new_session().unwrap();
+
+        TestContext {
+            cluster,
+            client,
+            session,
+        }
+    }
+
+    #[test]
+    fn concat() {
+        let mut ctx = ctx();
+        let a = ctx.session.blob(vec![1, 2, 3]);
+        let b = ctx.session.blob(vec![4, 5, 6]);
+        let c = ctx.session.concat(&[a, b]);
+        c.output().keep();
+        ctx.session.submit().unwrap();
+        ctx.session.wait(&[c.clone()], &[]).unwrap();
+        assert_eq!(
+            ctx.session.fetch(&c.output()).unwrap(),
+            vec![1, 2, 3, 4, 5, 6]
+        );
+    }
+}

rain_client/src/client/communicator.rs

@@ -0,0 +1,154 @@
+use capnp_rpc::rpc_twoparty_capnp;
+use futures::Future;
+use rain_core::comm::new_rpc_system;
+use std::error::Error;
+use std::net::SocketAddr;
+use tokio_core::net::TcpStream;
+use tokio_core::reactor::Core;
+
+use super::task::Task;
+use client::dataobject::DataObject;
+use rain_core::types::{DataObjectId, TaskId};
+use rain_core::utils::{FromCapnp, ToCapnp};
+use rain_core::{client_capnp, common_capnp, server_capnp};
+use std::cell::{RefCell, RefMut};
+
+pub struct Communicator {
+    core: RefCell<Core>,
+    service: client_capnp::client_service::Client,
+}
+
+impl Communicator {
+    pub fn new(scheduler: SocketAddr, version: i32) -> Result<Self, Box<Error>> {
+        let mut core = Core::new()?;
+        let handle = core.handle();
+        let stream = core.run(TcpStream::connect(&scheduler, &handle))?;
+        stream.set_nodelay(true)?;
+
+        debug!("Connection to server {} established", scheduler);
+
+        let mut rpc = Box::new(new_rpc_system(stream, None));
+        let bootstrap: server_capnp::server_bootstrap::Client =
+            rpc.bootstrap(rpc_twoparty_capnp::Side::Server);
+        handle.spawn(rpc.map_err(|err| panic!("RPC error: {}", err)));
+
+        let mut request = bootstrap.register_as_client_request();
+        request.get().set_version(version);
+
+        let service = core.run(request.send().promise)?.get()?.get_service()?;
+
+        Ok(Self {
+            core: RefCell::new(core),
+            service,
+        })
+    }
+
+    pub fn new_session(&self) -> Result<i32, Box<Error>> {
+        let id: i32 = self.comm()
+            .run(self.service.new_session_request().send().promise)?
+            .get()?
+            .get_session_id();
+
+        Ok(id)
+    }
+    pub fn close_session(&self, id: i32) -> Result<(), Box<Error>> {
+        self.comm().run({
+            let mut req = self.service.close_session_request();
+            req.get().set_session_id(id);
+            req.send().promise
+        })?;
+
+        Ok(())
+    }
+
+    pub fn submit<T, D>(&self, tasks: &[T], data_objects: &[D]) -> Result<(), Box<Error>>
+    where
+        T: AsRef<Task>,
+        D: AsRef<DataObject>,
+    {
+        let mut req = self.service.submit_request();
+
+        to_capnp_list!(
+            req.get(),
+            tasks.iter().map(|t| t.as_ref()).collect::<Vec<&Task>>(),
+            init_tasks
+        );
+        to_capnp_list!(
+            req.get(),
+            data_objects
+                .iter()
+                .map(|t| t.as_ref())
+                .collect::<Vec<&DataObject>>(),
+            init_objects
+        );
+        self.comm().run(req.send().promise)?;
+
+        Ok(())
+    }
+
+    pub fn unkeep(&self, objects: &[DataObjectId]) -> Result<(), Box<Error>> {
+        let mut req = self.service.unkeep_request();
+        to_capnp_list!(req.get(), objects, init_object_ids);
+        self.comm().run(req.send().promise)?;
+        Ok(())
+    }
+
+    pub fn wait(&self, tasks: &[TaskId], objects: &[DataObjectId]) -> Result<(), Box<Error>> {
+        let mut req = self.service.wait_request();
+        to_capnp_list!(req.get(), tasks, init_task_ids);
+        to_capnp_list!(req.get(), objects, init_object_ids);
+        self.comm().run(req.send().promise)?;
+        Ok(())
+    }
+    pub fn wait_some(
+        &self,
+        tasks: &[TaskId],
+        objects: &[DataObjectId],
+    ) -> Result<(Vec<TaskId>, Vec<DataObjectId>), Box<Error>> {
+        let mut req = self.service.wait_some_request();
+        to_capnp_list!(req.get(), tasks, init_task_ids);
+        to_capnp_list!(req.get(), objects, init_object_ids);
+        let res = self.comm().run(req.send().promise)?;
+
+        Ok((
+            from_capnp_list!(res.get()?, get_finished_tasks, TaskId),
+            from_capnp_list!(res.get()?, get_finished_objects, DataObjectId),
+        ))
+    }
+
+    pub fn fetch(&self, object_id: &DataObjectId) -> Result<Vec<u8>, Box<Error>> {
+        let mut req = self.service.fetch_request();
+        object_id.to_capnp(&mut req.get().get_id().unwrap());
+        req.get().set_size(1024);
+
+        let response = self.comm().run(req.send().promise)?;
+
+        // TODO: handle error states
+        let reader = response.get()?;
+        match reader.get_status().which()? {
+            common_capnp::fetch_result::status::Ok(()) => {
+                let data = reader.get_data()?;
+                Ok(Vec::from(data))
+            }
+            common_capnp::fetch_result::status::Removed(()) => {
+                print!("Removed");
+                Ok(vec![])
+            }
+            common_capnp::fetch_result::status::Error(err) => {
+                print!("Error: {:?}", err.unwrap().get_message());
+                Ok(vec![])
+            }
+            _ => bail!("Non-ok status"),
+        }
+    }
+
+    pub fn terminate_server(&self) -> Result<(), Box<Error>> {
+        self.comm()
+            .run(self.service.terminate_server_request().send().promise)?;
+        Ok(())
+    }
+
+    fn comm(&self) -> RefMut<Core> {
+        self.core.borrow_mut()
+    }
+}

rain_client/src/client/dataobject.rs

@@ -0,0 +1,17 @@
+use rain_core::types::{DataObjectId, ObjectSpec};
+use std::cell::Cell;
+
+pub struct DataObject {
+    pub keep: Cell<bool>,
+    pub data: Option<Vec<u8>>,
+    pub spec: ObjectSpec,
+}
+
+impl DataObject {
+    pub fn keep(&self) {
+        self.keep.set(true);
+    }
+    pub fn id(&self) -> DataObjectId {
+        self.spec.id
+    }
+}

rain_client/src/client/localcluster.rs

@@ -0,0 +1,48 @@
+use super::client::Client;
+use std::error::Error;
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Command, Stdio};
+
+pub struct LocalCluster {
+    listen_addr: SocketAddr,
+    binary: PathBuf,
+}
+
+impl LocalCluster {
+    pub fn new(binary: &str) -> Result<Self, Box<Error>> {
+        let mut cluster = LocalCluster {
+            binary: PathBuf::from(binary),
+            listen_addr: SocketAddr::new("127.0.0.1".parse()?, 7210),
+        };
+        cluster.start()?;
+
+        Ok(cluster)
+    }
+
+    fn start(&mut self) -> Result<(), Box<Error>> {
+        Command::new(&self.binary)
+            .arg("start")
+            .arg("--listen")
+            .arg(self.listen_addr.to_string())
+            .arg("--simple")
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()?
+            .wait()?;
+
+        Ok(())
+    }
+
+    pub fn create_client(&self) -> Result<Client, Box<Error>> {
+        Client::new(self.listen_addr)
+    }
+}
+
+impl Drop for LocalCluster {
+    #![allow(unused_must_use)]
+    fn drop(&mut self) {
+        Client::new(self.listen_addr).unwrap().terminate_server();
+    }
+}

rain_client/src/client/mod.rs

@@ -0,0 +1,14 @@
+pub mod client;
+pub mod localcluster;
+pub mod session;
+pub mod tasks;
+
+#[macro_use]
+mod rpc;
+mod communicator;
+mod dataobject;
+mod task;
+
+pub use self::client::Client;
+pub use self::localcluster::LocalCluster;
+pub use self::session::Session;