Add threadsafe/context.py

Author: peterhinch

v3/docs/THREADING.md

@@ -22,8 +22,10 @@ to WiFi and issue:
 import mip
 mip.install("github:peterhinch/micropython-async/v3/threadsafe")
 ```
-For non-networked targets use `mpremote` as described in
-[the official docs](http://docs.micropython.org/en/latest/reference/packages.html#installing-packages-with-mpremote).
+On any target `mpremote` may be used:
+```bash
+$ mpremote mip install github:peterhinch/micropython-async/v3/threadsafe
+```
 
 ###### [Main README](../README.md)
 ###### [Tutorial](./TUTORIAL.md)
@@ -47,6 +49,8 @@ For non-networked targets use `mpremote` as described in
   3.1 [Threadsafe Event](./THREADING.md#31-threadsafe-event)  
   3.2 [Message](./THREADING.md#32-message) A threadsafe event with data payload.  
  4. [Taming blocking functions](./THREADING.md#4-taming-blocking-functions) Enabling uasyncio to handle blocking code.  
+  4.1 [Basic approach](./THREADING.md#41-basic-approach)  
+  4.2 [More general solution](./THREADING,md#42-more-general-solution)  
  5. [Sharing a stream device](./THREADING.md#5-sharing-a-stream-device)  
  6. [Glossary](./THREADING.md#6-glossary) Terminology of realtime coding.  
 
@@ -188,7 +192,7 @@ thread safe classes offered here do not yet support Unix.
  is only required if mutual consistency of the three values is essential.
  3. In the absence of a GIL some operations on built-in objects are not thread
  safe. For example adding or deleting items in a `dict`. This extends to global
- variables which are implemented as a `dict`. See [Globals](./THREADING.md#15-globals).
+ variables because these are implemented as a `dict`. See [Globals](./THREADING.md#15-globals).
  4. The observations in 1.3 re user defined data structures and `uasyncio`
  interfacing apply.
  5. Code running on a core other than that running `uasyncio` may block for
@@ -643,7 +647,13 @@ again before it is accessed, the first data item will be lost.
 
 Blocking functions or methods have the potential of stalling the `uasyncio`
 scheduler. Short of rewriting them to work properly the only way to tame them
-is to run them in another thread. The following is a way to achieve this.
+is to run them in another thread. Any function to be run in this way must
+conform to the guiedelines above, notably with regard to allocation and side
+effects.
+
+## 4.1 Basic approach
+
+The following is a way to "unblock" a single function or method.
 ```python
 async def unblock(func, *args, **kwargs):
     def wrap(func, message, args, kwargs):
@@ -699,6 +709,42 @@ asyncio.run(main())
 ```
 ###### [Contents](./THREADING.md#contents)
 
+## 4.1 More general solution
+
+This provides a queueing mechanism. A task can assign a blocking function to a
+core even if the core is already busy. Further it allows for multiple cores or
+threads; these are defined as `Context` instances. Typical use:
+```python
+from threadsafe import Context
+
+core1 = Context()  # Has an instance of _thread, so a core on RP2
+
+def rats(t, n):  # Arbitrary blocking function or method
+    time.sleep(t)
+    return n * n
+
+async def some_task():
+    await core1.assign(rats, t=3, n=99)  # rats() runs on other core
+```
+#### Context class
+
+Constructor arg:
+ * `qsize=10` Size of function queue.
+
+Asynchronous method:
+ * `assign(func, *args, **kwargs)` Accepts a synchronous function with optional
+ args. These are placed on a queue for execution in the `Context` instance. The
+ method pauses until execution is complete, returning the fuction's return
+ value.
+
+The `Context` class constructor spawns a thread which waits on the `Context`
+queue. The`assign` method accepts a fuction and creates a `Job` instance. This
+includes a `ThreadSafeFlag` along with the function and its args. The `Assign`
+method places the `Job` on the queue and waits on the `ThreadSafeFlag`.
+
+The thread removes a `Job` from the queue and executes it. When complete it
+assigns the return value to the `Job` and sets the `ThreadSafeFlag`.
+
 # 5. Sharing a stream device
 
 Typical stream devices are a UART or a socket. These are typically employed to

v3/threadsafe/__init__.py

@@ -7,6 +7,7 @@
     "ThreadSafeEvent": "threadsafe_event",
     "ThreadSafeQueue": "threadsafe_queue",
     "Message": "message",
+    "Context": "context",
 }
 
 # Copied from uasyncio.__init__.py

v3/threadsafe/context.py

@@ -0,0 +1,31 @@
+# context.py: Run functions or methods on another core or in another thread
+
+import uasyncio as asyncio
+import _thread
+from threadsafe import ThreadSafeQueue
+
+# Object describing a job to be run on another core
+class Job:
+    def __init__(self, func, args, kwargs):
+        self.kwargs = kwargs
+        self.args = args
+        self.func = func
+        self.rval = None  # Return value
+        self.done = asyncio.ThreadSafeFlag()  # "done" indicator
+
+def worker(q):  # Runs forever on a core executing jobs as they arrive
+    while True:
+        job = q.get_sync(True)  # Block until a Job arrives
+        job.rval = job.func(*job.args, **job.kwargs)
+        job.done.set()
+
+class Context:
+    def __init__(self, qsize=10):
+        self.q = ThreadSafeQueue(qsize)
+        _thread.start_new_thread(worker, (self.q,))
+
+    async def assign(self, func, *args, **kwargs):
+        job = Job(func, args, kwargs)
+        await self.q.put(job)  # Will pause if q is full.
+        await job.done.wait()  # Pause until function has run
+        return job.rval

v3/threadsafe/package.json

@@ -4,6 +4,7 @@
     ["threadsafe/message.py", "github:peterhinch/micropython-async/v3/threadsafe/message.py"],
     ["threadsafe/threadsafe_event.py", "github:peterhinch/micropython-async/v3/threadsafe/threadsafe_event.py"],
     ["threadsafe/threadsafe_queue.py", "github:peterhinch/micropython-async/v3/threadsafe/threadsafe_queue.py"]
+    ["threadsafe/context.py", "github:peterhinch/micropython-async/v3/threadsafe/context.py"]
   ],
   "version": "0.1"
 }