[Notes] [Git][BuildStream/buildstream][tristan/remove-blessings] 7 commi

Tristan Van Berkom pushed to branch tristan/remove-blessings at BuildStream / buildstream

Commits:

ca811a4d

by Josh Smith at 2018-09-19T09:25:21Z

Rework Skipped usage

The SKIPPED message type is now used to indicate the end of a task which
was successful without having to perform the given task.

This overhauls the use of `Queue.done()` and therefore queues do not
need to provide a processed/skipped return value from `done()`. Instead
this is replaced with the action of raising a `SkipJob` exception from
within `Queue.process()`.

16fbc3e3

by Josh Smith at 2018-09-19T09:25:21Z

tests.py: Test skip on push

Adds a test to ensure that BuildStream alerts the user of a skipped push
when the remote already has the artifact cached.

ab56f693

by Josh Smith at 2018-09-19T09:25:21Z

element.py: Remove redundant timed_activity

This removes the timed_activity for an element _push action. This is
unnecessary as the job is already being timed elsewhere.

198fbfe8

by Josh Smith at 2018-09-19T09:25:21Z

cascache.py: Move push/pull messaging to cascache

Pulled/Pushed messages will no longer be produced from within
element.py, instead they will be produced during CasCache push() and
pull() appropriately.

Message consistency has also been improved.

213f7794

by Josh Smith at 2018-09-19T09:25:21Z

cascache.py: Modify messaging API calls

Both pulling and pushing INFO messages are now status messages.

Calls to the messaging API through `self.context.message()` have now
been switched to `element.info`.

dddd6025

by Qinusty at 2018-09-19T10:59:48Z

Merge branch 'Qinusty/skipped-rework' into 'master'

Add SkipJob for indicating a skipped activity

See merge request BuildStream/buildstream!765

16d9c6e5

by Tristan Van Berkom at 2018-09-19T11:08:33Z

_frontend/status.py: Completely remove the blessings dependency from BuildStream

This actually improves reliability of the status bar because we
now disable it completely in the case that not all of the terminal
escape sequences are supported on the given terminal.

This replaces the few functions we were using, to move the cursor
up one line, move it to the beginning of the line, and to clear a
line, with low level functions provided by the curses module in
the standard library.

This change makes it easier for downstream distro package maintainers
to package BuildStream, particularly on Fedora.

Asides from changing _frontend/status.py, this commit includes the
following changes:

  * _frontend/app.py: Use python isatty() function to determine if
                      we are connected to a tty, instead of relying
		      on blessings.

  * setup.py: Remove the dependency on blessings.

16 changed files:

buildstream/_artifactcache/cascache.py
buildstream/_exceptions.py
buildstream/_frontend/app.py
buildstream/_frontend/status.py
buildstream/_scheduler/jobs/job.py
buildstream/_scheduler/queues/buildqueue.py
buildstream/_scheduler/queues/fetchqueue.py
buildstream/_scheduler/queues/pullqueue.py
buildstream/_scheduler/queues/pushqueue.py
buildstream/_scheduler/queues/queue.py
buildstream/_scheduler/queues/trackqueue.py
buildstream/element.py
setup.py
tests/frontend/pull.py
tests/frontend/push.py
tests/testutils/runcli.py

Changes:

buildstream/_artifactcache/cascache.py

@@ -228,8 +228,8 @@ class CASCache(ArtifactCache):
          for remote in self._remotes[project]:
              try:
                  remote.init()
+-
 -                element.info("Pulling {} <- {}".format(element._get_brief_display_key(), remote.spec.url))
 +                display_key = element._get_brief_display_key()
 +                element.status("Pulling artifact {} <- {}".format(display_key, remote.spec.url))
                  request = buildstream_pb2.GetReferenceRequest()
                  request.key = ref
@@ -243,6 +243,7 @@ class CASCache(ArtifactCache):
                  self.set_ref(ref, tree)
 +                element.info("Pulled artifact {} <- {}".format(display_key, remote.spec.url))
                  # no need to pull from additional remotes
                  return True
@@ -251,11 +252,8 @@ class CASCache(ArtifactCache):
                      raise ArtifactError("Failed to pull artifact {}: {}".format(
                          element._get_brief_display_key(), e)) from e
                  else:
 -                    self.context.message(Message(
 -                        None,
 -                        MessageType.SKIPPED,
 -                        "Remote ({}) does not have {} cached".format(
 -                            remote.spec.url, element._get_brief_display_key())
 +                    element.info("Remote ({}) does not have {} cached".format(
 +                        remote.spec.url, element._get_brief_display_key()
                      ))
          return False
@@ -336,17 +334,15 @@ class CASCache(ArtifactCache):
          for remote in push_remotes:
              remote.init()
+-
 -            element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
 +            display_key = element._get_brief_display_key()
 +            element.status("Pushing artifact {} -> {}".format(display_key, remote.spec.url))
              if self._push_refs_to_remote(refs, remote):
 +                element.info("Pushed artifact {} -> {}".format(display_key, remote.spec.url))
                  pushed = True
              else:
 -                self.context.message(Message(
 -                    None,
 -                    MessageType.SKIPPED,
 -                    "Remote ({}) already has {} cached".format(
 -                        remote.spec.url, element._get_brief_display_key())
 +                element.info("Remote ({}) already has {} cached".format(
 +                    remote.spec.url, element._get_brief_display_key()
                  ))
          return pushed

buildstream/_exceptions.py

@@ -312,3 +312,12 @@ class StreamError(BstError):
  class AppError(BstError):
      def __init__(self, message, detail=None, reason=None):
          super().__init__(message, detail=detail, domain=ErrorDomain.APP, reason=reason)
++
++
 +# SkipJob
 +#
 +# Raised from a child process within a job when the job should be
 +# considered skipped by the parent process.
 +#
 +class SkipJob(Exception):
 +    pass

buildstream/_frontend/app.py

@@ -26,7 +26,6 @@ import datetime
  from textwrap import TextWrapper
  import click
  from click import UsageError
 -from blessings import Terminal
  # Import buildstream public symbols
  from .. import Scope
@@ -92,7 +91,7 @@ class App():
+         #
          # Earily initialization
+         #
 -        is_a_tty = Terminal().is_a_tty
 +        is_a_tty = sys.stdout.isatty() and sys.stderr.isatty()
          # Enable interactive mode if we're attached to a tty
          if main_options['no_interactive']:

buildstream/_frontend/status.py

@@ -16,8 +16,10 @@
+ #
  #  Authors:
  #        Tristan Van Berkom <tristan vanberkom codethink co uk>
 +import os
 +import sys
  import click
 -from blessings import Terminal
 +import curses
  # Import a widget internal for formatting time codes
  from .widget import TimeCode
@@ -43,6 +45,13 @@ from .._scheduler import ElementJob
+ #
  class Status():
 +    # Table of the terminal capabilities we require and use
 +    _TERM_CAPABILITIES = {
 +        'move_up': 'cuu1',
 +        'move_x': 'hpa',
 +        'clear_eol': 'el'
 +    }
++
      def __init__(self, context,
                   content_profile, format_profile,
                   success_profile, error_profile,
@@ -56,7 +65,6 @@ class Status():
          self._stream = stream
          self._jobs = []
          self._last_lines = 0  # Number of status lines we last printed to console
 -        self._term = Terminal()
          self._spacing = 1
          self._colors = colors
          self._header = _StatusHeader(context,
@@ -69,6 +77,7 @@ class Status():
          self._alloc_columns = None
          self._line_length = 0
          self._need_alloc = True
 +        self._term_caps = self._init_terminal()
      # add_job()
+     #
@@ -121,7 +130,7 @@ class Status():
+     #
      def clear(self):
 -        if not self._term.does_styling:
 +        if not self._term_caps:
              return
          for _ in range(self._last_lines):
@@ -138,7 +147,7 @@ class Status():
      # not necessary to call clear().
      def render(self):
 -        if not self._term.does_styling:
 +        if not self._term_caps:
              return
          elapsed = self._stream.elapsed_time
@@ -185,6 +194,55 @@ class Status():
      ###################################################
      #                 Private Methods                 #
      ###################################################
++
 +    # _init_terminal()
 +    #
 +    # Initialize the terminal and return the resolved terminal
 +    # capabilities dictionary.
 +    #
 +    # Returns:
 +    #    (dict|None): The resolved terminal capabilities dictionary,
 +    #                 or None if the terminal does not support all
 +    #                 of the required capabilities.
 +    #
 +    def _init_terminal(self):
++
 +        # We need both output streams to be connected to a terminal
 +        if not (sys.stdout.isatty() and sys.stderr.isatty()):
 +            return None
++
 +        # Initialized terminal, curses might decide it doesnt
 +        # support this terminal
 +        try:
 +            curses.setupterm(os.environ.get('TERM', 'dumb'))
 +        except curses.error:
 +            return None
++
 +        term_caps = {}
++
 +        # Resolve the string capabilities we need for the capability
 +        # names we need.
 +        #
 +        for capname, capval in self._TERM_CAPABILITIES.items():
 +            code = curses.tigetstr(capval)
++
 +            # If any of the required capabilities resolve empty strings or None,
 +            # then we don't have the capabilities we need for a status bar on
 +            # this terminal.
 +            if not code:
 +                return None
++
 +            # Decode sequences as latin1, as they are always 8-bit bytes,
 +            # so when b'\xff' is returned, this must be decoded to u'\xff'.
 +            #
 +            # This technique is employed by the python blessings library
 +            # as well, and should provide better compatibility with most
 +            # terminals.
 +            #
 +            term_caps[capname] = code.decode('latin1')
++
 +        return term_caps
++
      def _check_term_width(self):
          term_width, _ = click.get_terminal_size()
          if self._term_width != term_width:
@@ -192,12 +250,24 @@ class Status():
              self._need_alloc = True
      def _move_up(self):
 +        assert self._term_caps is not None
++
          # Explicitly move to beginning of line, fixes things up
          # when there was a ^C or ^Z printed to the terminal.
 -        click.echo(self._term.move_x(0) + self._term.move_up, nl=False, err=True)
 +        move_x = curses.tparm(self._term_caps['move_x'].encode('latin1'), 0)
 +        move_x = move_x.decode('latin1')
++
 +        move_up = curses.tparm(self._term_caps['move_up'].encode('latin1'))
 +        move_up = move_up.decode('latin1')
++
 +        click.echo(move_x + move_up, nl=False, err=True)
      def _clear_line(self):
 -        click.echo(self._term.clear_eol, nl=False, err=True)
 +        assert self._term_caps is not None
++
 +        clear_eol = curses.tparm(self._term_caps['clear_eol'].encode('latin1'))
 +        clear_eol = clear_eol.decode('latin1')
 +        click.echo(clear_eol, nl=False, err=True)
      def _allocate(self):
          if not self._need_alloc:

buildstream/_scheduler/jobs/job.py

@@ -31,7 +31,7 @@ import multiprocessing
  import psutil
  # BuildStream toplevel imports
 -from ..._exceptions import ImplError, BstError, set_last_task_error
 +from ..._exceptions import ImplError, BstError, set_last_task_error, SkipJob
  from ..._message import Message, MessageType, unconditional_messages
  from ... import _signals, utils
@@ -40,6 +40,7 @@ from ... import _signals, utils
  RC_OK = 0
  RC_FAIL = 1
  RC_PERM_FAIL = 2
 +RC_SKIPPED = 3
  # Used to distinguish between status messages and return values
@@ -117,7 +118,7 @@ class Job():
          self._max_retries = max_retries        # Maximum number of automatic retries
          self._result = None                    # Return value of child action in the parent
          self._tries = 0                        # Try count, for retryable jobs
+-
 +        self._skipped_flag = False             # Indicate whether the job was skipped.
          # If False, a retry will not be attempted regardless of whether _tries is less than _max_retries.
+         #
          self._retry_flag = True
@@ -277,6 +278,14 @@ class Job():
      def set_task_id(self, task_id):
          self._task_id = task_id
 +    # skipped
 +    #
 +    # Returns:
 +    #    bool: True if the job was skipped while processing.
 +    @property
 +    def skipped(self):
 +        return self._skipped_flag
++
      #######################################################
      #                  Abstract Methods                   #
      #######################################################
@@ -398,6 +407,13 @@ class Job():
              try:
                  # Try the task action
                  result = self.child_process()
 +            except SkipJob as e:
 +                elapsed = datetime.datetime.now() - starttime
 +                self.message(MessageType.SKIPPED, str(e),
 +                             elapsed=elapsed, logfile=filename)
++
 +                # Alert parent of skip by return code
 +                self._child_shutdown(RC_SKIPPED)
              except BstError as e:
                  elapsed = datetime.datetime.now() - starttime
                  self._retry_flag = e.temporary
@@ -545,14 +561,18 @@ class Job():
          # We don't want to retry if we got OK or a permanent fail.
          # This is set in _child_action but must also be set for the parent.
+         #
 -        self._retry_flag = returncode not in (RC_OK, RC_PERM_FAIL)
 +        self._retry_flag = returncode == RC_FAIL
++
 +        # Set the flag to alert Queue that this job skipped.
 +        self._skipped_flag = returncode == RC_SKIPPED
          if self._retry_flag and (self._tries <= self._max_retries) and not self._scheduler.terminated:
              self.spawn()
              return
 -        self.parent_complete(returncode == RC_OK, self._result)
 -        self._scheduler.job_completed(self, returncode == RC_OK)
 +        success = returncode in (RC_OK, RC_SKIPPED)
 +        self.parent_complete(success, self._result)
 +        self._scheduler.job_completed(self, success)
          # Force the deletion of the queue and process objects to try and clean up FDs
          self._queue = self._process = None

buildstream/_scheduler/queues/buildqueue.py

@@ -47,6 +47,7 @@ class BuildQueue(Queue):
                  to_queue.append(element)
                  continue
 +            # XXX: Fix this, See https://mail.gnome.org/archives/buildstream-list/2018-September/msg00029.html
              # Bypass queue processing entirely the first time it's tried.
              self._tried.add(element)
              _, description, detail = element._get_build_result()
@@ -113,5 +114,3 @@ class BuildQueue(Queue):
              # This has to be done after _assemble_done, such that the
              # element may register its cache key as required
              self._check_cache_size(job, element, result)
+-
 -        return True

buildstream/_scheduler/queues/fetchqueue.py

@@ -72,11 +72,9 @@ class FetchQueue(Queue):
      def done(self, _, element, result, success):
          if not success:
 -            return False
 +            return
          element._update_state()
          # Successful fetch, we must be CACHED now
          assert element._get_consistency() == Consistency.CACHED
+-
 -        return True

buildstream/_scheduler/queues/pullqueue.py

@@ -21,6 +21,7 @@
  # Local imports
  from . import Queue, QueueStatus
  from ..resources import ResourceType
 +from ..._exceptions import SkipJob
  # A queue which pulls element artifacts
@@ -33,7 +34,8 @@ class PullQueue(Queue):
      def process(self, element):
          # returns whether an artifact was downloaded or not
 -        return element._pull()
 +        if not element._pull():
 +            raise SkipJob(self.action_name)
      def status(self, element):
          # state of dependencies may have changed, recalculate element state
@@ -63,7 +65,3 @@ class PullQueue(Queue):
          # do not get an artifact size from pull jobs, we have to
          # actually check the cache size.
          self._scheduler.check_cache_size()
+-
 -        # Element._pull() returns True if it downloaded an artifact,
 -        # here we want to appear skipped if we did not download.
 -        return result

buildstream/_scheduler/queues/pushqueue.py

@@ -21,6 +21,7 @@
  # Local imports
  from . import Queue, QueueStatus
  from ..resources import ResourceType
 +from ..._exceptions import SkipJob
  # A queue which pushes element artifacts
@@ -33,20 +34,11 @@ class PushQueue(Queue):
      def process(self, element):
          # returns whether an artifact was uploaded or not
 -        return element._push()
 +        if not element._push():
 +            raise SkipJob(self.action_name)
      def status(self, element):
          if element._skip_push():
              return QueueStatus.SKIP
          return QueueStatus.READY
+-
 -    def done(self, _, element, result, success):
+-
 -        if not success:
 -            return False
+-
 -        # Element._push() returns True if it uploaded an artifact,
 -        # here we want to appear skipped if the remote already had
 -        # the artifact.
 -        return result

buildstream/_scheduler/queues/queue.py

@@ -136,10 +136,6 @@ class Queue():
      #    success (bool): True if the process() implementation did not
      #                    raise any exception
+     #
 -    # Returns:
 -    #    (bool): True if the element should appear to be processsed,
 -    #            Otherwise False will count the element as "skipped"
 -    #
      def done(self, job, element, result, success):
          pass
@@ -306,8 +302,7 @@ class Queue():
          # and determine if it should be considered as processed
          # or skipped.
          try:
 -            processed = self.done(job, element, result, success)
+-
 +            self.done(job, element, result, success)
          except BstError as e:
              # Report error and mark as failed
@@ -337,7 +332,7 @@ class Queue():
              self._done_queue.append(job)
              if success:
 -                if processed:
 +                if not job.skipped:
                      self.processed_elements.append(element)
                  else:
                      self.skipped_elements.append(element)

buildstream/_scheduler/queues/trackqueue.py

@@ -51,18 +51,11 @@ class TrackQueue(Queue):
      def done(self, _, element, result, success):
          if not success:
 -            return False
+-
 -        changed = False
 +            return
          # Set the new refs in the main process one by one as they complete
          for unique_id, new_ref in result:
              source = _plugin_lookup(unique_id)
 -            # We appear processed if at least one source has changed
 -            if source._save_ref(new_ref):
 -                changed = True
 +            source._save_ref(new_ref)
          element._tracking_done()
+-
 -        # We'll appear as a skipped element if tracking resulted in no change
 -        return changed

buildstream/element.py

@@ -1760,8 +1760,6 @@ class Element(Plugin):
              return False
          # Notify successfull download
 -        display_key = self._get_brief_display_key()
 -        self.info("Downloaded artifact {}".format(display_key))
          return True
      # _skip_push():
@@ -1800,16 +1798,13 @@ class Element(Plugin):
              self.warn("Not pushing tainted artifact.")
              return False
 -        display_key = self._get_brief_display_key()
 -        with self.timed_activity("Pushing artifact {}".format(display_key)):
 -            # Push all keys used for local commit
 -            pushed = self.__artifacts.push(self, self.__get_cache_keys_for_commit())
 -            if not pushed:
 -                return False
 +        # Push all keys used for local commit
 +        pushed = self.__artifacts.push(self, self.__get_cache_keys_for_commit())
 +        if not pushed:
 +            return False
 -            # Notify successful upload
 -            self.info("Pushed artifact {}".format(display_key))
 -            return True
 +        # Notify successful upload
 +        return True
      # _shell():
+     #

setup.py

@@ -297,7 +297,6 @@ setup(name='BuildStream',
            'ruamel.yaml < 0.15.52',
            'pluginbase',
            'Click',
 -          'blessings >= 1.6',
            'jinja2 >= 2.10',
            'protobuf >= 3.5',
            'grpcio >= 1.10',

tests/frontend/pull.py

@@ -356,4 +356,5 @@ def test_pull_missing_notifies_user(caplog, cli, tmpdir, datafiles):
          assert not result.get_pulled_elements(), \
              "No elements should have been pulled since the cache was empty"
 -        assert "SKIPPED Remote ({}) does not have".format(share.repo) in result.stderr
 +        assert "INFO    Remote ({}) does not have".format(share.repo) in result.stderr
 +        assert "SKIPPED Pull" in result.stderr

tests/frontend/push.py

@@ -386,3 +386,26 @@ def test_push_cross_junction(cli, tmpdir, datafiles):
          cache_key = cli.get_element_key(project, 'junction.bst:import-etc.bst')
          assert share.has_artifact('subtest', 'import-etc.bst', cache_key)
++
++
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_push_already_cached(caplog, cli, tmpdir, datafiles):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
 +    caplog.set_level(1)
++
 +    with create_artifact_share(os.path.join(str(tmpdir), 'artifactshare')) as share:
++
 +        cli.configure({
 +            'artifacts': {'url': share.repo, 'push': True}
 +        })
 +        result = cli.run(project=project, args=['build', 'target.bst'])
++
 +        result.assert_success()
 +        assert "SKIPPED Push" not in result.stderr
++
 +        result = cli.run(project=project, args=['push', 'target.bst'])
++
 +        result.assert_success()
 +        assert not result.get_pushed_elements(), "No elements should have been pushed since the cache was populated"
 +        assert "INFO    Remote ({}) already has ".format(share.repo) in result.stderr
 +        assert "SKIPPED Push" in result.stderr

tests/testutils/runcli.py

@@ -178,7 +178,7 @@ class Result():
          return list(pushed)
      def get_pulled_elements(self):
 -        pulled = re.findall(r'\[\s*pull:(\S+)\s*\]\s*INFO\s*Downloaded artifact', self.stderr)
 +        pulled = re.findall(r'\[\s*pull:(\S+)\s*\]\s*INFO\s*Pulled artifact', self.stderr)
          if pulled is None:
              return []

[Notes] [Git][BuildStream/buildstream][tristan/remove-blessings] 7 commits: Rework Skipped usage

Tristan Van Berkom pushed to branch tristan/remove-blessings at BuildStream / buildstream

Commits:

16 changed files:

Changes: