io.rs

/*
 * I/O core tools.
 */
cfg_if::cfg_if! {
    if #[cfg(any(not(target_arch = "wasm32"), target_os = "wasi"))] {
        use crate::common::crt_fd::Offset;
    } else {
        type Offset = i64;
    }
}

use crate::{
    PyObjectRef, PyRef, PyResult, TryFromObject, VirtualMachine,
    builtins::{PyBaseExceptionRef, PyModule},
    common::os::ErrorExt,
    convert::{IntoPyException, ToPyException},
};
pub use _io::{OpenArgs, io_open as open};

impl ToPyException for std::io::Error {
    fn to_pyexception(&self, vm: &VirtualMachine) -> PyBaseExceptionRef {
        let errno = self.posix_errno();
        let msg = self.to_string();
        #[allow(clippy::let_and_return)]
        let exc = vm.new_errno_error(errno, msg);

        #[cfg(windows)]
        {
            use crate::object::AsObject;
            let winerror = if let Some(winerror) = self.raw_os_error() {
                vm.new_pyobj(winerror)
            } else {
                vm.ctx.none()
            };

            // FIXME: manual setup winerror due to lack of OSError.__init__ support
            exc.as_object()
                .set_attr("winerror", vm.new_pyobj(winerror), vm)
                .unwrap();
        }
        exc
    }
}

impl IntoPyException for std::io::Error {
    fn into_pyexception(self, vm: &VirtualMachine) -> PyBaseExceptionRef {
        self.to_pyexception(vm)
    }
}

pub(crate) fn make_module(vm: &VirtualMachine) -> PyRef<PyModule> {
    let ctx = &vm.ctx;

    let module = _io::make_module(vm);

    #[cfg(any(not(target_arch = "wasm32"), target_os = "wasi"))]
    fileio::extend_module(vm, &module).unwrap();

    let unsupported_operation = _io::UNSUPPORTED_OPERATION
        .get_or_init(|| _io::make_unsupportedop(ctx))
        .clone();
    extend_module!(vm, &module, {
        "UnsupportedOperation" => unsupported_operation,
        "BlockingIOError" => ctx.exceptions.blocking_io_error.to_owned(),
    });

    module
}

// not used on all platforms
#[derive(Copy, Clone)]
#[repr(transparent)]
pub struct Fildes(pub i32);

impl TryFromObject for Fildes {
    fn try_from_object(vm: &VirtualMachine, obj: PyObjectRef) -> PyResult<Self> {
        use crate::builtins::int;
        let int = match obj.downcast::<int::PyInt>() {
            Ok(i) => i,
            Err(obj) => {
                let fileno_meth = vm.get_attribute_opt(obj, "fileno")?.ok_or_else(|| {
                    vm.new_type_error("argument must be an int, or have a fileno() method.")
                })?;
                fileno_meth
                    .call((), vm)?
                    .downcast()
                    .map_err(|_| vm.new_type_error("fileno() returned a non-integer"))?
            }
        };
        let fd = int.try_to_primitive(vm)?;
        if fd < 0 {
            return Err(vm.new_value_error(format!(
                "file descriptor cannot be a negative integer ({fd})"
            )));
        }
        Ok(Fildes(fd))
    }
}

#[cfg(unix)]
impl std::os::fd::AsFd for Fildes {
    fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
        // SAFETY: none, really. but, python's os api of passing around file descriptors
        //         everywhere isn't really io-safe anyway, so, this is passed to the user.
        unsafe { std::os::fd::BorrowedFd::borrow_raw(self.0) }
    }
}
#[cfg(unix)]
impl std::os::fd::AsRawFd for Fildes {
    fn as_raw_fd(&self) -> std::os::fd::RawFd {
        self.0
    }
}

