Saggi Mizrahi has uploaded a new change for review.
Change subject: Refactor communication infra ......................................................................
Refactor communication infra
This is one phase in many, there is still a lot to do but it's a good step in the right direction.
As I said, this is far from complete but it's enough code for a commit IMO.
- Properly layer code to allow multiple transports - Isolate json-rpc logic to the actual json-rpc layer - Someone more conferment but still incomplete json-rpc 2.0 implementation - Make sure tcp handling uses one thread - Write some more tests
Change-Id: I7a65364824cc8433263ccf6ed44f1f97141eff2c Signed-off-by: Saggi Mizrahi smizrahi@redhat.com --- M tests/Makefile.am M tests/apiTests.py A tests/jsonRpcTests.py M vdsm/clientIF.py M vdsm/utils.py M vdsm_api/BindingJsonRpc.py M vdsm_api/Bridge.py A vdsm_api/jsonrpc/__init__.py A vdsm_api/jsonrpc/tcpReactor.py 9 files changed, 617 insertions(+), 99 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/86/9986/1
diff --git a/tests/Makefile.am b/tests/Makefile.am index 8ad34e2..c26adc3 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -32,6 +32,7 @@ gluster_cli_tests.py \ guestIFTests.py \ hooksTests.py \ + jsonRpcTests.py \ libvirtvmTests.py \ lsblkTests.py \ main.py \ diff --git a/tests/apiTests.py b/tests/apiTests.py index bf40f9e..d770cf8 100644 --- a/tests/apiTests.py +++ b/tests/apiTests.py @@ -30,7 +30,9 @@ from testrunner import VdsmTestCase as TestCaseBase from vdsm import constants import BindingJsonRpc +import jsonrpc import apiData +
ip = '0.0.0.0' port = 9824 @@ -144,7 +146,9 @@ # Support parallel testing. Try hard to find an open port to use while True: try: - server = BindingJsonRpc.BindingJsonRpc(bridge, ip, port) + server = BindingJsonRpc.BindingJsonRpc(bridge, + [('tcp', {"ip": ip, + "port": port})]) break except socket.error as ex: if ex.errno == errno.EADDRINUSE: @@ -195,16 +199,21 @@ resp = msize + msg return resp
+ def _createRequest(self, method, reqId=None, params=()): + return {'jsonrpc': '2.0', "id": reqId, "method": method, + "params": params} + def sendMessage(self, msg): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: + sock.settimeout(3) try: sock.connect((ip, port)) except socket.error, e: raise ConnectionError("Unable to connect to server: %s", e) try: sock.sendall(msg) - except socket.error, e: + except (socket.error, socket.timeout), e: raise ProtocolError("Unable to send request: %s", e) try: data = sock.recv(JsonRawTest._Size.size) @@ -224,7 +233,9 @@ def testPing(self): self.clearAPI() self.programAPI("testPing") - msg = self.buildMessage({'id': 1, 'method': 'Host.ping', + msg = self.buildMessage({'jsonrpc': '2.0', + 'id': 1, + 'method': 'Host.ping', 'params': {}}) reply = self.sendMessage(msg) self.assertFalse('error' in reply) @@ -233,21 +244,26 @@ def testPingError(self): self.clearAPI() self.programAPI("testPingError") - msg = self.buildMessage({'id': 1, 'method': 'Host.ping', + msg = self.buildMessage({'jsonrpc': '2.0', + 'id': 1, 'method': 'Host.ping', 'params': {}}) reply = self.sendMessage(msg) self.assertEquals(1, reply['error']['code']) self.assertFalse('result' in reply)
def testNoMethod(self): - msg = self.buildMessage({'id': 1, 'method': 'Host.fake'}) + msg = self.buildMessage({'jsonrpc': '2.0', + 'id': 1, + 'method': 'Host.fake'}) reply = self.sendMessage(msg) - self.assertEquals(4, reply['error']['code']) + self.assertEquals(jsonrpc.JsonRpcMethodNotFoundError().code, + reply['error']['code'])
def testBadMethod(self): - msg = self.buildMessage({'id': 1, 'method': 'malformed''}) + msg = self.buildMessage(self._createRequest('malformed'', 1)) reply = self.sendMessage(msg) - self.assertEquals(4, reply['error']['code']) + self.assertEquals(jsonrpc.JsonRpcMethodNotFoundError().code, + reply['error']['code'])
def testMissingSize(self): self.assertRaises(ProtocolError, self.sendMessage, @@ -257,7 +273,9 @@ msg = "malformed message" msize = JsonRawTest._Size.pack(len(msg)) msg = msize + msg - self.assertRaises(ProtocolError, self.sendMessage, msg) + reply = self.sendMessage(msg) + self.assertEquals(jsonrpc.JsonRpcParseError().code, + reply['error']['code'])
def testSynchronization(self): def doPing(msg): @@ -266,7 +284,9 @@ ret = self.sendMessage(msg) self.assertFalse('error' in ret)
- msg = self.buildMessage({'id': 1, 'method': 'Host.ping'}) + msg = self.buildMessage({'jsonrpc': '2.0', + 'id': 1, + 'method': 'Host.ping'}) # Send Truncated message self.assertRaises(ProtocolError, doPing, msg[:-1])
@@ -280,6 +300,8 @@ doPing(msg)
def testInternalError(self): - msg = self.buildMessage({'id': 1, 'method': 'Host.ping'}) + msg = self.buildMessage({'jsonrpc': '2.0', + 'id': 1, 'method': 'Host.ping'}) reply = self.sendMessage(msg) - self.assertEquals(5, reply['error']['code']) + self.assertEquals(jsonrpc.JsonRpcInternalError().code, + reply['error']['code']) diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py new file mode 100644 index 0000000..85b98a6 --- /dev/null +++ b/tests/jsonRpcTests.py @@ -0,0 +1,158 @@ +# +# Copyright 2012 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# +import threading +import socket +from contextlib import contextmanager +from functools import partial +from contextlib import closing + +from testrunner import VdsmTestCase as TestCaseBase, \ + expandPermutations, \ + permutations + +from jsonrpc import tcpReactor + +PORT_RANGE = xrange(49152, 65535) + + +_distributedPorts = [] + + +def _getFreePort(): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + with closing(sock): + for port in PORT_RANGE: + if port in _distributedPorts: + continue + + try: + sock.bind(("0.0.0.0", port)) + except: + continue + + _distributedPorts.append(port) + return port + else: + raise Exception("Could not find a free port") + + +@contextmanager +def _tcpServerConstructor(messageHandler): + port = _getFreePort() + address = ("localhost", port) + reactor = tcpReactor.TCPReactor(address, messageHandler) + + try: + yield reactor, partial(TCPRactorClient, address) + finally: + reactor.stop() + + +REACTOR_CONSTRUCTORS = {"tcp": _tcpServerConstructor} +REACTOR_TYPE_PERMUTATIONS = [[r] for r in REACTOR_CONSTRUCTORS.iterkeys()] + + +@contextmanager +def constructReactor(tp, messageHandler): + with REACTOR_CONSTRUCTORS[tp](messageHandler) as res: + yield res + + +class _EchoMessageHandler(object): + def handleMessage(self, msgCtx): + msgCtx.sendReply(msgCtx.data) + + +class TCPRactorClient(object): + def __init__(self, address): + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.address = address + + def connect(self): + self.sock.connect(self.address) + + def sendMessage(self, msg, timeout=None): + msg = tcpReactor._Size.pack(len(msg)) + msg + self.sock.settimeout(timeout) + self.sock.send(msg) + + def recvMessage(self, timeout=None): + self.sock.settimeout(timeout) + rawSize = self.sock.recv(tcpReactor._Size.size) + size = tcpReactor._Size.unpack(rawSize)[0] + return self.sock.recv(size) + + def close(self): + self.sock.close() + + +@expandPermutations +class ReactorTests(TestCaseBase): + @permutations(REACTOR_TYPE_PERMUTATIONS) + def test(self, reactorType): + data = """I've seen things you people wouldn't believe. Attack ships on + fire off the shoulder of Orion. I've watched c-beams glitter + in the dark near the Tannhauser Gate. All those ... moments + will be lost in time, like tears...in rain. Time to die.""" + # (C) Blade Runner + + msgHandler = _EchoMessageHandler() + + def serve(reactor): + try: + reactor.serve_forever() + 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)) + + with constructReactor(reactorType, msgHandler) as (reactor, + clientFactory): + t = threading.Thread(target=serve, args=(reactor,)) + t.setDaemon(True) + t.start() + + clientNum = 1 + repeats = 1 + subRepeats = 1 + + clients = [] + try: + for i in range(clientNum): + client = clientFactory() + client.connect() + clients.append(client) + + for i in range(repeats): + for client in clients: + for i in range(subRepeats): + self.log.info("Sending message...") + client.sendMessage(data, 1) + + for i in range(repeats * subRepeats): + for client in clients: + self.log.info("Waiting for reply...") + retData = client.recvMessage(1) + self.log.info("Asserting reply...") + self.assertEquals(retData, data) + finally: + for client in clients: + client.close() diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py index 65b4e72..48a9cf8 100644 --- a/vdsm/clientIF.py +++ b/vdsm/clientIF.py @@ -156,8 +156,8 @@ schema = os.path.join(constants.P_VDSM, 'vdsmapi-schema.json') ip = self._getServerIP(config.get('addresses', 'management_ip')) port = config.getint('addresses', 'json_port') - self.bindings['json'] = BindingJsonRpc(DynamicBridge(schema), - ip, port) + conf = [('tcp', {"ip": ip, "port": port})] + self.bindings['json'] = BindingJsonRpc(DynamicBridge(schema), conf)
def _prepareBindings(self): self.bindings = {} diff --git a/vdsm/utils.py b/vdsm/utils.py index b4614e7..bb012d0 100644 --- a/vdsm/utils.py +++ b/vdsm/utils.py @@ -905,6 +905,38 @@ return unicode(self.cmd)
+class PollEvent(object): + def __init__(self): + self._r, self._w = os.pipe() + self._lock = threading.Lock() + self._isSet = False + + def fileno(self): + return self._r + + def set(self): + with self._lock: + if self._isSet: + return + + os.write(self._w, "a") + self._isSet = True + + def isSet(self): + return self._isSet + + def clear(self): + with self._lock: + if not self._isSet: + return + + os.read(self._r, 1) + + def __del__(self): + os.close(self._r) + os.close(self._w) + + def retry(func, expectedException=Exception, tries=None, timeout=None, sleep=1): """ diff --git a/vdsm_api/BindingJsonRpc.py b/vdsm_api/BindingJsonRpc.py index 1ddc863..b8c8658 100644 --- a/vdsm_api/BindingJsonRpc.py +++ b/vdsm_api/BindingJsonRpc.py @@ -14,86 +14,42 @@ # License along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import threading -import SocketServer -import json import logging - import struct -
_Size = struct.Struct("!Q")
+from jsonrpc import JsonRpcServer +from jsonrpc.tcpReactor import TCPReactor
-class BindingJsonRpc: + +class BindingJsonRpc(object): log = logging.getLogger('BindingJsonRpc')
- def __init__(self, bridge, ip, port): + def __init__(self, bridge, backendConfig): + reactors = [] self.bridge = bridge - self.serverPort = port - self.serverIP = ip - self._createServer() + self.server = JsonRpcServer(bridge) + for backendType, cfg in backendConfig: + if backendType == "tcp": + reactors.append(self._createTcpReactor(cfg))
- def _createServer(self): - ip = self.serverIP or '0.0.0.0' - self.server = JsonRpcServer((ip, self.serverPort), JsonRpcTCPHandler, - self.bridge) + self._reactors = reactors + + def _createTcpReactor(self, cfg): + address = cfg.get("ip", "0.0.0.0") + port = cfg.get("port", 4044) + return TCPReactor((address, port), self.server)
def start(self): - t = threading.Thread(target=self.server.serve_forever, - name='JsonRpc') - t.setDaemon(True) - t.start() + for reactor in self._reactors: + reactorName = reactor.__class__.__name__ + t = threading.Thread(target=reactor.serve_forever, + name='JsonRpc (%s)' % reactorName) + t.setDaemon(True) + t.start()
def prepareForShutdown(self): self.server.shutdown() - - -class JsonRpcServer(SocketServer.TCPServer): - def __init__(self, addrInfo, handler, bridge): - self.bridge = bridge - self.allow_reuse_address = True - SocketServer.TCPServer.__init__(self, addrInfo, handler) - - -class JsonRpcTCPHandler(SocketServer.StreamRequestHandler): - log = logging.getLogger('JsonRpcTCPHandler') - - def handle(self): - while True: - # self.request is the TCP socket connected to the client - try: - data = self.request.recv(_Size.size) - if len(data) != _Size.size: - self.log.debug("Connection closed") - return - msgLen = _Size.unpack(data)[0] - msg = json.loads(self.request.recv(msgLen)) - except: - self.log.warn("Unexpected exception", exc_info=True) - return - self.log.debug('Received request: %s', msg) - - try: - ret = self.server.bridge.dispatch(msg['method'], - msg.get('params', {})) - resp = self.buildResponse(msg['id'], ret) - except Exception, e: - self.log.warn("Dispatch error", exc_info=True) - err = {'error': {'code': 5, - 'message': 'Dispatch error: %s' % e}} - resp = self.buildResponse(msg['id'], err) - - self.wfile.write(resp) - self.wfile.flush() - - def buildResponse(self, msgId, result): - msgData = {'id': msgId} - if result['error']['code'] != 0: - msgData['error'] = result['error'] - else: - msgData['result'] = result['result'] - msg = json.dumps(msgData) - msg = msg.encode('utf-8') - self.log.debug('Sending reply: %s', msg) - msize = _Size.pack(len(msg)) - return msize + msg + for reactor in self._reactors: + reactor.stop() diff --git a/vdsm_api/Bridge.py b/vdsm_api/Bridge.py index d93dc22..08970ba 100644 --- a/vdsm_api/Bridge.py +++ b/vdsm_api/Bridge.py @@ -14,10 +14,12 @@ # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
from functools import partial + import vdsmapi import logging import types import API +import jsonrpc
class VdsmError(Exception): @@ -26,22 +28,26 @@ self.message = message
-class MethodBridge(object): +class DynamicBridge(object): + def __init__(self, schema): + self._parseSchema(schema) + def dispatch(self, name, argobj): methodName = name.replace('.', '_') result = None - error = {'code': 0, 'message': 'Success'} try: fn = getattr(self, methodName) except AttributeError: - error = {'code': 4, - 'message': "Operation '%s' not supported" % name} - return {'result': result, 'error': error} + raise jsonrpc.JsonRpcMethodNotFoundError() + try: result = fn(argobj) except VdsmError, e: - error = {'code': e.code, 'message': e.message} - return {'result': result, 'error': error} + # TBD: Do we really want to always log here + self.log.debug("Operation failed, returning error", exc_info=True) + raise jsonrpc.JsonRpcError(e.code, e.message) + + return result
def _getArgs(self, argobj, arglist): return tuple(argobj[arg] for arg in arglist if arg in argobj) @@ -53,11 +59,6 @@ return response[member] except KeyError: raise VdsmError(5, "Response is missing '%s' member" % member) - - -class DynamicBridge(MethodBridge): - def __init__(self, schema): - self._parseSchema(schema)
def _parseSchema(self, schema): self.commands = {} @@ -82,7 +83,7 @@ className, methodName = attr.split('_') return partial(self._dynamicMethod, className, methodName) else: - raise AttributeError() + raise AttributeError(attr)
def _getArgList(self, cmd): return self.commands[cmd].get('data', {}).keys() @@ -148,15 +149,30 @@ self._typeFixup('return', retType, result) return result
- def _dynamicMethod(self, className, methodName, argobj): + def _nameArgs(self, args, kwargs, arglist): + kwargs = kwargs.copy() + for i, arg in enumerate(args): + argName = arglist[i] + kwargs[argName] = arg + + return kwargs + + def _dynamicMethod(self, className, methodName, *args, **kwargs): cmd = '%s_%s' % (className, methodName) - ctorArgObj = argobj.get('__obj__', {}) + + # __obj__ is deprecated, support here is only for BC until the patches + # that remove it get pushed in. + ctorArgObj = kwargs.pop('__obj__', {}) ctorargs = self._getArgs(ctorArgObj, self._getObjargList(className)) + # This might look a bit wasteful because we unpack the argobj in the + # next line. I put this just so I need to do minimal changes to the + # rest of the code but still support an arglist + argobj = self._nameArgs(args, kwargs, self._getArgList(cmd)) args = self._getArgs(argobj, self._getArgList(cmd)) apiobj = self._getApiClass(className) api = apiobj(*ctorargs)
- self._fixupArgs(cmd, args) + self._fixupArgs(cmd, argobj)
# Call the override function (if given). Otherwise, just call directly fn = command_info.get(cmd, {}).get('call') @@ -169,7 +185,7 @@ if result['status']['code']: code = result['status']['code'] msg = result['status']['message'] - raise VdsmError(code, msg) + raise jsonrpc.JsonRpcError(code, msg)
retfield = command_info.get(cmd, {}).get('ret') if isinstance(retfield, types.FunctionType): diff --git a/vdsm_api/jsonrpc/__init__.py b/vdsm_api/jsonrpc/__init__.py new file mode 100644 index 0000000..43e4199 --- /dev/null +++ b/vdsm_api/jsonrpc/__init__.py @@ -0,0 +1,176 @@ +# Copyright (C) 2012 Saggi Mizrahi, Red Hat Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public +# License along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA +import json +import logging + +__all__ = ["tcpReactor"] + + +class JsonRpcError(RuntimeError): + def __init__(self, code, msg): + self.code = code + self._msg = msg + + +class JsonRpcParseError(JsonRpcError): + def __init__(self): + JsonRpcError.__init__(self, -32700, + "Invalid JSON was received by the server. " + "An error occurred on the server while parsing " + "the JSON text.") + + +class JsonRpcInvalidRequestError(JsonRpcError): + def __init__(self): + JsonRpcError.__init__(self, -32600, + "The JSON sent is not a valid Request object.") + + +class JsonRpcMethodNotFoundError(JsonRpcError): + def __init__(self): + JsonRpcError.__init__(self, -32601, + "The method does not exist / is not available.") + + +class JsonRpcInvalidParamsError(JsonRpcError): + def __init__(self): + JsonRpcError.__init__(self, -32602, + "Invalid method parameter(s).") + + +class JsonRpcInternalError(JsonRpcError): + def __init__(self): + JsonRpcError.__init__(self, -32603, + "Internal JSON-RPC error.") + + +class JsonRpcRequest(object): + def __init__(self, method, params, reqId): + self.method = method + self.params = params + self.id = reqId + + def invokeFunction(self, func): + if isinstance(self.params, list): + return func(*self.params) + else: + return func(**self.params) + + def isNotification(self): + return self.id is None + + +class JsonRpcServer(object): + log = logging.getLogger("JsonRpcServer") + + def __init__(self, bridge): + self._bridge = bridge + + def _parseMessage(self, msg): + try: + return json.loads(msg, 'utf-8') + except: + raise JsonRpcParseError() + + def _parseRequest(self, obj): + if obj.get("jsonrpc") != "2.0": + raise JsonRpcInvalidRequestError() + + method = obj.get("method") + if method is None: + raise JsonRpcInvalidRequestError() + + reqId = obj.get("id") + if not isinstance(reqId, int): + raise JsonRpcInvalidRequestError() + + params = obj.get('params', []) + if not isinstance(params, (list, dict)): + raise JsonRpcInvalidRequestError() + + return JsonRpcRequest(method, params, reqId) + + def _jsonError2Response(self, err, req): + respId = None + if req is not None: + respId = req.id + + return json.dumps({"jsonrpc": "2.0", + "error": {"code": err.code, "message": err.message}, + "id": respId}) + + def _generateResponse(self, req, result): + return json.dumps({"jsonrpc": "2.0", + "result": result, + "id": req.id}, 'utf-8') + + def _processRequest(self, srv, conn, req, method): + # TODO: Multithreading + resp = None + try: + res = req.invokeFunction(method) + resp = self._generateResponse(req, res) + + except JsonRpcError as e: + if not req.isNotification(): + resp = self._jsonError2Response(e, req) + + self.log.error("Error in request", exc_info=True) + except: + if not req.isNotification(): + resp = self._jsonError2Response(JsonRpcInternalError(), req) + + self.log.error("Unexpected error in request", exc_info=True) + + if req.isNotification(): + return + + srv.sendMessage(conn, self._jsonError2Response(e, resp)) + + def handleMessage(self, msgCtx): + #TODO: support batch requests + req = None + resp = None + try: + obj = self._parseMessage(msgCtx.data) + req = self._parseRequest(obj) + + mangledMethod = req.method.replace(".", "_") + self.log.debug("Looking for method '%s' in bridge", + mangledMethod) + try: + method = getattr(self._bridge, mangledMethod) + except AttributeError: + raise JsonRpcMethodNotFoundError() + else: + res = req.invokeFunction(method) + resp = self._generateResponse(req, res) + + except JsonRpcError as e: + resp = self._jsonError2Response(e, req) + self.log.error("Error processing request", exc_info=True) + except: + resp = self._jsonError2Response(JsonRpcInternalError(), req) + self.log.error("Unexpected error while processing request", + exc_info=True) + + if resp is None: + return + + # Notification don't respond even on errors + if req is not None and req.isNotification(): + return + + msgCtx.sendReply(resp) diff --git a/vdsm_api/jsonrpc/tcpReactor.py b/vdsm_api/jsonrpc/tcpReactor.py new file mode 100644 index 0000000..a815033 --- /dev/null +++ b/vdsm_api/jsonrpc/tcpReactor.py @@ -0,0 +1,157 @@ +# Copyright (C) 2012 Saggi Mizrahi, Red Hat Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public +# License along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA +import errno +import struct +import logging +import socket +from select import poll, POLLIN, POLLPRI, POLLERR, POLLHUP, POLLOUT + +from vdsm import utils + +_Size = struct.Struct("!Q") + + +class TCPConnection(object): + def __init__(self, conn, buffsize=(2 ** 20)): + self._conn = conn + conn.setblocking(False) + self.buffsize = buffsize + self._buffer = "" + self._sendData = [] + + def fileno(self): + return self._conn.fileno() + + def addSendData(self, data): + self._sendData.append(data) + + def hasSendData(self): + return len(self._sendData) > 0 + + def processOutput(self): + try: + if (len(self._sendData) > 0): + data = self._sendData[0] + bsent = self._conn.send(data) + if bsent < len(data): + self._sendData[0] = data[:bsent] + else: + self._sendData.pop(0) + except (OSError, IOError) as e: + if e.errno not in (errno.EINTR, errno.EAGAIN): + raise + + def processInput(self): + try: + self._buffer += self._conn.recv(self.buffsize - len(self._buffer)) + except (OSError, IOError) as e: + if e.errno not in (errno.EINTR, errno.EAGAIN): + raise + + buffLen = len(self._buffer) + if buffLen < _Size.size: + return + + if buffLen > self.buffsize: + self._conn.close() + self.sendData = [] + + msgLen = _Size.unpack(self._buffer[:_Size.size])[0] + if (buffLen - _Size.size) < msgLen: + return None + + msgStart = _Size.size + msgEnd = msgLen + _Size.size + res = self._buffer[msgStart:msgEnd] + self._buffer = self._buffer[msgEnd:] + return res + + +class TCPMessageContext(object): + + def __init__(self, server, conn, data): + self._server = server + self._conn = conn + self._data = data + + @property + def data(self): + return self._data + + def sendReply(self, data): + self._server.sendReply(self, data) + + +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 + self._messageHandler = messageHandler + self._inputEvent = utils.PollEvent() + # TODO: Close on exec + self._sock.bind(self._addr) + + def serve_forever(self): + sock = self._sock + sock.listen(1) + poller = poll() + poller.register(sock, POLLIN | POLLPRI) + poller.register(self._inputEvent, POLLIN | POLLPRI) + connections = {} + self.log.debug("Starting to accept clients") + while True: + for fd, conn in connections.iteritems(): + if conn.hasSendData(): + poller.modify(fd, (POLLIN | POLLPRI | POLLOUT)) + else: + poller.modify(fd, (POLLIN | POLLPRI)) + + for fd, ev in poller.poll(): + if ev & (POLLERR | POLLHUP): + if fd == sock.fileno(): + self.log.info("Listening socket closed, shutting down") + return + + self.log.debug("Connection closed") + poller.unregister(fd) + del connections[fd] + + elif fd == self._inputEvent.fileno(): + self._inputEvent.clear() + + elif fd == sock.fileno(): + conn, addr = sock.accept() + 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() + if msg is not None: + ctx = TCPMessageContext(self, conn, msg) + self._messageHandler.handleMessage(ctx) + if ev & POLLOUT: + conn.processOutput() + + def sendReply(self, ctx, message): + conn = ctx._conn + conn.addSendData(_Size.pack(len(message)) + message) + self._inputEvent.set() + + def stop(self): + self._sock.close()
-- To view, visit http://gerrit.ovirt.org/9986 To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange Gerrit-Change-Id: I7a65364824cc8433263ccf6ed44f1f97141eff2c Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Saggi Mizrahi smizrahi@redhat.com