ByteBuf Module

Features

DataKernel’s mission is to make efficient yet high-level I/O, which requires extensive usage of user-space byte buffers. Unfortunately, traditional Java ByteBuffers impose a heavy load on GC (either clogging young gen, or being allocated directly in old gen because of their large sizes).

To reduce GC overhead, DataKernel introduces its own GC-friendly and lightweight ByteBufs, which can be reused with ByteBufPool.

In addition, common I/O pattern is to treat ByteBuffers as a queue: I/O operation produces the data while application consumes the data or vice versa. ByteBufs are designed to facilitate this pattern as well, and also provide specialized ByteBufQueue with queue-like operations across multiple ByteBufs.

ByteBuf

An extremely lightweight and efficient implementation compared to the Java NIO ByteBuffer. There are no direct buffers, which simplifies and improves ByteBuf performance.

ByteBuf is similar to a FIFO byte queue and has two positions: head and tail. When you write data to your ByteBuf, it’s tail increases by the number of bytes written. Similarly, when you read data from your ByteBuf, it’s head increases by the number of bytes read.

You can read bytes from ByteBuf only when the tail is greater than the head. Also, you can write bytes to ByteBuf until the tail doesn’t exceed the length of the wrapped array. In this way, there is no need for ByteBuffer.flip() operations.

ByteBuf supports concurrent processes: while some data is being written to the ByteBuf by one process, another one can read it.

To create a ByteBuf you can either wrap your byte array into ByteBuf or allocate it from ByteBufPool.

Note: If you create a ByteBuf without allocating it from ByteBufPool, calling ByteBuf.recycle() will have no effect, such ByteBufs are simply collected by GC.

ByteBufPool

Allows to reuse ByteBufs, and as a result, reduces GC load. To make ByteBufPool usage more convenient, there are debugging and monitoring tools for allocated ByteBufs, including their stack traces.

To get a ByteBuf from the pool, use ByteBufPool.allocate(int size). A ByteBuf of rounding up to the next nearest power of 2 sizes will be allocated (for example, if the size is 29, a ByteBuf of 32 bytes will be allocated).

To return ByteBuf to the ByteBufPool, use ByteBuf.recycle(). Unlike languages like C/C++, it’s not required to recycle ByteBufs - in the worst case, it will be collected by the GC.

To make everything consistent, DataKernel relies on the concept of ‘ownership’ (like in Rust language) - after allocation, the components pass ByteBuf from one to another, until the last ‘owner’ recycles it to ByteBufPool.

You can explore an example of ByteBuf pool usage here

ByteBufQueue

ByteBufQueue class provides effective management of multiple ByteBufs. It creates an optimized queue of several ByteBufs with FIFO rules.

You can explore an example of ByteBuf queue usage here

Utility classes

ByteBuf module also contains utility classes to manage and resize the underlying byte buffer, String conversions, etc.

You can add ByteBuf module to your project by inserting a dependency in pom.xml:

<dependency>
    <groupId>io.datakernel</groupId>
    <artifactId>datakernel-bytebuf</artifactId>
    <version>3.1.0</version>
</dependency>

Examples

  1. ByteBuf Example - represents some basic ByteBuf possibilities, such as:
    • wrapping data in ByteBuf for writing/reading,
    • slicing particular parts out of data,
    • conversions.
  2. ByteBuf Pool Example - represents how to work with ByteBufPool.
  3. ByteBuf Queue Example - shows how queues of ByteBufs are created and processed.
Note: To run the examples, you need to clone DataKernel from GitHub:
$ git clone https://github.com/softindex/datakernel
And import it as a Maven project. Check out branch v3.1. Before running the examples, build the project.
These examples are located at datakernel -> examples -> core -> bytebuf.

ByteBuf Example

If you run the example, you’ll receive the following output:

0
1
2
3
4
5

[0, 1, 2, 3, 4, 5]

Hello

Sliced ByteBuf array: [1, 2, 3]

