You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by sa...@apache.org on 2018/07/19 08:40:21 UTC
[incubator-nemo] branch master updated: [NEMO-164] Reduce Overhead
of FreeSlotSchedulingConstraint (#79)
This is an automated email from the ASF dual-hosted git repository.
sanha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
The following commit(s) were added to refs/heads/master by this push:
new 3e9b400 [NEMO-164] Reduce Overhead of FreeSlotSchedulingConstraint (#79)
3e9b400 is described below
commit 3e9b400395430cf950a08d247f3c24f0c93e373e
Author: Jangho Seo <ja...@jangho.io>
AuthorDate: Thu Jul 19 17:40:19 2018 +0900
[NEMO-164] Reduce Overhead of FreeSlotSchedulingConstraint (#79)
JIRA: [NEMO-164: Reduce Overhead of FreeSlotSchedulingConstraint](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-164)
**Major changes:**
- Improves ExecutorRepresenter to track set of tasks with ExecutorSlotComplianceProperty=true and ExecutorSlotComplianceProperty=false separately.
- Improves FreeSlotSchedulingConstraint to use ExecutorRepresenter#getNumOfComplyingRunningTasks, instead of examining ExecutorSlotComplianceProperty for each task.
**Minor changes to note:**
- N/A
**Tests for the changes:**
- This PR modifies mock for ExecutorRepresenter in FreeSlotSchedulingConstraintTest in line with the changes on ExecutorRepresenter.
**Other comments:**
- @sanha reported the problem and demonstrated the solution.
resolves [NEMO-164](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-164)
---
.../master/resource/ExecutorRepresenter.java | 107 ++++++++++++++-------
.../scheduler/FreeSlotSchedulingConstraint.java | 7 +-
.../MinOccupancyFirstSchedulingPolicy.java | 4 +-
.../FreeSlotSchedulingConstraintTest.java | 29 +-----
4 files changed, 77 insertions(+), 70 deletions(-)
diff --git a/runtime/master/src/main/java/edu/snu/nemo/runtime/master/resource/ExecutorRepresenter.java b/runtime/master/src/main/java/edu/snu/nemo/runtime/master/resource/ExecutorRepresenter.java
index e78cb90..6a0a3ee 100644
--- a/runtime/master/src/main/java/edu/snu/nemo/runtime/master/resource/ExecutorRepresenter.java
+++ b/runtime/master/src/main/java/edu/snu/nemo/runtime/master/resource/ExecutorRepresenter.java
@@ -16,6 +16,7 @@
package edu.snu.nemo.runtime.master.resource;
import com.google.protobuf.ByteString;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorSlotComplianceProperty;
import edu.snu.nemo.runtime.common.RuntimeIdGenerator;
import edu.snu.nemo.runtime.common.comm.ControlMessage;
import edu.snu.nemo.runtime.common.message.MessageEnvironment;
@@ -28,6 +29,7 @@ import javax.annotation.concurrent.NotThreadSafe;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* (WARNING) This class is not thread-safe, and thus should only be accessed through ExecutorRegistry.
@@ -45,7 +47,8 @@ public final class ExecutorRepresenter {
private final String executorId;
private final ResourceSpecification resourceSpecification;
- private final Set<Task> runningTasks;
+ private final Map<String, Task> runningComplyingTasks;
+ private final Map<String, Task> runningNonComplyingTasks;
private final Map<Task, Integer> runningTaskToAttempt;
private final Set<Task> completeTasks;
private final Set<Task> failedTasks;
@@ -72,7 +75,8 @@ public final class ExecutorRepresenter {
this.executorId = executorId;
this.resourceSpecification = resourceSpecification;
this.messageSender = messageSender;
- this.runningTasks = new HashSet<>();
+ this.runningComplyingTasks = new HashMap<>();
+ this.runningNonComplyingTasks = new HashMap<>();
this.runningTaskToAttempt = new HashMap<>();
this.completeTasks = new HashSet<>();
this.failedTasks = new HashSet<>();
@@ -85,12 +89,13 @@ public final class ExecutorRepresenter {
* Marks all Tasks which were running in this executor as failed.
*/
public Set<String> onExecutorFailed() {
- failedTasks.addAll(runningTasks);
- final Set<String> snapshot = runningTasks.stream()
- .map(Task::getTaskId)
- .collect(Collectors.toSet());
- runningTasks.clear();
- return snapshot;
+ failedTasks.addAll(runningComplyingTasks.values());
+ failedTasks.addAll(runningNonComplyingTasks.values());
+ final Set<String> taskIds = Stream.concat(runningComplyingTasks.keySet().stream(),
+ runningNonComplyingTasks.keySet().stream()).collect(Collectors.toSet());
+ runningComplyingTasks.clear();
+ runningNonComplyingTasks.clear();
+ return taskIds;
}
/**
@@ -98,25 +103,23 @@ public final class ExecutorRepresenter {
* @param task
*/
public void onTaskScheduled(final Task task) {
- runningTasks.add(task);
+ (task.getPropertyValue(ExecutorSlotComplianceProperty.class).orElse(true)
+ ? runningComplyingTasks : runningNonComplyingTasks).put(task.getTaskId(), task);
runningTaskToAttempt.put(task, task.getAttemptIdx());
failedTasks.remove(task);
- serializationExecutorService.submit(new Runnable() {
- @Override
- public void run() {
- final byte[] serialized = SerializationUtils.serialize(task);
- sendControlMessage(
- ControlMessage.Message.newBuilder()
- .setId(RuntimeIdGenerator.generateMessageId())
- .setListenerId(MessageEnvironment.EXECUTOR_MESSAGE_LISTENER_ID)
- .setType(ControlMessage.MessageType.ScheduleTask)
- .setScheduleTaskMsg(
- ControlMessage.ScheduleTaskMsg.newBuilder()
- .setTask(ByteString.copyFrom(serialized))
- .build())
- .build());
- }
+ serializationExecutorService.submit(() -> {
+ final byte[] serialized = SerializationUtils.serialize(task);
+ sendControlMessage(
+ ControlMessage.Message.newBuilder()
+ .setId(RuntimeIdGenerator.generateMessageId())
+ .setListenerId(MessageEnvironment.EXECUTOR_MESSAGE_LISTENER_ID)
+ .setType(ControlMessage.MessageType.ScheduleTask)
+ .setScheduleTaskMsg(
+ ControlMessage.ScheduleTaskMsg.newBuilder()
+ .setTask(ByteString.copyFrom(serialized))
+ .build())
+ .build());
});
}
@@ -133,11 +136,7 @@ public final class ExecutorRepresenter {
*
*/
public void onTaskExecutionComplete(final String taskId) {
- Task completedTask = runningTasks.stream()
- .filter(task -> task.getTaskId().equals(taskId)).findFirst()
- .orElseThrow(() -> new RuntimeException("Completed task not found in its ExecutorRepresenter"));
-
- runningTasks.remove(completedTask);
+ final Task completedTask = removeFromRunningTasks(taskId);
runningTaskToAttempt.remove(completedTask);
completeTasks.add(completedTask);
}
@@ -147,11 +146,7 @@ public final class ExecutorRepresenter {
* @param taskId id of the Task
*/
public void onTaskExecutionFailed(final String taskId) {
- Task failedTask = runningTasks.stream()
- .filter(task -> task.getTaskId().equals(taskId)).findFirst()
- .orElseThrow(() -> new RuntimeException("Failed task not found in its ExecutorRepresenter"));
-
- runningTasks.remove(failedTask);
+ final Task failedTask = removeFromRunningTasks(taskId);
runningTaskToAttempt.remove(failedTask);
failedTasks.add(failedTask);
}
@@ -167,7 +162,29 @@ public final class ExecutorRepresenter {
* @return the current snapshot of set of Tasks that are running in this executor.
*/
public Set<Task> getRunningTasks() {
- return Collections.unmodifiableSet(new HashSet<>(runningTasks));
+ return Stream.concat(runningComplyingTasks.values().stream(),
+ runningNonComplyingTasks.values().stream()).collect(Collectors.toSet());
+ }
+
+ /**
+ * @return the number of running {@link Task}s.
+ */
+ public int getNumOfRunningTasks() {
+ return getNumOfComplyingRunningTasks() + getNumOfNonComplyingRunningTasks();
+ }
+
+ /**
+ * @return the number of running {@link Task}s that complies to the executor slot restriction.
+ */
+ public int getNumOfComplyingRunningTasks() {
+ return runningComplyingTasks.size();
+ }
+
+ /**
+ * @return the number of running {@link Task}s that does not comply to the executor slot restriction.
+ */
+ public int getNumOfNonComplyingRunningTasks() {
+ return runningNonComplyingTasks.size();
}
/**
@@ -202,10 +219,28 @@ public final class ExecutorRepresenter {
public String toString() {
final StringBuffer sb = new StringBuffer("ExecutorRepresenter{");
sb.append("executorId='").append(executorId).append('\'');
- sb.append(", runningTasks=").append(runningTasks);
+ sb.append(", runningTasks=").append(getRunningTasks());
sb.append(", failedTasks=").append(failedTasks);
sb.append('}');
return sb.toString();
}
+
+ /**
+ * Removes the specified {@link Task} from the map of running tasks.
+ *
+ * @param taskId id of the task to remove
+ * @return the removed {@link Task}
+ */
+ private Task removeFromRunningTasks(final String taskId) {
+ final Task task;
+ if (runningComplyingTasks.containsKey(taskId)) {
+ task = runningComplyingTasks.remove(taskId);
+ } else if (runningNonComplyingTasks.containsKey(taskId)) {
+ task = runningNonComplyingTasks.remove(taskId);
+ } else {
+ throw new RuntimeException(String.format("Task %s not found in its ExecutorRepresenter", taskId));
+ }
+ return task;
+ }
}
diff --git a/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraint.java b/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraint.java
index f25e4bc..c330cd4 100644
--- a/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraint.java
+++ b/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraint.java
@@ -38,11 +38,6 @@ public final class FreeSlotSchedulingConstraint implements SchedulingConstraint
return true;
}
- // Count the number of tasks which are running in this executor and complying the slot constraint.
- final long numOfComplyingTasks = executor.getRunningTasks().stream()
- .filter(runningTask -> runningTask.getPropertyValue(ExecutorSlotComplianceProperty.class)
- .orElseGet(() -> true))
- .count();
- return numOfComplyingTasks < executor.getExecutorCapacity();
+ return executor.getNumOfComplyingRunningTasks() < executor.getExecutorCapacity();
}
}
diff --git a/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/MinOccupancyFirstSchedulingPolicy.java b/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/MinOccupancyFirstSchedulingPolicy.java
index 1261d44..0fa51b5 100644
--- a/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/MinOccupancyFirstSchedulingPolicy.java
+++ b/runtime/master/src/main/java/edu/snu/nemo/runtime/master/scheduler/MinOccupancyFirstSchedulingPolicy.java
@@ -38,7 +38,7 @@ public final class MinOccupancyFirstSchedulingPolicy implements SchedulingPolicy
public ExecutorRepresenter selectExecutor(final Collection<ExecutorRepresenter> executors, final Task task) {
final OptionalInt minOccupancy =
executors.stream()
- .map(executor -> executor.getRunningTasks().size())
+ .map(executor -> executor.getNumOfRunningTasks())
.mapToInt(i -> i).min();
if (!minOccupancy.isPresent()) {
@@ -46,7 +46,7 @@ public final class MinOccupancyFirstSchedulingPolicy implements SchedulingPolicy
}
return executors.stream()
- .filter(executor -> executor.getRunningTasks().size() == minOccupancy.getAsInt())
+ .filter(executor -> executor.getNumOfRunningTasks() == minOccupancy.getAsInt())
.findFirst()
.orElseThrow(() -> new RuntimeException("No such executor"));
}
diff --git a/runtime/master/src/test/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraintTest.java b/runtime/master/src/test/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraintTest.java
index e387c70..871be92 100644
--- a/runtime/master/src/test/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraintTest.java
+++ b/runtime/master/src/test/java/edu/snu/nemo/runtime/master/scheduler/FreeSlotSchedulingConstraintTest.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorSlotCompliancePro
import edu.snu.nemo.runtime.common.plan.Task;
import edu.snu.nemo.runtime.master.resource.ExecutorRepresenter;
import org.apache.reef.tang.Tang;
-import org.apache.reef.tang.exceptions.InjectionException;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -28,7 +27,6 @@ import org.powermock.modules.junit4.PowerMockRunner;
import java.util.*;
import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.*;
@@ -46,42 +44,21 @@ public final class FreeSlotSchedulingConstraintTest {
@Before
public void setUp() throws Exception {
schedulingConstraint = Tang.Factory.getTang().newInjector().getInstance(FreeSlotSchedulingConstraint.class);
- a0 = mockExecutorRepresenter(1, 1, 1);
- a1 = mockExecutorRepresenter(2, 2, 3);
- }
-
- /**
- * Mock a task.
- *
- * @param taskId the ID of the task to mock.
- * @return the mocked task.
- */
- private static Task mockTask(final String taskId) {
- final Task task = mock(Task.class);
- when(task.getTaskId()).thenReturn(taskId);
- return task;
+ a0 = mockExecutorRepresenter(1, 1);
+ a1 = mockExecutorRepresenter(2, 3);
}
/**
* Mock an executor representer.
*
* @param numComplyingTasks the number of already running (mocked) tasks which comply slot constraint in the executor.
- * @param numIgnoringTasks the number of already running (mocked) tasks which ignore slot constraint in the executor.
* @param capacity the capacity of the executor.
* @return the mocked executor.
*/
private static ExecutorRepresenter mockExecutorRepresenter(final int numComplyingTasks,
- final int numIgnoringTasks,
final int capacity) {
final ExecutorRepresenter executorRepresenter = mock(ExecutorRepresenter.class);
- final Set<Task> runningTasks = new HashSet<>();
- IntStream.range(0, numComplyingTasks).forEach(i -> runningTasks.add(mockTask(String.valueOf(i))));
- IntStream.range(0, numIgnoringTasks).forEach(i -> {
- final Task task = mockTask(String.valueOf(numComplyingTasks + i));
- when(task.getPropertyValue(ExecutorSlotComplianceProperty.class)).thenReturn(Optional.of(false));
- runningTasks.add(task);
- });
- when(executorRepresenter.getRunningTasks()).thenReturn(runningTasks);
+ when(executorRepresenter.getNumOfComplyingRunningTasks()).thenReturn(numComplyingTasks);
when(executorRepresenter.getExecutorCapacity()).thenReturn(capacity);
return executorRepresenter;
}