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
|
/*
* Copyright (C) 2008 The Guava Authors
*
* Licensed 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
*
* http://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 com.google.common.util.concurrent;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import junit.framework.TestCase;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Tests {@link SerializingExecutor}.
*
* @author JJ Furman
*/
public class SerializingExecutorTest extends TestCase {
private static class FakeExecutor implements Executor {
Queue<Runnable> tasks = Lists.newLinkedList();
@Override public void execute(Runnable command) {
tasks.add(command);
}
boolean hasNext() {
return !tasks.isEmpty();
}
void runNext() {
assertTrue("expected at least one task to run", hasNext());
tasks.remove().run();
}
}
private FakeExecutor fakePool;
private SerializingExecutor e;
@Override
public void setUp() {
fakePool = new FakeExecutor();
e = new SerializingExecutor(fakePool);
}
public void testSerializingNullExecutor_fails() {
try {
new SerializingExecutor(null);
fail("Should have failed with NullPointerException.");
} catch (NullPointerException expected) {
}
}
public void testBasics() {
final AtomicInteger totalCalls = new AtomicInteger();
Runnable intCounter = new Runnable() {
@Override
public void run() {
totalCalls.incrementAndGet();
}
};
assertFalse(fakePool.hasNext());
e.execute(intCounter);
assertTrue(fakePool.hasNext());
e.execute(intCounter);
assertEquals(0, totalCalls.get());
fakePool.runNext(); // run just 1 sub task...
assertEquals(2, totalCalls.get());
assertFalse(fakePool.hasNext());
// Check that execute can be safely repeated
e.execute(intCounter);
e.execute(intCounter);
e.execute(intCounter);
assertEquals(2, totalCalls.get());
fakePool.runNext();
assertEquals(5, totalCalls.get());
assertFalse(fakePool.hasNext());
}
public void testOrdering() {
final List<Integer> callOrder = Lists.newArrayList();
class FakeOp implements Runnable {
final int op;
FakeOp(int op) {
this.op = op;
}
@Override
public void run() {
callOrder.add(op);
}
}
e.execute(new FakeOp(0));
e.execute(new FakeOp(1));
e.execute(new FakeOp(2));
fakePool.runNext();
assertEquals(ImmutableList.of(0, 1, 2), callOrder);
}
public void testExceptions() {
final AtomicInteger numCalls = new AtomicInteger();
Runnable runMe = new Runnable() {
@Override
public void run() {
numCalls.incrementAndGet();
throw new RuntimeException("FAKE EXCEPTION!");
}
};
e.execute(runMe);
e.execute(runMe);
fakePool.runNext();
assertEquals(2, numCalls.get());
}
public void testDelegateRejection() {
final AtomicInteger numCalls = new AtomicInteger();
final AtomicBoolean reject = new AtomicBoolean(true);
final SerializingExecutor executor = new SerializingExecutor(
new Executor() {
@Override public void execute(Runnable r) {
if (reject.get()) {
throw new RejectedExecutionException();
}
r.run();
}
});
Runnable task = new Runnable() {
@Override
public void run() {
numCalls.incrementAndGet();
}
};
try {
executor.execute(task);
fail();
} catch (RejectedExecutionException expected) {}
assertEquals(0, numCalls.get());
reject.set(false);
executor.execute(task);
assertEquals(2, numCalls.get());
}
public void testTaskThrowsError() throws Exception {
class MyError extends Error {}
final CyclicBarrier barrier = new CyclicBarrier(2);
// we need to make sure the error gets thrown on a different thread.
ExecutorService service = Executors.newSingleThreadExecutor();
try {
final SerializingExecutor executor = new SerializingExecutor(service);
Runnable errorTask = new Runnable() {
@Override
public void run() {
throw new MyError();
}
};
Runnable barrierTask = new Runnable() {
@Override
public void run() {
try {
barrier.await();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
};
executor.execute(errorTask);
service.execute(barrierTask); // submit directly to the service
// the barrier task runs after the error task so we know that the error has been observed by
// SerializingExecutor by the time the barrier is satified
barrier.await(10, TimeUnit.SECONDS);
executor.execute(barrierTask);
// timeout means the second task wasn't even tried
barrier.await(10, TimeUnit.SECONDS);
} finally {
service.shutdown();
}
}
}
|
