You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/04/04 19:32:05 UTC

[GitHub] [spark] ankurdave opened a new pull request, #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

ankurdave opened a new pull request, #36065:
URL: https://github.com/apache/spark/pull/36065

   ### What changes were proposed in this pull request?
   When calling a Python UDF on a DataFrame with large rows, a deadlock can occur involving the following three threads:
   
   1. The Scala task executor thread. During task execution, this is responsible for reading output produced by the Python process. However, in this case the task has finished early, and this thread is no longer reading output produced by the Python process. Instead, it is waiting for the Scala WriterThread to exit so that it can finish the task.
   2. The Scala WriterThread. This is trying to send a large row to the Python process, and is waiting for the Python process to read that row.
   3. The Python process. This is trying to send a large output to the Scala task executor thread, and is waiting for that thread to read that output.
   
   We considered the following three solutions for the deadlock:
   
   1. When the task completes, make the Scala task executor thread close the socket before waiting for the Scala WriterThread to exit. If the WriterThread is blocked on a large write, this would interrupt that write and allow the WriterThread to exit. However, it would prevent Python worker reuse.
   2. Modify PythonWorkerFactory to use interruptible I/O. [java.nio.channels.SocketChannel](https://docs.oracle.com/javase/6/docs/api/java/nio/channels/SocketChannel.html#write(java.nio.ByteBuffer)) supports interruptible blocking operations. The goal is that when the WriterThread is interrupted, it should exit even if it was blocked on a large write. However, this would be invasive.
   3. Modify the existing PythonRunner.MonitorThread to detect this deadlock and kill the Python worker. The MonitorThread currently kills the Python worker only if the task itself is interrupted. In this case, the task completes normally, so the MonitorThread does not take action. We want the MonitorThread to detect that the task is completed but the Python writer thread has not stopped, indicating a deadlock.
   
   This PR implements Option 3.
   
   ### Why are the changes needed?
   To fix a deadlock that can cause PySpark queries to hang.
   
   ### Does this PR introduce _any_ user-facing change?
   No.
   
   ### How was this patch tested?
   Added a test that previously encountered the deadlock and timed out, and now succeeds.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] dongjoon-hyun commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1097549512

   Or, since this is a broken test case, you may want to fix the test case as a follow-up on branch-3.2.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1095891580

   Merged to master and branch-3.3.
   
   Do we need this fix for other branches too, @ankurdave ?


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] ankurdave commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
ankurdave commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r847653028


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -646,6 +647,54 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       }
     }
   }
