Chronicle Queue

Chronicle Queue assumes disk space is cheap compared with memory. Chronicle Queue makes full use of the disk space you have, and so you are not limited by the main memory of your machine. If you use spinning HDD, you can store many TBs of disk space for little cost.

The only extra software that Chronicle Queue needs to run is the operating system. It doesn’t have a broker; instead it uses your operating system to do all the work. If your application dies, the operating system keeps running for seconds longer, so no data is lost; even without replication.

As Chronicle Queue stores all saved data in memory-mapped files, this has a trivial on-heap overhead, even if you have over 100 TB of data.

Chronicle put significant effort into achieving very low latency. In other products which focus on support of web applications, latencies of less than 40 milliseconds are fine as they are faster than you can see; for example, the frame rate of cinema is 24 Hz, or about 40 ms.

Chronicle Queue aims to achieve latencies of under 40 microseconds for 99% to 99.99% of the time. Using Chronicle Queue without replication, we support applications with latencies below 40 microseconds end-to-end across multiple services. Often the 99% latency of Chronicle Queue is entirely dependent on the choice of operating system and hard disk sub-system.

Replication for Chronicle Queue supports Chronicle Wire Enterprise. This supports a real-time compression which calculates the deltas for individual objects, as they are written. This can reduce the size of messages by a factor of 10, or better, without the need for batching; that is, without introducing significant latency.

Chronicle Queue also supports LZW, Snappy, and GZIP compression. These formats however add significant latency. These are only useful if you have strict limitations on network bandwidth.

Chronicle Queue supports a number of semantics:

• Every message is replayed on restart.

• Only new messages are played on restart.

• Restart from any known point using the index of the entry.

• Replay only the messages you have missed. This is supported directly using the methodReader/methodWriter builders.

On most systems System.nanoTime() is roughly the number of nanoseconds since the system last rebooted (although different JVMs may behave differently). This is the same across JVMs on the same machine, but wildly different between machines. The absolute difference when it comes to machines is meaningless. However, the information can be used to detect outliers; you can’t determine what the best latency is, but you can determine how far off the best latencies you are. This is useful if you are focusing on the 99th percentile latencies. We have a class called RunningMinimum to obtain timings from different machines, while compensating for a drift in the nanoTime between machines. The more often you take measurements, the more accurate this running minimum is.

Chronicle Queue manages storage by cycle. You can add a StoreFileListener which will notify you when a file is added, and when it is no longer retained. You can move, compress, or delete all the messages for a day, at once. NOTE : Unfortunately on Windows, if an IO operation is interrupted, it can close the underlying FileChannel.

Due to performance reasons, we have removed checking for interrupts in the chronicle queue code. Because of this, we recommend that you avoid using chronicle queue with code that generates interrupts. If you can not avoid generating interrupts then we suggest that you create a separate instance of Chronicle Queue per thread.

Most messaging systems are consumer-centric. Flow control is implemented to avoid the consumer ever getting overloaded; even momentarily. A common example is a server supporting multiple GUI users. Those users might be on different machines (OS and hardware), different qualities of network (latency and bandwidth), doing a variety of other things at different times. For this reason it makes sense for the client consumer to tell the producer when to back off, delaying any data until the consumer is ready to take more data.

Chronicle Queue is a producer-centric solution and does everything possible to never push back on the producer, or tell it to slow down. This makes it a powerful tool, providing a big buffer between your system, and an upstream producer over which you have little, or no, control.

Market data publishers don’t give you the option to push back on the producer for long; if at all. A few of our users consume data from CME OPRA. This produces peaks of 10 million events per minute, sent as UDP packets without any retry. If you miss, or drop a packet, then it is lost. You have to consume and record those packets as fast as they come to you, with very little buffering in the network adapter. For market data in particular, real time means in a few microseconds; it doesn’t mean intra-day (during the day).

