Revert "[Jormungandr] Use transient socket for instance, pt planner and asgard, AND remove socket reaper"

Author: Pierre-Etienne Bougué

source/jormungandr/jormungandr/__init__.py

@@ -32,10 +32,13 @@
 from __future__ import absolute_import, print_function, unicode_literals, division
 import logging
 import logging.config
+import os
 from flask import Flask, got_request_exception
 from flask_restful import Api
 from flask_caching import Cache
 from flask_cors import CORS
+import sys
+import six
 from jormungandr import init
 
 app = Flask(__name__)  # type: Flask
@@ -123,4 +126,7 @@ def setup_package():
     i_manager.stop()
 
 
+from jormungandr import transient_socket
+
+transient_socket.TransientSocket.init_socket_reaper(app.config)
 i_manager.is_ready = True

source/jormungandr/jormungandr/default_settings.py

@@ -11,10 +11,6 @@
 # path of the configuration file for each instances
 INSTANCES_DIR = os.getenv('JORMUNGANDR_INSTANCES_DIR', '/etc/jormungandr.d')
 
-
-INSTANCE_TIMEOUT = float(os.getenv('JORMUNGANDR_INSTANCE_TIMEOUT_S', 10))
-INSTANCE_FAST_TIMEOUT = float(os.getenv('JORMUNGANDR_INSTANCE_FAST_TIMEOUT_S', 1))
-
 # Patern that matches Jormungandr configuration files
 #  ex: '*.json' will match all json files within "INSTANCES_DIR" directory
 INSTANCES_FILENAME_PATTERN = os.getenv('JORMUNGANDR_INSTANCES_FILENAME_PATTERN', '*.json')
@@ -223,13 +219,12 @@
 MAX_JOURNEYS_CALLS = int(os.getenv('JORMUNGANDR_MAX_JOURNEYS_CALLS', 20))
 DEFAULT_AUTOCOMPLETE_BACKEND = os.getenv('JORMUNGANDR_DEFAULT_AUTOCOMPLETE_BACKEND', 'bragi')
 
-
-# ZMQ
+ZMQ_SOCKET_TTL_SECONDS = int(os.getenv('JORMUNGANDR_ZMQ_SOCKET_TTL_SECONDS', 10))
+ZMQ_SOCKET_REAPER_INTERVAL = int(os.getenv('JORMUNGANDR_ZMQ_SOCKET_REAPER_INTERVAL', 10))
 ZMQ_DEFAULT_SOCKET_TYPE = os.getenv('JORMUNGANDR_ZMQ_DEFAULT_SOCKET_TYPE', 'persistent')
 
-ZMQ_SOCKET_TTL_SECONDS = float(os.getenv('JORMUNGANDR_ZMQ_SOCKET_TTL_SECONDS', 10))
-ASGARD_ZMQ_SOCKET_TTL_SECONDS = float(os.getenv('JORMUNGANDR_ASGARD_ZMQ_SOCKET_TTL_SECONDS', 10))
-
+ASGARD_ZMQ_SOCKET_TTL_SECONDS = int(os.getenv('JORMUNGANDR_ASGARD_ZMQ_SOCKET_TTL_SECONDS', 30))
+ASGARD_ZMQ_SOCKET_REAPER_INTERVAL = int(os.getenv('JORMUNGANDR_ASGARD_ZMQ_SOCKET_REAPER_INTERVAL', 10))
 
 # Variable used only when deploying on aws
 ASGARD_ZMQ_SOCKET = os.getenv('JORMUNGANDR_ASGARD_ZMQ_SOCKET')

source/jormungandr/jormungandr/instance.py

@@ -65,7 +65,7 @@
 from jormungandr.equipments import EquipmentProviderManager
 from jormungandr.external_services import ExternalServiceManager
 from jormungandr.utils import can_connect_to_database
