optimisation: do not execute swing-destined tasks in pooled thread but pass them directly to invokeLater()

platform/platform-api/src/com/intellij/util/Alarm.java

@@ -25,6 +25,7 @@ import com.intellij.openapi.util.Disposer;
 import com.intellij.openapi.util.Pair;
 import com.intellij.openapi.wm.IdeFrame;
 import com.intellij.util.concurrency.AppExecutorUtil;
+import com.intellij.util.concurrency.EdtExecutorService;
 import com.intellij.util.concurrency.QueueProcessor;
 import com.intellij.util.messages.MessageBus;
 import com.intellij.util.messages.MessageBusConnection;
@@ -68,7 +69,9 @@ public class Alarm implements Disposable {
       myDisposed = true;
       cancelAllRequests();
 
-      myExecutorService.shutdownNow();
+      if (myThreadToUse != ThreadToUse.SWING_THREAD) {
+        myExecutorService.shutdownNow();
+      }
     }
   }
 
@@ -123,7 +126,12 @@ public class Alarm implements Disposable {
   public Alarm(@NotNull ThreadToUse threadToUse, @Nullable Disposable parentDisposable) {
     myThreadToUse = threadToUse;
 
-    myExecutorService = // have to restrict the number of running tasks because otherwise the (implicit) contract of
+    myExecutorService = threadToUse == ThreadToUse.SWING_THREAD ?
+                        // pass straight to EDT
+                        EdtExecutorService.getScheduledExecutorInstance() :
+
+                        // or pass to app pooled thread.
+                        // have to restrict the number of running tasks because otherwise the (implicit) contract of
                         // "addRequests with the same delay are executed in order" will be broken
                         AppExecutorUtil.createBoundedScheduledExecutorService(1);
 

platform/platform-tests/testSrc/com/intellij/util/AlarmTest.java

@@ -1,4 +1,4 @@
-/*
+ /*
  * Copyright 2000-2016 JetBrains s.r.o.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -15,12 +15,20 @@
  */
 package com.intellij.util;
 
+import com.intellij.openapi.application.ApplicationManager;
 import com.intellij.testFramework.PlatformTestCase;
 import com.intellij.util.ui.UIUtil;
+import org.jetbrains.annotations.NotNull;
 
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 public class AlarmTest extends PlatformTestCase {
   public void testTwoAddsWithZeroDelayMustExecuteSequentially() throws Exception {
@@ -43,7 +51,7 @@ public class AlarmTest extends PlatformTestCase {
     assertRequestsExecuteSequentially(alarm);
   }
 
-  private static void assertRequestsExecuteSequentially(Alarm alarm) throws InterruptedException, ExecutionException, TimeoutException {
+  private static void assertRequestsExecuteSequentially(@NotNull Alarm alarm) throws InterruptedException, ExecutionException, TimeoutException {
     int N = 100000;
     StringBuffer log = new StringBuffer(N*4);
     StringBuilder expected = new StringBuilder(N * 4);
@@ -55,9 +63,50 @@ public class AlarmTest extends PlatformTestCase {
     for (int i = 0; i < N; i++) {
       expected.append(i).append(" ");
     }
-    alarm.waitForAllExecuted(100, TimeUnit.SECONDS);
-    UIUtil.dispatchAllInvocationEvents();
+    Future<?> future = ApplicationManager.getApplication().executeOnPooledThread(() -> {
+      try {
+        alarm.waitForAllExecuted(100, TimeUnit.SECONDS);
+      }
+      catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    });
+    while (!future.isDone()) {
+      UIUtil.dispatchAllInvocationEvents();
+    }
+    future.get();
     assertEquals(0, alarm.getActiveRequestCount());
     assertEquals(expected.toString(), log.toString());
   }
+
+  public void testOneAlarmDoesNotStartTooManyThreads() throws InterruptedException, ExecutionException, TimeoutException {
+    Alarm alarm = new Alarm(getTestRootDisposable());
+    Map<Thread, StackTraceElement[]> before = Thread.getAllStackTraces();
+    AtomicInteger executed = new AtomicInteger();
+    int N = 100000;
+    for (int i = 0; i < N; i++) {
+      alarm.addRequest(executed::incrementAndGet, 100);
+    }
+    while (executed.get() != N) {
+      UIUtil.dispatchAllInvocationEvents();
+    }
+    Map<Thread, StackTraceElement[]> after = Thread.getAllStackTraces();
+    System.out.println("before: "+before.size()+"; after: "+after.size());
+    assertTrue(Math.abs(before.size() - after.size()) < 10);
+  }
+
+  public void testManyAlarmsDoNotStartTooManyThreads() throws InterruptedException, ExecutionException, TimeoutException {
+    Map<Thread, StackTraceElement[]> before = Thread.getAllStackTraces();
+    AtomicInteger executed = new AtomicInteger();
+    int N = 100000;
+    List<Alarm> alarms = Collections.nCopies(N, "").stream().map(__ -> new Alarm(getTestRootDisposable())).collect(Collectors.toList());
+    alarms.forEach(alarm -> alarm.addRequest(executed::incrementAndGet, 100));
+
+    while (executed.get() != N) {
+      UIUtil.dispatchAllInvocationEvents();
+    }
+    Map<Thread, StackTraceElement[]> after = Thread.getAllStackTraces();
+    System.out.println("before: "+before.size()+"; after: "+after.size());
+    assertTrue(Math.abs(before.size() - after.size()) < 10);
+  }
 }

platform/util/src/com/intellij/util/concurrency/AppScheduledExecutorService.java

@@ -79,7 +79,7 @@ public class AppScheduledExecutorService extends SchedulingWrapper {
     error();
   }
 
-  private static List<Runnable> error() {
+  static List<Runnable> error() {
     throw new IncorrectOperationException("You must not call this method on the global app pool");
   }
 

platform/util/src/com/intellij/util/concurrency/EdtExecutorService.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2000-2016 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.intellij.util.concurrency;
+
+import com.intellij.util.ui.EdtInvocationManager;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * ExecutorService implementation which delegates tasks for execution to the SwingUtilities.invokeLater(task)
+ */
+public class EdtExecutorService extends AbstractExecutorService {
+  @NotNull
+  public static ExecutorService getInstance() {
+    return INSTANCE;
+  }
+
+  @NotNull
+  public static ScheduledExecutorService getScheduledExecutorInstance() {
+    return SCHEDULED_INSTANCE;
+  }
+
+  @Override
+  public void execute(@NotNull Runnable command) {
+    EdtInvocationManager.getInstance().invokeLater(command);
+  }
+
+  @Override
+  public void shutdown() {
+    AppScheduledExecutorService.error();
+  }
+
+  @NotNull
+  @Override
+  public List<Runnable> shutdownNow() {
+    return AppScheduledExecutorService.error();
+  }
+
+  @Override
+  public boolean isShutdown() {
+    return false;
+  }
+
+  @Override
+  public boolean isTerminated() {
+    return false;
+  }
+
+  @Override
+  public boolean awaitTermination(long timeout, @NotNull TimeUnit unit) throws InterruptedException {
+    AppScheduledExecutorService.error();
+    return false;
+  }
+
+  private static final ExecutorService INSTANCE = new EdtExecutorService();
+  private static final ScheduledExecutorService SCHEDULED_INSTANCE = new SchedulingWrapper(INSTANCE, ((AppScheduledExecutorService)AppExecutorUtil.getAppScheduledExecutorService()).delayQueue){
+    @NotNull
+    @Override
+    public List<Runnable> shutdownNow() {
+      return AppScheduledExecutorService.error();
+    }
+
+    @Override
+    public void shutdown() {
+      AppScheduledExecutorService.error();
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, @NotNull TimeUnit unit) throws InterruptedException {
+      AppScheduledExecutorService.error();
+      return false;
+    }
+  };
+}