code: def test_disconnect_during_large_put(call_ray_start_shared):
      """
      Disconnect during a large (multi-chunk) put.
      """
      i = 0
      started = False
  
      def fail_halfway(_):
          # Inject an error halfway through the object transfer
          nonlocal i, started
          if not started:
              return
          i += 1
          if i == 8:
              raise RuntimeError
  
      with start_middleman_server(on_data_request=fail_halfway):
          started = True
          objref = ray.put(np.random.random((1024, 1024, 6)))
          assert i > 8  # Check that the failure was injected
          result = ray.get(objref)
          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 6 for calling a global function with example usage: test_disconnect_during_large_put(call_ray_start_shared) without return types