You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by andrewor14 <gi...@git.apache.org> on 2014/06/14 06:12:06 UTC
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
GitHub user andrewor14 opened a pull request:
https://github.com/apache/spark/pull/1083
[SPARK-1201] Do not fully materialize partitions for StorageLevel.MEMORY_*_SER
The deserialized version of a partition may occupy much more space than the serialized version. Therefore, if a partition is to be cached with `StorageLevel.MEMORY_*_SER`, we don't need to fully unroll it into an `ArrayBuffer`, but instead we can unroll it into a potentially much smaller `ByteBuffer`.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/andrewor14/spark unroll-them-partitions
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1083.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1083
----
commit a8f181d6483b509c29900de5f325a01ea0ef824f
Author: Andrew Or <an...@gmail.com>
Date: 2014-06-14T03:49:18Z
Add special handling for StorageLevel.MEMORY_*_SER
We only unroll the serialized form of each partition for this case,
because the deserialized form may be much larger and may not fit in
memory.
This commit also abstracts out part of the logic of getOrCompute to
make it more readable.
commit 2941c89baacacfc7573cde35a694bc18a7f5fd4f
Author: Andrew Or <an...@gmail.com>
Date: 2014-06-14T03:52:31Z
Clean up BlockStore (minor)
commit 44ef28246ad4f8116155b0db4969898cc09e5e5e
Author: Andrew Or <an...@gmail.com>
Date: 2014-06-14T03:53:25Z
Actually return updated blocks in putBytes
Previously we never returned the updated blocks in MemoryStore's
putBytes. This is a simple bug with a simple fix.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46735431
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46638158
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15935/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46078010
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46638153
Merged build finished. All automated tests passed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/1083#discussion_r13779778
--- Diff: core/src/main/scala/org/apache/spark/CacheManager.scala ---
@@ -46,79 +46,27 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
new InterruptibleIterator(context, values.asInstanceOf[Iterator[T]])
case None =>
- // Mark the split as loading (unless someone else marks it first)
- loading.synchronized {
- if (loading.contains(key)) {
- logInfo(s"Another thread is loading $key, waiting for it to finish...")
- while (loading.contains(key)) {
- try {
- loading.wait()
- } catch {
- case e: Exception =>
- logWarning(s"Got an exception while waiting for another thread to load $key", e)
- }
- }
- logInfo(s"Finished waiting for $key")
- /* See whether someone else has successfully loaded it. The main way this would fail
- * is for the RDD-level cache eviction policy if someone else has loaded the same RDD
- * partition but we didn't want to make space for it. However, that case is unlikely
- * because it's unlikely that two threads would work on the same RDD partition. One
- * downside of the current code is that threads wait serially if this does happen. */
- blockManager.get(key) match {
- case Some(values) =>
- return new InterruptibleIterator(context, values.asInstanceOf[Iterator[T]])
- case None =>
- logInfo(s"Whoever was loading $key failed; we'll try it ourselves")
- loading.add(key)
- }
- } else {
- loading.add(key)
- }
+ // Acquire a lock for loading this partition
+ // If another thread already holds the lock, wait for it to finish return its results
+ acquireLockForPartition(key).foreach { values =>
--- End diff --
Maybe better to avoid the foreach closure here because return within a closure is implemented as a try-catch, and it is very error prone in the future when we wrap this whole block with a try-catch that catches general Exception ... It has happened multiple times in Spark already, and those problems are really hard to find/debug. Best to just avoid it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46638156
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15933/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/1083#discussion_r13779767
--- Diff: core/src/main/scala/org/apache/spark/CacheManager.scala ---
@@ -128,4 +76,89 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
}
}
}
+
+ /**
+ * Acquire a loading lock for the partition identified by the given block ID.
+ *
+ * If the lock is free, just acquire it and return None. Otherwise, another thread is already
+ * loading the partition, so we wait for it to finish and return the values loaded by the thread.
+ */
+ private def acquireLockForPartition(id: RDDBlockId): Option[Iterator[Any]] = {
+ loading.synchronized {
+ if (!loading.contains(id)) {
+ // If the partition is free, acquire its lock and begin computing its value
+ loading.add(id)
+ None
+ } else {
+ // Otherwise, wait for another thread to finish and return its result
+ logInfo(s"Another thread is loading $id, waiting for it to finish...")
+ while (loading.contains(id)) {
+ try {
+ loading.wait()
+ } catch {
+ case e: Exception =>
+ logWarning(s"Exception while waiting for another thread to load $id", e)
+ }
+ }
+ logInfo(s"Finished waiting for $id")
+ /* See whether someone else has successfully loaded it. The main way this would fail
+ * is for the RDD-level cache eviction policy if someone else has loaded the same RDD
+ * partition but we didn't want to make space for it. However, that case is unlikely
+ * because it's unlikely that two threads would work on the same RDD partition. One
+ * downside of the current code is that threads wait serially if this does happen. */
+ val values = blockManager.get(id)
+ if (!values.isDefined) {
+ logInfo(s"Whoever was loading $id failed; we'll try it ourselves")
+ loading.add(id)
+ }
+ values
+ }
+ }
+ }
+
+ /**
+ * Cache the values of a partition, keeping track of any updates in the storage statuses
+ * of other blocks along the way.
+ */
+ private def cacheValues[T](
+ key: BlockId,
+ value: Iterator[T],
+ storageLevel: StorageLevel,
+ updatedBlocks: ArrayBuffer[(BlockId, BlockStatus)]): Iterator[T] = {
+
+ if (!storageLevel.useMemory) {
+ /* This RDD is not to be cached in memory, so we can just pass the computed values
+ * as an iterator directly to the BlockManager, rather than first fully unrolling
+ * it in memory. The latter option potentially uses much more memory and risks OOM
+ * exceptions that can be avoided. */
+ assume(storageLevel.useDisk || storageLevel.useOffHeap, s"Empty storage level for $key!")
--- End diff --
Might make sense to remove this assume; in case we add a new storage level in the future, this won't hold any more and because this code is so far away from the storage level code, we will likely forget to update this location.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1083
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46078656
Merged build finished. All automated tests passed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/1083#discussion_r13779775
--- Diff: core/src/main/scala/org/apache/spark/CacheManager.scala ---
@@ -128,4 +76,89 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
}
}
}
+
+ /**
+ * Acquire a loading lock for the partition identified by the given block ID.
+ *
+ * If the lock is free, just acquire it and return None. Otherwise, another thread is already
+ * loading the partition, so we wait for it to finish and return the values loaded by the thread.
+ */
+ private def acquireLockForPartition(id: RDDBlockId): Option[Iterator[Any]] = {
+ loading.synchronized {
+ if (!loading.contains(id)) {
+ // If the partition is free, acquire its lock and begin computing its value
+ loading.add(id)
+ None
+ } else {
+ // Otherwise, wait for another thread to finish and return its result
+ logInfo(s"Another thread is loading $id, waiting for it to finish...")
+ while (loading.contains(id)) {
+ try {
+ loading.wait()
+ } catch {
+ case e: Exception =>
+ logWarning(s"Exception while waiting for another thread to load $id", e)
+ }
+ }
+ logInfo(s"Finished waiting for $id")
+ /* See whether someone else has successfully loaded it. The main way this would fail
+ * is for the RDD-level cache eviction policy if someone else has loaded the same RDD
+ * partition but we didn't want to make space for it. However, that case is unlikely
+ * because it's unlikely that two threads would work on the same RDD partition. One
--- End diff --
This paragraph doesn't make a lot of sense to me. In general it is just unlikely for two threads to work on the same rdd partition. However, if we ever pass the first if (where it returns None), it already means we are in the "unlikely" case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46222973
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46235253
It's worth noting that the special handling of the memory serialized storage level actually introduce a regression. In particular, we first serialize the values, put it in block manager, and then deserialize the bytes to return the original values to the user. The tradeoff is that this adds an extra step to deserialize the bytes in the end, which could be slow for large partitions.
This special handling will most likely be superseded by a more general solution for [SPARK-1777](https://issues.apache.org/jira/browse/SPARK-1777), which avoids unrolling an entire partition if there is not enough space for it, regardless of the storage level. For now, I will put this PR on hold.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46229031
Merged build finished. All automated tests passed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46738155
Merged build finished. All automated tests passed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/1083#discussion_r13779776
--- Diff: core/src/main/scala/org/apache/spark/CacheManager.scala ---
@@ -128,4 +76,89 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
}
}
}
+
+ /**
+ * Acquire a loading lock for the partition identified by the given block ID.
+ *
+ * If the lock is free, just acquire it and return None. Otherwise, another thread is already
+ * loading the partition, so we wait for it to finish and return the values loaded by the thread.
+ */
+ private def acquireLockForPartition(id: RDDBlockId): Option[Iterator[Any]] = {
+ loading.synchronized {
+ if (!loading.contains(id)) {
+ // If the partition is free, acquire its lock and begin computing its value
+ loading.add(id)
+ None
+ } else {
+ // Otherwise, wait for another thread to finish and return its result
+ logInfo(s"Another thread is loading $id, waiting for it to finish...")
+ while (loading.contains(id)) {
+ try {
+ loading.wait()
+ } catch {
+ case e: Exception =>
+ logWarning(s"Exception while waiting for another thread to load $id", e)
+ }
+ }
+ logInfo(s"Finished waiting for $id")
+ /* See whether someone else has successfully loaded it. The main way this would fail
+ * is for the RDD-level cache eviction policy if someone else has loaded the same RDD
+ * partition but we didn't want to make space for it. However, that case is unlikely
+ * because it's unlikely that two threads would work on the same RDD partition. One
+ * downside of the current code is that threads wait serially if this does happen. */
+ val values = blockManager.get(id)
+ if (!values.isDefined) {
+ logInfo(s"Whoever was loading $id failed; we'll try it ourselves")
+ loading.add(id)
+ }
+ values
+ }
+ }
+ }
+
+ /**
+ * Cache the values of a partition, keeping track of any updates in the storage statuses
+ * of other blocks along the way.
+ */
+ private def cacheValues[T](
--- End diff --
instead of cacheValues, how about storeInBlockmanager?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46235712
(Whether or not we keep the special handling of the memory serialized storage level, I would still like to get the code style and formatting changes in eventually.)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46222984
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager, BlockStore, and...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46635528
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46636065
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46735447
LGTM.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1083#discussion_r13781786
--- Diff: core/src/main/scala/org/apache/spark/CacheManager.scala ---
@@ -46,79 +46,27 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
new InterruptibleIterator(context, values.asInstanceOf[Iterator[T]])
case None =>
- // Mark the split as loading (unless someone else marks it first)
- loading.synchronized {
- if (loading.contains(key)) {
- logInfo(s"Another thread is loading $key, waiting for it to finish...")
- while (loading.contains(key)) {
- try {
- loading.wait()
- } catch {
- case e: Exception =>
- logWarning(s"Got an exception while waiting for another thread to load $key", e)
- }
- }
- logInfo(s"Finished waiting for $key")
- /* See whether someone else has successfully loaded it. The main way this would fail
- * is for the RDD-level cache eviction policy if someone else has loaded the same RDD
- * partition but we didn't want to make space for it. However, that case is unlikely
- * because it's unlikely that two threads would work on the same RDD partition. One
- * downside of the current code is that threads wait serially if this does happen. */
- blockManager.get(key) match {
- case Some(values) =>
- return new InterruptibleIterator(context, values.asInstanceOf[Iterator[T]])
- case None =>
- logInfo(s"Whoever was loading $key failed; we'll try it ourselves")
- loading.add(key)
- }
- } else {
- loading.add(key)
- }
+ // Acquire a lock for loading this partition
+ // If another thread already holds the lock, wait for it to finish return its results
+ acquireLockForPartition(key).foreach { values =>
--- End diff --
Ok
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46080474
BTW can you construct a unit test for this in CacheManagerSuite?
Would be good also to add a unit test to test the lock (which existed earlier but had no test for it). Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager, BlockStore, and...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46635546
I have removed the special handling of the memory serialized case for the aforementioned reason. However, I would like get the style changes in master as I have a separate WIP PR that builds on top of this one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46078657
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15783/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/1083#discussion_r13781775
--- Diff: core/src/main/scala/org/apache/spark/CacheManager.scala ---
@@ -128,4 +76,89 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
}
}
}
+
+ /**
+ * Acquire a loading lock for the partition identified by the given block ID.
+ *
+ * If the lock is free, just acquire it and return None. Otherwise, another thread is already
+ * loading the partition, so we wait for it to finish and return the values loaded by the thread.
+ */
+ private def acquireLockForPartition(id: RDDBlockId): Option[Iterator[Any]] = {
+ loading.synchronized {
+ if (!loading.contains(id)) {
+ // If the partition is free, acquire its lock and begin computing its value
+ loading.add(id)
+ None
+ } else {
+ // Otherwise, wait for another thread to finish and return its result
+ logInfo(s"Another thread is loading $id, waiting for it to finish...")
+ while (loading.contains(id)) {
+ try {
+ loading.wait()
+ } catch {
+ case e: Exception =>
+ logWarning(s"Exception while waiting for another thread to load $id", e)
+ }
+ }
+ logInfo(s"Finished waiting for $id")
+ /* See whether someone else has successfully loaded it. The main way this would fail
+ * is for the RDD-level cache eviction policy if someone else has loaded the same RDD
+ * partition but we didn't want to make space for it. However, that case is unlikely
+ * because it's unlikely that two threads would work on the same RDD partition. One
--- End diff --
Yeah, this comment already existed and it wasn't exactly clear to me why we needed the lock in the first place. However, assuming the unlikely case that two threads do work on the same partition, it is technically possible for a race condition to occur where the block is kicked out of memory immediately after the other thread released the lock. In this case when we try to `get` the block we will return `None`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46229033
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15822/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46738158
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15972/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager, BlockStore, and...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46635533
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-1201] Do not fully materialize partitio...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46078013
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46738586
Merged in master. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46636074
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46735419
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [Minor] Clean up CacheManager et al.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1083#issuecomment-46638152
Merged build finished. All automated tests passed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---