simheft.py 文件源码

def get_schedule(self, simulation):
    """
    Overriden.
    """
    nxgraph = simulation.get_task_graph()
    platform_model = cscheduling.PlatformModel(simulation)
    state = cscheduling.SchedulerState(simulation)

    ordered_tasks = cscheduling.heft_order(nxgraph, platform_model)

    subgraph = networkx.DiGraph()

    # fork context is incompatible with SimGrid static variables
    ctx = multiprocessing.get_context("spawn")
    for task in ordered_tasks:
      _update_subgraph(nxgraph, subgraph, task)
      if cscheduling.try_schedule_boundary_task(task, platform_model, state):
        continue
      current_min = cscheduling.MinSelector()
      for host, timesheet in state.timetable.items():
        if cscheduling.is_master_host(host):
          continue
        current_state = state.copy()
        est = platform_model.est(host, nxgraph.pred[task], current_state)
        eet = platform_model.eet(task, host)
        # 'correct' way
        pos, start, finish = cscheduling.timesheet_insertion(timesheet, est, eet)
        # TODO: try aggressive inserts
        current_state.update(task, host, pos, start, finish)
        with tempfile.NamedTemporaryFile("w", suffix=".dot") as temp_file:
          _serialize_graph(subgraph, temp_file)
          subschedule = _serialize_schedule(current_state.timetable)
          with ctx.Pool(1) as process:
            serialized_state = process.apply(_run_simulation, (simulation.platform_path, temp_file.name, subschedule))
          current_state = _restore_state(simulation, serialized_state)
          current_min.update((current_state.max_time, host.speed, host.name), current_state)
      state = current_min.value
    expected_makespan = max([state["ect"] for state in state.task_states.values()])
    return state.schedule, expected_makespan