Netty 的内存池源码

_/* _ * Copyright 2013 The Netty Project * * The Netty Project licenses this file to you under the Apache License, * version 2.0 (the "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at: * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * License for the specific language governing permissions and limitations * under the License. */ package io.netty.util;

import io.netty.util.concurrent.FastThreadLocal; import io.netty.util.internal.SystemPropertyUtil; import io.netty.util.internal.logging.InternalLogger; import io.netty.util.internal.logging.InternalLoggerFactory;

import java.lang.ref.WeakReference; import java.util.Arrays; import java.util.Map; import java.util.WeakHashMap; import java.util.concurrent.atomic.AtomicInteger;

import static io.netty.util.internal.MathUtil.safeFindNextPositivePowerOfTwo; import static java.lang.Math.max; import static java.lang.Math.min;

_/** _ * Light-weight object pool based on a thread-local stack. * * @param <T> _the type of the pooled object _ */ public abstract class Recycler<T> {

1**private static final** InternalLogger _logger_ \= InternalLoggerFactory._getInstance_(Recycler.**class**);
2
3@SuppressWarnings(**"rawtypes"**)
4**private static final** Handle _NOOP\_HANDLE_ \= **new** Handle() {
5    @Override

public void recycle(Object object) { _// NOOP _ } }; private static final AtomicInteger ID_GENERATOR = new AtomicInteger(Integer.MIN_VALUE); private static final int OWN_THREAD_ID = ID_GENERATOR.getAndIncrement(); private static final int DEFAULT_INITIAL_MAX_CAPACITY_PER_THREAD = 4 * 1024; _// Use 4k instances as default. _ private static final int DEFAULT_MAX_CAPACITY_PER_THREAD; private static final int INITIAL_CAPACITY; private static final int MAX_SHARED_CAPACITY_FACTOR; private static final int MAX_DELAYED_QUEUES_PER_THREAD; private static final int LINK_CAPACITY; private static final int RATIO;

1**static** {
2    _// In the future, we might have different maxCapacity for different object types.

_ // e.g. io.netty.recycler.maxCapacity.writeTask // io.netty.recycler.maxCapacity.outboundBuffer int maxCapacityPerThread = SystemPropertyUtil.getInt("io.netty.recycler.maxCapacityPerThread", SystemPropertyUtil.getInt("io.netty.recycler.maxCapacity", DEFAULT_INITIAL_MAX_CAPACITY_PER_THREAD)); if (maxCapacityPerThread < 0) { maxCapacityPerThread = DEFAULT_INITIAL_MAX_CAPACITY_PER_THREAD; }

1    _DEFAULT\_MAX\_CAPACITY\_PER\_THREAD_ \= maxCapacityPerThread;
2
3    _MAX\_SHARED\_CAPACITY\_FACTOR_ \= _max_(2,
4            SystemPropertyUtil._getInt_(**"io.netty.recycler.maxSharedCapacityFactor"**,
5                    2));
6
7    _MAX\_DELAYED\_QUEUES\_PER\_THREAD_ \= _max_(0,
8            SystemPropertyUtil._getInt_(**"io.netty.recycler.maxDelayedQueuesPerThread"**,
9                    _// We use the same value as default EventLoop number

_ NettyRuntime.availableProcessors() * 2));

1    _LINK\_CAPACITY_ \= _safeFindNextPositivePowerOfTwo_(
2            _max_(SystemPropertyUtil._getInt_(**"io.netty.recycler.linkCapacity"**, 16), 16));
3
4    _// By default we allow one push to a Recycler for each 8th try on handles that were never recycled before.

_ // This should help to slowly increase the capacity of the recycler while not be too sensitive to allocation // bursts. RATIO = safeFindNextPositivePowerOfTwo(SystemPropertyUtil.getInt("io.netty.recycler.ratio", 8));

1    **if** (_logger_.isDebugEnabled()) {
2        **if** (_DEFAULT\_MAX\_CAPACITY\_PER\_THREAD_ \== 0) {
3            _logger_.debug(**"-Dio.netty.recycler.maxCapacityPerThread: disabled"**);
4            _logger_.debug(**"-Dio.netty.recycler.maxSharedCapacityFactor: disabled"**);
5            _logger_.debug(**"-Dio.netty.recycler.linkCapacity: disabled"**);
6            _logger_.debug(**"-Dio.netty.recycler.ratio: disabled"**);
7        } **else** {
8            _logger_.debug(**"-Dio.netty.recycler.maxCapacityPerThread: {}"**, _DEFAULT\_MAX\_CAPACITY\_PER\_THREAD_);
9            _logger_.debug(**"-Dio.netty.recycler.maxSharedCapacityFactor: {}"**, _MAX\_SHARED\_CAPACITY\_FACTOR_);
10            _logger_.debug(**"-Dio.netty.recycler.linkCapacity: {}"**, _LINK\_CAPACITY_);
11            _logger_.debug(**"-Dio.netty.recycler.ratio: {}"**, _RATIO_);
12        }
13    }
14
15    _INITIAL\_CAPACITY_ \= _min_(_DEFAULT\_MAX\_CAPACITY\_PER\_THREAD_, 256);
16}
17
18**private final int** **maxCapacityPerThread**;
19**private final int** **maxSharedCapacityFactor**;
20**private final int** **ratioMask**;
21**private final int** **maxDelayedQueuesPerThread**;
22
23**private final** FastThreadLocal<Stack<T\>> **threadLocal** \= **new** FastThreadLocal<Stack<T\>>() {
24    @Override

protected Stack<T> initialValue() { return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor, ratioMask, maxDelayedQueuesPerThread); }

    @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()) { DELAYED_RECYCLED.get().remove(value); } } } };

