You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ep...@apache.org on 2021/12/21 19:54:14 UTC
[hadoop] branch branch-2.10 updated: YARN-10178: Global Scheduler async thread crash caused by 'Comparison method violates its general contract. Contributed by Andras Gyori (gandras) and Qi Zhu (zhuqi).
This is an automated email from the ASF dual-hosted git repository.
epayne pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 9ee5265 YARN-10178: Global Scheduler async thread crash caused by 'Comparison method violates its general contract. Contributed by Andras Gyori (gandras) and Qi Zhu (zhuqi).
9ee5265 is described below
commit 9ee5265fb33b66421dc94034f54307fae7cff096
Author: Eric Payne <ep...@apache.org>
AuthorDate: Tue Dec 21 19:48:06 2021 +0000
YARN-10178: Global Scheduler async thread crash caused by 'Comparison method violates its general contract. Contributed by Andras Gyori (gandras) and Qi Zhu (zhuqi).
---
.../PriorityUtilizationQueueOrderingPolicy.java | 65 +++++++++++++++++-----
1 file changed, 51 insertions(+), 14 deletions(-)
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/policy/PriorityUtilizationQueueOrderingPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/policy/PriorityUtilizationQueueOrderingPolicy.java
index ada665d..c475be3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/policy/PriorityUtilizationQueueOrderingPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/policy/PriorityUtilizationQueueOrderingPolicy.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
@@ -95,24 +96,26 @@ public class PriorityUtilizationQueueOrderingPolicy implements QueueOrderingPoli
/**
* Comparator that both looks at priority and utilization
*/
- private class PriorityQueueComparator implements Comparator<CSQueue> {
+ private class PriorityQueueComparator
+ implements Comparator<PriorityQueueResourcesForSorting> {
@Override
- public int compare(CSQueue q1, CSQueue q2) {
+ public int compare(PriorityQueueResourcesForSorting q1Sort,
+ PriorityQueueResourcesForSorting q2Sort) {
String p = partitionToLookAt.get();
- int rc = compareQueueAccessToPartition(q1, q2, p);
+ int rc = compareQueueAccessToPartition(q1Sort.queue, q2Sort.queue, p);
if (0 != rc) {
return rc;
}
- float used1 = q1.getQueueCapacities().getUsedCapacity(p);
- float used2 = q2.getQueueCapacities().getUsedCapacity(p);
+ float used1 = q1Sort.usedCapacity;
+ float used2 = q2Sort.usedCapacity;
int p1 = 0;
int p2 = 0;
if (respectPriority) {
- p1 = q1.getPriority().getPriority();
- p2 = q2.getPriority().getPriority();
+ p1 = q1Sort.queue.getPriority().getPriority();
+ p2 = q2Sort.queue.getPriority().getPriority();
}
rc = PriorityUtilizationQueueOrderingPolicy.compare(used1, used2, p1, p2);
@@ -120,8 +123,8 @@ public class PriorityUtilizationQueueOrderingPolicy implements QueueOrderingPoli
// For queue with same used ratio / priority, queue with higher configured
// capacity goes first
if (0 == rc) {
- float abs1 = q1.getQueueCapacities().getAbsoluteCapacity(p);
- float abs2 = q2.getQueueCapacities().getAbsoluteCapacity(p);
+ float abs1 = q1Sort.absoluteCapacity;
+ float abs2 = q2Sort.absoluteCapacity;
return Float.compare(abs2, abs1);
}
@@ -156,6 +159,29 @@ public class PriorityUtilizationQueueOrderingPolicy implements QueueOrderingPoli
}
}
+ /**
+ * A simple storage class to represent a snapshot of a queue.
+ */
+ public static class PriorityQueueResourcesForSorting {
+ private final float usedCapacity;
+ private final float absoluteCapacity;
+ private final CSQueue queue;
+
+ PriorityQueueResourcesForSorting(CSQueue queue) {
+ this.queue = queue;
+ this.usedCapacity =
+ queue.getQueueCapacities().
+ getUsedCapacity(partitionToLookAt.get());
+ this.absoluteCapacity =
+ queue.getQueueCapacities().
+ getAbsoluteCapacity(partitionToLookAt.get());
+ }
+
+ public CSQueue getQueue() {
+ return queue;
+ }
+ }
+
public PriorityUtilizationQueueOrderingPolicy(boolean respectPriority) {
this.respectPriority = respectPriority;
}
@@ -167,12 +193,23 @@ public class PriorityUtilizationQueueOrderingPolicy implements QueueOrderingPoli
@Override
public Iterator<CSQueue> getAssignmentIterator(String partition) {
- // Since partitionToLookAt is a thread local variable, and every time we
- // copy and sort queues, so it's safe for multi-threading environment.
+ // partitionToLookAt is a thread local variable, therefore it is safe to mutate it.
PriorityUtilizationQueueOrderingPolicy.partitionToLookAt.set(partition);
- List<CSQueue> sortedQueue = new ArrayList<>(queues);
- Collections.sort(sortedQueue, new PriorityQueueComparator());
- return sortedQueue.iterator();
+
+ // Sort the snapshot of the queues in order to avoid breaking the prerequisites of TimSort.
+ // See YARN-10178 for details.
+ List<PriorityQueueResourcesForSorting> queueSnapshots = new ArrayList<>();
+ for (CSQueue queue : queues) {
+ queueSnapshots.add(new PriorityQueueResourcesForSorting(queue));
+ }
+ Collections.sort(queueSnapshots, new PriorityQueueComparator());
+
+ List<CSQueue> sortedQueues = new ArrayList<>();
+ for (PriorityQueueResourcesForSorting queueSnapshot : queueSnapshots) {
+ sortedQueues.add(queueSnapshot.queue);
+ }
+
+ return sortedQueues.iterator();
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org