@Override
public ByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
    checkIndex(index, length);
    if (dst == null) {
        throw new NullPointerException("dst");
    }
    if (dstIndex < 0 || dstIndex > dst.capacity() - length) {
        throw new IndexOutOfBoundsException("dstIndex: " + dstIndex);
    }

    if (dst.hasMemoryAddress()) {
        PlatformDependent.copyMemory(addr(index), dst.memoryAddress() + dstIndex, length);
    } else if (dst.hasArray()) {
        PlatformDependent.copyMemory(addr(index), dst.array(), dst.arrayOffset() + dstIndex, length);
    } else {
        dst.setBytes(dstIndex, this, index, length);
    }
    return this;
}
 

@Override
public final synchronized SSLParameters getSSLParameters() {
    SSLParameters sslParameters = super.getSSLParameters();

    int version = PlatformDependent.javaVersion();
    if (version >= 7) {
        sslParameters.setEndpointIdentificationAlgorithm(endPointIdentificationAlgorithm);
        Java7SslParametersUtils.setAlgorithmConstraints(sslParameters, algorithmConstraints);
        if (version >= 8) {
            if (sniHostNames != null) {
                Java8SslUtils.setSniHostNames(sslParameters, sniHostNames);
            }
            if (!isDestroyed()) {
                Java8SslUtils.setUseCipherSuitesOrder(
                        sslParameters, (SSL.getOptions(ssl) & SSL.SSL_OP_CIPHER_SERVER_PREFERENCE) != 0);
            }

            Java8SslUtils.setSNIMatchers(sslParameters, matchers);
        }
    }
    return sslParameters;
}
 

private void runFinishPeerReadTask(final VirtualChannel peer) {
    // If the peer is writing, we must wait until after reads are completed for that peer before we can read. So
    // we keep track of the task, and coordinate later that our read can't happen until the peer is done.
    final Runnable finishPeerReadTask = new Runnable() {
        @Override
        public void run() {
            finishPeerRead0(peer);
        }
    };
    try {
        if (peer.writeInProgress) {
            peer.finishReadFuture = peer.eventLoop().submit(finishPeerReadTask);
        } else {
            peer.eventLoop().execute(finishPeerReadTask);
        }
    } catch (Throwable cause) {
        close();
        peer.close();
        PlatformDependent.throwException(cause);
    }
}
 

private static int unsafeOnHeapWriteUTF(final CharSequence str, final byte[] bytes, final int index, final int length) {
   int charCount = index;
   for (int i = 0; i < length; i++) {
      char charAtPos = str.charAt(i);
      if (charAtPos <= 0x7f) {
         PlatformDependent.putByte(bytes, charCount++, (byte) charAtPos);
      } else if (charAtPos >= 0x800) {
         PlatformDependent.putByte(bytes, charCount++, (byte) (0xE0 | charAtPos >> 12 & 0x0F));
         PlatformDependent.putByte(bytes, charCount++, (byte) (0x80 | charAtPos >> 6 & 0x3F));
         PlatformDependent.putByte(bytes, charCount++, (byte) (0x80 | charAtPos >> 0 & 0x3F));
      } else {
         PlatformDependent.putByte(bytes, charCount++, (byte) (0xC0 | charAtPos >> 6 & 0x1F));
         PlatformDependent.putByte(bytes, charCount++, (byte) (0x80 | charAtPos >> 0 & 0x3F));
      }
   }

   final int writtenBytes = (charCount - index);
   return writtenBytes;
}
 

@Override
protected void memoryCopy(ByteBuffer src, int srcOffset, ByteBuffer dst, int dstOffset, int length) {
    if (length == 0) {
        return;
    }

    if (HAS_UNSAFE) {
        PlatformDependent.copyMemory(
                PlatformDependent.directBufferAddress(src) + srcOffset,
                PlatformDependent.directBufferAddress(dst) + dstOffset, length);
    } else {
        // We must duplicate the NIO buffers because they may be accessed by other Netty buffers.我们必须复制NIO缓冲区，因为它们可能被其他Netty缓冲区访问。
        src = src.duplicate();
        dst = dst.duplicate();
        src.position(srcOffset).limit(srcOffset + length);
        dst.position(dstOffset);
        dst.put(src);
    }
}
 

