@Override
  @ApiStatus.Internal
  public <T> boolean procInOrderAsync(@NotNull ProgressIndicator progress,
                                      int maxQueueSize,
                                      @NotNull PairProcessor<? super T, ? super QueueController<? super T>> thingProcessor,
                                      @NotNull Processor<? super QueueController<? super T>> otherActions) throws ProcessCanceledException {
    progress.checkCanceled(); // do not start up expensive threads if there's no need to
    // optimization: if we know the max number of elements in the queue, use the cheaper ABQ
    BlockingQueue<T> things = maxQueueSize < Integer.MAX_VALUE ?
                              new ArrayBlockingQueue<>(maxQueueSize + JobSchedulerImpl.getJobPoolParallelism())
                              : new LinkedBlockingQueue<>();
    // start (up to CPU cores) parallel tasks but no more than (queue size)
    int n = Math.max(1, Math.min(maxQueueSize, JobSchedulerImpl.getJobPoolParallelism() - 1));

    QueueController<? super T> addToQueue = new QueueController<T>() {
      @Override
      public void enqueue(T t) {
        boolean added = things.offer(t);
        if (!added) {
          dropEverythingAndPanic();
        }
      }

      @Override
      public void dropEverythingAndPanic() {
        // drop everything you are doing and stop
        if (things.peek() != TOMBSTONE()) {
          things.clear();
        }
        finish();
      }

      @Override
      public void finish() {
        // mark the queue end, for all workers to take
        for (int i = 0; i < n; i++) {
          if (!things.offer(TOMBSTONE())) {
            // if the queue is not able to accommodate all tombstones, then someone else has already put tombstones there, and it's safe to ignore the false result
            break;
          }
        }
      }


    };

    AtomicBoolean futureResult = new AtomicBoolean(true);
    AtomicReference<Callable<Boolean>> firstTask = new AtomicReference<>();
    CountedCompleter<Boolean> completer = new CountedCompleter<>() {
      @Override
      public void compute() {
        try {
          firstTask.get().call();
        }
        catch (Exception e) {
          completeExceptionally(e);
        }
      }

      @Override
      public Boolean getRawResult() {
        return futureResult.get();
      }
    };

    // spawn up to (JobSchedulerImpl.getJobPoolParallelism() - 1) tasks,
    // each one trying to dequeue as many elements off `things` as possible and handing them to `thingProcessor`, until `tombStone` is hit
    final class MyProcessQueueTask implements Callable<Boolean> {
      private final int mySeq;
      private final CoroutineContext myContext = ThreadContext.currentThreadContext();

      private MyProcessQueueTask(int seq) {
        mySeq = seq;
      }

      @Override
      public Boolean call() {
        boolean[] result = new boolean[1];
        try {
          ProgressManager.getInstance().executeProcessUnderProgress(() -> {
            try {
              while (true) {
                T element = things.take();

                if (element == TOMBSTONE()) {
                  result[0] = true;
                  break;
                }
                try (AccessToken ignored = ThreadContext.installThreadContext(myContext, true)) {
                  ProgressManager.checkCanceled();
                  if (!thingProcessor.process(element, addToQueue)) {
                    break;
                  }
                }
                catch (RuntimeException|Error e) {
                  if (logAllExceptions) {
                    LOG.info("Failed to process " + element + ". Add too failed query.", e);
                  }
                  throw e;
                }
              }
            }
            catch (InterruptedException e) {
              throw new RuntimeException(e);
            }
          }, progress);
          completer.tryComplete();
          return result[0];
        }
        catch (Throwable throwable) {
          completer.completeExceptionally(throwable);
          return false;
        }
        finally {
          if (!result[0]) {
            // in case of exception, we need to cancel all tasks as quick as possible, so if there is a task blocked in ".take()" we could unblock it by offering tombstone
            addToQueue.dropEverythingAndPanic();
            futureResult.set(false);
          }
        }
      }

      @Override
      public @NonNls String toString() {
        return super.toString() + " seq="+mySeq;
      }
    }
    completer.setPendingCount(n-1);
    for (int i = 1; i < n; i++) {
      ForkJoinPool.commonPool().submit(new MyProcessQueueTask(i));
    }
    firstTask.set(new MyProcessQueueTask(0));

    try {
      // execute other actions while we are processing enqueued elements
      if (!otherActions.process(addToQueue)) {
        futureResult.set(false);
      }
    }
    catch (Throwable t) {
      // in case of any exception or error in normal flow, terminate background tasks
      addToQueue.dropEverythingAndPanic();
      throw t;
    }
    finally {
      // FJP may execute tasks in the same stackframe.
      // This is the behavior similar to the manual pumping the event queue,
      // so we need to prepare the stackframe for installation of context.
      try (AccessToken ignored = ThreadContext.resetThreadContext()) {
        // do not call future.get() to avoid overcompensation
        completer.invoke();
      }
    }
    return futureResult.get();
  }
@Override
  @ApiStatus.Internal
  public <T> boolean procInOrderAsync(@NotNull ProgressIndicator progress,
                                      int maxQueueSize,
                                      @NotNull PairProcessor<? super T, ? super QueueController<? super T>> thingProcessor,
                                      @NotNull Processor<? super QueueController<? super T>> otherActions)