Dispatcher.java

1 
/*
2 
 * Copyright 2014-2026 Grzegorz Piwowarek, https://4comprehension.com/
3 
 *
4 
 * Licensed under the Apache License, Version 2.0 (the "License");
5 
 * you may not use this file except in compliance with the License.
6 
 * You may obtain a copy of the License at
7 
 *
8 
 * https://www.apache.org/licenses/LICENSE-2.0
9 
 *
10 
 * Unless required by applicable law or agreed to in writing, software
11 
 * distributed under the License is distributed on an "AS IS" BASIS,
12 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 
 * See the License for the specific language governing permissions and
14 
 * limitations under the License.
15 
 */
16 
package com.pivovarit.collectors;
17
18 
import java.lang.ref.Cleaner;
19 
import java.util.Objects;
20 
import java.util.Set;
21 
import java.util.concurrent.BlockingQueue;
22 
import java.util.concurrent.CompletableFuture;
23 
import java.util.concurrent.ConcurrentHashMap;
24 
import java.util.concurrent.Executor;
25 
import java.util.concurrent.FutureTask;
26 
import java.util.concurrent.LinkedBlockingQueue;
27 
import java.util.concurrent.RejectedExecutionException;
28 
import java.util.concurrent.Semaphore;
29 
import java.util.concurrent.ThreadFactory;
30 
import java.util.concurrent.atomic.AtomicReference;
31 
import java.util.function.Consumer;
32 
import java.util.function.Supplier;
33
34 
import static com.pivovarit.collectors.Preconditions.requireValidExecutor;
35
36 
/**
37 
 * @author Grzegorz Piwowarek
38 
 */
39 
final class Dispatcher<T> {
40
41 
    private static final Cleaner CLEANER = Cleaner.create();
42
43 
    private final AtomicReference<Throwable> failure = new AtomicReference<>();
44 
    private final Set<InterruptibleCompletableFuture<T>> pendingFutures = ConcurrentHashMap.newKeySet();
45 
    private final BlockingQueue<DispatchItem> workingQueue = new LinkedBlockingQueue<>();
46
47 
    private final ThreadFactory dispatcherThreadFactory = Thread.ofVirtual()
48 
      .name("parallel-collectors-dispatcher-", 0)
49 
      .factory();
50
51 
    private final Consumer<Thread> dispatcherThreadHook;
52
53 
    private final Executor executor;
54 
    private final Semaphore limiter;
55
56 
    // UNSTARTED -> RUNNING -> SHUTTING_DOWN
57 
    enum State {
58 
        UNSTARTED, RUNNING, SHUTTING_DOWN
59 
    }
60
61 
    private final AtomicReference<State> state = new AtomicReference<>(State.UNSTARTED);
62
63 
    Dispatcher(Executor executor, int permits, Consumer<Thread> dispatcherThreadHook) {
64 1 1. <init> : removed call to com/pivovarit/collectors/Preconditions::requireValidExecutor → SURVIVED
 
        requireValidExecutor(executor);
65 
        this.executor = executor;
66 
        this.dispatcherThreadHook = dispatcherThreadHook;
67 
        this.limiter = new Semaphore(permits);
68 
    }
69
70 
    Dispatcher(Executor executor, int permits) {
71 
        this(executor, permits, c -> {});
72 
    }
73
74 
    Dispatcher(Executor executor) {
75 
        this(executor, c -> {});
76 
    }
77
78 
    Dispatcher(Executor executor, Consumer<Thread> dispatcherThreadHook) {
79 1 1. <init> : removed call to com/pivovarit/collectors/Preconditions::requireValidExecutor → SURVIVED
 
        requireValidExecutor(executor);
80 
        this.executor = executor;
81 
        this.dispatcherThreadHook = dispatcherThreadHook;
82 
        this.limiter = null;
83 
    }
84
85 
    void start() {
86 1 1. start : negated conditional → TIMED_OUT
 
        if (state.compareAndSet(State.UNSTARTED, State.RUNNING)) {
87 
            var thread = dispatcherThreadFactory.newThread(() -> {
88 
                try {
89 
                    while (true) {
90 
                        switch (workingQueue.take()) {
91 
                            case DispatchItem.Task(FutureTask<?> task) -> {
92 
                                try {
93 1 1. lambda$start$0 : negated conditional → KILLED
 
                                    if (limiter != null) {
94 1 1. lambda$start$0 : removed call to java/util/concurrent/Semaphore::acquire → KILLED
 
                                        limiter.acquire();
95 
                                    }
96 
                                } catch (InterruptedException e) {
97 1 1. lambda$start$0 : removed call to com/pivovarit/collectors/Dispatcher::abort → NO_COVERAGE
 
                                    abort(e);
98 
                                    return;
99 
                                }
100 
                                try {
101 2 1. lambda$start$0 : removed call to com/pivovarit/collectors/Dispatcher::retry → TIMED_OUT
2. lambda$start$1 : removed call to java/util/concurrent/Executor::execute → TIMED_OUT
 
                                    retry(() -> executor.execute(() -> {
102 
                                        try {
103 1 1. lambda$start$2 : removed call to java/util/concurrent/FutureTask::run → TIMED_OUT
 
                                            task.run();
104 
                                        } finally {
105 1 1. lambda$start$2 : negated conditional → KILLED
 
                                            if (task.isCancelled()) {
106 
                                                Thread.interrupted();
107 
                                            }
108 1 1. lambda$start$2 : negated conditional → TIMED_OUT
 
                                            if (limiter != null) {
109 1 1. lambda$start$2 : removed call to java/util/concurrent/Semaphore::release → TIMED_OUT
 
                                                limiter.release();
110 
                                            }
111 
                                        }
112 
                                    }));
113 
                                } catch (RejectedExecutionException e) {
114 1 1. lambda$start$0 : negated conditional → SURVIVED
 
                                    if (limiter != null) {
115 1 1. lambda$start$0 : removed call to java/util/concurrent/Semaphore::release → SURVIVED
 
                                        limiter.release();
116 
                                    }
117
118 
                                    throw e;
119 
                                }
120 
                            }
121 
                            case DispatchItem.Stop ignored -> {
122 
                                return;
123 
                            }
124 
                        }
125 
                    }
126 
                } catch (Throwable e) {
127 1 1. lambda$start$0 : removed call to com/pivovarit/collectors/Dispatcher::abort → KILLED
 
                    abort(e);
128 
                }
129 
            });
130 1 1. start : removed call to java/util/function/Consumer::accept → KILLED
 
            dispatcherThreadHook.accept(thread);
131 1 1. start : removed call to java/lang/Thread::start → TIMED_OUT
 
            thread.start();
132 
        }
133 
    }
134
135 
    void stop() {
136 1 1. stop : negated conditional → TIMED_OUT
 
        if (state.compareAndSet(State.RUNNING, State.SHUTTING_DOWN)) {
137 
            workingQueue.add(DispatchItem.Stop.POISON_PILL);
138 
        }
139 
    }
140
141 
    Cleaner.Cleanable registerTerminationGuard(Object guard) {
142 
        var queue = workingQueue;
143 
        var currentState = state;
144
145 1 1. registerTerminationGuard : replaced return value with null for com/pivovarit/collectors/Dispatcher::registerTerminationGuard → KILLED
 
        return CLEANER.register(guard, () -> {
146 1 1. lambda$registerTerminationGuard$0 : negated conditional → TIMED_OUT
 
            if (currentState.compareAndSet(State.RUNNING, State.SHUTTING_DOWN)) {
147 
                queue.add(DispatchItem.Stop.POISON_PILL);
148 
            }
149 
        });
150 
    }
151
152 
    boolean wasStarted() {
153 1 1. wasStarted : replaced boolean return with false for com/pivovarit/collectors/Dispatcher::wasStarted → SURVIVED
 
        return switch (state.get()) {
154 
            case UNSTARTED -> false;
155 
            case RUNNING, SHUTTING_DOWN -> true;
156 
        };
157 
    }
158
159 
    boolean wasShutdown() {
160 1 1. wasShutdown : replaced boolean return with false for com/pivovarit/collectors/Dispatcher::wasShutdown → TIMED_OUT
 
        return switch (state.get()) {
161 
            case UNSTARTED, RUNNING -> false;
162 
            case SHUTTING_DOWN -> true;
163 
        };
164 
    }
165
166 
    CompletableFuture<T> submit(Supplier<T> supplier) {
167 1 1. submit : negated conditional → KILLED
 
        if (state.get() == State.SHUTTING_DOWN) {
168 
            throw new IllegalStateException("collector was already used and cannot be reused");
169 
        }
170 
        var future = new InterruptibleCompletableFuture<T>();
171 
        var task = new FutureTask<>(() -> {
172 
            try {
173 
                future.complete(supplier.get());
174 
            } catch (Throwable e) {
175 1 1. lambda$submit$0 : removed call to com/pivovarit/collectors/Dispatcher::abort → TIMED_OUT
 
                abort(e);
176 
                Thread.interrupted();
177 
            }
178 
        }, null);
179 1 1. submit : removed call to com/pivovarit/collectors/InterruptibleCompletableFuture::completedBy → TIMED_OUT
 
        future.completedBy(task);
180 
        pendingFutures.add(future);
181 
        future.whenComplete((result, ex) -> pendingFutures.remove(future));
182
183 
        var e = failure.get();
184 1 1. submit : negated conditional → KILLED
 
        if (e != null) {
185 1 1. submit : removed call to com/pivovarit/collectors/Dispatcher::abort → KILLED
 
            abort(future, e);
186 1 1. submit : replaced return value with null for com/pivovarit/collectors/Dispatcher::submit → KILLED
 
            return future;
187 
        }
188 
        workingQueue.add(new DispatchItem.Task(task));
189 1 1. submit : replaced return value with null for com/pivovarit/collectors/Dispatcher::submit → KILLED
 
        return future;
190 
    }
191
192 
    private void abort(Throwable e) {
193 1 1. abort : negated conditional → TIMED_OUT
 
        if (failure.compareAndSet(null, e)) {
194 
            for (var future : pendingFutures) {
195 1 1. abort : removed call to com/pivovarit/collectors/Dispatcher::abort → TIMED_OUT
 
                abort(future, e);
196 
            }
197 
        }
198 
    }
199
200 
    private static void abort(InterruptibleCompletableFuture<?> future, Throwable e) {
201 
        future.completeExceptionally(e);
202 
        future.cancel(true);
203 
    }
204
205 
    private static void retry(Runnable runnable) {
206 
        try {
207 1 1. retry : removed call to java/lang/Runnable::run → TIMED_OUT
 
            runnable.run();
208 
        } catch (RejectedExecutionException e) {
209 1 1. retry : removed call to java/lang/Thread::onSpinWait → SURVIVED
 
            Thread.onSpinWait();
210 1 1. retry : removed call to java/lang/Runnable::run → KILLED
 
            runnable.run();
211 
        }
212 
    }
213
214 
    sealed interface DispatchItem permits DispatchItem.Task, DispatchItem.Stop {
215 
        record Task(FutureTask<?> task) implements DispatchItem {
216 
            public Task {
217 
                Objects.requireNonNull(task);
218 
            }
219 
        }
220
221 
        enum Stop implements DispatchItem {
222 
            POISON_PILL
223 
        }
224 
    }
225 
}

Mutations

64

1.1
Location : <init>
Killed by : none removed call to com/pivovarit/collectors/Preconditions::requireValidExecutor → SURVIVED Covering tests

79

1.1
Location : <init>
Killed by : none removed call to com/pivovarit/collectors/Preconditions::requireValidExecutor → SURVIVED Covering tests

86

1.1
Location : start
Killed by : none negated conditional → TIMED_OUT

93

1.1
Location : lambda$start$0
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldNotExecuteTasksSubmittedAfterFailure()] negated conditional → KILLED

94

1.1
Location : lambda$start$0
Killed by : com.pivovarit.collectors.test.ExecutorPollutionTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.test.ExecutorPollutionTest]/[test-factory:shouldNotPolluteExecutorFactoryLimitedParallelism()]/[dynamic-test:#2] removed call to java/util/concurrent/Semaphore::acquire → KILLED

97

1.1
Location : lambda$start$0
Killed by : none removed call to com/pivovarit/collectors/Dispatcher::abort → NO_COVERAGE

101

1.1
Location : lambda$start$0
Killed by : none removed call to com/pivovarit/collectors/Dispatcher::retry → TIMED_OUT

2.2
Location : lambda$start$1
Killed by : none removed call to java/util/concurrent/Executor::execute → TIMED_OUT

103

1.1
Location : lambda$start$2
Killed by : none removed call to java/util/concurrent/FutureTask::run → TIMED_OUT

105

1.1
Location : lambda$start$2
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldNotLeakInterruptionIntoExecutorThreadOnSiblingCancellation()] negated conditional → KILLED

