[Solved] ZFuture Refactoring

[jerry73204]

It is related to eclipse-zenoh/zenoh#244, stating that the zenoh's ZFuture API suffers from the issue of blocked async functions. The async API should go on a rework, in terms of improving experience of API usage, avoiding blocking issues and providing fluent C API. The initial survey to this issue is in the post below.

[jerry73204]

On Design of ZFuture and Runnable

Background: Rust Async is Colored (or Not)

Since the async feature lands to Rust, it's been a heat debate about whether async functions should be distinguished from normal functions. Namely, async fns are red, others are blue. The function coloring is about how they are treated differently. There are several for and against articles on hackernews.

• In Defense of Async: Function Colors Are Rusty: https://www.thecodedmessage.com/posts/async-colors/
• Rust async is colored, and that’s not a big deal: https://morestina.net/blog/1686/rust-async-is-colored
• Rust's async isn't f#@king colored!: https://www.hobofan.com/blog/2021-03-10-rust-async-colored/

Rust has a simple coloring rule for async and normal functions: The functions calling an async (red) function must be red itself. In other words, a normal (blue) function cannot straitforwardly call a red function. It's due to the fact that red and blue functions are different in nature. Async functions must rely on a runtime to get executed by the design of polling model, and the lifetime judgement rules are different from normal functions.

The disagreement is mostly on the blocking behavior. It's fine that normal function call blocks for a long time. However, blocked async functions hurts the performance because it stucks the runtime. Hence, blocking async functions must be treated differently and have a special purple color. However, it relies on the programmer to ensure the written async function does not block. Even the well-known futures crate written by experienced programmers has an invisible blocking bomb (rust-lang/futures-rs#2552).

The Rust community would take months and years to finally address this issue. Before it happens, we would stick to several strategies to avoid writing blocking async code.

Why Runnable and ZFuture need re-work.

The ZFuture trait adds a fn wait() method to a future type. I guess it tries to provide sync API along with async's. As we learnt from function coloring, an async call cannot become sync itself. It must rely on a runtime to do so. It implies that the API becomes be runtime dependent. Programmers may accidentally call wait() inside an async context. That's undesired consequence because it causes the async fn to block.

The Runnable binds a (sync) function to a type. The derive_zfuture! macro turns a Runnable type to be a Future. It's done by letting the polling function to call the binded Runnable function . It's alarming to me because if Runnable blocks, and the polling function blocks as well. It's hidden beneath the derive_zfuture! and programmers can fall victime to it.

However, there are still imminent needs to make async functions to be sync. One is to allow async close in Drop::drop(). It's been a big deal and you can read the article to learn about it. The other is to provide C ABI, provided that C does not have async programming in mind.

On Sync API

There are several attempts to provide sync API along with async API. The reqwest crate provides a sync clone to async API. use The maybe-async provides a macro to switch from sync to async and vice versa. Both strategies could keep blocking code away from async context, while Zenoh's Runnable does not work well.

Zenoh's internal code heavily relies on async feature to run hundereds of concurrent tasks. There is little need to turn them to be sync because it loses the concurrency, and introduces blocking code. If Zenoh public API is small and manageable, they can have sync counterparts and better not to be mixed with async parts. For example, the zenoh::Session is totally async, while zenoh::blocking::Session is totally sync. That's a way to eliminate blocking code.

Async Drop

So far Rust is not capable of running async code in Drop::drop(). Zenoh's workaround is to call ZFuture::wait() on the future. However, the pitfall is that it could block Drop::drop(). For example, the Subscriber provides mostly async methods, and has to close the connection when dropped. I believe 90% of the case Subscriber is used inside an async context. The invisible blocking drop() will hurt the performance.

There are two common workaround on this. One is to let drop() to panic if close().await was not manually called by the programmer. The other way is to call async_std::task::spawn(me.close()), passing the closing task to the runtime. It will not block the drop anyway. However, it loses the guarantee that the dropped type is not finalized even drop() returns.

Async drop is still an open problem in Rust community. The withoutboats' article on async drop is fairy a good read on it.

[Mallets]

Hello @jerry73204 ,

I've been playing around a bit with Future and how to support the same kind of ergonomics provided by ZFuture but with a more safe approach: i.e. the blocking code is the one blocking on async code.

Please find attached some code that mimics in a certain way the zenoh internal architecture and how aPutBuilder could be implemented to provide .wait() and .await API calls.

Reporting here the implementation of PutBuilder for the sake of readability.