1**protected** Recycler() {
2    **this**(_DEFAULT\_MAX\_CAPACITY\_PER\_THREAD_);
3}
4
5**protected** Recycler(**int** maxCapacityPerThread) {
6    **this**(maxCapacityPerThread, _MAX\_SHARED\_CAPACITY\_FACTOR_);
7}
8
9**protected** Recycler(**int** maxCapacityPerThread, **int** maxSharedCapacityFactor) {
10    **this**(maxCapacityPerThread, maxSharedCapacityFactor, _RATIO_, _MAX\_DELAYED\_QUEUES\_PER\_THREAD_);
11}
12
13**protected** Recycler(**int** maxCapacityPerThread, **int** maxSharedCapacityFactor,
14                   **int** ratio, **int** maxDelayedQueuesPerThread) {
15    **ratioMask** \= _safeFindNextPositivePowerOfTwo_(ratio) - 1;
16    **if** (maxCapacityPerThread <= 0) {
17        **this**.**maxCapacityPerThread** \= 0;
18        **this**.**maxSharedCapacityFactor** \= 1;
19        **this**.**maxDelayedQueuesPerThread** \= 0;
20    } **else** {
21        **this**.**maxCapacityPerThread** \= maxCapacityPerThread;
22        **this**.**maxSharedCapacityFactor** \= _max_(1, maxSharedCapacityFactor);
23        **this**.**maxDelayedQueuesPerThread** \= _max_(0, maxDelayedQueuesPerThread);
24    }
25}
26
27@SuppressWarnings(**"unchecked"**)
28
29**public final** T get() {
30    **if** (**maxCapacityPerThread** \== 0) {
31        _//__表明没有开启池化

_ return newObject((Handle<T>) NOOP_HANDLE); } Stack<T> stack = threadLocal.get(); DefaultHandle<T> handle = stack.pop(); _//试图从“池”__中取出一个，没有就新建一个 _ if (handle == null) { handle = stack.newHandle(); handle.value = newObject(handle); } return (T) handle.value; }

_/\*\*

_ * @deprecated use {__@link _Handle#recycle(Object)}. _ */ @Deprecated public final boolean recycle(T o, Handle<T> handle) { if (handle == NOOP_HANDLE) { return false; }

1    DefaultHandle<T\> h = (DefaultHandle<T\>) handle;
2    **if** (h.**stack**.**parent** != **this**) {
3        **return false**;
4    }
5
6    h.recycle(o);
7    **return true**;
8}
9
10**final int** threadLocalCapacity() {
11    **return** **threadLocal**.get().**elements**.**length**;
12}
13
14**final int** threadLocalSize() {
15    **return** **threadLocal**.get().**size**;
16}
17
18**protected abstract** T newObject(Handle<T\> handle);
19
20**public interface** Handle<T\> {
21    **void** recycle(T object);
22}
23
24**static final class** DefaultHandle<T\> **implements** Handle<T\> {
25    **private int** **lastRecycledId**;
26    **private int** **recycleId**;
27
28    **boolean** **hasBeenRecycled**;
29
30    **private** Stack<?> **stack**;
31    **private** Object **value**;
32
33    DefaultHandle(Stack<?> stack) {
34        **this**.**stack** \= stack;
35    }
36
37    @Override

public void recycle(Object object) { if (object != value) { throw new IllegalArgumentException("object does not belong to handle"); }

1        Stack<?> stack = **this**.**stack**;
2        **if** (**lastRecycledId** != **recycleId** || stack == **null**) {
3            **throw new** IllegalStateException(**"recycled already"**);
4        }
5        _//__释放用完的对象到池里面去

_ stack.push(this); } }

1**private static final** FastThreadLocal<Map<Stack<?>, WeakOrderQueue>> _DELAYED\_RECYCLED_ \=
2        **new** FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>() {
3    @Override

protected Map<Stack<?>, WeakOrderQueue> initialValue() { **return new** WeakHashMap<Stack<?>, WeakOrderQueue>(); } };

_// a queue that makes only moderate guarantees about visibility: items are seen in the correct order,

_ // but we aren't absolutely guaranteed to ever see anything at all, thereby keeping the queue cheap to maintain private static final class WeakOrderQueue {

1    **static final** WeakOrderQueue _DUMMY_ \= **new** WeakOrderQueue();
2
3    _// Let Link extend AtomicInteger for intrinsics. The Link itself will be used as writerIndex.

_ @SuppressWarnings("serial") static final class Link extends AtomicInteger { private final DefaultHandle<?>[] elements = new DefaultHandle[LINK_CAPACITY];

1        **private int** **readIndex**;
2        Link **next**;
3    }
4
5    _// This act as a place holder for the head Link but also will reclaim space once finalized.

_ // Its important this does not hold any reference to either Stack or WeakOrderQueue. static final class Head { private final AtomicInteger availableSharedCapacity;

1        Link **link**;
2
3        Head(AtomicInteger availableSharedCapacity) {
4            **this**.**availableSharedCapacity** \= availableSharedCapacity;
5        }
6
7        _///_ _TODO: In the future when we move to Java9+ we should use java.lang.ref.Cleaner.

_ @Override protected void finalize() throws Throwable { try { super.finalize(); } finally { Link head = link; link = null; while (head != null) { reclaimSpace(LINK_CAPACITY); Link next = head.next; _// Unlink to help GC and guard against GC nepotism. _ head.next = null; head = next; } } }

1        **void** reclaimSpace(**int** space) {
2            **assert** space >= 0;
3            **availableSharedCapacity**.addAndGet(space);
4        }
5
6        **boolean** reserveSpace(**int** space) {
7            **return** _reserveSpace_(**availableSharedCapacity**, space);
8        }
9
10        **static boolean** reserveSpace(AtomicInteger availableSharedCapacity, **int** space) {
11            **assert** space >= 0;
12            **for** (;;) {
13                **int** available = availableSharedCapacity.get();
14                **if** (available < space) {
15                    **return false**;
16                }
17                **if** (availableSharedCapacity.compareAndSet(available, available - space)) {
18                    **return true**;
19                }
20            }
21        }
22    }
23
24    _// chain of data items

_ private final Head head; private Link tail; _// pointer to another queue of delayed items for the same stack _ private WeakOrderQueue next; private final WeakReference<Thread> owner; private final int id = ID_GENERATOR.getAndIncrement();

1    **private** WeakOrderQueue() {
2        **owner** \= **null**;
3        **head** \= **new** Head(**null**);
4    }
5
6    **private** WeakOrderQueue(Stack<?> stack, Thread thread) {
7        **tail** \= **new** Link();
8
9        _// Its important that we not store the Stack itself in the WeakOrderQueue as the Stack also is used in

_ // the WeakHashMap as key. So just store the enclosed AtomicInteger which should allow to have the // Stack itself GCed. head = new Head(stack.availableSharedCapacity); head.link = tail; owner = new WeakReference<Thread>(thread); }

1    **static** WeakOrderQueue newQueue(Stack<?> stack, Thread thread) {
2        **final** WeakOrderQueue queue = **new** WeakOrderQueue(stack, thread);
3        _// 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);

1        **return** queue;
2    }
3
4    **private void** setNext(WeakOrderQueue next) {
5        **assert** next != **this**;
6        **this**.**next** \= next;
7    }
8
9    _/\*\*

_ * Allocate a new {__@link WeakOrderQueue} or return {__@code _null} if not possible. _ */ static WeakOrderQueue allocate(Stack<?> stack, Thread thread) { _// We allocated a Link so reserve the space _ return Head.reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY) ? newQueue(stack, thread) : null; }

1    **void** add(DefaultHandle<?> handle) {
2        handle.**lastRecycledId** \= **id**;
3
4        Link tail = **this**.**tail**;
5        **int** writeIndex;
6        **if** ((writeIndex = tail.get()) == _LINK\_CAPACITY_) {
7            **if** (!**head**.reserveSpace(_LINK\_CAPACITY_)) {
8                _// Drop it.

_ return; } _// We allocate a Link so reserve the space _ this.tail = tail = tail.next = new Link();

1            writeIndex = tail.get();
2        }
3        tail.**elements**\[writeIndex\] = handle;
4        handle.**stack** \= **null**;
5        _// we lazy set to ensure that setting stack to null appears before we unnull it in the owning thread;

_ // this also means we guarantee visibility of an element in the queue if we see the index updated tail.lazySet(writeIndex + 1); }

1    **boolean** hasFinalData() {
2        **return** **tail**.**readIndex** != **tail**.get();
3    }
4
5    _// transfer as many items as we can from this queue to the stack, returning true if any were transferred

_ @SuppressWarnings("rawtypes") boolean transfer(Stack<?> dst) { Link head = this.head.link; if (head == null) { return false; }

1        **if** (head.**readIndex** \== _LINK\_CAPACITY_) {
2            **if** (head.**next** \== **null**) {
3                **return false**;
4            }
5            **this**.**head**.**link** \= head = head.**next**;
6            **this**.**head**.reclaimSpace(_LINK\_CAPACITY_);
7        }
8
9        **final int** srcStart = head.**readIndex**;
10        **int** srcEnd = head.get();
11        **final int** srcSize = srcEnd - srcStart;
12        **if** (srcSize == 0) {
13            **return false**;
14        }
15
16        **final int** dstSize = dst.**size**;
17        **final int** expectedCapacity = dstSize + srcSize;
18
19        **if** (expectedCapacity > dst.**elements**.**length**) {
20            **final int** actualCapacity = dst.increaseCapacity(expectedCapacity);
21            srcEnd = _min_(srcStart + actualCapacity - dstSize, srcEnd);
22        }
23
24        **if** (srcStart != srcEnd) {
25            **final** DefaultHandle\[\] srcElems = head.**elements**;
26            **final** DefaultHandle\[\] dstElems = dst.**elements**;
27            **int** newDstSize = dstSize;
28            **for** (**int** i = srcStart; i < srcEnd; i++) {
29                DefaultHandle element = srcElems\[i\];
30                **if** (element.**recycleId** \== 0) {
31                    element.**recycleId** \= element.**lastRecycledId**;
32                } **else if** (element.**recycleId** != element.**lastRecycledId**) {
33                    **throw new** IllegalStateException(**"recycled already"**);
34                }
35                srcElems\[i\] = **null**;
36
37                **if** (dst.dropHandle(element)) {
38                    _// Drop the object.

_ continue; } element.stack = dst; dstElems[newDstSize ++] = element; }

1            **if** (srcEnd == _LINK\_CAPACITY_ && head.**next** != **null**) {
2                _// Add capacity back as the Link is GCed.

_ this.head.reclaimSpace(LINK_CAPACITY); this.head.link = head.next; }

1            head.**readIndex** \= srcEnd;
2            **if** (dst.**size** \== newDstSize) {
3                **return false**;
4            }
5            dst.**size** \= newDstSize;
6            **return true**;
7        } **else** {
8            _// The destination stack is full already.

_ return false; } } }

1**static final class** Stack<T\> {
2
3    _// we keep a queue of per-thread queues, which is appended to once only, each time a new thread other

_ // than the stack owner recycles: when we run out of items in our stack we iterate this collection // to scavenge those that can be reused. this permits us to incur minimal thread synchronisation whilst // still recycling all items. final Recycler<T> parent;

    _// We store the Thread in a WeakReference as otherwise we may be the only ones that still hold a strong

_ // Reference to the Thread itself after it died because DefaultHandle will hold a reference to the Stack. // // The biggest issue is if we do not use a WeakReference the Thread may not be able to be collected at all if // the user will store a reference to the DefaultHandle somewhere and never clear this reference (or not clear // it in a timely manner). final WeakReference<Thread> threadRef; final AtomicInteger availableSharedCapacity; final int maxDelayedQueues;

1    **private final int** **maxCapacity**;
2    **private final int** **ratioMask**;
3    **private** DefaultHandle<?>\[\] **elements**;
4    **private int** **size**;
5    **private int** **handleRecycleCount** \= -1; _// Start with -1 so the first one will be recycled.

_ private WeakOrderQueue cursor, prev; private volatile WeakOrderQueue head;

1    Stack(Recycler<T\> parent, Thread thread, **int** maxCapacity, **int** maxSharedCapacityFactor,
2          **int** ratioMask, **int** maxDelayedQueues) {
3        **this**.**parent** \= parent;
4        **threadRef** \= **new** WeakReference<Thread>(thread);
5        **this**.**maxCapacity** \= maxCapacity;
6        **availableSharedCapacity** \= **new** AtomicInteger(_max_(maxCapacity / maxSharedCapacityFactor, _LINK\_CAPACITY_));
7        **elements** \= **new** DefaultHandle\[_min_(_INITIAL\_CAPACITY_, maxCapacity)\];
8        **this**.**ratioMask** \= ratioMask;
9        **this**.**maxDelayedQueues** \= maxDelayedQueues;
10    }
11
12    _// Marked as synchronized to ensure this is serialized.

_ synchronized void setHead(WeakOrderQueue queue) { queue.setNext(head); head = queue; }

1    **int** increaseCapacity(**int** expectedCapacity) {
2        **int** newCapacity = **elements**.**length**;
3        **int** maxCapacity = **this**.**maxCapacity**;
4        **do** {
5            newCapacity <<= 1;
6        } **while** (newCapacity < expectedCapacity && newCapacity < maxCapacity);
7
8        newCapacity = _min_(newCapacity, maxCapacity);
9        **if** (newCapacity != **elements**.**length**) {
10            **elements** \= Arrays._copyOf_(**elements**, newCapacity);
11        }
12
13        **return** newCapacity;
14    }
15
16    @SuppressWarnings({ **"unchecked"**, **"rawtypes"** })
17    DefaultHandle<T\> pop() {
18        **int** size = **this**.**size**;
19        **if** (size == 0) {
20            **if** (!scavenge()) {
21                **return null**;
22            }
23            size = **this**.**size**;
24        }
25        size --;
26        DefaultHandle ret = **elements**\[size\];
27        **elements**\[size\] = **null**;
28        **if** (ret.**lastRecycledId** != ret.**recycleId**) {
29            **throw new** IllegalStateException(**"recycled multiple times"**);
30        }
31        ret.**recycleId** \= 0;
32        ret.**lastRecycledId** \= 0;
33        **this**.**size** \= size;
34        **return** ret;
35    }
36
37    **boolean** scavenge() {
38        _// continue an existing scavenge, if any

_ if (scavengeSome()) { return true; }

        _// reset our scavenge cursor

_ prev = null; cursor = head; return false; }

1    **boolean** scavengeSome() {
2        WeakOrderQueue prev;
3        WeakOrderQueue cursor = **this**.**cursor**;
4        **if** (cursor == **null**) {
5            prev = **null**;
6            cursor = **head**;
7            **if** (cursor == **null**) {
8                **return false**;
9            }
10        } **else** {
11            prev = **this**.**prev**;
12        }
13
14        **boolean** success = **false**;
15        **do** {
16            **if** (cursor.transfer(**this**)) {
17                success = **true**;
18                **break**;
19            }
20            WeakOrderQueue next = cursor.**next**;
21            **if** (cursor.**owner**.get() == **null**) {
22                _// If the thread associated with the queue is gone, unlink it, after

_ // performing a volatile read to confirm there is no data left to collect. // We never unlink the first queue, as we don't want to synchronize on updating the head. if (cursor.hasFinalData()) { for (;;) { if (cursor.transfer(this)) { success = true; } else { break; } } }

1                **if** (prev != **null**) {
2                    prev.setNext(next);
3                }
4            } **else** {
5                prev = cursor;
6            }
7
8            cursor = next;
9
10        } **while** (cursor != **null** && !success);
11
12        **this**.**prev** \= prev;
13        **this**.**cursor** \= cursor;
14        **return** success;
15    }
16
17    **void** push(DefaultHandle<?> item) {
18        Thread currentThread = Thread._currentThread_();
19        **if** (**threadRef**.get() == currentThread) {
20            _// The current Thread is the thread that belongs to the Stack, we can try to push the object now.

_ pushNow(item); } else { _// The current Thread is not the one that belongs to the Stack _ // (or the Thread that belonged to the Stack was collected already), we need to signal that the push // happens later. pushLater(item, currentThread); } }

1    **private void** pushNow(DefaultHandle<?> item) {
2        **if** ((item.**recycleId** | item.**lastRecycledId**) != 0) {
3            **throw new** IllegalStateException(**"recycled already"**);
4        }
5        item.**recycleId** \= item.**lastRecycledId** \= _OWN\_THREAD\_ID_;
6
7        **int** size = **this**.**size**;
8        **if** (size >= **maxCapacity** || dropHandle(item)) {
9            _// Hit the maximum capacity or should drop - drop the possibly youngest object.

_ return; } if (size == elements.length) { elements = Arrays.copyOf(elements, min(size << 1, maxCapacity)); }

1        **elements**\[size\] = item;
2        **this**.**size** \= size + 1;
3    }
4
5    **private void** pushLater(DefaultHandle<?> item, Thread thread) {
6        _// we don't want to have a ref to the queue as the value in our weak map

_ // so we null it out; to ensure there are no races with restoring it later // we impose a memory ordering here (no-op on x86) Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get(); WeakOrderQueue queue = delayedRecycled.get(this); if (queue == null) { if (delayedRecycled.size() >= maxDelayedQueues) { _// Add a dummy queue so we know we should drop the object _ delayedRecycled.put(this, WeakOrderQueue.DUMMY); return; } _// Check if we already reached the maximum number of delayed queues and if we can allocate at all. _ if ((queue = WeakOrderQueue.allocate(this, thread)) == null) { _// drop object _ return; } delayedRecycled.put(this, queue); } else if (queue == WeakOrderQueue.DUMMY) { _// drop object _ return; }

1        queue.add(item);
2    }
3
4    **boolean** dropHandle(DefaultHandle<?> handle) {
5        **if** (!handle.**hasBeenRecycled**) {
6            **if** ((++**handleRecycleCount** & **ratioMask**) != 0) {
7                _// Drop the object.

_ return true; } handle.hasBeenRecycled = true; } return false; }

1    DefaultHandle<T\> newHandle() {
2        **return new** DefaultHandle<T\>(**this**);
3    }
4}

}