001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.ipc; 019 020import static org.apache.hadoop.hbase.ipc.IPCUtil.buildRequestHeader; 021import static org.apache.hadoop.hbase.ipc.IPCUtil.createRemoteException; 022import static org.apache.hadoop.hbase.ipc.IPCUtil.getTotalSizeWhenWrittenDelimited; 023import static org.apache.hadoop.hbase.ipc.IPCUtil.isFatalConnectionException; 024import static org.apache.hadoop.hbase.ipc.IPCUtil.setCancelled; 025import static org.apache.hadoop.hbase.ipc.IPCUtil.write; 026 027import io.opentelemetry.context.Scope; 028import java.io.BufferedInputStream; 029import java.io.BufferedOutputStream; 030import java.io.DataInputStream; 031import java.io.DataOutputStream; 032import java.io.IOException; 033import java.io.InputStream; 034import java.io.InterruptedIOException; 035import java.io.OutputStream; 036import java.net.InetSocketAddress; 037import java.net.Socket; 038import java.net.SocketTimeoutException; 039import java.security.PrivilegedExceptionAction; 040import java.util.ArrayDeque; 041import java.util.Locale; 042import java.util.Queue; 043import java.util.concurrent.ConcurrentHashMap; 044import java.util.concurrent.ConcurrentMap; 045import java.util.concurrent.ThreadLocalRandom; 046import javax.security.sasl.SaslException; 047import org.apache.hadoop.conf.Configuration; 048import org.apache.hadoop.hbase.CellScanner; 049import org.apache.hadoop.hbase.DoNotRetryIOException; 050import org.apache.hadoop.hbase.exceptions.ConnectionClosingException; 051import org.apache.hadoop.hbase.io.ByteArrayOutputStream; 052import org.apache.hadoop.hbase.ipc.HBaseRpcController.CancellationCallback; 053import org.apache.hadoop.hbase.log.HBaseMarkers; 054import org.apache.hadoop.hbase.security.HBaseSaslRpcClient; 055import org.apache.hadoop.hbase.security.SaslUtil; 056import org.apache.hadoop.hbase.security.SaslUtil.QualityOfProtection; 057import org.apache.hadoop.hbase.security.provider.SaslClientAuthenticationProvider; 058import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 059import org.apache.hadoop.hbase.util.ExceptionUtil; 060import org.apache.hadoop.io.IOUtils; 061import org.apache.hadoop.ipc.RemoteException; 062import org.apache.hadoop.net.NetUtils; 063import org.apache.hadoop.security.UserGroupInformation; 064import org.apache.hadoop.util.StringUtils; 065import org.apache.yetus.audience.InterfaceAudience; 066import org.slf4j.Logger; 067import org.slf4j.LoggerFactory; 068 069import org.apache.hbase.thirdparty.com.google.protobuf.Message; 070import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback; 071import org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf; 072import org.apache.hbase.thirdparty.io.netty.buffer.PooledByteBufAllocator; 073 074import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 075import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos; 076import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.CellBlockMeta; 077import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ConnectionHeader; 078import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ExceptionResponse; 079import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader; 080import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ResponseHeader; 081 082/** 083 * Thread that reads responses and notifies callers. Each connection owns a socket connected to a 084 * remote address. Calls are multiplexed through this socket: responses may be delivered out of 085 * order. 086 */ 087@InterfaceAudience.Private 088class BlockingRpcConnection extends RpcConnection implements Runnable { 089 090 private static final Logger LOG = LoggerFactory.getLogger(BlockingRpcConnection.class); 091 092 private final BlockingRpcClient rpcClient; 093 094 private final String threadName; 095 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "IS2_INCONSISTENT_SYNC", 096 justification = "We are always under lock actually") 097 private Thread thread; 098 099 // connected socket. protected for writing UT. 100 protected Socket socket = null; 101 private DataInputStream in; 102 private DataOutputStream out; 103 104 private HBaseSaslRpcClient saslRpcClient; 105 106 // currently active calls 107 private final ConcurrentMap<Integer, Call> calls = new ConcurrentHashMap<>(); 108 109 private final CallSender callSender; 110 111 private boolean closed = false; 112 113 private byte[] connectionHeaderPreamble; 114 115 private byte[] connectionHeaderWithLength; 116 117 private boolean waitingConnectionHeaderResponse = false; 118 119 /** 120 * If the client wants to interrupt its calls easily (i.e. call Thread#interrupt), it gets into a 121 * java issue: an interruption during a write closes the socket/channel. A way to avoid this is to 122 * use a different thread for writing. This way, on interruptions, we either cancel the writes or 123 * ignore the answer if the write is already done, but we don't stop the write in the middle. This 124 * adds a thread per region server in the client, so it's kept as an option. 125 * <p> 126 * The implementation is simple: the client threads adds their call to the queue, and then wait 127 * for an answer. The CallSender blocks on the queue, and writes the calls one after the other. On 128 * interruption, the client cancels its call. The CallSender checks that the call has not been 129 * canceled before writing it. 130 * </p> 131 * When the connection closes, all the calls not yet sent are dismissed. The client thread is 132 * notified with an appropriate exception, as if the call was already sent but the answer not yet 133 * received. 134 * </p> 135 */ 136 private class CallSender extends Thread { 137 138 private final Queue<Call> callsToWrite; 139 140 private final int maxQueueSize; 141 142 public CallSender(String name, Configuration conf) { 143 int queueSize = conf.getInt("hbase.ipc.client.write.queueSize", 1000); 144 callsToWrite = new ArrayDeque<>(queueSize); 145 this.maxQueueSize = queueSize; 146 setDaemon(true); 147 setName(name + " - writer"); 148 } 149 150 public void sendCall(final Call call) throws IOException { 151 if (callsToWrite.size() >= maxQueueSize) { 152 throw new IOException("Can't add " + call.toShortString() 153 + " to the write queue. callsToWrite.size()=" + callsToWrite.size()); 154 } 155 callsToWrite.offer(call); 156 BlockingRpcConnection.this.notifyAll(); 157 } 158 159 public void remove(Call call) { 160 callsToWrite.remove(call); 161 // By removing the call from the expected call list, we make the list smaller, but 162 // it means as well that we don't know how many calls we cancelled. 163 calls.remove(call.id); 164 call.setException(new CallCancelledException(call.toShortString() + ", waitTime=" 165 + (EnvironmentEdgeManager.currentTime() - call.getStartTime()) + ", rpcTimeout=" 166 + call.timeout)); 167 } 168 169 /** 170 * Reads the call from the queue, write them on the socket. 171 */ 172 @Override 173 public void run() { 174 synchronized (BlockingRpcConnection.this) { 175 while (!closed) { 176 if (callsToWrite.isEmpty()) { 177 // We should use another monitor object here for better performance since the read 178 // thread also uses ConnectionImpl.this. But this makes the locking schema more 179 // complicated, can do it later as an optimization. 180 try { 181 BlockingRpcConnection.this.wait(); 182 } catch (InterruptedException e) { 183 // Restore interrupt status 184 Thread.currentThread().interrupt(); 185 } 186 // check if we need to quit, so continue the main loop instead of fallback. 187 continue; 188 } 189 Call call = callsToWrite.poll(); 190 if (call.isDone()) { 191 continue; 192 } 193 try (Scope scope = call.span.makeCurrent()) { 194 writeRequest(call); 195 } catch (IOException e) { 196 // exception here means the call has not been added to the pendingCalls yet, so we need 197 // to fail it by our own. 198 LOG.debug("call write error for {}", call.toShortString()); 199 call.setException(e); 200 closeConn(e); 201 } 202 } 203 } 204 } 205 206 /** 207 * Cleans the call not yet sent when we finish. 208 */ 209 public void cleanup(IOException e) { 210 IOException ie = 211 new ConnectionClosingException("Connection to " + remoteId.getAddress() + " is closing."); 212 for (Call call : callsToWrite) { 213 call.setException(ie); 214 } 215 callsToWrite.clear(); 216 } 217 } 218 219 BlockingRpcConnection(BlockingRpcClient rpcClient, ConnectionId remoteId) throws IOException { 220 super(rpcClient.conf, AbstractRpcClient.WHEEL_TIMER, remoteId, rpcClient.clusterId, 221 rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.codec, rpcClient.compressor, 222 rpcClient.metrics); 223 this.rpcClient = rpcClient; 224 this.connectionHeaderPreamble = getConnectionHeaderPreamble(); 225 ConnectionHeader header = getConnectionHeader(); 226 ByteArrayOutputStream baos = new ByteArrayOutputStream(4 + header.getSerializedSize()); 227 DataOutputStream dos = new DataOutputStream(baos); 228 dos.writeInt(header.getSerializedSize()); 229 header.writeTo(dos); 230 assert baos.size() == 4 + header.getSerializedSize(); 231 this.connectionHeaderWithLength = baos.getBuffer(); 232 233 UserGroupInformation ticket = remoteId.ticket.getUGI(); 234 this.threadName = "BRPC Connection (" + this.rpcClient.socketFactory.hashCode() + ") to " 235 + remoteId.getAddress().toString() 236 + ((ticket == null) ? " from an unknown user" : (" from " + ticket.getUserName())); 237 238 if (this.rpcClient.conf.getBoolean(BlockingRpcClient.SPECIFIC_WRITE_THREAD, false)) { 239 callSender = new CallSender(threadName, this.rpcClient.conf); 240 callSender.start(); 241 } else { 242 callSender = null; 243 } 244 } 245 246 // protected for write UT. 247 protected void setupConnection() throws IOException { 248 short ioFailures = 0; 249 short timeoutFailures = 0; 250 while (true) { 251 try { 252 this.socket = this.rpcClient.socketFactory.createSocket(); 253 this.socket.setTcpNoDelay(this.rpcClient.isTcpNoDelay()); 254 this.socket.setKeepAlive(this.rpcClient.tcpKeepAlive); 255 if (this.rpcClient.localAddr != null) { 256 this.socket.bind(this.rpcClient.localAddr); 257 } 258 InetSocketAddress remoteAddr = getRemoteInetAddress(rpcClient.metrics); 259 NetUtils.connect(this.socket, remoteAddr, this.rpcClient.connectTO); 260 this.socket.setSoTimeout(this.rpcClient.readTO); 261 return; 262 } catch (SocketTimeoutException toe) { 263 /* 264 * The max number of retries is 45, which amounts to 20s*45 = 15 minutes retries. 265 */ 266 if (LOG.isDebugEnabled()) { 267 LOG.debug( 268 "Received exception in connection setup.\n" + StringUtils.stringifyException(toe)); 269 } 270 handleConnectionFailure(timeoutFailures++, this.rpcClient.maxRetries, toe); 271 } catch (IOException ie) { 272 if (LOG.isDebugEnabled()) { 273 LOG.debug( 274 "Received exception in connection setup.\n" + StringUtils.stringifyException(ie)); 275 } 276 handleConnectionFailure(ioFailures++, this.rpcClient.maxRetries, ie); 277 } 278 } 279 } 280 281 /** 282 * Handle connection failures If the current number of retries is equal to the max number of 283 * retries, stop retrying and throw the exception; Otherwise backoff N seconds and try connecting 284 * again. This Method is only called from inside setupIOstreams(), which is synchronized. Hence 285 * the sleep is synchronized; the locks will be retained. 286 * @param curRetries current number of retries 287 * @param maxRetries max number of retries allowed 288 * @param ioe failure reason 289 * @throws IOException if max number of retries is reached 290 */ 291 private void handleConnectionFailure(int curRetries, int maxRetries, IOException ioe) 292 throws IOException { 293 closeSocket(); 294 295 // throw the exception if the maximum number of retries is reached 296 if (curRetries >= maxRetries || ExceptionUtil.isInterrupt(ioe)) { 297 throw ioe; 298 } 299 300 // otherwise back off and retry 301 try { 302 Thread.sleep(this.rpcClient.failureSleep); 303 } catch (InterruptedException ie) { 304 ExceptionUtil.rethrowIfInterrupt(ie); 305 } 306 307 if (LOG.isInfoEnabled()) { 308 LOG.info("Retrying connect to server: " + remoteId.getAddress() + " after sleeping " 309 + this.rpcClient.failureSleep + "ms. Already tried " + curRetries + " time(s)."); 310 } 311 } 312 313 /* 314 * wait till someone signals us to start reading RPC response or it is idle too long, it is marked 315 * as to be closed, or the client is marked as not running. 316 * @return true if it is time to read a response; false otherwise. 317 */ 318 private synchronized boolean waitForWork() { 319 // beware of the concurrent access to the calls list: we can add calls, but as well 320 // remove them. 321 long waitUntil = EnvironmentEdgeManager.currentTime() + this.rpcClient.minIdleTimeBeforeClose; 322 for (;;) { 323 if (thread == null) { 324 return false; 325 } 326 if (!calls.isEmpty()) { 327 return true; 328 } 329 if (EnvironmentEdgeManager.currentTime() >= waitUntil) { 330 closeConn( 331 new IOException("idle connection closed with " + calls.size() + " pending request(s)")); 332 return false; 333 } 334 try { 335 wait(Math.min(this.rpcClient.minIdleTimeBeforeClose, 1000)); 336 } catch (InterruptedException e) { 337 // Restore interrupt status 338 Thread.currentThread().interrupt(); 339 } 340 } 341 } 342 343 @Override 344 public void run() { 345 if (LOG.isTraceEnabled()) { 346 LOG.trace(threadName + ": starting"); 347 } 348 while (waitForWork()) { 349 readResponse(); 350 } 351 if (LOG.isTraceEnabled()) { 352 LOG.trace(threadName + ": stopped"); 353 } 354 } 355 356 private void disposeSasl() { 357 if (saslRpcClient != null) { 358 saslRpcClient.dispose(); 359 saslRpcClient = null; 360 } 361 } 362 363 private boolean setupSaslConnection(final InputStream in2, final OutputStream out2) 364 throws IOException { 365 if (this.metrics != null) { 366 this.metrics.incrNsLookups(); 367 } 368 saslRpcClient = new HBaseSaslRpcClient(this.rpcClient.conf, provider, token, 369 socket.getInetAddress(), securityInfo, this.rpcClient.fallbackAllowed, 370 this.rpcClient.conf.get("hbase.rpc.protection", 371 QualityOfProtection.AUTHENTICATION.name().toLowerCase(Locale.ROOT)), 372 this.rpcClient.conf.getBoolean(CRYPTO_AES_ENABLED_KEY, CRYPTO_AES_ENABLED_DEFAULT)); 373 return saslRpcClient.saslConnect(in2, out2); 374 } 375 376 /** 377 * If multiple clients with the same principal try to connect to the same server at the same time, 378 * the server assumes a replay attack is in progress. This is a feature of kerberos. In order to 379 * work around this, what is done is that the client backs off randomly and tries to initiate the 380 * connection again. The other problem is to do with ticket expiry. To handle that, a relogin is 381 * attempted. 382 * <p> 383 * The retry logic is governed by the {@link SaslClientAuthenticationProvider#canRetry()} method. 384 * Some providers have the ability to obtain new credentials and then re-attempt to authenticate 385 * with HBase services. Other providers will continue to fail if they failed the first time -- for 386 * those, we want to fail-fast. 387 * </p> 388 */ 389 private void handleSaslConnectionFailure(final int currRetries, final int maxRetries, 390 final Exception ex, final UserGroupInformation user) throws IOException, InterruptedException { 391 closeSocket(); 392 user.doAs(new PrivilegedExceptionAction<Object>() { 393 @Override 394 public Object run() throws IOException, InterruptedException { 395 // A provider which failed authentication, but doesn't have the ability to relogin with 396 // some external system (e.g. username/password, the password either works or it doesn't) 397 if (!provider.canRetry()) { 398 LOG.warn("Exception encountered while connecting to the server : " + ex); 399 if (ex instanceof RemoteException) { 400 throw (RemoteException) ex; 401 } 402 if (ex instanceof SaslException) { 403 String msg = "SASL authentication failed." 404 + " The most likely cause is missing or invalid credentials."; 405 throw new RuntimeException(msg, ex); 406 } 407 throw new IOException(ex); 408 } 409 410 // Other providers, like kerberos, could request a new ticket from a keytab. Let 411 // them try again. 412 if (currRetries < maxRetries) { 413 LOG.debug("Exception encountered while connecting to the server", ex); 414 415 // Invoke the provider to perform the relogin 416 provider.relogin(); 417 418 // Get rid of any old state on the SaslClient 419 disposeSasl(); 420 421 // have granularity of milliseconds 422 // we are sleeping with the Connection lock held but since this 423 // connection instance is being used for connecting to the server 424 // in question, it is okay 425 Thread.sleep(ThreadLocalRandom.current().nextInt(reloginMaxBackoff) + 1); 426 return null; 427 } else { 428 String msg = 429 "Failed to initiate connection for " + UserGroupInformation.getLoginUser().getUserName() 430 + " to " + securityInfo.getServerPrincipal(); 431 throw new IOException(msg, ex); 432 } 433 } 434 }); 435 } 436 437 private void setupIOstreams() throws IOException { 438 if (socket != null) { 439 // The connection is already available. Perfect. 440 return; 441 } 442 443 if (this.rpcClient.failedServers.isFailedServer(remoteId.getAddress())) { 444 if (LOG.isDebugEnabled()) { 445 LOG.debug("Not trying to connect to " + remoteId.getAddress() 446 + " this server is in the failed servers list"); 447 } 448 throw new FailedServerException( 449 "This server is in the failed servers list: " + remoteId.getAddress()); 450 } 451 452 try { 453 if (LOG.isDebugEnabled()) { 454 LOG.debug("Connecting to " + remoteId.getAddress()); 455 } 456 457 short numRetries = 0; 458 int reloginMaxRetries = this.rpcClient.conf.getInt("hbase.security.relogin.maxretries", 5); 459 while (true) { 460 setupConnection(); 461 InputStream inStream = NetUtils.getInputStream(socket); 462 // This creates a socket with a write timeout. This timeout cannot be changed. 463 OutputStream outStream = NetUtils.getOutputStream(socket, this.rpcClient.writeTO); 464 // Write out the preamble -- MAGIC, version, and auth to use. 465 writeConnectionHeaderPreamble(outStream); 466 if (useSasl) { 467 final InputStream in2 = inStream; 468 final OutputStream out2 = outStream; 469 UserGroupInformation ticket = provider.getRealUser(remoteId.ticket); 470 boolean continueSasl; 471 if (ticket == null) { 472 throw new FatalConnectionException("ticket/user is null"); 473 } 474 try { 475 continueSasl = ticket.doAs(new PrivilegedExceptionAction<Boolean>() { 476 @Override 477 public Boolean run() throws IOException { 478 return setupSaslConnection(in2, out2); 479 } 480 }); 481 } catch (Exception ex) { 482 ExceptionUtil.rethrowIfInterrupt(ex); 483 handleSaslConnectionFailure(numRetries++, reloginMaxRetries, ex, ticket); 484 continue; 485 } 486 if (continueSasl) { 487 // Sasl connect is successful. Let's set up Sasl i/o streams. 488 inStream = saslRpcClient.getInputStream(); 489 outStream = saslRpcClient.getOutputStream(); 490 } else { 491 // fall back to simple auth because server told us so. 492 // do not change authMethod and useSasl here, we should start from secure when 493 // reconnecting because regionserver may change its sasl config after restart. 494 } 495 } 496 this.in = new DataInputStream(new BufferedInputStream(inStream)); 497 this.out = new DataOutputStream(new BufferedOutputStream(outStream)); 498 // Now write out the connection header 499 writeConnectionHeader(); 500 // process the response from server for connection header if necessary 501 processResponseForConnectionHeader(); 502 503 break; 504 } 505 } catch (Throwable t) { 506 closeSocket(); 507 IOException e = ExceptionUtil.asInterrupt(t); 508 if (e == null) { 509 this.rpcClient.failedServers.addToFailedServers(remoteId.getAddress(), t); 510 if (t instanceof LinkageError) { 511 // probably the hbase hadoop version does not match the running hadoop version 512 e = new DoNotRetryIOException(t); 513 } else if (t instanceof IOException) { 514 e = (IOException) t; 515 } else { 516 e = new IOException("Could not set up IO Streams to " + remoteId.getAddress(), t); 517 } 518 } 519 throw e; 520 } 521 522 // start the receiver thread after the socket connection has been set up 523 thread = new Thread(this, threadName); 524 thread.setDaemon(true); 525 thread.start(); 526 } 527 528 /** 529 * Write the RPC header: {@code <MAGIC WORD -- 'HBas'> <ONEBYTE_VERSION> <ONEBYTE_AUTH_TYPE>} 530 */ 531 private void writeConnectionHeaderPreamble(OutputStream out) throws IOException { 532 out.write(connectionHeaderPreamble); 533 out.flush(); 534 } 535 536 /** 537 * Write the connection header. 538 */ 539 private void writeConnectionHeader() throws IOException { 540 boolean isCryptoAesEnable = false; 541 // check if Crypto AES is enabled 542 if (saslRpcClient != null) { 543 boolean saslEncryptionEnabled = SaslUtil.QualityOfProtection.PRIVACY.getSaslQop() 544 .equalsIgnoreCase(saslRpcClient.getSaslQOP()); 545 isCryptoAesEnable = saslEncryptionEnabled 546 && conf.getBoolean(CRYPTO_AES_ENABLED_KEY, CRYPTO_AES_ENABLED_DEFAULT); 547 } 548 549 // if Crypto AES is enabled, set transformation and negotiate with server 550 if (isCryptoAesEnable) { 551 waitingConnectionHeaderResponse = true; 552 } 553 this.out.write(connectionHeaderWithLength); 554 this.out.flush(); 555 } 556 557 private void processResponseForConnectionHeader() throws IOException { 558 // if no response excepted, return 559 if (!waitingConnectionHeaderResponse) return; 560 try { 561 // read the ConnectionHeaderResponse from server 562 int len = this.in.readInt(); 563 byte[] buff = new byte[len]; 564 int readSize = this.in.read(buff); 565 if (LOG.isDebugEnabled()) { 566 LOG.debug("Length of response for connection header:" + readSize); 567 } 568 569 RPCProtos.ConnectionHeaderResponse connectionHeaderResponse = 570 RPCProtos.ConnectionHeaderResponse.parseFrom(buff); 571 572 // Get the CryptoCipherMeta, update the HBaseSaslRpcClient for Crypto Cipher 573 if (connectionHeaderResponse.hasCryptoCipherMeta()) { 574 negotiateCryptoAes(connectionHeaderResponse.getCryptoCipherMeta()); 575 } 576 waitingConnectionHeaderResponse = false; 577 } catch (SocketTimeoutException ste) { 578 LOG.error(HBaseMarkers.FATAL, 579 "Can't get the connection header response for rpc timeout, " 580 + "please check if server has the correct configuration to support the additional " 581 + "function.", 582 ste); 583 // timeout when waiting the connection header response, ignore the additional function 584 throw new IOException("Timeout while waiting connection header response", ste); 585 } 586 } 587 588 private void negotiateCryptoAes(RPCProtos.CryptoCipherMeta cryptoCipherMeta) throws IOException { 589 // initialize the Crypto AES with CryptoCipherMeta 590 saslRpcClient.initCryptoCipher(cryptoCipherMeta, this.rpcClient.conf); 591 // reset the inputStream/outputStream for Crypto AES encryption 592 this.in = new DataInputStream(new BufferedInputStream(saslRpcClient.getInputStream())); 593 this.out = new DataOutputStream(new BufferedOutputStream(saslRpcClient.getOutputStream())); 594 } 595 596 /** 597 * Initiates a call by sending the parameter to the remote server. Note: this is not called from 598 * the Connection thread, but by other threads. 599 * @see #readResponse() 600 */ 601 private void writeRequest(Call call) throws IOException { 602 ByteBuf cellBlock = null; 603 try { 604 cellBlock = this.rpcClient.cellBlockBuilder.buildCellBlock(this.codec, this.compressor, 605 call.cells, PooledByteBufAllocator.DEFAULT); 606 CellBlockMeta cellBlockMeta; 607 if (cellBlock != null) { 608 cellBlockMeta = CellBlockMeta.newBuilder().setLength(cellBlock.readableBytes()).build(); 609 } else { 610 cellBlockMeta = null; 611 } 612 RequestHeader requestHeader = buildRequestHeader(call, cellBlockMeta); 613 614 setupIOstreams(); 615 616 // Now we're going to write the call. We take the lock, then check that the connection 617 // is still valid, and, if so we do the write to the socket. If the write fails, we don't 618 // know where we stand, we have to close the connection. 619 if (Thread.interrupted()) { 620 throw new InterruptedIOException(); 621 } 622 623 calls.put(call.id, call); // We put first as we don't want the connection to become idle. 624 // from here, we do not throw any exception to upper layer as the call has been tracked in 625 // the pending calls map. 626 try { 627 call.callStats.setRequestSizeBytes(write(this.out, requestHeader, call.param, cellBlock)); 628 } catch (Throwable t) { 629 if (LOG.isTraceEnabled()) { 630 LOG.trace("Error while writing {}", call.toShortString()); 631 } 632 IOException e = IPCUtil.toIOE(t); 633 closeConn(e); 634 return; 635 } 636 } finally { 637 if (cellBlock != null) { 638 cellBlock.release(); 639 } 640 } 641 notifyAll(); 642 } 643 644 /* 645 * Receive a response. Because only one receiver, so no synchronization on in. 646 */ 647 private void readResponse() { 648 Call call = null; 649 boolean expectedCall = false; 650 try { 651 // See HBaseServer.Call.setResponse for where we write out the response. 652 // Total size of the response. Unused. But have to read it in anyways. 653 int totalSize = in.readInt(); 654 655 // Read the header 656 ResponseHeader responseHeader = ResponseHeader.parseDelimitedFrom(in); 657 int id = responseHeader.getCallId(); 658 call = calls.remove(id); // call.done have to be set before leaving this method 659 expectedCall = (call != null && !call.isDone()); 660 if (!expectedCall) { 661 // So we got a response for which we have no corresponding 'call' here on the client-side. 662 // We probably timed out waiting, cleaned up all references, and now the server decides 663 // to return a response. There is nothing we can do w/ the response at this stage. Clean 664 // out the wire of the response so its out of the way and we can get other responses on 665 // this connection. 666 int readSoFar = getTotalSizeWhenWrittenDelimited(responseHeader); 667 int whatIsLeftToRead = totalSize - readSoFar; 668 IOUtils.skipFully(in, whatIsLeftToRead); 669 if (call != null) { 670 call.callStats.setResponseSizeBytes(totalSize); 671 call.callStats 672 .setCallTimeMs(EnvironmentEdgeManager.currentTime() - call.callStats.getStartTime()); 673 } 674 return; 675 } 676 if (responseHeader.hasException()) { 677 ExceptionResponse exceptionResponse = responseHeader.getException(); 678 RemoteException re = createRemoteException(exceptionResponse); 679 call.setException(re); 680 call.callStats.setResponseSizeBytes(totalSize); 681 call.callStats 682 .setCallTimeMs(EnvironmentEdgeManager.currentTime() - call.callStats.getStartTime()); 683 if (isFatalConnectionException(exceptionResponse)) { 684 synchronized (this) { 685 closeConn(re); 686 } 687 } 688 } else { 689 Message value = null; 690 if (call.responseDefaultType != null) { 691 Message.Builder builder = call.responseDefaultType.newBuilderForType(); 692 ProtobufUtil.mergeDelimitedFrom(builder, in); 693 value = builder.build(); 694 } 695 CellScanner cellBlockScanner = null; 696 if (responseHeader.hasCellBlockMeta()) { 697 int size = responseHeader.getCellBlockMeta().getLength(); 698 byte[] cellBlock = new byte[size]; 699 IOUtils.readFully(this.in, cellBlock, 0, cellBlock.length); 700 cellBlockScanner = this.rpcClient.cellBlockBuilder.createCellScanner(this.codec, 701 this.compressor, cellBlock); 702 } 703 call.setResponse(value, cellBlockScanner); 704 call.callStats.setResponseSizeBytes(totalSize); 705 call.callStats 706 .setCallTimeMs(EnvironmentEdgeManager.currentTime() - call.callStats.getStartTime()); 707 } 708 } catch (IOException e) { 709 if (expectedCall) { 710 call.setException(e); 711 } 712 if (e instanceof SocketTimeoutException) { 713 // Clean up open calls but don't treat this as a fatal condition, 714 // since we expect certain responses to not make it by the specified 715 // {@link ConnectionId#rpcTimeout}. 716 if (LOG.isTraceEnabled()) { 717 LOG.trace("ignored", e); 718 } 719 } else { 720 synchronized (this) { 721 closeConn(e); 722 } 723 } 724 } 725 } 726 727 @Override 728 protected synchronized void callTimeout(Call call) { 729 // call sender 730 calls.remove(call.id); 731 } 732 733 // just close socket input and output. 734 private void closeSocket() { 735 IOUtils.closeStream(out); 736 IOUtils.closeStream(in); 737 IOUtils.closeSocket(socket); 738 out = null; 739 in = null; 740 socket = null; 741 } 742 743 // close socket, reader, and clean up all pending calls. 744 private void closeConn(IOException e) { 745 if (thread == null) { 746 return; 747 } 748 thread.interrupt(); 749 thread = null; 750 closeSocket(); 751 if (callSender != null) { 752 callSender.cleanup(e); 753 } 754 for (Call call : calls.values()) { 755 call.setException(e); 756 } 757 calls.clear(); 758 } 759 760 // release all resources, the connection will not be used any more. 761 @Override 762 public synchronized void shutdown() { 763 closed = true; 764 if (callSender != null) { 765 callSender.interrupt(); 766 } 767 closeConn(new IOException("connection to " + remoteId.getAddress() + " closed")); 768 } 769 770 @Override 771 public void cleanupConnection() { 772 // do nothing 773 } 774 775 @Override 776 public synchronized void sendRequest(final Call call, HBaseRpcController pcrc) 777 throws IOException { 778 pcrc.notifyOnCancel(new RpcCallback<Object>() { 779 780 @Override 781 public void run(Object parameter) { 782 setCancelled(call); 783 synchronized (BlockingRpcConnection.this) { 784 if (callSender != null) { 785 callSender.remove(call); 786 } else { 787 calls.remove(call.id); 788 } 789 } 790 } 791 }, new CancellationCallback() { 792 793 @Override 794 public void run(boolean cancelled) throws IOException { 795 if (cancelled) { 796 setCancelled(call); 797 return; 798 } 799 scheduleTimeoutTask(call); 800 if (callSender != null) { 801 callSender.sendCall(call); 802 } else { 803 // this is in the same thread with the caller so do not need to attach the trace context 804 // again. 805 writeRequest(call); 806 } 807 } 808 }); 809 } 810 811 @Override 812 public synchronized boolean isActive() { 813 return thread != null; 814 } 815}