You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2023/01/08 16:29:37 UTC
[GitHub] [spark] ivoson opened a new pull request, #39459: [WIP][SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache.
ivoson opened a new pull request, #39459:
URL: https://github.com/apache/spark/pull/39459
### What changes were proposed in this pull request?
Make rdd block(rdd cache) available only when a task generate the block succeed.
### Why are the changes needed?
Fixing the bug as described in [SPARK-41497)](https://issues.apache.org/jira/browse/SPARK-41497)
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Adding new UT.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1071815096
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
If a block is not visible, and it was not computed in this task, we should not be using that block's cached data at all - even if cached here or elsewhere.
Particularly for unordered or indeterminate computation, this can impact the accumulators generated - and it wont match the result iterator returned.
So we cant simply call `makeIterator` here - as @Ngone51 suggested, we should pass the flag to `getOrElseUpdate` and decide there based on visibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1098177596
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Not sure I follow - we cannot leverage the block until it is explicitly marked visible (unless within same task).
If the executor is lost before that happens, then we can never make it visible - and so fine to clean up from invisible blocks, right ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108038496
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
+ makeIterator()
Review Comment:
Creating the iterator is not enough, we have to drain it to update accumulators ...
`Utils.getIteratorSize(makeIterator())` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112123108
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). The happens right after the task finished
+ // computing/caching the block only when the block is not visible yet. And the rdd
+ // block will be marked as visible only when the corresponding task finished successfully.
+ context.reply(updateRDDBlockTaskInfo(blockId, taskId))
+
+ case GetRDDBlockVisibility(blockId) =>
+ // Get the visibility status of a specific rdd block.
+ if (!trackingCacheVisibility) {
Review Comment:
Sure, updated.
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). The happens right after the task finished
Review Comment:
Thanks, updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm closed pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm closed pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
URL: https://github.com/apache/spark/pull/39459
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1452774796
Thanks for the review and help on this. @mridulm @Ngone51
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1123978338
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully. This is relevant in context of consistent accumulator status.")
+ .version("3.4.0")
Review Comment:
According to the commit history, it seems this pr lands at master branch(3.5.0). Should this backport to 3.4.0 or update the version ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1103818933
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
A bit unclear on this actually ... why would a block status be cached on an executor (other than driver) which is not hosting the block ?
If executor is lost, driver will update its state as well due to the loss (evict the block) ... if there are other copies of the block, continues to be visible (since block exists). If task fails before block transitions to visible (or executor loss), driver will evict all blocks.
Note, when executor is fetching a block for computation from remote executor - it should be 'seeing' a block only if it is visible.
(If block is locally available due to replication - then it is invisible until marked visible. And so same logic as above would work).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1106753599
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> With above mechanism, do you think we still need another cache to store the visiblity information in executor or do we also need to cache the state in executors not having the cached block data stored?
You are right, we dont need it: since we are already checking for `blockInfoManager.isRDDBlockVisible(blockId)` - it will cover the case of (1) - and we will always query in case block is available, and we have to distinguish (2).
(2.1) would be an optimization we can attempt later on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1100333545
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Let me explain this further. If we track visible blocks, it's clear that we always know which blocks are visible.
If we track invisible blocks, the way we consider a block as visible is that at least one block exists and it's not in invisible lists. So if the existing blocks got lost, we will lose the information. Next time the cache is re-computed, we will do this again(firstly put it into invisible lists, then promote it to visible by removing it from invisible list once task finished successfully). And after doing the process again, the cache would be visible then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1071178376
##########
core/src/test/scala/org/apache/spark/AccumulatorSuite.scala:
##########
@@ -89,6 +89,38 @@ class AccumulatorSuite extends SparkFunSuite with Matchers with LocalSparkContex
assert(AccumulatorContext.get(100000).isEmpty)
}
+
+ test("SPARK-41497: accumulators should be reported in the case of task retry with rdd cache") {
+ // Set up a cluster with 2 executors
+ val conf = new SparkConf()
+ .setMaster("local-cluster[2, 1, 1024]").setAppName("TaskSchedulerImplSuite")
Review Comment:
The app name is not appropriate here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1096526714
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
One way in my mind is that, we cache the results(for visible rdd blocks) in the block manager. Once the rdd is removed, a broadcast message will be sent to each BlockManager to clean the cache.
I am wondering is it worth to do this? Since there is locality scheduling, if tasks got scheduled to the executor where cached block exists, there'll be no calls to master.
Let me know your thoughts about this, thanks. cc @mridulm @Ngone51
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108037035
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
Btw, can you add a todo for this here @ivoson ? Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1451232733
Hi @mridulm @Ngone51 thanks for the review. Please let me know if you have any other comments/concerns for 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
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1119075926
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,74 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
Review Comment:
Can you adress this @ivoson ? Thx
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
Review Comment:
Can you adress this @ivoson ? Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118028223
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerStorageEndpoint.scala:
##########
@@ -81,6 +81,8 @@ class BlockManagerStorageEndpoint(
case ReplicateBlock(blockId, replicas, maxReplicas) =>
context.reply(blockManager.replicateBlock(blockId, replicas.toSet, maxReplicas))
+ case MarkRDDBlockAsVisible(blockId) =>
Review Comment:
Of course. Updated.
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,160 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
Review Comment:
Thanks, done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090953345
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
This is a good point - and couple of issues that come to mind:
* We do not disable speculative execution for indeterminate computation currently - and data generated from two task attempts can vary (which could be cached, and so differ for same partition)
* We do not invalidate previously cached data, when there is a stage re-execution due to failures for an indeterminate computation,
This is something we should revisit @Ngone51, @jiangxb1987 - thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118028028
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -150,6 +150,12 @@ private[storage] class BlockInfoManager extends Logging {
*/
private[this] val blockInfoWrappers = new ConcurrentHashMap[BlockId, BlockInfoWrapper]
+ /**
+ * Record invisible rdd blocks stored in the block manager, entries will be removed when blocks
+ * are marked as visible or blocks are removed by [[removeBlock()]].
+ */
+ private[spark] val invisibleRDDBlocks = ConcurrentHashMap.newKeySet[RDDBlockId]
Review Comment:
Make sense. Thanks, updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1103943515
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
> If executor is lost, driver will update its state as well due to the loss (evict the block) ... if there are other copies of the block, continues to be visible (since block exists). If task fails before block transitions to visible (or executor loss), driver will evict all blocks.
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1096527017
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
Review Comment:
Updated.
##########
core/src/test/scala/org/apache/spark/AccumulatorSuite.scala:
##########
@@ -89,6 +89,38 @@ class AccumulatorSuite extends SparkFunSuite with Matchers with LocalSparkContex
assert(AccumulatorContext.get(100000).isEmpty)
}
+
+ test("SPARK-41497: accumulators should be reported in the case of task retry with rdd cache") {
+ // Set up a cluster with 2 executors
+ val conf = new SparkConf()
+ .setMaster("local-cluster[2, 1, 1024]").setAppName("TaskSchedulerImplSuite")
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1096527082
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
+ // Block exists and not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private[spark] def getOrElseUpdate[T](
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1106753599
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> With above mechanism, do you think we still need another cache to store the visiblity information in executor or do we also need to cache the state in executors not having the cached block data stored?
You are right, we dont need it: since we are already checking for `blockInfoManager.isRDDBlockVisible(blockId)`. Agree
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1107215072
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Will try to make the change, the scenario I mentioned above would not be a common case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090482420
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
Thanks, will make 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090483687
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
Yes, I've thought about this. Then we will add a mechanism to clean the cache. Let me re-think about 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090483687
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
Yes, I've thought about this. Then we will add a mechanism to clean the cache then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090479561
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -180,6 +186,14 @@ private[storage] class BlockInfoManager extends Logging {
// ----------------------------------------------------------------------------------------------
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ blockInfoWrappers.containsKey(blockId) && visibleRDDBlocks.contains(blockId)
+ }
Review Comment:
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
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1098176808
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
What I meant was, cache visibility - once a block becomes visible, we do not revert it back to being invisible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1098656449
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
Review Comment:
Wait, isn't it still the same as the previous way?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101665418
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Yes, that is the idea - block gets added to invisible list initially (since task has not completed), and then gets promoted to becoming visible (when task completes).
If existing blocks are lost - why would you need that information as they are gone ? In other words, how is it different from today's situation (without visibility) - if a block is lost, it is no longer in system.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111377539
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1470,28 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // Cached blocks are always visible if the feature flag is disabled.
+ if (!trackingCacheVisibility) {
+ return true
+ }
+
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+
+ if(master.isRDDBlockVisible(blockId)) {
+ // Cache the visibility status if block exists.
+ blockInfoManager.tryAddVisibleBlock(blockId)
+ true
Review Comment:
Should we still return `true` if `blockInfoManager` doesn't contain the block?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1125602608
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully. This is relevant in context of consistent accumulator status.")
+ .version("3.4.0")
Review Comment:
Sure, will do.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1071163490
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
Review Comment:
This way could lead to low efficiency if we just finish the remote value fetch but the cache is invisible, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1383971139
cc @juliuszsompolski @cloud-fan
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118005755
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
+ val res = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ if (res == null && trackingCacheVisibility) {
+ // Added to invisible blocks if it doesn't exist before.
+ blockId.asRDDId.foreach(invisibleRDDBlocks.add)
Review Comment:
Currently we syncrhonized all the write operations, and also the read operations which need to check both the variables. For other read operations, since the variabled won't change, I think synchronized block for such read operations may be not necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118013777
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -728,7 +800,22 @@ class BlockManagerMasterEndpoint(
}
if (storageLevel.isValid) {
+ val firstBlock = locations.isEmpty
locations.add(blockManagerId)
+
+ blockId.asRDDId.foreach { rddBlockId =>
+ (trackingCacheVisibility, firstBlock) match {
+ case (true, true) =>
+ // Mark as invisible for the first block.
+ invisibleRDDBlocks.add(rddBlockId)
+ case (true, false) if !invisibleRDDBlocks.contains(rddBlockId) =>
+ // If the rdd block is already visible, ask storage manager to update the visibility
+ // status.
+ blockManagerInfo(blockManagerId).storageEndpoint
+ .ask[Unit](MarkRDDBlockAsVisible(rddBlockId))
Review Comment:
> So ensure that the replica is also marked as visible, right ?
Yes, this is one scenario. Another scenario is that once a task failed fetching the cached block from remote executor, it'll compute and cache the block again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118028179
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +220,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). And this happens right after the task finished
+ // computing/caching the block only when the block is not visible yet. And the rdd
+ // block will be marked as visible when the corresponding task finished successfully.
+ context.reply(updateRDDBlockTaskInfo(blockId, taskId))
+
+ case GetRDDBlockVisibility(blockId) =>
+ // Get the visibility status of a specific rdd block.
+ context.reply(isRDDBlockVisible(blockId))
+
+ case UpdateRDDBlockVisibility(taskId, visible) =>
+ // This is to report the information that whether rdd blocks computed by task(with `taskId`)
+ // can be turned to be visible. This is reported by DAGScheduler right after task completes.
+ // If the task finished successfully, rdd blocks can be turned to be visible, otherwise rdd
+ // blocks' visibility status won't change.
+ context.reply(updateRDDBlockVisibility(taskId, visible))
+ }
+
+ private def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ if (trackingCacheVisibility) {
+ blockLocations.containsKey(blockId) &&
+ blockLocations.get(blockId).nonEmpty && !invisibleRDDBlocks.contains(blockId)
+ } else {
+ // Blocks should always be visible if the feature flag is disabled.
+ true
+ }
+ }
+
+ private def updateRDDBlockVisibility(taskId: Long, visible: Boolean): Unit = {
+ if (!trackingCacheVisibility) {
+ // Do nothing if the feature flag is disabled.
+ return
+ }
+
+ if (visible) {
+ tidToRddBlockIds.get(taskId).foreach { blockIds =>
+ blockIds.foreach { blockId =>
+ invisibleRDDBlocks.remove(blockId)
+ // Ask block managers to update the visibility status.
+ val msg = MarkRDDBlockAsVisible(blockId)
+ getLocations(blockId).flatMap(blockManagerInfo.get).foreach { managerInfo =>
+ managerInfo.storageEndpoint.ask[Unit](msg)
+ }
+ }
+ }
+ }
Review Comment:
Updated, and this is the jira: https://issues.apache.org/jira/browse/SPARK-42582
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1122750441
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
Review Comment:
done, 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
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101657140
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
To clarify, (my previous comment was poorly worded).
A block would be invisible for three reasons primarily (and minor variants of these) :
a) Some task T1 is still actively working on the partition which computed it.
b) Some task T1 on E1 generated the partition, which got pushed to E2 due to replication - and is not yet visible, since T1 is still running.
c) Some task T1 on E1 generated B1, but failed - and so block exists on E1 but is invisible.
In these cases, a block which is invisible would become visible once the task T1 successfully complete - or (in case of failure), is dropped (and so removed).
So the transition is only from invisible -> visible or invisible -> dropped, there is no state -> invisible (unless you count new_block -> invisible :-) )
When a block transitions to becoming visible (due to an explicit message from driver to the executor which generated it, and to the replicas), it remains in that state until it is dropped explicitly (which is again another message from driver).
So the proposal was:
1) Maintain cached state of block at executor - once it is visible, it never changes from that.
2) If invisible, it could be a potential race (T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1) - so we have to query driver, but we can cache the result if visible.
2.1) We can minimize this impact by dropping all B1's when the task T1 fails at E1 (driver would send the update anyway).
So essentially, we can cache only if the block is visible - if it is not visible, we always have to query driver (with (2.1) to minimize needing to query failures)
Thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108405970
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
> Yes, if the block already exists in the node, here just computed it again but not putting the result.
Shouldn't we replace the block with the new computed result?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1107226569
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
Yes, if the block already exists in the node, here just computed it again but not putting the result.
> Could we force put the result iterator even if the block exists in this case?
This will need to modify the locking mechanism a little bit. This is also a issue about indeterminate operation, for derterminate operations no need to replace the cache here. I am wondering can the inderterminate framework cover this case with some other solutions?
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
Yes, if the block already exists in the node, here just computed it again but not putting the result.
> Could we force put the result iterator even if the block exists in this case?
This will need to modify the locking mechanism a little bit. This is also a issue about indeterminate operation, for derterminate operations no need to replace the cache here. I am wondering can the inderterminate framework cover this case with some other solutions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1110773664
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
Yes, the recomputation is only for updating accumulators. The reulst should be the same unless the result is indeterminate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111226512
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Updated. Please take a look. Thanks. cc @Ngone51 @mridulm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118023335
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -728,7 +800,22 @@ class BlockManagerMasterEndpoint(
}
if (storageLevel.isValid) {
+ val firstBlock = locations.isEmpty
locations.add(blockManagerId)
+
+ blockId.asRDDId.foreach { rddBlockId =>
+ (trackingCacheVisibility, firstBlock) match {
+ case (true, true) =>
+ // Mark as invisible for the first block.
+ invisibleRDDBlocks.add(rddBlockId)
+ case (true, false) if !invisibleRDDBlocks.contains(rddBlockId) =>
+ // If the rdd block is already visible, ask storage manager to update the visibility
+ // status.
+ blockManagerInfo(blockManagerId).storageEndpoint
+ .ask[Unit](MarkRDDBlockAsVisible(rddBlockId))
Review Comment:
Ah yes, good point. Makes sense to do this irrespective of the path to block creation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1070632692
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -76,6 +76,8 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
Review Comment:
Thanks, fixed. When task completes, will clean up the records.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1096520993
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Just found one problem if we track the invisible RDD blocks.
If we track the invisible RDD blocks, then we would mark a RDD block as visible(cache can be used) only when it exists in `blockLocations` and not exists in `invisibleRDDBlocks`. When `blockLocations` removed the block(could be caused by executor lost), we will lose the information. Then the new cached data won't be leveraged as soon as possible(right after the cache is generated/reported 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1103799711
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
Hi @mridulm, in current implementation, once a block turns to be visible, driver would send a broadcast message to executors having the cached block data stored to mark the block as visible.
The state `visibleRDDBlocks` is cached in [BlockerInfoManager](https://github.com/apache/spark/pull/39459/files#diff-fdee2ef66ad5bea5323506395b453145c74f47c8da092dcacd34a66190a20a15). It is kind of cached visiblity state in executor side but only in executors which have the cached block stored. This is done in a push-based update style.
With above mechanism, do you think we still need another cache to store the visiblity information in executor or do we also need to cache the state in executors not having the cached block data stored?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1117987993
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -180,6 +186,27 @@ private[storage] class BlockInfoManager extends Logging {
// ----------------------------------------------------------------------------------------------
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ if (trackingCacheVisibility) {
+ invisibleRDDBlocks.synchronized {
+ blockInfoWrappers.containsKey(blockId) && !invisibleRDDBlocks.contains(blockId)
+ }
+ } else {
+ // Always be visible if the feature flag is disabled.
+ true
+ }
+ }
+
+ private[spark] def tryMarkBlockAsVisible(blockId: RDDBlockId): Unit = {
+ if (trackingCacheVisibility) {
+ invisibleRDDBlocks.synchronized {
+ if (blockInfoWrappers.containsKey(blockId)) {
+ invisibleRDDBlocks.remove(blockId)
Review Comment:
We dont need the `containsKey` check ...
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -150,6 +150,12 @@ private[storage] class BlockInfoManager extends Logging {
*/
private[this] val blockInfoWrappers = new ConcurrentHashMap[BlockId, BlockInfoWrapper]
+ /**
+ * Record invisible rdd blocks stored in the block manager, entries will be removed when blocks
+ * are marked as visible or blocks are removed by [[removeBlock()]].
+ */
+ private[spark] val invisibleRDDBlocks = ConcurrentHashMap.newKeySet[RDDBlockId]
Review Comment:
All accesses and updates to `invisibleRDDBlocks` are within the `synchronized` block - make it a regular `HashSet` instead ?
Also, let us make it `private[this]`.
For the test cases which are directly accessing variable right now, we can add a
```
private[storage] def containsInvisibleRDDBlock(blockId: RDDBlockId): Boolean = {
invisibleRDDBlocks.synchronized {
invisibleRDDBlocks.contains(blockId)
}
}
```
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,74 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
Review Comment:
Remove default value for `isCacheVisible`
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,160 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
+ val res2 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully and reported block task info.
+ assert(res2.isLeft && computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
Review Comment:
Note:
This specific check is not very accurate ... it is the current code's implementation detail. and not the behavior we want to test for. (I am fine with leaving this check here for now though !)
For example, we can skip reporting to master if the taskId has computed a particular block already (especially if it is in the invisible set). In actual spark usage, this cannot happen - and so this is an artifact of test cases IMO.
A better check would be to test for visibility is false, and computed == true (which we do already) - which is the behavior we want to test for.
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -139,7 +139,7 @@ private[storage] object BlockInfo {
*
* This class is thread-safe.
*/
-private[storage] class BlockInfoManager extends Logging {
+private[storage] class BlockInfoManager(trackingCacheVisibility: Boolean = true) extends Logging {
Review Comment:
Make it `false` by default ?
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully. This is relevant in context of consistent accumulator status.")
+ .version("3.4.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
Let us mark it `false` by default ?
Anyway we have to revisit this when fixing the general data correctness issue, with persisted blocks, we identified as part of this review.
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
Review Comment:
Let us if/else around `trackingCacheVisibility`.
```
if (trackingCacheVisibility) {
// existing code in this PR
} else {
blockInfoWrappers.putIfAbsent(blockId, wrapper)
}
```
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -525,6 +562,7 @@ private[storage] class BlockInfoManager extends Logging {
blockInfoWrappers.clear()
readLocksByTask.clear()
writeLocksByTask.clear()
+ invisibleRDDBlocks.clear()
Review Comment:
Note: when changing type of `invisibleRDDBlocks` to `HashSet`, this needs to be within `invisibleRDDBlocks.synchronized` as well
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +220,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). And this happens right after the task finished
+ // computing/caching the block only when the block is not visible yet. And the rdd
+ // block will be marked as visible when the corresponding task finished successfully.
+ context.reply(updateRDDBlockTaskInfo(blockId, taskId))
+
+ case GetRDDBlockVisibility(blockId) =>
+ // Get the visibility status of a specific rdd block.
+ context.reply(isRDDBlockVisible(blockId))
+
+ case UpdateRDDBlockVisibility(taskId, visible) =>
+ // This is to report the information that whether rdd blocks computed by task(with `taskId`)
+ // can be turned to be visible. This is reported by DAGScheduler right after task completes.
+ // If the task finished successfully, rdd blocks can be turned to be visible, otherwise rdd
+ // blocks' visibility status won't change.
+ context.reply(updateRDDBlockVisibility(taskId, visible))
+ }
+
+ private def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ if (trackingCacheVisibility) {
+ blockLocations.containsKey(blockId) &&
+ blockLocations.get(blockId).nonEmpty && !invisibleRDDBlocks.contains(blockId)
+ } else {
+ // Blocks should always be visible if the feature flag is disabled.
+ true
Review Comment:
This should not happen iirc (here and in the other methods added in this class) ... could be a precondition check.
But it is fine to include it for now.
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +220,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). And this happens right after the task finished
+ // computing/caching the block only when the block is not visible yet. And the rdd
+ // block will be marked as visible when the corresponding task finished successfully.
+ context.reply(updateRDDBlockTaskInfo(blockId, taskId))
+
+ case GetRDDBlockVisibility(blockId) =>
+ // Get the visibility status of a specific rdd block.
+ context.reply(isRDDBlockVisible(blockId))
+
+ case UpdateRDDBlockVisibility(taskId, visible) =>
+ // This is to report the information that whether rdd blocks computed by task(with `taskId`)
+ // can be turned to be visible. This is reported by DAGScheduler right after task completes.
+ // If the task finished successfully, rdd blocks can be turned to be visible, otherwise rdd
+ // blocks' visibility status won't change.
+ context.reply(updateRDDBlockVisibility(taskId, visible))
+ }
+
+ private def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ if (trackingCacheVisibility) {
+ blockLocations.containsKey(blockId) &&
+ blockLocations.get(blockId).nonEmpty && !invisibleRDDBlocks.contains(blockId)
+ } else {
+ // Blocks should always be visible if the feature flag is disabled.
+ true
+ }
+ }
+
+ private def updateRDDBlockVisibility(taskId: Long, visible: Boolean): Unit = {
+ if (!trackingCacheVisibility) {
+ // Do nothing if the feature flag is disabled.
+ return
+ }
+
+ if (visible) {
+ tidToRddBlockIds.get(taskId).foreach { blockIds =>
+ blockIds.foreach { blockId =>
+ invisibleRDDBlocks.remove(blockId)
+ // Ask block managers to update the visibility status.
+ val msg = MarkRDDBlockAsVisible(blockId)
+ getLocations(blockId).flatMap(blockManagerInfo.get).foreach { managerInfo =>
+ managerInfo.storageEndpoint.ask[Unit](msg)
+ }
+ }
+ }
+ }
Review Comment:
Note:
When `visible == false`, the task had failed - and we should be asking the block managers to evict the block.
This can come later, as part of the correctness fix though.
Can you add a TODO for it ? Please include reference to the correctness jira (for persisted block issue we discussed) in the comment as well. Thx
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -728,7 +800,22 @@ class BlockManagerMasterEndpoint(
}
if (storageLevel.isValid) {
+ val firstBlock = locations.isEmpty
locations.add(blockManagerId)
+
+ blockId.asRDDId.foreach { rddBlockId =>
+ (trackingCacheVisibility, firstBlock) match {
+ case (true, true) =>
+ // Mark as invisible for the first block.
+ invisibleRDDBlocks.add(rddBlockId)
+ case (true, false) if !invisibleRDDBlocks.contains(rddBlockId) =>
+ // If the rdd block is already visible, ask storage manager to update the visibility
+ // status.
+ blockManagerInfo(blockManagerId).storageEndpoint
+ .ask[Unit](MarkRDDBlockAsVisible(rddBlockId))
Review Comment:
To understand better, this is to handle case where block was replicated to another node - but in meantime the original task completed successfully (and so not invisible anymore).
So ensure that the replica is also marked as visible, right ?
Any other scenario where this is relevant ?
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerStorageEndpoint.scala:
##########
@@ -81,6 +81,8 @@ class BlockManagerStorageEndpoint(
case ReplicateBlock(blockId, replicas, maxReplicas) =>
context.reply(blockManager.replicateBlock(blockId, replicas.toSet, maxReplicas))
+ case MarkRDDBlockAsVisible(blockId) =>
Review Comment:
Can you add a comment for this as well ? Similar to `UpdateRDDBlockTaskInfo`, etc in `BlockManagerMasterEndpoint`
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,160 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
Review Comment:
Let us move this assertion to right after `verify` above ?
Essentially, that is the last check for the "Cache doesn't exist and is not visible." case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1106755256
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> I'm not trying to talk about the cache stuff. But just try to highlight this. So this should somehow be a behavior change, right? If T1 generates B1 in early time and T1 turns out to be a long-running task, it can be terrible for tasks like T2 which reads B1.
Do we actually have usecases where some other stage/task is depending on an earlier task generating a block as a prefix to its computation ?
Looks like a fairly brittle assumption, no ? (Unless I misunderstood the usecase here !)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108405970
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
> Yes, if the block already exists in the node, here just computed it again but not putting the result.
Shouldn't we replace the block with the new computed result?
Oh..the recomputation is only for accumulator updates? and the result should be same so no need to replace?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108038496
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
+ makeIterator()
Review Comment:
Creating the iterator is not enough, we have to drain it to update accumulators, right ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1104027347
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1787,6 +1792,12 @@ private[spark] class DAGScheduler(
case _: ExceptionFailure | _: TaskKilled => updateAccumulators(event)
case _ =>
}
+ if (trackingCacheVisibility) {
+ // Update rdd blocks' visibility status.
+ blockManagerMaster.updateRDDBlockVisibility(
Review Comment:
An asynchronous call seems enough here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112122792
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,150 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
+ val res2 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully and reported block task info.
+ assert(res2.isLeft && computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists and visible.
+ store.blockInfoManager.tryAddVisibleBlock(blockId)
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && store.isRDDBlockVisible(blockId))
+ val res3 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully but not report block task info.
+ assert(res3.isLeft && !computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
+ }
+
+ test("add block rdd visibility status") {
Review Comment:
Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112123963
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -150,6 +150,12 @@ private[storage] class BlockInfoManager extends Logging {
*/
private[this] val blockInfoWrappers = new ConcurrentHashMap[BlockId, BlockInfoWrapper]
+ /**
+ * Record visible rdd blocks stored in the block manager, entries will be removed
+ * by [[removeBlock()]]
+ */
+ private[spark] val visibleRDDBlocks = ConcurrentHashMap.newKeySet[RDDBlockId]
Review Comment:
Sounds good, will make the 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112113795
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1470,28 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // Cached blocks are always visible if the feature flag is disabled.
+ if (!trackingCacheVisibility) {
+ return true
+ }
+
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+
+ if(master.isRDDBlockVisible(blockId)) {
+ // Cache the visibility status if block exists.
+ blockInfoManager.tryAddVisibleBlock(blockId)
+ true
Review Comment:
Yes, I think so. Even though current executor doesn't have the cached block, we still can read the cache from a remote executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111412825
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). The happens right after the task finished
+ // computing/caching the block only when the block is not visible yet. And the rdd
+ // block will be marked as visible only when the corresponding task finished successfully.
+ context.reply(updateRDDBlockTaskInfo(blockId, taskId))
+
+ case GetRDDBlockVisibility(blockId) =>
+ // Get the visibility status of a specific rdd block.
+ if (!trackingCacheVisibility) {
Review Comment:
Could you push down the flag check into `isRDDBlockVisible` as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090953345
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
This is a good point - and couple of issues that come to mind:
* We do not disable speculative execution for indeterminate computation currently - and data generated from two task attempts can vary.
* We do not invalidate previously cached data, when there is a stage re-execution due to failures for an indeterminate computation,
This is something we should revisit @Ngone51, @jiangxb1987 - thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1098116796
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
No. We should fix that issue separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1098121312
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> Once the rdd is removed, a broadcast message will be sent to each BlockManager to clean the cache.
I think the problem in this way is that it introduces the race condition between the visible cache and the broadcast message.
`master.isRDDBlockVisible(blockId)` shouldn't be frequently called, right? I thought `blockInfoManager.isRDDBlockVisible(blockId)` would be the most case for us, 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1103794439
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
> If existing blocks are lost - why would you need that information as they are gone? In other words, how is it different from today's situation (without visibility) - if a block is lost, it is no longer in system.
Here is an example for the scenario I am trying to describe:
1. we have a cached block rdd_1_1 which has been successfully cached and marked as visible.
2. the cached block got lost due to executor lost;
3. another task on rdd1 got submitted and the 1st attempt failed after putting the cache block rdd_1_1, for the 2nd attempts, things could be different here:
a. if we still have the visiblily status, the 2nd attempt can use the cached block directly;
b otherwise, we still need to do the computing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1070632750
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,6 +1325,64 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ var getIterator = makeIterator
+
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ if (master.isRDDBlockVisible(blockId)) { // Read from cache only when the block is visible.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
+ // Initially we hold no locks on this block.
+ } else {
+ // Need to compute the block, since the block maybe already exists, force
+ // compute the block here.
+ val iterator = makeIterator()
+ getIterator = () => iterator
+ }
+
+ doPutIterator(blockId, getIterator, level, classTag, keepReadLock = true) match {
+ case None =>
+ // Report taskId -> blockId relationship to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
Review Comment:
In the latest implementation, the replication logic keeps the same as before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1071803105
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -180,6 +186,14 @@ private[storage] class BlockInfoManager extends Logging {
// ----------------------------------------------------------------------------------------------
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ blockInfoWrappers.containsKey(blockId) && visibleRDDBlocks.contains(blockId)
+ }
Review Comment:
Wondering if there is a possibility of both of these datastructures diverging - and driver ending up with metadata about blocks which have been removed ....
Need to take a closer look.
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Instead of keeping track of visible blocks, should we be tracking those which are not yet visible ?
The size would be much smaller - blocks by default get added to this `Set`, and get promoted to visible explicitly.
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
QQ: Should we be caching this instead of hitting driver each time ?
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
If a block is not visible, and it was not computed in this task, we should not be using that block at all - even if cached here or elsewhere.
Particularly for unordered or indeterminate computation, this can impact the accumulators generated - and it wont match the result returned.
So we cant simply call `makeIterator` here - as @Ngone51 suggested, we should pass the flag to `getOrElseUpdate` and decide there based on visibility.
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
+ // Block exists and not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private[spark] def getOrElseUpdate[T](
Review Comment:
Make this private instead ? And fix the tests which are depending on this directly to use `getOrElseUpdateRDDBlock` instead ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101657140
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
To clarify, (my previous comment was poorly worded).
A block would be invisible for three reasons primarily (and minor variants of these) :
a) Some task T1 is still actively working on the partition which computed it.
b) Some task T1 on E1 generated the partition, which got pushed to E2 due to replication - and is not yet visible, since T1 is still running.
c) Some task T1 on E1 generated B1, but failed - and so block exists on E1 but is invisible.
In these cases, a block which is invisible would become visible once the task T1 successfully complete - or (in case of failure), is dropped (and so removed).
So the transition is only from invisible -> visible or invisible -> dropped.
When a block transitions to becoming visible (due to an explicit message from driver to the executor which generated it, and to the replicas), it remains in that state until it is dropped explicitly (which is again another message from driver).
So the proposal was:
a) Maintain cached state of block at executor - once it is visible, it never changes from that.
b) If invisible, it could be a potential race (T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1) - so we have to query driver, but we can cache the result is visible.
b.1) We can minimize this impact by dropping all B1's when the task T1 fails at E1 (driver would send the update anyway).
So essentially, we can cache only if the block is visible - if it is not visible, we always have to query driver (with (b.1) to minimize needing to query failures)
Thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101657140
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
To clarify, (my previous comment was poorly worded).
A block would be invisible for three reasons primarily (and minor variants of these) :
a) Some task T1 is still actively working on the partition which computed it.
b) Some task T1 on E1 generated the partition, which got pushed to E2 due to replication - and is not yet visible, since T1 is still running.
c) Some task T1 on E1 generated B1, but failed - and so block exists on E1 but is invisible.
In these cases, a block which is invisible would become visible once the task T1 successfully complete - or (in case of failure), is dropped and becomes invisible.
So the transition is only from invisible -> visible or invisible -> dropped.
When a block transitions to becoming visible (due to an explicit message from driver to the executor which generated it, and to the replicas), it remains in that state until it is dropped explicitly (which is again another message from driver).
So the proposal was:
a) Maintain cached state of block at executor - once it is visible, it never changes from that.
b) If invisible, it could be a potential race (T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1) - so we have to query driver, but we can cache the result is visible.
b.1) We can minimize this impact by dropping all B1's when the task T1 fails at E1 (driver would send the update anyway).
So essentially, we can cache only if the block is visible - if it is not visible, we always have to query driver (with (b.1) to minimize needing to query failures)
Thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108397590
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> Do we actually have usecases where some other stage/task is depending on an earlier task generating a block as a prefix to its computation and the task itself has not completed ?
@mridulm Right...after a second thinking, it's not a valid concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1451486114
> @ivoson Did you get a chance to address the pending comments above ? ([here](https://github.com/apache/spark/pull/39459#discussion_r1117995723) and [here](https://github.com/apache/spark/pull/39459#discussion_r1117996417))
Thanks @mridulm for reminding. Missed these 2...
Updated, please take a look. 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
[GitHub] [spark] mridulm commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1451289469
@ivoson Did you get a chance to address the pending comments above ? ([here](https://github.com/apache/spark/pull/39459#discussion_r1117995723) and [here](https://github.com/apache/spark/pull/39459#discussion_r1117996417))
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1408399156
Hi, @mridulm @Ngone51 Sorry for the late response. Start working on this PR, just get back from a long vacation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090480637
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
+ // Block exists and not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private[spark] def getOrElseUpdate[T](
Review Comment:
Sounds more reasonable. Will make the 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111227309
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully.")
Review Comment:
Thanks, updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111400497
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +78,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
Review Comment:
visible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111415309
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,150 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
+ val res2 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully and reported block task info.
+ assert(res2.isLeft && computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists and visible.
+ store.blockInfoManager.tryAddVisibleBlock(blockId)
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && store.isRDDBlockVisible(blockId))
+ val res3 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully but not report block task info.
+ assert(res3.isLeft && !computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
+ }
+
+ test("add block rdd visibility status") {
Review Comment:
Please add the JIR number for the new tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090773361
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
> Particularly for unordered or indeterminate computation, this can impact the accumulators generated - and it wont match the result iterator returned.
There could be some other issues with such scenarios, like if we have blocks generated from different task attempts or cache re-compute, each cache may also have different data. Would this also be a problem?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1092858364
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
> We do not disable speculative execution for indeterminate computation currently - and data generated from two task attempts can vary (which could be cached, and so differ for same partition)
Right..in this case, the rdd block locations for different data can be attached to the same rdd block id. So the reader could get the different data for the same rdd block, which makes the rdd block data also indeterminate.
> We do not invalidate previously cached data, when there is a stage re-execution due to failures for an indeterminate computation,
This seems to be a missing point in the indeterminate computation framework. @cloud-fan could you help confirm?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1100268004
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> > Once the rdd is removed, a broadcast message will be sent to each BlockManager to clean the cache.
>
> I think the problem in this way is that it introduces the race condition between the visible cache and the broadcast message.
>
> `master.isRDDBlockVisible(blockId)` shouldn't be frequently called, right? I thought `blockInfoManager.isRDDBlockVisible(blockId)` would be the most case for us, no?
With locality scheduler I think `master.isRDDBlockVisible(blockId) ` should not be frequently called. If the assumption is true, then maybe we don't need to cache the results.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1420786054
@ivoson could you add a feature flag for the whole changes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1119073044
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,160 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
+ val res2 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully and reported block task info.
+ assert(res2.isLeft && computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
Review Comment:
That makes sense, thanks for clarifying !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118028518
##########
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:
##########
@@ -2266,6 +2270,160 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with PrivateMethodTe
}
}
+ test("SPARK-41497: getOrElseUpdateRDDBlock do compute based on cache visibility statue") {
+ val store = makeBlockManager(8000, "executor1")
+ val blockId = RDDBlockId(rddId = 1, splitIndex = 1)
+ var computed: Boolean = false
+ val data = Seq(1, 2, 3)
+ val makeIterator = () => {
+ computed = true
+ data.iterator
+ }
+
+ // Cache doesn't exist and is not visible.
+ assert(store.getStatus(blockId).isEmpty && !store.isRDDBlockVisible(blockId))
+ val res1 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Put cache successfully and reported block task info.
+ assert(res1.isLeft && computed)
+ verify(master, times(1)).updateRDDBlockTaskInfo(blockId, 1)
+
+ // Cache exists but not visible.
+ computed = false
+ assert(store.getStatus(blockId).nonEmpty && !store.isRDDBlockVisible(blockId))
+ val res2 = store.getOrElseUpdateRDDBlock(
+ 1, blockId, StorageLevel.MEMORY_ONLY, classTag[Int], makeIterator)
+ // Load cache successfully and reported block task info.
+ assert(res2.isLeft && computed)
+ verify(master, times(2)).updateRDDBlockTaskInfo(blockId, 1)
Review Comment:
HI @mridulm actually this is to make sure that task2blockId information will be reported only when the block is invisble as this is part of the work flow design. If this doesn't make sense, I can remove `verify` statement in a new iteration. 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118028046
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -525,6 +562,7 @@ private[storage] class BlockInfoManager extends Logging {
blockInfoWrappers.clear()
readLocksByTask.clear()
writeLocksByTask.clear()
+ invisibleRDDBlocks.clear()
Review Comment:
Thanks, done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1445280811
Hi @mridulm comments addressed. Please take a look when you have time. 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
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101657140
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
To clarify, (my previous comment was poorly worded).
A block would be invisible for three reasons primarily (and minor variants of these) :
a) Some task T1 is still actively working on the partition which computed it.
b) Some task T1 on E1 generated the partition, which got pushed to E2 due to replication - and is not yet visible, since T1 is still running.
c) Some task T1 on E1 generated B1, but failed - and so block exists on E1 but is invisible.
In these cases, a block which is invisible would become visible once the task T1 successfully complete - or (in case of failure), is dropped (and so removed).
So the transition is only from invisible -> visible or invisible -> dropped, there is no state -> invisible (unless you count new_block -> invisible :-) )
When a block transitions to becoming visible (due to an explicit message from driver to the executor which generated it, and to the replicas), it remains in that state until it is dropped explicitly (which is again another message from driver).
So the proposal was:
a) Maintain cached state of block at executor - once it is visible, it never changes from that.
b) If invisible, it could be a potential race (T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1) - so we have to query driver, but we can cache the result is visible.
b.1) We can minimize this impact by dropping all B1's when the task T1 fails at E1 (driver would send the update anyway).
So essentially, we can cache only if the block is visible - if it is not visible, we always have to query driver (with (b.1) to minimize needing to query failures)
Thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101657140
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
To clarify, (my previous comment was poorly worded).
A block would be invisible for three reasons primarily (and minor variants of these) :
a) Some task T1 is still actively working on the partition which computed it.
b) Some task T1 on E1 generated the partition, which got pushed to E2 due to replication - and is not yet visible, since T1 is still running.
c) Some task T1 on E1 generated B1, but failed - and so block exists on E1 but is invisible.
In these cases, a block which is invisible would become visible once the task T1 successfully complete - or (in case of failure), is dropped (and so removed).
So the transition is only from invisible -> visible or invisible -> dropped, there is no state -> invisible.
When a block transitions to becoming visible (due to an explicit message from driver to the executor which generated it, and to the replicas), it remains in that state until it is dropped explicitly (which is again another message from driver).
So the proposal was:
a) Maintain cached state of block at executor - once it is visible, it never changes from that.
b) If invisible, it could be a potential race (T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1) - so we have to query driver, but we can cache the result is visible.
b.1) We can minimize this impact by dropping all B1's when the task T1 fails at E1 (driver would send the update anyway).
So essentially, we can cache only if the block is visible - if it is not visible, we always have to query driver (with (b.1) to minimize needing to query failures)
Thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1103941913
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1
I'm not trying to talk about the cache stuff. But just try to highlight this. So this should somehow be a behavior change, right? If T1 generates B1 in early time and T1 turns out to be a long-running task, it can be terrible for tasks like T2 which reads B1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108047546
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,51 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
Review Comment:
Can we add the overall flow of the messages here ? (I want to make sure this is captured somewhere - either here, or as part of the message definitions).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111132438
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully.")
Review Comment:
Additionally add that this is relevant in context of consistent accumulator status
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111226580
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
Updated, please take a look. 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111226310
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,51 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
Review Comment:
Added some comments here, please take a look. 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
[GitHub] [spark] mridulm commented on pull request #39459: [WIP][SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1382693039
Please ping me when it is ready for review, and not WIP. 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1096526943
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
> Right..in this case, the rdd block locations for different data can be attached to the same rdd block id. So the reader could get the different data for the same rdd block, which makes the rdd block data also indeterminate.
@Ngone51 do you think shall we solve this issue in this PR? This looks like a more general issue about the indeterminate computation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101657140
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
To clarify, (my previous comment was poorly worded).
A block would be invisible for three reasons primarily (and minor variants of these) :
a) Some task T1 is still actively working on the partition which computed it.
b) Some task T1 on E1 generated the partition, which got pushed to E2 due to replication - and is not yet visible, since T1 is still running.
c) Some task T1 on E1 generated B1, but failed - and so block exists on E1 but is invisible.
In these cases, a block which is invisible would become visible once the task T1 successfully complete - or (in case of failure), is dropped (and so removed).
So the transition is only from invisible -> visible or invisible -> dropped, there is no state -> invisible (unless you count new_block -> invisible :-) )
When a block transitions to becoming visible (due to an explicit message from driver to the executor which generated it, and to the replicas), it remains in that state until it is dropped explicitly (which is again another message from driver).
So the proposal was:
a) Maintain cached state of block at executor - once it is visible, it never changes from that.
b) If invisible, it could be a potential race (T1 completed, driver yet to update B1 to E1, but launched some T2 which wants to read B1) - so we have to query driver, but we can cache the result if visible.
b.1) We can minimize this impact by dropping all B1's when the task T1 fails at E1 (driver would send the update anyway).
So essentially, we can cache only if the block is visible - if it is not visible, we always have to query driver (with (b.1) to minimize needing to query failures)
Thoughts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112457906
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -502,7 +536,10 @@ private[storage] class BlockInfoManager extends Logging {
throw new IllegalStateException(
s"Task $taskAttemptId called remove() on block $blockId without a write lock")
} else {
- blockInfoWrappers.remove(blockId)
+ invisibleRDDBlocks.synchronized {
+ blockInfoWrappers.remove(blockId)
Review Comment:
`blockInfoWrappers` isn't necessary to be protected by `invisibleRDDBlocks.synchronized`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108047546
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,51 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
Review Comment:
Can we add the overall flow of the messages here ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118028001
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -180,6 +186,27 @@ private[storage] class BlockInfoManager extends Logging {
// ----------------------------------------------------------------------------------------------
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ if (trackingCacheVisibility) {
+ invisibleRDDBlocks.synchronized {
+ blockInfoWrappers.containsKey(blockId) && !invisibleRDDBlocks.contains(blockId)
+ }
+ } else {
+ // Always be visible if the feature flag is disabled.
+ true
+ }
+ }
+
+ private[spark] def tryMarkBlockAsVisible(blockId: RDDBlockId): Unit = {
+ if (trackingCacheVisibility) {
+ invisibleRDDBlocks.synchronized {
+ if (blockInfoWrappers.containsKey(blockId)) {
+ invisibleRDDBlocks.remove(blockId)
Review Comment:
Thanks, you are right. Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1114380075
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
+ val res = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ if (res == null && trackingCacheVisibility) {
+ // Added to invisible blocks if it doesn't exist before.
+ blockId.asRDDId.foreach(invisibleRDDBlocks.add)
Review Comment:
Ok..but seem like the access to `blockInfoWrappers` is not always protected by `invisibleRDDBlocks.synchronized`? e.g., `BlockInfoManager#blockInfo`, `BlockInfoManager#acquireLock`. It doesn't make sense if we only protect it when it's accessed together with `invisibleRDDBlocks`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1122750718
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,74 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
Review Comment:
done, 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
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1124028359
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully. This is relevant in context of consistent accumulator status.")
+ .version("3.4.0")
Review Comment:
No, we should mark this as 3.5.0 instead ... slipped through the review (it was created way before 3.4 was cut).
Thanks for catching this @ulysses-you !
Can you create a follow up to change this to 3.5 @ivoson ? Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1124028359
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully. This is relevant in context of consistent accumulator status.")
+ .version("3.4.0")
Review Comment:
No, we should mark this as 3.5.0 instead ... slipped through the review (it was created way before 3.4 was cut).
Can you create a follow up to change this to 3.5 @ivoson ? Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1106755256
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> I'm not trying to talk about the cache stuff. But just try to highlight this. So this should somehow be a behavior change, right? If T1 generates B1 in early time and T1 turns out to be a long-running task, it can be terrible for tasks like T2 which reads B1.
Do we actually have usecases where some other stage/task is depending on an earlier task generating a block as a prefix to its computation and the task itself has not completed ?
Looks like a fairly brittle assumption, no ? (Unless I misunderstood the usecase here !)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1107261503
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> You are right, the current PR is handling it on a second read ... Since we are already checking for blockInfoManager.isRDDBlockVisible(blockId) first.
This should cover the case of (1) - and we will always query in case block is available, and we have to distinguish (2).
(2.1) would be an optimization we can attempt later on.
Thanks @mridulm . Just want to clarity that are you suggesting that we can also cache the state got from driver when it's already visible?
I think we can update the cache state if the block exists in the executor after getting the results from driver/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
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [WIP][SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1064364664
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,6 +1325,64 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ var getIterator = makeIterator
+
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ if (master.isRDDBlockVisible(blockId)) { // Read from cache only when the block is visible.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
+ // Initially we hold no locks on this block.
+ } else {
+ // Need to compute the block, since the block maybe already exists, force
+ // compute the block here.
+ val iterator = makeIterator()
+ getIterator = () => iterator
+ }
+
+ doPutIterator(blockId, getIterator, level, classTag, keepReadLock = true) match {
+ case None =>
+ // Report taskId -> blockId relationship to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
Review Comment:
How's the rdd replication being updated?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [WIP][SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1064361326
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -76,6 +76,8 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
Review Comment:
When do we clean up this if a task fails?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #39459: [WIP][SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1375792288
Can one of the admins verify this patch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1071164909
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
Review Comment:
Maybe, we should just pass `isCacheVisible` to `getOrElseUpdate` and only load the block cache when it's visible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1071815096
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
If a block is not visible, and it was not computed in this task, we should not be using that block's cached data at all - even if cached here or elsewhere.
Particularly for unordered or indeterminate computation, this can impact the accumulators generated - and it wont match the result returned.
So we cant simply call `makeIterator` here - as @Ngone51 suggested, we should pass the flag to `getOrElseUpdate` and decide there based on visibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by GitBox <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1383190851
cc @mridulm @Ngone51 The PR is ready for review. Please take a look when you have time. 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090477930
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
Review Comment:
Yes, that's true. Will make the 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090478629
##########
core/src/test/scala/org/apache/spark/AccumulatorSuite.scala:
##########
@@ -89,6 +89,38 @@ class AccumulatorSuite extends SparkFunSuite with Matchers with LocalSparkContex
assert(AccumulatorContext.get(100000).isEmpty)
}
+
+ test("SPARK-41497: accumulators should be reported in the case of task retry with rdd cache") {
+ // Set up a cluster with 2 executors
+ val conf = new SparkConf()
+ .setMaster("local-cluster[2, 1, 1024]").setAppName("TaskSchedulerImplSuite")
Review Comment:
Thanks, will update the app name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1090491731
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
That would be better. 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
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111400497
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +78,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
Review Comment:
visible? but the variable name is invisible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111414367
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -150,6 +150,12 @@ private[storage] class BlockInfoManager extends Logging {
*/
private[this] val blockInfoWrappers = new ConcurrentHashMap[BlockId, BlockInfoWrapper]
+ /**
+ * Record visible rdd blocks stored in the block manager, entries will be removed
+ * by [[removeBlock()]]
+ */
+ private[spark] val visibleRDDBlocks = ConcurrentHashMap.newKeySet[RDDBlockId]
Review Comment:
It's weird to me that we use the opposite visibility tracking logic between the driver and executor. Couldn't we use the same "invisible" logic at the executor 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112123512
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +78,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
Review Comment:
Thanks, fixed the 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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111226349
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
+ makeIterator()
Review Comment:
Thanks for pointing this out. Updated and added a UT for this.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1787,6 +1792,12 @@ private[spark] class DAGScheduler(
case _: ExceptionFailure | _: TaskKilled => updateAccumulators(event)
case _ =>
}
+ if (trackingCacheVisibility) {
+ // Update rdd blocks' visibility status.
+ blockManagerMaster.updateRDDBlockVisibility(
Review Comment:
Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111226606
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,14 +1325,47 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ var computed: Boolean = false
+ val getIterator = () => {
+ computed = true
+ makeIterator()
+ }
+
+ val res = getOrElseUpdate(blockId, level, classTag, getIterator)
+ if (res.isLeft && !isCacheVisible) {
+ if (!computed) {
+ // Loaded from cache, re-compute to update accumulators.
+ makeIterator()
+ }
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111297209
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
This is the same issue as [here](https://github.com/apache/spark/pull/39459#discussion_r1071815096) @Ngone51 ... I thought we were doing it as a follow up ?
Note that replacing cached block would also require replication (if relevant), master update, etc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1111402068
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -210,6 +219,65 @@ class BlockManagerMasterEndpoint(
case StopBlockManagerMaster =>
context.reply(true)
stop()
+
+ case UpdateRDDBlockTaskInfo(blockId, taskId) =>
+ // This is to report the information that a rdd block(with `blockId`) is computed
+ // and cached by task(with `taskId`). The happens right after the task finished
Review Comment:
"The" -> "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
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1113188852
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -502,7 +536,10 @@ private[storage] class BlockInfoManager extends Logging {
throw new IllegalStateException(
s"Task $taskAttemptId called remove() on block $blockId without a write lock")
} else {
- blockInfoWrappers.remove(blockId)
+ invisibleRDDBlocks.synchronized {
+ blockInfoWrappers.remove(blockId)
Review Comment:
Just want to make sure that state change for both `blockInfoWrappers` and `invisibleRDDBlocks` are processed in the same syncrhonized block to avoid potential concurrent issues since we depends on the state of the 2 variables to decide whether a block is visible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1112456693
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
+ val res = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ if (res == null && trackingCacheVisibility) {
+ // Added to invisible blocks if it doesn't exist before.
+ blockId.asRDDId.foreach(invisibleRDDBlocks.add)
Review Comment:
I'd suggest updating `invisibleRDDBlocks` within `lockForWriting()` when the write locking is acquired.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1113157159
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -399,7 +426,14 @@ private[storage] class BlockInfoManager extends Logging {
try {
val wrapper = new BlockInfoWrapper(newBlockInfo, lock)
while (true) {
- val previous = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ val previous = invisibleRDDBlocks.synchronized {
+ val res = blockInfoWrappers.putIfAbsent(blockId, wrapper)
+ if (res == null && trackingCacheVisibility) {
+ // Added to invisible blocks if it doesn't exist before.
+ blockId.asRDDId.foreach(invisibleRDDBlocks.add)
Review Comment:
Currently the way we check whether a block is visible is to check the information from both `blockInfoWrappers` and `invisibleRDDBlocks`, so we will need the synchronized block to modify the state to avoid concurrent issues (`invisibleRDDBlocks` should be updated once new item put into `blockInfoWrappers` within the synchronized block). It will be easier to have the syncrhonized block here.
```
private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
if (trackingCacheVisibility) {
invisibleRDDBlocks.synchronized {
blockInfoWrappers.containsKey(blockId) && !invisibleRDDBlocks.contains(blockId)
}
} else {
// Always be visible if the feature flag is disabled.
true
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1101665418
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Yes, that is the idea - block gets added to invisible list initially (since task has not completed), and then gets promoted to becoming visible (when task completes).
If existing blocks are lost - why would you need that information as they are gone ? In other words, how is it different from today's situation (without visibility) - if a block is lost, it is no longer in system.
I think I am missing something, if you can give an example of where this impacts, that would be great !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1098177596
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Not sure I follow - we cannot leverage the block until it is explicitly marked visible (unless within same task).
If the executor is lost before that happens, then we can never make it visible - and so fine to clean up from invisible blocks, right ? (Within task's executor, if for task, block can be used if visible or if tid == current task for invisible)
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -77,6 +77,11 @@ class BlockManagerMasterEndpoint(
// Mapping from block id to the set of block managers that have the block.
private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]
+ // Mapping from task id to the set of rdd blocks which are generated from the task.
+ private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
+ // Record the visible RDD blocks which have been generated at least from one successful task.
+ private val visibleRDDBlocks = new mutable.HashSet[RDDBlockId]
Review Comment:
Not sure I follow - we cannot leverage the block until it is explicitly marked visible (unless within same task).
If the executor is lost before that happens, then we can never make it visible - and so fine to clean up from invisible blocks, right ? (Within task's executor, if check is for task, block can be used if visible or if tid == current task for invisible)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1422195413
> @ivoson could you add a feature flag for the whole changes?
Sure, will add the feature flag.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1104018483
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1325,31 +1328,71 @@ private[spark] class BlockManager(
blockInfoManager.releaseAllLocksForTask(taskAttemptId)
}
+ /**
+ * Retrieve the given rdd block if it exists and is visible, otherwise call the provided
+ * `makeIterator` method to compute the block, persist it, and return its values.
+ *
+ * @return either a BlockResult if the block was successfully cached, or an iterator if the block
+ * could not be cached.
+ */
+ def getOrElseUpdateRDDBlock[T](
+ taskId: Long,
+ blockId: RDDBlockId,
+ level: StorageLevel,
+ classTag: ClassTag[T],
+ makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
+ val isCacheVisible = isRDDBlockVisible(blockId)
+ val res = getOrElseUpdate(blockId, level, classTag, makeIterator, isCacheVisible)
+ if (res.isLeft && !isCacheVisible) {
+ // Block exists but not visible, report taskId -> blockId info to master.
+ master.updateRDDBlockTaskInfo(blockId, taskId)
+ }
+
+ res
+ }
+
/**
* Retrieve the given block if it exists, otherwise call the provided `makeIterator` method
* to compute the block, persist it, and return its values.
*
* @return either a BlockResult if the block was successfully cached, or an iterator if the block
* could not be cached.
*/
- def getOrElseUpdate[T](
+ private def getOrElseUpdate[T](
blockId: BlockId,
level: StorageLevel,
classTag: ClassTag[T],
- makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {
- // Attempt to read the block from local or remote storage. If it's present, then we don't need
- // to go through the local-get-or-put path.
- get[T](blockId)(classTag) match {
- case Some(block) =>
- return Left(block)
- case _ =>
- // Need to compute the block.
+ makeIterator: () => Iterator[T],
+ isCacheVisible: Boolean = true): Either[BlockResult, Iterator[T]] = {
+ // Track whether the data is computed or not, force to do the computation later if need to.
+ // The reason we push the force computing later is that once the executor is decommissioned we
+ // will have a better chance to replicate the cache block because of the `checkShouldStore`
+ // validation when putting a new block.
+ var computed: Boolean = false
+ val iterator = () => {
+ computed = true
+ makeIterator()
+ }
+ if (isCacheVisible) {
+ // Attempt to read the block from local or remote storage. If it's present, then we don't need
+ // to go through the local-get-or-put path.
+ get[T](blockId)(classTag) match {
+ case Some(block) =>
+ return Left(block)
+ case _ =>
+ // Need to compute the block.
+ }
}
+
// Initially we hold no locks on this block.
- doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {
+ doPutIterator(blockId, iterator, level, classTag, keepReadLock = true) match {
case None =>
// doPut() didn't hand work back to us, so the block already existed or was successfully
// stored. Therefore, we now hold a read lock on the block.
+ if (!isCacheVisible && !computed) {
+ // Force compute to report accumulator updates.
Review Comment:
Seems like the result is only computed but not cached, no?
Could we force put the result iterator even if the block exists in this case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1106753599
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
> With above mechanism, do you think we still need another cache to store the visiblity information in executor or do we also need to cache the state in executors not having the cached block data stored?
You are right, the current PR is handling it on a second read ... Since we are already checking for `blockInfoManager.isRDDBlockVisible(blockId)` first.
This should cover the case of (1) - and we will always query in case block is available, and we have to distinguish (2).
(2.1) would be an optimization we can attempt later on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1107228885
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1787,6 +1792,12 @@ private[spark] class DAGScheduler(
case _: ExceptionFailure | _: TaskKilled => updateAccumulators(event)
case _ =>
}
+ if (trackingCacheVisibility) {
+ // Update rdd blocks' visibility status.
+ blockManagerMaster.updateRDDBlockVisibility(
Review Comment:
Yes, async call should be fine. Will make the 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
[GitHub] [spark] mridulm commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1108043615
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -1424,6 +1457,16 @@ private[spark] class BlockManager(
blockStoreUpdater.save()
}
+ // Check whether a rdd block is visible or not.
+ private[spark] def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
+ // If the rdd block visibility information not available in the block manager,
+ // asking master for the information.
+ if (blockInfoManager.isRDDBlockVisible(blockId)) {
+ return true
+ }
+ master.isRDDBlockVisible(blockId)
Review Comment:
Yes, `blockInfoManager.isRDDBlockVisible(blockId)` should check for the current executor side state of the block to see if it is visible or not.
If visible, immediately return true.
Else, query driver for visibility status - if driver returns true, cache that.
Also, when a block becomes visible at driver, it will update the block managers hosting the block about the visibility status - which is then maintained at executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ivoson commented on a diff in pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #39459:
URL: https://github.com/apache/spark/pull/39459#discussion_r1118027972
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2468,4 +2468,15 @@ package object config {
.version("3.4.0")
.booleanConf
.createWithDefault(false)
+
+ private[spark] val RDD_CACHE_VISIBILITY_TRACKING_ENABLED =
+ ConfigBuilder("spark.rdd.cache.visibilityTracking.enabled")
+ .internal()
+ .doc("Set to be true to enabled RDD cache block's visibility status. Once it's enabled," +
+ " a RDD cache block can be used only when it's marked as visible. And a RDD block will be" +
+ " marked as visible only when one of the tasks generating the cache block finished" +
+ " successfully. This is relevant in context of consistent accumulator status.")
+ .version("3.4.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
Sounds good. Thanks, updated.
##########
core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala:
##########
@@ -139,7 +139,7 @@ private[storage] object BlockInfo {
*
* This class is thread-safe.
*/
-private[storage] class BlockInfoManager extends Logging {
+private[storage] class BlockInfoManager(trackingCacheVisibility: Boolean = true) extends Logging {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #39459: [SPARK-41497][CORE] Fixing accumulator undercount in the case of the retry task with rdd cache
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39459:
URL: https://github.com/apache/spark/pull/39459#issuecomment-1452754371
Merged to master.
Thanks for fixing this @ivoson !
Thanks for the reviews @Ngone51 :-)
--
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