Change in vdsm[master]: jsonrpc: Make reactor clients 1st class objects
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: Make reactor clients 1st class objects
......................................................................
jsonrpc: Make reactor clients 1st class objects
Also add support for batch requests, tests to come soon
Change-Id: I64e4dfeb4eece4a85a3c6d36324b1e00f366a1cd
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, 360 insertions(+), 235 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/02/11402/1
diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index 180914d..57b18fa 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -25,6 +25,7 @@
import json
import uuid
import logging
+from Queue import Queue
from nose.plugins.skip import SkipTest
@@ -73,10 +74,10 @@
@contextmanager
-def _tcpServerConstructor(messageHandler):
+def _tcpServerConstructor():
port = _getFreePort()
address = ("localhost", port)
- reactor = tcpReactor.TCPReactor(messageHandler)
+ reactor = tcpReactor.TCPReactor()
try:
yield reactor, partial(TCPReactorClient, address), address
@@ -85,13 +86,13 @@
@contextmanager
-def _protonServerConstructor(messageHandler):
+def _protonServerConstructor():
if protonReactor is None:
raise SkipTest("qpid-proton python bindings are not installed")
port = _getFreePort()
serverAddress = "amqp://127.0.0.1:%d/vdsm_test" % (port,)
- reactor = protonReactor.ProtonReactor(messageHandler)
+ reactor = protonReactor.ProtonReactor()
try:
yield (reactor,
@@ -107,19 +108,24 @@
@contextmanager
-def constructReactor(tp, messageHandler):
- with REACTOR_CONSTRUCTORS[tp](messageHandler) as res:
+def constructReactor(tp):
+ with REACTOR_CONSTRUCTORS[tp]() as res:
yield res
@contextmanager
def constructServer(tp, bridge):
- server = JsonRpcServer(bridge)
- with constructReactor(tp, server) as (reactor, clientFactory, laddr):
+ queue = Queue()
+ server = JsonRpcServer(bridge, queue)
+ with constructReactor(tp) as (reactor, clientFactory, laddr):
t = threading.Thread(target=reactor.process_requests)
t.setDaemon(True)
t.start()
- reactor.createListener(laddr)
+
+ def _accept(listener, client):
+ client.setInbox(queue)
+
+ reactor.createListener(laddr, _accept)
t = threading.Thread(target=server.serve_requests)
t.setDaemon(True)
@@ -132,11 +138,6 @@
yield server, jsonClientFactory
finally:
server.stop()
-
-
-class _EchoMessageHandler(object):
- def handleMessage(self, msgCtx):
- msgCtx.sendReply(msgCtx.data)
class TCPReactorClient(object):
@@ -255,12 +256,36 @@
self._msngr.stop()
+class _EchoServer(object):
+ log = logging.getLogger("EchoServer")
+
+ def __init__(self):
+ self._queue = Queue()
+
+ def accept(self, l, c):
+ c.setInbox(self._queue)
+
+ def serve(self):
+ while True:
+ try:
+ client, msg = self._queue.get()
+ if client is None:
+ return
+
+ self.log.info("Echoing message")
+ client.send(msg)
+ except Exception:
+ self.log.error("EchoServer died unexpectedly", exc_info=True)
+
+
@expandPermutations
class ReactorTests(TestCaseBase):
@permutations(REACTOR_TYPE_PERMUTATIONS)
def test(self, reactorType):
data = dummyTextGenerator(((2 ** 10) * 200))
- msgHandler = _EchoMessageHandler()
+ queue = Queue()
+
+ echosrv = _EchoServer()
def serve(reactor):
try:
@@ -268,16 +293,20 @@
except socket.error as e:
pass
except Exception as e:
- self.log.error("Server died unexpectedly", exc_info=True)
- self.fail("Server died: (%s) %s" % (type(e), e))
+ self.log.error("Reactor died unexpectedly", exc_info=True)
+ self.fail("Reactor died: (%s) %s" % (type(e), e))
- with constructReactor(reactorType, msgHandler) as (reactor,
- clientFactory,
- laddr):
+ with constructReactor(reactorType) as \
+ (reactor, clientFactory, laddr):
t = threading.Thread(target=serve, args=(reactor,))
t.setDaemon(True)
t.start()
- reactor.createListener(laddr)
+
+ t = threading.Thread(target=echosrv.serve)
+ t.setDaemon(True)
+ t.start()
+
+ reactor.createListener(laddr, echosrv.accept)
clientNum = 1
repeats = 1
@@ -306,6 +335,8 @@
for client in clients:
client.close()
+ queue.put((None, None))
+
class _DummyBridge(object):
def echo(self, text):
diff --git a/vdsm_api/BindingJsonRpc.py b/vdsm_api/BindingJsonRpc.py
index a4c6d81..21f250c 100644
--- a/vdsm_api/BindingJsonRpc.py
+++ b/vdsm_api/BindingJsonRpc.py
@@ -16,6 +16,7 @@
import threading
import logging
import struct
+from Queue import Queue
_Size = struct.Struct("!Q")
@@ -40,7 +41,9 @@
def __init__(self, bridge, backendConfig):
reactors = {}
self.bridge = bridge
- self.server = JsonRpcServer(bridge, _simpleThreadFactory)
+ self._messageQueue = Queue()
+ self.server = JsonRpcServer(bridge, self._messageQueue,
+ _simpleThreadFactory)
self._cfg = backendConfig
for backendType, cfg in backendConfig:
@@ -62,7 +65,11 @@
except KeyError:
raise ValueError("cfg")
- return self._reactors["tcp"].createListener((address, port))
+ return self._reactors["tcp"].createListener((address, port),
+ self._onAccept)
+
+ def _onAccept(self, listener, client):
+ client.setInbox(self._messageQueue)
def _createProtonListener(self, cfg):
address = cfg.get("host", "0.0.0.0")
@@ -70,10 +77,10 @@
return self._reactors["amqp"].createListener((address, port))
def _createTcpReactor(self):
- return TCPReactor(self.server)
+ return TCPReactor()
def _createProtonReactor(self):
- return ProtonReactor(self.server)
+ return ProtonReactor()
def start(self):
t = threading.Thread(target=self.server.serve_requests,
diff --git a/vdsm_api/jsonrpc/__init__.py b/vdsm_api/jsonrpc/__init__.py
index 49e8788..c1ebf2d 100644
--- a/vdsm_api/jsonrpc/__init__.py
+++ b/vdsm_api/jsonrpc/__init__.py
@@ -14,7 +14,6 @@
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
import json
import logging
-from Queue import Queue
from functools import partial
__all__ = ["tcpReactor"]
@@ -70,21 +69,17 @@
self.params = params
self.id = reqId
- def encode(self):
- res = {'jsonrpc': '2.0',
- 'method': self.method,
- 'params': self.params,
- 'id': self.id}
-
- return json.dumps(res, 'utf-8')
-
- @staticmethod
- def decode(msg):
+ @classmethod
+ def decode(cls, msg):
try:
obj = json.loads(msg, 'utf-8')
except:
raise JsonRpcParseError()
+ return cls.fromRawObject(obj)
+
+ @staticmethod
+ def fromRawObject(obj):
if obj.get("jsonrpc") != "2.0":
raise JsonRpcInvalidRequestError()
@@ -101,6 +96,14 @@
raise JsonRpcInvalidRequestError()
return JsonRpcRequest(method, params, reqId)
+
+ def encode(self):
+ res = {'jsonrpc': '2.0',
+ 'method': self.method,
+ 'params': self.params,
+ 'id': self.id}
+
+ return json.dumps(res, 'utf-8')
def isNotification(self):
return (self.id is None)
@@ -138,37 +141,53 @@
return JsonRpcResponse(result, error, reqId)
-class _JsonRpcRequestContext(object):
- def __init__(self, ctx, queue, request):
- self.request = request
- self._ctx = ctx
- self._queue = queue
+class _JsonRpcServeRequestContext(object):
+ def __init__(self, client):
+ self._requests = []
+ self._client = client
+ self._counter = 0
+ self._requests = {}
+ self._responses = []
- def isNotification(self):
- return self.id is None
+ def setRequests(self, requests):
+ for request in requests:
+ if not request.isNotification():
+ self._counter += 1
+ self._requests[request.id] = request
- def sendReply(self, result, error):
- # TBD: Should calling this for a notification raise an error or be
- # ignored
- resp = JsonRpcResponse(result,
- error,
- self.request.id)
- self._queue.put_nowait(_JsonRpcResponseContext(self._ctx, resp))
+ self.sendReply()
+ @property
+ def counter(self):
+ return self._counter
-class _JsonRpcResponseContext(object):
- def __init__(self, ctx, response):
- self.response = response
- self.ctx = ctx
+ def sendReply(self):
+ if len(self._requests) > 0:
+ return
+ encodedObjects = []
+ for response in self._responses:
+ try:
+ encodedObjects.append(response.encode())
+ except: # Error encoding data
+ response = JsonRpcResponse(None, JsonRpcInternalError,
+ response.id)
+ encodedObjects.append(response.encode())
-class JsonRpcBatchRequest(object):
- def __init__(self, requests):
- self._requests = requests
+ if len(encodedObjects) == 1:
+ data = encodedObjects[0]
+ else:
+ data = '[' + ','.join(encodedObjects) + ']'
- def encode(self):
- obj = [r.toDict() for r in self._requests]
- return json.dumps(obj, 'utf-8')
+ self._client.send(data.encode('utf-8'))
+
+ def addResponse(self, response):
+ self._responses.append(response)
+
+ def requestDone(self, response):
+ del self._requests[response.id]
+ self.addResponse(response)
+ self.sendReply()
class JsonRpcCall(object):
@@ -202,45 +221,40 @@
class JsonRpcServer(object):
log = logging.getLogger("jsonrpc.JsonRpcServer")
- def __init__(self, bridge, threadFactory=None):
+ def __init__(self, bridge, messageQueue, threadFactory=None):
self._bridge = bridge
- self._workQueue = Queue()
+ self._workQueue = messageQueue
self._threadFactory = threadFactory
- def _serveRequest(self, ctx):
- req = ctx.request
+ def _serveRequest(self, ctx, req):
mangledMethod = req.method.replace(".", "_")
self.log.debug("Looking for method '%s' in bridge",
mangledMethod)
try:
method = getattr(self._bridge, mangledMethod)
except AttributeError:
- ctx.sendReply(None, JsonRpcMethodNotFoundError())
- else:
- try:
- params = req.params
- if isinstance(req.params, list):
- res = method(*params)
- else:
- res = method(**params)
- except JsonRpcError as e:
- ctx.sendReply(None, e)
- except Exception as e:
- ctx.sendReply(None, JsonRpcInternalError(str(e)))
- else:
- return ctx.sendReply(res, None)
+ if req.isNotification():
+ return
- def _processResponse(self, ctx):
+ ctx.requestDone(JsonRpcResponse(None,
+ JsonRpcMethodNotFoundError(),
+ req.id))
+ return
+
try:
- msg = ctx.response.encode()
+ params = req.params
+ if isinstance(req.params, list):
+ res = method(*params)
+ else:
+ res = method(**params)
+ except JsonRpcError as e:
+ ctx.requestDone(JsonRpcResponse(None, e, req.id))
except Exception as e:
- # Probably result failed to be serialized as json
- errResp = JsonRpcResponse(error=JsonRpcInternalError(str(e)),
- reqId=ctx.response.id)
-
- msg = errResp.encode()
-
- ctx.ctx.sendReply(msg)
+ ctx.requestDone(JsonRpcResponse(None,
+ JsonRpcInternalError(str(e)),
+ req.id))
+ else:
+ ctx.requestDone(JsonRpcResponse(res, None, req.id))
def serve_requests(self):
while True:
@@ -248,45 +262,58 @@
if obj is None:
break
- if isinstance(obj, _JsonRpcRequestContext):
- if self._threadFactory is None:
- self._serveRequest(obj)
- else:
- self._threadFactory(partial(self._serveRequest, obj))
- else:
- self._processResponse(obj)
+ client, msg = obj
+ self._parseMessage(client, msg)
- def handleMessage(self, msgCtx):
- #TODO: support batch requests
- req = None
- error = None
+ def _parseMessage(self, client, msg):
+ ctx = _JsonRpcServeRequestContext(client)
+
try:
- req = JsonRpcRequest.decode(msgCtx.data)
- ctx = _JsonRpcRequestContext(msgCtx, self._workQueue, req)
-
- self._workQueue.put_nowait(ctx)
- return
-
- except JsonRpcError as e:
- self.log.error("Error processing request", exc_info=True)
- error = e
+ rawRequests = json.loads(msg)
except:
- self.log.error("Unexpected error while processing request",
- exc_info=True)
-
- error = JsonRpcInternalError()
-
- # Notification, don't respond even on errors
- if req is not None and req.isNotification():
+ ctx.addResponse(JsonRpcResponse(None, JsonRpcParseError(), None))
+ ctx.sendReply()
return
- if req is None:
- resp = JsonRpcResponse(None, error, None)
+ if isinstance(rawRequests, list):
+ # Empty batch request
+ if len(rawRequests) == 0:
+ ctx.addResponse(
+ JsonRpcResponse(None, JsonRpcInvalidRequestError(),
+ None))
+ ctx.sendReply()
+ return
else:
- resp = JsonRpcResponse(None, error, req.id)
+ # From this point on we know it's always a list
+ rawRequests = [rawRequests]
- ctx = _JsonRpcResponseContext(msgCtx, resp)
- self._workQueue.put_nowait(ctx)
+ # JSON Parsed handling each request
+ requests = []
+ for rawRequest in rawRequests:
+ try:
+ req = JsonRpcRequest.fromRawObject(rawRequest)
+ requests.append(req)
+ except JsonRpcError as err:
+ ctx.addResponse(JsonRpcResponse(None, err, None))
+ except:
+ ctx.addResponse(JsonRpcResponse(None,
+ JsonRpcInternalError(),
+ None))
+
+ ctx.setRequests(requests)
+
+ # No request was built successfully or is only notifications
+ if ctx.counter == 0:
+ ctx.sendReply()
+
+ for request in requests:
+ self._runRequest(ctx, request)
+
+ def _runRequest(self, ctx, request):
+ if self._threadFactory is None:
+ self._serveRequest(ctx, request)
+ else:
+ self._threadFactory(partial(self._serveRequest, ctx, request))
def stop(self):
self._workQueue.put_nowait(None)
diff --git a/vdsm_api/jsonrpc/protonReactor.py b/vdsm_api/jsonrpc/protonReactor.py
index da903e5..307e526 100644
--- a/vdsm_api/jsonrpc/protonReactor.py
+++ b/vdsm_api/jsonrpc/protonReactor.py
@@ -27,45 +27,62 @@
AUTHENTICATING = 2
-class ProtonContext(object):
- log = logging.getLogger("jsonrpc.ProtonContext")
-
- def __init__(self, reactor, messageQueue, connector, msg):
- self._reactor = reactor
- self._msg = msg
- self._mq = messageQueue
- self._connector = connector
-
- @property
- def data(self):
- return self._msg.body
-
- def sendReply(self, data):
- msg = proton.Message()
- msg.address = self._msg.reply_to
- msg.body = data
- self._mq.put_nowait(msg)
- self.log.debug("Message Queued")
- self._reactor._activate(self._connector, proton.PN_CONNECTOR_WRITABLE)
- self._reactor._wakeup()
-
-
class ProtonListener(object):
- def __init__(self, address, reactor):
+ def __init__(self, listener, address, reactor, acceptHandler):
+ self._listener = listener
self._reactor = reactor
self._address = address
+ self._acceptHandler = acceptHandler
+
+ def fileno(self):
+ return proton.pn_listener_fd(self._listener)
def close(self):
- self._reactor._scheduleOp(False, self._reactor._stop_listening,
- self._address)
+ self._reactor._scheduleOp(False, proton.pn_listener_close,
+ self._listener)
+
+
+class ProtonClient(object):
+ def __init__(self, reactor, connection, connector, session):
+ self.connector = connector
+ self.connection = connection
+ self.session = session
+ self.sender = None
+ self.links = []
+ self._inbox = None
+ self._outbox = Queue()
+ self._reactor = reactor
+
+ def _pushIncomingMessage(self, msg):
+ try:
+ self._inbox.put_nowait((self, msg))
+ except AttributeError:
+ # Inbox not set
+ pass
+
+ def _popPendingMessage(self):
+ return self._outbox.get_nowait()
+
+ def fileno(self):
+ return proton.pn_connctor_fd(self._connector)
+
+ def setInbox(self, queue):
+ self._inbox = queue
+
+ def send(self, msg):
+ self._outbox.put_nowait(msg)
+ self._reactor._activate(self.connector,
+ proton.PN_CONNECTOR_WRITABLE)
+
+ def close(self):
+ #TODO
+ pass
class ProtonReactor(object):
log = logging.getLogger("jsonrpc.ProtonReactor")
- def __init__(self, messageHandler, deliveryTimeout=5):
- self._messageHandler = messageHandler
-
+ def __init__(self, deliveryTimeout=5):
self._isRunning = False
self._driver = proton.pn_driver()
@@ -162,15 +179,16 @@
self.log.debug("Connection Opened.")
proton.pn_connection_open(conn)
- def _openPendingSessions(self, conn):
+ def _openPendingSessions(self, conn, connector):
ssn = proton.pn_session_head(conn, proton.PN_LOCAL_UNINIT)
while ssn:
proton.pn_session_open(ssn)
- ctx = {'connection': conn,
- 'session': ssn,
- 'mqueue': Queue(),
- 'sender': None,
- 'links': []}
+ ctx = ProtonClient(self, conn, connector, ssn)
+ l = proton.pn_connector_listener(connector)
+ listener = proton.pn_listener_context(l)
+ listener._acceptHandler(listener, ctx)
+ self.log.debug("Found related listener")
+
self._sessionContexts.append(ctx)
proton.pn_session_set_context(ssn, ctx)
self.log.debug("Session Opened.")
@@ -249,11 +267,11 @@
self.log.debug("Closing Link")
proton.pn_link_close(link)
for ctx in self._sessionContexts:
- if link in ctx['links']:
- ctx['links'].remove(link)
+ if link in ctx.links:
+ ctx.links.remove(link)
- if link == ctx['sender']:
- ctx['sender'] = None
+ if link == ctx.sender:
+ ctx.sender = None
link = proton.pn_link_next(link, (proton.PN_LOCAL_ACTIVE |
proton.PN_REMOTE_CLOSED))
@@ -275,25 +293,26 @@
def _queueOutgoingDeliveries(self, conn):
ctxs = (ctx for ctx in self._sessionContexts
- if ctx['connection'] == conn)
+ if ctx.connection == conn)
for ctx in ctxs:
- sender = ctx['sender']
+ sender = ctx.sender
if sender is None:
# No sender link
- sender = proton.pn_sender(ctx['session'],
+ sender = proton.pn_sender(ctx.session,
"sender-%s" % str(uuid.uuid4()))
- ctx['sender'] = sender
+ ctx.sender = sender
proton.pn_link_open(sender)
continue
- mq = ctx['mqueue']
try:
- msg = mq.get_nowait()
+ data = ctx._popPendingMessage()
except Empty:
continue
else:
+ msg = proton.Message()
+ msg.body = data
self.log.debug("Creating delivery")
proton.pn_link_set_context(sender, msg.encode())
if proton.pn_link_credit(sender) == 0:
@@ -308,7 +327,7 @@
conn = proton.pn_connector_connection(connector)
self._initConnection(conn)
- self._openPendingSessions(conn)
+ self._openPendingSessions(conn, connector)
self._openLinks(conn)
self._queueOutgoingDeliveries(conn)
self._processDeliveries(conn, connector)
@@ -336,9 +355,7 @@
msgObj = proton.Message()
msgObj.decode(msg)
ctx = proton.pn_session_get_context(ssn)
- mq = ctx['mqueue']
- self._messageHandler.handleMessage(ProtonContext(self, mq, connector,
- msgObj))
+ ctx._pushIncomingMessage(msgObj.body)
proton.pn_delivery_settle(delivery)
proton.pn_link_advance(link)
@@ -364,24 +381,20 @@
proton.pn_delivery_set_context(delivery, time.time())
proton.pn_link_advance(link)
- def createListener(self, address):
+ def createListener(self, address, acceptHandler):
host, port = address
- l = self._scheduleOp(True, proton.pn_listener, self._driver,
- host, str(port), None)
+ return self._scheduleOp(True, self._createListener, address,
+ acceptHandler)
+
+ def _createListener(self, address, acceptHandler):
+ host, port = address
+ l = proton.pn_listener(self._driver, host, str(port), None)
if l is None:
raise RuntimeError("Could not listen on %s:%s" % (host, port))
- 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]
+ lObj = ProtonListener(l, address, self, acceptHandler)
+ proton.pn_listener_set_context(l, lObj)
+ return lObj
def _emptyCommandQueue(self):
while True:
@@ -417,8 +430,10 @@
self._acceptConnectionRequests()
self._processConnectors()
- for listener in self._listeners.keys():
- self._stop_listening(listener)
+ l = proton.pn_listener_head(self._driver)
+ while l:
+ proton.pn_listener_close(l)
+ l = proton.pn_listener_next(l)
def _wakeup(self):
proton.pn_driver_wakeup(self._driver)
diff --git a/vdsm_api/jsonrpc/tcpReactor.py b/vdsm_api/jsonrpc/tcpReactor.py
index bda9057..8c9cc7d 100644
--- a/vdsm_api/jsonrpc/tcpReactor.py
+++ b/vdsm_api/jsonrpc/tcpReactor.py
@@ -81,26 +81,68 @@
return res
-class TCPMessageContext(object):
- def __init__(self, server, conn, data):
- self._server = server
- self._conn = conn
- self._data = data
+class TCPClient(object):
+ def __init__(self, reactor, sock):
+ self._sock = sock
+ self._conn = TCPConnection(sock)
+ self._inbox = None
+ self._reactor = reactor
- @property
- def data(self):
- return self._data
+ def _pushRecievedMessage(self, msg):
+ try:
+ self._inbox.put_nowait((self, msg))
+ except AttributeError:
+ # Inbox not set
+ pass
- def sendReply(self, data):
- self._server.sendReply(self, data)
+ def fileno(self):
+ return self._sock.fileno()
+
+ def setInbox(self, queue):
+ self._inbox = queue
+
+ def send(self, message):
+ self._conn.addSendData(_Size.pack(len(message)) + message)
+ self._reactor.wakeup()
+
+ def _processInput(self):
+ msg = self._conn.processInput()
+ while msg is not None:
+ self._pushRecievedMessage(msg)
+ msg = self._conn.processInput()
+
+ def _processOutput(self):
+ self._conn.processOutput()
+
+ def _hasSendData(self):
+ return self._conn.hasSendData()
+
+ def close(self):
+ self._sock.close()
class TCPListener(object):
- def __init__(self, address):
+ log = logging.getLogger("jsonrpc.TCPListener")
+
+ def __init__(self, reactor, address, acceptHandler):
self._address = address
self.sock = sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(address)
sock.listen(10)
+ self._acceptHandler = acceptHandler
+ self._reactor = reactor
+
+ def _accept(self):
+ sock, addr = self.sock.accept()
+
+ client = TCPClient(self._reactor, sock)
+ try:
+ self._acceptHandler(self, client)
+ except:
+ self.log.warning("Accept handler threw an unexpected exception",
+ exc_info=True)
+
+ return client
def fileno(self):
return self.sock.fileno()
@@ -112,83 +154,86 @@
class TCPReactor(object):
log = logging.getLogger("jsonrpc.TCPReactor")
- def __init__(self, messageHandler):
- self._messageHandler = messageHandler
+ def __init__(self):
self._inputEvent = utils.PollEvent()
# TODO: Close on exec
- self._listeners = {}
+ self._trackedObjects = set()
+ self._isRunning = False
- def createListener(self, address):
- l = TCPListener(address)
+ def createListener(self, address, acceptHandler):
+ l = TCPListener(self, address, acceptHandler)
- self._listeners[address] = l
+ self._trackedObjects.add(l)
self._inputEvent.set()
return l
+ def wakeup(self):
+ self._inputEvent.set()
+
def process_requests(self):
poller = poll()
- connections = {}
self.log.debug("Starting to accept clients")
- listenerFDs = {}
+ objMap = {}
poller.register(self._inputEvent, POLLIN | POLLPRI)
# TODO: Exist condition
while True:
- for l in self._listeners.values():
+ for obj in self._trackedObjects:
try:
- fd = l.fileno()
+ fd = obj.fileno()
except:
continue
- if fd not in listenerFDs:
- listenerFDs[fd] = l
+ if fd not in objMap:
+ objMap[fd] = obj
poller.register(fd, POLLIN | POLLPRI)
- for fd, conn in connections.iteritems():
- if conn.hasSendData():
+ for fd, obj in objMap.iteritems():
+ if not isinstance(obj, TCPClient):
+ continue
+
+ if obj._hasSendData():
poller.modify(fd, (POLLIN | POLLPRI | POLLOUT))
else:
poller.modify(fd, (POLLIN | POLLPRI))
for fd, ev in poller.poll():
+ if fd == self._inputEvent.fileno():
+ self._inputEvent.clear()
+ continue
+
+ obj = objMap[fd]
if ev & (POLLERR | POLLHUP):
- if fd in listenerFDs:
+ if isinstance(obj, TCPListener):
self.log.info("Listening socket closed")
- del listenerFDs[fd]
else:
self.log.debug("Connection closed")
- del connections[fd]
+ self._trackedObjects.discard(obj)
+ del objMap[fd]
poller.unregister(fd)
- elif fd == self._inputEvent.fileno():
- self._inputEvent.clear()
-
- elif fd in listenerFDs:
+ elif isinstance(obj, TCPListener):
try:
- conn, addr = listenerFDs[fd].sock.accept()
+ client = obj._accept()
except (OSError, IOError):
continue
self.log.debug("Processing new connection")
- connections[conn.fileno()] = TCPConnection(conn)
- poller.register(conn, (POLLIN | POLLPRI))
- else:
- conn = connections[fd]
- if ev & (POLLIN | POLLPRI):
- msg = conn.processInput()
- while msg is not None:
- ctx = TCPMessageContext(self, conn, msg)
- self._messageHandler.handleMessage(ctx)
- msg = conn.processInput()
- if ev & POLLOUT:
- conn.processOutput()
+ self._trackedObjects.add(client)
+ else: # TCPClient
+ try:
+ if ev & (POLLIN | POLLPRI):
+ obj._processInput()
- def sendReply(self, ctx, message):
- conn = ctx._conn
- conn.addSendData(_Size.pack(len(message)) + message)
- self._inputEvent.set()
+ if ev & POLLOUT:
+ obj._processOutput()
+ except:
+ poller.unregister(fd)
+ obj.close()
+ self._trackedObjects.discard(obj)
+ del objMap[fd]
def stop(self):
- for listener in self._listeners.itervalues():
- listener.close()
+ for obj in self._trackedObjects:
+ obj.close()
--
To view, visit http://gerrit.ovirt.org/11402
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I64e4dfeb4eece4a85a3c6d36324b1e00f366a1cd
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