Change in vdsm[master]: jsonrpc: Add support for multiple listeners

smizrahi at redhat.com smizrahi at redhat.com
Sat Jan 26 00:08:24 UTC 2013


Saggi Mizrahi has uploaded a new change for review.

Change subject: jsonrpc: Add support for multiple listeners
......................................................................

jsonrpc: Add support for multiple listeners

Make is so that a reactor can have more then one listeners running

Change-Id: Ibe1b392eab2c30df5720caca11a80f44c6135437
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M tests/jsonRpcTests.py
M vdsm_api/BindingJsonRpc.py
M vdsm_api/jsonrpc/__init__.py
M vdsm_api/jsonrpc/protonReactor.py
M vdsm_api/jsonrpc/tcpReactor.py
5 files changed, 201 insertions(+), 73 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/00/11400/1

diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index 52c1486..cf47048 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -76,10 +76,10 @@
 def _tcpServerConstructor(messageHandler):
     port = _getFreePort()
     address = ("localhost", port)
-    reactor = tcpReactor.TCPReactor(address, messageHandler)
+    reactor = tcpReactor.TCPReactor(messageHandler)
 
     try:
-        yield reactor, partial(TCPReactorClient, address)
+        yield reactor, partial(TCPReactorClient, address), address
     finally:
         reactor.stop()
 
@@ -91,10 +91,12 @@
 
     port = _getFreePort()
     serverAddress = "amqp://127.0.0.1:%d/vdsm_test" % (port,)
-    reactor = protonReactor.ProtonReactor(("127.0.0.1", port), messageHandler)
+    reactor = protonReactor.ProtonReactor(messageHandler)
 
     try:
-        yield reactor, partial(ProtonReactorClient, serverAddress)
+        yield (reactor,
+               partial(ProtonReactorClient, serverAddress),
+               ("127.0.0.1", port))
     finally:
         reactor.stop()
 
@@ -113,11 +115,11 @@
 @contextmanager
 def constructServer(tp, bridge):
     server = JsonRpcServer(bridge)
-    with constructReactor(tp, server) as (reactor, clientFactory):
-        reactor.start_listening()
+    with constructReactor(tp, server) as (reactor, clientFactory, laddr):
         t = threading.Thread(target=reactor.process_requests)
         t.setDaemon(True)
         t.start()
+        reactor.start_listening(laddr)
 
         t = threading.Thread(target=server.serve_requests)
         t.setDaemon(True)
@@ -270,11 +272,12 @@
                 self.fail("Server died: (%s) %s" % (type(e), e))
 
         with constructReactor(reactorType, msgHandler) as (reactor,
-                                                           clientFactory):
-            reactor.start_listening()
+                                                           clientFactory,
+                                                           laddr):
             t = threading.Thread(target=serve, args=(reactor,))
             t.setDaemon(True)
             t.start()
+            reactor.start_listening(laddr)
 
             clientNum = 1
             repeats = 1
diff --git a/vdsm_api/BindingJsonRpc.py b/vdsm_api/BindingJsonRpc.py
index 538311d..11738c5 100644
--- a/vdsm_api/BindingJsonRpc.py
+++ b/vdsm_api/BindingJsonRpc.py
@@ -38,54 +38,67 @@
     log = logging.getLogger('BindingJsonRpc')
 
     def __init__(self, bridge, backendConfig):
-        reactors = []
+        reactors = {}
         self.bridge = bridge
         self.server = JsonRpcServer(bridge, _simpleThreadFactory)
-        for backendType, cfg in backendConfig:
-            if backendType == "tcp":
-                reactors.append(self._createTcpReactor(cfg))
-            elif backendType == "amqp":
-                if ProtonReactor is None:
-                    continue
+        self._cfg = backendConfig
 
-                reactors.append(self._createProtonReactor(cfg))
+        for backendType, cfg in backendConfig:
+            if backendType not in reactors:
+                if backendType == "tcp":
+                    reactors["tcp"] = self._createTcpReactor()
+                elif backendType == "amqp":
+                    if ProtonReactor is None:
+                        continue
+
+                    reactors["amqp"] = self._createProtonReactor()
 
         self._reactors = reactors
 
