Skip to content

Commit aeaed31

Browse files
benjchristensenbenjchristensen
committedFeb 15, 2013
Merge pull request #106 from benjchristensen/issue-102
HystrixThreadPool shutdown methods
2 parents 90d7713 + e2367fa commit aeaed31

File tree

6 files changed

+161
-13
lines changed

6 files changed

+161
-13
lines changed
 

‎hystrix-core/src/main/java/com/netflix/hystrix/Hystrix.java

+48
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
package com.netflix.hystrix;
2+
3+
import java.util.concurrent.TimeUnit;
4+
5+
/**
6+
* Lifecycle management of Hystrix.
7+
*/
8+
public class Hystrix {
9+
10+
/**
11+
* Reset state and release resources in use (such as thread-pools).
12+
* <p>
13+
* NOTE: This can result in race conditions if HystrixCommands are concurrently being executed.
14+
* </p>
15+
*/
16+
public static void reset() {
17+
// shutdown thread-pools
18+
HystrixThreadPool.Factory.shutdown();
19+
_reset();
20+
}
21+
22+
/**
23+
* Reset state and release resources in use (such as threadpools) and wait for completion.
24+
* <p>
25+
* NOTE: This can result in race conditions if HystrixCommands are concurrently being executed.
26+
* </p>
27+
*
28+
* @param time time to wait for thread-pools to shutdown
29+
* @param unit {@link TimeUnit} for <pre>time</pre> to wait for thread-pools to shutdown
30+
*/
31+
public static void reset(long time, TimeUnit unit) {
32+
// shutdown thread-pools
33+
HystrixThreadPool.Factory.shutdown(time, unit);
34+
_reset();
35+
}
36+
37+
/**
38+
* Reset logic that doesn't have time/TimeUnit arguments.
39+
*/
40+
private static void _reset() {
41+
// clear metrics
42+
HystrixCommandMetrics.reset();
43+
// clear collapsers
44+
HystrixCollapser.reset();
45+
// clear circuit breakers
46+
HystrixCircuitBreaker.Factory.reset();
47+
}
48+
}

‎hystrix-core/src/main/java/com/netflix/hystrix/HystrixCircuitBreaker.java

+7
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,13 @@ public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCo
113113
public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {
114114
return circuitBreakersByCommand.get(key.name());
115115
}
116+
117+
/**
118+
* Clears all circuit breakers. If new requests come in instances will be recreated.
119+
*/
120+
/* package */ static void reset() {
121+
circuitBreakersByCommand.clear();
122+
}
116123
}
117124

118125
/**

‎hystrix-core/src/main/java/com/netflix/hystrix/HystrixCollapser.java

+10-12
Original file line numberDiff line numberDiff line change
@@ -346,18 +346,6 @@ public Future<ResponseType> queue() {
346346
return response;
347347
}
348348

349-
/**
350-
* Reset the global and request-scoped state for the given collapser key.
351-
*
352-
* This is intended to support uses of collapsers in an environment like a REPL
353-
* where the definition of a collapser may change. It is not thread-safe and should
354-
* not be used in a production environment.
355-
*/
356-
public static void resetCollapser(HystrixCollapserKey key) {
357-
globalScopedCollapsers.remove(key.name());
358-
requestScopedCollapsers.remove(key.name());
359-
}
360-
361349
/**
362350
* Static global cache of RequestCollapsers for Scope.GLOBAL
363351
*/
@@ -994,6 +982,16 @@ protected String getCacheKey() {
994982
return null;
995983
}
996984

985+
/**
986+
* Clears all state. If new requests come in instances will be recreated and metrics started from scratch.
987+
*/
988+
/* package */ static void reset() {
989+
defaultNameCache.clear();
990+
globalScopedCollapsers.clear();
991+
requestScopedCollapsers.clear();
992+
RealCollapserTimer.timer.reset();
993+
}
994+
997995
private static String getDefaultNameFromClass(@SuppressWarnings("rawtypes") Class<? extends HystrixCollapser> cls) {
998996
String fromCache = defaultNameCache.get(cls);
999997
if (fromCache != null) {

‎hystrix-core/src/main/java/com/netflix/hystrix/HystrixCommandMetrics.java

+7
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,13 @@ public static Collection<HystrixCommandMetrics> getInstances() {
100100
return Collections.unmodifiableCollection(metrics.values());
101101
}
102102

103+
/**
104+
* Clears all state from metrics. If new requests come in instances will be recreated and metrics started from scratch.
105+
*/
106+
/* package */ static void reset() {
107+
metrics.clear();
108+
}
109+
103110
private final HystrixCommandProperties properties;
104111
private final HystrixRollingNumber counter;
105112
private final HystrixRollingPercentile percentileExecution;

‎hystrix-core/src/main/java/com/netflix/hystrix/HystrixThreadPool.java

+78-1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,9 @@
2626
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
2727
import com.netflix.hystrix.strategy.metrics.HystrixMetricsPublisherFactory;
2828
import com.netflix.hystrix.strategy.properties.HystrixPropertiesFactory;
29+
import org.junit.Test;
30+
31+
import static org.junit.Assert.*;
2932

3033
/**
3134
* ThreadPool used to executed {@link HystrixCommand#run()} on separate threads when configured to do so with {@link HystrixCommandProperties#executionIsolationStrategy()}.
@@ -113,13 +116,48 @@ public interface HystrixThreadPool {
113116
return poolForKey;
114117
}
115118
}
119+
120+
/**
121+
* Initiate the shutdown of all {@link HystrixThreadPool} instances.
122+
* <p>
123+
* NOTE: This is NOT thread-safe if HystrixCommands are concurrently being executed
124+
* and causing thread-pools to initialize while also trying to shutdown.
125+
* </p>
126+
*/
127+
/* package */ static synchronized void shutdown() {
128+
for (HystrixThreadPool pool : threadPools.values()) {
129+
pool.getExecutor().shutdown();
130+
}
131+
threadPools.clear();
132+
}
133+
134+
/**
135+
* Initiate the shutdown of all {@link HystrixThreadPool} instances and wait up to the given time on each pool to complete.
136+
* <p>
137+
* NOTE: This is NOT thread-safe if HystrixCommands are concurrently being executed
138+
* and causing thread-pools to initialize while also trying to shutdown.
139+
* </p>
140+
*/
141+
/* package */ static synchronized void shutdown(long timeout, TimeUnit unit) {
142+
for (HystrixThreadPool pool : threadPools.values()) {
143+
pool.getExecutor().shutdown();
144+
}
145+
for (HystrixThreadPool pool : threadPools.values()) {
146+
try {
147+
pool.getExecutor().awaitTermination(timeout, unit);
148+
} catch (InterruptedException e) {
149+
throw new RuntimeException("Interrupted while waiting for thread-pools to terminate. Pools may not be correctly shutdown or cleared.", e);
150+
}
151+
}
152+
threadPools.clear();
153+
}
116154
}
117155

118156
/**
119157
* @ExcludeFromJavadoc
120158
*/
121159
@ThreadSafe
122-
/* package */static class HystrixThreadPoolDefault implements HystrixThreadPool {
160+
/* package */ static class HystrixThreadPoolDefault implements HystrixThreadPool {
123161
private final HystrixThreadPoolProperties properties;
124162
private final BlockingQueue<Runnable> queue;
125163
private final ThreadPoolExecutor threadPool;
@@ -174,4 +212,43 @@ public boolean isQueueSpaceAvailable() {
174212

175213
}
176214

215+
public static class UnitTest {
216+
217+
@Test
218+
public void testShutdown() {
219+
// other unit tests will probably have run before this so get the count
220+
int count = Factory.threadPools.size();
221+
222+
HystrixThreadPool pool = Factory.getInstance(HystrixThreadPoolKey.Factory.asKey("threadPoolFactoryTest"),
223+
HystrixThreadPoolProperties.Setter.getUnitTestPropertiesBuilder());
224+
225+
assertEquals(count + 1, Factory.threadPools.size());
226+
assertFalse(pool.getExecutor().isShutdown());
227+
228+
Factory.shutdown();
229+
230+
// ensure all pools were removed from the cache
231+
assertEquals(0, Factory.threadPools.size());
232+
assertTrue(pool.getExecutor().isShutdown());
233+
}
234+
235+
@Test
236+
public void testShutdownWithWait() {
237+
// other unit tests will probably have run before this so get the count
238+
int count = Factory.threadPools.size();
239+
240+
HystrixThreadPool pool = Factory.getInstance(HystrixThreadPoolKey.Factory.asKey("threadPoolFactoryTest"),
241+
HystrixThreadPoolProperties.Setter.getUnitTestPropertiesBuilder());
242+
243+
assertEquals(count + 1, Factory.threadPools.size());
244+
assertFalse(pool.getExecutor().isShutdown());
245+
246+
Factory.shutdown(1, TimeUnit.SECONDS);
247+
248+
// ensure all pools were removed from the cache
249+
assertEquals(0, Factory.threadPools.size());
250+
assertTrue(pool.getExecutor().isShutdown());
251+
}
252+
}
253+
177254
}

‎hystrix-core/src/main/java/com/netflix/hystrix/util/HystrixTimer.java

+11
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,17 @@ public static HystrixTimer getInstance() {
6666
return INSTANCE;
6767
}
6868

69+
/**
70+
* Clears all listeners.
71+
* <p>
72+
* NOTE: This will result in race conditions if {@link #addTimerListener(com.netflix.hystrix.util.HystrixTimer.TimerListener)} is being concurrently called.
73+
* </p>
74+
*/
75+
public static void reset() {
76+
INSTANCE.listenersPerInterval.clear();
77+
INSTANCE.intervals.clear();
78+
}
79+
6980
private TickThread tickThread = new TickThread();
7081
private ConcurrentHashMap<Integer, ConcurrentLinkedQueue<Reference<TimerListener>>> listenersPerInterval = new ConcurrentHashMap<Integer, ConcurrentLinkedQueue<Reference<TimerListener>>>();
7182
private ConcurrentLinkedQueue<TimerInterval> intervals = new ConcurrentLinkedQueue<TimerInterval>();

0 commit comments

Comments
 (0)