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/09 13:15:20 UTC
[PR] [SPARK-49544] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
changgyoopark-db opened a new pull request, #48034:
URL: https://github.com/apache/spark/pull/48034
<!--
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.
-->
Replace the coarse-locking mechanism implemented in SparkConnectExecutionManager with ConcurrentMap in order to ameliorate lock contention.
### 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.
-->
When there are too many threads, e.g., ~10K threads on a 4-core node, OS scheduling may cause priority inversion that leads to a serious performance problems, e.g., a 1000s delay when reattaching to an execute holder.
### 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.
-->
Existing test cases.
### 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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1755009392
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,30 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
Review Comment:
Thanks, also instantiating an ExecuteHolder after checking that it's absent would be more efficient. I'll fix it.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751666770
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -108,43 +110,50 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
* execution if still running, free all resources.
*/
private[connect] def removeExecuteHolder(key: ExecuteKey, abandoned: Boolean = false): Unit = {
- var executeHolder: Option[ExecuteHolder] = None
+ val executeHolder = executions.get(key)
+
+ if (executeHolder == null) {
+ return
+ }
+
+ // Put into abandonedTombstones before removing it from executions, so that the client ends up
+ // getting an INVALID_HANDLE.OPERATION_ABANDONED error on a retry.
+ if (abandoned) {
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
+ }
+
+ // Remove the execution from the map *after* putting it in abandonedTombstones.
+ executions.remove(key)
+ executeHolder.sessionHolder.removeExecuteHolder(executeHolder.operationId)
+
executionsLock.synchronized {
- executeHolder = executions.remove(key)
- executeHolder.foreach { e =>
- // Put into abandonedTombstones under lock, so that if it's accessed it will end up
- // with INVALID_HANDLE.OPERATION_ABANDONED error.
- if (abandoned) {
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
- e.sessionHolder.removeExecuteHolder(e.operationId)
- }
if (executions.isEmpty) {
lastExecutionTimeMs = Some(System.currentTimeMillis())
}
- logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
}
- // close the execution outside the lock
- executeHolder.foreach { e =>
- e.close()
- if (abandoned) {
- // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
+
+ logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
+
+ executeHolder.close()
+ if (abandoned) {
+ // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
}
}
private[connect] def getExecuteHolder(key: ExecuteKey): Option[ExecuteHolder] = {
- executionsLock.synchronized {
- executions.get(key)
- }
+ Option(executions.get(key))
}
private[connect] def removeAllExecutionsForSession(key: SessionKey): Unit = {
- val sessionExecutionHolders = executionsLock.synchronized {
- executions.filter(_._2.sessionHolder.key == key)
- }
- sessionExecutionHolders.foreach { case (_, executeHolder) =>
+ var sessionExecutionHolders = mutable.ArrayBuffer[ExecuteHolder]()
+ executions.forEach((_, executeHolder) => {
+ if (executeHolder.sessionHolder.key == key) {
+ sessionExecutionHolders += executeHolder
+ }
+ })
+
+ sessionExecutionHolders.foreach { executeHolder =>
Review Comment:
You're right, this will be fine.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751597435
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -108,43 +110,50 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
* execution if still running, free all resources.
*/
private[connect] def removeExecuteHolder(key: ExecuteKey, abandoned: Boolean = false): Unit = {
- var executeHolder: Option[ExecuteHolder] = None
+ val executeHolder = executions.get(key)
+
+ if (executeHolder == null) {
+ return
+ }
+
+ // Put into abandonedTombstones before removing it from executions, so that the client ends up
+ // getting an INVALID_HANDLE.OPERATION_ABANDONED error on a retry.
+ if (abandoned) {
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
+ }
+
+ // Remove the execution from the map *after* putting it in abandonedTombstones.
+ executions.remove(key)
+ executeHolder.sessionHolder.removeExecuteHolder(executeHolder.operationId)
+
executionsLock.synchronized {
- executeHolder = executions.remove(key)
- executeHolder.foreach { e =>
- // Put into abandonedTombstones under lock, so that if it's accessed it will end up
- // with INVALID_HANDLE.OPERATION_ABANDONED error.
- if (abandoned) {
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
- e.sessionHolder.removeExecuteHolder(e.operationId)
- }
if (executions.isEmpty) {
lastExecutionTimeMs = Some(System.currentTimeMillis())
}
- logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
}
- // close the execution outside the lock
- executeHolder.foreach { e =>
- e.close()
- if (abandoned) {
- // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
+
+ logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
+
+ executeHolder.close()
+ if (abandoned) {
+ // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
}
}
private[connect] def getExecuteHolder(key: ExecuteKey): Option[ExecuteHolder] = {
- executionsLock.synchronized {
- executions.get(key)
- }
+ Option(executions.get(key))
}
private[connect] def removeAllExecutionsForSession(key: SessionKey): Unit = {
- val sessionExecutionHolders = executionsLock.synchronized {
- executions.filter(_._2.sessionHolder.key == key)
- }
- sessionExecutionHolders.foreach { case (_, executeHolder) =>
+ var sessionExecutionHolders = mutable.ArrayBuffer[ExecuteHolder]()
+ executions.forEach((_, executeHolder) => {
+ if (executeHolder.sessionHolder.key == key) {
+ sessionExecutionHolders += executeHolder
+ }
+ })
+
+ sessionExecutionHolders.foreach { executeHolder =>
Review Comment:
There's a comment in SessionHolder:
```
// Clean up all executions.
// After closedTimeMs is defined, SessionHolder.addExecuteHolder() will not allow new executions
// to be added for this session anymore. Because both SessionHolder.addExecuteHolder() and
// SparkConnectExecutionManager.removeAllExecutionsForSession() are executed under
// executionsLock, this guarantees that removeAllExecutionsForSession triggered here will
// remove all executions and no new executions will be added in the meanwhile.
SparkConnectService.executionManager.removeAllExecutionsForSession(this.key)
```
which needs to be adjusted (and thought through for correctness now...).
It's no longer under executionsLock, but removeAllExecutionsForSession is called after closedTimeMs is already defined...
If there is a `sessionHolder.addExecuteHolder(executeHolder)` from https://github.com/apache/spark/pull/48034/files#diff-a1ce7b58be6b0e0cd79bea6a5ee2b399d48d18e90e0ae50e1d3160d74f639ad7R96 that's in progress that started when closedTimeMs was not yet defined, then that executeHolder has already been put into the global list in https://github.com/apache/spark/pull/48034/files#diff-a1ce7b58be6b0e0cd79bea6a5ee2b399d48d18e90e0ae50e1d3160d74f639ad7R85, then it will be removed here by removeAllExecutionsForSession
So I think it's fine... Think about it.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,28 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ } else if (getAbandonedTombstone(executeHolder.key).isDefined) {
+ executions.remove(executeHolder.key)
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ }
+
+ sessionHolder.addExecuteHolder(executeHolder)
+
executionsLock.synchronized {
- // Check if the operation already exists, both in active executions, and in the graveyard
- // of tombstones of executions that have been abandoned.
- // The latter is to prevent double execution when a client retries execution, thinking it
- // never reached the server, but in fact it did, and already got removed as abandoned.
- if (executions.get(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- if (getAbandonedTombstone(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- sessionHolder.addExecuteHolder(executeHolder)
- executions.put(executeHolder.key, executeHolder)
Review Comment:
This needs some careful thought over whether it was needed that putting the execution into the session and into the global execution list under one lock was necessary had any implications...
I think it is fine, because all operations that add/remove executions operate on the global list here, and the list in sessionHolder is used only for operations internal to the session (like interrupting all operations)
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751656768
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -61,6 +60,7 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
private var lastExecutionTimeMs: Option[Long] = Some(System.currentTimeMillis())
Review Comment:
Thanks, I think I'll have to keep it, and need to deal with a potential data race between "lastExecutionTimeMs" and "executions" => maybe, by not setting None to it and making sure that it monotonically increases? I'll think about it.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #48034: [SPARK-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap
URL: https://github.com/apache/spark/pull/48034
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751722122
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,28 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ } else if (getAbandonedTombstone(executeHolder.key).isDefined) {
+ executions.remove(executeHolder.key)
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ }
+
+ sessionHolder.addExecuteHolder(executeHolder)
+
executionsLock.synchronized {
- // Check if the operation already exists, both in active executions, and in the graveyard
- // of tombstones of executions that have been abandoned.
- // The latter is to prevent double execution when a client retries execution, thinking it
- // never reached the server, but in fact it did, and already got removed as abandoned.
- if (executions.get(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- if (getAbandonedTombstone(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- sessionHolder.addExecuteHolder(executeHolder)
- executions.put(executeHolder.key, executeHolder)
Review Comment:
Yes, removing executions from SessionHolder will make the code simpler. My concern is, `interrupt*` will have to iterate over ExecutionManager.executions; if there are many sessions, `interrupt*` methods become inefficient.
-> Created a separate ticket for this: https://issues.apache.org/jira/browse/SPARK-49585 we can continue discuss about it in the ticket.
Agreed that there is no operation that reads ExecutionManager.sessions first assuming that there is an associated entry in SessionHolder.executions.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751600547
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -61,6 +60,7 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
private var lastExecutionTimeMs: Option[Long] = Some(System.currentTimeMillis())
Review Comment:
@juliuszsompolski While reviewing my own change, I noticed that lastExecutionTimeMs is never used in the code. Is this dead code? If so, can I just remove it in this PR?
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751672281
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,28 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ } else if (getAbandonedTombstone(executeHolder.key).isDefined) {
+ executions.remove(executeHolder.key)
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ }
+
+ sessionHolder.addExecuteHolder(executeHolder)
+
executionsLock.synchronized {
- // Check if the operation already exists, both in active executions, and in the graveyard
- // of tombstones of executions that have been abandoned.
- // The latter is to prevent double execution when a client retries execution, thinking it
- // never reached the server, but in fact it did, and already got removed as abandoned.
- if (executions.get(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- if (getAbandonedTombstone(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- sessionHolder.addExecuteHolder(executeHolder)
- executions.put(executeHolder.key, executeHolder)
Review Comment:
> But, is it now possible in some operations that an execution will be retrieved from the global map, validating that it exists, but then it will assume that it's present in the session map but race before it's added there?
As far as I looked, this looks fine, I don't see any places that first look for an operation in the global map, and then again look for it in the session map.
It would probably be a simplification to get rid of the executions in SessionHolder alltogether, and depend on the global map as a source of truth everywhere... The reason it is how it is is that before the ExecutionManager existed, the list of executions was only inside the sessions and there was no global list... Since ExecutionManager added a global utility to deal with it, it would be cleaner to get rid of the executions inside session altogether... they seem to be used in very few places anyway, but I didn't completely remove them...
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1755021926
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,30 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
Review Comment:
-> key is generated within ExecuteHolder, so deferring instantiation won't work. I'll just checking the tombstone part in the lambda.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1754990368
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,30 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
Review Comment:
You could use computeIfAbsent here... That will allow you to do the tombstone check there.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48034:
URL: https://github.com/apache/spark/pull/48034#issuecomment-2344975115
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751660500
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -108,43 +110,50 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
* execution if still running, free all resources.
*/
private[connect] def removeExecuteHolder(key: ExecuteKey, abandoned: Boolean = false): Unit = {
- var executeHolder: Option[ExecuteHolder] = None
+ val executeHolder = executions.get(key)
+
+ if (executeHolder == null) {
+ return
+ }
+
+ // Put into abandonedTombstones before removing it from executions, so that the client ends up
+ // getting an INVALID_HANDLE.OPERATION_ABANDONED error on a retry.
+ if (abandoned) {
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
+ }
+
+ // Remove the execution from the map *after* putting it in abandonedTombstones.
+ executions.remove(key)
+ executeHolder.sessionHolder.removeExecuteHolder(executeHolder.operationId)
+
executionsLock.synchronized {
- executeHolder = executions.remove(key)
- executeHolder.foreach { e =>
- // Put into abandonedTombstones under lock, so that if it's accessed it will end up
- // with INVALID_HANDLE.OPERATION_ABANDONED error.
- if (abandoned) {
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
- e.sessionHolder.removeExecuteHolder(e.operationId)
- }
if (executions.isEmpty) {
lastExecutionTimeMs = Some(System.currentTimeMillis())
}
- logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
}
- // close the execution outside the lock
- executeHolder.foreach { e =>
- e.close()
- if (abandoned) {
- // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
+
+ logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
+
+ executeHolder.close()
+ if (abandoned) {
+ // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
}
}
private[connect] def getExecuteHolder(key: ExecuteKey): Option[ExecuteHolder] = {
- executionsLock.synchronized {
- executions.get(key)
- }
+ Option(executions.get(key))
}
private[connect] def removeAllExecutionsForSession(key: SessionKey): Unit = {
- val sessionExecutionHolders = executionsLock.synchronized {
- executions.filter(_._2.sessionHolder.key == key)
- }
- sessionExecutionHolders.foreach { case (_, executeHolder) =>
+ var sessionExecutionHolders = mutable.ArrayBuffer[ExecuteHolder]()
+ executions.forEach((_, executeHolder) => {
+ if (executeHolder.sessionHolder.key == key) {
+ sessionExecutionHolders += executeHolder
+ }
+ })
+
+ sessionExecutionHolders.foreach { executeHolder =>
Review Comment:
```
1. Thread 1: check closedTimeMs -> None.
2. Thread 2: set closedTimeMs.
3. Thread 2: removeAllExecutionsForSession.
4. Thread 1: add the executeHolder to the map.
```
I think this is fine, because in 4. it will only add the executeHolder to the map in SessionHolder, and we’re throwing that SessionHolder away at that point.
I think it is not possible to add it to the global map in ExecutionManager at that point, because in 1. it was already added there.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751619058
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -108,43 +110,50 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
* execution if still running, free all resources.
*/
private[connect] def removeExecuteHolder(key: ExecuteKey, abandoned: Boolean = false): Unit = {
- var executeHolder: Option[ExecuteHolder] = None
+ val executeHolder = executions.get(key)
+
+ if (executeHolder == null) {
+ return
+ }
+
+ // Put into abandonedTombstones before removing it from executions, so that the client ends up
+ // getting an INVALID_HANDLE.OPERATION_ABANDONED error on a retry.
+ if (abandoned) {
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
+ }
+
+ // Remove the execution from the map *after* putting it in abandonedTombstones.
+ executions.remove(key)
+ executeHolder.sessionHolder.removeExecuteHolder(executeHolder.operationId)
+
executionsLock.synchronized {
- executeHolder = executions.remove(key)
- executeHolder.foreach { e =>
- // Put into abandonedTombstones under lock, so that if it's accessed it will end up
- // with INVALID_HANDLE.OPERATION_ABANDONED error.
- if (abandoned) {
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
- e.sessionHolder.removeExecuteHolder(e.operationId)
- }
if (executions.isEmpty) {
lastExecutionTimeMs = Some(System.currentTimeMillis())
}
- logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
}
- // close the execution outside the lock
- executeHolder.foreach { e =>
- e.close()
- if (abandoned) {
- // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
+
+ logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
+
+ executeHolder.close()
+ if (abandoned) {
+ // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
}
}
private[connect] def getExecuteHolder(key: ExecuteKey): Option[ExecuteHolder] = {
- executionsLock.synchronized {
- executions.get(key)
- }
+ Option(executions.get(key))
}
private[connect] def removeAllExecutionsForSession(key: SessionKey): Unit = {
- val sessionExecutionHolders = executionsLock.synchronized {
- executions.filter(_._2.sessionHolder.key == key)
- }
- sessionExecutionHolders.foreach { case (_, executeHolder) =>
+ var sessionExecutionHolders = mutable.ArrayBuffer[ExecuteHolder]()
+ executions.forEach((_, executeHolder) => {
+ if (executeHolder.sessionHolder.key == key) {
+ sessionExecutionHolders += executeHolder
+ }
+ })
+
+ sessionExecutionHolders.foreach { executeHolder =>
Review Comment:
Ah, that's a valid point, thanks! So, the following execution sequence will be really problematic.
Thread 1: check closedTimeMs -> None.
Thread 2: set closedTimeMs.
Thread 2: removeAllExecutionsForSession.
Thread 1: add the executeHolder to the map.
Furthermore, closedTimeMs is a plain Option, which is *semantically* not quite thread-safe (it will be 16-byte long so if not perfectly aligned in memory, it is unsafe).
I'll try to fix those issues in this PR too!
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751639159
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -108,43 +110,50 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
* execution if still running, free all resources.
*/
private[connect] def removeExecuteHolder(key: ExecuteKey, abandoned: Boolean = false): Unit = {
- var executeHolder: Option[ExecuteHolder] = None
+ val executeHolder = executions.get(key)
+
+ if (executeHolder == null) {
+ return
+ }
+
+ // Put into abandonedTombstones before removing it from executions, so that the client ends up
+ // getting an INVALID_HANDLE.OPERATION_ABANDONED error on a retry.
+ if (abandoned) {
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
+ }
+
+ // Remove the execution from the map *after* putting it in abandonedTombstones.
+ executions.remove(key)
+ executeHolder.sessionHolder.removeExecuteHolder(executeHolder.operationId)
+
executionsLock.synchronized {
- executeHolder = executions.remove(key)
- executeHolder.foreach { e =>
- // Put into abandonedTombstones under lock, so that if it's accessed it will end up
- // with INVALID_HANDLE.OPERATION_ABANDONED error.
- if (abandoned) {
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
- e.sessionHolder.removeExecuteHolder(e.operationId)
- }
if (executions.isEmpty) {
lastExecutionTimeMs = Some(System.currentTimeMillis())
}
- logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
}
- // close the execution outside the lock
- executeHolder.foreach { e =>
- e.close()
- if (abandoned) {
- // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
+
+ logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
+
+ executeHolder.close()
+ if (abandoned) {
+ // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
}
}
private[connect] def getExecuteHolder(key: ExecuteKey): Option[ExecuteHolder] = {
- executionsLock.synchronized {
- executions.get(key)
- }
+ Option(executions.get(key))
}
private[connect] def removeAllExecutionsForSession(key: SessionKey): Unit = {
- val sessionExecutionHolders = executionsLock.synchronized {
- executions.filter(_._2.sessionHolder.key == key)
- }
- sessionExecutionHolders.foreach { case (_, executeHolder) =>
+ var sessionExecutionHolders = mutable.ArrayBuffer[ExecuteHolder]()
+ executions.forEach((_, executeHolder) => {
+ if (executeHolder.sessionHolder.key == key) {
+ sessionExecutionHolders += executeHolder
+ }
+ })
+
+ sessionExecutionHolders.foreach { executeHolder =>
Review Comment:
Ignore the thread-safety of closedTimeMs -> it is marked volatile so should be safe.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751665785
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,28 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ } else if (getAbandonedTombstone(executeHolder.key).isDefined) {
+ executions.remove(executeHolder.key)
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ }
+
+ sessionHolder.addExecuteHolder(executeHolder)
+
executionsLock.synchronized {
- // Check if the operation already exists, both in active executions, and in the graveyard
- // of tombstones of executions that have been abandoned.
- // The latter is to prevent double execution when a client retries execution, thinking it
- // never reached the server, but in fact it did, and already got removed as abandoned.
- if (executions.get(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- if (getAbandonedTombstone(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- sessionHolder.addExecuteHolder(executeHolder)
- executions.put(executeHolder.key, executeHolder)
Review Comment:
For https://github.com/apache/spark/pull/48034#discussion_r1751660500 it is important that we now first put the execution in the global map, and then into the session map.
But, is it now possible in some operations that an execution will be retrieved from the global map, validating that it exists, but then it will assume that it's present in the session map but race before it's added there?
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1755031877
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,30 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
Review Comment:
you could move the key generation out :)...
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751619058
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -108,43 +110,50 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
* execution if still running, free all resources.
*/
private[connect] def removeExecuteHolder(key: ExecuteKey, abandoned: Boolean = false): Unit = {
- var executeHolder: Option[ExecuteHolder] = None
+ val executeHolder = executions.get(key)
+
+ if (executeHolder == null) {
+ return
+ }
+
+ // Put into abandonedTombstones before removing it from executions, so that the client ends up
+ // getting an INVALID_HANDLE.OPERATION_ABANDONED error on a retry.
+ if (abandoned) {
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
+ }
+
+ // Remove the execution from the map *after* putting it in abandonedTombstones.
+ executions.remove(key)
+ executeHolder.sessionHolder.removeExecuteHolder(executeHolder.operationId)
+
executionsLock.synchronized {
- executeHolder = executions.remove(key)
- executeHolder.foreach { e =>
- // Put into abandonedTombstones under lock, so that if it's accessed it will end up
- // with INVALID_HANDLE.OPERATION_ABANDONED error.
- if (abandoned) {
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
- e.sessionHolder.removeExecuteHolder(e.operationId)
- }
if (executions.isEmpty) {
lastExecutionTimeMs = Some(System.currentTimeMillis())
}
- logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
}
- // close the execution outside the lock
- executeHolder.foreach { e =>
- e.close()
- if (abandoned) {
- // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
- abandonedTombstones.put(key, e.getExecuteInfo)
- }
+
+ logInfo(log"ExecuteHolder ${MDC(LogKeys.EXECUTE_KEY, key)} is removed.")
+
+ executeHolder.close()
+ if (abandoned) {
+ // Update in abandonedTombstones: above it wasn't yet updated with closedTime etc.
+ abandonedTombstones.put(key, executeHolder.getExecuteInfo)
}
}
private[connect] def getExecuteHolder(key: ExecuteKey): Option[ExecuteHolder] = {
- executionsLock.synchronized {
- executions.get(key)
- }
+ Option(executions.get(key))
}
private[connect] def removeAllExecutionsForSession(key: SessionKey): Unit = {
- val sessionExecutionHolders = executionsLock.synchronized {
- executions.filter(_._2.sessionHolder.key == key)
- }
- sessionExecutionHolders.foreach { case (_, executeHolder) =>
+ var sessionExecutionHolders = mutable.ArrayBuffer[ExecuteHolder]()
+ executions.forEach((_, executeHolder) => {
+ if (executeHolder.sessionHolder.key == key) {
+ sessionExecutionHolders += executeHolder
+ }
+ })
+
+ sessionExecutionHolders.foreach { executeHolder =>
Review Comment:
Ah, that's a valid point, thanks! So, the following execution sequence will be really problematic.
Thread 1: check closedTimeMs -> None.
Thread 2: set closedTimeMs.
Thread 2: removeAllExecutionsForSession.
Thread 1: add the executeHolder to the map.
Furthermore, closedTimeMs is a plain Option, which is *semantically* not quite thread-safe (it will be 16-byte long so if not perfectly aligned in memory, it is true).
I'll try to fix those issues in this PR too!
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751714503
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -61,6 +60,7 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
private var lastExecutionTimeMs: Option[Long] = Some(System.currentTimeMillis())
Review Comment:
Yes, it seems safe, couldn't find a counter example. -> anyway I need to add a check (!empty) before setting it to None.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #48034:
URL: https://github.com/apache/spark/pull/48034#issuecomment-2343622499
@grundprinzip Hey Martin, can you review and merge this change? Thanks!
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #48034:
URL: https://github.com/apache/spark/pull/48034#issuecomment-2340666390
@hvanhovell @HyukjinKwon Hello Herman and Hyukjin, would you mind merging this PR? Thanks!
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #48034:
URL: https://github.com/apache/spark/pull/48034#issuecomment-2339872073
@juliuszsompolski Can you please review this PR too? Thanks!
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751678145
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -61,6 +60,7 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
private var lastExecutionTimeMs: Option[Long] = Some(System.currentTimeMillis())
Review Comment:
I think it's fine, because before setting it None, under lock, you check again whether the map is empty. So if some new execution was added concurrently, this will be fine. And on the side that adds that execution, it first puts it into the map (so the map becomes non-empty) and then sets it to non-None under lock.
So if we enter this block setting it to None after some concurrent add already added something to the map, we're fine because map will be non empty and we won't set it to None.
If the map was in fact empty, that means that any concurrent add didn't yet put the execution in the map. Then that add will set it back to non-None right after putting the execution in the map.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751639015
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -61,6 +60,7 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
private var lastExecutionTimeMs: Option[Long] = Some(System.currentTimeMillis())
Review Comment:
`lastExecutionTimeMs` was originally introduced so that some system managing the Spark cluster could check if it's idle and for how long it's idle. Such a monitoring system would call SparkConnectExecutionManager.listActiveExecutions to either know what queries are running in the system, or otherwise learn since when the system is idle.
I am not sure if it got to be used in practice.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1754501627
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,28 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ } else if (getAbandonedTombstone(executeHolder.key).isDefined) {
+ executions.remove(executeHolder.key)
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ }
+
+ sessionHolder.addExecuteHolder(executeHolder)
+
executionsLock.synchronized {
- // Check if the operation already exists, both in active executions, and in the graveyard
- // of tombstones of executions that have been abandoned.
- // The latter is to prevent double execution when a client retries execution, thinking it
- // never reached the server, but in fact it did, and already got removed as abandoned.
- if (executions.get(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- if (getAbandonedTombstone(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- sessionHolder.addExecuteHolder(executeHolder)
- executions.put(executeHolder.key, executeHolder)
Review Comment:
-> https://github.com/apache/spark/pull/48071
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1751626231
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,28 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ } else if (getAbandonedTombstone(executeHolder.key).isDefined) {
+ executions.remove(executeHolder.key)
+ throw new SparkSQLException(
+ errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
+ messageParameters = Map("handle" -> executeHolder.operationId))
+ }
+
+ sessionHolder.addExecuteHolder(executeHolder)
+
executionsLock.synchronized {
- // Check if the operation already exists, both in active executions, and in the graveyard
- // of tombstones of executions that have been abandoned.
- // The latter is to prevent double execution when a client retries execution, thinking it
- // never reached the server, but in fact it did, and already got removed as abandoned.
- if (executions.get(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- if (getAbandonedTombstone(executeHolder.key).isDefined) {
- throw new SparkSQLException(
- errorClass = "INVALID_HANDLE.OPERATION_ABANDONED",
- messageParameters = Map("handle" -> executeHolder.operationId))
- }
- sessionHolder.addExecuteHolder(executeHolder)
- executions.put(executeHolder.key, executeHolder)
Review Comment:
Yes, this will be fine; "exections" in SessionHolder was anyway used outside executionsLock previously.
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1755148377
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,30 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
Review Comment:
Now, the key is generated outside ExecuteHolder :)
--
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-49544][CONNECT] Replace coarse-locking in SparkConnectExecutionManager with ConcurrentMap [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #48034:
URL: https://github.com/apache/spark/pull/48034#discussion_r1755021926
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala:
##########
@@ -77,26 +77,30 @@ private[connect] class SparkConnectExecutionManager() extends Logging {
request.getSessionId,
previousSessionId)
val executeHolder = new ExecuteHolder(request, sessionHolder)
+
+ // Check if the operation already exists, either in the active execution map, or in the
+ // graveyard of tombstones of executions that have been abandoned. The latter is to prevent
+ // double executions when the client retries, thinking it never reached the server, but in
+ // fact it did, and already got removed as abandoned.
+ if (executions.putIfAbsent(executeHolder.key, executeHolder) != null) {
Review Comment:
-> key is generated within ExecuteHolder, so deferring instantiation won't work. I'll just checking the tombstone part into the lambda.
--
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