diff options
| -rw-r--r-- | changes/feature_6360_use-txzmq-in-backend | 1 | ||||
| -rw-r--r-- | pkg/requirements.pip | 2 | ||||
| -rw-r--r-- | src/leap/bitmask/backend/backend.py | 128 | ||||
| -rw-r--r-- | src/leap/bitmask/backend/backend_proxy.py | 239 | ||||
| -rw-r--r-- | src/leap/bitmask/backend/components.py | 3 | ||||
| -rw-r--r-- | src/leap/bitmask/gui/app.py | 1 |
6 files changed, 205 insertions, 169 deletions
diff --git a/changes/feature_6360_use-txzmq-in-backend b/changes/feature_6360_use-txzmq-in-backend new file mode 100644 index 00000000..6703f02b --- /dev/null +++ b/changes/feature_6360_use-txzmq-in-backend @@ -0,0 +1 @@ +- Use txzmq in backend (closes #6360). diff --git a/pkg/requirements.pip b/pkg/requirements.pip index 37cc18a3..13d4da14 100644 --- a/pkg/requirements.pip +++ b/pkg/requirements.pip @@ -9,6 +9,7 @@ argparse requests>=1.1.0 srp>=1.0.2 pyopenssl +coloredlogs # This won't be needed after we refactor leap.common.events to use zmq. python-dateutil @@ -24,6 +25,7 @@ zope.proxy # You will want to install this bundled if you don't have sodium in your system: # pip install pyzmq --install-option="--zmq=bundled" pyzmq +txzmq leap.common>=0.3.7 leap.soledad.client>=0.6.0 diff --git a/src/leap/bitmask/backend/backend.py b/src/leap/bitmask/backend/backend.py index cff731ba..fcbb19dc 100644 --- a/src/leap/bitmask/backend/backend.py +++ b/src/leap/bitmask/backend/backend.py @@ -17,17 +17,16 @@ # FIXME this is missing module documentation. It would be fine to say a couple # of lines about the whole backend architecture. -# TODO use txzmq bindings instead. import json import os -import threading import time import psutil -from twisted.internet import defer, reactor, threads +from twisted.internet import defer, reactor, threads, task +import txzmq import zmq try: from zmq.auth.thread import ThreadAuthenticator @@ -43,43 +42,42 @@ import logging logger = logging.getLogger(__name__) -class Backend(object): +class TxZmqREPConnection(object): """ - Backend server. - Receives signals from backend_proxy and emit signals if needed. + A twisted based zmq rep connection. """ - # XXX we might want to make this configurable per-platform, - # and use the most performant socket type on each one. - if flags.ZMQ_HAS_CURVE: - # XXX this should not be hardcoded. Make it configurable. - PORT = '5556' - BIND_ADDR = "tcp://127.0.0.1:%s" % PORT - else: - SOCKET_FILE = "/tmp/bitmask.socket.0" - BIND_ADDR = "ipc://%s" % SOCKET_FILE - PING_INTERVAL = 2 # secs - - def __init__(self, frontend_pid=None): + def __init__(self, server_address, process_request): """ - Backend constructor, create needed instances. + Initialize the connection. + + :param server_address: The address of the backend zmq server. + :type server: str + :param process_request: A callable used to process incoming requests. + :type process_request: callable(messageParts) """ - self._signaler = Signaler() + self._server_address = server_address + self._process_request = process_request + self._zmq_factory = None + self._zmq_connection = None + self._init_txzmq() - self._frontend_pid = frontend_pid + def _init_txzmq(self): + """ + Configure the txzmq components and connection. + """ + self._zmq_factory = txzmq.ZmqFactory() + self._zmq_factory.registerForShutdown() + self._zmq_connection = txzmq.ZmqREPConnection(self._zmq_factory) - self._do_work = threading.Event() # used to stop the worker thread. - self._zmq_socket = None + context = self._zmq_factory.context + socket = self._zmq_connection.socket - self._ongoing_defers = [] - self._init_zmq() + def _gotMessage(messageId, messageParts): + self._zmq_connection.reply(messageId, "OK") + self._process_request(messageParts) - def _init_zmq(self): - """ - Configure the zmq components and connection. - """ - context = zmq.Context() - socket = context.socket(zmq.REP) + self._zmq_connection.gotMessage = _gotMessage if flags.ZMQ_HAS_CURVE: # Start an authenticator for this context. @@ -95,37 +93,39 @@ class Backend(object): socket.curve_secretkey = secret socket.curve_server = True # must come before bind - socket.bind(self.BIND_ADDR) - if not flags.ZMQ_HAS_CURVE: - os.chmod(self.SOCKET_FILE, 0600) + proto, addr = self._server_address.split('://') # tcp/ipc, ip/socket + socket.bind(self._server_address) + if proto == 'ipc': + os.chmod(addr, 0600) - self._zmq_socket = socket - def _worker(self): - """ - Receive requests and send it to process. +class Backend(object): + """ + Backend server. + Receives signals from backend_proxy and emit signals if needed. + """ + # XXX we might want to make this configurable per-platform, + # and use the most performant socket type on each one. + if flags.ZMQ_HAS_CURVE: + # XXX this should not be hardcoded. Make it configurable. + PORT = '5556' + BIND_ADDR = "tcp://127.0.0.1:%s" % PORT + else: + SOCKET_FILE = "/tmp/bitmask.socket.0" + BIND_ADDR = "ipc://%s" % SOCKET_FILE - Note: we use a simple while since is less resource consuming than a - Twisted's LoopingCall. + PING_INTERVAL = 2 # secs + + def __init__(self, frontend_pid=None): """ - pid = self._frontend_pid - check_wait = 0 - while self._do_work.is_set(): - # Wait for next request from client - try: - request = self._zmq_socket.recv(zmq.NOBLOCK) - self._zmq_socket.send("OK") - # logger.debug("Received request: '{0}'".format(request)) - self._process_request(request) - except zmq.ZMQError as e: - if e.errno != zmq.EAGAIN: - raise - time.sleep(0.01) - - check_wait += 0.01 - if pid is not None and check_wait > self.PING_INTERVAL: - check_wait = 0 - self._check_frontend_alive() + Backend constructor, create needed instances. + """ + self._signaler = Signaler() + self._frontend_pid = frontend_pid + self._frontend_checker = None + self._ongoing_defers = [] + self._zmq_connection = TxZmqREPConnection( + self.BIND_ADDR, self._process_request) def _check_frontend_alive(self): """ @@ -160,25 +160,27 @@ class Backend(object): for d in self._ongoing_defers: d.cancel() + logger.debug("Stopping the Twisted reactor...") reactor.stop() - logger.debug("Twisted reactor stopped.") def run(self): """ Start the ZMQ server and run the loop to handle requests. """ self._signaler.start() - self._do_work.set() - threads.deferToThread(self._worker) + self._frontend_checker = task.LoopingCall(self._check_frontend_alive) + self._frontend_checker.start(self.PING_INTERVAL) + logger.debug("Starting Twisted reactor.") reactor.run() + logger.debug("Finished Twisted reactor.") def stop(self): """ Stop the server and the zmq request parse loop. """ - logger.debug("STOP received.") + logger.debug("Stopping the backend...") self._signaler.stop() - self._do_work.clear() + self._frontend_checker.stop() threads.deferToThread(self._stop_reactor) def _process_request(self, request_json): diff --git a/src/leap/bitmask/backend/backend_proxy.py b/src/leap/bitmask/backend/backend_proxy.py index 04046d3d..8d6930d6 100644 --- a/src/leap/bitmask/backend/backend_proxy.py +++ b/src/leap/bitmask/backend/backend_proxy.py @@ -21,11 +21,13 @@ to the backend. # XXX should document the relationship to the API here. import functools -import Queue import threading -import time import zmq +from zmq.eventloop import ioloop +from zmq.eventloop import zmqstream + +from taskthread import TimerTask from leap.bitmask.backend.api import API, STOP_REQUEST, PING_REQUEST from leap.bitmask.backend.settings import Settings @@ -37,35 +39,39 @@ import logging logger = logging.getLogger(__name__) -class BackendProxy(object): +class ZmqREQConnection(threading.Thread): """ - The BackendProxy handles calls from the GUI and forwards (through ZMQ) - to the backend. + A threaded zmq req connection. """ - if flags.ZMQ_HAS_CURVE: - PORT = '5556' - SERVER = "tcp://localhost:%s" % PORT - else: - SERVER = "ipc:///tmp/bitmask.socket.0" - - POLL_TIMEOUT = 4000 # ms - POLL_TRIES = 3 - - PING_INTERVAL = 2 # secs - - def __init__(self): - generate_zmq_certificates_if_needed() - - self._socket = None + def __init__(self, server_address, on_recv): + """ + Initialize the connection. - self.settings = Settings() + :param server_address: The address of the backend zmq server. + :type server: str + :param on_recv: The callback to be executed when a message is + received. + :type on_recv: callable(msg) + """ + threading.Thread.__init__(self) + self._server_address = server_address + self._on_recv = on_recv + self._stream = None + self._init_zmq() - # initialize ZMQ stuff: + def _init_zmq(self): + """ + Configure the zmq components and connection. + """ + logger.debug("Setting up ZMQ connection to server...") context = zmq.Context() - logger.debug("Connecting to server...") socket = context.socket(zmq.REQ) + # we use zmq's eventloop in order to asynchronously send requests + loop = ioloop.ZMQIOLoop.current() + self._stream = zmqstream.ZMQStream(socket, loop) + if flags.ZMQ_HAS_CURVE: # public, secret = zmq.curve_keypair() client_keys = zmq.curve_keypair() @@ -79,66 +85,128 @@ class BackendProxy(object): socket.setsockopt(zmq.RCVTIMEO, 1000) socket.setsockopt(zmq.LINGER, 0) # Terminate early - socket.connect(self.SERVER) - self._socket = socket - self._ping_at = 0 + self._stream.on_recv(self._on_recv) + + def run(self): + """ + Run the threaded stream connection loop. + """ + self._stream.socket.connect(self._server_address) + logger.debug("Starting ZMQ loop.") + self._stream.io_loop.start() + logger.debug("Finished ZMQ loop.") + + def stop(self): + """ + Stop the threaded connection loop. + """ + self._stream.io_loop.stop() + + def send(self, *args, **kwargs): + """ + Send a message through this connection. + """ + # Important note: calling send on the zmqstream from another + # thread doesn’t properly tell the IOLoop thread that there’s an + # event to process. This could cuase small delays if the IOLoop is + # already processing lots of events, but it can cause the message + # to never send if the zmq socket is the only one it’s handling. + # + # Because of that, we want ZmqREQConnection.send to hand off the + # stream.send to the IOLoop’s thread via IOLoop.add_callback: + self._stream.io_loop.add_callback( + lambda: self._stream.send(*args, **kwargs)) + + +class BackendProxy(object): + """ + The BackendProxy handles calls from the GUI and forwards (through ZMQ) + to the backend. + """ + + if flags.ZMQ_HAS_CURVE: + PORT = '5556' + SERVER = "tcp://localhost:%s" % PORT + else: + SERVER = "ipc:///tmp/bitmask.socket.0" + + PING_INTERVAL = 2 # secs + + def __init__(self): + """ + Initialize the backend proxy. + """ + generate_zmq_certificates_if_needed() + self._do_work = threading.Event() + self._work_lock = threading.Lock() + self._connection = ZmqREQConnection(self.SERVER, self._set_online) + self._heartbeat = TimerTask(self._ping, delay=self.PING_INTERVAL) + self._ping_event = threading.Event() self.online = False + self.settings = Settings() - self._call_queue = Queue.Queue() - self._worker_caller = threading.Thread(target=self._worker) + def _set_online(self, _): + """ + Mark the backend as being online. - def start(self): - self._worker_caller.start() + This is used as the zmq connection's on_recv callback, and so it is + passed the received message as a parameter. Because we currently don't + use that message, we just ignore it for now. + """ + self.online = True + # the following event is used when checking whether the backend is + # online + self._ping_event.set() + + def _set_offline(self): + """ + Mark the backend as being offline. + """ + self.online = False def check_online(self): """ Return whether the backend is accessible or not. You don't need to do `run` in order to use this. - :rtype: bool """ - # we use a small timeout in order to response quickly if the backend is - # offline - self._send_request(PING_REQUEST, retry=False, timeout=500) - self._socket.close() + logger.debug("Checking whether backend is online...") + self._send_request(PING_REQUEST) + # self._ping_event will eventually be set by the zmq connection's + # on_recv callback, so we use a small timeout in order to response + # quickly if the backend is offline + if not self._ping_event.wait(0.5): + logger.warning("Backend is offline!") + self._set_offline() return self.online - def _worker(self): + def start(self): """ - Worker loop that processes the Queue of pending requests to do. + Start the backend proxy. """ - while True: - try: - request = self._call_queue.get(block=False) - # break the loop after sending the 'stop' action to the - # backend. - if request == STOP_REQUEST: - break - - self._send_request(request) - except Queue.Empty: - pass - time.sleep(0.01) - self._ping() + logger.debug("Starting backend proxy...") + self._do_work.set() + self._connection.start() + self.check_online() + self._heartbeat.start() - logger.debug("BackendProxy worker stopped.") - - def _reset_ping(self): + def _stop(self): """ - Reset the ping timeout counter. - This is called for every ping and request. + Stop the backend proxy. """ - self._ping_at = time.time() + self.PING_INTERVAL + with self._work_lock: # avoid sending after connection was closed + self._do_work.clear() + self._heartbeat.stop() + self._connection.stop() + logger.debug("BackendProxy worker stopped.") def _ping(self): """ Heartbeat helper. Sends a PING request just to know that the server is alive. """ - if time.time() > self._ping_at: - self._send_request(PING_REQUEST) - self._reset_ping() + self._send_request(PING_REQUEST) def _api_call(self, *args, **kwargs): """ @@ -162,6 +230,8 @@ class BackendProxy(object): 'arguments': kwargs, } + request_json = None + try: request_json = zmq.utils.jsonapi.dumps(request) except Exception as e: @@ -172,12 +242,12 @@ class BackendProxy(object): raise # queue the call in order to handle the request in a thread safe way. - self._call_queue.put(request_json) + self._send_request(request_json) if api_method == STOP_REQUEST: - self._call_queue.put(STOP_REQUEST) + self._stop() - def _send_request(self, request, retry=True, timeout=None): + def _send_request(self, request): """ Send the given request to the server. This is used from a thread safe loop in order to avoid sending a @@ -185,49 +255,10 @@ class BackendProxy(object): :param request: the request to send. :type request: str - :param retry: whether we should retry or not in case of timeout. - :type retry: bool - :param timeout: a custom timeout (milliseconds) to wait for a response. - :type timeout: int """ - # logger.debug("Sending request to backend: {0}".format(request)) - self._socket.send(request) - - poll = zmq.Poller() - poll.register(self._socket, zmq.POLLIN) - - reply = None - - tries = 0 - if not retry: - tries = self.POLL_TRIES + 1 # this means: no retries left - - if timeout is None: - timeout = self.POLL_TIMEOUT - - while True: - socks = dict(poll.poll(timeout)) - if socks.get(self._socket) == zmq.POLLIN: - reply = self._socket.recv() - break - - tries += 1 - if tries < self.POLL_TRIES: - logger.warning('Retrying receive... {0}/{1}'.format( - tries, self.POLL_TRIES)) - else: - break - - if reply is None: - msg = "Timeout error contacting backend." - logger.critical(msg) - self.online = False - else: - # msg = "Received reply for '{0}' -> '{1}'".format(request, reply) - # logger.debug(msg) - self.online = True - # request received, no ping needed for other interval. - self._reset_ping() + with self._work_lock: # avoid sending after connection was closed + if self._do_work.is_set(): + self._connection.send(request) def __getattribute__(self, name): """ diff --git a/src/leap/bitmask/backend/components.py b/src/leap/bitmask/backend/components.py index 1efcda6c..0a2f1029 100644 --- a/src/leap/bitmask/backend/components.py +++ b/src/leap/bitmask/backend/components.py @@ -63,7 +63,8 @@ from leap.common import certs as leap_certs from leap.keymanager import openpgp -from leap.soledad.client.secrets import NoStorageSecret, PassphraseTooShort +from leap.soledad.client.secrets import PassphraseTooShort +from leap.soledad.client.secrets import NoStorageSecret logger = logging.getLogger(__name__) diff --git a/src/leap/bitmask/gui/app.py b/src/leap/bitmask/gui/app.py index 5fe031b1..46f276e1 100644 --- a/src/leap/bitmask/gui/app.py +++ b/src/leap/bitmask/gui/app.py @@ -20,7 +20,6 @@ and the signaler get signals from the backend. """ import logging -from functools import partial from PySide import QtCore, QtGui from leap.bitmask.config.leapsettings import LeapSettings |