9285268044

Committed 29 May 2024 11:28AM UTC coverage: 62.506%. Remained the same

Build # 9285268044

Build Type

push

github

Committed by

ylobankov

Commit Message

dispatcher: lift pipe buffer size restriction

A task queue dispatcher puts all the tasks to the task queue at startup.
Then workers are started and are taking the tasks from it.

If there are many tasks in a task group (which roughly corresponds to a
test suite), we can reach the pipe buffer size on putting into the
queue, because `multiprocessing.SimpleQueue` uses a pipe under the hood.

The solution is to use `multiprocessing.Queue`, which has an
intermediate buffer before the underlying pipe and writes to the pipe in
a background thread, without blocking a thread that calls
`<queue>.put()`.

The `Queue` API is a superset of the `SimpleQueue` API, so we can just
replace the implementation.

Let's also use `Queue` for the worker's output queue to be on the safe
side and for consistency.

Fixes #287

Run Details

759 of 1564 branches covered (48.53%)

Branch coverage included in aggregate %.

3 of 4 new or added lines in 1 file covered. (75.0%)

1 existing line in 1 file now uncovered.

2937 of 4349 relevant lines covered (67.53%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.11

/dispatcher.py

import os
import signal
import time
import select
import random
import functools
import yaml

import multiprocessing

# Queue is available from multiprocessing.queues on all Python
# versions known at the moment of writting the code (up to 3.12).
#
# However the mandatory argument 'ctx'
# (see multiprocessing.get_context()) was added to the constructor
# of Queue from multiprocessing.queues since Python 3.4 ([1]).
#
# So we should import Queue from multiprocessing on Python 3.4+
# to uniformly instantiate it (without constructor arguments).
#
# [1]: https://bugs.python.org/issue18999
try:
    # Python 3.4+
    from multiprocessing import Queue
except ImportError:
    # Python 2
    from multiprocessing.queues import Queue

from lib import Options
from lib.sampler import sampler
from lib.utils import set_fd_cloexec
from lib.worker import WorkerTaskResult, WorkerDone
from lib.colorer import color_stdout
from listeners import ArtifactsWatcher
from listeners import FailWatcher
from listeners import HangWatcher
from listeners import LogOutputWatcher
from listeners import OutputWatcher
from listeners import StatisticsWatcher


class Dispatcher:
    """Run specified count of worker processes ('max_workers_cnt' arg), pass
    task IDs (via 'task_queue'), receive results and output (via
    'result_queue') and pass it to listeners. Workers as well as tasks have
    types and certain task can be run only on worker of that type. To being
    abstract we get 'task_groups' argument contains worker generators (the
    callable working as factory of workers) and IDs of task that can be
    executed on such workers. The structure of this argument is the following:
    ```
    task_groups = {
        'some_key_1': {
            'gen_worker': function,
            'task_ids': list,
            'is_parallel': bool,
            'show_reproduce_content': bool,
        }
        ...
    }

    ```
    Usage (simplified and w/o exception catching):
    ```
    task_groups = ...
    dispatcher = Dispatcher(task_groups, max_workers_count=8, randomize=True)
    dispatcher.start()
    dispatcher.wait()
    dispatcher.statistics.print_statistics()
    dispatcher.wait_processes()
    ```
    """
    def __init__(self, task_groups, max_workers_cnt, randomize):
        self.pids = []
        self.processes = []
        self.result_queues = []
        self.task_queues = []
        self.workers_cnt = 0
        self.worker_next_id = 1

        tasks_cnt = 0
        self.task_queue_disps = dict()
        for key, task_group in task_groups.items():
            tasks_cnt += len(task_group['task_ids'])
            task_queue_disp = TaskQueueDispatcher(key, task_group, randomize)
            self.task_queue_disps[key] = task_queue_disp
            self.result_queues.append(task_queue_disp.result_queue)
            self.task_queues.append(task_queue_disp.task_queue)
        self.total_tasks_cnt = tasks_cnt

        self.report_timeout = 0.1

        self.statistics = None
        self.artifacts = None
        self.fail_watcher = None
        self.listeners = None
        self.init_listeners()

        self.max_workers_cnt = min(max_workers_cnt, tasks_cnt)

        self.pid_to_worker_id = dict()
        self.worker_id_to_pid = dict()

        self.randomize = randomize

    def terminate_all_workers(self):
        for process in self.processes:
            if process.is_alive():
                try:
                    process.terminate()
                except OSError:
                    pass

    def kill_all_workers(self):
        for pid in self.pids:
            try:
                os.kill(pid, signal.SIGKILL)
            except OSError:
                pass

    def init_listeners(self):
        args = Options().args
        watch_hang = args.no_output_timeout >= 0 and \
            not args.gdb and \
            not args.gdbserver and \
            not args.lldb and \
            not args.valgrind
        watch_fail = not Options().args.is_force

        log_output_watcher = LogOutputWatcher()
        self.statistics = StatisticsWatcher(log_output_watcher.get_logfile,
                                            self.total_tasks_cnt)
        self.artifacts = ArtifactsWatcher(log_output_watcher.get_logfile)
        output_watcher = OutputWatcher()
        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts]
        if sampler.is_enabled:
            self.listeners.append(sampler.watcher)
        if watch_fail:
            self.fail_watcher = FailWatcher(self.terminate_all_workers)
            self.listeners.append(self.fail_watcher)
        if watch_hang:
            warn_timeout = 60.0 if args.long else 10.0
            hang_watcher = HangWatcher(output_watcher.not_done_worker_ids,
                                       self.kill_all_workers, warn_timeout,
                                       float(args.no_output_timeout))
            self.listeners.append(hang_watcher)

    def run_max_workers(self):
        ok = True
        new_workers_cnt = self.max_workers_cnt - self.workers_cnt
        while ok and new_workers_cnt > 0:
            ok = self.add_worker()
            new_workers_cnt = self.max_workers_cnt - self.workers_cnt

    def start(self):
        self.run_max_workers()

    def find_nonempty_task_queue_disp(self):
        """Find TaskQueueDispatcher that doesn't reported it's 'done' (don't
        want more workers created for working on its task queue).
        """
        task_queue_disps_rnd = list(
            self.task_queue_disps.values())
        if self.randomize:
            random.shuffle(task_queue_disps_rnd)
        # run all parallel groups first
        for task_queue_disp in task_queue_disps_rnd:
            if not task_queue_disp.is_parallel:
                continue
            if task_queue_disp.done:
                continue
            return task_queue_disp
        # then run all rest groups in a sequence
        self.max_workers_cnt = 1
        for task_queue_disp in task_queue_disps_rnd:
            if len(task_queue_disp.worker_ids) > 0:
                continue
            if task_queue_disp.done:
                continue
            return task_queue_disp
        return None

    def get_task_queue_disp(self, worker_id):
        """Get TaskQueueDispatcher instance which contains certain worker by
        worker_id.
        """
        for task_queue_disp in self.task_queue_disps.values():
            if worker_id in task_queue_disp.worker_ids:
                return task_queue_disp
        return None

    def add_worker(self):
        # don't add new workers if fail occured and --force not passed
        if self.fail_watcher and self.fail_watcher.got_fail:
            return False
        task_queue_disp = self.find_nonempty_task_queue_disp()
        if not task_queue_disp:
            return False
        # self.max_workers_cnt can be changed in
        # find_nonempty_task_queue_disp()
        if self.workers_cnt >= self.max_workers_cnt:
            return False
        process = task_queue_disp.add_worker(self.worker_next_id)
        self.processes.append(process)
        self.pids.append(process.pid)
        self.pid_to_worker_id[process.pid] = self.worker_next_id
        self.worker_id_to_pid[self.worker_next_id] = process.pid

        self.workers_cnt += 1
        self.worker_next_id += 1

        return True

    def del_worker(self, worker_id):
        pid = self.worker_id_to_pid[worker_id]

        task_queue_disp = self.get_task_queue_disp(worker_id)
        task_queue_disp.del_worker(worker_id)
        self.workers_cnt -= 1

        self.pids.remove(pid)
        del self.worker_id_to_pid[worker_id]
        del self.pid_to_worker_id[pid]
        for process in self.processes:
            if process.pid == pid:
                self.processes.remove(process)
                break

    def mark_task_done(self, worker_id, task_id):
        task_queue_disp = self.get_task_queue_disp(worker_id)
        task_queue_disp.mark_task_done(task_id)

    def undone_tasks(self):
        res = []
        for task_queue_disp in self.task_queue_disps.values():
            res.extend(task_queue_disp.undone_tasks())
        return res

    def report_undone(self, verbose):
        undone = self.undone_tasks()
        if not bool(undone):
            return False
        if verbose:
            color_stdout(
                '[Internal test-run error] '
                'The following tasks were dispatched to some worker task '
                'queue, but were not reported as done (does not matters '
                'success or fail):\n', schema='test_var')
            for task_id in undone:
                task_id_str = yaml.safe_dump(task_id, default_flow_style=True)
                color_stdout('- %s' % task_id_str)
        else:
            # Visually continue StatisticsWatcher.print_statistics() output.
            color_stdout('* undone: %d\n' % len(undone), schema='test_var')
        return True

    def wait(self):
        """Wait all workers reported its done via result_queues. But in the
        case when some worker process terminated prematurely 'invoke_listeners'
        can add fake WorkerDone markers (see also 'check_for_dead_processes').
        """
        while self.workers_cnt > 0:
            try:
                inputs = [q._reader for q in self.result_queues]
                ready_inputs, _, _ = select.select(
                    inputs, [], [], self.report_timeout)
            except KeyboardInterrupt:
                self.flush_ready(inputs)
                raise

            objs = self.invoke_listeners(inputs, ready_inputs)
            for obj in objs:
                if isinstance(obj, WorkerTaskResult):
                    self.mark_task_done(obj.worker_id, obj.task_id)
                elif isinstance(obj, WorkerDone):
                    self.del_worker(obj.worker_id)
            if not objs:
                self.check_for_dead_processes()

            self.run_max_workers()

    def invoke_listeners(self, inputs, ready_inputs):
        """Returns received objects from result queue to allow Dispatcher
        update its structures.
        """
        # process timeout
        if not ready_inputs:
            for listener in self.listeners:
                listener.process_timeout(self.report_timeout)
            return []

        # collect received objects
        objs = []
        for ready_input in ready_inputs:
            result_queue = self.result_queues[inputs.index(ready_input)]
            while not result_queue.empty():
                objs.append(result_queue.get())

        # process received objects
        for obj in objs:
            for listener in self.listeners:
                listener.process_result(obj)

        return objs

    def flush_ready(self, inputs):
        """Write output from workers to stdout."""
        # leave only output listeners in self.listeners
        new_listeners = []
        for listener in self.listeners:
            if isinstance(listener, (LogOutputWatcher,
                                     OutputWatcher)):
                listener.report_at_timeout = False
                new_listeners.append(listener)
        self.listeners = new_listeners
        # wait some time until processes in our group get its SIGINTs and give
        # us some last output
        time.sleep(0.1)
        # collect and process ready inputs
        ready_inputs, _, _ = select.select(inputs, [], [], 0)
        self.invoke_listeners(inputs, ready_inputs)

    def check_for_dead_processes(self):
        for pid in self.pids:
            exited = False
            try:
                os.waitpid(pid, os.WNOHANG)
            except OSError:
                exited = True
            if exited:
                worker_id = self.pid_to_worker_id[pid]
                color_stdout(
                    "[Main process] Worker %d don't reported work "
                    "done using results queue, but the corresponding "
                    "process seems dead. Removing it from Dispatcher.\n"
                    % worker_id, schema='test_var')
                self.del_worker(worker_id)

    def wait_processes(self):
        for process in self.processes:
            process.join()
        self.processes = []


