= Async Mode
:reftext: Async Mode
:navtitle: Async Mode
:toclevels: 1

== Chronicle Async Mode

Chronicle Async Mode provides a high performance, multi-writer, multi-reader, buffered queue implementation
which can be used in your application code.

Additionally, it allows the underlying buffer to be accessed
through two different implementations of the `ChronicleQueue` interface, which allows it to be used to
mitigate pauses which the OS can introduce when using `ChronicleQueue` under load. Using Enterprise Chronicle
Queue together with Chronicle Async Mode allows a `ChronicleQueue` to be created that behaves in the same way
as the open source `ChronicleQueue` but with an asynchronous buffer behind the scenes covering up any pauses from the
underlying OS and hardware. This can be enabled by setting `readBufferMode` and `writeBufferMode` as specified below.
Tests on real production client workloads have shown that configuring these parameters on a loaded system can
immediately reduce latency outlier events by 98%.

=== The mechanics of Chronicle Async Mode

Chronicle Queue Enterprise Async Mode is a multi-writer, multi-reader, zero-garbage collection, monitored buffered queue
implementation which takes `Bytes`. As the Chronicle stack is built on `Bytes`, you can use the buffer for byte arrays, packed objects, or you can serialise
(or de-serialise objects out of, or into, it using https://github.com/OpenHFT/Chronicle-Wire[Chronicle Wire,window=_blank].

Backing the buffer with a persistent BytesStore (that is, one backed with a memory-mapped file) means it will be
visible to multiple processes. All processes can write to, and read from, the buffer.

==== Common scenarios:
- Back the buffer with an in-memory BytesStore. +
You can have multiple writers and multiple readers in the same process. The buffer does not survive longer than the process.

- Back the buffer with a memory-mapped BytesStore. +
You can have multiple writers and multiple readers in multiple processes. The buffer's data can outlast individual processes.

To use the basic buffer, you must create it by calling `newInstance`.

NOTE: The number of readers must be specified at creation. If not provided, the number of readers is assumed to be `1`. The `BytesStore` must be an
implementation of `NativeBytesStore`, and must be sized appropriately by using the `sizeFor` method. The base size must be a power of `2`.

```
  int size = BytesRingBuffer.sizeFor(2 << 20);
  NativeBytesStore nativeStore = BytesStore.nativeStoreWithFixedCapacity(size);
  BytesRingBuffer ringBuffer = BytesRingBuffer.newInstance(nativeStore)

  int numReaders = 2;
  NativeBytesStore nativeStore2 = BytesStore.nativeStoreWithFixedCapacity(size, numReaders);
  MultiReaderBytesRingBuffer ringBuffer2 = newInstance(nativeStore2, numReaders)
```

When created, it can be written to with `offer` (multiple threads can call `offer` at the same time). If `offer` returns `false` it means that the queue is full, and we are waiting for one or more readers.

```
  boolean succeeded = ringBuffer.offer(bytesToWrite);
```

When `offer` has returned, the data is available for all readers to consume. The `read` method consumes from the asynchronous buffer into the passed `BytesStore`, and returns a success boolean indicating whether anything was read. If the queue is empty, `false` will be returned.

```
  boolean succeeded = ringBuffer.read(bytesToRead);
```

=== Chronicle Async Mode and Chronicle Queue

There are two types of Chronicle Queue which are backed by the asynchronous buffer:

- `software.chronicle.enterprise.queue.EnterpriseSingleChronicleQueue`
- `software.chronicle.enterprise.queue.RingBufferChronicleQueue`

==== software.chronicle.enterprise.queue.EnterpriseSingleChronicleQueue

The asynchronous buffer can be used in one of two ways in EnterpriseSingleChronicleQueue, and this is controlled by
the builder's `readBufferMode` and `writeBufferMode`.

- Asynchronous writes
- Writes and reads using the asynchronous buffer

===== Asynchronous writes

If `writeBufferMode` is set to `BufferMode.Asynchronous` then any writes to the queue will be given to the buffer.
A background task is created in the builder's event loop to drain the buffer in to the underlying (disk-backed)
Chronicle Queue. Reads are made from the underlying queue. This use-case is designed to help with very "bursty" writes,
which cannot afford to be blocked by a slow underlying disk sub-system. The asynchronous buffer size must be carefully
chosen to allow for the maximum burst size.

NOTE: if you want multiple buffered queues to share the same drainer thread you can configure their
`SingleChronicleQueueBuilder`s `eventLoop` to share the same event loop.

====== Example:

```
    try (ChronicleQueue queue = SingleChronicleQueueBuilder
            .binary(path)
            .bufferCapacity(1 << 20)
            .writeBufferMode(BufferMode.Asynchronous)
            .build()) {
        ExcerptAppender appender = queue.acquireAppender();
        ...
        ExcerptTailer tailer = queue.createTailer();
        ...
    }
```

===== Writes and reads using Async Mode

If both `writeBufferMode` and `readBufferMode` are set to `BufferMode.Asynchronous` then any writes to the queue will
be given to the buffer and reads will read from the buffer. This gives the lowest possible latency as the
disk subsystem is not involved at all. As with asynchronous writes, a background task is created in the builder's event
loop to drain the buffer in to the underlying (disk-backed) Chronicle Queue.

====== Example for asynchronous reads and writes using a memory-mapped file and two processes

If you don't need access across processes (actually `ChronicleQueue` instances), then no need to set `SingleChronicleQueueBuilder.asyncBufferCreator`.
The default asyncBufferCreator will create an in-memory `ByteStore`. If you do not set this, and try and access
the buffered queue with multiple `ChronicleQueue` instances then each queue instance will create its own in-memory buffer.

NOTE: This example won't work on Windows as it uses /dev/shm.

[source,Java]
----
    // Common
    final File path = new File(".");
    final File ring = new File("/dev/shm/example.rb");
    final int TAILERS = 5;

    // Appender process
    SingleChronicleQueueBuilder builder = ChronicleQueue.singleBuilder(path);
    try (@NotNull ChronicleQueue queue = builder.bufferCapacity(1 << 20)
            .maxTailers(TAILERS)
            .readBufferMode(BufferMode.Asynchronous)
            .writeBufferMode(BufferMode.Asynchronous)
            .asyncBufferCreator(AsyncBufferCreators.MappedFileDevShm) // share a memory-mapped ring in /dev/shm
            .build()) {
        ExcerptAppender appender = queue.acquireAppender();
        ...
    }

    // Tailer processes - 1..TAILERS 
    SingleChronicleQueueBuilder builder = ChronicleQueue.singleBuilder(path); // same path as process 1
    try (@NotNull ChronicleQueue queue = builder.bufferCapacity(1 << 20)
            .maxTailers(TAILERS)
            .readBufferMode(BufferMode.Asynchronous)
            .writeBufferMode(BufferMode.Asynchronous)
            .asyncBufferCreator(AsyncBufferCreators.MappedFileDevShm) // share a memory-mapped ring in /dev/shm
            .build()) {
        ExcerptTailer tailer = queue.createTailer();
        ...
    }
----

===== Drainer thread
When the queue's event loop is closed, the drainer thread will wait up to 5 seconds to finish draining
to the underlying queue. If draining can not complete, a warning message is logged

===== Unsupported operations
The following operations are unsupported when using EnterpriseSingleChronicleQueue backed by an asynchronous buffer:

* writing and reading of metadata
* `ExcerptTailer.toStart()` and `ExcerptTailer.afterLastWritten()`
* `ExcerptTailer.index()` and `ExcerptTailer.moveToIndex()`

==== software.chronicle.enterprise.queue.RingBufferChronicleQueue

This queue does not implement the full contract of `ChronicleQueue`, and will throw a `UnsupportedOperationException`
from some methods. It does not contain the full functionality of EnterpriseSingleChronicleQueue - it does not
drain to an underlying queue, for example. It is created as in the example below, where:

- `ringBuffer` is the buffer to back this queue with,
- `wireType` is the wire type to use when appending and tailing,
- `pauser` is the pauser to use by the appender when waiting for readers.

```
    ChronicleQueue rbq = new RingBufferChronicleQueue(ringBuffer, wireType, pauser);
```

It can be used in the same way as any other Chronicle Queue implementation, but it is recommended to use
software.chronicle.enterprise.queue.EnterpriseSingleChronicleQueue in preference.

=== Gotchas

The maximum size of a message that can be written to an asynchronous buffer is 1/4 of its capacity. An exception will
be thrown if this is exceeded. The exception may be thrown at either read or write time depending on how
the buffer is written to, and whether asserts are enabled.

=== Licence

Chronicle Async Mode is a licenced product and is licenced separately to chronicle queue enterprise. If you would like further details on Async Mode, please contact sales@chronicle.software
