如何在单独的线程中的上下文管理器下运行代码？

Question

我想要一个上下文管理器，我可以在其中放置一些要在单独的线程中执行的代码。

到目前为止，我找不到一种方法来实现我想要的，最好的选择是编写闭包并在单独的线程中执行闭包。

我想要这样的东西

# code runs on main thread
print("this is main thread")

with amazingcontextmanager:
    # code to run in separate thread
    print("this is not main thread")

编辑：让我尝试再次问我的问题

@contextlib.contextmanager
def amazingcontextmanager():
    try:
        yield
    finally:
        print("thread done")

我想yield在新线程中执行。基本上我放在 contextmanager 下的任何内容都应该在单独的线程中执行。

Answer 1

尽管与问题类似Is it possible to access the context object (code block) inside the __exit__()method of a context manager? 在识别块的代码方面with，这个问题的不同之处在于上下文中的代码不能直接执行，因为你希望它在单独的线程中执行，所以你需要一种方法来阻止方法with之后的块的执行__enter__上下文管理器返回。

with一种通过引发异常来规避块体执行的方法。但是在方法中引发异常__enter__将导致上下文管理器之外发生彻底的异常__exit__，而无需调用我们想要启动线程的方法。因此，我们可以在方法返回后引发异常，方法是在为调用者框架__enter__设置的跟踪函数中执行此操作：sys.settrace

import sys
import threading
from linecache import getline
from tokenize import tokenize, INDENT, DEDENT

class thread_context:
    class EndContext(Exception):
        pass

    def _skip_execution(self, frame, event, arg):
        raise self.EndContext

    def __enter__(self):
        def readline():
            lineno = caller.f_lineno
            while line := getline(filename, lineno):
                if lineno == caller.f_lineno:  # dedent the with statement
                    line = line.lstrip()       # so it can be parsed alone
                yield line.encode()
                lineno += 1
            yield b''

        caller = sys._getframe(1)
        filename = caller.f_code.co_filename
        first = end = depth = 0
        try:
            for token, _, (start, _), (end, _), _ in tokenize(readline().__next__):
                if token == INDENT:
                    depth += 1
                    if not first:
                        first = start
                elif token == DEDENT:
                    if depth == 1:
                        break
                    depth -= 1
        except IndentationError:
            end += 1
        body = ''.join(
            getline(filename, caller.f_lineno + lineno - 1)
            for lineno in range(first, end)
        )
        self.namespace = {}
        self.thread = threading.Thread(
            target=exec,
            args=(
                compile('if 1:\n' + body, '\n' + body, 'exec'),
                caller.f_globals,
                self.namespace
            )
        )
        self.tracer = sys.gettrace()
        caller.f_trace = self._skip_execution
        sys.settrace(self._skip_execution)
        return self.thread

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is self.EndContext:
            caller = sys._getframe(1)
            caller.f_trace = self.tracer
            sys.settrace(self.tracer) # restore the original trace function
            self.namespace.update(caller.f_locals)
            self.thread.start()
            return True

以便：

from time import sleep

def main():
    foo = []
    with thread_context() as thread:
        for _ in range(3):
            sleep(.9)
            print(f'sleeping in {thread}')
        foo.append(1)
    while not foo:
        print('foo is empty')
        sleep(1)
    print('foo got', foo.pop())
    thread.join()

main()

输出：

foo is empty
sleeping in <Thread(Thread-1 (exec), started 139934645712576)>
foo is empty
sleeping in <Thread(Thread-1 (exec), started 139934645712576)>
foo is empty
sleeping in <Thread(Thread-1 (exec), started 139934645712576)>
foo got 1

演示： https: //replit.com/@blhsing1/DetailedDarlingSoftwareagent