No logging

No logging is a test conducted by the benchmark to measure the cost of a no-op log in the situation where the log message is below the threshold at which the logger is currently active. This can be thought of as the experiment’s control group.

Logback format

14:18:17.635 [Name Of Thread] INFO  c.e.NameOfLogger - Log message

The benchmarks used version 1.2.1 of logback.

java.util.logging format

Feb 08, 2017 2:09:19 PM com.example.NameOfLogger nameOfMethod
INFO: Log message

Log4j format

2017-02-08 14:16:29,651 [Name Of Thread] INFO com.example.NameOfLogger - message

The benchmarks used version 2.7 of Log4j.

Measurements

For purposes of comparison the benchmarks have been run on an iMac and on an AWS (Amazon Web Services) EC t2.2xlarge instance (see Tables 14-1 and 14-2). Profiling on macOS can cause problems due to various power-saving techniques, and AWS has the disadvantage that other containers can impact the results of the benchmark. No environment is perfect: there is always noise, and as discussed in Chapter 5, microbenchmarks are full of perils. Hopefully comparison between two datasets of the benchmarks will help uncover useful patterns for guidance in profiling real applications. Remember Feynman’s “you must not fool yourself” principle whenever experimental data needs to be handled.

The Java util logger performs its logging operations at between 42,404 ns and 86,054 ns per operation. The worst-case performance for this logger is using the java.util.logging format—over 2.5× worse than using the same format on Log4j.

Overall Logback offers the best performance from this benchmark run on the iMac, performing best using the Log4j style of logging formatter.

Looking at the benchmarks for AWS, we can see the results show a similar overall pattern to the iMac results. Logback is slightly quicker than Log4j. There are some key takeaways from these results.

AWS, though, has noticeably quicker execution speeds overall, which could be due to power saving kicking in on the iMac or other factors that have not been captured.

Logger results

The benchmarks reveal that there is a range of results depending on the logging format that is used, the logging framework, and the configuration used. Util logging generally offers the worst performance across the board in terms of execution time. The Log4j format appears to give the most consistent results across the board, with Logback executing the log statement in the best time.

In real systems, it is worth testing execution time performance on production kit, especially when the numbers are this close. Real systems are murky, and only the clearest of signals should be taken as evidence of anything underlying; a few tens of percent is not normally enough.

The danger with microbenchmarks, as discussed in Chapter 5, is that looking at the problem in the small can potentially disguise its impact on our application as a whole. We could make a decision based on these microbenchmarks that would influence the application in other unexpected ways.

The amount of garbage generated by a logging framework might be one such consideration, as would the amount of CPU time spent logging instead of processing business-critical parallel tasks. The design of the logging library and the mechanism by which it works are just as important as the straight-line execution results in the microbenchmarks.

Buffers

Richard Warburton has written an excellent article on buffers and the problems with buffers in Java.

Broadly, Java offers a ByteBuffer class, which offers an abstraction for a buffer that is either direct or nondirect.

A direct buffer lives outside of the usual Java heap (but still within the “C heap” of the overall JVM process). As a result, it can often have slower allocation and deallocation rates as compared to an on-heap (nondirect) buffer. The direct buffer has the advantage that the JVM will attempt to invoke instructions directly on that structure without an intermediate mapping.

The main problem with ByteBuffer is the generalized use case, meaning that optimizations specific to the type of buffer are not applied. For example, they don’t support atomic operations, which is a problem if you want to build a producer/consumer-style model across a buffer. ByteBuffer requires that every time you wrap a different structure a new underlying buffer is allocated. In Agrona the copy is avoided, and it supports four types of buffer with different characteristics, allowing you to define and control the interactions possible with each buffer object:

• The DirectBuffer interface provides the ability to read only from the buffer and forms the top level of the hierarchy.

• The MutableDirectBuffer interface extends DirectBuffer and adds write access to the buffer.

• The AtomicBuffer interface extends MutableDirectBuffer, offering ordering behavior.

• UnsafeBuffer is the class that uses Unsafe to implement AtomicBuffer.

Figure 14-3 shows the inheritance hierarchy of buffer classes for Agrona’s buffers.

Figure 14-3. Agrona buffers

The code in Agrona, as you might imagine, is very low-level and uses Unsafe extensively, including in this code fragment:

// This horrible filth is to encourage the JVM to call memset()
// when address is even.
// TODO: check if this still applies when Java 9 is out!!!
UNSAFE.putByte(byteArray, indexOffset, value);
UNSAFE.setMemory(byteArray, indexOffset + 1, length - 1, value);

