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