diff --git a/src/__init__.py b/src/__init__.py index fe3565c3..95d3e296 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -41,7 +41,6 @@ from .utils import \ reserve_port, \ release_port, \ - bound_ports, \ get_bin_path, \ get_bin_dir, \ get_pg_config, \ @@ -73,7 +72,7 @@ "NodeApp", "PostgresNode", "PortManager", - "reserve_port", "release_port", "bound_ports", "get_bin_path", "get_bin_dir", "get_pg_config", "get_pg_version", "parse_pg_version", + "reserve_port", "release_port", "get_bin_path", "get_bin_dir", "get_pg_config", "get_pg_version", "parse_pg_version", "First", "Any", "OsOperations", "LocalOperations", "RemoteOperations", "ConnectionParams" ] diff --git a/src/consts.py b/src/consts.py index 89c49ab7..d3589205 100644 --- a/src/consts.py +++ b/src/consts.py @@ -10,6 +10,10 @@ TMP_CACHE = 'tgsc_' TMP_BACKUP = 'tgsb_' +TMP_TESTGRES = "testgres" + +TMP_TESTGRES_PORTS = TMP_TESTGRES + "/ports" + # path to control file XLOG_CONTROL_FILE = "global/pg_control" diff --git a/src/impl/port_manager__generic2.py b/src/impl/port_manager__generic2.py new file mode 100755 index 00000000..b2b5f5c2 --- /dev/null +++ b/src/impl/port_manager__generic2.py @@ -0,0 +1,166 @@ +from testgres.operations.os_ops import OsOperations + +from ..port_manager import PortManager +from ..exceptions import PortForException +from .. import consts + +import threading +import random +import typing +import logging + + +class OsLockFsObj: + _os_ops: typing.Optional[OsOperations] + _path: typing.Optional[str] + + def __init__(self, os_ops: OsOperations, path: str): + assert isinstance(os_ops, OsOperations) + assert type(path) is str + + os_ops.makedir(path) # throw + + self._os_ops = os_ops + self._path = path + return + + def release(self) -> None: + assert type(self._path) is str + assert isinstance(self._os_ops, OsOperations) + assert self._os_ops.path_exists(self._path) + + self._os_ops.rmdir(self._path) # throw + + self._path = None + self._os_ops = None + return + + +class PortManager__Generic2(PortManager): + _C_MIN_PORT_NUMBER = 1024 + _C_MAX_PORT_NUMBER = 65535 + + _os_ops: OsOperations + _guard: typing.Any + + # TODO: is there better to use bitmap fot _available_ports? + _available_ports: typing.Set[int] + _reserved_ports: typing.Dict[int, OsLockFsObj] + + _lock_dir: typing.Optional[str] + + def __init__(self, os_ops: OsOperations): + assert __class__._C_MIN_PORT_NUMBER <= __class__._C_MAX_PORT_NUMBER + + assert os_ops is not None + assert isinstance(os_ops, OsOperations) + self._os_ops = os_ops + self._guard = threading.Lock() + + self._available_ports = set( + range(__class__._C_MIN_PORT_NUMBER, __class__._C_MAX_PORT_NUMBER + 1) + ) + assert len(self._available_ports) == ( + (__class__._C_MAX_PORT_NUMBER - __class__._C_MIN_PORT_NUMBER) + 1 + ) + self._reserved_ports = dict() + self._lock_dir = None + return + + def reserve_port(self) -> int: + assert self._guard is not None + assert type(self._available_ports) is set + assert type(self._reserved_ports) is dict + assert isinstance(self._os_ops, OsOperations) + + with self._guard: + if self._lock_dir is None: + temp_dir = self._os_ops.get_tempdir() + assert type(temp_dir) is str + lock_dir = self._os_ops.build_path(temp_dir, consts.TMP_TESTGRES_PORTS) + assert type(lock_dir) is str + self._os_ops.makedirs(lock_dir) + self._lock_dir = lock_dir + + assert self._lock_dir is not None + assert type(self._lock_dir) is str + + t = tuple(self._available_ports) + assert len(t) == len(self._available_ports) + sampled_ports = random.sample(t, min(len(t), 100)) + t = None + + for port in sampled_ports: + assert type(port) is int + assert port not in self._reserved_ports + assert port in self._available_ports + + assert port >= __class__._C_MIN_PORT_NUMBER + assert port <= __class__._C_MAX_PORT_NUMBER + + if not self._os_ops.is_port_free(port): + continue + + try: + lock_path = self.helper__make_lock_path(port) + lock_obj = OsLockFsObj(self._os_ops, lock_path) # raise + except: # noqa: E722 + continue + + assert isinstance(lock_obj, OsLockFsObj) + assert self._os_ops.path_exists(lock_path) + + try: + self._reserved_ports[port] = lock_obj + except: # noqa: E722 + assert port not in self._reserved_ports + lock_obj.release() + raise + + self._available_ports.discard(port) + assert port in self._reserved_ports + assert port not in self._available_ports + __class__.helper__send_debug_msg("Port {} is reserved.", port) + return port + + raise PortForException("Can't select a port.") + + def release_port(self, number: int) -> None: + assert type(number) is int + assert number >= __class__._C_MIN_PORT_NUMBER + assert number <= __class__._C_MAX_PORT_NUMBER + + assert self._guard is not None + assert type(self._reserved_ports) is dict + + with self._guard: + assert number in self._reserved_ports + assert number not in self._available_ports + self._available_ports.add(number) + lock_obj = self._reserved_ports.pop(number) + assert number not in self._reserved_ports + assert number in self._available_ports + assert isinstance(lock_obj, OsLockFsObj) + lock_obj.release() + __class__.helper__send_debug_msg("Port {} is released.", number) + return + + @staticmethod + def helper__send_debug_msg(msg_template: str, *args) -> None: + assert msg_template is not None + assert args is not None + assert type(msg_template) is str + assert type(args) is tuple + assert msg_template != "" + s = "[port manager] " + s += msg_template.format(*args) + logging.debug(s) + + def helper__make_lock_path(self, port_number: int) -> str: + assert type(port_number) is int + # You have to call the reserve_port at first! + assert type(self._lock_dir) is str + + result = self._os_ops.build_path(self._lock_dir, str(port_number) + ".lock") + assert type(result) is str + return result diff --git a/src/impl/port_manager__this_host.py b/src/impl/port_manager__this_host.py index c7520d80..6c3ef41d 100755 --- a/src/impl/port_manager__this_host.py +++ b/src/impl/port_manager__this_host.py @@ -3,10 +3,11 @@ from .. import utils import threading +import typing class PortManager__ThisHost(PortManager): - sm_single_instance: PortManager = None + sm_single_instance: typing.Optional[PortManager] = None sm_single_instance_guard = threading.Lock() @staticmethod diff --git a/src/node.py b/src/node.py index c642bddb..be1c04f4 100644 --- a/src/node.py +++ b/src/node.py @@ -77,7 +77,7 @@ from .port_manager import PortManager from .impl.port_manager__this_host import PortManager__ThisHost -from .impl.port_manager__generic import PortManager__Generic +from .impl.port_manager__generic2 import PortManager__Generic2 from .logger import TestgresLogger @@ -309,12 +309,12 @@ def _get_port_manager(os_ops: OsOperations) -> PortManager: if os_ops is LocalOperations.get_single_instance(): assert utils._old_port_manager is not None - assert type(utils._old_port_manager) is PortManager__Generic + assert type(utils._old_port_manager) is PortManager__Generic2 assert utils._old_port_manager._os_ops is os_ops return PortManager__ThisHost.get_single_instance() # TODO: Throw the exception "Please define a port manager." ? - return PortManager__Generic(os_ops) + return PortManager__Generic2(os_ops) def clone_with_new_name_and_base_dir(self, name: str, base_dir: str): assert name is None or type(name) is str diff --git a/src/utils.py b/src/utils.py index 6efd9a49..8561c7cc 100644 --- a/src/utils.py +++ b/src/utils.py @@ -26,7 +26,7 @@ from testgres.operations.local_ops import LocalOperations from testgres.operations.helpers import Helpers as OsHelpers -from .impl.port_manager__generic import PortManager__Generic +from .impl.port_manager__generic2 import PortManager__Generic2 from .impl.platforms import internal_platform_utils_factory from .impl import internal_utils @@ -37,10 +37,7 @@ # # The old, global "port manager" always worked with LOCAL system # -_old_port_manager = PortManager__Generic(LocalOperations.get_single_instance()) - -# ports used by nodes -bound_ports = _old_port_manager._reserved_ports +_old_port_manager = PortManager__Generic2(LocalOperations.get_single_instance()) # re-export version type @@ -55,7 +52,7 @@ def __init__(self, version: str) -> None: def internal__reserve_port(): """ - Generate a new port and add it to 'bound_ports'. + Generate a new port. """ return _old_port_manager.reserve_port() diff --git a/tests/helpers/global_data.py b/tests/helpers/global_data.py index fa49b24d..67004538 100644 --- a/tests/helpers/global_data.py +++ b/tests/helpers/global_data.py @@ -5,7 +5,7 @@ from src.node import PortManager from src.node import PortManager__ThisHost -from src.node import PortManager__Generic +from src.node import PortManager__Generic2 import os @@ -37,11 +37,11 @@ class OsOpsDescrs: class PortManagers: - sm_remote_port_manager = PortManager__Generic(OsOpsDescrs.sm_remote_os_ops) + sm_remote_port_manager = PortManager__Generic2(OsOpsDescrs.sm_remote_os_ops) sm_local_port_manager = PortManager__ThisHost.get_single_instance() - sm_local2_port_manager = PortManager__Generic(OsOpsDescrs.sm_local_os_ops) + sm_local2_port_manager = PortManager__Generic2(OsOpsDescrs.sm_local_os_ops) class PostgresNodeService: diff --git a/tests/test_testgres_local.py b/tests/test_testgres_local.py index 437d5f6d..012ec12d 100644 --- a/tests/test_testgres_local.py +++ b/tests/test_testgres_local.py @@ -20,7 +20,6 @@ from src import get_pg_version # NOTE: those are ugly imports -from src.utils import bound_ports from src.utils import PgVer from src.node import ProcessProxy @@ -91,40 +90,6 @@ def test_pg_config(self): b = get_pg_config() assert (id(a) != id(b)) - def test_ports_management(self): - assert bound_ports is not None - assert type(bound_ports) is set - - if len(bound_ports) != 0: - logging.warning("bound_ports is not empty: {0}".format(bound_ports)) - - stage0__bound_ports = bound_ports.copy() - - with get_new_node() as node: - assert bound_ports is not None - assert type(bound_ports) is set - - assert node.port is not None - assert type(node.port) is int - - logging.info("node port is {0}".format(node.port)) - - assert node.port in bound_ports - assert node.port not in stage0__bound_ports - - assert stage0__bound_ports <= bound_ports - assert len(stage0__bound_ports) + 1 == len(bound_ports) - - stage1__bound_ports = stage0__bound_ports.copy() - stage1__bound_ports.add(node.port) - - assert stage1__bound_ports == bound_ports - - # check that port has been freed successfully - assert bound_ports is not None - assert type(bound_ports) is set - assert bound_ports == stage0__bound_ports - def test_child_process_dies(self): # test for FileNotFound exception during child_processes() function cmd = ["timeout", "60"] if os.name == 'nt' else ["sleep", "60"]