Add UI to engine execution

src/python/pants/bin/local_pants_runner.py

@@ -189,7 +189,8 @@ def _maybe_run_v2(self):
     try:
       self._graph_session.run_console_rules(
         self._options.goals_and_possible_v2_goals,
-        self._target_roots
+        self._target_roots,
+        self._global_options.v2_ui
       )
     except GracefulTerminationException as e:
       logger.debug('Encountered graceful termination exception {}; exiting'.format(e))

src/python/pants/engine/BUILD

@@ -174,6 +174,7 @@ python_library(
     'src/python/pants/base:exceptions',
     'src/python/pants/base:specs',
     'src/python/pants/build_graph',
+    'src/python/pants/rules/core:core',
     'src/python/pants/util:objects',
   ]
 )

src/python/pants/engine/native.py

@@ -221,7 +221,7 @@
 Session* session_create(Scheduler*);
 void session_destroy(Session*);
 
-ExecutionRequest* execution_request_create(void);
+ExecutionRequest* execution_request_create(_Bool, uint64_t);
 void execution_request_destroy(ExecutionRequest*);
 
 uint64_t graph_len(Scheduler*);
@@ -788,8 +788,10 @@ def to_ids_buf(self, types):
   def new_tasks(self):
     return self.gc(self.lib.tasks_create(), self.lib.tasks_destroy)
 
-  def new_execution_request(self):
-    return self.gc(self.lib.execution_request_create(), self.lib.execution_request_destroy)
+  def new_execution_request(self, v2_ui, ui_worker_count):
+    return self.gc(
+      self.lib.execution_request_create(v2_ui, ui_worker_count),
+      self.lib.execution_request_destroy)
 
   def new_session(self, scheduler):
     return self.gc(self.lib.session_create(scheduler), self.lib.session_destroy)
@@ -873,7 +875,7 @@ def tc(constraint):
         execution_options.remote_store_chunk_bytes,
         execution_options.remote_store_chunk_upload_timeout_seconds,
         execution_options.process_execution_parallelism,
-        execution_options.process_execution_cleanup_local_dirs
+        execution_options.process_execution_cleanup_local_dirs,
       )
     return self.gc(scheduler, self.lib.scheduler_destroy)
 

src/python/pants/engine/scheduler.py

@@ -5,6 +5,7 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import logging
+import multiprocessing
 import os
 import time
 from builtins import object, open, str, zip
