Add `SinkWriter` in analogy to `StreamReader` to `tokio_util` #5070

dbischof90 · 2022-10-01T16:11:52Z

Motivation

Tokio provides StreamReader, enabling us to treat streams as file-like interfaces through AsyncRead. This however only exists for streams, not for general sinks, for which this PR provides an implementation. This is a new attempt on #5019 .

Solution

A sink is wrapped which takes owned buffers in form of a Vec<u8>. Each buffer is copied to a new vector, which is sent to the underlying sink.

Attempted alternatives

I have tried three other designs:

Send items byte by byte. This is very flexible, however, very inefficient.
Sending slices directly, via a higher-ranked trait bound to allow general lifetimes:

impl<S> AsyncWrite for SinkWriter<S>
where
    for<'r> S: Sink<&'r [u8], Error = io::Error>,
    [...]

This lead to non-trivial errors in the test case, mixing multiple lifetimes which I ultimately could not resolve:

error: implementation of `futures_util::Sink` is not general enough
  --> tokio-util/tests/io_sink_writer.rs:39:18
   |
39 |     let writer = SinkWriter::new(
   |                  ^^^^^^^^^^^^^^^ implementation of `futures_util::Sink` is not general enough
   |
   = note: `PollSender<&'2 [u8]>` must implement `futures_util::Sink<&'1 [u8]>`, for any lifetime `'1`...
   = note: ...but it actually implements `futures_util::Sink<&'2 [u8]>`, for some specific lifetime `'2`

Sending slices directly, via a specific, generic lifetime:

impl<'a, S> AsyncWrite for SinkWriter<'a, S>
where
    S: Sink<&'a [u8], Error = io::Error>,
    [...]

This lead to the problem that the buffer lifetime could not be restricted by the lifetime of the Sink.

error: lifetime may not live long enough
   --> tokio-util/src/io/sink_writer.rs:103:33
    |
92  | impl<'a, S> AsyncWrite for SinkWriter<'a, S>
    |      -- lifetime `'a` defined here
...
99  |         buf: &[u8],
    |              - let's call the lifetime of this reference `'1`
...
103 |                 if let Err(e) = self.as_mut().project().inner.start_send(buf) {
    |                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ argument requires that `'1` must outlive `'a`

The current proposal relies on copying the slice to a Vec<u8> and resolve the lifetime problems in that way.

Darksonn

You asked me how to avoid the Vec<u8> allocation. The changes below should do it.

tokio-util/src/io/sink_writer.rs

Darksonn · 2022-10-02T11:39:13Z

tokio-util/src/io/sink_writer.rs

+impl<S> SinkWriter<S>
+where
+    S: Sink<Vec<u8>, Error = io::Error>,
+{


Suggested change

impl<S> SinkWriter<S>

where

S: Sink<Vec<u8>, Error = io::Error>,

{

impl<S> SinkWriter<S> {

Darksonn · 2022-10-02T11:39:29Z

tokio-util/src/io/sink_writer.rs

+
+impl<S, E> AsyncWrite for SinkWriter<S>
+where
+    S: Sink<Vec<u8>, Error = E>,


Suggested change

S: Sink<Vec<u8>, Error = E>,

for<'a> S: Sink<&'a [u8], Error = E>,

Darksonn · 2022-10-02T11:39:38Z

tokio-util/src/io/sink_writer.rs

+    ) -> Poll<Result<usize, io::Error>> {
+        match self.as_mut().project().inner.poll_ready(cx) {
+            Poll::Ready(Ok(())) => {
+                if let Err(e) = self.as_mut().project().inner.start_send(buf.to_vec()) {


Suggested change

if let Err(e) = self.as_mut().project().inner.start_send(buf.to_vec()) {

if let Err(e) = self.as_mut().project().inner.start_send(buf) {

dbischof90 · 2022-10-02T19:25:53Z

You asked me how to avoid the Vec<u8> allocation. The changes below should do it.

@Darksonn, as mentioned, this doesn't work (or is at least not that simple), see commit 3d7097b together with an additional trait bound - adding that leads essentially to case 2) described above. Removing these trait bounds is your proposal and results in no error if I comment out line 21 in tokio-util/tests/io_sink_writer.rs, but writer won't implement 'AsyncWrite' which you'll see once you add it back. rust-analyzer however is convinced that AsyncWriteExt is implemented here.

Darksonn · 2022-10-03T10:52:20Z

That's because your Sink in the test doesn't implement for<'a> Sink<&' [u8], Error = io::Error> as we require it to.

farnz · 2022-10-03T11:17:16Z

You asked me how to avoid the Vec<u8> allocation. The changes below should do it.

@Darksonn, as mentioned, this doesn't work (or is at least not that simple), see commit 3d7097b together with an additional trait bound - adding that leads essentially to case 2) described above. Removing these trait bounds is your proposal and results in no error if I comment out line 21 in tokio-util/tests/io_sink_writer.rs, but writer won't implement 'AsyncWrite' which you'll see once you add it back. rust-analyzer however is convinced that AsyncWriteExt is implemented here.

The core problem here is that the mpsc channel you're using doesn't implement Sink for any lifetime - it implements it for a lifetime that's fixed at channel creation time. This is what Rust is trying to tell you with the error notes:

   = note: `PollSender<&'2 [u8]>` must implement `futures_util::Sink<&'1 [u8]>`, for any lifetime `'1`...
   = note: ...but it actually implements `futures_util::Sink<&'2 [u8]>`, for some specific lifetime `'2`

You'll need to implement the Sink in your test such that it accepts a buffer with any lifetime, not a buffer with a specific lifetime that's part of the Sink's type - in other words, you want the type parameter Item on your Sink<Item> to be for<'r> Item: &'r [u8] and not Item: &[u8].

dbischof90 · 2022-10-03T11:42:18Z

You asked me how to avoid the Vec<u8> allocation. The changes below should do it.

@Darksonn, as mentioned, this doesn't work (or is at least not that simple), see commit 3d7097b together with an additional trait bound - adding that leads essentially to case 2) described above. Removing these trait bounds is your proposal and results in no error if I comment out line 21 in tokio-util/tests/io_sink_writer.rs, but writer won't implement 'AsyncWrite' which you'll see once you add it back. rust-analyzer however is convinced that AsyncWriteExt is implemented here.

The core problem here is that the mpsc channel you're using doesn't implement Sink for any lifetime - it implements it for a lifetime that's fixed at channel creation time. This is what Rust is trying to tell you with the error notes:
   = note: `PollSender<&'2 [u8]>` must implement `futures_util::Sink<&'1 [u8]>`, for any lifetime `'1`...
   = note: ...but it actually implements `futures_util::Sink<&'2 [u8]>`, for some specific lifetime `'2`
You'll need to implement the Sink in your test such that it accepts a buffer with any lifetime, not a buffer with a specific lifetime that's part of the Sink's type - in other words, you want the type parameter Item on your Sink<Item> to be for<'r> Item: &'r [u8] and not Item: &[u8].

That's because your Sink in the test doesn't implement for<'a> Sink<&' [u8], Error = io::Error> as we require it to.

Hi both, that is clear (I can at least read :) ). But by using PollSender and channel(), can this be achieved? I also tried to build some very specific channel there but it also feels like this should not be so very specific. From looking at the test, that should work (in the sense that whatever the implementation of of AsyncWrite for some form of SinkWriter is supposed to look like, from an ergonomics point of view, that looks a reasonable way to use them.

So how does the implementation would need to change that it can be used like in the test here? As pushed initially, copying the buf: &[u8] to a Vec<u8> and use that works. I'd like to make sure that there is not a semi-obvious way such that this additional allocation and copy can be avoided.

farnz · 2022-10-03T12:08:08Z

But by using PollSender and channel(), can this be achieved? I also tried to build some very specific channel there but it also feels like this should not be so very specific. From looking at the test, that should work (in the sense that whatever the implementation of of AsyncWrite for some form of SinkWriter is supposed to look like, from an ergonomics point of view, that looks a reasonable way to use them.

So how does the implementation would need to change that it can be used like in the test here? As pushed initially, copying the buf: &[u8] to a Vec<u8> and use that works. I'd like to make sure that there is not a semi-obvious way such that this additional allocation and copy can be avoided.

Using PollSender and channel() is going to require a copy at some point, because of the lifetime of the buffer.

When I call poll_write, I do not pass you ownership ([u8]) or shared ownership (Arc<[u8]>) of the buffer; instead, I pass you a reference to the buffer (&[u8]) which is valid until poll_write returns. A channel needs the items passed through it to remain valid until the caller of Receiver::recv is done with them. This results in a lifetime issue if you're trying to pass &[u8] from poll_write through a channel to the Receiver, because you'll call Sender::send then immediately return from poll_write; importantly, this return from poll_write can happen before you call Receiver::recv.

For your test case, the simplest way to implement a Sink that just sends written data through a channel is to have the Sink convert from a reference to byte slice into a Vec<u8>, and send the Vec<u8> through the channel. However, for more complex Sinks, converting to a Vec<u8> is a wasted allocation, since the Sink will promptly convert it to an &[u8] in start_send, do something with the byte slice, and then not need to keep a copy of the data around. As a trivial example, imagine a Sink that counts the number of bytes you've tried to sink into it - its start_send would just call item.len() and not need the copy. As a more plausible example, you could also have a Sink that's writing buffers through the DisplayPort I2C over AUX interface, which allows you to write in 16 byte chunks - the first 16 bytes go directly without a copy, and the rest are buffered in an internal buffer that gets flushed out when each AUX transaction completes. If you sent a 16 byte (or smaller) buffer when the AUX interface is idle, you'd be able to write immediately without buffering.

dbischof90 · 2022-10-03T13:24:13Z

For your test case, the simplest way to implement a Sink that just sends written data through a channel is to have the Sink convert from a reference to byte slice into a Vec<u8>, and send the Vec<u8> through the channel. However, for more complex Sinks, converting to a Vec<u8> is a wasted allocation, since the Sink will promptly convert it to an &[u8] in start_send, do something with the byte slice, and then not need to keep a copy of the data around. As a trivial example, imagine a Sink that counts the number of bytes you've tried to sink into it - its start_send would just call item.len() and not need the copy. As a more plausible example, you could also have a Sink that's writing buffers through the DisplayPort I2C over AUX interface, which allows you to write in 16 byte chunks - the first 16 bytes go directly without a copy, and the rest are buffered in an internal buffer that gets flushed out when each AUX transaction completes. If you sent a 16 byte (or smaller) buffer when the AUX interface is idle, you'd be able to write immediately without buffering.

We're getting there, that is exactly what I was looking at. I also understand the requirement for the Vec here now. But you're describing two Sinks now, one in which I technically would not need a copy and one in which I need one (as in my testcase). That's hard to account for here, right? I'd be looking at the situation again that I Case 2) above is indeed the 'better' solution as it just passes the slice between both interfaces and I would indeed have to think about making a better test case (with that in mind). Looking at that PollSender, can that be made fit here?
Depending on the complexity, your (throughout) explanation might even be a hint towards a CopySinkWriter and a more generic, non-copying SinkWriter.

With that in mind, this quote here is not yet fully clear to me implication-wise:

You'll need to implement the Sink in your test such that it accepts a buffer with any lifetime, not a buffer with a specific lifetime

Is a specific lifetime not especially covered by a Sink that expects 'any' lifetime? I'm more than happy to admit that I used that initially to make the linter happy and thought I roughly understand the consequences.

farnz · 2022-10-03T15:34:42Z

For your test case, the simplest way to implement a Sink that just sends written data through a channel is to have the Sink convert from a reference to byte slice into a Vec<u8>, and send the Vec<u8> through the channel. However, for more complex Sinks, converting to a Vec<u8> is a wasted allocation, since the Sink will promptly convert it to an &[u8] in start_send, do something with the byte slice, and then not need to keep a copy of the data around. As a trivial example, imagine a Sink that counts the number of bytes you've tried to sink into it - its start_send would just call item.len() and not need the copy. As a more plausible example, you could also have a Sink that's writing buffers through the DisplayPort I2C over AUX interface, which allows you to write in 16 byte chunks - the first 16 bytes go directly without a copy, and the rest are buffered in an internal buffer that gets flushed out when each AUX transaction completes. If you sent a 16 byte (or smaller) buffer when the AUX interface is idle, you'd be able to write immediately without buffering.

We're getting there, that is exactly what I was looking at. I also understand the requirement for the Vec here now. But you're describing two Sinks now, one in which I technically would not need a copy and one in which I need one (as in my testcase). That's hard to account for here, right? I'd be looking at the situation again that I Case 2) above is indeed the 'better' solution as it just passes the slice between both interfaces and I would indeed have to think about making a better test case (with that in mind). Looking at that PollSender, can that be made fit here? Depending on the complexity, your (throughout) explanation might even be a hint towards a CopySinkWriter and a more generic, non-copying SinkWriter.

