I have a distributed application (YARN), which runs a WebApp.

This application use a default port to start (8008), before I start I need to check if port is in use.

A container may run in the same virtual machine, hence port may be in use. (Max I have 4 containers in WebApp).

I created the following code which seem to work, but want to see if there are some clean ups/improvements suggested.

def port_in_use(port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    result = sock.connect_ex(('', port))
    if result == 0:
      return True
      return False

def start_dashboard():    
    base_port = os.getenv('DASHBOARD_PORT_ENV_VAR', 8008)
    scan_ports = True
    attempts = 0
    max_attempts = 10
    while(scan_ports and attempts <= max_attempts):
      if port_in_use(base_port):
         base_port += 1
         attempts += 1
         scan_ports = False
    if attempts == max_attempts:
       raise IOError('Port in use')

1 Answer 1


Your code has some incorrect assumptions.

  • an application may listen on a specific address/port combination; can be available while *:port is not.

  • an application may bind a port without listening. Connects will fail, but so will your own bind.

  • a firewall or other mechanism can interfere with connections, generating false positives in your scan.

The reliable approach is to bind the port, just as your dashboard will, and then release it.

result = sock.bind(('', port))

You'll need to catch the exception and this is a good opportunity to move the whole thing into a function. That will make the start_dashboard logic cleaner and get rid of boolean loop-terminator scan_ports. Just exit the loop by returning the answer.

def next_free_port( port=1024, max_port=65535 ):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    while port <= max_port:
            sock.bind(('', port))
            return port
        except OSError:
            port += 1
    raise IOError('no free ports')

def start_dashboard():

    # pass optional second parameter "max_port" here, else scan until a free one is found
    port = next_free_port( os.getenv('DASHBOARD_PORT_ENV_VAR', 8008) )


You can use netcat to make ports in-use for testing: nc -l -p 9999 will listen on port 9999; press control-C to end it.

  • \$\begingroup\$ Thank you for the answer, check is local, so no firewall in place. \$\endgroup\$
    – gogasca
    Commented Mar 23, 2019 at 17:42
  • \$\begingroup\$ I mean a host firewall, like iptables on Linux. \$\endgroup\$ Commented Mar 23, 2019 at 19:06
  • \$\begingroup\$ Sounds good, any recommendation for Python style? Thanks \$\endgroup\$
    – gogasca
    Commented Mar 24, 2019 at 1:53
  • \$\begingroup\$ Using .bind I get: >>> port_in_use(22) if port is in use, which may just require to handle the OS exception. Traceback (most recent call last): File "<stdin>", line 1, in <module> File "<stdin>", line 3, in port_in_use OSError: [Errno 98] Address already in use \$\endgroup\$
    – gogasca
    Commented Mar 24, 2019 at 1:59
  • 1
    \$\begingroup\$ the code can be a little shorter and clearer; see edits for a fleshed-out example. \$\endgroup\$ Commented Mar 24, 2019 at 6:37

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.