python / python/cpython

Data races in CJK multibyte codec state under free-threading

Aperta
#156,169 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

Nessuno ha ancora preso questa issue.

extension-modules topic-free-threading type-bug
Lingua principale
Python
Stelle
77.2k
Fork
35.9k
Metriche PR in attesa

Descrizione

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 https://github.com/python/cpython/issues/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

Guida per i contributori

Apri la guida per i contributori

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Direzione di ricerca

Inizia eseguendo i reproducers forniti per .errors, incremental decoder/encoder, StreamReader e StreamWriter su una build di CPython free-threaded. Traccia gli accessi elencati a pending, pendingsize, state.c ed errors in questi punti di ingresso dei codec; il lavoro è completato quando le race sono eliminate nelle operazioni segnalate senza regressioni.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Valutazione

Stack tecnologico
c, python
Ambito
operating-systems
Tipo di issue
Bug
Difficoltà
5/5
Tempo stimato
Più di una settimana
Stato di attività
Attiva
Chiarezza
Abbastanza chiara
Idoneità per principianti
35/100

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.