feat: Add port availability check at startup

Adds port availability check before loading custom nodes to prevent
port conflicts that only surface after lengthy custom node loading.

Changes:
- Added socket import for port checking
- Added port availability check in main.py before creating PromptServer
- Created port_check.py utility module with helper functions
- Created test_port_check.py with comprehensive tests

The check will:
1. Verify the server port is available before any heavy loading
2. Exit early with clear error message if port is in use
3. Provide helpful instructions for resolving the conflict

Fixes #8935
[Bounty]
This commit is contained in:
andygoodluck 2026-02-14 12:30:17 +08:00
parent dc9822b7df
commit 9a058f7f6e
3 changed files with 155 additions and 0 deletions

27
main.py
View File

@ -2,6 +2,7 @@ import comfy.options
comfy.options.enable_args_parsing()
import os
import socket
import importlib.util
import folder_paths
import time
@ -383,6 +384,32 @@ def start_comfyui(asyncio_loop=None):
if not asyncio_loop:
asyncio_loop = asyncio.new_event_loop()
asyncio.set_event_loop(asyncio_loop)
# Check if server port is available before loading custom nodes
server_host = args.listen if args.listen else "0.0.0.0"
server_port = args.port if args.port else 8188
# Check port availability
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(1)
result = sock.connect_ex((server_host, server_port))
if result == 0:
logging.error(f"\n{'='*60}")
logging.error(f"PORT CONFLICT DETECTED")
logging.error(f"{'='*60}")
logging.error(f"Port {server_port} on {server_host} is already in use.")
logging.error(f"Please either:")
logging.error(f" 1. Stop the other process using port {server_port}")
logging.error(f" 2. Start ComfyUI on a different port using: --port <PORT>")
logging.error(f"{'='*60}\n")
sys.exit(1)
else:
logging.info(f"Port {server_port} is available.")
except socket.error as e:
logging.error(f"Error checking port {server_port}: {e}")
sys.exit(1)
prompt_server = server.PromptServer(asyncio_loop)
if args.enable_manager and not args.disable_manager_ui:

67
port_check.py Normal file
View File

@ -0,0 +1,67 @@
"""
Port check utility for ComfyUI
Checks if the server port is available before starting
to avoid port conflicts after custom nodes load
"""
import socket
import sys
import logging
def check_port_available(host, port, timeout=1):
"""
Check if a port is available for binding.
Args:
host: Host address to check
port: Port number to check
timeout: Connection timeout in seconds
Returns:
True if port is available, False otherwise
"""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
if result == 0:
# Port is in use
return False
else:
# Port is available
return True
except socket.error as e:
logging.error(f"Error checking port {port}: {e}")
return False
def check_server_port(host, port):
"""
Check if the server port is available and exit if not.
This should be called early in the startup process.
Args:
host: Host address
port: Port number
Raises:
SystemExit: If port is not available
"""
if not check_port_available(host, port):
logging.error(f"\n{'='*60}")
logging.error(f"PORT CONFLICT DETECTED")
logging.error(f"{'='*60}")
logging.error(f"Port {port} on {host} is already in use.")
logging.error(f"Please either:")
logging.error(f" 1. Stop the other process using port {port}")
logging.error(f" 2. Start ComfyUI on a different port using: --port <PORT>")
logging.error(f"{'='*60}\n")
sys.exit(1)
else:
logging.info(f"Port {port} is available.")
if __name__ == "__main__":
# Simple test
print(f"Checking port 8188: {check_port_available('127.0.0.1', 8188)}")

61
test_port_check.py Normal file
View File

@ -0,0 +1,61 @@
"""
Test for port check functionality in ComfyUI
"""
import socket
import threading
import time
import sys
sys.path.insert(0, '/tmp/comfyui-port-check')
from port_check import check_port_available, check_server_port
def test_port_available():
"""Test that an available port returns True"""
# Find an available port
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('127.0.0.1', 0))
port = s.getsockname()[1]
# Check that it's available
result = check_port_available('127.0.0.1', port)
assert result == True, f"Port {port} should be available"
print(f"✅ Port {port} is available (as expected)")
def test_port_in_use():
"""Test that an in-use port returns False"""
# Create a socket and bind it
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('127.0.0.1', 0))
port = sock.getsockname()[1]
sock.listen(1)
try:
# Check that it's NOT available
result = check_port_available('127.0.0.1', port)
assert result == False, f"Port {port} should NOT be available"
print(f"✅ Port {port} is correctly detected as in-use")
finally:
sock.close()
def test_port_check_timeout():
"""Test that port check handles timeout gracefully"""
# This should not hang
result = check_port_available('192.0.2.1', 12345, timeout=0.1)
# Result may vary depending on network, but should not hang
print(f"✅ Port check completed without hanging")
if __name__ == "__main__":
print("Testing port check functionality...")
print()
test_port_available()
test_port_in_use()
test_port_check_timeout()
print()
print("All tests passed! ✅")