python / python/cpython

Data races in CJK multibyte codec state under free-threading

Đang mở
#156,169 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Chưa có ai nhận issue này.

extension-modules topic-free-threading type-bug
Ngôn ngữ chính
Python
Star
77.2k
Fork
35.9k
Chỉ số pull request đang chờ

Mô tả

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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Hướng nghiên cứu

Bắt đầu bằng cách chạy các reproducer được cung cấp cho .errors, incremental decoder/encoder, StreamReader và StreamWriter trên một bản build CPython free-threaded. Theo dõi các lượt truy cập pending, pendingsize, state.c và errors được liệt kê trong các điểm vào codec đó; công việc được xem là hoàn tất khi các race được loại bỏ trên các thao tác đã báo cáo mà không gây hồi quy.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Đánh giá

Công nghệ
c, python
Lĩnh vực
operating-systems
Loại issue
Lỗi
Độ khó
5/5
Thời gian dự kiến
Hơn một tuần
Mức độ hoạt động
Sôi nổi
Độ rõ ràng
Khá rõ ràng
Mức phù hợp với người mới
35/100

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.