code: def test_search_alg_stalled(
      ray_start_4_cpus_2_gpus_extra, resource_manager_cls
  ):
      """Checks that runner and searcher state is maintained when stalled.
  
      We use a concurrency limit of 1, meaning each trial is added one-by-one
      from the searchers.
  
      We then run three samples. During the second trial, we stall the searcher,
      which means we don't suggest new trials after it finished.
  
      In this case, the runner should still be considered "running". Once we unstall,
      the experiment finishes regularly.
  
      Legacy test: test_trial_runner_3.py::TrialRunnerTest::testSearchAlgStalled
      """
      experiment_spec = {
          "run": "__fake",
          "num_samples": 3,
          "stop": {"training_iteration": 1},
      }
      experiments = [Experiment.from_json("test", experiment_spec)]
      search_alg = _MockSuggestionAlgorithm(max_concurrent=1)
      search_alg.add_configurations(experiments)
      searcher = search_alg.searcher
      runner = TestTuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          search_alg=search_alg,
      )
      runner.step()
      trials = runner.get_trials()
      while trials[0].status != Trial.TERMINATED:
          runner.step()
  
      # On next step, trials[1] is created
      runner.step()
  
      trials = runner.get_trials()
  
      while trials[1].status != Trial.RUNNING:
          runner.step()
  
      assert trials[1].status == Trial.RUNNING
      assert len(searcher.live_trials) == 1
  
      # Stall: We don't suggest new algorithms
      searcher.stall = True
  
      while trials[1].status != Trial.TERMINATED:
          runner.step()
  
      assert trials[1].status == Trial.TERMINATED
      assert len(searcher.live_trials) == 0
  
      assert all(trial.is_finished() for trial in trials)
      assert not search_alg.is_finished()
      assert not runner.is_finished()
  
      # Unstall
      searcher.stall = False
  
      # Create trials[2]
      runner.step()
  
      trials = runner.get_trials()
  
      while trials[2].status != Trial.RUNNING:
          runner.step()
  
      assert trials[2].status == Trial.RUNNING
      assert len(searcher.live_trials) == 1
  
      while trials[2].status != Trial.TERMINATED:
          runner.step()
  
      assert len(searcher.live_trials) == 0
      assert search_alg.is_finished()
      assert runner.is_finished()
  
imports: import os
  import pickle
  from collections import Counter
  import pytest
  import sys
  import ray
  from ray.air.constants import TRAINING_ITERATION
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.tune import Experiment, PlacementGroupFactory
  from ray.tune.execution.tune_controller import TuneController
  from ray.tune.experiment import Trial
  from ray.tune.schedulers import FIFOScheduler, TrialScheduler
  from ray.tune.search import Searcher, ConcurrencyLimiter, Repeater, SearchGenerator
  from ray.tune.search._mock import _MockSuggestionAlgorithm
  from ray.train.tests.util import mock_storage_context
label_desc: Use image node_id 5 for calling a global function with example usage: test_search_alg_stalled(ray_start_4_cpus_2_gpus_extra, resource_manager_cls) without return types