firmware: use dedicated error types for every protocol and thread.

Good riddance to io::Error::Unrecognized.

artiq/firmware/Cargo.lock

@@ -208,6 +208,7 @@ dependencies = [
  "failure_derive 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "io 0.0.0",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "std_artiq 0.0.0",
 ]
 
 [[package]]
@@ -226,6 +227,8 @@ dependencies = [
  "build_misoc 0.0.0",
  "byteorder 1.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "cslice 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "failure 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "failure_derive 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "fringe 1.1.0 (git+https://github.com/m-labs/libfringe?rev=bd23494)",
  "io 0.0.0",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -270,6 +273,9 @@ dependencies = [
 [[package]]
 name = "std_artiq"
 version = "0.0.0"
+dependencies = [
+ "failure 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+]
 
 [[package]]
 name = "syn"

artiq/firmware/libio/lib.rs

@@ -18,13 +18,13 @@ mod proto;
 pub use cursor::Cursor;
 #[cfg(feature = "byteorder")]
 pub use proto::{ProtoRead, ProtoWrite};
+#[cfg(all(feature = "byteorder", feature = "alloc"))]
+pub use proto::ReadStringError;
 
 #[derive(Fail, Debug, Clone, PartialEq)]
 pub enum Error<T> {
     #[fail(display = "unexpected end of stream")]
     UnexpectedEnd,
-    #[fail(display = "unrecognized input")]
-    Unrecognized,
     #[fail(display = "{}", _0)]
     Other(#[cause] T)
 }

artiq/firmware/libio/proto.rs

@@ -8,22 +8,11 @@ use ::{Read, Write, Error as IoError};
 #[derive(Fail, Debug, Clone, PartialEq)]
 pub enum ReadStringError<T> {
     #[fail(display = "invalid UTF-8: {}", _0)]
-    Utf8Error(Utf8Error),
+    Utf8(Utf8Error),
     #[fail(display = "{}", _0)]
     Other(T)
 }
 
-#[cfg(feature = "alloc")]
-impl<T> From<ReadStringError<IoError<T>>> for IoError<T>
-{
-    fn from(value: ReadStringError<IoError<T>>) -> IoError<T> {
-        match value {
-            ReadStringError::Utf8Error(_) => IoError::Unrecognized,
-            ReadStringError::Other(err) => err
-        }
-    }
-}
-
 pub trait ProtoRead {
     type ReadError;
 
@@ -75,7 +64,7 @@ pub trait ProtoRead {
     #[inline]
     fn read_string(&mut self) -> Result<::alloc::String, ReadStringError<Self::ReadError>> {
         let bytes = self.read_bytes().map_err(ReadStringError::Other)?;
-        String::from_utf8(bytes).map_err(|err| ReadStringError::Utf8Error(err.utf8_error()))
+        String::from_utf8(bytes).map_err(|err| ReadStringError::Utf8(err.utf8_error()))
     }
 }
 

artiq/firmware/libproto_artiq/Cargo.toml

@@ -15,6 +15,8 @@ cslice = { version = "0.3" }
 log = { version = "0.4", default-features = false, optional = true }
 io = { path = "../libio", features = ["byteorder"] }
 dyld = { path = "../libdyld" }
+# TODO: remove
+std_artiq = { path = "../libstd_artiq", optional = true }
 
 [features]
-alloc = ["io/alloc"]
+alloc = ["io/alloc", "std_artiq"]

artiq/firmware/libproto_artiq/analyzer_proto.rs

@@ -1,4 +1,4 @@
-use io::{Write, ProtoWrite, Error};
+use io::{Write, ProtoWrite, Error as IoError};
 
 #[derive(Debug)]
 pub struct Header {
@@ -10,7 +10,9 @@ pub struct Header {
 }
 
 impl Header {
-    pub fn write_to<T: Write>(&self, writer: &mut T) -> Result<(), Error<T::WriteError>> {
+    pub fn write_to<W>(&self, writer: &mut W) -> Result<(), IoError<W::WriteError>>
+        where W: Write + ?Sized
+    {
         writer.write_u32(self.sent_bytes)?;
         writer.write_u64(self.total_byte_count)?;
         writer.write_u8(self.overflow_occurred as u8)?;

artiq/firmware/libproto_artiq/drtioaux_proto.rs

@@ -1,4 +1,18 @@
-use io::{Read, ProtoRead, Write, ProtoWrite, Error};
+use io::{Read, ProtoRead, Write, ProtoWrite, Error as IoError};
+
+#[derive(Fail, Debug)]
+pub enum Error<T> {
+    #[fail(display = "unknown packet {:#02x}", _0)]
+    UnknownPacket(u8),
+    #[fail(display = "{}", _0)]
+    Io(#[cause] IoError<T>)
+}
+
+impl<T> From<IoError<T>> for Error<T> {
+    fn from(value: IoError<T>) -> Error<T> {
+        Error::Io(value)
+    }
+}
 
 #[derive(Debug)]
 pub enum Packet {
@@ -36,7 +50,9 @@ pub enum Packet {
 }
 
 impl Packet {
-    pub fn read_from<T: Read>(reader: &mut T) -> Result<Self, Error<T::ReadError>> {
+    pub fn read_from<R>(reader: &mut R) -> Result<Self, Error<R::ReadError>>
+        where R: Read + ?Sized
+    {
         Ok(match reader.read_u8()? {
             0x00 => Packet::EchoRequest,
             0x01 => Packet::EchoReply,
@@ -129,11 +145,13 @@ impl Packet {
                 succeeded: reader.read_bool()?
             },
 
-            _ => return Err(Error::Unrecognized)
+            ty => return Err(Error::UnknownPacket(ty))
         })
     }
 
-    pub fn write_to<T: Write>(&self, writer: &mut T) -> Result<(), Error<T::WriteError>> {
+    pub fn write_to<W>(&self, writer: &mut W) -> Result<(), IoError<W::WriteError>>
+        where W: Write + ?Sized
+    {
         match *self {
             Packet::EchoRequest =>
                 writer.write_u8(0x00)?,

artiq/firmware/libproto_artiq/lib.rs

@@ -1,6 +1,9 @@
 #![no_std]
 #![cfg_attr(feature = "alloc", feature(alloc))]
 
+extern crate failure;
+#[macro_use]
+extern crate failure_derive;
 #[cfg(feature = "alloc")]
 extern crate alloc;
 extern crate cslice;
@@ -8,6 +11,8 @@ extern crate cslice;
 #[macro_use]
 extern crate log;
 
+#[cfg(feature = "std_artiq")]
+extern crate std_artiq;
 extern crate io;
 extern crate dyld;
 

artiq/firmware/libproto_artiq/mgmt_proto.rs

@@ -2,7 +2,46 @@ use alloc::Vec;
 #[cfg(feature = "log")]
 use log;
 
-use io::{Read, ProtoRead, Write, ProtoWrite, Error};
+use io::{Read, ProtoRead, Write, ProtoWrite, Error as IoError};
+
+#[derive(Fail, Debug)]
+pub enum Error<T> {
+    #[fail(display = "incorrect magic")]
+    WrongMagic,
+    #[fail(display = "unknown packet {:#02x}", _0)]
+    UnknownPacket(u8),
+    #[fail(display = "unknown log level {}", _0)]
+    UnknownLogLevel(u8),
+    #[fail(display = "{}", _0)]
+    Io(#[cause] IoError<T>)
+}
+
+impl<T> From<IoError<T>> for Error<T> {
+    fn from(value: IoError<T>) -> Error<T> {
+        Error::Io(value)
+    }
+}
+
+#[cfg(feature = "std_artiq")]
+impl From<::std_artiq::io::Error> for Error<::std_artiq::io::Error> {
+    fn from(value: ::std_artiq::io::Error) -> Error<::std_artiq::io::Error> {
+        Error::Io(IoError::Other(value))
+    }
+}
+
+pub fn read_magic<R>(reader: &mut R) -> Result<(), Error<R::ReadError>>
+    where R: Read + ?Sized
+{
+    const MAGIC: &'static [u8] = b"ARTIQ management\n";
+
+    let mut magic: [u8; 17] = [0; 17];
+    reader.read_exact(&mut magic)?;
+    if magic != MAGIC {
+        Err(Error::WrongMagic)
+    } else {
+        Ok(())
+    }
+}
 
 #[derive(Debug)]
 pub enum Request {
@@ -40,7 +79,9 @@ pub enum Reply<'a> {
 }
 
 impl Request {
-    pub fn read_from<T: Read + ?Sized>(reader: &mut T) -> Result<Self, Error<T::ReadError>> {
+    pub fn read_from<R>(reader: &mut R) -> Result<Self, Error<R::ReadError>>
+        where R: Read + ?Sized
+    {
         #[cfg(feature = "log")]
         fn read_log_level_filter<T: Read + ?Sized>(reader: &mut T) ->
                 Result<log::LevelFilter, Error<T::ReadError>> {
@@ -51,7 +92,7 @@ impl Request {
                 3 => log::LevelFilter::Info,
                 4 => log::LevelFilter::Debug,
                 5 => log::LevelFilter::Trace,
-                _ => return Err(Error::Unrecognized)
+                lv => return Err(Error::UnknownLogLevel(lv))
             })
         }
 
@@ -73,13 +114,15 @@ impl Request {
             4 => Request::Hotswap(reader.read_bytes()?),
             5 => Request::Reboot,
             8 => Request::DebugAllocator,
-            _  => return Err(Error::Unrecognized)
+            ty => return Err(Error::UnknownPacket(ty))
         })
     }
 }
 
 impl<'a> Reply<'a> {
-    pub fn write_to<T: Write + ?Sized>(&self, writer: &mut T) -> Result<(), Error<T::WriteError>> {
+    pub fn write_to<W>(&self, writer: &mut W) -> Result<(), IoError<W::WriteError>>
+        where W: Write + ?Sized
+    {
         match *self {
             Reply::Success => {
                 writer.write_u8(1)?;

artiq/firmware/libproto_artiq/moninj_proto.rs

@@ -1,4 +1,41 @@
-use io::{Read, ProtoRead, Write, ProtoWrite, Error};
+use io::{Read, ProtoRead, Write, ProtoWrite, Error as IoError};
+
+#[derive(Fail, Debug)]
+pub enum Error<T> {
+    #[fail(display = "incorrect magic")]
+    WrongMagic,
+    #[fail(display = "unknown packet {:#02x}", _0)]
+    UnknownPacket(u8),
+    #[fail(display = "{}", _0)]
+    Io(#[cause] IoError<T>)
+}
+
+impl<T> From<IoError<T>> for Error<T> {
+    fn from(value: IoError<T>) -> Error<T> {
+        Error::Io(value)
+    }
+}
+
+#[cfg(feature = "std_artiq")]
+impl From<::std_artiq::io::Error> for Error<::std_artiq::io::Error> {
+    fn from(value: ::std_artiq::io::Error) -> Error<::std_artiq::io::Error> {
+        Error::Io(IoError::Other(value))
+    }
+}
+
+pub fn read_magic<R>(reader: &mut R) -> Result<(), Error<R::ReadError>>
+    where R: Read + ?Sized
+{
+    const MAGIC: &'static [u8] = b"ARTIQ moninj\n";
+
+    let mut magic: [u8; 13] = [0; 13];
+    reader.read_exact(&mut magic)?;
+    if magic != MAGIC {
+        Err(Error::WrongMagic)
+    } else {
+        Ok(())
+    }
+}
 
 #[derive(Debug)]
 pub enum HostMessage {
@@ -14,7 +51,9 @@ pub enum DeviceMessage {
 }
 
 impl HostMessage {
-    pub fn read_from<T: Read + ?Sized>(reader: &mut T) -> Result<Self, Error<T::ReadError>> {
+    pub fn read_from<R>(reader: &mut R) -> Result<Self, Error<R::ReadError>>
+        where R: Read + ?Sized
+    {
         Ok(match reader.read_u8()? {
             0 => HostMessage::Monitor {
                 enable: if reader.read_u8()? == 0 { false } else { true },
@@ -30,13 +69,15 @@ impl HostMessage {
                 channel: reader.read_u32()?,
                 overrd: reader.read_u8()?
             },
-            _ => return Err(Error::Unrecognized)
+            ty => return Err(Error::UnknownPacket(ty))
         })
     }
 }
 
 impl DeviceMessage {
-    pub fn write_to<T: Write + ?Sized>(&self, writer: &mut T) -> Result<(), Error<T::WriteError>> {
+    pub fn write_to<W>(&self, writer: &mut W) -> Result<(), IoError<W::WriteError>>
+        where W: Write + ?Sized
+    {
         match *self {
             DeviceMessage::MonitorStatus { channel, probe, value } => {
                 writer.write_u8(0)?;

artiq/firmware/libproto_artiq/rpc_proto.rs

@@ -4,10 +4,11 @@ use cslice::{CSlice, CMutSlice};
 use io::{ProtoRead, Read, Write, ProtoWrite, Error};
 use self::tag::{Tag, TagIterator, split_tag};
 
-unsafe fn recv_value<T>(reader: &mut T, tag: Tag, data: &mut *mut (),
+unsafe fn recv_value<R, E>(reader: &mut R, tag: Tag, data: &mut *mut (),
-                        alloc: &Fn(usize) -> Result<*mut (), Error<T::ReadError>>)
+                           alloc: &Fn(usize) -> Result<*mut (), E>)
-                       -> Result<(), Error<T::ReadError>>
+                          -> Result<(), E>
-    where T: Read + ?Sized
+    where R: Read + ?Sized,
+          E: From<Error<R::ReadError>>
 {
     macro_rules! consume_value {
         ($ty:ty, |$ptr:ident| $map:expr) => ({
@@ -74,10 +75,11 @@ unsafe fn recv_value<T>(reader: &mut T, tag: Tag, data: &mut *mut (),
     }
 }
 
-pub fn recv_return<T>(reader: &mut T, tag_bytes: &[u8], data: *mut (),
+pub fn recv_return<R, E>(reader: &mut R, tag_bytes: &[u8], data: *mut (),
-                      alloc: &Fn(usize) -> Result<*mut (), Error<T::ReadError>>)
+                         alloc: &Fn(usize) -> Result<*mut (), E>)
-                     -> Result<(), Error<T::ReadError>>
+                        -> Result<(), E>
-    where T: Read + ?Sized
+    where R: Read + ?Sized,
+          E: From<Error<R::ReadError>>
 {
     let mut it = TagIterator::new(tag_bytes);
     #[cfg(feature = "log")]
@@ -90,9 +92,9 @@ pub fn recv_return<T>(reader: &mut T, tag_bytes: &[u8], data: *mut (),
     Ok(())
 }
 
-unsafe fn send_value<T>(writer: &mut T, tag: Tag, data: &mut *const ())
+unsafe fn send_value<W>(writer: &mut W, tag: Tag, data: &mut *const ())
-                       -> Result<(), Error<T::WriteError>>
+                       -> Result<(), Error<W::WriteError>>
-    where T: Write + ?Sized
+    where W: Write + ?Sized
 {
     macro_rules! consume_value {
         ($ty:ty, |$ptr:ident| $map:expr) => ({
@@ -167,9 +169,9 @@ unsafe fn send_value<T>(writer: &mut T, tag: Tag, data: &mut *const ())
     }
 }
 
-pub fn send_args<T>(writer: &mut T, service: u32, tag_bytes: &[u8], data: *const *const ())
+pub fn send_args<W>(writer: &mut W, service: u32, tag_bytes: &[u8], data: *const *const ())
-                   -> Result<(), Error<T::WriteError>>
+                   -> Result<(), Error<W::WriteError>>
-    where T: Write + ?Sized
+    where W: Write + ?Sized
 {
     let (arg_tags_bytes, return_tag_bytes) = split_tag(tag_bytes);
 

artiq/firmware/libproto_artiq/session_proto.rs

@@ -1,8 +1,59 @@
+use core::str::Utf8Error;
 use alloc::{Vec, String};
 
-use io::{Read, ProtoRead, Write, ProtoWrite, Error};
+use io::{Read, ProtoRead, Write, ProtoWrite, Error as IoError, ReadStringError};
 
-fn read_sync<T: Read + ?Sized>(reader: &mut T) -> Result<(), Error<T::ReadError>> {
+#[derive(Fail, Debug)]
+pub enum Error<T> {
+    #[fail(display = "incorrect magic")]
+    WrongMagic,
+    #[fail(display = "unknown packet {:#02x}", _0)]
+    UnknownPacket(u8),
+    #[fail(display = "invalid UTF-8: {}", _0)]
+    Utf8(Utf8Error),
+    #[fail(display = "{}", _0)]
+    Io(#[cause] IoError<T>)
+}
+
+impl<T> From<IoError<T>> for Error<T> {
+    fn from(value: IoError<T>) -> Error<T> {
+        Error::Io(value)
+    }
+}
+
+impl<T> From<ReadStringError<IoError<T>>> for Error<T> {
+    fn from(value: ReadStringError<IoError<T>>) -> Error<T> {
+        match value {
+            ReadStringError::Utf8(err) => Error::Utf8(err),
+            ReadStringError::Other(err) => Error::Io(err)
+        }
+    }
+}
+
+#[cfg(feature = "std_artiq")]
+impl From<::std_artiq::io::Error> for Error<::std_artiq::io::Error> {
+    fn from(value: ::std_artiq::io::Error) -> Error<::std_artiq::io::Error> {
+        Error::Io(IoError::Other(value))
+    }
+}
+
+pub fn read_magic<R>(reader: &mut R) -> Result<(), Error<R::ReadError>>
+    where R: Read + ?Sized
+{
+    const MAGIC: &'static [u8] = b"ARTIQ coredev\n";
+
+    let mut magic: [u8; 14] = [0; 14];
+    reader.read_exact(&mut magic)?;
+    if magic != MAGIC {
+        Err(Error::WrongMagic)
+    } else {
+        Ok(())
+    }
+}
+
+fn read_sync<R>(reader: &mut R) -> Result<(), IoError<R::ReadError>>
+    where R: Read + ?Sized
+{
     let mut sync = [0; 4];
     for i in 0.. {
         sync[i % 4] = reader.read_u8()?;
@@ -11,7 +62,9 @@ fn read_sync<T: Read + ?Sized>(reader: &mut T) -> Result<(), Error<T::ReadError>
     Ok(())
 }
 
-fn write_sync<T: Write + ?Sized>(writer: &mut T) -> Result<(), Error<T::WriteError>> {
+fn write_sync<W>(writer: &mut W) -> Result<(), IoError<W::WriteError>>
+    where W: Write + ?Sized
+{
     writer.write_all(&[0x5a; 4])
 }
 
@@ -41,7 +94,9 @@ pub enum Request {
 }
 
 impl Request {
-    pub fn read_from<T: Read + ?Sized>(reader: &mut T) -> Result<Self, Error<T::ReadError>> {
+    pub fn read_from<R>(reader: &mut R) -> Result<Self, Error<R::ReadError>>
+        where R: Read + ?Sized
+    {
         read_sync(reader)?;
         Ok(match reader.read_u8()? {
             3  => Request::SystemInfo,
@@ -73,7 +128,7 @@ impl Request {
             12 => Request::FlashRemove {
                 key: reader.read_string()?
             },
-            _  => return Err(Error::Unrecognized)
+            ty  => return Err(Error::UnknownPacket(ty))
         })
     }
 }
@@ -114,7 +169,9 @@ pub enum Reply<'a> {
 }
 
 impl<'a> Reply<'a> {
-    pub fn write_to<T: Write + ?Sized>(&self, writer: &mut T) -> Result<(), Error<T::WriteError>> {
+    pub fn write_to<W>(&self, writer: &mut W) -> Result<(), IoError<W::WriteError>>
+        where W: Write + ?Sized
+    {
         write_sync(writer)?;
         match *self {
             Reply::SystemInfo { ident, finished_cleanly } => {

artiq/firmware/libstd_artiq/Cargo.toml

@@ -10,3 +10,6 @@ path = "lib.rs"
 [features]
 alloc = []
 io_error_alloc = []
+
+[dependencies]
+failure = { version = "0.1", default-features = false }

artiq/firmware/libstd_artiq/lib.rs

@@ -8,6 +8,7 @@ extern crate std_unicode;
 #[macro_use]
 #[macro_reexport(vec, format)]
 extern crate alloc;
+extern crate failure;
 
 pub use core::{any, cell, clone, cmp, convert, default, hash, iter, marker, mem, num,
     ops, option, ptr, result, sync,
@@ -38,3 +39,5 @@ impl<T> FakeBox<T> {
         val
     }
 }
+
+impl failure::Fail for error::Error + Send + Sync {}

artiq/firmware/runtime/Cargo.toml

@@ -14,6 +14,8 @@ build_misoc = { path = "../libbuild_misoc" }
 build_artiq = { path = "../libbuild_artiq" }
 
 [dependencies]
+failure = { version = "0.1", default-features = false }
+failure_derive = { version = "0.1", default-features = false }
 byteorder = { version = "1.0", default-features = false }
 cslice = { version = "0.3" }
 log = { version = "0.4", default-features = false }

artiq/firmware/runtime/kern_hwreq.rs

@@ -1,7 +1,6 @@
-use io;
 use kernel_proto as kern;
 use sched::Io;
-use session::{kern_acknowledge, kern_send};
+use session::{kern_acknowledge, kern_send, Error};
 #[cfg(has_rtio_core)]
 use rtio_mgt;
 
@@ -291,7 +290,7 @@ mod spi {
     }
 }
 
-pub fn process_kern_hwreq(io: &Io, request: &kern::Message) -> Result<bool, io::Error<::std::io::Error>> {
+pub fn process_kern_hwreq(io: &Io, request: &kern::Message) -> Result<bool, Error<::std::io::Error>> {
     match request {
         #[cfg(has_rtio_core)]
         &kern::RtioInitRequest => {

artiq/firmware/runtime/main.rs

@@ -2,6 +2,9 @@
 #![feature(lang_items, alloc, global_allocator, try_from, nonzero, nll)]
 
 extern crate alloc;
+extern crate failure;
+#[macro_use]
+extern crate failure_derive;
 extern crate cslice;
 #[macro_use]
 extern crate log;

artiq/firmware/runtime/mgmt.rs

@@ -1,27 +1,15 @@
 use log::{self, LevelFilter};
 
-use io::{self, Read, Write, ProtoWrite};
+use io::{Write, ProtoWrite, Error as IoError};
 use board_misoc::boot;
 use logger_artiq::BufferLogger;
+use mgmt_proto::*;
 use sched::Io;
 use sched::{TcpListener, TcpStream};
-use mgmt_proto::*;
 use profiler;
 
-fn check_magic(stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Error>> {
+fn worker(io: &Io, stream: &mut TcpStream) -> Result<(), Error<::std::io::Error>> {
-    const MAGIC: &'static [u8] = b"ARTIQ management\n";
+    read_magic(stream)?;
-
-    let mut magic: [u8; 17] = [0; 17];
-    stream.read_exact(&mut magic)?;
-    if magic != MAGIC {
-        Err(io::Error::Unrecognized)
-    } else {
-        Ok(())
-    }
-}
-
-fn worker(io: &Io, stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Error>> {
-    check_magic(stream)?;
     info!("new connection from {}", stream.remote_endpoint());
 
     loop {
@@ -34,7 +22,7 @@ fn worker(io: &Io, stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Er
             }
 
             Request::ClearLog => {
-                BufferLogger::with(|logger| -> Result<(), io::Error<::std::io::Error>> {
+                BufferLogger::with(|logger| -> Result<(), Error<::std::io::Error>> {
                     let mut buffer = io.until_ok(|| logger.buffer())?;
                     Ok(buffer.clear())
                 })?;
@@ -43,7 +31,7 @@ fn worker(io: &Io, stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Er
             }
 
             Request::PullLog => {
-                BufferLogger::with(|logger| -> Result<(), io::Error<::std::io::Error>> {
+                BufferLogger::with(|logger| -> Result<(), Error<::std::io::Error>> {
                     loop {
                         // Do this *before* acquiring the buffer, since that sets the log level
                         // to OFF.
@@ -165,7 +153,7 @@ pub fn thread(io: Io) {
             let mut stream = TcpStream::from_handle(&io, stream);
             match worker(&io, &mut stream) {
                 Ok(()) => (),
-                Err(io::Error::UnexpectedEnd) => (),
+                Err(Error::Io(IoError::UnexpectedEnd)) => (),
                 Err(err) => error!("aborted: {}", err)
             }
         });

artiq/firmware/runtime/moninj.rs

@@ -1,26 +1,12 @@
 use alloc::btree_map::BTreeMap;
 
-use io::{self, Read};
+use moninj_proto::*;
 use sched::Io;
 use sched::{TcpListener, TcpStream};
 use board_misoc::{clock, csr};
 #[cfg(has_drtio)]
 use drtioaux;
 
-use moninj_proto::*;
-
-fn check_magic(stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Error>> {
-    const MAGIC: &'static [u8] = b"ARTIQ moninj\n";
-
-    let mut magic: [u8; 13] = [0; 13];
-    stream.read_exact(&mut magic)?;
-    if magic != MAGIC {
-        Err(io::Error::Unrecognized)
-    } else {
-        Ok(())
-    }
-}
-
 #[cfg(has_rtio_moninj)]
 fn read_probe_local(channel: u16, probe: u8) -> u32 {
     unsafe {
@@ -159,11 +145,11 @@ fn read_injection_status(channel: u32, probe: u8) -> u8 {
     0
 }
 
-fn connection_worker(io: &Io, mut stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Error>> {
+fn connection_worker(io: &Io, mut stream: &mut TcpStream) -> Result<(), Error<::std::io::Error>> {
     let mut watch_list = BTreeMap::new();
     let mut next_check = 0;
 
-    check_magic(&mut stream)?;
+    read_magic(&mut stream)?;
     info!("new connection from {}", stream.remote_endpoint());
 
     loop {

artiq/firmware/runtime/session.rs

@@ -2,7 +2,7 @@ use core::{mem, str, cell::{Cell, RefCell}, fmt::Write as FmtWrite};
 use alloc::{Vec, String};
 use byteorder::{ByteOrder, NetworkEndian};
 
-use io::{self, Read, Write};
+use io::{self, Read, Write, Error as IoError};
 use board_misoc::{ident, cache, config};
 use {mailbox, rpc_queue, kernel};
 use urc::Urc;
@@ -19,17 +19,46 @@ use rpc_proto as rpc;
 use session_proto as host;
 use kernel_proto as kern;
 
-macro_rules! unexpected {
+#[derive(Fail, Debug)]
-    ($($arg:tt)*) => {
+pub enum Error<T> {
-        {
+    #[fail(display = "cannot load kernel: {}", _0)]
-            error!($($arg)*);
+    Load(String),
-            return Err(io::Error::Unrecognized)
+    #[fail(display = "kernel not found")]
-        }
+    KernelNotFound,
-    };
+    #[fail(display = "invalid kernel CPU pointer: {:#08x}", _0)]
+    InvalidPointer(usize),
+    #[fail(display = "RTIO clock failure")]
+    ClockFailure,
+    #[fail(display = "watchdog {} expired", _0)]
+    WatchdogExpired(usize),
+    #[fail(display = "out of watchdogs")]
+    OutOfWatchdogs,
+    #[fail(display = "protocol error: {}", _0)]
+    Protocol(#[cause] host::Error<T>),
+    #[fail(display = "{}", _0)]
+    Unexpected(String),
 }
 
-fn io_error(msg: &str) -> io::Error<::std::io::Error> {
+impl<T> From<host::Error<T>> for Error<T> {
-    io::Error::Other(::std::io::Error::new(::std::io::ErrorKind::Other, msg))
+    fn from(value: host::Error<T>) -> Error<T> {
+        Error::Protocol(value)
+    }
+}
+
+impl From<::std::io::Error> for Error<::std::io::Error> {
+    fn from(value: ::std::io::Error) -> Error<::std::io::Error> {
+        Error::Protocol(host::Error::Io(io::Error::Other(value)))
+    }
+}
+
+impl From<io::Error<::std::io::Error>> for Error<::std::io::Error> {
+    fn from(value: io::Error<::std::io::Error>) -> Error<::std::io::Error> {
+        Error::Protocol(host::Error::Io(value))
+    }
+}
+
+macro_rules! unexpected {
+     ($($arg:tt)*) => (return Err(Error::Unexpected(format!($($arg)*))));
 }
 
 // Persistent state
@@ -102,20 +131,10 @@ impl<'a> Drop for Session<'a> {
     }
 }
 
-fn check_magic(stream: &mut TcpStream) -> Result<(), io::Error<::std::io::Error>> {
+fn host_read<R>(reader: &mut R) -> Result<host::Request, Error<R::ReadError>>
-    const MAGIC: &'static [u8] = b"ARTIQ coredev\n";
+    where R: Read + ?Sized
-
+{
-    let mut magic: [u8; 14] = [0; 14];
+    let request = host::Request::read_from(reader)?;
-    stream.read_exact(&mut magic)?;
-    if magic != MAGIC {
-        Err(io::Error::Unrecognized)
-    } else {
-        Ok(())
-    }
-}
-
-fn host_read(stream: &mut TcpStream) -> Result<host::Request, io::Error<::std::io::Error>> {
-    let request = host::Request::read_from(stream)?;
     match &request {
         &host::Request::LoadKernel(_) => debug!("comm<-host LoadLibrary(...)"),
         _ => debug!("comm<-host {:?}", request)
@@ -123,12 +142,14 @@ fn host_read(stream: &mut TcpStream) -> Result<host::Request, io::Error<::std::i
     Ok(request)
 }
 
-fn host_write(stream: &mut TcpStream, reply: host::Reply) -> Result<(), io::Error<::std::io::Error>> {
+fn host_write<W>(writer: &mut W, reply: host::Reply) -> Result<(), IoError<W::WriteError>>
+    where W: Write + ?Sized
+{
     debug!("comm->host {:?}", reply);
-    reply.write_to(stream)
+    reply.write_to(writer)
 }
 
-pub fn kern_send(io: &Io, request: &kern::Message) -> Result<(), io::Error<::std::io::Error>> {
+pub fn kern_send(io: &Io, request: &kern::Message) -> Result<(), Error<::std::io::Error>> {
     match request {
         &kern::LoadRequest(_) => debug!("comm->kern LoadRequest(...)"),
         &kern::DmaRetrieveReply { trace, duration } => {
@@ -144,12 +165,11 @@ pub fn kern_send(io: &Io, request: &kern::Message) -> Result<(), io::Error<::std
     Ok(io.until(mailbox::acknowledged)?)
 }
 
-fn kern_recv_notrace<R, F>(io: &Io, f: F) -> Result<R, io::Error<::std::io::Error>>
+fn kern_recv_notrace<R, F>(io: &Io, f: F) -> Result<R, Error<::std::io::Error>>
-        where F: FnOnce(&kern::Message) -> Result<R, io::Error<::std::io::Error>> {
+        where F: FnOnce(&kern::Message) -> Result<R, Error<::std::io::Error>> {
     io.until(|| mailbox::receive() != 0)?;
     if !kernel::validate(mailbox::receive()) {
-        let message = format!("invalid kernel CPU pointer 0x{:x}", mailbox::receive());
+        return Err(Error::InvalidPointer(mailbox::receive()))
-        return Err(io::Error::Other(::std::io::Error::new(::std::io::ErrorKind::InvalidData, message)))
     }
 
     f(unsafe { &*(mailbox::receive() as *const kern::Message) })
@@ -171,20 +191,21 @@ fn kern_recv_dotrace(reply: &kern::Message) {
 }
 
 #[inline(always)]
-fn kern_recv<R, F>(io: &Io, f: F) -> Result<R, io::Error<::std::io::Error>>
+fn kern_recv<R, F>(io: &Io, f: F) -> Result<R, Error<::std::io::Error>>
-        where F: FnOnce(&kern::Message) -> Result<R, io::Error<::std::io::Error>> {
+        where F: FnOnce(&kern::Message) -> Result<R, Error<::std::io::Error>> {
     kern_recv_notrace(io, |reply| {
         kern_recv_dotrace(reply);
         f(reply)
     })
 }
 
-pub fn kern_acknowledge() -> Result<(), io::Error<::std::io::Error>> {
+pub fn kern_acknowledge() -> Result<(), Error<::std::io::Error>> {
     mailbox::acknowledge();
     Ok(())
 }
 
-unsafe fn kern_load(io: &Io, session: &mut Session, library: &[u8]) -> Result<(), io::Error<::std::io::Error>> {
+unsafe fn kern_load(io: &Io, session: &mut Session, library: &[u8])
+                   -> Result<(), Error<::std::io::Error>> {
     if session.running() {
         unexpected!("attempted to load a new kernel while a kernel was running")
     }
@@ -194,14 +215,13 @@ unsafe fn kern_load(io: &Io, session: &mut Session, library: &[u8]) -> Result<()
     kern_send(io, &kern::LoadRequest(&library))?;
     kern_recv(io, |reply| {
         match reply {
-            &kern::LoadReply(Ok(())) => {
+            kern::LoadReply(Ok(())) => {
                 session.kernel_state = KernelState::Loaded;
                 Ok(())
             }
-            &kern::LoadReply(Err(ref error)) => {
+            kern::LoadReply(Err(error)) => {
                 kernel::stop();
-                Err(io::Error::Other(::std::io::Error::new(::std::io::ErrorKind::Other,
+                Err(Error::Load(format!("{}", error)))
-                                                       format!("cannot load kernel: {}", error))))
             }
             other =>
                 unexpected!("unexpected reply from kernel CPU: {:?}", other)
@@ -209,7 +229,7 @@ unsafe fn kern_load(io: &Io, session: &mut Session, library: &[u8]) -> Result<()
     })
 }
 
-fn kern_run(session: &mut Session) -> Result<(), io::Error<::std::io::Error>> {
+fn kern_run(session: &mut Session) -> Result<(), Error<::std::io::Error>> {
     if session.kernel_state != KernelState::Loaded {
         unexpected!("attempted to run a kernel while not in Loaded state")
     }
@@ -221,15 +241,14 @@ fn kern_run(session: &mut Session) -> Result<(), io::Error<::std::io::Error>> {
 
 fn process_host_message(io: &Io,
                         stream: &mut TcpStream,
-                        session: &mut Session) -> Result<(), io::Error<::std::io::Error>> {
+                        session: &mut Session) -> Result<(), Error<::std::io::Error>> {
     match host_read(stream)? {
         host::Request::SystemInfo => {
             host_write(stream, host::Reply::SystemInfo {
                 ident: ident::read(&mut [0; 64]),
                 finished_cleanly: session.congress.finished_cleanly.get()
             })?;
-            session.congress.finished_cleanly.set(true);
+            session.congress.finished_cleanly.set(true)
-            Ok(())
         }
 
         // artiq_coreconfig
@@ -239,29 +258,26 @@ fn process_host_message(io: &Io,
                     Ok(value) => host_write(stream, host::Reply::FlashRead(&value)),
                     Err(_)    => host_write(stream, host::Reply::FlashError)
                 }
-            })
+            })?;
         }
-
         host::Request::FlashWrite { ref key, ref value } => {
             match config::write(key, value) {
                 Ok(_)  => host_write(stream, host::Reply::FlashOk),
                 Err(_) => host_write(stream, host::Reply::FlashError)
+            }?;
         }
-        }
-
         host::Request::FlashRemove { ref key } => {
             match config::remove(key) {
                 Ok(()) => host_write(stream, host::Reply::FlashOk),
                 Err(_) => host_write(stream, host::Reply::FlashError),
-            }
+            }?;
 
         }
-
         host::Request::FlashErase => {
             match config::erase() {
                 Ok(()) => host_write(stream, host::Reply::FlashOk),
                 Err(_) => host_write(stream, host::Reply::FlashError),
-            }
+            }?;
         }
 
         // artiq_run/artiq_master
@@ -273,9 +289,9 @@ fn process_host_message(io: &Io,
             #[cfg(has_rtio_core)]
             {
                 if rtio_mgt::crg::switch_clock(clk) {
-                    host_write(stream, host::Reply::ClockSwitchCompleted)
+                    host_write(stream, host::Reply::ClockSwitchCompleted)?;
                 } else {
-                    host_write(stream, host::Reply::ClockSwitchFailed)
+                    host_write(stream, host::Reply::ClockSwitchFailed)?;
                 }
             }
 
@@ -285,19 +301,18 @@ fn process_host_message(io: &Io,
 
         host::Request::LoadKernel(kernel) =>
             match unsafe { kern_load(io, session, &kernel) } {
-                Ok(()) => host_write(stream, host::Reply::LoadCompleted),
+                Ok(()) => host_write(stream, host::Reply::LoadCompleted)?,
                 Err(error) => {
                     let mut description = String::new();
                     write!(&mut description, "{}", error).unwrap();
                     host_write(stream, host::Reply::LoadFailed(&description))?;
-                    kern_acknowledge()
+                    kern_acknowledge()?;
                 }
             },
-
         host::Request::RunKernel =>
             match kern_run(session) {
-                Ok(()) => Ok(()),
+                Ok(()) => (),
-                Err(_) => host_write(stream, host::Reply::KernelStartupFailed)
+                Err(_) => host_write(stream, host::Reply::KernelStartupFailed)?
             },
 
         host::Request::RpcReply { tag } => {
@@ -311,7 +326,7 @@ fn process_host_message(io: &Io,
                     other => unexpected!("unexpected reply from kernel CPU: {:?}", other)
                 }
             })?;
-            rpc::recv_return(stream, &tag, slot, &|size| {
+            rpc::recv_return(stream, &tag, slot, &|size| -> Result<_, Error<::std::io::Error>> {
                 kern_send(io, &kern::RpcRecvReply(Ok(size)))?;
                 Ok(kern_recv(io, |reply| {
                     match reply {
@@ -322,8 +337,7 @@ fn process_host_message(io: &Io,
             })?;
             kern_send(io, &kern::RpcRecvReply(Ok(0)))?;
 
-            session.kernel_state = KernelState::Running;
+            session.kernel_state = KernelState::Running
-            Ok(())
         }
 
         host::Request::RpcException {
@@ -352,14 +366,15 @@ fn process_host_message(io: &Io,
             };
             kern_send(io, &kern::RpcRecvReply(Err(exn)))?;
 
-            session.kernel_state = KernelState::Running;
+            session.kernel_state = KernelState::Running
+        }
+    }
+
     Ok(())
-        }
-    }
 }
 
 fn process_kern_message(io: &Io, mut stream: Option<&mut TcpStream>,
-                        session: &mut Session) -> Result<bool, io::Error<::std::io::Error>> {
+                        session: &mut Session) -> Result<bool, Error<::std::io::Error>> {
     kern_recv_notrace(io, |request| {
         match (request, session.kernel_state) {
             (&kern::LoadReply(_), KernelState::Loaded) |
@@ -383,8 +398,9 @@ fn process_kern_message(io: &Io, mut stream: Option<&mut TcpStream>,
         match request {
             &kern::Log(args) => {
                 use std::fmt::Write;
-                session.log_buffer.write_fmt(args)
+                session.log_buffer
-                       .map_err(|_| io_error("cannot append to session log buffer"))?;
+                       .write_fmt(args)
+                       .unwrap_or_else(|_| warn!("cannot append to session log buffer"));
                 session.flush_log_buffer();
                 kern_acknowledge()
             }
@@ -430,11 +446,9 @@ fn process_kern_message(io: &Io, mut stream: Option<&mut TcpStream>,
             }
 
             &kern::WatchdogSetRequest { ms } => {
-                let id = session.watchdog_set.set_ms(ms)
+                let id = session.watchdog_set.set_ms(ms).map_err(|()| Error::OutOfWatchdogs)?;
-                                .map_err(|()| io_error("out of watchdogs"))?;
                 kern_send(io, &kern::WatchdogSetReply { id: id })
             }
-
             &kern::WatchdogClear { id } => {
                 session.watchdog_set.clear(id);
                 kern_acknowledge()
@@ -476,10 +490,9 @@ fn process_kern_message(io: &Io, mut stream: Option<&mut TcpStream>,
                 match stream {
                     None => return Ok(true),
                     Some(ref mut stream) =>
-                        host_write(stream, host::Reply::KernelFinished)
+                        host_write(stream, host::Reply::KernelFinished).map_err(|e| e.into())
                 }
             }
-
             &kern::RunException {
                 exception: kern::Exception { name, message, param, file, line, column, function },
                 backtrace
@@ -505,7 +518,7 @@ fn process_kern_message(io: &Io, mut stream: Option<&mut TcpStream>,
                             column:    column,
                             function:  function,
                             backtrace: backtrace
-                        })
+                        }).map_err(|e| e.into())
                     }
                 }
             }
@@ -516,7 +529,7 @@ fn process_kern_message(io: &Io, mut stream: Option<&mut TcpStream>,
 }
 
 fn process_kern_queued_rpc(stream: &mut TcpStream,
-                           _session: &mut Session) -> Result<(), io::Error<::std::io::Error>> {
+                           _session: &mut Session) -> Result<(), Error<::std::io::Error>> {
     rpc_queue::dequeue(|slice| {
         debug!("comm<-kern (async RPC)");
         let length = NetworkEndian::read_u32(slice) as usize;
@@ -529,7 +542,7 @@ fn process_kern_queued_rpc(stream: &mut TcpStream,
 
 fn host_kernel_worker(io: &Io,
                       stream: &mut TcpStream,
-                      congress: &mut Congress) -> Result<(), io::Error<::std::io::Error>> {
+                      congress: &mut Congress) -> Result<(), Error<::std::io::Error>> {
     let mut session = Session::new(congress);
 
     loop {
@@ -548,16 +561,16 @@ fn host_kernel_worker(io: &Io,
         }
 
         if session.kernel_state == KernelState::Running {
-            if session.watchdog_set.expired() {
+            if let Some(idx) = session.watchdog_set.expired() {
                 host_write(stream, host::Reply::WatchdogExpired)?;
-                return Err(io_error("watchdog expired"))
+                return Err(Error::WatchdogExpired(idx))
             }
 
             #[cfg(has_rtio_core)]
             {
                 if !rtio_mgt::crg::check() {
                     host_write(stream, host::Reply::ClockFailure)?;
-                    return Err(io_error("RTIO clock failure"))
+                    return Err(Error::ClockFailure)
                 }
             }
         }
@@ -568,7 +581,7 @@ fn host_kernel_worker(io: &Io,
 
 fn flash_kernel_worker(io: &Io,
                        congress: &mut Congress,
-                       config_key: &str) -> Result<(), io::Error<::std::io::Error>> {
+                       config_key: &str) -> Result<(), Error<::std::io::Error>> {
     let mut session = Session::new(congress);
 
     config::read(config_key, |result| {
@@ -578,15 +591,14 @@ fn flash_kernel_worker(io: &Io,
                 // so make a copy.
                 kern_load(io, &mut session, Vec::from(kernel).as_ref())
             },
-            _ => Err(io::Error::Other(::std::io::Error::new(::std::io::ErrorKind::NotFound,
+            _ => Err(Error::KernelNotFound)
-                                                        "kernel not found"))),
         }
     })?;
     kern_run(&mut session)?;
 
     loop {
         if !rpc_queue::empty() {
-            return Err(io_error("unexpected background RPC in flash kernel"))
+            unexpected!("unexpected background RPC in flash kernel")
         }
 
         if mailbox::receive() != 0 {
@@ -595,14 +607,14 @@ fn flash_kernel_worker(io: &Io,
             }
         }
 
-        if session.watchdog_set.expired() {
+        if let Some(idx) = session.watchdog_set.expired() {
-            return Err(io_error("watchdog expired"))
+            return Err(Error::WatchdogExpired(idx))
         }
 
         #[cfg(has_rtio_core)]
         {
             if !rtio_mgt::crg::check() {
-                return Err(io_error("RTIO clock failure"))
+                return Err(Error::ClockFailure)
             }
         }
 
@@ -639,10 +651,10 @@ pub fn thread(io: Io) {
             let mut congress = congress.borrow_mut();
             info!("running startup kernel");
             match flash_kernel_worker(&io, &mut congress, "startup_kernel") {
-                Ok(()) => info!("startup kernel finished"),
+                Ok(()) =>
-                Err(io::Error::Other(ref err)) if err.kind() == ::std::io::ErrorKind::NotFound => {
+                    info!("startup kernel finished"),
-                    info!("no startup kernel found")
+                Err(Error::KernelNotFound) =>
-                }
+                    info!("no startup kernel found"),
                 Err(err) => {
                     congress.finished_cleanly.set(false);
                     error!("startup kernel aborted: {}", err);
@@ -657,7 +669,7 @@ pub fn thread(io: Io) {
             stream.set_timeout(Some(1000));
             stream.set_keep_alive(Some(500));
 
-            match check_magic(&mut stream) {
+            match host::read_magic(&mut stream) {
                 Ok(()) => (),
                 Err(_) => {
                     warn!("wrong magic from {}", stream.remote_endpoint());
@@ -674,13 +686,11 @@ pub fn thread(io: Io) {
                 let mut stream = TcpStream::from_handle(&io, stream);
                 match host_kernel_worker(&io, &mut stream, &mut *congress) {
                     Ok(()) => (),
-                    Err(io::Error::UnexpectedEnd) => {
+                    Err(Error::Protocol(host::Error::Io(io::Error::UnexpectedEnd))) =>
-                        info!("connection closed");
+                        info!("connection closed"),
-                    }
+                    Err(Error::Protocol(host::Error::Io(io::Error::Other(ref err))))
-                    Err(io::Error::Other(ref err))
+                            if err.kind() == ::std::io::ErrorKind::Interrupted =>
-                            if err.kind() == ::std::io::ErrorKind::Interrupted => {
+                        info!("kernel interrupted"),
-                        info!("kernel interrupted");
-                    }
                     Err(err) => {
                         congress.finished_cleanly.set(false);
                         error!("session aborted: {}", err);
@@ -698,18 +708,16 @@ pub fn thread(io: Io) {
                 match flash_kernel_worker(&io, &mut *congress, "idle_kernel") {
                     Ok(()) =>
                         info!("idle kernel finished, standing by"),
-                    Err(io::Error::Other(ref err))
+                    Err(Error::Protocol(host::Error::Io(io::Error::Other(ref err))))
                             if err.kind() == ::std::io::ErrorKind::Interrupted => {
                         info!("idle kernel interrupted");
                     }
-                    Err(io::Error::Other(ref err))
+                    Err(Error::KernelNotFound) => {
-                            if err.kind() == ::std::io::ErrorKind::NotFound => {
                         info!("no idle kernel found");
                         while io.relinquish().is_ok() {}
                     }
-                    Err(err) => {
+                    Err(err) =>
-                        error!("idle kernel aborted: {}", err);
+                        error!("idle kernel aborted: {}", err)
-                    }
                 }
             })
         }

artiq/firmware/runtime/watchdog.rs

@@ -38,10 +38,12 @@ impl WatchdogSet {
         }
     }
 
-    pub fn expired(&self) -> bool {
+    pub fn expired(&self) -> Option<usize> {
-        self.watchdogs.iter()
+        self.watchdogs
-            .filter(|wd| wd.active)
+            .iter()
-            .min_by_key(|wd| wd.threshold)
+            .enumerate()
-            .map_or(false, |wd| clock::get_ms() > wd.threshold)
+            .filter(|(_, wd)| wd.active && clock::get_ms() > wd.threshold)
+            .min_by_key(|(_, wd)| wd.threshold)
+            .map_or(None, |(i, _)| Some(i))
     }
 }