[pypy-commit] pypy stdlib-2.7.3: CPython Issue #12213: Fix a buffering bug with interleaved reads and writes that

[amauryfa]

Author: Amaury Forgeot d'Arc <amauryfa at gmail.com>
Branch: stdlib-2.7.3
Changeset: r55689:845d29334d81
Date: 2012-06-15 22:33 +0200
http://bitbucket.org/pypy/pypy/changeset/845d29334d81/

Log:	CPython Issue #12213: Fix a buffering bug with interleaved reads and
	writes that could appear on io.BufferedRandom streams.

diff --git a/pypy/module/_io/interp_bufferedio.py b/pypy/module/_io/interp_bufferedio.py
--- a/pypy/module/_io/interp_bufferedio.py
+++ b/pypy/module/_io/interp_bufferedio.py
@@ -391,13 +391,7 @@
         self._check_init(space)
         with self.lock:
             if self.writable:
-                self._writer_flush_unlocked(space)
-            if self.readable:
-                if space.is_w(w_size, space.w_None):
-                    # Rewind the raw stream so that its position corresponds
-                    # to the current logical position
-                    self._raw_seek(space, -self._raw_offset(), 1)
-                self._reader_reset_buf()
+                self._flush_and_rewind_unlocked(space)
             # invalidate cached position
             self.abs_pos = -1
 
@@ -430,7 +424,7 @@
         self._check_init(space)
         with self.lock:
             if self.writable:
-                self._writer_flush_unlocked(space)
+                self._flush_and_rewind_unlocked(space)
             # Constraints:
             # 1. we don't want to advance the file position.
             # 2. we don't want to lose block alignment, so we can't shift the
@@ -464,9 +458,6 @@
             return space.wrap("")
 
         with self.lock:
-            if self.writable:
-                self._writer_flush_unlocked(space)
-
             # Return up to n bytes.  If at least one byte is buffered, we only
             # return buffered bytes.  Otherwise, we do one raw read.
 
@@ -477,6 +468,9 @@
 
             have = self._readahead()
             if have == 0:
+                if self.writable:
+                    self._flush_and_rewind_unlocked(space)
+
                 # Fill the buffer from the raw stream
                 self._reader_reset_buf()
                 self.pos = 0
@@ -493,6 +487,7 @@
 
     def _read_all(self, space):
         "Read all the file, don't update the cache"
+        # Must run with the lock held!
         builder = StringBuilder()
         # First copy what we have in the current buffer
         current_size = self._readahead()
@@ -500,10 +495,11 @@
         if current_size:
             data = ''.join(self.buffer[self.pos:self.pos + current_size])
             builder.append(data)
-        self._reader_reset_buf()
+            self.pos += current_size
         # We're going past the buffer's bounds, flush it
         if self.writable:
-            self._writer_flush_unlocked(space)
+            self._flush_and_rewind_unlocked(space)
+        self._reader_reset_buf()
 
         while True:
             # Read until EOF or until read() would block
@@ -559,6 +555,7 @@
     def _read_generic(self, space, n):
         """Generic read function: read from the stream until enough bytes are
            read, or until an EOF occurs or until read() would block."""
+        # Must run with the lock held!
         current_size = self._readahead()
         if n <= current_size:
             return self._read_fast(n)
@@ -572,13 +569,13 @@
                 result_buffer[written + i] = self.buffer[self.pos + i]
             remaining -= current_size
             written += current_size
+            self.pos += current_size
+
+        # Flush the write buffer if necessary
+        if self.writable:
+            self._writer_flush_unlocked(space)
         self._reader_reset_buf()
 
-        # XXX potential bug in CPython? The following is not enabled.
-        # We're going past the buffer's bounds, flush it
-        ## if self.writable:
-        ##     self._writer_flush_unlocked(space)
-
         # Read whole blocks, and don't buffer them
         while remaining > 0:
             r = self.buffer_size * (remaining // self.buffer_size)
@@ -755,13 +752,22 @@
         self._check_init(space)
         self._check_closed(space, "flush of closed file")
         with self.lock:
-            self._writer_flush_unlocked(space)
+            self._flush_and_rewind_unlocked(space)
             if self.readable:
                 # Rewind the raw stream so that its position corresponds to
                 # the current logical position.
                 self._raw_seek(space, -self._raw_offset(), 1)
                 self._reader_reset_buf()
 
+    def _flush_and_rewind_unlocked(self, space):
+        self._writer_flush_unlocked(space)
+        if self.readable:
+            # Rewind the raw stream so that its position corresponds to
+            # the current logical position.
+            try:
+                self._raw_seek(space, -self._raw_offset(), 1)
+            finally:
+                self._reader_reset_buf()
 
 class W_BufferedReader(BufferedMixin, W_BufferedIOBase):
     @unwrap_spec(buffer_size=int)
diff --git a/pypy/module/_io/test/test_bufferedio.py b/pypy/module/_io/test/test_bufferedio.py
--- a/pypy/module/_io/test/test_bufferedio.py
+++ b/pypy/module/_io/test/test_bufferedio.py
@@ -380,8 +380,8 @@
         self.test_nonblock_pipe_write(1024)
 
     def w_test_nonblock_pipe_write(self, bufsize):
-        import io
-        class NonBlockingPipe(io.BufferedIOBase):
+        import _io as io
+        class NonBlockingPipe(io._BufferedIOBase):
             "write() returns None when buffer is full"
             def __init__(self, buffersize=4096):
                 self.buffersize = buffersize
@@ -583,6 +583,44 @@
                 expected[i] = 1
                 assert raw.getvalue() == str(expected)
         
+    def test_interleaved_read_write(self):
+        import _io as io
+        # Test for issue #12213
+        with io.BytesIO(b'abcdefgh') as raw:
+            with io.BufferedRandom(raw, 100) as f:
+                f.write(b"1")
+                assert f.read(1) == b'b'
+                f.write(b'2')
+                assert f.read1(1) == b'd'
+                f.write(b'3')
+                buf = bytearray(1)
+                f.readinto(buf)
+                assert buf ==  b'f'
+                f.write(b'4')
+                assert f.peek(1) == b'h'
+                f.flush()
+                assert raw.getvalue() == b'1b2d3f4h'
+
+        with io.BytesIO(b'abc') as raw:
+            with io.BufferedRandom(raw, 100) as f:
+                assert f.read(1) == b'a'
+                f.write(b"2")
+                assert f.read(1) == b'c'
+                f.flush()
+                assert raw.getvalue() == b'a2c'
+
+    def test_interleaved_readline_write(self):
+        import _io as io
+        with io.BytesIO(b'ab\ncdef\ng\n') as raw:
+            with io.BufferedRandom(raw) as f:
+                f.write(b'1')
+                assert f.readline() == b'b\n'
+                f.write(b'2')
+                assert f.readline() == b'def\n'
+                f.write(b'3')
+                assert f.readline() == b'\n'
+                f.flush()
+                assert raw.getvalue() == b'1b\n2def\n3\n'
 
 class TestNonReentrantLock:
     def test_trylock(self):