GitHub user jrt1899 added a comment to the discussion: Parallelism bottleneck
in multi-UDF workflows
### Update: implementation is up for review
I've opened #8858 to track this and split the least-loaded approach (Design 2
above) into three stacked PRs. It is a new, opt-in partitioning:
`network-buffering.enable-least-loaded-routing`, default off.
`RoundRobinPartitioner` is not modified, and only links with no partitioning
requirement (the ones that use round robin today) are affected. Hash, range,
broadcast and one-to-one links are unchanged.
```mermaid
flowchart LR
subgraph RR["Round robin (today): n × m buffers"]
direction LR
S1["UDF1 worker"] --> B1["buffer → r1"] --> R1["UDF2 r1"]
S1 --> B2["buffer → r2"] --> R2["UDF2 r2"]
S1 --> B3["buffer → rm"] --> R3["UDF2 rm"]
end
subgraph LL["Least-loaded (new): n buffers"]
direction LR
S2["UDF1 worker"] --> B4["single buffer"] -->|"least backlog"| R4["UDF2 r*"]
end
```
**#8862: `LeastLoadedPartitioner` and backlog statistic** (inert)
- New `LeastLoadedPartitioning` / `LeastLoadedPartitioner` in Scala and Python.
Each sender fills one DTB-sized batch at a time and sends it to a single
preferred receiver.
- Workers report `received_tuple_count` on arrival, so received − processed
gives each receiver's backlog.
- Senders start on different receivers, so early batches are spread out before
the first ranking arrives.
**#8863: coordinator ranking and `UpdateRoutingPreference` RPC** (inert)
- On each statistics poll, the coordinator ranks receivers by backlog and
spreads senders across the least-loaded ones.
- A new lightweight RPC moves the sender's target index without rebuilding the
partitioner, so no partly filled batch is thrown away.
- Adds the config flag (default off) and a verbose logging flag.
**#8865: opt-in switch**
- With the flag on, links with no partitioning requirement use
`LeastLoadedPartitioning` instead of `RoundRobinPartitioning`.
| | Round robin (today) | Least-loaded (new) |
|---|---|---|
| Buffers on the link | n × m | n |
| Tuples before the first batch ships | up to n × m × DTB | DTB per sender |
| Receiver chosen | fixed rotation | smallest backlog, updated each poll |
| Batch size shipped | DTB | DTB (unchanged) |
**Next:** benchmark numbers on this implementation (the 34.3% above came from
an earlier prototype). The PRs build on each other, so #8862 is the place to
start reviewing.
GitHub link:
https://github.com/apache/texera/discussions/8519#discussioncomment-18762614
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