#[pymodule]
mod _io {
    use super::*;
    use crate::{
        AsObject, Context, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult,
        TryFromBorrowedObject, TryFromObject,
        builtins::{
            PyBaseExceptionRef, PyByteArray, PyBytes, PyBytesRef, PyIntRef, PyMemoryView, PyStr,
            PyStrRef, PyTuple, PyTupleRef, PyType, PyTypeRef,
        },
        class::StaticType,
        common::lock::{
            PyMappedThreadMutexGuard, PyRwLock, PyRwLockReadGuard, PyRwLockWriteGuard,
            PyThreadMutex, PyThreadMutexGuard,
        },
        common::wtf8::{Wtf8, Wtf8Buf},
        convert::ToPyObject,
        function::{
            ArgBytesLike, ArgIterable, ArgMemoryBuffer, ArgSize, Either, FuncArgs, IntoFuncArgs,
            OptionalArg, OptionalOption, PySetterValue,
        },
        protocol::{
            BufferDescriptor, BufferMethods, BufferResizeGuard, PyBuffer, PyIterReturn, VecBuffer,
        },
        recursion::ReprGuard,
        types::{
            Callable, Constructor, DefaultConstructor, Destructor, Initializer, IterNext, Iterable,
        },
        vm::VirtualMachine,
    };
    use bstr::ByteSlice;
    use crossbeam_utils::atomic::AtomicCell;
    use malachite_bigint::{BigInt, BigUint};
    use num_traits::ToPrimitive;
    use std::{
        borrow::Cow,
        io::{self, Cursor, SeekFrom, prelude::*},
        ops::Range,
    };

