[SPARK-22068][CORE]Reduce the duplicate code between putIteratorAsValues and putIteratorAsBytes #19285

ConeyLiu · 2017-09-20T00:42:08Z

What changes were proposed in this pull request?

The code logic between MemoryStore.putIteratorAsValues and Memory.putIteratorAsBytes are almost same, so we should reduce the duplicate code between them.

How was this patch tested?

Existing UT.

ConeyLiu · 2017-09-20T00:43:29Z

Hi @cloud-fan @jiangxb1987 , would you mind take a look ? Thanks a lot.

jerryshao · 2017-09-20T03:06:43Z

ok to test.

SparkQA · 2017-09-20T05:33:23Z

Test build #81961 has finished for PR 19285 at commit d2b8ccd.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-09-20T07:04:44Z

Test build #81971 has finished for PR 19285 at commit 9ea8f49.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-09-20T07:29:49Z

retest this please

SparkQA · 2017-09-20T07:34:24Z

Test build #81981 has finished for PR 19285 at commit 9ea8f49.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-09-20T07:52:52Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

-        new DeserializedMemoryEntry[T](arrayValues, SizeEstimator.estimate(arrayValues), classTag)
-      val size = entry.size
+      // get the precise size
+      val size = estimateSize(true)


Why we need estimateSize(true)? Is this just creating the entry and getting entry.size

We just unrolled the iterator successfully until here. But the size of underlying vector maybe greater than the unrollMemoryUsedByThisBlock which we requested memory for unroll the block. So we need check it again and determine whether we need request more memory. And we only should call bbos.toChunkedByteBuffer or vector.toArray after we requested enough memory.

Here, because the underlying storage is different. For putIteratorAsValues, it use SizeTrackingVector, while putIteratorAsBytes use ChunkedByteBufferOutputStream.

But the previous code just calls entry.size, are you fixing a new bug?

Previously, the putIteratorAsValues seems no problem. But the putIteratorAsBytes doesn't check again after unrolled the iterator. Now the putIterator is copied form previous putIteratorAsValues . For SizeTrackingVector, we could call arrayValues.toIterator to get a iterator again after call SizeTrackingVector.toArray. But for ChunkedByteBufferOutputStream, we can't back to stream after called ChunkedByteBufferOutputStream.toChunkedByteBuffer (the PartiallySerializedBlock need a stream).

It seems deserialized values do not have a precise size, even for SizeEstimator.estimate(arrayValues). This would be confused.

