Bytebuf

Overview

ActiveJ strives to make efficient yet high-level I/O. This requires extensive use of user-space byte buffers. Unfortunately, traditional Java ByteBuffers impose a heavy load on GC.

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

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

ByteBuf

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

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

You can read bytes from ByteBuf only when its tail value is greater than the head value. Similarly, you can write bytes to a ByteBuf as log as the tail does not exceed the length of the wrapped array. This eliminated the need for ByteBuffer.flip() operations.

ByteBuf supports concurrent processes: while one process writes some data to the ByteBuf, another process can read it.

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

note

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

ByteBufPool

The ByteBufPool allows ByteBufs to be reused and, as a result, reduces the load on the GC. To make ByteBufPool usage more convenient, there are tools for debugging and monitoring allocated ByteBufs, including their stack traces.

To get a ByteBuf from the pool, use ByteBufPool.allocate(int size). A ByteBuf of size rounded up to the nearest power of 2 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 the ByteBuf.recycle() method. In contrast to languages like C/C++, recycling ByteBufis not required - in the worst case, they will be collected by the GC.

To keep things consistent, ActiveJ relies on the concept of ‘ownership’ (as in Rust language) - once allocated, the components pass byteBuf from one to the other until the last ‘owner’ recycles it to ByteBufPool.

You can explore an example of ByteBufPool use case here

ByteBufs

The ByteBufs class provides efficient management of multiple ByteBufs. It is an optimized queue of multiple ByteBufs with FIFO rules.

You can explore an example of ByteBufs use case here

Utility classes

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

Examples

• ByteBuf Example - represents some basic ByteBuf possibilities, such as:

  • wrapping data in ByteBuf for writing/reading,
  • slicing particular parts out of data,
  • conversions.

• ByteBuf Pool Example - represents how to work with ByteBufPool.

• ByteBufs Example - shows how queues of ByteBufs are created and processed.

note

To run the examples, you need to clone ActiveJ from GitHub

git clone https://github.com/activej/activej

And import it as a Maven project. Check out tag v6.0-beta2. Before running the examples, build the project. These examples are located at activej/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

ByteBufs Example

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

bufs:2 bytes:7

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

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

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

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

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

// bufs consist of 2 Bufs at this moment

Then method BUFS.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 BUFS.takeRemaining() which takes all the remaining bufs from the queue.

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

ByteBuf takenBuf = BUFS.takeRemaining();

// Taken ByteBuf is combined of every ByteBuf that were in bufs

note

Pay attention to the difference between take() and poll() ByteBufs 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.

import io.activej.bytebuf.ByteBuf;
import io.activej.bytebuf.ByteBufs;

import java.util.Arrays;

public final class ByteBufsExample {
  private static final ByteBufs BUFS = new ByteBufs();

  private static void addingToBufs() {
    //[START REGION_1]
    BUFS.add(ByteBuf.wrapForReading(new byte[]{0, 1, 2, 3}));
    BUFS.add(ByteBuf.wrapForReading(new byte[]{3, 4, 5}));

    // bufs consist of 2 Bufs at this moment
    //[END REGION_1]
    System.out.println(BUFS);
    System.out.println();
  }

  private static void takingBufOutOfBufs() {
    ByteBuf takenBuf = BUFS.take();

    // Buf that is taken is the one that was put in bufs first
    System.out.println("Buf taken from bufs: " + Arrays.toString(takenBuf.asArray()));
    System.out.println();
  }

  private static void takingEverythingOutOfBufs() {
    //[START REGION_2]
    // Adding one more ByteBuf to bufs
    BUFS.add(ByteBuf.wrapForReading(new byte[]{6, 7, 8}));

    ByteBuf takenBuf = BUFS.takeRemaining();

    // Taken ByteBuf is combined of every ByteBuf that were in bufs
    //[END REGION_2]
    System.out.println("Buf taken from bufs: " + Arrays.toString(takenBuf.asArray()));
    System.out.println("Is 'ByteBufs' empty? " + BUFS.isEmpty());
    System.out.println();
  }

  public static void main(String[] args) {
    addingToBufs();
    takingBufOutOfBufs();
    takingEverythingOutOfBufs();
  }
}

See full example on GitHub