comphead commented on PR #2670:
URL:
https://github.com/apache/datafusion-ballista/pull/2670#issuecomment-6458001679
Thanks for the quick turnaround, the new head reads much tighter. A second
pass on `0d53dfca`, focused on simplifications. I prototyped the first group as
a patch against this head. It is not compiled or tested, so please treat it as
a sketch. I can share it if useful.
### Prototyped (untested)
1. **`SharedClient` as a flag instead of a closure.** `discard()` only sets
`discarded`, and the pool's `shared()` skips a discarded entry and reconnects.
That removes the `Box<dyn Fn>`, the `Weak<SharedSlot>`, the `generations`
counter and the two-mutex `SharedSlot`. The slot becomes one
`tokio::sync::Mutex<Option<Arc<SharedClient>>>` per key. `CountingPool` in
`tests.rs` shrinks the same way, and `open_remote_once` no longer needs
`(client, Option<shared>)`. The no-pool branch can wrap a fresh client in
`SharedClient`. One cost is that the hot path now takes an async lock per open
instead of a sync one, which probably does not matter but I have not measured
it.
2. **Registry slot on `tokio::sync::watch`.**
`ProducerSlot::{Waiting(Vec<oneshot::Sender>), Ready}` becomes
`watch::Sender<Option<Arc<ProducerOutputs>>>`. That drops the waiter list, the
`is_closed()` retain and the double match in `register` and `wait_for`.
3. **Broadcast replay on `watch`.** `Notify` plus `Mutex<ReplayState>`
becomes `watch::Sender<ReplayState>`. This removes the "register for a wakeup
before looking" protocol, the extra `Arc` handle, the `done` flag and
`.boxed()`, and `replay_stream` roughly halves.
4. **`failure` as `OnceLock<String>`** instead of `Mutex<Option<String>>`.
5. **Writer: coalescer inside `Output`.** `Output::write` and
`Output::finish` replace the parallel `coalescers` vector, the paired `.get()`
with its Internal error, `drain` and `Outputs::send(output, ..)`. The `open`
counter is derivable, so `all_closed()` can be `slots.iter().all(..)`. That is
O(P) per input batch instead of O(1), and the `write_time` timer would now also
cover coalescing. Both look negligible next to the hash and `take`, but I have
not measured.
6. `shared_connection_config` takes `ballista_config` twice and can take it
once.
Net for the prototype is about 190 added and 250 removed lines across 6
files.
### Proposed, not prototyped
- Drop `registry: Option<Arc<PipelinedRegistry>>` and `with_registry` from
the writer and reader, and use the session extension only. That removes two
fields, two setters and the re-apply branches in `with_new_children` and
`with_partitions`, at the cost of attaching the registry through
`SessionConfig` in about ten test call sites. `with_client_pool` can stay,
since `ShuffleReaderExec` already uses that pattern for a `dyn` pool.
- Merge `closed_jobs` and `closed_stages` into one bounded map keyed by
`JobId`, each entry holding an `all` flag and a stage set. That gives one
capacity constant, and `closed()` no longer clones a `JobId` into a tuple key
on every `register` and `open`.
- With (1), evicting discarded or never-connected shared slots in
`evict_idle` is a single `retain`, which would cover most of #2673 cheaply.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
