Better way to drive Service than poll_ready #110
Comments
|
In theory, I really want something like this as it would reduce the required overhead of using the Tower Service abstraction. That said, I thought more about it, and it is a bit more complicated... The "drive" function would have to do two things:
The second point also implies that there needs to be some way to tell the service to start shutting down. So, at the very least, there would need to be two additional functions:
(Note that This adds non trivial complexity to using a This implies that there would need to be a separate trait. How would that work? |
|
I don't really have an answer, just more observations about why the current situation isn't ideal: if you do work to "drive" the returned future(s) in |
|
The other advantage to have |
|
If you stash them in FuturesUnordered how do you call them? |
|
|
|
Bah, |
|
it would be possible to impl a similar structure that does enable lookup though... |
|
I actually think modelling |
|
For a more concrete example of where this applies, see: https://github.com/jonhoo/tokio-tower-wip/blob/861f273abea1682d1be90959c34a54d83f291b86/tests/pipeline/client.rs#L40 |
|
Some further discussion happened on Gitter. A brief-ish summary:
It seems like we need to figure out what methods trait Service {
/// Called to make room for another request.
///
/// Should call `poll_outstanding` as necessary to finish in-flight requests.
///
/// The default implementation always returns `Ready`, which must be safe because `call`
/// (per spec) must always be prepared to be called, even if not ready.
///
/// Implementors are encouraged to implement this if `call` can ever fail.
default fn poll_ready(&mut self) -> Result<Async<()>, Self::Error> {
Ok(Async::Ready(())
}
/// Called to make progress on in-flight requests.
///
/// Should return `Ready` when all outstanding requests have been serviced.
fn poll_outstanding(&mut self) -> Result<Async<()>, Self::Error>;
/// Called when there will be no more calls to `call` or `on_ready`.
///
/// This method will be called at most once.
/// The default implementation does nothing.
default fn on_shutdown(&mut self) { }
/// Called when there will be no more calls to `call`.
///
/// The returned `Future` resolves when the shutdown has been completed,
/// and all outstanding requests have been responded to (i.e., when
/// `poll_outstanding` next returns `Ready`.
///
/// This will implicitly call `on_shutdown` on the `Service` before polling
/// outstanding requests.
fn shutdown(self) -> ShutdownFuture<Self> { ... }
/// ...
fn call(&mut self, req: Self::Request) -> Self::Future;
/// Returns a `Future` that resolves when the given future resolves, and
/// internally calls [`poll_outstanding`] to drive this `Service` forward.
///
/// Akin to `future::select(fut, self)`.
fn finish<F>(self, fut: F) -> OutstandingPoller<Self, F> { ... }
/// Returns a `Future` that resolves the first time `call(req)` completes
/// after `poll_ready` has returned `Ready`. It produces `Self` and the
/// `Future` returned from `call`.
fn enqueue(self, req: Self::Request) -> ReadyCallPoller<Self> { ... }
} |
|
Following some further discussion on making the trait a bit more like trait Service {
/// Called to make room for another request.
///
/// Should call `poll_outstanding` as necessary to finish in-flight requests.
///
/// The default implementation always returns `Ready`, which must be safe because `call`
/// (per spec) must always be prepared to be called, even if not ready.
///
/// Implementors are encouraged to implement this if `call` can ever fail.
default fn poll_ready(&mut self) -> Result<Async<()>, Self::Error> {
Ok(Async::Ready(())
}
/// Called to make progress on in-flight requests.
///
/// Should return `Ready` when all outstanding requests have been serviced.
fn poll_outstanding(&mut self) -> Result<Async<()>, Self::Error>;
/// Called after there will be no more calls to `call`.
///
/// `poll_close` should ensure that all in-progress requests resolve, as well
/// as perform and finish any required service cleanup.
default fn poll_close(&mut self) -> Result<Async<()>, Self::Error> { self.poll_outstanding() }
/// ...
fn call(&mut self, req: Self::Request) -> Self::Future;
}
trait ServiceExt: Service {
/// Returns a `Future` that resolves when the given future resolves, and
/// internally calls [`poll_outstanding`] to drive this `Service` forward.
///
/// Akin to `future::select(fut, self)`.
fn finish<F>(self, fut: F) -> OutstandingPoller<Self, F> { ... }
/// Returns a `Future` that resolves the first time `call(req)` completes
/// after `poll_ready` has returned `Ready`. It produces `Self` and the
/// `Future` returned from `call`.
fn enqueue(self, req: Self::Request) -> ReadyCallPoller<Self> { ... }
} |
|
I may be missing something obvious, but I don't understand why there would be |
|
@seanmonstar the observation is that for some |
|
@seanmonstar Also, there are cases where the service needs to be driven when there are no futures (think h2's ping / pong). |
|
@jonhoo For clarity, could you move I'm also leaning towards not having any default implementations for functions. I think that the main implementors of For the simple case, we can defer to let my_service = service_fn(|request| {
// do something with request
Ok(response);
}); |
|
Also, we may want to bikeshed |
|
We could just call it |
|
Much more discussion on Gitter today, culminating in this question (all chat history edited for clarity):
Note also that we should not have
There's also the worry that we now need to have lots of things be able to take both
It seems like we're basically agreed that
So, we would have: trait Service<Request> {
/// Like today's Service::poll_ready.
/// Should this even be present on `Service`, or just on `DirectService`?
fn poll_ready(&mut self) -> Result<Async<()>, Self::Error>;
/// Like today's Service::call
fn call(&mut self, req: Request) -> Self::Future;
}
trait ServiceExt<Request>: Service<Request> {
/// Returns a `Future` that resolves the first time `call(req)` completes
/// after `poll_ready` has returned `Ready`. It produces `Self` and the
/// `Future` returned from `call`.
fn enqueue(self, req: Request) -> ReadyCallPoller<Self> { ... }
}
trait DirectService<Request> {
/// Called to make room for another request.
///
/// Should call `poll_outstanding` as necessary to finish in-flight requests.
fn poll_ready(&mut self) -> Result<Async<()>, Self::Error>;
/// Called to make progress on in-flight requests.
///
/// Should return `Ready` when all outstanding requests have been serviced.
fn poll_outstanding(&mut self) -> Result<Async<()>, Self::Error>;
/// Called after there will be no more calls to `call`.
///
/// `poll_close` should ensure that all in-progress requests resolve, as well
/// as perform and finish any required service cleanup.
default fn poll_close(&mut self) -> Result<Async<()>, Self::Error> { self.poll_outstanding() }
/// Like today's Service::call, but with the caveat that poll_outstanding must
/// continue to be called for the returned futures to resolve.
fn call(&mut self, req: Request) -> Self::Future;
}
We may also want an (internal) |
|
We'll probably also want to be very clear in the docs for |
This patch adds the `DirectService` trait, and related implementations
over it in `tower_balance` and `tower_buffer`. `DirectService` is
similar to a `Service`, but must be "driven" through calls to
`poll_service` for the futures returned by `call` to make progress.
The motivation behind adding this trait is that many current `Service`
implementations spawn long-running futures when the service is created,
which then drive the work necessary to turn requests into responses. A
simple example of this is a service that writes requests over a
`TcpStream` and reads responses over that same `TcpStream`. The
underlying stream must be read from to discover new responses, but there
is no single entity to drive that task. The returned futures would share
access to the stream (and worse yet, may get responses out of order),
and then service itself is not guaranteed to see any more calls to it as
the client is waiting for its requests to finish.
`DirectService` solves this by introducing a new method, `poll_service`,
which must be called to make progress on in-progress futures.
Furthermore, like `Future::poll`, `poll_service` must be called whenever
the associated task is notified so that the service can also respect
time-based operations like heartbeats.
The PR includes changes to both `tower_balance::Balance` and
`tower_buffer::Buffer` to add support for wrapping `DirectService`s. For
`Balance` this is straightforward: if the inner service is a `Service`,
the `Balance` also implements `Service`; if the inner service is a
`DirectService`, the `Balance` is itself also a `DirectService`. For
`Buffer`, this is more involved, as a `Buffer` turns any `DirectService`
*into* a `Service`. The `Buffer`'s `Worker` is spawned, and will
therefore drive the wrapped `DirectService`.
One complication arises in that `Buffer<T>` requires that `T: Service`,
but you can safely construct a `Buffer` over a `DirectService` per the
above. `Buffer` works around this by exposing
```rust
impl Service for HandleTo<S> where S: DirectService {}
```
And giving out `Buffer<HandleTo<S>>` when the `new_directed(s: S)`
constructor is invoked. Since `Buffer` never calls any methods on the
service it wraps, `HandleTo`'s implementation just consists of calls to
`unreachable!()`.
Note that `tower_buffer` now also includes a `DirectedService` type,
which is a wrapper around a `Service` that implements `DirectService`.
In theory, we could do away with this by adding a blanket impl:
```rust
impl<T> DirectedService for T where T: Service {}
```
but until we have specialization, this would prevent downstream users
from implementing `DirectService` themselves.
Fixes tower-rs#110.
This patch adds the `DirectService` trait, and related implementations
over it in `tower_balance` and `tower_buffer`. `DirectService` is
similar to a `Service`, but must be "driven" through calls to
`poll_service` for the futures returned by `call` to make progress.
The motivation behind adding this trait is that many current `Service`
implementations spawn long-running futures when the service is created,
which then drive the work necessary to turn requests into responses. A
simple example of this is a service that writes requests over a
`TcpStream` and reads responses over that same `TcpStream`. The
underlying stream must be read from to discover new responses, but there
is no single entity to drive that task. The returned futures would share
access to the stream (and worse yet, may get responses out of order),
and then service itself is not guaranteed to see any more calls to it as
the client is waiting for its requests to finish.
`DirectService` solves this by introducing a new method, `poll_service`,
which must be called to make progress on in-progress futures.
Furthermore, like `Future::poll`, `poll_service` must be called whenever
the associated task is notified so that the service can also respect
time-based operations like heartbeats.
The PR includes changes to both `tower_balance::Balance` and
`tower_buffer::Buffer` to add support for wrapping `DirectService`s. For
`Balance` this is straightforward: if the inner service is a `Service`,
the `Balance` also implements `Service`; if the inner service is a
`DirectService`, the `Balance` is itself also a `DirectService`. For
`Buffer`, this is more involved, as a `Buffer` turns any `DirectService`
*into* a `Service`. The `Buffer`'s `Worker` is spawned, and will
therefore drive the wrapped `DirectService`.
One complication arises in that `Buffer<T>` requires that `T: Service`,
but you can safely construct a `Buffer` over a `DirectService` per the
above. `Buffer` works around this by exposing
```rust
impl Service for HandleTo<S> where S: DirectService {}
```
And giving out `Buffer<HandleTo<S>>` when the `new_directed(s: S)`
constructor is invoked. Since `Buffer` never calls any methods on the
service it wraps, `HandleTo`'s implementation just consists of calls to
`unreachable!()`.
Note that `tower_buffer` now also includes a `DirectedService` type,
which is a wrapper around a `Service` that implements `DirectService`.
In theory, we could do away with this by adding a blanket impl:
```rust
impl<T> DirectedService for T where T: Service {}
```
but until we have specialization, this would prevent downstream users
from implementing `DirectService` themselves.
Finally, this also makes `Buffer` use a bounded mpsc channel, which
introduces a new capacity argument to `Buffer::new`.
Fixes tower-rs#110.
I'm looking to implement
tokio-tower::pipeline::Client, and as part of that I need to be able to "drive" theServiceimplementation forward to send pending requests, publish responses etc. Currently, the only way to do that is to do all the work inpoll_ready. While that works just fine, it feels like a bit of an abuse of the API. @carllerche mentioned here that we might want aService::drivemethod where we do this kind of heavy-lifting.The text was updated successfully, but these errors were encountered: