Too many Redis accesses in AppQueueManager

[ruanimal]

Self Checks

• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report (我已阅读并同意 Language Policy).
• [FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)
• Please do not modify this template :) and fill in all the required fields.

Content

Too many Redis accesses in AppQueueManager

Each time a message is generated, Redis is accessed twice to check if the workflow has stopped.
Especially when accessing LLM in a streaming mode, each SSE message will trigger Redis access.
I think this is redundant; accessing Redis once every few seconds, such as every 1 second, should be sufficient.

We can add memory caching for _is_stopped; if necessary, I can submit a pull request.

class AppQueueManager:
    ... # skip the irrelevant code.

    def listen(self):
           ....  # skip the irrelevant code.
                if elapsed_time >= listen_timeout or self._is_stopped():    # Redis accessed here
                    # publish two messages to make sure the client can receive the stop signal
                    # and stop listening after the stop signal processed
                    self.publish(
                        QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL), PublishFrom.TASK_PIPELINE
                    )

                if elapsed_time // 10 > last_ping_time:
                    self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
                    last_ping_time = elapsed_time // 10


class MessageBasedAppQueueManager(AppQueueManager):
    def _publish(self, event: AppQueueEvent, pub_from: PublishFrom):
        ...   # skip the irrelevant code.

        if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():    # Redis accessed here
            raise GenerateTaskStoppedError()

[ruanimal]

@dosu

[dosubot[bot]]

You're right—there's currently no in-memory caching for the _is_stopped method in AppQueueManager or its subclasses, so every check results in a Redis GET, which can be inefficient in streaming scenarios. Your suggestion to cache the stopped state in memory (for example, with a short TTL like 1 second) would help reduce redundant Redis calls and improve performance.

If you decide to implement this, consider how you'll handle cache invalidation and thread/process safety, especially if the manager is used in a multi-threaded or multi-process environment. A simple approach is to store the last checked value and timestamp, and only query Redis again if the cache has expired.

Feel free to submit a pull request with your optimization—this would be a valuable improvement! If you want feedback on your approach before submitting, just let me know.

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}