buildstream/_artifactcache/cascache.py

@@ -228,7 +228,7 @@ class CASCache(ArtifactCache):
             try:
                 remote.init()
 
-                element.info("Pulling {} <- {}".format(element._get_brief_display_key(), remote.spec.url))
+                element.status("Pulling {} <- {}".format(element._get_brief_display_key(), remote.spec.url))
 
                 request = buildstream_pb2.GetReferenceRequest()
                 request.key = ref
@@ -250,11 +250,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
@@ -279,7 +276,7 @@ class CASCache(ArtifactCache):
         for remote in push_remotes:
             remote.init()
             skipped_remote = True
-            element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
+            element.status("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
 
             try:
                 for ref in refs:
@@ -361,11 +358,8 @@ class CASCache(ArtifactCache):
                     raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
 
             if skipped_remote:
-                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

@@ -309,3 +309,17 @@ 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):
+    def __init__(self, *, detail=""):
+        super().__init__()
+        self._detail = detail
+
+    def __str__(self):
+        return self._detail

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
@@ -275,6 +276,10 @@ class Job():
     def set_task_id(self, task_id):
         self._task_id = task_id
 
+    @property
+    def skipped(self):
+        return self._skipped_flag
+
     #######################################################
     #                  Abstract Methods                   #
     #######################################################
@@ -396,6 +401,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
@@ -441,6 +453,7 @@ class Job():
                 self.message(MessageType.SUCCESS, self.action_name, elapsed=elapsed,
                              logfile=filename)
 
+                # XXX Verify below.
                 # Shutdown needs to stay outside of the above context manager,
                 # make sure we dont try to handle SIGTERM while the process
                 # is already busy in sys.exit()
@@ -547,14 +560,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)
 
     # _parent_process_envelope()
     #

buildstream/_scheduler/queues/buildqueue.py

@@ -54,12 +54,7 @@ class BuildQueue(Queue):
                           detail=detail, action_name=self.action_name,
                           elapsed=timedelta(seconds=0),
                           logfile=logfile)
-            job = ElementJob(self._scheduler, self.action_name,
-                             logfile, element=element, queue=self,
-                             resources=self.resources,
-                             action_cb=self.process,
-                             complete_cb=self._job_done,
-                             max_retries=self._max_retries)
+            job = self._create_job(element)
             self._done_queue.append(job)
             self.failed_elements.append(element)
             self._scheduler._job_complete_callback(job, False)
@@ -109,5 +104,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)
-
-        return True

buildstream/_scheduler/queues/fetchqueue.py

@@ -70,13 +70,8 @@ class FetchQueue(Queue):
         return QueueStatus.READY
 
     def done(self, _, element, result, success):
-
-        if not success:
-            return False
-
+        if success:
             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,12 @@ class PullQueue(Queue):
 
     def process(self, element):
         # returns whether an artifact was downloaded or not
-        return element._pull()
+        pulled = element._pull()
+
+        if not pulled:
+            raise SkipJob(detail=self.action_name)
+
+        return pulled
 
     def status(self, element):
         # state of dependencies may have changed, recalculate element state
@@ -53,17 +59,10 @@ class PullQueue(Queue):
             return QueueStatus.SKIP
 
     def done(self, _, element, result, success):
-
-        if not success:
-            return False
-
+        if success:
             element._pull_done()
 
             # Build jobs will check the "approximate" size first. Since we
             # do not get an artifact size from pull jobs, we have to
             # actually check the cache size.
             self._scheduler._check_cache_size_real()
-
-        # 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,15 @@ class PushQueue(Queue):
 
     def process(self, element):
         # returns whether an artifact was uploaded or not
-        return element._push()
+        pushed = element._push()
+
+        if not pushed:
+            raise SkipJob(detail=self.action_name)
+
+        return pushed
 
     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
 
@@ -158,20 +154,8 @@ class Queue():
         if not elts:
             return
 
-        # Note: The internal lists work with jobs. This is not
-        #       reflected in any external methods (except
-        #       pop/peek_ready_jobs).
-        def create_job(element):
-            logfile = self._element_log_path(element)
-            return ElementJob(self._scheduler, self.action_name,
-                              logfile, element=element, queue=self,
-                              resources=self.resources,
-                              action_cb=self.process,
-                              complete_cb=self._job_done,
-                              max_retries=self._max_retries)
-
         # Place skipped elements directly on the done queue
-        jobs = [create_job(elt) for elt in elts]
+        jobs = [self._create_job(elt) for elt in elts]
         skip = [job for job in jobs if self.status(job.element) == QueueStatus.SKIP]
         wait = [job for job in jobs if job not in skip]
 
@@ -308,8 +292,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
@@ -339,7 +322,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)
@@ -360,3 +343,15 @@ class Queue():
         logfile = "{key}-{action}".format(key=key, action=action)
 
         return os.path.join(project.name, element.normal_name, logfile)
+
+    # Note: The internal lists work with jobs. This is not
+    #       reflected in any external methods (except
+    #       pop/peek_ready_jobs).
+    def _create_job(self, element):
+        logfile = self._element_log_path(element)
+        return ElementJob(self._scheduler, self.action_name,
+                          logfile, element=element, queue=self,
+                          resources=self.resources,
+                          action_cb=self.process,
+                          complete_cb=self._job_done,
+                          max_retries=self._max_retries)

buildstream/_scheduler/queues/trackqueue.py

@@ -49,20 +49,10 @@ class TrackQueue(Queue):
         return QueueStatus.READY
 
     def done(self, _, element, result, success):
-
-        if not success:
-            return False
-
-        changed = False
-
+        if success:
             # 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

@@ -1746,7 +1746,7 @@ class Element(Plugin):
 
         # Notify successfull download
         display_key = self._get_brief_display_key()
-        self.info("Downloaded artifact {}".format(display_key))
+        self.info("Pulled artifact {}".format(display_key))
         return True
 
     # _skip_push():
@@ -1785,14 +1785,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
 
         # Notify successful upload
+        display_key = self._get_brief_display_key()
         self.info("Pushed artifact {}".format(display_key))
         return True
 

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 []