Skip to content

Commit 7fa11d4

Browse files
ben-manesben-manes
committed
Fix timer wheel cascading and support cancelating a scheduled task
When the timer wheel advances then higher level buckets evict to the lower wheels. In some cases this reorganization failed, such as when the nano time transitioned from a negative to a postive clock value. The number of steps to turn the wheel is now determined by a more accurate calculation. (fixes #541) When a scheduler is enabled then a future task may exist against the next expiration event to perform cache maintenance. If the cache becomes empty beforehand, e.g. Map.clear(), then this future can be canceled. (fixes #542)
1 parent d3488a7 commit 7fa11d4

7 files changed

Lines changed: 262 additions & 97 deletions

File tree

‎caffeine/src/main/java/com/github/benmanes/caffeine/cache/BoundedLocalCache.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -842,7 +842,9 @@ void expireEntries() {
842842
Pacer pacer = pacer();
843843
if (pacer != null) {
844844
long delay = getExpirationDelay(now);
845-
if (delay != Long.MAX_VALUE) {
845+
if (delay == Long.MAX_VALUE) {
846+
pacer.cancel();
847+
} else {
846848
pacer.schedule(executor, drainBuffersTask, now, delay);
847849
}
848850
}
@@ -1883,6 +1885,12 @@ public void clear() {
18831885
removeNode(entry.getValue(), now);
18841886
}
18851887

1888+
// Cancel the scheduled cleanup
1889+
Pacer pacer = pacer();
1890+
if (pacer != null) {
1891+
pacer.cancel();
1892+
}
1893+
18861894
// Discard all pending reads
18871895
readBuffer.drainTo(e -> {});
18881896
} finally {

‎caffeine/src/main/java/com/github/benmanes/caffeine/cache/LocalAsyncCache.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -567,7 +567,7 @@ public void invalidateAll() {
567567

568568
@Override
569569
public long estimatedSize() {
570-
return asyncCache().cache().size();
570+
return asyncCache().cache().estimatedSize();
571571
}
572572

573573
@Override

‎caffeine/src/main/java/com/github/benmanes/caffeine/cache/Pacer.java‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,10 +49,10 @@ public void schedule(Executor executor, Runnable command, long now, long delay)
4949

5050
if (future == null) {
5151
// short-circuit an immediate scheduler causing an infinite loop during initialization
52-
if (nextFireTime != 0) {
52+
if (nextFireTime != 0L) {
5353
return;
5454
}
55-
} else if ((nextFireTime - now) > 0) {
55+
} else if ((nextFireTime - now) > 0L) {
5656
// Determine whether to reschedule
5757
if (maySkip(scheduleAt)) {
5858
return;
@@ -63,12 +63,21 @@ public void schedule(Executor executor, Runnable command, long now, long delay)
6363
future = scheduler.schedule(executor, command, actualDelay, TimeUnit.NANOSECONDS);
6464
}
6565

66+
/** Attempts to cancel execution of the scheduled task, if present. */
67+
public void cancel() {
68+
if (future != null) {
69+
future.cancel(/* mayInterruptIfRunning */ false);
70+
nextFireTime = 0L;
71+
future = null;
72+
}
73+
}
74+
6675
/**
6776
* Returns if the current fire time is sooner, or if it is later and within the tolerance limit.
6877
*/
6978
boolean maySkip(long scheduleAt) {
7079
long delta = (scheduleAt - nextFireTime);
71-
return (delta >= 0) || (-delta <= TOLERANCE);
80+
return (delta >= 0L) || (-delta <= TOLERANCE);
7281
}
7382

7483
/** Returns the delay and sets the next fire time. */

‎caffeine/src/main/java/com/github/benmanes/caffeine/cache/TimerWheel.java‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,13 @@ public void advance(long currentTimeNanos) {
9797
long previousTimeNanos = nanos;
9898
try {
9999
nanos = currentTimeNanos;
100+
101+
// If wrapping, temporarily shift the clock for a positive comparison
102+
if ((previousTimeNanos < 0) && (currentTimeNanos > 0)) {
103+
previousTimeNanos += Long.MAX_VALUE;
104+
currentTimeNanos += Long.MAX_VALUE;
105+
}
106+
100107
for (int i = 0; i < SHIFT.length; i++) {
101108
long previousTicks = (previousTimeNanos >>> SHIFT[i]);
102109
long currentTicks = (currentTimeNanos >>> SHIFT[i]);
@@ -120,20 +127,14 @@ public void advance(long currentTimeNanos) {
120127
*/
121128
void expire(int index, long previousTicks, long currentTicks) {
122129
Node<K, V>[] timerWheel = wheel[index];
130+
int mask = timerWheel.length - 1;
123131

124-
int start, end;
125-
if ((currentTicks - previousTicks) >= timerWheel.length) {
126-
end = timerWheel.length;
127-
start = 0;
128-
} else {
129-
long mask = SPANS[index] - 1;
130-
start = (int) (previousTicks & mask);
131-
end = 1 + (int) (currentTicks & mask);
132-
}
132+
int steps = Math.min(1 + Math.abs((int) (currentTicks - previousTicks)), timerWheel.length);
133+
int start = (int) (previousTicks & mask);
134+
int end = start + steps;
133135

134-
int mask = timerWheel.length - 1;
135136
for (int i = start; i < end; i++) {
136-
Node<K, V> sentinel = timerWheel[(i & mask)];
137+
Node<K, V> sentinel = timerWheel[i & mask];
137138
Node<K, V> prev = sentinel.getPreviousInVariableOrder();
138139
Node<K, V> node = sentinel.getNextInVariableOrder();
139140
sentinel.setPreviousInVariableOrder(sentinel);

‎caffeine/src/test/java/com/github/benmanes/caffeine/cache/BoundedLocalCacheTest.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,9 @@
2121
import static com.github.benmanes.caffeine.cache.BLCHeader.DrainStatusRef.REQUIRED;
2222
import static com.github.benmanes.caffeine.cache.BoundedLocalCache.EXPIRE_WRITE_TOLERANCE;
2323
import static com.github.benmanes.caffeine.cache.BoundedLocalCache.PERCENT_MAIN_PROTECTED;
24+
import static com.github.benmanes.caffeine.cache.testing.CacheSpec.Expiration.AFTER_ACCESS;
25+
import static com.github.benmanes.caffeine.cache.testing.CacheSpec.Expiration.AFTER_WRITE;
26+
import static com.github.benmanes.caffeine.cache.testing.CacheSpec.Expiration.VARIABLE;
2427
import static com.github.benmanes.caffeine.cache.testing.RemovalListenerVerifier.verifyRemovalListener;
2528
import static com.github.benmanes.caffeine.cache.testing.StatsVerifier.verifyStats;
2629
import static com.github.benmanes.caffeine.testing.Awaits.await;
@@ -38,11 +41,14 @@
3841
import static org.hamcrest.Matchers.nullValue;
3942
import static org.mockito.ArgumentMatchers.any;
4043
import static org.mockito.ArgumentMatchers.anyLong;
44+
import static org.mockito.Mockito.doReturn;
45+
import static org.mockito.Mockito.verify;
4146
import static org.mockito.Mockito.when;
4247

4348
import java.util.List;
4449
import java.util.Map;
4550
import java.util.concurrent.Executor;
51+
import java.util.concurrent.Future;
4652
import java.util.concurrent.TimeUnit;
4753
import java.util.concurrent.atomic.AtomicBoolean;
4854
import java.util.concurrent.atomic.AtomicInteger;
@@ -60,6 +66,7 @@
6066
import com.github.benmanes.caffeine.cache.testing.CacheSpec;
6167
import com.github.benmanes.caffeine.cache.testing.CacheSpec.CacheExecutor;
6268
import com.github.benmanes.caffeine.cache.testing.CacheSpec.CacheExpiry;
69+
import com.github.benmanes.caffeine.cache.testing.CacheSpec.CacheScheduler;
6370
import com.github.benmanes.caffeine.cache.testing.CacheSpec.CacheWeigher;
6471
import com.github.benmanes.caffeine.cache.testing.CacheSpec.Compute;
6572
import com.github.benmanes.caffeine.cache.testing.CacheSpec.ExecutorFailure;
@@ -810,4 +817,54 @@ public void put_expireTolerance_expiry(Cache<Integer, Integer> cache, CacheConte
810817
assertThat(localCache.readBuffer.writes(), is(1L));
811818
assertThat(localCache.writeBuffer().producerIndex, is(8L));
812819
}
820+
821+
@Test(dataProvider = "caches")
822+
@CacheSpec(compute = Compute.SYNC, implementation = Implementation.Caffeine,
823+
population = Population.EMPTY, scheduler = CacheScheduler.MOCKITO,
824+
mustExpireWithAnyOf = { AFTER_ACCESS, AFTER_WRITE, VARIABLE },
825+
expiry = { CacheExpiry.DISABLED, CacheExpiry.CREATE, CacheExpiry.WRITE, CacheExpiry.ACCESS },
826+
expireAfterAccess = {Expire.DISABLED, Expire.ONE_MINUTE},
827+
expireAfterWrite = {Expire.DISABLED, Expire.ONE_MINUTE}, expiryTime = Expire.ONE_MINUTE)
828+
public void unschedule_cleanUp(Cache<Integer, Integer> cache, CacheContext context) {
829+
var future = Mockito.mock(Future.class);
830+
var localCache = asBoundedLocalCache(cache);
831+
doReturn(future).when(context.scheduler()).schedule(any(), any(), anyLong(), any());
832+
833+
for (int i = 0; i < 10; i++) {
834+
cache.put(i, -i);
835+
}
836+
assertThat(localCache.pacer().nextFireTime, is(not(0L)));
837+
assertThat(localCache.pacer().future, is(not(nullValue())));
838+
839+
context.ticker().advance(1, TimeUnit.HOURS);
840+
cache.cleanUp();
841+
842+
verify(future).cancel(false);
843+
assertThat(localCache.pacer().nextFireTime, is(0L));
844+
assertThat(localCache.pacer().future, is(nullValue()));
845+
}
846+
847+
@Test(dataProvider = "caches")
848+
@CacheSpec(compute = Compute.SYNC, implementation = Implementation.Caffeine,
849+
population = Population.EMPTY, scheduler = CacheScheduler.MOCKITO,
850+
mustExpireWithAnyOf = { AFTER_ACCESS, AFTER_WRITE, VARIABLE },
851+
expiry = { CacheExpiry.DISABLED, CacheExpiry.CREATE, CacheExpiry.WRITE, CacheExpiry.ACCESS },
852+
expireAfterAccess = {Expire.DISABLED, Expire.ONE_MINUTE},
853+
expireAfterWrite = {Expire.DISABLED, Expire.ONE_MINUTE}, expiryTime = Expire.ONE_MINUTE)
854+
public void unschedule_invalidateAll(Cache<Integer, Integer> cache, CacheContext context) {
855+
var future = Mockito.mock(Future.class);
856+
var localCache = asBoundedLocalCache(cache);
857+
doReturn(future).when(context.scheduler()).schedule(any(), any(), anyLong(), any());
858+
859+
for (int i = 0; i < 10; i++) {
860+
cache.put(i, -i);
861+
}
862+
assertThat(localCache.pacer().nextFireTime, is(not(0L)));
863+
assertThat(localCache.pacer().future, is(not(nullValue())));
864+
865+
cache.invalidateAll();
866+
verify(future).cancel(false);
867+
assertThat(localCache.pacer().nextFireTime, is(0L));
868+
assertThat(localCache.pacer().future, is(nullValue()));
869+
}
813870
}

‎caffeine/src/test/java/com/github/benmanes/caffeine/cache/PacerTest.java‎

Lines changed: 74 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,9 @@
1717

1818
import static org.hamcrest.MatcherAssert.assertThat;
1919
import static org.hamcrest.Matchers.is;
20-
import static org.mockito.ArgumentMatchers.anyBoolean;
20+
import static org.hamcrest.Matchers.not;
21+
import static org.hamcrest.Matchers.nullValue;
22+
import static org.mockito.Mockito.doAnswer;
2123
import static org.mockito.Mockito.doReturn;
2224
import static org.mockito.Mockito.verify;
2325
import static org.mockito.Mockito.verifyNoInteractions;
@@ -65,7 +67,54 @@ public void afterMethod() throws Exception {
6567
}
6668

6769
@Test
68-
public void scheduledAfterNextFireTime_skip() {
70+
public void schedule_initialize() {
71+
long delay = random.nextInt(Ints.saturatedCast(Pacer.TOLERANCE));
72+
doReturn(DisabledFuture.INSTANCE)
73+
.when(scheduler).schedule(executor, command, Pacer.TOLERANCE, TimeUnit.NANOSECONDS);
74+
pacer.schedule(executor, command, NOW, delay);
75+
76+
assertThat(pacer.future, is(DisabledFuture.INSTANCE));
77+
assertThat(pacer.nextFireTime, is(NOW + Pacer.TOLERANCE));
78+
}
79+
80+
@Test
81+
public void schedule_initialize_recurse() {
82+
long delay = random.nextInt(Ints.saturatedCast(Pacer.TOLERANCE));
83+
doAnswer(invocation -> {
84+
assertThat(pacer.future, is(nullValue()));
85+
assertThat(pacer.nextFireTime, is(not(0L)));
86+
pacer.schedule(executor, command, NOW, delay);
87+
return DisabledFuture.INSTANCE;
88+
}).when(scheduler).schedule(executor, command, Pacer.TOLERANCE, TimeUnit.NANOSECONDS);
89+
90+
pacer.schedule(executor, command, NOW, delay);
91+
assertThat(pacer.future, is(DisabledFuture.INSTANCE));
92+
assertThat(pacer.nextFireTime, is(NOW + Pacer.TOLERANCE));
93+
}
94+
95+
@Test
96+
public void schedule_cancel_schedule() {
97+
long fireTime = NOW + Pacer.TOLERANCE;
98+
long delay = random.nextInt(Ints.saturatedCast(Pacer.TOLERANCE));
99+
doReturn(future)
100+
.when(scheduler).schedule(executor, command, Pacer.TOLERANCE, TimeUnit.NANOSECONDS);
101+
102+
pacer.schedule(executor, command, NOW, delay);
103+
assertThat(pacer.nextFireTime, is(fireTime));
104+
assertThat(pacer.future, is(future));
105+
106+
pacer.cancel();
107+
verify(future).cancel(false);
108+
assertThat(pacer.nextFireTime, is(0L));
109+
assertThat(pacer.future, is(nullValue()));
110+
111+
pacer.schedule(executor, command, NOW, delay);
112+
assertThat(pacer.nextFireTime, is(fireTime));
113+
assertThat(pacer.future, is(future));
114+
}
115+
116+
@Test
117+
public void scheduled_afterNextFireTime_skip() {
69118
pacer.nextFireTime = NOW + ONE_MINUTE_IN_NANOS;
70119
pacer.future = future;
71120

@@ -78,7 +127,7 @@ public void scheduledAfterNextFireTime_skip() {
78127
}
79128

80129
@Test
81-
public void scheduledBeforeNextFireTime_skip() {
130+
public void schedule_beforeNextFireTime_skip() {
82131
pacer.nextFireTime = NOW + ONE_MINUTE_IN_NANOS;
83132
pacer.future = future;
84133

@@ -93,7 +142,7 @@ public void scheduledBeforeNextFireTime_skip() {
93142
}
94143

95144
@Test
96-
public void scheduledBeforeNextFireTime_minimumDelay() {
145+
public void schedule_beforeNextFireTime_minimumDelay() {
97146
pacer.nextFireTime = NOW + ONE_MINUTE_IN_NANOS;
98147
pacer.future = future;
99148

@@ -105,15 +154,15 @@ public void scheduledBeforeNextFireTime_minimumDelay() {
105154
assertThat(pacer.future, is(DisabledFuture.INSTANCE));
106155
assertThat(pacer.nextFireTime, is(NOW + Pacer.TOLERANCE));
107156

108-
verify(future).cancel(anyBoolean());
157+
verify(future).cancel(false);
109158
verify(scheduler).schedule(executor, command, Pacer.TOLERANCE, TimeUnit.NANOSECONDS);
110159

111160
verifyNoInteractions(executor, command);
112161
verifyNoMoreInteractions(scheduler, future);
113162
}
114163

115164
@Test
116-
public void scheduledBeforeNextFireTime_customDelay() {
165+
public void schedule_beforeNextFireTime_customDelay() {
117166
pacer.nextFireTime = NOW + ONE_MINUTE_IN_NANOS;
118167
pacer.future = future;
119168

@@ -125,10 +174,28 @@ public void scheduledBeforeNextFireTime_customDelay() {
125174
assertThat(pacer.future, is(DisabledFuture.INSTANCE));
126175
assertThat(pacer.nextFireTime, is(NOW + delay));
127176

128-
verify(future).cancel(anyBoolean());
177+
verify(future).cancel(false);
129178
verify(scheduler).schedule(executor, command, delay, TimeUnit.NANOSECONDS);
130179

131180
verifyNoInteractions(executor, command);
132181
verifyNoMoreInteractions(scheduler, future);
133182
}
183+
184+
@Test
185+
public void cancel_initialize() {
186+
pacer.cancel();
187+
assertThat(pacer.nextFireTime, is(0L));
188+
assertThat(pacer.future, is(nullValue()));
189+
}
190+
191+
@Test
192+
public void cancel_scheduled() {
193+
pacer.nextFireTime = NOW + ONE_MINUTE_IN_NANOS;
194+
pacer.future = future;
195+
196+
pacer.cancel();
197+
verify(future).cancel(false);
198+
assertThat(pacer.nextFireTime, is(0L));
199+
assertThat(pacer.future, is(nullValue()));
200+
}
134201
}

0 commit comments

Comments
 (0)