mirror of
https://github.com/wassname/ray.git
synced 2026-07-07 15:12:46 +08:00
Embedded task trace with object dependencies. (#818)
* Embedded timeline * Yeah * Fixed arrows not showing up. * Fixed arrows not showing up, and added check boxes for the kinds of dependencies that should be included in the trace. * first * Fixes * Fixed typo in comments, added more comments. fixed linting. * Added more comments. * Formatting. * fixes * Fixed state.py linting. * Fixed ui.py linting errors. * Fixed linting errors. * Renamed task dependencies and included instructions for viewing arrows. * Fixed according to PR comments. * Fixed bug. * Undid changes to metadata blocks. * Fixes according to comments. * Fixed linting. * Fixed linting. * NOQA keyword added to link line.
This commit is contained in:
alanamarzoev
committed by
Robert Nishihara
Robert Nishihara
parent
fc885bd918
commit
bfe473fa8c
@@ -61,6 +61,13 @@
|
||||
"#### Task trace timeline."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"To view arrows, go to View Options and select Flow Events."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
|
||||
@@ -2,6 +2,7 @@ from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
|
||||
import copy
|
||||
import heapq
|
||||
import json
|
||||
import pickle
|
||||
@@ -525,7 +526,13 @@ class GlobalState(object):
|
||||
|
||||
return task_info
|
||||
|
||||
def dump_catapult_trace(self, path, task_info, breakdowns=False):
|
||||
def dump_catapult_trace(self,
|
||||
path,
|
||||
task_info,
|
||||
breakdowns=True,
|
||||
task_dep=True,
|
||||
obj_dep=True):
|
||||
|
||||
"""Dump task profiling information to a file.
|
||||
|
||||
This information can be viewed as a timeline of profiling information
|
||||
@@ -534,9 +541,14 @@ class GlobalState(object):
|
||||
|
||||
Args:
|
||||
path: The filepath to dump the profiling information to.
|
||||
task_info: The task info to use to generate the trace.
|
||||
task_info: The task info to use to generate the trace. Should be
|
||||
the output of ray.global_state.task_profiles().
|
||||
breakdowns: Boolean indicating whether to break down the tasks into
|
||||
more fine-grained segments.
|
||||
more fine-grained segments.
|
||||
task_dep: Boolean indicating whether or not task submission edges
|
||||
should be included in the trace.
|
||||
obj_dep: Boolean indicating whether or not object dependency edges
|
||||
should be included in the trace.
|
||||
"""
|
||||
|
||||
workers = self.workers()
|
||||
@@ -552,80 +564,228 @@ class GlobalState(object):
|
||||
def micros_rel(ts):
|
||||
return micros(ts - start_time)
|
||||
|
||||
task_profiles = self.task_profiles(start=0, end=time.time())
|
||||
task_table = self.task_table()
|
||||
seen_obj = {}
|
||||
|
||||
full_trace = []
|
||||
for task_id, info in task_info.items():
|
||||
delta_info = dict()
|
||||
delta_info["task_id"] = task_id
|
||||
delta_info["get_arguments"] = (info["get_arguments_end"] -
|
||||
# total_info is what is displayed when selecting a task in the
|
||||
# timeline.
|
||||
total_info = dict()
|
||||
total_info["task_id"] = task_id
|
||||
total_info["get_arguments"] = (info["get_arguments_end"] -
|
||||
info["get_arguments_start"])
|
||||
delta_info["execute"] = (info["execute_end"] -
|
||||
total_info["execute"] = (info["execute_end"] -
|
||||
info["execute_start"])
|
||||
delta_info["store_outputs"] = (info["store_outputs_end"] -
|
||||
total_info["store_outputs"] = (info["store_outputs_end"] -
|
||||
info["store_outputs_start"])
|
||||
delta_info["function_name"] = info["function_name"]
|
||||
delta_info["worker_id"] = info["worker_id"]
|
||||
total_info["function_name"] = info["function_name"]
|
||||
total_info["worker_id"] = info["worker_id"]
|
||||
worker = workers[info["worker_id"]]
|
||||
task_t_info = task_table[task_id]
|
||||
task_spec = task_table[task_id]["TaskSpec"]
|
||||
task_spec["Args"] = [oid.hex() if isinstance(oid,
|
||||
ray.local_scheduler.ObjectID) else oid
|
||||
for oid in task_t_info["TaskSpec"]["Args"]]
|
||||
task_spec["ReturnObjectIDs"] = [oid.hex() for oid in
|
||||
(task_t_info["TaskSpec"]
|
||||
["ReturnObjectIDs"])]
|
||||
task_spec["LocalSchedulerID"] = task_t_info["LocalSchedulerID"]
|
||||
total_info = copy.copy(task_spec)
|
||||
|
||||
parent_info = task_info.get(
|
||||
task_table[task_id]["TaskSpec"]["ParentTaskID"])
|
||||
worker = workers[info["worker_id"]]
|
||||
# The catapult trace format documentation can be found here:
|
||||
# https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU/preview # NOQA
|
||||
if breakdowns:
|
||||
if "get_arguments_end" in info:
|
||||
get_args_trace = {
|
||||
"cat": "get_arguments",
|
||||
"pid": "Node " + str(worker["node_ip_address"]),
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"id": str(worker),
|
||||
"id": task_id,
|
||||
"ts": micros_rel(info["get_arguments_start"]),
|
||||
"ph": "X",
|
||||
"name": info["function_name"] + ":get_arguments",
|
||||
"args": delta_info,
|
||||
"args": total_info,
|
||||
"dur": micros(info["get_arguments_end"] -
|
||||
info["get_arguments_start"])
|
||||
info["get_arguments_start"]),
|
||||
"cname": "rail_idle"
|
||||
}
|
||||
full_trace.append(get_args_trace)
|
||||
|
||||
if "store_outputs_end" in info:
|
||||
outputs_trace = {
|
||||
"cat": "store_outputs",
|
||||
"pid": "Node " + str(worker["node_ip_address"]),
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"id": str(worker),
|
||||
"id": task_id,
|
||||
"ts": micros_rel(info["store_outputs_start"]),
|
||||
"ph": "X",
|
||||
"name": info["function_name"] + ":store_outputs",
|
||||
"args": delta_info,
|
||||
"args": total_info,
|
||||
"dur": micros(info["store_outputs_end"] -
|
||||
info["store_outputs_start"])
|
||||
info["store_outputs_start"]),
|
||||
"cname": "thread_state_runnable"
|
||||
}
|
||||
full_trace.append(outputs_trace)
|
||||
|
||||
if "execute_end" in info:
|
||||
execute_trace = {
|
||||
"cat": "execute",
|
||||
"pid": "Node " + str(worker["node_ip_address"]),
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"id": str(worker),
|
||||
"id": task_id,
|
||||
"ts": micros_rel(info["execute_start"]),
|
||||
"ph": "X",
|
||||
"name": info["function_name"] + ":execute",
|
||||
"args": delta_info,
|
||||
"args": total_info,
|
||||
"dur": micros(info["execute_end"] -
|
||||
info["execute_start"])
|
||||
info["execute_start"]),
|
||||
"cname": "rail_animation"
|
||||
}
|
||||
full_trace.append(execute_trace)
|
||||
|
||||
else:
|
||||
if parent_info:
|
||||
parent_worker = workers[parent_info["worker_id"]]
|
||||
parent_times = self._get_times(parent_info)
|
||||
parent = {
|
||||
"cat": "submit_task",
|
||||
"pid": "Node " + parent_worker["node_ip_address"],
|
||||
"tid": parent_info["worker_id"],
|
||||
"ts": micros_rel(task_profiles[task_table[task_id]
|
||||
["TaskSpec"]
|
||||
["ParentTaskID"]]
|
||||
["get_arguments_start"]),
|
||||
"ph": "s",
|
||||
"name": "SubmitTask",
|
||||
"args": {},
|
||||
"id": (parent_info["worker_id"] +
|
||||
str(micros(min(parent_times))))
|
||||
}
|
||||
full_trace.append(parent)
|
||||
|
||||
task_trace = {
|
||||
"cat": "submit_task",
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"ts": micros_rel(info["get_arguments_start"]),
|
||||
"ph": "f",
|
||||
"name": "SubmitTask",
|
||||
"args": {},
|
||||
"id": (info["worker_id"] +
|
||||
str(micros(min(parent_times)))),
|
||||
"bp": "e",
|
||||
"cname": "olive"
|
||||
}
|
||||
full_trace.append(task_trace)
|
||||
|
||||
task = {
|
||||
"cat": "task",
|
||||
"pid": "Node " + str(worker["node_ip_address"]),
|
||||
"tid": info["worker_id"],
|
||||
"id": str(worker),
|
||||
"ts": micros_rel(info["get_arguments_start"]),
|
||||
"ph": "X",
|
||||
"name": info["function_name"],
|
||||
"args": delta_info,
|
||||
"dur": micros(info["store_outputs_end"] -
|
||||
info["get_arguments_start"])
|
||||
"cat": "task",
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"id": task_id,
|
||||
"ts": micros_rel(info["get_arguments_start"]),
|
||||
"ph": "X",
|
||||
"name": info["function_name"],
|
||||
"args": total_info,
|
||||
"dur": micros(info["store_outputs_end"] -
|
||||
info["get_arguments_start"]),
|
||||
"cname": "thread_state_runnable"
|
||||
}
|
||||
full_trace.append(task)
|
||||
|
||||
print("dumping {}/{}".format(len(full_trace), len(task_info)))
|
||||
if task_dep:
|
||||
if parent_info:
|
||||
parent_worker = workers[parent_info["worker_id"]]
|
||||
parent_times = self._get_times(parent_info)
|
||||
parent = {
|
||||
"cat": "submit_task",
|
||||
"pid": "Node " + parent_worker["node_ip_address"],
|
||||
"tid": parent_info["worker_id"],
|
||||
"ts": micros_rel(task_profiles[task_table[task_id]
|
||||
["TaskSpec"]
|
||||
["ParentTaskID"]]
|
||||
["get_arguments_start"]),
|
||||
"ph": "s",
|
||||
"name": "SubmitTask",
|
||||
"args": {},
|
||||
"id": (parent_info["worker_id"] +
|
||||
str(micros(min(parent_times))))
|
||||
}
|
||||
full_trace.append(parent)
|
||||
|
||||
task_trace = {
|
||||
"cat": "submit_task",
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"ts": micros_rel(info["get_arguments_start"]),
|
||||
"ph": "f",
|
||||
"name": "SubmitTask",
|
||||
"args": {},
|
||||
"id": (info["worker_id"] +
|
||||
str(micros(min(parent_times)))),
|
||||
"bp": "e"
|
||||
}
|
||||
full_trace.append(task_trace)
|
||||
|
||||
if obj_dep:
|
||||
args = task_table[task_id]["TaskSpec"]["Args"]
|
||||
for arg in args:
|
||||
if isinstance(arg, ray.local_scheduler.ObjectID):
|
||||
continue
|
||||
object_info = self._object_table(arg)
|
||||
if object_info["IsPut"]:
|
||||
continue
|
||||
if arg not in seen_obj:
|
||||
seen_obj[arg] = 0
|
||||
seen_obj[arg] += 1
|
||||
owner_task = self._object_table(arg)["TaskID"]
|
||||
owner_worker = (workers[task_profiles
|
||||
[owner_task]["worker_id"]])
|
||||
# Adding/subtracting 2 to the time associated with the
|
||||
# beginning/ending of the flow event is necessary to
|
||||
# make the flow events show up reliably. When these times
|
||||
# are exact, this is presumably an edge case, and catapult
|
||||
# doesn't recognize that there is a duration event at that
|
||||
# exact point in time that the flow event should be bound
|
||||
# to. This issue is solved by adding the 2 ms to the
|
||||
# start/end time of the flow event, which guarantees
|
||||
# overlap with the duration event that it's associated
|
||||
# with, and the flow event therefore always gets drawn.
|
||||
owner = {
|
||||
"cat": "obj_dependency",
|
||||
"pid": "Node " + owner_worker["node_ip_address"],
|
||||
"tid": task_profiles[owner_task]["worker_id"],
|
||||
"ts": micros_rel(task_profiles[owner_task]
|
||||
["store_outputs_end"]) - 2,
|
||||
"ph": "s",
|
||||
"name": "ObjectDependency",
|
||||
"args": {},
|
||||
"bp": "e",
|
||||
"cname": "cq_build_attempt_failed",
|
||||
"id": "obj" + str(arg) + str(seen_obj[arg])
|
||||
}
|
||||
full_trace.append(owner)
|
||||
|
||||
dependent = {
|
||||
"cat": "obj_dependency",
|
||||
"pid": "Node " + worker["node_ip_address"],
|
||||
"tid": info["worker_id"],
|
||||
"ts": micros_rel(info["get_arguments_start"]) + 2,
|
||||
"ph": "f",
|
||||
"name": "ObjectDependency",
|
||||
"args": {},
|
||||
"cname": "cq_build_attempt_failed",
|
||||
"bp": "e",
|
||||
"id": "obj" + str(arg) + str(seen_obj[arg])
|
||||
}
|
||||
full_trace.append(dependent)
|
||||
|
||||
print("Creating JSON {}/{}".format(len(full_trace), len(task_info)))
|
||||
with open(path, "w") as outfile:
|
||||
json.dump(full_trace, outfile)
|
||||
|
||||
|
||||
+158
-76
@@ -3,11 +3,12 @@ import numpy as np
|
||||
import os
|
||||
import pprint
|
||||
import ray
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
from IPython.display import display
|
||||
from IPython.display import display, IFrame, clear_output
|
||||
|
||||
# Instances of this class maintains keep track of whether or not a
|
||||
# callback is currently executing. Since the execution of the callback
|
||||
@@ -101,8 +102,8 @@ def get_sliders(update):
|
||||
if breakdown_opt.value == total_tasks_value:
|
||||
num_tasks_box.value = -min(10000, num_tasks)
|
||||
range_slider.value = (int(100 -
|
||||
(100. * -num_tasks_box.value)
|
||||
/ num_tasks), 100)
|
||||
(100. * -num_tasks_box.value) /
|
||||
num_tasks), 100)
|
||||
else:
|
||||
low, high = map(lambda x: x / 100., range_slider.value)
|
||||
start_box.value = round(diff * low, 2)
|
||||
@@ -115,8 +116,8 @@ def get_sliders(update):
|
||||
elif start_box.value < 0:
|
||||
start_box.value = 0
|
||||
low, high = range_slider.value
|
||||
range_slider.value = (int((start_box.value * 100.)
|
||||
/ diff), high)
|
||||
range_slider.value = (int((start_box.value * 100.) /
|
||||
diff), high)
|
||||
|
||||
# Event was triggered by a change in the end_box value.
|
||||
elif event["owner"] == end_box:
|
||||
@@ -125,8 +126,8 @@ def get_sliders(update):
|
||||
elif end_box.value > diff:
|
||||
end_box.value = diff
|
||||
low, high = range_slider.value
|
||||
range_slider.value = (low, int((end_box.value * 100.)
|
||||
/ diff))
|
||||
range_slider.value = (low, int((end_box.value * 100.) /
|
||||
diff))
|
||||
|
||||
# Event was triggered by a change in the breakdown options
|
||||
# toggle.
|
||||
@@ -137,16 +138,16 @@ def get_sliders(update):
|
||||
num_tasks_box.disabled = False
|
||||
num_tasks_box.value = min(10000, num_tasks)
|
||||
range_slider.value = (int(100 -
|
||||
(100. * num_tasks_box.value)
|
||||
/ num_tasks), 100)
|
||||
(100. * num_tasks_box.value) /
|
||||
num_tasks), 100)
|
||||
else:
|
||||
start_box.disabled = False
|
||||
end_box.disabled = False
|
||||
num_tasks_box.disabled = True
|
||||
range_slider.value = (int((start_box.value * 100.)
|
||||
/ diff),
|
||||
int((end_box.value * 100.)
|
||||
/ diff))
|
||||
range_slider.value = (int((start_box.value * 100.) /
|
||||
diff),
|
||||
int((end_box.value * 100.) /
|
||||
diff))
|
||||
|
||||
# Event was triggered by a change in the range_slider
|
||||
# value.
|
||||
@@ -157,8 +158,8 @@ def get_sliders(update):
|
||||
new_low, new_high = event["new"]
|
||||
if old_low != new_low:
|
||||
range_slider.value = (new_low, 100)
|
||||
num_tasks_box.value = (-(100. - new_low)
|
||||
/ 100. * num_tasks)
|
||||
num_tasks_box.value = (-(100. - new_low) /
|
||||
100. * num_tasks)
|
||||
else:
|
||||
range_slider.value = (0, new_high)
|
||||
num_tasks_box.value = new_high / 100. * num_tasks
|
||||
@@ -171,13 +172,13 @@ def get_sliders(update):
|
||||
elif event["owner"] == num_tasks_box:
|
||||
if num_tasks_box.value > 0:
|
||||
range_slider.value = (0, int(100 *
|
||||
float(num_tasks_box.value)
|
||||
/ num_tasks))
|
||||
float(num_tasks_box.value) /
|
||||
num_tasks))
|
||||
elif num_tasks_box.value < 0:
|
||||
range_slider.value = (100 +
|
||||
int(100 *
|
||||
float(num_tasks_box.value)
|
||||
/ num_tasks), 100)
|
||||
float(num_tasks_box.value) /
|
||||
num_tasks), 100)
|
||||
|
||||
if not update:
|
||||
return
|
||||
@@ -194,16 +195,16 @@ def get_sliders(update):
|
||||
if breakdown_opt.value == total_time_value:
|
||||
tasks = ray.global_state.task_profiles(start=(smallest +
|
||||
diff * low),
|
||||
end=(smallest
|
||||
+ diff * high))
|
||||
end=(smallest +
|
||||
diff * high))
|
||||
|
||||
# (Querying based on % of total number of tasks that were
|
||||
# run.)
|
||||
elif breakdown_opt.value == total_tasks_value:
|
||||
if range_slider.value[0] == 0:
|
||||
tasks = ray.global_state.task_profiles(num_tasks=(int(
|
||||
num_tasks
|
||||
* high)),
|
||||
num_tasks *
|
||||
high)),
|
||||
fwd=True)
|
||||
else:
|
||||
tasks = ray.global_state.task_profiles(num_tasks=(int(
|
||||
@@ -263,6 +264,74 @@ def task_search_bar():
|
||||
task_search.on_submit(handle_submit)
|
||||
|
||||
|
||||
# Helper function that guarantees unique and writeable temp files.
|
||||
# Prevents clashes in task trace files when multiple notebooks are running.
|
||||
def _get_temp_file_path(**kwargs):
|
||||
temp_file = tempfile.NamedTemporaryFile(delete=False,
|
||||
dir=os.getcwd(),
|
||||
**kwargs)
|
||||
temp_file_path = temp_file.name
|
||||
temp_file.close()
|
||||
return os.path.relpath(temp_file_path)
|
||||
|
||||
|
||||
# Helper function that ensures that catapult is cloned to the correct location
|
||||
# and that the HTML files required for task trace embedding are in the same
|
||||
# directory as the web UI.
|
||||
def _setup_trace_dependencies():
|
||||
catapult_home = "/tmp/ray/catapult"
|
||||
catapult_commit = "33a9271eb3cf5caf925293ec6a4b47c94f1ac968"
|
||||
try:
|
||||
# Check if we're inside a git repo
|
||||
cmd = ["git",
|
||||
"-C",
|
||||
catapult_home,
|
||||
"rev-parse",
|
||||
"--is-inside-work-tree"]
|
||||
subprocess.check_call(cmd)
|
||||
|
||||
except subprocess.CalledProcessError:
|
||||
# Error on non-zero exit code (e.g. - ".git not found")
|
||||
if not os.path.exists(os.path.join(catapult_home)):
|
||||
print("Cloning catapult to {}.".format(catapult_home))
|
||||
cmd = ["git",
|
||||
"clone",
|
||||
"https://github.com/catapult-project/catapult.git",
|
||||
catapult_home]
|
||||
subprocess.check_call(cmd)
|
||||
|
||||
# Checks out the commit associated with allowing different arrow
|
||||
# colors. This can and should be removed after catapult's next
|
||||
# release.
|
||||
print("Checking out commit {}.".format(catapult_commit))
|
||||
cmd = ["git", "-C", catapult_home, "checkout", catapult_commit]
|
||||
subprocess.check_call(cmd)
|
||||
|
||||
# Path to the embedded trace viewer HTML file.
|
||||
embedded_trace_path = os.path.join(catapult_home,
|
||||
"tracing",
|
||||
"bin",
|
||||
"index.html")
|
||||
# Checks that the trace viewer renderer file exists, generates it if it
|
||||
# doesn't.
|
||||
if not os.path.exists("trace_viewer_full.html"):
|
||||
vulcanize_bin = os.path.join(catapult_home,
|
||||
"tracing",
|
||||
"bin",
|
||||
"vulcanize_trace_viewer")
|
||||
# TODO(rkn): The vulcanize_trace_viewer script currently requires
|
||||
# Python 2. Remove this dependency.
|
||||
cmd = ["python2",
|
||||
vulcanize_bin,
|
||||
"--config",
|
||||
"chrome",
|
||||
"--output",
|
||||
"trace_viewer_full.html"]
|
||||
subprocess.check_call(cmd)
|
||||
|
||||
return catapult_home, embedded_trace_path
|
||||
|
||||
|
||||
def task_timeline():
|
||||
path_input = widgets.Button(description="View task timeline")
|
||||
|
||||
@@ -275,29 +344,27 @@ def task_timeline():
|
||||
description="View options:",
|
||||
disabled=False,
|
||||
)
|
||||
obj_dep = widgets.Checkbox(
|
||||
value=True,
|
||||
description="Object dependencies",
|
||||
disabled=False
|
||||
)
|
||||
task_dep = widgets.Checkbox(
|
||||
value=True,
|
||||
description="Task submissions",
|
||||
disabled=False
|
||||
)
|
||||
|
||||
start_box, end_box, range_slider, time_opt = get_sliders(False)
|
||||
display(task_dep)
|
||||
display(obj_dep)
|
||||
display(breakdown_opt)
|
||||
display(path_input)
|
||||
|
||||
def find_trace2html():
|
||||
trace2html = "/tmp/ray/catapult/tracing/bin/trace2html"
|
||||
# Clone the catapult repository if it doesn't exist. TODO(rkn): We
|
||||
# could do this in the build.sh script later on.
|
||||
if not os.path.exists(trace2html):
|
||||
cmd = ["git",
|
||||
"clone",
|
||||
"https://github.com/catapult-project/catapult.git",
|
||||
"/tmp/ray/catapult"]
|
||||
subprocess.check_output(cmd)
|
||||
print("Cloning catapult to /tmp/ray/catapult.")
|
||||
assert os.path.exists(trace2html)
|
||||
return trace2html
|
||||
|
||||
def handle_submit(sender):
|
||||
tmp = tempfile.mktemp() + ".json"
|
||||
tmp2 = tempfile.mktemp() + ".html"
|
||||
json_tmp = tempfile.mktemp() + ".json"
|
||||
|
||||
# Determine whether task components should be displayed or not.
|
||||
if breakdown_opt.value == breakdown_basic:
|
||||
breakdown = False
|
||||
elif breakdown_opt.value == breakdown_task:
|
||||
@@ -312,42 +379,57 @@ def task_timeline():
|
||||
diff = largest - smallest
|
||||
|
||||
if time_opt.value == total_time_value:
|
||||
tasks = ray.global_state.task_profiles(
|
||||
start=smallest + diff * low,
|
||||
end=smallest + diff * high)
|
||||
tasks = ray.global_state.task_profiles(start=smallest + diff * low,
|
||||
end=smallest + diff * high)
|
||||
elif time_opt.value == total_tasks_value:
|
||||
if range_slider.value[0] == 0:
|
||||
tasks = ray.global_state.task_profiles(
|
||||
num_tasks=int(num_tasks * high),
|
||||
fwd=True)
|
||||
tasks = ray.global_state.task_profiles(num_tasks=int(
|
||||
num_tasks * high),
|
||||
fwd=True)
|
||||
else:
|
||||
tasks = ray.global_state.task_profiles(
|
||||
num_tasks=int(num_tasks * (high - low)),
|
||||
fwd=False)
|
||||
tasks = ray.global_state.task_profiles(num_tasks=int(
|
||||
num_tasks *
|
||||
(high - low)),
|
||||
fwd=False)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Unexpected time value '{}'".format(time_opt.value))
|
||||
|
||||
raise ValueError("Unexpected time value '{}'".format(
|
||||
time_opt.value))
|
||||
# Write trace to a JSON file
|
||||
print("{} tasks to trace".format(len(tasks)))
|
||||
print("Dumping task profiling data to " + tmp)
|
||||
ray.global_state.dump_catapult_trace(tmp,
|
||||
print("Dumping task profiling data to " + json_tmp)
|
||||
ray.global_state.dump_catapult_trace(json_tmp,
|
||||
tasks,
|
||||
breakdowns=breakdown)
|
||||
print("Converting chrome trace to " + tmp2)
|
||||
trace2html = find_trace2html()
|
||||
# TODO(rkn): The trace2html script currently requires Python 2.
|
||||
# Remove this dependency.
|
||||
subprocess.check_output(["python2",
|
||||
trace2html,
|
||||
tmp,
|
||||
"--output",
|
||||
tmp2])
|
||||
# Open the timeline in Chrome. TODO(rkn): We should remove the
|
||||
# dependency on Chrome and use whatever browser is currently being
|
||||
# used. Note that this currently does not work when Ray is being
|
||||
# used on a cluster and the browser is running locally.
|
||||
breakdowns=breakdown,
|
||||
obj_dep=obj_dep.value,
|
||||
task_dep=task_dep.value)
|
||||
|
||||
print("Opening html file in browser...")
|
||||
subprocess.Popen(["open", "-a", "Google Chrome", tmp2])
|
||||
|
||||
# Check that the catapult repo is cloned to the correct location
|
||||
print(_setup_trace_dependencies())
|
||||
catapult_home, trace_viewer_path = _setup_trace_dependencies()
|
||||
|
||||
html_file_path = _get_temp_file_path(suffix=".html")
|
||||
json_file_path = _get_temp_file_path(suffix=".json")
|
||||
|
||||
print("Pointing to {} named {}".format(json_tmp, json_file_path))
|
||||
shutil.copy(json_tmp, json_file_path)
|
||||
|
||||
with open(trace_viewer_path) as f:
|
||||
data = f.read()
|
||||
|
||||
# Replace the demo data path with our own
|
||||
# https://github.com/catapult-project/catapult/blob/
|
||||
# 33a9271eb3cf5caf925293ec6a4b47c94f1ac968/tracing/bin/index.html#L107
|
||||
data = data.replace("../test_data/big_trace.json", json_file_path)
|
||||
|
||||
with open(html_file_path, "w+") as f:
|
||||
f.write(data)
|
||||
|
||||
# Display the task trace within the Jupyter notebook
|
||||
clear_output(wait=True)
|
||||
display(IFrame(html_file_path, 900, 800))
|
||||
print("Displaying {}".format(html_file_path))
|
||||
|
||||
path_input.on_click(handle_submit)
|
||||
|
||||
@@ -462,17 +544,17 @@ def compute_utilizations(abs_earliest,
|
||||
task_start_time = data["get_arguments_start"]
|
||||
task_end_time = data["store_outputs_end"]
|
||||
|
||||
start_bucket = int((task_start_time - earliest_time)
|
||||
/ bucket_time_length)
|
||||
end_bucket = int((task_end_time - earliest_time)
|
||||
/ bucket_time_length)
|
||||
start_bucket = int((task_start_time - earliest_time) /
|
||||
bucket_time_length)
|
||||
end_bucket = int((task_end_time - earliest_time) /
|
||||
bucket_time_length)
|
||||
# Walk over each time bucket that this task intersects, adding the
|
||||
# amount of time that the task intersects within each bucket
|
||||
for bucket_idx in range(start_bucket, end_bucket + 1):
|
||||
bucket_start_time = (earliest_time + bucket_idx
|
||||
* bucket_time_length)
|
||||
bucket_end_time = (earliest_time + (bucket_idx + 1)
|
||||
* bucket_time_length)
|
||||
bucket_start_time = ((earliest_time + bucket_idx) *
|
||||
bucket_time_length)
|
||||
bucket_end_time = ((earliest_time + (bucket_idx + 1)) *
|
||||
bucket_time_length)
|
||||
|
||||
task_start_time_within_bucket = max(task_start_time,
|
||||
bucket_start_time)
|
||||
|
||||
Reference in New Issue
Block a user