Change in vdsm[ovirt-3.6]: jsonrpc: executor based thread factory
piotr.kliczewski at gmail.com
piotr.kliczewski at gmail.com
Mon Nov 9 14:16:32 UTC 2015
Piotr Kliczewski has uploaded a new change for review.
Change subject: jsonrpc: executor based thread factory
......................................................................
jsonrpc: executor based thread factory
Creating new thread for every request is not efficient so we introduce
usage of the executor for request processing.
Change-Id: I56b307633a8bf7e4aad8f87cc97a4129c9ed0970
Signed-off-by: pkliczewski <piotr.kliczewski at gmail.com>
Reviewed-on: https://gerrit.ovirt.org/43759
---
M lib/vdsm/config.py.in
M tests/integration/jsonRpcHelper.py
M tests/integration/jsonRpcTests.py
M vdsm/clientIF.py
M vdsm/rpc/bindingjsonrpc.py
M vdsm/vdsm
6 files changed, 70 insertions(+), 16 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/93/48293/1
diff --git a/lib/vdsm/config.py.in b/lib/vdsm/config.py.in
index 9ed11ff..3c53e5f 100644
--- a/lib/vdsm/config.py.in
+++ b/lib/vdsm/config.py.in
@@ -219,6 +219,18 @@
'There are 2 options: '
'"m2c" to use the m2crypto module '
'"ssl" to use the standard python ssl module'),
+
+ ]),
+
+ # Section: [rpc]
+ ('rpc', [
+
+ ('worker_threads', '8',
+ 'Number of worker threads to serve jsonrpc server.'),
+
+ ('tasks_per_worker', '10',
+ 'Max number of tasks which can be queued per workers.'),
+
]),
# Section: [mom]
diff --git a/tests/integration/jsonRpcHelper.py b/tests/integration/jsonRpcHelper.py
index 3abbfbc..b88189c 100644
--- a/tests/integration/jsonRpcHelper.py
+++ b/tests/integration/jsonRpcHelper.py
@@ -37,6 +37,7 @@
from protocoldetector import MultiProtocolAcceptor
from rpc.bindingjsonrpc import BindingJsonRpc
from vdsm.config import config
+from vdsm import schedule
from vdsm import utils
if config.get('vars', 'ssl_implementation') == 'm2c':
@@ -92,7 +93,11 @@
0,
sslctx,
)
- json_binding = BindingJsonRpc(jsonBridge, defaultdict(list), 60)
+
+ scheduler = schedule.Scheduler(name="test.Scheduler",
+ clock=utils.monotonic_time)
+ scheduler.start()
+ json_binding = BindingJsonRpc(jsonBridge, defaultdict(list), 60, scheduler)
json_binding.start()
cif = FakeClientIf(json_binding, dest)
@@ -118,6 +123,7 @@
acceptor.stop()
json_binding.stop()
xml_binding.stop()
+ scheduler.stop(wait=False)
@contextmanager
diff --git a/tests/integration/jsonRpcTests.py b/tests/integration/jsonRpcTests.py
index b7fd97d..16e7607 100644
--- a/tests/integration/jsonRpcTests.py
+++ b/tests/integration/jsonRpcTests.py
@@ -23,6 +23,7 @@
from contextlib import contextmanager
from monkeypatch import MonkeyPatch
from testValidation import slowtest
+from vdsm import executor
from testlib import VdsmTestCase as TestCaseBase, \
expandPermutations, \
@@ -78,6 +79,10 @@
def getInstance():
return FakeClientIf()
+
+
+def dispatch(callable, timeout=None):
+ raise executor.TooManyTasks
@expandPermutations
@@ -222,3 +227,20 @@
self.assertEquals(cm.exception.code,
JsonRpcNoResponseError().code)
+
+ @MonkeyPatch(clientIF, 'getInstance', getInstance)
+ @MonkeyPatch(executor.Executor, 'dispatch', dispatch)
+ @permutations(PERMUTATIONS)
+ def testFullExecutor(self, ssl, type):
+ bridge = _DummyBridge()
+ with constructClient(self.log, bridge, ssl, type) as clientFactory:
+ with self._client(clientFactory) as client:
+ if type == "xml":
+ # TODO start using executor for xmlrpc
+ pass
+ else:
+ with self.assertRaises(JsonRpcError) as cm:
+ self._callTimeout(client, "no_method", [], CALL_ID)
+
+ self.assertEquals(cm.exception.code,
+ JsonRpcInternalError().code)
diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index a7a0c66..918ca8b 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -71,7 +71,7 @@
_instance = None
_instanceLock = threading.Lock()
- def __init__(self, irs, log):
+ def __init__(self, irs, log, scheduler):
"""
Initialize the (single) clientIF instance
@@ -95,6 +95,7 @@
self.bindings = {}
self._broker_client = None
self._subscriptions = defaultdict(list)
+ self._scheduler = scheduler
if _glusterEnabled:
self.gluster = gapi.GlusterApi(self, log)
else:
@@ -182,14 +183,14 @@
vmObj.cont()
@classmethod
- def getInstance(cls, irs=None, log=None):
+ def getInstance(cls, irs=None, log=None, scheduler=None):
with cls._instanceLock:
if cls._instance is None:
if log is None:
raise Exception("Logging facility is required to create "
"the single clientIF instance")
else:
- cls._instance = clientIF(irs, log)
+ cls._instance = clientIF(irs, log, scheduler)
return cls._instance
def _createAcceptor(self, host, port):
@@ -248,7 +249,8 @@
bridge = Bridge.DynamicBridge()
json_binding = BindingJsonRpc(
bridge, self._subscriptions,
- config.getint('vars', 'connection_stats_timeout'))
+ config.getint('vars', 'connection_stats_timeout'),
+ self._scheduler)
self.bindings['jsonrpc'] = json_binding
stomp_detector = StompDetector(json_binding)
self._acceptor.add_detector(stomp_detector)
diff --git a/vdsm/rpc/bindingjsonrpc.py b/vdsm/rpc/bindingjsonrpc.py
index 80e3388..0e066eb 100644
--- a/vdsm/rpc/bindingjsonrpc.py
+++ b/vdsm/rpc/bindingjsonrpc.py
@@ -19,18 +19,26 @@
from yajsonrpc import JsonRpcServer
from yajsonrpc.stompreactor import StompReactor
+from vdsm import executor
+from vdsm.config import config
-def _simpleThreadFactory(func):
- t = threading.Thread(target=func)
- t.setDaemon(False)
- t.start()
+
+# TODO test what should be the default values
+_THREADS = config.getint('rpc', 'worker_threads')
+_TASK_PER_WORKER = config.getint('rpc', 'tasks_per_worker')
+_TASKS = _THREADS * _TASK_PER_WORKER
class BindingJsonRpc(object):
log = logging.getLogger('BindingJsonRpc')
- def __init__(self, bridge, subs, timeout):
- self._server = JsonRpcServer(bridge, timeout, _simpleThreadFactory)
+ def __init__(self, bridge, subs, timeout, scheduler):
+ self._executor = executor.Executor(name="jsonrpc.Executor",
+ workers_count=_THREADS,
+ max_tasks=_TASKS,
+ scheduler=scheduler)
+
+ self._server = JsonRpcServer(bridge, timeout, self._executor.dispatch)
self._reactor = StompReactor(subs)
self.startReactor()
@@ -45,6 +53,8 @@
return self._reactor
def start(self):
+ self._executor.start()
+
t = threading.Thread(target=self._server.serve_requests,
name='JsonRpcServer')
t.setDaemon(True)
@@ -60,3 +70,4 @@
def stop(self):
self._server.stop()
self._reactor.stop()
+ self._executor.stop()
diff --git a/vdsm/vdsm b/vdsm/vdsm
index 5f7b39e..867e900 100755
--- a/vdsm/vdsm
+++ b/vdsm/vdsm
@@ -93,14 +93,15 @@
except:
utils.panic("Error initializing IRS")
- from clientIF import clientIF # must import after config is read
- cif = clientIF.getInstance(irs, log)
-
- install_manhole({'irs': irs, 'cif': cif})
-
scheduler = schedule.Scheduler(name="vdsm.Scheduler",
clock=utils.monotonic_time)
scheduler.start()
+
+ from clientIF import clientIF # must import after config is read
+ cif = clientIF.getInstance(irs, log, scheduler)
+
+ install_manhole({'irs': irs, 'cif': cif})
+
cif.start()
periodic.start(cif, scheduler)
try:
--
To view, visit https://gerrit.ovirt.org/48293
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I56b307633a8bf7e4aad8f87cc97a4129c9ed0970
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: ovirt-3.6
Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski at gmail.com>
More information about the vdsm-patches
mailing list