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 io.opentelemetry.api.trace.Span;
021import io.opentelemetry.api.trace.StatusCode;
022import io.opentelemetry.context.Scope;
023import java.io.IOException;
024import java.net.InetAddress;
025import java.nio.ByteBuffer;
026import java.util.ArrayList;
027import java.util.List;
028import java.util.Optional;
029import java.util.concurrent.atomic.AtomicInteger;
030import org.apache.hadoop.hbase.CellScanner;
031import org.apache.hadoop.hbase.DoNotRetryIOException;
032import org.apache.hadoop.hbase.HBaseServerException;
033import org.apache.hadoop.hbase.exceptions.RegionMovedException;
034import org.apache.hadoop.hbase.io.ByteBuffAllocator;
035import org.apache.hadoop.hbase.io.ByteBufferListOutputStream;
036import org.apache.hadoop.hbase.ipc.RpcServer.CallCleanup;
037import org.apache.hadoop.hbase.security.User;
038import org.apache.hadoop.hbase.trace.TraceUtil;
039import org.apache.hadoop.hbase.util.ByteBufferUtils;
040import org.apache.hadoop.hbase.util.Bytes;
041import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
042import org.apache.hadoop.util.StringUtils;
043import org.apache.yetus.audience.InterfaceAudience;
044
045import org.apache.hbase.thirdparty.com.google.protobuf.BlockingService;
046import org.apache.hbase.thirdparty.com.google.protobuf.CodedOutputStream;
047import org.apache.hbase.thirdparty.com.google.protobuf.Descriptors.MethodDescriptor;
048import org.apache.hbase.thirdparty.com.google.protobuf.Message;
049
050import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
051import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.VersionInfo;
052import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.CellBlockMeta;
053import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ExceptionResponse;
054import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader;
055import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ResponseHeader;
056
057/**
058 * Datastructure that holds all necessary to a method invocation and then afterward, carries the
059 * result.
060 */
061@InterfaceAudience.Private
062public abstract class ServerCall<T extends ServerRpcConnection> implements RpcCall, RpcResponse {
063
064  protected final int id; // the client's call id
065  protected final BlockingService service;
066  protected final MethodDescriptor md;
067  protected final RequestHeader header;
068  protected Message param; // the parameter passed
069  // Optional cell data passed outside of protobufs.
070  protected final CellScanner cellScanner;
071  protected final T connection; // connection to client
072  protected final long receiveTime; // the time received when response is null
073  // the time served when response is not null
074  protected final int timeout;
075  protected long startTime;
076  protected final long deadline;// the deadline to handle this call, if exceed we can drop it.
077
078  protected final ByteBuffAllocator bbAllocator;
079
080  protected final CellBlockBuilder cellBlockBuilder;
081
082  /**
083   * Chain of buffers to send as response.
084   */
085  protected BufferChain response;
086
087  protected final long size; // size of current call
088  protected boolean isError;
089  protected ByteBufferListOutputStream cellBlockStream = null;
090  protected CallCleanup reqCleanup = null;
091
092  protected final User user;
093  protected final InetAddress remoteAddress;
094  protected RpcCallback rpcCallback;
095
096  private long responseCellSize = 0;
097  private long responseBlockSize = 0;
098  // cumulative size of serialized exceptions
099  private long exceptionSize = 0;
100  private final boolean retryImmediatelySupported;
101
102  // This is a dirty hack to address HBASE-22539. The highest bit is for rpc ref and cleanup, and
103  // the rest of the bits are for WAL reference count. We can only call release if all of them are
104  // zero. The reason why we can not use a general reference counting is that, we may call cleanup
105  // multiple times in the current implementation. We should fix this in the future.
106  // The refCount here will start as 0x80000000 and increment with every WAL reference and decrement
107  // from WAL side on release
108  private final AtomicInteger reference = new AtomicInteger(0x80000000);
109
110  private final Span span;
111
112  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "NP_NULL_ON_SOME_PATH",
113      justification = "Can't figure why this complaint is happening... see below")
114  ServerCall(int id, BlockingService service, MethodDescriptor md, RequestHeader header,
115    Message param, CellScanner cellScanner, T connection, long size, InetAddress remoteAddress,
116    long receiveTime, int timeout, ByteBuffAllocator byteBuffAllocator,
117    CellBlockBuilder cellBlockBuilder, CallCleanup reqCleanup) {
118    this.id = id;
119    this.service = service;
120    this.md = md;
121    this.header = header;
122    this.param = param;
123    this.cellScanner = cellScanner;
124    this.connection = connection;
125    this.receiveTime = receiveTime;
126    this.response = null;
127    this.isError = false;
128    this.size = size;
129    if (connection != null) {
130      this.user = connection.user;
131      this.retryImmediatelySupported = connection.retryImmediatelySupported;
132    } else {
133      this.user = null;
134      this.retryImmediatelySupported = false;
135    }
136    this.remoteAddress = remoteAddress;
137    this.timeout = timeout;
138    this.deadline = this.timeout > 0 ? this.receiveTime + this.timeout : Long.MAX_VALUE;
139    this.bbAllocator = byteBuffAllocator;
140    this.cellBlockBuilder = cellBlockBuilder;
141    this.reqCleanup = reqCleanup;
142    this.span = Span.current();
143  }
144
145  /**
146   * Call is done. Execution happened and we returned results to client. It is now safe to cleanup.
147   */
148  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "IS2_INCONSISTENT_SYNC",
149      justification = "Presume the lock on processing request held by caller is protection enough")
150  @Override
151  public void done() {
152    if (this.cellBlockStream != null) {
153      // This will return back the BBs which we got from pool.
154      this.cellBlockStream.releaseResources();
155      this.cellBlockStream = null;
156    }
157    // If the call was run successfuly, we might have already returned the BB
158    // back to pool. No worries..Then inputCellBlock will be null
159    cleanup();
160    span.end();
161  }
162
163  @Override
164  public void cleanup() {
165    for (;;) {
166      int ref = reference.get();
167      if ((ref & 0x80000000) == 0) {
168        return;
169      }
170      int nextRef = ref & 0x7fffffff;
171      if (reference.compareAndSet(ref, nextRef)) {
172        if (nextRef == 0) {
173          if (this.reqCleanup != null) {
174            this.reqCleanup.run();
175          }
176        }
177        return;
178      }
179    }
180  }
181
182  public void retainByWAL() {
183    reference.incrementAndGet();
184  }
185
186  public void releaseByWAL() {
187    // Here this method of decrementAndGet for releasing WAL reference count will work in both
188    // cases - i.e. highest bit (cleanup) 1 or 0. We will be decrementing a negative or positive
189    // value respectively in these 2 cases, but the logic will work the same way
190    if (reference.decrementAndGet() == 0) {
191      if (this.reqCleanup != null) {
192        this.reqCleanup.run();
193      }
194    }
195
196  }
197
198  @Override
199  public String toString() {
200    return toShortString() + " param: "
201      + (this.param != null ? ProtobufUtil.getShortTextFormat(this.param) : "") + " connection: "
202      + connection.toString();
203  }
204
205  @Override
206  public RequestHeader getHeader() {
207    return this.header;
208  }
209
210  @Override
211  public int getPriority() {
212    return this.header.getPriority();
213  }
214
215  /*
216   * Short string representation without param info because param itself could be huge depends on
217   * the payload of a command
218   */
219  @Override
220  public String toShortString() {
221    String serviceName = this.connection.service != null
222      ? this.connection.service.getDescriptorForType().getName()
223      : "null";
224    return "callId: " + this.id + " service: " + serviceName + " methodName: "
225      + ((this.md != null) ? this.md.getName() : "n/a") + " size: "
226      + StringUtils.TraditionalBinaryPrefix.long2String(this.size, "", 1) + " connection: "
227      + connection + " deadline: " + deadline;
228  }
229
230  @Override
231  public synchronized void setResponse(Message m, final CellScanner cells, Throwable t,
232    String errorMsg) {
233    if (this.isError) {
234      return;
235    }
236    if (t != null) {
237      this.isError = true;
238      TraceUtil.setError(span, t);
239    } else {
240      span.setStatus(StatusCode.OK);
241    }
242    BufferChain bc = null;
243    try {
244      ResponseHeader.Builder headerBuilder = ResponseHeader.newBuilder();
245      // Call id.
246      headerBuilder.setCallId(this.id);
247      if (t != null) {
248        setExceptionResponse(t, errorMsg, headerBuilder);
249      }
250      // Pass reservoir to buildCellBlock. Keep reference to returne so can add it back to the
251      // reservoir when finished. This is hacky and the hack is not contained but benefits are
252      // high when we can avoid a big buffer allocation on each rpc.
253      List<ByteBuffer> cellBlock = null;
254      int cellBlockSize = 0;
255      if (bbAllocator.isReservoirEnabled()) {
256        this.cellBlockStream = this.cellBlockBuilder.buildCellBlockStream(this.connection.codec,
257          this.connection.compressionCodec, cells, bbAllocator);
258        if (this.cellBlockStream != null) {
259          cellBlock = this.cellBlockStream.getByteBuffers();
260          cellBlockSize = this.cellBlockStream.size();
261        }
262      } else {
263        ByteBuffer b = this.cellBlockBuilder.buildCellBlock(this.connection.codec,
264          this.connection.compressionCodec, cells);
265        if (b != null) {
266          cellBlockSize = b.remaining();
267          cellBlock = new ArrayList<>(1);
268          cellBlock.add(b);
269        }
270      }
271
272      if (cellBlockSize > 0) {
273        CellBlockMeta.Builder cellBlockBuilder = CellBlockMeta.newBuilder();
274        // Presumes the cellBlock bytebuffer has been flipped so limit has total size in it.
275        cellBlockBuilder.setLength(cellBlockSize);
276        headerBuilder.setCellBlockMeta(cellBlockBuilder.build());
277      }
278      Message header = headerBuilder.build();
279      ByteBuffer headerBuf = createHeaderAndMessageBytes(m, header, cellBlockSize, cellBlock);
280      ByteBuffer[] responseBufs = null;
281      int cellBlockBufferSize = 0;
282      if (cellBlock != null) {
283        cellBlockBufferSize = cellBlock.size();
284        responseBufs = new ByteBuffer[1 + cellBlockBufferSize];
285      } else {
286        responseBufs = new ByteBuffer[1];
287      }
288      responseBufs[0] = headerBuf;
289      if (cellBlock != null) {
290        for (int i = 0; i < cellBlockBufferSize; i++) {
291          responseBufs[i + 1] = cellBlock.get(i);
292        }
293      }
294      bc = new BufferChain(responseBufs);
295    } catch (IOException e) {
296      RpcServer.LOG.warn("Exception while creating response " + e);
297    }
298    this.response = bc;
299    // Once a response message is created and set to this.response, this Call can be treated as
300    // done. The Responder thread will do the n/w write of this message back to client.
301    if (this.rpcCallback != null) {
302      try (Scope ignored = span.makeCurrent()) {
303        this.rpcCallback.run();
304      } catch (Exception e) {
305        // Don't allow any exception here to kill this handler thread.
306        RpcServer.LOG.warn("Exception while running the Rpc Callback.", e);
307        TraceUtil.setError(span, e);
308      }
309    }
310  }
311
312  static void setExceptionResponse(Throwable t, String errorMsg,
313    ResponseHeader.Builder headerBuilder) {
314    ExceptionResponse.Builder exceptionBuilder = ExceptionResponse.newBuilder();
315    exceptionBuilder.setExceptionClassName(t.getClass().getName());
316    exceptionBuilder.setStackTrace(errorMsg);
317    exceptionBuilder.setDoNotRetry(t instanceof DoNotRetryIOException);
318    if (t instanceof RegionMovedException) {
319      // Special casing for this exception. This is only one carrying a payload.
320      // Do this instead of build a generic system for allowing exceptions carry
321      // any kind of payload.
322      RegionMovedException rme = (RegionMovedException) t;
323      exceptionBuilder.setHostname(rme.getHostname());
324      exceptionBuilder.setPort(rme.getPort());
325    } else if (t instanceof HBaseServerException) {
326      HBaseServerException hse = (HBaseServerException) t;
327      exceptionBuilder.setServerOverloaded(hse.isServerOverloaded());
328    }
329    // Set the exception as the result of the method invocation.
330    headerBuilder.setException(exceptionBuilder.build());
331  }
332
333  static ByteBuffer createHeaderAndMessageBytes(Message result, Message header, int cellBlockSize,
334    List<ByteBuffer> cellBlock) throws IOException {
335    // Organize the response as a set of bytebuffers rather than collect it all together inside
336    // one big byte array; save on allocations.
337    // for writing the header, we check if there is available space in the buffers
338    // created for the cellblock itself. If there is space for the header, we reuse
339    // the last buffer in the cellblock. This applies to the cellblock created from the
340    // pool or even the onheap cellblock buffer in case there is no pool enabled.
341    // Possible reuse would avoid creating a temporary array for storing the header every time.
342    ByteBuffer possiblePBBuf = (cellBlockSize > 0) ? cellBlock.get(cellBlock.size() - 1) : null;
343    int headerSerializedSize = 0, resultSerializedSize = 0, headerVintSize = 0, resultVintSize = 0;
344    if (header != null) {
345      headerSerializedSize = header.getSerializedSize();
346      headerVintSize = CodedOutputStream.computeUInt32SizeNoTag(headerSerializedSize);
347    }
348    if (result != null) {
349      resultSerializedSize = result.getSerializedSize();
350      resultVintSize = CodedOutputStream.computeUInt32SizeNoTag(resultSerializedSize);
351    }
352    // calculate the total size
353    int totalSize = headerSerializedSize + headerVintSize + (resultSerializedSize + resultVintSize)
354      + cellBlockSize;
355    int totalPBSize = headerSerializedSize + headerVintSize + resultSerializedSize + resultVintSize
356      + Bytes.SIZEOF_INT;
357    // Only if the last buffer has enough space for header use it. Else allocate
358    // a new buffer. Assume they are all flipped
359    if (possiblePBBuf != null && possiblePBBuf.limit() + totalPBSize <= possiblePBBuf.capacity()) {
360      // duplicate the buffer. This is where the header is going to be written
361      ByteBuffer pbBuf = possiblePBBuf.duplicate();
362      // get the current limit
363      int limit = pbBuf.limit();
364      // Position such that we write the header to the end of the buffer
365      pbBuf.position(limit);
366      // limit to the header size
367      pbBuf.limit(totalPBSize + limit);
368      // mark the current position
369      pbBuf.mark();
370      writeToCOS(result, header, totalSize, pbBuf);
371      // reset the buffer back to old position
372      pbBuf.reset();
373      return pbBuf;
374    } else {
375      return createHeaderAndMessageBytes(result, header, totalSize, totalPBSize);
376    }
377  }
378
379  private static void writeToCOS(Message result, Message header, int totalSize, ByteBuffer pbBuf)
380    throws IOException {
381    ByteBufferUtils.putInt(pbBuf, totalSize);
382    // create COS that works on BB
383    CodedOutputStream cos = CodedOutputStream.newInstance(pbBuf);
384    if (header != null) {
385      cos.writeMessageNoTag(header);
386    }
387    if (result != null) {
388      cos.writeMessageNoTag(result);
389    }
390    cos.flush();
391    cos.checkNoSpaceLeft();
392  }
393
394  private static ByteBuffer createHeaderAndMessageBytes(Message result, Message header,
395    int totalSize, int totalPBSize) throws IOException {
396    ByteBuffer pbBuf = ByteBuffer.allocate(totalPBSize);
397    writeToCOS(result, header, totalSize, pbBuf);
398    pbBuf.flip();
399    return pbBuf;
400  }
401
402  @Override
403  public long disconnectSince() {
404    if (!this.connection.isConnectionOpen()) {
405      return EnvironmentEdgeManager.currentTime() - receiveTime;
406    } else {
407      return -1L;
408    }
409  }
410
411  @Override
412  public boolean isClientCellBlockSupported() {
413    return this.connection != null && this.connection.codec != null;
414  }
415
416  @Override
417  public long getResponseCellSize() {
418    return responseCellSize;
419  }
420
421  @Override
422  public void incrementResponseCellSize(long cellSize) {
423    responseCellSize += cellSize;
424  }
425
426  @Override
427  public long getResponseBlockSize() {
428    return responseBlockSize;
429  }
430
431  @Override
432  public void incrementResponseBlockSize(long blockSize) {
433    responseBlockSize += blockSize;
434  }
435
436  @Override
437  public long getResponseExceptionSize() {
438    return exceptionSize;
439  }
440
441  @Override
442  public void incrementResponseExceptionSize(long exSize) {
443    exceptionSize += exSize;
444  }
445
446  @Override
447  public long getSize() {
448    return this.size;
449  }
450
451  @Override
452  public long getDeadline() {
453    return deadline;
454  }
455
456  @Override
457  public Optional<User> getRequestUser() {
458    return Optional.ofNullable(user);
459  }
460
461  @Override
462  public InetAddress getRemoteAddress() {
463    return remoteAddress;
464  }
465
466  @Override
467  public VersionInfo getClientVersionInfo() {
468    return connection.getVersionInfo();
469  }
470
471  @Override
472  public synchronized void setCallBack(RpcCallback callback) {
473    this.rpcCallback = callback;
474  }
475
476  @Override
477  public boolean isRetryImmediatelySupported() {
478    return retryImmediatelySupported;
479  }
480
481  @Override
482  public BlockingService getService() {
483    return service;
484  }
485
486  @Override
487  public MethodDescriptor getMethod() {
488    return md;
489  }
490
491  @Override
492  public Message getParam() {
493    return param;
494  }
495
496  @Override
497  public CellScanner getCellScanner() {
498    return cellScanner;
499  }
500
501  @Override
502  public long getReceiveTime() {
503    return receiveTime;
504  }
505
506  @Override
507  public long getStartTime() {
508    return startTime;
509  }
510
511  @Override
512  public void setStartTime(long t) {
513    this.startTime = t;
514  }
515
516  @Override
517  public int getTimeout() {
518    return timeout;
519  }
520
521  @Override
522  public int getRemotePort() {
523    return connection.getRemotePort();
524  }
525
526  @Override
527  public synchronized BufferChain getResponse() {
528    return response;
529  }
530}