I/O completion based operations and cancellation

[Matthias247]

TLDR:

• I/O completion based operations might not be ideally modellable through futures.rs, since it requires all operations to be synchronously cancellable via Drop.
• An alternative approach could be to require all Futures to be polled to completion, and add cancellation signalization via another channel than Drop.

Longer version:

I spent some time during the last days to think about whether and how I/O completion based operations can be efficiently modeled with futures and async/await.

With I/O completion based operations, I'm e.g. referring to APIs like:

• Windows IOCP
• Linux io_submit
• Boost asio like operations
• Other higher level networking APIs, that allow to enqueue an operation, and allow to asynchronously wait for completion

What these operations have in common:

• When the operation is started, the operation and I/O buffers are provided to the IO subsytem.
• The application has to guarantee that buffers don't get modified during the IO operation
• As soon as the IO subsystem notifies the operation that the operation is complete (e.g. via a IO completion queue) the application can reuse buffers.
• Cancellation is often possible via a dedicated cancellation API. However it might not be synchronous, and the results of cancelled operations might still end up in the completion queue.

My current thought is that the support of futures.rs for these kinds of operations might not be as good as it could be:

With approaches as described in #1272 it should e.g. be possible to store I/O operation buffers inside Future implementations for the whole duration of the async operation, without passing the ownership of those buffers to the sink. However this isn't possible if the underlying API does not support synchronous cancellation. The operation (e.g. IOCP) might still refer to the data after the Future is dropped, up until the point where we get the confirmation through the completion queue. However blocking on a completion queue inside a constructor is not advisable for an asynchronous system.

The workaround is to hand over ownership of all the buffers to the implementor of the future - most likely in a refcounted fashion, so that the data still can be safely referred after the Future is destructed. This approach can e.g. be seen in traits like Sink. However the drawback here is that

• the operation has additional costs, e.g. a refcounted state needs to be allocated on the heap. The state can't be part of the state future anymore.
• if the operation is cancelled (via drop) the transferred item is completely lost and can no longer be referred to.

