Netty Source code ByteBuf(4.1.44)

An overview of the

Netty abandoned Java NIO’s ByteBuffer because it had limited functionality and was too complex to use. As a result, Netty created its own data container similar to ByteBuffer, called ByteBuf, and provided a number of cool features:

  • Capacity can be dynamically expanded on demand.
  • Different Pointers are used for reading and writing, and can be switched freely without the user calling flip() to switch the read and write mode.
  • Zero copy technology is implemented with built-in compound buffer types.
  • Support reference counting.
  • Cache pooling is supported.
  • Rich type, support for off-heap memory and in-heap memory.

Inheritance diagram

The boxes in the figure are some types of ByteBuf objects commonly used in Netty. Their function is known by their name.

  • ByteBufImplement ReferenceCounted and Comparable interfaces, respectivelyReference countingThe ability to compare with two bytebuFs.
  • ByteBufIs aAn abstract class, but can be defined as an interface, but Netty officially defined as abstract analogy definition interface performance is higher. Most of these are abstract methods, consisting of a series of apis for manipulating ByteBuf, as described below.
  • AbstractByteBuf is an important abstract class that is the skeleton for implementing a Buffer. Overrides the abstract methods of most of the ByteBuf abstract classes to encapsulate the common logic of ByteBuf operations, such as checking whether an index is valid before fetching data, etc. AbstractByteBuf AbstractByteBuf AbstractByteBuf AbstractByteBuf AbstractByteBuf AbstractByteBuf AbstractByteBuf AbstractByteBuf
    • readerIndex
    • writerIndex
  • AbstractReferenceCountedByteBuf is also important to abstract class, the main implementationReference countingCorrelation logic. Internally maintain a volatile int refCnt variable. But all operations on refCnt need to be done via the ReferenceCountUpdater instance. AbstractReferenceCountedByteBuf subclass implementation can be too much. There are important
    • PooledByteBuf: abstract class, which is ownedpoolingThe capabilities of ByteBuf’s skeleton internally define a lot aboutpoolingRelated classes and variables. We’ll talk more about that later.
    • UnpooledHeapByteBuf: implementation class,Unpooled heap memory ByteBuf. Byte [] array is used internally to store data. The bottom layer uses the HeapByteBufUtil static method to operate on the array. In order to improve performance, it is worth understanding to use displacement method to process data.
    • CompositeByteBuf: implementation class,Combinable ByteBuf. The underlying implementation wraps ByteBuf with an inner Component class, then implements the composite pattern by using Component[] to store multiple Component objects.
    • UnpooledDirectByteBuf: implementation class,Unpooled out-of-heap ByteBuf. The bottom layer holds a reference to a Java.nio.byteBuffer object.
    • FixedCompositeByteBuf: implementation class,Fixed composable ByteBuf. Allows ByteBuf arrays to be wrapped in read-only mode.

The above briefly describes the ByteBuf inheritance system without involving too many details. It only gives readers an overview of the ByteBuf design concept. Remember: context first, details later.

ReferenceCounted

Define andReference countingRelated interfaces. The API is pretty simple, the kind you can read. Methods are typically implemented in abstract classesio.netty.util.AbstractReferenceCountedIn the finish. As for how to achieve this, we will continue to talk about it below.

ByteBuf

ByteBuf is a very, very important abstract class that plays a major role in the Netty ecosystem. As we know, Netty is a high-performance network framework for receiving/sending data. The container for receiving data is ByteBuf, which is an important part of Netty’s high-performance network framework. It is not that java.nio.ByteBuffer cannot be used directly, but it is relatively complex to program and weak, while ByteBuf has a rich API and is easy to use.

To translate documents

A random and sequential accessible sequence of zero or more bytes (octets). This interface provides an abstract view for one or more primitive byte arrays (byte[]) and NIO buffers.

A random and sequentially accessible sequence of zero or more bytes (octet bytes). This interface provides an abstract view of one or more basic byte arrays (byte []) and NIO buffers.

Create a buffer

  • It is recommended to construct a Buffer object using an Allocator rather than using a separate class constructor.

Random access index

  • It can be accessed randomly just like a normal array. The index from0Start, the last byte is zerocapacity-1.

Sequential access index

  • This capability is based on two PointersreaderIndex 和 writerIndex. They divide the byte array into three regions (the capacity variable represents the size of the buffer)

Read the data

  • Under any nameread 或 skipThe initial operations will either get or skip the data in the current readerIndex and readerIndex will increase the number of bytes read. If the read operation is also a ByteBuf instance and no destination index is specified, the specified writerIndex of ByteBuf is added together.
  • If there is no more space left, it will throwIndexOutOfBoundsExceptionThe exception.
  • A newly created, wrapped, and copied Buffer has a readerIndex of 0.

Write data

  • Any operation whose name begins with write writes data to the current writerIndex and increases the number of bytes written. If the argument being written is also an instance of ByteBuf, and no source index is specified, the readerIndex of the ByteBuf object for that parameter is added together.
  • If there is no more space left, it will throwIndexOutOfBoundsExceptionThe exception.
  • The newly created writerIndex has a value of 0.
  • The writerIndex value of the wrapped and copied Buffer is the capacity value of the current ByteBuf.

Discarding the data

  • Discard content that has already been read.
  • throughdiscardReadBytes()Reclaim new available data storage space.

Please note that

When discardReadBytes() is called, the contents of writable sections are not guaranteed. Writable sections will not be moved in most cases and may even be filled with completely different data, depending on the implementation of the underlying buffer.

Clearing buffer indexes

  • The clear() method sets readerIndex and writerIndex to 0. It does not clear the contents of the buffer, only resets the two Pointers.

The search index

  • simplesingle-byteSearch.
    • indexOf(int, int, byte)
    • bytesBefore(int, int, byte)
  • It is especially useful for nul-terminated strings
    • bytesBefore(byte)
  • Powerful search
    • forEachByte(int, int, ByteProcessor)

Mark/reset

  • There are two marker indexes in each buffer. The values are writerIndex and readerIndex.
  • You can always relocate one of the two indexes by calling the reset method.

Derivative Buffer

  • Create one for the current Buffer by calling the following methodView:
    • duplicate()
    • slice()
    • slice(int, int)
    • readSlice(int)
    • retainedDuplicate()
    • retainedSlice()
    • retainedSlice(int, int)
    • readRetainedSlice(int)
  • The types of Pointers to derived buffers are independent. But share buffer data. It’s like a NIO buffer.
  • If you need a new Buffer, call the copy() method.

Derived buffers that are unreserved and reserved

  • As we know, buffers have reference counts, Retain () does not increment the reference count when we call methods such as duplicate(), slice(), slice(int, int) and readSlice(int). Do not let it be released). If you need to maintain a reference to this Buffer and not let it be released, Consider using retainedDuplicate(), retainedSlice(), retainedSlice(int, int), and readRetainedSlice(int), which return a Buffer that produces less garbage.

Byte[]

  • If a Buffer has a supporting array, it can be accessed directly through the array() method.
  • The hasArray() method determines whether the current Buffer has a supporting array. After calling discardReadBytes()

NIO Buffers

  • If a ByteBuf can be converted to a NIO ByteBuffer object, the object shares the content. You can get the NIO ByteBuffer object by using the nioBuffer() method.
  • The nioBufferCount() method determines whether a Buffer can be converted to a NIO ByteBuffer.

I/O streams

  • ByteBufInputStream
  • ByteBufOutputStream

We’ve had an overview of the Bytebuf-related apis, which, to be honest, is too much. It doesn’t have the same design ideals as Java.nio.Buffer.

  • Netty’s ByteBuf integrates all operations (apis) together, such as getBoolean(), getByte(), getLong(), and so on.
  • Java.nio.buffer, on the other hand, is decoupled through multiple classes. For example, for the basic Byte type, we first inherit Buffer using java.nio.byteBuffer objects, and then define abstract methods related to Byte to be implemented by subclasses. Which is better is a matter of opinion. Java uses good design patterns to decouple the functions of buffers, but there are plenty of classes. Netty is more compact and uses ByteBuf to unify other data types (such as int and long), probably thinking that the latter are composed of the smallest unit of bytes. That’s why the API is designed this way.

