[Maposmatic-dev] [PATCH maposmatic 7/9] Rework the daemon to use the new JobRenderers

[Maxime Petazzoni]

With the newest additions to the job rendering module 'render', the
daemon can be vastly simplified. This also lays the groundwork for the
processing and/or subprocess based implementations to come.

As of now, the daemon is fully usable in production with the same level
of functionality as before.
---
 scripts/daemon.py |   90 +++++++++++++++++++++++++++++++++-------------------
 1 files changed, 57 insertions(+), 33 deletions(-)

diff --git a/scripts/daemon.py b/scripts/daemon.py
index bddc7ff..be8a2fb 100755
--- a/scripts/daemon.py
+++ b/scripts/daemon.py
@@ -23,26 +23,34 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
-import time
 import sys
 import threading
-import subprocess
+import time
 
 import render
 from www.maposmatic.models import MapRenderingJob
 from www.settings import LOG
 from www.settings import RENDERING_RESULT_PATH, RENDERING_RESULT_MAX_SIZE_GB
 
-RESULT_SUCCESSFULL = 'ok'
-RESULT_INTERRUPTED = 'rendering interrupted'
-RESULT_FAILED      = 'rendering failed'
-RESULT_CANCELED    = 'rendering took too long, canceled'
+_DEFAULT_CLEAN_FREQUENCY = 20       # Clean thread polling frequency, in
+                                    # seconds.
+_DEFAULT_POLL_FREQUENCY = 10        # Daemon job polling frequency, in seconds
+
+_RESULT_MSGS = {
+    render.RESULT_SUCCESS: 'ok',
+    render.RESULT_KEYBOARD_INTERRUPT: 'rendering interrupted',
+    render.RESULT_RENDERING_EXCEPTION: 'rendering failed',
+    render.RESULT_TIMEOUT_REACHED: 'rendering took too long, canceled'
+}
 
 class MapOSMaticDaemon:
     """
     This is a basic rendering daemon, base class for the different
     implementations of rendering scheduling. By default, it acts as a
     standalone, single-process MapOSMatic rendering daemon.
+
+    It of course uses the TimingOutJobRenderer to ensure no long-lasting job
+    stalls the queue.
     """
 
     def __init__(self, frequency):
@@ -74,31 +82,48 @@ class MapOSMaticDaemon:
         LOG.info("MapOSMatic rendering daemon terminating.")
 
     def dispatch(self, job):
-        """Dispatch the given job. In this simple single-process daemon, this
-        is as simple as rendering it."""
-        self.render(job)
+        """In this simple single-process daemon, dispatching is as easy as
+        calling the render() method. Subclasses probably want to overload this
+        method too and implement a more clever dispatching mechanism.
+
+        Args:
+            job (MapRenderingJob): the job to process and render.
+
+        Returns True if the rendering was successful, False otherwise.
+        """
+
+        return self.render(job)
 
     def render(self, job):
-        """Render the given job, handling the different rendering outcomes
-        (success or failures)."""
+        """Render the given job using a timing out job renderer.
+
+        Args:
+            job (MapRenderingJob): the job to process and render.
+
+        Returns True if the rendering was successful, False otherwise.
+        """
 
-        LOG.info("Rendering job #%d '%s'..." %
-                 (job.id, job.maptitle))
+        renderer = render.TimingOutJobRenderer(job)
         job.start_rendering()
+        ret = renderer.run()
+        job.end_rendering(_RESULT_MSGS[ret])
+        return ret == 0
 
-        ret = render.render_job(job)
-        if ret == 0:
-            msg = RESULT_SUCCESSFULL
-            LOG.info("Finished rendering of job #%d." % job.id)
-        elif ret == 1:
-            msg = RESULT_INTERRUPTED
-            LOG.info("Rendering of job #%d interrupted!" % job.id)
-        else:
-            msg = RESULT_FAILED
-            LOG.info("Rendering of job #%d failed (exception occurred)!" %
-                     job.id)
+class ProcessingMapOSMaticDaemon(MapOSMaticDaemon):
+    """
+    A processing.Pool based implementation of the MapOSMaticDaemon rendering
+    daemon. Each job is handled by a worker process of the pool.
+    """
+
+    def __init__(self, frequency, n_workers=None):
+        MapOSMaticDaemon.__init__(self, frequency)
+        import multiprocessing
 
-        job.end_rendering(msg)
+        self.__pool = multiprocessing.Pool(n_workers)
+
+    def dispatch(self, job):
+        # self.__pool.apply_async(self.render, job)
+        raise NotImplementedError
 
 class SubprocessMapOSMaticDaemon(MapOSMaticDaemon):
     """
@@ -108,16 +133,12 @@ class SubprocessMapOSMaticDaemon(MapOSMaticDaemon):
 
     def __init__(self, frequency, max_process=4):
         MapOSMaticDaemon.__init__(self, frequency)
+        import subprocess
+
         self.max_process = max_process
         self.n_process = 0
 
     def dispatch(self, job):
-        LOG.info("Forking to render job #%d '%s'..." %
-                 (job.id, job.maptitle))
-        job.startrendering()
-
-    def render(self, job):
-        # Call render.py
         raise NotImplementedError
 
 class RenderingsGarbageCollector(threading.Thread):
@@ -231,14 +252,17 @@ class RenderingsGarbageCollector(threading.Thread):
 
 
 if __name__ == '__main__':
+    clean_freq = _DEFAULT_CLEAN_FREQUENCY
+    poll_freq = _DEFAULT_POLL_FREQUENCY
+
     if (not os.path.exists(RENDERING_RESULT_PATH)
         or not os.path.isdir(RENDERING_RESULT_PATH)):
         LOG.error("%s does not exist or is not a directory! "
                   "Please use a valid RENDERING_RESULT_PATH.")
         sys.exit(1)
 
-    daemon = MapOSMaticDaemon(10)
-    cleaner = RenderingsGarbageCollector(20)
+    cleaner = RenderingsGarbageCollector(clean_freq)
+    daemon = MapOSMaticDaemon(poll_freq)
 
     cleaner.start()
     daemon.serve()
-- 
1.6.3.3.277.g88938c