Expand settings

Author: ianmkenney

alchemiscale/compute/monitor.py

@@ -21,6 +21,11 @@ def __init__(self, settings):
         self._lock = Lock()
         self._terminate = False
 
+        if not hasattr(self, "sample_time"):
+            raise AttributeError(
+                f"{self.__class__.__name__} implementation requires definition of the `sample_time` attribute"
+            )
+
     def signal(self) -> ResourceSignal:
         with self._lock:
             return self._signal()
@@ -32,8 +37,7 @@ def monitor_cycle(self):
         while not self._terminate:
             with self._lock:
                 self._monitor_cycle()
-            # TODO: make configurable
-            time.sleep(1)
+            time.sleep(self.sample_time)
 
     @abstractmethod
     def _setup(self, settings):
@@ -53,7 +57,11 @@ class GPUMonitor(Monitor):
 
     def _setup(self, settings):
         self.history = []
-        self.gpu_index = settings.gpu_monitor_gpu_id
+        self.history_size = settings.gpu_monitor_sample_history_size
+        self.gpu_index = settings.gpu_monitor_gpu_index
+        self.grow_limit = settings.gpu_monitor_grow_limit
+        self.maintain_limit = settings.gpu_monitor_maintain_limit
+        self.sample_time = settings.gpu_monitor_sample_time
 
     @staticmethod
     def _nvidia_smi() -> int:
@@ -78,7 +86,7 @@ def _nvidia_smi() -> int:
     def _monitor_cycle(self):
         util = self._nvidia_smi()
         self.history.append(util)
-        self.history = self.history[-60:]
+        self.history = self.history[-self.history_size :]
 
     def _signal(self) -> ResourceSignal:
         utilization = sum(self.history) / len(self.history)
@@ -91,12 +99,12 @@ def _signal(self) -> ResourceSignal:
 
 class CPUMonitor(Monitor):
 
-    # TODO: make configurable
-    grow_limit = 0.50
-    maintain_limit = 0.75
-
     def _setup(self, settings):
         self.history = []
+        self.history_size = settings.cpu_monitor_sample_history_size
+        self.sample_time = settings.cpu_monitor_sample_time
+        self.grow_limit = settings.cpu_monitor_grow_limit
+        self.maintain_limit = settings.cpu_monitor_maintain_limit
 
     def _signal(self) -> ResourceSignal:
         total_load = sum(self.history) / len(self.history)
@@ -111,7 +119,7 @@ def _monitor_cycle(self):
         cpu_count = os.cpu_count()
         total_load = load / cpu_count
         self.history.append(total_load)
-        self.history = self.history[-60:]
+        self.history = self.history[-self.history_size :]
 
 
 class MemInfoParseError(Exception):
@@ -120,12 +128,12 @@ class MemInfoParseError(Exception):
 
 class MemoryMonitor(Monitor):
 
-    # TODO: make configurable
-    grow_limit = 0.65
-    maintain_limit = 0.85
-
     def _setup(self, settings):
         self.history = []
+        self.history_size = settings.memory_monitor_sample_history_size
+        self.sample_time = settings.memory_monitor_sample_time
+        self.grow_limit = settings.memory_monitor_grow_limit
+        self.maintain_limit = settings.memory_monitor_maintain_limit
 
     @staticmethod
     def _get_memory() -> tuple[int, int]:
@@ -157,7 +165,7 @@ def _monitor_cycle(self):
             fraction_used = (total - avail) / total
             self.history.append(fraction_used)
             # roughly the last minute of entries
-            self.history = self.history[-60:]
+            self.history = self.history[-self.history_size :]
         except Exception:
             self._terminate = True
 

alchemiscale/compute/service.py

@@ -692,35 +692,26 @@ class AsynchronousComputeService(SynchronousComputeService):
     _child_env: dict[str, str]
 
     def __init__(self, settings: ComputeServiceSettings):
-        self._child_env = dict()
+        self.settings = settings
+
+        # asynccomputeservice specific data structures and resource
+        # monitors.
+        self._child_env = dict()  # mods to child process env
         self._task_data = dict()
         self._initialize_dag_tree()
-        self._resource_monitors = None
-        self._initialize_resource_monitors(settings)
-
-        self.settings = settings
+        self._initialize_resource_monitors()
 
         self.api_url = self.settings.api_url
         self.name = self.settings.name
+        self.compute_manager_id = self.settings.compute_manager_id
         self.sleep_interval = self.settings.sleep_interval
+        self.deep_sleep_interval = self.settings.deep_sleep_interval
         self.heartbeat_interval = self.settings.heartbeat_interval
         self.claim_limit = self.settings.claim_limit
 
