@ApiStatus.Internal
  public <T> boolean processQueue(@NotNull BlockingQueue<@NotNull T> things,
                                  @NotNull Queue<@NotNull T> failedToProcess,
                                  @NotNull ProgressIndicator progress,
                                  @NotNull T tombStone,
                                  @NotNull Processor<? super T> thingProcessor) throws ProcessCanceledException {
    // 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 T myFirstTask;

      private final CoroutineContext myContext = ThreadContext.currentThreadContext();

      private MyProcessQueueTask(int seq, @Nullable T firstTask) {
        mySeq = seq;
        myFirstTask = firstTask;
      }

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

              if (element == tombStone) {
                things.put(tombStone); // return just popped tombStone to the 'things' queue for everybody else to see it
                // since the queue is drained up to the tombStone, there surely should be a place for one element, so "put" will not block
                result[0] = true;
                break;
              }
              try (AccessToken ignored = ThreadContext.installThreadContext(myContext, true)) {
                ProgressManager.checkCanceled();
                if (!thingProcessor.process(element)) {
                  break;
                }
              }
              catch (RuntimeException|Error e) {
                if (logAllExceptions) {
                  LOG.info("Failed to process " + element + ". Add too failed query.", e);
                }
                failedToProcess.add(element);
                throw e;
              }
              element = null;
            }
          }
          catch (InterruptedException e) {
            throw new RuntimeException(e);
          }
        }, progress);
        return result[0];
      }

      @Override
      public @NonNls String toString() {
        return super.toString() + " seq="+mySeq;
      }
    }
    progress.checkCanceled(); // do not start up expensive threads if there's no need to
    int size = things.size();
    boolean isQueueBounded = things.contains(tombStone);
    // start up (CPU cores) parallel tasks but no more than (queue size)
    int n = Math.max(1, Math.min(isQueueBounded ? size-1 : Integer.MAX_VALUE, JobSchedulerImpl.getJobPoolParallelism() - 1));
    List<ForkJoinTask<Boolean>> tasks = new ArrayList<>(n-1);
    List<T> firstElements = new ArrayList<>(n);
    things.drainTo(firstElements, n);
    // if the tombstone was removed by this batch operation, return it back to the queue to give chance tasks to stop themselves
    if (ContainerUtil.getLastItem(firstElements) == tombStone) {
      firstElements.remove(firstElements.size() - 1);
      try {
        things.put(tombStone);
      }
      catch (InterruptedException e) {
        LOG.error(e);
      }
    }
    for (int i = 1; i < n; i++) {
      tasks.add(ForkJoinPool.commonPool().submit(new MyProcessQueueTask(i, i < firstElements.size() ? firstElements.get(i) : null)));
    }
    MyProcessQueueTask firstTask = new MyProcessQueueTask(0, ContainerUtil.getFirstItem(firstElements));
    // execute the first task directly in this thread to avoid thread starvation
    boolean result = false;
    Throwable exception = null;
    try {
      result = firstTask.call();
    }
    catch (Throwable e) {
      exception = e;
    }
    for (ForkJoinTask<Boolean> task : tasks) {
      try {
        //noinspection NonShortCircuitBooleanExpression
        result &= task.join();
      }
      catch (Throwable e) {
        exception = e;
      }
    }
    if (exception != null) {
      ExceptionUtil.rethrow(exception);
    }
    return result;
  }
@ApiStatus.Internal
  public <T> boolean processQueue(@NotNull BlockingQueue<@NotNull T> things,
                                  @NotNull Queue<@NotNull T> failedToProcess,
                                  @NotNull ProgressIndicator progress,
                                  @NotNull T tombStone,
                                  @NotNull Processor<? super T> thingProcessor)