Server fixes (comnik#84)

* Match server dependencies * Fix server Scheduler use * Use RealtimeScheduler in csv source * Cargo fmt * Adjust scheduling test
PEZ · Oct 11, 2019 · db3c3ba · db3c3ba
1 parent b1c3fe8
commit db3c3ba
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 13 deletions.
diff --git a/server/Cargo.toml b/server/Cargo.toml
@@ -7,8 +7,8 @@ edition = "2018"
 
 [dependencies]
 jemallocator = "0.1.8"
-timely = { version = "0.10.0", features = ["bincode"] }
-differential-dataflow = { version = "0.10.0" }
+timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", features = ["bincode"] }
+differential-dataflow = { git = "https://github.com/TimelyDataflow/differential-dataflow" }
 declarative-dataflow = { path = "../", features = ["serde_json", "uuid"] }
 serde = "1"
 serde_derive = "1"

diff --git a/server/src/main.rs b/server/src/main.rs
@@ -20,8 +20,9 @@ use timely::synchronization::Sequencer;
 use differential_dataflow::logging::DifferentialEvent;
 use differential_dataflow::operators::Consolidate;
 
+use declarative_dataflow::scheduling::{AsScheduler, SchedulingEvent};
 use declarative_dataflow::server;
-use declarative_dataflow::server::{scheduler, CreateAttribute, Request, Server, TxId};
+use declarative_dataflow::server::{CreateAttribute, Request, Server, TxId};
 use declarative_dataflow::sinks::{Sinkable, SinkingContext};
 use declarative_dataflow::timestamp::{Coarsen, Time};
 use declarative_dataflow::{Output, ResultDiff};
@@ -306,10 +307,10 @@ fn main() {
 
             if server.scheduler.borrow().has_pending() {
                 let mut scheduler = server.scheduler.borrow_mut();
-                while let Some(activator) = scheduler.next() {
+                while let Some(activator) = scheduler.realtime.next() {
                     if let Some(event) = activator.schedule() {
                         match event {
-                            scheduler::Event::Tick => {
+                            SchedulingEvent::Tick => {
                                 sequencer.push(Command {
                                     owner: worker.index(),
                                     client: SYSTEM.0,
@@ -521,7 +522,7 @@ fn main() {
                                 if let Some(tick) = server_config.tick {
                                     let interval_end = Instant::now().duration_since(worker.timer()).coarsen(&tick);
                                     let at = worker.timer() + interval_end;
-                                    server.scheduler.borrow_mut().event_at(at, scheduler::Event::Tick);
+                                    server.scheduler.borrow_mut().realtime.event_at(at, SchedulingEvent::Tick);
                                 }
                             }
 
@@ -577,7 +578,7 @@ fn main() {
 
             // Finally, we give the CPU a chance to chill, if no work
             // remains.
-            let delay = server.scheduler.borrow().until_next().unwrap_or(Duration::from_millis(100));
+            let delay = server.scheduler.borrow().realtime.until_next().unwrap_or(Duration::from_millis(100));
             worker.step_or_park(Some(delay));
         }
 

diff --git a/src/scheduling/mod.rs b/src/scheduling/mod.rs
@@ -8,7 +8,8 @@ pub mod frontier_scheduler;
 pub use frontier_scheduler::FrontierScheduler;
 
 pub mod realtime_scheduler;
-pub use realtime_scheduler::{Event, RealtimeScheduler};
+pub use realtime_scheduler::Event as SchedulingEvent;
+pub use realtime_scheduler::RealtimeScheduler;
 
 /// Common scheduler behaviour.
 pub trait AsScheduler {

diff --git a/src/sources/csv_file.rs b/src/sources/csv_file.rs
@@ -193,6 +193,7 @@ impl<S: Scope<Timestamp = Duration>> Sourceable<S> for CsvFile {
                                 .upgrade()
                                 .unwrap()
                                 .borrow_mut()
+                                .realtime
                                 .schedule_after(interval, Rc::downgrade(&activator))
                         }
                     }

diff --git a/tests/scheduling_test.rs b/tests/scheduling_test.rs
@@ -1,4 +1,4 @@
-use declarative_dataflow::scheduling::{AsScheduler, Event, RealtimeScheduler};
+use declarative_dataflow::scheduling::{AsScheduler, RealtimeScheduler, SchedulingEvent};
 use std::time::Duration;
 
 #[test]
@@ -8,17 +8,20 @@ fn test_schedule_now() {
     assert!(!scheduler.has_pending());
     assert!(scheduler.until_next().is_none());
 
-    scheduler.event_after(Duration::from_secs(0), Event::Tick);
+    scheduler.event_after(Duration::from_secs(0), SchedulingEvent::Tick);
 
     assert!(scheduler.has_pending());
-    assert_eq!(scheduler.next().unwrap().schedule(), Some(Event::Tick));
+    assert_eq!(
+        scheduler.next().unwrap().schedule(),
+        Some(SchedulingEvent::Tick)
+    );
 }
 
 #[test]
 fn test_schedule_after() {
     let mut scheduler = RealtimeScheduler::new();
 
-    scheduler.event_after(Duration::from_secs(2), Event::Tick);
+    scheduler.event_after(Duration::from_secs(2), SchedulingEvent::Tick);
 
     assert!(!scheduler.has_pending());
     assert!(scheduler.next().is_none());
@@ -27,6 +30,9 @@ fn test_schedule_after() {
     std::thread::sleep(scheduler.until_next().unwrap());
 
     assert!(scheduler.has_pending());
-    assert_eq!(scheduler.next().unwrap().schedule(), Some(Event::Tick));
+    assert_eq!(
+        scheduler.next().unwrap().schedule(),
+        Some(SchedulingEvent::Tick)
+    );
     assert!(scheduler.until_next().is_none());
 }