You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "scwhittle (via GitHub)" <gi...@apache.org> on 2024/02/06 13:51:17 UTC
[PR] Add options giving more isolation between grpc streaming rpcs. [beam]
scwhittle opened a new pull request, #30233:
URL: https://github.com/apache/beam/pull/30233
- introduce an IsolationChannel class that uses a separate channel per-rpc. As rpcs complete, the channel will be available for subsequent rpcs. This isn't a good idea in general but given we have a few long-lived rpcs in the streaming client.
- use a separate GetDataStream rpc for heartbeats from read requests. This ensures that heartbeats are not delayed by other read requests.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481364418
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/options/StreamingDataflowWorkerOptions.java:
##########
@@ -130,6 +130,24 @@ public interface StreamingDataflowWorkerOptions extends DataflowWorkerHarnessOpt
void setWindmillMessagesBetweenIsReadyChecks(int value);
+ @Description("If true, a most a single active rpc will be used per channel.")
+ @Default.Boolean(false)
+ boolean getUseIsolatedChannels();
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #30233:
URL: https://github.com/apache/beam/pull/30233
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481361438
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java:
##########
@@ -245,11 +284,11 @@ public void refreshActiveWork(Map<String, List<HeartbeatRequest>> heartbeats) {
if (useStreamingRequests) {
// With streaming requests, always send the request even when it is empty, to ensure that
// we trigger health checks for the stream even when it is idle.
Review Comment:
Removed and changed logic. There is a AbstractWindmillStream.maybeSendHealthCheck method which is used for keeping streams alive via a registry that pings them now, it doesn't use the heartbeats themselves for this purpose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30233:
URL: https://github.com/apache/beam/pull/30233#issuecomment-1929695127
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1479974199
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java:
##########
@@ -59,28 +60,66 @@ public class MetricTrackingWindmillServerStub {
private final MemoryMonitor gcThrashingMonitor;
private final boolean useStreamingRequests;
+ private final WindmillStreamPool<GetDataStream> getDataStreamPool;
+
+ // May be the same instance as getDataStreamPool based upon options.
Review Comment:
```suggestion
// Maybe the same instance as getDataStreamPool based upon options.
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/options/StreamingDataflowWorkerOptions.java:
##########
@@ -130,6 +130,24 @@ public interface StreamingDataflowWorkerOptions extends DataflowWorkerHarnessOpt
void setWindmillMessagesBetweenIsReadyChecks(int value);
+ @Description("If true, a most a single active rpc will be used per channel.")
+ @Default.Boolean(false)
+ boolean getUseIsolatedChannels();
Review Comment:
thoughts on naming these getUseIsolatedWindmillChannels, getUseSeparateWindmillHeartbeatStreams, getWindmillGetDataStreamCount to differentiate from other channels(eg: BQ, GCS)?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("channelCache")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("channelCache")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("channelCache")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private ManagedChannel getChannel() {
+ synchronized (channelCache) {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ }
+ ManagedChannel result = channelFactory.get();
+ synchronized (channelCache) {
Review Comment:
I think there is a race where `isTerminated` could return false after returning true.
1. Thread 1 is at `result = channelFactory.get()`
2. Thread 2 calls isTerminated gets True
3. Thread 1 makes progress and does `result.shutdown()` and exits getChannel
4. Thread2 calls isTerminated again and gets False
Is this okay?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java:
##########
@@ -59,28 +60,66 @@ public class MetricTrackingWindmillServerStub {
private final MemoryMonitor gcThrashingMonitor;
private final boolean useStreamingRequests;
+ private final WindmillStreamPool<GetDataStream> getDataStreamPool;
+
+ // May be the same instance as getDataStreamPool based upon options.
+ private final WindmillStreamPool<GetDataStream> heartbeatStreamPool;
+
@GuardedBy("this")
private final List<ReadBatch> pendingReadBatches;
@GuardedBy("this")
private int activeReadThreads = 0;
- private WindmillStreamPool<GetDataStream> streamPool;
+ @Internal
+ @AutoBuilder(ofClass = MetricTrackingWindmillServerStub.class)
+ public abstract static class Builder {
+
+ abstract Builder setServer(WindmillServerStub server);
+
+ abstract Builder setGcThrashingMonitor(MemoryMonitor gcThrashingMonitor);
+
+ abstract Builder setUseStreamingRequests(boolean useStreamingRequests);
- public MetricTrackingWindmillServerStub(
- WindmillServerStub server, MemoryMonitor gcThrashingMonitor, boolean useStreamingRequests) {
+ abstract Builder setUseSeparateHeartbeatStreams(boolean useSeparateHeartbeatStreams);
+
+ abstract Builder setNumGetDataStreams(int numGetDataStreams);
+
+ abstract MetricTrackingWindmillServerStub build();
+ }
+
+ public static Builder builder(WindmillServerStub server, MemoryMonitor gcThrashingMonitor) {
+ return new AutoBuilder_MetricTrackingWindmillServerStub_Builder()
+ .setServer(server)
+ .setGcThrashingMonitor(gcThrashingMonitor)
+ .setUseStreamingRequests(false)
+ .setUseSeparateHeartbeatStreams(false)
+ .setNumGetDataStreams(1);
+ }
+
+ MetricTrackingWindmillServerStub(
+ WindmillServerStub server,
+ MemoryMonitor gcThrashingMonitor,
+ boolean useStreamingRequests,
+ boolean useSeparateHeartbeatStreams,
+ int numGetDataStreams) {
this.server = server;
this.gcThrashingMonitor = gcThrashingMonitor;
- // This is used as a queue but is expected to be less than 10 batches.
- this.pendingReadBatches = new ArrayList<>();
this.useStreamingRequests = useStreamingRequests;
- }
-
- public void start() {
if (useStreamingRequests) {
- streamPool =
- WindmillStreamPool.create(NUM_STREAMS, STREAM_TIMEOUT, this.server::getDataStream);
+ getDataStreamPool =
+ WindmillStreamPool.create(numGetDataStreams, STREAM_TIMEOUT, this.server::getDataStream);
Review Comment:
Can we cap numGetDataStreams to a min of 1?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("channelCache")
Review Comment:
Any specific reason to guard by channelCache? Can it be `@GuardedBy("this")`
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java:
##########
@@ -245,11 +284,11 @@ public void refreshActiveWork(Map<String, List<HeartbeatRequest>> heartbeats) {
if (useStreamingRequests) {
// With streaming requests, always send the request even when it is empty, to ensure that
// we trigger health checks for the stream even when it is idle.
Review Comment:
> // With streaming requests, always send the request even when it is empty, to ensure that
> // we trigger health checks for the stream even when it is idle.
I was going to say we don't do this anymore for the streams in geetDataPool, but it looks like this never was true.
refreshActiveWork() internally doesn't send any requests when heartbeats is empty. should we remove the comment?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("channelCache")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("channelCache")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("channelCache")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private ManagedChannel getChannel() {
+ synchronized (channelCache) {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ }
+ ManagedChannel result = channelFactory.get();
+ synchronized (channelCache) {
+ usedChannels.add(result);
+ if (shutdownStarted) result.shutdown();
+ }
+ return result;
+ }
+
+ @Override
+ public ManagedChannel shutdown() {
+ synchronized (channelCache) {
+ shutdownStarted = true;
+ for (ManagedChannel channel : usedChannels) {
+ channel.shutdown();
+ }
+ for (ManagedChannel channel : channelCache) {
+ channel.shutdown();
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public boolean isShutdown() {
+ synchronized (channelCache) {
+ if (!shutdownStarted) return false;
+ for (ManagedChannel channel : usedChannels) {
+ if (!channel.isShutdown()) return false;
+ }
+ for (ManagedChannel channel : channelCache) {
+ if (!channel.isShutdown()) return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ synchronized (channelCache) {
+ if (!shutdownStarted) return false;
+ for (ManagedChannel channel : usedChannels) {
+ if (!channel.isTerminated()) return false;
+ }
+ for (ManagedChannel channel : channelCache) {
+ if (!channel.isTerminated()) return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public ManagedChannel shutdownNow() {
+ synchronized (channelCache) {
+ shutdownStarted = true;
+ for (ManagedChannel channel : usedChannels) {
+ channel.shutdownNow();
+ }
+ for (ManagedChannel channel : channelCache) {
+ channel.shutdownNow();
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+ long endTimeNanos = System.nanoTime() + unit.toNanos(timeout);
+ if (isTerminated()) return true;
+ synchronized (channelCache) {
+ if (!shutdownStarted) return false;
+ for (ManagedChannel channel : usedChannels) {
+ long awaitTimeNanos = endTimeNanos - System.nanoTime();
+ if (awaitTimeNanos <= 0) {
+ break;
+ }
+ channel.awaitTermination(awaitTimeNanos, TimeUnit.NANOSECONDS);
+ }
+ for (ManagedChannel channel : channelCache) {
+ long awaitTimeNanos = endTimeNanos - System.nanoTime();
+ if (awaitTimeNanos <= 0) {
+ break;
+ }
+ channel.awaitTermination(awaitTimeNanos, TimeUnit.NANOSECONDS);
+ }
+ }
+ return isTerminated();
+ }
+
+ void releaseChannelAfterCall(ManagedChannel managedChannel) {
+ synchronized (channelCache) {
+ Preconditions.checkState(
+ usedChannels.remove(managedChannel), "Channel released that was not used");
+ channelCache.add(managedChannel);
+ }
+ }
+
+ /** ClientCall wrapper that adds channel back to the cache on completion. */
+ private static class ReleasingClientCall<ReqT, RespT>
+ extends SimpleForwardingClientCall<ReqT, RespT> {
+ private final IsolationChannel isolationChannel;
+
+ private final ManagedChannel channel;
+ private final AtomicBoolean wasReleased = new AtomicBoolean();
+
+ public ReleasingClientCall(
+ IsolationChannel isolationChannel,
+ ManagedChannel channel,
+ ClientCall<ReqT, RespT> delegate) {
+ super(delegate);
+ this.isolationChannel = isolationChannel;
+ this.channel = channel;
+ }
+
+ @Override
+ public void start(Listener<RespT> responseListener, Metadata headers) {
+ try {
+ super.start(
+ new SimpleForwardingClientCallListener<RespT>(responseListener) {
+ @Override
+ public void onClose(Status status, Metadata trailers) {
+ try {
+ super.onClose(status, trailers);
+ } finally {
+ releaseChannelOnce();
+ }
+ }
+ },
+ headers);
+ } catch (Exception e) {
+ releaseChannelOnce();
+ throw e;
+ }
+ }
+
+ private void releaseChannelOnce() {
+ if (!wasReleased.getAndSet(true)) {
+ isolationChannel.releaseChannelAfterCall(channel);
+ } else {
+ LOG.log(
Review Comment:
```suggestion
LOG.warning(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481351257
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java:
##########
@@ -59,28 +60,66 @@ public class MetricTrackingWindmillServerStub {
private final MemoryMonitor gcThrashingMonitor;
private final boolean useStreamingRequests;
+ private final WindmillStreamPool<GetDataStream> getDataStreamPool;
+
+ // May be the same instance as getDataStreamPool based upon options.
Review Comment:
May be is correct but changed to "This may be" so it's clearer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481399858
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("channelCache")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("channelCache")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("channelCache")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private ManagedChannel getChannel() {
+ synchronized (channelCache) {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ }
+ ManagedChannel result = channelFactory.get();
+ synchronized (channelCache) {
+ usedChannels.add(result);
+ if (shutdownStarted) result.shutdown();
+ }
+ return result;
+ }
+
+ @Override
+ public ManagedChannel shutdown() {
+ synchronized (channelCache) {
+ shutdownStarted = true;
+ for (ManagedChannel channel : usedChannels) {
+ channel.shutdown();
+ }
+ for (ManagedChannel channel : channelCache) {
+ channel.shutdown();
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public boolean isShutdown() {
+ synchronized (channelCache) {
+ if (!shutdownStarted) return false;
+ for (ManagedChannel channel : usedChannels) {
+ if (!channel.isShutdown()) return false;
+ }
+ for (ManagedChannel channel : channelCache) {
+ if (!channel.isShutdown()) return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ synchronized (channelCache) {
+ if (!shutdownStarted) return false;
+ for (ManagedChannel channel : usedChannels) {
+ if (!channel.isTerminated()) return false;
+ }
+ for (ManagedChannel channel : channelCache) {
+ if (!channel.isTerminated()) return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public ManagedChannel shutdownNow() {
+ synchronized (channelCache) {
+ shutdownStarted = true;
+ for (ManagedChannel channel : usedChannels) {
+ channel.shutdownNow();
+ }
+ for (ManagedChannel channel : channelCache) {
+ channel.shutdownNow();
+ }
+ }
+ return this;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+ long endTimeNanos = System.nanoTime() + unit.toNanos(timeout);
+ if (isTerminated()) return true;
+ synchronized (channelCache) {
+ if (!shutdownStarted) return false;
+ for (ManagedChannel channel : usedChannels) {
+ long awaitTimeNanos = endTimeNanos - System.nanoTime();
+ if (awaitTimeNanos <= 0) {
+ break;
+ }
+ channel.awaitTermination(awaitTimeNanos, TimeUnit.NANOSECONDS);
+ }
+ for (ManagedChannel channel : channelCache) {
+ long awaitTimeNanos = endTimeNanos - System.nanoTime();
+ if (awaitTimeNanos <= 0) {
+ break;
+ }
+ channel.awaitTermination(awaitTimeNanos, TimeUnit.NANOSECONDS);
+ }
+ }
+ return isTerminated();
+ }
+
+ void releaseChannelAfterCall(ManagedChannel managedChannel) {
+ synchronized (channelCache) {
+ Preconditions.checkState(
+ usedChannels.remove(managedChannel), "Channel released that was not used");
+ channelCache.add(managedChannel);
+ }
+ }
+
+ /** ClientCall wrapper that adds channel back to the cache on completion. */
+ private static class ReleasingClientCall<ReqT, RespT>
+ extends SimpleForwardingClientCall<ReqT, RespT> {
+ private final IsolationChannel isolationChannel;
+
+ private final ManagedChannel channel;
+ private final AtomicBoolean wasReleased = new AtomicBoolean();
+
+ public ReleasingClientCall(
+ IsolationChannel isolationChannel,
+ ManagedChannel channel,
+ ClientCall<ReqT, RespT> delegate) {
+ super(delegate);
+ this.isolationChannel = isolationChannel;
+ this.channel = channel;
+ }
+
+ @Override
+ public void start(Listener<RespT> responseListener, Metadata headers) {
+ try {
+ super.start(
+ new SimpleForwardingClientCallListener<RespT>(responseListener) {
+ @Override
+ public void onClose(Status status, Metadata trailers) {
+ try {
+ super.onClose(status, trailers);
+ } finally {
+ releaseChannelOnce();
+ }
+ }
+ },
+ headers);
+ } catch (Exception e) {
+ releaseChannelOnce();
+ throw e;
+ }
+ }
+
+ private void releaseChannelOnce() {
+ if (!wasReleased.getAndSet(true)) {
+ isolationChannel.releaseChannelAfterCall(channel);
+ } else {
+ LOG.log(
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481402475
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("channelCache")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("channelCache")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("channelCache")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private ManagedChannel getChannel() {
+ synchronized (channelCache) {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ }
+ ManagedChannel result = channelFactory.get();
+ synchronized (channelCache) {
Review Comment:
Changed so that we synchronize over channel creation and avoid creating new channels if we're shutting down, in that case we can use an existing channel. One is guaranteed to exist from constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #30233:
URL: https://github.com/apache/beam/pull/30233#issuecomment-1934137870
@arunpandianp PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481399315
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("channelCache")
Review Comment:
I wasn't sure if anything in grpc would synchronize on the channel object. But if it is, perhaps we'd want it to be synchronized with the state too. Changed to this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #30233:
URL: https://github.com/apache/beam/pull/30233#issuecomment-1929689271
part of #30230
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1482980836
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("this")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("this")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("this")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private synchronized ManagedChannel getChannel() {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ if (shutdownStarted) {
+ // Avoid creating a new channel if we're shutting down and just use an existing one
+ // that has already started shutting down.
+ Preconditions.checkArgument(!usedChannels.isEmpty());
+ return usedChannels.iterator().next();
+ }
+
+ ManagedChannel result = channelFactory.get();
+ usedChannels.add(result);
+ return result;
+ }
+
+ @Override
+ public synchronized ManagedChannel shutdown() {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481604628
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
Review Comment:
```suggestion
private static final Logger LOG = Logger.getLogger(IsolationChannel.class);
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("this")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("this")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("this")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private synchronized ManagedChannel getChannel() {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ if (shutdownStarted) {
+ // Avoid creating a new channel if we're shutting down and just use an existing one
+ // that has already started shutting down.
+ Preconditions.checkArgument(!usedChannels.isEmpty());
+ return usedChannels.iterator().next();
+ }
+
+ ManagedChannel result = channelFactory.get();
+ usedChannels.add(result);
+ return result;
+ }
+
+ @Override
+ public synchronized ManagedChannel shutdown() {
Review Comment:
can we remove the synchronized from method signature and add a synchronized(this) block inside? That'll allow changing internal locking without changing the method signature. With synchronized in the signature, other classes might start relying on the this locking behavior.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/IsolationChannel.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Metadata;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.MethodDescriptor;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Status;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+/**
+ * A {@link ManagedChannel} that creates a dynamic # of cached channels to the same endpoint such
+ * that each active rpc has its own channel.
+ */
+@Internal
+class IsolationChannel extends ManagedChannel {
+ static final Logger LOG = Logger.getLogger(IsolationChannel.class.getName());
+
+ private final Supplier<ManagedChannel> channelFactory;
+
+ @GuardedBy("this")
+ private final List<ManagedChannel> channelCache;
+
+ @GuardedBy("this")
+ private final Set<ManagedChannel> usedChannels;
+
+ @GuardedBy("this")
+ private boolean shutdownStarted = false;
+
+ private final String authority;
+
+ // Expected that supplier returns channels to the same endpoint with the same authority.
+ public static IsolationChannel create(Supplier<ManagedChannel> channelFactory) {
+ return new IsolationChannel(channelFactory);
+ }
+
+ @VisibleForTesting
+ IsolationChannel(Supplier<ManagedChannel> channelFactory) {
+ this.channelFactory = channelFactory;
+ this.channelCache = new ArrayList<>();
+ ManagedChannel channel = channelFactory.get();
+ this.authority = channel.authority();
+ this.channelCache.add(channel);
+ this.usedChannels = new HashSet<>();
+ }
+
+ @Override
+ public String authority() {
+ return authority;
+ }
+
+ /** Pick an unused channel from the set or create one and then create a {@link ClientCall}. */
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
+ ManagedChannel channel = getChannel();
+ return new ReleasingClientCall<>(this, channel, channel.newCall(methodDescriptor, callOptions));
+ }
+
+ private synchronized ManagedChannel getChannel() {
+ if (!channelCache.isEmpty()) {
+ ManagedChannel result = channelCache.remove(channelCache.size() - 1);
+ usedChannels.add(result);
+ return result;
+ }
+ if (shutdownStarted) {
+ // Avoid creating a new channel if we're shutting down and just use an existing one
+ // that has already started shutting down.
+ Preconditions.checkArgument(!usedChannels.isEmpty());
+ return usedChannels.iterator().next();
+ }
+
+ ManagedChannel result = channelFactory.get();
+ usedChannels.add(result);
+ return result;
+ }
+
+ @Override
+ public synchronized ManagedChannel shutdown() {
+ shutdownStarted = true;
+ for (ManagedChannel channel : usedChannels) {
+ channel.shutdown();
+ }
+ for (ManagedChannel channel : channelCache) {
+ channel.shutdown();
+ }
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isShutdown() {
+ if (!shutdownStarted) {
+ return false;
+ }
+ for (ManagedChannel channel : usedChannels) {
+ if (!channel.isShutdown()) return false;
+ }
+ for (ManagedChannel channel : channelCache) {
+ if (!channel.isShutdown()) return false;
+ }
+ return true;
+ }
+
+ @Override
+ public synchronized boolean isTerminated() {
+ if (!shutdownStarted) {
+ return false;
+ }
+ for (ManagedChannel channel : usedChannels) {
+ if (!channel.isTerminated()) return false;
+ }
+ for (ManagedChannel channel : channelCache) {
+ if (!channel.isTerminated()) return false;
+ }
+ return true;
+ }
+
+ @Override
+ public synchronized ManagedChannel shutdownNow() {
+ ArrayList<ManagedChannel> channels = new ArrayList<>();
+ synchronized (this) {
Review Comment:
duplicate synchronized lock
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #30233:
URL: https://github.com/apache/beam/pull/30233#discussion_r1481353403
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java:
##########
@@ -59,28 +60,66 @@ public class MetricTrackingWindmillServerStub {
private final MemoryMonitor gcThrashingMonitor;
private final boolean useStreamingRequests;
+ private final WindmillStreamPool<GetDataStream> getDataStreamPool;
+
+ // May be the same instance as getDataStreamPool based upon options.
+ private final WindmillStreamPool<GetDataStream> heartbeatStreamPool;
+
@GuardedBy("this")
private final List<ReadBatch> pendingReadBatches;
@GuardedBy("this")
private int activeReadThreads = 0;
- private WindmillStreamPool<GetDataStream> streamPool;
+ @Internal
+ @AutoBuilder(ofClass = MetricTrackingWindmillServerStub.class)
+ public abstract static class Builder {
+
+ abstract Builder setServer(WindmillServerStub server);
+
+ abstract Builder setGcThrashingMonitor(MemoryMonitor gcThrashingMonitor);
+
+ abstract Builder setUseStreamingRequests(boolean useStreamingRequests);
- public MetricTrackingWindmillServerStub(
- WindmillServerStub server, MemoryMonitor gcThrashingMonitor, boolean useStreamingRequests) {
+ abstract Builder setUseSeparateHeartbeatStreams(boolean useSeparateHeartbeatStreams);
+
+ abstract Builder setNumGetDataStreams(int numGetDataStreams);
+
+ abstract MetricTrackingWindmillServerStub build();
+ }
+
+ public static Builder builder(WindmillServerStub server, MemoryMonitor gcThrashingMonitor) {
+ return new AutoBuilder_MetricTrackingWindmillServerStub_Builder()
+ .setServer(server)
+ .setGcThrashingMonitor(gcThrashingMonitor)
+ .setUseStreamingRequests(false)
+ .setUseSeparateHeartbeatStreams(false)
+ .setNumGetDataStreams(1);
+ }
+
+ MetricTrackingWindmillServerStub(
+ WindmillServerStub server,
+ MemoryMonitor gcThrashingMonitor,
+ boolean useStreamingRequests,
+ boolean useSeparateHeartbeatStreams,
+ int numGetDataStreams) {
this.server = server;
this.gcThrashingMonitor = gcThrashingMonitor;
- // This is used as a queue but is expected to be less than 10 batches.
- this.pendingReadBatches = new ArrayList<>();
this.useStreamingRequests = useStreamingRequests;
- }
-
- public void start() {
if (useStreamingRequests) {
- streamPool =
- WindmillStreamPool.create(NUM_STREAMS, STREAM_TIMEOUT, this.server::getDataStream);
+ getDataStreamPool =
+ WindmillStreamPool.create(numGetDataStreams, STREAM_TIMEOUT, this.server::getDataStream);
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #30233:
URL: https://github.com/apache/beam/pull/30233#issuecomment-1929682363
R: @arunpandianp
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add options giving more isolation between grpc streaming rpcs. [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #30233:
URL: https://github.com/apache/beam/pull/30233#issuecomment-1934272266
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org