rllib/utils/debug/memory.py
def check_memory_leaks(
    trainer,
    to_check: Optional[Set[str]] = None,
    repeats: Optional[int] = None,
    max_num_trials: int = 3,
) -> DefaultDict[str, List[Suspect]]:
    """Diagnoses the given trainer for possible memory leaks.

    Isolates single components inside the trainer's local worker, e.g. the env,
    policy, etc.. and calls some of their methods repeatedly, while checking
    the memory footprints and keeping track of which lines in the code add
    un-GC'd items to memory.

    Args:
        trainer: The Trainer instance to test.
        to_check: Set of strings to indentify components to test. Allowed strings
            are: "env", "policy", "model", "rollout_worker". By default, check all
            of these.
        repeats: Number of times the test code block should get executed (per trial).
            If a trial fails, a new trial may get started with a larger number of
            repeats: actual_repeats = `repeats` * (trial + 1) (1st trial == 0).
        max_num_trials: The maximum number of trials to run each check for.

    Raises:
        A defaultdict(list) with keys being the `to_check` strings and values being
        lists of Suspect instances that were found.
    """
    local_worker = trainer.workers.local_worker()

    # Which components should we test?
    to_check = to_check or {"env", "model", "policy", "rollout_worker"}

    results_per_category = defaultdict(list)

    # Test a single sub-env (first in the VectorEnv)?
    if "env" in to_check:
        assert local_worker.async_env is not None, (
            "ERROR: Cannot test 'env' since given trainer does not have one "
            "in its local worker. Try setting `create_env_on_driver=True`."
        )

        # Isolate the first sub-env in the vectorized setup and test it.
        env = local_worker.async_env.get_sub_environments()[0]
        action_space = env.action_space
        # Always use same action to avoid numpy random caused memory leaks.
        action_sample = action_space.sample()

        def code():
            env.reset()
            while True:
                # If masking is used, try something like this:
                # np.random.choice(
                #    action_space.n, p=(obs["action_mask"] / sum(obs["action_mask"])))
                _, _, done, _ = env.step(action_sample)
                if done:
                    break

        test = _test_some_code_for_memory_leaks(
            desc="Looking for leaks in env, running through episodes.",
            init=None,
            code=code,
            # How many times to repeat the function call?
            repeats=repeats or 200,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["env"].extend(test)

    # Test the policy (single-agent case only so far).
    if "policy" in to_check:
        policy = local_worker.policy_map[DEFAULT_POLICY_ID]

        # Get a fixed obs (B=10).
        obs = tree.map_structure(
            lambda s: np.stack([s] * 10, axis=0), policy.observation_space.sample()
        )

        print("Looking for leaks in Policy")

        def code():
            policy.compute_actions_from_input_dict(
                {
                    "obs": obs,
                }
            )

        # Call `compute_actions_from_input_dict()` n times.
        test = _test_some_code_for_memory_leaks(
            desc="Calling `compute_actions_from_input_dict()`.",
            init=None,
            code=code,
            # How many times to repeat the function call?
            repeats=repeats or 400,
            # How many times to re-try if we find a suspicious memory
            # allocation?
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["policy"].extend(test)

        # Call `learn_on_batch()` n times.
        dummy_batch = policy._get_dummy_batch_from_view_requirements(batch_size=16)

        test = _test_some_code_for_memory_leaks(
            desc="Calling `learn_on_batch()`.",
            init=None,
            code=lambda: policy.learn_on_batch(dummy_batch),
            # How many times to repeat the function call?
            repeats=repeats or 100,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["policy"].extend(test)

    # Test only the model.
    if "model" in to_check:
        policy = local_worker.policy_map[DEFAULT_POLICY_ID]

        # Get a fixed obs.
        obs = tree.map_structure(lambda s: s[None], policy.observation_space.sample())

        print("Looking for leaks in Model")

        # Call `compute_actions_from_input_dict()` n times.
        test = _test_some_code_for_memory_leaks(
            desc="Calling `[model]()`.",
            init=None,
            code=lambda: policy.model({SampleBatch.OBS: obs}),
            # How many times to repeat the function call?
            repeats=repeats or 400,
            # How many times to re-try if we find a suspicious memory
            # allocation?
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["model"].extend(test)

    # Test the RolloutWorker.
    if "rollout_worker" in to_check:
        print("Looking for leaks in local RolloutWorker")

        def code():
            local_worker.sample()
            local_worker.get_metrics()

        # Call `compute_actions_from_input_dict()` n times.
        test = _test_some_code_for_memory_leaks(
            desc="Calling `sample()` and `get_metrics()`.",
            init=None,
            code=code,
            # How many times to repeat the function call?
            repeats=repeats or 200,
            # How many times to re-try if we find a suspicious memory
            # allocation?
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["rollout_worker"].extend(test)

    return results_per_category
Diagnoses the given trainer for possible memory leaks.

    Isolates single components inside the trainer's local worker, e.g. the env,
    policy, etc.. and calls some of their methods repeatedly, while checking
    the memory footprints and keeping track of which lines in the code add
    un-GC'd items to memory.

    Args:
        trainer: The Trainer instance to test.
        to_check: Set of strings to indentify components to test. Allowed strings
            are: "env", "policy", "model", "rollout_worker". By default, check all
            of these.
        repeats: Number of times the test code block should get executed (per trial).
            If a trial fails, a new trial may get started with a larger number of
            repeats: actual_repeats = `repeats` * (trial + 1) (1st trial == 0).
        max_num_trials: The maximum number of trials to run each check for.

    Raises:
        A defaultdict(list) with keys being the `to_check` strings and values being
        lists of Suspect instances that were found.
    
Diagnoses the given trainer for possible memory leaks.
Isolates single components inside the trainer's local worker, e.g. the env,
policy, etc.. and calls some of their methods repeatedly, while checking
the memory footprints and keeping track of which lines in the code add
un-GC'd items to memory.

The Trainer instance to test.
to_check: Set of strings to indentify components to test.

A defaultdict(list) with keys being the `to_check` strings and values being
lists of Suspect instances that were found.
Diagnoses the given trainer for possible memory leaks . Isolates single components inside the trainer ' s local worker e . g . the env policy etc .. and calls some of their methods repeatedly while checking the memory footprints and keeping track of which lines in the code add un - GC ' d items to memory . The Trainer instance to test . to_check : Set of strings to indentify components to test . A defaultdict ( list ) with keys being the ` to_check ` strings and values being lists of Suspect instances that were found .
def check_memory_leaks(
    trainer,
    to_check: Optional[Set[str]] = None,
    repeats: Optional[int] = None,
    max_num_trials: int = 3,
) -> DefaultDict[str, List[Suspect]]:
    local_worker = trainer.workers.local_worker()
    to_check = to_check or {"env", "model", "policy", "rollout_worker"}
    results_per_category = defaultdict(list)
    if "env" in to_check:
        assert local_worker.async_env is not None, (
            "ERROR: Cannot test 'env' since given trainer does not have one "
            "in its local worker. Try setting `create_env_on_driver=True`."
        )
        env = local_worker.async_env.get_sub_environments()[0]
        action_space = env.action_space
        action_sample = action_space.sample()
        def code():
            env.reset()
            while True:
                _, _, done, _ = env.step(action_sample)
                if done:
                    break
        test = _test_some_code_for_memory_leaks(
            desc="Looking for leaks in env, running through episodes.",
            init=None,
            code=code,
            repeats=repeats or 200,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["env"].extend(test)
    if "policy" in to_check:
        policy = local_worker.policy_map[DEFAULT_POLICY_ID]
        obs = tree.map_structure(
            lambda s: np.stack([s] * 10, axis=0), policy.observation_space.sample()
        )
        print("Looking for leaks in Policy")
        def code():
            policy.compute_actions_from_input_dict(
                {
                    "obs": obs,
                }
            )
        test = _test_some_code_for_memory_leaks(
            desc="Calling `compute_actions_from_input_dict()`.",
            init=None,
            code=code,
            repeats=repeats or 400,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["policy"].extend(test)
        dummy_batch = policy._get_dummy_batch_from_view_requirements(batch_size=16)
        test = _test_some_code_for_memory_leaks(
            desc="Calling `learn_on_batch()`.",
            init=None,
            code=lambda: policy.learn_on_batch(dummy_batch),
            repeats=repeats or 100,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["policy"].extend(test)
    if "model" in to_check:
        policy = local_worker.policy_map[DEFAULT_POLICY_ID]
        obs = tree.map_structure(lambda s: s[None], policy.observation_space.sample())
        print("Looking for leaks in Model")
        test = _test_some_code_for_memory_leaks(
            desc="Calling `[model]()`.",
            init=None,
            code=lambda: policy.model({SampleBatch.OBS: obs}),
            repeats=repeats or 400,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["model"].extend(test)
    if "rollout_worker" in to_check:
        print("Looking for leaks in local RolloutWorker")
        def code():
            local_worker.sample()
            local_worker.get_metrics()
        test = _test_some_code_for_memory_leaks(
            desc="Calling `sample()` and `get_metrics()`.",
            init=None,
            code=code,
            repeats=repeats or 200,
            max_num_trials=max_num_trials,
        )
        if test:
            results_per_category["rollout_worker"].extend(test)
    return results_per_category
def check_memory_leaks ( trainer , to_check : Optional [ Set [ str ] ] = None , repeats : Optional [ int ] = None , max_num_trials : int = 3 , ) -> DefaultDict [ str , List [ Suspect ] ] : local_worker = trainer . workers . local_worker ( ) to_check = to_check or { "env" , "model" , "policy" , "rollout_worker" } results_per_category = defaultdict ( list ) if "env" in to_check : assert local_worker . async_env is not None , ( "ERROR: Cannot test 'env' since given trainer does not have one " "in its local worker. Try setting `create_env_on_driver=True`." ) env = local_worker . async_env . get_sub_environments ( ) [ 0 ] action_space = env . action_space action_sample = action_space . sample ( ) def code ( ) : env . reset ( ) while True : _ , _ , done , _ = env . step ( action_sample ) if done : break test = _test_some_code_for_memory_leaks ( desc = "Looking for leaks in env, running through episodes." , init = None , code = code , repeats = repeats or 200 , max_num_trials = max_num_trials , ) if test : results_per_category [ "env" ] . extend ( test ) if "policy" in to_check : policy = local_worker . policy_map [ DEFAULT_POLICY_ID ] obs = tree . map_structure ( lambda s : np . stack ( [ s ] * 10 , axis = 0 ) , policy . observation_space . sample ( ) ) print ( "Looking for leaks in Policy" ) def code ( ) : policy . compute_actions_from_input_dict ( { "obs" : obs , } ) test = _test_some_code_for_memory_leaks ( desc = "Calling `compute_actions_from_input_dict()`." , init = None , code = code , repeats = repeats or 400 , max_num_trials = max_num_trials , ) if test : results_per_category [ "policy" ] . extend ( test ) dummy_batch = policy . _get_dummy_batch_from_view_requirements ( batch_size = 16 ) test = _test_some_code_for_memory_leaks ( desc = "Calling `learn_on_batch()`." , init = None , code = lambda : policy . learn_on_batch ( dummy_batch ) , repeats = repeats or 100 , max_num_trials = max_num_trials , ) if test : results_per_category [ "policy" ] . extend ( test ) if "model" in to_check : policy = local_worker . policy_map [ DEFAULT_POLICY_ID ] obs = tree . map_structure ( lambda s : s [ None ] , policy . observation_space . sample ( ) ) print ( "Looking for leaks in Model" ) test = _test_some_code_for_memory_leaks ( desc = "Calling `[model]()`." , init = None , code = lambda : policy . model ( { SampleBatch . OBS : obs } ) , repeats = repeats or 400 , max_num_trials = max_num_trials , ) if test : results_per_category [ "model" ] . extend ( test ) if "rollout_worker" in to_check : print ( "Looking for leaks in local RolloutWorker" ) def code ( ) : local_worker . sample ( ) local_worker . get_metrics ( ) test = _test_some_code_for_memory_leaks ( desc = "Calling `sample()` and `get_metrics()`." , init = None , code = code , repeats = repeats or 200 , max_num_trials = max_num_trials , ) if test : results_per_category [ "rollout_worker" ] . extend ( test ) return results_per_category
Diagnoses the given trainer for possible memory leaks.
Diagnoses the given trainer for possible memory leaks .
"""Diagnoses the given trainer for possible memory leaks.

    Isolates single components inside the trainer's local worker, e.g. the env,
    policy, etc.. and calls some of their methods repeatedly, while checking
    the memory footprints and keeping track of which lines in the code add
    un-GC'd items to memory.

    Args:
        trainer: The Trainer instance to test.
        to_check: Set of strings to indentify components to test. Allowed strings
            are: "env", "policy", "model", "rollout_worker". By default, check all
            of these.
        repeats: Number of times the test code block should get executed (per trial).
            If a trial fails, a new trial may get started with a larger number of
            repeats: actual_repeats = `repeats` * (trial + 1) (1st trial == 0).
        max_num_trials: The maximum number of trials to run each check for.

    Raises:
        A defaultdict(list) with keys being the `to_check` strings and values being
        lists of Suspect instances that were found.
    """ # Which components should we test? # Test a single sub-env (first in the VectorEnv)? # Isolate the first sub-env in the vectorized setup and test it. # Always use same action to avoid numpy random caused memory leaks. # If masking is used, try something like this: # np.random.choice( #    action_space.n, p=(obs["action_mask"] / sum(obs["action_mask"]))) # How many times to repeat the function call? # Test the policy (single-agent case only so far). # Get a fixed obs (B=10). # Call `compute_actions_from_input_dict()` n times. # How many times to repeat the function call? # How many times to re-try if we find a suspicious memory # allocation? # Call `learn_on_batch()` n times. # How many times to repeat the function call? # Test only the model. # Get a fixed obs. # Call `compute_actions_from_input_dict()` n times. # How many times to repeat the function call? # How many times to re-try if we find a suspicious memory # allocation? # Test the RolloutWorker. # Call `compute_actions_from_input_dict()` n times. # How many times to repeat the function call? # How many times to re-try if we find a suspicious memory # allocation?
trainer
to_check
Optional[Set[str]]
repeats
Optional[int]
max_num_trials
int