-
Notifications
You must be signed in to change notification settings - Fork 77
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor YamuxHandler.SendBuffer (#328)
* Introduce ByteBufQueue * Add ByteBufQueue tests * Writing exec path is always through fill/drain buffer * Adopt/fix existing tests * Add new test checking correct handling of negative sendWindowSize
- Loading branch information
1 parent
26efe02
commit 7dc2fa2
Showing
4 changed files
with
328 additions
and
87 deletions.
There are no files selected for viewing
44 changes: 44 additions & 0 deletions
44
libp2p/src/main/kotlin/io/libp2p/etc/util/netty/ByteBufQueue.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package io.libp2p.etc.util.netty | ||
|
|
||
| import io.netty.buffer.ByteBuf | ||
| import io.netty.buffer.Unpooled | ||
|
|
||
| class ByteBufQueue { | ||
| private val data: MutableList<ByteBuf> = mutableListOf() | ||
|
|
||
| fun push(buf: ByteBuf) { | ||
| data += buf | ||
| } | ||
|
|
||
| fun take(maxLength: Int): ByteBuf { | ||
| val wholeBuffers = mutableListOf<ByteBuf>() | ||
| var size = 0 | ||
| while (data.isNotEmpty()) { | ||
| val bufLen = data.first().readableBytes() | ||
| if (size + bufLen > maxLength) break | ||
| size += bufLen | ||
| wholeBuffers += data.removeFirst() | ||
| if (size == maxLength) break | ||
| } | ||
|
|
||
| val partialBufferSlice = | ||
| when { | ||
| data.isEmpty() -> null | ||
| size == maxLength -> null | ||
| else -> data.first() | ||
| } | ||
| ?.let { buf -> | ||
| val remainingBytes = maxLength - size | ||
| buf.readRetainedSlice(remainingBytes) | ||
| } | ||
|
|
||
| val allBuffers = wholeBuffers + listOfNotNull(partialBufferSlice) | ||
| return Unpooled.wrappedBuffer(*allBuffers.toTypedArray()) | ||
| } | ||
|
|
||
| fun dispose() { | ||
| data.forEach { it.release() } | ||
| } | ||
|
|
||
| fun readableBytes(): Int = data.sumOf { it.readableBytes() } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
167 changes: 167 additions & 0 deletions
167
libp2p/src/test/kotlin/io/libp2p/etc/util/netty/ByteBufQueueTest.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,167 @@ | ||
| package io.libp2p.etc.util.netty | ||
|
|
||
| import io.libp2p.tools.readAllBytesAndRelease | ||
| import io.netty.buffer.ByteBuf | ||
| import io.netty.buffer.Unpooled | ||
| import org.assertj.core.api.Assertions.assertThat | ||
| import org.junit.jupiter.api.AfterEach | ||
| import org.junit.jupiter.api.Test | ||
|
|
||
| class ByteBufQueueTest { | ||
|
|
||
| val queue = ByteBufQueue() | ||
|
|
||
| val allocatedBufs = mutableListOf<ByteBuf>() | ||
|
|
||
| @AfterEach | ||
| fun cleanUpAndCheck() { | ||
| allocatedBufs.forEach { | ||
| assertThat(it.refCnt()).isEqualTo(1) | ||
| } | ||
| } | ||
|
|
||
| fun allocateBuf(): ByteBuf { | ||
| val buf = Unpooled.buffer() | ||
| buf.retain() // ref counter to 2 to check that exactly 1 ref remains at the end | ||
| allocatedBufs += buf | ||
| return buf | ||
| } | ||
|
|
||
| fun allocateData(data: String): ByteBuf = | ||
| allocateBuf().writeBytes(data.toByteArray()) | ||
|
|
||
| fun ByteBuf.readString() = String(this.readAllBytesAndRelease()) | ||
|
|
||
| @Test | ||
| fun emptyTest() { | ||
| assertThat(queue.take(100).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun zeroTest() { | ||
| queue.push(allocateData("abc")) | ||
| assertThat(queue.take(0).readString()).isEqualTo("") | ||
| assertThat(queue.take(100).readString()).isEqualTo("abc") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyZeroTest() { | ||
| assertThat(queue.take(0).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyBuffersTest1() { | ||
| queue.push(allocateData("")) | ||
| assertThat(queue.take(10).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyBuffersTest2() { | ||
| queue.push(allocateData("")) | ||
| assertThat(queue.take(0).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyBuffersTest3() { | ||
| queue.push(allocateData("")) | ||
| queue.push(allocateData("a")) | ||
| queue.push(allocateData("")) | ||
| assertThat(queue.take(10).readString()).isEqualTo("a") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyBuffersTest4() { | ||
| queue.push(allocateData("a")) | ||
| queue.push(allocateData("")) | ||
| assertThat(queue.take(10).readString()).isEqualTo("a") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyBuffersTest5() { | ||
| queue.push(allocateData("a")) | ||
| queue.push(allocateData("")) | ||
| assertThat(queue.take(1).readString()).isEqualTo("a") | ||
| assertThat(queue.take(1).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun emptyBuffersTest6() { | ||
| queue.push(allocateData("a")) | ||
| queue.push(allocateData("")) | ||
| queue.push(allocateData("")) | ||
| queue.push(allocateData("b")) | ||
| assertThat(queue.take(10).readString()).isEqualTo("ab") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTake1() { | ||
| queue.push(allocateData("abc")) | ||
| queue.push(allocateData("def")) | ||
|
|
||
| assertThat(queue.take(4).readString()).isEqualTo("abcd") | ||
| assertThat(queue.take(1).readString()).isEqualTo("e") | ||
| assertThat(queue.take(100).readString()).isEqualTo("f") | ||
| assertThat(queue.take(100).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTake2() { | ||
| queue.push(allocateData("abc")) | ||
| queue.push(allocateData("def")) | ||
|
|
||
| assertThat(queue.take(2).readString()).isEqualTo("ab") | ||
| assertThat(queue.take(2).readString()).isEqualTo("cd") | ||
| assertThat(queue.take(2).readString()).isEqualTo("ef") | ||
| assertThat(queue.take(2).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTake3() { | ||
| queue.push(allocateData("abc")) | ||
| queue.push(allocateData("def")) | ||
|
|
||
| assertThat(queue.take(1).readString()).isEqualTo("a") | ||
| assertThat(queue.take(1).readString()).isEqualTo("b") | ||
| assertThat(queue.take(1).readString()).isEqualTo("c") | ||
| assertThat(queue.take(1).readString()).isEqualTo("d") | ||
| assertThat(queue.take(1).readString()).isEqualTo("e") | ||
| assertThat(queue.take(1).readString()).isEqualTo("f") | ||
| assertThat(queue.take(1).readString()).isEqualTo("") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTakePush1() { | ||
| queue.push(allocateData("abc")) | ||
| assertThat(queue.take(2).readString()).isEqualTo("ab") | ||
| queue.push(allocateData("def")) | ||
| assertThat(queue.take(2).readString()).isEqualTo("cd") | ||
| assertThat(queue.take(100).readString()).isEqualTo("ef") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTakePush2() { | ||
| queue.push(allocateData("abc")) | ||
| assertThat(queue.take(3).readString()).isEqualTo("abc") | ||
| queue.push(allocateData("def")) | ||
| assertThat(queue.take(2).readString()).isEqualTo("de") | ||
| assertThat(queue.take(100).readString()).isEqualTo("f") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTakePush3() { | ||
| queue.push(allocateData("abc")) | ||
| queue.push(allocateData("def")) | ||
| assertThat(queue.take(1).readString()).isEqualTo("a") | ||
| queue.push(allocateData("ghi")) | ||
| assertThat(queue.take(100).readString()).isEqualTo("bcdefghi") | ||
| } | ||
|
|
||
| @Test | ||
| fun pushTakePush4() { | ||
| queue.push(allocateData("abc")) | ||
| assertThat(queue.take(1).readString()).isEqualTo("a") | ||
| queue.push(allocateData("def")) | ||
| queue.push(allocateData("ghi")) | ||
| assertThat(queue.take(100).readString()).isEqualTo("bcdefghi") | ||
| } | ||
| } |
Oops, something went wrong.