refactor(`ipc`): use single buffer and remove manual wakers #69

mattsse · 2023-12-11T20:06:55Z

refactors the ReadJsonStream:

yield decoded items immediately
no manual wakers
remove BufReader and use single buffer

Evalir

this simplifies things, so lgtm. deferring to @prestwich for merge

Evalir · 2023-12-11T20:41:43Z

crates/transport-ipc/src/lib.rs

@@ -113,19 +112,17 @@ impl IpcBackend {
 pub struct ReadJsonStream<T> {
    /// The underlying reader.
    #[pin]
-    reader: BufReader<T>,
-    /// A buffer of bytes read from the reader.
+    reader: tokio_util::compat::Compat<T>,


TIL, although, what a name lol. I wasted some time looking for something like this 😅

!!! I was looking for that! i found the old 0.3 compat crate but couldn't find the new one

I previously wasted a few hours on this myself -.-

same with finding a properly maintained tokio::codec Json implementation.
using SerializeStream is an adequate solution I think

prestwich · 2023-12-11T20:09:21Z

crates/transport-ipc/src/lib.rs

            match item {
                Some(Ok(response)) => {
-                    this.items.push(response);
+                    return Ready(Some(response));


wont this now fail to wake up again until the socket has more data?

I guess the thing here is that you assume the read has EXACTLY ONE item in it. but that may not be the case if geth has sent multiple notifications

oh then i guess the followup would be that the underlying stream re-wakes the task as there's more data?

wont this now fail to wake up again until the socket has more data?

this yields a new value, the caller is supposed to call poll_next again, streams should only register the waker if they are pending

https://docs.rs/futures/latest/futures/stream/trait.Stream.html#tymethod.poll_next

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

the way this loop works is:

it drains the buffer by yielding deserialized values (advancing the buffer)
if there are no more items to serialize it fills the buffer by reading from the reader until pending

wont this now fail to wake up again until the socket has more data?

this yields a new value, the caller is supposed to call poll_next again, streams should only register the waker if they are pending

this doesn't mesh with my understanding of my understanding of wakers. AIUI the waker exists to notify the executor that there is additional work (i.e. that something that made us pending has become unblocked), and the executor may never poll a task that does not have a call to Waker::wake or Waker::wake_by_ref. E.g. the example executor would not re-poll this future until the waker stored in the underlying IO re-wakes the task because more data is available to read. So the observed behavior with that executor would be that any time 2 responses are in the buffer, the 2nd one is not read until the ipc socket has even more data available (and that data would not be read until more is available)

when they say "register for waking" I'm pretty sure what they mean is "store a copy of the waker in the local state. this is how, e.g., the tokio::time::sleep works under the hood. it stores the waker from the context of any task passed into it.

as a side note, your code may work in prod because tokio has an optimization where it polls futures again immediately after they return ready (even if the waker is not triggered), in case there's more work. While it may work, I'm not sure it is correct

tokio does not poll the future directly, the user of the stream does by calling poll_next. This happens in a Future, or poll function

the executor deals with the top-level task which contains the stream state machine, and any futures or other streams or other structs that contain that stream, and schedules that task for execution. futures and streams use the wakers to communicate to the executor that the task should be scheduled.

while let Some(next) = stream.next().await

this is a busy loop if the stream is always ready, yes. but as soon as the stream returns pending and the executor switches tasks, there is no guarantee that the executor returns to task before the task's Waker is used to re-wake it.

the implicit "guarantee" that poll_next registers the waker only exists if the stream is pending, per docs.

For example: mpsc::UnboundedReceiver Stream:

https://docs.rs/futures-channel/0.3.29/src/futures_channel/mpsc/mod.rs.html#1084-1090

this is "registering" the waker by adding it to task state to be re-woken when the channel has more items, ensuring that the executor calls poll when items are available, as i described above. ctx.waker().wake_by_ref() is not registering the waker in this way, it is requesting that the task be polled again by the executor immediately

ctx.waker().wake_by_ref() is not registering the waker in this way, it is requesting that the task be polled again by the executor immediately

right, that's why it isn't necessary to call it when returning an item

it is necessary if our buffer isn't empty, as we have work to do immediately, which wont get done if the task isnt rewoken

the new poll loop has 4 exits:

item decoded -> return the item

decoding failed -> terminate

no item decoded, nothing read from the reader -> pending (this ensures we're getting called again as soon as there's more to read)

reading failed -> terminate

I fail to see where we'd need to request another call.
But after writing this I think we can optimize this a bit more and skip the decoding attempt one time when we previously returned pending

if we read every item as soon as it's written, then we don't have a problem

however, if the other end of the socket ever writes 2 items between polls of this stream, thne then next time the stream is polled it will fill the buffer (getting the 2 items), and then deser and return an item. At that point, we have a serialized item in the buffer. If we do not re-wake properly, then the next poll will occur when the asyncread is ready to give us more serialized items for our buffer. That wake will again result in only 1 item being read from the buffer

we need to re-wake until the buffer is empty, otherwise we risk falling significantly behind and filling our buffer

prestwich · 2023-12-12T00:10:33Z

crates/transport-ipc/src/lib.rs

@@ -113,19 +112,17 @@ impl IpcBackend {
 pub struct ReadJsonStream<T> {
    /// The underlying reader.
    #[pin]
-    reader: BufReader<T>,
-    /// A buffer of bytes read from the reader.
+    reader: tokio_util::compat::Compat<T>,


!!! I was looking for that! i found the old 0.3 compat crate but couldn't find the new one

prestwich · 2023-12-12T18:28:12Z

crates/transport-ipc/src/lib.rs


+        loop {
+            // try decoding from the buffer
            tracing::debug!(buf_len = this.buf.len(), "Deserializing buffered IPC data");
            let mut de = serde_json::Deserializer::from_slice(this.buf.as_ref()).into_iter();


instantiating the deserializer performs a new alloc every time the poll is invoked. instead, let's check if the buffer is empty before instantiating, and then after instantiating, drain the whole buffer into an items: Vec<PubSubItem>. Will reduce total allocs significantly

added a toggle drained that is set when we have more data/no more items

use single buffer and remove manual wakers

c804e47

mattsse requested review from DaniPopes, Evalir, gakonst and prestwich as code owners December 11, 2023 20:06

Evalir changed the title ~~use single buffer and remove manual wakers~~ feat(ipc): use single buffer and remove manual wakers Dec 11, 2023

Evalir changed the title ~~feat(ipc): use single buffer and remove manual wakers~~ refactor(ipc): use single buffer and remove manual wakers Dec 11, 2023

Evalir approved these changes Dec 11, 2023

View reviewed changes

prestwich reviewed Dec 12, 2023

View reviewed changes

mattsse added 2 commits December 12, 2023 20:14

Merge branch 'main' into matt/work-on-ipc-reader

699647d

perf: use marker for decoding

f4e0802

prestwich approved these changes Dec 12, 2023

View reviewed changes

mattsse merged commit 51ee830 into main Dec 12, 2023
17 checks passed

DaniPopes deleted the matt/work-on-ipc-reader branch January 15, 2024 18:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(`ipc`): use single buffer and remove manual wakers #69

refactor(`ipc`): use single buffer and remove manual wakers #69

mattsse commented Dec 11, 2023

Evalir left a comment

Evalir Dec 11, 2023 •

edited

Loading

prestwich Dec 12, 2023

mattsse Dec 12, 2023

prestwich Dec 11, 2023

prestwich Dec 12, 2023

prestwich Dec 12, 2023

mattsse Dec 12, 2023 •

edited

Loading

prestwich Dec 12, 2023 •

edited

Loading

prestwich Dec 12, 2023

mattsse Dec 12, 2023

prestwich Dec 12, 2023

mattsse Dec 12, 2023 •

edited

Loading

prestwich Dec 12, 2023 •

edited

Loading

prestwich Dec 12, 2023

prestwich Dec 12, 2023

mattsse Dec 12, 2023

refactor(ipc): use single buffer and remove manual wakers #69

refactor(ipc): use single buffer and remove manual wakers #69

Conversation

mattsse commented Dec 11, 2023

Evalir left a comment

Choose a reason for hiding this comment

Evalir Dec 11, 2023 • edited Loading

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

mattsse Dec 12, 2023 • edited Loading

Choose a reason for hiding this comment

prestwich Dec 12, 2023 • edited Loading

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

mattsse Dec 12, 2023 • edited Loading

Choose a reason for hiding this comment

prestwich Dec 12, 2023 • edited Loading

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

refactor(`ipc`): use single buffer and remove manual wakers #69

refactor(`ipc`): use single buffer and remove manual wakers #69

Evalir Dec 11, 2023 •

edited

Loading

mattsse Dec 12, 2023 •

edited

Loading

prestwich Dec 12, 2023 •

edited

Loading

mattsse Dec 12, 2023 •

edited

Loading

prestwich Dec 12, 2023 •

edited

Loading