Add UI to engine execution

src/python/pants/bin/local_pants_runner.py

@@ -188,7 +188,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.render_v2_engine_ui
       )
     except Exception as e:
       logger.warn('Encountered unhandled exception {!r} during rule execution!'

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);
 void execution_request_destroy(ExecutionRequest*);
 
 uint64_t graph_len(Scheduler*);
@@ -781,8 +781,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, render_v2_engine_ui):
+    return self.gc(
+      self.lib.execution_request_create(render_v2_engine_ui),
+      self.lib.execution_request_destroy)
 
   def new_session(self, scheduler):
     return self.gc(self.lib.session_create(scheduler), self.lib.session_destroy)
@@ -866,7 +868,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

@@ -101,7 +101,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)
@@ -421,13 +420,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, render_v2_engine_ui):
+    native_execution_request = self._scheduler._native.new_execution_request(render_v2_engine_ui)
     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, render_v2_engine_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,
@@ -441,10 +440,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 render_v2_engine_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, render_v2_engine_ui)
 
   def invalidate_files(self, direct_filenames):
     """Invalidates the given filenames in an internal product Graph instance."""
@@ -512,14 +512,15 @@ def execute(self, execution_request):
     except TaskError as e:
       return ExecutionResult.failure(e)
 
-  def products_request(self, products, subjects):
+  def products_request(self, products, subjects, render_v2_engine_ui):
     """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.
+    :param bool render_v2_engine_ui: whether to render the v2 engine UI
     :returns: 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(products, subjects, render_v2_engine_ui)
     result = self.execute(request)
     if result.error:
       raise result.error
@@ -559,14 +560,15 @@ def products_request(self, products, subjects):
       product_results[product].append(state.value)
     return product_results
 
-  def product_request(self, product, subjects):
+  def product_request(self, product, subjects, render_v2_engine_ui=False):
     """Executes a request for a single product for some subjects, and returns the products.
 
     :param class product: A product type for the request.
     :param list subjects: A list of subjects for the request.
+    :param bool render_v2_engine_ui: whether to render the v2 engine UI
     :returns: A list of the requested products, with length match len(subjects).
     """
-    return self.products_request([product], subjects)[product]
+    return self.products_request([product], subjects, render_v2_engine_ui)[product]
 
   def capture_snapshots(self, path_globs_and_roots):
     """Synchronously captures Snapshots for each matching PathGlobs rooted at a its root directory.

src/python/pants/init/engine_initializer.py

@@ -192,21 +192,22 @@ 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, render_v2_engine_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 render_v2_engine_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)
     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.product_request(goal_product, subjects, render_v2_engine_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

@@ -223,6 +223,8 @@ def register_bootstrap_options(cls, register):
                   'of the directory will be overwritten if any filenames collide.')
     register('--print-exception-stacktrace', advanced=True, type=bool,
              help='Print to console the full exception stack trace if encountered.')
+    register('--render-v2-engine-ui', default=False, type=bool, daemon=False,
+             help='whether to show v2 engine execution progress.')
 
     # BinaryUtil options.
     register('--binaries-baseurls', type=list, advanced=True,

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, render_v2_engine_ui=False)
 
     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,8 @@ 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) -> *const ExecutionRequest {
+  Box::into_raw(Box::new(ExecutionRequest::new(should_render_ui)))
 }
 
 #[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,16 @@ 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,
 }
 
 impl ExecutionRequest {
-  pub fn new() -> ExecutionRequest {
-    ExecutionRequest { roots: Vec::new() }
+  pub fn new(should_render_ui: bool) -> ExecutionRequest {
+    ExecutionRequest {
+      roots: Vec::new(),
+      should_render_ui,
+    }
   }
 
   ///
@@ -261,12 +267,54 @@ impl Scheduler {
       core: self.core.clone(),
     };
     let (sender, receiver) = mpsc::channel();
+
+    // Setting up display
+    let display_worker_count = 8;
+    let mut optional_display = if request.should_render_ui {
+      Some(EngineDisplay::for_stdout(0))
+    } else {
+      None
+    };
+    if let Some(display) = optional_display.as_mut() {
+      display.start();
+      display.render();
+      let worker_ids: Vec<String> = (0..display_worker_count)
+        .map(|s| format!("{}", s))
+        .collect();
+      for worker_id in worker_ids {
+        display.add_worker(worker_id);
+        display.render();
+      }
+    }
+
     Scheduler::execute_helper(context, sender, request.roots.clone(), 8);
+    let roots: Vec<_> = 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() {
+        let ongoing_tasks = self.core.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.
+        for i in ongoing_tasks.len()..display_worker_count {
+          display.update(i.to_string(), "".to_string());
+        }
+        display.render();
       }
     };
+    if let Some(display) = optional_display.as_mut() {
+      display.render();
+      display.finish();
+    };
 
     request
       .roots

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]

src/rust/engine/ui/src/display.rs

@@ -244,7 +244,7 @@ impl EngineDisplay {
 
   fn render_for_pipe(&self) {
     // TODO: Handle non-tty output w polling interval adjustment + summary rendering.
-    panic!("TODO");
+    // Nothing needs to be printed to pipe by default.
   }
 
   // Paints one screen of rendering.
@@ -309,6 +309,10 @@ impl EngineDisplay {
 
   // Terminates the EngineDisplay and returns the cursor to a static position.
   pub fn finish(&mut self) {
+    // Don't do anything if it's not tty.
+    if !self.is_tty {
+      return;
+    }
     self.running = false;
     let current_pos = self.get_cursor_pos();
     let action_count = self.action_map.len() as u16;

src/rust/engine/ui/src/main.rs

@@ -27,15 +27,15 @@
 // Arc<Mutex> can be more clear than needing to grok Orderings:
 #![cfg_attr(feature = "cargo-clippy", allow(mutex_atomic))]
 
-extern crate engine_display;
 extern crate rand;
+extern crate ui;
 
 use std::thread;
 use std::time::Duration;
 
 use rand::Rng;
 
-use engine_display::EngineDisplay;
+use ui::EngineDisplay;
 
 // N.B. This is purely a demo/testing bin target for exercising the library.
 

tests/python/pants_test/engine/test_scheduler.py

@@ -40,7 +40,7 @@ def test_trace_includes_rule_exception_traceback(self):
     ]
 
     scheduler = create_scheduler(rules)
-    request = scheduler._native.new_execution_request()
+    request = scheduler._native.new_execution_request(render_v2_engine_ui=False)
     subject = B()
     scheduler.add_root_selection(request, subject, A)
     session = scheduler.new_session()