-        self.scheduler = sched.scheduler(time.monotonic, time.sleep)
-
-        self.client = AlchemiscaleComputeClient(
-            self.settings.api_url,
-            self.settings.identifier,
-            self.settings.key,
-            cache_directory=self.settings.client_cache_directory,
-            cache_size_limit=self.settings.client_cache_size_limit,
-            use_local_cache=self.settings.client_use_local_cache,
-            max_retries=self.settings.client_max_retries,
-            retry_base_seconds=self.settings.client_retry_base_seconds,
-            retry_max_seconds=self.settings.client_retry_max_seconds,
-            verify=self.settings.client_verify,
-        )
+        self.client = self._initialize_client()
         self.scopes = self.settings.scopes or [Scope()]
+
         self.shared_basedir = Path(self.settings.shared_basedir).absolute()
         self.shared_basedir.mkdir(exist_ok=True)
         self.keep_shared = self.settings.keep_shared
@@ -731,19 +722,59 @@ def __init__(self, settings: ComputeServiceSettings):
 
         self.compute_service_id = ComputeServiceID.new_from_name(self.name)
 
-    def _initialize_resource_monitors(self, settings):
+        self.int_sleep = InterruptableSleep()
+        self._initialize_logger()
+
+    def _initialize_logger(self):
+        extra = {"compute_service_id": str(self.compute_service_id)}
+        logger = logging.getLogger("AlchemiscaleAsynchronousComputeService")
+        logger.setLevel(self.settings.loglevel)
+
+        formatter = logging.Formatter(
+            "[%(asctime)s] [%(compute_service_id)s] [%(levelname)s] %(message)s"
+        )
+        formatter.converter = time.gmtime  # use utc time for logging timestamps
+
+        sh = logging.StreamHandler()
+        sh.setFormatter(formatter)
+        logger.addHandler(sh)
+
+        if self.settings.logfile is not None:
+            fh = logging.FileHandler(self.settings.logfile)
+            fh.setFormatter(formatter)
+            logger.addHandler(fh)
+
+        self.logger = logging.LoggerAdapter(logger, extra)
+
+    def _initialize_client(self):
+        return AlchemiscaleComputeClient(
+            api_url=self.settings.api_url,
+            identifier=self.settings.identifier,
+            key=self.settings.key,
+            cache_directory=self.settings.client_cache_directory,
+            cache_size_limit=self.settings.client_cache_size_limit,
+            use_local_cache=self.settings.client_use_local_cache,
+            max_retries=self.settings.client_max_retries,
+            retry_base_seconds=self.settings.client_retry_base_seconds,
+            retry_max_seconds=self.settings.client_retry_max_seconds,
+            verify=self.settings.client_verify,
+        )
+
+    def _initialize_resource_monitors(self):
         self._resource_monitors = []
 
-        if settings.memory_monitor_enabled:
-            self._resource_monitors.append(MemoryMonitor(settings))
+        if self.settings.memory_monitor_enabled:
+            self._resource_monitors.append(MemoryMonitor(self.settings))
 
-        if settings.cpu_monitor_enabled:
-            self._resource_monitors.append(CPUMonitor(settings))
+        if self.settings.cpu_monitor_enabled:
+            self._resource_monitors.append(CPUMonitor(self.settings))
 
-        if settings.gpu_monitor_enabled:
-            self._resource_monitors.append(GPUMonitor(settings))
+        if self.settings.gpu_monitor_enabled:
+            self._resource_monitors.append(GPUMonitor(self.settings))
             # reliable monitoring of the GPU requires pinning the GPU index
-            self._child_env |= {"CUDA_VISIBLE_DEVICES": settings.gpu_monitor_gpu_id}
+            self._child_env |= {
+                "CUDA_VISIBLE_DEVICES": self.settings.gpu_monitor_gpu_id
+            }
 
         for monitor in self._resource_monitors:
             threading.Thread(target=monitor.monitor_cycle, daemon=True).start()

alchemiscale/compute/settings.py

@@ -136,6 +136,74 @@ def validate_scopes(cls, values) -> list[Scope]:
         return _values
 
 
