| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| import ipaddress |
| import socket |
|
|
|
|
| def is_ipv4(ip_str: str) -> bool: |
| """ |
| Check if the given string is an IPv4 address |
| |
| Args: |
| ip_str: The IP address string to check |
| |
| Returns: |
| bool: Returns True if it's an IPv4 address, False otherwise |
| """ |
| try: |
| ipaddress.IPv4Address(ip_str) |
| return True |
| except ipaddress.AddressValueError: |
| return False |
|
|
|
|
| def is_ipv6(ip_str: str) -> bool: |
| """ |
| Check if the given string is an IPv6 address |
| |
| Args: |
| ip_str: The IP address string to check |
| |
| Returns: |
| bool: Returns True if it's an IPv6 address, False otherwise |
| """ |
| try: |
| ipaddress.IPv6Address(ip_str) |
| return True |
| except ipaddress.AddressValueError: |
| return False |
|
|
|
|
| def is_valid_ipv6_address(address: str) -> bool: |
| try: |
| ipaddress.IPv6Address(address) |
| return True |
| except ValueError: |
| return False |
|
|
|
|
| def get_free_port(address: str, with_alive_sock: bool = False) -> tuple[int, socket.socket | None]: |
| """Find a free port on the given address. |
| |
| By default the socket is closed internally, suitable for immediate use. |
| Set with_alive_sock=True to keep the socket open as a port reservation, |
| preventing other calls from getting the same port. The caller is |
| responsible for closing the socket before the port is actually bound |
| by the target service (e.g. NCCL, uvicorn). |
| """ |
| family = socket.AF_INET6 if is_valid_ipv6_address(address) else socket.AF_INET |
|
|
| sock = socket.socket(family=family, type=socket.SOCK_STREAM) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind((address, 0)) |
| port = sock.getsockname()[1] |
| if with_alive_sock: |
| return port, sock |
| sock.close() |
| return port, None |
|
|