Record some ByteBuf API usage instructions

// Discard all read data immediately (data copy is required to copy the unread content to the front)
// Discard even if there is only one byte left to write
public abstract ByteBuf discardReadBytes(a);

// Determine whether the readerIndex pointer exceeds half of capacity
// If it is exceeded, the discard action is performed
// This method is a bit smarter than discardReadBytes()
public abstract ByteBuf discardSomeReadBytes(a);

// Make sure minWritableBytes are writable
// If the capacity is insufficient, the expansion action is triggered
// Expanded capacity range [64Byte, 4MB]
public abstract ByteBuf ensureWritable(int minWritableBytes);

Return a value of type int
// 0: Current ByteBuf has sufficient writable capacity. Capacity remains unchanged
// 1: Current ByteBuf does not have sufficient writable capacity. Capacity remains unchanged
// 2: Current ByteBuf has sufficient writable capacity. Capacity increases
// 3: Currently, ByteBuf does not have enough writable capacity, but capacity has reached the maximum value
public abstract int ensureWritable(int minWritableBytes, boolean force);

/** * The set/get method is used to treat the underlying data as an array of bytes, and the * index values grow according to the length of the base type. * Set /get does not change the values of readerIndex and writerIndex. * You can think of it as a bit change operation. I don't understand this very well at this stage */
public abstract int   getInt(int index);
public abstract int   getIntLE(int index); * method getBytes (int, ByteBuf, int.int) can do the same thing. * The difference is that the "current method" increments the "writerIndex" value of the target Buffer object, and * getBytes(int, ByteBuf, int.int) method does not change./** * Transfers data from this Buffer to the specified target Buffer object, starting at the specified absolute index, until the target object becomes unwritable. * * "writerIndex" "readerIndex" * Data source: do not modify * Target object: add "writerIndex" * *@paramIndex Indicates the index value *@paramDST Target object *@returnThe source object * /
public abstract ByteBuf getBytes(int index, ByteBuf dst);

GetBytes (int, ByteBuf, int, int) does the same thing by transferring data from the specified absolute index to the specified target Buffer. * * "writerIndex" "readerIndex" * Data source: do not modify * Target object: add "writerIndex" *@paramIndex Indicates the index value *@paramDST Target object *@paramLength Indicates the copy length *@returnThe source object * /
public abstract ByteBuf getBytes(int index, ByteBuf dst, int length);

/** * Copy data to target array ** "writerIndex" "readerIndex" * data source: do not modify * Target object: none */
public abstract ByteBuf getBytes(int index, byte[] dst);

/** * Copy data to target array ** "writerIndex" "readerIndex" * data source: do not modify * Target object: none */
public abstract ByteBuf getBytes(int index, byte[] dst, int dstIndex, int length);

/** * Copy data to target array ** "writerIndex" "readerIndex" * data source: do not modify * Target object: add "writerIndex" */
public abstract ByteBuf getBytes(int index, ByteBuffer dst);

/** * Copy data to target object * the above methods for copying data to a ByteBuf object end up calling this method to copy data ** "writerIndex" "readerIndex" * Data source: none changes * Target object: none changes */
public abstract ByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length);

/** * Copy data to destination stream ** "writerIndex" "readerIndex" * Data source: do not modify * Target object: none */
public abstract ByteBuf getBytes(int index, OutputStream out, int length) throws IOException;

/** * Copy data to specified channel ** "writerIndex" "readerIndex" * data source: do not modify * Target object: none */
public abstract int getBytes(int index, GatheringByteChannel out, int length) throws IOException;

/** * Copies data to a specified channel without modifying the channel's "position" ** "writerIndex" "readerIndex" * Data source: none * Target object: none */
public abstract int getBytes(int index, FileChannel out, long position, int length) throws IOException;

/** * copy SRC's "writerIndex-readerIndex" to this.ByteBuf * The rest of the arguments with a ByteBuf object will follow the same logic. * setBytes(int index, ByteBuf SRC, int srcIndex, int Length) is a bit different * this method does not modify the values of either pointer variable. * * "writerIndex" "readerIndex" * SRC: increase the value of "readerIndex" * this: do not modify */
public abstract ByteBuf setBytes(int index, ByteBuf src);
public abstract ByteBuf setBytes(int index, ByteBuf src, int length);
public abstract ByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length);
public abstract ByteBuf setBytes(int index, byte[] src);
public abstract ByteBuf setBytes(int index, byte[] src, int srcIndex, int length);
public abstract ByteBuf setBytes(int index, ByteBuffer src);
public abstract int setBytes(int index, InputStream in, int length) throws IOException;
public abstract int setBytes(int index, ScatteringByteChannel in, int length) throws IOException;
public abstract int setBytes(int index, FileChannel in, long position, int length) throws IOException;
// Fill with NUL(0x00)
public abstract ByteBuf setZero(int index, int length);

/** * the following is the read operation * readerIndex grows according to the corresponding type. * Such as readerIndex+1 for readByte() and readerIndex+2 */ for readShort()
public abstract byte  readByte(a);
public abstract short readShort(a);
public abstract short readShortLE(a);
public abstract int   readUnsignedShort(a);
public abstract int   readUnsignedShortLE(a);
public abstract int   readMedium(a);

/** * Transfer the buffer's data to a newly created buffer, starting with the current readerIndex, and increment the readerIndex by the number of bytes (length) transferred. * The readerIndex and writerIndex of the returned buffer are 0 and length, respectively. * *@returnA newly created ByteBuf object */
public abstract ByteBuf readBytes(int length);

/** * returns a new ByteBuf object. It is a wrapper object with a reference to the source Buffer. * This object is just a view with several Pointers independent of the source Buffer * but readerIndex(0) and writerIndex(=length) values are initial. * Also, note that the current method does not call retain() to increase the reference count *@returnA newly created ByteBuf object */
public abstract ByteBuf readSlice(int length);
public abstract ByteBuf readRetainedSlice(int length);


/** * Reads data to DST until it is unreadable. * * "writerIndex" "readerIndex" * DST: increases the value of "writerIndex" * this: increases the value of "readerIndex" *@returnA newly created ByteBuf object */
public abstract ByteBuf readBytes(ByteBuf dst);
public abstract ByteBuf readBytes(ByteBuf dst, int length);

/** * Reads data to DST until it is unreadable. * * "writerIndex" "readerIndex" * DST: do not modify * this: do not modify *@returnA newly created ByteBuf object */
public abstract ByteBuf readBytes(ByteBuf dst, int dstIndex, int length);

public abstract CharSequence readCharSequence(int length, Charset charset);
public abstract int readBytes(FileChannel out, long position, int length) throws IOException;
public abstract ByteBuf skipBytes(int length);


/** * writes to memory with subscript writerIndex. * If the capacity is insufficient, an attempt will be made to expand the capacity * * "writerIndex" "readerIndex" * DST: None * This: "writerIndex" + 1 *@returnA newly created ByteBuf object */
public abstract ByteBuf writeByte(int value);

/** * writes to memory with subscript writerIndex. * If the capacity is insufficient, an attempt will be made to expand the capacity * * "writerIndex" "readerIndex" * DST: None * This: "writerIndex" + 1 *@returnA newly created ByteBuf object */
public abstract ByteBuf writeBytes(ByteBuf src);
public abstract ByteBuf writeBytes(ByteBuf src, int length);
public abstract ByteBuf writeBytes(ByteBuf src, int srcIndex, int length);
public abstract ByteBuf writeBytes(byte[] src);
public abstract ByteBuf writeBytes(byte[] src, int srcIndex, int length);
public abstract ByteBuf writeBytes(ByteBuffer src);
public abstract int writeBytes(FileChannel in, long position, int length) throws IOException;
public abstract ByteBuf writeZero(int length);
public abstract int writeCharSequence(CharSequence sequence, Charset charset);

