Comfyui集群版 by KoalaBryson · Pull Request #22 · devsapp/cap-comfyui

KoalaBryson · 2025-10-31T09:00:58Z

No description provided.

CiCi503

去掉无关的 md 文件和测试脚本，提交前要看一遍提交的内容
代码结构调整：在 service 下新增 gateway 模块，将在 route 里和 serverlessapi 下的关于 cpu 函数的复杂逻辑移至 gateway 下

CiCi503 · 2025-11-03T11:28:03Z

src/code/agent/routes/routes.py

-            service.start(constants.AUTO_LAUNCH_SNAPSHOT_NAME)
+
+            print(f"Initializing function with ComfyUI mode: {constants.COMFYUI_MODE}")
+            service.start('latest-dev', nodes_map={})  # 使用latest-dev快照，与_tryStartIGService保持一致


这不行，需要区分 dev 环境和 prod 环境。
prod 环境的 initialize 需要基于环境变量里配置的 prod-xxx 的 snapshot 启动

src/code/agent/services/process/websocket/websocket_manager.py

CiCi503 · 2025-11-03T14:32:28Z

src/code/agent/routes/routes.py

+            print(f"[Enhanced History] Error getting all persisted history: {e}")
+            return {}
+
+    def _get_persisted_history_by_prompt_id(self, api_service, prompt_id):


不要将这些复杂的业务逻辑放到 routes 里。
在 service 下新建一个 gateway 目录，将 get_history、status_poller 相关逻辑移动至 gateway 中

CiCi503 · 2025-11-03T14:35:36Z

src/code/agent/services/serverlessapi/serverless_api_service.py

        else:
            return []

+    def get_status_from_store_incremental(self, task_id: str):


这个逻辑和 serverless api 无关，移走

CiCi503 · 2025-11-10T07:12:20Z

src/code/agent/main.py

+
+_original_print = builtins.print
+
+def timestamped_print(*args, **kwargs):


可以 rebase cap，用 agent/utils 中的 logger

CiCi503 · 2025-11-10T07:40:24Z

src/code/agent/routes/routes.py

+
+            # 使用环境变量指定的snapshot，默认为latest-dev
+            snapshot_name = os.environ.get('AUTO_LAUNCH_SNAPSHOT_NAME', 'latest-dev')
+            print(f"Initializing function with ComfyUI mode: {constants.COMFYUI_MODE}, snapshot: {snapshot_name}")


这里的 print/logger 使用 util 中的 logger

系统相关的日志默认设置为 DEBUG，默认用户在日志中不需要看到

CiCi503 · 2025-11-11T03:44:18Z

src/code/agent/routes/routes.py

-            service.start(constants.AUTO_LAUNCH_SNAPSHOT_NAME)
+
+            # 使用环境变量指定的snapshot，默认为latest-dev
+            snapshot_name = os.environ.get('AUTO_LAUNCH_SNAPSHOT_NAME', 'latest-dev')


latest-dev 的 snapshot 是在哪里生成的？

CiCi503 · 2025-11-11T03:51:51Z

src/code/agent/routes/routes.py

+
+                    # 发送初始状态消息（模拟ComfyUI原生行为）
+                    client_id = f"cpu_client_{int(time.time() * 1000)}"
+                    ws.send(json.dumps({


add_connection 中有 _send_initial_status ，这里也 set status 了，为啥需要有两份呢？

CiCi503 · 2025-11-11T03:54:28Z

src/code/agent/services/process/websocket/websocket_manager.py

            print(f"ws connected: {json.dumps(conn_info, indent=2)}")
+
+            # 设置TCP_NODELAY禁用Nagle算法，确保消息立即发送
+            self._set_tcp_nodelay(ws)


禁用 Nagle 算法是有必要的吗？解决了什么问题？需要引入这个复杂度吗？

CiCi503 · 2025-11-11T04:06:46Z

src/code/agent/services/process/websocket/websocket_manager.py

+        # 任务状态推送功能
+        self._task_subscriptions: Dict[str, Set] = {}  # task_id -> set of websockets
+        self._client_subscriptions: Dict = {}  # websocket -> set of task_ids
+        self._client_id_mapping: Dict[str, Set] = {}  # client_id -> set of websockets


client_id_mapping 为啥需要一个 set 呢？一个 client_id 应该对应一个 ws？

CiCi503 · 2025-11-11T04:13:25Z

src/code/agent/services/process/websocket/websocket_manager.py

+        self._task_subscriptions: Dict[str, Set] = {}  # task_id -> set of websockets
+        self._client_subscriptions: Dict = {}  # websocket -> set of task_ids
+        self._client_id_mapping: Dict[str, Set] = {}  # client_id -> set of websockets
+        self._ws_client_id_mapping: Dict = {}  # websocket -> client_id


ws_client_id_mapping 没有存在的必要吧， getattr(ws, '_comfyui_client_id'） 可以获取到 client_id 呀

CiCi503 · 2025-11-11T06:26:46Z

src/code/agent/routes/routes.py

-                    "message": "Please start your comfyui/sd service first"
-                }), 500
+        # CPU模式：接收ComfyUI原生的WebSocket连接，但推送基于任务队列的真实状态
+        if constants.COMFYUI_MODE == "cpu":


当前的 routes 代码结构职责混乱，有代码有路由，还有很多 cpu/gpu 的判断散落在各处。建议将 cpu_router 和 gpu_router 分开。

CiCi503 · 2025-11-11T07:24:05Z

src/code/agent/services/gateway/task_queue.py

+    return _task_queue_manager
+
+# 为了向后兼容，提供一个属性访问器
+class TaskQueueManagerProxy:


这个 TaskQueueManagerProxy 有啥用？

CiCi503 · 2025-11-24T08:21:33Z

src/code/agent/routes/cpu_routes.py

+                    }
+                }), 500
+
+    def _register_userdata_block(self):


