[Notes] [Git][BuildStream/buildstream][master] 5 commits:

Tristan Van Berkom pushed to branch master at BuildStream / buildstream

Commits:

3e36e363

by Tristan Van Berkom at 2019-01-24T16:55:24Z

_scheduler/resources.py: Dont error out in unregister_exclusive_interest()

Don't require the interest to be registered, just discard any interest,
this function just sets a bit in a mask, and is not intended to maintain
a balance like the reserve() function is.

ce01f87e

by Tristan Van Berkom at 2019-01-24T16:55:24Z

_scheduler/scheduler.py: Run cache size exclusively at startup

When running any session that has Queues which require Resource.CACHE,
check if our loaded estimated size exceeds the quota, and if so;
lock the Resource.CACHE resource exclusively right away and run
an exclusive initial cache size job.

This ensures we cleanup first before doing anything which might
add to the cache at startup time, if deemed needed.

This is a partial fix for issue #737

2479e8df

by Tristan Van Berkom at 2019-01-24T16:55:24Z

_frontend/widget.py: Render core messages more like other messages

In order to test when core activities occur by parsing the stderr
in tests, we should make the messages conform more.

At the same time, this restores alignment of columns in core
messages with the element processing related messages.

Also, _scheduler/scheduler.py is updated to make it's activity names
conform to the (current) 5 character limit for the sake of alignment.

The tests/frontend/logging.py test gets it's regexes updated for
the log lines it checks for in stderr.

fdb8ff65

by Tristan Van Berkom at 2019-01-24T16:55:24Z

tests/artifactcache/expiry.py: Test that expiry happens first

acd0bf22

by Tristan Van Berkom at 2019-01-24T18:01:41Z

Merge branch 'tristan/cache-management' into 'master'

Cache management fixes

See merge request BuildStream/buildstream!1091

Changes:

buildstream/_frontend/widget.py

@@ -175,29 +175,22 @@ class TypeName(Widget):
  # A widget for displaying the Element name
  class ElementName(Widget):
 -    def __init__(self, context, content_profile, format_profile):
 -        super(ElementName, self).__init__(context, content_profile, format_profile)
+-
 -        # Pre initialization format string, before we know the length of
 -        # element names in the pipeline
 -        self._fmt_string = '{: <30}'
+-
      def render(self, message):
 +        action_name = message.action_name
          element_id = message.task_id or message.unique_id
 -        if element_id is None:
 -            return ""
+-
 -        plugin = _plugin_lookup(element_id)
 -        name = plugin._get_full_name()
 +        if element_id is not None:
 +            plugin = _plugin_lookup(element_id)
 +            name = plugin._get_full_name()
 +            name = '{: <30}'.format(name)
 +        else:
 +            name = 'core activity'
 +            name = '{: <30}'.format(name)
 -        # Sneak the action name in with the element name
 -        action_name = message.action_name
          if not action_name:
              action_name = "Main"
          return self.content_profile.fmt("{: >5}".format(action_name.lower())) + \
 -            self.format_profile.fmt(':') + \
 -            self.content_profile.fmt(self._fmt_string.format(name))
 +            self.format_profile.fmt(':') + self.content_profile.fmt(name)
  # A widget for displaying the primary message text
@@ -219,9 +212,12 @@ class CacheKey(Widget):
      def render(self, message):
          element_id = message.task_id or message.unique_id
 -        if element_id is None or not self._key_length:
 +        if not self._key_length:
              return ""
 +        if element_id is None:
 +            return ' ' * self._key_length
++
          missing = False
          key = ' ' * self._key_length
          plugin = _plugin_lookup(element_id)

buildstream/_scheduler/resources.py

@@ -163,4 +163,4 @@ class Resources():
      def unregister_exclusive_interest(self, resources, source):
          for resource in resources:
 -            self._exclusive_resources[resource].remove(source)
 +            self._exclusive_resources[resource].discard(source)

buildstream/_scheduler/scheduler.py

