Re: [pbs-devel] [RFC PATCH proxmox 1/4] proxmox-async: add SenderWriter helper

Re: [pbs-devel] [RFC PATCH proxmox 1/4] proxmox-async: add SenderWriter helper

[Dietmar Maurer]

This can block the executor? If so, should we use block_in_place?
Or does self.sender.blocking_send() consider that automatically?
 
> On 02/17/2022 10:40 AM Dominik Csapak <d.csapak@proxmox.com> wrote:
> 
>  
> this wraps around a tokio Sender for Vec<u8>, but implements a blocking
> write. We can use thas as an adapter for something that only takes a
> writer, and can read from it asynchonously
/listinfo/pbs-devel

Re: [pbs-devel] [RFC PATCH proxmox 1/4] proxmox-async: add SenderWriter helper

[Wolfgang Bumiller]

On Fri, Feb 18, 2022 at 08:36:05AM +0100, Dietmar Maurer wrote:
> This can block the executor? If so, should we use block_in_place?
> Or does self.sender.blocking_send() consider that automatically?

The sender in this case is a thread. The tokio side is the receiver.

The type implements `Write` which is blocking by definition, adding a
`block_in_place` inside it is never correct.

If mixing of blocking/async code is really a concern (which I don't
think it is), I'd rather not expose `from_sender` (or under a different
name as an `unsafe fn`) and instead only have a function creating the
*tuple* of `(SenderWriter, Receiver)` with appropriate documentation
about the sync/async nature of the two sides.

But I don't think this is truly warranted.

[pbs-devel] [RFC PATCH proxmox 1/4] proxmox-async: add SenderWriter helper

[Dominik Csapak]

this wraps around a tokio Sender for Vec<u8>, but implements a blocking
write. We can use thas as an adapter for something that only takes a
writer, and can read from it asynchonously

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 proxmox-async/src/blocking/mod.rs           |  3 ++
 proxmox-async/src/blocking/sender_writer.rs | 47 +++++++++++++++++++++
 2 files changed, 50 insertions(+)
 create mode 100644 proxmox-async/src/blocking/sender_writer.rs

diff --git a/proxmox-async/src/blocking/mod.rs b/proxmox-async/src/blocking/mod.rs
index 28247b3..06f821a 100644
--- a/proxmox-async/src/blocking/mod.rs
+++ b/proxmox-async/src/blocking/mod.rs
@@ -9,3 +9,6 @@ pub use tokio_writer_adapter::TokioWriterAdapter;
 
 mod wrapped_reader_stream;
 pub use wrapped_reader_stream::WrappedReaderStream;
+
+mod sender_writer;
+pub use sender_writer::SenderWriter;
diff --git a/proxmox-async/src/blocking/sender_writer.rs b/proxmox-async/src/blocking/sender_writer.rs
new file mode 100644
index 0000000..62682e5
--- /dev/null
+++ b/proxmox-async/src/blocking/sender_writer.rs
@@ -0,0 +1,47 @@
+use std::io;
+
+use anyhow::Error;
+use tokio::sync::mpsc::Sender;
+
+/// Wrapper struct around [`tokio::sync::mpsc::Sender`] for `Result<Vec<u8>, Error>` that implements [`std::io::Write`]
+pub struct SenderWriter {
+    sender: Sender<Result<Vec<u8>, Error>>,
+}
+
+impl SenderWriter {
+    pub fn from_sender(sender: tokio::sync::mpsc::Sender<Result<Vec<u8>, Error>>) -> Self {
+        Self { sender }
+    }
+
+    fn write_impl(&mut self, buf: &[u8]) -> io::Result<usize> {
+        if let Err(err) = self.sender.blocking_send(Ok(buf.to_vec())) {
+            return Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                format!("could not send: {}", err),
+            ));
+        }
+
+        Ok(buf.len())
+    }
+
+    fn flush_impl(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl io::Write for SenderWriter {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.write_impl(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.flush_impl()
+    }
+}
+
+impl Drop for SenderWriter {
+    fn drop(&mut self) {
+        // ignore errors
+        let _ = self.flush_impl();
+    }
+}
-- 
2.30.2