1 package org.apache.hadoop.hbase.ipc;
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
37
38
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
49
50
51
52
53
54
55 CallRunner(final RpcServerInterface rpcServer, final Call call, UserProvider userProvider) {
56 this.call = call;
57 this.rpcServer = rpcServer;
58
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
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
109
110 RequestContext.clear();
111 }
112 RpcServer.CurCall.set(null);
113 this.rpcServer.addCallSize(call.getSize() * -1);
114
115
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
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
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 }