netty源碼解解析(4.0)-23 ByteBuf記憶體管理:分配和釋放
- 2019 年 10 月 9 日
- 筆記
ByteBuf記憶體分配和釋放由具體實現負責，抽象類型只定義的記憶體分配和釋放的時機。
記憶體分配分兩個階段: 第一階段，初始化時分配記憶體。第二階段: 記憶體不夠用時分配新的記憶體。ByteBuf抽象層沒有定義第一階段的行為，但定義了第二階段的方法：
public abstract ByteBuf capacity(int newCapacity)
這個方法負責分配一個長度為newCapacity的新記憶體。
記憶體釋放的抽象實現在AbstractReferenceCountedByteBuf中實現，這個類實現引用計數，當調用release方法把引用計數變成0時，會調用
protected abstract void deallocate()
執行真正的記憶體釋放操作。
記憶體相關的屬性
ByteBuf定義了兩個記憶體相關的屬性:
capacity: 當前的當前的容量，也就是當前使用的記憶體大小。使用capacity()方法獲得。
maxCapacity: 最大容量，也就是可以使用的最大記憶體大小。使用newCapacity()方法獲得。
記憶體分配時機
AbstractByteBuf定義了記憶體分配的時機。當writeXX方法被調用的時候，如果如果發現可寫空間不足，就調用capacity分配新的記憶體。下面以writeInt為例詳細分析這個過程。
1 @Override 2 public ByteBuf writeInt(int value) { 3 ensureWritable0(4); 4 _setInt(writerIndex, value); 5 writerIndex += 4; 6 return this; 7 } 8 9 10 final void ensureWritable0(int minWritableBytes) { 11 ensureAccessible(); 12 if (minWritableBytes <= writableBytes()) { 13 return; 14 } 15 16 if (minWritableBytes > maxCapacity - writerIndex) { 17 throw new IndexOutOfBoundsException(String.format( 18 "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s", 19 writerIndex, minWritableBytes, maxCapacity, this)); 20 } 21 22 // Normalize the current capacity to the power of 2. 23 int newCapacity = calculateNewCapacity(writerIndex + minWritableBytes); 24 25 // Adjust to the new capacity. 26 capacity(newCapacity); 27 } 28
3行: ensureWritable確保當前能寫入4Byte的數據。
11行: 確保當前ByteBuf是可以訪問的。防止多執行緒環境下，ByteBuf記憶體被釋放後讀寫數據。
12，13行: 如果記憶體夠用，就此作罷。
16行: 如果需要記憶體大於maxCapacity拋出異常。
23, 26行行: 計算新記憶體的大小, 調用capacity(int)分配新記憶體。
重新分配記憶體之前的一個重要步驟的計算新記憶體的大小。這個工作由calculateNewCapacity方法完成，它的程式碼如下:
1 private int calculateNewCapacity(int minNewCapacity) { 2 final int maxCapacity = this.maxCapacity; 3 final int threshold = 1048576 * 4; // 4 MiB page 4 5 if (minNewCapacity == threshold) { 6 return threshold; 7 } 8 9 // If over threshold, do not double but just increase by threshold. 10 if (minNewCapacity > threshold) { 11 int newCapacity = minNewCapacity / threshold * threshold; 12 if (newCapacity > maxCapacity - threshold) { 13 newCapacity = maxCapacity; 14 } else { 15 newCapacity += threshold; 16 } 17 return newCapacity; 18 } 19 20 // Not over threshold. Double up to 4 MiB, starting from 64. 21 int newCapacity = 64; 22 while (newCapacity < minNewCapacity) { 23 newCapacity <<= 1; 24 } 25 26 return Math.min(newCapacity, maxCapacity); 27 }
1行：接受一個最小的新記憶體參數minNewCapacity。
3行： 定義一個4MB的閾值常量threshold。
5，6行: 如果minNewCapacity==threshold，那麼新記憶體大小就是threshold。
10-17行: 如果minNewCapacity>threshold, 新記憶體大小是min(maxCapacity, threshold * n)且threshold * n >= minNewCapacity。
21-26行: 如果minNewCapacity<threshold, 新記憶體大小是min(maxCapacity, 64 * 2n)且64 * 2n >= minNewCapacity。
記憶體分配和釋放的具體實現
本章涉及到的記憶體分配和釋放的具體實現只涉及到unpooled類型的ByteBuf，即:
UnpooledHeapByteBuf
UnpooledDirectByteBuf
UnpooledUnsafeHeapByteBuf
UnpooledUnsafeDirectByteBuf
這幾個具體實現中涉及到的記憶體分配和釋放程式碼比較簡潔，更容易明白ByteBuf記憶體管理的原理。相比之下，pooled類型的ByteBuf記憶體分配和釋放的程式碼要複雜很多，會在後面的章節獨立分析。
UnpooledHeapByteBuf和UnpooledUnsafeHeapByteBuf實現
UnpooledHeapByteBuf中，記憶體分配的實現程式碼主要集中在capacity(int)和allocateArray()方法中。capacity分配新記憶體的步驟是
- 調用allocateArray分配一塊新記憶體。
- 把舊記憶體中的實際複製到新記憶體中。
- 使用新記憶體替換舊記憶體(24行)。
- 釋放掉舊記憶體(25行)。
程式碼如下:
1 @Override 2 public ByteBuf capacity(int newCapacity) { 3 checkNewCapacity(newCapacity); 4 5 int oldCapacity = array.length; 6 byte[] oldArray = array; 7 if (newCapacity > oldCapacity) { 8 byte[] newArray = allocateArray(newCapacity); 9 System.arraycopy(oldArray, 0, newArray, 0, oldArray.length); 10 setArray(newArray); 11 freeArray(oldArray); 12 } else if (newCapacity < oldCapacity) { 13 byte[] newArray = allocateArray(newCapacity); 14 int readerIndex = readerIndex(); 15 if (readerIndex < newCapacity) { 16 int writerIndex = writerIndex(); 17 if (writerIndex > newCapacity) { 18 writerIndex(writerIndex = newCapacity); 19 } 20 System.arraycopy(oldArray, readerIndex, newArray, readerIndex, writerIndex - readerIndex); 21 } else { 22 setIndex(newCapacity, newCapacity); 23 } 24 setArray(newArray); 25 freeArray(oldArray); 26 } 27 return this; 28 } 29
capacity中複製舊記憶體數據到新記憶體中的時候分兩種情況(newCapacity，oldCapacity分別是新舊記憶體的大小):
- newCapacity>oldCapacity，這種情況比較簡單，直接複製就好了(第8行)。不影響readerIndex和writerIndex。
- newCapacity<oldCapacity, 這種情況比較複雜。capacity盡量把可讀的數據複製新記憶體中。
- 如果readerIndex<newCapacity且writerIndex<newCapacity。可讀數據會全部轉移到新記憶體中。readerIndex和writerIndex保持不變。
- 如果readerIndex<newCapacity且writeIndex>newCapacity。可端數據會部分轉移的新記憶體中，會丟失部分可讀數據。readerIndex不變，writerIndex變成newCapacity。
- 如果readerIndex>newCapacity，數據全部丟失，readerIndex和writerIndex都會變成newCapacity。
allocateArray方法負責分配一塊新記憶體，它的實現是new byte[]。freeArray方法負責釋放記憶體，這個方法是個空方法。
UnpooledUnsafeHeapByteBuf是UnloopedHeadpByteBuf的直接子類，在記憶體管理上的差別是allocateArray的實現，UnpooledUnsafeHeapByteBuf的實現是:
1 @Override 2 byte[] allocateArray(int initialCapacity) { 3 return PlatformDependent.allocateUninitializedArray(initialCapacity); 4 }
UnpooledDirectByteBuf和UnpooledUnsafeDirectByteBuf實現
UnpooledDirectByteBuf類使用DirectByteBuffer作為記憶體，使用了DirectByteBuffer的能力來實現ByteBuf介面。allocateDirect和freeDirect方法負責分配和釋放DirectByteBuffer。capacity(int)方法和UnloopedHeapByteBuf類似，使用allocateDirect創建一個新的DirectByteBuffer, 把舊記憶體數據複製到新記憶體中，然後使用新記憶體替換舊記憶體，最後調用freeDirect方法釋放掉舊的DirectByteBuffer。
1 protected ByteBuffer allocateDirect(int initialCapacity) { 2 return ByteBuffer.allocateDirect(initialCapacity); 3 } 4 5 protected void freeDirect(ByteBuffer buffer) { 6 PlatformDependent.freeDirectBuffer(buffer); 7 } 8 9 @Override 10 public ByteBuf capacity(int newCapacity) { 11 checkNewCapacity(newCapacity); 12 13 int readerIndex = readerIndex(); 14 int writerIndex = writerIndex(); 15 16 int oldCapacity = capacity; 17 if (newCapacity > oldCapacity) { 18 ByteBuffer oldBuffer = buffer; 19 ByteBuffer newBuffer = allocateDirect(newCapacity); 20 oldBuffer.position(0).limit(oldBuffer.capacity()); 21 newBuffer.position(0).limit(oldBuffer.capacity()); 22 newBuffer.put(oldBuffer); 23 newBuffer.clear(); 24 setByteBuffer(newBuffer); 25 } else if (newCapacity < oldCapacity) { 26 ByteBuffer oldBuffer = buffer; 27 ByteBuffer newBuffer = allocateDirect(newCapacity); 28 if (readerIndex < newCapacity) { 29 if (writerIndex > newCapacity) { 30 writerIndex(writerIndex = newCapacity); 31 } 32 oldBuffer.position(readerIndex).limit(writerIndex); 33 newBuffer.position(readerIndex).limit(writerIndex); 34 newBuffer.put(oldBuffer); 35 newBuffer.clear(); 36 } else { 37 setIndex(newCapacity, newCapacity); 38 } 39 setByteBuffer(newBuffer); 40 } 41 return this; 42 }
對比UnloopedHeapByteBuf的capacity(int)方法，發現這兩個實現非常類似，也分兩種情況處理:
- 18-24行，newCapacity > oldCapacity的情況。
- 26-39行, newCapacity < oldCapacity的情況。
兩種情況對readerIndex和writerIndex的影響也一樣，不同的是數據複製時的具體實現。
UnpooledUnsafeDirectByteBuf和UnpooledDirectByteBuf同屬於AbstractReferenceCountedByteBuf的派生類，它們之間沒有繼承關係。但記憶體分配和釋放實現是一樣的，不同的地方是記憶體I/O。UnpooledUnsafeDirectByteBuf使用UnsafeByteBufUtil類之間讀寫DirectByteBuffer的記憶體，沒有使用DirectByteBuffer的I/O能力。