@@ -40,8 +40,8 @@ class SchedStatus():
  # Some action names for the internal jobs we launch
+ #
 -_ACTION_NAME_CLEANUP = 'cleanup'
 -_ACTION_NAME_CACHE_SIZE = 'cache_size'
 +_ACTION_NAME_CLEANUP = 'clean'
 +_ACTION_NAME_CACHE_SIZE = 'size'
  # Scheduler()
@@ -151,6 +151,9 @@ class Scheduler():
          # Handle unix signals while running
          self._connect_signals()
 +        # Check if we need to start with some cache maintenance
 +        self._check_cache_management()
++
          # Run the queues
          self._sched()
          self.loop.run_forever()
@@ -272,6 +275,31 @@ class Scheduler():
      #                  Local Private Methods              #
      #######################################################
 +    # _check_cache_management()
 +    #
 +    # Run an initial check if we need to lock the cache
 +    # resource and check the size and possibly launch
 +    # a cleanup.
 +    #
 +    # Sessions which do not add to the cache are not affected.
 +    #
 +    def _check_cache_management(self):
++
 +        # Only trigger the check for a scheduler run which has
 +        # queues which require the CACHE resource.
 +        if not any(q for q in self.queues
 +                   if ResourceType.CACHE in q.resources):
 +            return
++
 +        # If the estimated size outgrows the quota, queue a job to
 +        # actually check the real cache size initially, this one
 +        # should have exclusive access to the cache to ensure nothing
 +        # starts while we are checking the cache.
 +        #
 +        artifacts = self.context.artifactcache
 +        if artifacts.has_quota_exceeded():
 +            self._sched_cache_size_job(exclusive=True)
++
      # _spawn_job()
+     #
      # Spanws a job
@@ -292,6 +320,11 @@ class Scheduler():
          self._cache_size_running = None
          self.resources.release([ResourceType.CACHE, ResourceType.PROCESS])
 +        # Unregister the exclusive interest if there was any
 +        self.resources.unregister_exclusive_interest(
 +            [ResourceType.CACHE], 'cache-size'
 +        )
++
          # Schedule a cleanup job if we've hit the threshold
          if status != JobStatus.OK:
              return
@@ -344,11 +377,35 @@ class Scheduler():
      # Runs a cache size job if one is scheduled to run now and
      # sufficient recources are available.
+     #
 -    def _sched_cache_size_job(self):
 +    # Args:
 +    #    exclusive (bool): Run a cache size job immediately and
 +    #                      hold the ResourceType.CACHE resource
 +    #                      exclusively (used at startup).
 +    #
 +    def _sched_cache_size_job(self, *, exclusive=False):
++
 +        # The exclusive argument is not intended (or safe) for arbitrary use.
 +        if exclusive:
 +            assert not self._cache_size_scheduled
 +            assert not self._cache_size_running
 +            assert not self._active_jobs
 +            self._cache_size_scheduled = True
          if self._cache_size_scheduled and not self._cache_size_running:
 -            if self.resources.reserve([ResourceType.CACHE, ResourceType.PROCESS]):
 +            # Handle the exclusive launch
 +            exclusive_resources = set()
 +            if exclusive:
 +                exclusive_resources.add(ResourceType.CACHE)
 +                self.resources.register_exclusive_interest(
 +                    exclusive_resources, 'cache-size'
 +                )
++
 +            # Reserve the resources (with the possible exclusive cache resource)
 +            if self.resources.reserve([ResourceType.CACHE, ResourceType.PROCESS],
 +                                      exclusive_resources):