struct PutBuilder {
    message: Option<Message>,
    queues: TargetQueues,
}

impl PutBuilder {
    async fn _await(mut pb: Pin<&mut Self>) -> bool {
        let message = pb.message.take().unwrap();
        match &mut pb.queues {
            TargetQueues::None => false,
            TargetQueues::One(q) => q.push(message).await,
            TargetQueues::Multiple(ref mut qq) => {
                let mut res = true;
                for q in qq.drain(..) {
                    res &= q.push(message.clone()).await;
                }
                res
            }
        }
    }

    pub fn wait(self) -> bool {
        task::block_on(self)
    }
}

impl Future for PutBuilder {
    type Output = bool;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        Future::poll(
            unsafe { Pin::new_unchecked(&mut Self::_await(self)) }.as_mut(),
            cx,
        )
    }
}

#[async_std::main]
async fn main() {
    // Stripping out the initialisation code...
    session.put(message.clone()).await;
    session.put(message.clone()).wait();
}

My only concern about this approach is about the use of unsafe in the poll function of the PutBuilder.
However, if I understood correctly how Pin works (see this guide), it should be safe to Pin on the stack in that way: I'm basically just mapping the PutBuilder future towards a specific async fn that accepts Pin<&mut Self> as argument.

What's your opinion on this?
Do you think is it a good approach to make ZFuture evolve?

builders.zip

[jerry73204]

In your PutBuilder example, I would simply turn async fn _await(mut pb: Pin<&mut Self>) into a sync fn wait(&mut self), given that the function has no point to wait or block. The poll() can safely call the wait() function. That would be a greater simplification.

Besides, the unsafe of Pin projection can be eliminated by pin-project (or lightweight pin-project-lite). Maybe it's what you're looking for.

use pin_project::pin_project;
use futures::ready;

#[pin-project]
struct SleepThenSend {
    is_sleeping: bool, // mapped to &mut bool
    #[pin]
    sleep: Sleep,  // mapped to Pin<&mut Sleep>
    #[pin]
    send: Send,  // mapped to Pin<&mut Sleep>
}

impl Future for SleepThenSend {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let me = self.project();
        if me.is_sleeping() {
            ready!(me.sleep.poll(cx));
            me.is_sleeping = false;
        } else {
            ready!(me.send().poll(cx));
        }
    }
}

[Mallets]

Maybe I didn't fully explain myself correctly.

DISCLAIMER 1: the queue implementation in this example is extremely simplified from what happens in reality.
The main point, in the real case, is that the push on the queue MAY block for long time when the queue is full.
This may be due e.g. to network congestion that forces zenoh to slow down and consequently to reduce the rate of draining the queue.

So, the _await in the PutBuilder simply wraps the push() operation on the queue which is truly async.
Therefore, the _await needs to be async.

DISCLAIMER 2: Whit this example I'm trying to find a good way to be able to express the semantic of wrapping an async fn into a fn poll().

In the example you provided by the pin-project, the sleep() function is being mapped. But, sleep() implements himself poll(). I personally find easier and more ergonomic from an algorithmic point of view to be able to write async fn and simply wrap them into a builder that impl Future.

DISCLAIMER 3: Ideally, the impl Future and the _await/wait actual code could be generated by a proc macro decorating any given builder.

[jerry73204]

Let me summarize the colors of functions and conversions among them. Here is the table of function colors. The sync/async means whether the function call returns a future (or async fn). The blocking/nonblocking means if the function will be the OS thread. For example, std::fs::read_to_string() blocks the OS thread, but async_std::fs::read_to_string().await does not.

kind	sync	async
nonblocking	SN	AN
blocking	SB	AB

Conversion Rules

The first rule is that it's fine to directly call X in Y, but only calling async fn in sync fn is prohibited. It can be worked around by delegating to runtime.

The second rule is that calling a blocking async/sync function causes the caller to be blocked.

Based on this rule, we have the following conversions. In the arrow A -> B, A is callee and B is caller. The => arrow means the direction needs a runtime to be done. The items with * are the special cases we'll discuss later.

• SN -> AN
• SB -> AB
• AN => SN
• AB => SB
• *AB => SN
• *SB => AN
• AN -> AN, AN -> AN, ...

Nonblocking Conversions

Based on the rules, this direction is fine because it's just done by wrapping a poll().
SN -> AN

The reverse has a similar effect, but requires a runtime to do so.
AN => SN

I would flavor SN -> AN because it only needs to provide a sync function, and let user to choose to call in sync or async way. AN => SN would make things complex.

