1 /*
2 * Copyright 2012 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a copy of the License at:
7 *
8 * https://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16
17 package io.netty.buffer;
18
19 import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
20
21 import io.netty.util.NettyRuntime;
22 import io.netty.util.concurrent.EventExecutor;
23 import io.netty.util.concurrent.FastThreadLocal;
24 import io.netty.util.concurrent.FastThreadLocalThread;
25 import io.netty.util.internal.PlatformDependent;
26 import io.netty.util.internal.StringUtil;
27 import io.netty.util.internal.SystemPropertyUtil;
28 import io.netty.util.internal.ThreadExecutorMap;
29 import io.netty.util.internal.logging.InternalLogger;
30 import io.netty.util.internal.logging.InternalLoggerFactory;
31
32 import java.nio.ByteBuffer;
33 import java.util.ArrayList;
34 import java.util.Collections;
35 import java.util.List;
36 import java.util.concurrent.TimeUnit;
37
38 public class PooledByteBufAllocator extends AbstractByteBufAllocator implements ByteBufAllocatorMetricProvider {
39
40 private static final InternalLogger logger = InternalLoggerFactory.getInstance(PooledByteBufAllocator.class);
41 private static final int DEFAULT_NUM_HEAP_ARENA;
42 private static final int DEFAULT_NUM_DIRECT_ARENA;
43
44 private static final int DEFAULT_PAGE_SIZE;
45 private static final int DEFAULT_MAX_ORDER; // 8192 << 9 = 4 MiB per chunk
46 private static final int DEFAULT_SMALL_CACHE_SIZE;
47 private static final int DEFAULT_NORMAL_CACHE_SIZE;
48 static final int DEFAULT_MAX_CACHED_BUFFER_CAPACITY;
49 private static final int DEFAULT_CACHE_TRIM_INTERVAL;
50 private static final long DEFAULT_CACHE_TRIM_INTERVAL_MILLIS;
51 private static final boolean DEFAULT_USE_CACHE_FOR_ALL_THREADS;
52 private static final int DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT;
53 static final int DEFAULT_MAX_CACHED_BYTEBUFFERS_PER_CHUNK;
54
55 private static final int MIN_PAGE_SIZE = 4096;
56 private static final int MAX_CHUNK_SIZE = (int) (((long) Integer.MAX_VALUE + 1) / 2);
57
58 private static final int CACHE_NOT_USED = 0;
59
60 private final Runnable trimTask = new Runnable() {
61 @Override
62 public void run() {
63 PooledByteBufAllocator.this.trimCurrentThreadCache();
64 }
65 };
66
67 static {
68 int defaultAlignment = SystemPropertyUtil.getInt(
69 "io.netty.allocator.directMemoryCacheAlignment", 0);
70 int defaultPageSize = SystemPropertyUtil.getInt("io.netty.allocator.pageSize", 8192);
71 Throwable pageSizeFallbackCause = null;
72 try {
73 validateAndCalculatePageShifts(defaultPageSize, defaultAlignment);
74 } catch (Throwable t) {
75 pageSizeFallbackCause = t;
76 defaultPageSize = 8192;
77 defaultAlignment = 0;
78 }
79 DEFAULT_PAGE_SIZE = defaultPageSize;
80 DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT = defaultAlignment;
81
82 int defaultMaxOrder = SystemPropertyUtil.getInt("io.netty.allocator.maxOrder", 9);
83 Throwable maxOrderFallbackCause = null;
84 try {
85 validateAndCalculateChunkSize(DEFAULT_PAGE_SIZE, defaultMaxOrder);
86 } catch (Throwable t) {
87 maxOrderFallbackCause = t;
88 defaultMaxOrder = 9;
89 }
90 DEFAULT_MAX_ORDER = defaultMaxOrder;
91
92 // Determine reasonable default for nHeapArena and nDirectArena.
93 // Assuming each arena has 3 chunks, the pool should not consume more than 50% of max memory.
94 final Runtime runtime = Runtime.getRuntime();
95
96 /*
97 * We use 2 * available processors by default to reduce contention as we use 2 * available processors for the
98 * number of EventLoops in NIO and EPOLL as well. If we choose a smaller number we will run into hot spots as
99 * allocation and de-allocation needs to be synchronized on the PoolArena.
100 *
101 * See https://github.com/netty/netty/issues/3888.
102 */
103 final int defaultMinNumArena = NettyRuntime.availableProcessors() * 2;
104 final int defaultChunkSize = DEFAULT_PAGE_SIZE << DEFAULT_MAX_ORDER;
105 DEFAULT_NUM_HEAP_ARENA = Math.max(0,
106 SystemPropertyUtil.getInt(
107 "io.netty.allocator.numHeapArenas",
108 (int) Math.min(
109 defaultMinNumArena,
110 runtime.maxMemory() / defaultChunkSize / 2 / 3)));
111 DEFAULT_NUM_DIRECT_ARENA = Math.max(0,
112 SystemPropertyUtil.getInt(
113 "io.netty.allocator.numDirectArenas",
114 (int) Math.min(
115 defaultMinNumArena,
116 PlatformDependent.maxDirectMemory() / defaultChunkSize / 2 / 3)));
117
118 // cache sizes
119 DEFAULT_SMALL_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.smallCacheSize", 256);
120 DEFAULT_NORMAL_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.normalCacheSize", 64);
121
122 // 32 kb is the default maximum capacity of the cached buffer. Similar to what is explained in
123 // 'Scalable memory allocation using jemalloc'
124 DEFAULT_MAX_CACHED_BUFFER_CAPACITY = SystemPropertyUtil.getInt(
125 "io.netty.allocator.maxCachedBufferCapacity", 32 * 1024);
126
127 // the number of threshold of allocations when cached entries will be freed up if not frequently used
128 DEFAULT_CACHE_TRIM_INTERVAL = SystemPropertyUtil.getInt(
129 "io.netty.allocator.cacheTrimInterval", 8192);
130
131 if (SystemPropertyUtil.contains("io.netty.allocation.cacheTrimIntervalMillis")) {
132 logger.warn("-Dio.netty.allocation.cacheTrimIntervalMillis is deprecated," +
133 " use -Dio.netty.allocator.cacheTrimIntervalMillis");
134
135 if (SystemPropertyUtil.contains("io.netty.allocator.cacheTrimIntervalMillis")) {
136 // Both system properties are specified. Use the non-deprecated one.
137 DEFAULT_CACHE_TRIM_INTERVAL_MILLIS = SystemPropertyUtil.getLong(
138 "io.netty.allocator.cacheTrimIntervalMillis", 0);
139 } else {
140 DEFAULT_CACHE_TRIM_INTERVAL_MILLIS = SystemPropertyUtil.getLong(
141 "io.netty.allocation.cacheTrimIntervalMillis", 0);
142 }
143 } else {
144 DEFAULT_CACHE_TRIM_INTERVAL_MILLIS = SystemPropertyUtil.getLong(
145 "io.netty.allocator.cacheTrimIntervalMillis", 0);
146 }
147
148 DEFAULT_USE_CACHE_FOR_ALL_THREADS = SystemPropertyUtil.getBoolean(
149 "io.netty.allocator.useCacheForAllThreads", false);
150
151 // Use 1023 by default as we use an ArrayDeque as backing storage which will then allocate an internal array
152 // of 1024 elements. Otherwise we would allocate 2048 and only use 1024 which is wasteful.
153 DEFAULT_MAX_CACHED_BYTEBUFFERS_PER_CHUNK = SystemPropertyUtil.getInt(
154 "io.netty.allocator.maxCachedByteBuffersPerChunk", 1023);
155
156 if (logger.isDebugEnabled()) {
157 logger.debug("-Dio.netty.allocator.numHeapArenas: {}", DEFAULT_NUM_HEAP_ARENA);
158 logger.debug("-Dio.netty.allocator.numDirectArenas: {}", DEFAULT_NUM_DIRECT_ARENA);
159 if (pageSizeFallbackCause == null) {
160 logger.debug("-Dio.netty.allocator.pageSize: {}", DEFAULT_PAGE_SIZE);
161 } else {
162 logger.debug("-Dio.netty.allocator.pageSize: {}", DEFAULT_PAGE_SIZE, pageSizeFallbackCause);
163 }
164 if (maxOrderFallbackCause == null) {
165 logger.debug("-Dio.netty.allocator.maxOrder: {}", DEFAULT_MAX_ORDER);
166 } else {
167 logger.debug("-Dio.netty.allocator.maxOrder: {}", DEFAULT_MAX_ORDER, maxOrderFallbackCause);
168 }
169 logger.debug("-Dio.netty.allocator.chunkSize: {}", DEFAULT_PAGE_SIZE << DEFAULT_MAX_ORDER);
170 logger.debug("-Dio.netty.allocator.smallCacheSize: {}", DEFAULT_SMALL_CACHE_SIZE);
171 logger.debug("-Dio.netty.allocator.normalCacheSize: {}", DEFAULT_NORMAL_CACHE_SIZE);
172 logger.debug("-Dio.netty.allocator.maxCachedBufferCapacity: {}", DEFAULT_MAX_CACHED_BUFFER_CAPACITY);
173 logger.debug("-Dio.netty.allocator.cacheTrimInterval: {}", DEFAULT_CACHE_TRIM_INTERVAL);
174 logger.debug("-Dio.netty.allocator.cacheTrimIntervalMillis: {}", DEFAULT_CACHE_TRIM_INTERVAL_MILLIS);
175 logger.debug("-Dio.netty.allocator.useCacheForAllThreads: {}", DEFAULT_USE_CACHE_FOR_ALL_THREADS);
176 logger.debug("-Dio.netty.allocator.maxCachedByteBuffersPerChunk: {}",
177 DEFAULT_MAX_CACHED_BYTEBUFFERS_PER_CHUNK);
178 }
179 }
180
181 public static final PooledByteBufAllocator DEFAULT =
182 new PooledByteBufAllocator(PlatformDependent.directBufferPreferred());
183
184 private final PoolArena<byte[]>[] heapArenas;
185 private final PoolArena<ByteBuffer>[] directArenas;
186 private final int smallCacheSize;
187 private final int normalCacheSize;
188 private final List<PoolArenaMetric> heapArenaMetrics;
189 private final List<PoolArenaMetric> directArenaMetrics;
190 private final PoolThreadLocalCache threadCache;
191 private final int chunkSize;
192 private final PooledByteBufAllocatorMetric metric;
193
194 public PooledByteBufAllocator() {
195 this(false);
196 }
197
198 @SuppressWarnings("deprecation")
199 public PooledByteBufAllocator(boolean preferDirect) {
200 this(preferDirect, DEFAULT_NUM_HEAP_ARENA, DEFAULT_NUM_DIRECT_ARENA, DEFAULT_PAGE_SIZE, DEFAULT_MAX_ORDER);
201 }
202
203 @SuppressWarnings("deprecation")
204 public PooledByteBufAllocator(int nHeapArena, int nDirectArena, int pageSize, int maxOrder) {
205 this(false, nHeapArena, nDirectArena, pageSize, maxOrder);
206 }
207
208 /**
209 * @deprecated use
210 * {@link PooledByteBufAllocator#PooledByteBufAllocator(boolean, int, int, int, int, int, int, boolean)}
211 */
212 @Deprecated
213 public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder) {
214 this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder,
215 0, DEFAULT_SMALL_CACHE_SIZE, DEFAULT_NORMAL_CACHE_SIZE);
216 }
217
218 /**
219 * @deprecated use
220 * {@link PooledByteBufAllocator#PooledByteBufAllocator(boolean, int, int, int, int, int, int, boolean)}
221 */
222 @Deprecated
223 public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
224 int tinyCacheSize, int smallCacheSize, int normalCacheSize) {
225 this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder, smallCacheSize,
226 normalCacheSize, DEFAULT_USE_CACHE_FOR_ALL_THREADS, DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT);
227 }
228
229 /**
230 * @deprecated use
231 * {@link PooledByteBufAllocator#PooledByteBufAllocator(boolean, int, int, int, int, int, int, boolean)}
232 */
233 @Deprecated
234 public PooledByteBufAllocator(boolean preferDirect, int nHeapArena,
235 int nDirectArena, int pageSize, int maxOrder, int tinyCacheSize,
236 int smallCacheSize, int normalCacheSize,
237 boolean useCacheForAllThreads) {
238 this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder,
239 smallCacheSize, normalCacheSize,
240 useCacheForAllThreads);
241 }
242
243 public PooledByteBufAllocator(boolean preferDirect, int nHeapArena,
244 int nDirectArena, int pageSize, int maxOrder,
245 int smallCacheSize, int normalCacheSize,
246 boolean useCacheForAllThreads) {
247 this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder,
248 smallCacheSize, normalCacheSize,
249 useCacheForAllThreads, DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT);
250 }
251
252 /**
253 * @deprecated use
254 * {@link PooledByteBufAllocator#PooledByteBufAllocator(boolean, int, int, int, int, int, int, boolean, int)}
255 */
256 @Deprecated
257 public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
258 int tinyCacheSize, int smallCacheSize, int normalCacheSize,
259 boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
260 this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder,
261 smallCacheSize, normalCacheSize,
262 useCacheForAllThreads, directMemoryCacheAlignment);
263 }
264
265 public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
266 int smallCacheSize, int normalCacheSize,
267 boolean useCacheForAllThreads, int directMemoryCacheAlignment) {
268 super(preferDirect);
269 threadCache = new PoolThreadLocalCache(useCacheForAllThreads);
270 this.smallCacheSize = smallCacheSize;
271 this.normalCacheSize = normalCacheSize;
272
273 if (directMemoryCacheAlignment != 0) {
274 if (!PlatformDependent.hasAlignDirectByteBuffer()) {
275 throw new UnsupportedOperationException("Buffer alignment is not supported. " +
276 "Either Unsafe or ByteBuffer.alignSlice() must be available.");
277 }
278
279 // Ensure page size is a whole multiple of the alignment, or bump it to the next whole multiple.
280 pageSize = (int) PlatformDependent.align(pageSize, directMemoryCacheAlignment);
281 }
282
283 chunkSize = validateAndCalculateChunkSize(pageSize, maxOrder);
284
285 checkPositiveOrZero(nHeapArena, "nHeapArena");
286 checkPositiveOrZero(nDirectArena, "nDirectArena");
287
288 checkPositiveOrZero(directMemoryCacheAlignment, "directMemoryCacheAlignment");
289 if (directMemoryCacheAlignment > 0 && !isDirectMemoryCacheAlignmentSupported()) {
290 throw new IllegalArgumentException("directMemoryCacheAlignment is not supported");
291 }
292
293 if ((directMemoryCacheAlignment & -directMemoryCacheAlignment) != directMemoryCacheAlignment) {
294 throw new IllegalArgumentException("directMemoryCacheAlignment: "
295 + directMemoryCacheAlignment + " (expected: power of two)");
296 }
297
298 int pageShifts = validateAndCalculatePageShifts(pageSize, directMemoryCacheAlignment);
299
300 if (nHeapArena > 0) {
301 heapArenas = newArenaArray(nHeapArena);
302 List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(heapArenas.length);
303 final SizeClasses sizeClasses = new SizeClasses(pageSize, pageShifts, chunkSize, 0);
304 for (int i = 0; i < heapArenas.length; i ++) {
305 PoolArena.HeapArena arena = new PoolArena.HeapArena(this, sizeClasses);
306 heapArenas[i] = arena;
307 metrics.add(arena);
308 }
309 heapArenaMetrics = Collections.unmodifiableList(metrics);
310 } else {
311 heapArenas = null;
312 heapArenaMetrics = Collections.emptyList();
313 }
314
315 if (nDirectArena > 0) {
316 directArenas = newArenaArray(nDirectArena);
317 List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(directArenas.length);
318 final SizeClasses sizeClasses = new SizeClasses(pageSize, pageShifts, chunkSize,
319 directMemoryCacheAlignment);
320 for (int i = 0; i < directArenas.length; i ++) {
321 PoolArena.DirectArena arena = new PoolArena.DirectArena(this, sizeClasses);
322 directArenas[i] = arena;
323 metrics.add(arena);
324 }
325 directArenaMetrics = Collections.unmodifiableList(metrics);
326 } else {
327 directArenas = null;
328 directArenaMetrics = Collections.emptyList();
329 }
330 metric = new PooledByteBufAllocatorMetric(this);
331 }
332
333 @SuppressWarnings("unchecked")
334 private static <T> PoolArena<T>[] newArenaArray(int size) {
335 return new PoolArena[size];
336 }
337
338 private static int validateAndCalculatePageShifts(int pageSize, int alignment) {
339 if (pageSize < MIN_PAGE_SIZE) {
340 throw new IllegalArgumentException("pageSize: " + pageSize + " (expected: " + MIN_PAGE_SIZE + ')');
341 }
342
343 if ((pageSize & pageSize - 1) != 0) {
344 throw new IllegalArgumentException("pageSize: " + pageSize + " (expected: power of 2)");
345 }
346
347 if (pageSize < alignment) {
348 throw new IllegalArgumentException("Alignment cannot be greater than page size. " +
349 "Alignment: " + alignment + ", page size: " + pageSize + '.');
350 }
351
352 // Logarithm base 2. At this point we know that pageSize is a power of two.
353 return Integer.SIZE - 1 - Integer.numberOfLeadingZeros(pageSize);
354 }
355
356 private static int validateAndCalculateChunkSize(int pageSize, int maxOrder) {
357 if (maxOrder > 14) {
358 throw new IllegalArgumentException("maxOrder: " + maxOrder + " (expected: 0-14)");
359 }
360
361 // Ensure the resulting chunkSize does not overflow.
362 int chunkSize = pageSize;
363 for (int i = maxOrder; i > 0; i --) {
364 if (chunkSize > MAX_CHUNK_SIZE / 2) {
365 throw new IllegalArgumentException(String.format(
366 "pageSize (%d) << maxOrder (%d) must not exceed %d", pageSize, maxOrder, MAX_CHUNK_SIZE));
367 }
368 chunkSize <<= 1;
369 }
370 return chunkSize;
371 }
372
373 @Override
374 protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
375 PoolThreadCache cache = threadCache.get();
376 PoolArena<byte[]> heapArena = cache.heapArena;
377
378 final ByteBuf buf;
379 if (heapArena != null) {
380 buf = heapArena.allocate(cache, initialCapacity, maxCapacity);
381 } else {
382 buf = PlatformDependent.hasUnsafe() ?
383 new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
384 new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
385 }
386
387 return toLeakAwareBuffer(buf);
388 }
389
390 @Override
391 protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
392 PoolThreadCache cache = threadCache.get();
393 PoolArena<ByteBuffer> directArena = cache.directArena;
394
395 final ByteBuf buf;
396 if (directArena != null) {
397 buf = directArena.allocate(cache, initialCapacity, maxCapacity);
398 } else {
399 buf = PlatformDependent.hasUnsafe() ?
400 UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
401 new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
402 }
403
404 return toLeakAwareBuffer(buf);
405 }
406
407 /**
408 * Default number of heap arenas - System Property: io.netty.allocator.numHeapArenas - default 2 * cores
409 */
410 public static int defaultNumHeapArena() {
411 return DEFAULT_NUM_HEAP_ARENA;
412 }
413
414 /**
415 * Default number of direct arenas - System Property: io.netty.allocator.numDirectArenas - default 2 * cores
416 */
417 public static int defaultNumDirectArena() {
418 return DEFAULT_NUM_DIRECT_ARENA;
419 }
420
421 /**
422 * Default buffer page size - System Property: io.netty.allocator.pageSize - default 8192
423 */
424 public static int defaultPageSize() {
425 return DEFAULT_PAGE_SIZE;
426 }
427
428 /**
429 * Default maximum order - System Property: io.netty.allocator.maxOrder - default 9
430 */
431 public static int defaultMaxOrder() {
432 return DEFAULT_MAX_ORDER;
433 }
434
435 /**
436 * Default thread caching behavior - System Property: io.netty.allocator.useCacheForAllThreads - default false
437 */
438 public static boolean defaultUseCacheForAllThreads() {
439 return DEFAULT_USE_CACHE_FOR_ALL_THREADS;
440 }
441
442 /**
443 * Default prefer direct - System Property: io.netty.noPreferDirect - default false
444 */
445 public static boolean defaultPreferDirect() {
446 return PlatformDependent.directBufferPreferred();
447 }
448
449 /**
450 * Default tiny cache size - default 0
451 *
452 * @deprecated Tiny caches have been merged into small caches.
453 */
454 @Deprecated
455 public static int defaultTinyCacheSize() {
456 return 0;
457 }
458
459 /**
460 * Default small cache size - System Property: io.netty.allocator.smallCacheSize - default 256
461 */
462 public static int defaultSmallCacheSize() {
463 return DEFAULT_SMALL_CACHE_SIZE;
464 }
465
466 /**
467 * Default normal cache size - System Property: io.netty.allocator.normalCacheSize - default 64
468 */
469 public static int defaultNormalCacheSize() {
470 return DEFAULT_NORMAL_CACHE_SIZE;
471 }
472
473 /**
474 * Return {@code true} if direct memory cache alignment is supported, {@code false} otherwise.
475 */
476 public static boolean isDirectMemoryCacheAlignmentSupported() {
477 return PlatformDependent.hasUnsafe();
478 }
479
480 @Override
481 public boolean isDirectBufferPooled() {
482 return directArenas != null;
483 }
484
485 /**
486 * @deprecated will be removed
487 * Returns {@code true} if the calling {@link Thread} has a {@link ThreadLocal} cache for the allocated
488 * buffers.
489 */
490 @Deprecated
491 public boolean hasThreadLocalCache() {
492 return threadCache.isSet();
493 }
494
495 /**
496 * @deprecated will be removed
497 * Free all cached buffers for the calling {@link Thread}.
498 */
499 @Deprecated
500 public void freeThreadLocalCache() {
501 threadCache.remove();
502 }
503
504 private final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache> {
505 private final boolean useCacheForAllThreads;
506
507 PoolThreadLocalCache(boolean useCacheForAllThreads) {
508 this.useCacheForAllThreads = useCacheForAllThreads;
509 }
510
511 @Override
512 protected synchronized PoolThreadCache initialValue() {
513 final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas);
514 final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas);
515
516 final Thread current = Thread.currentThread();
517 final EventExecutor executor = ThreadExecutorMap.currentExecutor();
518
519 if (useCacheForAllThreads ||
520 // If the current thread is a FastThreadLocalThread we will always use the cache
521 current instanceof FastThreadLocalThread ||
522 // The Thread is used by an EventExecutor, let's use the cache as the chances are good that we
523 // will allocate a lot!
524 executor != null) {
525 final PoolThreadCache cache = new PoolThreadCache(
526 heapArena, directArena, smallCacheSize, normalCacheSize,
527 DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL, true);
528
529 if (DEFAULT_CACHE_TRIM_INTERVAL_MILLIS > 0) {
530 if (executor != null) {
531 executor.scheduleAtFixedRate(trimTask, DEFAULT_CACHE_TRIM_INTERVAL_MILLIS,
532 DEFAULT_CACHE_TRIM_INTERVAL_MILLIS, TimeUnit.MILLISECONDS);
533 }
534 }
535 return cache;
536 }
537 // No caching so just use 0 as sizes.
538 return new PoolThreadCache(heapArena, directArena, 0, 0, 0, 0, false);
539 }
540
541 @Override
542 protected void onRemoval(PoolThreadCache threadCache) {
543 threadCache.free(false);
544 }
545
546 private <T> PoolArena<T> leastUsedArena(PoolArena<T>[] arenas) {
547 if (arenas == null || arenas.length == 0) {
548 return null;
549 }
550
551 PoolArena<T> minArena = arenas[0];
552 //optimized
553 //If it is the first execution, directly return minarena and reduce the number of for loop comparisons below
554 if (minArena.numThreadCaches.get() == CACHE_NOT_USED) {
555 return minArena;
556 }
557 for (int i = 1; i < arenas.length; i++) {
558 PoolArena<T> arena = arenas[i];
559 if (arena.numThreadCaches.get() < minArena.numThreadCaches.get()) {
560 minArena = arena;
561 }
562 }
563
564 return minArena;
565 }
566 }
567
568 @Override
569 public PooledByteBufAllocatorMetric metric() {
570 return metric;
571 }
572
573 /**
574 * Return the number of heap arenas.
575 *
576 * @deprecated use {@link PooledByteBufAllocatorMetric#numHeapArenas()}.
577 */
578 @Deprecated
579 public int numHeapArenas() {
580 return heapArenaMetrics.size();
581 }
582
583 /**
584 * Return the number of direct arenas.
585 *
586 * @deprecated use {@link PooledByteBufAllocatorMetric#numDirectArenas()}.
587 */
588 @Deprecated
589 public int numDirectArenas() {
590 return directArenaMetrics.size();
591 }
592
593 /**
594 * Return a {@link List} of all heap {@link PoolArenaMetric}s that are provided by this pool.
595 *
596 * @deprecated use {@link PooledByteBufAllocatorMetric#heapArenas()}.
597 */
598 @Deprecated
599 public List<PoolArenaMetric> heapArenas() {
600 return heapArenaMetrics;
601 }
602
603 /**
604 * Return a {@link List} of all direct {@link PoolArenaMetric}s that are provided by this pool.
605 *
606 * @deprecated use {@link PooledByteBufAllocatorMetric#directArenas()}.
607 */
608 @Deprecated
609 public List<PoolArenaMetric> directArenas() {
610 return directArenaMetrics;
611 }
612
613 /**
614 * Return the number of thread local caches used by this {@link PooledByteBufAllocator}.
615 *
616 * @deprecated use {@link PooledByteBufAllocatorMetric#numThreadLocalCaches()}.
617 */
618 @Deprecated
619 public int numThreadLocalCaches() {
620 PoolArena<?>[] arenas = heapArenas != null ? heapArenas : directArenas;
621 if (arenas == null) {
622 return 0;
623 }
624
625 int total = 0;
626 for (PoolArena<?> arena : arenas) {
627 total += arena.numThreadCaches.get();
628 }
629
630 return total;
631 }
632
633 /**
634 * Return the size of the tiny cache.
635 *
636 * @deprecated use {@link PooledByteBufAllocatorMetric#tinyCacheSize()}.
637 */
638 @Deprecated
639 public int tinyCacheSize() {
640 return 0;
641 }
642
643 /**
644 * Return the size of the small cache.
645 *
646 * @deprecated use {@link PooledByteBufAllocatorMetric#smallCacheSize()}.
647 */
648 @Deprecated
649 public int smallCacheSize() {
650 return smallCacheSize;
651 }
652
653 /**
654 * Return the size of the normal cache.
655 *
656 * @deprecated use {@link PooledByteBufAllocatorMetric#normalCacheSize()}.
657 */
658 @Deprecated
659 public int normalCacheSize() {
660 return normalCacheSize;
661 }
662
663 /**
664 * Return the chunk size for an arena.
665 *
666 * @deprecated use {@link PooledByteBufAllocatorMetric#chunkSize()}.
667 */
668 @Deprecated
669 public final int chunkSize() {
670 return chunkSize;
671 }
672
673 final long usedHeapMemory() {
674 return usedMemory(heapArenas);
675 }
676
677 final long usedDirectMemory() {
678 return usedMemory(directArenas);
679 }
680
681 private static long usedMemory(PoolArena<?>[] arenas) {
682 if (arenas == null) {
683 return -1;
684 }
685 long used = 0;
686 for (PoolArena<?> arena : arenas) {
687 used += arena.numActiveBytes();
688 if (used < 0) {
689 return Long.MAX_VALUE;
690 }
691 }
692 return used;
693 }
694
695 /**
696 * Returns the number of bytes of heap memory that is currently pinned to heap buffers allocated by a
697 * {@link ByteBufAllocator}, or {@code -1} if unknown.
698 * A buffer can pin more memory than its {@linkplain ByteBuf#capacity() capacity} might indicate,
699 * due to implementation details of the allocator.
700 */
701 public final long pinnedHeapMemory() {
702 return pinnedMemory(heapArenas);
703 }
704
705 /**
706 * Returns the number of bytes of direct memory that is currently pinned to direct buffers allocated by a
707 * {@link ByteBufAllocator}, or {@code -1} if unknown.
708 * A buffer can pin more memory than its {@linkplain ByteBuf#capacity() capacity} might indicate,
709 * due to implementation details of the allocator.
710 */
711 public final long pinnedDirectMemory() {
712 return pinnedMemory(directArenas);
713 }
714
715 private static long pinnedMemory(PoolArena<?>[] arenas) {
716 if (arenas == null) {
717 return -1;
718 }
719 long used = 0;
720 for (PoolArena<?> arena : arenas) {
721 used += arena.numPinnedBytes();
722 if (used < 0) {
723 return Long.MAX_VALUE;
724 }
725 }
726 return used;
727 }
728
729 final PoolThreadCache threadCache() {
730 PoolThreadCache cache = threadCache.get();
731 assert cache != null;
732 return cache;
733 }
734
735 /**
736 * Trim thread local cache for the current {@link Thread}, which will give back any cached memory that was not
737 * allocated frequently since the last trim operation.
738 *
739 * Returns {@code true} if a cache for the current {@link Thread} exists and so was trimmed, false otherwise.
740 */
741 public boolean trimCurrentThreadCache() {
742 PoolThreadCache cache = threadCache.getIfExists();
743 if (cache != null) {
744 cache.trim();
745 return true;
746 }
747 return false;
748 }
749
750 /**
751 * Returns the status of the allocator (which contains all metrics) as string. Be aware this may be expensive
752 * and so should not called too frequently.
753 */
754 public String dumpStats() {
755 int heapArenasLen = heapArenas == null ? 0 : heapArenas.length;
756 StringBuilder buf = new StringBuilder(512)
757 .append(heapArenasLen)
758 .append(" heap arena(s):")
759 .append(StringUtil.NEWLINE);
760 if (heapArenasLen > 0) {
761 for (PoolArena<byte[]> a: heapArenas) {
762 buf.append(a);
763 }
764 }
765
766 int directArenasLen = directArenas == null ? 0 : directArenas.length;
767
768 buf.append(directArenasLen)
769 .append(" direct arena(s):")
770 .append(StringUtil.NEWLINE);
771 if (directArenasLen > 0) {
772 for (PoolArena<ByteBuffer> a: directArenas) {
773 buf.append(a);
774 }
775 }
776
777 return buf.toString();
778 }
779 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！