How Dragonfly Improves Large Collection Replication

Motivation

Replication in Dragonfly must be both correct and efficient, even while data changes in real-time. In earlier versions up to v1.39, ensuring data consistency in replication required a coarse serialization lock on the replication channel to guarantee that concurrent writes do not corrupt the replication data.

Keeping in mind the thread-per-core architecture of Dragonfly, taking a lock while waiting for a network operation to finish is a major performance bottleneck. 

It stalls the other replication tasks which could have been performed by that core while waiting for the IO operation to finish. 

Those tasks need the same lock, but do not work with the same data. This is the specific area where the design has been changed to improve throughput, in particular the P99 latency.

The “tagged Chunks” extension to Dragonfly replication format has been developed to enable fine-grained locking during replication that maintains strict consistency without the performance penalties of coarse locking. 

Concretely, we observed doubling of throughput in our test scenario, along with up to 87 percent drop in P99 latency.

Glossary

• Baseline: The initial set of keys and values existing in the source table when data transfer is initiated.
• Journal: A log of concurrent mutations (commands like additions and deletions) occurring on the primary process during data transfer.
• Tagged Chunks: A wire format enhancement allowing large baseline entries to be split into non-contiguous chunks, wrapped in envelopes for proper reconstruction.
• DFS: The wire format used by Dragonfly for serializing and transferring data snapshots.

Replication Design Before Tagged Chunks

Dragonfly uses a process called snapshotting to transfer data to a replica connecting to a primary instance. During this process the primary Dragonfly process copies both the existing data set as well as any concurrent mutations (additions and deletions of keys) to the replica over a network connection. 

The replication progresses in phases. Once the full data set has been transferred, the connection moves to a steady streaming state, where changes are sent to the replica as they occur.

Baseline And Journal

When initiating a data transfer, Dragonfly considers data to be divided into two categories, baseline and journal. The baseline is the set of keys and values that exist in the source table when the transfer is initiated. 

The primary node also maintains a journal of incremental updates that happened while the baseline values are being copied to the replica.. This journal is streamed to the replica along with the baseline data transfer.

Unlike keys and values, journal entries are commands to be applied on the replica in the correct order to achieve data consistency. 

A critical invariant that we must maintain for data consistency is "baseline before journal". To understand this, consider the following sequence of commands:

• A key X is created before the replica joins. It is thus part of the baseline.
• The replica joins and baseline transfer begins via snapshotting.
• DEL X is issued. The journal command is part of the transfer.

If the DEL X command is received and applied by replica before baseline is sent, this results in resurrection of key X, a correctness bug. Therefore we maintain the invariant that baseline must always precede journal.

This is achieved through hooks built into our mutation path. A process observing a key registers itself within a central list of callbacks, and snapshot process registers itself via such callbacks

An update, such as a DEL command, has to wait until a serialization callback related to that key finishes serializing its baseline value, so the only valid sequence of events is:

• The snapshot is running on baseline data.
• DEL X is issued. It is blocked until the baseline entry is written.
• After baseline is written, DEL X is issued and then the journal command is streamed to the replica.

The critical requirement for correctness therefore is that a journal entry must only be written after the baseline entry for the related key. 

Locking Design Before Tagged Chunks

Before tagged chunks, a single serialized object had to be contiguous in the output stream, however large it may happen to be. Large values could be flushed midway through serialization to keep memory bounded, but even after such a flush the next bytes in the stream still had to be the continuation of the same object.

That created a coarse locking requirement around the shared snapshot serializer:

• A large entry could flush and yield while only partially serialized.
• During that yield, no other fiber could safely write journal data or another entry into the same serializer.
• Otherwise the replica would see another top-level record in the middle of an unfinished object, which the old loader could not parse.
• Holding the lock across flush/backpressure protected correctness, but it also stalled unrelated journal serialization and could increase write latency.

So the old design was correct and memory-bounded, but it held a coarse serialization lock around large values.

Replicating Data With Tagged Chunks

Tagged chunks change the wire format so a large baseline entry no longer has to be contiguous. When an entry is split because of a backpressure driven flush, each emitted piece is wrapped in a small envelope containing a stream id and a payload size. The stream id tells the loader which partial object this chunk belongs to, and the payload size tells it exactly where this chunk ends.

Small entries and raw records, such as journal records, remain unchanged in the wire format.

This allows Dragonfly to flush part of a large value, yield while the network write completes, and later resume that value. In between, other fibers may write to the same stream other baseline entries or journal records, while still preserving baseline-before-journal for the affected key or bucket.

Tagging

We assign a unique id to each data entry (a key-value pair) and attach to chunks a small header, which contains the id and the size of the chunk. The loader uses these identifiers to match the chunk to its corresponding partial entry in memory, reads exactly the size of the chunk, and adds it to the partially read entry.

This format is only applied where needed. For non-data entries in the stream as well as small entries which are never split mid-write, we do not add a tag header, thus saving space and parsing overhead.

Output Stream Management

With the new tagging format, the problem of writing becomes simpler. We identified points in the write path where we can yield control of the fiber, such as network IO, and ensure that we tag the data before sending it to the output stream. 

The in memory buffer now contains data that is either complete and un-tagged, or partial and tagged, or a mix of the two. The parser which runs on the replica is made aware of the format.

Loading Chunks

On the replica, RDB_OPCODE_TAGGED_CHUNK is treated as a top-level envelope around normal RDB object bytes. For the first chunk of a stream, the payload contains the usual object type, key, and beginning of the value. For later chunks, the loader restores the saved stream state and resumes reading the value.

The loader tracks partial objects by stream id. That state includes the database, key, type, object settings, the partially constructed value, and object-specific read progress. The payload size acts as a temporary read budget, forcing the loader to stop exactly at the chunk boundary. If the object is complete, it is applied to the database. Otherwise the continuation state is saved for the next chunk.

Fine-Grained Locking

With tagged chunks, the wire format no longer requires one shard-wide serializer lock to be held across a large value flush. The remaining correctness rule is: a journal record must not be written before the baseline data for the same affected bucket/key.

Now if a large value in one bucket is waiting on network backpressure, journal records or baseline entries for other buckets can still make progress. 

Comparing Performance Numbers

Running the same workload to compare tagged chunks shows a marked improvement. The workload is designed to perform a full sync between a Dragonfly primary and replica. 

Critically, during the sync, write operations are performed in parallel on the primary; to exercise baseline and journal transfer interleaved together.

Structure Of The Benchmark

The benchmark is a pytest based program, which executes Dragonfly processes and interacts with them using the python client.

To compare the performance improvements the following environment is set up which exercises replication and steady state background work:

• A Dragonfly primary
• A single Dragonfly replica

At first the replica is not connected to the primary. The primary is populated with large size sets and each set contains around 200,000 values. Each value in the set is chosen to be large, around 100 bytes. 

A combination of these two ensures that while serializing the baseline, memory usage grows high enough periodically, so that backpressure kicks in and flushes occur regularly.

Once the primary has saved the baseline data, a background workload process is started, it is a simple python program which runs in a loop, it continuously saves simple key value pairs while the benchmark is running. These keys and values are strings and small in size.

The purpose of this background process is to measure latencies observed during saving of these key value operations, which run while replication is in process.

With the baseline in place and background workload running, we execute the REPLICAOF command on the replica, which begins the replication process, and thus the data transfer. 

After the replica is up to date, the background process is stopped. The data we capture during the benchmark is primarily:

• latencies observed, ie time taken per operation by the background worker: p99, p50 and max latency
• throughput: the number of background operations we could perform while the benchmark ran
• preemptions: this is a metric that we expose in Dragonfly, which tells us whether the fiber flushed data during write. For the purpose of this benchmark this is a sanity check, whether the flushes are working as intended

The difference between tagged and non tagged path for this benchmark shows a clearly defined outcome:

• Throughput (operations per second performed during sync) doubles
• P99 latency dropped by around 87 percent.
• The number of write operations completed during full sync improves to around 9x.