A method of ensuring atomicity of transactions across a plurality of active hosts in a distributed environment, is provided. The method generally includes receiving, from a client, a second request to commit a second transaction subsequent to receiving a first request to commit a first transaction; assigning a second prepare identifier (ID) to the second transaction, wherein the second prepare ID assigned to the second transaction is greater than a first prepare ID assigned to the first transaction; transmitting, to the plurality of active hosts, instructions to prepare for committing the second transaction, the instructions including the second prepare ID; receiving, from each host, an acknowledgement indicating successful preparation for committing the second transaction; and transmitting, to the plurality of active hosts, instructions to commit the second transaction prior to receiving, from each host, an acknowledgement indicating successful preparation for committing the first transaction.

BACKGROUND

A transaction is a set of operations that form a logical unit of work. The essential idea of a transaction is indivisibility, i.e., either all or none of the operations of the transaction are permanently performed. Traditionally, transaction semantic is defined by ACID properties, namely atomicity, consistency, integrity, and durability.

Unlike a transaction on a local database, a distributed transaction involves altering data on multiple databases. Consequently, distributed transaction processing is more complicated because the database must coordinate the committing or aborting and rolling back of the changes in a transaction. In a distributed environment, a transaction is decomposed into a set of subtransactions, each of which executes at a single participating site, or server. In order to ensure the atomicity property of a distributed transaction, all sites participating in the transaction must coordinate their actions so that they either unanimously abort or unanimously commit the transaction. This is achieved by employing an atomic commit protocol (ACP) that executes a commit or an abort operation across multiple sites as a single logical operation. An ACP ensures that a distributed transaction is either committed and all its effects become persistent across all participating sites, or aborted and all its effects are obliterated as if the transaction had never executed at any site. In other words, the entire transaction commits, or the entire transaction is aborted and rolls back. In database management, saving data changes is known as a commit and undoing changes is known as a roll back.

The most common ACP for ensuring atomicity of multi-site transactions in a distributed environment is the two-phase commit (2PC) protocol. A distributed system that implements the 2PC protocol can include N sites, where N is an integer greater than zero, each of which includes a representation of the same data. For example, each site may be represented as a host, where each host has access to a separate data store that includes a representation of the same data. Data accessible by each host is updated such that data remains consistent, regardless of any host or network failures in the distributed system. To achieve this result, the 2PC protocol implements a 2PC coordinator (also referred to herein as a master site or 2PC engine) which is a separate entity connected to each of N hosts.

As the name implies, 2PC consists of two phases, namely a voting phase and a decision phase. During the voting phase, or prepare phase, the 2PC coordinator requests all hosts participating in the transaction's execution to prepare-to-commit whereas, during the decision phase, or commit phase, the 2PC coordinator either decides to commit the transaction if all the participating sites (e.g., hosts) are prepared to commit (e.g., voted “yes” or agreed to the transaction), or to abort if any participant has decided to abort (e.g., voted “no” or not agreed to the transaction).

Where all hosts involved in the transaction are prepared to commit (e.g., voted “yes” or agreed to the transaction), then the 2PC coordinator initiates the “commit” phase of the 2PC protocol. During the commit phase, the 2PC coordinator transmits a commit message to each host requesting the host to modify the data according to the transaction. According to the 2PC protocol, once the host receives the commit message, the host is required to perform the transaction, thereby guaranteeing the transaction. On the other hand, where at least one host involved in the transaction has decided to abort (e.g., voted “no” or does not agree to the transaction), then the 2PC coordinator transmits a message to each host that voted “yes” to the transaction, instructing the host to discard whatever temporary information the host has stored during the “prepare” phase.

Although the 2PC protocol may ensure that all participants in a distributed transaction reach consistent states, jitter associated with conventional distributed 2PC systems may adversely impact write input/output (I/O) performance of the system. Jitter is the variation in latency (or delay) during transmission or processing of a signal or packet. Jitter may occur where a signal or packet gets queued or delayed somewhere in transmission from one entity to another, while other signals or packets are not subject to delay. Jitter may result due to variations in I/O write paths at different hosts, network congestion, timing drift, poor hardware performance, and/or congestion at the disk layer.

In the context of distributed 2PC systems, prepare messages generated and transmitted by the 2PC coordinator likely encounter jitter, including both network jitter and disk jitter. For example, a network connection between the 2PC coordinator and one host may have greater latency than a network connection between the 2PC coordinator and another host due to congestion and/or quality of the network connection. This differentiation in latency may cause some prepare messages to arrive at one host earlier than another host also receiving the prepare message. Latency variation may also be apparent in preparation of the write request and return acknowledgement of the prepare message, thereby causing further delay in acknowledging the prepare message.

