HK1225119B - Checkpointing a collection of data units - Google Patents

Description

Checkpointing of a collection of data units

CROSS-REFERENCE TO RELATED APPLICATIONS

This application claims priority to U.S. application No. 61/893,439, filed on October 21, 2013.

Technical Field

The present application relates to checkpointing a collection of data units.

Background Art

There are various types of data processing systems in which the ability to recover or restart in response to a failure or other unexpected event is useful. For example, in real-time stream processing or complex event processing systems, it is useful to save system information such as input data and/or state information for computations performed on the input data. Checkpointing is an example of a method for periodically saving system information so that the system can be restored from the most recently saved consistent state. Checkpointing techniques for data processing systems operating on a continuous data stream are described in U.S. Patent No. 6,584,581, which is incorporated herein by reference.

Summary of the Invention

In one aspect, in general, a computing system for managing stored data includes: a memory module configured to store working data comprising a plurality of data units; a storage system configured to store recovery data comprising a plurality of groups of one or more data units; and at least one processor configured to manage transfers of the data units between the memory module and the storage system. The management includes: maintaining an order of the plurality of data units included in the working data, the order defining a first contiguous portion comprising one or more of the plurality of data units and a second contiguous portion comprising one or more of the plurality of data units; and, for each of a plurality of time intervals, identifying any data units accessed from the working data during the time interval and adding a group of two or more data units to the recovery data, the group of two or more data units including: one or more data units from the first contiguous portion including any accessed data units, and one or more data units from the second contiguous portion including at least one data unit previously added to the recovery data.

These aspects may include one or more of the following features.

The managing further comprises, for each of a plurality of time intervals, deleting from the recovery data at least one set of one or more data units for which any data units still included in the working data are stored in at least another set of one or more data units.

The managing further includes identifying, for each of a plurality of time intervals, any data units deleted from the working data during the time interval.

The second contiguous portion does not include any deleted data units.

The managing further includes, for each of the plurality of time intervals, adding information identifying any deleted data units to the restored data.

Identifying any data units accessed from the working data during the time interval includes moving any data units accessed from the working data during the time interval into the first contiguous portion.

The order among the plurality of data units included in the working data is established based on the order in which the data units were most recently accessed.

The first contiguous portion includes a most recently accessed data unit among all the plurality of data units included in the working data and a least recently accessed data unit of a subset of the plurality of data units that has not been added to the recovery data since being most recently accessed.

The second continuous portion does not overlap with the first continuous portion.

The second contiguous portion includes at least one data unit that has been added to the restored data since being most recently accessed.

The one or more data units from the second contiguous portion is limited to a number of data units that is between approximately half the number of data units in the one or more data units from the first contiguous portion and approximately twice the number of data units in the one or more data units from the first contiguous portion.

The first contiguous portion includes data units that were more recently accessed than any data units in the second contiguous portion.

The time indicating the time at which the data unit was most recently accessed corresponds to the time at which exclusive access to the data unit began.

The time indicating the time at which the data unit was most recently accessed corresponds to the time at which exclusive access to the data unit ends.

The managing also includes restoring the state of the working data using the recovery data in response to the failure.

Each of the plurality of data units included in the working data is associated with a key value.

At least one data unit included in the working data includes one or more values accessible based on a key value associated with the data unit.

The total number of different key values associated with different data units included in the working data is greater than approximately 1,000.

The time intervals do not overlap with each other.

Identifying any data unit accessed from the working data during the time interval includes identifying at least one of: any data unit added to the working data during the time interval, any data unit read from the working data during the time interval, or any data unit updated within the working data during the time interval.

The memory module includes a volatile memory device.

The storage system includes a non-volatile storage device.

On the other hand, in general, a method for managing the transfer of data units between a memory module of a computing system and a storage system of the computing system includes: storing working data including multiple data units in the memory module; storing recovery data including multiple groups of one or more data units in the storage system; maintaining an order of the multiple data units included in the working data, the order defining a first contiguous portion including one or more of the multiple data units and a second contiguous portion including one or more of the multiple data units; and for each of a plurality of time intervals, identifying any data unit accessed from the working data during the time interval, and adding a group of two or more data units to the recovery data, the group of two or more data units including: one or more data units from the first contiguous portion and including any accessed data unit, and one or more data units from the second contiguous portion and including at least one data unit that had previously been added to the recovery data.

In another aspect, generally, software is stored in a non-transitory manner on a computer-readable medium for managing the transfer of data units between a memory module of a computing system and a storage system of the computing system. The software includes instructions for causing the computing system to: store working data comprising a plurality of data units in the memory module; store recovery data comprising a plurality of groups of one or more data units in the storage system; maintain an order of the plurality of data units included in the working data, the order defining a first contiguous portion comprising one or more of the plurality of data units and a second contiguous portion comprising one or more of the plurality of data units; and for each of a plurality of time intervals, identify any data units accessed from the working data during the time interval and add a group of two or more data units to the recovery data, the group of two or more data units comprising: one or more data units from the first contiguous portion including any accessed data units and one or more data units from the second contiguous portion including at least one data unit previously added to the recovery data.

These aspects may include one or more of the following advantages.

In some data processing systems, the information maintained by the system includes working data, which comprises a collection of data units that are periodically updated as processing proceeds. In a Complex Event Processing (CEP) system, for example, event streams are processed and aggregated, and actions are taken based on the results. A set of working data for a CEP system may include the state of multiple data units, each representing an entry for which a different data stream is being received, such as prices associated with different stock symbols. The state represented by a particular data unit may be stored as a state object accessed by a unique key (e.g., a numeric value). In the stock symbol example, the system would maintain a state object for each stock symbol. Each state object may have one or more fields that store values such as scalars (e.g., a value associated with the stock) and vectors (e.g., historical price data). In some examples, a state object's fields may store calculated values representing the results of applying a function, such as an aggregation function (e.g., a sum, count, maximum, and average), which is incrementally updated with each new price value for the corresponding stock symbol. There may be multiple collections of state objects within the system, each of which may include a state object with a specific set of fields. The size of the collection can change as new state objects are added and old state objects are removed, but most of the changing data in the collection can be attributed to the actual data stored in the portion of the state objects that is being updated.

In some embodiments, the working data for the system is stored in a large amount of relatively fast memory, which can be volatile memory, such as dynamic random access memory (DRAM). To ensure the persistence of the working data, it can be particularly useful to employ a checkpointing scheme to periodically store the latest version of each state object on a more stable and reliable storage device (e.g., a hard disk drive, solid-state drive, or other non-volatile storage medium), while ensuring that the cost in terms of required resources (e.g., data transfer time and data storage space) is effectively managed. Although the working data can include a large number of data units (e.g., the state objects described above), at any given time between checkpoint operations (referred to as the "checkpoint interval"), a small fraction of the data units may have changed. The checkpointing scheme should enable the most recent state of each data unit to be restored in the event of a failure, including those data units that have recently changed as well as those that have not changed.

The art of efficient checkpointing in such a system is challenging, especially when the number of data units being managed is particularly large. For example, in one scenario, the working data comprises a collection of approximately one billion data units, each of which is several bytes, and during each checkpoint interval, approximately one thousandth of the data units (1 million data units) changes. Of course, the set of data units that have changed during each checkpoint interval can be a different (but possibly overlapping) set. For recovery purposes, it can also be assumed that only the most recent state of any particular data unit is required.

The first approach is to store the entire set of data units in the form of checkpoint files on the storage device at each checkpoint interval. To restore the latest state of each data unit in the set, the system can simply read the checkpoint file(s). This first approach is expensive in terms of data transfer costs for working data with gigabytes and checkpoint intervals in seconds, and may not even allow enough time to write all data units to the storage device.

The second approach is to store only the data units that have changed since the last checkpoint interval in the form of checkpoint files on the storage device at each checkpoint interval. While this second approach reduces the transfer time at each checkpoint, recovery becomes more expensive over time. This is because to restore the latest state of each data unit, the system must read each checkpoint file from the start of the checkpoint process to ensure that the latest state of the data units that have not changed since the first checkpoint file was written is restored.

A possible improvement to this second approach is for the system to perform an offline process in which the checkpoint files in the storage device are scanned and checkpoint files consisting entirely of old copies of data units with updated states represented by the most recently stored checkpoint files are deleted. Another improvement to this second approach is to scan the checkpoint files and rewrite them to delete old copies of data units, retaining at least one more recent copy of each deleted data unit in at least one more recent checkpoint file. When the number of data units in a checkpoint file drops to zero, the checkpoint file is deleted from the storage device to free up storage space. Some potential challenges of this improvement are: (a) the consolidation process introduces another process to be managed, which generally makes the system less reliable; and (b) the consolidation process has a computational time cost, including at least one read of the checkpoint data represented by the checkpoint file, but may also include several reads and writes of the checkpoint data.

A third approach is for the system to store a separate checkpoint file for each data unit. At each checkpoint interval, a new checkpoint file written for a particular updated data unit can replace the previous checkpoint file for that data unit. This will avoid high data transfer costs (since only the changed entries need to be stored), will avoid high data storage costs (since checkpoint files for obsolete data will not accumulate indefinitely), and will reduce complexity (since no cleanup process is required). However, a potential reason for the high price of this third approach is that files are relatively expensive relative to storage space. If the working data consists of billions of data units, each of which is only a few bytes, the file system may not be able to afford the file creation and management operations, and the storage space overhead for the file system metadata will end up consuming more storage space than the actual content of the data units.

