diff options
Diffstat (limited to 'src/sync')
| -rw-r--r-- | src/sync/barrier.rs | 130 | ||||
| -rw-r--r-- | src/sync/mod.rs | 44 | ||||
| -rw-r--r-- | src/sync/waiter.rs | 13 |
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>; +} |