code: def test_valid_actor_state_2(call_ray_start_shared):
      """
      Do a full disconnect (cancel channel) every 11 requests. Failure
      happens:
        - before request sent: request never reaches server
        - before response received: response never reaches server
        - while get's are being processed
      """
  
      @ray.remote
      class IncrActor:
          def __init__(self):
              self.val = 0
  
          def incr(self):
              self.val += 1
              return self.val
  
      i = 0
  
      with start_middleman_server() as middleman:
  
          def fail_every_eleven(_):
              nonlocal i
              i += 1
              if i % 11 == 0:
                  middleman.reset_channel()
  
          middleman.data_servicer.on_response = fail_every_eleven
          middleman.task_servicer.on_request = fail_every_eleven
          middleman.task_servicer.on_response = fail_every_eleven
  
          actor = IncrActor.remote()
          for _ in range(100):
              ref = actor.incr.remote()
          assert ray.get(ref) == 100
  
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 9 for calling a global function with example usage: test_valid_actor_state_2(call_ray_start_shared) and returns: self