EVOLUTION-MANAGER

Edit File: local_cli_wrapper.py

# Copyright 2016 The TensorFlow Authors. All Rights Reserved. # # Licensed 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. # ============================================================================== """Debugger Wrapper Session Consisting of a Local Curses-based CLI.""" from __future__ import absolute_import from __future__ import division from __future__ import print_function import argparse import os import sys import tempfile # Google-internal import(s). from tensorflow.python.debug.cli import analyzer_cli from tensorflow.python.debug.cli import cli_config from tensorflow.python.debug.cli import cli_shared from tensorflow.python.debug.cli import command_parser from tensorflow.python.debug.cli import debugger_cli_common from tensorflow.python.debug.cli import profile_analyzer_cli from tensorflow.python.debug.cli import ui_factory from tensorflow.python.debug.lib import common from tensorflow.python.debug.lib import debug_data from tensorflow.python.debug.wrappers import framework from tensorflow.python.lib.io import file_io _DUMP_ROOT_PREFIX = "tfdbg_" # TODO(donglin) Remove use_random_config_path after b/137652456 is fixed. class LocalCLIDebugWrapperSession(framework.BaseDebugWrapperSession): """Concrete subclass of BaseDebugWrapperSession implementing a local CLI. This class has all the methods that a `session.Session` object has, in order to support debugging with minimal code changes. Invoking its `run()` method will launch the command-line interface (CLI) of tfdbg. """ def __init__(self, sess, dump_root=None, log_usage=True, ui_type="curses", thread_name_filter=None, config_file_path=False): """Constructor of LocalCLIDebugWrapperSession. Args: sess: The TensorFlow `Session` object being wrapped. dump_root: (`str`) optional path to the dump root directory. Must be a directory that does not exist or an empty directory. If the directory does not exist, it will be created by the debugger core during debug `run()` calls and removed afterwards. If `None`, the debug dumps will be at tfdbg_<random_string> under the system temp directory. log_usage: (`bool`) whether the usage of this class is to be logged. ui_type: (`str`) requested UI type. Currently supported: (curses | readline) thread_name_filter: Regular-expression white list for thread name. See the doc of `BaseDebugWrapperSession` for details. config_file_path: Optional override to the default configuration file path, which is at `${HOME}/.tfdbg_config`. Raises: ValueError: If dump_root is an existing and non-empty directory or if dump_root is a file. """ if log_usage: pass # No logging for open-source. framework.BaseDebugWrapperSession.__init__( self, sess, thread_name_filter=thread_name_filter) if not dump_root: self._dump_root = tempfile.mktemp(prefix=_DUMP_ROOT_PREFIX) else: dump_root = os.path.expanduser(dump_root) if os.path.isfile(dump_root): raise ValueError("dump_root path points to a file: %s" % dump_root) elif os.path.isdir(dump_root) and os.listdir(dump_root): raise ValueError("dump_root path points to a non-empty directory: %s" % dump_root) self._dump_root = dump_root self._initialize_argparsers() # Registered tensor filters. self._tensor_filters = {} # Register frequently-used filter(s). self.add_tensor_filter("has_inf_or_nan", debug_data.has_inf_or_nan) # Below are the state variables of this wrapper object. # _active_tensor_filter: what (if any) tensor filter is in effect. If such # a filter is in effect, this object will call run() method of the # underlying TensorFlow Session object until the filter passes. This is # activated by the "-f" flag of the "run" command. # _run_through_times: keeps track of how many times the wrapper needs to # run through without stopping at the run-end CLI. It is activated by the # "-t" option of the "run" command. # _skip_debug: keeps track of whether the current run should be executed # without debugging. It is activated by the "-n" option of the "run" # command. # # _run_start_response: keeps track what OnRunStartResponse the wrapper # should return at the next run-start callback. If this information is # unavailable (i.e., is None), the run-start CLI will be launched to ask # the user. This is the case, e.g., right before the first run starts. self._active_tensor_filter = None self._active_filter_exclude_node_names = None self._active_tensor_filter_run_start_response = None self._run_through_times = 1 self._skip_debug = False self._run_start_response = None self._is_run_start = True self._ui_type = ui_type self._config = None if config_file_path: self._config = cli_config.CLIConfig(config_file_path=config_file_path) def _is_disk_usage_reset_each_run(self): # The dumped tensors are all cleaned up after every Session.run # in a command-line wrapper. return True def _initialize_argparsers(self): self._argparsers = {} ap = argparse.ArgumentParser( description="Run through, with or without debug tensor watching.", usage=argparse.SUPPRESS) ap.add_argument( "-t", "--times", dest="times", type=int, default=1, help="How many Session.run() calls to proceed with.") ap.add_argument( "-n", "--no_debug", dest="no_debug", action="store_true", help="Run through without debug tensor watching.") ap.add_argument( "-f", "--till_filter_pass", dest="till_filter_pass", type=str, default="", help="Run until a tensor in the graph passes the specified filter.") ap.add_argument( "-fenn", "--filter_exclude_node_names", dest="filter_exclude_node_names", type=str, default="", help="When applying the tensor filter, exclude node with names " "matching the regular expression. Applicable only if --tensor_filter " "or -f is used.") ap.add_argument( "--node_name_filter", dest="node_name_filter", type=str, default="", help="Regular-expression filter for node names to be watched in the " "run, e.g., loss, reshape.*") ap.add_argument( "--op_type_filter", dest="op_type_filter", type=str, default="", help="Regular-expression filter for op type to be watched in the run, " "e.g., (MatMul|Add), Variable.*") ap.add_argument( "--tensor_dtype_filter", dest="tensor_dtype_filter", type=str, default="", help="Regular-expression filter for tensor dtype to be watched in the " "run, e.g., (float32|float64), int.*") ap.add_argument( "-p", "--profile", dest="profile", action="store_true", help="Run and profile TensorFlow graph execution.") self._argparsers["run"] = ap ap = argparse.ArgumentParser( description="Display information about this Session.run() call.", usage=argparse.SUPPRESS) self._argparsers["run_info"] = ap self._argparsers["print_feed"] = command_parser.get_print_tensor_argparser( "Print the value of a feed in feed_dict.") def add_tensor_filter(self, filter_name, tensor_filter): """Add a tensor filter. Args: filter_name: (`str`) name of the filter. tensor_filter: (`callable`) the filter callable. See the doc string of `DebugDumpDir.find()` for more details about its signature. """ self._tensor_filters[filter_name] = tensor_filter def on_session_init(self, request): """Overrides on-session-init callback. Args: request: An instance of `OnSessionInitRequest`. Returns: An instance of `OnSessionInitResponse`. """ return framework.OnSessionInitResponse( framework.OnSessionInitAction.PROCEED) def on_run_start(self, request): """Overrides on-run-start callback. Args: request: An instance of `OnRunStartRequest`. Returns: An instance of `OnRunStartResponse`. """ self._is_run_start = True self._update_run_calls_state( request.run_call_count, request.fetches, request.feed_dict, is_callable_runner=request.is_callable_runner) if self._active_tensor_filter: # If we are running until a filter passes, we just need to keep running # with the previous `OnRunStartResponse`. return self._active_tensor_filter_run_start_response self._exit_if_requested_by_user() if self._run_call_count > 1 and not self._skip_debug: if self._run_through_times > 0: # Just run through without debugging. return framework.OnRunStartResponse( framework.OnRunStartAction.NON_DEBUG_RUN, []) elif self._run_through_times == 0: # It is the run at which the run-end CLI will be launched: activate # debugging. return (self._run_start_response or framework.OnRunStartResponse( framework.OnRunStartAction.DEBUG_RUN, self._get_run_debug_urls())) if self._run_start_response is None: self._prep_cli_for_run_start() self._run_start_response = self._launch_cli() if self._active_tensor_filter: self._active_tensor_filter_run_start_response = self._run_start_response if self._run_through_times > 1: self._run_through_times -= 1 self._exit_if_requested_by_user() return self._run_start_response def _exit_if_requested_by_user(self): if self._run_start_response == debugger_cli_common.EXPLICIT_USER_EXIT: # Explicit user "exit" command leads to sys.exit(1). print( "Note: user exited from debugger CLI: Calling sys.exit(1).", file=sys.stderr) sys.exit(1) def _prep_cli_for_run_start(self): """Prepare (but not launch) the CLI for run-start.""" self._run_cli = ui_factory.get_ui(self._ui_type, config=self._config) help_intro = debugger_cli_common.RichTextLines([]) if self._run_call_count == 1: # Show logo at the onset of the first run. help_intro.extend(cli_shared.get_tfdbg_logo()) help_intro.extend(debugger_cli_common.get_tensorflow_version_lines()) help_intro.extend(debugger_cli_common.RichTextLines("Upcoming run:")) help_intro.extend(self._run_info) self._run_cli.set_help_intro(help_intro) # Create initial screen output detailing the run. self._title = "run-start: " + self._run_description self._init_command = "run_info" self._title_color = "blue_on_white" def on_run_end(self, request): """Overrides on-run-end callback. Actions taken: 1) Load the debug dump. 2) Bring up the Analyzer CLI. Args: request: An instance of OnSessionInitRequest. Returns: An instance of OnSessionInitResponse. """ self._is_run_start = False if request.performed_action == framework.OnRunStartAction.DEBUG_RUN: partition_graphs = None if request.run_metadata and request.run_metadata.partition_graphs: partition_graphs = request.run_metadata.partition_graphs elif request.client_graph_def: partition_graphs = [request.client_graph_def] if request.tf_error and not os.path.isdir(self._dump_root): # It is possible that the dump root may not exist due to errors that # have occurred prior to graph execution (e.g., invalid device # assignments), in which case we will just raise the exception as the # unwrapped Session does. raise request.tf_error debug_dump = debug_data.DebugDumpDir( self._dump_root, partition_graphs=partition_graphs) debug_dump.set_python_graph(self._sess.graph) passed_filter = None passed_filter_exclude_node_names = None if self._active_tensor_filter: if not debug_dump.find( self._tensor_filters[self._active_tensor_filter], first_n=1, exclude_node_names=self._active_filter_exclude_node_names): # No dumped tensor passes the filter in this run. Clean up the dump # directory and move on. self._remove_dump_root() return framework.OnRunEndResponse() else: # Some dumped tensor(s) from this run passed the filter. passed_filter = self._active_tensor_filter passed_filter_exclude_node_names = ( self._active_filter_exclude_node_names) self._active_tensor_filter = None self._active_filter_exclude_node_names = None self._prep_debug_cli_for_run_end( debug_dump, request.tf_error, passed_filter, passed_filter_exclude_node_names) self._run_start_response = self._launch_cli() # Clean up the dump generated by this run. self._remove_dump_root() elif request.performed_action == framework.OnRunStartAction.PROFILE_RUN: self._prep_profile_cli_for_run_end(self._sess.graph, request.run_metadata) self._run_start_response = self._launch_cli() else: # No debug information to show following a non-debug run() call. self._run_start_response = None # Return placeholder response that currently holds no additional # information. return framework.OnRunEndResponse() def _remove_dump_root(self): if os.path.isdir(self._dump_root): file_io.delete_recursively(self._dump_root) def _prep_debug_cli_for_run_end(self, debug_dump, tf_error, passed_filter, passed_filter_exclude_node_names): """Prepare (but not launch) CLI for run-end, with debug dump from the run. Args: debug_dump: (debug_data.DebugDumpDir) The debug dump directory from this run. tf_error: (None or OpError) OpError that happened during the run() call (if any). passed_filter: (None or str) Name of the tensor filter that just passed and caused the preparation of this run-end CLI (if any). passed_filter_exclude_node_names: (None or str) Regular expression used with the tensor filter to exclude ops with names matching the regular expression. """ if tf_error: help_intro = cli_shared.get_error_intro(tf_error) self._init_command = "help" self._title_color = "red_on_white" else: help_intro = None self._init_command = "lt" self._title_color = "black_on_white" if passed_filter is not None: # Some dumped tensor(s) from this run passed the filter. self._init_command = "lt -f %s" % passed_filter if passed_filter_exclude_node_names: self._init_command += (" --filter_exclude_node_names %s" % passed_filter_exclude_node_names) self._title_color = "red_on_white" self._run_cli = analyzer_cli.create_analyzer_ui( debug_dump, self._tensor_filters, ui_type=self._ui_type, on_ui_exit=self._remove_dump_root, config=self._config) # Get names of all dumped tensors. dumped_tensor_names = [] for datum in debug_dump.dumped_tensor_data: dumped_tensor_names.append("%s:%d" % (datum.node_name, datum.output_slot)) # Tab completions for command "print_tensors". self._run_cli.register_tab_comp_context(["print_tensor", "pt"], dumped_tensor_names) # Tab completion for commands "node_info", "list_inputs" and # "list_outputs". The list comprehension is used below because nodes() # output can be unicodes and they need to be converted to strs. self._run_cli.register_tab_comp_context( ["node_info", "ni", "list_inputs", "li", "list_outputs", "lo"], [str(node_name) for node_name in debug_dump.nodes()]) # TODO(cais): Reduce API surface area for aliases vis-a-vis tab # completion contexts and registered command handlers. self._title = "run-end: " + self._run_description if help_intro: self._run_cli.set_help_intro(help_intro) def _prep_profile_cli_for_run_end(self, py_graph, run_metadata): self._init_command = "lp" self._run_cli = profile_analyzer_cli.create_profiler_ui( py_graph, run_metadata, ui_type=self._ui_type, config=self._run_cli.config) self._title = "run-end (profiler mode): " + self._run_description def _launch_cli(self): """Launch the interactive command-line interface. Returns: The OnRunStartResponse specified by the user using the "run" command. """ self._register_this_run_info(self._run_cli) response = self._run_cli.run_ui( init_command=self._init_command, title=self._title, title_color=self._title_color) return response def _run_info_handler(self, args, screen_info=None): output = debugger_cli_common.RichTextLines([]) if self._run_call_count == 1: output.extend(cli_shared.get_tfdbg_logo()) output.extend(debugger_cli_common.get_tensorflow_version_lines()) output.extend(self._run_info) if (not self._is_run_start and debugger_cli_common.MAIN_MENU_KEY in output.annotations): menu = output.annotations[debugger_cli_common.MAIN_MENU_KEY] if "list_tensors" not in menu.captions(): menu.insert( 0, debugger_cli_common.MenuItem("list_tensors", "list_tensors")) return output def _print_feed_handler(self, args, screen_info=None): np_printoptions = cli_shared.numpy_printoptions_from_screen_info( screen_info) if not self._feed_dict: return cli_shared.error( "The feed_dict of the current run is None or empty.") parsed = self._argparsers["print_feed"].parse_args(args) tensor_name, tensor_slicing = ( command_parser.parse_tensor_name_with_slicing(parsed.tensor_name)) feed_key = None feed_value = None for key in self._feed_dict: key_name = common.get_graph_element_name(key) if key_name == tensor_name: feed_key = key_name feed_value = self._feed_dict[key] break if feed_key is None: return cli_shared.error( "The feed_dict of the current run does not contain the key %s" % tensor_name) else: return cli_shared.format_tensor( feed_value, feed_key + " (feed)", np_printoptions, print_all=parsed.print_all, tensor_slicing=tensor_slicing, highlight_options=cli_shared.parse_ranges_highlight(parsed.ranges), include_numeric_summary=parsed.numeric_summary) def _run_handler(self, args, screen_info=None): """Command handler for "run" command during on-run-start.""" del screen_info # Currently unused. parsed = self._argparsers["run"].parse_args(args) parsed.node_name_filter = parsed.node_name_filter or None parsed.op_type_filter = parsed.op_type_filter or None parsed.tensor_dtype_filter = parsed.tensor_dtype_filter or None if parsed.filter_exclude_node_names and not parsed.till_filter_pass: raise ValueError( "The --filter_exclude_node_names (or -feon) flag is valid only if " "the --till_filter_pass (or -f) flag is used.") if parsed.profile: raise debugger_cli_common.CommandLineExit( exit_token=framework.OnRunStartResponse( framework.OnRunStartAction.PROFILE_RUN, [])) self._skip_debug = parsed.no_debug self._run_through_times = parsed.times if parsed.times > 1 or parsed.no_debug: # If requested -t times > 1, the very next run will be a non-debug run. action = framework.OnRunStartAction.NON_DEBUG_RUN debug_urls = [] else: action = framework.OnRunStartAction.DEBUG_RUN debug_urls = self._get_run_debug_urls() run_start_response = framework.OnRunStartResponse( action, debug_urls, node_name_regex_allowlist=parsed.node_name_filter, op_type_regex_allowlist=parsed.op_type_filter, tensor_dtype_regex_allowlist=parsed.tensor_dtype_filter) if parsed.till_filter_pass: # For the run-till-filter-pass (run -f) mode, use the DEBUG_RUN # option to access the intermediate tensors, and set the corresponding # state flag of the class itself to True. if parsed.till_filter_pass in self._tensor_filters: action = framework.OnRunStartAction.DEBUG_RUN self._active_tensor_filter = parsed.till_filter_pass self._active_filter_exclude_node_names = ( parsed.filter_exclude_node_names) self._active_tensor_filter_run_start_response = run_start_response else: # Handle invalid filter name. return debugger_cli_common.RichTextLines( ["ERROR: tensor filter \"%s\" does not exist." % parsed.till_filter_pass]) # Raise CommandLineExit exception to cause the CLI to exit. raise debugger_cli_common.CommandLineExit(exit_token=run_start_response) def _register_this_run_info(self, curses_cli): curses_cli.register_command_handler( "run", self._run_handler, self._argparsers["run"].format_help(), prefix_aliases=["r"]) curses_cli.register_command_handler( "run_info", self._run_info_handler, self._argparsers["run_info"].format_help(), prefix_aliases=["ri"]) curses_cli.register_command_handler( "print_feed", self._print_feed_handler, self._argparsers["print_feed"].format_help(), prefix_aliases=["pf"]) if self._tensor_filters: # Register tab completion for the filter names. curses_cli.register_tab_comp_context(["run", "r"], list(self._tensor_filters.keys())) if self._feed_dict and hasattr(self._feed_dict, "keys"): # Register tab completion for feed_dict keys. feed_keys = [common.get_graph_element_name(key) for key in self._feed_dict.keys()] curses_cli.register_tab_comp_context(["print_feed", "pf"], feed_keys) def _get_run_debug_urls(self): """Get the debug_urls value for the current run() call. Returns: debug_urls: (list of str) Debug URLs for the current run() call. Currently, the list consists of only one URL that is a file:// URL. """ return ["file://" + self._dump_root] def _update_run_calls_state(self, run_call_count, fetches, feed_dict, is_callable_runner=False): """Update the internal state with regard to run() call history. Args: run_call_count: (int) Number of run() calls that have occurred. fetches: a node/tensor or a list of node/tensor that are the fetches of the run() call. This is the same as the fetches argument to the run() call. feed_dict: None of a dict. This is the feed_dict argument to the run() call. is_callable_runner: (bool) whether a runner returned by Session.make_callable is being run. """ self._run_call_count = run_call_count self._feed_dict = feed_dict self._run_description = cli_shared.get_run_short_description( run_call_count, fetches, feed_dict, is_callable_runner=is_callable_runner) self._run_through_times -= 1 self._run_info = cli_shared.get_run_start_intro( run_call_count, fetches, feed_dict, self._tensor_filters, is_callable_runner=is_callable_runner)