mirror of
https://github.com/python/cpython.git
synced 2025-10-10 00:43:41 +00:00
Issue #11657: Fix sending file descriptors over 255 over a multiprocessing Pipe.
Also added some tests.
This commit is contained in:
parent
5bd8b8d80f
commit
bcb39d4846
3 changed files with 86 additions and 3 deletions
|
@ -35,7 +35,7 @@ import multiprocessing.managers
|
||||||
import multiprocessing.heap
|
import multiprocessing.heap
|
||||||
import multiprocessing.pool
|
import multiprocessing.pool
|
||||||
|
|
||||||
from multiprocessing import util
|
from multiprocessing import util, reduction
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from multiprocessing.sharedctypes import Value, copy
|
from multiprocessing.sharedctypes import Value, copy
|
||||||
|
@ -43,6 +43,11 @@ try:
|
||||||
except ImportError:
|
except ImportError:
|
||||||
HAS_SHAREDCTYPES = False
|
HAS_SHAREDCTYPES = False
|
||||||
|
|
||||||
|
try:
|
||||||
|
import msvcrt
|
||||||
|
except ImportError:
|
||||||
|
msvcrt = None
|
||||||
|
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
|
@ -72,6 +77,11 @@ HAVE_GETVALUE = not getattr(_multiprocessing,
|
||||||
|
|
||||||
WIN32 = (sys.platform == "win32")
|
WIN32 = (sys.platform == "win32")
|
||||||
|
|
||||||
|
try:
|
||||||
|
MAXFD = os.sysconf("SC_OPEN_MAX")
|
||||||
|
except:
|
||||||
|
MAXFD = 256
|
||||||
|
|
||||||
#
|
#
|
||||||
# Some tests require ctypes
|
# Some tests require ctypes
|
||||||
#
|
#
|
||||||
|
@ -1538,6 +1548,76 @@ class _TestConnection(BaseTestCase):
|
||||||
|
|
||||||
self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
|
self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def _is_fd_assigned(cls, fd):
|
||||||
|
try:
|
||||||
|
os.fstat(fd)
|
||||||
|
except OSError as e:
|
||||||
|
if e.errno == errno.EBADF:
|
||||||
|
return False
|
||||||
|
raise
|
||||||
|
else:
|
||||||
|
return True
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def _writefd(cls, conn, data, create_dummy_fds=False):
|
||||||
|
if create_dummy_fds:
|
||||||
|
for i in range(0, 256):
|
||||||
|
if not cls._is_fd_assigned(i):
|
||||||
|
os.dup2(conn.fileno(), i)
|
||||||
|
fd = reduction.recv_handle(conn)
|
||||||
|
if msvcrt:
|
||||||
|
fd = msvcrt.open_osfhandle(fd, os.O_WRONLY)
|
||||||
|
os.write(fd, data)
|
||||||
|
os.close(fd)
|
||||||
|
|
||||||
|
def test_fd_transfer(self):
|
||||||
|
if self.TYPE != 'processes':
|
||||||
|
self.skipTest("only makes sense with processes")
|
||||||
|
conn, child_conn = self.Pipe(duplex=True)
|
||||||
|
|
||||||
|
p = self.Process(target=self._writefd, args=(child_conn, b"foo"))
|
||||||
|
p.start()
|
||||||
|
with open(test.support.TESTFN, "wb") as f:
|
||||||
|
fd = f.fileno()
|
||||||
|
if msvcrt:
|
||||||
|
fd = msvcrt.get_osfhandle(fd)
|
||||||
|
reduction.send_handle(conn, fd, p.pid)
|
||||||
|
p.join()
|
||||||
|
with open(test.support.TESTFN, "rb") as f:
|
||||||
|
self.assertEqual(f.read(), b"foo")
|
||||||
|
|
||||||
|
@unittest.skipIf(sys.platform == "win32",
|
||||||
|
"test semantics don't make sense on Windows")
|
||||||
|
@unittest.skipIf(MAXFD <= 256,
|
||||||
|
"largest assignable fd number is too small")
|
||||||
|
@unittest.skipUnless(hasattr(os, "dup2"),
|
||||||
|
"test needs os.dup2()")
|
||||||
|
def test_large_fd_transfer(self):
|
||||||
|
# With fd > 256 (issue #11657)
|
||||||
|
if self.TYPE != 'processes':
|
||||||
|
self.skipTest("only makes sense with processes")
|
||||||
|
conn, child_conn = self.Pipe(duplex=True)
|
||||||
|
|
||||||
|
p = self.Process(target=self._writefd, args=(child_conn, b"bar", True))
|
||||||
|
p.start()
|
||||||
|
with open(test.support.TESTFN, "wb") as f:
|
||||||
|
fd = f.fileno()
|
||||||
|
for newfd in range(256, MAXFD):
|
||||||
|
if not self._is_fd_assigned(newfd):
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
self.fail("could not find an unassigned large file descriptor")
|
||||||
|
os.dup2(fd, newfd)
|
||||||
|
try:
|
||||||
|
reduction.send_handle(conn, newfd, p.pid)
|
||||||
|
finally:
|
||||||
|
os.close(newfd)
|
||||||
|
p.join()
|
||||||
|
with open(test.support.TESTFN, "rb") as f:
|
||||||
|
self.assertEqual(f.read(), b"bar")
|
||||||
|
|
||||||
|
|
||||||
class _TestListenerClient(BaseTestCase):
|
class _TestListenerClient(BaseTestCase):
|
||||||
|
|
||||||
ALLOWED_TYPES = ('processes', 'threads')
|
ALLOWED_TYPES = ('processes', 'threads')
|
||||||
|
|
|
@ -22,6 +22,9 @@ Core and Builtins
|
||||||
Library
|
Library
|
||||||
-------
|
-------
|
||||||
|
|
||||||
|
- Issue #11657: Fix sending file descriptors over 255 over a multiprocessing
|
||||||
|
Pipe.
|
||||||
|
|
||||||
- Issue #12213: Fix a buffering bug with interleaved reads and writes that
|
- Issue #12213: Fix a buffering bug with interleaved reads and writes that
|
||||||
could appear on BufferedRandom streams.
|
could appear on BufferedRandom streams.
|
||||||
|
|
||||||
|
|
|
@ -122,7 +122,7 @@ multiprocessing_sendfd(PyObject *self, PyObject *args)
|
||||||
cmsg->cmsg_type = SCM_RIGHTS;
|
cmsg->cmsg_type = SCM_RIGHTS;
|
||||||
cmsg->cmsg_len = CMSG_LEN(sizeof(int));
|
cmsg->cmsg_len = CMSG_LEN(sizeof(int));
|
||||||
msg.msg_controllen = cmsg->cmsg_len;
|
msg.msg_controllen = cmsg->cmsg_len;
|
||||||
*CMSG_DATA(cmsg) = fd;
|
* (int *) CMSG_DATA(cmsg) = fd;
|
||||||
|
|
||||||
Py_BEGIN_ALLOW_THREADS
|
Py_BEGIN_ALLOW_THREADS
|
||||||
res = sendmsg(conn, &msg, 0);
|
res = sendmsg(conn, &msg, 0);
|
||||||
|
@ -165,7 +165,7 @@ multiprocessing_recvfd(PyObject *self, PyObject *args)
|
||||||
if (res < 0)
|
if (res < 0)
|
||||||
return PyErr_SetFromErrno(PyExc_OSError);
|
return PyErr_SetFromErrno(PyExc_OSError);
|
||||||
|
|
||||||
fd = *CMSG_DATA(cmsg);
|
fd = * (int *) CMSG_DATA(cmsg);
|
||||||
return Py_BuildValue("i", fd);
|
return Py_BuildValue("i", fd);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue