text: Title: [BUG] Failed to create Mars DataFrame when mars object exists in a list
  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**
  
  Failed to create Mars DataFrame when mars object exists in a list.
  
  **To Reproduce**
  To help us reproducing this bug, please provide information below:
  1. Your Python version
  2. The version of Mars you use
  3. Versions of crucial packages, such as numpy, scipy and pandas
  4. Full stack of the error.
  5. Minimized code to reproduce the error.
  
  ```
  In [1]: import mars
  
  In [2]: mars.new_session()
  Web service started at http://0.0.0.0:24172
  Out[2]: <mars.deploy.oscar.session.SyncSession at 0x7f8ada249370>
  
  In [3]: import mars.dataframe as md
  
  In [5]: s = md.Series([1, 2, 3])
  
  In [6]: df2 = md.DataFrame({'a': [s.sum()]})
  100%|█████████████████████████████████████| 100.0/100 [00:00<00:00, 1592.00it/s]
  
  In [7]: df2
  Out[7]: DataFrame <op=DataFrameDataSource, key=5a704fd6d6ab7aee6f31d874c2f11347>
  
  In [12]: df2.execute()
    0%|                                                   | 0/100 [00:00<?, ?it/s]Failed to run subtask 00lm3BBKMBsieIow3LtlHrwv on band numa-0
  Traceback (most recent call last):
    File "/Users/qinxuye/Workspace/mars/mars/services/scheduling/worker/execution.py", line 331, in internal_run_subtask
      subtask_info.result = await self._retry_run_subtask(
    File "/Users/qinxuye/Workspace/mars/mars/services/scheduling/worker/execution.py", line 420, in _retry_run_subtask
      return await _retry_run(subtask, subtask_info, _run_subtask_once)
    File "/Users/qinxuye/Workspace/mars/mars/services/scheduling/worker/execution.py", line 107, in _retry_run
      raise ex
    File "/Users/qinxuye/Workspace/mars/mars/services/scheduling/worker/execution.py", line 67, in _retry_run
      return await target_async_func(*args)
    File "/Users/qinxuye/Workspace/mars/mars/services/scheduling/worker/execution.py", line 373, in _run_subtask_once
      return await asyncio.shield(aiotask)
    File "/Users/qinxuye/Workspace/mars/mars/services/subtask/api.py", line 68, in run_subtask_in_slot
      return await ref.run_subtask.options(profiling_context=profiling_context).send(
    File "/Users/qinxuye/Workspace/mars/mars/oscar/backends/context.py", line 183, in send
      future = await self._call(actor_ref.address, message, wait=False)
    File "/Users/qinxuye/Workspace/mars/mars/oscar/backends/context.py", line 61, in _call
      return await self._caller.call(
    File "/Users/qinxuye/Workspace/mars/mars/oscar/backends/core.py", line 95, in call
      await client.send(message)
    File "/Users/qinxuye/Workspace/mars/mars/oscar/backends/communication/base.py", line 258, in send
      return await self.channel.send(message)
    File "/Users/qinxuye/Workspace/mars/mars/oscar/backends/communication/socket.py", line 73, in send
      buffers = await serializer.run()
    File "/Users/qinxuye/Workspace/mars/mars/serialization/aio.py", line 80, in run
      return self._get_buffers()
    File "/Users/qinxuye/Workspace/mars/mars/serialization/aio.py", line 37, in _get_buffers
      headers, buffers = serialize(self._obj)
    File "/Users/qinxuye/Workspace/mars/mars/serialization/core.py", line 363, in serialize
      gen_result = gen_serializer.serialize(gen_to_serial, context)
    File "/Users/qinxuye/Workspace/mars/mars/serialization/core.py", line 72, in wrapped
      return func(self, obj, context)
    File "/Users/qinxuye/Workspace/mars/mars/serialization/core.py", line 151, in serialize
      return {}, pickle_buffers(obj)
    File "/Users/qinxuye/Workspace/mars/mars/serialization/core.py", line 88, in pickle_buffers
      buffers[0] = cloudpickle.dumps(
    File "/Users/qinxuye/miniconda3/envs/mars3.8/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
      cp.dump(obj)
    File "/Users/qinxuye/miniconda3/envs/mars3.8/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump
      return Pickler.dump(self, obj)
  TypeError: cannot pickle 'weakref' object
  100%|█████████████████████████████████████| 100.0/100 [00:00<00:00, 3581.29it/s]
  ---------------------------------------------------------------------------
  TypeError                                 Traceback (most recent call last)
  <ipython-input-12-fe40b754f95d> in <module>
  ----> 1 df2.execute()
  
  ~/Workspace/mars/mars/core/entity/tileables.py in execute(self, session, **kw)
      460 
      461     def execute(self, session=None, **kw):
  --> 462         result = self.data.execute(session=session, **kw)
      463         if isinstance(result, TILEABLE_TYPE):
      464             return self
  
  ~/Workspace/mars/mars/core/entity/executable.py in execute(self, session, **kw)
       96 
       97         session = _get_session(self, session)
  ---> 98         return execute(self, session=session, **kw)
       99 
      100     def _check_session(self, session: SessionType, action: str):
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in execute(tileable, session, wait, new_session_kwargs, show_progress, progress_update_interval, *tileables, **kwargs)
     1777         session = get_default_or_create(**(new_session_kwargs or dict()))
     1778     session = _ensure_sync(session)
  -> 1779     return session.execute(
     1780         tileable,
     1781         *tileables,
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in execute(self, tileable, show_progress, *tileables, **kwargs)
     1575         fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
     1576         try:
  -> 1577             execution_info: ExecutionInfo = fut.result(
     1578                 timeout=self._isolated_session.timeout
     1579             )
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/concurrent/futures/_base.py in result(self, timeout)
      437                 raise CancelledError()
      438             elif self._state == FINISHED:
  --> 439                 return self.__get_result()
      440             else:
      441                 raise TimeoutError()
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/concurrent/futures/_base.py in __get_result(self)
      386     def __get_result(self):
      387         if self._exception:
  --> 388             raise self._exception
      389         else:
      390             return self._result
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in _execute(session, wait, show_progress, progress_update_interval, cancelled, *tileables, **kwargs)
     1757             # set cancelled to avoid wait task leak
     1758             cancelled.set()
  -> 1759         await execution_info
     1760     else:
     1761         return execution_info
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in wait()
      100 
      101             async def wait():
  --> 102                 return await self._aio_task
      103 
      104             self._future_local.future = fut = asyncio.run_coroutine_threadsafe(
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in _run_in_background(self, tileables, task_id, progress, profiling)
      905                     )
      906                 if task_result.error:
  --> 907                     raise task_result.error.with_traceback(task_result.traceback)
      908             if cancelled:
      909                 return
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in internal_run_subtask(self, subtask, band_name)
      329 
      330             batch_quota_req = {(subtask.session_id, subtask.subtask_id): calc_size}
  --> 331             subtask_info.result = await self._retry_run_subtask(
      332                 subtask, band_name, subtask_api, batch_quota_req
      333             )
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _retry_run_subtask(self, subtask, band_name, subtask_api, batch_quota_req)
      418         #  any exceptions occurred.
      419         if subtask.retryable:
  --> 420             return await _retry_run(subtask, subtask_info, _run_subtask_once)
      421         else:
      422             try:
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _retry_run(subtask, subtask_info, target_async_func, *args)
      105                 )
      106             else:
  --> 107                 raise ex
      108 
      109 
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _retry_run(subtask, subtask_info, target_async_func, *args)
       65     while True:
       66         try:
  ---> 67             return await target_async_func(*args)
       68         except (OSError, MarsError) as ex:
       69             if subtask_info.num_retries < subtask_info.max_retries:
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _run_subtask_once()
      371                     subtask_api.run_subtask_in_slot(band_name, slot_id, subtask)
      372                 )
  --> 373                 return await asyncio.shield(aiotask)
      374             except asyncio.CancelledError as ex:
      375                 # make sure allocated slots are traced
  
  ~/Workspace/mars/mars/services/subtask/api.py in run_subtask_in_slot(self, band_name, slot_id, subtask)
       66             ProfilingContext(task_id=subtask.task_id) if enable_profiling else None
       67         )
  ---> 68         return await ref.run_subtask.options(profiling_context=profiling_context).send(
       69             subtask
       70         )
  
  ~/Workspace/mars/mars/oscar/backends/context.py in send(self, actor_ref, message, wait_response, profiling_context)
      181         ):
      182             detect_cycle_send(message, wait_response)
  --> 183             future = await self._call(actor_ref.address, message, wait=False)
      184             if wait_response:
      185                 result = await self._wait(future, actor_ref.address, message)
  
  ~/Workspace/mars/mars/oscar/backends/context.py in _call(self, address, message, wait)
       59         self, address: str, message: _MessageBase, wait: bool = True
       60     ) -> Union[ResultMessage, ErrorMessage, asyncio.Future]:
  ---> 61         return await self._caller.call(
       62             Router.get_instance_or_empty(), address, message, wait=wait
       63         )
  
  ~/Workspace/mars/mars/oscar/backends/core.py in call(self, router, dest_address, message, wait)
       93         with Timer() as timer:
       94             try:
  ---> 95                 await client.send(message)
       96             except ConnectionError:
       97                 try:
  
  ~/Workspace/mars/mars/oscar/backends/communication/base.py in send(self, message)
      256     @implements(Channel.send)
      257     async def send(self, message):
  --> 258         return await self.channel.send(message)
      259 
      260     @implements(Channel.recv)
  
  ~/Workspace/mars/mars/oscar/backends/communication/socket.py in send(self, message)
       71         compress = self.compression or 0
       72         serializer = AioSerializer(message, compress=compress)
  ---> 73         buffers = await serializer.run()
       74 
       75         # write buffers
  
  ~/Workspace/mars/mars/serialization/aio.py in run(self)
       78 
       79     async def run(self):
  ---> 80         return self._get_buffers()
       81 
       82 
  
  ~/Workspace/mars/mars/serialization/aio.py in _get_buffers(self)
       35 
       36     def _get_buffers(self):
  ---> 37         headers, buffers = serialize(self._obj)
       38 
       39         def _is_cuda_buffer(buf):  # pragma: no cover
  
  ~/Workspace/mars/mars/serialization/core.py in serialize(obj, context)
      361                 gen_to_serial = gen.send(last_serial)
      362                 gen_serializer = _serial_dispatcher.get_handler(type(gen_to_serial))
  --> 363                 gen_result = gen_serializer.serialize(gen_to_serial, context)
      364                 if isinstance(gen_result, types.GeneratorType):
      365                     # when intermediate result still generator, push its contexts
  
  ~/Workspace/mars/mars/serialization/core.py in wrapped(self, obj, context)
       70         else:
       71             context[id(obj)] = obj
  ---> 72             return func(self, obj, context)
       73 
       74     return wrapped
  
  ~/Workspace/mars/mars/serialization/core.py in serialize(self, obj, context)
      149     @buffered
      150     def serialize(self, obj, context: Dict):
  --> 151         return {}, pickle_buffers(obj)
      152 
      153     def deserialize(self, header: Dict, buffers: List, context: Dict):
  
  ~/Workspace/mars/mars/serialization/core.py in pickle_buffers(obj)
       86             buffers.append(memoryview(x))
       87 
  ---> 88         buffers[0] = cloudpickle.dumps(
       89             obj,
       90             buffer_callback=buffer_cb,
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py in dumps(obj, protocol, buffer_callback)
       71                 file, protocol=protocol, buffer_callback=buffer_callback
       72             )
  ---> 73             cp.dump(obj)
       74             return file.getvalue()
       75 
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py in dump(self, obj)
      561     def dump(self, obj):
      562         try:
  --> 563             return Pickler.dump(self, obj)
      564         except RuntimeError as e:
      565             if "recursion" in e.args[0]:
  
  TypeError: cannot pickle 'weakref' object
  ```
  
  