1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.ipc;
19
20 import java.util.Deque;
21 import java.util.concurrent.BlockingQueue;
22 import java.util.concurrent.ConcurrentLinkedDeque;
23 import java.util.concurrent.Semaphore;
24 import java.util.concurrent.atomic.AtomicInteger;
25
26 import org.apache.hadoop.conf.Configuration;
27 import org.apache.hadoop.hbase.Abortable;
28 import org.apache.hadoop.hbase.classification.InterfaceAudience;
29
30
31
32
33
34
35
36
37 @InterfaceAudience.Private
38 public class FastPathBalancedQueueRpcExecutor extends BalancedQueueRpcExecutor {
39
40
41
42
43
44 private final Deque<FastPathHandler> fastPathHandlerStack = new ConcurrentLinkedDeque<>();
45
46 public FastPathBalancedQueueRpcExecutor(final String name, final int handlerCount,
47 final int maxQueueLength, final PriorityFunction priority, final Configuration conf,
48 final Abortable abortable) {
49 super(name, handlerCount, maxQueueLength, priority, conf, abortable);
50
51 }
52
53 public FastPathBalancedQueueRpcExecutor(final String name, final int handlerCount,
54 final String callQueueType, final int maxQueueLength, final PriorityFunction priority,
55 final Configuration conf, final Abortable abortable) {
56 super(name, handlerCount, callQueueType, maxQueueLength, priority, conf, abortable);
57 }
58
59 @Override
60 protected Handler getHandler(String name, double handlerFailureThreshhold,
61 BlockingQueue<CallRunner> q, AtomicInteger activeHandlerCount) {
62 return new FastPathHandler(name, handlerFailureThreshhold, q, activeHandlerCount,
63 fastPathHandlerStack);
64 }
65
66 @Override
67 public boolean dispatch(CallRunner callTask) throws InterruptedException {
68 FastPathHandler handler = popReadyHandler();
69 return handler != null? handler.loadCallRunner(callTask): super.dispatch(callTask);
70 }
71
72
73
74
75 private FastPathHandler popReadyHandler() {
76 return this.fastPathHandlerStack.poll();
77 }
78
79 class FastPathHandler extends Handler {
80
81
82 final Deque<FastPathHandler> fastPathHandlerStack;
83
84 private Semaphore semaphore = new Semaphore(0);
85
86 private CallRunner loadedCallRunner;
87
88 FastPathHandler(String name, double handlerFailureThreshhold, BlockingQueue<CallRunner> q,
89 final AtomicInteger activeHandlerCount,
90 final Deque<FastPathHandler> fastPathHandlerStack) {
91 super(name, handlerFailureThreshhold, q, activeHandlerCount);
92 this.fastPathHandlerStack = fastPathHandlerStack;
93 }
94
95 protected CallRunner getCallRunner() throws InterruptedException {
96
97 CallRunner cr = this.q.poll();
98 if (cr == null) {
99
100
101 if (this.fastPathHandlerStack != null) {
102 this.fastPathHandlerStack.push(this);
103 this.semaphore.acquire();
104 cr = this.loadedCallRunner;
105 this.loadedCallRunner = null;
106 } else {
107
108 cr = super.getCallRunner();
109 }
110 }
111 return cr;
112 }
113
114
115
116
117
118 boolean loadCallRunner(final CallRunner cr) {
119 this.loadedCallRunner = cr;
120 this.semaphore.release();
121 return true;
122 }
123 }
124 }