From bfe473fa8c23a1e838456e57f32b27f277fe8117 Mon Sep 17 00:00:00 2001 From: alanamarzoev Date: Wed, 9 Aug 2017 23:00:14 -0700 Subject: [PATCH] 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. --- python/ray/WebUI.ipynb | 7 + python/ray/experimental/state.py | 226 ++++++++++++++++++++++++----- python/ray/experimental/ui.py | 234 +++++++++++++++++++++---------- 3 files changed, 358 insertions(+), 109 deletions(-) diff --git a/python/ray/WebUI.ipynb b/python/ray/WebUI.ipynb index 0ba0b65b7..390263827 100644 --- a/python/ray/WebUI.ipynb +++ b/python/ray/WebUI.ipynb @@ -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, diff --git a/python/ray/experimental/state.py b/python/ray/experimental/state.py index d191cd89e..3eb5362e6 100644 --- a/python/ray/experimental/state.py +++ b/python/ray/experimental/state.py @@ -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) diff --git a/python/ray/experimental/ui.py b/python/ray/experimental/ui.py index 82d86e6a1..6048ab6bc 100644 --- a/python/ray/experimental/ui.py +++ b/python/ray/experimental/ui.py @@ -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)