You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "JayajP (via GitHub)" <gi...@apache.org> on 2024/03/27 17:58:33 UTC
[PR] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
JayajP opened a new pull request, #30769:
URL: https://github.com/apache/beam/pull/30769
Implement `LockFreeHistogram` and use it in Dataflow's PerWorkerMetrics.
Currently, we record histogram metrics in `HistogramData`; however, this class locks the histogram on each metric update/extraction. I assume this is done for historic reasons since `HistogramData` also supports computing arbitrary percentiles on the recorded values and this computation would be inconsistent without locking on reads/writes. For PerWorkerMetrics/BigQueryMetrics we do not need percentiles so we can record metrics in a lock free manner.
LockFreeHistogram stores the following attributes on the values recorded:
- Count of number of values recorded in each bucket for recorded values that are in the bucket boundaries.
- Stored in a [AtomicLongArray](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicLongArray.html)
- Count/Mean of values the values recorded that are below the bucket boundaries (UnderflowStatistics)
- This is stored in an AtomicReference and updated with `AtomicReference::compareAndSet` similar to how existing [Distribution metrics](https://github.com/apache/beam/blob/45444209c776460ba67ac14a1c33e0775485266a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/DistributionData.java#L33) are recorded.
- Count/Mean of values the values recorded that are over the bucket boundaries (OverflowStatistic)
- stored similarly to UnderflowStatistic.
LockFreeHistogram supports extracting a delta update with the `getSnapshotAndReset` method. The delta update represents values that have been recorded in this histogram since the last time this method was called.
------------------------
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #30769:
URL: https://github.com/apache/beam/pull/30769#issuecomment-2040528515
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/30769?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `94.80519%` with `4 lines` in your changes are missing coverage. Please review.
> Project coverage is 68.45%. Comparing base [(`e894d8c`)](https://app.codecov.io/gh/apache/beam/commit/e894d8ca7fe10ed4771ce9d999b48361c25220ec?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`33d1ca1`)](https://app.codecov.io/gh/apache/beam/pull/30769?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 97 commits behind head on master.
| [Files](https://app.codecov.io/gh/apache/beam/pull/30769?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...eam/runners/dataflow/worker/LockFreeHistogram.java](https://app.codecov.io/gh/apache/beam/pull/30769?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FLockFreeHistogram.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvTG9ja0ZyZWVIaXN0b2dyYW0uamF2YQ==) | 93.33% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/30769?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #30769 +/- ##
=============================================
- Coverage 71.47% 68.45% -3.02%
- Complexity 0 14789 +14789
=============================================
Files 710 2611 +1901
Lines 104815 220252 +115437
Branches 0 11740 +11740
=============================================
+ Hits 74915 150772 +75857
- Misses 28268 63329 +35061
- Partials 1632 6151 +4519
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/30769/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [java](https://app.codecov.io/gh/apache/beam/pull/30769/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `65.71% <94.80%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/30769?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30769:
URL: https://github.com/apache/beam/pull/30769#issuecomment-2023779936
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30769:
URL: https://github.com/apache/beam/pull/30769#issuecomment-2023631280
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @jrmccluskey 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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "JayajP (via GitHub)" <gi...@apache.org>.
JayajP commented on PR #30769:
URL: https://github.com/apache/beam/pull/30769#issuecomment-2023777937
R: @m-trieu
--
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1544785415
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogramTest.java:
##########
@@ -0,0 +1,182 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link LockFreeHistogram}. */
+@RunWith(JUnit4.class)
+public class LockFreeHistogramTest {
+
+ @Test
+ public void testUpdate_OverflowValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(35, 40, 45);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedOverflow =
+ LockFreeHistogram.OutlierStatistic.create(120.0, 3L);
+ LockFreeHistogram.OutlierStatistic expectedUnderflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(0L, 0L, 0L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ assertThat(snapshot.get().underflowStatistic().mean(), equalTo(0.0));
+ assertThat(snapshot.get().overflowStatistic(), equalTo(expectedOverflow));
+ }
+
+ @Test
+ public void testUpdate_UnderflowValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(100, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(35, 40, 45);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedUnderflow =
+ LockFreeHistogram.OutlierStatistic.create(120.0, 3L);
+ LockFreeHistogram.OutlierStatistic expectedOverflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(0L, 0L, 0L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ assertThat(snapshot.get().underflowStatistic(), equalTo(expectedUnderflow));
+ }
+
+ @Test
+ public void testUpdate_InBoundsValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(5, 15, 25);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedOverflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ LockFreeHistogram.OutlierStatistic expectedUnderflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(1L, 1L, 1L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ }
+
+ @Test
+ public void testUpdate_EmptySnapshot() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(5, 15, 25);
+ Optional<LockFreeHistogram.Snapshot> snapshot_1 = histogram.getSnapshotAndReset();
+
+ assertThat(snapshot_1.isPresent(), equalTo(true));
+
+ Optional<LockFreeHistogram.Snapshot> snapshot_2 = histogram.getSnapshotAndReset();
+ assertThat(snapshot_2.isPresent(), equalTo(false));
+ }
+
+ /** A runnable records 200 values and then calls getSnapshotAndReset. */
+ private static class UpdateHistogramRunnable implements Runnable {
+ private final LockFreeHistogram histogram;
+ private final int val;
+ private Optional<LockFreeHistogram.Snapshot> snapshot;
+
+ private static final long valuesRecorded = 200L;
+
+ public UpdateHistogramRunnable(LockFreeHistogram histogram, int val) {
+ this.histogram = histogram;
+ this.val = val;
+ this.snapshot = Optional.empty();
+ }
+
+ @Override
+ public void run() {
+ for (long j = 0; j < valuesRecorded; j++) {
+ histogram.update(val);
+ }
+ snapshot = histogram.getSnapshotAndReset();
+ }
+
+ public long totalCountInSnapshot() {
+ if (snapshot.isPresent()) {
+ return snapshot.get().totalCount();
+ }
+ return 0;
+ }
+
+ public static long numValuesRecorded() {
+ return valuesRecorded;
+ }
+ }
+
+ @Test
+ public void testUpdateAndSnapshots_MultipleThreads() {
+ int numRunnables = 200;
+ ExecutorService executor = Executors.newFixedThreadPool(numRunnables);
+ HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 10);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+
+ UpdateHistogramRunnable[] runnables = new UpdateHistogramRunnable[numRunnables];
+ for (int i = 0; i < numRunnables; i++) {
+ runnables[i] = new UpdateHistogramRunnable(histogram, i);
+ executor.execute(runnables[i]);
+ }
+
+ try {
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
Review Comment:
right now it is executing the tasks 1 by 1 not (possibly) concurrently since execute is being called in the loop
Consider using https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CountDownLatch.html
you can do something like the following to make the test more deterministic (not bound bu
`CountDownLatch testLatch = new CountDownLatch(howManyTimesYouWantTheRunnableToRun)`
pass `testLatch` into `UpdateHistogramRunnable` constructor
at the end of `UpdateHistogramRunnable#run()` method call `testsLatch.countDown()`
then call `testLatch.await()` (wait until this has ran `howManyTimesYouWantTheRunnableToRun`)
and then use ExecutorService#submit to submit each runnable, this does not block until the runnable is done
another way is to not call execute in the `loop`, build the list of `UpdateHistogramRunnable` then use `ExecutorService.invokeAll(UpdateHistogramRunnable)`. I think right now it is executing 1 by 1.
based on lang specification invokeAll blocks but i haven't used 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
Re: [PR] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "JayajP (via GitHub)" <gi...@apache.org>.
JayajP commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1543498034
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogramTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+import java.util.Optional;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link LockFreeHistogram}. */
+@RunWith(JUnit4.class)
+public class LockFreeHistogramTest {
Review Comment:
Added `testUpdateAndSnapshots_MultipleThreads` which creates 200 threads. Each thread writes 200 updates and extracts a snapshot, the total values across all snapshots is 40000.
--
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #30769:
URL: https://github.com/apache/beam/pull/30769
--
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "JayajP (via GitHub)" <gi...@apache.org>.
JayajP commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1543499731
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
+
+ /**
+ * Represents the sum and mean of a collection of numbers. Used to represent the
+ * underflow/overflow statistics of a histogram.
+ */
+ @AutoValue
+ public abstract static class OutlierStatistic implements Serializable {
+ abstract double sum();
+
+ public abstract long count();
+
+ public static final OutlierStatistic EMPTY = create(0, 0);
+
+ public static OutlierStatistic create(double sum, long count) {
+ return new AutoValue_LockFreeHistogram_OutlierStatistic(sum, count);
+ }
+
+ public OutlierStatistic combine(double value) {
+ return create(sum() + value, count() + 1);
+ }
+
+ public double mean() {
+ if (count() == 0) {
+ return 0;
+ }
+ return sum() / count();
+ }
+ }
+
+ /**
+ * The snapshot of a histogram. The snapshot contains the overflow/underflow statistic, number of
+ * values recorded in each bucket, and the BucketType of the underlying histogram.
+ */
+ @AutoValue
+ public abstract static class Snapshot {
+ public abstract OutlierStatistic underflowStatistic();
+
+ public abstract OutlierStatistic overflowStatistic();
+
+ public abstract ImmutableLongArray buckets();
+
+ public abstract HistogramData.BucketType bucketType();
+
+ public static Snapshot create(
+ OutlierStatistic underflowStatistic,
+ OutlierStatistic overflowStatistic,
+ ImmutableLongArray buckets,
+ HistogramData.BucketType bucketType) {
+ return new AutoValue_LockFreeHistogram_Snapshot(
+ underflowStatistic, overflowStatistic, buckets, bucketType);
+ }
+
+ @Memoized
+ public long totalCount() {
+ long count = 0;
+ count += underflowStatistic().count();
+ count += overflowStatistic().count();
+ count += buckets().stream().sum();
+
+ return count;
+ }
+ }
+
+ /**
+ * Extract a delta update of this histogram. Update represents values that have been recorded in
+ * this histogram since the last time this method was called.
+ *
+ * <p>If this histogram is being updated concurrent to this method, then the returned snapshot is
+ * not guarenteed to contain those updates. However, those updates are not dropped and will be
+ * represented in a future call to this method.
+ *
+ * <p>If this histogram has not been updated since the last call to this method, an empty optional
+ * is returned.
+ */
+ public Optional<Snapshot> getSnapshotAndReset() {
+ if (!dirty.getAndSet(false)) {
+ return Optional.empty();
+ }
+
+ ImmutableLongArray.Builder bucketsSnapshotBuilder =
+ ImmutableLongArray.builder(buckets.length());
+ for (int i = 0; i < buckets.length(); i++) {
+ bucketsSnapshotBuilder.add(buckets.getAndSet(i, 0));
+ }
+ OutlierStatistic overflowSnapshot = overflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+ OutlierStatistic underflowSnapshot = underflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+
+ return Optional.of(
+ Snapshot.create(
+ underflowSnapshot, overflowSnapshot, bucketsSnapshotBuilder.build(), bucketType));
+ }
+
+ /** Create a histogram. */
+ public LockFreeHistogram(KV<MetricName, HistogramData.BucketType> kv) {
+ this.name = kv.getKey();
+ this.bucketType = kv.getValue();
+ this.buckets = new AtomicLongArray(bucketType.getNumBuckets());
+ }
+
+ @Override
+ public MetricName getName() {
+ return name;
+ }
+
+ private void updateInternal(double value) {
+ double rangeTo = bucketType.getRangeTo();
+ double rangeFrom = bucketType.getRangeFrom();
+ if (value >= rangeTo) {
+ recordTopRecordsValue(value);
+ } else if (value < rangeFrom) {
+ recordBottomRecordsValue(value);
+ } else {
+ recordInBoundsValue(value);
+ }
+ }
+
+ @Override
+ public void update(double value) {
+ updateInternal(value);
+ dirty.set(true);
Review Comment:
It won’t throw any ConcurrentModificationsErrors but, you’re right we don’t atomically flip this variable and update the metric.
I’m using this as an optimization for the thread that extracts metrics if there’s a LockFreeHistogram that’s not being written into. However, this variable is not necessary from a correctness standpoint (see details below), so I can remove it if you’d like.
This variable is updated when recording metrics/extracting updates. It does not affect the record path. On the extract path, we can either get a false positive (`dirty` is `true` when there’s no updates in the histogram) or a false negative (`dirty` is `false` when there are updates to the underlying histogram).
- In the false positive scenario, after we extract a snapshot we check that it’s non-empty before sending it to DFE so empty snapshots will just get dropped.
- In the false negative scenario, since we always set this value to true after recording a value in the histogram a future call to `getSnapshotAndReset` will see `dirty` set to `true` and extract the update.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
+
+ /**
+ * Represents the sum and mean of a collection of numbers. Used to represent the
+ * underflow/overflow statistics of a histogram.
+ */
+ @AutoValue
+ public abstract static class OutlierStatistic implements Serializable {
+ abstract double sum();
+
+ public abstract long count();
+
+ public static final OutlierStatistic EMPTY = create(0, 0);
+
+ public static OutlierStatistic create(double sum, long count) {
+ return new AutoValue_LockFreeHistogram_OutlierStatistic(sum, count);
+ }
+
+ public OutlierStatistic combine(double value) {
+ return create(sum() + value, count() + 1);
+ }
+
+ public double mean() {
+ if (count() == 0) {
+ return 0;
+ }
+ return sum() / count();
+ }
+ }
+
+ /**
+ * The snapshot of a histogram. The snapshot contains the overflow/underflow statistic, number of
+ * values recorded in each bucket, and the BucketType of the underlying histogram.
+ */
+ @AutoValue
+ public abstract static class Snapshot {
+ public abstract OutlierStatistic underflowStatistic();
+
+ public abstract OutlierStatistic overflowStatistic();
+
+ public abstract ImmutableLongArray buckets();
+
+ public abstract HistogramData.BucketType bucketType();
+
+ public static Snapshot create(
+ OutlierStatistic underflowStatistic,
+ OutlierStatistic overflowStatistic,
+ ImmutableLongArray buckets,
+ HistogramData.BucketType bucketType) {
+ return new AutoValue_LockFreeHistogram_Snapshot(
+ underflowStatistic, overflowStatistic, buckets, bucketType);
+ }
+
+ @Memoized
+ public long totalCount() {
+ long count = 0;
+ count += underflowStatistic().count();
+ count += overflowStatistic().count();
+ count += buckets().stream().sum();
+
+ return count;
+ }
+ }
+
+ /**
+ * Extract a delta update of this histogram. Update represents values that have been recorded in
+ * this histogram since the last time this method was called.
+ *
+ * <p>If this histogram is being updated concurrent to this method, then the returned snapshot is
+ * not guarenteed to contain those updates. However, those updates are not dropped and will be
+ * represented in a future call to this method.
+ *
+ * <p>If this histogram has not been updated since the last call to this method, an empty optional
+ * is returned.
+ */
+ public Optional<Snapshot> getSnapshotAndReset() {
+ if (!dirty.getAndSet(false)) {
+ return Optional.empty();
+ }
+
+ ImmutableLongArray.Builder bucketsSnapshotBuilder =
+ ImmutableLongArray.builder(buckets.length());
+ for (int i = 0; i < buckets.length(); i++) {
+ bucketsSnapshotBuilder.add(buckets.getAndSet(i, 0));
+ }
+ OutlierStatistic overflowSnapshot = overflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+ OutlierStatistic underflowSnapshot = underflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+
+ return Optional.of(
+ Snapshot.create(
+ underflowSnapshot, overflowSnapshot, bucketsSnapshotBuilder.build(), bucketType));
+ }
+
+ /** Create a histogram. */
+ public LockFreeHistogram(KV<MetricName, HistogramData.BucketType> kv) {
+ this.name = kv.getKey();
+ this.bucketType = kv.getValue();
+ this.buckets = new AtomicLongArray(bucketType.getNumBuckets());
+ }
+
+ @Override
+ public MetricName getName() {
+ return name;
+ }
+
+ private void updateInternal(double value) {
+ double rangeTo = bucketType.getRangeTo();
+ double rangeFrom = bucketType.getRangeFrom();
+ if (value >= rangeTo) {
+ recordTopRecordsValue(value);
+ } else if (value < rangeFrom) {
+ recordBottomRecordsValue(value);
+ } else {
+ recordInBoundsValue(value);
+ }
+ }
+
+ @Override
+ public void update(double value) {
+ updateInternal(value);
+ dirty.set(true);
+ }
+
+ public void update(double... values) {
Review Comment:
See comments above.
--
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "JayajP (via GitHub)" <gi...@apache.org>.
JayajP commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1544890781
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogramTest.java:
##########
@@ -0,0 +1,182 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link LockFreeHistogram}. */
+@RunWith(JUnit4.class)
+public class LockFreeHistogramTest {
+
+ @Test
+ public void testUpdate_OverflowValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(35, 40, 45);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedOverflow =
+ LockFreeHistogram.OutlierStatistic.create(120.0, 3L);
+ LockFreeHistogram.OutlierStatistic expectedUnderflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(0L, 0L, 0L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ assertThat(snapshot.get().underflowStatistic().mean(), equalTo(0.0));
+ assertThat(snapshot.get().overflowStatistic(), equalTo(expectedOverflow));
+ }
+
+ @Test
+ public void testUpdate_UnderflowValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(100, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(35, 40, 45);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedUnderflow =
+ LockFreeHistogram.OutlierStatistic.create(120.0, 3L);
+ LockFreeHistogram.OutlierStatistic expectedOverflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(0L, 0L, 0L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ assertThat(snapshot.get().underflowStatistic(), equalTo(expectedUnderflow));
+ }
+
+ @Test
+ public void testUpdate_InBoundsValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(5, 15, 25);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedOverflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ LockFreeHistogram.OutlierStatistic expectedUnderflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(1L, 1L, 1L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ }
+
+ @Test
+ public void testUpdate_EmptySnapshot() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(5, 15, 25);
+ Optional<LockFreeHistogram.Snapshot> snapshot_1 = histogram.getSnapshotAndReset();
+
+ assertThat(snapshot_1.isPresent(), equalTo(true));
+
+ Optional<LockFreeHistogram.Snapshot> snapshot_2 = histogram.getSnapshotAndReset();
+ assertThat(snapshot_2.isPresent(), equalTo(false));
+ }
+
+ /** A runnable records 200 values and then calls getSnapshotAndReset. */
+ private static class UpdateHistogramRunnable implements Runnable {
+ private final LockFreeHistogram histogram;
+ private final int val;
+ private Optional<LockFreeHistogram.Snapshot> snapshot;
+
+ private static final long valuesRecorded = 200L;
+
+ public UpdateHistogramRunnable(LockFreeHistogram histogram, int val) {
+ this.histogram = histogram;
+ this.val = val;
+ this.snapshot = Optional.empty();
+ }
+
+ @Override
+ public void run() {
+ for (long j = 0; j < valuesRecorded; j++) {
+ histogram.update(val);
+ }
+ snapshot = histogram.getSnapshotAndReset();
+ }
+
+ public long totalCountInSnapshot() {
+ if (snapshot.isPresent()) {
+ return snapshot.get().totalCount();
+ }
+ return 0;
+ }
+
+ public static long numValuesRecorded() {
+ return valuesRecorded;
+ }
+ }
+
+ @Test
+ public void testUpdateAndSnapshots_MultipleThreads() {
+ int numRunnables = 200;
+ ExecutorService executor = Executors.newFixedThreadPool(numRunnables);
+ HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 10);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+
+ UpdateHistogramRunnable[] runnables = new UpdateHistogramRunnable[numRunnables];
+ for (int i = 0; i < numRunnables; i++) {
+ runnables[i] = new UpdateHistogramRunnable(histogram, i);
+ executor.execute(runnables[i]);
+ }
+
+ try {
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1550659948
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogramTest.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link LockFreeHistogram}. */
+@RunWith(JUnit4.class)
+public class LockFreeHistogramTest {
+
+ @Test
+ public void testUpdate_OverflowValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(35, 40, 45);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedOverflow =
+ LockFreeHistogram.OutlierStatistic.create(120.0, 3L);
+ LockFreeHistogram.OutlierStatistic expectedUnderflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(0L, 0L, 0L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ assertThat(snapshot.get().underflowStatistic().mean(), equalTo(0.0));
+ assertThat(snapshot.get().overflowStatistic(), equalTo(expectedOverflow));
+ }
+
+ @Test
+ public void testUpdate_UnderflowValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(100, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(35, 40, 45);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedUnderflow =
+ LockFreeHistogram.OutlierStatistic.create(120.0, 3L);
+ LockFreeHistogram.OutlierStatistic expectedOverflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(0L, 0L, 0L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ assertThat(snapshot.get().underflowStatistic(), equalTo(expectedUnderflow));
+ }
+
+ @Test
+ public void testUpdate_InBoundsValues() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(5, 15, 25);
+ Optional<LockFreeHistogram.Snapshot> snapshot = histogram.getSnapshotAndReset();
+
+ LockFreeHistogram.OutlierStatistic expectedOverflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ LockFreeHistogram.OutlierStatistic expectedUnderflow = LockFreeHistogram.OutlierStatistic.EMPTY;
+ ImmutableLongArray expectedBuckets = ImmutableLongArray.of(1L, 1L, 1L);
+ LockFreeHistogram.Snapshot expectedSnapshot =
+ LockFreeHistogram.Snapshot.create(
+ expectedUnderflow, expectedOverflow, expectedBuckets, bucketType);
+
+ assertThat(snapshot.isPresent(), equalTo(true));
+ assertThat(snapshot.get(), equalTo(expectedSnapshot));
+ }
+
+ @Test
+ public void testUpdate_EmptySnapshot() {
+ HistogramData.BucketType bucketType = HistogramData.LinearBuckets.of(0, 10, 3);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+ histogram.update(5, 15, 25);
+ Optional<LockFreeHistogram.Snapshot> snapshot_1 = histogram.getSnapshotAndReset();
+
+ assertThat(snapshot_1.isPresent(), equalTo(true));
+
+ Optional<LockFreeHistogram.Snapshot> snapshot_2 = histogram.getSnapshotAndReset();
+ assertThat(snapshot_2.isPresent(), equalTo(false));
+ }
+
+ /** A runnable records 200 values and then calls getSnapshotAndReset. */
+ private static class UpdateHistogramRunnable implements Runnable {
+ private final LockFreeHistogram histogram;
+ private final int val;
+ private final CountDownLatch startSignal;
+ private final CountDownLatch doneSignal;
+ private Optional<LockFreeHistogram.Snapshot> snapshot;
+
+ private static final long valuesRecorded = 200L;
+
+ public UpdateHistogramRunnable(
+ LockFreeHistogram histogram,
+ int val,
+ CountDownLatch startSignal,
+ CountDownLatch doneSignal) {
+ this.histogram = histogram;
+ this.val = val;
+ this.startSignal = startSignal;
+ this.doneSignal = doneSignal;
+ this.snapshot = Optional.empty();
+ }
+
+ @Override
+ public void run() {
+ try {
+ startSignal.await();
+ } catch (InterruptedException e) {
+ return;
+ }
+
+ for (long j = 0; j < valuesRecorded; j++) {
+ histogram.update(val);
+ }
+ snapshot = histogram.getSnapshotAndReset();
+
+ doneSignal.countDown();
+ }
+
+ public long totalCountInSnapshot() {
+ if (snapshot.isPresent()) {
+ return snapshot.get().totalCount();
+ }
+ return 0;
+ }
+
+ public static long numValuesRecorded() {
+ return valuesRecorded;
+ }
+ }
+
+ @Test
+ public void testUpdateAndSnapshots_MultipleThreads() {
+ int numRunnables = 200;
+ CountDownLatch startSignal = new CountDownLatch(1);
+ CountDownLatch doneSignal = new CountDownLatch(numRunnables);
+ ExecutorService executor = Executors.newFixedThreadPool(numRunnables);
+
+ HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 10);
+ LockFreeHistogram histogram =
+ new LockFreeHistogram(KV.of(MetricName.named("name", "namespace"), bucketType));
+
+ UpdateHistogramRunnable[] runnables = new UpdateHistogramRunnable[numRunnables];
+ for (int i = 0; i < numRunnables; i++) {
+ runnables[i] = new UpdateHistogramRunnable(histogram, i, startSignal, doneSignal);
+ executor.execute(runnables[i]);
Review Comment:
you will want to use executor.submit in the loop
or create the array and then use executor.invokeAll() here
call execute 1 at a time will still execute it sequentially on 1 thread since execute blocks
--
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "JayajP (via GitHub)" <gi...@apache.org>.
JayajP commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1543498261
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
+
+ /**
+ * Represents the sum and mean of a collection of numbers. Used to represent the
+ * underflow/overflow statistics of a histogram.
+ */
+ @AutoValue
+ public abstract static class OutlierStatistic implements Serializable {
+ abstract double sum();
+
+ public abstract long count();
+
+ public static final OutlierStatistic EMPTY = create(0, 0);
+
+ public static OutlierStatistic create(double sum, long count) {
+ return new AutoValue_LockFreeHistogram_OutlierStatistic(sum, count);
+ }
+
+ public OutlierStatistic combine(double value) {
+ return create(sum() + value, count() + 1);
+ }
+
+ public double mean() {
+ if (count() == 0) {
+ return 0;
+ }
+ return sum() / count();
+ }
+ }
+
+ /**
+ * The snapshot of a histogram. The snapshot contains the overflow/underflow statistic, number of
+ * values recorded in each bucket, and the BucketType of the underlying histogram.
+ */
+ @AutoValue
+ public abstract static class Snapshot {
+ public abstract OutlierStatistic underflowStatistic();
+
+ public abstract OutlierStatistic overflowStatistic();
+
+ public abstract ImmutableLongArray buckets();
+
+ public abstract HistogramData.BucketType bucketType();
+
+ public static Snapshot create(
+ OutlierStatistic underflowStatistic,
+ OutlierStatistic overflowStatistic,
+ ImmutableLongArray buckets,
+ HistogramData.BucketType bucketType) {
+ return new AutoValue_LockFreeHistogram_Snapshot(
+ underflowStatistic, overflowStatistic, buckets, bucketType);
+ }
+
+ @Memoized
+ public long totalCount() {
+ long count = 0;
+ count += underflowStatistic().count();
+ count += overflowStatistic().count();
+ count += buckets().stream().sum();
+
+ return count;
+ }
+ }
+
+ /**
+ * Extract a delta update of this histogram. Update represents values that have been recorded in
+ * this histogram since the last time this method was called.
+ *
+ * <p>If this histogram is being updated concurrent to this method, then the returned snapshot is
+ * not guarenteed to contain those updates. However, those updates are not dropped and will be
+ * represented in a future call to this method.
+ *
+ * <p>If this histogram has not been updated since the last call to this method, an empty optional
+ * is returned.
+ */
+ public Optional<Snapshot> getSnapshotAndReset() {
+ if (!dirty.getAndSet(false)) {
+ return Optional.empty();
+ }
+
+ ImmutableLongArray.Builder bucketsSnapshotBuilder =
+ ImmutableLongArray.builder(buckets.length());
+ for (int i = 0; i < buckets.length(); i++) {
+ bucketsSnapshotBuilder.add(buckets.getAndSet(i, 0));
+ }
+ OutlierStatistic overflowSnapshot = overflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+ OutlierStatistic underflowSnapshot = underflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+
+ return Optional.of(
+ Snapshot.create(
+ underflowSnapshot, overflowSnapshot, bucketsSnapshotBuilder.build(), bucketType));
+ }
+
+ /** Create a histogram. */
+ public LockFreeHistogram(KV<MetricName, HistogramData.BucketType> kv) {
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] Implement LockFreeHistogram and use it for PerWorkerHistograms [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #30769:
URL: https://github.com/apache/beam/pull/30769#discussion_r1542045888
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
Review Comment:
public final class?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
Review Comment:
private final
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
Review Comment:
members should be private final*
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogramTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+import java.util.Optional;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link LockFreeHistogram}. */
+@RunWith(JUnit4.class)
+public class LockFreeHistogramTest {
Review Comment:
Can you add a test where multiple threads are trying to update an instance?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
+
+ /**
+ * Represents the sum and mean of a collection of numbers. Used to represent the
+ * underflow/overflow statistics of a histogram.
+ */
+ @AutoValue
+ public abstract static class OutlierStatistic implements Serializable {
+ abstract double sum();
+
+ public abstract long count();
+
+ public static final OutlierStatistic EMPTY = create(0, 0);
+
+ public static OutlierStatistic create(double sum, long count) {
+ return new AutoValue_LockFreeHistogram_OutlierStatistic(sum, count);
+ }
+
+ public OutlierStatistic combine(double value) {
+ return create(sum() + value, count() + 1);
+ }
+
+ public double mean() {
+ if (count() == 0) {
+ return 0;
+ }
+ return sum() / count();
+ }
+ }
+
+ /**
+ * The snapshot of a histogram. The snapshot contains the overflow/underflow statistic, number of
+ * values recorded in each bucket, and the BucketType of the underlying histogram.
+ */
+ @AutoValue
+ public abstract static class Snapshot {
+ public abstract OutlierStatistic underflowStatistic();
+
+ public abstract OutlierStatistic overflowStatistic();
+
+ public abstract ImmutableLongArray buckets();
+
+ public abstract HistogramData.BucketType bucketType();
+
+ public static Snapshot create(
+ OutlierStatistic underflowStatistic,
+ OutlierStatistic overflowStatistic,
+ ImmutableLongArray buckets,
+ HistogramData.BucketType bucketType) {
+ return new AutoValue_LockFreeHistogram_Snapshot(
+ underflowStatistic, overflowStatistic, buckets, bucketType);
+ }
+
+ @Memoized
+ public long totalCount() {
+ long count = 0;
+ count += underflowStatistic().count();
+ count += overflowStatistic().count();
+ count += buckets().stream().sum();
+
+ return count;
+ }
+ }
+
+ /**
+ * Extract a delta update of this histogram. Update represents values that have been recorded in
+ * this histogram since the last time this method was called.
+ *
+ * <p>If this histogram is being updated concurrent to this method, then the returned snapshot is
+ * not guarenteed to contain those updates. However, those updates are not dropped and will be
+ * represented in a future call to this method.
+ *
+ * <p>If this histogram has not been updated since the last call to this method, an empty optional
+ * is returned.
+ */
+ public Optional<Snapshot> getSnapshotAndReset() {
+ if (!dirty.getAndSet(false)) {
+ return Optional.empty();
+ }
+
+ ImmutableLongArray.Builder bucketsSnapshotBuilder =
+ ImmutableLongArray.builder(buckets.length());
+ for (int i = 0; i < buckets.length(); i++) {
+ bucketsSnapshotBuilder.add(buckets.getAndSet(i, 0));
+ }
+ OutlierStatistic overflowSnapshot = overflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+ OutlierStatistic underflowSnapshot = underflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+
+ return Optional.of(
+ Snapshot.create(
+ underflowSnapshot, overflowSnapshot, bucketsSnapshotBuilder.build(), bucketType));
+ }
+
+ /** Create a histogram. */
+ public LockFreeHistogram(KV<MetricName, HistogramData.BucketType> kv) {
Review Comment:
move constructor under members for readability.
nit: (might be my personal style preference, so up to you) but also initialize all members in constructor instead of where declared for readability
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
Review Comment:
future*
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
+
+ /**
+ * Represents the sum and mean of a collection of numbers. Used to represent the
+ * underflow/overflow statistics of a histogram.
+ */
+ @AutoValue
+ public abstract static class OutlierStatistic implements Serializable {
+ abstract double sum();
+
+ public abstract long count();
+
+ public static final OutlierStatistic EMPTY = create(0, 0);
+
+ public static OutlierStatistic create(double sum, long count) {
+ return new AutoValue_LockFreeHistogram_OutlierStatistic(sum, count);
+ }
+
+ public OutlierStatistic combine(double value) {
+ return create(sum() + value, count() + 1);
+ }
+
+ public double mean() {
+ if (count() == 0) {
+ return 0;
+ }
+ return sum() / count();
+ }
+ }
+
+ /**
+ * The snapshot of a histogram. The snapshot contains the overflow/underflow statistic, number of
+ * values recorded in each bucket, and the BucketType of the underlying histogram.
+ */
+ @AutoValue
+ public abstract static class Snapshot {
+ public abstract OutlierStatistic underflowStatistic();
+
+ public abstract OutlierStatistic overflowStatistic();
+
+ public abstract ImmutableLongArray buckets();
+
+ public abstract HistogramData.BucketType bucketType();
+
+ public static Snapshot create(
+ OutlierStatistic underflowStatistic,
+ OutlierStatistic overflowStatistic,
+ ImmutableLongArray buckets,
+ HistogramData.BucketType bucketType) {
+ return new AutoValue_LockFreeHistogram_Snapshot(
+ underflowStatistic, overflowStatistic, buckets, bucketType);
+ }
+
+ @Memoized
+ public long totalCount() {
+ long count = 0;
+ count += underflowStatistic().count();
+ count += overflowStatistic().count();
+ count += buckets().stream().sum();
+
+ return count;
+ }
+ }
+
+ /**
+ * Extract a delta update of this histogram. Update represents values that have been recorded in
+ * this histogram since the last time this method was called.
+ *
+ * <p>If this histogram is being updated concurrent to this method, then the returned snapshot is
+ * not guarenteed to contain those updates. However, those updates are not dropped and will be
+ * represented in a future call to this method.
+ *
+ * <p>If this histogram has not been updated since the last call to this method, an empty optional
+ * is returned.
+ */
+ public Optional<Snapshot> getSnapshotAndReset() {
+ if (!dirty.getAndSet(false)) {
+ return Optional.empty();
+ }
+
+ ImmutableLongArray.Builder bucketsSnapshotBuilder =
+ ImmutableLongArray.builder(buckets.length());
+ for (int i = 0; i < buckets.length(); i++) {
+ bucketsSnapshotBuilder.add(buckets.getAndSet(i, 0));
+ }
+ OutlierStatistic overflowSnapshot = overflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+ OutlierStatistic underflowSnapshot = underflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+
+ return Optional.of(
+ Snapshot.create(
+ underflowSnapshot, overflowSnapshot, bucketsSnapshotBuilder.build(), bucketType));
+ }
+
+ /** Create a histogram. */
+ public LockFreeHistogram(KV<MetricName, HistogramData.BucketType> kv) {
+ this.name = kv.getKey();
+ this.bucketType = kv.getValue();
+ this.buckets = new AtomicLongArray(bucketType.getNumBuckets());
+ }
+
+ @Override
+ public MetricName getName() {
+ return name;
+ }
+
+ private void updateInternal(double value) {
+ double rangeTo = bucketType.getRangeTo();
+ double rangeFrom = bucketType.getRangeFrom();
+ if (value >= rangeTo) {
+ recordTopRecordsValue(value);
+ } else if (value < rangeFrom) {
+ recordBottomRecordsValue(value);
+ } else {
+ recordInBoundsValue(value);
+ }
+ }
+
+ @Override
+ public void update(double value) {
+ updateInternal(value);
+ dirty.set(true);
+ }
+
+ public void update(double... values) {
Review Comment:
is this threadsafe? seems racy since it is not using the atomics API or locks
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
Review Comment:
annotate class with:
`@ThreadSafe`
`@Internal` (from beam lib)
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/LockFreeHistogram.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.ImmutableLongArray;
+
+/**
+ * A lock free implementation of {@link org.apache.beam.sdk.metrics.Histogram}. This class supports
+ * extracting delta updates with the {@link #getSnapshotAndReset} method.
+ */
+public class LockFreeHistogram implements Histogram {
+ private final HistogramData.BucketType bucketType;
+ private final AtomicLongArray buckets;
+ private final MetricName name;
+
+ private AtomicReference<OutlierStatistic> underflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+ private AtomicReference<OutlierStatistic> overflowStatistic =
+ new AtomicReference<LockFreeHistogram.OutlierStatistic>(OutlierStatistic.EMPTY);
+
+ /**
+ * Whether this histogram has updates that have not been extracted by {@code getSnapshotAndReset}.
+ * This values should be flipped to true AFTER recording a value, and flipped to false BEFORE
+ * extracting a snapshot. This ensures that recorded values will always be seen by a futrue {@code
+ * getSnapshotAndReset} call.
+ */
+ private AtomicBoolean dirty = new AtomicBoolean(false);
+
+ /**
+ * Represents the sum and mean of a collection of numbers. Used to represent the
+ * underflow/overflow statistics of a histogram.
+ */
+ @AutoValue
+ public abstract static class OutlierStatistic implements Serializable {
+ abstract double sum();
+
+ public abstract long count();
+
+ public static final OutlierStatistic EMPTY = create(0, 0);
+
+ public static OutlierStatistic create(double sum, long count) {
+ return new AutoValue_LockFreeHistogram_OutlierStatistic(sum, count);
+ }
+
+ public OutlierStatistic combine(double value) {
+ return create(sum() + value, count() + 1);
+ }
+
+ public double mean() {
+ if (count() == 0) {
+ return 0;
+ }
+ return sum() / count();
+ }
+ }
+
+ /**
+ * The snapshot of a histogram. The snapshot contains the overflow/underflow statistic, number of
+ * values recorded in each bucket, and the BucketType of the underlying histogram.
+ */
+ @AutoValue
+ public abstract static class Snapshot {
+ public abstract OutlierStatistic underflowStatistic();
+
+ public abstract OutlierStatistic overflowStatistic();
+
+ public abstract ImmutableLongArray buckets();
+
+ public abstract HistogramData.BucketType bucketType();
+
+ public static Snapshot create(
+ OutlierStatistic underflowStatistic,
+ OutlierStatistic overflowStatistic,
+ ImmutableLongArray buckets,
+ HistogramData.BucketType bucketType) {
+ return new AutoValue_LockFreeHistogram_Snapshot(
+ underflowStatistic, overflowStatistic, buckets, bucketType);
+ }
+
+ @Memoized
+ public long totalCount() {
+ long count = 0;
+ count += underflowStatistic().count();
+ count += overflowStatistic().count();
+ count += buckets().stream().sum();
+
+ return count;
+ }
+ }
+
+ /**
+ * Extract a delta update of this histogram. Update represents values that have been recorded in
+ * this histogram since the last time this method was called.
+ *
+ * <p>If this histogram is being updated concurrent to this method, then the returned snapshot is
+ * not guarenteed to contain those updates. However, those updates are not dropped and will be
+ * represented in a future call to this method.
+ *
+ * <p>If this histogram has not been updated since the last call to this method, an empty optional
+ * is returned.
+ */
+ public Optional<Snapshot> getSnapshotAndReset() {
+ if (!dirty.getAndSet(false)) {
+ return Optional.empty();
+ }
+
+ ImmutableLongArray.Builder bucketsSnapshotBuilder =
+ ImmutableLongArray.builder(buckets.length());
+ for (int i = 0; i < buckets.length(); i++) {
+ bucketsSnapshotBuilder.add(buckets.getAndSet(i, 0));
+ }
+ OutlierStatistic overflowSnapshot = overflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+ OutlierStatistic underflowSnapshot = underflowStatistic.getAndSet(OutlierStatistic.EMPTY);
+
+ return Optional.of(
+ Snapshot.create(
+ underflowSnapshot, overflowSnapshot, bucketsSnapshotBuilder.build(), bucketType));
+ }
+
+ /** Create a histogram. */
+ public LockFreeHistogram(KV<MetricName, HistogramData.BucketType> kv) {
+ this.name = kv.getKey();
+ this.bucketType = kv.getValue();
+ this.buckets = new AtomicLongArray(bucketType.getNumBuckets());
+ }
+
+ @Override
+ public MetricName getName() {
+ return name;
+ }
+
+ private void updateInternal(double value) {
+ double rangeTo = bucketType.getRangeTo();
+ double rangeFrom = bucketType.getRangeFrom();
+ if (value >= rangeTo) {
+ recordTopRecordsValue(value);
+ } else if (value < rangeFrom) {
+ recordBottomRecordsValue(value);
+ } else {
+ recordInBoundsValue(value);
+ }
+ }
+
+ @Override
+ public void update(double value) {
+ updateInternal(value);
+ dirty.set(true);
Review Comment:
is this threadsafe? seems racy since it is not using the atomics API or locks
--
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