Add mirror tasks to stream

9 files changed, 83 insertions, 3 deletions

diff --git a/buildstream/_context.py b/buildstream/_context.py
index c0d49b245..e38cc76e1 100644
--- a/buildstream/_context.py
+++ b/buildstream/_context.py
@@ -95,6 +95,9 @@ class Context():
         # Maximum number of push tasks
         self.sched_pushers = 4
 
+        # Maximum number of mirror tasks
+        self.sched_mirrorers = 4
+
         # Maximum number of retries for network tasks
         self.sched_network_retries = 2
 
@@ -186,12 +189,14 @@ class Context():
         scheduler = _yaml.node_get(defaults, Mapping, 'scheduler')
         _yaml.node_validate(scheduler, [
             'on-error', 'fetchers', 'builders',
-            'pushers', 'network-retries'
+            'pushers', 'network-retries',
+            'mirrorers'
         ])
         self.sched_error_action = _yaml.node_get(scheduler, str, 'on-error')
         self.sched_fetchers = _yaml.node_get(scheduler, int, 'fetchers')
         self.sched_builders = _yaml.node_get(scheduler, int, 'builders')
         self.sched_pushers = _yaml.node_get(scheduler, int, 'pushers')
+        self.sched_mirrorers = _yaml.node_get(scheduler, int, 'mirrorers')
         self.sched_network_retries = _yaml.node_get(scheduler, int, 'network-retries')
 
         # Load per-projects overrides
diff --git a/buildstream/_frontend/app.py b/buildstream/_frontend/app.py
index fa07a9a54..d6af94ade 100644
--- a/buildstream/_frontend/app.py
+++ b/buildstream/_frontend/app.py
@@ -193,6 +193,7 @@ class App():
             'fetchers': 'sched_fetchers',
             'builders': 'sched_builders',
             'pushers': 'sched_pushers',
+            'mirrorers': 'sched_mirrorers',
             'network_retries': 'sched_network_retries'
         }
         for cli_option, context_attr in override_map.items():
diff --git a/buildstream/_frontend/widget.py b/buildstream/_frontend/widget.py
index fe7229e8a..b0ea106cf 100644
--- a/buildstream/_frontend/widget.py
+++ b/buildstream/_frontend/widget.py
@@ -468,6 +468,7 @@ class LogLine(Widget):
         values["Maximum Fetch Tasks"] = context.sched_fetchers
         values["Maximum Build Tasks"] = context.sched_builders
         values["Maximum Push Tasks"] = context.sched_pushers
+        values["Maximum Mirror Tasks"] = context.sched_mirrorers
         values["Maximum Network Retries"] = context.sched_network_retries
         text += self._format_values(values)
         text += '\n'
diff --git a/buildstream/_scheduler/__init__.py b/buildstream/_scheduler/__init__.py
index 14cdebf8e..b5074b92e 100644
--- a/buildstream/_scheduler/__init__.py
+++ b/buildstream/_scheduler/__init__.py
@@ -25,5 +25,6 @@ from .trackqueue import TrackQueue
 from .buildqueue import BuildQueue
 from .pushqueue import PushQueue
 from .pullqueue import PullQueue
+from .mirrorqueue import MirrorQueue
 
 from .scheduler import Scheduler, SchedStatus
diff --git a/buildstream/_scheduler/mirrorqueue.py b/buildstream/_scheduler/mirrorqueue.py
new file mode 100644
index 000000000..7d6226fca
--- /dev/null
+++ b/buildstream/_scheduler/mirrorqueue.py
@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+#
+#  Copyright (C) 2018 Codethink Limited
+#
+#  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/>.
+#
+#  Authors:
+#        Valentin David <valentin.david@codethink.co.uk>
+
+from . import Queue, QueueType, QueueStatus
+
+
+class MirrorQueue(Queue):
+
+    action_name = "Mirror"
+    complete_name = "Mirrored"
+    queue_type = QueueType.MIRROR
+
+    def process(self, element):
+        for source in element.sources():
+            source.update_mirror()
+
+    def status(self, element):
+        if not list(element.sources()):
+            return QueueStatus.SKIP
+
+        return QueueStatus.READY
+
+    def done(self, element, result, success):
+        return success
diff --git a/buildstream/_scheduler/queue.py b/buildstream/_scheduler/queue.py
index 7c4ad6919..4305cf4be 100644
--- a/buildstream/_scheduler/queue.py
+++ b/buildstream/_scheduler/queue.py
@@ -45,6 +45,9 @@ class QueueType():
     # Tasks which upload stuff to the internet
     PUSH = 3
 
+    # Tasks which mirror sources
+    MIRROR = 4
+
 
 # Queue status for a given element
 #
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index 25e1e6790..fa992b4ec 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -96,7 +96,8 @@ class Scheduler():
         self._job_tokens = {
             QueueType.FETCH: context.sched_fetchers,
             QueueType.BUILD: context.sched_builders,
-            QueueType.PUSH: context.sched_pushers
+            QueueType.PUSH: context.sched_pushers,
+            QueueType.MIRROR: context.sched_mirrorers
         }
 
     # run()
diff --git a/buildstream/_stream.py b/buildstream/_stream.py
index f2806b4c8..5ca146b03 100644
--- a/buildstream/_stream.py
+++ b/buildstream/_stream.py
@@ -30,7 +30,7 @@ from tempfile import TemporaryDirectory
 
 from ._exceptions import StreamError, ImplError, BstError, set_last_task_error
 from ._message import Message, MessageType
-from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
+from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue, MirrorQueue
 from ._pipeline import Pipeline, PipelineSelection
 from ._platform import Platform
 from . import utils, _yaml, _site
@@ -213,6 +213,30 @@ class Stream():
         self._enqueue_plan(elements)
         self._run()
 
+    # mirror()
+    #
+    # Mirrors all sources on the pipeline.
+    #
+    # Args:
+    #    targets (list of str): Targets to fetch
+    #    selection (PipelineSelection): The selection mode for the specified targets
+    #    except_targets (list of str): Specified targets to except from fetching
+    #
+    def mirror(self, targets, *,
+               selection=PipelineSelection.PLAN,
+               except_targets=None):
+        elements, _ = \
+            self._load(targets, (),
+                       selection=selection,
+                       except_targets=except_targets,
+                       track_except_targets=except_targets,
+                       fetch_subprojects=True)
+
+        mirrorqueue = MirrorQueue(self._scheduler)
+        self._add_queue(mirrorqueue)
+        self._enqueue_plan(elements, queue=mirrorqueue)
+        self._run()
+
     # fetch()
     #
     # Fetches sources on the pipeline.
diff --git a/buildstream/data/userconfig.yaml b/buildstream/data/userconfig.yaml
index 6bb54ff9e..ce87ef8db 100644
--- a/buildstream/data/userconfig.yaml
+++ b/buildstream/data/userconfig.yaml
@@ -36,6 +36,9 @@ scheduler:
   # Maximum number of simultaneous artifact uploading tasks.
   pushers: 4
 
+  # Maximum number of simultaneous mirror tasks.
+  mirrorers: 10
+
   # Maximum number of retries for network tasks.
   network-retries: 2