WIP Back to threads by thomasballinger · Pull Request #927 · bpython/bpython
@@ -1,17 +1,14 @@
"""For running Python code that could interrupt itself at any time in order to,
for example, ask for a read on stdin, or a write on stdout
for example, ask for a read on stdin, or a write on stdout.
The CodeRunner spawns a greenlet to run code in, and that code can suspend its own execution to ask the main greenlet to refresh the display or get The CodeRunner spawns a thread to run code in. That code can block on a queue to ask the main (UI) thread to refresh the display or get information.
Greenlets are basically threads that can explicitly switch control to each other. You can replace the word "greenlet" with "thread" in these docs if that makes more sense to you. """
import code import greenlet from six.moves import queue import threading import logging import signal
Expand All
@@ -21,12 +18,12 @@
class SigintHappened: """If this class is returned, a SIGINT happened while the main greenlet""" """If this class is returned, a SIGINT happened while the main thread"""
class SystemExitFromCodeRunner(SystemExit): """If this class is returned, a SystemExit happened while in the code greenlet""" thread"""
class RequestFromCodeRunner:Expand Down
Expand Up
@@ -57,30 +54,38 @@ def __init__(self, args):
class CodeRunner: """Runs user code in an interpreter.
Running code requests a refresh by calling request_from_main_context(force_refresh=True), which suspends execution of the code and switches back to the main greenlet
After load_code() is called with the source code to be run, the run_code() method should be called to start running the code. The running code may request screen refreshes and user input by calling request_from_main_context. When this are called, the running source code cedes control, and the current run_code() method call returns.
The return value of run_code() determines whether the method ought to be called again to complete execution of the source code. """Runs user code in a pausable thread.
>>> cr = CodeRunner() >>> def get_input(): ... print('waiting for a number plz') ... return cr.request_from_main_thread() ... >>> i = InteractiveInterpreter(locals={'get_input': get_input}) >>> cr.interp = i >>> cr.load_code('x = get_input(); print(x * 2)') >>> finished = cr.run_code() waiting for a number plz >>> # do something else, user code thread is paused >>> finished = cr.run_code(for_code=21) 42
As user code executes it can make requests for values or simply request that the screen be refreshed with `request_from_main_thread()`. This pauses the user code execution thread and wakes up the main thread, where run_code() returns whether user code has finished executing. This is cooperative multitasking: even though there are two threads, the main thread and the user code thread, the two threads work cede control to one another like like green threads with no parallelism.
Once the screen refresh has occurred or the requested user input has been gathered, run_code() should be called again, passing in any requested user input. This continues until run_code returns Done. requested user input. This continues until run_code returns True.
The code greenlet is responsible for telling the main greenlet The code thread is responsible for telling the main thread what it wants returned in the next run_code call - CodeRunner just passes whatever is passed in to run_code(for_code) to the code greenlet code thread. """
def __init__(self, interp=None, request_refresh=lambda: None):Expand All
@@ -93,65 +98,67 @@ def __init__(self, interp=None, request_refresh=lambda: None):
"""
self.interp = interp or code.InteractiveInterpreter()
self.source = None
self.main_context = greenlet.getcurrent()
self.code_context = None
self.code_thread = None
self.requests_from_code_thread = queue.Queue(maxsize=0)
self.responses_for_code_thread = queue.Queue()
self.request_refresh = request_refresh
# waiting for response from main thread
self.code_is_waiting = False
# sigint happened while in main thread
self.sigint_happened_in_main_context = False
self.sigint_happened_in_main_thread = False
self.orig_sigint_handler = None
@property def running(self): """Returns greenlet if code has been loaded greenlet has been started""" return self.source and self.code_context """Returns the running thread if code has been loaded and started.""" return self.source and self.code_thread
def load_code(self, source): """Prep code to be run""" assert self.source is None, ( "you shouldn't load code when some is " "already running" ) self.source = source self.code_context = None self.code_thread = None
def _unload_code(self): """Called when done running code""" self.source = None self.code_context = None self.code_thread = None self.code_is_waiting = False
def run_code(self, for_code=None): """Returns Truthy values if code finishes, False otherwise
if for_code is provided, send that value to the code greenlet if for_code is provided, send that value to the code thread if source code is complete, returns "done" if source code is incomplete, returns "unfinished" """ if self.code_context is None: if self.code_thread is None: assert self.source is not None self.code_context = greenlet.greenlet(self._blocking_run_code) self.code_thread = threading.Thread( target=self._blocking_run_code, name="codethread" ) self.code_thread.daemon = True if is_main_thread(): self.orig_sigint_handler = signal.getsignal(signal.SIGINT) signal.signal(signal.SIGINT, self.sigint_handler) request = self.code_context.switch() self.code_thread.start() else: assert self.code_is_waiting self.code_is_waiting = False if is_main_thread(): signal.signal(signal.SIGINT, self.sigint_handler) if self.sigint_happened_in_main_context: self.sigint_happened_in_main_context = False request = self.code_context.switch(SigintHappened) if self.sigint_happened_in_main_thread: self.sigint_happened_in_main_thread = False self.responses_for_code_thread.put(SigintHappened) else: request = self.code_context.switch(for_code) self.responses_for_code_thread.put(for_code)
request = self.requests_from_code_thread.get() logger.debug("request received from code was %r", request) if not isinstance(request, RequestFromCodeRunner): raise ValueError( "Not a valid value from code greenlet: %r" % request ) raise ValueError("Not a valid value from code thread: %r" % request) if isinstance(request, (Wait, Refresh)): self.code_is_waiting = True if isinstance(request, Refresh):Expand All
@@ -170,32 +177,37 @@ def run_code(self, for_code=None):
def sigint_handler(self, *args):
"""SIGINT handler to use while code is running or request being
fulfilled"""
if greenlet.getcurrent() is self.code_context:
if threading.current_thread() is self.code_thread:
logger.debug("sigint while running user code!")
raise KeyboardInterrupt()
else:
logger.debug(
"sigint while fulfilling code request sigint handler "
"running!"
)
self.sigint_happened_in_main_context = True
self.sigint_happened_in_main_thread = True
def _blocking_run_code(self): try: unfinished = self.interp.runsource(self.source) except SystemExit as e: return SystemExitRequest(*e.args) return Unfinished() if unfinished else Done() self.requests_from_code_thread.push(SystemExitRequest(*e.args)) return self.requests_from_code_thread.put( Unfinished() if unfinished else Done() )
def request_from_main_context(self, force_refresh=False): def request_from_main_thread(self, force_refresh=False): """Return the argument passed in to .run_code(for_code)
Nothing means calls to run_code must be... ??? """ if force_refresh: value = self.main_context.switch(Refresh()) self.requests_from_code_thread.put(Refresh()) value = self.responses_for_code_thread.get() else: value = self.main_context.switch(Wait()) self.requests_from_code_thread.put(Wait()) value = self.responses_for_code_thread.get() if value is SigintHappened: raise KeyboardInterrupt() return valueExpand All
@@ -216,7 +228,7 @@ def __init__(self, coderunner, on_write, real_fileobj):
def write(self, s, *args, **kwargs): self.on_write(s, *args, **kwargs) return self.coderunner.request_from_main_context(force_refresh=True) return self.coderunner.request_from_main_thread(force_refresh=True)
# Some applications which use curses require that sys.stdout # have a method called fileno. One example is pwntools. ThisExpand Down
The CodeRunner spawns a greenlet to run code in, and that code can suspend its own execution to ask the main greenlet to refresh the display or get The CodeRunner spawns a thread to run code in. That code can block on a queue to ask the main (UI) thread to refresh the display or get information.
Greenlets are basically threads that can explicitly switch control to each other. You can replace the word "greenlet" with "thread" in these docs if that makes more sense to you. """
import code import greenlet from six.moves import queue import threading import logging import signal
class SigintHappened: """If this class is returned, a SIGINT happened while the main greenlet""" """If this class is returned, a SIGINT happened while the main thread"""
class SystemExitFromCodeRunner(SystemExit): """If this class is returned, a SystemExit happened while in the code greenlet""" thread"""
class RequestFromCodeRunner:
class CodeRunner: """Runs user code in an interpreter.
Running code requests a refresh by calling request_from_main_context(force_refresh=True), which suspends execution of the code and switches back to the main greenlet
After load_code() is called with the source code to be run, the run_code() method should be called to start running the code. The running code may request screen refreshes and user input by calling request_from_main_context. When this are called, the running source code cedes control, and the current run_code() method call returns.
The return value of run_code() determines whether the method ought to be called again to complete execution of the source code. """Runs user code in a pausable thread.
>>> cr = CodeRunner() >>> def get_input(): ... print('waiting for a number plz') ... return cr.request_from_main_thread() ... >>> i = InteractiveInterpreter(locals={'get_input': get_input}) >>> cr.interp = i >>> cr.load_code('x = get_input(); print(x * 2)') >>> finished = cr.run_code() waiting for a number plz >>> # do something else, user code thread is paused >>> finished = cr.run_code(for_code=21) 42
As user code executes it can make requests for values or simply request that the screen be refreshed with `request_from_main_thread()`. This pauses the user code execution thread and wakes up the main thread, where run_code() returns whether user code has finished executing. This is cooperative multitasking: even though there are two threads, the main thread and the user code thread, the two threads work cede control to one another like like green threads with no parallelism.
Once the screen refresh has occurred or the requested user input has been gathered, run_code() should be called again, passing in any requested user input. This continues until run_code returns Done. requested user input. This continues until run_code returns True.
The code greenlet is responsible for telling the main greenlet The code thread is responsible for telling the main thread what it wants returned in the next run_code call - CodeRunner just passes whatever is passed in to run_code(for_code) to the code greenlet code thread. """
def __init__(self, interp=None, request_refresh=lambda: None):
@property def running(self): """Returns greenlet if code has been loaded greenlet has been started""" return self.source and self.code_context """Returns the running thread if code has been loaded and started.""" return self.source and self.code_thread
def load_code(self, source): """Prep code to be run""" assert self.source is None, ( "you shouldn't load code when some is " "already running" ) self.source = source self.code_context = None self.code_thread = None
def _unload_code(self): """Called when done running code""" self.source = None self.code_context = None self.code_thread = None self.code_is_waiting = False
def run_code(self, for_code=None): """Returns Truthy values if code finishes, False otherwise
if for_code is provided, send that value to the code greenlet if for_code is provided, send that value to the code thread if source code is complete, returns "done" if source code is incomplete, returns "unfinished" """ if self.code_context is None: if self.code_thread is None: assert self.source is not None self.code_context = greenlet.greenlet(self._blocking_run_code) self.code_thread = threading.Thread( target=self._blocking_run_code, name="codethread" ) self.code_thread.daemon = True if is_main_thread(): self.orig_sigint_handler = signal.getsignal(signal.SIGINT) signal.signal(signal.SIGINT, self.sigint_handler) request = self.code_context.switch() self.code_thread.start() else: assert self.code_is_waiting self.code_is_waiting = False if is_main_thread(): signal.signal(signal.SIGINT, self.sigint_handler) if self.sigint_happened_in_main_context: self.sigint_happened_in_main_context = False request = self.code_context.switch(SigintHappened) if self.sigint_happened_in_main_thread: self.sigint_happened_in_main_thread = False self.responses_for_code_thread.put(SigintHappened) else: request = self.code_context.switch(for_code) self.responses_for_code_thread.put(for_code)
request = self.requests_from_code_thread.get() logger.debug("request received from code was %r", request) if not isinstance(request, RequestFromCodeRunner): raise ValueError( "Not a valid value from code greenlet: %r" % request ) raise ValueError("Not a valid value from code thread: %r" % request) if isinstance(request, (Wait, Refresh)): self.code_is_waiting = True if isinstance(request, Refresh):
def _blocking_run_code(self): try: unfinished = self.interp.runsource(self.source) except SystemExit as e: return SystemExitRequest(*e.args) return Unfinished() if unfinished else Done() self.requests_from_code_thread.push(SystemExitRequest(*e.args)) return self.requests_from_code_thread.put( Unfinished() if unfinished else Done() )
def request_from_main_context(self, force_refresh=False): def request_from_main_thread(self, force_refresh=False): """Return the argument passed in to .run_code(for_code)
Nothing means calls to run_code must be... ??? """ if force_refresh: value = self.main_context.switch(Refresh()) self.requests_from_code_thread.put(Refresh()) value = self.responses_for_code_thread.get() else: value = self.main_context.switch(Wait()) self.requests_from_code_thread.put(Wait()) value = self.responses_for_code_thread.get() if value is SigintHappened: raise KeyboardInterrupt() return value
def write(self, s, *args, **kwargs): self.on_write(s, *args, **kwargs) return self.coderunner.request_from_main_context(force_refresh=True) return self.coderunner.request_from_main_thread(force_refresh=True)
# Some applications which use curses require that sys.stdout # have a method called fileno. One example is pwntools. This