Data races in CJK multibyte codec state under free-threading

[maintain0404]

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