How to run the code under a contextmanager in a separate thread?

Question

I wanted to have a context manager where i can put some code which to be executed in a separate thread.

So far i couldn't find a way to achieve what i want, best alternative is writing closures and executing closures in a separate thread.

I want something like this

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

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

edit: let me try to ask my question again

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

I want yield to execute in a new thread. Basically whatever i put under contextmanager should be executed in a separate thread.

Remediless

What have you tried so far for the multithreading? There's really not enough information here for us to help. – Trehala

Edited the question the clarify further. My question is not about how to achieve multithreading, it is about how to use contextmanager in a specific way in which my thread should execute the code under contextmanager. – Remediless

I don't think this works. The only halfway idea I even have is having your context manager's __enter__ return a Thread instance and then overriding it's _target attribute. That still requires your method to do something - you don't just get the thread functionality without changes to the code. – Jocose

I'm commenting in the hopes this question gets some new attention, as I have the same requirement. Potential ideas I have had: duplicate a call stack for a new thread (so that the same code will get called twice) and then redirect the threads afterwards. OR capture the code within the context and bundle into a function somehow. I hope someone else has some ideas. – Switchback

Answer 1

Although similar to the question Is it possible to access the context object (code block) inside the __exit__() method of a context manager? in terms of identifying the code of a with block, this question differs in that the code in the context cannot be executed directly because you want it executed in a separate thread, so you will need a way to prevent the execution of the with block after the __enter__ method of the context manager returns.

One approach to circumventing the execution of the body of the with block by raising an exception. But raising an exception in the __enter__ method would result in an outright exception outside of the context manager without calling the __exit__ method, where we want to start a thread. So instead we can raise an exception after the __enter__ method returns by doing it in a trace function set for the sys.settrace and for the caller's frame:

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

so that:

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()

outputs:

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

Demo: https://replit.com/@blhsing1/DetailedDarlingSoftwareagent

Pneumoconiosis

Black magic answer right there! Nobody should probably use this in production, but I love the result! Perhaps one for python-ideas? – Georgenegeorges