1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.buffer;
17
18 import io.netty.util.ByteProcessor;
19 import io.netty.util.CharsetUtil;
20 import io.netty.util.IllegalReferenceCountException;
21 import io.netty.util.IntConsumer;
22 import io.netty.util.NettyRuntime;
23 import io.netty.util.Recycler;
24 import io.netty.util.Recycler.EnhancedHandle;
25 import io.netty.util.ReferenceCounted;
26 import io.netty.util.concurrent.ConcurrentSkipListIntObjMultimap;
27 import io.netty.util.concurrent.ConcurrentSkipListIntObjMultimap.IntEntry;
28 import io.netty.util.concurrent.FastThreadLocal;
29 import io.netty.util.concurrent.FastThreadLocalThread;
30 import io.netty.util.concurrent.MpscAtomicIntegerArrayQueue;
31 import io.netty.util.concurrent.MpscIntQueue;
32 import io.netty.util.internal.LongCounter;
33 import io.netty.util.internal.MathUtil;
34 import io.netty.util.internal.ObjectUtil;
35 import io.netty.util.internal.PlatformDependent;
36 import io.netty.util.internal.ReferenceCountUpdater;
37 import io.netty.util.internal.SuppressJava6Requirement;
38 import io.netty.util.internal.SystemPropertyUtil;
39 import io.netty.util.internal.ThreadExecutorMap;
40 import io.netty.util.internal.UnstableApi;
41
42 import java.io.IOException;
43 import java.io.InputStream;
44 import java.io.OutputStream;
45 import java.nio.ByteBuffer;
46 import java.nio.ByteOrder;
47 import java.nio.channels.ClosedChannelException;
48 import java.nio.channels.FileChannel;
49 import java.nio.channels.GatheringByteChannel;
50 import java.nio.channels.ScatteringByteChannel;
51 import java.nio.charset.Charset;
52 import java.util.ArrayList;
53 import java.util.Arrays;
54 import java.util.Iterator;
55 import java.util.Queue;
56 import java.util.concurrent.atomic.AtomicInteger;
57 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
58 import java.util.concurrent.atomic.AtomicLong;
59 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
60 import java.util.concurrent.atomic.LongAdder;
61 import java.util.concurrent.locks.StampedLock;
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88 @SuppressJava6Requirement(reason = "Guarded by version check")
89 @UnstableApi
90 final class AdaptivePoolingAllocator implements AdaptiveByteBufAllocator.AdaptiveAllocatorApi {
91 private static final int LOW_MEM_THRESHOLD = 512 * 1024 * 1024;
92 private static final boolean IS_LOW_MEM = SystemPropertyUtil.getBoolean(
93 "io.netty.allocator.lowMemory",
94 Runtime.getRuntime().maxMemory() <= LOW_MEM_THRESHOLD);
95
96
97
98
99
100 private static final boolean DISABLE_THREAD_LOCAL_MAGAZINES_ON_LOW_MEM = SystemPropertyUtil.getBoolean(
101 "io.netty.allocator.disableThreadLocalMagazinesOnLowMemory", true);
102
103
104
105
106
107
108
109
110 static final int MIN_CHUNK_SIZE = 128 * 1024;
111 private static final int EXPANSION_ATTEMPTS = 3;
112 private static final int INITIAL_MAGAZINES = 1;
113 private static final int RETIRE_CAPACITY = 256;
114 private static final int MAX_STRIPES = IS_LOW_MEM ? 1 : NettyRuntime.availableProcessors() * 2;
115 private static final int BUFS_PER_CHUNK = 8;
116
117
118
119
120
121
122 private static final int MAX_CHUNK_SIZE = IS_LOW_MEM ?
123 2 * 1024 * 1024 :
124 8 * 1024 * 1024;
125 private static final int MAX_POOLED_BUF_SIZE = MAX_CHUNK_SIZE / BUFS_PER_CHUNK;
126
127
128
129
130
131
132 static final int CHUNK_REUSE_QUEUE = Math.max(2, SystemPropertyUtil.getInt(
133 "io.netty.allocator.chunkReuseQueueCapacity", NettyRuntime.availableProcessors() * 2));
134
135 static final long CHUNK_PURGE_POLLS_THREAD_LOCAL = Math.max(1, SystemPropertyUtil.getLong(
136 "io.netty.allocator.chunkPurgePollsThreadLocal", 16L));
137
138 static final long CHUNK_PURGE_POLLS_SHARED = Math.max(1, SystemPropertyUtil.getLong(
139 "io.netty.allocator.chunkPurgePollsShared", 128L));
140
141 static final int CHUNK_PURGE_THRESHOLD = Math.max(1, SystemPropertyUtil.getInt(
142 "io.netty.allocator.chunkPurgeThreshold", 3));
143
144
145
146
147
148
149
150
151 static final int THREAD_LOCAL_CACHE_MAX_BYTES = Math.max(1, SystemPropertyUtil.getInt(
152 "io.netty.allocator.threadLocalChunkCacheMaxBytes", 8 * 1024 * 1024));
153
154
155
156
157
158
159
160 static final int THREAD_LOCAL_CACHE_MIN_BYTES = Math.min(THREAD_LOCAL_CACHE_MAX_BYTES,
161 Math.max(1, SystemPropertyUtil.getInt(
162 "io.netty.allocator.threadLocalChunkCacheMinBytes",
163 THREAD_LOCAL_CACHE_MAX_BYTES / 2)));
164
165
166
167
168
169 private static final int MAGAZINE_BUFFER_QUEUE_CAPACITY = SystemPropertyUtil.getInt(
170 "io.netty.allocator.magazineBufferQueueCapacity", 1024);
171
172
173
174
175
176
177
178
179
180
181
182
183
184 private static final int[] SIZE_CLASSES = {
185 32,
186 64,
187 128,
188 256,
189 512,
190 640,
191 1024,
192 1152,
193 2048,
194 2304,
195 4096,
196 4352,
197 8192,
198 8704,
199 16384,
200 16896,
201 };
202
203 private static final int SIZE_CLASSES_COUNT = SIZE_CLASSES.length;
204 private static final byte[] SIZE_INDEXES = new byte[SIZE_CLASSES[SIZE_CLASSES_COUNT - 1] / 32 + 1];
205
206 static {
207 if (MAGAZINE_BUFFER_QUEUE_CAPACITY < 2) {
208 throw new IllegalArgumentException("MAGAZINE_BUFFER_QUEUE_CAPACITY: " + MAGAZINE_BUFFER_QUEUE_CAPACITY
209 + " (expected: >= " + 2 + ')');
210 }
211 int lastIndex = 0;
212 for (int i = 0; i < SIZE_CLASSES_COUNT; i++) {
213 int sizeClass = SIZE_CLASSES[i];
214
215 assert (sizeClass & 31) == 0 : "Size class must be a multiple of 32";
216 int sizeIndex = sizeIndexOf(sizeClass);
217 Arrays.fill(SIZE_INDEXES, lastIndex + 1, sizeIndex + 1, (byte) i);
218 lastIndex = sizeIndex;
219 }
220 }
221
222 private final ChunkAllocator chunkAllocator;
223 private final ChunkRegistry chunkRegistry;
224 private final MagazineGroup[] sizeClassedMagazineGroups;
225 private final MagazineGroup largeBufferMagazineGroup;
226 private final FastThreadLocal<MagazineGroup[]> threadLocalGroup;
227
228 AdaptivePoolingAllocator(ChunkAllocator chunkAllocator, final boolean useCacheForNonEventLoopThreads) {
229 this.chunkAllocator = ObjectUtil.checkNotNull(chunkAllocator, "chunkAllocator");
230 chunkRegistry = new ChunkRegistry();
231 sizeClassedMagazineGroups = createMagazineGroupSizeClasses(this, false);
232 largeBufferMagazineGroup = new MagazineGroup(
233 this, chunkAllocator, new BuddyChunkManagementStrategy(), false);
234
235 boolean disableThreadLocalGroups = IS_LOW_MEM && DISABLE_THREAD_LOCAL_MAGAZINES_ON_LOW_MEM;
236 threadLocalGroup = disableThreadLocalGroups ? null : new FastThreadLocal<MagazineGroup[]>() {
237 @Override
238 protected MagazineGroup[] initialValue() {
239 if (useCacheForNonEventLoopThreads || ThreadExecutorMap.currentExecutor() != null) {
240 return createMagazineGroupSizeClasses(AdaptivePoolingAllocator.this, true);
241 }
242 return null;
243 }
244
245 @Override
246 protected void onRemoval(final MagazineGroup[] groups) throws Exception {
247 if (groups != null) {
248 for (MagazineGroup group : groups) {
249 group.free();
250 }
251 }
252 }
253 };
254 }
255
256 private static MagazineGroup[] createMagazineGroupSizeClasses(
257 AdaptivePoolingAllocator allocator, boolean isThreadLocal) {
258 MagazineGroup[] groups = new MagazineGroup[SIZE_CLASSES.length];
259 for (int i = 0; i < SIZE_CLASSES.length; i++) {
260 int segmentSize = SIZE_CLASSES[i];
261 groups[i] = new MagazineGroup(allocator, allocator.chunkAllocator,
262 new SizeClassChunkManagementStrategy(segmentSize), isThreadLocal);
263 }
264 return groups;
265 }
266
267 @Override
268 public ByteBuf allocate(int size, int maxCapacity) {
269 return allocate(size, maxCapacity, Thread.currentThread(), null);
270 }
271
272 private AdaptiveByteBuf allocate(int size, int maxCapacity, Thread currentThread, AdaptiveByteBuf buf) {
273 AdaptiveByteBuf allocated = null;
274 if (size <= MAX_POOLED_BUF_SIZE) {
275 final int index = sizeClassIndexOf(size);
276 MagazineGroup[] magazineGroups;
277 if (!FastThreadLocalThread.willCleanupFastThreadLocals(Thread.currentThread()) ||
278 IS_LOW_MEM ||
279 (magazineGroups = threadLocalGroup.get()) == null) {
280 magazineGroups = sizeClassedMagazineGroups;
281 }
282 if (index < magazineGroups.length) {
283 allocated = magazineGroups[index].allocate(size, maxCapacity, currentThread, buf);
284 } else if (!IS_LOW_MEM) {
285 allocated = largeBufferMagazineGroup.allocate(size, maxCapacity, currentThread, buf);
286 }
287 }
288 if (allocated == null) {
289 allocated = allocateFallback(size, maxCapacity, currentThread, buf);
290 }
291 return allocated;
292 }
293
294 private static int sizeIndexOf(final int size) {
295
296 return size + 31 >> 5;
297 }
298
299 static int sizeClassIndexOf(int size) {
300 int sizeIndex = sizeIndexOf(size);
301 if (sizeIndex < SIZE_INDEXES.length) {
302 return SIZE_INDEXES[sizeIndex];
303 }
304 return SIZE_CLASSES_COUNT;
305 }
306
307 static int[] getSizeClasses() {
308 return SIZE_CLASSES.clone();
309 }
310
311 private AdaptiveByteBuf allocateFallback(int size, int maxCapacity, Thread currentThread, AdaptiveByteBuf buf) {
312
313 Magazine magazine;
314 if (buf != null) {
315 Chunk chunk = buf.chunk;
316 if (chunk == null || chunk == Magazine.MAGAZINE_FREED || (magazine = chunk.currentMagazine()) == null) {
317 magazine = getFallbackMagazine(currentThread);
318 }
319 } else {
320 magazine = getFallbackMagazine(currentThread);
321 buf = magazine.newBuffer();
322 }
323
324 AbstractByteBuf innerChunk = chunkAllocator.allocate(size, maxCapacity);
325 Chunk chunk = new Chunk(innerChunk, magazine);
326 chunkRegistry.add(chunk);
327 try {
328 boolean success = chunk.readInitInto(buf, size, size, maxCapacity);
329 assert success : "Failed to initialize ByteBuf with dedicated chunk";
330 } finally {
331
332
333
334 chunk.release();
335 }
336 return buf;
337 }
338
339 private Magazine getFallbackMagazine(Thread currentThread) {
340 Magazine[] mags = largeBufferMagazineGroup.magazines;
341 return mags[(int) currentThread.getId() & mags.length - 1];
342 }
343
344
345
346
347 void reallocate(int size, int maxCapacity, AdaptiveByteBuf into) {
348 AdaptiveByteBuf result = allocate(size, maxCapacity, Thread.currentThread(), into);
349 assert result == into : "Re-allocation created separate buffer instance";
350 }
351
352 @Override
353 public long usedMemory() {
354 return chunkRegistry.totalCapacity();
355 }
356
357
358
359
360 @SuppressWarnings({"FinalizeDeclaration", "deprecation"})
361 @Override
362 protected void finalize() throws Throwable {
363 try {
364 free();
365 } finally {
366 super.finalize();
367 }
368 }
369
370 private void free() {
371 largeBufferMagazineGroup.free();
372 }
373
374 @SuppressJava6Requirement(reason = "Guarded by version check")
375 private static final class MagazineGroup {
376 private final AdaptivePoolingAllocator allocator;
377 private final ChunkAllocator chunkAllocator;
378 private final ChunkManagementStrategy chunkManagementStrategy;
379 private final ChunkCache chunkCache;
380 private final StampedLock magazineExpandLock;
381 private final Magazine threadLocalMagazine;
382 private Thread ownerThread;
383 private volatile Magazine[] magazines;
384 private volatile boolean freed;
385
386 MagazineGroup(AdaptivePoolingAllocator allocator,
387 ChunkAllocator chunkAllocator,
388 ChunkManagementStrategy chunkManagementStrategy,
389 boolean isThreadLocal) {
390 this.allocator = allocator;
391 this.chunkAllocator = chunkAllocator;
392 this.chunkManagementStrategy = chunkManagementStrategy;
393 chunkCache = chunkManagementStrategy.createChunkCache(isThreadLocal);
394 if (isThreadLocal) {
395 ownerThread = Thread.currentThread();
396 magazineExpandLock = null;
397 threadLocalMagazine = new Magazine(this, false, chunkManagementStrategy.createController(this));
398 } else {
399 ownerThread = null;
400 magazineExpandLock = new StampedLock();
401 threadLocalMagazine = null;
402 Magazine[] mags = new Magazine[INITIAL_MAGAZINES];
403 for (int i = 0; i < mags.length; i++) {
404 mags[i] = new Magazine(this, true, chunkManagementStrategy.createController(this));
405 }
406 magazines = mags;
407 }
408 }
409
410 public AdaptiveByteBuf allocate(int size, int maxCapacity, Thread currentThread, AdaptiveByteBuf buf) {
411 boolean reallocate = buf != null;
412
413
414 Magazine tlMag = threadLocalMagazine;
415 if (tlMag != null) {
416 if (buf == null) {
417 buf = tlMag.newBuffer();
418 }
419 boolean allocated = tlMag.tryAllocate(size, maxCapacity, buf, reallocate);
420 assert allocated : "Allocation of threadLocalMagazine must always succeed";
421 return buf;
422 }
423
424
425 long threadId = currentThread.getId();
426 Magazine[] mags;
427 int expansions = 0;
428 do {
429 mags = magazines;
430 int mask = mags.length - 1;
431 int index = (int) (threadId & mask);
432 for (int i = 0, m = mags.length << 1; i < m; i++) {
433 Magazine mag = mags[index + i & mask];
434 if (buf == null) {
435 buf = mag.newBuffer();
436 }
437 if (mag.tryAllocate(size, maxCapacity, buf, reallocate)) {
438
439 return buf;
440 }
441 }
442 expansions++;
443 } while (expansions <= EXPANSION_ATTEMPTS && tryExpandMagazines(mags.length));
444
445
446 if (!reallocate && buf != null) {
447 buf.release();
448 }
449 return null;
450 }
451
452 private boolean tryExpandMagazines(int currentLength) {
453 if (currentLength >= MAX_STRIPES) {
454 return true;
455 }
456 final Magazine[] mags;
457 long writeLock = magazineExpandLock.tryWriteLock();
458 if (writeLock != 0) {
459 try {
460 mags = magazines;
461 if (mags.length >= MAX_STRIPES || mags.length > currentLength || freed) {
462 return true;
463 }
464 Magazine[] expanded = new Magazine[mags.length * 2];
465 for (int i = 0, l = expanded.length; i < l; i++) {
466 expanded[i] = new Magazine(this, true, chunkManagementStrategy.createController(this));
467 }
468 magazines = expanded;
469 } finally {
470 magazineExpandLock.unlockWrite(writeLock);
471 }
472 for (Magazine magazine : mags) {
473 magazine.free();
474 }
475 }
476 return true;
477 }
478
479 Chunk pollChunk(int size) {
480 return chunkCache.pollChunk(size);
481 }
482
483 boolean offerChunk(Chunk chunk) {
484 if (freed) {
485 return false;
486 }
487
488 if (chunk.hasUnprocessedFreelistEntries()) {
489 chunk.processFreelistEntries();
490 }
491 boolean isAdded = chunkCache.offerChunk(chunk);
492
493 if (freed && isAdded) {
494
495 freeChunkReuseQueue(ownerThread);
496 }
497 return isAdded;
498 }
499
500 private void free() {
501 freed = true;
502 Thread ownerThread = this.ownerThread;
503 if (threadLocalMagazine != null) {
504 this.ownerThread = null;
505 threadLocalMagazine.free();
506 } else {
507 long stamp = magazineExpandLock.writeLock();
508 try {
509 Magazine[] mags = magazines;
510 for (Magazine magazine : mags) {
511 magazine.free();
512 }
513 } finally {
514 magazineExpandLock.unlockWrite(stamp);
515 }
516 }
517 freeChunkReuseQueue(ownerThread);
518 }
519
520 private void freeChunkReuseQueue(Thread ownerThread) {
521 if (ownerThread != null && chunkCache instanceof ThreadLocalSizeClassedChunkCache) {
522 ThreadLocalSizeClassedChunkCache tlCache = (ThreadLocalSizeClassedChunkCache) chunkCache;
523 int mask = tlCache.chunks.length - 1;
524 for (int i = 0; i < tlCache.count; i++) {
525 SizeClassedChunk chunk = tlCache.chunks[(tlCache.head + i) & mask];
526 assert ownerThread == chunk.ownerThread;
527 chunk.ownerThread = null;
528 }
529 }
530 chunkCache.free();
531 }
532 }
533
534 interface ChunkCache {
535 Chunk pollChunk(int size);
536
537 boolean offerChunk(Chunk chunk);
538
539 void free();
540
541 boolean isEmpty();
542 }
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572 abstract static class SizeClassedChunkCache implements ChunkCache {
573 static SizeClassedChunkCache create(boolean isThreadLocal, int chunkSize) {
574 return isThreadLocal ? new ThreadLocalSizeClassedChunkCache(chunkSize) :
575 new SharedSizeClassedChunkCache();
576 }
577
578 @Override
579 public abstract SizeClassedChunk pollChunk(int size);
580
581
582 abstract SizeClassedChunk forcePurge();
583 }
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674 static final class ThreadLocalSizeClassedChunkCache extends SizeClassedChunkCache {
675 SizeClassedChunk[] chunks;
676 int head;
677 int tail;
678 int count;
679 int notEmptyCount;
680 private long purgeBudget;
681 final int maxCachedChunks;
682 final int purgeRetentionFloor;
683
684 ThreadLocalSizeClassedChunkCache(int chunkSize) {
685 chunks = new SizeClassedChunk[8];
686 purgeBudget = CHUNK_PURGE_POLLS_THREAD_LOCAL;
687 maxCachedChunks = Math.max(1, THREAD_LOCAL_CACHE_MAX_BYTES / chunkSize);
688 purgeRetentionFloor = Math.min(maxCachedChunks,
689 Math.max(1, THREAD_LOCAL_CACHE_MIN_BYTES / chunkSize));
690 }
691
692 @Override
693 SizeClassedChunk forcePurge() {
694 purgeBudget = 1;
695 return pollChunk(0);
696 }
697
698 @Override
699 public SizeClassedChunk pollChunk(int size) {
700 if (--purgeBudget == 0) {
701 runPurgeScan();
702 }
703 return scanForCapacity();
704 }
705
706 private SizeClassedChunk scanForCapacity() {
707 if (notEmptyCount > 0) {
708 SizeClassedChunk chunk = chunks[head];
709 assert chunk.hasRemainingCapacity();
710 chunk.purgeEpoch = 0;
711 chunks[head] = null;
712 head = (head + 1) & (chunks.length - 1);
713 count--;
714 notEmptyCount--;
715 return chunk;
716 }
717 return scanForCapacityFallback();
718 }
719
720 private SizeClassedChunk scanForCapacityFallback() {
721 int mask = chunks.length - 1;
722 int emptyCount = count - notEmptyCount;
723 int pos = (head + notEmptyCount) & mask;
724 for (int i = 0; i < emptyCount; i++) {
725 SizeClassedChunk chunk = chunks[pos];
726 if (chunk.hasRemainingCapacity()) {
727 chunk.purgeEpoch = 0;
728 int lastIdx = (tail - 1) & mask;
729 chunks[pos] = chunks[lastIdx];
730 chunks[lastIdx] = null;
731 tail = lastIdx;
732 count--;
733 return chunk;
734 }
735 pos = (pos + 1) & mask;
736 }
737 return null;
738 }
739
740 private void runPurgeScan() {
741 int mask = chunks.length - 1;
742 int kept = 0;
743 int survivors = count;
744 for (int i = 0; i < count; i++) {
745 int readIdx = (head + i) & mask;
746 SizeClassedChunk chunk = chunks[readIdx];
747 if (chunk.purgeEpoch > 0) {
748 assert chunk.hasFullCapacity();
749 chunk.purgeEpoch++;
750 if (chunk.purgeEpoch > CHUNK_PURGE_THRESHOLD && survivors > purgeRetentionFloor) {
751 chunk.markToDeallocate();
752 chunks[readIdx] = null;
753 survivors--;
754 continue;
755 }
756 } else if (chunk.hasFullCapacity()) {
757 chunk.purgeEpoch = 1;
758 }
759 int writeIdx = (head + kept) & mask;
760 if (writeIdx != readIdx) {
761 chunks[writeIdx] = chunk;
762 chunks[readIdx] = null;
763 }
764 kept++;
765 }
766 tail = (head + kept) & mask;
767 count = kept;
768 partition(kept);
769 purgeBudget = CHUNK_PURGE_POLLS_THREAD_LOCAL;
770 }
771
772 private void partition(int size) {
773 int mask = chunks.length - 1;
774
775 int lo = 0;
776 int hi = size - 1;
777 while (lo <= hi) {
778 int loIdx = (head + lo) & mask;
779 if (chunks[loIdx].hasRemainingCapacity()) {
780 lo++;
781 } else {
782 int hiIdx = (head + hi) & mask;
783 SizeClassedChunk tmp = chunks[loIdx];
784 chunks[loIdx] = chunks[hiIdx];
785 chunks[hiIdx] = tmp;
786 hi--;
787 }
788 }
789 notEmptyCount = lo;
790
791
792
793
794
795
796
797
798
799
800
801 int elo = 0;
802 int emid = 0;
803 int ehi = lo - 1;
804 while (emid <= ehi) {
805 int emidIdx = (head + emid) & mask;
806 SizeClassedChunk c = chunks[emidIdx];
807 if (c.purgeEpoch == 0) {
808 if (elo != emid) {
809 int eloIdx = (head + elo) & mask;
810 chunks[emidIdx] = chunks[eloIdx];
811 chunks[eloIdx] = c;
812 }
813 elo++;
814 emid++;
815 } else if (c.purgeEpoch < CHUNK_PURGE_THRESHOLD) {
816 emid++;
817 } else {
818 int ehiIdx = (head + ehi) & mask;
819 chunks[emidIdx] = chunks[ehiIdx];
820 chunks[ehiIdx] = c;
821 ehi--;
822 }
823 }
824 }
825
826 @Override
827 public boolean offerChunk(Chunk chunk) {
828 if (count >= maxCachedChunks) {
829 return false;
830 }
831 if (count == chunks.length) {
832 SizeClassedChunk[] newChunks = new SizeClassedChunk[chunks.length * 2];
833 for (int i = 0; i < count; i++) {
834 newChunks[i] = chunks[(head + i) & (chunks.length - 1)];
835 }
836 chunks = newChunks;
837 head = 0;
838 tail = count;
839 }
840 chunks[tail] = (SizeClassedChunk) chunk;
841 tail = (tail + 1) & (chunks.length - 1);
842 count++;
843 return true;
844 }
845
846 @Override
847 public String toString() {
848 int mask = chunks.length - 1;
849 StringBuilder sb = new StringBuilder();
850 sb.append("ThreadLocalCache[head=").append(head)
851 .append(", tail=").append(tail)
852 .append(", count=").append(count)
853 .append(", notEmpty=").append(notEmptyCount)
854 .append(", length=").append(chunks.length)
855 .append("]\n ");
856 for (int i = 0; i < count; i++) {
857 if (i > 0) {
858 sb.append(", ");
859 }
860 if (i == notEmptyCount) {
861 sb.append("| ");
862 }
863 SizeClassedChunk c = chunks[(head + i) & mask];
864 String region = i < notEmptyCount ? "notEmpty" : "empty";
865 String actual = c == null ? "null" :
866 c.hasRemainingCapacity() ? "hasCap" : "noCap";
867 sb.append('[').append(region).append(':').append(actual)
868 .append(",ep=").append(c == null ? -1 : c.purgeEpoch).append(']');
869 }
870 return sb.toString();
871 }
872
873 @Override
874 public void free() {
875 int mask = chunks.length - 1;
876 for (int i = 0; i < count; i++) {
877 int idx = (head + i) & mask;
878 chunks[idx].markToDeallocate();
879 chunks[idx] = null;
880 }
881 head = 0;
882 tail = 0;
883 count = 0;
884 notEmptyCount = 0;
885 }
886
887 @Override
888 public boolean isEmpty() {
889 return count == 0;
890 }
891 }
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925 static final class SharedSizeClassedChunkCache extends SizeClassedChunkCache {
926
927
928 private static final int SHARED_CACHE_CAPACITY = Math.max(16, CHUNK_REUSE_QUEUE * 2);
929 private final Queue<SizeClassedChunk> queue;
930 private final AtomicLong purgeBudget;
931 private final ArrayList<SizeClassedChunk> deferredBuffer = new ArrayList<SizeClassedChunk>();
932 private long purgeGeneration;
933 private final AtomicLong scanGeneration = new AtomicLong();
934
935 SharedSizeClassedChunkCache() {
936 queue = PlatformDependent.newFixedMpmcQueue(SHARED_CACHE_CAPACITY);
937 purgeBudget = new AtomicLong(CHUNK_PURGE_POLLS_SHARED);
938 }
939
940 @Override
941 SizeClassedChunk forcePurge() {
942 purgeBudget.set(1);
943 return pollChunk(0);
944 }
945
946 @Override
947 public SizeClassedChunk pollChunk(int size) {
948 long budget = purgeBudget.decrementAndGet();
949 if (budget == 0) {
950 runPurgeScan();
951 }
952 return scanForCapacity();
953 }
954
955 private SizeClassedChunk scanForCapacity() {
956 SizeClassedChunk first = queue.poll();
957 if (first == null) {
958 return null;
959 }
960 if (first.purgeEpoch == 0 && first.hasRemainingCapacity()) {
961 return first;
962 }
963 long generation = scanGeneration.incrementAndGet();
964 first.lastScanGeneration = generation;
965 if (first.hasRemainingCapacity()) {
966 return scanForCapacitySlow(generation, first);
967 }
968 offerOrDeallocate(first);
969 return scanForCapacitySlow(generation, null);
970 }
971
972 private SizeClassedChunk scanForCapacitySlow(long generation, SizeClassedChunk fallback) {
973 SizeClassedChunk chunk;
974 while ((chunk = queue.poll()) != null) {
975 if (chunk.lastScanGeneration >= generation) {
976 offerOrDeallocate(chunk);
977 break;
978 }
979 if (chunk.hasRemainingCapacity()) {
980 if (chunk.purgeEpoch == 0) {
981 if (fallback != null) {
982 offerOrDeallocate(fallback);
983 }
984 return chunk;
985 }
986 if (fallback == null) {
987 fallback = chunk;
988 continue;
989 }
990 }
991 chunk.lastScanGeneration = generation;
992 offerOrDeallocate(chunk);
993 }
994 if (fallback != null) {
995 fallback.purgeEpoch = 0;
996 return fallback;
997 }
998 return null;
999 }
1000
1001 private boolean offerOrDeallocate(SizeClassedChunk chunk) {
1002 if (!queue.offer(chunk)) {
1003 chunk.markToDeallocate();
1004 return false;
1005 }
1006 return true;
1007 }
1008
1009 private boolean offerOrDeallocate(SizeClassedChunk chunk, long generation) {
1010 chunk.lastPurgeGeneration = generation;
1011 return offerOrDeallocate(chunk);
1012 }
1013
1014 private void runPurgeScan() {
1015 long generation = ++purgeGeneration;
1016 int retained = 0;
1017 ArrayList<SizeClassedChunk> deferred = deferredBuffer;
1018 SizeClassedChunk chunk;
1019 while ((chunk = queue.poll()) != null) {
1020 if (chunk.lastPurgeGeneration == generation) {
1021 offerOrDeallocate(chunk, generation);
1022 break;
1023 }
1024 retained++;
1025 if (chunk.hasFullCapacity()) {
1026 chunk.purgeEpoch++;
1027 if (chunk.purgeEpoch > CHUNK_PURGE_THRESHOLD) {
1028 deferred.add(chunk);
1029 continue;
1030 }
1031 } else {
1032 chunk.purgeEpoch = 0;
1033 }
1034 int remaining = chunk.remainingCapacity();
1035 if (remaining > 0) {
1036 if (!offerOrDeallocate(chunk, generation)) {
1037 retained--;
1038 }
1039 } else {
1040 deferred.add(chunk);
1041 }
1042 }
1043 for (int i = 0, size = deferred.size(); i < size; i++) {
1044 chunk = deferred.get(i);
1045 if (chunk.purgeEpoch > CHUNK_PURGE_THRESHOLD && retained > CHUNK_REUSE_QUEUE) {
1046 chunk.markToDeallocate();
1047 retained--;
1048 } else {
1049 if (!offerOrDeallocate(chunk, generation)) {
1050 retained--;
1051 }
1052 }
1053 }
1054 deferred.clear();
1055 purgeBudget.lazySet(CHUNK_PURGE_POLLS_SHARED);
1056 }
1057
1058 @Override
1059 public boolean offerChunk(Chunk chunk) {
1060 return queue.offer((SizeClassedChunk) chunk);
1061 }
1062
1063 @Override
1064 public void free() {
1065 SizeClassedChunk chunk;
1066 while ((chunk = queue.poll()) != null) {
1067 chunk.markToDeallocate();
1068 }
1069 }
1070
1071 @Override
1072 public boolean isEmpty() {
1073 return queue.isEmpty();
1074 }
1075 }
1076
1077 private static final class ConcurrentSkipListChunkCache implements ChunkCache {
1078 private final ConcurrentSkipListIntObjMultimap<Chunk> chunks;
1079
1080 private ConcurrentSkipListChunkCache() {
1081 chunks = new ConcurrentSkipListIntObjMultimap<Chunk>(-1);
1082 }
1083
1084 @Override
1085 public Chunk pollChunk(int size) {
1086 if (chunks.isEmpty()) {
1087 return null;
1088 }
1089 IntEntry<Chunk> entry = chunks.pollCeilingEntry(size);
1090 if (entry != null) {
1091 Chunk chunk = entry.getValue();
1092 if (chunk.hasUnprocessedFreelistEntries()) {
1093 chunk.processFreelistEntries();
1094 }
1095 return chunk;
1096 }
1097
1098 Chunk bestChunk = null;
1099 int bestRemainingCapacity = 0;
1100 Iterator<IntEntry<Chunk>> itr = chunks.iterator();
1101 while (itr.hasNext()) {
1102 entry = itr.next();
1103 final Chunk chunk;
1104 if (entry != null && (chunk = entry.getValue()).hasUnprocessedFreelistEntries()) {
1105 if (!chunks.remove(entry.getKey(), entry.getValue())) {
1106 continue;
1107 }
1108 chunk.processFreelistEntries();
1109 int remainingCapacity = chunk.remainingCapacity();
1110 if (remainingCapacity >= size &&
1111 (bestChunk == null || remainingCapacity > bestRemainingCapacity)) {
1112 if (bestChunk != null) {
1113 chunks.put(bestRemainingCapacity, bestChunk);
1114 }
1115 bestChunk = chunk;
1116 bestRemainingCapacity = remainingCapacity;
1117 } else {
1118 chunks.put(remainingCapacity, chunk);
1119 }
1120 }
1121 }
1122
1123 return bestChunk;
1124 }
1125
1126 @Override
1127 public boolean offerChunk(Chunk chunk) {
1128 chunks.put(chunk.remainingCapacity(), chunk);
1129
1130 int size = chunks.size();
1131 while (size > CHUNK_REUSE_QUEUE) {
1132 int key = -1;
1133 Chunk toDeallocate = null;
1134 for (IntEntry<Chunk> entry : chunks) {
1135 Chunk candidate = entry.getValue();
1136 if (candidate != null) {
1137 if (toDeallocate == null) {
1138 toDeallocate = candidate;
1139 key = entry.getKey();
1140 } else {
1141 int candidateRefCnt = candidate.refCnt();
1142 int toDeallocateRefCnt = toDeallocate.refCnt();
1143 if (candidateRefCnt < toDeallocateRefCnt ||
1144 candidateRefCnt == toDeallocateRefCnt &&
1145 candidate.capacity() < toDeallocate.capacity()) {
1146 toDeallocate = candidate;
1147 key = entry.getKey();
1148 }
1149 }
1150 }
1151 }
1152 if (toDeallocate == null) {
1153 break;
1154 }
1155 if (chunks.remove(key, toDeallocate)) {
1156 toDeallocate.markToDeallocate();
1157 }
1158 size = chunks.size();
1159 }
1160 return true;
1161 }
1162
1163 @Override
1164 public void free() {
1165 for (IntEntry<Chunk> entry : chunks) {
1166 Chunk chunk = entry.getValue();
1167 if (chunk != null && chunks.remove(entry.getKey(), chunk)) {
1168 chunk.markToDeallocate();
1169 }
1170 }
1171 }
1172
1173 @Override
1174 public boolean isEmpty() {
1175 return chunks.isEmpty();
1176 }
1177 }
1178
1179 private interface ChunkManagementStrategy {
1180 ChunkController createController(MagazineGroup group);
1181
1182 ChunkCache createChunkCache(boolean isThreadLocal);
1183 }
1184
1185 private interface ChunkController {
1186
1187
1188
1189 int computeBufferCapacity(int requestedSize, int maxCapacity, boolean isReallocation);
1190
1191
1192
1193
1194 Chunk newChunkAllocation(int promptingSize, Magazine magazine);
1195 }
1196
1197 private static final class SizeClassChunkManagementStrategy implements ChunkManagementStrategy {
1198
1199
1200
1201 private static final int MIN_SEGMENTS_PER_CHUNK = 32;
1202 private final int segmentSize;
1203 private final int chunkSize;
1204
1205 private SizeClassChunkManagementStrategy(int segmentSize) {
1206 this.segmentSize = ObjectUtil.checkPositive(segmentSize, "segmentSize");
1207 chunkSize = Math.max(MIN_CHUNK_SIZE, segmentSize * MIN_SEGMENTS_PER_CHUNK);
1208 }
1209
1210 @Override
1211 public ChunkController createController(MagazineGroup group) {
1212 return new SizeClassChunkController(group, segmentSize, chunkSize);
1213 }
1214
1215 @Override
1216 public ChunkCache createChunkCache(boolean isThreadLocal) {
1217 return SizeClassedChunkCache.create(isThreadLocal, chunkSize);
1218 }
1219 }
1220
1221 private static final class SizeClassChunkController implements ChunkController {
1222
1223 private final ChunkAllocator chunkAllocator;
1224 private final int segmentSize;
1225 private final int chunkSize;
1226 private final ChunkRegistry chunkRegistry;
1227
1228 private SizeClassChunkController(MagazineGroup group, int segmentSize, int chunkSize) {
1229 chunkAllocator = group.chunkAllocator;
1230 this.segmentSize = segmentSize;
1231 this.chunkSize = chunkSize;
1232 chunkRegistry = group.allocator.chunkRegistry;
1233 }
1234
1235 private MpscAtomicIntegerArrayQueue createEmptyFreeList() {
1236 return new MpscAtomicIntegerArrayQueue(chunkSize / segmentSize, SizeClassedChunk.FREE_LIST_EMPTY);
1237 }
1238
1239 private MpscAtomicIntegerArrayQueue createFreeList() {
1240 final int segmentsCount = chunkSize / segmentSize;
1241 final MpscAtomicIntegerArrayQueue freeList = new MpscAtomicIntegerArrayQueue(
1242 segmentsCount, SizeClassedChunk.FREE_LIST_EMPTY);
1243 int segmentOffset = 0;
1244 for (int i = 0; i < segmentsCount; i++) {
1245 freeList.offer(segmentOffset);
1246 segmentOffset += segmentSize;
1247 }
1248 return freeList;
1249 }
1250
1251 private IntStack createLocalFreeList() {
1252 final int segmentsCount = chunkSize / segmentSize;
1253 int segmentOffset = chunkSize;
1254 int[] offsets = new int[segmentsCount];
1255 for (int i = 0; i < segmentsCount; i++) {
1256 segmentOffset -= segmentSize;
1257 offsets[i] = segmentOffset;
1258 }
1259 return new IntStack(offsets);
1260 }
1261
1262 @Override
1263 public int computeBufferCapacity(
1264 int requestedSize, int maxCapacity, boolean isReallocation) {
1265 return Math.min(segmentSize, maxCapacity);
1266 }
1267
1268 @Override
1269 public Chunk newChunkAllocation(int promptingSize, Magazine magazine) {
1270 AbstractByteBuf chunkBuffer = chunkAllocator.allocate(chunkSize, chunkSize);
1271 assert chunkBuffer.capacity() == chunkSize;
1272 SizeClassedChunk chunk = new SizeClassedChunk(chunkBuffer, magazine, this);
1273 chunkRegistry.add(chunk);
1274 return chunk;
1275 }
1276 }
1277
1278 private static final class BuddyChunkManagementStrategy implements ChunkManagementStrategy {
1279 private final AtomicInteger maxChunkSize = new AtomicInteger();
1280
1281 @Override
1282 public ChunkController createController(MagazineGroup group) {
1283 return new BuddyChunkController(group, maxChunkSize);
1284 }
1285
1286 @Override
1287 public ChunkCache createChunkCache(boolean isThreadLocal) {
1288 return new ConcurrentSkipListChunkCache();
1289 }
1290 }
1291
1292 private static final class BuddyChunkController implements ChunkController {
1293 private final ChunkAllocator chunkAllocator;
1294 private final ChunkRegistry chunkRegistry;
1295 private final AtomicInteger maxChunkSize;
1296
1297 BuddyChunkController(MagazineGroup group, AtomicInteger maxChunkSize) {
1298 chunkAllocator = group.chunkAllocator;
1299 chunkRegistry = group.allocator.chunkRegistry;
1300 this.maxChunkSize = maxChunkSize;
1301 }
1302
1303 @Override
1304 public int computeBufferCapacity(int requestedSize, int maxCapacity, boolean isReallocation) {
1305 return MathUtil.safeFindNextPositivePowerOfTwo(requestedSize);
1306 }
1307
1308 @Override
1309 public Chunk newChunkAllocation(int promptingSize, Magazine magazine) {
1310 int maxChunkSize = this.maxChunkSize.get();
1311 int proposedChunkSize = MathUtil.safeFindNextPositivePowerOfTwo(BUFS_PER_CHUNK * promptingSize);
1312 int chunkSize = Math.min(MAX_CHUNK_SIZE, Math.max(maxChunkSize, proposedChunkSize));
1313 if (chunkSize > maxChunkSize) {
1314
1315 this.maxChunkSize.set(chunkSize);
1316 }
1317 BuddyChunk chunk = new BuddyChunk(chunkAllocator.allocate(chunkSize, chunkSize), magazine);
1318 chunkRegistry.add(chunk);
1319 return chunk;
1320 }
1321 }
1322
1323 @SuppressJava6Requirement(reason = "Guarded by version check")
1324 private static final class Magazine {
1325 private static final AtomicReferenceFieldUpdater<Magazine, Chunk> NEXT_IN_LINE;
1326
1327 static {
1328 NEXT_IN_LINE = AtomicReferenceFieldUpdater.newUpdater(Magazine.class, Chunk.class, "nextInLine");
1329 }
1330
1331 private static final Chunk MAGAZINE_FREED = new Chunk();
1332
1333 private static final class AdaptiveRecycler extends Recycler<AdaptiveByteBuf> {
1334
1335 private AdaptiveRecycler() {
1336 }
1337
1338 private AdaptiveRecycler(int maxCapacity) {
1339
1340 super(maxCapacity);
1341 }
1342
1343 @Override
1344 protected AdaptiveByteBuf newObject(final Handle<AdaptiveByteBuf> handle) {
1345 return new AdaptiveByteBuf((EnhancedHandle<AdaptiveByteBuf>) handle);
1346 }
1347
1348 public static AdaptiveRecycler threadLocal() {
1349 return new AdaptiveRecycler();
1350 }
1351
1352 public static AdaptiveRecycler sharedWith(int maxCapacity) {
1353 return new AdaptiveRecycler(maxCapacity);
1354 }
1355 }
1356
1357 private static final AdaptiveRecycler EVENT_LOOP_LOCAL_BUFFER_POOL = AdaptiveRecycler.threadLocal();
1358
1359 private Chunk current;
1360 @SuppressWarnings("unused")
1361 private volatile Chunk nextInLine;
1362 private final MagazineGroup group;
1363 private final ChunkController chunkController;
1364 private final StampedLock allocationLock;
1365 private final AdaptiveRecycler recycler;
1366
1367 Magazine(MagazineGroup group, boolean shareable, ChunkController chunkController) {
1368 this.group = group;
1369 this.chunkController = chunkController;
1370
1371 if (shareable) {
1372
1373 allocationLock = new StampedLock();
1374 recycler = AdaptiveRecycler.sharedWith(MAGAZINE_BUFFER_QUEUE_CAPACITY);
1375 } else {
1376 allocationLock = null;
1377 recycler = null;
1378 }
1379 }
1380
1381 public boolean tryAllocate(int size, int maxCapacity, AdaptiveByteBuf buf, boolean reallocate) {
1382 if (allocationLock == null) {
1383
1384 return allocate(size, maxCapacity, buf, reallocate);
1385 }
1386
1387
1388 long writeLock = allocationLock.tryWriteLock();
1389 if (writeLock != 0) {
1390 try {
1391 return allocate(size, maxCapacity, buf, reallocate);
1392 } finally {
1393 allocationLock.unlockWrite(writeLock);
1394 }
1395 }
1396 return allocateWithoutLock(size, maxCapacity, buf);
1397 }
1398
1399 private boolean allocateWithoutLock(int size, int maxCapacity, AdaptiveByteBuf buf) {
1400 Chunk curr = NEXT_IN_LINE.getAndSet(this, null);
1401 if (curr == MAGAZINE_FREED) {
1402
1403 restoreMagazineFreed();
1404 return false;
1405 }
1406 if (curr == null) {
1407 curr = group.pollChunk(size);
1408 if (curr == null) {
1409 return false;
1410 }
1411 curr.attachToMagazine(this);
1412 }
1413 boolean allocated = false;
1414 int remainingCapacity = curr.remainingCapacity();
1415 int startingCapacity = chunkController.computeBufferCapacity(
1416 size, maxCapacity, true );
1417 if (remainingCapacity >= size &&
1418 curr.readInitInto(buf, size, Math.min(remainingCapacity, startingCapacity), maxCapacity)) {
1419 allocated = true;
1420 remainingCapacity = curr.remainingCapacity();
1421 }
1422 try {
1423 if (remainingCapacity >= RETIRE_CAPACITY) {
1424 transferToNextInLineOrRelease(curr);
1425 curr = null;
1426 }
1427 } finally {
1428 if (curr != null) {
1429 curr.releaseFromMagazine();
1430 }
1431 }
1432 return allocated;
1433 }
1434
1435 private boolean allocate(int size, int maxCapacity, AdaptiveByteBuf buf, boolean reallocate) {
1436 int startingCapacity = chunkController.computeBufferCapacity(size, maxCapacity, reallocate);
1437 Chunk curr = current;
1438 if (curr != null) {
1439 boolean success = curr.readInitInto(buf, size, startingCapacity, maxCapacity);
1440 int remainingCapacity = curr.remainingCapacity();
1441 if (!success && remainingCapacity > 0) {
1442 current = null;
1443 transferToNextInLineOrRelease(curr);
1444 } else if (remainingCapacity == 0) {
1445 current = null;
1446 curr.releaseFromMagazine();
1447 }
1448 if (success) {
1449 return true;
1450 }
1451 }
1452
1453 assert current == null;
1454
1455
1456
1457
1458
1459
1460
1461
1462 curr = NEXT_IN_LINE.getAndSet(this, null);
1463 if (curr != null) {
1464 if (curr == MAGAZINE_FREED) {
1465
1466 restoreMagazineFreed();
1467 return false;
1468 }
1469
1470 int remainingCapacity = curr.remainingCapacity();
1471 if (remainingCapacity > startingCapacity &&
1472 curr.readInitInto(buf, size, startingCapacity, maxCapacity)) {
1473
1474 current = curr;
1475 return true;
1476 }
1477
1478 try {
1479 if (remainingCapacity >= size) {
1480
1481
1482 return curr.readInitInto(buf, size, remainingCapacity, maxCapacity);
1483 }
1484 } finally {
1485
1486
1487 curr.releaseFromMagazine();
1488 }
1489 }
1490
1491
1492 curr = group.pollChunk(size);
1493 if (curr == null) {
1494 curr = chunkController.newChunkAllocation(size, this);
1495 } else {
1496 curr.attachToMagazine(this);
1497
1498 int remainingCapacity = curr.remainingCapacity();
1499 if (remainingCapacity == 0 || remainingCapacity < size) {
1500
1501 if (remainingCapacity < RETIRE_CAPACITY) {
1502 curr.releaseFromMagazine();
1503 } else {
1504
1505
1506 transferToNextInLineOrRelease(curr);
1507 }
1508 curr = chunkController.newChunkAllocation(size, this);
1509 }
1510 }
1511
1512 current = curr;
1513 boolean success;
1514 try {
1515 int remainingCapacity = curr.remainingCapacity();
1516 assert remainingCapacity >= size;
1517 if (remainingCapacity > startingCapacity) {
1518 success = curr.readInitInto(buf, size, startingCapacity, maxCapacity);
1519 curr = null;
1520 } else {
1521 success = curr.readInitInto(buf, size, remainingCapacity, maxCapacity);
1522 }
1523 } finally {
1524 if (curr != null) {
1525
1526
1527 curr.releaseFromMagazine();
1528 current = null;
1529 }
1530 }
1531 return success;
1532 }
1533
1534 private void restoreMagazineFreed() {
1535 Chunk next = NEXT_IN_LINE.getAndSet(this, MAGAZINE_FREED);
1536 if (next != null && next != MAGAZINE_FREED) {
1537
1538 next.releaseFromMagazine();
1539 }
1540 }
1541
1542 private void transferToNextInLineOrRelease(Chunk chunk) {
1543 if (NEXT_IN_LINE.compareAndSet(this, null, chunk)) {
1544 return;
1545 }
1546
1547 Chunk nextChunk = NEXT_IN_LINE.get(this);
1548 if (nextChunk != null && nextChunk != MAGAZINE_FREED
1549 && chunk.remainingCapacity() > nextChunk.remainingCapacity()) {
1550 if (NEXT_IN_LINE.compareAndSet(this, nextChunk, chunk)) {
1551 nextChunk.releaseFromMagazine();
1552 return;
1553 }
1554 }
1555
1556
1557
1558
1559 chunk.releaseFromMagazine();
1560 }
1561
1562 void free() {
1563
1564 restoreMagazineFreed();
1565 long stamp = allocationLock != null ? allocationLock.writeLock() : 0;
1566 try {
1567 if (current != null) {
1568 current.releaseFromMagazine();
1569 current = null;
1570 }
1571 } finally {
1572 if (allocationLock != null) {
1573 allocationLock.unlockWrite(stamp);
1574 }
1575 }
1576 }
1577
1578 public AdaptiveByteBuf newBuffer() {
1579 AdaptiveRecycler recycler = this.recycler;
1580 AdaptiveByteBuf buf = recycler == null ? EVENT_LOOP_LOCAL_BUFFER_POOL.get() : recycler.get();
1581 buf.resetRefCnt();
1582 buf.discardMarks();
1583 return buf;
1584 }
1585
1586 boolean offerToQueue(Chunk chunk) {
1587 return group.offerChunk(chunk);
1588 }
1589 }
1590
1591 @SuppressJava6Requirement(reason = "Guarded by version check")
1592 private static final class ChunkRegistry {
1593 private final LongCounter totalCapacity = PlatformDependent.newLongCounter();
1594
1595 public long totalCapacity() {
1596 return totalCapacity.value();
1597 }
1598
1599 public void add(Chunk chunk) {
1600 totalCapacity.add(chunk.capacity());
1601 }
1602
1603 public void remove(Chunk chunk) {
1604 totalCapacity.add(-chunk.capacity());
1605 }
1606 }
1607
1608 static class Chunk implements ReferenceCounted {
1609 private static final long REFCNT_FIELD_OFFSET =
1610 ReferenceCountUpdater.getUnsafeOffset(Chunk.class, "refCnt");
1611 private static final AtomicIntegerFieldUpdater<Chunk> AIF_UPDATER =
1612 AtomicIntegerFieldUpdater.newUpdater(Chunk.class, "refCnt");
1613 protected final AbstractByteBuf delegate;
1614 protected Magazine magazine;
1615 private final AdaptivePoolingAllocator allocator;
1616 private final int capacity;
1617 protected int allocatedBytes;
1618
1619 private static final ReferenceCountUpdater<Chunk> updater =
1620 new ReferenceCountUpdater<Chunk>() {
1621 @Override
1622 protected AtomicIntegerFieldUpdater<Chunk> updater() {
1623 return AIF_UPDATER;
1624 }
1625 @Override
1626 protected long unsafeOffset() {
1627
1628
1629 return PlatformDependent.hasUnsafe() ? REFCNT_FIELD_OFFSET : -1;
1630 }
1631 };
1632
1633
1634 @SuppressWarnings({"unused", "FieldMayBeFinal"})
1635 private volatile int refCnt;
1636
1637 Chunk() {
1638
1639 delegate = null;
1640 magazine = null;
1641 allocator = null;
1642 capacity = 0;
1643 }
1644
1645 Chunk(AbstractByteBuf delegate, Magazine magazine) {
1646 this.delegate = delegate;
1647 capacity = delegate.capacity();
1648 updater.setInitialValue(this);
1649 attachToMagazine(magazine);
1650
1651
1652 allocator = magazine.group.allocator;
1653 }
1654
1655 Magazine currentMagazine() {
1656 return magazine;
1657 }
1658
1659 void detachFromMagazine() {
1660 if (magazine != null) {
1661 magazine = null;
1662 }
1663 }
1664
1665 void attachToMagazine(Magazine magazine) {
1666 assert this.magazine == null;
1667 this.magazine = magazine;
1668 }
1669
1670 @Override
1671 public Chunk touch(Object hint) {
1672 return this;
1673 }
1674
1675 @Override
1676 public int refCnt() {
1677 return updater.refCnt(this);
1678 }
1679
1680 @Override
1681 public Chunk retain() {
1682 return updater.retain(this);
1683 }
1684
1685 @Override
1686 public Chunk retain(int increment) {
1687 return updater.retain(this, increment);
1688 }
1689
1690 @Override
1691 public Chunk touch() {
1692 return this;
1693 }
1694
1695 @Override
1696 public boolean release() {
1697 if (updater.release(this)) {
1698 deallocate();
1699 return true;
1700 }
1701 return false;
1702 }
1703
1704 @Override
1705 public boolean release(int decrement) {
1706 if (updater.release(this, decrement)) {
1707 deallocate();
1708 return true;
1709 }
1710 return false;
1711 }
1712
1713
1714
1715
1716 void releaseFromMagazine() {
1717
1718
1719 Magazine mag = magazine;
1720 detachFromMagazine();
1721 if (!mag.offerToQueue(this)) {
1722 markToDeallocate();
1723 }
1724 }
1725
1726
1727
1728
1729 void releaseSegment(int ignoredSegmentId, int size) {
1730 release();
1731 }
1732
1733 void markToDeallocate() {
1734 release();
1735 }
1736
1737 protected void deallocate() {
1738 allocator.chunkRegistry.remove(this);
1739 delegate.release();
1740 }
1741
1742 public boolean readInitInto(AdaptiveByteBuf buf, int size, int startingCapacity, int maxCapacity) {
1743 int startIndex = allocatedBytes;
1744 allocatedBytes = startIndex + startingCapacity;
1745 Chunk chunk = this;
1746 chunk.retain();
1747 try {
1748 buf.init(delegate, chunk, 0, 0, startIndex, size, startingCapacity, maxCapacity);
1749 chunk = null;
1750 } finally {
1751 if (chunk != null) {
1752
1753
1754
1755 allocatedBytes = startIndex;
1756 chunk.release();
1757 }
1758 }
1759 return true;
1760 }
1761
1762 public int remainingCapacity() {
1763 return capacity - allocatedBytes;
1764 }
1765
1766 public boolean hasUnprocessedFreelistEntries() {
1767 return false;
1768 }
1769
1770 public void processFreelistEntries() {
1771 }
1772
1773 public int capacity() {
1774 return capacity;
1775 }
1776 }
1777
1778 private static final class IntStack {
1779
1780 private final int[] stack;
1781 private int top;
1782
1783 IntStack(int[] initialValues) {
1784 stack = initialValues;
1785 top = initialValues.length - 1;
1786 }
1787
1788 public boolean isEmpty() {
1789 return top == -1;
1790 }
1791
1792 public int pop() {
1793 final int last = stack[top];
1794 top--;
1795 return last;
1796 }
1797
1798 public void push(int value) {
1799 stack[top + 1] = value;
1800 top++;
1801 }
1802
1803 public int size() {
1804 return top + 1;
1805 }
1806 }
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827 static class SizeClassedChunk extends Chunk {
1828 private static final int FREE_LIST_EMPTY = -1;
1829 private static final int AVAILABLE = -1;
1830
1831
1832 private static final int DEALLOCATED = Integer.MIN_VALUE;
1833 private static final AtomicIntegerFieldUpdater<SizeClassedChunk> STATE =
1834 AtomicIntegerFieldUpdater.newUpdater(SizeClassedChunk.class, "state");
1835 private volatile int state;
1836 private final int segments;
1837 private final int segmentSize;
1838 private final MpscIntQueue externalFreeList;
1839 private final IntStack localFreeList;
1840 private Thread ownerThread;
1841 int purgeEpoch;
1842 long lastPurgeGeneration;
1843 long lastScanGeneration;
1844
1845 SizeClassedChunk(AbstractByteBuf delegate, Magazine magazine,
1846 SizeClassChunkController controller) {
1847 super(delegate, magazine);
1848 segmentSize = controller.segmentSize;
1849 segments = controller.chunkSize / segmentSize;
1850 STATE.lazySet(this, AVAILABLE);
1851 ownerThread = magazine.group.ownerThread;
1852 if (ownerThread == null) {
1853 externalFreeList = controller.createFreeList();
1854 localFreeList = null;
1855 } else {
1856 externalFreeList = controller.createEmptyFreeList();
1857 localFreeList = controller.createLocalFreeList();
1858 }
1859 }
1860
1861 @Override
1862 public boolean readInitInto(AdaptiveByteBuf buf, int size, int startingCapacity, int maxCapacity) {
1863 assert state == AVAILABLE;
1864 final int startIndex = nextAvailableSegmentOffset();
1865 if (startIndex == FREE_LIST_EMPTY) {
1866 return false;
1867 }
1868 allocatedBytes += segmentSize;
1869 try {
1870 buf.init(delegate, this, 0, 0, startIndex, size, startingCapacity, maxCapacity);
1871 } catch (Throwable t) {
1872 allocatedBytes -= segmentSize;
1873 releaseSegmentOffsetIntoFreeList(startIndex);
1874 PlatformDependent.throwException(t);
1875 }
1876 return true;
1877 }
1878
1879 private int nextAvailableSegmentOffset() {
1880 final int startIndex;
1881 IntStack localFreeList = this.localFreeList;
1882 if (localFreeList != null) {
1883 assert Thread.currentThread() == ownerThread;
1884 if (localFreeList.isEmpty()) {
1885 startIndex = externalFreeList.poll();
1886 } else {
1887 startIndex = localFreeList.pop();
1888 }
1889 } else {
1890 startIndex = externalFreeList.poll();
1891 }
1892 return startIndex;
1893 }
1894
1895
1896
1897
1898 public boolean hasRemainingCapacity() {
1899 int remaining = super.remainingCapacity();
1900 if (remaining > 0) {
1901 return true;
1902 }
1903 if (localFreeList != null) {
1904 return !localFreeList.isEmpty();
1905 }
1906 return !externalFreeList.isEmpty();
1907 }
1908
1909 boolean hasFullCapacity() {
1910 int free = externalFreeList.size();
1911 IntStack local = localFreeList;
1912 if (local != null) {
1913 free += local.size();
1914 }
1915 return free == segments;
1916 }
1917
1918 @Override
1919 public int remainingCapacity() {
1920 int remaining = super.remainingCapacity();
1921 return remaining > segmentSize ? remaining : updateRemainingCapacity(remaining);
1922 }
1923
1924 private int updateRemainingCapacity(int snapshotted) {
1925 int freeSegments = externalFreeList.size();
1926 IntStack localFreeList = this.localFreeList;
1927 if (localFreeList != null) {
1928 freeSegments += localFreeList.size();
1929 }
1930 int updated = freeSegments * segmentSize;
1931 if (updated != snapshotted) {
1932 allocatedBytes = capacity() - updated;
1933 }
1934 return updated;
1935 }
1936
1937 private void releaseSegmentOffsetIntoFreeList(int startIndex) {
1938 IntStack localFreeList = this.localFreeList;
1939 if (localFreeList != null && Thread.currentThread() == ownerThread) {
1940 localFreeList.push(startIndex);
1941 } else {
1942 boolean segmentReturned = externalFreeList.offer(startIndex);
1943 assert segmentReturned : "Unable to return segment " + startIndex + " to free list";
1944 }
1945 }
1946
1947 @Override
1948 void releaseSegment(int startIndex, int size) {
1949 IntStack localFreeList = this.localFreeList;
1950 if (localFreeList != null && Thread.currentThread() == ownerThread) {
1951 localFreeList.push(startIndex);
1952 int state = this.state;
1953 if (state != AVAILABLE) {
1954 updateStateOnLocalReleaseSegment(state, localFreeList);
1955 }
1956 } else {
1957 boolean segmentReturned = externalFreeList.offer(startIndex);
1958 assert segmentReturned;
1959
1960 int state = this.state;
1961 if (state != AVAILABLE) {
1962 deallocateIfNeeded(state);
1963 }
1964 }
1965 }
1966
1967 private void updateStateOnLocalReleaseSegment(int previousLocalSize, IntStack localFreeList) {
1968 int newLocalSize = localFreeList.size();
1969 boolean alwaysTrue = STATE.compareAndSet(this, previousLocalSize, newLocalSize);
1970 assert alwaysTrue : "this shouldn't happen unless double release in the local free list";
1971 deallocateIfNeeded(newLocalSize);
1972 }
1973
1974 private void deallocateIfNeeded(int localSize) {
1975
1976 int totalFreeSegments = localSize + externalFreeList.size();
1977 if (totalFreeSegments == segments && STATE.compareAndSet(this, localSize, DEALLOCATED)) {
1978 deallocate();
1979 }
1980 }
1981
1982 @Override
1983 void markToDeallocate() {
1984 IntStack localFreeList = this.localFreeList;
1985 int localSize = localFreeList != null ? localFreeList.size() : 0;
1986 STATE.set(this, localSize);
1987 deallocateIfNeeded(localSize);
1988 }
1989 }
1990
1991 private static final class BuddyChunk extends Chunk implements IntConsumer {
1992 private static final int MIN_BUDDY_SIZE = 32768;
1993 private static final byte IS_CLAIMED = (byte) (1 << 7);
1994 private static final byte HAS_CLAIMED_CHILDREN = 1 << 6;
1995 private static final byte SHIFT_MASK = ~(IS_CLAIMED | HAS_CLAIMED_CHILDREN);
1996 private static final int PACK_OFFSET_MASK = 0xFFFF;
1997 private static final int PACK_SIZE_SHIFT = Integer.SIZE - Integer.numberOfLeadingZeros(PACK_OFFSET_MASK);
1998
1999 private final MpscAtomicIntegerArrayQueue freeList;
2000
2001 private final byte[] buddies;
2002 private final int freeListCapacity;
2003
2004 BuddyChunk(AbstractByteBuf delegate, Magazine magazine) {
2005 super(delegate, magazine);
2006 freeListCapacity = delegate.capacity() / MIN_BUDDY_SIZE;
2007 int maxShift = Integer.numberOfTrailingZeros(freeListCapacity);
2008 assert maxShift <= 30;
2009
2010 freeList = new MpscAtomicIntegerArrayQueue(freeListCapacity, -1);
2011 buddies = new byte[freeListCapacity << 1];
2012
2013
2014 int index = 1;
2015 int runLength = 1;
2016 int currentRun = 0;
2017 while (maxShift > 0) {
2018 buddies[index++] = (byte) maxShift;
2019 if (++currentRun == runLength) {
2020 currentRun = 0;
2021 runLength <<= 1;
2022 maxShift--;
2023 }
2024 }
2025 }
2026
2027 @Override
2028 public boolean readInitInto(AdaptiveByteBuf buf, int size, int startingCapacity, int maxCapacity) {
2029 if (!freeList.isEmpty()) {
2030 freeList.drain(freeListCapacity, this);
2031 }
2032 int startIndex = chooseFirstFreeBuddy(1, startingCapacity, 0);
2033 if (startIndex == -1) {
2034 return false;
2035 }
2036 Chunk chunk = this;
2037 chunk.retain();
2038 try {
2039 buf.init(delegate, this, 0, 0, startIndex, size, startingCapacity, maxCapacity);
2040 allocatedBytes += startingCapacity;
2041 chunk = null;
2042 } finally {
2043 if (chunk != null) {
2044 unreserveMatchingBuddy(1, startingCapacity, startIndex, 0);
2045
2046
2047 chunk.release();
2048 }
2049 }
2050 return true;
2051 }
2052
2053 @Override
2054 public void accept(int packed) {
2055
2056 int size = unpackSize(packed);
2057 int offset = unpackOffset(packed);
2058 unreserveMatchingBuddy(1, size, offset, 0);
2059 allocatedBytes -= size;
2060 }
2061
2062 private static int unpackSize(int packed) {
2063 return MIN_BUDDY_SIZE << (packed >> PACK_SIZE_SHIFT);
2064 }
2065
2066 private static int unpackOffset(int packed) {
2067 return (packed & PACK_OFFSET_MASK) * MIN_BUDDY_SIZE;
2068 }
2069
2070 @Override
2071 void releaseSegment(int startingIndex, int size) {
2072 int packedOffset = startingIndex / MIN_BUDDY_SIZE;
2073 int packedSize = Integer.numberOfTrailingZeros(size / MIN_BUDDY_SIZE) << PACK_SIZE_SHIFT;
2074 int packed = packedOffset | packedSize;
2075 freeList.offer(packed);
2076 release();
2077 }
2078
2079 @Override
2080 public int remainingCapacity() {
2081 int capacityInFreeList = 0;
2082 if (!freeList.isEmpty()) {
2083 capacityInFreeList = freeList.weakPeekReduce(freeListCapacity, 0,
2084 new MpscAtomicIntegerArrayQueue.IntBinaryOperator() {
2085 @Override
2086 public int applyAsInt(int sum, int entry) {
2087 return sum + unpackSize(entry);
2088 }
2089 });
2090 }
2091 return super.remainingCapacity() + capacityInFreeList;
2092 }
2093
2094 @Override
2095 public boolean hasUnprocessedFreelistEntries() {
2096 return !freeList.isEmpty();
2097 }
2098
2099 @Override
2100 public void processFreelistEntries() {
2101 freeList.drain(freeListCapacity, this);
2102 }
2103
2104
2105
2106
2107 private int chooseFirstFreeBuddy(int index, int size, int currOffset) {
2108 byte[] buddies = this.buddies;
2109 while (index < buddies.length) {
2110 byte buddy = buddies[index];
2111 int currValue = MIN_BUDDY_SIZE << (buddy & SHIFT_MASK);
2112 if (currValue < size || (buddy & IS_CLAIMED) == IS_CLAIMED) {
2113 return -1;
2114 }
2115 if (currValue == size && (buddy & HAS_CLAIMED_CHILDREN) == 0) {
2116 buddies[index] |= IS_CLAIMED;
2117 return currOffset;
2118 }
2119 int found = chooseFirstFreeBuddy(index << 1, size, currOffset);
2120 if (found != -1) {
2121 buddies[index] |= HAS_CLAIMED_CHILDREN;
2122 return found;
2123 }
2124 index = (index << 1) + 1;
2125 currOffset += currValue >> 1;
2126 }
2127 return -1;
2128 }
2129
2130
2131
2132
2133 private boolean unreserveMatchingBuddy(int index, int size, int offset, int currOffset) {
2134 byte[] buddies = this.buddies;
2135 if (buddies.length <= index) {
2136 return false;
2137 }
2138 byte buddy = buddies[index];
2139 int currSize = MIN_BUDDY_SIZE << (buddy & SHIFT_MASK);
2140
2141 if (currSize == size) {
2142
2143 if (currOffset == offset) {
2144 buddies[index] &= SHIFT_MASK;
2145 return false;
2146 }
2147 throw new IllegalStateException("The intended segment was not found at index " +
2148 index + ", for size " + size + " and offset " + offset);
2149 }
2150
2151
2152 boolean claims;
2153 int siblingIndex;
2154 if (offset < currOffset + (currSize >> 1)) {
2155
2156 claims = unreserveMatchingBuddy(index << 1, size, offset, currOffset);
2157 siblingIndex = (index << 1) + 1;
2158 } else {
2159
2160 claims = unreserveMatchingBuddy((index << 1) + 1, size, offset, currOffset + (currSize >> 1));
2161 siblingIndex = index << 1;
2162 }
2163 if (!claims) {
2164
2165 byte sibling = buddies[siblingIndex];
2166 if ((sibling & SHIFT_MASK) == sibling) {
2167
2168 buddies[index] &= SHIFT_MASK;
2169 return false;
2170 }
2171 }
2172 return true;
2173 }
2174
2175 @Override
2176 public String toString() {
2177 int capacity = delegate.capacity();
2178 int remaining = capacity - allocatedBytes;
2179 return "BuddyChunk[capacity: " + capacity +
2180 ", remaining: " + remaining +
2181 ", free list: " + freeList.size() + ']';
2182 }
2183 }
2184
2185 static final class AdaptiveByteBuf extends AbstractReferenceCountedByteBuf {
2186
2187 private final EnhancedHandle<AdaptiveByteBuf> handle;
2188
2189
2190 private int startIndex;
2191 private AbstractByteBuf rootParent;
2192 Chunk chunk;
2193 private int length;
2194 private int maxFastCapacity;
2195 private ByteBuffer tmpNioBuf;
2196 private boolean hasArray;
2197 private boolean hasMemoryAddress;
2198
2199 AdaptiveByteBuf(EnhancedHandle<AdaptiveByteBuf> recyclerHandle) {
2200 super(0);
2201 handle = ObjectUtil.checkNotNull(recyclerHandle, "recyclerHandle");
2202 }
2203
2204 void init(AbstractByteBuf unwrapped, Chunk wrapped, int readerIndex, int writerIndex,
2205 int startIndex, int size, int capacity, int maxCapacity) {
2206 this.startIndex = startIndex;
2207 chunk = wrapped;
2208 length = size;
2209 maxFastCapacity = capacity;
2210 maxCapacity(maxCapacity);
2211 setIndex0(readerIndex, writerIndex);
2212 hasArray = unwrapped.hasArray();
2213 hasMemoryAddress = unwrapped.hasMemoryAddress();
2214 rootParent = unwrapped;
2215 tmpNioBuf = null;
2216 }
2217
2218 private AbstractByteBuf rootParent() {
2219 final AbstractByteBuf rootParent = this.rootParent;
2220 if (rootParent != null) {
2221 return rootParent;
2222 }
2223 throw new IllegalReferenceCountException();
2224 }
2225
2226 @Override
2227 public int capacity() {
2228 return length;
2229 }
2230
2231 @Override
2232 public int maxFastWritableBytes() {
2233 return Math.min(maxFastCapacity, maxCapacity()) - writerIndex;
2234 }
2235
2236 @Override
2237 public ByteBuf capacity(int newCapacity) {
2238 checkNewCapacity(newCapacity);
2239 if (length <= newCapacity && newCapacity <= maxFastCapacity) {
2240 length = newCapacity;
2241 return this;
2242 }
2243 if (newCapacity < capacity()) {
2244 length = newCapacity;
2245 trimIndicesToCapacity(newCapacity);
2246 return this;
2247 }
2248
2249
2250 Chunk chunk = this.chunk;
2251 AdaptivePoolingAllocator allocator = chunk.allocator;
2252 int readerIndex = this.readerIndex;
2253 int writerIndex = this.writerIndex;
2254 int baseOldRootIndex = startIndex;
2255 int oldLength = length;
2256 int oldCapacity = maxFastCapacity;
2257 AbstractByteBuf oldRoot = rootParent();
2258 allocator.reallocate(newCapacity, maxCapacity(), this);
2259 oldRoot.getBytes(baseOldRootIndex, this, 0, oldLength);
2260 chunk.releaseSegment(baseOldRootIndex, oldCapacity);
2261 assert oldCapacity < maxFastCapacity && newCapacity <= maxFastCapacity :
2262 "Capacity increase failed";
2263 this.readerIndex = readerIndex;
2264 this.writerIndex = writerIndex;
2265 return this;
2266 }
2267
2268 @Override
2269 public ByteBufAllocator alloc() {
2270 return rootParent().alloc();
2271 }
2272
2273 @SuppressWarnings("deprecation")
2274 @Override
2275 public ByteOrder order() {
2276 return rootParent().order();
2277 }
2278
2279 @Override
2280 public ByteBuf unwrap() {
2281 return null;
2282 }
2283
2284 @Override
2285 public boolean isDirect() {
2286 return rootParent().isDirect();
2287 }
2288
2289 @Override
2290 public int arrayOffset() {
2291 return idx(rootParent().arrayOffset());
2292 }
2293
2294 @Override
2295 public boolean hasMemoryAddress() {
2296 return hasMemoryAddress;
2297 }
2298
2299 @Override
2300 public long memoryAddress() {
2301 ensureAccessible();
2302 return rootParent().memoryAddress() + startIndex;
2303 }
2304
2305 @Override
2306 public ByteBuffer nioBuffer(int index, int length) {
2307 checkIndex(index, length);
2308 return rootParent().nioBuffer(idx(index), length);
2309 }
2310
2311 @Override
2312 public ByteBuffer internalNioBuffer(int index, int length) {
2313 checkIndex(index, length);
2314 return (ByteBuffer) internalNioBuffer().position(index).limit(index + length);
2315 }
2316
2317 private ByteBuffer internalNioBuffer() {
2318 if (tmpNioBuf == null) {
2319 tmpNioBuf = rootParent().nioBuffer(startIndex, maxFastCapacity);
2320 }
2321 return (ByteBuffer) tmpNioBuf.clear();
2322 }
2323
2324 @Override
2325 public ByteBuffer[] nioBuffers(int index, int length) {
2326 checkIndex(index, length);
2327 return rootParent().nioBuffers(idx(index), length);
2328 }
2329
2330 @Override
2331 public boolean hasArray() {
2332 return hasArray;
2333 }
2334
2335 @Override
2336 public byte[] array() {
2337 ensureAccessible();
2338 return rootParent().array();
2339 }
2340
2341 @Override
2342 public ByteBuf copy(int index, int length) {
2343 checkIndex(index, length);
2344 return rootParent().copy(idx(index), length);
2345 }
2346
2347 @Override
2348 public int nioBufferCount() {
2349 return rootParent().nioBufferCount();
2350 }
2351
2352 @Override
2353 protected byte _getByte(int index) {
2354 return rootParent()._getByte(idx(index));
2355 }
2356
2357 @Override
2358 protected short _getShort(int index) {
2359 return rootParent()._getShort(idx(index));
2360 }
2361
2362 @Override
2363 protected short _getShortLE(int index) {
2364 return rootParent()._getShortLE(idx(index));
2365 }
2366
2367 @Override
2368 protected int _getUnsignedMedium(int index) {
2369 return rootParent()._getUnsignedMedium(idx(index));
2370 }
2371
2372 @Override
2373 protected int _getUnsignedMediumLE(int index) {
2374 return rootParent()._getUnsignedMediumLE(idx(index));
2375 }
2376
2377 @Override
2378 protected int _getInt(int index) {
2379 return rootParent()._getInt(idx(index));
2380 }
2381
2382 @Override
2383 protected int _getIntLE(int index) {
2384 return rootParent()._getIntLE(idx(index));
2385 }
2386
2387 @Override
2388 protected long _getLong(int index) {
2389 return rootParent()._getLong(idx(index));
2390 }
2391
2392 @Override
2393 protected long _getLongLE(int index) {
2394 return rootParent()._getLongLE(idx(index));
2395 }
2396
2397 @Override
2398 public ByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
2399 checkIndex(index, length);
2400 rootParent().getBytes(idx(index), dst, dstIndex, length);
2401 return this;
2402 }
2403
2404 @Override
2405 public ByteBuf getBytes(int index, byte[] dst, int dstIndex, int length) {
2406 checkIndex(index, length);
2407 rootParent().getBytes(idx(index), dst, dstIndex, length);
2408 return this;
2409 }
2410
2411 @Override
2412 public ByteBuf getBytes(int index, ByteBuffer dst) {
2413 checkIndex(index, dst.remaining());
2414 rootParent().getBytes(idx(index), dst);
2415 return this;
2416 }
2417
2418 @Override
2419 protected void _setByte(int index, int value) {
2420 rootParent()._setByte(idx(index), value);
2421 }
2422
2423 @Override
2424 protected void _setShort(int index, int value) {
2425 rootParent()._setShort(idx(index), value);
2426 }
2427
2428 @Override
2429 protected void _setShortLE(int index, int value) {
2430 rootParent()._setShortLE(idx(index), value);
2431 }
2432
2433 @Override
2434 protected void _setMedium(int index, int value) {
2435 rootParent()._setMedium(idx(index), value);
2436 }
2437
2438 @Override
2439 protected void _setMediumLE(int index, int value) {
2440 rootParent()._setMediumLE(idx(index), value);
2441 }
2442
2443 @Override
2444 protected void _setInt(int index, int value) {
2445 rootParent()._setInt(idx(index), value);
2446 }
2447
2448 @Override
2449 protected void _setIntLE(int index, int value) {
2450 rootParent()._setIntLE(idx(index), value);
2451 }
2452
2453 @Override
2454 protected void _setLong(int index, long value) {
2455 rootParent()._setLong(idx(index), value);
2456 }
2457
2458 @Override
2459 protected void _setLongLE(int index, long value) {
2460 rootParent()._setLongLE(idx(index), value);
2461 }
2462
2463 @Override
2464 public ByteBuf setBytes(int index, byte[] src, int srcIndex, int length) {
2465 checkIndex(index, length);
2466 ByteBuffer tmp = (ByteBuffer) internalNioBuffer().clear().position(index);
2467 tmp.put(src, srcIndex, length);
2468 return this;
2469 }
2470
2471 @Override
2472 public ByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length) {
2473 checkIndex(index, length);
2474 ByteBuffer tmp = (ByteBuffer) internalNioBuffer().clear().position(index);
2475 tmp.put(src.nioBuffer(srcIndex, length));
2476 return this;
2477 }
2478
2479 @Override
2480 public ByteBuf setBytes(int index, ByteBuffer src) {
2481 checkIndex(index, src.remaining());
2482 ByteBuffer tmp = (ByteBuffer) internalNioBuffer().clear().position(index);
2483 tmp.put(src);
2484 return this;
2485 }
2486
2487 @Override
2488 public ByteBuf getBytes(int index, OutputStream out, int length)
2489 throws IOException {
2490 checkIndex(index, length);
2491 if (length != 0) {
2492 ByteBufUtil.readBytes(alloc(), internalNioBuffer().duplicate(), index, length, out);
2493 }
2494 return this;
2495 }
2496
2497 @Override
2498 public int getBytes(int index, GatheringByteChannel out, int length)
2499 throws IOException {
2500 ByteBuffer buf = internalNioBuffer().duplicate();
2501 buf.clear().position(index).limit(index + length);
2502 return out.write(buf);
2503 }
2504
2505 @Override
2506 public int getBytes(int index, FileChannel out, long position, int length)
2507 throws IOException {
2508 ByteBuffer buf = internalNioBuffer().duplicate();
2509 buf.clear().position(index).limit(index + length);
2510 return out.write(buf, position);
2511 }
2512
2513 @Override
2514 public int setBytes(int index, InputStream in, int length)
2515 throws IOException {
2516 checkIndex(index, length);
2517 final AbstractByteBuf rootParent = rootParent();
2518 if (rootParent.hasArray()) {
2519 return rootParent.setBytes(idx(index), in, length);
2520 }
2521 byte[] tmp = ByteBufUtil.threadLocalTempArray(length);
2522 int readBytes = in.read(tmp, 0, length);
2523 if (readBytes <= 0) {
2524 return readBytes;
2525 }
2526 setBytes(index, tmp, 0, readBytes);
2527 return readBytes;
2528 }
2529
2530 @Override
2531 public int setBytes(int index, ScatteringByteChannel in, int length)
2532 throws IOException {
2533 try {
2534 return in.read(internalNioBuffer(index, length));
2535 } catch (ClosedChannelException ignored) {
2536 return -1;
2537 }
2538 }
2539
2540 @Override
2541 public int setBytes(int index, FileChannel in, long position, int length)
2542 throws IOException {
2543 try {
2544 return in.read(internalNioBuffer(index, length), position);
2545 } catch (ClosedChannelException ignored) {
2546 return -1;
2547 }
2548 }
2549
2550 @Override
2551 public int setCharSequence(int index, CharSequence sequence, Charset charset) {
2552 return setCharSequence0(index, sequence, charset, false);
2553 }
2554
2555 private int setCharSequence0(int index, CharSequence sequence, Charset charset, boolean expand) {
2556 if (charset.equals(CharsetUtil.UTF_8)) {
2557 int length = ByteBufUtil.utf8MaxBytes(sequence);
2558 if (expand) {
2559 ensureWritable0(length);
2560 checkIndex0(index, length);
2561 } else {
2562 checkIndex(index, length);
2563 }
2564 return ByteBufUtil.writeUtf8(this, index, length, sequence, sequence.length());
2565 }
2566 if (charset.equals(CharsetUtil.US_ASCII) || charset.equals(CharsetUtil.ISO_8859_1)) {
2567 int length = sequence.length();
2568 if (expand) {
2569 ensureWritable0(length);
2570 checkIndex0(index, length);
2571 } else {
2572 checkIndex(index, length);
2573 }
2574 return ByteBufUtil.writeAscii(this, index, sequence, length);
2575 }
2576 byte[] bytes = sequence.toString().getBytes(charset);
2577 if (expand) {
2578 ensureWritable0(bytes.length);
2579
2580 }
2581 setBytes(index, bytes);
2582 return bytes.length;
2583 }
2584
2585 @Override
2586 public int writeCharSequence(CharSequence sequence, Charset charset) {
2587 int written = setCharSequence0(writerIndex, sequence, charset, true);
2588 writerIndex += written;
2589 return written;
2590 }
2591
2592 @Override
2593 public int forEachByte(int index, int length, ByteProcessor processor) {
2594 checkIndex(index, length);
2595 int ret = rootParent().forEachByte(idx(index), length, processor);
2596 return forEachResult(ret);
2597 }
2598
2599 @Override
2600 public int forEachByteDesc(int index, int length, ByteProcessor processor) {
2601 checkIndex(index, length);
2602 int ret = rootParent().forEachByteDesc(idx(index), length, processor);
2603 return forEachResult(ret);
2604 }
2605
2606 @Override
2607 public ByteBuf setZero(int index, int length) {
2608 checkIndex(index, length);
2609 rootParent().setZero(idx(index), length);
2610 return this;
2611 }
2612
2613 @Override
2614 public ByteBuf writeZero(int length) {
2615 ensureWritable(length);
2616 rootParent().setZero(idx(writerIndex), length);
2617 writerIndex += length;
2618 return this;
2619 }
2620
2621 private int forEachResult(int ret) {
2622 if (ret < startIndex) {
2623 return -1;
2624 }
2625 return ret - startIndex;
2626 }
2627
2628 @Override
2629 public boolean isContiguous() {
2630 return rootParent().isContiguous();
2631 }
2632
2633 private int idx(int index) {
2634 return index + startIndex;
2635 }
2636
2637 @Override
2638 protected void deallocate() {
2639 if (chunk != null) {
2640 chunk.releaseSegment(startIndex, maxFastCapacity);
2641 }
2642 tmpNioBuf = null;
2643 chunk = null;
2644 rootParent = null;
2645 handle.unguardedRecycle(this);
2646 }
2647 }
2648
2649
2650
2651
2652 interface ChunkAllocator {
2653
2654
2655
2656
2657
2658
2659
2660 AbstractByteBuf allocate(int initialCapacity, int maxCapacity);
2661 }
2662 }