YARN-6546. SLS is slow while loading 10k queues. (Yufei Gu via Haibo Chen)

This commit is contained in:
Haibo Chen 2017-10-17 16:03:46 -07:00
parent f27a4ad032
commit 46eb1033a8
7 changed files with 36 additions and 96 deletions

View File

@ -731,6 +731,7 @@ public class SLSRunner extends Configured implements Tool {
} }
queueAppNumMap.put(queueName, appNum); queueAppNumMap.put(queueName, appNum);
wrapper.getSchedulerMetrics().trackQueue(queueName);
} }
private void runNewAM(String jobType, String user, private void runNewAM(String jobType, String user,

View File

@ -28,9 +28,4 @@ public class CapacitySchedulerMetrics extends SchedulerMetrics {
public CapacitySchedulerMetrics() { public CapacitySchedulerMetrics() {
super(); super();
} }
@Override
public void trackQueue(String queueName) {
trackedQueues.add(queueName);
}
} }

View File

@ -167,8 +167,9 @@ public class FairSchedulerMetrics extends SchedulerMetrics {
} }
@Override @Override
public void trackQueue(String queueName) { protected void registerQueueMetrics(String queueName) {
trackedQueues.add(queueName); super.registerQueueMetrics(queueName);
FairScheduler fair = (FairScheduler) scheduler; FairScheduler fair = (FairScheduler) scheduler;
final FSQueue queue = fair.getQueueManager().getQueue(queueName); final FSQueue queue = fair.getQueueManager().getQueue(queueName);
registerQueueMetrics(queue, Metric.DEMAND); registerQueueMetrics(queue, Metric.DEMAND);
@ -209,16 +210,4 @@ public class FairSchedulerMetrics extends SchedulerMetrics {
} }
); );
} }
@Override
public void untrackQueue(String queueName) {
trackedQueues.remove(queueName);
for (Metric metric: Metric.values()) {
metrics.remove("variable.queue." + queueName + "." +
metric.value + ".memory");
metrics.remove("variable.queue." + queueName + "." +
metric.value + ".vcores");
}
}
} }

View File

@ -35,8 +35,9 @@ public class FifoSchedulerMetrics extends SchedulerMetrics {
} }
@Override @Override
public void trackQueue(String queueName) { protected void registerQueueMetrics(String queueName) {
trackedQueues.add(queueName); super.registerQueueMetrics(queueName);
FifoScheduler fifo = (FifoScheduler) scheduler; FifoScheduler fifo = (FifoScheduler) scheduler;
// for FifoScheduler, only DEFAULT_QUEUE // for FifoScheduler, only DEFAULT_QUEUE
// here the three parameters doesn't affect results // here the three parameters doesn't affect results

View File

@ -323,25 +323,6 @@ public class SLSCapacityScheduler extends CapacityScheduler implements
queueName); queueName);
} }
private void initQueueMetrics(CSQueue queue) {
if (queue instanceof LeafQueue) {
schedulerMetrics.initQueueMetric(queue.getQueueName());
return;
}
for (CSQueue child : queue.getChildQueues()) {
initQueueMetrics(child);
}
}
@Override
public void serviceInit(Configuration configuration) throws Exception {
super.serviceInit(configuration);
if (metricsON) {
initQueueMetrics(getRootQueue());
}
}
@Override @Override
public void serviceStop() throws Exception { public void serviceStop() throws Exception {
try { try {

View File

@ -306,25 +306,6 @@ public class SLSFairScheduler extends FairScheduler
queueName); queueName);
} }
private void initQueueMetrics(FSQueue queue) {
if (queue instanceof FSLeafQueue) {
schedulerMetrics.initQueueMetric(queue.getQueueName());
return;
}
for (FSQueue child : queue.getChildQueues()) {
initQueueMetrics(child);
}
}
@Override
public void serviceInit(Configuration conf) throws Exception {
super.serviceInit(conf);
if (metricsON) {
initQueueMetrics(getQueueManager().getRootQueue());
}
}
@Override @Override
public void serviceStop() throws Exception { public void serviceStop() throws Exception {
try { try {

View File

@ -234,14 +234,35 @@ public abstract class SchedulerMetrics {
} }
} }
public abstract void trackQueue(String queueName); /**
* Track a queue by registering its metrics.
public void untrackQueue(String queueName) { *
for (String m : queueTrackedMetrics) { * @param queue queue name
metrics.remove("variable.queue." + queueName + "." + m); */
public void trackQueue(String queue) {
queueLock.lock();
try {
if (!isTracked(queue)) {
trackedQueues.add(queue);
registerQueueMetrics(queue);
}
} finally {
queueLock.unlock();
} }
} }
protected void registerQueueMetrics(String queueName) {
SortedMap<String, Counter> counterMap = metrics.getCounters();
for (QueueMetric queueMetric : QueueMetric.values()) {
String metricName = getQueueMetricName(queueName, queueMetric);
if (!counterMap.containsKey(metricName)) {
metrics.counter(metricName);
queueTrackedMetrics.add(metricName);
}
}
}
public boolean isTracked(String queueName) { public boolean isTracked(String queueName) {
return trackedQueues.contains(queueName); return trackedQueues.contains(queueName);
} }
@ -547,40 +568,13 @@ public abstract class SchedulerMetrics {
return "counter.queue." + queue + "." + metric.value; return "counter.queue." + queue + "." + metric.value;
} }
private void traceQueueIfNotTraced(String queue) {
queueLock.lock();
try {
if (!isTracked(queue)) {
trackQueue(queue);
}
} finally {
queueLock.unlock();
}
}
void initQueueMetric(String queueName){
SortedMap<String, Counter> counterMap = metrics.getCounters();
for (QueueMetric queueMetric : QueueMetric.values()) {
String metricName = getQueueMetricName(queueName, queueMetric);
if (!counterMap.containsKey(metricName)) {
metrics.counter(metricName);
counterMap = metrics.getCounters();
}
}
traceQueueIfNotTraced(queueName);
}
void updateQueueMetrics(Resource pendingResource, Resource allocatedResource, void updateQueueMetrics(Resource pendingResource, Resource allocatedResource,
String queueName) { String queueName) {
trackQueue(queueName);
SortedMap<String, Counter> counterMap = metrics.getCounters(); SortedMap<String, Counter> counterMap = metrics.getCounters();
for(QueueMetric metric : QueueMetric.values()) { for(QueueMetric metric : QueueMetric.values()) {
String metricName = getQueueMetricName(queueName, metric); String metricName = getQueueMetricName(queueName, metric);
if (!counterMap.containsKey(metricName)) {
metrics.counter(metricName);
counterMap = metrics.getCounters();
}
if (metric == QueueMetric.PENDING_MEMORY) { if (metric == QueueMetric.PENDING_MEMORY) {
counterMap.get(metricName).inc(pendingResource.getMemorySize()); counterMap.get(metricName).inc(pendingResource.getMemorySize());
@ -592,8 +586,6 @@ public abstract class SchedulerMetrics {
counterMap.get(metricName).inc(allocatedResource.getVirtualCores()); counterMap.get(metricName).inc(allocatedResource.getVirtualCores());
} }
} }
traceQueueIfNotTraced(queueName);
} }
void updateQueueMetricsByRelease(Resource releaseResource, String queue) { void updateQueueMetricsByRelease(Resource releaseResource, String queue) {