+
+  /**
+   * This thread monitors the WriterThread and kills it in case of deadlock.
+   *
+   * A deadlock can arise if the task completes while the writer thread is sending input to the
+   * Python process (e.g. due to the use of `take()`), and the Python process is still producing
+   * output. When the inputs are sufficiently large, this can result in a deadlock due to the use of
+   * blocking I/O (SPARK-38677). To resolve the deadlock, we need to close the socket.
+   */
+  class WriterMonitorThread(
+      env: SparkEnv, worker: Socket, writerThread: WriterThread, context: TaskContext)
+    extends Thread(s"Writer Monitor for $pythonExec (writer thread id ${writerThread.getId})") {
+
+    /**
+     * How long to wait before closing the socket if the writer thread has not exited after the task
+     * ends.
+     */
+    private val taskKillTimeout = env.conf.get(PYTHON_TASK_KILL_TIMEOUT)
+
+    setDaemon(true)
+
+    override def run(): Unit = {
+      // Wait until the task is completed (or the writer thread exits, in which case this thread has
+      // nothing to do).
+      while (!context.isCompleted && writerThread.isAlive) {
+        Thread.sleep(2000)
+      }
+      if (writerThread.isAlive) {
+        Thread.sleep(taskKillTimeout)
+        // If the writer thread continues running, this indicates a deadlock. Kill the worker to
+        // resolve the deadlock.
+        if (writerThread.isAlive) {
+          try {
+            // Mimic the task name used in `Executor` to help the user find out the task to blame.
+            val taskName = s"${context.partitionId}.${context.attemptNumber} " +
+              s"in stage ${context.stageId} (TID ${context.taskAttemptId})"
+            logWarning(
+              s"Detected deadlock while completing task $taskName: " +
+                "Attempting to kill Python Worker")
+            env.destroyPythonWorker(pythonExec, envVars.asScala.toMap, worker)
+          } catch {
+            case e: Exception =>

Review Comment:
   Could you elaborate on this suggestion? Is it really appropriate to suppress e.g. OutOfMemoryError or StackOverflowError?
   
   I see the WriterThread has a more limited `catch` clause, maybe we could use that instead?
   
   ```
               case t: Throwable if (NonFatal(t) || t.isInstanceOf[Exception]) =>
                 // We must avoid throwing exceptions/NonFatals here, because the thread uncaught
                 // exception handler will kill the whole executor (see
                 // org.apache.spark.executor.Executor). Instead we log the exception and allow the
                 // thread to exit.
                 logError("Exception when trying to kill worker", t)
   ```



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] viirya commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r842236404


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -188,10 +188,10 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       // SPARK-35009: avoid creating multiple monitor threads for the same python worker
       // and task context
       if (PythonRunner.runningMonitorThreads.add(key)) {
-        new MonitorThread(SparkEnv.get, worker, context).start()
+        new MonitorThread(SparkEnv.get, worker, writerThread, context).start()

Review Comment:
   Hmm, seems only one `MonitorThread` is created for same python worker in reusing case. Then we will be only able to monitor one writer thread?



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon closed pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O
URL: https://github.com/apache/spark/pull/36065


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1095896275

   Do we need this fix for other branches too, @ankurdave? Seems like SPARK-33277 was ported back to other branches but It has a conflict with branch-3.1 and branch-3.0. I would appreciate if you open a backporting PR if it needs to be ported back


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r847900101


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -646,6 +647,54 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       }
     }
   }
+
+  /**
+   * This thread monitors the WriterThread and kills it in case of deadlock.
+   *
+   * A deadlock can arise if the task completes while the writer thread is sending input to the
+   * Python process (e.g. due to the use of `take()`), and the Python process is still producing
+   * output. When the inputs are sufficiently large, this can result in a deadlock due to the use of
+   * blocking I/O (SPARK-38677). To resolve the deadlock, we need to close the socket.
+   */
+  class WriterMonitorThread(
+      env: SparkEnv, worker: Socket, writerThread: WriterThread, context: TaskContext)
+    extends Thread(s"Writer Monitor for $pythonExec (writer thread id ${writerThread.getId})") {
+
+    /**
+     * How long to wait before closing the socket if the writer thread has not exited after the task
+     * ends.
+     */
+    private val taskKillTimeout = env.conf.get(PYTHON_TASK_KILL_TIMEOUT)
+
+    setDaemon(true)
+
+    override def run(): Unit = {
+      // Wait until the task is completed (or the writer thread exits, in which case this thread has
+      // nothing to do).
+      while (!context.isCompleted && writerThread.isAlive) {
+        Thread.sleep(2000)
+      }
+      if (writerThread.isAlive) {
+        Thread.sleep(taskKillTimeout)
+        // If the writer thread continues running, this indicates a deadlock. Kill the worker to
+        // resolve the deadlock.
+        if (writerThread.isAlive) {
+          try {
+            // Mimic the task name used in `Executor` to help the user find out the task to blame.
+            val taskName = s"${context.partitionId}.${context.attemptNumber} " +
+              s"in stage ${context.stageId} (TID ${context.taskAttemptId})"
+            logWarning(
+              s"Detected deadlock while completing task $taskName: " +
+                "Attempting to kill Python Worker")
+            env.destroyPythonWorker(pythonExec, envVars.asScala.toMap, worker)
+          } catch {
+            case e: Exception =>

Review Comment:
   👍  I think it's fine as is.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] ankurdave commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
ankurdave commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1087937265

   cc @tgravescs @HyukjinKwon @Ngone51 @viirya @cloud-fan 


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] cloud-fan commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r847897425


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -646,6 +647,54 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       }
     }
   }
+
+  /**
+   * This thread monitors the WriterThread and kills it in case of deadlock.
+   *
+   * A deadlock can arise if the task completes while the writer thread is sending input to the
+   * Python process (e.g. due to the use of `take()`), and the Python process is still producing
+   * output. When the inputs are sufficiently large, this can result in a deadlock due to the use of
+   * blocking I/O (SPARK-38677). To resolve the deadlock, we need to close the socket.
+   */
+  class WriterMonitorThread(
+      env: SparkEnv, worker: Socket, writerThread: WriterThread, context: TaskContext)
+    extends Thread(s"Writer Monitor for $pythonExec (writer thread id ${writerThread.getId})") {
+
+    /**
+     * How long to wait before closing the socket if the writer thread has not exited after the task
+     * ends.
+     */
+    private val taskKillTimeout = env.conf.get(PYTHON_TASK_KILL_TIMEOUT)
+
+    setDaemon(true)
+
+    override def run(): Unit = {
+      // Wait until the task is completed (or the writer thread exits, in which case this thread has
+      // nothing to do).
+      while (!context.isCompleted && writerThread.isAlive) {
+        Thread.sleep(2000)
+      }
+      if (writerThread.isAlive) {
+        Thread.sleep(taskKillTimeout)
+        // If the writer thread continues running, this indicates a deadlock. Kill the worker to
+        // resolve the deadlock.
+        if (writerThread.isAlive) {
+          try {
+            // Mimic the task name used in `Executor` to help the user find out the task to blame.
+            val taskName = s"${context.partitionId}.${context.attemptNumber} " +
+              s"in stage ${context.stageId} (TID ${context.taskAttemptId})"
+            logWarning(
+              s"Detected deadlock while completing task $taskName: " +
+                "Attempting to kill Python Worker")
+            env.destroyPythonWorker(pythonExec, envVars.asScala.toMap, worker)
+          } catch {
+            case e: Exception =>

Review Comment:
   +1



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] mridulm commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1088083648

   +CC @zhouyejoe 


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] viirya commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
viirya commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1094076729

   I'll merge this if no more comments in few days.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1097553971

   I reverted it. I will create a new PR for that


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] dongjoon-hyun commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1097554238

   Thank you, @HyukjinKwon !


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] cloud-fan commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1091037002

   looks reasonable to me.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] ankurdave commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
ankurdave commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r846461499


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -188,10 +188,10 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       // SPARK-35009: avoid creating multiple monitor threads for the same python worker
       // and task context
       if (PythonRunner.runningMonitorThreads.add(key)) {
-        new MonitorThread(SparkEnv.get, worker, context).start()
+        new MonitorThread(SparkEnv.get, worker, writerThread, context).start()

Review Comment:
   I haven't been able to repro this yet, but I did make the change to move to a separate WriterMonitorThread.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] ankurdave commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
ankurdave commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r844254756


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -188,10 +188,10 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       // SPARK-35009: avoid creating multiple monitor threads for the same python worker
       // and task context
       if (PythonRunner.runningMonitorThreads.add(key)) {
-        new MonitorThread(SparkEnv.get, worker, context).start()
+        new MonitorThread(SparkEnv.get, worker, writerThread, context).start()

Review Comment:
   Hmm, good catch. The lifetime of the MonitorThread is tied to the lifetime of the Python worker, so this won't work when the worker is reused.
   
   I think it would be easier to just create a new WriterMonitorThread that is tied to the lifetime of the writer thread, rather than trying to stuff the functionality into the MonitorThread.
   
   I'll prototype this and try to craft a unit test that exercises the case you mentioned. It seems like we should be able to see the effect with something like `rdd.mapPartitions(lambda iterator: iterator.take(1)).coalesce(1)`. It might be tricky to actually reproduce the deadlock in that case though.
   
   cc @attilapiros 



##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -613,14 +620,16 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
     setDaemon(true)
 
     private def monitorWorker(): Unit = {
-      // Kill the worker if it is interrupted, checking until task completion.
+      // Wait until the task is either canceled or completed.
       // TODO: This has a race condition if interruption occurs, as completed may still become true.
       while (!context.isInterrupted && !context.isCompleted) {

Review Comment:
   Yeah, I think we might want that, though I'm having trouble reasoning about the effect it would have on worker reuse. We still want the monitor thread's lifetime to be tied to the Python worker's lifetime.
   
   As I mentioned above, it seems like it would be easier to create a new kind of monitor thread for this case.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1097552984

   oh


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] mridulm commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r842328473


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -613,14 +620,16 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
     setDaemon(true)
 
     private def monitorWorker(): Unit = {
-      // Kill the worker if it is interrupted, checking until task completion.
+      // Wait until the task is either canceled or completed.
       // TODO: This has a race condition if interruption occurs, as completed may still become true.
       while (!context.isInterrupted && !context.isCompleted) {

Review Comment:
   Do we want `&& writerThread.isAlive` here ?



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] ankurdave commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
ankurdave commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1096911225

   Thanks @HyukjinKwon! The deadlock was introduced by https://issues.apache.org/jira/browse/SPARK-37088, which was only ported to master, branch-3.3, and branch-3.2. So those should be sufficient.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36065:
URL: https://github.com/apache/spark/pull/36065#discussion_r846946323


##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -646,6 +647,54 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
       }
     }
   }
+
+  /**
+   * This thread monitors the WriterThread and kills it in case of deadlock.
+   *
+   * A deadlock can arise if the task completes while the writer thread is sending input to the
+   * Python process (e.g. due to the use of `take()`), and the Python process is still producing
+   * output. When the inputs are sufficiently large, this can result in a deadlock due to the use of
+   * blocking I/O (SPARK-38677). To resolve the deadlock, we need to close the socket.
+   */
+  class WriterMonitorThread(
+      env: SparkEnv, worker: Socket, writerThread: WriterThread, context: TaskContext)
+    extends Thread(s"Writer Monitor for $pythonExec (writer thread id ${writerThread.getId})") {
+
+    /**
+     * How long to wait before closing the socket if the writer thread has not exited after the task
+     * ends.
+     */
+    private val taskKillTimeout = env.conf.get(PYTHON_TASK_KILL_TIMEOUT)
+
+    setDaemon(true)
+
+    override def run(): Unit = {
+      // Wait until the task is completed (or the writer thread exits, in which case this thread has
+      // nothing to do).
+      while (!context.isCompleted && writerThread.isAlive) {
+        Thread.sleep(2000)
+      }
+      if (writerThread.isAlive) {
+        Thread.sleep(taskKillTimeout)
+        // If the writer thread continues running, this indicates a deadlock. Kill the worker to
+        // resolve the deadlock.
+        if (writerThread.isAlive) {
+          try {
+            // Mimic the task name used in `Executor` to help the user find out the task to blame.
+            val taskName = s"${context.partitionId}.${context.attemptNumber} " +
+              s"in stage ${context.stageId} (TID ${context.taskAttemptId})"
+            logWarning(
+              s"Detected deadlock while completing task $taskName: " +
+                "Attempting to kill Python Worker")
+            env.destroyPythonWorker(pythonExec, envVars.asScala.toMap, worker)
+          } catch {
+            case e: Exception =>

Review Comment:
   ```suggestion
               case e: Throwable =>
   ```



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1097422844

   Thanks for clarification!


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] dongjoon-hyun commented on pull request #36065: [SPARK-38677][PYSPARK] Python MonitorThread should detect deadlock due to blocking I/O

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #36065:
URL: https://github.com/apache/spark/pull/36065#issuecomment-1097548915

   Hi, @ankurdave and @HyukjinKwon 
   
   This patch broke branch-3.2 because we don't have `CPickleSerializer` there.
   ```
   ======================================================================
   [1995](https://github.com/apache/spark/runs/5983322734?check_suite_focus=true#step:11:1995)
   ERROR: test_take_on_jrdd_with_large_rows_should_not_cause_deadlock (pyspark.tests.test_rdd.RDDTests)
   [1996](https://github.com/apache/spark/runs/5983322734?check_suite_focus=true#step:11:1996)
   ----------------------------------------------------------------------
   [1997](https://github.com/apache/spark/runs/5983322734?check_suite_focus=true#step:11:1997)
   Traceback (most recent call last):
   [1998](https://github.com/apache/spark/runs/5983322734?check_suite_focus=true#step:11:1998)
     File "/__w/spark/spark/python/pyspark/tests/test_rdd.py", line 727, in test_take_on_jrdd_with_large_rows_should_not_cause_deadlock
   [1999](https://github.com/apache/spark/runs/5983322734?check_suite_focus=true#step:11:1999)
       actual = CPickleSerializer().loads(df.rdd.map(list)._jrdd.first())
   [2000](https://github.com/apache/spark/runs/5983322734?check_suite_focus=true#step:11:2000)
   NameError: name 'CPickleSerializer' is not defined
   ```
   
   Could you revert this from `branch-3.2` and land it back with as a single piece?


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org