@Override
public void copy(long compoundAddr, final int srcStart, final int count) {
  copyWatch.start();

  final long[] dstAddrs = this.dstAddrs;

  long srcAddr = source.getDataBufferAddress() + srcStart * SIZE;

  final long max = compoundAddr + count * OFFSET_SIZE;
  for (; compoundAddr < max; compoundAddr +=OFFSET_SIZE, srcAddr += SIZE) {
    final int compoundIdx = PlatformDependent.getInt(compoundAddr);
    final int batchIdx = compoundIdx >>> 16;
    final int rowIdx = compoundIdx & 65535;

    final long dstAddr = dstAddrs[batchIdx] + rowIdx * SIZE;
    PlatformDependent.putLong(dstAddr, PlatformDependent.getLong(srcAddr));
    PlatformDependent.putLong(dstAddr + 8, PlatformDependent.getLong(srcAddr + 8));
  }

  copyWatch.stop();
}
 

private static int unsafeOffHeapWriteUTF(final CharSequence str, final long addressBytes, final int index, final int length) {
   int charCount = index;
   for (int i = 0; i < length; i++) {
      char charAtPos = str.charAt(i);
      if (charAtPos <= 0x7f) {
         PlatformDependent.putByte(addressBytes + charCount++, (byte) charAtPos);
      } else if (charAtPos >= 0x800) {
         PlatformDependent.putByte(addressBytes + charCount++, (byte) (0xE0 | charAtPos >> 12 & 0x0F));
         PlatformDependent.putByte(addressBytes + charCount++, (byte) (0x80 | charAtPos >> 6 & 0x3F));
         PlatformDependent.putByte(addressBytes + charCount++, (byte) (0x80 | charAtPos >> 0 & 0x3F));
      } else {
         PlatformDependent.putByte(addressBytes + charCount++, (byte) (0xC0 | charAtPos >> 6 & 0x1F));
         PlatformDependent.putByte(addressBytes + charCount++, (byte) (0x80 | charAtPos >> 0 & 0x3F));
      }
   }

   final int writtenBytes = (charCount - index);
   return writtenBytes;
}
 

@Override
public void release(ByteBuffer buffer) {
   Objects.requireNonNull(buffer);
   boolean directBuffer = buffer.isDirect();
   if (directBuffer == direct && !buffer.isReadOnly()) {
      final ByteBuffer byteBuffer = bytesPool.get();
      if (byteBuffer != buffer) {
         //replace with the current pooled only if greater or null
         if (byteBuffer == null || buffer.capacity() > byteBuffer.capacity()) {
            if (byteBuffer != null) {
               //free the smaller one
               if (directBuffer) {
                  PlatformDependent.freeDirectBuffer(byteBuffer);
               }
            }
            bytesPool.set(buffer);
         } else {
            if (directBuffer) {
               PlatformDependent.freeDirectBuffer(buffer);
            }
         }
      }
   }
}
 

