-
-
Notifications
You must be signed in to change notification settings - Fork 15.8k
/
MultithreadEventExecutorGroup.java
293 lines (263 loc) · 11.3 KB
/
MultithreadEventExecutorGroup.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
/*
* Copyright 2012 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.util.concurrent;
import io.netty.util.internal.EmptyArrays;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import static io.netty.util.internal.ObjectUtil.checkPositive;
/**
* {@link EventExecutorGroup} implementation that handles their tasks with multiple threads at
* the same time.
*/
public class MultithreadEventExecutorGroup implements EventExecutorGroup {
private final EventExecutor[] children;
private final List<EventExecutor> readonlyChildren;
private final AtomicInteger terminatedChildren = new AtomicInteger();
private final Promise<?> terminationFuture = GlobalEventExecutor.INSTANCE.newPromise();
private final boolean powerOfTwo;
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param threadFactory the ThreadFactory to use, or {@code null} if the default should be used.
*/
public MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory) {
this(nThreads, threadFactory, SingleThreadEventExecutor.DEFAULT_MAX_PENDING_EXECUTOR_TASKS,
RejectedExecutionHandlers.reject());
}
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param executor the {@link Executor} to use, or {@code null} if the default should be used.
*/
public MultithreadEventExecutorGroup(int nThreads, Executor executor) {
this(nThreads, executor, SingleThreadEventExecutor.DEFAULT_MAX_PENDING_EXECUTOR_TASKS,
RejectedExecutionHandlers.reject());
}
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param threadFactory the {@link ThreadFactory} to use, or {@code null} if the default should be used.
* @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
* @param rejectedHandler the {@link RejectedExecutionHandler} to use.
*/
public MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory,
int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {
this(nThreads, threadFactory, maxPendingTasks, rejectedHandler, EmptyArrays.EMPTY_OBJECTS);
}
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param executor the Executor to use, or {@code null} if the default should be used.
* @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
* @param rejectedHandler the {@link RejectedExecutionHandler} to use.
*/
public MultithreadEventExecutorGroup(int nThreads, Executor executor,
int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {
this(nThreads, executor, maxPendingTasks, rejectedHandler, EmptyArrays.EMPTY_OBJECTS);
}
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param threadFactory the ThreadFactory to use, or {@code null} if the default should be used.
* @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
* @param rejectedHandler the {@link RejectedExecutionHandler} to use.
* @param args arguments which will passed to each {@link #newChild(Executor, int,
* RejectedExecutionHandler, Object...)} call
*/
protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, int maxPendingTasks,
RejectedExecutionHandler rejectedHandler, Object... args) {
this(nThreads, threadFactory == null ? null : new ThreadPerTaskExecutor(threadFactory),
maxPendingTasks, rejectedHandler, args);
}
/**
* Create a new instance.
*
* @param nThreads the number of threads that will be used by this instance.
* @param executor the Executor to use, or {@code null} if the default should be used.
* @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
* @param rejectedHandler the {@link RejectedExecutionHandler} to use.
* @param args arguments which will passed to each {@link #newChild(Executor, int,
* RejectedExecutionHandler, Object...)} call
*/
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, int maxPendingTasks,
RejectedExecutionHandler rejectedHandler, Object... args) {
checkPositive(nThreads, "nThreads");
if (executor == null) {
executor = new ThreadPerTaskExecutor(new DefaultThreadFactory(getClass()));
}
children = new EventExecutor[nThreads];
powerOfTwo = isPowerOfTwo(children.length);
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(executor, maxPendingTasks, rejectedHandler, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event executor", e);
} finally {
if (!success) {
for (int j = 0; j < i; j ++) {
children[j].shutdownGracefully();
}
for (int j = 0; j < i; j ++) {
EventExecutor e = children[j];
try {
while (!e.isTerminated()) {
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
}
} catch (InterruptedException interrupted) {
// Let the caller handle the interruption.
Thread.currentThread().interrupt();
break;
}
}
}
}
}
final FutureListener<Object> terminationListener = future -> {
if (terminatedChildren.incrementAndGet() == children.length) {
terminationFuture.setSuccess(null);
}
};
for (EventExecutor e: children) {
e.terminationFuture().addListener(terminationListener);
}
readonlyChildren = Collections.unmodifiableList(Arrays.asList(children));
}
// Use a 'long' counter to avoid non-round-robin behaviour at the 32-bit overflow boundary.
// The 64-bit long solves this by placing the overflow so far into the future, that no system
// will encounter this in practice.
private final AtomicLong idx = new AtomicLong();
/**
* The {@link EventExecutor}s that are used by this {@link MultithreadEventExecutorGroup}.
*/
protected final List<EventExecutor> executors() {
return readonlyChildren;
}
/**
* Returns the next {@link EventExecutor} to use. The default implementation will use round-robin, but you may
* override this to change the selection algorithm.
*/
@Override
public EventExecutor next() {
if (powerOfTwo) {
return children[(int) idx.getAndIncrement() & children.length - 1];
}
return children[(int) Math.abs(idx.getAndIncrement() % children.length)];
}
private static boolean isPowerOfTwo(int val) {
return (val & -val) == val;
}
@Override
public Iterator<EventExecutor> iterator() {
return executors().iterator();
}
/**
* Return the number of {@link EventExecutor} this implementation uses. This number is the maps
* 1:1 to the threads it use.
*/
public final int executorCount() {
return executors().size();
}
/**
* Create a new EventExecutor which will later then accessible via the {@link #next()} method. This method will be
* called for each thread that will serve this {@link MultithreadEventExecutorGroup}.
*
* As this method is called from within the constructor you can only use the parameters passed into the method when
* overriding this method.
*/
protected EventExecutor newChild(Executor executor, int maxPendingTasks,
RejectedExecutionHandler rejectedExecutionHandler,
Object... args) {
assert args.length == 0;
return new SingleThreadEventExecutor(executor, maxPendingTasks, rejectedExecutionHandler);
}
@Override
public final Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
for (EventExecutor l: children) {
l.shutdownGracefully(quietPeriod, timeout, unit);
}
return terminationFuture();
}
@Override
public final Future<?> terminationFuture() {
return terminationFuture.asFuture();
}
@Override
@Deprecated
public final void shutdown() {
for (EventExecutor l: children) {
l.shutdown();
}
}
@Override
public final boolean isShuttingDown() {
for (EventExecutor l: children) {
if (!l.isShuttingDown()) {
return false;
}
}
return true;
}
@Override
public final boolean isShutdown() {
for (EventExecutor l: children) {
if (!l.isShutdown()) {
return false;
}
}
return true;
}
@Override
public final boolean isTerminated() {
for (EventExecutor l: children) {
if (!l.isTerminated()) {
return false;
}
}
return true;
}
@Override
public final boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long deadline = System.nanoTime() + unit.toNanos(timeout);
loop: for (EventExecutor l: children) {
for (;;) {
long timeLeft = deadline - System.nanoTime();
if (timeLeft <= 0) {
break loop;
}
if (l.awaitTermination(timeLeft, TimeUnit.NANOSECONDS)) {
break;
}
}
}
return isTerminated();
}
}