proposal: Remove Sync requirement for Body and AsyncBody

[Xuanwo]

Status: Rejected

Background

Isahc requires Body and AsyncBody to have Sync:

pub struct AsyncBody(Inner);

enum Inner {
    /// An empty body.
    Empty,

    /// A body stored in memory.
    Buffer(Cursor<Cow<'static, [u8]>>),

    /// An asynchronous reader.
    Reader(Pin<Box<dyn AsyncRead + Send + Sync>>, Option<u64>),
}

This makes users hard to meets the requirements.

Current Behavior

The root cause on why we need Sync for body is we need to store them in http::response::Builder's extension:

pub fn extension<T>(self, extension: T) -> Builder
where
    T: Any + Send + Sync + 'static,
{
    self.and_then(move |mut head| {
        head.extensions.insert(extension);
        Ok(head)
    })
}

To make our interceptors have the ability the get the body again, we store it in extension via:

builder = builder.extension(RequestBody(mem::take(&mut self.request_body)));

The only interceptor that depends on this is redirect:

let mut request_body = response
    .extensions_mut()
    .remove::<RequestBody>()
    .map(|v| v.0)
    .unwrap_or_default();

Proposal

We can remove the Sync requirement for Body, and use Mutex<Option<T>> instead to make our body Sync:

In response builder:

builder = builder.extension(Mutex::new(Some(RequestBody(mem::take(&mut self.request_body)))));

In redirect interceptor:

let mut request_body = response
    .extensions_mut()
    .remove::<Mutex<Option<RequestBody>>>()
    .map(|v| v.lock().unwrap().take().unwrap().0)
    .unwrap_or_default();

---

Any ideas? cc @sagebind

[sagebind]

Note that the same body types are used both for the request and for the response, so extra bounds that put extra burden on the request turn into extra benefits on the response. We made the body Sync in #172 to make the response body easier to use; the request body must also be Sync only as a consequence of using the same types for both.

I'm not entirely sure right now how valuable it is for the response body to be Sync; perhaps we could re-evaluate that decision based on some example usages and perhaps compare with other libraries to see what is most common.

[Xuanwo]

I don't know why we need to make our response Sync. In my opinion, it's up to users: If they want to access response from different threads, they can wrap our response inside Mutex<T>.

---

Ohter libs:

Hyper:

pub fn wrap_stream<S, O, E>(stream: S) -> Body
where
    S: Stream<Item = Result<O, E>> + Send + 'static,
    O: Into<Bytes> + 'static,
    E: Into<Box<dyn StdError + Send + Sync>> + 'static,
{
    let mapped = stream.map_ok(Into::into).map_err(Into::into);
    Body::new(Kind::Wrapped(SyncWrapper::new(Box::pin(mapped))))
}

Arrow2 (my real case):

https://docs.rs/arrow2/latest/arrow2/io/parquet/read/fn.read_metadata_async.html

pub async fn read_metadata_async<R: AsyncRead + AsyncSeek + Send + Unpin>(
    reader: &mut R,
) -> Result<FileMetaData> {
    Ok(_read_metadata_async(reader).await?)
}

NOTE: AsyncSeek is other thing handled by myself. This example is used to show that most libs don't need Sync.

[sagebind]

Well the underlying reader we use for the response body already implements Sync without a mutex, so if users do need the response body to be Sync then wrapping it in a Mutex will perform worse than what we have today.

[Xuanwo]

Well the underlying reader we use for the response body already implements Sync without a mutex, so if users do need the response body to be Sync then wrapping it in a Mutex will perform worse than what we have today.

Nice comments.

Also, here is the reqwest's body:

/// An asynchronous request body.
pub struct Body {
    inner: Inner,
}

// The `Stream` trait isn't stable, so the impl isn't public.
pub(crate) struct ImplStream(Body);

enum Inner {
    Reusable(Bytes),
    Streaming {
        body: Pin<
            Box<
                dyn HttpBody<Data = Bytes, Error = Box<dyn std::error::Error + Send + Sync>>
                    + Send
                    + Sync,
            >,
        >,
        timeout: Option<Pin<Box<Sleep>>>,
    },
}

---

Current design seems OK (at least we have the same requirement with reqwest).

[Xuanwo]

Thanks for replying. I'm going to set this proposal's status as rejected now.

However, this thread is open for any further discussion.

[Xuanwo]

Isahc will not access the same Reader from different threads at the same time. So I'm guessing users can wrap the reader in this way:

//! This mod is inspired by [async-compression](https://docs.rs/crate/async-compression/0.3.1/source/src/unshared.rs)
//!
//! More discussion could be found at [What shall Sync mean across an .await?](https://internals.rust-lang.org/t/what-shall-sync-mean-across-an-await/12020/26)

use std::pin::Pin;
use std::task::{Context, Poll};
use futures::AsyncRead;
use crate::BytesRead;

pub fn unshared_reader<R: BytesRead>(r: R) -> UnsharedReader<R> {
    UnsharedReader::new(r)
}

/// Wraps a type and only allows unique borrowing, the main usecase is to wrap a `!Sync` type and
/// implement `Sync` for it as this type blocks having multiple shared references to the inner
/// value.
///
/// # Safety
///
/// We must be careful when accessing `inner`, there must be no way to create a shared reference to
/// it from a shared reference to an `Unshared`, as that would allow creating shared references on
/// multiple threads.
///
/// As an example deriving or implementing `Clone` is impossible, two threads could attempt to
/// clone a shared `Unshared<T>` reference which would result in accessing the same inner value
/// concurrently.
pub struct UnsharedReader<R: BytesRead>(R);

impl<R: BytesRead> UnsharedReader<R> {
    pub fn new(inner: R) -> Self {
        UnsharedReader(inner)
    }

    pub fn get_mut(&mut self) -> &mut R {
        &mut self.0
    }
}

/// Safety: See comments on main docs for `UnsharedReader`
unsafe impl<R: BytesRead> Sync for UnsharedReader<R> {}

impl<R: BytesRead> AsyncRead for UnsharedReader<R> {
    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<std::io::Result<usize>> {
        Pin::new(self.get_mut().get_mut()).poll_read(cx, buf)
    }
}

So, we can expose a loosen API to end-users:

// Input `r` here only requires `Send`.
let mut req = self.put_object_request(&p, AsyncBody::from_reader_sized(unshared_reader(r), args.size()))?;