summaryrefslogtreecommitdiff
path: root/src/sync
diff options
context:
space:
mode:
Diffstat (limited to 'src/sync')
-rw-r--r--src/sync/barrier.rs130
-rw-r--r--src/sync/mod.rs44
-rw-r--r--src/sync/waiter.rs13
3 files changed, 187 insertions, 0 deletions
diff --git a/src/sync/barrier.rs b/src/sync/barrier.rs
new file mode 100644
index 0000000..7e53b12
--- /dev/null
+++ b/src/sync/barrier.rs
@@ -0,0 +1,130 @@
+use super::waiter::Waiter;
+use futures::sync::mpsc;
+use futures::sync::oneshot;
+use futures::prelude::*;
+
+pub trait BarrierSync<W: Waiter> {
+ fn wait_for(&mut self, waiter: W) -> Box<Future<Item=W::Item, Error=W::Error> + Send>;
+}
+
+pub struct Barrier<W: Waiter> {
+ sink: Option<mpsc::Sender<(W, oneshot::Sender<Result<W::Item, W::Error>>)>>,
+}
+
+impl<W: Waiter + 'static + Send> BarrierSync<W> for Barrier<W> {
+ fn wait_for(&mut self, waiter: W) -> Box<Future<Item=W::Item, Error=W::Error> + Send> {
+ let (resp_tx, resp_rx) = oneshot::channel();
+
+ if self.sink.is_none() {
+ let (barrier_tx, barrier_rx) = mpsc::channel(40);
+ self.barrier_task(barrier_rx);
+ self.sink.replace(barrier_tx);
+ }
+
+ let chan = self.sink.as_mut().unwrap().clone();
+
+ /*TODO: I want meaningful error types... */
+ let f1 = chan
+ .send((waiter, resp_tx))
+ .map_err(|err| W::Error::from(()))
+ .and_then(|_| {
+ resp_rx.then(|result| {
+ match result {
+ Ok(Ok(result)) => Ok(result),
+ Ok(Err(err)) => Err(err),
+ Err(err) => Err(W::Error::from(())),
+ }
+ })
+ });
+
+ Box::new(f1)
+ }
+}
+
+impl<W: Waiter + 'static + Send> Barrier<W> {
+ pub fn new() -> Barrier<W> {
+ Barrier {
+ sink: None,
+ }
+ }
+
+ fn barrier_task(&self, receiver: mpsc::Receiver<(W, oneshot::Sender<Result<W::Item, W::Error>>)>) {
+
+ enum Message<W: Waiter> {
+ Request((W, oneshot::Sender<Result<<W as Waiter>::Item, <W as Waiter>::Error>>)),
+ OnCondition(Result<(), <W as Waiter>::ConditionError>),
+ }
+
+ let mut polling = false;
+ let (on_condition_tx, on_condition_rx) = mpsc::unbounded();
+ let mut waiters = Vec::new();
+ let f1 = receiver.map(|request| Message::Request(request));
+ let f2 = on_condition_rx.map(|result| Message::OnCondition(result));
+
+ let inner_condition = on_condition_tx.clone();
+ let f =
+ f1.select(f2).for_each(move |message| {
+ match message {
+ Message::Request((waiter, backchan)) => {
+ if waiter.blocked() && !polling {
+ println!("locked");
+
+ let c1 = inner_condition.clone();
+ let f = waiter
+ .condition_poller()
+ .map(|_| ())
+ .then(|result| {
+ c1.send(result).wait();
+ Ok(())
+ });
+ tokio::spawn(f);
+ polling = true;
+
+ waiters.push((waiter, backchan));
+ } else if waiter.blocked() || polling {
+ println!("polling");
+ waiters.push((waiter, backchan));
+ } else {
+ println!("Pass along waiter!");
+ let f = waiter.into_future()
+ .then(|res| {
+ backchan.send(res);
+ Ok(())
+ });
+
+ tokio::spawn(f);
+ }
+ },
+ Message::OnCondition(result) => {
+ polling = false;
+ /*Resubmit all waiters back to the request channel
+ * At least one waiter will pass the barrier
+ */
+ match result {
+ Ok(_) => {
+ while waiters.len() > 0 {
+ let (waiter, backchan) = waiters.pop().unwrap();
+ let f = waiter.into_future()
+ .then(|res| {
+ backchan.send(res);
+ Ok(())
+ });
+
+ tokio::spawn(f);
+ }
+ },
+ _ => { panic!("condition channel closed") }
+ }
+ }
+ }
+
+
+
+ Ok(())
+ })
+ .map(|_| ())
+ .map_err(|_| ());
+
+ tokio::spawn(f);
+ }
+}
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
new file mode 100644
index 0000000..ca06d32
--- /dev/null
+++ b/src/sync/mod.rs
@@ -0,0 +1,44 @@
+//f.barrier(auth).barrier(ratelimit).and_then(|result| {})
+//A ratelimiter must be aware when a limit is hit, the upper limit,
+//and remaining requests. (use case specific)
+//
+//This can be done by either letting the ratelimiter drive the request
+//so it can inspect returned headers or by maybe? using a channel to inform
+//the limiter
+//
+//Submit task to ratelimiter.
+//Check if the limit is hit and if we are polling
+// 1 if we hit the limit and are not polling, add to the queue and start
+// polling.
+// 2. if we are polling add the request to the queue
+// 3. if we are not polling and not locked then
+// send the request and increment the in-flight counter.
+//
+// when the request has completed without errors then decrement
+// the in-flight counter, update limiter data, and return the
+// result to the requester.
+//
+// On error, EITHER:
+// 1. If the error is rate limiter related place the request
+// back in a queue, return other errors. (Prevents starvation)
+// 2. Return all errors back to the Requester they can resubmit
+// the request
+//
+// The main difference is that the condition is dependent on the waiter's
+// future result.
+//
+// For auth requests we can use an OkFuture that returns the waiter and never errs
+//
+// So waiters must provide IntoFuture, a future than can poll the condition,
+// and a is locked.
+// The lock check must be pure (no side effects) but IntoFuture may
+// have side effects (eg. increments in-flight counter)
+//
+// The result of the IntoFuture is returned to caller or the Err of the poll
+// Future. For simplicity these will be the same type.
+//
+// Should the poll condition trait be located on the Waiter or the Barrier?
+// All waiters in a barrier must use the same condition.
+
+pub mod barrier;
+pub mod waiter;
diff --git a/src/sync/waiter.rs b/src/sync/waiter.rs
new file mode 100644
index 0000000..656c42e
--- /dev/null
+++ b/src/sync/waiter.rs
@@ -0,0 +1,13 @@
+use futures::sync::oneshot;
+use futures::Future;
+
+pub trait Waiter {
+ type Item: Send + 'static;
+ type Error: From<Self::ConditionError>
+ + From<oneshot::Canceled> + From<()> + Send + 'static;
+ type ConditionError: Send + Clone + 'static;
+
+ fn blocked(&self) -> bool;
+ fn condition_poller(&self) -> Box<Future<Item=(), Error=Self::ConditionError> + Send>;
+ fn into_future(self) -> Box<Future<Item=Self::Item, Error=Self::Error> + Send>;
+}