Merge pull request #184 from buxx/166/dev/signals

tracim/tracim/config/app_cfg.py

@@ -14,6 +14,7 @@ convert them into boolean, for example, you should use the
 """
 import imp
 import importlib
+import os
 from urllib.parse import urlparse
 
 import tg
@@ -22,6 +23,7 @@ from tg.configuration.milestones import environment_loaded
 
 from tgext.pluggable import plug
 from tgext.pluggable import replace_template
+from tracim.lib.system import InterruptManager
 
 from tracim.lib.utils import lazy_ugettext as l_
 
@@ -109,6 +111,7 @@ def start_daemons(manager: DaemonsManager):
         manager.run('mail_sender', MailSenderDaemon)
 
 environment_loaded.register(lambda: start_daemons(daemons))
+interrupt_manager = InterruptManager(os.getpid(), daemons_manager=daemons)
 
 # Note: here are fake translatable strings that allow to translate messages for reset password email content
 duplicated_email_subject = l_('Password reset request')

tracim/tracim/lib/daemons.py

@@ -1,9 +1,8 @@
 import threading
+import collections
 from configparser import DuplicateSectionError
-from wsgiref.simple_server import make_server
-import signal
 
-import collections
+from wsgiref.simple_server import make_server
 
 from radicale import Application as RadicaleApplication
 from radicale import HTTPServer as BaseRadicaleHTTPServer
@@ -13,16 +12,18 @@ from radicale import config as radicale_config
 from rq import Connection as RQConnection
 from rq import Worker as BaseRQWorker
 from redis import Redis
+from rq.dummy import do_nothing
+from rq.worker import StopRequested
 
 from tracim.lib.base import logger
 from tracim.lib.exceptions import AlreadyRunningDaemon
-from tracim.lib.utils import add_signal_handler
+
+from tracim.lib.utils import get_rq_queue
 
 
 class DaemonsManager(object):
     def __init__(self):
         self._running_daemons = {}
-        add_signal_handler(signal.SIGTERM, self.stop_all)
 
     def run(self, name: str, daemon_class: object, **kwargs) -> None:
         """
@@ -61,7 +62,7 @@ class DaemonsManager(object):
             logger.info(self, 'Stopping daemon with name "{0}": OK'
                               .format(name))
 
-    def stop_all(self, *args, **kwargs) -> None:
+    def stop_all(self) -> None:
         """
         Stop all started daemons and wait for them.
         """
@@ -71,6 +72,10 @@ class DaemonsManager(object):
             daemon.stop()
 
         for name, daemon in self._running_daemons.items():
+            logger.info(
+                self,
+                'Stopping daemon "{0}" waiting confirmation'.format(name),
+            )
             daemon.join()
             logger.info(self, 'Stopping daemon "{0}" OK'.format(name))
 
@@ -158,7 +163,12 @@ class MailSenderDaemon(Daemon):
         pass
 
     def stop(self) -> None:
-        self.worker.request_stop('TRACIM STOP', None)
+        # When _stop_requested at False, tracim.lib.daemons.RQWorker
+        # will raise StopRequested exception in worker thread after receive a
+        # job.
+        self.worker._stop_requested = True
+        queue = get_rq_queue('mail_sender')
+        queue.enqueue(do_nothing)
 
     def run(self) -> None:
         from tracim.config.app_cfg import CFG
@@ -175,13 +185,19 @@ class MailSenderDaemon(Daemon):
 
 class RQWorker(BaseRQWorker):
     def _install_signal_handlers(self):
-        # TODO BS 20170126: RQ WWorker is designed to work in main thread
+        # RQ Worker is designed to work in main thread
         # So we have to disable these signals (we implement server stop in
-        # MailSenderDaemon.stop method). When bug
-        # https://github.com/tracim/tracim/issues/166 will be fixed, ensure
-        # This worker terminate correctly.
+        # MailSenderDaemon.stop method).
         pass
 
+    def dequeue_job_and_maintain_ttl(self, timeout):
+        # RQ Worker is designed to work in main thread, so we add behaviour
+        # here: if _stop_requested has been set to True, raise the standard way
+        # StopRequested exception to stop worker.
+        if self._stop_requested:
+            raise StopRequested()
+        return super().dequeue_job_and_maintain_ttl(timeout)
+
 
 class RadicaleHTTPSServer(TracimSocketServerMixin, BaseRadicaleHTTPSServer):
     pass

tracim/tracim/lib/email.py

@@ -6,37 +6,33 @@ from email.mime.text import MIMEText
 
 import typing
 from mako.template import Template
-from redis import Redis
-from rq import Queue
 from tg.i18n import ugettext as _
 
 from tracim.lib.base import logger
 from tracim.model import User
 
