Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/6508#discussion_r31487214
--- Diff:
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala ---
@@ -443,9 +466,27 @@ private[spark] class ExecutorAllocationManager(
private def onExecutorIdle(executorId: String): Unit = synchronized {
if (executorIds.contains(executorId)) {
if (!removeTimes.contains(executorId) &&
!executorsPendingToRemove.contains(executorId)) {
+
+ val hasCachedBlocks =
+ executorsWithCachedBlocks.contains(executorId) ||
+
executorEndpoints.get(executorId).exists(_.askWithRetry[Boolean](HasCachedBlocks))
+
+ if (hasCachedBlocks) executorsWithCachedBlocks += executorId
--- End diff --
It's not just taking it out of the synchronized block, is it?
If I follow the code correctly, this will be executed in the listener bus
event dispatcher thread; the RPC being made here is synchronous (note:
completely unrelated to being in a "synchronized" block), so for every idle
executor there will be, potentially, a few ms of round trip for this RPC.
IIRC event handlers should *not* block the listener bus thread because that
causes baby pandas to die or something.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
