Merge branch 'jonathan/job-progress' into 'master'fetch-committers

Render progress information for loading and processing elements

See merge request BuildStream/buildstream!1482

9 files changed, 410 insertions, 61 deletions

diff --git a/src/buildstream/_frontend/app.py b/src/buildstream/_frontend/app.py
index 0c140a83a..76d3b8dde 100644
--- a/src/buildstream/_frontend/app.py
+++ b/src/buildstream/_frontend/app.py
@@ -230,6 +230,9 @@ class App():
             # Propagate pipeline feedback to the user
             self.context.messenger.set_message_handler(self._message_handler)
 
+            # Allow the Messenger to write status messages
+            self.context.messenger.set_render_status_cb(self._maybe_render_status)
+
             # Preflight the artifact cache after initializing logging,
             # this can cause messages to be emitted.
             try:
diff --git a/src/buildstream/_frontend/status.py b/src/buildstream/_frontend/status.py
index 32fda1147..38e388818 100644
--- a/src/buildstream/_frontend/status.py
+++ b/src/buildstream/_frontend/status.py
@@ -19,6 +19,7 @@
 import os
 import sys
 import curses
+from collections import OrderedDict
 import click
 
 # Import a widget internal for formatting time codes
@@ -64,7 +65,7 @@ class Status():
         self._success_profile = success_profile
         self._error_profile = error_profile
         self._stream = stream
-        self._jobs = []
+        self._jobs = OrderedDict()
         self._last_lines = 0  # Number of status lines we last printed to console
         self._spacing = 1
         self._colors = colors
@@ -81,6 +82,7 @@ class Status():
 
         state.register_task_added_callback(self._add_job)
         state.register_task_removed_callback(self._remove_job)
+        state.register_task_changed_callback(self._job_changed)
 
     # clear()
     #
@@ -162,6 +164,21 @@ class Status():
     #                 Private Methods                 #
     ###################################################
 
+    # _job_changed()
+    #
+    # Reacts to a specified job being changed
+    #
+    # Args:
+    #    action_name (str): The action name for this job
+    #    full_name (str): The name of this specific job (e.g. element name)
+    #
+    def _job_changed(self, action_name, full_name):
+        job_key = (action_name, full_name)
+        task = self._state.tasks[job_key]
+        job = self._jobs[job_key]
+        if job.update(task):
+            self._need_alloc = True
+
     # _init_terminal()
     #
     # Initialize the terminal and return the resolved terminal
@@ -260,8 +277,9 @@ class Status():
         self._need_alloc = False
 
     def _job_lines(self, columns):
+        jobs_list = list(self._jobs.values())
         for i in range(0, len(self._jobs), columns):
-            yield self._jobs[i:i + columns]
+            yield jobs_list[i:i + columns]
 
     # Returns an array of integers representing the maximum
     # length in characters for each column, given the current
@@ -290,11 +308,11 @@ class Status():
     #
     def _add_job(self, action_name, full_name):
         task = self._state.tasks[(action_name, full_name)]
-        start_time = task.start_time
+        elapsed = task.elapsed_offset
         job = _StatusJob(self._context, action_name, full_name,
                          self._content_profile, self._format_profile,
-                         start_time)
-        self._jobs.append(job)
+                         elapsed)
+        self._jobs[(action_name, full_name)] = job
         self._need_alloc = True
 
     # _remove_job()
@@ -306,11 +324,7 @@ class Status():
     #    full_name (str): The name of this specific job (e.g. element name)
     #
     def _remove_job(self, action_name, full_name):
-        self._jobs = [
-            job for job in self._jobs
-            if not (job.full_name == full_name and
-                    job.action_name == action_name)
-        ]
+        del self._jobs[(action_name, full_name)]
         self._need_alloc = True
 
 
@@ -486,12 +500,59 @@ class _StatusJob():
         self._content_profile = content_profile
         self._format_profile = format_profile
         self._time_code = TimeCode(context, content_profile, format_profile)
+        self._current_progress = None  # Progress tally to render
+        self._maximum_progress = None  # Progress tally to render
 
