Added new built-in coroutines.

Author: patrick-kidger

README.md

@@ -26,9 +26,9 @@ assert out == (4, 5)
 - Somewhat unusually, our syntax uses `yield` rather than `await`, but the behaviour is the same. Await another coroutine with `yield coro`. Await on multiple with `yield [coro1, coro2, ...]` (a 'gather' in asyncio terminology; a 'nursery' in trio terminology).
 - An error in one coroutine will cancel all coroutines across the entire event loop.
     - If the erroring coroutine is sequentially depended on by a chain of other coroutines, then we chain their tracebacks for easier debugging.
-    - Errors even propagate to and from synchronous operations ran in threads.
+    - Errors propagate to and from synchronous operations ran in threads.
 - Can nest tinyio loops inside each other, none of this one-per-thread business.
-- Ludicrously simple. No need for futures, tasks, etc. Here's the full API:
+- Ludicrously simple. No need for futures, tasks, etc. Here's the entirety of the day-to-day API:
     ```python
     tinyio.Loop
     tinyio.run_in_thread
@@ -44,7 +44,7 @@ pip install tinyio
 
 ## Documentation
 
-#### Loops
+### Loops
 
 Create a loop with `tinyio.Loop()`. It has a single method, `.run(coro)`, which consumes a coroutine, and which returns the output of that coroutine.
 
@@ -57,9 +57,10 @@ Coroutines can `yield` four possible things:
 
 You can safely `yield` the same coroutine multiple times, e.g. perhaps four coroutines have a diamond dependency pattern, with two coroutines each depending on a single shared one.
 
-#### Threading
+### 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:
 
-Synchronous functions can be ran in threads using `tinyio.run_in_thread(fn, *args, **kwargs)`, which returns a coroutine you can `yield` on:
 ```python
 import time, tinyio
 
