Efficient recovery of transactional data stores

• US 8,954,385 B2
• Filed: 06/28/2011
• Issued: 02/10/2015
• Est. Priority Date: 06/28/2010
• Status: Active Grant

• Current Assignee: Jpmorgan Chase Bank NA As The Agent
• Original Assignee: SanDisk Enterprise IP LLC (Western Digital Corporation)
• Inventors: George, Johann, Dinker, Darpan, Krishnan, Manavalan, O'Krafka, Brian W.
• Primary Examiner(s): HICKS, MICHAEL J
• Application Number:
  US13/170,955
• Publication Number:
  US 20120005154A1
• Time in Patent Office: 1,323 Days
• US Class Current: 707/617
• CPC Class Codes: G06F 16/273 Asynchronous replication or reconciliation
• CPC Class Codes: G06F 16/275 Synchronous replication

Abstract

Approaches for recovering a node of a distributed transactional system. When a recovering node is being brought on-line, a copy of at least a portion of a first data store maintained by an existing node is created without ceasing to process transactions against the first data store at the existing node. The recovering node creates a second data store based on the copy of the first data store. While the recovering node creates the second data store, the recovering node stores committed transaction data received from a plurality of nodes. The committed transaction data describes transactions committed against data stores maintained by the sender of the committed transaction data. The recovering node may thereafter replay, against the second data store, any transactions identified by the committed transaction data that are associated with a global transaction number that is more recent than is a reference commit number.

Claims

1. A method of recovering a respective node of a distributed transactional system that comprises three or more nodes, the method comprising:

at a second node of the distributed transactional system having a second set of one or more storage devices;

receiving a copy of at least a portion of a first data store from a first node of the distributed transactional system having a first set of one or more storage devices;

in response to receiving the copy of at least the portion of the first data store, creating a second data store based on the copy of at least the portion the first data store;

concurrent with the second node creating the second data store;

receiving committed transaction data from a plurality of nodes other than the second node, wherein the plurality of nodes at least includes the first node, and a third node of the distributed transactional system having a third set of one or more storage devices, and wherein the first, second, and third nodes each execute at least one distinct instance of the distributed transaction system; and

storing the committed transaction data, wherein a respective portion of the committed transaction data describes a transaction committed against a respective data store maintained by a corresponding node of the plurality of nodes other than the second node; and

the method further comprising, at the second node;

identifying a reference commit number, wherein the reference commit number is a most recent global commit number for transactions committed against the copy of at least the portion of the first data store;

identifying, from the committed transaction data stored by the second node, one or more transactions having a global commit number that is lower than or equal to the reference commit number;

disregarding the identified one or more transactions having a global commit number that is lower than or equal to the reference commit number; and

replaying, against the second data store, any transactions identified by the committed transaction data as having a global commit number that is greater than the reference commit number.

2. The method of claim 1, further comprising:

the second node receiving buffer content information identifying a set of pages stored within a first in-memory buffer pool at the first node; and

the second node pre-fetching pages within a second in-memory buffer pool maintained at the second node based upon the received buffer content information.

3. The method of claim 2, wherein the second node simultaneously pre-fetches pages into the second in-memory buffer pool while creating the second data store or replaying, against the second data store, any transactions identified by the committed transaction data received by the second node.

4. The method of claim 1, wherein the distributed transactional system maintains a bandwidth limit for recovering the respective node of the distributed transactional system;

and

the method further comprising, in response to determining that the bandwidth limit is met or exceeded, limiting an amount of network bandwidth available for recovering the respective node of the distributed transactional system.

5. The method of claim 4, wherein the amount of bandwidth available for recovering the respective node of the distributed transactional system is limited using a bandwidth limiting utility.

6. The method of claim 4, wherein the amount of bandwidth available for recovering the respective node of the distributed transactional system is limited using one or more of the following networking utilities:

iptables, squid proxy, and tc.

7. The method of claim 4, wherein the amount of bandwidth available for recovering the respective node of the distributed transactional system is limited using a configurable threshold that is manually established.

8. The method of claim 4, wherein the amount of bandwidth available for recovering the respective node of the distributed transactional system is limited using a configurable threshold that is dynamically established by:

auto-detecting the bandwidth limit of a network between the first node and the second node; and

establishing a data transfer limit for recovering the respective node of the distributed transactional system to be a predefined percentage of the bandwidth limit of the network between the first node and the second node.

9. The method of claim 1, wherein storing the committed transaction data comprises storing the committed transaction data to hard-disk prior to the second node processing the transactions identified in the committed transaction data.

10. The method of claim 9, further comprising, the second node pre-fetching, from committed transaction data stored on the hard-disk, a plurality of buffered write operations to be processed, wherein the second node pre-fetches the plurality of buffered write operations in parallel with processing a different plurality of buffered write operations that were previously fetched from the hard-disk.

11. The method of claim 1, wherein the respective node of the distributed transactional system is the second node;

and

the method further comprising, determining a measure of progress of recovering the second node, wherein determining the measure of progress comprises;

determining a size of the copy of the first data store;

monitoring a size of the portion of the copy of the first data store received at the second data store; and

periodically computing a percentage of the amount of the copy of the first data store received at the second data store relative.

