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