View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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   * Balanced queue executor with a fastpath. Because this is FIFO, it has no respect for
32   * ordering so a fast path skipping the queuing of Calls if an Handler is available, is possible.
33   * Just pass the Call direct to waiting Handler thread. Try to keep the hot Handlers bubbling
34   * rather than let them go cold and lose context. Idea taken from Apace Kudu (incubating). See
35   * https://gerrit.cloudera.org/#/c/2938/7/src/kudu/rpc/service_queue.h
36   */
37  @InterfaceAudience.Private
38  public class FastPathBalancedQueueRpcExecutor extends BalancedQueueRpcExecutor {
39    // Depends on default behavior of BalancedQueueRpcExecutor being FIFO!
40  
41    /*
42     * Stack of Handlers waiting for work.
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     * @return Pop a Handler instance if one available ready-to-go or else return null.
74     */
75    private FastPathHandler popReadyHandler() {
76      return this.fastPathHandlerStack.poll();
77    }
78  
79    class FastPathHandler extends Handler {
80      // Below are for fast-path support. Push this Handler on to the fastPathHandlerStack Deque
81      // if an empty queue of CallRunners so we are available for direct handoff when one comes in.
82      final Deque<FastPathHandler> fastPathHandlerStack;
83      // Semaphore to coordinate loading of fastpathed loadedTask and our running it.
84      private Semaphore semaphore = new Semaphore(0);
85      // The task we get when fast-pathing.
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        // Get a callrunner if one in the Q.
97        CallRunner cr = this.q.poll();
98        if (cr == null) {
99          // Else, if a fastPathHandlerStack present and no callrunner in Q, register ourselves for
100         // the fastpath handoff done via fastPathHandlerStack.
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           // No fastpath available. Block until a task comes available.
108           cr = super.getCallRunner();
109         }
110       }
111       return cr;
112     }
113 
114     /**
115      * @param task Task gotten via fastpath.
116      * @return True if we successfully loaded our task
117      */
118     boolean loadCallRunner(final CallRunner cr) {
119       this.loadedCallRunner = cr;
120       this.semaphore.release();
121       return true;
122     }
123   }
124 }