async def METHOD_NAME(actor_pools):
    worker_pool = actor_pools
    config = {
        "services": ["storage"],
        "storage": {
            "backends": ["shared_memory"],
        },
    }
    await start_services(NodeRole.WORKER, config, address=worker_pool.external_address)
    api = await StorageAPI.create("mock_session", worker_pool.external_address)
    value1 = np.random.rand(10, 10)
    await api.put("data1", value1)
    get_value1 = await api.get("data1")
    np.testing.assert_array_equal(value1, get_value1)
    # test api in subpool
    subpool_address = list(worker_pool._sub_processes.keys())[0]
    api2 = await StorageAPI.create("mock_session", subpool_address)
    assert api2._storage_handler_ref.address == subpool_address
    get_value1 = await api2.get("data1")
    np.testing.assert_array_equal(value1, get_value1)
    sliced_value = await api2.get("data1", conditions=[slice(None, None), slice(0, 4)])
    np.testing.assert_array_equal(value1[:, :4], sliced_value)
    await api.unpin("data1")
    value2 = pd.DataFrame(value1)
    await api2.put("data2", value2)
    get_value2 = await api.get("data2")
    pd.testing.assert_frame_equal(value2, get_value2)
    # test writer and read
    buffers = await AioSerializer(value2).run()
    size = sum(getattr(buf, "nbytes", len(buf)) for buf in buffers)
    # test open_reader and open_writer
    writer = await api.open_writer("write_key", size, StorageLevel.MEMORY)
    async with writer:
        for buf in buffers:
            await writer.write(buf)
    reader = await api.open_reader("write_key")
    async with reader:
        read_value = await AioDeserializer(reader).run()
    pd.testing.assert_frame_equal(value2, read_value)
    await stop_services(
        NodeRole.WORKER, address=worker_pool.external_address, config=config
    )