17340928350

Committed 30 Aug 2025 07:15AM UTC coverage: 77.889% (+2.4%) from 75.467%

Build # 17340928350

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #32 from karellen/geventmp_forkserver

Instrument Multiprocessing forkserver

Coverage Stats

38 of 71 branches covered (53.52%)

Branch coverage included in aggregate %.

44 of 69 new or added lines in 5 files covered. (63.77%)

21 existing lines in 2 files now uncovered.

272 of 327 relevant lines covered (83.18%)

12.87 hits per line

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

49.25

/src/main/python/geventmp/_mp/3/_mp_forkserver.py

#   -*- coding: utf-8 -*-
#   Copyright 2019 Karellen, Inc. and contributors
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

import os
import signal
from multiprocessing.forkserver import ForkServer as _ForkServer, SIGNED_STRUCT

from gevent.os import make_nonblocking, nb_read, nb_write

__implements__ = ["ForkServer", "_forkserver", "ensure_running", "_serve_one",
                  "get_inherited_fds", "connect_to_new_process",
                  "set_forkserver_preload", "read_signed", "write_signed"]
__target__ = "multiprocessing.forkserver"


class ForkServer(_ForkServer):
    def connect_to_new_process(self, fds):
        parent_r, parent_w = super().connect_to_new_process(fds)
        make_nonblocking(parent_r)
        make_nonblocking(parent_w)
        return parent_r, parent_w

    def ensure_running(self):
        super().ensure_running()
        make_nonblocking(self._forkserver_alive_fd)


def read_signed(fd):
    data = b''
    length = SIGNED_STRUCT.size
    while len(data) < length:
        s = nb_read(fd, length - len(data))
        if not s:
            raise EOFError('unexpected EOF')
        data += s
    return SIGNED_STRUCT.unpack(data)[0]


def write_signed(fd, n):
    msg = SIGNED_STRUCT.pack(n)
    while msg:
        nbytes = nb_write(fd, msg)
        if nbytes == 0:
            raise RuntimeError('should not get here')
        msg = msg[nbytes:]


def _serve_one(child_r, fds, unused_fds, handlers):
    from multiprocessing import resource_tracker, spawn
    # close unnecessary stuff and reset signal handlers
    signal.set_wakeup_fd(-1)
    for sig, val in handlers.items():
        signal.signal(sig, val)
    for fd in unused_fds:
        os.close(fd)

    (_forkserver._forkserver_alive_fd,
     resource_tracker._resource_tracker._fd,
     *_forkserver._inherited_fds) = fds
    make_nonblocking(_forkserver._forkserver_alive_fd)
    make_nonblocking(resource_tracker._resource_tracker._fd)
    for ifd in _forkserver._inherited_fds:
        make_nonblocking(ifd)

    # Run process object received over pipe
    parent_sentinel = os.dup(child_r)
    make_nonblocking(child_r)
    make_nonblocking(parent_sentinel)
    code = spawn._main(child_r, parent_sentinel)

    return code


_forkserver = ForkServer()
ensure_running = _forkserver.ensure_running
get_inherited_fds = _forkserver.get_inherited_fds
connect_to_new_process = _forkserver.connect_to_new_process
set_forkserver_preload = _forkserver.set_forkserver_preload

1	# -- coding: utf-8 --
2	# Copyright 2019 Karellen, Inc. and contributors
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15
16	import os	18✔
17	import signal	18✔
18	from multiprocessing.forkserver import ForkServer as _ForkServer, SIGNED_STRUCT	18✔
19
20	from gevent.os import make_nonblocking, nb_read, nb_write	18✔
21
22	__implements__ = ["ForkServer", "_forkserver", "ensure_running", "_serve_one",	18✔
23	"get_inherited_fds", "connect_to_new_process",
24	"set_forkserver_preload", "read_signed", "write_signed"]
25	__target__ = "multiprocessing.forkserver"	18✔
26
27
28	class ForkServer(_ForkServer):	18✔
29	def connect_to_new_process(self, fds):	18✔
30	parent_r, parent_w = super().connect_to_new_process(fds)	18✔
31	make_nonblocking(parent_r)	18✔
32	make_nonblocking(parent_w)	16✔
33	return parent_r, parent_w	16✔
34
35	def ensure_running(self):	18✔
36	super().ensure_running()	18✔
37	make_nonblocking(self._forkserver_alive_fd)	18✔
38
39
40	def read_signed(fd):	18✔
41	data = b''	18✔
42	length = SIGNED_STRUCT.size	18✔
43	while len(data) < length:	18✔
44	s = nb_read(fd, length - len(data))	18✔
45	if not s:	18!
NEW 46	raise EOFError('unexpected EOF')	×
47	data += s	18✔
48	return SIGNED_STRUCT.unpack(data)[0]	18✔
49
50
51	def write_signed(fd, n):	18✔
NEW 52	msg = SIGNED_STRUCT.pack(n)	×
NEW 53	while msg:	×
NEW 54	nbytes = nb_write(fd, msg)	×
NEW 55	if nbytes == 0:	×
NEW 56	raise RuntimeError('should not get here')	×
NEW 57	msg = msg[nbytes:]	×
58
59
60	def _serve_one(child_r, fds, unused_fds, handlers):	18✔
NEW 61	from multiprocessing import resource_tracker, spawn	×
62	# close unnecessary stuff and reset signal handlers
NEW 63	signal.set_wakeup_fd(-1)	×
NEW 64	for sig, val in handlers.items():	×
NEW 65	signal.signal(sig, val)	×
NEW 66	for fd in unused_fds:	×
NEW 67	os.close(fd)	×
68
NEW 69	(_forkserver._forkserver_alive_fd,	×
70	resource_tracker._resource_tracker._fd,
71	*_forkserver._inherited_fds) = fds
NEW 72	make_nonblocking(_forkserver._forkserver_alive_fd)	×
NEW 73	make_nonblocking(resource_tracker._resource_tracker._fd)	×
NEW 74	for ifd in _forkserver._inherited_fds:	×
NEW 75	make_nonblocking(ifd)	×
76
77	# Run process object received over pipe
NEW 78	parent_sentinel = os.dup(child_r)	×
NEW 79	make_nonblocking(child_r)	×
NEW 80	make_nonblocking(parent_sentinel)	×
NEW 81	code = spawn._main(child_r, parent_sentinel)	×
82
NEW 83	return code	×
84
85
86	_forkserver = ForkServer()	18✔
87	ensure_running = _forkserver.ensure_running	18✔
88	get_inherited_fds = _forkserver.get_inherited_fds	18✔
89	connect_to_new_process = _forkserver.connect_to_new_process	18✔
90	set_forkserver_preload = _forkserver.set_forkserver_preload	18✔

karellen / geventmp / 17340928350

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