Array of ByteBuf converted from ByteBuffer: [1, 2, 3]
  • The first six lines are a result of wrapping a byte array to a ByteBuf wrapper for reading and then printing it:
byte[] data = new byte[]{0, 1, 2, 3, 4, 5};
ByteBuf byteBuf = ByteBuf.wrapForReading(data);
  • The line [0, 1, 2, 3, 4, 5] is a result of converting an empty array of bytes to ByteBuf and wrapping them for writing. Then the ByteBuf was filled with bytes with the help of while loop:
byte[] data = new byte[6];
ByteBuf byteBuf = ByteBuf.wrapForWriting(data);
byte value = 0;
while (byteBuf.canWrite()) {
	byteBuf.writeByte(value++);
}
  • “Hello” line was first converted from String to ByteBuf and wrapped for reading, then represented as a String for output with the help of byteBuf.asString():
String message = "Hello";
ByteBuf byteBuf = ByteBuf.wrapForReading(message.getBytes(UTF_8));
String unWrappedMessage = byteBuf.asString(UTF_8);
  • The last two outputs represent some other possibilities of ByteBuf, such as slicing:
byte[] data = new byte[]{0, 1, 2, 3, 4, 5};
ByteBuf byteBuf = ByteBuf.wrap(data, 0, data.length);
ByteBuf slice = byteBuf.slice(1, 3);

and conversions of default ByteBuffer to ByteBuf:

ByteBuf byteBuf = ByteBuf.wrap(new byte[20], 0, 0);
ByteBuffer buffer = byteBuf.toWriteByteBuffer();
buffer.put((byte) 1);
buffer.put((byte) 2);
buffer.put((byte) 3);
byteBuf.ofWriteByteBuffer(buffer);
See full example on GitHub

ByteBuf Pool Example

If you run the example, you’ll receive the following output:

Length of array of allocated ByteBuf: 128
Number of ByteBufs in pool before recycling: 0
Number of ByteBufs in pool after recycling: 1
Number of ByteBufs in pool: 0

Size of ByteBuf: 4
Remaining bytes of ByteBuf after 3 bytes have been written: 1
Remaining bytes of a new ByteBuf: 5

[0, 1, 2, 3, 4, 5]

Let’s have a look at the implementation:

public final class ByteBufPoolExample {
	/* Setting ByteBufPool minSize and maxSize properties here for illustrative purposes.
	 Otherwise, ByteBufs with size less than 32 would not be placed into pool
	 */
	static {
		System.setProperty("ByteBufPool.minSize", "1");
	}

	private static void allocatingBufs() {
		// Allocating a ByteBuf of 100 bytes
		ByteBuf byteBuf = ByteBufPool.allocate(100);

		// Allocated ByteBuf has an array with size equal to next power of 2, hence 128
		System.out.println("Length of array of allocated ByteBuf: " + byteBuf.writeRemaining());

		// Pool has 0 ByteBufs right now
		System.out.println("Number of ByteBufs in pool before recycling: " + ByteBufPool.getStats().getPoolItems());

		// Recycling ByteBuf to put it back to pool
		byteBuf.recycle();

		// Now pool consists of 1 ByteBuf that is the one we just recycled
		System.out.println("Number of ByteBufs in pool after recycling: " + ByteBufPool.getStats().getPoolItems());

		// Trying to allocate another ByteBuf
		ByteBuf anotherByteBuf = ByteBufPool.allocate(123);

		// Pool is now empty as the only ByteBuf in pool has just been taken from the pool
		System.out.println("Number of ByteBufs in pool: " + ByteBufPool.getStats().getPoolItems());
		System.out.println();
	}

