MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-09-27 10:03:11 +07:00
Code Issues Packages Projects Releases Wiki Activity

don't grant read privileges to pooled threads that start after write action is resumed (IDEA-CR-14502)

Browse Source
This commit is contained in:
peter
2016-10-20 17:26:10 +02:00
parent 3db867ce03
commit d799dad9ff
3 changed files with 154 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/platform-impl/src/com/intellij/openapi/application/impl/ApplicationImpl.java
+14 -18
View File
@@ -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 <T> Future<T> executeOnPooledThread(@NotNull final Callable<T> action) {
boolean privileged = myLock.isPrivilegedReader();
ReadMostlyRWLock.SuspensionId suspensionId = myLock.currentReadPrivilege();
return ourThreadExecutorsService.submit(new Callable<T>() {
@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();
}
});
}
platform/platform-impl/src/com/intellij/openapi/application/impl/ReadMostlyRWLock.java
+66 -37
View File
@@ -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<Reader> readers = ContainerUtil.createConcurrentList();
/** threads that can start read actions during a suspended write write action */
private final Set<Thread> privilegedReaders = ContainerUtil.newConcurrentSet();
private final Map<Thread, SuspensionId> 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<String> offenderNames = ContainerUtil.map(privilegedReaders, Thread::getName);
List<String> 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 {}
}
platform/platform-tests/testSrc/com/intellij/openapi/application/impl/ApplicationImplTest.java
+74 -14
View File
@@ -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<String> log = new ArrayList<>();
List<String> log = Collections.synchronizedList(new ArrayList<>());
Semaphore mayStartForeignRead = new Semaphore();
mayStartForeignRead.down();
List<Future> futures = Collections.synchronizedList(new ArrayList<>());
List<Future> 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> 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> 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");
}
}