beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbono...@apache.org
Subject [42/50] [abbrv] beam git commit: Increase the gRPC message size to max value
Date Thu, 20 Jul 2017 19:53:38 GMT
Increase the gRPC message size to max value


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/b424aa04
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/b424aa04
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/b424aa04

Branch: refs/heads/DSL_SQL
Commit: b424aa0409b507fe1c0c56a5f652d9be6458de66
Parents: 4d1db22
Author: Vikas Kedigehalli <vikasrk@google.com>
Authored: Tue Jul 18 10:06:46 2017 -0700
Committer: Luke Cwik <lcwik@google.com>
Committed: Wed Jul 19 13:17:37 2017 -0700

----------------------------------------------------------------------
 .../beam/fn/harness/channel/ManagedChannelFactory.java       | 6 ++++++
 sdks/python/apache_beam/runners/worker/data_plane.py         | 8 +++++++-
 2 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/b424aa04/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/channel/ManagedChannelFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/channel/ManagedChannelFactory.java
b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/channel/ManagedChannelFactory.java
index d26f4a5..3138bab 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/channel/ManagedChannelFactory.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/channel/ManagedChannelFactory.java
@@ -61,6 +61,9 @@ public abstract class ManagedChannelFactory {
               ? EpollDomainSocketChannel.class : EpollSocketChannel.class)
           .eventLoopGroup(new EpollEventLoopGroup())
           .usePlaintext(true)
+          // Set the message size to max value here. The actual size is governed by the
+          // buffer size in the layers above.
+          .maxInboundMessageSize(Integer.MAX_VALUE)
           .build();
     }
   }
@@ -74,6 +77,9 @@ public abstract class ManagedChannelFactory {
     public ManagedChannel forDescriptor(ApiServiceDescriptor apiServiceDescriptor) {
       return ManagedChannelBuilder.forTarget(apiServiceDescriptor.getUrl())
           .usePlaintext(true)
+          // Set the message size to max value here. The actual size is governed by the
+          // buffer size in the layers above.
+          .maxInboundMessageSize(Integer.MAX_VALUE)
           .build();
     }
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/b424aa04/sdks/python/apache_beam/runners/worker/data_plane.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/worker/data_plane.py b/sdks/python/apache_beam/runners/worker/data_plane.py
index 26f65ee..e713041 100644
--- a/sdks/python/apache_beam/runners/worker/data_plane.py
+++ b/sdks/python/apache_beam/runners/worker/data_plane.py
@@ -269,7 +269,13 @@ class GrpcClientDataChannelFactory(DataChannelFactory):
     url = remote_grpc_port.api_service_descriptor.url
     if url not in self._data_channel_cache:
       logging.info('Creating channel for %s', url)
-      grpc_channel = grpc.insecure_channel(url)
+      grpc_channel = grpc.insecure_channel(
+          url,
+          # Options to have no limits (-1) on the size of the messages
+          # received or sent over the data plane. The actual buffer size is
+          # controlled in a layer above.
+          options=[("grpc.max_receive_message_length", -1),
+                   ("grpc.max_send_message_length", -1)])
       self._data_channel_cache[url] = GrpcClientDataChannel(
           beam_fn_api_pb2.BeamFnDataStub(grpc_channel))
     return self._data_channel_cache[url]


Mime
View raw message