Change in vdsm[master]: Run JsonRpcServer requests in their own thread instead of th...
smizrahi at redhat.com
smizrahi at redhat.com
Wed Dec 19 21:35:36 UTC 2012
Saggi Mizrahi has uploaded a new change for review.
Change subject: Run JsonRpcServer requests in their own thread instead of the transport thread
......................................................................
Run JsonRpcServer requests in their own thread instead of the transport thread
This makes sure there is a single sync point for all json-rpc requests
and that handling a request never blocks the transport for processing
messages.
Change-Id: Ifccf3fd0e3b1bec7db475b2674cff09bb920738e
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M tests/jsonRpcTests.py
M vdsm_api/BindingJsonRpc.py
M vdsm_api/jsonrpc/__init__.py
3 files changed, 112 insertions(+), 34 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/52/10252/1
diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index b495541..da8ff5d 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -116,10 +116,17 @@
t.setDaemon(True)
t.start()
+ t = threading.Thread(target=server.serve_requests)
+ t.setDaemon(True)
+ t.start()
+
def jsonClientFactory():
return JsonRpcClient(clientFactory())
- yield server, jsonClientFactory
+ try:
+ yield server, jsonClientFactory
+ finally:
+ server.stop()
class _EchoMessageHandler(object):
diff --git a/vdsm_api/BindingJsonRpc.py b/vdsm_api/BindingJsonRpc.py
index 49f45df..645428e 100644
--- a/vdsm_api/BindingJsonRpc.py
+++ b/vdsm_api/BindingJsonRpc.py
@@ -71,6 +71,12 @@
t.setDaemon(True)
t.start()
+ t = threading.Thread(target=self.server.serve_requests,
+ name='JsonRpc (Rquest Processing)')
+ t.setDaemon(True)
+ t.start()
+
def prepareForShutdown(self):
+ self.server.stop()
for reactor in self._reactors:
reactor.stop()
diff --git a/vdsm_api/jsonrpc/__init__.py b/vdsm_api/jsonrpc/__init__.py
index f12c808..8e58ffe 100644
--- a/vdsm_api/jsonrpc/__init__.py
+++ b/vdsm_api/jsonrpc/__init__.py
@@ -14,6 +14,7 @@
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
import json
import logging
+from Queue import Queue
__all__ = ["tcpReactor"]
@@ -57,11 +58,13 @@
JsonRpcError.__init__(self, -32603, msg)
-class JsonRpcRequest(object):
- def __init__(self, method, params, reqId):
- self.method = method
+class _JsonRpcRequest(object):
+ def __init__(self, ctx, queue, methodName, params, reqId):
+ self.method = methodName
self.params = params
self.id = reqId
+ self._ctx = ctx
+ self._queue = queue
def invokeFunction(self, func):
if isinstance(self.params, list):
@@ -72,12 +75,27 @@
def isNotification(self):
return self.id is None
+ def sendReply(self, result, error):
+ # TBD: Should calling this for a notification raise an error or be
+ # ignored
+ self._queue.put_nowait(_JsonRpcResponse(self._ctx, self.id, result,
+ error))
+
+
+class _JsonRpcResponse(object):
+ def __init__(self, ctx, reqId, result, error):
+ self.ctx = ctx
+ self.result = result
+ self.error = error
+ self.id = reqId
+
class JsonRpcServer(object):
log = logging.getLogger("JsonRpcServer")
def __init__(self, bridge):
self._bridge = bridge
+ self._workQueue = Queue()
def _parseMessage(self, msg):
try:
@@ -85,7 +103,7 @@
except:
raise JsonRpcParseError()
- def _parseRequest(self, obj):
+ def _parseRequest(self, obj, ctx, queue):
if obj.get("jsonrpc") != "2.0":
raise JsonRpcInvalidRequestError()
@@ -101,7 +119,7 @@
if not isinstance(params, (list, dict)):
raise JsonRpcInvalidRequestError()
- return JsonRpcRequest(method, params, reqId)
+ return _JsonRpcRequest(ctx, queue, method, params, reqId)
def _jsonError2Response(self, err, req):
respId = None
@@ -112,43 +130,90 @@
"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 _generateResponse(self, resp):
+ res = {"jsonrpc": "2.0",
+ "id": resp.id}
+ if resp.error is not None:
+ res['error'] = {'code': resp.error.code,
+ 'message': resp.error.message}
+ else:
+ res['result'] = resp.result
+
+ return json.dumps(res, 'utf-8')
+
+ def _serveRequest(self, req):
+ mangledMethod = req.method.replace(".", "_")
+ self.log.debug("Looking for method '%s' in bridge",
+ mangledMethod)
+ try:
+ method = getattr(self._bridge, mangledMethod)
+ except AttributeError:
+ req.sendReply(None, JsonRpcMethodNotFoundError())
+ else:
+ try:
+ res = req.invokeFunction(method)
+ except JsonRpcError as e:
+ req.sendReply(None, e)
+ except Exception as e:
+ req.sendReply(None, JsonRpcInternalError(str(e)))
+ else:
+ return req.sendReply(res, None)
+
+ def _processResponse(self, resp):
+ try:
+ msg = self._generateResponse(resp)
+ except Exception as e:
+ # Probably result failed to be serialized as json
+ errResp = _JsonRpcResponse(resp.ctx,
+ resp.id,
+ None,
+ JsonRpcInternalError(str(e)))
+
+ msg = self._generateResponse(errResp)
+
+ resp.ctx.sendReply(msg)
+
+ def serve_requests(self):
+ while True:
+ obj = self._workQueue.get()
+ if obj is None:
+ break
+
+ if isinstance(obj, _JsonRpcRequest):
+ self._serveRequest(obj)
+ else:
+ self._processResponse(obj)
def handleMessage(self, msgCtx):
#TODO: support batch requests
req = None
- resp = None
+ error = None
try:
obj = self._parseMessage(msgCtx.data)
- req = self._parseRequest(obj)
+ req = self._parseRequest(obj, msgCtx, self._workQueue)
- 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:
+ self._workQueue.put_nowait(req)
return
- # Notification don't respond even on errors
+ except JsonRpcError as e:
+ self.log.error("Error processing request", exc_info=True)
+ error = e
+ 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():
return
- msgCtx.sendReply(resp)
+ if req is None:
+ resp = _JsonRpcResponse(msgCtx, None, None, error)
+ else:
+ resp = _JsonRpcResponse(msgCtx, req.id, None, error)
+
+ self._workQueue.put_nowait(resp)
+
+ def stop(self):
+ self._workQueue.put_nowait(None)
--
To view, visit http://gerrit.ovirt.org/10252
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ifccf3fd0e3b1bec7db475b2674cff09bb920738e
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