GFS Replication and Consistency

The Challenge of Concurrent Writes

When a client writes to a file in GFS, the data must be written to all replicas of the affected chunk (typically three). The fundamental challenge is ensuring that these writes are applied in the same order on all replicas. If two clients write to the same chunk concurrently, and replicas apply those writes in different orders, the replicas will diverge, leading to an inconsistent state. A reader might get different data depending on which replica it contacts.

How can GFS enforce a consistent order across all replicas without a complex and slow consensus protocol for every write? The answer lies in a combination of a primary/backup replication scheme and a mechanism called leases.

Leases and a Global Mutation Order

GFS designates one replica of a chunk as the primary. This primary replica is given the authority to choose a serial order for all mutations (writes or appends) to that chunk. All other replicas, called secondaries, must follow the order dictated by the primary.

But how is a replica chosen as the primary, and how do we prevent a "split-brain" scenario where two replicas both believe they are the primary? This is managed by the master using leases.

• The master grants a lease—essentially a temporary delegation of authority—to one replica for a specific chunk.
• A lease has a timeout (e.g., 60 seconds), but the primary can request extensions indefinitely through its regular HeartBeat messages to the master.
• The master guarantees it will not grant another lease for that chunk until the old one expires. This ensures there is only one primary for a given chunk at any time.

This lease mechanism defines a global mutation order: the master first defines the order of leases, and within each lease, the primary defines the serial order of mutations.

Let's walk through the full control and data flow for a write, which brilliantly separates the expensive data transfer from the lightweight ordering logic.

The process unfolds as follows:

1. The client asks the master for the location of the chunk replicas and which one holds the lease (the primary). If no lease exists, the master grants one.
2. The master replies with the identity of the primary and the other secondary replicas. The client caches this information for future writes to the same chunk.
3. Data Flow: The client pushes the data to all replicas. To use the network efficiently, this happens in a pipelined fashion. The client sends the data to the closest chunkserver in the chain, which then forwards it to the next closest, and so on. Each chunkserver stores the incoming data in an internal LRU buffer cache. Crucially, the data has been transferred but not yet applied.
4. Control Flow: Once all replicas acknowledge receiving the data, the client sends a write request to the primary.
5. The primary assigns a consecutive serial number to the mutation, defining its place in the global order. It applies the mutation to its own local state (by writing the data from its buffer to the chunk file).
6. The primary forwards the write request, including the serial number, to all secondary replicas. The secondaries apply the mutation in the same serial number order.
7. The secondaries reply to the primary after completing the operation. Once the primary has received all replies, it finally replies to the client.

If the write fails at any replica, an error is reported to the client. The client library then retries the operation. This can leave the chunk region in an inconsistent state, a point we'll return to shortly.

The GFS Consistency Model: Defined, Undefined, and Inconsistent

Now that we've seen the mechanics of a write, we can discuss the guarantees GFS provides. The model is intentionally relaxed to favor performance and availability for its target workloads.

The GFS paper specifies the state of a file region after a mutation. It's crucial to understand these distinctions, as they directly impact how applications must be written.

• A file region is consistent if all clients see the same data, regardless of which replica they read from.
• A region is defined if it is consistent and clients see what a particular mutation wrote in its entirety.

The outcome of a mutation depends on its type and whether it was concurrent with other mutations:

This model has significant implications. Namespace mutations (e.g., creating a file) are atomic because they are handled exclusively by the master. However, for data mutations, the application must be prepared for these different states. A successful write without concurrency results in a well-behaved defined state. But if concurrent writers succeed, the region is consistent (all replicas are byte-for-byte identical because they applied the same sequence of operations) but undefined (the content is a mix of writes). A failed write is the worst case, leaving the region inconsistent.

Atomic Record Appends

The "consistent but undefined" state for concurrent writes can be difficult for applications to handle. For the common use case of many producers writing to a shared log or result file, GFS provides a more powerful primitive: atomic record append.

With a record append, the client specifies only the data (the "record"). GFS guarantees that it will append this record to the file atomically at least once at an offset of GFS's choosing. It returns this offset to the client.

This operation elegantly solves the concurrent writer problem for producer-consumer queues. Each producer's record is written as a continuous sequence of bytes, without being interleaved with data from other clients.

The operation follows the same lease-based write protocol. The primary handles a special case: if the record doesn't fit in the current last chunk, it pads the chunk to its maximum size and instructs the client to retry on a new chunk. This ensures appends don't straddle chunk boundaries.

The "at-least-once" semantic means that if a client retries a record append after a timeout or error, the record may be written more than once. The regions containing successfully written records are defined, but the padding or duplicated data from retries are considered inconsistent. Therefore, consuming applications must be prepared to handle padding and potential duplicates, for example by using unique IDs within the records for deduplication.

Replication and Durability

Finally, let's solidify our understanding of the replication strategy. We know GFS defaults to three replicas, but where are they placed?

To protect against correlated failures (like a rack's power supply or network switch failing), GFS is rack-aware. It aims to store replicas of a chunk on at least two different racks. This maximizes data availability and reliability. The master, with its global view of the cluster topology, is responsible for making these intelligent placement decisions.

A critical piece of the consistency and durability story is how GFS handles replicas that miss mutations because their chunkserver was temporarily down. Such a replica is considered stale.

GFS uses a chunk version number to detect stale replicas.

• The master increments the version number for a chunk every time it grants a new lease.
• This new version number is persisted by the master and the up-to-date replicas.
• When a chunkserver that was down restarts, it reports the version numbers of its chunks to the master.
• If the master sees a version number lower than its current record for a chunk, it knows the replica is stale.
• Stale replicas are not included in responses to client requests and are eventually garbage collected. The master will also instruct another chunkserver to create a fresh replica from a valid one to restore the replication factor.

This mechanism ensures that clients are not given the locations of stale replicas and are always directed to up-to-date data.

Conclusion

In this lesson, we have explored the sophisticated mechanisms GFS uses to manage writes, concurrency, and replication. You now have a deep understanding of the trade-offs at the heart of its design.

The key takeaways are:

• Leases for Ordering: GFS uses a primary-secondary model where the master grants leases to a primary replica, which then defines a serial order for all mutations to a chunk.
• Decoupled Data/Control Flow: The expensive transfer of data is pipelined directly between chunkservers, separate from the lightweight control messages that order and commit the write.
• Relaxed Consistency: GFS provides a relaxed consistency model. Applications must be designed to handle defined, consistent but undefined, and inconsistent data states that can arise from serial, concurrent, or failed writes.
• Atomic Record Appends: For the common many-producer use case, GFS offers a specialized atomic append operation that guarantees atomicity at least once, simplifying application logic.
• Durable Replication: Replication is rack-aware to tolerate rack-level failures, and chunk version numbers are used to detect and isolate stale replicas, preventing clients from reading outdated data.

We have now covered the GFS architecture, its read path, and its write path. The final piece of the puzzle is understanding how the system as a whole recovers from the inevitable failures of its components. In our next lesson, we will analyze GFS fault recovery mechanisms for master and chunkserver failures, completing our study of this seminal system.