@@ -75,13 +76,12 @@ loop = tinyio.Loop()
 out = loop.run(foo(x=1))  # runs in one second, not three
 assert out == [2, 2, 2]
 ```
-The thread will call `fn(*args, **kwargs)`.
 
-#### Sleeping
+### Sleeping
 
 This is `tinyio.sleep(delay_in_seconds)`, which is a coroutine you can `yield` on.
 
-#### Error propagation
+### Error propagation
 
 If any coroutine raises an error, then:
 
@@ -91,6 +91,100 @@ If any coroutine raises an error, then:
 
 This gives every coroutine a chance to shut down gracefully. Debuggers like [`patdb`](https://github.com/patrick-kidger/patdb) offer the ability to navigate across exceptions in an exception group, allowing you to inspect the state of all coroutines that were related to the error.
 
+### Batteries-included
+
+We ship batteries-included with a collection of standard operations, all built on top of just the functionality you've already seen.
+
+<details><summary>Click to expand</summary>
+
+```python
+tinyio.add_done_callback        tinyio.Semaphore
+tinyio.AsCompleted              tinyio.ThreadPool
+tinyio.Barrier                  tinyio.timeout
+tinyio.Event                    tinyio.TimeoutError
+tinyio.Lock
+```
+None of these require special support from the event loop, they are all just simple implementations that you could have written yourself :)
+
+---
+
+- `tinyio.add_done_callback(coro, success_callback)`
+
+    Used as `yield {tinyio.add_done_callback(coro, success_callback)}`.
+
+    This wraps `coro` so that `success_callback(out)` is called on its output once it completes. Note the `{...}` above, indicating calling this in nonblocking fashion (otherwise you could just directly call the callbacks yourself).
+
+---
+
+- `tinyio.AsCompleted({coro1, coro2, ...})`
+
+    This schedules multiple coroutines in the background (like `yield {coro1, coro2, ...}`), and then offers their results in the order they complete.
+
+    This is iterated over in the following way, using its `.done()` and `.get()` methods:
+    ```python
+    def main():
+        iterator = tinyio.AsCompleted({coro1, coro2, coro3})
+        while not iterator.done():
+            x = yield iterator.get()
+    ```
+
+---
+
+- `tinyio.Barrier(value)`
+
+    This has a single method `barrier.wait()`, which is a coroutine you can `yield` on. Once `value` many coroutines have yielded on this method then it will unblock.
+
+---
+
+- `tinyio.Event()`
+
+    This has a method `.wait()`, which is a coroutine you can `yield` on. This will unblock once its `.set()` method is called (typically from another coroutine). It also has a `is_set()` method for checking whether it has been set.
+
+---
+
+- `tinyio.Lock()`
+
+    This is just a convenience for `tinyio.Semaphore(value=1)`, see below.
+
+---
+
+- `tinyio.Semaphore(value)`
+
+    This manages an internal counter that is initialised at `value`, is decremented when entering a region, and incremented when exiting. This blocks if this counter is at zero. In this way, at most `value` coroutines may acquire the semaphore at a time.
+
+    This is used as:
+    ```python
+    semaphore = Semaphore(value)
+
+    ...
+
+    with (yield semaphore()):
+        ...
+    ```
+
+---
+
+- `tinyio.timeout(coro, timeout_in_seconds)`
+
+    This is a coroutine you can `yield` on, used  as `output, success = yield tinyio.timeout(coro, timeout_in_seconds)`.
+    
+    This runs `coro` for at most `timeout_in_seconds`. If it succeeds in that time then the pair `(output, True)` is returned . Else this will return `(None, False)`, and `coro` will be halted by raising `tinyio.TimeoutError` inside it.
+
+---
+
+- `tinyio.ThreadPool(max_threads)`
+
+    This is equivalent to making multiple `tinyio.run_in_thread` calls, but will limit the number of threads to at most `max_threads`. Additional work after that will block until a thread becomes available.
+
+    This has two methods:
+
+    - `.run_in_thread(fn, *args, **kwargs)`, which is a coroutine you can `yield` on. This is equivalent to `yield tinyio.run_in_thread(fn, *args, **kwargs)`.
+    - `.map(fn, xs)`, which is a coroutine you can `yield` on. This is equivalent to `yield [tinyio.run_in_thread(fn, x) for x in xs]`.
+ 
+---
+
+</details>
+
 ## FAQ
 
 <details>

pyproject.toml

@@ -20,7 +20,7 @@ name = "tinyio"
 readme = "README.md"
 requires-python = ">=3.11"
 urls = {repository = "https://github.com/patrick-kidger/tinyio"}
-version = "0.1.2"
+version = "0.1.3"
 
 [project.optional-dependencies]
 dev = ["pre-commit", "pytest"]

tests/test_background.py

@@ -0,0 +1,75 @@
+import tinyio
+
+
+def _block(event1: tinyio.Event, event2: tinyio.Event, out):
+    yield event1.wait()
+    event2.set()
+    return out
+
+
+def _test_done_callback():
+    out = []
+    event1 = tinyio.Event()
+    event2 = tinyio.Event()
+    event3 = tinyio.Event()
+    yield {tinyio.add_done_callback(_block(event2, event3, 1), out.append)}
+    yield {tinyio.add_done_callback(_block(event1, event2, 2), out.append)}
+    for _ in range(20):
+        yield
+    assert len(out) == 0
+    event1.set()
+    yield event3.wait()
+    assert out == [2, 1]
+
+
+def test_done_callback():
+    loop = tinyio.Loop()
+    loop.run(_test_done_callback())
+
+
+# To reinstate if we ever reintroduce error callbacks.
+
+
+# def _raises_after(event: tinyio.Event):
+#     yield event.wait()
+#     raise RuntimeError("Kaboom")
+
+
+# def _test_error_callback(error_callback1, error_callback2):
+#     event1 = tinyio.Event()
+#     event2 = tinyio.Event()
+#     yield {tinyio.add_done_callback(_raises_after(event1), _assert_false, error_callback1)}
+#     yield {tinyio.add_done_callback(_raises_after(event2), _assert_false, error_callback2)}
+#     event1.set()
+
+
+# def test_error_callback():
+#     loop = tinyio.Loop()
+#     out1 = []
+#     out2 = []
+#     with pytest.raises(RuntimeError, match="Kaboom") as catcher:
+#         loop.run(_test_error_callback(out1.append, out2.append))
+#     assert len(out1) == 1 and out1[0] is catcher.value
+#     assert len(out2) == 1 and type(out2[0]) is tinyio.CancelledError
+
+
+# def test_error_callback_that_raises1():
+#     loop = tinyio.Loop()
+#     out = []
+#     def error_callback(_):
+#         raise ValueError("Eek")
+#     with pytest.raises(ValueError, match="Eek"):
+#         loop.run(_test_error_callback(error_callback, out.append))
+#     assert len(out) == 1 and type(out[0]) is tinyio.CancelledError
+
+
+# def test_error_callback_that_raises2():
+#     loop = tinyio.Loop()
+#     out = []
+#     def error_callback(_):
+#         raise ValueError("Eek")
+#     with pytest.raises(BaseExceptionGroup) as catcher, pytest.warns(RuntimeWarning, match="leak"):
+#         loop.run(_test_error_callback(out.append, error_callback))
+#     [runtime, value] = catcher.value.exceptions
+#     assert type(runtime) is RuntimeError and str(runtime) == "Kaboom"
+#     assert type(value) is ValueError and str(value) == "Eek"

tests/test_basic.py

@@ -84,23 +84,6 @@ def _mul():
     assert loop.run(_mul()) == 25
 
 
-def test_in_thread():
-    def _blocking_slow_add_one(x: int) -> int:
-        time.sleep(0.1)
-        return x + 1
-
-    def _big_gather(x: int):
-        out = yield [tinyio.run_in_thread(_blocking_slow_add_one, x) for _ in range(100)]
-        return out
-
-    loop = tinyio.Loop()
-    start = time.time()
-    out = loop.run(_big_gather(1))
-    end = time.time()
-    assert out == [2 for _ in range(100)]
-    assert end - start < 0.5
-
-
 def test_waiting_on_already_finished():
     def f():
         yield

tests/test_examples.py

@@ -0,0 +1,26 @@
+import random
+import time
+
+import tinyio
+
+
+def test_dataloading():
+    outs = []
+
+    def slow_transform(x):
+        time.sleep(random.uniform(0.01, 0.02))  # slow I/O bound work
+        return x * x
+
+    def main():
+        iterator = range(100)
+        pool = tinyio.ThreadPool(16)
+        async_iterator = tinyio.AsCompleted({pool.run_in_thread(slow_transform, item) for item in iterator})
+        while not async_iterator.done():
+            out = yield async_iterator.get()
+            outs.append(out)
+        return outs
+
+    loop = tinyio.Loop()
+    out = loop.run(main())
+    assert set(out) == {x**2 for x in range(100)}
+    assert out != [x**2 for x in range(100)]  # test not in order. Very low chance of failing, should be fine!

tests/test_sync.py

@@ -0,0 +1,100 @@
+import time
+
+import pytest
+import tinyio
+
+
+def test_semaphore():
+    counter = 0
+
+    def _count(semaphore, i):
+        nonlocal counter
+        with (yield semaphore()):
+            counter += 1
+            if counter > 2:
+                raise RuntimeError
+            yield
+            counter -= 1
+        return i
+
+    def _run(value):
+        semaphore = tinyio.Semaphore(value)
+        out = yield [_count(semaphore, i) for i in range(50)]
+        return out
+
+    loop = tinyio.Loop()
+    assert loop.run(_run(2)) == list(range(50))
+    with pytest.raises(RuntimeError):
+        loop.run(_run(3))
+
+
+def test_lock():
+    counter = 0
+
+    def _count(semaphore, i):
+        nonlocal counter
+        with (yield semaphore()):
+            counter += 1
+            if counter > 1:
+                raise RuntimeError
+            yield
+            counter -= 1
+        return i
+
+    def _run():
+        semaphore = tinyio.Lock()
+        out = yield [_count(semaphore, i) for i in range(50)]
+        return out
+
+    loop = tinyio.Loop()
+    assert loop.run(_run()) == list(range(50))
+
+
+def test_barrier():
+    barrier = tinyio.Barrier(3)
+    count = 0
+
+    def _foo():
+        nonlocal count
+        count += 1
+        i = yield barrier.wait()
+        time.sleep(0.1)
+        assert count == 3
+        return i
+
+    def _run():
+        out = yield [_foo() for _ in range(3)]
+        return out
+
+    loop = tinyio.Loop()
+    assert set(loop.run(_run())) == {0, 1, 2}
+
+
+def test_event():
+    event = tinyio.Event()
+    done = False
+    done2 = False
+
+    def _foo():
+        nonlocal done
+        assert event.is_set() is False
+        yield event.wait()
+        assert event.is_set() is True
+        done = True
+
+    def _bar():
+        nonlocal done2
+        yield {_foo()}
+        for _ in range(10):
+            yield
+        assert not done
+        assert event.is_set() is False
+        event.set()
+        assert event.is_set()
+        done2 = True
+
+    loop = tinyio.Loop()
+    loop.run(_bar())
+    assert done
+    assert done2
+    assert event.is_set()