rwols
commented
4 years ago This issue has raised questions in the Discord channel about the practically of an event loop, especially in light of the ST API. Let me preface that I do not ask to alter ST API functions. They can remain what they are.
The onus is on me to prove usefulness, so I'll try that with a tutorial of sorts.
Please create a file $packages/User/async_io.py and follow along :)
Starting and stopping an event loop
For now we want access to a loop. Let's create one ourselves in plugin_loaded and stop it in plugin_unloaded.
from threading import Thread
from typing import Optional
import asyncio
__loop: Optional[asyncio.AbstractEventLoop] = None
__thread: Optional[Thread] = None
def plugin_loaded() -> None:
print("loop: starting")
global __loop
global __thread
__loop = asyncio.new_event_loop()
__thread = Thread(target=__loop.run_forever)
__thread.start()
print("loop: started")
def __shutdown() -> None:
for task in asyncio.Task.all_tasks():
task.cancel()
asyncio.get_event_loop().stop()
def plugin_unloaded() -> None:
print("loop: stopping")
global __loop
global __thread
if __loop and __thread:
__loop.call_soon_threadsafe(__shutdown)
__thread.join()
__loop.run_until_complete(__loop.shutdown_asyncgens())
__loop.close()
__loop = None
__thread = None
print("loop: stopped")This will start and stop a new event loop whenever we save $packages/User/async_io.py. It shuts down in a clean manner by requesting all running tasks to cancel, then stopping the loop. The thread is joined on the UI thread of SublimeText (which should be fast, since the loop is stopping). Finally, we run the loop once more to allow coroutines to handle their CancelledError exception.
The loop is accessible through the global __loop variable.
Running blocking functions on the loop
Let's try to print to the console from a blocking function running on the loop. To do that we'll utilize sublime_plugin.EventListener and the call_soon_threadsafe method of AbstractEventLoop.
from sublime import View
from sublime_plugin import EventListener
def blocking_greet() -> None:
print("hello from a blocking function")
class MyEventListener(EventListener):
def on_selection_modified(self, view: View) -> None:
if view.settings().get("is_widget", False):
return
global __loop
assert __loop
__loop.call_soon_threadsafe(blocking_greet)Because the loop is running in a Python thread, we have to use call_soon_threadsafe, and not call_soon.
So far so good, but we could have accomplished the same with sublime.set_timeout_async(blocking_greet, 0). Let us continue on to coroutines...
Running asynchronous functions on the loop
Let's write a coroutine that prints to the console twice. The first time immediately, and the second time after one second.
async def greet_twice() -> None:
print("hello one!")
await asyncio.sleep(1)
print("hello two!")How do we run this coroutine? That is a two-step process. First, we have to get on the loop via call_soon_threadsafe. After that, we can schedule the coroutine to run with ensure_future. Let's abstract that away into a function:
from typing import Awaitable
def schedule(coro: Awaitable) -> None:
global __loop
if __loop:
__loop.call_soon_threadsafe(asyncio.ensure_future, coro)Now let's modify the MyEventListener class:
class MyEventListener(EventListener):
def on_selection_modified(self, view: View) -> None:
if view.settings().get("is_widget", False):
return
schedule(greet_twice())Check out the console and verify that you're running a coroutine from within SublimeText!
Cancellation
Asynchronous functions may receive an asyncio.CancelledError exception when awaiting an asynchronous result. To handle such an exception, simply catch it:
async def greet_twice_cancellation() -> None:
print("hello world!")
try:
await asyncio.sleep(1)
print("hello yet again world!")
except asyncio.CancelledError:
print("oh no! I was cancelled! Goodbye world!")Modify MyEventListener as follows:
def on_selection_modified(self, view: View) -> None:
if view.settings().get("is_widget", False):
return
schedule(greet_twice_cancellation())Now modify the selection, quickly save the file and verify that the coroutine is cancelled.
The Threadpool: No Need For threading.Thread
Along with scheduling coroutines, the AbstractEventLoop can also run intensive blocking functions concurrently using its run_in_executor method. Let's try that out. First let us write a blocking function that divides a number by two:
def blocking_compute(n: int) -> int:
print("compute: start", n)
import time
time.sleep(2) # zzz... working very hard
print("compute: done", n)
return n // 2Now we want to run a couple of invocations on the default threadpool and want to await the combined result.
async def compute_in_threadpool(view: View, point: int, *args: int) -> None:
loop = asyncio.get_event_loop()
coros = (loop.run_in_executor(None, blocking_compute, arg) for arg in args)
results = await asyncio.gather(*coros, loop=loop)
if view and view.is_valid():
view.show_popup(", ".join(map(str, results)), 0, point)Let's present it in a hover popup. Add an extra method to our MyEventListener:
def on_hover(self, view: View, point: int, hover_zone: int) -> None:
if view.settings().get("is_widget", False):
return
coro = compute_in_threadpool(
view, point, 2, 4, 8, 16, 32, 64, 128, 256)
schedule(coro)Case Study: Activity Monitor
Let's write an asynchronous activity monitor in less than twenty lines of Python code, in a single function:
GLYPHS = ("-", "\\", "|", "/", "-", "\\", "|", "/")
async def activity_monitor(view: View, key: str) -> None:
index = 0
try:
while True:
view.set_status(key, f"[{GLYPHS[index]}]")
index += 1
index %= len(GLYPHS)
await asyncio.sleep(0.1)
except Exception:
pass
finally:
try:
view.erase_status(key)
except Exception:
passWhen the view is no longer valid, the activity monitor stops. Let's use asyncio's cancellation facilities to make the monitor stop manually:
from sublime import Edit
class ActivityMonitorCommand(TextCommand):
keys: Dict[str, asyncio.Task] = {}
async def run_async(self, key: str) -> None:
task = self.keys.pop(key, None)
if task is None:
task = asyncio.create_task(activity_monitor(self.view, key))
self.keys[key] = task
else:
task.cancel()
def run(self, edit: Edit, key: str) -> None:
schedule(self.run_async(key))In the console, run
view.run_command("activity_monitor", {"key": "foo"})Run it again to stop the activity monitor. You can run an arbitrary number of activity monitors.
Case Study: Communicating With an External Process
More advanced plugins may communicate with external programs to retrieve useful information. The usual way to go about doing that is using the subprocess module and communicating with the process either in set_timeout_async or with a threading.Thread. Let's see how we can accomplish that with asyncio.
First let's write an external python script that prints a random string on stdout with a prefix given on stdin:
$ cat hello.py
#!/usr/bin/env python3
import time
import random
import sys
if __name__ == "__main__":
prefix = sys.stdin.read().strip()
time.sleep(1)
print(f"{prefix}:", random.choice(("foo", "bar", "baz")))Now we can write a command to call the external process:
from sublime import message_dialog
class RunExternalProgramCommand(TextCommand):
async def run_async(self, cmd: str, stdin_input: str) -> None:
process = await asyncio.create_subprocess_shell(
cmd,
stdout=PIPE,
stdin=PIPE)
encoding = "UTF-8"
errs = "replace"
inputbytes = stdin_input.encode(encoding, errs)
stdout_data, _ = await process.communicate(inputbytes)
message_dialog(stdout_data.decode(encoding, errs))
def run(self, edit: Edit, cmd: str, stdin_input: str) -> None:
schedule(self.run_async(cmd, stdin_input))Try running this locally to experience that the UI thread is not blocked at all.
Case Study: Asynchronous HTTP Requests
_This case study is hypothetical and has no code in our tutorial file async_io.py_. Suppose that the aiohttp library is available in SublimeText. We could write the following to fetch some HTML asynchronously:
import aiohttp
from sublime_plugin import TextCommand
from sublime import Edit
async def fetch(session: aiohttp.ClientSession, url: str) -> str:
async with session.get(url) as response:
return await response.text()
class FetchHtmlCommand(TextCommand):
async def run_async(self, url: str) -> None:
async with aiohttp.ClientSession() as session:
html = await fetch(session, url)
if self.view and self.view.is_valid():
window = self.view.window()
if not window:
return
window.new_file()
view = window.active_view()
if not view:
return
view.run_command("insert", {"characters": html})
def run(self, edit: Edit, url: str) -> None:
schedule(self.run_async(cmd, stdin_input))Case Study: Asynchronous Completions
Let's utilize asyncio to make our life easier with the new sublime.CompletionItems. First make a hypothetical event listener:
from sublime_plugin import ViewEventListener
from sublime import CompletionItem
from sublime import CompletionList
class MyViewEventListener(ViewEventListener):
# ...Now in order to handle asynchronous completions, let's make the promise object and schedule an async function to fulfill that promise:
def on_query_completions(
self,
prefix: str,
locations: List[int]
) -> Optional[CompletionList]:
if len(locations) != 1:
return None
promise = CompletionList()
coro = self.on_query_completions_async(prefix, locations[0], promise)
schedule(coro)
return promiseThe implementation of on_query_completions_async can be anything. Here is a mockup implementation:
async def on_query_completions_async(
self,
prefix: str,
location: int,
promise: CompletionList
) -> None:
await asyncio.sleep(1) # fetching completions from far away ...
snip = CompletionItem.snippet_completion
promise.set_completions([
snip("foo", "${1:foo}", "Does the foo"),
snip("bar", "${1:bar}", "Does the bar"),
snip("baz", "${1:baz}", "Does the baz")
])Notice how when you type the letter "b" the UI is not blocked and ST is awaiting the promise. After one second completions appear. Consider that one may combine this with the previous case studies using external process communication and HTTP requests.
Case Study: Implement set_timeout_async in terms of asyncio
We can implement sublime.set_timeout_async in terms of async functions and our now familiar and handy schedule function:
from typing import Callable
async def __wrapper(f: Callable[[], None], timeout_ms: int) -> None:
if timeout_ms > 0:
await asyncio.sleep(timeout_ms / 1000)
f()
def set_timeout_async(f: Callable[[], None], timeout_ms: int = 0) -> None:
schedule(__wrapper(f, timeout_ms))The function f blocks the event loop. This is okay if f is a "short" function. Otherwise, consider running f in the threadpool with run_in_executor.
Case Study: Implement Default/exec.py with asyncio
I have implemented our good old exec.py in terms of asyncio, removing the need for the subprocess and threading modules. It is a drop-in replacement. The file can be found as a gist. Check it out here. It uses our experimental async_io.py file to get access to our familiar schedule function.
- The
startandread_filenomethods ofAsyncProcesshave become async. - We redirect stderr to stdout.
- We schedule the
startmethod withschedulefrom ourasync_io.pyfile. - We
await self.read_fileno(self.proc.stdout).
Concrete Proposal
- Ensure a running loop when plugins call
asyncio.get_event_loop(). - Implement the
schedulefunction insublime.pyto easily schedule async functions on the loop.
The complete file async_io.py can be retrieved here as github gist.
deathaxe
wbond
kaste
daveleroy
Thom1729
FichteFoll
rchl
Problem description
Two plugins wanting to use the
asynciomodule will have to interact with an event loop. The "main" loop can be obtained viaasyncio.get_event_loop(). With build 4057 this loop is not yet running. This poses a "race condition" if two or more plugins want to use this global loop.All plugins could first check for a running loop, and start it if it hasn't started yet. Some plugin authors might forget, or might just not care and start it anyway. Some plugin authors might create their own separate loop with
my_loop = asyncio.new_event_loop()and starting it withthreading.Thread(target=my_loop.run_forever). This last possibility is especially annoying, because the global loop fromasyncio.get_event_loop()gives ST4 an opportunity to reduce its Python thread count. That is, using such a global loop is usually a superior alternative to firing up threads.Preferred solution
SublimeText should start the global event loop from
asynciobefore loading plugins.This guarantees all plugins a global running loop.
Additional Information (optional)
SublimeText build 4057.