In other scenarios, jitter during preparation of the client write may be increased due to a cache miss. Specifically, in some cases where deduplication techniques (e.g., data deduplication is a technique for eliminating duplicated or redundant data) are implemented, a caching policy may be introduced such that prior to preparing a write request, each host may determine whether the received write request corresponds to a unique write I/O for the given payload data by checking a cache. The cache may be stored in physical memory (e.g., random access memory (RAM)) configured within a host and act as a small, fast memory that can be used to determine whether the write request is for duplicated payload data already stored at the host by locating a hash corresponding to the payload data in the cache, without needing to access storage. A cache miss may occur when a request to retrieve a hash corresponding to the payload data is not currently in the cache memory. Where this occurs, acknowledgement of the prepare message may be even further delayed (e.g., in addition to the delay already encountered during transmission of the prepare message and/or delay caused by hardware performance problems at the host). In addition, other failures can occur, such as application failure or system failure, while waiting for a client write request to prepare or commit.

Such jitter issues and failures may cause latency amplification in conventional implementations of the distributed 2PC protocol. For example, a delay caused by jitter or failure may cause multiple hosts to receive and process a write I/O at a slower rate. Because write requests are committed sequentially and write requests require all hosts involved in the transaction to agree to the transaction before being committed, delay in receipt and processing of a prepare message may exponentially increase latency on the distributed 2PC system.

In turn, the system may become overloaded with prepared write requests waiting to be committed, thereby negatively affecting overall write I/O performance of the system. In cases where a client write request incurs substantial latency due to jitter or failure, client write requests accumulate and, when the number of accumulated inflight prepare requests reaches a configured threshold, the 2PC system may halt operation and stop working. This may especially occur in systems having a high write I/O workload. Hence, overall write I/O performance using the 2PC protocol may be compromised.

Accordingly, solutions for addressing the latency amplification problem in the distributed 2PC protocol are desired.

It should be noted that the information included in the Background section herein is simply meant to provide a reference for the discussion of certain embodiments in the Detailed Description None of the information included in this Background should be considered as an admission of prior art.

DETAILED DESCRIPTION

Aspects of the present disclosure introduce a jitter-tolerant distributed two-phase commit (2PC) solution to address latency issues associated with conventional distributed 2PC systems. As discussed above, a 2PC protocol breaks a database commit into two phases to ensure correctness and fault tolerance, across distributed transactions, in the distributed database system.

Each client write request received by the distributed 2PC system may be assigned a monotonically increasing log sequence number (LSN). An LSN represents the offset, in bytes, of a log record from the beginning of a database log file. In other words, an LSN corresponds to a given position in a log file and is monotonically increased for each incoming write received by the distributed 2PC system.

LSNs assigned to each incoming write request may have a sequential order property requiring write requests associated with lower LSNs to be committed prior to committing write requests associated with higher LSNs in conventional 2PC systems. In conventional 2PC systems, in the prepare phase, write requests associated with different LSNs can be prepared in parallel regardless of the LSN associated with the write request; however, during the commit phase, the write requests are committed sequentially. Thus, where at least one write request associated with a lower LSN experiences high latency, e.g., caused by jitter or failure as discussed above, the write requests associated with higher LSN values will be blocked. In some cases, especially in systems having a high write input/output (I/O) workload, the number of inflight write requests (e.g., write requests that have been successfully prepared and acknowledged) may exceed a configured threshold of the 2PC system, thereby causing the system to halt operation when a write request with a lower LSN is blocking many write requests with higher LSNs from being committed. Therefore, overall performance of the system may be compromised.

Accordingly, certain aspects of the present disclosure introduce a jitter-tolerant solution for distributed 2PC systems. In some embodiments, a monotonically increasing unique commit LSN is not assigned to a received client write I/O request until the commit phase. In some embodiments, in the prepare phase, a monotonically increasing unique prepare identifier (ID) is generated and assigned to each received client write I/O request. The prepare ID may be assigned to each client write I/O request at the prepare phase, as opposed to assigning the LSN to each client write I/O request at the prepare phase. The prepare ID does not have a sequential order property and, therefore, client write I/O requests prepared with a larger prepare ID may be committed without waiting for updates with lower prepare IDs to be successfully prepared and acknowledged. In this way, when a particular client write I/O request encounters large I/O latency during the prepare phase of the 2PC protocol, the other client write I/O requests received by the distributed 2PC system will not be delayed and, instead, can be committed without waiting for the successful preparation and acknowledgement of the delayed write request I/O, even when the delayed write request I/O is associated with smaller prepare ID values. This reduces latency for the other write request I/Os and helps prevent the number of inflight write requests from exceeding the threshold. Accordingly, the 2PC workflow may not be blocked by any jitter encountered during the prepare phase thereby improving performance of the distributed 2PC system.

