[SPARK-18231] Optimise SizeEstimator implementation #16196

a-roberts · 2016-12-07T10:42:48Z

What changes were proposed in this pull request?

Several improvements to the SizeEstimator for performance, most of the benefit comes from, when estimating, contending to not contending on multiple threads. There can be a small boost in uncontended scenarios from the removal of the synchronisation code but the cost of that synchronisation when not truly contended is low. On the PageRank workload for HiBench we see 10-15% performance improvements (measuring elapsed times on average) with both IBM's SDK for Java and OpenJDK 8. I don't see any changes other than noise for the other workloads on this benchmark.

How was this patch tested?

Existing unit tests but there are problems to resolve.

I see SizeEstimatorSuite and SizeTrackerSuite failing with at least IBM Java now due to smaller sizes being reported than the test expects (let's see what happens with OpenJDK on the community runs).

In SizeTrackerSuite I think the failures are caused by using ThreadLocalRandom and not Random - because with Random we see these tests passing again. Not sure how robust SizeTrackerSuite is though.

For performance testing I've used HiBench, large profile, with one executor ranging from 10g to 25g, experimenting with fixed and dynamic heaps. The Spark code I've based my results on is from December the 1st (master branch, so 2.1.0 snapshot).

More details on the optimisations (this being phase one and JDK agnostic) at www.spark.tc/improvements-to-the-sizeestimator-class

Several improvements to the SizeEstimator for performance, most of the benefit comes from, when estimating, contending to not contending on multiple threads. There can be a small boost in uncontended scenarios from the removal of the synchronisation code but the cost of that synchronisation when not truly contended is low. On the PageRank workload for HiBench we see 49~ second durations reduced to ~41 second durations. I don't see any changes for other workloads. Observed with both IBM's SDK for Java and OpenJDK.

mridulm · 2016-12-07T10:51:38Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

    }
    pointerSize = if (is64bit && !isCompressedOops) 8 else 4
-    classInfos.clear()
-    classInfos.put(classOf[Object], new ClassInfo(objectSize, Nil))


We should preserve behavior here.
Apply the same on classInfos.get()

No need to clear() because we're using ThreadLocal WeakHashMaps so the initialisation will occur once per thread, we put the new value in and we assume the size won't be changing which is going to be the case except for running in some debug modes*. As they're in a ThreadLocal WeakHashMap the classes can still be unloaded if no longer used.

I'm referring to JDK debug modes such as fullspeeddebug where we can change class layouts at runtime. Normal execution modes prohibit changes to the class layout during execution without the class being unloaded and reloaded which will trigger the map entry to clear and be recreated due to its weakness (so in normal cases the map will always be correct)

The map is expected to contain class info from the current execution; not from past runs (which might or might not be relevant - increasing the map size).
We would never have cleared it if it was not required.

mridulm · 2016-12-07T10:52:25Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

