View Javadoc

1   package org.apache.hadoop.hbase.ipc;
2   /**
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  import java.nio.channels.ClosedChannelException;
20  
21  import org.apache.hadoop.classification.InterfaceAudience;
22  import org.apache.hadoop.hbase.CellScanner;
23  import org.apache.hadoop.hbase.ipc.RpcServer.Call;
24  import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler;
25  import org.apache.hadoop.hbase.monitoring.TaskMonitor;
26  import org.apache.hadoop.hbase.security.UserProvider;
27  import org.apache.hadoop.hbase.util.Pair;
28  import org.apache.hadoop.security.UserGroupInformation;
29  import org.apache.hadoop.util.StringUtils;
30  import org.cloudera.htrace.Trace;
31  import org.cloudera.htrace.TraceScope;
32  
33  import com.google.protobuf.Message;
34  
35  /**
36   * The request processing logic, which is usually executed in thread pools provided by an
37   * {@link RpcScheduler}.  Call {@link #run()} to actually execute the contained
38   * {@link RpcServer.Call}
39   */
40  @InterfaceAudience.Private
41  public class CallRunner {
42    private final Call call;
43    private final RpcServerInterface rpcServer;
44    private final MonitoredRPCHandler status;
45    private UserProvider userProvider;
46  
47    /**
48     * On construction, adds the size of this call to the running count of outstanding call sizes.
49     * Presumption is that we are put on a queue while we wait on an executor to run us.  During this
50     * time we occupy heap.
51     * @param call The call to run.
52     * @param rpcServer
53     */
54    // The constructor is shutdown so only RpcServer in this class can make one of these.
55    CallRunner(final RpcServerInterface rpcServer, final Call call, UserProvider userProvider) {
56      this.call = call;
57      this.rpcServer = rpcServer;
58      // Add size of the call to queue size.
59      this.rpcServer.addCallSize(call.getSize());
60      this.status = getStatus();
61      this.userProvider = userProvider;
62    }
63  
64    public Call getCall() {
65      return call;
66    }
67  
68    public void run() {
69      try {
70        if (!call.connection.channel.isOpen()) {
71          if (RpcServer.LOG.isDebugEnabled()) {
72            RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call);
73          }
74          return;
75        }
76        this.status.setStatus("Setting up call");
77        this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort());
78        if (RpcServer.LOG.isDebugEnabled()) {
79          UserGroupInformation remoteUser = call.connection.user;
80          RpcServer.LOG.debug(call.toShortString() + " executing as " +
81              ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName()));
82        }
83        Throwable errorThrowable = null;
84        String error = null;
85        Pair<Message, CellScanner> resultPair = null;
86        RpcServer.CurCall.set(call);
87        TraceScope traceScope = null;
88        try {
89          if (!this.rpcServer.isStarted()) {
90            throw new ServerNotRunningYetException("Server is not running yet");
91          }
92          if (call.tinfo != null) {
93            traceScope = Trace.startSpan(call.toTraceString(), call.tinfo);
94          }
95          RequestContext.set(userProvider.create(call.connection.user), RpcServer.getRemoteIp(),
96            call.connection.service);
97          // make the call
98          resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner,
99            call.timestamp, this.status);
100       } catch (Throwable e) {
101         RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e);
102         errorThrowable = e;
103         error = StringUtils.stringifyException(e);
104       } finally {
105         if (traceScope != null) {
106           traceScope.close();
107         }
108         // Must always clear the request context to avoid leaking
109         // credentials between requests.
110         RequestContext.clear();
111       }
112       RpcServer.CurCall.set(null);
113       this.rpcServer.addCallSize(call.getSize() * -1);
114       // Set the response for undelayed calls and delayed calls with
115       // undelayed responses.
116       if (!call.isDelayed() || !call.isReturnValueDelayed()) {
117         Message param = resultPair != null ? resultPair.getFirst() : null;
118         CellScanner cells = resultPair != null ? resultPair.getSecond() : null;
119         call.setResponse(param, cells, errorThrowable, error);
120       }
121       call.sendResponseIfReady();
122       this.status.markComplete("Sent response");
123       this.status.pause("Waiting for a call");
124     } catch (OutOfMemoryError e) {
125       if (this.rpcServer.getErrorHandler() != null) {
126         if (this.rpcServer.getErrorHandler().checkOOME(e)) {
127           RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError");
128           return;
129         }
130       } else {
131         // rethrow if no handler
132         throw e;
133       }
134     } catch (ClosedChannelException cce) {
135       RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " +
136           "this means that the server was processing a " +
137           "request but the client went away. The error message was: " +
138           cce.getMessage());
139     } catch (Exception e) {
140       RpcServer.LOG.warn(Thread.currentThread().getName()
141           + ": caught: " + StringUtils.stringifyException(e));
142     }
143   }
144 
145   MonitoredRPCHandler getStatus() {
146     // It is ugly the way we park status up in RpcServer.  Let it be for now.  TODO.
147     MonitoredRPCHandler status = RpcServer.MONITORED_RPC.get();
148     if (status != null) {
149       return status;
150     }
151     status = TaskMonitor.get().createRPCStatus(Thread.currentThread().getName());
152     status.pause("Waiting for a call");
153     RpcServer.MONITORED_RPC.set(status);
154     return status;
155   }
156 }