code: from multiprocessing import Pool
  import os
  from typing import Sequence
  from typing import Tuple
  
  import numpy as np
  import pytest
  
  import optuna
  
  
  _STUDY_NAME = "_test_multiprocess"
  
  
  def f(x: float, y: float) -> float:
      return (x - 3) ** 2 + y
  
  
  def objective(trial: optuna.Trial) -> float:
      x = trial.suggest_float("x", -10, 10)
      y = trial.suggest_float("y", -10, 10)
      trial.report(x, 0)
      trial.report(y, 1)
      trial.set_user_attr("x", x)
      trial.set_system_attr("y", y)
      return f(x, y)
  
  
  def run_optimize(args: Tuple[str, str]) -> None:
      study_name = args[0]
      storage_url = args[1]
      # Create a study
      study = optuna.load_study(study_name=study_name, storage=storage_url)
      # Run optimization
      study.optimize(objective, n_trials=20)
  
  
  @pytest.fixture
  def storage_url() -> str:
      if "TEST_DB_URL" not in os.environ:
          pytest.skip("This test requires TEST_DB_URL.")
      storage_url = os.environ["TEST_DB_URL"]
      try:
          optuna.study.delete_study(_STUDY_NAME, storage_url)
      except KeyError:
          pass
      return storage_url
  
  
  def _check_trials(trials: Sequence[optuna.trial.FrozenTrial]) -> None:
      # Check trial states.
      assert all(trial.state == optuna.trial.TrialState.COMPLETE for trial in trials)
  
      # Check trial values and params.
      assert all("x" in trial.params for trial in trials)
      assert all("y" in trial.params for trial in trials)
      assert all(
          np.isclose(
              np.asarray([trial.value for trial in trials]),
              [f(trial.params["x"], trial.params["y"]) for trial in trials],
              atol=1e-4,
          ).tolist()
      )
  
      # Check intermediate values.
      assert all(len(trial.intermediate_values) == 2 for trial in trials)
      assert all(trial.params["x"] == trial.intermediate_values[0] for trial in trials)
      assert all(trial.params["y"] == trial.intermediate_values[1] for trial in trials)
  
      # Check attrs.
      assert all(
          np.isclose(
              [trial.user_attrs["x"] for trial in trials],
              [trial.params["x"] for trial in trials],
              atol=1e-4,
          ).tolist()
      )
      assert all(
          np.isclose(
              [trial.system_attrs["y"] for trial in trials],
              [trial.params["y"] for trial in trials],
              atol=1e-4,
          ).tolist()
      )
  
  
  def test_loaded_trials(storage_url: str) -> None:
      # Please create the tables by placing this function before the multi-process tests.
  
      N_TRIALS = 20
      study = optuna.create_study(study_name=_STUDY_NAME, storage=storage_url)
      # Run optimization
      study.optimize(objective, n_trials=N_TRIALS)
  
      trials = study.trials
      assert len(trials) == N_TRIALS
  
      _check_trials(trials)
  
      # Create a new study to confirm the study can load trial properly.
      loaded_study = optuna.load_study(study_name=_STUDY_NAME, storage=storage_url)
      _check_trials(loaded_study.trials)
  
  
  def test_multiprocess(storage_url: str) -> None:
      n_workers = 8
      study_name = _STUDY_NAME
      optuna.create_study(storage=storage_url, study_name=study_name)
      with Pool(n_workers) as pool:
          pool.map(run_optimize, [(study_name, storage_url)] * n_workers)
  
      study = optuna.load_study(study_name=study_name, storage=storage_url)
  
      trials = study.trials
      assert len(trials) == n_workers * 20
  
      _check_trials(trials)
  
apis: numpy.isclose numpy.asarray