+    override def initialValue(): java.util.WeakHashMap[Class[_], ClassInfo] = {
+      val toReturn = new WeakHashMap[Class[_], ClassInfo]()
+      toReturn.put(classOf[Object], new ClassInfo(objectSize, new Array[Int](0)))
+      return toReturn


Why not keep the returned value same as before ?
And move the initialization back into initialize() - so that use of classInfos Map across threads wont happen.

Built and profiled, averaging 42 sec run times with the initial commit, averaging 45 second run times with this. No changes = 48 sec.

My code as a diff (so using a ConcurrentHashMap and var not val so we can initialise it later) provided here:

import java.lang.management.ManagementFactory import java.lang.reflect.{Field, Modifier} import java.util.{IdentityHashMap, WeakHashMap} -import java.util.concurrent.ThreadLocalRandom +import java.util.concurrent.{ThreadLocalRandom, ConcurrentMap} import scala.collection.mutable.ArrayBuffer import scala.concurrent.util.Unsafe @@ -88,16 +88,6 @@ object SizeEstimator extends Logging { // TODO: Is this arch dependent ? private val ALIGN_SIZE = 8 - // A cache of ClassInfo objects for each class - // We use weakKeys to allow GC of dynamically created classes - private val classInfos = new ThreadLocal[WeakHashMap[Class[_], ClassInfo]] { - override def initialValue(): java.util.WeakHashMap[Class[_], ClassInfo] = { - val toReturn = new WeakHashMap[Class[_], ClassInfo]() - toReturn.put(classOf[Object], new ClassInfo(objectSize, new Array[Int](0))) - return toReturn - } - } - // Object and pointer sizes are arch dependent private var is64bit = false @@ -109,6 +99,8 @@ object SizeEstimator extends Logging { // Minimum size of a java.lang.Object private var objectSize = 8 + private var classInfos: ConcurrentMap[Class[_], ClassInfo] = null + initialize() // Sets object size, pointer size based on architecture and CompressedOops settings @@ -126,6 +118,9 @@ object SizeEstimator extends Logging { } } pointerSize = if (is64bit && !isCompressedOops) 8 else 4 + + classInfos = new MapMaker().weakKeys().makeMap[Class[_], ClassInfo]() + classInfos.put(classOf[Object], new ClassInfo(objectSize, new Array[Int](0))) } private def getIsCompressedOops: Boolean = { @@ -338,7 +333,7 @@ object SizeEstimator extends Logging { */ private def getClassInfo(cls: Class[_]): ClassInfo = { // Check whether we've already cached a ClassInfo for this class - val info = classInfos.get().get(cls) + val info = classInfos.get(cls) if (info != null) { return info } @@ -371,7 +366,7 @@ object SizeEstimator extends Logging { // Create and cache a new ClassInfo val newInfo = new ClassInfo(shellSize, fieldOffsets.toArray) - classInfos.get().put(cls, newInfo) + classInfos.put(cls, newInfo) newInfo }

What I meant was, continue to use ThreadLocal, but maintain the MapMaker's result for thlocal.get()

And move the initilization to initialize() instead of in initialValue()

Got it, thanks will give this a try

mridulm · 2016-12-07T11:02:07Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

+        val s1 = sampleArray(objArray, state, rand, drawn, length)
+        val s2 = sampleArray(objArray, state, rand, drawn, length)
        val size = math.min(s1, s2)
+


Changes to this method are excellent and should speed things up !

mridulm · 2016-12-07T11:02:50Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

+    // avoid the use of an iterator derrived from the range syntax here for performance
+    var count = 0
+    val end = ARRAY_SAMPLE_SIZE
+    while (count <= end) {


< end for until semantics

Ah yes, should be just < not <=, will add into the next commit

mridulm · 2016-12-07T11:07:35Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

    // Create and cache a new ClassInfo
-    val newInfo = new ClassInfo(shellSize, pointerFields)
-    classInfos.put(cls, newInfo)
+    val newInfo = new ClassInfo(shellSize, fieldOffsets.toArray)


We are loosing out on padding due to allignment here which the earlier code was computing. No ?

Will look into this and determine if the padding is needed

SparkQA · 2016-12-07T13:00:24Z

Test build #69795 has finished for PR 16196 at commit 50af8fc.

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

srowen

I'm kind of concerned that this is changing a lot, some of which isn't obviously without problems or risk, for marginal gains. I'd prefer to stick to obviously correct wins

srowen · 2016-12-07T19:53:53Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

+      array: Array[AnyRef],
      state: SearchState,
-      rand: Random,
+      rand: ThreadLocalRandom,


I don't think this has to change

srowen · 2016-12-07T19:56:50Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

  // We use weakKeys to allow GC of dynamically created classes
-  private val classInfos = new MapMaker().weakKeys().makeMap[Class[_], ClassInfo]()
+  private val classInfos = new ThreadLocal[WeakHashMap[Class[_], ClassInfo]] {
+    override def initialValue(): java.util.WeakHashMap[Class[_], ClassInfo] = {


Nit: remove java.util, and 'return' below. "map" is better than "toReturn"

This is going to expand the memory footprint, because redundant copies of this info will be maintained per thread. Is the contention that significant?

srowen · 2016-12-07T20:00:41Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

+          val fieldCount = classInfo.fieldOffsets.length
+          val us = Unsafe.instance
+          while (index < fieldCount) {
+            state.enqueue(us.getObject(obj, classInfo.fieldOffsets(index).toLong))


I understand avoiding reflection, but this is a dicier way to access fields of an object. I don't have a specific reason this would fail but the fact that it uses unsafe is riskier. Is this worth it?

srowen · 2016-12-07T20:01:48Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

-    for (i <- 0 until ARRAY_SAMPLE_SIZE) {
+    // avoid the use of an iterator derrived from the range syntax here for performance
+    var count = 0
+    val end = ARRAY_SAMPLE_SIZE


end is redundant here

srowen · 2016-12-07T20:03:20Z

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

        val fieldClass = field.getType
        if (fieldClass.isPrimitive) {
-          sizeCount(primitiveSize(fieldClass)) += 1
+          if (cls == classOf[Double] || cls == classOf[Long]) {


This and the logic changes below aren't obviously OK. this seems to lose a lot of logic. I think this has to be explained or backed out

rxin · 2016-12-08T08:45:31Z

+1 on @srowen's suggestion. This change is not surgical at all. It is going to be difficult to guarantee no behavior change. If anything I'd favor correctness over performance here.

srowen · 2016-12-11T09:41:14Z

Ping @a-roberts -- I think some sections of this are clearly a win, like near #16196 (comment) but maybe best to back out anything controversial. And touch up the style. Then I think this could be ready.

a-roberts · 2016-12-11T11:52:04Z

Agreed, I'll be back working on this and answering the queries after the 2.1.0 release vote passes, that's my current priority as we're nearing the Christmas break period

srowen · 2016-12-30T10:33:45Z

Ping to keep this on the radar; these couple PRs have been open a long time

a-roberts · 2017-01-02T19:51:20Z

It's on my to-do list, working on this once I'm back from an end of year break, can I get a list of concerns here please? I'll run with them and I think one concern is that we'll underestimate and this'll lead to insufficient memory problems at runtime

Once we figure this out I can add the scenario(s) above as unit tests to ensure any changes here are entirely correct

srowen · 2017-01-03T11:18:59Z

See the discussion above; I think the request is to back out much of this and leave only the clearly correct improvements like #16196 (comment) There are still a number of small comments and questions outstanding, which you can go back and browse.

srowen · 2017-01-09T12:51:28Z

Ping @a-roberts ; let's close this if not going to proceed, though at least part of it looks like a clear win.

a-roberts · 2017-01-09T13:39:50Z

We can close it for now and I'll reopen it once the changes are more conservative and I've done plenty of testing/profiling - or anybody else could do so by keeping in only the changes we deem as safe, unfortunately I have too much other Spark work on at the moment to give this PR the attention it deserves

Closes apache#15736 Closes apache#16309 Closes apache#16485 Closes apache#16502 Closes apache#16196 Closes apache#16498 Closes apache#12380 Closes apache#16764

mridulm reviewed Dec 7, 2016

View reviewed changes

srowen requested changes Dec 7, 2016

View reviewed changes

srowen added a commit to srowen/spark that referenced this pull request Feb 2, 2017

Close stale PRs

4a54bb4

Closes apache#15736 Closes apache#16309 Closes apache#16485 Closes apache#16502 Closes apache#16196 Closes apache#16498 Closes apache#12380 Closes apache#16764

srowen mentioned this pull request Feb 2, 2017

[BUILD] Close stale PRs #16778

Closed

asfgit closed this in 20b4ca1 Feb 3, 2017

[SPARK-18231] Optimise SizeEstimator implementation #16196

[SPARK-18231] Optimise SizeEstimator implementation #16196

Uh oh!

Conversation

a-roberts commented Dec 7, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mridulm Dec 7, 2016 • 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!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 7, 2016

Uh oh!

srowen left a comment

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rxin commented Dec 8, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

srowen commented Dec 11, 2016

Uh oh!

a-roberts commented Dec 11, 2016

Uh oh!

srowen commented Dec 30, 2016

Uh oh!

a-roberts commented Jan 2, 2017

Uh oh!

srowen commented Jan 3, 2017

Uh oh!

srowen commented Jan 9, 2017

Uh oh!

a-roberts commented Jan 9, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

mridulm Dec 7, 2016 •

edited

Loading

rxin commented Dec 8, 2016 •

edited

Loading