You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "m-trieu (via GitHub)" <gi...@apache.org> on 2024/10/22 22:24:28 UTC
[PR] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
m-trieu opened a new pull request, #32905:
URL: https://github.com/apache/beam/pull/32905
Factor out semaphore logic out of WeightBoundedQueue to allow for sharing the weigher
R: @scwhittle
------------------------
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829874920
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingApplianceWorkCommitter.java:
##########
@@ -55,7 +56,9 @@ private StreamingApplianceWorkCommitter(
this.commitWorkFn = commitWorkFn;
this.commitQueue =
WeightedBoundedQueue.create(
- MAX_COMMIT_QUEUE_BYTES, commit -> Math.min(MAX_COMMIT_QUEUE_BYTES, commit.getSize()));
+ WeightedSemaphore.create(
+ MAX_COMMIT_QUEUE_BYTES,
+ commit -> Math.min(MAX_COMMIT_QUEUE_BYTES, commit.getSize())));
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2458099028
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
Re: [PR] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829864394
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedBoundedQueue.java:
##########
@@ -18,49 +18,40 @@
package org.apache.beam.runners.dataflow.worker.streaming;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.checkerframework.checker.nullness.qual.Nullable;
-/** Bounded set of queues, with a maximum total weight. */
+/** Queue bounded by a {@link WeightedSemaphore}. */
public final class WeightedBoundedQueue<V> {
private final LinkedBlockingQueue<V> queue;
- private final int maxWeight;
- private final Semaphore limit;
- private final Function<V, Integer> weigher;
+ private final WeightedSemaphore<V> weightedSemaphore;
private WeightedBoundedQueue(
- LinkedBlockingQueue<V> linkedBlockingQueue,
- int maxWeight,
- Semaphore limit,
- Function<V, Integer> weigher) {
+ LinkedBlockingQueue<V> linkedBlockingQueue, WeightedSemaphore<V> weightedSemaphore) {
this.queue = linkedBlockingQueue;
- this.maxWeight = maxWeight;
- this.limit = limit;
- this.weigher = weigher;
+ this.weightedSemaphore = weightedSemaphore;
}
- public static <V> WeightedBoundedQueue<V> create(int maxWeight, Function<V, Integer> weigherFn) {
- return new WeightedBoundedQueue<>(
- new LinkedBlockingQueue<>(), maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ public static <V> WeightedBoundedQueue<V> create(WeightedSemaphore<V> weigher) {
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829048040
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedSemaphore.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming;
+
+import java.util.concurrent.Semaphore;
+import java.util.function.Function;
+
+public final class WeightedSemaphore<V> {
+ private final int maxWeight;
+ private final Semaphore limit;
+ private final Function<V, Integer> weigher;
+
+ private WeightedSemaphore(int maxWeight, Semaphore limit, Function<V, Integer> weigher) {
+ this.maxWeight = maxWeight;
+ this.limit = limit;
+ this.weigher = weigher;
+ }
+
+ public static <V> WeightedSemaphore<V> create(int maxWeight, Function<V, Integer> weigherFn) {
+ return new WeightedSemaphore<>(maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ }
+
+ public void acquireUninterruptibly(V value) {
+ limit.acquireUninterruptibly(weigher.apply(value));
Review Comment:
How about capping the calculated weight to not be more than the max so it will be the single item available to process?
Otherwise it seems that we coudl block forever. this is currently done in the specific weigher logic but seems safer to do here and in release.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +206,37 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread = new Thread(() -> queue2.put(MAX_WEIGHT));
+ putThread.start();
+ // Should only see the first value in the queue, since the queue is at capacity. putThread
+ // should be blocked. The weight should be the same however, since queue1 and queue2 are sharing
+ // the weigher.
+ Thread.sleep(100);
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(1, queue1.size());
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
Review Comment:
rm, doing this twice.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +206,37 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread = new Thread(() -> queue2.put(MAX_WEIGHT));
+ putThread.start();
+ // Should only see the first value in the queue, since the queue is at capacity. putThread
+ // should be blocked. The weight should be the same however, since queue1 and queue2 are sharing
+ // the weigher.
+ Thread.sleep(100);
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(1, queue1.size());
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(0, queue2.size());
+
+ // Poll queue1, pulling off the only value inside and freeing up the capacity in the weigher.
+ queue1.poll();
+
+ // Wait for the putThread which was previously blocked due to the weigher being at capacity.
+ putThread.join();
+
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(1, queue2.size());
Review Comment:
poll from queue2 and verify the weight goes to zero.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingApplianceWorkCommitter.java:
##########
@@ -55,7 +56,9 @@ private StreamingApplianceWorkCommitter(
this.commitWorkFn = commitWorkFn;
this.commitQueue =
WeightedBoundedQueue.create(
- MAX_COMMIT_QUEUE_BYTES, commit -> Math.min(MAX_COMMIT_QUEUE_BYTES, commit.getSize()));
+ WeightedSemaphore.create(
+ MAX_COMMIT_QUEUE_BYTES,
+ commit -> Math.min(MAX_COMMIT_QUEUE_BYTES, commit.getSize())));
Review Comment:
remove min.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedBoundedQueue.java:
##########
@@ -18,49 +18,40 @@
package org.apache.beam.runners.dataflow.worker.streaming;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.checkerframework.checker.nullness.qual.Nullable;
-/** Bounded set of queues, with a maximum total weight. */
+/** Queue bounded by a {@link WeightedSemaphore}. */
public final class WeightedBoundedQueue<V> {
private final LinkedBlockingQueue<V> queue;
- private final int maxWeight;
- private final Semaphore limit;
- private final Function<V, Integer> weigher;
+ private final WeightedSemaphore<V> weightedSemaphore;
private WeightedBoundedQueue(
- LinkedBlockingQueue<V> linkedBlockingQueue,
- int maxWeight,
- Semaphore limit,
- Function<V, Integer> weigher) {
+ LinkedBlockingQueue<V> linkedBlockingQueue, WeightedSemaphore<V> weightedSemaphore) {
this.queue = linkedBlockingQueue;
- this.maxWeight = maxWeight;
- this.limit = limit;
- this.weigher = weigher;
+ this.weightedSemaphore = weightedSemaphore;
}
- public static <V> WeightedBoundedQueue<V> create(int maxWeight, Function<V, Integer> weigherFn) {
- return new WeightedBoundedQueue<>(
- new LinkedBlockingQueue<>(), maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ public static <V> WeightedBoundedQueue<V> create(WeightedSemaphore<V> weigher) {
Review Comment:
nit: weightedSemaphore
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commits.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * 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.commits;
+
+import org.apache.beam.runners.dataflow.worker.streaming.WeightedSemaphore;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+
+/** Utility class for commits. */
+@Internal
+public final class Commits {
+
+ /** Max bytes of commits queued on the user worker. */
+ @VisibleForTesting static final int MAX_QUEUED_COMMITS_BYTES = 500 << 20; // 500MB
+
+ private Commits() {}
+
+ public static WeightedSemaphore<Commit> maxCommitByteSemaphore() {
+ return WeightedSemaphore.create(
+ MAX_QUEUED_COMMITS_BYTES, commit -> Math.min(MAX_QUEUED_COMMITS_BYTES, commit.getSize()));
Review Comment:
see other comment, remove the min here.
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1825321410
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -118,19 +118,17 @@ public final class StreamingDataflowWorker {
*/
public static final int MAX_SINK_BYTES = 10_000_000;
+ public static final String STREAMING_ENGINE_USE_JOB_SETTINGS_FOR_HEARTBEAT_POOL =
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1825323708
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +201,46 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread =
+ new Thread(
+ () -> {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ queue2.put(MAX_WEIGHT);
+ });
+ putThread.start();
+
+ // Should only see the first value in the queue, since the queue is at capacity. putThread
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2452912750
back to you @scwhittle thank you
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829872221
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedSemaphore.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming;
+
+import java.util.concurrent.Semaphore;
+import java.util.function.Function;
+
+public final class WeightedSemaphore<V> {
+ private final int maxWeight;
+ private final Semaphore limit;
+ private final Function<V, Integer> weigher;
+
+ private WeightedSemaphore(int maxWeight, Semaphore limit, Function<V, Integer> weigher) {
+ this.maxWeight = maxWeight;
+ this.limit = limit;
+ this.weigher = weigher;
+ }
+
+ public static <V> WeightedSemaphore<V> create(int maxWeight, Function<V, Integer> weigherFn) {
+ return new WeightedSemaphore<>(maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ }
+
+ public void acquireUninterruptibly(V value) {
+ limit.acquireUninterruptibly(weigher.apply(value));
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2459102296
failure appears unrelated
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2433849652
@scwhittle ready to for review, the failure to unrelated thanks!
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829886077
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +206,37 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread = new Thread(() -> queue2.put(MAX_WEIGHT));
+ putThread.start();
+ // Should only see the first value in the queue, since the queue is at capacity. putThread
+ // should be blocked. The weight should be the same however, since queue1 and queue2 are sharing
+ // the weigher.
+ Thread.sleep(100);
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(1, queue1.size());
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(0, queue2.size());
+
+ // Poll queue1, pulling off the only value inside and freeing up the capacity in the weigher.
+ queue1.poll();
+
+ // Wait for the putThread which was previously blocked due to the weigher being at capacity.
+ putThread.join();
+
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(1, queue2.size());
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2457977417
back to you thanks! @scwhittle
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1825321886
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedSemaphore.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming;
+
+import java.util.concurrent.Semaphore;
+import java.util.function.Function;
+
+public final class WeightedSemaphore<V> {
+ private final int maxWeight;
+ private final Semaphore limit;
+ private final Function<V, Integer> weigher;
+
+ private WeightedSemaphore(int maxWeight, Semaphore limit, Function<V, Integer> weigher) {
+ this.maxWeight = maxWeight;
+ this.limit = limit;
+ this.weigher = weigher;
+ }
+
+ public static <V> WeightedSemaphore<V> create(int maxWeight, Function<V, Integer> weigherFn) {
+ return new WeightedSemaphore<>(maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ }
+
+ void acquire(V value) {
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1826130150
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedBoundedQueue.java:
##########
@@ -18,49 +18,40 @@
package org.apache.beam.runners.dataflow.worker.streaming;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.checkerframework.checker.nullness.qual.Nullable;
-/** Bounded set of queues, with a maximum total weight. */
+/** Queue bounded by a {@link WeightedSemaphore}. */
public final class WeightedBoundedQueue<V> {
private final LinkedBlockingQueue<V> queue;
- private final int maxWeight;
- private final Semaphore limit;
- private final Function<V, Integer> weigher;
+ private final WeightedSemaphore<V> weigher;
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1826130936
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -86,16 +86,19 @@ public final class StreamingEngineWorkCommitter implements WorkCommitter {
public static Builder builder() {
return new AutoBuilder_StreamingEngineWorkCommitter_Builder()
.setBackendWorkerToken(NO_BACKEND_WORKER_TOKEN)
+ .setWeigher(
+ WeightedSemaphore.create(
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2436863685
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
Re: [PR] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2452921713
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @Abacn added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1825322508
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -61,11 +62,10 @@ public final class StreamingEngineWorkCommitter implements WorkCommitter {
Supplier<CloseableStream<CommitWorkStream>> commitWorkStreamFactory,
int numCommitSenders,
Consumer<CompleteCommit> onCommitComplete,
- String backendWorkerToken) {
+ String backendWorkerToken,
+ WeightedSemaphore<Commit> weigher) {
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829874400
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commits.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * 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.commits;
+
+import org.apache.beam.runners.dataflow.worker.streaming.WeightedSemaphore;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+
+/** Utility class for commits. */
+@Internal
+public final class Commits {
+
+ /** Max bytes of commits queued on the user worker. */
+ @VisibleForTesting static final int MAX_QUEUED_COMMITS_BYTES = 500 << 20; // 500MB
+
+ private Commits() {}
+
+ public static WeightedSemaphore<Commit> maxCommitByteSemaphore() {
+ return WeightedSemaphore.create(
+ MAX_QUEUED_COMMITS_BYTES, commit -> Math.min(MAX_QUEUED_COMMITS_BYTES, commit.getSize()));
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1825323343
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +201,46 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread =
+ new Thread(
+ () -> {
+ try {
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #32905:
URL: https://github.com/apache/beam/pull/32905
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32905:
URL: https://github.com/apache/beam/pull/32905#issuecomment-2430474916
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1822315996
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -118,19 +118,17 @@ public final class StreamingDataflowWorker {
*/
public static final int MAX_SINK_BYTES = 10_000_000;
+ public static final String STREAMING_ENGINE_USE_JOB_SETTINGS_FOR_HEARTBEAT_POOL =
Review Comment:
revert this file
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedSemaphore.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming;
+
+import java.util.concurrent.Semaphore;
+import java.util.function.Function;
+
+public final class WeightedSemaphore<V> {
+ private final int maxWeight;
+ private final Semaphore limit;
+ private final Function<V, Integer> weigher;
+
+ private WeightedSemaphore(int maxWeight, Semaphore limit, Function<V, Integer> weigher) {
+ this.maxWeight = maxWeight;
+ this.limit = limit;
+ this.weigher = weigher;
+ }
+
+ public static <V> WeightedSemaphore<V> create(int maxWeight, Function<V, Integer> weigherFn) {
+ return new WeightedSemaphore<>(maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ }
+
+ void acquire(V value) {
Review Comment:
public acquire and release? seems like basic functionality
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -86,16 +86,19 @@ public final class StreamingEngineWorkCommitter implements WorkCommitter {
public static Builder builder() {
return new AutoBuilder_StreamingEngineWorkCommitter_Builder()
.setBackendWorkerToken(NO_BACKEND_WORKER_TOKEN)
+ .setWeigher(
+ WeightedSemaphore.create(
Review Comment:
maybe require this to be set by caller instead of hiding it? It would be nice for the callers to know about it because then they could display it on their status page
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedBoundedQueue.java:
##########
@@ -78,24 +69,26 @@ public void put(V value) {
public @Nullable V poll(long timeout, TimeUnit unit) throws InterruptedException {
V result = queue.poll(timeout, unit);
if (result != null) {
- limit.release(weigher.apply(result));
+ weigher.release(result);
}
return result;
}
/** Returns and removes the next value, or blocks until one is available. */
public @Nullable V take() throws InterruptedException {
V result = queue.take();
- limit.release(weigher.apply(result));
+ weigher.release(result);
return result;
}
/** Returns the current weight of the queue. */
Review Comment:
can this be removed and just look at the weighted semaphore? It is confusing since it doesn't count just bytes in this queue.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java:
##########
@@ -61,11 +62,10 @@ public final class StreamingEngineWorkCommitter implements WorkCommitter {
Supplier<CloseableStream<CommitWorkStream>> commitWorkStreamFactory,
int numCommitSenders,
Consumer<CompleteCommit> onCommitComplete,
- String backendWorkerToken) {
+ String backendWorkerToken,
+ WeightedSemaphore<Commit> weigher) {
Review Comment:
rename weigher throughout to commitByteSemaphore or something capturing it is the limiter not just a weighing function?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedSemaphore.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming;
+
+import java.util.concurrent.Semaphore;
+import java.util.function.Function;
+
+public final class WeightedSemaphore<V> {
+ private final int maxWeight;
+ private final Semaphore limit;
+ private final Function<V, Integer> weigher;
+
+ private WeightedSemaphore(int maxWeight, Semaphore limit, Function<V, Integer> weigher) {
+ this.maxWeight = maxWeight;
+ this.limit = limit;
+ this.weigher = weigher;
+ }
+
+ public static <V> WeightedSemaphore<V> create(int maxWeight, Function<V, Integer> weigherFn) {
+ return new WeightedSemaphore<>(maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ }
+
+ void acquire(V value) {
Review Comment:
name acquireUninterruptibly?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +201,46 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread =
+ new Thread(
+ () -> {
+ try {
Review Comment:
remove the sleep? the element is already added by queue1 above so the sleep doesn't seem necessary
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedBoundedQueue.java:
##########
@@ -18,49 +18,40 @@
package org.apache.beam.runners.dataflow.worker.streaming;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.checkerframework.checker.nullness.qual.Nullable;
-/** Bounded set of queues, with a maximum total weight. */
+/** Queue bounded by a {@link WeightedSemaphore}. */
public final class WeightedBoundedQueue<V> {
private final LinkedBlockingQueue<V> queue;
- private final int maxWeight;
- private final Semaphore limit;
- private final Function<V, Integer> weigher;
+ private final WeightedSemaphore<V> weigher;
Review Comment:
name weightedSemaphore? weigher just sounds like it is the weigh function
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +201,46 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread =
+ new Thread(
+ () -> {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ queue2.put(MAX_WEIGHT);
+ });
+ putThread.start();
+
+ // Should only see the first value in the queue, since the queue is at capacity. putThread
Review Comment:
could put the sleep here if you want to give a little more time for putThread to run and become blocked (doesn't matter if it doesn't or not so not racy but better test if it does block).
--
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1826136243
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedBoundedQueue.java:
##########
@@ -78,24 +69,26 @@ public void put(V value) {
public @Nullable V poll(long timeout, TimeUnit unit) throws InterruptedException {
V result = queue.poll(timeout, unit);
if (result != null) {
- limit.release(weigher.apply(result));
+ weigher.release(result);
}
return result;
}
/** Returns and removes the next value, or blocks until one is available. */
public @Nullable V take() throws InterruptedException {
V result = queue.take();
- limit.release(weigher.apply(result));
+ weigher.release(result);
return result;
}
/** Returns the current weight of the queue. */
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1825321667
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/WeightedSemaphore.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.streaming;
+
+import java.util.concurrent.Semaphore;
+import java.util.function.Function;
+
+public final class WeightedSemaphore<V> {
+ private final int maxWeight;
+ private final Semaphore limit;
+ private final Function<V, Integer> weigher;
+
+ private WeightedSemaphore(int maxWeight, Semaphore limit, Function<V, Integer> weigher) {
+ this.maxWeight = maxWeight;
+ this.limit = limit;
+ this.weigher = weigher;
+ }
+
+ public static <V> WeightedSemaphore<V> create(int maxWeight, Function<V, Integer> weigherFn) {
+ return new WeightedSemaphore<>(maxWeight, new Semaphore(maxWeight, true), weigherFn);
+ }
+
+ void acquire(V value) {
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] extract semaphore logic out of WeightBoundedQueue to allow for sharing the weigher [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #32905:
URL: https://github.com/apache/beam/pull/32905#discussion_r1829876550
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/WeightBoundedQueueTest.java:
##########
@@ -194,4 +206,37 @@ public void testTake() throws InterruptedException {
assertEquals(MAX_WEIGHT, value.get());
}
+
+ @Test
+ public void testPut_sharedWeigher() throws InterruptedException {
+ WeightedSemaphore<Integer> weigher =
+ WeightedSemaphore.create(MAX_WEIGHT, i -> Math.min(MAX_WEIGHT, i));
+ WeightedBoundedQueue<Integer> queue1 = WeightedBoundedQueue.create(weigher);
+ WeightedBoundedQueue<Integer> queue2 = WeightedBoundedQueue.create(weigher);
+
+ // Insert value that takes all the weight into the queue1.
+ queue1.put(MAX_WEIGHT);
+
+ // Try to insert a value into the queue2. This will block since there is no capacity in the
+ // weigher.
+ Thread putThread = new Thread(() -> queue2.put(MAX_WEIGHT));
+ putThread.start();
+ // Should only see the first value in the queue, since the queue is at capacity. putThread
+ // should be blocked. The weight should be the same however, since queue1 and queue2 are sharing
+ // the weigher.
+ Thread.sleep(100);
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
+ assertEquals(1, queue1.size());
+ assertEquals(MAX_WEIGHT, weigher.currentWeight());
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