[BUG] The shared_memory storage backend (Python <= 3.7) may lose data when process crash

Nobody has claimed this yet.

**Describe the bug**
``` python
mars/deploy/oscar/tests/test_fail.py:73 (test_rerun_subtask_describe[fault_config0])
fault_cluster =
fault_config = [, {: 0}, []]

@pytest.mark.parametrize('fault_config',
[[FaultType.ProcessExit, {FaultPosition.ON_EXECUTE_OPERAND: 1},
[PSRSConcatPivot]]])
@pytest.mark.asyncio
async def test_rerun_subtask_describe(fault_cluster, fault_config):
fault_type, fault_count, fault_op_types = fault_config
name = await create_fault_injection_manager(
session_id=fault_cluster.session.session_id,
address=fault_cluster.session.address,
fault_count=fault_count,
fault_type=fault_type,
fault_op_types=fault_op_types)
extra_config = {ExtraConfigKey.FAULT_INJECTION_MANAGER_NAME: name}
session = get_default_async_session()

s = np.random.RandomState(0)
raw = pd.DataFrame(s.rand(100, 4), columns=list('abcd'))
df = md.DataFrame(raw, chunk_size=30)

r = df.describe()
> info = await session.execute(r, extra_config=extra_config)

test_fail.py:95:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../session.py:77: in wait
return await self._aio_task
../session.py:709: in _run_in_background
raise task_result.error.with_traceback(task_result.traceback)
../../../services/scheduling/worker/execution.py:266: in internal_run_subtask
subtask, band_name, subtask_api, batch_quota_req)
../../../services/scheduling/worker/execution.py:361: in _retry_run_subtask
return await _retry_run(subtask, subtask_info, _run_subtask_once)
../../../services/scheduling/worker/execution.py:75: in _retry_run
raise ex
../../../services/scheduling/worker/execution.py:68: in _retry_run
return await target_async_func(*args)
../../../services/scheduling/worker/execution.py:350: in _run_subtask_once
raise ex
../../../services/scheduling/worker/execution.py:319: in _run_subtask_once
return await asyncio.shield(aiotask)
../../../services/subtask/api.py:59: in run_subtask_in_slot
return await ref.run_subtask(subtask)
../../../oscar/backends/context.py:154: in send
return self._process_result_message(result)
../../../oscar/backends/context.py:59: in _process_result_message
raise message.error.with_traceback(message.traceback)
../../../oscar/backends/pool.py:496: in send
result = await future
../../../oscar/api.py:118: in __on_receive__
return await super().__on_receive__(message)
mars/oscar/core.pyx:351: in __on_receive__
raise ex
mars/oscar/core.pyx:345: in mars.oscar.core._BaseActor.__on_receive__
return await self._handle_actor_result(result)
mars/oscar/core.pyx:250: in _handle_actor_result
result = list(dones)[0].result()
mars/oscar/core.pyx:273: in _run_actor_async_generator
with debug_async_timeout('actor_lock_timeout',
mars/oscar/core.pyx:275: in mars.oscar.core._BaseActor._run_actor_async_generator
async with self._lock:
mars/oscar/core.pyx:279: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await gen.athrow(*res)
../../../services/subtask/worker/runner.py:102: in run_subtask
result = yield self._running_processor.run(subtask)
mars/oscar/core.pyx:284: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await self._handle_actor_result(res)
mars/oscar/core.pyx:219: in _handle_actor_result
result = await result
../../../oscar/backends/context.py:154: in send
return self._process_result_message(result)
../../../oscar/backends/context.py:59: in _process_result_message
raise message.error.with_traceback(message.traceback)
../../../oscar/backends/pool.py:496: in send
result = await future
../../../oscar/api.py:118: in __on_receive__
return await super().__on_receive__(message)
mars/oscar/core.pyx:351: in __on_receive__
raise ex
mars/oscar/core.pyx:345: in mars.oscar.core._BaseActor.__on_receive__
return await self._handle_actor_result(result)
mars/oscar/core.pyx:250: in _handle_actor_result
result = list(dones)[0].result()
mars/oscar/core.pyx:273: in _run_actor_async_generator
with debug_async_timeout('actor_lock_timeout',
mars/oscar/core.pyx:275: in mars.oscar.core._BaseActor._run_actor_async_generator
async with self._lock:
mars/oscar/core.pyx:279: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await gen.athrow(*res)
../../../services/subtask/worker/processor.py:470: in run
result = yield self._running_aio_task
mars/oscar/core.pyx:284: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await self._handle_actor_result(res)
mars/oscar/core.pyx:219: in _handle_actor_result
result = await result
../../../services/tests/fault_injection_patch.py:75: in run
return await super().run()
../../../services/subtask/worker/processor.py:356: in run
input_keys = await self._load_input_data()
../../../services/subtask/worker/processor.py:114: in _load_input_data
inputs = await self._storage_api.get.batch(*gets)
../../../oscar/batch.py:135: in _async_batch
return await self.batch_func(args_list, kwargs_list)
../../../services/storage/api/oscar.py:97: in batch_get
return await self._storage_handler_ref.get.batch(*gets)
../../../oscar/backends/context.py:154: in send
return self._process_result_message(result)
../../../oscar/backends/context.py:59: in _process_result_message
raise message.error.with_traceback(message.traceback)
../../../oscar/backends/pool.py:496: in send
result = await future
../../../oscar/api.py:118: in __on_receive__
return await super().__on_receive__(message)
mars/oscar/core.pyx:351: in __on_receive__
raise ex
mars/oscar/core.pyx:345: in mars.oscar.core._BaseActor.__on_receive__
return await self._handle_actor_result(result)
mars/oscar/core.pyx:250: in _handle_actor_result
result = list(dones)[0].result()
mars/oscar/core.pyx:273: in _run_actor_async_generator
with debug_async_timeout('actor_lock_timeout',
mars/oscar/core.pyx:275: in mars.oscar.core._BaseActor._run_actor_async_generator
async with self._lock:
mars/oscar/core.pyx:279: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await gen.athrow(*res)
../../../services/storage/handler.py:133: in batch_get
result = yield self._get_data(data_info, conditions)
mars/oscar/core.pyx:284: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await self._handle_actor_result(res)
mars/oscar/core.pyx:250: in _handle_actor_result
result = list(dones)[0].result()
mars/oscar/core.pyx:273: in _run_actor_async_generator
with debug_async_timeout('actor_lock_timeout',
mars/oscar/core.pyx:275: in mars.oscar.core._BaseActor._run_actor_async_generator
async with self._lock:
mars/oscar/core.pyx:279: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await gen.athrow(*res)
../../../services/storage/handler.py:80: in _get_data
data_info.object_id)
mars/oscar/core.pyx:284: in mars.oscar.core._BaseActor._run_actor_async_generator
res = await self._handle_actor_result(res)
mars/oscar/core.pyx:219: in _handle_actor_result
result = await result
../../../storage/shared_memory.py:152: in get
return await deserializer.run()
../../../serialization/aio.py:121: in run
return await self._get_obj()
../../../serialization/aio.py:112: in _get_obj
header = pickle.loads(await self._get_obj_header_bytes())
../../../serialization/aio.py:97: in _get_obj_header_bytes
header_bytes = bytes(await self._file.read(11))
../../../lib/aio/base.py:25: in method
return await self._loop.run_in_executor(self._executor, func)
/Users/po/.pyenv/versions/3.7.7/lib/python3.7/concurrent/futures/thread.py:57: in run
result = self.fn(*self.args, **self.kwargs)
../../../storage/core.py:94: in read
self._read_init()
../../../storage/shared_memory.py:78: in _read_init
self.shm = shm = SharedMemoryForRead(name=self._object_id)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

> mode=self._mode
)
E FileNotFoundError: [Errno 2] No such file or directory: '/rF38QVKnty29i3ER4PTdW42ww2Nqqo'

/Users/admin/.pyenv/versions/3.7.7/lib/python3.7/site-packages/shared_memory/shared_memory.py:103: FileNotFoundError
```

