Bug report
Bug description:
Bug Report
Bug description
On a free-threaded CPython build, concurrent calls on the same CJK
multibyte codec object race on unsynchronized mutable C state. The decoder and stream reader use pending, pendingsize, and
state.c; the encoder and stream writer use pending and state.c.
The .errors getter/setter additionally accesses self->errors
without synchronization.
This is a sub-issue of #153852 , gist
Reproducers
Cases and c struct fields has data race.
.errors getter <> setter of all codec classes - on self->errors
import sys, threading, codecs
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
ERROR_NAMES = ("tsan_error_handler_a", "tsan_error_handler_b")
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
dec = box[0]
for _ in range(6):
try:
dec.errors # reads self->errors; custom name takes Py_NewRef() path
except Exception:
pass
try:
dec.errors = ERROR_NAMES[wid & 1] # replaces and decrefs self->errors
except Exception:
pass
leave.wait()
ts = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for t in ts:
t.start()
factory = codecs.getincrementaldecoder("euc_jp")
for r in range(ROUNDS):
dec = factory(ERROR_NAMES[r & 1])
box[0] = dec # publish the shared decoder for this round
enter.wait()
leave.wait()
for t in ts:
t.join()
print("done, no crash")
IncrementalDecoder.decode <> .reset — on pendingsize
import codecs, sys, threading
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
DECODER = codecs.getincrementaldecoder("euc_jp")
DECODER_STATE = (b"\xa4", 0)
def new_decoder():
decoder = DECODER()
decoder.decode(b"\xa4") # leaves one byte in self->pending
return decoder
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
dec = box[0]
try:
dec.decode(b"\xa4", final=False)
except Exception:
pass
try:
dec.reset()
except Exception:
pass
leave.wait()
threads = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for thread in threads:
thread.start()
for _ in range(ROUNDS):
box[0] = new_decoder()
enter.wait()
leave.wait()
for thread in threads:
thread.join()
print(f"done, no crash")
IncrementalDecoder.getstate <> .setstate - on pending, pendingsize, state.c
import codecs, sys, threading
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
DECODER = codecs.getincrementaldecoder("euc_jp")
DECODER_STATE = (b"\xa4", 0)
def new_decoder():
decoder = DECODER()
decoder.decode(b"\xa4") # leaves one byte in self->pending
return decoder
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
dec = box[0]
for _ in range(6):
try:
dec.getstate() # reads self->pending/self->pendingsize
except Exception:
pass
try:
state = DECODER_STATE if wid & 1 else (b"", 0)
dec.setstate(state)
except Exception:
pass
leave.wait()
threads = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for thread in threads:
thread.start()
for _ in range(ROUNDS):
box[0] = new_decoder()
enter.wait()
leave.wait()
for thread in threads:
thread.join()
print(f"done, no crash")
IncrementalEncoder.encode <> .reset - on pending
import codecs, sys, threading
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
ENCODER = codecs.getincrementalencoder("euc_jis_2004")
empty_state = ENCODER().getstate()
def new_encoder():
encoder = ENCODER()
encoder.encode("\u00e6") # leaves a Unicode string in self->pending
return encoder
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
enc = box[0]
for _ in range(6):
try:
enc.encode("\u0300")
except Exception:
pass
try:
enc.reset()
except Exception:
pass
leave.wait()
threads = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for thread in threads:
thread.start()
factory = ENCODER
for _ in range(ROUNDS):
box[0] = new_encoder()
enter.wait()
leave.wait()
for thread in threads:
thread.join()
print(f"done, no crash")
IncrementalEncoder.getstate <> .setstate - on pending
import codecs, sys, threading
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
ENCODER = codecs.getincrementalencoder("euc_jis_2004")
empty_state = ENCODER().getstate()
def new_encoder():
encoder = ENCODER()
encoder.encode("\u00e6") # leaves a Unicode string in self->pending
return encoder
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
enc = box[0]
for _ in range(6):
try:
enc.getstate()
except Exception:
pass
try:
enc.setstate(empty_state)
except Exception:
pass
leave.wait()
threads = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for thread in threads:
thread.start()
factory = ENCODER
for _ in range(ROUNDS):
box[0] = new_encoder()
enter.wait()
leave.wait()
for thread in threads:
thread.join()
print(f"done, no crash")
StreamReader.read/readline/readlines <> .reset - on pendingsize
import codecs
import sys
import threading
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
class InputStream:
def read(self, size=-1):
return b"\xa4"
def readline(self, size=-1):
return b"\xa4"
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
dec = box[0]
try:
# Because StreamReader.read, .readline, .readlines all use mbstreamreader_iread,
# the code below proves every thread-conflict case of StreamReader.
box[0].reset() if wid & 1 else box[0].read(1)
except Exception:
pass
leave.wait()
threads = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for thread in threads:
thread.start()
factory = codecs.getreader("euc_jp")
for _ in range(ROUNDS):
box[0] = factory(InputStream())
enter.wait()
leave.wait()
for thread in threads:
thread.join()
print(f"done, no crash")
StreamWriter.write/writelines <> .reset - on pending
import codecs, sys, threading
assert not sys._is_gil_enabled(), "run free-threaded: PYTHON_GIL=0"
NT = 8
ROUNDS = 4000
box = [None]
enter = threading.Barrier(NT + 1)
leave = threading.Barrier(NT + 1)
class OutputStream:
def write(self, data):
return len(data)
def worker(wid):
for _ in range(ROUNDS):
enter.wait()
# Because StreamWriter.write, .writelines, all use encoder_encode_stateful,
# the code below proves every thread-conflict case of StreamWriter.
for _ in range(6):
try:
box[0].write("\u00e6")
except Exception:
pass
try:
box[0].reset()
except Exception:
pass
leave.wait()
threads = [threading.Thread(target=worker, args=(i,)) for i in range(NT)]
for thread in threads:
thread.start()
factory = codecs.getwriter("euc_jis_2004")
for _ in range(ROUNDS):
box[0] = factory(OutputStream())
enter.wait()
leave.wait()
for thread in threads:
thread.join()
print(f"done, no crash")
CPython versions tested on:
CPython main branch
Operating systems tested on:
macOS
Bug report
Bug description:
Bug Report
Bug description
On a free-threaded CPython build, concurrent calls on the same CJK
multibyte codec object race on unsynchronized mutable C state. The decoder and stream reader use
pending,pendingsize, andstate.c; the encoder and stream writer usependingandstate.c.The
.errorsgetter/setter additionally accessesself->errorswithout synchronization.
This is a sub-issue of #153852 , gist
Reproducers
Cases and c struct fields has data race.
.errors getter <> setter of all codec classes - on self->errors
IncrementalDecoder.decode <> .reset — on pendingsize
IncrementalDecoder.getstate <> .setstate - on pending, pendingsize, state.c
IncrementalEncoder.encode <> .reset - on pending
IncrementalEncoder.getstate <> .setstate - on pending
StreamReader.read/readline/readlines <> .reset - on pendingsize
StreamWriter.write/writelines <> .reset - on pending
CPython versions tested on:
CPython main branch
Operating systems tested on:
macOS