Merge pull request #419 from pigmej/ci-locks

Added locks for python CI and for CI processors
Mirantis · Dec 9, 2015 · 2c94df6 · 2c94df6
2 parents 5f1732b + 3467d48
commit 2c94df6
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 4 deletions.
diff --git a/solar/computable_inputs/ci_python.py b/solar/computable_inputs/ci_python.py
@@ -22,29 +22,44 @@
 from solar.computable_inputs import ComputablePassedTypes
 from solar.computable_inputs import HELPERS_PATH
 
+from functools import wraps
+from threading import RLock
+
 
 _PYTHON_WORKER = os.path.join(HELPERS_PATH, 'python_loop.py')
 _PYTHON_HELPERS = open(os.path.join(HELPERS_PATH, 'python_helpers.py')).read()
 
 
+def with_lock(f):
+    @wraps(f)
+    def _inner(obj, *args, **kwargs):
+        with obj.lock:
+            return f(obj, *args, **kwargs)
+    return _inner
+
+
 class Mgr(object):
 
     def __init__(self):
         self.child = None
+        self.lock = RLock()
 
+    @with_lock
     def run(self):
         self.child = subprocess.Popen(['/usr/bin/env', 'python',
                                        _PYTHON_WORKER],
                                       stdin=subprocess.PIPE,
                                       stdout=subprocess.PIPE)
         self.prepare()
 
+    @with_lock
     def prepare(self):
         self.run_code(fname=None,
                       code=_PYTHON_HELPERS,
                       kwargs={},
                       copy_env=False)
 
+    @with_lock
     def kill_child(self):
         self.child.kill()
 
@@ -77,6 +92,7 @@ def read(self):
         data = self.child.stdout.read(dlen)
         return json.loads(data)
 
+    @with_lock
     def run_code(self, fname, code, kwargs, copy_env=True):
         self.send({'fname': fname,
                    'code': code,

diff --git a/solar/computable_inputs/processor.py b/solar/computable_inputs/processor.py
@@ -14,6 +14,8 @@
 
 _av_processors = {}
 
+from threading import RLock
+
 try:
     from solar.computable_inputs.ci_lua import LuaProcessor
 except ImportError:
@@ -40,14 +42,16 @@
 
 
 _processors = {}
+_lock = RLock()
 
 
 def get_processor(resource, input_name, computable_type, data, other=None):
     computable = resource.meta_inputs[input_name]['computable']
     lang = computable['lang']
     funct = computable['func']
-    if lang not in _processors:
-        _processors[lang] = processor = _av_processors[lang]()
-    else:
-        processor = _processors[lang]
+    with _lock:
+        if lang not in _processors:
+            _processors[lang] = processor = _av_processors[lang]()
+        else:
+            processor = _processors[lang]
     return processor.process(resource.name, computable_type, funct, data)