Domains first

.gitignore

@@ -1,4 +1,5 @@
 /target
+/experiments/target
 /data
 
 .DS_Store

cli/examples/schema.json

@@ -12,7 +12,6 @@
         "input_semantics": "Distinct",
         "query_support": "Basic",
         "index_direction": "Forward",
-        "timeless": false
       }
     }
   },
@@ -29,7 +28,6 @@
         "input_semantics": "Distinct",
         "query_support": "Basic",
         "index_direction": "Forward",
-        "timeless": false
       }
     }
   },
@@ -46,7 +44,6 @@
         "input_semantics": "Distinct",
         "query_support": "Basic",
         "index_direction": "Forward",
-        "timeless": false
       }
     }
   },
@@ -63,7 +60,6 @@
         "input_semantics": "Distinct",
         "query_support": "Basic",
         "index_direction": "Forward",
-        "timeless": false
       }
     }
   }

experiments/src/bin/hector_wco.rs

@@ -82,9 +82,7 @@ fn main() {
 
         worker.dataflow::<u64, _, _>(|scope| {
             server
-                .context
-                .internal
-                .create_transactable_attribute("edge", AttributeConfig::tx_time(InputSemantics::Raw), scope)
+                .create_attribute(scope, "edge", AttributeConfig::tx_time(InputSemantics::Raw))
                 .unwrap();
 
             server

server/src/main.rs

@@ -371,6 +371,61 @@ fn main() {
 
                     let result = match req {
                         Request::Transact(req) => server.transact(req, owner, worker.index()),
+                        Request::Subscribe(aid) => {
+                            let interests = server.interests
+                                .entry(aid.clone())
+                                .or_insert_with(HashSet::new);
+
+                            // All workers keep track of every client's interests, s.t. they
+                            // know when to clean up unused dataflows.
+                            interests.insert(Token(client));
+
+                            if interests.len() > 1 {
+                                // We only want to setup the dataflow on
+                                // the first interest.
+                                Ok(())
+                            } else {
+                                let send_results = io.send.clone();
+
+                                let result = worker.dataflow::<T, _, _>(|scope| {
+                                    let (propose, shutdown) = server
+                                        .internal
+                                        .forward_propose(&aid)
+                                        .unwrap()
+                                        .import_frontier(scope, &aid);
+
+                                    // @TODO stash this somewhere
+                                    std::mem::forget(shutdown);
+
+                                    let pact = Exchange::new(move |_| owner as u64);
+
+                                    propose
+                                        .as_collection(|e, v| vec![e.clone(), v.clone()])
+                                        .inner
+                                        .unary(pact, "Subscription", move |_cap, _info| {
+                                            move |input, _output: &mut OutputHandle<_, ResultDiff<T>, _>| {
+                                                // Due to the exchange pact, this closure is only
+                                                // executed by the owning worker.
+
+                                                input.for_each(|_time, data| {
+                                                    let data = data.iter()
+                                                        .map(|(tuple, t, diff)| (tuple.clone(), t.clone().into(), *diff))
+                                                        .collect::<Vec<ResultDiff<Time>>>();
+
+                                                    send_results
+                                                        .send(Output::QueryDiff(aid.clone(), data))
+                                                        .expect("internal channel send failed");
+                                                });
+                                            }
+                                        })
+                                        .probe_with(&mut server.probe);
+
+                                    Ok(())
+                                });
+
+                                result
+                            }
+                        }
                         Request::Interest(req) => {
                             let interests = server.interests
                                 .entry(req.name.clone())
@@ -508,7 +563,7 @@ fn main() {
                             })
                         }
                         Request::AdvanceDomain(name, next) => server.advance_domain(name, next.into()),
-                        Request::CloseInput(name) => server.context.internal.close_input(name),
+                        Request::CloseInput(name) => server.internal.close_input(name),
                         Request::Disconnect => server.disconnect_client(Token(command.client)),
                         Request::Setup => unimplemented!(),
                         Request::Tick => {
@@ -557,7 +612,7 @@ fn main() {
                     #[cfg(feature = "bitemporal")]
                     let next = Pair::new(Instant::now().duration_since(worker.timer()), next_tx as u64);
 
-                    server.context.internal.advance_epoch(next).expect("failed to advance epoch");
+                    server.internal.advance_epoch(next).expect("failed to advance epoch");
                 }
             }
 
@@ -574,7 +629,7 @@ fn main() {
             // might take a decent amount of time, in case traces get
             // compacted. If that happens, we can park less before
             // scheduling the next activator.
-            server.context.internal.advance().expect("failed to advance domain");
+            server.internal.advance().expect("failed to advance domain");
 
             // Finally, we give the CPU a chance to chill, if no work
             // remains.