viirya commented on code in PR #59331:
URL: https://github.com/apache/spark/pull/59331#discussion_r4252738587
##########
core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java:
##########
@@ -113,10 +113,12 @@ public class TaskMemoryManager {
final MemoryMode tungstenMemoryMode;
/**
- * Tracks spillable memory consumers.
+ * Tracks spillable memory consumers. Consumers are tracked by identity
rather than by
+ * {@code equals}, so that distinct consumers which happen to be equal (for
example, Scala case
Review Comment:
Done, used your wording.
##########
core/src/main/java/org/apache/spark/memory/MemoryConsumer.java:
##########
@@ -156,6 +166,21 @@ public void freeMemory(long size) {
used.getAndAdd(-size);
}
+ /**
+ * Memory consumers are compared by identity. This is not final so that
existing subclasses which
+ * override it keep compiling, but its presence stops Scala case classes
from generating value
+ * equality.
+ */
+ @Override
+ public boolean equals(Object other) {
Review Comment:
Agreed. Whichever of the two PRs merges second will update the `HybridQueue`
comment on master to point at the `MemoryConsumer` contract instead of the
`HashSet`. The maintenance branches only get #59282, so its comment and
overrides stay as they are there.
##########
core/src/main/java/org/apache/spark/memory/MemoryConsumer.java:
##########
@@ -28,6 +28,13 @@
* A memory consumer of {@link TaskMemoryManager} that supports spilling.
*
* Note: this only supports allocation / spilling of Tungsten memory.
+ *
+ * Memory consumers have identity equality: {@link #equals} and {@link
#hashCode} are defined here
+ * by reference, so a subclass, including a Scala case class, does not get
value equality unless it
Review Comment:
Good point. Added "when compiled against this version" to the class Javadoc
and to the PR description, and noted on `equals` that a case class compiled
against an earlier version keeps its generated equality until it is rebuilt.
The PR description now also gives that as a reason for keeping the identity set
and `ne` in `Spillable`.
##########
core/src/main/java/org/apache/spark/memory/MemoryConsumer.java:
##########
@@ -156,6 +166,21 @@ public void freeMemory(long size) {
used.getAndAdd(-size);
}
+ /**
+ * Memory consumers are compared by identity. This is not final so that
existing subclasses which
Review Comment:
Done, the PR title and SPARK-60112 are now "Use identity equality for memory
consumers".
##########
core/src/test/scala/org/apache/spark/memory/MemoryConsumerSuite.scala:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.memory
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+
+class MemoryConsumerSuite extends SparkFunSuite {
+
+ test("case class consumers have identity equality") {
+ val taskMemoryManager = new TaskMemoryManager(new TestMemoryManager(new
SparkConf()), 0)
+ val consumer = CaseClassConsumer(taskMemoryManager, 1)
+ val other = CaseClassConsumer(taskMemoryManager, 1)
+
+ // The case class does not generate value-based equals/hashCode because
MemoryConsumer
+ // already defines them.
+ assert(consumer != other)
+ assert(consumer == consumer)
+ assert(consumer.hashCode === System.identityHashCode(consumer))
+ // A spill triggered by an equal-looking consumer is not mistaken for a
self-triggered one.
+ assert(consumer.spill(Long.MaxValue, other) === 1)
Review Comment:
Done. `CaseClassConsumer` now extends `TestMemoryConsumer`, and a new test
registers two consumers with equal arguments with `TaskMemoryManager`: when
`other` requests more memory than is free, `consumer` spills and releases all
of it. It fails without the identity `equals` in `MemoryConsumer`. I also added
to the helper's comment that real consumers should still use `eq`, as the
`spill` Javadoc says.
##########
core/src/main/java/org/apache/spark/memory/MemoryConsumer.java:
##########
@@ -28,6 +28,13 @@
* A memory consumer of {@link TaskMemoryManager} that supports spilling.
*
* Note: this only supports allocation / spilling of Tungsten memory.
+ *
+ * Memory consumers have identity equality: {@link #equals} and {@link
#hashCode} are defined here
+ * by reference, so a subclass, including a Scala case class, does not get
value equality unless it
+ * explicitly overrides them. {@link TaskMemoryManager} also tracks consumers
by identity, so every
+ * distinct instance is tracked, offered for spilling and reported in the
memory usage breakdown,
Review Comment:
Done, the class Javadoc now says only that distinct instances are never
merged into one entry, so each of them can be offered for spilling and reported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