/** * Find value from fromIndex to toIndex and return index *@returnThe first occurrence of the location index, -1 indicates that */ was not found
public abstract int indexOf(int fromIndex, int toIndex, byte value);

/** * positions the first match of the specified value in this buffer. * Search range [readerIndex, writerIndex]. * *@return-1 indicates that */ is not found
public abstract int bytesBefore(byte value);

/** * search scope [readerIndex, readerIndex + length] **@return-1 indicates that * * is not found@throws IndexOutOfBoundsException
 */
public abstract int bytesBefore(int length, byte value);

/** * search scope [index, idnex+length] **@return-1 indicates that * * is not found@throws IndexOutOfBoundsException
 */
public abstract int bytesBefore(int index, int length, byte value);

/** * Iterates the "readable bytes" of the buffer in ascending order with the specified processor **@return-1 indicates not found. If byteProcessor.process (byte) returns false, the last accessed index value */ is returned
public abstract int forEachByte(ByteProcessor processor);

/** * Iteration range [index, index+length-1) */
public abstract int forEachByte(int index, int length, ByteProcessor processor);
public abstract int forEachByteDesc(ByteProcessor processor);
public abstract int forEachByteDesc(int index, int length, ByteProcessor processor);

/** * returns a copy of the buffer's readable bytes. The two bytebufs are content independent. * Similar to buf.copy(buf.readerIndex(), buf.readableBytes()); * No pointer to the source ByteBuf will be modified */
public abstract ByteBuf copy(a);
public abstract ByteBuf copy(int index, int length);

/** * returns a fragment of the buffer's readable bytes. * Modifying the returned buffer or the contents of this buffer affects each other's contents, and they maintain separate indexes and tags. * This method is the same as buf.slice (buf.readerIndex (), buf.readableBytes (). * This method does not modify the readerIndex or writerIndex of this buffer. * /
public abstract ByteBuf slice(a);

/** * the same behavior as slice().retain() */
public abstract ByteBuf retainedSlice(a);
public abstract ByteBuf slice(int index, int length);
public abstract ByteBuf retainedSlice(int index, int length);

/** * Content sharing. Tags that maintain separate indexes. * The new ByteBuf readable content is the same as that returned by the slice() method. But because the underlying ByteBuf object is shared, * everything at the bottom is visible. * The read and write flags are not duplicated. Also note that this method does not call retain() to add +1 */ to the reference count
public abstract ByteBuf duplicate(a);
public abstract ByteBuf retainedDuplicate(a);

/** * Returns the maximum number of NIO bytebuffers that make up this buffer. The general default is 1, and the sum is computed for combined bytebufs. * *@return-1 indicates that there is no underlying ByteBuf *@see #nioBuffers(int, int)
 */
public abstract int nioBufferCount(a);

/** * Exposes the readable bytes of this buffer as NIO ByteBuffer. Share content. * buf.niobuffer (buf.readerIndex(), buf.readableBytes())) has the same result. * Note that if the buffer is a dynamic buffer and its capacity is adjusted, the returned NIO buffer will not see these changes */
public abstract ByteBuffer nioBuffer(a);
public abstract ByteBuffer nioBuffer(int index, int length);

/** * For internal use only: Exposes the internal NIO buffer. * /
public abstract ByteBuffer internalNioBuffer(int index, int length);
public abstract ByteBuffer[] nioBuffers();
public abstract ByteBuffer[] nioBuffers(int index, int length);

/** * Returns true */ if the current ByteBuf has supporting data
public abstract boolean hasArray(a);
public abstract byte[] array();

/** * returns the offset of the first byte in this buffer's support byte array. * /
public abstract int arrayOffset(a);

/** * Returns true if and only if this buffer has a reference to the low-level memory address of "backing data"
public abstract boolean hasMemoryAddress(a);
public abstract long memoryAddress(a);

/** * Returns true if this ByteBuf is internal to a single memory region. A compound type buffer must return false, even if it contains only a ByteBuf object. * /
public boolean isContiguous(a) {
    return false;
}

public abstract String toString(Charset charset);

public abstract String toString(int index, int length, Charset charset);

@Override
public abstract int hashCode(a);

@Override
public abstract boolean equals(Object obj);

@Override
public abstract int compareTo(ByteBuf buffer);

@Override
public abstract String toString(a);

@Override
public abstract ByteBuf retain(int increment);

@Override
public abstract ByteBuf retain(a);

@Override
public abstract ByteBuf touch(a);

@Override
public abstract ByteBuf touch(Object hint);

boolean isAccessible(a) {
    returnrefCnt() ! =0;
}
Copy the code
  • getXX()Copy data from the source Buffer to the target Buffer. It may be modifiedThe target BufferWriterIndex.
  • setXX()Copies data from the target Buffer to the source Buffer. It may be modifiedThe target BufferReaderIndex.
  • readXX()Represents reading data from a Buffer, which grows according to the base typeThe source BufferReaderIndex.
  • Get and set are both relative to this, such as this.getxx (), which means to get information about this.buffer and copy it to the target ByteBuf object. This.setxx (), in turn, copies data from the target ByteBuf object to this.buffer.

AbstractByteBuf

It is the basic framework of ByteBuf, which implements most of the abstract methods of IO.net ty.buffer.ByteBuf. Subclasses only need to implement corresponding abstract methods according to specific functions. The io.net ty. Buffer. AbstractByteBuf abstract class, do the following things:

  • Define and maintain five specified variables. These are readerIndex, writerIndex, markedReaderIndex, markedWriterIndex, and maxCapacity. Therefore, the main job of this abstract class is to maintain these five variables. For example, check whether it is satisfied before the getXX() method, and so on.
  • Initialize theResourceLeakDetector<ByteBuf>Memory leak detection object. It records various ByteBuf usage of Netty and monitors the objects that occupy resources, whether they are pooled or not, whether they are out of the heap or in the heap. There are 4 levels available:DISABLED, SIMPLE, ADVANCED, and PARANOID. The monitoring level changes from low to high. The higher the monitoring level, the more ByteBuFs that can be monitored, the more information that can be obtained, and the greater the impact on performance. You are advised to use ADVANCED or PARANOID in DEBUG mode and SMPLE in the production environment. In fact, the implementation logic is relatively simple, which is to wrap the ByteBuf object, record the necessary data during the execution of the relevant API, and then analyze where the memory leak occurs based on this data, and inform the user to troubleshoot through the log.

Excerpt part API

// io.netty.buffer.AbstractByteBuf
public abstract class AbstractByteBuf extends ByteBuf {
    static final ResourceLeakDetector<ByteBuf> leakDetector =
        ResourceLeakDetectorFactory.instance().newResourceLeakDetector(ByteBuf.class);
    int readerIndex;
    int writerIndex;
    private int markedReaderIndex;
    private int markedWriterIndex;
    private int maxCapacity;
    
    @Override
    public ByteBuf setByte(int index, int value) {
        checkIndex(index);
        _setByte(index, value);
        return this;
    }

    protected abstract void _setByte(int index, int value);

    @Override
    public byte getByte(int index) {
        checkIndex(index);
        return _getByte(index);
    }

    protected abstract byte _getByte(int index);

    @Override
    public byte readByte(a) {
        checkReadableBytes0(1);
        int i = readerIndex;
        byte b = _getByte(i);
        readerIndex = i + 1;
        return b;
    }
    // ...
}
Copy the code

AbstractReferenceCountedByteBuf

The ReferenceCountUpdater object is used internally to add/subtract refCnt. The only entry to the refCnt is the updater object. The internal implementation is relatively neat, since all operations are delegated to the ReferenceCountedByteBuf object. The relevant source code is as follows:

// io.netty.buffer.AbstractReferenceCountedByteBuf
public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
    // Get the offset of the variable "refCnt", underlying changed via Unsafe
    private static final long REFCNT_FIELD_OFFSET =
            ReferenceCountUpdater.getUnsafeOffset(AbstractReferenceCountedByteBuf.class, "refCnt");
    private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> AIF_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
	
    // The core object, the unique entry to the operation variable refCnt
    private static final ReferenceCountUpdater<AbstractReferenceCountedByteBuf> updater =
            new ReferenceCountUpdater<AbstractReferenceCountedByteBuf>() {
        @Override
        protected AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> updater(a) {
            return AIF_UPDATER;
        }
        @Override
        protected long unsafeOffset(a) {
            returnREFCNT_FIELD_OFFSET; }};// Value might not equal "real" reference count, all access should be via the updater
    @SuppressWarnings("unused")
    private volatile int refCnt = updater.initialValue();
    
    // ...
}
Copy the code

ReferenceCountUpdater

ReferenceCountUpdater is performed on ByteBuf that implements the ReferenceCounted interfaceReference countingRelated operations. Bottom through magic classjava.util.concurrent.atomic.AtomicIntegerFieldUpdaterTo complete the evaluation of the valueIncrease/decreaseOperation. So somebody said, why not just use AtomicLong? Don’t ask. Asking improves performance. Indeed, if you use an AtomicLong object, its memory overhead is higher than that of a base variable (I guess). Regarding the logic of ReferenceCountUpdater, we have the following cognition:

  • Each newly born ByteBuf object has a refCnt value of 2. Therefore, whenever reference counting logic +1, the corresponding refCnt physical +2, and whenever reference counting logic -1, the corresponding refCnt physical -2. Therefore, the internal reference count value is as long as there is a referenceAn even number. Can be passedrefCnt&1 == 0?To determine whether a reference is held.
  • In addition to this judgment, there are many places where bitwise operations can improve performance. It’s not much, but it’s the ultimate optimization.
  • The formularealCount = value>>>1Get count referenceBoolean value.
/** * Release ByteBuf, refcnt-2 **@param instance
 * @return* /
public final boolean release(T instance) {

    // #1 Retrieves the value of the current object's variable refCnt through Unsafe, which is non-atomic
    int rawCnt = nonVolatileRawCnt(instance);

    // #2 rawCnt==2(): refCnt = 1
    // tryFinalRelease0: Try only once, using CAS to set the refCnt value to 1
    // Attempt fails, retryRelease0, for(;) To update the value referenced by the count until success
    // rawCnt ! = 2, indicating that the release was not a complete release,
    return rawCnt == 2 ? tryFinalRelease0(instance, 2) || retryRelease0(instance, 1)
            : nonFinalRelease0(instance, 1, rawCnt, toLiveRealRefCnt(rawCnt, 1));
}

// io.netty.util.internal.ReferenceCountUpdater#tryFinalRelease0
private boolean tryFinalRelease0(T instance, int expectRawCnt) {
    // Change the refCnt value from the expected value expectRawCnt to 1
    return updater().compareAndSet(instance, expectRawCnt, 1); // any odd number will work
}

// io.netty.util.internal.ReferenceCountUpdater#retryRelease0
/** * Attempt to release the refCnt value of instance: logical -1, physical -2 */
private boolean retryRelease0(T instance, int decrement) {
    for (;;) {
        // #1 get the physical value of refCnt
        // Get the actual number of references, if odd, throw an exception,
        // Since there is no reference to ByteBuf, there is no release
        int rawCnt = updater().get(instance);
        
        // #2 get the refCnt logical value
        int realCnt = toLiveRealRefCnt(rawCnt, decrement);

        ByteBuf =realCnt; refCnt=1
        if (decrement == realCnt) {
            if (tryFinalRelease0(instance, rawCnt)) {
                return true; }}else if (decrement < realCnt) {
            // If the subtraction is less than the actual value, rawCNT-2 is updated
            if (updater().compareAndSet(instance, rawCnt, rawCnt - (decrement << 1))) {
                return false; }}else {
            // Otherwise an exception is thrown
            throw new IllegalReferenceCountException(realCnt, -decrement);
        }
        // In high concurrency cases, this helps improve throughput
        // Thread concession: Worried that the current thread is consuming too much CPU resources, the main reason is to change itself from the execution state to the ready state and compete with other threadsThread.yield(); }}public final boolean release(T instance, int decrement) {
    int rawCnt = nonVolatileRawCnt(instance);
    int realCnt = toLiveRealRefCnt(rawCnt, checkPositive(decrement, "decrement"));
    return decrement == realCnt ? tryFinalRelease0(instance, rawCnt) || retryRelease0(instance, decrement)
            : nonFinalRelease0(instance, decrement, rawCnt, realCnt);
}


// io.netty.util.internal.ReferenceCountUpdater#realRefCnt
// Get the true count
private static int realRefCnt(int rawCnt) {
    // (rawCnt & 1) ! = 0 to check whether it is an even number, even references will exist
    returnrawCnt ! =2&& rawCnt ! =4 && (rawCnt & 1) != 0 ? 0 : rawCnt >>> 1;
}

// io.netty.util.internal.ReferenceCountUpdater#retain0
/** * Maintains a reference to the ByteBuf object. * Logical +1, physical refCnt+2 *@paramThe instance instance *@paramIncrement *@paramRawIncrement Original value *@return* /
private T retain0(T instance, final int increment, final int rawIncrement) {
    
    // #1 gets the old value and increments the original value of the reference counter
    int oldRef = updater().getAndAdd(instance, rawIncrement);
    
    // #2 validates the old value
    // If the value is odd, the current ByteBuf object has been released and the reference to the freed ByteBuf object cannot be maintained
    if(oldRef ! =2&& oldRef ! =4 && (oldRef & 1) != 0) {
        throw new IllegalReferenceCountException(0, increment);
    }

    // #3 overflow processing
    // For example
    // oldRef =-1173741824, increment=1003741824 rawIncrement=2007483648
    // oldRef =2, increment=1103741824
    if ((oldRef <= 0 && oldRef + rawIncrement >= 0)
            || (oldRef >= 0 && oldRef + rawIncrement < oldRef)) {
        / / correction
        updater().getAndAdd(instance, -rawIncrement);
        // Throw an exception
        throw new IllegalReferenceCountException(realRefCnt(oldRef), increment);
    }
    
    / / # 4 to return
    return instance;
}
Copy the code

ReferenceCountUpdater is an abstract class, internal method to refCnt operation logic. Eventually to volatinle int refCnt assignment operation is accomplished by AtomicIntegerFieldUpdater. However, this object cannot be held in memory, so provide an abstract method given by the corresponding user:

// io.netty.util.internal.ReferenceCountUpdater
/ / get AtomicIntegerFieldUpdater class, the class is to provide the JDK concurrency change class of data
protected abstract AtomicIntegerFieldUpdater<T> updater(a);

// the offset address of the refCnt variable
protected abstract long unsafeOffset(a);
Copy the code

On the io.net ty. Buffer. AbstractReferenceCountedByteBuf abstraction is this:

// io.netty.buffer.AbstractReferenceCountedByteBuf
public abstract class AbstractReferenceCounted implements ReferenceCounted {
    // 
    private static final long REFCNT_FIELD_OFFSET =
            ReferenceCountUpdater.getUnsafeOffset(AbstractReferenceCounted.class, "refCnt");
    private static final AtomicIntegerFieldUpdater<AbstractReferenceCounted> AIF_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCounted.class, "refCnt");

    private static final ReferenceCountUpdater<AbstractReferenceCounted> updater =
            new ReferenceCountUpdater<AbstractReferenceCounted>() {
        / / get AtomicIntegerFieldUpdater object, the underlying use this object to manipulate properties
        @Override
        protected AtomicIntegerFieldUpdater<AbstractReferenceCounted> updater(a) {
            return AIF_UPDATER;
        }
        
        // Get the offset of the attribute variable refCnt
        @Override
        protected long unsafeOffset(a) {
            returnREFCNT_FIELD_OFFSET; }};// ...
}
Copy the code

Look at the source, I believe how to abstract classes AbstractReferenctCountedByteBuf for volatile variables int refCnt operation! The bottom is done through AtomicIntegerFieldUpdater finally.

AtomicIntegerFieldUpdater

A reflection-based utility written by concurrency guru Doug Lea that makes atomic updates to specified volatile int fields of specified classes. This class is designed for atomic data structures where multiple fields of the same node are updated atomically independently. Note that the guarantee of the compareAndSet method in this class is weaker than in other atomic classes. Because this class cannot ensure that all uses of the field are suitable for atomic access, it can only guarantee atomicity for other calls to compareAndSet, set on the same updater.

About the Unsafe

Unsafe’s approach to accessing object fields abstracts the layout of objects, providing the Sun.misc.Unsafe#objectFieldOffset method to retrieve the offset of a byte relative to the “start address” of a Java object. Methods such as getInt, getLong, and getObject are also provided to access the value of a variable by passing in its offset. See TODO for more.

AbstractReferenceCountedByteBuf subclass implementation

AbstractReferenceCountedByteBuf subclasses of basic included in our Netty ByteBuf examples of the most common type.

Talk more about de-pooled ByteBuf

So let’s take a lookThe pooling ByteBufRelated inheritance system: The pooling ByteBufWill fall into two categories, respectively

  • UnpooledHeapByteBuf
  • UnpooledDirectByteBuf

By comparing the

  • getByte(int);
  • getBytes(int, ByteBuf, int, int);
  • setBytes(int, int);
  • setBytes(int, ByteBuf, int, int);
  • deallocate();

The API compares two places that are the same and different.

UnpooledHeapByteBuf

// io.netty.buffer.UnpooledHeapByteBuf
/** * unpooled heap memory, using byte[] arrays to store data */
public class UnpooledHeapByteBuf extends AbstractReferenceCountedByteBuf {
	
    // Each ByteBuf object holds a reference to the ByteBufAllocator locator that created the ByteBuf
    The alloc() method returns the allocator corresponding to the current ByteBuf
    // Therefore, memory allocators can be obtained from any ByteBuf object, which can be used to allocate memory or do some judgment things like that
    private final ByteBufAllocator alloc;
    
    // Where the data is stored
    byte[] array;
    // ByteBuffer belongs to Java NIO, using Netty ByteBuf to ByteBuffer conversion
    private ByteBuffer tmpNioBuf;
    
    This method overrides the abstract class "AbstractByteBuf" method. * "AbstractByteBuf" contains the "refCnt" check and index check using checkIndex(index), while the current object only contains ensureAccessible() check */
    @Override
    public byte getByte(int index) {
        
        // #1 determine whether to read
        ensureAccessible();
        
        // #2 read index
        return _getByte(index);
    }

    @Override
    protected byte _getByte(int index) {
        HeapByteBufUtil (array[index])
        return HeapByteBufUtil.getByte(array, index);
    }

    /** * Sends data of length to the specified destination, starting with the specified absolute index value. * This method does not modify the readerIndex, writerIndex Pointers of source ByteBuf. * This API needs to be implemented by subclasses *@paramIndex Indicates the index value * of the source ByteBuf object@paramDST Indicates the destination ByteBuf object *@paramDstIndex Indicates the value of destination ByteBufindex *@paramThe length of the length *@return* /
    @Override
    public ByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
        
        // #1 Check whether index, length, capacity, refCnt meet the requirements
        checkDstIndex(index, length, dstIndex, dst.capacity());
        
        // #2 Use different copy policies according to the target ByteBuf type
        if (dst.hasMemoryAddress()) {
            // #2-1 If the target ByteBuf type contains memoryAddress, related to Unsafe,
            // Over "Unsafe", for copying data
            PlatformDependent.copyMemory(array, index, dst.memoryAddress() + dstIndex, length);
        } else if (dst.hasArray()) {
            // #2-2 if the target class ByteBuf type contains "support array", then it is associated with byte[] byte array,
            // Use the local system.arrayCopy () method to copy data
            getBytes(index, dst.array(), dst.arrayOffset() + dstIndex, length);
        } else {
            // #2-3 if neither of the two methods is used, the corresponding ByteBuf#setBytes method is called to complete the data replication
            // Target ByteBuf selects the appropriate method copy based on its implementation
            dst.setBytes(dstIndex, array, index, length);
        }
        return this;
    }
    
    /** ** Assignment */
    @Override
    public ByteBuf setByte(int index, int value) {
        ensureAccessible();
        _setByte(index, value);
        return this;
    }
    
    @Override
    protected void _setByte(int index, int value) {
        // Array [index]=value
        HeapByteBufUtil.setByte(array, index, value);
    }
    
    /** * Roughly the same as getBytes(int index, ByteBuf DST, int dstIndex, int length)
    @Override
    public ByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length) {
        
        // #1 Check whether index, length, capacity, refCnt meet the requirements
        checkSrcIndex(index, length, srcIndex, src.capacity());

        // #2 Use different copy policies for different data source ByteBuf instances
        if (src.hasMemoryAddress()) {
			// #2-1 with the "Unsafe" declaration
            PlatformDependent.copyMemory(src.memoryAddress() + srcIndex, array, index, length);
        } else  if (src.hasArray()) {
            // #2-2 complete with system.arrayCopy ()
            setBytes(index, src.array(), src.arrayOffset() + srcIndex, length);
        } else {
            // #2-3 do this with the "ByteBuf#getBytes" method
            src.getBytes(srcIndex, array, index, length);
        }
        return this;
    }
    
    /** * Release the ByteBuf object */
    @Override
    protected void deallocate(a) {
        freeArray(array);
        array = EmptyArrays.EMPTY_BYTES;
    }
    
    // Use GC for garbage collection
    protected void freeArray(byte[] array) {
        // NOOP}}// io.netty.util.internal.MathUtil#isOutOfBounds
public static boolean isOutOfBounds(int index, int length, int capacity) {
    return (index | length | (index + length) | (capacity - (index + length))) < 0;
}

// io.netty.util.internal.PlatformDependent#copyMemory(byte[], int, long, long)
public static void copyMemory(byte[] src, int srcIndex, long dstAddr, long length) {
    PlatformDependent0.copyMemory(src, BYTE_ARRAY_BASE_OFFSET + srcIndex, null, dstAddr, length);
}

// io.netty.util.internal.PlatformDependent0#copyMemory(java.lang.Object, long, java.lang.Object, long, long)
static void copyMemory(Object src, long srcOffset, Object dst, long dstOffset, long length) {
    // Manual safe-point polling is only needed prior Java9:
    // See https://bugs.openjdk.java.net/browse/JDK-8149596
    // Both layers copy data through unsafe.copymemory (), but due to JDK versions lower than 1.9, data is not fully copied due to safety points
    // Netty fixes this BUG by ensuring full copy with while+length
    if (javaVersion() <= 8) {
        copyMemoryWithSafePointPolling(src, srcOffset, dst, dstOffset, length);
    } else{ UNSAFE.copyMemory(src, srcOffset, dst, dstOffset, length); }}// io.netty.util.internal.PlatformDependent0#copyMemoryWithSafePointPolling
private static void copyMemoryWithSafePointPolling(
    Object src, long srcOffset, Object dst, long dstOffset, long length) {
    while (length > 0) {
        longsize = Math.min(length, UNSAFE_COPY_THRESHOLD); UNSAFE.copyMemory(src, srcOffset, dst, dstOffset, size); length -= size; srcOffset += size; dstOffset += size; }}// io.netty.buffer.HeapByteBufUtil
static void setByte(byte[] memory, int index, int value) {
    memory[index] = (byte) value;
}

Copy the code

The bottom layer of UnpooledHeapByteBuf is an array. According to the target ByteBuf object, different policies are selected to complete the read/write operations. Because the underlying data is stored by data, the implementation is relatively simple.

The subclass implementation

UnpooledHeapByteBuf has two subclass implementations:

  • UnpooledUnsafeHeapByteBuf: use UnsafeByteBufUtil management byte [] array. It also has a subclass
    • InstrumentedUnpooledUnsafeHeapByteBuf: belongs to the class of UnpooledByteBufAllocator private,
  • InstrumentedUnpooledHeapByteBuf: Instrumented said increased some device, is actually in the memory allocation and release after modified allocated memory size, used for memory management. With an Instrumented prefix has a total of five classes, they are all UnpooledByteBufAllocatory distributor private classes. In general, ByteBuf objects are created through allocators. One of the benefits is unified entry, so we can do some data logging:Allocated memory size. The value of the requested memory size is added when allocating memory, and the value of the corresponding returned memory size is subtracted when freeing memory. The corresponding 5 classes are as follows:
    • InstrumentedUnpooledUnsafeHeapByteBuf
    • InstrumentedUnpooledHeapByteBuf
    • InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf
    • InstrumentedUnpooledUnsafeDirectByteBuf
    • InstrumentedUnpooledDirectByteBuf

UnpooledDirectByteBuf

// io.netty.buffer.UnpooledDirectByteBuf
/** * Non-pooled out-of-heap memory, based on "java.nio.byteBuffer" implementation. * it is recommended to use "UnpooledByteBufAllocator. DirectBuffer (int, Int) "*" unpooled.directBuffer (int) "*" unpooled.wrappedBuffer (ByteBuffer) "to display allocation of ByteBuf objects instead of directly creating ByteBuf objects using the constructor */
public class UnpooledDirectByteBuf extends AbstractReferenceCountedByteBuf {

    private final ByteBufAllocator alloc;

    ByteBuffer buffer; // accessed by UnpooledUnsafeNoCleanerDirectByteBuf.reallocateDirect()
    private ByteBuffer tmpNioBuf;
    private int capacity;
    private boolean doNotFree;
    
    @Override
    public byte getByte(int index) {
        ensureAccessible();
        return _getByte(index);
    }

    @Override
    protected byte _getByte(int index) {
        return buffer.get(index);
    }
    
    @Override
    public short getShort(int index) {
        ensureAccessible();
        return _getShort(index);
    }

    @Override
    protected short _getShort(int index) {
        return buffer.getShort(index);
    }
    
    /** * copy the data in this.bytebuf to the target object "DST" */
    @Override
    public ByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
        
        // #1 Check strictly
        checkDstIndex(index, length, dstIndex, dst.capacity());
        
        // #2 Different replication strategies are adopted according to different target object implementations
        if (dst.hasArray()) {
            // #2-1 if the target ByteBuf holds a "support array", then the data copy is passed to the ByteBuffer#get() method.
            // Its underlying layer is also copied one by one through the for loop.
            getBytes(index, dst.array(), dst.arrayOffset() + dstIndex, length);
        } else if (dst.nioBufferCount() > 0) {
            // #2-2 consists of bytebuffers, which copy data one by one
            for (ByteBuffer bb: dst.nioBuffers(dstIndex, length)) {
                intbbLen = bb.remaining(); getBytes(index, bb); index += bbLen; }}else {
            In turn, the dst#setBytes method is called to copy the data
            dst.setBytes(dstIndex, this, index, length);
        }
        return this;
    }
    
    @Override
    public ByteBuf setByte(int index, int value) {
        ensureAccessible();
        _setByte(index, value);
        return this;
    }

    @Override
    protected void _setByte(int index, int value) {
        // Pass to ByteBuffer
        buffer.put(index, (byte) value);
    }
    
    @Override
    public ByteBuf setShort(int index, int value) {
        ensureAccessible();
        _setShort(index, value);
        return this;
    }

    @Override
    protected void _setShort(int index, int value) {
        buffer.putShort(index, (short) value);
    }
    
    /** * The bottom layer tries to clean ByteBuffer objects using Cleaner objects */
    @Override
    protected void deallocate(a) {
        ByteBuffer buffer = this.buffer;
        if (buffer == null) {
            return;
        }

        this.buffer = null;

        if (!doNotFree) {
            freeDirect(buffer);
        }
    }
    
    protected void freeDirect(ByteBuffer buffer) { PlatformDependent.freeDirectBuffer(buffer); }}Copy the code

UnpooledHeapByteBuf and UnpooledDirectByteBuf encapsulate a series of apis for byte[] and java.nio.bytebuffer. As we all know, java.nio.ByteBuffer’s main feature is that it is fast for low-level I/O operations, which reduces a memory copy. However, if it is used for in-memory calculations (such as byte lookups and other byte operations), the performance is not as good as using ByteBuf with byte[] at the bottom. Therefore, we need to select different types of ByteBuf instances according to the actual situation and pursue the ultimate performance.

The subclass implementation

As you can see from the figure above, there are many subclasses of ByteBuf, such as:

  • UnpooledUnsafeDirectByteBuf: Unpooled, uses the magic class Unsafe, and ByteBuf for out-of-heap memory. One of the key arguments is memoryAddress, which is reminiscent of the Unsafe class. Therefore, the allocation and release of that object is also related to Unsafe.
    • UnpooledUnsafeNoCleanerDirectByteBuf: no Cleaner collector. When the ByteBuf#release() method is called to freeMemory, the underlying layer does so through Unsafe#freeMemory(long address).
      • InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf: speak before, is mainly used for UnpooledByteBufAllocator allocator ByteBuf object.
    • ThreadLocalUnsafeDirectByteBuf: ByteBufUtil memory class. Improve ByteBuf allocation efficiency with lightweight object cache pools.
  • InstrumentedUnpooledDirectByteBuf: UnpooledByteBufAllocator Internal class that adds a memory capacity count to the logic that allocates memory.
  • ThreadLocalDirectByteBuf: ByteBufUtil inner class. useLightweight object cache poolImprove ByteBuf allocation efficiency.

UnpooledDirectByteBuf is a wrapper around the Java.nio.ByteBuffer object, but for Netty’s pursuit of extreme performance, how little is enough? So derived out the two different ideas of DirectByteBuf: UnpooledUnsafeDirectByteBuf and ThreadLocalDirectByteBuf. The former has a special variable, long Address, for Unsafe, which is used to read/write data. The latter is to use a lightweight object cache pool to improve the efficiency of ByteBuf allocation, as for the data read operation is still inherited from the parent class. UnpooledByteBuf UnpooledByteBuf UnpooledByteBuf You can tell from the name what this ByteBuf object has, mainly capturing:

  • Which data types/objects are used to store data. Byte [] is used for in-heap memory and java.nio.byteBuffer objects are used for out-of-heap memory.
  • There are two ways to reclaim out-of-heap memory, namely, Cleaner and Unsafe. (Of course, Cleaner also frees memory through Unsafe, but Cleaner is another way the JDK provides out-of-heap memory reclamation.)
  • Use a lightweight object cache pool to improve memory allocation efficiency. We’ll talk more about that later. In this way, allocation and reclamation actions are done by the buffer pool.

PooledByteBuf

Having analyzed a bunch of un-pooled BytebuFs, the implementation logic is not complicated. Pooled ByteBuf is much more complicated. First let’s look at PooledByteBuf. PooledByteBuf is an abstract class that provides a skeleton for subclasses to implement pooling, defining pooling-related properties and variables. In Netty 4.1.44 version before (including) using jemalloc3.x algorithm thought, and then using Jemalloc4. x algorithm thought for reconstruction. This article uses Netty 4.1.44 source code, the source code details for memory allocation is not detailed. If you are interested, please look at the big screen (TODO).

// io.netty.buffer.PooledByteBuf
/ * * * /
abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    
    // The allocator to which ByteBuf currently belongs
    private ByteBufAllocator allocator;

	// Object collector
    private final Handle<PooledByteBuf<T>> recyclerHandle;
	
    // T is generic and can be byte[] or ByteBuffer
    protected T memory;
    
    // PoolChunk to which the ByteBuf instance belongs
    protected PoolChunk<T> chunk;
    
    // Memory handle, 64 bits can be divided into two parts, each 32 bits, respectively representing different meanings.
    // Locate the memory location contained in the current ByteBuf object
    protected long handle;
    
    // Offset, starting at 0 (Byte), used for data container Byte []
    // First of all, we need to know that for non-HUGE memory, Netty requested 16MB of memory from the JVM.
    // This block of memory can be addressed in two ways. For "byte[]" this is the offset, while for direct memory "memoryAddress" + index is used.
    // Offset is the variable used for "byte[]". For example, when PooledHeapByteBuf is created, Netty will have a byte length of 16777216
    // Allocate an appropriate length of the array to the current ByteBuf. The offset is the offset with respect to the array subscript 0.
    // Write data to the array using offset+index to locate the memory region allocated by ByteBuf.
    // Simply put, the underlying byte[] (16MB) is shared by all, with the offset denoting the starting position.
    protected int offset;
    // Apply memory size
    protected int length;
    // Maximum length
    int maxLength;
    
    // Local cache
    PoolThreadCache cache;
    
    // Temporary ByteBuffer
    ByteBuffer tmpNioBuf;
    
    /** * this is a very important memory free code */
    @Override
    protected final void deallocate(a) {
        // #1 considering whether the handle variable is >0
        if (handle >= 0) {
            final long handle = this.handle;
            this.handle = -1;
            memory = null;
            // Release action via Arena (memory pool)
            chunk.arena.free(chunk, tmpNioBuf, handle, maxLength, cache);
            tmpNioBuf = null;
            chunk = null;
            // Reclaim the ByteBuf object (object pool)recycle(); }}private void recycle(a) {
        recyclerHandle.recycle(this);
    }
    
    // ...

}
Copy the code

Make a brief summary:

  • Netty recommends instantiating ByteBuf objects using allocators rather than constructors. Generally, each ByteBuf has its own ByteBufAllocator object.
  • Each ByteBuf holds a reference to the PoolChunk object, which holds an Arena reference that can be used to manage memory allocation. More on this later.
  • ByteBuf is a class representation of a segment of physical memory. We use the variable handle to indicate the location of the memory block.
  • Each ByteBuf holds a reference to the object recycle recyclerHandler. After the memory allocated by ByteBuf is recycled, the ByteBuf object is recycled and reused by recyclerHandler.
  • There are two representations of the underlying data store: byte[] and java.nio.byteBuffer objects. For convenience, generic (T) representations are used instead of separate classes for different underlying types.
  • The local thread pool PoolThreadCache is added to further improve the efficiency of memory allocation.

PooledByteBuf inheritance diagram

  • PooledHeapByteBuf: Poolable in-heap memory ByteBuf.
    • PooledUnsafeHeapByteBuf: Uses Unsafe to perform read/write operations.
  • PooledDirectByteBuf: Poolable out-of-heap memory ByteBuf.
  • PooledUnsafeDirectByteBuf: the use of pooling the Unsafe ByteBuf heap memory.

The difference between these bytebufs is the same as the name implies. The implementation of each PooleByteBuf subclass is not covered in detail, other than through Unsafe or java.nio.byteBuffer.

ByteBuf summary

  • We’re just going along one of the main ReferenceCounted->ByteBuf->AbstractByteBuf->AbstractReferenceCountedByteBuf->PooledByteBuf/UnpooledDirectByteBuf/Unpoole DHeapByteBuf explanation. Other types of ByteBuf you are interested in can look, the principle is not complicated.
  • ByteBuf is rightPhysical storage areaClass abstraction. It is classified according to two dimensions:Unpooled 和 Pooled,Heap 和 DirectByteBuffer. The two dimensions overlap and combine. Can be combined into:
    • UnpooledHeapByteBuf
    • UnpooledDirectByteBuf
    • PooledHeapByteBuf
    • PooledDirectByteBuf
    • There are also relatively important bytebuFs, such as CompositeByteBuf, that can improve programming efficiency.
  • Netty recommends that allocators be used to create ByteBuf objects. Therefore, ByteBuf objects named with the prefix Instrument are derived to track the life cycle of allocated ByteBuf objects and provide users with more details about memory allocation to help users better manage memory.
  • Reference counting is one of the key ways to reduce memory leaks.
  • Select an appropriate ByteBuf based on the actual situation and be familiar with the advantages and disadvantages of various ByetBuf instances:
    • Out of memory
      • advantages
        • Reduce a memory copy
        • Reduce GC pressure
        • Implement data sharing between processes and between JVM multiple instances
        • Suitable for large memory allocation scenarios
      • disadvantages
        • You need to release the memory manually. If you do not release the memory carelessly, it may leak out of the heap, and troubleshooting is difficult
    • Within the heap memory
      • advantages
        • Provides the ability to quickly allocate and free memory without pooling
        • Memory release is managed by the JVM. The user doesn’t have to worry about it
        • Suitable for scenarios with small memory allocation
      • disadvantages
        • For network I/O operations and file reads and writes, in-heap memory needs to be converted to out-of-heap memory and then interacts with the underlying device
    • pooling
      • advantages
        • Improve memory allocation speed and memory utilization
      • disadvantages
        • Before reference counting was mature, Netty assigned unpooled ByteBuf by default, but as various monitoring matures, Netty 4.1 assigns pooled ByteBuf by default
        • Managing memory requires some overhead
        • Memory leaks may occur
    • The pooling
      • advantages
        • Suitable for small memory allocation, fast allocation and fast release
      • disadvantages
        • No buffer

The above summary is just the tip of the iceberg for ByteBuf.

ByteBuf distribution

Allocate bytebug Locator on demand

ByteBufAllocator inheritance structure diagramIn fact, onlypoolingandThe poolingTwo kinds ofByteBufAllocatorAnd in order toPreferredIt’s just for convenience, yeahAbstractByteBufAllocatorSubclasses further encapsulate part of the API. So let’s focus onPoolBytesBufAllocator 和 UnpooledByteBufAllocatorTwo subclasses are implemented.

ByteBufAllocator API

AbstractByteBufAllocator

AbstractByteBufAllocator abstract class AbstractByteBufAllocator implements all interfaces of ByteBufAllocator and is the skeleton of ByteBufAllocator. As we know, Allocator is used to better manage the allocation of ByteBuf objects. It can determine whether the allocated memory capacity exceeds the limit, trace the allocated ByteBuf, and determine whether there are memory leaks. Therefore, AbstractByteBufAllocator has two methods inside that wrap ByteBuf and CompositeByteBuf objects, respectively, to detect memory leaks. Abstract classes don’t define many variables, but one important Boolean variable, directDefault, controls whether the object returned by the buffer() API is in or out of heap memory. The relevant source code is as follows:

public abstract class AbstractByteBufAllocator implements ByteBufAllocator {
    // Default initial capacity
    static final int DEFAULT_INITIAL_CAPACITY = 256;
    // Default maximum capacity
    static final int DEFAULT_MAX_CAPACITY = Integer.MAX_VALUE;
    ByteBuf by default
    static final int DEFAULT_MAX_COMPONENTS = 16;
    // If the capacity needs to be expanded very much, the new capacity needs to be calculated and increased by CALCULATE_THRESHOLD
    // Instead of being rude *2
    static final int CALCULATE_THRESHOLD = 1048576 * 4; // 4 MiB page

    static {
        ResourceLeakDetector.addExclusions(AbstractByteBufAllocator.class, "toLeakAwareBuffer");
    }
	
    /** * Trace the ByteBuf object to determine if a memory leak has occurred * For SIMPLE level, wrap ByteBuf with SimpleLeakAwareByteBuf * For ADVANCED, PARANOID level, Using AdvancedLeakAwareByteBuf to wrap ByteBuf * is a wrapper class that records data based on actions when calling bytebuf-related apis. For example, a Release () action will execute leak.Record (); Function, which understandably records the current usage of ByteBuf, * so it is possible to determine which ByteBuf objects have memory leaks */ by backtracking the records
    protected static ByteBuf toLeakAwareBuffer(ByteBuf buf) {
        ResourceLeakTracker<ByteBuf> leak;
        switch (ResourceLeakDetector.getLevel()) {
            case SIMPLE:
                leak = AbstractByteBuf.leakDetector.track(buf);
                if(leak ! =null) {
                    buf = new SimpleLeakAwareByteBuf(buf, leak);
                }
                break;
            case ADVANCED:
            case PARANOID:
                leak = AbstractByteBuf.leakDetector.track(buf);
                if(leak ! =null) {
                    buf = new AdvancedLeakAwareByteBuf(buf, leak);
                }
                break;
            default:
                break;
        }
        return buf;
    }
    private final boolean directByDefault;
    private final ByteBuf emptyBuf;
    
    protected AbstractByteBufAllocator(boolean preferDirect) {
        // Determined by whether preferDirect and platform support Unsafe
        directByDefault = preferDirect && PlatformDependent.hasUnsafe();
        emptyBuf = new EmptyByteBuf(this);
    }
    
    @Override
    public ByteBuf buffer(a) {
        if (directByDefault) {
            return directBuffer();
        }
        return heapBuffer();
    }
    
	// This is the abstract method that subclasses need to implement, returning ByteBuf for in-heap and out-of-heap memory
    protected abstract ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity);
    protected abstract ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity);
 
    // ...
}
Copy the code

