code: def test_environment_variables_env_caching(shutdown_only):
      """Test that workers with specified envs are cached and reused.
  
      When a new task or actor is created with a new runtime env, a
      new worker process is started.  If a subsequent task or actor
      uses the same runtime env, the same worker process should be
      used.  This function checks the pid of the worker to test this.
      """
      ray.init()
  
      env_var_name = "TEST123"
      val1 = "VAL1"
      val2 = "VAL2"
      assert os.environ.get(env_var_name) is None
  
      def task():
          return os.environ.get(env_var_name), os.getpid()
  
      @ray.remote
      def f():
          return task()
  
      @ray.remote
      def g():
          return task()
  
      def get_options(val):
          return {"runtime_env": {"env_vars": {env_var_name: val}}}
  
      # Empty runtime env does not set our env var.
      assert ray.get(f.remote())[0] is None
  
      # Worker pid1 should have an env var set.
      ret_val1, pid1 = ray.get(f.options(**get_options(val1)).remote())
      assert ret_val1 == val1
  
      # Worker pid2 should have an env var set to something different.
      ret_val2, pid2 = ray.get(g.options(**get_options(val2)).remote())
      assert ret_val2 == val2
  
      # Because the runtime env is different, it should use a different process.
      assert pid1 != pid2
  
      # Call g with an empty runtime env. It shouldn't reuse pid2, because
      # pid2 has an env var set.
      _, pid3 = ray.get(g.remote())
      assert pid2 != pid3
  
      # Call g with the same runtime env as pid2. Check it uses the same process.
      _, pid4 = ray.get(g.options(**get_options(val2)).remote())
      assert pid4 == pid2
  
      # Call f with a different runtime env from pid1.  Check that it uses a new
      # process.
      _, pid5 = ray.get(f.options(**get_options(val2)).remote())
      assert pid5 != pid1
  
      # Call f with the same runtime env as pid1.  Check it uses the same
      # process.
      _, pid6 = ray.get(f.options(**get_options(val1)).remote())
      assert pid6 == pid1
  
      # Same as above but with g instead of f.  Shouldn't affect the outcome.
      _, pid7 = ray.get(g.options(**get_options(val1)).remote())
      assert pid7 == pid1
  
file_path: ray/python/ray/tests/test_runtime_env_env_vars.py
imports: import os
  import sys
  import pytest
  import ray
label_desc: Use image node_id 7 for calling a global function with example usage: test_environment_variables_env_caching(shutdown_only) and returns: os,  os, task, task, dict