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.net.InetSocketAddress;
20 import java.nio.channels.ClosedChannelException;
21
22 import org.apache.hadoop.hbase.CellScanner;
23 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
24 import org.apache.hadoop.hbase.classification.InterfaceAudience;
25 import org.apache.hadoop.hbase.classification.InterfaceStability;
26 import org.apache.hadoop.hbase.ipc.RpcServer.Call;
27 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler;
28 import org.apache.hadoop.hbase.util.Pair;
29 import org.apache.hadoop.security.UserGroupInformation;
30 import org.apache.hadoop.util.StringUtils;
31 import org.apache.htrace.Trace;
32 import org.apache.htrace.TraceScope;
33
34 import com.google.protobuf.Message;
35
36
37
38
39
40
41 @InterfaceAudience.LimitedPrivate({HBaseInterfaceAudience.COPROC, HBaseInterfaceAudience.PHOENIX})
42 @InterfaceStability.Evolving
43 public class CallRunner {
44 private Call call;
45 private RpcServerInterface rpcServer;
46 private MonitoredRPCHandler status;
47
48
49
50
51
52
53
54 CallRunner(final RpcServerInterface rpcServer, final Call call) {
55 this.call = call;
56 this.rpcServer = rpcServer;
57
58 this.rpcServer.addCallSize(call.getSize());
59 }
60
61 public Call getCall() {
62 return call;
63 }
64
65 public void setStatus(MonitoredRPCHandler status) {
66 this.status = status;
67 }
68
69
70
71
72 private void cleanup() {
73 this.call = null;
74 this.rpcServer = null;
75 }
76
77 public void run() {
78 try {
79 if (!call.connection.channel.isOpen()) {
80 if (RpcServer.LOG.isDebugEnabled()) {
81 RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call);
82 }
83 return;
84 }
85 this.status.setStatus("Setting up call");
86 this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort());
87 if (RpcServer.LOG.isTraceEnabled()) {
88 UserGroupInformation remoteUser = call.connection.user;
89 RpcServer.LOG.trace(call.toShortString() + " executing as " +
90 ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName()));
91 }
92 Throwable errorThrowable = null;
93 String error = null;
94 Pair<Message, CellScanner> resultPair = null;
95 RpcServer.CurCall.set(call);
96 TraceScope traceScope = null;
97 try {
98 if (!this.rpcServer.isStarted()) {
99 InetSocketAddress address = rpcServer.getListenerAddress();
100 throw new ServerNotRunningYetException("Server " +
101 (address != null ? address : "(channel closed)") + " is not running yet");
102 }
103 if (call.tinfo != null) {
104 traceScope = Trace.startSpan(call.toTraceString(), call.tinfo);
105 }
106
107 resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner,
108 call.timestamp, this.status);
109 } catch (Throwable e) {
110 RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e);
111 errorThrowable = e;
112 error = StringUtils.stringifyException(e);
113 if (e instanceof Error) {
114 throw (Error)e;
115 }
116 } finally {
117 if (traceScope != null) {
118 traceScope.close();
119 }
120 RpcServer.CurCall.set(null);
121 }
122
123
124 if (!call.isDelayed() || !call.isReturnValueDelayed()) {
125 Message param = resultPair != null ? resultPair.getFirst() : null;
126 CellScanner cells = resultPair != null ? resultPair.getSecond() : null;
127 call.setResponse(param, cells, errorThrowable, error);
128 }
129 call.sendResponseIfReady();
130 this.status.markComplete("Sent response");
131 this.status.pause("Waiting for a call");
132 } catch (OutOfMemoryError e) {
133 if (this.rpcServer.getErrorHandler() != null) {
134 if (this.rpcServer.getErrorHandler().checkOOME(e)) {
135 RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError");
136 return;
137 }
138 } else {
139
140 throw e;
141 }
142 } catch (ClosedChannelException cce) {
143 InetSocketAddress address = rpcServer.getListenerAddress();
144 RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " +
145 "this means that the server " + (address != null ? address : "(channel closed)") +
146 " was processing a request but the client went away. The error message was: " +
147 cce.getMessage());
148 } catch (Exception e) {
149 RpcServer.LOG.warn(Thread.currentThread().getName()
150 + ": caught: " + StringUtils.stringifyException(e));
151 } finally {
152
153 this.rpcServer.addCallSize(call.getSize() * -1);
154 cleanup();
155 }
156 }
157 }