From da5740ecdc1346d93c4961cb00f9368c63be18ac Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sun, 16 Aug 2026 19:04:01 -0400 Subject: [PATCH] Remove legacy dynamic-loading server --- Cargo.toml | 5 - server/Cargo.toml | 15 -- server/README.md | 158 ------------------ server/dataflows/degr_dist/Cargo.toml | 14 -- server/dataflows/degr_dist/src/lib.rs | 26 --- server/dataflows/neighborhood/Cargo.toml | 13 -- server/dataflows/neighborhood/src/lib.rs | 35 ---- server/dataflows/random_graph/Cargo.toml | 17 -- server/dataflows/random_graph/src/lib.rs | 196 ----------------------- server/dataflows/reachability/Cargo.toml | 13 -- server/dataflows/reachability/src/lib.rs | 33 ---- server/src/bin/dd_server.rs | 140 ---------------- server/src/lib.rs | 94 ----------- 13 files changed, 759 deletions(-) delete mode 100644 server/Cargo.toml delete mode 100644 server/README.md delete mode 100644 server/dataflows/degr_dist/Cargo.toml delete mode 100644 server/dataflows/degr_dist/src/lib.rs delete mode 100644 server/dataflows/neighborhood/Cargo.toml delete mode 100644 server/dataflows/neighborhood/src/lib.rs delete mode 100644 server/dataflows/random_graph/Cargo.toml delete mode 100644 server/dataflows/random_graph/src/lib.rs delete mode 100644 server/dataflows/reachability/Cargo.toml delete mode 100644 server/dataflows/reachability/src/lib.rs delete mode 100644 server/src/bin/dd_server.rs delete mode 100644 server/src/lib.rs diff --git a/Cargo.toml b/Cargo.toml index 88102a7e3..6786d490b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,11 +5,6 @@ members = [ "dogsdogsdogs", "experiments", "interactive", - "server", - "server/dataflows/degr_dist", - "server/dataflows/neighborhood", - "server/dataflows/random_graph", - "server/dataflows/reachability", #"tpchlike", #"doop", "mdbook", diff --git a/server/Cargo.toml b/server/Cargo.toml deleted file mode 100644 index 4a67a1489..000000000 --- a/server/Cargo.toml +++ /dev/null @@ -1,15 +0,0 @@ -[package] -name = "dd_server" -version = "0.1.0" -authors = ["Frank McSherry "] -edition.workspace = true -rust-version.workspace = true -publish = false - -[lints] -workspace = true - -[dependencies] -libloading = "0.8" -differential-dataflow = { workspace = true } -timely = { workspace = true, features = ["getopts"] } diff --git a/server/README.md b/server/README.md deleted file mode 100644 index 7072c63dc..000000000 --- a/server/README.md +++ /dev/null @@ -1,158 +0,0 @@ -# Graph Server - -A differential dataflow server for continually changing graphs - -## Overview - -The graph server is a system that hosts continually changing graph datasets, and computation defined over them. Users are able to build and load shared libraries that can both define computation over existing graph datasets or define and share new graphs. The user code is able to share the same underlying representations for the graphs, reducing the overhead and streamlining the execution. - -## A Sketch - -The graph server is very much in progress, and its goals are mostly to exercise timely and differential dataflow as systems, and to see where this leads. At the moment, the server provides a small set of graph definitions and graph analyses one can load against them. This is mostly so that we can see what it is like to start up and shut down graph computations with shared state, and understand if we are doing this well. - -Several computations are defined in `./dataflows/` which give examples of defining your own computation. These projects have no special status, and could each be re-implemented by users such as yourself. The intent is that once your project is compiled, the server can load the associated shared libraries and bring your computation into the same shared address space as the other graph computations. - -## Examples - -The main binary is `bin/server.rs`, which can be invoked by running - - cargo run --bin server - -This .. doesn't appear to do anything: - - Echidnatron% cargo run --bin server - Finished dev [unoptimized + debuginfo] target(s) in 0.0 secs - Running `target/debug/server` - -What has happened is that the server is up and running and waiting for some input! Pretty exciting. - -Before getting ahead of ourselves, we'll need to build some code to run. Let's quit out of the server for now and do the following: - - Echidnatron% cd dataflows/random_graph - Echidnatron% cargo build - Compiling random_graph v0.1.0 (file:///Users/mcsherry/Projects/differential-dataflow/grapht/dataflows/random_graph) - Finished dev [unoptimized + debuginfo] target(s) in 5.90 secs - Echidnatron% - -and then - - Echidnatron% cd ../degr_dist - Echidnatron% cargo build - Compiling degr_dist v0.1.0 (file:///Users/mcsherry/Projects/differential-dataflow/grapht/dataflows/degr_dist) - Finished dev [unoptimized + debuginfo] target(s) in 8.14 secs - Echidnatron% - -These commands will build two shared libraries, `librandom_graph.dylib` and `libdegr_dist.dylib`, which we will use in our server! - -Ok, back to the server now. Load that puppy up again and type something after it: - - Echidnatron% cargo run --bin server - Finished dev [unoptimized + debuginfo] target(s) in 0.0 secs - Running `target/debug/server` - load ./dataflows/random_graph/target/debug/librandom_graph.dylib build 1000 2000 10 - -Ok. We have now bound to the string `` a random graph on 1,000 nodes comprising a sliding window over 2,000 edges, which changes ten times every second. If you would like to, you can change any of the arguments passed, though if you increase the rate of change too much you may overload the system. Let's leave it how it is for now. - -Up next, let's attach the `degr_dist` computation to `` and see what we get: - - load ./dataflows/degr_dist/target/debug/libdegr_dist.dylib build - -This will attach our pre-defined degree distribution computation, which goes and computes the stream of changes to the counts of nodes with each out-degree. - -Rather than blast all that information at us (it would only be tens of changes per second, but still), it reports on the distribution of latencies of changes: how long did it take a change from the time it should have entered the system, through to being observed in the output? - -The first line you'll see may look like so (it will depend on the performance of your system): - - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8, 6, 16, 18, 64, 66, 0, 9, 0, 0, 0, 0, 0, 0, 0] - -These counts report the number of observed latencies for each power-of-two number of microseconds. It seems that the lowest latency here is `(1 << 17)` microseconds, or roughly 131 milliseconds. That is a large number, but what is going on here is that the first line is the `degr_dist` computation catching up on historical data. Subsequent lines should look better: - - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - -We are now at entries 11 and 12, corresponding to between one and four millisecond latencies. This number is still pretty large, and it is due to us running without release optimizations. If we turn those on, things look a bit better: - - delays: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8, 0, 16, 6, 26, 46, 14, 0, 9, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - delays: [0, 0, 0, 0, 0, 0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - -We see a similar first batch of latencies, reflecting the gap between historical times and when I typed the second `load` line, followed by latencies that are all in the range of 65 to 128 microseconds range (the index incremented comes from `delay_us.next_power_of_two().trailing_zeros()` methods). - -This is an especially lightly loaded computation, and you can play around with the data generation parameters to get a feel for how the latencies respond as you increase the load. - -## An example computation - -Let's take a closer look at the `degr_dist` computation. What does it look like? - -You can find [the source](https://github.com/frankmcsherry/differential-dataflow/tree/master/grapht/dataflows/degr_dist) in the repository, but it's intentionally concise so let's just take a peek at [`dataflows/degr_dist/src/lib.rs`](https://github.com/frankmcsherry/differential-dataflow/blob/master/grapht/dataflows/degr_dist/src/lib.rs) where all of the logic lives: - -First, there is the standard Rust boilerplate, calling out which crates and traits we want to use. These are complicated only because `degr_dist` uses these types, and could be much simpler with some wrapping. - -```rust -extern crate timely; -extern crate differential_dataflow; -extern crate dd_server; - -use std::rc::Rc; -use std::cell::RefCell; - -use timely::dataflow::operators::inspect::Inspect; -use timely::dataflow::operators::Probe; -use differential_dataflow::operators::CountTotal; -use dd_server::{Environment, TraceHandle}; -``` - -Once we get past the boilerplate, we get to define a method that takes some context about the larger world, and is free to build up some dataflow! - -```rust -#[no_mangle] -pub fn build((dataflow, handles, probe, _timer, args): Environment) -> Result<(), String> { - - if args.len() != 1 { return Err(format!("expected one argument, instead: {:?}", args)); } - - handles - .get_mut::>>>(&args[0])? - .borrow_mut().as_mut().unwrap() - .import(dataflow) - .as_collection(|k,v| (k.clone(), v.clone())) - .map(|(src, _dst)| src as usize).count_total() - .map(|(_src, cnt)| cnt as usize).count_total() - .probe_with(probe); - - Ok(()) -} -``` - -I've deleted the code that does all the printing to the screen, because ideally real computations don't actually write that sort of stuff, but you can check it out in the repository link up above. - -You can drill down on some of the types, but `handles` is a `HashMap>` in which we stash various things, including access to the random graph we created and are continually updating. We can look up the graph by name, import it in to our dataflow, and then write code using standard differential dataflow operators. - -### Stashing outputs - -Watching `println!` statements fly past is only so interesting. Which is to say: "not very". Step one is obviously to comment out the `.inspect()` line, but where do we go from there? Probably, we would want to publish the output, which we can do with our mutable access to `handles`. - -Instead, let's look at what the `random_graph` library does. Now, we aren't going to look at all the code, because there is a lot of random graph stuff, but from the point where we have a differential dataflow collection of edges, which we `probe`, onwards it looks like: - -```rust - let trace = - // .. lots of stuff .. - .probe_with(probe) - .as_collection() - .arrange_by_key_u() - .trace; - - *trace_handle.borrow_mut() = Some(trace); - handles.set::>>>(name.to_owned(), trace_handle); -``` - -This looks like a bit of a mess, which is fair, but we are roughly wrapping up a trace handle so that it can be shared with others, and then registering it under `name.to_owned()` in the `handles` map. - -In fact, we stash a few other things in the map, which allows any program that knows what to look for to get access to shared state. For example, we stash the capability that the random graph uses to produce its changes, so that anyone could drop the capability and cause the graph generation to cease. \ No newline at end of file diff --git a/server/dataflows/degr_dist/Cargo.toml b/server/dataflows/degr_dist/Cargo.toml deleted file mode 100644 index e9dd8d9e1..000000000 --- a/server/dataflows/degr_dist/Cargo.toml +++ /dev/null @@ -1,14 +0,0 @@ -[package] -name = "degr_dist" -version = "0.1.0" -authors = ["Frank McSherry "] -edition.workspace = true -publish = false - -[dependencies] -timely = { workspace = true } -differential-dataflow = { workspace = true } -dd_server = { path = "../../" } - -[lib] -crate-type = ["dylib"] diff --git a/server/dataflows/degr_dist/src/lib.rs b/server/dataflows/degr_dist/src/lib.rs deleted file mode 100644 index 95d10bee1..000000000 --- a/server/dataflows/degr_dist/src/lib.rs +++ /dev/null @@ -1,26 +0,0 @@ -use std::rc::Rc; -use std::cell::RefCell; - -// use timely::dataflow::operators::{Probe, Operator}; -use differential_dataflow::operators::CountTotal; -use dd_server::{Environment, TraceHandle}; - -// load ./dataflows/degr_dist/target/release/libdegr_dist.dylib build - -#[no_mangle] -pub fn build((dataflow, handles, probe, _timer, args): Environment) -> Result<(), String> { - - if args.len() != 1 { return Err(format!("expected one argument, instead: {:?}", args)); } - - handles - .get_mut::>>>(&args[0])? - .borrow_mut().as_mut().unwrap() - .import(dataflow) - .as_collection(|&src,_dst| src) - .count_total() - .map(|(_deg, cnt)| cnt as usize) - .count_total() - .probe_with(probe); - - Ok(()) -} \ No newline at end of file diff --git a/server/dataflows/neighborhood/Cargo.toml b/server/dataflows/neighborhood/Cargo.toml deleted file mode 100644 index 3d850278c..000000000 --- a/server/dataflows/neighborhood/Cargo.toml +++ /dev/null @@ -1,13 +0,0 @@ -[package] -name = "neighborhood" -version = "0.1.0" -authors = ["Frank McSherry "] -edition.workspace = true -publish = false - -[dependencies] -differential-dataflow = { workspace = true } -dd_server = { path = "../../" } - -[lib] -crate-type = ["dylib"] diff --git a/server/dataflows/neighborhood/src/lib.rs b/server/dataflows/neighborhood/src/lib.rs deleted file mode 100644 index 8d5be369b..000000000 --- a/server/dataflows/neighborhood/src/lib.rs +++ /dev/null @@ -1,35 +0,0 @@ -use std::rc::Rc; -use std::cell::RefCell; - -use differential_dataflow::input::Input; - -use dd_server::{Environment, TraceHandle}; - -// load ./dataflows/neighborhood/target/release/libneighborhood.dylib build 0 - -#[no_mangle] -pub fn build((dataflow, handles, probe, _timer, args): Environment) -> Result<(), String> { - - if args.len() != 2 { return Err(format!("expected two arguments; instead: {:?}", args)); } - - let edges = handles - .get_mut::>>>(&args[0])? - .borrow_mut().as_mut().unwrap().import(dataflow); - - let source = args[1].parse::().map_err(|_| format!("parse error, source: {:?}", args[1]))?; - let (_input, query) = dataflow.new_collection_from(Some(source)); - - let timer = ::std::time::Instant::now(); - - query - .map(|x| (x, x)) - .join_core(edges.clone(), |_n, &q, &d| Some((d, q))) // one hop - .join_core(edges.clone(), |_n, &q, &d| Some((d, q))) // two hops - .join_core(edges, |_n, &q, &d| Some((d, q))) // three hops - .map(|x| x.1) - .consolidate() - .inspect(move |x| println!("{:?}:\t{:?}", timer.elapsed(), x)) - .probe_with(probe); - - Ok(()) -} \ No newline at end of file diff --git a/server/dataflows/random_graph/Cargo.toml b/server/dataflows/random_graph/Cargo.toml deleted file mode 100644 index c6053c3d1..000000000 --- a/server/dataflows/random_graph/Cargo.toml +++ /dev/null @@ -1,17 +0,0 @@ -[package] -name = "random_graph" -version = "0.1.0" -authors = ["Frank McSherry "] -edition.workspace = true -publish = false - -[dependencies] -timely = {workspace = true} -differential-dataflow = { workspace = true } -dd_server = { path = "../../" } -rand="0.3.13" - -hdrhist = { git = "https://github.com/utaal/hdrhist" } - -[lib] -crate-type = ["dylib"] diff --git a/server/dataflows/random_graph/src/lib.rs b/server/dataflows/random_graph/src/lib.rs deleted file mode 100644 index 971646c04..000000000 --- a/server/dataflows/random_graph/src/lib.rs +++ /dev/null @@ -1,196 +0,0 @@ -use std::rc::Rc; -use std::cell::RefCell; - -use rand::{Rng, SeedableRng, StdRng}; - - -use timely::dataflow::operators::Probe; -use timely::dataflow::operators::generic::operator::source; -use timely::progress::Antichain; - -use differential_dataflow::AsCollection; -use differential_dataflow::trace::TraceReader; - -use dd_server::{Environment, TraceHandle}; - -// load ./dataflows/random_graph/target/release/librandom_graph.dylib build 1000 2000 1000000 -// load ./dataflows/random_graph/target/release/librandom_graph.dylib build 10000000 100000000 1000000 -// drop -capability - -#[no_mangle] -pub fn build((dataflow, handles, probe, timer, args): Environment) -> Result<(), String> { - - // This call either starts the production of random graph edges. - // - // The arguments should be - // - // - // - // where is the target number of edge changes per second. The source - // will play out changes to keep up with this, and timestamp them as if they - // were emitted at the correct time. The timestamps use the system `timer`, - // but only start whenever the method is called. This means that the data are - // not deterministic, but if you subtract the elapsed time between system start - // up and method call, they should be deterministic. - // - // The method also registers a capability with name `-capability`, - // and will continue to execute until this capability is dropped from `handles`. - // To terminate the operator it is sufficient to drop the capability, as the - // operator holds only a weak reference to it. - // - // The operator also holds an `Weak>>` which it will - // attempt to borrow and call `set_logical_compaction` in order to advance the capability - // as it runs, to allow compaction and the maintenance of bounded state. - - if args.len() != 4 { return Err(format!("expected four arguments, instead: {:?}", args)); } - - let name = &args[0]; - let nodes: usize = args[1].parse().map_err(|_| format!("parse error, nodes: {:?}", args[1]))?; - let edges: usize = args[2].parse().map_err(|_| format!("parse error, edges: {:?}", args[2]))?; - let rate: usize = args[3].parse().map_err(|_| format!("parse error, rate: {:?}", args[3]))?; - - let requests_per_sec = rate; - let ns_per_request = 1000000000 / requests_per_sec; - - // shared capability keeps graph generation going. - let capability = Rc::new(RefCell::new(None)); - - // shared (optional) trace handle, so that the operator can advance capabilities. - let trace_handle: Rc>> = Rc::new(RefCell::new(None)); - let trace_handle_weak = Rc::downgrade(&trace_handle); - - let timer = timer.clone(); - - // create a trace from a source of random graph edges. - let mut trace = - source(dataflow, "RandomGraph", |cap, info| { - - let activator = dataflow.activator_for(info.address); - let mut hist = hdrhist::HDRHist::new(); - - let probe2 = probe.clone(); - - let index = dataflow.index(); - let peers = dataflow.peers(); - - // RNGs for edge addition and deletion. - let seed: &[_] = &[1, 2, 3, index]; - let mut rng1: StdRng = SeedableRng::from_seed(seed); - let mut rng2: StdRng = SeedableRng::from_seed(seed); - - // numbers of times we've stepped each RNG. - let mut additions = 0; - let mut deletions = 0; - - // record delay between system start-up and operator start-up. - let delay = timer.elapsed(); - let delay_ns = (delay.as_secs() as usize) * 1_000_000_000 + (delay.subsec_nanos() as usize); - - println!("{:?}: random graph generation started", delay); - - // stash capability in a rc::Weak. - *capability.borrow_mut() = Some(cap); - let capability = ::std::rc::Rc::downgrade(&capability); - - let mut recorded_ns = delay_ns; // don't worry about recording latencies before this. - let recording_step_ns = 1000; // record "per microsecond". - - let mut dirty = false; - - move |output| { - - activator.activate(); - - // Open-loop latency-throughput test, parameterized by offered rate `ns_per_request`. - let elapsed = timer.elapsed(); - let elapsed_ns = (elapsed.as_secs() as usize) * 1_000_000_000 + (elapsed.subsec_nanos() as usize); - - // Determine completed ns. - let acknowledged_ns: usize = probe2.with_frontier(|frontier| frontier[0]); - - if dirty { - if (recorded_ns >> 30) != (acknowledged_ns >> 30) { - println!("CCDF:"); - for (val,prob,_count) in hist.ccdf() { - println!("\t{}\t{}", val, prob); - } - } - } - - while recorded_ns < acknowledged_ns { - dirty = true; - hist.add_value((elapsed_ns - recorded_ns) as u64); - recorded_ns += recording_step_ns; - } - - // attempt to advance the frontier of the trace handle. - if let Some(trace_handle) = trace_handle_weak.upgrade() { - let mut borrow = trace_handle.borrow_mut(); - if let Some(ref mut trace_handle) = borrow.as_mut() { - trace_handle.set_logical_compaction(Antichain::from_elem(elapsed_ns).borrow()); - } - } - - // if our capability has not been cancelled ... - if let Some(capability) = capability.upgrade() { - - let mut borrow = capability.borrow_mut(); - let capability = borrow.as_mut().unwrap(); - // let mut time = capability.time().clone(); - - { // scope to allow session to drop, un-borrow. - let mut session = output.session(&capability); - - // load initial graph. - while additions < edges + deletions { - if additions % peers == index { - let src = rng1.gen_range(0, nodes); - let dst = rng1.gen_range(0, nodes); - session.give(((src, dst), 0, 1)); - } - additions += 1; - } - - // ship any scheduled edge additions. - while ns_per_request * (additions - edges) < (elapsed_ns - delay_ns) { - if additions % peers == index { - let time = delay_ns + ns_per_request * (additions - edges); - let src = rng1.gen_range(0, nodes); - let dst = rng1.gen_range(0, nodes); - session.give(((src, dst), time, 1)); - } - additions += 1; - } - - // ship any scheduled edge deletions. - while ns_per_request * deletions < (elapsed_ns - delay_ns) { - if deletions % peers == index { - let time = delay_ns + ns_per_request * deletions; - let src = rng2.gen_range(0, nodes); - let dst = rng2.gen_range(0, nodes); - session.give(((src, dst), time, -1)); - } - deletions += 1; - } - } - - capability.downgrade(&elapsed_ns); - } - } - }) - .probe_with(probe) - .as_collection() - .arrange_by_key() - .trace; - - // release all blocks on merging. - trace.set_physical_compaction(Antichain::new().borrow()); - *trace_handle.borrow_mut() = Some(trace); - - handles.set::>>>(name.to_owned(), trace_handle); - handles.set(format!("{}-capability", name), capability); - - println!("handles set"); - - Ok(()) -} \ No newline at end of file diff --git a/server/dataflows/reachability/Cargo.toml b/server/dataflows/reachability/Cargo.toml deleted file mode 100644 index d97567389..000000000 --- a/server/dataflows/reachability/Cargo.toml +++ /dev/null @@ -1,13 +0,0 @@ -[package] -name = "reachability" -version = "0.1.0" -authors = ["Frank McSherry "] -edition.workspace = true -publish = false - -[dependencies] -differential-dataflow = { workspace = true } -dd_server = { path = "../../" } - -[lib] -crate-type = ["dylib"] diff --git a/server/dataflows/reachability/src/lib.rs b/server/dataflows/reachability/src/lib.rs deleted file mode 100644 index 6ad5e9d16..000000000 --- a/server/dataflows/reachability/src/lib.rs +++ /dev/null @@ -1,33 +0,0 @@ -use std::rc::Rc; -use std::cell::RefCell; - -use differential_dataflow::input::Input; -use differential_dataflow::operators::Iterate; - -use dd_server::{Environment, TraceHandle}; - -#[no_mangle] -pub fn build((dataflow, handles, probe, _timer, args): Environment) -> Result<(), String> { - - if args.len() != 2 { return Err(format!("expected two arguments; instead: {:?}", args)); } - - let edges = handles - .get_mut::>>>(&args[0])? - .borrow_mut().as_mut().unwrap().import(dataflow); - - let source = args[1].parse::().map_err(|_| format!("parse error, source: {:?}", args[1]))?; - let (_input, roots) = dataflow.new_collection_from(Some(source)); - - // repeatedly update minimal distances each node can be reached from each root - roots.clone().iterate(|scope, dists| { - let edges = edges.enter(scope); - let roots = roots.enter(scope); - dists.arrange_by_self() - .join_core(edges, |_src, _, &dst| Some(dst)) - .concat(roots) - .distinct() - }) - .probe_with(probe); - - Ok(()) -} \ No newline at end of file diff --git a/server/src/bin/dd_server.rs b/server/src/bin/dd_server.rs deleted file mode 100644 index 8fa7bd269..000000000 --- a/server/src/bin/dd_server.rs +++ /dev/null @@ -1,140 +0,0 @@ -use std::io::BufRead; -use std::io::Write; - -use std::sync::{Arc, Mutex}; - -use timely::synchronization::Sequencer; - -use libloading::{Library, Symbol}; - -use dd_server::{Environment, TraceHandler}; - -fn main() { - - // shared queue of commands to serialize (in the "put in an order" sense). - let (send, recv) = std::sync::mpsc::channel::>(); - let recv = Arc::new(Mutex::new(recv)); - - // demonstrate dynamic loading of dataflows via shared libraries. - let guards = timely::execute_from_args(std::env::args(), move |worker| { - - let timer = ::std::time::Instant::now(); - let recv = recv.clone(); - - // map from string name to arranged graph. - let mut handles = TraceHandler::new(); - - // common probe used by all dataflows to express progress information. - let mut probe = timely::dataflow::operators::probe::Handle::new(); - - // queue shared between serializer (producer) and command loop (consumer). - let mut sequencer = Sequencer::new(worker, timer); - - let mut done = false; - while !done { - - // Check out channel status. - let lock = recv.lock().expect("Mutex poisoned"); - use std::sync::mpsc::TryRecvError; - match lock.try_recv() { - Ok(command) => { sequencer.push(command); }, - Err(TryRecvError::Empty) => { }, - Err(TryRecvError::Disconnected) => { done = true; }, - }; - - // Dequeue and act on commands. - while let Some(mut command) = sequencer.next() { - - let index = worker.index(); - println!("worker {:?}: received command: {:?}", index, command); - - if command.len() > 1 { - let operation = command.remove(0); - match operation.as_str() { - "list" => { - println!("worker {:?} listing", index); - for key in handles.keys() { - println!("worker {:?} list: {:?}", index, key); - } - } - "load" => { - - if command.len() >= 2 { - - let library_path = &command[0]; - let symbol_name = &command[1]; - - if let Ok(lib) = unsafe { Library::new(library_path) } { - worker.dataflow_core("dataflow", None, lib, |lib, child| { - let result = unsafe { - lib.get::Result<(),String>>>(symbol_name.as_bytes()) - .map(|func| func((child, &mut handles, &mut probe, &timer, &command[2..]))) - }; - - match result { - Err(_) => { println!("worker {:?}: failed to find symbol {:?} in shared library {:?}.", index, symbol_name, library_path); }, - Ok(Err(x)) => { println!("worker {:?}: error: {:?}", index, x); }, - Ok(Ok(())) => { /* Good news, everyone! */ }, - } - }); - } - else { - println!("worker {:?}: failed to open shared library: {:?}", index, library_path); - } - } - }, - "drop" => { - for name in command.iter() { - handles.remove(name); - } - } - _ => { - println!("worker {:?}: unrecognized command: {:?}", index, operation); - } - } - } - - // arguably we should pick a time (now) and `step_while` until it has passed. - // this should ensure that we actually fully drain ranges of updates, rather - // than providing no guaranteed progress for e.g. iterative computations. - - worker.step(); - } - } - - println!("worker {}: command queue unavailable; exiting command loop.", worker.index()); - }); - - // the main thread now continues, to read from the console and issue work to the shared queue. - - std::io::stdout().flush().unwrap(); - let input = std::io::stdin(); - - let mut done = false; - - while !done { - - if let Some(line) = input.lock().lines().map(|x| x.unwrap()).next() { - let elts: Vec<_> = line.split_whitespace().map(|x| x.to_owned()).collect(); - - if elts.len() > 0 { - match elts[0].as_str() { - "help" => { println!("valid commands are currently: bind, drop, exit, help, list, load"); }, - "bind" => { println!("ideally this would load and bind a library to some delightful name"); }, - "drop" => { send.send(elts).expect("failed to send command"); } - "exit" => { done = true; }, - "load" => { send.send(elts).expect("failed to send command"); }, - "list" => { send.send(elts).expect("failed to send command"); }, - _ => { println!("unrecognized command: {:?}", elts[0]); }, - } - } - - std::io::stdout().flush().unwrap(); - } - } - - println!("main: exited command loop"); - drop(send); - - guards.unwrap(); -} diff --git a/server/src/lib.rs b/server/src/lib.rs deleted file mode 100644 index cdf674b68..000000000 --- a/server/src/lib.rs +++ /dev/null @@ -1,94 +0,0 @@ -use std::any::Any; -use std::rc::Rc; -use std::collections::HashMap; -use std::time::Instant; - -use libloading::Library; - -use timely::dataflow::Scope; -use timely::dataflow::operators::probe::Handle as ProbeHandle; - -// stuff for talking about shared trace types ... -use differential_dataflow::operators::arrange::TraceAgent; -use differential_dataflow::trace::implementations::ValSpine; - -// These are all defined here so that users can be assured a common layout. -pub type RootTime = usize; -type TraceSpine = ValSpine; -pub type TraceHandle = TraceAgent; - -/// Arguments provided to each shared library to help build their dataflows and register their results. -pub type Environment<'a, 'b> = ( - Scope<'b, usize>, - &'a mut TraceHandler, - &'a mut ProbeHandle, - &'a Instant, - &'a [String] -); - -/// A wrapper around types that keep their source libraries alive. -/// -/// This type is meant to be a smart pointer for a type `T` that needs to keep -/// a `Library` alive, perhaps because its methods would call in to the library. -/// The type should have a specified drop order (viz RFC 1857) which guarantees -/// that the shared library reference drops only after the element itself is -/// dropped. It also implements `Deref` and `DerefMut` to provide the experience -/// of a `T` itself. -/// -/// Nothing about this type actually *guarantees* that the code for type `T` is -/// found in the wrapped library, and it is up to the user to wrap the correct -/// libraries here. -pub struct LibraryWrapper { - /// The wrapped element. - element: T, - /// An unused field used to keep the reference count alive. - _library: Rc, -} - -impl LibraryWrapper { - /// Creates a new instance of LibraryWrapper. - pub fn new(element: T, library: &Rc) -> Self { - LibraryWrapper { - element: element, - _library: library.clone(), - } - } -} - -impl ::std::ops::Deref for LibraryWrapper { - type Target = T; - fn deref(&self) -> &T { &self.element } -} - -impl ::std::ops::DerefMut for LibraryWrapper { - fn deref_mut(&mut self) -> &mut T { &mut self.element } -} - -/// A wrapper around a `HashMap>` that handles downcasting. -pub struct TraceHandler { - handles: HashMap>, -} - -impl TraceHandler { - /// Create a new trace handler. - pub fn new() -> Self { TraceHandler { handles: HashMap::new() } } - /// Acquire a mutable borrow of the value for `name`, if it is of type `T`. - pub fn get_mut<'a, T: Any>(&'a mut self, name: &str) -> Result<&'a mut T, String> { - let boxed = self.handles.get_mut(name).ok_or(format!("failed to find handle: {:?}", name))?; - boxed.downcast_mut::().ok_or(format!("failed to downcast: {}", name)) - } - /// Enumerates the keys maintained in storage (for the `list` operation). - pub fn keys(&self) -> ::std::collections::hash_map::Keys<'_, String, Box> { - self.handles.keys() - } - /// Assign a thing to key `name`, boxed as `Box`. - pub fn set(&mut self, name: String, thing: T) { - let boxed: Box = Box::new(thing); - assert!(boxed.downcast_ref::().is_some()); - self.handles.insert(name, boxed); - } - /// Removes the resource associated with `name`. - pub fn remove(&mut self, name: &str) -> Option> { - self.handles.remove(name) - } -}