Some of the checkpointing techniques described below share at least some of the advantages of the aforementioned methods and their improvements. For example, based on identifying recovery data, new copies of potentially changed data units and old copies of data units previously added to (different) checkpoint files can be included in the same checkpoint file, thereby limiting data storage costs (including file system overhead) and data transfer time. The process of incrementally migrating old copies to newer checkpoint files can be accomplished in a manner that limits the amount of additional work performed during a checkpoint interval and gradually consolidates old copies of the least recently accessed data units until the old checkpoint file can be discarded, as it only stores redundant backup copies of the data units. This consolidation process can be performed by the same checkpoint process that stores the checkpoint files and, therefore, may be simpler and more reliable than an offline consolidation process. Even if old checkpoint files are not deleted, storing the latest states of multiple data units (i.e., data units with different keys) in the same checkpoint file reduces potential data storage costs due to file system overhead, particularly when the number of different keys assigned to the data units is large (e.g., greater than approximately 1,000, or greater than approximately 1,000,000, or greater than approximately 1,000,000,000).

Other features and advantages of the invention will become apparent from the following description and from the claims.

BRIEF DESCRIPTION OF THE DRAWINGS

FIG1 is a block diagram of a data processing system.

2A and 2B are timelines of checkpointing and data processing operations.

3A and 3B are flow diagrams of checkpointing and recovery algorithms, respectively.

4A , 4E and 4F are schematic diagrams of working data states.

4B , 4C and 4D are schematic diagrams of checkpoint files.

FIG5 is a schematic diagram showing the movement of a pointer over time.

DETAILED DESCRIPTION

Figure 1 shows an example of a data processing system 100 to which checkpointing techniques may be applied. System 100 includes a data source 102, which may include one or more data sources, such as storage devices or online data sources, each of which may store or provide data in any of a variety of formats (e.g., a database table, a spreadsheet file, a plain text file, or a native format used by a host). An execution environment 104 includes a processing module 106 (e.g., at least one central processing unit having one or more processing cores) and a memory module 108 (e.g., one or more storage devices, such as DRAM or other forms of relatively fast storage media). Execution environment 104 may be hosted on one or more general-purpose computers controlled by an appropriate operating system (e.g., a version of the UNIX operating system). For example, the execution environment 104 may include a multi-node parallel computing environment that includes a configuration of a computer system using multiple central processing units (CPUs) or processor cores, which may be local (e.g., a multi-processor system such as a symmetric multi-processing (SMP) computer), or locally distributed (e.g., multiple processors coupled as a cluster or massively parallel processing (MPP) system), or remote or remotely distributed (e.g., multiple processors coupled via a local area network (LAN) and/or a wide area network (WAN)), or a combination of the above.

The storage device providing the data source 102 can be a storage device local to the execution environment 104, for example, stored on a storage medium 110 connected to the computer hosting the execution environment 104, or a storage device remote from the execution environment 104, for example, hosted on a remote system (e.g., host computer 112) that communicates with the computer hosting the execution environment 104 via a remote connection (e.g., a server connection for streaming data feeds). Output data generated by data processing within the execution environment 104 can be stored back to the data source 102 or other storage medium, or used in other ways.

Processing module 106 processes data from data source 102 for any of a variety of applications (e.g., complex event processing) and, during such processing, accesses working data 114 stored in memory module 108. Processing module 106 also periodically executes a checkpointing process that stores portions of working data 114 in a data storage system 116 accessible within execution environment 104 (e.g., a hard drive of a computer hosting execution environment 104). The checkpointing process may store certain portions of working data 114 in their entirety and only selectively store other portions to avoid redundantly backing up certain data that has not changed since the last checkpoint interval. For example, working data 114 may include a set of data units that the checkpointing process selectively stores in a set of checkpoint files 120 based on the order in which the data units are maintained. Other portions of working data 114, such as other in-memory states associated with data processing, may be stored in separate checkpoint files.

The order of the data units is maintained by a hypervisor, which can be part of a larger data processing program or a separate process that manages the working data 114. In some embodiments, the data units are stored in the memory module 108 in an associative array of key-value pair entries organized in the order of least-recently-accessed (LRA) to most-recently-accessed (MRA). For example, the associative array can be implemented as a hash table, and the entries in the hash table can be strung together using a doubly linked list pointer arrangement according to the maintenance order. Each entry in the hash table is accessed based on a unique key and can store data corresponding to the key (representing any number of independent values of a variable or other state information) in an enclosed data object, such as the keyed state object described above. The hypervisor maintains an MRA pointer to the MRA entries in the hash table and an LRA pointer to the LRA entries in the hash table. The hypervisor also maintains a pointer to the least recently set checkpoint (LRC) and a pointer to the most recently set checkpoint (MRC). Each entry in the hash table also has an associated attribute that stores the checkpoint number (CPN) corresponding to the checkpoint file that was last saved to the checkpoint file (if any). These pointers and fields are used to selectively determine which entries will be copied to the new checkpoint file in any given checkpoint interval, as described in more detail below.

