Construct: Rebuffered stream

Question

This is part of the Construct library. Rebuffered is used to turn a seekless/tellless stream like from a socket or pipe into a seekable/tellable one like a BytesIO or a file.

If tailcutoff is provided, only that amount of bytes is kept counting from the end of last read write operation backwards. Otherwise entire stream is cached.

class RebufferedBytesIO:
    __slots__ = ["substream","offset","rwbuffer","moved","tailcutoff"]

    def __init__(self, substream, tailcutoff=None):
        self.substream = substream
        self.offset = 0
        self.rwbuffer = b""
        self.moved = 0
        self.tailcutoff = tailcutoff

    def read(self, count):
        startsat = self.offset
        endsat = startsat + count
        if startsat < self.moved:
            raise IOError("could not read because tail was cut off")
        while self.moved + len(self.rwbuffer) < endsat:
            newdata = self.substream.read(128*1024)
            self.rwbuffer += newdata
            if not newdata:
                sleep(0)
        data = self.rwbuffer[startsat-self.moved:endsat-self.moved]
        self.offset += count
        if self.tailcutoff is not None and self.moved < self.offset - self.tailcutoff:
            removed = self.offset - self.tailcutoff - self.moved
            self.moved += removed
            self.rwbuffer = self.rwbuffer[removed:]
        if len(data) < count:
            raise IOError("could not read enough bytes, something went wrong")
        return data

    def write(self, data):
        startsat = self.offset
        endsat = startsat + len(data)
        while self.moved + len(self.rwbuffer) < startsat:
            newdata = self.substream.read(128*1024)
            self.rwbuffer += newdata
            if not newdata:
                sleep(0)
        self.rwbuffer = self.rwbuffer[:startsat-self.moved] + data + self.rwbuffer[endsat-self.moved:]
        self.offset = endsat
        if self.tailcutoff is not None and self.moved < self.offset - self.tailcutoff:
            removed = self.offset - self.tailcutoff - self.moved
            self.moved += removed
            self.rwbuffer = self.rwbuffer[removed:]
        return len(data)

    def seek(self, at, whence=0):
        if whence == 0:
            self.offset = at
            return self.offset
        elif whence == 1:
            self.offset += at
            return self.offset
        else:
            raise ValueError("seeks only with whence 0 and 1")

    def seekable(self):
        return True

    def tell(self):
        return self.offset

    def tellable(self):
        return True

    def cachedfrom(self):
        return self.moved

    def cachedto(self):
        return self.moved + len(self.rwbuffer)

Answer 1

1. There's no docstring for the class. What kind of thing does an object belonging to this class represent? The text from your post would make a good start.

2. There's no docstring for the __init__ method. What do I pass for substream? What does tailcutoff mean? Again, the text from your post would make a good start.

3. There are no docstrings for the cachedfrom and cachedto methods. What do these return?

4. The cachedfrom and cachedto methods do not take any arguments. Would it make sense for these to be properties instead of methods?

5. Attributes like substream and rwbuffer do not appear to be intended to be used outside of the class implementation. It is conventional to give these names starting with _. See PEP8:

   Use one leading underscore only for non-public methods and instance variables.

6. The idea seems to be to create an object that acts like Python's built-in buffered I/O objects. But it doesn't implement the whole interface — for example, there's no close method, no closed property, and so on.

   It would make sense to inherit from the abstract base class io.BufferedIOBase so that you get implementations of all the methods (even if some of them are only stubs).

7. The number 128*1024 appears in two places without explanation. It should have a name, something like _chunk_size, and it might make sense for it to be a keyword argument to the constructor.

8. In the seek method I would use the constants io.SEEK_SET and io.SEEK_CUR instead of the numbers 0 and 1. That's because the named constants make the intention clear.

9. BufferedIOBase.read has this behaviour:

   If the argument is omitted, None, or negative, data is read and returned until EOF is reached.

   This behaviour is not implemented by RebufferedBytesIO.read. This kind of incompatibility might prevent someone from being able to pass a RebufferedBytesIO object to a function that was expecting an io.BufferedIOBase object.

10. RebufferedBytesIO.read raises an exception on a short read:

    if len(data) < count:
        raise IOError("could not read enough bytes, something went wrong")
    

    But for BufferedIOBase streams, short reads are possible.

11. The code for reading more chunks from the substream until the target position is reached is duplicated between the read and write methods. This should be refactored into its own method.

12. The code for doing the tail cut-off is duplicated between the read and write methods. This should be refactored into its own method.

13. Cutting off the tail using:

     self.rwbuffer = self.rwbuffer[removed:]
    

    is inefficient. In the worst case this code ends up copying the whole of rwbuffer on each read or write, which would be disastrous for performance.

14. The logic for reading a single chunk from the underlying stream looks like this:

    newdata = self.substream.read(128*1024)
    self.rwbuffer += newdata
    

    but the documentation for RawIOBase.read says:

    If the object is in non-blocking mode and no bytes are available, None is returned.

    If this happens, the += will fail with TypeError: can't concat bytes to NoneType.

15. The logic for reading chunks from the underlying stream looks like this:

    while self.moved + len(self.rwbuffer) < endsat:
        newdata = self.substream.read(128*1024)
        self.rwbuffer += newdata
        if not newdata:
            sleep(0)
    

    If the underlying stream is blocking, the sleep(0) is not necessary (because the read blocks if there is nothing to return. But if the underlying stream is non-blocking, then:

    A BlockingIOError is raised if the underlying raw stream is in non blocking-mode, and has no data available at the moment.

    So in neither the blocking case nor the non-blocking case is there any obvious reason to have a sleep here.