goingforstudying-ctrl commented on code in PR #2779:
URL:
https://github.com/apache/datafusion-ballista/pull/2779#discussion_r4487970123
##########
ballista/executor/src/client_pool.rs:
##########
@@ -183,7 +219,40 @@ fn evict(inner: &Inner) {
}
!deque.is_empty()
});
- inner.shared.retain(|_, slot| !slot.is_dead());
+ inner
+ .shared
+ .retain(|_, slot| slot.survives_eviction(now, inner.idle_timeout));
+}
+
+/// Re-attach `slot`, carrying the connection `ours` just opened, to the map.
+/// Eviction can drop the slot while its `shared` caller was still
+/// connecting; without this the fresh connection would be invisible to the
+/// next caller, which would open another one. When a different slot has
+/// already reconnected, defer to its connection and let `ours` close on
+/// drop instead.
+fn reattach_shared_slot(
+ inner: &Inner,
+ key: &PoolKey,
+ slot: Arc<SharedSlot>,
+ ours: Arc<SharedClient>,
+) -> Arc<SharedClient> {
+ match inner.shared.entry(key.clone()) {
+ Entry::Occupied(mut e) => {
+ if Arc::ptr_eq(e.get(), &slot) {
+ ours
+ } else if let Some(other) = e.get().handout() {
Review Comment:
Reworded the comment in f47631b: callers holding the detached slot keep
using `ours` and it closes once they and their in-flight streams finish. Left
the defer alone since the extra short-lived connection is harmless.
##########
ballista/core/src/client_pool.rs:
##########
@@ -108,11 +113,21 @@ impl SharedClient {
self.client.clone()
}
+ /// Installs the hook the first [discard](Self::discard) runs. Discards
+ /// after the first do not run it again.
+ pub fn set_discard_hook(&self, hook: Box<dyn FnOnce() + Send + Sync>) {
+ *self.on_discard.lock().unwrap() = Some(hook);
+ }
+
/// Marks the connection as gone, after a failure that suggests it is, so
/// that the pool opens a new one for the next caller. Callers already on
/// it keep it until they are done.
pub fn discard(&self) {
- self.discarded.store(true, Ordering::Release);
+ if !self.discarded.swap(true, Ordering::AcqRel)
+ && let Some(hook) = self.on_discard.lock().unwrap().take()
+ {
+ hook();
Review Comment:
Done in f47631b. The hook is taken out of the lock first now, and the new
unit test pins it: the hook asserts try_lock succeeds while it runs.
--
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]