Blocking Conversions

The blocking case is interesting because SB -> AB call is the source of chaos. In fact, the AB case should never exist in the library because it will starves the runtime resources. Though, it's not easy to stop people from calling a blocking function inside an async, either intentionally or accidentally.

The right way is writing blocking ops inside a sync function, and they can only be in sync functions. To turn them async, call tokio's and async-std's provides spawn_blocking() to do this job. It run blocking ops on a OS thread and let the async caller to wait for the thread to finish. It's effectively a SB => AN call.

fn blocking() {
    for _ in 1e8 {
        println!("lots of print");
    }
}

async {
    spawn_blocking(move || { blocking(); }).await;
}

As far as a blocking operation cannot be avoided, it's suggested to write a sync function first.

Must-be-async Cases

In addition to the arguments above suggesting to write sync functions before async counterparts, there is one scenario that we prefer writing async functions, that is mostly I/O operations. That's the place where the conversion AN => SB would happen. In this case, we would wrap the future by block_on() to be sync.

async fn read_large_file() -> String {
    async_std::fs::read_to_string("my_large_file.txt").await.unwrap()
}

move || {
    let text = block_on(read_large_file());
}

A Little Summary

To summarize, we have the first class citizens in several use cases.

• AN: I/O, synchronization jobs (locks and channels, etc)
• SN: Cheap computation
• SB: Unavoidable blocking ops, mostly heavy computation
• AB: No way

and they give rise the the conversions:

• Direct: AN -> AN, SN -> SN, SN -> AN
• spawn_blocking(): SB => AN
• block_on(): AN => SN

Implications for Zenoh

Let's break down the Zenoh to see the usage of the functions:

• AN: socket connections, internal channels
• SB: intense message routing
• SN: cheap ops, hashmap lookup, etc

Considering that the async I/O serve most part of Zenoh and plus several computation-bounded routing tasks, I would adopt the strategies:

• Write async I/O, cheap sync and blocking sync functions as it is. No block_on() indirections.
• Spawn dedicated OS threads for routing tasks and provide channels/atomics to bridge the boundary of async/sync.
• Make a clear distinction of async I/O parts and sync parts internally. For example, TransportLinkUnicast provides mostly async methods, and SerializationBatch provides only sync methods.
• The dual sync/async API only makes sense on public interface. That the place we can provide a wrapper for AN -> SB conversion.

[jerry73204]

TL:DR

• Write async for I/O, sync for costly computations. Just write as it is.
• For public interface, provide a block_on() wrapper to turn an async fn to sync, and Provide the sync fn if it's already sync.

[Mallets]

I fully agree with the explanation on function colouring. I don't have nothing to add on your description that is very comprehensive. Thanks!

Few comments inline on your considerations and and implications for zenoh (starting from your last point and following in pseudo-random order):

The dual sync/async API only makes sense on public interface. That the place we can provide a wrapper for AN -> SB conversion.

That's exactly the intention and the rationale behind this discussion. I'm not envisioning this approach in the internal code.
And in my initial idea of the example, PutBuilder is part of the public API.

Write async I/O, cheap sync and blocking sync functions as it is. No block_on() indirections.

I agree w.r.t. all the internal code.

Make a clear distinction of async I/O parts and sync parts internally. For example, TransportLinkUnicast provides mostly async methods, and SerializationBatch provides only sync methods.

That's the idea I tried to capture in the simplified example:

• The push() on the queue is async
• Routing is sync
• PutBuilder, as part of the API, will call route() to retrieve the handles to push then asynchronously on the queue in an

Spawn dedicated OS threads for routing tasks and provide channels/atomics to bridge the boundary of async/sync.

I believe this is not really an option. Such approach would have a negative impact on the performance due to:

• More context switch introduced by moving messages around different threads and tasks
• I would avoid spawning ourselves dedicated threads since that would take out the control from the user w.r.t. number of threads a zenoh application/runtime uses.

[jerry73204]

We're having a consensus on the public interface. I think the way of writing is concerned here. My concern is that the signature async fn _await(mut pb: Pin<&mut Self>) -> bool is less friendly to most devs. People have to learn the Pin concept and that's a long journey to go. I'm looking for alternatives that is more comprehensible.

Let's focus on the problem. ParBuilder provides several methods that is supposed to be truly async. Generate a sync counterparts for these methods in a comprehensive way with little overheads. Say ParBuilder::build() and Parbuilder::reset() async methods are concerned here.