Chronicle Queue is fast and efficient, and has been used to increase the speed that data is passed between threads. In addition, it also keeps a record of every message passed allowing you to significantly reduce the amount of logging that you need to do.

Compliance systems are required by more and more systems these days. Everyone has to have them, but no one wants to be slowed down by them. By using Chronicle Queue to buffer data between monitored systems and the compliance system, you don’t need to worry about the impact of compliance recording for your monitored systems. Again, Chronicle Queue can support millions of events per-second, per-server, and access data which has been retained for years.

Chronicle Queue supports low latency IPC (Inter Process Communication) between JVMs on the same machine in the order of magnitude of 1 microsecond; as well as between machines with a typical latency of 10 microseconds for modest throughputs of a few hundred thousands. Chronicle Queue supports throughputs of millions of events per second, with stable microsecond latencies.

See Articles on the use of Chronicle Queue in Microservices

A Chronicle Queue can be used to build state machines. All the information about the state of those components can be reproduced externally, without direct access to the components, or to their state. This significantly reduces the need for additional logging. However, any logging you do need can be recorded in great detail. This makes enabling DEBUG logging in production practical. This is because the cost of logging is very low; less than 10 microseconds. Logs can be replicated centrally for log consolidation. Chronicle Queue is being used to store 100+ TB of data, which can be replayed from any point in time.

Non-batching streaming components are highly performant, deterministic, and reproducible. You can reproduce bugs which only show up after a million events played in a particular order, with accelerated realistic timings. This makes using Stream processing attractive for systems which need a high degree of quality outcomes.

Chronicle Queue (CQ) can be configured via a number of methods on the SingleChronicleQueueBuilder class. A few of the parameters that were most queried by our customers are explained below .

• RollCycle

The RollCycle parameter configures the rate at which CQ will roll the underlying queue files. For instance, using the following code snippet will result in the queue files being rolled (i.e. a new file created) every hour:

Once a queue’s roll-cycle has been set, it cannot be changed at a later date. Any further instances of SingleChronicleQueue configured to use the same path should be configured to use the same roll-cycle, and if they are not, then the roll-cycle will be updated to match the persisted roll-cycle. In this case, a warning log message will be printed in order to notify the library user of the situation:

console output:

The maximum number of messages that can be stored in a queue file depends on roll cycle. See FAQ for more information on this.

In Chronicle Queue, the rollover time is based on UTC. The Timezone Rollover Enterprise feature extends Chronicle Queue’s ability to specify the time and periodicity of queue rollovers, rather than UTC. For more information see Timezone Queue Rollover.

The Chronicle Queue FileUtil class provides useful methods for managing queue files. See Managing Roll Files Directly.

• wireType

It’s possible to configure how Chronicle Queue will store the data by explicitly set the WireType:

For example:

Although it’s possible to explicitly provide WireType when creating a builder, it is discouraged as not all wire types are supported by Chronicle Queue yet. In particular, the following wire types are not supported:

• TEXT (and essentially all based on text, including JSON and CSV)

• RAW

• READ_ANY

• blockSize

When a queue is read/written, part of the file currently being read/written is mapped to a memory segment. This parameter controls the size of the memory mapping block. You can change this parameter using the method SingleChronicleQueueBuilder.blockSize(long blockSize) if it is necessary.

If you are sending large messages then you should set a large blockSize i.e. the blockSize should be at least four times the message size.

We recommend that you use the same blockSize for each queue instance when replicating queues, the blockSize is not written to the queue’s metadata, so should ideally be set to the same value when creating your instances of chronicle queue (this is recommended but if you wish to run with a different blocksize you can).

• indexSpacing

This parameter shows the space between excerpts that are explicitly indexed. A higher number means higher sequential write performance but slower random access read. The sequential read performance is not affected by this property. For example, the following default index spacing can be returned:

• 16 (MINUTELY)

• 64 (DAILY)

You can change this parameter using the method SingleChronicleQueueBuilder.indexSpacing(int indexSpacing).

• indexCount