public void accumulate(final long memoryAddr, final int count) {
  final long maxMemAddr = memoryAddr + count * WIDTH_ORDINAL;
  FieldVector inputVector = getInput();
  final long incomingBit = inputVector.getValidityBufferAddress();
  final long incomingValue = inputVector.getDataBufferAddress();
  final long[] bitAddresses = this.bitAddresses;
  final long[] valueAddresses = this.valueAddresses;
  final int scale = ((DecimalVector)inputVector).getScale();

  int incomingIndex = 0;
  for(long ordinalAddr = memoryAddr; ordinalAddr < maxMemAddr; ordinalAddr += WIDTH_ORDINAL, incomingIndex++){
    java.math.BigDecimal newVal = DecimalAccumulatorUtilsNoSpill.getBigDecimal(incomingValue + (incomingIndex * WIDTH_INPUT), valBuf, scale);
    final int tableIndex = PlatformDependent.getInt(ordinalAddr);
    int chunkIndex = tableIndex >>> LBlockHashTableNoSpill.BITS_IN_CHUNK;
    int chunkOffset = tableIndex & LBlockHashTableNoSpill.CHUNK_OFFSET_MASK;
    final long maxAddr = valueAddresses[chunkIndex] + (chunkOffset) * WIDTH_ACCUMULATOR;
    final long bitUpdateAddr = bitAddresses[chunkIndex] + ((chunkOffset >>> 5) * 4);
    final int bitVal = (PlatformDependent.getByte(incomingBit + ((incomingIndex >>> 3))) >>> (incomingIndex & 7)) & 1;
    final int bitUpdateVal = bitVal << (chunkOffset & 31);
    PlatformDependent.putLong(maxAddr, Double.doubleToLongBits(max(Double.longBitsToDouble(PlatformDependent.getLong(maxAddr)), newVal.doubleValue(), bitVal)));
    PlatformDependent.putInt(bitUpdateAddr, PlatformDependent.getInt(bitUpdateAddr) | bitUpdateVal);
  }
}
 

private static Constructor<?> customClassConstructor(String customLeakDetector) {
    try {
        final Class<?> detectorClass = Class.forName(customLeakDetector, true,
                PlatformDependent.getSystemClassLoader());

        if (ResourceLeakDetector.class.isAssignableFrom(detectorClass)) {
            return detectorClass.getConstructor(Class.class, int.class);
        } else {
            logger.error("Class {} does not inherit from ResourceLeakDetector.", customLeakDetector);
        }
    } catch (Throwable t) {
        logger.error("Could not load custom resource leak detector class provided: {}",
                customLeakDetector, t);
    }
    return null;
}
 

private static Constructor<?> obsoleteCustomClassConstructor(String customLeakDetector) {
    try {
        final Class<?> detectorClass = Class.forName(customLeakDetector, true,
                PlatformDependent.getSystemClassLoader());

        if (ResourceLeakDetector.class.isAssignableFrom(detectorClass)) {
            return detectorClass.getConstructor(Class.class, int.class, long.class);
        } else {
            logger.error("Class {} does not inherit from ResourceLeakDetector.", customLeakDetector);
        }
    } catch (Throwable t) {
        logger.error("Could not load custom resource leak detector class provided: {}",
                customLeakDetector, t);
    }
    return null;
}
 

@Setup(Level.Iteration)
public final void setup() throws Exception {
    ByteBufAllocator allocator = new PooledByteBufAllocator(true);
    initEngines(allocator);
    initHandshakeBuffers();

    wrapDstBuffer = allocateBuffer(clientEngine.getSession().getPacketBufferSize() << 2);
    wrapSrcBuffer = allocateBuffer(messageSize);

    byte[] bytes = new byte[messageSize];
    PlatformDependent.threadLocalRandom().nextBytes(bytes);
    wrapSrcBuffer.put(bytes);
    wrapSrcBuffer.flip();

    // Complete the initial TLS handshake.
    if (!doHandshake()) {
        throw new IllegalStateException();
    }
    doSetup();
}
 

public void accumulate(final long memoryAddr, final int count) {
  final long maxAddr = memoryAddr + count * 4;
  final long incomingBit = getInput().getValidityBufferAddress();
  final long incomingValue =  getInput().getDataBufferAddress();
  final long[] bitAddresses = this.bitAddresses;
  final long[] valueAddresses = this.valueAddresses;

  int incomingIndex = 0;
  for(long ordinalAddr = memoryAddr; ordinalAddr < maxAddr; ordinalAddr += 4, incomingIndex++){
    final int bitVal = (PlatformDependent.getByte(incomingBit + ((incomingIndex >>> 3))) >>> (incomingIndex & 7)) & 1;
    final long newVal = PlatformDependent.getLong(incomingValue + (incomingIndex * WIDTH)) * bitVal;
    final int tableIndex = PlatformDependent.getInt(ordinalAddr);
    int chunkIndex = tableIndex >>> LBlockHashTableNoSpill.BITS_IN_CHUNK;
    int chunkOffset = tableIndex & LBlockHashTableNoSpill.CHUNK_OFFSET_MASK;
    final long sumAddr = valueAddresses[chunkIndex] + (chunkOffset) * 8;
    final long bitUpdateAddr = bitAddresses[chunkIndex] + ((chunkOffset >>> 5) * 4);
    final int bitUpdateVal = bitVal << (chunkOffset & 31);
    PlatformDependent.putLong(sumAddr, PlatformDependent.getLong(sumAddr) + newVal);
    PlatformDependent.putInt(bitUpdateAddr, PlatformDependent.getInt(bitUpdateAddr) | bitUpdateVal);
  }
}
 

/**
 * Write encrypted data to the OpenSSL network BIO.将加密数据写入OpenSSL网络BIO。
 */
private ByteBuf writeEncryptedData(final ByteBuffer src, int len) {
    final int pos = src.position();
    if (src.isDirect()) {
        SSL.bioSetByteBuffer(networkBIO, bufferAddress(src) + pos, len, false);
    } else {
        final ByteBuf buf = alloc.directBuffer(len);
        try {
            final int limit = src.limit();
            src.limit(pos + len);
            buf.writeBytes(src);
            // Restore the original position and limit because we don't want to consume from `src`.
            src.position(pos);
            src.limit(limit);

            SSL.bioSetByteBuffer(networkBIO, memoryAddress(buf), len, false);
            return buf;
        } catch (Throwable cause) {
            buf.release();
            PlatformDependent.throwException(cause);
        }
    }
    return null;
}
 

@Override
public void validateName(CharSequence name) {
    if (name == null || name.length() == 0) {
        throw new IllegalArgumentException("empty headers are not allowed [" + name + "]");
    }
    if (name instanceof AsciiString) {
        try {
            ((AsciiString) name).forEachByte(HEADER_NAME_VALIDATOR);
        } catch (Exception e) {
            PlatformDependent.throwException(e);
        }
    } else {
        // Go through each character in the name检查名称中的每个字符
        for (int index = 0; index < name.length(); ++index) {
            validateHeaderNameElement(name.charAt(index));
        }
    }
}
 

protected static boolean equals(byte[] bytes0, int offset0, int length0, byte[] bytes1,
    int offset1, int length1) {
  if (length0 != length1) {
    return false;
  }
  return PlatformDependent.equals(bytes0, offset0, bytes1, offset1, length0);
}
 

/**
 * Checks for the presence of an {@link Exception}.
 */
private ChannelFuture checkException(ChannelPromise promise) {
  Throwable t = lastException;
  if (t != null) {
    lastException = null;

    if (promise.isVoid()) {
        PlatformDependent.throwException(t);
    }

    return promise.setFailure(t);
  }

  return promise.setSuccess();
}
 

@Override
public void accumulate(final long memoryAddr, final int count) {
  final long maxAddr = memoryAddr + count * 4;
  final long incomingBit = getInput().getValidityBufferAddress();
  final long incomingValue =  getInput().getDataBufferAddress();

  int incomingIndex = 0;

  final ArrowBuf inputOffsetBuf = getInput().getOffsetBuffer();
  final ArrowBuf inputBuf = getInput().getDataBuffer();

  for(long ordinalAddr = memoryAddr; ordinalAddr < maxAddr; ordinalAddr += 4, incomingIndex++){
    final int bitVal = (PlatformDependent.getByte(incomingBit + ((incomingIndex >>> 3))) >>> (incomingIndex & 7)) & 1;

    //incoming record is null, skip it
    if (bitVal == 0) {
      continue;
    }

    //get the proper chunk from the ordinal
    final int tableIndex = PlatformDependent.getInt(ordinalAddr);
    //System.out.println("record idx: " + incomingIndex + " ordinal: " + tableIndex);
    final int chunkIndex = tableIndex >>> LBlockHashTableNoSpill.BITS_IN_CHUNK;
    final int chunkOffset = tableIndex & LBlockHashTableNoSpill.CHUNK_OFFSET_MASK;

    final HllAccumHolder ah =  this.accumulators[chunkIndex];
    final HllSketch sketch = ah.getAccums()[chunkOffset];

    //get the offset of incoming record
    final int startOffset = inputOffsetBuf.getInt(incomingIndex * BaseVariableWidthVector.OFFSET_WIDTH);
    final int endOffset = inputOffsetBuf.getInt((incomingIndex + 1) * BaseVariableWidthVector.OFFSET_WIDTH);
    final int len = endOffset - startOffset;
    final ByteBuffer buffer = inputBuf.nioBuffer(startOffset, len);

    //apply the update
    sketch.update(Memory.wrap(buffer), 0, len);
  } //for
}
 

private static final int memEqual(final long laddr, int lStart, int lEnd, final long raddr, int rStart,
    final int rEnd) {

  int n = lEnd - lStart;
  if (n == rEnd - rStart) {
    long lPos = laddr + lStart;
    long rPos = raddr + rStart;

    while (n > 7) {
      long leftLong = PlatformDependent.getLong(lPos);
      long rightLong = PlatformDependent.getLong(rPos);
      if (leftLong != rightLong) {
        return 0;
      }
      lPos += 8;
      rPos += 8;
      n -= 8;
    }
    while (n-- != 0) {
      byte leftByte = PlatformDependent.getByte(lPos);
      byte rightByte = PlatformDependent.getByte(rPos);
      if (leftByte != rightByte) {
        return 0;
      }
      lPos++;
      rPos++;
    }
    return 1;
  } else {
    return 0;
  }
}
 

static void setShort(long address, int value) {
    if (UNALIGNED) {
        PlatformDependent.putShort(
                address, BIG_ENDIAN_NATIVE_ORDER ? (short) value : Short.reverseBytes((short) value));
    } else {
        PlatformDependent.putByte(address, (byte) (value >>> 8));
        PlatformDependent.putByte(address + 1, (byte) value);
    }
}
 

@Override
public void endRow() {
  long rowHeaderPos = recordStartAddr();
  // curOffset is equivalent to a byte length of this row.
  PlatformDependent.putInt(rowHeaderPos, curOffset);

  //forward (record offset + fields offset)
  rowHeaderPos += SizeOf.SIZE_OF_INT + curFieldOffset;
  // set remain header field length
  for (int i = curFieldIdx; i < dataTypes.length; i++) {
    PlatformDependent.putInt(rowHeaderPos, MemoryRowBlock.NULL_FIELD_OFFSET);
    rowHeaderPos += SizeOf.SIZE_OF_INT;
  }
  curOffset = 0;
}
 

protected void startAfter(ChannelFuture future){
    //有异常抛出
    Throwable cause = future.cause();
    if(cause != null){
        PlatformDependent.throwException(cause);
    }
    logger.info("{} start (port = {}, pid = {}, os = {}) ...",
            getName(),
            getPort()+"",
            HostUtil.getPid()+"",
            HostUtil.getOsName());
}
 

@Override
public ByteBuf writeLong(long value) {
  BoundsChecking.ensureWritable(this, 8);
  PlatformDependent.putLong(addr(writerIndex), value);
  writerIndex += 8;
  return this;
}
 

