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}