@@ -102,7 +103,6 @@ def __init__(
 
     self._native = native
     self.include_trace_on_error = include_trace_on_error
-
     # Validate and register all provided and intrinsic tasks.
     rule_index = RuleIndex.create(list(rules))
     self._root_subject_types = sorted(rule_index.roots, key=repr)
@@ -422,13 +422,13 @@ def visualize_graph_to_file(self, filename):
   def visualize_rule_graph_to_file(self, filename):
     self._scheduler.visualize_rule_graph_to_file(filename)
 
-  def execution_request_literal(self, request_specs):
-    native_execution_request = self._scheduler._native.new_execution_request()
+  def execution_request_literal(self, request_specs, v2_ui):
+    native_execution_request = self._scheduler._native.new_execution_request(v2_ui, multiprocessing.cpu_count())
     for subject, product in request_specs:
       self._scheduler.add_root_selection(native_execution_request, subject, product)
     return ExecutionRequest(request_specs, native_execution_request)
 
-  def execution_request(self, products, subjects):
+  def execution_request(self, products, subjects, v2_ui=False):
     """Create and return an ExecutionRequest for the given products and subjects.
 
     The resulting ExecutionRequest object will contain keys tied to this scheduler's product Graph,
@@ -442,10 +442,11 @@ def execution_request(self, products, subjects):
     :param subjects: A list of Spec and/or PathGlobs objects.
     :type subject: list of :class:`pants.base.specs.Spec`, `pants.build_graph.Address`, and/or
       :class:`pants.engine.fs.PathGlobs` objects.
+    :param bool v2_ui: whether to render the v2 engine UI
     :returns: An ExecutionRequest for the given products and subjects.
     """
     roots = (tuple((s, p) for s in subjects for p in products))
-    return self.execution_request_literal(roots)
+    return self.execution_request_literal(roots, v2_ui)
 
   def invalidate_files(self, direct_filenames):
     """Invalidates the given filenames in an internal product Graph instance."""
@@ -513,51 +514,73 @@ def execute(self, execution_request):
     except TaskError as e:
       return ExecutionResult.failure(e)
 
-  def products_request(self, products, subjects):
-    """Executes a request for multiple products for some subjects, and returns the products.
+  def _trace_on_error(self, unique_exceptions, request):
+    exception_noun = pluralize(len(unique_exceptions), 'Exception')
+    if self._scheduler.include_trace_on_error:
+      cumulative_trace = '\n'.join(self.trace(request))
+      raise ExecutionError(
+        '{} encountered:\n{}'.format(exception_noun, cumulative_trace),
+        unique_exceptions,
+      )
+    else:
+      raise ExecutionError(
+        '{} encountered:\n  {}'.format(
+          exception_noun,
+          '\n  '.join('{}: {}'.format(type(t).__name__, str(t)) for t in unique_exceptions)),
+        unique_exceptions
+      )
+
+  def run_console_rule(self, product, subject, v2_ui):
+    """
 
-    :param list products: A list of product type for the request.
-    :param list subjects: A list of subjects for the request.
-    :returns: A dict from product type to lists of products each with length matching len(subjects).
+    :param product: product type for the request.
+    :param subject: subject for the request.
+    :param v2_ui: whether to render the v2 engine UI
+    :return: A dict from product type to lists of products each with length matching len(subjects).
     """
-    request = self.execution_request(products, subjects)
+    request = self.execution_request([product], [subject], v2_ui)
     result = self.execute(request)
     if result.error:
       raise result.error
 
+    self._state_validation(result)
+    assert len(result.root_products) == 1
+    root, state = result.root_products[0]
+    if type(state) is Throw:
+      exc = state.exc
+      if isinstance(exc, GracefulTerminationException):
+        raise exc
+      self._trace_on_error([exc], request)
+    return {result.root_products[0]: [state.value]}
+
+  def _state_validation(self, result):
     # State validation.
     unknown_state_types = tuple(
       type(state) for _, state in result.root_products if type(state) not in (Throw, Return)
     )
     if unknown_state_types:
       State.raise_unrecognized(unknown_state_types)
 
+  def products_request(self, products, subjects):
+    """Executes a request for multiple products for some subjects, and returns the products.
+
+    :param list products: A list of product type for the request.
+    :param list subjects: A list of subjects for the request.
+    :returns: A dict from product type to lists of products each with length matching len(subjects).
+    """
+    request = self.execution_request(products, subjects)
+    result = self.execute(request)
+    if result.error:
+      raise result.error
+
+    self._state_validation(result)
+
     # Throw handling.
     # TODO: See https://github.com/pantsbuild/pants/issues/3912
     throw_root_states = tuple(state for root, state in result.root_products if type(state) is Throw)
     if throw_root_states:
       unique_exceptions = tuple({t.exc for t in throw_root_states})
-
-      # TODO: consider adding a new top-level function adjacent to products_request used for running console tasks,
-      # so that this code doesn't need to exist in this form.
-      if len(unique_exceptions) == 1 and isinstance(unique_exceptions[0], GracefulTerminationException):
-        raise unique_exceptions[0]
-
-      exception_noun = pluralize(len(unique_exceptions), 'Exception')
-
-      if self._scheduler.include_trace_on_error:
-        cumulative_trace = '\n'.join(self.trace(request))
-        raise ExecutionError(
-          '{} encountered:\n{}'.format(exception_noun, cumulative_trace),
-          unique_exceptions,
-        )
-      else:
-        raise ExecutionError(
-          '{} encountered:\n  {}'.format(
-            exception_noun,
-            '\n  '.join('{}: {}'.format(type(t).__name__, str(t)) for t in unique_exceptions)),
-          unique_exceptions
-        )
+      self._trace_on_error(unique_exceptions, request)
 
     # Everything is a Return: we rely on the fact that roots are ordered to preserve subject
     # order in output lists.

src/python/pants/init/engine_initializer.py

@@ -192,21 +192,24 @@ def validate_goals(self, goals):
     if invalid_goals:
       raise self.InvalidGoals(invalid_goals)
 
-  def run_console_rules(self, goals, target_roots):
+  def run_console_rules(self, goals, target_roots, v2_ui):
     """Runs @console_rules sequentially and interactively by requesting their implicit Goal products.
 
     For retryable failures, raises scheduler.ExecutionError.
 
     :param list goals: The list of requested goal names as passed on the commandline.
     :param TargetRoots target_roots: The targets root of the request.
+    :param bool v2_ui: whether to render the v2 engine UI
     """
     # Reduce to only applicable goals - with validation happening by way of `validate_goals()`.
     goals = [goal for goal in goals if goal in self.goal_map]
     subjects = self._determine_subjects(target_roots)
+    # Console rule can only have one subject.
+    assert len(subjects) == 1
     for goal in goals:
       goal_product = self.goal_map[goal]
       logger.debug('requesting {} to satisfy execution of `{}` goal'.format(goal_product, goal))
-      self.scheduler_session.product_request(goal_product, subjects)
+      self.scheduler_session.run_console_rule(goal_product, subjects[0], v2_ui)
 
   def create_build_graph(self, target_roots, build_root=None):
     """Construct and return a `BuildGraph` given a set of input specs.

src/python/pants/option/global_options.py

@@ -330,6 +330,9 @@ def register_options(cls, register):
              help='Enables execution of v1 Tasks.')
     register('--v2', advanced=True, type=bool, default=False,
              help='Enables execution of v2 @console_rules.')