Each time an entry is accessed, it becomes the most recently accessed entry, i.e., an MRA pointer is assigned to the memory address of the entry, and other pointers within the linked list are adjusted appropriately (e.g., for adjacent entries at older locations in the table). In some embodiments, an entry is considered accessed whenever its key is used to retrieve data stored in the entry, or when the entry has been added to the table. In such embodiments, an entry is still considered accessed when a program retrieves the entry's data for reading without changing it. In some embodiments, the hypervisor maintains order based on when the entry actually changed (e.g., most recently changed or least recently changed), and does not consider reading an entry's data without changing it to affect the order. Such embodiments may, for example, employ a "dirty bit" for each entry that changes due to an access. In the following example, a simpler approach is employed, assuming that any access may have changed the data stored in the entry. The hypervisor also determines when an entry is no longer needed and should be deleted from the table. For example, in some embodiments, entries are deleted from the table starting with the least recently accessed entry when memory or time constraints dictate.

The data processing system 100 provides an incremental checkpointing solution by executing a checkpointing process at checkpoint intervals. For example, a checkpointing process can be triggered after a predetermined time or after the system receives a predetermined number of input records. When the checkpointing process is triggered, the checkpoint file is stored with "new" copies of data units that have been accessed during the most recent checkpoint interval and a similar number of "old" copies of data units that have not been accessed during the most recent checkpoint interval and are already stored in the checkpoint file. The MRC pointers are adjusted as described in more detail below. These checkpointing operations performed by the checkpointing process can occur while the hypervisor continues to manage access to the working data 114 for data processing operations, or access to the working data 114 can be temporarily blocked while the checkpointing process is executed. Figures 2A and 2B illustrate example timelines (increasing time toward the right) illustrating the timing of checkpointing process operations for the most recent checkpoint interval. In the example of Figure 2A, the checkpointing process performs a checkpointing operation 200 to save the working data state of the activity that occurred during the previous checkpoint interval 202A while executing data processing operations 204 that continued during the checkpoint interval 202B. Alternatively, in the example of FIG2B , the checkpoint process completes a checkpoint operation 200 ′ for activity that occurred during the previous checkpoint interval 202A before data processing operations 204 ′ are resumed during the checkpoint interval 202B. To enable proper reconstruction of the working data 114 during recovery, the checkpoint process stores data units in a manner that enables new copies to be distinguished from old copies (e.g., using markers for each data unit or writing data units to a checkpoint file in a separate portion). As the checkpoint process writes older data to newer checkpoint files, the checkpoint files that previously stored those data units (now stored in the newer checkpoint files) can be deleted if certain conditions are met.

As the hypervisor manages access to the working data 114 during normal data processing, it updates the MRA pointer and the MRC pointer appropriately in preparation for the checkpoint process that will occur at the next checkpoint interval. For example, for a table of entries organized as a linked list as described above, if an MRC entry is accessed, the MRC entry becomes an MRA entry at one end of the table, and the MRC pointer is adjusted to reference the entry one step toward the LRA end of the table. When the next checkpoint process occurs, only the entries between the MRA (inclusive) pointer and the MRC (exclusive) pointer need to be stored. After the entries between the MRC pointer and the MRA pointer are stored in the checkpoint file, the checkpoint process sets the MRC pointer to the entry identified by the MRA pointer in preparation for the next checkpoint interval.

Another aspect of managing the working data 114 and the checkpoint files 120 that are relevant to the proper reconstruction of tables during recovery is tracking data units (e.g., table entries) that have been deleted (or indicated as no longer in use) from the working data 114. In order to clearly identify each data unit that has existed since the start of data processing, including those that have been deleted, each data unit is assigned a unique identifier (ID). Since the keys of table entries are unique, as long as the keys of deleted entries are not reused, these keys can be used as unique IDs. Otherwise, if the keys are to be reused, additional unique IDs can be assigned to each entry. In the following example, an 8-byte integer that is incremented each time a new entry is added to the table will serve as both such a unique ID and the key for the entry. When an entry is deleted from the table during a checkpoint interval, the hypervisor adds its ID to the list of deleted entries for that checkpoint interval and stores it as part of the checkpoint file.

Thus, a checkpoint file for a particular checkpoint interval may include a data structure (e.g., a table) storing the following two types of items:

(1) an entry storing the keys of entries that have been deleted since the last checkpoint interval, and

(2) A project that stores copies of entries (keys and corresponding data) that have been accessed since the last checkpoint interval.

A possible recovery process includes reading each checkpoint file in the order in which it was created. For each checkpoint file, the recovery process performed by the processing module 106 will perform the following steps:

(1) Delete any entry whose key is stored in an item of type (1) in the checkpoint file, and then

(2) Add or update an entry in the type (2) item stored in the checkpoint file.

While reading each checkpoint file from the first checkpoint file stored will result in correct behavior, improvements can be made so that the recovery process does not need to read every checkpoint file from the beginning of the checkpoint and reproduce every change to the table when recovering. By incrementally copying entries that were previously copied to old checkpoint files to new checkpoint files, the checkpoint process can eventually delete older checkpoint files that are no longer needed. This can speed up the recovery process and reduce data storage costs.

