Split gRPC Server and client settings and set connection backoff to 100 ms

.gitignore

@@ -61,4 +61,3 @@ sqlite*
 
 # Miscellaneous
 /executor-py/~
-

indexify/src/indexify/executor/function_executor/health_checker.py

@@ -4,7 +4,6 @@
 
 from grpc.aio import AioRpcError
 
-from indexify.function_executor.proto.configuration import HEALTH_CHECK_TIMEOUT_SEC
 from indexify.function_executor.proto.function_executor_pb2 import (
     HealthCheckRequest,
     HealthCheckResponse,
@@ -13,6 +12,8 @@
     FunctionExecutorStub,
 )
 
+from .server.client_configuration import HEALTH_CHECK_TIMEOUT_SEC
+
 HEALTH_CHECK_POLL_PERIOD_SEC = 10
 
 

indexify/src/indexify/executor/function_executor/server/client_configuration.py

@@ -0,0 +1,20 @@
+# We send function inputs and outputs over gRPC.
+# -1 means unlimited. We don't want to limit the size of data customers are using.
+# The effective max message size in this case is about 1.9 GB, see the max payload test.
+# This is due to internal hard gRPC limits. When we want to increase the message sizes
+# we'll have to implement chunking for large messages.
+_MAX_GRPC_MESSAGE_LENGTH = -1
+
+# Optimize the channels for low latency connection establishement as we are running on the same host.
+_RECONNECT_BACKOFF_MS = 100
+
+GRPC_CHANNEL_OPTIONS = [
+    ("grpc.max_receive_message_length", _MAX_GRPC_MESSAGE_LENGTH),
+    ("grpc.max_send_message_length", _MAX_GRPC_MESSAGE_LENGTH),
+    ("grpc.min_reconnect_backoff_ms", _RECONNECT_BACKOFF_MS),
+    ("grpc.max_reconnect_backoff_ms", _RECONNECT_BACKOFF_MS),
+    ("grpc.initial_reconnect_backoff_ms", _RECONNECT_BACKOFF_MS),
+]
+
+# If a health check takes more than this duration then the server is considered unhealthy.
+HEALTH_CHECK_TIMEOUT_SEC = 5

indexify/src/indexify/executor/function_executor/server/subprocess_function_executor_server.py

@@ -3,8 +3,7 @@
 
 import grpc
 
-from indexify.function_executor.proto.configuration import GRPC_CHANNEL_OPTIONS
-
+from .client_configuration import GRPC_CHANNEL_OPTIONS
 from .function_executor_server import FunctionExecutorServer
 
 

indexify/src/indexify/function_executor/proto/configuration.py → indexify/src/indexify/function_executor/proto/server_configuration.py

@@ -17,8 +17,3 @@
     ("grpc.max_send_message_length", _MAX_GRPC_MESSAGE_LENGTH),
     ("grpc.so_reuseport", _REUSE_SERVER_PORT),
 ]
-
-GRPC_CHANNEL_OPTIONS = GRPC_SERVER_OPTIONS
-
-# If a health check takes more than this duration then the server is considered unhealthy.
-HEALTH_CHECK_TIMEOUT_SEC = 5

indexify/src/indexify/function_executor/server.py

@@ -2,8 +2,8 @@
 
 import grpc
 
-from .proto.configuration import GRPC_SERVER_OPTIONS
 from .proto.function_executor_pb2_grpc import add_FunctionExecutorServicer_to_server
+from .proto.server_configuration import GRPC_SERVER_OPTIONS
 from .service import Service
 
 # Temporary limit until we have a better way to control this.

indexify/tests/function_executor/test_health_check.py

@@ -15,7 +15,6 @@
     run_task,
 )
 
-from indexify.function_executor.proto.configuration import HEALTH_CHECK_TIMEOUT_SEC
 from indexify.function_executor.proto.function_executor_pb2 import (
     HealthCheckRequest,
     HealthCheckResponse,
@@ -30,6 +29,7 @@
 
 # Lower - faster tests but more CPU usage.
 HEALTH_CHECK_POLL_PERIOD_SEC = 0.1
+HEALTH_CHECK_TIMEOUT_SEC = 5
 
 
 @tensorlake_function()

indexify/tests/function_executor/testing.py

@@ -5,7 +5,6 @@
 import grpc
 from tensorlake.functions_sdk.object_serializer import CloudPickleSerializer
 
-from indexify.function_executor.proto.configuration import GRPC_CHANNEL_OPTIONS
 from indexify.function_executor.proto.function_executor_pb2 import (
     FunctionOutput,
     RunTaskRequest,
@@ -15,6 +14,7 @@
 from indexify.function_executor.proto.function_executor_pb2_grpc import (
     FunctionExecutorStub,
 )
+from indexify.function_executor.proto.server_configuration import GRPC_SERVER_OPTIONS
 
 # Default Executor range is 50000:51000.
 # Use a value outside of this range to not conflict with other tests.
@@ -43,9 +43,10 @@ def __exit__(self, exc_type, exc_value, traceback):
 
 
 def rpc_channel(context_manager: FunctionExecutorProcessContextManager) -> grpc.Channel:
+    # The GRPC_SERVER_OPTIONS include the maximum message size which we need to set in the client channel.
     channel: grpc.Channel = grpc.insecure_channel(
         f"localhost:{context_manager.port}",
-        options=GRPC_CHANNEL_OPTIONS,
+        options=GRPC_SERVER_OPTIONS,
     )
     try:
         SERVER_STARTUP_TIMEOUT_SEC = 5