Netty source object pool
An overview of the
Pooling is one of the things we do to improve performance, such as thread pooling, memory pooling, and object pooling next. The purpose of pooling is to reduce the overhead of creating and destroying objects, increase concurrency, and also reduce GC stress. Netty designed a simple object pool that is widely used in Netty memory. Such as for PooledUnsafeDirectByteBuf object pooling, such as io.net. Ty buffer. PoolThreadCache. MemoryRegionCache. Entry Object, which is used to encapsulate the memory block information in the local cache. Not all scenarios require object pooling, and it should be practical. It is also very simple to use, and sample code is shown below.
Object pool pattern is a design pattern. An object pool contains a set of objects that are already initialized and ready to use, and objects can be created and destroyed as needed. Users of the pool can take objects from the pool, manipulate them, and return them to the pool when they are no longer needed rather than destroying them directly. This is a special kind of factory object. If the cost of initialization and instantiation is high and frequent instantiation is required, the use of object pool can achieve significant performance improvement. The time it takes to fetch an object from the pool is predictable, but the time it takes to create a new instance is uncertain.
The sample code
class User {
private String name;
private Integer age;
private Recycler.Handle<User> handler;
public User(a) {}public User(Recycler.Handle<User> handler) {
this.handler = handler;
}
// SETTER & GETTER
public Recycler.Handle<User> getHandler(a) {
returnhandler; }}public class RecyclerTest {
private static final Recycler<User> userRecycler = new Recycler<User>() {
@Override
protected User newObject(Handle<User> handle) {
return newUser(handle); }};@Test
public void testRecycleDemo(a) {
// Get the User object from the object pool
User userFromCache1 = userRecycler.get();
System.out.println(userFromCache1);
// Get the User object from the object pool
User userFromCache2 = userRecycler.get();
System.out.println(userFromCache2);
Reclaim the userFromCache1 object
userFromCache1.getHandler().recycle(userFromCache1);
// Find that the existing acquired object is the same as the userFromCache1 objectUser userFromCache3 = userRecycler.get(); System.out.println(userFromCache3); }}// OUTPUT
// User@1963006a
// User@7fbe847c
// User@1963006a
Copy the codeThe code above shows how to quickly create a pool of objects. The core class is Recycler. Think of it as a utility class that you can use in your daily development. When we instantiate Recycler, we need to override the newObject(Handle
Handle) method, which associates objects with a Handle
. This is because the underlying Handle wraps the object to be reclaimed.
Reasoning to achieve
Before reading the source code, let’s try to design an object pool ourselves.
Single thread
The single-threaded model is the simplest, and we only need to construct a stack (or queue) for offloading and offloading operations to allocate and reclaim, respectively. If there are no objects in the stack, create a new object. To be safe, we need to set the stack size and the threshold. Check whether the number of reclaimed objects exceeds the threshold. If the number exceeds the threshold, discard the objects to avoid memory waste caused by too many pooled objects.
multithreading
With multithreading, the simplest solution is to lock related stack operations, but this will inevitably affect performance. Therefore, we can design to be lock-free with ThreadLocal. One of the things we haven’t considered yet is that we should call A thread_1-related method to collect (object A was created and collected by thread Thread_1, but in Thread_2 we call A thread_1-related method to collect). However, this method must be locked, which can also cause performance problems. Instead of pushing the reclaimed object, the local thread holds the reclaimed object and waits for Thread_1 to reclaim the reclaimed object of each thread. Moreover, Thread_1 has more freedom, such that when a certain threshold is exceeded, it can discard some of the recycled objects of other threads completely under its own control. Thread_2 does not mindlessly store other threads’ recycled objects. Some threads cache objects, while others discard them. Object A is allocated by Thread_1, but there are two threads, Thread_2 and Thread_3, that can recycle an object twice. Netty does not solve this problem. Instead, you throw an exception directly.
Recognize related classes
Before reading the source code, we need to make a note of the relevant classes. Look at the big screen:
- The object to be reclaimed is wrapped with the DefautlHandler class.
- Each recycled object type corresponds to a Stack object, a globally static Recycler and a class of WeakOrderQueue that serve this type of object.
- There is one inside Recycler
FastThreadLocal<Map<Stack, WeakOrderQueue>>This is related to heterogeneous thread collection, because a thread faces more than one Stack. - WeakOrderQueue is also related to the collection of different threads and has a linked list structure inside. More on that later.
DefaultHandle
In the lower right corner, it is Recycler’s packaging for Recycler, Recycler’s low-level operations are recyclable but not recyclable. It implements the Handle interface, which containsrecycle(Object obj)Recycling method. There are several important internal variables, explained as follows:
// io.netty.util.Recycler.DefaultHandle
private static final class DefaultHandle<T> implements Handle<T> {
// Recycle the RecycleId of this Handle last time
int lastRecycledId;
// Create a RecycleId for this Handle. Used together with lastRecycledId for repeated recycling detection
int recycleId;
// Whether the object has been reclaimed
boolean hasBeenRecycled;
// Create a Stack object for DefaultHandleStack<? > stack;// Object to be reclaimedObject value; DefaultHandle(Stack<? > stack) {this.stack = stack;
}
/** * Reclaim the object "value" held by this "Handle" * if the object is not equal, raise an exception */
@Override
public void recycle(Object object) {
if(object ! = value) {throw new IllegalArgumentException("object does not belong to handle"); } Stack<? > stack =this.stack;
if(lastRecycledId ! = recycleId || stack ==null) {
throw new IllegalStateException("recycled already");
}
// Push the recycle object onto the Stack, give it to the Stack, and leave the rest to the Stack
stack.push(this); }}Copy the codeDefaultHandler wraps the object to be reclaimed and also adds some information. RecycledId and lastRecycleId for recycling tests. LastRecycleId Is the thread ID used to store the last Recycler. RecycledId is used to store Recycler ID. This value is created by the Recycler global variable ID_GENERATOR. Every time we recycle a DefaultHandle object (that wraps our real recyclables inside), we get a unique ID from Recycler, and recycledId is equal to lastRecycledId at first. Then, by determining whether the two values are equal, we can determine whether the object is recycled and throw an exception.
Stack
This object is very important. It is the core class of the Netty object pool. It internally defines the object pool array structure and the core methods of fetch and reclaim (including local and foreign thread reclaim methods). First, recognize the relevant variables:
// io.netty.util.Recycler.Stack
private static final class Stack<T> {
// Stack is created by which Recycler
final Recycler<T> parent;
// We store the thread in a WeakReference,
// Otherwise we might be the only reference that still holds a strong reference to the thread itself after the thread dies, because DefaultHandle will hold a reference to the stack.
// The biggest problem is that if we don't use WeakReference, then if the user stores a reference to DefaultHandle somewhere,
// And the reference is never cleared (or not cleared in time), Thread may not be collected at all.
final WeakReference<Thread> threadRef;
// The number of cacheable object instances left by other threads corresponding to the Stack (i.e. the number of cacheable object instances left by other threads)
// Default value: 2048
final AtomicInteger availableSharedCapacity;
// How many Stack objects can be cached simultaneously by a thread. This "Stack" can be interpreted as another thread's "Stack."
// It's impossible to cache all of the Stack, so you need to limit it a bit
// Default value: 8
private final int maxDelayedQueues;
// The maximum size of the array. Default value: 4096
private final int maxCapacity;
// it can be interpreted as limiting the flow of the recycle action. Default value: 8
// Instead of limiting traffic until it is blocked, do so from the start
private final int interval;
// An array to store cached data. Default value: 256DefaultHandle<? >[] elements;// The number of non-empty elements in the array. Default value: 0
int size;
// Skip the number of reclaimed objects. Counting from 0, +1 for each recycled object skipped.
// Reset handleRecycleCount=0 when handleRecycleCount>interval
// Default value: 8. Initial value and "interval" so that the first element can be reclaimed
private int handleRecycleCount;
// Three nodes related to the collection of different threads
private WeakOrderQueue cursor, prev;
private volatile WeakOrderQueue head;
}
Copy the codeWe see the DefaultHandler[] array used to store the data, which is of type DefaultHandler, the wrapper class described above. Internally, a WeakOrderQueue is defined that is related to the collection of different threads, as well as some safety variables and information variables. Stack objects are retrieved from FastThreadLocal objects inside Recycle, so each thread has its own Stack object, creating a lock-free environment and connecting with other threads through weakOrderQueue. The related methods are described in the implementation logic.
WeakOrderQueue
WeakOrderQueue is used to store objects that are allocated by different threads to reclaim the same thread, which is the bridge we talked about earlier. For example, object A is allocated by thread Thread_1, but in Thread_2, according to the principle of “who allocates who retrieves”, Thread_2 cannot reclaim the object, so Thread_2 will put the object in the corresponding WeakOrderQueue linked list. This list is created by Thread_2, so how can it be associated with Thread_1? So there’s a trick here, and another thread will maintain oneMap<Stack<? > WeakOrderQueue>The local thread cache, Thread_2 gets the corresponding WeakOrderQueue based on the Stack (because DefaultHandler is internally wrapped to reclaim the object, and that DefaultHandler also holds the Stack reference that created it). If not, Then create and update the Head node of the Stack (lock). This establishes the association between Thread_1 and Thread_2 about object A, and Thread_1 can then reclaim the object from WeakOrderQueue. Related variables are explained as follows:
/** * The internal list of "WorkOrderQueue" consists of "Head" and "tail" nodes. * Team data is not immediately visible to other threads, using the idea of final consistency. * There is no need to guarantee immediate visibility, just eventual visibility */
// io.netty.util.Recycler.WeakOrderQueue
private static final class WeakOrderQueue extends WeakReference<Thread> {
// Head manages the creation of the Link object. The internal next points to the next "Link" node, forming a linked list structure
private final Head head;
// Data storage node
private Link tail;
// A list of workOrderQueues that point to other threads
private WeakOrderQueue next;
/ / the only ID
private final int id = ID_GENERATOR.getAndIncrement();
// it can be interpreted as limiting the flow of the recycle action. Default value: 8
// Instead of limiting traffic until it is blocked, do so from the start
private final int interval;
// Number of discarded reclaimed objects
private int handleRecycleCount;
// The LINK node inherits AtomicInteger and has a readIndex pointer inside
static final class Link extends AtomicInteger {
finalDefaultHandle<? >[] elements =new DefaultHandle[LINK_CAPACITY];
intreadIndex; Link next; }}Copy the codeWeakOrderQueue inherits WeakReference. When the owning thread is reclaimed, the corresponding WeakOrderQueue will also be reclaimed. Internally, Link objects constitute a linked list structure, and Link maintains an array of DefaultHandle[] to temporarily store objects reclaimed by different threads. If it does not, it succeeds. Otherwise, create a new Link node and add a reclaim object. Then update the list structure to make the tail pointer point to the newly created Link object. Since there is more than one thread, there will be multiple corresponding WeakOrderQueues, and a linked list structure is formed between WeakOrderQueues. The interval variable is used to recycle **” stream limiting “**, which limits the recycle rate from the beginning by collecting one object for every eight and discarking the rest.
Recycler
Recycler is our users’ direct contact, it’s our portal to build a simple pool of recyclers. The main function of this class is to read configuration information and initialize it, and it builds two very important FastThreadLocal objects, respectivelyFastThreadLocal<Stack<T>>As well asFastThreadLocal<Map<Stack<? >, WeakOrderQueue>>. Related variables are explained as follows:
// io.netty.util.Recycler
public abstract class Recycler<T> {
// Globally unique ID can be used in two places, one is to initialize each Recycler OWN_THREAD_ID
// The other is the initialization ID for each WeakOrderQueue
private static final AtomicInteger ID_GENERATOR = new AtomicInteger(Integer.MIN_VALUE);
// Global ID generator
private static final int OWN_THREAD_ID = ID_GENERATOR.getAndIncrement();
// Maximum cacheable object capacity per thread. Default value :4096
private static final int DEFAULT_INITIAL_MAX_CAPACITY_PER_THREAD = 4 * 1024;
// Maximum cacheable object size per thread. Default :4096
private static final int DEFAULT_MAX_CAPACITY_PER_THREAD;
// Initial capacity. Default value :256
private static final int INITIAL_CAPACITY;
// Maximum shared capacity factor. Default value: 2
private static final int MAX_SHARED_CAPACITY_FACTOR;
// Maximum queue delay per thread. Default: 8
private static final int MAX_DELAYED_QUEUES_PER_THREAD;
// Link is used to store objects collected by different threads. There is an array inside. The array length is =LINK_CAPACITY
// Default value: 16
private static final int LINK_CAPACITY;
// Ratio of objects discarded by different threads. Default value :8
// In a different thread, only one object out of eight is reclaimed and the rest are discarded
private static final int RATIO;
// The maximum length of the array in the State
data structure held by each thread
private final int maxCapacityPerThread;
// Share the capacity factor. The larger the value, the smaller the total number of objects to be reclaimed outside the thread.
/ / stay outside of this thread, because they are recycled object = maxCapacityPerThread/maxSharedCapacityFactor total
private final int maxSharedCapacityFactor;
// For objects that have never been reclaimed, Netty chooses to discard them at a certain percentage (when) to avoid the excessive growth of objects in the pool, which will affect the main thread service function.
// Default value: 8
private final int interval;
// Maximum number of instance objects can be cached per thread object pool
private final int maxDelayedQueuesPerThread;
// Each thread has its own Stack.
private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {
@Override
protected Stack<T> initialValue(a) {
return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,
interval, maxDelayedQueuesPerThread);
}
// Remove the callback method
@Override
protected void onRemoval(Stack<T> value) {
// Let us remove the WeakOrderQueue from the WeakHashMap directly if its safe to remove some overhead
if (value.threadRef.get() == Thread.currentThread()) {
if (DELAYED_RECYCLED.isSet()) {
// Safely remove "WeakOrderQueue"DELAYED_RECYCLED.get().remove(value); }}}};WeakOrderQueue () {WeakOrderQueue (); WeakOrderQueue () {WeakOrderQueue ();
private static finalFastThreadLocal<Map<Stack<? >, WeakOrderQueue>> DELAYED_RECYCLED =newFastThreadLocal<Map<Stack<? >, WeakOrderQueue>>() {// Initialize a "WeakHashMap" object for each thread to ensure that the object can be reclaimed without strong references
/ / key = > Stack,
/ / value = > WeakOrderQueue,
@Override
protectedMap<Stack<? >, WeakOrderQueue> initialValue() {return new WeakHashMap<Stack<?>, WeakOrderQueue>();
}
};
}
Copy the codeBefore talking about the source code, we first review the reasoning to achieve multithreading that section, through the source code to feel Netty on the object pool lock free design ideas.
Gets objects from the pool
Recycler#get
This method is Netty’s entry point to the lightweight object pool. The main logic is to get the thread-private Stack object and use it to get DefaultHandle objects from the object pool. If the object fetched from Stack is empty, it passesnewObject(Default)Method creates a new target object.newObject(Default)This method needs to be implemented by the user.
/** * Get the pooled object from the object pool */
// io.netty.util.Recycler#get
public final T get(a) {
if (maxCapacityPerThread == 0) {
return newObject((Handle<T>) NOOP_HANDLE);
}
// #1 get the current thread cache "Stack" object,
Stack<T> stack = threadLocal.get();
// #2 Pops a DefaultHandle object from the Stack
DefaultHandle<T> handle = stack.pop();
// #3 If the popup object is empty, there is no cached object inside.
// Create a new Handle and a new Object().
// newObject(Handler): This method is abstract and requires a user - defined implementation
if (handle == null) {
handle = stack.newHandle();
// Handle and the object hold references to each other
// More importantly, the Handle holds a Stack reference, so it can be pushed directly onto the Stack when it is recycled
handle.value = newObject(handle);
}
/ / # 5 to return
return (T) handle.value;
}
Copy the codeStack#pop
Stack#pop()Or more concise, first will judgesizeIs greater than 0 (greater than 0 indicates the existence of cache objects), if greater than 0 fromDefaultHandle[]The value of size is also the subscript of the array. If equal to 0, it enters the world of collecting foreign threads:Other threads are collectively called different threads) * *WeakOrderQueueRetrieves objects to be reclaimed from the queue. This method also includes a simple recycle check logic that throws an exception if the created thread and the recycle thread are not equal.
// io.netty.util.Recycler.Stack#pop
/** * pop a "DefaultHandler" object from the stack * ① if size>0, pop elements in the array * ② if size=0, try to steal some cache objects from other threads */
// io.netty.util.Recycler.Stack#pop
DefaultHandle<T> pop(a) {
int size = this.size;
if (size == 0) {
// The number of objects in the current cache is 0
// Try to get a recyclable object from a different-thread related "WeakOrderQueue" queue
if(! scavenge()) {// The thread has no recyclable object and returns null
return null;
}
size = this.size;
if (size <= 0) {
// double check, avoid races
return null; }}// There are cache objects in the stack
size --;
// Use array stack internally
DefaultHandle ret = elements[size];
elements[size] = null;
/ / update the size
this.size = size;
// Recycle check. If the recycle ids are different, they are different
if(ret.lastRecycledId ! = ret.recycleId) {// If the last object reclaim ID is different from this one, throw an exception
throw new IllegalStateException("recycled multiple times");
}
// Reset the recycling ID of DefaultHandler
ret.recycleId = 0;
ret.lastRecycledId = 0;
return ret;
}
Copy the codeStack#scavenge
Stack# esome () is called to attempt to reclaim some objects from the WeakOrderQueue linked list related to different threads. If the reclaim is successful, return true. If the reclaim fails, reset the curson pointer and return to the head node again.
/** * scavenge. This method is used to reclaim some objects from the WeakOrderQueue */
// io.netty.util.Recycler.Stack#scavenge
private boolean scavenge(a) {
// Try to retrieve some of the objects to be reclaimed from "WeakOrderQueue" and convert them to the "Stack#DefaultHandler[]" array
if (scavengeSome()) {
// The transfer succeeded
return true;
}
// Failed to get prev, cursor pointer reset
prev = null;
cursor = head;
return false;
}
Copy the codeStack#scavengeSome
Stack#scavengeSome()The WeakOrderQueue () method attempts to reclaim objects from a different thread. Note that this method is executed in a Stack object, and Stack has a WeakOrderQueue linked list that is related to a different thread, so we can find reclaimed objects by traversing the list. Another important point is that if the thread bound to WeakOrderQueue has died, it means that the corresponding WeakOrderQueue will not add new recycle objects, so we need to clean the WeakOrderQueue accordingly to ensure that it can be GC to avoid memory leakage. Corresponding clearing methodcursor.reclaimAllSpaceAndUnlink().scavengeSome()Return true as long as the number of successes is greater than 1, otherwise continue to iterate over other weakOrderQueues of different threads, directly iterate over all WeakOrderQueues or find available reclaim objects.
// io.netty.util.Recycler.Stack#scavengeSome
private boolean scavengeSome(a) {
WeakOrderQueue prev;
WeakOrderQueue cursor = this.cursor;
// The current cursor pointer is empty
if (cursor == null) {
prev = null;
// point to the header
cursor = head;
if (cursor == null) {
// If the header is null, it indicates that no foreign thread has been saved. Return false
return false; }}else {
prev = this.prev;
}
boolean success = false;
// There is a different thread "WeakOrderQueue", ready to reclaim the object
do {
// Try to move as many reclaimed objects as possible from a different thread "WeakOrderQueue". But the pointer is not updated
// Return true when the number of transferred objects >0
if (cursor.transfer(this)) {
// The transition succeeds, and the loop is broken
success = true;
break;
}
// Save the next reference to "WeakOrderQueue", because this time "WeakOrderQueue" may be reclaimed
WeakOrderQueue next = cursor.getNext();
if (cursor.get() == null) {
// If the thread associated with the queue terminates (cursor.get()==null),
// Determine if there are any recyclable objects, and disconnect them after they have been transferred
// We never disconnect the first connection because we don't want to synchronize when updating the header
if (cursor.hasFinalData()) {
// Move objects proportionally
for (;;) {
if (cursor.transfer(this)) {
success = true;
} else {
break; }}}if(prev ! =null) {
// Make sure to reclaim all Spaces before removing the WeakOrderQueue for GC.
cursor.reclaimAllSpaceAndUnlink();
// Update node information, delete cursor node,prev.setNext(next); }}else {
prev = cursor;
}
// Update cursor pointer to the next object under "WeakOrderQueue"
cursor = next;
} while(cursor ! =null && !success);
this.prev = prev;
this.cursor = cursor;
return success;
}
Copy the codeWeakOrderQueue#transfer
WeakOrderQueue#transfer()Move some objects from the other-thread WeakOrderQueue queue to the Stack array. WeakOrderQueue is a linked list of Link objects inside, while the underlying Link list uses an array of length 16 to store reclaimed objects and maintains a read pointerreadIndex, records the position of the read operation.transfer()Method will process only one Link node (not all objects to be reclaimed can be successfully written into the Stack, and will be discarded in a certain proportion). When the expected capacity exceeds the Stack array length, it needs to be expanded. If the read pointer reaches the end of the Link node, determine whether the link. next object is empty. If not, update the head pointer to reclaim the current Link object.
/** * Move the "WeakOrderQueue" memory cache to the "Stack" array * Remember, this method is called by the thread that is going to reclaim the object from another thread, so no lock is required * move one Link node at a time (length: 16) */
// io.netty.util.Recycler.WeakOrderQueue#transfer
@SuppressWarnings("rawtypes")
boolean transfer(Stack
dst) {
// Get the header of the list
Link head = this.head.link;
if (head == null) {
// If the header is null, no data is available
return false;
}
// LINK_CAPACITY Capacity of a Link node. Default value: 16
// Check whether the read pointer "readIndex" reaches the boundary, if it reaches the boundary and points to the next node is NULL
if (head.readIndex == LINK_CAPACITY) {
if (head.next == null) {
return false;
}
head = head.next;
// Update the header to point to next
this.head.relink(head);
}
// The "Head" object has two pointer variables: readIndex and AtomicInteger
// srcStart: indicates the readable start point
final int srcStart = head.readIndex;
// head: indicates the number of additions since the creation
int srcEnd = head.get();
// The current number of recyclable objects
final int srcSize = srcEnd - srcStart;
if (srcSize == 0) {
return false;
}
final int dstSize = dst.size;
// Calculate the expected value
final int expectedCapacity = dstSize + srcSize;
// The Stack array needs to be expanded
if (expectedCapacity > dst.elements.length) {
// Expand the Stack array. The actual capacity expansion rule is calculated according to the Stack rule
// It may not be the same as the expected capacity
final int actualCapacity = dst.increaseCapacity(expectedCapacity);
// According to the actual capacity of "Stack" to determine the final reclamation end index
srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd);
}
if(srcStart ! = srcEnd) {// Loop over and copy
final DefaultHandle[] srcElems = head.elements;
final DefaultHandle[] dstElems = dst.elements;
int newDstSize = dstSize;
for (inti = srcStart; i < srcEnd; i++) { DefaultHandle<? > element = srcElems[i];if (element.recycleId == 0) {
element.recycleId = element.lastRecycledId;
} else if(element.recycleId ! = element.lastRecycledId) {// If the present value is not equal, it means that the creator thread and the collector thread are not the same thread
// Throw an exception
throw new IllegalStateException("recycled already");
}
// Easy for GC to recycle object Link object
srcElems[i] = null;
// This collection requires discarding objects in proportion to a certain percentage, rather than receiving them all
if (dst.dropHandle(element)) {
/ / throw away
continue;
}
// Copy recycled objects into the Stack target array without discarding them
element.stack = dst;
dstElems[newDstSize ++] = element;
}
if(srcEnd == LINK_CAPACITY && head.next ! =null) {
// The current Link object has already been recycled, and the next node it points to is not NULL
// Update the head to point to the next Link node, the current Link needs to be GC
this.head.relink(head.next);
}
// Update the read node pointer
head.readIndex = srcEnd;
// Again determine whether the target array writes new data
if (dst.size == newDstSize) {
Return false if nothing is retrieved
return false;
}
// Update the value of the used capacity of the target Stack
dst.size = newDstSize;
return true;
} else {
// The destination stack is full already.
return false; }}Copy the codeTo obtain objects from the object pool, first judge whether there are objects in the local thread Stack, if there are, it will directly return, if not, you need to try to recover objects from the WeakOrderQueue linked list of different threads.
Another thread holds objects temporarily
Earlier we talked about the main thread allocating memory, including retrieving some objects from the WeakOrderQueue linked list of different threads. Now let’s look at how different threads temporarily store objects to be reclaimed that do not belong to their own threads in the WeakOrderQueue linked list. Object A is created by Thread_1, but reclaimed by Thread_2. In this case, Thread_2 temporarily stores object A in the WeakOrderQueue list of the local thread. WeakOrderQueue forms a linked list with Thread_1’s Stack#head node, which Thread_1 walks through to reclaim objects.
DefaultHandle#recycle
DefaultHandle#recycle()Does some authentication, such as target objects, reclaim threads, etc. And then callStack#push()Method into the memory pool.
// io.netty.util.Recycler.DefaultHandle#recycle
/** * Reclaim an object that holds a reference to "DefaultHandle" internally, * and "DefaultHandle" only retrieves itself */
@Override
public void recycle(Object object) {
if(object ! = value) {throw an exception if the target recycle object is not itselfthrow new IllegalArgumentException("object does not belong to handle"); } Stack<? > stack =this.stack;
if(lastRecycledId ! = recycleId || stack ==null) {
// If the collector thread and the allocation thread are not the same, raise an exception
throw new IllegalStateException("recycled already");
}
// Press "Stack"
stack.push(this);
}
Copy the codeStack#push
Select different reclamation policies based on local threads or different threads. Local thread can directly recycle, but different thread recycling needs to write WeakOrderQueue linked list.
void push(DefaultHandle
item) {
Thread currentThread = Thread.currentThread();
if (threadRef.get() == currentThread) {
// This thread is recycled, directly written to the memory pool
pushNow(item);
} else {
// Different threads recycle objects, delaying collectionpushLater(item, currentThread); }}Copy the codeStack#pushLater
Stack#pushLater()The main purpose of the method is to obtain thread-private WeakOrderQueue objects from the local thread cache and write the reclaim queue to the queue.
/** * Writes the reclaimed object to the "WeakOrderQueue" linked list associated with the different thread. * There are two thresholds in the process of creating a "WeakOrderQueue" that will cause the creation to fail: WeakOrderQueue number exceeded maxDelayedQueues limit, failed to create * ② Failed to apply capacity of length 16 to target object "Stack" * Put newly created object into Map array for next use. * and write item to "WeakOrderQueue" */
// io.netty.util.Recycler.Stack#pushLater
private void pushLater(DefaultHandle
item, Thread thread) {
if (maxDelayedQueues == 0) {
// Does not support the collection of different threads, directly discarded
return;
}
// Use "FastThreadLocal" to cache the relationship between Stack and WeakOrderQueue
// Because a thread can correspond to multiple stacks (i.e. multiple object pools of different types)Map<Stack<? >, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get(); WeakOrderQueue queue = delayedRecycled.get(this);
if (queue == null) {
// The current thread does not have a "WeakOrderQueue" linked list associated with it and needs to create a new one
// Check to see if the quota is full before creating it
if (delayedRecycled.size() >= maxDelayedQueues) {
// Add a placeholder object when the value is full
delayedRecycled.put(this, WeakOrderQueue.DUMMY);
return;
}
AvailableSharedCapacity
// This will result in NULL being returned.
// If the object is not NULl, it is successfully created, and update the head pointer of "Stack" to point to a new "WeakOrderQueue".
if ((queue = newWeakOrderQueue(thread)) == null) {
// Discard it directly
return;
}
delayedRecycled.put(this, queue);
} else if (queue == WeakOrderQueue.DUMMY) {
// drop object
return;
}
// Add the reclaim object to the WeakOrderQueue queue
queue.add(item);
}
Copy the codeStack#newWeakOrderQueue()
Create a brand new oneWeakOrderQueueAnd register with the correspondingStack#headIn the header.
// io.netty.util.Recycler.Stack#newWeakOrderQueue
private WeakOrderQueue newWeakOrderQueue(Thread thread) {
return WeakOrderQueue.newQueue(this, thread);
}
// io.netty.util.Recycler.WeakOrderQueue#newQueue
static WeakOrderQueue newQueue(Stack
stack, Thread thread) {
// We allocated a Link so reserve the space
if(! Head.reserveSpaceForLink(stack.availableSharedCapacity)) {return null;
}
final WeakOrderQueue queue = new WeakOrderQueue(stack, thread);
// Done outside of the constructor to ensure WeakOrderQueue.this does not escape the constructor and so
// may be accessed while its still constructed.
stack.setHead(queue);
return queue;
}
/ / add lock
// io.netty.util.Recycler.Stack#setHead
synchronized void setHead(WeakOrderQueue queue) {
queue.setNext(head);
head = queue;
}
Copy the codeWeakOrderQueue#add
Write the reclaimed object to the WeakOrderQueue queue. WeakOrderQueue is a private variable of the local thread cache. Thread writes to its WeakOrderQueue without locking. WriteIndex and readIndex of Link node also do not have concurrency problems, but there are problems of data visibility. This visibility is visible to the reclaim thread, because the reclaim thread needs to determine the termination sequence with writeIndexvolatileDeferred update write index usage. WeakOrderQueue (writeIndex) allows you to retrieve data from the WeakOrderQueue (WeakOrderQueue) of other threads without being stuck on the same tree. You’ll see it the next time you come in, which improves write performance.
/** * Adding data to the list is actually adding data to the "tail" node. * If the write pointer to the node to which "tail" points (tail.get() and AtomicInteger holds the write pointer) exceeds LINK_CAPACITY *, a new Link object */ needs to be created
// io.netty.util.Recycler.WeakOrderQueue#add
void add(DefaultHandle
handle) {
handle.lastRecycledId = id;
// Only one object in every eight is reclaimed
if (handleRecycleCount < interval) {
handleRecycleCount++;
return;
}
handleRecycleCount = 0;
Link tail = this.tail;
int writeIndex;
// When the write pointer reaches the end of the Link node, a new "Link" node needs to be applied
if ((writeIndex = tail.get()) == LINK_CAPACITY) {
The availableSharedCapacity may be exceeded and the creation may fail
Link link = head.newLink();
if (link == null) {
// Failed to create, and was discarded
return;
}
// Update node information.
this.tail = tail = tail.next = link;
// Get the write node
writeIndex = tail.get();
}
/ / update the value
tail.elements[writeIndex] = handle;
// The handle of "DefaultHandle" is left empty because it is possible that the "Stack" will be reclaimed. If there is a strong reference,
// The Stack object cannot be reclaimed. The value will be reset later when the Stack object is reclaimed
handle.stack = null;
// Delay index update (the index is updated until the queue element is visible)
// This is an advanced use of "volatile". Since there are multiple producers and a consumer, there is no need to update the "WeakOrderQueue" link
// Immediately visible, it can retrieve reclaimed objects from other "WeakOrderQueue" queues. Use lazySet to maintain final visibility, but there is a delay,
// This improves write time performance (essentially reducing memory barrier overhead)
tail.lazySet(writeIndex + 1);
}
Copy the codeconclusion
There are many things we can learn from Netty’s Recycle lightweight object pool, the most important of which is the lock-free design. The entire Recycle class consists of a single synchronized keyword; the rest is implemented through CAS or atomic variables. The core of lock-free design is the FastThreadLocal class, which is designed by Netty itself to adapt to the Netty environment of high performance local thread cache variables, using the design strategy of space for time. Another good design idea is the heterothread object collection strategy: The foreign thread does not actively push the object to be reclaimed to the target thread. Instead, the WeakOrderQueue caches the object to be reclaimed, so that locking is not required. Then the write pointer is set to AtomicInteger visible to the target thread, but Netty does not meet the requirement. It also uses the advanced use of AtomicInteger AtomicInteger#lazySet to delay updating the write pointer because, depending on the actual situation, the target thread does not only reclaim the WeakOrderQueue of the current foreign thread, but other foreign threads can also be reclaimed. Therefore, for the target thread, the write pointer can not be displayed in time, which can improve the writing efficiency of the foreign thread.