To ensure that old checkpoint files can be safely deleted without losing any saved state necessary to restore the most recent state of every entry in the table (since the most recently completed checkpoint interval), the hypervisor and checkpointing process together cause the LRC pointer to incrementally scan entries from the LRA side toward the MRA side, recording old entries that have been copied to newer checkpoint files. For each checkpoint interval, the checkpointing process saves as many old entries from the LRA side as it saves new entries from the MRA side. This way, the checkpointing process limits data transfer costs to a ratio proportional to the number of entries accessed since the last checkpoint interval. Because the checkpointing process writes old entries from the LRA side to newer checkpoint files, it is possible to delete previously part of the checkpoint files as long as they do not store old entries that have not been copied to newer checkpoint files. The recovery can then restore the latest table by reading the remaining checkpoint files in order from oldest to newest.

When the data processing system 100 begins processing data and managing working data 114 stored in the memory module 108, there may be an initial number of checkpoint intervals, wherein the initial checkpoint process creates an initial set of one or more checkpoint files 120 using only new data units. For example, this initial checkpoint process stores new entries at the MRA end of the table between the MRC pointer (exclusive) and the MRA pointer (inclusive). The LRC pointer is not used for this initial checkpoint process. After a certain number of initial checkpoint files 120 have been stored, a normal (steady-state) checkpoint process also begins storing old data units (i.e., the table of entries) and new data units. When a normal checkpoint process begins, the LRC pointer is initially set to the LRA pointer. The checkpoint process then stores a limited number of old entries at the LRA end of the table between the LRC pointer (exclusive) and the MRC pointer (exclusive), or the number of new entries from the LRC pointer (inclusive) to the checkpointed entry, whichever is less.

The following example of the algorithm employed by the checkpoint process is written in pseudocode. The pseudocode includes functional comments that describe the statements and functions of the pseudocode. The pseudocode uses standard C programming language syntax for conditional statements (e.g., 'if' statements) and loops (e.g., 'while' loops and 'for' loops), as well as for comments (prefixed with '//'). In the pseudocode listing, the MRA pointer is in the variable 'mra', the LRA pointer is in the variable 'lra', the MRC pointer is in the variable 'mrc', and the LRC pointer is in the variable 'lrc'. Dot notation is used with these variables to represent the portions of the entries identified by these pointers. In particular, the dot notation 'pointer.prev' and 'pointer.next' are used to represent the position in the table that is one step closer to the MRA end and the LRA end from the entry pointed to by 'pointer', and the dot notation 'pointer.checkpoint_number', 'pointer.key', and 'pointer.data' are used to represent the CPN, key, and data of the entry pointed to by 'pointer', respectively. Dot notation is also used to represent calling some function associated with a variable, such as 'item.is_<property>()' to test whether the item represented by the variable 'item' has the property '<property>'. The following algorithm may be executed by the checkpoint process for each checkpoint interval.

//Start with an empty list of checkpoint files to delete

files_to_remove=empty_list();

//Open a new (empty) checkpoint file

//(named after the current checkpoint number)

checkpoint_file=open_checkpoint_file(checkpoint_number);

//Advance the MRC pointer (find the first new entry)

mrc = mrc.prev;

//while loop to copy all visited items

//During the most recent checkpoint interval

// and an equal number of old entries:

while (mrc!=mra) {

//Write the current new entry into the current checkpoint file

// Set 'New?' to true

write(checkpoint_file,checkpointed_entry(mrc,true));

//Record the checkpoint number to the current new entry

mrc.checkpoint_number=checkpoint_number;

//Advance the MRC pointer

mrc = mrc.prev;

// If LRC hasn't caught up to MRC yet...

if(lrc!=mrc){

//...Write the LRC entries to the current checkpoint file

//Set 'New?' to false

write(checkpoint_file,checkpointed_entry(lrc,false));

// If the LRC entry is the latest entry in its old checkpoint file

// Then delete the file

if(lrc.checkpoint_number!=lrc.prev.checkpoint_number)

files_to_remove.add(lrc.checkpoint_number);

//Record the new checkpoint number

lrc.checkpoint_number=checkpoint_number;

//Advance the LRC pointer

lrc = lrc.prev;

}

}

//MRC is now MRA (exit from while loop)

//If LRC catches up with MRC, then set it back to LRA

if(lrc == mrc)

lrc=lra;

// Record deleted entries with the keys of all entries deleted in this interval:

for(key in removed_keys)

write(checkpoint_file,checkpointed_removal(key));

//Record LRC-type entries with LCR keys:

write(checkpoint_file,checkpointed_lrc(lrc.key));

//Advance checkpoint number

checkpoint_number++;

// Delete the files listed for deletion

for(file in files_to_remove)

remove_checkpoint_file(file);

