Skip to content

Commit 72cb6b1

Browse files
lenin-jaganathanlenin-jaganathan
committed
Add concurrency to exponential histogram
- Rescaling is not frequent behaviour (for cumulative scale should normalize after initial few recording, for delta it could increase/decrease but would be mostly stable). Instead of opting for a full synchronization on writes (which would have less performance), the choice of ReadWriteLock is chosen.
1 parent 9018b50 commit 72cb6b1

File tree

5 files changed

+95
-37
lines changed

5 files changed

+95
-37
lines changed

‎implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/internal/Base2ExponentialHistogram.java

Lines changed: 63 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import java.util.List;
2323
import java.util.concurrent.TimeUnit;
2424
import java.util.concurrent.atomic.LongAdder;
25+
import java.util.concurrent.locks.ReentrantReadWriteLock;
2526
import java.util.stream.Collectors;
2627

2728
import io.micrometer.common.lang.Nullable;
@@ -49,6 +50,21 @@
4950
*/
5051
public abstract class Base2ExponentialHistogram implements Histogram {
5152

53+
/*
54+
* This is an alternative to full-blown synchronization on the Histogram which would
55+
* severely affect the performance under high concurrent cases.
56+
*
57+
* Read lock guards adding data into histogram i,e multiple values can be recorded at
58+
* the same time for a given scale. If scale needs to be changed, write-lock need to
59+
* be acquired and all the operations are blocked until new scale is determined and
60+
* updated with.
61+
*/
62+
private final ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
63+
64+
private final ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
65+
66+
private final ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
67+
5268
private final int maxScale;
5369

5470
private final int maxBucketsCount;
@@ -131,10 +147,24 @@ public HistogramSnapshot takeSnapshot(final long count, final double total, fina
131147
}
132148

133149
/**
134-
* Returns the snapshot of current recorded values.
150+
* Returns the snapshot of current recorded values. This method is thread-safe and
151+
* block concurrent recordings to the underlying histogram.
135152
*/
136-
ExponentialHistogramSnapShot getCurrentValuesSnapshot() {
137-
return (circularCountHolder.isEmpty() && zeroCount.longValue() == 0)
153+
ExponentialHistogramSnapShot getCurrentValuesSnapshot(final boolean shouldResetHistogram) {
154+
writeLock.lock();
155+
try {
156+
ExponentialHistogramSnapShot exponentialHistogramSnapShot = getCurrentValuesSnapshot();
157+
if (shouldResetHistogram)
158+
reset();
159+
return exponentialHistogramSnapShot;
160+
}
161+
finally {
162+
writeLock.unlock();
163+
}
164+
}
165+
166+
private ExponentialHistogramSnapShot getCurrentValuesSnapshot() {
167+
return circularCountHolder.isEmpty() && zeroCount.longValue() == 0
138168
? DefaultExponentialHistogramSnapShot.getEmptySnapshotForScale(scale)
139169
: new DefaultExponentialHistogramSnapShot(scale, zeroCount.longValue(), zeroThreshold,
140170
new ExponentialBuckets(getOffset(), getBucketCounts()), EMPTY_EXPONENTIAL_BUCKET);
@@ -168,24 +198,46 @@ public void recordDouble(double value) {
168198
return;
169199
}
170200

171-
int index = base2IndexProvider.getIndexForValue(value);
172-
if (!circularCountHolder.increment(index, 1)) {
173-
synchronized (this) {
174-
int downScaleFactor = getDownScaleFactor(index);
201+
recordInternal(value);
202+
}
203+
204+
private void recordInternal(double value) {
205+
boolean hasRecorded;
206+
readLock.lock();
207+
try {
208+
hasRecorded = incrementValue(value);
209+
}
210+
finally {
211+
readLock.unlock();
212+
}
213+
214+
if (!hasRecorded) {
215+
// If recording fails, then we MIGHT have to re-scale. Try re-scaling.
216+
writeLock.lock();
217+
try {
218+
final int downScaleFactor = getDownScaleFactor(base2IndexProvider.getIndexForValue(value));
175219
downScale(downScaleFactor);
176-
index = base2IndexProvider.getIndexForValue(value);
177-
circularCountHolder.increment(index, 1);
220+
// Record the value within the write lock to guarantee write.
221+
incrementValue(value);
222+
}
223+
finally {
224+
writeLock.unlock();
178225
}
179226
}
180227
}
181228

229+
private boolean incrementValue(final double value) {
230+
return circularCountHolder.increment(base2IndexProvider.getIndexForValue(value), 1);
231+
}
232+
182233
/**
183234
* Reduces the scale of the histogram by downScaleFactor. The buckets are merged to
184235
* align with the exponential scale.
185236
* @param downScaleFactor - the factor to downscale this histogram.
186237
*/
187238
private void downScale(int downScaleFactor) {
188239
if (downScaleFactor == 0) {
240+
// Should never happen.
189241
return;
190242
}
191243

@@ -286,12 +338,12 @@ private List<Long> getBucketCounts() {
286338
* Reset the current values and possibly increase the scale based on current recorded
287339
* values;
288340
*/
289-
synchronized void reset() {
341+
// VisibleForTesting
342+
void reset() {
290343
int upscaleFactor = getUpscaleFactor();
291344
if (upscaleFactor > 0) {
292345
this.updateScale(this.scale + upscaleFactor);
293346
}
294-
295347
this.circularCountHolder.reset();
296348
this.zeroCount.reset();
297349
}

‎implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/internal/CircularCountHolder.java

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,18 @@ else if (index < startIndex) {
8181
startIndex = index;
8282
}
8383

84-
counts.addAndGet(getRelativeIndex(index), incrementBy);
85-
return true;
84+
final int relativeIndex = getRelativeIndex(index);
85+
if (relativeIndex >= 0 && relativeIndex < length) {
86+
counts.addAndGet(relativeIndex, incrementBy);
87+
return true;
88+
}
89+
/*
90+
* This case is only possible (rarely) during initial writes on the histogram
91+
* where there might be concurrent threads modifying the index and not fully
92+
* synchronized externally. In this case, we would just notify the caller that
93+
* write has failed and let the caller take a call on what to do.
94+
*/
95+
return false;
8696
}
8797

8898
private int getRelativeIndex(int index) {

‎implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/internal/CumulativeBase2ExponentialHistogram.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,8 +59,8 @@ public ExponentialHistogramSnapShot getLatestExponentialHistogramSnapshot() {
5959
}
6060

6161
@Override
62-
synchronized void takeExponentialHistogramSnapShot() {
63-
this.exponentialHistogramSnapShot = getCurrentValuesSnapshot();
62+
void takeExponentialHistogramSnapShot() {
63+
this.exponentialHistogramSnapShot = getCurrentValuesSnapshot(false);
6464
}
6565

6666
}

‎implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/internal/DeltaBase2ExponentialHistogram.java

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ public ExponentialHistogramSnapShot getLatestExponentialHistogramSnapshot() {
6666
}
6767

6868
@Override
69-
synchronized void takeExponentialHistogramSnapShot() {
69+
void takeExponentialHistogramSnapShot() {
7070
stepExponentialHistogramSnapShot.poll();
7171
}
7272

@@ -82,12 +82,8 @@ public StepExponentialHistogramSnapShot(final Clock clock, final long stepMillis
8282
}
8383

8484
@Override
85-
protected synchronized Supplier<ExponentialHistogramSnapShot> valueSupplier() {
86-
return () -> {
87-
ExponentialHistogramSnapShot latestSnapShot = getCurrentValuesSnapshot();
88-
reset();
89-
return latestSnapShot;
90-
};
85+
protected Supplier<ExponentialHistogramSnapShot> valueSupplier() {
86+
return () -> getCurrentValuesSnapshot(true);
9187
}
9288

9389
@Override

‎implementations/micrometer-registry-otlp/src/test/java/io/micrometer/registry/otlp/internal/Base2ExponentialHistogramTest.java

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,8 @@ void testRecordDouble() {
5151
// 1 Always belongs to index 0.
5252
base2ExponentialHistogram.recordDouble(1.000000000001);
5353
assertThat(base2ExponentialHistogram.getScale()).isEqualTo(MAX_SCALE);
54-
assertThat(base2ExponentialHistogram.getCurrentValuesSnapshot().zeroCount()).isZero();
55-
assertThat(getAllBucketsCountSum(base2ExponentialHistogram.getCurrentValuesSnapshot())).isEqualTo(1);
54+
assertThat(base2ExponentialHistogram.getCurrentValuesSnapshot(false).zeroCount()).isZero();
55+
assertThat(getAllBucketsCountSum(base2ExponentialHistogram.getCurrentValuesSnapshot(false))).isEqualTo(1);
5656
}
5757

5858
@Test
@@ -66,7 +66,7 @@ void testRecordTimeBased() {
6666
// calling
6767
// recordDouble(2).
6868

69-
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
69+
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
7070
assertThat(currentSnapshot.zeroCount()).isZero();
7171
assertThat(currentSnapshot.scale()).isLessThan(MAX_SCALE);
7272
assertThat(getAllBucketsCountSum(currentSnapshot)).isEqualTo(2);
@@ -84,13 +84,13 @@ void testRecordTimeBasedInSeconds() {
8484
// This should be same as calling recordDouble(0.05).
8585
base2ExponentialHistogram.recordLong(Duration.ofMillis(50).toNanos());
8686

87-
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
87+
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
8888
assertThat(currentSnapshot.zeroCount()).isZero();
8989
assertThat(currentSnapshot.scale()).isLessThan(MAX_SCALE);
9090
assertThat(getAllBucketsCountSum(currentSnapshot)).isEqualTo(2);
9191

9292
base2ExponentialHistogram.recordLong(Duration.ofMillis(90).toNanos());
93-
currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
93+
currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
9494
assertThat(currentSnapshot.scale()).isEqualTo(1);
9595
assertThat(getAllBucketsCountSum(currentSnapshot)).isEqualTo(3);
9696
}
@@ -101,7 +101,7 @@ void testZeroThreshHold() {
101101
base2ExponentialHistogram.recordDouble(0.0);
102102
base2ExponentialHistogram.recordDouble(0.5);
103103

104-
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
104+
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
105105
assertThat(currentSnapshot.zeroThreshold()).isLessThan(1).isGreaterThan(0);
106106
assertThat(currentSnapshot.zeroCount()).isEqualTo(2);
107107
assertThat(currentSnapshot.scale()).isEqualTo(MAX_SCALE);
@@ -113,7 +113,7 @@ void testZeroThreshHold() {
113113
base2ExponentialHistogramWithZeroAsMin.recordDouble(Math.nextUp(0.0));
114114

115115
ExponentialHistogramSnapShot snapshotWithZeroAsMin = base2ExponentialHistogramWithZeroAsMin
116-
.getCurrentValuesSnapshot();
116+
.getCurrentValuesSnapshot(false);
117117
assertThat(snapshotWithZeroAsMin.zeroThreshold()).isEqualTo(0.0);
118118
assertThat(snapshotWithZeroAsMin.zeroCount()).isEqualTo(1);
119119
assertThat(snapshotWithZeroAsMin.scale()).isEqualTo(MAX_SCALE);
@@ -124,7 +124,7 @@ void testZeroThreshHold() {
124124
void testDownScale() {
125125
base2ExponentialHistogram.recordDouble(1.0001);
126126

127-
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
127+
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
128128
assertThat(currentSnapshot.zeroCount()).isZero();
129129
assertThat(currentSnapshot.scale()).isEqualTo(MAX_SCALE);
130130
assertThat(getAllBucketsCountSum(currentSnapshot)).isEqualTo(1);
@@ -182,11 +182,11 @@ void testUpscale() {
182182

183183
@Test
184184
void testValuesAtIndices() {
185-
ExponentialHistogramSnapShot currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
185+
ExponentialHistogramSnapShot currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
186186
assertThat(currentValueSnapshot.positive().bucketCounts()).isEmpty();
187187

188188
base2ExponentialHistogram.recordDouble(1.0001);
189-
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
189+
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
190190
assertThat(currentValueSnapshot.positive().offset()).isZero();
191191
assertThat(currentValueSnapshot.positive().bucketCounts().get(0)).isEqualTo(1);
192192
assertThat(currentValueSnapshot.positive().bucketCounts()).filteredOn(value -> value == 0).isEmpty();
@@ -195,7 +195,7 @@ void testValuesAtIndices() {
195195

196196
base2ExponentialHistogram.recordDouble(1.0076);
197197
base2ExponentialHistogram.recordDouble(1.008);
198-
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
198+
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
199199
assertThat(currentValueSnapshot.positive().offset()).isZero();
200200
assertThat(base2ExponentialHistogram.getScale()).isEqualTo(MAX_SCALE);
201201
assertThat(currentValueSnapshot.positive().bucketCounts().get(0)).isEqualTo(1);
@@ -206,7 +206,7 @@ void testValuesAtIndices() {
206206
// We will record a value that will downscale by 1 and this should merge 2
207207
// consecutive buckets into one.
208208
base2ExponentialHistogram.recordDouble(1.012);
209-
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
209+
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
210210
assertThat(currentValueSnapshot.positive().offset()).isZero();
211211
assertThat(base2ExponentialHistogram.getScale()).isEqualTo(MAX_SCALE - 1);
212212
assertThat(currentValueSnapshot.positive().bucketCounts().get(0)).isEqualTo(2);
@@ -216,7 +216,7 @@ void testValuesAtIndices() {
216216

217217
// The base will reduced by a factor of more than one,
218218
base2ExponentialHistogram.recordDouble(4);
219-
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
219+
currentValueSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
220220
assertThat(currentValueSnapshot.positive().offset()).isZero();
221221
assertThat(base2ExponentialHistogram.getScale()).isEqualTo(3);
222222
assertThat(currentValueSnapshot.positive().bucketCounts().get(0)).isEqualTo(5);
@@ -242,15 +242,15 @@ void reset() {
242242
base2ExponentialHistogram.recordDouble(1);
243243
base2ExponentialHistogram.recordDouble(2);
244244

245-
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
245+
ExponentialHistogramSnapShot currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
246246
final int intialScale = currentSnapshot.scale();
247247
assertThat(currentSnapshot.zeroCount()).isEqualTo(1);
248248
assertThat(currentSnapshot.scale()).isLessThan(MAX_SCALE);
249249
assertThat(currentSnapshot.positive().offset()).isNotZero();
250250
assertThat(getAllBucketsCountSum(currentSnapshot)).isEqualTo(2);
251251

252252
base2ExponentialHistogram.reset();
253-
currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot();
253+
currentSnapshot = base2ExponentialHistogram.getCurrentValuesSnapshot(false);
254254
assertThat(currentSnapshot.zeroCount()).isZero();
255255
assertThat(currentSnapshot.scale()).isEqualTo(intialScale);
256256
assertThat(currentSnapshot.positive().offset()).isZero();

0 commit comments

Comments
 (0)