如何为子进程选择空闲端口? [英] How to pick a free port for a subprocess?

问题描述

我正在围绕Appium服务器编写Python包装器.Appium接受用于绑定到本地端口的命令行参数.不幸的是，Appium无法为其自身自动选择一个空闲端口，因此它要么绑定到明确指定的端口，要么失败，并显示 EADDRINUSE .即使告诉它绑定到端口 0 ，它也将成功启动，但不会显示其绑定到的端口.

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.

如果我自己在Python包装器中找到了一个免费端口，则无法保证在我将其传递给Appium的同时，其他某些进程也不会绑定到同一端口.而且，如果我自己不首先发布它，Appium将无法绑定到它，所以我必须这么做.

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.

我知道这在实践中几乎不可能发生，但是在以跨平台方式将其传递给另一个进程(Linux，macOS，Windows)之前，保留"本地端口号的正确方法"是什么??

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)?

推荐答案

感谢@rodrigo的建议，我最终得到了以下代码:

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))

它可在Linux，macOS和Windows上运行，并找出处于LISTEN状态的给定进程的所有本地绑定的TCPv4端口.它还跳过所有类型的主机/端口/用户名反向查找，以使其更快，并且不需要提升的特权.

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.

因此，最后的想法是让Appium(或其他任何东西)从 0.0.0.0:0 开始，它将自身绑定到OS提供的第一个可用端口，并且然后检查它现在正在侦听哪些端口.没有比赛条件.

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.

这篇关于如何为子进程选择空闲端口?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！