FIG. 1depicts example physical and virtual network components of a networking environment100with which embodiments of the present disclosure may be implemented.

The networking environment100includes a data center120connected to a network110. The network110is generally representative of a network of computing entities, such as a local area network (“LAN”) or a wide area network (“WAN”), a network of networks, such as the Internet, or any connection over which data may be transmitted between connected computing entities. For example, data can be transmitted in and out of data center120via network110.

Data center120generally represents a set of networked computing entities, and may comprise a logical overlay network. As shown, data center120includes a host cluster101, storage116, and a gateway150. Gateway150provides components in data center120with connectivity to network110, and is used to communicate with destinations external to data center120. Gateway150may be a virtual computing instance (VCI), a physical device, or a software module running within a host102.

Host cluster101includes hosts102(1) . . . (n) (collectively referred to as hosts102). Each of hosts102may be constructed on a server grade hardware platform, such as hardware platforms108(1) . . . (n) (collectively referred to as hardware platforms108). An example of a hardware platform is an x86 architecture platform. Hosts102may be geographically co-located servers on the same rack or on different racks. Each host102is configured to provide a virtualization layer, also referred to as a hypervisor106(1) . . . (n) (collectively referred to as hypervisors106). Hypervisors106abstract processor, memory, storage, and networking physical resources of hardware platforms108into a number of VCIs or virtual machines (VMs)105(1)1-105(1)x,105(2)1-105(2)y,105(n)1-105(n)z(collectively referred to as VMs105) on hosts102. As shown, multiple VMs105may run concurrently on the same host102.

Each of hypervisors106may run in conjunction with an operating system (OS) (not shown) in its respective host102. In some embodiments, hypervisors106can be installed as system level software directly on hardware platforms108of its respective host102(e.g., referred to as “bare metal” installation) and be conceptually interposed between the physical hardware and the guest OSs executing in the VMs105. Though certain aspects are described herein with respect to VMs running on host machines, it should be understood that such aspects are similarly applicable to physical machines, like host machines102, without the use of virtualization.

In certain embodiments, a 2PC protocol is implemented in networking environment100to provide a distributed 2PC system. Accordingly, one of hosts102is designated as a 2PC coordinator (also referred to as a master site) while remaining active hosts102in networking environment100are designated as the participants (also referred to as components). As shown inFIG. 1, host102(1) is designated as 2PC coordinator and, thus hypervisor106(1) implements a 2PC engine (referred to herein as 2PC coordinator114). It should be noted that the 2PC coordinator114may instead be implemented on host102directly, run on an OS running on host102, etc. 2PC coordinator114is configured to coordinate all activities and synchronizations between active participants in the distributed 2PC system.

Remaining hosts102(2) . . . (n) that are active (e.g., connected and operative) in networking environment100represent participants in the 2PC protocol. Each host102of hosts102(2) . . . (n) has access to a separate storage116(2) . . . (n) (collectively referred to as storage116), respectively, that includes a representation of the same data.

Storage116provides VMs105and/or physical machines102access to consolidated, block-level data storage. As shown in one embodiment, each storage116is storage directly coupled to a host102. In another embodiment, storage116are collectively a virtual storage area network (vSAN) that aggregates local or direct-attached capacity devices of host cluster101and creates a single storage pool shared across all hosts102in host cluster101. In another embodiment, storage116includes local storage in hardware platform108of each host102.

Storage116manages storage of data at a block granularity. For example, storage116is divided into a number of physical blocks (e.g., 4096 bytes or “4K” size blocks), each physical block having a corresponding physical block address (PBA) that indexes the physical block in storage. The physical blocks of storage116are used to store blocks of data (also referred to as data blocks) used by VMs105and/or physical machines102, which may be referenced by logical block addresses (LBAs). Blocks of data may be stored as compressed data or uncompressed data in storage116, such that there may or may not be a one to one correspondence between a physical block on storage116and a data block referenced by an LBA. Storage116receives110requests for a data block from a VM105and/or physical machine102, which VM105and/or physical machine102refers to using an LBA that is in an address space used by the VM and/or physical machines to address blocks of data. Such an LBA may be referred to as an LBA of the data block.

In one embodiment, storage116is a write-ahead-log (WAL)-based storage system including a WAL118(e.g., WAL118(2) . . . (n), collectively referred to as WAL118). In particular, WAL118provides atomicity and durability guarantees in storage116by persisting every change as a command to an append-only log before they are written to storage116. For example, client requests to write data to storage116may be processed by recording the received client write request in WAL118(e.g., as a log record). As described in more detail below, WAL118may be a prepare log, and in some cases, converted to a commit log after both receiving a command from 2PC coordinator114including instructions to commit data and committing the data in data store116, in accordance with the command.

