diff --git a/platform/platform-impl/src/com/intellij/openapi/application/impl/ApplicationImpl.java b/platform/platform-impl/src/com/intellij/openapi/application/impl/ApplicationImpl.java index 930bfc4b7a3b..bf3580c55a03 100644 --- a/platform/platform-impl/src/com/intellij/openapi/application/impl/ApplicationImpl.java +++ b/platform/platform-impl/src/com/intellij/openapi/application/impl/ApplicationImpl.java @@ -296,7 +296,7 @@ public class ApplicationImpl extends PlatformComponentManagerImpl implements App @NotNull @Override public Future executeOnPooledThread(@NotNull final Runnable action) { - boolean privileged = myLock.isPrivilegedReader(); + ReadMostlyRWLock.SuspensionId suspensionId = myLock.currentReadPrivilege(); return ourThreadExecutorsService.submit(new Runnable() { @Override public String toString() { @@ -305,7 +305,7 @@ public class ApplicationImpl extends PlatformComponentManagerImpl implements App @Override public void run() { - try (AccessToken ignored = myLock.setupReadPrivilege(privileged)) { + try (AccessToken ignored = myLock.applyReadPrivilege(suspensionId)) { action.run(); } catch (ProcessCanceledException e) { @@ -324,11 +324,11 @@ public class ApplicationImpl extends PlatformComponentManagerImpl implements App @NotNull @Override public Future executeOnPooledThread(@NotNull final Callable action) { - boolean privileged = myLock.isPrivilegedReader(); + ReadMostlyRWLock.SuspensionId suspensionId = myLock.currentReadPrivilege(); return ourThreadExecutorsService.submit(new Callable() { @Override public T call() { - try (AccessToken ignored = myLock.setupReadPrivilege(privileged)) { + try (AccessToken ignored = myLock.applyReadPrivilege(suspensionId)) { return action.call(); } catch (ProcessCanceledException e) { @@ -1245,34 +1245,30 @@ public class ApplicationImpl extends PlatformComponentManagerImpl implements App public void executeSuspendingWriteAction(@Nullable Project project, @NotNull String title, @NotNull Runnable runnable) { assertIsDispatchThread(); if (!myLock.isWriteLocked()) { - runModalProgress(project, title, false, runnable); + runModalProgress(project, title, runnable); return; } TransactionGuard.getInstance().submitTransactionAndWait(() -> { int prevBase = myWriteStackBase; myWriteStackBase = myWriteActionsStack.size(); - myLock.writeSuspend(); - try { - runModalProgress(project, title, true, runnable); + try (AccessToken ignored = myLock.writeSuspend()) { + runModalProgress(project, title, () -> { + try (AccessToken ignored1 = myLock.grantReadPrivilege()) { + runnable.run(); + } + }); } finally { - try { - myLock.writeResume(); - } - finally { - myWriteStackBase = prevBase; - } + myWriteStackBase = prevBase; } }); } - private void runModalProgress(@Nullable Project project, @NotNull String title, boolean withReadPrivileges, @NotNull Runnable runnable) { + private static void runModalProgress(@Nullable Project project, @NotNull String title, @NotNull Runnable runnable) { ProgressManager.getInstance().run(new Task.Modal(project, title, false) { @Override public void run(@NotNull ProgressIndicator indicator) { - try (AccessToken ignored = myLock.setupReadPrivilege(withReadPrivileges)) { - runnable.run(); - } + runnable.run(); } }); } diff --git a/platform/platform-impl/src/com/intellij/openapi/application/impl/ReadMostlyRWLock.java b/platform/platform-impl/src/com/intellij/openapi/application/impl/ReadMostlyRWLock.java index 2ff56fe918f2..3d0f90c3a8ac 100644 --- a/platform/platform-impl/src/com/intellij/openapi/application/impl/ReadMostlyRWLock.java +++ b/platform/platform-impl/src/com/intellij/openapi/application/impl/ReadMostlyRWLock.java @@ -23,11 +23,12 @@ import com.intellij.openapi.progress.ProgressManager; import com.intellij.util.containers.ConcurrentList; import com.intellij.util.containers.ContainerUtil; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import java.util.ArrayList; import java.util.List; -import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.LockSupport; /** @@ -48,12 +49,12 @@ class ReadMostlyRWLock { private final Thread writeThread; private volatile boolean writeRequested; // this writer is requesting or obtained the write access private volatile boolean writeAcquired; // this writer obtained the write lock - private volatile AtomicInteger writeSuspended = new AtomicInteger(); // All reader threads are registered here. Dead readers are garbage collected in writeUnlock(). private final ConcurrentList readers = ContainerUtil.createConcurrentList(); - /** threads that can start read actions during a suspended write write action */ - private final Set privilegedReaders = ContainerUtil.newConcurrentSet(); + private final Map privilegedReaders = new ConcurrentHashMap<>(); + + private volatile SuspensionId currentSuspension; ReadMostlyRWLock(@NotNull Thread writeThread) { this.writeThread = writeThread; @@ -91,31 +92,29 @@ class ReadMostlyRWLock { checkReadThreadAccess(); Reader status = R.get(); - // be optimistic - if (tryReadLock(status)) { - return; - } - - for(int iter=0;;iter++) { - if (tryReadLock(status)) { + for (int iter = 0; ; iter++) { + if (tryReadLock(status, true)) { return; } ProgressManager.checkCanceled(); + waitABit(status, iter); + } + } - if (iter > SPIN_TO_WAIT_FOR_LOCK) { - status.blocked = true; - try { - LockSupport.parkNanos(this, 1000000); // unparked by writeUnlock - } - finally { - status.blocked = false; - } + private void waitABit(Reader status, int iteration) { + if (iteration > SPIN_TO_WAIT_FOR_LOCK) { + status.blocked = true; + try { + LockSupport.parkNanos(this, 1000000); // unparked by writeUnlock } - else { - Thread.yield(); + finally { + status.blocked = false; } } + else { + Thread.yield(); + } } void readUnlock() { @@ -130,12 +129,12 @@ class ReadMostlyRWLock { boolean tryReadLock() { checkReadThreadAccess(); Reader status = R.get(); - return tryReadLock(status); + return tryReadLock(status, true); } - private boolean tryReadLock(Reader status) { + private boolean tryReadLock(Reader status, boolean checkPrivileges) { if (!writeRequested) { - if (writeSuspended.get() > 0 && !isPrivilegedReader()) { + if (checkPrivileges && currentSuspension != null && !privilegedReaders.containsKey(Thread.currentThread())) { return false; } status.readRequested = true; @@ -169,32 +168,60 @@ class ReadMostlyRWLock { } } - void writeSuspend() { - writeSuspended.incrementAndGet(); + AccessToken writeSuspend() { + SuspensionId prevSuspension = currentSuspension; + if (prevSuspension == null) { + currentSuspension = new SuspensionId(); + } writeUnlock(); + return new AccessToken() { + @Override + public void finish() { + writeLock(); + currentSuspension = prevSuspension; + if (prevSuspension == null) { + ensureNoPrivilegedReaders(); + } + } + }; } - void writeResume() { - writeLock(); - writeSuspended.decrementAndGet(); - + private void ensureNoPrivilegedReaders() { if (!privilegedReaders.isEmpty()) { - List offenderNames = ContainerUtil.map(privilegedReaders, Thread::getName); + List offenderNames = ContainerUtil.map(privilegedReaders.keySet(), Thread::getName); privilegedReaders.clear(); LOG.error("Pooled threads created during write action suspension should have been terminated: " + offenderNames, new Attachment("threadDump.txt", ThreadDumper.dumpThreadsToString())); } } - boolean isPrivilegedReader() { - return privilegedReaders.contains(Thread.currentThread()); + @Nullable + SuspensionId currentReadPrivilege() { + return privilegedReaders.get(Thread.currentThread()); } - AccessToken setupReadPrivilege(boolean allow) { - if (!allow || isPrivilegedReader()) return AccessToken.EMPTY_ACCESS_TOKEN; + @NotNull AccessToken applyReadPrivilege(@Nullable SuspensionId context) { + Reader status = R.get(); + int iter = 0; + while (context != null && context == currentSuspension) { + if (tryReadLock(status, false)) { + try { + return context == currentSuspension ? grantReadPrivilege() : AccessToken.EMPTY_ACCESS_TOKEN; + } + finally { + readUnlock(); + } + } + waitABit(status, iter++); + } + return AccessToken.EMPTY_ACCESS_TOKEN; + } + + @NotNull + AccessToken grantReadPrivilege() { Thread thread = Thread.currentThread(); - privilegedReaders.add(thread); + privilegedReaders.put(thread, currentSuspension); return new AccessToken() { @Override public void finish() { @@ -259,4 +286,6 @@ class ReadMostlyRWLock { boolean isWriteLocked() { return writeAcquired; } + + static class SuspensionId {} } diff --git a/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/ApplicationImplTest.java b/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/ApplicationImplTest.java index 25a9e188c1a1..7bc8a47028c8 100644 --- a/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/ApplicationImplTest.java +++ b/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/ApplicationImplTest.java @@ -26,11 +26,13 @@ import com.intellij.openapi.progress.impl.ProgressManagerImpl; import com.intellij.openapi.progress.util.ProgressIndicatorBase; import com.intellij.openapi.util.Disposer; import com.intellij.openapi.util.EmptyRunnable; +import com.intellij.openapi.util.Ref; import com.intellij.openapi.util.ThrowableComputable; import com.intellij.testFramework.LightPlatformTestCase; import com.intellij.testFramework.LoggedErrorProcessor; import com.intellij.testFramework.PlatformTestUtil; import com.intellij.util.ArrayUtil; +import com.intellij.util.ExceptionUtil; import com.intellij.util.TimeoutUtil; import com.intellij.util.concurrency.Semaphore; import com.intellij.util.containers.ContainerUtil; @@ -546,13 +548,18 @@ public class ApplicationImplTest extends LightPlatformTestCase { }); } + private static void safeWrite(Runnable r) { + ApplicationManager.getApplication().invokeLater(() -> WriteAction.run(r::run)); + UIUtil.dispatchAllInvocationEvents(); + } + public void testSuspendWriteActionDelaysForeignReadActions() throws Exception { - List log = new ArrayList<>(); + List log = Collections.synchronizedList(new ArrayList<>()); Semaphore mayStartForeignRead = new Semaphore(); mayStartForeignRead.down(); - List futures = Collections.synchronizedList(new ArrayList<>()); + List futures = new ArrayList<>(); ApplicationImpl app = (ApplicationImpl)ApplicationManager.getApplication(); futures.add(app.executeOnPooledThread(() -> { @@ -560,7 +567,7 @@ public class ApplicationImplTest extends LightPlatformTestCase { ReadAction.run(() -> log.add("foreign read")); })); - app.invokeLater(() -> WriteAction.run(() -> { + safeWrite(() -> { log.add("write started"); app.executeSuspendingWriteAction(ourProject, "", () -> { app.invokeAndWait(() -> @@ -574,8 +581,7 @@ public class ApplicationImplTest extends LightPlatformTestCase { waitForFuture(app.executeOnPooledThread(() -> ReadAction.run(() -> log.add("forked read")))); }); log.add("write finished"); - })); - UIUtil.dispatchAllInvocationEvents(); + }); futures.forEach(ApplicationImplTest::waitForFuture); assertOrderedEquals(log, "write started", "progress read", "nested write", "forked read", "write finished", "foreign read", "foreign read"); @@ -595,29 +601,83 @@ public class ApplicationImplTest extends LightPlatformTestCase { ApplicationImpl app = (ApplicationImpl)ApplicationManager.getApplication(); assertFalse(app.hasWriteAction(actionClass)); - app.invokeLater(() -> WriteAction.run(() -> { + safeWrite(() -> { assertTrue(app.hasWriteAction(actionClass)); app.executeSuspendingWriteAction(ourProject, "", () -> ReadAction.run(() -> { assertTrue(app.hasWriteAction(actionClass)); waitForFuture(app.executeOnPooledThread(() -> ReadAction.run(() -> assertTrue(app.hasWriteAction(actionClass))))); })); - })); - UIUtil.dispatchAllInvocationEvents(); + }); } public void testPooledThreadsThatHappenInSuspendedWriteActionStayInSuspendedWriteAction() { LoggedErrorProcessor.getInstance().disableStderrDumping(getTestRootDisposable()); + Ref future = Ref.create(); ApplicationImpl app = (ApplicationImpl)ApplicationManager.getApplication(); - app.invokeLater(() -> WriteAction.run(() -> { + safeWrite(() -> { try { - app.executeSuspendingWriteAction(ourProject, "", () -> app.executeOnPooledThread(() -> TimeoutUtil.sleep(1000))); - fail(); + Semaphore started = new Semaphore(); + started.down(); + app.executeSuspendingWriteAction(ourProject, "", () -> { + future.set(app.executeOnPooledThread(() -> { + started.up(); + TimeoutUtil.sleep(1000); + })); + assertTrue(started.waitFor(1000)); + }); + fail("should not allow pooled thread to stay there"); } catch (AssertionError e) { - assertTrue(e.getMessage().contains("should have been terminated")); + assertTrue(ExceptionUtil.getThrowableText(e), isEscapingThreadAssertion(e)); } - })); - UIUtil.dispatchAllInvocationEvents(); + }); + waitForFuture(future.get()); + } + + public void testPooledThreadsStartedAfterQuickSuspendedWriteActionDontGetReadPrivileges() { + ApplicationImpl app = (ApplicationImpl)ApplicationManager.getApplication(); + safeWrite(new Runnable() { + @Override + public void run() { + for (int i = 0; i < 1000; i++) { + checkPooledThreadsDontGetWrongPrivileges(); + UIUtil.dispatchAllInvocationEvents(); + } + } + + private void checkPooledThreadsDontGetWrongPrivileges() { + Ref future = Ref.create(); + + Disposable disableStderrDumping = Disposer.newDisposable(); + LoggedErrorProcessor.getInstance().disableStderrDumping(disableStderrDumping); + + Semaphore mayFinish = new Semaphore(); + mayFinish.down(); + try { + app.executeSuspendingWriteAction(ourProject, "", () -> + future.set(app.executeOnPooledThread( + () -> assertTrue(mayFinish.waitFor(1000))))); + } + catch (AssertionError e) { + if (!isEscapingThreadAssertion(e)) { + e.printStackTrace(); + throw e; + } + } + finally { + Disposer.dispose(disableStderrDumping); + } + + app.executeSuspendingWriteAction(ourProject, "", () -> {}); + mayFinish.up(); + waitForFuture(future.get()); + } + + }); + } + + private static boolean isEscapingThreadAssertion(AssertionError e) { + return e.getMessage().contains("should have been terminated"); } }