impl PutBuilder {
    pub async fn build(self) -> bool { /* omit */ }
    pub async fn reset(&mut self) {/* omit */}
}

Strategy 1: ZFuture wrapper but without Box

The problem is to provide a sync block_on() method on the anonymous future returned from build(). We can do what ZFuture did, but without Box::pin(). The merit is that it works for every future type, either it's from async fn or it has poll() explicitly defined. The drawback is that it makes the public API dirty.

impl PutBuilder {
    pub fn build(self) -> ZFuture<impl Future<Output = bool>> {
        ZFuture::new(async move { /* omit */ })
    }

    pub fn reset(&mut self) -> ZFuture<impl Future<Output = ()>> {
        ZFuture::new(async move { /* omit */ })
    }
}

struct ZFuture<F: Future> {
    inner: F
}

impl<F: Future> ZFuture<F> {
    pub fn block_on(self) -> F::Output {
        block_on(self.future)
    }
}
impl<F: Future> Future for ZFuture<F> { /* omit */ }
impl<F: Future> Deref for ZFuture<F> { /* omit */ }
impl<F: Future> DerefMut for ZFuture<F> { /* omit */ }

A concern is that it allows users to call async/sync interchangeably. Careless users can block the callee inside an async context, but it also provides a workaround to do sync call inside drop(). That is, ZFuture is either a feature or a bug depending on the usage.

async move {
    builder.reset().await;
    builder.build().block_on(); // blocking the thread here
}

impl Drop for MySuperBuilder {
    fn drop(&mut self) {
        self.builder().block_on(); // yay!
    }
}

Strategy 2: Procedural macro

It elaborates your procedural macro approach. It's an automatic rewrite on the existing async method.

impl PutBuilder {
    #[zfuture]
    pub async fn build(self) -> bool { /* omit */ }
    #[zfuture]
    pub async fn reset(&mut self) { /* omit */ }
}

Or we can derive a whole new type with member methods being sync. It prevents interleaving sync/async on user side. The implication is discussed above.

#[zfuture(PutBuilderBlocking)] // derives an impl on PutBuilderBlocking type
impl PutBuilder {
    pub async fn build(self) -> bool { /* omit */ }
    pub async fn reset(&mut self) { /* omit */ }
}

Discussion on Blocking Code

In addition to the original ZFuture approach, I would remove the Runnable trait because it enables the SB -> AB loophole. IMO, I would eliminate the blocking methods by wrapping spawn_blocking() manually. The # of threads for spawn_blocking() is controlled by the runtime. It is manageable. The big problem is to determine whether a code blocks or not is still unsolved today. Maybe we can keep an eye on #[must_not_suspend] RFC.

As for your concern on performance impact, it depends on the actual implementation. For example, the TransmissionPipeline consumes thousands of messages per second and holds an internal lock for every message. The lock is short-lived and it's fine to be sync. Using async would add up lots of polling overheads. I'm not sure about the actual difference but we can have an experiment to go.

[Mallets]

Discussion on Blocking code

I'm fine about removing the Runnable trait and preclude the possibility of the loophole for the SB -> AB kind of code.

For what concerns the actual implementation, the push() of operation of the TransmissionPipeline is sync. But this is a problem today and it should be async since it could block for a long time because of network congestion.
You should really look at the push() like a socket.write(): the problem is not the writing in the buffer per-se (which takes little time) but rather that the queue can fill up and we would need to wait until the underline network gets some spare bandwidth. And in that case, it would be preferable to not block the whole thread.

_await and Pin<&mut self>

zenoh users will not need to write it. It would be the duty of zenoh committers to provide a builder to the users with the whole implementation. To give an example, the PutBuilder would be provided by zenoh and it is not intended to be implemented nor defined by a zenoh user.

Ergonomicis of the API

As provided by my initial example, I would like to have for the API something like:

Sync

fn main() {
    // ...
    session.put(message).wait();
}

The PutBuilder (or any other builder) exposes a wait() function for sync operations.

Async

In general what I would like is to avoid the following:

#[async_std::main]
async fn main() {
    // ...
    session.put(message).build().await; // BAD
}

Which is very verbose and makes the build() redundant to every API call.
Every time a put() operation needs to be performed

I would prefer to have the same API feeling as provided by ZFuture today:

#[async_std::main]
async fn main() {
    // ...
    session.put(message).await; // GOOD
}

