Cleanup

Author: comnik

Cargo.toml

@@ -22,6 +22,7 @@ serde_derive = "1"
 serde_json = "1"
 rmp-serde = "0.13.7"
 mio = "0.6.16"
+mio-extras = "2.0.5"
 slab = "0.4.1"
 ws = { git = "https://github.com/comnik/ws-rs" }
 log = "0.4"

src/bin/server.rs

@@ -1,24 +1,12 @@
 #[global_allocator]
 static ALLOCATOR: jemallocator::Jemalloc = jemallocator::Jemalloc;
 
-extern crate declarative_dataflow;
-extern crate differential_dataflow;
-extern crate getopts;
-extern crate mio;
 #[macro_use]
 extern crate serde_derive;
-extern crate serde_json;
-extern crate slab;
-extern crate timely;
-extern crate ws;
-
-#[macro_use]
-extern crate log;
-extern crate env_logger;
-
 #[macro_use]
 extern crate abomonation_derive;
-extern crate abomonation;
+#[macro_use]
+extern crate log;
 
 use std::collections::{HashSet, VecDeque};
 use std::io::BufRead;
@@ -91,8 +79,9 @@ fn main() {
 
     let args: Vec<String> = std::env::args().collect();
     let timely_args = std::env::args().take_while(|ref arg| *arg != "--");
+    let timely_config = timely::Configuration::from_args(timely_args).unwrap();
 
-    timely::execute_from_args(timely_args, move |worker| {
+    timely::execute(timely_config, move |worker| {
         // read configuration
         let server_args = args.iter().rev().take_while(|arg| *arg != "--");
         let default_config: Config = Default::default();
@@ -139,13 +128,13 @@ fn main() {
         };
 
         // setup CLI channel
-        let (send_cli, recv_cli) = mio::channel::channel();
+        let (send_cli, recv_cli) = mio_extras::channel::channel();
 
         // setup results channel
-        let (send_results, recv_results) = mio::channel::channel::<(String, Vec<ResultDiff<T>>)>();
+        let (send_results, recv_results) = mio_extras::channel::channel::<(String, Vec<ResultDiff<T>>)>();
 
         // setup errors channel
-        let (send_errors, recv_errors) = mio::channel::channel::<(Vec<Token>, Vec<(Error, TxId)>)>();
+        let (send_errors, recv_errors) = mio_extras::channel::channel::<(Vec<Token>, Vec<(Error, TxId)>)>();
 
         // setup server socket
         // let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), config.port);
@@ -416,14 +405,12 @@ fn main() {
                     _ => {
                         let token = event.token();
                         let active = {
-                            let readiness = event.readiness();
-                            let conn_events = connections[token.into()].events();
+                            let event_readiness = event.readiness();
+                            let conn_readiness = connections[token.into()].events();
 
-                            // @TODO refactor connection to accept a
-                            // vector in which to place events and
-                            // rename conn_events to avoid name clash
+                            // @TODO refactor connection to accept a vector in which to place events
 
-                            if (readiness & conn_events).is_readable() {
+                            if (event_readiness & conn_readiness).is_readable() {
                                 match connections[token.into()].read() {
                                     Err(err) => {
                                         trace!(
@@ -493,9 +480,9 @@ fn main() {
                                 }
                             }
 
-                            let conn_events = connections[token.into()].events();
+                            let conn_readiness = connections[token.into()].events();
 
-                            if (readiness & conn_events).is_writable() {
+                            if (event_readiness & conn_readiness).is_writable() {
                                 if let Err(err) = connections[token.into()].write() {
                                     trace!(
                                         "[WORKER {}] error while writing: {}",
@@ -589,7 +576,7 @@ fn main() {
                                                         .consolidate()
                                                 }
                                                 #[cfg(not(feature = "real-time"))]
-                                                Some(delay) => {
+                                                Some(granularity) => {
                                                     relation
                                                         .delay(move |t| (t/granularity + 1) * granularity)
                                                         .consolidate()

src/domain/mod.rs

@@ -243,18 +243,16 @@ where
 
                     self.forward
                         .get_mut(aid)
-                        .expect(&format!(
-                            "Configuration available for unknown attribute {}",
-                            aid
-                        ))
+                        .unwrap_or_else(|| {
+                            panic!("Configuration available for unknown attribute {}", aid)
+                        })
                         .advance_by(frontier);
 
                     self.reverse
                         .get_mut(aid)
-                        .expect(&format!(
-                            "Configuration available for unknown attribute {}",
-                            aid
-                        ))
+                        .unwrap_or_else(|| {
+                            panic!("Configuration available for unknown attribute {}", aid)
+                        })
                         .advance_by(frontier);
                 }
             }
@@ -265,10 +263,9 @@ where
 
                     self.arrangements
                         .get_mut(name)
-                        .expect(&format!(
-                            "Configuration available for unknown relation {}",
-                            name
-                        ))
+                        .unwrap_or_else(|| {
+                            panic!("Configuration available for unknown relation {}", name)
+                        })
                         .advance_by(frontier);
                 }
             }

src/server/mod.rs

@@ -5,7 +5,7 @@ use std::collections::{HashMap, HashSet};
 use std::hash::Hash;
 use std::ops::Sub;
 use std::rc::Rc;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use timely::dataflow::{ProbeHandle, Scope};
 use timely::order::TotalOrder;
@@ -420,9 +420,9 @@ impl<Token: Hash> Server<u64, Token> {
 }
 
 #[cfg(feature = "real-time")]
-impl<Token: Hash> Server<Duration, Token> {
+impl<Token: Hash> Server<std::time::Duration, Token> {
     /// Handle a RegisterSource request.
-    pub fn register_source<S: Scope<Timestamp = Duration>>(
+    pub fn register_source<S: Scope<Timestamp = std::time::Duration>>(
         &mut self,
         source: Source,
         scope: &mut S,
@@ -437,7 +437,7 @@ impl<Token: Hash> Server<Duration, Token> {
     }
 
     /// Handle a RegisterSink request.
-    pub fn register_sink<S: Scope<Timestamp = Duration>>(
+    pub fn register_sink<S: Scope<Timestamp = std::time::Duration>>(
         &mut self,
         req: RegisterSink,
         scope: &mut S,

src/server/scheduler.rs

@@ -12,6 +12,7 @@ use timely::scheduling::activate::Activator;
 /// (optionally) block for input from within the event loop without
 /// unnecessarily delaying sources that have run out of fuel during
 /// the current step.
+#[derive(Default)]
 pub struct Scheduler {
     activator_queue: BinaryHeap<TimedActivator>,
 }
@@ -28,11 +29,7 @@ impl Scheduler {
     /// scheduled.
     pub fn has_pending(&self) -> bool {
         if let Some(ref timed_activator) = self.activator_queue.peek() {
-            if Instant::now() <= timed_activator.at {
-                true
-            } else {
-                false
-            }
+            Instant::now() <= timed_activator.at
         } else {
             false
         }

src/sources/mod.rs

@@ -2,7 +2,7 @@
 
 use std::cell::RefCell;
 use std::rc::Weak;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use timely::dataflow::{Scope, Stream};
 use timely::order::TotalOrder;
@@ -53,13 +53,13 @@ pub enum Source {
 }
 
 #[cfg(feature = "real-time")]
-impl Sourceable<Duration> for Source {
-    fn source<S: Scope<Timestamp = Duration>>(
+impl Sourceable<std::time::Duration> for Source {
+    fn source<S: Scope<Timestamp = std::time::Duration>>(
         &self,
         scope: &mut S,
         t0: Instant,
         scheduler: Weak<RefCell<Scheduler>>,
-    ) -> Vec<(Aid, Stream<S, ((Value, Value), Duration, isize)>)> {
+    ) -> Vec<(Aid, Stream<S, ((Value, Value), std::time::Duration, isize)>)> {
         match *self {
             Source::TimelyLogging(ref source) => source.source(scope, t0, scheduler),
             Source::DifferentialLogging(ref source) => source.source(scope, t0, scheduler),