+        self.size = self.calculate_size()
+
+    # calculate_size()
+    #
+    # Calculates the amount of space the job takes up when rendered
+    #
+    # Returns:
+    #    int: The size of the job when rendered
+    #
+    def calculate_size(self):
         # Calculate the size needed to display
-        self.size = 10  # Size of time code with brackets
-        self.size += len(action_name)
-        self.size += len(self.full_name)
-        self.size += 3  # '[' + ':' + ']'
+        size = 10  # Size of time code with brackets
+        size += len(self.action_name)
+        size += len(self.full_name)
+        size += 3  # '[' + ':' + ']'
+        if self._current_progress is not None:
+            size += len(str(self._current_progress))
+            size += 1  # ':'
+            if self._maximum_progress is not None:
+                size += len(str(self._maximum_progress))
+                size += 1  # '/'
+        return size
+
+    # update()
+    #
+    # Synchronises its internal data with the provided Task,
+    # and returns whether its size has changed
+    #
+    # Args:
+    #    task (Task): The task associated with this job
+    #
+    # Returns:
+    #    bool: Whether the size of the job has changed
+    #
+    def update(self, task):
+        changed = False
+        size_changed = False
+        if task.current_progress != self._current_progress:
+            changed = True
+            self._current_progress = task.current_progress
+        if task.maximum_progress != self._maximum_progress:
+            changed = True
+            self._maximum_progress = task.maximum_progress
+        if changed:
+            old_size = self.size
+            self.size = self.calculate_size()
+            if self.size != old_size:
+                size_changed = True
+
+        return size_changed
 
     # render()
     #
@@ -506,12 +567,20 @@ class _StatusJob():
             self._time_code.render_time(elapsed - self._offset) + \
             self._format_profile.fmt(']')
 
-        # Add padding after the display name, before terminating ']'
-        name = self.full_name + (' ' * padding)
         text += self._format_profile.fmt('[') + \
             self._content_profile.fmt(self.action_name) + \
             self._format_profile.fmt(':') + \
-            self._content_profile.fmt(name) + \
-            self._format_profile.fmt(']')
+            self._content_profile.fmt(self.full_name)
+
+        if self._current_progress is not None:
+            text += self._format_profile.fmt(':') + \
+                self._content_profile.fmt(str(self._current_progress))
+            if self._maximum_progress is not None:
+                text += self._format_profile.fmt('/') + \
+                    self._content_profile.fmt(str(self._maximum_progress))
+
+        # Add padding before terminating ']'
+        terminator = (' ' * padding) + ']'
+        text += terminator
 
         return text
diff --git a/src/buildstream/_loader/loader.py b/src/buildstream/_loader/loader.py
index 6108f10b8..0ec9b9e17 100644
--- a/src/buildstream/_loader/loader.py
+++ b/src/buildstream/_loader/loader.py
@@ -88,11 +88,12 @@ class Loader():
     #                       this is a bit more expensive due to deep copies
     #    ticker (callable): An optional function for tracking load progress
     #    targets (list of str): Target, element-path relative bst filenames in the project
+    #    task (Task): A task object to report progress to
     #
     # Raises: LoadError
     #
     # Returns: The toplevel LoadElement
-    def load(self, targets, rewritable=False, ticker=None):
+    def load(self, targets, rewritable=False, ticker=None, task=None):
 
         for filename in targets:
             if os.path.isabs(filename):
@@ -141,7 +142,7 @@ class Loader():
 
             # Finally, wrap what we have into LoadElements and return the target
             #
-            ret.append(loader._collect_element(element))
+            ret.append(loader._collect_element(element, task))
 
         self._clean_caches()
 
@@ -411,11 +412,12 @@ class Loader():
     #
     # Args:
     #    element (LoadElement): The element for which to load a MetaElement
+    #    task (Task): A task to write progress information to
     #
     # Returns:
     #    (MetaElement): A partially loaded MetaElement
     #
-    def _collect_element_no_deps(self, element):
+    def _collect_element_no_deps(self, element, task):
         # Return the already built one, if we already built it
         meta_element = self._meta_elements.get(element.name)
         if meta_element:
