Per execution lock in scheduler
@@ -62,6 +62,7 @@ class ExecutionQueueHandler(object):
self.message_type = LiveActionDB
self._shutdown = False
self._coordinator = coordination.get_coordinator()
+ self._locks = {}
def loop(self):
LOG.debug('Entering scheduler loop')
@@ -71,9 +72,13 @@ class ExecutionQueueHandler(object):
with metrics.Timer(key='scheduler.loop'):
with self._coordinator.get_lock('st2shcedulerqueue'):
queued_executions = _next_executions()
+ for execution in queued_executions:
+ self._locks[str(execution.id)] = self._coordinator.get_lock(str(execution.id))
for execution in queued_executions:
with metrics.Timer(key='scheduler.handle_execution'):
+ lock = self._locks[str(execution.id)]
self._handle_execution(execution)
+ lock.release()
# try:
# except Exception as error:
# self._coordinator = coordination.get_coordinator()