    #[allow(clippy::let_and_return)]
    fn validate_whence(whence: i32) -> bool {
        let x = (0..=2).contains(&whence);
        cfg_if::cfg_if! {
            if #[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "linux"))] {
                x || matches!(whence, libc::SEEK_DATA | libc::SEEK_HOLE)
            } else {
                x
            }
        }
    }

    fn ensure_unclosed(file: &PyObject, msg: &str, vm: &VirtualMachine) -> PyResult<()> {
        if file.get_attr("closed", vm)?.try_to_bool(vm)? {
            Err(vm.new_value_error(msg))
        } else {
            Ok(())
        }
    }

    pub fn new_unsupported_operation(vm: &VirtualMachine, msg: String) -> PyBaseExceptionRef {
        vm.new_exception_msg(UNSUPPORTED_OPERATION.get().unwrap().clone(), msg)
    }

    fn _unsupported<T>(vm: &VirtualMachine, zelf: &PyObject, operation: &str) -> PyResult<T> {
        Err(new_unsupported_operation(
            vm,
            format!("{}.{}() not supported", zelf.class().name(), operation),
        ))
    }

    #[derive(FromArgs)]
    pub(super) struct OptionalSize {
        // In a few functions, the default value is -1 rather than None.
        // Make sure the default value doesn't affect compatibility.
        #[pyarg(positional, default)]
        size: Option<ArgSize>,
    }

    impl OptionalSize {
        #[allow(clippy::wrong_self_convention)]
        pub fn to_usize(self) -> Option<usize> {
            self.size?.to_usize()
        }

        pub fn try_usize(self, vm: &VirtualMachine) -> PyResult<Option<usize>> {
            self.size
                .map(|v| {
                    let v = *v;
                    if v >= 0 {
                        Ok(v as usize)
                    } else {
                        Err(vm.new_value_error(format!("Negative size value {v}")))
                    }
                })
                .transpose()
        }
    }

    fn os_err(vm: &VirtualMachine, err: io::Error) -> PyBaseExceptionRef {
        #[cfg(any(not(target_arch = "wasm32"), target_os = "wasi"))]
        {
            use crate::convert::ToPyException;
            err.to_pyexception(vm)
        }
        #[cfg(all(target_arch = "wasm32", not(target_os = "wasi")))]
        {
            vm.new_os_error(err.to_string())
        }
    }

    pub(super) fn io_closed_error(vm: &VirtualMachine) -> PyBaseExceptionRef {
        vm.new_value_error("I/O operation on closed file")
    }

    #[pyattr]
    const DEFAULT_BUFFER_SIZE: usize = 8 * 1024;

    pub(super) fn seekfrom(
        vm: &VirtualMachine,
        offset: PyObjectRef,
        how: OptionalArg<i32>,
    ) -> PyResult<SeekFrom> {
        let seek = match how {
            OptionalArg::Present(0) | OptionalArg::Missing => {
                SeekFrom::Start(offset.try_into_value(vm)?)
            }
            OptionalArg::Present(1) => SeekFrom::Current(offset.try_into_value(vm)?),
            OptionalArg::Present(2) => SeekFrom::End(offset.try_into_value(vm)?),
            _ => return Err(vm.new_value_error("invalid value for how")),
        };
        Ok(seek)
    }

    #[derive(Debug)]
    struct BufferedIO {
        cursor: Cursor<Vec<u8>>,
    }

    impl BufferedIO {
        fn new(cursor: Cursor<Vec<u8>>) -> BufferedIO {
            BufferedIO { cursor }
        }

        fn write(&mut self, data: &[u8]) -> Option<u64> {
            if data.is_empty() {
                return Some(0);
            }
            let length = data.len();
            self.cursor.write_all(data).ok()?;
            Some(length as u64)
        }

        //return the entire contents of the underlying
        fn getvalue(&self) -> Vec<u8> {
            self.cursor.clone().into_inner()
        }

        //skip to the jth position
        fn seek(&mut self, seek: SeekFrom) -> io::Result<u64> {
            self.cursor.seek(seek)
        }

        //Read k bytes from the object and return.
        fn read(&mut self, bytes: Option<usize>) -> Option<Vec<u8>> {
            let pos = self.cursor.position().to_usize()?;
            let avail_slice = self.cursor.get_ref().get(pos..)?;
            // if we don't specify the number of bytes, or it's too big, give the whole rest of the slice
            let n = bytes.map_or_else(
                || avail_slice.len(),
                |n| std::cmp::min(n, avail_slice.len()),
            );
            let b = avail_slice[..n].to_vec();
            self.cursor.set_position((pos + n) as u64);
            Some(b)
        }

        fn tell(&self) -> u64 {
            self.cursor.position()
        }

        fn readline(&mut self, size: Option<usize>, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
            self.read_until(size, b'\n', vm)
        }

        fn read_until(
            &mut self,
            size: Option<usize>,
            byte: u8,
            vm: &VirtualMachine,
        ) -> PyResult<Vec<u8>> {
            let size = match size {
                None => {
                    let mut buf: Vec<u8> = Vec::new();
                    self.cursor
                        .read_until(byte, &mut buf)
                        .map_err(|err| os_err(vm, err))?;
                    return Ok(buf);
                }
                Some(0) => {
                    return Ok(Vec::new());
                }
                Some(size) => size,
            };

            let available = {
                // For Cursor, fill_buf returns all of the remaining data unlike other BufReads which have outer reading source.
                // Unless we add other data by write, there will be no more data.
                let buf = self.cursor.fill_buf().map_err(|err| os_err(vm, err))?;
                if size < buf.len() { &buf[..size] } else { buf }
            };
            let buf = match available.find_byte(byte) {
                Some(i) => available[..=i].to_vec(),
                _ => available.to_vec(),
            };
            self.cursor.consume(buf.len());
            Ok(buf)
        }

        fn truncate(&mut self, pos: Option<usize>) -> usize {
            let pos = pos.unwrap_or_else(|| self.tell() as usize);
            self.cursor.get_mut().truncate(pos);
            pos
        }
    }

    fn file_closed(file: &PyObject, vm: &VirtualMachine) -> PyResult<bool> {
        file.get_attr("closed", vm)?.try_to_bool(vm)
    }
    fn check_closed(file: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
        if file_closed(file, vm)? {
            Err(io_closed_error(vm))
        } else {
            Ok(())
        }
    }

    fn check_readable(file: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
        if vm.call_method(file, "readable", ())?.try_to_bool(vm)? {
            Ok(())
        } else {
            Err(new_unsupported_operation(
                vm,
                "File or stream is not readable".to_owned(),
            ))
        }
    }

    fn check_writable(file: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
        if vm.call_method(file, "writable", ())?.try_to_bool(vm)? {
            Ok(())
        } else {
            Err(new_unsupported_operation(
                vm,
                "File or stream is not writable.".to_owned(),
            ))
        }
    }

    fn check_seekable(file: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
        if vm.call_method(file, "seekable", ())?.try_to_bool(vm)? {
            Ok(())
        } else {
            Err(new_unsupported_operation(
                vm,
                "File or stream is not seekable".to_owned(),
            ))
        }
    }

    fn check_decoded(decoded: PyObjectRef, vm: &VirtualMachine) -> PyResult<PyStrRef> {
        decoded.downcast().map_err(|obj| {
            vm.new_type_error(format!(
                "decoder should return a string result, not '{}'",
                obj.class().name()
            ))
        })
    }

    #[pyattr]
    #[pyclass(name = "_IOBase")]
    #[derive(Debug, PyPayload)]
    pub struct _IOBase;

    #[pyclass(with(IterNext, Iterable, Destructor), flags(BASETYPE, HAS_DICT))]
    impl _IOBase {
        #[pymethod]
        fn seek(
            zelf: PyObjectRef,
            _pos: PyObjectRef,
            _whence: OptionalArg,
            vm: &VirtualMachine,
        ) -> PyResult {
            _unsupported(vm, &zelf, "seek")
        }
        #[pymethod]
        fn tell(zelf: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            vm.call_method(&zelf, "seek", (0, 1))
        }
        #[pymethod]
        fn truncate(zelf: PyObjectRef, _pos: OptionalArg, vm: &VirtualMachine) -> PyResult {
            _unsupported(vm, &zelf, "truncate")
        }
        #[pymethod]
        fn fileno(zelf: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            _unsupported(vm, &zelf, "truncate")
        }

        #[pyattr]
        fn __closed(ctx: &Context) -> PyIntRef {
            ctx.new_bool(false)
        }

        #[pymethod]
        fn __enter__(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            check_closed(&instance, vm)?;
            Ok(instance)
        }

        #[pymethod]
        fn __exit__(instance: PyObjectRef, _args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> {
            vm.call_method(&instance, "close", ())?;
            Ok(())
        }

        #[pymethod]
        fn flush(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
            // just check if this is closed; if it isn't, do nothing
            check_closed(&instance, vm)
        }

        #[pymethod]
        fn seekable(_self: PyObjectRef) -> bool {
            false
        }
        #[pymethod]
        fn readable(_self: PyObjectRef) -> bool {
            false
        }
        #[pymethod]
        fn writable(_self: PyObjectRef) -> bool {
            false
        }

        #[pymethod]
        fn isatty(_self: PyObjectRef) -> bool {
            false
        }

        #[pygetset]
        fn closed(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            instance.get_attr("__closed", vm)
        }

        #[pymethod]
        fn close(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
            iobase_close(&instance, vm)
        }

        #[pymethod]
        fn readline(
            instance: PyObjectRef,
            size: OptionalSize,
            vm: &VirtualMachine,
        ) -> PyResult<Vec<u8>> {
            let size = size.to_usize();
            let read = instance.get_attr("read", vm)?;
            let mut res = Vec::new();
            while size.is_none_or(|s| res.len() < s) {
                let read_res = ArgBytesLike::try_from_object(vm, read.call((1,), vm)?)?;
                if read_res.with_ref(|b| b.is_empty()) {
                    break;
                }
                read_res.with_ref(|b| res.extend_from_slice(b));
                if res.ends_with(b"\n") {
                    break;
                }
            }
            Ok(res)
        }

        #[pymethod]
        fn readlines(
            instance: PyObjectRef,
            hint: OptionalOption<isize>,
            vm: &VirtualMachine,
        ) -> PyResult<Vec<PyObjectRef>> {
            let hint = hint.flatten().unwrap_or(-1);
            if hint <= 0 {
                return instance.try_to_value(vm);
            }
            let hint = hint as usize;
            let mut ret = Vec::new();
            let it = ArgIterable::<PyObjectRef>::try_from_object(vm, instance)?;
            let mut full_len = 0;
            for line in it.iter(vm)? {
                let line = line?;
                let line_len = line.length(vm)?;
                ret.push(line.clone());
                full_len += line_len;
                if full_len > hint {
                    break;
                }
            }
            Ok(ret)
        }

        #[pymethod]
        fn writelines(
            instance: PyObjectRef,
            lines: ArgIterable,
            vm: &VirtualMachine,
        ) -> PyResult<()> {
            check_closed(&instance, vm)?;
            for line in lines.iter(vm)? {
                vm.call_method(&instance, "write", (line?,))?;
            }
            Ok(())
        }

        #[pymethod(name = "_checkClosed")]
        fn check_closed(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
            check_closed(&instance, vm)
        }

        #[pymethod(name = "_checkReadable")]
        fn check_readable(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
            check_readable(&instance, vm)
        }

        #[pymethod(name = "_checkWritable")]
        fn check_writable(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
            check_writable(&instance, vm)
        }

        #[pymethod(name = "_checkSeekable")]
        fn check_seekable(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
            check_seekable(&instance, vm)
        }
    }

    impl Destructor for _IOBase {
        fn slot_del(zelf: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
            let _ = vm.call_method(zelf, "close", ());
            Ok(())
        }

        #[cold]
        fn del(_zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<()> {
            unreachable!("slot_del is implemented")
        }
    }

    impl Iterable for _IOBase {
        fn slot_iter(zelf: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            check_closed(&zelf, vm)?;
            Ok(zelf)
        }

        fn iter(_zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyResult {
            unreachable!("slot_iter is implemented")
        }
    }

    impl IterNext for _IOBase {
        fn slot_iternext(zelf: &PyObject, vm: &VirtualMachine) -> PyResult<PyIterReturn> {
            let line = vm.call_method(zelf, "readline", ())?;
            Ok(if !line.clone().try_to_bool(vm)? {
                PyIterReturn::StopIteration(None)
            } else {
                PyIterReturn::Return(line)
            })
        }

        fn next(_zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<PyIterReturn> {
            unreachable!("slot_iternext is implemented")
        }
    }

    pub(super) fn iobase_close(file: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
        if !file_closed(file, vm)? {
            let res = vm.call_method(file, "flush", ());
            file.set_attr("__closed", vm.new_pyobj(true), vm)?;
            res?;
        }
        Ok(())
    }

    #[pyattr]
    #[pyclass(name = "_RawIOBase", base = "_IOBase")]
    pub(super) struct _RawIOBase;

    #[pyclass(flags(BASETYPE, HAS_DICT))]
    impl _RawIOBase {
        #[pymethod]
        fn read(instance: PyObjectRef, size: OptionalSize, vm: &VirtualMachine) -> PyResult {
            if let Some(size) = size.to_usize() {
                // FIXME: unnecessary zero-init
                let b = PyByteArray::from(vec![0; size]).into_ref(&vm.ctx);
                let n = <Option<usize>>::try_from_object(
                    vm,
                    vm.call_method(&instance, "readinto", (b.clone(),))?,
                )?;
                Ok(n.map(|n| {
                    let mut bytes = b.borrow_buf_mut();
                    bytes.truncate(n);
                    // FIXME: try to use Arc::unwrap on the bytearray to get at the inner buffer
                    bytes.clone()
                })
                .to_pyobject(vm))
            } else {
                vm.call_method(&instance, "readall", ())
            }
        }

        #[pymethod]
        fn readall(instance: PyObjectRef, vm: &VirtualMachine) -> PyResult<Option<Vec<u8>>> {
            let mut chunks = Vec::new();
            let mut total_len = 0;
            loop {
                let data = vm.call_method(&instance, "read", (DEFAULT_BUFFER_SIZE,))?;
                let data = <Option<PyBytesRef>>::try_from_object(vm, data)?;
                match data {
                    None => {
                        if chunks.is_empty() {
                            return Ok(None);
                        }
                        break;
                    }
                    Some(b) => {
                        if b.as_bytes().is_empty() {
                            break;
                        }
                        total_len += b.as_bytes().len();
                        chunks.push(b)
                    }
                }
            }
            let mut ret = Vec::with_capacity(total_len);
            for b in chunks {
                ret.extend_from_slice(b.as_bytes())
            }
            Ok(Some(ret))
        }
    }

    #[pyattr]
    #[pyclass(name = "_BufferedIOBase", base = "_IOBase")]
    struct _BufferedIOBase;

    #[pyclass(flags(BASETYPE))]
    impl _BufferedIOBase {
        #[pymethod]
        fn read(zelf: PyObjectRef, _size: OptionalArg, vm: &VirtualMachine) -> PyResult {
            _unsupported(vm, &zelf, "read")
        }
        #[pymethod]
        fn read1(zelf: PyObjectRef, _size: OptionalArg, vm: &VirtualMachine) -> PyResult {
            _unsupported(vm, &zelf, "read1")
        }
        fn _readinto(
            zelf: PyObjectRef,
            buf_obj: PyObjectRef,
            method: &str,
            vm: &VirtualMachine,
        ) -> PyResult<usize> {
            let b = ArgMemoryBuffer::try_from_borrowed_object(vm, &buf_obj)?;
            let l = b.len();
            let data = vm.call_method(&zelf, method, (l,))?;
            if data.is(&buf_obj) {
                return Ok(l);
            }
            let mut buf = b.borrow_buf_mut();
            let data = ArgBytesLike::try_from_object(vm, data)?;
            let data = data.borrow_buf();
            match buf.get_mut(..data.len()) {
                Some(slice) => {
                    slice.copy_from_slice(&data);
                    Ok(data.len())
                }
                None => {
                    Err(vm.new_value_error("readinto: buffer and read data have different lengths"))
                }
            }
        }
        #[pymethod]
        fn readinto(zelf: PyObjectRef, b: PyObjectRef, vm: &VirtualMachine) -> PyResult<usize> {
            Self::_readinto(zelf, b, "read", vm)
        }
        #[pymethod]
        fn readinto1(zelf: PyObjectRef, b: PyObjectRef, vm: &VirtualMachine) -> PyResult<usize> {
            Self::_readinto(zelf, b, "read1", vm)
        }
        #[pymethod]
        fn write(zelf: PyObjectRef, _b: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            _unsupported(vm, &zelf, "write")
        }
        #[pymethod]
        fn detach(zelf: PyObjectRef, vm: &VirtualMachine) -> PyResult {
            _unsupported(vm, &zelf, "detach")
        }
    }

    // TextIO Base has no public constructor
    #[pyattr]
    #[pyclass(name = "_TextIOBase", base = "_IOBase")]
    #[derive(Debug, PyPayload)]
    struct _TextIOBase;

    #[pyclass(flags(BASETYPE))]
    impl _TextIOBase {
        #[pygetset]
        fn encoding(_zelf: PyObjectRef, vm: &VirtualMachine) -> PyObjectRef {
            vm.ctx.none()
        }
    }

    #[derive(FromArgs, Clone)]
    struct BufferSize {
        #[pyarg(any, optional)]
        buffer_size: OptionalArg<isize>,
    }

    bitflags::bitflags! {
        #[derive(Copy, Clone, Debug, PartialEq, Default)]
        struct BufferedFlags: u8 {
            const DETACHED = 1 << 0;
            const WRITABLE = 1 << 1;
            const READABLE = 1 << 2;
        }
    }

    #[derive(Debug, Default)]
    struct BufferedData {
        raw: Option<PyObjectRef>,
        flags: BufferedFlags,
        abs_pos: Offset,
        buffer: Vec<u8>,
        pos: Offset,
        raw_pos: Offset,
        read_end: Offset,
        write_pos: Offset,
        write_end: Offset,
    }

    impl BufferedData {
        fn check_init(&self, vm: &VirtualMachine) -> PyResult<&PyObject> {
            if let Some(raw) = &self.raw {
                Ok(raw)
            } else {
                let msg = if self.flags.contains(BufferedFlags::DETACHED) {
                    "raw stream has been detached"
                } else {
                    "I/O operation on uninitialized object"
                };
                Err(vm.new_value_error(msg))
            }
        }

        #[inline]
        fn writable(&self) -> bool {
            self.flags.contains(BufferedFlags::WRITABLE)
        }
        #[inline]
        fn readable(&self) -> bool {
            self.flags.contains(BufferedFlags::READABLE)
        }

        #[inline]
        fn valid_read(&self) -> bool {
            self.readable() && self.read_end != -1
        }
        #[inline]
        fn valid_write(&self) -> bool {
            self.writable() && self.write_end != -1
        }

        #[inline]
        fn raw_offset(&self) -> Offset {
            if (self.valid_read() || self.valid_write()) && self.raw_pos >= 0 {
                self.raw_pos - self.pos
            } else {
                0
            }
        }
        #[inline]
        fn readahead(&self) -> Offset {
            if self.valid_read() {
                self.read_end - self.pos
            } else {
                0
            }
        }

        fn reset_read(&mut self) {
            self.read_end = -1;
        }
        fn reset_write(&mut self) {
            self.write_pos = 0;
            self.write_end = -1;
        }

        fn flush(&mut self, vm: &VirtualMachine) -> PyResult<()> {
            if !self.valid_write() || self.write_pos == self.write_end {
                self.reset_write();
                return Ok(());
            }

            let rewind = self.raw_offset() + (self.pos - self.write_pos);
            if rewind != 0 {
                self.raw_seek(-rewind, 1, vm)?;
                self.raw_pos = -rewind;
            }

            while self.write_pos < self.write_end {
                let n =
                    self.raw_write(None, self.write_pos as usize..self.write_end as usize, vm)?;
                let n = n.ok_or_else(|| {
                    vm.new_exception_msg(
                        vm.ctx.exceptions.blocking_io_error.to_owned(),
                        "write could not complete without blocking".to_owned(),
                    )
                })?;
                self.write_pos += n as Offset;
                self.raw_pos = self.write_pos;
                vm.check_signals()?;
            }

            self.reset_write();

            Ok(())
        }

        fn flush_rewind(&mut self, vm: &VirtualMachine) -> PyResult<()> {
            self.flush(vm)?;
            if self.readable() {
                let res = self.raw_seek(-self.raw_offset(), 1, vm);
                self.reset_read();
                res?;
            }
            Ok(())
        }

        fn raw_seek(&mut self, pos: Offset, whence: i32, vm: &VirtualMachine) -> PyResult<Offset> {
            let ret = vm.call_method(self.check_init(vm)?, "seek", (pos, whence))?;
            let offset = get_offset(ret, vm)?;
            if offset < 0 {
                return Err(
                    vm.new_os_error(format!("Raw stream returned invalid position {offset}"))
                );
            }
            self.abs_pos = offset;
            Ok(offset)
        }

        fn seek(&mut self, target: Offset, whence: i32, vm: &VirtualMachine) -> PyResult<Offset> {
            if matches!(whence, 0 | 1) && self.readable() {
                let current = self.raw_tell_cache(vm)?;
                let available = self.readahead();
                if available > 0 {
                    let offset = if whence == 0 {
                        target - (current - self.raw_offset())
                    } else {
                        target
                    };
                    if offset >= -self.pos && offset <= available {
                        self.pos += offset;
                        return Ok(current - available + offset);
                    }
                }
            }
            // raw.get_attr("seek", vm)?.call(args, vm)
            if self.writable() {
                self.flush(vm)?;
            }
            let target = if whence == 1 {
                target - self.raw_offset()
            } else {
                target
            };
            let res = self.raw_seek(target, whence, vm);
            self.raw_pos = -1;
            if res.is_ok() && self.readable() {
                self.reset_read();
            }
            res
        }

        fn raw_tell(&mut self, vm: &VirtualMachine) -> PyResult<Offset> {
            let raw = self.check_init(vm)?;
            let ret = vm.call_method(raw, "tell", ())?;
            let offset = get_offset(ret, vm)?;
            if offset < 0 {
                return Err(
                    vm.new_os_error(format!("Raw stream returned invalid position {offset}"))
                );
            }
            self.abs_pos = offset;
            Ok(offset)
        }

        fn raw_tell_cache(&mut self, vm: &VirtualMachine) -> PyResult<Offset> {
            if self.abs_pos == -1 {
                self.raw_tell(vm)
            } else {
                Ok(self.abs_pos)
            }
        }

        /// None means non-blocking failed
        fn raw_write(
            &mut self,
            buf: Option<PyBuffer>,
            buf_range: Range<usize>,
            vm: &VirtualMachine,
        ) -> PyResult<Option<usize>> {
            let len = buf_range.len();
            let res = if let Some(buf) = buf {
                let mem_obj = PyMemoryView::from_buffer_range(buf, buf_range, vm)?.to_pyobject(vm);

                // TODO: loop if write() raises an interrupt
                vm.call_method(self.raw.as_ref().unwrap(), "write", (mem_obj,))?
            } else {
                let v = std::mem::take(&mut self.buffer);
                let write_buf = VecBuffer::from(v).into_ref(&vm.ctx);
                let mem_obj = PyMemoryView::from_buffer_range(
                    write_buf.clone().into_pybuffer(true),
                    buf_range,
                    vm,
                )?
                .into_ref(&vm.ctx);

                // TODO: loop if write() raises an interrupt
                let res = vm.call_method(self.raw.as_ref().unwrap(), "write", (mem_obj.clone(),));

                mem_obj.release();
                self.buffer = write_buf.take();

                res?
            };

            if vm.is_none(&res) {
                return Ok(None);
            }
            let n = isize::try_from_object(vm, res)?;
            if n < 0 || n as usize > len {
                return Err(vm.new_os_error(format!(
                    "raw write() returned invalid length {n} (should have been between 0 and {len})"
                )));
            }
            if self.abs_pos != -1 {
                self.abs_pos += n as Offset
            }
            Ok(Some(n as usize))
        }

        fn write(&mut self, obj: ArgBytesLike, vm: &VirtualMachine) -> PyResult<usize> {
            if !self.valid_read() && !self.valid_write() {
                self.pos = 0;
                self.raw_pos = 0;
            }
            let avail = self.buffer.len() - self.pos as usize;
            let buf_len;
            {
                let buf = obj.borrow_buf();
                buf_len = buf.len();
                if buf.len() <= avail {
                    self.buffer[self.pos as usize..][..buf.len()].copy_from_slice(&buf);
                    if !self.valid_write() || self.write_pos > self.pos {
                        self.write_pos = self.pos
                    }
                    self.adjust_position(self.pos + buf.len() as Offset);
                    if self.pos > self.write_end {
                        self.write_end = self.pos
                    }
                    return Ok(buf.len());
                }
            }

            // TODO: something something check if error is BlockingIOError?
            let _ = self.flush(vm);

            let offset = self.raw_offset();
            if offset != 0 {
                self.raw_seek(-offset, 1, vm)?;
                self.raw_pos -= offset;