@@ -452,6 +454,8 @@ class Loader():
 
         # Cache it now, make sure it's already there before recursing
         self._meta_elements[element.name] = meta_element
+        if task:
+            task.add_current_progress()
 
         return meta_element
 
@@ -461,13 +465,14 @@ class Loader():
     #
     # Args:
     #    top_element (LoadElement): The element for which to load a MetaElement
+    #    task (Task): The task to update with progress changes
     #
     # Returns:
     #    (MetaElement): A fully loaded MetaElement
     #
-    def _collect_element(self, top_element):
+    def _collect_element(self, top_element, task):
         element_queue = [top_element]
-        meta_element_queue = [self._collect_element_no_deps(top_element)]
+        meta_element_queue = [self._collect_element_no_deps(top_element, task)]
 
         while element_queue:
             element = element_queue.pop()
@@ -484,7 +489,7 @@ class Loader():
                 name = dep.element.name
 
                 if name not in loader._meta_elements:
-                    meta_dep = loader._collect_element_no_deps(dep.element)
+                    meta_dep = loader._collect_element_no_deps(dep.element, task)
                     element_queue.append(dep.element)
                     meta_element_queue.append(meta_dep)
                 else:
@@ -555,7 +560,9 @@ class Loader():
                 return None
 
         # meta junction element
-        meta_element = self._collect_element(self._elements[filename])
+        # XXX: This is a likely point for progress reporting to end up
+        # missing some elements, but it currently doesn't appear to be the case.
+        meta_element = self._collect_element(self._elements[filename], None)
         if meta_element.kind != 'junction':
             raise LoadError("{}{}: Expected junction but element kind is {}"
                             .format(provenance_str, filename, meta_element.kind),
diff --git a/src/buildstream/_messenger.py b/src/buildstream/_messenger.py
index d83b464ff..d768abf0c 100644
--- a/src/buildstream/_messenger.py
+++ b/src/buildstream/_messenger.py
@@ -28,6 +28,18 @@ from ._message import Message, MessageType
 from .plugin import Plugin
 
 
+_RENDER_INTERVAL = datetime.timedelta(seconds=1)
+
+
+# TimeData class to contain times in an object that can be passed around
+# and updated from different places
+class _TimeData():
+    __slots__ = ['start_time']
+
+    def __init__(self, start_time):
+        self.start_time = start_time
+
+
 class Messenger():
 
     def __init__(self):
@@ -35,6 +47,10 @@ class Messenger():
         self._silence_scope_depth = 0
         self._log_handle = None
         self._log_filename = None
+        self._state = None
+        self._next_render = None  # A Time object
+        self._active_simple_tasks = 0
+        self._render_status_cb = None
 
     # set_message_handler()
     #
@@ -51,6 +67,29 @@ class Messenger():
     def set_message_handler(self, handler):
         self._message_handler = handler
 
+    # set_state()
+    #
+    # Sets the State object within the Messenger
+    #
+    # Args:
+    #    state (State): The state to set
+    #
+    def set_state(self, state):
+        self._state = state
+
+    # set_render_status_cb()
+    #
+    # Sets the callback to use to render status
+    #
+    # Args:
+    #    callback (function): The Callback to be notified
+    #
+    # Callback Args:
+    #    There are no arguments to the callback
+    #
+    def set_render_status_cb(self, callback):
+        self._render_status_cb = callback
+
     # _silent_messages():
     #
     # Returns:
@@ -110,28 +149,13 @@ class Messenger():
     #
     # Args:
     #    activity_name (str): The name of the activity
-    #    context (Context): The invocation context object
     #    unique_id (int): Optionally, the unique id of the plugin related to the message
     #    detail (str): An optional detailed message, can be multiline output
     #    silent_nested (bool): If True, all but _message.unconditional_messages are silenced
     #
     @contextmanager
     def timed_activity(self, activity_name, *, unique_id=None, detail=None, silent_nested=False):
-
-        starttime = datetime.datetime.now()
-        stopped_time = None
-
-        def stop_time():
-            nonlocal stopped_time
-            stopped_time = datetime.datetime.now()
-
-        def resume_time():
-            nonlocal stopped_time
-            nonlocal starttime
-            sleep_time = datetime.datetime.now() - stopped_time
-            starttime += sleep_time
-
-        with _signals.suspendable(stop_time, resume_time):
+        with self._timed_suspendable() as timedata:
             try:
                 # Push activity depth for status messages
                 message = Message(unique_id, MessageType.START, activity_name, detail=detail)
@@ -142,15 +166,75 @@ class Messenger():
             except BstError:
                 # Note the failure in status messages and reraise, the scheduler
                 # expects an error when there is an error.
-                elapsed = datetime.datetime.now() - starttime
+                elapsed = datetime.datetime.now() - timedata.start_time
                 message = Message(unique_id, MessageType.FAIL, activity_name, elapsed=elapsed)
                 self.message(message)
                 raise
 
-            elapsed = datetime.datetime.now() - starttime
+            elapsed = datetime.datetime.now() - timedata.start_time
             message = Message(unique_id, MessageType.SUCCESS, activity_name, elapsed=elapsed)
             self.message(message)
 
+    # simple_task()
+    #
+    # Context manager for creating a task to report progress to.
+    #
+    # Args:
+    #    activity_name (str): The name of the activity
+    #    unique_id (int): Optionally, the unique id of the plugin related to the message
+    #    full_name (str): Optionally, the distinguishing name of the activity, e.g. element name
+    #    silent_nested (bool): If True, all but _message.unconditional_messages are silenced
+    #
+    # Yields:
+    #    Task: A Task object that represents this activity, principally used to report progress
+    #
+    @contextmanager
+    def simple_task(self, activity_name, *, unique_id=None, full_name=None, silent_nested=False):
+        # Bypass use of State when none exists (e.g. tests)
+        if not self._state:
+            with self.timed_activity(activity_name, unique_id=unique_id, silent_nested=silent_nested):
+                yield
+            return
+
+        if not full_name:
+            full_name = activity_name
+
+        with self._timed_suspendable() as timedata:
+            try:
+                message = Message(unique_id, MessageType.START, activity_name)
+                self.message(message)
+
+                task = self._state.add_task(activity_name, full_name)
+                task.set_render_cb(self._render_status)
+                self._active_simple_tasks += 1
+                if not self._next_render:
+                    self._next_render = datetime.datetime.now() + _RENDER_INTERVAL
+
+                with self.silence(actually_silence=silent_nested):
+                    yield task
+
+            except BstError:
+                elapsed = datetime.datetime.now() - timedata.start_time
+                message = Message(unique_id, MessageType.FAIL, activity_name, elapsed=elapsed)
+                self.message(message)
+                raise
+            finally:
+                self._state.remove_task(activity_name, full_name)
+                self._active_simple_tasks -= 1
+                if self._active_simple_tasks == 0:
+                    self._next_render = None
+
+            elapsed = datetime.datetime.now() - timedata.start_time
+            if task.current_progress is not None:
+                if task.maximum_progress is not None:
+                    detail = "{} of {} subtasks processed".format(task.current_progress, task.maximum_progress)
+                else:
+                    detail = "{} subtasks processed".format(task.current_progress)
+            else:
+                detail = None
+            message = Message(unique_id, MessageType.SUCCESS, activity_name, elapsed=elapsed, detail=detail)
+            self.message(message)
+
     # recorded_messages()
     #
     # Records all messages in a log file while the context manager
@@ -311,4 +395,53 @@ class Messenger():
         #
         del state['_message_handler']
 
+        # The render status callback is only used in the main process
+        #
+        del state['_render_status_cb']
+
+        # The State object is not needed outside the main process
+        del state['_state']
+
         return state
+
+    # _render_status()
+    #
+    # Calls the render status callback set in the messenger, but only if a
+    # second has passed since it last rendered.
+    #
+    def _render_status(self):
+        assert self._next_render
+
+        # self._render_status_cb()
+        now = datetime.datetime.now()
+        if self._render_status_cb and now >= self._next_render:
+            self._render_status_cb()
+            self._next_render = now + _RENDER_INTERVAL
+
+    # _timed_suspendable()
+    #
+    # A contextmanager that allows an activity to be suspended and can
+    # adjust for clock drift caused by suspending
+    #
+    # Yields:
+    #    TimeData: An object that contains the time the activity started
+    #
+    @contextmanager
+    def _timed_suspendable(self):
+        # Note: timedata needs to be in a namedtuple so that values can be
+        # yielded that will change
+        timedata = _TimeData(start_time=datetime.datetime.now())
+        stopped_time = None
+
+        def stop_time():
+            nonlocal stopped_time
+            stopped_time = datetime.datetime.now()
+
+        def resume_time():
+            nonlocal timedata
+            nonlocal stopped_time
+            sleep_time = datetime.datetime.now() - stopped_time
+            timedata.start_time += sleep_time
+
+        with _signals.suspendable(stop_time, resume_time):
+            yield timedata
diff --git a/src/buildstream/_project.py b/src/buildstream/_project.py
index b14109630..f4a7466de 100644
--- a/src/buildstream/_project.py
+++ b/src/buildstream/_project.py
@@ -423,12 +423,18 @@ class Project():
     #    (list): A list of loaded Element
     #
     def load_elements(self, targets, *, rewritable=False):
-        with self._context.messenger.timed_activity("Loading elements", silent_nested=True):
-            meta_elements = self.loader.load(targets, rewritable=rewritable, ticker=None)
+        with self._context.messenger.simple_task("Loading elements", silent_nested=True) as task:
+            meta_elements = self.loader.load(targets, rewritable=rewritable, ticker=None, task=task)
 
-        with self._context.messenger.timed_activity("Resolving elements"):
+            # workaround for task potentially being None (because no State object)
+            if task:
+                total_elements = task.current_progress
+
+        with self._context.messenger.simple_task("Resolving elements") as task:
+            if task:
+                task.set_maximum_progress(total_elements)
             elements = [
-                Element._new_from_meta(meta)
+                Element._new_from_meta(meta, task)
                 for meta in meta_elements
             ]
 
diff --git a/src/buildstream/_state.py b/src/buildstream/_state.py
index b2f0b705d..388ed8151 100644
--- a/src/buildstream/_state.py
+++ b/src/buildstream/_state.py
@@ -15,7 +15,7 @@
 #  License along with this library. If not, see <http://www.gnu.org/licenses/>.
 #
 
-
+import datetime
 from collections import OrderedDict
 
 
@@ -92,8 +92,14 @@ class TaskGroup():
 # BuildStream's Core is responsible for making changes to this data.
 # BuildStream's Frontend may register callbacks with State to be notified
 # when parts of State change, and read State to know what has changed.
+#
+# Args:
+#    session_start (datetime): The time the session started
+#
 class State():
-    def __init__(self):
+    def __init__(self, session_start):
+        self._session_start = session_start
+
         self.task_groups = OrderedDict()  # key is TaskGroup name
 
         # Note: A Task's full_name is technically unique, but only accidentally.
@@ -101,6 +107,7 @@ class State():
 
         self._task_added_cbs = []
         self._task_removed_cbs = []
+        self._task_changed_cbs = []
         self._task_groups_changed_cbs = []
         self._task_failed_cbs = []
 
@@ -162,6 +169,33 @@ class State():
     def unregister_task_removed_callback(self, callback):
         self._task_removed_cbs.remove(callback)
 
+    # register_task_changed_callback()
+    #
+    # Register a callback to be notified when a task has changed
+    #
+    # Args:
+    #    callback (function): The callback to be notified
+    #
+    # Callback Args:
+    #    action_name (str): The name of the action, e.g. 'build'
+    #    full_name (str): The full name of the task, distinguishing
+    #                     it from other tasks with the same action name
+    #                     e.g. an element's name.
+    #
+    def register_task_changed_callback(self, callback):
+        self._task_changed_cbs.append(callback)
+
+    # unregister_task_changed_callback()
+    #
+    # Unregisters a callback previously registered by
+    # register_task_changed_callback()
+    #
+    # Args:
+    #    callback (function): The callback to be notified
+    #
+    def unregister_task_changed_callback(self, callback):
+        self._task_changed_cbs.remove(callback)
+
     # register_task_failed_callback()
     #
     # Registers a callback to be notified when a task has failed
@@ -238,20 +272,25 @@ class State():
     #    full_name (str): The full name of the task, distinguishing
     #                     it from other tasks with the same action name
     #                     e.g. an element's name.
-    #    start_time (timedelta): The time the task started, relative to
-    #                            buildstream's start time.
+    #    elapsed_offset (timedelta): (Optional) The time the task started, relative
+    #                                to buildstream's start time.
     #
-    def add_task(self, action_name, full_name, start_time):
+    def add_task(self, action_name, full_name, elapsed_offset=None):
         task_key = (action_name, full_name)
         assert task_key not in self.tasks, \
             "Trying to add task '{}:{}' to '{}'".format(action_name, full_name, self.tasks)
 
-        task = _Task(action_name, full_name, start_time)
+        if not elapsed_offset:
+            elapsed_offset = datetime.datetime.now() - self._session_start
+
+        task = _Task(self, action_name, full_name, elapsed_offset)
         self.tasks[task_key] = task
 
         for cb in self._task_added_cbs:
             cb(action_name, full_name)
 
+        return task
+
     # remove_task()
     #
     # Remove the task and send appropriate notifications
@@ -297,14 +336,55 @@ class State():
 # The state data stored for an individual task
 #
 # Args:
+#    state (State): The State object
 #    action_name (str): The name of the action, e.g. 'build'
 #    full_name (str): The full name of the task, distinguishing
 #                     it from other tasks with the same action name
 #                     e.g. an element's name.
-#    start_time (timedelta): The time the task started, relative to
-#                            buildstream's start time.
+#    elapsed_offset (timedelta): The time the task started, relative to
+#                                buildstream's start time.
 class _Task():
-    def __init__(self, action_name, full_name, start_time):
+    def __init__(self, state, action_name, full_name, elapsed_offset):
+        self._state = state
         self.action_name = action_name
         self.full_name = full_name
-        self.start_time = start_time
+        self.elapsed_offset = elapsed_offset
+        self.current_progress = None
+        self.maximum_progress = None
+
+        self._render_cb = None  # Callback to call when something could be rendered
+
+    # set_render_cb()
+    #
+    # Sets the callback to be called when the Task has changed and should be rendered
+    #
+    # NOTE: This should probably be removed once the frontend is running
+    #       separately from the scheduler, since renders could be triggered
+    #       by the scheduler.
+    def set_render_cb(self, callback):
+        self._render_cb = callback
+
+    def set_current_progress(self, progress):
+        self.current_progress = progress
+        for cb in self._state._task_changed_cbs:
+            cb(self.action_name, self.full_name)
+        if self._render_cb:
+            self._render_cb()
+
+    def set_maximum_progress(self, progress):
+        self.maximum_progress = progress
+        for cb in self._state._task_changed_cbs:
+            cb(self.action_name, self.full_name)
+
+        if self._render_cb:
+            self._render_cb()
+
+    def add_current_progress(self):
+        if self.current_progress is None:
+            new_progress = 1
+        else:
+            new_progress = self.current_progress + 1
+        self.set_current_progress(new_progress)
+
+    def add_maximum_progress(self):
+        self.set_maximum_progress(self.maximum_progress or 0 + 1)
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 0f320c569..3713c87a7 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -78,7 +78,10 @@ class Stream():
         self._sourcecache = None
         self._project = None
         self._pipeline = None
-        self._state = State()        # Owned by Stream, used by Core to set state
+        self._state = State(session_start)  # Owned by Stream, used by Core to set state
+
+        context.messenger.set_state(self._state)
+
         self._scheduler = Scheduler(context, session_start, self._state,
                                     interrupt_callback=interrupt_callback,
                                     ticker_callback=ticker_callback)
diff --git a/src/buildstream/element.py b/src/buildstream/element.py
index efa876c73..ffdd1511e 100644
--- a/src/buildstream/element.py
+++ b/src/buildstream/element.py
@@ -936,12 +936,13 @@ class Element(Plugin):
     #
     # Args:
     #    meta (MetaElement): The meta element
+    #    task (Task): A task object to report progress to
     #
     # Returns:
     #    (Element): A newly created Element instance
     #
     @classmethod
-    def _new_from_meta(cls, meta):
+    def _new_from_meta(cls, meta, task=None):
 
         if not meta.first_pass:
             meta.project.ensure_fully_loaded()
@@ -967,7 +968,7 @@ class Element(Plugin):
 
         # Instantiate dependencies
         for meta_dep in meta.dependencies:
-            dependency = Element._new_from_meta(meta_dep)
+            dependency = Element._new_from_meta(meta_dep, task)
             element.__runtime_dependencies.append(dependency)
             dependency.__reverse_runtime_deps.add(element)
         no_of_runtime_deps = len(element.__runtime_dependencies)
@@ -976,7 +977,7 @@ class Element(Plugin):
         element.__runtime_deps_uncached = no_of_runtime_deps
 
         for meta_dep in meta.build_dependencies:
-            dependency = Element._new_from_meta(meta_dep)
+            dependency = Element._new_from_meta(meta_dep, task)
             element.__build_dependencies.append(dependency)
             dependency.__reverse_build_deps.add(element)
         no_of_build_deps = len(element.__build_dependencies)
@@ -986,6 +987,9 @@ class Element(Plugin):
 
         element.__preflight()
 
+        if task:
+            task.add_current_progress()
+
         return element
 
     # _clear_meta_elements_cache()
diff --git a/tests/frontend/show.py b/tests/frontend/show.py
index d173fd80a..0aa720681 100644
--- a/tests/frontend/show.py
+++ b/tests/frontend/show.py
@@ -40,6 +40,16 @@ def test_show(cli, datafiles, target, fmt, expected):
                              .format(expected, result.output))
 
 
