Nov. 8, 2021
Here is the link.
I did spend over two hours to read the paper. Here is the blog.
2.6.3 Operation Log
The operation log contains a historical record of critical
metadata changes. It is central to GFS. Not only is it the
only persistent record of metadata, but it also serves as a
logical time line that defines the order of concurrent operations. Files and chunks, as well as their versions (see
Section 4.5), are all uniquely and eternally identified by the
logical times at which they were created.
Since the operation log is critical, we must store it reliably and not make changes visible to clients until metadata
changes are made persistent. Otherwise, we effectively lose
the whole file system or recent client operations even if the
chunks themselves survive. Therefore, we replicate it on
multiple remote machines and respond to a client operation only after flushing the corresponding log record to disk
both locally and remotely. The master batches several log
records together before flushing thereby reducing the impact
of flushing and replication on overall system throughput.
The master recovers its file system state by replaying the
operation log. To minimize startup time, we must keep the
log small.
The master checkpoints its state whenever the log
grows beyond a certain size so that it can recover by loading
the latest checkpoint from local disk and replaying only the limited number of log records after that. The checkpoint is
in a compact B-tree like form that can be directly mapped
into memory and used for namespace lookup without extra parsing. This further speeds up recovery and improves
availability.
Because building a checkpoint can take a while, the master’s internal state is structured in such a way that a new
checkpoint can be created without delaying incoming mutations. The master switches to a new log file and creates the
new checkpoint in a separate thread. The new checkpoint
includes all mutations before the switch. It can be created
in a minute or so for a cluster with a few million files. When
completed, it is written to disk both locally and remotely.
Recovery needs only the latest complete checkpoint and
subsequent log files. Older checkpoints and log files can
be freely deleted, though we keep a few around to guard
against catastrophes. A failure during checkpointing does
not affect correctness because the recovery code detects and
skips incomplete checkpoints.
2.7 Consistency Model
GFS has a relaxed consistency model that supports our
highly distributed applications well but remains relatively
simple and efficient to implement. We now discuss GFS’s
guarantees and what they mean to applications. We also
highlight how GFS maintains these guarantees but leave the
details to other parts of the paper.
2.7.1 Guarantees by GFS
File namespace mutations (e.g., file creation) are atomic.
They are handled exclusively by the master: namespace
locking guarantees atomicity and correctness (Section 4.1);
the master’s operation log defines a global total order of
these operations (Section 2.6.3).
The state of a file region after a data mutation depends
on the type of mutation, whether it succeeds or fails, and
whether there are concurrent mutations. Table 1 summarizes the result. A file region is consistent if all clients will
always see the same data, regardless of which replicas they
read from. A region is defined after a file data mutation if it
is consistent and clients will see what the mutation writes in
its entirety. When a mutation succeeds without interference
from concurrent writers, the affected region is defined (and
by implication consistent): all clients will always see what
the mutation has written. Concurrent successful mutations
leave the region undefined but consistent: all clients see the
same data, but it may not reflect what any one mutation
has written. Typically, it consists of mingled fragments from
multiple mutations. A failed mutation makes the region inconsistent (hence also undefined): different clients may see
different data at different times. We describe below how our
applications can distinguish defined regions from undefined regions. The applications do not need to further distinguish
between different kinds of undefined regions.
Data mutations may be writes or record appends. A write
causes data to be written at an application-specified file
offset. A record append causes data (the “record”) to be
appended atomically at least once even in the presence of
concurrent mutations, but at an offset of GFS’s choosing
(Section 3.3). (In contrast, a “regular” append is merely a
write at an offset that the client believes to be the current
end of file.) The offset is returned to the client and marks
the beginning of a defined region that contains the record.
In addition, GFS may insert padding or record duplicates in
between. They occupy regions considered to be inconsistent
and are typically dwarfed by the amount of user data.
After a sequence of successful mutations, the mutated file
region is guaranteed to be defined and contain the data written by the last mutation. GFS achieves this by (a) applying
mutations to a chunk in the same order on all its replicas
(Section 3.1), and (b) using chunk version numbers to detect
any replica that has become stale because it has missed mutations while its chunkserver was down (Section 4.5). Stale
replicas will never be involved in a mutation or given to
clients asking the master for chunk locations. They are
garbage collected at the earliest opportunity.
Since clients cache chunk locations, they may read from a
stale replica before that information is refreshed. This window is limited by the cache entry’s timeout and the next
open of the file, which purges from the cache all chunk information for that file. Moreover, as most of our files are
append-only, a stale replica usually returns a premature
end of chunk rather than outdated data. When a reader
retries and contacts the master, it will immediately get current chunk locations.
Questions:
- Checksumming - what is to be used for?
- irreversibly - Google translate -
Long after a successful mutation, component failures can
of course still corrupt or destroy data. GFS identifies failed
chunkservers by regular handshakes between master and all
chunkservers and detects data corruption by checksumming
(Section 5.2). Once a problem surfaces, the data is restored
from valid replicas as soon as possible (Section 4.3). A chunk
is lost irreversibly only if all its replicas are lost before GFS
can react, typically within minutes. Even in this case, it becomes unavailable, not corrupted: applications receive clear
errors rather than corrupt data.
2.7.2 Implications for Applications
GFS applications can accommodate the relaxed consistency model with a few simple techniques already needed for
other purposes: relying on appends rather than overwrites,
checkpointing, and writing self-validating, self-identifying
records.
Practically all our applications mutate files by appending
rather than overwriting. In one typical use, a writer generates a file from beginning to end. It atomically renames the
file to a permanent name after writing all the data, or periodically checkpoints how much has been successfully written. Checkpoints may also include application-level checksums. Readers verify and process only the file region up
to the last checkpoint, which is known to be in the defined
state. Regardless of consistency and concurrency issues, this
approach has served us well. Appending is far more efficient and more resilient to application failures than random
writes. Checkpointing allows writers to restart incrementally and keeps readers from processing successfully written file data that is still incomplete from the application’s perspective.
In the other typical use, many writers concurrently append to a file for merged results or as a producer-consumer
queue. Record append’s append-at-least-once semantics preserves each writer’s output. Readers deal with the occasional padding and duplicates as follows. Each record prepared by the writer contains extra information like checksums so that its validity can be verified. A reader can
identify and discard extra padding and record fragments
using the checksums. If it cannot tolerate the occasional
duplicates (e.g., if they would trigger non-idempotent operations), it can filter them out using unique identifiers in
the records, which are often needed anyway to name corresponding application entities such as web documents. These
functionalities for record I/O (except duplicate removal) are
in library code shared by our applications and applicable to
other file interface implementations at Google. With that,
the same sequence of records, plus rare duplicates, is always
delivered to the record reader.
3. SYSTEM INTERACTIONS
We designed the system to minimize the master’s involvement in all operations. With that background, we now describe how the client, master, and chunkservers interact to
implement data mutations, atomic record append, and snapshot.
3.1 Leases and Mutation Order
A mutation is an operation that changes the contents or
metadata of a chunk such as a write or an append operation. Each mutation is performed at all the chunk’s replicas.
We use leases to maintain a consistent mutation order across
replicas. The master grants a chunk lease to one of the replicas, which we call the primary. The primary picks a serial
order for all mutations to the chunk. All replicas follow this
order when applying mutations. Thus, the global mutation
order is defined first by the lease grant order chosen by the
master, and within a lease by the serial numbers assigned
by the primary.
The lease mechanism is designed to minimize management overhead at the master. A lease has an initial timeout
of 60 seconds. However, as long as the chunk is being mutated, the primary can request and typically receive extensions from the master indefinitely. These extension requests
and grants are piggybacked on the HeartBeat messages regularly exchanged between the master and all chunkservers.
The master may sometimes try to revoke a lease before it
expires (e.g., when the master wants to disable mutations
on a file that is being renamed). Even if the master loses
communication with a primary, it can safely grant a new
lease to another replica after the old lease expires.
In Figure 2, we illustrate this process by following the
control flow of a write through these numbered steps.
1. The client asks the master which chunkserver holds
the current lease for the chunk and the locations of
the other replicas. If no one has a lease, the master
grants one to a replica it chooses (not shown).
2. The master replies with the identity of the primary and
the locations of the other (secondary) replicas. The
client caches this data for future mutations. It needs
to contact the master again only when the primary becomes unreachable or replies that it no longer holds
a lease.
3. The client pushes the data to all the replicas. A client
can do so in any order. Each chunkserver will store
the data in an internal LRU buffer cache until the
data is used or aged out. By decoupling the data flow
from the control flow, we can improve performance by
scheduling the expensive data flow based on the network topology regardless of which chunkserver is the
primary. Section 3.2 discusses this further.
4. Once all the replicas have acknowledged receiving the
data, the client sends a write request to the primary.
The request identifies the data pushed earlier to all of
the replicas. The primary assigns consecutive serial
numbers to all the mutations it receives, possibly from
multiple clients, which provides the necessary serialization. It applies the mutation to its own local state
in serial number order.
5. The primary forwards the write request to all secondary replicas. Each secondary replica applies mutations in the same serial number order assigned by
the primary.
6. The secondaries all reply to the primary indicating
that they have completed the operation.
7. The primary replies to the client. Any errors encountered at any of the replicas are reported to the client.
In case of errors, the write may have succeeded at the
primary and an arbitrary subset of the secondary replicas. (If it had failed at the primary, it would not
have been assigned a serial number and forwarded.)
The client request is considered to have failed, and the
modified region is left in an inconsistent state. Our
client code handles such errors by retrying the failed
mutation. It will make a few attempts at steps (3)
through (7) before falling back to a retry from the beginning of the write.
If a write by the application is large or straddles a chunk
boundary, GFS client code breaks it down into multiple
write operations. They all follow the control flow described
above but may be interleaved with and overwritten by concurrent operations from other clients. Therefore, the shared file region may end up containing fragments from different
clients, although the replicas will be identical because the individual operations are completed successfully in the same
order on all replicas. This leaves the file region in consistent
but undefined state as noted in Section 2.7.
3.2 Data Flow
We decouple the flow of data from the flow of control to
use the network efficiently. While control flows from the
client to the primary and then to all secondaries, data is
pushed linearly along a carefully picked chain of chunkservers
in a pipelined fashion. Our goals are to fully utilize each
machine’s network bandwidth, avoid network bottlenecks
and high-latency links, and minimize the latency to push
through all the data.
To fully utilize each machine’s network bandwidth, the
data is pushed linearly along a chain of chunkservers rather
than distributed in some other topology (e.g., tree). Thus,
each machine’s full outbound bandwidth is used to transfer the data as fast as possible rather than divided among
multiple recipients.
To avoid network bottlenecks and high-latency links (e.g.,
inter-switch links are often both) as much as possible, each
machine forwards the data to the “closest” machine in the
network topology that has not received it. Suppose the
client is pushing data to chunkservers S1 through S4. It
sends the data to the closest chunkserver, say S1. S1 forwards it to the closest chunkserver S2 through S4 closest to
S1, say S2. Similarly, S2 forwards it to S3 or S4, whichever
is closer to S2, and so on. Our network topology is simple
enough that “distances” can be accurately estimated from
IP addresses.
Pipelining - Google and learn more about it. HTTP pipelining - https://en.wikipedia.org/wiki/HTTP_pipelining
Finally, we minimize latency by pipelining the data transfer over TCP connections. Once a chunkserver receives some
data, it starts forwarding immediately. Pipelining is especially helpful to us because we use a switched network with
full-duplex links. Sending the data immediately does not
reduce the receive rate. Without network congestion, the
ideal elapsed time for transferring B bytes to R replicas is
B/T + RL where T is the network throughput and L is latency to transfer bytes between two machines. Our network
links are typically 100 Mbps (T), and L is far below 1 ms.
Therefore, 1 MB can ideally be distributed in about 80 ms.
3.3 Atomic Record Appends
GFS provides an atomic append operation called record
append. In a traditional write, the client specifies the offset at which data is to be written. Concurrent writes to
the same region are not serializable: the region may end up
containing data fragments from multiple clients. In a record
append, however, the client specifies only the data. GFS
appends it to the file at least once atomically (i.e., as one
continuous sequence of bytes) at an offset of GFS’s choosing
and returns that offset to the client. This is similar to writing to a file opened in O APPEND mode in Unix without the
race conditions when multiple writers do so concurrently.
Record append is heavily used by our distributed applications in which many clients on different machines append
to the same file concurrently. Clients would need additional complicated and expensive synchronization, for example through a distributed lock manager, if they do so
with traditional writes. In our workloads, such files often serve as multiple-producer/single-consumer queues or contain merged results from many different clients.
Record append is a kind of mutation and follows the control flow in Section 3.1 with only a little extra logic at the
primary. The client pushes the data to all replicas of the
last chunk of the file Then, it sends its request to the primary. The primary checks to see if appending the record
to the current chunk would cause the chunk to exceed the
maximum size (64 MB). If so, it pads the chunk to the maximum size, tells secondaries to do the same, and replies to
the client indicating that the operation should be retried
on the next chunk. (Record append is restricted to be at
most one-fourth of the maximum chunk size to keep worst-case fragmentation at an acceptable level.) If the record
fits within the maximum size, which is the common case,
the primary appends the data to its replica, tells the secondaries to write the data at the exact offset where it has, and
finally replies success to the client.
If a record append fails at any replica, the client retries the
operation. As a result, replicas of the same chunk may contain different data possibly including duplicates of the same
record in whole or in part. GFS does not guarantee that all
replicas are bytewise identical. It only guarantees that the
data is written at least once as an atomic unit. This property follows readily from the simple observation that for the
operation to report success, the data must have been written
at the same offset on all replicas of some chunk. Furthermore, after this, all replicas are at least as long as the end
of record and therefore any future record will be assigned a
higher offset or a different chunk even if a different replica
later becomes the primary. In terms of our consistency guarantees, the regions in which successful record append operations have written their data are defined (hence consistent),
whereas intervening regions are inconsistent (hence undefined). Our applications can deal with inconsistent regions
as we discussed in Section 2.7.2.
