Change in vdsm[master]: executor: Optimize locking
nsoffer at redhat.com
nsoffer at redhat.com
Sun Aug 3 23:58:20 UTC 2014
Nir Soffer has uploaded a new change for review.
Change subject: executor: Optimize locking
......................................................................
executor: Optimize locking
Profiles show that locking is our biggest overhead in the executor. This
patch optimize locking in the executor:
- Remove the _Worker lock, it should work correctly without a lock.
- Separate the Executor and _TaskQueue lockinng. The Eecutor uses now a
lock, use to start, stop and resize the exectutor. These operations
are rare.
- Move the Condition into the _TaskQueue, and use it only for waiting
when the deque is empty. deque is thread safe providing fast atomic
append and popleft methods.
Change-Id: I1df60abe8c85b57030494a5c031a0db35e6d8354
Signed-off-by: Nir Soffer <nsoffer at redhat.com>
---
M lib/vdsm/executor.py
1 file changed, 46 insertions(+), 41 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/86/30986/1
diff --git a/lib/vdsm/executor.py b/lib/vdsm/executor.py
index 64384f3..02a80b2 100644
--- a/lib/vdsm/executor.py
+++ b/lib/vdsm/executor.py
@@ -52,12 +52,12 @@
self._tasks = _TaskQueue(max_tasks)
self._scheduler = scheduler
self._workers = set()
- self._cond = threading.Condition(threading.Lock())
+ self._lock = threading.Lock()
self._running = False
def start(self):
self._log.debug('Starting executor')
- with self._cond:
+ with self._lock:
if self._running:
raise AlreadyStarted()
self._running = True
@@ -66,11 +66,12 @@
def stop(self, wait=True):
self._log.debug('Stopping executor')
- with self._cond:
+ with self._lock:
self._running = False
self._tasks.clear()
+ for i in range(self._workers_count):
+ self._tasks.put((_STOP, 0))
workers = tuple(self._workers) if wait else ()
- self._cond.notify_all() # wake up all the workers
for worker in workers:
try:
worker.join()
@@ -83,11 +84,9 @@
The task may be any callable.
The task will be executed as soon as possible
in one of the active workers of the executor."""
- with self._cond:
- if not self._running:
- raise NotRunning()
- self._tasks.put((callable, timeout))
- self._cond.notify() # wake up one worker to handle this task
+ if not self._running:
+ raise NotRunning()
+ self._tasks.put((callable, timeout))
# Serving workers
@@ -96,7 +95,7 @@
Called from shceduler thread when worker was discarded. The worker
thread is blocked on a task, and will exit when the task finish.
"""
- with self._cond:
+ with self._lock:
if self._running:
self._add_worker()
@@ -104,7 +103,7 @@
"""
Called from worker thread before it exit.
"""
- with self._cond:
+ with self._lock:
self._workers.remove(worker)
def _next_task(self):
@@ -112,18 +111,19 @@
Called from worker thread to get the next task from the taks queue.
Raises NotRunning exception if executor was stopped.
"""
- with self._cond:
- while self._running and self._tasks.empty():
- self._cond.wait()
- if not self._running:
- raise NotRunning()
- return self._tasks.get()
+ task, timeout = self._tasks.get()
+ if task is _STOP:
+ raise NotRunning()
+ return task, timeout
# Private
def _add_worker(self):
worker = _Worker(self, self._scheduler)
self._workers.add(worker)
+
+
+_STOP = object()
class _WorkerBusy(Exception):
@@ -142,7 +142,6 @@
def __init__(self, executor, scheduler):
self._executor = executor
self._scheduler = scheduler
- self._lock = threading.Lock()
self._busy = False
self._discarded = False
_Worker._id += 1
@@ -157,9 +156,8 @@
return self._thread.name
def join(self):
- with self._lock:
- if self._busy:
- raise _WorkerBusy()
+ if self._busy:
+ raise _WorkerBusy()
self._log.debug('Waiting for worker %s', self.name)
self._thread.join()
@@ -179,17 +177,15 @@
def _execute_task(self):
callable, timeout = self._executor._next_task()
discard = self._discard_after(timeout)
- with self._lock:
- self._busy = True
+ self._busy = True
try:
callable()
except Exception:
self._log.exception("Unhandled exception in %s", callable)
finally:
- with self._lock:
- self._busy = False
- if self._discarded:
- raise _WorkerDiscarded()
+ self._busy = False
+ if self._discarded:
+ raise _WorkerDiscarded()
if discard is not None:
discard.cancel()
@@ -199,34 +195,43 @@
return None
def _discard(self):
- with self._lock:
- if self._discarded:
- raise AssertionError("Attempt to discard worker twice")
- if not self._busy:
- return
- self._log.debug("Discarding worker %s", self.name)
- self._discarded = True
+ if self._discarded:
+ raise AssertionError("Attempt to discard worker twice")
+ if not self._busy:
+ return
+ self._discarded = True
+ self._log.debug("Worker %s discarded", self.name)
self._executor._worker_discarded(self)
class _TaskQueue(object):
- """Queue.Queue doesn't support the clear() operation.
- All the locking must be provided by the caller."""
+ """ Queue.Queue doesn't support the clear() operation. """
def __init__(self, max_tasks):
self._max_tasks = max_tasks
self._tasks = collections.deque()
-
- def empty(self):
- return len(self._tasks) == 0
+ # deque is thread safe - we can append and pop from both ends without
+ # additioanl locking. We need this condition only for waiting.
+ self._cond = threading.Condition(threading.Lock())
+ self._waiters = 0
def put(self, task):
if len(self._tasks) == self._max_tasks:
raise TooManyTasks()
- self._tasks.appendleft(task)
+ self._tasks.append(task)
+ if self._waiters > 0:
+ with self._cond:
+ self._cond.notify()
def get(self):
- return self._tasks.pop()
+ while True:
+ try:
+ return self._tasks.popleft()
+ except IndexError:
+ with self._cond:
+ self._waiters += 1
+ self._cond.wait()
+ self._waiters -= 1
def clear(self):
self._tasks.clear()
--
To view, visit http://gerrit.ovirt.org/30986
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I1df60abe8c85b57030494a5c031a0db35e6d8354
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer at redhat.com>
More information about the vdsm-patches
mailing list