Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Robert Wojciechowski
|
14fcf5c0c3 |
@@ -7,18 +7,14 @@ on:
|
||||
paths:
|
||||
- "pyproject.toml"
|
||||
|
||||
permissions:
|
||||
issues: write
|
||||
|
||||
jobs:
|
||||
publish-node:
|
||||
name: Publish Custom Node to registry
|
||||
runs-on: ubuntu-latest
|
||||
if: ${{ github.repository_owner == 'robertvoy' }}
|
||||
steps:
|
||||
- name: Check out code
|
||||
uses: actions/checkout@v4
|
||||
- name: Publish Custom Node
|
||||
uses: Comfy-Org/publish-node-action@v1
|
||||
uses: Comfy-Org/publish-node-action@main
|
||||
with:
|
||||
personal_access_token: ${{ secrets.REGISTRY_ACCESS_TOKEN }}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
<div align="center">
|
||||
<img width="250" src="https://github.com/user-attachments/assets/533bb98d-0c4a-499f-9bca-5c937e361087" />
|
||||
<img width="320" src="https://github.com/user-attachments/assets/537d38cc-2e34-4527-bca7-0d79f4120cce" />
|
||||
<br><br>
|
||||
<a href="https://www.youtube.com/watch?v=p6eE3IlAbOs"><img src="https://img.shields.io/badge/Video_Tutorial-grey?style=flat&logo=youtube&logoColor=white" alt="Video Tutorial"></a>
|
||||
<a href="/docs/worker-setup-guides.md"><img src="https://img.shields.io/badge/Setup_Guides-grey?style=flat&logo=gitbook&logoColor=white" alt="Setup Guides"></a>
|
||||
@@ -9,7 +9,7 @@
|
||||
<br><br>
|
||||
</div>
|
||||
|
||||
> **A powerful extension for ComfyUI that enables distributed and parallel processing across multiple GPUs and machines. Generate more images and videos and accelerate your upscaling workflows by leveraging all available GPU resources in your network and cloud.**
|
||||
> **A powerful extension for ComfyUI that enables distributed and parallel processing across multiple GPUs and machines. Generate more images and videos and accelerate your upscaling workflows by leveraging all available GPU resources in your network.**
|
||||
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
## Key Features
|
||||
|
||||
#### Parallel Workflow Processing
|
||||
- Run your workflow on multiple GPUs simultaneously with varied seeds, collect results on the master
|
||||
- Run workflows on multiple GPUs simultaneously with varied seeds
|
||||
- Scale output with more workers
|
||||
- Supports images and videos
|
||||
|
||||
@@ -34,26 +34,11 @@
|
||||
|
||||
---
|
||||
|
||||
## Worker Types
|
||||
|
||||
<img width="200" align="right" alt="ComfyUI_temp_khvcc_00034_@0 25x" src="https://github.com/user-attachments/assets/651e4912-7c23-4e32-bd88-250f5175e129" />
|
||||
|
||||
ComfyUI Distributed supports three types of workers:
|
||||
|
||||
- **Local Workers** - Additional GPUs on the same machine (auto-configured on first launch)
|
||||
- **Remote Workers** - GPUs on other computers in your network
|
||||
- **Cloud Workers** - GPUs hosted on a cloud service like Runpod, accessible via secure tunnels
|
||||
|
||||
> For detailed setup instructions, see the [setup guide](/docs/worker-setup-guides.md)
|
||||
|
||||
---
|
||||
|
||||
## Requirements
|
||||
|
||||
- ComfyUI
|
||||
> Note: Desktop app not currently supported
|
||||
- Multiple NVIDIA GPUs
|
||||
> No additional GPUs? Use [Cloud Workers](https://github.com/robertvoy/ComfyUI-Distributed/blob/main/docs/worker-setup-guides.md#cloud-workers)
|
||||
- Multiple GPUs
|
||||
- That's it
|
||||
|
||||
---
|
||||
@@ -65,18 +50,9 @@ ComfyUI Distributed supports three types of workers:
|
||||
git clone https://github.com/robertvoy/ComfyUI-Distributed.git
|
||||
```
|
||||
|
||||
2. **Restart ComfyUI**
|
||||
- If you'll be using remote/cloud workers, add `--enable-cors-header` to your launch arguments on the master
|
||||
2. **Restart ComfyUI** - If you'll be using remote/cloud workers, add `--enable-cors-header` to your launch arguments on the master
|
||||
|
||||
3. Read the [setup guide](/docs/worker-setup-guides.md) for adding workers
|
||||
|
||||
---
|
||||
|
||||
## Official Sponsor
|
||||
|
||||
[<img width="1500" height="339" src="https://github.com/user-attachments/assets/c5f75e1f-3e19-4c57-b05d-151311cd1cf0" />](https://get.runpod.io/0bw29uf3ug0p)
|
||||
|
||||
Join Runpod with [this link](https://get.runpod.io/0bw29uf3ug0p) and unlock a special bonus.
|
||||
3. Read the [setup guide](https://github.com/robertvoy/ComfyUI-Distributed/blob/feature/cloud-workers/docs/worker-setup-guides.md) for adding workers
|
||||
|
||||
---
|
||||
|
||||
@@ -112,7 +88,7 @@ Generate multiple videos in the time it takes to generate one. Each worker uses
|
||||
8. Run the workflow!
|
||||
|
||||
### Distributed Upscaling
|
||||
Accelerate Ultimate SD Upscaler by distributing tiles across multiple workers, with speed scaling as you add more GPUs.
|
||||
Accelerate the process by distributing tiles across multiple workers, with speed scaling as you add more GPUs.
|
||||
|
||||
|
||||
|
||||
@@ -122,7 +98,6 @@ Accelerate Ultimate SD Upscaler by distributing tiles across multiple workers, w
|
||||
2. Upscale with ESRGAN or similar
|
||||
3. Connect to **Ultimate SD Upscale Distributed**
|
||||
4. Configure tile settings
|
||||
> If your GPUs are similar, set `static_distribution` to true; otherwise, false
|
||||
5. Enable workers for faster processing
|
||||
|
||||
---
|
||||
@@ -168,14 +143,4 @@ This software is provided "as is" without any warranties, express or implied, in
|
||||
---
|
||||
|
||||
## Support the Project
|
||||
|
||||
<img width="200" align="right" src="https://github.com/user-attachments/assets/84291921-c44e-4556-94f2-a3b16500f4f9" />
|
||||
|
||||
If my custom nodes have added value to your workflow, consider fueling future development with a coffee!
|
||||
|
||||
Your support helps keep this project thriving.
|
||||
|
||||
Buy me a coffee at: https://buymeacoffee.com/robertvoy
|
||||
|
||||
|
||||
|
||||
If my custom nodes have added value to your workflow, consider fueling future development with a coffee! Your support helps keep this project thriving. Buy me a coffee at: https://buymeacoffee.com/robertvoy
|
||||
|
||||
+1
-1
@@ -62,4 +62,4 @@ __all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS']
|
||||
|
||||
debug_log("Loaded Distributed nodes.")
|
||||
debug_log(f"Config file: {CONFIG_FILE}")
|
||||
debug_log(f"Available nodes: {list(NODE_CLASS_MAPPINGS.keys())}")
|
||||
debug_log(f"Available nodes: {list(NODE_CLASS_MAPPINGS.keys())}")
|
||||
+4
-15
@@ -1740,9 +1740,6 @@ async def job_complete_endpoint(request):
|
||||
image_data_list.sort(key=lambda x: x[0])
|
||||
tensors = [tensor for _, tensor in image_data_list]
|
||||
|
||||
# Keep the indices for later use
|
||||
indices = [idx for idx, _ in image_data_list]
|
||||
|
||||
# Validate final order
|
||||
if metadata:
|
||||
debug_log(f"Reordered {len(tensors)} images based on metadata indices")
|
||||
@@ -1771,11 +1768,10 @@ async def job_complete_endpoint(request):
|
||||
debug_log(f"Current pending jobs: {list(prompt_server.distributed_pending_jobs.keys())}")
|
||||
if multi_job_id in prompt_server.distributed_pending_jobs:
|
||||
if batch_size > 0:
|
||||
# Put batch as single item with indices
|
||||
# Put batch as single item
|
||||
await prompt_server.distributed_pending_jobs[multi_job_id].put({
|
||||
'worker_id': worker_id,
|
||||
'tensors': tensors,
|
||||
'indices': indices,
|
||||
'is_last': is_last
|
||||
})
|
||||
debug_log(f"Received batch result for job {multi_job_id} from worker {worker_id}, size={len(tensors)}")
|
||||
@@ -1938,7 +1934,6 @@ class DistributedCollectorNode:
|
||||
|
||||
# Check if batch mode
|
||||
tensors = result.get('tensors', [])
|
||||
indices = result.get('indices', []) # Get the indices
|
||||
if tensors:
|
||||
# Batch mode
|
||||
debug_log(f"Master - Got batch from worker {worker_id}, size={len(tensors)}, is_last={is_last}")
|
||||
@@ -1946,15 +1941,9 @@ class DistributedCollectorNode:
|
||||
if worker_id not in worker_images:
|
||||
worker_images[worker_id] = {}
|
||||
|
||||
# Use actual indices if available, otherwise fall back to sequential
|
||||
if indices:
|
||||
for i, tensor in enumerate(tensors):
|
||||
actual_idx = indices[i]
|
||||
worker_images[worker_id][actual_idx] = tensor
|
||||
else:
|
||||
# Fallback for backward compatibility
|
||||
for idx, tensor in enumerate(tensors):
|
||||
worker_images[worker_id][idx] = tensor
|
||||
# Add all tensors with sequential indices
|
||||
for idx, tensor in enumerate(tensors):
|
||||
worker_images[worker_id][idx] = tensor
|
||||
|
||||
|
||||
collected_count += len(tensors)
|
||||
|
||||
+885
-637
File diff suppressed because it is too large
Load Diff
@@ -1,14 +1,15 @@
|
||||
## Automating ComfyUI Model Downloads
|
||||
> This guide will walk you through creating a shell script to automatically download the necessary models for your ComfyUI workflow, leveraging an advanced Large Language Model (LLM).
|
||||
|
||||
1. In ComfyUI (on your local machine), export your workflow as an API workflow
|
||||
2. Copy the below prompt and upload the API workflow to an LLM **that has access to the internet**
|
||||
|
||||
<details>
|
||||
<summary><strong>📋 Click to expand the full prompt</strong></summary>
|
||||
This guide will walk you through creating a shell script to automatically download the necessary models for your ComfyUI workflow, leveraging an advanced Large Language Model (LLM).
|
||||
|
||||
1. Launched the Pod with these Environment Variables:
|
||||
- CIVITAI_API_TOKEN ([get your token here](https://civitai.com/user/account))
|
||||
- HF_API_TOKEN ([get your token here](https://huggingface.co/settings/tokens))
|
||||
2. In ComfyUI, export your workflow as an API workflow
|
||||
3. Copy the below prompt and upload the API workflow to a LLM **that has access to the internet**
|
||||
```
|
||||
Create a sh script that will download the models from this workflow into the correct folders. For reference, these are the paths:
|
||||
|
||||
base_path: /workspace/ComfyUI
|
||||
checkpoints: models/checkpoints/
|
||||
clip: models/clip/
|
||||
@@ -24,30 +25,32 @@ text_encoders: models/text_encoders/
|
||||
unet: models/unet/
|
||||
upscale_models: models/upscale_models/
|
||||
vae: models/vae/
|
||||
|
||||
---
|
||||
|
||||
Important:
|
||||
Make sure you find the correct URLs for the models online.
|
||||
Use comfy cli to download the models: `comfy model download --url <URL> [--relative-path <PATH>] [--set-civitai-api-token <TOKEN>] [--set-hf-api-token <TOKEN>]`
|
||||
Make sure you add `--set-civitai-api-token $CIVITAI_API_TOKEN` for CivitAI download and `--set-hf-api-token $HF_API_TOKEN` for Hugging Face downloads.
|
||||
|
||||
---
|
||||
|
||||
Example:
|
||||
|
||||
#!/bin/bash
|
||||
|
||||
# Download from CivitAI
|
||||
comfy model download --url https://civitai.com/api/download/models/1759168 --relative-path /workspace/ComfyUI/models/checkpoints --set-civitai-api-token $CIVITAI_API_TOKEN
|
||||
|
||||
# Download model from Hugging Face
|
||||
comfy model download --url https://huggingface.co/black-forest-labs/FLUX.1-dev/resolve/main/flux1-dev.safetensors --relative-path /workspace/ComfyUI/models/unet --set-hf-api-token $HF_API_TOKEN
|
||||
|
||||
# If a model in the workflow was in a subfolder
|
||||
comfy model download --url https://civitai.com/api/download/models/1759168 --relative-path /workspace/ComfyUI/models/checkpoints/SDXL --set-civitai-api-token $CIVITAI_API_TOKEN
|
||||
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
3. Review the LLMs output to make sure all download links are correct and save it as a .sh file, for example `download_models.sh`
|
||||
4. Launch the [ComfyUI Distributed Pod](https://console.runpod.io/deploy?template=m21ynvo8yo&ref=ak218p52) with these Environment Variables:
|
||||
- `CIVITAI_API_TOKEN`: [get your token here](https://civitai.com/user/account)
|
||||
- `HF_API_TOKEN`: [get your token here](https://huggingface.co/settings/tokens)
|
||||
5. Upload the .sh file to your Runpod instance, into `/workspace`
|
||||
4. Review the LLMs output to make sure all download links are correct and save it as a .sh file, for example `download_models.sh`
|
||||
5. Upload it onto your Runpod instance, into `/workspace`
|
||||
6. Then run these commands:
|
||||
- `chmod 755 /workspace/download_models.sh`
|
||||
- `/workspace/download_models.sh`
|
||||
7. Confirm each model name (sometimes you might need to rename them to match the name on your local machine)
|
||||
|
||||
+25
-54
@@ -1,11 +1,11 @@
|
||||
## Worker Setup Guide
|
||||
## ComfyUI Distributed Worker Setup Guide
|
||||
|
||||
ComfyUI-Distributed allows you to distribute your workflows across multiple GPUs, whether they're in the same machine, on your local network, or in the cloud.
|
||||
|
||||
**Master**: The main ComfyUI instance that coordinates and distributes work. This is where you load workflows, manage the queue, and view results.
|
||||
|
||||
**Worker**: A ComfyUI instance that receives and processes tasks from the master. Workers handle just the GPU computation and send results back to the master. You can have multiple workers connected to a single master, each utilizing their own GPU.
|
||||
|
||||
<img width="600" src="https://github.com/user-attachments/assets/609c42aa-8a1c-4a3f-939e-f3552fa1d54f" />
|
||||
|
||||
### Types of Workers
|
||||
|
||||
- **Local workers**: Additional GPUs on the same machine as the master
|
||||
@@ -13,14 +13,8 @@
|
||||
- **Cloud workers**: GPUs hosted on cloud services like Runpod
|
||||
|
||||
## Local workers
|
||||
|
||||
<img align="right" width="200" src="https://github.com/user-attachments/assets/651e4912-7c23-4e32-bd88-250f5175e129" />
|
||||
|
||||
> These are added automatically on first launch, but you can add them manually if you need to.
|
||||
|
||||
|
||||
📺 [Watch Tutorial](https://youtu.be/p6eE3IlAbOs?si=K7Km0_flmPHwRQwz&t=43)
|
||||
|
||||
1. **Open** the Distributed GPU panel.
|
||||
2. **Click** "Add Worker" in the UI.
|
||||
3. **Configure** your local worker:
|
||||
@@ -31,14 +25,8 @@
|
||||
4. **Save** and launch the local worker.
|
||||
|
||||
## Remote workers
|
||||
|
||||
<img align="right" width="200" src="https://github.com/user-attachments/assets/84291921-c44e-4556-94f2-a3b16500f4f9" />
|
||||
|
||||
|
||||
> ComfyUI instances running on completely different computers on your network. These allow you to harness GPU power from other machines. Remote workers must be manually started on their respective computers and are connected via IP address.
|
||||
|
||||
📺 [Watch Tutorial](https://youtu.be/p6eE3IlAbOs?si=Oxj3EzPyf4jKDvfG&t=140)
|
||||
|
||||
**On the Remote Worker Machine:**
|
||||
1. **Launch** ComfyUI with the `--listen --enable-cors-header` arguments. ⚠️ **Required!**
|
||||
- This ComfyUI instance will serve as a worker for your main master.
|
||||
@@ -61,55 +49,38 @@
|
||||
6. **Save** the remote worker configuration.
|
||||
|
||||
## Cloud workers
|
||||
|
||||
<img align="right" width="200" src="https://github.com/user-attachments/assets/a053f3ae-22f0-4e1c-8f2e-f26a1f660adf" />
|
||||
|
||||
> ComfyUI instances running on a cloud service like Runpod.
|
||||
> ComfyUI instances running on a cloud service like [Runpod](https://get.runpod.io/0bw29uf3ug0p).
|
||||
|
||||
### Deploy Cloud Worker on Runpod
|
||||
|
||||
📺 [Watch Tutorial](https://www.youtube.com/watch?v=wxKKWMQhYTk)
|
||||
|
||||
**On Runpod:**
|
||||
> If using your own template, make sure you launch ComfyUI with the `--enable-cors-header` argument and you `git clone ComfyUI-Distributed` into custom_nodes. ⚠️ **Required!**
|
||||
|
||||
1. Register a [Runpod](https://get.runpod.io/0bw29uf3ug0p) account.
|
||||
2. On Runpod, go to Storage > New Network Volume and create a volume that will store the models you need. Start with 40 GB, you can always add more later. Learn more [about Network Volumes](https://docs.runpod.io/pods/storage/create-network-volumes).
|
||||
3. Use the [ComfyUI Distributed Pod](https://console.runpod.io/deploy?template=m21ynvo8yo&ref=0bw29uf3ug0p) template.
|
||||
4. Make sure your Network Volume is mounted and choose a suitable GPU.
|
||||
> ⚠️ To use the ComfyUI Distributed Pod template, you will need to filter instances by CUDA 12.8 (add filter in Additional Filters).
|
||||
6. Press Edit Template to configure the pod's Environment Variables:
|
||||
- CIVITAI_API_TOKEN: [get your token here](https://civitai.com/user/account)
|
||||
- HF_API_TOKEN: [get your token here](https://huggingface.co/settings/tokens)
|
||||
- SAGE_ATTENTION: optional optimisation (set to true/false)
|
||||
5. Deploy your pod.
|
||||
6. Connect to your pod using JupyterLabs. This gives us access to the pod's file system.
|
||||
7. Download models into /workspaces/ComfyUI/models/ (these will remain on your network drive even after you terminate the pod). Example commands below:
|
||||
```
|
||||
# Download from CivitAI
|
||||
comfy model download --url https://civitai.com/api/download/models/1759168 --relative-path /workspace/ComfyUI/models/checkpoints --set-civitai-api-token $CIVITAI_API_TOKEN
|
||||
# Download model from Hugging Face
|
||||
comfy model download --url https://huggingface.co/black-forest-labs/FLUX.1-dev/resolve/main/flux1-dev.safetensors --relative-path /workspace/ComfyUI/models/unet --set-hf-api-token $HF_API_TOKEN
|
||||
```
|
||||
> ℹ️ Use [this guide](model-download-script.md) to make this process easy. It will generate a shell script that automatically downloads the models for a given workflow.
|
||||
9. Access ComfyUI through the Runpod URL.
|
||||
10. Download any additional custom nodes you need using the ComfyUI Manager.
|
||||
2. On Runpod, go to Storage > New Network Volume and create a volume that will store the models you need. Start with 40 GB, you can always add more later.
|
||||
3. Now go to Pods and find a suitable GPU for your workflows.
|
||||
4. Choose the [ComfyUI Distributed Pod](https://console.runpod.io/deploy?template=m21ynvo8yo&ref=ak218p52) template and make sure your network drive is mounted.
|
||||
> To use the ComfyUI Distributed Pod template, you will need to filter instances by CUDA 12.8.
|
||||
6. Launch your pod.
|
||||
7. Access your pod using JupyterLabs.
|
||||
8. Download models into /workspaces/ComfyUI/models/ (these will remain on your network drive even after you terminate the pod).
|
||||
> You can use [this guide](model-download-script.md) to make this process easy for you. It will generate a shell script that will automatically download the models you need for a given workflow.
|
||||
8. If using your own template, make sure you launch ComfyUI with the `--enable-cors-header` argument and you git clone ComfyUI-Distributed into custom_nodes. ⚠️ **Required!**
|
||||
9. Download any additional custom nodes using the ComfyUI Manager.
|
||||
|
||||
**On the Main Machine:**
|
||||
1. **Launch** a Cloudflare tunnel.
|
||||
1. Launch a Cloudflare tunnel
|
||||
- Download from here: [https://github.com/cloudflare/cloudflared/releases](https://github.com/cloudflare/cloudflared/releases)
|
||||
- Then run, for example: `cloudflared-windows-amd64.exe tunnel --url http://localhost:8188`
|
||||
> ℹ️ Cloudflare tunnels create secure connections without exposing ports directly to the internet and are required for Cloud Workers.
|
||||
2. **Copy** the Cloudflare address
|
||||
> Cloudflare tunnels create secure connections without exposing ports directly to the internet
|
||||
2. Copy the Cloudflare address
|
||||
3. **Launch** ComfyUI with `--enable-cors-header` launch argument.
|
||||
4. **Open** the Distributed GPU panel (sidebar on the left).
|
||||
5. **Edit** the Master's settings to change the host address to the Cloudflare address.
|
||||
6. **Click** "Add Worker."
|
||||
7. **Choose** "Cloud".
|
||||
8. **Configure** your cloud worker:
|
||||
- **Host**: The ComfyUI Runpod address. For example: `wcegfo9tbbml9l-8188.proxy.runpod.net`
|
||||
- **Port**: 443
|
||||
9. **Save** the remote worker configuration.
|
||||
5. **Edit** the Master's host address and replace it with the Cloudflare address.
|
||||
- **Click** "Add Worker."
|
||||
- **Choose** "Cloud".
|
||||
- **Configure** your cloud worker:
|
||||
- **Host**: The Runpod address
|
||||
- **Port**: 443
|
||||
- **Save** the remote worker configuration.
|
||||
|
||||
---
|
||||
|
||||
@@ -120,7 +91,7 @@ comfy model download --url https://huggingface.co/black-forest-labs/FLUX.1-dev/r
|
||||
- If your cloud platform doesn't provide a secure connection, use Cloudflare to create a tunnel for the worker. Each GPU needs their own tunnel for their respective port.
|
||||
- For example: `./cloudflared tunnel --url http://localhost:8188`
|
||||
1. **Launch** ComfyUI with the `--listen --enable-cors-header` arguments. ⚠️ **Required!**
|
||||
2. **Add** workers in the UI panel if the cloud machine has more than one GPU.
|
||||
2. **Add** workers in the UI panel if the remote machine has more than one GPU.
|
||||
- Make sure that they also have `--listen` set in `Extra Args`.
|
||||
- Then launch them.
|
||||
|
||||
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
name = "ComfyUI-Distributed"
|
||||
description = "ComfyUI extension that enables multi-GPU processing locally, remotely and in the cloud"
|
||||
version = "1.0.9"
|
||||
description = "ComfyUI extension that enables multi-GPU processing locally and remotely "
|
||||
version = "1.0.1"
|
||||
license = {file = "LICENSE"}
|
||||
dependencies = []
|
||||
|
||||
@@ -12,4 +12,4 @@ Repository = "https://github.com/robertvoy/ComfyUI-Distributed"
|
||||
[tool.comfy]
|
||||
PublisherId = "robertvoy"
|
||||
DisplayName = "ComfyUI-Distributed"
|
||||
Icon = "https://raw.githubusercontent.com/robertvoy/ComfyUI-Distributed/refs/heads/main/web/distributed-logo-icon.png"
|
||||
Icon = ""
|
||||
|
||||
+408
-217
@@ -7,96 +7,19 @@ import io
|
||||
from aiohttp import web
|
||||
import server
|
||||
from PIL import Image
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
# Import from other utilities
|
||||
from .logging import debug_log, log
|
||||
from .network import handle_api_error, get_client_session
|
||||
# We avoid converting to tensors on the master for tiles; blending uses PIL
|
||||
from .image import tensor_to_pil
|
||||
|
||||
# Configure maximum payload size (50MB default, configurable via environment variable)
|
||||
MAX_PAYLOAD_SIZE = int(os.environ.get('COMFYUI_MAX_PAYLOAD_SIZE', str(50 * 1024 * 1024)))
|
||||
|
||||
# Import HEARTBEAT_TIMEOUT from constants
|
||||
from .constants import HEARTBEAT_TIMEOUT
|
||||
from .config import load_config
|
||||
|
||||
|
||||
def _parse_tiles_from_form(data):
|
||||
"""Parse tiles submitted via multipart/form-data into a list of tile dicts.
|
||||
|
||||
Expects the following fields in the aiohttp form data:
|
||||
- 'tiles_metadata': JSON list with per-tile metadata items containing at least
|
||||
'tile_idx', 'x', 'y', 'extracted_width', 'extracted_height'. Optional
|
||||
'batch_idx' and 'global_idx' are included when available.
|
||||
- 'tile_{i}': image bytes for each tile described in tiles_metadata (PNG).
|
||||
- 'padding': integer padding used during extraction (optional; defaults 0).
|
||||
|
||||
Returns: list of dicts with keys: 'image', 'tile_idx', 'x', 'y',
|
||||
'extracted_width', 'extracted_height', and optional 'batch_idx', 'global_idx',
|
||||
plus 'padding'.
|
||||
"""
|
||||
try:
|
||||
# Parse padding if present
|
||||
padding = int(data.get('padding', 0)) if data.get('padding') is not None else 0
|
||||
except Exception:
|
||||
padding = 0
|
||||
|
||||
# Parse tiles metadata (JSON list)
|
||||
meta_raw = data.get('tiles_metadata')
|
||||
if meta_raw is None:
|
||||
raise ValueError("Missing tiles_metadata")
|
||||
|
||||
try:
|
||||
metadata = json.loads(meta_raw)
|
||||
except Exception as e:
|
||||
raise ValueError(f"Invalid tiles_metadata JSON: {e}")
|
||||
|
||||
if not isinstance(metadata, list):
|
||||
raise ValueError("tiles_metadata must be a list")
|
||||
|
||||
tiles = []
|
||||
# Iterate over metadata items and corresponding uploaded files tile_0, tile_1, ...
|
||||
for i, meta in enumerate(metadata):
|
||||
file_field = data.get(f'tile_{i}')
|
||||
if file_field is None or not hasattr(file_field, 'file'):
|
||||
raise ValueError(f"Missing tile data for index {i}")
|
||||
|
||||
# Read image bytes and decode to PIL
|
||||
raw = file_field.file.read()
|
||||
try:
|
||||
img = Image.open(io.BytesIO(raw)).convert("RGB")
|
||||
except Exception as e:
|
||||
raise ValueError(f"Invalid image data for tile {i}: {e}")
|
||||
|
||||
# Build tile dictionary (store PIL only; master blends via PIL)
|
||||
try:
|
||||
tile_info = {
|
||||
'image': img,
|
||||
'tile_idx': int(meta.get('tile_idx', i)),
|
||||
'x': int(meta.get('x', 0)),
|
||||
'y': int(meta.get('y', 0)),
|
||||
'extracted_width': int(meta.get('extracted_width', img.width)),
|
||||
'extracted_height': int(meta.get('extracted_height', img.height)),
|
||||
'padding': int(padding),
|
||||
}
|
||||
except Exception as e:
|
||||
raise ValueError(f"Invalid metadata values for tile {i}: {e}")
|
||||
|
||||
# Optional fields
|
||||
if 'batch_idx' in meta:
|
||||
try:
|
||||
tile_info['batch_idx'] = int(meta['batch_idx'])
|
||||
except Exception:
|
||||
pass
|
||||
if 'global_idx' in meta:
|
||||
try:
|
||||
tile_info['global_idx'] = int(meta['global_idx'])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
tiles.append(tile_info)
|
||||
|
||||
return tiles
|
||||
|
||||
|
||||
# Unified Job Data Structure Keys
|
||||
@@ -113,46 +36,7 @@ JOB_NUM_TILES_PER_IMAGE = 'num_tiles_per_image' # For static
|
||||
TASK_TYPE_TILE = 'tile'
|
||||
TASK_TYPE_IMAGE = 'image'
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
async def init_dynamic_job(multi_job_id: str, batch_size: int, enabled_workers: List[str], all_indices: Optional[List[int]] = None):
|
||||
"""Initialize queue for dynamic mode (per-image), with collector fields.
|
||||
|
||||
- Creates JOB_PENDING_TASKS with image indices
|
||||
- Adds 'completed_images' dict and 'pending_images' alias used by collectors
|
||||
"""
|
||||
await _init_job_queue(
|
||||
multi_job_id,
|
||||
'dynamic',
|
||||
batch_size=batch_size,
|
||||
all_indices=all_indices or list(range(batch_size)),
|
||||
enabled_workers=enabled_workers,
|
||||
)
|
||||
|
||||
prompt_server = ensure_tile_jobs_initialized()
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
job_data['completed_images'] = {}
|
||||
job_data['pending_images'] = job_data[JOB_PENDING_TASKS]
|
||||
debug_log(f"Job {multi_job_id} initialized with {batch_size} images")
|
||||
|
||||
|
||||
async def init_static_job_batched(multi_job_id: str, batch_size: int, num_tiles_per_image: int, enabled_workers: List[str]):
|
||||
"""Initialize queue for static mode (batched-per-tile).
|
||||
|
||||
- Populates JOB_PENDING_TASKS with tile ids [0..num_tiles_per_image-1]
|
||||
"""
|
||||
await _init_job_queue(
|
||||
multi_job_id,
|
||||
'static',
|
||||
batch_size=batch_size,
|
||||
num_tiles_per_image=num_tiles_per_image,
|
||||
enabled_workers=enabled_workers,
|
||||
batched_static=True,
|
||||
)
|
||||
# Initialization handled by master; avoid duplicate init logs here
|
||||
|
||||
async def _init_job_queue(multi_job_id, mode, batch_size=None, num_tiles_per_image=None, all_indices=None, enabled_workers=None, task_assignments=None, batched_static: bool = False):
|
||||
async def _init_job_queue(multi_job_id, mode, batch_size=None, num_tiles_per_image=None, all_indices=None, enabled_workers=None, task_assignments=None):
|
||||
"""Unified initialization for job queues in static and dynamic modes."""
|
||||
prompt_server = ensure_tile_jobs_initialized()
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
@@ -172,22 +56,17 @@ async def _init_job_queue(multi_job_id, mode, batch_size=None, num_tiles_per_ima
|
||||
if mode == 'dynamic':
|
||||
job_data[JOB_BATCH_SIZE] = batch_size
|
||||
pending_queue = job_data[JOB_PENDING_TASKS]
|
||||
for i in (all_indices or range(batch_size)):
|
||||
for i in all_indices or range(batch_size):
|
||||
await pending_queue.put(i)
|
||||
debug_log(f"Initialized image queue with {batch_size} pending items")
|
||||
debug_log(f"Initialized dynamic queue with {batch_size} pending images")
|
||||
elif mode == 'static':
|
||||
job_data[JOB_NUM_TILES_PER_IMAGE] = num_tiles_per_image
|
||||
job_data[JOB_BATCH_SIZE] = batch_size
|
||||
job_data['batched_static'] = bool(batched_static)
|
||||
# For batched static distribution, populate only tile ids [0..num_tiles_per_image-1]
|
||||
# For static with dynamic distribution, populate all tile indices in pending queue
|
||||
pending_queue = job_data[JOB_PENDING_TASKS]
|
||||
if batched_static and num_tiles_per_image is not None:
|
||||
for i in range(num_tiles_per_image):
|
||||
await pending_queue.put(i)
|
||||
else:
|
||||
total_tiles = batch_size * num_tiles_per_image
|
||||
for i in range(total_tiles):
|
||||
await pending_queue.put(i)
|
||||
total_tiles = batch_size * num_tiles_per_image
|
||||
for i in range(total_tiles):
|
||||
await pending_queue.put(i)
|
||||
debug_log(f"Initialized static queue with {total_tiles} pending tiles for dynamic distribution")
|
||||
|
||||
# Keep backward compatibility - if task assignments provided, still track them
|
||||
if task_assignments and enabled_workers:
|
||||
@@ -201,13 +80,37 @@ async def _init_job_queue(multi_job_id, mode, batch_size=None, num_tiles_per_ima
|
||||
|
||||
prompt_server.distributed_pending_tile_jobs[multi_job_id] = job_data
|
||||
|
||||
# Note: legacy task distribution and queue pull helpers removed
|
||||
def _distribute_tasks(items: list, num_participants: int) -> list[list[any]]:
|
||||
"""Distribute a list of items among N participants (master + workers)."""
|
||||
if num_participants <= 1:
|
||||
return [items]
|
||||
|
||||
items_per_participant = len(items) // num_participants
|
||||
remainder = len(items) % num_participants
|
||||
assignments = []
|
||||
start_idx = 0
|
||||
for i in range(num_participants):
|
||||
count = items_per_participant + (1 if i < remainder else 0)
|
||||
end_idx = start_idx + count
|
||||
assignments.append(items[start_idx:end_idx])
|
||||
start_idx = end_idx
|
||||
return assignments
|
||||
|
||||
async def _get_next_task(multi_job_id):
|
||||
"""Get next task from pending queue (generalized for tiles/images)."""
|
||||
prompt_server = ensure_tile_jobs_initialized()
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs.get(multi_job_id)
|
||||
if not job_data or JOB_PENDING_TASKS not in job_data:
|
||||
return None
|
||||
try:
|
||||
task_id = await asyncio.wait_for(job_data[JOB_PENDING_TASKS].get(), timeout=1.0)
|
||||
return task_id
|
||||
except asyncio.TimeoutError:
|
||||
return None
|
||||
|
||||
async def _drain_results_queue(multi_job_id):
|
||||
"""Drain pending results from queue and update completed_tasks. Returns count drained.
|
||||
|
||||
Uses non-blocking get_nowait to avoid await timeouts and reduce latency.
|
||||
"""
|
||||
"""Drain pending results from queue and update completed_tasks. Returns count drained."""
|
||||
prompt_server = ensure_tile_jobs_initialized()
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs.get(multi_job_id)
|
||||
@@ -217,46 +120,45 @@ async def _drain_results_queue(multi_job_id):
|
||||
completed_tasks = job_data[JOB_COMPLETED_TASKS]
|
||||
|
||||
collected = 0
|
||||
while True:
|
||||
while not q.empty():
|
||||
try:
|
||||
result = q.get_nowait()
|
||||
except asyncio.QueueEmpty:
|
||||
break
|
||||
|
||||
worker_id = result['worker_id']
|
||||
is_last = result.get('is_last', False)
|
||||
|
||||
if 'image_idx' in result and 'image' in result:
|
||||
task_id = result['image_idx']
|
||||
if task_id not in completed_tasks:
|
||||
completed_tasks[task_id] = result['image']
|
||||
collected += 1
|
||||
elif 'tiles' in result:
|
||||
for tile_data in result['tiles']:
|
||||
task_id = tile_data.get('global_idx', tile_data['tile_idx'])
|
||||
result = await asyncio.wait_for(q.get(), timeout=0.1)
|
||||
worker_id = result['worker_id']
|
||||
is_last = result.get('is_last', False)
|
||||
|
||||
if 'image_idx' in result and 'image' in result:
|
||||
task_id = result['image_idx']
|
||||
if task_id not in completed_tasks:
|
||||
completed_tasks[task_id] = tile_data
|
||||
completed_tasks[task_id] = result['image']
|
||||
collected += 1
|
||||
elif 'tiles' in result:
|
||||
for tile_data in result['tiles']:
|
||||
task_id = tile_data.get('global_idx', tile_data['tile_idx'])
|
||||
if task_id not in completed_tasks:
|
||||
completed_tasks[task_id] = tile_data
|
||||
collected += 1
|
||||
elif 'tensor' in result and 'tile_idx' in result: # Single tile backward compat
|
||||
task_id = result.get('global_idx', result['tile_idx'])
|
||||
if task_id not in completed_tasks:
|
||||
completed_tasks[task_id] = {
|
||||
'tensor': result['tensor'],
|
||||
'tile_idx': result['tile_idx'],
|
||||
'x': result['x'],
|
||||
'y': result['y'],
|
||||
'extracted_width': result['extracted_width'],
|
||||
'extracted_height': result['extracted_height'],
|
||||
'padding': result['padding'],
|
||||
'batch_idx': result.get('batch_idx', 0),
|
||||
'global_idx': task_id
|
||||
}
|
||||
collected += 1
|
||||
elif 'tensor' in result and 'tile_idx' in result: # Single tile backward compat
|
||||
task_id = result.get('global_idx', result['tile_idx'])
|
||||
if task_id not in completed_tasks:
|
||||
completed_tasks[task_id] = {
|
||||
'tensor': result['tensor'],
|
||||
'tile_idx': result['tile_idx'],
|
||||
'x': result['x'],
|
||||
'y': result['y'],
|
||||
'extracted_width': result['extracted_width'],
|
||||
'extracted_height': result['extracted_height'],
|
||||
'padding': result['padding'],
|
||||
'batch_idx': result.get('batch_idx', 0),
|
||||
'global_idx': task_id
|
||||
}
|
||||
collected += 1
|
||||
|
||||
if is_last:
|
||||
# Track worker completion
|
||||
if worker_id in job_data[JOB_WORKER_STATUS]:
|
||||
del job_data[JOB_WORKER_STATUS][worker_id]
|
||||
if is_last:
|
||||
# Track worker completion
|
||||
if worker_id in job_data[JOB_WORKER_STATUS]:
|
||||
del job_data[JOB_WORKER_STATUS][worker_id]
|
||||
except asyncio.TimeoutError:
|
||||
break
|
||||
|
||||
return collected
|
||||
|
||||
@@ -272,35 +174,13 @@ async def _check_and_requeue_timed_out_workers(multi_job_id, total_tasks):
|
||||
requeued_count = 0
|
||||
completed_tasks = job_data.get(JOB_COMPLETED_TASKS, {})
|
||||
|
||||
# Allow override via config setting 'worker_timeout_seconds'
|
||||
cfg = load_config()
|
||||
hb_timeout = int(cfg.get('settings', {}).get('worker_timeout_seconds', HEARTBEAT_TIMEOUT))
|
||||
|
||||
for worker, last_heartbeat in list(job_data.get(JOB_WORKER_STATUS, {}).items()):
|
||||
if current_time - last_heartbeat > hb_timeout:
|
||||
if current_time - last_heartbeat > HEARTBEAT_TIMEOUT:
|
||||
log(f"Worker {worker} timed out")
|
||||
for task_id in job_data.get(JOB_ASSIGNED_TO_WORKERS, {}).get(worker, []):
|
||||
# If batched_static, task_id is a tile_idx; consider it complete only if
|
||||
# all corresponding global_idx entries are present in completed_tasks.
|
||||
batched_static = bool(job_data.get('batched_static', False))
|
||||
if batched_static:
|
||||
num_tiles_per_image = job_data.get(JOB_NUM_TILES_PER_IMAGE, 1)
|
||||
batch_size = job_data.get(JOB_BATCH_SIZE, 1)
|
||||
# Check all global indices for this tile across the batch
|
||||
all_done = True
|
||||
for b in range(batch_size):
|
||||
gidx = b * num_tiles_per_image + task_id
|
||||
if gidx not in completed_tasks:
|
||||
all_done = False
|
||||
break
|
||||
if not all_done:
|
||||
await job_data[JOB_PENDING_TASKS].put(task_id)
|
||||
requeued_count += 1
|
||||
else:
|
||||
# Legacy/global-idx mode: task_id is a global index key
|
||||
if task_id not in completed_tasks:
|
||||
await job_data[JOB_PENDING_TASKS].put(task_id)
|
||||
requeued_count += 1
|
||||
if task_id not in completed_tasks:
|
||||
await job_data[JOB_PENDING_TASKS].put(task_id)
|
||||
requeued_count += 1
|
||||
if JOB_WORKER_STATUS in job_data:
|
||||
del job_data[JOB_WORKER_STATUS][worker]
|
||||
if JOB_ASSIGNED_TO_WORKERS in job_data:
|
||||
@@ -346,6 +226,42 @@ async def _cleanup_job(multi_job_id):
|
||||
|
||||
# API Endpoints (generalized)
|
||||
|
||||
@server.PromptServer.instance.routes.post("/distributed/request_task")
|
||||
async def request_task_endpoint(request):
|
||||
try:
|
||||
data = await request.json()
|
||||
worker_id = data.get('worker_id')
|
||||
multi_job_id = data.get('multi_job_id')
|
||||
|
||||
if not worker_id or not multi_job_id:
|
||||
return await handle_api_error(request, "Missing worker_id or multi_job_id", 400)
|
||||
|
||||
prompt_server = ensure_tile_jobs_initialized()
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
mode = job_data.get(JOB_MODE)
|
||||
pending_queue = job_data.get(JOB_PENDING_TASKS)
|
||||
if pending_queue:
|
||||
try:
|
||||
task_id = await asyncio.wait_for(pending_queue.get(), timeout=0.1)
|
||||
if JOB_ASSIGNED_TO_WORKERS in job_data and worker_id in job_data[JOB_ASSIGNED_TO_WORKERS]:
|
||||
job_data[JOB_ASSIGNED_TO_WORKERS][worker_id].append(task_id)
|
||||
if JOB_WORKER_STATUS in job_data:
|
||||
job_data[JOB_WORKER_STATUS][worker_id] = time.time()
|
||||
remaining = pending_queue.qsize()
|
||||
debug_log(f"Assigned task {task_id} to worker {worker_id} in {mode} mode")
|
||||
return web.json_response({"task_id": task_id, "estimated_remaining": remaining, "mode": mode})
|
||||
except asyncio.TimeoutError:
|
||||
return web.json_response({"task_id": None})
|
||||
else:
|
||||
return await handle_api_error(request, "No pending tasks", 400)
|
||||
else:
|
||||
return await handle_api_error(request, "Job not found", 404)
|
||||
except Exception as e:
|
||||
return await handle_api_error(request, e, 500)
|
||||
|
||||
|
||||
@server.PromptServer.instance.routes.post("/distributed/heartbeat")
|
||||
async def heartbeat_endpoint(request):
|
||||
try:
|
||||
@@ -398,7 +314,7 @@ async def submit_tiles_endpoint(request):
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_MODE in job_data and job_data[JOB_MODE] != 'static':
|
||||
return await handle_api_error(request, "Job not configured for tile submissions", 400)
|
||||
return await handle_api_error(request, "Mode mismatch: expected static mode", 400)
|
||||
if JOB_QUEUE in job_data:
|
||||
await job_data[JOB_QUEUE].put({
|
||||
'worker_id': worker_id,
|
||||
@@ -408,17 +324,115 @@ async def submit_tiles_endpoint(request):
|
||||
debug_log(f"Received completion signal from worker {worker_id}")
|
||||
return web.json_response({"status": "success"})
|
||||
|
||||
try:
|
||||
tiles = _parse_tiles_from_form(data)
|
||||
except ValueError as e:
|
||||
return await handle_api_error(request, str(e), 400)
|
||||
# Handle batch tiles with metadata
|
||||
if batch_size > 0:
|
||||
padding = int(data.get('padding', 32))
|
||||
metadata_field = data.get('tiles_metadata')
|
||||
if metadata_field:
|
||||
if hasattr(metadata_field, 'file'):
|
||||
metadata_str = metadata_field.file.read().decode('utf-8')
|
||||
elif isinstance(metadata_field, (bytes, bytearray)):
|
||||
metadata_str = metadata_field.decode('utf-8')
|
||||
else:
|
||||
metadata_str = str(metadata_field)
|
||||
|
||||
metadata = json.loads(metadata_str)
|
||||
if len(metadata) != batch_size:
|
||||
return await handle_api_error(request, "Metadata length mismatch", 400)
|
||||
|
||||
tile_data_list = []
|
||||
|
||||
for i in range(batch_size):
|
||||
tile_field = data.get(f'tile_{i}')
|
||||
if tile_field is None:
|
||||
continue
|
||||
|
||||
img_data = tile_field.file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
img_np = np.array(img).astype(np.float32) / 255.0
|
||||
tensor = torch.from_numpy(img_np)[None,]
|
||||
|
||||
if i < len(metadata):
|
||||
tile_meta = metadata[i]
|
||||
tile_idx = tile_meta.get('tile_idx', i)
|
||||
tile_info = {
|
||||
'tensor': tensor,
|
||||
'tile_idx': tile_idx,
|
||||
'x': tile_meta['x'],
|
||||
'y': tile_meta['y'],
|
||||
'extracted_width': tile_meta['extracted_width'],
|
||||
'extracted_height': tile_meta['extracted_height'],
|
||||
'padding': padding,
|
||||
'batch_idx': tile_meta.get('batch_idx', 0),
|
||||
'global_idx': tile_meta.get('global_idx', tile_idx)
|
||||
}
|
||||
tile_data_list.append(tile_info)
|
||||
|
||||
tile_data_list.sort(key=lambda x: x['tile_idx'])
|
||||
tiles.extend(tile_data_list)
|
||||
else:
|
||||
# Legacy format
|
||||
for i in range(batch_size):
|
||||
tile_field = data.get(f'tile_{i}')
|
||||
if tile_field is None:
|
||||
continue
|
||||
|
||||
tile_idx = int(data.get(f'tile_{i}_idx', i))
|
||||
x = int(data.get(f'tile_{i}_x', 0))
|
||||
y = int(data.get(f'tile_{i}_y', 0))
|
||||
extracted_width = int(data.get(f'tile_{i}_width', 512))
|
||||
extracted_height = int(data.get(f'tile_{i}_height', 512))
|
||||
|
||||
img_data = tile_field.file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
img_np = np.array(img).astype(np.float32) / 255.0
|
||||
tensor = torch.from_numpy(img_np)[None,]
|
||||
|
||||
tiles.append({
|
||||
'tensor': tensor,
|
||||
'tile_idx': tile_idx,
|
||||
'x': x,
|
||||
'y': y,
|
||||
'extracted_width': extracted_width,
|
||||
'extracted_height': extracted_height,
|
||||
'padding': padding
|
||||
})
|
||||
else:
|
||||
# Single tile legacy
|
||||
image_file = data.get('image')
|
||||
if not image_file:
|
||||
return await handle_api_error(request, "Missing image data", 400)
|
||||
|
||||
tile_idx = int(data.get('tile_idx', 0))
|
||||
x = int(data.get('x', 0))
|
||||
y = int(data.get('y', 0))
|
||||
extracted_width = int(data.get('extracted_width', 512))
|
||||
extracted_height = int(data.get('extracted_height', 512))
|
||||
padding = int(data.get('padding', 32))
|
||||
|
||||
img_data = image_file.file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
img_np = np.array(img).astype(np.float32) / 255.0
|
||||
tensor = torch.from_numpy(img_np)[None,]
|
||||
|
||||
tiles = [{
|
||||
'tensor': tensor,
|
||||
'tile_idx': tile_idx,
|
||||
'x': x,
|
||||
'y': y,
|
||||
'extracted_width': extracted_width,
|
||||
'extracted_height': extracted_height,
|
||||
'padding': padding,
|
||||
'batch_idx': 0,
|
||||
'global_idx': tile_idx
|
||||
}]
|
||||
|
||||
# Submit tiles to queue
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_MODE in job_data and job_data[JOB_MODE] != 'static':
|
||||
return await handle_api_error(request, "Job not configured for tile submissions", 400)
|
||||
return await handle_api_error(request, "Mode mismatch: expected static mode", 400)
|
||||
|
||||
q = job_data[JOB_QUEUE]
|
||||
if batch_size > 0 or len(tiles) > 0:
|
||||
@@ -427,7 +441,6 @@ async def submit_tiles_endpoint(request):
|
||||
'tiles': tiles,
|
||||
'is_last': is_last
|
||||
})
|
||||
debug_log(f"Received {len(tiles)} tiles from worker {worker_id} (is_last={is_last})")
|
||||
else:
|
||||
await q.put({
|
||||
'worker_id': worker_id,
|
||||
@@ -471,7 +484,7 @@ async def submit_image_endpoint(request):
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_MODE in job_data and job_data[JOB_MODE] != 'dynamic':
|
||||
return await handle_api_error(request, "Job not configured for image submissions", 400)
|
||||
return await handle_api_error(request, "Mode mismatch: expected dynamic mode", 400)
|
||||
if JOB_QUEUE in job_data:
|
||||
await job_data[JOB_QUEUE].put({
|
||||
'worker_id': worker_id,
|
||||
@@ -487,7 +500,7 @@ async def submit_image_endpoint(request):
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_MODE in job_data and job_data[JOB_MODE] != 'dynamic':
|
||||
return await handle_api_error(request, "Job not configured for image submissions", 400)
|
||||
return await handle_api_error(request, "Mode mismatch: expected dynamic mode", 400)
|
||||
if JOB_QUEUE in job_data:
|
||||
await job_data[JOB_QUEUE].put({
|
||||
'worker_id': worker_id,
|
||||
@@ -503,7 +516,188 @@ async def submit_image_endpoint(request):
|
||||
except Exception as e:
|
||||
return await handle_api_error(request, e, 500)
|
||||
|
||||
# Note: Removed legacy /distributed/tile_complete endpoint. Use /distributed/submit_tiles.
|
||||
# Keep legacy endpoint for backward compatibility
|
||||
@server.PromptServer.instance.routes.post("/distributed/tile_complete")
|
||||
async def tile_complete_endpoint(request):
|
||||
try:
|
||||
content_length = request.headers.get('content-length')
|
||||
if content_length and int(content_length) > MAX_PAYLOAD_SIZE:
|
||||
return await handle_api_error(request, f"Payload too large: {content_length} bytes", 413)
|
||||
|
||||
data = await request.post()
|
||||
multi_job_id = data.get('multi_job_id')
|
||||
worker_id = data.get('worker_id')
|
||||
is_last = data.get('is_last', 'False').lower() == 'true'
|
||||
|
||||
if multi_job_id is None or worker_id is None:
|
||||
return await handle_api_error(request, "Missing multi_job_id or worker_id", 400)
|
||||
|
||||
prompt_server = ensure_tile_jobs_initialized()
|
||||
|
||||
if 'full_image' in data and 'image_idx' in data:
|
||||
image_idx = int(data.get('image_idx'))
|
||||
img_data = data['full_image'].file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
|
||||
debug_log(f"Received full image {image_idx} from worker {worker_id}")
|
||||
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_MODE in job_data and job_data[JOB_MODE] != 'dynamic':
|
||||
return await handle_api_error(request, "Mode mismatch for image submission", 400)
|
||||
if JOB_QUEUE in job_data:
|
||||
await job_data[JOB_QUEUE].put({
|
||||
'worker_id': worker_id,
|
||||
'image_idx': image_idx,
|
||||
'image': img,
|
||||
'is_last': is_last
|
||||
})
|
||||
return web.json_response({"status": "success"})
|
||||
|
||||
batch_size = int(data.get('batch_size', 0))
|
||||
tiles = []
|
||||
|
||||
if batch_size == 0 and is_last:
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_QUEUE in job_data:
|
||||
await job_data[JOB_QUEUE].put({
|
||||
'worker_id': worker_id,
|
||||
'is_last': True,
|
||||
'tiles': []
|
||||
})
|
||||
debug_log(f"Received completion signal from worker {worker_id}")
|
||||
return web.json_response({"status": "success"})
|
||||
|
||||
if batch_size > 0:
|
||||
padding = int(data.get('padding', 32))
|
||||
metadata_field = data.get('tiles_metadata')
|
||||
if metadata_field:
|
||||
if hasattr(metadata_field, 'file'):
|
||||
metadata_str = metadata_field.file.read().decode('utf-8')
|
||||
elif isinstance(metadata_field, (bytes, bytearray)):
|
||||
metadata_str = metadata_field.decode('utf-8')
|
||||
else:
|
||||
metadata_str = str(metadata_field)
|
||||
|
||||
metadata = json.loads(metadata_str)
|
||||
if len(metadata) != batch_size:
|
||||
return await handle_api_error(request, "Metadata length mismatch", 400)
|
||||
|
||||
tile_data_list = []
|
||||
|
||||
for i in range(batch_size):
|
||||
tile_field = data.get(f'tile_{i}')
|
||||
if tile_field is None:
|
||||
continue
|
||||
|
||||
img_data = tile_field.file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
img_np = np.array(img).astype(np.float32) / 255.0
|
||||
tensor = torch.from_numpy(img_np)[None,]
|
||||
|
||||
if i < len(metadata):
|
||||
tile_meta = metadata[i]
|
||||
tile_idx = tile_meta.get('tile_idx', i)
|
||||
tile_info = {
|
||||
'tensor': tensor,
|
||||
'tile_idx': tile_idx,
|
||||
'x': tile_meta['x'],
|
||||
'y': tile_meta['y'],
|
||||
'extracted_width': tile_meta['extracted_width'],
|
||||
'extracted_height': tile_meta['extracted_height'],
|
||||
'padding': padding,
|
||||
'batch_idx': tile_meta.get('batch_idx', 0),
|
||||
'global_idx': tile_meta.get('global_idx', tile_idx)
|
||||
}
|
||||
tile_data_list.append(tile_info)
|
||||
|
||||
tile_data_list.sort(key=lambda x: x['tile_idx'])
|
||||
tiles.extend(tile_data_list)
|
||||
else:
|
||||
# Legacy
|
||||
for i in range(batch_size):
|
||||
tile_field = data.get(f'tile_{i}')
|
||||
if tile_field is None:
|
||||
continue
|
||||
|
||||
tile_idx = int(data.get(f'tile_{i}_idx', i))
|
||||
x = int(data.get(f'tile_{i}_x', 0))
|
||||
y = int(data.get(f'tile_{i}_y', 0))
|
||||
extracted_width = int(data.get(f'tile_{i}_width', 512))
|
||||
extracted_height = int(data.get(f'tile_{i}_height', 512))
|
||||
|
||||
img_data = tile_field.file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
img_np = np.array(img).astype(np.float32) / 255.0
|
||||
tensor = torch.from_numpy(img_np)[None,]
|
||||
|
||||
tiles.append({
|
||||
'tensor': tensor,
|
||||
'tile_idx': tile_idx,
|
||||
'x': x,
|
||||
'y': y,
|
||||
'extracted_width': extracted_width,
|
||||
'extracted_height': extracted_height,
|
||||
'padding': padding
|
||||
})
|
||||
else:
|
||||
# Single tile legacy
|
||||
image_file = data.get('image')
|
||||
if not image_file:
|
||||
return await handle_api_error(request, "Missing image data", 400)
|
||||
|
||||
tile_idx = int(data.get('tile_idx', 0))
|
||||
x = int(data.get('x', 0))
|
||||
y = int(data.get('y', 0))
|
||||
extracted_width = int(data.get('extracted_width', 512))
|
||||
extracted_height = int(data.get('extracted_height', 512))
|
||||
padding = int(data.get('padding', 32))
|
||||
|
||||
img_data = image_file.file.read()
|
||||
img = Image.open(io.BytesIO(img_data)).convert("RGB")
|
||||
img_np = np.array(img).astype(np.float32) / 255.0
|
||||
tensor = torch.from_numpy(img_np)[None,]
|
||||
|
||||
tiles = [{
|
||||
'tensor': tensor,
|
||||
'tile_idx': tile_idx,
|
||||
'x': x,
|
||||
'y': y,
|
||||
'extracted_width': extracted_width,
|
||||
'extracted_height': extracted_height,
|
||||
'padding': padding,
|
||||
'batch_idx': 0,
|
||||
'global_idx': tile_idx
|
||||
}]
|
||||
|
||||
async with prompt_server.distributed_tile_jobs_lock:
|
||||
if multi_job_id in prompt_server.distributed_pending_tile_jobs:
|
||||
job_data = prompt_server.distributed_pending_tile_jobs[multi_job_id]
|
||||
if JOB_MODE in job_data and job_data[JOB_MODE] != 'static':
|
||||
return await handle_api_error(request, "Mode mismatch for tile submission", 400)
|
||||
|
||||
q = job_data[JOB_QUEUE]
|
||||
if batch_size > 0 or len(tiles) > 0:
|
||||
await q.put({
|
||||
'worker_id': worker_id,
|
||||
'tiles': tiles,
|
||||
'is_last': is_last
|
||||
})
|
||||
else:
|
||||
await q.put({
|
||||
'worker_id': worker_id,
|
||||
'is_last': True,
|
||||
'tiles': []
|
||||
})
|
||||
|
||||
return web.json_response({"status": "success"})
|
||||
else:
|
||||
return await handle_api_error(request, "Job not found", 404)
|
||||
except Exception as e:
|
||||
return await handle_api_error(request, e, 500)
|
||||
|
||||
|
||||
|
||||
@@ -514,8 +708,7 @@ def clone_control_chain(control, clone_hint=True):
|
||||
return None
|
||||
new_control = copy.copy(control) # Shallow copy (shares model)
|
||||
if clone_hint and hasattr(control, 'cond_hint_original'):
|
||||
hint = getattr(control, 'cond_hint_original', None)
|
||||
new_control.cond_hint_original = hint.clone() if hint is not None else None
|
||||
new_control.cond_hint_original = control.cond_hint_original.clone()
|
||||
if hasattr(control, 'previous_controlnet'):
|
||||
new_control.previous_controlnet = clone_control_chain(control.previous_controlnet, clone_hint)
|
||||
return new_control
|
||||
@@ -529,12 +722,10 @@ def clone_conditioning(cond_list, clone_hints=True):
|
||||
if 'control' in new_dict:
|
||||
new_dict['control'] = clone_control_chain(new_dict['control'], clone_hints)
|
||||
if 'mask' in new_dict:
|
||||
if new_dict['mask'] is not None:
|
||||
new_dict['mask'] = new_dict['mask'].clone()
|
||||
new_dict['mask'] = new_dict['mask'].clone()
|
||||
# Handle other potential fields if needed
|
||||
if 'pooled_output' in new_dict:
|
||||
if new_dict['pooled_output'] is not None:
|
||||
new_dict['pooled_output'] = new_dict['pooled_output'].clone()
|
||||
new_dict['pooled_output'] = new_dict['pooled_output'].clone()
|
||||
if 'area' in new_dict:
|
||||
new_dict['area'] = new_dict['area'][:] # Shallow copy list/tuple
|
||||
new_cond.append([new_emb, new_dict])
|
||||
@@ -585,7 +776,7 @@ async def request_image_endpoint(request):
|
||||
elif mode == 'static' and JOB_PENDING_TASKS in job_data:
|
||||
pending_queue = job_data[JOB_PENDING_TASKS]
|
||||
else:
|
||||
return await handle_api_error(request, "Invalid job configuration", 400)
|
||||
return await handle_api_error(request, f"Invalid {mode} mode configuration", 400)
|
||||
|
||||
try:
|
||||
task_idx = await asyncio.wait_for(pending_queue.get(), timeout=0.1)
|
||||
@@ -604,7 +795,7 @@ async def request_image_endpoint(request):
|
||||
return web.json_response({"image_idx": task_idx, "estimated_remaining": remaining})
|
||||
else: # static
|
||||
debug_log(f"UltimateSDUpscale API - Assigned tile {task_idx} to worker {worker_id}")
|
||||
return web.json_response({"tile_idx": task_idx, "estimated_remaining": remaining, "batched_static": job_data.get('batched_static', False)})
|
||||
return web.json_response({"tile_idx": task_idx, "estimated_remaining": remaining})
|
||||
except asyncio.TimeoutError:
|
||||
if mode == 'dynamic':
|
||||
return web.json_response({"image_idx": None})
|
||||
|
||||
+1
-56
@@ -4,6 +4,7 @@ import torch
|
||||
import torch.nn.functional as F
|
||||
from torchvision.transforms import GaussianBlur
|
||||
import math
|
||||
import os
|
||||
|
||||
if (not hasattr(Image, 'Resampling')): # For older versions of Pillow
|
||||
Image.Resampling = Image
|
||||
@@ -441,61 +442,6 @@ def crop_mask(cond_dict, region, init_size, canvas_size, tile_size, w_pad, h_pad
|
||||
|
||||
cond_dict["mask"] = torch.cat(masks, dim=0) # (B, H, W)
|
||||
|
||||
# Added Flux-Kontext Support crop_reference_latents by TBG ETUR
|
||||
def crop_reference_latents(cond_dict, region, init_size, canvas_size, tile_size, w_pad, h_pad):
|
||||
"""
|
||||
1. Resize each latent to `canvas_size` in latent units.
|
||||
2. Crop the rectangle `region` (pixel coordinates).
|
||||
3. Down-sample the crop to latent-space `tile_size`.
|
||||
Expects a list of BCHW tensors under "reference_latents".
|
||||
"""
|
||||
|
||||
latents = cond_dict.get("reference_latents")
|
||||
if not isinstance(latents, list):
|
||||
return # nothing to do
|
||||
|
||||
k = 8 # down-sample factor from pixel space → latent space (SD-type models)
|
||||
|
||||
W_can_px, H_can_px = canvas_size
|
||||
# canvas size expressed in latent units
|
||||
W_can_lat, H_can_lat = W_can_px // k, H_can_px // k
|
||||
|
||||
W_tile_px, H_tile_px = tile_size
|
||||
W_tile_lat, H_tile_lat = max(1, W_tile_px // k), max(1, H_tile_px // k)
|
||||
|
||||
x1_px, y1_px, x2_px, y2_px = region
|
||||
|
||||
new_latents = []
|
||||
for t in latents: # (B,C,H_lat_in,W_lat_in)
|
||||
if t.ndim != 4:
|
||||
raise ValueError(f"expected BCHW, got {t.shape}")
|
||||
|
||||
# 1. Resize to canvas resolution in latent units only if needed
|
||||
if t.shape[-2:] != (H_can_lat, W_can_lat):
|
||||
t = F.interpolate(t,
|
||||
size=(H_can_lat, W_can_lat),
|
||||
mode="bilinear",
|
||||
align_corners=False)
|
||||
|
||||
# 2. Convert pixel crop → latent slice
|
||||
w0_lat = int(round(x1_px / k))
|
||||
w1_lat = int(round(x2_px / k))
|
||||
h0_lat = int(round(y1_px / k))
|
||||
h1_lat = int(round(y2_px / k))
|
||||
|
||||
cropped = t[:, :, h0_lat:h1_lat, w0_lat:w1_lat] # view
|
||||
|
||||
# 3. Down-sample to latent-tile size
|
||||
cropped = F.interpolate(cropped,
|
||||
size=(H_tile_lat, W_tile_lat),
|
||||
mode="bilinear",
|
||||
align_corners=False)
|
||||
|
||||
new_latents.append(cropped)
|
||||
|
||||
cond_dict["reference_latents"] = new_latents
|
||||
|
||||
|
||||
|
||||
def crop_cond(cond, region, init_size, canvas_size, tile_size, w_pad=0, h_pad=0):
|
||||
cropped = []
|
||||
@@ -506,6 +452,5 @@ def crop_cond(cond, region, init_size, canvas_size, tile_size, w_pad=0, h_pad=0)
|
||||
crop_gligen(cond_dict, region, init_size, canvas_size, tile_size, w_pad, h_pad)
|
||||
crop_area(cond_dict, region, init_size, canvas_size, tile_size, w_pad, h_pad)
|
||||
crop_mask(cond_dict, region, init_size, canvas_size, tile_size, w_pad, h_pad)
|
||||
crop_reference_latents(cond_dict, region, init_size, canvas_size, tile_size, w_pad, h_pad)
|
||||
cropped.append(n)
|
||||
return cropped
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 4.7 KiB |
+6
-18
@@ -19,25 +19,13 @@ function convertPathsForPlatform(apiPrompt, targetSeparator) {
|
||||
const isLikelyFilename = (value) => {
|
||||
return value.match(/\.(ckpt|safetensors|pt|pth|bin|yaml|json|png|jpg|jpeg|webp|gif|bmp|latent|txt|vae|lora|embedding)(\s*\[\w+\])?$/i);
|
||||
};
|
||||
const isImageOrVideo = (value) => {
|
||||
return value.match(/\.(png|jpg|jpeg|webp|gif|bmp|mp4|avi|mov|mkv|webm)(\s*\[\w+\])?$/i);
|
||||
};
|
||||
|
||||
function convert(obj) {
|
||||
if (typeof obj === 'string') {
|
||||
// Only convert strings that look like file paths
|
||||
if ((obj.includes('\\') || obj.includes('/')) && isLikelyFilename(obj)) {
|
||||
const trimmed = obj.trim();
|
||||
const hasDrive = /^[A-Za-z]:\\\\|^[A-Za-z]:\//.test(trimmed);
|
||||
const isAbsolute = trimmed.startsWith('/') || trimmed.startsWith('\\\\');
|
||||
const hasProtocol = /^\w+:\/\//.test(trimmed);
|
||||
|
||||
// For annotated relative image/video paths, keep forward slashes
|
||||
if (!hasDrive && !isAbsolute && !hasProtocol && isImageOrVideo(trimmed)) {
|
||||
return trimmed.replace(/[\\\\]/g, '/');
|
||||
}
|
||||
// Otherwise replace any path separator with the worker's target separator
|
||||
return trimmed.replace(/[\\\\\/]/g, targetSeparator);
|
||||
// Replace any path separator with the target one
|
||||
return obj.replace(/[\\\/]/g, targetSeparator);
|
||||
}
|
||||
return obj;
|
||||
} else if (Array.isArray(obj)) {
|
||||
@@ -519,9 +507,9 @@ export async function uploadImagesToWorker(extension, workerUrl, images) {
|
||||
let cleanName = imageData.name;
|
||||
let subfolder = '';
|
||||
|
||||
// Extract subfolder if present (handle both slash styles)
|
||||
if (cleanName.includes('/') || cleanName.includes('\\')) {
|
||||
const parts = cleanName.replace(/\\/g, '/').split('/');
|
||||
// Extract subfolder if present
|
||||
if (cleanName.includes('/')) {
|
||||
const parts = cleanName.split('/');
|
||||
subfolder = parts.slice(0, -1).join('/');
|
||||
cleanName = parts[parts.length - 1];
|
||||
}
|
||||
@@ -599,4 +587,4 @@ export async function performPreflightCheck(extension, workers) {
|
||||
});
|
||||
|
||||
return activeWorkers;
|
||||
}
|
||||
}
|
||||
+17
-2
@@ -217,7 +217,22 @@ class DistributedExtension {
|
||||
this.panelElement = null;
|
||||
}
|
||||
|
||||
// updateSummary removed
|
||||
updateSummary() {
|
||||
const summaryEl = document.getElementById('distributed-summary');
|
||||
if (summaryEl) {
|
||||
// Count active workers from state
|
||||
const activeWorkers = this.config.workers.filter(w => {
|
||||
const status = this.state.getWorkerStatus(w.id);
|
||||
return status.online;
|
||||
}).length;
|
||||
const totalGPUs = activeWorkers + 1;
|
||||
if (this.isEnabled) {
|
||||
summaryEl.textContent = `If Collector node is present, total generation = (${totalGPUs} GPUs × Batch Size)`;
|
||||
} else {
|
||||
summaryEl.textContent = "Only the master GPU will be used.";
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// --- Core Logic & Execution ---
|
||||
|
||||
@@ -1318,4 +1333,4 @@ app.registerExtension({
|
||||
async setup() {
|
||||
new DistributedExtension();
|
||||
}
|
||||
});
|
||||
});
|
||||
+31
-71
@@ -148,67 +148,57 @@ export async function renderSidebarContent(extension, el) {
|
||||
|
||||
// Settings section
|
||||
const settingsSection = document.createElement("div");
|
||||
// Top separator only; spacing handled by the clickable toggle area for equal top/bottom spacing
|
||||
settingsSection.style.cssText = "border-top: 1px solid #444; margin-bottom: 10px;";
|
||||
settingsSection.style.cssText = "border-top: 1px solid #444; padding-top: 10px; margin-bottom: 10px;";
|
||||
|
||||
// Settings header with toggle (full-area clickable between separators)
|
||||
const settingsToggleArea = document.createElement("div");
|
||||
// Equal spacing above header (to top separator) and below header (to bottom separator)
|
||||
settingsToggleArea.style.cssText = "padding: 16.5px 0; cursor: pointer; user-select: none;";
|
||||
|
||||
// Settings header with toggle
|
||||
const settingsHeader = document.createElement("div");
|
||||
settingsHeader.style.cssText = "display: flex; align-items: center; justify-content: space-between;";
|
||||
settingsHeader.style.cssText = "display: flex; align-items: center; justify-content: space-between; cursor: pointer; user-select: none;";
|
||||
|
||||
const workerSettingsTitle = document.createElement("h4");
|
||||
workerSettingsTitle.textContent = "Settings";
|
||||
workerSettingsTitle.style.cssText = "margin: 0; font-size: 14px;";
|
||||
|
||||
const workerSettingsToggle = document.createElement("span");
|
||||
workerSettingsToggle.textContent = "▶"; // Right arrow when collapsed
|
||||
workerSettingsToggle.style.cssText = "font-size: 12px; color: #888; transition: all 0.2s ease;";
|
||||
|
||||
settingsHeader.appendChild(workerSettingsTitle);
|
||||
settingsHeader.appendChild(workerSettingsToggle);
|
||||
settingsToggleArea.appendChild(settingsHeader);
|
||||
// Hover effect for toggle area
|
||||
settingsToggleArea.onmouseover = () => { workerSettingsToggle.style.color = "#fff"; };
|
||||
settingsToggleArea.onmouseout = () => { workerSettingsToggle.style.color = "#888"; };
|
||||
|
||||
// A small separator shown only when collapsed (to make the section boundary obvious)
|
||||
const settingsSeparator = document.createElement("div");
|
||||
// No margin so the bottom spacing is controlled by settingsToggleArea padding-bottom
|
||||
settingsSeparator.style.cssText = "border-bottom: 1px solid #444; margin: 0;";
|
||||
|
||||
// Hover effect for header
|
||||
settingsHeader.onmouseover = () => {
|
||||
workerSettingsToggle.style.color = "#fff";
|
||||
};
|
||||
settingsHeader.onmouseout = () => {
|
||||
workerSettingsToggle.style.color = "#888";
|
||||
};
|
||||
|
||||
// Collapsible settings content
|
||||
const settingsContent = document.createElement("div");
|
||||
settingsContent.style.cssText = "max-height: 0; overflow: hidden; opacity: 0; transition: max-height 0.3s ease, opacity 0.3s ease;";
|
||||
|
||||
const settingsDiv = document.createElement("div");
|
||||
settingsDiv.style.cssText = "display: grid; grid-template-columns: 1fr auto; row-gap: 10px; column-gap: 10px; padding-top: 10px; align-items: center;";
|
||||
settingsDiv.style.cssText = "display: flex; flex-direction: column; gap: 8px; padding-top: 10px;";
|
||||
|
||||
// Toggle functionality
|
||||
let settingsExpanded = false;
|
||||
settingsToggleArea.onclick = () => {
|
||||
settingsHeader.onclick = () => {
|
||||
settingsExpanded = !settingsExpanded;
|
||||
if (settingsExpanded) {
|
||||
settingsContent.style.maxHeight = "200px";
|
||||
settingsContent.style.opacity = "1";
|
||||
workerSettingsToggle.style.transform = "rotate(90deg)";
|
||||
settingsSeparator.style.display = "none";
|
||||
} else {
|
||||
settingsContent.style.maxHeight = "0";
|
||||
settingsContent.style.opacity = "0";
|
||||
workerSettingsToggle.style.transform = "rotate(0deg)";
|
||||
settingsSeparator.style.display = "block";
|
||||
}
|
||||
};
|
||||
|
||||
// Debug mode setting
|
||||
// Section: General
|
||||
const generalLabel = document.createElement("div");
|
||||
generalLabel.textContent = "GENERAL";
|
||||
generalLabel.style.cssText = "grid-column: 1 / span 2; font-size: 11px; color: #888; letter-spacing: 0.06em; padding-top: 2px;";
|
||||
|
||||
const debugGroup = document.createElement("div");
|
||||
debugGroup.style.cssText = "grid-column: 1 / span 2; display: flex; align-items: center; gap: 8px;";
|
||||
|
||||
debugGroup.style.cssText = "display: flex; align-items: center; gap: 8px;";
|
||||
|
||||
const debugCheckbox = document.createElement("input");
|
||||
debugCheckbox.type = "checkbox";
|
||||
debugCheckbox.id = "setting-debug";
|
||||
@@ -219,14 +209,13 @@ export async function renderSidebarContent(extension, el) {
|
||||
debugLabel.htmlFor = "setting-debug";
|
||||
debugLabel.textContent = "Debug Mode";
|
||||
debugLabel.style.cssText = "font-size: 12px; color: #ccc; cursor: pointer;";
|
||||
debugLabel.title = "Enable verbose logging in the browser console.";
|
||||
|
||||
debugGroup.appendChild(debugCheckbox);
|
||||
debugGroup.appendChild(debugLabel);
|
||||
|
||||
// Auto-launch workers setting
|
||||
const autoLaunchGroup = document.createElement("div");
|
||||
autoLaunchGroup.style.cssText = "grid-column: 1 / span 2; display: flex; align-items: center; gap: 8px;";
|
||||
autoLaunchGroup.style.cssText = "display: flex; align-items: center; gap: 8px;";
|
||||
|
||||
const autoLaunchCheckbox = document.createElement("input");
|
||||
autoLaunchCheckbox.type = "checkbox";
|
||||
@@ -238,14 +227,13 @@ export async function renderSidebarContent(extension, el) {
|
||||
autoLaunchLabel.htmlFor = "setting-auto-launch";
|
||||
autoLaunchLabel.textContent = "Auto-launch Local Workers on Startup";
|
||||
autoLaunchLabel.style.cssText = "font-size: 12px; color: #ccc; cursor: pointer;";
|
||||
autoLaunchLabel.title = "Start local worker processes automatically when the master starts.";
|
||||
|
||||
autoLaunchGroup.appendChild(autoLaunchCheckbox);
|
||||
autoLaunchGroup.appendChild(autoLaunchLabel);
|
||||
|
||||
// Stop workers on exit setting (under General)
|
||||
// Stop workers on exit setting
|
||||
const stopOnExitGroup = document.createElement("div");
|
||||
stopOnExitGroup.style.cssText = "grid-column: 1 / span 2; display: flex; align-items: center; gap: 8px;";
|
||||
stopOnExitGroup.style.cssText = "display: flex; align-items: center; gap: 8px;";
|
||||
|
||||
const stopOnExitCheckbox = document.createElement("input");
|
||||
stopOnExitCheckbox.type = "checkbox";
|
||||
@@ -257,56 +245,28 @@ export async function renderSidebarContent(extension, el) {
|
||||
stopOnExitLabel.htmlFor = "setting-stop-on-exit";
|
||||
stopOnExitLabel.textContent = "Stop Local Workers on Master Exit";
|
||||
stopOnExitLabel.style.cssText = "font-size: 12px; color: #ccc; cursor: pointer;";
|
||||
stopOnExitLabel.title = "Stop local worker processes automatically when the master exits.";
|
||||
|
||||
stopOnExitGroup.appendChild(stopOnExitCheckbox);
|
||||
stopOnExitGroup.appendChild(stopOnExitLabel);
|
||||
|
||||
settingsDiv.appendChild(generalLabel);
|
||||
settingsDiv.appendChild(debugGroup);
|
||||
settingsDiv.appendChild(autoLaunchGroup);
|
||||
settingsDiv.appendChild(stopOnExitGroup);
|
||||
|
||||
// Worker Timeout setting (seconds)
|
||||
// Section: Timeouts
|
||||
const timeoutsLabel = document.createElement("div");
|
||||
timeoutsLabel.textContent = "TIMEOUTS";
|
||||
timeoutsLabel.style.cssText = "grid-column: 1 / span 2; font-size: 11px; color: #888; letter-spacing: 0.06em; padding-top: 4px;";
|
||||
|
||||
const timeoutGroup = document.createElement("div");
|
||||
timeoutGroup.style.cssText = "grid-column: 1 / span 2; display: flex; align-items: center; gap: 6px;";
|
||||
|
||||
const timeoutLabel = document.createElement("label");
|
||||
timeoutLabel.htmlFor = "setting-worker-timeout";
|
||||
timeoutLabel.textContent = "Worker Timeout";
|
||||
timeoutLabel.style.cssText = "font-size: 12px; color: #ccc; cursor: default;";
|
||||
timeoutLabel.title = "Seconds without a heartbeat before a worker is considered timed out and its tasks are requeued. Default 60. For WAN, consider 300–600.";
|
||||
|
||||
const timeoutInput = document.createElement("input");
|
||||
timeoutInput.type = "number";
|
||||
timeoutInput.id = "setting-worker-timeout";
|
||||
timeoutInput.min = "10";
|
||||
timeoutInput.step = "1";
|
||||
timeoutInput.style.cssText = "width: 80px; padding: 2px 6px; background: #222; color: #ddd; border: 1px solid #333; border-radius: 3px;";
|
||||
timeoutInput.value = (extension.config?.settings?.worker_timeout_seconds ?? 60);
|
||||
timeoutInput.onchange = (e) => {
|
||||
const v = parseInt(e.target.value, 10);
|
||||
if (!Number.isFinite(v) || v <= 0) return;
|
||||
extension._updateSetting('worker_timeout_seconds', v);
|
||||
};
|
||||
|
||||
timeoutGroup.appendChild(timeoutLabel);
|
||||
timeoutGroup.appendChild(timeoutInput);
|
||||
settingsDiv.appendChild(timeoutsLabel);
|
||||
settingsDiv.appendChild(timeoutGroup);
|
||||
settingsContent.appendChild(settingsDiv);
|
||||
|
||||
settingsSection.appendChild(settingsToggleArea);
|
||||
settingsSection.appendChild(settingsSeparator);
|
||||
settingsSection.appendChild(settingsHeader);
|
||||
settingsSection.appendChild(settingsContent);
|
||||
container.appendChild(settingsSection);
|
||||
|
||||
const summarySection = document.createElement("div");
|
||||
summarySection.style.cssText = "border-top: 1px solid #444; padding-top: 10px;";
|
||||
const summary = document.createElement("div");
|
||||
summary.id = "distributed-summary";
|
||||
summary.style.cssText = "font-size: 11px; color: #888;";
|
||||
summarySection.appendChild(summary);
|
||||
container.appendChild(summarySection);
|
||||
el.appendChild(container);
|
||||
extension.updateSummary();
|
||||
|
||||
// Start checking worker statuses immediately in parallel
|
||||
setTimeout(() => extension.checkAllWorkerStatuses(), 0);
|
||||
@@ -314,4 +274,4 @@ export async function renderSidebarContent(extension, el) {
|
||||
// Always reset the rendering flag
|
||||
extension._isRendering = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -697,7 +697,7 @@ export class DistributedUI {
|
||||
// Create powered by Runpod text (initially hidden)
|
||||
const runpodText = document.createElement("a");
|
||||
runpodText.id = `runpod-text-${worker.id}`;
|
||||
runpodText.href = "https://github.com/robertvoy/ComfyUI-Distributed/blob/main/docs/worker-setup-guides.md#cloud-workers";
|
||||
runpodText.href = "https://get.runpod.io/0bw29uf3ug0p";
|
||||
runpodText.target = "_blank";
|
||||
runpodText.textContent = "Deploy Cloud Worker with Runpod";
|
||||
runpodText.style.cssText = "font-size: 12px; color: #4a90e2; text-decoration: none; margin-top: 4px; display: none; cursor: pointer;";
|
||||
@@ -1233,4 +1233,4 @@ export class DistributedUI {
|
||||
|
||||
return card;
|
||||
}
|
||||
}
|
||||
}
|
||||
+3
-5
@@ -105,15 +105,13 @@ export function findImageReferences(extension, apiPrompt) {
|
||||
if (typeof mediaValue === 'string') {
|
||||
// Clean special suffixes like [input] or [output]
|
||||
const cleanValue = mediaValue.replace(/\s*\[\w+\]$/, '').trim();
|
||||
// Normalize to forward slashes so subfolder/filename derivation is consistent on Windows
|
||||
const normalizedValue = cleanValue.replace(/\\/g, '/');
|
||||
if (imageExtensions.test(normalizedValue)) {
|
||||
images.set(normalizedValue, {
|
||||
if (imageExtensions.test(cleanValue)) {
|
||||
images.set(cleanValue, {
|
||||
nodeId,
|
||||
nodeType: node.class_type,
|
||||
inputName: 'image' // Keep as 'image' for compatibility
|
||||
});
|
||||
extension.log(`Found media reference: ${normalizedValue} in node ${nodeId} (${node.class_type})`, "debug");
|
||||
extension.log(`Found media reference: ${cleanValue} in node ${nodeId} (${node.class_type})`, "debug");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user