++
 +                # Update state and launch
                  self._cache_size_scheduled = False
                  self._cache_size_running = \
                      CacheSizeJob(self, _ACTION_NAME_CACHE_SIZE,

tests/artifactcache/expiry.py

@@ -18,6 +18,7 @@
+ #
  import os
 +import re
  from unittest import mock
  import pytest
@@ -425,3 +426,66 @@ def test_extract_expiry(cli, datafiles, tmpdir):
      assert os.path.isdir(refsdirtarget2)
      assert not os.path.exists(refsdirtarget)
++
++
 +# Ensures that when launching BuildStream with a full artifact cache,
 +# the cache size and cleanup jobs are run before any other jobs.
 +#
 +@pytest.mark.datafiles(DATA_DIR)
 +def test_cleanup_first(cli, datafiles, tmpdir):
 +    project = os.path.join(datafiles.dirname, datafiles.basename)
 +    element_path = 'elements'
 +    cache_location = os.path.join(project, 'cache', 'artifacts', 'ostree')
 +    checkout = os.path.join(project, 'checkout')
++
 +    cli.configure({
 +        'cache': {
 +            'quota': 10000000,
 +        }
 +    })
++
 +    # Create an element that uses almost the entire cache (an empty
 +    # ostree cache starts at about ~10KiB, so we need a bit of a
 +    # buffer)
 +    create_element_size('target.bst', project, element_path, [], 8000000)
 +    res = cli.run(project=project, args=['build', 'target.bst'])
 +    res.assert_success()
++
 +    assert cli.get_element_state(project, 'target.bst') == 'cached'
++
 +    # Now configure with a smaller quota, create a situation
 +    # where the cache must be cleaned up before building anything else.
 +    #
 +    # Fix the fetchers and builders just to ensure a predictable
 +    # sequence of events (although it does not effect this test)
 +    cli.configure({
 +        'cache': {
 +            'quota': 5000000,
 +        },
 +        'scheduler': {
 +            'fetchers': 1,
 +            'builders': 1
 +        }
 +    })
++
 +    # Our cache is now more than full, BuildStream
 +    create_element_size('target2.bst', project, element_path, [], 4000000)
 +    res = cli.run(project=project, args=['build', 'target2.bst'])
 +    res.assert_success()
++
 +    # Find all of the activity (like push, pull, fetch) lines
 +    results = re.findall(r'\[.*\]\[.*\]\[\s*(\S+):.*\]\s*START\s*.*\.log', res.stderr)
++
 +    # Don't bother checking the order of 'fetch', it is allowed to start
 +    # before or after the initial cache size job, runs in parallel, and does
 +    # not require ResourceType.CACHE.
 +    results.remove('fetch')
 +    print(results)
++
 +    # Assert the expected sequence of events
 +    assert results == ['size', 'clean', 'build']
++
 +    # Check that the correct element remains in the cache
 +    states = cli.get_element_states(project, ['target.bst', 'target2.bst'])
 +    assert states['target.bst'] != 'cached'
 +    assert states['target2.bst'] == 'cached'

tests/frontend/logging.py

@@ -41,7 +41,7 @@ def test_default_logging(cli, tmpdir, datafiles):
      result = cli.run(project=project, args=['source', 'fetch', element_name])
      result.assert_success()
 -    m = re.search(r"\[\d\d:\d\d:\d\d\]\[\]\[\] SUCCESS Checking sources", result.stderr)
 +    m = re.search(r"\[\d\d:\d\d:\d\d\]\[\s*\]\[.*\] SUCCESS Checking sources", result.stderr)
      assert(m is not None)
@@ -77,7 +77,7 @@ def test_custom_logging(cli, tmpdir, datafiles):
      result = cli.run(project=project, args=['source', 'fetch', element_name])
      result.assert_success()
 -    m = re.search(r"\d\d:\d\d:\d\d,\d\d:\d\d:\d\d.\d{6},\d\d:\d\d:\d\d,,,SUCCESS,Checking sources", result.stderr)
 +    m = re.search(r"\d\d:\d\d:\d\d,\d\d:\d\d:\d\d.\d{6},\d\d:\d\d:\d\d,\s*,.*,SUCCESS,Checking sources", result.stderr)
      assert(m is not None)

[Notes] [Git][BuildStream/buildstream][master] 5 commits: _scheduler/resources.py: Dont error out in unregister_exclusive_interest()

Tristan Van Berkom pushed to branch master at BuildStream / buildstream

Commits:

5 changed files:

Changes: