code: def test_searcher_save_restore(
      ray_start_8_cpus, resource_manager_cls, tmpdir
  ):
      """Searchers state should be saved and restored in the experiment checkpoint.
  
      Legacy test: test_trial_runner_3.py::TrialRunnerTest::testSearcherSaveRestore
      """
  
      def create_searcher():
          class TestSuggestion(Searcher):
              def __init__(self, index):
                  self.index = index
                  self.returned_result = []
                  super().__init__(
                      metric="episode_reward_mean", mode="max"
                  )
  
              def suggest(self, trial_id):
                  self.index += 1
                  return {"test_variable": self.index}
  
              def on_trial_complete(
                  self, trial_id, result=None, **kwargs
              ):
                  self.returned_result.append(result)
  
              def save(self, checkpoint_path):
                  with open(checkpoint_path, "wb") as f:
                      pickle.dump(self.__dict__, f)
  
              def restore(self, checkpoint_path):
                  with open(checkpoint_path, "rb") as f:
                      self.__dict__.update(pickle.load(f))
  
          searcher = TestSuggestion(0)
          searcher = ConcurrencyLimiter(searcher, max_concurrent=2)
          searcher = Repeater(searcher, repeat=3, set_index=False)
          search_alg = SearchGenerator(searcher)
          experiment_spec = {
              "run": "__fake",
              "num_samples": 20,
              "config": {"sleep": 10},
              "stop": {"training_iteration": 2},
              "resources_per_trial": PlacementGroupFactory(
                  [{"CPU": 1}]
              ),
          }
          experiments = [Experiment.from_json("test", experiment_spec)]
          search_alg.add_configurations(experiments)
          return search_alg
  
      searcher = create_searcher()
  
      runner = TestTuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          search_alg=searcher,
          checkpoint_period=-1,
          experiment_path=str(tmpdir),
      )
  
      while len(runner.get_trials()) < 6:
          runner.step()
  
      assert len(runner.get_trials()) == 6, [
          t.config for t in runner.get_trials()
      ]
      runner.checkpoint()
      trials = runner.get_trials()
      [
          runner._schedule_trial_stop(t)
          for t in trials
          if t.status is not Trial.ERROR
      ]
  
      runner.cleanup()
  
      del runner
  
      searcher = create_searcher()
  
      runner2 = TestTuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          search_alg=searcher,
          experiment_path=str(tmpdir),
          resume="LOCAL",
      )
  
      assert len(runner2.get_trials()) == 6, [
          t.config for t in runner2.get_trials()
      ]
  
      def trial_statuses():
          return [t.status for t in runner2.get_trials()]
  
      def num_running_trials():
          return sum(
              t.status == Trial.RUNNING for t in runner2.get_trials()
          )
  
      while num_running_trials() < 6:
          runner2.step()
  
      assert len(set(trial_statuses())) == 1
      assert Trial.RUNNING in trial_statuses()
  
      for i in range(20):
          runner2.step()
          assert 1 <= num_running_trials() <= 6
  
      evaluated = [
          t.evaluated_params["test_variable"]
          for t in runner2.get_trials()
      ]
      count = Counter(evaluated)
      assert all(v <= 3 for v in count.values())
  
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 7 for calling a global function with example usage: test_searcher_save_restore(ray_start_8_cpus, resource_manager_cls, tmpdir) and returns: search_alg, unknown, sum, dict