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