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.

Đánh giá

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

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.

Mô tả

extension-modules topic-free-threading type-bug

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

Ngôn ngữ chính
Python
Star
77.2k
Fork
36k
Merge trung bình
1 ngày 9 giờ
Pull request đã merge (30 ngày)
558

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.

Issue khác của python/cpython

Tất cả issue của python/cpython

Issue tương tự

Thêm issue về Python

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.