From f03e08d126c1e92255efc10c32aa54fff805de62 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 12 Sep 2023 15:48:02 +0100 Subject: [PATCH] Lint --- README.md | 2 +- permchain/pubsub.py | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 0e00491ca..cec57e5d3 100644 --- a/README.md +++ b/README.md @@ -27,11 +27,11 @@ Check `tests` and `examples` for more examples. - [x] Implement OUT as regular topic - [x] Implement IN as regular topic - [x] Add Connection.peek() to monitor past messages from all topics +- [x] Enable resuming PubSub from the "middle" of the computation - [ ] Move tracking of inflight processes/messages to Connection - [ ] Use this to build retry mechanism, where any inflight messages are moved back to the respective topics when restarting - [ ] Detect cycles (aka. infinite loops) and throw an error - [ ] Allow user to catch that error (by subcribing to an error topic?) -- [x] Enable resuming PubSub from the "middle" of the computation - [ ] Add "human in the loop" pattern - [ ] Add "wait until topic X is done" pattern, aka. `Topic.reduce()` - [ ] Add Redis-backed Connection implementation diff --git a/permchain/pubsub.py b/permchain/pubsub.py index 123cbb171..6c1aaf402 100644 --- a/permchain/pubsub.py +++ b/permchain/pubsub.py @@ -109,11 +109,11 @@ class PubSub(Serializable, Runnable[Any, Any], ABC): fut.add_done_callback(cleanup_run) # Listen on all subscribed topics - listeners_by_topic = groupby( + processes_by_topic = groupby( sorted(self.processes, key=lambda p: p.topic.name), lambda p: p.topic.name, ) - for topic_name, processes in listeners_by_topic: + for topic_name, processes in processes_by_topic: self.connection.listen( topic_prefix, topic_name, @@ -126,6 +126,8 @@ class PubSub(Serializable, Runnable[Any, Any], ABC): try: if inflight: # Yield output until all processes are done + # This blocks the current thread, all other work needs to go + # through the executor for chunk in self.connection.iterate(topic_prefix, OUTPUT_TOPIC): yield chunk else: