Patent Publication Number: US-10310904-B2

Title: Distributed technique for allocating long-lived jobs among worker processes

Description:
BACKGROUND 
     Field 
     The disclosed embodiments generally relate to data storage systems. More specifically, the disclosed embodiments relate to the design of a data storage system that facilitates allocating long-lived jobs among a set of worker processes. 
     Related Art 
     Organizations are presently using cloud-based storage systems to store large volumes of data. These cloud-based storage systems are typically operated by hosting companies that maintain a sizable storage infrastructure, often comprising thousands of servers that are sited in geographically distributed data centers. Customers typically buy or lease storage capacity from these hosting companies. In turn, the hosting companies provision storage resources according to the customers&#39; requirements and enable the customers to access these storage resources. 
     Within such cloud-based storage systems, a number of long-lived jobs are typically executed to perform various housekeeping tasks, such as propagating updates among machines, and performing garbage-collection operations. These long-lived jobs are typically executed by a set of worker processes distributed across machines in the system. Over time, these worker processes can fail due to various hardware and software errors, so it is often necessary to reallocate jobs among the remaining worker processes. During this re-allocation process, it is desirable to balance the jobs among worker processes as evenly as possible to prevent a specific worker process from becoming overly busy and possibly failing. Moreover, to facilitate fault-tolerance and scalability, it is desirable not to rely on a central arbiter to perform these reallocation operations. It is also desirable to minimize the number of jobs that need to be moved among worker processes during a reallocation operation, because moving an ongoing job can be a time-consuming and complicated task. 
     Hence, what is needed is a distributed technique for efficiently allocating long-lived jobs among worker processes. 
     SUMMARY 
     The disclosed embodiments relate to a distributed computing system that uses worker processes to execute long-lived jobs. During operation, each worker process performs the following operations. First, the worker process identifies a set of jobs to be executed, and also identifies a set of worker processes that are available to execute the set of jobs. Next, the worker process sorts the set of worker processes based on unique identifiers for the worker processes. Then, the worker process assigns jobs to each worker process in the set of worker processes, wherein approximately the same number of jobs is assigned to each worker process, and wherein jobs are assigned to the worker processes in sorted order. Assigning jobs to each worker process involves using a unique identifier for each worker process to seed a deterministic pseudorandom number generator, and then using the seeded deterministic pseudorandom number generator to select the one or more jobs for each worker process to execute. 
     In some embodiments, after assigning the one or more jobs to each worker process, the worker process commences executing the one or more jobs that are assigned to it, and stops executing jobs that are no longer assigned to it. 
     In some embodiments, identifying the set of worker processes involves querying a distributed consensus service that operates by sending heartbeats to different machines in the distributed computing system to identify worker processes that are available to execute jobs. 
     In some embodiments, each new worker process that comes online performs initialization operations, wherein the initialization operations include: registering the new worker process with a distributed consensus service to obtain an identifier for the worker process; and subscribing to an online worker list to obtain notifications about changes to the set of worker processes that are available to execute jobs. 
     In some embodiments, using the deterministic pseudorandom number generator to select the one or more jobs for each worker process involves: determining a desired number of jobs that the worker process will execute by dividing a number of jobs in the set of jobs by a number of worker processes in the set of worker processes and, if necessary, adding one to account for a remainder. It also involves repeating the following operations until the desired number of jobs is selected for the worker process to execute. First, the system obtains a random number from the deterministic pseudorandom number generator. Next, the system uses the random number and a modulo operation to randomly select a job from the set of jobs. If the job has already been selected, the system repeats the process of obtaining a random number and using the random number to select a job until an unselected job is selected. 
     In some embodiments, each worker process repeats the process of assigning jobs to worker processes whenever the set of worker processes changes. 
     In some embodiments, the operations involved in executing a job are idempotent, thereby allowing multiple worker processes to repeat the operations without violating correctness when jobs are reassigned among worker processes. 
     In some embodiments, the set of jobs that are long-lived includes jobs that are responsible for managing queues of updates and deletes directed to replicated copies of data items located in different zones, wherein each zone comprises a separate geographic storage location. 
    
    
     
       BRIEF DESCRIPTION OF THE FIGURES 
         FIG. 1  illustrates a content-management environment in accordance with the disclosed embodiments. 
         FIG. 2  illustrates a set of data centers in accordance with the disclosed embodiments. 
         FIG. 3  illustrates the logical structure of the data storage system in accordance with the disclosed embodiments. 
         FIG. 4A  illustrates the structure of an object storage device (OSD) in accordance with the disclosed embodiments. 
         FIG. 4B  illustrates the structure of a write-ahead log (WAL) in accordance with the disclosed embodiments. 
         FIG. 5  presents a flow chart illustrating how a get( ) operation is processed in accordance with the disclosed embodiments. 
         FIG. 6  presents a flow chart illustrating how a put( ) operation is processed in accordance with the disclosed embodiments. 
         FIG. 7  presents a flow chart illustrating how a failure of a storage device is handled in accordance with the disclosed embodiments. 
         FIG. 8  presents a flow chart illustrating how an extent can be accessed in the open state and the closed state in accordance with the disclosed embodiments. 
         FIG. 9A  presents a flow chart illustrating operations that can be performed while changing an extent from the open state to the closed state in accordance with the disclosed embodiments. 
         FIG. 9B  illustrates the structure of an extent in accordance with the disclosed embodiments. 
         FIG. 9C  illustrates a hash table entry in accordance with the disclosed embodiments. 
         FIG. 10  illustrates how asynchronous puts and deletes are communicated between zones in accordance with the disclosed embodiments. 
         FIG. 11  illustrates the structure of a hash database (HDB) including a cross-zone replication (XZR) queue in accordance with the disclosed embodiments. 
         FIG. 12  presents a flow chart illustrating how a delete operation received from a front end is processed in accordance with the disclosed embodiments. 
         FIG. 13  presents a flow chart illustrating how an asynchronous delete operation received from another zone is processed in accordance with the disclosed embodiments. 
         FIG. 14  presents a flow chart illustrating how a put operation received from a front end is processed in accordance with the disclosed embodiments. 
         FIG. 15  presents a flow chart illustrating how an asynchronous put operation received from another zone is processed in accordance with the disclosed embodiments. 
         FIG. 16  illustrates how long-lived jobs can be allocated among worker processes in accordance with the disclosed embodiments. 
         FIG. 17  presents a flow chart illustrating how a worker process identifies one or more jobs to execute in accordance with the disclosed embodiments. 
         FIG. 18  presents a flow chart illustrating how a worker process assigns a set of jobs to a set of worker processes in accordance with the disclosed embodiments. 
     
    
    
     DETAILED DESCRIPTION 
     The following description is presented to enable any person skilled in the art to make and use the present embodiments, and is provided in the context of a particular application and its requirements. Various modifications to the disclosed embodiments will be readily apparent to those skilled in the art, and the general principles defined herein may be applied to other embodiments and applications without departing from the spirit and scope of the present embodiments. Thus, the present embodiments are not limited to the embodiments shown, but are to be accorded the widest scope consistent with the principles and features disclosed herein. 
     The data structures and code described in this detailed description are typically stored on a computer-readable storage medium, which may be any device or medium that can store code and/or data for use by a computer system. The computer-readable storage medium includes, but is not limited to, volatile memory, non-volatile memory, magnetic and optical storage devices such as disk drives, magnetic tape, CDs (compact discs), DVDs (digital versatile discs or digital video discs), or other media capable of storing computer-readable media now known or later developed. 
     The methods and processes described in the detailed description section can be embodied as code and/or data, which can be stored in a computer-readable storage medium as described above. When a computer system reads and executes the code and/or data stored on the computer-readable storage medium, the computer system performs the methods and processes embodied as data structures and code and stored within the computer-readable storage medium. Furthermore, the methods and processes described below can be included in hardware modules. For example, the hardware modules can include, but are not limited to, application-specific integrated circuit (ASIC) chips, field-programmable gate arrays (FPGAs), and other programmable-logic devices now known or later developed. When the hardware modules are activated, the hardware modules perform the methods and processes included within the hardware modules. 
     Before describing the details of the data storage system, we first describe the structure of an exemplary online content-management system  120  that includes such a data storage system and that operates within such a content-management environment  105 . 
     Content-Management Environment 
       FIG. 1  illustrates content-management environment  105  according to various embodiments. As may be understood from this figure, content-management environment  105  includes a plurality of client devices  110 A and  110 B (collectively  110 ) and an online content-management system  120  that are interconnected by one or more networks  118 . Various aspects of the client devices  110  and online content-management system  120  are discussed below. 
     Client Devices 
     In various embodiments, each client device  110  may selectively execute a content-management client application  112 A and  112 B (collectively  112 ) (also referred to as a “content-management client”) that may be used to access content items stored within online content-management system  120 . In some embodiments, synchronized copies of a content item  114 A,  114 B and  114 C are maintained on client devices  110 A and  110 B and within online content-management system  120 , respectively. (Note that a “content item” can include a file, a folder, a set of folders, or any other type of data object.) In some embodiments, client devices  110  may provide a file-browser type interface (not shown) for directly manipulating the content items stored on online content-management system  120  without maintaining a local copy. Client devices  110  may also include applications  116 A and  116 B (collectively  116 ) that manipulate copies of content items  114 A and  114 B. 
     While only two client devices  110 A and  110 B are shown in  FIG. 1  for purposes of clarity, it should be understood by those skilled in the art that many client devices  110  may simultaneously connect through network(s)  118  to online content-management system  120  at any given time. Examples of suitable client devices  110  include, but are not limited to, a desktop computer; mobile computing devices, such as a laptop or a tablet; and handheld devices, such as a smartphone (e.g., an IPHONE®, BLACKBERRY®, or ANDROID™-based smartphone). Each client device  110  may store a local, synched copy of one or more content items from within online content-management system  120 , and the content items may be stored in any suitable format. When content-management client  112  presents content items that are stored within the online content-management system  120  to a user, the content items may be arranged in folders and the folders themselves may be arranged in other folders, or in any other arbitrary arrangement supported by online content-management system  120 , as determined by the user. However, one of skill in the art should understand in light of this disclosure that each user&#39;s content item storage architecture may be considerably different from the next, and in some instances, the content item storage architecture may be implemented to maximize storage and content item retrieval efficiency. 
     Content-Management System 
     Online content-management system  120  stores content items and manages access to those content items via client devices  110 . Online content-management system  120  and its components may be implemented using any appropriate hardware and software that supports file serving, storage, and retrieval functions. For example, online content-management system  120  may be implemented in a single server or multiple servers. 
     In various embodiments, online content-management system  120  includes interface module  122 , account module  124 , content-item-updating module  126 , and data store  128 . Some of the elements of online content-management system  120  are discussed below. 
     Content-Management System—Interface Module 
     In particular embodiments, interface module  122  may facilitate content item access and content item storage operations among online content-management system  120  and client devices  110 . Interface module  122  may receive content items from and send content items to client devices  110  consistent with the user&#39;s preferences for sharing content items. Interface module  122  may also act as the counterpart to a client-side file-explorer style user interface that allows a user to manipulate content items directly stored on online content-management system  120 . In some embodiments, software on client devices  110  may integrate network-stored content items with the client&#39;s local file system to enable a user to manipulate network-stored content items through the same user interface (UI) used to manipulate content items on the local file system, e.g., via a file explorer, file finder or browser application. As an alternative or supplement to the client-side file-explorer interface, interface module  122  may provide a web interface for client devices  110  to access (e.g., via an application  116 ) and allow a user to manipulate content items stored within online content-management system  120 . In this way, the user can directly manipulate content items stored within online content-management system  120 . 
     Content-Management System—Data Store 
     In various embodiments, data store  128  may store content items such as those uploaded using client devices  110 , or using any other suitable computing device. In the embodiment illustrated in  FIG. 1 , client device  110 A, which is associated with a first user, is shown as locally storing at least one content item  114 A, and client device  110 B, which is associated with a second user, is shown as locally storing at least one content item  114 B. As shown in  FIG. 1 , a copy of the locally stored content item  114 C is maintained in data store  128  of online content-management system  120 . In the embodiment illustrated in  FIG. 1 , content items  114 A,  114 B and  114 C are local versions of the same shared document that reside on client devices  110 A,  110 B and online content-management system  120 , respectively. Note that client devices  110 A and  110 B may also store other content items that are replicated on online content-management system  120 , but are not shared with other client devices. 
     In various embodiments, data store  128  may maintain information identifying the user, information describing the user&#39;s file directory, and other information in a file journal that is maintained for each user. In some embodiments, the file journal may be maintained on online content-management system  120 , and in other embodiments, a file journal (e.g., a “server-side file journal”) may be maintained on both online content-management system  120  and locally on each client device  110 . In various embodiments, the file journal may be used to facilitate the synchronization of the various copies of a particular content item that are associated with a user&#39;s account. 
     As a particular example, in the embodiment shown in  FIG. 1 , the system may be configured so that any changes that are made to content item  114 A on particular client device  110 A may also be automatically reflected in the copy of content item  114 C stored within online content-management system  120  and the copy of the content item  114 B on client device  110 B. Similarly, any changes that are made to content item  114 C on online content-management system  120  may also be automatically reflected in the copy of content item  114 A stored on client device  110 A, and the copy of content item  114 B stored on client device  110 B. 
     Also, content items  114 A and  114 B may be stored in local caches within content-management clients  112 A and  112 B, respectively. Alternatively, content items  114 A and  114 B may be stored in local file systems within content-management clients  112 A and  112 B. In some situations, content items  114 A and  114 B are stored in file system space that is reserved for content-management clients  112 A and  112 B. In other situations, content items  114 A and  114 B are stored in normal file system space that is not reserved for content-management clients  112 A and  112 B. 
     Content-Management System—Account Module 
     In particular embodiments, account module  124  may track content items stored in data store  128  and entries in the server-side file journal for each content item. As users grant content-item-access permissions to other users, account module  124  may update the server-side file journal associated with each relevant user in data store  128 . Account module  124  may also track client devices  110  that are associated with each user&#39;s account. For example, a user may want to share all their content items among their desktop computer, tablet computer, and mobile device. To make such a sharing arrangement seamless to the user, the user&#39;s single account on online content-management system  120  may be associated with each of the user&#39;s respective client devices. In some embodiments, an application running on each respective client device  110  may help to coordinate synchronization of content items on the client device with corresponding versions of the content items within the user&#39;s account in online content-management system  120 , and also with corresponding versions of the content items stored on the user&#39;s various other client devices. 
     Content-Management System—Content-Item-Updating Module 
     In particular embodiments, content-item-updating module  126  is configured to maintain consistency among different copies (versions) of a content item  114 A,  114 B and  114 C that are located in client device  110 A, client device  110 B and online content-management system  120 , respectively. This can be complicated, because the different versions  114 A,  114 B and  114 C of the same shared content items may be simultaneously changed at client devices  110 A-B and online content-management system  120 . Hence, online content-management system  120  needs to enforce an updating policy to resolve conflicting updates to different versions of the same content item. For example, the copy of the content item  114 C on online content-management system  120  can be the master copy of the content item, and updates to the various copies  114 A and  114 B of the content item can be serialized and applied one-at-a-time to the master copy  114 C before being propagated back to the copies  114 A and  114 B located on client devices  110 A and  110 B. 
     Data Centers 
     We next describe the data centers that provide the infrastructure for the data storage system.  FIG. 2  illustrates an exemplary data store  128  (from  FIG. 1 ) that comprises a set of data centers  201 - 203  in accordance with the disclosed embodiments. Note that data store  128  can be smaller than the system illustrated in  FIG. 2 . (For example, data store  128  can comprise a single server that is connected to a number of disk drives, a single rack that houses a number of servers, a row of racks, or a single data center with multiple rows of racks.) As illustrated in  FIG. 2 , data store  128  can include a set of geographically distributed data centers  201 - 203  that may be located in different states, different countries or even on different continents. 
     Data centers  201 - 203  are coupled together through a network  200 , wherein network  200  can be a private network with dedicated communication links, or a public network, such as the Internet, or a virtual-private network (VPN) that operates over a public network. 
     Communications to each data center pass through a set of routers that route the communications to specific storage nodes within each data center. More specifically, communications with data center  201  pass through routers  205 , communications with data center  202  pass through routers  206 , and communications with data center  203  pass through routers  207 . 
     As illustrated in  FIG. 2 , routers  205 - 207  channel communications to storage devices within the data centers, wherein the storage devices are incorporated into servers that are housed in racks, wherein the racks are organized into rows within each data center. For example, the racks within data center  201  are organized into rows  210 ,  220  and  230 , wherein row  210  includes racks  211 - 214 , row  220  includes racks  221 - 224  and row  230  includes racks  231 - 234 . The racks within data center  202  are organized into rows  240 ,  250  and  260 , wherein row  240  includes racks  241 - 244 , row  250  includes racks  251 - 254  and row  260  includes racks  261 - 264 . Finally, the racks within data center  203  are organized into rows  270 ,  280  and  290 , wherein row  270  includes racks  271 - 274 , row  280  includes racks  281 - 284  and row  290  includes racks  291 - 294 . 
     As illustrated in  FIG. 2 , data store  128  is organized hierarchically, comprising multiple data centers, wherein machines within each data center are organized into rows, wherein each row includes one or more racks, wherein each rack includes one or more servers, and wherein each server (also referred to as an “object storage device” (OSD)) includes one or more storage devices (e.g., disk drives). 
     Data Storage System 
       FIG. 3  illustrates the logical structure of the data storage system  300  in accordance with the disclosed embodiments. As illustrated in  FIG. 3 , data storage system  300  includes a logical entity called a “pocket”  302  that in some embodiments is equivalent to an Amazon S3™ bucket. Each pocket is completely distinct; nothing is shared between pockets. For example, in an exemplary implementation, the system provides a “block storage pocket” to store data files, and a “thumbnail pocket” to store thumbnail images for data objects. Note that the applications specify which pockets are to be accessed. 
     Within a pocket one or more “zones” exist that are associated with physical data centers, and these physical data centers can reside at different geographic locations. For example, one data center might be located in California, another data center might be located in Virginia, and another data center might be located in Europe. For fault-tolerance purposes, data can be stored redundantly by maintaining multiple copies of the data on different servers within a single data center and also across multiple data centers. 
     For example, when a data item first enters a data center, it can be initially replicated to improve availability and provide fault tolerance. It can then be asynchronously propagated to other data centers. 
     Note that storing the data redundantly can simply involve making copies of data items, or alternatively using a more space-efficient encoding scheme, such as erasure codes (e.g., Reed-Solomon codes) or Hamming codes to provide fault tolerance. 
     Within each zone (such as zone  304  in  FIG. 3 ), there exist a set of front ends  321 - 324 , a hash database (HDB)  330  and a set of “cells,” such as cell  340  illustrated in  FIG. 3 . A typical cell  340  includes a number of object storage devices (OSDs)  343 - 346 , wherein the individual OSDs  343 - 346  include storage devices that actually store data blocks. Cell  340  also includes a “master”  341 , which is in charge of managing OSDs  343 - 346  and a bucket database (BDB)  342  described in more detail below. (Note that HDB  330  and BDB  342  are logical databases which can be stored redundantly in multiple physical databases to provide fault tolerance.) 
     Master  341  performs a number of actions. For example, master  341  can determine how many writeable buckets the system has at any point in time. If the system runs out of buckets, master  341  can create new buckets and allocate them to the storage devices. Master  341  can also monitor OSDs and associated storage devices, and if any OSD or storage device fails, master  341  can migrate the associated buckets to other OSDs. 
     As illustrated in  FIG. 3 , a number of block servers  316 - 319 , which are typically located in a data center associated with a zone, can service requests from a number of clients  311 - 314 . For example, clients  311 - 314  can comprise applications running on client machines and/or devices that access data items in data storage system  300 . Block servers  316 - 319  in turn forward the requests to front ends  321 - 324  that are located within specific zones, such as zone  304  illustrated in  FIG. 3 . Note that clients  311 - 314  communicate with front ends  321 - 324  through block servers  316 - 319 , and the front ends  321 - 324  are the only machines within the zones that have public IP addresses. 
     Files to be stored in data storage system  300  comprise one or more data blocks that are individually stored in data storage system  300 . For example, a large file can be associated with multiple data blocks, wherein each data block is 1 MB to 4 MBs in size. 
     Moreover, each data block is associated with a “hash” that serves as a global identifier for the data block. The hash can be computed from the data block by running the data block through a hash function, such as a SHA-256 hash function. (The SHA-256 hash function is defined as a Federal Information Processing Standard (FIPS) by the U.S. National Institute of Standards and Technology (NIST).) The hash is used by data storage system  300  to determine where the associated data block is stored. 
     Get( ) Operation 
     The system performs a number of operations while processing data accesses on behalf of clients  311 - 314 . For example, when a get( ) operation is received along with an associated hash, the hash is used to perform a lookup in HDB  330 . This lookup returns an identifier for a “bucket” and associated cell where the data block is stored. 
     To streamline failure-recovery operations, a large number of data blocks can be aggregated into larger buckets. For example, a number of 1-4 MB data blocks can be aggregated into a single 1 GB bucket, wherein each bucket is stored in a specific cell. This enables the system to manipulate a small number of buckets during a failure-recovery operation instead of manipulating a large number of individual data blocks. Aggregating data blocks into buckets also greatly decreases the amount of metadata the system has to maintain and manipulate; this is advantageous because metadata is computationally expensive to maintain and manipulate. 
     Because a large number of data blocks can exist in data storage system  300 , HDB  330  can potentially be very large. If HDB  330  is very large, it is advantageous to structure HDB  330  as a “sharded” database. For example, when performing a lookup based on a hash in HDB  330 , the first 8 bits of the hash can be used to associate the hash with one of 256 possible shards, and this shard can be used to direct the lookup to an associated instance of HDB  330 . For example, as illustrated in  FIG. 3 , HDB  330  can comprise 4 instances  331 - 334 , wherein instance  331  is associated with shards  1 - 64 , instance  332  is associated with shards  65 - 128 , instance  333  is associated with shards  129 - 192  and instance  334  is associated with shards  193 - 256 . In other embodiments, HDB  330  can be divided into more or fewer instances. (Note that a zone can include a “ZooKeeper™ cluster” that is responsible for mapping shards to specific target cells and also mapping shards to physical HDB machines.) 
     HDB instances  331 - 334  are logical databases that are mapped to physical databases, and to provide fault tolerance, each logical database can be redundantly stored in multiple physical databases. For example, in one embodiment, each HDB instance maps to three physical databases. If data storage system  300  is very large (for example containing trillions of data blocks), HDB  330  will be too large to fit in random-access memory. In this case HDB  330  will mainly be stored in non-volatile storage, which for example, can comprise flash drives or disk drives. 
     After the bucket and associated cell are identified for the get( ) operation, the system performs a lookup in a bucket database (BDB)  342  in the associated cell  340 . This lookup returns an identifier for an object storage device (OSD)  343  where the bucket is located. Note that because each bucket is fairly large (e.g., 1 GB) and contains a large number of data blocks, BDB  342  is relatively small and can typically be stored in random-access memory, which greatly speeds up the lookup process. 
     Finally, within the OSD, the system performs a lookup based on the bucket and the hash to determine an offset and a length for the data block in a write-ahead log that stores data blocks for the bucket. The system then returns the data block from the determined offset in the write-ahead log. Note that because data storage system  300  is designed to store “immutable data” that does not change after it is written, it is efficient to store the immutable data in a write-ahead log, as opposed to a random-access structure. Because the data is never overwritten, writes do not require more complex and time-consuming random-access lookup mechanisms. 
     Put( ) Operation 
     During a put( ) operation, the system receives a data block to be written from a client. To process the put( ) operation, the system first computes a hash from the data block, for example using the SHA-256 technique described above. Next, the system selects a writeable bucket and an associated cell for the data block. Note that front ends  321 - 324  periodically poll all the BDBs to identify and then cache writeable buckets. This enables front ends  321 - 324  to keep track of a number of buckets (e.g., 10 to 100 buckets) that they know are writeable at any given time. Then, when a put( ) operation is subsequently received, a front end simply selects a cached bucket that it knows is writable. 
     Within the associated cell, the system uses an identifier for the selected bucket to perform a lookup in the BDB. This lookup returns one or more OSDs for the bucket. (Note that the bucket may be replicated across multiple OSDs to provide fault tolerance.) Within the OSDs, the system appends the data block to a write-ahead log that stores data blocks for the bucket. After the data is stably written to the OSDs, the system writes the hash-to-bucket mapping to the HDB  330 . 
     Note that the master  341  modifies the BDB  342  and the front end  321  modifies the HDB  330 . In general, master  341  is concerned with reliability of storage, and hence performs operations to facilitate redundancy and rebalancing, while the front end  321  is generally concerned with finding information and simply maps hashes to logical constructs, such as buckets. 
     Master  341  performs various operations to detect and handle failures. More specifically, master  341  periodically performs health checks on OSDs. If master  341  detects a failure in an OSD, the associated buckets are degraded and the master sets the buckets to be non-writable. Note that get( ) operations have to access the buckets where the blocks are stored, but put( ) operations can be directed to any bucket that is currently writeable, so when a problem happens with a bucket, the system simply marks the bucket as non-writeable. The system can continue performing get( ) operations on the degraded bucket, because there exist multiple copies of the degraded bucket. 
     To handle a failure associated with a bucket, master  341  tells the associated OSDs to freeze the bucket. Master  341  then tells the OSDs to replicate the bucket to a new OSD. The system then adds the new OSD to the cluster, increments the generation number for the OSD, and marks the bucket as writeable. (Note that when a degraded OSD is restarted after a failure, it will not accept any reads because its generation number is old.) The system guarantees that every OSD in the current generation has valid data. 
     The system also includes mechanisms to perform compaction operations. Although the data stored in data storage system  300  is immutable, the system often needs to delete data items when users remove them from the system. In some embodiments, the system tracks deleted data items in a log, and when the usable storage in a given bucket falls below a threshold, the system compacts the bucket. 
     Object Storage Device 
       FIG. 4A  illustrates the structure of an exemplary object storage device (OSD)  343  in accordance with the disclosed embodiments. As illustrated in  FIG. 4 , OSD  343  includes a processor  406  that is connected to a memory  408  through a bridge  407 . Processor  406  is also coupled to Serial Attached SCSI (SAS) expanders  410  and  420 , where SAS expander  410  is coupled to disk drives  411 - 414  and SAS expander  420  is coupled to disk drives  421 - 424 . (Note that SAS expanders  410  and  420  may be coupled to more or fewer disk drives.) Also, note that a failure in OSD  343  can involve a failure of a single one of the disk drives  411 - 414  or  421 - 424 , or a failure that affects all or most of OSD  343 , such as a failure in processor  406 , bridge  407 , memory  408 , SAS expanders  410  and  420  or one of the associated data paths. 
     Write-Ahead Log 
       FIG. 4B  illustrates the structure of a write-ahead log (WAL)  450  which is maintained within an OSD (such as OSD  343 ) in accordance with the disclosed embodiments. WAL  450  provides a log-structured data store which is advantageous for storing immutable data. WAL  450  comprises one or more 1 GB extents which can be associated with the logical buckets described above. As illustrated in  FIG. 4B , an extent can include a “data portion”  452  that has already been written to, and an unwritten portion that contains available space  454 . The data blocks that are stored within data portion  452  are associated with metadata that, for example, contains hashes and the offsets for the data blocks. To improve performance, metadata associated with recently written data blocks  458  can be stored in a memory buffer. When the system recovers from a failure, all of the metadata can be reconstructed by scanning through WAL  450  starting from a last known pointer  453 . 
     During a put( ) operation, the system synchronously appends the data block and an associated header to the WAL  450 , wherein the header includes a number of data items associated with the block, including the hash and the length of the block. At the same time, the system synchronously adds metadata to the memory buffer. When a bucket becomes full, the system seals the bucket, and the bucket never gets modified again. 
     During a get( ) operation, the system checks the memory buffer to find the offset and length for the data block. The system then uses the offset and length to read the data block from WAL  450 . 
     Get( ) Operation 
       FIG. 5  presents a flow chart illustrating how a “get( )” operation is processed in accordance with the disclosed embodiments. At the start of this process, the system receives a request to retrieve a data block from the data storage system, wherein the request was generated by a client performing a get( ) operation. This request includes a hash parameter that functions as a global identifier for the data block (step  502 ). Next, the system uses the hash to look up a bucket and an associated cell that contains the data block, wherein the lookup is performed in an HDB for the data storage system (step  504 ). Recall that the HDB can be a sharded database, and this lookup can involve examining the first byte of the hash to identify a corresponding shard, and then performing the lookup in a specific HDB instance associated with the shard. Also, recall that this lookup is likely to involve a random disk seek. 
     Within the cell, the system uses the bucket to look up an OSD that contains the bucket, wherein the lookup is performed in a local BDB for the cell (step  506 ). Recall that because the local BDB can be stored in memory, this lookup can be fast. Then, within the OSD, the system uses the bucket and the hash to determine an offset and a length for the data block in a write-ahead log that stores data blocks for the bucket (step  508 ). Finally, the system returns the data block from the determined offset in the write-ahead log (step  510 ). 
     Put( ) Operation 
       FIG. 6  presents a flow chart illustrating how a “put( )” operation is processed in accordance with the disclosed embodiments. At the start of this process, the system receives a request to write a data block to the data storage system, wherein the request was generated by a client performing a put( ) operation (step  602 ). Next, the system uses the data block to compute a hash that functions as a global identifier for the data block (step  604 ). As mentioned above, computing this hash can involve feeding the data block through a SHA-256 hash module. Then, the system selects a writeable bucket and an associated cell for the data block (step  606 ). Recall that the system maintains a pool of available buckets. Within the associated cell, the system uses the selected bucket to look up an OSD for the data block, wherein the lookup is performed in a local BDB for the selected cell (step  608 ). Because the BDB is typically located in memory, this lookup can be fast. (Also, note that the bucket can be replicated across a number of cells, in which case the lookup returns multiple OSDs containing copies of the bucket, and the data block is written to each of the multiple OSDs.) 
     Next, within the OSD, the system appends the data block to a write-ahead log that stores data blocks for the bucket (step  610 ). Note that committing the write-ahead log to disk typically involves a sequential disk seek. Finally, the system updates the HDB to include an entry that maps the hash to the selected bucket and associated cell (step  612 ). 
     Recovering from a Storage Device Failure 
       FIG. 7  presents a flow chart illustrating how a failure of a storage device is handled in accordance with the disclosed embodiments. At the start of this process, the system detects a failure associated with a bucket in a cell (step  702 ). (For example, the system can detect a failure in an OSD that stores a copy of the bucket.) Next, the system marks the bucket as non-writable (step  704 ). The system then performs a fast block-copy of the bucket to a new OSD in the cell (step  706 ). The systems also updates the BDB for the cell to indicate that the bucket is associated with the new OSD (step  708 ). As part of this updating operation, the system increments the new generation number for the OSD and updates the BDB with this new generation number (step  710 ). Recall that when a degraded OSD is restarted after a failure, it will not accept any reads or writes because its generation number is old. Note that it is important to update the BDB with the new generation number at the same time the BDB is updated to associate the bucket with the new OSD. This ensures there is no period of time where the BDB points to a new OSD in the wrong generation. 
     Storage Model Supporting Open and Closed Extents 
     As mentioned above, data storage systems often rely on locks to prevent updates to data blocks during repair operations and other operations that are adversely affected by such updates. However, locks can create performance problems for applications that need to access the data blocks. Instead of using locks, the disclosed embodiments solve this problem by placing extents (and associated volumes) in either an open state or a closed state, and then enforcing a strict state-machine separation between operations performed on extents in the open state and the closed state. 
     Note that an open extent (or associated open volume) is “internally mutable,” which means that data can be written to it. However, an open extent is also “externally immutable,” which means that the system cannot perform certain operations on the extent, including moving, merging, repairing, compacting, erasure-coding or garbage-collecting the extent. (Note that, at any given time, only a very small fraction of extents is in the open state.) In contrast, a closed extent (or associated closed volume) is internally immutable, which means that the system can move, merge, repair, compact, erasure-code or garbage-collect the closed extent. 
     Moreover, extents in the open state have different data formats than extents in the closed state because of the differing workloads for extents in the different states. For example, extents in the open state are tailored for append-only writes. In contrast, extents in the closed state have a more sophisticated index design to facilitate fast lookups with a low memory footprint. 
     Before describing how this data storage system operates, we first describe how extents are associated with volumes and buckets. As mentioned above, a “bucket” is a logical grouping of data blocks into a single conceptual unit, wherein each hash for a data block maps to a single bucket. Note that buckets are easier to manage than extremely large numbers of independent data blocks. For example, as mentioned above, a number of 1-4 MB data blocks can be aggregated into a single 1 GB bucket. Also, recall that the BDB maps hashes for data blocks to buckets, and this mapping is generally immutable. A “volume” is a mapping from one or more buckets to a set of OSDs. Note that the BDB has a separate table that maps each bucket to the volume that the bucket resides in and the OSDs that store extents for the volume. When the system uses a basic replication scheme, each volume contains a single bucket. On the other hand, when the system uses an erasure-coding storage scheme, a volume may contain multiple buckets, which are combined and coded across the set of OSDs. An “extent” contains the actual data for a given volume that is written to an OSD. More specifically, an extent is the part of a volume that resides on a single OSD. When using a basic replication scheme, there is only one bucket in each volume, and thus each extent is just the entire copy of the bucket. However, when using an erasure-coding storage scheme, each extent in the volume is different, and may either store the plaintext contents of a bucket, or parity data that is used to facilitate recovery from failures. Note that an OSD deals with extents because it rarely cares about the semantics of the data inside the extents. Moreover, a master typically deals with volumes, because it cares about moving bulk data around. Finally, front ends care about buckets, because buckets are logical containers for hashes, but they also need to map these hashes to volumes and extents to perform puts and gets on the corresponding OSDs. 
     We now describe how this data storage system operates on open and closed extents. In particular,  FIG. 8  presents a flow chart illustrating how an extent can be accessed in the open state and the closed state in accordance with the disclosed embodiments. When an extent is in the open state, the system allows data blocks to be appended to the extent, and disallows operations to be performed on the extent that are incompatible with data being concurrently appended to the extent (step  802 ). For example, operations that are incompatible with data being concurrently appended to the extent can include, but are not limited to: moving the extent; deleting the extent; merging the extent with another extent; repairing the extent; compacting the extent; garbage-collecting the extent; and erasure-coding the extent. 
     Next, when the extent becomes full, the system changes the extent from the open state to the closed state (step  804 ). Then, while the extent is in the closed state, the system disallows data blocks to be appended to the extent, and allows operations to be performed on the extent that are incompatible with data being concurrently appended to the extent (step  806 ). 
     Changing an Extent from the Open State to the Closed State 
       FIG. 9A  presents a flow chart illustrating operations that are performed to change an extent from the open state to the closed state in accordance with the disclosed embodiments. At the start of this process, when the extent becomes full (e.g., the amount of data in the extent exceeds a threshold value) the master tells all OSDs containing copies of the extent to close the extent (step  902 ). The close operation succeeds if any OSD closes its local copy of the extent and refuses to perform subsequent writes to its local copy of the extent. Hence, the master considers the extent to be closed as soon as it receives an acknowledgment from one of the OSDs that the extent has been closed. The master does not wait to receive acknowledgments from all of the OSDs holding copies of the extent because it is possible for one of the OSDs to go down before it can provide such an acknowledgment. 
     Next, the master waits for a period of time (e.g., 24 hours) to ensure that all changes to the extent have been committed to the HDB (step  904 ). Then, the master tells the OSDs to synchronize the extent to ensure that all copies of the extent contain the same data blocks (step  906 ). 
     During the synchronization operation, the master is informed if any of the copies of the extent are missing data blocks. The master assumes that such identified copies of the extent have been subject to a truncation operation. Note that if an extent is truncated, it is hard to determine precisely how much of the extent was truncated. Hence, after the synchronization operation is complete, the master tells the OSDs to delete any copies of the extent that are missing data blocks (step  908 ). After a copy of the extent is deleted, the system will eventually determine that the system is short one copy of the extent, and will replace the deleted copy by replicating another copy of the extent, which is not missing any data blocks, to another storage device. 
     Next, before closing the extent, all OSDs that hold copies of the extent construct an index containing entries that specify offsets and lengths for data blocks in the extent, and also generate an associated key list and deleted list (step  910 ). (This process is described in more detail below with reference to the closed extent illustrated in  FIG. 9B .) The OSDs subsequently append the index, key list and deleted list to the end of their copy of the extent in non-volatile storage (step  912 ). 
     Finally, to complete the process of changing the extent to the closed state, the OSDs update their copies of the extent in non-volatile storage to indicate the extent is in the closed state (step  914 ). A close operation can possibly fail if the OSD crashes, which can possibly lead to a half-generated index. It is, therefore, important that the commit point for changing an extent from the open state to the closed state occurs when the header of the extent is updated in non-volatile storage to indicate that the extent is in the closed state. 
       FIG. 9B  illustrates the structure of an exemplary closed extent  918  in accordance with the disclosed embodiments. Extent  918  includes a header that contains various metadata associated with the extent, including offsets for index  924 , key list  926  and deleted list  928 , which are described in more detail below. Extent  918  also includes a set of data blocks  922 , which has been appended to the extent. Note that each copy of the extent contains the same set of data blocks. However, the system does not perform any serialization operations while writing to the different copies of the extent, so the data blocks may be appended to each extent in a different order. 
     Extent  918  also includes an index  924  that facilitates looking up locations for data blocks in the extent. In general, any type of indexing structure can be used for this purpose. Some embodiments implement index  924  using a hash table, wherein each entry in the hash table is accessed using a hash key for the associated data block. Moreover, each hash table entry specifies an offset and a length for the data block within the extent to facilitate subsequently accessing the data block. 
     In some embodiments, index  924  is a variation of a cuckoo hash table that can be accessed to retrieve a data block as follows. First, a hash key that was generated from the data block is divided into three integers and a tag. (For example, a 16-byte hash key can be divided into three four-byte unsigned integers, and a four-byte tag.) The three integers are used as probe locations into the cuckoo hash table. The system constructs the cuckoo hash table by inserting each data block into the hash table. While inserting a given data block, if the system runs into a cycle, or if all three probe locations are taken, the system aborts the insertion process, increases the size of the cuckoo hash table and restarts the insertion process. The four-byte tag is stored in the hash table entry and, during subsequent lookup operations for a data block, is compared against four corresponding bytes of the hash key for the data block to determine whether the hash table entry is associated with the data block. After all of the data blocks are inserted into the hash table, the system performs one last pass through the data blocks and performs a lookup based on a hash key for the data block to ensure that the hash key matches a corresponding entry in one of the three probe locations. 
     For example, a typical hash table entry  930  is illustrated in  FIG. 9C . This hash table entry  930  includes an offset  932  that specifies a location for the start of the data block in the extent, and also a length  934  for the extent. Hash table entry  930  also includes a tag  938  that is matched against a portion of a hash key for a data block during a lookup. Entry  930  also includes a key length field  936  that specifies the length of the hash key that is used to access the hash table. 
     Note that the variation of the cuckoo hash table described above does not actually store a full copy of the hash key for each data block. The full hash keys are instead stored in an associated key list  926 . Within key list  926 , the hash keys are stored in the order that their corresponding entries appear in the hash table. 
     Extent  918  also includes a deleted list  928  that specifies locations in the index for data blocks that have been garbage-collected from extent  918 . 
     When closing an extent, the system can also compute and store internal checksums to cover the header  920 , the data blocks  922 , the index  924  and the key list  926 . These internal checksums can facilitate detecting subsequent data corruption errors in extent  918 . 
     Facilitating Cross-Zone Replication 
     As discussed above, to provide fault tolerance, data items are often replicated between zones. However, this cross-zone replication can complicate puts and deletes, because when a put or a delete operation is applied against a copy of a data item in a zone, it needs to be propagated to other zones to be applied against other copies of the data item. However, this can create a race condition between a put operation and a delete operation that are directed to the same data item. This race condition can cause inconsistencies about the ordering of the put and delete operations in different zones. In some cases, the put is applied before the delete, and in other cases the delete is applied before the put. These inconsistencies in the ordering of puts and deletes can ultimately cause inconsistencies between different copies of the data item in different zones. 
     This race condition problem is hard to resolve in the above-described storage system architecture because there exists no central arbiter that can determine the ordering of updates. Each zone maintains its own HDB and its own time base that can be used to serialize operations within the zone. However, there exists no global arbiter and no global time base to similarly serialize operations across zones. 
     The disclosed embodiments solve this problem without a central arbiter or a global time base. Instead, each zone makes decisions about the ordering of operations on its own. Also, a special conditional delete operation is used to avoid race conditions. This conditional delete operation will only delete a data item if the data item has not been updated for a period of time that is referred to as a “minimum lifetime” for a data item. This will ensure that if a put and a delete are directed to the same data item at the same time, the put will win, unless for some reason the put was delayed for more than the minimum lifetime. If this minimum lifetime is sufficiently long (e.g., seven days), the likelihood that the put will be delayed for longer than the minimum lifetime will be insignificant. Hence, it is possible that a data item that was legitimately deleted in a zone does not end up being deleted from one of the zones. However, this possibility is extremely remote if the minimum lifetime is sufficiently long (e.g., seven days). Moreover, this type of error will eventually be rectified by a background process that periodically examines a “zones table” (that contains entries indicating the zones the data items should reside in) and deletes copies of data items that should not exist in specific zones. 
     More specifically, referring to  FIG. 10 , each zone maintains its own HDB and also supports an associated HDB service that is responsible for propagating put and delete operations to other zones. In particular, zone  1002  maintains a local HDB  1006  and supports an associated HDB service  1004  that is responsible for propagating asynchronous puts and deletes  1008  to zone  1012 . Similarly, zone  1012  maintains a local HDB  1016  and supports an associated HDB service  1014  that is responsible for propagating asynchronous puts and deletes  1018  to zone  1002 . 
       FIG. 11  illustrates various data structures within HDB  1006  in accordance with the disclosed embodiments. These data structures include a table of hashes  1102  that contains an entry for each data item, wherein each entry includes: (1) a hash for the data item; (2) a checksum for the data item; (3) a last update time that specifies the time that the data item was last updated; and (4) various other fields, containing other types of information, such as location information for the data item. Note that this last update time is used to facilitate distributed delete operations as is described in more detail below. 
     The data structures also include a cross-zone replication (XZR) queue  1104  implemented as a first-in, first-out (FIFO) buffer that stores entries for asynchronous put and delete operations that are waiting to be propagated to a remote zone. After the asynchronous put and delete operations are communicated to the remote zone and associated acknowledgments are received, the associated entries can be removed from XZR queue  1104 . We next describe how put and delete operations are processed in a local zone. 
     Delete Operations 
       FIG. 12  presents a flow chart illustrating how a delete operation received from a front end is processed in accordance with the disclosed embodiments. First, a delete operation that is directed to a first data item is received at a local zone at a time t d  (step  1202 ). Next, the system computes a “maximum last update time” for the delete operation t mlu =t d −t min , wherein t min  is a minimum lifetime for a data item (step  1204 ). The system then determines whether a copy of the first data item exists in the local zone (step  1206 ). If not, the system asynchronously propagates the delete operation to other zones in the data storage system along with t mlu , wherein the delete operation is performed in another zone if the other zone determines that the first data item was last updated before t mlu , (step  1212 ). Otherwise, if a copy of the first data item exists in the local zone, the system determines, from a local index in the local zone, a time t lu  that the first data item was last updated (step  1208 ). For example, the system can perform a lookup in the table of hashes in the HDB to retrieve the last update time stored in an entry for the first data item. Next, if t lu &lt;t mlu , the system deletes the copy of the first data item in the local zone (step  1210 ). Otherwise, the system does not perform the delete operation. Finally, the system asynchronously propagates the delete operation to other zones in the data storage system along with t mlu , wherein the delete operation is performed in another zone if the other zone determines that the first data item was last updated before t mlu  (step  1212 ). 
     Note that the data item is not actually deleted right away. The HDB is updated to indicate that the data item has been deleted, and the client that sent the delete operation receives an acknowledgment that the data item has been deleted. However, the copy of the deleted data item actually persists in non-volatile storage until a background process eventually deletes the data item. 
       FIG. 13  presents a flow chart illustrating how an asynchronous delete operation received from another zone is processed in accordance with the disclosed embodiments. At the start of this process, the system receives an asynchronous delete operation at the local zone from a remote zone, wherein the asynchronous delete operation is directed to a second data item and includes a time t mlu  (step  1302 ). Next, the system determines if a copy of the second data item exists in the local zone (step  1304 ). If not, the process completes. Otherwise, if a copy of the second data item exists in the local zone, the system determines, from a local index maintained in the local zone, a time t lu  that the second data item was last updated (step  1306 ). If t lu &lt;t mlu , the system deletes the copy of the second data item in the local zone (step  1308 ). 
     Put Operations 
       FIG. 14  presents a flow chart illustrating how a put operation received from a front end is processed in accordance with the disclosed embodiments. At the start of this process, the system receives a put operation at a local zone from the front end, wherein the put operation is directed to a third data item (step  1402 ). Next, the system applies the put operation to either update or instantiate a copy of the third data item in the local zone (step  1404 ). Note that if the third data item does not already exist in the local zone, the put operation instantiates a copy of the third data item in the local zone. Next, the system updates a last update time t lu  for the third data item in a local index maintained at the local zone to specify a time that the put operation was applied to the third data item (step  1406 ). For example, the system can update the last update time t lu  in an entry for the third data item in the HDB. Finally, the system asynchronously propagates the put operation to other zones in the data storage system along with t lu  (step  1408 ). 
       FIG. 15  presents a flow chart illustrating how an asynchronous put operation received from another zone is processed in accordance with the disclosed embodiments. At the start of this process, the system receives an asynchronous put operation at the local zone from a remote zone, wherein the asynchronous put operation is directed to a fourth data item and includes a time t lu  indicating when the fourth data item was last updated (step  1502 ). Next, the system applies the asynchronous put operation to either update or instantiate a copy of the fourth data item in the local zone (step  1504 ). Finally, the system updates an entry for the fourth data item in a local index maintained at the local zone to indicate that the fourth data item was last updated at the later of: the time t lu , and the time the fourth data item was actually updated or instantiated in the local zone (step  1506 ). 
     Allocating Long-Lived Jobs Among Worker Processes 
     As mentioned above, a distributed storage system typically executes a number of ongoing long-lived jobs to perform various housekeeping operations, such as propagating updates among machines and performing garbage-collection operations. For example,  FIG. 16  illustrates how a set of long-lived jobs  1621 - 1636  can be allocated among a set of worker processes  1601 - 1604  in accordance with the disclosed embodiments. Worker processes  1601 - 1604  can generally execute on any computing node (e.g., central processing unit) in the distributed storage system. In some cases, each worker process  1601 - 1604  resides on a different computing node. In other cases, multiple worker processes share a single computing node. 
     Worker processes  1601 - 1604  are responsible for executing a number of long-lived jobs  1621 - 1636 . For example, the long-lived jobs  1621 - 1636  can include jobs that are responsible for processing cross-zone replication (XZR) queues as is described above with reference to  FIGS. 10-11 . Such long-lived jobs typically persist for long periods of time and can be thought of as ongoing “duties” that must be continually attended to, instead of as fixed-length jobs that can be processed within a time interval. 
     In some cases, the expected life span of such long-lived jobs  1621 - 1636  greatly exceeds the expected lifetime of a typical worker process  1601 - 1604 . Hence, if a worker process fails, the long-lived jobs that were being processed by the worker process need to be reallocated among the remaining worker processes. If this reallocation task is performed naively, for example by simply shifting the jobs among consecutive worker processes in an ordered list of the worker processes, a large number of jobs can potentially be moved among worker processes. This can cause a large amount of additional work to be performed in moving jobs among processes, and this additional work interferes with system performance. It is, therefore, preferable to keep jobs with the same worker processes if possible. Also, as mentioned above, to facilitate fault-tolerance and scalability, it is desirable not to rely on a central arbiter to perform these reallocation operations, but instead to use a distributed technique to perform the reallocation operations. 
     The embodiments described below implement a distributed technique for allocating long-lived jobs among worker processes. This distributed technique operates by executing the same code on each worker process to allocate the jobs among the available worker processes. In this way, no central arbiter is required to generate the allocation. Instead, each worker process computes the same allocation on its own. 
     More specifically,  FIG. 17  presents a flow chart illustrating how a worker process computes such an allocation in accordance with the disclosed embodiments. First, a new worker process comes on line (step  1702 ). Next, the worker process registers with a distributed consensus service and obtains a unique identifier (step  1704 ). The worker process also subscribes to receive push notifications about changes to the set of worker processes that are available to execute jobs (step  1706 ). For example, the worker process can register with an instance of the Apache ZooKeeper™ distributed service to obtain the unique identifier and to register for push notifications about changes to worker processes. Also, in some embodiments, the unique identifier is generated using an incrementing counter, or another technique that issues identifiers in increasing order. In this way, the system ensures that worker processes with higher identifier values joined the work pool more recently than worker processes with lower identifier values. (In cases where the worker process is already online, and the system is reallocating jobs because the set of worker processes has changed, the worker process skips the initialization operations in steps  1702 - 1706  and starts with step  1708 .) 
     Next, the worker process identifies a set of jobs to be executed, and also identifies a set of worker processes that are available to execute the set of jobs (step  1708 ). For example, the set of jobs to be executed can be determined from a static system configuration, and the set of available worker processes can obtained from the distributed consensus service. Next, the system sorts the set of worker processes based on their unique identifiers (step  1710 ). 
     Then, the worker process assigns one or more jobs to each worker process in the set of worker processes (step  1712 ) using a process that is outlined in the flow chart in  FIG. 18 . Note that the steps in this flow chart are repeated for each worker process in the sorted order. (In this way, each worker process simulates the decisions made by all other worker processes, and this enables each worker process to locally determine the job distribution that would otherwise be determined by global consensus.) First, the system determines a desired number of jobs that the worker process will execute by dividing the number of jobs in the set of jobs by the number of worker processes in the set of worker processes. If necessary, the system adds one to the result of this division for some of the worker processes to account for a remainder of the division operation (step  1802 ). Next, the system uses the unique ID for the work process to seed a deterministic pseudorandom number generator (step  1804 ). Then, the system repeats the following operations until the desired number of jobs is selected for the worker process to execute. First, the system obtains a random number from the deterministic pseudorandom number generator (step  1806 ). Next, the system uses the random number and a modulo operation to randomly select a job from the set of jobs (step  1808 ). If the job has already been selected, the system repeats the process of obtaining a random number and using the random number to select a job until an unselected job is selected (step  1810 ). 
     Finally, after the worker process finishes assigning the set of jobs to the set of processes, the worker process commences executing the one or more jobs that are assigned to the worker process itself (step  1714 ) and, if necessary, stops executing jobs that are no longer assigned to the worker process (step  1716 ). 
     Note that because jobs are allocated to worker processes in sorted order, and the identifiers are issued to worker processes using an incrementing counter, older worker processes will select their jobs first. Moreover, because each worker process always seeds the deterministic pseudorandom number generator with the same unique identifier, the older worker processes will tend to select the same jobs that they selected previously. This means that when jobs are reassigned older worker processes will tend to keep the same jobs, and newer worker processes (that have the higher identifiers) will tend to have more volatile job assignments. 
     Also, some amount of lag time is involved in starting and stopping jobs. Hence, if a job is reassigned from a first process to a second process, the system either needs to use locks to ensure that the same work is not performed by the first and second processes, or alternatively, the system needs to ensure that having multiple processes repeat the same work does not lead to incorrectness. Because locks can severely degrade performance, it is undesirable to use locks. Hence, in many scenarios the system ensures that the work performed by each job is “idempotent” so that the work can be repeated by multiple processes without violating correctness. 
     For example, if the work involves processing puts and deletes stored in a cross-zone replication queue, multiple processes can perform the same work in parallel without violating correctness. The processing of a cross-zone replication queue can be further optimized by requiring that the queue pointer can only be advanced and can never be set backwards. In this case, a process that completes an item of work first will successfully increment the queue pointer, whereas a process that completes the same item of work at a later time will not increment the queue pointer. This restriction on advancing the queue pointer can be enforced by maintaining the queue pointer in a SQL database that supports ACID transactions, and requiring that any SQL clause that updates the queue pointer must include a clause that ensures the queue pointer can only be advanced in a forward direction and can never be set backwards. 
     The foregoing descriptions of embodiments have been presented for purposes of illustration and description only. They are not intended to be exhaustive or to limit the present description to the forms disclosed. Accordingly, many modifications and variations will be apparent to practitioners skilled in the art. Additionally, the above disclosure is not intended to limit the present description. The scope of the present description is defined by the appended claims.