[Python-checkins] r65686 - in python/trunk: Lib/io.py Lib/test/test_io.py Misc/NEWS
antoine.pitrou
python-checkins at python.org
Thu Aug 14 23:04:31 CEST 2008
Author: antoine.pitrou
Date: Thu Aug 14 23:04:30 2008
New Revision: 65686
Log:
Issue #3476: make BufferedReader and BufferedWriter thread-safe
Modified:
python/trunk/Lib/io.py
python/trunk/Lib/test/test_io.py
python/trunk/Misc/NEWS
Modified: python/trunk/Lib/io.py
==============================================================================
--- python/trunk/Lib/io.py (original)
+++ python/trunk/Lib/io.py Thu Aug 14 23:04:30 2008
@@ -63,6 +63,7 @@
import codecs
import _fileio
import warnings
+import threading
# open() uses st_blksize whenever we can
DEFAULT_BUFFER_SIZE = 8 * 1024 # bytes
@@ -908,6 +909,7 @@
_BufferedIOMixin.__init__(self, raw)
self.buffer_size = buffer_size
self._reset_read_buf()
+ self._read_lock = threading.Lock()
def _reset_read_buf(self):
self._read_buf = b""
@@ -921,6 +923,10 @@
mode. If n is negative, read until EOF or until read() would
block.
"""
+ with self._read_lock:
+ return self._read_unlocked(n)
+
+ def _read_unlocked(self, n=None):
nodata_val = b""
empty_values = (b"", None)
buf = self._read_buf
@@ -973,6 +979,10 @@
do at most one raw read to satisfy it. We never return more
than self.buffer_size.
"""
+ with self._read_lock:
+ return self._peek_unlocked(n)
+
+ def _peek_unlocked(self, n=0):
want = min(n, self.buffer_size)
have = len(self._read_buf) - self._read_pos
if have < want:
@@ -989,18 +999,21 @@
# only return buffered bytes. Otherwise, we do one raw read.
if n <= 0:
return b""
- self.peek(1)
- return self.read(min(n, len(self._read_buf) - self._read_pos))
+ with self._read_lock:
+ self._peek_unlocked(1)
+ return self._read_unlocked(
+ min(n, len(self._read_buf) - self._read_pos))
def tell(self):
return self.raw.tell() - len(self._read_buf) + self._read_pos
def seek(self, pos, whence=0):
- if whence == 1:
- pos -= len(self._read_buf) - self._read_pos
- pos = self.raw.seek(pos, whence)
- self._reset_read_buf()
- return pos
+ with self._read_lock:
+ if whence == 1:
+ pos -= len(self._read_buf) - self._read_pos
+ pos = self.raw.seek(pos, whence)
+ self._reset_read_buf()
+ return pos
class BufferedWriter(_BufferedIOMixin):
@@ -1022,43 +1035,51 @@
if max_buffer_size is None
else max_buffer_size)
self._write_buf = bytearray()
+ self._write_lock = threading.Lock()
def write(self, b):
if self.closed:
raise ValueError("write to closed file")
if isinstance(b, unicode):
raise TypeError("can't write unicode to binary stream")
- # XXX we can implement some more tricks to try and avoid partial writes
- if len(self._write_buf) > self.buffer_size:
- # We're full, so let's pre-flush the buffer
- try:
- self.flush()
- except BlockingIOError as e:
- # We can't accept anything else.
- # XXX Why not just let the exception pass through?
- raise BlockingIOError(e.errno, e.strerror, 0)
- before = len(self._write_buf)
- self._write_buf.extend(b)
- written = len(self._write_buf) - before
- if len(self._write_buf) > self.buffer_size:
- try:
- self.flush()
- except BlockingIOError as e:
- if (len(self._write_buf) > self.max_buffer_size):
- # We've hit max_buffer_size. We have to accept a partial
- # write and cut back our buffer.
- overage = len(self._write_buf) - self.max_buffer_size
- self._write_buf = self._write_buf[:self.max_buffer_size]
- raise BlockingIOError(e.errno, e.strerror, overage)
- return written
+ with self._write_lock:
+ # XXX we can implement some more tricks to try and avoid
+ # partial writes
+ if len(self._write_buf) > self.buffer_size:
+ # We're full, so let's pre-flush the buffer
+ try:
+ self._flush_unlocked()
+ except BlockingIOError as e:
+ # We can't accept anything else.
+ # XXX Why not just let the exception pass through?
+ raise BlockingIOError(e.errno, e.strerror, 0)
+ before = len(self._write_buf)
+ self._write_buf.extend(b)
+ written = len(self._write_buf) - before
+ if len(self._write_buf) > self.buffer_size:
+ try:
+ self._flush_unlocked()
+ except BlockingIOError as e:
+ if len(self._write_buf) > self.max_buffer_size:
+ # We've hit max_buffer_size. We have to accept a
+ # partial write and cut back our buffer.
+ overage = len(self._write_buf) - self.max_buffer_size
+ self._write_buf = self._write_buf[:self.max_buffer_size]
+ raise BlockingIOError(e.errno, e.strerror, overage)
+ return written
def truncate(self, pos=None):
- self.flush()
- if pos is None:
- pos = self.raw.tell()
- return self.raw.truncate(pos)
+ with self._write_lock:
+ self._flush_unlocked()
+ if pos is None:
+ pos = self.raw.tell()
+ return self.raw.truncate(pos)
def flush(self):
+ with self._write_lock:
+ self._flush_unlocked()
+
+ def _flush_unlocked(self):
if self.closed:
raise ValueError("flush of closed file")
written = 0
@@ -1077,8 +1098,9 @@
return self.raw.tell() + len(self._write_buf)
def seek(self, pos, whence=0):
- self.flush()
- return self.raw.seek(pos, whence)
+ with self._write_lock:
+ self._flush_unlocked()
+ return self.raw.seek(pos, whence)
class BufferedRWPair(BufferedIOBase):
@@ -1168,7 +1190,8 @@
# First do the raw seek, then empty the read buffer, so that
# if the raw seek fails, we don't lose buffered data forever.
pos = self.raw.seek(pos, whence)
- self._reset_read_buf()
+ with self._read_lock:
+ self._reset_read_buf()
return pos
def tell(self):
@@ -1205,8 +1228,9 @@
def write(self, b):
if self._read_buf:
# Undo readahead
- self.raw.seek(self._read_pos - len(self._read_buf), 1)
- self._reset_read_buf()
+ with self._read_lock:
+ self.raw.seek(self._read_pos - len(self._read_buf), 1)
+ self._reset_read_buf()
return BufferedWriter.write(self, b)
Modified: python/trunk/Lib/test/test_io.py
==============================================================================
--- python/trunk/Lib/test/test_io.py (original)
+++ python/trunk/Lib/test/test_io.py Thu Aug 14 23:04:30 2008
@@ -6,8 +6,10 @@
import sys
import time
import array
+import threading
+import random
import unittest
-from itertools import chain
+from itertools import chain, cycle
from test import test_support
import codecs
@@ -390,6 +392,49 @@
# this test. Else, write it.
pass
+ def testThreads(self):
+ try:
+ # Write out many bytes with exactly the same number of 0's,
+ # 1's... 255's. This will help us check that concurrent reading
+ # doesn't duplicate or forget contents.
+ N = 1000
+ l = range(256) * N
+ random.shuffle(l)
+ s = bytes(bytearray(l))
+ with io.open(test_support.TESTFN, "wb") as f:
+ f.write(s)
+ with io.open(test_support.TESTFN, "rb", buffering=0) as raw:
+ bufio = io.BufferedReader(raw, 8)
+ errors = []
+ results = []
+ def f():
+ try:
+ # Intra-buffer read then buffer-flushing read
+ for n in cycle([1, 19]):
+ s = bufio.read(n)
+ if not s:
+ break
+ # list.append() is atomic
+ results.append(s)
+ except Exception as e:
+ errors.append(e)
+ raise
+ threads = [threading.Thread(target=f) for x in range(20)]
+ for t in threads:
+ t.start()
+ time.sleep(0.02) # yield
+ for t in threads:
+ t.join()
+ self.assertFalse(errors,
+ "the following exceptions were caught: %r" % errors)
+ s = b''.join(results)
+ for i in range(256):
+ c = bytes(bytearray([i]))
+ self.assertEqual(s.count(c), N)
+ finally:
+ test_support.unlink(test_support.TESTFN)
+
+
class BufferedWriterTest(unittest.TestCase):
@@ -446,6 +491,38 @@
self.assertEquals(b"abc", writer._write_stack[0])
+ def testThreads(self):
+ # BufferedWriter should not raise exceptions or crash
+ # when called from multiple threads.
+ try:
+ # We use a real file object because it allows us to
+ # exercise situations where the GIL is released before
+ # writing the buffer to the raw streams. This is in addition
+ # to concurrency issues due to switching threads in the middle
+ # of Python code.
+ with io.open(test_support.TESTFN, "wb", buffering=0) as raw:
+ bufio = io.BufferedWriter(raw, 8)
+ errors = []
+ def f():
+ try:
+ # Write enough bytes to flush the buffer
+ s = b"a" * 19
+ for i in range(50):
+ bufio.write(s)
+ except Exception as e:
+ errors.append(e)
+ raise
+ threads = [threading.Thread(target=f) for x in range(20)]
+ for t in threads:
+ t.start()
+ time.sleep(0.02) # yield
+ for t in threads:
+ t.join()
+ self.assertFalse(errors,
+ "the following exceptions were caught: %r" % errors)
+ finally:
+ test_support.unlink(test_support.TESTFN)
+
class BufferedRWPairTest(unittest.TestCase):
Modified: python/trunk/Misc/NEWS
==============================================================================
--- python/trunk/Misc/NEWS (original)
+++ python/trunk/Misc/NEWS Thu Aug 14 23:04:30 2008
@@ -48,6 +48,9 @@
Library
-------
+- Issue #3476: binary buffered reading through the new "io" library is now
+ thread-safe.
+
- Silence the DeprecationWarning of rfc822 when it is imported by mimetools
since mimetools itself is deprecated. Because modules are cached, all
subsequent imports of rfc822 will not raise a visible DeprecationWarning.
More information about the Python-checkins
mailing list