The 2PC protocol implemented in networking environment100breaks a database commit into two phases to ensure correctness and fault tolerance, across hosts102, in the distributed database system. As mentioned previously, the two phases may include a “prepare” phase, also referred to as a voting phase, and a “commit” phase, also referred to as a decision phase.

Conventional distributed 2PC implementations assign an LSN to each write request received by 2PC coordinator114when the write request is received. Accordingly, during the prepare phase of the conventional 2PC protocol, 2PC coordinator114transmits a prepare message to each host102that includes the LSN and the details of the transaction related to the LSN. The LSNs are assigned in a sequential fashion. In particular, a second incoming write request received by 2PC coordinator114is assigned an LSN greater than a first incoming write request received by 2PC coordinator114at an earlier time.

In the conventional distributed 2PC implementations, each host102, after receiving the prepare message, persists payload data associated with the received LSN to its respective storage116. In an illustrative example of the conventional distributed 2PC implementation, 2PC coordinator114receives three client write I/O requests over a period of time, the first received client write I/O request is assigned LSN1, the second received client write I/O request is assigned LSN2, and the third received client write I/O request is assigned LSN3 by 2PC coordinator114. Each host102generates a prepare log record in WAL118for storing metadata associated with the payload data, wherein the metadata includes at least the LSN. Each host102may receive multiple prepare messages, each related to different LSNs or client writes; however, preparation of each write request to relevant disk components of each host102(e.g., persisting the payload data and generating a record in the prepare log) may be independent of the LSN associated with each write. In other words, a write request assigned a larger LSN may be prepared by host102prior to a write request assigned a lower LSN. Following preparation of each write request, each host102sends an acknowledgement message to 2PC coordinator114.

In the conventional distributed 2PC implementations, although a client write may be prepared irrespective of its assigned LSN, a prepared client write with a higher LSN may not be committed until client write I/O requests associated with lower LSNs have been committed (and also, therefore, prepared) by all hosts102receiving the prepare message. Thus, in the illustrative example, the write request associated with LSN3 is not committed at each host102until the write request associated with LSN1 and the write request associated with LSN2 have been committed (and prepared and acknowledged) by all hosts, and similarly, the write request associated with LSN2 is not be committed at each host102until the write request associated with LSN1 has been committed by all hosts. In other words, 2PC coordinator114may not initiate the commit phase for the write request associated with LSN3 until 2PC coordinator114receives acknowledgment of preparation of the write request associated with LSN1 and the write request associated with LSN2 at each host102, and 2PC coordination114may not initiate the commit phase for the write request associated with LSN2 until 2PC coordinator114receives acknowledgement of preparation of the write request associated with LSN1 at each host102. Unfortunately, as discussed above, this may cause blockage of prepared write requests due to jitter and/or failure. For example, if a host102does not prepare and send a successful acknowledgement for the write request associated with LSN1, due to jitter, then the write requests having LSNs behind the unprepared write request, such as the write requests associated with LSN2 and LSN3 are blocked, which ties up resources at 2PC coordinator114until all write requests with lower LSNs (e.g., LSN1) have been prepared and acknowledged.

Accordingly, aspects of the present disclosure introduce a jitter-tolerant solution for distributed 2PC systems in which a prepare ID is assigned in the prepare phase and the LSN is not assigned until the commit phase.FIG. 2is a call flow diagram illustrating example operations200for a jitter-tolerant 2PC protocol in distributed 2PC systems, according to an example embodiment of the present application. As illustrated inFIG. 2, four hosts, host102(1)-102(4) may be active within the distributed 2PC system, and host102(1) may be designated as 2PC coordinator114. Although the embodiment illustrated inFIG. 2illustrates host102(1) designated as 2PC coordinator114, any of active hosts102(2)-102(4) may be designed as 2PC coordinator114in other embodiments. Further, although the embodiment illustrated inFIG. 2includes only four active hosts, any number of active hosts may be considered in other embodiments. Operations202through212ofFIG. 2may relate to the prepare phase of the 2PC protocol, while operations214through228may relate to the commit phase of the 2PC protocol.

Beginning at operation202, 2PC coordinator114receives a client write I/O request. For example, the client write I/O request requests to write data for an LBA, e.g., LBA3, in storage116.

