Initial, untested version of controller and worker

Worker has been tested to a marginal extent, but the controller is
litereally untested.

7 files changed, 147 insertions, 0 deletions

diff --git a/utils/Cargo.toml b/utils/Cargo.toml
new file mode 100644
index 0000000..53767d0
--- /dev/null
+++ b/utils/Cargo.toml
@@ -0,0 +1,7 @@
+[package]
+name = "dd-utils"
+version = "0.1.0"
+authors = ["Tom Smeding <tom.smeding@gmail.com>"]
+edition = "2018"
+
+[dependencies]
diff --git a/utils/src/enums.rs b/utils/src/enums.rs
new file mode 100644
index 0000000..d08c7ea
--- /dev/null
+++ b/utils/src/enums.rs
@@ -0,0 +1,11 @@
+pub enum MessageBody {
+    Version(u32),
+    NewCore(String, Vec<u8>),
+    Job(u64, Vec<u8>),
+}
+
+pub enum Reply {
+    Version(bool),
+    NewCore,
+    Job(i32, Vec<u8>),
+}
diff --git a/utils/src/error.rs b/utils/src/error.rs
new file mode 100644
index 0000000..f25bc5d
--- /dev/null
+++ b/utils/src/error.rs
@@ -0,0 +1,27 @@
+use std::io;
+
+pub trait IntoIOError {
+    fn ioerr(self) -> io::Error;
+    fn perror(self, parent: io::Error) -> io::Error;
+}
+
+// This impl bound is taken directly from the io::Error::new function.
+impl<E: Into<Box<dyn std::error::Error + Send + Sync>>> IntoIOError for E {
+    fn ioerr(self) -> io::Error {
+        io::Error::new(io::ErrorKind::Other, self)
+    }
+
+    fn perror(self, parent: io::Error) -> io::Error {
+        io::Error::new(parent.kind(), format!("{}: {}", self.into(), parent))
+    }
+}
+
+pub trait IntoIOResult<T> {
+    fn iores(self) -> io::Result<T>;
+}
+
+impl<T, E: IntoIOError> IntoIOResult<T> for Result<T, E> {
+    fn iores(self) -> io::Result<T> {
+        self.map_err(|e| e.ioerr())
+    }
+}
diff --git a/utils/src/idgen.rs b/utils/src/idgen.rs
new file mode 100644
index 0000000..20f13ff
--- /dev/null
+++ b/utils/src/idgen.rs
@@ -0,0 +1,16 @@
+#[derive(Debug)]
+pub struct IdGen {
+    next: u64,
+}
+
+impl IdGen {
+    pub fn new(start: u64) -> Self {
+        IdGen { next: start }
+    }
+
+    pub fn gen(&mut self) -> u64 {
+        let res = self.next;
+        self.next += 1;
+        res
+    }
+}
diff --git a/utils/src/lib.rs b/utils/src/lib.rs
new file mode 100644
index 0000000..867fd97
--- /dev/null
+++ b/utils/src/lib.rs
@@ -0,0 +1,4 @@
+pub mod error;
+pub mod idgen;
+pub mod protocol;
+pub mod read_ext;
diff --git a/utils/src/protocol.rs b/utils/src/protocol.rs
new file mode 100644
index 0000000..4820f79
--- /dev/null
+++ b/utils/src/protocol.rs
@@ -0,0 +1,44 @@
+use std::convert::TryInto;
+use std::io::{self, BufReader, ErrorKind, Read};
+use std::net::TcpStream;
+
+pub enum MessageBody {
+    Version(u32),
+    NewCore(String, Vec<u8>),
+    Job(u64, Vec<u8>),
+}
+
+pub enum Reply {
+    Version(bool),
+    NewCore,
+    Job(i32, Vec<u8>),
+}
+
+pub struct RawMessage {
+    pub typ: u8,
+    pub id: u64,
+    pub payload: Vec<u8>,
+}
+
+impl RawMessage {
+    pub fn receive(reader: &mut BufReader<TcpStream>) -> io::Result<Option<Self>> {
+        let mut header = [0u8; 17];
+        if let Err(e) = reader.read(&mut header) {
+            if e.kind() == ErrorKind::UnexpectedEof { return Ok(None); }
+            else { return Err(e); }
+        }
+
+        let typ = header[0];
+        let id = u64::from_le_bytes(header[1..9].try_into().unwrap());
+        let length = usize::from_le_bytes(header[9..17].try_into().unwrap());
+
+        let mut payload = Vec::new();
+        payload.resize(length, 0u8);
+        if let Err(e) = reader.read(&mut payload) {
+            if e.kind() == ErrorKind::UnexpectedEof { return Ok(None); }
+            else { return Err(e); }
+        }
+
+        Ok(Some(Self { typ, id, payload }))
+    }
+}
diff --git a/utils/src/read_ext.rs b/utils/src/read_ext.rs
new file mode 100644
index 0000000..7695ada
--- /dev/null
+++ b/utils/src/read_ext.rs
@@ -0,0 +1,38 @@
+use std::convert::TryInto;
+
+pub trait ReadExt {
+    fn read_exact(&mut self, n: usize) -> Option<&[u8]>;
+
+    fn read_le_i32(&mut self) -> Option<i32> {
+        self.read_exact(4).map(|bytes| i32::from_le_bytes(bytes.try_into().unwrap()))
+    }
+
+    fn read_le_u32(&mut self) -> Option<u32> {
+        self.read_exact(4).map(|bytes| u32::from_le_bytes(bytes.try_into().unwrap()))
+    }
+
+    fn read_le_u64(&mut self) -> Option<u64> {
+        self.read_exact(8).map(|bytes| u64::from_le_bytes(bytes.try_into().unwrap()))
+    }
+
+    fn read_pascal_blob(&mut self) -> Option<&[u8]> {
+        let len = self.read_le_u64()?;
+        self.read_exact(len as usize)
+    }
+
+    fn read_pascal_string(&mut self) -> Option<Result<String, std::string::FromUtf8Error>> {
+        self.read_pascal_blob().map(|b| String::from_utf8(b.to_vec()))
+    }
+}
+
+impl ReadExt for &[u8] {
+    fn read_exact(&mut self, n: usize) -> Option<&[u8]> {
+        if self.len() >= n {
+            let (res, newlist) = self.split_at(n);
+            *self = newlist;
+            Some(res)
+        } else {
+            None
+        }
+    }
+}