text: Title: [BUG] Ray context GC bug
  Body: <!--
  Thank you for your contribution!
  
  Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
  -->
  
  **Describe the bug**
  A clear and concise description of what the bug is.
  
  Set the `DEFAULT_SUBTASK_MONITOR_INTERVAL` to 0 in `mars/services/task/execution/ray/config.py`, then run `mars/dataframe/base/tests/test_base_execution.py::test_cut_execution` with env `MARS_CI_BACKEND=ray`. The case will fail with the following exception:
  
  ``` python
  mars/dataframe/base/tests/test_base_execution.py::test_cut_execution FAILED [100%]
  
  mars/dataframe/base/tests/test_base_execution.py:777 (test_cut_execution)
  setup = <mars.deploy.oscar.session.SyncSession object at 0x12d589e10>
  
      @pytest.mark.ray_dag
      def test_cut_execution(setup):
          session = setup
      
          rs = np.random.RandomState(0)
          raw = rs.random(15) * 1000
          s = pd.Series(raw, index=[f"i{i}" for i in range(15)])
          bins = [10, 100, 500]
          ii = pd.interval_range(10, 500, 3)
          labels = ["a", "b"]
      
          t = tensor(raw, chunk_size=4)
          series = from_pandas_series(s, chunk_size=4)
          iii = from_pandas_index(ii, chunk_size=2)
      
          # cut on Series
          r = cut(series, bins)
          result = r.execute().fetch()
          pd.testing.assert_series_equal(result, pd.cut(s, bins))
      
          r, b = cut(series, bins, retbins=True)
          r_result = r.execute().fetch()
          b_result = b.execute().fetch()
          r_expected, b_expected = pd.cut(s, bins, retbins=True)
          pd.testing.assert_series_equal(r_result, r_expected)
          np.testing.assert_array_equal(b_result, b_expected)
      
          # cut on tensor
          r = cut(t, bins)
          # result and expected is array whose dtype is CategoricalDtype
          result = r.execute().fetch()
          expected = pd.cut(raw, bins)
          assert len(result) == len(expected)
          for r, e in zip(result, expected):
              np.testing.assert_equal(r, e)
      
          # one chunk
          r = cut(s, tensor(bins, chunk_size=2), right=False, include_lowest=True)
          result = r.execute().fetch()
          pd.testing.assert_series_equal(
              result, pd.cut(s, bins, right=False, include_lowest=True)
          )
      
          # test labels
          r = cut(t, bins, labels=labels)
          # result and expected is array whose dtype is CategoricalDtype
          result = r.execute().fetch()
          expected = pd.cut(raw, bins, labels=labels)
          assert len(result) == len(expected)
          for r, e in zip(result, expected):
              np.testing.assert_equal(r, e)
      
          r = cut(t, bins, labels=False)
          # result and expected is array whose dtype is CategoricalDtype
          result = r.execute().fetch()
          expected = pd.cut(raw, bins, labels=False)
          np.testing.assert_array_equal(result, expected)
      
          # test labels which is tensor
          labels_t = tensor(["a", "b"], chunk_size=1)
          r = cut(raw, bins, labels=labels_t, include_lowest=True)
          # result and expected is array whose dtype is CategoricalDtype
          result = r.execute().fetch()
          expected = pd.cut(raw, bins, labels=labels, include_lowest=True)
          assert len(result) == len(expected)
          for r, e in zip(result, expected):
              np.testing.assert_equal(r, e)
      
          # test labels=False
          r, b = cut(raw, ii, labels=False, retbins=True)
          # result and expected is array whose dtype is CategoricalDtype
          r_result, b_result = session.fetch(*session.execute(r, b))
          r_expected, b_expected = pd.cut(raw, ii, labels=False, retbins=True)
          for r, e in zip(r_result, r_expected):
              np.testing.assert_equal(r, e)
          pd.testing.assert_index_equal(b_result, b_expected)
      
          # test bins which is md.IntervalIndex
          r, b = cut(series, iii, labels=tensor(labels, chunk_size=1), retbins=True)
          r_result = r.execute().fetch()
  >       b_result = b.execute().fetch()
  
  mars/dataframe/base/tests/test_base_execution.py:858: 
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  mars/core/entity/executable.py:164: in fetch
      return self._fetch(session=session, **kw)
  mars/core/entity/executable.py:161: in _fetch
      return fetch(self, session=session, **kw)
  mars/deploy/oscar/session.py:1941: in fetch
      return session.fetch(tileable, *tileables, **kwargs)
  mars/deploy/oscar/session.py:1720: in fetch
      return asyncio.run_coroutine_threadsafe(coro, self._loop).result()
  ../../.pyenv/versions/3.7.7/lib/python3.7/concurrent/futures/_base.py:435: in result
      return self.__get_result()
  ../../.pyenv/versions/3.7.7/lib/python3.7/concurrent/futures/_base.py:384: in __get_result
      raise self._exception
  mars/deploy/oscar/session.py:1909: in _fetch
      data = await session.fetch(tileable, *tileables, **kwargs)
  mars/deploy/oscar/tests/session.py:68: in fetch
      results = await super().fetch(*tileables)
  mars/deploy/oscar/session.py:1126: in fetch
      chunk_metas = await self._meta_api.get_chunk_meta.batch(*get_chunk_metas)
  mars/oscar/batch.py:146: in _async_batch
      return [await self._async_call(*args_list[0], **kwargs_list[0])]
  mars/oscar/batch.py:95: in _async_call
      return await self.func(*args, **kwargs)
  mars/services/meta/api/oscar.py:179: in get_chunk_meta
      return await self._meta_store.get_meta(object_id, fields=fields, error=error)
  mars/oscar/core.pyx:263: in __pyx_actor_method_wrapper
      async with lock:
  mars/oscar/core.pyx:266: in mars.oscar.core.__pyx_actor_method_wrapper
      result = await result
  mars/oscar/batch.py:95: in _async_call
      return await self.func(*args, **kwargs)
  mars/services/meta/store/dictionary.py:95: in get_meta
      return self._get_meta(object_id, fields=fields, error=error)
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  
  self = <mars.services.meta.store.dictionary.DictMetaStore object at 0x12ed69210>
  object_id = '8553d79ca62df9bbf3150edc97f20b79_0', fields = ('object_refs',)
  error = 'raise'
  
      def _get_meta(
          self, object_id: str, fields: List[str] = None, error: str = "raise"
      ) -> Dict:
          if error not in ("raise", "ignore"):  # pragma: no cover
              raise ValueError("error must be raise or ignore")
          try:
  >           meta = self._store[object_id]
  E           KeyError: '8553d79ca62df9bbf3150edc97f20b79_0'
  
  mars/services/meta/store/dictionary.py:80: KeyError
  
  ```
  
  The bug was introduced by https://github.com/mars-project/mars/pull/3061.
  
  **To Reproduce**
  To help us reproducing this bug, please provide information below:
  1. Your Python version 3.7.7
  2. The version of Mars you use Latest master
  3. Versions of crucial packages, such as numpy, scipy and pandas
  4. Full stack of the error.
  5. Minimized code to reproduce the error.
  
  **Expected behavior**
  A clear and concise description of what you expected to happen.
  
  **Additional context**
  Add any other context about the problem here.
  