The size of each index array, as well as the total number of index arrays per queue file.

• readBufferMode, writeBufferMode

These parameters define BufferMode for reads or writes that have the following options:

• None - The default (and the only one available for open source users), no buffering;

• Copy - used in conjunction with encryption;

• Asynchronous - use an asynchronous buffer when reading and/or writing, provided by Chronicle Async Mode.

• bufferCapacity

RingBuffer capacity in bytes when using bufferMode: Asynchronous

In Chronicle Queue we refer to the act of writing your data to the Chronicle Queue, as storing an excerpt. This data could be made up from any data type, including text, numbers, or serialised blobs. Ultimately, all your data, regardless of what it is, is stored as a series of bytes.

Just before storing your excerpt, Chronicle Queue reserves a 4-byte header. Chronicle Queue writes the length of your data into this header. This way, when Chronicle Queue comes to read your excerpt, it knows how long each blob of data is. We refer to this 4-byte header, along with your excerpt, as a document. Strictly speaking Chronicle Queue can be used to read and write documents.

As stated before, Chronicle Queue uses an appender to write to the queue and a tailer to read from the queue. Unlike other java queuing solutions, messages are not lost when they are read with a tailer. This is covered in more detail in the section below on "Reading from a queue using a tailer". To write data to a Chronicle Queue, you must first create an appender:

Chronicle Queue uses the following low-level interface to write the data:

The close on the try-with-resources, is the point when the length of the data is written to the header. You can also use the DocumentContext to find out the index that your data has just been assigned (see below). You can later use this index to move-to/look up this excerpt. Each Chronicle Queue excerpt has a unique index.

The high-level methods below such as writeText() are convenience methods on calling appender.writingDocument(), but both approaches essentially do the same thing. The actual code of writeText(CharSequence text) looks like this:

So you have a choice of a number of high-level interfaces, down to a low-level API, to raw memory.

This is the highest-level API which hides the fact you are writing to messaging at all. The benefit is that you can swap calls to the interface with a real component, or an interface to a different protocol.

You can write a "self-describing message". Such messages can support schema changes. They are also easier to understand when debugging or diagnosing problems.

You can write "raw data" which is self-describing. The types will always be correct; position is the only indication as to the meaning of those values.

You can write "raw data" which is not self-describing. Your reader must know what this data means, and the types that were used.

Below, the lowest level way to write data is illustrated. You get an address to raw memory and you can write whatever you want.

You can print the contents of the queue. You can see the first two, and last two messages store the same data.

prints:

Reading the queue follows the same pattern as writing, except there is a possibility there is not a message when you attempt to read it.

You can turn each message into a method call based on the content of the message, and have Chronicle Queue automatically deserialize the method arguments. Calling reader.readOne() will automatically skip over (filter out) any messages that do not match your method reader.

You can decode the message yourself.

You can read self-describing data values. This will check the types are correct, and convert as required.

You can read raw data as primitives and strings.

or, you can get the underlying memory address and access the native memory.

An abstraction can be added to filter messages, or assign messages to just one message processor. However, in general you only need one main tailer for a topic, with possibly, some supporting tailers for monitoring etc.

As Chronicle Queue doesn’t partition its topics, you get total ordering of all messages within that topic. Across topics, there is no guarantee of ordering; if you want to replay deterministically from a system which consumes from multiple topics, we suggest replaying from that system’s output.

Chronicle Queue tailers may create file handlers, the file handlers are cleaned up whenever the associated chronicle queue’s close() method is invoked or whenever the Jvm runs a Garbage Collection. If you are writing your code not have GC pauses and you explicitly want to clean up the file handlers, you can call the following:

In some applications, it may be necessary to start reading from the end of the queue (e.g. in a restart scenario). For this use-case, ExcerptTailer provides the toEnd() method. When the tailer direction is FORWARD (by default, or as set by the ExcerptTailer.direction method), then calling toEnd() will place the tailer just after the last existing record in the queue. In this case, the tailer is now ready for reading any new records appended to the queue. Until any new messages are appended to the queue, there will be no new DocumentContext available for reading:

If it is necessary to read backwards through the queue from the end, then the tailer can be set to read backwards:

When reading backwards, then the toEnd() method will move the tailer to the last record in the queue. If the queue is not empty, then there will be a DocumentContext available for reading:

AKA named tailers.

It can be useful to have a tailer which continues from where it was up to on restart of the application.

1. Tailer "a" last reads message 2

2. Tailer "a" next reads message 3

3. Tailer "b" last reads message 0

4. Tailer "b" next reads message 1

This is from the RestartableTailerTest where there are two tailers, each with a unique name. These tailers store their index within the Queue itself and this index is maintained as the tailer uses toStart(), toEnd(), moveToIndex() or reads a message.

Chronicle Queue stores its data in binary format, with a file extension of cq4:

This can often be a bit difficult to read, so it is better to dump the cq4 files as text. This can also help you fix your production issues, as it gives you the visibility as to what has been stored in the queue, and in what order.

You can dump the queue to the terminal using net.openhft.chronicle.queue.main.DumpMain or net.openhft.chronicle.queue.ChronicleReaderMain. DumpMain performs a simple dump to the terminal while ChronicleReaderMain handles more complex operations, e.g. tailing a queue. They can both be run from the command line in a number of ways described below.

If you have a project pom file that includes the Chronicle-Queue artifact, you can read a cq4 file with the following command:

In the above command myqueue is the directory containing your .cq4 files

You can also set up any dependent files manually. This requires the chronicle-queue.jar, from any version 4.5.3 or later, and that all dependent files are present on the class path. The dependent jars are listed below:

Once the dependencies are present on the class path, you can run:

This will dump the 19700101-02.cq4 file out as text, as shown below:

There is also a script named dump_queue.sh located in the Chonicle-Queue/bin-folder that gathers the needed dependencies in a shaded jar and uses it to dump the queue with DumpMain. The script can be run from the Chronicle-Queue root folder like this:

The second tool for logging the contents of the chronicle queue is the ChronicleReaderMain (in the Chronicle Queue project). As mentioned above, it is able to perform several operations beyond printing the file content to the console. For example, it can be used to tail a queue to detect whenever new messages are added (rather like $tail -f).

Below is the command line interface used to configure ChronicleReaderMain:

Just as with DumpQueue you need the classes in the example above present on the class path. This can again be achieved by manually adding them and then run:

Another option is to create an Uber Jar using the Maven shade plugin. It is configured as follows:

Once the Uber jar is present, you can run ChronicleReaderMain from the command line via:

Lastly, there is a script for running the reader named queue_reader.sh which again is located in the Chonicle-Queue/bin-folder. It automatically gathers the needed dependencies in a shaded jar and uses it to run ChronicleReaderMain. The script can be run from the Chronicle-Queue root folder like this:

If using MethodReader and MethodWriter then you can write single-argument method calls to a queue using net.openhft.chronicle.queue.ChronicleWriterMain or the shell script queue_writer.sh e.g.

If you want to write to the below "doit" method

Then you can call ChronicleWriterMain -d queue doit x.yaml with either (or both) of the below Yamls:

or

If DTO makes use of custom serialisation then you should specify the interface to write to with -i

Chronicle v4.4+ supports the use of proxies to write and read messages. You start by defining an asynchronous interface, where all methods have:

• arguments which are only inputs

• no return value or exceptions expected.

To write to the queue you can call a proxy which implements this interface.

These calls produce messages which can be dumped as follows.

To read the messages, you can provide a reader which calls your implementation with the same calls that you made.

Running this example prints:

• For more details see, Using Method Reader/Writers and MessageReaderWriterTest

Chronicle Queue supports explicit, or implicit, nanosecond resolution timing for messages as they pass end-to-end over across your system. We support using nano-time across machines, without the need for specialist hardware. To enable this, set the sourceId of the queue.

A timestamp is added for each read and write as it passes from service to service.

1. First write

2. First read

3. Write of the result of the read.

4. What triggered this event.

In the following section you will find how to work with the excerpt index.

• Finding the index at the end of a Chronicle Queue

Chronicle Queue appenders are thread-local. In fact when you ask for:

the acquireAppender() uses a thread-local pool to give you an appender which will be reused to reduce object creation. As such, the method call to:

will only give you the last index appended by this appender; not the last index appended by any appender. If you wish to find the index of the last record written to the queue, then you have to call:

Which will return the index of the last excerpt present in the queue (or -1 for an empty queue). Note that if the queue is being written to concurrently it’s possible the value may be an under-estimate, as subsequent entries may have been written even before it was returned.

• The number of messages between two indexes

To count the number of messages between two indexes you can use:

for more information on this see :

• Move to a specific message and read it

The following example shows how to write 10 messages, then move to the 5th message to read it

You can add a StoreFileListener to notify you when a file is added, or no longer used. This can be used to delete files after a period of time. However, by default, files are retained forever. Our largest users have over 100 TB of data stored in queues.

Appenders and tailers are cheap as they don’t even require a TCP connection; they are just a few Java objects. The only thing each tailer retains is an index which is composed from:

• a cycle number. For example, days since epoch, and

• a sequence number within that cycle.

In the case of a DAILY cycle, the sequence number is 32 bits, and the index = ((long) cycle << 32) | sequenceNumber providing up to 4 billion entries per day. if more messages per day are anticipated, the XLARGE_DAILY cycle, for example, provides up 4 trillion entries per day using a 48-bit sequence number. Printing the index in hexadecimal is common in our libraries, to make it easier to see these two components.

Rather than partition the queue files across servers, we support each server, storing as much data as you have disk space. This is much more scalable than being limited to the amount of memory space that you have. You can buy a redundant pair of 6TB of enterprise disks very much more cheaply than 6TB of memory.

Chronicle Queue runs a background thread to watch for low disk space (see net.openhft.chronicle.threads.DiskSpaceMonitor class) as the JVM can crash when allocating a new memory mapped file if disk space becomes low enough. The disk space monitor checks (for each FileStore you are using Chronicle Queues on): that there is less than 200MB free. If so you will see:

otherwise it will check for the threshold percentage and log out this message:

The threshold percentage is controlled by the chronicle.disk.monitor.threshold.percent system property. The default value is 0.

As mentioned previously Chronicle Queue stores its data off-heap in a ‘.cq4’ file. So whenever you wish to append data to this file or read data into this file, chronicle queue will create a file handle . Typically, Chronicle Queue will create a new ‘.cq4’ file every day. However, this could be changed so that you can create a new file every hour, every minute or even every second.

If we create a queue file every second, we would refer to this as SECONDLY rolling. Of course, creating a new file every second is a little extreme, but it’s a good way to illustrate the following point. When using secondly rolling, If you had written 10 seconds worth of data and then you wish to read this data, chronicle would have to scan across 10 files. To reduce the creation of the file handles, chronicle queue cashes them lazily and when it comes to writing data to the queue files, care-full consideration must be taken when closing the files, because on most OS’s a close of the file, will force any data that has been appended to the file, to be flushed to disk, and if we are not careful this could stall your application.

Pretoucher is a class designed to be called from a long-lived thread. The purpose of the Pretoucher is to accelerate writing in a queue. Upon invocation of the execute() method, this object will pre-touch pages in the queue’s underlying store file, so that they are resident in the page-cache (i.e. loaded from storage) before they are required by appenders to the queue. Resources held by this object will be released when the underlying queue is closed. Alternatively, the shutdown() method can be called to close the supplied queue and release any other resources. Invocation of the execute() method after shutdown() has been called will cause an IllegalStateException to be thrown.

