goingforstudying-ctrl commented on code in PR #2779:
URL:
https://github.com/apache/datafusion-ballista/pull/2779#discussion_r4499292382
##########
ballista/executor/src/client_pool.rs:
##########
@@ -290,8 +364,32 @@ impl BallistaClientPool for DefaultBallistaClientPool {
.await?;
let shared = Arc::new(SharedClient::new(client));
- *slot.connection.lock().unwrap() = Some(shared.clone());
- Ok(shared)
+ {
+ let inner = Arc::downgrade(&self.inner);
+ let slot_ref = Arc::downgrade(&slot);
+ let me = Arc::downgrade(&shared);
+ let hook_key = key.clone();
+ shared.set_discard_hook(Box::new(move || {
Review Comment:
Removed in b8d4305. The discard hook, the public `set_discard_hook` API, and
the race it reopened are all gone. Eviction reclaims discarded slots on its
next pass.
##########
ballista/executor/src/client_pool.rs:
##########
@@ -66,24 +68,59 @@ type IdleMap = DashMap<PoolKey, VecDeque<IdleEntry>>;
#[derive(Default)]
struct SharedSlot {
connecting: tokio::sync::Mutex<()>,
- connection: Mutex<Option<Arc<SharedClient>>>,
+ state: Mutex<SharedSlotState>,
+}
+
+#[derive(Default)]
+struct SharedSlotState {
+ connection: Option<Arc<SharedClient>>,
+ /// When the connection was last handed out by `shared`; eviction uses it
+ /// to tell a connection nobody asks for any more from an active one.
+ last_handout: Option<Instant>,
}
impl SharedSlot {
- /// The open connection, unless there is none or it was discarded.
- fn live(&self) -> Option<Arc<SharedClient>> {
- self.connection
+ /// The open connection for a new caller, unless there is none or it was
+ /// discarded, stamped as requested just now.
+ fn handout(&self) -> Option<Arc<SharedClient>> {
+ let mut state = self.state.lock().unwrap();
+ let client = state
+ .connection
+ .as_ref()
+ .filter(|client| !client.is_discarded())
+ .cloned()?;
+ state.last_handout = Some(Instant::now());
+ Some(client)
+ }
+
+ /// Whether `client` is the connection this slot currently holds. A
+ /// discard from a connection the slot has since replaced must not drop
+ /// the slot.
+ fn holds(&self, client: &Arc<SharedClient>) -> bool {
+ self.state
.lock()
.unwrap()
+ .connection
.as_ref()
- .filter(|client| !client.is_discarded())
- .cloned()
+ .is_some_and(|current| Arc::ptr_eq(current, client))
}
- /// Whether eviction can drop the slot: it holds no live connection and
- /// none is being opened.
- fn is_dead(&self) -> bool {
- self.live().is_none() && self.connecting.try_lock().is_ok()
+ /// Whether the slot survives an eviction pass: without a live
+ /// connection, only while a connect is in flight on it; with one, until
+ /// it has gone unrequested for `idle_timeout`.
+ fn survives_eviction(&self, now: Instant, idle_timeout: Duration) -> bool {
Review Comment:
The overlap is acceptable. When eviction retires a slot mid-stream, the
connection stays alive until hyper finishes the open h2 streams, so nothing is
lost. The next caller opens a fresh connection — a short-lived duplicate that
converges once the old streams drain. A usage guard would work too but adds
tracking complexity for a case that resolves itself. Docs updated in b8d4305:
the pool rustdoc now notes that shared connections expire on `idle_timeout`
(the executor's `client_ttl`).
##########
ballista/executor/src/client_pool.rs:
##########
@@ -259,17 +333,17 @@ impl BallistaClientPool for DefaultBallistaClientPool {
customize_endpoint: Option<Arc<BallistaConfigGrpcEndpoint>>,
) -> Result<Arc<SharedClient>> {
let key = (host.to_string(), port, config.clone());
- let slot = self.inner.shared.entry(key).or_default().clone();
+ let slot = self.inner.shared.entry(key.clone()).or_default().clone();
- if let Some(client) = slot.live() {
+ if let Some(client) = slot.handout() {
Review Comment:
Fixed in b8d4305. Both the fast path and the post-connect path now route
through `reattach_shared_slot()`, so an evicted slot gets reinserted before the
caller uses it.
--
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]