优化函数名，_register_userdata_block => _register_userdata_handler？

设置一个环境变量吧，默认不允许 save userdata，可以通过环境变量开关控制是否 save userdata

src/code/agent/routes/serverless_api_routes.py

CiCi503 · 2025-11-24T13:21:25Z

src/code/agent/services/gateway/queue/task_queue.py

+    def _async_handle_task_failure(self, task_id: str, status_data: dict):
+        """异步处理任务失败（向后兼容方法）
+
+        注意：此方法已被 _async_handle_task_failure_after_update 替代


此方法已经被替代了，还留着干啥呢？

Change-Id: I45dc92f37937f2159a8141918718ddb1e5c97bc0 Co-developed-by: Cursor <noreply@cursor.com>

Change-Id: Iad50a0d65fb3137f369c36294584cee2c3b97783 Co-developed-by: Cursor <noreply@cursor.com>

Change-Id: Id6716878f840d1c44ef7f861f1cd459e59b166d8

Change-Id: I7a5940740cd62ea924248a701e18e81826ee99b1

Change-Id: Ia8ebf0ec505dd33071e48c2f26eb984e55d071f9

CiCi503 · 2025-12-03T13:19:54Z

src/code/agent/services/gateway/handlers/gpu_forwarder.py

+            if resp.status_code == 202:
+                return task_id, resp
+            else:
+                return None, (500, "async_invocation_error", f"Failed to invoke GPU function asynchronously: HTTP {resp.status_code}")


记录错误日志

更新 error code && error message

src/code/agent/services/gateway/handlers/gpu_forwarder.py

CiCi503 · 2025-12-03T13:34:04Z

src/code/agent/services/gateway/handlers/gpu_forwarder.py

+            except Exception as e:
+                error_msg = f"Failed to add task to queue: {e}"
+                log("ERROR", f"[TaskId={task_id}][RequestId={request_id}] {error_msg}\n{traceback.format_exc()}")
+                return None, (500, "queue_error", error_msg)


error_code 梳理一下

src/code/agent/services/gateway/handlers/gpu_forwarder.py

src/code/agent/services/gateway/handlers/history_gateway.py

Change-Id: I23283079c5872909a09bee8a87d5d6ef2077d3d4

CiCi503 reviewed Nov 3, 2025

View reviewed changes

CiCi503 requested changes Nov 3, 2025

View reviewed changes

KoalaBryson force-pushed the comfyui-cluster branch 2 times, most recently from 969857b to 12b563e Compare November 6, 2025 06:22

CiCi503 reviewed Nov 10, 2025

View reviewed changes

CiCi503 requested changes Nov 11, 2025

View reviewed changes

CiCi503 reviewed Nov 11, 2025

View reviewed changes

KoalaBryson force-pushed the comfyui-cluster branch 3 times, most recently from 047dbcf to 5326170 Compare November 24, 2025 02:51

CiCi503 reviewed Nov 24, 2025

View reviewed changes

src/code/agent/routes/serverless_api_routes.py Show resolved Hide resolved

CiCi503 reviewed Nov 24, 2025

View reviewed changes

KoalaBryson force-pushed the comfyui-cluster branch 2 times, most recently from aa5234b to c1206fa Compare November 25, 2025 13:39

CiCi503 and others added 4 commits November 30, 2025 11:32

feat: support zstd

247c3e1

Change-Id: I45dc92f37937f2159a8141918718ddb1e5c97bc0 Co-developed-by: Cursor <noreply@cursor.com>

bug fix

4f0723c

Change-Id: Iad50a0d65fb3137f369c36294584cee2c3b97783 Co-developed-by: Cursor <noreply@cursor.com>

CPU/GPU解耦

3004182

Change-Id: Id6716878f840d1c44ef7f861f1cd459e59b166d8

refactor

28553c8

Change-Id: I7a5940740cd62ea924248a701e18e81826ee99b1

KoalaBryson force-pushed the comfyui-cluster branch 5 times, most recently from 86e9a9a to 82c465d Compare December 3, 2025 11:57

refactor

50a9900

Change-Id: Ia8ebf0ec505dd33071e48c2f26eb984e55d071f9

KoalaBryson force-pushed the comfyui-cluster branch from 82c465d to 50a9900 Compare December 3, 2025 12:16

CiCi503 reviewed Dec 3, 2025

View reviewed changes

src/code/agent/services/gateway/handlers/gpu_forwarder.py Show resolved Hide resolved

CiCi503 reviewed Dec 3, 2025

View reviewed changes

src/code/agent/services/gateway/handlers/gpu_forwarder.py Show resolved Hide resolved

CiCi503 reviewed Dec 3, 2025

View reviewed changes

src/code/agent/services/gateway/handlers/history_gateway.py Show resolved Hide resolved

review

0d8a191

Change-Id: I23283079c5872909a09bee8a87d5d6ef2077d3d4


		_original_print = builtins.print

		def timestamped_print(args, *kwargs):

Conversation

KoalaBryson commented Oct 31, 2025

Uh oh!

CiCi503 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants