SubmissionReceipt (@NotNull T task) { List<T> disposeQueue = new ArrayList<>(1); T newTask = task; SubmissionReceipt receipt; var childContext = AppExecutorUtil.propagateContextOrCancellation() ? Propagation.createChildContext() : null; CoroutineContext currentContext = childContext == null ? null : childContext.getContext(); Job job = childContext == null ? null : childContext.getJob(); synchronized (myLock) { for (int i = myTasksQueue.size() - 1; i >= 0; i--) { T oldTask = myTasksQueue.get(i); ProgressIndicatorBase indicator = myProgresses.get(oldTask); ChildContext otherChildContext = myThreadContexts.get(oldTask); CoroutineContext otherContext = otherChildContext == null ? null : otherChildContext.getContext(); //dispose cancelled tasks if (indicator == null || indicator.isCanceled()) { myTasksQueue.remove(i); disposeQueue.add(oldTask); continue; } // note that parent may know nothing about children, so the following may happen (just like in case with `equals` with inheritance): // class Parent; class Child extends Parent // parent.tryMergeWith(child) != child.tryMergeWith(parent) // At the moment we prevent accidental errors by forcing tasks' class equality. // More permissive strategy (that we don't apply) would be to check "isAssignableFrom" and always use `child.tryMergeWith` if (task.getClass() != oldTask.getClass()) { continue; } T mergedTask = task.tryMergeWith(oldTask); boolean contextsEqual; if (AppExecutorUtil.propagateContextOrCancellation()) { if ((currentContext == null) != (otherContext == null)) { contextsEqual = false; } else { contextsEqual = currentContext == null || ThreadContext.getContextSkeleton(currentContext).equals(ThreadContext.getContextSkeleton(otherContext)); } } else { contextsEqual = true; } if (mergedTask == oldTask && contextsEqual) { // new task completely absorbed by the old task which means that we don't need to modify the queue newTask = null; if (job != null) { job.cancel(null); } disposeQueue.add(task); break; } if (mergedTask != null) { LOG.debug("Merged " + task + " with " + oldTask); newTask = mergedTask; myTasksQueue.remove(i); disposeQueue.add(oldTask); break; } } //register the new task last, preserving FIFO order T taskToAdd = newTask; if (taskToAdd != null) { myTasksQueue.add(taskToAdd); mySubmittedTasksCount.incrementAndGet(); ProgressIndicatorBase progress = new ProgressIndicatorBase(); myProgresses.put(taskToAdd, progress); myThreadContexts.put(taskToAdd, childContext); Disposer.register(taskToAdd, () -> { //a removed progress means the task would be ignored on queue processing synchronized (myLock) { myProgresses.remove(taskToAdd); myThreadContexts.remove(taskToAdd); if (job != null) { job.cancel(null); } } progress.cancel(); }); } receipt = new SubmissionReceipt(mySubmittedTasksCount.get()); } disposeSafe(disposeQueue); return receipt; }