Skip to content

Commit

Permalink
Auto-select port for server
Browse files Browse the repository at this point in the history
Add a function `_find_available_port` to find a port in the range
[50051, 60000] that is available to open a listening socket on.  Use
this function to automatically select a port when either
`launch_session` or `session` are used to launch a gRPC server session
and `port_num` is `None` (the default).
  • Loading branch information
isaacwaldron committed Jan 10, 2025
1 parent d023cc5 commit acba55c
Show file tree
Hide file tree
Showing 2 changed files with 47 additions and 12 deletions.
39 changes: 32 additions & 7 deletions src/ansys/edb/core/session.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
"""Session manager for gRPC."""

from contextlib import contextmanager
from enum import Enum
import errno
from shutil import which
import socket
from struct import pack, unpack
import subprocess
from sys import modules
Expand Down Expand Up @@ -142,6 +145,8 @@
from ansys.edb.core.inner.exceptions import EDBSessionException, ErrorCode
from ansys.edb.core.inner.interceptors import ExceptionInterceptor, LoggingInterceptor

DEFAULT_ADDRESS = "localhost"

# The session module singleton
MOD = modules[__name__]
MOD.current_session = None
Expand Down Expand Up @@ -172,8 +177,8 @@ def __init__(self, ip_address, port_num, ansys_em_root):
if MOD.current_session is not None:
raise EDBSessionException(ErrorCode.STARTUP_MULTI_SESSIONS)

self.ip_address = ip_address or "localhost"
self.port_num = port_num
self.ip_address = ip_address or DEFAULT_ADDRESS
self.port_num = port_num or _find_available_port()
self.ansys_em_root = ansys_em_root
self.channel = None
self.local_server_proc = None
Expand Down Expand Up @@ -457,8 +462,8 @@ def launch_session(ansys_em_root, port_num=None):
ansys_em_root : str
Directory where the ``EDB_RPC_Server.exe`` file is installed.
port_num : int, default: None
Port number to listen on. The default is ``None``, in which
case localhost is used.
Port number to listen on. The default is ``None``, in which case a port in [50051, 60000]
is selected.
Examples
--------
Expand All @@ -480,15 +485,16 @@ def launch_session(ansys_em_root, port_num=None):


@contextmanager
def session(ansys_em_root, port_num, ip_address=None):
def session(ansys_em_root: str, port_num: int = None, ip_address: str = None):
r"""Launch a local session to an EDB API server in a context manager.
Parameters
----------
ansys_em_root : str
Directory where the ``EDB_RPC_Server.exe`` file is installed.
port_num : int
Port number to listen on.
port_num : int, default: None
Port number to listen on. The default is ``None``, in which case a port in [50051, 60000]
is selected.
ip_address : str, default: None
IP address where the server executable file is running. The default is ``None``, in which
case localhost is used.
Expand Down Expand Up @@ -573,3 +579,22 @@ def _ensure_session(ansys_em_root, port_num, ip_address):
else:
MOD.current_session = _Session(ip_address, port_num, ansys_em_root)
MOD.current_session.connect()


def _find_available_port(interface: str = None, start_port: int = 50051, end_port: int = 60000):
"""Find an available port in the given range.
Parameters
----------
interface : str, default: :data:`DEFAULT_ADDRESS`
Interface to check for available ports.
start_port : int, default: ``50051``
First port number to check.
end_port : int, default: ``60000``
Last port number to check.
"""
for port in range(start_port, end_port):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
if sock.connect_ex((interface or DEFAULT_ADDRESS, port)) == errno.ECONNREFUSED:
return port
raise RuntimeError("No available ports found")
20 changes: 15 additions & 5 deletions tests/e2e/unit_tests/test_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,16 +5,26 @@
import settings

from ansys.edb.core.database import Database
from ansys.edb.core.session import _Session, launch_session
import ansys.edb.core.session as session


def test_default_port_in_use(tmp_path: Path):
def test_launch_session_when_default_port_in_use(new_database_path: Path):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 50051))
sess: _Session = launch_session(settings.server_exe_dir())
sess: session._Session = session.launch_session(settings.server_exe_dir())
try:
edb_path = tmp_path / "p1.aedb"
with closing(Database.create(edb_path.as_posix())) as db:
assert isinstance(sess.port_num, int) and sess.port_num != 50051
with closing(Database.create(new_database_path)) as db:
assert not db.is_null
finally:
sess.disconnect()


def test_session_when_default_port_in_use(new_database_path: Path):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 50051))
with session.session(settings.server_exe_dir()):
sess: session._Session = session.MOD.current_session
assert isinstance(sess.port_num, int) and sess.port_num != 50051
with closing(Database.create(new_database_path)) as db:
assert not db.is_null

0 comments on commit acba55c

Please sign in to comment.