code: def test_concurent_producers(monkeypatch):
      """Test concurrent production of new trials."""
      with create_producer() as (producer, storage):
          trials_in_db_before = len(storage._fetch_trials({}))
          new_trials_in_db_before = len(
              storage._fetch_trials({"status": "new"})
          )
  
          producer.experiment.algorithm.algorithm.possible_values = [
              (1,)
          ]
          # Make sure it starts from index 0
          producer.experiment.algorithm.seed_rng(0)
  
          second_producer = Producer(producer.experiment)
          second_producer.experiment = copy.deepcopy(
              producer.experiment
          )
  
          sleep = 0.5
  
          def suggest(self, num):
              time.sleep(sleep)
              return producer.experiment.algorithm.algorithm.possible_values[
                  0
              ]
  
          monkeypatch.setattr(
              producer.experiment.algorithm.algorithm,
              "suggest",
              suggest,
          )
  
          pool = threading.Pool(2)
          first_result = pool.apply_async(producer.produce)
          second_result = pool.apply_async(
              second_producer.produce, dict(timeout=0)
          )
  
          assert first_result.get(sleep * 5) == 1
  
          # TODO: Use Oríon's custom AcquireLockTimeoutError
          with pytest.raises(TimeoutError):
              second_result.get(sleep * 5)
  
          # `num_new_trials` new trials were registered at database
          assert (
              len(storage._fetch_trials({})) == trials_in_db_before + 1
          )
          assert (
              len(storage._fetch_trials({"status": "new"}))
              == new_trials_in_db_before + 1
          )
          random_dt = NotImplemented  # todo: undefined variable.
          new_trials = list(
              storage._fetch_trials(
                  {"status": "new", "submit_time": random_dt}
              )
          )
          assert new_trials[0].experiment == producer.experiment.id
          assert new_trials[0].start_time is None
          assert new_trials[0].end_time is None
          assert new_trials[0].results == []
          assert new_trials[0].params == {
              "/decoding_layer": "gru",
              "/encoding_layer": "rnn",
          }
  
          assert new_trials[1].params == {
              "/decoding_layer": "gru",
              "/encoding_layer": "gru",
          }
  
file_path: orion/tests/unittests/core/worker/test_producer.py
imports: import contextlib
  import copy
  import threading
  import time
  import pytest
  from orion.core.io.experiment_builder import build
  from orion.core.worker.producer import Producer
  from orion.testing import OrionState, generate_trials
label_desc: Use image node_id 6 for calling a global function with example usage: test_concurent_producers(monkeypatch) and returns: producer