PCE left classesBeingProcessed dirty fixed

java/java-indexing-impl/src/com/intellij/psi/impl/search/LazyConcurrentCollection.java

@@ -30,6 +30,7 @@ import org.jetbrains.annotations.NotNull;
 import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Collection of elements of type V which is
@@ -53,9 +54,9 @@ class LazyConcurrentCollection<T,V> implements Iterable<V> {
   //    If more elements requested for this iterator, the processMoreSubclasses() is called which tries to populate 'subClasses' with more inheritors.
   private final HashSetQueue<T> subClasses; // guarded by lock
   private final Object lock = new Object(); // MUST NOT acquire read action inside this lock
-  @NotNull private final Function<T, V> myConvertor;
+  @NotNull private final Function<T, V> myAnchorToValueConvertor;
   @NotNull private final MoreElementsGenerator<T,V> myGenerator;
-  @NotNull private final Predicate<? super V> myApplicableFilter;
+  @NotNull private final Predicate<? super V> myApplicableForGenerationFilter;
   private final Semaphore currentlyProcessingClasses = new Semaphore();
 
   private final HashSetQueue.PositionalIterator<T> candidatesToFindSubclassesIterator; // guarded by lock
@@ -67,13 +68,13 @@ class LazyConcurrentCollection<T,V> implements Iterable<V> {
 
   LazyConcurrentCollection(@NotNull T seedElement,
                            @NotNull Function<T, V> convertor,
-                           @NotNull Predicate<? super V> applicableFilter,
+                           @NotNull Predicate<? super V> applicableForGenerationFilter,
                            @NotNull MoreElementsGenerator<T, V> generator) {
     subClasses = new HashSetQueue<>();
     subClasses.add(seedElement);
-    myConvertor = convertor;
+    myAnchorToValueConvertor = convertor;
     myGenerator = generator;
-    myApplicableFilter = applicableFilter;
+    myApplicableForGenerationFilter = applicableForGenerationFilter;
     candidatesToFindSubclassesIterator = subClasses.iterator();
   }
 
@@ -111,55 +112,58 @@ class LazyConcurrentCollection<T,V> implements Iterable<V> {
         synchronized (lock) {
           next = subClassIterator.next();
         }
-        return myConvertor.fun(next);
+        return myAnchorToValueConvertor.fun(next);
       }
     };
   }
 
-  private Pair.NonNull<T,V> findNextClassInQueue(@NotNull HashSetQueue.PositionalIterator.IteratorPosition<T> position) {
-    // find the first class which is fit (not anonymous and not final and retrievable from PsiAnchor) and not already processed (flag PROCESSING_SUBCLASSES_STATUS in class user data)
-    // couldn't call iterator.next() until class is processed, so use position.peek()/position.next() which don't advance iterator
-    while (position != null) {
-      ProgressManager.checkCanceled();
-      T anchor = position.peek();
-      V value = myConvertor.fun(anchor);
-      boolean isAccepted = value != null && myApplicableFilter.apply(value);
-
-      if (isAccepted && !classesProcessed.contains(anchor) && classesBeingProcessed.add(anchor)) {
-        return Pair.createNonNull(anchor, value);
-      }
-      // the candidate is already being processed in the other thread, try the next one (not advancing iterator!)
-      position = position.next();
-    }
-    return null;
-  }
-
   // polls 'subClasses' for more sub classes and call DirectClassInheritorsSearch for them
