Source code for galaxy.util.monitors

import logging
import threading

from .sleeper import Sleeper

log = logging.getLogger(__name__)

DEFAULT_MONITOR_THREAD_JOIN_TIMEOUT = 5


[docs]class Monitors:
    def _init_monitor_thread(self, name, target_name=None, target=None, start=False, config=None):
        self.monitor_join_sleep = getattr(config, "monitor_thread_join_timeout", DEFAULT_MONITOR_THREAD_JOIN_TIMEOUT)
        self.monitor_join = self.monitor_join_sleep > 0
        self.monitor_running = True

        if target is not None:
            assert target_name is None
            monitor_func = target
        else:
            target_name = target_name or "monitor"
            monitor_func = getattr(self, target_name)
        self.sleeper = Sleeper()
        self.monitor_thread = threading.Thread(name=name, target=monitor_func)
        self.monitor_thread.daemon = True
        self._start = start
        self.start_monitoring()

    def _init_noop_monitor(self):
        self.sleeper = None
        self.monitor_join = False

[docs]    def start_monitoring(self):
        if self._start:
            self.monitor_thread.start()

[docs]    def stop_monitoring(self):
        self.monitor_running = False

    def _monitor_sleep(self, sleep_amount):
        self.sleeper.sleep(sleep_amount)

[docs]    def shutdown_monitor(self):
        self.stop_monitoring()
        if self.sleeper is not None:
            self.sleeper.wake()
        if self.monitor_join:
            log.debug("Joining monitor thread")
            self.monitor_thread.join(self.monitor_join_sleep)