passed to the C :c:func:`fcntl` call. The return value after a successful
call is the contents of the buffer, converted to a :class:`bytes` object.
The length of the returned object will be the same as the length of the
- *arg* argument. This is limited to 1024 bytes.
+ *arg* argument.
If the :c:func:`fcntl` call fails, an :exc:`OSError` is raised.
.. note::
- If the type or the size of *arg* does not match the type or size
- of the argument of the operation (for example, if an integer is
+ If the type or size of *arg* does not match the type or size
+ of the operation's argument (for example, if an integer is
passed when a pointer is expected, or the information returned in
- the buffer by the operating system is larger than 1024 bytes),
+ the buffer by the operating system is larger than the size of *arg*),
this is most likely to result in a segmentation violation or
a more subtle data corruption.
Add support of arbitrary :term:`bytes-like objects <bytes-like object>`,
not only :class:`bytes`.
+ .. versionchanged:: next
+ The size of bytes-like objects is no longer limited to 1024 bytes.
+
.. function:: ioctl(fd, request, arg=0, mutate_flag=True, /)
If the type or size of *arg* does not match the type or size
of the operation's argument (for example, if an integer is
passed when a pointer is expected, or the information returned in
- the buffer by the operating system is larger than 1024 bytes,
- or the size of the mutable bytes-like object is too small),
+ the buffer by the operating system is larger than the size of *arg*),
this is most likely to result in a segmentation violation or
a more subtle data corruption.
The GIL is always released during a system call.
System calls failing with EINTR are automatically retried.
+ .. versionchanged:: next
+ The size of not mutated bytes-like objects is no longer
+ limited to 1024 bytes.
+
.. function:: flock(fd, operation, /)
Perform the lock operation *operation* on file descriptor *fd* (file objects providing
os.close(test_pipe_r)
os.close(test_pipe_w)
+ def _check_fcntl_not_mutate_len(self, nbytes=None):
+ self.f = open(TESTFN, 'wb')
+ buf = struct.pack('ii', fcntl.F_OWNER_PID, os.getpid())
+ if nbytes is not None:
+ buf += b' ' * (nbytes - len(buf))
+ else:
+ nbytes = len(buf)
+ save_buf = bytes(buf)
+ r = fcntl.fcntl(self.f, fcntl.F_SETOWN_EX, buf)
+ self.assertIsInstance(r, bytes)
+ self.assertEqual(len(r), len(save_buf))
+ self.assertEqual(buf, save_buf)
+ type, pid = memoryview(r).cast('i')[:2]
+ self.assertEqual(type, fcntl.F_OWNER_PID)
+ self.assertEqual(pid, os.getpid())
+
+ buf = b' ' * nbytes
+ r = fcntl.fcntl(self.f, fcntl.F_GETOWN_EX, buf)
+ self.assertIsInstance(r, bytes)
+ self.assertEqual(len(r), len(save_buf))
+ self.assertEqual(buf, b' ' * nbytes)
+ type, pid = memoryview(r).cast('i')[:2]
+ self.assertEqual(type, fcntl.F_OWNER_PID)
+ self.assertEqual(pid, os.getpid())
+
+ buf = memoryview(b' ' * nbytes)
+ r = fcntl.fcntl(self.f, fcntl.F_GETOWN_EX, buf)
+ self.assertIsInstance(r, bytes)
+ self.assertEqual(len(r), len(save_buf))
+ self.assertEqual(bytes(buf), b' ' * nbytes)
+ type, pid = memoryview(r).cast('i')[:2]
+ self.assertEqual(type, fcntl.F_OWNER_PID)
+ self.assertEqual(pid, os.getpid())
+
+ @unittest.skipUnless(
+ hasattr(fcntl, "F_SETOWN_EX") and hasattr(fcntl, "F_GETOWN_EX"),
+ "requires F_SETOWN_EX and F_GETOWN_EX")
+ def test_fcntl_small_buffer(self):
+ self._check_fcntl_not_mutate_len()
+
+ @unittest.skipUnless(
+ hasattr(fcntl, "F_SETOWN_EX") and hasattr(fcntl, "F_GETOWN_EX"),
+ "requires F_SETOWN_EX and F_GETOWN_EX")
+ def test_fcntl_large_buffer(self):
+ self._check_fcntl_not_mutate_len(2024)
+
if __name__ == '__main__':
unittest.main()
self._check_ioctl_not_mutate_len(1024)
def test_ioctl_mutate_2048(self):
- # Test with a larger buffer, just for the record.
self._check_ioctl_mutate_len(2048)
- self.assertRaises(ValueError, self._check_ioctl_not_mutate_len, 2048)
+ self._check_ioctl_not_mutate_len(1024)
@unittest.skipUnless(hasattr(os, 'openpty'), "need os.openpty()")
--- /dev/null
+:func:`fcntl.fcntl` and :func:`fcntl.ioctl`: Remove the 1024 bytes limit
+on the size of not mutated bytes-like argument.
return NULL;
}
Py_ssize_t len = view.len;
- if (len > FCNTL_BUFSZ) {
- PyErr_SetString(PyExc_ValueError,
- "fcntl argument 3 is too long");
+ if (len <= FCNTL_BUFSZ) {
+ memcpy(buf, view.buf, len);
+ memcpy(buf + len, guard, GUARDSZ);
PyBuffer_Release(&view);
- return NULL;
- }
- memcpy(buf, view.buf, len);
- memcpy(buf + len, guard, GUARDSZ);
- PyBuffer_Release(&view);
- do {
- Py_BEGIN_ALLOW_THREADS
- ret = fcntl(fd, code, buf);
- Py_END_ALLOW_THREADS
- } while (ret == -1 && errno == EINTR && !(async_err = PyErr_CheckSignals()));
- if (ret < 0) {
- return !async_err ? PyErr_SetFromErrno(PyExc_OSError) : NULL;
+ do {
+ Py_BEGIN_ALLOW_THREADS
+ ret = fcntl(fd, code, buf);
+ Py_END_ALLOW_THREADS
+ } while (ret == -1 && errno == EINTR && !(async_err = PyErr_CheckSignals()));
+ if (ret < 0) {
+ return !async_err ? PyErr_SetFromErrno(PyExc_OSError) : NULL;
+ }
+ if (memcmp(buf + len, guard, GUARDSZ) != 0) {
+ PyErr_SetString(PyExc_SystemError, "buffer overflow");
+ return NULL;
+ }
+ return PyBytes_FromStringAndSize(buf, len);
}
- if (memcmp(buf + len, guard, GUARDSZ) != 0) {
- PyErr_SetString(PyExc_SystemError, "buffer overflow");
- return NULL;
+ else {
+ PyObject *result = PyBytes_FromStringAndSize(NULL, len);
+ if (result == NULL) {
+ PyBuffer_Release(&view);
+ return NULL;
+ }
+ char *ptr = PyBytes_AsString(result);
+ memcpy(ptr, view.buf, len);
+ PyBuffer_Release(&view);
+
+ do {
+ Py_BEGIN_ALLOW_THREADS
+ ret = fcntl(fd, code, ptr);
+ Py_END_ALLOW_THREADS
+ } while (ret == -1 && errno == EINTR && !(async_err = PyErr_CheckSignals()));
+ if (ret < 0) {
+ if (async_err) {
+ PyErr_SetFromErrno(PyExc_OSError);
+ }
+ Py_DECREF(result);
+ return NULL;
+ }
+ if (ptr[len] != '\0') {
+ PyErr_SetString(PyExc_SystemError, "buffer overflow");
+ return NULL;
+ }
+ return result;
}
- return PyBytes_FromStringAndSize(buf, len);
#undef FCNTL_BUFSZ
}
PyErr_Format(PyExc_TypeError,
return NULL;
}
Py_ssize_t len = view.len;
- if (len > IOCTL_BUFSZ) {
- PyErr_SetString(PyExc_ValueError,
- "ioctl argument 3 is too long");
+ if (len <= IOCTL_BUFSZ) {
+ memcpy(buf, view.buf, len);
+ memcpy(buf + len, guard, GUARDSZ);
PyBuffer_Release(&view);
- return NULL;
- }
- memcpy(buf, view.buf, len);
- memcpy(buf + len, guard, GUARDSZ);
- PyBuffer_Release(&view);
- do {
- Py_BEGIN_ALLOW_THREADS
- ret = ioctl(fd, code, buf);
- Py_END_ALLOW_THREADS
- } while (ret == -1 && errno == EINTR && !(async_err = PyErr_CheckSignals()));
- if (ret < 0) {
- return !async_err ? PyErr_SetFromErrno(PyExc_OSError) : NULL;
+ do {
+ Py_BEGIN_ALLOW_THREADS
+ ret = ioctl(fd, code, buf);
+ Py_END_ALLOW_THREADS
+ } while (ret == -1 && errno == EINTR && !(async_err = PyErr_CheckSignals()));
+ if (ret < 0) {
+ return !async_err ? PyErr_SetFromErrno(PyExc_OSError) : NULL;
+ }
+ if (memcmp(buf + len, guard, GUARDSZ) != 0) {
+ PyErr_SetString(PyExc_SystemError, "buffer overflow");
+ return NULL;
+ }
+ return PyBytes_FromStringAndSize(buf, len);
}
- if (memcmp(buf + len, guard, GUARDSZ) != 0) {
- PyErr_SetString(PyExc_SystemError, "buffer overflow");
- return NULL;
+ else {
+ PyObject *result = PyBytes_FromStringAndSize(NULL, len);
+ if (result == NULL) {
+ PyBuffer_Release(&view);
+ return NULL;
+ }
+ char *ptr = PyBytes_AsString(result);
+ memcpy(ptr, view.buf, len);
+ PyBuffer_Release(&view);
+
+ do {
+ Py_BEGIN_ALLOW_THREADS
+ ret = ioctl(fd, code, ptr);
+ Py_END_ALLOW_THREADS
+ } while (ret == -1 && errno == EINTR && !(async_err = PyErr_CheckSignals()));
+ if (ret < 0) {
+ if (async_err) {
+ PyErr_SetFromErrno(PyExc_OSError);
+ }
+ Py_DECREF(result);
+ return NULL;
+ }
+ if (ptr[len] != '\0') {
+ PyErr_SetString(PyExc_SystemError, "buffer overflow");
+ return NULL;
+ }
+ return result;
}
- return PyBytes_FromStringAndSize(buf, len);
#undef IOCTL_BUFSZ
}
PyErr_Format(PyExc_TypeError,
projects / thirdparty / Python / cpython.git / commitdiff