+@pytest.mark.datafiles(os.path.join(DATA_DIR, 'project'))
+def test_show_progress_tally(cli, datafiles):
+    # Check that the progress reporting messages give correct tallies
+    project = str(datafiles)
+    result = cli.run(project=project, args=['show', 'compose-all.bst'])
+    result.assert_success()
+    assert "  3 subtasks processed" in result.stderr
+    assert "3 of 3 subtasks processed" in result.stderr
+
+
 @pytest.mark.datafiles(os.path.join(
     os.path.dirname(os.path.realpath(__file__)),
     "invalid_element_path",
@@ -387,6 +397,40 @@ def test_fetched_junction(cli, tmpdir, datafiles, element_name, workspaced):
     assert 'junction.bst:import-etc.bst-buildable' in results
 
 
+@pytest.mark.datafiles(os.path.join(DATA_DIR, 'project'))
+def test_junction_tally(cli, tmpdir, datafiles):
+    # Check that the progress reporting messages count elements in junctions
+    project = str(datafiles)
+    subproject_path = os.path.join(project, 'files', 'sub-project')
+    junction_path = os.path.join(project, 'elements', 'junction.bst')
+    element_path = os.path.join(project, 'elements', 'junction-dep.bst')
+
+    # Create a repo to hold the subproject and generate a junction element for it
+    generate_junction(tmpdir, subproject_path, junction_path, store_ref=True)
+
+    # Create a stack element to depend on a cross junction element
+    #
+    element = {
+        'kind': 'stack',
+        'depends': [
+            {
+                'junction': 'junction.bst',
+                'filename': 'import-etc.bst'
+            }
+        ]
+    }
+    _yaml.roundtrip_dump(element, element_path)
+
+    result = cli.run(project=project, silent=True, args=[
+        'source', 'fetch', 'junction.bst'])
+    result.assert_success()
+
+    # Assert the correct progress tallies are in the logging
+    result = cli.run(project=project, args=['show', 'junction-dep.bst'])
+    assert "  2 subtasks processed" in result.stderr
+    assert "2 of 2 subtasks processed" in result.stderr
+
+
 ###############################################################
 #                   Testing recursion depth                   #
 ###############################################################