This is not to say that Agrona is hacky in any way—quite the opposite. The need for this code was to bypass an old optimization that was applied inside the JVM that is now an antioptimization. The library has gone down to this level of detail in ensuring maximum performance can be gained.

Agrona buffers allow access to the underlying data through various get methods—for example, getLong(int index). Even though the buffer is wrapped, it is up to the developer to know at what point the index of their data resides. In addition, put operations allow putting the long value at a specific point on the buffer. Note that the buffer is not of any single type; it is down to the developer to select and manage the appropriate structure for their data. Bounds checking can be enabled/disabled, so dead code can be optimized away by the JIT compiler.

Lists, maps, and sets

Agrona provides a series of list implementations that are backed by arrays of ints or long primitives. As mentioned in “Optimizing Collections”, Java does not have a mechanism of laying out objects side by side in an array, and in standard collections the result is an array of references. Forcing the usage of an object rather than a primitive in standard collections results in autoboxing and unboxing, in addition to the size overhead of the object itself. Agrona also supplies ArrayList utilities that allow fast removal from an ArrayList but spoil the ordering.

Agrona’s map and set implementations store keys and values side by side in a hash table structure. If a collision of keys occurs, the next value is stored immediately after that position in the table. This kind of structure lends itself well to quick access of primitive mappings present on the same cache line.

Queues

Agrona has its own concurrent package that contains useful concurrent utilities and structures including queues and ring buffers.

Queues follow the standard java.util.Queue interface, and they can be used in place of a standard queue implementation. Agrona queues also implement the org.agrona.concurrent.Pipe interface, which adds support for a container processed in sequence. In particular, Pipe adds support for counting, capacity, and draining operations, to interact easily with consumers of the queue. The queues are all lockless and use Unsafe to make them appropriate for use in low-latency systems. The org.agrona.concurrent.AbstractConcurrentArrayQueue provides the first level of support for a series of queues that will provide different producer/consumer models. One interesting part of this API are these classes:

/**
 * Pad out a cache line to the left of a producer fields
 * to prevent false sharing.
 */
class AbstractConcurrentArrayQueuePadding1
{
    @SuppressWarnings("unused")
    protected long p1, p2, p3, p4, p5, p6, p7, p8, p9, p10, p11, p12,
	p13, p14, p15;
}

/**
 * Values for the producer that are expected to be padded.
 */
class AbstractConcurrentArrayQueueProducer
	extends AbstractConcurrentArrayQueuePadding1
{
    protected volatile long tail;
    protected long headCache;
    protected volatile long sharedHeadCache;
}

/**
 * Pad out a cache line between the producer and consumer fields to prevent
 * false sharing.
 */
class AbstractConcurrentArrayQueuePadding2
	extends AbstractConcurrentArrayQueueProducer
{
    @SuppressWarnings("unused")
    protected long p16, p17, p18, p19, p20, p21, p22, p23, p24, p25,
	p26, p27, p28, p29, p30;
}

/**
 * Values for the consumer that are expected to be padded.
 */
class AbstractConcurrentArrayQueueConsumer
	extends AbstractConcurrentArrayQueuePadding2
{
    protected volatile long head;
}

/**
 * Pad out a cache line between the producer and consumer fields to
 * prevent false sharing.
 */
class AbstractConcurrentArrayQueuePadding3
    extends AbstractConcurrentArrayQueuePadding2
{
    @SuppressWarnings("unused")
    protected long p31, p32, p33, p34, p35, p36, p37, p38, p39, p40,
	p41, p42, p43, p44, p45;
}

/**
 * Leftover immutable queue fields.
 */
public abstract class AbstractConcurrentArrayQueue<E>
    extends AbstractConcurrentArrayQueuePadding3
    implements QueuedPipe<E> {...}

The code fragment from AbstractConcurrentArrayQueue shows a clever (forced) arrangement of the queue memory to avoid false sharing when the queue is accessed by the consumer and producer concurrently. The reason that we require this padding is because the layout of fields in memory is not guaranteed by Java or the JVM.

Putting the producer and consumer on separate cache lines ensures that the structure can adequately perform in a low-latency, high-throughput situation. Cache lines are used to access memory, and if the producer and consumer shared the same cache line, problems would occur when the cache line was accessed concurrently.

There are three concrete implementations, and separating out the implementation in this way enables coordination in code only where necessary:

OneToOneConcurrentArrayQueue