+from tracim.lib.utils import get_rq_queue
+
 
 def send_email_through(
-        send_callable: typing.Callable[[Message], None],
+        sendmail_callable: typing.Callable[[Message], None],
         message: Message,
 ) -> None:
     """
     Send mail encapsulation to send it in async or sync mode.
     TODO BS 20170126: A global mail/sender management should be a good
                       thing. Actually, this method is an fast solution.
-    :param send_callable: A callable who get message on first parameter
+    :param sendmail_callable: A callable who get message on first parameter
     :param message: The message who have to be sent
     """
     from tracim.config.app_cfg import CFG
     cfg = CFG.get_instance()
 
     if cfg.EMAIL_PROCESSING_MODE == CFG.CST.SYNC:
-        send_callable(message)
+        sendmail_callable(message)
     elif cfg.EMAIL_PROCESSING_MODE == CFG.CST.ASYNC:
-        queue = Queue('mail_sender', connection=Redis(
-            host=cfg.EMAIL_SENDER_REDIS_HOST,
-            port=cfg.EMAIL_SENDER_REDIS_PORT,
-            db=cfg.EMAIL_SENDER_REDIS_DB,
-        ))
-        queue.enqueue(send_callable, message)
+        queue = get_rq_queue('mail_sender')
+        queue.enqueue(sendmail_callable, message)
     else:
         raise NotImplementedError(
             'Mail sender processing mode {} is not implemented'.format(

tracim/tracim/lib/system.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+import os
+import signal
+
+from tracim.lib.daemons import DaemonsManager
+
+
+class InterruptManager(object):
+    """
+    Manager interruption of tracim components.
+
+    Stop all tracim daemons, then:
+
+    With a specific production server like uWSGI, we should use master
+    FIFO system to exit properly the program:
+    https://github.com/unbit/uwsgi/issues/849. But to be generic, we resend the
+    signal after intercept it.
+    """
+    def __init__(
+            self,
+            tracim_process_pid: int,
+            daemons_manager: DaemonsManager,
+    ) -> None:
+        """
+        :param tracim_process_pid: pid of tracim.
+        :param daemons_manager: Tracim daemons manager
+        """
+        self.daemons_manager = daemons_manager
+        self.tracim_process_pid = tracim_process_pid
+        self._install_sgnal_handlers()
+
+    def _install_sgnal_handlers(self) -> None:
+        """
+        Install signal handler to intercept SIGINT and SIGTERM signals
+        """
+        signal.signal(signal.SIGTERM, self.stop)
+        signal.signal(signal.SIGINT, self.stop)
+
+    def _remove_signal_handlers(self) -> None:
+        """
+        Remove installed signals to permit stop of main thread.
+        """
+        signal.signal(signal.SIGTERM, signal.SIG_DFL)
+        signal.signal(signal.SIGINT, signal.SIG_DFL)
+
+    def stop(self, signum, frame) -> None:
+        """
+        Run stopping process needed when tracim have to stop.
+        :param signum: signal interruption value
+        :param frame: frame of signal origin
+        """
+        self._remove_signal_handlers()
+        self.daemons_manager.stop_all()
+        os.kill(self.tracim_process_pid, signum)

tracim/tracim/lib/utils.py

@@ -13,6 +13,8 @@ from tg.i18n import ugettext
 from tg.support.registry import StackedObjectProxy
 from tg.util import LazyString as BaseLazyString
 from tg.util import lazify
+from redis import Redis
+from rq import Queue
 
 from tracim.lib.base import logger
 from webob import Response
@@ -63,20 +65,6 @@ def NotImplemented():
     raise NotImplementedError()
 
 
-def add_signal_handler(signal_id, handler) -> None:
-    """
-    Add a callback attached to python signal.
-    :param signal_id: signal identifier (eg. signal.SIGTERM)
-    :param handler: callback to execute when signal trig
-    """
-    def _handler(*args, **kwargs):
-        handler()
-        signal.signal(signal_id, signal.SIG_DFL)
-        os.kill(os.getpid(), signal_id)  # Rethrow signal
-
-    signal.signal(signal_id, _handler)
-
-
 class APIWSGIHTTPException(WSGIHTTPException):
     def json_formatter(self, body, status, title, environ):
         if self.comment:
@@ -171,3 +159,18 @@ def _lazy_ugettext(text: str):
         return text
 
 lazy_ugettext = lazify(_lazy_ugettext)
+
+
+def get_rq_queue(queue_name: str= 'default') -> Queue:
+    """
+    :param queue_name: name of queue
+    :return: wanted queue
+    """
+    from tracim.config.app_cfg import CFG
+    cfg = CFG.get_instance()
+
+    return Queue(queue_name, connection=Redis(
+        host=cfg.EMAIL_SENDER_REDIS_HOST,
+        port=cfg.EMAIL_SENDER_REDIS_PORT,
+        db=cfg.EMAIL_SENDER_REDIS_DB,
+    ))