+    register('--v2-ui', default=False, type=bool, daemon=False,
+             help='Whether to show v2 engine execution progress. '
+                  'This requires the --v2 flag to take effect.')
 
     loop_flag = '--loop'
     register(loop_flag, type=bool,
@@ -373,3 +376,6 @@ def validate_instance(cls, opts):
                          '`--v2 --no-v1` to function as expected.')
     if opts.loop and not opts.enable_pantsd:
       raise OptionsError('The --loop option requires `--enable-pantsd`, in order to watch files.')
+
+    if opts.v2_ui and not opts.v2:
+      raise OptionsError('The --v2-ui option requires --v2 to be enabled together.')

src/python/pants/pantsd/service/scheduler_service.py

@@ -214,7 +214,7 @@ def _prefork_body(self, session, options):
         session.validate_goals(options.goals_and_possible_v2_goals)
 
       # N.B. @console_rules run pre-fork in order to cache the products they request during execution.
-      session.run_console_rules(options.goals_and_possible_v2_goals, target_roots)
+      session.run_console_rules(options.goals_and_possible_v2_goals, target_roots, v2_ui=global_options.v2_ui)
 
     return target_roots
 

src/rust/engine/Cargo.toml

@@ -86,3 +86,4 @@ resettable = { path = "resettable" }
 smallvec = "0.6"
 tokio = "0.1"
 tempfile = "3"
+ui = { path = "ui" }

src/rust/engine/src/lib.rs

@@ -67,6 +67,7 @@ extern crate resettable;
 #[macro_use]
 extern crate smallvec;
 extern crate tokio;
+extern crate ui;
 
 use std::ffi::CStr;
 use std::fs::File;
@@ -571,8 +572,14 @@ pub extern "C" fn session_destroy(ptr: *mut Session) {
 }
 
 #[no_mangle]