private void initControlBlock(final long controlBlockAddr) {
  final long addr = controlBlockAddr;
  final long max = addr + MAX_VALUES_PER_BATCH * CONTROL_WIDTH;
  for (long l = addr; l < max; l += LBlockHashTable.CONTROL_WIDTH) {
    PlatformDependent.putLong(l, LBlockHashTable.LFREE);
  }
}
 

static void getBytes(AbstractByteBuf buf, long addr, int index, OutputStream out, int length) throws IOException {
    buf.checkIndex(index, length);
    if (length != 0) {
        ByteBuf tmpBuf = buf.alloc().heapBuffer(length);
        try {
            byte[] tmp = tmpBuf.array();
            int offset = tmpBuf.arrayOffset();
            PlatformDependent.copyMemory(addr, tmp, offset, length);
            out.write(tmp, offset, length);
        } finally {
            tmpBuf.release();
        }
    }
}
 

static void setShort(byte[] array, int index, int value) {
    if (UNALIGNED) {
        PlatformDependent.putShort(array, index,
                                   BIG_ENDIAN_NATIVE_ORDER ? (short) value : Short.reverseBytes((short) value));
    } else {
        PlatformDependent.putByte(array, index, (byte) (value >>> 8));
        PlatformDependent.putByte(array, index + 1, (byte) value);
    }
}
 

/**
 * Start of insertion. Record the state before insertion
 */
public void traceInsertStart(int numRecords) {
  if (traceBuf == null) {
    return;
  }
  PlatformDependent.putInt(traceBufNext + 0 * 4, capacity);
  PlatformDependent.putInt(traceBufNext + 1 * 4, maxSize);
  PlatformDependent.putInt(traceBufNext + 2 * 4, batches);
  PlatformDependent.putInt(traceBufNext + 3 * 4, currentOrdinal);
  PlatformDependent.putInt(traceBufNext + 4 * 4, rehashCount);
  PlatformDependent.putInt(traceBufNext + 5 * 4, numRecords);
  traceBufNext += 6 * 4;
}
 

public static ZlibDecoder newZlibDecoder(ZlibWrapper wrapper) {
    if (PlatformDependent.javaVersion() < 7 || noJdkZlibDecoder) {
        return new JZlibDecoder(wrapper);
    } else {
        return new JdkZlibDecoder(wrapper, true);
    }
}
 

@Test
public void testUnsafeHeapBufferAndUnsafeDirectBuffer() {
    T allocator = newUnpooledAllocator();
    ByteBuf directBuffer = allocator.directBuffer();
    assertInstanceOf(directBuffer,
            PlatformDependent.hasUnsafe() ? UnpooledUnsafeDirectByteBuf.class : UnpooledDirectByteBuf.class);
    directBuffer.release();

    ByteBuf heapBuffer = allocator.heapBuffer();
    assertInstanceOf(heapBuffer,
            PlatformDependent.hasUnsafe() ? UnpooledUnsafeHeapByteBuf.class : UnpooledHeapByteBuf.class);
    heapBuffer.release();
}
 

public int getVarKeyLength(int ordinal) {
  if (fixedOnly) {
    return 0;
  } else {
    final int blockWidth = pivot.getBlockWidth();
    final int dataChunkIndex = ordinal >>> BITS_IN_CHUNK;
    final long tableVarOffsetAddr = tableFixedAddresses[dataChunkIndex] + ((ordinal & CHUNK_OFFSET_MASK) * blockWidth) + blockWidth - VAR_OFFSET_SIZE;
    final int tableVarOffset = PlatformDependent.getInt(tableVarOffsetAddr);
    // VAR_LENGTH_SIZE is not added to varLen when pivot it in pivotVariableLengths method, so we need to add it here
    final int varLen = PlatformDependent.getInt(initVariableAddresses[dataChunkIndex] + tableVarOffset) + VAR_LENGTH_SIZE;
    return varLen;
  }
}