In response to receiving the client write request I/O, at operation204, 2PC coordinator114assigns a monotonically increasing prepare ID to the client write request I/O to write data for LBA3. For example, where previously 2PC coordinator received two client write I/O requests requesting to write data for LBA1 and LBA2 and assigned a first prepare ID, Prepare ID1, to the client write I/O request associated with LBA1 and assigned a second prepare ID, Prepare ID2, to the client write I/O request associated with LBA2, then at block204, 2PC coordinator114may assign a third prepare ID, Prepare ID3, to this incoming write I/O request associated with LBA3 received at operation202. This is unlike conventional distributed 2PC implementations where an LSN, as opposed to a prepare ID, is assigned to each client write I/O request received by 2PC coordinator114. As mentioned previously, where as an LSN has a sequential order property, a prepare ID does not have sequential order property. In other words, a client write I/O request assigned a prepare ID may be prepared and committed irrespective of its prepare ID value or the prepare ID value of other client write I/O requests.

At operation206, 2PC coordinator114transmits a prepare message (e.g., a prepare operation command) to each host102including instructions instructing each host102to prepare the client write I/O request to write data for LBA3. The prepare message may include Prepare ID3and details of the transaction related to the Prepare ID, including payload data for LBA3 to be written to storage116.

At operation208, each host102persists the payload data for LBA3 to their respective storage116, and at operation210, adds the client write I/O request for LBA3 to their respective WAL118. In particular, each host102adds a record to their respective WAL118, the record including metadata associated with the client write I/O request for LBA3. Recorded metadata may include one or more of the LBA associated with the client write request, a number of uncompressed contiguous blocks for which payload data associated with the LBA is stored within (also referred to herein as “numBocks”), the assigned Prepare ID, etc. Accordingly, inFIG. 2, hosts102(2)-(4) create a record in WALs118(2)-(4), respectively, which includes at least LBA3 and Prepare ID3.

At operation212, after successfully preparing the client write for LBA3 at operation208and operation210, each host102transmits an acknowledgement message back to 2PC coordinator114. The acknowledgement message received from each host102indicates to 2PC coordinator114that the client write I/O request for LBA3 has been successfully prepared at that host102.

While the embodiment ofFIG. 2illustrates reception of the prepare message from 2PC coordinator114at each host102at the same time, as mentioned above, this may not be the case in some embodiments, and especially in embodiments involving a distributed 2PC system with increased jitter. As mentioned, jitter in the distributed 2PC system may cause the prepare message received at host102(2) to be received at a different time than the prepare message received at host102(3) and also, in some cases, at a different time than the prepare message received at host102(4). Accordingly, hosts102may begin to persist the payload data at operation208at different times, at least due to different reception times of the prepare message from 2PC coordinator114. For example, the prepare message intended for host102(4) may be delayed such that it is not received by host102(4) until after host102(2) and/or host102(3) has already begun persisting payload data associated with LBA3 at208, adding the client write I/O request to WAL118at operation210, and/or transmitting an acknowledgement message to 2PC coordinator at operation212.

In addition to jitter in transmission, jitter at operation208and operation210(e.g., during preparation) may also cause transmission of the acknowledgement message at operation212to be delayed at some hosts102compared to others. In particular, latency caused by, for example, disk jitter, a cache miss, or poor hardware at each host102as described previously, may contribute to the variation in preparation of the write at different hosts102. This variation for the preparation of the client write I/O request for LBA3 is illustrated inFIG. 2by the staggering pattern of operation208and operation210at each host102.

While timing variation in transmitting the acknowledgement message to 2PC coordinator114at operation212may be due to latency associated with the reception and preparation of the client write request I/O for LBA3, in some cases, the variation may be due to latency associated with the reception and preparation of other client write request I/Os. For example, client write I/O requests requesting to write data for LBA1 and LBA2 were received prior to the client write I/O request requesting to write data for LBA3, thus it is possible that latency in the reception and preparation of write requests for LBA1 and LBA2 may also further delay the preparation and acknowledgment of successful preparation at one host102compared to another host102. However, it is to be noted that each host102may prepare each write request irrespective of its prepare ID. Accordingly, in some cases, the preparation of the write request associated with Prepare ID3 associated with LBA3 may be prepared prior to the write request associated with Prepare ID1 associated with LBA1 and/or the write request associated with Prepare ID2 associated with LBA2, thus the preparation and acknowledgment of successful preparation of the client write I/O request for LBA3 may not be delayed by these writes corresponding to earlier Prepare IDs.