	private static void ensuringWriteRemaining() {
		ByteBuf byteBuf = ByteBufPool.allocate(3);

		// Size is equal to power of 2 that is larger than 3, hence 4
		System.out.println("Size of ByteBuf: " + byteBuf.writeRemaining());

		byteBuf.write(new byte[]{0, 1, 2});

		// After writing 3 bytes into ByteBuf we have only 1 spare byte in ByteBuf
		System.out.println("Remaining bytes of ByteBuf after 3 bytes have been written: " + byteBuf.writeRemaining());

		// We need to write 3 more bytes so we have to ensure that there are 3 spare bytes in ByteBuf
		// and if there are not - create new ByteBuf with enough room for 3 bytes (old ByteBuf will get recycled)
		ByteBuf newByteBuf = ByteBufPool.ensureWriteRemaining(byteBuf, 3);
		System.out.println("Amount of ByteBufs in pool:" + ByteBufPool.getStats().getPoolItems());

		// As we need to write 3 more bytes, we need a ByteBuf that can hold 6 bytes.
		// The next power of 2 is 8, so considering 3 bytes that have already been written, new ByteBuf
		// can store (8-3=5) more bytes
		System.out.println("Remaining bytes of a new ByteBuf: " + newByteBuf.writeRemaining());

		// Recycling a new ByteBuf (remember, the old one has already been recycled)
		newByteBuf.recycle();
		System.out.println();
	}

	private static void appendingBufs() {
		ByteBuf bufOne = ByteBuf.wrapForReading(new byte[]{0, 1, 2});
		ByteBuf bufTwo = ByteBuf.wrapForReading(new byte[]{3, 4, 5});

		ByteBuf appendedBuf = ByteBufPool.append(bufOne, bufTwo);

		// Appended ByteBuf consists of two ByteBufs, you don't have to worry about allocating ByteBuf
		// with enough capacity or how to properly copy bytes, ByteBufPool will handle it for you
		System.out.println(Arrays.toString(appendedBuf.asArray()));
		System.out.println();
	}

	public static void main(String[] args) {
		allocatingBufs();
		ensuringWriteRemaining();
		appendingBufs();
	}
}
See full example on GitHub

ByteBuf Queue Example

If you run the example, you’ll receive the following output:

bufs:2 bytes:7

Buf taken from queue: [0, 1, 2, 3]

Buf taken from queue: [3, 4, 5, 6, 7, 8]

[1, 2, 3, 4]
[5, 6, 7, 8]
Is queue empty? true

The first line represents our queue after we added two bufs: [0, 1, 2, 3] and [3, 4, 5] with QUEUE.add() method.

QUEUE.add(ByteBuf.wrapForReading(new byte[]{0, 1, 2, 3}));
QUEUE.add(ByteBuf.wrapForReading(new byte[]{3, 4, 5}));

// queue consists of 2 Bufs at this moment

Then method QUEUE.take() is applied and the first added buf, which is [0, 1, 2, 3], is taken from the queue.

The next line represents the result of two operations: adding a new [6, 7, 8] buf and then applying QUEUE.takeRemaining() which takes all remaining bufs from the queue.

// Adding one more ByteBuf to queue
QUEUE.add(ByteBuf.wrapForReading(new byte[]{6, 7, 8}));

ByteBuf takenBuf = QUEUE.takeRemaining();

// Taken ByteBuf is combined of every ByteBuf that were in Queue
Note: pay attention to the difference between take() and poll() ByteBufQueue methods. When using take(), you must be sure that there is at least one ByteBuf remaining in the queue, otherwise use poll() which can return null.

Finally, the last three lines represent the following operations:

  • Creating two bufs: [1, 2, 3, 4] and [5, 6, 7, 8].
  • Draining the queue to the consumer which prints the bufs.
  • Then we check if the queue is empty now.
QUEUE.add(ByteBuf.wrapForReading(new byte[]{1, 2, 3, 4}));
QUEUE.add(ByteBuf.wrapForReading(new byte[]{5, 6, 7, 8}));

// Draining queue to some ByteBuf consumer
QUEUE.drainTo(buf -> System.out.println(Arrays.toString(buf.getArray())));

// Queue is empty after draining
See full example on GitHub