-    def _createTcpReactor(self, cfg):
+    def _createTcpListener(self, cfg):
         address = cfg.get("ip", "0.0.0.0")
         try:
             port = cfg["port"]
         except KeyError:
             raise ValueError("cfg")
 
-        return TCPReactor((address, port), self.server)
+        return self._reactors["tcp"].start_listening((address, port))
 
-    def _createProtonReactor(self, cfg):
+    def _createProtonListener(self, cfg):
         address = cfg.get("host", "0.0.0.0")
         port = cfg.get("port", 5672)
-        return ProtonReactor((address, port), self.server)
+        return self._reactors["amqp"].start_listening((address, port))
+
+    def _createTcpReactor(self):
+        return TCPReactor(self.server)
+
+    def _createProtonReactor(self):
+        return ProtonReactor(self.server)
 
     def start(self):
-        for reactor in self._reactors:
-            reactorName = reactor.__class__.__name__
-            try:
-                reactor.start_listening()
-            except:
-                # TBD: propegate error and crash VDSM
-                self.log.warning("Could not listen on for rector '%s'",
-                                 reactorName)
-            else:
-                t = threading.Thread(target=reactor.process_requests,
-                                     name='JsonRpc (%s)' % reactorName)
-                t.setDaemon(True)
-                t.start()
-
         t = threading.Thread(target=self.server.serve_requests,
-                             name='JsonRpc (Rquest Processing)')
+                             name='JsonRpcServer')
         t.setDaemon(True)
         t.start()
 
+        for reactor in self._reactors.itervalues():
+            reactorName = reactor.__class__.__name__
+            t = threading.Thread(target=reactor.process_requests,
+                                 name='JsonRpc (%s)' % reactorName)
+            t.setDaemon(True)
+            t.start()
+
+        for backendType, cfg in self._cfg:
+            try:
+                if backendType == "tcp":
+                    self._createTcpListener(cfg)
+                elif backendType == "amqp":
+                    self._createProtonListener(cfg)
+            except:
+                # TBD: propegate error and crash VDSM
+                self.log.warning("Could not listen on reactor '%s'",
+                                 reactorName, exc_info=True)
+
     def prepareForShutdown(self):
         self.server.stop()
         for reactor in self._reactors:
diff --git a/vdsm_api/jsonrpc/__init__.py b/vdsm_api/jsonrpc/__init__.py
index 3e6a6aa..49e8788 100644
--- a/vdsm_api/jsonrpc/__init__.py
+++ b/vdsm_api/jsonrpc/__init__.py
@@ -20,6 +20,11 @@
 __all__ = ["tcpReactor"]
 
 
+_STATE_INCOMING = 1
+_STATE_OUTGOING = 2
+_STATE_ONESHOT = 4
+
+
 class JsonRpcError(RuntimeError):
     def __init__(self, code, msg):
         self.code = code
@@ -107,7 +112,7 @@
         self.error = error
         self.id = reqId
 
-    def encode(self):
+    def toDict(self):
         res = {'jsonrpc': '2.0',
                'id': self.id}
 
@@ -117,6 +122,10 @@
         else:
             res['result'] = self.result
 
+        return res
+
+    def encode(self):
+        res = self.toDict()
         return json.dumps(res, 'utf-8')
 
     @staticmethod
@@ -153,6 +162,43 @@
         self.ctx = ctx
 
 
