Python scanner for the first free port in a range

Question

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(('127.0.0.1', port))
 if result == 0:
 return True
 else:
 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
 else:
 scan_ports = False
 if attempts == max_attempts:
 raise IOError('Port in use')
 dashboard.configure(port=base_port)
 dashboard.launch()

Answer 1

Your code has some incorrect assumptions.

• an application may listen on a specific address/port combination; 127.0.0.1:port 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))
sock.close()

---

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:
 try:
 sock.bind(('', port))
 sock.close()
 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) )
 dashboard.configure(port=port)
 dashboard.launch()

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.

• gogasca: Thank you for the answer, check is local, so no firewall in place.
• Oh My Goodness: I mean a host firewall, like iptables on Linux.
• gogasca: Sounds good, any recommendation for Python style? Thanks
• gogasca: 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
• Oh My Goodness: the code can be a little shorter and clearer; see edits for a fleshed-out example.