PooledByteBufAllocator (4.1.44)

Initialization phase

poolingdistributorPooledByteBufAllocatorInitialization is divided into two phases, a static code block and a constructor. The default parameters for static block initialization are:

The PooledByteBufAllocator constructor is not complex. Its main functions are as follows:

  • Validation. Determine whether the set value is out of bounds and whether the status is abnormal.
  • Initialize thePoolThreadLocalCache.
  • Initialize theArena. Including DirectArena and HeapArena. Arena is an important concept of Jemalloc algorithm. A pooledbytefallocator object has multiple arenas, which aims to reduce resource competition and improve memory allocation efficiency in multi-threaded environment. For details on Arena, go to TODO.
  • Configure monitoring.

Interface implementation

PooledByteBufAllocatorTwo abstract methods are implemented that are closely related to creating ByteBuf objects:

Summary (Pooled by Bug locator)

The PooledByteBufAllocator doesn’t actually do much. There are mainly

  • Initialize the
    • Static parameter initialization(including verification). Such asDEFAULT_PAGE_SIZE(page size),DEFALUT_MAX_ORDER(tree height)And so on. Complete in static code block.
    • instantiationheapArenas 和 directArenasTwo arrays.ArenaIt has to do with memory allocation,AllocatorDelegate memory allocation to the correspondingArenasTo complete.
    • All kinds ofsize. Such assmallCacheSize,normalCacheSize,chunkSize 。
    • Distributor monitoring.List<PoolArenaMetric.
    • Local thread cachePoolThreadLocalCache.
  • Memory allocation is a delegateArenaObject complete.
  • There is an important inner classPoolThreadLocalCacheIt belongs to the local thread cache and is used to improve memory allocation efficiency.