The Pretoucher’s configuration parameters (set via the system properties) are as follows:

• SingleChronicleQueueExcerpts.earlyAcquireNextCycle (defaults to false): Causes the Pretoucher to create the next cycle file while the queue is still writing to the current one in order to mitigate the impact of stalls in the OS when creating new files.

• SingleChronicleQueueExcerpts.pretoucherPrerollTimeMs (defaults to 2,000 milliseconds) The pretoucher will create new cycle files this amount of time in advanced of them being written to. Effectively moves the Pretoucher’s notion of which cycle is "current" into the future by pretoucherPrerollTimeMs.

• SingleChronicleQueueExcerpts.dontWrite (defaults to false): Tells the Pretoucher to never create cycle files that do not already exist. As opposed to the default behaviour where if the Pretoucher runs inside a cycle where no excerpts have been written, it will create the "current" cycle file. Obviously enabling this will prevent earlyAcquireNextCycle from working.

  • Pretoucher usage example

The configuration parameters of Pretoucher that were described above should be set via system properties. For example, in the following excerpt earlyAcquireNextCycle is set to true and pretoucherPrerollTimeMs to 100ms.

The constructor of Pretoucher takes the name of the queue that this Pretoucher is assigned to and creates a new Pretoucher. Then, by invoking the execute() method the Pretoucher starts.

The method close(), closes the Pretoucher and releases its resources.

28	at java.util.concurrent.ConcurrentHashMap.putVal(ConcurrentHashMap.java:1012)
	at java.util.concurrent.ConcurrentHashMap.put(ConcurrentHashMap.java:1006)
	at net.openhft.chronicle.core.util.WeakReferenceCleaner.newCleaner(WeakReferenceCleaner.java:43)
	at net.openhft.chronicle.bytes.NativeBytesStore.<init>(NativeBytesStore.java:90)
	at net.openhft.chronicle.bytes.MappedBytesStore.<init>(MappedBytesStore.java:31)
	at net.openhft.chronicle.bytes.MappedFile$$Lambda$4/1732398722.create(Unknown Source)
	at net.openhft.chronicle.bytes.MappedFile.acquireByteStore(MappedFile.java:297)
	at net.openhft.chronicle.bytes.MappedFile.acquireByteStore(MappedFile.java:246)

25	at net.openhft.chronicle.queue.jitter.QueueWriteJitterMain.lambda$main$1(QueueWriteJitterMain.java:58)
	at net.openhft.chronicle.queue.jitter.QueueWriteJitterMain$$Lambda$11/967627249.run(Unknown Source)
	at java.lang.Thread.run(Thread.java:748)

21	at java.util.concurrent.ConcurrentHashMap.putVal(ConcurrentHashMap.java:1027)
	at java.util.concurrent.ConcurrentHashMap.put(ConcurrentHashMap.java:1006)
	at net.openhft.chronicle.core.util.WeakReferenceCleaner.newCleaner(WeakReferenceCleaner.java:43)
	at net.openhft.chronicle.bytes.NativeBytesStore.<init>(NativeBytesStore.java:90)
	at net.openhft.chronicle.bytes.MappedBytesStore.<init>(MappedBytesStore.java:31)
	at net.openhft.chronicle.bytes.MappedFile$$Lambda$4/1732398722.create(Unknown Source)
	at net.openhft.chronicle.bytes.MappedFile.acquireByteStore(MappedFile.java:297)
	at net.openhft.chronicle.bytes.MappedFile.acquireByteStore(MappedFile.java:246)

14	at net.openhft.chronicle.queue.jitter.QueueWriteJitterMain.lambda$main$1(QueueWriteJitterMain.java:54)
	at net.openhft.chronicle.queue.jitter.QueueWriteJitterMain$$Lambda$11/967627249.run(Unknown Source)
	at java.lang.Thread.run(Thread.java:748)

• Chronicle support on StackOverflow

• Chronicle support on Google Groups