Further,FIG. 2illustrates all active hosts102agreeing to perform the transaction, or in other words, voting “yes” in response to the prepare message transmitted by 2PC coordinator114at operation206. However, in some implementations, one or more hosts102may not agree to perform the transaction, or in other words, vote “no” in response to the prepare message transmitted by 2PC coordinator114. In such a case, 2PC coordinator114initiates an abort phase of the 2PC protocol. Accordingly, remaining operations214through228ofFIG. 2may not be performed, and instead 2PC coordinator114transmits a message to each host102that voted “yes” to the transaction instructing each host102that voted “yes” to discard whatever temporary information each host102had stored during the prepare phase. For purposes of this illustration however, it is assumed that all active hosts102agree to perform the transaction, thus the process illustrated inFIG. 2may continue beyond operation212.

At operation214, after receiving an acknowledgement message from each host102indicating successful preparation of the client write I/O request assigned Prepare ID3 at each host102, 2PC coordinator114acknowledges the client requesting data to be written for LBA3.

At operation216, 2PC coordinator114adds the client write I/O request assigned Prepare ID3 to a commit batch. The commit batch includes one or multiple writes prepared and acknowledged by each active host102. For example, the commit batch includes the client write I/O request assigned to Prepare ID3, the client write I/O request assigned Prepare ID2, and the client write request I/O assigned Prepare ID1.

Further at operation216, each client write I/O request added to the commit batch, and the commit batch is assigned a monotonically increased unique LSN. An LSN may be assigned to each client write I/O request irrespective of its Prepare ID value. For example, for a batch commit including the client write I/O request assigned Prepare ID3 and the client write I/O request assigned Prepare ID2, but not the client write I/O request assigned Prepare ID1, 2PC coordinator114may assign a lower value LSN, e.g., LSN1, to the batch containing the client write I/O requests assigned Prepare ID3 and Prepare ID2 than an LSN assigned to a batch with the client write I/O request assigned Prepare ID1, e.g., LSN 2. The assigned LSN may indicate to each host102the order in which the client write I/O requests in the commit batch are to be committed by each host102as compared to another batch of commit (e.g., a commit message for a commit batch assigned LSN2 may not be sent by 2PC coordinator to each host102and each host may not commit writes in the commit batch assigned LSN2 until writes in a commit batch assigned LSN1 are globally committed). Waiting to assign an LSN to each client write I/O request until after the client write I/O request has been successfully prepared at each active host102helps to ensure that no client write I/O requests are blocked by earlier received client write I/O requests that may have incurred excessive latency during the prepare phase (e.g., due to jitter in the distributed 2PC system).

Accordingly, at operation218, 2PC coordinator114transmits a commit message (e.g., a commit operation command) to each host102instructing each host102to commit the client write I/O requests for the commit batch associated with the assigned LSN, e.g., LSN1 for the commit batch with the write request assigned Prepare ID3 and the write request assigned Prepare ID2. The commit message may include the Prepare IDs and the LSN assigned to the commit batch.

At operation220, each host102commits the previously persisted payload data for LBA3 to their respective storage116, and at operation222, adds the LSN to the original record created for the client write I/O request in WAL118at operation210. In other words, each host102converts the prepare log associated with LBA3 and Prepare ID3 to a commit log by adding the assigned LSN to the log record. Thus, after operation222, a commit log, but no prepare log, for LBA3 and Prepare ID3 may exist at each host102.

At operation224, after successfully committing the payload data for LBA3 to storage116and converting the prepare log to a commit log at operation220and operation222, respectively, each host102transmits an acknowledgement message back to 2PC coordinator114. The acknowledgement message received from each host102indicates to 2PC coordinator114that the client write I/O request for LBA3 has been successfully committed at that host102.

Similar to operations206-212of the prepare phase of the 2PC protocol, operations218-224of the commit phase of the 2PC protocol may also not all occur or complete at the same time at each host102, at least in part due to jitter in the distributed 2PC system. For example, althoughFIG. 2illustrates reception of the commit message at operation218by each host102at the same time, jitter may cause the commit message to be received at host102(2) to be received at a different time than the commit message received at host102(3) and also, in some cases, at a different time than the commit message received at host102(4).

At operation226, after receiving an acknowledgement message from each host102indicating successful commitment of the client write I/O request assigned Prepare ID3 in storage116at each host102, 2PC coordinator114determines the client write I/O request assigned Prepare ID3 has been globally committed (e.g., committed on all active hosts102) and marks it as such.

At operation228, once the client write I/O request has been globally committed, each host102may remove the commit log record corresponding to Prepare ID3 and its assigned LSN. AlthoughFIG. 2illustrates each host102removing the commit log record corresponding to Prepare ID3 and its assigned LSN at the same time, removal of the commit log record may be independent of removal at other hosts102. In other words, host102(2) may remove its commit log record at a time independent of when host102(3) removes its commit log record, and in some cases, further at a time independent of when host102(4) removes its commit log record. In some cases, a host102may determine the client write I/O request associated with Prepare ID3 and its assigned LSN has been globally committed when receiving a new commit message for another batch of client write I/O requests to commit, and thus at this time, determine to remove any commit log records for client write I/O requests that are determined to be globally committed.