If we choose one producer and one consumer, we are opting into a policy that the only concurrent access occurring is when the producer and consumer are accessing the structure at the same time. The main point of interest is the head and tail positions, as these are only updated by one thread at a time.

The head can only be updated by a poll or drain operation to take from the queue, and the tail can only be updated by a put() operation. Selecting this mode avoids unnecessary loss of performance due to additional coordination checks, which are required by the other two types of queue.

ManyToManyConcurrentArrayQueue

Alternatively, if we choose to have many producers, then additional controls are required for the updating of the tail position (as this may have been updated by another producer). Using Unsafe.compareAndSwapLong in a while loop until the tail is updated ensures a lock-free way of updating the queue tail safely. Again, there is no such contention on the consumer side, as we are guaranteeing one consumer.

ManyToOneConcurrentArrayQueue

Finally, if we choose to have many producers and consumers, then coordination controls are needed for the update of the head or the tail. This level of coordination and control is achieved by a while loop wrapping a compareAndSwap. This will require the most coordination of all of the alternatives and so should be used only if that level of safety is required.

Ring buffers

Agrona provides org.agrona.concurrent.RingBuffer as an interface for exchanging binary-encoded messages for interprocess communication. It uses a DirectBuffer to manage the storage of the message off-heap. Thanks to some ASCII art in the source code, we can see that messages are stored in a RecordDescriptor structure:

*   0                   1                   2                   3
*   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
*  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*  |R|                         Length                              |
*  +-+-------------------------------------------------------------+
*  |                            Type                               |
*  +---------------------------------------------------------------+
*  |                       Encoded Message                        ...
* ...                                                              |
*  +---------------------------------------------------------------+

There are two types of ring buffer implemented in Agrona, the OneToOneRingBuffer and the ManyToOneRingBuffer. The write operation takes a source buffer to write a message to the buffer, whereas read operations are called back on an onMessage style of handler. Where many producers are writing in the ManyToOneRingBuffer, a call to Unsafe.storeFence() manually controls memory synchronization. The store fence “ensures lack of reordering of stores before the fence with loads or stores after the fence.”

Agrona has many low-level structures and interesting aspects. If the software system that you are looking to build is at this level of detail, this project is a great first step to experiment with.

There are other projects available, such as JCTools, that also provide variations for concurrent queue structures. Unless a very specific use case is required, developers should not ignore the existence of these open source libraries and should avoid reinventing the wheel.

Copy-free and native type mapping

Copying comes at a cost, and anyone who has done any programming in C++ will probably have been caught out from time to time copying objects accidentally. Copying isn’t expensive when objects are small, but as the size of objects increases so does the cost of the copy.

High-level Java programmers may not always consider this problem, as they are so used to working with references and having memory automatically managed. The copy-free technique in SBE is designed to ensure that no intermediate buffers are used when encoding or decoding messages.

Writing to the underlying buffer directly does, however, come at a design cost: larger messages that cannot fit in the buffer aren’t supported. In order to support them, a protocol would have to be built in to segment the messages and reassemble them.

Types that natively map to sensible assembly instructions also help when you are working with a copy-free design. Having a mapping that corresponds to a good selection of assembly operations dramatically improves the performance of field retrieval.

Steady-state allocation

Allocating objects in Java introduces a natural problem where low-latency applications are concerned. The allocation itself requires CPU cycles (even if it’s very small, such as a TLAB allocation), and then there’s the problem of deleting the object after the usage is complete.

GC is often stop-the-world, which introduces a pause. This is true even of more advanced collectors that work mostly concurrently. Even when limiting the absolute pause time, the process of GC can introduce a meaningful variance in the performance model.

SBE is allocation-free because it uses the flyweight pattern over the underlying buffer.

Streaming access and word-aligned access

Access to memory is something that we would not normally have control over in Java. In “Optimizing Collections”, we discussed ObjectLayout, which is a proposal to store objects in alignment like in C++ vectors. Normally arrays in Java are arrays of references, meaning reading in memory sequentially would not be possible.

SBE is designed to encode and decode messages in a forward progression that also is framed correctly to allow for good word alignment. Without good alignment, performance issues can start to occur at the processor level.

Working with SBE

SBE messages are defined as XML schema files specifying the layout of the messages. Although XML is nowadays quite generally disliked, schemas do give us a good mechanism for specifying a message interface accurately. XML also has instant toolchain support in IDEs such as Eclipse and Intellij.

