You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "hubcio (via GitHub)" <gi...@apache.org> on 2026/10/27 07:28:05 UTC
[PR] feat(message_bus): spread replica traffic over k links per peer [iggy]
hubcio opened a new pull request, #4656:
URL: https://github.com/apache/iggy/pull/4656
Each pair of nodes shared one replica link, and one shard per node
owned it. Every other shard forwarded its consensus frames through
that shard. In September OMB runs with 100 topics, one node led every
partition, and its link shard used a full core. In October bench runs
on AWS, each replica link carried at most about 1.2 GB/s. If that cap
holds per TCP connection, k links raise it.
cluster.coordinator.replica_connections_per_peer sets k. Link j
carries the frames of a contiguous range of shards, and on each node
one shard of that range owns it. Frames for the metadata group take
link 0. No benchmark used k > 1 yet, so the default stays 1.
Before TLS and the hello, the dialer sends an 8-byte preamble with its
replica id and the link. Shard 0 of the acceptor reads only those
bytes and hands the socket to the owner of the link. The hello carries
the shard count, k and the link, and with auth on the MAC covers them.
The owner refuses a hello that does not match the preamble or its own
values.
Frames of a shard range have only their own link, because a fallback
to another link can reorder them. So shard 0 tracks the k links to a
peer as one bundle. A node sends to that peer only while its bundle is
up. If a link of the bundle closes, or the bundle stays incomplete for
message_bus.handshake_grace, shard 0 closes all k. The node with the
lower id then dials them again. A failed handshake changes no bundle
on its node, because an inbound hello is not authenticated before it
completes.
A peer that died without a FIN or RST left a half-open link. VSR
heartbeats told consensus that the peer was gone, but nothing closed
the socket. The link kept its owner slot, so the reconnect sweep
skipped the peer. With Linux defaults, the kernel closed the link only
after at least 15 minutes of retransmits. Replica sockets now get
TCP_USER_TIMEOUT on Linux, equal to cluster.heartbeat_timeout. From
Linux 5.11 it also covers a zero window, so a peer that stops reading
loses its link too.
Every link sends the preamble, also at k = 1. So a node with this
change and a node without it never link, even on the same release. An
upgrade must restart every node at once. Nodes with another shard
count or k refuse each other, also at k = 1. Before, nodes with
different shard counts linked. The default sharding.cpu_allocation
takes the shard count from the cores of the host, so hosts with
different core counts need an explicit, equal shard count.
Shard metrics now count bundles that are up, teardowns, preamble
refusals, handshake failures and frames dropped at a full peer queue.
The connection cache had no users, so this change deletes it and rand,
which only the cache used in message_bus.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(message_bus): spread replica traffic over k links per peer [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4656:
URL: https://github.com/apache/iggy/pull/4656#issuecomment-6438652208
agreed, it breaks the wire at every k. added an upgrading note to `core/server/README.md` in 18691f9b4, and the title gets `!` plus a `BREAKING CHANGE:` footer. the shard count part is gone, see the handshake thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(message_bus)!: spread replica traffic over k links per peer [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #4656:
URL: https://github.com/apache/iggy/pull/4656
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(message_bus): spread replica traffic over k links per peer [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4656:
URL: https://github.com/apache/iggy/pull/4656#issuecomment-6437015545
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4656?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `66.87898%` with `468 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.35%. Comparing base ([`dc4006d`](https://app.codecov.io/gh/apache/iggy/commit/dc4006d92f0565dd0ae26e842a3433b51e3c3f1b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`0cdd345`](https://app.codecov.io/gh/apache/iggy/commit/0cdd34587bfa433cf18731b445f46bd753e28184?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4656?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [core/message\_bus/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvbGliLnJz) | 67.68% | [139 Missing and 9 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/message\_bus/src/replica/links.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Freplica%2Flinks.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvcmVwbGljYS9saW5rcy5ycw==) | 48.30% | [61 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/message\_bus/src/replica/bundle.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Freplica%2Fbundle.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvcmVwbGljYS9idW5kbGUucnM=) | 70.12% | [46 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/message\_bus/src/replica/auth.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Freplica%2Fauth.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvcmVwbGljYS9hdXRoLnJz) | 42.66% | [43 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/shard/src/coordinator.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fshard%2Fsrc%2Fcoordinator.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaGFyZC9zcmMvY29vcmRpbmF0b3IucnM=) | 46.83% | [39 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/message\_bus/src/replica/preamble.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Freplica%2Fpreamble.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvcmVwbGljYS9wcmVhbWJsZS5ycw==) | 63.33% | [31 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/shard/src/metrics.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fshard%2Fsrc%2Fmetrics.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaGFyZC9zcmMvbWV0cmljcy5ycw==) | 61.76% | [26 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/message\_bus/src/installer/replica.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Finstaller%2Freplica.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvaW5zdGFsbGVyL3JlcGxpY2EucnM=) | 88.39% | [10 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/shard/src/partition\_io.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fshard%2Fsrc%2Fpartition_io.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaGFyZC9zcmMvcGFydGl0aW9uX2lvLnJz) | 33.33% | [12 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/server\_common/src/consensus\_message.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fserver_common%2Fsrc%2Fconsensus_message.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXJfY29tbW9uL3NyYy9jb25zZW5zdXNfbWVzc2FnZS5ycw==) | 35.71% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [9 more](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
> :exclamation: There is a different number of reports uploaded between BASE (dc4006d) and HEAD (0cdd345). Click for more details.
>
> <details><summary>HEAD has 2 uploads less than BASE</summary>
>
>| Flag | BASE (dc4006d) | HEAD (0cdd345) |
>|------|------|------|
>|rust|3|1|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4656 +/- ##
=============================================
- Coverage 80.89% 58.35% -22.55%
Complexity 3104 3104
=============================================
Files 1465 1468 +3
Lines 356188 356944 +756
Branches 300421 301179 +758
=============================================
- Hits 288155 208294 -79861
- Misses 61202 141861 +80659
+ Partials 6831 6789 -42
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.00% <66.87%> (-27.16%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.21% <ø> (+0.01%)` | :arrow_up: |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.56% <ø> (-0.14%)` | :arrow_down: |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.67% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.71% <ø> (+0.04%)` | :arrow_up: |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.44% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.05% <ø> (+0.21%)` | :arrow_up: |
| [C++ SDK](https://app.codecov.io/gh/apache/iggy/pull/4656/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.51% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4656?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/configs/src/server\_config/cluster.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fconfigs%2Fsrc%2Fserver_config%2Fcluster.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25maWdzL3NyYy9zZXJ2ZXJfY29uZmlnL2NsdXN0ZXIucnM=) | `54.26% <100.00%> (-29.88%)` | :arrow_down: |
| [core/message\_bus/src/config.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Fconfig.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvY29uZmlnLnJz) | `78.16% <100.00%> (+13.04%)` | :arrow_up: |
| [core/message\_bus/src/installer/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Finstaller%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvaW5zdGFsbGVyL21vZC5ycw==) | `96.82% <100.00%> (-1.47%)` | :arrow_down: |
| [...e/message\_bus/src/lifecycle/connection\_registry.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Flifecycle%2Fconnection_registry.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvbGlmZWN5Y2xlL2Nvbm5lY3Rpb25fcmVnaXN0cnkucnM=) | `52.31% <100.00%> (-27.44%)` | :arrow_down: |
| [core/message\_bus/src/replica/handshake.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fmessage_bus%2Fsrc%2Freplica%2Fhandshake.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXNzYWdlX2J1cy9zcmMvcmVwbGljYS9oYW5kc2hha2UucnM=) | `74.31% <100.00%> (-7.69%)` | :arrow_down: |
| [core/server/src/boot/recovery.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Frecovery.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvcmVjb3ZlcnkucnM=) | `61.71% <ø> (-22.97%)` | :arrow_down: |
| [core/server/src/boot/topology.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Ftopology.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvdG9wb2xvZ3kucnM=) | `64.06% <100.00%> (-21.11%)` | :arrow_down: |
| [core/server/src/server\_error.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fserver_error.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL3NlcnZlcl9lcnJvci5ycw==) | `14.51% <ø> (-69.36%)` | :arrow_down: |
| [core/shard/src/config.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fshard%2Fsrc%2Fconfig.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaGFyZC9zcmMvY29uZmlnLnJz) | `100.00% <ø> (ø)` | |
| [core/shard/src/router.rs](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree&filepath=core%2Fshard%2Fsrc%2Frouter.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaGFyZC9zcmMvcm91dGVyLnJz) | `64.16% <100.00%> (-13.55%)` | :arrow_down: |
| ... and [20 more](https://app.codecov.io/gh/apache/iggy/pull/4656?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [606 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4656/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(message_bus): spread replica traffic over k links per peer [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4656:
URL: https://github.com/apache/iggy/pull/4656#discussion_r4386292382
##########
core/message_bus/src/replica/handshake.rs:
##########
@@ -201,6 +214,34 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>(
)
.await;
}
+ let announced = LinkPlacement::read(&header.reserved_command);
+ let placement_status = if announced.shards != placement.shards {
Review Comment:
This refuses a peer whose shard count differs, but nothing else reads the remote count. Each node places its links from its own `LinkLayout`, the acceptor picks the owner from its own layout (`core/message_bus/src/lib.rs:1155`), and the receiving shard hands every frame to `shard.dispatch`, which picks the target with the local shard count (`core/shard/src/router.rs:173`). The sender picks the link from its own shard, so ordering does not depend on the remote count either, and at k = 1 `range_of` is 0 for every shard. The default `cpu_allocation = "numa:auto"` takes the shard count from the host cores, so a cluster on mixed instance sizes that links today stops linking after this change, and the only operator doc that says so is the `replica_connections_per_peer` comment in `core/server/config.toml`.
##########
core/message_bus/src/replica/links.rs:
##########
@@ -0,0 +1,278 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Placement of the k replica links to each peer.
+//!
+//! Shard `s` belongs to range `s * k / shards`, and range `j` sends over
+//! link `j`. Link `j` to one peer lives on one usable shard of range `j`,
+//! picked by the peer slot. Both ends derive placement from configuration
+//! alone, and a consensus group has the same shard index on every node, so
+//! a group frame leaves from and arrives in the same range.
+
+use crate::MAX_INFLIGHT_REPLICA_HANDSHAKES;
+use thiserror::Error;
+
+/// Shard count, links per peer and the shard-0 skip flag of one node.
+///
+/// Every node of a cluster must run the same shard count and link count:
+/// the replica handshake refuses a peer whose values differ.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct LinkLayout {
+ shards: u16,
+ links: u16,
+ skip_shard_zero: bool,
+}
+
+/// A link count the node cannot run.
+#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
+pub enum LinkLayoutError {
+ #[error("replica_connections_per_peer must be at least 1")]
+ NoLinks,
+ #[error("replica_connections_per_peer = {links} exceeds the shard count {shards}")]
+ MoreLinksThanShards { links: u16, shards: u16 },
+ #[error(
+ "replica_connections_per_peer = {links} with {peers} peers opens more than \
+ {cap} links, the inbound handshake cap",
+ cap = MAX_INFLIGHT_REPLICA_HANDSHAKES
+ )]
+ TooManyLinks { links: u16, peers: usize },
+}
+
+impl LinkLayout {
+ /// # Errors
+ ///
+ /// Refuses no links, more links than shards, and more links in total
+ /// than [`MAX_INFLIGHT_REPLICA_HANDSHAKES`], which would let a full
+ /// redial of every peer fill the inbound handshake cap.
+ pub fn new(
+ shards: u16,
+ links: u16,
+ skip_shard_zero: bool,
+ peers: usize,
+ ) -> Result<Self, LinkLayoutError> {
+ if links == 0 {
+ return Err(LinkLayoutError::NoLinks);
+ }
+ if links > shards {
+ return Err(LinkLayoutError::MoreLinksThanShards { links, shards });
+ }
+ if usize::from(links) * peers > MAX_INFLIGHT_REPLICA_HANDSHAKES {
+ return Err(LinkLayoutError::TooManyLinks { links, peers });
+ }
+ Ok(Self {
+ shards,
+ links,
+ skip_shard_zero,
+ })
+ }
+
+ #[must_use]
+ pub const fn shards(self) -> u16 {
+ self.shards
+ }
+
+ #[must_use]
+ pub const fn links(self) -> u16 {
+ self.links
+ }
+
+ /// The link that `shard` sends over.
+ ///
+ /// A shard id at or above the shard count, which only test buses use,
+ /// falls into the last range.
+ #[must_use]
+ pub fn range_of(self, shard: u16) -> u16 {
+ let range = u32::from(shard) * u32::from(self.links) / u32::from(self.shards);
+ u16::try_from(range).map_or(self.links - 1, |range| range.min(self.links - 1))
+ }
+
+ /// The shard that owns `link` to the peer at `peer_slot`.
+ ///
+ /// Shard 0 hosts the listener, the dialer and the metadata group, so the
+ /// skip flag keeps links off it while its range has another shard.
+ #[must_use]
+ pub fn link_owner(self, link: u16, peer_slot: u8) -> u16 {
+ debug_assert!(link < self.links, "link {link} of {}", self.links);
+ let first = self.first_of_range(link);
+ let end = self.first_of_range(link + 1);
+ let first_usable = if first == 0 && self.skip_shard_zero && end > 1 {
Review Comment:
With the skip flag on (the default), link 0 to every peer lands on shard 1 whenever range 0 holds two shards, so at small shard counts k > 1 leaves the busiest link shard as loaded as before. On 4 shards with k = 2 and 3 nodes, range 0 is {0, 1}, `first_usable` is 1 and `end - first_usable` is 1, so shard 1 owns link 0 to both peers and moves the traffic of two shards to each of them, the same load it carried at k = 1 as the owner of one link with all four shards. Shards 2 and 3 carry half that. On 8 shards with k = 4 range 0 is {0, 1} again, and the hot shard drops to a half instead of a quarter. Partitions hash over every shard including 0 (`core/shard/src/shards_table.rs:131`), so range 0 carries full traffic with one usable owner.
##########
core/message_bus/src/lib.rs:
##########
@@ -979,42 +1077,261 @@ impl IggyMessageBus {
.is_some()
}
- /// Record that an outbound connection to `replica_id` was delegated
- /// and its dialer handshake is in flight on the owning shard. The
- /// reconnect sweep skips pending peers; the entry is cleared by the
+ /// Record that link `link` to `replica_id` was dialed and delegated and
+ /// its dialer handshake is in flight on the owning shard. The
+ /// reconnect sweep skips pending links; the entry is cleared by the
/// handshake-outcome ack ([`Self::clear_dial_pending`]) or expires
/// on its own after `handshake_grace` plus margin.
- pub fn mark_dial_pending(&self, replica_id: u8) {
+ pub fn mark_dial_pending(&self, replica_id: u8, link: u16) {
self.pending_dials
.borrow_mut()
- .insert(replica_id, self.replica_handshake_deadline());
+ .insert((replica_id, link), self.replica_handshake_deadline());
}
- /// Whether an outbound handshake to `replica_id` is still in flight.
- /// An expired entry is removed here (lazy expiry) and reported as
- /// not pending, re-enabling the redial.
+ /// Whether an outbound handshake of link `link` to `replica_id` is
+ /// still in flight. An expired entry is removed here (lazy expiry) and
+ /// reported as not pending, re-enabling the redial.
#[must_use]
- pub fn check_dial_pending(&self, replica_id: u8) -> bool {
+ pub fn check_dial_pending(&self, replica_id: u8, link: u16) -> bool {
let mut pending = self.pending_dials.borrow_mut();
- match pending.get(&replica_id) {
+ match pending.get(&(replica_id, link)) {
Some(deadline) if *deadline > Instant::now() => true,
Some(_) => {
- pending.remove(&replica_id);
+ pending.remove(&(replica_id, link));
false
}
None => false,
}
}
- /// Clear the pending-dial entry for `replica_id`. Idempotent; returns
- /// whether an entry was present.
- pub fn clear_dial_pending(&self, replica_id: u8) -> bool {
+ /// Clear the pending-dial entry of link `link` to `replica_id`.
+ /// Idempotent; returns whether an entry was present.
+ pub fn clear_dial_pending(&self, replica_id: u8, link: u16) -> bool {
self.pending_dials
.borrow_mut()
- .remove(&replica_id)
+ .remove(&(replica_id, link))
.is_some()
}
+ /// Shard 0: read the placement preamble of an accepted replica socket
+ /// and place its link. Returns the link, in the current epoch of the
+ /// dialer's bundle, and the shard that owns it.
+ ///
+ /// Reads exactly the preamble bytes, within `handshake_grace`, so the
+ /// owner shard still finds the TLS `ClientHello` or the hello first.
+ ///
+ /// # Errors
+ ///
+ /// See [`PreambleError`]. The caller closes the socket.
+ #[allow(clippy::future_not_send)]
+ pub async fn place_replica_inbound(
+ &self,
+ stream: &mut compio::net::TcpStream,
+ ) -> Result<(ReplicaLinkId, u16), PreambleError> {
+ let placed = self.read_and_place_preamble(stream).await;
+ if placed.is_err() {
+ self.count_link_event(|stats| stats.preamble_refusals += 1);
+ }
+ placed
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn read_and_place_preamble(
+ &self,
+ stream: &mut compio::net::TcpStream,
+ ) -> Result<(ReplicaLinkId, u16), PreambleError> {
+ let (self_id, replica_count) = self
+ .replica_handshake_ctx()
+ .map(|ctx| (ctx.self_id, ctx.replica_count))
+ .ok_or(PreambleError::NoHandshakeContext)?;
+ let bytes = replica::preamble::read_preamble(stream, self.config.handshake_grace).await?;
Review Comment:
Shard 0 waits up to `handshake_grace` for these 8 bytes, then `spawn_replica_install` starts a fresh `timeout(grace, ...)` on the owner shard (`core/message_bus/src/installer/replica.rs:90`), so an inbound replica socket can stay in admission for two grace windows, 20 s at the default. The in-flight slot is taken before this read (`core/message_bus/src/replica/listener.rs:105`) with a deadline of one grace plus the 5 s margin, so slots of live handshakes expire and `try_acquire_replica_handshake_slot` prunes them, and more than `MAX_INFLIGHT_REPLICA_HANDSHAKES` sockets can be open at once. Both docs that rule this out are now wrong: `core/message_bus/src/installer/replica.rs:111` says a peer cannot stretch admission to two grace windows, and `core/message_bus/src/lib.rs:140` says slot expiry matters only for a lost ack. The `handshake_grace` comment in `core/server/config.toml:1245` still describes only client handshakes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(message_bus): spread replica traffic over k links per peer [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4656:
URL: https://github.com/apache/iggy/pull/4656#discussion_r4387206434
##########
core/message_bus/src/replica/handshake.rs:
##########
@@ -201,6 +214,34 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>(
)
.await;
}
+ let announced = LinkPlacement::read(&header.reserved_command);
+ let placement_status = if announced.shards != placement.shards {
Review Comment:
right, nothing reads the remote shard count. dropped the check for every k in 358523ab3: the hello carries only the link count and link index, `ShardCountMismatch` is gone and the two newer statuses moved to 7 and 8 (both new in this pr). the link count check stays, since with k = 1 vs k = 2 one side waits forever for link 1. `nodes_with_other_shard_counts_connect` covers 4 vs 2 shards.
##########
core/message_bus/src/replica/links.rs:
##########
@@ -0,0 +1,278 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Placement of the k replica links to each peer.
+//!
+//! Shard `s` belongs to range `s * k / shards`, and range `j` sends over
+//! link `j`. Link `j` to one peer lives on one usable shard of range `j`,
+//! picked by the peer slot. Both ends derive placement from configuration
+//! alone, and a consensus group has the same shard index on every node, so
+//! a group frame leaves from and arrives in the same range.
+
+use crate::MAX_INFLIGHT_REPLICA_HANDSHAKES;
+use thiserror::Error;
+
+/// Shard count, links per peer and the shard-0 skip flag of one node.
+///
+/// Every node of a cluster must run the same shard count and link count:
+/// the replica handshake refuses a peer whose values differ.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct LinkLayout {
+ shards: u16,
+ links: u16,
+ skip_shard_zero: bool,
+}
+
+/// A link count the node cannot run.
+#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
+pub enum LinkLayoutError {
+ #[error("replica_connections_per_peer must be at least 1")]
+ NoLinks,
+ #[error("replica_connections_per_peer = {links} exceeds the shard count {shards}")]
+ MoreLinksThanShards { links: u16, shards: u16 },
+ #[error(
+ "replica_connections_per_peer = {links} with {peers} peers opens more than \
+ {cap} links, the inbound handshake cap",
+ cap = MAX_INFLIGHT_REPLICA_HANDSHAKES
+ )]
+ TooManyLinks { links: u16, peers: usize },
+}
+
+impl LinkLayout {
+ /// # Errors
+ ///
+ /// Refuses no links, more links than shards, and more links in total
+ /// than [`MAX_INFLIGHT_REPLICA_HANDSHAKES`], which would let a full
+ /// redial of every peer fill the inbound handshake cap.
+ pub fn new(
+ shards: u16,
+ links: u16,
+ skip_shard_zero: bool,
+ peers: usize,
+ ) -> Result<Self, LinkLayoutError> {
+ if links == 0 {
+ return Err(LinkLayoutError::NoLinks);
+ }
+ if links > shards {
+ return Err(LinkLayoutError::MoreLinksThanShards { links, shards });
+ }
+ if usize::from(links) * peers > MAX_INFLIGHT_REPLICA_HANDSHAKES {
+ return Err(LinkLayoutError::TooManyLinks { links, peers });
+ }
+ Ok(Self {
+ shards,
+ links,
+ skip_shard_zero,
+ })
+ }
+
+ #[must_use]
+ pub const fn shards(self) -> u16 {
+ self.shards
+ }
+
+ #[must_use]
+ pub const fn links(self) -> u16 {
+ self.links
+ }
+
+ /// The link that `shard` sends over.
+ ///
+ /// A shard id at or above the shard count, which only test buses use,
+ /// falls into the last range.
+ #[must_use]
+ pub fn range_of(self, shard: u16) -> u16 {
+ let range = u32::from(shard) * u32::from(self.links) / u32::from(self.shards);
+ u16::try_from(range).map_or(self.links - 1, |range| range.min(self.links - 1))
+ }
+
+ /// The shard that owns `link` to the peer at `peer_slot`.
+ ///
+ /// Shard 0 hosts the listener, the dialer and the metadata group, so the
+ /// skip flag keeps links off it while its range has another shard.
+ #[must_use]
+ pub fn link_owner(self, link: u16, peer_slot: u8) -> u16 {
+ debug_assert!(link < self.links, "link {link} of {}", self.links);
+ let first = self.first_of_range(link);
+ let end = self.first_of_range(link + 1);
+ let first_usable = if first == 0 && self.skip_shard_zero && end > 1 {
Review Comment:
the math is right, but no owner choice fixes it there. with the skip flag, 4 shards, k = 2 and 3 nodes, each node places 4 links (2 peers x 2) on 3 usable shards, so one shard always owns two. sizing ranges over shards 1..S hits the same wall. i kept `link_owner` and documented the sizing rule at `replica_connections_per_peer` in 18691f9b4: a range needs at least as many shards as peers, and range 0 one more under the skip flag.
##########
core/message_bus/src/lib.rs:
##########
@@ -979,42 +1077,261 @@ impl IggyMessageBus {
.is_some()
}
- /// Record that an outbound connection to `replica_id` was delegated
- /// and its dialer handshake is in flight on the owning shard. The
- /// reconnect sweep skips pending peers; the entry is cleared by the
+ /// Record that link `link` to `replica_id` was dialed and delegated and
+ /// its dialer handshake is in flight on the owning shard. The
+ /// reconnect sweep skips pending links; the entry is cleared by the
/// handshake-outcome ack ([`Self::clear_dial_pending`]) or expires
/// on its own after `handshake_grace` plus margin.
- pub fn mark_dial_pending(&self, replica_id: u8) {
+ pub fn mark_dial_pending(&self, replica_id: u8, link: u16) {
self.pending_dials
.borrow_mut()
- .insert(replica_id, self.replica_handshake_deadline());
+ .insert((replica_id, link), self.replica_handshake_deadline());
}
- /// Whether an outbound handshake to `replica_id` is still in flight.
- /// An expired entry is removed here (lazy expiry) and reported as
- /// not pending, re-enabling the redial.
+ /// Whether an outbound handshake of link `link` to `replica_id` is
+ /// still in flight. An expired entry is removed here (lazy expiry) and
+ /// reported as not pending, re-enabling the redial.
#[must_use]
- pub fn check_dial_pending(&self, replica_id: u8) -> bool {
+ pub fn check_dial_pending(&self, replica_id: u8, link: u16) -> bool {
let mut pending = self.pending_dials.borrow_mut();
- match pending.get(&replica_id) {
+ match pending.get(&(replica_id, link)) {
Some(deadline) if *deadline > Instant::now() => true,
Some(_) => {
- pending.remove(&replica_id);
+ pending.remove(&(replica_id, link));
false
}
None => false,
}
}
- /// Clear the pending-dial entry for `replica_id`. Idempotent; returns
- /// whether an entry was present.
- pub fn clear_dial_pending(&self, replica_id: u8) -> bool {
+ /// Clear the pending-dial entry of link `link` to `replica_id`.
+ /// Idempotent; returns whether an entry was present.
+ pub fn clear_dial_pending(&self, replica_id: u8, link: u16) -> bool {
self.pending_dials
.borrow_mut()
- .remove(&replica_id)
+ .remove(&(replica_id, link))
.is_some()
}
+ /// Shard 0: read the placement preamble of an accepted replica socket
+ /// and place its link. Returns the link, in the current epoch of the
+ /// dialer's bundle, and the shard that owns it.
+ ///
+ /// Reads exactly the preamble bytes, within `handshake_grace`, so the
+ /// owner shard still finds the TLS `ClientHello` or the hello first.
+ ///
+ /// # Errors
+ ///
+ /// See [`PreambleError`]. The caller closes the socket.
+ #[allow(clippy::future_not_send)]
+ pub async fn place_replica_inbound(
+ &self,
+ stream: &mut compio::net::TcpStream,
+ ) -> Result<(ReplicaLinkId, u16), PreambleError> {
+ let placed = self.read_and_place_preamble(stream).await;
+ if placed.is_err() {
+ self.count_link_event(|stats| stats.preamble_refusals += 1);
+ }
+ placed
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn read_and_place_preamble(
+ &self,
+ stream: &mut compio::net::TcpStream,
+ ) -> Result<(ReplicaLinkId, u16), PreambleError> {
+ let (self_id, replica_count) = self
+ .replica_handshake_ctx()
+ .map(|ctx| (ctx.self_id, ctx.replica_count))
+ .ok_or(PreambleError::NoHandshakeContext)?;
+ let bytes = replica::preamble::read_preamble(stream, self.config.handshake_grace).await?;
Review Comment:
good catch, fixed in 6229e2094. shard 0 sets one deadline when it takes the slot, and the preamble read and the owner shard handshake both end by it, so admission fits in one grace and the slot's extra 5 s only covers a lost ack. `acceptor_closes_a_slow_link_one_grace_after_connect` sends the preamble at 0.8 grace and fails on the old code (closed at 1.8 s).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org