[futures] introduce a new memory bounded channel#5025
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 947f0b56aa
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| value: t, | ||
| }); | ||
| } | ||
| let Ok(permit) = self.semaphore.acquire_many(size).await else { |
There was a problem hiding this comment.
When estimated_size() returns 0 (for example an empty payload implementation that reports its byte length), acquire_many(0) does not reduce the semaphore capacity but the value is still pushed into the unbounded mpsc queue. That lets producers enqueue an unlimited number of zero-estimated messages despite any channel capacity, while each queued entry still has allocation/queue overhead; consider rejecting zero estimates or charging at least one permit.
Useful? React with 👍 / 👎.
This PR introduces a new generic memory bounded channel to be used in various places across the codebase.
This channel is backed by an unbounded tokio channel under the hood with a semaphore on top (where the permits is number of bytes).
Elements of this channel need to implement the `EstimatedSize` trait. I know we have a similar trait in `restate-platform` but both
crates live at the bottom of our dependency tree, if we're comfortable of building a dependency, I think I can add a blanket implementation restate-platform to implement that trait for everything that implements the `memory::EstimatedMemorySize` trait. One annoying caveat though is that the estimated size returns u32 because that's the max size of the permit we can acquire from the semaphore.
**Benchmaks (note codex wrote those):**
Benchmarks against a normal bounded channel shows ~17% overhead in the no contention scenarios (no concurrent readers/writers):
```
Benchmarking mem_bounded/uncontended/mem_bounded: Collecting 100 samples in estimated 5.
mem_bounded/uncontended/mem_bounded
time: [575.24 µs 575.41 µs 575.58 µs]
thrpt: [17.374 Melem/s 17.379 Melem/s 17.384 Melem/s]
Benchmarking mem_bounded/uncontended/tokio_bounded: Collecting 100 samples in estimated
mem_bounded/uncontended/tokio_bounded
time: [490.96 µs 491.06 µs 491.17 µs]
thrpt: [20.360 Melem/s 20.364 Melem/s 20.368 Melem/s]
```
and that difference collapses when there are multiple concurrent readers and writers:
```
mem_bounded/pipelined/mem_bounded/1
time: [16.685 ms 16.865 ms 17.049 ms]
thrpt: [5.8654 Melem/s 5.9296 Melem/s 5.9933 Melem/s]
mem_bounded/pipelined/tokio_bounded/1
time: [14.104 ms 14.270 ms 14.434 ms]
thrpt: [6.9279 Melem/s 7.0076 Melem/s 7.0904 Melem/s]
mem_bounded/pipelined/mem_bounded/4
time: [17.914 ms 18.103 ms 18.302 ms]
thrpt: [5.4639 Melem/s 5.5240 Melem/s 5.5822 Melem/s]
mem_bounded/pipelined/tokio_bounded/4
time: [17.305 ms 17.411 ms 17.520 ms]
thrpt: [5.7078 Melem/s 5.7433 Melem/s 5.7786 Melem/s]
```
[futures] Buffer locally
1 participant
This PR introduces a new generic memory bounded channel to be used in various places across the codebase.
This channel is backed by an unbounded tokio channel under the hood with a semaphore on top (where the permits is number of bytes).
Elements of this channel need to implement the
EstimatedSizetrait. I know we have a similar trait inrestate-platformbut bothcrates live at the bottom of our dependency tree, if we're comfortable of building a dependency, I think I can add a blanket implementation restate-platform to implement that trait for everything that implements the
memory::EstimatedMemorySizetrait. One annoying caveat though is that the estimated size returns u32 because that's the max size of the permit we can acquire from the semaphore.Benchmaks (note codex wrote those):
Benchmarks against a normal bounded channel shows ~17% overhead in the no contention scenarios (no concurrent readers/writers):
and that difference collapses when there are multiple concurrent readers and writers:
[futures] Buffer locally