code: def test_disconnects_during_large_async_get(call_ray_start_shared):
      """
      Disconnect repeatedly during a large (multi-chunk) async get.
      """
      i = 0
      started = False
  
      def fail_every_three(_):
          # Inject an error every third time this method is called
          nonlocal i, started
          if not started:
              return
          i += 1
          if i % 3 == 0:
              raise RuntimeError
  
      @ray.remote
      def large_result():
          # 1024x1024x6 float64 matrix (96 MiB). With 5MiB chunk size,
          # it will take at least 16 chunks to transfer this object. Since
          # the failure is injected every 3 chunks, this transfer can only
          # work if the chunked get request retries at the last received chunk
          # (instead of starting from the beginning each retry)
          return np.random.random((1024, 1024, 6))
  
      with start_middleman_server(on_data_response=fail_every_three):
          started = True
  
          async def get_large_result():
              return await large_result.remote()
  
          result = get_or_create_event_loop().run_until_complete(
              get_large_result()
          )
          assert result.shape == (1024, 1024, 6)
  
imports: from concurrent import futures
  import contextlib
  import os
  import threading
  import sys
  import grpc
  from mock import Mock
  import numpy
  import time
  import random
  import pytest
  from typing import Any, Callable, Optional
  from unittest.mock import patch
  import ray
  from ray._private.utils import get_or_create_event_loop
  import ray.core.generated.ray_client_pb2
  import ray.core.generated.ray_client_pb2_grpc
  from ray.tests.conftest import call_ray_start_context
  from ray.util.client.common import CLIENT_SERVER_MAX_THREADS, GRPC_OPTIONS
label_desc: Use image node_id 5 for calling a global function with example usage: test_disconnects_during_large_async_get(call_ray_start_shared) and returns: np, unknown