**To Reproduce**
To help us reproducing this bug, please provide information below:
1. Python 3.6 & 3.7
2. The latest master

```python
# Copyright 1999-2021 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import pytest
import numpy as np
import pandas as pd

import mars.dataframe as md
from mars.tensor.base.psrs import PSRSConcatPivot
from mars.services.tests.fault_injection_manager import (
AbstractFaultInjectionManager,
ExtraConfigKey,
FaultPosition,
FaultType,
)
from mars.deploy.oscar.local import new_cluster
from mars.deploy.oscar.service import load_config
from mars.deploy.oscar.session import get_default_async_session

@pytest.fixture
async def fault_cluster(request):
start_method = os.environ.get('POOL_START_METHOD', None)
config = load_config()
config['third_party_modules'] = ['mars.services.tests.fault_injection_patch']
config['scheduling']['subtask_max_retries'] = 2
config['storage']['backends'] = ['shared_memory']
client = await new_cluster(subprocess_start_method=start_method,
config=config,
n_worker=2,
n_cpu=2)
async with client:
yield client

async def create_fault_injection_manager(session_id, address, fault_count, fault_type, fault_op_types=None):
class FaultInjectionManager(AbstractFaultInjectionManager):
def __init__(self):
self._fault_count = fault_count

def set_fault_count(self, count):
self._fault_count = count

def get_fault_count(self):
return self._fault_count

def get_fault(self, pos: FaultPosition, ctx=None) -> FaultType:
# Check op types if fault_op_types provided.
if fault_op_types and type(ctx.get('operand')) not in fault_op_types:
return FaultType.NoFault
if self._fault_count.get(pos, 0) > 0:
self._fault_count[pos] -= 1
if fault_op_types:
print(f"get_fault {fault_type} {pos} {ctx}")
return fault_type
return FaultType.NoFault

await FaultInjectionManager.create(session_id, address)
return FaultInjectionManager.name

@pytest.mark.parametrize('fault_config',
[[FaultType.ProcessExit, {FaultPosition.ON_EXECUTE_OPERAND: 1},
[PSRSConcatPivot]]])
@pytest.mark.asyncio
async def test_rerun_subtask_describe(fault_cluster, fault_config):
fault_type, fault_count, fault_op_types = fault_config
name = await create_fault_injection_manager(
session_id=fault_cluster.session.session_id,
address=fault_cluster.session.address,
fault_count=fault_count,
fault_type=fault_type,
fault_op_types=fault_op_types)
extra_config = {ExtraConfigKey.FAULT_INJECTION_MANAGER_NAME: name}
session = get_default_async_session()

s = np.random.RandomState(0)
raw = pd.DataFrame(s.rand(100, 4), columns=list('abcd'))
df = md.DataFrame(raw, chunk_size=30)

r = df.describe()
info = await session.execute(r, extra_config=extra_config)
await info
assert info.result() is None
assert info.exception() is None
assert info.progress() == 1
res = await session.fetch(r)
pd.testing.assert_frame_equal(res, raw.describe())

fault_injection_manager = await session.get_remote_object(
fault_cluster.session.session_id, name)
remain_fault_count = await fault_injection_manager.get_fault_count()
for key in fault_count:
assert remain_fault_count[key] == 0
```

**Additional context**
The data is lost in the shared memory storage backend. I noticed that the shared memory is a backport https://pypi.org/project/shared-memory38/ in Python 3.6 & 3.7, and this bug is reproduced only in Python 3.6 & 3.7. I think it may be a backport bug in the `shared-memory38`.

I digged into the bug (_the screen shot shows a different key because i reproduce this problem many times_), it shows that the lost object is putted into the shared memory success, then the process is crashed.
After that, getting the object will raise the above exception, the object does not exists in the shared memory store.

Start with storage/shared_memory.py and serialization/aio.py, then run mars/deploy/oscar/tests/test_fail.py::test_rerun_subtask_describe using the shared_memory backend on Python 3.6 or 3.7. Trace the process-exit retry path shown in the report and verify that the rerun no longer fails with FileNotFoundError or loses its data.

Written by the indexing model from the issue text.