#3431

pending completion

Build # #3431

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

<a href="https://github.com/mosquito/aiomisc/commit/<a class=hub.com/mosquito/aiomisc/commit/<a class="double-link" href="https://git"><a class=hub.com/mosquito/aiomisc/commit/<a class="double-link" href="https://git"><a class=hub.com/mosquito/aiomisc/commit/<a class="double-link" href="https://git"><a class=hub.com/mosquito/aiomisc/commit/<a class="double-link" href="https://git"><a class=hub.com/mosquito/aiomisc/commit/a3ac00e698b3f8917c4f2f101c47865c94b346ea">a3ac00e69">&lt;a href=&quot;https://github.com/mosquito/aiomisc/commit/</a><a class="double-link" href="https://github.com/mosquito/aiomisc/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/mosquito/aiomisc/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/mosquito/aiomisc/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/mosquito/aiomisc/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/mosquito/aiomisc/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/mosquito/aiomisc/commit/a3ac00e698b3f8917c4f2f101c47865c94b346ea">a3ac00e69</a><a href="https://github.com/mosquito/aiomisc/commit/a3ac00e698b3f8917c4f2f101c47865c94b346ea">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/mosquito/aiomisc/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/mosquito/aiomisc/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/mosquito/aiomisc/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/mosquito/aiomisc/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/mosquito/aiomisc/commit/a3ac00e698b3f8917c4f2f101c47865c94b346ea&quot;&gt;a3ac00e69&lt;/a&gt;&lt;a href=&quot;https://github.com/mosquito/aiomisc/commit/a3ac00e698b3f8917c4f2f101c47865c94b346ea&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/mosquito/aiomisc/commit/a3ac00e698b3f8917c4f2f101c47865c94b346ea&amp;quot;&amp;gt;&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;lt;a href=&amp;amp;amp;quot;https://github.com/mosquito/aiomisc/commit/&amp;amp;lt;/a&amp;amp;gt;&amp;amp;lt;a class=&... (continued)

Run Details

453 of 636 branches covered (71.23%)

Branch coverage included in aggregate %.

59 of 59 new or added lines in 8 files covered. (100.0%)

1706 of 1984 relevant lines covered (85.99%)

0.86 hits per line

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

80.87

/aiomisc/worker_pool.py

import asyncio
import os
import pickle
import socket
import sys
import uuid
import warnings
from inspect import Traceback
from itertools import chain
from multiprocessing import AuthenticationError, ProcessError
from os import chmod, urandom
from subprocess import PIPE, Popen
from tempfile import mktemp
from types import MappingProxyType
from typing import (
    Any, Callable, Coroutine, Dict, Mapping, Optional, Set, Tuple, Type,
)

from aiomisc.counters import Statistic
from aiomisc.thread_pool import threaded
from aiomisc.utils import bind_socket, cancel_tasks
from aiomisc_log import LOG_FORMAT, LOG_LEVEL
from aiomisc_worker import (
    COOKIE_SIZE, HASHER, INET_AF, SIGNAL, AddressType, Header, PacketTypes, T,
    log,
)


if sys.version_info < (3, 7):
    warnings.warn(
        "Python 3.6 works not well see https://bugs.python.org/issue37380",
        RuntimeWarning,
    )


class WorkerPoolStatistic(Statistic):
    processes: int
    queue_size: int
    submitted: int
    sum_time: float
    done: int
    success: int
    error: int
    bad_auth: int
    task_added: int


class WorkerPool:
    tasks: asyncio.Queue
    server: asyncio.AbstractServer
    address: AddressType
    initializer: Optional[Callable[[], Any]]
    initializer_args: Tuple[Any, ...]
    initializer_kwargs: Mapping[str, Any]

    if hasattr(socket, "AF_UNIX"):
        def _create_socket(self) -> None:
            path = mktemp(suffix=".sock", prefix="worker-")
            self.socket = bind_socket(
                socket.AF_UNIX,
                socket.SOCK_STREAM,
                address=path,
            )
            self.address = path
            chmod(path, 0o600)
    else:
        def _create_socket(self) -> None:
            self.socket = bind_socket(
                INET_AF,
                socket.SOCK_STREAM,
                address="localhost",
            )
            self.address = self.socket.getsockname()[:2]

    @staticmethod
    def _kill_process(process: Popen) -> None:
        if process.returncode is not None:
            return None
        log.debug("Terminating worker pool process PID: %s", process.pid)
        process.kill()

    @threaded
    def __create_process(self, identity: str) -> Popen:
        if self.__closing:
            raise RuntimeError("Pool closed")

        env = dict(os.environ)
        env["AIOMISC_NO_PLUGINS"] = ""
        process = Popen(
            [sys.executable, "-m", "aiomisc_worker"],
            stdin=PIPE, env=env,
        )
        self.__spawning[identity] = process
        log.debug("Spawning new worker pool process PID: %s", process.pid)

        assert process.stdin

        log_level = (
            log.getEffectiveLevel() if LOG_LEVEL is None else LOG_LEVEL.get()
        )
        log_format = "color" if LOG_FORMAT is None else LOG_FORMAT.get().value

        process.stdin.write(
            pickle.dumps((
                self.address, self.__cookie, identity,
                log_level, log_format,
            )),
        )
        process.stdin.close()

        return process

    def __init__(
        self, workers: int, max_overflow: int = 0, *,
        process_poll_time: float = 0.1,
        initializer: Optional[Callable[[], Any]] = None,
        initializer_args: Tuple[Any, ...] = (),
        initializer_kwargs: Mapping[str, Any] = MappingProxyType({}),
    ):
        self._create_socket()
        self.__cookie = urandom(COOKIE_SIZE)
        self.__loop: Optional[asyncio.AbstractEventLoop] = None
        self.__futures: Set[asyncio.Future] = set()
        self.__spawning: Dict[str, Popen] = dict()
        self.__task_store: Set[asyncio.Task] = set()
        self.__closing = False
        self.__starting: Dict[str, asyncio.Future] = dict()
        self._statistic = WorkerPoolStatistic()
        self.processes: Set[Popen] = set()
        self.workers = workers
        self.tasks = asyncio.Queue(maxsize=max_overflow)
        self.process_poll_time = process_poll_time
        self.initializer = initializer
        self.initializer_args = initializer_args
        self.initializer_kwargs = initializer_kwargs

    async def __wait_process(self, process: Popen) -> None:
        while process.poll() is None:
            await asyncio.sleep(self.process_poll_time)

    @property
    def loop(self) -> asyncio.AbstractEventLoop:
        if self.__loop is None:
            self.__loop = asyncio.get_event_loop()
        return self.__loop

    @staticmethod
    async def __wait_closed(writer: asyncio.StreamWriter) -> None:
        if writer.is_closing():
            return

        writer.close()

        if hasattr(writer, "wait_closed"):
            await writer.wait_closed()

    async def __handle_client(
        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
    ) -> None:
        async def receive() -> Tuple[PacketTypes, Any]:
            header = await reader.readexactly(Header.size)
            packet_type, payload_length = Header.unpack(header)
            payload = await reader.readexactly(payload_length)
            data = pickle.loads(payload)
            return PacketTypes(packet_type), data

        async def send(packet_type: PacketTypes, data: Any) -> None:
            payload = pickle.dumps(data)
            header = Header.pack(packet_type.value, len(payload))
            writer.write(header)
            writer.write(payload)

        async def step(
            func: Callable, args: Tuple[Any, ...],
            kwargs: Dict[str, Any], result_future: asyncio.Future,
        ) -> None:
            await send(PacketTypes.REQUEST, (func, args, kwargs))
            self._statistic.submitted += 1

            packet_type, result = await receive()
            self._statistic.done += 1

            if packet_type == PacketTypes.RESULT:
                result_future.set_result(result)
                self._statistic.success += 1
                return None

            if packet_type == PacketTypes.EXCEPTION:
                result_future.set_exception(result)
                self._statistic.error += 1
                return None

            if packet_type == PacketTypes.CANCELLED:
                if not result_future.done():
                    result_future.set_exception(asyncio.CancelledError)
                return None

            raise ValueError("Unknown packet type")

        async def handler(start_event: asyncio.Event) -> None:
            log.debug("Starting to handle client")

            packet_type, salt = await receive()
            assert packet_type == PacketTypes.AUTH_SALT

            packet_type, digest = await receive()
            assert packet_type == PacketTypes.AUTH_DIGEST

            hasher = HASHER()
            hasher.update(salt)
            hasher.update(self.__cookie)

            if digest != hasher.digest():
                exc = AuthenticationError("Invalid cookie")
                self._statistic.bad_auth += 1
                await send(PacketTypes.EXCEPTION, exc)
                raise exc

            await send(PacketTypes.AUTH_OK, True)

            log.debug("Client authorized")

            packet_type, identity = await receive()
            assert packet_type == PacketTypes.IDENTITY
            process = self.__spawning.pop(identity)
            starting: asyncio.Future = self.__starting.pop(identity)

            if self.initializer is not None:
                initializer_done = self.__create_future()

                await step(
                    self.initializer,
                    self.initializer_args,
                    dict(self.initializer_kwargs),
                    initializer_done,
                )

                try:
                    await initializer_done
                except Exception as e:
                    starting.set_exception(e)
                    raise
                else:
                    starting.set_result(None)
                finally:
                    start_event.set()
            else:
                starting.set_result(None)
                start_event.set()

            try:
                while True:
                    func: Callable
                    args: Tuple[Any, ...]
                    kwargs: Dict[str, Any]
                    result_future: asyncio.Future
                    process_future: asyncio.Future

                    self._statistic.processes += 1
                    (
                        func, args, kwargs, result_future, process_future,
                    ) = await self.tasks.get()

                    try:
                        if process_future.done():
                            continue

                        process_future.set_result(process)

                        if result_future.done():
                            continue

                        await step(func, args, kwargs, result_future)
                    except asyncio.IncompleteReadError:
                        await self.__wait_process(process)
                        self.__on_exit(process)

                        result_future.set_exception(
                            ProcessError(
                                "Process {!r} exited with code {!r}".format(
                                    process, process.returncode,
                                ),
                            ),
                        )
                        break
                    except Exception as e:
                        if not result_future.done():
                            self.loop.call_soon(result_future.set_exception, e)

                        if not writer.is_closing():
                            self.loop.call_soon(writer.close)

                        await self.__wait_process(process)
                        self.__on_exit(process)

                        raise
                    finally:
                        self._statistic.processes -= 1
            finally:
                await self.__wait_closed(writer)

        start_event = asyncio.Event()
        task = self.loop.create_task(handler(start_event))
        await start_event.wait()
        self.__task_add(task)

        await task

    def __task_add(self, task: asyncio.Task) -> None:
        self._statistic.task_added += 1
        task.add_done_callback(self.__task_store.remove)
        self.__task_store.add(task)

    def __task(self, coroutine: Coroutine) -> asyncio.Task:
        task = self.loop.create_task(coroutine)
        self.__task_add(task)
        return task

    async def start(self) -> None:
        self.server = await asyncio.start_server(
            self.__handle_client,
            sock=self.socket,
        )

        tasks = []

        for n in range(self.workers):
            log.debug("Starting worker %d", n)
            tasks.append(self.__spawn_process())

        await asyncio.gather(*tasks)

    def __on_exit(self, process: Popen) -> None:
        async def respawn() -> None:
            if self.__closing:
                return None
            self.processes.remove(process)
            await self.__spawn_process()

        self.__task(respawn())

    async def __spawn_process(self) -> None:
        log.debug("Spawning new process")

        identity = uuid.uuid4().hex
        start_future = self.__create_future()
        self.__starting[identity] = start_future
        process = await self.__create_process(identity)
        await start_future
        self.processes.add(process)

    def __create_future(self) -> asyncio.Future:
        future = self.loop.create_future()
        self.__futures.add(future)
        future.add_done_callback(self.__futures.remove)
        return future

    def __reject_futures(self) -> None:
        for future in self.__futures:
            if future.done():
                continue
            future.set_exception(RuntimeError("Pool closed"))

    async def close(self) -> None:
        if self.__closing:
            return

        self.__closing = True

        await cancel_tasks(
            chain(tuple(self.__task_store), tuple(self.__futures)),
        )

        @threaded
        def killer() -> None:
            while self.processes:
                self._kill_process(self.processes.pop())

        await killer()

    async def create_task(
        self, func: Callable[..., T],
        *args: Any, **kwargs: Any
    ) -> T:
        result_future = self.__create_future()
        process_future = self.__create_future()

        await self.tasks.put((
            func, args, kwargs, result_future, process_future,
        ))

        process: Popen = await process_future

        try:
            return await result_future
        except asyncio.CancelledError:
            os.kill(process.pid, SIGNAL)
            raise

    async def __aenter__(self) -> "WorkerPool":
        await self.start()
        return self

    async def __aexit__(
        self, exc_type: Type[Exception],
        exc_val: Exception, exc_tb: Traceback,
    ) -> None:
        await self.close()

1	import asyncio	1✔
2	import os	1✔
3	import pickle	1✔
4	import socket	1✔
5	import sys	1✔
6	import uuid	1✔
7	import warnings	1✔
8	from inspect import Traceback	1✔
9	from itertools import chain	1✔
10	from multiprocessing import AuthenticationError, ProcessError	1✔
11	from os import chmod, urandom	1✔
12	from subprocess import PIPE, Popen	1✔
13	from tempfile import mktemp	1✔
14	from types import MappingProxyType	1✔
15	from typing import (	1✔
16	Any, Callable, Coroutine, Dict, Mapping, Optional, Set, Tuple, Type,
17	)
18
19	from aiomisc.counters import Statistic	1✔
20	from aiomisc.thread_pool import threaded	1✔
21	from aiomisc.utils import bind_socket, cancel_tasks	1✔
22	from aiomisc_log import LOG_FORMAT, LOG_LEVEL	1✔
23	from aiomisc_worker import (	1✔
24	COOKIE_SIZE, HASHER, INET_AF, SIGNAL, AddressType, Header, PacketTypes, T,
25	log,
26	)
27
28
29	if sys.version_info < (3, 7):	1!
30	warnings.warn(	×
31	"Python 3.6 works not well see https://bugs.python.org/issue37380",
32	RuntimeWarning,
33	)
34
35
36	class WorkerPoolStatistic(Statistic):	1✔
37	processes: int	1✔
38	queue_size: int	1✔
39	submitted: int	1✔
40	sum_time: float	1✔
41	done: int	1✔
42	success: int	1✔
43	error: int	1✔
44	bad_auth: int	1✔
45	task_added: int	1✔
46
47
48	class WorkerPool:	1✔
49	tasks: asyncio.Queue	1✔
50	server: asyncio.AbstractServer	1✔
51	address: AddressType	1✔
52	initializer: Optional[Callable[[], Any]]	1✔
53	initializer_args: Tuple[Any, ...]	1✔
54	initializer_kwargs: Mapping[str, Any]	1✔
55
56	if hasattr(socket, "AF_UNIX"):	1!
57	def _create_socket(self) -> None:	1✔
58	path = mktemp(suffix=".sock", prefix="worker-")	1✔
59	self.socket = bind_socket(	1✔
60	socket.AF_UNIX,
61	socket.SOCK_STREAM,
62	address=path,
63	)
64	self.address = path
65	chmod(path, 0o600)
66	else:
67	def _create_socket(self) -> None:	×
68	self.socket = bind_socket(	×
69	INET_AF,
70	socket.SOCK_STREAM,
71	address="localhost",
72	)
73	self.address = self.socket.getsockname()[:2]
74
75	@staticmethod	1✔
76	def _kill_process(process: Popen) -> None:	1!
77	if process.returncode is not None:	1!
78	return None	×
79	log.debug("Terminating worker pool process PID: %s", process.pid)	1✔
80	process.kill()	1✔
81
82	@threaded	1✔
83	def __create_process(self, identity: str) -> Popen:	1!
84	if self.__closing:	1!
85	raise RuntimeError("Pool closed")	×
86
87	env = dict(os.environ)	1✔
88	env["AIOMISC_NO_PLUGINS"] = ""	1✔
89	process = Popen(	1✔
90	[sys.executable, "-m", "aiomisc_worker"],
91	stdin=PIPE, env=env,
92	)
93	self.__spawning[identity] = process
94	log.debug("Spawning new worker pool process PID: %s", process.pid)
95
96	assert process.stdin
97
98	log_level = (
99	log.getEffectiveLevel() if LOG_LEVEL is None else LOG_LEVEL.get()
100	)
101	log_format = "color" if LOG_FORMAT is None else LOG_FORMAT.get().value
102
103	process.stdin.write(
104	pickle.dumps((
105	self.address, self.__cookie, identity,
106	log_level, log_format,
107	)),
108	)
109	process.stdin.close()
110
111	return process
112
113	def __init__(
114	self, workers: int, max_overflow: int = 0, *,
115	process_poll_time: float = 0.1,
116	initializer: Optional[Callable[[], Any]] = None,
117	initializer_args: Tuple[Any, ...] = (),
118	initializer_kwargs: Mapping[str, Any] = MappingProxyType({}),
119	):
120	self._create_socket()	1✔
121	self.__cookie = urandom(COOKIE_SIZE)	1✔
122	self.__loop: Optional[asyncio.AbstractEventLoop] = None	1✔
123	self.__futures: Set[asyncio.Future] = set()	1✔
124	self.__spawning: Dict[str, Popen] = dict()	1✔
125	self.__task_store: Set[asyncio.Task] = set()	1✔
126	self.__closing = False	1✔
127	self.__starting: Dict[str, asyncio.Future] = dict()	1✔
128	self._statistic = WorkerPoolStatistic()	1✔
129	self.processes: Set[Popen] = set()	1✔
130	self.workers = workers	1✔
131	self.tasks = asyncio.Queue(maxsize=max_overflow)	1✔
132	self.process_poll_time = process_poll_time	1✔
133	self.initializer = initializer	1✔
134	self.initializer_args = initializer_args	1✔
135	self.initializer_kwargs = initializer_kwargs	1✔
136
137	async def __wait_process(self, process: Popen) -> None:
138	while process.poll() is None:	1✔
139	await asyncio.sleep(self.process_poll_time)	1✔
140
141	@property
142	def loop(self) -> asyncio.AbstractEventLoop:
143	if self.__loop is None:	1✔
144	self.__loop = asyncio.get_event_loop()	1✔
145	return self.__loop	1✔
146
147	@staticmethod
148	async def __wait_closed(writer: asyncio.StreamWriter) -> None:
149	if writer.is_closing():	1!
150	return	×
151
152	writer.close()	1✔
153
154	if hasattr(writer, "wait_closed"):	1!
155	await writer.wait_closed()	1✔
156
157	async def __handle_client(
158	self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
159	) -> None:
160	async def receive() -> Tuple[PacketTypes, Any]:	1✔
161	header = await reader.readexactly(Header.size)	1✔
162	packet_type, payload_length = Header.unpack(header)	1✔
163	payload = await reader.readexactly(payload_length)	1✔
164	data = pickle.loads(payload)	1✔
165	return PacketTypes(packet_type), data	1✔
166
167	async def send(packet_type: PacketTypes, data: Any) -> None:	1✔
168	payload = pickle.dumps(data)	1✔
169	header = Header.pack(packet_type.value, len(payload))	1✔
170	writer.write(header)	1✔
171	writer.write(payload)	1✔
172
173	async def step(	1✔
174	func: Callable, args: Tuple[Any, ...],
175	kwargs: Dict[str, Any], result_future: asyncio.Future,
176	) -> None:
177	await send(PacketTypes.REQUEST, (func, args, kwargs))	1✔
178	self._statistic.submitted += 1	1✔
179
180	packet_type, result = await receive()	1✔
181	self._statistic.done += 1	1✔
182
183	if packet_type == PacketTypes.RESULT:	1✔
184	result_future.set_result(result)	1✔
185	self._statistic.success += 1	1✔
186	return None	1✔
187
188	if packet_type == PacketTypes.EXCEPTION:	1✔
189	result_future.set_exception(result)	1✔
190	self._statistic.error += 1	1✔
191	return None	1✔
192
193	if packet_type == PacketTypes.CANCELLED:	1!
194	if not result_future.done():	1!
195	result_future.set_exception(asyncio.CancelledError)	×
196	return None	1✔
197
198	raise ValueError("Unknown packet type")	×
199
200	async def handler(start_event: asyncio.Event) -> None:
201	log.debug("Starting to handle client")	1✔
202
203	packet_type, salt = await receive()	1✔
204	assert packet_type == PacketTypes.AUTH_SALT	1✔
205
206	packet_type, digest = await receive()	1✔
207	assert packet_type == PacketTypes.AUTH_DIGEST	1✔
208
209	hasher = HASHER()	1✔
210	hasher.update(salt)	1✔
211	hasher.update(self.__cookie)	1✔
212
213	if digest != hasher.digest():	1!
214	exc = AuthenticationError("Invalid cookie")	×
215	self._statistic.bad_auth += 1	×
216	await send(PacketTypes.EXCEPTION, exc)	×
217	raise exc	×
218
219	await send(PacketTypes.AUTH_OK, True)	1✔
220
221	log.debug("Client authorized")	1✔
222
223	packet_type, identity = await receive()	1✔
224	assert packet_type == PacketTypes.IDENTITY	1✔
225	process = self.__spawning.pop(identity)	1✔
226	starting: asyncio.Future = self.__starting.pop(identity)	1✔
227
228	if self.initializer is not None:	1!
229	initializer_done = self.__create_future()	1✔
230
231	await step(	1✔
232	self.initializer,
233	self.initializer_args,
234	dict(self.initializer_kwargs),
235	initializer_done,
236	)
237
238	try:
239	await initializer_done
240	except Exception as e:
241	starting.set_exception(e)
242	raise
243	else:
244	starting.set_result(None)
245	finally:
246	start_event.set()
247	else:
248	starting.set_result(None)
249	start_event.set()
250
251	try:
252	while True:
253	func: Callable
254	args: Tuple[Any, ...]
255	kwargs: Dict[str, Any]
256	result_future: asyncio.Future
257	process_future: asyncio.Future
258
259	self._statistic.processes += 1
260	(
261	func, args, kwargs, result_future, process_future,
262	) = await self.tasks.get()
263
264	try:
265	if process_future.done():
266	continue
267
268	process_future.set_result(process)
269
270	if result_future.done():
271	continue
272
273	await step(func, args, kwargs, result_future)
274	except asyncio.IncompleteReadError:
275	await self.__wait_process(process)
276	self.__on_exit(process)
277
278	result_future.set_exception(
279	ProcessError(
280	"Process {!r} exited with code {!r}".format(
281	process, process.returncode,
282	),
283	),
284	)
285	break
286	except Exception as e:	×
287	if not result_future.done():
288	self.loop.call_soon(result_future.set_exception, e)
289
290	if not writer.is_closing():
291	self.loop.call_soon(writer.close)
292		×
293	await self.__wait_process(process)
294	self.__on_exit(process)
295
296	raise
297	finally:
298	self._statistic.processes -= 1
299	finally:
300	await self.__wait_closed(writer)
301
302	start_event = asyncio.Event()
303	task = self.loop.create_task(handler(start_event))
304	await start_event.wait()	1✔
305	self.__task_add(task)
306
307	await task
308
309	def __task_add(self, task: asyncio.Task) -> None:
310	self._statistic.task_added += 1	1✔
311	task.add_done_callback(self.__task_store.remove)	1✔
312	self.__task_store.add(task)	1✔
313
314	def __task(self, coroutine: Coroutine) -> asyncio.Task:
315	task = self.loop.create_task(coroutine)	1✔
316	self.__task_add(task)	1✔
317	return task	1✔
318
319	async def start(self) -> None:
320	self.server = await asyncio.start_server(	1✔
321	self.__handle_client,
322	sock=self.socket,
323	)
324		×
325	tasks = []
326
327	for n in range(self.workers):
328	log.debug("Starting worker %d", n)
329	tasks.append(self.__spawn_process())	1✔
330
331	await asyncio.gather(*tasks)
332
333	def __on_exit(self, process: Popen) -> None:
334	async def respawn() -> None:	1✔
335	if self.__closing:	1!
336	return None	×
337	self.processes.remove(process)	1✔
338	await self.__spawn_process()	1✔
339
340	self.__task(respawn())	1✔
341
342	async def __spawn_process(self) -> None:
343	log.debug("Spawning new process")	1✔
344
345	identity = uuid.uuid4().hex	1✔
346	start_future = self.__create_future()	1✔
347	self.__starting[identity] = start_future	1✔
348	process = await self.__create_process(identity)	1✔
349	await start_future	1✔
350	self.processes.add(process)	1✔
351
352	def __create_future(self) -> asyncio.Future:	1✔
353	future = self.loop.create_future()	1✔
354	self.__futures.add(future)	1✔
355	future.add_done_callback(self.__futures.remove)	1✔
356	return future	1✔
357
358	def __reject_futures(self) -> None:
359	for future in self.__futures:	×
360	if future.done():	×
361	continue	×
362	future.set_exception(RuntimeError("Pool closed"))	×
363
364	async def close(self) -> None:
365	if self.__closing:	1!
366	return	×
367
368	self.__closing = True	1✔
369		×
370	await cancel_tasks(	1✔
371	chain(tuple(self.__task_store), tuple(self.__futures)),
372	)
373		×
374	@threaded	1✔
375	def killer() -> None:
376	while self.processes:	1!
377	self._kill_process(self.processes.pop())	1✔
378
379	await killer()
380
381	async def create_task(
382	self, func: Callable[..., T],
383	args: Any, *kwargs: Any
384	) -> T:
385	result_future = self.__create_future()	1✔
386	process_future = self.__create_future()	1✔
387
388	await self.tasks.put((	1✔
389	func, args, kwargs, result_future, process_future,
390	))
391
392	process: Popen = await process_future
393
394	try:
395	return await result_future
396	except asyncio.CancelledError:
397	os.kill(process.pid, SIGNAL)
398	raise
399
400	async def __aenter__(self) -> "WorkerPool":
401	await self.start()	1✔
402	return self	1✔
403
404	async def __aexit__(
405	self, exc_type: Type[Exception],
406	exc_val: Exception, exc_tb: Traceback,
407	) -> None:
408	await self.close()	1✔

mosquito / aiomisc / #3431

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