108

1.1
Location : lambda$start$2
Killed by : none negated conditional → TIMED_OUT

109

1.1
Location : lambda$start$2
Killed by : none removed call to java/util/concurrent/Semaphore::release → TIMED_OUT

114

1.1
Location : lambda$start$0
Killed by : none negated conditional → SURVIVED Covering tests

115

1.1
Location : lambda$start$0
Killed by : none removed call to java/util/concurrent/Semaphore::release → SURVIVED Covering tests

127

1.1
Location : lambda$start$0
Killed by : com.pivovarit.collectors.test.TaskRejectionTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.test.TaskRejectionTest]/[test-factory:shouldHandleRejection()]/[dynamic-test:#7] removed call to com/pivovarit/collectors/Dispatcher::abort → KILLED

130

1.1
Location : start
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldNotExecuteTasksSubmittedAfterFailure()] removed call to java/util/function/Consumer::accept → KILLED

131

1.1
Location : start
Killed by : none removed call to java/lang/Thread::start → TIMED_OUT

136

1.1
Location : stop
Killed by : none negated conditional → TIMED_OUT

145

1.1
Location : registerTerminationGuard
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldStopWhenTerminationGuardCleanedOnInterruptedThread()] replaced return value with null for com/pivovarit/collectors/Dispatcher::registerTerminationGuard → KILLED

146

1.1
Location : lambda$registerTerminationGuard$0
Killed by : none negated conditional → TIMED_OUT

153

1.1
Location : wasStarted
Killed by : none replaced boolean return with false for com/pivovarit/collectors/Dispatcher::wasStarted → SURVIVED Covering tests

160

1.1
Location : wasShutdown
Killed by : none replaced boolean return with false for com/pivovarit/collectors/Dispatcher::wasShutdown → TIMED_OUT

167

1.1
Location : submit
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldRejectSubmitAfterShutdown()] negated conditional → KILLED

175

1.1
Location : lambda$submit$0
Killed by : none removed call to com/pivovarit/collectors/Dispatcher::abort → TIMED_OUT

179

1.1
Location : submit
Killed by : none removed call to com/pivovarit/collectors/InterruptibleCompletableFuture::completedBy → TIMED_OUT

184

1.1
Location : submit
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldRejectSubmitAfterShutdown()] negated conditional → KILLED

185

1.1
Location : submit
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldNotExecuteTasksSubmittedAfterFailure()] removed call to com/pivovarit/collectors/Dispatcher::abort → KILLED

186

1.1
Location : submit
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldNotExecuteTasksSubmittedAfterFailure()] replaced return value with null for com/pivovarit/collectors/Dispatcher::submit → KILLED

189

1.1
Location : submit
Killed by : com.pivovarit.collectors.DispatcherTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.DispatcherTest]/[method:shouldNotExecuteTasksSubmittedAfterFailure()] replaced return value with null for com/pivovarit/collectors/Dispatcher::submit → KILLED

193

1.1
Location : abort
Killed by : none negated conditional → TIMED_OUT

195

1.1
Location : abort
Killed by : none removed call to com/pivovarit/collectors/Dispatcher::abort → TIMED_OUT

207

1.1
Location : retry
Killed by : none removed call to java/lang/Runnable::run → TIMED_OUT

209

1.1
Location : retry
Killed by : none removed call to java/lang/Thread::onSpinWait → SURVIVED Covering tests

210

1.1
Location : retry
Killed by : com.pivovarit.collectors.test.TaskRejectionTest.[engine:junit-jupiter]/[class:com.pivovarit.collectors.test.TaskRejectionTest]/[test-factory:shouldHandleRejection()]/[dynamic-test:#7] removed call to java/lang/Runnable::run → KILLED

Active mutators

Tests examined


Report generated by PIT 1.30.0 support