path: root/src/buildstream/_state.py

#
#  Copyright (C) 2019 Bloomberg Finance LP
#
#  This program is free software; you can redistribute it and/or
#  modify it under the terms of the GNU Lesser General Public
#  License as published by the Free Software Foundation; either
#  version 2 of the License, or (at your option) any later version.
#
#  This library is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
#  Lesser General Public License for more details.
#
#  You should have received a copy of the GNU Lesser General Public
#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
#

import datetime
from typing import Optional, Tuple, List, Dict, Callable
from .types import _DisplayKey


# TaskGroup
#
# The state data stored for a group of tasks (usually scheduler queues)
#
# Args:
#    name: The name of the Task Group, e.g. 'build'
#    state: The state object
#    complete_name: Optional name for frontend status rendering, e.g. 'built'
#
class TaskGroup:
    def __init__(self, name: str, state: "State", complete_name: Optional[str] = None) -> None:

        #
        # Public members
        #
        self.name: str = name  # The name of tasks in this group
        self.complete_name: Optional[str] = complete_name  # Optional name for frontend status rendering, e.g. 'built'

        self.processed_tasks: int = 0  # Number of processed tasks
        self.skipped_tasks: int = 0  # Number of skipped tasks
        self.failed_tasks: List[str] = []  # List of element full names which failed

        #
        # Private members
        #
        self._state: "State" = state

    ###########################################
    # Core-facing APIs to drive notifications #
    ###########################################

    # add_processed_task()
    #
    # Update the TaskGroup's count of processed tasks and notify of changes
    #
    # This is a core-facing API and should not be called from the frontend
    #
    def add_processed_task(self) -> None:
        self.processed_tasks += 1
        for cb in self._state._task_groups_changed_cbs:
            cb()

    # add_skipped_task()
    #
    # Update the TaskGroup's count of skipped tasks and notify of changes
    #
    # This is a core-facing API and should not be called from the frontend
    #
    def add_skipped_task(self) -> None:
        self.skipped_tasks += 1

        for cb in self._state._task_groups_changed_cbs:
            cb()

    # add_failed_task()
    #
    # Update the TaskGroup's list of failed tasks and notify of changes
    #
    # Args:
    #    full_name: The full name of the task, distinguishing
    #               it from other tasks with the same action name
    #               e.g. an element's name.
    #
    # This is a core-facing API and should not be called from the frontend
    #
    def add_failed_task(self, full_name: str) -> None:
        self.failed_tasks.append(full_name)

        for cb in self._state._task_groups_changed_cbs:
            cb()


# Task
#
# The state data stored for an individual task
#
# Args:
#    state: The State object
#    task_id: The unique identifier of the task
#    action_name: The name of the action, e.g. 'build'
#    full_name: The full name of the task, distinguishing
#                     it from other tasks with the same action name
#                     e.g. an element's name.
#    elapsed_offset: The time the task started, relative to
#                                buildstream's start time.
class Task:
    def __init__(
        self, state: "State", task_id: str, action_name: str, full_name: str, elapsed_offset: datetime.timedelta
    ) -> None:

        #
        # Public members
        #
        self.id: str = task_id
        self.action_name: str = action_name
        self.full_name: str = full_name
        self.elapsed_offset: datetime.timedelta = elapsed_offset
        self.current_progress: Optional[int] = None
        self.maximum_progress: Optional[int] = None

        #
        # Private members
        #
        self._state: "State" = state
        self._task_changed_cb: Optional[Callable[[], None]] = None  # Callback to call when something could be rendered

    ##############################################
    # Core-facing APIs for driving notifications #
    ##############################################

    # set_task_changed_callback()
    #
    # Sets the callback to be called when this task has
    # changed.
    #
    # This is just a convenience codepath for the Messenger object
    # run simple tasks outside of the scheduler context, rather
    # than connecting to the State callbacks which are there for the
    # purpose of the frontend to get notifications of task progress.
    #
    # Args:
    #    callback: The callback to call when progress changed
    #
    def set_task_changed_callback(self, callback: Optional[Callable[[], None]]) -> None:
        self._task_changed_cb = callback

    # set_maximum_progress()
    #
    # Sets the maximum progress possible for this task.
    #
    # Args:
    #    progress: The maximum progress possible for this task
    #
    def set_maximum_progress(self, progress: int) -> None:
        self.maximum_progress = progress
        self._notify_task_changed()

    # set_current_progress()
    #
    # Sets the current progress of the task, this should
    # be a number between 0 and the maximum progress, if a
    # maximum progress has been set.
    #
    # Args:
    #    progress: The current progress
    #
    def set_current_progress(self, progress: int) -> None:
        self.current_progress = progress
        self._notify_task_changed()

    # add_current_progress()
    #
    # A convenience function for incrementing the current
    # progress of this task by 1.
    #
    def add_current_progress(self) -> None:
        if self.current_progress is None:
            new_progress = 1
        else:
            new_progress = self.current_progress + 1
        self.set_current_progress(new_progress)

    ##############################################
    #             Private methods                #
    ##############################################
    def _notify_task_changed(self) -> None:
        for cb in self._state._task_changed_cbs:
            cb(self.id)

        if self._task_changed_cb:
            self._task_changed_cb()


