if nvdocker: # Define device requests for NVIDIA GPUs device_requests += [ docker.types.DeviceRequest( count=-1, # -1 specifies all available GPUs capabilities=[['gpu']], # This is the equivalent of `--gpus all` driver='nvidia' ) ]
nvidia-cuda: # for testing with GPU support image:nvidia/cuda:11.0.3-base-ubuntu20.04 container_name:d-gui-cuda entrypoint: ["echo", "CUDA image ready"]
try: # Run a container that utilizes GPU container = client.containers.run( test_image, command=test_command, runtime='nvidia', # Specify the NVIDIA runtime detach=True, auto_remove=True, # Remove container after execution )
# Fetch logs to check if nvidia-smi command was successful #logs = container.logs() #print(logs.decode('utf-8')) # Optional: Print output for verification
# If the command was successful, NVIDIA Docker is available returnTrue except (APIError, ContainerError) as e: # If there was an error, log it and return False print(f"Error checking NVIDIA Docker availability: {e}") returnFalse finally: # Cleanup: Stop container if it's still running try: container.stop() except Exception: pass# If container doesn't exist or is already removed, ignore
defcheck_port_in_use(port) -> bool: # check if port is in use in the host with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: # If connect_ex returns 0, the port is in use return s.connect_ex(('host.docker.internal', port)) == 0
defis_port_used_by_container(port:int) -> bool: client = docker.from_env() for container in client.containers.list(all=True): port_bindings:dict = client.api.inspect_container(container.id)['HostConfig'].get('PortBindings', {}) # 關鍵是這邊直接inspect去看container設定的port ports = parse_ports(port_bindings) ifstr(port) in ports.values(): returnTrue returnFalse
最後一件事!
我們完成前面兩件事之後,就可以從最小的port開始往上找,找到沒被使用的port
這邊要注意的是,找到所有的port非常耗時(畢竟是使用socket去測試)
所以我們要定義數量,找到足夠的port就可以停止了
原始碼可以參照find_multiple_free_ports.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
deffind_multiple_free_ports(count): # find multiple free ports in the host
base_port = 1024# Start scanning from port 1024 free_ports = []
whilelen(free_ports) < count and base_port < 65535: ifnot check_port_in_use(base_port) andnot is_port_used_by_container(base_port): # find port is not in use in the host free_ports.append(base_port) base_port += 1