[#247] Distributed port manager (PortManager__Generic2)

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"
 ]

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"
 

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

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

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

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

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:

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"]