python / python/cpython

Data races in CJK multibyte codec state under free-threading

Ouverte
#156,169 0 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Personne n'a encore pris cette issue.

extension-modules topic-free-threading type-bug
Langage dominant
Python
Étoiles
77.2k
Forks
35.9k
Métriques de PR en attente

Description

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

Guide de contribution

Ouvrir le guide de contribution

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Piste de recherche

Commencez par exécuter les reproducers fournis pour .errors, incremental decoder/encoder, StreamReader et StreamWriter sur un build CPython free-threaded. Suivez les accès listés à pending, pendingsize, state.c et errors dans ces points d’entrée de codec ; le travail est terminé lorsque les races sont éliminées pour les opérations signalées sans régression.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Évaluation

Stack technique
c, python
Domaine
operating-systems
Type d'issue
Bug
Difficulté
5/5
Temps estimé
Plus d'une semaine
Activité
Active
Clarté
Plutôt claire
Accessibilité débutants
35/100

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.