使用打字和守护线程时的错误

问题描述 投票:0回答:1

当我在 Python 3.11 上运行此代码时,它运行良好,但如果我在

typing_minimal
中导入
typing
(或
reader.py
),则会收到此错误:

Fatal Python error: _enter_buffered_busy: could not acquire lock for <_io.BufferedReader name='<stdin>'> at interpreter shutdown, possibly due to daemon threads
Python runtime state: finalizing (tstate=0x00007fff13ab6960)

Current thread 0x00003944 (most recent call first):
  <no Python frame>

main.py:

import reader
from typing_minimal import Generic, TypeVar


class BaseClass(Generic[TypeVar("VALUE")]):
    def __init__(self):
        pass


class SubClass(BaseClass[None]):
    pass


reader.start_reading()

reader.py:

from sys import stdin
from threading import Thread

# BUG: I get an error when I uncomment this import:
# import typing_minimal


def read():
    stdin.buffer.read()


def start_reading():
    Thread(target=read, daemon=True).start()

typing_minimal.py:

from functools import lru_cache, wraps


def _tp_cache(func):
    cached = lru_cache()(func)

    @wraps(func)
    def inner(*args, **kwds):
        return cached(*args, **kwds)

    return inner


class TypeVar:
    def __init__(self, _):
        pass


class _GenericAlias:
    def __init__(self, origin):
        self.origin = origin

    def __eq__(self, other):
        if isinstance(other, _GenericAlias):
            return self.origin is other.origin

        return NotImplemented

    def __hash__(self):
        return hash(self.origin)

    def __mro_entries__(self, _):
        return (self.origin,)


class Generic:
    @_tp_cache
    def __class_getitem__(cls, *_):
        return _GenericAlias(cls)

有人知道这是怎么回事吗?我很困惑。

python multithreading
1个回答
0
投票

在 main.py 中更改

class SubClass(BaseClass[None]):

class SubClass(BaseClass):

事情似乎有效。这似乎是某种竞争条件,但如果没有解释为什么在 main.py 中需要这个非法代码,我认为最好解决这个竞争条件的触发因素,然后解释底层机制。

© www.soinside.com 2019 - 2024. All rights reserved.