Skip to content

Issue #1138 Replacement of sliding time window. #1139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
arteam merged 13 commits into from
Jun 14, 2017
Merged

Issue #1138 Replacement of sliding time window. #1139

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
7abc2cf
Drop-in replacement of sliding time window reservoir.
storozhukBM Apr 5, 2017
d30f41c
chunks with two indexes
storozhukBM Apr 5, 2017
1f9ecb1
snapshot gathering optimisation
storozhukBM Apr 6, 2017
36acdc4
jcstress tests and few fixes
storozhukBM Apr 24, 2017
ce1e0e6
JCStress tests improvements and additional JMH benchmark for concurre…
storozhukBM Jun 7, 2017
3ab321a
Load test plus results for new implementation.
storozhukBM Jun 12, 2017
b55701a
PR preparation. Clean-up
storozhukBM Jun 12, 2017
36d60c6
Load test results for old implementation.
storozhukBM Jun 12, 2017
8fd7bc2
Merge branch 'double_indexed_chunk' into 1138_replacement_of_sliding_…
storozhukBM Jun 12, 2017
13be79c
Test results clean up
storozhukBM Jun 12, 2017
ade445e
Code clean-up.
storozhukBM Jun 12, 2017
e60e167
Changes to ticks calculations in order to delay overflow and make onc…
storozhukBM Jun 13, 2017
f17c64b
Change synchronization mechanism.
storozhukBM Jun 14, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
snapshot gathering optimisation
  • Loading branch information
