You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@nemo.apache.org by GitBox <gi...@apache.org> on 2020/02/27 09:46:39 UTC

[GitHub] [incubator-nemo] wonook opened a new pull request #288: implementing scheduler

wonook opened a new pull request #288: implementing scheduler
URL: https://github.com/apache/incubator-nemo/pull/288
 
 
   JIRA: [NEMO-438: Create a Simulator for Simulating an Execution of an Execution Plan](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-438)
   
   **Major changes:**
   - Implements a SimulationScheduler that schedules a physical plan
   - Implements a SimulatedExecutor that acts like an executor
   
   **Minor changes to note:**
   - Refactors executor's ResourceSpecification to be able to be accessed in the driver as well
   - Enables IRDAG to get the input size of the workload
   - Added Job Duration to the Job Metric.
   
   **Tests for the changes:**
   - SimulationSchedulerTest.java tests for the implementations
   
   **Other comments:**
   - None
   
   

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385644817
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
 
 Review comment:
   The purpose of this class is to somehow 'mirror' the behaviors of batch scheduler with similar code. Does this mean this class should be modified when the batch scheduler is modified? Can you leave a comment about this so we don't forget later when we do edit the batch scheduler?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on issue #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on issue #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#issuecomment-595898957
 
 
   @johnyangk ping :)

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r386952780
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SchedulerUtils.java
 ##########
 @@ -0,0 +1,361 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.collect.Sets;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.dag.DAG;
