Do not re-use objects in the EdgePartition/EdgeTriplet iterators.

[darabos]

This avoids a silent data corruption issue (https://spark-project.atlassian.net/browse/SPARK-1188) and has no performance impact by my measurements. It also simplifies the code. As far as I can tell the object re-use was nothing but premature optimization.

I did actual benchmarks for all the included changes, and there is no performance difference. I am not sure where to put the benchmarks. Does Spark not have a benchmark suite?

This is an example benchmark I did:

test("benchmark") {
val builder = new EdgePartitionBuilder[Int]
for (i <- (1 to 10000000)) {
builder.add(i.toLong, i.toLong, i)
}
val p = builder.toEdgePartition
p.map(_.attr + 1).iterator.toList
}

It ran for 10 seconds both before and after this change.

[AmplabJenkins]

Can one of the admins verify this patch?

[darabos]

Sorry, the new JIRA link is https://issues.apache.org/jira/browse/SPARK-1188. Thanks!

[rxin]

Thanks for sending this in. It's something that is extremely confusing in graphx and we need to fix it. However, I am not sure if taking out the object reuse in edges is the way to fix this problem.

This is actually hard to test in micro benchmarks, because you rarely get GCs in micro benchmarks. In these cases where an edge/triplet is returned by an iterator, JVM's escape analysis cannot capture the scope and cannot do on-stack allocation. As a result, there are lots of temporary objects allocated in the heap.

Allocations of short-lived objects are supposed to be cheap. The most expensive objects to gc are medium-lived objects. However, the more temporary objects we allocate, the more frequent a young gen gc happens. The more frequent young gen gc happens, the more likely for random objects to become medium lived.

Maybe a better way to fix this is to leave the object reuse as is, but in all places where we return the object to the user, we should make sure it copies it.

I just looked at the code, and I think we can accomplish that by just adding a copy to EdgeRDD.compute, i.e.

  override def compute(part: Partition, context: TaskContext): Iterator[Edge[ED]] = {
    firstParent[(PartitionID, EdgePartition[ED])].iterator(part, context).next._2.iterator.map(_.copy())
  }

[rxin]

And for EdgeTripletIterator we should just keep your change (i.e. always create a new row) since that is not used internally at all.

[darabos]

Thanks for the comments! The description of the GC effects was very educational. I made the suggested changes. Let me know if you'd like to see something else changed.

[rxin]

Can we revert this change? This is where Scala fails us. The while loop is ~ 2x faster than iter.toArray in my micro benchmark.

[darabos]

Done. Yeah, I didn't benchmark this, sorry.

I guess toArray must work with a dynamic array, since it does not know the size in advance. There is an Iterator.copyToArray, but regrettably it does not return the number of elements copied, so we would lose the important assertion. I filed a feature request for this. (https://issues.scala-lang.org/browse/SI-8469) Hopefully I'm not being obtuse.

[rxin]

Thanks @darabos. The change looks good to me other than the one place I pointed out.

[rxin]

Merged. Thanks! This is a very important usability improvement.

cc @jegonzal @ankurdave