You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/10/18 22:41:20 UTC
[GitHub] [beam] lukecwik opened a new pull request #15747: Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
lukecwik opened a new pull request #15747:
URL: https://github.com/apache/beam/pull/15747
This is towards a migration to save 3-5% of CPU for small bundles by effectively merging the QueueingBeamFnDataClient at a higher level within the inbound observer removing some unnecessary timed polling/take calls. The existing implementation relied on the timeouts to prevent a stuckness issue since they were not able to propagate the cancellation cleanly to the producer or consumer.
JMH benchmarks based upon the complete change:
```
Result "org.apache.beam.fn.harness.ProcessBundleBenchmark.testTinyBundle":
55980.588 ±(99.9%) 1410.852 ops/s [Average]
(min, avg, max) = (51231.160, 55980.588, 59024.162), stdev = 1883.447
CI (99.9%): [54569.736, 57391.440] (assumes normal distribution)
```
Benchmark before the complete change:
```
Result "org.apache.beam.fn.harness.ProcessBundleBenchmark.testTinyBundle":
54228.952 ±(99.9%) 1415.909 ops/s [Average]
(min, avg, max) = (47791.452, 54228.952, 57266.584), stdev = 1890.198
CI (99.9%): [52813.043, 55644.862] (assumes normal distribution)
```
Please ignore the cancellable queue part of this change since it is being reviewed independently in #15736
Also note that I'm breaking this change up across multiple PRs since the end goal is to deprecate and replace `BeamFnDataGrpcMultiplexer` with `BeamFnDataGrpcMultiplexer2` and similarly for the inbound observer which is why the `2` in the name.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] 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://beam.apache.org/contribute/#make-reviewers-job-easier).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
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)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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
[GitHub] [beam] TheNeuralBit commented on a change in pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #15747:
URL: https://github.com/apache/beam/pull/15747#discussion_r733924809
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
Review comment:
A labeled break was unfamiliar to me, I had to look it up to be able to parse, maybe the below would help clarify?
```suggestion
SINGLE_INSTRUCTION_ID:
{
String instructionId = null;
for (BeamFnApi.Elements.Data data : value.getDataList()) {
if (instructionId == null) {
instructionId = data.getInstructionId();
} else if (!instructionId.equals(data.getInstructionId())) {
// Multiple instruction ids detected, break out of this block
break SINGLE_INSTRUCTION_ID;
}
}
for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
if (instructionId == null) {
instructionId = timers.getInstructionId();
} else if (!instructionId.equals(timers.getInstructionId())) {
// Multiple instruction ids detected, break out of this block
break SINGLE_INSTRUCTION_ID;
}
}
```
Alternatively, I think an approach that uses Streams could improve clarity:
```java
Stream<String> instructionIds =
Streams.concat(
value.getDataList().stream().map(BeamFnApi.Elements.Data::getInstructionId),
value.getTimersList().stream().map(BeamFnApi.Elements.Timers::getInstructionId));
if (instructionIds.distinct().limit(2).count() <= 1) {
// fast path
} else {
// multi instruction path
}
```
But I understand it may not be as performant, we'd need to benchmark it. Up to you what you'd like to do here.
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ if (instructionId == null) {
+ return;
+ }
+ forwardToConsumerForInstructionId(instructionId, value);
+ return;
+ }
+
+ // Handle the case if there are multiple instruction ids.
+ HashSet<String> instructionIds = new HashSet<>();
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ instructionIds.add(data.getInstructionId());
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ instructionIds.add(timers.getInstructionId());
+ }
+ for (String instructionId : instructionIds) {
Review comment:
Why not do this at the same time that you're collecting the instruction ids?
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/DataEndpoint.java
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.sdk.fn.data;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.coders.Coder;
+
+@AutoValue
+public abstract class DataEndpoint<T> {
Review comment:
Could this and TimerEndpoint be nested inside the `BeamFnDataInboundObserver2`?
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
Review comment:
nit: could you prefer either `BeamFnApi.Elements` or just `Elements` everywhere?
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ if (instructionId == null) {
+ return;
+ }
+ forwardToConsumerForInstructionId(instructionId, value);
+ return;
+ }
+
+ // Handle the case if there are multiple instruction ids.
+ HashSet<String> instructionIds = new HashSet<>();
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ instructionIds.add(data.getInstructionId());
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ instructionIds.add(timers.getInstructionId());
+ }
+ for (String instructionId : instructionIds) {
+ BeamFnApi.Elements.Builder builder = BeamFnApi.Elements.newBuilder();
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId.equals(data.getInstructionId())) {
+ builder.addData(data);
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId.equals(timers.getInstructionId())) {
+ builder.addTimers(timers);
+ }
+ }
+ forwardToConsumerForInstructionId(instructionId, builder.build());
+ }
+ }
+
+ private void forwardToConsumerForInstructionId(String instructionId, Elements value) {
+ CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> consumerFuture =
+ receiverFuture(instructionId);
+ if (erroredInstructionIds.containsKey(instructionId)) {
+ LOG.debug("Ignoring inbound data for failed instruction {}", instructionId);
+ return;
+ }
Review comment:
It looks like the current implementation doesn't have any logic like this, is it new?
--
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
[GitHub] [beam] lukecwik commented on pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15747:
URL: https://github.com/apache/beam/pull/15747#issuecomment-946246636
R: @TheNeuralBit
--
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
[GitHub] [beam] lukecwik commented on a change in pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15747:
URL: https://github.com/apache/beam/pull/15747#discussion_r734066644
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
Review comment:
It is not as performant.
I initially used a HashSet variant and that cost us 0.3% of CPU time which is a significant amount when the whole change will improve performance by 3-5%
This thread that reads from gRPC is quite sensitive to code changes since each message must pass through it.
--
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
[GitHub] [beam] lukecwik commented on pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15747:
URL: https://github.com/apache/beam/pull/15747#issuecomment-949060103
Run Java PreCommit
--
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
[GitHub] [beam] lukecwik commented on pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15747:
URL: https://github.com/apache/beam/pull/15747#issuecomment-947803505
> What is the migration plan?
1) Swap the usage within the BeamFnDataClient.
2) Wait for Dataflow JRH deprecation and code deletion
3) Swap the usage within the shared libraries for runners like Flink/Spark.
I have a prototype for 3) and got stuck when trying to update the JRH for Dataflow since it was going to be too much effort there which is why I opted for code deletion. JRH deletion is upcoming with Python streaming using Dataflow runner v2 all the time.
--
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
[GitHub] [beam] lukecwik commented on pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15747:
URL: https://github.com/apache/beam/pull/15747#issuecomment-949132795
Run Java PreCommit
--
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
[GitHub] [beam] lukecwik commented on a change in pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15747:
URL: https://github.com/apache/beam/pull/15747#discussion_r734066644
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
Review comment:
Thanks for the suggestion to help clarify since they are somewhat uncommon but necessary if using `break` in some nested for loop situations.
Using Stream is definitely not as performant and I initially used a HashSet variant that you see below with multiple instructions ids and that cost us 0.3% of CPU time which is a significant amount when the whole change will improve performance by 3-5%. The common case is a single instruction id so the optimization is worth it.
This thread that reads from gRPC is quite sensitive to code changes since each message must pass through it while bundle processing is less sensitive since we can use more CPU cores.
--
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
[GitHub] [beam] lukecwik commented on a change in pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15747:
URL: https://github.com/apache/beam/pull/15747#discussion_r734070557
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
Review comment:
Done.
I typically prefer importing `BeamFnApi.Elements` since `Elements` is too generic but many times Intellij doesn't do what I want and I forget about it.
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ if (instructionId == null) {
+ return;
+ }
+ forwardToConsumerForInstructionId(instructionId, value);
+ return;
+ }
+
+ // Handle the case if there are multiple instruction ids.
+ HashSet<String> instructionIds = new HashSet<>();
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ instructionIds.add(data.getInstructionId());
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ instructionIds.add(timers.getInstructionId());
+ }
+ for (String instructionId : instructionIds) {
Review comment:
Are you suggesting I use a map from `instructionId -> BeamFnApi.Elements.Builder` and then loop over the map's entries in a subsequent loop or something else?
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/DataEndpoint.java
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.sdk.fn.data;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.coders.Coder;
+
+@AutoValue
+public abstract class DataEndpoint<T> {
Review comment:
There are other usages within the shared runner code that does bundle processing allowing us to replace LogicalEndpoint.
##########
File path: sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer2.java
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.sdk.fn.data;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.fn.CancellableQueue;
+import org.apache.beam.sdk.fn.stream.OutboundObserverFactory;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.Status;
+import org.apache.beam.vendor.grpc.v1p36p0.io.grpc.stub.StreamObserver;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A gRPC multiplexer for a specific {@link Endpoints.ApiServiceDescriptor}.
+ *
+ * <p>Multiplexes data for inbound consumers based upon their {@code instructionId}.
+ *
+ * <p>Multiplexing inbound and outbound streams is as thread safe as the consumers of those streams.
+ * For inbound streams, this is as thread safe as the inbound observers. For outbound streams, this
+ * is as thread safe as the underlying stream observer.
+ *
+ * <p>TODO: Add support for multiplexing over multiple outbound observers by stickying the output
+ * location with a specific outbound observer.
+ */
+public class BeamFnDataGrpcMultiplexer2 implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(BeamFnDataGrpcMultiplexer2.class);
+ private final Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor;
+ private final StreamObserver<BeamFnApi.Elements> inboundObserver;
+ private final StreamObserver<BeamFnApi.Elements> outboundObserver;
+ private final ConcurrentMap<
+ /*instructionId=*/ String, CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>>>
+ receivers;
+ private final ConcurrentMap<String, Boolean> erroredInstructionIds;
+ private final List<CancellableQueue<Elements>> unusedQueues;
+
+ public BeamFnDataGrpcMultiplexer2(
+ Endpoints.@Nullable ApiServiceDescriptor apiServiceDescriptor,
+ OutboundObserverFactory outboundObserverFactory,
+ OutboundObserverFactory.BasicFactory<BeamFnApi.Elements, BeamFnApi.Elements>
+ baseOutboundObserverFactory) {
+ this.apiServiceDescriptor = apiServiceDescriptor;
+ this.receivers = new ConcurrentHashMap<>();
+ this.inboundObserver = new InboundObserver();
+ this.outboundObserver =
+ outboundObserverFactory.outboundObserverFor(baseOutboundObserverFactory, inboundObserver);
+ this.erroredInstructionIds = new ConcurrentHashMap<>();
+ this.unusedQueues = new ArrayList<>(100);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .omitNullValues()
+ .add("apiServiceDescriptor", apiServiceDescriptor)
+ .add("consumers", receivers)
+ .toString();
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getInboundObserver() {
+ return inboundObserver;
+ }
+
+ public StreamObserver<BeamFnApi.Elements> getOutboundObserver() {
+ return outboundObserver;
+ }
+
+ private CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture(
+ String instructionId) {
+ return receivers.computeIfAbsent(instructionId, (unused) -> new CompletableFuture<>());
+ }
+
+ /**
+ * Registers a consumer for the specified intruction id.
+ *
+ * <p>The {@link BeamFnDataGrpcMultiplexer2} partitions {@link BeamFnApi.Elements} with multiple
+ * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
+ * single instruction id.
+ *
+ * <p>The caller must {@link #unregisterConsumer unregister the consumer} when they no longer wish
+ * to receive messages.
+ */
+ public void registerConsumer(
+ String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
+ receiverFuture(instructionId).complete(receiver);
+ }
+
+ /** Unregisters a consumer. */
+ public void unregisterConsumer(String instructionId) {
+ receivers.remove(instructionId);
+ }
+
+ @VisibleForTesting
+ boolean hasConsumer(String instructionId) {
+ return receivers.containsKey(instructionId);
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
+ ImmutableList.copyOf(receivers.values())) {
+ // Cancel any observer waiting for the client to complete. If the receiver has already been
+ // completed or cancelled, this call will be ignored.
+ receiver.cancel(true);
+ if (!receiver.isCompletedExceptionally()) {
+ try {
+ receiver.get().close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ }
+ // Cancel any outbound calls and complete any inbound calls, as this multiplexer is hanging up
+ outboundObserver.onError(
+ Status.CANCELLED.withDescription("Multiplexer hanging up").asException());
+ inboundObserver.onCompleted();
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ /**
+ * A multiplexing {@link StreamObserver} that selects the inbound {@link Consumer} to pass the
+ * elements to.
+ *
+ * <p>The inbound observer blocks until the {@link Consumer} is bound allowing for the sending
+ * harness to initiate transmitting data without needing for the receiving harness to signal that
+ * it is ready to consume that data.
+ */
+ private final class InboundObserver implements StreamObserver<BeamFnApi.Elements> {
+ @Override
+ public void onNext(BeamFnApi.Elements value) {
+ // Have a fast path to handle the common case and provide a short circuit to exit if we detect
+ // multiple instruction ids.
+ SINGLE:
+ {
+ String instructionId = null;
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId == null) {
+ instructionId = data.getInstructionId();
+ } else if (!instructionId.equals(data.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId == null) {
+ instructionId = timers.getInstructionId();
+ } else if (!instructionId.equals(timers.getInstructionId())) {
+ break SINGLE;
+ }
+ }
+ if (instructionId == null) {
+ return;
+ }
+ forwardToConsumerForInstructionId(instructionId, value);
+ return;
+ }
+
+ // Handle the case if there are multiple instruction ids.
+ HashSet<String> instructionIds = new HashSet<>();
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ instructionIds.add(data.getInstructionId());
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ instructionIds.add(timers.getInstructionId());
+ }
+ for (String instructionId : instructionIds) {
+ BeamFnApi.Elements.Builder builder = BeamFnApi.Elements.newBuilder();
+ for (BeamFnApi.Elements.Data data : value.getDataList()) {
+ if (instructionId.equals(data.getInstructionId())) {
+ builder.addData(data);
+ }
+ }
+ for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {
+ if (instructionId.equals(timers.getInstructionId())) {
+ builder.addTimers(timers);
+ }
+ }
+ forwardToConsumerForInstructionId(instructionId, builder.build());
+ }
+ }
+
+ private void forwardToConsumerForInstructionId(String instructionId, Elements value) {
+ CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> consumerFuture =
+ receiverFuture(instructionId);
+ if (erroredInstructionIds.containsKey(instructionId)) {
+ LOG.debug("Ignoring inbound data for failed instruction {}", instructionId);
+ return;
+ }
Review comment:
The existing implementation was supposed to stick around forever if an instruction failed and it would drop the data since the future would have been completed here: https://github.com/apache/beam/blob/8c143ab8dc2289c12fe4d7fc0b3042b0b518fda7/sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java#L53
This removes the need for storing that extra state around and the consumer can be removed on failure.
--
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
[GitHub] [beam] lukecwik merged pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
lukecwik merged pull request #15747:
URL: https://github.com/apache/beam/pull/15747
--
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
[GitHub] [beam] TheNeuralBit commented on pull request #15747: [BEAM-13015] Create a multiplexer that sends Elements based upon instruction id allowing for an inbound observer responsible for the entire instruction id.
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #15747:
URL: https://github.com/apache/beam/pull/15747#issuecomment-946900318
What is the migration plan?
--
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