Fix first sweeper run

Author: artikz

cms/io/triggeredservice.py

@@ -212,12 +212,10 @@ def __init__(self, shard):
 
         self._executors = []
 
-        # Set up and spawn the sweeper.
-        #
-        # TODO: link to greenlet and react to its death.
         self._sweeper_start = None
         self._sweeper_event = Event()
-        gevent.spawn(self._sweeper_loop)
+        self._sweeper_started = False
+        self._sweeper_timeout = None
 
     def add_executor(self, executor):
         """Add an executor for the service.
@@ -264,21 +262,27 @@ def dequeue(self, operation):
         for executor in self._executors:
             executor.dequeue(operation)
 
-    def _sweeper_timeout(self):
-        """Return how frequently to run the sweeper loop.
+    def start_sweeper(self, timeout):
+        """Start sweeper loop with given timeout.
 
-        return (float|None): timeout in seconds, or None for no
-            sweeping.
+        timeout (float): timeout in seconds.
 
         """
-        return None
+        if not self._sweeper_started:
+            self._sweeper_started = True
+            self._sweeper_timeout = timeout
+
+            # TODO: link to greenlet and react to its death.
+            gevent.spawn(self._sweeper_loop)
+        else:
+            logger.warning("Service tried to start the sweeper loop twice.")
 
     def _sweeper_loop(self):
         """Regularly check for missed operations.
 
-        Run the sweep once every _sweeper_timeout() seconds but make
+        Run the sweep once every _sweeper_timeout seconds but make
         sure that no two sweeps run simultaneously. That is, start a
-        new sweep _sweeper_timeout() seconds after the previous one
+        new sweep _sweeper_timeout seconds after the previous one
         started or when the previous one finished, whatever comes
         last.
 
@@ -291,10 +295,6 @@ def _sweeper_loop(self):
         suppressed, because the loop must go on.
 
         """
-        # If the timeout is None, it means the subclass does not want
-        # a sweeper.
-        if self._sweeper_timeout() is None:
-            return
         while True:
             self._sweeper_start = monotonic_time()
             self._sweeper_event.clear()
@@ -306,7 +306,7 @@ def _sweeper_loop(self):
                              "operations.", exc_info=True)
 
             self._sweeper_event.wait(max(self._sweeper_start +
-                                         self._sweeper_timeout() -
+                                         self._sweeper_timeout -
                                          monotonic_time(), 0))
 
     def _sweep(self):

cms/service/EvaluationService.py

@@ -847,6 +847,7 @@ def __init__(self, shard, contest_id):
             ServiceCoord("ScoringService", 0))
 
         self.add_executor(EvaluationExecutor(self))
+        self.start_sweeper(117.0)
 
         self.add_timeout(self.check_workers_timeout, None,
                          EvaluationService.WORKER_TIMEOUT_CHECK_TIME
@@ -857,9 +858,6 @@ def __init__(self, shard, contest_id):
                          .total_seconds(),
                          immediately=False)
 
-    def _sweeper_timeout(self):
-        return 117.0
-
     def submission_enqueue_operations(self, submission, check_again=False):
         """Push in queue the operations required by a submission.
 

cms/service/PrintingService.py

@@ -231,15 +231,13 @@ def __init__(self, shard):
         self.file_cacher = FileCacher(self)
 
         self.add_executor(PrintingExecutor(self.file_cacher))
+        self.start_sweeper(61.0)
 
         if config.printer is None:
             logger.info("Printing is disabled, so the PrintingService is "
                         "idle.")
             return
 
-    def _sweeper_timeout(self):
-        return 61.0
-
     def _missing_operations(self):
         """Enqueue unprinted print jobs.
 

cms/service/ProxyService.py

@@ -261,13 +261,11 @@ def __init__(self, shard, contest_id):
         self.rankings = list()
         for ranking in config.rankings:
             self.add_executor(ProxyExecutor(ranking.encode('utf-8')))
+        self.start_sweeper(347.0)
 
         # Send some initial data to rankings.
         self.initialize()
 
-    def _sweeper_timeout(self):
-        return 347.0
-
     def _missing_operations(self):
         """Return a generator of data to be sent to the rankings..
 

cms/service/ScoringService.py

@@ -153,9 +153,7 @@ def __init__(self, shard):
             must_be_present=ranking_enabled)
 
         self.add_executor(ScoringExecutor(self.proxy_service))
-
-    def _sweeper_timeout(self):
-        return 347.0
+        self.start_sweeper(347.0)
 
     def _missing_operations(self):
         """Return a generator of unscored submission results.

cmstestsuite/unit_tests/io/triggeredservice_test.py

@@ -86,9 +86,6 @@ def __init__(self, shard, timeout):
     def add_missing_operation(self, operation):
         self._operations.append(operation)
 
-    def _sweeper_timeout(self):
-        return self._timeout
-
     def _missing_operations(self):
         counter = 0
         while self._operations != []:
@@ -114,6 +111,7 @@ def setUpService(self, timeout=None, batch=False):
         self.service = FakeTriggeredService(0, timeout)
         for notifier in self.notifiers:
             self.service.add_executor(FakeExecutor(notifier))
+        self.service.start_sweeper(timeout)
 
     def test_success(self):
         """Test a simple success case."""