(
fd: crt_fd::Borrowed<'_>,
buffer: ArgMemoryBuffer,
vm: &VirtualMachine,
)
| 302 | |
| 303 | #[pyfunction] |
| 304 | fn readinto( |
| 305 | fd: crt_fd::Borrowed<'_>, |
| 306 | buffer: ArgMemoryBuffer, |
| 307 | vm: &VirtualMachine, |
| 308 | ) -> PyResult<usize> { |
| 309 | buffer.with_ref(|buf| { |
| 310 | loop { |
| 311 | match vm.allow_threads(|| crt_fd::read(fd, buf)) { |
| 312 | Ok(n) => return Ok(n), |
| 313 | Err(e) if e.raw_os_error() == Some(libc::EINTR) => { |
| 314 | vm.check_signals()?; |
| 315 | continue; |
| 316 | } |
| 317 | Err(e) => return Err(e.into_pyexception(vm)), |
| 318 | } |
| 319 | } |
| 320 | }) |
| 321 | } |
| 322 | |
| 323 | #[pyfunction] |
| 324 | fn write(fd: crt_fd::Borrowed<'_>, data: ArgBytesLike, vm: &VirtualMachine) -> PyResult<usize> { |
nothing calls this directly
no test coverage detected