feat(s2n-quic-platform): add socket tasks for sync rings #1789
Conversation
camshaft
force-pushed
the
camshaft/socket-task
branch
2 times, most recently
from
132f7af
to
4cc0c3d
Compare
camshaft
force-pushed
the
camshaft/socket-task
branch
from
4cc0c3d
to
2d265e4
Compare
| } | ||
| Interrupted => { | ||
| // if we got interrupted break and have the task try again | ||
| ControlFlow::Break(()) |
There was a problem hiding this comment.
I find it a bit odd that we should break if we want to re-try, rather than continue. I think the trait needs better documentation for what callers do on Continue vs. Break.
There was a problem hiding this comment.
yeah needs better docs
| } | ||
| _ => { | ||
| // ignore all other errors and have the task try again | ||
| ControlFlow::Break(()) |
There was a problem hiding this comment.
Here we break, but in tx on_error we continue. Why the difference? (Maybe this is explained by the docs requested in the previous comment)
| pub struct Receiver<T: Message, S: Socket<T>> { | ||
| ring: Producer<T>, | ||
| rx: S, | ||
| pending: u32, |
There was a problem hiding this comment.
I think this is the number of slots in the producer ring which we have already filled (but not yet release()'d). Some of the comments/names feel a bit confusing though in relation to that; comments left below.
| loop { | ||
| let count = ready!(self.ring.poll_acquire(watermark, cx)); | ||
|
|
||
| // if the number of items increased since last time then yield |
There was a problem hiding this comment.
This is not the number of items, but rather the spare capacity -- we're not acquiring "things" here, but rather "empty slots".
| let mut events = events::RxEvents::default(); | ||
|
|
||
| while !events.take_blocked() { | ||
| let pending = match ready!(this.poll_ring(u32::MAX, cx)) { |
There was a problem hiding this comment.
This pending is always this.pending -- I wonder if this code would be clearer if we avoided retrieving it from the function here? Polling the ring to figure out how much we've put in it (but only in our private area, not yet release'd) doesn't make sense to me, because that information isn't related to the ring doing anything, it's purely our data at this point. Really the poll here is asking "did you free up more capacity since we last asked", i.e., when we do this.ring.data()[pending..] is that going to be empty. Or at least that's my mental model?
There was a problem hiding this comment.
Really the poll here is asking "did you free up more capacity since we last asked", i.e., when we do this.ring.data()[pending..] is that going to be empty. Or at least that's my mental model?
yeah that's correct. are you suggesting the poll_ring function just return Option<()> and using this.pending directly?
There was a problem hiding this comment.
Yeah, I think so. Or bool.
| @@ -0,0 +1,106 @@ | |||
| // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | |||
There was a problem hiding this comment.
I suspect that much of the rx side applies here as well - let's discuss and figure that out and then we can talk about tx if needed separately.
There was a problem hiding this comment.
Yeah they're both very similar. I can apply the same feedback before reviewing.
camshaft
force-pushed
the
camshaft/socket-task
branch
from
2d265e4
to
e884483
Compare
Description of changes:
Using the socket ring buffer added in #1787, this PR implements async tasks that interact with the syscalls and either send or receive messages into the ring buffer. Note that the tasks are generic over the message/syscall type.
Call-outs:
The implementation that interacts with the s2n-quic endpoint is in #1788.
Testing:
In this PR, I don't have explicit tests for these implementations, since it requires a task on the other side of the ring doing something with it. In the final PR, there will be some tests that show it all working end-to-end.
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.