+class AsynchronousComputeServiceSettings(ComputeServiceSettings):
+
+    stack_size: int = Field(
+        2,
+        description="The number of concurrent protocol units that are able to run at once.",
+    )
+
+    gpu_monitor_enabled: bool = Field(
+        True, description="If the GPU monitor is enabled."
+    )
+    gpu_monitor_gpu_index: str = Field(
+        "0",
+        description="The GPU index to perform calculations on. This sets the CUDA_VISIBLE_DEVICES environment variable for spawned compute tasks.",
+    )
+    gpu_monitor_grow_limit: float = Field(
+        0.7,
+        description="GPU utilization percentage below this value will allow greater concurrency. See utilization.gpu in nvidia-smi for more information.",
+    )
+    gpu_monitor_maintain_limit: float = Field(
+        1.1,
+        description="GPU utilization percentage above this value will scale back concurrency. See utilization.gpu in nvidia-smi for more information.",
+    )
+    gpu_monitor_sample_time: int = Field(
+        1,
+        description="Number of seconds between collecting GPU utilization measurements.",
+    )
+    gpu_monitor_sample_history_size: int = Field(
+        60,
+        description="Maximum number of samples to use when considering reactive concurrency behavior.",
+    )
+    memory_monitor_enabled: bool = Field(
+        True, description="If the memory monitor is enabled."
+    )
+    memory_monitor_grow_limit: float = Field(
+        0.7,
+        description="Memory usage percentage below this value will allow greater concurrency.",
+    )
+    memory_monitor_maintain_limit: float = Field(
+        0.9,
+        description="Memory usage percentage above this value will scale back concurrency.",
+    )
+    memory_monitor_sample_time: int = Field(
+        1, description="Number of seconds between collecting memory usage measurements."
+    )
+    memory_monitor_sample_history_size: int = Field(
+        60,
+        description="Maximum number of samples to use when considering reactive concurrency behavior.",
+    )
+    cpu_monitor_enabled: bool = Field(
+        True, description="If the CPU monitor is enabled."
+    )
+    cpu_monitor_grow_limit: float = Field(
+        0.8,
+        description="CPU usage percentage below this value will allow greater concurrency.",
+    )
+    cpu_monitor_maintain_limit: float = Field(
+        1.2,
+        description="CPU usage percentage above this value will scale back concurrency.",
+    )
+    cpu_monitor_sample_time: int = Field(
+        1, description="Number of seconds between collecting CPU usage measurements."
+    )
+    cpu_monitor_sample_history_size: int = Field(
+        60,
+        description="Maximum number of samples to use when considering reactive concurrency behavior.",
+    )
+
+
 class ComputeManagerSettings(BaseModel):
     name: str = Field(
         ...,

service.py

@@ -8,18 +8,45 @@
 @dataclass
 class MockSettings:
     memory_monitor_enabled: bool
+    memory_monitor_sample_time: int
+    memory_monitor_sample_history_size: int
+    memory_monitor_grow_limit: float
+    memory_monitor_maintain_limit: float
     cpu_monitor_enabled: bool
+    cpu_monitor_sample_time: int
+    cpu_monitor_sample_history_size: int
+    cpu_monitor_grow_limit: float
+    cpu_monitor_maintain_limit: float
     gpu_monitor_enabled: bool
     gpu_monitor_gpu_id: int
+    gpu_monitor_sample_time: int
+    gpu_monitor_sample_history_size: int
+    gpu_monitor_grow_limit: float
+    gpu_monitor_maintain_limit: float
 
 class MockService(AsynchronousComputeService):
 
     def __init__(self, scratch_basedir, shared_basedir, stack_size, keep_scratch, keep_shared, n_retries, claim_limit, task_generator):
         self._child_env = dict()
         self._initialize_dag_tree()
-        settings = MockSettings(memory_monitor_enabled=True, cpu_monitor_enabled=True, gpu_monitor_enabled=False, gpu_monitor_gpu_id="0")
-        #settings = MockSettings(memory_monitor_enabled=False, cpu_monitor_enabled=False, gpu_monitor_enabled=False, gpu_monitor_gpu_id="0")
-        self._initialize_resource_monitors(settings)
+        self.settings = MockSettings(memory_monitor_enabled=True,
+                                     cpu_monitor_enabled=True,
+                                     gpu_monitor_enabled=False,
+                                     gpu_monitor_gpu_id="0",
+                                     gpu_monitor_sample_time=1,
+                                     gpu_monitor_sample_history_size=60,
+                                     gpu_monitor_grow_limit=0.7,
+                                     gpu_monitor_maintain_limit=0.9,
+                                     memory_monitor_sample_time=1,
+                                     memory_monitor_sample_history_size=60,
+                                     memory_monitor_grow_limit=0.7,
+                                     memory_monitor_maintain_limit=0.9,
+                                     cpu_monitor_sample_time=1,
+                                     cpu_monitor_sample_history_size=60,
+                                     cpu_monitor_grow_limit=0.9,
+                                     cpu_monitor_maintain_limit=1.2,
+                                     )
+        self._initialize_resource_monitors()
         self._task_data = dict()
         self._executor_stack = ExecutorStack(stack_size)