In some cases, a failure may occur prior to completion of the jitter-tolerant 2PC protocol described with respect toFIG. 2. In some examples, the failure may constitute an application failure. For example, an application failure may include the failure of a host102to process a prepare message (such as the prepare message received at operation206inFIG. 2) (e.g., a no-memory transient error). Accordingly, a host102which has failed to process the prepare message may return a corresponding failure code in its acknowledgement message (such as the acknowledgement message transmitted to 2PC coordinator114at operation212or operation224inFIG. 2) such that 2PC coordinator114is aware of the failure. In some other examples, the failure may constitute a system failure. For example, a system failure may include the failure of an originally-active host102in the distributed 2PC system. Accordingly, 2PC coordinator114may learn of this failure from another system that monitors the status of hosts102in the distributed 2PC system (e.g., via a notification from the other system). To ensure consensus when such a failure occurs during the 2PC protocol, 2PC coordinator114may be configured to request prepare logs and commit logs from all hosts102in the distributed 2PC system, and based at least in part on the type and number of received logs, determine appropriate action with respect to one or more client write I/O requests that were in the process of being prepared or committed when the failure occurred. Appropriate action may include determining to abort one or more transactions, determining to re-commit one or more transactions with prior assigned LSN(s), determining to re-commit one or more transactions with newly assigned LSN(s), or determining to take no further action with respect to one or more transactions (e.g., neither commit nor abort the one or more transactions).

FIG. 3is an example workflow300for failure recovery for a jitter-tolerant 2PC protocol in distributed 2PC systems, according to an example embodiment of the present application. Workflow300may be used to determine appropriate action for one or more client write I/O requests not yet committed in storage116at each host102when failure occurs in the distributed 2PC system. Workflow300ofFIG. 3may be performed, for example, by 2PC coordinator114.

Workflow300may begin, at operation302, by 2PC coordinator114determining a failure. The failure, as described herein, may be an application failure or a system failure. In response to determining the failure at operation302, at operation304, 2PC coordinator114requests all logs from active hosts102. It may be noted that after determination of a failure, a number of currently active hosts102may be less than a number of hosts102active at the time of receipt of a prepare message (referred to herein as initially active hosts102) for one or more client write I/O requests not yet committed in storage116at each host102involved in the distributed transaction when the failure occurred. At operation306, 2PC coordinator114determines whether logs requested from currently active hosts102include only prepare logs, e.g., log records recorded in WAL118with corresponding Prepare IDs, and no corresponding LSNs.

In an example failure recovery scenario, where 2PC coordinator114receives only prepare logs from currently active hosts102, at operation308, 2PC coordinator114determines whether a number of received prepare logs corresponding to a same prepare ID is less than a number of currently active hosts102. For example, assuming there are currently three active hosts102, 2PC coordinator114may determine whether three prepare logs with Prepare ID1 are received or less than three prepare logs with Prepare ID1 are received to ascertain whether the failure occurred before all currently active hosts102were able to prepare the client write I/O request for Prepare ID1.

Where 2PC coordinator114determines the number of received prepare logs corresponding to a same prepare ID is less than a number of currently active hosts102when a prepare message for the Prepare ID in question was sent, then 2PC coordinator114may conclude that the failure occurred before all currently active hosts102were able to prepare the client write I/O request for a same Prepare ID. Accordingly, at operation310, 2PC coordinator114may abort the transaction by transmitting an abort command for this Prepare ID to all currently active hosts102. The abort command may include instructions instructing each of the currently active hosts102to abort the prepare for the client write I/O request indicated by the Prepare ID.

In another failure recovery scenario, where 2PC coordinator114determines the number of received prepare logs corresponding to a same Prepare ID is not less than a number of currently active hosts102, then at operation312, 2PC coordinator114may determine whether a commit message was previously transmitted for this Prepare ID. Where 2PC coordinator114determines a commit message was prior transmitted at operation312, 2PC coordinator114may conclude that the failure occurred after all active hosts102were able to prepare the client write I/O request for the same Prepare ID. Thus, 2PC coordinator114may determine that the failure is associated with disconnection of an initially active host102(e.g., a number of currently active hosts102is less than a number of initially active hosts102). In particular, given currently active hosts102are all able to provide a prepare log for the same Prepare ID, but are not able to provide any commit logs, and a commit message was transmitted to hosts102for the Prepare ID, 2PC coordinator114is able to ascertain that an initially active host102has become disconnected prior to commitment of the write.

Accordingly, at operation314, 2PC coordinator114may assign a new LSN to the client write I/O request and, at operation316, transmit a re-commit command for this client write I/O request corresponding to the newly assigned LSN. The command may instruct all currently active hosts102to re-commit using the newly assigned LSN. 2PC coordinator114may assign a new LSN in case the disconnected host(s)102are reconnected. In particular, the disconnected host(s)102may possess the originally assigned LSN, but not the newly assigned LSN, thus, when the disconnected host(s)102reconnect they may resync the client write I/O request, for the LSN the disconnected host(s)102possesses does not match the newly assigned LSN.

Referring back to operation312, where 2PC coordinator114determines a commit message was not previously transmitted for this Prepare ID, 2PC coordinator114may conclude that the prepare of the write for this Prepare ID was successful on all hosts102, but no commit message had been transmitted prior to the failure occurring. Accordingly, 2PC coordinator114may resume the 2PC procedure for committing of the client write I/O request corresponding to this Prepare ID and add the client write I/O request to a commit batch, assign a monotonically increasing LSN to the commit batch, and transmit a commit message (e.g., a commit operation command) to each host102instructing each host102to commit the client write I/O requests for the commit batch associated with the assigned LSN.

In yet another failure recovery scenario, returning back to operation306, where 2PC coordinator114does not receive only prepare logs from currently active hosts102, at operation320, 2PC coordinator114determines whether logs requested from currently active hosts102include only commit logs, e.g., log records recorded in WAL118with Prepare IDs and corresponding LSNs. Where 2PC coordinator114receives only commit logs from currently active hosts102, at operation322, 2PC coordinator114may determine no further action is necessary, e.g., neither a commit command nor an abort command is necessary for the client write I/O request because the write was previously committed on all hosts102. Thus, receiving only commit logs from currently active hosts102indicates that the failure occurred after commitment of the client write I/O requests on all hosts102. In some cases, the 2PC coordinator114may further determine the failure occurred after commitment of the client write I/O request on all hosts102but before all commit logs were removed at each host102. In particular, where 2PC coordinator114receives a number of commit logs corresponding to a same prepare ID (and LSN) that is less than a number of currently active hosts102, 2PC coordinator may determine the failure occurred after commitment of the client write I/O request on all hosts102but before all commit logs were removed at each host102. For example, assuming Prepare ID1 (corresponding to LSN1) is a prepare ID that was in the process of being committed when the failure occurred, where 2PC coordinator114receives only commit logs with Prepare ID1 and LSN1, but the number of commit logs received is less than a number of currently active hosts102, 2PC coordinator may conclude the failure occurred after commitment of the client write I/O request but before all commit logs were removed at each host102.

In yet another failure recovery scenario, where 2PC coordinator114receives both prepare logs and commit logs, for a same Prepare ID, from currently active hosts102(e.g., does not receive only prepare logs from currently active hosts102at operation306and does not receive only commit logs from currently active hosts102at operation320), then 2PC coordinator114may determine that the failure occurred after a commit message was transmitted by 2PC coordinator114but before all hosts102(e.g., for which the commit message was transmitted to) were able to successfully commit a client write I/O request corresponding to the Prepare ID found on both the received prepare and commit logs. Accordingly, at operation324, 2PC coordinator114may determine the prior assigned LSN for the client write I/O using one or more of the received commit logs (e.g., commit logs include both the Prepare ID and the assigned LSN). At operation326, 2PC coordinator114may transmit a re-commit command for this client write I/O request corresponding to the previously assigned LSN. The command may instruct all currently active hosts102to re-commit using the previously assigned LSN.

One or more embodiments may be implemented as one or more computer programs or as one or more computer program modules embodied in one or more computer readable media. The term computer readable medium refers to any data storage device that can store data which can thereafter be input to a computer system computer readable media may be based on any existing or subsequently developed technology for embodying computer programs in a manner that enables them to be read by a computer. Examples of a computer readable medium include a hard drive, network attached storage (NAS), read-only memory, random-access memory (e.g., a flash memory device), NVMe storage, Persistent Memory storage, a CD (Compact Discs), CD-ROM, a CD-R, or a CD-RW, a DVD (Digital Versatile Disc), a magnetic tape, and other optical and non-optical data storage devices. The computer readable medium can be a non-transitory computer readable medium. The computer readable medium can also be distributed over a network coupled computer system so that the computer readable code is stored and executed in a distributed fashion. In particular, one or more embodiments may be implemented as a non-transitory computer readable medium comprising instructions that, when executed by one or more processors of a computing system, cause the computing system to perform a method, as described herein.