-  private void processMoreSubclasses(@NotNull Iterator<T> subClassIterator) {
+  // returns true if some classes were found
+  private boolean processMoreSubclasses(@NotNull Iterator<T> subClassIterator) {
     while (true) {
       ProgressManager.checkCanceled();
 
       Pair.NonNull<T,V> pair =
-        ApplicationManager.getApplication().runReadAction(new Computable<Pair.NonNull<T,V>>() {
-          @Override
-          public Pair.NonNull<T,V> compute() {
-            synchronized (lock) {
-              // Find the classes in subClasses collection to operate on
-              // (without advancing the candidatesToFindSubclassesIterator iterator - it will be moved after the class successfully handled - to protect against PCE, INRE, etc)
-              // The found class will be marked as being analyzed - placed in classesBeingProcessed collection
-              HashSetQueue.PositionalIterator.IteratorPosition<T> startPosition = candidatesToFindSubclassesIterator.position().next();
-              Pair.NonNull<T,V> pair = startPosition == null ? null : findNextClassInQueue(startPosition);
-              if (pair != null) {
-                currentlyProcessingClasses.down();
-              }
-              return pair;
+        ApplicationManager.getApplication().runReadAction((Computable<Pair.NonNull<T,V>>)() -> {
+          synchronized (lock) {
+            // Find the classes in subClasses collection to operate on
+            // (without advancing the candidatesToFindSubclassesIterator iterator - it will be moved after the class successfully handled - to protect against PCE, INRE, etc)
+            // The found class will be marked as being analyzed - placed in classesBeingProcessed collection
+            HashSetQueue.PositionalIterator.IteratorPosition<T> startPosition = candidatesToFindSubclassesIterator.position().next();
+            Pair.NonNull<T,V> next = startPosition == null ? null : findNextClassInQueue(startPosition);
+            if (next != null) {
+              currentlyProcessingClasses.down();
+              classesBeingProcessed.add(next.getFirst());
             }
+            return next;
           }
         });
       if (pair == null) {
         // no candidates left in queue, exit
         // but first, wait for other threads to process their candidates
-        break;
+        synchronized (lock) {
+          advanceIteratorOnSuccess(); // to skip unsuitable classes like final etc from the queue
+          if (subClassIterator.hasNext()) {
+            return true;
+          }
+        }
+
+        boolean producedSomething = waitForOtherThreadsToFinishProcessing(subClassIterator);
+        if (producedSomething) {
+          return true;
+        }
+
+        // aaaaaaaa! Other threads were unable to produce anything. That can be because:
+        // - the whole queue has been processed. => exit, return false
+        // - the other thread has been interrupted. => check the queue again to pickup the work it dropped.
+        synchronized (lock) {
+          advanceIteratorOnSuccess(); // to skip unsuitable classes like final etc from the queue
+          if (!candidatesToFindSubclassesIterator.hasNext()) {
+            return false;
+          }
+        }
+
+        continue; // check again
       }
 
       V candidate = pair.getSecond();
@@ -170,29 +174,33 @@ class LazyConcurrentCollection<T,V> implements Iterable<V> {
             subClasses.add(generatedElement);
           }
         });
+        synchronized (lock) {
+          classesProcessed.add(anchor);
+          advanceIteratorOnSuccess();
+          if (subClassIterator.hasNext()) {
+            // we've added something to subClasses so we can return and the iterator can move forward at least once;
+            // more elements will be added on the subsequent call to .next()
+            return true;
+          }
+        }
       }
       finally {
-        currentlyProcessingClasses.up();
-      }
-
-      synchronized (lock) {
-        classesBeingProcessed.remove(anchor);
-        classesProcessed.add(anchor);
-        advanceIteratorOnSuccess();
-        if (subClassIterator.hasNext()) {
-          // we've added something to subClasses so we can return and the iterator can move forward at least once;
-          // more elements will be added on the subsequent call to .next()
-          return;
+        synchronized (lock) {
+          classesBeingProcessed.remove(anchor);
+          currentlyProcessingClasses.up();
         }
       }
     }
+  }
 
+  private boolean waitForOtherThreadsToFinishProcessing(@NotNull final Iterator<T> subClassIterator) {
     // Found nothing, have to wait for other threads because:
     // The first thread comes and takes a class off the queue to search for inheritors,
     // the second thread comes and sees there is no classes in the queue.
     // The second thread should not return nothing, it should wait for the first thread to finish.
     //
     // Wait within managedBlock to signal FJP this thread is locked (to avoid thread starvation and deadlocks)
+    AtomicBoolean hasNext = new AtomicBoolean();
     try {
       ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
         @Override
@@ -204,7 +212,10 @@ class LazyConcurrentCollection<T,V> implements Iterable<V> {
         @Override
         public boolean isReleasable() {
           synchronized (lock) {
-            return !currentlyProcessingClasses.isDown() || subClassIterator.hasNext();
+            // other thread produced something or all of them reached the end of list
+            boolean producedSomething = subClassIterator.hasNext();
+            hasNext.set(producedSomething); // store the result to avoid locking again after exit
+            return producedSomething || !candidatesToFindSubclassesIterator.hasNext() || classesBeingProcessed.isEmpty();
           }
         }
       });
@@ -212,20 +223,41 @@ class LazyConcurrentCollection<T,V> implements Iterable<V> {
     catch (InterruptedException e) {
       throw new RuntimeException(e);
     }
+    return hasNext.get();
   }
 
-  private void advanceIteratorOnSuccess() {
-    HashSetQueue.PositionalIterator.IteratorPosition<T> position = candidatesToFindSubclassesIterator.position().next();
+  // under lock
+  private Pair.NonNull<T,V> findNextClassInQueue(@NotNull HashSetQueue.PositionalIterator.IteratorPosition<T> position) {
+    // find the first class suitable for analyzing inheritors of (not anonymous and not final and retrievable from PsiAnchor) and not already processed or being processed (by other thread)
+    // couldn't call iterator.next() until class is processed, so use position.peek()/position.next() which don't advance iterator
     while (position != null) {
-      T next = position.peek();
-      if (classesProcessed.contains(next)) {
+      ProgressManager.checkCanceled();
+      T anchor = position.peek();
+      if (!classesProcessed.contains(anchor) && !classesBeingProcessed.contains(anchor)) {
+        V value = myAnchorToValueConvertor.fun(anchor);
+        boolean isAccepted = value != null && myApplicableForGenerationFilter.apply(value);
+        if (isAccepted) {
+          return Pair.createNonNull(anchor, value);
+        }
+        classesProcessed.add(anchor);
+      }
+      // the candidate is already being processed in the other thread, try the next one (not advancing iterator!)
+      position = position.next();
+    }
+    return null;
+  }
+
+  // under lock
+  private void advanceIteratorOnSuccess() {
+    while (candidatesToFindSubclassesIterator.hasNext()) {
+      T next = candidatesToFindSubclassesIterator.position().next().peek();
+      boolean removed = classesProcessed.remove(next);
+      if (removed) {
         candidatesToFindSubclassesIterator.next();
-        classesProcessed.remove(next);
       }
       else {
         break;
       }
-      position = position.next();
     }
   }
 }

java/java-tests/testSrc/com/intellij/psi/search/ClassInheritorsTest.java

@@ -16,17 +16,30 @@
 package com.intellij.psi.search;
 
 import com.intellij.codeInsight.daemon.DaemonAnalyzerTestCase;
+import com.intellij.concurrency.JobScheduler;
+import com.intellij.openapi.application.ApplicationManager;
+import com.intellij.openapi.progress.ProcessCanceledException;
+import com.intellij.openapi.progress.ProgressManager;
+import com.intellij.openapi.progress.util.StandardProgressIndicatorBase;
 import com.intellij.openapi.projectRoots.Sdk;
+import com.intellij.openapi.vfs.VirtualFile;
 import com.intellij.psi.PsiClass;
+import com.intellij.psi.PsiJavaFile;
 import com.intellij.psi.impl.compiled.ClsClassImpl;
 import com.intellij.psi.impl.source.PsiClassImpl;
 import com.intellij.psi.search.searches.ClassInheritorsSearch;
 import com.intellij.psi.search.searches.DirectClassInheritorsSearch;
 import com.intellij.testFramework.IdeaTestUtil;
+import com.intellij.util.CommonProcessors;
+import gnu.trove.THashSet;
 
+import java.io.File;
+import java.io.IOException;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 public class ClassInheritorsTest extends DaemonAnalyzerTestCase {
@@ -49,4 +62,43 @@ public class ClassInheritorsTest extends DaemonAnalyzerTestCase {
     List<String> allFqn = allSubClasses.stream().map(PsiClass::getQualifiedName).sorted().collect(Collectors.toList());
     assertEquals(allFqn.toString(), allFqn.size(), new HashSet<>(allFqn).size());
   }
+
+  public void testStressInPresenceOfPCEs() throws IOException {
+    ApplicationManager.getApplication().assertIsDispatchThread(); // no write action can go through while we test
+    File tempDirectory = createTempDirectory();
+    VirtualFile dir = getVirtualFile(tempDirectory);
+    addSourceContentToRoots(getModule(), dir);
+    int N = 1000;
+    PsiJavaFile file0 = (PsiJavaFile)createFile(myModule, dir, "C0.java", "class C0 { }");
+    for (int i=1;i<N ;i++) {
+      int extI = i - 1 - (i - 1) % 10; // 10 inheritors
+      createFile(myModule, dir, "C" + i + ".java", "class C" + i + " extends C" + extI + " { }");
+    }
+    PsiClass class0 = file0.getClasses()[0];
+
+    int delayToCancel = 100;
+    for (int i=0;i<1000;i++) {
+      //System.out.println("i = " + i+ "; delayToCancel="+delayToCancel);
+      StandardProgressIndicatorBase progress = new StandardProgressIndicatorBase();
+      JobScheduler.getScheduler().schedule(progress::cancel, delayToCancel, TimeUnit.MILLISECONDS);
+      try {
+        Collections.nCopies(Runtime.getRuntime().availableProcessors(), "").stream().parallel().forEach(__ -> {
+          Collection<PsiClass> inheritors = Collections.synchronizedSet(new THashSet<>());
+          ProgressManager.getInstance().executeProcessUnderProgress(()-> {
+            boolean success = ClassInheritorsSearch.search(class0).forEach(new CommonProcessors.CollectProcessor<>(inheritors));
+            if (N - 1 != inheritors.size() || !success) {
+              assertEquals(N - 1, inheritors.size());
+            }
+          }, progress);
+        });
+        myPsiManager.dropResolveCaches();
+        //System.out.println("Iterated all");
+        delayToCancel--;
+      }
+      catch (ProcessCanceledException e) {
+        //System.out.println("e = " + e);
+        delayToCancel++;
+      }
+    }
+  }
 }