diff --git a/platform/platform-api/src/com/intellij/util/Alarm.java b/platform/platform-api/src/com/intellij/util/Alarm.java index e8215fe5d190..90fc638b0463 100644 --- a/platform/platform-api/src/com/intellij/util/Alarm.java +++ b/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); diff --git a/platform/platform-tests/testSrc/com/intellij/util/AlarmTest.java b/platform/platform-tests/testSrc/com/intellij/util/AlarmTest.java index 10106c38cbd9..b2f8f747cbd3 100644 --- a/platform/platform-tests/testSrc/com/intellij/util/AlarmTest.java +++ b/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 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 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 before = Thread.getAllStackTraces(); + AtomicInteger executed = new AtomicInteger(); + int N = 100000; + List 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 after = Thread.getAllStackTraces(); + System.out.println("before: "+before.size()+"; after: "+after.size()); + assertTrue(Math.abs(before.size() - after.size()) < 10); + } } \ No newline at end of file diff --git a/platform/util/src/com/intellij/util/concurrency/AppScheduledExecutorService.java b/platform/util/src/com/intellij/util/concurrency/AppScheduledExecutorService.java index d1c309d9f7ca..81e03f3fd5b9 100644 --- a/platform/util/src/com/intellij/util/concurrency/AppScheduledExecutorService.java +++ b/platform/util/src/com/intellij/util/concurrency/AppScheduledExecutorService.java @@ -79,7 +79,7 @@ public class AppScheduledExecutorService extends SchedulingWrapper { error(); } - private static List error() { + static List error() { throw new IncorrectOperationException("You must not call this method on the global app pool"); } diff --git a/platform/util/src/com/intellij/util/concurrency/EdtExecutorService.java b/platform/util/src/com/intellij/util/concurrency/EdtExecutorService.java new file mode 100644 index 000000000000..61c81e4c10e7 --- /dev/null +++ b/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 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 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; + } + }; +}