SBE provides sbe-tool, a command-line tool that, when given a schema, allows you to generate the appropriate encoders and decoders. The steps to get this to work are as follows:

# Fork or clone the project
git clone git@github.com:real-logic/simple-binary-encoding.git

# Build the project using your favorite build tool
gradle

# The sbe-tool will be created in
sbe-tool/build/libs

# Run the sbe-tool with a schema-a sample schema is provided at
# https://github.com/real-logic/simple-binary-encoding/blob/master/
    sbe-samples/src/main/resources/example-schema.xml
java -jar sbe-tool-1.7.5-SNAPSHOT-all.jar message-schema.xml

# When this command completes it will generate a series of .java files in the
    baseline directory
$ ls
BooleanType.java                 GroupSizeEncodingEncoder.java
BoostType.java                   MessageHeaderDecoder.java
BoosterDecoder.java              MessageHeaderEncoder.java
BoosterEncoder.java              MetaAttribute.java
CarDecoder.java                  Model.java
CarEncoder.java                  OptionalExtrasDecoder.java
EngineDecoder.java               OptionalExtrasEncoder.java
EngineEncoder.java               VarStringEncodingDecoder.java
GroupSizeEncodingDecoder.java    VarStringEncodingEncoder.java

It is important to remember that one of the core parts of the SBE protocol is that the messages must be read in order, which means essentially as defined in the schema. The tutorial to start working with these messages is outlined in the SBE Java Users Guide.

Why build something new?

One of the primary reasons for building a new product such as Aeron is that some products in the market have become more general. This is not meant as a criticism, but when customers demand features (and are usually paying for them), it can push a product in a particular direction. The products can become bloated and provide many more features than originally intended, perhaps even becoming frameworks.

Messaging systems in low-level Java are quite fun to build and may start life as a pet project inside a company or in the open source community. The potential problem is that a lack of experience with what is needed from a low-latency perspective can make it hard to take these pet projects and make them production-ready. Ensuring that a product is built for low latency from the ground up without sacrificing performance remains a difficult challenge.

As this chapter has highlighted, the strong design principle behind Aeron is that it is built as a library of components. This means that you are not bound to a framework, and if you only require a low-level data structure then Agrona will provide that, without the need to bring in many other dependencies.

Publishers

Before we discuss Aeron in depth, it is useful to understand some of the higher-level components depicted in Figure 14-4.

Figure 14-4. Architecture overview of Aeron’s major components

Specifically:

• Media refers to the mechanism over which Aeron will communicate; for example, this could be UDP or IPC. It could also be InfiniBand or another medium. The main point is Aeron as a client has an abstraction away from this.

• The media driver refers to the connection between the media and Aeron, allowing configuration to be set and communication to occur with that transport.

• The conductor is responsible for administration, such as setting up the buffers and listening to requests for new subscribers and publishers. It will also look for NAKs (Negative Acknowledgments) and arrange for retransmission. The conductor allows the sender and receiver to just deal with shifting bytes for maximum throughput.

• The sender reads data from producers and sends it out on sockets.

• The receiver reads data from the sockets and forwards it on to the corresponding channel and session.

The media driver is usually a separate process that provides the buffers to be used for transmitting and receiving messages. Different media drivers can be used to exploit optimizations on different hardware deployments; the MediaDriver.Context is the configuration that sets up the optimizations for that media driver. The media driver can also be launched in an embedded way within the same process; the embedded process can be configured with a context or through system properties. Starting up an embedded media driver can be done as follows:

final MediaDriver driver = MediaDriver.launch();

The Aeron applications need to connect to the media driver as either publishers or subscribers. The Aeron class makes this fairly straightforward. Aeron also has an inner Context class that can be used to configure the settings:

final Aeron.Context ctx = new Aeron.Context();

Aeron can then connect a publication to communicate over a given channel and stream. Because a Publication is AutoClosable, it will be automatically cleaned up when the try block finishes executing:

try (Publication publication = aeron.addPublication(CHANNEL, STREAM_ID))
{...}

To send a message a buffer is offered to the publisher, with the result of the offer determining the state of the message. Publication has a series of long constants that represent a mapping of errors that can be compared to the long result from the offer() method:

final long result = publication.offer(BUFFER, 0, messageBytes.length);

Sending a message is as simple as that, but in order for this to be of any use a subscriber should be listening to the same media driver.

Subscribers

The subscriber’s startup is similar to the publisher’s: the media driver needs to be accessed and then the Aeron client connects. The components of a consumer mirror the image shown in Figure 14-4. The consumer registers a callback function that is triggered when a message is received:

