You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by JoshRosen <gi...@git.apache.org> on 2016/03/05 06:11:54 UTC
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
GitHub user JoshRosen opened a pull request:
https://github.com/apache/spark/pull/11534
[SPARK-13696][WIP] Remove BlockStore class & simplify interfaces of mem. & disk stores
Today, both the MemoryStore and DiskStore implement a common `BlockStore` API, but I feel that this API is inappropriate because it abstracts away important distinctions between the behavior of these two stores.
For instance, the disk store doesn't have a notion of storing deserialized objects, so it's confusing for it to expose object-based APIs like putIterator() and getValues() instead of only exposing binary APIs and pushing the responsibilities of serialization and deserialization to the client. Similarly, the DiskStore put() methods accepted a `StorageLevel` parameter even though the disk store can only store blocks in one form.
As part of a larger BlockManager interface cleanup, this patch remove the BlockStore interface and refine the MemoryStore and DiskStore interfaces to reflect more narrow sets of responsibilities for those components. Some of the benefits of this interface cleanup are reflected in simplifications to several unit tests to eliminate now-unnecessary mocking, significant simplification of the BlockManager's `getLocal()` method, and a narrower API between the MemoryStore and DiskStore.
This patch is a work-in-progress and is currently rebased on top of two other open PRs:
- [SPARK-13695] Don't cache MEMORY_AND_DISK blocks as bytes in memory after spills https://github.com/apache/spark/pull/11533
- [SPARK-13659] Refactor BlockStore put*() APIs to remove returnValues: https://github.com/apache/spark/pull/11502
Next week I'll update this PR description to have a more comprehensive list of improvements. In the meantime, though, the commit messages provide lots of detail into my rationale for the intermediate steps in this refactoring.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/JoshRosen/spark remove-blockstore-interface
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11534.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 #11534
----
commit 6339db4a830580817e5020c2830a278b297e856e
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T21:37:25Z
Refactor synchronous block replication code to call doGetLocal():
The goal behind this change is to rely on doGetLocal() for returning bytes
rather than having to perform the serialization in the replication code.
commit 4d940fb23fae42d79028d13563d8ed923c34599d
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T21:41:00Z
Don't request returnValues when putting values in doPut().
commit 2587546579983f7511816eeea416c63a0d72eaac
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T21:57:03Z
Remove unused PutResult.droppedBlocks field
commit 34ba61cd777190f755be9dff803873abfbba41fa
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T22:11:44Z
Clarify that putBytes() never has returnValues == true
commit ca2d23655e5a27dc95d14609a03900de2afecd81
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T22:16:23Z
Don't call putIterator() with returnValues == true in tests.
commit 68b5e7ceba2b4dc41cc856f7adda72e66ab128ed
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T22:19:25Z
Remove returnValues from the BlockStore interface.
commit ae66ccc8ae55f0171e301ddf03bb15d0b3fcb9a6
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T22:46:43Z
Change putIterator return type to be an Either.
commit 5c9259459c24038d94a763436d389619ab48d0c3
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T22:55:06Z
Remove last accesses of PutResult.data.
commit 2a2d50d4e2eb84c071c4c667b2ae2b4e391ced63
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T22:57:32Z
Remove the PutResult.data field.
commit b38ee81ea5048be5ee213f76f4d4687e27a3250a
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T23:12:14Z
Access DiskStore and MemoryStore directly instead of through BlockStore interface in doPut().
commit 83a2dfb78a6d0058118d284dcf73361fadb39477
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T23:14:29Z
Remove unused DiskStore.getBytes() overload.
commit e24aea1153e182c2ff1bb498246b86d85f052557
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T23:29:59Z
Change putBytes' return type to Unit, since it's never checked.
Eventually it would make sense to give this a Boolean return type, but
I'd like to do this later since there's actually a tricky semantics issue
when considering what the return type is when saving to the MemoryStore.
commit 6381b00a94c7bf4ea0693fc4ae6868ef0f866dc4
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-03T23:31:50Z
Remove PutResult.
commit 9d469362ee679d4cd214b366794e55cbdb57d23f
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:07:59Z
Delete the BlockStore interface.
commit 02a46073d20dd09b1319531660f33d005a22a23c
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:13:14Z
Remove unused DiskStore.getBytes() overloads.
commit 0fc843acf8165bb19b1e1e9466f76e9da26fda30
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:14:36Z
DiskStore.getBytes() never returns None, so it shouldn't return an Option.
commit f0e6d937620569bc27ae46e9eee95297c48cc051
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:31:02Z
Simplify DiskStore.putIterator's return type.
commit de3222467011108d0ece3422f6b3ae2c3580ddfc
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:33:03Z
Remove MemoryStore.putIterator() overload.
commit 40a3cf7daed415bbc259670c51b1afa8cb66fe87
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:39:06Z
DiskStore put() methods don't need to take a StorageLevel.
commit 3756b9684ce536e6c9ba6dd75df679a8f4d8b9c0
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:05:23Z
Factor common error-handling code in DiskStore.put*() into helper function.
commit 7b5aacd1fa12b2f9b9a3544c29e749f2cdfb6e49
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:09:45Z
Remove DiskStore.putIterator().
commit dbe7145b3005e0a9d82339e96263b2b0468f05d2
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:21:57Z
Remove DiskStore's dependency on BlockManager.
commit 1f08a015cdbcfe800a4fe0170f118a1b4a9786a7
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:25:31Z
Minor simplifications in DiskStoreSuite test.
commit a2ea6409bdbd0c723a6db4cffeca81c475507b4d
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:29:18Z
Remove outdated comment in DiskBlockManager.
commit 9009bc59f8f508acbf978d7e03529f20c9dbb306
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:39:25Z
Remove DiskStore's dependency on BlockManager.
commit f5a89ec2a99d5aec15f109616ffc6089748ae142
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T03:28:49Z
Shorten period of holding memoryManager lock.
commit a3f2ec7725f7d4d1225adae71fb2642e68a103ce
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T03:48:35Z
MemoryStore.put() no longer handles dropping to disk.
This is now handled by the caller.
commit 149f20aca4c9dee1469b1578aebe820a59d48a0a
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T04:06:36Z
MemoryStore.putBytes() shouldn't perform deserialization.
commit fde66133742c86e97acab59a145eacea27a34aa5
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T19:31:46Z
MemoryStore should take its own conf, not obtain it from BlockManager.
commit 277f87760f23d5c2076ae8ee4befc4985c98c779
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T19:53:47Z
Move MemoryManager into new o.a.s.storage.memory package
----
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194667138
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52787/
Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194598943
**[Test build #52792 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52792/consoleFull)** for PR 11534 at commit [`84e4d76`](https://github.com/apache/spark/commit/84e4d768dfc52a7a610cb935c838e76eb28fbebf).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55413618
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -34,13 +35,15 @@ private case class MemoryEntry(value: Any, size: Long, deserialized: Boolean)
* Stores blocks in memory, either as Arrays of deserialized Java objects or as
* serialized ByteBuffers.
*/
-private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: MemoryManager)
- extends BlockStore(blockManager) {
+private[spark] class MemoryStore(
+ conf: SparkConf,
+ blockManager: BlockManager,
+ memoryManager: MemoryManager)
+ extends Logging {
// Note: all changes to memory allocations, notably putting blocks, evicting blocks, and
// acquiring or releasing unroll memory, must be synchronized on `memoryManager`!
- private val conf = blockManager.conf
private val entries = new LinkedHashMap[BlockId, MemoryEntry](32, 0.75f, true)
--- End diff --
Actually, nevermind: we _do_ need this synchronization to guarantee the memory safety / consistency of the bookkeeping structures themselves. This synchronization is distinct from the per-block locking that's done higher up the stack.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194061779
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192576749
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52506/
Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195024308
**[Test build #52850 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52850/consoleFull)** for PR 11534 at commit [`2752655`](https://github.com/apache/spark/commit/2752655518c9907dd2c9195f7da8cff28824d71e).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55739221
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
--- End diff --
Yeah, I'm working on this now.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194647472
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55614897
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
--- End diff --
There's not really any reason why we should put it back in memory only for the deserialized case. If I have `MEM_AND_DISK_SER` then this won't even try to the bytes in memory.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55618498
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
--- End diff --
This is in `doPutBytes`. Similar to another comment I had above, there's no reason why we shouldn't attempt to drop it to disk here if it's left. Right now if `memoryStore.putIterator` doesn't fit in memory then we sometimes drop it to disk but sometimes not. That's confusing.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194618341
**[Test build #52777 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52777/consoleFull)** for PR 11534 at commit [`94d2bd6`](https://github.com/apache/spark/commit/94d2bd65f4fc70e75ddff4fe70c8fc4c68b481af).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194644310
Merged build finished. Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193679863
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52638/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193679857
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194593904
**[Test build #52787 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52787/consoleFull)** for PR 11534 at commit [`e3a4c22`](https://github.com/apache/spark/commit/e3a4c223fcfd943266590a479f7e43d78bafabd5).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194061785
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52702/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194025752
**[Test build #52702 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52702/consoleFull)** for PR 11534 at commit [`92c5125`](https://github.com/apache/spark/commit/92c5125f2f4736335971e779fc39e9fa74f8c310).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55280347
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -744,7 +732,7 @@ private[spark] class BlockManager(
level: StorageLevel,
tellMaster: Boolean = true,
effectiveStorageLevel: Option[StorageLevel] = None,
- keepReadLock: Boolean = false): Option[PutResult] = {
+ keepReadLock: Boolean = false): Option[Option[Iterator[Any]]] = {
--- End diff --
Why do levels of options? What does it mean if the outer is set but the inner is not
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55411711
--- Diff: core/src/main/scala/org/apache/spark/storage/DiskStore.scala ---
@@ -17,125 +17,100 @@
package org.apache.spark.storage
-import java.io.{File, FileOutputStream, IOException, RandomAccessFile}
+import java.io.{FileOutputStream, IOException, RandomAccessFile}
import java.nio.ByteBuffer
import java.nio.channels.FileChannel.MapMode
-import org.apache.spark.Logging
+import com.google.common.io.Closeables
+
+import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.util.Utils
/**
* Stores BlockManager blocks on disk.
*/
-private[spark] class DiskStore(blockManager: BlockManager, diskManager: DiskBlockManager)
- extends BlockStore(blockManager) with Logging {
+private[spark] class DiskStore(conf: SparkConf, diskManager: DiskBlockManager) extends Logging {
- val minMemoryMapBytes = blockManager.conf.getSizeAsBytes("spark.storage.memoryMapThreshold", "2m")
+ private val minMemoryMapBytes = conf.getSizeAsBytes("spark.storage.memoryMapThreshold", "2m")
- override def getSize(blockId: BlockId): Long = {
+ def getSize(blockId: BlockId): Long = {
diskManager.getFile(blockId.name).length
}
- override def putBytes(blockId: BlockId, _bytes: ByteBuffer, level: StorageLevel): PutResult = {
- // So that we do not modify the input offsets !
- // duplicate does not copy buffer, so inexpensive
- val bytes = _bytes.duplicate()
+ /**
+ * Invokes the provided callback function to write the specific block.
+ *
+ * @throws IllegalStateException if the block already exists in the disk store.
+ */
+ def put(blockId: BlockId)(writeFunc: FileOutputStream => Unit): Unit = {
+ if (contains(blockId)) {
+ throw new IllegalStateException(s"Block $blockId is already present in the disk store")
+ }
logDebug(s"Attempting to put block $blockId")
val startTime = System.currentTimeMillis
val file = diskManager.getFile(blockId)
- val channel = new FileOutputStream(file).getChannel
- Utils.tryWithSafeFinally {
- while (bytes.remaining > 0) {
- channel.write(bytes)
+ val fileOutputStream = new FileOutputStream(file)
+ var threwException: Boolean = true
+ try {
+ writeFunc(fileOutputStream)
+ threwException = false
+ } finally {
+ try {
+ Closeables.close(fileOutputStream, threwException)
+ } finally {
+ if (threwException) {
+ remove(blockId)
+ }
}
- } {
- channel.close()
}
val finishTime = System.currentTimeMillis
logDebug("Block %s stored as %s file on disk in %d ms".format(
- file.getName, Utils.bytesToString(bytes.limit), finishTime - startTime))
- PutResult(bytes.limit(), Right(bytes.duplicate()))
+ file.getName,
+ Utils.bytesToString(file.length()),
+ finishTime - startTime))
}
- override def putIterator(
- blockId: BlockId,
- values: Iterator[Any],
- level: StorageLevel,
- returnValues: Boolean): PutResult = {
-
- logDebug(s"Attempting to write values for block $blockId")
- val startTime = System.currentTimeMillis
- val file = diskManager.getFile(blockId)
- val outputStream = new FileOutputStream(file)
- try {
+ def putBytes(blockId: BlockId, _bytes: ByteBuffer): Unit = {
+ // So that we do not modify the input offsets !
+ // duplicate does not copy buffer, so inexpensive
+ val bytes = _bytes.duplicate()
+ put(blockId) { fileOutputStream =>
+ val channel = fileOutputStream.getChannel
Utils.tryWithSafeFinally {
- blockManager.dataSerializeStream(blockId, outputStream, values)
+ while (bytes.remaining > 0) {
+ channel.write(bytes)
+ }
} {
- // Close outputStream here because it should be closed before file is deleted.
- outputStream.close()
+ channel.close()
}
- } catch {
- case e: Throwable =>
- if (file.exists()) {
- if (!file.delete()) {
- logWarning(s"Error deleting ${file}")
- }
- }
- throw e
- }
-
- val length = file.length
-
- val timeTaken = System.currentTimeMillis - startTime
- logDebug("Block %s stored as %s file on disk in %d ms".format(
- file.getName, Utils.bytesToString(length), timeTaken))
-
- if (returnValues) {
- // Return a byte buffer for the contents of the file
- val buffer = getBytes(blockId).get
- PutResult(length, Right(buffer))
- } else {
- PutResult(length, null)
}
}
- private def getBytes(file: File, offset: Long, length: Long): Option[ByteBuffer] = {
+ def getBytes(blockId: BlockId): ByteBuffer = {
+ val file = diskManager.getFile(blockId.name)
val channel = new RandomAccessFile(file, "r").getChannel
Utils.tryWithSafeFinally {
// For small files, directly read rather than memory map
- if (length < minMemoryMapBytes) {
- val buf = ByteBuffer.allocate(length.toInt)
- channel.position(offset)
+ if (file.length < minMemoryMapBytes) {
+ val buf = ByteBuffer.allocate(file.length.toInt)
+ channel.position(0)
while (buf.remaining() != 0) {
if (channel.read(buf) == -1) {
throw new IOException("Reached EOF before filling buffer\n" +
- s"offset=$offset\nfile=${file.getAbsolutePath}\nbuf.remaining=${buf.remaining}")
+ s"offset=0\nfile=${file.getAbsolutePath}\nbuf.remaining=${buf.remaining}")
}
}
buf.flip()
- Some(buf)
+ buf
} else {
- Some(channel.map(MapMode.READ_ONLY, offset, length))
+ channel.map(MapMode.READ_ONLY, 0, file.length)
}
} {
channel.close()
}
}
- override def getBytes(blockId: BlockId): Option[ByteBuffer] = {
- val file = diskManager.getFile(blockId.name)
- getBytes(file, 0, file.length)
- }
-
- def getBytes(segment: FileSegment): Option[ByteBuffer] = {
- getBytes(segment.file, segment.offset, segment.length)
- }
-
- override def getValues(blockId: BlockId): Option[Iterator[Any]] = {
- getBytes(blockId).map(buffer => blockManager.dataDeserialize(blockId, buffer))
- }
-
- override def remove(blockId: BlockId): Boolean = {
+ def remove(blockId: BlockId): Boolean = {
--- End diff --
Removals from the stores are idempotent, so the return value lets the caller check whether the removal actually deleted a block (see usage in `BlockManager.removeBlock()`).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193968479
**[Test build #52679 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52679/consoleFull)** for PR 11534 at commit [`988f003`](https://github.com/apache/spark/commit/988f00393676eabfc11e665f20f9ce26388e4c11).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195091525
Merged into master.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194618554
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52777/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55286562
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -34,13 +35,15 @@ private case class MemoryEntry(value: Any, size: Long, deserialized: Boolean)
* Stores blocks in memory, either as Arrays of deserialized Java objects or as
* serialized ByteBuffers.
*/
-private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: MemoryManager)
- extends BlockStore(blockManager) {
+private[spark] class MemoryStore(
+ conf: SparkConf,
+ blockManager: BlockManager,
+ memoryManager: MemoryManager)
+ extends Logging {
// Note: all changes to memory allocations, notably putting blocks, evicting blocks, and
// acquiring or releasing unroll memory, must be synchronized on `memoryManager`!
- private val conf = blockManager.conf
private val entries = new LinkedHashMap[BlockId, MemoryEntry](32, 0.75f, true)
--- End diff --
Comment on thread safety. Why do you need locks for entries?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55617159
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
--- End diff --
That's correct.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619176
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -318,19 +291,26 @@ private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: Memo
blockId.asRDDId.map(_.rddId)
}
+ private def acquireStorageMemory(blockId: BlockId, size: Long): Boolean = {
--- End diff --
IIUC this is only called in 1 place. Why bother abstracting it? It's just kind of weird to see that it's releasing pending unroll memory outside the context of putting a block. I actually think this fits better where it used to be.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619301
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -318,19 +291,26 @@ private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: Memo
blockId.asRDDId.map(_.rddId)
}
+ private def acquireStorageMemory(blockId: BlockId, size: Long): Boolean = {
--- End diff --
Sure, I can move it back. This method is going away anyways in my next patch, though.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195010289
LGTM, can't wait to merge this!
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55280667
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -801,43 +780,63 @@ private[spark] class BlockManager(
}
var blockWasSuccessfullyStored = false
- var result: PutResult = null
+ var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
putBlockInfo.synchronized {
--- End diff --
What is this synchronized protecting? You have an exclusive write lock now no?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55733263
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -422,108 +462,52 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- val level = info.level
- logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
--- End diff --
Done.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55735070
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -739,46 +696,105 @@ private[spark] class BlockManager(
blockId: BlockId,
bytes: ByteBuffer,
level: StorageLevel,
- tellMaster: Boolean = true,
- effectiveStorageLevel: Option[StorageLevel] = None): Boolean = {
+ tellMaster: Boolean = true): Boolean = {
require(bytes != null, "Bytes is null")
- val result = doPut(blockId, ByteBufferValues(bytes), level, tellMaster, effectiveStorageLevel)
- result == DoPutSucceeded
+ doPutBytes(blockId, bytes, level, tellMaster)
}
/**
- * Put the given block according to the given level in one of the block stores, replicating
+ * Put the given bytes according to the given level in one of the block stores, replicating
* the values if necessary.
*
* If the block already exists, this method will not overwrite it.
*
- * @param effectiveStorageLevel the level according to which the block will actually be handled.
- * This allows the caller to specify an alternate behavior of doPut
- * while preserving the original level specified by the user.
* @param keepReadLock if true, this method will hold the read lock when it returns (even if the
* block already exists). If false, this method will hold no locks when it
* returns.
- * @return [[DoPutSucceeded]] if the block was already present or if the put succeeded, or
- * [[DoPutBytesFailed]] if the put failed and we were storing bytes, or
- * [[DoPutIteratorFailed]] if the put failed and we were storing an iterator.
+ * @return true if the block was already present or if the put succeeded, false otherwise.
*/
- private def doPut(
+ private def doPutBytes(
blockId: BlockId,
- data: BlockValues,
+ bytes: ByteBuffer,
level: StorageLevel,
tellMaster: Boolean = true,
- effectiveStorageLevel: Option[StorageLevel] = None,
- keepReadLock: Boolean = false): DoPutResult = {
+ keepReadLock: Boolean = false): Boolean = {
+ doPut(blockId, level, tellMaster = tellMaster, keepReadLock = keepReadLock) { putBlockInfo =>
+ val startTimeMs = System.currentTimeMillis
+ // Since we're storing bytes, initiate the replication before storing them locally.
+ // This is faster as data is already serialized and ready to send.
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
+ }
+
+ bytes.rewind()
+ val size = bytes.limit()
+
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
+ } else {
+ memoryStore.putBytes(blockId, size, () => bytes)
+ }
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
+ }
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
+
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ val blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
+ }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ if (level.replication > 1) {
+ // Wait for asynchronous replication to finish
+ Await.ready(replicationFuture, Duration.Inf)
+ }
+ if (blockWasSuccessfullyStored) {
+ None
+ } else {
+ Some(bytes)
--- End diff --
I don't think we ever use this anywhere. It's simpler to just return `blockWasSuccessfullyStored` here so this method can just return the result of `doPut` directly instead of `doPut(...) { ... }.isEmpty`
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55620372
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
--- End diff --
Done.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193982549
Merged build finished. Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11534
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194642411
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55617427
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
}
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
+ }
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
+ * if the put failed.
+ */
+ private def doPutIterator(
+ blockId: BlockId,
+ iterator: () => Iterator[Any],
+ level: StorageLevel,
+ tellMaster: Boolean = true,
+ keepReadLock: Boolean = false): Option[Iterator[Any]] = {
+
+ require(blockId != null, "BlockId is null")
+ require(level != null && level.isValid, "StorageLevel is null or invalid")
+
+ /* Remember the block's storage level so that we can correctly drop it to disk if it needs
+ * to be dropped right after it got put into memory. Note, however, that other threads will
+ * not be able to get() this block until we call markReady on its BlockInfo. */
+ val putBlockInfo = {
+ val newInfo = new BlockInfo(level, tellMaster)
+ if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo)) {
+ newInfo
+ } else {
+ logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ if (!keepReadLock) {
+ // lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
+ releaseLock(blockId)
+ }
--- End diff --
This is the only hesitation about this patch that I have. There's now a non-trivial amount of code that's duplicated in `doPutIterator` and `doPutBytes`. Is there any way to abstract it? Otherwise I'm not sure if we have a net reduction in complexity here. :/
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194644148
**[Test build #52791 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52791/consoleFull)** for PR 11534 at commit [`0147471`](https://github.com/apache/spark/commit/01474717ac30d472e24ebdb35e0ea5f8b9570e3c).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55411807
--- Diff: core/src/main/scala/org/apache/spark/storage/DiskStore.scala ---
@@ -17,125 +17,100 @@
package org.apache.spark.storage
-import java.io.{File, FileOutputStream, IOException, RandomAccessFile}
+import java.io.{FileOutputStream, IOException, RandomAccessFile}
import java.nio.ByteBuffer
import java.nio.channels.FileChannel.MapMode
-import org.apache.spark.Logging
+import com.google.common.io.Closeables
+
+import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.util.Utils
/**
* Stores BlockManager blocks on disk.
--- End diff --
Yes, that's right. Thread-safety is ensured by locking at higher levels of the system.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193969226
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194596386
**[Test build #52791 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52791/consoleFull)** for PR 11534 at commit [`0147471`](https://github.com/apache/spark/commit/01474717ac30d472e24ebdb35e0ea5f8b9570e3c).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55616812
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -701,16 +666,15 @@ private[spark] class BlockManager(
blockId: BlockId,
values: Iterator[Any],
level: StorageLevel,
- tellMaster: Boolean = true,
- effectiveStorageLevel: Option[StorageLevel] = None): Boolean = {
+ tellMaster: Boolean = true): Boolean = {
require(values != null, "Values is null")
- val result = doPut(
+ val maybeIterator = doPutIterator(
blockId,
- IteratorValues(() => values),
+ () => values,
level,
- tellMaster,
- effectiveStorageLevel)
- result == DoPutSucceeded
+ tellMaster)
--- End diff --
nit: just put these in 1 line. It fits.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194600143
@JoshRosen This looks great. The only concern I have is the duplicate code in `doPutIterator` and `doPutBytes`. In the future if we find a bug in one of them and forget to fix it in the other place things will get really hard to maintain. I do agree on the general idea of separating the two instead of having one giant monolithic `doPut` method, but I really think we need to fix the duplicate code here and I would prefer to do it in this patch.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194642023
**[Test build #52794 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52794/consoleFull)** for PR 11534 at commit [`e2ee2e2`](https://github.com/apache/spark/commit/e2ee2e2af503f759a1b152d1dc9e9c7be3616d94).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193490117
Overall changes look good. Need to take another pass or two to understand this code better.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55620146
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
/**
- * Get a local block from the block manager.
- * Assumes that the caller holds a read lock on the block.
+ * Get block from the local block manager as serialized bytes.
+ *
+ * Must be called while holding a read lock on the block.
+ * Releases the read lock upon exception; keeps the read lock upon successful return.
*/
- private def doGetLocal(
- blockId: BlockId,
- info: BlockInfo,
- asBlockResult: Boolean): Option[Any] = {
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
val level = info.level
logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
- if (level.useMemory) {
- logDebug(s"Getting block $blockId from memory")
- val result = if (asBlockResult) {
- memoryStore.getValues(blockId).map { iter =>
- val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
- new BlockResult(ci, DataReadMethod.Memory, info.size)
- }
+ // In order, try to read the serialized bytes from memory, then from disk, then fall back to
+ // serializing in-memory objects, and, finally, throw an exception if the block does not exist.
+ if (level.deserialized) {
+ // Try to avoid expensive serialization by reading a pre-serialized copy from disk:
+ if (level.useDisk && diskStore.contains(blockId)) {
+ diskStore.getBytes(blockId)
--- End diff --
Comment added.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194646709
**[Test build #52797 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52797/consoleFull)** for PR 11534 at commit [`08bc93b`](https://github.com/apache/spark/commit/08bc93b4f0734c7739534a47388238beab39efe8).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195006923
**[Test build #52845 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52845/consoleFull)** for PR 11534 at commit [`2e8b531`](https://github.com/apache/spark/commit/2e8b531ff6393b6d022e36b53f11befd3576f0b5).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195074808
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52850/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194030709
**[Test build #52704 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52704/consoleFull)** for PR 11534 at commit [`7a08a17`](https://github.com/apache/spark/commit/7a08a179f8951abbdcb7e70f6bfb53821fbc7352).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194074040
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52705/
Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619205
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -114,89 +106,70 @@ private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: Memo
if (putSuccess) {
assert(bytes.limit == size)
}
- }
-
- override def putIterator(
- blockId: BlockId,
- values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
- require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- putIterator(blockId, values, level, allowPersistToDisk = true)
+ putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * There may not be enough space to fully unroll the iterator in memory, in which case we
- * optionally drop the values to disk if
- * (1) the block's storage level specifies useDisk, and
- * (2) `allowPersistToDisk` is true.
--- End diff --
great! I hated this flag.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193982552
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52683/
Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193969232
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52679/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194558289
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55476210
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +752,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
-
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
+ }
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
}
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
+ * if the put failed.
+ */
+ private def doPutIterator(
+ blockId: BlockId,
+ iterator: () => Iterator[Any],
+ level: StorageLevel,
+ tellMaster: Boolean = true,
+ keepReadLock: Boolean = false): Option[Iterator[Any]] = {
+
+ require(blockId != null, "BlockId is null")
+ require(level != null && level.isValid, "StorageLevel is null or invalid")
+
+ /* Remember the block's storage level so that we can correctly drop it to disk if it needs
+ * to be dropped right after it got put into memory. Note, however, that other threads will
+ * not be able to get() this block until we call markReady on its BlockInfo. */
+ val putBlockInfo = {
+ val newInfo = new BlockInfo(level, tellMaster)
+ if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo)) {
+ newInfo
+ } else {
+ logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ if (!keepReadLock) {
+ // lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
+ releaseLock(blockId)
+ }
+ return None
+ }
+ }
+
+ val startTimeMs = System.currentTimeMillis
+
+ // Size of the block in bytes
+ var size = 0L
+
+ var blockWasSuccessfullyStored = false
+ var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ memoryStore.putIterator(blockId, iterator(), level) match {
+ case Right(s) =>
+ size = s
+ case Left(iter) =>
+ // Not enough space to unroll this block; drop to disk if applicable
+ if (level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.put(blockId) { fileOutputStream =>
+ dataSerializeStream(blockId, fileOutputStream, iter)
--- End diff --
Nope, the entire block is now on disk because `iter` contains all of the elements of the original input `iterator()`.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55615052
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
/**
- * Get a local block from the block manager.
- * Assumes that the caller holds a read lock on the block.
+ * Get block from the local block manager as serialized bytes.
+ *
+ * Must be called while holding a read lock on the block.
+ * Releases the read lock upon exception; keeps the read lock upon successful return.
*/
- private def doGetLocal(
- blockId: BlockId,
- info: BlockInfo,
- asBlockResult: Boolean): Option[Any] = {
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
val level = info.level
logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
- if (level.useMemory) {
- logDebug(s"Getting block $blockId from memory")
- val result = if (asBlockResult) {
- memoryStore.getValues(blockId).map { iter =>
- val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
- new BlockResult(ci, DataReadMethod.Memory, info.size)
- }
+ // In order, try to read the serialized bytes from memory, then from disk, then fall back to
+ // serializing in-memory objects, and, finally, throw an exception if the block does not exist.
+ if (level.deserialized) {
+ // Try to avoid expensive serialization by reading a pre-serialized copy from disk:
+ if (level.useDisk && diskStore.contains(blockId)) {
+ diskStore.getBytes(blockId)
--- End diff --
I would add a comment here saying don't bother putting it back in memory here because we'd have to deserialize the data and that might be slow.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55613252
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
+ // Cache the values before returning them
+ memoryStore.putIterator(blockId, valuesFromDisk, level) match {
+ case Left(iter) =>
+ // The memory store put() failed, so it returned the iterator back to us:
+ iter
+ case Right(_) =>
+ // The put() succeeded, so we can read the values back:
+ memoryStore.getValues(blockId).get
+ }
+ } else {
+ valuesFromDisk
+ }
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iterToReturn, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Disk, info.size))
+ } else {
+ releaseLock(blockId)
--- End diff --
can you add a comment in the javadoc on when the lock will be released?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55278560
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -422,108 +462,52 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- val level = info.level
- logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
--- End diff --
Comment that this is called with the read lock taken.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192746032
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193636595
Splitting up `doPut()` into two smaller methods, one concerned with putting bytes and another for putting iterators of objects, would let me simplify its return type and might lead to other simplifications (e.g. by eliminating impossible error-handling cases). I might try this tomorrow.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55618229
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
}
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
+ }
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
+ * if the put failed.
+ */
+ private def doPutIterator(
+ blockId: BlockId,
+ iterator: () => Iterator[Any],
+ level: StorageLevel,
+ tellMaster: Boolean = true,
+ keepReadLock: Boolean = false): Option[Iterator[Any]] = {
+
+ require(blockId != null, "BlockId is null")
+ require(level != null && level.isValid, "StorageLevel is null or invalid")
+
+ /* Remember the block's storage level so that we can correctly drop it to disk if it needs
+ * to be dropped right after it got put into memory. Note, however, that other threads will
+ * not be able to get() this block until we call markReady on its BlockInfo. */
+ val putBlockInfo = {
+ val newInfo = new BlockInfo(level, tellMaster)
+ if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo)) {
+ newInfo
+ } else {
+ logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ if (!keepReadLock) {
+ // lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
+ releaseLock(blockId)
+ }
+ return None
}
+ }
+
+ val startTimeMs = System.currentTimeMillis
+
+ // Size of the block in bytes
+ var size = 0L
+
+ var blockWasSuccessfullyStored = false
+ var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ memoryStore.putIterator(blockId, iterator(), level) match {
+ case Right(s) =>
+ size = s
+ case Left(iter) =>
+ // Not enough space to unroll this block; drop to disk if applicable
+ if (level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.put(blockId) { fileOutputStream =>
+ dataSerializeStream(blockId, fileOutputStream, iter)
+ }
+ size = diskStore.getSize(blockId)
+ } else {
+ iteratorFromFailedMemoryStorePut = Some(iter)
+ }
+ }
+ } else if (level.useDisk) {
+ diskStore.put(blockId) { fileOutputStream =>
+ dataSerializeStream(blockId, fileOutputStream, iterator())
+ }
+ size = diskStore.getSize(blockId)
+ }
+
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
+ }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
+ } else {
+ blockInfoManager.unlock(blockId)
+ }
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
+ }
+ }
+ logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ if (level.replication > 1 && blockWasSuccessfullyStored) {
+ val remoteStartTime = System.currentTimeMillis
+ val bytesToReplicate = doGetLocalBytes(blockId, putBlockInfo)
--- End diff --
I'll just move this code inside of the `try` block.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194641750
**[Test build #52792 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52792/consoleFull)** for PR 11534 at commit [`84e4d76`](https://github.com/apache/spark/commit/84e4d768dfc52a7a610cb935c838e76eb28fbebf).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619390
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
--- End diff --
ah ok, never mind
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195074806
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193982170
**[Test build #52683 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52683/consoleFull)** for PR 11534 at commit [`14857b3`](https://github.com/apache/spark/commit/14857b3e53d1b496b7ff07099c53ab2d775f950a).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192736752
Spotted the problem: we need an extra `releaseLock()` call in `getOrElseUpdate`'s "block already exists" branch; see 9dc510c. I can spin this change off into a smaller patch against master in case we want to clean up the log noise sooner.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619576
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
/**
- * Get a local block from the block manager.
- * Assumes that the caller holds a read lock on the block.
+ * Get block from the local block manager as serialized bytes.
+ *
+ * Must be called while holding a read lock on the block.
+ * Releases the read lock upon exception; keeps the read lock upon successful return.
*/
- private def doGetLocal(
- blockId: BlockId,
- info: BlockInfo,
- asBlockResult: Boolean): Option[Any] = {
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
val level = info.level
logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
- if (level.useMemory) {
- logDebug(s"Getting block $blockId from memory")
- val result = if (asBlockResult) {
- memoryStore.getValues(blockId).map { iter =>
- val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
- new BlockResult(ci, DataReadMethod.Memory, info.size)
- }
+ // In order, try to read the serialized bytes from memory, then from disk, then fall back to
+ // serializing in-memory objects, and, finally, throw an exception if the block does not exist.
+ if (level.deserialized) {
+ // Try to avoid expensive serialization by reading a pre-serialized copy from disk:
+ if (level.useDisk && diskStore.contains(blockId)) {
+ diskStore.getBytes(blockId)
--- End diff --
yeah, that's what I meant
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55616553
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
}
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
+ }
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
+ * if the put failed.
+ */
+ private def doPutIterator(
+ blockId: BlockId,
+ iterator: () => Iterator[Any],
+ level: StorageLevel,
+ tellMaster: Boolean = true,
+ keepReadLock: Boolean = false): Option[Iterator[Any]] = {
+
+ require(blockId != null, "BlockId is null")
+ require(level != null && level.isValid, "StorageLevel is null or invalid")
+
+ /* Remember the block's storage level so that we can correctly drop it to disk if it needs
+ * to be dropped right after it got put into memory. Note, however, that other threads will
+ * not be able to get() this block until we call markReady on its BlockInfo. */
+ val putBlockInfo = {
+ val newInfo = new BlockInfo(level, tellMaster)
+ if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo)) {
+ newInfo
+ } else {
+ logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ if (!keepReadLock) {
+ // lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
+ releaseLock(blockId)
+ }
+ return None
}
+ }
+
+ val startTimeMs = System.currentTimeMillis
+
+ // Size of the block in bytes
+ var size = 0L
+
+ var blockWasSuccessfullyStored = false
+ var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ memoryStore.putIterator(blockId, iterator(), level) match {
+ case Right(s) =>
+ size = s
+ case Left(iter) =>
+ // Not enough space to unroll this block; drop to disk if applicable
+ if (level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.put(blockId) { fileOutputStream =>
+ dataSerializeStream(blockId, fileOutputStream, iter)
+ }
+ size = diskStore.getSize(blockId)
+ } else {
+ iteratorFromFailedMemoryStorePut = Some(iter)
+ }
+ }
+ } else if (level.useDisk) {
+ diskStore.put(blockId) { fileOutputStream =>
+ dataSerializeStream(blockId, fileOutputStream, iterator())
+ }
+ size = diskStore.getSize(blockId)
+ }
+
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
+ }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
+ } else {
+ blockInfoManager.unlock(blockId)
+ }
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
+ }
+ }
+ logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ if (level.replication > 1 && blockWasSuccessfullyStored) {
+ val remoteStartTime = System.currentTimeMillis
+ val bytesToReplicate = doGetLocalBytes(blockId, putBlockInfo)
--- End diff --
if `keepReadLock` is false we don't actually hold the read lock here! That's not what `doGetLocalBytes` assumes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55618676
--- Diff: core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala ---
@@ -26,18 +26,14 @@ import org.apache.spark.util.{ShutdownHookManager, Utils}
/**
* Creates and maintains the logical mapping between logical blocks and physical on-disk
- * locations. By default, one block is mapped to one file with a name given by its BlockId.
- * However, it is also possible to have a block map to only a segment of a file, by calling
- * mapBlockToFileSegment().
+ * locations. One block is mapped to one file with a name given by its BlockId.
*
* Block files are hashed among the directories listed in spark.local.dir (or in
* SPARK_LOCAL_DIRS, if it's set).
*/
-private[spark] class DiskBlockManager(blockManager: BlockManager, conf: SparkConf)
- extends Logging {
+private[spark] class DiskBlockManager(conf: SparkConf, deleteFilesOnStop: Boolean) extends Logging {
--- End diff --
nice!
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193678971
**[Test build #52638 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52638/consoleFull)** for PR 11534 at commit [`04f8d15`](https://github.com/apache/spark/commit/04f8d15b8877a508fba70ba1ee118bf9d199973b).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194600693
@andrewor14, let me take another shot at reducing that duplication and I'll get back to you.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55279346
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -422,108 +462,52 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- val level = info.level
- logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.deserialized) {
--- End diff --
Is this backwards? If the level is deserialized we serialize it? comment.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55412172
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -34,13 +35,15 @@ private case class MemoryEntry(value: Any, size: Long, deserialized: Boolean)
* Stores blocks in memory, either as Arrays of deserialized Java objects or as
* serialized ByteBuffers.
*/
-private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: MemoryManager)
- extends BlockStore(blockManager) {
+private[spark] class MemoryStore(
+ conf: SparkConf,
+ blockManager: BlockManager,
+ memoryManager: MemoryManager)
+ extends Logging {
// Note: all changes to memory allocations, notably putting blocks, evicting blocks, and
// acquiring or releasing unroll memory, must be synchronized on `memoryManager`!
- private val conf = blockManager.conf
private val entries = new LinkedHashMap[BlockId, MemoryEntry](32, 0.75f, true)
--- End diff --
Good catch. In the old code, there were multiple places that called into the MemoryStore, some in the BlockManager and others in the CahceManager, so I think we needed these locks to ensure thread-safety. Now, though, I think that all accesses should be coming from the BlockManager so I think we can remove this locking and rely on the caller holding the appropriate BlockManager locks instead.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55472870
--- Diff: core/src/main/scala/org/apache/spark/storage/DiskStore.scala ---
@@ -17,112 +17,100 @@
package org.apache.spark.storage
-import java.io.{File, FileOutputStream, IOException, RandomAccessFile}
+import java.io.{FileOutputStream, IOException, RandomAccessFile}
import java.nio.ByteBuffer
import java.nio.channels.FileChannel.MapMode
-import org.apache.spark.Logging
+import com.google.common.io.Closeables
+
+import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.util.Utils
/**
* Stores BlockManager blocks on disk.
*/
-private[spark] class DiskStore(blockManager: BlockManager, diskManager: DiskBlockManager)
- extends BlockStore(blockManager) with Logging {
+private[spark] class DiskStore(conf: SparkConf, diskManager: DiskBlockManager) extends Logging {
- val minMemoryMapBytes = blockManager.conf.getSizeAsBytes("spark.storage.memoryMapThreshold", "2m")
+ private val minMemoryMapBytes = conf.getSizeAsBytes("spark.storage.memoryMapThreshold", "2m")
- override def getSize(blockId: BlockId): Long = {
+ def getSize(blockId: BlockId): Long = {
diskManager.getFile(blockId.name).length
}
- override def putBytes(blockId: BlockId, _bytes: ByteBuffer, level: StorageLevel): Unit = {
- // So that we do not modify the input offsets !
- // duplicate does not copy buffer, so inexpensive
- val bytes = _bytes.duplicate()
+ /**
+ * Invokes the provided callback function to write the specific block.
+ *
+ * @throws IllegalStateException if the block already exists in the disk store.
+ */
+ def put(blockId: BlockId)(writeFunc: FileOutputStream => Unit): Unit = {
+ if (contains(blockId)) {
+ throw new IllegalStateException(s"Block $blockId is already present in the disk store")
+ }
logDebug(s"Attempting to put block $blockId")
val startTime = System.currentTimeMillis
val file = diskManager.getFile(blockId)
- val channel = new FileOutputStream(file).getChannel
- Utils.tryWithSafeFinally {
- while (bytes.remaining > 0) {
- channel.write(bytes)
+ val fileOutputStream = new FileOutputStream(file)
+ var threwException: Boolean = true
+ try {
+ writeFunc(fileOutputStream)
+ threwException = false
+ } finally {
+ try {
+ Closeables.close(fileOutputStream, threwException)
+ } finally {
+ if (threwException) {
+ remove(blockId)
+ }
}
- } {
- channel.close()
}
val finishTime = System.currentTimeMillis
logDebug("Block %s stored as %s file on disk in %d ms".format(
- file.getName, Utils.bytesToString(bytes.limit), finishTime - startTime))
+ file.getName,
+ Utils.bytesToString(file.length()),
+ finishTime - startTime))
}
- override def putIterator(
- blockId: BlockId,
- values: Iterator[Any],
- level: StorageLevel): Right[Iterator[Any], Long] = {
- logDebug(s"Attempting to write values for block $blockId")
- val startTime = System.currentTimeMillis
- val file = diskManager.getFile(blockId)
- val outputStream = new FileOutputStream(file)
- try {
+ def putBytes(blockId: BlockId, _bytes: ByteBuffer): Unit = {
--- End diff --
Comment function and guarantees on _bytes on return.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55471546
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -34,13 +35,15 @@ private case class MemoryEntry(value: Any, size: Long, deserialized: Boolean)
* Stores blocks in memory, either as Arrays of deserialized Java objects or as
* serialized ByteBuffers.
*/
-private[spark] class MemoryStore(blockManager: BlockManager, memoryManager: MemoryManager)
- extends BlockStore(blockManager) {
+private[spark] class MemoryStore(
+ conf: SparkConf,
+ blockManager: BlockManager,
+ memoryManager: MemoryManager)
+ extends Logging {
// Note: all changes to memory allocations, notably putting blocks, evicting blocks, and
// acquiring or releasing unroll memory, must be synchronized on `memoryManager`!
- private val conf = blockManager.conf
private val entries = new LinkedHashMap[BlockId, MemoryEntry](32, 0.75f, true)
--- End diff --
Hmm okay. DiskStore also needs to be thread safe right? It just happens it doesn't maintain any state right now that's interesting. I think it's worth commenting so people modifying this know what the expected properties are.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194642010
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194666644
**[Test build #52787 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52787/consoleFull)** for PR 11534 at commit [`e3a4c22`](https://github.com/apache/spark/commit/e3a4c223fcfd943266590a479f7e43d78bafabd5).
* This patch **fails from timeout after a configured wait of \`250m\`**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192576748
Merged build finished. Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194642413
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52794/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194600377
**[Test build #52794 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52794/consoleFull)** for PR 11534 at commit [`e2ee2e2`](https://github.com/apache/spark/commit/e2ee2e2af503f759a1b152d1dc9e9c7be3616d94).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55285972
--- Diff: core/src/main/scala/org/apache/spark/storage/DiskStore.scala ---
@@ -17,125 +17,100 @@
package org.apache.spark.storage
-import java.io.{File, FileOutputStream, IOException, RandomAccessFile}
+import java.io.{FileOutputStream, IOException, RandomAccessFile}
import java.nio.ByteBuffer
import java.nio.channels.FileChannel.MapMode
-import org.apache.spark.Logging
+import com.google.common.io.Closeables
+
+import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.util.Utils
/**
* Stores BlockManager blocks on disk.
--- End diff --
This not thread safe correct?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55278923
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -401,11 +403,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
+ // Cache the values before returning them
+ memoryStore.putIterator(blockId, valuesFromDisk, level) match {
--- End diff --
How would you simplify it? We have to return an iterator on failure because of the one-shot iterator semantics.
Do you think that we should just throw away the partially-unrolled iterator returned from `putIterator()` and perform another disk scan instead?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194031761
**[Test build #52705 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52705/consoleFull)** for PR 11534 at commit [`a16276e`](https://github.com/apache/spark/commit/a16276e1cfb92fba88f2963b0d918aa14f2500a4).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192733973
There seems to be a lot of
```
12:54:15.739 WARN org.apache.spark.storage.BlockManager: Block rdd_38548_0 already exists on this machine; not re-adding it
12:54:15.742 ERROR org.apache.spark.executor.Executor: 1 block locks were not released by TID = 103091:
[rdd_38548_0]
```
entries in the test log during HiveCompatibilitySuite. I'll try to investigate to see if I can spot what's going on. This isn't leading to test failures but likely indicates a problem somewhere.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194033260
I went ahead and split up `doPut()`, which greatly simplified the return types.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55618054
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
}
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
+ }
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
+ * if the put failed.
+ */
+ private def doPutIterator(
+ blockId: BlockId,
+ iterator: () => Iterator[Any],
+ level: StorageLevel,
+ tellMaster: Boolean = true,
+ keepReadLock: Boolean = false): Option[Iterator[Any]] = {
+
+ require(blockId != null, "BlockId is null")
+ require(level != null && level.isValid, "StorageLevel is null or invalid")
+
+ /* Remember the block's storage level so that we can correctly drop it to disk if it needs
+ * to be dropped right after it got put into memory. Note, however, that other threads will
+ * not be able to get() this block until we call markReady on its BlockInfo. */
+ val putBlockInfo = {
+ val newInfo = new BlockInfo(level, tellMaster)
+ if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo)) {
+ newInfo
+ } else {
+ logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ if (!keepReadLock) {
+ // lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
+ releaseLock(blockId)
+ }
--- End diff --
I've tried factoring it out into a common method but that method has a weird name and signature and the size of the comment needed to describe it meant that we didn't end up saving much space. Basically, I'd like to defer that cleanup to later since it seems non-essential and I think this is already more understandable than the old code.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194644312
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52791/
Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194618552
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194061280
**[Test build #52702 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52702/consoleFull)** for PR 11534 at commit [`92c5125`](https://github.com/apache/spark/commit/92c5125f2f4736335971e779fc39e9fa74f8c310).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192746037
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52521/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192758756
**[Test build #52522 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52522/consoleFull)** for PR 11534 at commit [`9dc510c`](https://github.com/apache/spark/commit/9dc510ca373ff02d71bf54b952396b0c1284ef0f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55612253
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
--- End diff --
I considered this but thought that doing so would be more confusing to readers.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619575
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -701,16 +666,15 @@ private[spark] class BlockManager(
blockId: BlockId,
values: Iterator[Any],
level: StorageLevel,
- tellMaster: Boolean = true,
- effectiveStorageLevel: Option[StorageLevel] = None): Boolean = {
+ tellMaster: Boolean = true): Boolean = {
require(values != null, "Values is null")
- val result = doPut(
+ val maybeIterator = doPutIterator(
blockId,
- IteratorValues(() => values),
+ () => values,
level,
- tellMaster,
- effectiveStorageLevel)
- result == DoPutSucceeded
+ tellMaster)
--- End diff --
Done.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194075328
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52704/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192711148
**[Test build #52521 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52521/consoleFull)** for PR 11534 at commit [`18c07e6`](https://github.com/apache/spark/commit/18c07e6da858be167500220732d3bb4565e98df2).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55617710
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -548,16 +521,9 @@ private[spark] class BlockManager(
* This does not acquire a lock on this block in this JVM.
*/
def getRemote(blockId: BlockId): Option[BlockResult] = {
--- End diff --
Done.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55611895
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
--- End diff --
don't know if we care about the debug message, but a nicer way to write this is:
```
blockInfoManager.lockForReading(blockId).map { info => doGetLocalBytes(blockId, info) }
```
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55615659
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
--- End diff --
I think it's pretty clear, but we don't have to do 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55610963
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
+ // Cache the values before returning them
+ memoryStore.putIterator(blockId, valuesFromDisk, level) match {
+ case Left(iter) =>
+ // The memory store put() failed, so it returned the iterator back to us:
+ iter
+ case Right(_) =>
+ // The put() succeeded, so we can read the values back:
+ memoryStore.getValues(blockId).get
+ }
+ } else {
+ valuesFromDisk
+ }
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iterToReturn, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Disk, info.size))
+ } else {
+ releaseLock(blockId)
--- End diff --
Nope; we want the caller to hold a read lock if this method returns successfully.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193636217
**[Test build #52638 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52638/consoleFull)** for PR 11534 at commit [`04f8d15`](https://github.com/apache/spark/commit/04f8d15b8877a508fba70ba1ee118bf9d199973b).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619226
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +742,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
--- End diff --
If `isRight` here returns `false` then in the `if (!putSucceeded && level.useDisk) {` branch four lines down we'll try to put the original bytes to disk.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194092278
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55473363
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +752,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
-
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
+ }
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
}
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
+ * if the put failed.
+ */
+ private def doPutIterator(
+ blockId: BlockId,
+ iterator: () => Iterator[Any],
+ level: StorageLevel,
+ tellMaster: Boolean = true,
+ keepReadLock: Boolean = false): Option[Iterator[Any]] = {
+
+ require(blockId != null, "BlockId is null")
+ require(level != null && level.isValid, "StorageLevel is null or invalid")
+
+ /* Remember the block's storage level so that we can correctly drop it to disk if it needs
+ * to be dropped right after it got put into memory. Note, however, that other threads will
+ * not be able to get() this block until we call markReady on its BlockInfo. */
+ val putBlockInfo = {
+ val newInfo = new BlockInfo(level, tellMaster)
+ if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo)) {
+ newInfo
+ } else {
+ logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ if (!keepReadLock) {
+ // lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
+ releaseLock(blockId)
+ }
+ return None
+ }
+ }
+
+ val startTimeMs = System.currentTimeMillis
+
+ // Size of the block in bytes
+ var size = 0L
+
+ var blockWasSuccessfullyStored = false
+ var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ memoryStore.putIterator(blockId, iterator(), level) match {
+ case Right(s) =>
+ size = s
+ case Left(iter) =>
+ // Not enough space to unroll this block; drop to disk if applicable
+ if (level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.put(blockId) { fileOutputStream =>
+ dataSerializeStream(blockId, fileOutputStream, iter)
--- End diff --
Does this mean this block is now split between the memStore and the diskStore?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55735901
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
--- End diff --
@JoshRosen by the way this comment still needs to be addressed, but I think everything else is covered.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192758829
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52522/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194601341
**[Test build #52797 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52797/consoleFull)** for PR 11534 at commit [`08bc93b`](https://github.com/apache/spark/commit/08bc93b4f0734c7739534a47388238beab39efe8).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194092098
**[Test build #52707 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52707/consoleFull)** for PR 11534 at commit [`66796b5`](https://github.com/apache/spark/commit/66796b5bf89ddedc9644b9f5692441293c0c0aaa).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55617134
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -916,11 +945,9 @@ private[spark] class BlockManager(
}
if (blockWasSuccessfullyStored) {
- DoPutSucceeded
- } else if (iteratorFromFailedMemoryStorePut.isDefined) {
- DoPutIteratorFailed(iteratorFromFailedMemoryStorePut.get)
+ None
} else {
- DoPutBytesFailed
+ Some(iteratorFromFailedMemoryStorePut.get)
--- End diff --
I would just return `iteratorFromFailedMemoryStorePut` here directly. Doing the `Some(opt.get)` is a strange pattern. If you want you can also add an `assert(blockWasSuccessfullyStored == iteratorFromFailedMemoryStorePut.isEmpty)`.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194074039
Merged build finished. Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192576667
**[Test build #52506 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52506/consoleFull)** for PR 11534 at commit [`7d2aba3`](https://github.com/apache/spark/commit/7d2aba3997ba3770c45585daee4e023a172a4f50).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194667136
Merged build finished. Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55279952
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -422,108 +462,52 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- val level = info.level
- logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.deserialized) {
--- End diff --
`level-deserialized` specifies that this block's in-memory values will be stored as Java objects. For a block at this storage level, there are three possibilities:
1. it's stored as java objects in memory
2. it's stored as bytes on disk
3. an error has occurred and the block is missing
The logic inside of this `if` case tries to check case 2 ahead of case 1 so that we avoid unnecessary serialization if a serialized copy already exists on disk.
Were you confused about the fact that we can call `getLocalBytes()` on a block which is cached as java objects?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192738253
**[Test build #52522 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52522/consoleFull)** for PR 11534 at commit [`9dc510c`](https://github.com/apache/spark/commit/9dc510ca373ff02d71bf54b952396b0c1284ef0f).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55741963
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
--- End diff --
I agree. I've updated this to put the bytes into memory.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194570815
**[Test build #52777 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52777/consoleFull)** for PR 11534 at commit [`94d2bd6`](https://github.com/apache/spark/commit/94d2bd65f4fc70e75ddff4fe70c8fc4c68b481af).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55285926
--- Diff: core/src/main/scala/org/apache/spark/storage/DiskStore.scala ---
@@ -17,125 +17,100 @@
package org.apache.spark.storage
-import java.io.{File, FileOutputStream, IOException, RandomAccessFile}
+import java.io.{FileOutputStream, IOException, RandomAccessFile}
import java.nio.ByteBuffer
import java.nio.channels.FileChannel.MapMode
-import org.apache.spark.Logging
+import com.google.common.io.Closeables
+
+import org.apache.spark.{Logging, SparkConf}
import org.apache.spark.util.Utils
/**
* Stores BlockManager blocks on disk.
*/
-private[spark] class DiskStore(blockManager: BlockManager, diskManager: DiskBlockManager)
- extends BlockStore(blockManager) with Logging {
+private[spark] class DiskStore(conf: SparkConf, diskManager: DiskBlockManager) extends Logging {
- val minMemoryMapBytes = blockManager.conf.getSizeAsBytes("spark.storage.memoryMapThreshold", "2m")
+ private val minMemoryMapBytes = conf.getSizeAsBytes("spark.storage.memoryMapThreshold", "2m")
- override def getSize(blockId: BlockId): Long = {
+ def getSize(blockId: BlockId): Long = {
diskManager.getFile(blockId.name).length
}
- override def putBytes(blockId: BlockId, _bytes: ByteBuffer, level: StorageLevel): PutResult = {
- // So that we do not modify the input offsets !
- // duplicate does not copy buffer, so inexpensive
- val bytes = _bytes.duplicate()
+ /**
+ * Invokes the provided callback function to write the specific block.
+ *
+ * @throws IllegalStateException if the block already exists in the disk store.
+ */
+ def put(blockId: BlockId)(writeFunc: FileOutputStream => Unit): Unit = {
+ if (contains(blockId)) {
+ throw new IllegalStateException(s"Block $blockId is already present in the disk store")
+ }
logDebug(s"Attempting to put block $blockId")
val startTime = System.currentTimeMillis
val file = diskManager.getFile(blockId)
- val channel = new FileOutputStream(file).getChannel
- Utils.tryWithSafeFinally {
- while (bytes.remaining > 0) {
- channel.write(bytes)
+ val fileOutputStream = new FileOutputStream(file)
+ var threwException: Boolean = true
+ try {
+ writeFunc(fileOutputStream)
+ threwException = false
+ } finally {
+ try {
+ Closeables.close(fileOutputStream, threwException)
+ } finally {
+ if (threwException) {
+ remove(blockId)
+ }
}
- } {
- channel.close()
}
val finishTime = System.currentTimeMillis
logDebug("Block %s stored as %s file on disk in %d ms".format(
- file.getName, Utils.bytesToString(bytes.limit), finishTime - startTime))
- PutResult(bytes.limit(), Right(bytes.duplicate()))
+ file.getName,
+ Utils.bytesToString(file.length()),
+ finishTime - startTime))
}
- override def putIterator(
- blockId: BlockId,
- values: Iterator[Any],
- level: StorageLevel,
- returnValues: Boolean): PutResult = {
-
- logDebug(s"Attempting to write values for block $blockId")
- val startTime = System.currentTimeMillis
- val file = diskManager.getFile(blockId)
- val outputStream = new FileOutputStream(file)
- try {
+ def putBytes(blockId: BlockId, _bytes: ByteBuffer): Unit = {
+ // So that we do not modify the input offsets !
+ // duplicate does not copy buffer, so inexpensive
+ val bytes = _bytes.duplicate()
+ put(blockId) { fileOutputStream =>
+ val channel = fileOutputStream.getChannel
Utils.tryWithSafeFinally {
- blockManager.dataSerializeStream(blockId, outputStream, values)
+ while (bytes.remaining > 0) {
+ channel.write(bytes)
+ }
} {
- // Close outputStream here because it should be closed before file is deleted.
- outputStream.close()
+ channel.close()
}
- } catch {
- case e: Throwable =>
- if (file.exists()) {
- if (!file.delete()) {
- logWarning(s"Error deleting ${file}")
- }
- }
- throw e
- }
-
- val length = file.length
-
- val timeTaken = System.currentTimeMillis - startTime
- logDebug("Block %s stored as %s file on disk in %d ms".format(
- file.getName, Utils.bytesToString(length), timeTaken))
-
- if (returnValues) {
- // Return a byte buffer for the contents of the file
- val buffer = getBytes(blockId).get
- PutResult(length, Right(buffer))
- } else {
- PutResult(length, null)
}
}
- private def getBytes(file: File, offset: Long, length: Long): Option[ByteBuffer] = {
+ def getBytes(blockId: BlockId): ByteBuffer = {
+ val file = diskManager.getFile(blockId.name)
val channel = new RandomAccessFile(file, "r").getChannel
Utils.tryWithSafeFinally {
// For small files, directly read rather than memory map
- if (length < minMemoryMapBytes) {
- val buf = ByteBuffer.allocate(length.toInt)
- channel.position(offset)
+ if (file.length < minMemoryMapBytes) {
+ val buf = ByteBuffer.allocate(file.length.toInt)
+ channel.position(0)
while (buf.remaining() != 0) {
if (channel.read(buf) == -1) {
throw new IOException("Reached EOF before filling buffer\n" +
- s"offset=$offset\nfile=${file.getAbsolutePath}\nbuf.remaining=${buf.remaining}")
+ s"offset=0\nfile=${file.getAbsolutePath}\nbuf.remaining=${buf.remaining}")
}
}
buf.flip()
- Some(buf)
+ buf
} else {
- Some(channel.map(MapMode.READ_ONLY, offset, length))
+ channel.map(MapMode.READ_ONLY, 0, file.length)
}
} {
channel.close()
}
}
- override def getBytes(blockId: BlockId): Option[ByteBuffer] = {
- val file = diskManager.getFile(blockId.name)
- getBytes(file, 0, file.length)
- }
-
- def getBytes(segment: FileSegment): Option[ByteBuffer] = {
- getBytes(segment.file, segment.offset, segment.length)
- }
-
- override def getValues(blockId: BlockId): Option[Iterator[Any]] = {
- getBytes(blockId).map(buffer => blockManager.dataDeserialize(blockId, buffer))
- }
-
- override def remove(blockId: BlockId): Boolean = {
+ def remove(blockId: BlockId): Boolean = {
--- End diff --
is this return code useful?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193937837
**[Test build #52683 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52683/consoleFull)** for PR 11534 at commit [`14857b3`](https://github.com/apache/spark/commit/14857b3e53d1b496b7ff07099c53ab2d775f950a).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55616267
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
--- End diff --
need to clarify in javadocs the locking semantics of `getLocalValues` and `getLocalBytes`. In particular, both of these methods acquire the read lock. In the iterator case we release the lock when the iterator is drained. In the bytes case the caller is responsible for releasing the locks. In both cases, if we couldn't get the blocks for some reason, we release the lock and throw an exception instead. (Is that right?)
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55616774
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -701,16 +666,15 @@ private[spark] class BlockManager(
blockId: BlockId,
values: Iterator[Any],
level: StorageLevel,
- tellMaster: Boolean = true,
- effectiveStorageLevel: Option[StorageLevel] = None): Boolean = {
+ tellMaster: Boolean = true): Boolean = {
require(values != null, "Values is null")
- val result = doPut(
+ val maybeIterator = doPutIterator(
blockId,
- IteratorValues(() => values),
+ () => values,
level,
- tellMaster,
- effectiveStorageLevel)
- result == DoPutSucceeded
+ tellMaster)
+ // If doPut() didn't hand work back to us, then block already existed or was successfully stored
+ maybeIterator.isEmpty
--- End diff --
I would call this `failedIterator` or something to be more specific.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55610821
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -407,11 +399,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
+ // Cache the values before returning them
+ memoryStore.putIterator(blockId, valuesFromDisk, level) match {
+ case Left(iter) =>
+ // The memory store put() failed, so it returned the iterator back to us:
+ iter
+ case Right(_) =>
+ // The put() succeeded, so we can read the values back:
+ memoryStore.getValues(blockId).get
+ }
+ } else {
+ valuesFromDisk
+ }
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iterToReturn, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Disk, info.size))
+ } else {
+ releaseLock(blockId)
--- End diff --
do we need to release in finally?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194642013
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52792/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192745799
**[Test build #52521 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52521/consoleFull)** for PR 11534 at commit [`18c07e6`](https://github.com/apache/spark/commit/18c07e6da858be167500220732d3bb4565e98df2).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55278386
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -401,11 +403,49 @@ private[spark] class BlockManager(
}
/**
- * Get block from local block manager.
+ * Get block from local block manager as an iterator of Java objects.
*/
- def getLocal(blockId: BlockId): Option[BlockResult] = {
+ def getLocalValues(blockId: BlockId): Option[BlockResult] = {
logDebug(s"Getting local block $blockId")
- doGetLocal(blockId, asBlockResult = true).asInstanceOf[Option[BlockResult]]
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ val level = info.level
+ logDebug(s"Level for block $blockId is $level")
+ if (level.useMemory && memoryStore.contains(blockId)) {
+ val iter: Iterator[Any] = if (level.deserialized) {
+ memoryStore.getValues(blockId).get
+ } else {
+ dataDeserialize(blockId, memoryStore.getBytes(blockId).get)
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
+ Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
+ } else if (level.useDisk && diskStore.contains(blockId)) {
+ val valuesFromDisk = dataDeserialize(blockId, diskStore.getBytes(blockId))
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
+ // Cache the values before returning them
+ memoryStore.putIterator(blockId, valuesFromDisk, level) match {
--- End diff --
This seems like confusing putIterator semantics...
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55476161
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +752,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
-
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
+ }
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
}
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
--- End diff --
Yep, that's right. The returned iterator, if present, contains the same elements as the original iterator.
This is going to become a little clearer after yet another followup patch to refactor the equivalent memorystore code.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55619604
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -701,16 +666,15 @@ private[spark] class BlockManager(
blockId: BlockId,
values: Iterator[Any],
level: StorageLevel,
- tellMaster: Boolean = true,
- effectiveStorageLevel: Option[StorageLevel] = None): Boolean = {
+ tellMaster: Boolean = true): Boolean = {
require(values != null, "Values is null")
- val result = doPut(
+ val maybeIterator = doPutIterator(
blockId,
- IteratorValues(() => values),
+ () => values,
level,
- tellMaster,
- effectiveStorageLevel)
- result == DoPutSucceeded
+ tellMaster)
+ // If doPut() didn't hand work back to us, then block already existed or was successfully stored
+ maybeIterator.isEmpty
--- End diff --
I just removed the variable, since the entire line and the `isEmpty` fit together.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55615777
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -548,16 +521,9 @@ private[spark] class BlockManager(
* This does not acquire a lock on this block in this JVM.
*/
def getRemote(blockId: BlockId): Option[BlockResult] = {
--- End diff --
can you rename this `getRemoteValues` to be consistent? Now we have
```
getLocalValues
getLocalBytes
getRemote
getRemoteBytes
```
so this one stands out
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193919924
**[Test build #52679 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52679/consoleFull)** for PR 11534 at commit [`988f003`](https://github.com/apache/spark/commit/988f00393676eabfc11e665f20f9ce26388e4c11).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55473498
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,117 +752,193 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return true
}
}
val startTimeMs = System.currentTimeMillis
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
+ // Since we're storing bytes, initiate the replication before storing them locally.
// This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
+ val replicationFuture = if (level.replication > 1) {
+ // Duplicate doesn't copy the bytes, but just creates a wrapper
+ val bufferView = bytes.duplicate()
+ Future {
+ // This is a blocking action and should run in futureExecutionContext which is a cached
+ // thread pool
+ replicate(blockId, bufferView, level)
+ }(futureExecutionContext)
+ } else {
+ null
}
var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
-
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
+ bytes.rewind()
+ val size = bytes.limit()
+
+ try {
+ if (level.useMemory) {
+ // Put it in memory first, even if it also has useDisk set to true;
+ // We will drop it to disk later if the memory store can't hold it.
+ val putSucceeded = if (level.deserialized) {
+ val values = dataDeserialize(blockId, bytes.duplicate())
+ memoryStore.putIterator(blockId, values, level).isRight
} else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
+ memoryStore.putBytes(blockId, size, () => bytes)
+ }
+ if (!putSucceeded && level.useDisk) {
+ logWarning(s"Persisting block $blockId to disk instead.")
+ diskStore.putBytes(blockId, bytes)
}
+ } else if (level.useDisk) {
+ diskStore.putBytes(blockId, bytes)
+ }
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
+ val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
+ blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
+ if (blockWasSuccessfullyStored) {
+ // Now that the block is in either the memory, externalBlockStore, or disk store,
+ // tell the master about it.
+ putBlockInfo.size = size
+ if (tellMaster) {
+ reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
}
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ Option(TaskContext.get()).foreach { c =>
+ c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
+ }
+ }
+ } finally {
+ if (blockWasSuccessfullyStored) {
+ if (keepReadLock) {
+ blockInfoManager.downgradeLock(blockId)
} else {
- blockInfoManager.removeBlock(blockId)
- logWarning(s"Putting block $blockId failed")
+ blockInfoManager.unlock(blockId)
}
+ } else {
+ blockInfoManager.removeBlock(blockId)
+ logWarning(s"Putting block $blockId failed")
}
}
logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))
- if (replicationFuture != null) {
+ if (level.replication > 1) {
// Wait for asynchronous replication to finish
Await.ready(replicationFuture, Duration.Inf)
- } else if (putLevel.replication > 1 && blockWasSuccessfullyStored) {
- val remoteStartTime = System.currentTimeMillis
- val bytesToReplicate: ByteBuffer = {
- doGetLocal(blockId, putBlockInfo, asBlockResult = false)
- .map(_.asInstanceOf[ByteBuffer])
- .getOrElse {
- throw new SparkException(s"Block $blockId was not found even though it was just stored")
- }
+ logDebug("Putting block %s with replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ } else {
+ logDebug("Putting block %s without replication took %s"
+ .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
+ }
+
+ blockWasSuccessfullyStored
+ }
+
+ /**
+ * Put the given block according to the given level in one of the block stores, replicating
+ * the values if necessary.
+ *
+ * If the block already exists, this method will not overwrite it.
+ *
+ * @param keepReadLock if true, this method will hold the read lock when it returns (even if the
+ * block already exists). If false, this method will hold no locks when it
+ * returns.
+ * @return None if the block was already present or if the put succeeded, or Some(iterator)
--- End diff --
Can you comment on any properties of the returned iterator in this case?
Just so I understand, if this returns None, we've walked the input iterator to the end. If it returns an iterator, it has been advanced arbitrarily far along?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195074477
**[Test build #52850 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52850/consoleFull)** for PR 11534 at commit [`2752655`](https://github.com/apache/spark/commit/2752655518c9907dd2c9195f7da8cff28824d71e).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194075324
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194073883
**[Test build #52705 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52705/consoleFull)** for PR 11534 at commit [`a16276e`](https://github.com/apache/spark/commit/a16276e1cfb92fba88f2963b0d918aa14f2500a4).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55280741
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -744,7 +732,7 @@ private[spark] class BlockManager(
level: StorageLevel,
tellMaster: Boolean = true,
effectiveStorageLevel: Option[StorageLevel] = None,
- keepReadLock: Boolean = false): Option[PutResult] = {
+ keepReadLock: Boolean = false): Option[Option[Iterator[Any]]] = {
--- End diff --
This was discussed extensively on https://github.com/apache/spark/pull/11502 and will be fixed once I merge that PR.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55617424
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -428,77 +458,46 @@ private[spark] class BlockManager(
Option(
shuffleBlockResolver.getBlockData(blockId.asInstanceOf[ShuffleBlockId]).nioByteBuffer())
} else {
- doGetLocal(blockId, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
- }
- }
-
- private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {
- blockInfoManager.lockForReading(blockId) match {
- case None =>
- logDebug(s"Block $blockId was not found")
- None
- case Some(info) =>
- doGetLocal(blockId, info, asBlockResult)
+ blockInfoManager.lockForReading(blockId) match {
+ case None =>
+ logDebug(s"Block $blockId was not found")
+ None
+ case Some(info) =>
+ Some(doGetLocalBytes(blockId, info))
+ }
}
}
/**
- * Get a local block from the block manager.
- * Assumes that the caller holds a read lock on the block.
+ * Get block from the local block manager as serialized bytes.
+ *
+ * Must be called while holding a read lock on the block.
+ * Releases the read lock upon exception; keeps the read lock upon successful return.
*/
- private def doGetLocal(
- blockId: BlockId,
- info: BlockInfo,
- asBlockResult: Boolean): Option[Any] = {
+ private def doGetLocalBytes(blockId: BlockId, info: BlockInfo): ByteBuffer = {
val level = info.level
logDebug(s"Level for block $blockId is $level")
-
- // Look for the block in memory
- if (level.useMemory) {
- logDebug(s"Getting block $blockId from memory")
- val result = if (asBlockResult) {
- memoryStore.getValues(blockId).map { iter =>
- val ci = CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
- new BlockResult(ci, DataReadMethod.Memory, info.size)
- }
+ // In order, try to read the serialized bytes from memory, then from disk, then fall back to
+ // serializing in-memory objects, and, finally, throw an exception if the block does not exist.
+ if (level.deserialized) {
+ // Try to avoid expensive serialization by reading a pre-serialized copy from disk:
+ if (level.useDisk && diskStore.contains(blockId)) {
+ diskStore.getBytes(blockId)
--- End diff --
To be more specific, we don't put the bytes into memory here because a) the block is stored at a deserialized level, so per our contract it should _only_ be stored as values in memory and not bytes, and b) the caller of this method (`doGetLocalBytes()`) has requested bytes and not values, so deserializing them is just extra work that's likely not to have a payoff.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194092280
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52707/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195050846
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52845/
Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55618661
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -916,11 +945,9 @@ private[spark] class BlockManager(
}
if (blockWasSuccessfullyStored) {
- DoPutSucceeded
- } else if (iteratorFromFailedMemoryStorePut.isDefined) {
- DoPutIteratorFailed(iteratorFromFailedMemoryStorePut.get)
+ None
} else {
- DoPutBytesFailed
+ Some(iteratorFromFailedMemoryStorePut.get)
--- End diff --
Done.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55735303
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -789,138 +805,113 @@ private[spark] class BlockManager(
// lockNewBlockForWriting returned a read lock on the existing block, so we must free it:
releaseLock(blockId)
}
- return DoPutSucceeded
+ return None
}
}
val startTimeMs = System.currentTimeMillis
-
- // Size of the block in bytes
- var size = 0L
-
- // The level we actually use to put the block
- val putLevel = effectiveStorageLevel.getOrElse(level)
-
- // If we're storing bytes, then initiate the replication before storing them locally.
- // This is faster as data is already serialized and ready to send.
- val replicationFuture = data match {
- case b: ByteBufferValues if putLevel.replication > 1 =>
- // Duplicate doesn't copy the bytes, but just creates a wrapper
- val bufferView = b.buffer.duplicate()
- Future {
- // This is a blocking action and should run in futureExecutionContext which is a cached
- // thread pool
- replicate(blockId, bufferView, putLevel)
- }(futureExecutionContext)
- case _ => null
- }
-
- var blockWasSuccessfullyStored = false
- var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
-
- putBlockInfo.synchronized {
- logTrace("Put for block %s took %s to get into synchronized block"
- .format(blockId, Utils.getUsedTimeMs(startTimeMs)))
-
- try {
- if (putLevel.useMemory) {
- // Put it in memory first, even if it also has useDisk set to true;
- // We will drop it to disk later if the memory store can't hold it.
- data match {
- case IteratorValues(iterator) =>
- memoryStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- case Left(iter) =>
- iteratorFromFailedMemoryStorePut = Some(iter)
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- memoryStore.putBytes(blockId, bytes, putLevel)
- }
- } else if (putLevel.useDisk) {
- data match {
- case IteratorValues(iterator) =>
- diskStore.putIterator(blockId, iterator(), putLevel) match {
- case Right(s) =>
- size = s
- // putIterator() will never return Left (see its return type).
- }
- case ByteBufferValues(bytes) =>
- bytes.rewind()
- size = bytes.limit()
- diskStore.putBytes(blockId, bytes, putLevel)
- }
- } else {
- assert(putLevel == StorageLevel.NONE)
- throw new BlockException(
- blockId, s"Attempted to put block $blockId without specifying storage level!")
- }
-
- val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
- blockWasSuccessfullyStored = putBlockStatus.storageLevel.isValid
- if (blockWasSuccessfullyStored) {
- // Now that the block is in either the memory, externalBlockStore, or disk store,
- // let other threads read it, and tell the master about it.
- putBlockInfo.size = size
- if (tellMaster) {
- reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
- }
- Option(TaskContext.get()).foreach { c =>
- c.taskMetrics().incUpdatedBlockStatuses(Seq((blockId, putBlockStatus)))
- }
- }
- } finally {
- if (blockWasSuccessfullyStored) {
- if (keepReadLock) {
- blockInfoManager.downgradeLock(blockId)
- } else {
- blockInfoManager.unlock(blockId)
- }
+ var blockWasSuccessfullyStored: Boolean = false
+ var result: Option[T] = None
--- End diff --
no need to be a var
```
val result = try {
val res = putBody(putBlockInfo)
blockWasSuccessfullyStored = res.isEmpty
res
} finally {
...
}
...
result
```
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194033595
**[Test build #52707 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52707/consoleFull)** for PR 11534 at commit [`66796b5`](https://github.com/apache/spark/commit/66796b5bf89ddedc9644b9f5692441293c0c0aaa).
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195050842
Merged build finished. Test FAILed.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192758828
Merged build finished. Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-192576747
**[Test build #52506 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52506/consoleFull)** for PR 11534 at commit [`7d2aba3`](https://github.com/apache/spark/commit/7d2aba3997ba3770c45585daee4e023a172a4f50).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-193635954
FYI I rebased this to fix the merge conflicts, so the diff + commit log is now smaller.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194074726
**[Test build #52704 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52704/consoleFull)** for PR 11534 at commit [`7a08a17`](https://github.com/apache/spark/commit/7a08a179f8951abbdcb7e70f6bfb53821fbc7352).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-194647478
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52797/
Test 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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696] Remove BlockStore class & simpli...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11534#issuecomment-195050637
**[Test build #52845 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52845/consoleFull)** for PR 11534 at commit [`2e8b531`](https://github.com/apache/spark/commit/2e8b531ff6393b6d022e36b53f11befd3576f0b5).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13696][WIP] Remove BlockStore class & s...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11534#discussion_r55411268
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -801,43 +780,63 @@ private[spark] class BlockManager(
}
var blockWasSuccessfullyStored = false
- var result: PutResult = null
+ var iteratorFromFailedMemoryStorePut: Option[Iterator[Any]] = None
putBlockInfo.synchronized {
--- End diff --
I think this was a carryover from the very old code where we synchronized on the BlockInfo objects themselves instead of using our new lock interface. I'll remove this to clean things up.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org