-from jormungandr import pt_planners_manager, transient_socket
+from jormungandr import pt_planners_manager
 
 type_to_pttype = {
     "stop_area": request_pb2.PlaceCodeRequest.StopArea,  # type: ignore
@@ -101,8 +101,9 @@ def _getter(self):
     return property(_getter)
 
 
-class Instance(transient_socket.TransientSocket):
+class Instance(object):
     name = None  # type: Text
+    _sockets = None  # type: Deque[Tuple[zmq.Socket, float]]
 
     def __init__(
         self,
@@ -121,15 +122,9 @@ def __init__(
         ghost_words=None,
         instance_db=None,
     ):
-        super(Instance, self).__init__(
-            name=name,
-            zmq_context=context,
-            zmq_socket=zmq_socket,
-            socket_ttl=app.config.get(str('ZMQ_SOCKET_TTL_SECONDS'), 10),
-        )
-
         self.geom = None
         self.geojson = None
+        self._sockets = deque()
         self.socket_path = zmq_socket
         self._scenario = None
         self._scenario_name = None
@@ -722,6 +717,66 @@ def places_proximity_radius(self):
         instance_db = self.get_models()
         return get_value_or_default('places_proximity_radius', instance_db, self.name)
 
+    def reap_socket(self, ttl):
+        # type: (int) -> None
+        if self.zmq_socket_type != 'transient':
+            return
+        logger = logging.getLogger(__name__)
+        now = time.time()
+
+        def _reap_sockets(connector):
+            while True:
+                try:
+                    socket, t = connector._sockets.popleft()
+                    if now - t > ttl:
+                        logger.debug("closing one socket for %s", connector.name)
+                        socket.setsockopt(zmq.LINGER, 0)
+                        socket.close()
+                    else:
+                        connector._sockets.appendleft((socket, t))
+                        break  # remaining socket are still in "keep alive" state
+                except IndexError:
+                    break
+
+        for _, planner in self._pt_planner_manager.get_all_pt_planners():
+            if planner.is_zmq_socket():
+                _reap_sockets(planner)
+        _reap_sockets(self)
+
+    @contextmanager
+    def socket(self, context):
+        sockets = self._sockets
+        socket_path = self.socket_path
+
+        t = None
+        try:
+            socket, t = sockets.pop()
+        except IndexError:  # there is no socket available: lets create one
+            start = time.time()
+            socket = context.socket(zmq.REQ)
+            socket.connect(socket_path)
+            logging.getLogger(__name__).info(
+                "it took %s ms to open a instance socket of %s during a request",
+                '%.2e' % ((time.time() - start) * 1000),
+                self.name,
+            )
+
+        try:
+            yield socket
+        finally:
+            if not socket.closed:
+                if t is not None and time.time() - t > app.config.get("ZMQ_SOCKET_TTL_SECONDS", 10):
+                    start = time.time()
+                    socket.setsockopt(zmq.LINGER, 0)
+                    socket.close()
+                    logging.getLogger(__name__).info(
+                        "it took %s ms to close a instance socket in %s",
+                        '%.2e' % ((time.time() - start) * 1000),
+                        self.name,
+                    )
+                else:
+                    sockets.append((socket, t or time.time()))
+
     def send_and_receive(self, *args, **kwargs):
         """
         encapsulate all call to kraken in a circuit breaker, this way we don't loose time calling dead instance
@@ -731,25 +786,37 @@ def send_and_receive(self, *args, **kwargs):
         except pybreaker.CircuitBreakerError as e:
             raise DeadSocketException(self.name, self.socket_path)
 
-    def _send_and_receive(self, request, timeout=app.config.get('INSTANCE_TIMEOUT', 10), quiet=False, **kwargs):
-        deadline = datetime.utcnow() + timedelta(milliseconds=timeout * 1000)
+    def _send_and_receive(
+        self, request, timeout=app.config.get('INSTANCE_TIMEOUT', 10000), quiet=False, **kwargs
+    ):
+        logger = logging.getLogger(__name__)
+        deadline = datetime.utcnow() + timedelta(milliseconds=timeout)
         request.deadline = deadline.strftime('%Y%m%dT%H%M%S,%f')
 
-        if 'request_id' in kwargs and kwargs['request_id']:
-            request.request_id = kwargs['request_id']
-        else:
-            try:
-                request.request_id = flask.request.id
-            except RuntimeError:
-                # we aren't in a flask context, so there is no request
-                if 'flask_request_id' in kwargs and kwargs['flask_request_id']:
-                    request.request_id = kwargs['flask_request_id']
-
-        pb = self.call(request.SerializeToString(), timeout, debug_cb=lambda: six.text_type(request))
-        resp = response_pb2.Response()
-        resp.ParseFromString(pb)
-        self.update_property(resp)
-        return resp
+        with self.socket(self.context) as socket:
+            if 'request_id' in kwargs and kwargs['request_id']:
+                request.request_id = kwargs['request_id']
+            else:
+                try:
+                    request.request_id = flask.request.id
+                except RuntimeError:
+                    # we aren't in a flask context, so there is no request
+                    if 'flask_request_id' in kwargs and kwargs['flask_request_id']:
+                        request.request_id = kwargs['flask_request_id']
+
+            socket.send(request.SerializeToString())
+            if socket.poll(timeout=timeout) > 0:
+                pb = socket.recv()
+                resp = response_pb2.Response()
+                resp.ParseFromString(pb)
+                self.update_property(resp)  # we update the timezone and geom of the instances at each request
+                return resp
+            else:
+                socket.setsockopt(zmq.LINGER, 0)
+                socket.close()
+                if not quiet:
+                    logger.error('request on %s failed: %s', self.socket_path, six.text_type(request))
+                raise DeadSocketException(self.name, self.socket_path)
 
     def get_id(self, id_):
         """
@@ -758,7 +825,7 @@ def get_id(self, id_):
         req = request_pb2.Request()
         req.requested_api = type_pb2.place_uri
         req.place_uri.uri = id_
-        return self.send_and_receive(req, timeout=app.config.get(str('INSTANCE_FAST_TIMEOUT'), 1))
+        return self.send_and_receive(req, timeout=app.config.get('INSTANCE_FAST_TIMEOUT', 1000))
 
     def has_id(self, id_):
         """
@@ -796,7 +863,7 @@ def get_external_codes(self, type_, id_):
         req.place_code.type_code = "external_code"
         req.place_code.code = id_
         # we set the timeout to 1s
-        return self.send_and_receive(req, timeout=app.config.get(str('INSTANCE_FAST_TIMEOUT'), 1))
+        return self.send_and_receive(req, timeout=app.config.get('INSTANCE_FAST_TIMEOUT', 1000))
 
     def has_external_code(self, type_, id_):
         """
@@ -852,7 +919,7 @@ def init(self):
         request_id = "instance_init"
         try:
             # we use _send_and_receive to avoid the circuit breaker, we don't want fast fail on init :)
-            resp = self._send_and_receive(req, request_id=request_id, timeout=1, quiet=True)
+            resp = self._send_and_receive(req, request_id=request_id, timeout=1000, quiet=True)
             # the instance is automatically updated on a call
             if self.publication_date != pub_date:
                 return True

source/jormungandr/jormungandr/instance_manager.py

@@ -100,7 +100,10 @@ def __init__(
         self.start_ping = start_ping
         self.instances = {}
         self.context = zmq.Context()
+        self.socket_ttl = app.config.get("ZMQ_SOCKET_TTL_SECONDS", 10)
+        self.reaper_interval = app.config.get("ZMQ_SOCKET_REAPER_INTERVAL", 10)
         self.is_ready = False
+        self.init_socket_reaper()
 
     def __repr__(self):
         return '<InstanceManager>'
@@ -206,6 +209,52 @@ def init_kraken_instances(self):
                 self._clear_cache()
                 break
 
+    def init_socket_reaper(self):
+        # start a greenlet that handle connection closing when idle
+        logging.getLogger(__name__).info("spawning a socket reaper with gevent")
+        gevent.spawn(self.socket_reaper_thread)
+
+        # Use uwsgi timer if we are running in uwsgi without gevent.
+        # When we are using uwsgi without gevent, idle workers won't run the greenlet, it will only
+        # be scheduled when waiting for a response of an external service (kraken mostly)
+        try:
+            import uwsgi
+
+            # In gevent mode we stop, no need to add a timer, the greenlet will be scheduled while waiting
+            # for incomming request.
+            if 'gevent' in uwsgi.opt:
+                return
+
+            logging.getLogger(__name__).info("spawning a socket reaper with  uwsgi timer")
+
+            # Register a signal handler for the signal 1 of uwsgi
+            # this signal will trigger the socket reaper and can be run by any worker
+            def reaper_timer(signal):
+                self.reap_sockets()
+
+            uwsgi.register_signal(1, 'active-workers', reaper_timer)
+            # Add a timer that trigger this signal every reaper_interval second
+            uwsgi.add_timer(1, self.reaper_interval)
+        except (ImportError, ValueError):
+            # ImportError is triggered if we aren't in uwsgi
+            # ValueError is raise if there is no more timer availlable: only 64 timers can be created
+            # workers that didn't create a timer can still run the signal handler
+            # if uwsgi dispatch the signal to them
+            # signal are dispatched randomly to workers (not round robbin :()
+            logging.getLogger(__name__).info(
+                "No more uwsgi timer available or not running in uwsgi, only gevent will be used"
+            )
+
+    def reap_sockets(self):
+        for instance in self.instances.values():
+            instance.reap_socket(self.socket_ttl)
+            gevent.idle(-1)  # request handling has the priority
+
+    def socket_reaper_thread(self, disable_gevent=False):
+        while True:
+            self.reap_sockets()
+            gevent.sleep(self.reaper_interval)
+
     def thread_ping(self, timer=10):
         """
         fetch krakens metadata