@storozhukBM
storozhukBM committed Apr 6, 2017
commit 1f9ecb1938bda4cc0e94f0fb3c6b6b29f53c47e7
32 changes: 16 additions & 16 deletions metrics-benchmarks/src/main/java/com/codahale/metrics/benchmarks/ReservoirBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,22 +30,6 @@ public class ReservoirBenchmark {
// It's intentionally not declared as final to avoid constant folding
private long nextValue = 0xFBFBABBA;

public static void main(String[] args) throws RunnerException {
Options opt = new OptionsBuilder()
.include(".*" + ReservoirBenchmark.class.getSimpleName() + ".*")
.warmupIterations(10)
.measurementIterations(10)
.addProfiler(GCProfiler.class)
.measurementTime(TimeValue.seconds(3))
.timeUnit(TimeUnit.MICROSECONDS)
.mode(Mode.AverageTime)
.threads(4)
.forks(1)
.build();

new Runner(opt).run();
}

@Benchmark
public Object perfUniformReservoir() {
uniform.update(nextValue);
Expand Down Expand Up @@ -76,4 +60,20 @@ public Object perfSlidingTimeWindowReservoir() {
return slidingTime;
}

public static void main(String[] args) throws RunnerException {
Options opt = new OptionsBuilder()
.include(".*" + ReservoirBenchmark.class.getSimpleName() + ".*")
.warmupIterations(10)
.measurementIterations(10)
.addProfiler(GCProfiler.class)
.measurementTime(TimeValue.seconds(3))
.timeUnit(TimeUnit.MICROSECONDS)
.mode(Mode.AverageTime)
.threads(4)
.forks(1)
.build();

new Runner(opt).run();
}

}
36 changes: 18 additions & 18 deletions metrics-benchmarks/src/main/java/com/codahale/metrics/benchmarks/reservoirs_bench_result.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,31 +1,31 @@
# Run complete. Total time: 00:03:51
# Run complete. Total time: 00:03:53

Benchmark Mode Samples Score Score error Units
c.c.m.b.ReservoirBenchmark.arrReservoir avgt 10 1.074 0.007 us/op
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.count.profiled avgt 10 48.000 NaN counts
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.count.total avgt 10 59.000 NaN counts
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.time.profiled avgt 10 142.000 NaN ms
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.time.total avgt 10 216.000 NaN ms
c.c.m.b.ReservoirBenchmark.arrReservoir avgt 10 0.581 0.030 us/op
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.count.profiled avgt 10 20.000 NaN counts
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.count.total avgt 10 27.000 NaN counts
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.time.profiled avgt 10 202.000 NaN ms
c.c.m.b.ReservoirBenchmark.arrReservoir:@gc.time.total avgt 10 300.000 NaN ms

c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir avgt 10 0.610 0.051 us/op
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.count.profiled avgt 10 494.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.count.total avgt 10 596.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.time.profiled avgt 10 248.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.time.total avgt 10 310.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir avgt 10 0.655 0.008 us/op
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.count.profiled avgt 10 483.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.count.total avgt 10 574.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.time.profiled avgt 10 228.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfExponentiallyDecayingReservoir:@gc.time.total avgt 10 285.000 NaN ms

c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir avgt 10 2.851 0.884 us/op
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.count.profiled avgt 10 20.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.count.total avgt 10 34.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.time.profiled avgt 10 9506.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.time.total avgt 10 14598.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir avgt 10 2.833 1.303 us/op
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.count.profiled avgt 10 19.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.count.total avgt 10 32.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.time.profiled avgt 10 10838.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfSlidingTimeWindowReservoir:@gc.time.total avgt 10 16019.000 NaN ms

c.c.m.b.ReservoirBenchmark.perfSlidingWindowReservoir avgt 10 0.228 0.003 us/op
c.c.m.b.ReservoirBenchmark.perfSlidingWindowReservoir avgt 10 0.216 0.006 us/op
c.c.m.b.ReservoirBenchmark.perfSlidingWindowReservoir:@gc.count.profiled avgt 10 0.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfSlidingWindowReservoir:@gc.count.total avgt 10 0.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfSlidingWindowReservoir:@gc.time.profiled avgt 10 0.000 NaN ms
c.c.m.b.ReservoirBenchmark.perfSlidingWindowReservoir:@gc.time.total avgt 10 0.000 NaN ms

c.c.m.b.ReservoirBenchmark.perfUniformReservoir avgt 10 0.088 0.000 us/op
c.c.m.b.ReservoirBenchmark.perfUniformReservoir avgt 10 0.087 0.001 us/op
c.c.m.b.ReservoirBenchmark.perfUniformReservoir:@gc.count.profiled avgt 10 0.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfUniformReservoir:@gc.count.total avgt 10 0.000 NaN counts
c.c.m.b.ReservoirBenchmark.perfUniformReservoir:@gc.time.profiled avgt 10 0.000 NaN ms
Expand Down
138 changes: 49 additions & 89 deletions metrics-core/src/main/java/com/codahale/metrics/ChunkedAssociativeLongArray.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
package com.codahale.metrics;

import static java.lang.Math.min;
import static java.util.Arrays.binarySearch;

import java.util.ArrayDeque;
Expand All @@ -12,9 +11,9 @@
*/
public class ChunkedAssociativeLongArray {
private static final long[] EMPTY = new long[0];
private static final int DEFAULT_CHUNK_SIZE = 100;
private final ReentrantLock activeChunkLock = new ReentrantLock();
private static final int DEFAULT_CHUNK_SIZE = 512;

private final ReentrantLock activeChunkLock = new ReentrantLock();
private final int chunkSize;
private Chunk activeChunk;

Expand All @@ -40,114 +39,72 @@ public void put(long key, long value) {
}
}

public int traverse(long minKey, Deque<Chunk> traversedChunksDeque) {
public int traverse(Deque<Chunk> traversedChunksDeque) {
Chunk currentChunk;
int currentChunkCursor;
int currentChunkStartIndex;
activeChunkLock.lock();
try {
currentChunk = activeChunk;
currentChunkStartIndex = activeChunk.startIndex;
currentChunkCursor = activeChunk.cursor;
} finally {
activeChunkLock.unlock();
}

if (!isFirstElementIsGreaterEqualThanKey(currentChunk, minKey)) {
return 0;
}

int valuesSize = 0;
while (true) {
int realIndex = findFirstIndexOfGreaterEqualElements(
currentChunk.keys, currentChunkStartIndex, currentChunkCursor, minKey
);
int actualElementsCount = currentChunkCursor - realIndex;
valuesSize += actualElementsCount;
valuesSize += currentChunk.cursor - currentChunk.startIndex;
if (traversedChunksDeque != null) {
traversedChunksDeque.addLast(currentChunk);
}
if (realIndex != currentChunkStartIndex || currentChunk.tailChunk == null) {
if (currentChunk.tailChunk == null) {
break;
}

currentChunk = currentChunk.tailChunk;
Copy link
Contributor

@vladimir-bukhtoyarov vladimir-bukhtoyarov Jun 13, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reading mutable fields without synchronization. trim and put do mutation of:

  • cursor
  • startIndex
  • tailChunk
    So result can be totaly wrong, even negative.

currentChunkCursor = currentChunk.cursor;
currentChunkStartIndex = currentChunk.startIndex;
}

return valuesSize;
}

public long[] values(long minKey) {
public long[] values() {
Deque<Chunk> chunksDeque = new ArrayDeque<Chunk>();
int valuesSize = traverse(minKey, chunksDeque);
int valuesSize = traverse(chunksDeque);
if (valuesSize == 0) {
return EMPTY;
}

long[] values = new long[valuesSize];
int valuesIndex = 0;
Chunk copySourceChunk = chunksDeque.removeLast();
int copyStartIndex = findFirstIndexOfGreaterEqualElements(
copySourceChunk.keys, copySourceChunk.startIndex, copySourceChunk.cursor, minKey
);

while (valuesIndex < valuesSize) {
int canBeCopiedFromChunk = copySourceChunk.cursor - copyStartIndex;
int leftToCopy = valuesSize - valuesIndex;
int length = min(canBeCopiedFromChunk, leftToCopy);
if (copyStartIndex + length != 0) {
System.arraycopy(copySourceChunk.values, copyStartIndex, values, valuesIndex, length);
}
while (!chunksDeque.isEmpty()) {
Chunk copySourceChunk = chunksDeque.removeLast();
int length = copySourceChunk.cursor - copySourceChunk.startIndex;
System.arraycopy(copySourceChunk.values, copySourceChunk.startIndex, values, valuesIndex, length);
valuesIndex += length;

if (!chunksDeque.isEmpty()) {
copySourceChunk = chunksDeque.removeLast();
copyStartIndex = copySourceChunk.startIndex;
}
}

return values;
}

public int size(long minKey) {
int valueSize = traverse(minKey, null);
public int size() {
int valueSize = traverse(null);
return valueSize;
}

String out(long minKey) {
String out() {
ArrayDeque<Chunk> chunksDeque = new ArrayDeque<Chunk>();
int valuesSize = traverse(minKey, chunksDeque);
int valuesSize = traverse(chunksDeque);
if (valuesSize == 0) {
return "[]";
}

StringBuilder builder = new StringBuilder();
int valuesIndex = 0;
Chunk copySourceChunk = chunksDeque.removeLast();
int copyStartIndex = findFirstIndexOfGreaterEqualElements(
copySourceChunk.keys, copySourceChunk.startIndex, copySourceChunk.cursor, minKey
);

while (valuesIndex < valuesSize) {
int canBeCopiedFromChunk = copySourceChunk.cursor - copyStartIndex;
int leftToCopy = valuesSize - valuesIndex;
int length = min(canBeCopiedFromChunk, leftToCopy);
if (copyStartIndex + length != 0) {
builder.append('[');
for (int i = copyStartIndex; i < copyStartIndex + length; i++) {
long key = copySourceChunk.keys[i];
long value = copySourceChunk.values[i];
builder.append('(').append(key).append(": ").append(value).append(')').append(' ');
}
builder.append(']');
while (!chunksDeque.isEmpty()) {
Chunk copySourceChunk = chunksDeque.removeLast();
builder.append('[');
for (int i = copySourceChunk.startIndex; i < copySourceChunk.cursor; i++) {
long key = copySourceChunk.keys[i];
long value = copySourceChunk.values[i];
builder.append('(').append(key).append(": ").append(value).append(')').append(' ');
}
valuesIndex += length;

builder.append(']');
if (!chunksDeque.isEmpty()) {
copySourceChunk = chunksDeque.removeLast();
copyStartIndex = copySourceChunk.startIndex;
builder.append("->");
}
}
Expand All @@ -167,25 +124,27 @@ public void trim(long startKey, long endKey) {
* |5______________________________23| :: trim(5, 23)
* [5, 9] -> [10, 13, 14, 15] -> [21] :: result layout
*/
Chunk head;
activeChunkLock.lock();
try {
Chunk head = findChunkWhereKeyShouldBe(activeChunk, endKey);
head = findChunkWhereKeyShouldBe(activeChunk, endKey);
activeChunk = head;
int newEndIndex = findFirstIndexOfGreaterEqualElements(
activeChunk.keys, activeChunk.startIndex, activeChunk.cursor, endKey
);
activeChunk.cursor = newEndIndex;

Chunk tail = findChunkWhereKeyShouldBe(head, startKey);
int newStartIndex = findFirstIndexOfGreaterEqualElements(
tail.keys, tail.startIndex, tail.cursor, startKey
);
tail.startIndex = newStartIndex;
tail.chunkSize = tail.cursor - tail.startIndex;
tail.tailChunk = null; // get rid of all forward chunks
} finally {
activeChunkLock.unlock();
}

Chunk tail = findChunkWhereKeyShouldBe(head, startKey);
int newStartIndex = findFirstIndexOfGreaterEqualElements(
tail.keys, tail.startIndex, tail.cursor, startKey
);
tail.startIndex = newStartIndex;
tail.chunkSize = tail.cursor - tail.startIndex;
tail.tailChunk = null; // get rid of all forward chunks
}

/**
Expand All @@ -201,25 +160,26 @@ public void clear(long startKey, long endKey) {
* |5______________________________23| :: clear(5, 23)
* [3, 4] -> [24, 29, 30] -> [31] :: result layout
*/
Chunk tail;
activeChunkLock.lock();
try {
Chunk tail = findChunkWhereKeyShouldBe(activeChunk, endKey);
Chunk gapStartChunk = splitChunkOnTwoSeparateChunks(tail, endKey);

// now we should skip specified gap [startKey, endKey]
// and concatenate our tail with new head four after gap
Chunk afterGapHead = findChunkWhereKeyShouldBe(gapStartChunk, startKey);
if (afterGapHead != null) {
int newEndIndex = findFirstIndexOfGreaterEqualElements(
afterGapHead.keys, afterGapHead.startIndex, afterGapHead.cursor, startKey
);
afterGapHead.cursor = newEndIndex;
afterGapHead.chunkSize = afterGapHead.cursor - afterGapHead.startIndex;
}
tail.tailChunk = afterGapHead; // concat
tail = findChunkWhereKeyShouldBe(activeChunk, endKey);
} finally {
activeChunkLock.unlock();
}

Chunk gapStartChunk = splitChunkOnTwoSeparateChunks(tail, endKey);
// now we should skip specified gap [startKey, endKey]
// and concatenate our tail with new head four after gap
Chunk afterGapHead = findChunkWhereKeyShouldBe(gapStartChunk, startKey);
if (afterGapHead != null) {
int newEndIndex = findFirstIndexOfGreaterEqualElements(
afterGapHead.keys, afterGapHead.startIndex, afterGapHead.cursor, startKey
);
afterGapHead.cursor = newEndIndex;
afterGapHead.chunkSize = afterGapHead.cursor - afterGapHead.startIndex;
}
tail.tailChunk = afterGapHead; // concat
}

private Chunk splitChunkOnTwoSeparateChunks(Chunk chunk, long key) {
Expand Down Expand Up @@ -265,7 +225,7 @@ public void clear() {
try {
activeChunk.tailChunk = null;
activeChunk.startIndex = 0;
activeChunk.chunkSize = this.chunkSize;
activeChunk.chunkSize = activeChunk.keys.length;
activeChunk.cursor = 0;
} finally {
activeChunkLock.unlock();
Expand Down
14 changes: 3 additions & 11 deletions metrics-core/src/main/java/com/codahale/metrics/SlidingTimeWindowArrayReservoir.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ public SlidingTimeWindowArrayReservoir(long window, TimeUnit windowUnit) {
*/
public SlidingTimeWindowArrayReservoir(long window, TimeUnit windowUnit, Clock clock) {
this.clock = clock;
this.measurements = new ChunkedAssociativeLongArray(512);
this.measurements = new ChunkedAssociativeLongArray();
this.window = windowUnit.toNanos(window) * COLLISION_BUFFER;
this.lastTick = new AtomicLong(clock.getTick() * COLLISION_BUFFER);
this.count = new AtomicLong();
Expand All @@ -48,9 +48,7 @@ public SlidingTimeWindowArrayReservoir(long window, TimeUnit windowUnit, Clock c
@Override
public int size() {
trim();
final long now = lastTick.get();
final long windowStart = now - window;
return measurements.size(windowStart);
return measurements.size();
}

@Override
Expand All @@ -70,9 +68,7 @@ public void update(long value) {
@Override
public Snapshot getSnapshot() {
trim();
final long now = lastTick.get();
final long windowStart = now - window;
return new UniformSnapshot(measurements.values(windowStart));
return new UniformSnapshot(measurements.values());
}

private long getTick() {
Expand All @@ -82,7 +78,6 @@ private long getTick() {
// ensure the tick is strictly incrementing even if there are duplicate ticks
final long newTick = tick - oldTick > 0 ? tick : oldTick + 1;
if (lastTick.compareAndSet(oldTick, newTick)) {
// System.out.println("arr tick: " + newTick);
return newTick;
}
}
Expand All @@ -92,9 +87,6 @@ private void trim() {
final long now = getTick();
final long windowStart = now - window;
final long windowEnd = now + CLEAR_BUFFER;
// System.out.println("arr");
// System.out.println(windowStart);
// System.out.println(windowEnd);
if (windowStart < windowEnd) {
measurements.trim(windowStart, windowEnd);
} else {
Expand Down
Loading