Skip to content
Advertisement

How to pick a free port for a subprocess?

I am writing a Python wrapper around Appium server. Appium accepts command-line parameter for a local port to bind to. Unfortunately, Appium cannot autoselect a free port for itself, so it either binds to explicitly specified port, or fails with EADDRINUSE. Even when telling it to bind to port 0, it will start successfully, but won’t display what port it had bound to.

If I find a free port myself in the Python wrapper, there is no guarantee that some other process won’t bind to the same port meanwhile I am passing it to Appium. And if I don’t release it first myself, Appium won’t be able to bind to it, so I have to.

I know this is unlikely to ever happen in practice, but what would be the “right way” to “reserve” a local port number before passing it to another process in a cross-platform way (Linux, macOS, Windows)?

Advertisement

Answer

Thanks to @rodrigo suggestion in comments, I have ended up with this code:

import platform
import re
import subprocess
from typing import Set

if platform.system() == 'Windows':
    def _get_ports(pid):
        sp = subprocess.run(['netstat', '-anop', 'TCP'],
                            stdout=subprocess.PIPE,
                            stderr=subprocess.DEVNULL,
                            check=True)

        rx_socket = re.compile(br'''(?x) ^
                                    s* TCP
                                    s+ 127.0.0.1 : (?P<port>d{1,5})
                                    s+ .*?
                                    s+ LISTENING
                                    s+ (?P<pid>d+)
                                    s* $''')

        for line in sp.stdout.splitlines():
            rxm = rx_socket.match(line)
            if rxm is None:
                continue

            sock_port, sock_pid = map(int, rxm.groups())
            if sock_pid == pid:
                yield sock_port
else:
    def _get_ports(pid):
        sp = subprocess.run(['lsof', '-anlPFn', '+w',
                             f'-p{pid}', '-i4TCP@127.0.0.1', '-sTCP:LISTEN'],
                            stdout=subprocess.PIPE,
                            stderr=subprocess.DEVNULL,
                            check=True)

        for line in sp.stdout.splitlines():
            if line.startswith(b'n'):
                host, port = line.rsplit(b':', 1)
                port = int(port)
                yield port


def get_ports(pid: int) -> Set[int]:
    """Get set of local-bound listening TCPv4 ports for given process.

    :param pid: process ID to inspect
    :returns: set of ports
    """

    return set(_get_ports(pid))

print(get_ports(12345))

It works on Linux, macOS and Windows, and finds out all locally-bound TCPv4 ports for given process that are in LISTEN state. It also skips all kinds of host/port/username reverse look-ups to make it faster, and does not require elevated privileges.

So, finally, the idea is to just let Appium (or anything else) start on 0.0.0.0:0, it will bind itself to the first available port, as provided by OS, and then inspect what ports is it now listening on. No race conditions.

User contributions licensed under: CC BY-SA
2 People found this is helpful
Advertisement