Executing distributed globally-ordered transactional workloads in replicated state machines

• US 9,465,649 B2
• Filed: 04/15/2013
• Issued: 10/11/2016
• Est. Priority Date: 04/15/2013
• Status: Active Grant

• Current Assignee: International Business Machines Corporation
• Original Assignee: International Business Machines Corporation
• Inventors: Meng, Shicong, Zhang, Li
• Primary Examiner(s): Waldron, Scott A
• Assistant Examiner(s): Wang, Dongming
• Application Number:
  US13/863,065
• Publication Number:
  US 20140310239A1
• Time in Patent Office: 1,275 Days
• Field of Search: 707/615
• US Class Current: 1/1
• CPC Class Codes: G06F 9/466 Transaction processing

Abstract

A method of transaction replication includes transmitting at least one transaction received during an epoch from a local node to remote nodes of a domain of 2N+1 nodes at the end of an epoch (N is an integer greater than or equal to 1). The remote nodes log receipt of the at least one transaction, notify the local node of the receipt of the at least one transaction, transmit the at least one transaction to all of the 2N+1 nodes, and add the at least one transaction to an execution order upon receiving at least N+1 copies of the at least one transaction.

Claims

1. A method of transaction replication, comprising:

transmitting at least one transaction received during an epoch, by a local datastore node, to at least one remote datastore node of a plurality of remote datastore nodes upon an end of the epoch, wherein the epoch corresponds to a predefined time period having a fixed length, and the local datastore node and the plurality of remote datastore nodes are part of a domain of 2N+1 datastore nodes, wherein N is an integer greater than or equal to 1;

logging a receipt of the at least one transaction, by the at least one remote datastore node, upon receiving the at least one transaction at the at least one remote datastore node;

notifying the local datastore node of the receipt of the at least one transaction, by the at least one remote datastore node, upon receiving the at least one transaction at the at least one remote datastore node;

transmitting the at least one transaction to all datastore nodes in the domain other than the local datastore node and the at least one remote datastore node, by the at least one remote datastore node, upon receiving the at least one transaction at the at least one remote datastore node; and

adding the at least one transaction to an execution queue at each of the datastore nodes of the domain, wherein each datastore node does not add the at least one transaction to its execution queue until at least N+1 copies of the at least one transaction have been received at that datastore node,

wherein the at least one transaction is transmitted from the local datastore node to the at least one remote datastore node, and from the at least one remote datastore node to all datastore nodes other than the local datastore node and the at least one remote datastore node, without being transmitted to a centralized sequencer.

2. The method of claim 1, further comprising:

measuring an elapsed time, by the at least one remote datastore node, since last logging the receipt of the at least one transaction;

transmitting a timeout message to all datastore nodes in the domain, by the at least one remote datastore node, upon the elapsed time exceeding a predefined timeout threshold;

transmitting a timeout confirmation message to all datastore nodes in the domain, by each of the datastore nodes of the domain, upon receiving at least N+1 timeout confirmation messages regarding the local datastore node at each of the datastore nodes; and

removing the local datastore node from a participating datastore node list, by each of the datastore nodes of the domain, upon receiving at least N+1 timeout confirmation messages regarding the local datastore node at each of the datastore nodes.

3. The method of claim 2, further comprising:

discarding unfinished processes that have occurred prior to removal from the participating datastore node list, by the local datastore node, upon being removed from the participating datastore node list;

transmitting a recovery notification message, to each of the datastore nodes in the domain, by the local datastore node; and

adding the local datastore node to the participating datastore node list, by each of the datastore nodes of the domain, upon each of the datastore nodes receiving at least N+1 copies of the recovery notification message.

4. The method of claim 3, wherein the local datastore node is added to the participating datastore node list by each of the datastore nodes of the domain during different epochs.

5. The method of claim 1, further comprising:

generating a local signature corresponding to the local datastore node, by the local datastore node;

transmitting the local signature, by the local datastore node, to all datastore nodes in the domain;

generating a remote signature corresponding to the at least one remote datastore node, by the at least one remote datastore node;

transmitting the remote signature, by the at least one remote datastore node, to all datastore nodes in the domain;

transmitting a signature mismatch notification linked to the remote signature, by the at least one remote datastore node, to the local datastore node upon determining that the remote signature does not match the local signature; and

disconnecting the local datastore node from the domain of datastore nodes upon the local datastore node receiving at least N+1 signature mismatch notifications linked to the remote signature.

6. The method of claim 5, further comprising:

restoring the disconnected local datastore node using the remote signature.

7. A method of establishing a global consistent execution order of transactions, comprising:

dividing a total amount of time into a plurality of epochs, wherein each epoch is a predefined time period having a fixed length;

assigning a plurality of epochIds to the plurality of epochs, wherein the epochIds are a sequentially increasing sequence of numbers that define the epochs;

receiving a first local transaction during a first epoch, and a second local transaction during a second epoch, at a local datastore node;

receiving a first remote transaction during the first epoch, and a second remote transaction during the second epoch, at a remote datastore node, wherein the local and remote datastore nodes are part of a domain of 2N+1 datastore nodes, wherein N is an integer greater than or equal to 1;

assigning a timestamp to each of the transactions upon respectively receiving the transactions, wherein the timestamps indicate a respective arrival time of the transactions;

assigning a first epochId of the plurality of epochIds to the first local transaction and the first remote transaction upon respectively receiving the first local transaction and the first remote transaction at the local and remote datastore nodes;

assigning a second epochId of the plurality of epochIds to the second local transaction and the second remote transaction upon respectively receiving the second local transaction and the second remote transaction at the local and remote datastore nodes;

combining the first local transaction and the first remote transaction into a first transaction set based on the first epochId assigned to the first local transaction and the first remote transaction;

combining the second local transaction and the second remote transaction into a second transaction set based on the second epochId assigned to the second local transaction and the second remote transaction;

assigning a first global order to the first local transaction and the first remote transaction within the first transaction set based on the timestamps assigned to the first local transaction and the first remote transaction without transmitting the first local transaction and the first remote transaction to a centralized sequencer; and

assigning a second global order to the second local transaction and the second remote transaction within the second transaction set based on the timestamps assigned to the second local transaction and the second remote transaction without transmitting the second local transaction and the second remote transaction to the centralized sequencer.

8. The method of claim 7, wherein assigning the first and second global orders is further based on respective node IDs of all transactions within the first and second transaction sets, and the respective node IDs correspond to a node at which the transactions were received.

9. The method of claim 7, wherein the first and second local transactions are read-only transactions, and the read-only transactions are not combined into the first and second transaction sets.

10. The method of claim 7, wherein the fixed length of each epoch is about 20 milliseconds.