code: def test_executor_context_gc(ray_start_regular_shared2, config):
      session = new_session(
          backend=config["backend"],
          n_cpu=2,
          web=False,
          use_uvloop=False,
          config={
              "task.execution_config.ray.monitor_interval_seconds": 0
          },
      )
  
      assert session._session.client.web_address is None
      assert session.get_web_endpoint() is None
  
      def f1(c):
          time.sleep(0.5)
          return c
  
      with session:
          t1 = mt.random.randint(10, size=(100, 10), chunk_size=100)
          t2 = mt.random.randint(10, size=(100, 10), chunk_size=50)
          t3 = t2 + t1
          t4 = t3.sum(0)
          t5 = t4.map_chunk(f1)
          r = t5.execute()
          result = r.fetch()
          assert result is not None
          assert len(result) == 10
          context = get_context()
          assert len(context._task_context) < 5
  
      session.stop_server()
      assert get_default_async_session() is None
  
file_path: mars/mars/deploy/oscar/tests/test_ray_dag.py
imports: import copy
  import os
  import time
  import pytest
  from ....None import get_context
  from ....None import tensor
  from ....session import new_session, get_default_async_session
  from ....tests import test_session
  from ....tests.core import DICT_NOT_EMPTY, require_ray
  from ....utils import lazy_import
  from ..local import new_cluster
  from ..tests import test_local
  from ..tests.session import new_test_session
  from ..tests.test_local import _cancel_when_tile, _cancel_when_execute
  from .modules.utils import cleanup_third_party_modules_output, get_output_filenames
label_desc: Use image node_id 4 for calling a global function with example usage: test_executor_context_gc(ray_start_regular_shared2, config) and returns: c