The algorithm described above is also shown in the flowchart of FIG3A . The checkpoint process initializes 300 a checkpoint file, deletes the empty list of checkpoint files, adds a new empty checkpoint file, and advances the MRC pointer to the first new entry. The process executes a while loop 302 with a loop condition, which loops until the MRC pointer equals the MRA pointer. Within while loop 302, the process writes 304 the current new entry to the current checkpoint file, records 306 the current CPN into the current MRC entry, and advances 308 the MRC pointer. The process checks 310 to determine whether the LRC has reached the MRC. If so, it returns to the next while loop iteration; if not, it continues. If the while loop 302 continues, the process writes 312 the LRC entry to the current checkpoint file. The process then checks 314 to determine whether the LRC entry is the latest entry in its old checkpoint file. If so, it marks 316 the checkpoint file for deletion by adding it to the list of checkpoint files to be deleted. The process then records 318 the current CPN to the current LRC entry and advances 320 the LRC pointer before returning to the next while loop iteration. After exiting the while loop 302, the process checks 322 to determine whether the LRC has reached the MRC and, if so, sets the LRC back to the LRA. The process then records 326 a deletion-type entry with the key of all entries deleted during the checkpoint interval, records 328 an LRC-type entry, advances 330 the CPN, and deletes 332 the checkpoint file marked for deletion.

The following is an example of an algorithm, written in pseudocode, used by the recovery process.The following algorithm can be implemented to recover the most recent consistent state (ie, the most recently checkpointed state) of the entry table after a failure.

The above algorithm is also shown in the process flow of Figure 3B. The recovery process initializes 340 the MRA pointer and the LRA pointer to null values. The recovery process then executes a nested For loop, with an outer For loop 344 iterating over the checkpoint files (oldest to newest) and an inner For loop 346 iterating over the items in each checkpoint file. Within the inner For loop 346, if the item is a delete-type item, the process deletes the item. The process then checks 350 whether the item is an entry-type item. If so, it searches 352 for an entry with the specified key and updates the entry with the specified data, or if no entry is found, it creates an entry with the specified key and data. The updated or created entry is inserted 354 into the appropriate position in the table by setting pointers. After insertion, or if the item is not an entry-type item, if it is an LRC-type item, the process sets 356 the LRC to the entry with the specified key. The two For loops then return for the next iteration. After both For loops exit, the process sets the MRC to the MRA.

Other examples of algorithms that can be used by the checkpoint process may include other steps. For example, the checkpoint files may be compressed. Checkpoint files may be merged into fewer than one physical file per checkpoint interval. For example, there may be periodic switching that allows the checkpoint process to delete older copies of table entries that already have updated copies stored in a constant-time operation. A list of older entries may be maintained based on entry modification rather than simple access. The process can simply delete all checkpoint files associated with checkpoint numbers less than the checkpoint number for the last LRC entry, rather than recording the checkpoint number of each checkpoint file to be deleted when the LRC pointer is advanced. When writing new entries, the process can count these new entries and write the same (or similar) number of old entries after writing these new entries to separate sections of the checkpoint file (e.g., a section for new entries, a section for old entries, a section for entries with deleted keys, and a section for entries with keys of LRC entries), rather than interleaving new and old entries in the checkpoint file and using markers to distinguish between them.

4A through 4F illustrate examples of different states of a table of entries and other in-memory states within working data 114, as well as examples of different states of a checkpoint file 120. FIG4A shows a portion of a table 400 storing entries for key/data pairs and a CPN identifying the most recent checkpoint file in which the key/data pairs are stored, as well as the locations of the MRA pointer, LRA pointer, MRC pointer, and LRC pointer. FIG4A also shows a list 402 of deleted keys. Table 400 and list 402 are shown for one state before the checkpoint process generates a checkpoint file with CPN 201. For simplicity, the content of the data for a given key is represented by a single letter in the example.

Figures 4B and 4C show some collections of existing checkpoint files (CPN193-CPN200) before generating a checkpoint file with CPN201. Figure 4B shows earlier checkpoint files 193-194, and Figure 4C shows the latest checkpoint files 198-200. In this example, the checkpoint files are displayed as a table with items as rows in the table. Each item has one of three possible values for the 'Type' field: E (Entry Type), R (Delete Type), or L (LRC Type). Entry type items have a key value for the 'Key' field, a data value for the 'Data' field, and a T (True) or F (False) for the 'True (New)?' field, indicating whether the entry is a "new" entry or an "old" entry, respectively. As described above, an "old" entry is an entry that has been stored in a previous checkpoint file and is now being stored again in a newer checkpoint file. A delete type item or an LRC type item has a key value for the 'Key' field, but does not have a key value for the 'Data' or 'True (New)?' field. ’ field value.

Executing the illustrative algorithm described in the pseudocode above for a checkpoint interval ending with table 400 and list 402 as shown in FIG4A and checkpoint files as shown in FIG4B-4C, the checkpoint process will store a new checkpoint file with CPN 201 as shown in FIG4D. The checkpoint process will also delete the checkpoint file with CPN 193 because the transition of the LRC pointer from the last checkpointed entry in that checkpoint file to the last checkpointed entry in the checkpoint file with CPN 194 indicates that any copies of the entry in the checkpoint file with CPN 193 are no longer needed. After the checkpoint process has stored the checkpoint file with CPN 201, and before any entry has had a chance to be accessed since the state of FIG4A, table 400 and list 402 have the state shown in FIG4E.

