Add a lock to keep self._buffer and self._pos in sync

This requires updating pickling to exclude the lock

Author: cmaloney

Lib/_pyio.py

@@ -876,16 +876,27 @@ class BytesIO(BufferedIOBase):
     _buffer = None
 
     def __init__(self, initial_bytes=None):
+        # Use to keep self._buffer and self._pos consistent.
+        self._lock = Lock()
+
         buf = bytearray()
         if initial_bytes is not None:
             buf += initial_bytes
-        self._buffer = buf
-        self._pos = 0
+
+        with self._lock:
+            self._buffer = buf
+            self._pos = 0
 
     def __getstate__(self):
         if self.closed:
             raise ValueError("__getstate__ on closed file")
-        return self.__dict__.copy()
+        state = self.__dict__.copy()
+        del state['_lock']
+        return state
+
+    def __setstate__(self, state):
+        self.__dict__.update(state)
+        self._lock = Lock()
 
     def getvalue(self):
         """Return the bytes value (contents) of the buffer
@@ -909,23 +920,25 @@ def close(self):
     def read(self, size=-1):
         if self.closed:
             raise ValueError("read from closed file")
-        if size is None:
-            size = -1
-        else:
-            try:
-                size_index = size.__index__
-            except AttributeError:
-                raise TypeError(f"{size!r} is not an integer")
+
+        with self._lock:
+            if size is None:
+                size = -1
             else:
-                size = size_index()
-        if size < 0:
-            size = len(self._buffer)
-        if len(self._buffer) <= self._pos:
-            return b""
-        newpos = min(len(self._buffer), self._pos + size)
-        b = self._buffer[self._pos : newpos]
-        self._pos = newpos
-        return bytes(b)
+                try:
+                    size_index = size.__index__
+                except AttributeError:
+                    raise TypeError(f"{size!r} is not an integer")
+                else:
+                    size = size_index()
+            if size < 0:
+                size = len(self._buffer)
+            if len(self._buffer) <= self._pos:
+                return b""
+            newpos = min(len(self._buffer), self._pos + size)
+            b = self._buffer[self._pos : newpos]
+            self._pos = newpos
+            return bytes(b)
 
     def read1(self, size=-1):
         """This is the same as read.
@@ -941,12 +954,14 @@ def write(self, b):
             n = view.nbytes  # Size of any bytes-like object
         if n == 0:
             return 0
-        pos = self._pos
-        if pos > len(self._buffer):
-            # Pad buffer to pos with null bytes.
-            self._buffer.resize(pos)
-        self._buffer[pos:pos + n] = b
-        self._pos += n
+
+        with self._lock:
+            pos = self._pos
+            if pos > len(self._buffer):
+                # Pad buffer to pos with null bytes.
+                self._buffer.resize(pos)
+            self._buffer[pos:pos + n] = b
+            self._pos += n
         return n
 
     def seek(self, pos, whence=0):
@@ -978,18 +993,20 @@ def tell(self):
     def truncate(self, pos=None):
         if self.closed:
             raise ValueError("truncate on closed file")
-        if pos is None:
-            pos = self._pos
-        else:
-            try:
-                pos_index = pos.__index__
-            except AttributeError:
-                raise TypeError(f"{pos!r} is not an integer")
+
+        with self._lock:
+            if pos is None:
+                pos = self._pos
             else:
-                pos = pos_index()
-            if pos < 0:
-                raise ValueError("negative truncate position %r" % (pos,))
-        del self._buffer[pos:]
+                try:
+                    pos_index = pos.__index__
+                except AttributeError:
+                    raise TypeError(f"{pos!r} is not an integer")
+                else:
+                    pos = pos_index()
+                if pos < 0:
+                    raise ValueError("negative truncate position %r" % (pos,))
+            del self._buffer[pos:]
         return pos
 
     def readable(self):

Lib/test/test_free_threading/test_io.py

@@ -104,9 +104,8 @@ def sizeof(barrier, b, *ignore):
         self.check([truncate] + [getbuffer] * 10, self.ioclass(b'0\n'*204800))
         self.check([truncate] + [iter] * 10, self.ioclass(b'0\n'*20480))
         self.check([truncate] + [getstate] * 10, self.ioclass(b'0\n'*204800))
-        # _pyio uses default __setstate__
-        if hasattr(self.ioclass, '__setstate__'):
-            self.check([truncate] + [setstate] * 10, self.ioclass(b'0\n'*204800), (b'123', 0, None))
+        state = self.ioclass(b'123').__getstate__()
+        self.check([truncate] + [setstate] * 10, self.ioclass(b'0\n'*204800), state)
         self.check([truncate] + [sizeof] * 10, self.ioclass(b'0\n'*204800))
 
         # no tests for seek or tell because they don't break anything