The PutBuilder (or any other builder) is itself a Future and .await can directly be called upon.
That's way, in my initial example, I created the 'async fn _await' that is just referred to by the corresponding impl Future.
The impl Future and fn wait() actual code could be then automatically derived by a procedural macro as suggested in your Strategy 2.
We could also have a feature that enables/disable the mix of sync/async API.

Summary

Desired sync API:

fn main() {
    // ...
    session.put(message).wait();
}

Desired async API:

#[async_std::main]
async fn main() {
    // ...
    session.put(message).await;
}

How to provide the async API was the initial intention of my first reply.
To investigate a method that allows to do it by potentially wrapping any async fn under the hood.

[jerry73204]

Learned from your vision, I had a some misconception of PutBuilder that it can be evaluated at any point. I think it should be really named PutExpr because the builder pattern has a build() to mark completion.

Let me go on the detour about my experience. According to zenoh 0.6 doc, the usage looks like this. One configuration is done for each put operation.

session
    .put("/key/expression", "value")
    .encoding(Encoding::TEXT_PLAIN)
    .congestion_control(CongestionControl::Block)
    .await
    .unwrap();

put() is mostly exercised in loops with the same configuration. We have to pass a bunch of things, such as key paths and encoding above, to call put() if it resides in another function, and repeat it over and over. An explicit builder will come to the rescue.

let builder = session
    .put_builder("/key/expression")
    .encoding(Encoding::TEXT_PLAIN)
    .congestion_control(CongestionControl::Block);

for value in 0..100 {
    builder.put(value).await;
}

Having a builder also enables us to configure the senders an receivers at once, and run them in parallel.

struct Putter {
    builder: PutBuilder,
}

struct Getter {
    builder: RecvBuilder,
}

let putter = Putter::new(&session);
let getter = Putter::new(&getter);

futures::join!(
    spawn(async { loop { putter.put().await; } }),
    spawn(async { loop { getter.get().await; } }),
);

If we can really have a builder, I would add Session::put_builder() for builder creation, and let Session::put() to run a put with default configurations.

impl Session {
    pub fn put_builder(&self, key: impl KeyExpr) -> PutBuilder { /* omit */ }

    pub async fn put(&self, key: impl KeyExpr, value: Value) -> Result<()> {
        self.put_builder(key).put(value).await
    }
}

// shortcut
session.put(key, value).await?;

// for experienced users
session
    .put_builder(key)
    .encoding(Encoding::TEXT_PLAIN)
    .congestion_control(CongestionControl::Block)
    .put(value)
    .await;

This way also avoids writing the Future::poll() by hand. That's also a relief to developer's side.

Plus, the ZFuture macro approach becomes promising to us if we can write the code like this:

impl PutBuilder {
    #[zfuture]
    pub async fn put(&self, msg: Message) -> Result<()> { /* omit */ }
}

[jerry73204]

I made an attempt to write #[zfuture] macro. Here is a running example.
https://github.com/NEWSLabNTU/zenoh/blob/c4175807ff73fa2d6dcd0ac31777f3cb0f80c40c/commons/zenoh-async-rt/tests/zfuture.rs

[Mallets]

We have debated quite extensively the look-and-feel of the new API across multiple languages (see #23).

The discussion on the API has taken into account has taken into account the points made in this ZFuture Refactoring discussion.
As summary, the following approach has been adopted:

• sync and async API should be exposed separately to users. This helps avoiding to call sync functions into async context by mistake.
• Resolving the builder pattern in zenoh's Rust API is done by a dedicated function: res(), which can be sync or async depending on the API being imported. Each API might hence have a sync and async version. As a first step, the sync API could be based on the async API but it doesn't have to. The actual sync and async API implementations are therefore free to evolve independently.

// SYNC
use zenoh::prelude::sync::*;

fn main() {
    let session = zenoh::open(...).res().unwrap();
    session.put("/a", vec![0u8; 64]).res().unwrap();
}

// ASYNC
use zenoh::prelude::r#async::*;

async fn main() {
    let session = zenoh::open(...).res().await.unwrap();
    session.put("/a", vec![0u8; 64]).res().await.unwrap();
}

• Drop is not async. Therefore, undeclaring subscribers/publisher/etc. and closing sessions are made explicit and use the Drop only as a safety measure to perform the necessary cleanup.

// SYNC
use zenoh::prelude::sync::*;

fn main() {
    session.close().res().unwrap();
}

// ASYNC
use zenoh::prelude::r#async::*;

async fn main() {
    session.close().res().await.unwrap();
}