class TaskQueueDispatcher:
    """Incapsulate data structures necessary for dispatching workers working on
    the one task queue.
    """
    def __init__(self, key, task_group, randomize):
        self.key = key
        self.gen_worker = task_group['gen_worker']
        self.task_ids = task_group['task_ids']
        self.is_parallel = task_group['is_parallel']
        if self.is_parallel:
            self.randomize = randomize
            if self.randomize:
                random.shuffle(self.task_ids)
        else:
            self.randomize = False
        self.result_queue = Queue()
        self.task_queue = Queue()

        # Don't expose queues file descriptors over Popen to, say, tarantool
        # running tests.
        set_fd_cloexec(self.result_queue._reader.fileno())
        set_fd_cloexec(self.result_queue._writer.fileno())
        set_fd_cloexec(self.task_queue._reader.fileno())
        set_fd_cloexec(self.task_queue._writer.fileno())

        for task_id in self.task_ids:
            self.task_queue.put(task_id)
        self.worker_ids = set()
        self.done = False
        self.done_task_ids = set()

    def _run_worker(self, worker_id):
        """Entry function for worker processes."""
        os.environ['TEST_RUN_WORKER_ID'] = str(worker_id)
        color_stdout.queue = self.result_queue
        worker = self.gen_worker(worker_id)
        sampler.set_queue(self.result_queue, worker_id, worker.name)
        worker.run_all(self.task_queue, self.result_queue)

    def add_worker(self, worker_id):
        # Note: each of our workers should consume only one None, but for the
        # case of abnormal circumstances we listen for processes termination
        # (method 'check_for_dead_processes') and for time w/o output from
        # workers (class 'HangWatcher').
        self.task_queue.put(None)  # 'stop worker' marker

        entry = functools.partial(self._run_worker, worker_id)

        self.worker_ids.add(worker_id)
        process = multiprocessing.Process(target=entry)
        process.start()
        return process

    def del_worker(self, worker_id):
        self.worker_ids.remove(worker_id)
        # mark task queue as done when the first worker done to prevent cycling
        # with add-del workers
        self.done = True

    def mark_task_done(self, task_id):
        self.done_task_ids.add(task_id)

    def undone_tasks(self):
        # keeps an original order
        res = []
        for task_id in self.task_ids:
            if task_id not in self.done_task_ids:
                res.append(task_id)
        return res