+class JsonRpcBatchRequest(object):
+    def __init__(self, requests):
+        self._requests = requests
+
+    def encode(self):
+        obj = [r.toDict() for r in self._requests]
+        return json.dumps(obj, 'utf-8')
+
+
+class JsonRpcCall(object):
+    def __init__(self, client, request):
+        self._request = request
+        self._client = client
+        self._state = _STATE_INCOMING
+
+    def fileno(self):
+        return self.client.fileno()
+
+    def state(self):
+        return
+
+
+class JsonRpcClient(object):
+    def __init__(self, transport):
+        self._transport = transport
+
+    def sendRequest(self, request):
+        request.encode()
+        self.transport.sendMessage()
+
+    def fileno(self):
+        return self._transport.fileno()
+
+    def process():
+        return None
+
+
 class JsonRpcServer(object):
     log = logging.getLogger("jsonrpc.JsonRpcServer")
 
diff --git a/vdsm_api/jsonrpc/protonReactor.py b/vdsm_api/jsonrpc/protonReactor.py
index aa2bf26..7b561e7 100644
--- a/vdsm_api/jsonrpc/protonReactor.py
+++ b/vdsm_api/jsonrpc/protonReactor.py
@@ -17,6 +17,8 @@
 import uuid
 from Queue import Queue, Empty
 import time
+from functools import partial
+from threading import Event
 
 import proton
 
@@ -48,14 +50,21 @@
         self._reactor._wakeup()
 
 
+class ProtonListener(object):
+    def __init__(self, address, reactor):
+        self._reactor = reactor
+        self._address = address
+
+    def close(self):
+        self._reactor._scheduleOp(False, self._reactor._stop_listening,
+                                  self._address)
+
+
 class ProtonReactor(object):
     log = logging.getLogger("jsonrpc.ProtonReactor")
 
-    def __init__(self, address, messageHandler, deliveryTimeout=5):
+    def __init__(self, messageHandler, deliveryTimeout=5):
         self._messageHandler = messageHandler
-        host, port = address
-        self.host = host
-        self.port = port
 
         self._isRunning = False
 
@@ -63,10 +72,11 @@
 
         self._sessionContexts = []
         self._deliveryTimeout = deliveryTimeout
-        self._activationQeue = Queue()
+        self._commandQueue = Queue()
+        self._listeners = {}
 
     def _activate(self, connector, cond):