You can't make PollSender work with poll_write and a zero-copy setup. When poll_write returns, the reference buf: &[u8] is no longer valid, but PollSender::start_send needs the reference to remain valid until after Receiver::recv returns the buffer to the user. You don't control when Receiver::recv is called, so you can't enforce this requirement.

With that in mind, this quote here is not yet fully clear to me implication-wise:

You'll need to implement the Sink in your test such that it accepts a buffer with any lifetime, not a buffer with a specific lifetime

Is a specific lifetime not especially covered by a Sink that expects 'any' lifetime? I'm more than happy to admit that I used that initially to make the linter happy and thought I roughly understand the consequences.

A specific lifetime 'a is a subset of all possible lifetimes for<'a> 'a. The distinction, and it's subtle, is in which piece of code gets to set the smallest region permitted for the lifetime.

With S: Sink<&'a [u8]>, you're telling the compiler that there is a lifetime, and the Sink gets to set the smallest permissible region for the lifetime (which can thus be longer than the lifetime that poll_write has given you). With for<'a> S: Sink<&'a [u8]>, you're still saying there is a lifetime, but now you're saying that the Sink has to handle the lifetime 'a that the caller supplies.

farnz · 2022-10-03T15:37:21Z

FWIW, my inclination would be to have a SinkWriter that writes to any Sink with a wide-open lifetime, and a CopySink that does the copying from &'_[T] to Vec<T>. You can then combine the two to make your test case work - SinkWriter can write to the CopySink, and the CopySink can write to a PollSender whose internal channel passes Vec<u8> around.

