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