goingforstudying-ctrl commented on code in PR #2779:
URL:
https://github.com/apache/datafusion-ballista/pull/2779#discussion_r4487969128
##########
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 {
+ let state = self.state.lock().unwrap();
+ let live = state
+ .connection
+ .as_ref()
+ .is_some_and(|client| !client.is_discarded());
+ if !live {
+ return self.connecting.try_lock().is_err();
+ }
+ match state.last_handout {
+ Some(last) => now.duration_since(last) < idle_timeout,
Review Comment:
Added in f47631b. The test reads one batch from a local flight server, drops
the SharedClient Arc, ages the handout and runs evict_idle; the detached stream
still reads to EOF. The last batch is only sent after eviction, so nothing was
pre-buffered.
--
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]
