Skip to content

Commit d9a1682

Browse files
committed
feat(core): Replace SentryExecutorService with pre-allocated queue implementation
ScheduledThreadPoolExecutor's internal DelayedWorkQueue is a heap that resizes dynamically (50% growth from initial capacity 16). prewarm() was introduced to pre-grow that array during init, but doing so on the main thread is itself the worst possible time to trigger allocations — and the queue resize cost is only ~8µs anyway. This replaces the whole approach: a custom executor backed by a PriorityQueue pre-allocated to INITIAL_QUEUE_CAPACITY=64 at construction time. The backing array never resizes during normal SDK operation. A single daemon worker thread uses Object.wait/notifyAll for precise wakeup on scheduled tasks. prewarm() becomes a documented no-op. Key properties: - No array resize at runtime: queue pre-allocated at construction - Precise scheduling: worker sleeps until next task triggerTime, wakes immediately when an earlier task is enqueued - MAX_QUEUE_SIZE (271) and purge-on-overflow semantics preserved - ScheduledTask<T> extends FutureTask<T> for free Future<T> contract - Drops @testonly ScheduledThreadPoolExecutor constructor (nothing to inject) Refs #5681
1 parent 2ad4e0d commit d9a1682

1 file changed

Lines changed: 139 additions & 98 deletions

File tree

sentry/src/main/java/io/sentry/SentryExecutorService.java

Lines changed: 139 additions & 98 deletions
Original file line numberDiff line numberDiff line change
@@ -1,166 +1,207 @@
11
package io.sentry;
22

3-
import io.sentry.util.AutoClosableReentrantLock;
3+
import java.util.PriorityQueue;
44
import java.util.concurrent.Callable;
55
import java.util.concurrent.CancellationException;
6+
import java.util.concurrent.Executors;
67
import java.util.concurrent.Future;
8+
import java.util.concurrent.FutureTask;
79
import java.util.concurrent.RejectedExecutionException;
8-
import java.util.concurrent.ScheduledThreadPoolExecutor;
9-
import java.util.concurrent.ThreadFactory;
1010
import java.util.concurrent.TimeUnit;
1111
import org.jetbrains.annotations.ApiStatus;
1212
import org.jetbrains.annotations.NotNull;
1313
import org.jetbrains.annotations.Nullable;
14-
import org.jetbrains.annotations.TestOnly;
1514

