readded executor queue limit for submit()

submit() now tries to purge if it reaches the limit

Author: stefanosiano

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

@@ -2,6 +2,7 @@
 
 import io.sentry.util.AutoClosableReentrantLock;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -22,6 +23,12 @@ public final class SentryExecutorService implements ISentryExecutorService {
    */
   private static final int INITIAL_QUEUE_SIZE = 40;
 
+  /**
+   * 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.
+   */
+  private static final int MAX_QUEUE_SIZE = 271;
+
   private final @NotNull ScheduledThreadPoolExecutor executorService;
   private final @NotNull AutoClosableReentrantLock lock = new AutoClosableReentrantLock();
 
@@ -46,16 +53,41 @@ public SentryExecutorService() {
     this(new ScheduledThreadPoolExecutor(1, new SentryExecutorServiceThreadFactory()), null);
   }
 
+  private boolean isQueueAvailable() {
+    // If limit is reached, purge cancelled tasks from the queue
+    if (executorService.getQueue().size() >= MAX_QUEUE_SIZE) {
+      executorService.purge();
+    }
+    // Check limit again after purge
+    return executorService.getQueue().size() < MAX_QUEUE_SIZE;
+  }
+
   @Override
   public @NotNull Future<?> submit(final @NotNull Runnable runnable)
       throws RejectedExecutionException {
-    return executorService.submit(runnable);
+    if (isQueueAvailable()) {
+      return executorService.submit(runnable);
+    }
+    if (options != null) {
+      options
+        .getLogger()
+        .log(SentryLevel.WARNING, "Task " + runnable + " rejected from " + executorService);
+    }
+    return new CancelledFuture<>();
   }
 
   @Override
   public @NotNull <T> Future<T> submit(final @NotNull Callable<T> callable)
       throws RejectedExecutionException {
-    return executorService.submit(callable);
+    if (isQueueAvailable()) {
+      return executorService.submit(callable);
+    }
+    if (options != null) {
+      options
+        .getLogger()
+        .log(SentryLevel.WARNING, "Task " + callable + " rejected from " + executorService);
+    }
+    return new CancelledFuture<>();
   }
 
   @Override
@@ -127,4 +159,30 @@ private static final class SentryExecutorServiceThreadFactory implements ThreadF
       return ret;
     }
   }
+  private static final class CancelledFuture<T> implements Future<T> {
+    @Override
+    public boolean cancel(final boolean mayInterruptIfRunning) {
+      return true;
+    }
+
+    @Override
+    public boolean isCancelled() {
+      return true;
+    }
+
+    @Override
+    public boolean isDone() {
+      return true;
+    }
+
+    @Override
+    public T get() {
+      throw new CancellationException();
+    }
+
+    @Override
+    public T get(final long timeout, final @NotNull TimeUnit unit) {
+      throw new CancellationException();
+    }
+  }
 }

sentry/src/test/java/io/sentry/SentryExecutorServiceTest.kt

@@ -14,8 +14,13 @@ import org.mockito.kotlin.any
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.never
+import org.mockito.kotlin.spy
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.Callable
+import java.util.concurrent.CancellationException
+import kotlin.test.assertFailsWith
 
 class SentryExecutorServiceTest {
   @Test
@@ -105,6 +110,86 @@ class SentryExecutorServiceTest {
     whenever(executor.isShutdown).thenReturn(false)
     assertFalse(sentryExecutor.isClosed)
   }
+  @Test
+  fun `SentryExecutorService submit runnable returns cancelled future when queue size exceeds limit`() {
+    val queue = mock<BlockingQueue<Runnable>>()
+    whenever(queue.size).thenReturn(272) // Above MAX_QUEUE_SIZE (271)
+
+    val executor = mock<ScheduledThreadPoolExecutor> { on { getQueue() } doReturn queue }
+
+    val options = mock<SentryOptions>()
+    val logger = mock<ILogger>()
+    whenever(options.logger).thenReturn(logger)
+
+    val sentryExecutor = SentryExecutorService(executor, options)
+    val future = sentryExecutor.submit {}
+
+    assertTrue(future.isCancelled)
+    assertTrue(future.isDone)
+    assertFailsWith<CancellationException> { future.get() }
+    verify(executor, never()).submit(any<Runnable>())
+    verify(logger).log(any<SentryLevel>(), any<String>())
+  }
+
+  @Test
+  fun `SentryExecutorService submit runnable accepts when queue size is within limit`() {
+    val queue = mock<BlockingQueue<Runnable>>()
+    whenever(queue.size).thenReturn(270) // Below MAX_QUEUE_SIZE (271)
+
+    val executor = mock<ScheduledThreadPoolExecutor> { on { getQueue() } doReturn queue }
+
+    val sentryExecutor = SentryExecutorService(executor, null)
+    sentryExecutor.submit {}
+
+    verify(executor).submit(any<Runnable>())
+  }
+
+  @Test
+  fun `SentryExecutorService submit callable returns cancelled future when queue size exceeds limit`() {
+    val queue = mock<BlockingQueue<Runnable>>()
+    whenever(queue.size).thenReturn(272) // Above MAX_QUEUE_SIZE (271)
+
+    val executor = mock<ScheduledThreadPoolExecutor> { on { getQueue() } doReturn queue }
+
+    val options = mock<SentryOptions>()
+    val logger = mock<ILogger>()
+    whenever(options.logger).thenReturn(logger)
+
+    val sentryExecutor = SentryExecutorService(executor, options)
+    val future = sentryExecutor.submit(Callable { "result" })
+
+    assertTrue(future.isCancelled)
+    assertTrue(future.isDone)
+    assertFailsWith<CancellationException> { future.get() }
+    verify(executor, never()).submit(any<Callable<String>>())
+    verify(logger).log(any<SentryLevel>(), any<String>())
+  }
+
+  @Test
+  fun `SentryExecutorService submit callable accepts when queue size is within limit`() {
+    val queue = mock<BlockingQueue<Runnable>>()
+    whenever(queue.size).thenReturn(270) // Below MAX_QUEUE_SIZE (271)
+
+    val executor = mock<ScheduledThreadPoolExecutor> { on { getQueue() } doReturn queue }
+
+    val sentryExecutor = SentryExecutorService(executor, null)
+    sentryExecutor.submit(Callable { "result" })
+
+    verify(executor).submit(any<Callable<String>>())
+  }
+
+  @Test
+  fun `SentryExecutorService schedule accepts when queue size is within limit`() {
+    val queue = mock<BlockingQueue<Runnable>>()
+    whenever(queue.size).thenReturn(270) // Below MAX_QUEUE_SIZE (271)
+
+    val executor = mock<ScheduledThreadPoolExecutor> { on { getQueue() } doReturn queue }
+
+    val sentryExecutor = SentryExecutorService(executor, null)
+    sentryExecutor.schedule({}, 1000L)
+
+    verify(executor).schedule(any<Runnable>(), any(), any())
+  }
 
   @Test
   fun `SentryExecutorService prewarm schedules dummy tasks and clears queue`() {
@@ -122,13 +207,11 @@ class SentryExecutorServiceTest {
   }
 
   @Test
-  fun `SentryExecutorService runs any number of job`() {
-    val sentryExecutor = SentryExecutorService()
-    var called = false
-    // Post 1k jobs after 1 day, to test new jobs are accepted
+  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)) }
-    sentryExecutor.submit { called = true }
-    await.until { called }
-    assertTrue(called)
+    assertEquals(1000, executor.queue.size)
   }
 }