+import org.apache.nemo.common.exception.UnknownFailureCauseException;
+import org.apache.nemo.common.ir.Readable;
+import org.apache.nemo.common.ir.edge.executionproperty.MessageIdEdgeProperty;
+import org.apache.nemo.common.ir.vertex.executionproperty.IgnoreSchedulingTempDataReceiverProperty;
+import org.apache.nemo.common.ir.vertex.executionproperty.MessageIdVertexProperty;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.plan.*;
+import org.apache.nemo.runtime.common.state.BlockState;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Utlity methods regarding schedulers.
+ */
+public final class SchedulerUtils {
 
 Review comment:
   I'll rename the class

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385640787
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SchedulerUtils.java
 ##########
 @@ -0,0 +1,361 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.collect.Sets;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.dag.DAG;
+import org.apache.nemo.common.exception.UnknownFailureCauseException;
+import org.apache.nemo.common.ir.Readable;
+import org.apache.nemo.common.ir.edge.executionproperty.MessageIdEdgeProperty;
+import org.apache.nemo.common.ir.vertex.executionproperty.IgnoreSchedulingTempDataReceiverProperty;
+import org.apache.nemo.common.ir.vertex.executionproperty.MessageIdVertexProperty;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.plan.*;
+import org.apache.nemo.runtime.common.state.BlockState;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Utlity methods regarding schedulers.
+ */
+public final class SchedulerUtils {
 
 Review comment:
   BatchSchedulerUtils?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385643826
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
+                              final SchedulingConstraintRegistry schedulingConstraintRegistry,
+                              final BlockManagerMaster blockManagerMaster,
+                              @Parameter(JobConf.ExecutorJSONContents.class) final String resourceSpecificationString,
+                              @Parameter(JobConf.ScheduleSerThread.class) final int scheduleSerThread,
+                              @Parameter(JobConf.DAGDirectory.class) final String dagDirectory) {
+    this.planRewriter = planRewriter;
+    this.blockManagerMaster = blockManagerMaster;
+    this.pendingTaskCollectionPointer = PendingTaskCollectionPointer.newInstance();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.schedulingConstraintRegistry = schedulingConstraintRegistry;
+    this.schedulingPolicy = new SimulationSchedulingPolicy();
+    this.resourceSpecificationString = resourceSpecificationString;
+    this.dagDirectory = dagDirectory;
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.serializationExecutorService = Executors.newFixedThreadPool(scheduleSerThread);
+    this.actualMetricStore = MetricStore.getStore();
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(this.metricStore);
+    this.simulatedTaskExecutorMap = new HashMap<>();
+    setUpExecutors();
+  }
+
+  /**
+   * Simulate the launch of executors.
+   */
+  private void setUpExecutors() {
+    final List<Pair<Integer, ResourceSpecification>> resourceSpecs =
+      Util.parseResourceSpecificationString(resourceSpecificationString);
+    // Role of ActiveContextHandler + RuntimeMaster.onExecuterLaunched.
+    final AtomicInteger executorIdGenerator = new AtomicInteger(0);
+    final AtomicInteger resourceRequestCount = new AtomicInteger(0);
+    resourceSpecs.forEach(p -> {
+      for (int i = 0; i < p.left(); i++) {
+        resourceRequestCount.getAndIncrement();
+        final ActiveContext ac = new SimulationEvaluatorActiveContext(executorIdGenerator.getAndIncrement());
+        this.onExecutorAdded(new DefaultExecutorRepresenter(ac.getId(), p.right(),
+          new SimulationMessageSender(ac.getId(), this), ac, serializationExecutorService, ac.getId()));
+      }
+    });
+    this.metricCountDownLatch = new CountDownLatch(resourceRequestCount.get());
+  }
+
+  /**
+   * Reset the instance to its initial state.
+   */
+  public void reset() {
+    this.terminate();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.pendingTaskCollectionPointer.getAndSetNull();
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(metricStore);
+    this.simulatedTaskExecutorMap.clear();
+    setUpExecutors();
+  }
+
+  @VisibleForTesting
+  public PlanStateManager getPlanStateManager() {
+    return planStateManager;
+  }
+
+  /**
+   * The entrance point of the simulator. Simulate a plan by submitting a plan through this method.
+   * @param submittedPhysicalPlan the plan to simulate.
+   * @param maxScheduleAttempt the max number of times this plan/sub-part of the plan should be attempted.
+   */
+  @Override
+  public void schedulePlan(final PhysicalPlan submittedPhysicalPlan, final int maxScheduleAttempt) {
+    // Execute the given plan.
+    LOG.info("Plan to schedule: {}", submittedPhysicalPlan.getPlanId());
+
+    if (!planStateManager.isInitialized()) {
+      // First scheduling.
+      taskDispatcher.run();
+      updatePlan(submittedPhysicalPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("submitted");
+    } else {
+      // Append the submitted plan to the original plan.
+      final PhysicalPlan appendedPlan =
+        PlanAppender.appendPlan(planStateManager.getPhysicalPlan(), submittedPhysicalPlan);
+      updatePlan(appendedPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("appended");
+    }
+
+    doSchedule();
+
+    try {
+      planStateManager.waitUntilFinish();
+    } finally {
+      planStateManager.storeJSON("final");
+    }
+
+    final Long jobDuration = this.simulatedTaskExecutorMap.values().stream()
+      .mapToLong(SimulatedTaskExecutor::getElapsedTime)
+      .max().orElse(0);
+    LOG.info("Simulation of {} is complete with job duration of {}!", submittedPhysicalPlan.getPlanId(), jobDuration);
+    this.metricStore.getOrCreateMetric(JobMetric.class, submittedPhysicalPlan.getPlanId()).setJobDuration(jobDuration);
+    executorRegistry.viewExecutors(executors -> executors.forEach(executor -> metricCountDownLatch.countDown()));
+  }
+
+  /**
+   * The main entry point for task scheduling.
+   * This operation can be invoked at any point during job execution, as it is designed to be free of side-effects.
+   * <p>
+   * These are the reasons why.
+   * - We 'reset' {@link PendingTaskCollectionPointer}, and not 'add' new tasks to it
+   * - We make {@link TaskDispatcher} dispatch only the tasks that are READY.
+   */
+  private void doSchedule() {
+    final java.util.Optional<List<Stage>> earliest =
+      SchedulerUtils.selectEarliestSchedulableGroup(sortedScheduleGroups, planStateManager);
+
+    if (earliest.isPresent()) {
+      final List<Task> tasksToSchedule = earliest.get().stream()
+        .flatMap(stage -> SchedulerUtils.selectSchedulableTasks(planStateManager, blockManagerMaster, stage).stream())
+        .collect(Collectors.toList());
+      if (!tasksToSchedule.isEmpty()) {
+        LOG.info("Scheduling some tasks in {}, which are in the same ScheduleGroup", tasksToSchedule.stream()
+          .map(Task::getTaskId)
+          .map(RuntimeIdManager::getStageIdFromTaskId)
+          .collect(Collectors.toSet()));
+
+        // Set the pointer to the schedulable tasks.
+        pendingTaskCollectionPointer.setToOverwrite(tasksToSchedule);
+
+        // Notify the dispatcher that a new collection is available.
+        taskDispatcher.onNewPendingTaskCollectionAvailable();
+      }
+    } else {
+      LOG.info("Skipping this round as no ScheduleGroup is schedulable.");
+    }
+  }
+
+  @Override
+  public void updatePlan(final PhysicalPlan newPhysicalPlan) {
+    // update the physical plan in the scheduler.
+    // NOTE: what's already been executed is not modified in the new physical plan.
+    // TODO #182: Consider reshaping in run-time optimization. At now, we only consider plan appending.
+    updatePlan(newPhysicalPlan, planStateManager.getMaxScheduleAttempt());
+  }
+
+  /**
+   * Update the physical plan in the scheduler.
+   *
+   * @param newPhysicalPlan    the new physical plan to update.
+   * @param maxScheduleAttempt the maximum number of task scheduling attempt.
+   */
+  private void updatePlan(final PhysicalPlan newPhysicalPlan,
+                          final int maxScheduleAttempt) {
+    planStateManager.updatePlan(newPhysicalPlan, maxScheduleAttempt);
+    this.sortedScheduleGroups = newPhysicalPlan.getStageDAG().getVertices().stream()
+      .collect(Collectors.groupingBy(Stage::getScheduleGroup))
+      .entrySet().stream()
+      .sorted(Map.Entry.comparingByKey())
+      .map(Map.Entry::getValue)
+      .collect(Collectors.toList());
+  }
+
+  @Override
+  public void onExecutorAdded(final ExecutorRepresenter executorRepresenter) {
+    LOG.info("{} added (node: {})", executorRepresenter.getExecutorId(), executorRepresenter.getNodeName());
+    executorRegistry.registerExecutor(executorRepresenter);
+    this.simulatedTaskExecutorMap.put(executorRepresenter.getExecutorId(),
+      new SimulatedTaskExecutor(this, executorRepresenter, actualMetricStore));
+    taskDispatcher.onExecutorSlotAvailable();
+  }
+
+  @Override
+  public void onExecutorRemoved(final String executorId) {
+    // Role of FailedEvaluatorHandler + onExecutorFailed.
+    metricCountDownLatch.countDown();
+    LOG.info("{} removed", executorId);
+
+    // These are tasks that were running at the time of executor removal.
+    final Set<String> interruptedTasks = new HashSet<>();
+    executorRegistry.updateExecutor(executorId, (executor, state) -> {
+      interruptedTasks.addAll(executor.onExecutorFailed());
+      return Pair.of(executor, ExecutorRegistry.ExecutorState.FAILED);
+    });
+
+    // Retry the interrupted tasks (and required parents)
+    SchedulerUtils.retryTasksAndRequiredParents(planStateManager, blockManagerMaster, interruptedTasks);
+
+    // Trigger the scheduling of SHOULD_RETRY tasks in the earliest scheduleGroup
+    doSchedule();
+  }
+
+  /**
+   * Process the RuntimePassMessage.
+   *
+   * @param taskId that generated the message.
+   * @param data   of the message.
+   */
+  public void onRunTimePassMessage(final String taskId, final Object data) {
+    // TODO #436: Dynamic task resizing.
+    SchedulerUtils.onRunTimePassMessage(planStateManager, planRewriter, taskId, data);
+  }
+
+  @Override
+  public synchronized void onTaskStateReportFromExecutor(final String executorId,
+                                                         final String taskId,
+                                                         final int attemptIdx,
+                                                         final TaskState.State newState,
+                                                         @Nullable final String taskPutOnHold,
+                                                         final TaskState.RecoverableTaskFailureCause failureCause) {
+    // Role of MasterControlMessageReceiver + handleControlMessage --> onTaskStateChanged.
+    // Do change state, as this notification is for the current task attempt.
+    planStateManager.onTaskStateChanged(taskId, newState);
+    switch (newState) {
+      case COMPLETE:
+        SchedulerUtils.onTaskExecutionComplete(executorRegistry, executorId, taskId);
+        break;
+      case SHOULD_RETRY:
+        // SHOULD_RETRY from an executor means that the task ran into a recoverable failure
+        SchedulerUtils.onTaskExecutionFailedRecoverable(planStateManager, blockManagerMaster, executorRegistry,
+          executorId, taskId, failureCause);
+        break;
+      case ON_HOLD:
+        final java.util.Optional<PhysicalPlan> optionalPhysicalPlan =
+          SchedulerUtils.onTaskExecutionOnHold(planStateManager, executorRegistry, planRewriter, executorId, taskId);
+        optionalPhysicalPlan.ifPresent(this::updatePlan);
+        break;
+      case FAILED:
+        throw new UnrecoverableFailureException(new Exception(String.format("The plan failed on %s in %s",
+          taskId, executorId)));
+      case READY:
+      case EXECUTING:
+        throw new SimulationException("The states READY/EXECUTING cannot occur at this point");
+      default:
+        throw new UnknownExecutionStateException(new Exception("This TaskState is unknown: " + newState));
+    }
+
+    // Invoke doSchedule()
+    switch (newState) {
+      case COMPLETE:
+      case ON_HOLD:
+        // If the stage has completed
+        final String stageIdForTaskUponCompletion = RuntimeIdManager.getStageIdFromTaskId(taskId);
+        if (planStateManager.getStageState(stageIdForTaskUponCompletion).equals(StageState.State.COMPLETE)
+          && !planStateManager.isPlanDone()) {
+          doSchedule();
+        }
+        break;
+      case SHOULD_RETRY:
+        // Do retry
+        doSchedule();
+        break;
+      default:
+        break;
+    }
+
+    // Invoke taskDispatcher.onExecutorSlotAvailable()
+    switch (newState) {
+      // These three states mean that a slot is made available.
+      case COMPLETE:
+      case ON_HOLD:
+      case SHOULD_RETRY:
+        taskDispatcher.onExecutorSlotAvailable();
+        break;
+      default:
+        break;
+    }
+  }
+
+  /**
+   * Handle metric message, as it would have been sent from org.apache.nemo.runtime.executor.MetricManagerWorker#send.
+   * @param metricType  type of metric.
+   * @param metricId    id of metric.
+   * @param metricField field of metric.
+   * @param metricValue value of metric.
+   */
+  void handleMetricMessage(final String metricType, final String metricId,
+                           final String metricField, final byte[] metricValue) {
+    final Class<Metric> metricClass = metricStore.getMetricClassByName(metricType);
+    // process metric message
+    try {
+      if (metricStore.getOrCreateMetric(metricClass, metricId).processMetricMessage(metricField, metricValue)) {
+        metricStore.triggerBroadcast(metricClass, metricId);
+      }
+    } catch (final Exception e) {
+      LOG.warn("Error when processing metric message for {}, {}, {}.", metricType, metricId, metricField);
+    }
+  }
+
+  /**
+   * The endpoint of the simulator. Collect the metric store, and terminate the simulator.
+   * @return the metrics of the simulation.
+   */
+  public MetricStore collectMetricStore() {
+    try {
+      // wait for metric flush
+      if (!metricCountDownLatch.await(10000, TimeUnit.MILLISECONDS)) {
+        LOG.warn("Terminating master before all executor terminated messages arrived.");
+      }
+    } catch (final InterruptedException e) {
+      LOG.warn("Waiting executor terminating process interrupted: ", e);
+      // clean up state...
+      Thread.currentThread().interrupt();
+    }
+
+    final MetricStore res = this.metricStore;
+    this.reset();
+    return res;
+  }
+
+  @Override
+  public void onSpeculativeExecutionCheck() {
+    // we don't simulate speculate execution yet.
+    return;
+  }
+
+  @Override
+  public void terminate() {
+    this.taskDispatcher.terminate();
+    this.executorRegistry.terminate();
+  }
+
+  /**
+   * Evaluator ActiveContext for the Simulation.
+   */
+  private static final class SimulationEvaluatorActiveContext implements ActiveContext {
 
 Review comment:
   Would it be possible to not introduce this?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r386952862
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulatedTaskExecutor.java
 ##########
 @@ -0,0 +1,254 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.Streams;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.dag.DAG;
+import org.apache.nemo.common.ir.vertex.IRVertex;
+import org.apache.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageEnvironment;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.StateTransitionEvent;
+import org.apache.nemo.runtime.common.metric.TaskMetric;
+import org.apache.nemo.runtime.common.plan.RuntimeEdge;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+/**
+ * Class for simulated task execution.
+ */
+public final class SimulatedTaskExecutor {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulatedTaskExecutor.class.getName());
+  private static final String TASK_METRIC_ID = "TaskMetric";
+
+  /**
+   * the simulation scheduler that the executor is associated with.
+   */
+  private final SimulationScheduler scheduler;
+  private final ExecutorRepresenter executorRepresenter;
+  private Long executorInitializationTime;
+  private final AtomicLong currentTime;
+  private Long timeCheckpoint;
+  private final MetricStore actualMetricStore;
+  private final ConcurrentMap<String, DAG<IRVertex, RuntimeEdge<IRVertex>>> stageIDToStageIRDAG;
+
+  /**
+   * Constructor.
+   * @param scheduler the simulation scheduler that the executor is associated with.
+   */
+  SimulatedTaskExecutor(final SimulationScheduler scheduler,
+                        final ExecutorRepresenter executorRepresenter,
+                        final MetricStore actualMetricStore) {
+    this.scheduler = scheduler;
+    this.executorRepresenter = executorRepresenter;
+    this.actualMetricStore = actualMetricStore;
+    this.stageIDToStageIRDAG = new ConcurrentHashMap<>();
+    this.currentTime = new AtomicLong(-1L);
+    this.executorInitializationTime = -1L;
+  }
+
+  /**
+   * Calculate the expected task duration.
+   * This only works if there exists metrics in the actual MetricStore, that contains information about the stage,
+   * as well as with the parallelism the task is associated with.
+   *
+   * @param task the task to calculate the task duration for.
+   * @return the expected task duration.
+   */
+  private long calculateExpectedTaskDuration(final Task task) {
+    final DAG<IRVertex, RuntimeEdge<IRVertex>> stageIRDAG = stageIDToStageIRDAG.computeIfAbsent(task.getStageId(),
+      i -> SerializationUtils.deserialize(task.getSerializedIRDag()));
+
+    final Map<String, Object> jobMetricMap = this.actualMetricStore.getMetricMap(JobMetric.class);
+    if (jobMetricMap.size() > 1) {
+      LOG.warn("MetricStore has more than one JobMetric. The results could be misleading.");
+    }
+    // Fetch first element.
+    final JobMetric jobMetric = (JobMetric) jobMetricMap.entrySet().iterator().next().getValue();
+    final JsonNode stageDAG = jobMetric.getStageDAG();
+
+    // Gather ID of stages that have the characteristics that the task possesses.
+    final Set<String> stageIdsToGatherMetricsFrom = Streams.stream(() -> stageDAG.get("vertices").iterator())
+      .filter(s -> s.get("properties").get("irDag").get("vertices").size()
+        == stageIRDAG.getVertices().size())  // same # of vertices.
+      .filter(s -> s.get("properties").get("irDag").get("edges").size()
+        == stageIRDAG.getEdges().size())  // same # of edges.
+      .filter(s -> s.get("properties").get("executionProperties")
+        .get("org.apache.nemo.common.ir.vertex.executionproperty.ParallelismProperty").asInt(0)
+        == task.getPropertyValue(ParallelismProperty.class).orElse(0))  // same parallelism.
+      .map(s -> s.get("id").asText())
+      .collect(Collectors.toSet());
+
+    // Derive the average task duration from the stages.
+    final OptionalDouble average = this.actualMetricStore.getMetricMap(TaskMetric.class).entrySet().stream()
+      .filter(e -> stageIdsToGatherMetricsFrom.contains(RuntimeIdManager.getStageIdFromTaskId(e.getKey())))
+      .map(Map.Entry::getValue)  // stream of TaskMetric.
+      .mapToLong(tm -> ((TaskMetric) tm).getTaskDuration())
+      .filter(l -> l > 0)
+      .average();
+
+    // convert to long and save.
+    return (long) (average.orElse(0) + 0.5);  // 0 to indicate something went wrong
+  }
+
+  /**
+   * Handle the task and record metrics, as a real Executor#onTaskReceived would.
+   *
+   * @param task the task to execute.
+   */
+  public void onTaskReceived(final Task task) {
+    if (executorInitializationTime < 0) {
+      executorInitializationTime = System.currentTimeMillis();
+      currentTime.set(executorInitializationTime);
+      timeCheckpoint = executorInitializationTime;
+    }
+    final String taskId = task.getTaskId();
+    final int attemptIdx = task.getAttemptIdx();
+    String idOfVertexPutOnHold = null;
+
+    final long schedulingOverhead = System.currentTimeMillis() - this.timeCheckpoint;
+    this.timeCheckpoint = System.currentTimeMillis();
+    this.sendMetric(TASK_METRIC_ID, taskId, "schedulingOverhead",
+      SerializationUtils.serialize(schedulingOverhead));
+    final long executionStartTime = this.currentTime.getAndAdd(schedulingOverhead);
+
+    // Prepare (constructor of TaskExecutor)
+
+    // Deserialize task
+
+    // Connect incoming / outgoing edges.
+
+    // Execute
+    LOG.debug("{} started", taskId);
+
+    // Fetch external data (Read) and process them
 
 Review comment:
   ✅ 

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk merged pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk merged pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288
 
 
   

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r386952122
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
+                              final SchedulingConstraintRegistry schedulingConstraintRegistry,
+                              final BlockManagerMaster blockManagerMaster,
+                              @Parameter(JobConf.ExecutorJSONContents.class) final String resourceSpecificationString,
+                              @Parameter(JobConf.ScheduleSerThread.class) final int scheduleSerThread,
+                              @Parameter(JobConf.DAGDirectory.class) final String dagDirectory) {
+    this.planRewriter = planRewriter;
+    this.blockManagerMaster = blockManagerMaster;
+    this.pendingTaskCollectionPointer = PendingTaskCollectionPointer.newInstance();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.schedulingConstraintRegistry = schedulingConstraintRegistry;
+    this.schedulingPolicy = new SimulationSchedulingPolicy();
+    this.resourceSpecificationString = resourceSpecificationString;
+    this.dagDirectory = dagDirectory;
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.serializationExecutorService = Executors.newFixedThreadPool(scheduleSerThread);
+    this.actualMetricStore = MetricStore.getStore();
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(this.metricStore);
+    this.simulatedTaskExecutorMap = new HashMap<>();
+    setUpExecutors();
+  }
+
+  /**
+   * Simulate the launch of executors.
+   */
+  private void setUpExecutors() {
+    final List<Pair<Integer, ResourceSpecification>> resourceSpecs =
+      Util.parseResourceSpecificationString(resourceSpecificationString);
+    // Role of ActiveContextHandler + RuntimeMaster.onExecuterLaunched.
+    final AtomicInteger executorIdGenerator = new AtomicInteger(0);
+    final AtomicInteger resourceRequestCount = new AtomicInteger(0);
+    resourceSpecs.forEach(p -> {
+      for (int i = 0; i < p.left(); i++) {
+        resourceRequestCount.getAndIncrement();
+        final ActiveContext ac = new SimulationEvaluatorActiveContext(executorIdGenerator.getAndIncrement());
+        this.onExecutorAdded(new DefaultExecutorRepresenter(ac.getId(), p.right(),
+          new SimulationMessageSender(ac.getId(), this), ac, serializationExecutorService, ac.getId()));
+      }
+    });
+    this.metricCountDownLatch = new CountDownLatch(resourceRequestCount.get());
+  }
+
+  /**
+   * Reset the instance to its initial state.
+   */
+  public void reset() {
+    this.terminate();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.pendingTaskCollectionPointer.getAndSetNull();
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(metricStore);
+    this.simulatedTaskExecutorMap.clear();
+    setUpExecutors();
+  }
+
+  @VisibleForTesting
+  public PlanStateManager getPlanStateManager() {
+    return planStateManager;
+  }
+
+  /**
+   * The entrance point of the simulator. Simulate a plan by submitting a plan through this method.
+   * @param submittedPhysicalPlan the plan to simulate.
+   * @param maxScheduleAttempt the max number of times this plan/sub-part of the plan should be attempted.
+   */
+  @Override
+  public void schedulePlan(final PhysicalPlan submittedPhysicalPlan, final int maxScheduleAttempt) {
+    // Execute the given plan.
+    LOG.info("Plan to schedule: {}", submittedPhysicalPlan.getPlanId());
+
+    if (!planStateManager.isInitialized()) {
+      // First scheduling.
+      taskDispatcher.run();
+      updatePlan(submittedPhysicalPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("submitted");
+    } else {
+      // Append the submitted plan to the original plan.
+      final PhysicalPlan appendedPlan =
+        PlanAppender.appendPlan(planStateManager.getPhysicalPlan(), submittedPhysicalPlan);
+      updatePlan(appendedPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("appended");
+    }
+
+    doSchedule();
+
+    try {
+      planStateManager.waitUntilFinish();
+    } finally {
+      planStateManager.storeJSON("final");
+    }
+
+    final Long jobDuration = this.simulatedTaskExecutorMap.values().stream()
+      .mapToLong(SimulatedTaskExecutor::getElapsedTime)
+      .max().orElse(0);
+    LOG.info("Simulation of {} is complete with job duration of {}!", submittedPhysicalPlan.getPlanId(), jobDuration);
+    this.metricStore.getOrCreateMetric(JobMetric.class, submittedPhysicalPlan.getPlanId()).setJobDuration(jobDuration);
+    executorRegistry.viewExecutors(executors -> executors.forEach(executor -> metricCountDownLatch.countDown()));
+  }
+
+  /**
+   * The main entry point for task scheduling.
+   * This operation can be invoked at any point during job execution, as it is designed to be free of side-effects.
+   * <p>
+   * These are the reasons why.
+   * - We 'reset' {@link PendingTaskCollectionPointer}, and not 'add' new tasks to it
+   * - We make {@link TaskDispatcher} dispatch only the tasks that are READY.
+   */
+  private void doSchedule() {
+    final java.util.Optional<List<Stage>> earliest =
+      SchedulerUtils.selectEarliestSchedulableGroup(sortedScheduleGroups, planStateManager);
+
+    if (earliest.isPresent()) {
+      final List<Task> tasksToSchedule = earliest.get().stream()
+        .flatMap(stage -> SchedulerUtils.selectSchedulableTasks(planStateManager, blockManagerMaster, stage).stream())
+        .collect(Collectors.toList());
+      if (!tasksToSchedule.isEmpty()) {
+        LOG.info("Scheduling some tasks in {}, which are in the same ScheduleGroup", tasksToSchedule.stream()
+          .map(Task::getTaskId)
+          .map(RuntimeIdManager::getStageIdFromTaskId)
+          .collect(Collectors.toSet()));
+
+        // Set the pointer to the schedulable tasks.
+        pendingTaskCollectionPointer.setToOverwrite(tasksToSchedule);
+
+        // Notify the dispatcher that a new collection is available.
+        taskDispatcher.onNewPendingTaskCollectionAvailable();
+      }
+    } else {
+      LOG.info("Skipping this round as no ScheduleGroup is schedulable.");
+    }
+  }
+
+  @Override
+  public void updatePlan(final PhysicalPlan newPhysicalPlan) {
+    // update the physical plan in the scheduler.
+    // NOTE: what's already been executed is not modified in the new physical plan.
+    // TODO #182: Consider reshaping in run-time optimization. At now, we only consider plan appending.
+    updatePlan(newPhysicalPlan, planStateManager.getMaxScheduleAttempt());
+  }
+
+  /**
+   * Update the physical plan in the scheduler.
+   *
+   * @param newPhysicalPlan    the new physical plan to update.
+   * @param maxScheduleAttempt the maximum number of task scheduling attempt.
+   */
+  private void updatePlan(final PhysicalPlan newPhysicalPlan,
+                          final int maxScheduleAttempt) {
+    planStateManager.updatePlan(newPhysicalPlan, maxScheduleAttempt);
+    this.sortedScheduleGroups = newPhysicalPlan.getStageDAG().getVertices().stream()
+      .collect(Collectors.groupingBy(Stage::getScheduleGroup))
+      .entrySet().stream()
+      .sorted(Map.Entry.comparingByKey())
+      .map(Map.Entry::getValue)
+      .collect(Collectors.toList());
+  }
+
+  @Override
+  public void onExecutorAdded(final ExecutorRepresenter executorRepresenter) {
+    LOG.info("{} added (node: {})", executorRepresenter.getExecutorId(), executorRepresenter.getNodeName());
+    executorRegistry.registerExecutor(executorRepresenter);
+    this.simulatedTaskExecutorMap.put(executorRepresenter.getExecutorId(),
+      new SimulatedTaskExecutor(this, executorRepresenter, actualMetricStore));
+    taskDispatcher.onExecutorSlotAvailable();
+  }
+
+  @Override
+  public void onExecutorRemoved(final String executorId) {
+    // Role of FailedEvaluatorHandler + onExecutorFailed.
+    metricCountDownLatch.countDown();
+    LOG.info("{} removed", executorId);
+
+    // These are tasks that were running at the time of executor removal.
+    final Set<String> interruptedTasks = new HashSet<>();
+    executorRegistry.updateExecutor(executorId, (executor, state) -> {
+      interruptedTasks.addAll(executor.onExecutorFailed());
+      return Pair.of(executor, ExecutorRegistry.ExecutorState.FAILED);
+    });
+
+    // Retry the interrupted tasks (and required parents)
+    SchedulerUtils.retryTasksAndRequiredParents(planStateManager, blockManagerMaster, interruptedTasks);
+
 
 Review comment:
   I tried to refactor the common parts to the BatchSchedulerUtil, so it seems like this would be alright.

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on issue #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on issue #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#issuecomment-593903863
 
 
   @johnyangk I've addressed your comments. Thanks! 😄 

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385645460
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
+                              final SchedulingConstraintRegistry schedulingConstraintRegistry,
+                              final BlockManagerMaster blockManagerMaster,
+                              @Parameter(JobConf.ExecutorJSONContents.class) final String resourceSpecificationString,
+                              @Parameter(JobConf.ScheduleSerThread.class) final int scheduleSerThread,
+                              @Parameter(JobConf.DAGDirectory.class) final String dagDirectory) {
+    this.planRewriter = planRewriter;
+    this.blockManagerMaster = blockManagerMaster;
+    this.pendingTaskCollectionPointer = PendingTaskCollectionPointer.newInstance();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.schedulingConstraintRegistry = schedulingConstraintRegistry;
+    this.schedulingPolicy = new SimulationSchedulingPolicy();
+    this.resourceSpecificationString = resourceSpecificationString;
+    this.dagDirectory = dagDirectory;
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.serializationExecutorService = Executors.newFixedThreadPool(scheduleSerThread);
+    this.actualMetricStore = MetricStore.getStore();
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(this.metricStore);
+    this.simulatedTaskExecutorMap = new HashMap<>();
+    setUpExecutors();
+  }
+
+  /**
+   * Simulate the launch of executors.
+   */
+  private void setUpExecutors() {
+    final List<Pair<Integer, ResourceSpecification>> resourceSpecs =
+      Util.parseResourceSpecificationString(resourceSpecificationString);
+    // Role of ActiveContextHandler + RuntimeMaster.onExecuterLaunched.
+    final AtomicInteger executorIdGenerator = new AtomicInteger(0);
+    final AtomicInteger resourceRequestCount = new AtomicInteger(0);
+    resourceSpecs.forEach(p -> {
+      for (int i = 0; i < p.left(); i++) {
+        resourceRequestCount.getAndIncrement();
+        final ActiveContext ac = new SimulationEvaluatorActiveContext(executorIdGenerator.getAndIncrement());
+        this.onExecutorAdded(new DefaultExecutorRepresenter(ac.getId(), p.right(),
+          new SimulationMessageSender(ac.getId(), this), ac, serializationExecutorService, ac.getId()));
+      }
+    });
+    this.metricCountDownLatch = new CountDownLatch(resourceRequestCount.get());
+  }
+
+  /**
+   * Reset the instance to its initial state.
+   */
+  public void reset() {
+    this.terminate();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.pendingTaskCollectionPointer.getAndSetNull();
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(metricStore);
+    this.simulatedTaskExecutorMap.clear();
+    setUpExecutors();
+  }
+
+  @VisibleForTesting
+  public PlanStateManager getPlanStateManager() {
+    return planStateManager;
+  }
+
+  /**
+   * The entrance point of the simulator. Simulate a plan by submitting a plan through this method.
+   * @param submittedPhysicalPlan the plan to simulate.
+   * @param maxScheduleAttempt the max number of times this plan/sub-part of the plan should be attempted.
+   */
+  @Override
+  public void schedulePlan(final PhysicalPlan submittedPhysicalPlan, final int maxScheduleAttempt) {
+    // Execute the given plan.
+    LOG.info("Plan to schedule: {}", submittedPhysicalPlan.getPlanId());
+
+    if (!planStateManager.isInitialized()) {
+      // First scheduling.
+      taskDispatcher.run();
+      updatePlan(submittedPhysicalPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("submitted");
+    } else {
+      // Append the submitted plan to the original plan.
+      final PhysicalPlan appendedPlan =
+        PlanAppender.appendPlan(planStateManager.getPhysicalPlan(), submittedPhysicalPlan);
+      updatePlan(appendedPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("appended");
+    }
+
+    doSchedule();
+
+    try {
+      planStateManager.waitUntilFinish();
+    } finally {
+      planStateManager.storeJSON("final");
+    }
+
+    final Long jobDuration = this.simulatedTaskExecutorMap.values().stream()
+      .mapToLong(SimulatedTaskExecutor::getElapsedTime)
+      .max().orElse(0);
+    LOG.info("Simulation of {} is complete with job duration of {}!", submittedPhysicalPlan.getPlanId(), jobDuration);
+    this.metricStore.getOrCreateMetric(JobMetric.class, submittedPhysicalPlan.getPlanId()).setJobDuration(jobDuration);
+    executorRegistry.viewExecutors(executors -> executors.forEach(executor -> metricCountDownLatch.countDown()));
+  }
+
+  /**
+   * The main entry point for task scheduling.
+   * This operation can be invoked at any point during job execution, as it is designed to be free of side-effects.
+   * <p>
+   * These are the reasons why.
+   * - We 'reset' {@link PendingTaskCollectionPointer}, and not 'add' new tasks to it
+   * - We make {@link TaskDispatcher} dispatch only the tasks that are READY.
+   */
+  private void doSchedule() {
+    final java.util.Optional<List<Stage>> earliest =
+      SchedulerUtils.selectEarliestSchedulableGroup(sortedScheduleGroups, planStateManager);
+
+    if (earliest.isPresent()) {
+      final List<Task> tasksToSchedule = earliest.get().stream()
+        .flatMap(stage -> SchedulerUtils.selectSchedulableTasks(planStateManager, blockManagerMaster, stage).stream())
+        .collect(Collectors.toList());
+      if (!tasksToSchedule.isEmpty()) {
+        LOG.info("Scheduling some tasks in {}, which are in the same ScheduleGroup", tasksToSchedule.stream()
+          .map(Task::getTaskId)
+          .map(RuntimeIdManager::getStageIdFromTaskId)
+          .collect(Collectors.toSet()));
+
+        // Set the pointer to the schedulable tasks.
+        pendingTaskCollectionPointer.setToOverwrite(tasksToSchedule);
+
+        // Notify the dispatcher that a new collection is available.
+        taskDispatcher.onNewPendingTaskCollectionAvailable();
+      }
+    } else {
+      LOG.info("Skipping this round as no ScheduleGroup is schedulable.");
+    }
+  }
+
+  @Override
+  public void updatePlan(final PhysicalPlan newPhysicalPlan) {
+    // update the physical plan in the scheduler.
+    // NOTE: what's already been executed is not modified in the new physical plan.
+    // TODO #182: Consider reshaping in run-time optimization. At now, we only consider plan appending.
+    updatePlan(newPhysicalPlan, planStateManager.getMaxScheduleAttempt());
+  }
+
+  /**
+   * Update the physical plan in the scheduler.
+   *
+   * @param newPhysicalPlan    the new physical plan to update.
+   * @param maxScheduleAttempt the maximum number of task scheduling attempt.
+   */
+  private void updatePlan(final PhysicalPlan newPhysicalPlan,
+                          final int maxScheduleAttempt) {
+    planStateManager.updatePlan(newPhysicalPlan, maxScheduleAttempt);
+    this.sortedScheduleGroups = newPhysicalPlan.getStageDAG().getVertices().stream()
+      .collect(Collectors.groupingBy(Stage::getScheduleGroup))
+      .entrySet().stream()
+      .sorted(Map.Entry.comparingByKey())
+      .map(Map.Entry::getValue)
+      .collect(Collectors.toList());
+  }
+
+  @Override
+  public void onExecutorAdded(final ExecutorRepresenter executorRepresenter) {
+    LOG.info("{} added (node: {})", executorRepresenter.getExecutorId(), executorRepresenter.getNodeName());
+    executorRegistry.registerExecutor(executorRepresenter);
+    this.simulatedTaskExecutorMap.put(executorRepresenter.getExecutorId(),
+      new SimulatedTaskExecutor(this, executorRepresenter, actualMetricStore));
+    taskDispatcher.onExecutorSlotAvailable();
+  }
+
+  @Override
+  public void onExecutorRemoved(final String executorId) {
+    // Role of FailedEvaluatorHandler + onExecutorFailed.
+    metricCountDownLatch.countDown();
+    LOG.info("{} removed", executorId);
+
+    // These are tasks that were running at the time of executor removal.
+    final Set<String> interruptedTasks = new HashSet<>();
+    executorRegistry.updateExecutor(executorId, (executor, state) -> {
+      interruptedTasks.addAll(executor.onExecutorFailed());
+      return Pair.of(executor, ExecutorRegistry.ExecutorState.FAILED);
+    });
+
+    // Retry the interrupted tasks (and required parents)
+    SchedulerUtils.retryTasksAndRequiredParents(planStateManager, blockManagerMaster, interruptedTasks);
+
 
 Review comment:
   I guess if something is added to the place in the batch scheduler corresponding to this line, then the same code should be added to this line?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r386952697
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
+                              final SchedulingConstraintRegistry schedulingConstraintRegistry,
+                              final BlockManagerMaster blockManagerMaster,
+                              @Parameter(JobConf.ExecutorJSONContents.class) final String resourceSpecificationString,
+                              @Parameter(JobConf.ScheduleSerThread.class) final int scheduleSerThread,
+                              @Parameter(JobConf.DAGDirectory.class) final String dagDirectory) {
+    this.planRewriter = planRewriter;
+    this.blockManagerMaster = blockManagerMaster;
+    this.pendingTaskCollectionPointer = PendingTaskCollectionPointer.newInstance();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.schedulingConstraintRegistry = schedulingConstraintRegistry;
+    this.schedulingPolicy = new SimulationSchedulingPolicy();
+    this.resourceSpecificationString = resourceSpecificationString;
+    this.dagDirectory = dagDirectory;
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.serializationExecutorService = Executors.newFixedThreadPool(scheduleSerThread);
+    this.actualMetricStore = MetricStore.getStore();
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(this.metricStore);
+    this.simulatedTaskExecutorMap = new HashMap<>();
+    setUpExecutors();
+  }
+
+  /**
+   * Simulate the launch of executors.
+   */
+  private void setUpExecutors() {
+    final List<Pair<Integer, ResourceSpecification>> resourceSpecs =
+      Util.parseResourceSpecificationString(resourceSpecificationString);
+    // Role of ActiveContextHandler + RuntimeMaster.onExecuterLaunched.
+    final AtomicInteger executorIdGenerator = new AtomicInteger(0);
+    final AtomicInteger resourceRequestCount = new AtomicInteger(0);
+    resourceSpecs.forEach(p -> {
+      for (int i = 0; i < p.left(); i++) {
+        resourceRequestCount.getAndIncrement();
+        final ActiveContext ac = new SimulationEvaluatorActiveContext(executorIdGenerator.getAndIncrement());
+        this.onExecutorAdded(new DefaultExecutorRepresenter(ac.getId(), p.right(),
+          new SimulationMessageSender(ac.getId(), this), ac, serializationExecutorService, ac.getId()));
+      }
+    });
+    this.metricCountDownLatch = new CountDownLatch(resourceRequestCount.get());
+  }
+
+  /**
+   * Reset the instance to its initial state.
+   */
+  public void reset() {
+    this.terminate();
+    this.executorRegistry = ExecutorRegistry.newInstance();
+    this.planStateManager = PlanStateManager.newInstance(dagDirectory);
+    this.pendingTaskCollectionPointer.getAndSetNull();
+    this.taskDispatcher = TaskDispatcher.newInstance(schedulingConstraintRegistry, schedulingPolicy,
+      pendingTaskCollectionPointer, executorRegistry, planStateManager);
+    this.metricStore = MetricStore.newInstance();
+    this.planStateManager.setMetricStore(metricStore);
+    this.simulatedTaskExecutorMap.clear();
+    setUpExecutors();
+  }
+
+  @VisibleForTesting
+  public PlanStateManager getPlanStateManager() {
+    return planStateManager;
+  }
+
+  /**
+   * The entrance point of the simulator. Simulate a plan by submitting a plan through this method.
+   * @param submittedPhysicalPlan the plan to simulate.
+   * @param maxScheduleAttempt the max number of times this plan/sub-part of the plan should be attempted.
+   */
+  @Override
+  public void schedulePlan(final PhysicalPlan submittedPhysicalPlan, final int maxScheduleAttempt) {
+    // Execute the given plan.
+    LOG.info("Plan to schedule: {}", submittedPhysicalPlan.getPlanId());
+
+    if (!planStateManager.isInitialized()) {
+      // First scheduling.
+      taskDispatcher.run();
+      updatePlan(submittedPhysicalPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("submitted");
+    } else {
+      // Append the submitted plan to the original plan.
+      final PhysicalPlan appendedPlan =
+        PlanAppender.appendPlan(planStateManager.getPhysicalPlan(), submittedPhysicalPlan);
+      updatePlan(appendedPlan, maxScheduleAttempt);
+      planStateManager.storeJSON("appended");
+    }
+
+    doSchedule();
+
+    try {
+      planStateManager.waitUntilFinish();
+    } finally {
+      planStateManager.storeJSON("final");
+    }
+
+    final Long jobDuration = this.simulatedTaskExecutorMap.values().stream()
+      .mapToLong(SimulatedTaskExecutor::getElapsedTime)
+      .max().orElse(0);
+    LOG.info("Simulation of {} is complete with job duration of {}!", submittedPhysicalPlan.getPlanId(), jobDuration);
+    this.metricStore.getOrCreateMetric(JobMetric.class, submittedPhysicalPlan.getPlanId()).setJobDuration(jobDuration);
+    executorRegistry.viewExecutors(executors -> executors.forEach(executor -> metricCountDownLatch.countDown()));
+  }
+
+  /**
+   * The main entry point for task scheduling.
+   * This operation can be invoked at any point during job execution, as it is designed to be free of side-effects.
+   * <p>
+   * These are the reasons why.
+   * - We 'reset' {@link PendingTaskCollectionPointer}, and not 'add' new tasks to it
+   * - We make {@link TaskDispatcher} dispatch only the tasks that are READY.
+   */
+  private void doSchedule() {
+    final java.util.Optional<List<Stage>> earliest =
+      SchedulerUtils.selectEarliestSchedulableGroup(sortedScheduleGroups, planStateManager);
+
+    if (earliest.isPresent()) {
+      final List<Task> tasksToSchedule = earliest.get().stream()
+        .flatMap(stage -> SchedulerUtils.selectSchedulableTasks(planStateManager, blockManagerMaster, stage).stream())
+        .collect(Collectors.toList());
+      if (!tasksToSchedule.isEmpty()) {
+        LOG.info("Scheduling some tasks in {}, which are in the same ScheduleGroup", tasksToSchedule.stream()
+          .map(Task::getTaskId)
+          .map(RuntimeIdManager::getStageIdFromTaskId)
+          .collect(Collectors.toSet()));
+
+        // Set the pointer to the schedulable tasks.
+        pendingTaskCollectionPointer.setToOverwrite(tasksToSchedule);
+
+        // Notify the dispatcher that a new collection is available.
+        taskDispatcher.onNewPendingTaskCollectionAvailable();
+      }
+    } else {
+      LOG.info("Skipping this round as no ScheduleGroup is schedulable.");
+    }
+  }
+
+  @Override
+  public void updatePlan(final PhysicalPlan newPhysicalPlan) {
+    // update the physical plan in the scheduler.
+    // NOTE: what's already been executed is not modified in the new physical plan.
+    // TODO #182: Consider reshaping in run-time optimization. At now, we only consider plan appending.
+    updatePlan(newPhysicalPlan, planStateManager.getMaxScheduleAttempt());
+  }
+
+  /**
+   * Update the physical plan in the scheduler.
+   *
+   * @param newPhysicalPlan    the new physical plan to update.
+   * @param maxScheduleAttempt the maximum number of task scheduling attempt.
+   */
+  private void updatePlan(final PhysicalPlan newPhysicalPlan,
+                          final int maxScheduleAttempt) {
+    planStateManager.updatePlan(newPhysicalPlan, maxScheduleAttempt);
+    this.sortedScheduleGroups = newPhysicalPlan.getStageDAG().getVertices().stream()
+      .collect(Collectors.groupingBy(Stage::getScheduleGroup))
+      .entrySet().stream()
+      .sorted(Map.Entry.comparingByKey())
+      .map(Map.Entry::getValue)
+      .collect(Collectors.toList());
+  }
+
+  @Override
+  public void onExecutorAdded(final ExecutorRepresenter executorRepresenter) {
+    LOG.info("{} added (node: {})", executorRepresenter.getExecutorId(), executorRepresenter.getNodeName());
+    executorRegistry.registerExecutor(executorRepresenter);
+    this.simulatedTaskExecutorMap.put(executorRepresenter.getExecutorId(),
+      new SimulatedTaskExecutor(this, executorRepresenter, actualMetricStore));
+    taskDispatcher.onExecutorSlotAvailable();
+  }
+
+  @Override
+  public void onExecutorRemoved(final String executorId) {
+    // Role of FailedEvaluatorHandler + onExecutorFailed.
+    metricCountDownLatch.countDown();
+    LOG.info("{} removed", executorId);
+
+    // These are tasks that were running at the time of executor removal.
+    final Set<String> interruptedTasks = new HashSet<>();
+    executorRegistry.updateExecutor(executorId, (executor, state) -> {
+      interruptedTasks.addAll(executor.onExecutorFailed());
+      return Pair.of(executor, ExecutorRegistry.ExecutorState.FAILED);
+    });
+
+    // Retry the interrupted tasks (and required parents)
+    SchedulerUtils.retryTasksAndRequiredParents(planStateManager, blockManagerMaster, interruptedTasks);
+
+    // Trigger the scheduling of SHOULD_RETRY tasks in the earliest scheduleGroup
+    doSchedule();
+  }
+
+  /**
+   * Process the RuntimePassMessage.
+   *
+   * @param taskId that generated the message.
+   * @param data   of the message.
+   */
+  public void onRunTimePassMessage(final String taskId, final Object data) {
+    // TODO #436: Dynamic task resizing.
+    SchedulerUtils.onRunTimePassMessage(planStateManager, planRewriter, taskId, data);
+  }
+
+  @Override
+  public synchronized void onTaskStateReportFromExecutor(final String executorId,
+                                                         final String taskId,
+                                                         final int attemptIdx,
+                                                         final TaskState.State newState,
+                                                         @Nullable final String taskPutOnHold,
+                                                         final TaskState.RecoverableTaskFailureCause failureCause) {
+    // Role of MasterControlMessageReceiver + handleControlMessage --> onTaskStateChanged.
+    // Do change state, as this notification is for the current task attempt.
+    planStateManager.onTaskStateChanged(taskId, newState);
+    switch (newState) {
+      case COMPLETE:
+        SchedulerUtils.onTaskExecutionComplete(executorRegistry, executorId, taskId);
+        break;
+      case SHOULD_RETRY:
+        // SHOULD_RETRY from an executor means that the task ran into a recoverable failure
+        SchedulerUtils.onTaskExecutionFailedRecoverable(planStateManager, blockManagerMaster, executorRegistry,
+          executorId, taskId, failureCause);
+        break;
+      case ON_HOLD:
+        final java.util.Optional<PhysicalPlan> optionalPhysicalPlan =
+          SchedulerUtils.onTaskExecutionOnHold(planStateManager, executorRegistry, planRewriter, executorId, taskId);
+        optionalPhysicalPlan.ifPresent(this::updatePlan);
+        break;
+      case FAILED:
+        throw new UnrecoverableFailureException(new Exception(String.format("The plan failed on %s in %s",
+          taskId, executorId)));
+      case READY:
+      case EXECUTING:
+        throw new SimulationException("The states READY/EXECUTING cannot occur at this point");
+      default:
+        throw new UnknownExecutionStateException(new Exception("This TaskState is unknown: " + newState));
+    }
+
+    // Invoke doSchedule()
+    switch (newState) {
+      case COMPLETE:
+      case ON_HOLD:
+        // If the stage has completed
+        final String stageIdForTaskUponCompletion = RuntimeIdManager.getStageIdFromTaskId(taskId);
+        if (planStateManager.getStageState(stageIdForTaskUponCompletion).equals(StageState.State.COMPLETE)
+          && !planStateManager.isPlanDone()) {
+          doSchedule();
+        }
+        break;
+      case SHOULD_RETRY:
+        // Do retry
+        doSchedule();
+        break;
+      default:
+        break;
+    }
+
+    // Invoke taskDispatcher.onExecutorSlotAvailable()
+    switch (newState) {
+      // These three states mean that a slot is made available.
+      case COMPLETE:
+      case ON_HOLD:
+      case SHOULD_RETRY:
+        taskDispatcher.onExecutorSlotAvailable();
+        break;
+      default:
+        break;
+    }
+  }
+
+  /**
+   * Handle metric message, as it would have been sent from org.apache.nemo.runtime.executor.MetricManagerWorker#send.
+   * @param metricType  type of metric.
+   * @param metricId    id of metric.
+   * @param metricField field of metric.
+   * @param metricValue value of metric.
+   */
+  void handleMetricMessage(final String metricType, final String metricId,
+                           final String metricField, final byte[] metricValue) {
+    final Class<Metric> metricClass = metricStore.getMetricClassByName(metricType);
+    // process metric message
+    try {
+      if (metricStore.getOrCreateMetric(metricClass, metricId).processMetricMessage(metricField, metricValue)) {
+        metricStore.triggerBroadcast(metricClass, metricId);
+      }
+    } catch (final Exception e) {
+      LOG.warn("Error when processing metric message for {}, {}, {}.", metricType, metricId, metricField);
+    }
+  }
+
+  /**
+   * The endpoint of the simulator. Collect the metric store, and terminate the simulator.
+   * @return the metrics of the simulation.
+   */
+  public MetricStore collectMetricStore() {
+    try {
+      // wait for metric flush
+      if (!metricCountDownLatch.await(10000, TimeUnit.MILLISECONDS)) {
+        LOG.warn("Terminating master before all executor terminated messages arrived.");
+      }
+    } catch (final InterruptedException e) {
+      LOG.warn("Waiting executor terminating process interrupted: ", e);
+      // clean up state...
+      Thread.currentThread().interrupt();
+    }
+
+    final MetricStore res = this.metricStore;
+    this.reset();
+    return res;
+  }
+
+  @Override
+  public void onSpeculativeExecutionCheck() {
+    // we don't simulate speculate execution yet.
+    return;
+  }
+
+  @Override
+  public void terminate() {
+    this.taskDispatcher.terminate();
+    this.executorRegistry.terminate();
+  }
+
+  /**
+   * Evaluator ActiveContext for the Simulation.
+   */
+  private static final class SimulationEvaluatorActiveContext implements ActiveContext {
 
 Review comment:
   I'm afraid that the active context is required to create executor representers that simulates executors

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385642020
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
 
 Review comment:
   Are all these variables used?
   Maybe briefly comment on how they are used?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r386957643
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
 
 Review comment:
   I'll add comments on each of the components.

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385644817
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
 
 Review comment:
   The purpose of this class seems to somehow 'mirror' the behaviors of batch scheduler with similar code. Does this mean this class should be modified when the batch scheduler is modified? Can you leave a comment about this so we don't forget later when we make changes to the batch scheduler?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on issue #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on issue #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#issuecomment-592455936
 
 
   @johnyangk Can I ask for a review? Thanks!!

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk merged pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk merged pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288
 
 
   

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

johnyangk commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r385639413
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulatedTaskExecutor.java
 ##########
 @@ -0,0 +1,254 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.Streams;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.dag.DAG;
+import org.apache.nemo.common.ir.vertex.IRVertex;
+import org.apache.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageEnvironment;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.StateTransitionEvent;
+import org.apache.nemo.runtime.common.metric.TaskMetric;
+import org.apache.nemo.runtime.common.plan.RuntimeEdge;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+/**
+ * Class for simulated task execution.
+ */
+public final class SimulatedTaskExecutor {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulatedTaskExecutor.class.getName());
+  private static final String TASK_METRIC_ID = "TaskMetric";
+
+  /**
+   * the simulation scheduler that the executor is associated with.
+   */
+  private final SimulationScheduler scheduler;
+  private final ExecutorRepresenter executorRepresenter;
+  private Long executorInitializationTime;
+  private final AtomicLong currentTime;
+  private Long timeCheckpoint;
+  private final MetricStore actualMetricStore;
+  private final ConcurrentMap<String, DAG<IRVertex, RuntimeEdge<IRVertex>>> stageIDToStageIRDAG;
+
+  /**
+   * Constructor.
+   * @param scheduler the simulation scheduler that the executor is associated with.
+   */
+  SimulatedTaskExecutor(final SimulationScheduler scheduler,
+                        final ExecutorRepresenter executorRepresenter,
+                        final MetricStore actualMetricStore) {
+    this.scheduler = scheduler;
+    this.executorRepresenter = executorRepresenter;
+    this.actualMetricStore = actualMetricStore;
+    this.stageIDToStageIRDAG = new ConcurrentHashMap<>();
+    this.currentTime = new AtomicLong(-1L);
+    this.executorInitializationTime = -1L;
+  }
+
+  /**
+   * Calculate the expected task duration.
+   * This only works if there exists metrics in the actual MetricStore, that contains information about the stage,
+   * as well as with the parallelism the task is associated with.
+   *
+   * @param task the task to calculate the task duration for.
+   * @return the expected task duration.
+   */
+  private long calculateExpectedTaskDuration(final Task task) {
+    final DAG<IRVertex, RuntimeEdge<IRVertex>> stageIRDAG = stageIDToStageIRDAG.computeIfAbsent(task.getStageId(),
+      i -> SerializationUtils.deserialize(task.getSerializedIRDag()));
+
+    final Map<String, Object> jobMetricMap = this.actualMetricStore.getMetricMap(JobMetric.class);
+    if (jobMetricMap.size() > 1) {
+      LOG.warn("MetricStore has more than one JobMetric. The results could be misleading.");
+    }
+    // Fetch first element.
+    final JobMetric jobMetric = (JobMetric) jobMetricMap.entrySet().iterator().next().getValue();
+    final JsonNode stageDAG = jobMetric.getStageDAG();
+
+    // Gather ID of stages that have the characteristics that the task possesses.
+    final Set<String> stageIdsToGatherMetricsFrom = Streams.stream(() -> stageDAG.get("vertices").iterator())
+      .filter(s -> s.get("properties").get("irDag").get("vertices").size()
+        == stageIRDAG.getVertices().size())  // same # of vertices.
+      .filter(s -> s.get("properties").get("irDag").get("edges").size()
+        == stageIRDAG.getEdges().size())  // same # of edges.
+      .filter(s -> s.get("properties").get("executionProperties")
+        .get("org.apache.nemo.common.ir.vertex.executionproperty.ParallelismProperty").asInt(0)
+        == task.getPropertyValue(ParallelismProperty.class).orElse(0))  // same parallelism.
+      .map(s -> s.get("id").asText())
+      .collect(Collectors.toSet());
+
+    // Derive the average task duration from the stages.
+    final OptionalDouble average = this.actualMetricStore.getMetricMap(TaskMetric.class).entrySet().stream()
+      .filter(e -> stageIdsToGatherMetricsFrom.contains(RuntimeIdManager.getStageIdFromTaskId(e.getKey())))
+      .map(Map.Entry::getValue)  // stream of TaskMetric.
+      .mapToLong(tm -> ((TaskMetric) tm).getTaskDuration())
+      .filter(l -> l > 0)
+      .average();
+
+    // convert to long and save.
+    return (long) (average.orElse(0) + 0.5);  // 0 to indicate something went wrong
+  }
+
+  /**
+   * Handle the task and record metrics, as a real Executor#onTaskReceived would.
+   *
+   * @param task the task to execute.
+   */
+  public void onTaskReceived(final Task task) {
+    if (executorInitializationTime < 0) {
+      executorInitializationTime = System.currentTimeMillis();
+      currentTime.set(executorInitializationTime);
+      timeCheckpoint = executorInitializationTime;
+    }
+    final String taskId = task.getTaskId();
+    final int attemptIdx = task.getAttemptIdx();
+    String idOfVertexPutOnHold = null;
+
+    final long schedulingOverhead = System.currentTimeMillis() - this.timeCheckpoint;
+    this.timeCheckpoint = System.currentTimeMillis();
+    this.sendMetric(TASK_METRIC_ID, taskId, "schedulingOverhead",
+      SerializationUtils.serialize(schedulingOverhead));
+    final long executionStartTime = this.currentTime.getAndAdd(schedulingOverhead);
+
+    // Prepare (constructor of TaskExecutor)
+
+    // Deserialize task
+
+    // Connect incoming / outgoing edges.
+
+    // Execute
+    LOG.debug("{} started", taskId);
+
+    // Fetch external data (Read) and process them
 
 Review comment:
   Remove comments?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

[GitHub] [incubator-nemo] wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan

Posted by GitBox <gi...@apache.org>.

wonook commented on a change in pull request #288: [NEMO-438] Create a Simulator for Simulating an Execution of an Execution Plan
URL: https://github.com/apache/incubator-nemo/pull/288#discussion_r386952252
 
 

 ##########
 File path: runtime/master/src/main/java/org/apache/nemo/runtime/master/scheduler/SimulationScheduler.java
 ##########
 @@ -0,0 +1,579 @@
+/*
+ * 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.nemo.runtime.master.scheduler;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.SerializationUtils;
+import org.apache.nemo.common.Pair;
+import org.apache.nemo.common.Util;
+import org.apache.nemo.common.exception.IllegalMessageException;
+import org.apache.nemo.common.exception.SimulationException;
+import org.apache.nemo.common.exception.UnknownExecutionStateException;
+import org.apache.nemo.common.exception.UnrecoverableFailureException;
+import org.apache.nemo.common.ir.executionproperty.ResourceSpecification;
+import org.apache.nemo.conf.JobConf;
+import org.apache.nemo.runtime.common.RuntimeIdManager;
+import org.apache.nemo.runtime.common.comm.ControlMessage;
+import org.apache.nemo.runtime.common.message.MessageSender;
+import org.apache.nemo.runtime.common.message.MessageUtils;
+import org.apache.nemo.runtime.common.metric.JobMetric;
+import org.apache.nemo.runtime.common.metric.Metric;
+import org.apache.nemo.runtime.common.plan.PhysicalPlan;
+import org.apache.nemo.runtime.common.plan.PlanRewriter;
+import org.apache.nemo.runtime.common.plan.Stage;
+import org.apache.nemo.runtime.common.plan.Task;
+import org.apache.nemo.runtime.common.state.StageState;
+import org.apache.nemo.runtime.common.state.TaskState;
+import org.apache.nemo.runtime.master.BlockManagerMaster;
+import org.apache.nemo.runtime.master.PlanAppender;
+import org.apache.nemo.runtime.master.PlanStateManager;
+import org.apache.nemo.runtime.master.metric.MetricStore;
+import org.apache.nemo.runtime.master.resource.DefaultExecutorRepresenter;
+import org.apache.nemo.runtime.master.resource.ExecutorRepresenter;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.inject.Inject;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Scheduler for simulating an execution, not controlled by the runtime master.
+ */
+@DriverSide
+@NotThreadSafe
+public final class SimulationScheduler implements Scheduler {
+  private static final Logger LOG = LoggerFactory.getLogger(SimulationScheduler.class.getName());
+
+  /**
+   * Run-time optimizations.
+   */
+  private final PlanRewriter planRewriter;
+
+  /**
+   * Components related to scheduling the given plan.
+   */
+  private TaskDispatcher taskDispatcher;
+  private final PendingTaskCollectionPointer pendingTaskCollectionPointer;
+  private ExecutorRegistry executorRegistry;
+  private PlanStateManager planStateManager;
+  private final ExecutorService serializationExecutorService; // Executor service for scheduling message serialization.
+  private final BlockManagerMaster blockManagerMaster;
+  private final MetricStore actualMetricStore;
+  private MetricStore metricStore;
+  private CountDownLatch metricCountDownLatch;
+
+  private final SchedulingConstraintRegistry schedulingConstraintRegistry;
+  private final SchedulingPolicy schedulingPolicy;
+  private final String resourceSpecificationString;
+  private final String dagDirectory;
+
+  private final Map<String, SimulatedTaskExecutor> simulatedTaskExecutorMap;
+
+  /**
+   * The below variables depend on the submitted plan to execute.
+   */
+  private List<List<Stage>> sortedScheduleGroups;
+
+  @Inject
+  private SimulationScheduler(final PlanRewriter planRewriter,
 
 Review comment:
   I'll leave comments accordingly.

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services