You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "changgyoopark-db (via GitHub)" <gi...@apache.org> on 2024/09/23 07:55:22 UTC
[PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
changgyoopark-db opened a new pull request, #48208:
URL: https://github.com/apache/spark/pull/48208
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Execution posts "closed" even when the execution has yet to be started.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Execution can be interrupted before started, thus causing the "closed" message to be omitted.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
SparkConnectServiceSuite.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1771661488
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -68,10 +68,19 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
}
/**
- * Register a callback that gets executed after completion/interruption of the execution thread.
+ * Execute the supplied lambda if no execute runner thread has started, or pass it to the
+ * execute runner thread so that the lambda will get executed on completion. Failure details
+ * will only be passed to the lambda only when an error was raised during actual execution.
*/
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ private[connect] def processNowOrOnCompletion(callback: Try[Unit] => Unit): Unit = {
+ lock.synchronized {
+ if (started) {
+ promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ } else {
+ // No thread has been spawned, and therefore there was no failure, passing `Success`.
+ callback(Success(()))
Review Comment:
Does this callback have to be executed within the lock?
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #48208:
URL: https://github.com/apache/spark/pull/48208#issuecomment-2371182324
New test failure caused by the updated `interrupt` method semantics: to be fixed late today.
```
[info] - interrupt all - background queries, foreground interrupt *** FAILED *** (127 milliseconds)
[info] ListBuffer("18e26c82-502f-48bb-8028-328963d8e6d3") had length 1 instead of expected length 2 Interrupted operations: ListBuffer(18e26c82-502f-48bb-8028-328963d8e6d3). (SparkSessionE2ESuite.scala:79)
[info] org.scalatest.exceptions.TestFailedException:
```
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1771663028
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -68,10 +68,19 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
}
/**
- * Register a callback that gets executed after completion/interruption of the execution thread.
+ * Execute the supplied lambda if no execute runner thread has started, or pass it to the
+ * execute runner thread so that the lambda will get executed on completion. Failure details
+ * will only be passed to the lambda only when an error was raised during actual execution.
*/
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ private[connect] def processNowOrOnCompletion(callback: Try[Unit] => Unit): Unit = {
+ lock.synchronized {
+ if (started) {
+ promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ } else {
+ // No thread has been spawned, and therefore there was no failure, passing `Success`.
+ callback(Success(()))
Review Comment:
HI, that's what I'm fixing; incoming new change in 1 minute.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #48208:
URL: https://github.com/apache/spark/pull/48208#issuecomment-2370595119
@juliuszsompolski Hi again! I know... this change is a bit complicated, so I wouldn't urge you to review the code unless you have plenty of free time. To ease reviewing the code,
- ExecuteThreadRunner can be in one of the five states below.
1. "notStarted": initial state / can be interrupted / the completion callback will be executed by ExecuteHolder.close.
2. "interrupted": the thread has never been started, and was interrupted / terminal state / the completion callback will be executed by ExecuteHolder.close.
3. "started": the thread is running / can be interrupted / -.
4. "startedInterrupted": the thread is running and interrupted / cannot be completed / the thread executes the completion callback.
5. "completed": the thread will be terminated / terminal state / -.
So, 4. "startedInterrupted" is the only state in which ExecuteThreadRunner is allowed to execute the completion callback.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1773214868
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -139,23 +129,50 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
}
}
} catch {
- ErrorUtils.handleError(
- "execute",
- executeHolder.responseObserver,
- executeHolder.sessionHolder.userId,
- executeHolder.sessionHolder.sessionId,
- Some(executeHolder.eventsManager),
- interrupted)
+ case e: Throwable if state.getAcquire() != ThreadState.startedInterrupted =>
+ ErrorUtils.handleError(
+ "execute",
+ executeHolder.responseObserver,
+ executeHolder.sessionHolder.userId,
+ executeHolder.sessionHolder.sessionId,
+ Some(executeHolder.eventsManager),
+ false)(e)
+ } finally {
+ // Make sure to transition to completed in order to prevent the thread from being interrupted
+ // afterwards.
+ var currentState = state.getAcquire()
+ while (currentState == ThreadState.started ||
+ currentState == ThreadState.startedInterrupted) {
+ val interrupted = currentState == ThreadState.startedInterrupted
+ val prevState = state.compareAndExchangeRelease(currentState, ThreadState.completed)
+ if (prevState == currentState) {
+ if (interrupted) {
+ try {
+ ErrorUtils.handleError(
+ "execute",
+ executeHolder.responseObserver,
+ executeHolder.sessionHolder.userId,
+ executeHolder.sessionHolder.sessionId,
+ Some(executeHolder.eventsManager),
+ true)(new SparkSQLException("OPERATION_CANCELED", Map.empty))
+ } finally {
+ executeHolder.cleanup()
+ }
+ }
+ return
+ }
+ currentState = prevState
+ }
}
}
// Inner executeInternal is wrapped by execute() for error handling.
- private def executeInternal() = {
- // synchronized - check if already got interrupted while starting.
- lock.synchronized {
- if (interrupted) {
- throw new InterruptedException()
- }
+ private def executeInternal(): Unit = {
+ val prevState = state.compareAndExchangeRelease(ThreadState.notStarted, ThreadState.started)
+ if (prevState != ThreadState.notStarted && prevState != ThreadState.started) {
+ // Silently return, expecting that the caller would handle the interruption.
+ assert(prevState != ThreadState.completed)
Review Comment:
If we remove the state transition in `start`, then you're right. I'll make it clearer.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1773222971
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -32,76 +32,73 @@ import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{ExecuteHolder, ExecuteSessionTag, SparkConnectService}
import org.apache.spark.sql.connect.utils.ErrorUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.Utils
/**
* This class launches the actual execution in an execution thread. The execution pushes the
* responses to a ExecuteResponseObserver in executeHolder.
*/
private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends Logging {
- private val promise: Promise[Unit] = Promise[Unit]()
+ /** The thread state. */
+ private val state: AtomicInteger = new AtomicInteger(ThreadState.notStarted)
// The newly created thread will inherit all InheritableThreadLocals used by Spark,
// e.g. SparkContext.localProperties. If considering implementing a thread-pool,
// forwarding of thread locals needs to be taken into account.
- private val executionThread: ExecutionThread = new ExecutionThread(promise)
-
- private var started: Boolean = false
-
- private var interrupted: Boolean = false
-
- private var completed: Boolean = false
-
- private val lock = new Object
+ private val executionThread: ExecutionThread = new ExecutionThread()
/** Launches the execution in a background thread, returns immediately. */
private[connect] def start(): Unit = {
- lock.synchronized {
- assert(!started)
- // Do not start if already interrupted.
- if (!interrupted) {
- executionThread.start()
- started = true
- }
- }
- }
+ if (state.getAcquire() == ThreadState.notStarted) {
+ executionThread.start()
- /**
- * Register a callback that gets executed after completion/interruption of the execution thread.
- */
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ // If the thread is started earlier than this, the thread will change the state itself.
+ state.compareAndExchangeRelease(ThreadState.notStarted, ThreadState.started)
+ }
}
/**
- * Interrupt the executing thread.
+ * Interrupts the execution thread if the thread is running and has yet to be completed.
+ *
* @return
- * true if it was not interrupted before, false if it was already interrupted or completed.
+ * true if the thread is running and interrupted.
*/
private[connect] def interrupt(): Boolean = {
- lock.synchronized {
- if (!started && !interrupted) {
- // execution thread hasn't started yet, and will not be started.
- // handle the interrupted error here directly.
- interrupted = true
- ErrorUtils.handleError(
- "execute",
- executeHolder.responseObserver,
- executeHolder.sessionHolder.userId,
- executeHolder.sessionHolder.sessionId,
- Some(executeHolder.eventsManager),
- interrupted)(new SparkSQLException("OPERATION_CANCELED", Map.empty))
- true
- } else if (!interrupted && !completed) {
- // checking completed prevents sending interrupt onError after onCompleted
- interrupted = true
- executionThread.interrupt()
- true
+ var currentState = state.getAcquire()
+ while (currentState == ThreadState.notStarted || currentState == ThreadState.started) {
+ val newState = if (currentState == ThreadState.notStarted) {
+ ThreadState.interrupted
} else {
- false
+ ThreadState.startedInterrupted
}
+
+ val prevState = state.compareAndExchangeRelease(currentState, newState)
+ if (prevState == currentState) {
+ if (prevState == ThreadState.notStarted) {
+ // The execution thread has not been started, and will never be started.
Review Comment:
You're right, the thread will silently exit; need to update the comment.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48208:
URL: https://github.com/apache/spark/pull/48208#issuecomment-2372595470
Merged to master.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1773221283
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -32,76 +32,73 @@ import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{ExecuteHolder, ExecuteSessionTag, SparkConnectService}
import org.apache.spark.sql.connect.utils.ErrorUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.Utils
/**
* This class launches the actual execution in an execution thread. The execution pushes the
* responses to a ExecuteResponseObserver in executeHolder.
*/
private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends Logging {
- private val promise: Promise[Unit] = Promise[Unit]()
+ /** The thread state. */
+ private val state: AtomicInteger = new AtomicInteger(ThreadState.notStarted)
// The newly created thread will inherit all InheritableThreadLocals used by Spark,
// e.g. SparkContext.localProperties. If considering implementing a thread-pool,
// forwarding of thread locals needs to be taken into account.
- private val executionThread: ExecutionThread = new ExecutionThread(promise)
-
- private var started: Boolean = false
-
- private var interrupted: Boolean = false
-
- private var completed: Boolean = false
-
- private val lock = new Object
+ private val executionThread: ExecutionThread = new ExecutionThread()
/** Launches the execution in a background thread, returns immediately. */
private[connect] def start(): Unit = {
- lock.synchronized {
- assert(!started)
- // Do not start if already interrupted.
- if (!interrupted) {
- executionThread.start()
- started = true
- }
- }
- }
+ if (state.getAcquire() == ThreadState.notStarted) {
+ executionThread.start()
- /**
- * Register a callback that gets executed after completion/interruption of the execution thread.
- */
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ // If the thread is started earlier than this, the thread will change the state itself.
+ state.compareAndExchangeRelease(ThreadState.notStarted, ThreadState.started)
Review Comment:
At the start of executeInternal is enough; I'll remove this.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1773106069
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -139,23 +129,50 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
}
}
} catch {
- ErrorUtils.handleError(
- "execute",
- executeHolder.responseObserver,
- executeHolder.sessionHolder.userId,
- executeHolder.sessionHolder.sessionId,
- Some(executeHolder.eventsManager),
- interrupted)
+ case e: Throwable if state.getAcquire() != ThreadState.startedInterrupted =>
+ ErrorUtils.handleError(
+ "execute",
+ executeHolder.responseObserver,
+ executeHolder.sessionHolder.userId,
+ executeHolder.sessionHolder.sessionId,
+ Some(executeHolder.eventsManager),
+ false)(e)
+ } finally {
+ // Make sure to transition to completed in order to prevent the thread from being interrupted
+ // afterwards.
+ var currentState = state.getAcquire()
+ while (currentState == ThreadState.started ||
+ currentState == ThreadState.startedInterrupted) {
+ val interrupted = currentState == ThreadState.startedInterrupted
+ val prevState = state.compareAndExchangeRelease(currentState, ThreadState.completed)
+ if (prevState == currentState) {
+ if (interrupted) {
+ try {
+ ErrorUtils.handleError(
+ "execute",
+ executeHolder.responseObserver,
+ executeHolder.sessionHolder.userId,
+ executeHolder.sessionHolder.sessionId,
+ Some(executeHolder.eventsManager),
+ true)(new SparkSQLException("OPERATION_CANCELED", Map.empty))
+ } finally {
+ executeHolder.cleanup()
+ }
+ }
+ return
+ }
+ currentState = prevState
+ }
}
}
// Inner executeInternal is wrapped by execute() for error handling.
- private def executeInternal() = {
- // synchronized - check if already got interrupted while starting.
- lock.synchronized {
- if (interrupted) {
- throw new InterruptedException()
- }
+ private def executeInternal(): Unit = {
+ val prevState = state.compareAndExchangeRelease(ThreadState.notStarted, ThreadState.started)
+ if (prevState != ThreadState.notStarted && prevState != ThreadState.started) {
+ // Silently return, expecting that the caller would handle the interruption.
+ assert(prevState != ThreadState.completed)
Review Comment:
`assert(prevState == ThreadState.interrupted)` should be true here, because the only way this could happen is if it got interrupted right as it was starting? (it should not be `startedInterrupted` here?)
Maybe extend this comment with more explanation?
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -32,76 +32,73 @@ import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{ExecuteHolder, ExecuteSessionTag, SparkConnectService}
import org.apache.spark.sql.connect.utils.ErrorUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.Utils
/**
* This class launches the actual execution in an execution thread. The execution pushes the
* responses to a ExecuteResponseObserver in executeHolder.
*/
private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends Logging {
- private val promise: Promise[Unit] = Promise[Unit]()
+ /** The thread state. */
+ private val state: AtomicInteger = new AtomicInteger(ThreadState.notStarted)
// The newly created thread will inherit all InheritableThreadLocals used by Spark,
// e.g. SparkContext.localProperties. If considering implementing a thread-pool,
// forwarding of thread locals needs to be taken into account.
- private val executionThread: ExecutionThread = new ExecutionThread(promise)
-
- private var started: Boolean = false
-
- private var interrupted: Boolean = false
-
- private var completed: Boolean = false
-
- private val lock = new Object
+ private val executionThread: ExecutionThread = new ExecutionThread()
/** Launches the execution in a background thread, returns immediately. */
private[connect] def start(): Unit = {
- lock.synchronized {
- assert(!started)
- // Do not start if already interrupted.
- if (!interrupted) {
- executionThread.start()
- started = true
- }
- }
- }
+ if (state.getAcquire() == ThreadState.notStarted) {
+ executionThread.start()
- /**
- * Register a callback that gets executed after completion/interruption of the execution thread.
- */
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ // If the thread is started earlier than this, the thread will change the state itself.
+ state.compareAndExchangeRelease(ThreadState.notStarted, ThreadState.started)
+ }
}
/**
- * Interrupt the executing thread.
+ * Interrupts the execution thread if the thread is running and has yet to be completed.
+ *
* @return
- * true if it was not interrupted before, false if it was already interrupted or completed.
+ * true if the thread is running and interrupted.
*/
private[connect] def interrupt(): Boolean = {
- lock.synchronized {
- if (!started && !interrupted) {
- // execution thread hasn't started yet, and will not be started.
- // handle the interrupted error here directly.
- interrupted = true
- ErrorUtils.handleError(
- "execute",
- executeHolder.responseObserver,
- executeHolder.sessionHolder.userId,
- executeHolder.sessionHolder.sessionId,
- Some(executeHolder.eventsManager),
- interrupted)(new SparkSQLException("OPERATION_CANCELED", Map.empty))
- true
- } else if (!interrupted && !completed) {
- // checking completed prevents sending interrupt onError after onCompleted
- interrupted = true
- executionThread.interrupt()
- true
+ var currentState = state.getAcquire()
+ while (currentState == ThreadState.notStarted || currentState == ThreadState.started) {
+ val newState = if (currentState == ThreadState.notStarted) {
+ ThreadState.interrupted
} else {
- false
+ ThreadState.startedInterrupted
}
+
+ val prevState = state.compareAndExchangeRelease(currentState, newState)
+ if (prevState == currentState) {
+ if (prevState == ThreadState.notStarted) {
+ // The execution thread has not been started, and will never be started.
Review Comment:
If this happens before the thread reaches the transition in executeInternal, executeInternal will silently exit?
So it's techically not true that the thread has not been started; it might have been started but not reaching that transition?
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -32,76 +32,73 @@ import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{ExecuteHolder, ExecuteSessionTag, SparkConnectService}
import org.apache.spark.sql.connect.utils.ErrorUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.Utils
/**
* This class launches the actual execution in an execution thread. The execution pushes the
* responses to a ExecuteResponseObserver in executeHolder.
*/
private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends Logging {
- private val promise: Promise[Unit] = Promise[Unit]()
+ /** The thread state. */
+ private val state: AtomicInteger = new AtomicInteger(ThreadState.notStarted)
// The newly created thread will inherit all InheritableThreadLocals used by Spark,
// e.g. SparkContext.localProperties. If considering implementing a thread-pool,
// forwarding of thread locals needs to be taken into account.
- private val executionThread: ExecutionThread = new ExecutionThread(promise)
-
- private var started: Boolean = false
-
- private var interrupted: Boolean = false
-
- private var completed: Boolean = false
-
- private val lock = new Object
+ private val executionThread: ExecutionThread = new ExecutionThread()
/** Launches the execution in a background thread, returns immediately. */
private[connect] def start(): Unit = {
- lock.synchronized {
- assert(!started)
- // Do not start if already interrupted.
- if (!interrupted) {
- executionThread.start()
- started = true
- }
- }
- }
+ if (state.getAcquire() == ThreadState.notStarted) {
+ executionThread.start()
- /**
- * Register a callback that gets executed after completion/interruption of the execution thread.
- */
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ // If the thread is started earlier than this, the thread will change the state itself.
+ state.compareAndExchangeRelease(ThreadState.notStarted, ThreadState.started)
Review Comment:
Is this needed here, or is the transition to started at the start of executeInternal enough?
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -32,76 +32,73 @@ import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{ExecuteHolder, ExecuteSessionTag, SparkConnectService}
import org.apache.spark.sql.connect.utils.ErrorUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.Utils
/**
* This class launches the actual execution in an execution thread. The execution pushes the
* responses to a ExecuteResponseObserver in executeHolder.
*/
private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends Logging {
- private val promise: Promise[Unit] = Promise[Unit]()
+ /** The thread state. */
+ private val state: AtomicInteger = new AtomicInteger(ThreadState.notStarted)
// The newly created thread will inherit all InheritableThreadLocals used by Spark,
// e.g. SparkContext.localProperties. If considering implementing a thread-pool,
// forwarding of thread locals needs to be taken into account.
- private val executionThread: ExecutionThread = new ExecutionThread(promise)
-
- private var started: Boolean = false
-
- private var interrupted: Boolean = false
-
- private var completed: Boolean = false
-
- private val lock = new Object
+ private val executionThread: ExecutionThread = new ExecutionThread()
/** Launches the execution in a background thread, returns immediately. */
private[connect] def start(): Unit = {
- lock.synchronized {
- assert(!started)
- // Do not start if already interrupted.
- if (!interrupted) {
- executionThread.start()
- started = true
- }
- }
- }
+ if (state.getAcquire() == ThreadState.notStarted) {
+ executionThread.start()
Review Comment:
The other possible state here is interrupted, if it got interrupted before started?
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -226,17 +243,13 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
observedMetrics ++ accumulatedInPython))
}
- lock.synchronized {
- // Synchronized before sending ResultComplete, and up until completing the result stream
- // to prevent a situation in which a client of reattachable execution receives
- // ResultComplete, and proceeds to send ReleaseExecute, and that triggers an interrupt
- // before it finishes.
-
- if (interrupted) {
- // check if it got interrupted at the very last moment
- throw new InterruptedException()
- }
- completed = true // no longer interruptible
+ // State transition should be atomic to prevent a situation in which a client of reattachable
+ // execution receives ResultComplete, and proceeds to send ReleaseExecute, and that triggers
+ // an interrupt before it finishes.
+ if (state.compareAndExchangeRelease(
+ ThreadState.started,
+ ThreadState.completed) == ThreadState.started) {
+ // Now, the execution cannot be interrupted.
Review Comment:
Why do we not have to throw InterruptedException anymore in an else to this if?
I guess because the check for this is now in a finally and not in catch?
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #48208:
URL: https://github.com/apache/spark/pull/48208#issuecomment-2371140000
@vicennial could you also take a detailed look next to @hvanhovell 's brief look? :)
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1771722046
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -68,10 +68,19 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
}
/**
- * Register a callback that gets executed after completion/interruption of the execution thread.
+ * Execute the supplied lambda if no execute runner thread has started, or pass it to the
+ * execute runner thread so that the lambda will get executed on completion. Failure details
+ * will only be passed to the lambda only when an error was raised during actual execution.
*/
- private[connect] def processOnCompletion(callback: Try[Unit] => Unit): Unit = {
- promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ private[connect] def processNowOrOnCompletion(callback: Try[Unit] => Unit): Unit = {
+ lock.synchronized {
+ if (started) {
+ promise.future.onComplete(callback)(ExecuteThreadRunner.namedExecutionContext)
+ } else {
+ // No thread has been spawned, and therefore there was no failure, passing `Success`.
+ callback(Success(()))
Review Comment:
the whole "promise" and locking won't be needed for this as what this "promise" does is pretty static. I'll try to get rid of most of those complicated code pieces.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1773216355
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -32,76 +32,73 @@ import org.apache.spark.sql.connect.common.ProtoUtils
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
import org.apache.spark.sql.connect.service.{ExecuteHolder, ExecuteSessionTag, SparkConnectService}
import org.apache.spark.sql.connect.utils.ErrorUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
+import org.apache.spark.util.Utils
/**
* This class launches the actual execution in an execution thread. The execution pushes the
* responses to a ExecuteResponseObserver in executeHolder.
*/
private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends Logging {
- private val promise: Promise[Unit] = Promise[Unit]()
+ /** The thread state. */
+ private val state: AtomicInteger = new AtomicInteger(ThreadState.notStarted)
// The newly created thread will inherit all InheritableThreadLocals used by Spark,
// e.g. SparkContext.localProperties. If considering implementing a thread-pool,
// forwarding of thread locals needs to be taken into account.
- private val executionThread: ExecutionThread = new ExecutionThread(promise)
-
- private var started: Boolean = false
-
- private var interrupted: Boolean = false
-
- private var completed: Boolean = false
-
- private val lock = new Object
+ private val executionThread: ExecutionThread = new ExecutionThread()
/** Launches the execution in a background thread, returns immediately. */
private[connect] def start(): Unit = {
- lock.synchronized {
- assert(!started)
- // Do not start if already interrupted.
- if (!interrupted) {
- executionThread.start()
- started = true
- }
- }
- }
+ if (state.getAcquire() == ThreadState.notStarted) {
+ executionThread.start()
Review Comment:
Exactly: I'll add an assertion.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48208:
URL: https://github.com/apache/spark/pull/48208#discussion_r1773212180
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -226,17 +243,13 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
observedMetrics ++ accumulatedInPython))
}
- lock.synchronized {
- // Synchronized before sending ResultComplete, and up until completing the result stream
- // to prevent a situation in which a client of reattachable execution receives
- // ResultComplete, and proceeds to send ReleaseExecute, and that triggers an interrupt
- // before it finishes.
-
- if (interrupted) {
- // check if it got interrupted at the very last moment
- throw new InterruptedException()
- }
- completed = true // no longer interruptible
+ // State transition should be atomic to prevent a situation in which a client of reattachable
+ // execution receives ResultComplete, and proceeds to send ReleaseExecute, and that triggers
+ // an interrupt before it finishes.
+ if (state.compareAndExchangeRelease(
+ ThreadState.started,
+ ThreadState.completed) == ThreadState.started) {
+ // Now, the execution cannot be interrupted.
Review Comment:
Exactly. I'll add more comments.
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #48208:
URL: https://github.com/apache/spark/pull/48208#issuecomment-2371135501
@juliuszsompolski Thanks a lot! @hvanhovell I know you're super busy, but can you please take a brief look at the change?
--
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
Re: [PR] [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #48208: [SPARK-49688][CONNECT] Fix a data race between interrupt and execute plan
URL: https://github.com/apache/spark/pull/48208
--
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