-        self._activationQeue.put_nowait((connector, cond))
+        self._scheduleOp(False, proton.pn_connector_activate, connector, cond)
 
     def _convertTimeout(self, timeout):
         """
@@ -354,31 +364,61 @@
                 proton.pn_delivery_set_context(delivery, time.time())
                 proton.pn_link_advance(link)
 
-    def start_listening(self):
-        self.listener = proton.pn_listener(self._driver, self.host,
-                                           str(self.port), None)
-        if self.listener is None:
-            raise RuntimeError("Could not listen on %s:%s" % (self.host,
-                                                              self.port))
+    def start_listening(self, address):
+        host, port = address
+        l = self._scheduleOp(True, proton.pn_listener, self._driver,
+                             host, str(port), None)
+        if l is None:
+            raise RuntimeError("Could not listen on %s:%s" % (host, port))
 
-    def _emptyActivationQueue(self):
+        self._listeners[address] = l
+        return ProtonListener(address, self)
+
+    def _stop_listening(self, address):
+        try:
+            l = self._listeners[address]
+        except KeyError:
+            return
+
+        proton.pn_listener_close(l)
+        del self._listeners[address]
+
+    def _emptyCommandQueue(self):
         while True:
             try:
-                args = self._activationQeue.get_nowait()
+                r = self._commandQueue.get_nowait()
             except Empty:
                 return
             else:
-                proton.pn_connector_activate(*args)
+                cmd, evt, _ = r
+                res = cmd()
+                if evt is not None:
+                    r[2] = res
+                    evt.set()
+
+    def _scheduleOp(self, sync, op, *args, **kwargs):
+        if sync:
+            r = [partial(op, *args, **kwargs), Event(), None]
+        else:
+            r = [partial(op, *args, **kwargs), None, None]
+
+        self._commandQueue.put_nowait(r)
+        self._wakeup()
+
+        if sync:
+            r[1].wait()
+            return r[2]
 
     def process_requests(self):
         self._isRunning = True
         while self._isRunning:
             self._waitDriverEvent()
-            self._emptyActivationQueue()
+            self._emptyCommandQueue()
             self._acceptConnectionRequests()
             self._processConnectors()
 
-        proton.pn_listener_close(self.listener)
+        for listener in self._listeners.keys():
+            self._stop_listening(listener)
 
     def _wakeup(self):
         proton.pn_driver_wakeup(self._driver)
diff --git a/vdsm_api/jsonrpc/tcpReactor.py b/vdsm_api/jsonrpc/tcpReactor.py
index a9cf40c..2ef8766 100644
--- a/vdsm_api/jsonrpc/tcpReactor.py
+++ b/vdsm_api/jsonrpc/tcpReactor.py
@@ -82,7 +82,6 @@
 
 
 class TCPMessageContext(object):
-
     def __init__(self, server, conn, data):
         self._server = server
         self._conn = conn
@@ -96,30 +95,55 @@
         self._server.sendReply(self, data)
 
 
+class TCPListener(object):
+    def __init__(self, address):
+        self._address = address
+        self.sock = sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.bind(address)
+        sock.listen(10)
+
+    def fileno(self):
+        return self.sock.fileno()
+
+    def close(self):
+        self.sock.close()
+
+
 class TCPReactor(object):
     log = logging.getLogger("jsonrpc.TCPReactor")
 
-    def __init__(self, address, messageHandler):
-        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self._addr = address
+    def __init__(self, messageHandler):
         self._messageHandler = messageHandler
         self._inputEvent = utils.PollEvent()
         # TODO: Close on exec
-        self._sock.bind(self._addr)
+        self._listeners = {}
 
-    def start_listening(self):
-        sock = self._sock
-        sock.listen(10)
+    def start_listening(self, address):
+        l = TCPListener(address)
+
+        self._listeners[address] = l
+        self._inputEvent.set()
+        return l
 
     def process_requests(self):
-        sock = self._sock
         poller = poll()
-        poller.register(sock, POLLIN | POLLPRI)
-        poller.register(self._inputEvent, POLLIN | POLLPRI)
         connections = {}
         self.log.debug("Starting to accept clients")
-        sockFD = sock.fileno()
+
+        listenerFDs = {}
+        poller.register(self._inputEvent, POLLIN | POLLPRI)
+        # TODO: Exist condition
         while True:
+            for l in self._listeners.values():
+                try:
+                    fd = l.fileno()
+                except:
+                    continue
+
+                if fd not in listenerFDs:
+                    listenerFDs[fd] = l
+                    poller.register(fd, POLLIN | POLLPRI)
+
             for fd, conn in connections.iteritems():
                 if conn.hasSendData():
                     poller.modify(fd, (POLLIN | POLLPRI | POLLOUT))
@@ -128,20 +152,21 @@
 
             for fd, ev in poller.poll():
                 if ev & (POLLERR | POLLHUP):
-                    if fd == sockFD:
-                        self.log.info("Listening socket closed, shutting down")
-                        return
+                    if fd in listenerFDs:
+                        self.log.info("Listening socket closed")
+                        del listenerFDs[fd]
+                    else:
+                        self.log.debug("Connection closed")
+                        del connections[fd]
 
-                    self.log.debug("Connection closed")
                     poller.unregister(fd)
-                    del connections[fd]
 
                 elif fd == self._inputEvent.fileno():
                     self._inputEvent.clear()
 
-                elif fd == sockFD:
+                elif fd in listenerFDs:
                     try:
-                        conn, addr = sock.accept()
+                        conn, addr = listenerFDs[fd].sock.accept()
                     except (OSError, IOError):
                         continue
 
@@ -165,4 +190,5 @@
         self._inputEvent.set()
 
     def stop(self):
-        self._sock.close()
+        for listener in self._listeners.itervalues():
+            listener.close()


--
To view, visit http://gerrit.ovirt.org/11400
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ibe1b392eab2c30df5720caca11a80f44c6135437
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Saggi Mizrahi <smizrahi at redhat.com>


More information about the vdsm-patches mailing list