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