diff --git a/sentry/src/main/java/io/sentry/SentryExecutorService.java b/sentry/src/main/java/io/sentry/SentryExecutorService.java index adb50b232e..15de0446e8 100644 --- a/sentry/src/main/java/io/sentry/SentryExecutorService.java +++ b/sentry/src/main/java/io/sentry/SentryExecutorService.java @@ -1,166 +1,207 @@ package io.sentry; -import io.sentry.util.AutoClosableReentrantLock; +import java.util.PriorityQueue; import java.util.concurrent.Callable; import java.util.concurrent.CancellationException; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import org.jetbrains.annotations.ApiStatus; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; -import org.jetbrains.annotations.TestOnly; +/** + * Custom {@link ISentryExecutorService} backed by a single daemon worker thread and a {@link + * PriorityQueue} pre-allocated to {@link #INITIAL_QUEUE_CAPACITY}. + * + *

Because the backing array is sized at construction time, no array resize ever occurs during + * normal SDK operation, and {@link #prewarm()} is a no-op. + */ @ApiStatus.Internal public final class SentryExecutorService implements ISentryExecutorService { /** - * ScheduledThreadPoolExecutor grows work queue by 50% each time. With the initial capacity of 16 - * it will have to resize 4 times to reach 40, which is a decent middle-ground for prewarming. - * This will prevent from growing in unexpected areas of the SDK. + * Pre-allocated initial capacity for the task queue. Sized to comfortably exceed the maximum + * number of tasks the SDK queues concurrently so the backing array never needs to grow at + * runtime. */ - private static final int INITIAL_QUEUE_SIZE = 40; + static final int INITIAL_QUEUE_CAPACITY = 64; /** - * By default, the work queue is unbounded so it can grow as much as the memory allows. We want to - * limit it by 271 which would be x8 times growth from the default initial capacity. + * Hard limit on the number of pending tasks. Tasks submitted beyond this limit are silently + * dropped and a cancelled {@link Future} is returned. */ private static final int MAX_QUEUE_SIZE = 271; - private final @NotNull ScheduledThreadPoolExecutor executorService; - private final @NotNull AutoClosableReentrantLock lock = new AutoClosableReentrantLock(); - - @SuppressWarnings("UnnecessaryLambda") - private final @NotNull Runnable dummyRunnable = () -> {}; - + private final @NotNull PriorityQueue> queue; + private final @NotNull Object lock = new Object(); + private final @NotNull Thread workerThread; private final @Nullable SentryOptions options; + private volatile boolean closed = false; - @TestOnly - SentryExecutorService( - final @NotNull ScheduledThreadPoolExecutor executorService, - final @Nullable SentryOptions options) { - this.executorService = executorService; - this.options = options; + public SentryExecutorService() { + this(null); } public SentryExecutorService(final @Nullable SentryOptions options) { - this(new ScheduledThreadPoolExecutor(1, new SentryExecutorServiceThreadFactory()), options); - } - - public SentryExecutorService() { - this(new ScheduledThreadPoolExecutor(1, new SentryExecutorServiceThreadFactory()), null); + this.options = options; + this.queue = new PriorityQueue<>(INITIAL_QUEUE_CAPACITY); + this.workerThread = new Thread(this::loop, "SentryExecutorService"); + this.workerThread.setDaemon(true); + this.workerThread.start(); } - private boolean isQueueAvailable() { - // If limit is reached, purge cancelled tasks from the queue - if (executorService.getQueue().size() >= MAX_QUEUE_SIZE) { - executorService.purge(); + private void loop() { + while (!closed) { + ScheduledTask task = null; + synchronized (lock) { + while (!closed) { + final ScheduledTask head = queue.peek(); + if (head == null) { + try { + lock.wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } else { + final long delayNs = head.triggerTimeNs - System.nanoTime(); + if (delayNs <= 0L) { + task = queue.poll(); + break; + } else { + // Sleep until the task is due, or until a new earlier task wakes us. + final long delayMs = Math.max(1L, delayNs / 1_000_000L); + try { + lock.wait(delayMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + } + } + } + // Execute outside the lock so producers are not blocked during task execution. + if (task != null && !task.isCancelled()) { + task.run(); + } } - // Check limit again after purge - return executorService.getQueue().size() < MAX_QUEUE_SIZE; } @Override public @NotNull Future submit(final @NotNull Runnable runnable) throws RejectedExecutionException { - if (isQueueAvailable()) { - return executorService.submit(runnable); - } - if (options != null) { - options - .getLogger() - .log(SentryLevel.WARNING, "Task " + runnable + " rejected from " + executorService); - } - return new CancelledFuture<>(); + return submit(Executors.callable(runnable, (Void) null)); } @Override public @NotNull Future submit(final @NotNull Callable callable) throws RejectedExecutionException { - if (isQueueAvailable()) { - return executorService.submit(callable); - } - if (options != null) { - options - .getLogger() - .log(SentryLevel.WARNING, "Task " + callable + " rejected from " + executorService); + synchronized (lock) { + if (closed) { + return new CancelledFuture<>(); + } + if (queue.size() >= MAX_QUEUE_SIZE) { + // Purge cancelled tasks before declaring the queue full. + queue.removeIf(ScheduledTask::isCancelled); + } + if (queue.size() >= MAX_QUEUE_SIZE) { + if (options != null) { + options + .getLogger() + .log( + SentryLevel.WARNING, + "Task " + callable + " rejected from SentryExecutorService: queue full"); + } + return new CancelledFuture<>(); + } + final ScheduledTask task = new ScheduledTask<>(callable, System.nanoTime()); + queue.offer(task); + lock.notifyAll(); + return task; } - return new CancelledFuture<>(); } @Override public @NotNull Future schedule(final @NotNull Runnable runnable, final long delayMillis) throws RejectedExecutionException { - return executorService.schedule(runnable, delayMillis, TimeUnit.MILLISECONDS); + synchronized (lock) { + if (closed) { + return new CancelledFuture<>(); + } + final ScheduledTask task = + new ScheduledTask<>( + Executors.callable(runnable, (Void) null), + System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delayMillis)); + queue.offer(task); + // Wake the worker only if this task is now the earliest — avoids spurious wakeups. + final ScheduledTask head = queue.peek(); + if (head == task) { + lock.notifyAll(); + } + return task; + } } @Override public void close(final long timeoutMillis) { - try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) { - if (!executorService.isShutdown()) { - executorService.shutdown(); - try { - if (!executorService.awaitTermination(timeoutMillis, TimeUnit.MILLISECONDS)) { - executorService.shutdownNow(); - } - } catch (InterruptedException e) { - executorService.shutdownNow(); - Thread.currentThread().interrupt(); - } + synchronized (lock) { + if (closed) { + return; } + closed = true; + queue.clear(); + lock.notifyAll(); + } + try { + workerThread.join(timeoutMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + if (workerThread.isAlive()) { + workerThread.interrupt(); } } @Override public boolean isClosed() { - try (final @NotNull ISentryLifecycleToken ignored = lock.acquire()) { - return executorService.isShutdown(); - } + return closed; } - @SuppressWarnings({"FutureReturnValueIgnored"}) + /** + * No-op. The task queue is pre-allocated at construction time ({@link #INITIAL_QUEUE_CAPACITY} + * slots), so no warm-up is required. + * + * @deprecated Pre-allocation makes this unnecessary. Will be removed in a future release. + */ @Override - public void prewarm() { - try { - executorService.submit( - () -> { - try { - // schedule a bunch of dummy runnables in the future that will never execute to - // trigger - // queue growth and then purge the queue - for (int i = 0; i < INITIAL_QUEUE_SIZE; i++) { - final Future future = - executorService.schedule(dummyRunnable, 365L, TimeUnit.DAYS); - future.cancel(true); - } - executorService.purge(); - } catch (RejectedExecutionException ignored) { - // ignore - } - }); - } catch (RejectedExecutionException e) { - if (options != null) { - options - .getLogger() - .log(SentryLevel.WARNING, "Prewarm task rejected from " + executorService, e); - } - } - } + @Deprecated + public void prewarm() {} - private static final class SentryExecutorServiceThreadFactory implements ThreadFactory { - private int cnt; + // ---- internals ---- + + private static final class ScheduledTask extends FutureTask + implements Comparable> { + + /** Absolute trigger time in nanoseconds ({@link System#nanoTime()}). */ + final long triggerTimeNs; + + ScheduledTask(final @NotNull Callable callable, final long triggerTimeNs) { + super(callable); + this.triggerTimeNs = triggerTimeNs; + } @Override - public @NotNull Thread newThread(final @NotNull Runnable r) { - final Thread ret = new Thread(r, "SentryExecutorServiceThreadFactory-" + cnt++); - ret.setDaemon(true); - return ret; + public int compareTo(final @NotNull ScheduledTask other) { + return Long.compare(this.triggerTimeNs, other.triggerTimeNs); } } private static final class CancelledFuture implements Future { + @Override public boolean cancel(final boolean mayInterruptIfRunning) { return true; diff --git a/sentry/src/test/java/io/sentry/SentryExecutorServiceTest.kt b/sentry/src/test/java/io/sentry/SentryExecutorServiceTest.kt index 2ccba650ad..76975e1c36 100644 --- a/sentry/src/test/java/io/sentry/SentryExecutorServiceTest.kt +++ b/sentry/src/test/java/io/sentry/SentryExecutorServiceTest.kt @@ -1,235 +1,267 @@ package io.sentry -import io.sentry.test.getProperty -import java.util.concurrent.BlockingQueue import java.util.concurrent.Callable import java.util.concurrent.CancellationException -import java.util.concurrent.LinkedBlockingQueue -import java.util.concurrent.ScheduledThreadPoolExecutor +import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import kotlin.test.Test -import kotlin.test.assertEquals import kotlin.test.assertFailsWith import kotlin.test.assertFalse import kotlin.test.assertTrue import org.awaitility.kotlin.await import org.mockito.kotlin.any -import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock import org.mockito.kotlin.never import org.mockito.kotlin.verify import org.mockito.kotlin.whenever class SentryExecutorServiceTest { + + // region submit(Runnable) + @Test - fun `SentryExecutorService forwards submit call to ExecutorService`() { - val executor = mock { on { queue } doReturn LinkedBlockingQueue() } - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.submit {} - verify(executor).submit(any()) + fun `executes submitted runnable`() { + val executor = SentryExecutorService() + val executed = AtomicBoolean(false) + executor.submit { executed.set(true) } + await.untilTrue(executed) + executor.close(15_000) } @Test - fun `SentryExecutorService forwards schedule call to ExecutorService`() { - val executor = mock { on { queue } doReturn LinkedBlockingQueue() } - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.schedule({}, 0L) - verify(executor).schedule(any(), any(), any()) + fun `submit runnable returns non-cancelled future`() { + val executor = SentryExecutorService() + val future = executor.submit {} + future.get(5, TimeUnit.SECONDS) + assertFalse(future.isCancelled) + executor.close(15_000) } + // endregion + + // region submit(Callable) + @Test - fun `SentryExecutorService forwards close call to ExecutorService`() { - val executor = mock() - val sentryExecutor = SentryExecutorService(executor, null) - whenever(executor.isShutdown).thenReturn(false) - whenever(executor.awaitTermination(any(), any())).thenReturn(true) - sentryExecutor.close(15000) - verify(executor).shutdown() + fun `executes submitted callable and returns result`() { + val executor = SentryExecutorService() + val future = executor.submit(Callable { 42 }) + assertTrue(future.get(5, TimeUnit.SECONDS) == 42) + executor.close(15_000) } + // endregion + + // region schedule + @Test - fun `SentryExecutorService forwards close and call shutdownNow if not enough time`() { - val executor = mock() - val sentryExecutor = SentryExecutorService(executor, null) - whenever(executor.isShutdown).thenReturn(false) - whenever(executor.awaitTermination(any(), any())).thenReturn(false) - sentryExecutor.close(15000) - verify(executor).shutdownNow() + fun `executes scheduled runnable after delay`() { + val executor = SentryExecutorService() + val executed = AtomicBoolean(false) + val startMs = System.currentTimeMillis() + executor.schedule({ executed.set(true) }, 200L) + await.untilTrue(executed) + val elapsedMs = System.currentTimeMillis() - startMs + assertTrue(elapsedMs >= 150L, "Expected >= 150ms delay, got ${elapsedMs}ms") + executor.close(15_000) } @Test - fun `SentryExecutorService forwards close and call shutdownNow if await throws`() { - val executor = mock() - val sentryExecutor = SentryExecutorService(executor, null) - whenever(executor.isShutdown).thenReturn(false) - whenever(executor.awaitTermination(any(), any())).thenThrow(InterruptedException()) - sentryExecutor.close(15000) - verify(executor).shutdownNow() + fun `schedule returns non-cancelled future`() { + val executor = SentryExecutorService() + val future = executor.schedule({}, 0L) + future.get(5, TimeUnit.SECONDS) + assertFalse(future.isCancelled) + executor.close(15_000) } + // endregion + + // region close / isClosed + @Test - fun `SentryExecutorService forwards close but do not shutdown if its already closed`() { - val executor = mock() - val sentryExecutor = SentryExecutorService(executor, null) - whenever(executor.isShutdown).thenReturn(true) - sentryExecutor.close(15000) - verify(executor, never()).shutdown() + fun `isClosed returns false before close`() { + val executor = SentryExecutorService() + assertFalse(executor.isClosed) + executor.close(15_000) } @Test - fun `SentryExecutorService forwards close call to ExecutorService and close it`() { - val executor = ScheduledThreadPoolExecutor(1) - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.close(15000) - assertTrue(executor.isShutdown) + fun `isClosed returns true after close`() { + val executor = SentryExecutorService() + executor.close(15_000) + assertTrue(executor.isClosed) } @Test - fun `SentryExecutorService executes runnable`() { - val sentryExecutor = SentryExecutorService() - val atomicBoolean = AtomicBoolean(true) - sentryExecutor.submit { atomicBoolean.set(false) } - await.untilFalse(atomicBoolean) - sentryExecutor.close(15000) + fun `close is idempotent`() { + val executor = SentryExecutorService() + executor.close(15_000) + executor.close(15_000) // second call must not throw + assertTrue(executor.isClosed) } @Test - fun `SentryExecutorService isClosed returns true if executor is shutdown`() { - val executor = mock() - val sentryExecutor = SentryExecutorService(executor, null) - whenever(executor.isShutdown).thenReturn(true) - assertTrue(sentryExecutor.isClosed) + fun `close waits for in-flight task to complete`() { + val executor = SentryExecutorService() + val started = AtomicBoolean(false) + val completed = AtomicBoolean(false) + executor.submit { + started.set(true) + Thread.sleep(200) + completed.set(true) + } + await.untilTrue(started) + executor.close(15_000) + assertTrue(completed.get()) } @Test - fun `SentryExecutorService isClosed returns false if executor is not shutdown`() { - val executor = mock() - val sentryExecutor = SentryExecutorService(executor, null) - whenever(executor.isShutdown).thenReturn(false) - assertFalse(sentryExecutor.isClosed) + fun `submit after close returns cancelled future`() { + val executor = SentryExecutorService() + executor.close(15_000) + val future = executor.submit {} + assertTrue(future.isCancelled) + assertTrue(future.isDone) + assertFailsWith { future.get() } } @Test - fun `SentryExecutorService submit runnable returns cancelled future when queue size exceeds limit`() { - val queue = mock>() - whenever(queue.size).thenReturn(272) // Above MAX_QUEUE_SIZE (271) + fun `schedule after close returns cancelled future`() { + val executor = SentryExecutorService() + executor.close(15_000) + val future = executor.schedule({}, 1_000L) + assertTrue(future.isCancelled) + assertTrue(future.isDone) + assertFailsWith { future.get() } + } + + // endregion - val executor = mock { on { getQueue() } doReturn queue } + // region queue limit + @Test + fun `submit runnable returns cancelled future when queue exceeds limit`() { val options = mock() val logger = mock() whenever(options.logger).thenReturn(logger) - val sentryExecutor = SentryExecutorService(executor, options) - val future = sentryExecutor.submit {} + val executor = SentryExecutorService(options) + // Fill queue past MAX_QUEUE_SIZE with far-future tasks + repeat(272) { executor.schedule({}, TimeUnit.DAYS.toMillis(1)) } + val future = executor.submit {} assertTrue(future.isCancelled) assertTrue(future.isDone) assertFailsWith { future.get() } - verify(executor, never()).submit(any()) verify(logger).log(any(), any()) + executor.close(100) } @Test - fun `SentryExecutorService submit runnable accepts when queue size is within limit`() { - val queue = mock>() - whenever(queue.size).thenReturn(270) // Below MAX_QUEUE_SIZE (271) - - val executor = mock { on { getQueue() } doReturn queue } - - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.submit {} - - verify(executor).submit(any()) - } - - @Test - fun `SentryExecutorService submit callable returns cancelled future when queue size exceeds limit`() { - val queue = mock>() - whenever(queue.size).thenReturn(272) // Above MAX_QUEUE_SIZE (271) - - val executor = mock { on { getQueue() } doReturn queue } - + fun `submit callable returns cancelled future when queue exceeds limit`() { val options = mock() val logger = mock() whenever(options.logger).thenReturn(logger) - val sentryExecutor = SentryExecutorService(executor, options) - val future = sentryExecutor.submit(Callable { "result" }) + val executor = SentryExecutorService(options) + repeat(272) { executor.schedule({}, TimeUnit.DAYS.toMillis(1)) } + val future = executor.submit(Callable { "result" }) assertTrue(future.isCancelled) - assertTrue(future.isDone) - assertFailsWith { future.get() } - verify(executor, never()).submit(any>()) verify(logger).log(any(), any()) + executor.close(100) } @Test - fun `SentryExecutorService submit callable accepts when queue size is within limit`() { - val queue = mock>() - whenever(queue.size).thenReturn(270) // Below MAX_QUEUE_SIZE (271) - - val executor = mock { on { getQueue() } doReturn queue } - - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.submit(Callable { "result" }) - - verify(executor).submit(any>()) + fun `submit purges cancelled tasks when queue limit is reached`() { + val executor = SentryExecutorService() + // Fill and cancel all + val futures = (1..272).map { executor.schedule({}, TimeUnit.DAYS.toMillis(1)) } + futures.forEach { it.cancel(true) } + + // Next submit should succeed after purge + val executed = AtomicBoolean(false) + val future = executor.submit { executed.set(true) } + assertFalse(future.isCancelled) + await.untilTrue(executed) + executor.close(15_000) } @Test - fun `SentryExecutorService schedule accepts when queue size is within limit`() { - val queue = mock>() - whenever(queue.size).thenReturn(270) // Below MAX_QUEUE_SIZE (271) + fun `submit logs nothing when queue size is within limit`() { + val options = mock() + val logger = mock() + whenever(options.logger).thenReturn(logger) - val executor = mock { on { getQueue() } doReturn queue } + val executor = SentryExecutorService(options) + executor.submit {} + verify(logger, never()).log(any(), any()) + executor.close(15_000) + } - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.schedule({}, 1000L) + // endregion - verify(executor).schedule(any(), any(), any()) - } + // region ordering @Test - fun `SentryExecutorService prewarm schedules dummy tasks and clears queue`() { - val executor = ScheduledThreadPoolExecutor(1) - - val sentryExecutor = SentryExecutorService(executor, null) - sentryExecutor.prewarm() + fun `tasks run in trigger-time order`() { + val executor = SentryExecutorService() + val order = mutableListOf() + val latch = CountDownLatch(3) + // Schedule out of order; single worker ensures serialised execution. + executor.schedule( + { + synchronized(order) { order.add(3) } + latch.countDown() + }, + 300L, + ) + executor.schedule( + { + synchronized(order) { order.add(1) } + latch.countDown() + }, + 100L, + ) + executor.schedule( + { + synchronized(order) { order.add(2) } + latch.countDown() + }, + 200L, + ) + latch.await(10, TimeUnit.SECONDS) + assertTrue(order == listOf(1, 2, 3), "Expected [1,2,3] but got $order") + executor.close(15_000) + } - Thread.sleep(1000) + // endregion - // the internal queue/array should be resized 4 times to 54 - assertEquals(54, (executor.queue.getProperty("queue") as Array<*>).size) - // the queue should be empty - assertEquals(0, executor.queue.size) - } + // region prewarm @Test - fun `SentryExecutorService schedules any number of job`() { - val executor = ScheduledThreadPoolExecutor(1) - val sentryExecutor = SentryExecutorService(executor, null) - // Post 1k jobs after 1 day, to test they are all accepted - repeat(1000) { sentryExecutor.schedule({}, TimeUnit.DAYS.toMillis(1)) } - assertEquals(1000, executor.queue.size) + fun `prewarm is a no-op`() { + val executor = SentryExecutorService() + executor.prewarm() // must not throw or block + executor.close(15_000) } - @Test - fun `SentryExecutorService purges cancelled jobs when limit is reached`() { - val executor = ScheduledThreadPoolExecutor(1) - val sentryExecutor = SentryExecutorService(executor, null) - // Post 1k jobs after 1 day, to test they are all accepted - repeat(1000) { - val future = sentryExecutor.schedule({}, TimeUnit.DAYS.toMillis(1)) - future.cancel(true) - } - assertEquals(1000, executor.queue.size) + // endregion + + // region initial capacity - // Submit should purge cancelled scheduled jobs - sentryExecutor.submit {} - // The queue size should be 1, but if the executor thread runs right before we check the size, - // it returns 0 - assertTrue(executor.queue.size < 2) + @Test + fun `initial queue capacity constant is in expected range`() { + assertTrue( + SentryExecutorService.INITIAL_QUEUE_CAPACITY >= 32, + "INITIAL_QUEUE_CAPACITY should be at least 32", + ) + assertTrue( + SentryExecutorService.INITIAL_QUEUE_CAPACITY <= 128, + "INITIAL_QUEUE_CAPACITY should be at most 128", + ) } + + // endregion }