Darksonn · 2022-10-04T10:00:25Z

I think that if we add a wrapper like this:

impl<'a, S> Sink<&'a [u8]> for CopyToBytes<S>
where
    S: Sink<Bytes>,
{
    type Error = S::Error;
    
    ...
}

Then that's good enough. You would be able to do SinkWriter::new(CopyToBytes::new(poll_sender)) in the example.

dbischof90 · 2022-10-04T20:57:29Z

Thank you two for those considerations. From what I understand, I agree.

Can you think of a construct fit for a separate test which would work with a SinkWriter in isolation instead of both SinkWriter and CopyToBytes? As you see, that was close to Case 3) above which (what I now understand) did not work with the PollSender. I could of course implement a toy sender that just sends the first byte of a buffer or clones it internally or something else but I find it always helpful if one uses somewhat realistic situations in a test. This here is now of course not an embedded environment but maybe you have something in mind. Otherwise it has to be some constructed thing ofc.

farnz · 2022-10-05T08:49:43Z

Thank you two for those considerations. From what I understand, I agree.

Can you think of a construct fit for a separate test which would work with a SinkWriter in isolation instead of both SinkWriter and CopyToBytes? As you see, that was close to Case 3) above which (what I now understand) did not work with the PollSender. I could of course implement a toy sender that just sends the first byte of a buffer or clones it internally or something else but I find it always helpful if one uses somewhat realistic situations in a test. This here is now of course not an embedded environment but maybe you have something in mind. Otherwise it has to be some constructed thing ofc.

Three non-toy Sinks that don't you could build easily for a test:

A Sink that hashes or CRCs the bytes sent to it. You will need a hash crate like sha3 or crc32fast to do the computation on the slices you get given.
A compressing Sink that uses something like GzEncoder to compress the slices as it copies them into your private buffer.
A packetizer that puts a trivial header (MPEG PES header for example, with start code 0x000001BF, no PES extension, and packet length not zero, so the packets begin 00 00 01 BF si ze and then the data, split into 64 KiB chunks).

All of these are plausible things to want, and all of them do their own internal copies that don't benefit from being passed Vec<u8> instead of &[u8]

Darksonn · 2022-10-05T11:01:42Z

If you just need the sink for a test, you can write a sink that just appends the given bytes to a Vec<u8>. However, I think testing it together with CopyToBytes is also fine.

farnz

And note that CopyToBytes has the same "inadvisable" comment that needs thinking about.

StreamReader has this comment because the unit of poll_read is bytes, but the unit of the Stream is Bufs, and thus StreamReader can have a partially read item buffered inside it, making it hard (but not impossible) to avoid getting buffers out of order if you mix StreamReader::poll_read with Stream::next.

SinkWriter doesn't have any internal buffering, and thus doesn't have this issue - there's no way to get writes out of order with code that appears to handle it all in-order.

tokio-util/src/io/sink_writer.rs

tokio-util/tests/io_sink_writer.rs

Darksonn

The code looks good, but the documentation needs work.

tokio-util/src/io/sink_writer.rs

Co-authored-by: Alice Ryhl <aliceryhl@google.com>

farnz

Code looks great to me - just documentation to tweak now.

tokio-util/src/io/sink_writer.rs

tokio-util/src/io/copy_to_bytes.rs

Co-authored-by: Simon Farnsworth <simon@farnz.org.uk>

farnz

I'd also suggest looking at the final docs generated by rustdoc - I find I'm better at spotting errors in the docs when I do that, as compared to looking in the source or at rust-analyzer overlays.

tokio-util/src/io/copy_to_bytes.rs

tokio-util/src/io/sink_writer.rs

tokio-util/tests/io_sink_writer.rs

dbischof90 · 2022-10-10T13:02:12Z

Ah you're right, I was a bit careless there. I'll take some time and check the finalised documentation.

farnz

Otherwise, now LGTM. @Darksonn and the rest of the Tokio team have the final say, though (I'm just a contributor).

tokio-util/src/io/copy_to_bytes.rs

farnz · 2022-10-13T10:45:46Z

tokio-util/src/io/sink_writer.rs

+    ) -> Poll<Result<usize, io::Error>> {
+        match self.as_mut().project().inner.poll_ready(cx) {
+            Poll::Ready(Ok(())) => {
+                if let Err(e) = self.as_mut().project().inner.start_send(buf) {


And I think this as_mut is also not needed.

Both of those calls need a pinned mutable reference here.

You are passed a pinned mutable reference - indeed, looking at this code, I think there's room to refactor with let this = self.project() before the match, and then using this instead of self.as_mut().project().

Hmm that doesn't work I think - if I use this this approach I will have an ownership problem since poll_readytakes self and doesn't borrow. So I have to project twice there.

You should only need to project once, and then use the projection throughout - inner should have type Pin<&mut S> after projection, so you can put the as_mut() after this.inner instead.

Oh that does indeed work - meaning I also don't need to restrict poll_write to be mut. Out of interest, what would (besides that) be the practical implication for this? Does it add a cost if I project twice versus pinning a reference twice?

Projection is free, so it doesn't really have any impact how you do it. Also, the part of the arguments that affects the signature is the type written after the colon, and adding mut on the argument name is before the colon, so it doesn't actually affect the signature of the function.

tokio-util/src/io/copy_to_bytes.rs

tokio-util/src/io/sink_writer.rs

Darksonn · 2022-10-18T07:38:40Z

tokio-util/src/io/copy_to_bytes.rs

+    /// A helper which wraps a `Sink<Bytes>` and converts it into
+    /// a `Sink<&'a [u8]>` by copying each byte slice into an owned [`Bytes`].


Suggested change

/// A helper which wraps a `Sink<Bytes>` and converts it into

/// a `Sink<&'a [u8]>` by copying each byte slice into an owned [`Bytes`].

/// A helper that wraps a `Sink<Bytes>` and converts it into

/// a `Sink<&'a [u8]>` by copying each byte slice into an owned [`Bytes`].

Darksonn · 2022-10-18T07:41:47Z

tokio-util/src/io/sink_writer.rs

+            Poll::Pending => {
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }


Adding a wake call here will be a busy loop that consumes 100% CPU waiting for it to become ready. We shouldn't do that.

The call to poll_ready has returned Pending, so it promises to wake the waker for us when poll_ready can make progress.

Suggested change

Poll::Pending => {

cx.waker().wake_by_ref();

Poll::Pending

}

Poll::Pending => Poll::Pending,

Darksonn · 2022-10-18T07:48:33Z

tokio-util/src/io/sink_writer.rs

+    /// This adapter implements [`AsyncWrite`] for `Sink<&[u8]>`. If you want to
+    /// implement `Sink<_>` for [`AsyncWrite`], see the [`codec`] module; if you need to implement
+    /// [`AsyncWrite`] for `Sink<Bytes>`, see [`CopyToBytes`].


Suggested change

/// This adapter implements [`AsyncWrite`] for `Sink<&[u8]>`. If you want to

/// implement `Sink<_>` for [`AsyncWrite`], see the [`codec`] module; if you need to implement

/// [`AsyncWrite`] for `Sink<Bytes>`, see [`CopyToBytes`].

/// This adapter implements [`AsyncWrite`] for `Sink<&[u8]>`. Because of the

/// lifetime, this trait is relatively rarely implemented. The main ways to

/// get a `Sink<&[u8]>` that you can use with this type are:

///

/// * With the codec module by implementing the [`Encoder<&[u8]>`] trait.

/// * By wrapping a `Sink<Bytes>` in a [`CopyToBytes`].

/// * Manually implementing `Sink<&[u8]>` directly.

///

/// The opposite conversion of implementing `Sink<_>` for an [`AsyncWrite`]

/// is done using the [`codec`] module.

///

/// [`Encoder<&[u8]>`]: crate::codec::Encoder

Darksonn · 2022-10-23T18:20:07Z

I think the only missing changes here are documentation. Would you be willing to improve the documentation further? Otherwise I can also merge this and have it fixed in a follow-up PR.

Darksonn

I've added a commit that improves the documentation. Thanks for the PR.

Per review comment tokio-rs#5070 (comment): > Adding a wake call here will be a busy loop that consumes 100% CPU > waiting for it to become ready. We shouldn't do that. Furthermore, according to https://docs.rs/futures-sink/latest/futures_sink/trait.Sink.html#tymethod.poll_ready, poll_ready will register current task to be notified. Discussion: https://discord.com/channels/500028886025895936/500336346770964480/1072534504981418024

Per review comment tokio-rs#5070 (comment): > Adding a wake call here will be a busy loop that consumes 100% CPU > waiting for it to become ready. We shouldn't do that. Furthermore, according to https://docs.rs/futures-sink/latest/futures_sink/trait.Sink.html#tymethod.poll_ready, poll_ready will make sure that the current task is notified. Discussion: https://discord.com/channels/500028886025895936/500336346770964480/1072534504981418024

dbischof90 added 2 commits October 1, 2022 18:06

Reworked implementation, used Vec to circumvent lifetime problem

b1cfd5a

Reworked implementation, used Vec to circumvent lifetime problem

1f89c2d

Darksonn added A-tokio-util Area: The tokio-util crate M-io Module: tokio/io labels Oct 2, 2022

Darksonn reviewed Oct 2, 2022

View reviewed changes

dbischof90 added 3 commits October 2, 2022 20:55

Rustfmt fixes

ee33761

Applied suggested changes, nonworking test

d2130d2

Nonworking test, changed sink type from Vec to slice

3d7097b

dbischof90 added 4 commits October 6, 2022 08:20

Added new wrapping structure to own slices

f2fe6a2

Left-over cleanup in tests

bdcd41e

Left-over cleanup in tests

cbe0702

Merge branch 'master' into sink_writer

715a73f

farnz suggested changes Oct 6, 2022

View reviewed changes

Darksonn reviewed Oct 6, 2022

View reviewed changes

tokio-util/src/io/sink_writer.rs Outdated Show resolved Hide resolved

tokio-util/src/io/sink_writer.rs Outdated Show resolved Hide resolved

tokio-util/src/io/sink_writer.rs Outdated Show resolved Hide resolved

tokio-util/src/io/sink_writer.rs Outdated Show resolved Hide resolved

dbischof90 and others added 3 commits October 7, 2022 19:49

Added next round of feedback

52fefb3

Merge branch 'master' into sink_writer

c3069b8

Apply suggestions from code review

60c1b4d

Co-authored-by: Alice Ryhl <aliceryhl@google.com>

farnz reviewed Oct 10, 2022

View reviewed changes

tokio-util/src/io/sink_writer.rs Outdated Show resolved Hide resolved

tokio-util/src/io/copy_to_bytes.rs Outdated Show resolved Hide resolved

Update tokio-util/src/io/sink_writer.rs

29ed45e

Co-authored-by: Simon Farnsworth <simon@farnz.org.uk>

farnz reviewed Oct 10, 2022

View reviewed changes

Next round of documentation review

e6865ad

farnz approved these changes Oct 13, 2022

View reviewed changes

Darksonn reviewed Oct 14, 2022

View reviewed changes

tokio-util/src/io/copy_to_bytes.rs Outdated Show resolved Hide resolved

tokio-util/src/io/sink_writer.rs Outdated Show resolved Hide resolved

dbischof90 added 4 commits October 15, 2022 16:19

Removed from CopyToBytes and fixed documentation

b55a084

Fixed documentation

1252fe3

Refactor projection

8753859

Small detail in documentation

99e9802

Darksonn reviewed Oct 18, 2022

View reviewed changes

Darksonn approved these changes Oct 30, 2022

View reviewed changes

Improve documentation

1652b29

Darksonn enabled auto-merge (squash) October 30, 2022 10:22

Fix doc test

dc9fc2e

Darksonn merged commit 620880f into tokio-rs:master Oct 30, 2022

huonw mentioned this pull request Jan 22, 2023

Release a new version of tokio-util #5388

Closed

funbringer mentioned this pull request Feb 7, 2023

bugfix: Remove erroneous wake call in SinkWriter #5436

Merged

Darksonn mentioned this pull request Feb 9, 2023

Prepare tokio-util v0.7.5 #5442

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add `SinkWriter` in analogy to `StreamReader` to `tokio_util` #5070

Add `SinkWriter` in analogy to `StreamReader` to `tokio_util` #5070

dbischof90 commented Oct 1, 2022

Darksonn left a comment

Darksonn Oct 2, 2022

Darksonn Oct 2, 2022

Darksonn Oct 2, 2022

dbischof90 commented Oct 2, 2022 •

edited

Loading

Darksonn commented Oct 3, 2022 •

edited

Loading

farnz commented Oct 3, 2022

dbischof90 commented Oct 3, 2022

farnz commented Oct 3, 2022

dbischof90 commented Oct 3, 2022

farnz commented Oct 3, 2022

farnz commented Oct 3, 2022

Darksonn commented Oct 4, 2022

dbischof90 commented Oct 4, 2022 •

edited

Loading

farnz commented Oct 5, 2022

Darksonn commented Oct 5, 2022

farnz left a comment

Darksonn left a comment

farnz left a comment

farnz left a comment

dbischof90 commented Oct 10, 2022

farnz left a comment

farnz Oct 13, 2022

dbischof90 Oct 13, 2022

farnz Oct 13, 2022

dbischof90 Oct 15, 2022

farnz Oct 15, 2022

dbischof90 Oct 15, 2022

Darksonn Oct 18, 2022

Darksonn Oct 18, 2022

Darksonn Oct 18, 2022

Darksonn Oct 18, 2022 •

edited

Loading

Darksonn commented Oct 23, 2022

Darksonn left a comment

	S: Sink<Vec<u8>, Error = E>,
	for<'a> S: Sink<&'a [u8], Error = E>,

	if let Err(e) = self.as_mut().project().inner.start_send(buf.to_vec()) {
	if let Err(e) = self.as_mut().project().inner.start_send(buf) {

		/// A helper which wraps a `Sink<Bytes>` and converts it into
		/// a `Sink<&'a [u8]>` by copying each byte slice into an owned [`Bytes`].

-    /// This adapter implements [`AsyncWrite`] for `Sink<&[u8]>`. If you want to
-    /// implement `Sink<_>` for [`AsyncWrite`], see the [`codec`] module; if you need to implement
-    /// [`AsyncWrite`] for `Sink<Bytes>`, see [`CopyToBytes`].
+    /// This adapter implements [`AsyncWrite`] for `Sink<&[u8]>`. Because of the
+    /// lifetime, this trait is relatively rarely implemented. The main ways to
+    /// get a `Sink<&[u8]>` that you can use with this type are:
+    ///
+    ///  * With the codec module by implementing the [`Encoder<&[u8]>`] trait.
+    ///  * By wrapping a `Sink<Bytes>` in a [`CopyToBytes`].
+    ///  * Manually implementing `Sink<&[u8]>` directly.
+    ///
+    /// The opposite conversion of implementing `Sink<_>` for an [`AsyncWrite`]
+    /// is done using the [`codec`] module.
+    ///
+    /// [`Encoder<&[u8]>`]: crate::codec::Encoder

Add SinkWriter in analogy to StreamReader to tokio_util #5070

Add SinkWriter in analogy to StreamReader to tokio_util #5070

Conversation

dbischof90 commented Oct 1, 2022

Motivation

Solution

Attempted alternatives

Darksonn left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dbischof90 commented Oct 2, 2022 • edited Loading

Darksonn commented Oct 3, 2022 • edited Loading

farnz commented Oct 3, 2022

dbischof90 commented Oct 3, 2022

farnz commented Oct 3, 2022

dbischof90 commented Oct 3, 2022

farnz commented Oct 3, 2022

farnz commented Oct 3, 2022

Darksonn commented Oct 4, 2022

dbischof90 commented Oct 4, 2022 • edited Loading

farnz commented Oct 5, 2022

Darksonn commented Oct 5, 2022

farnz left a comment

Choose a reason for hiding this comment

Darksonn left a comment

Choose a reason for hiding this comment

farnz left a comment

Choose a reason for hiding this comment

farnz left a comment

Choose a reason for hiding this comment

dbischof90 commented Oct 10, 2022

farnz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Darksonn Oct 18, 2022 • edited Loading

Choose a reason for hiding this comment

Darksonn commented Oct 23, 2022

Darksonn left a comment

Choose a reason for hiding this comment

Add `SinkWriter` in analogy to `StreamReader` to `tokio_util` #5070

Add `SinkWriter` in analogy to `StreamReader` to `tokio_util` #5070

dbischof90 commented Oct 2, 2022 •

edited

Loading

Darksonn commented Oct 3, 2022 •

edited

Loading

dbischof90 commented Oct 4, 2022 •

edited

Loading

Darksonn Oct 18, 2022 •

edited

Loading