If a system failure occurs after storing the checkpoint file with CPN 201, the system can perform a recovery process to restore the state of table 400 and list 402 (as shown in FIG4E ) by processing only the remaining checkpoint files (with CPNs 194-201) in the order of their CPNs. FIG4F shows a series of snapshots of the state of table 400 as it is rebuilt. A portion of the state of table 400 is shown after processing the first checkpoint file 194 and after processing each of checkpoint files 198-200. Then, after processing checkpoint file 201, the fully recovered state of table 400 is restored, including the pointers in table 400 (as shown in FIG4E ). The MRA pointer and MRC pointer are set to the first entry in table 400, and the LRA pointer is set to the last entry in table 400. The LRC pointer value is restored using the LRC-type entry of the last checkpoint file, which in this example is checkpoint file 201.

Figure 5 shows a depiction of how different pointers move over time for a series of nine different checkpoint intervals, with the textured bars for each checkpoint interval representing different portions of the table. The changes in the sizes of the different portions over time illustrate the LRC pointer catching up with the MRC pointer and looping back to the LRA pointer. In this example, there is a steady number of data units being accessed during the series of checkpoint intervals, and the size of the table is steady. Below each bar is the current CPN of the checkpoint file being written for that checkpoint interval, and below that is a list of the CPNs of the remaining valid checkpoint files that store the state of the table (after any old checkpoint files have been deleted). In most use cases, including this example, the set of remaining checkpoint files is fairly stable, with an average of one checkpoint file being deleted per checkpoint interval.

For example, the checkpointing method described above can be implemented using a programmable computing system that executes appropriate software instructions, or it can be implemented in appropriate hardware such as a field programmable gate array (FPGA), or in some hybrid form. For example, in a programmatic approach, the software can include processes in one or more computer programs that execute on one or more programmed or programmable computing systems (which can have various architectures, such as distributed, client/server, or grid-based), each computing system including at least one processor, at least one data storage system (including volatile and/or non-volatile memory and/or storage elements), and at least one user interface (for receiving input using at least one input device or port, and for providing output using at least one output device or port). The software can include one or more modules of a larger program, for example, the larger program provides other services related to the design, configuration, and execution of data flow graphs. The modules of the program (e.g., elements of a data flow graph) can be implemented as data structures or other organized data that conform to a data model stored in a database.

The software can be provided on a tangible permanent storage medium such as a CD-ROM or other computer-readable medium (which can be read by a general or special computing system or device) or delivered (encoded into a propagation signal) to a tangible permanent medium of a computing system that executes the software via a communication medium of a network. Some or all of the processing can be performed on a dedicated computer, or performed using dedicated hardware such as a coprocessor or a field programmable gate array (FPGA) or an application-specific integrated circuit (ASIC). The processing can be implemented in a distributed manner in which different computing parts specified by the software are performed by different computing elements. Each such computer program is preferably stored in or downloaded to a computer-readable storage medium (e.g., a solid-state memory or medium, or a magnetic or optical medium) of a storage device that can be read by a general or special programmable computer, for configuring and operating the computer when the computer reads the storage device medium to perform the processing described herein. It is also contemplated that the system of the present invention can be implemented as a tangible permanent storage medium configured with a computer program, wherein the storage medium so configured causes the computer to operate in a specific and predefined manner to perform one or more processing steps described herein.

Several embodiments of the present invention have been described. However, it should be understood that the foregoing description is intended to illustrate, not to limit, the scope of the present invention, which is defined by the following claims. Therefore, other embodiments are also within the scope of the following claims. For example, various modifications may be made without departing from the scope of the present invention. Furthermore, some of the steps described above may be sequence-independent and, therefore, may be performed in an order different from that described.

Claims (25)