# State
#
# The state data that is stored for the purpose of sharing with the frontend.
#
# 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: The time the session started
#
class State:
    def __init__(self, session_start: datetime.datetime) -> None:

        #
        # Public members
        #
        self.task_groups: Dict[str, TaskGroup] = {}  # Dictionary of active task groups by group name
        self.tasks: Dict[str, Task] = {}  # Dictionary of active tasks by unique task ID

        #
        # Private members
        #
        self._session_start: datetime.datetime = session_start
        self._task_added_cbs: List[Callable[[str], None]] = []
        self._task_removed_cbs: List[Callable[[str], None]] = []
        self._task_changed_cbs: List[Callable[[str], None]] = []
        self._task_failed_cbs: List[Callable[[str, Optional[Tuple[int, _DisplayKey]]], None]] = []
        self._task_groups_changed_cbs: List[Callable[[], None]] = []

    #####################################
    # Frontend-facing notification APIs #
    #####################################

    # register_task_added_callback()
    #
    # Registers a callback to be notified when a task is added
    #
    # Args:
    #    callback: The callback to be notified
    #
    # Callback Args:
    #    task_id: The unique identifier of the task
    #
    def register_task_added_callback(self, callback: Callable[[str], None]) -> None:
        self._task_added_cbs.append(callback)

    # unregister_task_added_callback()
    #
    # Unregisters a callback previously registered by
    # register_task_added_callback()
    #
    # Args:
    #    callback: The callback to be removed
    #
    def unregister_task_added_callback(self, callback: Callable[[str], None]) -> None:
        self._task_added_cbs.remove(callback)

    # register_task_removed_callback()
    #
    # Registers a callback to be notified when a task is removed
    #
    # Args:
    #    callback: The callback to be notified
    #
    # Callback Args:
    #    task_id: The unique identifier of the task
    #
    def register_task_removed_callback(self, callback: Callable[[str], None]) -> None:
        self._task_removed_cbs.append(callback)

    # unregister_task_removed_callback()
    #
    # Unregisters a callback previously registered by
    # register_task_removed_callback()
    #
    # Args:
    #    callback: The callback to be notified
    #
    def unregister_task_removed_callback(self, callback: Callable[[str], None]) -> None:
        self._task_removed_cbs.remove(callback)

    # register_task_changed_callback()
    #
    # Register a callback to be notified when a task has changed
    #
    # Args:
    #    callback: The callback to be notified
    #
    # Callback Args:
    #    task_id: The unique identifier of the task
    #
    def register_task_changed_callback(self, callback: Callable[[str], None]) -> None:
        self._task_changed_cbs.append(callback)

    # unregister_task_changed_callback()
    #
    # Unregisters a callback previously registered by
    # register_task_changed_callback()
    #
    # Args:
    #    callback: The callback to be notified
    #
    def unregister_task_changed_callback(self, callback: Callable[[str], None]) -> None:
        self._task_changed_cbs.remove(callback)

    # register_task_failed_callback()
    #
    # Registers a callback to be notified when a task has failed
    #
    # Args:
    #    callback (function): The callback to be notified
    #
    # Callback Args:
    #    task_id: The unique identifier of the task
    #    element: (optionally) The element unique_id and DisplayKey of an element job
    #
    def register_task_failed_callback(
        self, callback: Callable[[str, Optional[Tuple[int, _DisplayKey]]], None]
    ) -> None:
        self._task_failed_cbs.append(callback)

    # unregister_task_failed_callback()
    #
    # Unregisters a callback previously registered by
    # register_task_failed_callback()
    #
    # Args:
    #    callback (function): The callback to be removed
    #
    def unregister_task_failed_callback(
        self, callback: Callable[[str, Optional[Tuple[int, _DisplayKey]]], None]
    ) -> None:
        self._task_failed_cbs.remove(callback)

    # register_task_groups_changed_callback()
    #
    # Registers a callback to be notified whenever the task groups info has changed
    #
    # Args:
    #    callback: The callback to be notified
    #
    # Callback Args:
    #    task_id: The unique identifier of the task
    #    element: (optionally) The element unique_id and DisplayKey of an element job
    #
    def register_task_groups_changed_callback(self, callback: Callable[[], None]) -> None:
        self._task_groups_changed_cbs.append(callback)

    # unregister_task_groups_changed_callback()
    #
    # Unregisters a callback previously registered by register_task_groups_changed_callback()
    #
    # Args:
    #    callback (function): The callback to be removed
    #
    def unregister_task_groups_changed_callback(self, callback: Callable[[], None]) -> None:
        self._task_groups_changed_cbs.remove(callback)

    ##############################################
    # Core-facing APIs for driving notifications #
    ##############################################

    # add_task_group()
    #
    # Notification that a new task group has been added
    #
    # This is a core-facing API and should not be called from the frontend
    #
    # Args:
    #    name (str): The name of the task group, e.g. 'build'
    #    complete_name (str): Optional name to be used for frontend status rendering, e.g. 'built'
    #
    # Returns:
    #    TaskGroup: The task group created
    #
    def add_task_group(self, name, complete_name=None) -> TaskGroup:
        assert name not in self.task_groups, "Trying to add task group '{}' to '{}'".format(name, self.task_groups)
        group = TaskGroup(name, self, complete_name)
        self.task_groups[name] = group

        return group

    # remove_task_group()
    #
    # Notification that a task group has been removed
    #
    # This is a core-facing API and should not be called from the frontend
    #
    # Args:
    #    name (str): The name of the task group, e.g. 'build'
    #
    def remove_task_group(self, name) -> None:
        # Rely on 'del' to raise an error when removing nonexistent task groups
        del self.task_groups[name]

    # add_task()
    #
    # Add a task and send appropriate notifications
    #
    # This is a core-facing API and should not be called from the frontend
    #
    # Args:
    #    task_id: The unique identifier of the task
    #    action_name: The name of the action, e.g. 'build'
    #    full_name: The full name of the task, distinguishing
    #                     it from other tasks with the same action name
    #                     e.g. an element's name.
    #    elapsed_offset (timedelta): (Optional) The time the task started, relative
    #                                to buildstream's start time. Note scheduler tasks
    #                                use this as they don't report relative to wallclock time
    #                                if the Scheduler has been suspended.
    #
    # Returns:
    #    The new task
    #
    def add_task(
        self, task_id: str, action_name: str, full_name: str, elapsed_offset: Optional[datetime.timedelta] = None
    ) -> Task:
        assert task_id not in self.tasks, "Trying to add task '{}:{}' with ID '{}' to '{}'".format(
            action_name, full_name, task_id, self.tasks
        )

        if not elapsed_offset:
            elapsed_offset = self.elapsed_time()

        task = Task(self, task_id, action_name, full_name, elapsed_offset)
        self.tasks[task_id] = task

        for cb in self._task_added_cbs:
            cb(task_id)

        return task

    # remove_task()
    #
    # Remove the task and send appropriate notifications
    #
    # This is a core-facing API and should not be called from the frontend
    #
    # Args:
    #    task_id: The unique identifier of the task
    #
    def remove_task(self, task_id: str) -> None:
        # Rely on 'del' to raise an error when removing nonexistent tasks
        del self.tasks[task_id]

        for cb in self._task_removed_cbs:
            cb(task_id)

    # fail_task()
    #
    # Notify all registered callbacks that a task has failed.
    #
    # This is separate from the tasks changed callbacks because a failed task
    # requires the frontend to intervene to decide what happens next.
    #
    # This is a core-facing API and should not be called from the frontend
    #
    # Args:
    #    task_id: The unique identifier of the task
    #    element: (optionally) The element unique_id and display keys if an
    #                                  element job
    #
    def fail_task(self, task_id: str, element: Optional[Tuple[int, _DisplayKey]] = None) -> None:
        for cb in self._task_failed_cbs:
            cb(task_id, element)

    # elapsed_time()
    #
    # Fetches the current session elapsed time
    #
    # Args:
    #    start_time: Optional explicit start time, relative to caller.
    #
    # Returns:
    #    The amount of time since the start of the session,
    #    discounting any time spent while jobs were suspended if
    #    start_time given relative to the Scheduler
    #
    def elapsed_time(self, start_time: Optional[datetime.datetime] = None) -> datetime.timedelta:
        time_now = datetime.datetime.now()
        if start_time is None:
            start_time = self._session_start or time_now
        return time_now - start_time

    # offset_start_time()
    #
    # Update the 'start' time of the application by a given offset
    #
    # This allows modifying the time spent building when BuildStream
    # gets paused then restarted, to give an accurate view of the real
    # time spend building.
    #
    # Args:
    #   offset: the offset to add to the start time
    #
    def offset_start_time(self, offset: datetime.timedelta) -> None:
        self._session_start += offset