In the current state of the ecosystem patterns like passing refcounted owned refcounted byte buffers had been established, and there is a preference for readiness based APIs instead of completion based ones (e.g. in the Sink trait). However for more resource constrained systems (embedded) and for some kinds of API types these things might not be preferable. Lets e.g. pretend I want to build a minimal resource using Websocket client, where client.send(message) returns a Future that represents the transfer. Since the websocket framing may not be corrupted, we can not stop sending the message at arbitrary points (e.g. when only half of it's bytes had been written). So we would need to provide the client a copy of the full message, and not only provide it a reference to it. We also might not be able to go further and e.g. use the send futures state as an intrusive operation, which is queued inside the Websocket client, as thought about in #1272 .

I would be interested to learn whether others have also already thought about the problem, and whether there are other good ways to model these kinds of operations in a zero-overhead fashion.

I came up with a modification to Future semantics, which should be able to allow us operating on borrowed data:

• Futures would always need to be polled until completion. Where completion is when the I/O subsystem has signalled that the resources of the future are clearly no longer needed.
• Dropping a future should in this approach only be valid if it either has never been polled, or has been polled to completion.
• Since dropping a future can no longer be used for completion, another approach to signal cancellation to operations would be required. That could be e.g. another parameter that is passed to Future creating methods, like a CancellationToken, as pioneered in other languages. The parameter could obviously be either part of a standard library, or non-standardized (like there is no standard cancellation defined for Javascript Promises, but library authors could still add extra facilities).

There is definitely a lot of precedent for APIs which differentiate between signalling cancellation and waiting for it to complete, e.g. Go's Context APIs, .NET Task cancellation facilities, Kotlin coroutines. It also matches well to the described C APIs. So this seems to make sense.
Always waiting for things to complete also seems to match better to the "structured concurrency" methodology which currently seems to get a lot of positive tailwind.

However a change in semantics like this is obviously very invasive. It would not only require leaf Futures to be modified to, but also would require all combinators to be adopted (they can't drop any non-completed path anymore). So this would need to be carefully evaluated. If a model like this is chosen, it would be great if Futures in the non-completed state can't even be dropped by users to avoid possible usage errors and leaks/corruptions. Maybe through something like #[must_use] in the non-completed state.

[Thomasdezeeuw]

I'm not sure I understand the exact problem. I'm not familiar with Windows IOCP, but for Linux aio I don't see any major problems. However it must be supported by the futures' executor (so many this isn't an issue for the futures crate).

Take aio_write for example, if it owns both the File (or rather the file descriptor) and the buffer then I don't forsee any problems. This does require the future to be pinned, which is a requirement of the Future trait anyway. Cancelling can be done via aio_cancel in Drop, I don't see any problems here either. The future itself could just include the buffer and/or file in the return value. It wouldn't make for the greatest API, but I think it would work.

P.S. I could be missing something here, please let me know.

[Nemo157]

Here's some docs with an example of cancelling async IO on windows. They block and wait for the cancellation to complete before continuing.

I wonder if it's easy enough for the IO handle to record that it has requested cancellation, then handle waiting for it to complete the next time an IO operation happens. It could cause issues if there were errors cancelling, but maybe you don't care about those?

[Thomasdezeeuw]

If Windows functions are blocking can't I/O pools be used? Much like tokio-fs uses it for blocking I/O operations on files? If so it would make it an issue for the futures executor.

As for the handling of errors when cancelling we could follow the standard library when closing a file, ignore it. It's always a race when cancelling something, I wouldn't know what else could be done at that point (other then informing the caller).

[Nemo157]

If Windows functions are blocking can't I/O pools be used?

It's not that the cancellation itself is blocking, the cancellation returns an async completion and they block in the example waiting for it to complete. An I/O pool can't be used because Drop is synchronous, I/O pools can make synchronous operations look async, but not the inverse.

As for the handling of errors when cancelling we could follow the standard library when closing a file, ignore it. It's always a race when cancelling something, I wouldn't know what else could be done at that point (other then informing the caller).

This is just cancelling a single operation though, the I/O handle is still open and can be re-used for another operation e.g.

async {
    let mut reader: impl AsyncRead = { ... };
    let first: Either<io::Result<[u8; 32]>, Timeout>
        = await!(reader.read_exact([0; 32]).timeout(Duration::from_secs(1)));
    let second: io::Result<[u8; 32]>
        = await!(reader.read_exact([0; 32]));
}

If the first operation times out and is synchronously cancelled by the timeout adaptor, but then asynchronously errors, what happens for the second read call?

[Matthias247]

IOCP cancellation and Linux io_cancel (or also posix aio_cancel) work similar: They may cancel the operation synchronously, but in some situations (operation has already been started, I/O resource or driver doesn't support it, etc) it might not be possible. In that worst-case one would always need to block synchronously for the operation to complete if that work would need to be done in drop. The only different here between IOCP and io_cancel is that IOCP will always deliver an event in the completion queue, whether cancellation was successful or not.

So the drop method would look somewhat like this:

impl Drop for IocpFuture {
    fn drop(&mut self) {
        CancelIo(self.handle);
        // Block here until the completion queue signaled the real completion
        // Since the completion queue might be handled by an IOCP reactor thread,
        // we must synchronize here somewhat with it.
        self.wait_handle.wait(); // Assuming the IO reactor signals the wait_handle
        // Cleanup resources
    }
}

Since the cancel operation is best effort, blocking here might take a long time, and thereby might block the futures executor, which is definitely not intended.

The completion based networking API on the higher level might be even harder to solve: We shouldn't be able to cancel while the message/packet is half-transmitted/received, since that would corrupt the state of the messaging channel. So we would need to synchronously block here until the transmitting component has fully completed the operation. Which is not the natural thing, if the component (e.g. Websocket client, MQTT client, etc) is working asynchronously on top of futures too. With the other approach we would signal cancellation to the client, and perform the normal asynchronous wait / poll to wait for the operation to complete (either successfully, or in a cancelled form).

[Thomasdezeeuw]

@Nemo157 @Matthias247 Good points. I no longer see any way cancellation is possible in a non-blocking synchronous way. So for memory reasons alone the buffer must be forgotten (mem::forget) when dropped, which is less then ideal.

@Nemo157 I think the output of your example would be hard to define, because the first operation may or may not succeed after being cancelled. Which means that in some executions the second read will read the 0-32 bytes, in other executions the 32-64 bytes. So I think its hard to define what the expected result should be.

[Matthias247]

@cramertj, @aturon, @carllerche Would be interesting to get an input from you too.

I found another example of the behavior after reading the article from @sdroege about glib/gio and futures integration (which is very cool btw).

Here I checked the original GTK APIs (e.g. https://developer.gnome.org/gio/stable/GOutputStream.html#g-output-stream-write-async) against their Rust wrappers ( http://gtk-rs.org/docs/gio/trait.OutputStreamExt.html#tymethod.write_bytes_async_future). It seems like to fulfill the futures contract the transmitted buffers utilize some extra boxing (and potentially copying). And the cancellation might not be fully synchronous (operation might continue to run for a while after the future was dropped, until the underlying GTK callback was delivered), which could lead to unexpected side effects when the user wants to directly start another async operation after dropping the last one.

While the wrappers are already very cool, I think they are not yet as much zero-cost as Rust futures potentially could be, since they force additional heap allocation of some arguments as well as asynchronous operation state.

[sdroege]

@Matthias247 Thanks for bringing this up.

In the context of GIO async operations there is not that much that could be improved here though, but generally for completion-based/callback-based async APIs there some opportunity and thanks for working on finding some solutions to improve the situation :)

In case of GLib/GIO:

1. You'll have allocations anyway as both GLib and GIO will allocate quite some stuff on the heap for each operation already, and on the Rust side we need an allocation to be able to store the closure for the callback somewhere and being able to pass it as a pointer to the C callback
2. Performance-wise, a bigger problem seems to be that the callback only triggers a oneshot channel and then the actual future needs to be polled again from the event loop. This could be prevented in two ways
   a) it's possible to get the future via the GLib API and then poll it directly from the callback again,
   b) if combinators like and_then were part of the Future trait those could be overridden in a meaningful way here and we could have access to the "continuation" of our future, and make use of that knowledge directly.
   Both is not a solution in case of GLib though, as there is API that gives you the currently executed GSource (basically what our Future here is). We would bypass that by directly polling our Future from the callback, and that API would then still return the GSource of the GIO async operation. While probably not a problem in practice, it nonetheless violates the API contract.

3. Cancellation: That is true but that's already the case in GLib/GIO so there's not much we can do about that really :)

4. Boxed Futures are returned from all the functions currently simply because impl trait on trait functions is not implemented or stable yet.

5. For the function you looked at in C, the corresponding Rust version is this one here: http://gtk-rs.org/docs/gio/prelude/trait.OutputStreamExtManual.html#tymethod.write_async . Different signature, and the bytes are passed as an arbitrary AsRef<[u8]> (which is then boxed to pass it around).

[Matthias247]

Thanks for your comment!

1. and 5. I fully agree that for the callback based API (http://gtk-rs.org/docs/gio/trait.OutputStreamExt.html#tymethod.write_bytes_async) the boxing is necessary. For the future based interface it could ideally be avoided, if the method creates a Future<'a> which stores the slice with the same lifetime, and which is guaranteed to outlive the underlying async operation. But that would require the future to live as long as the underlying glib callback returns, which might not happen if the user drops it at an arbitrary point.
2. True, that's another level of indirection, which always costs an extra eventloop iteration. I'm not sure how costly this is, depending on the executor it might be fairly cheap (or at least doesn't require an extra allocation). Directly polling the task from the callback might reduce interoperability with other task executors (maybe it's possible to poll gio futures from a future tokio eventloop? Not sure). But this is probably a separate discussion point from this issue.
3. I think the underlying GIO cancellation mechanism makes sense, it equals how most other frameworks model cancellation too. However with the original C API the cancellation is safe, in regards that one knows when the operation is finished by waiting for the callback (after the cancellation). By wrapping the asynchronous operation and cancellation into a future, the cancellation is started when the future is dropped. But the user has no way to determine when the operation had been finished, since the associated handle (future) is no longer visible. This might cause issues if another IO operation should be started on the same resource afterwards - I read somewhere that GIO only supports one outstanding async operation per stream.

[Matthias247]

My summary of the situation is:

In order to decrease requirements on boxing parameters to future based operations, and to better support asynchronous cancellation, mit might be helpful to enforce that some (or all) kinds of futures are driven to completion. This is already possible with the current APIs. However the current ecosystem relies on the fact that dropping futures at any point of time is possible. If a future type which relies on being driven to completion gets prematurely dropped it can cause memory issues, e.g. if the pinned future acted as storage location for the underlying IO resource while executing the requested action.

So I think this would only be a safe feature if it can be avoided through either the type system or lints. I don't yet have a concrete proposal on how this could be expressed, but I'm pretty sure others would have ideas (if there is interest in it).

If some futures must be driven to completion some of the currently existing combinators would obviously no longer work, since they partly on being able to cancel some operations synchronously. That could be either fixed by standardizing another kind of cancellation mechanism, or by supporting them only for the safely droppable futures. But this kind of discussion could behind the more general one (if there should be run-to-completion futures).

I'm also a bit torn back and forth if these ideas should be supported in the futures ecosystem:

• Implementing these kinds of futures, which e.g. act as data storage for IO resources, will require quite a bit of unsafe code and experience. Which is something that is usually not promoted within Rust.
• However on the other hand it is already possible to references and slices across yield points thanks to pinning in compiler-generated code. This is somewhat trying to apply the same pattern within hand-written "leaf-futures".
• Futures should also be a zero-cost abstraction, and requiring resources to be refcounted (or otherwise-said: garbage collected) to support them doesn't follow that principle.

[sdroege]

1. [...] and which is guaranteed to outlive the underlying async operation

The problem here is that you don't statically know any shorter lifetime than 'static for this.

3\. [...] read somewhere that GIO only supports one outstanding async operation per stream.

That's generally true but depends on the actual implementation. E.g. a socket allows a concurrent read while a write is running.

But good point on this, it's a problem as you have no way of knowing when the actual operation is cancelled with the futures-based API. The non-futures API will call your callback with an error Cancelled to signal that, which you'd also get from the futures API if you cancel in a different way than dropping (which is currently not possible).

---

Anyway, we're getting off-topic here: https://github.com/gtk-rs/gio/issues/156 :)

[ishitatsuyuki]

Cancellation has some interesting traits, and based on that I'm going to propose that those operations need asynchronous cancellation implement safety by owning all required objects (buffers, handles) in the futures itself.

First, on cancellation:

• Coorperative cancellation is annoying to implement by hand, and doing that with Drop seems to be a simple way that works.
• Just like Drop we often don't care about the result of cancellation (or any asynchronous resource deallocations): they can happen at anytime (after use), and should not block the main flow from running.

Therefore, implementing cancellation in a way similar to garbage collection sounds good to me. If anything need to Drop asynchronously, they move the needed objects out of the struct then hand it to the executor so they complete at some time.

[Matthias247]

Coorperative cancellation is annoying to implement by hand, and doing that with Drop seems to be a simple way that works.

I don't think Drop is able to perform any kind of cooperative cancellation, since it can't wait. It can however initiate a cancellation (and the operation might continue to run in the background).

Just like Drop we often don't care about the result of cancellation (or any asynchronous resource deallocations): they can happen at anytime (after use), and should not block the main flow from running.

I partially agree. From a program correctness point of view one probably doesn't care whether the resource is still in use - the program potentially wants to do something else after the cancellation anyway.

However there is some downside, that you already tangent with the term "Garbage Collection":
Allowing those operations to continue to run in the background makes resource usage less deterministic. E.g. let's say we build a webserver, and it handles requests and sockets in a IO completion based way. The request times out on a read and gets dropped. From the application logics point of view the request and socket have now been freed, and another request can get handled without increasing resource usage. However in reality some of the resources are still in use in the background, and won't get freed until that operation finally finishes. That introduces tiny bit of non-determinism, which certain kinds of programs (e.g. the ones that try to provide deterministic latencies) should try to avoid. It can probably be worked around in other mechanisms, e.g. by putting an [async] semaphore around connect/accept calls and limiting the amount of sockets in the system in that way on a lower level: The resource would only count as freed when all pending IO operations have finished. And the application level would then observe the latency when starting a new transaction instead of waiting for an existing one to be fully cancelled.

I think it's then also important for implementations to avoid undefined behavior after a cancellation - e.g. by making sure that the original IO resource always gets closed (or at least is marked as closed).

Otherwise the following piece of code will lead to serious undefined behavior:

trait FutureBasedAsyncWrite {
    type Result<'a>: Future<Output = Result<usize, Error>> + 'a;
    fn write(&'a mut self, bytes: &'a [u8]) -> Self::Result<'a>;
}

async fn writeCancelWrite<IO: FutureBasedAsyncWrite>(io: IO) {
    let writeFut1 = io.write(buffer);
    let timer = startTimer(...);
    select! {
        _ = writeFut1 => { /* Not interesting here */ return; },
       _ = timer => {},
    }
    // The operation is cancelled and the future is dropped. However the IO resource might still
    // be in use.
    // If we now start a new operation from the view of the underlying primitive and OS 2 async
    // operations might get in progress on the same resource, which is often not supported.
     io.write(buffer2).await;
}

Maybe we can try to solve this issue with exhaustive documentation on how to model those
operations. E.g. the upper program would be well-defined if cancelling writeFut1 would lead
to an internal close of io, so that the second operation would always immediately error instead of trying to perform actual IO.

Another obvious downside is that we can expose those operations only via "owned" buffers (e.g. Bytes), which are not as universal and interoperable as pure byte slices. And the interfaces need to be different than current AsyncWrite or a hypothetical future-based AsyncWrite as defined above.

[cramertj]

I don't have plans to add any aio-specific features to futures-rs specifically, so I'm going to close this for now. that said, if anyone is working to develop Rust AIO systems that could benefit from futures-rs changes, let me know and I'll be happy to discuss them!