1. A computing system for managing stored data, the computing system comprising: The memory module is configured to store working data comprising multiple data units; The storage system is configured to store recovery data comprising multiple sets of one or more data units; and At least one processor is configured to manage the transfer of data units between the memory module and the memory system, the management including: Maintaining the order of the plurality of data units included in the working data, the order defining a first consecutive portion including one or more data units and a second consecutive portion including one or more data units; and For each of the plurality of time intervals, any data unit accessed from the working data during that time interval is identified, and a set of two or more data units is added to the recovery data, the set of two or more data units comprising: one or more data units from the first consecutive portion, the one or more data units comprising any data unit that has been accessed, and one or more data units from the second consecutive portion, the one or more data units comprising at least one data unit that has been previously added to the recovery data. 2. The computing system of claim 1, wherein the management further comprises: for each of a plurality of time intervals, deleting at least one set of one or more data units from the recovery data, wherein for the set of one or more data units, any data units still included in the working data are stored in at least another set of one or more data units. 3. The computing system of claim 1, wherein the management further comprises: for each of a plurality of time intervals, identifying any data unit deleted from the working data during the time interval. 4. The computing system of claim 3, wherein the second continuous portion does not include any deleted data units. 5. The computing system of claim 3, wherein the management further comprises: for each of a plurality of time intervals, adding information identifying any deleted data units to the recovered data. 6. The computing system of claim 1, wherein identifying any data unit accessed from the working data during the time interval includes moving the any data unit accessed from the working data during the time interval into the first continuous portion. 7. The computing system of claim 1, wherein the order among the plurality of data units included in the working data is based on the order in which the data units were recently accessed. 8. The computing system of claim 7, wherein the first continuous portion comprises: the most recently accessed data unit among all the plurality of data units included in the working data, and the least recently accessed data unit in a subset of the plurality of data units that have not been added to the recovery data since the most recent access. 9. The computing system of claim 8, wherein the second continuous portion does not overlap with the first continuous portion. 10. The computing system of claim 9, wherein the second continuum comprises at least one data unit that has been added to the recovered data since the most recent access. 11. The computing system of claim 10, wherein the one or more data units from the second consecutive portion are limited to a number of data units, the number of data units being between approximately half the number of data units in the one or more data units from the first consecutive portion and approximately twice the number of data units in the one or more data units from the first consecutive portion. 12. The computing system of claim 11, wherein the first contiguous portion includes data units that are accessed more closely than any data unit in the second contiguous portion. 13. The computing system of claim 1, wherein the time at which the data unit was recently accessed corresponds to the time at which exclusive access to the data unit began. 14. The computing system of claim 1, wherein the time at which the data unit was recently accessed corresponds to the time at which exclusive access to the data unit ended. 15. The computing system of claim 1, wherein the management further includes using the recovery data to restore the state of the working data in response to a fault. 16. The computing system of claim 1, wherein each of the plurality of data units included in the working data is associated with a key value. 17. The computing system of claim 16, wherein at least one data unit included in the working data comprises one or more values accessible based on a key associated with the data unit. 18. The computing system of claim 16, wherein the total number of different key values associated with the different data units included in the working data is greater than approximately 1,000. 19. The computing system of claim 1, wherein the time intervals do not overlap with each other. 20. The computing system of claim 1, wherein identifying any data unit accessed from the working data during the time interval includes identifying at least one of the following: any data unit added to the working data during the time interval, any data unit read from the working data during the time interval, or any data unit updated within the working data during the time interval. 21. The computing system of claim 1, wherein the memory module comprises a volatile storage device. 22. The computing system of claim 1, wherein the storage system comprises a non-volatile storage device. 23. A computing system for managing stored data, the computing system comprising: The memory module is configured to store working data comprising multiple data units; The storage system is configured to store recovery data comprising multiple sets of one or more data units; and A means for managing the transfer of data units between the memory module and the storage system, the management including: Maintaining the order of the plurality of data units included in the working data, the order defining a first consecutive portion including one or more data units and a second consecutive portion including one or more data units; and For each of the multiple time intervals, any data unit accessed from the working data during that time interval is identified, and a set of two or more data units is added to the recovered data, the set of two or more data units comprising: one or more data units from the first consecutive portion. Multiple data units, including any accessed data unit, and one or more data units from the second contiguous portion, including at least one data unit previously added to the recovered data. 24. A method for managing the transfer of data units between a memory module of a computing system and a storage system of the computing system, the method comprising: The working data, which includes multiple data units, is stored in the memory module; The recovery data, comprising multiple groups of one or more data units, is stored in the storage system; Maintaining the order of the plurality of data units included in the working data, the order defining a first consecutive portion including one or more data units and a second consecutive portion including one or more data units; and For each of the plurality of time intervals, any data unit accessed from the working data during that time interval is identified, and a set of two or more data units is added to the recovery data, the set of two or more data units comprising: one or more data units from the first consecutive portion, the one or more data units comprising any data unit that has been accessed, and one or more data units from the second consecutive portion, the one or more data units comprising at least one data unit that has been previously added to the recovery data. 25. A computer-readable medium having a computer program stored thereon, the computer program being used to manage the transfer of data units between a memory module of a computing system and a storage system of the computing system, the computer program including instructions for causing the computing system to perform the following operations: The working data, which includes multiple data units, is stored in the memory module; The recovery data, comprising multiple groups of one or more data units, is stored in the storage system; Maintaining the order of the plurality of data units included in the working data, the order defining a first consecutive portion including one or more data units and a second consecutive portion including one or more data units; and For each of the plurality of time intervals, any data unit accessed from the working data during that time interval is identified, and a set of two or more data units is added to the recovery data, the set of two or more data units comprising: one or more data units from the first consecutive portion, the one or more data units comprising any data unit that has been accessed, and one or more data units from the second consecutive portion, the one or more data units comprising at least one data unit that has been previously added to the recovery data.