diff options
author | Ross Brattain <ross.b.brattain@intel.com> | 2017-09-29 15:39:04 -0700 |
---|---|---|
committer | Ross Brattain <ross.b.brattain@intel.com> | 2017-10-01 13:18:08 -0700 |
commit | 75c0e3a54b8f6e8fd77c7d9d95decab830159929 (patch) | |
tree | 8c978f396e010d7031349afe7f2a5ecbd385dc0a | |
parent | 2cc3ef430319e1b3b4984737abc656349ae1bc5c (diff) |
cancel all queue join threads
In some cases we are blocking in base.Runner join() because the
queues are not empty
call cancel_join_thread to prevent the Queue from blocking the
Process exit
https://docs.python.org/3.3/library/multiprocessing.html#all-platforms
Joining processes that use queues
Bear in mind that a process that has put items in a queue will wait
before terminating until all the buffered items are fed by the
"feeder" thread to the underlying pipe. (The child process can call
the cancel_join_thread() method of the queue to avoid this behaviour.)
This means that whenever you use a queue you need to make sure that
all items which have been put on the queue will eventually be removed
before the process is joined. Otherwise you cannot be sure that
processes which have put items on the queue will terminate. Remember
also that non-daemonic processes will be joined automatically.
Warning
As mentioned above, if a child process has put items on a queue (and
it has not used JoinableQueue.cancel_join_thread), then that process
will not terminate until all buffered items have been flushed to the
pipe.
This means that if you try joining that process you may get a deadlock
unless you are sure that all items which have been put on the queue
have been consumed. Similarly, if the child process is non-daemonic
then the parent process may hang on exit when it tries to join all its
non-daemonic children.
cancel_join_thread()
Prevent join_thread() from blocking. In particular, this prevents the
background thread from being joined automatically when the process
exits – see join_thread().
A better name for this method might be allow_exit_without_flush(). It
is likely to cause enqueued data to lost, and you almost certainly
will not need to use it. It is really only there if you need the
current process to exit immediately without waiting to flush enqueued
data to the underlying pipe, and you don’t care about lost data.
Change-Id: I345c722a752bddf9f0824a11cdf52ae9f04669af
Signed-off-by: Ross Brattain <ross.b.brattain@intel.com>
-rwxr-xr-x | yardstick/benchmark/runners/arithmetic.py | 5 | ||||
-rwxr-xr-x | yardstick/benchmark/runners/base.py | 4 | ||||
-rwxr-xr-x | yardstick/benchmark/runners/dynamictp.py | 1 | ||||
-rw-r--r-- | yardstick/benchmark/runners/iteration.py | 5 | ||||
-rw-r--r-- | yardstick/benchmark/scenarios/availability/monitor/basemonitor.py | 1 |
5 files changed, 16 insertions, 0 deletions
diff --git a/yardstick/benchmark/runners/arithmetic.py b/yardstick/benchmark/runners/arithmetic.py index 7898ae2bc..974fb21b3 100755 --- a/yardstick/benchmark/runners/arithmetic.py +++ b/yardstick/benchmark/runners/arithmetic.py @@ -46,6 +46,11 @@ def _worker_process(queue, cls, method_name, scenario_cfg, sequence = 1 + # if we don't do this we can hang waiting for the queue to drain + # have to do this in the subprocess + queue.cancel_join_thread() + output_queue.cancel_join_thread() + runner_cfg = scenario_cfg['runner'] interval = runner_cfg.get("interval", 1) diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py index a69811f8a..57903ebb9 100755 --- a/yardstick/benchmark/runners/base.py +++ b/yardstick/benchmark/runners/base.py @@ -47,6 +47,7 @@ def _execute_shell_command(command): def _single_action(seconds, command, queue): """entrypoint for the single action process""" + queue.cancel_join_thread() log.debug("single action, fires after %d seconds (from now)", seconds) time.sleep(seconds) log.debug("single action: executing command: '%s'", command) @@ -61,6 +62,7 @@ def _single_action(seconds, command, queue): def _periodic_action(interval, command, queue): """entrypoint for the periodic action process""" + queue.cancel_join_thread() log.debug("periodic action, fires every: %d seconds", interval) time_spent = 0 while True: @@ -137,7 +139,9 @@ class Runner(object): self.config = config self.periodic_action_process = None self.output_queue = multiprocessing.Queue() + self.output_queue.cancel_join_thread() self.result_queue = multiprocessing.Queue() + self.result_queue.cancel_join_thread() self.process = None self.aborted = multiprocessing.Event() Runner.runners.append(self) diff --git a/yardstick/benchmark/runners/dynamictp.py b/yardstick/benchmark/runners/dynamictp.py index afff27d75..2f5f7e4f4 100755 --- a/yardstick/benchmark/runners/dynamictp.py +++ b/yardstick/benchmark/runners/dynamictp.py @@ -33,6 +33,7 @@ LOG = logging.getLogger(__name__) def _worker_process(queue, cls, method_name, scenario_cfg, context_cfg, aborted): # pragma: no cover + queue.cancel_join_thread() runner_cfg = scenario_cfg['runner'] iterations = runner_cfg.get("iterations", 1) interval = runner_cfg.get("interval", 1) diff --git a/yardstick/benchmark/runners/iteration.py b/yardstick/benchmark/runners/iteration.py index 50fe106bd..822e67723 100644 --- a/yardstick/benchmark/runners/iteration.py +++ b/yardstick/benchmark/runners/iteration.py @@ -36,6 +36,11 @@ def _worker_process(queue, cls, method_name, scenario_cfg, sequence = 1 + # if we don't do this we can hang waiting for the queue to drain + # have to do this in the subprocess + queue.cancel_join_thread() + output_queue.cancel_join_thread() + runner_cfg = scenario_cfg['runner'] interval = runner_cfg.get("interval", 1) diff --git a/yardstick/benchmark/scenarios/availability/monitor/basemonitor.py b/yardstick/benchmark/scenarios/availability/monitor/basemonitor.py index a6c1a28bd..871f13f88 100644 --- a/yardstick/benchmark/scenarios/availability/monitor/basemonitor.py +++ b/yardstick/benchmark/scenarios/availability/monitor/basemonitor.py @@ -90,6 +90,7 @@ class BaseMonitor(multiprocessing.Process): self._config = config self._context = context self._queue = multiprocessing.Queue() + self._queue.cancel_join_thread() self._event = multiprocessing.Event() self.monitor_data = data self.setup_done = False |