12. The method of claim 1, determining a measure of progress of recovering the respective node of the distributed transactional system, wherein determining the measure of progress comprises:

determining an amount of committed transaction data stored on the hard-disk prior to reading any of the buffered write operations from the hard-disk;

monitoring a change in the amount of committed transaction data stored on the hard-disk as buffered write operations are read from the hard-disk; and

periodically computing a percentage of the amount of committed transaction data read from the hard-disk.

13. The method of claim 1, wherein the first data store is persistently maintained on a first set of solid state devices;

wherein the second data store is persistently maintained on a second set of solid state devices; and

the method further comprising, at the second node, pre-fetching one or more data blocks referenced by a plurality of buffered write operations prior to performing, in parallel, the plurality of buffered write operations against the second data store.

14. The method of claim 1, wherein the plurality of nodes other than the second node are configured to not send, to the second node, any committed transaction data for transactions having a global commit number lower than or equal to the reference commit number.

15. The method of claim 1, wherein:

the respective node of the distributed transactional system is the second node;

prior to recovering the second node, the second data store stores a partial set of data; and

receiving the copy of at least the portion of the first data store, at the second node, comprises;

at the second node;

communicating with the first node to determine if the first node possesses replication logs; and

upon determining that the first node possesses replication logs, instructing the first node to create a copy of the difference between the current state of the first data store and the second data store.

16. The method of claim 1, wherein the copy of at least the portion of a first data store is a full copy of the first data store.

17. A non-transitory machine-readable storage medium storing one or more sequences of instructions for recovering a respective node of a distributed transactional system that comprises three or more nodes, which when executed by one or more processors, cause the distributed transactional system to:

at a second node of the distributed transactional system having a second set of one or more storage devices;

receive a copy of at least a portion of a first data store from a first node of the distributed transactional system having a first set of one or more storage devices;

in response to receiving the copy of at least the portion of the first data store, create a second data store based on the copy of the first data store;

concurrent with the second node creating the second data store;

receive committed transaction data from a plurality of nodes other than the second node, wherein the plurality of nodes at least includes the first node, and a third node of the distributed transactional system having a third set of one or more storage devices, and wherein the first, second, and third nodes each execute at least one distinct instance of the distributed transaction system; and

store the committed transaction data, wherein a respective portion of the committed transaction data describes a transaction committed against a respective data store maintained by a corresponding node of the plurality of nodes other than the second node; and

the one or more sequences of instructions, further cause the second node of the distributed transaction system to;

identify a reference commit number, wherein the reference commit number is a most recent global commit number for transactions committed against the copy of at least the portion of the first data store;

identify, from the committed transaction data stored by the second node, one or more transactions having a global commit number that is lower than or equal to the reference commit number;

disregard the identified one or more transactions having a global commit number that is lower than or equal to the reference commit number; and

replay, against the second data store, any transactions identified by the committed transaction data as having a global commit number that is greater than the reference commit number.

18. The storage medium of claim 17, wherein:

the respective node of the distributed transactional system is the second node;

prior to recovering the second node, the second data store stores a partial set of data; and

receiving the copy of at least the portion of the first data store, at the second node, comprises;

at the second node;

communicating with the first node to determine if the first node possesses replication logs; and

upon determining that the first node possesses replication logs, instructing the first node to create a copy of the difference between the current state of the first data store and the second data store.

19. A system for recovering a respective node of a distributed transactional system that comprises three or more nodes, the system comprising:

one or more processors; and

a non-transitory machine-readable storage medium storing one or more sequences of instructions, which when executed by the one or more processors, cause the distributed transactional system to;

at a second node of the distributed transactional system having a second set of one or more storage devices;

receive a copy of at least a portion of a first data store from a first node of the distributed transactional system having a first set of one or more storage devices;

in response to receiving the copy of at least the portion of the first data store, create a second data store based on the copy of the first data store;

concurrent with the second node creating the second data store;

receive committed transaction data from a plurality of nodes other than the second node, wherein the plurality of nodes at least includes the first node, and a third node of the distributed transactional system having a third set of one or more storage devices, and wherein the first, second, and third nodes each execute at least one distinct instance of the distributed transaction system; and

store the committed transaction data, wherein a respective portion of the committed transaction data describes a transaction committed against a respective data store maintained by a corresponding node of the plurality of nodes other than the second node; and

the one or more sequences of instructions, further cause the second node of the distributed transaction system to;

identify a reference commit number, wherein the reference commit number is a most recent global commit number for transactions committed against the copy of at least the portion of the first data store;

identify, from the committed transaction data stored by the second node, one or more transactions having a global commit number that is lower than or equal to the reference commit number;

disregard the identified one or more transactions having a global commit number that is lower than or equal to the reference commit number; and

replay, against the second data store, any transactions identified by the committed transaction data as having a global commit number that is greater than the reference commit number.

20. The system of claim 19, wherein:

the respective node of the distributed transactional system is the second node;

prior to recovering the second node, the second data store stores a partial set of data; and

receiving the copy of at least the portion of the first data store, at the second node, comprises;

at the second node;

communicating with the first node to determine if the first node possesses replication logs; and

upon determining that the first node possesses replication logs, instructing the first node to create a copy of the difference between the current state of the first data store and the second data store.