1	import os	1✔
2	import signal	1✔
3	import time	1✔
4	import select	1✔
5	import random	1✔
6	import functools	1✔
7	import yaml	1✔
8
9	import multiprocessing	1✔
10
11	# Queue is available from multiprocessing.queues on all Python
12	# versions known at the moment of writting the code (up to 3.12).
13	#
14	# However the mandatory argument 'ctx'
15	# (see multiprocessing.get_context()) was added to the constructor
16	# of Queue from multiprocessing.queues since Python 3.4 ([1]).
17	#
18	# So we should import Queue from multiprocessing on Python 3.4+
19	# to uniformly instantiate it (without constructor arguments).
20	#
21	# [1]: https://bugs.python.org/issue18999
22	try:	1✔
23	# Python 3.4+
24	from multiprocessing import Queue	1✔
UNCOV 25	except ImportError:	×
26	# Python 2
NEW 27	from multiprocessing.queues import Queue	×
28
29	from lib import Options	1✔
30	from lib.sampler import sampler	1✔
31	from lib.utils import set_fd_cloexec	1✔
32	from lib.worker import WorkerTaskResult, WorkerDone	1✔
33	from lib.colorer import color_stdout	1✔
34	from listeners import ArtifactsWatcher	1✔
35	from listeners import FailWatcher	1✔
36	from listeners import HangWatcher	1✔
37	from listeners import LogOutputWatcher	1✔
38	from listeners import OutputWatcher	1✔
39	from listeners import StatisticsWatcher	1✔
40
41
42	class Dispatcher:	1✔
43	"""Run specified count of worker processes ('max_workers_cnt' arg), pass
44	task IDs (via 'task_queue'), receive results and output (via
45	'result_queue') and pass it to listeners. Workers as well as tasks have
46	types and certain task can be run only on worker of that type. To being
47	abstract we get 'task_groups' argument contains worker generators (the
48	callable working as factory of workers) and IDs of task that can be
49	executed on such workers. The structure of this argument is the following:
50	```
51	task_groups = {
52	'some_key_1': {
53	'gen_worker': function,
54	'task_ids': list,
55	'is_parallel': bool,
56	'show_reproduce_content': bool,
57	}
58	...
59	}
60
61	```
62	Usage (simplified and w/o exception catching):
63	```
64	task_groups = ...
65	dispatcher = Dispatcher(task_groups, max_workers_count=8, randomize=True)
66	dispatcher.start()
67	dispatcher.wait()
68	dispatcher.statistics.print_statistics()
69	dispatcher.wait_processes()
70	```
71	"""
72	def __init__(self, task_groups, max_workers_cnt, randomize):	1✔
73	self.pids = []	1✔
74	self.processes = []	1✔
75	self.result_queues = []	1✔
76	self.task_queues = []	1✔
77	self.workers_cnt = 0	1✔
78	self.worker_next_id = 1	1✔
79
80	tasks_cnt = 0	1✔
81	self.task_queue_disps = dict()	1✔
82	for key, task_group in task_groups.items():	1✔
83	tasks_cnt += len(task_group['task_ids'])	1✔
84	task_queue_disp = TaskQueueDispatcher(key, task_group, randomize)	1✔
85	self.task_queue_disps[key] = task_queue_disp	1✔
86	self.result_queues.append(task_queue_disp.result_queue)	1✔
87	self.task_queues.append(task_queue_disp.task_queue)	1✔
88	self.total_tasks_cnt = tasks_cnt	1✔
89
90	self.report_timeout = 0.1	1✔
91
92	self.statistics = None	1✔
93	self.artifacts = None	1✔
94	self.fail_watcher = None	1✔
95	self.listeners = None	1✔
96	self.init_listeners()	1✔
97
98	self.max_workers_cnt = min(max_workers_cnt, tasks_cnt)	1✔
99
100	self.pid_to_worker_id = dict()	1✔
101	self.worker_id_to_pid = dict()	1✔
102
103	self.randomize = randomize	1✔
104
105	def terminate_all_workers(self):	1✔
106	for process in self.processes:	×
107	if process.is_alive():	×
108	try:	×
109	process.terminate()	×
110	except OSError:	×
111	pass	×
112
113	def kill_all_workers(self):	1✔
114	for pid in self.pids:	×
115	try:	×
116	os.kill(pid, signal.SIGKILL)	×
117	except OSError:	×
118	pass	×
119
120	def init_listeners(self):	1✔
121	args = Options().args	1✔
122	watch_hang = args.no_output_timeout >= 0 and \	1✔
123	not args.gdb and \
124	not args.gdbserver and \
125	not args.lldb and \
126	not args.valgrind
127	watch_fail = not Options().args.is_force	1✔
128
129	log_output_watcher = LogOutputWatcher()	1✔
130	self.statistics = StatisticsWatcher(log_output_watcher.get_logfile,	1✔
131	self.total_tasks_cnt)
132	self.artifacts = ArtifactsWatcher(log_output_watcher.get_logfile)	1✔
133	output_watcher = OutputWatcher()	1✔
134	self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts]	1✔
135	if sampler.is_enabled:	1!
136	self.listeners.append(sampler.watcher)	1✔
137	if watch_fail:	1!
138	self.fail_watcher = FailWatcher(self.terminate_all_workers)	×
139	self.listeners.append(self.fail_watcher)	×
140	if watch_hang:	1!
141	warn_timeout = 60.0 if args.long else 10.0	1✔
142	hang_watcher = HangWatcher(output_watcher.not_done_worker_ids,	1✔
143	self.kill_all_workers, warn_timeout,
144	float(args.no_output_timeout))
145	self.listeners.append(hang_watcher)	1✔
146
147	def run_max_workers(self):	1✔
148	ok = True	1✔
149	new_workers_cnt = self.max_workers_cnt - self.workers_cnt	1✔
150	while ok and new_workers_cnt > 0:	1✔
151	ok = self.add_worker()	1✔
152	new_workers_cnt = self.max_workers_cnt - self.workers_cnt	1✔
153
154	def start(self):	1✔
155	self.run_max_workers()	1✔
156
157	def find_nonempty_task_queue_disp(self):	1✔
158	"""Find TaskQueueDispatcher that doesn't reported it's 'done' (don't
159	want more workers created for working on its task queue).
160	"""
161	task_queue_disps_rnd = list(	1✔
162	self.task_queue_disps.values())
163	if self.randomize:	1!
164	random.shuffle(task_queue_disps_rnd)	1✔
165	# run all parallel groups first
166	for task_queue_disp in task_queue_disps_rnd:	1✔
167	if not task_queue_disp.is_parallel:	1✔
168	continue	1✔
169	if task_queue_disp.done:	1✔
170	continue	1✔
171	return task_queue_disp	1✔
172	# then run all rest groups in a sequence
173	self.max_workers_cnt = 1	1✔
174	for task_queue_disp in task_queue_disps_rnd:	1✔
175	if len(task_queue_disp.worker_ids) > 0:	1!
176	continue	×
177	if task_queue_disp.done:	1✔
178	continue	1✔
179	return task_queue_disp	1✔
180	return None	1✔
181
182	def get_task_queue_disp(self, worker_id):	1✔
183	"""Get TaskQueueDispatcher instance which contains certain worker by
184	worker_id.
185	"""
186	for task_queue_disp in self.task_queue_disps.values():	1!
187	if worker_id in task_queue_disp.worker_ids:	1✔
188	return task_queue_disp	1✔
189	return None	×
190
191	def add_worker(self):	1✔
192	# don't add new workers if fail occured and --force not passed
193	if self.fail_watcher and self.fail_watcher.got_fail:	1!
194	return False	×
195	task_queue_disp = self.find_nonempty_task_queue_disp()	1✔
196	if not task_queue_disp:	1✔
197	return False	1✔
198	# self.max_workers_cnt can be changed in
199	# find_nonempty_task_queue_disp()
200	if self.workers_cnt >= self.max_workers_cnt:	1✔
201	return False	1✔
202	process = task_queue_disp.add_worker(self.worker_next_id)	1✔
203	self.processes.append(process)	1✔
204	self.pids.append(process.pid)	1✔
205	self.pid_to_worker_id[process.pid] = self.worker_next_id	1✔
206	self.worker_id_to_pid[self.worker_next_id] = process.pid	1✔
207
208	self.workers_cnt += 1	1✔
209	self.worker_next_id += 1	1✔
210
211	return True	1✔
212
213	def del_worker(self, worker_id):	1✔
214	pid = self.worker_id_to_pid[worker_id]	1✔
215
216	task_queue_disp = self.get_task_queue_disp(worker_id)	1✔
217	task_queue_disp.del_worker(worker_id)	1✔
218	self.workers_cnt -= 1	1✔
219
220	self.pids.remove(pid)	1✔
221	del self.worker_id_to_pid[worker_id]	1✔
222	del self.pid_to_worker_id[pid]	1✔
223	for process in self.processes:	1!
224	if process.pid == pid:	1✔
225	self.processes.remove(process)	1✔
226	break	1✔
227
228	def mark_task_done(self, worker_id, task_id):	1✔
229	task_queue_disp = self.get_task_queue_disp(worker_id)	1✔
230	task_queue_disp.mark_task_done(task_id)	1✔
231
232	def undone_tasks(self):	1✔
233	res = []	1✔
234	for task_queue_disp in self.task_queue_disps.values():	1✔
235	res.extend(task_queue_disp.undone_tasks())	1✔
236	return res	1✔
237
238	def report_undone(self, verbose):	1✔
239	undone = self.undone_tasks()	1✔
240	if not bool(undone):	1!
241	return False	1✔
242	if verbose:	×
243	color_stdout(	×
244	'[Internal test-run error] '
245	'The following tasks were dispatched to some worker task '
246	'queue, but were not reported as done (does not matters '
247	'success or fail):\n', schema='test_var')
248	for task_id in undone:	×
249	task_id_str = yaml.safe_dump(task_id, default_flow_style=True)	×
250	color_stdout('- %s' % task_id_str)	×
251	else:
252	# Visually continue StatisticsWatcher.print_statistics() output.
253	color_stdout('* undone: %d\n' % len(undone), schema='test_var')	×
254	return True	×
255
256	def wait(self):	1✔
257	"""Wait all workers reported its done via result_queues. But in the
258	case when some worker process terminated prematurely 'invoke_listeners'
259	can add fake WorkerDone markers (see also 'check_for_dead_processes').
260	"""
261	while self.workers_cnt > 0:	1✔
262	try:	1✔
263	inputs = [q._reader for q in self.result_queues]	1✔
264	ready_inputs, _, _ = select.select(	1✔
265	inputs, [], [], self.report_timeout)
266	except KeyboardInterrupt:	×
267	self.flush_ready(inputs)	×
268	raise	×
269
270	objs = self.invoke_listeners(inputs, ready_inputs)	1✔
271	for obj in objs:	1✔
272	if isinstance(obj, WorkerTaskResult):	1✔
273	self.mark_task_done(obj.worker_id, obj.task_id)	1✔
274	elif isinstance(obj, WorkerDone):	1✔
275	self.del_worker(obj.worker_id)	1✔
276	if not objs:	1✔
277	self.check_for_dead_processes()	1✔
278
279	self.run_max_workers()	1✔
280
281	def invoke_listeners(self, inputs, ready_inputs):	1✔
282	"""Returns received objects from result queue to allow Dispatcher
283	update its structures.
284	"""
285	# process timeout
286	if not ready_inputs:	1✔
287	for listener in self.listeners:	1✔
288	listener.process_timeout(self.report_timeout)	1✔
289	return []	1✔
290
291	# collect received objects
292	objs = []	1✔
293	for ready_input in ready_inputs:	1✔
294	result_queue = self.result_queues[inputs.index(ready_input)]	1✔
295	while not result_queue.empty():	1✔
296	objs.append(result_queue.get())	1✔
297
298	# process received objects
299	for obj in objs:	1✔
300	for listener in self.listeners:	1✔
301	listener.process_result(obj)	1✔
302
303	return objs	1✔
304
305	def flush_ready(self, inputs):	1✔
306	"""Write output from workers to stdout."""
307	# leave only output listeners in self.listeners
308	new_listeners = []	×
309	for listener in self.listeners:	×
310	if isinstance(listener, (LogOutputWatcher,	×
311	OutputWatcher)):
312	listener.report_at_timeout = False	×
313	new_listeners.append(listener)	×
314	self.listeners = new_listeners	×
315	# wait some time until processes in our group get its SIGINTs and give
316	# us some last output
317	time.sleep(0.1)	×
318	# collect and process ready inputs
319	ready_inputs, _, _ = select.select(inputs, [], [], 0)	×
320	self.invoke_listeners(inputs, ready_inputs)	×
321
322	def check_for_dead_processes(self):	1✔
323	for pid in self.pids:	1✔
324	exited = False	1✔
325	try:	1✔
326	os.waitpid(pid, os.WNOHANG)	1✔
327	except OSError:	×
328	exited = True	×
329	if exited:	1!
330	worker_id = self.pid_to_worker_id[pid]	×
331	color_stdout(	×
332	"[Main process] Worker %d don't reported work "
333	"done using results queue, but the corresponding "
334	"process seems dead. Removing it from Dispatcher.\n"
335	% worker_id, schema='test_var')
336	self.del_worker(worker_id)	×
337
338	def wait_processes(self):	1✔
339	for process in self.processes:	1!
340	process.join()	×
341	self.processes = []	1✔
342
343
344	class TaskQueueDispatcher:	1✔
345	"""Incapsulate data structures necessary for dispatching workers working on
346	the one task queue.
347	"""
348	def __init__(self, key, task_group, randomize):	1✔
349	self.key = key	1✔
350	self.gen_worker = task_group['gen_worker']	1✔
351	self.task_ids = task_group['task_ids']	1✔
352	self.is_parallel = task_group['is_parallel']	1✔
353	if self.is_parallel:	1✔
354	self.randomize = randomize	1✔
355	if self.randomize:	1!
356	random.shuffle(self.task_ids)	1✔
357	else:
358	self.randomize = False	1✔
359	self.result_queue = Queue()	1✔
360	self.task_queue = Queue()	1✔
361
362	# Don't expose queues file descriptors over Popen to, say, tarantool
363	# running tests.
364	set_fd_cloexec(self.result_queue._reader.fileno())	1✔
365	set_fd_cloexec(self.result_queue._writer.fileno())	1✔
366	set_fd_cloexec(self.task_queue._reader.fileno())	1✔
367	set_fd_cloexec(self.task_queue._writer.fileno())	1✔
368
369	for task_id in self.task_ids:	1✔
370	self.task_queue.put(task_id)	1✔
371	self.worker_ids = set()	1✔
372	self.done = False	1✔
373	self.done_task_ids = set()	1✔
374
375	def _run_worker(self, worker_id):	1✔
376	"""Entry function for worker processes."""
377	os.environ['TEST_RUN_WORKER_ID'] = str(worker_id)	1✔
378	color_stdout.queue = self.result_queue	1✔
379	worker = self.gen_worker(worker_id)	1✔
380	sampler.set_queue(self.result_queue, worker_id, worker.name)	1✔
381	worker.run_all(self.task_queue, self.result_queue)	1✔
382
383	def add_worker(self, worker_id):	1✔
384	# Note: each of our workers should consume only one None, but for the
385	# case of abnormal circumstances we listen for processes termination
386	# (method 'check_for_dead_processes') and for time w/o output from
387	# workers (class 'HangWatcher').
388	self.task_queue.put(None) # 'stop worker' marker	1✔
389
390	entry = functools.partial(self._run_worker, worker_id)	1✔
391
392	self.worker_ids.add(worker_id)	1✔
393	process = multiprocessing.Process(target=entry)	1✔
394	process.start()	1✔
395	return process	1✔
396
397	def del_worker(self, worker_id):	1✔
398	self.worker_ids.remove(worker_id)	1✔
399	# mark task queue as done when the first worker done to prevent cycling
400	# with add-del workers
401	self.done = True	1✔
402
403	def mark_task_done(self, task_id):	1✔
404	self.done_task_ids.add(task_id)	1✔
405
406	def undone_tasks(self):	1✔
407	# keeps an original order
408	res = []	1✔
409	for task_id in self.task_ids:	1✔
410	if task_id not in self.done_task_ids:	1!
411	res.append(task_id)	×
412	return res	1✔

tarantool / test-run / 9285268044

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous