package com.codahale.metrics;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public class SlidingTimeWindowReservoir implements Reservoir {
private static final int COLLISION_BUFFER = 256;
private static final int TRIM_THRESHOLD = 256;
private static final long CLEAR_BUFFER = TimeUnit.HOURS.toNanos(1) * COLLISION_BUFFER;
private final Clock clock;
private final ConcurrentSkipListMap<Long, Long> measurements;
private final long window;
private final AtomicLong lastTick;
private final AtomicLong count;
public SlidingTimeWindowReservoir(long window, TimeUnit windowUnit) {
this(window, windowUnit, Clock.defaultClock());
}
public SlidingTimeWindowReservoir(long window, TimeUnit windowUnit, Clock clock) {
this.clock = clock;
this.measurements = new ConcurrentSkipListMap<>();
this.window = windowUnit.toNanos(window) * COLLISION_BUFFER;
this.lastTick = new AtomicLong(clock.getTick() * COLLISION_BUFFER);
this.count = new AtomicLong();
}
@Override
public int size() {
trim();
return measurements.size();
}
@Override
public void update(long value) {
if (count.incrementAndGet() % TRIM_THRESHOLD == 0) {
trim();
}
measurements.put(getTick(), value);
}
@Override
public Snapshot getSnapshot() {
trim();
return new UniformSnapshot(measurements.values());
}
private long getTick() {
for ( ;; ) {
final long oldTick = lastTick.get();
final long tick = clock.getTick() * COLLISION_BUFFER;
final long newTick = tick - oldTick > 0 ? tick : oldTick + 1;
if (lastTick.compareAndSet(oldTick, newTick)) {
return newTick;
}
}
}
private void trim() {
final long now = getTick();
final long windowStart = now - window;
final long windowEnd = now + CLEAR_BUFFER;
if (windowStart < windowEnd) {
measurements.headMap(windowStart).clear();
measurements.tailMap(windowEnd).clear();
} else {
measurements.subMap(windowEnd, windowStart).clear();
}
}
}