UnpooledByteBufAllocator (4.1.44)

UnpooledByteBufAllocator is simpler than PooledByteBufAllocator and does not have complex memory management variables or logic. There are five internal classes that start with the Instrumented prefix. They have been analyzed before and will not be described here. UnpooledByteBufAllocator Specifies which ByteBuf to select based on:

  • Platform supportUnsafe.
  • With or withoutCleaner.

Let’s look directly at the source code:

// io.netty.buffer.UnpooledByteBufAllocator
public final class UnpooledByteBufAllocator 
    extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {

    private final UnpooledByteBufAllocatorMetric metric = new UnpooledByteBufAllocatorMetric();
    private final boolean disableLeakDetector;

    private final boolean noCleaner;
    
        public UnpooledByteBufAllocator(boolean preferDirect, boolean disableLeakDetector, boolean tryNoCleaner) {
        super(preferDirect);
        this.disableLeakDetector = disableLeakDetector;
        // Initialize the "noCleaner" variable
        noCleaner = tryNoCleaner && PlatformDependent.hasUnsafe()
                && PlatformDependent.hasDirectBufferNoCleanerConstructor();
    }

    /** * get a non-pooled instance of in-heap memory "ByteBuf" */
    @Override
    protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
        // Create different types of ByteBuf objects, depending on whether the platform supports Unsafe
        return PlatformDependent.hasUnsafe() ?
                new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
                new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
    }

    /** * get a non-pooled instance of out-of-heap memory "ByteBuf" */
    @Override
    protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
        final ByteBuf buf;
        // Not only does the platform support Unsafe, it also checks whether there are Cleaner platforms
        if (PlatformDependent.hasUnsafe()) {
            buf = noCleaner ? new InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf(this, initialCapacity, maxCapacity) :
                    new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
        } else {
            buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
        }
        returndisableLeakDetector ? buf : toLeakAwareBuffer(buf); }}Copy the code