final FragmentHandler fragmentHandler =
    SamplesUtil.printStringMessage(STREAM_ID);

try (Aeron aeron = Aeron.connect(ctx);
     Subscription subscription = aeron.addSubscription(CHANNEL, STREAM_ID))
{
    SamplesUtil.subscriberLoop(fragmentHandler,
        FRAGMENT_COUNT_LIMIT, running).accept(subscription);

}

These examples have just explored the basic setup; the Aeron project has samples that explore more advanced examples.

Transport requirements

Aeron is an OSI layer 4 transport for messaging, which means it has a series of responsibilities that it must comply with:

Ordering

Packets will be received out of order from lower-level transports, and it is responsible for reordering out-of-sequence messages.

Reliability

Problems will occur when data is lost, and it must make a request to retransmit that data. While the request for old data is in progress, the process of receiving new data should not be impeded. Reliability in this context refers to connection-level reliability rather than session-level reliability (e.g., fault tolerance over multiple process restarts).

Back pressure

Subscribers will come under pressure under high-volume scenarios, so the service must support flow control and back-pressure measures.

Congestion

This is a problem that can occur on saturated networks, but if a low-latency application is being built it should not be a primary concern. Aeron provides an optional feature to activate congestion control; users that are on low-latency networks can turn it off and users sensitive to other traffic can turn it on. Congestion control can impact a product that in the optimal path has adequate network capacity.

Multiplexing

The transport should be capable of handling multiple streams of information across a single channel without compromising the overall performance.

How it works under the hood

Many existing protocols introduce complicated data structures, such as skip lists, to try to build effective message processing systems. This complexity, mainly due to indirection of pointers, leads to systems that have unpredictable latency characteristics.

Fundamentally Aeron creates a replicated persistent log of messages.

Martin Thompson

Aeron was designed to provide the cleanest and simplest way of building a sequence of messages in a structure. Although it might not seem like the most likely choice initially, Aeron makes significant use of the concept of a file. Files are structures that can be shared across interested processes, and using the memory-mapped file feature of Linux directs all calls to the file to memory rather than a physical file write.

The tail pointer is used to keep track of where the last message was written. Figure 14-5 shows a single message has been written into our current file along with the header.

Figure 14-5. Messages written to a file

The sequencing of events is quite interesting here. The tail pointer reserves the space for the message in the file. The increment of the tail is atomic, and therefore the writer knows the beginning and end of its section.

This allows multiple writers to update the file in a lock-free manner, which is what effectively establishes the file writing protocol. The message is then written, but how can we tell that it’s finished? The header is the last thing to be written atomically to the file. Its presence tells us that the message is complete.

Files are persistent structures that grow when written and do not mutate. Reading records from the file does not require locks, as the file can be opened and read only by an observing process. But is it possible to simply have a file that grows infinitely?

This introduces a lot of problems due to page faults and page churn being introduced in our previously nicely memory-mapped file. We address this concern by having three files: active, dirty, and clean. Active represents the file currently being written to, dirty is the file that was previously written to, and clean is the next file for writing to. The files are rotated to avoid latency caused by bigger files.

Messages are never allowed to go across files. If the tail is pushed off the end of the active file, the insertion process pads the remainder of the file and writes the message into the clean file. From the dirty file it is possible to archive and deep-store the transaction log permanently.

The mechanism for handling missing messages is also really clever and avoids the skip lists and other structures mentioned earlier. The header of the message contains the message’s ordering. When a message is inserted, if it comes out of order a space is left for the previous messages(s). When the missing message is received, it can be inserted in the correct position in the file. This gives an ever-increasing series of messages with no gaps or other structures involved. Ordering data incrementally also has the added benefit of being incredibly quick to work with from a mechanical sympathy perspective.

A watermark represents the current position of the last received message. If the watermark and tail end up being different for a period of time, it indicates missing messages. To resolve the missing messages, NAKs are sent to request them. A NAK can be sent for a message and populated once the message is received.

One interesting side effect of this protocol is that every message received has a unique way of identifying the bytes in each message, based on the streamId, sessionId, termId, and termOffset. The Aeron Archive can be used to record and replay messaging streams. By combining the archive and this unique representation it is possible to uniquely identify all messages throughout the history.

The logfile is at the heart of Aeron’s ability to maintain speed and state. It is also a simple, elegantly executed design that allows the product to compete with—and in many cases beat—well-established (and pricey) multicast products.