Threading

Blocking functions can be ran in threads using tinyio.run_in_thread(fn, *args, **kwargs), which gives a coroutine you can yield on.

Example

import time, tinyio

def slow_blocking_add_one(x: int) -> int:
    time.sleep(1)
    return x + 1

def foo(x: int):
    out = yield [tinyio.run_in_thread(slow_blocking_add_one, x) for _ in range(3)]
    return out

loop = tinyio.Loop()
out = loop.run(foo(x=1))  # runs in one second, not three
assert out == [2, 2, 2]

---

tinyio.run_in_thread(fn: Callable[~_Params, ~_Return], /, *args: _Params.args, **kwargs: _Params.kwargs) -> Coro[~_Return]

A tinyio coroutine for running the blocking function fn(*args, **kwargs) in a thread.

If this coroutine is cancelled then the cancellation will be raised in the thread as well; vice-versa if the function call raises an error then this will propagate to the coroutine. (Note that this is unlike normal Python threads, which do not have this feature.)

Arguments:

• fn: the function to call.
• *args: arguments to call fn with.
• **kwargs: keyword arguments to call fn with.

Returns:

A coroutine that can be yielded on, returning the output of fn(*args, **kwargs).

---

tinyio.ThreadPool

A wrapper around tinyio.run_in_thread to launch at most value many threads at a time.

__init__(max_threads: int)

Arguments:

• value: the maximum number of threads to launch at a time.

run_in_thread(fn: Callable[~_Params, ~_Return], /, *args: _Params.args, **kwargs: _Params.kwargs) -> Coro[~_Return]

Like tinyio.run_in_thread(fn, *args, **kwargs).

Usage is output = yield pool.run_in_thread(...)

map(fn: Callable[[~_T], ~_Return], /, xs: Iterable[~_T]) -> Coro[list[~_Return]]

Like [tinyio.run_in_thread(fn, x) for x in xs].

Usage is output_list = pool.map(...)