ConeyLiu · 2017-09-20T09:23:13Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

      }
      // Acquire storage memory if necessary to store this block in memory.
      val enoughStorageMemory = {
        if (unrollMemoryUsedByThisBlock <= size) {


Here the size of underlying vector or bytebuffer maybe greater than the unrollMemoryUsedByThisBlock .

cloud-fan · 2017-09-20T14:27:54Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

-        reserveAdditionalMemoryIfNecessary()
+    def estimateSize(precise: Boolean): Long = {
+      if (precise) {
+        serializationStream.flush()


I don't see anywhere in the previous code call flush.

Because there are some data cached in the serializationStream, we can't get the precise size if don't call flush. Previous we don't check again after unrolled the block, and it directly call the serializationStream.close(). But here we maybe need the serializationStream again if we can't get anther unroll memory, so we only should call flush.

can you send a PR to fix this issue for putIteratorAsBytes first? It will make this PR easier to review

OK, I'll do it tomorrow.

@cloud-fan Sorry for the previous saying, I read the code again. Here seems call serializationStream .close is also OK. Because the the iterator is has not value need write, that's meaning the serializationStream don't need anymore.

SparkQA · 2017-09-22T14:07:26Z

Test build #82084 has started for PR 19285 at commit d0fcf4f.

jiangxb1987 · 2017-11-06T11:05:15Z

@ConeyLiu Could you rebase this with the latest master so we can continue review it? Thanks!

ConeyLiu · 2017-11-07T02:06:25Z

It's updated. Thanks a lot.

SparkQA · 2017-11-07T05:28:10Z

Test build #83527 has finished for PR 19285 at commit 3a90ad1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jiangxb1987 · 2017-11-07T14:02:28Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

-   *         original input iterator. The caller must either fully consume this iterator or call
-   *         `close()` on it in order to free the storage memory consumed by the partially-unrolled
-   *         block.
+   * @param memoryMode The values saved mode.


nit: also add param description for blockId、 values and classTag.

jiangxb1987 · 2017-11-07T14:18:47Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+        // We only call need the precise size after all values unrolled.
+        arrayValues = vector.toArray
+        preciseSize = SizeEstimator.estimate(arrayValues)
+        vector = null


It looks scary to put vector to null in the function estimateSize.

jiangxb1987 · 2017-11-07T14:21:14Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+    def createMemoryEntry(): MemoryEntry[T] = {
+      // We successfully unrolled the entirety of this block
+      assert(arrayValues != null, "arrayValue shouldn't be null!")
+      assert(preciseSize != -1, "preciseSize shouldn't be -1")


Under which condition would preciseSize be -1?

jiangxb1987 · 2017-11-07T14:21:37Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+      // We successfully unrolled the entirety of this block
+      assert(arrayValues != null, "arrayValue shouldn't be null!")
+      assert(preciseSize != -1, "preciseSize shouldn't be -1")
+      val entry = new DeserializedMemoryEntry[T](arrayValues, preciseSize, classTag)


Why do we need to create the val entry?

SparkQA · 2017-11-08T04:57:31Z

Test build #83575 has finished for PR 19285 at commit bc3ad4e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-01-19T04:10:12Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+      memoryMode: MemoryMode,
+      storeValue: T => Unit,
+      estimateSize: Boolean => Long,
+      createMemoryEntry: () => MemoryEntry[T]): Either[Long, Long] = {


instead of passing 3 functions, I'd like to introduce

class ValuesHolder { def storeValue(value) def esitimatedSize() def buildEntry(): MemoryEntry }

cloud-fan · 2018-01-19T04:11:01Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+   * OOM exceptions, this method will gradually unroll the iterator while periodically checking
+   * whether there is enough free memory. If the block is successfully materialized, then the
+   * temporary unroll memory used during the materialization is "transferred" to storage memory,
+   * so we won't acquire more memory than is actually needed to store the block.


let's not duplicated this document

cloud-fan · 2018-01-19T04:11:49Z

overall looks good

jerryshao · 2018-01-19T04:49:45Z

Are we targeting this to 2.3 or 2.4?

cloud-fan · 2018-01-19T07:08:23Z

It's just a refactor so I'd like to target it for 2.4

ConeyLiu · 2018-01-24T01:46:01Z

Thanks for your valuable suggestion, the code has been updated.

SparkQA · 2018-01-24T05:06:07Z

Test build #86557 has finished for PR 19285 at commit c988762.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-01-24T05:18:28Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

it can be a local variable.

SparkQA · 2018-01-24T08:05:01Z

Test build #86565 has finished for PR 19285 at commit f392217.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-01-24T11:45:27Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+    val valuesHolder = new SerializedValuesHolder[T](blockId, chunkSize, classTag,
+      memoryMode, serializerManager)

-    if (keepUnrolling) {


is it better to use this code structure?

if (keepUnrolling) { // get precise size and reserve extra memory if needed } if (keepUnrolling) { // create the entry }

I do not understand what you mean, could you explain it more?

putIteratorAsValues and putIteratorAsBytes have different code structure for the last step. In the new putIterator method, you followed the code structure of putIteratorAsValues, is it better to follow the one from putIteratorAsBytes?

Thanks for the detailed explanation. I have been updated, the code looks more clearly now.

cloud-fan · 2018-01-24T11:47:32Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala


+private trait ValuesHolder[T] {
+  def storeValue(value: T): Unit
+  def estimatedSize(roughly: Boolean): Long


this is not a good API design, we can do

trait ValuesHolder { def putValue(value: T) def estimatedSize: Long def getBuilder(): ValuesBuilder } trait ValuesBuilder { def preciseSize: Long def build(): MemoryEntry }

an example

class DeserializedValuesHolder extends ValuesHolder { ... def getBuilder = new ValuesBuilder { val valuesArray = vector.toArray def preciseSize = SizeEstimator.estimate(valuesArray) def buid = ... } } class SerializedValuesHolder extends ValuesHolder { ... def getBuilder = new ValuesBuilder { serializationStream.close() def preciseSize = bbos.size def build = ... } }

Very thanks, I'll update it tomorrow.

cloud-fan · 2018-01-25T04:44:57Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

-        }
+    }
+
+    if (keepUnrolling) {


a little improvement

if (keepUnrolling) { val builder = valuesHolder.getBuilder() ... if (keepUnrolling) { val entry = builder.build() ... Right(entry.size) } else { ... logUnrollFailureMessage(blockId, builder.preciseSize) Left(unrollMemoryUsedByThisBlock) } } else { ... logUnrollFailureMessage(blockId, valueHolder.estimatedSize) Left(unrollMemoryUsedByThisBlock) }

cloud-fan · 2018-01-25T04:55:41Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+    // We successfully unrolled the entirety of this block
+    serializationStream.close()
+
+    override val preciseSize: Long = bbos.size


this can be a def?

cloud-fan · 2018-01-25T04:56:39Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+private trait ValuesHolder[T] {
+  def storeValue(value: T): Unit
+  def estimatedSize(): Long
+  def getBuilder(): ValuesBuilder[T]


add a comment to say that, after getBuilder is called, this ValuesHolder becomes invalid.

SparkQA · 2018-01-25T06:58:13Z

Test build #86620 has finished for PR 19285 at commit b41f1bb.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-01-25T07:28:54Z

Test build #86619 has finished for PR 19285 at commit ded080d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-01-25T08:05:01Z

Test build #86629 has finished for PR 19285 at commit 9e0759f.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-01-25T08:05:02Z

Test build #86630 has finished for PR 19285 at commit 40bdcac.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-01-25T08:28:58Z

retest this please

Ngone51 · 2018-01-25T10:49:49Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

 }

+private trait ValuesBuilder[T] {
+  def preciseSize: Long


Hey guys, why not name the trait as MemoryEntryBuilder? As I see from the code, it is used to build the MemoryEntry.

SparkQA · 2018-01-25T12:43:33Z

Test build #86634 has finished for PR 19285 at commit 40bdcac.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-01-26T05:20:17Z

Test build #86674 has finished for PR 19285 at commit 9d1aeef.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-01-26T11:19:00Z

thanks, merging to master!

ConeyLiu · 2018-01-27T04:09:18Z

thanks all.

ConeyLiu added 4 commits September 17, 2017 17:53

refactor memorystore

2c20dcb

fix conflicts

1205643

fix bug and add some comments

92e1d51

better variable name

6e2e29b

small fix

d2b8ccd

fix unit test errors

9ea8f49

cloud-fan reviewed Sep 20, 2017

View reviewed changes

ConeyLiu commented Sep 20, 2017

View reviewed changes

cloud-fan reviewed Sep 20, 2017

View reviewed changes

small gix

d0fcf4f

ConeyLiu added 2 commits November 7, 2017 09:49

fix conflicts

714c37f

Merge remote-tracking branch 'origin/rmemorystore' into rmemorystore

3a90ad1

jiangxb1987 reviewed Nov 7, 2017

View reviewed changes

address comments

bc3ad4e

cloud-fan reviewed Jan 19, 2018

View reviewed changes

cloud-fan reviewed Jan 24, 2018

View reviewed changes

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala Outdated

Copy link

Contributor

cloud-fan Jan 24, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it can be a local variable.

address comments

f392217

ConeyLiu force-pushed the rmemorystore branch from c988762 to f392217 Compare January 24, 2018 06:08

cloud-fan reviewed Jan 24, 2018

View reviewed changes

ConeyLiu added 2 commits January 25, 2018 11:24

address comments

ded080d

small fix

b41f1bb

cloud-fan reviewed Jan 25, 2018

View reviewed changes

address comments

9e0759f

address comments

40bdcac

Ngone51 reviewed Jan 25, 2018

View reviewed changes

address comments

9d1aeef

asfgit closed this in 3e25251 Jan 26, 2018

ConeyLiu deleted the rmemorystore branch January 27, 2018 04:09

Ngone51 mentioned this pull request Feb 27, 2018

[SPARK-23516][CORE] It is unnecessary to transfer unroll memory to storage memory #20676

Closed

[SPARK-22068][CORE]Reduce the duplicate code between putIteratorAsValues and putIteratorAsBytes #19285

[SPARK-22068][CORE]Reduce the duplicate code between putIteratorAsValues and putIteratorAsBytes #19285

Uh oh!

Conversation

ConeyLiu commented Sep 20, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

ConeyLiu commented Sep 20, 2017

Uh oh!

jerryshao commented Sep 20, 2017

Uh oh!

SparkQA commented Sep 20, 2017

Uh oh!

SparkQA commented Sep 20, 2017

Uh oh!

cloud-fan commented Sep 20, 2017

Uh oh!

SparkQA commented Sep 20, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ConeyLiu Sep 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ConeyLiu Sep 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ConeyLiu Sep 21, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 22, 2017

Uh oh!

jiangxb1987 commented Nov 6, 2017

Uh oh!

ConeyLiu commented Nov 7, 2017

Uh oh!

SparkQA commented Nov 7, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Nov 8, 2017

Uh oh!

cloud-fan Jan 19, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jan 19, 2018

Uh oh!

jerryshao commented Jan 19, 2018

Uh oh!

cloud-fan commented Jan 19, 2018

Uh oh!

ConeyLiu Sep 20, 2017 •

edited

Loading

ConeyLiu Sep 20, 2017 •

edited

Loading

ConeyLiu Sep 21, 2017 •

edited

Loading

cloud-fan Jan 19, 2018 •

edited

Loading