As can be seen from the source code, there are two conditions that can be left or rightUnpooledByteBufAllocatorAllocation policy. , respectively,UnsafeandnoCleaner, and finally return different types of ByteBuf implementation classes.

Unpooled

Unpooled makes it easy to create a non-pooled instance of ByteBuf, which can be thought of as a utility class. An internal UnpooledByteBufAllocator object is used to allocate memory.

One of Netty’s zero-copy ideas is to create view-based implementations. I only need to manage the individual Pointers without having to copy the underlying data all over again, reducing the number of memory copies. So you can create a view by wrapping it. Note, however, that the wrapper object modifies the underlying data and is also visible to the source ByteBuf object. If you want to avoid this, copy it.

ByteBufUtil

ByteBufUtil provides static helper methods for manipulating ByteBuf.

methods describe
hexdump() Prints the contents of ByteBuf in a hexadecimal representation
equals(ByteBuf, ByteBuf) Used to determine the equality of two ByteBuf instances

Reference counting

The idea behind reference counting is not complicated, ** it is mainly concerned with tracking the number of active references to a particular object. ** An instance of the ReferenceCounted implementation will typically start with a reference count of 1 for the activity. As long as the reference count is greater than zero, it is guaranteed that the object will not be freed. When the number of active references drops to zero, the instance is released.

Note: The exact semantics of freeing may be implementation-specific, as memory may be reclaimed immediately or later, but it is at least clear that freed objects are not available in the language.

Reference counting is Paramount to pooling because it reduces the overhead of memory allocation.

Channel channel = ... ; ByteBufAllocator allocator = channel.alloc();// ...
ByteBuf buffer = allocator.directBuffer();
int count = buffer.refCnt();

// Release the object
boolean released = buffer.release();
Copy the code

conclusion

This is the end of Netty’s Architecture for ByteBuf, and there are a number of things that are missing, such as CompositeByteBuf, but the implementation principle is not that complicated. I hope you can have a systematic understanding of the whole ByteBuf through this article. This goes a long way toward developing a high-performance Netty. Netty introduces reference counting to optimize memory usage and performance. Read the source code to understand Netty’s pursuit of performance optimization. The idea of reference counting is not complicated, and Netty’s source code implementation is very efficient. What needs to be strengthened is an understanding of concurrency. Come back to this article when you have a better understanding of concurrency. It is now only known from the source code that the author wrote it, but it is not clear how it evolved, like reading a book that tells you a conclusion but does not give you the derivation.

My official account