python/ray/_private/test_utils.py
def monitor_memory_usage(
    print_interval_s: int = 30,
    record_interval_s: int = 5,
    warning_threshold: float = 0.9,
):
    """Run the memory monitor actor that prints the memory usage.

    The monitor will run on the same node as this function is called.

    Params:
        interval_s (int): The interval memory usage information is printed
        warning_threshold (float): The threshold where the
            memory usage warning is printed.

    Returns:
        The memory monitor actor.
    """
    assert ray.is_initialized(), "The API is only available when Ray is initialized."

    @ray.remote(num_cpus=0)
    class MemoryMonitorActor:
        def __init__(
            self,
            print_interval_s: float = 20,
            record_interval_s: float = 5,
            warning_threshold: float = 0.9,
            n: int = 10,
        ):
            """The actor that monitor the memory usage of the cluster.

            Params:
                print_interval_s (float): The interval where
                    memory usage is printed.
                record_interval_s (float): The interval where
                    memory usage is recorded.
                warning_threshold (float): The threshold where
                    memory warning is printed
                n (int): When memory usage is printed,
                    top n entries are printed.
            """
            # -- Interval the monitor prints the memory usage information. --
            self.print_interval_s = print_interval_s
            # -- Interval the monitor records the memory usage information. --
            self.record_interval_s = record_interval_s
            # -- Whether or not the monitor is running. --
            self.is_running = False
            # -- The used_gb/total_gb threshold where warning message omits. --
            self.warning_threshold = warning_threshold
            # -- The monitor that calculates the memory usage of the node. --
            self.monitor = memory_monitor.MemoryMonitor()
            # -- The top n memory usage of processes are printed. --
            self.n = n
            # -- The peak memory usage in GB during lifetime of monitor. --
            self.peak_memory_usage = 0
            # -- The top n memory usage of processes
            # during peak memory usage. --
            self.peak_top_n_memory_usage = ""
            # -- The last time memory usage was printed --
            self._last_print_time = 0
            # -- logger. --
            logging.basicConfig(level=logging.INFO)

        def ready(self):
            pass

        async def run(self):
            """Run the monitor."""
            self.is_running = True
            while self.is_running:
                now = time.time()
                used_gb, total_gb = self.monitor.get_memory_usage()
                top_n_memory_usage = memory_monitor.get_top_n_memory_usage(n=self.n)
                if used_gb > self.peak_memory_usage:
                    self.peak_memory_usage = used_gb
                    self.peak_top_n_memory_usage = top_n_memory_usage

                if used_gb > total_gb * self.warning_threshold:
                    logging.warning(
                        "The memory usage is high: " f"{used_gb / total_gb * 100}%"
                    )
                if now - self._last_print_time > self.print_interval_s:
                    logging.info(f"Memory usage: {used_gb} / {total_gb}")
                    logging.info(f"Top {self.n} process memory usage:")
                    logging.info(top_n_memory_usage)
                    self._last_print_time = now
                await asyncio.sleep(self.record_interval_s)

        async def stop_run(self):
            """Stop running the monitor.

            Returns:
                True if the monitor is stopped. False otherwise.
            """
            was_running = self.is_running
            self.is_running = False
            return was_running

        async def get_peak_memory_info(self):
            """Return the tuple of the peak memory usage and the
            top n process information during the peak memory usage.
            """
            return self.peak_memory_usage, self.peak_top_n_memory_usage

    current_node_ip = ray.worker.global_worker.node_ip_address
    # Schedule the actor on the current node.
    memory_monitor_actor = MemoryMonitorActor.options(
        resources={f"node:{current_node_ip}": 0.001}
    ).remote(
        print_interval_s=print_interval_s,
        record_interval_s=record_interval_s,
        warning_threshold=warning_threshold,
    )
    print("Waiting for memory monitor actor to be ready...")
    ray.get(memory_monitor_actor.ready.remote())
    print("Memory monitor actor is ready now.")
    memory_monitor_actor.run.remote()
    return memory_monitor_actor
Run the memory monitor actor that prints the memory usage.

    The monitor will run on the same node as this function is called.

    Params:
        interval_s (int): The interval memory usage information is printed
        warning_threshold (float): The threshold where the
            memory usage warning is printed.

    Returns:
        The memory monitor actor.
    
Run the memory monitor actor that prints the memory usage.
The monitor will run on the same node as this function is called.
Run the memory monitor actor that prints the memory usage . The monitor will run on the same node as this function is called .
def monitor_memory_usage(
    print_interval_s: int = 30,
    record_interval_s: int = 5,
    warning_threshold: float = 0.9,
):
    assert ray.is_initialized(), "The API is only available when Ray is initialized."
    @ray.remote(num_cpus=0)
    class MemoryMonitorActor:
        def __init__(
            self,
            print_interval_s: float = 20,
            record_interval_s: float = 5,
            warning_threshold: float = 0.9,
            n: int = 10,
        ):
            self.print_interval_s = print_interval_s
            self.record_interval_s = record_interval_s
            self.is_running = False
            self.warning_threshold = warning_threshold
            self.monitor = memory_monitor.MemoryMonitor()
            self.n = n
            self.peak_memory_usage = 0
            self.peak_top_n_memory_usage = ""
            self._last_print_time = 0
            logging.basicConfig(level=logging.INFO)
        def ready(self):
            pass
        async def run(self):
            self.is_running = True
            while self.is_running:
                now = time.time()
                used_gb, total_gb = self.monitor.get_memory_usage()
                top_n_memory_usage = memory_monitor.get_top_n_memory_usage(n=self.n)
                if used_gb > self.peak_memory_usage:
                    self.peak_memory_usage = used_gb
                    self.peak_top_n_memory_usage = top_n_memory_usage
                if used_gb > total_gb * self.warning_threshold:
                    logging.warning(
                        "The memory usage is high: " f"{used_gb / total_gb * 100}%"
                    )
                if now - self._last_print_time > self.print_interval_s:
                    logging.info(f"Memory usage: {used_gb} / {total_gb}")
                    logging.info(f"Top {self.n} process memory usage:")
                    logging.info(top_n_memory_usage)
                    self._last_print_time = now
                await asyncio.sleep(self.record_interval_s)
        async def stop_run(self):
            was_running = self.is_running
            self.is_running = False
            return was_running
        async def get_peak_memory_info(self):
            return self.peak_memory_usage, self.peak_top_n_memory_usage
    current_node_ip = ray.worker.global_worker.node_ip_address
    memory_monitor_actor = MemoryMonitorActor.options(
        resources={f"node:{current_node_ip}": 0.001}
    ).remote(
        print_interval_s=print_interval_s,
        record_interval_s=record_interval_s,
        warning_threshold=warning_threshold,
    )
    print("Waiting for memory monitor actor to be ready...")
    ray.get(memory_monitor_actor.ready.remote())
    print("Memory monitor actor is ready now.")
    memory_monitor_actor.run.remote()
    return memory_monitor_actor
def monitor_memory_usage ( print_interval_s : int = 30 , record_interval_s : int = 5 , warning_threshold : float = 0.9 , ) : assert ray . is_initialized ( ) , "The API is only available when Ray is initialized." @ ray . remote ( num_cpus = 0 ) class MemoryMonitorActor : def __init__ ( self , print_interval_s : float = 20 , record_interval_s : float = 5 , warning_threshold : float = 0.9 , n : int = 10 , ) : """The actor that monitor the memory usage of the cluster.

            Params:
                print_interval_s (float): The interval where
                    memory usage is printed.
                record_interval_s (float): The interval where
                    memory usage is recorded.
                warning_threshold (float): The threshold where
                    memory warning is printed
                n (int): When memory usage is printed,
                    top n entries are printed.
            """ self . print_interval_s = print_interval_s self . record_interval_s = record_interval_s self . is_running = False self . warning_threshold = warning_threshold self . monitor = memory_monitor . MemoryMonitor ( ) self . n = n self . peak_memory_usage = 0 self . peak_top_n_memory_usage = "" self . _last_print_time = 0 logging . basicConfig ( level = logging . INFO ) def ready ( self ) : pass async def run ( self ) : """Run the monitor.""" self . is_running = True while self . is_running : now = time . time ( ) used_gb , total_gb = self . monitor . get_memory_usage ( ) top_n_memory_usage = memory_monitor . get_top_n_memory_usage ( n = self . n ) if used_gb > self . peak_memory_usage : self . peak_memory_usage = used_gb self . peak_top_n_memory_usage = top_n_memory_usage if used_gb > total_gb * self . warning_threshold : logging . warning ( "The memory usage is high: " f"{used_gb / total_gb * 100}%" ) if now - self . _last_print_time > self . print_interval_s : logging . info ( f"Memory usage: {used_gb} / {total_gb}" ) logging . info ( f"Top {self.n} process memory usage:" ) logging . info ( top_n_memory_usage ) self . _last_print_time = now await asyncio . sleep ( self . record_interval_s ) async def stop_run ( self ) : """Stop running the monitor.

            Returns:
                True if the monitor is stopped. False otherwise.
            """ was_running = self . is_running self . is_running = False return was_running async def get_peak_memory_info ( self ) : """Return the tuple of the peak memory usage and the
            top n process information during the peak memory usage.
            """ return self . peak_memory_usage , self . peak_top_n_memory_usage current_node_ip = ray . worker . global_worker . node_ip_address memory_monitor_actor = MemoryMonitorActor . options ( resources = { f"node:{current_node_ip}" : 0.001 } ) . remote ( print_interval_s = print_interval_s , record_interval_s = record_interval_s , warning_threshold = warning_threshold , ) print ( "Waiting for memory monitor actor to be ready..." ) ray . get ( memory_monitor_actor . ready . remote ( ) ) print ( "Memory monitor actor is ready now." ) memory_monitor_actor . run . remote ( ) return memory_monitor_actor
Run the memory monitor actor that prints the memory usage.
Run the memory monitor actor that prints the memory usage .
"""Run the memory monitor actor that prints the memory usage.

    The monitor will run on the same node as this function is called.

    Params:
        interval_s (int): The interval memory usage information is printed
        warning_threshold (float): The threshold where the
            memory usage warning is printed.

    Returns:
        The memory monitor actor.
    """ """The actor that monitor the memory usage of the cluster.

            Params:
                print_interval_s (float): The interval where
                    memory usage is printed.
                record_interval_s (float): The interval where
                    memory usage is recorded.
                warning_threshold (float): The threshold where
                    memory warning is printed
                n (int): When memory usage is printed,
                    top n entries are printed.
            """ # -- Interval the monitor prints the memory usage information. -- # -- Interval the monitor records the memory usage information. -- # -- Whether or not the monitor is running. -- # -- The used_gb/total_gb threshold where warning message omits. -- # -- The monitor that calculates the memory usage of the node. -- # -- The top n memory usage of processes are printed. -- # -- The peak memory usage in GB during lifetime of monitor. -- # -- The top n memory usage of processes # during peak memory usage. -- # -- The last time memory usage was printed -- # -- logger. -- """Run the monitor.""" """Stop running the monitor.

            Returns:
                True if the monitor is stopped. False otherwise.
            """ """Return the tuple of the peak memory usage and the
            top n process information during the peak memory usage.
            """ # Schedule the actor on the current node.
The memory monitor actor.
The memory monitor actor .
print_interval_s
int
record_interval_s
int
warning_threshold
float
The threshold where the
memory usage warning is printed.
The threshold where the memory usage warning is printed .
interval_s
The interval memory usage information is printed
The interval memory usage information is printed