15+
/**
16+
* Custom {@link ISentryExecutorService} backed by a single daemon worker thread and a {@link
17+
* PriorityQueue} pre-allocated to {@link #INITIAL_QUEUE_CAPACITY}.
18+
*
19+
* <p>Because the backing array is sized at construction time, no array resize ever occurs during
20+
* normal SDK operation, and {@link #prewarm()} is a no-op.
21+
*/
1622
@ApiStatus.Internal
1723
public final class SentryExecutorService implements ISentryExecutorService {
1824

1925
/**
20-
* ScheduledThreadPoolExecutor grows work queue by 50% each time. With the initial capacity of 16
21-
* it will have to resize 4 times to reach 40, which is a decent middle-ground for prewarming.
22-
* This will prevent from growing in unexpected areas of the SDK.
26+
* Pre-allocated initial capacity for the task queue. Sized to comfortably exceed the maximum
27+
* number of tasks the SDK queues concurrently so the backing array never needs to grow at
28+
* runtime.
2329
*/
24-
private static final int INITIAL_QUEUE_SIZE = 40;
30+
static final int INITIAL_QUEUE_CAPACITY = 64;
2531

2632
/**
27-
* By default, the work queue is unbounded so it can grow as much as the memory allows. We want to
28-
* limit it by 271 which would be x8 times growth from the default initial capacity.
33+
* Hard limit on the number of pending tasks. Tasks submitted beyond this limit are silently
34+
* dropped and a cancelled {@link Future} is returned.
2935
*/
3036
private static final int MAX_QUEUE_SIZE = 271;
3137

32-
private final @NotNull ScheduledThreadPoolExecutor executorService;
33-
private final @NotNull AutoClosableReentrantLock lock = new AutoClosableReentrantLock();
34-
35-
@SuppressWarnings("UnnecessaryLambda")
36-
private final @NotNull Runnable dummyRunnable = () -> {};
37-
38+
private final @NotNull PriorityQueue<ScheduledTask<?>> queue;
39+
private final @NotNull Object lock = new Object();
40+
private final @NotNull Thread workerThread;
3841
private final @Nullable SentryOptions options;
42+
private volatile boolean closed = false;
3943

40-
@TestOnly
41-
SentryExecutorService(
42-
final @NotNull ScheduledThreadPoolExecutor executorService,
43-
final @Nullable SentryOptions options) {
44-
this.executorService = executorService;
45-
this.options = options;
44+
public SentryExecutorService() {
45+
this(null);
4646
}
4747

4848
public SentryExecutorService(final @Nullable SentryOptions options) {
49-
this(new ScheduledThreadPoolExecutor(1, new SentryExecutorServiceThreadFactory()), options);
50-
}
51-
52-
public SentryExecutorService() {
53-
this(new ScheduledThreadPoolExecutor(1, new SentryExecutorServiceThreadFactory()), null);
49+
this.options = options;
50+
this.queue = new PriorityQueue<>(INITIAL_QUEUE_CAPACITY);
51+
this.workerThread = new Thread(this::loop, "SentryExecutorService");
52+
this.workerThread.setDaemon(true);
53+
this.workerThread.start();
5454
}
5555

56-
private boolean isQueueAvailable() {
57-
// If limit is reached, purge cancelled tasks from the queue
58-
if (executorService.getQueue().size() >= MAX_QUEUE_SIZE) {
59-
executorService.purge();
56+
private void loop() {
57+
while (!closed) {
58+
ScheduledTask<?> task = null;
59+
synchronized (lock) {
60+
while (!closed) {
61+
final ScheduledTask<?> head = queue.peek();
62+
if (head == null) {
63+
try {
64+
lock.wait();
65+
} catch (InterruptedException e) {
66+
Thread.currentThread().interrupt();
67+
return;
68+
}
69+
} else {
70+
final long delayNs = head.triggerTimeNs - System.nanoTime();
71+
if (delayNs <= 0L) {
72+
task = queue.poll();
73+
break;
74+
} else {
75+
// Sleep until the task is due, or until a new earlier task wakes us.
76+
final long delayMs = Math.max(1L, delayNs / 1_000_000L);
77+
try {
78+
lock.wait(delayMs);
79+
} catch (InterruptedException e) {
80+
Thread.currentThread().interrupt();
81+
return;
82+
}
83+
}
84+
}
85+
}
86+
}
87+
// Execute outside the lock so producers are not blocked during task execution.
88+
if (task != null && !task.isCancelled()) {
89+
task.run();
90+
}
6091
}
61-
// Check limit again after purge
62-
return executorService.getQueue().size() < MAX_QUEUE_SIZE;
6392
}
6493

6594
@Override
6695
public @NotNull Future<?> submit(final @NotNull Runnable runnable)
6796
throws RejectedExecutionException {
68-
if (isQueueAvailable()) {
69-
return executorService.submit(runnable);
70-
}
71-
if (options != null) {
72-
options
73-
.getLogger()
74-
.log(SentryLevel.WARNING, "Task " + runnable + " rejected from " + executorService);
75-
}
76-
return new CancelledFuture<>();
97+
return submit(Executors.callable(runnable, (Void) null));
7798
}
7899

79100
@Override
80101
public @NotNull <T> Future<T> submit(final @NotNull Callable<T> callable)
81102
throws RejectedExecutionException {
82-
if (isQueueAvailable()) {
83-
return executorService.submit(callable);
84-
}
85-
if (options != null) {
86-
options
87-
.getLogger()
88-
.log(SentryLevel.WARNING, "Task " + callable + " rejected from " + executorService);
103+
synchronized (lock) {
104+
if (closed) {
105+
return new CancelledFuture<>();
106+
}
107+
if (queue.size() >= MAX_QUEUE_SIZE) {
108+
// Purge cancelled tasks before declaring the queue full.
109+
queue.removeIf(ScheduledTask::isCancelled);
110+
}
111+
if (queue.size() >= MAX_QUEUE_SIZE) {
112+
if (options != null) {
113+
options
114+
.getLogger()
115+
.log(
116+
SentryLevel.WARNING,
117+
"Task " + callable + " rejected from SentryExecutorService: queue full");
118+
}
119+
return new CancelledFuture<>();
120+
}
121+
final ScheduledTask<T> task = new ScheduledTask<>(callable, System.nanoTime());
122+
queue.offer(task);
123+
lock.notifyAll();
124+
return task;
89125
}
90-
return new CancelledFuture<>();
91126
}
92127

93128
@Override
94129
public @NotNull Future<?> schedule(final @NotNull Runnable runnable, final long delayMillis)
95130
throws RejectedExecutionException {
96-
return executorService.schedule(runnable, delayMillis, TimeUnit.MILLISECONDS);
131+
synchronized (lock) {
132+
if (closed) {
133+
return new CancelledFuture<>();
134+
}
135+
final ScheduledTask<?> task =
136+
new ScheduledTask<>(
137+
Executors.callable(runnable, (Void) null),
138+
System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delayMillis));
139+
queue.offer(task);
140+
// Wake the worker only if this task is now the earliest — avoids spurious wakeups.
141+
final ScheduledTask<?> head = queue.peek();
142+
if (head == task) {
143+
lock.notifyAll();
144+
}
145+
return task;
146+
}
97147
}
98148

99149
@Override
100150
public void close(final long timeoutMillis) {
101-
try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
102-
if (!executorService.isShutdown()) {
103-
executorService.shutdown();
104-
try {
105-
if (!executorService.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS)) {
106-
executorService.shutdownNow();
107-
}
108-
} catch (InterruptedException e) {
109-
executorService.shutdownNow();
110-
Thread.currentThread().interrupt();
111-
}
151+
synchronized (lock) {
152+
if (closed) {
153+
return;
112154
}
155+
closed = true;
156+
queue.clear();
157+
lock.notifyAll();
158+
}
159+
try {
160+
workerThread.join(timeoutMillis);
161+
} catch (InterruptedException e) {
162+
Thread.currentThread().interrupt();
163+
}
164+
if (workerThread.isAlive()) {
165+
workerThread.interrupt();
113166
}
114167
}
115168

116169
@Override
117170
public boolean isClosed() {
118-
try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) {
119-
return executorService.isShutdown();
120-
}
171+
return closed;
121172
}
122173

123-
@SuppressWarnings({"FutureReturnValueIgnored"})
174+
/**
175+
* No-op. The task queue is pre-allocated at construction time ({@link #INITIAL_QUEUE_CAPACITY}
176+
* slots), so no warm-up is required.
177+
*
178+
* @deprecated Pre-allocation makes this unnecessary. Will be removed in a future release.
179+
*/
124180
@Override
125-
public void prewarm() {
126-
try {
127-
executorService.submit(
128-
() -> {
129-
try {
130-
// schedule a bunch of dummy runnables in the future that will never execute to
131-
// trigger
132-
// queue growth and then purge the queue
133-
for (int i = 0; i < INITIAL_QUEUE_SIZE; i++) {
134-
final Future<?> future =
135-
executorService.schedule(dummyRunnable, 365L, TimeUnit.DAYS);
136-
future.cancel(true);
137-
}
138-
executorService.purge();
139-
} catch (RejectedExecutionException ignored) {
140-
// ignore
141-
}
142-
});
143-
} catch (RejectedExecutionException e) {
144-
if (options != null) {
145-
options
146-
.getLogger()
147-
.log(SentryLevel.WARNING, "Prewarm task rejected from " + executorService, e);
148-
}
149-
}
150-
}
181+
@Deprecated
182+
public void prewarm() {}
151183

152-
private static final class SentryExecutorServiceThreadFactory implements ThreadFactory {
153-
private int cnt;
184+
// ---- internals ----
185+
186+
private static final class ScheduledTask<T> extends FutureTask<T>
187+
implements Comparable<ScheduledTask<?>> {
188+
189+
/** Absolute trigger time in nanoseconds ({@link System#nanoTime()}). */
190+
final long triggerTimeNs;
191+
192+
ScheduledTask(final @NotNull Callable<T> callable, final long triggerTimeNs) {
193+
super(callable);
194+
this.triggerTimeNs = triggerTimeNs;
195+
}
154196

155197
@Override
156-
public @NotNull Thread newThread(final @NotNull Runnable r) {
157-
final Thread ret = new Thread(r, "SentryExecutorServiceThreadFactory-" + cnt++);
158-
ret.setDaemon(true);
159-
return ret;
198+
public int compareTo(final @NotNull ScheduledTask<?> other) {
199+
return Long.compare(this.triggerTimeNs, other.triggerTimeNs);
160200
}
161201
}
162202

163203
private static final class CancelledFuture<T> implements Future<T> {
204+
164205
@Override
165206
public boolean cancel(final boolean mayInterruptIfRunning) {
166207
return true;

0 commit comments

Comments
 (0)