update impl; add tests

Author: zhenyu-ms

testplan/common/entity/base.py

@@ -12,6 +12,7 @@
 import traceback
 from collections import OrderedDict, deque
 from contextlib import suppress
+from multiprocessing import TimeoutError
 from typing import (
     Any,
     Callable,
@@ -34,7 +35,7 @@
 from testplan.common.utils.path import default_runpath, makedirs, makeemptydirs
 from testplan.common.utils.strings import slugify, uuid4
 from testplan.common.utils.thread import execute_as_thread, interruptible_join
-from testplan.common.utils.timing import wait, Timer
+from testplan.common.utils.timing import Timer, wait
 from testplan.common.utils.validation import is_subclass
 
 
@@ -239,32 +240,32 @@ def start(self):
             else:
                 resource.logger.info("%s started", resource)
 
-    def start_in_pool(self, pool: mp.ThreadPool):
+    def start_in_pool(
+        self, pool: mp.ThreadPool, timeout: Optional[float] = None
+    ):
         """
         Start all resources concurrently in thread pool and log exceptions.
 
         :param pool: thread pool
         """
-
-        for resource in self._resources.values():
-            if not resource.async_start:
-                raise RuntimeError(
-                    f"Cannot start resource {resource} in thread pool,"
-                    " its `async_start` attribute is set to False"
-                )
-
-        def start_in_thread(resource):
-            resource.start()
-            resource.wait(resource.STATUS.STARTED)
-            resource.logger.info("%s started", resource)
+        # async_start is meaningless here...
+        for r in self._resources.values():
+            if r.auto_start:
+                r.cfg.set_local("async_start", False)
 
         async_r = pool.map_async(
             self._apply_resource_exception_logged(
-                self.start_exceptions, start_in_thread
+                self.start_exceptions, lambda r: r.start()
             ),
             (r for r in self._resources.values() if r.auto_start),
         )
-        async_r.wait()
+        try:
+            async_r.get(timeout)
+        except TimeoutError:
+            raise RuntimeError(
+                f"timeout after {timeout}s when starting resources "
+                f"{self._resources.values()} in internal threading pool."
+            )
 
     def sync_stop_resource(self, resource: "Resource"):
         """
@@ -346,42 +347,33 @@ def stop(self, is_reversed: bool = False):
                 resource.force_stop()
             resource.logger.info("%s stopped", resource)
 
-    def stop_in_pool(self, pool: mp.ThreadPool):
+    def stop_in_pool(
+        self, pool: mp.ThreadPool, timeout: Optional[float] = None
+    ):
         """
         Stop all resources concurrently in thread pool and log exceptions.
 
         :param pool: thread pool
         """
-        for resource in self._resources.values():
-            if not resource.async_start:
-                raise RuntimeError(
-                    f"Cannot stop resource {resource} in thread pool,"
-                    " its `async_start` attribute is set to False"
-                )
-
-        def stop_in_thread(resource: "Resource"):
-            try:
-                if resource.status not in (
-                    resource.STATUS.STOPPING,
-                    resource.STATUS.STOPPED,
-                ):
-                    resource.stop()
-                if resource.status == resource.STATUS.STOPPING:
-                    resource.wait(resource.STATUS.STOPPED)
-            except:
-                # Resource status should be STOPPED even it failed to stop
-                resource.force_stop()
-                raise
-            finally:
-                resource.logger.info("%s stopped", resource)
+        # async_start is meaningless here...
+        # in practice, stop_in_pool must be called in pair of start_in_pool
+        for r in self._resources.values():
+            if r.auto_start:
+                r.cfg.set_local("async_start", False)
 
         async_r = pool.map_async(
             self._apply_resource_exception_logged(
-                self.stop_exceptions, stop_in_thread
+                self.stop_exceptions, lambda r: r.stop()
             ),
             self._resources.values(),
         )
-        async_r.wait()
+        try:
+            async_r.get(timeout)
+        except TimeoutError:
+            raise RuntimeError(
+                f"timeout after {timeout}s when stopping resources "
+                f"{self._resources.values()} in internal threading pool."
+            )
 
     def _apply_resource_exception_logged(
         self, exception_record, func: Callable[["Resource"], None]

testplan/common/remote/remote_resource.py

@@ -143,7 +143,7 @@ class RemoteResource(Entity):
     :param paramiko_config: Paramiko SSH client extra configuration.
     :param remote_runtime_builder: RuntimeBuilder instance to prepare remote python env.
         Default is ``SourceTransferBuilder()``.
-    :param status_wait_timeout: remote resource start/stop timeout, default is 60.
+    :param status_wait_timeout: remote resource start/stop timeout, default is 600.
     """
 
     CONFIG = RemoteResourceConfig
@@ -170,7 +170,7 @@ def __init__(
         setup_script: List[str] = None,
         paramiko_config: Optional[dict] = None,
         remote_runtime_builder: Optional[RuntimeBuilder] = None,
-        status_wait_timeout: int = 60,
+        status_wait_timeout: int = 600,
         **options,
     ) -> None:
         if not worker_is_remote(remote_host):

testplan/runners/pools/remote.py

@@ -364,20 +364,26 @@ def _start_workers(self) -> None:
         for worker in self._workers:
             self._conn.register(worker)
         if self.pool:
-            self._workers.start_in_pool(self.pool)
+            self._workers.start_in_pool(
+                self.pool,
+                self.cfg.status_wait_timeout,
+            )
         else:
             self._workers.start()
 
     def _stop_workers(self) -> None:
         if self.pool:
-            self._workers.stop_in_pool(self.pool)
+            self._workers.stop_in_pool(
+                self.pool,
+                self.cfg.status_wait_timeout,
+            )
         else:
             self._workers.stop()
 
     def _start_thread_pool(self) -> None:
         size = len(self._instances)
         try:
-            if size > 2:
+            if size >= 2:
                 self.pool = ThreadPool(min(size, cpu_count()))
         except Exception as exc:
             if isinstance(exc, AttributeError):