class_docstr: Manages a list of grpc connections.
  
  :param logger: the logger to use
  :param compression: The compression algorithm to be used by this GRPCConnectionPool when sending data to GRPC
code: def __init__(
      self,
      runtime_name,
      logger: Optional[JinaLogger] = None,
      compression: Optional[str] = None,
      metrics_registry: Optional["CollectorRegistry"] = None,
      meter: Optional["Meter"] = None,
      aio_tracing_client_interceptors: Optional[
          Sequence["ClientInterceptor"]
      ] = None,
      tracing_client_interceptor: Optional[
          "OpenTelemetryClientInterceptor"
      ] = None,
      channel_options: Optional[list] = None,
  ):
      self._logger = logger or JinaLogger(self.__class__.__name__)
      self.channel_options = channel_options
  
      self.compression = (
          getattr(grpc.Compression, compression)
          if compression
          else grpc.Compression.NoCompression
      )
  
      if metrics_registry:
          with ImportExtensions(
              required=True,
              help_text="You need to install the `prometheus_client` to use the montitoring functionality of jina",
          ):
              from prometheus_client import Summary
  
          sending_requests_time_metrics = Summary(
              "sending_request_seconds",
              "Time spent between sending a request to the Executor/Head and receiving the response",
              registry=metrics_registry,
              namespace="jina",
              labelnames=("runtime_name",),
          ).labels(runtime_name)
  
          received_response_bytes = Summary(
              "received_response_bytes",
              "Size in bytes of the response returned from the Head/Executor",
              registry=metrics_registry,
              namespace="jina",
              labelnames=("runtime_name",),
          ).labels(runtime_name)
  
          send_requests_bytes_metrics = Summary(
              "sent_request_bytes",
              "Size in bytes of the request sent to the Head/Executor",
              registry=metrics_registry,
              namespace="jina",
              labelnames=("runtime_name",),
          ).labels(runtime_name)
      else:
          sending_requests_time_metrics = None
          received_response_bytes = None
          send_requests_bytes_metrics = None
  
      self._metrics = _NetworkingMetrics(
          sending_requests_time_metrics,
          received_response_bytes,
          send_requests_bytes_metrics,
      )
  
      if meter:
          self._histograms = _NetworkingHistograms(
              sending_requests_time_metrics=meter.create_histogram(
                  name="jina_sending_request_seconds",
                  unit="s",
                  description="Time spent between sending a request to the Executor/Head and receiving the response",
              ),
              received_response_bytes=meter.create_histogram(
                  name="jina_received_response_bytes",
                  unit="By",
                  description="Size in bytes of the response returned from the Head/Executor",
              ),
              send_requests_bytes_metrics=meter.create_histogram(
                  name="jina_sent_request_bytes",
                  unit="By",
                  description="Size in bytes of the request sent to the Head/Executor",
              ),
              histogram_metric_labels={"runtime_name": runtime_name},
          )
      else:
          self._histograms = _NetworkingHistograms()
  
      self.aio_tracing_client_interceptors = (
          aio_tracing_client_interceptors
      )
      self.tracing_client_interceptor = tracing_client_interceptor
      self._connections = _ConnectionPoolMap(
          runtime_name=runtime_name,
          logger=self._logger,
          metrics=self._metrics,
          histograms=self._histograms,
          aio_tracing_client_interceptors=self.aio_tracing_client_interceptors,
          tracing_client_interceptor=self.tracing_client_interceptor,
          channel_options=self.channel_options,
      )
      self._deployment_address_map = {}
  
imports: import asyncio
  from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Set, Tuple, Union, AsyncGenerator
  import grpc
  from grpc.aio import AioRpcError
  from jina.constants import __default_endpoint__
  from jina.enums import PollingType
  from jina.excepts import InternalNetworkError
  from jina.importer import ImportExtensions
  from jina.logging.logger import JinaLogger
  from jina.proto import jina_pb2
  from jina.serve.helper import format_grpc_error
  from jina.serve.networking.connection_pool_map import _ConnectionPoolMap
  from jina.serve.networking.connection_stub import create_async_channel_stub
  from jina.serve.networking.instrumentation import _NetworkingHistograms, _NetworkingMetrics
  from jina.serve.networking.replica_list import _ReplicaList
  from jina.serve.networking.utils import DEFAULT_MINIMUM_RETRIES
  from jina.types.request import Request
  from jina.types.request.data import SingleDocumentRequest
label_desc: Use image node_id 1 to create a new GrpcConnectionPool object  with example: obj = GrpcConnectionPool(runtime_name, logger, compression, metrics_registry, meter, aio_tracing_client_interceptors, tracing_client_interceptor, channel_options)