-pub extern "C" fn execution_request_create() -> *const ExecutionRequest {
-  Box::into_raw(Box::new(ExecutionRequest::new()))
+pub extern "C" fn execution_request_create(
+  should_render_ui: bool,
+  ui_worker_count: u64,
+) -> *const ExecutionRequest {
+  Box::into_raw(Box::new(ExecutionRequest::new(
+    should_render_ui,
+    ui_worker_count,
+  )))
 }
 
 #[no_mangle]

src/rust/engine/src/scheduler.rs

@@ -18,6 +18,7 @@ use nodes::{NodeKey, Select, Tracer, TryInto, Visualizer};
 use parking_lot::Mutex;
 use rule_graph;
 use selectors;
+use ui::EngineDisplay;
 
 ///
 /// A Session represents a related series of requests (generally: one run of the pants CLI) on an
@@ -55,11 +56,18 @@ impl Session {
 pub struct ExecutionRequest {
   // Set of roots for an execution, in the order they were declared.
   pub roots: Vec<Root>,
+  // Flag used to determine whether to show engine execution progress.
+  pub should_render_ui: bool,
+  pub ui_worker_count: u64,
 }
 
 impl ExecutionRequest {
-  pub fn new() -> ExecutionRequest {
-    ExecutionRequest { roots: Vec::new() }
+  pub fn new(should_render_ui: bool, ui_worker_count: u64) -> ExecutionRequest {
+    ExecutionRequest {
+      roots: Vec::new(),
+      should_render_ui,
+      ui_worker_count,
+    }
   }
 
   ///
@@ -261,12 +269,29 @@ impl Scheduler {
       core: self.core.clone(),
     };
     let (sender, receiver) = mpsc::channel();
+
+    // Setting up display
+    let mut optional_display: Option<EngineDisplay> =
+      EngineDisplay::create(request.ui_worker_count as usize, request.should_render_ui);
+
     Scheduler::execute_helper(context, sender, request.roots.clone(), 8);
+    let roots: Vec<NodeKey> = request
+      .roots
+      .clone()
+      .into_iter()
+      .map(|s| s.into())
+      .collect();
+
     let results = loop {
       if let Ok(res) = receiver.recv_timeout(Duration::from_millis(100)) {
         break res;
+      } else if let Some(display) = optional_display.as_mut() {
+        Scheduler::display_ongoing_tasks(&self.core.graph, &roots, display);
       }
     };
+    if let Some(display) = optional_display.as_mut() {
+      display.finish();
+    };
 
     request
       .roots
@@ -276,6 +301,21 @@ impl Scheduler {
       .collect()
   }
 
+  fn display_ongoing_tasks(graph: &Graph<NodeKey>, roots: &[NodeKey], display: &mut EngineDisplay) {
+    let display_worker_count = display.worker_count();
+    let ongoing_tasks = graph.heavy_hitters(&roots, display_worker_count);
+    for (i, task) in ongoing_tasks.iter().enumerate() {
+      display.update(i.to_string(), format!("{:?}", task));
+    }
+    // If the number of ongoing tasks is less than the number of workers,
+    // fill the rest of the workers with empty string.
+    // TODO(yic): further improve the UI. https://github.com/pantsbuild/pants/issues/6666
+    for i in ongoing_tasks.len()..display_worker_count {
+      display.update(i.to_string(), "".to_string());
+    }
+    display.render();
+  }
+
   pub fn capture_snapshot_from_arbitrary_root<P: AsRef<Path>>(
     &self,
     root_path: P,

src/rust/engine/ui/Cargo.toml

@@ -4,11 +4,10 @@ version = "0.0.1"
 authors = [ "Pants Build <pantsbuild@gmail.com>" ]
 
 [lib]
-name = "engine_display"
 path = "src/display.rs"
 
 [[bin]]
-name = "ui"
+name = "demo"
 path = "src/main.rs"
 
 [dependencies]